diff --git a/spinifex/daemon/instance_records.go b/spinifex/daemon/instance_records.go new file mode 100644 index 000000000..0179e0793 --- /dev/null +++ b/spinifex/daemon/instance_records.go @@ -0,0 +1,168 @@ +package daemon + +import ( + "context" + "errors" + "log/slog" + + "github.com/mulgadc/spinifex/spinifex/kvstore" + "github.com/mulgadc/spinifex/spinifex/vm" +) + +// 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. +// +// 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. +const InstanceRecordPrefix = "i/" + +// instanceRecordKey is the only place the prefix and the ID are joined, so a +// reader and a writer cannot disagree about the key shape. +func instanceRecordKey(instanceID string) string { + return InstanceRecordPrefix + instanceID +} + +// WriteInstanceRecord writes an instance record to the shared KV store, +// replacing any existing record for instanceID. Replace rather than Set: the +// write is wholesale, and doing it under CAS is what stops a racing update +// landing out of order. +func (m *JetStreamManager) WriteInstanceRecord(instanceID string, record *vm.InstanceRecord) error { + if m.records == nil { + return errors.New("KV bucket not initialized") + } + + key := instanceRecordKey(instanceID) + if err := m.records.Replace(context.Background(), key, record); err != nil { + return err + } + + slog.Debug("Wrote instance record to JetStream KV", "key", key, "instanceId", instanceID) + return nil +} + +// LoadInstanceRecord loads one instance record from the shared KV store. +// Returns nil, nil if the key does not exist, matching the accessors it will +// replace. +func (m *JetStreamManager) LoadInstanceRecord(instanceID string) (*vm.InstanceRecord, error) { + if m.records == nil { + return nil, errors.New("KV bucket not initialized") + } + return loadRecord(m.records, instanceRecordKey(instanceID)) +} + +// UpdateInstanceRecord atomically applies mutate to the current record for +// instanceID and writes it back under CAS, retrying a concurrent writer's +// revision conflict. Returns kvstore.ErrNotFound if no record exists: a mutation +// racing a delete must not resurrect what it deleted. +func (m *JetStreamManager) UpdateInstanceRecord(instanceID string, mutate func(*vm.InstanceRecord)) (*vm.InstanceRecord, error) { + if m.records == nil { + return nil, errors.New("KV bucket not initialized") + } + return mutateRecord(m.records, instanceRecordKey(instanceID), mutate) +} + +// DeleteInstanceRecord removes an instance record from the shared KV store. +// It is idempotent — deleting a non-existent key is not an error. +func (m *JetStreamManager) DeleteInstanceRecord(instanceID string) error { + if m.records == nil { + return errors.New("KV bucket not initialized") + } + + key := instanceRecordKey(instanceID) + if err := m.records.Delete(context.Background(), key); err != nil { + return err + } + + slog.Debug("Deleted instance record from JetStream KV", "key", key) + return nil +} + +// ListInstanceRecords returns every instance record in the shared KV store. +// This is the scan that replaces the single Get of a node's blob, so it must +// not skip a record it cannot decode: a dropped instance reads as terminated. +func (m *JetStreamManager) ListInstanceRecords() ([]*vm.InstanceRecord, error) { + if m.records == nil { + return nil, errors.New("KV bucket not initialized") + } + return listRecords(m.records, InstanceRecordPrefix) +} + +// WriteTerminatedInstanceRecord writes a terminated instance record to the +// terminated KV bucket, replacing any existing record for instanceID. The entry +// expires with the bucket's TTL. +func (m *JetStreamManager) WriteTerminatedInstanceRecord(instanceID string, record *vm.InstanceRecord) error { + if m.termRecords == nil { + return errors.New("terminated instance KV bucket not initialized") + } + + key := instanceRecordKey(instanceID) + if err := m.termRecords.Replace(context.Background(), key, record); err != nil { + return err + } + + slog.Debug("Wrote terminated instance record to JetStream KV", "key", key, "instanceId", instanceID) + return nil +} + +// LoadTerminatedInstanceRecord loads one terminated instance record. +// Returns nil, nil if the key does not exist. +func (m *JetStreamManager) LoadTerminatedInstanceRecord(instanceID string) (*vm.InstanceRecord, error) { + if m.termRecords == nil { + return nil, errors.New("terminated instance KV bucket not initialized") + } + return loadRecord(m.termRecords, instanceRecordKey(instanceID)) +} + +// UpdateTerminatedInstanceRecord atomically applies mutate to the current +// terminated record for instanceID and writes it back under CAS. The teardown +// reaper merges per-dependent progress this way rather than replacing the +// record, so it cannot clobber a mark a concurrent update wrote. +func (m *JetStreamManager) UpdateTerminatedInstanceRecord(instanceID string, mutate func(*vm.InstanceRecord)) (*vm.InstanceRecord, error) { + if m.termRecords == nil { + return nil, errors.New("terminated instance KV bucket not initialized") + } + return mutateRecord(m.termRecords, instanceRecordKey(instanceID), mutate) +} + +// DeleteTerminatedInstanceRecord removes a terminated instance record. It is +// idempotent, and the bucket's TTL removes anything this misses. +func (m *JetStreamManager) DeleteTerminatedInstanceRecord(instanceID string) error { + if m.termRecords == nil { + return errors.New("terminated instance KV bucket not initialized") + } + + key := instanceRecordKey(instanceID) + if err := m.termRecords.Delete(context.Background(), key); err != nil { + return err + } + + slog.Debug("Deleted terminated instance record from JetStream KV", "key", key) + return nil +} + +// ListTerminatedInstanceRecords returns every terminated instance record. +func (m *JetStreamManager) ListTerminatedInstanceRecords() ([]*vm.InstanceRecord, error) { + if m.termRecords == nil { + return nil, errors.New("terminated instance KV bucket not initialized") + } + return listRecords(m.termRecords, InstanceRecordPrefix) +} + +// 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 +// sentinel the store uses. +func loadRecord[T any](store *kvstore.Store[T], key string) (*T, error) { + record, _, err := store.Get(context.Background(), key) + if err != nil { + if errors.Is(err, kvstore.ErrNotFound) { + return nil, nil + } + return nil, err + } + return record, nil +} diff --git a/spinifex/daemon/instance_records_test.go b/spinifex/daemon/instance_records_test.go new file mode 100644 index 000000000..a7b6e723f --- /dev/null +++ b/spinifex/daemon/instance_records_test.go @@ -0,0 +1,253 @@ +package daemon_test + +import ( + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/daemon" + "github.com/mulgadc/spinifex/spinifex/kvstore" + "github.com/mulgadc/spinifex/spinifex/resource" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/mulgadc/spinifex/spinifex/vm" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// newRecordManager returns a JetStreamManager over an embedded server with both +// 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() + _, nc, _ := testutil.StartTestJetStream(t) + + m, err := daemon.NewJetStreamManager(nc, 1) + require.NoError(t, err) + require.NoError(t, m.InitKVBucket()) + require.NoError(t, m.InitTerminatedInstanceBucket()) + return m +} + +func testRecord(id string) *vm.InstanceRecord { + return &vm.InstanceRecord{ + Metadata: resource.Metadata{Name: id, AccountID: "111122223333"}, + Spec: vm.InstanceSpec{InstanceType: "m7i.large", DesiredState: vm.DesiredStopped}, + Status: vm.InstanceStatus{Status: vm.StateStopped, LastNode: "node-1"}, + } +} + +func TestInstanceRecord_RoundTrips(t *testing.T) { + m := newRecordManager(t) + + require.NoError(t, m.WriteInstanceRecord("i-1", testRecord("i-1"))) + + got, err := m.LoadInstanceRecord("i-1") + require.NoError(t, err) + require.NotNil(t, got) + assert.Equal(t, "i-1", got.Metadata.Name) + assert.Equal(t, "111122223333", got.Metadata.AccountID) + assert.Equal(t, vm.DesiredStopped, got.Spec.DesiredState) + assert.Equal(t, "node-1", got.Status.LastNode) +} + +// An instance nobody has heard of is not an error. Every Load accessor on the +// manager answers that way, and the callers that will move onto these are +// written against it. +func TestLoadInstanceRecord_AbsentIsNil(t *testing.T) { + m := newRecordManager(t) + + got, err := m.LoadInstanceRecord("i-never-written") + + require.NoError(t, err) + 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) { + 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") + 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) + + // And deleting one leaves the other, which is what makes the dual-write + // release safe to roll back. + require.NoError(t, m.DeleteInstanceRecord("i-1")) + + stopped, err = m.ListStoppedInstances() + require.NoError(t, err) + assert.Len(t, stopped, 1, "deleting the record must not touch the key it will replace") +} + +func TestTerminatedInstanceRecords_AreDisjointFromTheKeysTheyReplace(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) + require.Len(t, terminated, 1) + assert.Equal(t, "t3.nano", terminated[0].InstanceType) + + records, err := m.ListTerminatedInstanceRecords() + require.NoError(t, err) + require.Len(t, records, 1) + assert.Equal(t, "m7i.large", records[0].Spec.InstanceType) +} + +// The two buckets are separate stores over the same prefix, so a record in one +// must not be visible through the other. +func TestInstanceRecords_TheTwoBucketsDoNotShareRecords(t *testing.T) { + m := newRecordManager(t) + + require.NoError(t, m.WriteInstanceRecord("i-1", testRecord("i-1"))) + + got, err := m.LoadTerminatedInstanceRecord("i-1") + require.NoError(t, err) + assert.Nil(t, got, "a live record must not be readable from the terminated bucket") + + terminated, err := m.ListTerminatedInstanceRecords() + require.NoError(t, err) + assert.Empty(t, terminated) +} + +func TestUpdateInstanceRecord_CommitsTheMutation(t *testing.T) { + m := newRecordManager(t) + require.NoError(t, m.WriteInstanceRecord("i-1", testRecord("i-1"))) + + updated, err := m.UpdateInstanceRecord("i-1", func(r *vm.InstanceRecord) { + r.Status.Status = vm.StateRunning + r.Status.LastNode = "node-2" + }) + + require.NoError(t, err) + assert.Equal(t, vm.StateRunning, updated.Status.Status) + + reread, err := m.LoadInstanceRecord("i-1") + require.NoError(t, err) + assert.Equal(t, "node-2", reread.Status.LastNode, "the mutation must be the committed one, not a local copy") +} + +// A mutation racing a delete must fail rather than resurrect what the delete +// removed. This is the guarantee the claim in the next change is built on. +func TestUpdateInstanceRecord_AbsentDoesNotResurrect(t *testing.T) { + m := newRecordManager(t) + + _, err := m.UpdateInstanceRecord("i-gone", func(r *vm.InstanceRecord) { + r.Status.LastNode = "node-2" + }) + + assert.ErrorIs(t, err, kvstore.ErrNotFound) + + got, err := m.LoadInstanceRecord("i-gone") + require.NoError(t, err) + assert.Nil(t, got, "a failed mutation must leave no record behind") +} + +func TestDeleteInstanceRecord_IsIdempotent(t *testing.T) { + m := newRecordManager(t) + require.NoError(t, m.WriteInstanceRecord("i-1", testRecord("i-1"))) + + require.NoError(t, m.DeleteInstanceRecord("i-1")) + require.NoError(t, m.DeleteInstanceRecord("i-1"), "deleting an absent record is not an error") + + got, err := m.LoadInstanceRecord("i-1") + require.NoError(t, err) + assert.Nil(t, got) +} + +// An empty listing is nil and no error, matching the listings these replace — +// a caller distinguishing "no instances" from "could not read" reads the error. +func TestListInstanceRecords_EmptyIsNotAnError(t *testing.T) { + m := newRecordManager(t) + + records, err := m.ListInstanceRecords() + + require.NoError(t, err) + assert.Empty(t, records) +} + +func TestListInstanceRecords_ReturnsEveryRecord(t *testing.T) { + m := newRecordManager(t) + for _, id := range []string{"i-1", "i-2", "i-3"} { + require.NoError(t, m.WriteInstanceRecord(id, testRecord(id))) + } + + records, err := m.ListInstanceRecords() + + require.NoError(t, err) + names := make([]string, 0, len(records)) + for _, r := range records { + names = append(names, r.Metadata.Name) + } + assert.ElementsMatch(t, []string{"i-1", "i-2", "i-3"}, names) +} + +// A write replaces the record wholesale rather than merging into it, so a field +// cleared by the writer stays cleared. +func TestWriteInstanceRecord_ReplacesWholesale(t *testing.T) { + m := newRecordManager(t) + require.NoError(t, m.WriteInstanceRecord("i-1", testRecord("i-1"))) + + replacement := testRecord("i-1") + replacement.Status.LastNode = "" + require.NoError(t, m.WriteInstanceRecord("i-1", replacement)) + + got, err := m.LoadInstanceRecord("i-1") + require.NoError(t, err) + assert.Empty(t, got.Status.LastNode) +} + +// The DeletionTimestamp the envelope carries has to survive the round trip, or +// the teardown path cannot tell a deleting instance from a live one. +func TestInstanceRecord_CarriesDeletionTimestamp(t *testing.T) { + m := newRecordManager(t) + record := testRecord("i-1") + deleted := time.Date(2026, 8, 28, 1, 0, 0, 0, time.UTC) + record.Metadata.DeletionTimestamp = &deleted + + require.NoError(t, m.WriteInstanceRecord("i-1", record)) + + got, err := m.LoadInstanceRecord("i-1") + require.NoError(t, err) + require.NotNil(t, got.Metadata.DeletionTimestamp) + assert.True(t, got.Metadata.MarkedForDeletion()) + assert.Equal(t, deleted, got.Metadata.DeletionTimestamp.UTC()) +} + +// Every accessor must report an unopened bucket rather than panic on a nil +// store: Tier 1 boot runs before cluster KV exists. +func TestInstanceRecordAccessors_ReportAnUnopenedBucket(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + m, err := daemon.NewJetStreamManager(nc, 1) + require.NoError(t, err) + + assert.Error(t, m.WriteInstanceRecord("i-1", testRecord("i-1"))) + assert.Error(t, m.DeleteInstanceRecord("i-1")) + assert.Error(t, m.WriteTerminatedInstanceRecord("i-1", testRecord("i-1"))) + assert.Error(t, m.DeleteTerminatedInstanceRecord("i-1")) + + _, err = m.LoadInstanceRecord("i-1") + assert.Error(t, err) + _, err = m.ListInstanceRecords() + assert.Error(t, err) + _, err = m.UpdateInstanceRecord("i-1", func(*vm.InstanceRecord) {}) + assert.Error(t, err) + _, err = m.LoadTerminatedInstanceRecord("i-1") + assert.Error(t, err) + _, err = m.ListTerminatedInstanceRecords() + assert.Error(t, err) + _, err = m.UpdateTerminatedInstanceRecord("i-1", func(*vm.InstanceRecord) {}) + assert.Error(t, err) +} diff --git a/spinifex/daemon/jetstream.go b/spinifex/daemon/jetstream.go index 1e9bf43d1..dc61205fd 100644 --- a/spinifex/daemon/jetstream.go +++ b/spinifex/daemon/jetstream.go @@ -49,19 +49,24 @@ type KVSyncObserver interface { // JetStreamManager manages JetStream KV store operations for instance state. // -// The instance-state bucket carries two record types, so it is one kvstore -// bucket with two typed views over it: they share its handle, and a recovery -// driven through either repairs both. A node's running set is stored as -// LocalState, the same envelope the local state file uses. +// The instance-state bucket carries several record types, so it is one kvstore +// bucket with a typed view per type: they share its handle, and a recovery +// driven through any of them repairs all of them. A node's running set is +// stored as LocalState, the same envelope the local state file uses. type JetStreamManager struct { js jetstream.JetStream stateB *kvstore.Bucket // spinifex-instance-state nodeState *kvstore.Store[LocalState] // node. records stopped *kvstore.Store[vm.VM] // instance. records term *kvstore.Store[vm.VM] // spinifex-terminated-instances - clusterKV jetstream.KeyValue // spinifex-cluster-state - replicas int - obs KVSyncObserver + // 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/, instance-state bucket + termRecords *kvstore.Store[vm.InstanceRecord] // i/, terminated bucket + clusterKV jetstream.KeyValue // spinifex-cluster-state + replicas int + obs KVSyncObserver } // checkNodeStateVersion reports whether a node record read from KV is one this @@ -141,12 +146,13 @@ func terminatedInstanceConfig(replicas int) kvstore.Config { } } -// setInstanceStateBucket points the two typed views at b. They share its -// handle, so a recovery driven through either repairs both. +// setInstanceStateBucket points the typed views at b. They share its handle, +// so a recovery driven through any of them repairs all of them. func (m *JetStreamManager) setInstanceStateBucket(b *kvstore.Bucket) { m.stateB = b m.nodeState = kvstore.On[LocalState](b) m.stopped = kvstore.On[vm.VM](b) + m.records = kvstore.On[vm.InstanceRecord](b) } // InitKVBucket initializes the KV bucket, creating it if it doesn't exist. @@ -194,6 +200,9 @@ func (m *JetStreamManager) InitClusterStateBucket() error { // JetStream automatically purges keys after 1 hour, matching AWS behavior for terminated instances. func (m *JetStreamManager) InitTerminatedInstanceBucket() error { m.term = kvstore.New[vm.VM](m.js, terminatedInstanceConfig(m.replicas)) + // A second view over the same handle, for the same reason the instance-state + // bucket carries several. + m.termRecords = kvstore.On[vm.InstanceRecord](m.term.Bucket) _, err := m.term.KV(context.Background()) return err } @@ -653,15 +662,15 @@ func (m *JetStreamManager) UpdateStoppedInstance(instanceID string, mutate func( if m.stopped == nil { return nil, errors.New("KV bucket not initialized") } - return mutateVM(m.stopped, StoppedInstancePrefix+instanceID, mutate) + return mutateRecord(m.stopped, StoppedInstancePrefix+instanceID, mutate) } -// mutateVM 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. -func mutateVM(store *kvstore.Store[vm.VM], key string, mutate func(*vm.VM)) (*vm.VM, error) { - var updated *vm.VM - err := store.Mutate(context.Background(), key, func(v *vm.VM) (bool, error) { +// 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. +func mutateRecord[T any](store *kvstore.Store[T], key string, mutate func(*T)) (*T, error) { + var updated *T + err := store.Mutate(context.Background(), key, func(v *T) (bool, error) { mutate(v) updated = v return true, nil @@ -677,25 +686,26 @@ func (m *JetStreamManager) ListStoppedInstances() ([]*vm.VM, error) { if m.stopped == nil { return nil, errors.New("KV bucket not initialized") } - return listVMs(m.stopped, StoppedInstancePrefix) + return listRecords(m.stopped, StoppedInstancePrefix) } -// listVMs returns every record under prefix. Unlike the raw scan it replaces it -// fails on an undecodable record rather than skipping it: both callers feed -// DescribeInstances, where a silently dropped instance reads as terminated. -func listVMs(store *kvstore.Store[vm.VM], prefix string) ([]*vm.VM, error) { +// listRecords returns every record under prefix. Unlike the raw scan it +// replaces it fails on an undecodable record rather than skipping it: these +// listings feed DescribeInstances, where a silently dropped instance reads as +// terminated. +func listRecords[T any](store *kvstore.Store[T], prefix string) ([]*T, error) { records, err := store.List(context.Background(), prefix) if err != nil { return nil, err } - instances := make([]*vm.VM, 0, len(records)) + out := make([]*T, 0, len(records)) for i := range records { - instances = append(instances, &records[i]) + out = append(out, &records[i]) } - if len(instances) == 0 { + if len(out) == 0 { return nil, nil } - return instances, nil + return out, nil } // WriteTerminatedInstance writes a terminated instance to the terminated KV @@ -730,7 +740,7 @@ func (m *JetStreamManager) UpdateTerminatedInstance(instanceID string, mutate fu if m.term == nil { return nil, errors.New("terminated instance KV bucket not initialized") } - return mutateVM(m.term, TerminatedInstancePrefix+instanceID, mutate) + return mutateRecord(m.term, TerminatedInstancePrefix+instanceID, mutate) } // ListTerminatedInstances returns all terminated instances from the terminated KV bucket. @@ -738,7 +748,7 @@ func (m *JetStreamManager) ListTerminatedInstances() ([]*vm.VM, error) { if m.term == nil { return nil, errors.New("terminated instance KV bucket not initialized") } - return listVMs(m.term, TerminatedInstancePrefix) + return listRecords(m.term, TerminatedInstancePrefix) } // DeleteTerminatedInstance removes a terminated instance from the terminated KV bucket.