From 196b7c81111263ccceae700c577e7206634b1f87 Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Thu, 27 Aug 2026 21:41:50 +1000 Subject: [PATCH] refactor: report kvstore.ErrNotFound from the StateStore interfaces (#920) --- spinifex/daemon/jetstream.go | 7 ++---- spinifex/daemon/jetstream_test.go | 24 +++++++++---------- spinifex/handlers/ec2/instance/service.go | 2 +- .../handlers/ec2/instance/service_impl.go | 12 +++++----- spinifex/vm/mock/mock.go | 7 +++--- spinifex/vm/state_store.go | 3 ++- 6 files changed, 27 insertions(+), 28 deletions(-) diff --git a/spinifex/daemon/jetstream.go b/spinifex/daemon/jetstream.go index 8e4890ed2..8f1779ed2 100644 --- a/spinifex/daemon/jetstream.go +++ b/spinifex/daemon/jetstream.go @@ -644,8 +644,8 @@ func (m *JetStreamManager) UpdateStoppedInstance(instanceID string, mutate func( } // mutateVM applies mutate under the store's CAS loop and returns the committed -// record. An absent key comes back as jetstream.ErrKeyNotFound, which is the -// sentinel the callers and both StateStore interfaces are written against. +// 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) { @@ -654,9 +654,6 @@ func mutateVM(store *kvstore.Store[vm.VM], key string, mutate func(*vm.VM)) (*vm return true, nil }) if err != nil { - if errors.Is(err, kvstore.ErrNotFound) { - return nil, jetstream.ErrKeyNotFound - } return nil, err } return updated, nil diff --git a/spinifex/daemon/jetstream_test.go b/spinifex/daemon/jetstream_test.go index 1fafb4d87..4e048617a 100644 --- a/spinifex/daemon/jetstream_test.go +++ b/spinifex/daemon/jetstream_test.go @@ -610,7 +610,7 @@ func TestJetStreamManager_UpdateStoppedInstance_ConflictRetry(t *testing.T) { } // TestJetStreamManager_UpdateStoppedInstance_NotFound verifies -// UpdateStoppedInstance surfaces jetstream.ErrKeyNotFound rather than silently +// UpdateStoppedInstance surfaces kvstore.ErrNotFound rather than silently // creating a record when nothing exists to update. func TestJetStreamManager_UpdateStoppedInstance_NotFound(t *testing.T) { nc, err := nats.Connect(sharedJSNATSURL) @@ -622,14 +622,14 @@ func TestJetStreamManager_UpdateStoppedInstance_NotFound(t *testing.T) { require.NoError(t, jsm.InitKVBucket()) _, err = jsm.UpdateStoppedInstance("i-does-not-exist", func(v *vm.VM) {}) - assert.ErrorIs(t, err, jetstream.ErrKeyNotFound) + assert.ErrorIs(t, err, kvstore.ErrNotFound) } // TestJetStreamManager_UpdateStoppedInstance_NoResurrectAfterClaim is the // core TOCTOU regression test: a claim (delete) that lands between an // UpdateStoppedInstance caller's own Load and its CAS write must not be // undone. createIfAbsent=false means the CAS write observes the key gone and -// fails with ErrKeyNotFound instead of recreating the stopped record after +// fails with kvstore.ErrNotFound instead of recreating the stopped record after // ClaimStoppedInstance already handed it off to a winning start. func TestJetStreamManager_UpdateStoppedInstance_NoResurrectAfterClaim(t *testing.T) { nc, err := nats.Connect(sharedJSNATSURL) @@ -658,7 +658,7 @@ func TestJetStreamManager_UpdateStoppedInstance_NoResurrectAfterClaim(t *testing _, err = jsm.UpdateStoppedInstance(testVM.ID, func(v *vm.VM) { v.InstanceType = "should-not-land" }) - assert.ErrorIs(t, err, jetstream.ErrKeyNotFound, "a claim that deleted the record must not be resurrected by a losing racer's update") + assert.ErrorIs(t, err, kvstore.ErrNotFound, "a claim that deleted the record must not be resurrected by a losing racer's update") stillGone, err := jsm.LoadStoppedInstance(testVM.ID) require.NoError(t, err) @@ -1394,7 +1394,7 @@ func TestJetStreamManager_UpdateTerminatedInstance_NotFound(t *testing.T) { require.NoError(t, jsm.InitTerminatedInstanceBucket()) _, err = jsm.UpdateTerminatedInstance("i-does-not-exist", func(v *vm.VM) {}) - assert.ErrorIs(t, err, jetstream.ErrKeyNotFound) + assert.ErrorIs(t, err, kvstore.ErrNotFound) } // TestJetStreamManager_WriteStoppedInstance_OverwritesConcurrentValue verifies @@ -1771,11 +1771,11 @@ func TestJetStreamManager_ListStoppedInstances_FailsOnAnUndecodableRecord(t *tes assert.ErrorContains(t, err, "i-corrupt") } -// TestJetStreamManager_UpdateStoppedInstance_AbsentKeyKeepsItsSentinel guards -// the boundary mapping. Store.Mutate reports kvstore.ErrNotFound, but three -// call sites and both StateStore interfaces are written against -// jetstream.ErrKeyNotFound, so the manager converts rather than leaking it. -func TestJetStreamManager_UpdateStoppedInstance_AbsentKeyKeepsItsSentinel(t *testing.T) { +// TestJetStreamManager_UpdateStoppedInstance_AbsentKeyIsNotFound pins the +// sentinel both StateStore interfaces are written against, for the stopped and +// terminated stores alike. Mutating an absent key must report it, not create +// the record. +func TestJetStreamManager_UpdateStoppedInstance_AbsentKeyIsNotFound(t *testing.T) { nc, err := nats.Connect(sharedJSNATSURL) require.NoError(t, err) defer nc.Close() @@ -1788,10 +1788,10 @@ func TestJetStreamManager_UpdateStoppedInstance_AbsentKeyKeepsItsSentinel(t *tes _, err = jsm.UpdateStoppedInstance("i-absent", func(*vm.VM) { t.Error("mutate must not run against an absent key") }) - require.ErrorIs(t, err, jetstream.ErrKeyNotFound) + require.ErrorIs(t, err, kvstore.ErrNotFound) _, err = jsm.UpdateTerminatedInstance("i-absent", func(*vm.VM) { t.Error("mutate must not run against an absent key") }) - require.ErrorIs(t, err, jetstream.ErrKeyNotFound) + require.ErrorIs(t, err, kvstore.ErrNotFound) } diff --git a/spinifex/handlers/ec2/instance/service.go b/spinifex/handlers/ec2/instance/service.go index 04fbb6cca..b602fb870 100644 --- a/spinifex/handlers/ec2/instance/service.go +++ b/spinifex/handlers/ec2/instance/service.go @@ -88,7 +88,7 @@ type StoppedInstanceStore interface { // UpdateStoppedInstance atomically applies mutate to the current stopped // record under optimistic concurrency (CAS) with createIfAbsent=false, so // a caller racing a winning ClaimStoppedInstance gets a clean - // jetstream.ErrKeyNotFound instead of resurrecting a deleted record. + // kvstore.ErrNotFound instead of resurrecting a deleted record. UpdateStoppedInstance(instanceID string, mutate func(*vm.VM)) (*vm.VM, error) WriteTerminatedInstance(instanceID string, instance *vm.VM) error // ClaimStoppedInstance atomically removes instanceID's record and diff --git a/spinifex/handlers/ec2/instance/service_impl.go b/spinifex/handlers/ec2/instance/service_impl.go index cdc0f0092..37f371e7b 100644 --- a/spinifex/handlers/ec2/instance/service_impl.go +++ b/spinifex/handlers/ec2/instance/service_impl.go @@ -26,13 +26,13 @@ import ( handlers_dns "github.com/mulgadc/spinifex/spinifex/handlers/dns" handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" "github.com/mulgadc/spinifex/spinifex/instancetypes" + "github.com/mulgadc/spinifex/spinifex/kvstore" "github.com/mulgadc/spinifex/spinifex/network/topology" "github.com/mulgadc/spinifex/spinifex/objectstore" spxtypes "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/mulgadc/spinifex/spinifex/vm" "github.com/nats-io/nats.go" - "github.com/nats-io/nats.go/jetstream" ) // bytesPerGiB converts the byte sizes the launch path works in to the GiB @@ -403,14 +403,14 @@ func (s *InstanceServiceImpl) TagStoppedInstance(ctx context.Context, instanceID // CAS the tag mutation into the shared KV record instead of a wholesale // WriteStoppedInstance: createIfAbsent=false means a claim that deletes // the record between the Load above and this write fails cleanly with - // jetstream.ErrKeyNotFound rather than resurrecting a stale stopped entry. + // kvstore.ErrNotFound rather than resurrecting a stale stopped entry. newTags := instance.Instance.Tags if _, err := s.stoppedStore.UpdateStoppedInstance(instanceID, func(v *vm.VM) { if v.Instance != nil { v.Instance.Tags = newTags } }); err != nil { - if errors.Is(err, jetstream.ErrKeyNotFound) { + if errors.Is(err, kvstore.ErrNotFound) { slog.WarnContext(ctx, "TagStoppedInstance: instance claimed concurrently, not resurrecting", "instanceId", instanceID) return errors.New(awserrors.ErrorInvalidInstanceIDNotFound) } @@ -455,7 +455,7 @@ func (s *InstanceServiceImpl) SetStoppedInstanceMonitoring(ctx context.Context, } v.RunInstancesInput.Monitoring.Enabled = aws.Bool(enabled) }); err != nil { - if errors.Is(err, jetstream.ErrKeyNotFound) { + if errors.Is(err, kvstore.ErrNotFound) { slog.WarnContext(ctx, "SetStoppedInstanceMonitoring: instance claimed concurrently, not resurrecting", "instanceId", instanceID) return errors.New(awserrors.ErrorInvalidInstanceIDNotFound) } @@ -2117,7 +2117,7 @@ func (s *InstanceServiceImpl) ModifyInstanceAttribute(ctx context.Context, input // CAS the mutations into the shared KV record instead of a wholesale // WriteStoppedInstance: createIfAbsent=false means a claim that deletes // the record between the Load above and this write fails cleanly with - // jetstream.ErrKeyNotFound rather than resurrecting a stale stopped entry. + // kvstore.ErrNotFound rather than resurrecting a stale stopped entry. _, err = s.stoppedStore.UpdateStoppedInstance(instanceID, func(v *vm.VM) { if input.InstanceType != nil && input.InstanceType.Value != nil && v.Instance != nil { newType := *input.InstanceType.Value @@ -2146,7 +2146,7 @@ func (s *InstanceServiceImpl) ModifyInstanceAttribute(ctx context.Context, input } }) if err != nil { - if errors.Is(err, jetstream.ErrKeyNotFound) { + if errors.Is(err, kvstore.ErrNotFound) { // The record existed moments ago (Load above) but a concurrent // claim removed it: the instance is mid-transition, not gone. slog.WarnContext(ctx, "ModifyInstanceAttribute: instance claimed concurrently, not resurrecting", "instanceId", instanceID) diff --git a/spinifex/vm/mock/mock.go b/spinifex/vm/mock/mock.go index fc19c2e0d..1633ade04 100644 --- a/spinifex/vm/mock/mock.go +++ b/spinifex/vm/mock/mock.go @@ -8,11 +8,12 @@ package mock import ( "errors" + "fmt" "maps" "sync" + "github.com/mulgadc/spinifex/spinifex/kvstore" "github.com/mulgadc/spinifex/spinifex/vm" - "github.com/nats-io/nats.go/jetstream" ) // StateStore is an in-memory vm.StateStore. Per-method error fields let a @@ -197,7 +198,7 @@ func (s *StateStore) ClaimStoppedInstance(id string) (*vm.VM, error) { // UpdateStoppedInstance mimics the real CAS semantics: mutate runs under the // store lock against the stored value, and a missing record returns -// jetstream.ErrKeyNotFound rather than resurrecting it. +// kvstore.ErrNotFound rather than resurrecting it. func (s *StateStore) UpdateStoppedInstance(id string, mutate func(*vm.VM)) (*vm.VM, error) { s.mu.Lock() defer s.mu.Unlock() @@ -207,7 +208,7 @@ func (s *StateStore) UpdateStoppedInstance(id string, mutate func(*vm.VM)) (*vm. } v, ok := s.Stopped[id] if !ok { - return nil, jetstream.ErrKeyNotFound + return nil, fmt.Errorf("%w: %s", kvstore.ErrNotFound, id) } mutate(v) return v, nil diff --git a/spinifex/vm/state_store.go b/spinifex/vm/state_store.go index 87fea96a9..c5f2d1b9b 100644 --- a/spinifex/vm/state_store.go +++ b/spinifex/vm/state_store.go @@ -32,6 +32,7 @@ type StateStore interface { // record for id and writes it back using optimistic concurrency (CAS), // so a concurrent writer to the same record (e.g. two teardown-reaper // sweeps advancing different dependents) cannot clobber the other's - // progress. Returns an error if no record exists for id. + // progress. Reports kvstore.ErrNotFound if no record exists for id, rather + // than creating one: a record the reaper already retired must stay retired. UpdateTerminatedInstance(id string, mutate func(*VM)) (*VM, error) }