Implement the peer mesh behind the cluster interface

The cluster package now does the thing the seam described. Nothing constructs
it: the server still holds no cluster, and cmd still refuses to start with the
options set, so this is inert like the pieces before it.

A publish marshals once and drops the line into one bounded queue per peer. A
worker batches what it finds, up to the batch linger and flushed early at 100
messages or 256KB, and POSTs NDJSON to that peer's /v1/cluster/message. Fire
and forget: a wedged peer backs up only its own queue, and a full queue drops
rather than blocking the caller. Inbound batches are authenticated by shared
secret, carry an origin header for loop prevention, and are decoded line by
line into the server's delivery callback.

The heartbeat loop refreshes this node's registry row, reads the live set back
and reconciles the per-peer queues, so a departing peer's queue is retired and
a new one is reachable on the next publish. Members reports this node plus its
peers, with peer health being the freshness of this node's own registry view
(registry.Fresh, added here), because an isolated node answering from a frozen
view must not be mistaken for current membership. A node that lost the
database while a peer is still healthy calls back to close its subscribers, and
keeps them when no peer is healthy, since that is an outage rather than
isolation.

Delivery gaps are not here: a failed batch is logged and counted. They come
with the recovery mechanism that uses them.

cluster/cluster.go

@@ -2,10 +2,6 @@
 // register themselves in a PostgreSQL node registry (control plane) and fan published messages
 // out to each other directly over HTTP (data plane); PostgreSQL is never on the message path.
 // The single-node default is the nop cluster, which does nothing.
-//
-// This is the seam only: the interface the server talks to, the single-node default behind it,
-// and the message wire format. The peer mesh that implements the interface for real is the next
-// piece, so on a single node this package is the nop and nothing else.
 package cluster
 
 import (
@@ -42,6 +38,24 @@ const (
 // IS a node, which is why peer values carry a NodeID.
 type NodeID string
 
+const (
+	// secretHeader carries the shared secret authenticating node-to-node fan-out requests.
+	secretHeader = "X-Cluster-Secret"
+
+	// originHeader carries the sending node's ID on fan-out requests, so a node can skip
+	// requests that carry its own broadcasts (loop prevention).
+	originHeader = "X-Cluster-Origin"
+)
+
+// Content types of the peer API: message bodies are NDJSON (one JSON message per line, matching
+// the framing of ntfy's own /topic/json subscribe stream), state bodies are plain JSON. Future
+// node-to-node request types get their own paths on the cluster listener; an old node answering
+// 404 on an unknown path keeps mixed-version clusters working during rolling deploys.
+const (
+	contentTypeNDJSON = "application/x-ndjson"
+	contentTypeJSON   = "application/json"
+)
+
 // tag is the log tag for everything cluster-related, so "tag=cluster -> trace" turns on
 // per-message decisions without raising the level anywhere else
 const tag = "cluster"
@@ -108,8 +122,5 @@ func New(conf *Config, pool *db.DB, deliver DeliverFunc) (Cluster, error) {
 	if conf.NodeTTL == 0 {
 		conf.NodeTTL = defaultNodeTTL
 	}
-	// Everything above is the config contract the mesh relies on. The mesh itself is the next
-	// piece, and cmd refuses to start a server with experimental-cluster-listen set, so nothing
-	// but a test reaches this line.
-	return nil, errors.New("clustering is not implemented in this build")
+	return newMeshCluster(conf, pool, deliver)
 }

cluster/cluster_load_test.go

@@ -0,0 +1,133 @@
+package cluster
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	dbtest "heckel.io/ntfy/v2/db/test"
+	"heckel.io/ntfy/v2/model"
+)
+
+// TestMesh_Soak floods the mesh with concurrent publishers and asserts exact delivery: every
+// message reaches the peer exactly once, nothing is dropped, and batching keeps the request
+// count far below the message count. Skipped unless NTFY_TEST_SOAK is set (it takes a few
+// seconds and is meant for pre-deploy verification, not the regular suite).
+func TestMesh_Soak(t *testing.T) {
+	if os.Getenv("NTFY_TEST_SOAK") == "" {
+		t.Skip("NTFY_TEST_SOAK not set")
+	}
+	// ~1000 msg/s aggregate (10x the ntfy.sh peak of ~88 msg/s): each publisher paces itself to
+	// 100 msg/s. Unthrottled publishing intentionally overruns the bounded per-peer queue (load
+	// shedding by design), so a zero-drop assertion only holds below the drain ceiling.
+	const (
+		publishers           = 10
+		messagesPerPublisher = 300
+		publishInterval      = 10 * time.Millisecond
+		total                = publishers * messagesPerPublisher
+	)
+	schemaDSN := dbtest.CreateTestPostgresSchema(t)
+	pool := openTestPool(t, schemaDSN)
+	var mu sync.Mutex
+	received := make(map[string]int, total) // message body -> count, to catch duplicates
+	requests := 0
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, err := io.ReadAll(r.Body)
+		require.Nil(t, err)
+		messages, err := unmarshalMessageBody(body, 1<<20)
+		require.Nil(t, err)
+		mu.Lock()
+		requests++
+		for _, m := range messages {
+			received[m.Message]++
+		}
+		mu.Unlock()
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer srv.Close()
+	conf := newTestMeshConfig("node-a", "http://127.0.0.1:1")
+	conf.BatchLinger = 50 * time.Millisecond
+	conf.NodeTTL = time.Minute // The fake peer never heartbeats; liveness is not under test here
+	registerFakePeer(t, pool, "node-peer", srv.URL)
+	mesh, err := newMeshCluster(conf, pool, nil)
+	require.Nil(t, err)
+	defer mesh.Close()
+	start := time.Now()
+	var wg sync.WaitGroup
+	for p := 0; p < publishers; p++ {
+		wg.Add(1)
+		go func(p int) {
+			defer wg.Done()
+			ticker := time.NewTicker(publishInterval)
+			defer ticker.Stop()
+			for i := 0; i < messagesPerPublisher; i++ {
+				require.Nil(t, mesh.ForwardMessage(model.NewDefaultMessage("mytopic", fmt.Sprintf("p%d-m%d", p, i))))
+				<-ticker.C
+			}
+		}(p)
+	}
+	wg.Wait()
+	waitFor(t, func() bool {
+		mu.Lock()
+		defer mu.Unlock()
+		return len(received) == total
+	})
+	elapsed := time.Since(start)
+	mu.Lock()
+	defer mu.Unlock()
+	for body, count := range received {
+		require.Equalf(t, 1, count, "message %s delivered %d times", body, count)
+	}
+	require.Less(t, requests, total/10, "expected strong batching under load")
+	t.Logf("soak: %d messages, %d requests (%.1f msgs/request), %.0f msgs/s",
+		total, requests, float64(total)/float64(requests), float64(total)/elapsed.Seconds())
+}
+
+// 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 BenchmarkForwardMessage(b *testing.B) {
+	if os.Getenv("NTFY_TEST_DATABASE_URL") == "" {
+		b.Skip("NTFY_TEST_DATABASE_URL not set")
+	}
+	schemaDSN := dbtest.CreateTestPostgresSchema(b)
+	pool := openTestPool(b, schemaDSN)
+	conf := newTestMeshConfig("node-a", "http://127.0.0.1:1")
+	conf.BatchLinger = time.Minute // Never flush; we measure enqueue only
+	mesh, err := newMeshCluster(conf, pool, nil)
+	require.Nil(b, err)
+	defer mesh.Close()
+	registerFakePeer(b, pool, "node-peer", "http://127.0.0.1:1")
+	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.ForwardMessage(m); err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+// BenchmarkDecodeFanout measures the receive-path cost of decoding a 100-message NDJSON body.
+func BenchmarkDecodeFanout(b *testing.B) {
+	frags := make([][]byte, 100)
+	for i := range frags {
+		data, err := marshalMessage(model.NewDefaultMessage("mytopic", fmt.Sprintf("benchmark message %d", i)))
+		require.Nil(b, err)
+		frags[i] = data
+	}
+	body := assembleMessageBody(frags)
+	b.SetBytes(int64(len(body)))
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		messages, err := unmarshalMessageBody(body, 1<<20)
+		if err != nil || len(messages) != 100 {
+			b.Fatal("decode failed")
+		}
+	}
+}

cluster/cluster_mesh.go

@@ -0,0 +1,504 @@
+package cluster
+
+import (
+	"bytes"
+	"context"
+	"crypto/subtle"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"sync"
+	"time"
+
+	"heckel.io/ntfy/v2/cluster/registry"
+	"heckel.io/ntfy/v2/db"
+	"heckel.io/ntfy/v2/log"
+	"heckel.io/ntfy/v2/metrics"
+	"heckel.io/ntfy/v2/model"
+	"heckel.io/ntfy/v2/util"
+)
+
+const (
+	meshHTTPTimeout   = 5 * time.Second
+	peerHealthTimeout = 2 * time.Second // Per-peer health probe while this node is unhealthy
+	peerQueueSize     = 1024            // Bounded per-peer fan-out queue (drop on overflow)
+	batchMaxMessages  = 100             // Flush a batch early when it reaches this many messages
+	batchMaxBytes     = 256 * 1024      // Flush a batch early when it reaches this size
+	stateMaxBytes     = 1024 * 1024     // Upper bound for inbound state bodies (announcements, cancels)
+)
+
+// meshCluster fans messages out directly to peer nodes over HTTP (the data plane), using
+// PostgreSQL only as a control plane: the node_registry table for membership/discovery, and a
+// Postgres advisory lock for singleton-job leader election. The message path never touches the
+// database (not even for membership: see registry.Peers).
+//
+// A published message goes to EVERY live peer, which then drops it unless it has a subscriber
+// for the topic. Routing by subscription knowledge was tried and removed: it made delivery
+// correctness depend on the ordering of state messages, and at the cluster sizes this is built
+// for the saving is a fraction of a few hundred KB/s (see docs/10-cluster.md).
+//
+// 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
+	registry       *registry.Registry
+	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
+	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
+}
+
+// newMeshCluster creates the mesh cluster: it sets up the registry schema, registers this node
+// (synchronously, so it is discoverable before New returns), and starts the heartbeat loop.
+// Peer delivery workers are started lazily as peers appear in the registry.
+func newMeshCluster(conf *Config, pool *db.DB, deliver DeliverFunc) (*meshCluster, error) {
+	reg, err := registry.New(pool, string(conf.NodeID), conf.AdvertiseURL, conf.NodeTTL)
+	if err != nil {
+		return nil, err
+	}
+	// Register synchronously so the node is discoverable before the constructor returns; the
+	// heartbeat loop refreshes the registration from here on. The first peer snapshot is taken
+	// here too, so fan-out works from the first publish rather than the first heartbeat.
+	if err := reg.Register(); err != nil {
+		return nil, err
+	}
+	if _, err := reg.Refresh(); err != nil {
+		return nil, err
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	c := &meshCluster{
+		conf:           conf,
+		deliver:        deliver,
+		registry:       reg,
+		httpClient:     &http.Client{Timeout: meshHTTPTimeout},
+		queues:         make(map[NodeID]*peerQueue),
+		lastRegistered: time.Now(), // The synchronous Register above just succeeded
+		knownPeers:     make(map[NodeID]string),
+		ctx:            ctx,
+		cancel:         cancel,
+	}
+	c.mux = http.NewServeMux()
+	c.mux.HandleFunc("POST "+MessagePath, c.authenticated(c.handleMessage))
+	c.mux.HandleFunc("POST "+StatePath, c.authenticated(c.handleState))
+	c.mux.HandleFunc("GET "+MembersPath, c.secretAuthenticated(c.handleMembers))
+	c.wg.Add(2)
+	go c.heartbeatLoop()
+	go c.isolationLoop()
+	return c, nil
+}
+
+// ServeHTTP serves the internal peer API. Auth lives in the authenticated middleware, so every
+// endpoint gets the same shared-secret and origin handling.
+func (c *meshCluster) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	c.mux.ServeHTTP(w, r)
+}
+
+// authenticated wraps a peer API handler with the checks every endpoint needs: the shared
+// secret (constant-time compare, rejected before any body is read), a present origin, and the
+// origin self-skip (a request carrying this node's own traffic is acknowledged but ignored).
+func (c *meshCluster) authenticated(h func(origin NodeID, w http.ResponseWriter, r *http.Request)) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		if c.conf.Secret == "" || subtle.ConstantTimeCompare([]byte(r.Header.Get(secretHeader)), []byte(c.conf.Secret)) != 1 {
+			w.WriteHeader(http.StatusUnauthorized)
+			return
+		}
+		origin := NodeID(r.Header.Get(originHeader))
+		if origin == "" {
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+		if origin == c.conf.NodeID {
+			w.WriteHeader(http.StatusOK) // Our own traffic; nothing to do
+			return
+		}
+		h(origin, w, r)
+	}
+}
+
+// secretAuthenticated wraps a handler with the shared-secret check only. Unlike authenticated
+// it expects no origin node, because the callers are the load balancers' agents, not peers.
+func (c *meshCluster) secretAuthenticated(h http.HandlerFunc) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		if c.conf.Secret == "" || subtle.ConstantTimeCompare([]byte(r.Header.Get(secretHeader)), []byte(c.conf.Secret)) != 1 {
+			w.WriteHeader(http.StatusUnauthorized)
+			return
+		}
+		h(w, r)
+	}
+}
+
+// handleMembers lists the live cluster members for the load balancers' agents
+func (c *meshCluster) handleMembers(w http.ResponseWriter, _ *http.Request) {
+	w.Header().Set("Content-Type", contentTypeJSON)
+	if err := json.NewEncoder(w).Encode(c.Members()); err != nil {
+		log.Tag(tag).Err(err).Warn("Cannot write member list")
+	}
+}
+
+// Members returns this node plus the peers the registry currently considers live, for the LB
+// agents. A peer's health here is this node's knowledge of it, not a probe: it is healthy if the
+// registry view is still fresh, and unknown (reported unhealthy) once this node can no longer
+// read the registry. An isolated node therefore reports nobody as healthy, including itself,
+// which is what an agent needs to disregard its answer and keep the upstreams it has.
+func (c *meshCluster) Members() []Member {
+	fresh := c.registry.Fresh()
+	members := []Member{{NodeID: c.conf.NodeID, AdvertiseURL: c.conf.AdvertiseURL, Healthy: c.Healthy()}}
+	for _, p := range c.registry.Peers() {
+		members = append(members, Member{NodeID: NodeID(p.NodeID), AdvertiseURL: p.AdvertiseURL, Healthy: fresh})
+	}
+	return members
+}
+
+// heartbeatLoop runs one heartbeat immediately (the ticker first fires a full interval after
+// startup, and a fresh node should be leader-capable and state-visible right away), then one per
+// interval until shutdown.
+func (c *meshCluster) heartbeatLoop() {
+	defer c.wg.Done()
+	ticker := time.NewTicker(c.conf.HeartbeatInterval)
+	defer ticker.Stop()
+	if err := c.heartbeat(); err != nil {
+		log.Tag(tag).Err(err).Warn("Cluster heartbeat failed")
+	}
+	for {
+		select {
+		case <-c.ctx.Done():
+			return
+		case <-ticker.C:
+			if err := c.heartbeat(); err != nil {
+				log.Tag(tag).Err(err).Warn("Cluster heartbeat failed")
+			}
+		}
+	}
+}
+
+// isolationLoop runs the isolation check on its own ticker: a partitioned database can make
+// heartbeat calls hang for a while, and Healthy is time-based, so this still notices
+func (c *meshCluster) isolationLoop() {
+	defer c.wg.Done()
+	ticker := time.NewTicker(c.conf.HeartbeatInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-c.ctx.Done():
+			return
+		case <-ticker.C:
+			c.maybeIsolated()
+		}
+	}
+}
+
+// maybeIsolated calls IsolatedFunc while this node's registration is stale (peers no longer
+// forward to it) but at least one known peer is healthy. With no healthy peer (e.g. a full
+// database outage) nothing happens: the mesh keeps delivering on its cached peer view.
+func (c *meshCluster) maybeIsolated() {
+	if c.conf.IsolatedFunc == nil || c.Healthy() {
+		return
+	}
+	c.mu.Lock()
+	urls := make([]string, 0, len(c.knownPeers))
+	for _, url := range c.knownPeers {
+		urls = append(urls, url)
+	}
+	c.mu.Unlock()
+	for _, url := range urls {
+		if c.peerHealthy(url) {
+			log.Tag(tag).Warn("This node lost its cluster registration while peer %s is healthy; closing local subscribers so they reconnect elsewhere", url)
+			c.conf.IsolatedFunc()
+			return
+		}
+	}
+}
+
+func (c *meshCluster) peerHealthy(url string) bool {
+	ctx, cancel := context.WithTimeout(c.ctx, peerHealthTimeout)
+	defer cancel()
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url+HealthPath, nil)
+	if err != nil {
+		return false
+	}
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		return false
+	}
+	resp.Body.Close()
+	return resp.StatusCode == http.StatusOK
+}
+
+// heartbeat is one control-plane tick: refresh this node's registry row, retry/confirm the
+// leader lock, prune long-dead registry rows (as leader), and reconcile the per-peer queues.
+//
+// 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 --
+// ForwardMessage keeps serving the last peer snapshot.
+func (c *meshCluster) heartbeat() error {
+	if err := c.registry.Register(); err != nil {
+		return err
+	}
+	c.mu.Lock()
+	c.lastRegistered = time.Now()
+	c.mu.Unlock()
+	// Effective leadership: the lowest live node id, with a hold-off that guarantees a
+	// no-leader gap on failover rather than two leaders (see registry.IsLeader)
+	if c.registry.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
+		}
+	} else {
+		metrics.ClusterLeader.Set(0)
+	}
+	// A fresh read (not the cached view): heartbeat cadence bounds how long a newly joined
+	// peer can go unseen; ForwardMessage keeps using the cache on the hot path
+	peers, err := c.registry.Refresh()
+	if err != nil {
+		return err
+	}
+	c.reconcilePeers(peers)
+	return nil
+}
+
+// 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). 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)))
+	alive := make(map[NodeID]string, len(peers)) // node ID -> advertise URL
+	for _, p := range peers {
+		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
+			delete(c.queues, nodeID)
+		}
+	}
+	c.mu.Unlock()
+}
+
+// queueFor returns the send queue for the given peer, creating it (and its delivery worker) if it
+// does not exist yet. The caller must hold c.mu.
+func (c *meshCluster) queueFor(p *registry.Peer) *peerQueue {
+	nodeID := NodeID(p.NodeID)
+	q, ok := c.queues[nodeID]
+	if ok {
+		return q
+	}
+	q = &peerQueue{
+		advertiseURL: p.AdvertiseURL,
+		queue: util.NewLingerQueue(peerQueueSize, batchMaxMessages, batchMaxBytes,
+			func(line []byte) int { return len(line) }, c.conf.BatchLinger),
+	}
+	c.queues[nodeID] = q
+	c.wg.Add(1)
+	go c.peerWorker(nodeID, q)
+	return q
+}
+
+// ForwardMessage enqueues the message for delivery to every live peer node; a peer without
+// subscribers for the topic drops it on arrival. 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) ForwardMessage(msg *model.Message) error {
+	peers := c.registry.Peers() // The heartbeat's snapshot; never a database call on the publish path
+	if len(peers) == 0 {
+		return nil // Cluster of one; skip the marshal
+	}
+	data, err := marshalMessage(msg)
+	if err != nil {
+		return err
+	}
+
+	metrics.ClusterMessagesForwarded.Inc()
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.closed {
+		return nil // Shutting down; the message is dropped like any other in-flight fan-out
+	}
+	for _, p := range peers {
+		if !c.queueFor(p).queue.TryEnqueue(data) {
+			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
+}
+
+// peerWorker delivers batches of queued fan-out messages to a single peer. Batches form in the
+// peer's LingerQueue (up to BatchLinger delay, flushed early on size/count caps); the worker
+// exits when the queue is closed (peer left the registry, or mesh shutdown) and drained.
+//
+// A batch that cannot be delivered is dropped: recovering it is the delivery-gap mechanism, which
+// is not in this piece yet, so a failed batch is counted and logged and nothing else.
+func (c *meshCluster) peerWorker(nodeID NodeID, q *peerQueue) {
+	defer c.wg.Done()
+	for lines := range q.queue.Dequeue() {
+		body := assembleMessageBody(lines)
+		log.Tag(tag).Debug("Sending batch of %d message(s) (%d bytes) to peer %s", len(lines), len(body), nodeID)
+		if err := c.postToPeer(nodeID, messageURL(q.advertiseURL), contentTypeNDJSON, body); err != nil {
+			log.Tag(tag).Err(err).Warn("Batch to peer %s was lost", nodeID)
+		}
+		metrics.ClusterBatchesSent.Inc()
+	}
+}
+
+// postToPeer POSTs a peer API payload, authenticated with the shared cluster secret. Failures
+// are logged, counted and returned; the caller decides what a failure means (a message batch
+// is dropped, state is retried on the next heartbeat).
+func (c *meshCluster) postToPeer(nodeID NodeID, url, contentType string, payload []byte) error {
+	req, err := http.NewRequestWithContext(c.ctx, http.MethodPost, url, bytes.NewReader(payload))
+	if err != nil {
+		metrics.ClusterSendErrors.Inc()
+		log.Tag(tag).Err(err).Warn("Failed to build request for peer %s", nodeID)
+		return err
+	}
+	req.Header.Set("Content-Type", contentType)
+	req.Header.Set(secretHeader, c.conf.Secret)
+	req.Header.Set(originHeader, string(c.conf.NodeID))
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		if c.ctx.Err() == nil {
+			metrics.ClusterSendErrors.Inc()
+			log.Tag(tag).Err(err).Warn("Failed to send to peer %s (%s)", nodeID, url)
+		}
+		return err
+	}
+	resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		metrics.ClusterSendErrors.Inc()
+		log.Tag(tag).Warn("Peer %s (%s) rejected request with HTTP %d", nodeID, url, resp.StatusCode)
+		return fmt.Errorf("peer %s rejected request with HTTP %d", nodeID, resp.StatusCode)
+	}
+	return nil
+}
+
+// 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(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
+	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)
+}
+
+// handleState receives a peer's state envelope and applies each section it carries.
+func (c *meshCluster) handleState(origin NodeID, w http.ResponseWriter, r *http.Request) {
+	body, err := io.ReadAll(io.LimitReader(r.Body, stateMaxBytes))
+	if err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+	var state apiState
+	if err := json.Unmarshal(body, &state); err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+	if state.Topics != nil && len(state.Topics.Added) > 0 && c.conf.TopicsAddedFunc != nil {
+		log.Tag(tag).Debug("Received %d announced topic(s) from peer %s", len(state.Topics.Added), origin)
+		c.conf.TopicsAddedFunc(state.Topics.Added)
+	}
+	if len(state.Cancels) > 0 && c.conf.CancelFunc != nil {
+		log.Tag(tag).Debug("Received %d subscriber cancel(s) from peer %s", len(state.Cancels), origin)
+		for _, cancel := range state.Cancels {
+			c.conf.CancelFunc(cancel)
+		}
+	}
+	w.WriteHeader(http.StatusOK)
+}
+
+// BroadcastState tells all live peers about state changes that must not wait: topics that
+// gained their first local subscriber (a hint for peers' cached rate-visitor lookups, never
+// used for routing) and subscriber cancels.
+func (c *meshCluster) BroadcastState(state *State) {
+	if len(state.AddedTopics) == 0 && len(state.SubscriberCancels) == 0 {
+		return
+	}
+	peers := c.registry.Peers()
+	if len(peers) == 0 {
+		return
+	}
+	envelope := &apiState{Cancels: state.SubscriberCancels}
+	if len(state.AddedTopics) > 0 {
+		envelope.Topics = &apiStateTopics{Added: state.AddedTopics}
+	}
+	body, err := json.Marshal(envelope)
+	if err != nil {
+		return
+	}
+	log.Tag(tag).Debug("Broadcasting state (%d new topics, %d cancels) to %d peer(s)", len(state.AddedTopics), len(state.SubscriberCancels), len(peers))
+	for _, p := range peers {
+		go c.postToPeer(NodeID(p.NodeID), stateURL(p.AdvertiseURL), contentTypeJSON, body)
+	}
+}
+
+// IsLeader reports whether this node currently holds singleton-job leadership.
+func (c *meshCluster) IsLeader() bool {
+	return c.registry.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 {
+	c.cancel() // Stops the heartbeat loop and aborts in-flight peer deliveries
+	// Close the peer queues so their workers flush and exit; final sends are best-effort since
+	// the context is already canceled (parity with fire-and-forget delivery)
+	c.mu.Lock()
+	c.closed = true
+	for nodeID, q := range c.queues {
+		q.queue.Close()
+		delete(c.queues, nodeID)
+	}
+	c.mu.Unlock()
+	// Wait for the loops BEFORE deregistering: an in-flight heartbeat's Register would otherwise
+	// re-insert our row right after Deregister deleted it
+	c.wg.Wait()
+	if err := c.registry.Deregister(); err != nil {
+		log.Tag(tag).Err(err).Warn("Failed to deregister node") // Its row goes stale on its own
+	}
+	metrics.ClusterLeader.Set(0)
+	return nil
+}

cluster/cluster_mesh_test.go

@@ -0,0 +1,721 @@
+package cluster
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"heckel.io/ntfy/v2/cluster/registry"
+	"heckel.io/ntfy/v2/db"
+	"heckel.io/ntfy/v2/db/pg"
+	dbtest "heckel.io/ntfy/v2/db/test"
+	"heckel.io/ntfy/v2/model"
+)
+
+const (
+	testSecret = "s3cret"
+)
+
+// openTestPool opens a dedicated connection pool to the given test schema, so that each simulated
+// node has its own pool like real nodes would.
+func openTestPool(t testing.TB, dsn string) *db.DB {
+	host, err := pg.Open(dsn)
+	require.Nil(t, err)
+	d := db.New(host, nil)
+	t.Cleanup(func() { d.Close() })
+	return d
+}
+
+func newTestMeshConfig(nodeID, advertiseURL string) *Config {
+	return &Config{
+		Enabled:           true,
+		NodeID:            NodeID(nodeID),
+		AdvertiseURL:      advertiseURL,
+		Secret:            testSecret,
+		HeartbeatInterval: 100 * time.Millisecond,
+		// The liveness window AND the leadership hold-off. One second is the floor: heartbeats
+		// are stored in whole seconds, so a shorter TTL truncates the SQL cutoff to zero and a
+		// peer counts as live only inside the wall-clock second it registered in, which makes
+		// every test that waits for a peer to appear or leave a coin flip
+		NodeTTL:         time.Second,
+		MaxMessageBytes: 1 << 20,
+	}
+}
+
+// registerFakePeer registers a fake peer via the registry (creating the table if the mesh has
+// not been constructed yet): tests register fakes before the mesh boots, since its first
+// heartbeat caches the peer list. The fake never refreshes its heartbeat.
+func registerFakePeer(t testing.TB, pool *db.DB, nodeID NodeID, url string) {
+	t.Helper()
+	reg, err := registry.New(pool, string(nodeID), url, time.Minute)
+	require.Nil(t, err)
+	require.Nil(t, reg.Register())
+}
+
+func waitFor(t *testing.T, f func() bool) {
+	t.Helper()
+	for i := 0; i < 100; i++ {
+		if f() {
+			return
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+	t.Fatal("timed out waiting for condition")
+}
+
+func TestMesh_CrossNodeDelivery(t *testing.T) {
+	schemaDSN := dbtest.CreateTestPostgresSchema(t)
+	poolA, poolB := openTestPool(t, schemaDSN), openTestPool(t, schemaDSN)
+	var mu sync.Mutex
+	var received []*model.Message
+	var meshB *meshCluster
+	srvB := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		meshB.ServeHTTP(w, r)
+	}))
+	defer srvB.Close()
+	meshB, err := newMeshCluster(newTestMeshConfig("node-b", srvB.URL), poolB, func(m *model.Message) {
+		mu.Lock()
+		defer mu.Unlock()
+		received = append(received, m)
+	})
+	require.Nil(t, err)
+	defer meshB.Close()
+	meshA, err := newMeshCluster(newTestMeshConfig("node-a", "http://127.0.0.1:1"), poolA, func(m *model.Message) {
+		t.Error("node A must not receive its own relayed message")
+	})
+	require.Nil(t, err)
+	defer meshA.Close()
+	msg := model.NewDefaultMessage("mytopic", "hello cross-node")
+	require.Nil(t, meshA.ForwardMessage(msg))
+	waitFor(t, func() bool {
+		mu.Lock()
+		defer mu.Unlock()
+		return len(received) == 1
+	})
+	mu.Lock()
+	defer mu.Unlock()
+	require.Equal(t, "mytopic", received[0].Topic)
+	require.Equal(t, "hello cross-node", received[0].Message)
+}
+
+func TestMesh_PeerAPI_Auth(t *testing.T) {
+	schemaDSN := dbtest.CreateTestPostgresSchema(t)
+	pool := openTestPool(t, schemaDSN)
+	var delivered int
+	mesh, err := newMeshCluster(newTestMeshConfig("node-a", "http://127.0.0.1:1"), pool, func(m *model.Message) {
+		delivered++
+	})
+	require.Nil(t, err)
+	defer mesh.Close()
+	frag, err := marshalMessage(model.NewDefaultMessage("mytopic", "hi"))
+	require.Nil(t, err)
+	payload := assembleMessageBody([][]byte{frag})
+
+	// Wrong secret -> 401, not delivered
+	rr := httptest.NewRecorder()
+	req := httptest.NewRequest("POST", MessagePath, strings.NewReader(string(payload)))
+	req.Header.Set(secretHeader, "wrong")
+	req.Header.Set(originHeader, "node-b")
+	mesh.ServeHTTP(rr, req)
+	require.Equal(t, 401, rr.Code)
+
+	// Missing secret -> 401, not delivered
+	rr = httptest.NewRecorder()
+	mesh.ServeHTTP(rr, httptest.NewRequest("POST", MessagePath, strings.NewReader(string(payload))))
+	require.Equal(t, 401, rr.Code)
+	require.Equal(t, 0, delivered)
+
+	// Missing origin -> 400, not delivered
+	rr = httptest.NewRecorder()
+	req = httptest.NewRequest("POST", MessagePath, strings.NewReader(string(payload)))
+	req.Header.Set(secretHeader, testSecret)
+	mesh.ServeHTTP(rr, req)
+	require.Equal(t, 400, rr.Code)
+	require.Equal(t, 0, delivered)
+
+	// Correct secret and origin -> 200, delivered
+	rr = httptest.NewRecorder()
+	req = httptest.NewRequest("POST", MessagePath, strings.NewReader(string(payload)))
+	req.Header.Set(secretHeader, testSecret)
+	req.Header.Set(originHeader, "node-b")
+	mesh.ServeHTTP(rr, req)
+	require.Equal(t, 200, rr.Code)
+	require.Equal(t, 1, delivered)
+}
+
+func TestMesh_PeerAPI_SelfOrigin(t *testing.T) {
+	schemaDSN := dbtest.CreateTestPostgresSchema(t)
+	pool := openTestPool(t, schemaDSN)
+	var delivered int
+	mesh, err := newMeshCluster(newTestMeshConfig("node-a", "http://127.0.0.1:1"), pool, func(m *model.Message) {
+		delivered++
+	})
+	require.Nil(t, err)
+	defer mesh.Close()
+
+	// A request that carries this node's own broadcasts must not be re-delivered (loop prevention)
+	frag, err := marshalMessage(model.NewDefaultMessage("mytopic", "loop"))
+	require.Nil(t, err)
+	payload := assembleMessageBody([][]byte{frag})
+	rr := httptest.NewRecorder()
+	req := httptest.NewRequest("POST", MessagePath, strings.NewReader(string(payload)))
+	req.Header.Set(secretHeader, testSecret)
+	req.Header.Set(originHeader, "node-a") // Same as the receiving node's ID
+	mesh.ServeHTTP(rr, req)
+	require.Equal(t, 200, rr.Code)
+	require.Equal(t, 0, delivered)
+}
+
+func TestMesh_SlowPeerIsolation(t *testing.T) {
+	// A wedged peer must not delay delivery to healthy peers: each peer has its own queue and
+	// delivery worker. With a shared send queue (the design this replaces), the slow peer's
+	// requests would occupy all delivery workers and starve the fast peer.
+	schemaDSN := dbtest.CreateTestPostgresSchema(t)
+	pool := openTestPool(t, schemaDSN)
+	var mu sync.Mutex
+	fastReceived := 0 // Messages, not requests: with batching, one request can carry many
+	srvFast := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, err := io.ReadAll(r.Body)
+		require.Nil(t, err)
+		messages, err := unmarshalMessageBody(body, 1<<20)
+		require.Nil(t, err)
+		mu.Lock()
+		fastReceived += len(messages)
+		mu.Unlock()
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer srvFast.Close()
+	release := make(chan struct{})
+	srvSlow := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		<-release // Wedged until the end of the test
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer srvSlow.Close()
+	defer close(release)
+	// Register the fake peers before the mesh boots; its first heartbeat caches the peer list
+	for i, url := range []string{srvFast.URL, srvSlow.URL} {
+		registerFakePeer(t, pool, NodeID(fmt.Sprintf("node-fake-%d", i)), url)
+	}
+	mesh, err := newMeshCluster(newTestMeshConfig("node-a", "http://127.0.0.1:1"), pool, nil)
+	require.Nil(t, err)
+	defer mesh.Close()
+	const n = 20
+	for i := 0; i < n; i++ {
+		require.Nil(t, mesh.ForwardMessage(model.NewDefaultMessage("mytopic", fmt.Sprintf("message %d", i))))
+	}
+	waitFor(t, func() bool {
+		mu.Lock()
+		defer mu.Unlock()
+		return fastReceived == n
+	})
+}
+
+func TestMesh_BatchCoalescing(t *testing.T) {
+	// Messages published within the linger window arrive as batches: fewer HTTP requests than
+	// messages, with nothing lost. Fails against a one-request-per-message sender.
+	schemaDSN := dbtest.CreateTestPostgresSchema(t)
+	pool := openTestPool(t, schemaDSN)
+	var mu sync.Mutex
+	requests, messages := 0, 0
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, err := io.ReadAll(r.Body)
+		require.Nil(t, err)
+		decoded, err := unmarshalMessageBody(body, 1<<20)
+		require.Nil(t, err)
+		mu.Lock()
+		requests++
+		messages += len(decoded)
+		mu.Unlock()
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer srv.Close()
+	registerFakePeer(t, pool, "node-fake", srv.URL)
+	conf := newTestMeshConfig("node-a", "http://127.0.0.1:1")
+	conf.BatchLinger = 150 * time.Millisecond
+	mesh, err := newMeshCluster(conf, pool, nil)
+	require.Nil(t, err)
+	defer mesh.Close()
+	const n = 20
+	for i := 0; i < n; i++ {
+		require.Nil(t, mesh.ForwardMessage(model.NewDefaultMessage("mytopic", fmt.Sprintf("message %d", i))))
+	}
+	waitFor(t, func() bool {
+		mu.Lock()
+		defer mu.Unlock()
+		return messages == n
+	})
+	mu.Lock()
+	defer mu.Unlock()
+	require.Less(t, requests, 5, "expected %d messages coalesced into few requests, got %d", n, requests)
+}
+
+func TestMesh_DeadPeerRemovedAndRejoin(t *testing.T) {
+	// A peer that dies ungracefully (no Deregister) stops refreshing its heartbeat: after the
+	// TTL it no longer counts as live (no more sends), its queue/worker are reconciled away, the
+	// leader prunes its registry row, and a re-registered peer starts receiving again.
+	schemaDSN := dbtest.CreateTestPostgresSchema(t)
+	pool := openTestPool(t, schemaDSN)
+	var mu sync.Mutex
+	received := 0
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, err := io.ReadAll(r.Body)
+		require.Nil(t, err)
+		messages, err := unmarshalMessageBody(body, 1<<20)
+		require.Nil(t, err)
+		mu.Lock()
+		received += len(messages)
+		mu.Unlock()
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer srv.Close()
+	conf := newTestMeshConfig("node-a", "http://127.0.0.1:1")
+	conf.NodeTTL = 300 * time.Millisecond // Fast expiry so the test observes TTL-based removal
+	mesh, err := newMeshCluster(conf, pool, nil)
+	require.Nil(t, err)
+	defer mesh.Close()
+	// The fake peer registers once and then "dies": its heartbeat is never refreshed
+	registerFakePeer(t, pool, "node-dead", srv.URL)
+	_, err = mesh.registry.Refresh() // See the peer now; the short TTL would otherwise expire it first
+	require.Nil(t, err)
+	require.Nil(t, mesh.ForwardMessage(model.NewDefaultMessage("mytopic", "while alive")))
+	waitFor(t, func() bool {
+		mu.Lock()
+		defer mu.Unlock()
+		return received == 1
+	})
+	// After the TTL, the peer is no longer live: its queue is reconciled away and its registry
+	// row is pruned by the leader (this mesh is the only real node, so it holds the lock)
+	waitFor(t, func() bool {
+		mesh.mu.Lock()
+		defer mesh.mu.Unlock()
+		return len(mesh.queues) == 0
+	})
+	waitFor(t, func() bool {
+		var count int
+		require.Nil(t, pool.QueryRow(`SELECT COUNT(*) FROM node_registry WHERE node_id = 'node-dead'`).Scan(&count))
+		return count == 0
+	})
+	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
+	mu.Unlock()
+	// The peer comes back (same node ID, fresh heartbeat) and receives messages again; the
+	// 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.ForwardMessage(model.NewDefaultMessage("mytopic", "after rejoin")))
+		mu.Lock()
+		defer mu.Unlock()
+		return received > 1
+	})
+}
+
+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)
+	pool := openTestPool(t, schemaDSN)
+	mesh, err := newMeshCluster(newTestMeshConfig("node-a", "http://127.0.0.1:1"), pool, nil)
+	require.Nil(t, err)
+	registerFakePeer(t, pool, "node-peer", "http://127.0.0.1:1")
+	require.Nil(t, mesh.Close())
+	require.Nil(t, mesh.ForwardMessage(model.NewDefaultMessage("mytopic", "too late"))) // Dropped silently
+	mesh.mu.Lock()
+	defer mesh.mu.Unlock()
+	require.Empty(t, mesh.queues)
+}
+
+func TestMesh_LeaderFailover(t *testing.T) {
+	schemaDSN := dbtest.CreateTestPostgresSchema(t)
+	poolA, poolB := openTestPool(t, schemaDSN), openTestPool(t, schemaDSN)
+	meshA, err := newMeshCluster(newTestMeshConfig("node-a", "http://127.0.0.1:1"), poolA, nil)
+	require.Nil(t, err)
+	defer meshA.Close()
+	meshB, err := newMeshCluster(newTestMeshConfig("node-b", "http://127.0.0.1:1"), poolB, nil)
+	require.Nil(t, err)
+	defer meshB.Close()
+	// Exactly one node becomes leader
+	waitFor(t, func() bool {
+		return meshA.IsLeader() != meshB.IsLeader() // Exactly one
+	})
+	// The leader steps down; the follower takes over
+	leader, follower := meshA, meshB
+	if meshB.IsLeader() {
+		leader, follower = meshB, meshA
+	}
+	require.Nil(t, leader.Close())
+	waitFor(t, follower.IsLeader)
+}
+
+func TestMesh_CloseDeregisters(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)
+	require.Nil(t, err)
+	var count int
+	require.Nil(t, pool.QueryRow(`SELECT COUNT(*) FROM node_registry WHERE node_id = 'node-a'`).Scan(&count))
+	require.Equal(t, 1, count)
+	require.Nil(t, mesh.Close())
+	require.Nil(t, pool.QueryRow(`SELECT COUNT(*) FROM node_registry WHERE node_id = 'node-a'`).Scan(&count))
+	require.Equal(t, 0, count)
+}
+
+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)
+	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())
+}
+
+func TestMesh_SubscriberCancelBroadcast(t *testing.T) {
+	// A subscriber-cancel broadcast from node A must invoke node B's CancelFunc; node A's own
+	// callback must not fire (no loop).
+	schemaDSN := dbtest.CreateTestPostgresSchema(t)
+	poolA, poolB := openTestPool(t, schemaDSN), openTestPool(t, schemaDSN)
+	var mu sync.Mutex
+	var receivedB []*SubscriberCancel
+	canceledA := 0
+
+	listenerB, err := net.Listen("tcp", "127.0.0.1:0")
+	require.Nil(t, err)
+	confB := newTestMeshConfig("node-b", "http://"+listenerB.Addr().String())
+	confB.CancelFunc = func(cancel *SubscriberCancel) {
+		mu.Lock()
+		defer mu.Unlock()
+		receivedB = append(receivedB, cancel)
+	}
+	meshB, err := newMeshCluster(confB, poolB, nil)
+	require.Nil(t, err)
+	defer meshB.Close()
+	srvB := &http.Server{Handler: meshB}
+	go srvB.Serve(listenerB)
+	defer srvB.Close()
+
+	confA := newTestMeshConfig("node-a", "http://127.0.0.1:1")
+	confA.CancelFunc = func(_ *SubscriberCancel) {
+		mu.Lock()
+		defer mu.Unlock()
+		canceledA++
+	}
+	meshA, err := newMeshCluster(confA, poolA, nil)
+	require.Nil(t, err)
+	defer meshA.Close()
+
+	meshA.BroadcastState(&State{SubscriberCancels: []*SubscriberCancel{
+		{Topic: "mytopic", ExceptUserID: "u_owner"},
+		{Topic: "up*", UserID: "u_revoked"},
+	}})
+	waitFor(t, func() bool {
+		mu.Lock()
+		defer mu.Unlock()
+		return len(receivedB) == 2
+	})
+	mu.Lock()
+	defer mu.Unlock()
+	require.Equal(t, "mytopic", receivedB[0].Topic)
+	require.Equal(t, "u_owner", receivedB[0].ExceptUserID)
+	require.Equal(t, "up*", receivedB[1].Topic)
+	require.Equal(t, "u_revoked", receivedB[1].UserID)
+	require.Equal(t, 0, canceledA)
+}
+
+func TestMesh_ForwardsEveryTopicToEveryPeer(t *testing.T) {
+	// Fan-out is unconditional: a peer gets every message and drops the ones it has no
+	// subscriber for. Nothing about a peer's subscriptions may gate delivery, which is what
+	// routing by subscription knowledge used to do (removed: ordering made it lose messages).
+	schemaDSN := dbtest.CreateTestPostgresSchema(t)
+	pool := openTestPool(t, schemaDSN)
+	var mu sync.Mutex
+	var topics []string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, err := io.ReadAll(r.Body)
+		require.Nil(t, err)
+		messages, err := unmarshalMessageBody(body, 1<<20)
+		require.Nil(t, err)
+		mu.Lock()
+		for _, m := range messages {
+			topics = append(topics, m.Topic)
+		}
+		mu.Unlock()
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer srv.Close()
+	registerFakePeer(t, pool, "node-b", srv.URL)
+	mesh, err := newMeshCluster(newTestMeshConfig("node-a", "http://127.0.0.1:1"), pool, nil)
+	require.Nil(t, err)
+	defer mesh.Close()
+	require.Nil(t, mesh.ForwardMessage(model.NewDefaultMessage("some-topic", "one")))
+	require.Nil(t, mesh.ForwardMessage(model.NewDefaultMessage("other-topic", "two")))
+	waitFor(t, func() bool {
+		mu.Lock()
+		defer mu.Unlock()
+		return len(topics) == 2
+	})
+	mu.Lock()
+	defer mu.Unlock()
+	require.ElementsMatch(t, []string{"some-topic", "other-topic"}, topics)
+}
+
+func TestMesh_TopicAnnouncementInvokesTopicsAddedFunc(t *testing.T) {
+	// A first-subscriber announcement from node A must reach node B's TopicsAddedFunc, which
+	// the server uses to drop a stale UnifiedPush rate-visitor miss
+	schemaDSN := dbtest.CreateTestPostgresSchema(t)
+	poolA, poolB := openTestPool(t, schemaDSN), openTestPool(t, schemaDSN)
+	var mu sync.Mutex
+	var addedB []string
+
+	listenerB, err := net.Listen("tcp", "127.0.0.1:0")
+	require.Nil(t, err)
+	confB := newTestMeshConfig("node-b", "http://"+listenerB.Addr().String())
+	confB.TopicsAddedFunc = func(topics []string) {
+		mu.Lock()
+		defer mu.Unlock()
+		addedB = append(addedB, topics...)
+	}
+	meshB, err := newMeshCluster(confB, poolB, nil)
+	require.Nil(t, err)
+	defer meshB.Close()
+	srvB := &http.Server{Handler: meshB}
+	go srvB.Serve(listenerB)
+	defer srvB.Close()
+
+	meshA, err := newMeshCluster(newTestMeshConfig("node-a", "http://127.0.0.1:1"), poolA, nil)
+	require.Nil(t, err)
+	defer meshA.Close()
+
+	meshA.BroadcastState(&State{AddedTopics: []string{"up123456789012"}})
+	waitFor(t, func() bool {
+		mu.Lock()
+		defer mu.Unlock()
+		return len(addedB) > 0
+	})
+	mu.Lock()
+	defer mu.Unlock()
+	require.Equal(t, []string{"up123456789012"}, addedB)
+}
+
+func TestMesh_IsolatedFuncWhenPeersHealthy(t *testing.T) {
+	// A node that lost its database (registration goes stale) while a peer is still healthy
+	// is isolated: peers stop forwarding to it, so its subscribers would silently receive
+	// nothing. IsolatedFunc tells the server to close them so clients reconnect elsewhere.
+	isolatedTest(t, http.StatusOK, true)
+}
+
+func TestMesh_NoIsolatedFuncWhenNoPeerHealthy(t *testing.T) {
+	// Full database outage: every node is unhealthy, the mesh keeps delivering on its cached
+	// peer view, so subscribers must be kept (fail open)
+	isolatedTest(t, http.StatusServiceUnavailable, false)
+}
+
+func isolatedTest(t *testing.T, peerHealthStatus int, wantIsolated bool) {
+	schemaDSN := dbtest.CreateTestPostgresSchema(t)
+	peer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/v1/health" {
+			w.WriteHeader(peerHealthStatus)
+		}
+	}))
+	defer peer.Close()
+
+	host, err := pg.Open(schemaDSN)
+	require.Nil(t, err)
+	pool := db.New(host, nil)
+	var isolated atomic.Int32
+	conf := newTestMeshConfig("node-a", "http://127.0.0.1:1")
+	conf.NodeTTL = 300 * time.Millisecond
+	conf.IsolatedFunc = func() { isolated.Add(1) }
+	// Registered last: the fake peer never heartbeats again, so its row must still be fresh
+	// when the mesh takes its first peer snapshot
+	registerFakePeer(t, openTestPool(t, schemaDSN), "node-peer", peer.URL)
+	mesh, err := newMeshCluster(conf, pool, nil)
+	require.Nil(t, err)
+	defer mesh.Close()
+	waitFor(t, func() bool {
+		mesh.mu.Lock()
+		defer mesh.mu.Unlock()
+		return len(mesh.knownPeers) == 1
+	})
+	require.Equal(t, int32(0), isolated.Load()) // Healthy node: never isolated
+
+	pool.Close() // Database lost: registration fails from now on
+	time.Sleep(time.Second)
+	require.False(t, mesh.Healthy())
+	require.Equal(t, wantIsolated, isolated.Load() > 0)
+}
+
+func TestMesh_IsolatedFuncWhenDatabaseHangs(t *testing.T) {
+	// A network partition makes database calls hang (dropped packets) instead of failing, which
+	// blocks the heartbeat loop. Isolation must still be detected (Healthy is time-based).
+	schemaDSN := dbtest.CreateTestPostgresSchema(t)
+	peer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
+	defer peer.Close()
+
+	proxy := newFreezableProxy(t, schemaDSN)
+	var isolated atomic.Int32
+	conf := newTestMeshConfig("node-a", "http://127.0.0.1:1")
+	conf.NodeTTL = 300 * time.Millisecond
+	conf.IsolatedFunc = func() { isolated.Add(1) }
+	// Registered last, and through its own pool: the fake peer never heartbeats again, so its
+	// row must still be fresh when the mesh takes its first peer snapshot
+	registerFakePeer(t, openTestPool(t, schemaDSN), "node-peer", peer.URL)
+	mesh, err := newMeshCluster(conf, openTestPool(t, proxy.dsn), nil)
+	require.Nil(t, err)
+	defer mesh.Close()
+	waitFor(t, func() bool {
+		mesh.mu.Lock()
+		defer mesh.mu.Unlock()
+		return len(mesh.knownPeers) == 1
+	})
+
+	proxy.freeze()
+	waitFor(t, func() bool { return isolated.Load() > 0 })
+}
+
+// freezableProxy forwards TCP to the test database until frozen; then it stops moving bytes
+// (and new connections hang), like a partition that drops packets
+type freezableProxy struct {
+	dsn    string
+	frozen atomic.Bool
+}
+
+func newFreezableProxy(t *testing.T, dsn string) *freezableProxy {
+	u, err := url.Parse(dsn)
+	require.Nil(t, err)
+	target := u.Host
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.Nil(t, err)
+	t.Cleanup(func() { listener.Close() })
+	p := &freezableProxy{}
+	u.Host = listener.Addr().String()
+	p.dsn = u.String()
+	pipe := func(dst, src net.Conn) {
+		buf := make([]byte, 32*1024)
+		for {
+			n, err := src.Read(buf)
+			for p.frozen.Load() {
+				time.Sleep(50 * time.Millisecond) // Hold everything, never close: a black hole
+			}
+			if err != nil {
+				dst.Close()
+				return
+			}
+			if _, err := dst.Write(buf[:n]); err != nil {
+				return
+			}
+		}
+	}
+	go func() {
+		for {
+			client, err := listener.Accept()
+			if err != nil {
+				return
+			}
+			server, err := net.Dial("tcp", target)
+			if err != nil {
+				client.Close()
+				continue
+			}
+			go pipe(server, client)
+			go pipe(client, server)
+		}
+	}()
+	return p
+}
+
+func (p *freezableProxy) freeze() {
+	p.frozen.Store(true)
+}
+
+func TestMesh_MembersEndpoint(t *testing.T) {
+	// The LB agents ask a node which nodes are live, so each LB can maintain its own upstream
+	// list instead of a central monitor pushing it. Secret-authenticated, read-only.
+	schemaDSN := dbtest.CreateTestPostgresSchema(t)
+	pool := openTestPool(t, schemaDSN)
+	registerFakePeer(t, pool, "node-peer", "http://192.168.1.50:2587")
+	mesh, err := newMeshCluster(newTestMeshConfig("node-a", "http://192.168.1.10:2587"), pool, nil)
+	require.Nil(t, err)
+	defer mesh.Close()
+	waitFor(t, func() bool {
+		mesh.mu.Lock()
+		defer mesh.mu.Unlock()
+		return len(mesh.knownPeers) == 1
+	})
+
+	// Without the shared secret: rejected
+	rr := httptest.NewRecorder()
+	mesh.ServeHTTP(rr, httptest.NewRequest("GET", MembersPath, nil))
+	require.Equal(t, http.StatusUnauthorized, rr.Code)
+
+	// With it: this node plus its live peers, with the addresses an agent needs
+	rr = httptest.NewRecorder()
+	req := httptest.NewRequest("GET", MembersPath, nil)
+	req.Header.Set(secretHeader, testSecret)
+	mesh.ServeHTTP(rr, req)
+	require.Equal(t, http.StatusOK, rr.Code)
+	var members []Member
+	require.Nil(t, json.Unmarshal(rr.Body.Bytes(), &members))
+	byID := make(map[NodeID]Member)
+	for _, m := range members {
+		byID[m.NodeID] = m
+	}
+	require.Equal(t, 2, len(byID))
+	require.Equal(t, "http://192.168.1.10:2587", byID["node-a"].AdvertiseURL)
+	require.True(t, byID["node-a"].Healthy)
+	require.Equal(t, "http://192.168.1.50:2587", byID["node-peer"].AdvertiseURL)
+}
+
+func TestMesh_MembersOfAnIsolatedNodeAreNotReportedHealthy(t *testing.T) {
+	// A node that lost the database keeps serving its last peer view, so it answers the LB
+	// agents with a membership that may be minutes out of date, and used to mark every one of
+	// those peers healthy. An agent that happens to ask this node then pins its upstreams to a
+	// frozen list. The answer has to carry how much this node still knows: nothing.
+	schemaDSN := dbtest.CreateTestPostgresSchema(t)
+	pool := openTestPool(t, schemaDSN)
+	registerFakePeer(t, pool, "node-peer", "http://192.168.1.50:2587")
+	conf := newTestMeshConfig("node-a", "http://192.168.1.10:2587")
+	mesh, err := newMeshCluster(conf, pool, nil)
+	require.Nil(t, err)
+	defer mesh.Close()
+	waitFor(t, func() bool {
+		mesh.mu.Lock()
+		defer mesh.mu.Unlock()
+		return len(mesh.knownPeers) == 1
+	})
+	healthyPeers := func() int {
+		n := 0
+		for _, m := range mesh.Members() {
+			if m.NodeID != conf.NodeID && m.Healthy {
+				n++
+			}
+		}
+		return n
+	}
+	require.Equal(t, 1, healthyPeers()) // While the view is fresh, the peer counts
+
+	require.Nil(t, pool.Close()) // The database goes away: heartbeats and reads fail
+	waitFor(t, func() bool { return healthyPeers() == 0 })
+	for _, m := range mesh.Members() {
+		require.False(t, m.Healthy, "an isolated node reported %s as healthy", m.NodeID)
+	}
+}

cluster/registry/registry.go

@@ -136,6 +136,16 @@ func (r *Registry) Peers() []*Peer {
 	return r.peers
 }
 
+// Fresh reports whether the snapshot Peers serves is recent enough to act on. Peers keeps
+// serving the last known list when the database is unreachable, which is what keeps fan-out
+// alive through a hiccup, but a view that old must not be reported outwards as the current
+// membership (see cluster.Member).
+func (r *Registry) Fresh() bool {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return !r.refreshedAt.IsZero() && time.Since(r.refreshedAt) < r.ttl
+}
+
 // Prune deletes registry rows whose heartbeat is long expired. Only the leader calls this; the
 // grace period of 3x the TTL avoids deleting rows of nodes that are merely slow to heartbeat.
 func (r *Registry) Prune() error {

cluster/registry/registry_test.go

@@ -471,3 +471,26 @@ func TestRegistry_LeaseLapsesBeforeAnySuccessorBegins(t *testing.T) {
 	// short that it ends before peers would even notice this node is gone
 	require.GreaterOrEqual(t, leaseFactor, 1, "the lease must cover the liveness window")
 }
+
+func TestRegistry_ViewIsOnlyFreshAfterARecentRefresh(t *testing.T) {
+	// Peers serves the last known list when the database is unreachable, which keeps fan-out
+	// alive but must not be mistaken for the current membership: anything that reports
+	// membership outwards (the member list the LB agents read) needs to know the difference.
+	t.Parallel()
+	pool := openTestPool(t, dbtest.CreateTestPostgresSchema(t))
+	r, err := New(pool, "node-a", "http://127.0.0.1:1", leaderTestTTL)
+	require.Nil(t, err)
+	require.False(t, r.Fresh(), "a registry that has never read the database is not fresh")
+
+	require.Nil(t, r.Register())
+	_, err = r.Refresh()
+	require.Nil(t, err)
+	require.True(t, r.Fresh())
+
+	// The view ages out on its own, without any failed call
+	deadline := time.Now().Add(leaderTestWait)
+	for r.Fresh() && time.Now().Before(deadline) {
+		time.Sleep(leaderTestPoll)
+	}
+	require.False(t, r.Fresh(), "the view never went stale")
+}

cluster/types.go

@@ -4,6 +4,7 @@ import (
 	"time"
 
 	"heckel.io/ntfy/v2/model"
+	"heckel.io/ntfy/v2/util"
 )
 
 // Config configures the cluster. It is assembled by the server from its own config, which keeps
@@ -54,6 +55,26 @@ type CancelFunc func(cancel *SubscriberCancel)
 // announcement costs a stale lookup until its TTL, never a lost message.
 type TopicsAddedFunc func(topics []string)
 
+// apiState is the peer state-exchange envelope. Each concern is an optional section; future
+// concerns (rate limit counters, stats) become siblings of Topics.
+type apiState struct {
+	Topics  *apiStateTopics     `json:"topics,omitempty"`
+	Cancels []*SubscriberCancel `json:"cancels,omitempty"`
+}
+
+// apiStateTopics carries topics that just gained their first subscriber on the sending node.
+type apiStateTopics struct {
+	Added []string `json:"added,omitempty"`
+}
+
+// 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; ForwardMessage creates a fresh one on demand).
+type peerQueue struct {
+	advertiseURL string
+	queue        *util.LingerQueue[[]byte]
+}
+
 // 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

cluster/util.go

@@ -6,11 +6,22 @@ import (
 	"encoding/json"
 	"io"
 	"net/netip"
+	"strings"
 
 	"heckel.io/ntfy/v2/log"
 	"heckel.io/ntfy/v2/model"
 )
 
+// messageURL derives the peer's message endpoint URL from its advertise URL.
+func messageURL(advertiseURL string) string {
+	return strings.TrimRight(advertiseURL, "/") + MessagePath
+}
+
+// stateURL derives the peer's state endpoint URL from its advertise URL.
+func stateURL(advertiseURL string) string {
+	return strings.TrimRight(advertiseURL, "/") + StatePath
+}
+
 // marshalMessage serializes one message and its non-JSON fields (Sender, User) as an
 // apiMessage line. Lines are marshaled once per publish and shared across all per-peer
 // queues; assembleMessageBody joins them without re-marshaling.

db/test/test.go

@@ -21,7 +21,7 @@ const testPoolMaxConns = "2"
 // CreateTestPostgresSchema creates a temporary PostgreSQL schema and returns the DSN pointing to it.
 // It registers a cleanup function to drop the schema when the test finishes.
 // If NTFY_TEST_DATABASE_URL is not set, the test is skipped.
-func CreateTestPostgresSchema(t *testing.T) string {
+func CreateTestPostgresSchema(t testing.TB) string {
 	t.Helper()
 	dsn := os.Getenv("NTFY_TEST_DATABASE_URL")
 	if dsn == "" {
@@ -55,7 +55,7 @@ func CreateTestPostgresSchema(t *testing.T) string {
 // CreateTestPostgres creates a temporary PostgreSQL schema and returns an open *db.DB connection to it.
 // It registers cleanup functions to close the DB and drop the schema when the test finishes.
 // If NTFY_TEST_DATABASE_URL is not set, the test is skipped.
-func CreateTestPostgres(t *testing.T) *db.DB {
+func CreateTestPostgres(t testing.TB) *db.DB {
 	t.Helper()
 	schemaDSN := CreateTestPostgresSchema(t)
 	testHost, err := pg.Open(schemaDSN)

metrics/metrics.go

@@ -75,6 +75,27 @@ var (
 	HTTPRequests = prometheus.NewCounterVec(prometheus.CounterOpts{
 		Name: "ntfy_http_requests_total",
 	}, []string{"http_code", "ntfy_code", "http_method"})
+	ClusterPeers = prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "ntfy_cluster_peers",
+	})
+	ClusterMessagesForwarded = prometheus.NewCounter(prometheus.CounterOpts{
+		Name: "ntfy_cluster_messages_forwarded_total",
+	})
+	ClusterBatchesSent = prometheus.NewCounter(prometheus.CounterOpts{
+		Name: "ntfy_cluster_batches_sent_total",
+	})
+	ClusterSendErrors = prometheus.NewCounter(prometheus.CounterOpts{
+		Name: "ntfy_cluster_send_errors_total",
+	})
+	ClusterQueueDropped = prometheus.NewCounter(prometheus.CounterOpts{
+		Name: "ntfy_cluster_queue_dropped_total",
+	})
+	ClusterMessagesWasted = prometheus.NewCounter(prometheus.CounterOpts{
+		Name: "ntfy_cluster_messages_wasted_total",
+	})
+	ClusterLeader = prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "ntfy_cluster_leader",
+	})
 )
 
 // init registers all collectors with the default Prometheus registry. Registration is
@@ -98,6 +119,13 @@ func init() {
 		MatrixPublishedSuccess,
 		MatrixPublishedFailure,
 		AttachmentsTotalSize,
+		ClusterPeers,
+		ClusterMessagesForwarded,
+		ClusterBatchesSent,
+		ClusterSendErrors,
+		ClusterQueueDropped,
+		ClusterMessagesWasted,
+		ClusterLeader,
 		Visitors,
 		Users,
 		Subscribers,

metrics/metrics_test.go

@@ -15,6 +15,13 @@ var expectedMetricNames = []string{
 	"ntfy_attachments_total_size",
 	"ntfy_calls_made_failure",
 	"ntfy_calls_made_success",
+	"ntfy_cluster_batches_sent_total",
+	"ntfy_cluster_leader",
+	"ntfy_cluster_messages_forwarded_total",
+	"ntfy_cluster_messages_wasted_total",
+	"ntfy_cluster_peers",
+	"ntfy_cluster_queue_dropped_total",
+	"ntfy_cluster_send_errors_total",
 	"ntfy_emails_received_failure",
 	"ntfy_emails_received_success",
 	"ntfy_emails_sent_failure",