mirror of
https://github.com/binwiederhier/ntfy.git
synced 2026-11-10 22:50:59 +08:00
binwiederhier
859e6443c7
A cluster message whose topic is known on the receiving node but has no subscriber there was published to nobody and uncounted, and its Keepalive kept the topic from ever expiring: a passive node held every topic the active one published to. It is now counted in ntfy_cluster_messages_wasted_total and not published, so idle topics expire after topicExpungeAfter.
53 lines
2.2 KiB
Go
53 lines
2.2 KiB
Go
package server
|
|
|
|
import (
|
|
"time"
|
|
|
|
"heckel.io/ntfy/v2/log"
|
|
"heckel.io/ntfy/v2/metrics"
|
|
"heckel.io/ntfy/v2/model"
|
|
)
|
|
|
|
// 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
|
|
// origin node, so this only publishes to the local topic, and never re-relays.
|
|
//
|
|
// A message whose topic has no subscriber here is counted as wasted and not published: publishing
|
|
// would keep the idle topic alive (Keepalive), so a node would never expunge a topic the other
|
|
// nodes keep publishing to. A subscriber that comes along later gets it from its since= read of
|
|
// the cache, or from the tail (which sends what never went out here).
|
|
func (s *Server) deliverFromBus(m *model.Message) {
|
|
s.mu.RLock()
|
|
t, ok := s.topics[m.Topic]
|
|
s.mu.RUnlock()
|
|
if !ok || t.SubscribersCount() == 0 {
|
|
metrics.ClusterMessagesWasted.Inc() // Relayed here needlessly: no subscriber here wanted the message
|
|
return
|
|
}
|
|
v := s.visitor(m.Sender, nil)
|
|
if err := t.Publish(v, m); err != nil {
|
|
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()
|
|
}
|
|
}
|