diff --git a/spinifex/daemon/instance_records.go b/spinifex/daemon/instance_records.go index 0179e0793..2385391e6 100644 --- a/spinifex/daemon/instance_records.go +++ b/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. 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 diff --git a/spinifex/daemon/instance_records_migrate.go b/spinifex/daemon/instance_records_migrate.go new file mode 100644 index 000000000..df19cef9d --- /dev/null +++ b/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/. 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/ are +// the ones a node that predates it wrote afterwards. +// +// The node. 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/ 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/ key space", + Run: func(ctx context.Context, kvc migrate.KVContext) error { + return copyInstancesForward(ctx, kvc, TerminatedInstancePrefix) + }, + }) +} + +// copyInstancesForward writes an i/ 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 +} diff --git a/spinifex/daemon/instance_records_test.go b/spinifex/daemon/instance_records_test.go index a7b6e723f..732b9601c 100644 --- a/spinifex/daemon/instance_records_test.go +++ b/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 diff --git a/spinifex/daemon/instance_records_transition_test.go b/spinifex/daemon/instance_records_transition_test.go new file mode 100644 index 000000000..d0d213952 --- /dev/null +++ b/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/ 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) +} diff --git a/spinifex/daemon/jetstream.go b/spinifex/daemon/jetstream.go index dc61205fd..b160a10a2 100644 --- a/spinifex/daemon/jetstream.go +++ b/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/; 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. first and is then mirrored onto i/. +// That order is not arbitrary: a node that predates the per-resource key space +// serialises its own writes against instance., 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. decides whether the instance exists and i/ 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/: 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/ 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. and the committed result mirrored +// onto i/, 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/ 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/ 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. decides existence, i/ 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) } diff --git a/spinifex/migrate/VERSIONS.md b/spinifex/migrate/VERSIONS.md index 44706394b..5667ff77f 100644 --- a/spinifex/migrate/VERSIONS.md +++ b/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/` 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'