mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 09:01:31 +00:00
feat: add the i/<id> instance record store (#930)
Adds the accessors for the per-resource instance record key space, over both buckets that carry instance records. Nothing calls them yet. The prefix is disjoint from the three it replaces: List and DeletePrefix match on a plain string prefix rather than on NATS subject tokens, so i/ shares a bucket with instance., node. and terminated. without colliding with any of them. mutateVM and listVMs become generic so the record accessors reuse them rather than carrying a second copy.
This commit is contained in:
Josh Siviour
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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.<id> records
|
||||
stopped *kvstore.Store[vm.VM] // instance.<id> 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/<id>, instance-state bucket
|
||||
termRecords *kvstore.Store[vm.InstanceRecord] // i/<id>, 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.
|
||||
|
||||
Reference in New Issue
Block a user