diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 58bff6741..a457ea23d 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -2603,11 +2603,6 @@ func (d *Daemon) persistState(nodeID string, vms map[string]*vm.VM) error { if err != nil { return fmt.Errorf("marshal state: %w", err) } - kvData, err := marshalInstanceState(vms) - if err != nil { - return fmt.Errorf("marshal state: %w", err) - } - // The KV write is independent of local disk health (JetStream, not this // disk), so a local failure must not skip it — attempt both, then report // the local failure. Revision only advances when the local write (the @@ -2620,12 +2615,13 @@ func (d *Daemon) persistState(nodeID string, vms map[string]*vm.VM) error { d.stateRevision.Add(1) } - // The mirror follows the blob it mirrors, so a record on the new key space - // is never newer than the running set it belongs to. vms is the caller's - // snapshot and cannot change underneath either write. + // The marker is written before the records so a reader that sees the node + // at all can only under-report it, never conclude it has no instances from + // a set that has not been written yet. vms is the caller's snapshot and + // cannot change underneath either write. if d.jsManager != nil { - d.jsManager.WriteStateBytesBestEffort(nodeID, kvData, kvSyncTimeout) - d.jsManager.MirrorRunningSet(nodeID, vms) + d.jsManager.WriteNodeMarkerBestEffort(nodeID, kvSyncTimeout) + d.jsManager.WriteRunningSet(nodeID, vms) } if localErr != nil { diff --git a/spinifex/daemon/daemon_handlers_test.go b/spinifex/daemon/daemon_handlers_test.go index e8e721f49..8318aef30 100644 --- a/spinifex/daemon/daemon_handlers_test.go +++ b/spinifex/daemon/daemon_handlers_test.go @@ -1193,20 +1193,26 @@ func TestHandleEC2StartStoppedInstance_MissingInstanceID(t *testing.T) { assert.Equal(t, awserrors.ErrorMissingParameter, errResp["Code"]) } +// Starting an instance that is not stopped is IncorrectInstanceState rather +// than NotFound, which is the distinction a caller retries on. +// +// The instance is put in the vmMgr and nowhere else on purpose. Writing a +// running instance through WriteStoppedInstance no longer constructs this: one +// key holds an instance for its whole life, so that write stamps the state it +// means instead of storing a record the listing would then decline to find. +// What is left is the race the taxonomy was written for — the stopped record is +// gone because a claim already started the instance here. func TestHandleEC2StartStoppedInstance_NotStoppedState(t *testing.T) { natsURL := sharedJSNATSURL daemon := createFullTestDaemonWithJetStream(t, natsURL) - // Write an instance in running state to shared KV - runningVM := &vm.VM{ + daemon.vmMgr.Insert(&vm.VM{ ID: "i-running-shared", Status: vm.StateRunning, InstanceType: getTestInstanceType(t), AccountID: testAccountID, - } - err := daemon.jsManager.WriteStoppedInstance(runningVM.ID, runningVM) - require.NoError(t, err) + }) sub, err := daemon.natsConn.QueueSubscribe("ec2.start", "spinifex-workers", asMsgHandler(daemon.handleEC2StartStoppedInstance)) require.NoError(t, err) @@ -1220,9 +1226,6 @@ func TestHandleEC2StartStoppedInstance_NotStoppedState(t *testing.T) { err = json.Unmarshal(reply.Data, &errResp) require.NoError(t, err) assert.Equal(t, awserrors.ErrorIncorrectInstanceState, errResp["Code"]) - - // Cleanup - _ = daemon.jsManager.DeleteStoppedInstance(runningVM.ID) } // --- handleEC2DescribeStoppedInstances tests --- diff --git a/spinifex/daemon/dns_reconcile.go b/spinifex/daemon/dns_reconcile.go index c36705bbd..bc46bfd9f 100644 --- a/spinifex/daemon/dns_reconcile.go +++ b/spinifex/daemon/dns_reconcile.go @@ -26,10 +26,13 @@ func (d *Daemon) dnsWatchSources() []reconciler.Source { } } -// instanceStateWatchFilter matches the per-node instance blobs. EC2 records are -// the one DNS input that is node-local, so a change to any node's blob is the -// only signal that this node's own VMs may need re-asserting. -const instanceStateWatchFilter = InstanceStatePrefix + "*" +// instanceStateWatchFilter matches the instance records. EC2 records are the +// one DNS input that is node-local, so a change to any instance is the only +// signal that this node's own VMs may need re-asserting. +// +// This is a NATS subject filter, so "*" spans one dot-delimited token and the +// prefix has to end in a dot for it to match anything at all. +const instanceStateWatchFilter = InstanceRecordPrefix + "*" // instanceStateWatchBuckets returns the shared instance-state bucket. It is a // fresh kvstore.Bucket rather than the JetStreamManager's handle so a recovery diff --git a/spinifex/daemon/dns_reconcile_test.go b/spinifex/daemon/dns_reconcile_test.go index 7daf1140e..207f3590f 100644 --- a/spinifex/daemon/dns_reconcile_test.go +++ b/spinifex/daemon/dns_reconcile_test.go @@ -33,7 +33,7 @@ func TestDNSWatchSources_OneSourcePerDesiredSetInput(t *testing.T) { filters = append(filters, s.Filter()) } assert.ElementsMatch(t, []string{ - InstanceStatePrefix + "*", + InstanceRecordPrefix + "*", handlers_elbv2.KeyPrefixLB + "*", ">", ">", diff --git a/spinifex/daemon/export_test.go b/spinifex/daemon/export_test.go new file mode 100644 index 000000000..f615279a4 --- /dev/null +++ b/spinifex/daemon/export_test.go @@ -0,0 +1,13 @@ +package daemon + +//test:in-package — an export_test.go is in-package by definition; that is the +//whole of what it does. The tests using these live in daemon_test. + +// The membership predicates, exported for the external test package. They are +// unexported in production because nothing outside the package should be +// deciding which set an instance is in, but they are worth testing directly: +// the key prefix used to answer this and could not be wrong, and these can. +var ( + OperatorStopped = operatorStopped + RunsOn = runsOn +) diff --git a/spinifex/daemon/instance_membership.go b/spinifex/daemon/instance_membership.go new file mode 100644 index 000000000..438c664bc --- /dev/null +++ b/spinifex/daemon/instance_membership.go @@ -0,0 +1,37 @@ +package daemon + +import "github.com/mulgadc/spinifex/spinifex/vm" + +// One key now holds an instance for its whole life, so which set it belongs to +// is a predicate over the record rather than the prefix it sits under. Before +// the cutover it was structural — a stopped instance sat at instance. and a +// running one inside node. — and could not be wrong. It can be now, so +// it is decided here once instead of at each call site. + +// operatorStopped reports whether a record is stopped in the sense +// DescribeStoppedInstances means: stopped because someone asked for it. +// +// Status alone will not do. A node's DRAIN sequence also leaves an instance in +// StateStopped, and restore relaunches those rather than listing them. +// DesiredState is what separates the two, and this is the same test +// classifyRestoredInstances already makes. +func operatorStopped(record *vm.InstanceRecord) bool { + if record == nil { + return false + } + return record.Status.Status == vm.StateStopped && + record.Spec.DesiredState == vm.DesiredStopped +} + +// runsOn reports whether a record belongs to nodeID's running set: everything +// that node last owned except what an operator stopped. +// +// StateTerminated is deliberately included. Restore migrates those to the +// terminated bucket, and a record it cannot see is one it cannot migrate — +// which is the "void" its own comment warns about. +func runsOn(record *vm.InstanceRecord, nodeID string) bool { + if record == nil || nodeID == "" { + return false + } + return record.Status.LastNode == nodeID && !operatorStopped(record) +} diff --git a/spinifex/daemon/instance_membership_test.go b/spinifex/daemon/instance_membership_test.go new file mode 100644 index 000000000..f8bf76f24 --- /dev/null +++ b/spinifex/daemon/instance_membership_test.go @@ -0,0 +1,72 @@ +package daemon_test + +import ( + "testing" + + "github.com/mulgadc/spinifex/spinifex/daemon" + "github.com/mulgadc/spinifex/spinifex/vm" + "github.com/stretchr/testify/assert" +) + +// allInstanceStates is every state an instance can be persisted in. The table +// below covers all of them on purpose: before the cutover the key an instance +// sat under decided which set it was in, so no state could put it in the wrong +// one. Now one does, and an unconsidered state is how that goes wrong quietly. +var allInstanceStates = []vm.InstanceState{ + vm.StateProvisioning, + vm.StatePending, + vm.StateRunning, + vm.StateStopping, + vm.StateStopped, + vm.StateShuttingDown, + vm.StateTerminated, + vm.StateError, +} + +func membershipRecord(status vm.InstanceState, desired vm.DesiredState, node string) *vm.InstanceRecord { + return (&vm.VM{ID: "i-1", Status: status, DesiredState: desired, LastNode: node}).Record() +} + +func TestOperatorStopped_OnlyStoppedAndAskedFor(t *testing.T) { + for _, status := range allInstanceStates { + for _, desired := range []vm.DesiredState{vm.DesiredRunning, vm.DesiredStopped} { + want := status == vm.StateStopped && desired == vm.DesiredStopped + assert.Equal(t, want, daemon.OperatorStopped(membershipRecord(status, desired, "node-1")), + "status=%s desired=%q", status, desired) + } + } +} + +// The case that makes the pair necessary rather than the status alone: a DRAIN +// stop leaves StateStopped with DesiredRunning, and restore relaunches it. Read +// as operator-stopped it would be listed as stopped and never come back. +func TestOperatorStopped_DrainStoppedIsNotOperatorStopped(t *testing.T) { + drained := membershipRecord(vm.StateStopped, vm.DesiredRunning, "node-1") + assert.False(t, daemon.OperatorStopped(drained)) + assert.True(t, daemon.RunsOn(drained, "node-1"), "a drain-stopped instance is still the node's to relaunch") +} + +func TestRunsOn_EverythingTheNodeOwnsExceptOperatorStopped(t *testing.T) { + for _, status := range allInstanceStates { + stopped := status == vm.StateStopped + assert.Equal(t, !stopped, daemon.RunsOn(membershipRecord(status, vm.DesiredStopped, "node-1"), "node-1"), + "desired-stopped, status=%s", status) + assert.True(t, daemon.RunsOn(membershipRecord(status, vm.DesiredRunning, "node-1"), "node-1"), + "desired-running, status=%s", status) + } +} + +// Terminated records stay in the running set until restore has moved them to +// the terminated bucket. Excluding them here is what its own comment calls a +// void: gone from the running set and not yet anywhere else. +func TestRunsOn_TerminatedIsStillTheNodesToMigrate(t *testing.T) { + assert.True(t, daemon.RunsOn(membershipRecord(vm.StateTerminated, vm.DesiredRunning, "node-1"), "node-1")) +} + +func TestRunsOn_AnotherNodesRecordIsNotOurs(t *testing.T) { + assert.False(t, daemon.RunsOn(membershipRecord(vm.StateRunning, vm.DesiredRunning, "node-2"), "node-1")) + assert.False(t, daemon.RunsOn(membershipRecord(vm.StateRunning, vm.DesiredRunning, ""), "node-1")) + assert.False(t, daemon.RunsOn(membershipRecord(vm.StateRunning, vm.DesiredRunning, "node-1"), "")) + assert.False(t, daemon.RunsOn(nil, "node-1")) + assert.False(t, daemon.OperatorStopped(nil)) +} diff --git a/spinifex/daemon/instance_records.go b/spinifex/daemon/instance_records.go index 316c28f27..cdd058e37 100644 --- a/spinifex/daemon/instance_records.go +++ b/spinifex/daemon/instance_records.go @@ -3,7 +3,6 @@ package daemon import ( "context" "errors" - "fmt" "log/slog" "github.com/mulgadc/spinifex/spinifex/kvstore" @@ -157,111 +156,51 @@ 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. +// writeRecord writes instance to the per-resource key space. // -// 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 +// Replace rather than Set: the write is wholesale, and doing it under CAS is +// what stops a racing update landing out of order. +func writeRecord(store *kvstore.Store[vm.InstanceRecord], instanceID string, instance *vm.VM) error { + return store.Replace(context.Background(), instanceRecordKey(instanceID), instance.Record()) } -// 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 - } +// loadInstance reads one instance from the per-resource key space, reporting an +// absent key as nil. want filters what counts as present: a key holds an +// instance for its whole life now, so "is there a record" and "is it the kind +// of instance I asked for" are different questions. +func loadInstance(store *kvstore.Store[vm.InstanceRecord], instanceID string, + want func(*vm.InstanceRecord) bool) (*vm.VM, error) { + record, err := loadRecord(store, instanceRecordKey(instanceID)) + if err != nil || record == 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 { + if want != nil && !want(record) { return nil, nil } + return vm.VMFromRecord(record), nil +} - newer, err := listRecords(records, InstanceRecordPrefix) +// listInstances returns every instance in the key space that want accepts. +func listInstances(store *kvstore.Store[vm.InstanceRecord], + want func(*vm.InstanceRecord) bool) ([]*vm.VM, error) { + records, err := listRecords(store, 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)) + out := make([]*vm.VM, 0, len(records)) + for _, record := range records { + if want != nil && !want(record) { continue } - out = append(out, instance) + out = append(out, vm.VMFromRecord(record)) + } + if len(out) == 0 { + return nil, nil } 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 index c06e1c80b..d5759f3e8 100644 --- a/spinifex/daemon/instance_records_migrate.go +++ b/spinifex/daemon/instance_records_migrate.go @@ -7,11 +7,16 @@ import ( "fmt" "strings" + "github.com/mulgadc/spinifex/spinifex/kvutil" "github.com/mulgadc/spinifex/spinifex/migrate" "github.com/mulgadc/spinifex/spinifex/vm" "github.com/nats-io/nats.go/jetstream" ) +// errRecordAbsent marks a blob member with no record left to own, so the scan +// can tell that apart from a read that failed. +var errRecordAbsent = errors.New("instance record absent") + // The copy-forward onto the per-resource key space. 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 @@ -44,6 +49,12 @@ func init() { Description: "re-key instance records from i/ to i.", Run: rekeyRecordSeparator, }) + migrate.DefaultRegistry.RegisterKV(InstanceStateBucket, migrate.KVMigration{ + FromVersion: 4, + ToVersion: 5, + Description: "carry each node's presence and ownership out of its running-set blob", + Run: carryNodeOwnershipForward, + }) migrate.DefaultRegistry.RegisterKV(TerminatedInstanceBucket, migrate.KVMigration{ FromVersion: 1, ToVersion: 2, @@ -111,6 +122,123 @@ func rekeyRecordSeparator(ctx context.Context, kvc migrate.KVContext) error { return nil } +// carryNodeOwnershipForward moves the two things the node. key carried in +// its name into the key space that replaces it: that the node exists, and which +// instances are its. +// +// The presence marker comes first. Without it a node reads no marker until its +// own first state write, and a node that boots before it writes would be told +// the cluster has no record of it. That is the safe direction — restore declines +// to adopt rather than adopting an empty set — but it also means a node cannot +// recover its instances from KV across the crossing, which is the one time it +// most needs to. +// +// The blob is read, not consumed. It is what a node rolled back to the previous +// release reads, and this is the step that stops being reversible if it moves. +func carryNodeOwnershipForward(ctx context.Context, kvc migrate.KVContext) 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) + } + + marker, err := json.Marshal(LocalState{SchemaVersion: LocalStateSchemaVersion}) + if err != nil { + return fmt.Errorf("encode presence marker: %w", err) + } + + seeded, owned := 0, 0 + for _, key := range keys { + if !strings.HasPrefix(key, InstanceStatePrefix) { + continue + } + nodeID := strings.TrimPrefix(key, InstanceStatePrefix) + + if _, err := kvc.KV.Create(ctx, NodePresencePrefix+nodeID, marker); err != nil { + // A node that has already upgraded wrote its own marker; it is the + // same value, and the newer write is not overwritten with this one. + if !errors.Is(err, jetstream.ErrKeyExists) { + return fmt.Errorf("write %s: %w", NodePresencePrefix+nodeID, err) + } + } else { + seeded++ + } + + stamped, err := stampBlobOwnership(ctx, kvc, key, nodeID) + if err != nil { + return err + } + owned += stamped + } + + kvc.Logger.Info("Carried node presence and ownership onto the record key space", + "markersSeeded", seeded, "recordsStamped", owned) + return nil +} + +// stampBlobOwnership names the owner of every instance in a node's blob on the +// record that replaces it, reporting how many it had to write. +// +// This is the half of the split that is easy to miss. Which node ran an instance +// used to be the name of the key its blob sat under, so nothing wrote it into +// the instance, and the records copied forward from those blobs carry no owner +// at all. After the cutover ownership is read off the record, and an unowned +// record is one no node lists — so the node that boots next reads a marker +// saying the cluster knows it and a running set saying it runs nothing, and +// adopts that over the instances it actually has. +func stampBlobOwnership(ctx context.Context, kvc migrate.KVContext, key, nodeID string) (int, error) { + entry, err := kvc.KV.Get(ctx, key) + if err != nil { + if errors.Is(err, jetstream.ErrKeyNotFound) { + return 0, nil + } + return 0, fmt.Errorf("read %s: %w", key, err) + } + + var state LocalState + if err := json.Unmarshal(entry.Value(), &state); err != nil { + return 0, fmt.Errorf("decode %s: %w", key, err) + } + if err := checkNodeStateVersion(key, state.SchemaVersion); err != nil { + return 0, err + } + + stamped := 0 + for id := range state.VMS { + recordKey := instanceRecordKey(id) + wrote := false + _, err := kvutil.Update(ctx, kvc.KV, recordKey, + kvutil.CASConfig{NotFound: errRecordAbsent}, + func(record *vm.InstanceRecord) (bool, error) { + wrote = false + // An owner already on the record is the newer fact, whoever it + // names: this one is reconstructed from a blob that stopped + // being written the moment the cutover landed. + if record.Status.LastNode != "" { + return false, nil + } + record.Status.LastNode = nodeID + wrote = true + return true, nil + }) + if err != nil { + // No record to own. The split that creates them has already run, so + // what is missing here was deleted afterwards, and the delete is the + // newer fact. + if errors.Is(err, errRecordAbsent) { + continue + } + return 0, fmt.Errorf("stamp owner on %s: %w", recordKey, err) + } + if wrote { + stamped++ + } + } + return stamped, nil +} + // copyInstancesForward writes a record for every key under prefix. // // Two nodes can reach this at once — the version stamp is a read-then-write, diff --git a/spinifex/daemon/instance_records_mirror.go b/spinifex/daemon/instance_records_mirror.go deleted file mode 100644 index 581fbf504..000000000 --- a/spinifex/daemon/instance_records_mirror.go +++ /dev/null @@ -1,176 +0,0 @@ -package daemon - -import ( - "context" - "encoding/json" - "hash/fnv" - "log/slog" - "sync" - - "github.com/mulgadc/spinifex/spinifex/vm" -) - -// mirrorState tracks what this node last put on the per-resource key space, so -// a change to one instance writes one record. -// -// Without it every state change would rewrite the whole running set one key at -// a time, which is worse than the single blob it replaces rather than better: -// splitting the blob exists to stop two instances on a node serialising against -// each other. -type mirrorState struct { - mu sync.Mutex - seeded bool - digest map[string]uint64 -} - -// MirrorRunningSet reconciles this node's running instances onto the -// per-resource key space, after the blob it mirrors has been written. -// -// Best-effort, like that blob write: the local state file is the source of -// truth, the blob still decides which instances are running here, and a read -// that finds no mirror answers from the blob member. So a failure degrades a -// read rather than breaking one, and is logged rather than returned. -func (m *JetStreamManager) MirrorRunningSet(nodeID string, vms map[string]*vm.VM) { - if m.records == nil || m.stopped == nil { - return - } - - m.mirror.mu.Lock() - defer m.mirror.mu.Unlock() - - if !m.mirror.seeded { - if err := m.seedRunningMirror(nodeID); err != nil { - slog.Warn("Could not read the existing instance records; the next state write retries", - "node", nodeID, "err", err) - return - } - } - - for id, instance := range vms { - if instance == nil { - continue - } - sum, err := recordDigest(instance.Record()) - if err != nil { - slog.Error("Could not encode an instance record", "instanceId", id, "err", err) - continue - } - if current, ok := m.mirror.digest[id]; ok && current == sum { - continue - } - - // A failed mirror forgets its digest as well as its record, so the next - // state write repeats the attempt rather than reading the failure as - // already mirrored. - if err := mirrorRecord(m.records, id, instance); err != nil { - slog.Error("Could not mirror an instance record", "instanceId", id, "err", err) - delete(m.mirror.digest, id) - continue - } - m.mirror.digest[id] = sum - } - - for id := range m.mirror.digest { - if _, ok := vms[id]; ok { - continue - } - if m.retireMirror(id) { - delete(m.mirror.digest, id) - } - } -} - -// retireMirror drops the record of an instance that has left this node's -// running set, reporting whether the question is settled. -// -// It is not settled by deleting unconditionally. The stopped key space shares -// the record key, and an instance that stops leaves the running set as -// WriteStoppedInstance mirrors it there, so an unconditional delete would race -// a live write — and would keep deleting, on every state change, the records of -// instances stopped on this node. Where the stopped key holds the instance the -// record is handed over rather than retired, which is also what it becomes at -// the cutover, when the two key spaces are one. -func (m *JetStreamManager) retireMirror(instanceID string) bool { - stopped, err := loadRecord(m.stopped, StoppedInstancePrefix+instanceID) - if err != nil { - slog.Warn("Could not tell whether a departed instance is stopped; leaving its record in place", - "instanceId", instanceID, "err", err) - return false - } - if stopped != nil { - return true - } - - if err := m.records.Delete(context.Background(), instanceRecordKey(instanceID)); err != nil { - slog.Error("Could not remove the record of an instance that left this node", - "instanceId", instanceID, "err", err) - return false - } - return true -} - -// seedRunningMirror primes the digest map from the records already on the key -// space for this node, so the first state write after boot rewrites only what -// actually changed while the node was down — and so a record left behind by a -// previous process is still a candidate for retirement. -// -// Ownership is read from status.LastNode, the same field the GC uses to decide -// whose records a node may touch. -func (m *JetStreamManager) seedRunningMirror(nodeID string) error { - existing, err := listRecords(m.records, InstanceRecordPrefix) - if err != nil { - return err - } - - digest := make(map[string]uint64, len(existing)) - for _, record := range existing { - if record.Status.LastNode != nodeID { - continue - } - sum, err := recordDigest(record) - if err != nil { - return err - } - digest[record.Metadata.Name] = sum - } - - m.mirror.digest = digest - m.mirror.seeded = true - slog.Debug("Seeded the instance record mirror", "node", nodeID, "records", len(digest)) - return nil -} - -// recordDigest identifies a record by its wire form, so a state change that -// leaves an instance untouched does not become a KV write. encoding/json sorts -// map keys, so the same record always encodes to the same bytes. -func recordDigest(record *vm.InstanceRecord) (uint64, error) { - data, err := json.Marshal(record) - if err != nil { - return 0, err - } - h := fnv.New64a() - _, _ = h.Write(data) - return h.Sum64(), nil -} - -// answerFromMirrors replaces each member of a node's running set with its -// mirror where there is one. The blob still decides membership — see -// loadPreferringRecord for why a mirror cannot be trusted to — so a record with -// no member behind it is ignored here. -func (m *JetStreamManager) answerFromMirrors(vms map[string]*vm.VM) error { - if m.records == nil || len(vms) == 0 { - return nil - } - - mirrored, err := listRecords(m.records, InstanceRecordPrefix) - if err != nil { - return err - } - for _, record := range mirrored { - if _, ok := vms[record.Metadata.Name]; !ok { - continue - } - vms[record.Metadata.Name] = vm.VMFromRecord(record) - } - return nil -} diff --git a/spinifex/daemon/instance_records_mirror_test.go b/spinifex/daemon/instance_records_mirror_test.go deleted file mode 100644 index 19f53d490..000000000 --- a/spinifex/daemon/instance_records_mirror_test.go +++ /dev/null @@ -1,314 +0,0 @@ -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" -) - -const mirrorNode = "node-1" - -func runningVM(id, instanceType string) *vm.VM { - return &vm.VM{ID: id, InstanceType: instanceType, Status: vm.StateRunning, LastNode: mirrorNode} -} - -func runningSet(vms ...*vm.VM) map[string]*vm.VM { - set := make(map[string]*vm.VM, len(vms)) - for _, v := range vms { - set[v.ID] = v - } - return set -} - -// writeRunningSet does what persistState does: commit the blob, then mirror -// what it committed. -func writeRunningSet(t *testing.T, m *daemon.JetStreamManager, vms ...*vm.VM) { - t.Helper() - require.NoError(t, m.WriteState(mirrorNode, runningSet(vms...))) -} - -// recordRevision reads the revision of a record straight from the bucket, so a -// test can tell a write that did nothing from one that did not happen. -func recordRevision(t *testing.T, nc *nats.Conn, instanceID string) uint64 { - t.Helper() - js, err := jetstream.New(nc) - require.NoError(t, err) - kv, err := js.KeyValue(context.Background(), daemon.InstanceStateBucket) - require.NoError(t, err) - entry, err := kv.Get(context.Background(), daemon.InstanceRecordPrefix+instanceID) - require.NoError(t, err) - return entry.Revision() -} - -func TestMirrorRunningSet_WritesARecordPerMember(t *testing.T) { - m := newRecordManager(t) - - writeRunningSet(t, m, runningVM("i-1", "t3.nano"), runningVM("i-2", "t3.micro")) - - records, err := m.ListInstanceRecords() - require.NoError(t, err) - byID := make(map[string]string, len(records)) - for _, record := range records { - byID[record.Metadata.Name] = record.Spec.InstanceType - } - assert.Equal(t, map[string]string{"i-1": "t3.nano", "i-2": "t3.micro"}, byID, - "one blob of two instances must become two records") -} - -func TestLoadState_AnswersFromTheMirrorWhereThereIsOne(t *testing.T) { - m := newRecordManager(t) - require.NoError(t, m.WriteState(mirrorNode, runningSet(runningVM("i-1", "t3.nano")))) - - // Written directly rather than through the mirror, standing in for a record - // committed after the blob it belongs to. - require.NoError(t, m.WriteInstanceRecord("i-1", runningVM("i-1", "m7i.large").Record())) - - loaded, found, err := m.LoadState(mirrorNode) - - require.NoError(t, err) - require.True(t, found) - require.Contains(t, loaded, "i-1") - assert.Equal(t, "m7i.large", loaded["i-1"].InstanceType, "the record must win over the blob member") -} - -func TestLoadState_FallsBackToTheBlobMember(t *testing.T) { - m := newRecordManager(t) - require.NoError(t, m.WriteState(mirrorNode, runningSet(runningVM("i-1", "t3.nano")))) - - loaded, found, err := m.LoadState(mirrorNode) - - require.NoError(t, err) - require.True(t, found) - require.Contains(t, loaded, "i-1", "an instance only the blob holds must still be found") - assert.Equal(t, "t3.nano", loaded["i-1"].InstanceType) -} - -// The blob decides which instances are running here. A record with no member -// behind it is what a node predating the key space leaves when it stops an -// instance, and reading it as running is the fault env19 found. -func TestLoadState_IgnoresARecordWithNoBlobMember(t *testing.T) { - m := newRecordManager(t) - require.NoError(t, m.WriteState(mirrorNode, runningSet(runningVM("i-1", "t3.nano")))) - require.NoError(t, m.WriteInstanceRecord("i-orphan", runningVM("i-orphan", "m7i.large").Record())) - - loaded, found, err := m.LoadState(mirrorNode) - - require.NoError(t, err) - require.True(t, found) - assert.NotContains(t, loaded, "i-orphan") - assert.Len(t, loaded, 1) -} - -// The conversion's totality is guarded in the vm package; what this adds is -// that the running-set path actually routes through it, on fields from both -// halves of the record rather than the instance type alone. -func TestMirrorRunningSet_CarriesBothHalvesOfTheRecord(t *testing.T) { - m := newRecordManager(t) - instance := runningVM("i-1", "t3.nano") - instance.DesiredState = vm.DesiredRunning - instance.PublicIP = "10.0.0.9" - instance.ENIId = "eni-1" - instance.HostfwdPorts = []int{2222} - - writeRunningSet(t, m, instance) - - loaded, _, err := m.LoadState(mirrorNode) - require.NoError(t, err) - require.Contains(t, loaded, "i-1") - got := loaded["i-1"] - assert.Equal(t, vm.DesiredRunning, got.DesiredState) - assert.Equal(t, []int{2222}, got.HostfwdPorts) - assert.Equal(t, "10.0.0.9", got.PublicIP) - assert.Equal(t, "eni-1", got.ENIId) - assert.Equal(t, mirrorNode, got.LastNode) -} - -func TestMirrorRunningSet_RetiresAMemberThatLeft(t *testing.T) { - m := newRecordManager(t) - writeRunningSet(t, m, runningVM("i-1", "t3.nano"), runningVM("i-2", "t3.micro")) - - writeRunningSet(t, m, runningVM("i-1", "t3.nano")) - - gone, err := m.LoadInstanceRecord("i-2") - require.NoError(t, err) - assert.Nil(t, gone, "a record left behind by a terminated instance never expires from this bucket") - - kept, err := m.LoadInstanceRecord("i-1") - require.NoError(t, err) - assert.NotNil(t, kept) -} - -// An instance that stops leaves the running set at the moment the stopped key -// space takes it over, and that path has already written the record. Retiring -// it here would race a live write and would keep re-deleting the records of -// every instance ever stopped on this node. -func TestMirrorRunningSet_HandsOverAMemberThatStopped(t *testing.T) { - m := newRecordManager(t) - writeRunningSet(t, m, runningVM("i-1", "t3.nano")) - - require.NoError(t, m.WriteStoppedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano", Status: vm.StateStopped})) - writeRunningSet(t, m) - - record, err := m.LoadInstanceRecord("i-1") - require.NoError(t, err) - require.NotNil(t, record, "the stopped key space owns this record now") - - stopped, err := m.ListStoppedInstances() - require.NoError(t, err) - require.Len(t, stopped, 1) - assert.Equal(t, "i-1", stopped[0].ID) -} - -// Rewriting every record on every state change would reintroduce, one key at a -// time, the contention splitting the blob exists to remove. -func TestMirrorRunningSet_WritesOnlyWhatChanged(t *testing.T) { - m, nc := newRecordManagerConn(t) - writeRunningSet(t, m, runningVM("i-1", "t3.nano"), runningVM("i-2", "t3.micro")) - before1, before2 := recordRevision(t, nc, "i-1"), recordRevision(t, nc, "i-2") - - writeRunningSet(t, m, runningVM("i-1", "t3.nano"), runningVM("i-2", "t3.micro")) - - assert.Equal(t, before1, recordRevision(t, nc, "i-1"), "an unchanged instance must not be rewritten") - assert.Equal(t, before2, recordRevision(t, nc, "i-2")) - - changed := runningVM("i-1", "t3.nano") - changed.PublicIP = "10.0.0.9" - writeRunningSet(t, m, changed, runningVM("i-2", "t3.micro")) - - assert.Greater(t, recordRevision(t, nc, "i-1"), before1, "a changed instance must be rewritten") - assert.Equal(t, before2, recordRevision(t, nc, "i-2"), "and only that instance") -} - -// A restarted process has no memory of what it last wrote, so it primes itself -// from the bucket rather than rewriting every record to find out. -func TestMirrorRunningSet_SeedsFromWhatIsAlreadyThere(t *testing.T) { - first, nc := newRecordManagerConn(t) - writeRunningSet(t, first, runningVM("i-1", "t3.nano")) - before := recordRevision(t, nc, "i-1") - - second, err := daemon.NewJetStreamManager(nc, 1) - require.NoError(t, err) - require.NoError(t, second.InitKVBucket()) - writeRunningSet(t, second, runningVM("i-1", "t3.nano")) - - assert.Equal(t, before, recordRevision(t, nc, "i-1")) -} - -// Ownership is read from status.LastNode, the same field the GC uses. A node -// must not retire a record for an instance that is running somewhere else. -func TestMirrorRunningSet_LeavesAnotherNodesRecordsAlone(t *testing.T) { - m := newRecordManager(t) - elsewhere := &vm.VM{ID: "i-elsewhere", InstanceType: "t3.nano", Status: vm.StateRunning, LastNode: "node-2"} - require.NoError(t, m.WriteInstanceRecord("i-elsewhere", elsewhere.Record())) - - writeRunningSet(t, m, runningVM("i-1", "t3.nano")) - - record, err := m.LoadInstanceRecord("i-elsewhere") - require.NoError(t, err) - assert.NotNil(t, record) -} - -// seedNodeBlobBucket creates the instance-state bucket at the version before -// the blob split, holding one node's running set as a single record. -func seedNodeBlobBucket(t *testing.T, nc *nats.Conn, nodeID string, vms map[string]*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: daemon.InstanceStateBucket, History: 1}) - require.NoError(t, err) - - data, err := json.Marshal(daemon.LocalState{SchemaVersion: daemon.LocalStateSchemaVersion, VMS: vms}) - require.NoError(t, err) - _, err = kv.Put(ctx, daemon.InstanceStatePrefix+nodeID, data) - require.NoError(t, err) - - require.NoError(t, kvutil.WriteVersion(ctx, kv, 2)) - return kv -} - -func TestInstanceStateMigration_SplitsTheNodeBlob(t *testing.T) { - _, nc, _ := testutil.StartTestJetStream(t) - seedNodeBlobBucket(t, nc, mirrorNode, runningSet(runningVM("i-1", "t3.nano"), runningVM("i-2", "t3.micro"))) - - m, err := daemon.NewJetStreamManager(nc, 1) - require.NoError(t, err) - require.NoError(t, m.InitKVBucket()) - - records, err := m.ListInstanceRecords() - require.NoError(t, err) - byID := make(map[string]string, len(records)) - for _, record := range records { - byID[record.Metadata.Name] = record.Spec.InstanceType - } - assert.Equal(t, map[string]string{"i-1": "t3.nano", "i-2": "t3.micro"}, byID) -} - -// Release 1 keeps writing the blob, so the split must copy it rather than -// consume it: a node still on the previous release reads nothing else. -func TestInstanceStateMigration_KeepsTheNodeBlob(t *testing.T) { - _, nc, _ := testutil.StartTestJetStream(t) - seedNodeBlobBucket(t, nc, mirrorNode, runningSet(runningVM("i-1", "t3.nano"))) - - m, err := daemon.NewJetStreamManager(nc, 1) - require.NoError(t, err) - require.NoError(t, m.InitKVBucket()) - - loaded, found, err := m.LoadState(mirrorNode) - require.NoError(t, err) - require.True(t, found) - require.Contains(t, loaded, "i-1") -} - -// Every node's set is copied, not just the one running the migration: it runs -// once per bucket, and a node that is down would otherwise have no records -// until it came back. -func TestInstanceStateMigration_SplitsEveryNodesBlob(t *testing.T) { - _, nc, _ := testutil.StartTestJetStream(t) - kv := seedNodeBlobBucket(t, nc, mirrorNode, runningSet(runningVM("i-1", "t3.nano"))) - - other := &vm.VM{ID: "i-2", InstanceType: "t3.micro", Status: vm.StateRunning, LastNode: "node-2"} - data, err := json.Marshal(daemon.LocalState{ - SchemaVersion: daemon.LocalStateSchemaVersion, - VMS: map[string]*vm.VM{other.ID: other}, - }) - require.NoError(t, err) - _, err = kv.Put(context.Background(), daemon.InstanceStatePrefix+"node-2", data) - require.NoError(t, err) - - m, err := daemon.NewJetStreamManager(nc, 1) - require.NoError(t, err) - require.NoError(t, m.InitKVBucket()) - - record, err := m.LoadInstanceRecord("i-2") - require.NoError(t, err) - require.NotNil(t, record) - assert.Equal(t, "node-2", record.Status.LastNode) -} - -func TestInstanceStateMigration_BlobSplitIsSafeToRunTwice(t *testing.T) { - _, nc, _ := testutil.StartTestJetStream(t) - kv := seedNodeBlobBucket(t, nc, mirrorNode, runningSet(runningVM("i-1", "t3.nano"))) - - first, err := daemon.NewJetStreamManager(nc, 1) - require.NoError(t, err) - require.NoError(t, first.InitKVBucket()) - before := recordRevision(t, nc, "i-1") - - require.NoError(t, kvutil.WriteVersion(context.Background(), kv, 2)) - second, err := daemon.NewJetStreamManager(nc, 1) - require.NoError(t, err) - require.NoError(t, second.InitKVBucket()) - - assert.Equal(t, before, recordRevision(t, nc, "i-1"), "a second run must not overwrite what the first copied") -} diff --git a/spinifex/daemon/instance_records_test.go b/spinifex/daemon/instance_records_test.go index 732b9601c..2e880a125 100644 --- a/spinifex/daemon/instance_records_test.go +++ b/spinifex/daemon/instance_records_test.go @@ -70,36 +70,31 @@ func TestLoadInstanceRecord_AbsentIsNil(t *testing.T) { assert.Nil(t, got) } -// 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) { +// A stopped instance is written once, to the record key, and read back from +// there through both the instance accessors and the record accessors. +func TestStoppedInstances_LiveAtTheRecordKey(t *testing.T) { m := newRecordManager(t) require.NoError(t, m.WriteStoppedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"})) stopped, err := m.ListStoppedInstances() require.NoError(t, err) - require.Len(t, stopped, 1, "an instance held at both keys must be listed once") + require.Len(t, stopped, 1, "an instance 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 write must have been mirrored onto the record key") + require.Len(t, records, 1) assert.Equal(t, "t3.nano", records[0].Spec.InstanceType) - // 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")) got, err := m.LoadStoppedInstance("i-1") require.NoError(t, err) - require.NotNil(t, got, "deleting the mirror must not touch the key it mirrors") - assert.Equal(t, "t3.nano", got.InstanceType) + assert.Nil(t, got, "there is one key, so removing it removes the instance") } -func TestTerminatedInstances_SpanBothKeySpaces(t *testing.T) { +func TestTerminatedInstances_LiveAtTheRecordKey(t *testing.T) { m := newRecordManager(t) require.NoError(t, m.WriteTerminatedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"})) @@ -118,8 +113,26 @@ func TestTerminatedInstances_SpanBothKeySpaces(t *testing.T) { got, err := m.LoadTerminatedInstance("i-1") require.NoError(t, err) - require.NotNil(t, got) - assert.Equal(t, "t3.nano", got.InstanceType) + assert.Nil(t, got) +} + +// Four key spaces share the instance-state bucket and a listing selects between +// them by string prefix, not by subject token. "i." is a prefix of nothing else +// here and nothing else is a prefix of it, which is the whole of what keeps a +// record listing from picking up a frozen key or a node's marker. +func TestInstanceRecords_ThePrefixesInOneBucketAreDisjoint(t *testing.T) { + m, nc := newRecordManagerConn(t) + + require.NoError(t, m.WriteInstanceRecord("i-1", testRecord("i-1"))) + require.NoError(t, m.WriteNodeMarker("i-1")) + seedFrozenKey(t, nc, daemon.InstanceStateBucket, daemon.StoppedInstancePrefix+"i-1") + seedFrozenKey(t, nc, daemon.InstanceStateBucket, daemon.InstanceStatePrefix+"i-1") + + records, err := m.ListInstanceRecords() + + require.NoError(t, err) + require.Len(t, records, 1, "only the record key may answer a record listing") + assert.Equal(t, "i-1", records[0].Metadata.Name) } // 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 index 05a58cc34..c3e97d1ba 100644 --- a/spinifex/daemon/instance_records_transition_test.go +++ b/spinifex/daemon/instance_records_transition_test.go @@ -15,108 +15,32 @@ import ( "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) { +// seedFrozenKey writes raw bytes at one of the key spaces the record replaced. +// Nothing does this any more, which is the point: it stages what a cluster that +// crossed the cutover still has lying around. +func seedFrozenKey(t *testing.T, nc *nats.Conn, bucket, key string) { t.Helper() js, err := jetstream.New(nc) require.NoError(t, err) - kv, err := js.KeyValue(context.Background(), daemon.InstanceStateBucket) + kv, err := js.KeyValue(context.Background(), bucket) + require.NoError(t, err) + _, err = kv.Put(context.Background(), key, []byte(`{"id":"i-1"}`)) 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") +// frozenKeyExists reports whether a key in one of the replaced key spaces is +// still there. +func frozenKeyExists(t *testing.T, nc *nats.Conn, bucket, key string) bool { + t.Helper() + js, err := jetstream.New(nc) require.NoError(t, err) - require.NotNil(t, record, "the older node cannot have removed a key it does not know") - - stopped, err := m.ListStoppedInstances() + kv, err := js.KeyValue(context.Background(), bucket) 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") + _, err = kv.Get(context.Background(), key) + return err == nil } -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) { +func TestUpdateStoppedInstance_CommitsToTheRecord(t *testing.T) { m := newRecordManager(t) require.NoError(t, m.WriteStoppedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"})) @@ -126,12 +50,12 @@ func TestUpdateStoppedInstance_MirrorsTheCommittedValue(t *testing.T) { 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") + assert.Equal(t, "node-2", record.Status.LastNode, "the record 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) { +// A mutation that finds nothing to mutate must not create it. The record would +// then hold an instance nothing ever stopped. +func TestUpdateStoppedInstance_AbsentCreatesNothing(t *testing.T) { m := newRecordManager(t) _, err := m.UpdateStoppedInstance("i-gone", func(v *vm.VM) { v.LastNode = "node-2" }) @@ -142,25 +66,33 @@ func TestUpdateStoppedInstance_AbsentLeavesNoMirror(t *testing.T) { assert.Nil(t, record) } -func TestDeleteStoppedInstance_ClearsBothKeySpaces(t *testing.T) { - m := newRecordManager(t) +// The delete drains the key the record replaced as well as the record. The +// frozen space is kept so the cutover can be rolled back, but an instance that +// is gone should not be recoverable from it, and draining on this path is what +// stops it growing without bound. +func TestDeleteStoppedInstance_DrainsTheFrozenKey(t *testing.T) { + m, nc := newRecordManagerConn(t) require.NoError(t, m.WriteStoppedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"})) + seedFrozenKey(t, nc, daemon.InstanceStateBucket, daemon.StoppedInstancePrefix+"i-1") 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") + assert.Nil(t, record) stopped, err := m.ListStoppedInstances() require.NoError(t, err) assert.Empty(t, stopped) + + assert.False(t, frozenKeyExists(t, nc, daemon.InstanceStateBucket, daemon.StoppedInstancePrefix+"i-1"), + "the key the record replaced must not outlive the instance") } -// 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) { +// Exclusivity is a compare-and-set on the record now rather than a delete of +// it. One key holds an instance for its whole life, so a claim that deleted +// would be deleting the instance it is about to start. +func TestClaimStoppedInstance_OnlyOneCallerWins(t *testing.T) { m := newRecordManager(t) require.NoError(t, m.WriteStoppedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"})) @@ -169,19 +101,58 @@ func TestClaimStoppedInstance_ClearsTheMirror(t *testing.T) { 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) { +// What the winner changes is the intent, not the key. Clearing DesiredStopped +// is what makes the record no longer claimable and no longer stopped, and it is +// the same edit that stops it reading back as an instance nobody started. +func TestClaimStoppedInstance_KeepsTheRecordAndClearsTheStopIntent(t *testing.T) { + m := newRecordManager(t) + require.NoError(t, m.WriteStoppedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"})) + + _, err := m.ClaimStoppedInstance("i-1") + require.NoError(t, err) + + record, err := m.LoadInstanceRecord("i-1") + require.NoError(t, err) + require.NotNil(t, record, "the claim moves the record between sets, it does not delete it") + assert.Equal(t, vm.DesiredRunning, record.Spec.DesiredState) + + stopped, err := m.ListStoppedInstances() + require.NoError(t, err) + assert.Empty(t, stopped, "a claimed instance must not keep reading back as stopped") + + got, err := m.LoadStoppedInstance("i-1") + require.NoError(t, err) + assert.Nil(t, got) +} + +// LastNode is left alone deliberately. The claimant has not run the instance +// yet, and the node that last did is the one that should recover it if the +// launch that follows this claim never happens. +func TestClaimStoppedInstance_LeavesOwnershipWithTheLastNodeToRunIt(t *testing.T) { + m := newRecordManager(t) + require.NoError(t, m.WriteStoppedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano", LastNode: "node-2"})) + + _, err := m.ClaimStoppedInstance("i-1") + 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) +} + +func TestClaimStoppedInstance_AbsentIsNotClaimable(t *testing.T) { + m := newRecordManager(t) + + _, err := m.ClaimStoppedInstance("i-never-existed") + assert.ErrorIs(t, err, vm.ErrStoppedInstanceClaimed) +} + +func TestWriteTerminatedInstance_WritesTheRecord(t *testing.T) { m := newRecordManager(t) require.NoError(t, m.WriteTerminatedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"})) @@ -192,7 +163,7 @@ func TestWriteTerminatedInstance_MirrorsOntoTheRecordKey(t *testing.T) { assert.Equal(t, "t3.nano", record.Spec.InstanceType) } -func TestUpdateTerminatedInstance_MirrorsTheCommittedValue(t *testing.T) { +func TestUpdateTerminatedInstance_CommitsToTheRecord(t *testing.T) { m := newRecordManager(t) require.NoError(t, m.WriteTerminatedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"})) @@ -207,29 +178,11 @@ func TestUpdateTerminatedInstance_MirrorsTheCommittedValue(t *testing.T) { 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) +func TestDeleteTerminatedInstance_DrainsTheFrozenKey(t *testing.T) { + m, nc := newRecordManagerConn(t) + require.NoError(t, m.InitTerminatedInstanceBucket()) require.NoError(t, m.WriteTerminatedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano"})) + seedFrozenKey(t, nc, daemon.TerminatedInstanceBucket, daemon.TerminatedInstancePrefix+"i-1") require.NoError(t, m.DeleteTerminatedInstance("i-1")) @@ -240,18 +193,8 @@ func TestDeleteTerminatedInstance_ClearsBothKeySpaces(t *testing.T) { 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) + assert.False(t, frozenKeyExists(t, nc, daemon.TerminatedInstanceBucket, daemon.TerminatedInstancePrefix+"i-1")) } // seedLegacyBucket creates bucket at schema version 1 holding instance at the diff --git a/spinifex/daemon/instance_running_set.go b/spinifex/daemon/instance_running_set.go new file mode 100644 index 000000000..67668f17b --- /dev/null +++ b/spinifex/daemon/instance_running_set.go @@ -0,0 +1,185 @@ +package daemon + +import ( + "context" + "encoding/json" + "hash/fnv" + "log/slog" + "sync" + + "github.com/mulgadc/spinifex/spinifex/vm" +) + +// runningSetState tracks what this node last wrote to the per-resource key +// space, so a change to one instance writes one record. +// +// Without it every state change would rewrite the whole running set one key at +// a time, which is worse than the single blob it replaces rather than better: +// splitting the blob exists to stop two instances on a node serialising against +// each other. +type runningSetState struct { + mu sync.Mutex + // seededFor is the node the digest describes, not a bool: the digest is + // per-node state and carries no node in it, so a manager asked to write a + // second node's set would otherwise measure it against the first node's and + // retire every record the first one owns. + seededFor string + digest map[string]uint64 +} + +// WriteRunningSet reconciles this node's running instances onto the +// per-resource key space. +// +// Best-effort, like the marker write it follows: the local state file is the +// source of truth for the node's own instances, so a failure here leaves the +// cluster's view stale rather than losing anything, and is logged rather than +// returned. The next state change repeats whatever did not land. +func (m *JetStreamManager) WriteRunningSet(nodeID string, vms map[string]*vm.VM) { + if m.records == nil { + return + } + + m.running.mu.Lock() + defer m.running.mu.Unlock() + + if m.running.seededFor != nodeID { + if err := m.seedRunningSet(nodeID); err != nil { + slog.Warn("Could not read the existing instance records; the next state write retries", + "node", nodeID, "err", err) + return + } + } + + for id, instance := range vms { + if instance == nil { + continue + } + // Ownership is stamped here rather than taken from the instance. This + // is the only place that knows whose running set is being written, and + // a record that reaches the key space without it belongs to no node — + // so nothing lists it and nothing retires it. + record := instance.Record() + record.Status.LastNode = nodeID + + sum, err := recordDigest(record) + if err != nil { + slog.Error("Could not encode an instance record", "instanceId", id, "err", err) + continue + } + if current, ok := m.running.digest[id]; ok && current == sum { + continue + } + + // A failed write forgets its digest as well as its record, so the next + // state write repeats the attempt rather than reading the failure as + // already written. + if err := m.records.Replace(context.Background(), instanceRecordKey(id), record); err != nil { + slog.Error("Could not write an instance record", "instanceId", id, "err", err) + delete(m.running.digest, id) + continue + } + m.running.digest[id] = sum + } + + for id := range m.running.digest { + if _, ok := vms[id]; ok { + continue + } + if m.retireRecord(nodeID, id) { + delete(m.running.digest, id) + } + } +} + +// retireRecord drops the record of an instance that has left nodeID's running +// set, reporting whether the question is settled. +// +// It is not settled by deleting unconditionally. One key now holds an instance +// for its whole life, so the record of a departed instance is the same record +// whoever took it over is now using. Two ways to leave are not deaths: an +// operator stop, where WriteStoppedInstance has just rewritten that key, and a +// move, where another node has claimed it. Both are handed over, not retired. +func (m *JetStreamManager) retireRecord(nodeID, instanceID string) bool { + record, err := loadRecord(m.records, instanceRecordKey(instanceID)) + if err != nil { + slog.Warn("Could not tell whether a departed instance is stopped; leaving its record in place", + "instanceId", instanceID, "err", err) + return false + } + if record == nil { + return true + } + if operatorStopped(record) { + return true + } + if record.Status.LastNode != nodeID { + slog.Debug("An instance that left this node is owned elsewhere; leaving its record alone", + "instanceId", instanceID, "owner", record.Status.LastNode) + return true + } + + if err := m.records.Delete(context.Background(), instanceRecordKey(instanceID)); err != nil { + slog.Error("Could not remove the record of an instance that left this node", + "instanceId", instanceID, "err", err) + return false + } + return true +} + +// seedRunningSet primes the digest map from the records already on the key +// space for this node, so the first state write after boot rewrites only what +// actually changed while the node was down — and so a record left behind by a +// previous process is still a candidate for retirement. +func (m *JetStreamManager) seedRunningSet(nodeID string) error { + existing, err := listRecords(m.records, InstanceRecordPrefix) + if err != nil { + return err + } + + digest := make(map[string]uint64, len(existing)) + for _, record := range existing { + if !runsOn(record, nodeID) { + continue + } + sum, err := recordDigest(record) + if err != nil { + return err + } + digest[record.Metadata.Name] = sum + } + + m.running.digest = digest + m.running.seededFor = nodeID + slog.Debug("Seeded the instance record set", "node", nodeID, "records", len(digest)) + return nil +} + +// recordDigest identifies a record by its wire form, so a state change that +// leaves an instance untouched does not become a KV write. encoding/json sorts +// map keys, so the same record always encodes to the same bytes. +func recordDigest(record *vm.InstanceRecord) (uint64, error) { + data, err := json.Marshal(record) + if err != nil { + return 0, err + } + h := fnv.New64a() + _, _ = h.Write(data) + return h.Sum64(), nil +} + +// nodeRunningRecords returns the records nodeID owns, as instances. +func (m *JetStreamManager) nodeRunningRecords(nodeID string) (map[string]*vm.VM, error) { + records, err := listRecords(m.records, InstanceRecordPrefix) + if err != nil { + return nil, err + } + + vms := make(map[string]*vm.VM, len(records)) + for _, record := range records { + if !runsOn(record, nodeID) { + continue + } + vms[record.Metadata.Name] = vm.VMFromRecord(record) + } + return vms, nil +} diff --git a/spinifex/daemon/instance_running_set_test.go b/spinifex/daemon/instance_running_set_test.go new file mode 100644 index 000000000..6f278fa77 --- /dev/null +++ b/spinifex/daemon/instance_running_set_test.go @@ -0,0 +1,462 @@ +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" +) + +const localNode = "node-1" + +func runningVM(id, instanceType string) *vm.VM { + return &vm.VM{ID: id, InstanceType: instanceType, Status: vm.StateRunning, LastNode: localNode} +} + +func runningSet(vms ...*vm.VM) map[string]*vm.VM { + set := make(map[string]*vm.VM, len(vms)) + for _, v := range vms { + set[v.ID] = v + } + return set +} + +// writeRunningSet does what persistState does: the presence marker, then the +// records. Both, because a set written without a marker is a set no reader will +// admit exists. +func writeRunningSet(t *testing.T, m *daemon.JetStreamManager, vms ...*vm.VM) { + t.Helper() + require.NoError(t, m.WriteNodeMarker(localNode)) + m.WriteRunningSet(localNode, runningSet(vms...)) +} + +// recordRevision reads the revision of a record straight from the bucket, so a +// test can tell a write that did nothing from one that did not happen. +func recordRevision(t *testing.T, nc *nats.Conn, instanceID string) uint64 { + t.Helper() + js, err := jetstream.New(nc) + require.NoError(t, err) + kv, err := js.KeyValue(context.Background(), daemon.InstanceStateBucket) + require.NoError(t, err) + entry, err := kv.Get(context.Background(), daemon.InstanceRecordPrefix+instanceID) + require.NoError(t, err) + return entry.Revision() +} + +func TestWriteRunningSet_WritesARecordPerMember(t *testing.T) { + m := newRecordManager(t) + + writeRunningSet(t, m, runningVM("i-1", "t3.nano"), runningVM("i-2", "t3.micro")) + + records, err := m.ListInstanceRecords() + require.NoError(t, err) + byID := make(map[string]string, len(records)) + for _, record := range records { + byID[record.Metadata.Name] = record.Spec.InstanceType + } + assert.Equal(t, map[string]string{"i-1": "t3.nano", "i-2": "t3.micro"}, byID, + "a set of two instances must become two records") +} + +func TestLoadState_ReadsTheNodesRecords(t *testing.T) { + m := newRecordManager(t) + writeRunningSet(t, m, runningVM("i-1", "t3.nano"), runningVM("i-2", "t3.micro")) + + loaded, found, err := m.LoadState(localNode) + + require.NoError(t, err) + require.True(t, found) + require.Len(t, loaded, 2) + assert.Equal(t, "t3.nano", loaded["i-1"].InstanceType) + assert.Equal(t, "t3.micro", loaded["i-2"].InstanceType) +} + +// The key space is shared by every node, so a listing is not a running set +// until it is filtered by who owns each record. +func TestLoadState_IgnoresAnotherNodesRecords(t *testing.T) { + m := newRecordManager(t) + writeRunningSet(t, m, runningVM("i-1", "t3.nano")) + + elsewhere := &vm.VM{ID: "i-2", InstanceType: "t3.micro", Status: vm.StateRunning, LastNode: "node-2"} + require.NoError(t, m.WriteInstanceRecord("i-2", elsewhere.Record())) + + loaded, found, err := m.LoadState(localNode) + + require.NoError(t, err) + require.True(t, found) + assert.NotContains(t, loaded, "i-2") + assert.Len(t, loaded, 1) +} + +// The distinction the presence marker exists for. Both cases scan no records, +// and restore replaces a node's local state with what it reads here whenever it +// believes the cluster has a record of that node — so reading "no instances" as +// "no such node", or the reverse, is the difference between a node coming back +// with its instances and coming back empty. +func TestLoadState_FoundIsTheMarkerNotTheRecordCount(t *testing.T) { + m := newRecordManager(t) + + loaded, found, err := m.LoadState(localNode) + require.NoError(t, err) + assert.False(t, found, "a node that has never written a marker is not in the cluster's record") + assert.Empty(t, loaded) + + require.NoError(t, m.WriteNodeMarker(localNode)) + + loaded, found, err = m.LoadState(localNode) + require.NoError(t, err) + assert.True(t, found, "a node with a marker and no instances is a known, empty node") + assert.Empty(t, loaded) +} + +// An operator-stopped instance belongs to the stopped set, and the running set +// is the same keys filtered. Returning it here would relaunch it on restore. +func TestLoadState_OmitsAnOperatorStoppedRecord(t *testing.T) { + m := newRecordManager(t) + writeRunningSet(t, m, runningVM("i-1", "t3.nano")) + require.NoError(t, m.WriteStoppedInstance("i-1", runningVM("i-1", "t3.nano"))) + + loaded, found, err := m.LoadState(localNode) + + require.NoError(t, err) + require.True(t, found) + assert.NotContains(t, loaded, "i-1") +} + +// The conversion's totality is guarded in the vm package; what this adds is +// that the running-set path actually routes through it, on fields from both +// halves of the record rather than the instance type alone. +func TestWriteRunningSet_CarriesBothHalvesOfTheRecord(t *testing.T) { + m := newRecordManager(t) + instance := runningVM("i-1", "t3.nano") + instance.DesiredState = vm.DesiredRunning + instance.PublicIP = "10.0.0.9" + instance.ENIId = "eni-1" + instance.HostfwdPorts = []int{2222} + + writeRunningSet(t, m, instance) + + loaded, _, err := m.LoadState(localNode) + require.NoError(t, err) + require.Contains(t, loaded, "i-1") + got := loaded["i-1"] + assert.Equal(t, vm.DesiredRunning, got.DesiredState) + assert.Equal(t, []int{2222}, got.HostfwdPorts) + assert.Equal(t, "10.0.0.9", got.PublicIP) + assert.Equal(t, "eni-1", got.ENIId) + assert.Equal(t, localNode, got.LastNode) +} + +// Ownership is stamped by the write rather than carried in from the caller: +// this is the only place that knows whose running set is being written. +func TestWriteRunningSet_StampsOwnership(t *testing.T) { + m := newRecordManager(t) + unowned := &vm.VM{ID: "i-1", InstanceType: "t3.nano", Status: vm.StateRunning} + + require.NoError(t, m.WriteNodeMarker(localNode)) + m.WriteRunningSet(localNode, runningSet(unowned)) + + record, err := m.LoadInstanceRecord("i-1") + require.NoError(t, err) + require.NotNil(t, record) + assert.Equal(t, localNode, record.Status.LastNode) +} + +func TestWriteRunningSet_RetiresAMemberThatLeft(t *testing.T) { + m := newRecordManager(t) + writeRunningSet(t, m, runningVM("i-1", "t3.nano"), runningVM("i-2", "t3.micro")) + + writeRunningSet(t, m, runningVM("i-1", "t3.nano")) + + gone, err := m.LoadInstanceRecord("i-2") + require.NoError(t, err) + assert.Nil(t, gone, "a record left behind by a terminated instance never expires from this bucket") + + kept, err := m.LoadInstanceRecord("i-1") + require.NoError(t, err) + assert.NotNil(t, kept) +} + +// One key holds an instance for its whole life, so the record an instance +// leaves the running set with is the same one the stopped set has just taken +// over. An unconditional retire would delete the stopped instance it became. +func TestWriteRunningSet_HandsOverAMemberThatStopped(t *testing.T) { + m := newRecordManager(t) + writeRunningSet(t, m, runningVM("i-1", "t3.nano")) + + require.NoError(t, m.WriteStoppedInstance("i-1", &vm.VM{ID: "i-1", InstanceType: "t3.nano", Status: vm.StateStopped})) + writeRunningSet(t, m) + + record, err := m.LoadInstanceRecord("i-1") + require.NoError(t, err) + require.NotNil(t, record, "the stopped set owns this record now") + + stopped, err := m.ListStoppedInstances() + require.NoError(t, err) + require.Len(t, stopped, 1) + assert.Equal(t, "i-1", stopped[0].ID) +} + +// Rewriting every record on every state change would reintroduce, one key at a +// time, the contention splitting the blob exists to remove. +func TestWriteRunningSet_WritesOnlyWhatChanged(t *testing.T) { + m, nc := newRecordManagerConn(t) + writeRunningSet(t, m, runningVM("i-1", "t3.nano"), runningVM("i-2", "t3.micro")) + before1, before2 := recordRevision(t, nc, "i-1"), recordRevision(t, nc, "i-2") + + writeRunningSet(t, m, runningVM("i-1", "t3.nano"), runningVM("i-2", "t3.micro")) + + assert.Equal(t, before1, recordRevision(t, nc, "i-1"), "an unchanged instance must not be rewritten") + assert.Equal(t, before2, recordRevision(t, nc, "i-2")) + + changed := runningVM("i-1", "t3.nano") + changed.PublicIP = "10.0.0.9" + writeRunningSet(t, m, changed, runningVM("i-2", "t3.micro")) + + assert.Greater(t, recordRevision(t, nc, "i-1"), before1, "a changed instance must be rewritten") + assert.Equal(t, before2, recordRevision(t, nc, "i-2"), "and only that instance") +} + +// A restarted process has no memory of what it last wrote, so it primes itself +// from the bucket rather than rewriting every record to find out. +func TestWriteRunningSet_SeedsFromWhatIsAlreadyThere(t *testing.T) { + first, nc := newRecordManagerConn(t) + writeRunningSet(t, first, runningVM("i-1", "t3.nano")) + before := recordRevision(t, nc, "i-1") + + second, err := daemon.NewJetStreamManager(nc, 1) + require.NoError(t, err) + require.NoError(t, second.InitKVBucket()) + writeRunningSet(t, second, runningVM("i-1", "t3.nano")) + + assert.Equal(t, before, recordRevision(t, nc, "i-1")) +} + +// Ownership is read from status.LastNode, the same field the GC uses. A node +// must not retire a record for an instance that is running somewhere else. +func TestWriteRunningSet_LeavesAnotherNodesRecordsAlone(t *testing.T) { + m := newRecordManager(t) + elsewhere := &vm.VM{ID: "i-elsewhere", InstanceType: "t3.nano", Status: vm.StateRunning, LastNode: "node-2"} + require.NoError(t, m.WriteInstanceRecord("i-elsewhere", elsewhere.Record())) + + writeRunningSet(t, m, runningVM("i-1", "t3.nano")) + + record, err := m.LoadInstanceRecord("i-elsewhere") + require.NoError(t, err) + assert.NotNil(t, record) +} + +// seedNodeBlobBucket creates the instance-state bucket at the version before +// the blob split, holding one node's running set as a single record. +func seedNodeBlobBucket(t *testing.T, nc *nats.Conn, nodeID string, vms map[string]*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: daemon.InstanceStateBucket, History: 1}) + require.NoError(t, err) + + data, err := json.Marshal(daemon.LocalState{SchemaVersion: daemon.LocalStateSchemaVersion, VMS: vms}) + require.NoError(t, err) + _, err = kv.Put(ctx, daemon.InstanceStatePrefix+nodeID, data) + require.NoError(t, err) + + require.NoError(t, kvutil.WriteVersion(ctx, kv, 2)) + return kv +} + +// readNodeBlob returns the running-set blob at the key the record space +// replaced, or nil if it is gone. +func readNodeBlob(t *testing.T, kv jetstream.KeyValue, nodeID string) *daemon.LocalState { + t.Helper() + entry, err := kv.Get(context.Background(), daemon.InstanceStatePrefix+nodeID) + if err != nil { + return nil + } + var blob daemon.LocalState + require.NoError(t, json.Unmarshal(entry.Value(), &blob)) + return &blob +} + +func TestInstanceStateMigration_SplitsTheNodeBlob(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + seedNodeBlobBucket(t, nc, localNode, runningSet(runningVM("i-1", "t3.nano"), runningVM("i-2", "t3.micro"))) + + m, err := daemon.NewJetStreamManager(nc, 1) + require.NoError(t, err) + require.NoError(t, m.InitKVBucket()) + + records, err := m.ListInstanceRecords() + require.NoError(t, err) + byID := make(map[string]string, len(records)) + for _, record := range records { + byID[record.Metadata.Name] = record.Spec.InstanceType + } + assert.Equal(t, map[string]string{"i-1": "t3.nano", "i-2": "t3.micro"}, byID) +} + +// The split copies the blob rather than consuming it, and nothing after the +// cutover writes over it. It is what a node rolled back to the release before +// the cutover reads to find out what it was running, so leaving it intact is +// the whole of what makes the crossing reversible. +func TestInstanceStateMigration_LeavesTheNodeBlobIntact(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + kv := seedNodeBlobBucket(t, nc, localNode, runningSet(runningVM("i-1", "t3.nano"))) + + m, err := daemon.NewJetStreamManager(nc, 1) + require.NoError(t, err) + require.NoError(t, m.InitKVBucket()) + writeRunningSet(t, m, runningVM("i-1", "t3.nano")) + + blob := readNodeBlob(t, kv, localNode) + require.NotNil(t, blob, "the blob the split copied from must survive the migration") + assert.Contains(t, blob.VMS, "i-1", "and must not be emptied by the marker that replaces it") +} + +// The marker is a key of its own for that reason: written onto the blob it +// would answer the presence question by destroying the answer to the other one. +func TestInstanceStateMigration_SeedsAPresenceMarkerPerNode(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + kv := seedNodeBlobBucket(t, nc, localNode, runningSet(runningVM("i-1", "t3.nano"))) + + m, err := daemon.NewJetStreamManager(nc, 1) + require.NoError(t, err) + require.NoError(t, m.InitKVBucket()) + + _, err = kv.Get(context.Background(), daemon.NodePresencePrefix+localNode) + require.NoError(t, err, "a node that had a blob must be present without having written anything yet") + + loaded, found, err := m.LoadState(localNode) + require.NoError(t, err) + assert.True(t, found, "so it can recover its instances from KV across the crossing") + assert.Contains(t, loaded, "i-1") +} + +// Every node's set is copied, not just the one running the migration: it runs +// once per bucket, and a node that is down would otherwise have no records +// until it came back. +func TestInstanceStateMigration_SplitsEveryNodesBlob(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + kv := seedNodeBlobBucket(t, nc, localNode, runningSet(runningVM("i-1", "t3.nano"))) + + other := &vm.VM{ID: "i-2", InstanceType: "t3.micro", Status: vm.StateRunning, LastNode: "node-2"} + data, err := json.Marshal(daemon.LocalState{ + SchemaVersion: daemon.LocalStateSchemaVersion, + VMS: map[string]*vm.VM{other.ID: other}, + }) + require.NoError(t, err) + _, err = kv.Put(context.Background(), daemon.InstanceStatePrefix+"node-2", data) + require.NoError(t, err) + + m, err := daemon.NewJetStreamManager(nc, 1) + require.NoError(t, err) + require.NoError(t, m.InitKVBucket()) + + record, err := m.LoadInstanceRecord("i-2") + require.NoError(t, err) + require.NotNil(t, record) + assert.Equal(t, "node-2", record.Status.LastNode) +} + +// unownedVM is what a running-set blob actually holds. The key it sat under +// named the node, so nothing wrote the node onto the instance inside it, and a +// record copied forward from one carries no owner at all. +func unownedVM(id, instanceType string) *vm.VM { + return &vm.VM{ID: id, InstanceType: instanceType, Status: vm.StateRunning} +} + +// Ownership is the other half of what the node. key carried in its name, +// and the half a reader cannot do without: after the cutover an unowned record +// is one no node lists. +func TestInstanceStateMigration_CarriesOwnershipOffTheBlobKey(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + seedNodeBlobBucket(t, nc, localNode, runningSet(unownedVM("i-1", "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) + assert.Equal(t, localNode, record.Status.LastNode, "the blob key is the only thing that knew this") +} + +// The case that makes the stamp necessary rather than tidy. The marker says the +// cluster knows the node, so restore adopts what comes back with it — and an +// unowned running set comes back empty, which it would then adopt over the +// instances the node is actually running. +func TestInstanceStateMigration_ANodeRecoversAnUnownedRunningSet(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + seedNodeBlobBucket(t, nc, localNode, runningSet(unownedVM("i-1", "t3.nano"), unownedVM("i-2", "t3.micro"))) + + m, err := daemon.NewJetStreamManager(nc, 1) + require.NoError(t, err) + require.NoError(t, m.InitKVBucket()) + + loaded, found, err := m.LoadState(localNode) + + require.NoError(t, err) + require.True(t, found) + assert.Len(t, loaded, 2, "an adopted empty set here is every instance on the node dropped") +} + +// An owner already on the record is the newer fact. The blob it would be +// overwritten from stopped being written when the cutover landed. +func TestInstanceStateMigration_LeavesAnOwnedRecordAlone(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + moved := &vm.VM{ID: "i-1", InstanceType: "t3.nano", Status: vm.StateRunning, LastNode: "node-2"} + seedNodeBlobBucket(t, nc, localNode, runningSet(moved)) + + 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, "node-2", record.Status.LastNode) +} + +func TestInstanceStateMigration_OwnershipStampIsSafeToRunTwice(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + kv := seedNodeBlobBucket(t, nc, localNode, runningSet(unownedVM("i-1", "t3.nano"))) + + first, err := daemon.NewJetStreamManager(nc, 1) + require.NoError(t, err) + require.NoError(t, first.InitKVBucket()) + before := recordRevision(t, nc, "i-1") + + require.NoError(t, kvutil.WriteVersion(context.Background(), kv, 4)) + second, err := daemon.NewJetStreamManager(nc, 1) + require.NoError(t, err) + require.NoError(t, second.InitKVBucket()) + + assert.Equal(t, before, recordRevision(t, nc, "i-1"), "a record that already names its owner must not be rewritten") +} + +func TestInstanceStateMigration_BlobSplitIsSafeToRunTwice(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + kv := seedNodeBlobBucket(t, nc, localNode, runningSet(runningVM("i-1", "t3.nano"))) + + first, err := daemon.NewJetStreamManager(nc, 1) + require.NoError(t, err) + require.NoError(t, first.InitKVBucket()) + before := recordRevision(t, nc, "i-1") + + require.NoError(t, kvutil.WriteVersion(context.Background(), kv, 2)) + second, err := daemon.NewJetStreamManager(nc, 1) + require.NoError(t, err) + require.NoError(t, second.InitKVBucket()) + + assert.Equal(t, before, recordRevision(t, nc, "i-1"), "a second run must not overwrite what the first copied") +} diff --git a/spinifex/daemon/jetstream.go b/spinifex/daemon/jetstream.go index 82871744e..9ae004d47 100644 --- a/spinifex/daemon/jetstream.go +++ b/spinifex/daemon/jetstream.go @@ -23,8 +23,17 @@ const ( InstanceStateBucket = "spinifex-instance-state" // ClusterStateBucket is the name of the KV bucket for cluster state (heartbeats, shutdown markers, service maps). ClusterStateBucket = "spinifex-cluster-state" - // InstanceStatePrefix is the key prefix for per-node instance state entries. + // InstanceStatePrefix is the key prefix for the per-node running-set blobs + // the record space replaces. Nothing writes it after the cutover; the + // migration reads it and a rolled-back node still finds what it left. InstanceStatePrefix = "node." + // NodePresencePrefix is the key prefix for a node's presence marker. + // + // Deliberately not InstanceStatePrefix. The marker holds no instances, so + // writing it there would empty the blob it shares a key with, and a node + // rolled back to the release before the cutover reads that blob to find out + // what it was running. Frozen means not written, not written empty. + NodePresencePrefix = "nodepresence." // StoppedInstancePrefix is the key prefix for stopped instances in shared KV. StoppedInstancePrefix = "instance." // TerminatedInstanceBucket is the name of the KV bucket for terminated instances (auto-expiry via TTL). @@ -34,9 +43,13 @@ const ( // Schema versions for daemon KV buckets. Both instance buckets copied their // per-instance keys onto the record space at 2; instance-state took the node - // blobs at 3; both re-keyed that space from "i/" to "i." last. See + // blobs at 3; both re-keyed that space from "i/" to "i." next. See // instance_records_migrate.go. - InstanceStateBucketVersion = 4 + // + // instance-state 5 is the cutover: the record space became the only copy. + // The bump is also the compatibility stamp — a build that predates it stops + // on a bucket stamped 5 rather than reading the keys it no longer owns. + InstanceStateBucketVersion = 5 ClusterStateBucketVersion = 1 TerminatedInstanceBucketVersion = 3 ) @@ -59,18 +72,18 @@ type KVSyncObserver interface { 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 + nodeState *kvstore.Store[LocalState] // nodepresence. markers + stopped *kvstore.Store[vm.VM] // instance., frozen: drained, never written term *kvstore.Store[vm.VM] // spinifex-terminated-instances // 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 + 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 - mirror mirrorState + running runningSetState } // checkNodeStateVersion reports whether a node record read from KV is one this @@ -406,36 +419,35 @@ func (m *JetStreamManager) WriteServiceManifest(nodeID string, services []string return err } -// WriteState writes the instance state to the KV store for the given node and -// mirrors it onto the per-resource key space. -// vms must be a snapshot owned by the caller — JetStreamManager does not lock. +// WriteNodeMarker records that this node has cluster state at all, without +// recording what it is. // -// It mirrors even though persistState is the path production takes, because a -// blob writer that skips the mirror leaves a record staler than the blob it -// belongs to, and reads answer from the record. -func (m *JetStreamManager) WriteState(nodeID string, vms map[string]*vm.VM) error { +// Reads need it because "this node has no instances" and "there is no cluster +// record of this node" are different answers and only one of them may replace a +// node's local state. Scanning the records cannot tell them apart — both scan +// empty — and restore adopts the cluster's set wholesale when it believes there +// is one, so conflating them drops every instance on a node whose records are +// briefly unreadable. +// +// The marker carries no instances, so a state change no longer rewrites the +// node's whole set. That is the cost the split exists to remove; keeping an +// empty envelope keeps the answer and not the cost. +func (m *JetStreamManager) WriteNodeMarker(nodeID string) error { if m.nodeState == nil { return errors.New("KV bucket not initialized") } - key := InstanceStatePrefix + nodeID - record := LocalState{SchemaVersion: LocalStateSchemaVersion, VMS: vms} - if err := m.nodeState.Set(context.Background(), key, &record); err != nil { - return err - } - m.MirrorRunningSet(nodeID, vms) - - slog.Debug("Wrote state to JetStream KV", "key", key, "instances", len(vms)) - return nil + key := NodePresencePrefix + nodeID + marker := LocalState{SchemaVersion: LocalStateSchemaVersion} + return m.nodeState.Set(context.Background(), key, &marker) } -// WriteStateBytesBestEffort attempts to push pre-marshalled instance state to KV -// with a deadline. On timeout or error, it logs a warning and returns — never -// blocks the caller past `timeout` and never returns an error. Used when the -// local state file is the source of truth and KV is a best-effort cache; hot -// paths marshal under a short-lived lock and commit lock-free. -func (m *JetStreamManager) WriteStateBytesBestEffort(nodeID string, jsonData []byte, timeout time.Duration) { - if m.stateB == nil { +// WriteNodeMarkerBestEffort writes this node's presence marker with a deadline. +// On timeout or error it logs and returns — never blocks the caller past +// timeout and never returns an error. The local state file is the source of +// truth; KV is the cluster's view of it, and the next state change retries. +func (m *JetStreamManager) WriteNodeMarkerBestEffort(nodeID string, timeout time.Duration) { + if m.nodeState == nil { slog.Debug("KV bucket not initialized, skipping cluster sync", "node", nodeID) return } @@ -443,15 +455,9 @@ func (m *JetStreamManager) WriteStateBytesBestEffort(nodeID string, jsonData []b ctx, cancel := context.WithTimeout(context.Background(), timeout) defer cancel() - key := InstanceStatePrefix + nodeID - // Pre-marshalled bytes, so this goes to the raw handle rather than through - // a typed view. Best-effort by contract: a failure is logged, not recovered - // from here, and the next typed write repairs the bucket for both. - kv, err := m.stateB.KV(ctx) - if err == nil { - _, err = kv.Put(ctx, key, jsonData) - } - if err != nil { + key := NodePresencePrefix + nodeID + marker := LocalState{SchemaVersion: LocalStateSchemaVersion} + if err := m.nodeState.Set(ctx, key, &marker); err != nil { if m.obs != nil { m.obs.RecordKVSyncFailure(InstanceStateBucket, err) } @@ -465,31 +471,22 @@ func (m *JetStreamManager) WriteStateBytesBestEffort(nodeID string, jsonData []b if m.obs != nil { m.obs.RecordKVSyncSuccess(InstanceStateBucket) } - slog.Debug("Wrote state to KV (best-effort)", "key", key, "bytes", len(jsonData)) + slog.Debug("Wrote node marker to KV (best-effort)", "key", key) } -// marshalInstanceState produces the JSON wire form of vms, for the best-effort -// path that marshals under the caller's lock and commits lock-free. The KV -// record and the local state file are the same envelope, so this is the same -// marshaller: the two cannot drift. -func marshalInstanceState(vms map[string]*vm.VM) ([]byte, error) { - return MarshalLocalState(vms) -} - -// LoadState loads the instance state from the KV store for the given node, -// answering each member from its i/ mirror where there is one. +// LoadState loads the instances the given node owns, from one record each. // -// The bool reports whether a record exists at all. A node that has never -// written state and a node whose record was lost both read as empty, and a -// caller that would drop instances on the strength of that must tell them -// apart — so it stays the blob's answer, not the mirrors'. +// The bool reports whether the cluster has a record of this node at all, and it +// is the marker's answer rather than the records'. A node with no instances and +// a node whose records could not be read both scan empty, and only the first of +// them may replace what the node has locally. func (m *JetStreamManager) LoadState(nodeID string) (map[string]*vm.VM, bool, error) { if m.nodeState == nil { return nil, false, errors.New("KV bucket not initialized") } - key := InstanceStatePrefix + nodeID - state, _, err := m.nodeState.Get(context.Background(), key) + key := NodePresencePrefix + nodeID + marker, _, err := m.nodeState.Get(context.Background(), key) if err != nil { if errors.Is(err, kvstore.ErrNotFound) { slog.Debug("No existing state in JetStream KV", "key", key) @@ -497,18 +494,17 @@ func (m *JetStreamManager) LoadState(nodeID string) (map[string]*vm.VM, bool, er } return nil, false, err } - if err := checkNodeStateVersion(key, state.SchemaVersion); err != nil { - return nil, false, err - } - if state.VMS == nil { - state.VMS = make(map[string]*vm.VM) - } - if err := m.answerFromMirrors(state.VMS); err != nil { + if err := checkNodeStateVersion(key, marker.SchemaVersion); err != nil { return nil, false, err } - slog.Debug("Loaded state from JetStream KV", "key", key, "instances", len(state.VMS)) - return state.VMS, true, nil + vms, err := m.nodeRunningRecords(nodeID) + if err != nil { + return nil, false, err + } + + slog.Debug("Loaded state from JetStream KV", "node", nodeID, "instances", len(vms)) + return vms, true, nil } // DeleteState removes the instance state from the KV store for the given node. @@ -517,7 +513,7 @@ func (m *JetStreamManager) DeleteState(nodeID string) error { return errors.New("KV bucket not initialized") } - key := InstanceStatePrefix + nodeID + key := NodePresencePrefix + nodeID if err := m.nodeState.Delete(context.Background(), key); err != nil { return err } @@ -590,54 +586,53 @@ func (m *JetStreamManager) UpdateReplicas(newReplicas int) error { // 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. +// The instance keeps the one key it has held since it was launched. Stopping +// it rewrites that key rather than moving it, so nothing can observe the +// instance at neither key or at both. func (m *JetStreamManager) WriteStoppedInstance(instanceID string, instance *vm.VM) error { - if m.stopped == nil { + if m.records == nil { return errors.New("KV bucket not initialized") } - key := StoppedInstancePrefix + instanceID - // Replace rather than Set: this overwrites the record wholesale, and doing - // it under CAS is what stops a racing update landing out of order. - if err := m.stopped.Replace(context.Background(), key, instance); err != nil { - return err - } - if err := mirrorRecord(m.records, instanceID, instance); err != nil { + // The write stamps what it means rather than trusting the caller to have. + // Membership is a predicate over the record now, so an instance written + // here without both fields set would be stored and then not found — and the + // caller that forgot is not the one that discovers it. + record := instance.Record() + record.Status.Status = vm.StateStopped + record.Spec.DesiredState = vm.DesiredStopped + if err := m.records.Replace(context.Background(), instanceRecordKey(instanceID), record); err != nil { return err } - slog.Debug("Wrote stopped instance to JetStream KV", "key", key, "instanceId", instanceID) + slog.Debug("Wrote stopped instance to JetStream KV", "instanceId", instanceID) return nil } // LoadStoppedInstance loads a stopped instance from the shared KV store. -// 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. +// Returns nil, nil when no record exists, and also when one does but holds an +// instance that is not stopped: the key is shared with this node's running set, +// so the record has to be asked what it is. func (m *JetStreamManager) LoadStoppedInstance(instanceID string) (*vm.VM, error) { - if m.stopped == nil { + if m.records == nil { return nil, errors.New("KV bucket not initialized") } - return loadPreferringRecord(m.records, m.stopped, StoppedInstancePrefix+instanceID, instanceID) + return loadInstance(m.records, instanceID, operatorStopped) } -// 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. +// DeleteStoppedInstance removes a stopped instance from 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 { + if m.records == nil { return errors.New("KV bucket not initialized") } - key := StoppedInstancePrefix + instanceID - if err := deleteBothRecords(m.records, m.stopped, key, instanceID); err != nil { + if err := m.records.Delete(context.Background(), instanceRecordKey(instanceID)); err != nil { return err } + m.drainFrozenKey(m.stopped, StoppedInstancePrefix+instanceID) - slog.Debug("Deleted stopped instance from JetStream KV", "key", key) + slog.Debug("Deleted stopped instance from JetStream KV", "instanceId", instanceID) return nil } @@ -651,32 +646,49 @@ func (m *JetStreamManager) DeleteStoppedInstance(instanceID string) error { // 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. +// Exclusivity is a compare-and-set on the record rather than a delete of it. +// A delete was safe while the key held nothing but a stopped instance; the key +// holds the instance for its whole life now, so deleting to claim it is +// deleting the instance, and a claimant that fails after winning takes the +// instance with it. +// +// The winner clears DesiredStopped, which is what makes the record no longer +// claimable. A loser's CAS fails on the revision, re-reads, finds an instance +// nobody asked to be stopped, and reports the claim lost — so exactly one +// caller can win, which is the property the delete had. +// +// LastNode is left alone. The claimant has not run the instance yet, and the +// node that last did is the one that should recover it if this launch never +// happens. func (m *JetStreamManager) ClaimStoppedInstance(instanceID string) (*vm.VM, error) { - if m.stopped == nil { + if m.records == 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) + key := instanceRecordKey(instanceID) + record, rev, err := m.records.Get(context.Background(), key) if err != nil { + if errors.Is(err, kvstore.ErrNotFound) { + return nil, vm.ErrStoppedInstanceClaimed + } return nil, err } - if notFound { + if !operatorStopped(record) { return nil, vm.ErrStoppedInstanceClaimed } - slog.Debug("Claimed stopped instance from JetStream KV", "key", key, "instanceId", instanceID) - return instance, nil + claimed := *record + claimed.Spec.DesiredState = vm.DesiredRunning + if err := m.records.CompareAndSet(context.Background(), key, &claimed, rev); err != nil { + if errors.Is(err, kvstore.ErrConflict) { + return nil, vm.ErrStoppedInstanceClaimed + } + return nil, err + } + m.drainFrozenKey(m.stopped, StoppedInstancePrefix+instanceID) + + slog.Debug("Claimed stopped instance from JetStream KV", "instanceId", instanceID) + return vm.VMFromRecord(&claimed), nil } // UpdateStoppedInstance atomically applies mutate to the current KV-stored @@ -688,24 +700,58 @@ func (m *JetStreamManager) ClaimStoppedInstance(instanceID string) (*vm.VM, erro // 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. +// The mutation runs against the record, on the one key the instance has, so +// there is a single CAS anchor rather than two that can diverge. func (m *JetStreamManager) UpdateStoppedInstance(instanceID string, mutate func(*vm.VM)) (*vm.VM, error) { - if m.stopped == nil { + if m.records == nil { return nil, errors.New("KV bucket not initialized") } + return mutateInstance(m.records, instanceID, operatorStopped, mutate) +} - updated, err := mutateRecord(m.stopped, StoppedInstancePrefix+instanceID, mutate) +// mutateInstance applies mutate to the instance a record holds, under the +// store's CAS loop, and returns what committed. The callers speak vm.VM and the +// store speaks records, so the conversion happens inside the loop rather than +// around it — a retry has to re-read and re-convert, not replay a stale copy. +// +// want is checked inside the loop, and a record it rejects is reported absent. +// The key outlives the set it is in now, so "still stopped" has to be re-tested +// on the value each attempt reads: a claim that landed between a caller's own +// load and this write leaves the key there, and without the test the caller's +// mutation would land on an instance somebody else has already started. +func mutateInstance(store *kvstore.Store[vm.InstanceRecord], instanceID string, + want func(*vm.InstanceRecord) bool, mutate func(*vm.VM)) (*vm.VM, error) { + var updated *vm.VM + err := store.Mutate(context.Background(), instanceRecordKey(instanceID), + func(record *vm.InstanceRecord) (bool, error) { + if want != nil && !want(record) { + return false, kvstore.ErrNotFound + } + instance := vm.VMFromRecord(record) + mutate(instance) + *record = *instance.Record() + updated = instance + return true, nil + }) if err != nil { return nil, err } - if err := mirrorRecord(m.records, instanceID, updated); err != nil { - return nil, err - } return updated, nil } +// drainFrozenKey removes the key the record replaced. The old key spaces are +// frozen rather than deleted at the cutover, so the crossing can be rolled +// back — but an instance that is gone should not be recoverable, and draining +// them on the delete path is what stops the frozen space growing forever. +func (m *JetStreamManager) drainFrozenKey(store *kvstore.Store[vm.VM], key string) { + if store == nil { + return + } + if err := store.Delete(context.Background(), key); err != nil { + slog.Debug("Could not drain a frozen key", "key", key, "err", err) + } +} + // 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. @@ -722,13 +768,14 @@ 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, -// answering each from its i/ mirror where there is one. +// ListStoppedInstances returns the stopped instances in the shared KV store. +// The key space also holds every running instance, so membership is the +// record's answer rather than the prefix's. func (m *JetStreamManager) ListStoppedInstances() ([]*vm.VM, error) { - if m.stopped == nil { + if m.records == nil { return nil, errors.New("KV bucket not initialized") } - return listPreferringRecords(m.records, m.stopped, StoppedInstancePrefix) + return listInstances(m.records, operatorStopped) } // listRecords returns every record under prefix. Unlike the raw scan it @@ -759,19 +806,15 @@ func listRecords[T any](store *kvstore.Store[T], prefix string) ([]*T, error) { // record instead of replacing it wholesale should use // UpdateTerminatedInstance. func (m *JetStreamManager) WriteTerminatedInstance(instanceID string, instance *vm.VM) error { - if m.term == nil { + if m.termRecords == nil { return errors.New("terminated instance KV bucket not initialized") } - key := TerminatedInstancePrefix + instanceID - if err := m.term.Replace(context.Background(), key, instance); err != nil { - return err - } - if err := mirrorRecord(m.termRecords, instanceID, instance); err != nil { + if err := writeRecord(m.termRecords, instanceID, instance); err != nil { return err } - slog.Debug("Wrote terminated instance to JetStream KV", "key", key, "instanceId", instanceID) + slog.Debug("Wrote terminated instance to JetStream KV", "instanceId", instanceID) return nil } @@ -782,51 +825,44 @@ func (m *JetStreamManager) WriteTerminatedInstance(instanceID string, instance * // progress without clobbering marks written by a concurrent update to the // same record. Returns jetstream.ErrKeyNotFound if no record exists yet. func (m *JetStreamManager) UpdateTerminatedInstance(instanceID string, mutate func(*vm.VM)) (*vm.VM, error) { - if m.term == nil { + if m.termRecords == nil { return nil, errors.New("terminated instance KV bucket not initialized") } - - 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 + // No predicate: the terminated bucket is its own key space, so being in it + // is the whole of the membership test. + return mutateInstance(m.termRecords, instanceID, nil, mutate) } // ListTerminatedInstances returns all terminated instances from the terminated -// KV bucket, answering each from its i/ mirror where there is one. +// KV bucket. Its key space holds nothing else, so every record is one. func (m *JetStreamManager) ListTerminatedInstances() ([]*vm.VM, error) { - if m.term == nil { + if m.termRecords == nil { return nil, errors.New("terminated instance KV bucket not initialized") } - return listPreferringRecords(m.termRecords, m.term, TerminatedInstancePrefix) + return listInstances(m.termRecords, nil) } -// DeleteTerminatedInstance removes a terminated instance from both key spaces -// in the terminated KV bucket. +// DeleteTerminatedInstance removes a terminated instance from the terminated KV +// bucket. func (m *JetStreamManager) DeleteTerminatedInstance(instanceID string) error { - if m.term == nil { + if m.termRecords == nil { return errors.New("terminated instance KV bucket not initialized") } - key := TerminatedInstancePrefix + instanceID - if err := deleteBothRecords(m.termRecords, m.term, key, instanceID); err != nil { + if err := m.termRecords.Delete(context.Background(), instanceRecordKey(instanceID)); err != nil { return err } + m.drainFrozenKey(m.term, TerminatedInstancePrefix+instanceID) - slog.Debug("Deleted terminated instance from JetStream KV", "key", key) + slog.Debug("Deleted terminated instance from JetStream KV", "instanceId", instanceID) return nil } // 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. +// KV bucket. Returns nil, nil if the instance does not exist. func (m *JetStreamManager) LoadTerminatedInstance(instanceID string) (*vm.VM, error) { - if m.term == nil { + if m.termRecords == nil { return nil, errors.New("terminated instance KV bucket not initialized") } - return loadPreferringRecord(m.termRecords, m.term, TerminatedInstancePrefix+instanceID, instanceID) + return loadInstance(m.termRecords, instanceID, nil) } diff --git a/spinifex/daemon/jetstream_test.go b/spinifex/daemon/jetstream_test.go index 6645ca20a..ef8a4007b 100644 --- a/spinifex/daemon/jetstream_test.go +++ b/spinifex/daemon/jetstream_test.go @@ -3,6 +3,7 @@ package daemon import ( "encoding/json" "errors" + "github.com/mulgadc/spinifex/spinifex/testutil" "sync" "testing" "time" @@ -16,6 +17,19 @@ import ( "github.com/stretchr/testify/require" ) +// writeNodeState does what persistState does: the presence marker, then one +// record per running instance. Two calls rather than one because the marker +// answers whether the cluster knows this node and the records answer what it +// is running, and only the first of those may be inferred from the second. +func writeNodeState(t *testing.T, m *JetStreamManager, nodeID string, vms map[string]*vm.VM) error { + t.Helper() + if err := m.WriteNodeMarker(nodeID); err != nil { + return err + } + m.WriteRunningSet(nodeID, vms) + return nil +} + // TestJetStreamManager_WriteAndLoadState tests round-trip write and load of instance state. func TestJetStreamManager_WriteAndLoadState(t *testing.T) { natsURL := sharedJSNATSURL @@ -49,7 +63,7 @@ func TestJetStreamManager_WriteAndLoadState(t *testing.T) { } // Write state - err = jsm.WriteState(testNodeID, testInstances) + err = writeNodeState(t, jsm, testNodeID, testInstances) require.NoError(t, err, "Failed to write state") // Load state @@ -133,7 +147,7 @@ func TestJetStreamManager_BucketReconnection(t *testing.T) { Status: vm.StateRunning, }, } - err = jsm1.WriteState("persist-node", testInstances) + err = writeNodeState(t, jsm1, "persist-node", testInstances) require.NoError(t, err) nc1.Close() @@ -180,7 +194,7 @@ func TestJetStreamManager_DeleteState(t *testing.T) { Status: vm.StateRunning, }, } - err = jsm.WriteState(testNodeID, testInstances) + err = writeNodeState(t, jsm, testNodeID, testInstances) require.NoError(t, err) // Verify state exists @@ -242,7 +256,7 @@ func TestJetStreamManager_WriteState_UpdateExisting(t *testing.T) { Status: vm.StateRunning, }, } - err = jsm.WriteState(testNodeID, initialInstances) + err = writeNodeState(t, jsm, testNodeID, initialInstances) require.NoError(t, err) // Update state with different instances @@ -256,7 +270,7 @@ func TestJetStreamManager_WriteState_UpdateExisting(t *testing.T) { Status: vm.StateRunning, }, } - err = jsm.WriteState(testNodeID, updatedInstances) + err = writeNodeState(t, jsm, testNodeID, updatedInstances) require.NoError(t, err) // Load and verify updated state @@ -285,7 +299,7 @@ func TestJetStreamManager_MultipleNodes(t *testing.T) { node1Instances := map[string]*vm.VM{ "i-node1-001": {ID: "i-node1-001", Status: vm.StateRunning}, } - err = jsm.WriteState("node-1", node1Instances) + err = writeNodeState(t, jsm, "node-1", node1Instances) require.NoError(t, err) // Write state for node-2 @@ -293,7 +307,7 @@ func TestJetStreamManager_MultipleNodes(t *testing.T) { "i-node2-001": {ID: "i-node2-001", Status: vm.StateStopped}, "i-node2-002": {ID: "i-node2-002", Status: vm.StateRunning}, } - err = jsm.WriteState("node-2", node2Instances) + err = writeNodeState(t, jsm, "node-2", node2Instances) require.NoError(t, err) // Load and verify node-1 state @@ -327,7 +341,7 @@ func TestJetStreamManager_KVNotInitialized(t *testing.T) { require.NoError(t, err) testInstances := make(map[string]*vm.VM) - err = jsm.WriteState("test-node", testInstances) + err = writeNodeState(t, jsm, "test-node", testInstances) assert.Error(t, err, "WriteState should error when KV not initialized") _, _, err = jsm.LoadState("test-node") @@ -625,12 +639,14 @@ func TestJetStreamManager_UpdateStoppedInstance_NotFound(t *testing.T) { 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 kvstore.ErrNotFound instead of recreating the stopped record after -// ClaimStoppedInstance already handed it off to a winning start. +// TestJetStreamManager_UpdateStoppedInstance_NoResurrectAfterClaim is the core +// TOCTOU regression test: a claim landing between an UpdateStoppedInstance +// caller's own Load and its CAS write must not be undone. +// +// The claim no longer deletes anything, so the key is still there and the CAS +// write would succeed on revision alone. What refuses it is the membership +// test — the record it reads is no longer stopped, and to this API an instance +// that is not stopped is one that is not there. func TestJetStreamManager_UpdateStoppedInstance_NoResurrectAfterClaim(t *testing.T) { nc, err := nats.Connect(sharedJSNATSURL) require.NoError(t, err) @@ -644,7 +660,7 @@ func TestJetStreamManager_UpdateStoppedInstance_NoResurrectAfterClaim(t *testing require.NoError(t, jsm.WriteStoppedInstance(testVM.ID, testVM)) // A caller loads the record (as TagStoppedInstance / ModifyInstanceAttribute - // do) before a winning claim removes it. + // do) before a winning claim takes it. loaded, err := jsm.LoadStoppedInstance(testVM.ID) require.NoError(t, err) require.NotNil(t, loaded) @@ -654,15 +670,24 @@ func TestJetStreamManager_UpdateStoppedInstance_NoResurrectAfterClaim(t *testing require.NotNil(t, claimed) // The tag/attribute caller's CAS write now races the already-completed - // claim: it must fail cleanly, not resurrect the record. + // claim: it must fail cleanly, not land on the claimed instance. _, err = jsm.UpdateStoppedInstance(testVM.ID, func(v *vm.VM) { v.InstanceType = "should-not-land" }) - assert.ErrorIs(t, err, kvstore.ErrNotFound, "a claim that deleted the record must not be resurrected by a losing racer's update") + assert.ErrorIs(t, err, kvstore.ErrNotFound, + "an update that lost the race to a claim must not land on the instance it started") stillGone, err := jsm.LoadStoppedInstance(testVM.ID) require.NoError(t, err) - assert.Nil(t, stillGone, "the stopped record must stay deleted after the losing update") + assert.Nil(t, stillGone, "and the instance must stay claimed, not fall back into the stopped set") + + // The record itself survives the claim — that is the point of claiming by + // compare-and-set — so the mutation must be absent from it rather than the + // record absent from the bucket. + record, err := jsm.LoadInstanceRecord(testVM.ID) + require.NoError(t, err) + require.NotNil(t, record, "the claim moves the record between sets, it does not delete it") + assert.Equal(t, "t3.micro", record.Spec.InstanceType, "the losing update must not have committed") } // TestJetStreamManager_ListStoppedInstances tests listing multiple stopped instances. @@ -722,7 +747,7 @@ func TestJetStreamManager_StoppedInstances_NoInterference(t *testing.T) { nodeInstances := map[string]*vm.VM{ "i-running-001": {ID: "i-running-001", Status: vm.StateRunning}, } - err = jsm.WriteState("interference-test-node", nodeInstances) + err = writeNodeState(t, jsm, "interference-test-node", nodeInstances) require.NoError(t, err) // Write stopped instance @@ -882,7 +907,7 @@ func TestJetStreamManager_WriteState_RecoverAfterStreamLost(t *testing.T) { testInstances := map[string]*vm.VM{ "i-recover-001": {ID: "i-recover-001", Status: vm.StateRunning}, } - err = jsm.WriteState("recovery-node", testInstances) + err = writeNodeState(t, jsm, "recovery-node", testInstances) require.NoError(t, err, "WriteState should recover after stream loss") // Verify data was written @@ -1053,7 +1078,7 @@ func TestJetStreamManager_WriteState_RecoveryFailure(t *testing.T) { swapToNonJSContext(t, jsm) testInstances := map[string]*vm.VM{"i-fail": {ID: "i-fail"}} - err = jsm.WriteState("fail-node", testInstances) + err = writeNodeState(t, jsm, "fail-node", testInstances) assert.Error(t, err, "WriteState should return error when recovery fails") } @@ -1614,7 +1639,7 @@ func TestJetStreamManager_BestEffort_Success_NotifiesObserver(t *testing.T) { obs := &fakeKVObserver{} jsm.SetSyncObserver(obs) - jsm.WriteStateBytesBestEffort("obs-success", []byte(`{"vms":{}}`), 5*time.Second) + jsm.WriteNodeMarkerBestEffort("obs-success", 5*time.Second) successes, failures := obs.snapshot() require.Len(t, successes, 1) @@ -1636,7 +1661,7 @@ func TestJetStreamManager_BestEffort_PutError_NotifiesObserver(t *testing.T) { // Closing the connection forces the inflight Put to fail without timing out. nc.Close() - jsm.WriteStateBytesBestEffort("obs-fail", []byte(`{"vms":{}}`), 5*time.Second) + jsm.WriteNodeMarkerBestEffort("obs-fail", 5*time.Second) successes, failures := obs.snapshot() assert.Empty(t, successes) @@ -1655,7 +1680,7 @@ func TestJetStreamManager_BestEffort_NilObserver_NoPanic(t *testing.T) { require.NoError(t, jsm.InitKVBucket()) // No observer set — must not panic on success or failure paths. - jsm.WriteStateBytesBestEffort("obs-nil", []byte(`{"vms":{}}`), 5*time.Second) + jsm.WriteNodeMarkerBestEffort("obs-nil", 5*time.Second) } // --- UpdateMgmtIPAM CAS tests --- @@ -1750,10 +1775,11 @@ func TestJetStreamManager_UpdateMgmtIPAM_ClusterKVNotInitialized(t *testing.T) { // logged an undecodable record and carried on, so the instance simply vanished // from DescribeInstances and the caller concluded it was gone. A tenant-facing // listing is complete or it fails. +// Its own JetStream rather than the package's: one key space holds every +// instance now, so an undecodable record fails every listing that reads it and +// not just the one under test. func TestJetStreamManager_ListStoppedInstances_FailsOnAnUndecodableRecord(t *testing.T) { - nc, err := nats.Connect(sharedJSNATSURL) - require.NoError(t, err) - defer nc.Close() + _, nc, _ := testutil.StartTestJetStream(t) jsm, err := NewJetStreamManager(nc, 1) require.NoError(t, err) @@ -1763,7 +1789,7 @@ func TestJetStreamManager_ListStoppedInstances_FailsOnAnUndecodableRecord(t *tes kv, err := jsm.stateB.KV(t.Context()) require.NoError(t, err) - _, err = kv.Put(t.Context(), StoppedInstancePrefix+"i-corrupt", []byte("{not json")) + _, err = kv.Put(t.Context(), InstanceRecordPrefix+"i-corrupt", []byte("{not json")) require.NoError(t, err) _, err = jsm.ListStoppedInstances() @@ -1802,12 +1828,15 @@ func seedNodeRecord(t *testing.T, jsm *JetStreamManager, nodeID string, raw []by t.Helper() kv, err := jsm.stateB.KV(t.Context()) require.NoError(t, err) - _, err = kv.Put(t.Context(), InstanceStatePrefix+nodeID, raw) + _, err = kv.Put(t.Context(), NodePresencePrefix+nodeID, raw) require.NoError(t, err) } // A node record predates schema versioning, so it carries no version at all. // Reading it as current is what makes an in-place upgrade need no migration. +// +// The instances come from the records, not from the marker, so a marker that +// still carries a vms member is read for its presence and nothing else. func TestJetStreamManager_LoadState_UnversionedRecordStillReads(t *testing.T) { nc, err := nats.Connect(sharedJSNATSURL) require.NoError(t, err) @@ -1819,12 +1848,15 @@ func TestJetStreamManager_LoadState_UnversionedRecordStillReads(t *testing.T) { seedNodeRecord(t, jsm, "legacy-node", []byte(`{"vms":{"i-legacy-001":{"id":"i-legacy-001","status":"running"}}}`)) + require.NoError(t, jsm.WriteInstanceRecord("i-legacy-002", + (&vm.VM{ID: "i-legacy-002", Status: vm.StateRunning, LastNode: "legacy-node"}).Record())) loaded, found, err := jsm.LoadState("legacy-node") require.NoError(t, err, "A record written before versioning must still read") require.True(t, found) require.Len(t, loaded, 1) - assert.Equal(t, vm.StateRunning, loaded["i-legacy-001"].Status) + assert.Equal(t, vm.StateRunning, loaded["i-legacy-002"].Status) + assert.NotContains(t, loaded, "i-legacy-001", "the marker is not a running set") } // A record from a newer binary is refused rather than parsed on a guess: this @@ -1849,7 +1881,7 @@ func TestJetStreamManager_LoadState_FutureRecordIsRefused(t *testing.T) { } // Both writers stamp the version: the typed WriteState and the best-effort -// pre-marshalled path, which is the one that would silently skip it. +// marker path, which is the one that would silently skip it. func TestJetStreamManager_WriteState_StampsSchemaVersion(t *testing.T) { nc, err := nats.Connect(sharedJSNATSURL) require.NoError(t, err) @@ -1860,14 +1892,14 @@ func TestJetStreamManager_WriteState_StampsSchemaVersion(t *testing.T) { require.NoError(t, jsm.InitKVBucket()) vms := map[string]*vm.VM{"i-stamp-001": {ID: "i-stamp-001", Status: vm.StateRunning}} - require.NoError(t, jsm.WriteState("stamp-node", vms)) + require.NoError(t, writeNodeState(t, jsm, "stamp-node", vms)) kv, err := jsm.stateB.KV(t.Context()) require.NoError(t, err) read := func(nodeID string) LocalState { t.Helper() - entry, err := kv.Get(t.Context(), InstanceStatePrefix+nodeID) + entry, err := kv.Get(t.Context(), NodePresencePrefix+nodeID) require.NoError(t, err) var got LocalState require.NoError(t, json.Unmarshal(entry.Value(), &got)) @@ -1876,24 +1908,10 @@ func TestJetStreamManager_WriteState_StampsSchemaVersion(t *testing.T) { assert.Equal(t, LocalStateSchemaVersion, read("stamp-node").SchemaVersion) - data, err := marshalInstanceState(vms) - require.NoError(t, err) - jsm.WriteStateBytesBestEffort("bytes-node", data, 5*time.Second) + jsm.WriteNodeMarkerBestEffort("bytes-node", 5*time.Second) assert.Equal(t, LocalStateSchemaVersion, read("bytes-node").SchemaVersion, - "the best-effort path marshals its own bytes and must stamp the same version") -} - -// The KV node record and the local state file are one envelope. Asserted on the -// bytes rather than the type, because a second marshaller is how they drift. -func TestMarshalInstanceState_MatchesLocalStateFileBytes(t *testing.T) { - vms := map[string]*vm.VM{ - "i-envelope-001": {ID: "i-envelope-001", Status: vm.StateRunning, InstanceType: "t3.micro"}, - } - - kvBytes, err := marshalInstanceState(vms) - require.NoError(t, err) - fileBytes, err := MarshalLocalState(vms) - require.NoError(t, err) - - assert.JSONEq(t, string(fileBytes), string(kvBytes)) + "the best-effort path must stamp the same version") + + assert.Empty(t, read("stamp-node").VMS, + "the node key is a presence marker: carrying the running set is the cost the split removes") } diff --git a/spinifex/daemon/state_test.go b/spinifex/daemon/state_test.go index b09c61141..266bc2469 100644 --- a/spinifex/daemon/state_test.go +++ b/spinifex/daemon/state_test.go @@ -758,7 +758,7 @@ func TestRestoreInstances_ClusterRecordWinsForSharedInstance(t *testing.T) { require.NoError(t, daemon.WriteState()) // The cluster record covers only i-shared, and disagrees about it. - require.NoError(t, daemon.jsManager.WriteState(daemon.node, map[string]*vm.VM{ + require.NoError(t, writeNodeState(t, daemon.jsManager, daemon.node, map[string]*vm.VM{ "i-shared": {ID: "i-shared", Status: vm.StateError, InstanceType: "t3.small"}, })) diff --git a/spinifex/migrate/VERSIONS.md b/spinifex/migrate/VERSIONS.md index cba67fbb8..58abaed28 100644 --- a/spinifex/migrate/VERSIONS.md +++ b/spinifex/migrate/VERSIONS.md @@ -19,7 +19,7 @@ Single source of truth for the current schema version of every config file Spini | Bucket | Current version | Constant | |---|---|---| -| `spinifex-instance-state` | `4` | `daemon.InstanceStateBucketVersion` | +| `spinifex-instance-state` | `5` | `daemon.InstanceStateBucketVersion` | | `spinifex-cluster-state` | `1` | `daemon.ClusterStateBucketVersion` | | `spinifex-terminated-instances` | `3` | `daemon.TerminatedInstanceBucketVersion` | @@ -34,11 +34,14 @@ Single source of truth for the current schema version of every config file Spini | `spinifex-instance-state` | `1`→`2` | copy the `instance.` keys onto the record space | | `spinifex-instance-state` | `2`→`3` | split each `node.` blob, one record per instance in it | | `spinifex-instance-state` | `3`→`4` | re-key the record space from `i/` to `i.` | +| `spinifex-instance-state` | `4`→`5` | carry each `node.` blob's presence and ownership onto the record space: a `nodepresence.` marker, and that node named as `last_node` on every record in the blob that has no owner | | `spinifex-terminated-instances` | `1`→`2` | copy the `terminated.` keys onto the record space | | `spinifex-terminated-instances` | `2`→`3` | re-key the record space from `i/` to `i.` | The re-key steps exist because a watch filter is a NATS subject filter: `*` matches one dot-delimited token, so `i/` is a single token and `i/*` matches nothing. Only a build that shipped the slash has anything for them to move. +Version `5` of `spinifex-instance-state` is the cutover: the record space became the only copy, and `instance.` and `node.` are frozen — read by nothing, written by nothing, and left in place so the crossing can be rolled back. The bump is also what a build predating the cutover trips over: `RunKV` refuses a bucket stamped past what it understands, with `SchemaAheadError`, rather than opening it and reading half a key space. + Every other bucket has no registered migration, so `RunKV` stamps its target version directly on first init. **Object store:** none, so `RunObject` stamps directly too. diff --git a/spinifex/migrate/migrate.go b/spinifex/migrate/migrate.go index af58b2b9e..ff213d837 100644 --- a/spinifex/migrate/migrate.go +++ b/spinifex/migrate/migrate.go @@ -152,7 +152,10 @@ func (r *Registry) RunObject(ctx context.Context, target string, objects objects return fmt.Errorf("read version for %s: %w", target, err) } - if current >= targetVersion { + if current > targetVersion { + return SchemaAheadError{Bucket: target, Found: current, Understood: targetVersion} + } + if current == targetVersion { return nil } @@ -207,6 +210,25 @@ func (r *Registry) RunObject(ctx context.Context, target string, objects objects return nil } +// SchemaAheadError reports a bucket migrated past what this build understands, +// which means another node in the cluster is running a newer release. +// +// Stopping is the point. A migration only ever adds steps, so a newer schema is +// one this build has no code for: it would read the keys it knows, miss the +// ones it does not, and write back a view of the bucket assembled from half of +// it. Refusing to open is recoverable; that is not. +type SchemaAheadError struct { + Bucket string + Found int + Understood int +} + +func (e SchemaAheadError) Error() string { + return fmt.Sprintf( + "%s is at schema version %d but this build understands %d: another node is running a newer release of Spinifex; upgrade this node to match", + e.Bucket, e.Found, e.Understood) +} + // RunKVWithJetStream is RunKV with a JetStream handle attached to each // migration's KVContext, enabling cross-bucket reads (e.g. owner-attribution // during a backfill). Prefer plain RunKV when the migration is self-contained. @@ -226,7 +248,10 @@ func (r *Registry) runKV(ctx context.Context, bucket string, kv jetstream.KeyVal return fmt.Errorf("read version for %s: %w", bucket, err) } - if current >= targetVersion { + if current > targetVersion { + return SchemaAheadError{Bucket: bucket, Found: current, Understood: targetVersion} + } + if current == targetVersion { return nil } diff --git a/spinifex/migrate/migrate_test.go b/spinifex/migrate/migrate_test.go index e79744c1a..f91277cea 100644 --- a/spinifex/migrate/migrate_test.go +++ b/spinifex/migrate/migrate_test.go @@ -242,6 +242,33 @@ func TestRunKV_Idempotent(t *testing.T) { assert.Equal(t, 1, runCount) // Not incremented. } +// A bucket stamped past what this build understands was migrated by a node on +// a newer release. Opening it anyway is the mixed-cluster failure: this build +// reads the keys it knows, misses the ones it does not, and writes back a view +// of the bucket assembled from half of it. +func TestRunKV_SchemaAheadOfThisBuildIsRefused(t *testing.T) { + r := NewRegistry() + r.RegisterKV("test-bucket", KVMigration{ + FromVersion: 1, ToVersion: 2, Description: "add field", + Run: func(ctx context.Context, kvc KVContext) error { return nil }, + }) + + _, nc := startTestNATS(t) + kv := createTestBucket(t, nc, "test-bucket") + _, err := kv.PutString(t.Context(), "_version", "5") + require.NoError(t, err) + + err = r.RunKV(t.Context(), "test-bucket", kv, 2) + + var ahead SchemaAheadError + require.ErrorAs(t, err, &ahead) + assert.Equal(t, 5, ahead.Found) + assert.Equal(t, 2, ahead.Understood) + assert.Contains(t, err.Error(), "test-bucket") + assert.Contains(t, err.Error(), "upgrade this node", + "the error has to say what to do about it, not just that it happened") +} + // --- Config version reader tests --- func TestTOMLVersionReader_ReadStringVersion(t *testing.T) {