Merge pull request #2118 from binwiederhier/visitor-per-request

Build the visitor per request instead of sharing it

docs/releases.md

@@ -2274,6 +2274,7 @@ and the [ntfy Android app](https://github.com/binwiederhier/ntfy-android/release
 
 * A user without a tier shares its rate limits with anonymous requests from the same IP, and a login from that IP could switch off the auth failure and account action limits for a concurrent request ([#2119](https://github.com/binwiederhier/ntfy/pull/2119))
 * A tier with a daily email limit of zero sends no emails, a tier edited in place no longer loses pending user stats, and a clock that goes back no longer takes tokens away
+* Every request gets its own visitor instead of sharing one per IP, so a login or logout from the same IP can no longer change the user under a concurrent request, which could panic the account handlers or treat an anonymous request as the logged-in user ([#1051](https://github.com/binwiederhier/ntfy/issues/1051), [#2118](https://github.com/binwiederhier/ntfy/pull/2118))
 
 **Security:**
 

limit/service.go

@@ -166,19 +166,43 @@ func (s *Service) ResetUsage() {
 	}
 }
 
-// Prune forgets keys that have not been used for longer than idle. A key that comes back starts
-// with full buckets and zero counts.
-func (s *Service) Prune(idle time.Duration) {
+// Prune forgets keys that have not been used for longer than idle, and returns how many. A key
+// that comes back starts with full buckets and zero counts.
+func (s *Service) Prune(idle time.Duration) int {
 	cutoff := s.now().Add(-idle).UnixNano()
+	pruned := 0
 	for _, sh := range s.shards {
 		sh.mu.Lock()
 		for key, e := range sh.entries {
 			if e.seen < cutoff {
 				delete(sh.entries, key)
+				pruned++
 			}
 		}
 		sh.mu.Unlock()
 	}
+	return pruned
+}
+
+// Touch marks the key as used now, so Prune keeps it, e.g. for a long-lived subscription
+func (s *Service) Touch(key string) {
+	now := s.now()
+	sh := s.shards[shardIndex(key)]
+	sh.mu.Lock()
+	defer sh.mu.Unlock()
+	sh.entryNoLock(key, now).seen = now.UnixNano()
+}
+
+// Seen returns when the key was last used, or the zero time if the service does not know it
+func (s *Service) Seen(key string) time.Time {
+	sh := s.shards[shardIndex(key)]
+	sh.mu.Lock()
+	defer sh.mu.Unlock()
+	e, ok := sh.entries[key]
+	if !ok {
+		return time.Time{}
+	}
+	return time.Unix(0, e.seen).UTC()
 }
 
 // Limiter returns a util.Limiter view of the key's bucket for the rule, for the readers and

limit/service_test.go

@@ -383,6 +383,27 @@ func BenchmarkAllow_Parallel(b *testing.B) {
 	})
 }
 
+func TestService_TouchAndSeen(t *testing.T) {
+	s, clock := newTestServiceWithClock(t)
+	require.True(t, s.Seen("k").IsZero())
+	s.Touch("k")
+	require.Equal(t, clock.now(), s.Seen("k"))
+	clock.advance(10 * time.Minute)
+	s.Touch("k")
+	s.Prune(5 * time.Minute)
+	require.Equal(t, 1, s.Size()) // Touched, so not idle
+	require.Equal(t, clock.now(), s.Seen("k"))
+	clock.advance(10 * time.Minute)
+	s.Prune(5 * time.Minute)
+	require.True(t, s.Seen("k").IsZero())
+}
+
+func TestService_FirebaseKind(t *testing.T) {
+	kind, ok := ParseKind("firebase")
+	require.True(t, ok)
+	require.Equal(t, KindFirebase, kind)
+}
+
 func TestService_ResetUsage_KeepsBuckets(t *testing.T) {
 	// The daily reset starts the counts over, but the buckets keep their tokens: an email bucket
 	// emptied right before the reset is not refilled by it (it used to be)

limit/types.go

@@ -18,12 +18,13 @@ const (
 	KindTopics                // Topics created in the server's in-memory topic map
 	KindAccount               // Account actions: signups and password reset requests
 	KindAuth                  // Failed authentication attempts
+	KindFirebase              // Firebase sends, paused for a while after Firebase reports the quota exceeded
 	numKinds                  // Sentinel, keep last
 )
 
 // kindNames are the stable names of the kinds, used in the usage table and on the wire; they
 // must never change once released
-var kindNames = [numKinds]string{"requests", "messages", "emails", "calls", "bandwidth", "topics", "account", "auth"}
+var kindNames = [numKinds]string{"requests", "messages", "emails", "calls", "bandwidth", "topics", "account", "auth", "firebase"}
 
 // Rule says how much of a kind a key may use: a token bucket of Burst tokens refilling one token
 // per Refill, a Daily cap, or both. A zero Rule allows everything.

server/server.go

@@ -64,9 +64,9 @@ type Server struct {
 	smtpServerBackend *smtpBackend
 	mailer            mail.Sender
 	topics            map[string]*topic
-	visitors          map[string]*visitor // ip:<ip> or user:<user>
-	limiter           *limit.Service      // Decides on and counts the visitors' usage, keyed by visitor key
-	ban               *ban.Service        // Abuse ban-feed; nil when the feature is disabled (no ban file)
+	limiter           *limit.Service // Decides on and counts the visitors' usage, keyed by visitor key
+	visitors          *visitorEnv    // Shared by all visitors, which are built per request
+	ban               *ban.Service   // Abuse ban-feed; nil when the feature is disabled (no ban file)
 	firebaseClient    *firebaseClient
 	twilio            *twilio.Client
 	messages          int64                               // Total number of messages (persisted if messageCache enabled)
@@ -301,7 +301,6 @@ func New(conf *Config) (*Server, error) {
 		messages:        messages,
 		messagesHistory: []int64{messages},
 		messagesFlushed: messages, // The loaded total is already persisted; only new publishes are deltas
-		visitors:        make(map[string]*visitor),
 		limiter:         limit.New(&limit.Config{ResetTime: conf.VisitorStatsResetTime}),
 		stripe:          stripe,
 		streams:         newStreamRegistry(),
@@ -349,6 +348,7 @@ func New(conf *Config) (*Server, error) {
 		firebaseClient = newFirebaseClient(sender, auther)
 	}
 	s.userManager = userManager
+	s.visitors = newVisitorEnv(conf, s.limiter, messageCache, userManager)
 	s.firebaseClient = firebaseClient
 	if pool != nil {
 		metrics.SetDBPoolsFunc(s.dbPoolStats)
@@ -1116,10 +1116,6 @@ func (s *Server) handlePublishInternal(r *http.Request, v *visitor) (*model.Mess
 			return nil, err
 		}
 	}
-	u := v.User()
-	if s.userManager != nil && u != nil && u.Tier != nil {
-		go v.EnqueueUserStatsDelta()
-	}
 	s.mu.Lock()
 	s.messages++
 	s.mu.Unlock()
@@ -2164,9 +2160,6 @@ func (s *Server) resetStats() {
 	s.mu.Lock()
 	defer s.mu.Unlock() // Includes the database query to avoid races with other processes
 	s.limiter.ResetUsage()
-	for _, v := range s.visitors {
-		v.ResetStats()
-	}
 	if s.userManager != nil {
 		if err := s.userManager.ResetStats(); err != nil {
 			log.Tag(tagResetter).Warn("Failed to write to database: %s", err.Error())
@@ -2178,7 +2171,7 @@ func (s *Server) runFirebaseKeepaliver() {
 	if s.firebaseClient == nil {
 		return
 	}
-	v := newVisitor(s.config, s.limiter, s.messageCache, s.userManager, netip.IPv4Unspecified(), nil) // Background process, not a real visitor, uses IP 0.0.0.0
+	v := newVisitor(s.visitors, netip.IPv4Unspecified(), nil) // Background process, not a real visitor, uses IP 0.0.0.0
 	for {
 		select {
 		case <-time.After(s.config.FirebaseKeepaliveInterval):
@@ -2357,18 +2350,10 @@ func (s *Server) transformMatrixJSON(next handleFunc) handleFunc {
 	}
 }
 
+// visitor returns the visitor for this request. It is built per request, so concurrent requests from
+// the same IP (which share a limiter key) cannot see each other's user (#1051).
 func (s *Server) visitor(ip netip.Addr, user *user.User) *visitor {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	id := visitorID(ip, user, s.config)
-	v, exists := s.visitors[id]
-	if !exists {
-		s.visitors[id] = newVisitor(s.config, s.limiter, s.messageCache, s.userManager, ip, user)
-		return s.visitors[id]
-	}
-	v.Keepalive()
-	v.SetUser(user) // Always update with the latest user, may be nil!
-	return v
+	return newVisitor(s.visitors, ip, user)
 }
 
 func (s *Server) writeJSON(w http.ResponseWriter, v any) error {

server/server_account.go

@@ -326,9 +326,8 @@ func (s *Server) handleAccountTokenCreate(w http.ResponseWriter, r *http.Request
 func (s *Server) handleAccountTokenUpdate(w http.ResponseWriter, r *http.Request, v *visitor) error {
 	u := v.User()
 	if u == nil {
-		// ensureUser already checked, but the user can be nil'd between that check and this
-		// re-read: IP-keyed visitors are shared, and a concurrent anonymous request on the
-		// same IP calls SetUser(nil) on this very object (recurring production panic)
+		// ensureUser already checked; kept as a guard, since a nil user here was a recurring
+		// production panic while visitors were shared between requests (#1051)
 		return errHTTPUnauthorized
 	}
 	req, err := readJSONWithLimit[apiAccountTokenUpdateRequest](r.Body, jsonBodyBytesLimit, true) // Allow empty body!

server/server_account_test.go

@@ -223,7 +223,7 @@ func TestAccount_Get_Anonymous(t *testing.T) {
 		require.Equal(t, int64(0), account.Stats.Messages)
 		require.Equal(t, int64(1004), account.Stats.MessagesRemaining)
 		require.Equal(t, int64(0), account.Stats.Emails)
-		require.Equal(t, int64(16), account.Stats.EmailsRemaining) // What the bucket holds now (burst 16), below the day's 24
+		require.Equal(t, int64(24), account.Stats.EmailsRemaining)
 		require.Equal(t, int64(0), account.Stats.Calls)
 		require.Equal(t, int64(0), account.Stats.CallsRemaining)
 
@@ -240,7 +240,7 @@ func TestAccount_Get_Anonymous(t *testing.T) {
 		require.Equal(t, int64(2), account.Stats.Messages)
 		require.Equal(t, int64(1002), account.Stats.MessagesRemaining)
 		require.Equal(t, int64(1), account.Stats.Emails)
-		require.Equal(t, int64(15), account.Stats.EmailsRemaining) // 16 - 1 in the bucket
+		require.Equal(t, int64(23), account.Stats.EmailsRemaining)
 	})
 }
 
@@ -928,25 +928,25 @@ func TestAccount_Reservation_Delete_Messages_And_Attachments(t *testing.T) {
 	require.Equal(t, int64(2), account.Stats.Messages) // Is not reset!
 }*/
 
-func TestAccount_TokenUpdate_NilUserRace_NoPanic(t *testing.T) {
+func TestAccount_TokenUpdate_NilUser_NoPanic(t *testing.T) {
 	forEachBackend(t, func(t *testing.T, databaseURL string) {
-		// Regression for the recurring production panic in handleAccountTokenUpdate: the
-		// handler re-reads v.User() after ensureUser already checked it, and a concurrent
-		// anonymous request on the same IP-keyed visitor can SetUser(nil) in between. The
-		// handler must return 401, not dereference a nil user.
+		// Regression for a recurring production panic in handleAccountTokenUpdate: the handler
+		// re-reads v.User() after ensureUser checked it, and must return 401 for a nil user, not
+		// dereference it. Visitors were shared between requests then (#1051).
 		conf := newTestConfigWithAuthFile(t, databaseURL)
 		conf.EnableLogin = true
 		s := newTestServer(t, conf)
+		ip := netip.MustParseAddr("9.9.9.9")
 		require.Nil(t, s.userManager.AddUser("phil", "phil12345", user.RoleUser, false))
 		u, err := s.userManager.User("phil")
 		require.Nil(t, err)
-
-		v := s.visitor(netip.MustParseAddr("9.9.9.9"), u)
-		v.SetUser(nil) // The concurrent anonymous request, made deterministic
+		authed := s.visitor(ip, u)
+		anon := s.visitor(ip, nil) // A concurrent anonymous request from the same IP
+		require.NotNil(t, authed.User())
 
 		r, _ := http.NewRequest("PATCH", "/v1/account/token", strings.NewReader("{}"))
 		rr := httptest.NewRecorder()
-		err = s.handleAccountTokenUpdate(rr, r, v)
+		err = s.handleAccountTokenUpdate(rr, r, anon)
 		require.Error(t, err)
 		httpErr, ok := err.(*errHTTP)
 		require.True(t, ok)

server/server_auth.go

@@ -25,8 +25,7 @@ func (s *Server) maybeAuthenticate(r *http.Request) (*http.Request, *visitor, er
 	// Read the "Authorization" header value and exit out early if it's not set
 	ip := extractIPAddress(r, s.config.BehindProxy, s.config.ProxyForwardedHeader, s.config.ProxyTrustedPrefixes)
 	// Stash the extracted client IP in the request context so downstream code (the abuse ban-feed in
-	// handleError) can reuse it without re-parsing headers, and so an account-keyed (tier'd) visitor --
-	// whose shared visitor object has a stale v.ip -- is still attributed to the actual request IP.
+	// handleError) can reuse it without re-parsing headers.
 	r = withContext(r, map[contextKey]any{contextVisitorIP: ip})
 	vip := s.visitor(ip, nil)
 	if s.userManager == nil {

server/server_firebase_test.go

@@ -346,7 +346,7 @@ func TestMaybeTruncateFCMMessage_NotTooLong(t *testing.T) {
 func TestToFirebaseSender_Abuse(t *testing.T) {
 	sender := &testFirebaseSender{allowed: 2}
 	client := newFirebaseClient(sender, &testAuther{})
-	visitor := newVisitor(newTestConfig(t, ""), limit.New(&limit.Config{}), newMemTestCache(t), nil, netip.MustParseAddr("1.2.3.4"), nil)
+	visitor := newVisitor(newVisitorEnv(newTestConfig(t, ""), limit.New(&limit.Config{}), newMemTestCache(t), nil), netip.MustParseAddr("1.2.3.4"), nil)
 
 	require.Nil(t, client.Send(visitor, &model.Message{Topic: "mytopic"}))
 	require.Equal(t, 1, len(sender.Messages()))

server/server_manager.go

@@ -75,8 +75,9 @@ func (s *Server) execManager() {
 
 	// Print stats
 	s.mu.RLock()
-	messagesCount, topicsCount, visitorsCount := s.messages, len(s.topics), len(s.visitors)
+	messagesCount, topicsCount := s.messages, len(s.topics)
 	s.mu.RUnlock()
+	visitorsCount := s.limiter.Size()
 
 	// Update stats
 	s.updateAndWriteStats(messagesCount)
@@ -110,23 +111,14 @@ func (s *Server) execManager() {
 }
 
 func (s *Server) pruneVisitors() {
-	staleVisitors := 0
+	pruned := 0
 	log.
 		Tag(tagManager).
 		Timing(func() {
-			s.mu.Lock()
-			defer s.mu.Unlock()
-			for ip, v := range s.visitors {
-				if v.Stale() {
-					log.Tag(tagManager).With(v).Trace("Deleting stale visitor")
-					delete(s.visitors, ip)
-					staleVisitors++
-				}
-			}
+			pruned = s.limiter.Prune(visitorExpungeAfter) // Visitors are per request; only their keys' buckets and counts stay
 		}).
-		Field("stale_visitors", staleVisitors).
-		Debug("Finished deleting stale visitors")
-	s.limiter.Prune(visitorExpungeAfter) // The keys' buckets and counts go with the visitors
+		Field("pruned_keys", pruned).
+		Debug("Finished deleting stale visitor keys")
 }
 
 func (s *Server) pruneTokens() {

server/server_middleware_test.go

@@ -0,0 +1,62 @@
+package server
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"net/netip"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"heckel.io/ntfy/v2/user"
+)
+
+func TestServer_EnsureUser_AnonymousNotAuthorizedAsConcurrentUser(t *testing.T) {
+	// An anonymous request that overlaps with an authenticated request from the same IP must not
+	// see that request's user, or it could mint a token for the victim's account (#1051)
+	c := newTestConfigWithAuthFile(t, "")
+	c.AuthDefault = user.PermissionDenyAll
+	s := newTestServer(t, c)
+	require.Nil(t, s.userManager.AddUser("phil", "phil12345", user.RoleUser, false))
+	u, err := s.userManager.User("phil")
+	require.Nil(t, err)
+	ip := netip.MustParseAddr("9.9.9.9")
+
+	anon := s.visitor(ip, nil)
+	s.visitor(ip, u) // The concurrent authenticated request, made deterministic
+
+	nextCalled := false
+	next := func(w http.ResponseWriter, r *http.Request, v *visitor) error {
+		nextCalled = true
+		return nil
+	}
+	r, _ := http.NewRequest("POST", "/v1/account/token", nil)
+	err = s.ensureUser(next)(httptest.NewRecorder(), r, anon)
+	require.Equal(t, errHTTPUnauthorized, err)
+	require.False(t, nextCalled)
+}
+
+func TestServer_AuthorizeTopic_UserSurvivesConcurrentAnonymousRequest(t *testing.T) {
+	// The reverse direction: an authenticated request must not be authorized as nobody (and be
+	// rejected under deny-all) because an anonymous request from the same IP overlapped with it
+	// (#1051, #2113)
+	c := newTestConfigWithAuthFile(t, "")
+	c.AuthDefault = user.PermissionDenyAll
+	s := newTestServer(t, c)
+	require.Nil(t, s.userManager.AddUser("phil", "phil12345", user.RoleUser, false))
+	require.Nil(t, s.userManager.AllowAccess("phil", "mytopic", user.PermissionReadWrite))
+	u, err := s.userManager.User("phil")
+	require.Nil(t, err)
+	ip := netip.MustParseAddr("9.9.9.9")
+
+	authed := s.visitor(ip, u)
+	s.visitor(ip, nil) // The concurrent anonymous request, made deterministic
+
+	nextCalled := false
+	next := func(w http.ResponseWriter, r *http.Request, v *visitor) error {
+		nextCalled = true
+		return nil
+	}
+	r, _ := http.NewRequest("PUT", "/mytopic", nil)
+	require.Nil(t, s.authorizeTopicWrite(next)(httptest.NewRecorder(), r, authed))
+	require.True(t, nextCalled)
+}

server/server_payments.go

@@ -63,7 +63,7 @@ func (s *Server) handleBillingTiersGet(w http.ResponseWriter, _ *http.Request, _
 	if err != nil {
 		return err
 	}
-	freeTier := vis.LimitsFor(visitorConfigFrom(s.config), nil)
+	freeTier := s.visitors.ipLimits
 	response := []*apiAccountBillingTier{
 		{
 			// This is a bit of a hack: This is the "Free" tier. It has no tier code, name or price.
@@ -211,7 +211,7 @@ func (s *Server) handleAccountBillingSubscriptionCreateSuccess(w http.ResponseWr
 	if err != nil {
 		return err
 	}
-	v.SetUser(u)
+	v = s.visitor(v.IP(), u) // The webhook acts on behalf of the user
 	logvr(v, r).
 		With(tier).
 		Tag(tagStripe).
@@ -435,7 +435,7 @@ func (s *Server) handleAccountBillingWebhookSubscriptionUpdated(r *http.Request,
 	if err != nil {
 		return err
 	}
-	v.SetUser(u)
+	v = s.visitor(v.IP(), u) // The webhook acts on behalf of the user
 	tier, err := s.userManager.TierByStripePrice(priceID)
 	if err != nil {
 		return err
@@ -458,7 +458,7 @@ func (s *Server) handleAccountBillingWebhookSubscriptionDeleted(r *http.Request,
 	if err != nil {
 		return err
 	}
-	v.SetUser(u)
+	v = s.visitor(v.IP(), u) // The webhook acts on behalf of the user
 	logvr(v, r).
 		Tag(tagStripe).
 		Field("stripe_webhook_type", event.Type).

server/server_payments_test.go

@@ -12,7 +12,6 @@ import (
 	"heckel.io/ntfy/v2/user"
 	"heckel.io/ntfy/v2/util"
 	"io"
-	"net/netip"
 	"os"
 	"path/filepath"
 	"strings"
@@ -261,7 +260,7 @@ func TestPayments_AccountDelete_Cancels_Subscription(t *testing.T) {
 	})
 }
 
-func TestPayments_Checkout_Success_And_Increase_Rate_Limits_Reset_Visitor(t *testing.T) {
+func TestPayments_Checkout_Success_And_Increase_Rate_Limits(t *testing.T) {
 	forEachBackend(t, func(t *testing.T, databaseURL string) {
 		// This test is too overloaded, but it's also a great end-to-end a test.
 		//
@@ -395,12 +394,11 @@ func TestPayments_Checkout_Success_And_Increase_Rate_Limits_Reset_Visitor(t *tes
 		})
 		require.Equal(t, 429, rr.Code)
 
-		// Now let's test the message limiter by faking a ridiculously generous rate limiter
-		v := s.visitor(netip.MustParseAddr("9.9.9.9"), u)
-		v.mu.Lock()
-		v.limits.RequestLimitBurst = 1000000
-		v.limits.RequestLimitReplenish = time.Nanosecond
-		v.mu.Unlock()
+		// Now let's test the message limiter by faking a ridiculously generous request limit; the
+		// tier's request burst never goes below the configured one, and tier limits are resolved
+		// per request from the visitor config the server built at startup
+		s.visitors.limitsConfig.RequestLimitBurst = 1000000
+		s.visitors.limitsConfig.RequestLimitReplenish = time.Nanosecond
 
 		var wg sync.WaitGroup
 		for i := 0; i < 209; i++ {

server/server_test.go

@@ -3627,25 +3627,26 @@ func TestServer_SubscriberRateLimiting_VisitorExpiration(t *testing.T) {
 		rr := request(t, s, "GET", "/upAAAAAAAAAAAA/json?poll=1", "", nil, subscriberFn)
 		require.Equal(t, 200, rr.Code)
 		require.Equal(t, "1.2.3.4", s.topics["upAAAAAAAAAAAA"].rateVisitor.ip.String())
-		require.Equal(t, s.visitors["ip:1.2.3.4"], s.topics["upAAAAAAAAAAAA"].rateVisitor)
+		require.Equal(t, "ip:1.2.3.4", s.topics["upAAAAAAAAAAAA"].rateVisitor.key)
 
 		// Publish message, observe rate visitor tokens being decreased
+		publisher := s.visitor(netip.MustParseAddr("9.9.9.9"), nil)
 		response := request(t, s, "POST", "/upAAAAAAAAAAAA", "some message", nil)
 		require.Equal(t, 200, response.Code)
-		require.Equal(t, int64(0), s.visitors["ip:9.9.9.9"].Stats().Messages)
+		require.Equal(t, int64(0), publisher.Stats().Messages)
 		require.Equal(t, int64(1), s.topics["upAAAAAAAAAAAA"].rateVisitor.Stats().Messages)
-		require.Equal(t, s.visitors["ip:1.2.3.4"], s.topics["upAAAAAAAAAAAA"].rateVisitor)
 
-		// Expire visitor
-		s.visitors["ip:1.2.3.4"].seen = time.Now().Add(-1 * 25 * time.Hour)
-		s.pruneVisitors()
+		// Expire visitor: its subscriber has been gone for a day, and its key is pruned after a day without use
+		s.topics["upAAAAAAAAAAAA"].rateVisitor.seen.Store(time.Now().Add(-25 * time.Hour).UnixNano())
+		s.limiter.Prune(-time.Hour) // Everything is idle for longer than -1h
+		require.True(t, s.topics["upAAAAAAAAAAAA"].rateVisitor.Stale())
 
 		// Publish message again, observe that rateVisitor is not used anymore and is reset
 		response = request(t, s, "POST", "/upAAAAAAAAAAAA", "some message", nil)
 		require.Equal(t, 200, response.Code)
-		require.Equal(t, int64(1), s.visitors["ip:9.9.9.9"].Stats().Messages)
+		require.Equal(t, int64(1), publisher.Stats().Messages)
 		require.Nil(t, s.topics["upAAAAAAAAAAAA"].rateVisitor)
-		require.Nil(t, s.visitors["ip:1.2.3.4"])
+		require.Equal(t, int64(0), s.visitor(netip.MustParseAddr("1.2.3.4"), nil).Stats().Messages)
 	})
 }
 
@@ -5575,7 +5576,7 @@ func TestServer_SubscribeHTTP_NoWriteAfterHandlerReturn(t *testing.T) {
 		// that copied the subscriber list before Unsubscribe ran. In production, this is exactly
 		// how the panic occurs: the goroutine spawned by topic.Publish calls sub() after the
 		// handler has already returned and Go has cleaned up the response writer.
-		v := newVisitor(s.config, s.limiter, s.messageCache, s.userManager, netip.MustParseAddr("9.9.9.9"), nil)
+		v := newVisitor(s.visitors, netip.MustParseAddr("9.9.9.9"), nil)
 		msg := model.NewDefaultMessage("mytopic", "straggler message")
 		_ = copiedSub(v, msg)
 
@@ -5595,7 +5596,7 @@ func TestServer_HandleError_SkipsWriteHeaderOnHijackedConnection(t *testing.T) {
 		r, _ := http.NewRequest("GET", "/mytopic/ws", nil)
 		r.Header.Set("Upgrade", "websocket")
 		r.Header.Set("Connection", "Upgrade")
-		v := newVisitor(s.config, s.limiter, s.messageCache, s.userManager, netip.MustParseAddr("1.2.3.4"), nil)
+		v := newVisitor(s.visitors, netip.MustParseAddr("1.2.3.4"), nil)
 
 		// Test post-upgrade errors wrapped with errWebSocketPostUpgrade (should NOT call WriteHeader)
 		postUpgradeErr := &errWebSocketPostUpgrade{errors.New("websocket: close 1000 (normal)")}
@@ -6025,3 +6026,34 @@ func TestServer_Account_RemainingMatchesAdmissions(t *testing.T) {
 		s.resetStats()
 	}
 }
+
+func TestServer_SubscriptionLimit_ReleasedWhenStreamCloses(t *testing.T) {
+	c := newTestConfig(t, "")
+	c.VisitorSubscriptionLimit = 1
+	s := newTestServer(t, c)
+
+	// The open stream takes the only slot
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		request(t, s, "GET", "/mytopic/json", "", nil, func(r *http.Request) {
+			*r = *r.WithContext(ctx)
+		})
+		close(done)
+	}()
+	waitFor(t, func() bool {
+		s.visitors.subscriptions.mu.Lock()
+		defer s.visitors.subscriptions.mu.Unlock()
+		return len(s.visitors.subscriptions.counts) == 1
+	})
+	rr := request(t, s, "GET", "/mytopic/json?poll=1", "", nil)
+	require.Equal(t, 429, rr.Code)
+	require.Equal(t, 42903, toHTTPError(t, rr.Body.String()).Code)
+
+	// Closing the stream gives it back
+	cancel()
+	<-done
+	rr = request(t, s, "GET", "/mytopic/json?poll=1", "", nil)
+	require.Equal(t, 200, rr.Code)
+	require.Empty(t, s.visitors.subscriptions.counts)
+}

server/util_test.go

@@ -4,12 +4,14 @@ import (
 	"bytes"
 	"crypto/rand"
 	"fmt"
-	"heckel.io/ntfy/v2/user"
 	"net/http"
 	"net/netip"
 	"strings"
 	"testing"
 
+	"heckel.io/ntfy/v2/user"
+	vis "heckel.io/ntfy/v2/visitor"
+
 	"github.com/stretchr/testify/require"
 )
 
@@ -151,14 +153,14 @@ func TestVisitorID(t *testing.T) {
 		ID:   "u_123",
 		Tier: &user.Tier{},
 	}
-	require.Equal(t, "ip:1.2.3.4", visitorID(netip.MustParseAddr("1.2.3.4"), nil, confWithDefaults))
-	require.Equal(t, "ip:2a01:599:b26:2397::", visitorID(netip.MustParseAddr("2a01:599:b26:2397:dbe7:5aa2:95ce:1e83"), nil, confWithDefaults))
-	require.Equal(t, "ip:2001:db8:25:86::", visitorID(netip.MustParseAddr("2001:db8:25:86:1::1"), nil, confWithDefaults))
-	require.Equal(t, "ip:2001:db8:25:86::", visitorID(netip.MustParseAddr("2001:db8:25:86:2::1"), nil, confWithDefaults))
+	require.Equal(t, "ip:1.2.3.4", vis.Key(netip.MustParseAddr("1.2.3.4"), nil, visitorConfigFrom(confWithDefaults)))
+	require.Equal(t, "ip:2a01:599:b26:2397::", vis.Key(netip.MustParseAddr("2a01:599:b26:2397:dbe7:5aa2:95ce:1e83"), nil, visitorConfigFrom(confWithDefaults)))
+	require.Equal(t, "ip:2001:db8:25:86::", vis.Key(netip.MustParseAddr("2001:db8:25:86:1::1"), nil, visitorConfigFrom(confWithDefaults)))
+	require.Equal(t, "ip:2001:db8:25:86::", vis.Key(netip.MustParseAddr("2001:db8:25:86:2::1"), nil, visitorConfigFrom(confWithDefaults)))
 
-	require.Equal(t, "user:u_123", visitorID(netip.MustParseAddr("1.2.3.4"), userWithTier, confWithDefaults))
-	require.Equal(t, "user:u_123", visitorID(netip.MustParseAddr("2a01:599:b26:2397:dbe7:5aa2:95ce:1e83"), userWithTier, confWithDefaults))
+	require.Equal(t, "user:u_123", vis.Key(netip.MustParseAddr("1.2.3.4"), userWithTier, visitorConfigFrom(confWithDefaults)))
+	require.Equal(t, "user:u_123", vis.Key(netip.MustParseAddr("2a01:599:b26:2397:dbe7:5aa2:95ce:1e83"), userWithTier, visitorConfigFrom(confWithDefaults)))
 
-	require.Equal(t, "ip:1.2.0.0", visitorID(netip.MustParseAddr("1.2.3.4"), nil, confWithShortenedPrefixes))
-	require.Equal(t, "ip:2a01:599:b26:2300::", visitorID(netip.MustParseAddr("2a01:599:b26:2397:dbe7:5aa2:95ce:1e83"), nil, confWithShortenedPrefixes))
+	require.Equal(t, "ip:1.2.0.0", vis.Key(netip.MustParseAddr("1.2.3.4"), nil, visitorConfigFrom(confWithShortenedPrefixes)))
+	require.Equal(t, "ip:2a01:599:b26:2300::", vis.Key(netip.MustParseAddr("2a01:599:b26:2397:dbe7:5aa2:95ce:1e83"), nil, visitorConfigFrom(confWithShortenedPrefixes)))
 }

server/visitor.go

@@ -2,7 +2,6 @@ package server
 
 import (
 	"net/netip"
-	"sync"
 	"sync/atomic"
 	"time"
 
@@ -18,33 +17,36 @@ const (
 	// oneDay is an approximation of a day as a time.Duration
 	oneDay = 24 * time.Hour
 
-	// visitorExpungeAfter defines how long a visitor is active before it is removed from memory. This number
-	// has to be very high to prevent e-mail abuse, but it doesn't really affect the other limits anyway, since
-	// they are replenished faster (typically).
+	// visitorExpungeAfter is how long a limiter key may go unused before its buckets and counts are
+	// forgotten (see Server.pruneVisitors), and how long a topic keeps a rate visitor whose subscriber
+	// is gone. This number has to be very high to prevent e-mail abuse, but it doesn't really affect
+	// the other limits anyway, since they are replenished faster (typically).
 	visitorExpungeAfter = oneDay
 )
 
-// visitor represents an API user: its identity (the IP, or the user) and the limits that apply to it.
-// The counting itself happens in the limit service, keyed by the visitor key, so two visitor objects
-// with the same key share one set of buckets and counts.
+// visitor represents one request's API user: its identity (the IP, or the user) and the limits that
+// apply to it. It is built per request and never changed afterwards, so concurrent requests from the
+// same IP cannot see each other's user (#1051). The counting happens in the limit service, keyed by
+// key, so two visitors with the same key share one set of buckets and counts.
 type visitor struct {
+	env    *visitorEnv
+	key    string       // The limiter key: "ip:<prefix>" or "user:<id>", see vis.Key
+	ip     netip.Addr   // Visitor IP address
+	user   *user.User   // Only set if authenticated user, otherwise nil
+	limits *vis.Limits  // Resolved from the config or the user's tier
+	seen   atomic.Int64 // Unix nanos of the last keepalive: the liveness of a subscriber kept as a topic's rate visitor (see Stale)
+}
+
+// visitorEnv is what every visitor of a server shares; it is built once, so building a visitor per
+// request costs little
+type visitorEnv struct {
 	config        *Config
 	limitsConfig  *vis.Config    // The parts of config that keying and limit resolution need (see visitorConfigFrom)
+	ipLimits      *vis.Limits    // The config-based limits of visitors without a tier; never modified
 	limiter       *limit.Service // Decides on and counts usage, keyed by key
+	subscriptions *subscriptionGauge
 	messageCache  *message.Cache
-	userManager   *user.Manager      // May be nil
-	key           string             // The limiter key: "ip:<prefix>" or "user:<id>", see vis.Key
-	ip            netip.Addr         // Visitor IP address
-	user          *user.User         // Only set if authenticated user, otherwise nil
-	limits        *vis.Limits        // Resolved from the config or the user's tier; replaced by SetUser when the tier changes
-	subscriptions *util.FixedLimiter // Gauge of this node's open subscriptions for the visitor; never shared, it protects this node
-	firebase      time.Time          // Next allowed Firebase message
-	seen          time.Time          // Last seen time of this visitor (needed for removal of stale visitors)
-	localMessages atomic.Int64       // Messages this visitor object admitted; the user stats deltas are measured on these, not on the shared counts (see enqueueUserStatsDeltaNoLock)
-	localEmails   atomic.Int64       // Emails this visitor object admitted
-	localCalls    atomic.Int64       // Calls this visitor object admitted
-	statsEnqueued user.Stats         // The local counts already handed to the user manager's write queue
-	mu            sync.RWMutex
+	userManager   *user.Manager // May be nil
 }
 
 type visitorInfo struct {
@@ -65,55 +67,72 @@ type visitorStats struct {
 	AttachmentTotalSizeRemaining int64
 }
 
-func newVisitor(conf *Config, limiter *limit.Service, messageCache *message.Cache, userManager *user.Manager, ip netip.Addr, user *user.User) *visitor {
-	v := &visitor{
+func newVisitorEnv(conf *Config, limiter *limit.Service, messageCache *message.Cache, userManager *user.Manager) *visitorEnv {
+	limitsConfig := visitorConfigFrom(conf)
+	return &visitorEnv{
 		config:        conf,
-		limitsConfig:  visitorConfigFrom(conf),
+		limitsConfig:  limitsConfig,
+		ipLimits:      vis.LimitsFor(limitsConfig, nil),
 		limiter:       limiter,
+		subscriptions: newSubscriptionGauge(),
 		messageCache:  messageCache,
 		userManager:   userManager, // May be nil
-		ip:            ip,
-		user:          user,
-		firebase:      time.Unix(0, 0),
-		seen:          time.Now(),
-		subscriptions: util.NewFixedLimiter(int64(conf.VisitorSubscriptionLimit)),
 	}
-	v.resetLimitsNoLock()
+}
+
+func newVisitor(env *visitorEnv, ip netip.Addr, u *user.User) *visitor {
+	limits := env.ipLimits
+	var tier *user.Tier
+	if u != nil && u.Tier != nil {
+		tier = u.Tier
+		limits = vis.LimitsFor(env.limitsConfig, tier) // Resolved per request: the tier's fields may change at any time
+	}
+	v := &visitor{
+		env:    env,
+		key:    vis.Key(ip, u, env.limitsConfig),
+		ip:     ip,
+		user:   u,
+		limits: limits,
+	}
+	if tier != nil {
+		// Carry the persisted stats into the key's daily counts, if the key is new to the limiter
+		var usage limit.Usage
+		usage.Set(limit.KindMessages, u.Stats.Messages)
+		usage.Set(limit.KindEmails, u.Stats.Emails)
+		usage.Set(limit.KindCalls, u.Stats.Calls)
+		env.limiter.Seed(v.key, usage)
+	}
+	env.limiter.Touch(v.key) // The key is seen now, as the visitor's request is (after Seed, which only fills unknown keys)
+	v.seen.Store(time.Now().UnixNano())
 	return v
 }
 
 func (v *visitor) Context() log.Context {
-	v.mu.RLock()
-	defer v.mu.RUnlock()
-	return v.contextNoLock()
-}
-
-func (v *visitor) contextNoLock() log.Context {
-	info := v.infoLightNoLock()
+	info := v.infoLight()
 	fields := log.Context{
 		"visitor_id":                 v.key,
 		"visitor_ip":                 v.ip.String(),
-		"visitor_seen":               util.FormatTime(v.seen),
+		"visitor_seen":               util.FormatTime(time.Unix(0, v.seen.Load())),
 		"visitor_messages":           info.Stats.Messages,
 		"visitor_messages_limit":     info.Limits.MessageLimit,
 		"visitor_messages_remaining": info.Stats.MessagesRemaining,
-		"visitor_requests_remaining": v.limiter.Tokens(v.key, v.requestRule()),
+		"visitor_requests_remaining": v.env.limiter.Tokens(v.key, v.requestRule()),
 	}
-	if v.config.SMTPSenderFrom != "" {
+	if v.env.config.SMTPSenderFrom != "" {
 		fields["visitor_emails"] = info.Stats.Emails
 		fields["visitor_emails_limit"] = info.Limits.EmailLimit
 		fields["visitor_emails_remaining"] = info.Stats.EmailsRemaining
 	}
-	if v.config.TwilioAccount != "" {
+	if v.env.config.TwilioAccount != "" {
 		fields["visitor_calls"] = info.Stats.Calls
 		fields["visitor_calls_limit"] = info.Limits.CallLimit
 		fields["visitor_calls_remaining"] = info.Stats.CallsRemaining
 	}
 	if v.user == nil {
-		fields["visitor_auth_attempts_remaining"] = v.limiter.Tokens(v.key, v.authRule())
+		fields["visitor_auth_attempts_remaining"] = v.env.limiter.Tokens(v.key, v.authRule())
 	}
 	if v.topicCreationLimited() {
-		fields["visitor_topic_creations_remaining"] = v.limiter.Tokens(v.key, v.topicCreationRule())
+		fields["visitor_topic_creations_remaining"] = v.env.limiter.Tokens(v.key, v.topicCreationRule())
 	}
 	if v.user != nil {
 		fields["user_id"] = v.user.ID
@@ -146,159 +165,152 @@ func visitorExtendedInfoContext(info *visitorInfo) log.Context {
 }
 
 func (v *visitor) RequestAllowed() bool {
-	v.mu.RLock() // limits could be replaced!
-	defer v.mu.RUnlock()
 	if v.limits.RequestLimitBurst <= 0 {
 		return false // A bucket of nothing admits nothing, as before
 	}
-	return v.limiter.Allow(v.key, v.requestRule(), 1).OK
+	return v.env.limiter.Allow(v.key, v.requestRule(), 1).OK
 }
 
+// FirebaseAllowed returns false while Firebase sends are paused for the key (see FirebaseTemporarilyDeny)
 func (v *visitor) FirebaseAllowed() bool {
-	v.mu.RLock()
-	defer v.mu.RUnlock()
-	return !time.Now().Before(v.firebase)
+	rule := v.firebaseRule()
+	if rule.Refill <= 0 {
+		return true
+	}
+	return v.env.limiter.Tokens(v.key, rule) >= 1
 }
 
+// FirebaseTemporarilyDeny pauses Firebase sends for the key for the configured penalty duration
 func (v *visitor) FirebaseTemporarilyDeny() {
-	v.mu.Lock()
-	defer v.mu.Unlock()
-	v.firebase = time.Now().Add(v.config.FirebaseQuotaExceededPenaltyDuration)
+	rule := v.firebaseRule()
+	if rule.Refill <= 0 {
+		return
+	}
+	v.env.limiter.Allow(v.key, rule, 1)
 }
 
 func (v *visitor) MessageAllowed() bool {
-	v.mu.RLock() // limits could be replaced!
-	defer v.mu.RUnlock()
 	if v.limits.MessageLimit <= 0 {
 		return false // A daily limit of zero admits nothing; the limit service reads it as "no cap"
 	}
-	allowed := v.limiter.Allow(v.key, v.messageRule(), 1).OK
+	allowed := v.env.limiter.Allow(v.key, v.messageRule(), 1).OK
 	if allowed {
-		v.localMessages.Add(1)
+		v.enqueueUserStats(&user.Stats{
+			Messages: 1,
+		})
 	}
 	return allowed
 }
 
 func (v *visitor) EmailAllowed() bool {
-	v.mu.RLock() // limits could be replaced!
-	defer v.mu.RUnlock()
 	if v.limits.EmailLimitBurst <= 0 || v.limits.EmailLimit <= 0 {
 		return false // No email bucket, or no emails on the tier: the limit service reads a daily limit of zero as "no cap"
 	}
-	allowed := v.limiter.Allow(v.key, v.emailRule(), 1).OK
+	allowed := v.env.limiter.Allow(v.key, v.emailRule(), 1).OK
 	if allowed {
-		v.localEmails.Add(1)
+		v.enqueueUserStats(&user.Stats{
+			Emails: 1,
+		})
 	}
 	return allowed
 }
 
 func (v *visitor) CallAllowed() bool {
-	v.mu.RLock() // limits could be replaced!
-	defer v.mu.RUnlock()
 	if v.limits.CallLimit <= 0 {
 		return false // A daily limit of zero admits nothing (e.g. a tier without calls, verified number or not)
 	}
-	allowed := v.limiter.Allow(v.key, v.callRule(), 1).OK
+	allowed := v.env.limiter.Allow(v.key, v.callRule(), 1).OK
 	if allowed {
-		v.localCalls.Add(1)
+		v.enqueueUserStats(&user.Stats{
+			Calls: 1,
+		})
 	}
 	return allowed
 }
 
+// enqueueUserStats hands what a tier user's request admitted to the user manager's write queue, which
+// adds it to the persisted stats (deltas add up across requests and cluster nodes)
+func (v *visitor) enqueueUserStats(delta *user.Stats) {
+	if v.env.userManager == nil || v.user == nil || v.user.Tier == nil {
+		return
+	}
+	v.env.userManager.EnqueueUserStats(v.user.ID, delta)
+}
+
+// SubscriptionAllowed takes one of the visitor key's subscription slots on this node, if one is free.
+// Every successful call must be paired with RemoveSubscription.
 func (v *visitor) SubscriptionAllowed() bool {
-	v.mu.RLock()
-	defer v.mu.RUnlock()
-	return v.subscriptions.Allow()
+	return v.env.subscriptions.Acquire(v.key, int64(v.env.config.VisitorSubscriptionLimit))
 }
 
 // TopicCreationAllowed returns true if the visitor is allowed to cause a new topic to be
 // inserted into the server's in-memory topic map. Returns true if no limiter is configured.
 func (v *visitor) TopicCreationAllowed() bool {
-	v.mu.RLock() // limits could be replaced!
-	defer v.mu.RUnlock()
 	if !v.topicCreationLimited() {
 		return true
 	}
-	return v.limiter.Allow(v.key, v.topicCreationRule(), 1).OK
+	return v.env.limiter.Allow(v.key, v.topicCreationRule(), 1).OK
 }
 
 // AuthAllowed returns true if an auth request can be attempted (a token is available)
-// It does not look at v.user: a concurrent request may set it on a shared IP-keyed visitor (#1051)
 func (v *visitor) AuthAllowed() bool {
-	v.mu.RLock() // limits could be replaced!
-	defer v.mu.RUnlock()
-	return v.limiter.Tokens(v.key, v.authRule()) >= 1
+	return v.env.limiter.Tokens(v.key, v.authRule()) >= 1
 }
 
 // AuthFailed records an auth failure
 func (v *visitor) AuthFailed() {
-	v.mu.RLock() // limits could be replaced!
-	defer v.mu.RUnlock()
-	v.limiter.Allow(v.key, v.authRule(), 1)
+	v.env.limiter.Allow(v.key, v.authRule(), 1)
 }
 
 // AccountActionAllowed returns true if a rate-limited account action (signup or password-reset
-// request) is currently allowed for this visitor. Like AuthAllowed, it does not look at v.user.
+// request) is currently allowed for this visitor
 func (v *visitor) AccountActionAllowed() bool {
-	v.mu.RLock() // limits could be replaced!
-	defer v.mu.RUnlock()
-	return v.limiter.Tokens(v.key, v.accountRule()) >= 1
+	return v.env.limiter.Tokens(v.key, v.accountRule()) >= 1
 }
 
 // AccountActionPerformed decreases the account limiter. This is to be called after a rate-limited
 // account action (signup or password-reset request).
 func (v *visitor) AccountActionPerformed() {
-	v.mu.RLock() // limits could be replaced!
-	defer v.mu.RUnlock()
-	v.limiter.Allow(v.key, v.accountRule(), 1)
+	v.env.limiter.Allow(v.key, v.accountRule(), 1)
 }
 
 func (v *visitor) BandwidthAllowed(bytes int64) bool {
-	v.mu.RLock() // limits could be replaced!
-	defer v.mu.RUnlock()
 	if v.limits.AttachmentBandwidthLimit <= 0 {
 		return false // A bandwidth limit of zero admits nothing, as before
 	}
-	return v.limiter.Allow(v.key, v.bandwidthRule(), bytes).OK
+	return v.env.limiter.Allow(v.key, v.bandwidthRule(), bytes).OK
 }
 
+// RemoveSubscription gives back a slot taken by SubscriptionAllowed
 func (v *visitor) RemoveSubscription() {
-	v.mu.RLock()
-	defer v.mu.RUnlock()
-	v.subscriptions.AllowN(-1)
+	v.env.subscriptions.Release(v.key)
 }
 
+// Keepalive marks the visitor and its key as seen. The subscribe handlers call it on the visitor a
+// topic keeps as its rate visitor; nothing else does, so Stale is the subscriber's liveness.
 func (v *visitor) Keepalive() {
-	v.mu.Lock()
-	defer v.mu.Unlock()
-	v.seen = time.Now()
+	v.seen.Store(time.Now().UnixNano())
+	v.env.limiter.Touch(v.key)
 }
 
 // BandwidthLimiter returns the visitor's bandwidth bucket as a util.Limiter, for the readers and
 // writers that meter bytes. A revert on the limiter (a negative AllowN) refunds the bytes.
 func (v *visitor) BandwidthLimiter() util.Limiter {
-	v.mu.RLock() // limits could be replaced!
-	defer v.mu.RUnlock()
 	if v.limits.AttachmentBandwidthLimit <= 0 {
 		return util.NewFixedLimiter(0) // A bandwidth limit of zero admits nothing, as BandwidthAllowed
 	}
-	return v.limiter.Limiter(v.key, v.bandwidthRule())
+	return v.env.limiter.Limiter(v.key, v.bandwidthRule())
 }
 
+// Stale is true when the visitor has not been kept alive for visitorExpungeAfter. It reads the
+// visitor's own keepalive, not the limiter key's last use: publishers charging a topic's rate
+// visitor must not keep a vanished subscriber's limits on the topic.
 func (v *visitor) Stale() bool {
-	v.mu.RLock()
-	defer v.mu.RUnlock()
-	return time.Since(v.seen) > visitorExpungeAfter
+	return time.Since(time.Unix(0, v.seen.Load())) > visitorExpungeAfter
 }
 
 func (v *visitor) Stats() *user.Stats {
-	v.mu.RLock()
-	defer v.mu.RUnlock()
-	return v.statsNoLock()
-}
-
-func (v *visitor) statsNoLock() *user.Stats {
-	usage := v.limiter.Usage(v.key)
+	usage := v.env.limiter.Usage(v.key)
 	return &user.Stats{
 		Messages: usage.Count(limit.KindMessages),
 		Emails:   usage.Count(limit.KindEmails),
@@ -306,143 +318,36 @@ func (v *visitor) statsNoLock() *user.Stats {
 	}
 }
 
-// ResetStats zeroes the visitor's local counts and with them the enqueued seeds its deltas are
-// measured against, at the daily rollover, because the caller zeroes the user's database row and
-// the limiter's counts in the same pass (in a cluster that row is reset once, by the leader).
-// Only the stats resetter calls this.
-func (v *visitor) ResetStats() {
-	v.mu.Lock() // Also protects statsEnqueued
-	defer v.mu.Unlock()
-	v.resetLocalStatsNoLock() // The database row is zeroed by the (leader's) daily reset
-}
-
-// EnqueueUserStatsDelta enqueues the visitor's not-yet-persisted stats increments to the user
-// manager's write queue. Deltas add up across cluster nodes (see Manager.EnqueueUserStats).
-func (v *visitor) EnqueueUserStatsDelta() {
-	v.mu.Lock()
-	defer v.mu.Unlock()
-	v.enqueueUserStatsDeltaNoLock()
-}
-
-// enqueueUserStatsDeltaNoLock measures the deltas on what this visitor object admitted, not on
-// the limiter's counts: those are shared by every visitor object with the same key, and would be
-// enqueued by each of them. The caller must hold v.mu.
-func (v *visitor) enqueueUserStatsDeltaNoLock() {
-	u := v.user // Pinned for the whole function: SetUser(nil) must not turn this into a nil deref
-	if v.userManager == nil || u == nil {
-		return
-	}
-	current := user.Stats{
-		Messages: v.localMessages.Load(),
-		Emails:   v.localEmails.Load(),
-		Calls:    v.localCalls.Load(),
-	}
-	delta := user.Stats{
-		Messages: current.Messages - v.statsEnqueued.Messages,
-		Emails:   current.Emails - v.statsEnqueued.Emails,
-		Calls:    current.Calls - v.statsEnqueued.Calls,
-	}
-	if delta == (user.Stats{}) {
-		return
-	}
-	v.statsEnqueued = current
-	v.userManager.EnqueueUserStats(u.ID, &delta)
-}
-
-// resetLocalStatsNoLock zeroes the local counts and what was enqueued of them. The caller must hold v.mu.
-func (v *visitor) resetLocalStatsNoLock() {
-	v.localMessages.Store(0)
-	v.localEmails.Store(0)
-	v.localCalls.Store(0)
-	v.statsEnqueued = user.Stats{}
-}
-
 // User returns the visitor user, or nil if there is none
 func (v *visitor) User() *user.User {
-	v.mu.RLock()
-	defer v.mu.RUnlock()
 	return v.user // May be nil
 }
 
 // IP returns the visitor IP address
 func (v *visitor) IP() netip.Addr {
-	v.mu.RLock()
-	defer v.mu.RUnlock()
 	return v.ip
 }
 
 // Authenticated returns true if a user successfully authenticated
 func (v *visitor) Authenticated() bool {
-	v.mu.RLock()
-	defer v.mu.RUnlock()
 	return v.user != nil
 }
 
-// SetUser sets the visitors user to the given value. A changed tier (gained, lost, or a tier
-// with other limits) re-resolves the key and the limits; the next call to the limiter uses them,
-// with the buckets as they are: nothing is handed back.
-func (v *visitor) SetUser(u *user.User) {
-	v.mu.Lock()
-	defer v.mu.Unlock()
-	shouldResetLimits := v.user.TierID() != u.TierID() // TierID works with nil receiver
-	tierChanged := shouldResetLimits || (u != nil && u.Tier != nil && *v.user.Tier != *u.Tier)
-	if tierChanged && v.user != nil && v.user.Tier != nil {
-		// Flush increments counted under the old key, under the old account, before the key and
-		// the local counts are reset below; the new key is re-seeded from the new user's persisted
-		// stats. Usage of an IP-keyed visitor (no user, or no tier) is never persisted, so it must
-		// not land on the new account. A tier edited in place resets the local counts too.
-		v.enqueueUserStatsDeltaNoLock()
-	}
-	v.user = u // u may be nil!
-	if tierChanged {
-		v.resetLimitsNoLock()
-	}
-}
-
 // MaybeUserID returns the user ID of the visitor (if any). If this is an anonymous visitor,
 // an empty string is returned.
 func (v *visitor) MaybeUserID() string {
-	v.mu.RLock()
-	defer v.mu.RUnlock()
 	if v.user != nil {
 		return v.user.ID
 	}
 	return ""
 }
 
-// resetLimitsNoLock re-resolves the key and the limits from the current user, and seeds the key's
-// daily counts from the user's persisted stats if the key is new to the limiter. The caller must
-// hold v.mu.
-func (v *visitor) resetLimitsNoLock() {
-	v.key = vis.Key(v.ip, v.user, v.limitsConfig)
-	v.limits = v.limitsNoLock()
-	if v.user != nil && v.user.Tier != nil {
-		var usage limit.Usage
-		usage.Set(limit.KindMessages, v.user.Stats.Messages)
-		usage.Set(limit.KindEmails, v.user.Stats.Emails)
-		usage.Set(limit.KindCalls, v.user.Stats.Calls)
-		v.limiter.Seed(v.key, usage)
-	}
-	v.resetLocalStatsNoLock()                                            // What was admitted so far belongs to the previous key and user, and was flushed by SetUser if it was owed
-	log.Fields(v.contextNoLock()).Debug("Rate limits reset for visitor") // Must be after function, because contextNoLock() describes rate limits
-}
-
 func (v *visitor) Limits() *vis.Limits {
-	v.mu.RLock()
-	defer v.mu.RUnlock()
 	return v.limits
 }
 
-func (v *visitor) limitsNoLock() *vis.Limits {
-	var tier *user.Tier
-	if v.user != nil {
-		tier = v.user.Tier // May be nil, which resolves the config-based limits
-	}
-	return vis.LimitsFor(v.limitsConfig, tier)
-}
-
 // The rules the visitor's limits translate to, built per call (they are small values); the
-// limiter keeps the state. The caller must hold v.mu.
+// limiter keeps the state.
 
 func (v *visitor) requestRule() limit.Rule {
 	return limit.Rule{
@@ -490,44 +395,51 @@ func (v *visitor) bandwidthRule() limit.Rule {
 func (v *visitor) topicCreationRule() limit.Rule {
 	return limit.Rule{
 		Kind:   limit.KindTopics,
-		Burst:  int64(v.config.VisitorTopicCreationLimitBurst),
-		Refill: v.config.VisitorTopicCreationLimitReplenish,
+		Burst:  int64(v.env.config.VisitorTopicCreationLimitBurst),
+		Refill: v.env.config.VisitorTopicCreationLimitReplenish,
 	}
 }
 
 func (v *visitor) topicCreationLimited() bool {
-	return v.config.VisitorTopicCreationLimitBurst > 0 && v.config.VisitorTopicCreationLimitReplenish > 0
+	return v.env.config.VisitorTopicCreationLimitBurst > 0 && v.env.config.VisitorTopicCreationLimitReplenish > 0
 }
 
 func (v *visitor) accountRule() limit.Rule {
 	return limit.Rule{
 		Kind:   limit.KindAccount,
-		Burst:  int64(v.config.VisitorAccountCreationLimitBurst),
-		Refill: v.config.VisitorAccountCreationLimitReplenish,
+		Burst:  int64(v.env.config.VisitorAccountCreationLimitBurst),
+		Refill: v.env.config.VisitorAccountCreationLimitReplenish,
 	}
 }
 
 func (v *visitor) authRule() limit.Rule {
 	return limit.Rule{
 		Kind:   limit.KindAuth,
-		Burst:  int64(v.config.VisitorAuthFailureLimitBurst),
-		Refill: v.config.VisitorAuthFailureLimitReplenish,
+		Burst:  int64(v.env.config.VisitorAuthFailureLimitBurst),
+		Refill: v.env.config.VisitorAuthFailureLimitReplenish,
+	}
+}
+
+// firebaseRule is a single token that takes the penalty duration to come back: taking it pauses sends
+func (v *visitor) firebaseRule() limit.Rule {
+	return limit.Rule{
+		Kind:   limit.KindFirebase,
+		Burst:  1,
+		Refill: v.env.config.FirebaseQuotaExceededPenaltyDuration,
 	}
 }
 
 func (v *visitor) Info() (*visitorInfo, error) {
-	v.mu.RLock()
-	info := v.infoLightNoLock()
-	v.mu.RUnlock()
+	info := v.infoLight()
 
 	// Attachment stats from database
 	var attachmentsBytesUsed int64
 	var err error
 	u := v.User()
 	if u != nil {
-		attachmentsBytesUsed, err = v.messageCache.AttachmentBytesUsedByUser(u.ID)
+		attachmentsBytesUsed, err = v.env.messageCache.AttachmentBytesUsedByUser(u.ID)
 	} else {
-		attachmentsBytesUsed, err = v.messageCache.AttachmentBytesUsedBySender(v.IP().String())
+		attachmentsBytesUsed, err = v.env.messageCache.AttachmentBytesUsedBySender(v.IP().String())
 	}
 	if err != nil {
 		return nil, err
@@ -537,8 +449,8 @@ func (v *visitor) Info() (*visitorInfo, error) {
 
 	// Reservation stats from database
 	var reservations int64
-	if v.userManager != nil && u != nil {
-		reservations, err = v.userManager.ReservationsCount(u.Name)
+	if v.env.userManager != nil && u != nil {
+		reservations, err = v.env.userManager.ReservationsCount(u.Name)
 		if err != nil {
 			return nil, err
 		}
@@ -549,8 +461,8 @@ func (v *visitor) Info() (*visitorInfo, error) {
 	return info, nil
 }
 
-func (v *visitor) infoLightNoLock() *visitorInfo {
-	stats := v.statsNoLock()
+func (v *visitor) infoLight() *visitorInfo {
+	stats := v.Stats()
 	limits := v.limits
 	return &visitorInfo{
 		Limits: limits,
@@ -558,23 +470,13 @@ func (v *visitor) infoLightNoLock() *visitorInfo {
 			Messages:          stats.Messages,
 			MessagesRemaining: zeroIfNegative(limits.MessageLimit - stats.Messages),
 			Emails:            stats.Emails,
-			EmailsRemaining:   v.emailsRemainingNoLock(stats.Emails),
+			EmailsRemaining:   zeroIfNegative(limits.EmailLimit - stats.Emails),
 			Calls:             stats.Calls,
 			CallsRemaining:    zeroIfNegative(limits.CallLimit - stats.Calls),
 		},
 	}
 }
 
-// emailsRemainingNoLock returns what the visitor can send now: the day's remaining emails, but no
-// more than the email bucket holds (it may be empty while the day has room, e.g. after the reset)
-func (v *visitor) emailsRemainingNoLock(emails int64) int64 {
-	remaining := zeroIfNegative(v.limits.EmailLimit - emails)
-	if rule := v.emailRule(); rule.Burst > 0 {
-		remaining = min(remaining, v.limiter.Tokens(v.key, rule))
-	}
-	return remaining
-}
-
 func zeroIfNegative(value int64) int64 {
 	if value < 0 {
 		return 0
@@ -582,11 +484,6 @@ func zeroIfNegative(value int64) int64 {
 	return value
 }
 
-// visitorID returns a unique identifier for a visitor based on user or IP, using configurable prefix bits for IPv4/IPv6
-func visitorID(ip netip.Addr, u *user.User, conf *Config) string {
-	return vis.Key(ip, u, visitorConfigFrom(conf))
-}
-
 // visitorConfigFrom assembles the keying and limit-resolution config from the server config
 func visitorConfigFrom(conf *Config) *vis.Config {
 	return &vis.Config{

server/visitor_subscriptions.go

@@ -0,0 +1,39 @@
+package server
+
+import (
+	"sync"
+)
+
+// subscriptionGauge counts the open subscriptions per visitor key on this node. It is never shared
+// with other nodes: it protects this node's connections.
+type subscriptionGauge struct {
+	counts map[string]int64
+	mu     sync.Mutex // Protects counts
+}
+
+func newSubscriptionGauge() *subscriptionGauge {
+	return &subscriptionGauge{
+		counts: make(map[string]int64),
+	}
+}
+
+// Acquire counts one more subscription for the key, unless the key already has limit of them
+func (g *subscriptionGauge) Acquire(key string, limit int64) bool {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	if g.counts[key] >= limit {
+		return false
+	}
+	g.counts[key]++
+	return true
+}
+
+// Release counts one subscription less for the key, and forgets the key when it has none
+func (g *subscriptionGauge) Release(key string) {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	g.counts[key]--
+	if g.counts[key] <= 0 {
+		delete(g.counts, key)
+	}
+}

server/visitor_subscriptions_test.go

@@ -0,0 +1,51 @@
+package server
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestSubscriptionGauge_AcquireUpToLimit(t *testing.T) {
+	g := newSubscriptionGauge()
+	for i := 0; i < 3; i++ {
+		require.True(t, g.Acquire("ip:1.2.3.4", 3))
+	}
+	require.False(t, g.Acquire("ip:1.2.3.4", 3))
+	require.True(t, g.Acquire("ip:5.6.7.8", 3)) // Keys are counted separately
+}
+
+func TestSubscriptionGauge_ReleaseForgetsKey(t *testing.T) {
+	g := newSubscriptionGauge()
+	require.True(t, g.Acquire("ip:1.2.3.4", 2))
+	require.True(t, g.Acquire("ip:1.2.3.4", 2))
+	require.False(t, g.Acquire("ip:1.2.3.4", 2))
+	g.Release("ip:1.2.3.4")
+	require.True(t, g.Acquire("ip:1.2.3.4", 2)) // A released slot is free again
+	g.Release("ip:1.2.3.4")
+	g.Release("ip:1.2.3.4")
+	require.Empty(t, g.counts) // No key is left behind once its subscriptions are gone
+}
+
+func TestSubscriptionGauge_ConcurrentNeverExceedsLimit(t *testing.T) {
+	g := newSubscriptionGauge()
+	var acquired atomic.Int64
+	var wg sync.WaitGroup
+	for i := 0; i < 100; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if g.Acquire("ip:1.2.3.4", 10) {
+				acquired.Add(1)
+			}
+		}()
+	}
+	wg.Wait()
+	require.Equal(t, int64(10), acquired.Load())
+	for i := 0; i < 10; i++ {
+		g.Release("ip:1.2.3.4")
+	}
+	require.Empty(t, g.counts)
+}

server/visitor_test.go

@@ -2,6 +2,7 @@ package server
 
 import (
 	"io"
+	"net/http/httptest"
 	"net/netip"
 	"strings"
 	"sync"
@@ -43,9 +44,9 @@ func TestVisitor_UserStatsDeltaFlush(t *testing.T) {
 	})
 }
 
-func TestVisitor_SetUser_DoesNotChargeAnonymousUsage(t *testing.T) {
-	// A user who gets a tier takes over their IP-keyed visitor (e.g. the billing callback). What
-	// that visitor counted anonymously must not land on the account.
+func TestVisitor_TierUserDoesNotGetAnonymousUsage(t *testing.T) {
+	// What a user's IP counted anonymously must not land on the account once the user (with a
+	// tier) shows up from that IP.
 	c := newTestConfigWithAuthFile(t, "")
 	c.AuthStatsQueueWriterInterval = 100 * time.Millisecond
 	s := newTestServer(t, c)
@@ -59,7 +60,7 @@ func TestVisitor_SetUser_DoesNotChargeAnonymousUsage(t *testing.T) {
 		require.True(t, v.MessageAllowed())
 	}
 
-	v.SetUser(u)
+	require.True(t, s.visitor(netip.MustParseAddr("9.9.9.9"), u).RequestAllowed())
 	time.Sleep(500 * time.Millisecond)
 	u, err = s.userManager.User("phil")
 	require.Nil(t, err)
@@ -123,9 +124,9 @@ func TestVisitor_TierChangeKeepsBuckets(t *testing.T) {
 	})
 }
 
-func TestVisitor_AccountEmailsRemainingFollowsBucket(t *testing.T) {
-	// The account page shows what the user can send now: no more than the email bucket holds,
-	// even when the daily limit has room left (e.g. right after the daily reset)
+func TestVisitor_AccountEmailsRemainingCountsTheDay(t *testing.T) {
+	// The account shows the day's remaining emails, like it does for messages. The email bucket
+	// may still refuse a send while the day has room left.
 	c := newTestConfig(t, "")
 	c.VisitorEmailLimitBurst = 2
 	c.VisitorEmailLimitReplenish = time.Hour // Daily limit 24
@@ -142,7 +143,7 @@ func TestVisitor_AccountEmailsRemainingFollowsBucket(t *testing.T) {
 	account, err := util.UnmarshalJSON[apiAccountResponse](io.NopCloser(response.Body))
 	require.Nil(t, err)
 	require.Equal(t, int64(2), account.Stats.Emails)
-	require.Equal(t, int64(0), account.Stats.EmailsRemaining)
+	require.Equal(t, int64(22), account.Stats.EmailsRemaining)
 	response = request(t, s, "PUT", "/mytopic", "test", map[string]string{
 		"Email": "test@email.com",
 	})
@@ -233,8 +234,75 @@ func TestVisitor_ZeroLimitsAdmitNothing(t *testing.T) {
 	})
 }
 
+func TestVisitor_UserIsPerRequest(t *testing.T) {
+	// Untiered users and anonymous requests from one IP share a limiter key, but each request
+	// must keep its own user (#1051)
+	s := newTestServer(t, newTestConfigWithAuthFile(t, ""))
+	require.Nil(t, s.userManager.AddUser("phil", "phil12345", user.RoleUser, false))
+	u, err := s.userManager.User("phil")
+	require.Nil(t, err)
+	ip := netip.MustParseAddr("9.9.9.9")
+
+	// The concurrent requests, made deterministic
+	authed := s.visitor(ip, u)
+	anon := s.visitor(ip, nil)
+	require.NotNil(t, authed.User())
+	require.Equal(t, u.ID, authed.User().ID)
+	require.Nil(t, anon.User())
+	authedAgain := s.visitor(ip, u)
+	require.Nil(t, anon.User())
+	require.NotNil(t, authedAgain.User())
+	require.Equal(t, u.ID, authedAgain.User().ID)
+
+	// Limits are still shared per IP: requests and the node-local subscription gauge
+	for authed.RequestAllowed() {
+	}
+	require.False(t, anon.RequestAllowed())
+	for authed.SubscriptionAllowed() {
+	}
+	require.False(t, anon.SubscriptionAllowed())
+	authed.RemoveSubscription()
+	require.True(t, anon.SubscriptionAllowed())
+}
+
+func TestVisitor_SubscriptionLimitSharedPerKey(t *testing.T) {
+	c := newTestConfig(t, "")
+	c.VisitorSubscriptionLimit = 2
+	s := newTestServer(t, c)
+	ip := netip.MustParseAddr("9.9.9.9")
+	v1 := s.visitor(ip, nil)
+	v2 := s.visitor(ip, nil) // A second request from the same IP shares the slots
+	require.True(t, v1.SubscriptionAllowed())
+	require.True(t, v2.SubscriptionAllowed())
+	require.False(t, v1.SubscriptionAllowed())
+	require.True(t, s.visitor(netip.MustParseAddr("1.2.3.4"), nil).SubscriptionAllowed())
+	v1.RemoveSubscription()
+	require.True(t, v2.SubscriptionAllowed())
+}
+
+func TestVisitor_FirebasePenaltyIsSharedPerKey(t *testing.T) {
+	s := newTestServer(t, newTestConfig(t, ""))
+	ip := netip.MustParseAddr("9.9.9.9")
+	v1 := s.visitor(ip, nil)
+	v2 := s.visitor(ip, nil)
+	require.True(t, v2.FirebaseAllowed())
+	v1.FirebaseTemporarilyDeny()
+	require.False(t, v2.FirebaseAllowed())
+	require.True(t, s.visitor(netip.MustParseAddr("1.2.3.4"), nil).FirebaseAllowed())
+}
+
+func TestVisitor_FirebasePenaltyExpires(t *testing.T) {
+	c := newTestConfig(t, "")
+	c.FirebaseQuotaExceededPenaltyDuration = 300 * time.Millisecond
+	s := newTestServer(t, c)
+	v := s.visitor(netip.MustParseAddr("9.9.9.9"), nil)
+	v.FirebaseTemporarilyDeny()
+	require.False(t, v.FirebaseAllowed())
+	require.Eventually(t, v.FirebaseAllowed, 3*time.Second, 50*time.Millisecond) // Sends resume once the penalty has passed
+}
+
 func TestVisitor_AuthLimiter_NotBypassedByConcurrentLogin(t *testing.T) {
-	// A user without a tier shares the IP-keyed visitor with anonymous requests from the same IP.
+	// A user without a tier shares the IP limiter key with anonymous requests from the same IP.
 	// Their login must not switch off the auth failure limiter for a concurrent guesser (#1051).
 	c := newTestConfigWithAuthFile(t, "")
 	c.VisitorAuthFailureLimitBurst = 3
@@ -269,10 +337,48 @@ func TestVisitor_AccountLimiter_NotDeniedByConcurrentLogin(t *testing.T) {
 	require.True(t, vip.AccountActionAllowed())
 }
 
+func TestVisitor_NewVisitorIsNotStale(t *testing.T) {
+	// A visitor is seen when it is built for a request, like it was when visitors were shared
+	s := newTestServer(t, newTestConfig(t, ""))
+	v := s.visitor(netip.MustParseAddr("9.9.9.9"), nil)
+	require.False(t, v.Stale())
+}
+
+func TestVisitor_RateVisitorKeptForExemptSubscriber(t *testing.T) {
+	// A subscriber from a request-limit-exempt prefix spends nothing before it becomes a topic's
+	// rate visitor; the topic must still keep it. Fails if a new visitor counts as stale.
+	c := newTestConfig(t, "")
+	c.VisitorSubscriberRateLimiting = true
+	c.VisitorRequestExemptPrefixes = []netip.Prefix{netip.MustParsePrefix("9.9.9.0/24")}
+	s := newTestServer(t, c)
+	_, err := s.topicsFromIDs(s.visitor(netip.MustParseAddr("1.2.3.4"), nil), "mytopic") // The topic exists already
+	require.Nil(t, err)
+	v := s.visitor(netip.MustParseAddr("9.9.9.9"), nil)
+	topics, err := s.topicsFromIDs(v, "mytopic")
+	require.Nil(t, err)
+	require.Nil(t, s.setRateVisitors(httptest.NewRequest("GET", "/mytopic/json", nil), v, topics))
+	require.NotNil(t, topics[0].RateVisitor())
+}
+
+func TestVisitor_AuthAndAccountLimitsIgnoreUser(t *testing.T) {
+	// The auth and account limits are per key and do not look at the user (as on main since #2119):
+	// a logged-in user may request a password reset, and is charged like everyone on the key
+	c := newTestConfigWithAuthFile(t, "")
+	c.VisitorAuthFailureLimitBurst = 1
+	s := newTestServer(t, c)
+	require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleUser, false))
+	u, err := s.userManager.User("phil")
+	require.Nil(t, err)
+	v := s.visitor(netip.MustParseAddr("9.9.9.9"), u)
+	require.True(t, v.AccountActionAllowed())
+	require.True(t, v.AuthAllowed())
+	v.AuthFailed()
+	require.False(t, v.AuthAllowed())
+}
+
 func TestVisitor_SharedKeyCountsOnce(t *testing.T) {
-	// Two visitor objects with the same user key (the user's own visitor, and the IP-keyed one the
-	// billing callback promotes with SetUser) share one count in the limiter, but each persists
-	// only what it admitted itself, so the user's stats are not double counted
+	// Two visitors with the same user key (the user from two addresses) share one count in the
+	// limiter, and each admitted unit is persisted once, so the user's stats are not double counted
 	forEachBackend(t, func(t *testing.T, databaseURL string) {
 		c := newTestConfigWithAuthFile(t, databaseURL)
 		c.AuthStatsQueueWriterInterval = 100 * time.Millisecond
@@ -286,9 +392,8 @@ func TestVisitor_SharedKeyCountsOnce(t *testing.T) {
 		u, err := s.userManager.User("phil")
 		require.Nil(t, err)
 		v1 := s.visitor(netip.MustParseAddr("1.1.1.1"), u)
-		v2 := s.visitor(netip.MustParseAddr("2.2.2.2"), nil)
-		v2.SetUser(u)
-		require.NotSame(t, v1, v2)
+		v2 := s.visitor(netip.MustParseAddr("2.2.2.2"), u)
+		require.Equal(t, v1.key, v2.key)
 		for i := 0; i < 3; i++ {
 			require.True(t, v1.MessageAllowed())
 		}
@@ -299,9 +404,6 @@ func TestVisitor_SharedKeyCountsOnce(t *testing.T) {
 		require.Equal(t, int64(5), v2.Stats().Messages)
 
 		// Persisted once: 3 + 2, not 5 + 5
-		v1.EnqueueUserStatsDelta()
-		v2.EnqueueUserStatsDelta()
-		v1.EnqueueUserStatsDelta() // Nothing new since the last one
 		waitFor(t, func() bool {
 			u, err := s.userManager.User("phil")
 			require.Nil(t, err)
@@ -328,31 +430,6 @@ func TestVisitor_SharedKeyCountsOnce(t *testing.T) {
 	})
 }
 
-func TestVisitor_ExpungedVisitorKeepsCounts(t *testing.T) {
-	// Expunging a stale visitor object from memory is not a reset: the limiter keeps the key's
-	// counts until the key itself goes idle, so a re-created visitor continues where it left off
-	c := newTestConfig(t, "")
-	c.VisitorMessageDailyLimit = 3
-	s := newTestServer(t, c)
-	ip := netip.MustParseAddr("9.9.9.9")
-	v := s.visitor(ip, nil)
-	require.True(t, v.MessageAllowed())
-	require.True(t, v.MessageAllowed())
-	v.mu.Lock()
-	v.seen = time.Now().Add(-2 * visitorExpungeAfter)
-	v.mu.Unlock()
-	s.pruneVisitors()
-	s.mu.Lock()
-	require.Empty(t, s.visitors)
-	s.mu.Unlock()
-	require.Equal(t, 1, s.limiter.Size()) // The key was used just now, so it stays
-	v2 := s.visitor(ip, nil)
-	require.NotSame(t, v, v2)
-	require.Equal(t, int64(2), v2.Stats().Messages)
-	require.True(t, v2.MessageAllowed())
-	require.False(t, v2.MessageAllowed())
-}
-
 func TestVisitor_DailyResetKeepsEmailBucket(t *testing.T) {
 	// The daily reset starts the day's counts over, but it does not refill the email bucket: a
 	// visitor who emptied it right before the reset still has to wait for the refill
@@ -381,7 +458,7 @@ func TestVisitor_DailyResetKeepsEmailBucket(t *testing.T) {
 	require.Equal(t, int64(0), account.Stats.Messages)
 	require.Equal(t, int64(2), account.Stats.MessagesRemaining)
 	require.Equal(t, int64(0), account.Stats.Emails)
-	require.Equal(t, int64(0), account.Stats.EmailsRemaining)
+	require.Equal(t, int64(24), account.Stats.EmailsRemaining)
 	response = request(t, s, "PUT", "/mytopic", "test", nil)
 	require.Equal(t, 200, response.Code)
 	response = request(t, s, "PUT", "/mytopic", "test", map[string]string{
@@ -459,13 +536,13 @@ func TestVisitor_TierChangeDoesNotReseedCounts(t *testing.T) {
 		u, err = s.userManager.User("phil")
 		require.Nil(t, err)
 		require.Equal(t, int64(0), u.Stats.Messages) // Stale
-		require.Same(t, v, s.visitor(ip, u))
+		v = s.visitor(ip, u)                         // The next request carries the new tier
 		require.Equal(t, int64(4), v.Stats().Messages)
 		for i := 0; i < 6; i++ {
 			require.True(t, v.MessageAllowed())
 		}
 		require.False(t, v.MessageAllowed())
-		fresh := newVisitor(s.config, s.limiter, s.messageCache, s.userManager, ip, u)
+		fresh := s.visitor(ip, u)
 		require.Equal(t, int64(10), fresh.Stats().Messages)
 		require.False(t, fresh.MessageAllowed())
 	})
@@ -485,19 +562,19 @@ func TestVisitor_IPPrefixKeying(t *testing.T) {
 	require.True(t, v.MessageAllowed())
 	require.True(t, v.MessageAllowed())
 	peer := s.visitor(netip.MustParseAddr("1.2.3.200"), nil)
-	require.Same(t, v, peer)
+	require.Equal(t, v.key, peer.key)
 	require.Equal(t, int64(2), peer.Stats().Messages)
 	require.True(t, s.visitor(netip.MustParseAddr("1.2.4.1"), nil).MessageAllowed())
 	vu := s.visitor(netip.MustParseAddr("1.2.3.77"), u)
-	require.Same(t, v, vu)
+	require.Equal(t, v.key, vu.key)
 	require.True(t, vu.MessageAllowed())
 	require.False(t, vu.MessageAllowed())
 	require.False(t, s.visitor(netip.MustParseAddr("1.2.3.4"), nil).MessageAllowed())
 
 	// IPv6: the default /64
 	v6 := s.visitor(netip.MustParseAddr("2001:db8:1:2::1"), nil)
-	require.Same(t, v6, s.visitor(netip.MustParseAddr("2001:db8:1:2:ffff::1"), nil))
-	require.NotSame(t, v6, s.visitor(netip.MustParseAddr("2001:db8:1:3::1"), nil))
+	require.Equal(t, v6.key, s.visitor(netip.MustParseAddr("2001:db8:1:2:ffff::1"), nil).key)
+	require.NotEqual(t, v6.key, s.visitor(netip.MustParseAddr("2001:db8:1:3::1"), nil).key)
 }
 
 func TestVisitor_TierDowngradeReturnsToIPBucket(t *testing.T) {
@@ -522,7 +599,7 @@ func TestVisitor_TierDowngradeReturnsToIPBucket(t *testing.T) {
 	u, err := s.userManager.User("phil")
 	require.Nil(t, err)
 	vu := s.visitor(ip, u)
-	require.NotSame(t, anon, vu)
+	require.NotEqual(t, anon.key, vu.key)
 	require.Equal(t, int64(0), vu.Stats().Messages)
 	for i := 0; i < 5; i++ {
 		require.True(t, vu.MessageAllowed())
@@ -534,7 +611,7 @@ func TestVisitor_TierDowngradeReturnsToIPBucket(t *testing.T) {
 	require.Nil(t, err)
 	require.Nil(t, u.Tier)
 	vd := s.visitor(ip, u)
-	require.Same(t, anon, vd)
+	require.Equal(t, anon.key, vd.key)
 	require.Equal(t, int64(2), vd.Stats().Messages)
 	require.True(t, vd.MessageAllowed())
 	require.False(t, vd.MessageAllowed())
@@ -544,14 +621,13 @@ func TestVisitor_TierDowngradeReturnsToIPBucket(t *testing.T) {
 	u, err = s.userManager.User("phil")
 	require.Nil(t, err)
 	vr := s.visitor(ip, u)
-	require.Same(t, vu, vr)
+	require.Equal(t, vu.key, vr.key)
 	require.Equal(t, int64(5), vr.Stats().Messages)
 }
 
 func TestVisitor_ConcurrentSharedKey(t *testing.T) {
-	// Visitor objects are shared per key and mutated by SetUser on every request. Under
-	// contention, the daily count must stay exact (no over-admission) and the reads must not
-	// race the user swaps; run with -race.
+	// Every request builds its own visitor on a shared key. Under contention, the daily count
+	// must stay exact (no over-admission) and the reads must not race the writes; run with -race.
 	c := newTestConfigWithAuthFile(t, "")
 	c.VisitorRequestLimitBurst = 10000
 	s := newTestServer(t, c)
@@ -644,33 +720,20 @@ func TestVisitor_TierWithoutEmailsAdmitsNone(t *testing.T) {
 	require.Equal(t, 42902, toHTTPError(t, response.Body.String()).Code)
 }
 
-func TestVisitor_TierFieldChangeFlushesStats(t *testing.T) {
-	// A tier edited in place (same ID, other limits) resets the visitor's local counts like a tier
-	// change does, so what was admitted but not yet enqueued must be flushed first, or the user's
-	// persisted stats lose it
-	c := newTestConfigWithAuthFile(t, "")
-	c.AuthStatsQueueWriterInterval = 100 * time.Millisecond
+func TestVisitor_RateVisitorLivenessIsTheSubscribers(t *testing.T) {
+	// A topic keeps its rate visitor for as long as the subscriber is around (its keepalive), not
+	// for as long as its limiter key is in use: the limiter forgetting the key must not drop it,
+	// and publishers charging the key must not keep it
+	c := newTestConfig(t, "")
+	c.VisitorSubscriberRateLimiting = true
 	s := newTestServer(t, c)
-	require.Nil(t, s.userManager.AddTier(&user.Tier{
-		Code:         "test",
-		MessageLimit: 10,
-	}))
-	require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleUser, false))
-	require.Nil(t, s.userManager.ChangeTier("phil", "test"))
-	u, err := s.userManager.User("phil")
+	v := s.visitor(netip.MustParseAddr("9.9.9.9"), nil)
+	topics, err := s.topicsFromIDs(v, "mytopic")
 	require.Nil(t, err)
-	v := s.visitor(netip.MustParseAddr("9.9.9.9"), u)
-	require.True(t, v.MessageAllowed()) // Admitted, not yet enqueued
-	tier, err := s.userManager.Tier("test")
-	require.Nil(t, err)
-	tier.MessageLimit = 20
-	require.Nil(t, s.userManager.UpdateTier(tier))
-	u, err = s.userManager.User("phil")
-	require.Nil(t, err)
-	v.SetUser(u)
-	v.EnqueueUserStatsDelta()
-	time.Sleep(500 * time.Millisecond)
-	u, err = s.userManager.User("phil")
-	require.Nil(t, err)
-	require.Equal(t, int64(1), u.Stats.Messages)
+	require.Nil(t, s.setRateVisitors(httptest.NewRequest("GET", "/mytopic/json", nil), v, topics))
+	s.limiter.Prune(0) // The key is gone, the subscriber is not
+	require.NotNil(t, topics[0].RateVisitor())
+	v.seen.Store(time.Now().Add(-2 * visitorExpungeAfter).UnixNano()) // The subscriber is gone
+	require.True(t, s.limiter.Allow(v.key, v.messageRule(), 1).OK)    // A publisher charges the key
+	require.Nil(t, topics[0].RateVisitor())
 }

visitor/key.go

@@ -1,7 +1,6 @@
 package visitor
 
 import (
-	"fmt"
 	"net/netip"
 
 	"heckel.io/ntfy/v2/user"
@@ -12,12 +11,12 @@ import (
 // tier shares the IP key with everyone behind the same prefix.
 func Key(ip netip.Addr, u *user.User, conf *Config) string {
 	if u != nil && u.Tier != nil {
-		return fmt.Sprintf("user:%s", u.ID)
+		return "user:" + u.ID
 	}
 	if ip.Is4() {
 		ip = netip.PrefixFrom(ip, conf.PrefixBitsIPv4).Masked().Addr()
 	} else if ip.Is6() {
 		ip = netip.PrefixFrom(ip, conf.PrefixBitsIPv6).Masked().Addr()
 	}
-	return fmt.Sprintf("ip:%s", ip.String())
+	return "ip:" + ip.String()
 }