Merge pull request #2070 from binwiederhier/stats-deltas

Write the published-messages total as per-node deltas

docs/releases.md

@@ -2222,6 +2222,10 @@ and the [ntfy Android app](https://github.com/binwiederhier/ntfy-android/release
 * Reject phone number verification unless Twilio confirms the code as approved, instead of accepting any successful response
 * Ignore duplicate topics in a subscription and limit a subscription to 150 topics, so a single request cannot replay the same topic's cache over and over
 
+**Bug fixes + maintenance:**
+
+* Count published messages from every cluster node: each node adds its own publishes to the stored total instead of the leader overwriting it with its own count (experimental clustering)
+
 ### ntfy iOS app v1.8.0 (UNRELEASED)
 
 **Features:**

message/cache.go

@@ -590,11 +590,12 @@ func (c *Cache) readAttachmentBytesUsed(rows *sql.Rows) (int64, error) {
 	return size, nil
 }
 
-// UpdateStats updates the total message count statistic
-func (c *Cache) UpdateStats(messages int64) error {
+// AddStats adds the given delta to the total message count statistic. Increments (rather than
+// absolute writes) let multiple cluster nodes contribute to one shared counter.
+func (c *Cache) AddStats(delta int64) error {
 	c.maybeLock()
 	defer c.maybeUnlock()
-	_, err := c.db.Exec(c.queries.updateStats, messages)
+	_, err := c.db.Exec(c.queries.updateStats, delta)
 	return err
 }
 

message/cache_postgres.go

@@ -124,7 +124,7 @@ const (
 		RETURNING mid, sequence_id, time, event, expires, topic, message, title, priority, tags, click, icon, actions, attachment_name, attachment_type, attachment_size, attachment_expires, attachment_url, sender, user_id, content_type, encoding
 	`
 
-	postgresUpdateStatsQuery       = `UPDATE message_stats SET value = $1 WHERE key = 'messages'`
+	postgresUpdateStatsQuery       = `UPDATE message_stats SET value = value + $1 WHERE key = 'messages'`
 	postgresUpdateMessageTimeQuery = `UPDATE message SET time = $1 WHERE mid = $2`
 )
 

message/cache_sqlite.go

@@ -102,7 +102,7 @@ const (
 	sqliteSelectAttachmentsWithSizesQuery    = `SELECT mid, attachment_size FROM messages WHERE attachment_expires > ? AND attachment_deleted = 0`
 
 	sqliteSelectStatsQuery       = `SELECT value FROM stats WHERE key = 'messages'`
-	sqliteUpdateStatsQuery       = `UPDATE stats SET value = ? WHERE key = 'messages'`
+	sqliteUpdateStatsQuery       = `UPDATE stats SET value = value + ? WHERE key = 'messages'`
 	sqliteUpdateMessageTimeQuery = `UPDATE messages SET time = ? WHERE mid = ?`
 )
 

message/cache_test.go

@@ -714,14 +714,14 @@ func TestStore_Stats(t *testing.T) {
 		require.Nil(t, err)
 		require.Equal(t, int64(0), messages)
 
-		// Update stats
-		require.Nil(t, s.UpdateStats(42))
+		// Add stats
+		require.Nil(t, s.AddStats(42))
 		messages, err = s.Stats()
 		require.Nil(t, err)
 		require.Equal(t, int64(42), messages)
 
-		// Update again (overwrites)
-		require.Nil(t, s.UpdateStats(100))
+		// Add again (increments)
+		require.Nil(t, s.AddStats(58))
 		messages, err = s.Stats()
 		require.Nil(t, err)
 		require.Equal(t, int64(100), messages)
@@ -1110,6 +1110,30 @@ func TestStore_AddMessage_InvalidUTF8BatchDoesNotDropValidMessages(t *testing.T)
 	})
 }
 
+func TestStore_AddStats_SumsAcrossNodes(t *testing.T) {
+	// The global message counter is written by every cluster node; writes must be additive
+	// increments, not absolute values, or nodes overwrite each other's counts
+	schemaDSN := dbtest.CreateTestPostgresSchema(t)
+	openStore := func() *message.Cache {
+		host, err := pg.Open(schemaDSN)
+		require.Nil(t, err)
+		d := db.New(host, nil)
+		store, err := message.NewPostgresStore(d, 0, 0)
+		require.Nil(t, err)
+		t.Cleanup(func() {
+			store.Close()
+		})
+		return store
+	}
+	a, b := openStore(), openStore()
+	require.Nil(t, a.AddStats(5))
+	require.Nil(t, b.AddStats(3))
+	require.Nil(t, a.AddStats(2))
+	messages, err := a.Stats()
+	require.Nil(t, err)
+	require.Equal(t, int64(10), messages)
+}
+
 func TestStore_CloseGivesUpOnAStuckBatchWrite(t *testing.T) {
 	// A wedged database must not make Close, and with it the entire shutdown, block forever.
 	// systemd would SIGKILL us eventually, but only after its stop timeout.
@@ -1280,7 +1304,9 @@ func TestStore_MessagesSinceID_ReplicaLagFallsBackToPrimary(t *testing.T) {
 		d.MarkReplicasHealthyForTest() // Route ReadOnly() to the lagged replica deterministically
 		store, err := message.NewPostgresStore(d, 0, 0)
 		require.Nil(t, err)
-		t.Cleanup(func() { store.Close() })
+		t.Cleanup(func() {
+			store.Close()
+		})
 		return store
 	}
 	// The "replica" is a separate schema holding only the first 3 messages (fully lagged after

server/server.go

@@ -69,6 +69,7 @@ type Server struct {
 	twilio            *twilio.Client
 	messages          int64                               // Total number of messages (persisted if messageCache enabled)
 	messagesHistory   []int64                             // Last n values of the messages counter, used to determine rate
+	messagesFlushed   int64                               // Value of the messages counter at the last stats flush; the difference is written as a delta
 	userManager       *user.Manager                       // Might be nil!
 	messageCache      *message.Cache                      // Database that stores the messages
 	webPush           *webpush.Store                      // Database that stores web push subscriptions
@@ -336,6 +337,7 @@ func New(conf *Config) (*Server, error) {
 		userManager:     userManager,
 		messages:        messages,
 		messagesHistory: []int64{messages},
+		messagesFlushed: messages, // The loaded total is already persisted; only new publishes are deltas
 		visitors:        make(map[string]*visitor),
 		stripe:          stripe,
 	}
@@ -2314,11 +2316,23 @@ func (s *Server) updateAndWriteStats(messagesCount int64) {
 	if len(s.messagesHistory) > messagesHistoryMax {
 		s.messagesHistory = s.messagesHistory[1:]
 	}
+	snapshot, delta := s.messages, s.messages-s.messagesFlushed
 	s.mu.Unlock()
-	if !s.cluster.IsLeader() {
-		return // Overwritten, not added to: a second writer would replace the total with its own count
-	}
-	if err := s.messageCache.UpdateStats(messagesCount); err != nil {
+	// Write this node's delta and fold the cluster-wide total back in, so every node's counter
+	// (and /v1/stats) converges on the shared sum instead of overwriting it
+	if err := s.messageCache.AddStats(delta); err != nil {
 		log.Tag(tagManager).Err(err).Warn("Cannot write messages stats")
+		return // Delta remains unflushed; retried on the next tick
 	}
+	total, err := s.messageCache.Stats()
+	if err != nil {
+		log.Tag(tagManager).Err(err).Warn("Cannot read messages stats")
+		total = snapshot // Keep the local view; only the flush marker moves
+	}
+	s.mu.Lock()
+	// The marker is the folded total (everything the database already has, incl. peer counts),
+	// so the next delta is only this node's new publishes; peer counts must never be re-added
+	s.messagesFlushed = total
+	s.messages = total + (s.messages - snapshot) // Publishes that arrived while flushing stay counted
+	s.mu.Unlock()
 }

server/server_manager_test.go

@@ -46,8 +46,8 @@ func TestServer_Manager_ExpiresDeliveries(t *testing.T) {
 
 func TestServer_Manager_SharedJobsRunOnLeaderOnly(t *testing.T) {
 	forEachBackend(t, func(t *testing.T, databaseURL string) {
-		// A passive (non-leader) node must not prune the shared database, nor overwrite the
-		// stored message total with its own count, which only covers its own publishes
+		// A passive (non-leader) node must not prune the shared database. It does add its own
+		// publishes to the stored message total, as a delta, so another node's count survives
 		s := newTestServer(t, newTestConfig(t, databaseURL))
 		b := &fakeCluster{}
 		b.setLeader(false)
@@ -58,7 +58,7 @@ func TestServer_Manager_SharedJobsRunOnLeaderOnly(t *testing.T) {
 		require.Equal(t, 200, rr.Code)
 		m := toMessage(t, rr.Body.String())
 		require.Nil(t, s.messageCache.ExpireMessages("mytopic"))
-		require.Nil(t, s.messageCache.UpdateStats(500))
+		require.Nil(t, s.messageCache.AddStats(500))
 
 		// Not the leader: nothing shared is touched
 		s.execManager()
@@ -66,15 +66,40 @@ func TestServer_Manager_SharedJobsRunOnLeaderOnly(t *testing.T) {
 		require.Nil(t, err)
 		messages, err := s.messageCache.Stats()
 		require.Nil(t, err)
-		require.Equal(t, int64(500), messages)
+		require.Equal(t, int64(501), messages)
 
-		// Leader: expired messages are pruned and the total is written
+		// Leader: expired messages are pruned, and the total is not counted twice
 		b.setLeader(true)
 		s.execManager()
 		_, err = s.messageCache.Message(m.ID)
 		require.Equal(t, model.ErrMessageNotFound, err)
 		messages, err = s.messageCache.Stats()
 		require.Nil(t, err)
-		require.Equal(t, s.messages, messages)
+		require.Equal(t, int64(501), messages)
+		require.Equal(t, int64(501), s.messages)
+	})
+}
+
+func TestServer_Manager_MessagesTotalSumsAcrossNodes(t *testing.T) {
+	forEachBackend(t, func(t *testing.T, databaseURL string) {
+		// Two nodes on one database: each adds only its own publishes to the stored total, and
+		// a node that loaded the total at startup does not add it again on its first flush
+		c := newTestConfig(t, databaseURL)
+		s1 := newTestServer(t, c)
+		require.Equal(t, 200, request(t, s1, "POST", "/mytopic", "one", nil).Code)
+		require.Equal(t, 200, request(t, s1, "POST", "/mytopic", "two", nil).Code)
+		s1.execManager()
+
+		s2 := newTestServer(t, c)
+		require.Equal(t, int64(2), s2.messages)
+		require.Equal(t, 200, request(t, s2, "POST", "/mytopic", "three", nil).Code)
+		s2.execManager()
+		s1.execManager()
+
+		messages, err := s1.messageCache.Stats()
+		require.Nil(t, err)
+		require.Equal(t, int64(3), messages)
+		require.Equal(t, int64(3), s1.messages)
+		require.Equal(t, int64(3), s2.messages)
 	})
 }