diff --git a/.context/LEARNINGS.md b/.context/LEARNINGS.md index 753580cf8..105dab46a 100644 --- a/.context/LEARNINGS.md +++ b/.context/LEARNINGS.md @@ -28,6 +28,16 @@ DO NOT UPDATE FOR: --> +## [2026-09-08-153411] Hub cluster mode never started: two defects hid each other + +**Context**: Wiring issue #96's leadership fields through the Status RPC looked like plumbing until the built binary was exercised live: ctx hub start --daemon --peers started a standalone hub (RunDaemon built its re-exec argv from --port and --data-dir only, so --peers was parsed and never forwarded), and in the foreground Run advertised fmt.Sprintf(':%d', port+1) to raft.NewTCPTransport, which refuses an unspecified advertise address. + +**Lesson**: A daemon re-exec argv is a second, silent flag surface: a flag missing there is a flag the process never sees, while the parent still reports success. The daemon path was also hiding the startup crash the foreground path would have shown, so neither bug was visible from the other side alone. Package tests passed throughout. + +**Application**: For any fork/re-exec path, extract the argv into a testable function and pin every flag it must carry (daemonArgs + TestDaemonArgs_ForwardsClusterFlags). Before claiming a CLI feature works, run the built binary through the documented flow, not just the package tests. + +--- + ## [2026-08-23-170949] Hook commands must survive four shells and hostile cwds; hosts punish pre-ctx aborts **Context**: Adversarial audit of every ctx hook surface (Claude/Codex/Copilot manifests, 16 Copilot wrapper scripts, OpenCode plugin, trace hook, plugin-reload) after the Codex non-repo-cwd anchor bug: 20 confirmed defects in 7 classes. diff --git a/.context/TASKS.md b/.context/TASKS.md index 15de64343..4b45d0b78 100644 --- a/.context/TASKS.md +++ b/.context/TASKS.md @@ -1810,7 +1810,14 @@ links back to the spec for detail. `--bootstrap` node calls `BootstrapCluster`, others join via `AddVoter`. Persist a `bootstrapped` flag in the raft data dir to avoid double-bootstrapping - on restart. #priority:medium #added:2026-04-11 #pr:60 + on restart. Partially done in Phase HL: `--join` + starts a node that bootstraps nothing and + `ctx hub peer add` calls `AddVoter`, and + `ErrCantBootstrap` is tolerated on restart. Still + open: making one node the designated bootstrapper + instead of every `--peers` node bootstrapping the + same list, and persisting the flag. + #priority:medium #added:2026-04-11 #pr:60 #audit:H-12 - [ ] **H-13** Follower-side replication validation: call `validateEntry` on every entry received from @@ -1844,11 +1851,15 @@ links back to the spec for detail. by the sysadmin. Precondition for any non-localhost multi-node deployment. #priority:critical #added:2026-04-11 #pr:60 #audit:H-10,H-11 -- [ ] **H-28** Decouple Raft bind port from gRPC port. +- [x] **H-28** Decouple Raft bind port from gRPC port. Accept a dedicated `--raft-bind` flag; default to a random high port or refuse to start. Makes port - scanning less productive. #priority:low - #added:2026-04-11 #pr:60 #audit:H-28 + scanning less productive. Done in Phase HL: the flag + is required for cluster mode and rejects wildcard, + bare-port and host-less addresses, so a hub refuses + to start rather than binding an address no peer can + dial. Spec: specs/hub-status-cluster-leadership.md + #priority:low #added:2026-04-11 #pr:60 #audit:H-28 - [ ] Signed-entry mode: publishing clients sign their entries with a per-client signing key; followers verify on replication. Eliminates the "trust the @@ -3102,3 +3113,32 @@ work is the delivery layer: plugin root, manifests, deployer, parser, docs. - [x] [CX6] Verification gate: make lint, make test, make audit green; live ctx setup codex --write + codex exec hook run (SessionStart context injection, UserPromptSubmit nudges, SessionEnd journal import) recorded in the PR; DECISIONS entries for plugin-root placement, TOML append strategy, skill generation, memories non-goal. Spec: specs/codex-integration.md #priority:medium #session:581183bc #branch:feat/codex-integration #commit:ce5a8328 #added:2026-08-23-120739 - [ ] [CX7] Follow-up: Windows commandWindows overrides for the Codex hooks manifest (hooks currently require a POSIX shell with git on PATH). Spec: specs/codex-integration.md #priority:medium #session:581183bc #branch:feat/codex-integration #commit:ce5a8328 #added:2026-08-23-120739 + +### Phase HL: Hub Status Cluster Leadership (issue #96) + +Spec: `specs/hub-status-cluster-leadership.md`. Read it before starting any +HL task. The Raft `Cluster` on `Server` is never read by the Status RPC, and +the three cluster-ish lines `ctx hub status` prints today (role, leader, +peers) are derived from listener counts, the dialed address and the project +count. HL wires the real state through response → handler → render. +Issue: https://github.com/ActiveMemory/ctx/issues/96 + +- [x] [HL1] Cluster: `LeaderAddr` returns the Raft address it is named for (not the ServerID), `Peers()` reads the committed configuration, and `BootstrapCluster`'s error is checked (tolerating `raft.ErrCantBootstrap` on restart). Spec: specs/hub-status-cluster-leadership.md #priority:medium #branch:fix/hub-status-cluster-leadership #issue:96 #added:2026-09-08-134421 + +- [x] [HL2] Wire: `StatusResponse` gains `ClusterEnabled`, `IsLeader`, `LeaderAddr`, `ClusterPeers`; `hubStatus` populates them from `s.cluster`; `cfgWarn.HubClusterPeers` for a failed configuration read. Spec: specs/hub-status-cluster-leadership.md #priority:medium #branch:fix/hub-status-cluster-leadership #issue:96 #added:2026-09-08-134421 + +- [x] [HL3] Render: `ClusterStatus` takes `ClusterStatusInfo`; standalone prints role + entries only, cluster mode prints the real leader and peer count; `RoleLeader`/`RoleStandalone` replace the `RoleActive` listener-count heuristic. Spec: specs/hub-status-cluster-leadership.md #priority:medium #branch:fix/hub-status-cluster-leadership #issue:96 #added:2026-09-08-134421 + +- [x] [HL4] Docs: hub-cluster recipe expected output, docs/cli/hub.md, commands.yaml description, docs/operations/hub.md monitoring section (drops the nonexistent `--exit-code` flag and the per-peer lag claim), internal/hub/doc.go. Spec: specs/hub-status-cluster-leadership.md #priority:medium #branch:fix/hub-status-cluster-leadership #issue:96 #added:2026-09-08-134421 + +- [x] [HL5] Tests: standalone vs single-node-Raft Status contract, `Peers()` excludes self, three rendered shapes; `make lint` and `make test` green. Spec: specs/hub-status-cluster-leadership.md #priority:medium #branch:fix/hub-status-cluster-leadership #issue:96 #added:2026-09-08-134421 + +- [x] [HL6] `ctx hub stepdown` printed "Leadership transferred" and `ctx hub peer add|remove` printed a peer confirmation, none of which reached the Raft node (`Cluster.Stepdown()` had no caller and raft's AddVoter/RemoveServer were never called). Wired in this PR as two admin-token-gated RPCs (Peer, Stepdown) with leader-only preconditions; see HL9. Spec: specs/hub-status-cluster-leadership.md #priority:medium #branch:fix/hub-status-cluster-leadership #issue:96 #added:2026-09-08-134421 + +- [x] [HL7] Daemon cluster flags: `RunDaemon` built the re-exec argv from --port and --data-dir only, so `ctx hub start --daemon --peers ...` started a standalone hub and reported success. argv extracted to a testable `daemonArgs` that forwards both cluster flags. Spec: specs/hub-status-cluster-leadership.md #priority:medium #branch:fix/hub-status-cluster-leadership #issue:96 #added:2026-09-08-134421 + +- [x] [HL8] Raft bind address (closes H-28): `--raft-bind` binds and advertises the Raft transport, `--peers` becomes the other nodes' Raft addresses, wildcard/bare-port/host-less values are rejected with a message naming the flag, and --raft-bind alone runs a self-electing single node. Before this, `Run` advertised ":port+1" and every cluster start died on "local bind address is not advertisable". Spec: specs/hub-status-cluster-leadership.md #priority:medium #branch:fix/hub-status-cluster-leadership #issue:96 #added:2026-09-08-134421 + +- [x] [HL9] Cluster commands wired: admin-gated Peer and Stepdown RPCs (`Cluster.AddPeer`/`RemovePeer` over raft AddVoter/RemoveServer, `Cluster.Stepdown` over LeadershipTransfer), `raft.ErrNotLeader` mapped to FailedPrecondition naming `ctx hub status`, `--token`/`CTX_HUB_ADMIN_TOKEN` resolution shared by revoke/peer/stepdown via core/admin. Spec: specs/hub-status-cluster-leadership.md #priority:medium #branch:fix/hub-status-cluster-leadership #issue:96 #added:2026-09-08-163000 + +- [x] [HL10] Join mode: `ctx hub start --join` brings up the Raft transport without bootstrapping so a leader can add the node with `ctx hub peer add` (the AddVoter half of H-12; `peer add` is meaningless without it). `--join` with `--peers` is rejected. Verified live: peer add took the leader from Peers: 0 to Peers: 1 and the joiner reported Role: Follower. Spec: specs/hub-status-cluster-leadership.md #priority:medium #branch:fix/hub-status-cluster-leadership #issue:96 #added:2026-09-08-163000 diff --git a/docs/cli/hub.md b/docs/cli/hub.md index f8233c637..cd4feaeb8 100644 --- a/docs/cli/hub.md +++ b/docs/cli/hub.md @@ -62,13 +62,34 @@ the daemon with `ctx hub stop` (see below). #### Cluster Mode For high availability, run multiple hubs with Raft-based -leader election: +leader election. `--raft-bind` is the address this node binds +its Raft transport to and advertises to the others, and +`--peers` lists the `--raft-bind` addresses of the other +nodes — Raft addresses, not hub ports: ```bash ctx hub start --port 9900 \ + --raft-bind host1:9901 \ --peers host2:9901,host3:9901 ``` +`--raft-bind` must name a host a peer can dial. A bare port +(`:9901`) or a wildcard (`0.0.0.0:9901`) is rejected at +startup, because Raft refuses to advertise an address that +does not identify this node to anyone else. + +`--raft-bind` on its own — with no `--peers` — runs a +single-node Raft cluster that elects itself. That is the +cheapest way to see the leadership fields of +[`ctx hub status`](#ctx-hub-status) before adding nodes. + +To add a node to a cluster that is already running, start it +with `--join` instead of `--peers`: it brings up its Raft +transport, bootstraps nothing, and waits for +[`ctx hub peer add`](#ctx-hub-peer) on the leader to hand it a +configuration. `--join` and `--peers` together are an error — +a node either bootstraps a cluster or joins one. + Raft is used **only** for leader election. Data replication uses sequence-based gRPC sync on the append-only JSONL log; there is no multi-node consensus on writes. See the @@ -77,12 +98,14 @@ setup and the Raft-lite durability caveat. #### Flags -| Flag | Description | Default | -|--------------|--------------------------------------------------|------------------| -| `--port` | Hub listen port | `9900` | -| `--data-dir` | Hub data directory | `~/.ctx/hub-data/` | -| `--daemon` | Run the hub server in the background | `false` | -| `--peers` | Comma-separated peer addresses for cluster mode | *(none)* | +| Flag | Description | Default | +|---------------|---------------------------------------------------|------------------| +| `--port` | Hub listen port | `9900` | +| `--data-dir` | Hub data directory | `~/.ctx/hub-data/` | +| `--daemon` | Run the hub server in the background | `false` | +| `--raft-bind` | Raft address this node binds and advertises | *(none)* | +| `--peers` | Comma-separated peer Raft addresses | *(none)* | +| `--join` | Wait to be added by a leader (no bootstrap) | `false` | #### Validation @@ -111,8 +134,34 @@ Safe to rerun: if no daemon is running, returns a ### `ctx hub status` -Show cluster status: role, peers, sync state, entry count, -and uptime. +Show what the hub reports about itself: its role, the current +leader, the entry count and the peer count. + +A hub started without `--peers` runs no Raft node, so it has no +leader and no peers to name: + +``` +Role: Standalone +Entries: 1248 +``` + +A hub started with peers answers from its Raft node. The role is +`Leader` or `Follower`, the leader is the address Raft holds for +the current term, and the peer count is the committed cluster +configuration minus the node answering: + +``` +Role: Leader +Leader: 10.0.0.5:9901 +Entries: 1248 Peers: 2 +``` + +While an election is in progress — or after quorum is lost — +Raft knows no leader, and the line says so: + +``` +Leader: unknown (election in progress) +``` When the hub has disconnected any slow listeners, the output gains a `Dropped listeners:` line with the cumulative count. @@ -128,15 +177,33 @@ ctx hub status ### `ctx hub peer` -Add or remove peers from the cluster at runtime. Useful for -scaling up or replacing a decommissioned node without -restarting the leader. +Add or remove peers in the cluster's Raft configuration at +runtime. Useful for scaling up or replacing a decommissioned +node without restarting the leader. + +The address is the peer's **Raft** address (its `--raft-bind`), +not its hub port. Membership changes are admin-gated, like +[`ctx hub revoke`](#ctx-hub-revoke): pass `--token` or set +`CTX_HUB_ADMIN_TOKEN`. + +Only the leader can change the configuration. Run the command +against the leader — [`ctx hub status`](#ctx-hub-status) names +it — or the hub answers `not the leader`. + +A node being added must already be running with +`--raft-bind --join`, so that it is waiting for a +configuration instead of bootstrapping one of its own. **Examples**: ```bash -ctx hub peer add host2:9901 -ctx hub peer remove host2:9901 +# On the new node: +ctx hub start --daemon --port 9900 \ + --raft-bind host4:9901 --join + +# On the leader: +ctx hub peer add host4:9901 --token ctx_adm_... +ctx hub peer remove host3:9901 --token ctx_adm_... ``` ### `ctx hub stepdown` @@ -146,10 +213,17 @@ new election among the remaining followers before the current leader steps down. Use before taking the leader offline for maintenance. +Admin-gated like [`ctx hub peer`](#ctx-hub-peer), and +leader-only: a follower answers `not the leader` instead of +reporting a transfer that did not happen. The confirmation +prints after the transfer returns; +[`ctx hub status`](#ctx-hub-status) names the node that won. + **Examples**: ```bash -ctx hub stepdown +ctx hub stepdown --token ctx_adm_... +CTX_HUB_ADMIN_TOKEN=ctx_adm_... ctx hub stepdown ``` ### See Also diff --git a/docs/operations/hub-failure-modes.md b/docs/operations/hub-failure-modes.md index 3048158ef..ec3e2a600 100644 --- a/docs/operations/hub-failure-modes.md +++ b/docs/operations/hub-failure-modes.md @@ -146,9 +146,24 @@ or accept the higher sequence by regenerating `meta.json` from ### Leader Crash, Clean Shutdown -**What happens:** `ctx hub stop` triggers `stepdown` first, so -a new leader is elected before the old one exits. In-flight -writes drain. Clients reconnect to the new leader transparently. +**What happens:** `ctx hub stop` sends SIGTERM; the hub shuts +its Raft node down and drains in-flight RPCs. It does **not** +hand off leadership on its own — the survivors notice the +missing heartbeat and elect a new leader a couple of seconds +later, and clients whose streams were on the old leader have +to be re-run (reconnect is manual, see +[Client Loses Connection Mid-Stream](#client-loses-connection-mid-stream)). + +**What you should do:** for a planned restart, hand off first: + +```bash +ctx hub stepdown --token ctx_adm_... # on the leader +ctx hub status # confirm the new leader +ctx hub stop +``` + +That way the election happens while the old leader is still +serving, instead of after it is gone. ### Leader Crash, Hard Fail (Kill -9, Power Loss) diff --git a/docs/operations/hub.md b/docs/operations/hub.md index ae739770b..aa08517b1 100644 --- a/docs/operations/hub.md +++ b/docs/operations/hub.md @@ -174,20 +174,31 @@ the sequence counter and loses writes. Liveness probe: ```bash -ctx hub status --exit-code +ctx hub status ``` -Exit code `0` means the node is healthy (leader or in-sync -follower); non-zero means degraded. Wire this into your monitoring -of choice. +The command exits non-zero when the RPC fails — hub unreachable, +token rejected — so a wrapper can treat that as the liveness +signal. On a reachable node it prints what the node knows about +itself, and the grading is yours to do from those lines: + +``` +Role: Leader +Leader: 10.0.0.5:9901 +Entries: 1248 Peers: 2 +``` For cluster deployments, watch for: -- **Role flaps**: the leader changing more than once per hour - suggests network instability or disk contention. -- **Replication lag**: `ctx hub status` shows per-peer sequence - offsets. Sustained lag > 100 sequences on a follower is worth - investigating. +- **Role flaps**: the `Role:` line changing more than once per + hour suggests network instability or disk contention. +- **A leader nobody can name**: `Leader: unknown (election in + progress)` is normal for a second or two after a node starts + or a leader dies. Persisting past that, on a node that is + itself reachable, means it cannot see a quorum. +- **A peer count that disagrees between nodes**: each node + reports its own committed Raft configuration, so two nodes + disagreeing on `Peers:` means they never agreed on membership. - **`entries.jsonl` growth rate**: sudden spikes often indicate a misbehaving `ctx connection listen` reconnect loop. diff --git a/docs/recipes/hub-cluster.md b/docs/recipes/hub-cluster.md index d18c5a182..b4d8e119c 100644 --- a/docs/recipes/hub-cluster.md +++ b/docs/recipes/hub-cluster.md @@ -48,7 +48,8 @@ it doubles failure probability without providing quorum. | | | +---v---+ +---v---+ +---v---+ | hub A | | hub B | | hub C | -| :9900 | | :9900 | | :9900 | +| :9900 | | :9900 | | :9900 | gRPC (clients, data sync) +| :9901 | | :9901 | | :9901 | Raft (leader election) +-------+ +-------+ +-------+ ^ ^ ^ +-----------+-----------+ @@ -56,15 +57,30 @@ it doubles failure probability without providing quorum. gRPC (data sync) ``` +Each node runs two listeners: the hub's gRPC port that clients +dial (`--port`), and the Raft port the other nodes dial +(`--raft-bind`). They are separate addresses; the peer list is +made of **Raft** addresses. + ## Step 1: Bootstrap the First Node ```bash ctx hub start --daemon \ --port 9900 \ - --peers hub-b.lan:9900,hub-c.lan:9900 + --raft-bind hub-a.lan:9901 \ + --peers hub-b.lan:9901,hub-c.lan:9901 ``` +`--raft-bind` is the address this node advertises to the other +two, so it has to be a host they can dial: a bare port +(`:9901`) or a wildcard (`0.0.0.0:9901`) is rejected at +startup. Every node's `--raft-bind` appears in the other nodes' +`--peers` lists, and each node bootstraps that same set. + The node starts a Raft election as soon as it sees its peers. +Until a quorum answers, `ctx hub status` reports +`Leader: unknown (election in progress)` — expected while the +other nodes are still coming up. ## Step 2: Start the Other Nodes @@ -73,7 +89,8 @@ On `hub-b.lan`: ```bash ctx hub start --daemon \ --port 9900 \ - --peers hub-a.lan:9900,hub-c.lan:9900 + --raft-bind hub-b.lan:9901 \ + --peers hub-a.lan:9901,hub-c.lan:9901 ``` On `hub-c.lan`: @@ -81,7 +98,8 @@ On `hub-c.lan`: ```bash ctx hub start --daemon \ --port 9900 \ - --peers hub-a.lan:9900,hub-b.lan:9900 + --raft-bind hub-c.lan:9901 \ + --peers hub-a.lan:9901,hub-b.lan:9901 ``` After a few seconds, one node wins the election and becomes the @@ -95,17 +113,32 @@ From any node: ctx hub status ``` -Expected output: +Expected output on the node that won the election: + +``` +Role: Leader +Leader: hub-a.lan:9901 +Entries: 1248 Peers: 2 +``` + +and on either of the others: ``` -role: leader -peers: hub-a.lan:9900 (leader) - hub-b.lan:9900 (follower, in-sync) - hub-c.lan:9900 (follower, in-sync) -entries: 1248 -uptime: 3h42m +Role: Follower +Leader: hub-a.lan:9901 +Entries: 1248 Peers: 2 ``` +The leader is named by its Raft address, which is what the +cluster agrees on. Clients still dial the hub port. + +`Peers:` counts the servers in the committed Raft configuration +other than the one answering, so a three-node cluster reports +two from every node. If the line reads +`Leader: unknown (election in progress)`, Raft has no leader for +the current term: either the election is still running, or the +node you asked cannot see a quorum. + ## Step 4: Register Clients with Failover Peers The `ctx hub *` commands above run on the hub nodes themselves and @@ -129,29 +162,53 @@ always land on the right node. ## Runtime Membership Changes -Add a new peer without downtime: +Membership changes are admin-gated and leader-only: pass +`--token` (or set `CTX_HUB_ADMIN_TOKEN`) and run them against +the leader that `ctx hub status` names. A follower answers +`not the leader` rather than pretending. + +Adding a node is two steps, because a new node must not +bootstrap a configuration of its own — it waits for one: ```bash -ctx hub peer add hub-d.lan:9900 +# On hub-d.lan, the new node: +ctx hub start --daemon \ + --port 9900 \ + --raft-bind hub-d.lan:9901 \ + --join + +# On the leader: +ctx hub peer add hub-d.lan:9901 --token ctx_adm_... ``` -Remove a decommissioned peer: +`ctx hub status` on hub-d.lan then reports `Role: Follower` +and names the leader; every node's `Peers:` count goes up by +one. + +Remove a decommissioned peer, again on the leader: ```bash -ctx hub peer remove hub-c.lan:9900 +ctx hub peer remove hub-c.lan:9901 --token ctx_adm_... ``` +Removal shrinks the quorum, which is the point: a node you +have taken away should stop counting against liveness. Removing +one of three leaves two, and a two-node cluster needs both to +be up — plan the next addition accordingly. + ## Planned Maintenance Before taking a leader offline, hand off leadership: ```bash -ssh hub-a.lan 'ctx hub stepdown' +ssh hub-a.lan 'ctx hub stepdown --token ctx_adm_...' ``` -`stepdown` triggers a new election among the remaining followers -before the leader goes offline. In-flight clients briefly pause, -then reconnect to the new leader. +`stepdown` asks Raft to transfer leadership to a follower that +is caught up, and returns once the transfer completes. Run +`ctx hub status` afterwards to see which node won. Then stop +the old leader; the cluster keeps serving throughout, because +the handoff happened before the process went away. ## Failure Modes at a Glance diff --git a/docs/recipes/index.md b/docs/recipes/index.md index 7f34e91bf..63cf9fcb3 100644 --- a/docs/recipes/index.md +++ b/docs/recipes/index.md @@ -555,6 +555,6 @@ Raft-based leader election across three or more nodes for redundancy. Covers bootstrap, runtime peer management, graceful stepdown, and the Raft-lite durability caveat. -**Uses**: `ctx hub start --peers`, `ctx hub status`, +**Uses**: `ctx hub start --raft-bind --peers`, `ctx hub status`, `ctx hub peer add/remove`, `ctx hub stepdown` diff --git a/internal/assets/commands/commands.yaml b/internal/assets/commands/commands.yaml index f5d7eae65..96b92ad60 100644 --- a/internal/assets/commands/commands.yaml +++ b/internal/assets/commands/commands.yaml @@ -384,9 +384,12 @@ hub.start: file to /hub.pid for later shutdown via `ctx hub stop`. - With --peers, joins a Raft cluster for leader election - across multiple nodes. Data replication happens via - sequence-based gRPC sync on the append-only JSONL log. + With --raft-bind, runs a Raft node for leader election. + The address is what this node advertises to its peers, so + it must name a host they can dial; --peers lists the + --raft-bind addresses of the other nodes. Data replication + happens via sequence-based gRPC sync on the append-only + JSONL log. short: Start the ctx Hub server hub.stop: long: |- @@ -399,12 +402,22 @@ hub.stop: short: Stop a running ctx Hub daemon hub.status: long: |- - Show cluster status: role, peers, sync state, entry - count, and uptime. + Show what the hub reports about itself: node role, + current leader, entry count, and peer count. A hub + running without peers reports Standalone and names + no leader. short: Show cluster status hub.peer: long: |- - Add or remove peers from the cluster at runtime. + Add or remove peers in the cluster's Raft configuration + at runtime. + + Admin-gated: pass --token or set CTX_HUB_ADMIN_TOKEN. + Only the leader can change the configuration, and the + address is the peer's --raft-bind address, not its hub + port. A node being added must already be running with + --join, so it is waiting for a configuration instead of + bootstrapping one of its own. short: Add or remove cluster peers hub.stepdown: long: |- @@ -413,6 +426,11 @@ hub.stepdown: Triggers a new election among the remaining followers before the current leader steps down. Use before taking a node offline for maintenance. + + Admin-gated: pass --token or set CTX_HUB_ADMIN_TOKEN. Run + it against the leader; a follower has no leadership to + hand over. The confirmation prints once the transfer + returns, and `ctx hub status` names the node that won. short: Transfer leadership hub.revoke: long: |- diff --git a/internal/assets/commands/examples.yaml b/internal/assets/commands/examples.yaml index 43288b36a..87f487cd5 100644 --- a/internal/assets/commands/examples.yaml +++ b/internal/assets/commands/examples.yaml @@ -122,7 +122,9 @@ hub.start: ctx hub start # Foreground, port 9900 ctx hub start --port 8080 # Custom port ctx hub start --daemon # Background, writes hub.pid - ctx hub start --peers host2:9900,host3:9900 # Raft cluster member + ctx hub start --raft-bind host1:9901 # Single-node Raft + ctx hub start --raft-bind host1:9901 --peers host2:9901 + ctx hub start --raft-bind host4:9901 --join # Wait to be added hub.stop: short: |2- @@ -134,11 +136,11 @@ hub.status: hub.peer: short: |2- - ctx hub peer add host2:9900 - ctx hub peer remove host2:9900 + ctx hub peer add host2:9901 --token ctx_adm_... + ctx hub peer remove host2:9901 --token ctx_adm_... hub.stepdown: - short: ' ctx hub stepdown' + short: ' ctx hub stepdown --token ctx_adm_...' initialize: short: |2- diff --git a/internal/assets/commands/flags.yaml b/internal/assets/commands/flags.yaml index e8c4d7037..65547f30e 100644 --- a/internal/assets/commands/flags.yaml +++ b/internal/assets/commands/flags.yaml @@ -297,12 +297,20 @@ hub.start.data-dir: short: Hub data directory (default ~/.ctx/hub-data/) hub.start.peers: short: Comma-separated peer addresses for cluster mode +hub.start.raft-bind: + short: Raft address this node binds and advertises to peers +hub.start.join: + short: Wait to be added by a leader instead of bootstrapping hub.start.port: short: Hub listen port (default 9900) hub.stop.data-dir: short: Hub data directory (default ~/.ctx/hub-data/) hub.revoke.token: short: Admin credential from hub startup (or $CTX_HUB_ADMIN_TOKEN) +hub.peer.token: + short: Admin credential from hub startup (or $CTX_HUB_ADMIN_TOKEN) +hub.stepdown.token: + short: Admin credential from hub startup (or $CTX_HUB_ADMIN_TOKEN) watch.dry-run: short: Show updates without applying watch.log: diff --git a/internal/assets/commands/text/errors.yaml b/internal/assets/commands/text/errors.yaml index 5eb53a54e..679441546 100644 --- a/internal/assets/commands/text/errors.yaml +++ b/internal/assets/commands/text/errors.yaml @@ -668,6 +668,12 @@ err.hub.admin-token-required: short: 'admin token required: pass --token or set $CTX_HUB_ADMIN_TOKEN' err.hub.invalid-peer-action: short: "action must be 'add' or 'remove', got %q" +err.hub.raft-bind-required: + short: 'cluster mode needs --raft-bind: the address this node advertises to its peers, e.g. --raft-bind 10.0.0.5:9901' +err.hub.join-with-peers: + short: '--join and --peers are mutually exclusive: a joining node gets its configuration from the leader that adds it, so run --join alone and then `ctx hub peer add ` on the leader' +err.hub.raft-bind-unroutable: + short: '--raft-bind %q is not an address a peer can dial: give a concrete host and port, not a wildcard or bare port' err.serve.no-running-hub: short: 'no running hub: %w' err.serve.invalid-pid: diff --git a/internal/assets/commands/text/write.yaml b/internal/assets/commands/text/write.yaml index 1e1a4c4d6..75f58e39f 100644 --- a/internal/assets/commands/text/write.yaml +++ b/internal/assets/commands/text/write.yaml @@ -1125,6 +1125,8 @@ write.connect-hub-stats: short: 'Entries: %d Clients: %d' write.hub-cluster-stats: short: 'Entries: %d Peers: %d' +write.hub-entries: + short: 'Entries: %d' write.hub-dropped-listeners: short: 'Dropped listeners: %d (slow subscribers disconnected)' write.agent-section-hub: @@ -1141,6 +1143,8 @@ write.hub-leadership-transferred: short: Leadership transferred write.hub-leader: short: 'Leader: %s' +write.hub-leader-unknown: + short: 'Leader: unknown (election in progress)' write.hub-role: short: 'Role: %s' diff --git a/internal/cli/hub/cmd/peer/cmd.go b/internal/cli/hub/cmd/peer/cmd.go index b79386698..c8309fbbf 100644 --- a/internal/cli/hub/cmd/peer/cmd.go +++ b/internal/cli/hub/cmd/peer/cmd.go @@ -10,9 +10,13 @@ import ( "github.com/spf13/cobra" "github.com/ActiveMemory/ctx/internal/assets/read/desc" + coreAdmin "github.com/ActiveMemory/ctx/internal/cli/hub/core/admin" corePeer "github.com/ActiveMemory/ctx/internal/cli/hub/core/peer" "github.com/ActiveMemory/ctx/internal/config/cli" "github.com/ActiveMemory/ctx/internal/config/embed/cmd" + "github.com/ActiveMemory/ctx/internal/config/embed/flag" + cFlag "github.com/ActiveMemory/ctx/internal/config/flag" + "github.com/ActiveMemory/ctx/internal/flagbind" ) // Cmd returns the hub peer subcommand. @@ -20,9 +24,11 @@ import ( // Returns: // - *cobra.Command: The peer subcommand func Cmd() *cobra.Command { + var adminToken string + short, long := desc.Command(cmd.DescKeyHubPeer) - return &cobra.Command{ + c := &cobra.Command{ Use: cmd.UseHubPeer, Short: short, Long: long, @@ -31,6 +37,24 @@ func Cmd() *cobra.Command { // Hub stores at ~/.ctx/hub-data/, not .context/. // Spec: specs/single-source-context-anchor.md. Annotations: map[string]string{cli.AnnotationSkipInit: cli.AnnotationTrue}, - RunE: corePeer.Run, + RunE: func( + cobraCmd *cobra.Command, args []string, + ) error { + token, tokenErr := coreAdmin.Token(adminToken) + if tokenErr != nil { + cobraCmd.SilenceUsage = true + return tokenErr + } + return corePeer.Run( + cobraCmd, args[0], args[1], token, + ) + }, } + + flagbind.StringFlag( + c, &adminToken, + cFlag.Token, flag.DescKeyHubPeerAuth, + ) + + return c } diff --git a/internal/cli/hub/cmd/revoke/cmd.go b/internal/cli/hub/cmd/revoke/cmd.go index 7cf2de2aa..0284397d8 100644 --- a/internal/cli/hub/cmd/revoke/cmd.go +++ b/internal/cli/hub/cmd/revoke/cmd.go @@ -7,18 +7,15 @@ package revoke import ( - "os" - "github.com/spf13/cobra" "github.com/ActiveMemory/ctx/internal/assets/read/desc" + coreAdmin "github.com/ActiveMemory/ctx/internal/cli/hub/core/admin" coreRevoke "github.com/ActiveMemory/ctx/internal/cli/hub/core/revoke" "github.com/ActiveMemory/ctx/internal/config/cli" "github.com/ActiveMemory/ctx/internal/config/embed/cmd" "github.com/ActiveMemory/ctx/internal/config/embed/flag" - "github.com/ActiveMemory/ctx/internal/config/env" cFlag "github.com/ActiveMemory/ctx/internal/config/flag" - errHub "github.com/ActiveMemory/ctx/internal/err/hub" "github.com/ActiveMemory/ctx/internal/flagbind" ) @@ -43,15 +40,10 @@ func Cmd() *cobra.Command { RunE: func( cobraCmd *cobra.Command, args []string, ) error { - // Admin token: --token flag takes precedence, then - // the CTX_HUB_ADMIN_TOKEN environment variable. - token := adminToken - if token == "" { - token = os.Getenv(env.HubAdmin) - } - if token == "" { + token, tokenErr := coreAdmin.Token(adminToken) + if tokenErr != nil { cobraCmd.SilenceUsage = true - return errHub.AdminTokenRequired() + return tokenErr } return coreRevoke.Run(cobraCmd, args[0], token) }, diff --git a/internal/cli/hub/cmd/start/cmd.go b/internal/cli/hub/cmd/start/cmd.go index 7d1ba48e7..0f29c058e 100644 --- a/internal/cli/hub/cmd/start/cmd.go +++ b/internal/cli/hub/cmd/start/cmd.go @@ -21,8 +21,9 @@ import ( // Cmd returns the hub start subcommand. // // Starts the ctx Hub gRPC server either in the foreground or -// as a detached daemon. When --peers is set, joins a Raft -// cluster for leader election. +// as a detached daemon. When --raft-bind is set, runs a Raft +// node for leader election, bootstrapped with the nodes named +// by --peers, or waiting to be added by a leader with --join. // // Returns: // - *cobra.Command: The start subcommand @@ -32,6 +33,8 @@ func Cmd() *cobra.Command { port int dataDir string peersStr string + raftBind string + join bool ) short, long := desc.Command(cmd.DescKeyHubStart) @@ -51,11 +54,13 @@ func Cmd() *cobra.Command { if isDaemon { return server.RunDaemon( cobraCmd, port, dataDir, + raftBind, peersStr, join, ) } peers := server.ParsePeers(peersStr) return server.Run( - cobraCmd, port, dataDir, peers, + cobraCmd, port, dataDir, + raftBind, peers, join, ) }, } @@ -77,6 +82,14 @@ func Cmd() *cobra.Command { c, &peersStr, cFlag.Peers, flag.DescKeyHubStartPeers, ) + flagbind.StringFlag( + c, &raftBind, + cFlag.RaftBind, flag.DescKeyHubStartRaftBind, + ) + flagbind.BoolFlag( + c, &join, + cFlag.Join, flag.DescKeyHubStartJoin, + ) return c } diff --git a/internal/cli/hub/cmd/stepdown/cmd.go b/internal/cli/hub/cmd/stepdown/cmd.go index b915cf0e7..f66d1db7b 100644 --- a/internal/cli/hub/cmd/stepdown/cmd.go +++ b/internal/cli/hub/cmd/stepdown/cmd.go @@ -10,9 +10,13 @@ import ( "github.com/spf13/cobra" "github.com/ActiveMemory/ctx/internal/assets/read/desc" + coreAdmin "github.com/ActiveMemory/ctx/internal/cli/hub/core/admin" coreStep "github.com/ActiveMemory/ctx/internal/cli/hub/core/stepdown" "github.com/ActiveMemory/ctx/internal/config/cli" "github.com/ActiveMemory/ctx/internal/config/embed/cmd" + "github.com/ActiveMemory/ctx/internal/config/embed/flag" + cFlag "github.com/ActiveMemory/ctx/internal/config/flag" + "github.com/ActiveMemory/ctx/internal/flagbind" ) // Cmd returns the hub stepdown subcommand. @@ -20,9 +24,11 @@ import ( // Returns: // - *cobra.Command: The stepdown subcommand func Cmd() *cobra.Command { + var adminToken string + short, long := desc.Command(cmd.DescKeyHubStepdown) - return &cobra.Command{ + c := &cobra.Command{ Use: cmd.UseHubStepdown, Short: short, Long: long, @@ -31,6 +37,22 @@ func Cmd() *cobra.Command { // Hub stores at ~/.ctx/hub-data/, not .context/. // Spec: specs/single-source-context-anchor.md. Annotations: map[string]string{cli.AnnotationSkipInit: cli.AnnotationTrue}, - RunE: coreStep.Run, + RunE: func( + cobraCmd *cobra.Command, _ []string, + ) error { + token, tokenErr := coreAdmin.Token(adminToken) + if tokenErr != nil { + cobraCmd.SilenceUsage = true + return tokenErr + } + return coreStep.Run(cobraCmd, token) + }, } + + flagbind.StringFlag( + c, &adminToken, + cFlag.Token, flag.DescKeyHubStepdownAuth, + ) + + return c } diff --git a/internal/cli/hub/core/admin/admin.go b/internal/cli/hub/core/admin/admin.go new file mode 100644 index 000000000..56acc6d50 --- /dev/null +++ b/internal/cli/hub/core/admin/admin.go @@ -0,0 +1,37 @@ +// / ctx: https://ctx.ist +// ,'`./ do you remember? +// `.,'\ +// \ Copyright 2026-present Context contributors. +// SPDX-License-Identifier: Apache-2.0 + +package admin + +import ( + "os" + + "github.com/ActiveMemory/ctx/internal/config/env" + errHub "github.com/ActiveMemory/ctx/internal/err/hub" +) + +// Token resolves the hub admin credential. +// +// The --token flag takes precedence, then the +// CTX_HUB_ADMIN_TOKEN environment variable. +// +// Parameters: +// - flagValue: value of the command's --token flag +// +// Returns: +// - string: the resolved admin token +// - error: non-nil when neither source supplied one +func Token(flagValue string) (string, error) { + if flagValue != "" { + return flagValue, nil + } + + if fromEnv := os.Getenv(env.HubAdmin); fromEnv != "" { + return fromEnv, nil + } + + return "", errHub.AdminTokenRequired() +} diff --git a/internal/cli/hub/core/admin/doc.go b/internal/cli/hub/core/admin/doc.go new file mode 100644 index 000000000..14129f12a --- /dev/null +++ b/internal/cli/hub/core/admin/doc.go @@ -0,0 +1,30 @@ +// / ctx: https://ctx.ist +// ,'`./ do you remember? +// `.,'\ +// \ Copyright 2026-present Context contributors. +// SPDX-License-Identifier: Apache-2.0 + +// Package admin resolves the hub admin credential for the +// operator-facing ctx hub subcommands. +// +// # Overview +// +// Revoke, peer and stepdown are admin-token-gated on the hub: +// they are operator actions, not client ones. Each command +// accepts the token the same way, and [Token] is that one +// way, so the three cannot drift apart. +// +// # Behavior +// +// [Token] takes the --token flag value and falls back to the +// CTX_HUB_ADMIN_TOKEN environment variable. An empty result +// is an error naming both, rather than an RPC that fails with +// PermissionDenied at the far end. +// +// # Usage +// +// token, tokenErr := admin.Token(flagValue) +// if tokenErr != nil { +// return tokenErr +// } +package admin diff --git a/internal/cli/hub/core/peer/doc.go b/internal/cli/hub/core/peer/doc.go index cbfd3186e..41c46afae 100644 --- a/internal/cli/hub/core/peer/doc.go +++ b/internal/cli/hub/core/peer/doc.go @@ -17,22 +17,27 @@ // // # Behavior // -// [Run] dispatches on the action argument ("add" or "remove") -// to register or deregister a peer address in the cluster. +// [Run] validates the action, dials the hub named by the +// saved connection config, and calls the admin-gated Peer +// RPC. The confirmation prints only after the hub has +// committed the configuration change; a follower answers +// FailedPrecondition, because only the leader can change +// cluster membership. // // # Data Flow // // The peer management pipeline works as follows: // -// 1. The cmd layer invokes [Run] with cobra args -// containing [action, address]. -// 2. [Run] switches on the action string, matching -// against the configured add and remove constants -// from the hub config package. -// 3. For "add", a confirmation message is printed -// via writeHub.PeerAdded. -// 4. For "remove", a confirmation message is printed -// via writeHub.PeerRemoved. -// 5. An invalid action returns an error from the -// hub error package. +// 1. The cmd layer resolves the admin token and invokes +// [Run] with the action and the peer's Raft address. +// 2. [Run] rejects an action that is neither add nor +// remove, using the configured constants from the hub +// config package. +// 3. The connection config supplies the hub address; the +// client dials without a bearer token, since the RPC is +// authenticated by the admin credential. +// 4. The hub applies the change through raft AddVoter or +// RemoveServer and returns. +// 5. writeHub.PeerAdded or writeHub.PeerRemoved reports +// what the cluster accepted. package peer diff --git a/internal/cli/hub/core/peer/peer.go b/internal/cli/hub/core/peer/peer.go index 402aa38af..9505e8014 100644 --- a/internal/cli/hub/core/peer/peer.go +++ b/internal/cli/hub/core/peer/peer.go @@ -7,32 +7,77 @@ package peer import ( + "context" + "github.com/spf13/cobra" + connectCfg "github.com/ActiveMemory/ctx/internal/cli/connection/core/config" cfgHub "github.com/ActiveMemory/ctx/internal/config/hub" + cfgWarn "github.com/ActiveMemory/ctx/internal/config/warn" errHub "github.com/ActiveMemory/ctx/internal/err/hub" + "github.com/ActiveMemory/ctx/internal/hub" + logWarn "github.com/ActiveMemory/ctx/internal/log/warn" writeHub "github.com/ActiveMemory/ctx/internal/write/hub" ) -// Run handles peer add/remove subcommands. +// Run adds or removes a peer in the hub's Raft configuration. +// +// The hub address comes from the saved connection config (same +// as ctx hub status) and authentication uses the admin token, +// so a fresh client is dialed without a bearer token. Only the +// leader can change the configuration; a follower answers +// FailedPrecondition and the message names ctx hub status as +// the way to find the leader. +// +// The address is the peer's --raft-bind address, not its hub +// port: Raft membership is keyed on the Raft transport. // // Parameters: // - cmd: cobra command for output -// - args: [action, address] where action is add or remove +// - action: "add" or "remove" +// - addr: Raft address of the peer +// - adminToken: hub admin token, already resolved // // Returns: -// - error: non-nil if action is invalid -func Run(cmd *cobra.Command, args []string) error { - action := args[0] - addr := args[1] +// - error: non-nil on a bad action, config load, dial, or a +// rejected configuration change +func Run( + cmd *cobra.Command, + action string, + addr string, + adminToken string, +) error { + if action != cfgHub.ActionAdd && + action != cfgHub.ActionRemove { + return errHub.InvalidPeerAction(action) + } + + cfg, loadErr := connectCfg.Load() + if loadErr != nil { + return loadErr + } - switch action { - case cfgHub.ActionAdd: + client, dialErr := hub.NewClient(cfg.HubAddr, "") + if dialErr != nil { + return dialErr + } + defer func() { + if cerr := client.Close(); cerr != nil { + logWarn.Warn(cfgWarn.CloseHubClient, cerr) + } + }() + + if peerErr := client.Peer( + context.Background(), adminToken, action, addr, + ); peerErr != nil { + return peerErr + } + + if action == cfgHub.ActionAdd { writeHub.PeerAdded(cmd, addr) - case cfgHub.ActionRemove: + } else { writeHub.PeerRemoved(cmd, addr) - default: - return errHub.InvalidPeerAction(action) } + return nil } diff --git a/internal/cli/hub/core/server/daemon.go b/internal/cli/hub/core/server/daemon.go index b4e59ebd4..10f83ca8a 100644 --- a/internal/cli/hub/core/server/daemon.go +++ b/internal/cli/hub/core/server/daemon.go @@ -15,7 +15,6 @@ import ( "github.com/spf13/cobra" - cfgFlag "github.com/ActiveMemory/ctx/internal/config/flag" "github.com/ActiveMemory/ctx/internal/config/fs" cfgHub "github.com/ActiveMemory/ctx/internal/config/hub" cfgWarn "github.com/ActiveMemory/ctx/internal/config/warn" @@ -35,11 +34,17 @@ import ( // - cmd: cobra command for output // - port: TCP port to listen on // - dataDir: hub data directory (empty = default) +// - raftBind: Raft address advertised to peers (empty = none) +// - peers: comma-separated peer addresses (empty = no cluster) +// - join: wait to be added instead of bootstrapping // // Returns: // - error: non-nil if fork or PID file write fails func RunDaemon( - cmd *cobra.Command, port int, dataDir string, + cmd *cobra.Command, + port int, + dataDir, raftBind, peers string, + join bool, ) error { if dataDir == "" { defaultDir, dirErr := defaultDataDir() @@ -54,13 +59,10 @@ func RunDaemon( return lookErr } - args := []string{ - cfgHub.ArgHub, cfgHub.ArgStart, - cfgHub.FmtFlagPrefix + cfgFlag.Port, strconv.Itoa(port), - cfgHub.FmtFlagPrefix + cfgFlag.DataDir, dataDir, - } - - pid, startErr := execDaemon.Start(binPath, args) + pid, startErr := execDaemon.Start( + binPath, + daemonArgs(port, dataDir, raftBind, peers, join), + ) if startErr != nil { return startErr } diff --git a/internal/cli/hub/core/server/daemon_test.go b/internal/cli/hub/core/server/daemon_test.go new file mode 100644 index 000000000..375676361 --- /dev/null +++ b/internal/cli/hub/core/server/daemon_test.go @@ -0,0 +1,96 @@ +// / ctx: https://ctx.ist +// ,'`./ do you remember? +// `.,'\ +// \ Copyright 2026-present Context contributors. +// SPDX-License-Identifier: Apache-2.0 + +package server + +import ( + "slices" + "testing" +) + +// TestDaemonArgs_ForwardsClusterFlags pins the flags that turn +// Raft on. The daemon is a re-executed process, so a flag left +// out of this argv is a flag the hub never sees: --peers used to +// be dropped, which started every daemonized hub standalone +// while ctx hub start reported success and the cluster recipe +// said otherwise. +func TestDaemonArgs_ForwardsClusterFlags(t *testing.T) { + args := daemonArgs( + 9900, "/tmp/hub", "10.0.0.5:9901", + "h2:9901,h3:9901", false, + ) + + for flag, want := range map[string]string{ + "--raft-bind": "10.0.0.5:9901", + "--peers": "h2:9901,h3:9901", + } { + idx := slices.Index(args, flag) + if idx < 0 { + t.Fatalf("%s not forwarded: %v", flag, args) + } + if got := args[idx+1]; got != want { + t.Errorf("%s = %q, want %q", flag, got, want) + } + } +} + +// TestDaemonArgs_ForwardsJoin pins the boolean flag, which has +// no value to carry and so is the easiest one to drop. A +// daemonized joiner that lost it would bootstrap a cluster of +// one instead of waiting to be added. +func TestDaemonArgs_ForwardsJoin(t *testing.T) { + args := daemonArgs( + 9900, "/tmp/hub", "10.0.0.5:9901", "", true, + ) + + if !slices.Contains(args, "--join") { + t.Errorf("--join not forwarded: %v", args) + } +} + +// TestDaemonArgs_OmitsClusterFlags keeps a standalone daemon's +// argv as it was: an empty --raft-bind would put the hub into +// cluster mode with no address to advertise. +func TestDaemonArgs_OmitsClusterFlags(t *testing.T) { + args := daemonArgs(9900, "/tmp/hub", "", "", false) + + want := []string{ + "hub", "start", "--port", "9900", + "--data-dir", "/tmp/hub", + } + if !slices.Equal(args, want) { + t.Errorf("args = %v, want %v", args, want) + } +} + +// TestValidateRaftBind pins the addresses Raft cannot advertise. +// raft.NewTCPTransport rejects an unspecified address with an +// error naming neither the flag nor the value, and the wildcard +// forms are exactly what an operator reaches for when told to +// give a bind address. +func TestValidateRaftBind(t *testing.T) { + for _, tc := range []struct { + addr string + wantErr bool + }{ + {"10.0.0.5:9901", false}, + {"hub-a.lan:9901", false}, + {"127.0.0.1:9901", false}, + {"", true}, + {":9901", true}, + {"0.0.0.0:9901", true}, + {"[::]:9901", true}, + {"10.0.0.5", true}, + } { + bindErr := validateRaftBind(tc.addr) + if tc.wantErr && bindErr == nil { + t.Errorf("%q accepted, want rejected", tc.addr) + } + if !tc.wantErr && bindErr != nil { + t.Errorf("%q rejected: %v", tc.addr, bindErr) + } + } +} diff --git a/internal/cli/hub/core/server/run.go b/internal/cli/hub/core/server/run.go index 70c66ed12..9b1af43bd 100644 --- a/internal/cli/hub/core/server/run.go +++ b/internal/cli/hub/core/server/run.go @@ -45,13 +45,26 @@ func DefaultPort() int { return defaultPort } // On first run, generates an admin token and prints it. // On subsequent runs, loads the existing token. // If dataDir is empty, uses ~/.ctx/hub-data/. -// If peers is non-empty, starts Raft cluster for HA. +// +// A raftBind address starts the Raft node that makes +// leadership queryable through the Status RPC; peers are the +// raftBind addresses of the other nodes, so every node +// bootstraps the same configuration. Without raftBind the hub +// runs standalone, and asking for peers without it is an error +// rather than a hub that quietly is not in a cluster. +// +// A joining node bootstraps nothing and waits for a leader to +// add it with ctx hub peer add. That is the only way into an +// existing cluster: a node that bootstrapped its own +// configuration would be a second cluster of one. // // Parameters: // - cmd: cobra command for output // - port: TCP port to listen on // - dataDir: hub data directory (empty = default) -// - peers: peer addresses for cluster mode (may be nil) +// - raftBind: address this node advertises to peers +// - peers: peer Raft addresses (may be nil) +// - join: wait to be added instead of bootstrapping // // Returns: // - error: non-nil if setup or server startup fails @@ -59,7 +72,9 @@ func Run( cmd *cobra.Command, port int, dataDir string, + raftBind string, peers []string, + join bool, ) error { dataDir, resolveErr := resolveDataDir(dataDir) if resolveErr != nil { @@ -80,17 +95,16 @@ func Run( srv := hub.NewServer(store, adminToken) - // Start Raft cluster if peers are configured. - if len(peers) > 0 { - bindAddr := fmt.Sprintf(cfgHub.FmtPort, port+1) - cluster, clusterErr := hub.NewCluster( - fmt.Sprintf(cfgHub.FmtPort, port), - bindAddr, dataDir, peers, - ) - if clusterErr != nil { + // Start the Raft node when the operator named an address + // for it. The node registers under that address as both its + // ID and its transport address, which is the shape peers are + // given, so every node bootstraps the same configuration. + if raftBind != "" || len(peers) > 0 || join { + if clusterErr := startCluster( + srv, dataDir, raftBind, peers, join, + ); clusterErr != nil { return clusterErr } - srv.SetCluster(cluster) } addr := fmt.Sprintf(cfgHub.FmtPort, port) diff --git a/internal/cli/hub/core/server/setup.go b/internal/cli/hub/core/server/setup.go index bfce1fe99..ce556e77b 100644 --- a/internal/cli/hub/core/server/setup.go +++ b/internal/cli/hub/core/server/setup.go @@ -7,14 +7,19 @@ package server import ( + "net" "os" "path/filepath" + "strconv" + "strings" "github.com/spf13/cobra" "github.com/ActiveMemory/ctx/internal/config/dir" + cfgFlag "github.com/ActiveMemory/ctx/internal/config/flag" "github.com/ActiveMemory/ctx/internal/config/fs" cfgHub "github.com/ActiveMemory/ctx/internal/config/hub" + errHub "github.com/ActiveMemory/ctx/internal/err/hub" "github.com/ActiveMemory/ctx/internal/hub" "github.com/ActiveMemory/ctx/internal/io" writeServe "github.com/ActiveMemory/ctx/internal/write/serve" @@ -90,3 +95,121 @@ func loadOrCreateAdmin( return adminToken, nil } + +// daemonArgs builds the argv the background hub is re-executed +// with. The daemon is a fresh process, so a flag missing here is +// a flag the hub never sees: the cluster flags used to be left +// out, which started every daemonized hub standalone while the +// command reported success. +// +// Parameters: +// - port: TCP port to listen on +// - dataDir: resolved hub data directory +// - raftBind: Raft address (empty = standalone) +// - peers: comma-separated peer addresses (empty = none) +// - join: wait to be added instead of bootstrapping +// +// Returns: +// - []string: arguments for the re-executed binary +func daemonArgs( + port int, dataDir, raftBind, peers string, + join bool, +) []string { + args := []string{ + cfgHub.ArgHub, cfgHub.ArgStart, + cfgHub.FmtFlagPrefix + cfgFlag.Port, + strconv.Itoa(port), + cfgHub.FmtFlagPrefix + cfgFlag.DataDir, dataDir, + } + + if raftBind != "" { + args = append(args, + cfgHub.FmtFlagPrefix+cfgFlag.RaftBind, raftBind, + ) + } + + if peers != "" { + args = append(args, + cfgHub.FmtFlagPrefix+cfgFlag.Peers, peers, + ) + } + + if join { + args = append(args, + cfgHub.FmtFlagPrefix+cfgFlag.Join, + ) + } + + return args +} + +// validateRaftBind rejects a Raft address before Raft does. +// +// raft.NewTCPTransport refuses to advertise an unspecified +// address, and the error it returns for one ("local bind address +// is not advertisable") names neither the flag nor the value. +// The check runs here so the operator is told which address was +// rejected and what to pass instead. +// +// Parameters: +// - addr: the --raft-bind value +// +// Returns: +// - error: non-nil if the address is empty or unroutable +func validateRaftBind(addr string) error { + if strings.TrimSpace(addr) == "" { + return errHub.RaftBindRequired() + } + + host, _, splitErr := net.SplitHostPort(addr) + if splitErr != nil || host == "" { + return errHub.RaftBindUnroutable(addr) + } + + if ip := net.ParseIP(host); ip != nil && + ip.IsUnspecified() { + return errHub.RaftBindUnroutable(addr) + } + + return nil +} + +// startCluster validates the cluster flags and attaches a Raft +// node to the server. +// +// Parameters: +// - srv: hub server to attach the cluster to +// - dataDir: resolved hub data directory +// - raftBind: address this node advertises to peers +// - peers: peer Raft addresses (may be nil) +// - join: wait to be added instead of bootstrapping +// +// Returns: +// - error: non-nil if the flags or Raft setup are invalid +func startCluster( + srv *hub.Server, + dataDir, raftBind string, + peers []string, + join bool, +) error { + if bindErr := validateRaftBind(raftBind); bindErr != nil { + return bindErr + } + if join && len(peers) > 0 { + return errHub.JoinWithPeers() + } + + cluster, clusterErr := hub.NewCluster(hub.ClusterConfig{ + NodeID: raftBind, + BindAddr: raftBind, + DataDir: dataDir, + Peers: peers, + Join: join, + }) + if clusterErr != nil { + return clusterErr + } + srv.SetCluster(cluster) + + return nil +} diff --git a/internal/cli/hub/core/status/doc.go b/internal/cli/hub/core/status/doc.go index c0dba6c13..db3e443a0 100644 --- a/internal/cli/hub/core/status/doc.go +++ b/internal/cli/hub/core/status/doc.go @@ -11,13 +11,15 @@ // // This package queries a remote hub for its cluster // state and renders a summary showing the node role, -// address, total entries, and project count. +// the leader, the entry count and the peer count. // // # Behavior // -// [Run] dials the hub via gRPC, retrieves cluster metrics -// (connected clients, entry count, projects), and renders -// a summary showing node role, address, and totals. +// [Run] dials the hub via gRPC and renders what the +// Status RPC reports. The role, the leader and the +// peer count are the hub's answers, read from its +// Raft node: a hub started without peers reports +// Standalone and names no leader. // // # Data Flow // @@ -26,13 +28,13 @@ // 1. Loads connection config to obtain the hub // address and authentication token. // 2. Dials the hub via gRPC using hub.NewClient. -// 3. Calls the Status RPC to retrieve cluster -// metrics including connected clients, total -// entries, and per-project breakdowns. -// 4. Determines the node role: if there are -// connected clients the node is marked active, -// otherwise it is a follower. +// 3. Calls the Status RPC, whose response carries +// the entry count, the listener counts and the +// cluster leadership fields. +// 4. Maps the response onto writeHub's render +// fields: Standalone when no Raft node is +// attached, otherwise Leader or Follower from +// the reported leadership state. // 5. Delegates to writeHub.ClusterStatus to render -// the role, address, entry count, and project -// count for the user. +// the result. package status diff --git a/internal/cli/hub/core/status/info.go b/internal/cli/hub/core/status/info.go new file mode 100644 index 000000000..e7e198dad --- /dev/null +++ b/internal/cli/hub/core/status/info.go @@ -0,0 +1,48 @@ +// / ctx: https://ctx.ist +// ,'`./ do you remember? +// `.,'\ +// \ Copyright 2026-present Context contributors. +// SPDX-License-Identifier: Apache-2.0 + +package status + +import ( + cfgHub "github.com/ActiveMemory/ctx/internal/config/hub" + "github.com/ActiveMemory/ctx/internal/hub" + writeHub "github.com/ActiveMemory/ctx/internal/write/hub" +) + +// renderInfo maps a Status response onto the rendered fields. +// +// The role, the leader and the peer count are the hub's answers, +// not the client's: a hub with no Raft node reports Standalone +// and has neither a leader nor peers to name. +// +// Parameters: +// - resp: response from the hub Status RPC +// +// Returns: +// - writeHub.ClusterStatusInfo: fields to render +func renderInfo( + resp *hub.StatusResponse, +) writeHub.ClusterStatusInfo { + info := writeHub.ClusterStatusInfo{ + Role: cfgHub.RoleStandalone, + Entries: resp.TotalEntries, + Dropped: resp.DroppedListeners, + } + + if !resp.ClusterEnabled { + return info + } + + info.Clustered = true + info.Role = cfgHub.RoleFollower + if resp.IsLeader { + info.Role = cfgHub.RoleLeader + } + info.Leader = resp.LeaderAddr + info.Peers = resp.ClusterPeers + + return info +} diff --git a/internal/cli/hub/core/status/status.go b/internal/cli/hub/core/status/status.go index 8412506f8..3404d1181 100644 --- a/internal/cli/hub/core/status/status.go +++ b/internal/cli/hub/core/status/status.go @@ -12,7 +12,6 @@ import ( "github.com/spf13/cobra" connectCfg "github.com/ActiveMemory/ctx/internal/cli/connection/core/config" - cfgHub "github.com/ActiveMemory/ctx/internal/config/hub" cfgWarn "github.com/ActiveMemory/ctx/internal/config/warn" "github.com/ActiveMemory/ctx/internal/hub" logWarn "github.com/ActiveMemory/ctx/internal/log/warn" @@ -52,16 +51,7 @@ func Run(cmd *cobra.Command, _ []string) error { return statusErr } - role := cfgHub.RoleFollower - if resp.ConnectedClients > 0 { - role = cfgHub.RoleActive - } + writeHub.ClusterStatus(cmd, renderInfo(resp)) - writeHub.ClusterStatus( - cmd, role, cfg.HubAddr, - resp.TotalEntries, - len(resp.EntriesByProject), - resp.DroppedListeners, - ) return nil } diff --git a/internal/cli/hub/core/status/status_test.go b/internal/cli/hub/core/status/status_test.go new file mode 100644 index 000000000..f2a462450 --- /dev/null +++ b/internal/cli/hub/core/status/status_test.go @@ -0,0 +1,87 @@ +// / ctx: https://ctx.ist +// ,'`./ do you remember? +// `.,'\ +// \ Copyright 2026-present Context contributors. +// SPDX-License-Identifier: Apache-2.0 + +package status + +import ( + "testing" + + cfgHub "github.com/ActiveMemory/ctx/internal/config/hub" + "github.com/ActiveMemory/ctx/internal/hub" +) + +// TestRenderInfo_Standalone pins the case the old renderer got +// wrong in every field: a hub with no Raft node used to report +// its role from the listener count, name itself the leader, and +// report the project count as a peer count. +func TestRenderInfo_Standalone(t *testing.T) { + info := renderInfo(&hub.StatusResponse{ + TotalEntries: 42, + ConnectedClients: 3, + EntriesByProject: map[string]uint64{ + "alpha": 1, "beta": 2, + }, + }) + + if info.Role != cfgHub.RoleStandalone { + t.Errorf("role = %q, want %q", + info.Role, cfgHub.RoleStandalone) + } + if info.Clustered { + t.Error("standalone response rendered as clustered") + } + if info.Leader != "" { + t.Errorf("leader = %q, want empty", info.Leader) + } + if info.Peers != 0 { + t.Errorf("peers = %d, want 0 (projects are not peers)", + info.Peers) + } +} + +// TestRenderInfo_Leader covers the node that answers yes to +// "am I the leader?". +func TestRenderInfo_Leader(t *testing.T) { + info := renderInfo(&hub.StatusResponse{ + TotalEntries: 42, + ClusterEnabled: true, + IsLeader: true, + LeaderAddr: "10.0.0.5:9901", + ClusterPeers: 2, + }) + + if info.Role != cfgHub.RoleLeader { + t.Errorf("role = %q, want %q", + info.Role, cfgHub.RoleLeader) + } + if !info.Clustered { + t.Error("clustered response rendered as standalone") + } + if info.Leader != "10.0.0.5:9901" { + t.Errorf("leader = %q, want the reported address", + info.Leader) + } + if info.Peers != 2 { + t.Errorf("peers = %d, want 2", info.Peers) + } +} + +// TestRenderInfo_FollowerWithNoListeners is the inversion of the +// old heuristic: a follower with subscribers used to read Active, +// and a leader with none read Follower. The role now comes from +// Raft, so the listener count does not move it. +func TestRenderInfo_FollowerWithNoListeners(t *testing.T) { + info := renderInfo(&hub.StatusResponse{ + ClusterEnabled: true, + LeaderAddr: "10.0.0.5:9901", + ClusterPeers: 2, + }) + + if info.Role != cfgHub.RoleFollower { + t.Errorf("role = %q, want %q", + info.Role, cfgHub.RoleFollower) + } +} diff --git a/internal/cli/hub/core/stepdown/doc.go b/internal/cli/hub/core/stepdown/doc.go index e8331f60c..12e431d35 100644 --- a/internal/cli/hub/core/stepdown/doc.go +++ b/internal/cli/hub/core/stepdown/doc.go @@ -17,20 +17,23 @@ // // # Behavior // -// [Run] signals the current hub node to relinquish its -// leader role and prints a confirmation once the transfer -// is initiated. +// [Run] asks the hub to hand leadership to a follower and +// prints a confirmation once the transfer returns. Only a +// leader can transfer leadership: a follower answers +// FailedPrecondition rather than reporting a handoff that +// did not happen. // // # Data Flow // // The stepdown pipeline works as follows: // -// 1. The cmd layer invokes [Run] with the cobra -// command and unused args. -// 2. [Run] calls writeHub.SteppedDown to print a -// confirmation message indicating the node has -// initiated leadership transfer. -// 3. The function returns nil on success. Future -// implementations may add gRPC calls to -// coordinate the transfer with the cluster. +// 1. The cmd layer resolves the admin token and invokes +// [Run]. +// 2. The connection config supplies the hub address; the +// client dials without a bearer token, since the RPC is +// authenticated by the admin credential. +// 3. The hub calls raft LeadershipTransfer and returns +// when it completes or fails. +// 4. writeHub.SteppedDown reports the handoff. Which node +// won is answered by ctx hub status. package stepdown diff --git a/internal/cli/hub/core/stepdown/stepdown.go b/internal/cli/hub/core/stepdown/stepdown.go index a3429bec9..b3c437779 100644 --- a/internal/cli/hub/core/stepdown/stepdown.go +++ b/internal/cli/hub/core/stepdown/stepdown.go @@ -7,20 +7,60 @@ package stepdown import ( + "context" + "github.com/spf13/cobra" + connectCfg "github.com/ActiveMemory/ctx/internal/cli/connection/core/config" + cfgWarn "github.com/ActiveMemory/ctx/internal/config/warn" + "github.com/ActiveMemory/ctx/internal/hub" + logWarn "github.com/ActiveMemory/ctx/internal/log/warn" writeHub "github.com/ActiveMemory/ctx/internal/write/hub" ) -// Run requests leadership transfer from the current node. +// Run asks the hub to hand leadership to another node. +// +// The hub address comes from the saved connection config (same +// as ctx hub status) and authentication uses the admin token, +// so a fresh client is dialed without a bearer token. Only a +// leader can transfer leadership: asking a follower fails with +// FailedPrecondition rather than printing a confirmation for +// something that did not happen. +// +// The confirmation prints only after the transfer returns; run +// ctx hub status to see which node won. // // Parameters: // - cmd: cobra command for output -// - args: unused (cobra signature) +// - adminToken: hub admin token, already resolved // // Returns: -// - error: non-nil if transfer fails -func Run(cmd *cobra.Command, _ []string) error { +// - error: non-nil if config load, dial, or transfer fails +func Run( + cmd *cobra.Command, adminToken string, +) error { + cfg, loadErr := connectCfg.Load() + if loadErr != nil { + return loadErr + } + + client, dialErr := hub.NewClient(cfg.HubAddr, "") + if dialErr != nil { + return dialErr + } + defer func() { + if cerr := client.Close(); cerr != nil { + logWarn.Warn(cfgWarn.CloseHubClient, cerr) + } + }() + + if stepErr := client.Stepdown( + context.Background(), adminToken, + ); stepErr != nil { + return stepErr + } + writeHub.SteppedDown(cmd) + return nil } diff --git a/internal/config/embed/flag/hub.go b/internal/config/embed/flag/hub.go index e86867dc7..0be057777 100644 --- a/internal/config/embed/flag/hub.go +++ b/internal/config/embed/flag/hub.go @@ -16,8 +16,18 @@ const ( DescKeyHubStartDaemon = "hub.start.daemon" // DescKeyHubStartPeers is the text key for hub start --peers. DescKeyHubStartPeers = "hub.start.peers" + // DescKeyHubStartRaftBind is the text key for hub start + // --raft-bind. + DescKeyHubStartRaftBind = "hub.start.raft-bind" + // DescKeyHubStartJoin is the text key for hub start --join. + DescKeyHubStartJoin = "hub.start.join" // DescKeyHubStopDataDir is the text key for hub stop --data-dir. DescKeyHubStopDataDir = "hub.stop.data-dir" // DescKeyHubRevokeAuth is the text key for hub revoke --token. DescKeyHubRevokeAuth = "hub.revoke.token" + // DescKeyHubPeerAuth is the text key for hub peer --token. + DescKeyHubPeerAuth = "hub.peer.token" + // DescKeyHubStepdownAuth is the text key for hub stepdown + // --token. + DescKeyHubStepdownAuth = "hub.stepdown.token" ) diff --git a/internal/config/embed/text/err_hub.go b/internal/config/embed/text/err_hub.go index bec042e2b..b887a3a9a 100644 --- a/internal/config/embed/text/err_hub.go +++ b/internal/config/embed/text/err_hub.go @@ -23,6 +23,15 @@ const ( // DescKeyErrHubAdminRequired is the text key for an // admin-gated command invoked with no admin token supplied. DescKeyErrHubAdminRequired = "err.hub.admin-token-required" + // DescKeyErrHubRaftBindRequired is the text key for a + // cluster start with no --raft-bind address. + DescKeyErrHubRaftBindRequired = "err.hub.raft-bind-required" + // DescKeyErrHubRaftBindUnroutable is the text key for a + // --raft-bind address peers could not dial. + DescKeyErrHubRaftBindUnroutable = "err.hub.raft-bind-unroutable" + // DescKeyErrHubJoinWithPeers is the text key for a start + // that asks to join and to bootstrap at the same time. + DescKeyErrHubJoinWithPeers = "err.hub.join-with-peers" // DescKeyErrHubInvalidPeerAction is the text key for // unrecognized peer action errors. DescKeyErrHubInvalidPeerAction = "err.hub.invalid-peer-action" diff --git a/internal/config/embed/text/write_hub.go b/internal/config/embed/text/write_hub.go index d20923ebc..7ca4064ad 100644 --- a/internal/config/embed/text/write_hub.go +++ b/internal/config/embed/text/write_hub.go @@ -23,9 +23,17 @@ const ( // DescKeyWriteHubRole is the text key for hub role // messages. DescKeyWriteHubRole = "write.hub-role" + // DescKeyWriteHubLeaderUnknown is the text key for the + // leader line while no leader is known: an election is in + // progress, or quorum was lost. + DescKeyWriteHubLeaderUnknown = "write.hub-leader-unknown" // DescKeyWriteHubClusterStats is the text key for hub // cluster statistics. DescKeyWriteHubClusterStats = "write.hub-cluster-stats" + // DescKeyWriteHubEntries is the text key for the entry + // count on a standalone hub, where the cluster-stats line + // would report a peer count that does not exist. + DescKeyWriteHubEntries = "write.hub-entries" // DescKeyWriteHubDroppedListeners is the text key for the // cumulative slow-listener disconnect count. Printed only // when the count is non-zero. diff --git a/internal/config/flag/flag.go b/internal/config/flag/flag.go index e8f7c0170..57d2ccf6e 100644 --- a/internal/config/flag/flag.go +++ b/internal/config/flag/flag.go @@ -76,6 +76,7 @@ const ( Hook = "hook" JSON = "json" JSONFile = "json-file" + Join = "join" KeepFrontmatter = "keep-frontmatter" Key = "key" Label = "label" @@ -98,6 +99,7 @@ const ( Project = "project" Prompt = "prompt" Quiet = "quiet" + RaftBind = "raft-bind" Raw = "raw" Record = "record" Regenerate = "regenerate" diff --git a/internal/config/hub/doc.go b/internal/config/hub/doc.go index 63dd522b5..9fe90502b 100644 --- a/internal/config/hub/doc.go +++ b/internal/config/hub/doc.go @@ -88,7 +88,8 @@ // // - ArgHub, ArgStart: re-exec argument tokens // - ActionAdd, ActionRemove: peer action names -// - RoleFollower, RoleActive: status role labels +// - RoleFollower, RoleLeader, RoleStandalone: status +// role labels, reported from Raft state // - ReplicateInterval (5s): follower retry // interval // - HubSyncTimeout (10s): session-start pull diff --git a/internal/config/hub/hub.go b/internal/config/hub/hub.go index 072ddca97..7948ab9c5 100644 --- a/internal/config/hub/hub.go +++ b/internal/config/hub/hub.go @@ -29,6 +29,10 @@ const ( MethodStatus = "Status" // MethodRevoke is the Revoke RPC method name. MethodRevoke = "Revoke" + // MethodPeer is the Peer RPC method name. + MethodPeer = "Peer" + // MethodStepdown is the Stepdown RPC method name. + MethodStepdown = "Stepdown" ) // Full gRPC method paths (ServicePath + MethodName). @@ -45,6 +49,10 @@ const ( PathStatus = ServicePath + MethodStatus // PathRevoke is the full gRPC path for Revoke. PathRevoke = ServicePath + MethodRevoke + // PathPeer is the full method path for Peer. + PathPeer = ServicePath + MethodPeer + // PathStepdown is the full method path for Stepdown. + PathStepdown = ServicePath + MethodStepdown ) // Authorization header. @@ -137,12 +145,18 @@ const ( FileAdminToken = "admin.token" ) -// Status role labels. +// Status role labels. Reported from Raft state, so a hub +// started without peers is Standalone rather than a follower +// of nothing. const ( - // RoleFollower is the role label for a follower node. + // RoleFollower is the role label for a clustered node that + // is not the current leader. RoleFollower = "Follower" - // RoleActive is the role label for an active node. - RoleActive = "Active" + // RoleLeader is the role label for the current Raft leader. + RoleLeader = "Leader" + // RoleStandalone is the role label for a hub running with + // no Raft node attached. + RoleStandalone = "Standalone" ) // Address formatting. @@ -214,6 +228,18 @@ const ( ErrInvalidAdminToken = "invalid admin token" // ErrProjectNameRequired is the gRPC error for missing project name. ErrProjectNameRequired = "project_name required" + // ErrClusterDisabled is the gRPC error for a cluster command + // sent to a hub running without a Raft node. + ErrClusterDisabled = "cluster not enabled: " + + "hub started without --raft-bind" + // ErrNotLeader is the gRPC error for a configuration change + // or leadership transfer asked of a node that is not the + // leader. + ErrNotLeader = "not the leader: run this against the " + + "leader named by ctx hub status" + // ErrPeerAddressRequired is the gRPC error for a peer change + // with no address. + ErrPeerAddressRequired = "address required" // ErrClientIDRequired is the gRPC error for a missing client ID // on the Revoke RPC. ErrClientIDRequired = "client_id required" diff --git a/internal/config/warn/warn.go b/internal/config/warn/warn.go index a574d9c4e..b30c6ace4 100644 --- a/internal/config/warn/warn.go +++ b/internal/config/warn/warn.go @@ -122,6 +122,14 @@ const ( HubFanOutSlowListener = "hub fanout: disconnected slow listener " + "(buffer full); cumulative disconnects: %d" + // HubClusterPeers is the stderr format for a failed Raft + // configuration read inside the Status RPC. Takes the error. + // Status is a diagnostic: losing the peer count is not a + // reason to deny the operator the rest of the response, so + // the handler warns and reports zero peers instead of + // failing the call. + HubClusterPeers = "hub cluster: read raft configuration: %v" + // HubReplicateAppend is the stderr format for a failed // [Store.Append] inside the follower replication stream. The // loop is best-effort and has no return path, so a dropped diff --git a/internal/err/hub/hub.go b/internal/err/hub/hub.go index e012d0b40..3a7186860 100644 --- a/internal/err/hub/hub.go +++ b/internal/err/hub/hub.go @@ -81,6 +81,47 @@ func AdminTokenRequired() error { ) } +// RaftBindRequired returns an error when cluster mode is +// requested without a Raft bind address. Raft refuses to +// advertise a wildcard address, so there is no address to +// derive: the operator has to name the one their peers dial. +// +// Returns: +// - error: guidance on supplying --raft-bind +func RaftBindRequired() error { + return errors.New( + desc.Text(text.DescKeyErrHubRaftBindRequired), + ) +} + +// JoinWithPeers returns an error when a start asks both to +// join an existing cluster and to bootstrap one. A joining +// node has no configuration of its own by design. +// +// Returns: +// - error: guidance on the join flow +func JoinWithPeers() error { + return errors.New( + desc.Text(text.DescKeyErrHubJoinWithPeers), + ) +} + +// RaftBindUnroutable returns an error for a Raft bind address +// no peer could dial: a bare port, an empty host, or a +// wildcard such as 0.0.0.0 or [::]. +// +// Parameters: +// - addr: the rejected address +// +// Returns: +// - error: "--raft-bind is not an address a peer can +// dial: ..." +func RaftBindUnroutable(addr string) error { + return fmt.Errorf( + desc.Text(text.DescKeyErrHubRaftBindUnroutable), addr, + ) +} + // InvalidPeerAction returns an error for an unrecognized // peer action. // diff --git a/internal/hub/client.go b/internal/hub/client.go index 6e204257f..e82eeb317 100644 --- a/internal/hub/client.go +++ b/internal/hub/client.go @@ -96,6 +96,63 @@ func (c *Client) Revoke( ) } +// Peer calls the Peer RPC to change cluster membership. +// +// Authenticated with the admin token, like Revoke: the client +// is dialed without a bearer token. +// +// Parameters: +// - ctx: context for the call +// - adminToken: hub admin credential +// - action: "add" or "remove" +// - addr: Raft address of the peer +// +// Returns: +// - error: non-nil if the configuration change fails +func (c *Client) Peer( + ctx context.Context, + adminToken string, + action string, + addr string, +) error { + resp := &PeerResponse{} + return c.conn.Invoke( + ctx, + cfgHub.PathPeer, + &PeerRequest{ + AdminToken: adminToken, + Action: action, + Address: addr, + }, + resp, + ) +} + +// Stepdown calls the Stepdown RPC, asking the leader to hand +// leadership to another node. +// +// Authenticated with the admin token, like Revoke. +// +// Parameters: +// - ctx: context for the call +// - adminToken: hub admin credential +// +// Returns: +// - error: non-nil if the transfer fails or the node is not +// the leader +func (c *Client) Stepdown( + ctx context.Context, + adminToken string, +) error { + resp := &StepdownResponse{} + return c.conn.Invoke( + ctx, + cfgHub.PathStepdown, + &StepdownRequest{AdminToken: adminToken}, + resp, + ) +} + // Publish calls the Publish RPC. // // Parameters: diff --git a/internal/hub/cluster.go b/internal/hub/cluster.go index 050298b53..694f04036 100644 --- a/internal/hub/cluster.go +++ b/internal/hub/cluster.go @@ -7,6 +7,7 @@ package hub import ( + "errors" "net" "os" "path/filepath" @@ -27,21 +28,26 @@ import ( // replicated via sequence-based gRPC sync. Raft only // determines which node is the current master. // +// A joining node ([ClusterConfig.Join]) skips bootstrap and +// waits to be added by a leader, which is the only way a new +// node can enter an existing cluster: a node that bootstraps +// its own configuration would be a second cluster of one, not +// a member of the first. +// // Parameters: -// - nodeID: unique identifier for this node -// - bindAddr: address for Raft communication -// - dataDir: directory for Raft state -// - peers: other cluster nodes (empty = single node) +// - cfg: node identity, transport address, state directory +// and the servers to bootstrap with // // Returns: // - *Cluster: initialized Raft cluster node // - error: non-nil if setup fails func NewCluster( - nodeID string, - bindAddr string, - dataDir string, - peers []string, + clusterCfg ClusterConfig, ) (*Cluster, error) { + nodeID := clusterCfg.NodeID + bindAddr := clusterCfg.BindAddr + dataDir := clusterCfg.DataDir + peers := clusterCfg.Peers raftDir := filepath.Join(dataDir, cfgHub.RaftDir) if mkErr := io.SafeMkdirAll( raftDir, fs.PermKeyDir, @@ -87,35 +93,41 @@ func NewCluster( return nil, raftErr } - // Bootstrap if single node or first startup. - if len(peers) == 0 { - config := raft.Configuration{ - Servers: []raft.Server{ - { - ID: raft.ServerID(nodeID), - Address: raft.ServerAddress(bindAddr), - }, - }, - } - r.BootstrapCluster(config) - } else { - servers := make( - []raft.Server, 0, len(peers)+1, - ) + // A joining node has no configuration of its own: the + // leader that adds it sends one. + if clusterCfg.Join { + return &Cluster{ + raftNode: r, + transport: transport, + }, nil + } + + // Bootstrap this node plus any configured peers. A single + // node bootstraps a one-server cluster and elects itself. + servers := make([]raft.Server, 0, len(peers)+1) + servers = append(servers, raft.Server{ + ID: raft.ServerID(nodeID), + Address: raft.ServerAddress(bindAddr), + }) + for _, p := range peers { servers = append(servers, raft.Server{ - ID: raft.ServerID(nodeID), - Address: raft.ServerAddress(bindAddr), + ID: raft.ServerID(p), + Address: raft.ServerAddress(p), }) - for _, p := range peers { - servers = append(servers, raft.Server{ - ID: raft.ServerID(p), - Address: raft.ServerAddress(p), - }) - } - config := raft.Configuration{ - Servers: servers, - } - r.BootstrapCluster(config) + } + + // ErrCantBootstrap is what a restart against existing Raft + // state returns: that node is already bootstrapped and the + // on-disk configuration wins. Any other error leaves a node + // that never elects anyone, which is precisely the state the + // Status RPC's leadership fields exist to make visible -- so + // it surfaces here instead of being discarded. + bootErr := r.BootstrapCluster(raft.Configuration{ + Servers: servers, + }).Error() + if bootErr != nil && + !errors.Is(bootErr, raft.ErrCantBootstrap) { + return nil, bootErr } return &Cluster{ @@ -132,13 +144,82 @@ func (c *Cluster) IsLeader() bool { return c.raftNode.State() == raft.Leader } -// LeaderAddr returns the address of the current leader. +// LeaderAddr returns the Raft address of the current leader. +// +// Empty while no leader is known: during an election, or once +// quorum is lost. // // Returns: // - string: leader address, or empty if unknown func (c *Cluster) LeaderAddr() string { - _, id := c.raftNode.LeaderWithID() - return string(id) + addr, _ := c.raftNode.LeaderWithID() + return string(addr) +} + +// Peers reports how many servers other than this node the +// committed Raft configuration holds. A single-node cluster +// reports zero. +// +// Returns: +// - uint32: server count excluding this node +// - error: non-nil if the configuration read fails +func (c *Cluster) Peers() (uint32, error) { + future := c.raftNode.GetConfiguration() + if cfgErr := future.Error(); cfgErr != nil { + return 0, cfgErr + } + + // Counted rather than converted from len() so the wire type + // needs no int-to-uint32 narrowing. + var peers uint32 + for range future.Configuration().Servers { + peers++ + } + if peers > 0 { + peers-- + } + + return peers, nil +} + +// AddPeer adds a voting server to the cluster configuration. +// +// The address is both the new server's ID and its transport +// address, matching how every node registers itself. Only the +// leader can change the configuration; a follower returns +// [raft.ErrNotLeader]. The added node must be running in join +// mode ([ClusterConfig.Join]), or it is already a cluster of +// its own and will not accept this one's configuration. +// +// Parameters: +// - addr: Raft address of the server to add +// +// Returns: +// - error: non-nil if the configuration change fails +func (c *Cluster) AddPeer(addr string) error { + return c.raftNode.AddVoter( + raft.ServerID(addr), + raft.ServerAddress(addr), + 0, 0, + ).Error() +} + +// RemovePeer removes a server from the cluster configuration. +// +// Only the leader can change the configuration; a follower +// returns [raft.ErrNotLeader]. Removing a server shrinks the +// quorum, which is what makes a decommissioned node stop +// counting against liveness. +// +// Parameters: +// - addr: Raft address of the server to remove +// +// Returns: +// - error: non-nil if the configuration change fails +func (c *Cluster) RemovePeer(addr string) error { + return c.raftNode.RemoveServer( + raft.ServerID(addr), 0, 0, + ).Error() } // Stepdown transfers leadership to another node. diff --git a/internal/hub/cluster_test.go b/internal/hub/cluster_test.go new file mode 100644 index 000000000..d64778eb6 --- /dev/null +++ b/internal/hub/cluster_test.go @@ -0,0 +1,444 @@ +// / ctx: https://ctx.ist +// ,'`./ do you remember? +// `.,'\ +// \ Copyright 2026-present Context contributors. +// SPDX-License-Identifier: Apache-2.0 + +package hub + +import ( + "testing" + "time" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + cfgHub "github.com/ActiveMemory/ctx/internal/config/hub" +) + +// clusterNodeID is the Raft ServerID the test node registers +// under. It is deliberately not an address: LeaderWithID returns +// (address, id), and an ID that cannot be mistaken for the bound +// address is what makes the LeaderAddr assertions meaningful. +const clusterNodeID = "test-node" + +// singleNodeCluster starts a one-server Raft node on a free +// loopback port and blocks until it elects itself. A single node +// is quorum on its own, so the election is deterministic; the +// deadline only bounds how long the test waits for the default +// election timeout to fire. +// +// Returns the cluster and the address it bound, which is the +// leader address every caller asserts against. +func singleNodeCluster(t *testing.T) (*Cluster, string) { + t.Helper() + + return newSingleNode(t, clusterNodeID, freeAddrs(t, 1)[0]) +} + +// singleNodeClusterAt is singleNodeCluster on a caller-chosen +// address, registered under that address the way +// ctx hub start --raft-bind registers a node. +func singleNodeClusterAt( + t *testing.T, addr string, +) (*Cluster, string) { + t.Helper() + + return newSingleNode(t, addr, addr) +} + +// newSingleNode boots one self-electing Raft node and blocks +// until it has elected itself. +func newSingleNode( + t *testing.T, nodeID, addr string, +) (*Cluster, string) { + t.Helper() + + cluster, clusterErr := NewCluster(ClusterConfig{ + NodeID: nodeID, + BindAddr: addr, + DataDir: t.TempDir(), + }) + if clusterErr != nil { + t.Fatal(clusterErr) + } + t.Cleanup(func() { + if shutErr := cluster.Shutdown(); shutErr != nil { + t.Log(shutErr) + } + }) + + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + if cluster.IsLeader() { + return cluster, addr + } + time.Sleep(10 * time.Millisecond) + } + + t.Fatal("raft node never became leader") + + return nil, "" +} + +// TestCluster_PeersExcludesSelf pins what the Peers: line counts. +// The committed configuration of a bootstrapped single node holds +// exactly one server — itself — so the peer count is zero, not +// one. +func TestCluster_PeersExcludesSelf(t *testing.T) { + cluster, _ := singleNodeCluster(t) + + peers, peersErr := cluster.Peers() + if peersErr != nil { + t.Fatal(peersErr) + } + if peers != 0 { + t.Errorf("peers = %d, want 0 for a single node", peers) + } +} + +// TestCluster_LeaderAddrIsTheAddress pins the method to its name. +// LeaderWithID returns (address, id); the method used to discard +// the address and return the ID. The node registers under an ID +// that is not an address, so returning the ID again fails here. +func TestCluster_LeaderAddrIsTheAddress(t *testing.T) { + cluster, addr := singleNodeCluster(t) + + if leader := cluster.LeaderAddr(); leader != addr { + t.Errorf("leader addr = %q, want %q", + leader, addr) + } +} + +// TestHubStatus_StandaloneReportsClusterDisabled pins the +// disambiguator. A hub with no Raft node must report +// ClusterEnabled false: without that flag, a standalone hub and a +// clustered node that has lost its leader are indistinguishable, +// since both report IsLeader false and an empty LeaderAddr. +func TestHubStatus_StandaloneReportsClusterDisabled( + t *testing.T, +) { + srv := listenTestServer(t) + + resp, statusErr := srv.hubStatus(testCtx()) + if statusErr != nil { + t.Fatal(statusErr) + } + + if resp.ClusterEnabled { + t.Error("cluster reported enabled with no cluster set") + } + if resp.IsLeader { + t.Error("standalone hub claims leadership") + } + if resp.LeaderAddr != "" { + t.Errorf("leader addr = %q, want empty", + resp.LeaderAddr) + } + if resp.ClusterPeers != 0 { + t.Errorf("peers = %d, want 0", resp.ClusterPeers) + } +} + +// TestHubStatus_ClusterReportsLeadershipState is the contract +// issue #96 asks for: with a Raft node attached, Status answers +// the leadership question instead of leaving the operator to +// guess from a role derived from the listener count. +func TestHubStatus_ClusterReportsLeadershipState(t *testing.T) { + srv := listenTestServer(t) + cluster, addr := singleNodeCluster(t) + srv.SetCluster(cluster) + + resp, statusErr := srv.hubStatus(testCtx()) + if statusErr != nil { + t.Fatal(statusErr) + } + + if !resp.ClusterEnabled { + t.Error("cluster reported disabled with a cluster set") + } + if !resp.IsLeader { + t.Error("elected node does not report leadership") + } + if resp.LeaderAddr != addr { + t.Errorf("leader addr = %q, want %q", + resp.LeaderAddr, addr) + } + if resp.ClusterPeers != 0 { + t.Errorf("peers = %d, want 0 for a single node", + resp.ClusterPeers) + } +} + +// freeAddrs returns n loopback addresses with nothing listening +// on them. Raft binds these itself, so the listeners only serve +// to have the kernel pick free ports. +func freeAddrs(t *testing.T, n int) []string { + t.Helper() + + addrs := make([]string, 0, n) + for range n { + lis := listenRandom(t) + addrs = append(addrs, lis.Addr().String()) + if closeErr := lis.Close(); closeErr != nil { + t.Fatal(closeErr) + } + } + + return addrs +} + +// startCluster brings up one Raft node per address, each +// bootstrapped with the full server list. Every node registers +// under its own address as both ID and transport address, which +// is the shape ctx hub start --raft-bind gives them. +func startCluster(t *testing.T, addrs []string) []*Cluster { + t.Helper() + + clusters := make([]*Cluster, 0, len(addrs)) + for i, addr := range addrs { + peers := make([]string, 0, len(addrs)-1) + for j, peer := range addrs { + if i != j { + peers = append(peers, peer) + } + } + + cluster, clusterErr := NewCluster(ClusterConfig{ + NodeID: addr, + BindAddr: addr, + DataDir: t.TempDir(), + Peers: peers, + }) + if clusterErr != nil { + t.Fatal(clusterErr) + } + t.Cleanup(func() { + if shutErr := cluster.Shutdown(); shutErr != nil { + t.Log(shutErr) + } + }) + clusters = append(clusters, cluster) + } + + return clusters +} + +// TestCluster_ThreeNodesElectOneLeader is the contract the +// Status fields exist to report, over a real three-node Raft +// cluster: exactly one node leads, all three name the same +// leader, and each counts the other two as peers. +// +// It is also the regression test for the address arithmetic +// this branch removed. Binding Raft to ":port+1" and advertising +// the same wildcard made NewCluster fail with "local bind +// address is not advertisable" on every node, so no ctx cluster +// could start at all. +func TestCluster_ThreeNodesElectOneLeader(t *testing.T) { + addrs := freeAddrs(t, 3) + clusters := startCluster(t, addrs) + + leader := waitForLeader(t, clusters) + + for i, cluster := range clusters { + if got := cluster.LeaderAddr(); got != leader { + t.Errorf("node %d names leader %q, want %q", + i, got, leader) + } + + peers, peersErr := cluster.Peers() + if peersErr != nil { + t.Fatal(peersErr) + } + if peers != 2 { + t.Errorf("node %d reports %d peers, want 2", + i, peers) + } + } +} + +// waitForLeader blocks until exactly one node reports leadership +// and every node agrees on its address, then returns that +// address. +func waitForLeader( + t *testing.T, clusters []*Cluster, +) string { + t.Helper() + + deadline := time.Now().Add(20 * time.Second) + for time.Now().Before(deadline) { + leaders := 0 + addr := "" + agreed := true + + for _, cluster := range clusters { + if cluster.IsLeader() { + leaders++ + } + switch { + case cluster.LeaderAddr() == "": + agreed = false + case addr == "": + addr = cluster.LeaderAddr() + case cluster.LeaderAddr() != addr: + agreed = false + } + } + + if leaders == 1 && agreed { + return addr + } + time.Sleep(50 * time.Millisecond) + } + + t.Fatal("cluster never settled on a single leader") + + return "" +} + +// joinCluster starts a Raft node that bootstraps nothing and +// waits to be added by a leader, which is what +// ctx hub start --join does. +func joinCluster(t *testing.T, addr string) *Cluster { + t.Helper() + + cluster, clusterErr := NewCluster(ClusterConfig{ + NodeID: addr, + BindAddr: addr, + DataDir: t.TempDir(), + Join: true, + }) + if clusterErr != nil { + t.Fatal(clusterErr) + } + t.Cleanup(func() { + if shutErr := cluster.Shutdown(); shutErr != nil { + t.Log(shutErr) + } + }) + + return cluster +} + +// waitFor polls cond until it holds or the deadline passes. +func waitFor(t *testing.T, what string, cond func() bool) { + t.Helper() + + deadline := time.Now().Add(20 * time.Second) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(50 * time.Millisecond) + } + + t.Fatalf("timed out waiting for %s", what) +} + +// TestCluster_PeerAddJoinsNode is what ctx hub peer add has to +// mean: a node started in join mode holds no configuration +// until the leader adds it, and then it is a member -- it +// counts as a peer and it names the same leader. Before this +// change the command printed "Added peer" and reached nothing. +func TestCluster_PeerAddJoinsNode(t *testing.T) { + addrs := freeAddrs(t, 2) + leader, leaderAddr := singleNodeClusterAt(t, addrs[0]) + joiner := joinCluster(t, addrs[1]) + + if addErr := leader.AddPeer(addrs[1]); addErr != nil { + t.Fatal(addErr) + } + + waitFor(t, "the joiner to follow the leader", func() bool { + return joiner.LeaderAddr() == leaderAddr + }) + + peers, peersErr := leader.Peers() + if peersErr != nil { + t.Fatal(peersErr) + } + if peers != 1 { + t.Errorf("leader reports %d peers, want 1", peers) + } + if joiner.IsLeader() { + t.Error("the joiner elected itself instead of joining") + } +} + +// TestCluster_PeerRemoveShrinksCluster pins the other half: +// removing a decommissioned node takes it back out of the +// committed configuration, so it stops counting toward quorum. +func TestCluster_PeerRemoveShrinksCluster(t *testing.T) { + addrs := freeAddrs(t, 2) + leader, leaderAddr := singleNodeClusterAt(t, addrs[0]) + joiner := joinCluster(t, addrs[1]) + + if addErr := leader.AddPeer(addrs[1]); addErr != nil { + t.Fatal(addErr) + } + waitFor(t, "the joiner to follow the leader", func() bool { + return joiner.LeaderAddr() == leaderAddr + }) + + if remErr := leader.RemovePeer(addrs[1]); remErr != nil { + t.Fatal(remErr) + } + + waitFor(t, "the peer count to drop", func() bool { + peers, peersErr := leader.Peers() + return peersErr == nil && peers == 0 + }) +} + +// TestCluster_StepdownHandsOffLeadership pins ctx hub stepdown: +// the node that was leading is not leading afterwards, and the +// other node is. The command used to print "Leadership +// transferred" without asking anyone. +func TestCluster_StepdownHandsOffLeadership(t *testing.T) { + addrs := freeAddrs(t, 2) + clusters := startCluster(t, addrs) + waitForLeader(t, clusters) + + var leader, other *Cluster + if clusters[0].IsLeader() { + leader, other = clusters[0], clusters[1] + } else { + leader, other = clusters[1], clusters[0] + } + + if stepErr := leader.Stepdown(); stepErr != nil { + t.Fatal(stepErr) + } + + waitFor(t, "leadership to move", func() bool { + return !leader.IsLeader() && other.IsLeader() + }) +} + +// TestPeer_FollowerIsPrecondition pins the error mapping that +// makes a follower's refusal actionable: raft.ErrNotLeader +// becomes FailedPrecondition with a message pointing at +// ctx hub status, not an opaque Internal. +func TestPeer_FollowerIsPrecondition(t *testing.T) { + addrs := freeAddrs(t, 2) + clusters := startCluster(t, addrs) + waitForLeader(t, clusters) + + follower := clusters[0] + if follower.IsLeader() { + follower = clusters[1] + } + + srv := listenTestServer(t) + srv.SetCluster(follower) + + _, peerErr := srv.peer(testCtx(), &PeerRequest{ + AdminToken: srv.adminToken, + Action: cfgHub.ActionRemove, + Address: addrs[0], + }) + + if got := status.Code(peerErr); got != codes.FailedPrecondition { + t.Errorf("code = %v, want FailedPrecondition", got) + } +} diff --git a/internal/hub/doc.go b/internal/hub/doc.go index 5c9826599..5e1f26009 100644 --- a/internal/hub/doc.go +++ b/internal/hub/doc.go @@ -17,7 +17,8 @@ // - Storage ([Store]): append-only JSONL with // sequence numbers and per-client tokens. // - Transport ([Server]): gRPC Register / Publish -// / Sync / Listen / Status RPCs. +// / Sync / Listen / Status / Revoke / Peer / +// Stepdown RPCs. // - Cluster ([Cluster]): HashiCorp Raft for leader // election only (see Raft-Lite below). // - Client ([Client]): connection registration, @@ -72,6 +73,30 @@ // mutex) and bumps a cumulative counter reported as // DroppedListeners by the Status RPC. // +// # Cluster Leadership +// +// [Cluster] runs Raft for leader election only. The +// Status RPC reports that election: ClusterEnabled +// says whether a Raft node is attached at all, and +// IsLeader, LeaderAddr and ClusterPeers describe the +// current term when one is. Without ClusterEnabled a +// standalone hub could not be told apart from a +// clustered node that has lost its leader, since both +// report no leadership. +// +// Two admin-token-gated RPCs change that state rather +// than report it. Peer adds or removes a server +// ([Cluster.AddPeer], [Cluster.RemovePeer]) and +// Stepdown hands leadership to a follower +// ([Cluster.Stepdown]). Both are leader-only: raft +// refuses a configuration change or a transfer on a +// follower, and the handler turns that into a +// FailedPrecondition naming ctx hub status. A node +// being added starts with [ClusterConfig.Join] set, +// bootstrapping nothing, because a node that +// bootstraps its own configuration is a second +// cluster of one rather than a member of the first. +// // # Encryption // // Client-side connection state is encrypted at rest diff --git a/internal/hub/err_check.go b/internal/hub/err_check.go index 450771e86..7f0f85823 100644 --- a/internal/hub/err_check.go +++ b/internal/hub/err_check.go @@ -7,6 +7,9 @@ package hub import ( + "errors" + + "github.com/hashicorp/raft" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" @@ -23,6 +26,30 @@ var errSlowListener = status.Error( codes.ResourceExhausted, cfgHub.ErrSlowListener, ) +// clusterOpErr maps a Raft configuration or leadership error +// onto a gRPC status. +// +// raft.ErrNotLeader is a precondition the operator can act on +// -- ask the leader instead -- so it travels as +// FailedPrecondition with an answer, rather than as an opaque +// internal error. Everything else is the hub's problem, not +// the caller's. +// +// Parameters: +// - err: error from a Cluster configuration or transfer call +// +// Returns: +// - error: gRPC status carrying the right code +func clusterOpErr(err error) error { + if errors.Is(err, raft.ErrNotLeader) { + return status.Error( + codes.FailedPrecondition, cfgHub.ErrNotLeader, + ) + } + + return status.Error(codes.Internal, err.Error()) +} + // authErr reports whether err is an authentication or // authorization failure. // diff --git a/internal/hub/grpc.go b/internal/hub/grpc.go index c8437944a..8d811c1d7 100644 --- a/internal/hub/grpc.go +++ b/internal/hub/grpc.go @@ -58,6 +58,14 @@ func serviceDesc(s *Server) *grpc.ServiceDesc { MethodName: cfgHub.MethodRevoke, Handler: makeRevokeHandler(s), }, + { + MethodName: cfgHub.MethodPeer, + Handler: makePeerHandler(s), + }, + { + MethodName: cfgHub.MethodStepdown, + Handler: makeStepdownHandler(s), + }, }, Streams: []grpc.StreamDesc{ { @@ -128,6 +136,52 @@ func makeRevokeHandler(s *Server) grpc.MethodHandler { } } +// makePeerHandler creates the Peer handler. +// Peer uses admin token auth, not bearer (matches Register): +// cluster membership is an operator action. +// +// Parameters: +// - s: hub server for request dispatch +// +// Returns: +// - grpc.MethodHandler: unary handler for Peer RPC +func makePeerHandler(s *Server) grpc.MethodHandler { + return func( + _ any, ctx context.Context, + dec func(any) error, + _ grpc.UnaryServerInterceptor, + ) (any, error) { + req := &PeerRequest{} + if decErr := dec(req); decErr != nil { + return nil, decErr + } + return s.peer(ctx, req) + } +} + +// makeStepdownHandler creates the Stepdown handler. +// Stepdown uses admin token auth, not bearer (matches +// Register): handing off leadership is an operator action. +// +// Parameters: +// - s: hub server for request dispatch +// +// Returns: +// - grpc.MethodHandler: unary handler for Stepdown RPC +func makeStepdownHandler(s *Server) grpc.MethodHandler { + return func( + _ any, ctx context.Context, + dec func(any) error, + _ grpc.UnaryServerInterceptor, + ) (any, error) { + req := &StepdownRequest{} + if decErr := dec(req); decErr != nil { + return nil, decErr + } + return s.stepdown(ctx, req) + } +} + // makePublishHandler creates the Publish handler. // // Parameters: diff --git a/internal/hub/handler.go b/internal/hub/handler.go index c9fc2ab9b..33291be0d 100644 --- a/internal/hub/handler.go +++ b/internal/hub/handler.go @@ -14,7 +14,9 @@ import ( "google.golang.org/grpc/status" cfgHub "github.com/ActiveMemory/ctx/internal/config/hub" + cfgWarn "github.com/ActiveMemory/ctx/internal/config/warn" errHub "github.com/ActiveMemory/ctx/internal/err/hub" + logWarn "github.com/ActiveMemory/ctx/internal/log/warn" ) // register handles the Register RPC. @@ -106,6 +108,106 @@ func (s *Server) revoke( return &RevokeResponse{}, nil } +// peer handles the Peer RPC. +// +// Admin-token-gated (mirrors register): changing cluster +// membership is an operator action, not a client one. Only the +// leader can commit a configuration change, so a follower +// answers FailedPrecondition rather than silently doing +// nothing -- which is what the CLI did before this RPC existed. +// +// Parameters: +// - ctx: request context (unused) +// - req: peer request with admin token, action and address +// +// Returns: +// - *PeerResponse: empty on success +// - error: PermissionDenied on bad admin token, +// FailedPrecondition with no cluster or on a follower, +// InvalidArgument on a bad action or empty address +func (s *Server) peer( + _ context.Context, req *PeerRequest, +) (*PeerResponse, error) { + if req.AdminToken != s.adminToken { + return nil, status.Error( + codes.PermissionDenied, + cfgHub.ErrInvalidAdminToken, + ) + } + if s.cluster == nil { + return nil, status.Error( + codes.FailedPrecondition, + cfgHub.ErrClusterDisabled, + ) + } + if req.Address == "" { + return nil, status.Error( + codes.InvalidArgument, + cfgHub.ErrPeerAddressRequired, + ) + } + + var changeErr error + switch req.Action { + case cfgHub.ActionAdd: + changeErr = s.cluster.AddPeer(req.Address) + case cfgHub.ActionRemove: + changeErr = s.cluster.RemovePeer(req.Address) + default: + return nil, status.Error( + codes.InvalidArgument, + errHub.InvalidPeerAction(req.Action).Error(), + ) + } + if changeErr != nil { + return nil, clusterOpErr(changeErr) + } + + return &PeerResponse{}, nil +} + +// stepdown handles the Stepdown RPC. +// +// Admin-token-gated (mirrors register). Leadership transfer is +// a leader-only operation; asking a follower is a precondition +// failure, not a no-op. +// +// Parameters: +// - ctx: request context (unused) +// - req: stepdown request with admin token +// +// Returns: +// - *StepdownResponse: empty once the transfer completes +// - error: PermissionDenied on bad admin token, +// FailedPrecondition with no cluster or on a follower +func (s *Server) stepdown( + _ context.Context, req *StepdownRequest, +) (*StepdownResponse, error) { + if req.AdminToken != s.adminToken { + return nil, status.Error( + codes.PermissionDenied, + cfgHub.ErrInvalidAdminToken, + ) + } + if s.cluster == nil { + return nil, status.Error( + codes.FailedPrecondition, + cfgHub.ErrClusterDisabled, + ) + } + if !s.cluster.IsLeader() { + return nil, status.Error( + codes.FailedPrecondition, cfgHub.ErrNotLeader, + ) + } + + if transferErr := s.cluster.Stepdown(); transferErr != nil { + return nil, clusterOpErr(transferErr) + } + + return &StepdownResponse{}, nil +} + // publish handles the Publish RPC. // // Parameters: @@ -246,21 +348,41 @@ func (s *Server) listenEntries( // hubStatus handles the Status RPC. // +// The cluster fields stay at their zero values when no Raft node +// is attached; ClusterEnabled tells the caller which case it is +// looking at. A failed configuration read warns to stderr and +// reports zero peers rather than failing the call: Status is a +// diagnostic, and the rest of the response is still worth having. +// // Parameters: // - ctx: request context (unused) // // Returns: -// - *StatusResponse: hub statistics +// - *StatusResponse: hub statistics and cluster leadership // - error: always nil func (s *Server) hubStatus( _ context.Context, ) (*StatusResponse, error) { total, byType, byProject := s.store.Stats() - return &StatusResponse{ + resp := &StatusResponse{ TotalEntries: total, ConnectedClients: s.listeners.count(), DroppedListeners: s.listeners.droppedCount(), EntriesByType: byType, EntriesByProject: byProject, - }, nil + } + + if s.cluster != nil { + resp.ClusterEnabled = true + resp.IsLeader = s.cluster.IsLeader() + resp.LeaderAddr = s.cluster.LeaderAddr() + + peers, peersErr := s.cluster.Peers() + if peersErr != nil { + logWarn.Warn(cfgWarn.HubClusterPeers, peersErr) + } + resp.ClusterPeers = peers + } + + return resp, nil } diff --git a/internal/hub/handler_test.go b/internal/hub/handler_test.go index 4f1d6aa40..d1ed73a08 100644 --- a/internal/hub/handler_test.go +++ b/internal/hub/handler_test.go @@ -18,6 +18,7 @@ import ( "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + cfgHub "github.com/ActiveMemory/ctx/internal/config/hub" logWarn "github.com/ActiveMemory/ctx/internal/log/warn" ) @@ -165,3 +166,92 @@ func TestListenEntries_ContextCancelEndsStream(t *testing.T) { t.Errorf("droppedCount = %d, want 0 for a clean exit", got) } } + +// TestPeer_RejectsBadAdminToken keeps membership an operator +// action: the Peer RPC is gated the way Register is, so a +// client token cannot reshape the cluster. +func TestPeer_RejectsBadAdminToken(t *testing.T) { + srv := listenTestServer(t) + + _, peerErr := srv.peer(testCtx(), &PeerRequest{ + AdminToken: "not-the-admin-token", + Action: cfgHub.ActionAdd, + Address: "10.0.0.6:9901", + }) + + if got := status.Code(peerErr); got != codes.PermissionDenied { + t.Errorf("code = %v, want PermissionDenied", got) + } +} + +// TestPeer_NoClusterIsPrecondition answers the standalone hub +// honestly. Printing "Added peer" against a hub with no Raft +// node is what the command did before it had an RPC. +func TestPeer_NoClusterIsPrecondition(t *testing.T) { + srv := listenTestServer(t) + + _, peerErr := srv.peer(testCtx(), &PeerRequest{ + AdminToken: srv.adminToken, + Action: cfgHub.ActionAdd, + Address: "10.0.0.6:9901", + }) + + if got := status.Code(peerErr); got != codes.FailedPrecondition { + t.Errorf("code = %v, want FailedPrecondition", got) + } +} + +// TestPeer_ValidatesRequest covers the two malformed shapes: +// an action that is neither add nor remove, and an empty +// address. +func TestPeer_ValidatesRequest(t *testing.T) { + srv := listenTestServer(t) + cluster, _ := singleNodeCluster(t) + srv.SetCluster(cluster) + + for name, req := range map[string]*PeerRequest{ + "unknown action": { + AdminToken: srv.adminToken, + Action: "promote", + Address: "10.0.0.6:9901", + }, + "empty address": { + AdminToken: srv.adminToken, + Action: cfgHub.ActionAdd, + }, + } { + _, peerErr := srv.peer(testCtx(), req) + if got := status.Code(peerErr); got != codes.InvalidArgument { + t.Errorf("%s: code = %v, want InvalidArgument", + name, got) + } + } +} + +// TestStepdown_RejectsBadAdminToken gates leadership transfer +// the same way. +func TestStepdown_RejectsBadAdminToken(t *testing.T) { + srv := listenTestServer(t) + + _, stepErr := srv.stepdown(testCtx(), &StepdownRequest{ + AdminToken: "not-the-admin-token", + }) + + if got := status.Code(stepErr); got != codes.PermissionDenied { + t.Errorf("code = %v, want PermissionDenied", got) + } +} + +// TestStepdown_NoClusterIsPrecondition pins the standalone +// answer: there is no leadership to hand over. +func TestStepdown_NoClusterIsPrecondition(t *testing.T) { + srv := listenTestServer(t) + + _, stepErr := srv.stepdown(testCtx(), &StepdownRequest{ + AdminToken: srv.adminToken, + }) + + if got := status.Code(stepErr); got != codes.FailedPrecondition { + t.Errorf("code = %v, want FailedPrecondition", got) + } +} diff --git a/internal/hub/types.go b/internal/hub/types.go index d20340758..394aae184 100644 --- a/internal/hub/types.go +++ b/internal/hub/types.go @@ -274,20 +274,63 @@ type EntryMsg struct { Meta EntryMeta `json:"meta"` } +// PeerRequest is the input for the Peer RPC. +// +// Fields: +// - AdminToken: admin credential (same gate as Register) +// - Action: "add" or "remove" +// - Address: Raft address of the peer to add or remove +type PeerRequest struct { + AdminToken string `json:"admin_token"` + Action string `json:"action"` + Address string `json:"address"` +} + +// PeerResponse is the output of the Peer RPC. Empty: the +// configuration change either committed or returned an error. +type PeerResponse struct{} + +// StepdownRequest is the input for the Stepdown RPC. +// +// Fields: +// - AdminToken: admin credential (same gate as Register) +type StepdownRequest struct { + AdminToken string `json:"admin_token"` +} + +// StepdownResponse is the output of the Stepdown RPC. Empty: +// the transfer either completed or returned an error, and the +// node that won is reported by the next Status call. +type StepdownResponse struct{} + // StatusResponse is the output of the Status RPC. // +// The cluster fields are zero values on a hub started without +// peers, where no Raft node exists. ClusterEnabled is the +// disambiguator: without it a standalone hub is indistinguishable +// from a clustered node that has lost its leader, since both +// report IsLeader false and an empty LeaderAddr. +// // Fields: // - TotalEntries: total number of entries // - ConnectedClients: active listener count // - DroppedListeners: cumulative slow-listener disconnects // - EntriesByType: entry count per type // - EntriesByProject: entry count per origin project +// - ClusterEnabled: a Raft node is attached to this hub +// - IsLeader: this node is the current Raft leader +// - LeaderAddr: Raft address of the leader, empty if unknown +// - ClusterPeers: Raft servers other than this node type StatusResponse struct { TotalEntries uint64 `json:"total_entries"` ConnectedClients uint32 `json:"connected_clients"` DroppedListeners uint64 `json:"dropped_listeners"` EntriesByType map[string]uint64 `json:"entries_by_type"` EntriesByProject map[string]uint64 `json:"entries_by_project"` + ClusterEnabled bool `json:"cluster_enabled"` + IsLeader bool `json:"is_leader,omitempty"` + LeaderAddr string `json:"leader_addr,omitempty"` + ClusterPeers uint32 `json:"cluster_peers,omitempty"` } // Client is a gRPC client for the ctx Hub. @@ -300,6 +343,26 @@ type Client struct { token string } +// ClusterConfig is the input for [NewCluster]. +// +// Join and Peers are mutually exclusive: a node either +// bootstraps a configuration (itself, plus Peers) or waits for +// a leader to send it one. +// +// Fields: +// - NodeID: Raft ServerID for this node +// - BindAddr: address the Raft transport binds and advertises +// - DataDir: hub data directory holding the Raft state +// - Peers: other servers to bootstrap with (empty = alone) +// - Join: skip bootstrap and wait to be added by a leader +type ClusterConfig struct { + NodeID string + BindAddr string + DataDir string + Peers []string + Join bool +} + // Cluster wraps a Raft node for leader election only. // // Fields: diff --git a/internal/write/hub/cluster.go b/internal/write/hub/cluster.go new file mode 100644 index 000000000..da55a0be8 --- /dev/null +++ b/internal/write/hub/cluster.go @@ -0,0 +1,42 @@ +// / ctx: https://ctx.ist +// ,'`./ do you remember? +// `.,'\ +// \ Copyright 2026-present Context contributors. +// SPDX-License-Identifier: Apache-2.0 + +package hub + +import ( + "fmt" + + "github.com/spf13/cobra" + + "github.com/ActiveMemory/ctx/internal/assets/read/desc" + "github.com/ActiveMemory/ctx/internal/config/embed/text" +) + +// clusterLines prints the leader and the combined entry/peer +// line for a hub running with a Raft node attached. +// +// Parameters: +// - cmd: Cobra command for output +// - info: status fields as reported by the Status RPC +func clusterLines( + cmd *cobra.Command, info ClusterStatusInfo, +) { + if info.Leader == "" { + cmd.Println(desc.Text( + text.DescKeyWriteHubLeaderUnknown, + )) + } else { + cmd.Println(fmt.Sprintf( + desc.Text(text.DescKeyWriteHubLeader), + info.Leader, + )) + } + + cmd.Println(fmt.Sprintf( + desc.Text(text.DescKeyWriteHubClusterStats), + info.Entries, info.Peers, + )) +} diff --git a/internal/write/hub/doc.go b/internal/write/hub/doc.go index 7602ae62c..e64f34808 100644 --- a/internal/write/hub/doc.go +++ b/internal/write/hub/doc.go @@ -9,9 +9,12 @@ // // # Cluster Status // -// [ClusterStatus] prints the full cluster dashboard: -// the current node role (Leader or Follower), the -// leader address, total entry count, and peer count. +// [ClusterStatus] prints the status dashboard from a +// [ClusterStatusInfo], whose fields all come from the +// hub's Status RPC. A standalone hub prints its role +// (Standalone) and entry count; a clustered one adds +// the leader address -- or an election-in-progress +// note while Raft has no leader -- and the peer count. // It also prints the cumulative slow-listener // disconnect count, but only when that count is // non-zero, so a healthy hub keeps its former output. @@ -35,9 +38,13 @@ // // # Usage // -// hub.ClusterStatus( -// cmd, role, leader, entries, peers, dropped, -// ) +// hub.ClusterStatus(cmd, hub.ClusterStatusInfo{ +// Role: role, +// Clustered: true, +// Leader: leaderAddr, +// Entries: entries, +// Peers: peers, +// }) // hub.PeerAdded(cmd, peerAddr) // hub.SteppedDown(cmd) package hub diff --git a/internal/write/hub/hub.go b/internal/write/hub/hub.go index 01a9138c8..e828af4e7 100644 --- a/internal/write/hub/hub.go +++ b/internal/write/hub/hub.go @@ -15,38 +15,38 @@ import ( "github.com/ActiveMemory/ctx/internal/config/embed/text" ) -// ClusterStatus prints cluster role and stats. The dropped-listener -// line is omitted when the count is zero so a healthy hub keeps its -// current output. +// ClusterStatus prints the node role and hub statistics. +// +// A standalone hub prints its role and entry count and stops +// there: it has no leader and no peers to report. A clustered +// hub adds the leader address -- or a note that an election is +// in progress when Raft has no leader yet -- and the peer +// count. The dropped-listener line is omitted when the count is +// zero, so a healthy hub keeps its current output. // // Parameters: // - cmd: Cobra command for output -// - role: current node role (Leader/Follower) -// - leader: leader address -// - entries: total entry count -// - peers: number of peers -// - dropped: cumulative slow-listener disconnects +// - info: status fields as reported by the Status RPC func ClusterStatus( - cmd *cobra.Command, - role, leader string, - entries uint64, - peers int, - dropped uint64, + cmd *cobra.Command, info ClusterStatusInfo, ) { cmd.Println(fmt.Sprintf( - desc.Text(text.DescKeyWriteHubRole), role, - )) - cmd.Println(fmt.Sprintf( - desc.Text(text.DescKeyWriteHubLeader), leader, - )) - cmd.Println(fmt.Sprintf( - desc.Text(text.DescKeyWriteHubClusterStats), - entries, peers, + desc.Text(text.DescKeyWriteHubRole), info.Role, )) - if dropped > 0 { + + if info.Clustered { + clusterLines(cmd, info) + } else { + cmd.Println(fmt.Sprintf( + desc.Text(text.DescKeyWriteHubEntries), + info.Entries, + )) + } + + if info.Dropped > 0 { cmd.Println(fmt.Sprintf( desc.Text(text.DescKeyWriteHubDroppedListeners), - dropped, + info.Dropped, )) } } diff --git a/internal/write/hub/hub_test.go b/internal/write/hub/hub_test.go index e830e9423..381660809 100644 --- a/internal/write/hub/hub_test.go +++ b/internal/write/hub/hub_test.go @@ -17,22 +17,94 @@ import ( ) // clusterStatus renders ClusterStatus into a buffer. -func clusterStatus(dropped uint64) string { +func clusterStatus( + info writeHub.ClusterStatusInfo, +) string { var buf bytes.Buffer cmd := &cobra.Command{} cmd.SetOut(&buf) - writeHub.ClusterStatus( - cmd, "leader", "127.0.0.1:9901", 42, 2, dropped, - ) + writeHub.ClusterStatus(cmd, info) return buf.String() } +// clustered returns the info a healthy three-node leader +// reports. +func clustered() writeHub.ClusterStatusInfo { + return writeHub.ClusterStatusInfo{ + Role: "Leader", + Clustered: true, + Leader: "127.0.0.1:9901", + Entries: 42, + Peers: 2, + } +} + +// TestClusterStatus_Cluster pins the clustered shape: the role, +// the leader address the hub reported, and the peer count. +func TestClusterStatus_Cluster(t *testing.T) { + out := clusterStatus(clustered()) + + for _, want := range []string{ + "Role: Leader", + "Leader: 127.0.0.1:9901", + "Entries: 42 Peers: 2", + } { + if !strings.Contains(out, want) { + t.Errorf("want %q, got:\n%s", want, out) + } + } +} + +// TestClusterStatus_Standalone pins the standalone shape. A hub +// with no Raft node has no leader and no peers; printing either +// would be inventing them, which is what the old renderer did. +func TestClusterStatus_Standalone(t *testing.T) { + out := clusterStatus(writeHub.ClusterStatusInfo{ + Role: "Standalone", + Entries: 42, + }) + + if !strings.Contains(out, "Role: Standalone") { + t.Errorf("want the standalone role, got:\n%s", out) + } + if !strings.Contains(out, "Entries: 42") { + t.Errorf("want the entry count, got:\n%s", out) + } + if strings.Contains(out, "Leader") { + t.Errorf("want no leader line, got:\n%s", out) + } + if strings.Contains(out, "Peers") { + t.Errorf("want no peer count, got:\n%s", out) + } +} + +// TestClusterStatus_LeaderUnknown covers the clustered node with +// no leader: mid-election, or after quorum loss. The line has to +// say so rather than render an empty address. +func TestClusterStatus_LeaderUnknown(t *testing.T) { + info := clustered() + info.Role = "Follower" + info.Leader = "" + + out := clusterStatus(info) + + if !strings.Contains(out, "Leader: unknown") { + t.Errorf("want the unknown-leader line, got:\n%s", out) + } + if !strings.Contains(out, "Entries: 42 Peers: 2") { + t.Errorf("want the stats line intact, got:\n%s", out) + } +} + // TestClusterStatus_DroppedListeners pins the conditional // slow-listener line. desc.Text returns "" for an unknown key, so a // renamed text key would silently blank the line; asserting on the // rendered count catches that. func TestClusterStatus_DroppedListeners(t *testing.T) { - out := clusterStatus(3) + info := clustered() + info.Dropped = 3 + + out := clusterStatus(info) if !strings.Contains(out, "Dropped listeners: 3") { t.Errorf("want dropped-listener line with count, got:\n%s", out) @@ -42,7 +114,7 @@ func TestClusterStatus_DroppedListeners(t *testing.T) { // TestClusterStatus_NoDroppedListeners pins the omission at zero so // a healthy hub's output stays what it was. func TestClusterStatus_NoDroppedListeners(t *testing.T) { - out := clusterStatus(0) + out := clusterStatus(clustered()) if strings.Contains(out, "Dropped listeners") { t.Errorf("want no dropped-listener line at zero, got:\n%s", out) diff --git a/internal/write/hub/types.go b/internal/write/hub/types.go new file mode 100644 index 000000000..eae7b090d --- /dev/null +++ b/internal/write/hub/types.go @@ -0,0 +1,30 @@ +// / ctx: https://ctx.ist +// ,'`./ do you remember? +// `.,'\ +// \ Copyright 2026-present Context contributors. +// SPDX-License-Identifier: Apache-2.0 + +package hub + +// ClusterStatusInfo carries what [ClusterStatus] renders. +// +// Clustered decides the shape: a hub with no Raft node has no +// leader and no peers, and printing either would be inventing +// them. Every field comes from the Status RPC response, not +// from the client's view of the connection. +// +// Fields: +// - Role: node role label (Leader, Follower, Standalone) +// - Clustered: a Raft node is attached to the hub +// - Leader: Raft address of the leader, empty if unknown +// - Entries: total entry count in the hub's store +// - Peers: Raft servers other than the one answering +// - Dropped: cumulative slow-listener disconnects +type ClusterStatusInfo struct { + Role string + Clustered bool + Leader string + Entries uint64 + Peers uint32 + Dropped uint64 +} diff --git a/specs/hub-status-cluster-leadership.md b/specs/hub-status-cluster-leadership.md new file mode 100644 index 000000000..fa0c2740f --- /dev/null +++ b/specs/hub-status-cluster-leadership.md @@ -0,0 +1,335 @@ +# Hub Status: Cluster Leadership + +The hub keeps a Raft `Cluster` on `Server` and shuts it down in +`GracefulStop`, but nothing ever reads its leadership state. The +Status RPC does not carry it, so `ctx hub status` cannot answer +"who is the leader?" — and the three lines it does print about +cluster state are fabricated from data that has nothing to do +with Raft. Upstream issue: +[ActiveMemory/ctx#96](https://github.com/ActiveMemory/ctx/issues/96). + +## Problem + +### The leadership state no RPC carries — `internal/hub/handler.go` + +`Cluster.IsLeader()` and `Cluster.LeaderAddr()` exist and work. +`hubStatus` never calls either: it reports store stats and +listener counts and returns. `StatusResponse` has no field to +put leadership in even if it did, so an operator running the +hub in HA mode has no supported way to ask which node leads. + +### The three lines that answer from the wrong data — `internal/cli/hub/core/status/status.go` + +`ctx hub status` prints a role, a leader and a peer count today. +None of the three is what its label claims: + +- **`Role:`** is `Active` when `ConnectedClients > 0` and + `Follower` otherwise. It reports whether anyone is subscribed, + not what this node's Raft role is. A leader with no listeners + reads `Follower`. +- **`Leader:`** is `cfg.HubAddr` — the address the CLI just + dialed. It says "the hub you asked is the leader" for every + hub, including a follower and including a standalone node with + no Raft at all. +- **`Peers:`** is `len(resp.EntriesByProject)`, the number of + distinct origin projects in the store. A single-node hub + holding entries from three projects reports `Peers: 3`. + +This is worse than silence: the failure-modes runbook tells +operators to run `ctx hub status` on each peer when they see +"no leader" errors, and the answer they get back is +manufactured. + +### `LeaderAddr` returns an ID, not an address — `internal/hub/cluster.go` + +```go +func (c *Cluster) LeaderAddr() string { + _, id := c.raftNode.LeaderWithID() + return string(id) +} +``` + +`LeaderWithID` returns `(ServerAddress, ServerID)`. The method +named `LeaderAddr`, documented as "leader address", discards the +address and returns the ID. Surfacing that value on the wire as +`leader_addr` would make the mismatch a published contract. + +### The cluster that could never start — `internal/cli/hub/core/server` + +Two defects made the leadership fields unreachable from the CLI +no matter how correctly they were wired. + +`RunDaemon` built the re-exec argv from `--port` and +`--data-dir` only. `--peers` was accepted by the flag parser, +never forwarded, and never seen by the process that actually +ran: every `ctx hub start --daemon --peers ...` started a +standalone hub while reporting success, and the HA recipe's +three `--daemon` commands produced three unrelated hubs. + +Running the same command in the foreground, where the flag did +arrive, failed anyway. `Run` derived the Raft address as +`fmt.Sprintf(":%d", port+1)` and passed it to +`raft.NewTCPTransport` as both bind and advertise address. +Raft refuses to advertise an unspecified address, so every +cluster start died on: + +``` +Error: local bind address is not advertisable +``` + +Cluster mode has never started on any machine. The two defects +hid each other: the daemon path silently dropped the flag that +would have surfaced the crash. + +### The three commands that printed and returned — `internal/cli/hub` + +`ctx hub peer add`, `ctx hub peer remove` and `ctx hub +stepdown` each called a `write` helper and returned nil. +`Cluster.Stepdown()` had no caller anywhere in the tree, and +raft's `AddVoter` / `RemoveServer` were never called at all. +The HA recipe documented all three as working cluster +operations, so an operator handing off leadership before +maintenance got "Leadership transferred" from a process that +had asked nobody anything. + +### The bootstrap error nothing checks — `internal/hub/cluster.go` + +`NewCluster` calls `r.BootstrapCluster(config)` in both branches +and discards the future. A node whose bootstrap fails returns a +healthy-looking `*Cluster` that never elects anyone — the exact +state the new Status fields are meant to make visible, arriving +with no error and no log line. + +## Solution + +### Make the cluster answerable + +1. `internal/hub/cluster.go` — `LeaderAddr` returns the Raft + `ServerAddress` of the leader, matching its name and its + docstring. Empty while no leader is known (election in + progress, or quorum lost). +2. `internal/hub/cluster.go` — `Peers()` reports the number of + *other* servers in the committed Raft configuration + (`GetConfiguration`), returning the future's error rather + than swallowing it. The count is accumulated in a `uint32` + instead of converting `len()`, so the wire type needs no + `gosec` narrowing suppression. +3. `internal/hub/cluster.go` — the two bootstrap branches + collapse into one server list plus one `BootstrapCluster` + call whose error is checked. `raft.ErrCantBootstrap` is + tolerated: it is what a restart against existing Raft state + returns, and that node is already bootstrapped. + +### Give the cluster an address it can advertise + +4. `internal/config/flag`, `internal/config/embed/flag`, + `internal/assets/commands/flags.yaml` — `--raft-bind`, the + address this node binds its Raft transport to and + advertises to its peers. `--peers` becomes the list of the + other nodes' `--raft-bind` addresses, so every node + bootstraps the same `{ID, Address}` set; each node + registers under its own Raft address as both. +5. `internal/cli/hub/core/server/setup.go` — + `validateRaftBind` rejects an empty, host-less or wildcard + address before Raft does, because Raft's own error names + neither the flag nor the value. `errHub.RaftBindRequired` + and `errHub.RaftBindUnroutable` carry the text, keyed in + `errors.yaml` like every other hub error. +6. `internal/cli/hub/core/server/run.go` — the Raft node + starts when `--raft-bind` is given, with or without peers: + a lone node bootstraps a one-server cluster and elects + itself, which is the cheapest way for an operator to see + the new Status fields. Asking for `--peers` without + `--raft-bind` is an error rather than a hub that quietly + is not in a cluster. +7. `internal/cli/hub/core/server` — `daemonArgs` (extracted + from `RunDaemon` so it can be tested without forking) + forwards both cluster flags. + +### Make the cluster commands do what they print + +8. `internal/hub` — two admin-token-gated RPCs, `Peer` and + `Stepdown`, gated the way `Register` and `Revoke` are: + reshaping a cluster is an operator action, not a client + one. `Cluster.AddPeer` / `RemovePeer` wrap raft's + `AddVoter` / `RemoveServer` keyed on the node's Raft + address, which is also its ID. +9. `internal/hub/err_check.go` — `clusterOpErr` maps + `raft.ErrNotLeader` to `FailedPrecondition` with a message + naming `ctx hub status` as the way to find the leader. A + follower's refusal is something the operator can act on; + an opaque `Internal` is not. +10. `internal/cli/hub/core/{peer,stepdown}` — both dial the + hub from the saved connection config and call the RPC. + `internal/cli/hub/core/admin.Token` resolves `--token` + then `CTX_HUB_ADMIN_TOKEN` for all three admin commands, + including `revoke`, whose inline copy it replaces. +11. `--join` (`internal/cli/hub`, `internal/hub.ClusterConfig`) + — a node that bootstraps its own configuration is a second + cluster of one, not a member of the first, so a node being + added has to start with no configuration and wait. `--join` + with `--peers` is an error: a node either bootstraps or + joins. + +### Carry it on the wire + +12. `internal/hub/types.go` — `StatusResponse` gains + `ClusterEnabled`, `IsLeader`, `LeaderAddr` and + `ClusterPeers`. All four are additive and JSON-omitempty + where a zero value is meaningless, so an older client + decoding a newer response is unaffected. + `ClusterEnabled` is the disambiguator: without it a + standalone hub is indistinguishable from a clustered node + that has lost its leader — both report `IsLeader: false`, + `LeaderAddr: ""`. +13. `internal/hub/handler.go` — `hubStatus` fills those fields + from `s.cluster` when one is attached, and warns to stderr + (`cfgWarn.HubClusterPeers`) if the configuration read fails + rather than failing the RPC: a Status call is a diagnostic, + and losing the peer count is not a reason to deny the + operator the rest of it. +14. `internal/config/warn/warn.go` — `HubClusterPeers`, the + stderr format for that read. + +### Render what the hub actually said + +15. `internal/config/hub/hub.go` — `RoleLeader` and + `RoleStandalone` join `RoleFollower`. `RoleActive` is + deleted: it labelled the listener-count heuristic and has no + meaning once the role comes from Raft. +16. `internal/write/hub` — `ClusterStatus` takes a + `ClusterStatusInfo` struct rather than growing a seventh + positional parameter, and renders two shapes: + + ``` + Role: Standalone Role: Leader + Entries: 1248 Leader: 10.0.0.5:9901 + Entries: 1248 Peers: 2 + ``` + + The leader line reads `Leader: unknown (election in + progress)` when the cluster is up but `LeaderAddr` is empty. + `Entries:` gets its own text key for the standalone shape, + where the combined `Entries: %d Peers: %d` line would be + printing a peer count that does not exist. The + `Dropped listeners:` line keeps its non-zero condition. +17. `internal/cli/hub/core/status/status.go` — the role, + the leader and the peer count all come from the response. + Standalone hubs print no leader and no peer count instead + of inventing both. + +### Correct the docs the change falsifies + +18. `docs/recipes/hub-cluster.md` — the "expected output" block + is replaced with what the command prints. It currently + shows a per-peer table with sync state and uptime that no + version of this code has ever produced. +19. `docs/cli/hub.md`, `internal/assets/commands/commands.yaml` + — `ctx hub status` is described as role, leader, entries, + peers and dropped listeners; "sync state" and "uptime" are + dropped because neither is reported. +20. `docs/operations/hub.md` — the monitoring section drops + `ctx hub status --exit-code` (no such flag exists; the + command exits non-zero only on RPC failure) and the + per-peer replication-lag claim (the response carries no + per-peer sequence). Role flaps stay: with a truthful + `Role:` line they are now actually observable. +21. `internal/hub/doc.go` — the Status paragraph names the + leadership fields. +22. `docs/cli/hub.md`, `docs/recipes/hub-cluster.md` — + `--raft-bind` and `--join` in the start reference, the + cluster recipe's start commands, and its topology diagram, + which showed one port per node where there are two. The + membership and maintenance sections document what the + commands now do: admin-gated, leader-only, addressed by + Raft address, and — for an addition — preceded by starting + the new node with `--join`. + +## Tests + +- `TestCluster_ThreeNodesElectOneLeader` — three real Raft + nodes on loopback ports, each bootstrapped with the full + server list: exactly one leads, all three name the same + leader address, and each counts the other two as peers. This + is the regression test for the address arithmetic: on `main` + the nodes cannot start at all. +- `TestValidateRaftBind` — the wildcard, bare-port and + host-less forms an operator reaches for, each rejected with + a message naming the flag. +- `TestDaemonArgs_ForwardsClusterFlags` / + `TestDaemonArgs_OmitsClusterFlags` — the argv the daemon is + re-executed with. `--peers` used to be missing from it, so a + daemonized "cluster" node ran standalone while reporting + success; the second test keeps a standalone daemon's argv + unchanged. +- `TestHubStatus_StandaloneReportsClusterDisabled` — a Server + with no `SetCluster` reports `ClusterEnabled == false` and + zero values for the rest. Pins the disambiguator. +- `TestHubStatus_ClusterReportsLeadershipState` — a real + single-node Raft node elects itself; Status then reports + `ClusterEnabled`, `IsLeader`, a non-empty `LeaderAddr` and + `ClusterPeers == 0` (one server, no peers). Verified by + mutation: returning early from the `s.cluster != nil` block + fails it. +- `TestCluster_PeersExcludesSelf` — the count excludes self on + a single-node configuration. +- `TestCluster_LeaderAddrIsTheAddress` — the node registers + under an ID that is not an address, so the old + address-discarding implementation fails here. +- `TestRenderInfo_*` — the CLI mapping: Standalone with no + leader and no peers, Leader when the hub says so, and + Follower on a clustered node with no listeners (the case the + old listener-count heuristic labelled backwards). +- `TestClusterStatus_Standalone` / `TestClusterStatus_Cluster` + / `TestClusterStatus_LeaderUnknown` — the three rendered + shapes. `desc.Text` returns `""` for an unknown key, so a + renamed text key blanks a line silently; asserting on the + rendered strings catches it. `TestClusterStatus_*Dropped*` + keep their existing contract under the new struct argument. + +- `TestCluster_PeerAddJoinsNode` / + `TestCluster_PeerRemoveShrinksCluster` — a node started in + join mode holds no configuration until the leader adds it, + then follows that leader and counts as a peer; removing it + shrinks the configuration again. Real Raft nodes, not mocks. +- `TestCluster_StepdownHandsOffLeadership` — after a transfer + the old leader is not leading and the other node is. +- `TestPeer_RejectsBadAdminToken` / + `TestStepdown_RejectsBadAdminToken` — the admin gate. +- `TestPeer_NoClusterIsPrecondition` / + `TestStepdown_NoClusterIsPrecondition` / + `TestPeer_FollowerIsPrecondition` — the three refusals that + used to be confirmations: no Raft node at all, and a + configuration change asked of a follower (the + `raft.ErrNotLeader` mapping). +- `TestPeer_ValidatesRequest` — unknown action and empty + address. +- `TestDaemonArgs_ForwardsJoin` — the boolean flag, which + carries no value and is the easiest one to drop. + +## Out of Scope + +- **A `Leadership` streaming RPC** and a `ctx hub leader` + shortcut, both named as non-goals by the issue. Status is + the minimum useful surface; a subscription and a shortcut + are sugar until a caller asks for them. +- **Raft term, commit index, log position.** Useful for + debugging quorum, a different question from "who leads now". +- **Deterministic bootstrap (H-12).** A node with `--peers` + still bootstraps the full server list, which works because + every node is given the same list and `ErrCantBootstrap` is + tolerated on restart. `--join` + `ctx hub peer add` is the + `AddVoter` half of H-12 and lands here because `peer add` + is meaningless without it; the single-`--bootstrap`-node + flow and the persisted bootstrapped flag stay with H-12. +- **Client-side failover to the new leader.** `ctx hub + stepdown` now moves leadership, and a client whose stream + was on the old leader still has to be re-run: `Listen` is + called once with `sinceSequence` hardcoded to `0`, the same + latent full-refetch `fix-hub-silent-error-suppression.md` + parked. Reconnect stays manual and the failure-modes doc + says so. +- **Authenticated Raft transport (H-10/H-11).** `--raft-bind` + now makes a real cluster possible on a LAN, and the Raft + transport is still unauthenticated and unencrypted. The + security docs already say so; nothing here changes it.