From ccfbc2309d71c63724b8869bf766b36b3f6d15db Mon Sep 17 00:00:00 2001 From: binwiederhier Date: Sun, 2 Aug 2026 23:54:35 +0200 Subject: [PATCH] Refactor --- cluster/cluster.go | 46 +++------- cluster/cluster_load_test.go | 8 +- cluster/cluster_mesh.go | 132 ++++++++++++++++++--------- cluster/cluster_mesh_test.go | 60 ++++++++----- cluster/cluster_nop.go | 6 +- cluster/cluster_test.go | 3 +- cluster/types.go | 33 ++++++- db/pg/leader.go | 162 +++++++++++++++++++++++++--------- db/pg/leader_internal_test.go | 41 +++++++++ db/pg/leader_test.go | 98 ++++++++++---------- metrics/metrics.go | 6 +- metrics/metrics_test.go | 2 +- server/server.go | 30 ++++--- server/server_cluster_test.go | 53 +++++++++-- 14 files changed, 459 insertions(+), 221 deletions(-) create mode 100644 db/pg/leader_internal_test.go diff --git a/cluster/cluster.go b/cluster/cluster.go index 8c6dd27d..167a7490 100644 --- a/cluster/cluster.go +++ b/cluster/cluster.go @@ -61,52 +61,26 @@ const ( DefaultBatchLinger = 500 * time.Millisecond ) -// Config configures the cluster. It is assembled by the server from its own config, which keeps -// this package free of server types. -type Config struct { - Enabled bool // Master switch; when false, New returns the nop cluster - NodeID NodeID // Stable per-node identifier; required - AdvertiseURL string // Base URL peers use to reach this node's fan-out endpoint - Secret string // Shared secret authenticating node-to-node fan-out requests - HeartbeatInterval time.Duration // How often the node registry heartbeat is refreshed - NodeTTL time.Duration // Registry rows older than this do not count as live peers - BatchLinger time.Duration // How long messages wait in a peer queue to form a batch; 0 = send immediately - StateInterval time.Duration // How often the full subscription state is pushed to peers - MaxMessageBytes int64 // Upper bound for a single message on the wire (batch limits derive from this) -} - -// DeliverFunc hands a message received from a peer node to this node's local subscribers. The -// server supplies it, which inverts the dependency: this package never imports the server. -type DeliverFunc func(m *model.Message) - -// TopicsFunc returns the topics that currently have at least one live subscriber, computed -// fresh on every call: membership is never tracked as a list, so topics "leave" simply by not -// appearing in the next snapshot. The server supplies it (same inversion as DeliverFunc). -type TopicsFunc func() []string - // Cluster fans published messages out to peer cluster nodes and receives their fan-out requests. // Local delivery to a node's own subscribers still happens inline in the server; the cluster // only covers the cross-node hop. type Cluster interface { - // The internal peer API (/v1/internal/*), served on the dedicated cluster listener. The - // nop cluster answers 404. http.Handler - // Relay sends a locally published message on to the peer nodes that may have subscribers + // ForwardMessage sends a locally published message on to the peer nodes that may have subscribers // for its topic (all of them, when subscription knowledge is missing or stale). It is // fire-and-forget and must not block the caller's request path. - Relay(m *model.Message) error - // AnnounceTopics tells peers that these topics just gained their first local subscriber, - // closing the routing-knowledge window to ~one round trip. Nop in single-node mode. - AnnounceTopics(topics []string) + ForwardMessage(m *model.Message) error + // BroadcastState pushes a subscription-state delta to ALL peers (unlike ForwardMessage, + // which routes), closing the routing-knowledge window to ~one round trip. Nop single-node. + BroadcastState(state *State) // IsLeader reports whether this node holds the cluster leader lock. Singleton background // jobs (e.g. the Firebase keepaliver) are gated on the leader. IsLeader() bool - // TODO(T8): add Healthy() bool reflecting registration health (unhealthy when the last - // successful Register is older than NodeTTL, i.e. when peers stop relaying to this node), - // so /v1/health can pull a DB-partitioned node out of DNS rotation. Needs a fail-open - // answer at the health checker first: during a full DB outage ALL nodes fail to register - // while the mesh keeps delivering on stale peer caches, and naively pulling every node - // would turn a control-plane blip into a total outage. + // Healthy reports whether this node is fit to serve: its registry heartbeat is fresh + // enough (within NodeTTL) that peers still forward messages to it. Health checkers must + // fail open (never pull ALL nodes): during a full database outage every node reports + // unhealthy while the mesh keeps delivering on stale peer caches. + Healthy() bool // Close stops the cluster and releases its resources. Close() error } diff --git a/cluster/cluster_load_test.go b/cluster/cluster_load_test.go index a2689603..1c92934b 100644 --- a/cluster/cluster_load_test.go +++ b/cluster/cluster_load_test.go @@ -67,7 +67,7 @@ func TestMesh_Soak(t *testing.T) { ticker := time.NewTicker(publishInterval) defer ticker.Stop() for i := 0; i < messagesPerPublisher; i++ { - require.Nil(t, mesh.Relay(model.NewDefaultMessage("mytopic", fmt.Sprintf("p%d-m%d", p, i)))) + require.Nil(t, mesh.ForwardMessage(model.NewDefaultMessage("mytopic", fmt.Sprintf("p%d-m%d", p, i)))) <-ticker.C } }(p) @@ -89,10 +89,10 @@ func TestMesh_Soak(t *testing.T) { total, requests, float64(total)/float64(requests), float64(total)/elapsed.Seconds()) } -// BenchmarkRelay measures the publish-path cost of Relay: marshal + peer lookup (cached) +// BenchmarkForwardMessage measures the publish-path cost of ForwardMessage: marshal + peer lookup (cached) // + enqueue. The peer never drains, so enqueued fragments are dropped once the queue fills; // the benchmark measures the hot path, not HTTP delivery. -func BenchmarkRelay(b *testing.B) { +func BenchmarkForwardMessage(b *testing.B) { if os.Getenv("NTFY_TEST_DATABASE_URL") == "" { b.Skip("NTFY_TEST_DATABASE_URL not set") } @@ -107,7 +107,7 @@ func BenchmarkRelay(b *testing.B) { m := model.NewDefaultMessage("mytopic", "benchmark message body of typical size for a push") b.ResetTimer() for i := 0; i < b.N; i++ { - if err := mesh.Relay(m); err != nil { + if err := mesh.ForwardMessage(m); err != nil { b.Fatal(err) } } diff --git a/cluster/cluster_mesh.go b/cluster/cluster_mesh.go index 0404b3c0..24beb504 100644 --- a/cluster/cluster_mesh.go +++ b/cluster/cluster_mesh.go @@ -37,22 +37,24 @@ const ( // Each peer has its own bounded send queue and delivery worker, so a slow or wedged peer only // backs up (and eventually drops) its own queue and never delays delivery to healthy peers. type meshCluster struct { - conf *Config - deliver DeliverFunc - topics TopicsFunc - registry *registry.Registry - leader *pg.Leader - httpClient *http.Client - mux *http.ServeMux // The internal peer API; Cluster is an http.Handler - queues map[NodeID]*peerQueue // per-peer send queues; reconciled against the registry - closed bool // Guards against Relay spawning new workers after Close - states map[NodeID]*peerState // what each peer last told us (subscription knowledge) - lastStatePush time.Time // Only touched by the heartbeat goroutine - ctx context.Context - cancel context.CancelFunc - wg sync.WaitGroup - mu sync.Mutex // Protects queues and closed - statesMu sync.Mutex // Protects states + conf *Config + deliver DeliverFunc + topics TopicsFunc + registry *registry.Registry + leader *pg.Leader + httpClient *http.Client + mux *http.ServeMux // The internal peer API; Cluster is an http.Handler + queues map[NodeID]*peerQueue // per-peer send queues; reconciled against the registry + closed bool // Guards against ForwardMessage spawning new workers after Close + states map[NodeID]*peerState // what each peer last told us (subscription knowledge) + lastStatePush time.Time // Only touched by the heartbeat goroutine + knownPeers map[NodeID]string // Peers seen in the last reconcile, for join/leave logging + lastRegistered time.Time // Last successful registry heartbeat, for Healthy + ctx context.Context + cancel context.CancelFunc + wg sync.WaitGroup + mu sync.Mutex // Protects queues, closed, knownPeers and lastRegistered + statesMu sync.Mutex // Protects states } // newMeshCluster creates the mesh cluster: it sets up the registry schema, registers this node @@ -73,16 +75,19 @@ func newMeshCluster(conf *Config, pool *db.DB, deliver DeliverFunc, topics Topic } ctx, cancel := context.WithCancel(context.Background()) c := &meshCluster{ - conf: conf, - deliver: deliver, - topics: topics, - registry: reg, - leader: pg.NewLeader(pool.Primary(), pg.LeaderLockKey), - httpClient: &http.Client{Timeout: meshHTTPTimeout}, - queues: make(map[NodeID]*peerQueue), - states: make(map[NodeID]*peerState), - ctx: ctx, - cancel: cancel, + conf: conf, + deliver: deliver, + topics: topics, + registry: reg, + // Renews its lease on its own fixed cadence; see pg.Leader for the semantics + leader: pg.NewLeader(pool.Primary(), pg.LeaderLockKey, conf.LeaderRenewInterval), + httpClient: &http.Client{Timeout: meshHTTPTimeout}, + queues: make(map[NodeID]*peerQueue), + lastRegistered: time.Now(), // The synchronous Register above just succeeded + states: make(map[NodeID]*peerState), + knownPeers: make(map[NodeID]string), + ctx: ctx, + cancel: cancel, } c.mux = http.NewServeMux() c.mux.HandleFunc("POST "+MessagePath, c.authenticated(c.handleMessage)) @@ -147,16 +152,19 @@ func (c *meshCluster) heartbeatLoop() { // periodically push our subscription state to peers. // // A node that cannot even register itself aborts the tick: the remaining database work would -// fail against the same database, and everything downstream degrades safely without it -- Relay +// fail against the same database, and everything downstream degrades safely without it -- ForwardMessage // serves the stale peer cache on its own, and peers fall back to broadcasting to us once our // last pushed state expires. func (c *meshCluster) heartbeat() error { - // TODO(T8): record the last successful Register here to back the future Healthy() method - // (see the Cluster interface) if err := c.registry.Register(); err != nil { return err } - if c.leader.TryAcquire(c.ctx) { + c.mu.Lock() + c.lastRegistered = time.Now() + c.mu.Unlock() + // Effective leadership: pg.Leader's lease semantics guarantee a no-leader gap on + // failover, never two leaders + if c.leader.IsLeader() { metrics.ClusterLeader.Set(1) if err := c.registry.Prune(); err != nil { log.Tag(tag).Err(err).Warn("Failed to prune stale nodes") // Housekeeping only; not fatal for the tick @@ -179,7 +187,7 @@ func (c *meshCluster) heartbeat() error { // reconcilePeers aligns this node's per-peer attachments with the live peer set: it retires the // queues (and workers) of peers that have left the registry or re-registered under a new // advertise URL (the retired queue's remainder was headed for a dead address anyway), and prunes -// the stale state of departed peers. New and replacement queues are created lazily by Relay, not +// the stale state of departed peers. New and replacement queues are created lazily by ForwardMessage, not // here, so a freshly joined peer is reachable immediately. func (c *meshCluster) reconcilePeers(peers []*registry.Peer) { metrics.ClusterPeers.Set(float64(len(peers))) @@ -188,6 +196,18 @@ func (c *meshCluster) reconcilePeers(peers []*registry.Peer) { alive[NodeID(p.NodeID)] = p.AdvertiseURL } c.mu.Lock() + // Log joins and leaves (as seen through the up-to-NodeTTL-stale registry view) + for nodeID, url := range alive { + if _, ok := c.knownPeers[nodeID]; !ok { + log.Tag(tag).Info("Peer %s (%s) joined the cluster", nodeID, url) + } + } + for nodeID := range c.knownPeers { + if _, ok := alive[nodeID]; !ok { + log.Tag(tag).Info("Peer %s left the cluster", nodeID) + } + } + c.knownPeers = alive for nodeID, q := range c.queues { if url, ok := alive[nodeID]; !ok || q.advertiseURL != url { q.queue.Close() // Flushes the remainder; the worker exits when the queue is drained @@ -227,11 +247,11 @@ func (c *meshCluster) queueFor(p *registry.Peer) *peerQueue { return q } -// Relay enqueues the message for delivery to every live peer node that may have subscribers for +// ForwardMessage enqueues the message for delivery to every live peer node that may have subscribers for // its topic (all of them, absent fresh knowledge). Delivery is fire-and-forget via each peer's // bounded batching queue; if a peer's queue is full the message is dropped for that peer // (subscribers reconnect and re-poll history from the database). -func (c *meshCluster) Relay(msg *model.Message) error { +func (c *meshCluster) ForwardMessage(msg *model.Message) error { peers, err := c.registry.Peers() if err != nil { return err @@ -243,7 +263,7 @@ func (c *meshCluster) Relay(msg *model.Message) error { if err != nil { return err } - metrics.ClusterMessagesRelayed.Inc() + metrics.ClusterMessagesForwarded.Inc() c.mu.Lock() defer c.mu.Unlock() if c.closed { @@ -254,11 +274,16 @@ func (c *meshCluster) Relay(msg *model.Message) error { // certain (no state, stale state) falls back to broadcasting if !c.mayNeed(NodeID(p.NodeID), msg.Topic) { metrics.ClusterRouteSkipped.Inc() + if ev := log.Tag(tag); ev.IsTrace() { + ev.Trace("Skipping peer %s for message %s: no subscribers for topic %s", p.NodeID, msg.ID, msg.Topic) + } continue } if !c.queueFor(p).queue.TryEnqueue(frag) { metrics.ClusterQueueDropped.Inc() log.Tag(tag).Warn("Fan-out queue for peer %s full, dropping message %s", p.NodeID, msg.ID) + } else if ev := log.Tag(tag); ev.IsTrace() { + ev.Trace("Enqueued message %s (topic %s) for peer %s", msg.ID, msg.Topic, p.NodeID) } } return nil @@ -284,7 +309,9 @@ func (c *meshCluster) mayNeed(peer NodeID, topic string) bool { func (c *meshCluster) peerWorker(nodeID NodeID, q *peerQueue) { defer c.wg.Done() for frags := range q.queue.Dequeue() { - c.postToPeer(nodeID, messageURL(q.advertiseURL), contentTypeNDJSON, assembleMessageBody(frags)) + body := assembleMessageBody(frags) + log.Tag(tag).Debug("Sending batch of %d message(s) (%d bytes) to peer %s", len(frags), len(body), nodeID) + c.postToPeer(nodeID, messageURL(q.advertiseURL), contentTypeNDJSON, body) metrics.ClusterBatchesSent.Inc() } } @@ -319,13 +346,22 @@ func (c *meshCluster) postToPeer(nodeID NodeID, url, contentType string, payload // handleMessage receives a batch of peer messages (NDJSON) and streams them to local // subscribers line by line, delivering each message as it is decoded. -func (c *meshCluster) handleMessage(_ NodeID, w http.ResponseWriter, r *http.Request) { +func (c *meshCluster) handleMessage(origin NodeID, w http.ResponseWriter, r *http.Request) { // A batch can exceed its byte cap by one message, plus framing overhead maxBodyBytes := int64(batchMaxBytes) + c.conf.MaxMessageBytes + 1024 - if err := decodeMessageBody(io.LimitReader(r.Body, maxBodyBytes), int(c.conf.MaxMessageBytes), c.deliver); err != nil { + received := 0 + deliver := func(m *model.Message) { + received++ + if ev := log.Tag(tag); ev.IsTrace() { + ev.Trace("Delivering message %s (topic %s) from peer %s", m.ID, m.Topic, origin) + } + c.deliver(m) + } + if err := decodeMessageBody(io.LimitReader(r.Body, maxBodyBytes), int(c.conf.MaxMessageBytes), deliver); err != nil { w.WriteHeader(http.StatusBadRequest) return } + log.Tag(tag).Debug("Received batch of %d message(s) from peer %s", received, origin) w.WriteHeader(http.StatusOK) } @@ -362,6 +398,7 @@ func (c *meshCluster) applyTopicState(origin NodeID, topics *apiStateTopics) err return err } c.states[origin] = &peerState{topics: filter, updatedAt: time.Now()} + log.Tag(tag).Debug("Received subscription state from peer %s (%d filter bytes)", origin, len(topics.Filter)) return nil } if state, ok := c.states[origin]; ok { @@ -369,6 +406,7 @@ func (c *meshCluster) applyTopicState(origin NodeID, topics *apiStateTopics) err state.topics.Add(topic) } state.updatedAt = time.Now() + log.Tag(tag).Debug("Received %d announced topic(s) from peer %s", len(topics.Added), origin) } return nil } @@ -394,27 +432,29 @@ func (c *meshCluster) pushState(peers []*registry.Peer) { if err != nil { return } + log.Tag(tag).Debug("Pushing subscription state (%d topics, %d bytes) to %d peer(s)", len(topics), len(body), len(peers)) for _, p := range peers { go c.postToPeer(NodeID(p.NodeID), stateURL(p.AdvertiseURL), contentTypeJSON, body) } metrics.ClusterStatePushes.Inc() } -// AnnounceTopics immediately tells all live peers that these topics gained their first local +// BroadcastState immediately tells all live peers that these topics gained their first local // subscriber, shrinking the window in which a publisher could wrongly skip this node from a // full state interval down to about one round trip. -func (c *meshCluster) AnnounceTopics(topics []string) { - if len(topics) == 0 { +func (c *meshCluster) BroadcastState(state *State) { + if len(state.AddedTopics) == 0 { return } peers, err := c.registry.Peers() if err != nil || len(peers) == 0 { return } - body, err := json.Marshal(&apiState{Topics: &apiStateTopics{Added: topics}}) + body, err := json.Marshal(&apiState{Topics: &apiStateTopics{Added: state.AddedTopics}}) if err != nil { return } + log.Tag(tag).Debug("Broadcasting state (%d new topics) to %d peer(s)", len(state.AddedTopics), len(peers)) for _, p := range peers { go c.postToPeer(NodeID(p.NodeID), stateURL(p.AdvertiseURL), contentTypeJSON, body) } @@ -425,6 +465,14 @@ func (c *meshCluster) IsLeader() bool { return c.leader.IsLeader() } +// Healthy reports whether this node's registry heartbeat is fresh enough that peers still +// forward messages to it (see the Cluster interface for the checker's fail-open duty). +func (c *meshCluster) Healthy() bool { + c.mu.Lock() + defer c.mu.Unlock() + return time.Since(c.lastRegistered) < c.conf.NodeTTL +} + // Close stops the mesh: it deregisters this node, releases leadership, stops all peer workers, // and waits for them to exit. func (c *meshCluster) Close() error { @@ -444,7 +492,7 @@ func (c *meshCluster) Close() error { if err := c.registry.Deregister(); err != nil { log.Tag(tag).Err(err).Warn("Failed to deregister node") } - c.leader.Release() + c.leader.Close() metrics.ClusterLeader.Set(0) return nil } diff --git a/cluster/cluster_mesh_test.go b/cluster/cluster_mesh_test.go index 49f19774..3af942bf 100644 --- a/cluster/cluster_mesh_test.go +++ b/cluster/cluster_mesh_test.go @@ -37,14 +37,15 @@ func openTestPool(t testing.TB, dsn string) *db.DB { func newTestMeshConfig(nodeID, advertiseURL string) *Config { return &Config{ - Enabled: true, - NodeID: NodeID(nodeID), - AdvertiseURL: advertiseURL, - Secret: testSecret, - HeartbeatInterval: 100 * time.Millisecond, - NodeTTL: time.Second, // Also the peer cache bound; short so fake peers registered mid-test are seen quickly - MaxMessageBytes: 1 << 20, - StateInterval: time.Minute, // Individual tests lower this to exercise state pushes + Enabled: true, + NodeID: NodeID(nodeID), + AdvertiseURL: advertiseURL, + Secret: testSecret, + HeartbeatInterval: 100 * time.Millisecond, + LeaderRenewInterval: 20 * time.Millisecond, // Lease duration 60ms, hold-off 120ms; keeps leadership tests fast + NodeTTL: time.Second, // Also the peer cache bound; short so fake peers registered mid-test are seen quickly + MaxMessageBytes: 1 << 20, + StateInterval: time.Minute, // Individual tests lower this to exercise state pushes } } @@ -92,7 +93,7 @@ func TestMesh_CrossNodeDelivery(t *testing.T) { require.Nil(t, err) defer meshA.Close() msg := model.NewDefaultMessage("mytopic", "hello cross-node") - require.Nil(t, meshA.Relay(msg)) + require.Nil(t, meshA.ForwardMessage(msg)) waitFor(t, func() bool { mu.Lock() defer mu.Unlock() @@ -207,7 +208,7 @@ func TestMesh_SlowPeerIsolation(t *testing.T) { defer mesh.Close() const n = 20 for i := 0; i < n; i++ { - require.Nil(t, mesh.Relay(model.NewDefaultMessage("mytopic", fmt.Sprintf("message %d", i)))) + require.Nil(t, mesh.ForwardMessage(model.NewDefaultMessage("mytopic", fmt.Sprintf("message %d", i)))) } waitFor(t, func() bool { mu.Lock() @@ -243,7 +244,7 @@ func TestMesh_BatchCoalescing(t *testing.T) { defer mesh.Close() const n = 20 for i := 0; i < n; i++ { - require.Nil(t, mesh.Relay(model.NewDefaultMessage("mytopic", fmt.Sprintf("message %d", i)))) + require.Nil(t, mesh.ForwardMessage(model.NewDefaultMessage("mytopic", fmt.Sprintf("message %d", i)))) } waitFor(t, func() bool { mu.Lock() @@ -281,7 +282,7 @@ func TestMesh_DeadPeerRemovedAndRejoin(t *testing.T) { defer mesh.Close() // The fake peer registers once and then "dies": its heartbeat is never refreshed registerFakePeer(t, pool, "node-dead", srv.URL) - require.Nil(t, mesh.Relay(model.NewDefaultMessage("mytopic", "while alive"))) + require.Nil(t, mesh.ForwardMessage(model.NewDefaultMessage("mytopic", "while alive"))) waitFor(t, func() bool { mu.Lock() defer mu.Unlock() @@ -299,7 +300,7 @@ func TestMesh_DeadPeerRemovedAndRejoin(t *testing.T) { require.Nil(t, pool.QueryRow(`SELECT COUNT(*) FROM node_registry WHERE node_id = 'node-dead'`).Scan(&count)) return count == 0 }) - require.Nil(t, mesh.Relay(model.NewDefaultMessage("mytopic", "while dead"))) + require.Nil(t, mesh.ForwardMessage(model.NewDefaultMessage("mytopic", "while dead"))) time.Sleep(250 * time.Millisecond) // Give a wrong implementation time to deliver anyway mu.Lock() require.Equal(t, 1, received) // Only the first message arrived @@ -308,15 +309,15 @@ func TestMesh_DeadPeerRemovedAndRejoin(t *testing.T) { // relay retries because the peer list is cached for up to the node TTL registerFakePeer(t, pool, "node-dead", srv.URL) waitFor(t, func() bool { - require.Nil(t, mesh.Relay(model.NewDefaultMessage("mytopic", "after rejoin"))) + require.Nil(t, mesh.ForwardMessage(model.NewDefaultMessage("mytopic", "after rejoin"))) mu.Lock() defer mu.Unlock() return received > 1 }) } -func TestMesh_RelayAfterClose(t *testing.T) { - // A Relay racing shutdown (e.g. an in-flight publish during server Stop) must not spawn +func TestMesh_ForwardAfterClose(t *testing.T) { + // A ForwardMessage racing shutdown (e.g. an in-flight publish during server Stop) must not spawn // a new peer queue and worker after Close: the worker would never exit (its queue is never // closed) and nothing waits for it. schemaDSN := dbtest.CreateTestPostgresSchema(t) @@ -325,7 +326,7 @@ func TestMesh_RelayAfterClose(t *testing.T) { require.Nil(t, err) registerFakePeer(t, pool, "node-peer", "http://127.0.0.1:1") require.Nil(t, mesh.Close()) - require.Nil(t, mesh.Relay(model.NewDefaultMessage("mytopic", "too late"))) // Dropped silently + require.Nil(t, mesh.ForwardMessage(model.NewDefaultMessage("mytopic", "too late"))) // Dropped silently mesh.mu.Lock() defer mesh.mu.Unlock() require.Empty(t, mesh.queues) @@ -418,13 +419,13 @@ func TestMesh_RouteSkipsUnsubscribedPeer(t *testing.T) { rr := postState(mesh, "node-b", &apiState{Topics: &apiStateTopics{Filter: topicFilter(t, "subscribed-topic")}}) require.Equal(t, 200, rr.Code) // A topic outside the peer's state is skipped - require.Nil(t, mesh.Relay(model.NewDefaultMessage("other-topic", "skipped"))) + require.Nil(t, mesh.ForwardMessage(model.NewDefaultMessage("other-topic", "skipped"))) time.Sleep(300 * time.Millisecond) // Give a wrong implementation time to deliver anyway mu.Lock() require.Equal(t, 0, received) mu.Unlock() // A topic inside the peer's state is delivered - require.Nil(t, mesh.Relay(model.NewDefaultMessage("subscribed-topic", "delivered"))) + require.Nil(t, mesh.ForwardMessage(model.NewDefaultMessage("subscribed-topic", "delivered"))) waitFor(t, func() bool { mu.Lock() defer mu.Unlock() @@ -457,7 +458,7 @@ func TestMesh_RouteBroadcastsOnStaleState(t *testing.T) { mesh.statesMu.Lock() mesh.states["node-b"].updatedAt = time.Now().Add(-time.Hour) mesh.statesMu.Unlock() - require.Nil(t, mesh.Relay(model.NewDefaultMessage("other-topic", "broadcast anyway"))) + require.Nil(t, mesh.ForwardMessage(model.NewDefaultMessage("other-topic", "broadcast anyway"))) waitFor(t, func() bool { mu.Lock() defer mu.Unlock() @@ -534,7 +535,7 @@ func TestMesh_AnnounceClosesWindow(t *testing.T) { return ok }) // Announcements merge into the baseline right away - meshA.AnnounceTopics([]string{"fresh-topic"}) + meshA.BroadcastState(&State{AddedTopics: []string{"fresh-topic"}}) waitFor(t, func() bool { meshB.statesMu.Lock() defer meshB.statesMu.Unlock() @@ -570,3 +571,20 @@ func TestMesh_StateOfDepartedPeerPruned(t *testing.T) { mesh.statesMu.Unlock() require.False(t, ok) } + +func TestMesh_HealthyReflectsRegistration(t *testing.T) { + schemaDSN := dbtest.CreateTestPostgresSchema(t) + pool := openTestPool(t, schemaDSN) + mesh, err := newMeshCluster(newTestMeshConfig("node-a", "http://127.0.0.1:1"), pool, nil, nil) + require.Nil(t, err) + defer mesh.Close() + require.True(t, mesh.Healthy()) // Registered synchronously at construction + // Stale heartbeat: peers stop forwarding to this node, so it must report unhealthy + mesh.mu.Lock() + mesh.lastRegistered = time.Now().Add(-2 * mesh.conf.NodeTTL) + mesh.mu.Unlock() + require.False(t, mesh.Healthy()) + // A successful heartbeat restores health + require.Nil(t, mesh.heartbeat()) + require.True(t, mesh.Healthy()) +} diff --git a/cluster/cluster_nop.go b/cluster/cluster_nop.go index 06231f52..a1e06505 100644 --- a/cluster/cluster_nop.go +++ b/cluster/cluster_nop.go @@ -11,14 +11,16 @@ import ( // no special-casing in single-node mode). type nopCluster struct{} -func (c *nopCluster) Relay(_ *model.Message) error { return nil } +func (c *nopCluster) ForwardMessage(_ *model.Message) error { return nil } func (c *nopCluster) ServeHTTP(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusNotFound) } -func (c *nopCluster) AnnounceTopics(_ []string) {} +func (c *nopCluster) BroadcastState(_ *State) {} func (c *nopCluster) IsLeader() bool { return true } +func (c *nopCluster) Healthy() bool { return true } + func (c *nopCluster) Close() error { return nil } diff --git a/cluster/cluster_test.go b/cluster/cluster_test.go index b5ae2290..12f7e23e 100644 --- a/cluster/cluster_test.go +++ b/cluster/cluster_test.go @@ -67,9 +67,10 @@ func TestNop(t *testing.T) { b, err := New(&Config{}, nil, nil, nil) // not enabled -> nop cluster, no database required require.Nil(t, err) require.IsType(t, &nopCluster{}, b) - require.Nil(t, b.Relay(model.NewDefaultMessage("mytopic", "hi"))) + require.Nil(t, b.ForwardMessage(model.NewDefaultMessage("mytopic", "hi"))) // A single node is trivially the leader, so leader-gated jobs run without special-casing require.True(t, b.IsLeader()) + require.True(t, b.Healthy()) rr := httptest.NewRecorder() b.ServeHTTP(rr, httptest.NewRequest("POST", MessagePath, nil)) require.Equal(t, 404, rr.Code) diff --git a/cluster/types.go b/cluster/types.go index b85247da..5a88bae3 100644 --- a/cluster/types.go +++ b/cluster/types.go @@ -7,6 +7,35 @@ import ( "heckel.io/ntfy/v2/util" ) +// Config configures the cluster. It is assembled by the server from its own config, which keeps +// this package free of server types. +type Config struct { + Enabled bool // Master switch; when false, New returns the nop cluster + NodeID NodeID // Stable per-node identifier; required + AdvertiseURL string // Base URL peers use to reach this node's fan-out endpoint + Secret string // Shared secret authenticating node-to-node fan-out requests + HeartbeatInterval time.Duration // How often the node registry heartbeat is refreshed + NodeTTL time.Duration // Registry rows older than this do not count as live peers + BatchLinger time.Duration // How long messages wait in a peer queue to form a batch; 0 = send immediately + StateInterval time.Duration // How often the full subscription state is pushed to peers + MaxMessageBytes int64 // Upper bound for a single message on the wire (batch limits derive from this) + LeaderRenewInterval time.Duration // Overrides the leader lease renewal cadence; tests only, 0 = default +} + +// DeliverFunc hands a message received from a peer node to this node's local subscribers. The +// server supplies it, which inverts the dependency: this package never imports the server. +type DeliverFunc func(m *model.Message) + +// State is a subscription-state delta for Cluster.BroadcastState. +type State struct { + AddedTopics []string // Topics that just gained their first local subscriber on this node +} + +// TopicsFunc returns the topics that currently have at least one live subscriber, computed +// fresh on every call: membership is never tracked as a list, so topics "leave" simply by not +// appearing in the next snapshot. The server supplies it (same inversion as DeliverFunc). +type TopicsFunc func() []string + // apiMessage is one line of a message request body (NDJSON: one message per line; a single // message is just a one-line body). It carries the two fields that model.Message does not // serialize to JSON (Sender and User), which are needed to reconstruct the visitor on the @@ -31,7 +60,7 @@ type apiStateTopics struct { Added []string `json:"added,omitempty"` } -// peerState is what a peer last told us about itself; Relay routes around peers whose +// peerState is what a peer last told us about itself; ForwardMessage routes around peers whose // fresh state provably excludes a topic. type peerState struct { topics *util.BloomFilter @@ -40,7 +69,7 @@ type peerState struct { // peerQueue is the bounded, batching send queue for a single peer, pinned to the advertise URL // the peer was created with: a peer re-registering under a different advertise URL is treated -// as a replacement (reconcile retires the old queue; Relay creates a fresh one on demand). +// as a replacement (reconcile retires the old queue; ForwardMessage creates a fresh one on demand). type peerQueue struct { advertiseURL string queue *util.LingerQueue[[]byte] // pre-marshaled apiMessage fragments diff --git a/db/pg/leader.go b/db/pg/leader.go index 8bf88369..f9338508 100644 --- a/db/pg/leader.go +++ b/db/pg/leader.go @@ -4,80 +4,160 @@ import ( "context" "database/sql" "sync" + "time" + + "heckel.io/ntfy/v2/log" ) const ( + tagLeader = "leader" + tryAdvisoryLockQuery = `SELECT pg_try_advisory_lock($1)` advisoryUnlockQuery = `SELECT pg_advisory_unlock($1)` + + defaultRenewInterval = 5 * time.Second + leaderMissedRenewals = 3 + leaderHoldoffFactor = 2 ) -// Leader implements singleton-job leader election via a Postgres advisory lock held on a pinned -// connection. The lock auto-releases if the holding connection dies, so a crashed leader is -// replaced without manual fencing or lease bookkeeping. Multiple independent leaderships can -// coexist by using distinct lock keys. +// Leader implements singleton-job leader election via a Postgres advisory lock held on a +// pinned connection. The lock auto-releases when the holding connection dies, so a crashed +// leader is replaced without manual fencing; distinct keys elect independently. The Leader +// renews its lease on its own loop; callers only ask IsLeader and eventually Close. +// +// Holding the lock is not the same as believing to be the leader: IsLeader also requires a +// recent renewal (lease duration) and a completed hold-off after winning the lock. The +// hold-off outlasts the lease duration by construction, so on failover the old belief always +// expires before the new one begins: a short no-leader gap, never two leaders. Defaults: +// renew every 5s, lease duration 15s, hold-off 30s -> up to ~35s without a leader. type Leader struct { - db *sql.DB - key int64 - conn *sql.Conn // holds the advisory lock while this process is leader - held bool - mu sync.Mutex // Protects conn and held + db *sql.DB + key int64 + renewInterval time.Duration + conn *sql.Conn // holds the advisory lock while this process is leader + acquiredAt time.Time // When the lock was won (this tenure), for the hold-off + renewedAt time.Time // Last successful renewal, for the lease duration; zero = lock not held + cancel context.CancelFunc // Stops the renew loop and aborts its in-flight query on Close + closeOnce sync.Once + wg sync.WaitGroup + mu sync.Mutex // Protects conn, acquiredAt and renewedAt } -// NewLeader creates a Leader competing for the advisory lock identified by key. It does not -// attempt to acquire the lock; call TryAcquire periodically. -func NewLeader(db *sql.DB, key int64) *Leader { - return &Leader{db: db, key: key} +// NewLeader creates a Leader competing for the lock identified by key and starts its renew +// loop. renewInterval is for tests; pass 0 for the default. +func NewLeader(db *sql.DB, key int64, renewInterval time.Duration) *Leader { + if renewInterval <= 0 { + renewInterval = defaultRenewInterval + } + ctx, cancel := context.WithCancel(context.Background()) + l := &Leader{ + db: db, + key: key, + renewInterval: renewInterval, + cancel: cancel, + } + l.wg.Add(1) + go l.runAcquireOrRenewLoop(ctx) + return l } -// TryAcquire attempts to grab (or confirm) the advisory lock on a pinned connection, without -// blocking. It returns whether this process is the leader after the attempt, i.e. it acquired -// the lock now or still holds it. It is meant to be called periodically; on a healthy leader it -// is a cheap ping, on a follower it retries the lock. -func (l *Leader) TryAcquire(ctx context.Context) bool { +// IsLeader reports whether this process should act as the leader: lock held, lease renewed +// recently, hold-off elapsed (see the Leader doc comment). +func (l *Leader) IsLeader() bool { l.mu.Lock() - conn, held := l.conn, l.held - l.mu.Unlock() - if held { - if conn != nil && conn.PingContext(ctx) == nil { - return true // Still leader, connection healthy + defer l.mu.Unlock() + leaseDuration := leaderMissedRenewals * l.renewInterval + holdoff := leaderHoldoffFactor * leaseDuration + return time.Since(l.renewedAt) < leaseDuration && time.Since(l.acquiredAt) >= holdoff +} + +// Close stops competing for leadership and releases the lock. Idempotent. +func (l *Leader) Close() { + l.closeOnce.Do(func() { + l.cancel() // Also aborts an in-flight renewal query + l.wg.Wait() + if l.IsLeader() { + log.Tag(tagLeader).Info("Lost leadership: closed (lock key %d)", l.key) } - l.Release() // Connection died; the lock is already gone, re-acquire below + l.release() + }) +} + +// runAcquireOrRenewLoop acquires or renews the lock every renewInterval until ctx is canceled +func (l *Leader) runAcquireOrRenewLoop(ctx context.Context) { + defer l.wg.Done() + ticker := time.NewTicker(l.renewInterval) + defer ticker.Stop() + wasLeader := false + for { + attemptCtx, cancel := context.WithTimeout(ctx, l.renewInterval) + l.tryAcquireOrRenew(attemptCtx) + cancel() + if isLeader := l.IsLeader(); isLeader != wasLeader { + wasLeader = isLeader + if isLeader { + log.Tag(tagLeader).Info("Became leader (lock key %d)", l.key) + } else { + log.Tag(tagLeader).Info("Lost leadership (lock key %d)", l.key) + } + } + select { + case <-ticker.C: + case <-ctx.Done(): + return + } + } +} + +// tryAcquireOrRenew renews the lock on a healthy leader (a cheap ping) or retries acquiring +// it on a follower, on a pinned connection. +func (l *Leader) tryAcquireOrRenew(ctx context.Context) { + l.mu.Lock() + conn := l.conn + l.mu.Unlock() + if conn != nil { + if conn.PingContext(ctx) == nil { + // Still holding the lock, connection healthy: renew the lease + l.mu.Lock() + l.renewedAt = time.Now() + l.mu.Unlock() + log.Tag(tagLeader).Trace("Renewed leader lease (lock key %d)", l.key) + return + } + log.Tag(tagLeader).Debug("Leader lock connection died, lock lost (lock key %d)", l.key) + l.release() // Connection died; the lock is already gone, re-acquire below } newConn, err := l.db.Conn(ctx) if err != nil { - return false + log.Tag(tagLeader).Debug("Cannot get connection to compete for leader lock (lock key %d): %s", l.key, err.Error()) + return } var acquired bool if err := newConn.QueryRowContext(ctx, tryAdvisoryLockQuery, l.key).Scan(&acquired); err != nil || !acquired { newConn.Close() - return false + log.Tag(tagLeader).Trace("Leader lock held elsewhere (lock key %d)", l.key) + return } + log.Tag(tagLeader).Debug("Acquired leader lock (lock key %d); leadership after the hold-off", l.key) l.mu.Lock() l.conn = newConn - l.held = true + l.acquiredAt = time.Now() + l.renewedAt = l.acquiredAt l.mu.Unlock() - return true } -// Release unlocks the advisory lock and returns the pinned connection to the pool. -func (l *Leader) Release() { +// release unlocks the advisory lock and returns the pinned connection to the pool +func (l *Leader) release() { l.mu.Lock() conn := l.conn l.conn = nil - l.held = false + l.renewedAt = time.Time{} // Zero revokes belief; without it, IsLeader would linger a lease duration l.mu.Unlock() if conn != nil { - // Unlock explicitly before returning the connection to the pool: sql.Conn.Close() returns - // the physical connection to the pool rather than closing it, so the session-scoped lock - // would otherwise stay held. + // Unlock explicitly: sql.Conn.Close() returns the connection to the pool, so the + // session-scoped lock would otherwise stay held conn.ExecContext(context.Background(), advisoryUnlockQuery, l.key) conn.Close() + log.Tag(tagLeader).Debug("Released leader lock (lock key %d)", l.key) } } - -// IsLeader reports whether this process currently holds the advisory lock. -func (l *Leader) IsLeader() bool { - l.mu.Lock() - defer l.mu.Unlock() - return l.held -} diff --git a/db/pg/leader_internal_test.go b/db/pg/leader_internal_test.go new file mode 100644 index 00000000..68881fb2 --- /dev/null +++ b/db/pg/leader_internal_test.go @@ -0,0 +1,41 @@ +package pg + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// The lease logic is pure time arithmetic, so it is unit-tested here without a database; the +// external leader tests cover the loop end to end. + +func TestLeader_Lease_HoldoffMeansNoLeaderRatherThanTwo(t *testing.T) { + // Freshly acquired lock: belief must wait out the hold-off + l := &Leader{renewInterval: 20 * time.Second} // Lease duration 1m, hold-off 2m + l.acquiredAt = time.Now() + l.renewedAt = l.acquiredAt + require.False(t, l.IsLeader()) + // Once the hold-off has passed (and verification is fresh), belief begins + l.acquiredAt = time.Now().Add(-3 * time.Minute) + l.renewedAt = time.Now() + require.True(t, l.IsLeader()) +} + +func TestLeader_Lease_ExpiredLeaseRevokesLeadership(t *testing.T) { + // A leader that cannot renew its lease (wedged process, long GC pause) must stop + // believing once the lease expires, even though the lock may still be held + l := &Leader{renewInterval: 20 * time.Second} // Lease duration 1m, hold-off 2m + l.acquiredAt = time.Now().Add(-time.Hour) + l.renewedAt = time.Now().Add(-2 * time.Minute) // Lease expired + require.False(t, l.IsLeader()) + l.renewedAt = time.Now() // Fresh renewal restores belief + require.True(t, l.IsLeader()) +} + +func TestLeader_Lease_ReleasedIsNeverLeader(t *testing.T) { + // release() zeroes renewedAt, which fails the lease check no matter how old the tenure + l := &Leader{renewInterval: 20 * time.Second} // Lease duration 1m, hold-off 2m + l.acquiredAt = time.Now().Add(-time.Hour) + require.False(t, l.IsLeader()) +} diff --git a/db/pg/leader_test.go b/db/pg/leader_test.go index 9f5ecbe2..e03f839e 100644 --- a/db/pg/leader_test.go +++ b/db/pg/leader_test.go @@ -1,7 +1,6 @@ package pg_test import ( - "context" "testing" "time" @@ -10,33 +9,33 @@ import ( dbtest "heckel.io/ntfy/v2/db/test" ) +const testRenewInterval = 20 * time.Millisecond // Lease duration 60ms, hold-off 120ms + func TestLeader_AcquireAndFailover(t *testing.T) { testDB := dbtest.CreateTestPostgres(t) // skips if NTFY_TEST_DATABASE_URL is unset const key = int64(42) - ctx := context.Background() - l1 := pg.NewLeader(testDB.Primary(), key) - l2 := pg.NewLeader(testDB.Primary(), key) - defer l1.Release() - defer l2.Release() - // First to try wins; the second stays follower - require.True(t, l1.TryAcquire(ctx)) - require.False(t, l2.TryAcquire(ctx)) + l1 := pg.NewLeader(testDB.Primary(), key, testRenewInterval) + defer l1.Close() + // Belief follows the hold-off, it is never instant + require.False(t, l1.IsLeader()) + waitForLeader(t, l1) + // A competitor never becomes leader while the leader lives + l2 := pg.NewLeader(testDB.Primary(), key, testRenewInterval) + defer l2.Close() + time.Sleep(300 * time.Millisecond) // Several verification rounds + require.False(t, l2.IsLeader()) require.True(t, l1.IsLeader()) - require.False(t, l2.IsLeader()) - // Repeated TryAcquire on the leader is a no-op ping and reports leadership - require.True(t, l1.TryAcquire(ctx)) - // Release -> the follower can take over - l1.Release() + // Close -> the follower takes over + l1.Close() + require.False(t, l1.IsLeader()) + waitForLeader(t, l2) require.False(t, l1.IsLeader()) - require.True(t, l2.TryAcquire(ctx)) - require.True(t, l2.IsLeader()) } func TestLeader_ConnectionLossFailover(t *testing.T) { - // A leader that dies without calling Release (crash, network loss) must not wedge the - // cluster: the advisory lock is session-scoped, so Postgres releases it when the pinned - // connection dies, and a follower can take over. Simulated by terminating the lock-holding - // backend server-side. + // A crashed leader must not wedge the cluster: Postgres releases the session-scoped lock + // when the pinned connection dies (simulated by terminating the backend), and someone + // re-acquires. Either node may win; the invariant is one leader eventually, never two. schemaDSN := dbtest.CreateTestPostgresSchema(t) hostA, err := pg.Open(schemaDSN) require.Nil(t, err) @@ -45,40 +44,47 @@ func TestLeader_ConnectionLossFailover(t *testing.T) { require.Nil(t, err) defer hostB.DB.Close() const key = int64(43) - ctx := context.Background() - l1 := pg.NewLeader(hostA.DB, key) - l2 := pg.NewLeader(hostB.DB, key) - defer l1.Release() - defer l2.Release() - require.True(t, l1.TryAcquire(ctx)) - require.False(t, l2.TryAcquire(ctx)) + l1 := pg.NewLeader(hostA.DB, key, testRenewInterval) + defer l1.Close() + waitForLeader(t, l1) + l2 := pg.NewLeader(hostB.DB, key, testRenewInterval) + defer l2.Close() // Kill the backend holding the lock (advisory lock keys map to classid/objid) _, err = hostB.DB.Exec(`SELECT pg_terminate_backend(pid) FROM pg_locks WHERE locktype = 'advisory' AND objid = $1 AND granted`, key) require.Nil(t, err) - // The lock is auto-released; the follower takes over - waitForCond(t, func() bool { return l2.TryAcquire(ctx) }) - // The old leader discovers its dead connection on the next attempt and stays follower - require.False(t, l1.TryAcquire(ctx)) -} - -func waitForCond(t *testing.T, f func() bool) { - t.Helper() - for i := 0; i < 100; i++ { - if f() { + // Eventually exactly one leader again, and never two along the way + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + leader1, leader2 := l1.IsLeader(), l2.IsLeader() + require.False(t, leader1 && leader2, "two leaders at once") + if leader1 != leader2 { return } - time.Sleep(20 * time.Millisecond) + time.Sleep(10 * time.Millisecond) } - t.Fatal("timed out waiting for condition") + t.Fatal("no leader re-emerged after connection loss") } func TestLeader_DistinctKeysAreIndependent(t *testing.T) { testDB := dbtest.CreateTestPostgres(t) - ctx := context.Background() - l1 := pg.NewLeader(testDB.Primary(), 1) - l2 := pg.NewLeader(testDB.Primary(), 2) - defer l1.Release() - defer l2.Release() - require.True(t, l1.TryAcquire(ctx)) - require.True(t, l2.TryAcquire(ctx)) // Different keys do not compete + l1 := pg.NewLeader(testDB.Primary(), 1, testRenewInterval) + defer l1.Close() + l2 := pg.NewLeader(testDB.Primary(), 2, testRenewInterval) + defer l2.Close() + // Different keys do not compete: both become effective leaders + waitForLeader(t, l1) + waitForLeader(t, l2) +} + +// waitForLeader waits until the node believes it is the leader, or fails the test +func waitForLeader(t *testing.T, l *pg.Leader) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if l.IsLeader() { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatal("node never became effective leader") } diff --git a/metrics/metrics.go b/metrics/metrics.go index 00b9741c..b81c6ce4 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -78,8 +78,8 @@ var ( ClusterPeers = prometheus.NewGauge(prometheus.GaugeOpts{ Name: "ntfy_cluster_peers", }) - ClusterMessagesRelayed = prometheus.NewCounter(prometheus.CounterOpts{ - Name: "ntfy_cluster_messages_relayed_total", + ClusterMessagesForwarded = prometheus.NewCounter(prometheus.CounterOpts{ + Name: "ntfy_cluster_messages_forwarded_total", }) ClusterSendErrors = prometheus.NewCounter(prometheus.CounterOpts{ Name: "ntfy_cluster_send_errors_total", @@ -131,7 +131,7 @@ func init() { Topics, HTTPRequests, ClusterPeers, - ClusterMessagesRelayed, + ClusterMessagesForwarded, ClusterSendErrors, ClusterQueueDropped, ClusterBatchesSent, diff --git a/metrics/metrics_test.go b/metrics/metrics_test.go index 6bc9e223..e913bd2d 100644 --- a/metrics/metrics_test.go +++ b/metrics/metrics_test.go @@ -17,7 +17,7 @@ var expectedMetricNames = []string{ "ntfy_calls_made_success", "ntfy_cluster_batches_sent_total", "ntfy_cluster_leader", - "ntfy_cluster_messages_relayed_total", + "ntfy_cluster_messages_forwarded_total", "ntfy_cluster_messages_wasted_total", "ntfy_cluster_peers", "ntfy_cluster_queue_dropped_total", diff --git a/server/server.go b/server/server.go index 27b8b1b1..8970e126 100644 --- a/server/server.go +++ b/server/server.go @@ -370,11 +370,14 @@ func New(conf *Config) (*Server, error) { // be reached from the outside even before any firewalling. func (s *Server) clusterHandler() http.Handler { mux := http.NewServeMux() - // TODO(T8): reflect s.cluster.Healthy() here (and in handleHealth) instead of a static - // response, so health checks can pull a node that lost its registry heartbeat mux.HandleFunc(apiHealthPath, func(w http.ResponseWriter, _ *http.Request) { w.Header().Set("Content-Type", "application/json") - io.WriteString(w, `{"healthy":true}`+"\n") + if s.cluster.Healthy() { + io.WriteString(w, `{"healthy":true}`+"\n") + } else { + w.WriteHeader(http.StatusServiceUnavailable) + io.WriteString(w, `{"healthy":false}`+"\n") + } }) mux.Handle("/", s.cluster) return mux @@ -396,15 +399,15 @@ func (s *Server) liveTopics() []string { } // topicAnnouncer returns the first-subscriber hook for a topic: it tells peer nodes right away -// that this node now wants messages for it (see Cluster.AnnounceTopics). +// that this node now wants messages for it (see Cluster.BroadcastState). func (s *Server) topicAnnouncer(id string) func() { return func() { - s.cluster.AnnounceTopics([]string{id}) + s.cluster.BroadcastState(&cluster.State{AddedTopics: []string{id}}) } } // 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.Relay: local +// 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. func (s *Server) deliverFromBus(m *model.Message) { @@ -829,12 +832,13 @@ func (s *Server) handleTopicAuth(w http.ResponseWriter, _ *http.Request, _ *visi } func (s *Server) handleHealth(w http.ResponseWriter, _ *http.Request, _ *visitor) error { - // TODO(T8): in cluster mode, reflect s.cluster.Healthy() so DNS/LB health checks stop - // routing NEW clients to a node whose peers no longer relay to it (see Cluster interface) - response := &apiHealthResponse{ - Healthy: true, + // Unhealthy = the registry heartbeat went stale and peers stopped forwarding to this + // node; 503 lets status-code LB checks pull it (checkers must fail open, see cluster.Cluster) + healthy := s.cluster.Healthy() + if !healthy { + w.WriteHeader(http.StatusServiceUnavailable) } - return s.writeJSON(w, response) + return s.writeJSON(w, &apiHealthResponse{Healthy: healthy}) } // handleMetrics returns Prometheus metrics. This endpoint is only called if enable-metrics is set, @@ -965,8 +969,8 @@ func (s *Server) dispatch(v *visitor, t *topic, m *model.Message, opts dispatchO return err } } - // Relay to peer cluster nodes, whose subscribers do not show up in this node's topics map - if err := s.cluster.Relay(m); err != nil { + // ForwardMessage to peer cluster nodes, whose subscribers do not show up in this node's topics map + if err := s.cluster.ForwardMessage(m); err != nil { logvm(v, m).Err(err).Warn("Cluster: unable to relay message to peer nodes") } // Fire the requested side-effect targets diff --git a/server/server_cluster_test.go b/server/server_cluster_test.go index 0daf20cf..9720e470 100644 --- a/server/server_cluster_test.go +++ b/server/server_cluster_test.go @@ -12,6 +12,7 @@ import ( "time" "github.com/stretchr/testify/require" + "heckel.io/ntfy/v2/cluster" dbtest "heckel.io/ntfy/v2/db/test" "heckel.io/ntfy/v2/model" "heckel.io/ntfy/v2/user" @@ -20,13 +21,14 @@ import ( // fakeCluster records relayed messages and topic announcements so tests can assert that every // publish path passes through the cluster exactly once, and that subscription hooks fire. type fakeCluster struct { - mu sync.Mutex - messages []*model.Message - announced []string - notLeader bool + mu sync.Mutex + messages []*model.Message + announced []string + notLeader bool + notHealthy bool } -func (b *fakeCluster) Relay(m *model.Message) error { +func (b *fakeCluster) ForwardMessage(m *model.Message) error { b.mu.Lock() defer b.mu.Unlock() b.messages = append(b.messages, m) @@ -35,10 +37,22 @@ func (b *fakeCluster) Relay(m *model.Message) error { func (b *fakeCluster) ServeHTTP(_ http.ResponseWriter, _ *http.Request) {} -func (b *fakeCluster) AnnounceTopics(topics []string) { +func (b *fakeCluster) BroadcastState(state *cluster.State) { b.mu.Lock() defer b.mu.Unlock() - b.announced = append(b.announced, topics...) + b.announced = append(b.announced, state.AddedTopics...) +} + +func (b *fakeCluster) Healthy() bool { + b.mu.Lock() + defer b.mu.Unlock() + return !b.notHealthy +} + +func (b *fakeCluster) setHealthy(healthy bool) { + b.mu.Lock() + defer b.mu.Unlock() + b.notHealthy = !healthy } func (b *fakeCluster) IsLeader() bool { @@ -67,7 +81,7 @@ func (b *fakeCluster) Announced() []string { return append([]string{}, b.announced...) } -func TestServer_Cluster_PublishRelaysOnce(t *testing.T) { +func TestServer_Cluster_PublishForwardsOnce(t *testing.T) { s := newTestServer(t, newTestConfig(t, "")) b := &fakeCluster{} s.cluster = b @@ -79,7 +93,7 @@ func TestServer_Cluster_PublishRelaysOnce(t *testing.T) { require.Equal(t, "hi there", messages[0].Message) } -func TestServer_Cluster_SyncEventRelays(t *testing.T) { +func TestServer_Cluster_SyncEventForwards(t *testing.T) { // Account sync events are delivered via the user's st_... sync topic; without relaying // them, cross-device account sync silently breaks when a user's devices land on different // cluster nodes. @@ -316,3 +330,24 @@ func TestServer_Cluster_FirebaseKeepaliverOnlyOnLeader(t *testing.T) { cl.setLeader(true) waitFor(t, func() bool { return len(sender.Messages()) > 0 }) } + +func TestServer_Cluster_HealthReflectsCluster(t *testing.T) { + // A node whose registry heartbeat went stale no longer receives forwarded messages, so + // health checks must pull it from rotation (the fail-open policy lives in the checker) + s := newTestServer(t, newTestConfig(t, "")) + cl := &fakeCluster{} + s.cluster = cl + rr := request(t, s, "GET", "/v1/health", "", nil) + require.Equal(t, 200, rr.Code) + require.Contains(t, rr.Body.String(), `"healthy":true`) + cl.setHealthy(false) + rr = request(t, s, "GET", "/v1/health", "", nil) + require.Equal(t, 503, rr.Code) + require.Contains(t, rr.Body.String(), `"healthy":false`) + // The cluster listener's health endpoint reflects the same state + rr2 := httptest.NewRecorder() + req, err := http.NewRequest("GET", "/v1/health", nil) + require.Nil(t, err) + s.clusterHandler().ServeHTTP(rr2, req) + require.Equal(t, 503, rr2.Code) +}