Write a UnifiedPush topic's subscriber activity about once an hour, spread over time

v2.37.0 wrote every topic with an open stream every 5 minutes, and the topics touched together after a restart stayed in step, so ntfy.sh wrote ~45k topic rows in the same minute every 5 minutes (~145 rows/s on average, more than the message inserts), stalling the primary. Each topic now has its own interval between 30 and 60 minutes (by hash), and a subscribe or poll that only repeats the same rate visitor is a touch.

Claude-Session: https://claude.ai/code/session_01DeXuhSjq5Gta5qPJgSceFe

topic/service.go

@@ -9,6 +9,7 @@ package topic
 
 import (
 	"context"
+	"hash/fnv"
 	"maps"
 	"slices"
 	"sync"
@@ -35,8 +36,11 @@ const (
 	pruneAfter    = 7 * 24 * time.Hour
 
 	// touchInterval is how often a touch moves a topic's subscriber activity at most, so an open
-	// stream, touched by the manager every minute, writes only every few minutes
-	touchInterval = 5 * time.Minute
+	// stream, touched by the manager every minute, writes about once an hour. Each topic has its own
+	// interval between half of it and all of it (touchIntervalFor), so topics touched together,
+	// like the streams that reconnect after a restart, are not written together again. Hourly is
+	// plenty for the rate visitor's day (rateVisitorLiveFor) and the 410's four days.
+	touchInterval = time.Hour
 
 	// maxRefused caps the topics known only from refused pushes, so a push flood cannot fill memory;
 	// rows read from other nodes count too, so it is effectively cluster-wide
@@ -118,6 +122,9 @@ func (s *Service) Close() {
 func (s *Service) SetRateVisitor(topic, userID, ip string) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
+	if e, ok := s.entries[topic]; ok && e.RateVisitorUserID == userID && e.RateVisitorIP == ip && s.touchedRecentlyNoLock(e) {
+		return // The same rate visitor again (a poll, a reconnect) is a touch
+	}
 	e := s.touchNoLock(topic)
 	e.RateVisitorUserID, e.RateVisitorIP = userID, ip
 	s.addPendingNoLock(e, true)
@@ -128,7 +135,7 @@ func (s *Service) SetRateVisitor(topic, userID, ip string) {
 func (s *Service) Touch(topic string) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	if e, ok := s.entries[topic]; ok && e.SubscribedAt != 0 && s.now().Unix()-e.SubscribedAt < int64(touchInterval/time.Second) {
+	if e, ok := s.entries[topic]; ok && s.touchedRecentlyNoLock(e) {
 		return
 	}
 	s.addPendingNoLock(s.touchNoLock(topic), false)
@@ -354,6 +361,12 @@ func (s *Service) addPendingNoLock(e *State, rateVisitor bool) {
 	s.pending[e.Topic] = &r
 }
 
+// touchedRecentlyNoLock returns true if the topic's subscriber activity was recorded within its own
+// touch interval. The caller must hold s.mu.
+func (s *Service) touchedRecentlyNoLock(e *State) bool {
+	return e.SubscribedAt != 0 && s.now().Unix()-e.SubscribedAt < int64(touchIntervalFor(e.Topic)/time.Second)
+}
+
 // stateNoLock returns a copy of the topic's state, or nil if it is unknown. The caller must hold s.mu.
 func (s *Service) stateNoLock(topic string) *State {
 	e, ok := s.entries[topic]
@@ -363,3 +376,12 @@ func (s *Service) stateNoLock(topic string) *State {
 	c := *e
 	return &c
 }
+
+// touchIntervalFor returns the topic's own touch interval, between half of touchInterval and all
+// of it, the same on every node and after a restart
+func touchIntervalFor(topic string) time.Duration {
+	h := fnv.New32a()
+	h.Write([]byte(topic))
+	half := touchInterval / 2
+	return half + time.Duration(h.Sum32()%uint32(half/time.Second))*time.Second
+}

topic/service_test.go

@@ -100,12 +100,12 @@ func TestService_State_ReturnsCopy(t *testing.T) {
 }
 
 func TestService_Touch_Throttled(t *testing.T) {
-	// A touch moves subscriber activity at most once per touchInterval, so an open stream the
-	// manager touches every minute costs a write every few minutes only
+	// A touch moves subscriber activity at most once per the topic's own touch interval, so an open
+	// stream the manager touches every minute costs a write about once an hour only
 	s, clock := newTestServiceWithClock(t)
 	s.SetRateVisitor("upA", "", "1.2.3.4")
 	set := clock.unix()
-	clock.advance(touchInterval - time.Second)
+	clock.advance(touchIntervalFor("upA") - time.Second)
 	s.Touch("upA")
 	require.Equal(t, set, s.State("upA").SubscribedAt)
 	clock.advance(time.Second)
@@ -121,6 +121,55 @@ func TestService_Touch_Throttled(t *testing.T) {
 	require.Equal(t, clock.unix(), s.State("upC").SubscribedAt)
 }
 
+func TestService_Touch_SpreadOverTime(t *testing.T) {
+	// Topics touched together, like the streams that all reconnect after a restart, are written
+	// again spread over the second half of touchInterval, not all in the same minute
+	s, clock := newTestServiceWithClock(t)
+	topics := make([]string, 1000)
+	for i := range topics {
+		topics[i] = fmt.Sprintf("up%012d", i)
+		s.Touch(topics[i])
+	}
+	lastWritten := make(map[string]int64)
+	for _, topic := range topics {
+		lastWritten[topic] = clock.unix()
+	}
+	rewritten := make(map[string]bool)
+	for minute := 1; minute <= int(touchInterval/time.Minute); minute++ {
+		clock.advance(time.Minute)
+		written := 0
+		for _, topic := range topics {
+			s.Touch(topic) // The manager touches every topic with an open stream every minute
+			if s.State(topic).SubscribedAt == clock.unix() {
+				written++
+				require.GreaterOrEqual(t, clock.unix()-lastWritten[topic], int64(touchInterval/2/time.Second), "%s written again before half of touchInterval", topic)
+				lastWritten[topic] = clock.unix()
+				rewritten[topic] = true
+			}
+		}
+		require.LessOrEqual(t, written, len(topics)/15, "minute %d: %d of %d topics written at once", minute, written, len(topics))
+	}
+	require.Len(t, rewritten, len(topics), "every topic written again within touchInterval")
+}
+
+func TestService_SetRateVisitor_UnchangedThrottled(t *testing.T) {
+	// A subscribe or poll that only repeats the same rate visitor counts as a touch, so a client
+	// polling every minute does not write every minute; a new rate visitor is written right away
+	s, clock := newTestServiceWithClock(t)
+	s.SetRateVisitor("upA", "u_1", "1.2.3.4")
+	set := clock.unix()
+	clock.advance(time.Minute)
+	s.SetRateVisitor("upA", "u_1", "1.2.3.4")
+	require.Equal(t, set, s.State("upA").SubscribedAt)
+
+	s.SetRateVisitor("upA", "", "5.6.7.8")
+	require.Equal(t, state("upA", set, clock.unix(), "", "5.6.7.8"), stateOf(s, "upA"))
+
+	clock.advance(touchInterval)
+	s.SetRateVisitor("upA", "", "5.6.7.8")
+	require.Equal(t, clock.unix(), s.State("upA").SubscribedAt)
+}
+
 func TestService_Refused_Bounded(t *testing.T) {
 	// Topics known only from refused pushes are capped; a subscriber frees a slot
 	s, _ := newTestServiceWithClock(t)