feat: cut instance records over to one key per instance (#938)

* feat: decide instance set membership from the record, not the key prefix

* feat: cut instance records over to one key per instance

The record space becomes the only copy. Which set an instance belongs to
was structural before this — a stopped instance sat at instance.<id> and a
running one inside the node.<nodeID> blob, so it could not be wrong — and
is now a predicate over the record, decided in one place rather than at
each call site.

The old key spaces are frozen, not deleted. A migration that removed them
would make the crossing unrollbackable, and the node blob is the only
record of what a node was running that a build predating the cutover can
read. instance.<id> and terminated.<id> drain on the delete path.

The per-node key survives as a presence marker at its own prefix. Reads
need it because a node with no instances and a node the cluster has no
record of both scan empty, and only one of them may replace what the node
has locally. It carries no instances, so a state change no longer rewrites
the node's whole running set.

Claiming a stopped instance becomes a compare-and-set. It cannot wait for
a later change: with one key holding an instance for its whole life, a
claim that deletes is deleting the instance it is about to start. The
winner clears the stop intent and leaves LastNode alone, so an instance
whose launch never happens is still recoverable by the node that ran it.

A build predating the cutover now refuses to open a bucket stamped past
what it understands, rather than reading the keys it knows and writing
back a view assembled from half a key space.

* fix: carry node ownership out of the blob key during the cutover migration

The node.<id> key carried two facts in its name: that the node exists, and
which instances are its. The 4->5 migration moved the first and not the
second.

A blob member's own LastNode is empty, because the key was the owner, and so
is last_node on the record copied forward from it. After the cutover
ownership is read off the record, so an unowned record is one no node lists
- and with the presence marker seeded, LoadRunningState returns found=true
with an empty set, which restore adopts over the instances the node is
actually running.

The migration now stamps last_node from the blob's key onto every record in
it that has no owner, and leaves a record that already names one alone.

Found on a live cluster before crossing it. The test helpers built blob
members with LastNode already set, which production blobs never have; the
new tests seed them the way the bucket actually holds them.

spinifex/daemon/daemon.go

@@ -2603,11 +2603,6 @@ func (d *Daemon) persistState(nodeID string, vms map[string]*vm.VM) error {
 	if err != nil {
 		return fmt.Errorf("marshal state: %w", err)
 	}
-	kvData, err := marshalInstanceState(vms)
-	if err != nil {
-		return fmt.Errorf("marshal state: %w", err)
-	}
-
 	// The KV write is independent of local disk health (JetStream, not this
 	// disk), so a local failure must not skip it — attempt both, then report
 	// the local failure. Revision only advances when the local write (the
@@ -2620,12 +2615,13 @@ func (d *Daemon) persistState(nodeID string, vms map[string]*vm.VM) error {
 		d.stateRevision.Add(1)
 	}
 
-	// The mirror follows the blob it mirrors, so a record on the new key space
-	// is never newer than the running set it belongs to. vms is the caller's
-	// snapshot and cannot change underneath either write.
+	// The marker is written before the records so a reader that sees the node
+	// at all can only under-report it, never conclude it has no instances from
+	// a set that has not been written yet. vms is the caller's snapshot and
+	// cannot change underneath either write.
 	if d.jsManager != nil {
-		d.jsManager.WriteStateBytesBestEffort(nodeID, kvData, kvSyncTimeout)
-		d.jsManager.MirrorRunningSet(nodeID, vms)
+		d.jsManager.WriteNodeMarkerBestEffort(nodeID, kvSyncTimeout)
+		d.jsManager.WriteRunningSet(nodeID, vms)
 	}
 
 	if localErr != nil {

spinifex/daemon/daemon_handlers_test.go

@@ -1193,20 +1193,26 @@ func TestHandleEC2StartStoppedInstance_MissingInstanceID(t *testing.T) {
 	assert.Equal(t, awserrors.ErrorMissingParameter, errResp["Code"])
 }
 
+// Starting an instance that is not stopped is IncorrectInstanceState rather
+// than NotFound, which is the distinction a caller retries on.
+//
+// The instance is put in the vmMgr and nowhere else on purpose. Writing a
+// running instance through WriteStoppedInstance no longer constructs this: one
+// key holds an instance for its whole life, so that write stamps the state it
+// means instead of storing a record the listing would then decline to find.
+// What is left is the race the taxonomy was written for — the stopped record is
+// gone because a claim already started the instance here.
 func TestHandleEC2StartStoppedInstance_NotStoppedState(t *testing.T) {
 	natsURL := sharedJSNATSURL
 
 	daemon := createFullTestDaemonWithJetStream(t, natsURL)
 
-	// Write an instance in running state to shared KV
-	runningVM := &vm.VM{
+	daemon.vmMgr.Insert(&vm.VM{
 		ID:           "i-running-shared",
 		Status:       vm.StateRunning,
 		InstanceType: getTestInstanceType(t),
 		AccountID:    testAccountID,
-	}
-	err := daemon.jsManager.WriteStoppedInstance(runningVM.ID, runningVM)
-	require.NoError(t, err)
+	})
 
 	sub, err := daemon.natsConn.QueueSubscribe("ec2.start", "spinifex-workers", asMsgHandler(daemon.handleEC2StartStoppedInstance))
 	require.NoError(t, err)
@@ -1220,9 +1226,6 @@ func TestHandleEC2StartStoppedInstance_NotStoppedState(t *testing.T) {
 	err = json.Unmarshal(reply.Data, &errResp)
 	require.NoError(t, err)
 	assert.Equal(t, awserrors.ErrorIncorrectInstanceState, errResp["Code"])
-
-	// Cleanup
-	_ = daemon.jsManager.DeleteStoppedInstance(runningVM.ID)
 }
 
 // --- handleEC2DescribeStoppedInstances tests ---

spinifex/daemon/dns_reconcile.go

@@ -26,10 +26,13 @@ func (d *Daemon) dnsWatchSources() []reconciler.Source {
 	}
 }
 
-// instanceStateWatchFilter matches the per-node instance blobs. EC2 records are
-// the one DNS input that is node-local, so a change to any node's blob is the
-// only signal that this node's own VMs may need re-asserting.
-const instanceStateWatchFilter = InstanceStatePrefix + "*"
+// instanceStateWatchFilter matches the instance records. EC2 records are the
+// one DNS input that is node-local, so a change to any instance is the only
+// signal that this node's own VMs may need re-asserting.
+//
+// This is a NATS subject filter, so "*" spans one dot-delimited token and the
+// prefix has to end in a dot for it to match anything at all.
+const instanceStateWatchFilter = InstanceRecordPrefix + "*"
 
 // instanceStateWatchBuckets returns the shared instance-state bucket. It is a
 // fresh kvstore.Bucket rather than the JetStreamManager's handle so a recovery

spinifex/daemon/dns_reconcile_test.go

@@ -33,7 +33,7 @@ func TestDNSWatchSources_OneSourcePerDesiredSetInput(t *testing.T) {
 		filters = append(filters, s.Filter())
 	}
 	assert.ElementsMatch(t, []string{
-		InstanceStatePrefix + "*",
+		InstanceRecordPrefix + "*",
 		handlers_elbv2.KeyPrefixLB + "*",
 		">",
 		">",

spinifex/daemon/export_test.go

@@ -0,0 +1,13 @@
+package daemon
+
+//test:in-package — an export_test.go is in-package by definition; that is the
+//whole of what it does. The tests using these live in daemon_test.
+
+// The membership predicates, exported for the external test package. They are
+// unexported in production because nothing outside the package should be
+// deciding which set an instance is in, but they are worth testing directly:
+// the key prefix used to answer this and could not be wrong, and these can.
+var (
+	OperatorStopped = operatorStopped
+	RunsOn          = runsOn
+)

spinifex/daemon/instance_membership.go

@@ -0,0 +1,37 @@
+package daemon
+
+import "github.com/mulgadc/spinifex/spinifex/vm"
+
+// One key now holds an instance for its whole life, so which set it belongs to
+// is a predicate over the record rather than the prefix it sits under. Before
+// the cutover it was structural — a stopped instance sat at instance.<id> and a
+// running one inside node.<nodeID> — and could not be wrong. It can be now, so
+// it is decided here once instead of at each call site.
+
+// operatorStopped reports whether a record is stopped in the sense
+// DescribeStoppedInstances means: stopped because someone asked for it.
+//
+// Status alone will not do. A node's DRAIN sequence also leaves an instance in
+// StateStopped, and restore relaunches those rather than listing them.
+// DesiredState is what separates the two, and this is the same test
+// classifyRestoredInstances already makes.
+func operatorStopped(record *vm.InstanceRecord) bool {
+	if record == nil {
+		return false
+	}
+	return record.Status.Status == vm.StateStopped &&
+		record.Spec.DesiredState == vm.DesiredStopped
+}
+
+// runsOn reports whether a record belongs to nodeID's running set: everything
+// that node last owned except what an operator stopped.
+//
+// StateTerminated is deliberately included. Restore migrates those to the
+// terminated bucket, and a record it cannot see is one it cannot migrate —
+// which is the "void" its own comment warns about.
+func runsOn(record *vm.InstanceRecord, nodeID string) bool {
+	if record == nil || nodeID == "" {
+		return false
+	}
+	return record.Status.LastNode == nodeID && !operatorStopped(record)
+}

spinifex/daemon/instance_membership_test.go

@@ -0,0 +1,72 @@
+package daemon_test
+
+import (
+	"testing"
+
+	"github.com/mulgadc/spinifex/spinifex/daemon"
+	"github.com/mulgadc/spinifex/spinifex/vm"
+	"github.com/stretchr/testify/assert"
+)
+
+// allInstanceStates is every state an instance can be persisted in. The table
+// below covers all of them on purpose: before the cutover the key an instance
+// sat under decided which set it was in, so no state could put it in the wrong
+// one. Now one does, and an unconsidered state is how that goes wrong quietly.
+var allInstanceStates = []vm.InstanceState{
+	vm.StateProvisioning,
+	vm.StatePending,
+	vm.StateRunning,
+	vm.StateStopping,
+	vm.StateStopped,
+	vm.StateShuttingDown,
+	vm.StateTerminated,
+	vm.StateError,
+}
+
+func membershipRecord(status vm.InstanceState, desired vm.DesiredState, node string) *vm.InstanceRecord {
+	return (&vm.VM{ID: "i-1", Status: status, DesiredState: desired, LastNode: node}).Record()
+}
+
+func TestOperatorStopped_OnlyStoppedAndAskedFor(t *testing.T) {
+	for _, status := range allInstanceStates {
+		for _, desired := range []vm.DesiredState{vm.DesiredRunning, vm.DesiredStopped} {
+			want := status == vm.StateStopped && desired == vm.DesiredStopped
+			assert.Equal(t, want, daemon.OperatorStopped(membershipRecord(status, desired, "node-1")),
+				"status=%s desired=%q", status, desired)
+		}
+	}
+}
+
+// The case that makes the pair necessary rather than the status alone: a DRAIN
+// stop leaves StateStopped with DesiredRunning, and restore relaunches it. Read
+// as operator-stopped it would be listed as stopped and never come back.
+func TestOperatorStopped_DrainStoppedIsNotOperatorStopped(t *testing.T) {
+	drained := membershipRecord(vm.StateStopped, vm.DesiredRunning, "node-1")
+	assert.False(t, daemon.OperatorStopped(drained))
+	assert.True(t, daemon.RunsOn(drained, "node-1"), "a drain-stopped instance is still the node's to relaunch")
+}
+
+func TestRunsOn_EverythingTheNodeOwnsExceptOperatorStopped(t *testing.T) {
+	for _, status := range allInstanceStates {
+		stopped := status == vm.StateStopped
+		assert.Equal(t, !stopped, daemon.RunsOn(membershipRecord(status, vm.DesiredStopped, "node-1"), "node-1"),
+			"desired-stopped, status=%s", status)
+		assert.True(t, daemon.RunsOn(membershipRecord(status, vm.DesiredRunning, "node-1"), "node-1"),
+			"desired-running, status=%s", status)
+	}
+}
+
+// Terminated records stay in the running set until restore has moved them to
+// the terminated bucket. Excluding them here is what its own comment calls a
+// void: gone from the running set and not yet anywhere else.
+func TestRunsOn_TerminatedIsStillTheNodesToMigrate(t *testing.T) {
+	assert.True(t, daemon.RunsOn(membershipRecord(vm.StateTerminated, vm.DesiredRunning, "node-1"), "node-1"))
+}
+
+func TestRunsOn_AnotherNodesRecordIsNotOurs(t *testing.T) {
+	assert.False(t, daemon.RunsOn(membershipRecord(vm.StateRunning, vm.DesiredRunning, "node-2"), "node-1"))
+	assert.False(t, daemon.RunsOn(membershipRecord(vm.StateRunning, vm.DesiredRunning, ""), "node-1"))
+	assert.False(t, daemon.RunsOn(membershipRecord(vm.StateRunning, vm.DesiredRunning, "node-1"), ""))
+	assert.False(t, daemon.RunsOn(nil, "node-1"))
+	assert.False(t, daemon.OperatorStopped(nil))
+}

spinifex/daemon/instance_records.go

@@ -3,7 +3,6 @@ package daemon
 import (
 	"context"
 	"errors"
-	"fmt"
 	"log/slog"
 
 	"github.com/mulgadc/spinifex/spinifex/kvstore"
@@ -157,111 +156,51 @@ func (m *JetStreamManager) ListTerminatedInstanceRecords() ([]*vm.InstanceRecord
 	return listRecords(m.termRecords, InstanceRecordPrefix)
 }
 
-// mirrorRecord writes instance to the per-resource key space, after the write
-// it mirrors has already committed at the key it replaces.
+// writeRecord writes instance to the per-resource key space.
 //
-// Reads answer from the mirror wherever one exists, so what they depend on is
-// that a record present there is never older than the one at the key it
-// mirrors. A mirror write that fails takes the new key with it, which restores
-// that by leaving the reader the value it mirrors — a degraded write, not a
-// failed one, so it is logged rather than returned. Only a failure to remove
-// the stale mirror is returned: that is the one outcome leaving a read able to
-// serve a record older than the one just committed.
-func mirrorRecord(store *kvstore.Store[vm.InstanceRecord], instanceID string, instance *vm.VM) error {
-	key := instanceRecordKey(instanceID)
-	err := store.Replace(context.Background(), key, instance.Record())
-	if err == nil {
-		return nil
-	}
-
-	if delErr := store.Delete(context.Background(), key); delErr != nil {
-		return fmt.Errorf("mirror %s failed and its stale record could not be removed: %w",
-			key, errors.Join(err, delErr))
-	}
-	slog.Error("Instance record mirror failed; reads fall back to the record it mirrors",
-		"key", key, "instanceId", instanceID, "err", err)
-	return nil
+// Replace rather than Set: the write is wholesale, and doing it under CAS is
+// what stops a racing update landing out of order.
+func writeRecord(store *kvstore.Store[vm.InstanceRecord], instanceID string, instance *vm.VM) error {
+	return store.Replace(context.Background(), instanceRecordKey(instanceID), instance.Record())
 }
 
-// Until the cutover the per-resource key space is a shadow: the key it
-// replaces decides whether an instance exists, and the record supplies its
-// value.
-//
-// The read cannot be the other way round while any node predates the new
-// space. Such a node removes an instance by deleting only the key it knows —
-// a claim is exactly that, one atomic delete — so the mirror outlives it, and
-// a reader taking the mirror as proof of existence reports a started instance
-// as stopped for as long as that record survives. A three-node rolling deploy
-// on env19 produced exactly that: one instance listed running and stopped at
-// once, indefinitely.
-//
-// Answering from the record still runs the conversion on every read, which is
-// the point of a transition release. Only the existence question moves.
-func loadPreferringRecord(records *kvstore.Store[vm.InstanceRecord], legacy *kvstore.Store[vm.VM], legacyKey, instanceID string) (*vm.VM, error) {
-	instance, _, err := legacy.Get(context.Background(), legacyKey)
-	if err != nil {
-		if errors.Is(err, kvstore.ErrNotFound) {
-			return nil, nil
-		}
+// loadInstance reads one instance from the per-resource key space, reporting an
+// absent key as nil. want filters what counts as present: a key holds an
+// instance for its whole life now, so "is there a record" and "is it the kind
+// of instance I asked for" are different questions.
+func loadInstance(store *kvstore.Store[vm.InstanceRecord], instanceID string,
+	want func(*vm.InstanceRecord) bool) (*vm.VM, error) {
+	record, err := loadRecord(store, instanceRecordKey(instanceID))
+	if err != nil || record == nil {
 		return nil, err
 	}
-
-	// A decode failure at the record key is returned rather than fallen back
-	// on: it means the conversion is wrong, and hiding it behind the key the
-	// record mirrors is how that reaches a cutover unnoticed.
-	record, err := loadRecord(records, instanceRecordKey(instanceID))
-	if err != nil {
-		return nil, err
-	}
-	if record != nil {
-		return vm.VMFromRecord(record), nil
-	}
-	return instance, nil
-}
-
-// listPreferringRecords lists the key space being replaced and answers each
-// entry from its mirror where one exists. A mirror with nothing behind it is
-// not listed; see loadPreferringRecord for why one can be there at all.
-func listPreferringRecords(records *kvstore.Store[vm.InstanceRecord], legacy *kvstore.Store[vm.VM], legacyPrefix string) ([]*vm.VM, error) {
-	older, err := listRecords(legacy, legacyPrefix)
-	if err != nil {
-		return nil, err
-	}
-	if len(older) == 0 {
+	if want != nil && !want(record) {
 		return nil, nil
 	}
+	return vm.VMFromRecord(record), nil
+}
 
-	newer, err := listRecords(records, InstanceRecordPrefix)
+// listInstances returns every instance in the key space that want accepts.
+func listInstances(store *kvstore.Store[vm.InstanceRecord],
+	want func(*vm.InstanceRecord) bool) ([]*vm.VM, error) {
+	records, err := listRecords(store, InstanceRecordPrefix)
 	if err != nil {
 		return nil, err
 	}
-	mirrored := make(map[string]*vm.InstanceRecord, len(newer))
-	for _, record := range newer {
-		mirrored[record.Metadata.Name] = record
-	}
 
-	out := make([]*vm.VM, 0, len(older))
-	for _, instance := range older {
-		if record, ok := mirrored[instance.ID]; ok {
-			out = append(out, vm.VMFromRecord(record))
+	out := make([]*vm.VM, 0, len(records))
+	for _, record := range records {
+		if want != nil && !want(record) {
 			continue
 		}
-		out = append(out, instance)
+		out = append(out, vm.VMFromRecord(record))
+	}
+	if len(out) == 0 {
+		return nil, nil
 	}
 	return out, nil
 }
 
-// deleteBothRecords removes an instance from both key spaces. Both deletes are
-// idempotent, so a retry repairs a partial failure; until then the instance is
-// still readable through whichever key survived, which is why the error is
-// returned rather than logged.
-func deleteBothRecords(records *kvstore.Store[vm.InstanceRecord], legacy *kvstore.Store[vm.VM], legacyKey, instanceID string) error {
-	if err := records.Delete(context.Background(), instanceRecordKey(instanceID)); err != nil {
-		return err
-	}
-	return legacy.Delete(context.Background(), legacyKey)
-}
-
 // loadRecord reads one record, reporting an absent key as nil rather than as an
 // error. Every Load accessor on this manager answers that way, so a caller
 // asking after an instance that is simply not there does not have to know which

spinifex/daemon/instance_records_migrate.go

@@ -7,11 +7,16 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/mulgadc/spinifex/spinifex/kvutil"
 	"github.com/mulgadc/spinifex/spinifex/migrate"
 	"github.com/mulgadc/spinifex/spinifex/vm"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
+// errRecordAbsent marks a blob member with no record left to own, so the scan
+// can tell that apart from a read that failed.
+var errRecordAbsent = errors.New("instance record absent")
+
 // The copy-forward onto the per-resource key space. Running it as a migration
 // rather than lazily on read is what lets a listing merge the two key spaces
 // cheaply instead of falling back per key: after this, the only records missing
@@ -44,6 +49,12 @@ func init() {
 		Description: "re-key instance records from i/<id> to i.<id>",
 		Run:         rekeyRecordSeparator,
 	})
+	migrate.DefaultRegistry.RegisterKV(InstanceStateBucket, migrate.KVMigration{
+		FromVersion: 4,
+		ToVersion:   5,
+		Description: "carry each node's presence and ownership out of its running-set blob",
+		Run:         carryNodeOwnershipForward,
+	})
 	migrate.DefaultRegistry.RegisterKV(TerminatedInstanceBucket, migrate.KVMigration{
 		FromVersion: 1,
 		ToVersion:   2,
@@ -111,6 +122,123 @@ func rekeyRecordSeparator(ctx context.Context, kvc migrate.KVContext) error {
 	return nil
 }
 
+// carryNodeOwnershipForward moves the two things the node.<id> key carried in
+// its name into the key space that replaces it: that the node exists, and which
+// instances are its.
+//
+// The presence marker comes first. Without it a node reads no marker until its
+// own first state write, and a node that boots before it writes would be told
+// the cluster has no record of it. That is the safe direction — restore declines
+// to adopt rather than adopting an empty set — but it also means a node cannot
+// recover its instances from KV across the crossing, which is the one time it
+// most needs to.
+//
+// The blob is read, not consumed. It is what a node rolled back to the previous
+// release reads, and this is the step that stops being reversible if it moves.
+func carryNodeOwnershipForward(ctx context.Context, kvc migrate.KVContext) error {
+	keys, err := kvc.KV.Keys(ctx)
+	if err != nil {
+		if errors.Is(err, jetstream.ErrNoKeysFound) {
+			return nil
+		}
+		return fmt.Errorf("list keys: %w", err)
+	}
+
+	marker, err := json.Marshal(LocalState{SchemaVersion: LocalStateSchemaVersion})
+	if err != nil {
+		return fmt.Errorf("encode presence marker: %w", err)
+	}
+
+	seeded, owned := 0, 0
+	for _, key := range keys {
+		if !strings.HasPrefix(key, InstanceStatePrefix) {
+			continue
+		}
+		nodeID := strings.TrimPrefix(key, InstanceStatePrefix)
+
+		if _, err := kvc.KV.Create(ctx, NodePresencePrefix+nodeID, marker); err != nil {
+			// A node that has already upgraded wrote its own marker; it is the
+			// same value, and the newer write is not overwritten with this one.
+			if !errors.Is(err, jetstream.ErrKeyExists) {
+				return fmt.Errorf("write %s: %w", NodePresencePrefix+nodeID, err)
+			}
+		} else {
+			seeded++
+		}
+
+		stamped, err := stampBlobOwnership(ctx, kvc, key, nodeID)
+		if err != nil {
+			return err
+		}
+		owned += stamped
+	}
+
+	kvc.Logger.Info("Carried node presence and ownership onto the record key space",
+		"markersSeeded", seeded, "recordsStamped", owned)
+	return nil
+}
+
+// stampBlobOwnership names the owner of every instance in a node's blob on the
+// record that replaces it, reporting how many it had to write.
+//
+// This is the half of the split that is easy to miss. Which node ran an instance
+// used to be the name of the key its blob sat under, so nothing wrote it into
+// the instance, and the records copied forward from those blobs carry no owner
+// at all. After the cutover ownership is read off the record, and an unowned
+// record is one no node lists — so the node that boots next reads a marker
+// saying the cluster knows it and a running set saying it runs nothing, and
+// adopts that over the instances it actually has.
+func stampBlobOwnership(ctx context.Context, kvc migrate.KVContext, key, nodeID string) (int, error) {
+	entry, err := kvc.KV.Get(ctx, key)
+	if err != nil {
+		if errors.Is(err, jetstream.ErrKeyNotFound) {
+			return 0, nil
+		}
+		return 0, fmt.Errorf("read %s: %w", key, err)
+	}
+
+	var state LocalState
+	if err := json.Unmarshal(entry.Value(), &state); err != nil {
+		return 0, fmt.Errorf("decode %s: %w", key, err)
+	}
+	if err := checkNodeStateVersion(key, state.SchemaVersion); err != nil {
+		return 0, err
+	}
+
+	stamped := 0
+	for id := range state.VMS {
+		recordKey := instanceRecordKey(id)
+		wrote := false
+		_, err := kvutil.Update(ctx, kvc.KV, recordKey,
+			kvutil.CASConfig{NotFound: errRecordAbsent},
+			func(record *vm.InstanceRecord) (bool, error) {
+				wrote = false
+				// An owner already on the record is the newer fact, whoever it
+				// names: this one is reconstructed from a blob that stopped
+				// being written the moment the cutover landed.
+				if record.Status.LastNode != "" {
+					return false, nil
+				}
+				record.Status.LastNode = nodeID
+				wrote = true
+				return true, nil
+			})
+		if err != nil {
+			// No record to own. The split that creates them has already run, so
+			// what is missing here was deleted afterwards, and the delete is the
+			// newer fact.
+			if errors.Is(err, errRecordAbsent) {
+				continue
+			}
+			return 0, fmt.Errorf("stamp owner on %s: %w", recordKey, err)
+		}
+		if wrote {
+			stamped++
+		}
+	}
+	return stamped, nil
+}
+
 // copyInstancesForward writes a record for every key under prefix.
 //
 // Two nodes can reach this at once — the version stamp is a read-then-write,

spinifex/daemon/instance_records_mirror.go

@@ -1,176 +0,0 @@
-package daemon
-
-import (
-	"context"
-	"encoding/json"
-	"hash/fnv"
-	"log/slog"
-	"sync"
-
-	"github.com/mulgadc/spinifex/spinifex/vm"
-)
-
-// mirrorState tracks what this node last put on the per-resource key space, so
-// a change to one instance writes one record.
-//
-// Without it every state change would rewrite the whole running set one key at
-// a time, which is worse than the single blob it replaces rather than better:
-// splitting the blob exists to stop two instances on a node serialising against
-// each other.
-type mirrorState struct {
-	mu     sync.Mutex
-	seeded bool
-	digest map[string]uint64
-}
-
-// MirrorRunningSet reconciles this node's running instances onto the
-// per-resource key space, after the blob it mirrors has been written.
-//
-// Best-effort, like that blob write: the local state file is the source of
-// truth, the blob still decides which instances are running here, and a read
-// that finds no mirror answers from the blob member. So a failure degrades a
-// read rather than breaking one, and is logged rather than returned.
-func (m *JetStreamManager) MirrorRunningSet(nodeID string, vms map[string]*vm.VM) {
-	if m.records == nil || m.stopped == nil {
-		return
-	}
-
-	m.mirror.mu.Lock()
-	defer m.mirror.mu.Unlock()
-
-	if !m.mirror.seeded {
-		if err := m.seedRunningMirror(nodeID); err != nil {
-			slog.Warn("Could not read the existing instance records; the next state write retries",
-				"node", nodeID, "err", err)
-			return
-		}
-	}
-
-	for id, instance := range vms {
-		if instance == nil {
-			continue
-		}
-		sum, err := recordDigest(instance.Record())
-		if err != nil {
-			slog.Error("Could not encode an instance record", "instanceId", id, "err", err)
-			continue
-		}
-		if current, ok := m.mirror.digest[id]; ok && current == sum {
-			continue
-		}
-
-		// A failed mirror forgets its digest as well as its record, so the next
-		// state write repeats the attempt rather than reading the failure as
-		// already mirrored.
-		if err := mirrorRecord(m.records, id, instance); err != nil {
-			slog.Error("Could not mirror an instance record", "instanceId", id, "err", err)
-			delete(m.mirror.digest, id)
-			continue
-		}
-		m.mirror.digest[id] = sum
-	}
-
-	for id := range m.mirror.digest {
-		if _, ok := vms[id]; ok {
-			continue
-		}
-		if m.retireMirror(id) {
-			delete(m.mirror.digest, id)
-		}
-	}
-}
-
-// retireMirror drops the record of an instance that has left this node's
-// running set, reporting whether the question is settled.
-//
-// It is not settled by deleting unconditionally. The stopped key space shares
-// the record key, and an instance that stops leaves the running set as
-// WriteStoppedInstance mirrors it there, so an unconditional delete would race
-// a live write — and would keep deleting, on every state change, the records of
-// instances stopped on this node. Where the stopped key holds the instance the
-// record is handed over rather than retired, which is also what it becomes at
-// the cutover, when the two key spaces are one.
-func (m *JetStreamManager) retireMirror(instanceID string) bool {
-	stopped, err := loadRecord(m.stopped, StoppedInstancePrefix+instanceID)
-	if err != nil {
-		slog.Warn("Could not tell whether a departed instance is stopped; leaving its record in place",
-			"instanceId", instanceID, "err", err)
-		return false
-	}
-	if stopped != nil {
-		return true
-	}
-
-	if err := m.records.Delete(context.Background(), instanceRecordKey(instanceID)); err != nil {
-		slog.Error("Could not remove the record of an instance that left this node",
-			"instanceId", instanceID, "err", err)
-		return false
-	}
-	return true
-}
-
-// seedRunningMirror primes the digest map from the records already on the key
-// space for this node, so the first state write after boot rewrites only what
-// actually changed while the node was down — and so a record left behind by a
-// previous process is still a candidate for retirement.
-//
-// Ownership is read from status.LastNode, the same field the GC uses to decide
-// whose records a node may touch.
-func (m *JetStreamManager) seedRunningMirror(nodeID string) error {
-	existing, err := listRecords(m.records, InstanceRecordPrefix)
-	if err != nil {
-		return err
-	}
-
-	digest := make(map[string]uint64, len(existing))
-	for _, record := range existing {
-		if record.Status.LastNode != nodeID {
-			continue
-		}
-		sum, err := recordDigest(record)
-		if err != nil {
-			return err
-		}
-		digest[record.Metadata.Name] = sum
-	}
-
-	m.mirror.digest = digest
-	m.mirror.seeded = true
-	slog.Debug("Seeded the instance record mirror", "node", nodeID, "records", len(digest))
-	return nil
-}
-
-// 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.
-func recordDigest(record *vm.InstanceRecord) (uint64, error) {
-	data, err := json.Marshal(record)
-	if err != nil {
-		return 0, err
-	}
-	h := fnv.New64a()
-	_, _ = h.Write(data)
-	return h.Sum64(), nil
-}
-
-// answerFromMirrors replaces each member of a node's running set with its
-// mirror where there is one. The blob still decides membership — see
-// loadPreferringRecord for why a mirror cannot be trusted to — so a record with
-// no member behind it is ignored here.
-func (m *JetStreamManager) answerFromMirrors(vms map[string]*vm.VM) error {
-	if m.records == nil || len(vms) == 0 {
-		return nil
-	}
-
-	mirrored, err := listRecords(m.records, InstanceRecordPrefix)
-	if err != nil {
-		return err
-	}
-	for _, record := range mirrored {
-		if _, ok := vms[record.Metadata.Name]; !ok {
-			continue
-		}
-		vms[record.Metadata.Name] = vm.VMFromRecord(record)
-	}
-	return nil
-}

spinifex/daemon/instance_records_mirror_test.go

@@ -1,314 +0,0 @@
-package daemon_test
-
-import (
-	"context"
-	"encoding/json"
-	"testing"
-
-	"github.com/mulgadc/spinifex/spinifex/daemon"
-	"github.com/mulgadc/spinifex/spinifex/kvutil"
-	"github.com/mulgadc/spinifex/spinifex/testutil"
-	"github.com/mulgadc/spinifex/spinifex/vm"
-	"github.com/nats-io/nats.go"
-	"github.com/nats-io/nats.go/jetstream"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-const mirrorNode = "node-1"
-
-func runningVM(id, instanceType string) *vm.VM {
-	return &vm.VM{ID: id, InstanceType: instanceType, Status: vm.StateRunning, LastNode: mirrorNode}
-}
-
-func runningSet(vms ...*vm.VM) map[string]*vm.VM {
-	set := make(map[string]*vm.VM, len(vms))
-	for _, v := range vms {
-		set[v.ID] = v
-	}
-	return set
-}
-
-// writeRunningSet does what persistState does: commit the blob, then mirror
-// what it committed.
-func writeRunningSet(t *testing.T, m *daemon.JetStreamManager, vms ...*vm.VM) {
-	t.Helper()
-	require.NoError(t, m.WriteState(mirrorNode, runningSet(vms...)))
-}
-
-// recordRevision reads the revision of a record straight from the bucket, so a
-// test can tell a write that did nothing from one that did not happen.
-func recordRevision(t *testing.T, nc *nats.Conn, instanceID string) uint64 {
-	t.Helper()
-	js, err := jetstream.New(nc)
-	require.NoError(t, err)
-	kv, err := js.KeyValue(context.Background(), daemon.InstanceStateBucket)
-	require.NoError(t, err)
-	entry, err := kv.Get(context.Background(), daemon.InstanceRecordPrefix+instanceID)
-	require.NoError(t, err)
-	return entry.Revision()
-}
-
-func TestMirrorRunningSet_WritesARecordPerMember(t *testing.T) {
-	m := newRecordManager(t)
-
-	writeRunningSet(t, m, runningVM("i-1", "t3.nano"), runningVM("i-2", "t3.micro"))
-
-	records, err := m.ListInstanceRecords()
-	require.NoError(t, err)
-	byID := make(map[string]string, len(records))
-	for _, record := range records {
-		byID[record.Metadata.Name] = record.Spec.InstanceType
-	}
-	assert.Equal(t, map[string]string{"i-1": "t3.nano", "i-2": "t3.micro"}, byID,
-		"one blob of two instances must become two records")
-}
-
-func TestLoadState_AnswersFromTheMirrorWhereThereIsOne(t *testing.T) {
-	m := newRecordManager(t)
-	require.NoError(t, m.WriteState(mirrorNode, runningSet(runningVM("i-1", "t3.nano"))))
-
-	// Written directly rather than through the mirror, standing in for a record
-	// committed after the blob it belongs to.
-	require.NoError(t, m.WriteInstanceRecord("i-1", runningVM("i-1", "m7i.large").Record()))
-
-	loaded, found, err := m.LoadState(mirrorNode)
-
-	require.NoError(t, err)
-	require.True(t, found)
-	require.Contains(t, loaded, "i-1")
-	assert.Equal(t, "m7i.large", loaded["i-1"].InstanceType, "the record must win over the blob member")
-}
-
-func TestLoadState_FallsBackToTheBlobMember(t *testing.T) {
-	m := newRecordManager(t)
-	require.NoError(t, m.WriteState(mirrorNode, runningSet(runningVM("i-1", "t3.nano"))))
-
-	loaded, found, err := m.LoadState(mirrorNode)
-
-	require.NoError(t, err)
-	require.True(t, found)
-	require.Contains(t, loaded, "i-1", "an instance only the blob holds must still be found")
-	assert.Equal(t, "t3.nano", loaded["i-1"].InstanceType)
-}
-
-// The blob decides which instances are running here. A record with no member
-// behind it is what a node predating the key space leaves when it stops an
-// instance, and reading it as running is the fault env19 found.
-func TestLoadState_IgnoresARecordWithNoBlobMember(t *testing.T) {
-	m := newRecordManager(t)
-	require.NoError(t, m.WriteState(mirrorNode, runningSet(runningVM("i-1", "t3.nano"))))
-	require.NoError(t, m.WriteInstanceRecord("i-orphan", runningVM("i-orphan", "m7i.large").Record()))
-
-	loaded, found, err := m.LoadState(mirrorNode)
-
-	require.NoError(t, err)
-	require.True(t, found)
-	assert.NotContains(t, loaded, "i-orphan")
-	assert.Len(t, loaded, 1)
-}
-
-// The conversion's totality is guarded in the vm package; what this adds is
-// that the running-set path actually routes through it, on fields from both
-// halves of the record rather than the instance type alone.
-func TestMirrorRunningSet_CarriesBothHalvesOfTheRecord(t *testing.T) {
-	m := newRecordManager(t)
-	instance := runningVM("i-1", "t3.nano")
-	instance.DesiredState = vm.DesiredRunning
-	instance.PublicIP = "10.0.0.9"
-	instance.ENIId = "eni-1"
-	instance.HostfwdPorts = []int{2222}
-
-	writeRunningSet(t, m, instance)
-
-	loaded, _, err := m.LoadState(mirrorNode)
-	require.NoError(t, err)
-	require.Contains(t, loaded, "i-1")
-	got := loaded["i-1"]
-	assert.Equal(t, vm.DesiredRunning, got.DesiredState)
-	assert.Equal(t, []int{2222}, got.HostfwdPorts)
-	assert.Equal(t, "10.0.0.9", got.PublicIP)
-	assert.Equal(t, "eni-1", got.ENIId)
-	assert.Equal(t, mirrorNode, got.LastNode)
-}
-
-func TestMirrorRunningSet_RetiresAMemberThatLeft(t *testing.T) {
-	m := newRecordManager(t)
-	writeRunningSet(t, m, runningVM("i-1", "t3.nano"), runningVM("i-2", "t3.micro"))
-
-	writeRunningSet(t, m, runningVM("i-1", "t3.nano"))
-
-	gone, err := m.LoadInstanceRecord("i-2")
-	require.NoError(t, err)
-	assert.Nil(t, gone, "a record left behind by a terminated instance never expires from this bucket")
-
-	kept, err := m.LoadInstanceRecord("i-1")
-	require.NoError(t, err)
-	assert.NotNil(t, kept)
-}
-
-// An instance that stops leaves the running set at the moment the stopped key
-// space takes it over, and that path has already written the record. Retiring
-// it here would race a live write and would keep re-deleting the records of
-// every instance ever stopped on this node.
-func TestMirrorRunningSet_HandsOverAMemberThatStopped(t *testing.T) {
-	m := newRecordManager(t)
-	writeRunningSet(t, m, runningVM("i-1", "t3.nano"))
-
-	require.NoError(t, m.WriteStoppedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano", Status: vm.StateStopped}))
-	writeRunningSet(t, m)
-
-	record, err := m.LoadInstanceRecord("i-1")
-	require.NoError(t, err)
-	require.NotNil(t, record, "the stopped key space owns this record now")
-
-	stopped, err := m.ListStoppedInstances()
-	require.NoError(t, err)
-	require.Len(t, stopped, 1)
-	assert.Equal(t, "i-1", stopped[0].ID)
-}
-
-// Rewriting every record on every state change would reintroduce, one key at a
-// time, the contention splitting the blob exists to remove.
-func TestMirrorRunningSet_WritesOnlyWhatChanged(t *testing.T) {
-	m, nc := newRecordManagerConn(t)
-	writeRunningSet(t, m, runningVM("i-1", "t3.nano"), runningVM("i-2", "t3.micro"))
-	before1, before2 := recordRevision(t, nc, "i-1"), recordRevision(t, nc, "i-2")
-
-	writeRunningSet(t, m, runningVM("i-1", "t3.nano"), runningVM("i-2", "t3.micro"))
-
-	assert.Equal(t, before1, recordRevision(t, nc, "i-1"), "an unchanged instance must not be rewritten")
-	assert.Equal(t, before2, recordRevision(t, nc, "i-2"))
-
-	changed := runningVM("i-1", "t3.nano")
-	changed.PublicIP = "10.0.0.9"
-	writeRunningSet(t, m, changed, runningVM("i-2", "t3.micro"))
-
-	assert.Greater(t, recordRevision(t, nc, "i-1"), before1, "a changed instance must be rewritten")
-	assert.Equal(t, before2, recordRevision(t, nc, "i-2"), "and only that instance")
-}
-
-// A restarted process has no memory of what it last wrote, so it primes itself
-// from the bucket rather than rewriting every record to find out.
-func TestMirrorRunningSet_SeedsFromWhatIsAlreadyThere(t *testing.T) {
-	first, nc := newRecordManagerConn(t)
-	writeRunningSet(t, first, runningVM("i-1", "t3.nano"))
-	before := recordRevision(t, nc, "i-1")
-
-	second, err := daemon.NewJetStreamManager(nc, 1)
-	require.NoError(t, err)
-	require.NoError(t, second.InitKVBucket())
-	writeRunningSet(t, second, runningVM("i-1", "t3.nano"))
-
-	assert.Equal(t, before, recordRevision(t, nc, "i-1"))
-}
-
-// Ownership is read from status.LastNode, the same field the GC uses. A node
-// must not retire a record for an instance that is running somewhere else.
-func TestMirrorRunningSet_LeavesAnotherNodesRecordsAlone(t *testing.T) {
-	m := newRecordManager(t)
-	elsewhere := &vm.VM{ID: "i-elsewhere", InstanceType: "t3.nano", Status: vm.StateRunning, LastNode: "node-2"}
-	require.NoError(t, m.WriteInstanceRecord("i-elsewhere", elsewhere.Record()))
-
-	writeRunningSet(t, m, runningVM("i-1", "t3.nano"))
-
-	record, err := m.LoadInstanceRecord("i-elsewhere")
-	require.NoError(t, err)
-	assert.NotNil(t, record)
-}
-
-// seedNodeBlobBucket creates the instance-state bucket at the version before
-// the blob split, holding one node's running set as a single record.
-func seedNodeBlobBucket(t *testing.T, nc *nats.Conn, nodeID string, vms map[string]*vm.VM) jetstream.KeyValue {
-	t.Helper()
-	ctx := context.Background()
-
-	js, err := jetstream.New(nc)
-	require.NoError(t, err)
-	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: daemon.InstanceStateBucket, History: 1})
-	require.NoError(t, err)
-
-	data, err := json.Marshal(daemon.LocalState{SchemaVersion: daemon.LocalStateSchemaVersion, VMS: vms})
-	require.NoError(t, err)
-	_, err = kv.Put(ctx, daemon.InstanceStatePrefix+nodeID, data)
-	require.NoError(t, err)
-
-	require.NoError(t, kvutil.WriteVersion(ctx, kv, 2))
-	return kv
-}
-
-func TestInstanceStateMigration_SplitsTheNodeBlob(t *testing.T) {
-	_, nc, _ := testutil.StartTestJetStream(t)
-	seedNodeBlobBucket(t, nc, mirrorNode, runningSet(runningVM("i-1", "t3.nano"), runningVM("i-2", "t3.micro")))
-
-	m, err := daemon.NewJetStreamManager(nc, 1)
-	require.NoError(t, err)
-	require.NoError(t, m.InitKVBucket())
-
-	records, err := m.ListInstanceRecords()
-	require.NoError(t, err)
-	byID := make(map[string]string, len(records))
-	for _, record := range records {
-		byID[record.Metadata.Name] = record.Spec.InstanceType
-	}
-	assert.Equal(t, map[string]string{"i-1": "t3.nano", "i-2": "t3.micro"}, byID)
-}
-
-// Release 1 keeps writing the blob, so the split must copy it rather than
-// consume it: a node still on the previous release reads nothing else.
-func TestInstanceStateMigration_KeepsTheNodeBlob(t *testing.T) {
-	_, nc, _ := testutil.StartTestJetStream(t)
-	seedNodeBlobBucket(t, nc, mirrorNode, runningSet(runningVM("i-1", "t3.nano")))
-
-	m, err := daemon.NewJetStreamManager(nc, 1)
-	require.NoError(t, err)
-	require.NoError(t, m.InitKVBucket())
-
-	loaded, found, err := m.LoadState(mirrorNode)
-	require.NoError(t, err)
-	require.True(t, found)
-	require.Contains(t, loaded, "i-1")
-}
-
-// Every node's set is copied, not just the one running the migration: it runs
-// once per bucket, and a node that is down would otherwise have no records
-// until it came back.
-func TestInstanceStateMigration_SplitsEveryNodesBlob(t *testing.T) {
-	_, nc, _ := testutil.StartTestJetStream(t)
-	kv := seedNodeBlobBucket(t, nc, mirrorNode, runningSet(runningVM("i-1", "t3.nano")))
-
-	other := &vm.VM{ID: "i-2", InstanceType: "t3.micro", Status: vm.StateRunning, LastNode: "node-2"}
-	data, err := json.Marshal(daemon.LocalState{
-		SchemaVersion: daemon.LocalStateSchemaVersion,
-		VMS:           map[string]*vm.VM{other.ID: other},
-	})
-	require.NoError(t, err)
-	_, err = kv.Put(context.Background(), daemon.InstanceStatePrefix+"node-2", data)
-	require.NoError(t, err)
-
-	m, err := daemon.NewJetStreamManager(nc, 1)
-	require.NoError(t, err)
-	require.NoError(t, m.InitKVBucket())
-
-	record, err := m.LoadInstanceRecord("i-2")
-	require.NoError(t, err)
-	require.NotNil(t, record)
-	assert.Equal(t, "node-2", record.Status.LastNode)
-}
-
-func TestInstanceStateMigration_BlobSplitIsSafeToRunTwice(t *testing.T) {
-	_, nc, _ := testutil.StartTestJetStream(t)
-	kv := seedNodeBlobBucket(t, nc, mirrorNode, runningSet(runningVM("i-1", "t3.nano")))
-
-	first, err := daemon.NewJetStreamManager(nc, 1)
-	require.NoError(t, err)
-	require.NoError(t, first.InitKVBucket())
-	before := recordRevision(t, nc, "i-1")
-
-	require.NoError(t, kvutil.WriteVersion(context.Background(), kv, 2))
-	second, err := daemon.NewJetStreamManager(nc, 1)
-	require.NoError(t, err)
-	require.NoError(t, second.InitKVBucket())
-
-	assert.Equal(t, before, recordRevision(t, nc, "i-1"), "a second run must not overwrite what the first copied")
-}

spinifex/daemon/instance_records_test.go

@@ -70,36 +70,31 @@ func TestLoadInstanceRecord_AbsentIsNil(t *testing.T) {
 	assert.Nil(t, got)
 }
 
-// The two key spaces are distinct keys sharing one bucket: a write lands at
-// both, a listing merges them so the instance appears once rather than twice,
-// and removing one copy leaves the other readable. Nothing but the prefix
-// keeps them apart, so this fails if it is ever changed to something colliding.
-func TestStoppedInstances_SpanBothKeySpaces(t *testing.T) {
+// A stopped instance is written once, to the record key, and read back from
+// there through both the instance accessors and the record accessors.
+func TestStoppedInstances_LiveAtTheRecordKey(t *testing.T) {
 	m := newRecordManager(t)
 
 	require.NoError(t, m.WriteStoppedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"}))
 
 	stopped, err := m.ListStoppedInstances()
 	require.NoError(t, err)
-	require.Len(t, stopped, 1, "an instance held at both keys must be listed once")
+	require.Len(t, stopped, 1, "an instance must be listed once")
 	assert.Equal(t, "t3.nano", stopped[0].InstanceType)
 
 	records, err := m.ListInstanceRecords()
 	require.NoError(t, err)
-	require.Len(t, records, 1, "the write must have been mirrored onto the record key")
+	require.Len(t, records, 1)
 	assert.Equal(t, "t3.nano", records[0].Spec.InstanceType)
 
-	// Dropping the mirror leaves the key it mirrors untouched, which is what
-	// makes the fallback a real fallback rather than a second name for one key.
 	require.NoError(t, m.DeleteInstanceRecord("i-1"))
 
 	got, err := m.LoadStoppedInstance("i-1")
 	require.NoError(t, err)
-	require.NotNil(t, got, "deleting the mirror must not touch the key it mirrors")
-	assert.Equal(t, "t3.nano", got.InstanceType)
+	assert.Nil(t, got, "there is one key, so removing it removes the instance")
 }
 
-func TestTerminatedInstances_SpanBothKeySpaces(t *testing.T) {
+func TestTerminatedInstances_LiveAtTheRecordKey(t *testing.T) {
 	m := newRecordManager(t)
 
 	require.NoError(t, m.WriteTerminatedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"}))
@@ -118,8 +113,26 @@ func TestTerminatedInstances_SpanBothKeySpaces(t *testing.T) {
 
 	got, err := m.LoadTerminatedInstance("i-1")
 	require.NoError(t, err)
-	require.NotNil(t, got)
-	assert.Equal(t, "t3.nano", got.InstanceType)
+	assert.Nil(t, got)
+}
+
+// Four key spaces share the instance-state bucket and a listing selects between
+// them by string prefix, not by subject token. "i." is a prefix of nothing else
+// here and nothing else is a prefix of it, which is the whole of what keeps a
+// record listing from picking up a frozen key or a node's marker.
+func TestInstanceRecords_ThePrefixesInOneBucketAreDisjoint(t *testing.T) {
+	m, nc := newRecordManagerConn(t)
+
+	require.NoError(t, m.WriteInstanceRecord("i-1", testRecord("i-1")))
+	require.NoError(t, m.WriteNodeMarker("i-1"))
+	seedFrozenKey(t, nc, daemon.InstanceStateBucket, daemon.StoppedInstancePrefix+"i-1")
+	seedFrozenKey(t, nc, daemon.InstanceStateBucket, daemon.InstanceStatePrefix+"i-1")
+
+	records, err := m.ListInstanceRecords()
+
+	require.NoError(t, err)
+	require.Len(t, records, 1, "only the record key may answer a record listing")
+	assert.Equal(t, "i-1", records[0].Metadata.Name)
 }
 
 // The two buckets are separate stores over the same prefix, so a record in one

spinifex/daemon/instance_records_transition_test.go

@@ -15,108 +15,32 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-// legacyOnly leaves an instance at the key i/<id> replaces and nowhere else,
-// which is the state a node that predates the per-resource key space leaves
-// behind on every write it makes.
-func legacyOnly(t *testing.T, m *daemon.JetStreamManager, instance *vm.VM) {
-	t.Helper()
-	require.NoError(t, m.WriteStoppedInstance(instance.ID, instance))
-	require.NoError(t, m.DeleteInstanceRecord(instance.ID))
-}
-
-func TestLoadStoppedInstance_PrefersTheRecordKey(t *testing.T) {
-	m := newRecordManager(t)
-	require.NoError(t, m.WriteStoppedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"}))
-
-	// Only reachable by writing the record key directly; the accessors keep the
-	// two in step. It stands in for a mirror written after the key it mirrors.
-	fresher := testRecord("i-1")
-	require.NoError(t, m.WriteInstanceRecord("i-1", fresher))
-
-	got, err := m.LoadStoppedInstance("i-1")
-	require.NoError(t, err)
-	require.NotNil(t, got)
-	assert.Equal(t, "m7i.large", got.InstanceType, "the record key must win over the key it replaces")
-}
-
-func TestLoadStoppedInstance_FallsBackToTheKeyItReplaces(t *testing.T) {
-	m := newRecordManager(t)
-	legacyOnly(t, m, &vm.VM{ID: "i-1", InstanceType: "t3.nano"})
-
-	got, err := m.LoadStoppedInstance("i-1")
-
-	require.NoError(t, err)
-	require.NotNil(t, got, "an instance only the old key holds must still be found")
-	assert.Equal(t, "t3.nano", got.InstanceType)
-}
-
-// An instance held only at the key being replaced is listed from there, and
-// one held at both is answered from its mirror. A mirror with nothing behind
-// it is not an instance — see TestOlderNodeClaim_* below.
-func TestListStoppedInstances_AnswersFromTheMirrorWhereThereIsOne(t *testing.T) {
-	m := newRecordManager(t)
-
-	legacyOnly(t, m, &vm.VM{ID: "i-old", InstanceType: "t3.nano"})
-	require.NoError(t, m.WriteStoppedInstance("i-both", &vm.VM{ID: "i-both", InstanceType: "t3.micro"}))
-
-	stopped, err := m.ListStoppedInstances()
-
-	require.NoError(t, err)
-	byID := make(map[string]string, len(stopped))
-	for _, instance := range stopped {
-		byID[instance.ID] = instance.InstanceType
-	}
-	assert.Equal(t, map[string]string{"i-old": "t3.nano", "i-both": "t3.micro"}, byID)
-}
-
-// olderNodeClaim reproduces what a node predating the per-resource space does
-// when it claims: one atomic delete of the only key it knows, leaving the
-// mirror behind. Staged underneath the manager because no accessor can do it.
-func olderNodeClaim(t *testing.T, nc *nats.Conn, instanceID string) {
+// seedFrozenKey writes raw bytes at one of the key spaces the record replaced.
+// Nothing does this any more, which is the point: it stages what a cluster that
+// crossed the cutover still has lying around.
+func seedFrozenKey(t *testing.T, nc *nats.Conn, bucket, key string) {
 	t.Helper()
 	js, err := jetstream.New(nc)
 	require.NoError(t, err)
-	kv, err := js.KeyValue(context.Background(), daemon.InstanceStateBucket)
+	kv, err := js.KeyValue(context.Background(), bucket)
+	require.NoError(t, err)
+	_, err = kv.Put(context.Background(), key, []byte(`{"id":"i-1"}`))
 	require.NoError(t, err)
-	require.NoError(t, kv.Delete(context.Background(), daemon.StoppedInstancePrefix+instanceID))
 }
 
-// A rolling deploy on env19 listed one instance as running and stopped at
-// once, permanently, because its mirror outlived the claim that started it.
-// The key being replaced decides existence for exactly this reason.
-func TestOlderNodeClaim_LeavesNoStoppedInstanceBehind(t *testing.T) {
-	m, nc := newRecordManagerConn(t)
-	require.NoError(t, m.WriteStoppedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"}))
-
-	olderNodeClaim(t, nc, "i-1")
-
-	// The mirror is still there — nothing removed it, which is the point.
-	record, err := m.LoadInstanceRecord("i-1")
+// frozenKeyExists reports whether a key in one of the replaced key spaces is
+// still there.
+func frozenKeyExists(t *testing.T, nc *nats.Conn, bucket, key string) bool {
+	t.Helper()
+	js, err := jetstream.New(nc)
 	require.NoError(t, err)
-	require.NotNil(t, record, "the older node cannot have removed a key it does not know")
-
-	stopped, err := m.ListStoppedInstances()
+	kv, err := js.KeyValue(context.Background(), bucket)
 	require.NoError(t, err)
-	assert.Empty(t, stopped, "a started instance must not keep reading back as stopped")
-
-	got, err := m.LoadStoppedInstance("i-1")
-	require.NoError(t, err)
-	assert.Nil(t, got, "the mirror is not evidence the instance is still claimable")
+	_, err = kv.Get(context.Background(), key)
+	return err == nil
 }
 
-func TestOlderNodeClaim_DoesNotBlockASecondClaim(t *testing.T) {
-	m, nc := newRecordManagerConn(t)
-	require.NoError(t, m.WriteStoppedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"}))
-
-	olderNodeClaim(t, nc, "i-1")
-
-	// Exclusivity never moved off the key the older node claimed, so this node
-	// must lose rather than launch a second copy of a running instance.
-	_, err := m.ClaimStoppedInstance("i-1")
-	assert.ErrorIs(t, err, vm.ErrStoppedInstanceClaimed)
-}
-
-func TestUpdateStoppedInstance_MirrorsTheCommittedValue(t *testing.T) {
+func TestUpdateStoppedInstance_CommitsToTheRecord(t *testing.T) {
 	m := newRecordManager(t)
 	require.NoError(t, m.WriteStoppedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"}))
 
@@ -126,12 +50,12 @@ func TestUpdateStoppedInstance_MirrorsTheCommittedValue(t *testing.T) {
 	record, err := m.LoadInstanceRecord("i-1")
 	require.NoError(t, err)
 	require.NotNil(t, record)
-	assert.Equal(t, "node-2", record.Status.LastNode, "the mirror must carry what the mutation committed")
+	assert.Equal(t, "node-2", record.Status.LastNode, "the record must carry what the mutation committed")
 }
 
-// A mutation that finds nothing to mutate must not conjure a mirror: the
-// record key would then hold an instance the key it mirrors does not.
-func TestUpdateStoppedInstance_AbsentLeavesNoMirror(t *testing.T) {
+// A mutation that finds nothing to mutate must not create it. The record would
+// then hold an instance nothing ever stopped.
+func TestUpdateStoppedInstance_AbsentCreatesNothing(t *testing.T) {
 	m := newRecordManager(t)
 
 	_, err := m.UpdateStoppedInstance("i-gone", func(v *vm.VM) { v.LastNode = "node-2" })
@@ -142,25 +66,33 @@ func TestUpdateStoppedInstance_AbsentLeavesNoMirror(t *testing.T) {
 	assert.Nil(t, record)
 }
 
-func TestDeleteStoppedInstance_ClearsBothKeySpaces(t *testing.T) {
-	m := newRecordManager(t)
+// The delete drains the key the record replaced as well as the record. The
+// frozen space is kept so the cutover can be rolled back, but an instance that
+// is gone should not be recoverable from it, and draining on this path is what
+// stops it growing without bound.
+func TestDeleteStoppedInstance_DrainsTheFrozenKey(t *testing.T) {
+	m, nc := newRecordManagerConn(t)
 	require.NoError(t, m.WriteStoppedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"}))
+	seedFrozenKey(t, nc, daemon.InstanceStateBucket, daemon.StoppedInstancePrefix+"i-1")
 
 	require.NoError(t, m.DeleteStoppedInstance("i-1"))
 
 	record, err := m.LoadInstanceRecord("i-1")
 	require.NoError(t, err)
-	assert.Nil(t, record, "a delete that leaves the mirror behind resurrects the instance on the next read")
+	assert.Nil(t, record)
 
 	stopped, err := m.ListStoppedInstances()
 	require.NoError(t, err)
 	assert.Empty(t, stopped)
+
+	assert.False(t, frozenKeyExists(t, nc, daemon.InstanceStateBucket, daemon.StoppedInstancePrefix+"i-1"),
+		"the key the record replaced must not outlive the instance")
 }
 
-// The claim stays on the key it always used — two atomic deletes would be two
-// winners — and clears the mirror so the winner cannot leave the instance
-// readable as though it were still claimable.
-func TestClaimStoppedInstance_ClearsTheMirror(t *testing.T) {
+// Exclusivity is a compare-and-set on the record now rather than a delete of
+// it. One key holds an instance for its whole life, so a claim that deleted
+// would be deleting the instance it is about to start.
+func TestClaimStoppedInstance_OnlyOneCallerWins(t *testing.T) {
 	m := newRecordManager(t)
 	require.NoError(t, m.WriteStoppedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"}))
 
@@ -169,19 +101,58 @@ func TestClaimStoppedInstance_ClearsTheMirror(t *testing.T) {
 	require.NotNil(t, claimed)
 	assert.Equal(t, "t3.nano", claimed.InstanceType)
 
-	record, err := m.LoadInstanceRecord("i-1")
-	require.NoError(t, err)
-	assert.Nil(t, record, "a claimed instance must not survive at the record key")
-
-	stopped, err := m.ListStoppedInstances()
-	require.NoError(t, err)
-	assert.Empty(t, stopped)
-
 	_, err = m.ClaimStoppedInstance("i-1")
 	assert.ErrorIs(t, err, vm.ErrStoppedInstanceClaimed, "only one caller may win a claim")
 }
 
-func TestWriteTerminatedInstance_MirrorsOntoTheRecordKey(t *testing.T) {
+// What the winner changes is the intent, not the key. Clearing DesiredStopped
+// is what makes the record no longer claimable and no longer stopped, and it is
+// the same edit that stops it reading back as an instance nobody started.
+func TestClaimStoppedInstance_KeepsTheRecordAndClearsTheStopIntent(t *testing.T) {
+	m := newRecordManager(t)
+	require.NoError(t, m.WriteStoppedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"}))
+
+	_, err := m.ClaimStoppedInstance("i-1")
+	require.NoError(t, err)
+
+	record, err := m.LoadInstanceRecord("i-1")
+	require.NoError(t, err)
+	require.NotNil(t, record, "the claim moves the record between sets, it does not delete it")
+	assert.Equal(t, vm.DesiredRunning, record.Spec.DesiredState)
+
+	stopped, err := m.ListStoppedInstances()
+	require.NoError(t, err)
+	assert.Empty(t, stopped, "a claimed instance must not keep reading back as stopped")
+
+	got, err := m.LoadStoppedInstance("i-1")
+	require.NoError(t, err)
+	assert.Nil(t, got)
+}
+
+// LastNode is left alone deliberately. The claimant has not run the instance
+// yet, and the node that last did is the one that should recover it if the
+// launch that follows this claim never happens.
+func TestClaimStoppedInstance_LeavesOwnershipWithTheLastNodeToRunIt(t *testing.T) {
+	m := newRecordManager(t)
+	require.NoError(t, m.WriteStoppedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano", LastNode: "node-2"}))
+
+	_, err := m.ClaimStoppedInstance("i-1")
+	require.NoError(t, err)
+
+	record, err := m.LoadInstanceRecord("i-1")
+	require.NoError(t, err)
+	require.NotNil(t, record)
+	assert.Equal(t, "node-2", record.Status.LastNode)
+}
+
+func TestClaimStoppedInstance_AbsentIsNotClaimable(t *testing.T) {
+	m := newRecordManager(t)
+
+	_, err := m.ClaimStoppedInstance("i-never-existed")
+	assert.ErrorIs(t, err, vm.ErrStoppedInstanceClaimed)
+}
+
+func TestWriteTerminatedInstance_WritesTheRecord(t *testing.T) {
 	m := newRecordManager(t)
 
 	require.NoError(t, m.WriteTerminatedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"}))
@@ -192,7 +163,7 @@ func TestWriteTerminatedInstance_MirrorsOntoTheRecordKey(t *testing.T) {
 	assert.Equal(t, "t3.nano", record.Spec.InstanceType)
 }
 
-func TestUpdateTerminatedInstance_MirrorsTheCommittedValue(t *testing.T) {
+func TestUpdateTerminatedInstance_CommitsToTheRecord(t *testing.T) {
 	m := newRecordManager(t)
 	require.NoError(t, m.WriteTerminatedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"}))
 
@@ -207,29 +178,11 @@ func TestUpdateTerminatedInstance_MirrorsTheCommittedValue(t *testing.T) {
 	assert.Equal(t, map[string]string{"eni": "done"}, record.Status.Teardown)
 }
 
-func TestListTerminatedInstances_AnswersFromTheMirrorWhereThereIsOne(t *testing.T) {
-	m := newRecordManager(t)
-
-	require.NoError(t, m.WriteTerminatedInstance("i-old", &vm.VM{ID: "i-old", InstanceType: "t3.nano"}))
-	require.NoError(t, m.DeleteTerminatedInstanceRecord("i-old"))
-	require.NoError(t, m.WriteTerminatedInstance("i-both", &vm.VM{ID: "i-both", InstanceType: "t3.micro"}))
-
-	// A mirror with nothing behind it is not a terminated instance either.
-	require.NoError(t, m.WriteTerminatedInstanceRecord("i-orphan", testRecord("i-orphan")))
-
-	terminated, err := m.ListTerminatedInstances()
-
-	require.NoError(t, err)
-	ids := make([]string, 0, len(terminated))
-	for _, instance := range terminated {
-		ids = append(ids, instance.ID)
-	}
-	assert.ElementsMatch(t, []string{"i-old", "i-both"}, ids)
-}
-
-func TestDeleteTerminatedInstance_ClearsBothKeySpaces(t *testing.T) {
-	m := newRecordManager(t)
+func TestDeleteTerminatedInstance_DrainsTheFrozenKey(t *testing.T) {
+	m, nc := newRecordManagerConn(t)
+	require.NoError(t, m.InitTerminatedInstanceBucket())
 	require.NoError(t, m.WriteTerminatedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"}))
+	seedFrozenKey(t, nc, daemon.TerminatedInstanceBucket, daemon.TerminatedInstancePrefix+"i-1")
 
 	require.NoError(t, m.DeleteTerminatedInstance("i-1"))
 
@@ -240,18 +193,8 @@ func TestDeleteTerminatedInstance_ClearsBothKeySpaces(t *testing.T) {
 	terminated, err := m.ListTerminatedInstances()
 	require.NoError(t, err)
 	assert.Empty(t, terminated)
-}
 
-func TestLoadTerminatedInstance_FallsBackToTheKeyItReplaces(t *testing.T) {
-	m := newRecordManager(t)
-	require.NoError(t, m.WriteTerminatedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"}))
-	require.NoError(t, m.DeleteTerminatedInstanceRecord("i-1"))
-
-	got, err := m.LoadTerminatedInstance("i-1")
-
-	require.NoError(t, err)
-	require.NotNil(t, got)
-	assert.Equal(t, "t3.nano", got.InstanceType)
+	assert.False(t, frozenKeyExists(t, nc, daemon.TerminatedInstanceBucket, daemon.TerminatedInstancePrefix+"i-1"))
 }
 
 // seedLegacyBucket creates bucket at schema version 1 holding instance at the

spinifex/daemon/instance_running_set.go

@@ -0,0 +1,185 @@
+package daemon
+
+import (
+	"context"
+	"encoding/json"
+	"hash/fnv"
+	"log/slog"
+	"sync"
+
+	"github.com/mulgadc/spinifex/spinifex/vm"
+)
+
+// runningSetState tracks what this node last wrote to the per-resource key
+// space, so a change to one instance writes one record.
+//
+// Without it every state change would rewrite the whole running set one key at
+// a time, which is worse than the single blob it replaces rather than better:
+// splitting the blob exists to stop two instances on a node serialising against
+// each other.
+type runningSetState struct {
+	mu sync.Mutex
+	// seededFor is the node the digest describes, not a bool: the digest is
+	// per-node state and carries no node in it, so a manager asked to write a
+	// second node's set would otherwise measure it against the first node's and
+	// retire every record the first one owns.
+	seededFor string
+	digest    map[string]uint64
+}
+
+// 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 string, vms map[string]*vm.VM) {
+	if m.records == nil {
+		return
+	}
+
+	m.running.mu.Lock()
+	defer m.running.mu.Unlock()
+
+	if m.running.seededFor != nodeID {
+		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
+		}
+	}
+
+	for id, instance := range vms {
+		if instance == nil {
+			continue
+		}
+		// Ownership is stamped here rather than taken from the instance. This
+		// is the only place that knows whose running set is being written, and
+		// a record that reaches the key space without it belongs to no node —
+		// so nothing lists it and nothing retires it.
+		record := instance.Record()
+		record.Status.LastNode = nodeID
+
+		sum, err := recordDigest(record)
+		if err != nil {
+			slog.Error("Could not encode an instance record", "instanceId", id, "err", err)
+			continue
+		}
+		if current, ok := m.running.digest[id]; ok && current == sum {
+			continue
+		}
+
+		// A failed write forgets its digest as well as its record, so the next
+		// state write repeats the attempt rather than reading the failure as
+		// already written.
+		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)
+			continue
+		}
+		m.running.digest[id] = sum
+	}
+
+	for id := range m.running.digest {
+		if _, ok := vms[id]; ok {
+			continue
+		}
+		if m.retireRecord(nodeID, id) {
+			delete(m.running.digest, id)
+		}
+	}
+}
+
+// retireRecord drops the record of an instance that has left nodeID's running
+// set, reporting whether the question is settled.
+//
+// It is not settled by deleting unconditionally. One key now holds an instance
+// for its whole life, so the record of a departed instance is the same record
+// whoever took it over is now using. Two ways to leave are not deaths: an
+// operator stop, where WriteStoppedInstance has just rewritten that key, and a
+// move, where another node has claimed it. Both are handed over, not retired.
+func (m *JetStreamManager) retireRecord(nodeID, instanceID string) bool {
+	record, err := loadRecord(m.records, instanceRecordKey(instanceID))
+	if err != nil {
+		slog.Warn("Could not tell whether a departed instance is stopped; leaving its record in place",
+			"instanceId", instanceID, "err", err)
+		return false
+	}
+	if record == nil {
+		return true
+	}
+	if operatorStopped(record) {
+		return true
+	}
+	if record.Status.LastNode != nodeID {
+		slog.Debug("An instance that left this node is owned elsewhere; leaving its record alone",
+			"instanceId", instanceID, "owner", record.Status.LastNode)
+		return true
+	}
+
+	if err := m.records.Delete(context.Background(), instanceRecordKey(instanceID)); err != nil {
+		slog.Error("Could not remove the record of an instance that left this node",
+			"instanceId", instanceID, "err", err)
+		return false
+	}
+	return true
+}
+
+// seedRunningSet primes the digest map from the records already on the key
+// space for this node, so the first state write after boot rewrites only what
+// actually changed while the node was down — and so a record left behind by a
+// previous process is still a candidate for retirement.
+func (m *JetStreamManager) seedRunningSet(nodeID string) error {
+	existing, err := listRecords(m.records, InstanceRecordPrefix)
+	if err != nil {
+		return err
+	}
+
+	digest := make(map[string]uint64, len(existing))
+	for _, record := range existing {
+		if !runsOn(record, nodeID) {
+			continue
+		}
+		sum, err := recordDigest(record)
+		if err != nil {
+			return err
+		}
+		digest[record.Metadata.Name] = sum
+	}
+
+	m.running.digest = digest
+	m.running.seededFor = nodeID
+	slog.Debug("Seeded the instance record set", "node", nodeID, "records", len(digest))
+	return nil
+}
+
+// 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.
+func recordDigest(record *vm.InstanceRecord) (uint64, error) {
+	data, err := json.Marshal(record)
+	if err != nil {
+		return 0, err
+	}
+	h := fnv.New64a()
+	_, _ = h.Write(data)
+	return h.Sum64(), nil
+}
+
+// nodeRunningRecords returns the records nodeID owns, as instances.
+func (m *JetStreamManager) nodeRunningRecords(nodeID string) (map[string]*vm.VM, error) {
+	records, err := listRecords(m.records, InstanceRecordPrefix)
+	if err != nil {
+		return nil, err
+	}
+
+	vms := make(map[string]*vm.VM, len(records))
+	for _, record := range records {
+		if !runsOn(record, nodeID) {
+			continue
+		}
+		vms[record.Metadata.Name] = vm.VMFromRecord(record)
+	}
+	return vms, nil
+}

spinifex/daemon/instance_running_set_test.go

@@ -0,0 +1,462 @@
+package daemon_test
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+
+	"github.com/mulgadc/spinifex/spinifex/daemon"
+	"github.com/mulgadc/spinifex/spinifex/kvutil"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
+	"github.com/mulgadc/spinifex/spinifex/vm"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const localNode = "node-1"
+
+func runningVM(id, instanceType string) *vm.VM {
+	return &vm.VM{ID: id, InstanceType: instanceType, Status: vm.StateRunning, LastNode: localNode}
+}
+
+func runningSet(vms ...*vm.VM) map[string]*vm.VM {
+	set := make(map[string]*vm.VM, len(vms))
+	for _, v := range vms {
+		set[v.ID] = v
+	}
+	return set
+}
+
+// writeRunningSet does what persistState does: the presence marker, then the
+// records. Both, because a set written without a marker is a set no reader will
+// admit exists.
+func writeRunningSet(t *testing.T, m *daemon.JetStreamManager, vms ...*vm.VM) {
+	t.Helper()
+	require.NoError(t, m.WriteNodeMarker(localNode))
+	m.WriteRunningSet(localNode, runningSet(vms...))
+}
+
+// recordRevision reads the revision of a record straight from the bucket, so a
+// test can tell a write that did nothing from one that did not happen.
+func recordRevision(t *testing.T, nc *nats.Conn, instanceID string) uint64 {
+	t.Helper()
+	js, err := jetstream.New(nc)
+	require.NoError(t, err)
+	kv, err := js.KeyValue(context.Background(), daemon.InstanceStateBucket)
+	require.NoError(t, err)
+	entry, err := kv.Get(context.Background(), daemon.InstanceRecordPrefix+instanceID)
+	require.NoError(t, err)
+	return entry.Revision()
+}
+
+func TestWriteRunningSet_WritesARecordPerMember(t *testing.T) {
+	m := newRecordManager(t)
+
+	writeRunningSet(t, m, runningVM("i-1", "t3.nano"), runningVM("i-2", "t3.micro"))
+
+	records, err := m.ListInstanceRecords()
+	require.NoError(t, err)
+	byID := make(map[string]string, len(records))
+	for _, record := range records {
+		byID[record.Metadata.Name] = record.Spec.InstanceType
+	}
+	assert.Equal(t, map[string]string{"i-1": "t3.nano", "i-2": "t3.micro"}, byID,
+		"a set of two instances must become two records")
+}
+
+func TestLoadState_ReadsTheNodesRecords(t *testing.T) {
+	m := newRecordManager(t)
+	writeRunningSet(t, m, runningVM("i-1", "t3.nano"), runningVM("i-2", "t3.micro"))
+
+	loaded, found, err := m.LoadState(localNode)
+
+	require.NoError(t, err)
+	require.True(t, found)
+	require.Len(t, loaded, 2)
+	assert.Equal(t, "t3.nano", loaded["i-1"].InstanceType)
+	assert.Equal(t, "t3.micro", loaded["i-2"].InstanceType)
+}
+
+// The key space is shared by every node, so a listing is not a running set
+// until it is filtered by who owns each record.
+func TestLoadState_IgnoresAnotherNodesRecords(t *testing.T) {
+	m := newRecordManager(t)
+	writeRunningSet(t, m, runningVM("i-1", "t3.nano"))
+
+	elsewhere := &vm.VM{ID: "i-2", InstanceType: "t3.micro", Status: vm.StateRunning, LastNode: "node-2"}
+	require.NoError(t, m.WriteInstanceRecord("i-2", elsewhere.Record()))
+
+	loaded, found, err := m.LoadState(localNode)
+
+	require.NoError(t, err)
+	require.True(t, found)
+	assert.NotContains(t, loaded, "i-2")
+	assert.Len(t, loaded, 1)
+}
+
+// The distinction the presence marker exists for. Both cases scan no records,
+// and restore replaces a node's local state with what it reads here whenever it
+// believes the cluster has a record of that node — so reading "no instances" as
+// "no such node", or the reverse, is the difference between a node coming back
+// with its instances and coming back empty.
+func TestLoadState_FoundIsTheMarkerNotTheRecordCount(t *testing.T) {
+	m := newRecordManager(t)
+
+	loaded, found, err := m.LoadState(localNode)
+	require.NoError(t, err)
+	assert.False(t, found, "a node that has never written a marker is not in the cluster's record")
+	assert.Empty(t, loaded)
+
+	require.NoError(t, m.WriteNodeMarker(localNode))
+
+	loaded, found, err = m.LoadState(localNode)
+	require.NoError(t, err)
+	assert.True(t, found, "a node with a marker and no instances is a known, empty node")
+	assert.Empty(t, loaded)
+}
+
+// An operator-stopped instance belongs to the stopped set, and the running set
+// is the same keys filtered. Returning it here would relaunch it on restore.
+func TestLoadState_OmitsAnOperatorStoppedRecord(t *testing.T) {
+	m := newRecordManager(t)
+	writeRunningSet(t, m, runningVM("i-1", "t3.nano"))
+	require.NoError(t, m.WriteStoppedInstance("i-1", runningVM("i-1", "t3.nano")))
+
+	loaded, found, err := m.LoadState(localNode)
+
+	require.NoError(t, err)
+	require.True(t, found)
+	assert.NotContains(t, loaded, "i-1")
+}
+
+// The conversion's totality is guarded in the vm package; what this adds is
+// that the running-set path actually routes through it, on fields from both
+// halves of the record rather than the instance type alone.
+func TestWriteRunningSet_CarriesBothHalvesOfTheRecord(t *testing.T) {
+	m := newRecordManager(t)
+	instance := runningVM("i-1", "t3.nano")
+	instance.DesiredState = vm.DesiredRunning
+	instance.PublicIP = "10.0.0.9"
+	instance.ENIId = "eni-1"
+	instance.HostfwdPorts = []int{2222}
+
+	writeRunningSet(t, m, instance)
+
+	loaded, _, err := m.LoadState(localNode)
+	require.NoError(t, err)
+	require.Contains(t, loaded, "i-1")
+	got := loaded["i-1"]
+	assert.Equal(t, vm.DesiredRunning, got.DesiredState)
+	assert.Equal(t, []int{2222}, got.HostfwdPorts)
+	assert.Equal(t, "10.0.0.9", got.PublicIP)
+	assert.Equal(t, "eni-1", got.ENIId)
+	assert.Equal(t, localNode, got.LastNode)
+}
+
+// Ownership is stamped by the write rather than carried in from the caller:
+// this is the only place that knows whose running set is being written.
+func TestWriteRunningSet_StampsOwnership(t *testing.T) {
+	m := newRecordManager(t)
+	unowned := &vm.VM{ID: "i-1", InstanceType: "t3.nano", Status: vm.StateRunning}
+
+	require.NoError(t, m.WriteNodeMarker(localNode))
+	m.WriteRunningSet(localNode, runningSet(unowned))
+
+	record, err := m.LoadInstanceRecord("i-1")
+	require.NoError(t, err)
+	require.NotNil(t, record)
+	assert.Equal(t, localNode, record.Status.LastNode)
+}
+
+func TestWriteRunningSet_RetiresAMemberThatLeft(t *testing.T) {
+	m := newRecordManager(t)
+	writeRunningSet(t, m, runningVM("i-1", "t3.nano"), runningVM("i-2", "t3.micro"))
+
+	writeRunningSet(t, m, runningVM("i-1", "t3.nano"))
+
+	gone, err := m.LoadInstanceRecord("i-2")
+	require.NoError(t, err)
+	assert.Nil(t, gone, "a record left behind by a terminated instance never expires from this bucket")
+
+	kept, err := m.LoadInstanceRecord("i-1")
+	require.NoError(t, err)
+	assert.NotNil(t, kept)
+}
+
+// One key holds an instance for its whole life, so the record an instance
+// leaves the running set with is the same one the stopped set has just taken
+// over. An unconditional retire would delete the stopped instance it became.
+func TestWriteRunningSet_HandsOverAMemberThatStopped(t *testing.T) {
+	m := newRecordManager(t)
+	writeRunningSet(t, m, runningVM("i-1", "t3.nano"))
+
+	require.NoError(t, m.WriteStoppedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano", Status: vm.StateStopped}))
+	writeRunningSet(t, m)
+
+	record, err := m.LoadInstanceRecord("i-1")
+	require.NoError(t, err)
+	require.NotNil(t, record, "the stopped set owns this record now")
+
+	stopped, err := m.ListStoppedInstances()
+	require.NoError(t, err)
+	require.Len(t, stopped, 1)
+	assert.Equal(t, "i-1", stopped[0].ID)
+}
+
+// Rewriting every record on every state change would reintroduce, one key at a
+// time, the contention splitting the blob exists to remove.
+func TestWriteRunningSet_WritesOnlyWhatChanged(t *testing.T) {
+	m, nc := newRecordManagerConn(t)
+	writeRunningSet(t, m, runningVM("i-1", "t3.nano"), runningVM("i-2", "t3.micro"))
+	before1, before2 := recordRevision(t, nc, "i-1"), recordRevision(t, nc, "i-2")
+
+	writeRunningSet(t, m, runningVM("i-1", "t3.nano"), runningVM("i-2", "t3.micro"))
+
+	assert.Equal(t, before1, recordRevision(t, nc, "i-1"), "an unchanged instance must not be rewritten")
+	assert.Equal(t, before2, recordRevision(t, nc, "i-2"))
+
+	changed := runningVM("i-1", "t3.nano")
+	changed.PublicIP = "10.0.0.9"
+	writeRunningSet(t, m, changed, runningVM("i-2", "t3.micro"))
+
+	assert.Greater(t, recordRevision(t, nc, "i-1"), before1, "a changed instance must be rewritten")
+	assert.Equal(t, before2, recordRevision(t, nc, "i-2"), "and only that instance")
+}
+
+// A restarted process has no memory of what it last wrote, so it primes itself
+// from the bucket rather than rewriting every record to find out.
+func TestWriteRunningSet_SeedsFromWhatIsAlreadyThere(t *testing.T) {
+	first, nc := newRecordManagerConn(t)
+	writeRunningSet(t, first, runningVM("i-1", "t3.nano"))
+	before := recordRevision(t, nc, "i-1")
+
+	second, err := daemon.NewJetStreamManager(nc, 1)
+	require.NoError(t, err)
+	require.NoError(t, second.InitKVBucket())
+	writeRunningSet(t, second, runningVM("i-1", "t3.nano"))
+
+	assert.Equal(t, before, recordRevision(t, nc, "i-1"))
+}
+
+// Ownership is read from status.LastNode, the same field the GC uses. A node
+// must not retire a record for an instance that is running somewhere else.
+func TestWriteRunningSet_LeavesAnotherNodesRecordsAlone(t *testing.T) {
+	m := newRecordManager(t)
+	elsewhere := &vm.VM{ID: "i-elsewhere", InstanceType: "t3.nano", Status: vm.StateRunning, LastNode: "node-2"}
+	require.NoError(t, m.WriteInstanceRecord("i-elsewhere", elsewhere.Record()))
+
+	writeRunningSet(t, m, runningVM("i-1", "t3.nano"))
+
+	record, err := m.LoadInstanceRecord("i-elsewhere")
+	require.NoError(t, err)
+	assert.NotNil(t, record)
+}
+
+// seedNodeBlobBucket creates the instance-state bucket at the version before
+// the blob split, holding one node's running set as a single record.
+func seedNodeBlobBucket(t *testing.T, nc *nats.Conn, nodeID string, vms map[string]*vm.VM) jetstream.KeyValue {
+	t.Helper()
+	ctx := context.Background()
+
+	js, err := jetstream.New(nc)
+	require.NoError(t, err)
+	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: daemon.InstanceStateBucket, History: 1})
+	require.NoError(t, err)
+
+	data, err := json.Marshal(daemon.LocalState{SchemaVersion: daemon.LocalStateSchemaVersion, VMS: vms})
+	require.NoError(t, err)
+	_, err = kv.Put(ctx, daemon.InstanceStatePrefix+nodeID, data)
+	require.NoError(t, err)
+
+	require.NoError(t, kvutil.WriteVersion(ctx, kv, 2))
+	return kv
+}
+
+// readNodeBlob returns the running-set blob at the key the record space
+// replaced, or nil if it is gone.
+func readNodeBlob(t *testing.T, kv jetstream.KeyValue, nodeID string) *daemon.LocalState {
+	t.Helper()
+	entry, err := kv.Get(context.Background(), daemon.InstanceStatePrefix+nodeID)
+	if err != nil {
+		return nil
+	}
+	var blob daemon.LocalState
+	require.NoError(t, json.Unmarshal(entry.Value(), &blob))
+	return &blob
+}
+
+func TestInstanceStateMigration_SplitsTheNodeBlob(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	seedNodeBlobBucket(t, nc, localNode, runningSet(runningVM("i-1", "t3.nano"), runningVM("i-2", "t3.micro")))
+
+	m, err := daemon.NewJetStreamManager(nc, 1)
+	require.NoError(t, err)
+	require.NoError(t, m.InitKVBucket())
+
+	records, err := m.ListInstanceRecords()
+	require.NoError(t, err)
+	byID := make(map[string]string, len(records))
+	for _, record := range records {
+		byID[record.Metadata.Name] = record.Spec.InstanceType
+	}
+	assert.Equal(t, map[string]string{"i-1": "t3.nano", "i-2": "t3.micro"}, byID)
+}
+
+// The split copies the blob rather than consuming it, and nothing after the
+// cutover writes over it. It is what a node rolled back to the release before
+// the cutover reads to find out what it was running, so leaving it intact is
+// the whole of what makes the crossing reversible.
+func TestInstanceStateMigration_LeavesTheNodeBlobIntact(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	kv := seedNodeBlobBucket(t, nc, localNode, runningSet(runningVM("i-1", "t3.nano")))
+
+	m, err := daemon.NewJetStreamManager(nc, 1)
+	require.NoError(t, err)
+	require.NoError(t, m.InitKVBucket())
+	writeRunningSet(t, m, runningVM("i-1", "t3.nano"))
+
+	blob := readNodeBlob(t, kv, localNode)
+	require.NotNil(t, blob, "the blob the split copied from must survive the migration")
+	assert.Contains(t, blob.VMS, "i-1", "and must not be emptied by the marker that replaces it")
+}
+
+// The marker is a key of its own for that reason: written onto the blob it
+// would answer the presence question by destroying the answer to the other one.
+func TestInstanceStateMigration_SeedsAPresenceMarkerPerNode(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	kv := seedNodeBlobBucket(t, nc, localNode, runningSet(runningVM("i-1", "t3.nano")))
+
+	m, err := daemon.NewJetStreamManager(nc, 1)
+	require.NoError(t, err)
+	require.NoError(t, m.InitKVBucket())
+
+	_, err = kv.Get(context.Background(), daemon.NodePresencePrefix+localNode)
+	require.NoError(t, err, "a node that had a blob must be present without having written anything yet")
+
+	loaded, found, err := m.LoadState(localNode)
+	require.NoError(t, err)
+	assert.True(t, found, "so it can recover its instances from KV across the crossing")
+	assert.Contains(t, loaded, "i-1")
+}
+
+// Every node's set is copied, not just the one running the migration: it runs
+// once per bucket, and a node that is down would otherwise have no records
+// until it came back.
+func TestInstanceStateMigration_SplitsEveryNodesBlob(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	kv := seedNodeBlobBucket(t, nc, localNode, runningSet(runningVM("i-1", "t3.nano")))
+
+	other := &vm.VM{ID: "i-2", InstanceType: "t3.micro", Status: vm.StateRunning, LastNode: "node-2"}
+	data, err := json.Marshal(daemon.LocalState{
+		SchemaVersion: daemon.LocalStateSchemaVersion,
+		VMS:           map[string]*vm.VM{other.ID: other},
+	})
+	require.NoError(t, err)
+	_, err = kv.Put(context.Background(), daemon.InstanceStatePrefix+"node-2", data)
+	require.NoError(t, err)
+
+	m, err := daemon.NewJetStreamManager(nc, 1)
+	require.NoError(t, err)
+	require.NoError(t, m.InitKVBucket())
+
+	record, err := m.LoadInstanceRecord("i-2")
+	require.NoError(t, err)
+	require.NotNil(t, record)
+	assert.Equal(t, "node-2", record.Status.LastNode)
+}
+
+// unownedVM is what a running-set blob actually holds. The key it sat under
+// named the node, so nothing wrote the node onto the instance inside it, and a
+// record copied forward from one carries no owner at all.
+func unownedVM(id, instanceType string) *vm.VM {
+	return &vm.VM{ID: id, InstanceType: instanceType, Status: vm.StateRunning}
+}
+
+// Ownership is the other half of what the node.<id> key carried in its name,
+// and the half a reader cannot do without: after the cutover an unowned record
+// is one no node lists.
+func TestInstanceStateMigration_CarriesOwnershipOffTheBlobKey(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	seedNodeBlobBucket(t, nc, localNode, runningSet(unownedVM("i-1", "t3.nano")))
+
+	m, err := daemon.NewJetStreamManager(nc, 1)
+	require.NoError(t, err)
+	require.NoError(t, m.InitKVBucket())
+
+	record, err := m.LoadInstanceRecord("i-1")
+	require.NoError(t, err)
+	require.NotNil(t, record)
+	assert.Equal(t, localNode, record.Status.LastNode, "the blob key is the only thing that knew this")
+}
+
+// The case that makes the stamp necessary rather than tidy. The marker says the
+// cluster knows the node, so restore adopts what comes back with it — and an
+// unowned running set comes back empty, which it would then adopt over the
+// instances the node is actually running.
+func TestInstanceStateMigration_ANodeRecoversAnUnownedRunningSet(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	seedNodeBlobBucket(t, nc, localNode, runningSet(unownedVM("i-1", "t3.nano"), unownedVM("i-2", "t3.micro")))
+
+	m, err := daemon.NewJetStreamManager(nc, 1)
+	require.NoError(t, err)
+	require.NoError(t, m.InitKVBucket())
+
+	loaded, found, err := m.LoadState(localNode)
+
+	require.NoError(t, err)
+	require.True(t, found)
+	assert.Len(t, loaded, 2, "an adopted empty set here is every instance on the node dropped")
+}
+
+// An owner already on the record is the newer fact. The blob it would be
+// overwritten from stopped being written when the cutover landed.
+func TestInstanceStateMigration_LeavesAnOwnedRecordAlone(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	moved := &vm.VM{ID: "i-1", InstanceType: "t3.nano", Status: vm.StateRunning, LastNode: "node-2"}
+	seedNodeBlobBucket(t, nc, localNode, runningSet(moved))
+
+	m, err := daemon.NewJetStreamManager(nc, 1)
+	require.NoError(t, err)
+	require.NoError(t, m.InitKVBucket())
+
+	record, err := m.LoadInstanceRecord("i-1")
+	require.NoError(t, err)
+	require.NotNil(t, record)
+	assert.Equal(t, "node-2", record.Status.LastNode)
+}
+
+func TestInstanceStateMigration_OwnershipStampIsSafeToRunTwice(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	kv := seedNodeBlobBucket(t, nc, localNode, runningSet(unownedVM("i-1", "t3.nano")))
+
+	first, err := daemon.NewJetStreamManager(nc, 1)
+	require.NoError(t, err)
+	require.NoError(t, first.InitKVBucket())
+	before := recordRevision(t, nc, "i-1")
+
+	require.NoError(t, kvutil.WriteVersion(context.Background(), kv, 4))
+	second, err := daemon.NewJetStreamManager(nc, 1)
+	require.NoError(t, err)
+	require.NoError(t, second.InitKVBucket())
+
+	assert.Equal(t, before, recordRevision(t, nc, "i-1"), "a record that already names its owner must not be rewritten")
+}
+
+func TestInstanceStateMigration_BlobSplitIsSafeToRunTwice(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	kv := seedNodeBlobBucket(t, nc, localNode, runningSet(runningVM("i-1", "t3.nano")))
+
+	first, err := daemon.NewJetStreamManager(nc, 1)
+	require.NoError(t, err)
+	require.NoError(t, first.InitKVBucket())
+	before := recordRevision(t, nc, "i-1")
+
+	require.NoError(t, kvutil.WriteVersion(context.Background(), kv, 2))
+	second, err := daemon.NewJetStreamManager(nc, 1)
+	require.NoError(t, err)
+	require.NoError(t, second.InitKVBucket())
+
+	assert.Equal(t, before, recordRevision(t, nc, "i-1"), "a second run must not overwrite what the first copied")
+}

spinifex/daemon/jetstream.go

@@ -23,8 +23,17 @@ const (
 	InstanceStateBucket = "spinifex-instance-state"
 	// ClusterStateBucket is the name of the KV bucket for cluster state (heartbeats, shutdown markers, service maps).
 	ClusterStateBucket = "spinifex-cluster-state"
-	// InstanceStatePrefix is the key prefix for per-node instance state entries.
+	// InstanceStatePrefix is the key prefix for the per-node running-set blobs
+	// the record space replaces. Nothing writes it after the cutover; the
+	// migration reads it and a rolled-back node still finds what it left.
 	InstanceStatePrefix = "node."
+	// NodePresencePrefix is the key prefix for a node's presence marker.
+	//
+	// Deliberately not InstanceStatePrefix. The marker holds no instances, so
+	// writing it there would empty the blob it shares a key with, and a node
+	// rolled back to the release before the cutover reads that blob to find out
+	// what it was running. Frozen means not written, not written empty.
+	NodePresencePrefix = "nodepresence."
 	// StoppedInstancePrefix is the key prefix for stopped instances in shared KV.
 	StoppedInstancePrefix = "instance."
 	// TerminatedInstanceBucket is the name of the KV bucket for terminated instances (auto-expiry via TTL).
@@ -34,9 +43,13 @@ const (
 
 	// Schema versions for daemon KV buckets. Both instance buckets copied their
 	// per-instance keys onto the record space at 2; instance-state took the node
-	// blobs at 3; both re-keyed that space from "i/" to "i." last. See
+	// blobs at 3; both re-keyed that space from "i/" to "i." next. See
 	// instance_records_migrate.go.
-	InstanceStateBucketVersion      = 4
+	//
+	// instance-state 5 is the cutover: the record space became the only copy.
+	// The bump is also the compatibility stamp — a build that predates it stops
+	// on a bucket stamped 5 rather than reading the keys it no longer owns.
+	InstanceStateBucketVersion      = 5
 	ClusterStateBucketVersion       = 1
 	TerminatedInstanceBucketVersion = 3
 )
@@ -59,18 +72,18 @@ type KVSyncObserver interface {
 type JetStreamManager struct {
 	js        jetstream.JetStream
 	stateB    *kvstore.Bucket            // spinifex-instance-state
-	nodeState *kvstore.Store[LocalState] // node.<id> records
-	stopped   *kvstore.Store[vm.VM]      // instance.<id> records
+	nodeState *kvstore.Store[LocalState] // nodepresence.<id> markers
+	stopped   *kvstore.Store[vm.VM]      // instance.<id>, frozen: drained, never written
 	term      *kvstore.Store[vm.VM]      // spinifex-terminated-instances
 	// The per-resource key space the three views above are moving onto. Each
 	// is a third view over a bucket one of them already holds, so the two
 	// spaces share a handle and a recovery driven through either repairs both.
-	records     *kvstore.Store[vm.InstanceRecord] // i/<id>, instance-state bucket
-	termRecords *kvstore.Store[vm.InstanceRecord] // i/<id>, terminated bucket
+	records     *kvstore.Store[vm.InstanceRecord] // i.<id>, instance-state bucket
+	termRecords *kvstore.Store[vm.InstanceRecord] // i.<id>, terminated bucket
 	clusterKV   jetstream.KeyValue                // spinifex-cluster-state
 	replicas    int
 	obs         KVSyncObserver
-	mirror      mirrorState
+	running     runningSetState
 }
 
 // checkNodeStateVersion reports whether a node record read from KV is one this
@@ -406,36 +419,35 @@ func (m *JetStreamManager) WriteServiceManifest(nodeID string, services []string
 	return err
 }
 
-// WriteState writes the instance state to the KV store for the given node and
-// mirrors it onto the per-resource key space.
-// vms must be a snapshot owned by the caller — JetStreamManager does not lock.
+// WriteNodeMarker records that this node has cluster state at all, without
+// recording what it is.
 //
-// It mirrors even though persistState is the path production takes, because a
-// blob writer that skips the mirror leaves a record staler than the blob it
-// belongs to, and reads answer from the record.
-func (m *JetStreamManager) WriteState(nodeID string, vms map[string]*vm.VM) error {
+// Reads need it because "this node has no instances" and "there is no cluster
+// record of this node" are different answers and only one of them may replace a
+// node's local state. Scanning the records cannot tell them apart — both scan
+// empty — and restore adopts the cluster's set wholesale when it believes there
+// is one, so conflating them drops every instance on a node whose records are
+// briefly unreadable.
+//
+// The marker carries no instances, so a state change no longer rewrites the
+// node's whole set. That is the cost the split exists to remove; keeping an
+// empty envelope keeps the answer and not the cost.
+func (m *JetStreamManager) WriteNodeMarker(nodeID string) error {
 	if m.nodeState == nil {
 		return errors.New("KV bucket not initialized")
 	}
 
-	key := InstanceStatePrefix + nodeID
-	record := LocalState{SchemaVersion: LocalStateSchemaVersion, VMS: vms}
-	if err := m.nodeState.Set(context.Background(), key, &record); err != nil {
-		return err
-	}
-	m.MirrorRunningSet(nodeID, vms)
-
-	slog.Debug("Wrote state to JetStream KV", "key", key, "instances", len(vms))
-	return nil
+	key := NodePresencePrefix + nodeID
+	marker := LocalState{SchemaVersion: LocalStateSchemaVersion}
+	return m.nodeState.Set(context.Background(), key, &marker)
 }
 
-// WriteStateBytesBestEffort attempts to push pre-marshalled 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; hot
-// paths marshal under a short-lived lock and commit lock-free.
-func (m *JetStreamManager) WriteStateBytesBestEffort(nodeID string, jsonData []byte, timeout time.Duration) {
-	if m.stateB == nil {
+// WriteNodeMarkerBestEffort writes this node's presence marker with a deadline.
+// On timeout or error it logs and returns — never blocks the caller past
+// timeout and never returns an error. The local state file is the source of
+// truth; KV is the cluster's view of it, and the next state change retries.
+func (m *JetStreamManager) WriteNodeMarkerBestEffort(nodeID string, timeout time.Duration) {
+	if m.nodeState == nil {
 		slog.Debug("KV bucket not initialized, skipping cluster sync", "node", nodeID)
 		return
 	}
@@ -443,15 +455,9 @@ func (m *JetStreamManager) WriteStateBytesBestEffort(nodeID string, jsonData []b
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
 
-	key := InstanceStatePrefix + nodeID
-	// Pre-marshalled bytes, so this goes to the raw handle rather than through
-	// a typed view. Best-effort by contract: a failure is logged, not recovered
-	// from here, and the next typed write repairs the bucket for both.
-	kv, err := m.stateB.KV(ctx)
-	if err == nil {
-		_, err = kv.Put(ctx, key, jsonData)
-	}
-	if err != nil {
+	key := NodePresencePrefix + nodeID
+	marker := LocalState{SchemaVersion: LocalStateSchemaVersion}
+	if err := m.nodeState.Set(ctx, key, &marker); err != nil {
 		if m.obs != nil {
 			m.obs.RecordKVSyncFailure(InstanceStateBucket, err)
 		}
@@ -465,31 +471,22 @@ func (m *JetStreamManager) WriteStateBytesBestEffort(nodeID string, jsonData []b
 	if m.obs != nil {
 		m.obs.RecordKVSyncSuccess(InstanceStateBucket)
 	}
-	slog.Debug("Wrote state to KV (best-effort)", "key", key, "bytes", len(jsonData))
+	slog.Debug("Wrote node marker to KV (best-effort)", "key", key)
 }
 
-// marshalInstanceState produces the JSON wire form of vms, for the best-effort
-// path that marshals under the caller's lock and commits lock-free. The KV
-// record and the local state file are the same envelope, so this is the same
-// marshaller: the two cannot drift.
-func marshalInstanceState(vms map[string]*vm.VM) ([]byte, error) {
-	return MarshalLocalState(vms)
-}
-
-// LoadState loads the instance state from the KV store for the given node,
-// answering each member from its i/<id> mirror where there is one.
+// LoadState loads the instances the given node owns, from one record each.
 //
-// The bool reports whether a record exists at all. A node that has never
-// written state and a node whose record was lost both read as empty, and a
-// caller that would drop instances on the strength of that must tell them
-// apart — so it stays the blob's answer, not the mirrors'.
+// The bool reports whether the cluster has a record of this node at all, and it
+// is the marker's answer rather than the records'. A node with no instances and
+// a node whose records could not be read both scan empty, and only the first of
+// them may replace what the node has locally.
 func (m *JetStreamManager) LoadState(nodeID string) (map[string]*vm.VM, bool, error) {
 	if m.nodeState == nil {
 		return nil, false, errors.New("KV bucket not initialized")
 	}
 
-	key := InstanceStatePrefix + nodeID
-	state, _, err := m.nodeState.Get(context.Background(), key)
+	key := NodePresencePrefix + nodeID
+	marker, _, err := m.nodeState.Get(context.Background(), key)
 	if err != nil {
 		if errors.Is(err, kvstore.ErrNotFound) {
 			slog.Debug("No existing state in JetStream KV", "key", key)
@@ -497,18 +494,17 @@ func (m *JetStreamManager) LoadState(nodeID string) (map[string]*vm.VM, bool, er
 		}
 		return nil, false, err
 	}
-	if err := checkNodeStateVersion(key, state.SchemaVersion); err != nil {
-		return nil, false, err
-	}
-	if state.VMS == nil {
-		state.VMS = make(map[string]*vm.VM)
-	}
-	if err := m.answerFromMirrors(state.VMS); err != nil {
+	if err := checkNodeStateVersion(key, marker.SchemaVersion); err != nil {
 		return nil, false, err
 	}
 
-	slog.Debug("Loaded state from JetStream KV", "key", key, "instances", len(state.VMS))
-	return state.VMS, true, nil
+	vms, err := m.nodeRunningRecords(nodeID)
+	if err != nil {
+		return nil, false, err
+	}
+
+	slog.Debug("Loaded state from JetStream KV", "node", nodeID, "instances", len(vms))
+	return vms, true, nil
 }
 
 // DeleteState removes the instance state from the KV store for the given node.
@@ -517,7 +513,7 @@ func (m *JetStreamManager) DeleteState(nodeID string) error {
 		return errors.New("KV bucket not initialized")
 	}
 
-	key := InstanceStatePrefix + nodeID
+	key := NodePresencePrefix + nodeID
 	if err := m.nodeState.Delete(context.Background(), key); err != nil {
 		return err
 	}
@@ -590,54 +586,53 @@ func (m *JetStreamManager) UpdateReplicas(newReplicas int) error {
 // callers that read-modify-write a stopped instance (e.g. tag/attribute
 // mutations) build on.
 //
-// The write lands at instance.<id> first and is then mirrored onto i/<id>.
-// That order is not arbitrary: a node that predates the per-resource key space
-// serialises its own writes against instance.<id>, so it stays the CAS anchor
-// until nothing is left that reads it.
+// The instance keeps the one key it has held since it was launched. Stopping
+// it rewrites that key rather than moving it, so nothing can observe the
+// instance at neither key or at both.
 func (m *JetStreamManager) WriteStoppedInstance(instanceID string, instance *vm.VM) error {
-	if m.stopped == nil {
+	if m.records == nil {
 		return errors.New("KV bucket not initialized")
 	}
 
-	key := StoppedInstancePrefix + instanceID
-	// Replace rather than Set: this overwrites the record wholesale, and doing
-	// it under CAS is what stops a racing update landing out of order.
-	if err := m.stopped.Replace(context.Background(), key, instance); err != nil {
-		return err
-	}
-	if err := mirrorRecord(m.records, instanceID, instance); err != nil {
+	// The write stamps what it means rather than trusting the caller to have.
+	// Membership is a predicate over the record now, so an instance written
+	// here without both fields set would be stored and then not found — and the
+	// caller that forgot is not the one that discovers it.
+	record := instance.Record()
+	record.Status.Status = vm.StateStopped
+	record.Spec.DesiredState = vm.DesiredStopped
+	if err := m.records.Replace(context.Background(), instanceRecordKey(instanceID), record); err != nil {
 		return err
 	}
 
-	slog.Debug("Wrote stopped instance to JetStream KV", "key", key, "instanceId", instanceID)
+	slog.Debug("Wrote stopped instance to JetStream KV", "instanceId", instanceID)
 	return nil
 }
 
 // LoadStoppedInstance loads a stopped instance from the shared KV store.
-// instance.<id> decides whether the instance exists and i/<id> supplies its
-// value, for the reason loadPreferringRecord gives.
-// Returns nil, nil if the instance does not exist.
+// Returns nil, nil when no record exists, and also when one does but holds an
+// instance that is not stopped: the key is shared with this node's running set,
+// so the record has to be asked what it is.
 func (m *JetStreamManager) LoadStoppedInstance(instanceID string) (*vm.VM, error) {
-	if m.stopped == nil {
+	if m.records == nil {
 		return nil, errors.New("KV bucket not initialized")
 	}
-	return loadPreferringRecord(m.records, m.stopped, StoppedInstancePrefix+instanceID, instanceID)
+	return loadInstance(m.records, instanceID, operatorStopped)
 }
 
-// DeleteStoppedInstance removes a stopped instance from both key spaces in the
-// shared KV store. It is idempotent — deleting a non-existent key is not an
-// error.
+// DeleteStoppedInstance removes a stopped instance from the shared KV store. It
+// is idempotent — deleting a non-existent key is not an error.
 func (m *JetStreamManager) DeleteStoppedInstance(instanceID string) error {
-	if m.stopped == nil {
+	if m.records == nil {
 		return errors.New("KV bucket not initialized")
 	}
 
-	key := StoppedInstancePrefix + instanceID
-	if err := deleteBothRecords(m.records, m.stopped, key, instanceID); err != nil {
+	if err := m.records.Delete(context.Background(), instanceRecordKey(instanceID)); err != nil {
 		return err
 	}
+	m.drainFrozenKey(m.stopped, StoppedInstancePrefix+instanceID)
 
-	slog.Debug("Deleted stopped instance from JetStream KV", "key", key)
+	slog.Debug("Deleted stopped instance from JetStream KV", "instanceId", instanceID)
 	return nil
 }
 
@@ -651,32 +646,49 @@ func (m *JetStreamManager) DeleteStoppedInstance(instanceID string) error {
 // retry after the instance was already claimed) gets vm.ErrStoppedInstanceClaimed
 // instead of a VM and must not proceed to allocate resources or launch qemu.
 //
-// The claim cannot be dual-homed onto i/<id>: exclusivity here is one atomic
-// delete, and two of them are two winners. It stays on the key a node that
-// predates the per-resource space also claims, and the mirror is cleared
-// first — a win can then never leave the instance readable at i/<id> as though
-// it were still claimable, and a loss costs only a mirror the next write
-// restores.
+// Exclusivity is a compare-and-set on the record rather than a delete of it.
+// A delete was safe while the key held nothing but a stopped instance; the key
+// holds the instance for its whole life now, so deleting to claim it is
+// deleting the instance, and a claimant that fails after winning takes the
+// instance with it.
+//
+// The winner clears DesiredStopped, which is what makes the record no longer
+// claimable. A loser's CAS fails on the revision, re-reads, finds an instance
+// nobody asked to be stopped, and reports the claim lost — so exactly one
+// caller can win, which is the property the delete had.
+//
+// LastNode is left alone. The claimant has not run the instance yet, and the
+// node that last did is the one that should recover it if this launch never
+// happens.
 func (m *JetStreamManager) ClaimStoppedInstance(instanceID string) (*vm.VM, error) {
-	if m.stopped == nil {
+	if m.records == nil {
 		return nil, errors.New("KV bucket not initialized")
 	}
 
-	if err := m.records.Delete(context.Background(), instanceRecordKey(instanceID)); err != nil {
-		return nil, err
-	}
-
-	key := StoppedInstancePrefix + instanceID
-	instance, notFound, err := m.stopped.Claim(context.Background(), key)
+	key := instanceRecordKey(instanceID)
+	record, rev, err := m.records.Get(context.Background(), key)
 	if err != nil {
+		if errors.Is(err, kvstore.ErrNotFound) {
+			return nil, vm.ErrStoppedInstanceClaimed
+		}
 		return nil, err
 	}
-	if notFound {
+	if !operatorStopped(record) {
 		return nil, vm.ErrStoppedInstanceClaimed
 	}
 
-	slog.Debug("Claimed stopped instance from JetStream KV", "key", key, "instanceId", instanceID)
-	return instance, nil
+	claimed := *record
+	claimed.Spec.DesiredState = vm.DesiredRunning
+	if err := m.records.CompareAndSet(context.Background(), key, &claimed, rev); err != nil {
+		if errors.Is(err, kvstore.ErrConflict) {
+			return nil, vm.ErrStoppedInstanceClaimed
+		}
+		return nil, err
+	}
+	m.drainFrozenKey(m.stopped, StoppedInstancePrefix+instanceID)
+
+	slog.Debug("Claimed stopped instance from JetStream KV", "instanceId", instanceID)
+	return vm.VMFromRecord(&claimed), nil
 }
 
 // UpdateStoppedInstance atomically applies mutate to the current KV-stored
@@ -688,24 +700,58 @@ func (m *JetStreamManager) ClaimStoppedInstance(instanceID string) (*vm.VM, erro
 // stopped instance so they cannot race a claim into recreating a stale
 // record. Returns jetstream.ErrKeyNotFound if no record exists.
 //
-// The mutation is applied to instance.<id> and the committed result mirrored
-// onto i/<id>, rather than each key being mutated in turn: one CAS anchor and
-// a copy of what it committed cannot diverge, two CAS loops over two keys can.
+// The mutation runs against the record, on the one key the instance has, so
+// there is a single CAS anchor rather than two that can diverge.
 func (m *JetStreamManager) UpdateStoppedInstance(instanceID string, mutate func(*vm.VM)) (*vm.VM, error) {
-	if m.stopped == nil {
+	if m.records == nil {
 		return nil, errors.New("KV bucket not initialized")
 	}
+	return mutateInstance(m.records, instanceID, operatorStopped, mutate)
+}
 
-	updated, err := mutateRecord(m.stopped, StoppedInstancePrefix+instanceID, mutate)
+// mutateInstance applies mutate to the instance a record holds, under the
+// store's CAS loop, and returns what committed. The callers speak vm.VM and the
+// store speaks records, so the conversion happens inside the loop rather than
+// around it — a retry has to re-read and re-convert, not replay a stale copy.
+//
+// want is checked inside the loop, and a record it rejects is reported absent.
+// The key outlives the set it is in now, so "still stopped" has to be re-tested
+// on the value each attempt reads: a claim that landed between a caller's own
+// load and this write leaves the key there, and without the test the caller's
+// mutation would land on an instance somebody else has already started.
+func mutateInstance(store *kvstore.Store[vm.InstanceRecord], instanceID string,
+	want func(*vm.InstanceRecord) bool, mutate func(*vm.VM)) (*vm.VM, error) {
+	var updated *vm.VM
+	err := store.Mutate(context.Background(), instanceRecordKey(instanceID),
+		func(record *vm.InstanceRecord) (bool, error) {
+			if want != nil && !want(record) {
+				return false, kvstore.ErrNotFound
+			}
+			instance := vm.VMFromRecord(record)
+			mutate(instance)
+			*record = *instance.Record()
+			updated = instance
+			return true, nil
+		})
 	if err != nil {
 		return nil, err
 	}
-	if err := mirrorRecord(m.records, instanceID, updated); err != nil {
-		return nil, err
-	}
 	return updated, nil
 }
 
+// drainFrozenKey removes the key the record replaced. The old key spaces are
+// frozen rather than deleted at the cutover, so the crossing can be rolled
+// back — but an instance that is gone should not be recoverable, and draining
+// them on the delete path is what stops the frozen space growing forever.
+func (m *JetStreamManager) drainFrozenKey(store *kvstore.Store[vm.VM], key string) {
+	if store == nil {
+		return
+	}
+	if err := store.Delete(context.Background(), key); err != nil {
+		slog.Debug("Could not drain a frozen key", "key", key, "err", err)
+	}
+}
+
 // mutateRecord applies mutate under the store's CAS loop and returns the
 // committed record. An absent key surfaces as kvstore.ErrNotFound, which is what
 // both StateStore interfaces are written against.
@@ -722,13 +768,14 @@ func mutateRecord[T any](store *kvstore.Store[T], key string, mutate func(*T)) (
 	return updated, nil
 }
 
-// ListStoppedInstances returns all stopped instances from the shared KV store,
-// answering each from its i/<id> mirror where there is one.
+// ListStoppedInstances returns the stopped instances in the shared KV store.
+// The key space also holds every running instance, so membership is the
+// record's answer rather than the prefix's.
 func (m *JetStreamManager) ListStoppedInstances() ([]*vm.VM, error) {
-	if m.stopped == nil {
+	if m.records == nil {
 		return nil, errors.New("KV bucket not initialized")
 	}
-	return listPreferringRecords(m.records, m.stopped, StoppedInstancePrefix)
+	return listInstances(m.records, operatorStopped)
 }
 
 // listRecords returns every record under prefix. Unlike the raw scan it
@@ -759,19 +806,15 @@ func listRecords[T any](store *kvstore.Store[T], prefix string) ([]*T, error) {
 // record instead of replacing it wholesale should use
 // UpdateTerminatedInstance.
 func (m *JetStreamManager) WriteTerminatedInstance(instanceID string, instance *vm.VM) error {
-	if m.term == nil {
+	if m.termRecords == nil {
 		return errors.New("terminated instance KV bucket not initialized")
 	}
 
-	key := TerminatedInstancePrefix + instanceID
-	if err := m.term.Replace(context.Background(), key, instance); err != nil {
-		return err
-	}
-	if err := mirrorRecord(m.termRecords, instanceID, instance); err != nil {
+	if err := writeRecord(m.termRecords, instanceID, instance); err != nil {
 		return err
 	}
 
-	slog.Debug("Wrote terminated instance to JetStream KV", "key", key, "instanceId", instanceID)
+	slog.Debug("Wrote terminated instance to JetStream KV", "instanceId", instanceID)
 	return nil
 }
 
@@ -782,51 +825,44 @@ func (m *JetStreamManager) WriteTerminatedInstance(instanceID string, instance *
 // progress without clobbering marks written by a concurrent update to the
 // same record. Returns jetstream.ErrKeyNotFound if no record exists yet.
 func (m *JetStreamManager) UpdateTerminatedInstance(instanceID string, mutate func(*vm.VM)) (*vm.VM, error) {
-	if m.term == nil {
+	if m.termRecords == nil {
 		return nil, errors.New("terminated instance KV bucket not initialized")
 	}
-
-	updated, err := mutateRecord(m.term, TerminatedInstancePrefix+instanceID, mutate)
-	if err != nil {
-		return nil, err
-	}
-	if err := mirrorRecord(m.termRecords, instanceID, updated); err != nil {
-		return nil, err
-	}
-	return updated, nil
+	// No predicate: the terminated bucket is its own key space, so being in it
+	// is the whole of the membership test.
+	return mutateInstance(m.termRecords, instanceID, nil, mutate)
 }
 
 // ListTerminatedInstances returns all terminated instances from the terminated
-// KV bucket, answering each from its i/<id> mirror where there is one.
+// KV bucket. Its key space holds nothing else, so every record is one.
 func (m *JetStreamManager) ListTerminatedInstances() ([]*vm.VM, error) {
-	if m.term == nil {
+	if m.termRecords == nil {
 		return nil, errors.New("terminated instance KV bucket not initialized")
 	}
-	return listPreferringRecords(m.termRecords, m.term, TerminatedInstancePrefix)
+	return listInstances(m.termRecords, nil)
 }
 
-// DeleteTerminatedInstance removes a terminated instance from both key spaces
-// in the terminated KV bucket.
+// DeleteTerminatedInstance removes a terminated instance from the terminated KV
+// bucket.
 func (m *JetStreamManager) DeleteTerminatedInstance(instanceID string) error {
-	if m.term == nil {
+	if m.termRecords == nil {
 		return errors.New("terminated instance KV bucket not initialized")
 	}
 
-	key := TerminatedInstancePrefix + instanceID
-	if err := deleteBothRecords(m.termRecords, m.term, key, instanceID); err != nil {
+	if err := m.termRecords.Delete(context.Background(), instanceRecordKey(instanceID)); err != nil {
 		return err
 	}
+	m.drainFrozenKey(m.term, TerminatedInstancePrefix+instanceID)
 
-	slog.Debug("Deleted terminated instance from JetStream KV", "key", key)
+	slog.Debug("Deleted terminated instance from JetStream KV", "instanceId", instanceID)
 	return nil
 }
 
 // LoadTerminatedInstance loads a single terminated instance from the terminated
-// KV bucket. terminated.<id> decides existence, i/<id> supplies the value.
-// Returns nil, nil if the instance does not exist.
+// KV bucket. Returns nil, nil if the instance does not exist.
 func (m *JetStreamManager) LoadTerminatedInstance(instanceID string) (*vm.VM, error) {
-	if m.term == nil {
+	if m.termRecords == nil {
 		return nil, errors.New("terminated instance KV bucket not initialized")
 	}
-	return loadPreferringRecord(m.termRecords, m.term, TerminatedInstancePrefix+instanceID, instanceID)
+	return loadInstance(m.termRecords, instanceID, nil)
 }

spinifex/daemon/jetstream_test.go

@@ -3,6 +3,7 @@ package daemon
 import (
 	"encoding/json"
 	"errors"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"sync"
 	"testing"
 	"time"
@@ -16,6 +17,19 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+// writeNodeState does what persistState does: the presence marker, then one
+// record per running instance. Two calls rather than one because the marker
+// answers whether the cluster knows this node and the records answer what it
+// is running, and only the first of those may be inferred from the second.
+func writeNodeState(t *testing.T, m *JetStreamManager, nodeID string, vms map[string]*vm.VM) error {
+	t.Helper()
+	if err := m.WriteNodeMarker(nodeID); err != nil {
+		return err
+	}
+	m.WriteRunningSet(nodeID, vms)
+	return nil
+}
+
 // TestJetStreamManager_WriteAndLoadState tests round-trip write and load of instance state.
 func TestJetStreamManager_WriteAndLoadState(t *testing.T) {
 	natsURL := sharedJSNATSURL
@@ -49,7 +63,7 @@ func TestJetStreamManager_WriteAndLoadState(t *testing.T) {
 	}
 
 	// Write state
-	err = jsm.WriteState(testNodeID, testInstances)
+	err = writeNodeState(t, jsm, testNodeID, testInstances)
 	require.NoError(t, err, "Failed to write state")
 
 	// Load state
@@ -133,7 +147,7 @@ func TestJetStreamManager_BucketReconnection(t *testing.T) {
 			Status: vm.StateRunning,
 		},
 	}
-	err = jsm1.WriteState("persist-node", testInstances)
+	err = writeNodeState(t, jsm1, "persist-node", testInstances)
 	require.NoError(t, err)
 
 	nc1.Close()
@@ -180,7 +194,7 @@ func TestJetStreamManager_DeleteState(t *testing.T) {
 			Status: vm.StateRunning,
 		},
 	}
-	err = jsm.WriteState(testNodeID, testInstances)
+	err = writeNodeState(t, jsm, testNodeID, testInstances)
 	require.NoError(t, err)
 
 	// Verify state exists
@@ -242,7 +256,7 @@ func TestJetStreamManager_WriteState_UpdateExisting(t *testing.T) {
 			Status: vm.StateRunning,
 		},
 	}
-	err = jsm.WriteState(testNodeID, initialInstances)
+	err = writeNodeState(t, jsm, testNodeID, initialInstances)
 	require.NoError(t, err)
 
 	// Update state with different instances
@@ -256,7 +270,7 @@ func TestJetStreamManager_WriteState_UpdateExisting(t *testing.T) {
 			Status: vm.StateRunning,
 		},
 	}
-	err = jsm.WriteState(testNodeID, updatedInstances)
+	err = writeNodeState(t, jsm, testNodeID, updatedInstances)
 	require.NoError(t, err)
 
 	// Load and verify updated state
@@ -285,7 +299,7 @@ func TestJetStreamManager_MultipleNodes(t *testing.T) {
 	node1Instances := map[string]*vm.VM{
 		"i-node1-001": {ID: "i-node1-001", Status: vm.StateRunning},
 	}
-	err = jsm.WriteState("node-1", node1Instances)
+	err = writeNodeState(t, jsm, "node-1", node1Instances)
 	require.NoError(t, err)
 
 	// Write state for node-2
@@ -293,7 +307,7 @@ func TestJetStreamManager_MultipleNodes(t *testing.T) {
 		"i-node2-001": {ID: "i-node2-001", Status: vm.StateStopped},
 		"i-node2-002": {ID: "i-node2-002", Status: vm.StateRunning},
 	}
-	err = jsm.WriteState("node-2", node2Instances)
+	err = writeNodeState(t, jsm, "node-2", node2Instances)
 	require.NoError(t, err)
 
 	// Load and verify node-1 state
@@ -327,7 +341,7 @@ func TestJetStreamManager_KVNotInitialized(t *testing.T) {
 	require.NoError(t, err)
 
 	testInstances := make(map[string]*vm.VM)
-	err = jsm.WriteState("test-node", testInstances)
+	err = writeNodeState(t, jsm, "test-node", testInstances)
 	assert.Error(t, err, "WriteState should error when KV not initialized")
 
 	_, _, err = jsm.LoadState("test-node")
@@ -625,12 +639,14 @@ func TestJetStreamManager_UpdateStoppedInstance_NotFound(t *testing.T) {
 	assert.ErrorIs(t, err, kvstore.ErrNotFound)
 }
 
-// TestJetStreamManager_UpdateStoppedInstance_NoResurrectAfterClaim is the
-// core TOCTOU regression test: a claim (delete) that lands between an
-// UpdateStoppedInstance caller's own Load and its CAS write must not be
-// undone. createIfAbsent=false means the CAS write observes the key gone and
-// fails with kvstore.ErrNotFound instead of recreating the stopped record after
-// ClaimStoppedInstance already handed it off to a winning start.
+// TestJetStreamManager_UpdateStoppedInstance_NoResurrectAfterClaim is the core
+// TOCTOU regression test: a claim landing between an UpdateStoppedInstance
+// caller's own Load and its CAS write must not be undone.
+//
+// The claim no longer deletes anything, so the key is still there and the CAS
+// write would succeed on revision alone. What refuses it is the membership
+// test — the record it reads is no longer stopped, and to this API an instance
+// that is not stopped is one that is not there.
 func TestJetStreamManager_UpdateStoppedInstance_NoResurrectAfterClaim(t *testing.T) {
 	nc, err := nats.Connect(sharedJSNATSURL)
 	require.NoError(t, err)
@@ -644,7 +660,7 @@ func TestJetStreamManager_UpdateStoppedInstance_NoResurrectAfterClaim(t *testing
 	require.NoError(t, jsm.WriteStoppedInstance(testVM.ID, testVM))
 
 	// A caller loads the record (as TagStoppedInstance / ModifyInstanceAttribute
-	// do) before a winning claim removes it.
+	// do) before a winning claim takes it.
 	loaded, err := jsm.LoadStoppedInstance(testVM.ID)
 	require.NoError(t, err)
 	require.NotNil(t, loaded)
@@ -654,15 +670,24 @@ func TestJetStreamManager_UpdateStoppedInstance_NoResurrectAfterClaim(t *testing
 	require.NotNil(t, claimed)
 
 	// The tag/attribute caller's CAS write now races the already-completed
-	// claim: it must fail cleanly, not resurrect the record.
+	// claim: it must fail cleanly, not land on the claimed instance.
 	_, err = jsm.UpdateStoppedInstance(testVM.ID, func(v *vm.VM) {
 		v.InstanceType = "should-not-land"
 	})
-	assert.ErrorIs(t, err, kvstore.ErrNotFound, "a claim that deleted the record must not be resurrected by a losing racer's update")
+	assert.ErrorIs(t, err, kvstore.ErrNotFound,
+		"an update that lost the race to a claim must not land on the instance it started")
 
 	stillGone, err := jsm.LoadStoppedInstance(testVM.ID)
 	require.NoError(t, err)
-	assert.Nil(t, stillGone, "the stopped record must stay deleted after the losing update")
+	assert.Nil(t, stillGone, "and the instance must stay claimed, not fall back into the stopped set")
+
+	// The record itself survives the claim — that is the point of claiming by
+	// compare-and-set — so the mutation must be absent from it rather than the
+	// record absent from the bucket.
+	record, err := jsm.LoadInstanceRecord(testVM.ID)
+	require.NoError(t, err)
+	require.NotNil(t, record, "the claim moves the record between sets, it does not delete it")
+	assert.Equal(t, "t3.micro", record.Spec.InstanceType, "the losing update must not have committed")
 }
 
 // TestJetStreamManager_ListStoppedInstances tests listing multiple stopped instances.
@@ -722,7 +747,7 @@ func TestJetStreamManager_StoppedInstances_NoInterference(t *testing.T) {
 	nodeInstances := map[string]*vm.VM{
 		"i-running-001": {ID: "i-running-001", Status: vm.StateRunning},
 	}
-	err = jsm.WriteState("interference-test-node", nodeInstances)
+	err = writeNodeState(t, jsm, "interference-test-node", nodeInstances)
 	require.NoError(t, err)
 
 	// Write stopped instance
@@ -882,7 +907,7 @@ func TestJetStreamManager_WriteState_RecoverAfterStreamLost(t *testing.T) {
 	testInstances := map[string]*vm.VM{
 		"i-recover-001": {ID: "i-recover-001", Status: vm.StateRunning},
 	}
-	err = jsm.WriteState("recovery-node", testInstances)
+	err = writeNodeState(t, jsm, "recovery-node", testInstances)
 	require.NoError(t, err, "WriteState should recover after stream loss")
 
 	// Verify data was written
@@ -1053,7 +1078,7 @@ func TestJetStreamManager_WriteState_RecoveryFailure(t *testing.T) {
 	swapToNonJSContext(t, jsm)
 
 	testInstances := map[string]*vm.VM{"i-fail": {ID: "i-fail"}}
-	err = jsm.WriteState("fail-node", testInstances)
+	err = writeNodeState(t, jsm, "fail-node", testInstances)
 	assert.Error(t, err, "WriteState should return error when recovery fails")
 }
 
@@ -1614,7 +1639,7 @@ func TestJetStreamManager_BestEffort_Success_NotifiesObserver(t *testing.T) {
 	obs := &fakeKVObserver{}
 	jsm.SetSyncObserver(obs)
 
-	jsm.WriteStateBytesBestEffort("obs-success", []byte(`{"vms":{}}`), 5*time.Second)
+	jsm.WriteNodeMarkerBestEffort("obs-success", 5*time.Second)
 
 	successes, failures := obs.snapshot()
 	require.Len(t, successes, 1)
@@ -1636,7 +1661,7 @@ func TestJetStreamManager_BestEffort_PutError_NotifiesObserver(t *testing.T) {
 	// Closing the connection forces the inflight Put to fail without timing out.
 	nc.Close()
 
-	jsm.WriteStateBytesBestEffort("obs-fail", []byte(`{"vms":{}}`), 5*time.Second)
+	jsm.WriteNodeMarkerBestEffort("obs-fail", 5*time.Second)
 
 	successes, failures := obs.snapshot()
 	assert.Empty(t, successes)
@@ -1655,7 +1680,7 @@ func TestJetStreamManager_BestEffort_NilObserver_NoPanic(t *testing.T) {
 	require.NoError(t, jsm.InitKVBucket())
 
 	// No observer set — must not panic on success or failure paths.
-	jsm.WriteStateBytesBestEffort("obs-nil", []byte(`{"vms":{}}`), 5*time.Second)
+	jsm.WriteNodeMarkerBestEffort("obs-nil", 5*time.Second)
 }
 
 // --- UpdateMgmtIPAM CAS tests ---
@@ -1750,10 +1775,11 @@ func TestJetStreamManager_UpdateMgmtIPAM_ClusterKVNotInitialized(t *testing.T) {
 // logged an undecodable record and carried on, so the instance simply vanished
 // from DescribeInstances and the caller concluded it was gone. A tenant-facing
 // listing is complete or it fails.
+// Its own JetStream rather than the package's: one key space holds every
+// instance now, so an undecodable record fails every listing that reads it and
+// not just the one under test.
 func TestJetStreamManager_ListStoppedInstances_FailsOnAnUndecodableRecord(t *testing.T) {
-	nc, err := nats.Connect(sharedJSNATSURL)
-	require.NoError(t, err)
-	defer nc.Close()
+	_, nc, _ := testutil.StartTestJetStream(t)
 
 	jsm, err := NewJetStreamManager(nc, 1)
 	require.NoError(t, err)
@@ -1763,7 +1789,7 @@ func TestJetStreamManager_ListStoppedInstances_FailsOnAnUndecodableRecord(t *tes
 
 	kv, err := jsm.stateB.KV(t.Context())
 	require.NoError(t, err)
-	_, err = kv.Put(t.Context(), StoppedInstancePrefix+"i-corrupt", []byte("{not json"))
+	_, err = kv.Put(t.Context(), InstanceRecordPrefix+"i-corrupt", []byte("{not json"))
 	require.NoError(t, err)
 
 	_, err = jsm.ListStoppedInstances()
@@ -1802,12 +1828,15 @@ func seedNodeRecord(t *testing.T, jsm *JetStreamManager, nodeID string, raw []by
 	t.Helper()
 	kv, err := jsm.stateB.KV(t.Context())
 	require.NoError(t, err)
-	_, err = kv.Put(t.Context(), InstanceStatePrefix+nodeID, raw)
+	_, err = kv.Put(t.Context(), NodePresencePrefix+nodeID, raw)
 	require.NoError(t, err)
 }
 
 // A node record predates schema versioning, so it carries no version at all.
 // Reading it as current is what makes an in-place upgrade need no migration.
+//
+// The instances come from the records, not from the marker, so a marker that
+// still carries a vms member is read for its presence and nothing else.
 func TestJetStreamManager_LoadState_UnversionedRecordStillReads(t *testing.T) {
 	nc, err := nats.Connect(sharedJSNATSURL)
 	require.NoError(t, err)
@@ -1819,12 +1848,15 @@ func TestJetStreamManager_LoadState_UnversionedRecordStillReads(t *testing.T) {
 
 	seedNodeRecord(t, jsm, "legacy-node",
 		[]byte(`{"vms":{"i-legacy-001":{"id":"i-legacy-001","status":"running"}}}`))
+	require.NoError(t, jsm.WriteInstanceRecord("i-legacy-002",
+		(&vm.VM{ID: "i-legacy-002", Status: vm.StateRunning, LastNode: "legacy-node"}).Record()))
 
 	loaded, found, err := jsm.LoadState("legacy-node")
 	require.NoError(t, err, "A record written before versioning must still read")
 	require.True(t, found)
 	require.Len(t, loaded, 1)
-	assert.Equal(t, vm.StateRunning, loaded["i-legacy-001"].Status)
+	assert.Equal(t, vm.StateRunning, loaded["i-legacy-002"].Status)
+	assert.NotContains(t, loaded, "i-legacy-001", "the marker is not a running set")
 }
 
 // A record from a newer binary is refused rather than parsed on a guess: this
@@ -1849,7 +1881,7 @@ func TestJetStreamManager_LoadState_FutureRecordIsRefused(t *testing.T) {
 }
 
 // Both writers stamp the version: the typed WriteState and the best-effort
-// pre-marshalled path, which is the one that would silently skip it.
+// marker path, which is the one that would silently skip it.
 func TestJetStreamManager_WriteState_StampsSchemaVersion(t *testing.T) {
 	nc, err := nats.Connect(sharedJSNATSURL)
 	require.NoError(t, err)
@@ -1860,14 +1892,14 @@ func TestJetStreamManager_WriteState_StampsSchemaVersion(t *testing.T) {
 	require.NoError(t, jsm.InitKVBucket())
 
 	vms := map[string]*vm.VM{"i-stamp-001": {ID: "i-stamp-001", Status: vm.StateRunning}}
-	require.NoError(t, jsm.WriteState("stamp-node", vms))
+	require.NoError(t, writeNodeState(t, jsm, "stamp-node", vms))
 
 	kv, err := jsm.stateB.KV(t.Context())
 	require.NoError(t, err)
 
 	read := func(nodeID string) LocalState {
 		t.Helper()
-		entry, err := kv.Get(t.Context(), InstanceStatePrefix+nodeID)
+		entry, err := kv.Get(t.Context(), NodePresencePrefix+nodeID)
 		require.NoError(t, err)
 		var got LocalState
 		require.NoError(t, json.Unmarshal(entry.Value(), &got))
@@ -1876,24 +1908,10 @@ func TestJetStreamManager_WriteState_StampsSchemaVersion(t *testing.T) {
 
 	assert.Equal(t, LocalStateSchemaVersion, read("stamp-node").SchemaVersion)
 
-	data, err := marshalInstanceState(vms)
-	require.NoError(t, err)
-	jsm.WriteStateBytesBestEffort("bytes-node", data, 5*time.Second)
+	jsm.WriteNodeMarkerBestEffort("bytes-node", 5*time.Second)
 	assert.Equal(t, LocalStateSchemaVersion, read("bytes-node").SchemaVersion,
-		"the best-effort path marshals its own bytes and must stamp the same version")
-}
-
-// The KV node record and the local state file are one envelope. Asserted on the
-// bytes rather than the type, because a second marshaller is how they drift.
-func TestMarshalInstanceState_MatchesLocalStateFileBytes(t *testing.T) {
-	vms := map[string]*vm.VM{
-		"i-envelope-001": {ID: "i-envelope-001", Status: vm.StateRunning, InstanceType: "t3.micro"},
-	}
-
-	kvBytes, err := marshalInstanceState(vms)
-	require.NoError(t, err)
-	fileBytes, err := MarshalLocalState(vms)
-	require.NoError(t, err)
-
-	assert.JSONEq(t, string(fileBytes), string(kvBytes))
+		"the best-effort path must stamp the same version")
+
+	assert.Empty(t, read("stamp-node").VMS,
+		"the node key is a presence marker: carrying the running set is the cost the split removes")
 }

spinifex/daemon/state_test.go

@@ -758,7 +758,7 @@ func TestRestoreInstances_ClusterRecordWinsForSharedInstance(t *testing.T) {
 	require.NoError(t, daemon.WriteState())
 
 	// The cluster record covers only i-shared, and disagrees about it.
-	require.NoError(t, daemon.jsManager.WriteState(daemon.node, map[string]*vm.VM{
+	require.NoError(t, writeNodeState(t, daemon.jsManager, daemon.node, map[string]*vm.VM{
 		"i-shared": {ID: "i-shared", Status: vm.StateError, InstanceType: "t3.small"},
 	}))
 

spinifex/migrate/VERSIONS.md

@@ -19,7 +19,7 @@ Single source of truth for the current schema version of every config file Spini
 
 | Bucket | Current version | Constant |
 |---|---|---|
-| `spinifex-instance-state` | `4` | `daemon.InstanceStateBucketVersion` |
+| `spinifex-instance-state` | `5` | `daemon.InstanceStateBucketVersion` |
 | `spinifex-cluster-state` | `1` | `daemon.ClusterStateBucketVersion` |
 | `spinifex-terminated-instances` | `3` | `daemon.TerminatedInstanceBucketVersion` |
 
@@ -34,11 +34,14 @@ Single source of truth for the current schema version of every config file Spini
 | `spinifex-instance-state` | `1`→`2` | copy the `instance.<id>` keys onto the record space |
 | `spinifex-instance-state` | `2`→`3` | split each `node.<id>` blob, one record per instance in it |
 | `spinifex-instance-state` | `3`→`4` | re-key the record space from `i/<id>` to `i.<id>` |
+| `spinifex-instance-state` | `4`→`5` | carry each `node.<id>` blob's presence and ownership onto the record space: a `nodepresence.<id>` marker, and that node named as `last_node` on every record in the blob that has no owner |
 | `spinifex-terminated-instances` | `1`→`2` | copy the `terminated.<id>` keys onto the record space |
 | `spinifex-terminated-instances` | `2`→`3` | re-key the record space from `i/<id>` to `i.<id>` |
 
 The re-key steps exist because a watch filter is a NATS subject filter: `*` matches one dot-delimited token, so `i/<id>` is a single token and `i/*` matches nothing. Only a build that shipped the slash has anything for them to move.
 
+Version `5` of `spinifex-instance-state` is the cutover: the record space became the only copy, and `instance.<id>` and `node.<id>` are frozen — read by nothing, written by nothing, and left in place so the crossing can be rolled back. The bump is also what a build predating the cutover trips over: `RunKV` refuses a bucket stamped past what it understands, with `SchemaAheadError`, rather than opening it and reading half a key space.
+
 Every other bucket has no registered migration, so `RunKV` stamps its target version directly on first init.
 
 **Object store:** none, so `RunObject` stamps directly too.

spinifex/migrate/migrate.go

@@ -152,7 +152,10 @@ func (r *Registry) RunObject(ctx context.Context, target string, objects objects
 		return fmt.Errorf("read version for %s: %w", target, err)
 	}
 
-	if current >= targetVersion {
+	if current > targetVersion {
+		return SchemaAheadError{Bucket: target, Found: current, Understood: targetVersion}
+	}
+	if current == targetVersion {
 		return nil
 	}
 
@@ -207,6 +210,25 @@ func (r *Registry) RunObject(ctx context.Context, target string, objects objects
 	return nil
 }
 
+// SchemaAheadError reports a bucket migrated past what this build understands,
+// which means another node in the cluster is running a newer release.
+//
+// Stopping is the point. A migration only ever adds steps, so a newer schema is
+// one this build has no code for: it would read the keys it knows, miss the
+// ones it does not, and write back a view of the bucket assembled from half of
+// it. Refusing to open is recoverable; that is not.
+type SchemaAheadError struct {
+	Bucket     string
+	Found      int
+	Understood int
+}
+
+func (e SchemaAheadError) Error() string {
+	return fmt.Sprintf(
+		"%s is at schema version %d but this build understands %d: another node is running a newer release of Spinifex; upgrade this node to match",
+		e.Bucket, e.Found, e.Understood)
+}
+
 // RunKVWithJetStream is RunKV with a JetStream handle attached to each
 // migration's KVContext, enabling cross-bucket reads (e.g. owner-attribution
 // during a backfill). Prefer plain RunKV when the migration is self-contained.
@@ -226,7 +248,10 @@ func (r *Registry) runKV(ctx context.Context, bucket string, kv jetstream.KeyVal
 		return fmt.Errorf("read version for %s: %w", bucket, err)
 	}
 
-	if current >= targetVersion {
+	if current > targetVersion {
+		return SchemaAheadError{Bucket: bucket, Found: current, Understood: targetVersion}
+	}
+	if current == targetVersion {
 		return nil
 	}
 

spinifex/migrate/migrate_test.go

@@ -242,6 +242,33 @@ func TestRunKV_Idempotent(t *testing.T) {
 	assert.Equal(t, 1, runCount) // Not incremented.
 }
 
+// A bucket stamped past what this build understands was migrated by a node on
+// a newer release. Opening it anyway is the mixed-cluster failure: this build
+// reads the keys it knows, misses the ones it does not, and writes back a view
+// of the bucket assembled from half of it.
+func TestRunKV_SchemaAheadOfThisBuildIsRefused(t *testing.T) {
+	r := NewRegistry()
+	r.RegisterKV("test-bucket", KVMigration{
+		FromVersion: 1, ToVersion: 2, Description: "add field",
+		Run: func(ctx context.Context, kvc KVContext) error { return nil },
+	})
+
+	_, nc := startTestNATS(t)
+	kv := createTestBucket(t, nc, "test-bucket")
+	_, err := kv.PutString(t.Context(), "_version", "5")
+	require.NoError(t, err)
+
+	err = r.RunKV(t.Context(), "test-bucket", kv, 2)
+
+	var ahead SchemaAheadError
+	require.ErrorAs(t, err, &ahead)
+	assert.Equal(t, 5, ahead.Found)
+	assert.Equal(t, 2, ahead.Understood)
+	assert.Contains(t, err.Error(), "test-bucket")
+	assert.Contains(t, err.Error(), "upgrade this node",
+		"the error has to say what to do about it, not just that it happened")
+}
+
 // --- Config version reader tests ---
 
 func TestTOMLVersionReader_ReadStringVersion(t *testing.T) {