feat: repair instance records on a ticker, not only on state change (#988)

* feat: repair instance records on a ticker, not only on state change

A record write is best-effort, and a failed one forgets its digest so the
next reconcile retries it. But the only caller is persistState, reached
from WriteState, which fires on state change. An instance that launches,
has its record write fail, and then simply runs has no next state change,
so its record is absent for as long as it lives.

A 60s sweep on the node now reconciles the live instance set against the
digest of what it has published. The digest guard means a node with
nothing outstanding writes nothing, so the cost tracks failures rather
than instance count. The interval is the point of the change — it is what
turns an unbounded window into a bounded one — so it is asserted directly
rather than inferred from a sweep.

The sweep does not go 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.

WriteRunningSet now reports what it did, so a sweep that found work can
say so and a healthy node stays silent.

* fix: keep the QMP liveness stamp out of record change detection

Found on env19: the node owning a running instance rewrote its record on
every sweep, forever. The digest guard was working exactly as coded — it
correctly found a real difference each tick, because Health.LastQMPSuccess
advances on every successful QMP poll and the collector runs on its own
cadence. Left alone this is a KV write per running instance per node per
minute, indefinitely.

The stamp is now held out of the digest. Nothing reads it back: impairment
is judged from QMPConsecutiveFailures and ImpairedSince, and node liveness
comes from the heartbeat store, so holding it out costs no reader anything
and restores the publication cadence the record had before the sweep
existed. QMPConsecutiveFailures and ImpairedSince stay in, because an
impairment is a real transition that has to reach the record space.

Both halves are covered: a stamp moving on its own must not write, and an
impairment must still write. The first fails without this change.

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

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)
+}

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")
+}

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
 	}