ntfy/cluster/cluster.go

// Package cluster implements cross-node message delivery for a multi-node ntfy cluster. Nodes
// 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.
package cluster

import (
	"errors"
	"net/http"
	"time"

	"heckel.io/ntfy/v2/db"
	"heckel.io/ntfy/v2/model"
)

// The internal peer API: every kind of node-to-node communication is a path under
// /v1/internal/, served only on the dedicated cluster listener. Future concerns (rate limit
// counters, stats) become new paths or new sections of the state envelope.
const (
	// MessagePath receives batches of published messages (NDJSON, one apiMessage per line).
	MessagePath = "/v1/internal/message"
	// StatePath receives peer state (JSON apiState): full subscription snapshots and
	// incremental updates.
	StatePath = "/v1/internal/state"
)

// NodeID identifies a cluster node; it keys the registry, the per-peer queues, and the peer
// state table.
//
// Naming convention: a "node" is any cluster member in the absolute sense (identity, registry,
// config); a "peer" is another node as seen from this one (Peers, peerQueue, peerState). A peer
// 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"
)

const (
	defaultHeartbeatInterval = 3 * time.Second  // How often a node refreshes its registry heartbeat
	defaultNodeTTL           = 30 * time.Second // A node counts as live if its heartbeat is newer than this; generous to avoid false-dead flapping (see plans)
	defaultStateInterval     = 15 * time.Second // How often the full subscription state is pushed to peers

	// DefaultBatchLinger is how long a fan-out message may wait in a peer's queue for more
	// messages to arrive, so they are delivered as one batch. It trades up to this much
	// cross-node latency for a bounded request rate per peer.
	DefaultBatchLinger = 500 * time.Millisecond
)

// 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 {
	http.Handler
	// 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.
	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
	// 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
}

// New creates the cluster for the given config: the nop cluster when clustering is disabled (the
// single-node default), or the peer-mesh cluster otherwise.
func New(conf *Config, pool *db.DB, deliver DeliverFunc, topics TopicsFunc) (Cluster, error) {
	if !conf.Enabled {
		return &nopCluster{}, nil
	}
	if pool == nil {
		return nil, errors.New("cluster mode requires a PostgreSQL database (set database-url)")
	}
	if conf.AdvertiseURL == "" {
		return nil, errors.New("cluster mode requires an advertise URL (set cluster-advertise-url)")
	}
	if conf.NodeID == "" {
		return nil, errors.New("cluster mode requires a stable node ID (set cluster-node-id)")
	}
	if conf.HeartbeatInterval == 0 {
		conf.HeartbeatInterval = defaultHeartbeatInterval
	}
	if conf.NodeTTL == 0 {
		conf.NodeTTL = defaultNodeTTL
	}
	if conf.StateInterval == 0 {
		conf.StateInterval = defaultStateInterval
	}
	return newMeshCluster(conf, pool, deliver, topics)
}