Keep the payers of UnifiedPush topics in the limit service, synced through the topic_payer table

limit/payer.go

@@ -0,0 +1,169 @@
+package limit
+
+import (
+	"context"
+	"time"
+
+	"heckel.io/ntfy/v2/log"
+)
+
+const (
+	// payerTouchInterval is how often a touch of a topic's payer is written at most; an open
+	// stream is touched about this often, so other nodes may see it up to this much older
+	payerTouchInterval = 5 * time.Minute
+
+	// payerPruneAfter is how long a payer row is kept after it was last written. It must be well
+	// above the server's 404 threshold for an abandoned UnifiedPush topic (7 days), so the row
+	// that starts the countdown is still there when the threshold is reached.
+	payerPruneAfter = 30 * oneDay
+)
+
+// SetPayer makes p the topic's payer; the latest subscriber wins. p.SeenAt is the node's clock.
+func (s *Service) SetPayer(topic string, p *Payer) {
+	s.payers.mu.Lock()
+	defer s.payers.mu.Unlock()
+	s.payers.entries[topic] = &payerEntry{
+		key:    p.Key,
+		userID: p.UserID,
+		seenAt: p.SeenAt.Unix(),
+		write:  payerWriteSet,
+	}
+}
+
+// TouchPayer marks subscriber activity on the topic now (an open or closed stream), keeping its
+// payer. It does nothing if the topic was seen within payerTouchInterval.
+func (s *Service) TouchPayer(topic string) {
+	now := s.now().Unix()
+	s.payers.mu.Lock()
+	defer s.payers.mu.Unlock()
+	e, ok := s.payers.entries[topic]
+	if !ok {
+		s.payers.addNoLock(topic, now)
+		return
+	} else if now-e.seenAt < int64(payerTouchInterval/time.Second) {
+		return
+	}
+	e.seenAt = now
+	e.write = max(e.write, payerWriteTouch)
+}
+
+// NoteTopic remembers a topic this node does not know yet, as seen now and without a payer, e.g.
+// on a refused push, so that the topic's abandonment can be told apart from a new topic. It is
+// written as a touch: a row this node has not read yet (written in the last seconds, or while its
+// load failed) keeps its payer, and its seen_at moves to now at most.
+func (s *Service) NoteTopic(topic string) {
+	now := s.now().Unix()
+	s.payers.mu.Lock()
+	defer s.payers.mu.Unlock()
+	if _, ok := s.payers.entries[topic]; !ok {
+		s.payers.addNoLock(topic, now)
+	}
+}
+
+// Payer returns the topic's payer, or nil if the topic is unknown
+func (s *Service) Payer(topic string) *Payer {
+	s.payers.mu.Lock()
+	defer s.payers.mu.Unlock()
+	e, ok := s.payers.entries[topic]
+	if !ok {
+		return nil
+	}
+	return &Payer{
+		Key:    e.key,
+		UserID: e.userID,
+		SeenAt: time.Unix(e.seenAt, 0).UTC(),
+	}
+}
+
+// writePayers writes the payers changed here. A failed write marks them changed again, for the
+// next round; rewriting the ones that were written is harmless.
+func (s *Service) writePayers(ctx context.Context) {
+	rows := s.unwrittenPayers()
+	if len(rows) == 0 {
+		return
+	}
+	if err := s.store.writePayers(ctx, rows, s.now().Unix()); err != nil {
+		s.markPayersUnwritten(rows)
+		log.Tag(tag).Err(err).Warn("Cannot write %d payer(s); retrying in the next round", len(rows))
+	}
+}
+
+// unwrittenPayers returns the rows of the payers changed here, and marks them written
+func (s *Service) unwrittenPayers() []payerRow {
+	s.payers.mu.Lock()
+	defer s.payers.mu.Unlock()
+	rows := make([]payerRow, 0)
+	for topic, e := range s.payers.entries {
+		if e.write == payerWriteNone {
+			continue
+		}
+		r := payerRow{
+			topic:  topic,
+			seenAt: e.seenAt,
+		}
+		if e.write == payerWriteSet {
+			r.key, r.userID = e.key, e.userID
+		}
+		rows = append(rows, r)
+		e.write = payerWriteNone
+	}
+	return rows
+}
+
+// markPayersUnwritten marks the payers of rows that failed to write as changed again
+func (s *Service) markPayersUnwritten(rows []payerRow) {
+	s.payers.mu.Lock()
+	defer s.payers.mu.Unlock()
+	for _, r := range rows {
+		e, ok := s.payers.entries[r.topic]
+		if !ok {
+			continue
+		}
+		write := payerWriteTouch
+		if r.key != "" {
+			write = payerWriteSet
+		}
+		e.write = max(e.write, write)
+	}
+}
+
+// readPayers applies the payer rows written since the given time, except to entries with a change
+// not written yet (that change is newer). It returns how many rows it applied.
+func (s *Service) readPayers(ctx context.Context, since int64) (int, error) {
+	applied := 0
+	err := s.store.readPayersSince(ctx, since, func(r *payerRow) {
+		s.payers.mu.Lock()
+		defer s.payers.mu.Unlock()
+		e, ok := s.payers.entries[r.topic]
+		if ok && e.write != payerWriteNone {
+			return
+		} else if !ok {
+			e = &payerEntry{}
+			s.payers.entries[r.topic] = e
+		}
+		e.key, e.userID, e.seenAt = r.key, r.userID, r.seenAt
+		applied++
+	})
+	return applied, err
+}
+
+// prunePayers forgets the payers not seen for payerPruneAfter
+func (s *Service) prunePayers(now time.Time) {
+	cutoff := now.Add(-payerPruneAfter).Unix()
+	s.payers.mu.Lock()
+	defer s.payers.mu.Unlock()
+	for topic, e := range s.payers.entries {
+		if e.seenAt < cutoff {
+			delete(s.payers.entries, topic)
+		}
+	}
+}
+
+// addNoLock adds the topic as seen at now without a payer, to be written as a touch. The caller
+// must hold pt.mu.
+func (pt *payerTable) addNoLock(topic string, now int64) {
+	pt.entries[topic] = &payerEntry{
+		seenAt: now,
+		write:  payerWriteTouch,
+	}
+}

limit/payer_test.go

@@ -0,0 +1,183 @@
+package limit
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestService_Payer_SetTouchNote(t *testing.T) {
+	s, clock := newTestServiceWithClock(t)
+	require.Nil(t, s.Payer("upAAAAAAAAAAAA"))
+
+	// A refused push makes the topic known, without a payer; a second one does not move its time
+	s.NoteTopic("upAAAAAAAAAAAA")
+	noted := clock.now()
+	clock.advance(time.Hour)
+	s.NoteTopic("upAAAAAAAAAAAA")
+	require.Equal(t, &Payer{
+		SeenAt: noted,
+	}, s.Payer("upAAAAAAAAAAAA"))
+
+	// A subscriber becomes the payer; the latest one wins
+	s.SetPayer("upAAAAAAAAAAAA", &Payer{
+		Key:    "ip:1.2.3.4",
+		SeenAt: clock.now(),
+	})
+	s.SetPayer("upAAAAAAAAAAAA", &Payer{
+		Key:    "user:u_1",
+		UserID: "u_1",
+		SeenAt: clock.now(),
+	})
+	set := clock.now()
+	require.Equal(t, userPayer(set), s.Payer("upAAAAAAAAAAAA"))
+
+	// A touch keeps the payer, and moves its time only once payerTouchInterval has passed
+	clock.advance(payerTouchInterval - time.Second)
+	s.TouchPayer("upAAAAAAAAAAAA")
+	require.Equal(t, userPayer(set), s.Payer("upAAAAAAAAAAAA"))
+	clock.advance(time.Second)
+	s.TouchPayer("upAAAAAAAAAAAA")
+	require.Equal(t, userPayer(clock.now()), s.Payer("upAAAAAAAAAAAA"))
+
+	// A touch of an unknown topic makes it known, without a payer
+	s.TouchPayer("upBBBBBBBBBBBB")
+	require.Equal(t, &Payer{
+		SeenAt: clock.now(),
+	}, s.Payer("upBBBBBBBBBBBB"))
+}
+
+func TestService_Payer_Prune(t *testing.T) {
+	s, clock := newTestServiceWithClock(t)
+	s.NoteTopic("upAAAAAAAAAAAA")
+	clock.advance(time.Hour)
+	s.NoteTopic("upBBBBBBBBBBBB")
+	clock.advance(payerPruneAfter - 30*time.Minute)
+	s.Prune(time.Hour)
+	require.Nil(t, s.Payer("upAAAAAAAAAAAA"))
+	require.NotNil(t, s.Payer("upBBBBBBBBBBBB"))
+}
+
+func TestService_Persist_PayerSharedAcrossNodes(t *testing.T) {
+	// A payer set on one node is the payer on the other after a sync; a touch on the other node
+	// moves its time for both, and keeps the payer
+	forEachBackend(t, func(t *testing.T, newStore func() *Store) {
+		a := startTestService(t, newStore(), time.Now)
+		b := startTestService(t, newStore(), time.Now)
+		aged := time.Unix(time.Now().Add(-25*time.Hour).Unix(), 0).UTC()
+		a.SetPayer("upAAAAAAAAAAAA", &Payer{
+			Key:    "user:u_1",
+			UserID: "u_1",
+			SeenAt: aged,
+		})
+		a.sync()
+		b.sync()
+		require.Equal(t, userPayer(aged), b.Payer("upAAAAAAAAAAAA"))
+		b.TouchPayer("upAAAAAAAAAAAA")
+		require.NotNil(t, b.Payer("upAAAAAAAAAAAA"))
+		touched := b.Payer("upAAAAAAAAAAAA").SeenAt
+		require.True(t, touched.After(aged))
+		b.sync()
+		a.sync()
+		require.Equal(t, userPayer(touched), a.Payer("upAAAAAAAAAAAA"))
+	})
+}
+
+func TestService_Persist_PayerSurvivesRestart(t *testing.T) {
+	forEachBackend(t, func(t *testing.T, newStore func() *Store) {
+		s := startTestService(t, newStore(), time.Now)
+		seen := time.Unix(time.Now().Unix(), 0).UTC()
+		s.SetPayer("upAAAAAAAAAAAA", &Payer{
+			Key:    "ip:1.2.3.4",
+			SeenAt: seen,
+		})
+		s.Close() // Writes what is unwritten
+		s = startTestService(t, newStore(), time.Now)
+		require.Equal(t, ipPayer(seen), s.Payer("upAAAAAAAAAAAA"))
+	})
+}
+
+func TestService_Persist_NoteKeepsStoredPayer(t *testing.T) {
+	// A refused push on a node that has not read the payer yet must not wipe it
+	forEachBackend(t, func(t *testing.T, newStore func() *Store) {
+		a := startTestService(t, newStore(), time.Now)
+		b := startTestService(t, newStore(), time.Now)
+		a.SetPayer("upAAAAAAAAAAAA", &Payer{
+			Key:    "ip:1.2.3.4",
+			SeenAt: time.Now(),
+		})
+		a.sync()
+		b.NoteTopic("upAAAAAAAAAAAA")
+		b.sync()
+		a.sync()
+		for _, s := range []*Service{a, b} {
+			require.NotNil(t, s.Payer("upAAAAAAAAAAAA"))
+			require.Equal(t, "ip:1.2.3.4", s.Payer("upAAAAAAAAAAAA").Key)
+		}
+	})
+}
+
+func TestService_Persist_ReadBackKeepsUnwrittenPayer(t *testing.T) {
+	// A payer set here and not written yet is not overwritten by an older row read back
+	forEachBackend(t, func(t *testing.T, newStore func() *Store) {
+		a := startTestService(t, newStore(), time.Now)
+		b := startTestService(t, newStore(), time.Now)
+		a.SetPayer("upAAAAAAAAAAAA", &Payer{
+			Key:    "ip:1.2.3.4",
+			SeenAt: time.Now(),
+		})
+		a.sync()
+		b.SetPayer("upAAAAAAAAAAAA", &Payer{
+			Key:    "ip:5.6.7.8",
+			SeenAt: time.Now(),
+		})
+		_, err := b.readPayers(context.Background(), 0)
+		require.Nil(t, err)
+		require.NotNil(t, b.Payer("upAAAAAAAAAAAA"))
+		require.Equal(t, "ip:5.6.7.8", b.Payer("upAAAAAAAAAAAA").Key)
+		b.sync()
+		a.sync()
+		require.NotNil(t, a.Payer("upAAAAAAAAAAAA"))
+		require.Equal(t, "ip:5.6.7.8", a.Payer("upAAAAAAAAAAAA").Key)
+	})
+}
+
+func TestService_Persist_PayerWriteFailureRetries(t *testing.T) {
+	forEachBackend(t, func(t *testing.T, newStore func() *Store) {
+		store := newStore()
+		s := startTestService(t, store, time.Now)
+		s.SetPayer("upAAAAAAAAAAAA", &Payer{
+			Key:    "ip:1.2.3.4",
+			SeenAt: time.Now(),
+		})
+		writePayerChunk := store.writePayerChunk
+		store.writePayerChunk = func(ctx context.Context, rows []payerRow, now int64) error {
+			return errors.New("database is down")
+		}
+		s.sync()
+		require.Empty(t, readAllPayers(t, store, 0))
+		store.writePayerChunk = writePayerChunk
+		s.sync()
+		require.Equal(t, "ip:1.2.3.4", readAllPayers(t, store, 0)["upAAAAAAAAAAAA"].key)
+	})
+}
+
+// userPayer returns the payer of user u_1, seen at the given time
+func userPayer(seen time.Time) *Payer {
+	return &Payer{
+		Key:    "user:u_1",
+		UserID: "u_1",
+		SeenAt: seen,
+	}
+}
+
+// ipPayer returns the payer of the anonymous visitor 1.2.3.4, seen at the given time
+func ipPayer(seen time.Time) *Payer {
+	return &Payer{
+		Key:    "ip:1.2.3.4",
+		SeenAt: seen,
+	}
+}

limit/service.go

@@ -15,6 +15,9 @@
 // current rows. Store failures never reach the request path: counts stay unflushed and are
 // written once the store is back (fail open, per-node limits meanwhile).
 //
+// The service also keeps the payers of the UnifiedPush topics (see SetPayer), synced through the
+// topic_payer table by the same worker, so a push on any node is billed to the topic's subscriber.
+//
 // Periods, updated_at and the read-back watermark use the node's clock, so the nodes must
 // keep their clocks in sync (NTP); a skew below a second or so is harmless, a skew beyond
 // readOverlap may skip rows in the read-back. The database's clock is never used.
@@ -64,6 +67,7 @@ type Service struct {
 	now       func() time.Time          // Injectable clock, for tests
 	counters  [shardCount]*counterShard // Daily counts, kept until their period ends
 	buckets   [shardCount]*bucketShard  // Token buckets, per node, evicted when idle and full
+	payers    *payerTable               // Payers of the UnifiedPush topics, kept for payerPruneAfter
 	store     *Store                    // Nil: memory only
 	watermark int64                     // Time of the last read, unix seconds; the sync worker's only
 	prunedAt  int64                     // Time of the last prune, unix seconds; the sync worker's only
@@ -85,6 +89,9 @@ func newService(conf *Config, now func() time.Time) *Service {
 		now:      now,
 		store:    conf.Store,
 		interval: syncInterval,
+		payers: &payerTable{
+			entries: make(map[string]*payerEntry),
+		},
 	}
 	s.ctx, s.cancel = context.WithCancel(context.Background())
 	for i := 0; i < shardCount; i++ {
@@ -165,6 +172,7 @@ func (s *Service) Close() {
 		ctx, cancel := context.WithTimeout(context.Background(), statementTimeout)
 		defer cancel()
 		s.write(ctx)
+		s.writePayers(ctx)
 	})
 }
 
@@ -263,6 +271,7 @@ func (s *Service) Prune(idle time.Duration) int {
 		}
 		cs.mu.Unlock()
 	}
+	s.prunePayers(now)
 	return pruned
 }
 
@@ -344,24 +353,38 @@ func (s *Service) load() bool {
 		log.Tag(tag).Err(err).Warn("Cannot load usage; retrying in %v", s.interval)
 		return false
 	}
+	payers, err := s.readPayers(ctx, now-int64(payerPruneAfter/time.Second))
+	if err != nil {
+		log.Tag(tag).Err(err).Warn("Cannot load payers; retrying in %v", s.interval)
+		return false
+	}
 	s.watermark, s.prunedAt = now, now
-	log.Tag(tag).Info("Loaded usage of %d key(s) in %v", n, time.Since(started).Round(time.Millisecond))
+	log.Tag(tag).Info("Loaded usage of %d key(s) and %d payer(s) in %v", n, payers, time.Since(started).Round(time.Millisecond))
 	return true
 }
 
-// sync is one round of the sync worker: write the unflushed counts of the promoted keys, read
-// back the rows changed since the last round, and prune once in a while. Every error is logged
-// and leaves the state as it was, for the next round.
+// sync is one round of the sync worker: write the unflushed counts of the promoted keys and the
+// changed payers, read back the rows changed since the last round, and prune once in a while.
+// Every error is logged and leaves the state as it was, for the next round.
 func (s *Service) sync() {
 	now := s.now().Unix()
 	ctx, cancel := context.WithTimeout(s.ctx, statementTimeout)
 	s.write(ctx)
 	cancel()
 	ctx, cancel = context.WithTimeout(s.ctx, statementTimeout)
-	_, err := s.read(ctx, s.watermark-int64(readOverlap/time.Second))
+	s.writePayers(ctx)
 	cancel()
+	since := s.watermark - int64(readOverlap/time.Second)
+	ctx, cancel = context.WithTimeout(s.ctx, statementTimeout)
+	_, err := s.read(ctx, since)
+	cancel()
+	if err == nil {
+		ctx, cancel = context.WithTimeout(s.ctx, statementTimeout)
+		_, err = s.readPayers(ctx, since)
+		cancel()
+	}
 	if err != nil {
-		log.Tag(tag).Err(err).Warn("Cannot read usage; counts of other nodes are stale until the next round")
+		log.Tag(tag).Err(err).Warn("Cannot read usage or payers; those of other nodes are stale until the next round")
 	} else {
 		s.watermark = now
 	}
@@ -375,6 +398,14 @@ func (s *Service) sync() {
 		} else {
 			log.Tag(tag).Debug("Pruned %d usage row(s)", deleted)
 		}
+		ctx, cancel = context.WithTimeout(s.ctx, statementTimeout)
+		deleted, err = s.store.prunePayers(ctx, now-int64(payerPruneAfter/time.Second))
+		cancel()
+		if err != nil {
+			log.Tag(tag).Err(err).Warn("Cannot prune payers")
+		} else {
+			log.Tag(tag).Debug("Pruned %d payer row(s)", deleted)
+		}
 	}
 }
 

limit/store.go

@@ -32,20 +32,33 @@ type usageRow struct {
 	counts      [numDaily]int64
 }
 
+// payerRow is one topic's row of the topic_payer table. On the way to the store, an empty key
+// keeps the stored payer and only moves seen_at forward (a touch).
+type payerRow struct {
+	topic  string
+	key    string
+	userID string
+	seenAt int64 // Unix seconds
+}
+
 // Store reads and writes the usage table, on SQLite (the user database) or Postgres. Rows are
 // stamped (updated_at) with the writing node's clock, like the reader's watermark, so the
 // database's clock does not matter. Only the service's sync worker calls it.
 type Store struct {
-	db         *db.DB
-	queries    *queries
-	writeChunk func(ctx context.Context, rows []usageRow, now int64) error // Dialect specific: unnest on Postgres, a prepared statement per row on SQLite
+	db              *db.DB
+	queries         *queries
+	writeChunk      func(ctx context.Context, rows []usageRow, now int64) error // Dialect specific: unnest on Postgres, a prepared statement per row on SQLite
+	writePayerChunk func(ctx context.Context, rows []payerRow, now int64) error // Dialect specific, as writeChunk
 }
 
 // queries are the dialect's SQL statements
 type queries struct {
-	selectSince string
-	upsert      string
-	prune       string
+	selectSince       string
+	upsert            string
+	prune             string
+	selectPayersSince string
+	upsertPayer       string
+	prunePayers       string
 }
 
 // write adds the rows' counts to their keys' rows if the periods match, or replaces them if the
@@ -93,9 +106,15 @@ func (s *Store) readSince(ctx context.Context, since int64, fn func(r *usageRow)
 // prune deletes the rows not written since the given time (unix seconds), in chunks, and returns how
 // many it deleted
 func (s *Store) prune(ctx context.Context, before int64) (int64, error) {
+	return s.pruneChunked(ctx, s.queries.prune, before)
+}
+
+// pruneChunked runs the prune query (age, chunk size) until a chunk deletes fewer rows than the
+// chunk size, and returns how many rows it deleted
+func (s *Store) pruneChunked(ctx context.Context, query string, before int64) (int64, error) {
 	var deleted int64
 	for {
-		result, err := s.db.ExecContext(ctx, s.queries.prune, before, pruneChunkRows)
+		result, err := s.db.ExecContext(ctx, query, before, pruneChunkRows)
 		if err != nil {
 			return deleted, err
 		}
@@ -109,3 +128,42 @@ func (s *Store) prune(ctx context.Context, before int64) (int64, error) {
 		}
 	}
 }
+
+// writePayers writes the payer rows, sorted by topic in chunks (see write), stamped with now
+// (unix seconds). A set replaces the stored payer and seen_at; a touch keeps the payer and moves
+// seen_at forward. Rows are state, not deltas, so a failed write is simply written again.
+func (s *Store) writePayers(ctx context.Context, rows []payerRow, now int64) error {
+	slices.SortFunc(rows, func(a, b payerRow) int {
+		return strings.Compare(a.topic, b.topic)
+	})
+	for chunk := range slices.Chunk(rows, writeChunkRows) {
+		if err := s.writePayerChunk(ctx, chunk, now); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// readPayersSince calls fn for every payer row written at or after the given time (unix seconds),
+// streaming them from the primary
+func (s *Store) readPayersSince(ctx context.Context, since int64, fn func(r *payerRow)) error {
+	rows, err := s.db.QueryContext(ctx, s.queries.selectPayersSince, since)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	var r payerRow
+	for rows.Next() {
+		if err := rows.Scan(&r.topic, &r.key, &r.userID, &r.seenAt); err != nil {
+			return err
+		}
+		fn(&r)
+	}
+	return rows.Err()
+}
+
+// prunePayers deletes the payer rows not written since the given time (unix seconds), in chunks,
+// and returns how many it deleted
+func (s *Store) prunePayers(ctx context.Context, before int64) (int64, error) {
+	return s.pruneChunked(ctx, s.queries.prunePayers, before)
+}

limit/store_postgres.go

@@ -28,18 +28,39 @@ const (
 	postgresPruneQuery = `DELETE FROM usage WHERE updated_at < $1 AND key IN (SELECT key FROM usage WHERE updated_at < $1 LIMIT $2)` // The outer age check is rechecked on a row updated while the delete waits for it
 )
 
+// The payer upsert, one fixed statement per chunk like the usage upsert. A set (a payer key)
+// replaces the payer and seen_at; a touch (no payer key) keeps the payer and moves seen_at forward.
+const (
+	postgresSelectPayersSinceQuery = `SELECT topic, payer_key, payer_user_id, seen_at FROM topic_payer WHERE updated_at >= $1`
+	postgresUpsertPayerQuery       = `
+		INSERT INTO topic_payer AS p (topic, payer_key, payer_user_id, seen_at, updated_at)
+		SELECT t, k, u, s, $5::BIGINT
+		FROM unnest($1::TEXT[], $2::TEXT[], $3::TEXT[], $4::BIGINT[]) AS d(t, k, u, s)
+		ON CONFLICT (topic) DO UPDATE SET
+			payer_key = CASE WHEN EXCLUDED.payer_key = '' THEN p.payer_key ELSE EXCLUDED.payer_key END,
+			payer_user_id = CASE WHEN EXCLUDED.payer_key = '' THEN p.payer_user_id ELSE EXCLUDED.payer_user_id END,
+			seen_at = CASE WHEN EXCLUDED.payer_key = '' THEN GREATEST(p.seen_at, EXCLUDED.seen_at) ELSE EXCLUDED.seen_at END,
+			updated_at = EXCLUDED.updated_at
+	`
+	postgresPrunePayersQuery = `DELETE FROM topic_payer WHERE updated_at < $1 AND topic IN (SELECT topic FROM topic_payer WHERE updated_at < $1 LIMIT $2)` // As the usage prune
+)
+
 // NewPostgresStore returns the store on the shared pool; the usage table is the user schema's
 // (see package user, which creates and migrates it)
 func NewPostgresStore(d *db.DB) *Store {
 	s := &Store{
 		db: d,
 		queries: &queries{
-			selectSince: postgresSelectSinceQuery,
-			upsert:      postgresUpsertQuery,
-			prune:       postgresPruneQuery,
+			selectSince:       postgresSelectSinceQuery,
+			upsert:            postgresUpsertQuery,
+			prune:             postgresPruneQuery,
+			selectPayersSince: postgresSelectPayersSinceQuery,
+			upsertPayer:       postgresUpsertPayerQuery,
+			prunePayers:       postgresPrunePayersQuery,
 		},
 	}
 	s.writeChunk = s.writePostgres
+	s.writePayerChunk = s.writePayersPostgres
 	return s
 }
 
@@ -65,6 +86,17 @@ func (s *Store) writePostgres(ctx context.Context, rows []usageRow, now int64) e
 	return err
 }
 
+// writePayersPostgres writes the payer rows in one statement, as arrays
+func (s *Store) writePayersPostgres(ctx context.Context, rows []payerRow, now int64) error {
+	topics, keys, userIDs := make([]string, len(rows)), make([]string, len(rows)), make([]string, len(rows))
+	seenAts := make([]int64, len(rows))
+	for i, r := range rows {
+		topics[i], keys[i], userIDs[i], seenAts[i] = r.topic, r.key, r.userID, r.seenAt
+	}
+	_, err := s.db.ExecContext(ctx, s.queries.upsertPayer, topics, keys, userIDs, seenAts, now)
+	return err
+}
+
 // mayHaveCommittedPostgres reports whether a failed write may have committed anyway: a timeout
 // after the statement was sent, e.g. a commit acknowledged after the deadline. An error from the
 // server, or one before the statement was sent, means nothing was written.

limit/store_sqlite.go

@@ -24,18 +24,37 @@ const (
 	sqlitePruneQuery = `DELETE FROM usage WHERE updated_at < ?1 AND key IN (SELECT key FROM usage WHERE updated_at < ?1 LIMIT ?2)`
 )
 
+// Same as the Postgres payer queries (see there), one row per statement
+const (
+	sqliteSelectPayersSinceQuery = `SELECT topic, payer_key, payer_user_id, seen_at FROM topic_payer WHERE updated_at >= ?`
+	sqliteUpsertPayerQuery       = `
+		INSERT INTO topic_payer (topic, payer_key, payer_user_id, seen_at, updated_at)
+		VALUES (?, ?, ?, ?, ?)
+		ON CONFLICT (topic) DO UPDATE SET
+			payer_key = CASE WHEN excluded.payer_key = '' THEN payer_key ELSE excluded.payer_key END,
+			payer_user_id = CASE WHEN excluded.payer_key = '' THEN payer_user_id ELSE excluded.payer_user_id END,
+			seen_at = CASE WHEN excluded.payer_key = '' THEN MAX(seen_at, excluded.seen_at) ELSE excluded.seen_at END,
+			updated_at = excluded.updated_at
+	`
+	sqlitePrunePayersQuery = `DELETE FROM topic_payer WHERE updated_at < ?1 AND topic IN (SELECT topic FROM topic_payer WHERE updated_at < ?1 LIMIT ?2)`
+)
+
 // NewSQLiteStore returns the store on the user database (auth-file), which holds the usage
 // table (see package user, which creates and migrates it)
 func NewSQLiteStore(d *db.DB) *Store {
 	s := &Store{
 		db: d,
 		queries: &queries{
-			selectSince: sqliteSelectSinceQuery,
-			upsert:      sqliteUpsertQuery,
-			prune:       sqlitePruneQuery,
+			selectSince:       sqliteSelectSinceQuery,
+			upsert:            sqliteUpsertQuery,
+			prune:             sqlitePruneQuery,
+			selectPayersSince: sqliteSelectPayersSinceQuery,
+			upsertPayer:       sqliteUpsertPayerQuery,
+			prunePayers:       sqlitePrunePayersQuery,
 		},
 	}
 	s.writeChunk = s.writeSQLite
+	s.writePayerChunk = s.writePayersSQLite
 	return s
 }
 
@@ -55,3 +74,20 @@ func (s *Store) writeSQLite(ctx context.Context, rows []usageRow, now int64) err
 		return nil
 	})
 }
+
+// writePayersSQLite writes the payer rows in one transaction, one prepared statement per row
+func (s *Store) writePayersSQLite(ctx context.Context, rows []payerRow, now int64) error {
+	return db.ExecTx(s.db, func(tx *sql.Tx) error {
+		stmt, err := tx.PrepareContext(ctx, s.queries.upsertPayer)
+		if err != nil {
+			return err
+		}
+		defer stmt.Close()
+		for _, r := range rows {
+			if _, err := stmt.ExecContext(ctx, r.topic, r.key, r.userID, r.seenAt, now); err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+}

limit/store_test.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"path/filepath"
 	"slices"
+	"strings"
 	"testing"
 	"time"
 
@@ -213,3 +214,83 @@ func keysOf(rows map[string]usageRow) []string {
 	slices.Sort(keys)
 	return keys
 }
+
+func TestStore_WritePayers_SetReplacesTouchKeeps(t *testing.T) {
+	forEachBackend(t, func(t *testing.T, newStore func() *Store) {
+		s := newStore()
+		write := func(topic, key string, seenAt, now int64) {
+			require.Nil(t, s.writePayers(context.Background(), []payerRow{newPayerRow(topic, key, seenAt)}, now))
+		}
+		write("up1", "ip:1.2.3.4", 100, 101)
+		write("up1", "", 200, 201) // A touch keeps the payer
+		require.Equal(t, newPayerRow("up1", "ip:1.2.3.4", 200), readAllPayers(t, s, 0)["up1"])
+		write("up1", "", 150, 202) // An older touch does not move seen_at back
+		require.Equal(t, newPayerRow("up1", "ip:1.2.3.4", 200), readAllPayers(t, s, 0)["up1"])
+		write("up1", "user:u_1", 50, 203) // A set replaces both
+		require.Equal(t, newPayerRow("up1", "user:u_1", 50), readAllPayers(t, s, 0)["up1"])
+		write("up2", "", 300, 301) // A touch of an unknown topic: no payer
+		require.Equal(t, newPayerRow("up2", "", 300), readAllPayers(t, s, 0)["up2"])
+
+		// The read-back is by write time
+		require.Equal(t, []string{"up2"}, payerTopicsOf(readAllPayers(t, s, 301)))
+	})
+}
+
+func TestStore_WritePayers_Chunked(t *testing.T) {
+	forEachBackend(t, func(t *testing.T, newStore func() *Store) {
+		s := newStore()
+		rows := make([]payerRow, 0)
+		for i := 0; i < writeChunkRows+5; i++ {
+			rows = append(rows, newPayerRow(fmt.Sprintf("up%012d", i), "ip:1.2.3.4", 100))
+		}
+		require.Nil(t, s.writePayers(context.Background(), rows, 100))
+		require.Len(t, readAllPayers(t, s, 0), writeChunkRows+5)
+	})
+}
+
+func TestStore_PrunePayers(t *testing.T) {
+	forEachBackend(t, func(t *testing.T, newStore func() *Store) {
+		s := newStore()
+		for i := 0; i < 5; i++ {
+			require.Nil(t, s.writePayers(context.Background(), []payerRow{newPayerRow(fmt.Sprintf("up%d", i), "", 1)}, int64(100+i)))
+		}
+		deleted, err := s.prunePayers(context.Background(), 103)
+		require.Nil(t, err)
+		require.Equal(t, int64(3), deleted)
+		require.Equal(t, []string{"up3", "up4"}, payerTopicsOf(readAllPayers(t, s, 0)))
+	})
+}
+
+// readAllPayers returns the payer rows written at or after since, by topic
+func readAllPayers(t *testing.T, s *Store, since int64) map[string]payerRow {
+	t.Helper()
+	rows := make(map[string]payerRow)
+	require.Nil(t, s.readPayersSince(context.Background(), since, func(r *payerRow) {
+		rows[r.topic] = *r
+	}))
+	return rows
+}
+
+// payerTopicsOf returns the map's topics, sorted
+func payerTopicsOf(rows map[string]payerRow) []string {
+	topics := make([]string, 0, len(rows))
+	for topic := range rows {
+		topics = append(topics, topic)
+	}
+	slices.Sort(topics)
+	return topics
+}
+
+// newPayerRow returns a payer row; a user key ("user:<id>") has its user ID
+func newPayerRow(topic, key string, seenAt int64) payerRow {
+	userID, _ := strings.CutPrefix(key, "user:")
+	if userID == key {
+		userID = ""
+	}
+	return payerRow{
+		topic:  topic,
+		key:    key,
+		userID: userID,
+		seenAt: seenAt,
+	}
+}

limit/types.go

@@ -53,6 +53,14 @@ type Usage struct {
 	counts [numKinds]int64
 }
 
+// Payer is who pays for the pushes to a UnifiedPush topic (its subscriber), and when the topic last
+// had subscriber activity (see Service.SetPayer)
+type Payer struct {
+	Key    string    // The subscriber's limiter key; empty for a topic with subscriber activity or a refused push, but no payer
+	UserID string    // The subscriber's user ID, empty if anonymous
+	SeenAt time.Time // Last subscribe, poll or open stream, or the refused push that made the topic known; node clock
+}
+
 // Config configures the service
 type Config struct {
 	Store *Store // Where the daily counts are shared and kept across restarts; nil for memory only
@@ -87,6 +95,29 @@ type bucketShard struct {
 	mu      sync.Mutex // Protects entries and everything inside them
 }
 
+// payerWrite is what a payer entry still has to write to the store; a set outranks a touch
+type payerWrite uint8
+
+const (
+	payerWriteNone  payerWrite = iota
+	payerWriteTouch            // Only seen_at: keep the stored payer (TouchPayer, NoteTopic)
+	payerWriteSet              // Payer and seen_at (SetPayer)
+)
+
+// payerEntry is one topic's payer, as last set here or read from the store
+type payerEntry struct {
+	key    string
+	userID string
+	seenAt int64      // Unix seconds
+	write  payerWrite // Unwritten change; the read-back leaves an entry with one alone
+}
+
+// payerTable is the payers of the UnifiedPush topics, by topic
+type payerTable struct {
+	entries map[string]*payerEntry
+	mu      sync.Mutex // Protects entries and everything inside them
+}
+
 // bucket is a token bucket. It is created full: last is zero until the first use, and tokens
 // then start at the rule's burst. Tokens may go negative (debt) when a burst shrinks.
 type bucket struct {