From c67c06d8e6979992e1ea20fae6e3c105b2797421 Mon Sep 17 00:00:00 2001 From: Paul Payne Date: Tue, 7 Jul 2026 05:03:07 -0700 Subject: [PATCH] feat(mesh): fleet role + shared config/presence KV (Phase 2) - add static 'role' (authority|follower) to NodeConfig/CastleConfig, wired from castle.yaml through the registry to the mesh wire; civil pinned authority - castle-presence: TTL KV bucket each node renews (churn signal); delete-on-stop - castle-config: authority-gated get/put + change-watch SSE (follower reconcile hook hangs off it) - bound the NATS drain on stop() so shutdown can't hang - tests: role config/registry round-trip, wire round-trip, role write-gating Single-node verified; follower reconcile pending the second node. --- castle-api/src/castle_api/mesh_wire.py | 3 + castle-api/src/castle_api/nats_client.py | 91 +++++++++++++++++++++--- castle-api/tests/test_mesh_wire.py | 12 ++++ castle-api/tests/test_nats_client.py | 30 ++++++++ core/src/castle_core/config.py | 4 ++ core/src/castle_core/deploy.py | 1 + core/src/castle_core/registry.py | 8 +++ core/tests/test_fleet_role.py | 48 +++++++++++++ docs/fleet-mesh-plan.md | 24 ++++++- 9 files changed, 211 insertions(+), 10 deletions(-) create mode 100644 castle-api/tests/test_nats_client.py create mode 100644 core/tests/test_fleet_role.py diff --git a/castle-api/src/castle_api/mesh_wire.py b/castle-api/src/castle_api/mesh_wire.py index 9f66509..1a79d16 100644 --- a/castle-api/src/castle_api/mesh_wire.py +++ b/castle-api/src/castle_api/mesh_wire.py @@ -26,6 +26,8 @@ def registry_to_json(registry: NodeRegistry) -> str: # acme domain — lets peers build launch URLs (.) # for this node's exposed apps. Omitted when the node has no domain. "gateway_domain": registry.node.gateway_domain, + # fleet role — so peers know which node is the config/secret authority. + "role": registry.node.role, }, "deployed": {}, } @@ -73,6 +75,7 @@ def json_to_registry(payload: str) -> NodeRegistry: castle_root=node_data.get("castle_root"), gateway_port=node_data.get("gateway_port", 9000), gateway_domain=node_data.get("gateway_domain"), + role=node_data.get("role", "follower"), ) deployed: dict[str, Deployment] = {} for key, comp_data in data.get("deployed", {}).items(): diff --git a/castle-api/src/castle_api/nats_client.py b/castle-api/src/castle_api/nats_client.py index 0194058..3a71b2b 100644 --- a/castle-api/src/castle_api/nats_client.py +++ b/castle-api/src/castle_api/nats_client.py @@ -34,8 +34,11 @@ from castle_api.stream import broadcast logger = logging.getLogger(__name__) REGISTRY_BUCKET = "castle-registry" +PRESENCE_BUCKET = "castle-presence" +CONFIG_BUCKET = "castle-config" HEARTBEAT_SEC = 30.0 PRUNE_SEC = 30.0 +PRESENCE_TTL = 90.0 # a node whose presence key expires within this is gone class CastleNATSClient: @@ -52,6 +55,8 @@ class CastleNATSClient: self._servers = servers self._nc: nats.NATS | None = None self._kv = None + self._presence_kv = None + self._config_kv = None self._tasks: list[asyncio.Task] = [] self._last_json: dict[str, str] = {} self._online: set[str] = set() @@ -64,6 +69,11 @@ class CastleNATSClient: def servers(self) -> str | list[str]: return self._servers + @property + def role(self) -> str: + """This node's fleet role — 'authority' or 'follower'.""" + return self._local_registry.node.role + async def start(self) -> None: """Connect, publish our registry, seed state, and start watchers.""" self._nc = await nats.connect( @@ -72,22 +82,39 @@ class CastleNATSClient: max_reconnect_attempts=-1, # reconnect forever — nodes come and go ) js = self._nc.jetstream() - try: - self._kv = await js.key_value(REGISTRY_BUCKET) - except Exception: - self._kv = await js.create_key_value( - config=KeyValueConfig(bucket=REGISTRY_BUCKET, history=1) - ) + self._kv = await self._ensure_bucket(js, REGISTRY_BUCKET, history=1) + # Presence: a short-TTL key each node renews; its expiry = the node is gone. + self._presence_kv = await self._ensure_bucket( + js, PRESENCE_BUCKET, history=1, ttl=PRESENCE_TTL + ) + # Shared config: authority-written, followers watch + reconcile. + self._config_kv = await self._ensure_bucket(js, CONFIG_BUCKET, history=5) await self.publish_registry(self._local_registry) + await self._presence_kv.put(self._local_hostname, b"online") await self._seed_existing() self._tasks = [ asyncio.create_task(self._watch_loop()), + asyncio.create_task(self._config_watch_loop()), asyncio.create_task(self._heartbeat_loop()), asyncio.create_task(self._prune_loop()), ] - logger.info("NATS mesh client started (servers=%s)", self._servers) + logger.info( + "NATS mesh client started (servers=%s, role=%s)", + self._servers, + self.role, + ) + + @staticmethod + async def _ensure_bucket(js, bucket: str, *, history: int = 1, ttl=None): + """Bind an existing KV bucket or create it.""" + try: + return await js.key_value(bucket) + except Exception: + return await js.create_key_value( + config=KeyValueConfig(bucket=bucket, history=history, ttl=ttl) + ) async def stop(self) -> None: """Delete our key (immediate offline to peers) and disconnect.""" @@ -100,9 +127,15 @@ class CastleNATSClient: if self._kv is not None: with contextlib.suppress(Exception): await self._kv.delete(self._local_hostname) - if self._nc is not None: + if self._presence_kv is not None: with contextlib.suppress(Exception): - await self._nc.drain() + await self._presence_kv.delete(self._local_hostname) + if self._nc is not None: + # Bound the drain so a wedged connection can't hang systemd shutdown. + with contextlib.suppress(Exception): + await asyncio.wait_for(self._nc.drain(), timeout=5.0) + with contextlib.suppress(Exception): + await self._nc.close() self._nc = None logger.info("NATS mesh client stopped") @@ -172,6 +205,46 @@ class CastleNATSClient: await asyncio.sleep(HEARTBEAT_SEC) with contextlib.suppress(Exception): await self.publish_registry(self._local_registry) + if self._presence_kv is not None: + with contextlib.suppress(Exception): + await self._presence_kv.put(self._local_hostname, b"online") + + # --- Shared config (authority writes, followers reconcile) --- + + async def get_shared_config(self, key: str) -> str | None: + """Read a shared-config value from the mesh (None if unset).""" + if self._config_kv is None: + return None + try: + entry = await self._config_kv.get(key) + return entry.value.decode() if entry.value else None + except Exception: + return None + + async def put_shared_config(self, key: str, value: str) -> None: + """Write a shared-config value. Only the authority may write.""" + if self.role != "authority": + raise PermissionError("only the authority node may write shared config") + if self._config_kv is None: + raise RuntimeError("config bucket not available") + await self._config_kv.put(key, value.encode()) + + async def _config_watch_loop(self) -> None: + """Watch shared config; announce changes so followers can reconcile. + + The reconcile action (trigger `castle apply` on a follower) hangs off this + SSE event; it's inert on the authority and on a single node. + """ + if self._config_kv is None: + return + watcher = await self._config_kv.watchall() + async for entry in watcher: + if entry is None: + continue + op = "delete" if entry.operation in ("DEL", "PURGE") else "put" + await broadcast( + "mesh", {"event": "config_changed", "key": entry.key, "op": op} + ) async def _prune_loop(self) -> None: """Mark crashed peers (no refresh within the stale TTL) offline.""" diff --git a/castle-api/tests/test_mesh_wire.py b/castle-api/tests/test_mesh_wire.py index 1467b69..9693172 100644 --- a/castle-api/tests/test_mesh_wire.py +++ b/castle-api/tests/test_mesh_wire.py @@ -95,6 +95,18 @@ class TestRegistrySerialization: assert bare.schedule is None assert bare.managed is False + def test_node_role_preserved(self) -> None: + reg = NodeRegistry( + node=NodeConfig(hostname="civil", role="authority"), deployed={} + ) + restored = json_to_registry(registry_to_json(reg)) + assert restored.node.role == "authority" + + def test_node_role_defaults_follower(self) -> None: + reg = NodeRegistry(node=NodeConfig(hostname="n"), deployed={}) + restored = json_to_registry(registry_to_json(reg)) + assert restored.node.role == "follower" + def test_no_secrets_in_payload(self) -> None: """env vars, run_cmd, and castle_root must never appear on the wire.""" original = _make_registry() diff --git a/castle-api/tests/test_nats_client.py b/castle-api/tests/test_nats_client.py new file mode 100644 index 0000000..ec47124 --- /dev/null +++ b/castle-api/tests/test_nats_client.py @@ -0,0 +1,30 @@ +"""CastleNATSClient role gating — hermetic (no live NATS server needed). + +The write-gate is checked before touching the KV bucket, so a follower is denied +without any connection. +""" + +from __future__ import annotations + +import asyncio + +import pytest +from castle_core.registry import NodeConfig, NodeRegistry + +from castle_api.nats_client import CastleNATSClient + + +def _client(role: str) -> CastleNATSClient: + reg = NodeRegistry(node=NodeConfig(hostname="n", role=role), deployed={}) + return CastleNATSClient("n", reg, servers="nats://localhost:4222") + + +def test_role_property() -> None: + assert _client("authority").role == "authority" + assert _client("follower").role == "follower" + + +def test_follower_cannot_write_shared_config() -> None: + client = _client("follower") + with pytest.raises(PermissionError): + asyncio.run(client.put_shared_config("fleet/key", "value")) diff --git a/core/src/castle_core/config.py b/core/src/castle_core/config.py index 5b7bc71..7e47969 100644 --- a/core/src/castle_core/config.py +++ b/core/src/castle_core/config.py @@ -184,6 +184,9 @@ class CastleConfig: # built-in defaults so tests/callers that don't care stay valid. data_dir: Path = field(default_factory=lambda: _DEFAULT_DATA_DIR) repos_dir: Path = field(default_factory=lambda: _DEFAULT_REPOS_DIR) + # Fleet role: "authority" (may write shared config/secrets to the mesh) or + # "follower" (reconciles from it). Static — pinned here, no election. + role: str = "follower" # Construction convenience only (not stored): a flat name→spec dict is routed # into the per-kind stores by kind_for. Lets callers/tests hand us a flat map # without pre-splitting it; there is still no flat `deployments` attribute. @@ -465,6 +468,7 @@ def load_config(root: Path | None = None) -> CastleConfig: agents=agents, data_dir=data_dir, repos_dir=repos_dir, + role=data.get("role", "follower"), **stores, ) return config diff --git a/core/src/castle_core/deploy.py b/core/src/castle_core/deploy.py index 5db67b0..ecfa090 100644 --- a/core/src/castle_core/deploy.py +++ b/core/src/castle_core/deploy.py @@ -189,6 +189,7 @@ def _node_config(config: CastleConfig) -> NodeConfig: public_domain=config.gateway.public_domain, tunnel_id=config.gateway.tunnel_id, cert_hook=config.gateway.cert_hook, + role=config.role, ) diff --git a/core/src/castle_core/registry.py b/core/src/castle_core/registry.py index 5cbb08c..668a799 100644 --- a/core/src/castle_core/registry.py +++ b/core/src/castle_core/registry.py @@ -32,6 +32,11 @@ class NodeConfig: tunnel_id: str | None = None # Emit the cert_obtained → `castle tls reconcile` hook (needs events-exec plugin). cert_hook: bool = False + # Fleet role: "authority" may write shared config/secrets to the mesh; + # "follower" reconciles from it. Static (no election) — the authority is + # pinned in castle.yaml. When the authority is down, shared state is + # read-only and every node keeps serving its own deployments from cache. + role: str = "follower" def __post_init__(self) -> None: if not self.hostname: @@ -152,6 +157,7 @@ def load_registry(path: Path | None = None) -> NodeRegistry: public_domain=node_data.get("public_domain"), tunnel_id=node_data.get("tunnel_id"), cert_hook=node_data.get("cert_hook", False), + role=node_data.get("role", "follower"), ) deployed: dict[str, Deployment] = {} @@ -241,6 +247,8 @@ def save_registry(registry: NodeRegistry, path: Path | None = None) -> None: data["node"]["tunnel_id"] = registry.node.tunnel_id if registry.node.cert_hook: data["node"]["cert_hook"] = registry.node.cert_hook + if registry.node.role and registry.node.role != "follower": + data["node"]["role"] = registry.node.role for key, comp in registry.deployed.items(): entry: dict = { diff --git a/core/tests/test_fleet_role.py b/core/tests/test_fleet_role.py new file mode 100644 index 0000000..2d440e5 --- /dev/null +++ b/core/tests/test_fleet_role.py @@ -0,0 +1,48 @@ +"""Fleet role (authority/follower) — config loading + registry round-trip.""" + +from __future__ import annotations + +from pathlib import Path + +import yaml +from castle_core.config import load_config +from castle_core.registry import ( + NodeConfig, + NodeRegistry, + load_registry, + save_registry, +) + + +def _write_min_config(root: Path, role: str | None) -> None: + data: dict = {"gateway": {"port": 18000}} + if role is not None: + data["role"] = role + (root / "castle.yaml").write_text(yaml.safe_dump(data)) + + +def test_role_defaults_to_follower(tmp_path: Path) -> None: + _write_min_config(tmp_path, None) + assert load_config(tmp_path).role == "follower" + + +def test_role_loaded_from_yaml(tmp_path: Path) -> None: + _write_min_config(tmp_path, "authority") + assert load_config(tmp_path).role == "authority" + + +def test_registry_role_round_trip(tmp_path: Path) -> None: + reg = NodeRegistry( + node=NodeConfig(hostname="civil", role="authority"), deployed={} + ) + path = tmp_path / "registry.yaml" + save_registry(reg, path) + assert load_registry(path).node.role == "authority" + + +def test_registry_role_defaults_follower(tmp_path: Path) -> None: + """A node with no explicit role round-trips as follower.""" + reg = NodeRegistry(node=NodeConfig(hostname="n"), deployed={}) + path = tmp_path / "registry.yaml" + save_registry(reg, path) + assert load_registry(path).node.role == "follower" diff --git a/docs/fleet-mesh-plan.md b/docs/fleet-mesh-plan.md index 8af6288..ccd13f8 100644 --- a/docs/fleet-mesh-plan.md +++ b/docs/fleet-mesh-plan.md @@ -1,7 +1,7 @@ # Fleet Mesh Plan — OpenBao + NATS **Status:** in progress on branch `feat/fleet-mesh-nats-openbao`. -Phases 0–1 complete + single-node verified (live cutover done). Phases 2–4 pending. +Phases 0–2 complete + single-node verified (live). Phases 3–4 pending (need 2nd node). ## Context @@ -196,6 +196,28 @@ so revert = `git checkout main -- castle-api` then re-apply. The old Mosquitto `mqtt` service is left running (dormant) as a safety net; retire it once the second node is proven on NATS. +### Phase 2 — DONE + single-node verified + tested (2026-07-07) + +- **`role` field** on `NodeConfig` + `CastleConfig`, wired end-to-end: + castle.yaml top-level `role:` → config → `_node_config` → registry.yaml → + mesh wire. **`civil` pinned `role: authority`**; default `follower`. +- **Presence** (`castle-presence`) — a TTL KV bucket each node renews on the + heartbeat; expiry = the node is gone. Delete-on-stop for immediate departure. +- **Shared config** (`castle-config`) — `get_shared_config` / `put_shared_config` + (authority-gated: followers raise `PermissionError`), plus a watch loop that + broadcasts a `config_changed` SSE (the follower `castle apply` reconcile hook + hangs off this — inert on one node). +- Hardened `stop()` to bound the NATS drain (can't hang systemd shutdown). +- Verified live: all three buckets present, `civil=online` presence, + `"role": "authority"` on the wire, authority write + follower-deny exercised + against the running `castle-nats`. +- **Tests added:** `core/tests/test_fleet_role.py` (role config load + registry + round-trip), `castle-api/tests/test_nats_client.py` (role gating, hermetic), + role round-trip in `test_mesh_wire.py`. Suites: **200 core + 93 api** pass. + +**Pending second node:** the follower-side reconcile (watch `castle-config` → +`castle apply`) is wired as an SSE hook but only meaningful with a peer. + ## Decisions (resolved) 1. **Discovery — both.** Keep mDNS for zero-config LAN peer discovery *and*