Bill UnifiedPush pushes to the topic's payer on every node, answer 404 for topics abandoned for 7 days

server/errors.go

@@ -163,6 +163,7 @@ var (
 	errHTTPBadRequestBaseURLNotConfigured            = &errHTTP{40068, http.StatusBadRequest, "invalid request: base-url must be configured on the server to generate links", "https://ntfy.sh/docs/config/", nil}
 	errHTTPBadRequestPasswordHashInvalid             = &errHTTP{40069, http.StatusBadRequest, "invalid request: password hash must be a bcrypt hash with cost 10 or higher, use 'ntfy user hash' to generate", "", nil}
 	errHTTPNotFound                                  = &errHTTP{40401, http.StatusNotFound, "page not found", "", nil}
+	errHTTPNotFoundUnifiedPushTopicAbandoned         = &errHTTP{40402, http.StatusNotFound, "UnifiedPush topic has had no subscriber for too long, subscription expired", "", nil}
 	errHTTPUnauthorized                              = &errHTTP{40101, http.StatusUnauthorized, "unauthorized", "https://ntfy.sh/docs/publish/#authentication", nil}
 	errHTTPForbidden                                 = &errHTTP{40301, http.StatusForbidden, "forbidden", "https://ntfy.sh/docs/publish/#authentication", nil}
 	errHTTPForbiddenAdminSocketOnly                  = &errHTTP{40302, http.StatusForbidden, "forbidden: this operation is only allowed over the local admin socket", "https://ntfy.sh/docs/config/#admin-socket", nil}

server/payer.go

@@ -0,0 +1,137 @@
+package server
+
+import (
+	"errors"
+	"strings"
+	"sync"
+	"time"
+
+	"heckel.io/ntfy/v2/user"
+)
+
+const (
+	// unifiedPushAbandonedAfter is how long a UnifiedPush topic may go without subscriber activity
+	// (subscribe, poll, open stream) before a push to it answers 404 instead of 507: RFC 8030
+	// section 7.3 signals an expired subscription with 404, and app servers like Mastodon then
+	// delete it (#664). It is long enough for a phone that is offline for a few days.
+	unifiedPushAbandonedAfter = 7 * oneDay
+
+	// payerUserCacheTTL is how long a payer's user is cached (see payerUserCache)
+	payerUserCacheTTL = time.Minute
+)
+
+// payerUserCache caches the users of the topics' payers by ID, so a push does not read the user
+// database; a change to the user (a new tier) applies to pushes within payerUserCacheTTL
+type payerUserCache struct {
+	lookup  func(id string) (*user.User, error) // userManager.UserByID, nil without a user manager; replaced in tests
+	entries map[string]*payerUserEntry
+	mu      sync.Mutex // Protects entries
+}
+
+// payerUserEntry is a cached user; nil if it does not exist (anymore)
+type payerUserEntry struct {
+	user    *user.User
+	fetched time.Time
+}
+
+func newPayerUserCache(lookup func(id string) (*user.User, error)) *payerUserCache {
+	return &payerUserCache{
+		lookup:  lookup,
+		entries: make(map[string]*payerUserEntry),
+	}
+}
+
+// payerVisitor returns the visitor of the topic's live payer, or nil if it has none. A payer is
+// live while the topic has a stream open here, or for visitorExpungeAfter after the subscriber was
+// last seen on any node (the manager of a node with an open stream touches it). A deleted
+// user pays for nothing; a failed user lookup is returned, so the publisher retries.
+func (s *Server) payerVisitor(t *topic) (*visitor, error) {
+	if !s.config.VisitorSubscriberRateLimiting || !isUnifiedPushTopic(t.ID) {
+		return nil, nil
+	}
+	p := s.limiter.Payer(t.ID)
+	if p == nil || p.Key == "" {
+		return nil, nil
+	} else if t.SubscribersCount() == 0 && time.Since(p.SeenAt) > visitorExpungeAfter {
+		return nil, nil
+	}
+	var u *user.User
+	if p.UserID != "" {
+		var err error
+		if u, err = s.payerUsers.User(p.UserID); err != nil {
+			return nil, err
+		} else if u == nil {
+			return nil, nil
+		}
+	}
+	return newPayerVisitor(s.visitors, p.Key, u), nil
+}
+
+// unifiedPushTopicAbandoned returns true if the topic has had no subscriber activity, on any node,
+// for unifiedPushAbandonedAfter, counted from its first refused push if it never had any
+func (s *Server) unifiedPushTopicAbandoned(t *topic) bool {
+	if !isUnifiedPushTopic(t.ID) || t.SubscribersCount() > 0 {
+		return false
+	}
+	p := s.limiter.Payer(t.ID)
+	return p != nil && time.Since(p.SeenAt) > unifiedPushAbandonedAfter
+}
+
+// touchPayers marks subscriber activity on the UnifiedPush topics now, keeping their payers
+func (s *Server) touchPayers(topics []*topic) {
+	if !s.config.VisitorSubscriberRateLimiting {
+		return
+	}
+	for _, t := range topics {
+		if isUnifiedPushTopic(t.ID) {
+			s.limiter.TouchPayer(t.ID)
+		}
+	}
+}
+
+// User returns the user with the given ID, from the cache if it was looked up within
+// payerUserCacheTTL, or nil if it does not exist or is deleted
+func (c *payerUserCache) User(id string) (*user.User, error) {
+	c.mu.Lock()
+	e, ok := c.entries[id]
+	c.mu.Unlock()
+	if ok && time.Since(e.fetched) < payerUserCacheTTL {
+		return e.user, nil
+	}
+	var u *user.User
+	if c.lookup != nil {
+		var err error
+		u, err = c.lookup(id)
+		if errors.Is(err, user.ErrUserNotFound) {
+			u = nil
+		} else if err != nil {
+			return nil, err
+		} else if u.Deleted {
+			u = nil
+		}
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.entries[id] = &payerUserEntry{
+		user:    u,
+		fetched: time.Now(),
+	}
+	return u, nil
+}
+
+// Prune forgets the users looked up longer than payerUserCacheTTL ago
+func (c *payerUserCache) Prune() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	for id, e := range c.entries {
+		if time.Since(e.fetched) >= payerUserCacheTTL {
+			delete(c.entries, id)
+		}
+	}
+}
+
+// isUnifiedPushTopic returns true if the topic may have a payer: an "up..." topic of the
+// UnifiedPush length
+func isUnifiedPushTopic(id string) bool {
+	return strings.HasPrefix(id, unifiedPushTopicPrefix) && len(id) == unifiedPushTopicLength
+}

server/payer_test.go

@@ -0,0 +1,339 @@
+package server
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/netip"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"heckel.io/ntfy/v2/db"
+	"heckel.io/ntfy/v2/db/pg"
+	"heckel.io/ntfy/v2/limit"
+	"heckel.io/ntfy/v2/user"
+	"heckel.io/ntfy/v2/util"
+)
+
+func TestServer_UnifiedPush_PayerOnOtherNode(t *testing.T) {
+	// Two nodes on one database: a push on the other node than the subscriber's is billed to the
+	// subscriber once the nodes have synced, not answered with a 507
+	forEachBackend(t, func(t *testing.T, databaseURL string) {
+		c := newTestConfigWithAuthFile(t, databaseURL)
+		c.VisitorSubscriberRateLimiting = true
+		s1 := newTestServer(t, c)
+		s2 := newTestServer(t, c)
+		rr := request(t, s1, "GET", "/upAAAAAAAAAAAA/json?poll=1", "", nil, func(r *http.Request) {
+			r.RemoteAddr = "1.2.3.4:1234"
+		})
+		require.Equal(t, 200, rr.Code)
+		waitForWithMaxWait(t, 3*syncWait, func() bool {
+			return payerKey(s2, "upAAAAAAAAAAAA") == "ip:1.2.3.4"
+		})
+		response := request(t, s2, "PUT", "/upAAAAAAAAAAAA?up=1", "some message", nil)
+		require.Equal(t, 200, response.Code)
+		require.Equal(t, int64(1), s2.visitor(netip.MustParseAddr("1.2.3.4"), nil).Stats().Messages)
+		require.Equal(t, int64(0), s2.visitor(netip.MustParseAddr("9.9.9.9"), nil).Stats().Messages)
+	})
+}
+
+func TestServer_UnifiedPush_StreamOnOtherNodeKeepsPayer(t *testing.T) {
+	// A stream open on one node keeps the topic's payer live on the other nodes too: the manager
+	// touches it, so it does not expire a day after the subscribe
+	forEachBackend(t, func(t *testing.T, databaseURL string) {
+		c := newTestConfigWithAuthFile(t, databaseURL)
+		c.VisitorSubscriberRateLimiting = true
+		s1 := newTestServer(t, c)
+		s2 := newTestServer(t, c)
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		go func() {
+			request(t, s1, "GET", "/upAAAAAAAAAAAA/json", "", nil, func(r *http.Request) {
+				r.RemoteAddr = "1.2.3.4:1234"
+				*r = *r.WithContext(ctx)
+			})
+		}()
+		waitFor(t, func() bool {
+			to := lockedTopic(s1, "upAAAAAAAAAAAA")
+			return to != nil && to.SubscribersCount() == 1
+		})
+
+		// A day later, the stream is still open on s1
+		agePayer(s1, "upAAAAAAAAAAAA", 25*time.Hour)
+		s1.execManager()
+		waitForWithMaxWait(t, 3*syncWait, func() bool {
+			p := s2.limiter.Payer("upAAAAAAAAAAAA")
+			return p != nil && p.Key == "ip:1.2.3.4" && time.Since(p.SeenAt) < time.Hour
+		})
+		response := request(t, s2, "PUT", "/upAAAAAAAAAAAA?up=1", "some message", nil)
+		require.Equal(t, 200, response.Code)
+	})
+}
+
+func TestServer_UnifiedPush_AbandonedTopicAnswers404(t *testing.T) {
+	// A UnifiedPush push to a topic without subscriber activity for unifiedPushAbandonedAfter is
+	// told to go away (404, RFC 8030 section 7.3); before that, it may be retried (507)
+	c := newTestConfig(t, "")
+	c.VisitorSubscriberRateLimiting = true
+	s := newTestServer(t, c)
+	push := func() *errHTTP {
+		response := request(t, s, "PUT", "/upAAAAAAAAAAAA?up=1", "some message", nil)
+		if response.Code == 200 {
+			return nil
+		}
+		return toHTTPError(t, response.Body.String())
+	}
+
+	// Never subscribed: 507 for a week from the first push, then 404
+	require.Equal(t, 50701, push().Code)
+	agePayer(s, "upAAAAAAAAAAAA", unifiedPushAbandonedAfter-time.Hour)
+	require.Equal(t, 50701, push().Code)
+	agePayer(s, "upAAAAAAAAAAAA", unifiedPushAbandonedAfter+time.Hour)
+	require.Equal(t, 40402, push().Code)
+	require.Equal(t, 404, push().HTTPCode)
+
+	// A subscriber comes (back): it pays again; a week after it is gone, 404
+	rr := request(t, s, "GET", "/upAAAAAAAAAAAA/json?poll=1", "", nil, func(r *http.Request) {
+		r.RemoteAddr = "1.2.3.4:1234"
+	})
+	require.Equal(t, 200, rr.Code)
+	require.Nil(t, push())
+	agePayer(s, "upAAAAAAAAAAAA", 2*oneDay)
+	require.Equal(t, 50701, push().Code)
+	agePayer(s, "upAAAAAAAAAAAA", unifiedPushAbandonedAfter+time.Hour)
+	require.Equal(t, 40402, push().Code)
+
+	// An open stream keeps the topic, however long ago it was seen otherwise
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go func() {
+		request(t, s, "GET", "/upAAAAAAAAAAAA/json", "", nil, func(r *http.Request) {
+			r.RemoteAddr = "1.2.3.4:1234"
+			*r = *r.WithContext(ctx)
+		})
+	}()
+	waitFor(t, func() bool {
+		return lockedTopic(s, "upAAAAAAAAAAAA").SubscribersCount() == 1
+	})
+	agePayer(s, "upAAAAAAAAAAAA", unifiedPushAbandonedAfter+time.Hour)
+	require.Nil(t, push())
+}
+
+func TestServer_UnifiedPush_AbandonedTopicOnOtherNode(t *testing.T) {
+	// The first refused push starts the countdown for every node: a node started a week later
+	// answers 404
+	forEachBackend(t, func(t *testing.T, databaseURL string) {
+		c := newTestConfigWithAuthFile(t, databaseURL)
+		c.VisitorSubscriberRateLimiting = true
+		s1 := newTestServer(t, c)
+		response := request(t, s1, "PUT", "/upAAAAAAAAAAAA?up=1", "some message", nil)
+		require.Equal(t, 507, response.Code)
+		s1.Stop() // Writes the topic
+		execUserDB(t, c, fmt.Sprintf(`UPDATE topic_payer SET seen_at = seen_at - %d`, int64((unifiedPushAbandonedAfter+time.Hour)/time.Second)))
+		s2 := newTestServer(t, c)
+		response = request(t, s2, "PUT", "/upAAAAAAAAAAAA?up=1", "some message", nil)
+		require.Equal(t, 404, response.Code)
+		require.Equal(t, 40402, toHTTPError(t, response.Body.String()).Code)
+	})
+}
+
+func TestServer_UnifiedPush_NonUnifiedPushTopicIsNotAbandoned(t *testing.T) {
+	// A topic that cannot have a payer (not an "up..." topic) keeps answering 507
+	c := newTestConfig(t, "")
+	c.VisitorSubscriberRateLimiting = true
+	s := newTestServer(t, c)
+	response := request(t, s, "PUT", "/mytopic?up=1", "some message", nil)
+	require.Equal(t, 507, response.Code)
+	require.Nil(t, s.limiter.Payer("mytopic"))
+}
+
+func TestServer_MatrixGateway_AbandonedTopicRejectsPushKey(t *testing.T) {
+	// The Matrix gateway answers an abandoned topic with a rejected pushkey, not with the 404
+	c := newTestConfig(t, "")
+	c.VisitorSubscriberRateLimiting = true
+	s := newTestServer(t, c)
+	notification := `{"notification":{"devices":[{"pushkey":"http://127.0.0.1:12345/upAAAAAAAAAAAA?up=1"}]}}`
+	response := request(t, s, "POST", "/_matrix/push/v1/notify", notification, nil)
+	require.Equal(t, 507, response.Code)
+	agePayer(s, "upAAAAAAAAAAAA", unifiedPushAbandonedAfter+time.Hour)
+	response = request(t, s, "POST", "/_matrix/push/v1/notify", notification, nil)
+	require.Equal(t, 200, response.Code)
+	require.Equal(t, `{"rejected":["http://127.0.0.1:12345/upAAAAAAAAAAAA?up=1"]}`+"\n", response.Body.String())
+}
+
+func TestServer_UnifiedPush_PayerLivenessIsTheSubscribers(t *testing.T) {
+	// A topic keeps its payer for as long as the subscriber is around, 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)
+	rr := request(t, s, "GET", "/upAAAAAAAAAAAA/json?poll=1", "", nil, func(r *http.Request) {
+		r.RemoteAddr = "1.2.3.4:1234"
+	})
+	require.Equal(t, 200, rr.Code)
+	s.limiter.Prune(0) // The key is gone, the subscriber is not
+	payer, err := s.payerVisitor(lockedTopic(s, "upAAAAAAAAAAAA"))
+	require.Nil(t, err)
+	require.NotNil(t, payer)
+	agePayer(s, "upAAAAAAAAAAAA", 2*visitorExpungeAfter) // The subscriber is gone
+	require.True(t, payer.MessageAllowed())              // A publisher charges the key
+	payer, err = s.payerVisitor(lockedTopic(s, "upAAAAAAAAAAAA"))
+	require.Nil(t, err)
+	require.Nil(t, payer)
+}
+
+func TestServer_UnifiedPush_TierChangeAppliesAfterUserCache(t *testing.T) {
+	// Pushes are billed with the subscriber's current tier, as of the cached user lookup
+	forEachBackend(t, func(t *testing.T, databaseURL string) {
+		c := newTestConfigWithAuthFile(t, databaseURL)
+		c.VisitorSubscriberRateLimiting = true
+		s := newTestServer(t, c)
+		require.Nil(t, s.userManager.AddTier(&user.Tier{
+			Code:         "small",
+			MessageLimit: 1,
+		}))
+		require.Nil(t, s.userManager.AddTier(&user.Tier{
+			Code:         "large",
+			MessageLimit: 10,
+		}))
+		require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleUser, false))
+		require.Nil(t, s.userManager.ChangeTier("phil", "small"))
+		rr := request(t, s, "GET", "/upAAAAAAAAAAAA/json?poll=1", "", map[string]string{
+			"Authorization": util.BasicAuth("phil", "phil"),
+		})
+		require.Equal(t, 200, rr.Code)
+		push := func() int {
+			return request(t, s, "PUT", "/upAAAAAAAAAAAA?up=1", "some message", nil).Code
+		}
+		require.Equal(t, 200, push())
+		require.Equal(t, 429, push())
+
+		// The new tier applies once the cached user expires
+		require.Nil(t, s.userManager.ChangeTier("phil", "large"))
+		require.Equal(t, 429, push())
+		expirePayerUsers(s)
+		require.Equal(t, 200, push())
+	})
+}
+
+func TestServer_UnifiedPush_PayerUserIsCached(t *testing.T) {
+	// Pushes do not read the subscriber's user every time
+	c := newTestConfigWithAuthFile(t, "")
+	c.VisitorSubscriberRateLimiting = true
+	s := newTestServer(t, c)
+	require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleUser, false))
+	rr := request(t, s, "GET", "/upAAAAAAAAAAAA/json?poll=1", "", map[string]string{
+		"Authorization": util.BasicAuth("phil", "phil"),
+	})
+	require.Equal(t, 200, rr.Code)
+	lookups := 0
+	s.payerUsers.lookup = func(id string) (*user.User, error) {
+		lookups++
+		return s.userManager.UserByID(id)
+	}
+	for i := 0; i < 3; i++ {
+		require.Equal(t, 200, request(t, s, "PUT", "/upAAAAAAAAAAAA?up=1", "some message", nil).Code)
+	}
+	require.Equal(t, 1, lookups)
+}
+
+func TestServer_UnifiedPush_DeletedSubscriberIsNoPayer(t *testing.T) {
+	// A deleted subscriber pays for nothing, and its IP does not pay in its place: the push gets
+	// the 507 of a topic without subscriber
+	forEachBackend(t, func(t *testing.T, databaseURL string) {
+		c := newTestConfigWithAuthFile(t, databaseURL)
+		c.VisitorSubscriberRateLimiting = true
+		s := newTestServer(t, c)
+		require.Nil(t, s.userManager.AddUser("alice", "alice", user.RoleUser, false))
+		require.Nil(t, s.userManager.AddUser("bob", "bob", user.RoleUser, false))
+		rr := request(t, s, "GET", "/upAAAAAAAAAAAA/json?poll=1", "", map[string]string{
+			"Authorization": util.BasicAuth("alice", "alice"),
+		})
+		require.Equal(t, 200, rr.Code)
+		rr = request(t, s, "GET", "/upBBBBBBBBBBBB/json?poll=1", "", map[string]string{
+			"Authorization": util.BasicAuth("bob", "bob"),
+		})
+		require.Equal(t, 200, rr.Code)
+
+		// Alice deletes her account (soft delete), bob is removed by an admin (hard delete)
+		alice, err := s.userManager.User("alice")
+		require.Nil(t, err)
+		require.Nil(t, s.userManager.MarkUserRemoved(alice))
+		require.Nil(t, s.userManager.RemoveUser("bob"))
+		for _, topic := range []string{"upAAAAAAAAAAAA", "upBBBBBBBBBBBB"} {
+			response := request(t, s, "PUT", "/"+topic+"?up=1", "some message", nil)
+			require.Equal(t, 507, response.Code, topic)
+			require.Equal(t, 50701, toHTTPError(t, response.Body.String()).Code)
+		}
+	})
+}
+
+func TestServer_UnifiedPush_PayerUserLookupErrorIsRetryable(t *testing.T) {
+	// If the subscriber cannot be looked up, the push fails with a 5xx the publisher retries: it is
+	// not a topic without subscriber, which would make the Matrix gateway reject the pushkey
+	c := newTestConfigWithAuthFile(t, "")
+	c.VisitorSubscriberRateLimiting = true
+	s := newTestServer(t, c)
+	require.Nil(t, s.userManager.AddUser("alice", "alice", user.RoleUser, false))
+	rr := request(t, s, "GET", "/upAAAAAAAAAAAA/json?poll=1", "", map[string]string{
+		"Authorization": util.BasicAuth("alice", "alice"),
+	})
+	require.Equal(t, 200, rr.Code)
+	s.payerUsers.lookup = func(id string) (*user.User, error) {
+		return nil, errors.New("database is down")
+	}
+	response := request(t, s, "PUT", "/upAAAAAAAAAAAA?up=1", "some message", nil)
+	require.Equal(t, 500, response.Code)
+}
+
+// syncWait is how long a payer takes at most to reach another node in healthy operation: one
+// sync round to be written, one to be read
+const syncWait = 10 * time.Second
+
+// payerKey returns the key of the topic's payer, or "" if it has none
+func payerKey(s *Server, topic string) string {
+	if p := s.limiter.Payer(topic); p != nil {
+		return p.Key
+	}
+	return ""
+}
+
+// agePayer makes the topic's payer (or the topic, if it has none) look as if it was last seen d ago
+func agePayer(s *Server, topic string, d time.Duration) {
+	p := s.limiter.Payer(topic)
+	if p == nil {
+		p = &limit.Payer{} // Unknown: make it known, so the caller's assertion fails, not the helper
+	}
+	p.SeenAt = time.Now().Add(-d)
+	s.limiter.SetPayer(topic, p)
+}
+
+// expirePayerUsers makes the cached users of the payers look older than payerUserCacheTTL
+func expirePayerUsers(s *Server) {
+	s.payerUsers.mu.Lock()
+	defer s.payerUsers.mu.Unlock()
+	for _, e := range s.payerUsers.entries {
+		e.fetched = e.fetched.Add(-2 * payerUserCacheTTL)
+	}
+}
+
+// execUserDB runs the query on the config's user database: the auth-file, or the Postgres schema
+func execUserDB(t *testing.T, c *Config, query string) {
+	t.Helper()
+	var d *db.DB
+	if c.DatabaseURL != "" {
+		host, err := pg.Open(c.DatabaseURL)
+		require.Nil(t, err)
+		d = db.New(host, nil)
+	} else {
+		var err error
+		d, err = user.OpenSQLite(c.AuthFile)
+		require.Nil(t, err)
+	}
+	defer d.Close()
+	_, err := d.Exec(query)
+	require.Nil(t, err)
+}

server/server.go

@@ -65,9 +65,10 @@ type Server struct {
 	smtpServerBackend *smtpBackend
 	mailer            mail.Sender
 	topics            map[string]*topic
-	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)
+	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
+	payerUsers        *payerUserCache // Users of the UnifiedPush topics' payers, by ID
+	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)
@@ -360,6 +361,10 @@ func New(conf *Config) (*Server, error) {
 		Store: createUsageStore(pool, userDB),
 	})
 	s.visitors = newVisitorEnv(conf, s.limiter, messageCache, userManager)
+	s.payerUsers = newPayerUserCache(nil)
+	if userManager != nil {
+		s.payerUsers.lookup = userManager.UserByID
+	}
 	s.firebaseClient = firebaseClient
 	if pool != nil {
 		metrics.SetDBPoolsFunc(s.dbPoolStats)
@@ -1099,11 +1104,17 @@ func (s *Server) handlePublishInternal(r *http.Request, v *visitor) (*model.Mess
 	if e != nil {
 		return nil, e.With(t)
 	}
-	if unifiedpush && s.config.VisitorSubscriberRateLimiting && t.RateVisitor() == nil {
+	if unifiedpush && s.config.VisitorSubscriberRateLimiting && vrate == v { // No payer, see limitRequestsWithTopic
 		// UnifiedPush clients must subscribe before publishing to allow proper subscriber-based rate limiting.
 		// The 5xx response is because some app servers (in particular Mastodon) will remove
 		// the subscription as invalid if any 400-499 code (except 429/408) is returned.
 		// See https://github.com/mastodon/mastodon/blob/730bb3e211a84a2f30e3e2bbeae3f77149824a68/app/workers/web/push_notification_worker.rb#L35-L46
+		// That is what a topic abandoned for unifiedPushAbandonedAfter gets: a 404, so the subscription is removed.
+		if s.unifiedPushTopicAbandoned(t) {
+			return nil, errHTTPNotFoundUnifiedPushTopicAbandoned.With(t)
+		} else if isUnifiedPushTopic(t.ID) {
+			s.limiter.NoteTopic(t.ID) // Starts the countdown to the 404 of a topic nobody ever subscribed to
+		}
 		return nil, errHTTPInsufficientStorageUnifiedPush.With(t)
 	} else if !util.ContainsIP(s.config.VisitorRequestExemptPrefixes, v.ip) && !vrate.MessageAllowed() {
 		return nil, errHTTPTooManyRequestsLimitMessages.With(t)
@@ -1216,7 +1227,13 @@ func (s *Server) handlePublishMatrix(w http.ResponseWriter, r *http.Request, v *
 	if err != nil {
 		metrics.MessagesPublishedFailure.Inc()
 		metrics.MatrixPublishedFailure.Inc()
-		if e, ok := err.(*errHTTP); ok && e.HTTPCode == errHTTPInsufficientStorageUnifiedPush.HTTPCode {
+		if e, ok := err.(*errHTTP); ok && e.Code == errHTTPNotFoundUnifiedPushTopicAbandoned.Code {
+			pushKey, err := fromContext[string](r, contextMatrixPushKey)
+			if err != nil {
+				return err
+			}
+			return writeMatrixResponse(w, pushKey)
+		} else if ok && e.HTTPCode == errHTTPInsufficientStorageUnifiedPush.HTTPCode {
 			topic, err := fromContext[*topic](r, contextTopic)
 			if err != nil {
 				return err
@@ -1710,7 +1727,7 @@ func (s *Server) handleSubscribeHTTP(w http.ResponseWriter, r *http.Request, v *
 		}
 		return nil
 	}
-	if err := s.maybeSetRateVisitors(r, v, topics); err != nil {
+	if err := s.maybeSetPayers(r, v, topics); err != nil {
 		return err
 	}
 	w.Header().Set("Access-Control-Allow-Origin", s.config.AccessControlAllowOrigin) // CORS, allow cross-origin requests
@@ -1738,6 +1755,7 @@ func (s *Server) handleSubscribeHTTP(w http.ResponseWriter, r *http.Request, v *
 		for i, subscriberID := range subscriberIDs {
 			topics[i].Unsubscribe(subscriberID) // Order!
 		}
+		s.touchPayers(topics) // The subscriber was last seen now
 	}()
 	if err := s.authorizeSubscribed(v, topics); err != nil {
 		return err
@@ -1874,7 +1892,7 @@ func (s *Server) handleSubscribeWS(w http.ResponseWriter, r *http.Request, v *vi
 		}
 		return conn.WriteJSON(msg)
 	}
-	if err := s.maybeSetRateVisitors(r, v, topics); err != nil {
+	if err := s.maybeSetPayers(r, v, topics); err != nil {
 		return err
 	}
 	w.Header().Set("Access-Control-Allow-Origin", s.config.AccessControlAllowOrigin) // CORS, allow cross-origin requests
@@ -1897,6 +1915,7 @@ func (s *Server) handleSubscribeWS(w http.ResponseWriter, r *http.Request, v *vi
 		for i, subscriberID := range subscriberIDs {
 			topics[i].Unsubscribe(subscriberID) // Order!
 		}
+		s.touchPayers(topics) // The subscriber was last seen now
 	}()
 	if err := s.authorizeSubscribed(v, topics); err != nil {
 		return err
@@ -1932,38 +1951,39 @@ func parseSubscribeParams(r *http.Request) (poll bool, since model.SinceMarker,
 	return
 }
 
-// maybeSetRateVisitors sets the rate visitor on a topic (v.SetRateVisitor), indicating that all messages published
-// to that topic will be rate limited against the rate visitor instead of the publishing visitor.
+// maybeSetPayers makes v the payer of a topic (limit.Service.SetPayer), indicating that all messages published
+// to that topic will be rate limited against the payer instead of the publishing visitor.
 //
-// Setting the rate visitor is ony allowed if the `visitor-subscriber-rate-limiting` setting is enabled, AND
+// Setting the payer is ony allowed if the `visitor-subscriber-rate-limiting` setting is enabled, AND
 // - auth-file is not set (everything is open by default)
 // - or the topic is reserved, and v.user is the owner
 // - or the topic is not reserved, and v.user has write access
 //
 // This only applies to UnifiedPush topics ("up...").
-func (s *Server) maybeSetRateVisitors(r *http.Request, v *visitor, topics []*topic) error {
+func (s *Server) maybeSetPayers(r *http.Request, v *visitor, topics []*topic) error {
 	// Bail out if not enabled
 	if !s.config.VisitorSubscriberRateLimiting {
 		return nil
 	}
 
-	// Make a list of topics that we'll actually set the RateVisitor on
+	// Make a list of topics that we'll actually set the payer on; the others count as subscriber activity only
 	eligibleRateTopics := make([]*topic, 0)
 	for _, t := range topics {
-		if strings.HasPrefix(t.ID, unifiedPushTopicPrefix) && len(t.ID) == unifiedPushTopicLength {
+		if isUnifiedPushTopic(t.ID) {
 			eligibleRateTopics = append(eligibleRateTopics, t)
 		}
 	}
 	if len(eligibleRateTopics) == 0 {
 		return nil
 	}
+	s.touchPayers(eligibleRateTopics)
 
-	// If access controls are turned off, v has access to everything, and we can set the rate visitor
+	// If access controls are turned off, v has access to everything, and we can set the payer
 	if s.userManager == nil {
-		return s.setRateVisitors(r, v, eligibleRateTopics)
+		return s.setPayers(r, v, eligibleRateTopics)
 	}
 
-	// If access controls are enabled, only set rate visitor if
+	// If access controls are enabled, only set payer if
 	// - topic is reserved, and v.user is the owner
 	// - topic is not reserved, and v.user has write access
 	writableRateTopics := make([]*topic, 0)
@@ -1983,16 +2003,21 @@ func (s *Server) maybeSetRateVisitors(r *http.Request, v *visitor, topics []*top
 			writableRateTopics = append(writableRateTopics, t)
 		}
 	}
-	return s.setRateVisitors(r, v, writableRateTopics)
+	return s.setPayers(r, v, writableRateTopics)
 }
 
-func (s *Server) setRateVisitors(r *http.Request, v *visitor, rateTopics []*topic) error {
+func (s *Server) setPayers(r *http.Request, v *visitor, rateTopics []*topic) error {
 	for _, t := range rateTopics {
 		logvr(v, r).
 			Tag(tagSubscribe).
 			With(t).
-			Debug("Setting visitor as rate visitor for topic %s", t.ID)
-		t.SetRateVisitor(v)
+			Debug("Setting visitor as payer for topic %s", t.ID)
+		s.limiter.SetPayer(t.ID, &limit.Payer{
+			Key:    v.key,
+			UserID: v.MaybeUserID(),
+			SeenAt: time.Now(),
+		})
+		t.Keepalive()
 	}
 	return nil
 }

server/server_manager.go

@@ -14,6 +14,7 @@ func (s *Server) execManager() {
 	// database (and the web push job, which also sends expiry warnings) run on the cluster
 	// leader only. Without clustering, IsLeader is always true.
 	s.pruneVisitors()
+	s.payerUsers.Prune()
 	if s.cluster.IsLeader() {
 		s.pruneTokens()
 		s.pruneAttachments()
@@ -48,6 +49,9 @@ func (s *Server) execManager() {
 						ev.Trace("- topic %s: %d subscribers, accessed %s", t.ID, subs, util.FormatTime(lastAccess))
 					}
 					t.ExpireDelivered()
+					if subs > 0 {
+						s.touchPayers([]*topic{t}) // An open stream keeps the payer live on the other nodes too
+					}
 					subscribers += subs
 				}
 			}

server/server_middleware.go

@@ -36,8 +36,11 @@ func (s *Server) limitRequestsWithTopic(next handleFunc) handleFunc {
 			return err
 		}
 		vrate := v
-		if rateVisitor := t.RateVisitor(); rateVisitor != nil {
-			vrate = rateVisitor
+		payer, err := s.payerVisitor(t) // Once per request, so the publish handler sees the same payer
+		if err != nil {
+			return err
+		} else if payer != nil {
+			vrate = payer
 		}
 		r = withContext(r, map[contextKey]any{
 			contextRateVisitor: vrate,

server/server_test.go

@@ -1976,7 +1976,7 @@ func TestServer_MatrixGateway_Push_Failure_NoSubscriber_After13Hours(t *testing.
 		response := request(t, s, "POST", "/_matrix/push/v1/notify", notification, nil)
 		require.Equal(t, 507, response.Code)
 		require.Equal(t, 50701, toHTTPError(t, response.Body.String()).Code)
-		require.Nil(t, s.topics["mytopic"].rateVisitor)
+		require.Nil(t, s.limiter.Payer("mytopic"))
 
 		// Fake: This topic has been around for 13 hours without a rate visitor
 		s.topics["mytopic"].lastAccess = time.Now().Add(-13 * time.Hour)
@@ -3270,7 +3270,7 @@ func TestServer_SubscriberRateLimiting_Success(t *testing.T) {
 		rr := request(t, s, "GET", "/upAAAAAAAAAAAA/json?poll=1", "", nil, subscriber1Fn)
 		require.Equal(t, 200, rr.Code)
 		require.Equal(t, "", rr.Body.String())
-		require.Equal(t, "1.2.3.4", s.topics["upAAAAAAAAAAAA"].rateVisitor.ip.String())
+		require.Equal(t, "ip:1.2.3.4", payerKey(s, "upAAAAAAAAAAAA"))
 
 		// "Register" visitor 8.7.7.1 to topic "up012345678912" as a rate limit visitor (implicitly via topic name)
 		subscriber2Fn := func(r *http.Request) {
@@ -3279,7 +3279,7 @@ func TestServer_SubscriberRateLimiting_Success(t *testing.T) {
 		rr = request(t, s, "GET", "/up012345678912/json?poll=1", "", nil, subscriber2Fn)
 		require.Equal(t, 200, rr.Code)
 		require.Equal(t, "", rr.Body.String())
-		require.Equal(t, "8.7.7.1", s.topics["up012345678912"].rateVisitor.ip.String())
+		require.Equal(t, "ip:8.7.7.1", payerKey(s, "up012345678912"))
 
 		// Publish 2 messages to "subscriber1topic" as visitor 9.9.9.9. It'd be 3 normally, but the
 		// GET request before is also counted towards the request limiter.
@@ -3324,9 +3324,9 @@ func TestServer_SubscriberRateLimiting_NotWrongTopic(t *testing.T) {
 		rr := request(t, s, "GET", "/alerts,upAAAAAAAAAAAA,upBBBBBBBBBBBB/json?poll=1", "", nil, subscriberFn)
 		require.Equal(t, 200, rr.Code)
 		require.Equal(t, "", rr.Body.String())
-		require.Nil(t, s.topics["alerts"].rateVisitor)
-		require.Equal(t, "1.2.3.4", s.topics["upAAAAAAAAAAAA"].rateVisitor.ip.String())
-		require.Equal(t, "1.2.3.4", s.topics["upBBBBBBBBBBBB"].rateVisitor.ip.String())
+		require.Equal(t, "", payerKey(s, "alerts"))
+		require.Equal(t, "ip:1.2.3.4", payerKey(s, "upAAAAAAAAAAAA"))
+		require.Equal(t, "ip:1.2.3.4", payerKey(s, "upBBBBBBBBBBBB"))
 	})
 }
 
@@ -3345,7 +3345,7 @@ func TestServer_SubscriberRateLimiting_NotEnabled_Failed(t *testing.T) {
 		})
 		require.Equal(t, 200, rr.Code)
 		require.Equal(t, "", rr.Body.String())
-		require.Nil(t, s.topics["upAAAAAAAAAAAA"].rateVisitor)
+		require.Equal(t, "", payerKey(s, "upAAAAAAAAAAAA"))
 
 		// Registering visitor 8.7.7.1 to topic has no effect
 		rr = request(t, s, "GET", "/up012345678912/json?poll=1", "", nil, func(r *http.Request) {
@@ -3353,7 +3353,7 @@ func TestServer_SubscriberRateLimiting_NotEnabled_Failed(t *testing.T) {
 		})
 		require.Equal(t, 200, rr.Code)
 		require.Equal(t, "", rr.Body.String())
-		require.Nil(t, s.topics["up012345678912"].rateVisitor)
+		require.Equal(t, "", payerKey(s, "up012345678912"))
 
 		// Publish 3 messages to "upAAAAAAAAAAAA" as visitor 9.9.9.9
 		for i := 0; i < 3; i++ {
@@ -3436,26 +3436,23 @@ 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, "ip:1.2.3.4", s.topics["upAAAAAAAAAAAA"].rateVisitor.key)
+		require.Equal(t, "ip:1.2.3.4", payerKey(s, "upAAAAAAAAAAAA"))
 
 		// 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), publisher.Stats().Messages)
-		require.Equal(t, int64(1), s.topics["upAAAAAAAAAAAA"].rateVisitor.Stats().Messages)
+		require.Equal(t, int64(1), s.visitor(netip.MustParseAddr("1.2.3.4"), nil).Stats().Messages)
 
 		// Expire visitor: its subscriber has been gone for a day, and its key is pruned after a day without use
-		ageRateVisitor(s.topics["upAAAAAAAAAAAA"], 25*time.Hour)
+		agePayer(s, "upAAAAAAAAAAAA", 25*time.Hour)
 		s.limiter.Prune(-time.Hour) // Everything is idle for longer than -1h
-		require.True(t, s.topics["upAAAAAAAAAAAA"].rateVisitorStaleNoLock())
 
-		// Publish message again, observe that rateVisitor is not used anymore and is reset
+		// Publish message again, observe that the payer is not used anymore
 		response = request(t, s, "POST", "/upAAAAAAAAAAAA", "some message", nil)
 		require.Equal(t, 200, response.Code)
 		require.Equal(t, int64(1), publisher.Stats().Messages)
-		require.Nil(t, s.topics["upAAAAAAAAAAAA"].rateVisitor)
 		require.Equal(t, int64(1), s.visitor(netip.MustParseAddr("1.2.3.4"), nil).Stats().Messages) // Daily counts outlive pruning until their period ends
 	})
 }
@@ -3486,7 +3483,7 @@ func TestServer_SubscriberRateLimiting_PollKeepsStreamAsRateVisitor(t *testing.T
 	require.Equal(t, 200, rr.Code)
 
 	// A day later, the stream is still there and keeps the topic's rate visitor
-	ageRateVisitor(lockedTopic(s, "upAAAAAAAAAAAA"), 25*time.Hour)
+	agePayer(s, "upAAAAAAAAAAAA", 25*time.Hour)
 	response := request(t, s, "PUT", "/upAAAAAAAAAAAA?up=1", "some message", nil)
 	require.Equal(t, 200, response.Code)
 }
@@ -3503,14 +3500,10 @@ func TestServer_SubscriberRateLimiting_PollKeepsPollingSubscriberAlive(t *testin
 	require.Equal(t, 200, rr.Code)
 
 	// Polled 23 hours ago; polling again keeps it the rate visitor for another day
-	to := lockedTopic(s, "upAAAAAAAAAAAA")
-	ageRateVisitor(to, 23*time.Hour)
+	agePayer(s, "upAAAAAAAAAAAA", 23*time.Hour)
 	rr = request(t, s, "GET", "/upAAAAAAAAAAAA/json?poll=1", "", nil, subscriberFn)
 	require.Equal(t, 200, rr.Code)
-	to.mu.Lock()
-	seen := to.rateSeen
-	to.mu.Unlock()
-	require.WithinDuration(t, time.Now(), seen, time.Minute)
+	require.WithinDuration(t, time.Now(), s.limiter.Payer("upAAAAAAAAAAAA").SeenAt, time.Minute)
 }
 
 func TestServer_SubscriberRateLimiting_OverlappingStreamsKeepTopicAlive(t *testing.T) {
@@ -3550,7 +3543,7 @@ func TestServer_SubscriberRateLimiting_OverlappingStreamsKeepTopicAlive(t *testi
 	<-doneB
 
 	// A day later, stream A is still there
-	ageRateVisitor(lockedTopic(s, "upAAAAAAAAAAAA"), 25*time.Hour)
+	agePayer(s, "upAAAAAAAAAAAA", 25*time.Hour)
 	response := request(t, s, "PUT", "/upAAAAAAAAAAAA?up=1", "some message", nil)
 	require.Equal(t, 200, response.Code)
 }
@@ -3579,7 +3572,7 @@ func TestServer_SubscriberRateLimiting_OpenStreamKeepsPollerAsRateVisitor(t *tes
 	require.Equal(t, 200, rr.Code)
 
 	// A day later, only the stream is still there; the publish is billed to the poller's key
-	ageRateVisitor(lockedTopic(s, "upAAAAAAAAAAAA"), 25*time.Hour)
+	agePayer(s, "upAAAAAAAAAAAA", 25*time.Hour)
 	response := request(t, s, "PUT", "/upAAAAAAAAAAAA?up=1", "some message", nil)
 	require.Equal(t, 200, response.Code)
 	require.Equal(t, int64(1), s.visitor(netip.MustParseAddr("1.2.3.4"), nil).Stats().Messages)
@@ -3618,8 +3611,9 @@ func TestServer_SubscriberRateLimiting_StreamDoesNotReclaimTopicAfterWriteRevoke
 		to := lockedTopic(s, "upAAAAAAAAAAAA")
 		return to != nil && to.SubscribersCount() == 1
 	})
-	to := lockedTopic(s, "upAAAAAAAAAAAA")
-	require.True(t, to.RateVisitor().MessageAllowed())
+	payer, err := s.payerVisitor(lockedTopic(s, "upAAAAAAAAAAAA"))
+	require.Nil(t, err)
+	require.True(t, payer.MessageAllowed())
 	require.Nil(t, s.userManager.AllowAccess("alice", "upAAAAAAAAAAAA", user.PermissionRead))
 
 	// Bob takes the topic with a poll, and is gone a day later
@@ -3627,7 +3621,7 @@ func TestServer_SubscriberRateLimiting_StreamDoesNotReclaimTopicAfterWriteRevoke
 		r.RemoteAddr = "2.2.2.2:1234"
 	})
 	require.Equal(t, 200, response.Code)
-	ageRateVisitor(to, 25*time.Hour)
+	agePayer(s, "upAAAAAAAAAAAA", 25*time.Hour)
 
 	// Alice's open stream keeps the topic alive, but it is still billed to Bob
 	response = request(t, s, "PUT", "/upAAAAAAAAAAAA", "some message", bobHeaders, func(r *http.Request) {
@@ -3655,7 +3649,7 @@ func TestServer_SubscriberRateLimiting_OpenStreamKeepsTopicAliveWithoutKeepalive
 		to := lockedTopic(s, "upAAAAAAAAAAAA")
 		return to != nil && to.SubscribersCount() == 1
 	})
-	ageRateVisitor(lockedTopic(s, "upAAAAAAAAAAAA"), 25*time.Hour)
+	agePayer(s, "upAAAAAAAAAAAA", 25*time.Hour)
 	response := request(t, s, "PUT", "/upAAAAAAAAAAAA?up=1", "some message", nil)
 	require.Equal(t, 200, response.Code)
 	require.Equal(t, int64(1), s.visitor(netip.MustParseAddr("1.2.3.4"), nil).Stats().Messages)
@@ -3681,26 +3675,18 @@ func TestServer_SubscriberRateLimiting_ClosedStreamIsLastSeen(t *testing.T) {
 		to := lockedTopic(s, "upAAAAAAAAAAAA")
 		return to != nil && to.SubscribersCount() == 1
 	})
-	to := lockedTopic(s, "upAAAAAAAAAAAA")
-	ageRateVisitor(to, 25*time.Hour) // Subscribed more than a day ago
+	agePayer(s, "upAAAAAAAAAAAA", 25*time.Hour) // Subscribed more than a day ago
 	cancel()
 	<-done
 	response := request(t, s, "PUT", "/upAAAAAAAAAAAA?up=1", "some message", nil)
 	require.Equal(t, 200, response.Code)
 
 	// A day after the stream closed, the subscriber is gone
-	ageRateVisitor(to, 25*time.Hour)
+	agePayer(s, "upAAAAAAAAAAAA", 25*time.Hour)
 	response = request(t, s, "PUT", "/upAAAAAAAAAAAA?up=1", "some message", nil)
 	require.Equal(t, 507, response.Code)
 }
 
-// ageRateVisitor makes the topic's rate visitor look as if it was last seen d ago
-func ageRateVisitor(t *topic, d time.Duration) {
-	t.mu.Lock()
-	defer t.mu.Unlock()
-	t.rateSeen = time.Now().Add(-d)
-}
-
 // lockedTopic returns the topic with the given ID, or nil, reading the topics map under the server's lock
 func lockedTopic(s *Server, id string) *topic {
 	s.mu.RLock()
@@ -3725,8 +3711,8 @@ func TestServer_SubscriberRateLimiting_ProtectedTopics_WithDefaultReadWrite(t *t
 			r.RemoteAddr = "1.2.3.4:1234"
 		})
 		require.Equal(t, 200, rr.Code)
-		require.Equal(t, "1.2.3.4", s.topics["up123456789012"].rateVisitor.ip.String())
-		require.Nil(t, s.topics["announcements"].rateVisitor)
+		require.Equal(t, "ip:1.2.3.4", payerKey(s, "up123456789012"))
+		require.Equal(t, "", payerKey(s, "announcements"))
 	})
 }
 

server/topic.go

@@ -29,8 +29,6 @@ const (
 type topic struct {
 	ID          string
 	subscribers map[int]*topicSubscriber
-	rateVisitor *visitor
-	rateSeen    time.Time // When the rate visitor subscribed, or a stream on the topic last closed
 	lastAccess  time.Time
 	delivered   map[string]time.Time // When recent messages went out to the subscribers, by message ID
 	mu          sync.RWMutex
@@ -110,9 +108,6 @@ func (t *topic) ExpireDelivered() {
 func (t *topic) Stale() bool {
 	t.mu.Lock()
 	defer t.mu.Unlock()
-	if !t.rateVisitorStaleNoLock() {
-		return false
-	}
 	return len(t.subscribers) == 0 && time.Since(t.lastAccess) > topicExpungeAfter
 }
 
@@ -122,41 +117,11 @@ func (t *topic) LastAccess() time.Time {
 	return t.lastAccess
 }
 
-// SetRateVisitor makes the subscriber's visitor the topic's rate visitor; the latest subscriber wins
-func (t *topic) SetRateVisitor(v *visitor) {
-	t.mu.Lock()
-	defer t.mu.Unlock()
-	t.rateVisitor = v
-	t.rateSeen = time.Now()
-	t.lastAccess = time.Now()
-}
-
-// RateVisitor returns the topic's rate visitor, or nil if there is none or it is stale
-func (t *topic) RateVisitor() *visitor {
-	t.mu.Lock()
-	defer t.mu.Unlock()
-	if t.rateVisitorStaleNoLock() {
-		t.rateVisitor = nil
-	}
-	return t.rateVisitor
-}
-
-// rateVisitorStaleNoLock is true if there is no rate visitor, or the topic has no open stream and
-// none closed or polled within visitorExpungeAfter. Any open stream keeps the rate visitor alive,
-// which is the subscriber's own, or whoever subscribed last.
-func (t *topic) rateVisitorStaleNoLock() bool {
-	if t.rateVisitor == nil {
-		return true
-	}
-	return len(t.subscribers) == 0 && time.Since(t.rateSeen) > visitorExpungeAfter
-}
-
 // Unsubscribe removes the subscription from the list of subscribers
 func (t *topic) Unsubscribe(id int) {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 	delete(t.subscribers, id)
-	t.rateSeen = time.Now() // The subscriber was last seen now; the rate visitor stays for another visitorExpungeAfter
 }
 
 // Publish sends a live message to the subscribers that have not had it (see deliver), without
@@ -326,11 +291,6 @@ func (t *topic) Context() log.Context {
 		"topic_subscribers": len(t.subscribers),
 		"topic_last_access": util.FormatTime(t.lastAccess),
 	}
-	if t.rateVisitor != nil {
-		for k, v := range t.rateVisitor.Context() {
-			fields["topic_rate_"+k] = v
-		}
-	}
 	return fields
 }
 

server/visitor.go

@@ -17,9 +17,9 @@ const (
 	oneDay = 24 * time.Hour
 
 	// 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).
+	// forgotten (see Server.pruneVisitors), and how long a topic keeps a payer whose subscriber is
+	// gone (see Server.payerVisitor). 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
 
 	// promoteRate is the divisor of the daily message cap (cap / promoteRate, i.e. 10%) from which
@@ -93,21 +93,35 @@ func newVisitorEnv(conf *Config, limiter *limit.Service, messageCache *message.C
 }
 
 func newVisitor(env *visitorEnv, ip netip.Addr, u *user.User) *visitor {
-	limits := env.ipLimits
-	var tier *user.Tier
+	v := newVisitorWithKey(env, vis.Key(ip, u, env.limitsConfig), u)
+	v.ip = ip
+	env.limiter.Touch(v.key) // The key is seen now, as the visitor's request is
+	return v
+}
+
+// newPayerVisitor returns the visitor of a UnifiedPush topic's payer (see Server.payerVisitor):
+// the subscriber's key, with its user's current limits. A user with a tier now is keyed by the
+// user, as vis.Key would, even if it had none when it subscribed. Its IP is unknown.
+func newPayerVisitor(env *visitorEnv, key string, u *user.User) *visitor {
 	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
+		key = vis.Key(netip.Addr{}, u, env.limitsConfig)
 	}
-	v := &visitor{
+	return newVisitorWithKey(env, key, u)
+}
+
+// newVisitorWithKey returns a visitor with the given key, and the limits of the user's tier, or
+// the config's without one
+func newVisitorWithKey(env *visitorEnv, key string, u *user.User) *visitor {
+	limits := env.ipLimits
+	if u != nil && u.Tier != nil {
+		limits = vis.LimitsFor(env.limitsConfig, u.Tier) // Resolved per request: the tier's fields may change at any time
+	}
+	return &visitor{
 		env:    env,
-		key:    vis.Key(ip, u, env.limitsConfig),
-		ip:     ip,
+		key:    key,
 		user:   u,
 		limits: limits,
 	}
-	env.limiter.Touch(v.key) // The key is seen now, as the visitor's request is
-	return v
 }
 
 func (v *visitor) Context() log.Context {

server/visitor_test.go

@@ -2,7 +2,6 @@ package server
 
 import (
 	"io"
-	"net/http/httptest"
 	"net/netip"
 	"strings"
 	"sync"
@@ -313,22 +312,6 @@ func TestVisitor_AccountLimiter_NotDeniedByConcurrentLogin(t *testing.T) {
 	require.True(t, vip.AccountActionAllowed())
 }
 
-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 rate 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)
-	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
@@ -636,24 +619,6 @@ func TestVisitor_TierWithoutEmailsAdmitsNone(t *testing.T) {
 	require.Equal(t, 42902, toHTTPError(t, response.Body.String()).Code)
 }
 
-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)
-	v := s.visitor(netip.MustParseAddr("9.9.9.9"), nil)
-	topics, err := s.topicsFromIDs(v, "mytopic")
-	require.Nil(t, err)
-	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())
-	ageRateVisitor(topics[0], 2*visitorExpungeAfter)               // 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())
-}
-
 func TestVisitor_EmailsWithoutTierHaveDailyCap(t *testing.T) {
 	// Without a tier, the daily limit derived from the refill (24 with a refill of an hour) is a
 	// cap, as for tiers: a burst of 30 is cut at 24