Rename closeLocalSubscribers to maybeCloseLocalSubscribers and move it to the bottom

server/server.go

@@ -348,7 +348,7 @@ func New(conf *Config) (*Server, error) {
 		Role:            conf.ClusterRole,
 		Secret:          conf.ClusterSecret,
 		BatchLinger:     conf.ClusterBatchLinger,
-		IsolatedFunc:    s.closeLocalSubscribers,
+		IsolatedFunc:    s.maybeCloseLocalSubscribers,
 		MaxMessageBytes: int64(conf.MessageSizeLimit)*4 + 1024, // Envelope overhead over the raw message
 	}, pool, s.deliverFromBus)
 	if err != nil {

server/server_cluster.go

@@ -8,26 +8,6 @@ import (
 	"heckel.io/ntfy/v2/model"
 )
 
-// closeLocalSubscribers closes every subscriber connection on this node. The cluster calls it
-// while this node is isolated (lost its registration, peers healthy): peers no longer forward
-// to it, so its subscribers would silently receive nothing; they reconnect to a healthy node.
-//
-// Not while the tail still reads the database: then the subscribers get every message anyway,
-// a little late. Heartbeats time out when the database is slow, not only when it is gone, and
-// closing tens of thousands of connections then sends them to nodes that are just as slow, as
-// a reconnect storm that keeps the database slow.
-func (s *Server) closeLocalSubscribers() {
-	if s.tail != nil && time.Since(s.tail.LastRead()) < tailStaleAfter {
-		log.Tag(tagTail).Info("Keeping local subscribers although the cluster registration is stale: the tail still reads the database")
-		return
-	}
-	s.mu.RLock()
-	defer s.mu.RUnlock()
-	for _, t := range s.topics {
-		t.CancelAllSubscribers()
-	}
-}
-
 // deliverFromBus delivers a message received from a peer node (via the cluster) to this
 // node's local subscribers. It is the receive-side counterpart to Cluster.ForwardMessage: local
 // delivery and all global side effects (Firebase, email, web push, upstream) already ran on the
@@ -45,3 +25,23 @@ func (s *Server) deliverFromBus(m *model.Message) {
 		logvm(v, m).Err(err).Warn("Cluster: unable to deliver fan-out message to local subscribers")
 	}
 }
+
+// maybeCloseLocalSubscribers closes every subscriber connection on this node. The cluster calls it
+// while this node is isolated (lost its registration, peers healthy): peers no longer forward
+// to it, so its subscribers would silently receive nothing; they reconnect to a healthy node.
+//
+// Not while the tail still reads the database: then the subscribers get every message anyway,
+// a little late. Heartbeats time out when the database is slow, not only when it is gone, and
+// closing tens of thousands of connections then sends them to nodes that are just as slow, as
+// a reconnect storm that keeps the database slow.
+func (s *Server) maybeCloseLocalSubscribers() {
+	if s.tail != nil && time.Since(s.tail.LastRead()) < tailStaleAfter {
+		log.Tag(tagTail).Info("Keeping local subscribers although the cluster registration is stale: the tail still reads the database")
+		return
+	}
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	for _, t := range s.topics {
+		t.CancelAllSubscribers()
+	}
+}

server/server_cluster_test.go

@@ -302,7 +302,7 @@ func TestServer_Cluster_IsolatedNodeClosesSubscribers(t *testing.T) {
 		t, ok := topics["mytopic"]
 		return ok && t.SubscribersCount() == 1
 	})
-	s.closeLocalSubscribers()
+	s.maybeCloseLocalSubscribers()
 	select {
 	case <-done:
 	case <-time.After(3 * time.Second):
@@ -328,7 +328,7 @@ func TestServer_Cluster_IsolatedNodeKeepsSubscribersWhileTheTailReads(t *testing
 		return ok && t.SubscribersCount() == 1
 	})
 	waitFor(t, func() bool { return !s.tail.LastRead().IsZero() })
-	s.closeLocalSubscribers()
+	s.maybeCloseLocalSubscribers()
 	select {
 	case <-done:
 		t.Fatal("subscriber connection closed although the tail reads")
@@ -337,7 +337,7 @@ func TestServer_Cluster_IsolatedNodeKeepsSubscribersWhileTheTailReads(t *testing
 }
 
 func TestServer_Cluster_StopDoesNotWaitForCallbackUnderServerLock(t *testing.T) {
-	// The mesh's isolation loop calls back into the server (closeLocalSubscribers, which takes
+	// The mesh's isolation loop calls back into the server (maybeCloseLocalSubscribers, which takes
 	// s.mu). Stop must not hold s.mu while it waits for that loop, or a callback already
 	// waiting for the lock deadlocks shutdown before the message cache is drained.
 	s, err := New(newTestConfig(t, ""))
@@ -363,7 +363,7 @@ func TestServer_Cluster_StopDoesNotWaitForCallbackUnderServerLock(t *testing.T)
 		IsolatedFunc: func() {
 			once.Do(func() { close(entered) })
 			<-release
-			s.closeLocalSubscribers()
+			s.maybeCloseLocalSubscribers()
 		},
 	}, pool, func(*model.Message) {})
 	require.NoError(t, err)

server/server_tail.go

@@ -11,7 +11,7 @@ import (
 const (
 	tagTail        = "tail"
 	tailInterval   = time.Second      // How often the tail reads what was written; bounds how late a missed message arrives
-	tailStaleAfter = 30 * time.Second // A tail that has not read for this long cannot stand in for the fan-out (see closeLocalSubscribers)
+	tailStaleAfter = 30 * time.Second // A tail that has not read for this long cannot stand in for the fan-out (see maybeCloseLocalSubscribers)
 
 	// unwrittenMarkerMargin is how far before an unwritten marker's delivery here its replay
 	// starts: the fan-out linger, and publish times having whole seconds (see sinceForTopic)