Add the experimental-cluster-role option

cmd/serve.go

@@ -19,6 +19,7 @@ import (
 	"github.com/urfave/cli/v2"
 	"github.com/urfave/cli/v2/altsrc"
 	"heckel.io/ntfy/v2/ban"
+	"heckel.io/ntfy/v2/cluster"
 	"heckel.io/ntfy/v2/log"
 	"heckel.io/ntfy/v2/payments"
 	"heckel.io/ntfy/v2/server"
@@ -47,6 +48,7 @@ var flagsServe = append(
 	altsrc.NewStringFlag(&cli.StringFlag{Name: "experimental-cluster-listen", Aliases: []string{"experimental_cluster_listen"}, EnvVars: []string{"NTFY_EXPERIMENTAL_CLUSTER_LISTEN"}, Usage: "ip:port for the dedicated cluster fan-out listener; bind it to the private network (e.g. 10.0.0.5:2587)"}),
 	altsrc.NewStringFlag(&cli.StringFlag{Name: "experimental-cluster-advertise-url", Aliases: []string{"experimental_cluster_advertise_url"}, EnvVars: []string{"NTFY_EXPERIMENTAL_CLUSTER_ADVERTISE_URL"}, Usage: "base URL peer nodes use to reach this node's fan-out listener (defaults to http://<experimental-cluster-listen>)"}),
 	altsrc.NewStringFlag(&cli.StringFlag{Name: "experimental-cluster-secret", Aliases: []string{"experimental_cluster_secret"}, EnvVars: []string{"NTFY_EXPERIMENTAL_CLUSTER_SECRET"}, Usage: "shared secret authenticating node-to-node fan-out requests"}),
+	altsrc.NewStringFlag(&cli.StringFlag{Name: "experimental-cluster-role", Aliases: []string{"experimental_cluster_role"}, EnvVars: []string{"NTFY_EXPERIMENTAL_CLUSTER_ROLE"}, Value: string(server.DefaultClusterRole), Usage: "role of this cluster node: active (takes client traffic, may lead) or passive (hidden: never leads, never fences)"}),
 	altsrc.NewStringFlag(&cli.StringFlag{Name: "experimental-cluster-batch-linger", Aliases: []string{"experimental_cluster_batch_linger"}, EnvVars: []string{"NTFY_EXPERIMENTAL_CLUSTER_BATCH_LINGER"}, Value: util.FormatDuration(server.DefaultClusterBatchLinger), Usage: "how long fan-out messages wait to form a batch per peer node (0 = send immediately)"}),
 	altsrc.NewStringFlag(&cli.StringFlag{Name: "cache-file", Aliases: []string{"cache_file", "C"}, EnvVars: []string{"NTFY_CACHE_FILE"}, Usage: "cache file used for message caching"}),
 	altsrc.NewStringFlag(&cli.StringFlag{Name: "cache-duration", Aliases: []string{"cache_duration", "b"}, EnvVars: []string{"NTFY_CACHE_DURATION"}, Value: util.FormatDuration(server.DefaultCacheDuration), Usage: "buffer messages for this time to allow `since` requests"}),
@@ -167,6 +169,7 @@ func execServe(c *cli.Context) error {
 	clusterAdvertiseURL := c.String("experimental-cluster-advertise-url")
 	clusterSecret := c.String("experimental-cluster-secret")
 	clusterBatchLingerStr := c.String("experimental-cluster-batch-linger")
+	clusterRoleStr := c.String("experimental-cluster-role")
 	webPushPrivateKey := c.String("web-push-private-key")
 	webPushPublicKey := c.String("web-push-public-key")
 	webPushFile := c.String("web-push-file")
@@ -266,6 +269,10 @@ func execServe(c *cli.Context) error {
 	if err != nil || clusterBatchLinger < 0 {
 		return fmt.Errorf("invalid cluster batch linger: %s", clusterBatchLingerStr)
 	}
+	clusterRole := cluster.Role(clusterRoleStr)
+	if !clusterRole.Valid() {
+		return fmt.Errorf("invalid experimental-cluster-role: %s (must be %s or %s)", clusterRoleStr, cluster.RoleActive, cluster.RolePassive)
+	}
 	managerInterval, err := util.ParseDuration(managerIntervalStr)
 	if err != nil {
 		return fmt.Errorf("invalid manager interval: %s", managerIntervalStr)
@@ -591,6 +598,7 @@ func execServe(c *cli.Context) error {
 	conf.ClusterAdvertiseURL = clusterAdvertiseURL
 	conf.ClusterSecret = clusterSecret
 	conf.ClusterBatchLinger = clusterBatchLinger
+	conf.ClusterRole = clusterRole
 	conf.WebPushPrivateKey = webPushPrivateKey
 	conf.WebPushPublicKey = webPushPublicKey
 	conf.WebPushFile = webPushFile

cmd/serve_test.go

@@ -571,6 +571,11 @@ func TestCLI_Serve_ClusterValidation(t *testing.T) {
 	err = app.Run([]string{"ntfy", "serve", "--config=" + configFile, "--experimental-cluster-batch-linger=-1s"})
 	require.Error(t, err)
 	require.Contains(t, err.Error(), "cluster batch linger")
+	// The role must be one of the known ones: a typo would make a node half passive
+	app, _, _, _ = newTestApp()
+	err = app.Run([]string{"ntfy", "serve", "--config=" + configFile, "--experimental-cluster-listen=:2587", "--database-url=postgres://user:pass@localhost:1/na", "--experimental-cluster-secret=s3cret", "--experimental-cluster-node-id=node-a", "--experimental-cluster-role=leader"})
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "experimental-cluster-role")
 }
 
 func TestCLI_Serve_AttachmentExpiryLongerThanCacheDurationWarning(t *testing.T) {

docs/config.md

@@ -2368,6 +2368,7 @@ variable before running the `ntfy` command (e.g. `export NTFY_LISTEN_HTTP=:80`).
 | `experimental-cluster-advertise-url`       | `NTFY_EXPERIMENTAL_CLUSTER_ADVERTISE_URL`       | *URL*                                               | `http://<experimental-cluster-listen>` | Experimental, not ready for production. URL under which other nodes reach this node's cluster listener.                                                                                             |
 | `experimental-cluster-secret`              | `NTFY_EXPERIMENTAL_CLUSTER_SECRET`              | *string*                                            | -                 | Experimental, not ready for production. Shared secret authenticating node-to-node traffic; must match on all nodes.                                                                                                                      |
 | `experimental-cluster-batch-linger`        | `NTFY_EXPERIMENTAL_CLUSTER_BATCH_LINGER`        | *duration*                                          | `500ms`           | Experimental, not ready for production. How long messages wait to form a node-to-node delivery batch; `0` sends immediately.                                                                                                             |
+| `experimental-cluster-role`                | `NTFY_EXPERIMENTAL_CLUSTER_ROLE`                | `active` or `passive`                               | `active`          | Experimental, not ready for production. A `passive` node never leads the cluster's singleton jobs and never closes its subscribers to send them to another node; other active nodes do not count it as a place for theirs. Use it for a node that is not behind the load balancer. |
 | `database-replica-urls`                    | `NTFY_DATABASE_REPLICA_URLS`                    | *list of strings (connection URLs)*                 | -                 | PostgreSQL read replica connection strings. Non-critical read-only queries are distributed across replicas (round-robin) with automatic fallback to primary. Requires `database-url`.                                                   |
 | `cache-file`                               | `NTFY_CACHE_FILE`                               | *filename*                                          | -                 | If set, messages are cached in a local SQLite database instead of only in-memory. This allows for service restarts without losing messages in support of the since= parameter. See [message cache](#message-cache).                     |
 | `cache-duration`                           | `NTFY_CACHE_DURATION`                           | *duration*                                          | 12h               | Duration for which messages will be buffered before they are deleted. This is required to support the `since=...` and `poll=1` parameter. Set this to `0` to disable the cache entirely.                                                |

server/config.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"heckel.io/ntfy/v2/ban"
+	"heckel.io/ntfy/v2/cluster"
 	"heckel.io/ntfy/v2/user"
 )
 
@@ -30,6 +31,7 @@ const (
 	DefaultFirebaseQuotaExceededPenaltyDuration = 10 * time.Minute       // Time that over-users are locked out of Firebase if it returns "quota exceeded"
 	DefaultStripePriceCacheDuration             = 3 * time.Hour          // Time to keep Stripe prices cached in memory before a refresh is needed
 	DefaultClusterBatchLinger                   = 500 * time.Millisecond // How long fan-out messages wait to form a per-peer batch (experimental clustering)
+	DefaultClusterRole                          = cluster.RoleActive     // Active nodes take client traffic, lead and fence (experimental clustering)
 )
 
 // Platform-specific default paths (set in config_unix.go or config_windows.go)
@@ -137,6 +139,7 @@ type Config struct {
 	ClusterAdvertiseURL                  string        // Base URL peers use to reach this node's fan-out listener (defaults to "http://<cluster-listen>")
 	ClusterSecret                        string        `hash:"-"` // Shared secret authenticating node-to-node fan-out requests
 	ClusterBatchLinger                   time.Duration // How long fan-out messages wait to form a batch per peer; 0 sends immediately
+	ClusterRole                          cluster.Role  // Active (default) or passive: a passive node never leads and never fences
 	FirebaseKeyFile                      string
 	CacheFile                            string
 	CacheDuration                        time.Duration
@@ -258,6 +261,7 @@ func NewConfig() *Config {
 		ClusterAdvertiseURL:                  "",
 		ClusterSecret:                        "",
 		ClusterBatchLinger:                   DefaultClusterBatchLinger,
+		ClusterRole:                          DefaultClusterRole,
 		FirebaseKeyFile:                      "",
 		CacheFile:                            "",
 		CacheDuration:                        DefaultCacheDuration,

server/server.go

@@ -344,6 +344,7 @@ func New(conf *Config) (*Server, error) {
 		Enabled:         conf.ClusterListen != "", // Setting experimental-cluster-listen implicitly enables clustering
 		NodeID:          cluster.NodeID(conf.ClusterNodeID),
 		AdvertiseURL:    advertiseURL,
+		Role:            conf.ClusterRole,
 		Secret:          conf.ClusterSecret,
 		BatchLinger:     conf.ClusterBatchLinger,
 		IsolatedFunc:    s.closeLocalSubscribers,

server/server.yml

@@ -78,6 +78,12 @@
 # - experimental-cluster-batch-linger is how long fan-out messages may wait to form a batch per
 #   peer node, trading up to that much cross-node delivery latency for a bounded request rate
 #   between nodes. Set to 0 to send each message immediately.
+# - experimental-cluster-role is "active" (default) or "passive". A passive node is meant to run
+#   outside the load balancer (e.g. to observe a new node under real load): it receives fan-out
+#   and serves clients that connect to it directly, but it never leads the cluster's singleton
+#   jobs, and it never closes its subscribers to send them elsewhere. Active nodes do not count
+#   it as a place to send theirs either. The member list reports each node's role, so load
+#   balancer agents can route to active nodes only.
 #
 # Clustering also requires "database-url": the shared PostgreSQL database carries the durable
 # state and the cluster's own coordination tables.
@@ -91,6 +97,7 @@
 # experimental-cluster-advertise-url: "http://<ip:port>"
 # experimental-cluster-secret: <secret>
 # experimental-cluster-batch-linger: 500ms
+# experimental-cluster-role: active
 
 # If "cache-file" is set, messages are cached in a local SQLite database instead of only in-memory.
 # This allows for service restarts without losing messages in support of the since= parameter.

server/server_cluster_test.go

@@ -157,6 +157,21 @@ func TestServer_Cluster_DeliverNotOnPublicHandler(t *testing.T) {
 	})
 }
 
+func TestServer_Cluster_Role(t *testing.T) {
+	// The configured role reaches the cluster, which reports it to the LB agents
+	schemaDSN := dbtest.CreateTestPostgresSchema(t)
+	conf := newTestConfig(t, schemaDSN)
+	conf.ClusterNodeID = "node-a"
+	conf.ClusterListen = "127.0.0.1:1" // Enables clustering; not bound since Run() is not called
+	conf.ClusterSecret = "s3cret"
+	conf.ClusterAdvertiseURL = "http://127.0.0.1:1"
+	conf.ClusterRole = cluster.RolePassive
+	s := newTestServer(t, conf)
+	members := s.cluster.Members()
+	require.Len(t, members, 1)
+	require.Equal(t, cluster.RolePassive, members[0].Role)
+}
+
 func TestServer_Cluster_EndToEnd(t *testing.T) {
 	// Two full servers sharing one Postgres schema: a message published to node A over HTTP must
 	// reach a subscriber connected to node B, via the node registry and the fan-out endpoint.