feat: dual write and read the i/<id> instance record key space (#932)

* feat: dual write and read the i/<id> instance record key space

* fix: make the i/<id> key space a shadow until the cutover

spinifex/daemon/instance_records.go

@@ -3,6 +3,7 @@ package daemon
 import (
 	"context"
 	"errors"
+	"fmt"
 	"log/slog"
 
 	"github.com/mulgadc/spinifex/spinifex/kvstore"
@@ -10,14 +11,14 @@ import (
 )
 
 // InstanceRecordPrefix is the key prefix for the per-resource instance record
-// space. Nothing writes it yet; it exists so the accessors can be settled and
-// tested before the keys move onto it.
+// space. The stopped and terminated key spaces are mirrored onto it; the
+// node.<id> blob is not, and still holds a node's whole running set.
 //
-// It cannot collide with the three prefixes it will replace. Those separate
-// their prefix from the ID with ".", and a key beginning "i/" is neither
-// "instance." nor "node." nor "terminated." — List and DeletePrefix match on a
-// plain string prefix, not on NATS subject tokens, so the two spaces are
-// disjoint while they share a bucket.
+// It cannot collide with the three prefixes it replaces. Those separate their
+// prefix from the ID with ".", and a key beginning "i/" is neither "instance."
+// nor "node." nor "terminated." — List and DeletePrefix match on a plain
+// string prefix, not on NATS subject tokens, so the two spaces are disjoint
+// while they share a bucket.
 const InstanceRecordPrefix = "i/"
 
 // instanceRecordKey is the only place the prefix and the ID are joined, so a
@@ -152,6 +153,111 @@ 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.
+//
+// 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
+}
+
+// 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
+		}
+		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 {
+		return nil, nil
+	}
+
+	newer, err := listRecords(records, 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))
+			continue
+		}
+		out = append(out, instance)
+	}
+	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

@@ -0,0 +1,96 @@
+package daemon
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/mulgadc/spinifex/spinifex/migrate"
+	"github.com/mulgadc/spinifex/spinifex/vm"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// The copy-forward onto i/<id>. 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 from i/<id> are
+// the ones a node that predates it wrote afterwards.
+//
+// The node.<id> blob is deliberately not copied. It holds a whole node's
+// running set in one record, so moving it is a split rather than a copy.
+func init() {
+	migrate.DefaultRegistry.RegisterKV(InstanceStateBucket, migrate.KVMigration{
+		FromVersion: 1,
+		ToVersion:   2,
+		Description: "copy stopped instances forward to the i/<id> key space",
+		Run: func(ctx context.Context, kvc migrate.KVContext) error {
+			return copyInstancesForward(ctx, kvc, StoppedInstancePrefix)
+		},
+	})
+	migrate.DefaultRegistry.RegisterKV(TerminatedInstanceBucket, migrate.KVMigration{
+		FromVersion: 1,
+		ToVersion:   2,
+		Description: "copy terminated instances forward to the i/<id> key space",
+		Run: func(ctx context.Context, kvc migrate.KVContext) error {
+			return copyInstancesForward(ctx, kvc, TerminatedInstancePrefix)
+		},
+	})
+}
+
+// copyInstancesForward writes an i/<id> record for every key under prefix.
+//
+// Two nodes can reach this at once — the version stamp is a read-then-write,
+// not a CAS — and a node that has already upgraded can be writing the
+// destination while the scan runs. Both are handled by creating rather than
+// putting: a destination that already exists holds either the same copy or a
+// fresher one, and neither may be overwritten with what this scan read.
+func copyInstancesForward(ctx context.Context, kvc migrate.KVContext, prefix string) 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)
+	}
+
+	copied := 0
+	for _, key := range keys {
+		if !strings.HasPrefix(key, prefix) {
+			continue
+		}
+
+		entry, err := kvc.KV.Get(ctx, key)
+		if err != nil {
+			// Deleted between the listing and the read: there is nothing left to
+			// copy, and the delete is the newer fact.
+			if errors.Is(err, jetstream.ErrKeyNotFound) {
+				continue
+			}
+			return fmt.Errorf("read %s: %w", key, err)
+		}
+
+		var instance vm.VM
+		if err := json.Unmarshal(entry.Value(), &instance); err != nil {
+			return fmt.Errorf("decode %s: %w", key, err)
+		}
+
+		record, err := json.Marshal(instance.Record())
+		if err != nil {
+			return fmt.Errorf("encode record for %s: %w", key, err)
+		}
+
+		dest := instanceRecordKey(strings.TrimPrefix(key, prefix))
+		if _, err := kvc.KV.Create(ctx, dest, record); err != nil {
+			if errors.Is(err, jetstream.ErrKeyExists) {
+				continue
+			}
+			return fmt.Errorf("write %s: %w", dest, err)
+		}
+		copied++
+	}
+
+	kvc.Logger.Info("Copied instances onto the per-resource key space",
+		"prefix", prefix, "copied", copied)
+	return nil
+}

spinifex/daemon/instance_records_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/resource"
 	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/mulgadc/spinifex/spinifex/vm"
+	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -17,6 +18,14 @@ import (
 // record-carrying buckets open. Each test gets its own server, so tests never
 // see each other's keys.
 func newRecordManager(t *testing.T) *daemon.JetStreamManager {
+	t.Helper()
+	m, _ := newRecordManagerConn(t)
+	return m
+}
+
+// newRecordManagerConn also hands back the connection, for tests that have to
+// reach the bucket underneath the manager to stage what another node did.
+func newRecordManagerConn(t *testing.T) (*daemon.JetStreamManager, *nats.Conn) {
 	t.Helper()
 	_, nc, _ := testutil.StartTestJetStream(t)
 
@@ -24,7 +33,7 @@ func newRecordManager(t *testing.T) *daemon.JetStreamManager {
 	require.NoError(t, err)
 	require.NoError(t, m.InitKVBucket())
 	require.NoError(t, m.InitTerminatedInstanceBucket())
-	return m
+	return m, nc
 }
 
 func testRecord(id string) *vm.InstanceRecord {
@@ -61,39 +70,39 @@ func TestLoadInstanceRecord_AbsentIsNil(t *testing.T) {
 	assert.Nil(t, got)
 }
 
-// The whole point of a fresh prefix: an old binary reading the keys this
-// replaces must not see these records, and this one must not see those. They
-// share a bucket, so nothing but the prefix keeps them apart.
-func TestInstanceRecords_AreDisjointFromTheKeysTheyReplace(t *testing.T) {
+// 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) {
 	m := newRecordManager(t)
 
 	require.NoError(t, m.WriteStoppedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"}))
-	require.NoError(t, m.WriteInstanceRecord("i-1", testRecord("i-1")))
 
 	stopped, err := m.ListStoppedInstances()
 	require.NoError(t, err)
-	require.Len(t, stopped, 1, "the record must not appear in the stopped listing")
+	require.Len(t, stopped, 1, "an instance held at both keys 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 stopped instance must not appear in the record listing")
-	assert.Equal(t, "m7i.large", records[0].Spec.InstanceType)
+	require.Len(t, records, 1, "the write must have been mirrored onto the record key")
+	assert.Equal(t, "t3.nano", records[0].Spec.InstanceType)
 
-	// And deleting one leaves the other, which is what makes the dual-write
-	// release safe to roll back.
+	// 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"))
 
-	stopped, err = m.ListStoppedInstances()
+	got, err := m.LoadStoppedInstance("i-1")
 	require.NoError(t, err)
-	assert.Len(t, stopped, 1, "deleting the record must not touch the key it will replace")
+	require.NotNil(t, got, "deleting the mirror must not touch the key it mirrors")
+	assert.Equal(t, "t3.nano", got.InstanceType)
 }
 
-func TestTerminatedInstanceRecords_AreDisjointFromTheKeysTheyReplace(t *testing.T) {
+func TestTerminatedInstances_SpanBothKeySpaces(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.WriteTerminatedInstanceRecord("i-1", testRecord("i-1")))
 
 	terminated, err := m.ListTerminatedInstances()
 	require.NoError(t, err)
@@ -103,7 +112,14 @@ func TestTerminatedInstanceRecords_AreDisjointFromTheKeysTheyReplace(t *testing.
 	records, err := m.ListTerminatedInstanceRecords()
 	require.NoError(t, err)
 	require.Len(t, records, 1)
-	assert.Equal(t, "m7i.large", records[0].Spec.InstanceType)
+	assert.Equal(t, "t3.nano", records[0].Spec.InstanceType)
+
+	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)
 }
 
 // The two buckets are separate stores over the same prefix, so a record in one

spinifex/daemon/instance_records_transition_test.go

@@ -0,0 +1,383 @@
+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"
+)
+
+// 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) {
+	t.Helper()
+	js, err := jetstream.New(nc)
+	require.NoError(t, err)
+	kv, err := js.KeyValue(context.Background(), daemon.InstanceStateBucket)
+	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")
+	require.NoError(t, err)
+	require.NotNil(t, record, "the older node cannot have removed a key it does not know")
+
+	stopped, err := m.ListStoppedInstances()
+	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")
+}
+
+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) {
+	m := newRecordManager(t)
+	require.NoError(t, m.WriteStoppedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"}))
+
+	_, err := m.UpdateStoppedInstance("i-1", func(v *vm.VM) { v.LastNode = "node-2" })
+	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, "the mirror 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) {
+	m := newRecordManager(t)
+
+	_, err := m.UpdateStoppedInstance("i-gone", func(v *vm.VM) { v.LastNode = "node-2" })
+	require.Error(t, err)
+
+	record, err := m.LoadInstanceRecord("i-gone")
+	require.NoError(t, err)
+	assert.Nil(t, record)
+}
+
+func TestDeleteStoppedInstance_ClearsBothKeySpaces(t *testing.T) {
+	m := newRecordManager(t)
+	require.NoError(t, m.WriteStoppedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"}))
+
+	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")
+
+	stopped, err := m.ListStoppedInstances()
+	require.NoError(t, err)
+	assert.Empty(t, stopped)
+}
+
+// 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) {
+	m := newRecordManager(t)
+	require.NoError(t, m.WriteStoppedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"}))
+
+	claimed, err := m.ClaimStoppedInstance("i-1")
+	require.NoError(t, err)
+	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) {
+	m := newRecordManager(t)
+
+	require.NoError(t, m.WriteTerminatedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"}))
+
+	record, err := m.LoadTerminatedInstanceRecord("i-1")
+	require.NoError(t, err)
+	require.NotNil(t, record)
+	assert.Equal(t, "t3.nano", record.Spec.InstanceType)
+}
+
+func TestUpdateTerminatedInstance_MirrorsTheCommittedValue(t *testing.T) {
+	m := newRecordManager(t)
+	require.NoError(t, m.WriteTerminatedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"}))
+
+	_, err := m.UpdateTerminatedInstance("i-1", func(v *vm.VM) {
+		v.Teardown = map[string]string{"eni": "done"}
+	})
+	require.NoError(t, err)
+
+	record, err := m.LoadTerminatedInstanceRecord("i-1")
+	require.NoError(t, err)
+	require.NotNil(t, record)
+	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)
+	require.NoError(t, m.WriteTerminatedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"}))
+
+	require.NoError(t, m.DeleteTerminatedInstance("i-1"))
+
+	record, err := m.LoadTerminatedInstanceRecord("i-1")
+	require.NoError(t, err)
+	assert.Nil(t, record)
+
+	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)
+}
+
+// seedLegacyBucket creates bucket at schema version 1 holding instance at the
+// key the migration copies from, as a cluster that has never run the
+// per-resource key space would leave it.
+func seedLegacyBucket(t *testing.T, nc *nats.Conn, bucket, prefix string, instance *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: bucket, History: 1})
+	require.NoError(t, err)
+
+	data, err := json.Marshal(instance)
+	require.NoError(t, err)
+	_, err = kv.Put(ctx, prefix+instance.ID, data)
+	require.NoError(t, err)
+
+	require.NoError(t, kvutil.WriteVersion(ctx, kv, 1))
+	return kv
+}
+
+func TestInstanceStateMigration_CopiesStoppedInstancesForward(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	seedLegacyBucket(t, nc, daemon.InstanceStateBucket, daemon.StoppedInstancePrefix,
+		&vm.VM{ID: "i-1", InstanceType: "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, "opening the bucket must copy the old keys forward")
+	assert.Equal(t, "t3.nano", record.Spec.InstanceType)
+}
+
+// The node blob holds a whole node's running set in one record, so it is a
+// split rather than a copy and is not this migration's to make.
+func TestInstanceStateMigration_LeavesTheNodeBlobAlone(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	kv := seedLegacyBucket(t, nc, daemon.InstanceStateBucket, daemon.StoppedInstancePrefix,
+		&vm.VM{ID: "i-1", InstanceType: "t3.nano"})
+
+	blob, err := json.Marshal(map[string]any{"instances": map[string]any{}})
+	require.NoError(t, err)
+	_, err = kv.Put(context.Background(), daemon.InstanceStatePrefix+"node-1", blob)
+	require.NoError(t, err)
+
+	m, err := daemon.NewJetStreamManager(nc, 1)
+	require.NoError(t, err)
+	require.NoError(t, m.InitKVBucket())
+
+	records, err := m.ListInstanceRecords()
+	require.NoError(t, err)
+	require.Len(t, records, 1)
+	assert.Equal(t, "i-1", records[0].Metadata.Name)
+}
+
+// A node that has already upgraded can be writing the destination while the
+// scan runs, and the version stamp is a read-then-write rather than a CAS, so
+// two nodes can run this at once. Neither may lose a fresher record.
+func TestInstanceStateMigration_DoesNotOverwriteAFresherRecord(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	kv := seedLegacyBucket(t, nc, daemon.InstanceStateBucket, daemon.StoppedInstancePrefix,
+		&vm.VM{ID: "i-1", InstanceType: "t3.nano"})
+
+	fresher, err := json.Marshal((&vm.VM{ID: "i-1", InstanceType: "m7i.large"}).Record())
+	require.NoError(t, err)
+	_, err = kv.Put(context.Background(), daemon.InstanceRecordPrefix+"i-1", fresher)
+	require.NoError(t, err)
+
+	// A second instance nobody has copied yet, so the run this test measures is
+	// one that does work rather than one that skips everything.
+	untouched, err := json.Marshal(&vm.VM{ID: "i-2", InstanceType: "t3.micro"})
+	require.NoError(t, err)
+	_, err = kv.Put(context.Background(), daemon.StoppedInstancePrefix+"i-2", untouched)
+	require.NoError(t, err)
+
+	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, "m7i.large", record.Spec.InstanceType)
+
+	copied, err := m.LoadInstanceRecord("i-2")
+	require.NoError(t, err)
+	require.NotNil(t, copied, "skipping an existing destination must not stop the scan")
+	assert.Equal(t, "t3.micro", copied.Spec.InstanceType)
+}
+
+func TestInstanceStateMigration_IsSafeToRunTwice(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	kv := seedLegacyBucket(t, nc, daemon.InstanceStateBucket, daemon.StoppedInstancePrefix,
+		&vm.VM{ID: "i-1", InstanceType: "t3.nano"})
+
+	first, err := daemon.NewJetStreamManager(nc, 1)
+	require.NoError(t, err)
+	require.NoError(t, first.InitKVBucket())
+
+	require.NoError(t, kvutil.WriteVersion(context.Background(), kv, 1))
+	second, err := daemon.NewJetStreamManager(nc, 1)
+	require.NoError(t, err)
+	require.NoError(t, second.InitKVBucket())
+
+	records, err := second.ListInstanceRecords()
+	require.NoError(t, err)
+	require.Len(t, records, 1)
+	assert.Equal(t, "t3.nano", records[0].Spec.InstanceType)
+}
+
+func TestTerminatedInstanceMigration_CopiesTerminatedInstancesForward(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	seedLegacyBucket(t, nc, daemon.TerminatedInstanceBucket, daemon.TerminatedInstancePrefix,
+		&vm.VM{ID: "i-1", InstanceType: "t3.nano"})
+
+	m, err := daemon.NewJetStreamManager(nc, 1)
+	require.NoError(t, err)
+	require.NoError(t, m.InitTerminatedInstanceBucket())
+
+	record, err := m.LoadTerminatedInstanceRecord("i-1")
+	require.NoError(t, err)
+	require.NotNil(t, record)
+	assert.Equal(t, "t3.nano", record.Spec.InstanceType)
+}

spinifex/daemon/jetstream.go

@@ -32,10 +32,11 @@ const (
 	// TerminatedInstancePrefix is the key prefix for terminated instances.
 	TerminatedInstancePrefix = "terminated."
 
-	// Schema versions for daemon KV buckets.
-	InstanceStateBucketVersion      = 1
+	// Schema versions for daemon KV buckets. The two instance buckets are at 2
+	// for the copy-forward onto i/<id>; see instance_records_migrate.go.
+	InstanceStateBucketVersion      = 2
 	ClusterStateBucketVersion       = 1
-	TerminatedInstanceBucketVersion = 1
+	TerminatedInstanceBucketVersion = 2
 )
 
 // KVSyncObserver receives best-effort KV sync outcomes from
@@ -574,6 +575,11 @@ func (m *JetStreamManager) UpdateReplicas(newReplicas int) error {
 // rather than silently overwriting it out of order. This is the substrate
 // 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.
 func (m *JetStreamManager) WriteStoppedInstance(instanceID string, instance *vm.VM) error {
 	if m.stopped == nil {
 		return errors.New("KV bucket not initialized")
@@ -585,37 +591,35 @@ func (m *JetStreamManager) WriteStoppedInstance(instanceID string, instance *vm.
 	if err := m.stopped.Replace(context.Background(), key, instance); err != nil {
 		return err
 	}
+	if err := mirrorRecord(m.records, instanceID, instance); err != nil {
+		return err
+	}
 
 	slog.Debug("Wrote stopped instance to JetStream KV", "key", key, "instanceId", instanceID)
 	return nil
 }
 
 // LoadStoppedInstance loads a stopped instance from the shared KV store.
-// Returns nil, nil if the key does not exist.
+// 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.
 func (m *JetStreamManager) LoadStoppedInstance(instanceID string) (*vm.VM, error) {
 	if m.stopped == nil {
 		return nil, errors.New("KV bucket not initialized")
 	}
-
-	instance, _, err := m.stopped.Get(context.Background(), StoppedInstancePrefix+instanceID)
-	if err != nil {
-		if errors.Is(err, kvstore.ErrNotFound) {
-			return nil, nil
-		}
-		return nil, err
-	}
-	return instance, nil
+	return loadPreferringRecord(m.records, m.stopped, StoppedInstancePrefix+instanceID, instanceID)
 }
 
-// DeleteStoppedInstance removes a stopped instance from the shared KV store.
-// It is idempotent — deleting a non-existent key is not an error.
+// 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.
 func (m *JetStreamManager) DeleteStoppedInstance(instanceID string) error {
 	if m.stopped == nil {
 		return errors.New("KV bucket not initialized")
 	}
 
 	key := StoppedInstancePrefix + instanceID
-	if err := m.stopped.Delete(context.Background(), key); err != nil {
+	if err := deleteBothRecords(m.records, m.stopped, key, instanceID); err != nil {
 		return err
 	}
 
@@ -632,11 +636,22 @@ func (m *JetStreamManager) DeleteStoppedInstance(instanceID string) error {
 // racing a forwarded call, two nodes racing the same forwarded call, or a
 // 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.
 func (m *JetStreamManager) ClaimStoppedInstance(instanceID string) (*vm.VM, error) {
 	if m.stopped == 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)
 	if err != nil {
@@ -658,11 +673,23 @@ func (m *JetStreamManager) ClaimStoppedInstance(instanceID string) (*vm.VM, erro
 // resurrecting it. Used by tag/attribute mutations that read-modify-write a
 // 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.
 func (m *JetStreamManager) UpdateStoppedInstance(instanceID string, mutate func(*vm.VM)) (*vm.VM, error) {
 	if m.stopped == nil {
 		return nil, errors.New("KV bucket not initialized")
 	}
-	return mutateRecord(m.stopped, StoppedInstancePrefix+instanceID, mutate)
+
+	updated, err := mutateRecord(m.stopped, StoppedInstancePrefix+instanceID, mutate)
+	if err != nil {
+		return nil, err
+	}
+	if err := mirrorRecord(m.records, instanceID, updated); err != nil {
+		return nil, err
+	}
+	return updated, nil
 }
 
 // mutateRecord applies mutate under the store's CAS loop and returns the
@@ -681,12 +708,13 @@ 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.
+// ListStoppedInstances returns all stopped instances from the shared KV store,
+// answering each from its i/<id> mirror where there is one.
 func (m *JetStreamManager) ListStoppedInstances() ([]*vm.VM, error) {
 	if m.stopped == nil {
 		return nil, errors.New("KV bucket not initialized")
 	}
-	return listRecords(m.stopped, StoppedInstancePrefix)
+	return listPreferringRecords(m.records, m.stopped, StoppedInstancePrefix)
 }
 
 // listRecords returns every record under prefix. Unlike the raw scan it
@@ -725,6 +753,9 @@ func (m *JetStreamManager) WriteTerminatedInstance(instanceID string, instance *
 	if err := m.term.Replace(context.Background(), key, instance); err != nil {
 		return err
 	}
+	if err := mirrorRecord(m.termRecords, instanceID, instance); err != nil {
+		return err
+	}
 
 	slog.Debug("Wrote terminated instance to JetStream KV", "key", key, "instanceId", instanceID)
 	return nil
@@ -740,25 +771,35 @@ func (m *JetStreamManager) UpdateTerminatedInstance(instanceID string, mutate fu
 	if m.term == nil {
 		return nil, errors.New("terminated instance KV bucket not initialized")
 	}
-	return mutateRecord(m.term, TerminatedInstancePrefix+instanceID, mutate)
+
+	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
 }
 
-// ListTerminatedInstances returns all terminated instances from the terminated KV bucket.
+// ListTerminatedInstances returns all terminated instances from the terminated
+// KV bucket, answering each from its i/<id> mirror where there is one.
 func (m *JetStreamManager) ListTerminatedInstances() ([]*vm.VM, error) {
 	if m.term == nil {
 		return nil, errors.New("terminated instance KV bucket not initialized")
 	}
-	return listRecords(m.term, TerminatedInstancePrefix)
+	return listPreferringRecords(m.termRecords, m.term, TerminatedInstancePrefix)
 }
 
-// DeleteTerminatedInstance removes a terminated instance from the terminated KV bucket.
+// DeleteTerminatedInstance removes a terminated instance from both key spaces
+// in the terminated KV bucket.
 func (m *JetStreamManager) DeleteTerminatedInstance(instanceID string) error {
 	if m.term == nil {
 		return errors.New("terminated instance KV bucket not initialized")
 	}
 
 	key := TerminatedInstancePrefix + instanceID
-	if err := m.term.Delete(context.Background(), key); err != nil {
+	if err := deleteBothRecords(m.termRecords, m.term, key, instanceID); err != nil {
 		return err
 	}
 
@@ -766,19 +807,12 @@ func (m *JetStreamManager) DeleteTerminatedInstance(instanceID string) error {
 	return nil
 }
 
-// LoadTerminatedInstance loads a single terminated instance from the terminated KV bucket.
-// Returns nil, nil if the key does not exist.
+// 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.
 func (m *JetStreamManager) LoadTerminatedInstance(instanceID string) (*vm.VM, error) {
 	if m.term == nil {
 		return nil, errors.New("terminated instance KV bucket not initialized")
 	}
-
-	instance, _, err := m.term.Get(context.Background(), TerminatedInstancePrefix+instanceID)
-	if err != nil {
-		if errors.Is(err, kvstore.ErrNotFound) {
-			return nil, nil
-		}
-		return nil, err
-	}
-	return instance, nil
+	return loadPreferringRecord(m.termRecords, m.term, TerminatedInstancePrefix+instanceID, instanceID)
 }

spinifex/migrate/VERSIONS.md

@@ -15,17 +15,29 @@ Single source of truth for the current schema version of every config file Spini
 1. **Fresh install (`spx admin init`)** — the version string is baked into the embedded template file and written verbatim to disk. There is no Go constant; the template is the source of truth.
 2. **Upgrade (`spx admin upgrade`)** — the migration framework (`Registry.RunConfig`) calls `ConfigVersionReader.WriteVersion` after each registered migration step. A target with no registered migration is left alone, and the upgrade command reports nothing pending for it.
 
+## KV bucket versions
+
+| Bucket | Current version | Constant |
+|---|---|---|
+| `spinifex-instance-state` | `2` | `daemon.InstanceStateBucketVersion` |
+| `spinifex-cluster-state` | `1` | `daemon.ClusterStateBucketVersion` |
+| `spinifex-terminated-instances` | `2` | `daemon.TerminatedInstanceBucketVersion` |
+
 ## Registered migrations
 
-None. The migrations that used to live here predated a breaking change that required `spx admin init --force`, so no install can reach the current versions by migrating and the steps were dropped rather than left as dead code.
+**Config:** none. The migrations that used to live here predated a breaking change that required `spx admin init --force`, so no install can reach the current versions by migrating and the steps were dropped rather than left as dead code. `spinifex.toml` is still registered as a config target in `migrate.go` so `spx admin upgrade` reports its on-disk version.
 
-`spinifex.toml` is still registered as a config target in `migrate.go` so `spx admin upgrade` reports its on-disk version. KV and object-store buckets have no registered migrations either, so `RunKV` and `RunObject` stamp their target version directly on first init.
+**KV:** `spinifex-instance-state` and `spinifex-terminated-instances` each carry a `1`→`2` step registered from `daemon/instance_records_migrate.go`, copying instance records onto the `i/<id>` per-resource key space. Every other bucket has none, so `RunKV` stamps its target version directly on first init.
+
+**Object store:** none, so `RunObject` stamps directly too.
 
 ## Where migrations live
 
 Register a `ConfigMigration` against `DefaultRegistry` in a new numbered file under `spinifex/migrate/`, and bump the version in both the template and the table above.
 
-There are no migrations in the tree to copy from. For worked examples of all three kinds, read the deleted files out of git history:
+A KV migration goes in the package that owns the bucket instead, next to the constant it bumps — `migrate` cannot import the record types it would have to decode. `daemon/instance_records_migrate.go` is the worked example.
+
+For worked examples of the config and object-store kinds, read the deleted files out of git history:
 
 ```bash
 git log --diff-filter=D --name-only -- 'spinifex/migrate/0*.go'