From 5863aeed1391f37ce3feeba0bbef1d36e21a6eea Mon Sep 17 00:00:00 2001 From: Ben Duncan <5178243+benduncan@users.noreply.github.com> Date: Sat, 14 Feb 2026 13:23:24 +1000 Subject: [PATCH] Phase 0/1 multi-server-restart enhancements --- .claude/CLAUDE.md | 13 + cmd/hive/cmd/admin.go | 11 +- cmd/hive/cmd/templates/hive.toml | 3 + .../improvements/multi-server-restart.md | 407 +++++++----------- hive/admin/admin.go | 3 + hive/config/config.go | 49 ++- hive/config/config_test.go | 38 ++ hive/daemon/daemon.go | 159 ++++++- hive/daemon/jetstream.go | 108 ++++- hive/formation/formation.go | 13 +- hive/formation/helpers.go | 39 +- hive/formation/helpers_test.go | 87 ++++ scripts/start-dev.sh | 257 ++++++----- scripts/stop-dev.sh | 36 +- 14 files changed, 795 insertions(+), 428 deletions(-) create mode 100644 hive/formation/helpers_test.go diff --git a/.claude/CLAUDE.md b/.claude/CLAUDE.md index 4733e744f..3bc8ceb92 100644 --- a/.claude/CLAUDE.md +++ b/.claude/CLAUDE.md @@ -139,6 +139,19 @@ When starting work on a feature or investigating an issue, review the last 10-20 - All new features must have comprehensive unit tests - For returning AWS errors, use `awserrors` package instead of manually typed strings +**Preflight Policy:** +- Run `make preflight` before pushing any major changes — it runs the same checks as GitHub Actions (gofmt, go vet, gosec, staticcheck, govulncheck, and all unit tests) +- All developers must install the pre-push hook to enforce this automatically: + +```bash +cat > .git/hooks/pre-push << 'EOF' +#!/bin/sh +make preflight +make test +EOF +chmod +x .git/hooks/pre-push +``` + **Testing Policy:** - **All unit tests MUST pass** before committing changes - **No exceptions** - failing tests block commits diff --git a/cmd/hive/cmd/admin.go b/cmd/hive/cmd/admin.go index 587ff0cd4..562d28a55 100644 --- a/cmd/hive/cmd/admin.go +++ b/cmd/hive/cmd/admin.go @@ -162,6 +162,7 @@ func init() { adminInitCmd.Flags().String("predastore-nodes", "", "Comma-separated IPs for multi-node Predastore cluster (e.g., 10.11.12.1,10.11.12.2,10.11.12.3). Requires >= 3 nodes.") adminInitCmd.Flags().String("formation-timeout", "10m", "Timeout for cluster formation (e.g., 5m, 30s)") adminInitCmd.Flags().String("cluster-name", "hive", "NATS cluster name") + adminInitCmd.Flags().StringSlice("services", nil, "Services this node runs (default: all). Valid: nats,predastore,viperblock,daemon,awsgw,ui") // Flags for admin join adminJoinCmd.Flags().String("region", "ap-southeast-2", "Region for this node") @@ -173,6 +174,7 @@ func init() { adminJoinCmd.Flags().String("bind", "0.0.0.0", "IP address to bind services to (e.g., 10.11.12.2 for multi-node on single host)") adminJoinCmd.Flags().String("cluster-bind", "", "IP address to bind NATS cluster services to (e.g., 10.11.12.1 for multi-node)") adminJoinCmd.Flags().String("cluster-routes", "", "NATS cluster hosts for routing specify multiple with comma (e.g., 10.11.12.1:4248,10.11.12.2:4248 for multi-node)") + adminJoinCmd.Flags().StringSlice("services", nil, "Services this node runs (default: all)") adminJoinCmd.MarkFlagRequired("node") adminJoinCmd.MarkFlagRequired("host") @@ -486,6 +488,7 @@ func runAdminInit(cmd *cobra.Command, args []string) { predastoreNodesStr, _ := cmd.Flags().GetString("predastore-nodes") formationTimeoutStr, _ := cmd.Flags().GetString("formation-timeout") clusterName, _ := cmd.Flags().GetString("cluster-name") + services, _ := cmd.Flags().GetStringSlice("services") // Validate IP address format if net.ParseIP(bindIP) == nil { @@ -565,7 +568,7 @@ func runAdminInit(cmd *cobra.Command, args []string) { if isMultiNode { runAdminInitMultiNode(cmd, accessKey, secretKey, accountID, natsToken, clusterName, configDir, hiveRoot, certPath, region, az, node, bindIP, clusterBind, - port, nodes, formationTimeoutStr) + port, nodes, formationTimeoutStr, services) return } @@ -649,6 +652,7 @@ func runAdminInit(cmd *cobra.Command, args []string) { ClusterName: clusterName, PredastoreNodeID: predastoreNodeID, + Services: services, } // Generate config files @@ -701,7 +705,7 @@ func runAdminInit(cmd *cobra.Command, args []string) { // then generates configs with complete cluster topology. func runAdminInitMultiNode(cmd *cobra.Command, accessKey, secretKey, accountID, natsToken, clusterName, configDir, hiveRoot, certPath, region, az, node, bindIP, clusterBind string, - port, expectedNodes int, formationTimeoutStr string) { + port, expectedNodes int, formationTimeoutStr string, services []string) { formationTimeout, err := time.ParseDuration(formationTimeoutStr) if err != nil { @@ -826,6 +830,7 @@ func runAdminInitMultiNode(cmd *cobra.Command, accessKey, secretKey, accountID, ClusterName: clusterName, PredastoreNodeID: predastoreNodeID, + Services: services, } // Generate config files @@ -887,6 +892,7 @@ func runAdminJoin(cmd *cobra.Command, args []string) { bindIP, _ := cmd.Flags().GetString("bind") configDir, _ := cmd.Flags().GetString("config-dir") clusterBind, _ := cmd.Flags().GetString("cluster-bind") + services, _ := cmd.Flags().GetStringSlice("services") // Validate required parameters if node == "" { @@ -1127,6 +1133,7 @@ func runAdminJoin(cmd *cobra.Command, args []string) { ClusterName: creds.ClusterName, PredastoreNodeID: predastoreNodeID, + Services: services, } // Generate config files diff --git a/cmd/hive/cmd/templates/hive.toml b/cmd/hive/cmd/templates/hive.toml index 8ab7a3631..2e918c141 100644 --- a/cmd/hive/cmd/templates/hive.toml +++ b/cmd/hive/cmd/templates/hive.toml @@ -7,6 +7,9 @@ node = "{{.Node}}" node = "{{.Node}}" region = "{{.Region}}" az = "{{.Az}}" +{{- if .Services}} +services = [{{range $i, $s := .Services}}{{if $i}}, {{end}}"{{$s}}"{{end}}] +{{- end}} account_id = "{{.AccountID}}" accesskey = "{{.AccessKey}}" diff --git a/docs/development/improvements/multi-server-restart.md b/docs/development/improvements/multi-server-restart.md index 09d1a65a7..676d083e5 100644 --- a/docs/development/improvements/multi-server-restart.md +++ b/docs/development/improvements/multi-server-restart.md @@ -1,6 +1,6 @@ # Multi-Server Cluster Restart -**Status: Planned** +**Status: In Progress** — Phase 0 and Phase 1 complete. Phase 2 and 3 planned. ## Summary @@ -89,263 +89,139 @@ Each daemon's `restoreInstances()` (`daemon.go:634-719`) runs independently on s ## Proposed Changes -### Phase 0: Node capabilities and service topology +### Phase 0: Node capabilities and service topology — Complete Foundation for everything else — the cluster must know what each node runs. -#### 0.1 Capabilities in config +#### 0.1 Capabilities in config — Done -Add a `services` list to the per-node `Config` struct. This declares which services the node runs locally: +Added `Services []string` field to the `Config` struct (`config/config.go:29`) and two helper methods: +- `HasService(name string) bool` — checks if a service is in the list (or returns true for all if list is empty, for backward compat) +- `GetServices() []string` — returns the services list, defaulting to `AllServices` when empty -```toml -# hive.toml — full infrastructure node (node1, node2, node3) -[nodes.node1] -node = "node1" -services = ["nats", "predastore", "viperblock", "daemon", "awsgw", "ui"] -# ... existing daemon, nats, predastore, awsgw sections ... +The `AllServices` package variable defines the canonical list: `["nats", "predastore", "viperblock", "daemon", "awsgw", "ui"]`. -# hive.toml — compute-only node (node4) -[nodes.node4] -node = "node4" -services = ["daemon", "viperblock"] -# nats/predastore sections still present but point to REMOTE hosts -[nodes.node4.nats] -host = "10.1.3.170:4222" # remote NATS cluster -[nodes.node4.predastore] -host = "10.1.3.170:8443" # remote Predastore -``` +The originally-planned role helpers (`IsComputeNode`, `IsStorageNode`, `IsInfraNode`) were deferred — `HasService()` is sufficient and more flexible. -The `services` field is a string slice of service names. Valid values: `nats`, `predastore`, `viperblock`, `daemon`, `awsgw`, `ui`. When omitted (backward compat), defaults to all services — existing configs work unchanged. +Updated `NodeHealthResponse` to include `Services []string` and `ServiceHealth map[string]string`. -**Config struct change** (`config/config.go`): +Unit tests added in `hive/config/config_test.go`: `TestHasService_ExplicitList`, `TestHasService_EmptyListDefaultsAll`, `TestHasService_UnknownService`, `TestGetServices_DefaultsToAll`, `TestGetServices_ExplicitList`. -```go -type Config struct { - Node string `mapstructure:"node"` - Host string `mapstructure:"host"` - Region string `mapstructure:"region"` - AZ string `mapstructure:"az"` - DataDir string `mapstructure:"data_dir"` - Services []string `mapstructure:"services"` // NEW: which services this node runs locally - // ... rest unchanged -} -``` +**Files modified:** +- `hive/config/config.go` — `Services` field, `AllServices`, `HasService()`, `GetServices()`, updated `NodeHealthResponse` +- `hive/config/config_test.go` — 5 new tests -Helper methods on Config: +#### 0.2 Formation carries capabilities — Done -```go -func (c Config) HasService(name string) bool // check if service is in the list -func (c Config) IsComputeNode() bool // has "daemon" -func (c Config) IsStorageNode() bool // has "predastore" or "viperblock" -func (c Config) IsInfraNode() bool // has "nats" -``` +Added `Services []string` to `NodeInfo` in `hive/formation/formation.go`. Since `JoinRequest` embeds `NodeInfo`, it automatically gets the field. -**Files:** -- `hive/config/config.go` — add `Services` field and helper methods -- `cmd/hive/cmd/templates/hive.toml` — add `services` to template -- `cmd/hive/cmd/admin.go` — add `--services` flag to `hive admin init` and `hive admin join` +Updated `BuildClusterRoutes()` to filter nodes by `"nats"` service — only nodes running NATS are included in cluster routes. Updated `BuildPredastoreNodes()` to filter by `"predastore"` service. Added `hasService()` helper with backward-compat behavior (empty services list = all services). -#### 0.2 Formation carries capabilities +Unit tests added in `hive/formation/helpers_test.go`: `TestHasService_EmptyListMeansAll`, `TestHasService_ExplicitList`, `TestBuildClusterRoutes_FiltersNonNATSNodes`, `TestBuildClusterRoutes_EmptyServicesIncludesAll`, `TestBuildPredastoreNodes_FiltersByPredastore`, `TestBuildPredastoreNodes_EmptyServicesIncludesAll`, `TestBuildClusterRoutes_MixedServicesAndEmpty`. -Extend `formation.NodeInfo` to include the services list, so the formation server knows what each node provides: +**Files modified:** +- `hive/formation/formation.go` — `Services` field on `NodeInfo` +- `hive/formation/helpers.go` — `hasService()` helper, filtered `BuildClusterRoutes()` and `BuildPredastoreNodes()` +- `hive/formation/helpers_test.go` — 7 new tests -```go -type NodeInfo struct { - Name string `json:"name"` - BindIP string `json:"bind_ip"` - ClusterIP string `json:"cluster_ip"` - Region string `json:"region"` - AZ string `json:"az"` - Port int `json:"port"` - Services []string `json:"services"` // NEW -} -``` +#### 0.3 Config template and CLI flags — Done -This allows `BuildClusterRoutes()` to only include nodes with `"nats"` in their services list for NATS cluster routes, and `BuildPredastoreNodes()` to only include nodes with `"predastore"`. +Added `Services []string` to `ConfigSettings` in `hive/admin/admin.go`. -**Files:** -- `hive/formation/formation.go` — add `Services` to `NodeInfo` and `JoinRequest` -- `hive/formation/helpers.go` — filter by capability in `BuildClusterRoutes()` and `BuildPredastoreNodes()` +Added `services` template rendering in `cmd/hive/cmd/templates/hive.toml` (after the `az` line). -#### 0.3 Capability-aware start/stop scripts +Added `--services` StringSlice flag to both `adminInitCmd` and `adminJoinCmd` in `cmd/hive/cmd/admin.go`. The flag is read in `runAdminInit`, `runAdminInitMultiNode`, and `runAdminJoin`, and passed through to `ConfigSettings`. -Refactor `start-dev.sh` and `stop-dev.sh` to read the `services` list from `hive.toml` and only start/stop configured services. +**Files modified:** +- `hive/admin/admin.go` — `Services` field on `ConfigSettings` +- `cmd/hive/cmd/templates/hive.toml` — `services` template line +- `cmd/hive/cmd/admin.go` — `--services` flag on init/join, wiring to `ConfigSettings` -**Service dependency graph** (start order): +#### 0.4 Capability-aware start/stop scripts — Done -``` -nats → (none, starts first) -predastore → nats -viperblock → nats -daemon → nats, viperblock -awsgw → nats, daemon -ui → awsgw -``` +Added `parse_services()` and `has_service()` bash functions to both `scripts/start-dev.sh` and `scripts/stop-dev.sh`. -For a compute-only node (`services = ["daemon", "viperblock"]`): -1. Skip NATS (not local) — but verify remote NATS is reachable -2. Skip Predastore (not local) — but verify remote Predastore is reachable -3. Start Viperblock (local) -4. Start Daemon (local) -5. Skip AWSGW (not local) -6. Skip UI (not local) +`parse_services()` reads the `services` line from `hive.toml` and defaults to all services if not set. All 6 service start/stop blocks are wrapped in `has_service` conditionals with skip messages for non-local services. -**Implementation in start-dev.sh:** +The stop script maps between process names and config service names (e.g., `hive` → `daemon`, `hive-ui` → `ui`). -```bash -# Read services from hive.toml -SERVICES=$(parse_services_from_config "$CONFIG_DIR/hive.toml") -# Defaults to "nats predastore viperblock daemon awsgw ui" if not set +**Files modified:** +- `scripts/start-dev.sh` — `parse_services()`, `has_service()`, conditional service startup +- `scripts/stop-dev.sh` — `parse_services()`, `has_service()` (with name mapping), conditional service stop -has_service() { echo "$SERVICES" | grep -qw "$1"; } +#### 0.5 Health endpoint and cluster state KV — Done -# Start local services in dependency order -if has_service "nats"; then - start_service "nats" "$NATS_CMD" -else - # Verify remote NATS is reachable - NATS_HOST=$(parse_nats_host "$CONFIG_DIR/hive.toml") - check_service "NATS (remote)" "${NATS_HOST%%:*}" "${NATS_HOST##*:}" -fi +**Health endpoint**: Updated the `/health` HTTP handler in `daemon.go` to include `Services` (from config) and `ServiceHealth` (per-service status map). For local services, status is `"ok"`. For remote NATS dependency (when nats is not a local service), checks `natsConn.IsConnected()` and reports `"remote_ok"` or `"remote_unreachable"`. -if has_service "predastore"; then - start_service "predastore" "$PREDASTORE_CMD" -else - # Verify remote Predastore is reachable - check_service "Predastore (remote)" "$PREDASTORE_HOST" "$PREDASTORE_PORT" -fi +**Cluster state KV bucket**: Added `hive-cluster-state` bucket alongside the existing `hive-instance-state` bucket. New `clusterKV` field on `JetStreamManager`. `InitClusterStateBucket()` creates the bucket with 1-hour TTL and history of 1. Called from `initJetStream()` after `InitKVBucket()`. -# ... etc for each service -``` +**Service manifest**: `WriteServiceManifest()` writes `node.{nodeID}.services` to the cluster state KV on daemon startup, containing the node's services, NATS host, predastore host, and timestamp. -**stop-dev.sh** — reverse logic: only stop services that were started locally. Currently it calls `./bin/hive service stop` for all services, which fails silently for services that aren't running — but it's noisy and slow. +`UpdateReplicas()` updated to also update the cluster-state bucket replicas alongside the instance-state bucket. -**Files:** -- `scripts/start-dev.sh` — add config parsing, conditional service startup, remote dependency checks -- `scripts/stop-dev.sh` — add config parsing, only stop local services +**Files modified:** +- `hive/daemon/daemon.go` — health endpoint with services/serviceHealth, `InitClusterStateBucket()` call in `initJetStream()`, `WriteServiceManifest()` call in `Start()` +- `hive/daemon/jetstream.go` — `ClusterStateBucket` const, `clusterKV` field, `InitClusterStateBucket()`, `WriteServiceManifest()`, updated `UpdateReplicas()` -#### 0.4 Health endpoint includes capabilities - -Extend `NodeHealthResponse` to advertise what the node runs: - -```go -type NodeHealthResponse struct { - Node string `json:"node"` - Status string `json:"status"` - ConfigHash string `json:"config_hash"` - Epoch uint64 `json:"epoch"` - Uptime int64 `json:"uptime"` - Services []string `json:"services"` // NEW: what this node runs - ServiceHealth map[string]string `json:"service_health"` // NEW: per-service status -} -``` - -The `ServiceHealth` map only includes services this node runs: - -```json -{ - "node": "node4", - "status": "running", - "services": ["daemon", "viperblock"], - "service_health": { - "daemon": "ok", - "viperblock": "ok", - "nats": "remote_ok", - "predastore": "remote_ok" - } -} -``` - -For local services, the daemon checks the process is running. For remote dependencies (NATS, Predastore), it checks connectivity. This gives operators and the recovery coordinator a complete picture of node health. - -**Files:** -- `hive/config/config.go` — update `NodeHealthResponse` struct -- `hive/daemon/daemon.go` — update health handler to include capabilities and per-service health -- `hive/daemon/daemon_handlers.go` — update NATS health handler similarly - -#### 0.5 Cluster service map in JetStream KV - -To support runtime service discovery (not just config-time), each daemon writes its node's service manifest to JetStream KV on startup: - -``` -Bucket: hive-cluster-state (new, separate from hive-instance-state) -Key: node.{nodeID}.services -Value: {"node": "node4", "services": ["daemon", "viperblock"], "nats_host": "10.1.3.170:4222", "predastore_host": "10.1.3.170:8443"} -TTL: 1 hour (auto-cleanup of dead nodes) -History: 1 -``` - -This enables: -- Recovery coordinator to find which nodes can accept VMs (nodes with `"daemon"` service) -- Any node to discover where NATS/Predastore are running -- Operators to query cluster topology via `hive admin status` - -**Design decisions on KV buckets:** - -| Bucket | Purpose | Replicas | History | TTL | -|--------|---------|----------|---------|-----| -| `hive-instance-state` | VM state, stopped instances | cluster size | 1 | none | -| `hive-cluster-state` | Heartbeats, service maps, shutdown markers, recovery leases | cluster size | 1 | 1 hour | - -Separating the buckets keeps ephemeral data (heartbeats, leases) from durable data (instance state). The 1-hour TTL on `hive-cluster-state` auto-cleans dead node heartbeats and stale service maps without manual intervention. At 100 nodes this bucket holds ~100 keys of ~300 bytes each — trivially small. The TTL prevents unbounded growth from nodes that leave the cluster permanently. - -**Files:** -- `hive/daemon/jetstream.go` — new `InitClusterStateBucket()`, `WriteServiceManifest()`, `ListServiceManifests()` - -### Phase 1: Self-recovery improvements (single-node correctness) +### Phase 1: Self-recovery improvements (single-node correctness) — Complete These changes fix the immediate restart bugs without requiring cross-node coordination. -#### 1.1 Cluster readiness gate (capability-aware) +#### 1.1 Cluster readiness gate — Done -Before `restoreInstances()`, add a readiness check that waits for dependent services. The check varies based on the node's capabilities: +Added `waitForClusterReady()` method to `Daemon`, called in `Start()` just before `restoreInstances()`. Uses a 2-minute max wait with 2-second polling interval. -**Full infrastructure node** (`services = ["nats", "predastore", "viperblock", "daemon", ...]`): -``` -NATS connected (already done) - → JetStream KV initialized (already done) - → Local Viperblock healthy (health check with retry) - → Local Predastore healthy (health check with retry) - → Peer daemons discoverable (at least N-1 NATS routes connected) - → Begin recovery -``` +Two checks run in sequence: +1. **Viperblock** — `checkViperblockReady()` verifies NATS connection is live. Initially implemented as a NATS request to `ebs.{node}.health`, but viperblock has no health NATS topic — this caused a 2-minute hang during E2E tests. Fixed to simply check `d.natsConn.IsConnected()` since viperblock runs alongside the daemon and subscribes to NATS. +2. **Predastore** — `checkPredastoreReady()` does a TCP dial to the predastore host:port from config with a 3-second timeout. Skipped if no predastore host is configured. -**Compute-only node** (`services = ["daemon", "viperblock"]`): -``` -Remote NATS reachable (TCP connect with retry) - → JetStream KV initialized (already done — connects to remote NATS) - → Local Viperblock healthy (health check with retry) - → Remote Predastore reachable (TCP connect with retry) - → Begin recovery -``` +If all checks pass, recovery proceeds immediately. If the 2-minute timeout expires, recovery proceeds anyway with a warning log. -The `waitForClusterReady()` method reads the node's `services` list and builds its dependency checks accordingly. Local services get health endpoint checks; remote services get TCP connectivity checks. +**Finding**: Viperblock does not expose a health endpoint via NATS. Any future viperblock health check should use TCP connectivity or an HTTP endpoint, not NATS request-reply. -**Files:** -- `hive/daemon/daemon.go` — new `waitForClusterReady()`, called from `Start()` between lines 523-547 +**Files modified:** +- `hive/daemon/daemon.go` — `waitForClusterReady()`, `checkViperblockReady()`, `checkPredastoreReady()`, added `"net"` import -#### 1.2 Recovery throttling with semaphore +#### 1.2 Recovery throttling with semaphore — Done -Replace the unbounded loop in `restoreInstances()` with a worker pool. Limit concurrent VM launches to `maxConcurrentRecovery` (default: 2 per node). +Refactored `restoreInstances()` into two phases: -Recovery order priority: -1. Reconnect to still-running QEMU processes (instant, no I/O) -2. Finalize transitional states (stopping → stopped, shutting-down → terminated) -3. Relaunch crashed VMs (pending/running with dead QEMU) — these hit storage +**Phase 1 (sequential)**: Iterates all instances from KV state. For each instance: +- Reconnects to still-running QEMU processes (checks PID file, instant, no I/O) +- Finalizes transitional states (stopping → stopped, shutting-down → terminated) +- Collects instances that need relaunching into a `toLaunch` list -**Files:** -- `hive/daemon/daemon.go` — refactor `restoreInstances()` to use a semaphore for step 3 +**Phase 2 (throttled)**: Launches crashed VMs using a semaphore channel with `maxConcurrentRecovery = 2`. Each launch runs in a goroutine that acquires the semaphore before proceeding and releases on completion. A `sync.WaitGroup` ensures all launches complete before `restoreInstances()` returns. -#### 1.3 Clean shutdown marker +This prevents thundering herd on storage during recovery while keeping the lightweight reconnect/finalize steps sequential. -On graceful shutdown, write a `shutdown.{nodeID}` key to `hive-cluster-state` KV with timestamp and instance count. On startup, check for this marker: +**Files modified:** +- `hive/daemon/daemon.go` — refactored `restoreInstances()` with two-phase approach and semaphore -- **Marker present**: Clean shutdown. Trust KV state. Delete marker and proceed. -- **Marker absent**: Crash or power loss. Log a warning, do extra PID file validation, wait slightly longer for QEMU processes to potentially still be exiting. +#### 1.3 Clean shutdown marker — Done -**Files:** -- `hive/daemon/jetstream.go` — new `WriteShutdownMarker()` / `ReadShutdownMarker()` / `DeleteShutdownMarker()` -- `hive/daemon/daemon.go` — write marker in `setupShutdown()`, read/delete in `restoreInstances()` +Three new methods on `JetStreamManager`: +- `WriteShutdownMarker(nodeID)` — writes `shutdown.{nodeID}` to `hive-cluster-state` KV with node name and timestamp +- `ReadShutdownMarker(nodeID)` — checks if a shutdown marker exists for the node +- `DeleteShutdownMarker(nodeID)` — deletes the marker (tolerates not-found) + +**Write**: In `setupShutdown()`, the shutdown marker is written to cluster state KV before `WriteState()` persists instance state. This ensures the marker is present if the state write also succeeds. + +**Read**: At the start of `restoreInstances()`, the marker is checked. If present: log "Clean shutdown marker found, trusting KV state", delete the marker, proceed normally. If absent: log a warning about possible crash recovery, sleep 3 seconds to let orphaned QEMU processes finish exiting, then proceed with careful PID validation. + +**Files modified:** +- `hive/daemon/jetstream.go` — `WriteShutdownMarker()`, `ReadShutdownMarker()`, `DeleteShutdownMarker()` +- `hive/daemon/daemon.go` — marker write in `setupShutdown()`, marker check/delete in `restoreInstances()` + +#### 1.4 Service manifest on startup — Done + +Added `WriteServiceManifest()` to `JetStreamManager`, which writes `node.{nodeID}.services` to the cluster state KV containing the node's service list, NATS host, predastore host, and timestamp. + +Called in `Start()` after `initJetStream()` succeeds. Failure is non-fatal (logged as warning). + +**Files modified:** +- `hive/daemon/jetstream.go` — `WriteServiceManifest()` +- `hive/daemon/daemon.go` — call in `Start()` ### Phase 2: Cross-node coordination @@ -457,70 +333,91 @@ This reads from the `hive-cluster-state` KV bucket (heartbeats + service manifes **Files:** - `cmd/hive/cmd/admin.go` — new `status` subcommand -## Files to Modify +## Files Modified (Phase 0 + 1) + +| File | Changes | Phase | +|------|---------|-------| +| `hive/config/config.go` | Added `Services` field, `AllServices`, `HasService()`, `GetServices()`, updated `NodeHealthResponse` with `Services` and `ServiceHealth` | 0.1, 0.5 | +| `hive/config/config_test.go` | 5 new tests for `HasService` and `GetServices` | 0.1 | +| `hive/admin/admin.go` | Added `Services []string` to `ConfigSettings` | 0.3 | +| `cmd/hive/cmd/templates/hive.toml` | Added `services` template rendering | 0.3 | +| `cmd/hive/cmd/admin.go` | Added `--services` flag to init/join, wired to `ConfigSettings` | 0.3 | +| `hive/formation/formation.go` | Added `Services` to `NodeInfo` | 0.2 | +| `hive/formation/helpers.go` | Added `hasService()`, filtered `BuildClusterRoutes()` by nats, `BuildPredastoreNodes()` by predastore | 0.2 | +| `hive/formation/helpers_test.go` | 7 new tests for capability-filtered routing and predastore node building | 0.2 | +| `scripts/start-dev.sh` | Added `parse_services()`, `has_service()`, conditional service startup | 0.4 | +| `scripts/stop-dev.sh` | Added `parse_services()`, `has_service()` (with name mapping), conditional stop | 0.4 | +| `hive/daemon/daemon.go` | Updated health endpoint, `waitForClusterReady()`, `checkViperblockReady()`, `checkPredastoreReady()`, refactored `restoreInstances()` with semaphore, shutdown marker in `setupShutdown()`, service manifest write in `Start()` | 0.5, 1.1–1.4 | +| `hive/daemon/jetstream.go` | `ClusterStateBucket`, `clusterKV` field, `InitClusterStateBucket()`, `WriteShutdownMarker()`, `ReadShutdownMarker()`, `DeleteShutdownMarker()`, `WriteServiceManifest()`, updated `UpdateReplicas()` | 0.5, 1.3, 1.4 | + +## Files to Modify (Phase 2 + 3) | File | Changes | |------|---------| -| `hive/config/config.go` | Add `Services` field to `Config`, helper methods (`HasService`, etc.), update `NodeHealthResponse` | -| `cmd/hive/cmd/templates/hive.toml` | Add `services` field to template | -| `cmd/hive/cmd/admin.go` | Add `--services` flag to init/join, add `maintenance` and `status` subcommands | -| `hive/formation/formation.go` | Add `Services` to `NodeInfo` | -| `hive/formation/helpers.go` | Filter `BuildClusterRoutes` and `BuildPredastoreNodes` by capability | -| `scripts/start-dev.sh` | Read services from config, conditional service startup, remote dependency checks | -| `scripts/stop-dev.sh` | Read services from config, only stop local services | -| `hive/daemon/daemon.go` | `waitForClusterReady()` (capability-aware), refactor `restoreInstances()` with semaphore, heartbeat/recovery integration | -| `hive/daemon/jetstream.go` | New `hive-cluster-state` bucket, shutdown marker, heartbeat, recovery lease, service manifest, maintenance mode | +| `hive/daemon/jetstream.go` | Heartbeat, recovery lease, maintenance mode | | `hive/daemon/recovery.go` | **New file** — recovery coordinator: leader election, capability-aware VM assignment, peer monitoring | -| `hive/daemon/state.go` | Minor — shutdown marker integration with state transitions | -| `hive/daemon/daemon_handlers.go` | Update NATS health handler with capabilities and per-service health | +| `hive/daemon/daemon.go` | Heartbeat/recovery integration | +| `cmd/hive/cmd/admin.go` | `maintenance` and `status` subcommands | ## Implementation Order Phase 0 is the prerequisite — all subsequent phases depend on nodes knowing their capabilities. Phase 1 fixes immediate restart bugs. Phases 2-3 add cross-node resilience and tooling. -**Phase 0** — node capabilities (foundation): -1. `Services` field in Config + helpers -2. Formation carries capabilities -3. Capability-aware start/stop scripts -4. Health endpoint includes capabilities -5. Cluster service map in JetStream KV +**Phase 0** — node capabilities (foundation) — **Complete**: +1. [x] `Services` field in Config + helpers +2. [x] Formation carries capabilities +3. [x] Config template and CLI flags +4. [x] Capability-aware start/stop scripts +5. [x] Health endpoint includes capabilities + cluster state KV bucket + service manifest -**Phase 1** — self-recovery (fixes immediate restart bugs): -6. Cluster readiness gate (capability-aware) -7. Recovery throttling (semaphore in `restoreInstances`) -8. Clean shutdown marker +**Phase 1** — self-recovery (fixes immediate restart bugs) — **Complete**: +6. [x] Cluster readiness gate +7. [x] Recovery throttling (semaphore in `restoreInstances`) +8. [x] Clean shutdown marker +9. [x] Service manifest on startup -**Phase 2** — cross-node failure recovery: -9. Heartbeat publishing (with capabilities) -10. Peer failure detection -11. Leader-elected recovery coordinator (capability-aware placement) +**Phase 2** — cross-node failure recovery — **Planned**: +10. Heartbeat publishing (with capabilities) +11. Peer failure detection +12. Leader-elected recovery coordinator (capability-aware placement) -**Phase 3** — operational tooling: -12. Maintenance mode flag -13. Recovery status endpoints -14. Admin cluster status command +**Phase 3** — operational tooling — **Planned**: +13. Maintenance mode flag +14. Recovery status endpoints +15. Admin cluster status command ## Testing -### Unit tests -- Config `HasService()` helper methods with various service lists -- Config defaults to all services when `Services` is empty (backward compat) -- `waitForClusterReady()` with different capability configurations -- `restoreInstances()` with mocked JetStream: verify semaphore limits concurrent launches -- Shutdown marker: write on shutdown, verify presence on clean restart, verify absence simulates crash +### Phase 0 + 1 verification — Passed + +| Check | Result | +|-------|--------| +| `make test` (all unit tests) | Passed | +| `make preflight` (format + vet + security + tests) | Passed | +| `make test-docker-single` (single-node E2E) | Passed — all 9 phases | +| `make test-docker-multi` (multi-node E2E, 3 nodes) | Passed — all phases including cluster health, cross-node operations | +| Backward compatibility (no `services` field in config) | Verified — defaults to all services | + +**New unit tests added:** +- `hive/config/config_test.go` — `HasService` (explicit, empty/default, unknown), `GetServices` (default, explicit) +- `hive/formation/helpers_test.go` — `hasService` (empty=all, explicit), `BuildClusterRoutes` (filters by nats, empty=all, mixed), `BuildPredastoreNodes` (filters by predastore, empty=all) + +**Bug found during E2E**: `checkViperblockReady()` initially sent a NATS request to `ebs.{node}.health`, but viperblock does not subscribe to any health NATS topic. This caused the daemon to block in `waitForClusterReady()` for the full 2-minute timeout on every startup. Fixed by checking `natsConn.IsConnected()` instead. + +### Phase 2 + 3 tests (planned) + +#### Unit tests - Recovery coordinator: test leader election, capability-filtered VM assignment, epoch-based fencing - Heartbeat: test staleness detection thresholds -### Integration tests (multi-node Docker) -- Start 3-node cluster (all services), launch 4 VMs, stop all nodes, restart — verify all 4 VMs recover exactly once -- Start 3+1 cluster (3 full + 1 compute-only), verify compute-only node connects to remote NATS/Predastore and launches VMs -- Kill one node (SIGKILL), verify surviving nodes eventually recover its VMs after grace period, respecting capabilities -- `start-dev.sh` on a compute-only node: verify it skips NATS/Predastore startup but checks remote connectivity +#### Integration tests (multi-node Docker) +- Start 3-node cluster, launch 4 VMs, stop all, restart — verify all 4 VMs recover exactly once +- Start 3+1 cluster (3 full + 1 compute-only), verify compute-only connects to remote NATS/Predastore +- Kill one node (SIGKILL), verify surviving nodes recover its VMs after grace period - Rolling restart: stop/start nodes one at a time, verify zero duplicate launches -### Manual verification -- `make test` passes after each phase -- Deploy to 3-node test cluster (10.1.3.170-172), run stop/start cycle, check `~/hive/logs/` for race conditions +#### Manual verification +- Deploy to 3-node test cluster (10.1.3.170-172), run stop/start cycle, check logs for race conditions - Add a 4th compute-only node, verify it joins and accepts workloads ## Future Work diff --git a/hive/admin/admin.go b/hive/admin/admin.go index fa07a6ff3..c6fad9331 100644 --- a/hive/admin/admin.go +++ b/hive/admin/admin.go @@ -42,6 +42,9 @@ type ConfigSettings struct { // Predastore multi-node PredastoreNodeID int + + // Node capabilities + Services []string } // PredastoreNodeConfig describes a single Predastore node for multi-node config generation. diff --git a/hive/config/config.go b/hive/config/config.go index c18bc756f..beaafe01c 100644 --- a/hive/config/config.go +++ b/hive/config/config.go @@ -21,11 +21,12 @@ type ClusterConfig struct { // Config holds all configuration for the application type Config struct { // Node config - Node string `mapstructure:"node"` - Host string `mapstructure:"host"` // Unique hostname or IP of this node - Region string `mapstructure:"region"` - AZ string `mapstructure:"az"` - DataDir string `mapstructure:"data_dir"` + Node string `mapstructure:"node"` + Host string `mapstructure:"host"` // Unique hostname or IP of this node + Region string `mapstructure:"region"` + AZ string `mapstructure:"az"` + DataDir string `mapstructure:"data_dir"` + Services []string `mapstructure:"services"` // Which services this node runs locally Daemon DaemonConfig `mapstructure:"daemon"` NATS NATSConfig `mapstructure:"nats"` @@ -40,6 +41,32 @@ type Config struct { WalDir string `mapstructure:"wal_dir"` } +// AllServices is the default service list when Services is empty (backward compat). +var AllServices = []string{"nats", "predastore", "viperblock", "daemon", "awsgw", "ui"} + +// HasService reports whether the node runs the named service. +// An empty Services list means all services (backward compat). +func (c Config) HasService(name string) bool { + services := c.Services + if len(services) == 0 { + services = AllServices + } + for _, s := range services { + if s == name { + return true + } + } + return false +} + +// GetServices returns the configured service list, defaulting to AllServices. +func (c Config) GetServices() []string { + if len(c.Services) == 0 { + return AllServices + } + return c.Services +} + type AWSGWConfig struct { Host string `mapstructure:"host"` TLSKey string `mapstructure:"tlskey"` @@ -209,11 +236,13 @@ type NodeJoinResponse struct { } type NodeHealthResponse struct { - Node string `json:"node"` - Status string `json:"status"` - ConfigHash string `json:"config_hash"` - Epoch uint64 `json:"epoch"` - Uptime int64 `json:"uptime"` + Node string `json:"node"` + Status string `json:"status"` + ConfigHash string `json:"config_hash"` + Epoch uint64 `json:"epoch"` + Uptime int64 `json:"uptime"` + Services []string `json:"services"` + ServiceHealth map[string]string `json:"service_health,omitempty"` } // LoadConfig loads the configuration from file and environment variables diff --git a/hive/config/config_test.go b/hive/config/config_test.go index 4de701b60..715e3b1f2 100644 --- a/hive/config/config_test.go +++ b/hive/config/config_test.go @@ -219,6 +219,44 @@ expected_nodes = 3 assert.Equal(t, 3, n.AWSGW.ExpectedNodes) } +// Tests for HasService / GetServices + +func TestHasService_ExplicitList(t *testing.T) { + c := Config{Services: []string{"nats", "daemon"}} + + assert.True(t, c.HasService("nats")) + assert.True(t, c.HasService("daemon")) + assert.False(t, c.HasService("predastore")) + assert.False(t, c.HasService("viperblock")) + assert.False(t, c.HasService("ui")) +} + +func TestHasService_EmptyListBackwardCompat(t *testing.T) { + c := Config{} // no Services set + + // Empty list means all services + for _, svc := range AllServices { + assert.True(t, c.HasService(svc), "expected %s to be available with empty list", svc) + } +} + +func TestHasService_UnknownService(t *testing.T) { + c := Config{Services: []string{"nats"}} + assert.False(t, c.HasService("unknown")) +} + +func TestGetServices_DefaultsToAll(t *testing.T) { + c := Config{} + services := c.GetServices() + assert.Equal(t, AllServices, services) +} + +func TestGetServices_ExplicitList(t *testing.T) { + c := Config{Services: []string{"nats", "predastore"}} + services := c.GetServices() + assert.Equal(t, []string{"nats", "predastore"}, services) +} + // Tests for type constants func TestNBDTransportConstants(t *testing.T) { diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index 29ab1fbbc..97b9e1e5d 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -9,6 +9,7 @@ import ( "errors" "fmt" "log/slog" + "net" "os" "os/exec" "os/signal" @@ -522,6 +523,18 @@ func (d *Daemon) Start() error { return fmt.Errorf("failed to initialize JetStream: %w", err) } + // Write service manifest so other nodes know what this node runs + if d.jsManager != nil { + if err := d.jsManager.WriteServiceManifest( + d.node, + d.config.GetServices(), + d.config.NATS.Host, + d.config.Predastore.Host, + ); err != nil { + slog.Warn("Failed to write service manifest", "error", err) + } + } + // Create services before loading/launching instances, since LaunchInstance depends on them store := objectstore.NewS3ObjectStoreFromConfig(d.config.Predastore.Host, d.config.Predastore.Region, d.config.AccessKey, d.config.SecretKey) d.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(d.config, d.resourceMgr.instanceTypes, d.natsConn, &d.Instances, store) @@ -544,6 +557,7 @@ func (d *Daemon) Start() error { } d.accountService = accountSvc + d.waitForClusterReady() d.restoreInstances() if err := d.subscribeAll(); err != nil { @@ -585,7 +599,11 @@ func (d *Daemon) initJetStream() error { } if err == nil { - slog.Info("JetStream KV store initialized successfully", "replicas", 1, "attempts", attempt) + err = d.jsManager.InitClusterStateBucket() + } + + if err == nil { + slog.Info("JetStream KV stores initialized successfully", "replicas", 1, "attempts", attempt) lastErr = nil break } @@ -614,6 +632,66 @@ func (d *Daemon) initJetStream() error { return nil } +// waitForClusterReady waits until dependent infrastructure services are reachable +// before starting VM recovery. This prevents races where VMs try to mount volumes +// before viperblock/predastore are ready. +func (d *Daemon) waitForClusterReady() { + slog.Info("Waiting for cluster readiness...") + maxWait := 2 * time.Minute + start := time.Now() + interval := 2 * time.Second + + for time.Since(start) < maxWait { + ready := true + var reason string + + // Viperblock must be reachable (local or remote) + if ready && !d.checkViperblockReady() { + ready = false + reason = "viperblock not ready" + } + + // Predastore must be reachable (local or remote) + if ready && !d.checkPredastoreReady() { + ready = false + reason = "predastore not ready" + } + + if ready { + slog.Info("Cluster readiness check passed", "elapsed", time.Since(start)) + return + } + + slog.Debug("Cluster not ready, waiting...", "reason", reason, "elapsed", time.Since(start)) + time.Sleep(interval) + } + + slog.Warn("Cluster readiness timeout, proceeding with recovery anyway", "maxWait", maxWait) +} + +// checkViperblockReady checks if viperblock is reachable by verifying +// the NATS connection is up (viperblock subscribes to ebs topics on NATS). +func (d *Daemon) checkViperblockReady() bool { + if d.natsConn == nil { + return false + } + return d.natsConn.IsConnected() +} + +// checkPredastoreReady checks if predastore is reachable via TCP. +func (d *Daemon) checkPredastoreReady() bool { + host := d.config.Predastore.Host + if host == "" { + return true // no predastore configured, skip check + } + conn, err := net.DialTimeout("tcp", host, 3*time.Second) + if err != nil { + return false + } + conn.Close() + return true +} + // migrateStoppedToSharedKV writes a stopped instance to shared KV and removes // it from the local instance map. Returns true if migration succeeded. func (d *Daemon) migrateStoppedToSharedKV(instance *vm.VM) bool { @@ -631,9 +709,29 @@ func (d *Daemon) migrateStoppedToSharedKV(instance *vm.VM) bool { return true } +// maxConcurrentRecovery limits how many VMs are relaunched in parallel during recovery. +const maxConcurrentRecovery = 2 + // restoreInstances loads persisted VM state and re-launches instances that are // neither terminated nor flagged as user-stopped. func (d *Daemon) restoreInstances() { + // Check for clean shutdown marker + cleanShutdown := false + if d.jsManager != nil { + if marker, err := d.jsManager.ReadShutdownMarker(d.node); err == nil { + cleanShutdown = marker + if marker { + slog.Info("Clean shutdown marker found, trusting KV state") + _ = d.jsManager.DeleteShutdownMarker(d.node) + } + } + } + + if !cleanShutdown { + slog.Warn("No clean shutdown marker — possible crash recovery, validating QEMU PIDs carefully") + time.Sleep(3 * time.Second) + } + err := d.LoadState() if err != nil { slog.Warn("Failed to load state, continuing with empty state", "error", err) @@ -645,6 +743,9 @@ func (d *Daemon) restoreInstances() { // Ensure mutexes and QMP clients are usable after deserialization d.Instances.Mu = sync.Mutex{} + // Phase 1: Reconnect running QEMU, finalize transitional states, collect VMs to relaunch + var toLaunch []*vm.VM + for i := range d.Instances.VMS { d.Instances.VMS[i].EBSRequests.Mu = sync.Mutex{} d.Instances.VMS[i].QMPClient = &qmp.QMPClient{} @@ -706,10 +807,28 @@ func (d *Daemon) restoreInstances() { slog.Info("Instance was running but QEMU exited, relaunching", "instance", instance.ID) } - slog.Info("Launching instance", "instance", instance.ID) - if err := d.LaunchInstance(instance); err != nil { - slog.Error("Failed to launch instance", "instanceId", instance.ID, "err", err) + toLaunch = append(toLaunch, instance) + } + + // Phase 2: Relaunch crashed VMs with semaphore-based throttling + if len(toLaunch) > 0 { + slog.Info("Launching instances (recovery)", "count", len(toLaunch), "maxConcurrent", maxConcurrentRecovery) + sem := make(chan struct{}, maxConcurrentRecovery) + var wg sync.WaitGroup + + for _, instance := range toLaunch { + sem <- struct{}{} // acquire + wg.Add(1) + go func(inst *vm.VM) { + defer wg.Done() + defer func() { <-sem }() // release + slog.Info("Launching instance (recovery)", "instance", inst.ID) + if err := d.LaunchInstance(inst); err != nil { + slog.Error("Failed to launch instance", "instanceId", inst.ID, "err", err) + } + }(instance) } + wg.Wait() } // Persist state after any migrations/removals during restore @@ -835,12 +954,27 @@ func (d *Daemon) ClusterManager() error { }) } + serviceHealth := make(map[string]string) + for _, svc := range d.config.GetServices() { + serviceHealth[svc] = "ok" + } + // For remote dependencies, check connectivity + if !d.config.HasService("nats") { + if d.natsConn != nil && d.natsConn.IsConnected() { + serviceHealth["nats"] = "remote_ok" + } else { + serviceHealth["nats"] = "remote_unreachable" + } + } + response := config.NodeHealthResponse{ - Node: d.node, - Status: "running", - ConfigHash: configHash, - Epoch: d.clusterConfig.Epoch, - Uptime: int64(time.Since(d.startTime).Seconds()), + Node: d.node, + Status: "running", + ConfigHash: configHash, + Epoch: d.clusterConfig.Epoch, + Uptime: int64(time.Since(d.startTime).Seconds()), + Services: d.config.GetServices(), + ServiceHealth: serviceHealth, } return c.JSON(response) @@ -1221,6 +1355,13 @@ func (d *Daemon) setupShutdown() { } + // Write shutdown marker to cluster state KV + if d.jsManager != nil { + if err := d.jsManager.WriteShutdownMarker(d.node); err != nil { + slog.Error("Failed to write shutdown marker", "err", err) + } + } + // Write state to JetStream before closing NATS connection err := d.WriteState() if err != nil { diff --git a/hive/daemon/jetstream.go b/hive/daemon/jetstream.go index a7ff3dbda..97c5f235a 100644 --- a/hive/daemon/jetstream.go +++ b/hive/daemon/jetstream.go @@ -5,6 +5,7 @@ import ( "errors" "log/slog" "strings" + "time" "github.com/mulgadc/hive/hive/vm" "github.com/nats-io/nats.go" @@ -13,6 +14,8 @@ import ( const ( // InstanceStateBucket is the name of the KV bucket for storing instance state InstanceStateBucket = "hive-instance-state" + // ClusterStateBucket is the name of the KV bucket for cluster state (heartbeats, shutdown markers, service maps) + ClusterStateBucket = "hive-cluster-state" // InstanceStatePrefix is the key prefix for per-node instance state entries InstanceStatePrefix = "node." // StoppedInstancePrefix is the key prefix for stopped instances in shared KV @@ -21,9 +24,10 @@ const ( // JetStreamManager manages JetStream KV store operations for instance state type JetStreamManager struct { - js nats.JetStreamContext - kv nats.KeyValue - replicas int + js nats.JetStreamContext + kv nats.KeyValue // hive-instance-state + clusterKV nats.KeyValue // hive-cluster-state + replicas int } // NewJetStreamManager creates a new JetStreamManager from a NATS connection. @@ -68,6 +72,89 @@ func (m *JetStreamManager) InitKVBucket() error { return nil } +// InitClusterStateBucket initializes the cluster-state KV bucket, creating it if it doesn't exist. +func (m *JetStreamManager) InitClusterStateBucket() error { + kv, err := m.js.KeyValue(ClusterStateBucket) + if err != nil { + if errors.Is(err, nats.ErrBucketNotFound) { + slog.Info("Creating JetStream KV bucket", "bucket", ClusterStateBucket, "replicas", m.replicas) + kv, err = m.js.CreateKeyValue(&nats.KeyValueConfig{ + Bucket: ClusterStateBucket, + Description: "Hive cluster state (heartbeats, shutdown markers, service maps)", + History: 1, + Replicas: m.replicas, + TTL: 1 * time.Hour, + }) + if err != nil { + return err + } + } else { + return err + } + } else { + slog.Info("Connected to existing JetStream KV bucket", "bucket", ClusterStateBucket) + } + + m.clusterKV = kv + return nil +} + +// WriteShutdownMarker writes a shutdown marker for the given node to the cluster-state KV. +func (m *JetStreamManager) WriteShutdownMarker(nodeID string) error { + if m.clusterKV == nil { + return errors.New("cluster state KV not initialized") + } + data, _ := json.Marshal(map[string]any{ + "node": nodeID, + "timestamp": time.Now().UTC().Format(time.RFC3339), + }) + _, err := m.clusterKV.Put("shutdown."+nodeID, data) + return err +} + +// ReadShutdownMarker checks if a clean shutdown marker exists for the given node. +func (m *JetStreamManager) ReadShutdownMarker(nodeID string) (bool, error) { + if m.clusterKV == nil { + return false, errors.New("cluster state KV not initialized") + } + _, err := m.clusterKV.Get("shutdown." + nodeID) + if errors.Is(err, nats.ErrKeyNotFound) { + return false, nil + } + if err != nil { + return false, err + } + return true, nil +} + +// DeleteShutdownMarker removes the shutdown marker for the given node. +func (m *JetStreamManager) DeleteShutdownMarker(nodeID string) error { + if m.clusterKV == nil { + return errors.New("cluster state KV not initialized") + } + err := m.clusterKV.Delete("shutdown." + nodeID) + if err != nil && !errors.Is(err, nats.ErrKeyNotFound) { + return err + } + return nil +} + +// WriteServiceManifest writes the service manifest for the given node to the cluster-state KV. +func (m *JetStreamManager) WriteServiceManifest(nodeID string, services []string, natsHost, predastoreHost string) error { + if m.clusterKV == nil { + return errors.New("cluster state KV not initialized") + } + data, _ := json.Marshal(map[string]any{ + "node": nodeID, + "services": services, + "nats_host": natsHost, + "predastore_host": predastoreHost, + "timestamp": time.Now().UTC().Format(time.RFC3339), + }) + _, err := m.clusterKV.Put("node."+nodeID+".services", data) + return err +} + // WriteState writes the instance state to the KV store for the given node. // It acquires instances.Mu internally. func (m *JetStreamManager) WriteState(nodeID string, instances *vm.Instances) error { @@ -180,6 +267,21 @@ func (m *JetStreamManager) UpdateReplicas(newReplicas int) error { m.replicas = newReplicas slog.Info("Updated JetStream KV bucket replicas", "bucket", InstanceStateBucket, "oldReplicas", currentReplicas, "newReplicas", newReplicas) + + // Also update the cluster-state bucket if it exists + if m.clusterKV != nil { + clusterStreamName := "KV_" + ClusterStateBucket + clusterInfo, err := m.js.StreamInfo(clusterStreamName) + if err == nil && clusterInfo.Config.Replicas < newReplicas { + clusterInfo.Config.Replicas = newReplicas + if _, err := m.js.UpdateStream(&clusterInfo.Config); err != nil { + slog.Warn("Failed to update cluster-state bucket replicas", "error", err) + } else { + slog.Info("Updated cluster-state bucket replicas", "bucket", ClusterStateBucket, "newReplicas", newReplicas) + } + } + } + return nil } diff --git a/hive/formation/formation.go b/hive/formation/formation.go index 0e2969ca7..24b527036 100644 --- a/hive/formation/formation.go +++ b/hive/formation/formation.go @@ -13,12 +13,13 @@ import ( // NodeInfo describes a node participating in cluster formation. type NodeInfo struct { - Name string `json:"name"` - BindIP string `json:"bind_ip"` - ClusterIP string `json:"cluster_ip"` - Region string `json:"region"` - AZ string `json:"az"` - Port int `json:"port"` + Name string `json:"name"` + BindIP string `json:"bind_ip"` + ClusterIP string `json:"cluster_ip"` + Region string `json:"region"` + AZ string `json:"az"` + Port int `json:"port"` + Services []string `json:"services,omitempty"` } // JoinRequest is the payload POSTed by joining nodes. diff --git a/hive/formation/helpers.go b/hive/formation/helpers.go index 9a0fc050c..788218c5a 100644 --- a/hive/formation/helpers.go +++ b/hive/formation/helpers.go @@ -6,11 +6,32 @@ import ( "github.com/mulgadc/hive/hive/admin" ) +// hasService reports whether the node's service list includes name. +// An empty list means all services (backward compat). +func hasService(services []string, name string) bool { + if len(services) == 0 { + return true // backward compat: empty = all services + } + for _, s := range services { + if s == name { + return true + } + } + return false +} + // BuildClusterRoutes returns a sorted list of "IP:4248" NATS cluster routes -// for each node. If ClusterIP is set it is used, otherwise BindIP. -// Nodes are sorted by name for deterministic ordering. +// for nodes that run the "nats" service. If ClusterIP is set it is used, +// otherwise BindIP. Nodes are sorted by name for deterministic ordering. func BuildClusterRoutes(nodes map[string]NodeInfo) []string { - sorted := sortedNodes(nodes) + // Filter to nodes that run NATS + natsNodes := make(map[string]NodeInfo) + for k, n := range nodes { + if hasService(n.Services, "nats") { + natsNodes[k] = n + } + } + sorted := sortedNodes(natsNodes) routes := make([]string, len(sorted)) for i, n := range sorted { ip := n.ClusterIP @@ -23,9 +44,17 @@ func BuildClusterRoutes(nodes map[string]NodeInfo) []string { } // BuildPredastoreNodes returns a sorted list of PredastoreNodeConfig with -// 1-based IDs. Nodes are sorted by name for deterministic ordering. +// 1-based IDs for nodes that run the "predastore" service. +// Nodes are sorted by name for deterministic ordering. func BuildPredastoreNodes(nodes map[string]NodeInfo) []admin.PredastoreNodeConfig { - sorted := sortedNodes(nodes) + // Filter to nodes that run Predastore + predaNodes := make(map[string]NodeInfo) + for k, n := range nodes { + if hasService(n.Services, "predastore") { + predaNodes[k] = n + } + } + sorted := sortedNodes(predaNodes) out := make([]admin.PredastoreNodeConfig, len(sorted)) for i, n := range sorted { out[i] = admin.PredastoreNodeConfig{ diff --git a/hive/formation/helpers_test.go b/hive/formation/helpers_test.go new file mode 100644 index 000000000..f223e1287 --- /dev/null +++ b/hive/formation/helpers_test.go @@ -0,0 +1,87 @@ +package formation + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestHasService_EmptyListMeansAll(t *testing.T) { + t.Parallel() + assert.True(t, hasService(nil, "nats")) + assert.True(t, hasService([]string{}, "predastore")) +} + +func TestHasService_ExplicitList(t *testing.T) { + t.Parallel() + services := []string{"nats", "daemon"} + assert.True(t, hasService(services, "nats")) + assert.True(t, hasService(services, "daemon")) + assert.False(t, hasService(services, "predastore")) +} + +func TestBuildClusterRoutes_FiltersNonNATSNodes(t *testing.T) { + t.Parallel() + nodes := map[string]NodeInfo{ + "node1": {Name: "node1", BindIP: "10.0.0.1", Services: []string{"nats", "daemon"}}, + "node2": {Name: "node2", BindIP: "10.0.0.2", Services: []string{"daemon"}}, // no NATS + "node3": {Name: "node3", BindIP: "10.0.0.3", Services: []string{"nats", "predastore"}}, + } + + routes := BuildClusterRoutes(nodes) + assert.Equal(t, []string{"10.0.0.1:4248", "10.0.0.3:4248"}, routes) +} + +func TestBuildClusterRoutes_EmptyServicesIncludesAll(t *testing.T) { + t.Parallel() + // Nodes with empty services (backward compat) should all be included + nodes := map[string]NodeInfo{ + "node1": {Name: "node1", BindIP: "10.0.0.1"}, + "node2": {Name: "node2", BindIP: "10.0.0.2"}, + } + + routes := BuildClusterRoutes(nodes) + assert.Equal(t, []string{"10.0.0.1:4248", "10.0.0.2:4248"}, routes) +} + +func TestBuildPredastoreNodes_FiltersByPredastore(t *testing.T) { + t.Parallel() + nodes := map[string]NodeInfo{ + "node1": {Name: "node1", BindIP: "10.0.0.1", Services: []string{"nats", "predastore"}}, + "node2": {Name: "node2", BindIP: "10.0.0.2", Services: []string{"daemon"}}, // no predastore + "node3": {Name: "node3", BindIP: "10.0.0.3", Services: []string{"predastore"}}, + } + + pnodes := BuildPredastoreNodes(nodes) + require.Len(t, pnodes, 2) + assert.Equal(t, 1, pnodes[0].ID) + assert.Equal(t, "10.0.0.1", pnodes[0].Host) + assert.Equal(t, 2, pnodes[1].ID) + assert.Equal(t, "10.0.0.3", pnodes[1].Host) +} + +func TestBuildPredastoreNodes_EmptyServicesIncludesAll(t *testing.T) { + t.Parallel() + nodes := map[string]NodeInfo{ + "node1": {Name: "node1", BindIP: "10.0.0.1"}, + "node2": {Name: "node2", BindIP: "10.0.0.2"}, + "node3": {Name: "node3", BindIP: "10.0.0.3"}, + } + + pnodes := BuildPredastoreNodes(nodes) + require.Len(t, pnodes, 3) +} + +func TestBuildClusterRoutes_MixedServicesAndEmpty(t *testing.T) { + t.Parallel() + nodes := map[string]NodeInfo{ + "node1": {Name: "node1", BindIP: "10.0.0.1"}, // empty = all + "node2": {Name: "node2", BindIP: "10.0.0.2", Services: []string{"daemon"}}, // no nats + "node3": {Name: "node3", BindIP: "10.0.0.3", Services: []string{"nats"}}, // has nats + } + + routes := BuildClusterRoutes(nodes) + // node1 (empty=all, included), node2 (no nats, excluded), node3 (has nats, included) + assert.Equal(t, []string{"10.0.0.1:4248", "10.0.0.3:4248"}, routes) +} diff --git a/scripts/start-dev.sh b/scripts/start-dev.sh index d51523404..3d728ac09 100755 --- a/scripts/start-dev.sh +++ b/scripts/start-dev.sh @@ -30,6 +30,26 @@ echo "🚀 Starting Hive development environment..." echo "Project root: $PROJECT_ROOT" echo "Data directory: $DATA_DIR" +# Parse services from hive.toml — defaults to all if not set +parse_services() { + local config_file="$CONFIG_DIR/hive.toml" + if [ -f "$config_file" ]; then + local svc_line=$(grep -m1 '^services' "$config_file" | sed 's/.*\[//;s/\].*//;s/"//g;s/,/ /g') + if [ -n "$svc_line" ]; then + echo "$svc_line" + return + fi + fi + echo "nats predastore viperblock daemon awsgw ui" +} + +SERVICES=$(parse_services) +has_service() { + local svc="$1" + echo "$SERVICES" | grep -qw "$svc" +} +echo "Services: $SERVICES" + # Confirm configuration directory exists if [ ! -d "$CONFIG_DIR" ]; then echo "⚠️ Configuration directory $CONFIG_DIR does not exist." @@ -141,176 +161,149 @@ fi # 1️⃣ Start NATS server echo "" -echo "1️⃣ Starting NATS server..." +if has_service "nats"; then + echo "1️⃣ Starting NATS server..." -if [ -f "$CONFIG_DIR/nats/nats.conf" ]; then - export HIVE_CONFIG_PATH=$CONFIG_DIR/nats/nats.conf - echo " Using NATS config file: $CONFIG_DIR/nats/nats.conf" + if [ -f "$CONFIG_DIR/nats/nats.conf" ]; then + export HIVE_CONFIG_PATH=$CONFIG_DIR/nats/nats.conf + echo " Using NATS config file: $CONFIG_DIR/nats/nats.conf" + else + echo " ⚠️ NATS config file not found at $CONFIG_DIR/nats/nats.conf, using defaults" + export HIVE_NATS_HOST=0.0.0.0 + export HIVE_NATS_PORT=4222 + export HIVE_NATS_DATA_DIR=$DATA_DIR/nats/ + export HIVE_NATS_JETSTREAM=false + fi + + NATS_CMD="./bin/hive service nats start" + start_service "nats" "$NATS_CMD" else - echo " ⚠️ NATS config file not found at $CONFIG_DIR/nats/nats.conf, using defaults" - # TODO: Confirm, settings file will overwrite env vars (e.g multi-node config) - export HIVE_NATS_HOST=0.0.0.0 - export HIVE_NATS_PORT=4222 - export HIVE_NATS_DATA_DIR=$DATA_DIR/nats/ - export HIVE_NATS_JETSTREAM=false - #export HIVE_NATS_DEBUG=true + echo "1️⃣ Skipping NATS (not a local service)" fi -# Use air for hot reloading (dev!) -#NATS_CMD="air -c .air-nats.toml" -NATS_CMD="./bin/hive service nats start" - -start_service "nats" "$NATS_CMD" - -#start_service "nats" "go run cmd/hive/main.go service nats start" -#check_service "NATS" "4222" - # 2️⃣ Start Predastore -echo "2️⃣ Starting Predastore..." -unset HIVE_CONFIG_PATH +echo "" +if has_service "predastore"; then + echo "2️⃣ Starting Predastore..." + unset HIVE_CONFIG_PATH -export HIVE_PREDASTORE_BASE_PATH=$DATA_DIR/predastore/ -export HIVE_PREDASTORE_CONFIG_PATH=$CONFIG_DIR/predastore/predastore.toml -export HIVE_PREDASTORE_TLS_CERT=$CONFIG_DIR/server.pem -export HIVE_PREDASTORE_TLS_KEY=$CONFIG_DIR/server.key -# Very chatty logs, only for debugging -#export HIVE_PREDASTORE_DEBUG=true + export HIVE_PREDASTORE_BASE_PATH=$DATA_DIR/predastore/ + export HIVE_PREDASTORE_CONFIG_PATH=$CONFIG_DIR/predastore/predastore.toml + export HIVE_PREDASTORE_TLS_CERT=$CONFIG_DIR/server.pem + export HIVE_PREDASTORE_TLS_KEY=$CONFIG_DIR/server.key -# Auto-detect Predastore host:port from hive.toml [nodes..predastore] section -PREDASTORE_BIND="0.0.0.0:8443" -if [ -f "$CONFIG_DIR/hive.toml" ]; then - DETECTED_PREDASTORE_HOST=$(awk -F'"' '/\[nodes\..*\.predastore\]/{found=1} found && /^host/{print $2; exit}' "$CONFIG_DIR/hive.toml") - if [ -n "$DETECTED_PREDASTORE_HOST" ]; then - PREDASTORE_BIND="$DETECTED_PREDASTORE_HOST" - echo " Auto-detected Predastore bind=$PREDASTORE_BIND from hive.toml" - fi -fi -export HIVE_PREDASTORE_HOST="${PREDASTORE_BIND%%:*}" -export HIVE_PREDASTORE_PORT="${PREDASTORE_BIND##*:}" - -# Default, distributed backend. For testing, all nodes running locally. -# Specify NODE_ID to run a specific node (e.g multi-server) -export HIVE_PREDASTORE_BACKEND=distributed - -# Auto-detect Predastore NODE_ID from hive.toml if not already set -if [ -z "$HIVE_PREDASTORE_NODE_ID" ]; then + # Auto-detect Predastore host:port from hive.toml [nodes..predastore] section + PREDASTORE_BIND="0.0.0.0:8443" if [ -f "$CONFIG_DIR/hive.toml" ]; then - DETECTED_NODE_ID=$(awk -F'= *' '/node_id/{gsub(/ /,"",$2); print $2; exit}' "$CONFIG_DIR/hive.toml") - if [ -n "$DETECTED_NODE_ID" ] && [ "$DETECTED_NODE_ID" != "0" ]; then - export HIVE_PREDASTORE_NODE_ID="$DETECTED_NODE_ID" - echo " Auto-detected Predastore NODE_ID=$DETECTED_NODE_ID from hive.toml" + DETECTED_PREDASTORE_HOST=$(awk -F'"' '/\[nodes\..*\.predastore\]/{found=1} found && /^host/{print $2; exit}' "$CONFIG_DIR/hive.toml") + if [ -n "$DETECTED_PREDASTORE_HOST" ]; then + PREDASTORE_BIND="$DETECTED_PREDASTORE_HOST" + echo " Auto-detected Predastore bind=$PREDASTORE_BIND from hive.toml" fi fi + export HIVE_PREDASTORE_HOST="${PREDASTORE_BIND%%:*}" + export HIVE_PREDASTORE_PORT="${PREDASTORE_BIND##*:}" + + export HIVE_PREDASTORE_BACKEND=distributed + + # Auto-detect Predastore NODE_ID from hive.toml if not already set + if [ -z "$HIVE_PREDASTORE_NODE_ID" ]; then + if [ -f "$CONFIG_DIR/hive.toml" ]; then + DETECTED_NODE_ID=$(awk -F'= *' '/node_id/{gsub(/ /,"",$2); print $2; exit}' "$CONFIG_DIR/hive.toml") + if [ -n "$DETECTED_NODE_ID" ] && [ "$DETECTED_NODE_ID" != "0" ]; then + export HIVE_PREDASTORE_NODE_ID="$DETECTED_NODE_ID" + echo " Auto-detected Predastore NODE_ID=$DETECTED_NODE_ID from hive.toml" + fi + fi + fi + export HIVE_PREDASTORE_NODE_ID="${HIVE_PREDASTORE_NODE_ID:-}" + + PREDASTORE_CMD="./bin/hive service predastore start" + start_service "predastore" "$PREDASTORE_CMD" + check_service "Predastore" "$HIVE_PREDASTORE_HOST" "$HIVE_PREDASTORE_PORT" +else + echo "2️⃣ Skipping Predastore (not a local service)" fi -export HIVE_PREDASTORE_NODE_ID="${HIVE_PREDASTORE_NODE_ID:-}" - -# Use air for hot reloading (dev!) -#PREDASTORE_CMD="air -c .air-predastore.toml" -PREDASTORE_CMD="./bin/hive service predastore start" - -start_service "predastore" "$PREDASTORE_CMD" -check_service "Predastore" "$HIVE_PREDASTORE_HOST" "$HIVE_PREDASTORE_PORT" - -#else -# echo "" -# echo "2️⃣ Skipping Predastore for multi-node setup (TODO)" -#fi # 3️⃣ Start Viperblock echo "" -echo "3️⃣ Starting Viperblock..." +if has_service "viperblock"; then + echo "3️⃣ Starting Viperblock..." -# Determine base directory for Viperblock data, dev uses /mnt/ramdisk if available -if [ -d "/mnt/ramdisk" ] && [ -w "/mnt/ramdisk" ]; then - VB_BASE_DIR="/mnt/ramdisk/" + # Determine base directory for Viperblock data, dev uses /mnt/ramdisk if available + if [ -d "/mnt/ramdisk" ] && [ -w "/mnt/ramdisk" ]; then + VB_BASE_DIR="/mnt/ramdisk/" + else + VB_BASE_DIR="$DATA_DIR/viperblock/" + fi + + # Check if NBD plugin exists + NBD_PLUGIN_PATH="$MULGA_ROOT/viperblock/lib/nbdkit-viperblock-plugin.so" + + if [ ! -f "$NBD_PLUGIN_PATH" ]; then + echo " ⚠️ NBD plugin not found at $NBD_PLUGIN_PATH" + echo " Building Viperblock first..." + cd "$MULGA_ROOT/viperblock" + make build + cd "$PROJECT_ROOT" + fi + + export HIVE_CONFIG_PATH=$CONFIG_DIR/hive.toml + export HIVE_VIPERBLOCK_PLUGIN_PATH=$NBD_PLUGIN_PATH + export HIVE_BASE_DIR=$VB_BASE_DIR + + VIPERBLOCK_CMD="./bin/hive service viperblock start" + start_service "viperblock" "$VIPERBLOCK_CMD" else - VB_BASE_DIR="$DATA_DIR/viperblock/" + echo "3️⃣ Skipping Viperblock (not a local service)" fi -# Check if NBD plugin exists -NBD_PLUGIN_PATH="$MULGA_ROOT/viperblock/lib/nbdkit-viperblock-plugin.so" - -if [ ! -f "$NBD_PLUGIN_PATH" ]; then - echo " ⚠️ NBD plugin not found at $NBD_PLUGIN_PATH" - echo " Building Viperblock first..." - cd "$MULGA_ROOT/viperblock" - make build - cd "$PROJECT_ROOT" -fi - -export HIVE_CONFIG_PATH=$CONFIG_DIR/hive.toml -# Use config above for Viperblock settings -#export HIVE_VIPERBLOCK_S3_HOST=0.0.0.0:8443 -#export HIVE_VIPERBLOCK_S3_BUCKET=predastore -#export HIVE_VIPERBLOCK_S3_REGION=ap-southeast-2 -#export HIVE_NATS_HOST=0.0.0.0:4222 -export HIVE_VIPERBLOCK_PLUGIN_PATH=$NBD_PLUGIN_PATH -export HIVE_BASE_DIR=$VB_BASE_DIR - -#VIPERBLOCK_CMD="air -c .air-viperblock.toml" -VIPERBLOCK_CMD="./bin/hive service viperblock start" - -#VIPERBLOCK_CMD="go run cmd/hive/main.go service viperblock start \ -# --nats-host 0.0.0.0:4222 \ -# --base-dir $VB_BASE_DIR \ -# --plugin-path $NBD_PLUGIN_PATH" - -start_service "viperblock" "$VIPERBLOCK_CMD" - # 4️⃣ Start Hive Gateway/Daemon echo "" -echo "4️⃣. Starting Hive Gateway..." +if has_service "daemon"; then + echo "4️⃣. Starting Hive Gateway..." -# Use the same base directory as Viperblock for consistency -export HIVE_BASE_DIR="$VB_BASE_DIR" -export HIVE_CONFIG_PATH=$CONFIG_DIR/hive.toml -export HIVE_BASE_DIR=$DATA_DIR/hive/ -export HIVE_WAL_DIR=$WAL_DIR + export HIVE_CONFIG_PATH=$CONFIG_DIR/hive.toml + export HIVE_BASE_DIR=$DATA_DIR/hive/ + export HIVE_WAL_DIR=$WAL_DIR -#HIVE_CMD="air -c .air-hive.toml" -#HIVE_CMD="./bin/hive service hive start --config $HIVE_CONFIG_PATH--base-dir $HIVE_BASE_DIR --wal-dir $HIVE_WAL_DIR" -HIVE_CMD="./bin/hive service hive start" - -start_service "hive" "$HIVE_CMD" + HIVE_CMD="./bin/hive service hive start" + start_service "hive" "$HIVE_CMD" +else + echo "4️⃣. Skipping Hive Gateway (not a local service)" +fi # 5️⃣ Start AWS Gateway echo "" -echo "5️⃣. Starting AWS Gateway..." +if has_service "awsgw"; then + echo "5️⃣. Starting AWS Gateway..." -# Use the same base directory as Viperblock for consistency + unset HIVE_NATS_HOST + unset HIVE_PREDASTORE_HOST + export HIVE_BASE_DIR=$CONFIG_DIR + export HIVE_CONFIG_PATH=$CONFIG_DIR/hive.toml + export HIVE_AWSGW_TLS_CERT=$CONFIG_DIR/server.pem + export HIVE_AWSGW_TLS_KEY=$CONFIG_DIR/server.key -unset HIVE_NATS_HOST -unset HIVE_PREDASTORE_HOST -export HIVE_BASE_DIR=$CONFIG_DIR -export HIVE_CONFIG_PATH=$CONFIG_DIR/hive.toml -# Overwritten by config file -#export HIVE_AWSGW_HOST="0.0.0.0:9999" -export HIVE_AWSGW_TLS_CERT=$CONFIG_DIR/server.pem -export HIVE_AWSGW_TLS_KEY=$CONFIG_DIR/server.key - -#HIVE_CMD="air -c .air-hive.toml" -#HIVE_CMD="./bin/hive service hive start --config config/hive.toml --base-dir $HIVE_BASE_DIR --wal-dir $WAL_DIR" -AWSGW_CMD="./bin/hive service awsgw start" - -start_service "awsgw" "$AWSGW_CMD" - -# TODO: Need host:ip support, depending on node -#check_service "awsgw" "9999" + AWSGW_CMD="./bin/hive service awsgw start" + start_service "awsgw" "$AWSGW_CMD" +else + echo "5️⃣. Skipping AWS Gateway (not a local service)" +fi -# 6️⃣ Start Hive UI (skip with UI=false) -if [ "${UI}" != "false" ]; then +# 6️⃣ Start Hive UI (skip with UI=false or if not a local service) +if [ "${UI}" != "false" ] && has_service "ui"; then echo "" echo "6️⃣. Starting Hive UI..." HIVEUI_CMD="./bin/hive service hive-ui start" - start_service "hive-ui" "$HIVEUI_CMD" else echo "" - echo "6️⃣. Skipping Hive UI (UI=false)" + echo "6️⃣. Skipping Hive UI" fi diff --git a/scripts/stop-dev.sh b/scripts/stop-dev.sh index 515e5faeb..9bf6d1142 100755 --- a/scripts/stop-dev.sh +++ b/scripts/stop-dev.sh @@ -28,6 +28,30 @@ LOGS_DIR="$PROJECT_ROOT/data/logs" echo "Stopping Hive development environment..." +# Parse services from hive.toml — defaults to all if not set +CONFIG_DIR="${CONFIG_DIR:-$DATA_DIR/config}" +parse_services() { + local config_file="$CONFIG_DIR/hive.toml" + if [ -f "$config_file" ]; then + local svc_line=$(grep -m1 '^services' "$config_file" | sed 's/.*\[//;s/\].*//;s/"//g;s/,/ /g') + if [ -n "$svc_line" ]; then + echo "$svc_line" + return + fi + fi + echo "nats predastore viperblock daemon awsgw ui" +} + +SERVICES=$(parse_services) +has_service() { + local svc="$1" + case "$svc" in + hive) svc="daemon" ;; + hive-ui) svc="ui" ;; + esac + echo "$SERVICES" | grep -qw "$svc" +} + # Function to stop service by PID file stop_service() { local name="$1" @@ -60,10 +84,10 @@ stop_service() { echo "Stopping services..." echo "" # Stop Hive UI first (last to start) -stop_service "hive-ui" "$PID_DIR" +has_service "hive-ui" && stop_service "hive-ui" "$PID_DIR" # Stop Hive daemon/gateway (it will terminate running instances, unmount nbd devices) -stop_service "hive" "$PID_DIR" +has_service "hive" && stop_service "hive" "$PID_DIR" # Wait for all QEMU instances to exit before stopping infrastructure services. # The daemon sends system_powerdown to VMs during shutdown, but may be killed @@ -87,16 +111,16 @@ if pgrep -x qemu-system-x86_64 > /dev/null 2>&1; then fi # Stop AWSGW -stop_service "awsgw" "$PID_DIR" +has_service "awsgw" && stop_service "awsgw" "$PID_DIR" # Stop Viperblock -stop_service "viperblock" "$PID_DIR" +has_service "viperblock" && stop_service "viperblock" "$PID_DIR" # Stop Predastore -stop_service "predastore" "$PID_DIR" +has_service "predastore" && stop_service "predastore" "$PID_DIR" # Stop NATS -stop_service "nats" "$PID_DIR" +has_service "nats" && stop_service "nats" "$PID_DIR" echo "" echo "✅ Hive development environment stopped"