gsadmin/spinifex
1
0
Fork 0
mirror of https://github.com/mulgadc/spinifex.git synced 2026-10-21 03:36:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: report kvstore.ErrNotFound from the StateStore interfaces (#920)

Browse Source
This commit is contained in:
Josh Siviour
2026-08-27 21:41:50 +10:00
committed by GitHub
parent 5760b74bf6
commit 196b7c8111
6 changed files with 27 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
spinifex/daemon/jetstream.go
+2 -5
View File
@@ -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
spinifex/daemon/jetstream_test.go
+12 -12
View File
@@ -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)
}
spinifex/handlers/ec2/instance/service.go
+1 -1
View File
@@ -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
spinifex/handlers/ec2/instance/service_impl.go
+6 -6
View File
@@ -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)
spinifex/vm/mock/mock.go
+4 -3
View File
@@ -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
spinifex/vm/state_store.go
+2 -1
View File
@@ -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)
}