diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 000c30005..618b8170c 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -807,7 +807,9 @@ func (d *Daemon) Start() error { d.waitForClusterReady() d.upgradeJetStreamReplicas() - d.restoreInstances() + if err := d.restoreInstances(); err != nil { + return fmt.Errorf("restore instances: %w", err) + } // Rebuild mgmt IP allocator from restored VMs so we don't re-allocate IPs // that are already in use by running system instances. @@ -1034,8 +1036,10 @@ func (d *Daemon) migrateTerminatedToKV(instance *vm.VM) bool { 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() { +// neither terminated nor flagged as user-stopped. Returns an error only when +// local state is unreadable (corrupt JSON or unknown schema_version) — those +// are fatal because silently dropping instance state would orphan running VMs. +func (d *Daemon) restoreInstances() error { // Check for clean shutdown marker cleanShutdown := false if d.jsManager != nil { @@ -1053,10 +1057,8 @@ func (d *Daemon) restoreInstances() { time.Sleep(3 * time.Second) } - err := d.LoadState() - if err != nil { - slog.Warn("Failed to load state, continuing with empty state", "error", err) - return + if err := d.LoadState(); err != nil { + return fmt.Errorf("load state: %w", err) } slog.Info("Loaded state", "instance count", len(d.Instances.VMS)) @@ -1258,6 +1260,7 @@ func (d *Daemon) restoreInstances() { if err := d.WriteState(); err != nil { slog.Error("Failed to persist state after restore", "error", err) } + return nil } // isInstanceProcessRunning checks if the QEMU process for an instance is still alive. @@ -1481,33 +1484,64 @@ func (d *Daemon) ClusterManager() error { return nil } -// WriteState writes the instance state to JetStream KV store (required). -// It acquires d.Instances.Mu internally. +// kvSyncTimeout bounds the best-effort cluster sync so a degraded NATS does +// not stall every state transition. 1s is well above healthy KV.Put latency +// and well below a user-visible delay. +const kvSyncTimeout = time.Second + +// localStatePath returns the on-disk path to this daemon's instance state file. +func (d *Daemon) localStatePath() string { + if d.config == nil { + return LocalStatePath("", "") + } + return LocalStatePath(d.config.DataDir, d.config.BaseDir) +} + +// WriteState persists the instance state. Local file is the source of truth; +// JetStream KV is best-effort cluster cache. The local write is fatal on +// failure; KV failures are logged and swallowed so partition-time clients +// never see "KV down" errors. +// +// Acquires d.Instances.Mu internally. Lock is held across the local write and +// the bounded KV sync so the snapshot stays consistent. func (d *Daemon) WriteState() error { - if d.jsManager == nil { - return fmt.Errorf("JetStream manager not initialized - cannot write state") + d.Instances.Mu.Lock() + defer d.Instances.Mu.Unlock() + + path := d.localStatePath() + if err := WriteLocalState(path, &d.Instances); err != nil { + slog.Error("Local state write failed", "path", path, "error", err) + return fmt.Errorf("write local state: %w", err) } - if err := d.jsManager.WriteState(d.node, &d.Instances); err != nil { - slog.Error("JetStream write failed", "error", err) - return fmt.Errorf("failed to write state to JetStream: %w", err) + + if d.jsManager != nil { + d.jsManager.WriteStateBestEffort(d.node, &d.Instances, kvSyncTimeout) } + return nil } -// LoadState loads the instance state from JetStream KV store (required) +// LoadState loads the instance state from the local file. Missing file is the +// fresh-install signal (start with empty map). Corrupt or unknown-schema files +// are fatal — caller refuses start rather than silently losing data. There is +// no KV fallback: a fresh node has empty KV anyway, so silent KV-fallback on +// file loss would just mask bugs. func (d *Daemon) LoadState() error { - if d.jsManager == nil { - return fmt.Errorf("JetStream manager not initialized - cannot load state") - } - - instances, err := d.jsManager.LoadState(d.node) + path := d.localStatePath() + state, err := ReadLocalState(path) if err != nil { - slog.Error("JetStream load failed", "error", err) - return fmt.Errorf("failed to load state from JetStream: %w", err) + slog.Error("Local state load failed", "path", path, "error", err) + return fmt.Errorf("read local state: %w", err) } - // Copy only the VMS map, not the mutex - d.Instances.VMS = instances.VMS + if state == nil { + d.Instances.VMS = make(map[string]*vm.VM) + slog.Info("No local state file, starting with empty instance map", "path", path) + return nil + } + + d.Instances.VMS = state.VMS + slog.Info("Loaded local state", "path", path, "instances", len(state.VMS)) return nil } diff --git a/spinifex/daemon/daemon_test.go b/spinifex/daemon/daemon_test.go index 8ae4364ae..74d3b7efa 100644 --- a/spinifex/daemon/daemon_test.go +++ b/spinifex/daemon/daemon_test.go @@ -915,9 +915,10 @@ func TestDaemon_BootAllocation(t *testing.T) { err = daemon.jsManager.InitKVBucket() require.NoError(t, err) - // Pre-populate JetStream with test state + // Pre-populate the local state file with test state. Post-1a, LoadState + // reads from the local file (KV is best-effort cache only). testInstances := &vm.Instances{VMS: vms} - err = daemon.jsManager.WriteState("node-1", testInstances) + err = WriteLocalState(daemon.localStatePath(), testInstances) require.NoError(t, err) // Manually trigger the LoadState and allocation logic normally found in Start() @@ -3123,19 +3124,62 @@ func TestInstanceTypeMemoryMiB_NilSafety(t *testing.T) { } // --- Daemon.WriteState / Daemon.LoadState nil jsManager --- +// +// Post-1a: local file is the source of truth, KV is best-effort. A nil +// jsManager is a valid configuration (e.g. standalone daemon, fresh install +// before NATS is up) and must not block local persistence. func TestDaemon_WriteState_NilJSManager(t *testing.T) { - d := &Daemon{jsManager: nil} - err := d.WriteState() - assert.Error(t, err) - assert.Contains(t, err.Error(), "JetStream manager not initialized") + tmpDir := t.TempDir() + d := &Daemon{ + jsManager: nil, + config: &config.Config{BaseDir: tmpDir}, + Instances: vm.Instances{VMS: map[string]*vm.VM{"i-1": {ID: "i-1"}}}, + } + require.NoError(t, d.WriteState()) + + state, err := ReadLocalState(LocalStatePath("", tmpDir)) + require.NoError(t, err) + require.NotNil(t, state) + assert.Contains(t, state.VMS, "i-1") } func TestDaemon_LoadState_NilJSManager(t *testing.T) { - d := &Daemon{jsManager: nil} + tmpDir := t.TempDir() + require.NoError(t, WriteLocalState(LocalStatePath("", tmpDir), &vm.Instances{ + VMS: map[string]*vm.VM{"i-seed": {ID: "i-seed"}}, + })) + + d := &Daemon{ + jsManager: nil, + config: &config.Config{BaseDir: tmpDir}, + } + require.NoError(t, d.LoadState()) + assert.Contains(t, d.Instances.VMS, "i-seed") +} + +func TestDaemon_LoadState_MissingFileIsFreshInstall(t *testing.T) { + d := &Daemon{ + jsManager: nil, + config: &config.Config{BaseDir: t.TempDir()}, + } + require.NoError(t, d.LoadState()) + assert.Empty(t, d.Instances.VMS) +} + +func TestDaemon_LoadState_CorruptFileFatal(t *testing.T) { + tmpDir := t.TempDir() + path := LocalStatePath("", tmpDir) + require.NoError(t, os.MkdirAll(filepath.Dir(path), 0o750)) + require.NoError(t, os.WriteFile(path, []byte("{not json"), 0o600)) + + d := &Daemon{ + jsManager: nil, + config: &config.Config{BaseDir: tmpDir}, + } err := d.LoadState() - assert.Error(t, err) - assert.Contains(t, err.Error(), "JetStream manager not initialized") + require.Error(t, err) + assert.Contains(t, err.Error(), "read local state") } // --- GetAvailableInstanceTypeInfos edge cases --- diff --git a/spinifex/daemon/jetstream.go b/spinifex/daemon/jetstream.go index 838947f58..58ec55d22 100644 --- a/spinifex/daemon/jetstream.go +++ b/spinifex/daemon/jetstream.go @@ -382,23 +382,14 @@ func (m *JetStreamManager) WriteServiceManifest(nodeID string, services []string } // WriteState writes the instance state to the KV store for the given node. -// It acquires instances.Mu internally. +// Caller must hold instances.Mu — caller is the local-state writer and the lock +// is needed across the local file write + KV sync to keep the snapshot consistent. func (m *JetStreamManager) WriteState(nodeID string, instances *vm.Instances) error { - instances.Mu.Lock() - defer instances.Mu.Unlock() - if m.kv == nil { return errors.New("KV bucket not initialized") } - // Create a struct without the mutex to avoid copying the lock - state := struct { - VMS map[string]*vm.VM `json:"vms"` - }{ - VMS: instances.VMS, - } - - jsonData, err := json.Marshal(state) + jsonData, err := marshalInstanceState(instances) if err != nil { return err } @@ -425,6 +416,59 @@ func (m *JetStreamManager) WriteState(nodeID string, instances *vm.Instances) er return nil } +// WriteStateBestEffort attempts to push instance state to KV with a deadline. +// On timeout or error, it logs a warning and returns — never blocks the caller +// past `timeout` and never returns an error. Used when the local state file is +// the source of truth and KV is a best-effort cache. +// +// Caller must hold instances.Mu so the marshalled snapshot is consistent. +// +// Note: kv.Put has no context API. On timeout, the in-flight Put goroutine +// continues and completes (or fails) on its own. This leaks at most one +// goroutine per write; under sustained partition the leak is bounded by +// WriteState call cadence (per-state-transition, not a tight loop). +func (m *JetStreamManager) WriteStateBestEffort(nodeID string, instances *vm.Instances, timeout time.Duration) { + if m.kv == nil { + slog.Debug("KV bucket not initialized, skipping cluster sync", "node", nodeID) + return + } + + jsonData, err := marshalInstanceState(instances) + if err != nil { + slog.Warn("KV sync skipped: marshal failed", "node", nodeID, "err", err) + return + } + + key := InstanceStatePrefix + nodeID + done := make(chan error, 1) + go func() { + _, putErr := m.kv.Put(key, jsonData) + done <- putErr + }() + + select { + case putErr := <-done: + if putErr != nil { + slog.Warn("KV sync failed (best-effort)", "key", key, "err", putErr) + return + } + slog.Debug("Wrote state to KV (best-effort)", "key", key, "instances", len(instances.VMS)) + case <-time.After(timeout): + slog.Warn("KV sync timed out (best-effort)", "key", key, "timeout", timeout) + } +} + +// marshalInstanceState produces the JSON wire form of instances without copying +// the embedded sync.Mutex. Caller must hold instances.Mu. +func marshalInstanceState(instances *vm.Instances) ([]byte, error) { + state := struct { + VMS map[string]*vm.VM `json:"vms"` + }{ + VMS: instances.VMS, + } + return json.Marshal(state) +} + // LoadState loads the instance state from the KV store for the given node func (m *JetStreamManager) LoadState(nodeID string) (*vm.Instances, error) { if m.kv == nil { diff --git a/spinifex/daemon/local_state.go b/spinifex/daemon/local_state.go new file mode 100644 index 000000000..64a1d91fe --- /dev/null +++ b/spinifex/daemon/local_state.go @@ -0,0 +1,138 @@ +package daemon + +import ( + "encoding/json" + "errors" + "fmt" + "log/slog" + "os" + "path/filepath" + + "github.com/mulgadc/spinifex/spinifex/vm" +) + +const ( + // LocalStateSchemaVersion is the on-disk schema version for instance-state.json. + // Bump on any breaking change to the LocalState shape; daemon refuses to start + // on an unknown version rather than silently losing data. + LocalStateSchemaVersion = 1 + + // DefaultLocalStateDir is the default directory under DataDir for local state files. + DefaultLocalStateDir = "state" + + // LocalStateFileName is the on-disk filename for per-node instance state. + LocalStateFileName = "instance-state.json" + + // DefaultDataDir is the fallback DataDir when config.DataDir is empty. + DefaultDataDir = "/var/lib/spinifex" +) + +// LocalState is the on-disk representation of a node's instance state. +// SchemaVersion gates compatibility; unknown versions are rejected. +type LocalState struct { + SchemaVersion int `json:"schema_version"` + VMS map[string]*vm.VM `json:"vms"` +} + +// LocalStatePath returns the absolute path to the per-node instance state file. +// Resolution order: dataDir, then baseDir, then DefaultDataDir. The first +// non-empty value wins. Production sets dataDir (/var/lib/spinifex); tests +// typically set baseDir to a t.TempDir(). +func LocalStatePath(dataDir, baseDir string) string { + root := dataDir + if root == "" { + root = baseDir + } + if root == "" { + root = DefaultDataDir + } + return filepath.Join(root, DefaultLocalStateDir, LocalStateFileName) +} + +// WriteLocalState atomically writes the instance state to path. +// Caller must hold instances.Mu. +// +// Atomicity: marshal → write to .tmp → fsync → rename. The rename is +// atomic on POSIX so a concurrent reader sees either the old file or the new +// one, never a half-written file. +func WriteLocalState(path string, instances *vm.Instances) error { + state := LocalState{ + SchemaVersion: LocalStateSchemaVersion, + VMS: instances.VMS, + } + + data, err := json.Marshal(state) + if err != nil { + return fmt.Errorf("marshal local state: %w", err) + } + + dir := filepath.Dir(path) + if err := os.MkdirAll(dir, 0o750); err != nil { + return fmt.Errorf("mkdir %s: %w", dir, err) + } + + tmp := path + ".tmp" + f, err := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600) + if err != nil { + return fmt.Errorf("open tmp %s: %w", tmp, err) + } + + if _, err := f.Write(data); err != nil { + f.Close() + os.Remove(tmp) + return fmt.Errorf("write tmp %s: %w", tmp, err) + } + + if err := f.Sync(); err != nil { + f.Close() + os.Remove(tmp) + return fmt.Errorf("fsync tmp %s: %w", tmp, err) + } + + if err := f.Close(); err != nil { + os.Remove(tmp) + return fmt.Errorf("close tmp %s: %w", tmp, err) + } + + if err := os.Rename(tmp, path); err != nil { + os.Remove(tmp) + return fmt.Errorf("rename %s -> %s: %w", tmp, path, err) + } + + return nil +} + +// ReadLocalState reads the per-node instance state from path. +// +// Returns (nil, nil) if the file does not exist — fresh-install signal, +// caller should start with an empty instance map. +// +// Returns an error on: +// - JSON parse failure (corruption — caller refuses start) +// - Unknown SchemaVersion (caller refuses start; safer than silent migration) +func ReadLocalState(path string) (*LocalState, error) { + data, err := os.ReadFile(path) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + slog.Debug("Local state file missing, treating as fresh install", "path", path) + return nil, nil + } + return nil, fmt.Errorf("read local state %s: %w", path, err) + } + + var state LocalState + if err := json.Unmarshal(data, &state); err != nil { + return nil, fmt.Errorf("parse local state %s: %w", path, err) + } + + if state.SchemaVersion != LocalStateSchemaVersion { + return nil, fmt.Errorf("local state %s: unknown schema_version %d (expected %d)", + path, state.SchemaVersion, LocalStateSchemaVersion) + } + + if state.VMS == nil { + state.VMS = make(map[string]*vm.VM) + } + + return &state, nil +} diff --git a/spinifex/daemon/local_state_test.go b/spinifex/daemon/local_state_test.go new file mode 100644 index 000000000..8b577b5cc --- /dev/null +++ b/spinifex/daemon/local_state_test.go @@ -0,0 +1,122 @@ +package daemon + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" + + "github.com/mulgadc/spinifex/spinifex/vm" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestLocalStatePath_PrefersDataDir(t *testing.T) { + got := LocalStatePath("/data", "/base") + assert.Equal(t, "/data/state/instance-state.json", got) +} + +func TestLocalStatePath_FallsBackToBaseDir(t *testing.T) { + got := LocalStatePath("", "/base") + assert.Equal(t, "/base/state/instance-state.json", got) +} + +func TestLocalStatePath_DefaultWhenBothEmpty(t *testing.T) { + got := LocalStatePath("", "") + assert.Equal(t, "/var/lib/spinifex/state/instance-state.json", got) +} + +func TestWriteLocalState_RoundTrip(t *testing.T) { + path := filepath.Join(t.TempDir(), "state", "instance-state.json") + + instances := &vm.Instances{ + VMS: map[string]*vm.VM{ + "i-aaa": {ID: "i-aaa", InstanceType: "t3.micro"}, + "i-bbb": {ID: "i-bbb", InstanceType: "m5.large"}, + }, + } + + require.NoError(t, WriteLocalState(path, instances)) + + loaded, err := ReadLocalState(path) + require.NoError(t, err) + require.NotNil(t, loaded) + assert.Equal(t, LocalStateSchemaVersion, loaded.SchemaVersion) + assert.Len(t, loaded.VMS, 2) + assert.Equal(t, "t3.micro", loaded.VMS["i-aaa"].InstanceType) + assert.Equal(t, "m5.large", loaded.VMS["i-bbb"].InstanceType) +} + +func TestWriteLocalState_MkdirParent(t *testing.T) { + path := filepath.Join(t.TempDir(), "deep", "nested", "state", "instance-state.json") + require.NoError(t, WriteLocalState(path, &vm.Instances{VMS: map[string]*vm.VM{}})) + _, err := os.Stat(path) + require.NoError(t, err) +} + +func TestWriteLocalState_NoTmpLeftBehind(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "instance-state.json") + require.NoError(t, WriteLocalState(path, &vm.Instances{VMS: map[string]*vm.VM{}})) + + entries, err := os.ReadDir(dir) + require.NoError(t, err) + for _, e := range entries { + assert.NotContains(t, e.Name(), ".tmp", "stale tmp file left behind") + } +} + +func TestWriteLocalState_AtomicReplace(t *testing.T) { + path := filepath.Join(t.TempDir(), "instance-state.json") + + require.NoError(t, WriteLocalState(path, &vm.Instances{VMS: map[string]*vm.VM{ + "i-1": {ID: "i-1"}, + }})) + + require.NoError(t, WriteLocalState(path, &vm.Instances{VMS: map[string]*vm.VM{ + "i-2": {ID: "i-2"}, + }})) + + loaded, err := ReadLocalState(path) + require.NoError(t, err) + require.NotNil(t, loaded) + assert.Len(t, loaded.VMS, 1) + _, ok := loaded.VMS["i-2"] + assert.True(t, ok, "second write should replace, not merge") +} + +func TestReadLocalState_Missing(t *testing.T) { + state, err := ReadLocalState(filepath.Join(t.TempDir(), "absent.json")) + require.NoError(t, err) + assert.Nil(t, state, "missing file is the fresh-install signal") +} + +func TestReadLocalState_CorruptJSON(t *testing.T) { + path := filepath.Join(t.TempDir(), "instance-state.json") + require.NoError(t, os.WriteFile(path, []byte("{not json"), 0o600)) + + _, err := ReadLocalState(path) + require.Error(t, err) + assert.Contains(t, err.Error(), "parse local state") +} + +func TestReadLocalState_UnknownSchema(t *testing.T) { + path := filepath.Join(t.TempDir(), "instance-state.json") + data, _ := json.Marshal(map[string]any{"schema_version": 99, "vms": map[string]any{}}) + require.NoError(t, os.WriteFile(path, data, 0o600)) + + _, err := ReadLocalState(path) + require.Error(t, err) + assert.Contains(t, err.Error(), "unknown schema_version 99") +} + +func TestReadLocalState_NilVMSMapInitialized(t *testing.T) { + path := filepath.Join(t.TempDir(), "instance-state.json") + data, _ := json.Marshal(map[string]any{"schema_version": LocalStateSchemaVersion}) + require.NoError(t, os.WriteFile(path, data, 0o600)) + + state, err := ReadLocalState(path) + require.NoError(t, err) + require.NotNil(t, state) + assert.NotNil(t, state.VMS) +} diff --git a/spinifex/daemon/state_test.go b/spinifex/daemon/state_test.go index ced7724f5..8f9760e1c 100644 --- a/spinifex/daemon/state_test.go +++ b/spinifex/daemon/state_test.go @@ -573,7 +573,7 @@ func TestTransitionState_MultipleInstancesIndependent(t *testing.T) { func simulateCleanRestore(t *testing.T, daemon *Daemon) { t.Helper() require.NoError(t, daemon.jsManager.WriteShutdownMarker(daemon.node)) - daemon.restoreInstances() + require.NoError(t, daemon.restoreInstances()) } // TestRestoreInstances_StoppingFinalizedToStopped verifies that an instance