56a2ec9 fix(web): count each visitor against the per-client event stream limit · ermite

M cmd/ermite/web.go +11 -6


@@ -4,11 +4,13 @@ package main
4	4
5	5		import (
6	6		"net/http"
	7	+	"net/netip"
7	8
8	9		"bnema.dev/ermite-api/gen/ermite/ermiteconnect"
9	10		"bnema.dev/ermite/internal/adapters/httpserver"
10	11		"bnema.dev/ermite/internal/adapters/rpc"
11	12		"bnema.dev/ermite/internal/adapters/web"
	13	+	"bnema.dev/ermite/internal/clientlimit"
12	14		"bnema.dev/ermite/internal/config"
13	15		"bnema.dev/ermite/internal/domain"
14	16		)

@@ -28,15 +30,18 @@ func newWeb(cfg config.Config, api *rpc.API) (http.Handler, error) {
28	30		if err != nil {
29	31		return nil, err
30	32		}
31		-	return asPrincipal(domain.Anonymous(), srv), nil
	33	+	return asPrincipal(domain.Anonymous(), cfg.TrustedProxies, srv), nil
32	34		}
33	35
34		-	// asPrincipal runs next with p as the caller of every request: the API calls
35		-	// the web makes carry it (rpc.API.InProcessClient). Browsers are anonymous
36		-	// until the web login of P5b brings a session.
37		-	func asPrincipal(p domain.Principal, next http.Handler) http.Handler {
	36	+	// asPrincipal runs next with p as the caller of every request, attributed to
	37	+	// the browser's address: the API calls the web makes carry both
	38	+	// (rpc.API.InProcessClient), so the per-client limits of the API count each
	39	+	// visitor. Browsers are anonymous until the web login of P5b brings a session.
	40	+	func asPrincipal(p domain.Principal, trusted []netip.Prefix, next http.Handler) http.Handler {
38	41		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
39		-	next.ServeHTTP(w, r.WithContext(domain.ContextWithPrincipal(r.Context(), p)))
	42	+	ctx := domain.ContextWithPrincipal(r.Context(), p)
	43	+	ctx = rpc.ContextWithClient(ctx, clientlimit.ClientAddr(r, trusted))
	44	+	next.ServeHTTP(w, r.WithContext(ctx))
40	45		})
41	46		}
42	47

 import (
 	"net/http"
+	"net/netip"
 	"bnema.dev/ermite-api/gen/ermite/ermiteconnect"
 	"bnema.dev/ermite/internal/adapters/httpserver"
 	"bnema.dev/ermite/internal/adapters/rpc"
 	"bnema.dev/ermite/internal/adapters/web"
+	"bnema.dev/ermite/internal/clientlimit"
 	"bnema.dev/ermite/internal/config"
 	"bnema.dev/ermite/internal/domain"
 )
 	if err != nil {
 		return nil, err
 	}
-	return asPrincipal(domain.Anonymous(), srv), nil
+	return asPrincipal(domain.Anonymous(), cfg.TrustedProxies, srv), nil
 }
-// asPrincipal runs next with p as the caller of every request: the API calls
+// asPrincipal runs next with p as the caller of every request, attributed to
-// the web makes carry it (rpc.API.InProcessClient). Browsers are anonymous
+// the browser's address: the API calls the web makes carry both
-// until the web login of P5b brings a session.
+// (rpc.API.InProcessClient), so the per-client limits of the API count each
-func asPrincipal(p domain.Principal, next http.Handler) http.Handler {
+// visitor. Browsers are anonymous until the web login of P5b brings a session.
+func asPrincipal(p domain.Principal, trusted []netip.Prefix, next http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		next.ServeHTTP(w, r.WithContext(domain.ContextWithPrincipal(r.Context(), p)))
+		ctx := domain.ContextWithPrincipal(r.Context(), p)
+		ctx = rpc.ContextWithClient(ctx, clientlimit.ClientAddr(r, trusted))
+		next.ServeHTTP(w, r.WithContext(ctx))
 	})
 }

M cmd/ermite/web_events_test.go +17 -0


@@ -376,6 +376,23 @@ func TestWebEventsEndOnShutdown(t *testing.T) {
376	376		require.Eventually(t, func() bool { return h.forge.EventLog.Watchers() == 0 }, 5time.Second, 5time.Millisecond)
377	377		}
378	378
	379	+	// The web counts each visitor against the API's limit of streams per client:
	380	+	// one address cannot take every watch slot of the forge.
	381	+	func TestWebEventsLimitPerVisitor(t *testing.T) {
	382	+	h := newHarness(t, harnessOpts{})
	383	+	for range app.MaxWatchersPerClient {
	384	+	h.sse("topic=home", "")
	385	+	}
	386	+	req, err := http.NewRequestWithContext(h.ctx, http.MethodGet, h.url("/-/events?topic=home"), nil)
	387	+	require.NoError(t, err)
	388	+	resp, err := http.DefaultClient.Do(req)
	389	+	require.NoError(t, err)
	390	+	defer func() { _ = resp.Body.Close() }()
	391	+	assert.Equal(t, http.StatusServiceUnavailable, resp.StatusCode)
	392	+	assert.NotEmpty(t, resp.Header.Get("Retry-After"))
	393	+	assert.Equal(t, app.MaxWatchersPerClient, h.forge.EventLog.Watchers())
	394	+	}
	395	+
379	396		// A visitor that goes away frees its watch.
380	397		func TestWebEventsClientDisconnect(t *testing.T) {
381	398		h := newHarness(t, harnessOpts{})


@@ -376,6 +376,23 @@ func TestWebEventsEndOnShutdown(t *testing.T) {
376	require.Eventually(t, func() bool { return h.forge.EventLog.Watchers() == 0 }, 5time.Second, 5time.Millisecond)	376		require.Eventually(t, func() bool { return h.forge.EventLog.Watchers() == 0 }, 5time.Second, 5time.Millisecond)
377	}	377		}
378		378
		379	+	// The web counts each visitor against the API's limit of streams per client:
		380	+	// one address cannot take every watch slot of the forge.
		381	+	func TestWebEventsLimitPerVisitor(t *testing.T) {
		382	+	h := newHarness(t, harnessOpts{})
		383	+	for range app.MaxWatchersPerClient {
		384	+	h.sse("topic=home", "")
		385	+	}
		386	+	req, err := http.NewRequestWithContext(h.ctx, http.MethodGet, h.url("/-/events?topic=home"), nil)
		387	+	require.NoError(t, err)
		388	+	resp, err := http.DefaultClient.Do(req)
		389	+	require.NoError(t, err)
		390	+	defer func() { _ = resp.Body.Close() }()
		391	+	assert.Equal(t, http.StatusServiceUnavailable, resp.StatusCode)
		392	+	assert.NotEmpty(t, resp.Header.Get("Retry-After"))
		393	+	assert.Equal(t, app.MaxWatchersPerClient, h.forge.EventLog.Watchers())
		394	+	}
		395	+
379	// A visitor that goes away frees its watch.	396		// A visitor that goes away frees its watch.
380	func TestWebEventsClientDisconnect(t *testing.T) {	397		func TestWebEventsClientDisconnect(t *testing.T) {
381	h := newHarness(t, harnessOpts{})	398		h := newHarness(t, harnessOpts{})

M internal/adapters/rpc/interceptor.go +12 -2


@@ -89,8 +89,18 @@ func (i *interceptor) authenticate(ctx context.Context, procedure, peerAddr stri
89	89
90	90		type clientKey struct{}
91	91
92		-	// clientFromContext returns the normalized address of the client of a network
93		-	// call; empty for an in-process call.
	92	+	// ContextWithClient attributes the in-process calls made with ctx to the
	93	+	// client at addr (clientlimit.ClientAddr of the browser's request), so that
	94	+	// the per-client limits of the API, such as the event streams one, apply to
	95	+	// each visitor of the web rather than to the web as a whole. A network call
	96	+	// is always attributed to its own peer: the interceptor overwrites it.
	97	+	func ContextWithClient(ctx context.Context, addr string) context.Context {
	98	+	return context.WithValue(ctx, clientKey{}, clientlimit.Key(addr))
	99	+	}
	100	+
	101	+	// clientFromContext returns the normalized address of the client of a call:
	102	+	// its peer for a network call, the one given to ContextWithClient for an
	103	+	// in-process call, empty when there is none.
94	104		func clientFromContext(ctx context.Context) string {
95	105		c, _ := ctx.Value(clientKey{}).(string)
96	106		return c


@@ -89,8 +89,18 @@ func (i *interceptor) authenticate(ctx context.Context, procedure, peerAddr stri
89			89
90		type clientKey struct{}	90		type clientKey struct{}
91			91
92	-	// clientFromContext returns the normalized address of the client of a network	92	+	// ContextWithClient attributes the in-process calls made with ctx to the
93	-	// call; empty for an in-process call.	93	+	// client at addr (clientlimit.ClientAddr of the browser's request), so that
			94	+	// the per-client limits of the API, such as the event streams one, apply to
			95	+	// each visitor of the web rather than to the web as a whole. A network call
			96	+	// is always attributed to its own peer: the interceptor overwrites it.
			97	+	func ContextWithClient(ctx context.Context, addr string) context.Context {
			98	+	return context.WithValue(ctx, clientKey{}, clientlimit.Key(addr))
			99	+	}
			100	+
			101	+	// clientFromContext returns the normalized address of the client of a call:
			102	+	// its peer for a network call, the one given to ContextWithClient for an
			103	+	// in-process call, empty when there is none.
94		func clientFromContext(ctx context.Context) string {	104		func clientFromContext(ctx context.Context) string {
95		c, _ := ctx.Value(clientKey{}).(string)	105		c, _ := ctx.Value(clientKey{}).(string)
96		return c	106		return c

M internal/app/events.go +3 -2


@@ -236,8 +236,9 @@ type WatchParams struct {
236	236		// receive only new events.
237	237		ResumeAfter string
238	238		// Client identifies where the watch comes from, for the limit per client
239		-	// (MaxWatchersPerClient): an opaque key such as a normalized address. Empty
240		-	// for callers in this process, which only count against MaxWatchers.
	239	+	// (MaxWatchersPerClient): an opaque key such as a normalized address; the
	240	+	// web passes its visitor's. Empty only for callers with no client, which
	241	+	// count against MaxWatchers alone.
241	242		Client string
242	243		// Ready, when not nil, is called once the topics are valid and the watch
243	244		// is registered: events published from then on are not missed. Nothing is

 	// receive only new events.
 	ResumeAfter string
 	// Client identifies where the watch comes from, for the limit per client
-	// (MaxWatchersPerClient): an opaque key such as a normalized address. Empty
+	// (MaxWatchersPerClient): an opaque key such as a normalized address; the
-	// for callers in this process, which only count against MaxWatchers.
+	// web passes its visitor's. Empty only for callers with no client, which
+	// count against MaxWatchers alone.
 	Client string
 	// Ready, when not nil, is called once the topics are valid and the watch
 	// is registered: events published from then on are not missed. Nothing is