diff --git a/server/server.go b/server/server.go index 77f5c42c..d738f24f 100644 --- a/server/server.go +++ b/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 { diff --git a/server/server_cluster.go b/server/server_cluster.go index 428367b8..1f509070 100644 --- a/server/server_cluster.go +++ b/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() + } +} diff --git a/server/server_cluster_test.go b/server/server_cluster_test.go index a4f2b97c..a936d820 100644 --- a/server/server_cluster_test.go +++ b/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) diff --git a/server/server_tail.go b/server/server_tail.go index 6bb4d58c..c003ef80 100644 --- a/server/server_tail.go +++ b/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)