diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index e9535c209..887a51a30 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -259,6 +259,10 @@ type Daemon struct { // stateRevision: incremented on each successful WriteState. stateRevision atomic.Uint64 + // recordRepairInterval: overrides recordRepairInterval, for tests that must + // observe a sweep without waiting a minute. Zero uses the real bound. + recordRepairInterval time.Duration + // kvSyncFailures: best-effort JetStream KV sync failures since start. kvSyncFailures atomic.Int64 // lastKVSyncAt: unix-nano timestamp of the last successful KV sync. @@ -2043,6 +2047,7 @@ func (d *Daemon) startCluster() error { d.resourceMgr.initSubscriptions(d.natsConn, d.handleEC2RunInstances, d.handleSystemLaunchInstance, d.node) d.startHeartbeat() + d.startRecordRepair() d.vmMgr.StartPendingWatchdog(d.ctx) // Reality→desired GC backstop (ADR-0003 §3): finish teardown interrupted by diff --git a/spinifex/daemon/instance_record_repair.go b/spinifex/daemon/instance_record_repair.go new file mode 100644 index 000000000..dc9d9f472 --- /dev/null +++ b/spinifex/daemon/instance_record_repair.go @@ -0,0 +1,75 @@ +package daemon + +import ( + "log/slog" + "time" + + "github.com/mulgadc/spinifex/spinifex/otelsetup" + "github.com/mulgadc/spinifex/spinifex/vm" +) + +// recordRepairInterval bounds how long an instance record can stay unpublished. +// +// Record writes are best-effort and a failed one is otherwise retried only by +// the next state change, so an instance that launches, fails its write and then +// simply runs would have no record for as long as it lived. This sweep is the +// bound on that window, and the bound is the point of it. +const recordRepairInterval = 60 * time.Second + +// startRecordRepair republishes instance records whose write failed, on a +// ticker, independent of state change. The goroutine exits when d.ctx is +// cancelled. +// +// It does not fire immediately: the daemon writes its state during startup, and +// a sweep racing that would reconcile against a set still being populated. +func (d *Daemon) startRecordRepair() { + if d.jsManager == nil { + slog.Warn("JetStream not initialized, skipping instance record repair") + return + } + + interval := d.recordRepairInterval + if interval <= 0 { + interval = recordRepairInterval + } + + d.shutdownWg.Go(func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + + for { + select { + case <-d.ctx.Done(): + slog.Info("Instance record repair stopping") + return + case <-ticker.C: + d.repairInstanceRecords() + } + } + }) + + slog.Info("Instance record repair started", "interval_ms", otelsetup.Millis(interval)) +} + +// repairInstanceRecords runs one sweep. It reconciles the node's live instances +// against the digest of what it has already published, so a node with nothing +// outstanding writes nothing and the cost tracks failures rather than instance +// count. +// +// Deliberately not routed through persistState: the local state file is already +// correct and rewriting it would advance the state revision for no reason. Only +// the record half needs repairing. +func (d *Daemon) repairInstanceRecords() { + var result RunningSetResult + d.vmMgr.View(func(vms map[string]*vm.VM) { + result = d.jsManager.WriteRunningSet(d.node, d.config.AZ, vms) + }) + + if result.Written == 0 && result.Retired == 0 && result.Failed == 0 { + return + } + // Only a sweep that found work says anything. Any output here means a record + // write had previously failed, so silence is the healthy state. + slog.Info("Instance record repair swept", "node", d.node, + "written", result.Written, "retired", result.Retired, "failed", result.Failed) +} diff --git a/spinifex/daemon/instance_record_repair_test.go b/spinifex/daemon/instance_record_repair_test.go new file mode 100644 index 000000000..7c54c13b6 --- /dev/null +++ b/spinifex/daemon/instance_record_repair_test.go @@ -0,0 +1,187 @@ +//test:in-package — the sweep's ticker interval and its Daemon wiring are both unexported seams +package daemon + +import ( + "context" + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/config" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/mulgadc/spinifex/spinifex/vm" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// newRecordManagerInPackage is the in-package twin of the daemon_test fixture, +// needed here because these tests build a Daemon out of unexported fields. +func newRecordManagerInPackage(t *testing.T) *JetStreamManager { + t.Helper() + _, nc, _ := testutil.StartTestJetStream(t) + + m, err := NewJetStreamManager(nc, 1) + require.NoError(t, err) + require.NoError(t, m.InitKVBucket()) + require.NoError(t, m.InitTerminatedInstanceBucket()) + return m +} + +// recordRevisionInPackage reads a record's KV revision, so a test can tell a +// sweep that wrote nothing from one that rewrote the same bytes. +func recordRevisionInPackage(t *testing.T, m *JetStreamManager, instanceID string) uint64 { + t.Helper() + _, revision, err := m.records.Get(context.Background(), instanceRecordKey(instanceID)) + require.NoError(t, err) + return revision +} + +// The interval is the bound this whole phase exists to establish, so it is +// asserted directly rather than inferred from a sweep. +func TestRecordRepairInterval_IsTheDocumentedBound(t *testing.T) { + t.Parallel() + assert.Equal(t, 60*time.Second, recordRepairInterval) +} + +func TestStartRecordRepair_NoJetStreamIsNotFatal(t *testing.T) { + t.Parallel() + d := &Daemon{ctx: t.Context(), config: &config.Config{}} + d.startRecordRepair() + d.shutdownWg.Wait() +} + +// The sweep must stop with the daemon rather than outliving it, since it writes +// to a store the shutdown is tearing down. +func TestStartRecordRepair_StopsOnContextCancel(t *testing.T) { + t.Parallel() + m := newRecordManagerInPackage(t) + + ctx, cancel := context.WithCancel(context.Background()) + d := &Daemon{ + ctx: ctx, + node: "node-1", + config: &config.Config{AZ: "us-east-1a"}, + jsManager: m, + vmMgr: vm.NewManager(), + recordRepairInterval: time.Millisecond, + } + d.startRecordRepair() + + cancel() + done := make(chan struct{}) + go func() { d.shutdownWg.Wait(); close(done) }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("the repair sweep did not stop when the daemon context was cancelled") + } +} + +// The sweep republishes a record whose write failed without the instance +// changing, which is the whole guarantee: an instance that launches, fails its +// write and then simply runs still gets a record. +func TestRepairInstanceRecords_RepublishesAFailedWriteWithNoStateChange(t *testing.T) { + t.Parallel() + m := newRecordManagerInPackage(t) + require.NoError(t, m.WriteNodeMarker("node-1")) + + // The state a failed write leaves behind: no record on the key space and no + // digest entry claiming there is one. WriteRunningSet drops the digest entry + // on failure precisely so this is what a later reconcile sees. + instance := &vm.VM{ID: "i-1", InstanceType: "t3.nano", Status: vm.StateRunning, LastNode: "node-1"} + record, err := loadRecord(m.records, instanceRecordKey("i-1")) + require.NoError(t, err) + require.Nil(t, record) + + d := &Daemon{ + ctx: context.Background(), node: "node-1", + config: &config.Config{AZ: "us-east-1a"}, jsManager: m, vmMgr: vm.NewManager(), + } + d.vmMgr.Insert(instance) + + d.repairInstanceRecords() + + record, err = loadRecord(m.records, instanceRecordKey("i-1")) + require.NoError(t, err) + require.NotNil(t, record, "the sweep should have published the record the failed write dropped") + assert.Equal(t, "node-1", record.Status.LastNode) + assert.Equal(t, "us-east-1a", record.Status.AZ) +} + +// A node with nothing outstanding must not turn the sweep into traffic: the +// digest guard is what keeps the cost proportional to failures. +func TestRepairInstanceRecords_HealthyNodeWritesNothing(t *testing.T) { + t.Parallel() + m := newRecordManagerInPackage(t) + require.NoError(t, m.WriteNodeMarker("node-1")) + + instance := &vm.VM{ID: "i-1", InstanceType: "t3.nano", Status: vm.StateRunning, LastNode: "node-1"} + first := m.WriteRunningSet("node-1", "us-east-1a", map[string]*vm.VM{instance.ID: instance}) + require.Equal(t, 1, first.Written) + + d := &Daemon{ + ctx: context.Background(), node: "node-1", + config: &config.Config{AZ: "us-east-1a"}, jsManager: m, vmMgr: vm.NewManager(), + } + d.vmMgr.Insert(instance) + + before := recordRevisionInPackage(t, m, "i-1") + d.repairInstanceRecords() + assert.Equal(t, before, recordRevisionInPackage(t, m, "i-1"), + "a sweep over an already-published set must not write") +} + +// The QMP collector advances LastQMPSuccess on its own cadence, so digesting it +// made every running instance look changed once a minute and turned the sweep +// into permanent cluster-wide write traffic. Caught on env19, not here. +func TestRepairInstanceRecords_LivenessStampAloneDoesNotWrite(t *testing.T) { + t.Parallel() + m := newRecordManagerInPackage(t) + require.NoError(t, m.WriteNodeMarker("node-1")) + + instance := &vm.VM{ID: "i-1", InstanceType: "t3.nano", Status: vm.StateRunning, LastNode: "node-1"} + instance.Health.LastQMPSuccess = time.Now() + + d := &Daemon{ + ctx: context.Background(), node: "node-1", + config: &config.Config{AZ: "us-east-1a"}, jsManager: m, vmMgr: vm.NewManager(), + } + d.vmMgr.Insert(instance) + d.repairInstanceRecords() + before := recordRevisionInPackage(t, m, "i-1") + + for i := range 3 { + d.vmMgr.UpdateState("i-1", func(v *vm.VM) { + v.Health.LastQMPSuccess = time.Now().Add(time.Duration(i+1) * time.Minute) + }) + d.repairInstanceRecords() + } + + assert.Equal(t, before, recordRevisionInPackage(t, m, "i-1"), + "a moving liveness stamp is not a state change and must not be republished") +} + +// The exclusion must be narrow: an impairment is a real transition and has to +// reach the record space, or a describe cannot see it. +func TestRepairInstanceRecords_RealHealthChangeStillWrites(t *testing.T) { + t.Parallel() + m := newRecordManagerInPackage(t) + require.NoError(t, m.WriteNodeMarker("node-1")) + + instance := &vm.VM{ID: "i-1", InstanceType: "t3.nano", Status: vm.StateRunning, LastNode: "node-1"} + d := &Daemon{ + ctx: context.Background(), node: "node-1", + config: &config.Config{AZ: "us-east-1a"}, jsManager: m, vmMgr: vm.NewManager(), + } + d.vmMgr.Insert(instance) + d.repairInstanceRecords() + before := recordRevisionInPackage(t, m, "i-1") + + d.vmMgr.UpdateState("i-1", func(v *vm.VM) { + v.Health.QMPConsecutiveFailures = 3 + v.Health.ImpairedSince = time.Now() + }) + d.repairInstanceRecords() + + assert.Greater(t, recordRevisionInPackage(t, m, "i-1"), before, + "an impairment must still be published") +} diff --git a/spinifex/daemon/instance_running_set.go b/spinifex/daemon/instance_running_set.go index 810e69804..0653e12e9 100644 --- a/spinifex/daemon/instance_running_set.go +++ b/spinifex/daemon/instance_running_set.go @@ -6,6 +6,7 @@ import ( "hash/fnv" "log/slog" "sync" + "time" "github.com/mulgadc/spinifex/spinifex/vm" ) @@ -27,16 +28,27 @@ type runningSetState struct { digest map[string]uint64 } +// RunningSetResult counts what one reconcile actually did. A healthy node that +// has published everything scores zero across the board, which is what makes it +// usable as a repair signal rather than as traffic. +type RunningSetResult struct { + Written int // records published, whether new, changed or retried after a failure + Retired int // instances released from the digest, whether deleted or handed over + Failed int // records this pass could not publish or settle +} + // WriteRunningSet reconciles this node's running instances onto the // per-resource key space. // // Best-effort, like the marker write it follows: the local state file is the // source of truth for the node's own instances, so a failure here leaves the // cluster's view stale rather than losing anything, and is logged rather than -// returned. The next state change repeats whatever did not land. -func (m *JetStreamManager) WriteRunningSet(nodeID, az string, vms map[string]*vm.VM) { +// returned. Whatever did not land is repeated by the next state change or by +// the repair sweep, whichever comes first. +func (m *JetStreamManager) WriteRunningSet(nodeID, az string, vms map[string]*vm.VM) RunningSetResult { + var result RunningSetResult if m.records == nil { - return + return result } m.running.mu.Lock() @@ -46,7 +58,8 @@ func (m *JetStreamManager) WriteRunningSet(nodeID, az string, vms map[string]*vm if err := m.seedRunningSet(nodeID); err != nil { slog.Warn("Could not read the existing instance records; the next state write retries", "node", nodeID, "err", err) - return + result.Failed++ + return result } } @@ -65,6 +78,7 @@ func (m *JetStreamManager) WriteRunningSet(nodeID, az string, vms map[string]*vm sum, err := recordDigest(record) if err != nil { slog.Error("Could not encode an instance record", "instanceId", id, "err", err) + result.Failed++ continue } if current, ok := m.running.digest[id]; ok && current == sum { @@ -77,9 +91,11 @@ func (m *JetStreamManager) WriteRunningSet(nodeID, az string, vms map[string]*vm if err := m.records.Replace(context.Background(), instanceRecordKey(id), record); err != nil { slog.Error("Could not write an instance record", "instanceId", id, "err", err) delete(m.running.digest, id) + result.Failed++ continue } m.running.digest[id] = sum + result.Written++ } for id := range m.running.digest { @@ -88,8 +104,12 @@ func (m *JetStreamManager) WriteRunningSet(nodeID, az string, vms map[string]*vm } if m.retireRecord(nodeID, id) { delete(m.running.digest, id) + result.Retired++ + continue } + result.Failed++ } + return result } // retireRecord drops the record of an instance that has left nodeID's running @@ -158,8 +178,18 @@ func (m *JetStreamManager) seedRunningSet(nodeID string) error { // recordDigest identifies a record by its wire form, so a state change that // leaves an instance untouched does not become a KV write. encoding/json sorts // map keys, so the same record always encodes to the same bytes. +// +// LastQMPSuccess is held out of it. It advances on every successful poll, so +// digesting it would make every running instance look changed once a minute and +// turn the repair sweep into permanent write traffic. Nothing reads it back — +// impairment is judged from QMPConsecutiveFailures and ImpairedSince, and node +// liveness from the heartbeat store — so holding it out costs no reader +// anything and keeps publication on the cadence it had before the sweep existed. func recordDigest(record *vm.InstanceRecord) (uint64, error) { - data, err := json.Marshal(record) + stable := *record + stable.Status.Health.LastQMPSuccess = time.Time{} + + data, err := json.Marshal(&stable) if err != nil { return 0, err }