From b1d0eba8920314d839310a1ffad7c38416a97ce4 Mon Sep 17 00:00:00 2001 From: joshsiv-mulga Date: Tue, 29 Sep 2026 15:30:18 +1000 Subject: [PATCH] refactor: extract KV migration engine to state foundation --- spinifex/daemon/instance_records_migrate.go | 2 +- spinifex/daemon/jetstream.go | 2 +- spinifex/foundation/state/migrate/kv.go | 137 ++++++++++++++++++ spinifex/handlers/ec2/account/service_impl.go | 2 +- spinifex/handlers/ec2/eigw/service_impl.go | 4 +- spinifex/handlers/ec2/eip/service_impl.go | 4 +- spinifex/handlers/ec2/igw/service_impl.go | 4 +- .../ec2/launchtemplate/service_impl.go | 4 +- spinifex/handlers/ec2/natgw/service_impl.go | 4 +- .../ec2/placementgroup/service_impl.go | 2 +- .../handlers/ec2/routetable/service_impl.go | 4 +- .../handlers/ec2/snapshot/service_impl.go | 8 +- .../handlers/ec2/spotinstance/service_impl.go | 2 +- spinifex/handlers/ec2/vpc/ipam.go | 2 +- spinifex/handlers/ec2/vpc/service_impl.go | 2 +- spinifex/handlers/ecs/store.go | 4 +- spinifex/handlers/eks/store.go | 4 +- spinifex/handlers/elbv2/store.go | 2 +- spinifex/handlers/iam/service_impl.go | 2 +- spinifex/handlers/rds/store.go | 4 +- spinifex/handlers/sts/session_credentials.go | 4 +- spinifex/kvlease/bucket.go | 2 +- spinifex/migrate/migrate.go | 133 +---------------- spinifex/migrate/migrate_test.go | 65 +++++---- spinifex/network/external/static_pool.go | 2 +- 25 files changed, 211 insertions(+), 194 deletions(-) create mode 100644 spinifex/foundation/state/migrate/kv.go diff --git a/spinifex/daemon/instance_records_migrate.go b/spinifex/daemon/instance_records_migrate.go index 7caf06dca..51025d175 100644 --- a/spinifex/daemon/instance_records_migrate.go +++ b/spinifex/daemon/instance_records_migrate.go @@ -8,7 +8,7 @@ import ( "strings" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" - "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" "github.com/mulgadc/spinifex/spinifex/runtime/compute/vm" "github.com/nats-io/nats.go/jetstream" ) diff --git a/spinifex/daemon/jetstream.go b/spinifex/daemon/jetstream.go index f3203df26..67c2c99ca 100644 --- a/spinifex/daemon/jetstream.go +++ b/spinifex/daemon/jetstream.go @@ -13,7 +13,7 @@ import ( "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" - "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" "github.com/mulgadc/spinifex/spinifex/foundation/telemetry" "github.com/mulgadc/spinifex/spinifex/runtime/compute/vm" "github.com/nats-io/nats.go" diff --git a/spinifex/foundation/state/migrate/kv.go b/spinifex/foundation/state/migrate/kv.go new file mode 100644 index 000000000..bb0a48894 --- /dev/null +++ b/spinifex/foundation/state/migrate/kv.go @@ -0,0 +1,137 @@ +// Package migrate applies versioned transformations to shared KV state. +// +// Individual domains register their migrations here; this package owns the +// chain validation and version-stamping mechanics, not any domain record. +package migrate + +import ( + "context" + "fmt" + "log/slog" + "sort" + + "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" + "github.com/nats-io/nats.go/jetstream" +) + +// KVMigration represents a versioned transformation of KV bucket data. +type KVMigration struct { + FromVersion int + ToVersion int + Description string + Run func(ctx context.Context, kvc KVContext) error +} + +// KVContext provides a migration with its target bucket. JetStream is non-nil +// only for RunKVWithJetStream, for migrations that must read sibling buckets. +type KVContext struct { + KV jetstream.KeyValue + JetStream jetstream.JetStream + Logger *slog.Logger +} + +// Registry holds migrations keyed by KV bucket name. +type Registry struct { + kvMigrations map[string][]KVMigration +} + +// DefaultRegistry is the process-wide KV migration registry. Domains register +// their versioned state transitions during package initialization. +var DefaultRegistry = NewRegistry() + +// NewRegistry creates an empty KV migration registry. +func NewRegistry() *Registry { + return &Registry{kvMigrations: make(map[string][]KVMigration)} +} + +// RegisterKV adds a bucket migration. Migrations execute by FromVersion order. +func (r *Registry) RegisterKV(bucket string, m KVMigration) { + r.kvMigrations[bucket] = append(r.kvMigrations[bucket], m) + sort.Slice(r.kvMigrations[bucket], func(i, j int) bool { + return r.kvMigrations[bucket][i].FromVersion < r.kvMigrations[bucket][j].FromVersion + }) +} + +// SchemaAheadError reports a bucket migrated past what this build understands. +// A node must stop rather than write an incomplete view of newer state. +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 context, for migrations that need sibling-bucket reads. +func (r *Registry) RunKVWithJetStream(ctx context.Context, bucket string, kv jetstream.KeyValue, js jetstream.JetStream, targetVersion int) error { + return r.runKV(ctx, bucket, kv, js, targetVersion) +} + +// RunKV applies pending migrations up to targetVersion. A fresh bucket with no +// registered migrations is stamped directly; an incomplete chain is refused. +func (r *Registry) RunKV(ctx context.Context, bucket string, kv jetstream.KeyValue, targetVersion int) error { + return r.runKV(ctx, bucket, kv, nil, targetVersion) +} + +func (r *Registry) runKV(ctx context.Context, bucket string, kv jetstream.KeyValue, js jetstream.JetStream, targetVersion int) error { + current, err := kvutil.ReadVersion(ctx, kv) + if err != nil { + return fmt.Errorf("read version for %s: %w", bucket, err) + } + + if current > targetVersion { + return SchemaAheadError{Bucket: bucket, Found: current, Understood: targetVersion} + } + if current == targetVersion { + return nil + } + + all := r.kvMigrations[bucket] + if current == 0 && len(all) == 0 { + return kvutil.WriteVersion(ctx, kv, targetVersion) + } + + // A fresh bucket has no v0 schema; begin at the lowest registered step. + if current == 0 { + current = all[0].FromVersion + } + + var pending []KVMigration + for _, m := range all { + if m.FromVersion >= current && m.ToVersion <= targetVersion { + pending = append(pending, m) + } + } + if len(pending) == 0 { + return fmt.Errorf("no migrations registered for %s from version %d to %d", bucket, current, targetVersion) + } + + expected := current + for _, m := range pending { + if m.FromVersion != expected { + return fmt.Errorf("migration chain gap for %s: expected from %d, got from %d", bucket, expected, m.FromVersion) + } + expected = m.ToVersion + } + if expected != targetVersion { + return fmt.Errorf("migration chain for %s ends at version %d, target is %d", bucket, expected, targetVersion) + } + + logger := slog.Default() + for _, m := range pending { + logger.Info("Running KV migration", "bucket", bucket, "from", m.FromVersion, "to", m.ToVersion, "description", m.Description) + kvc := KVContext{KV: kv, JetStream: js, Logger: logger} + if err := m.Run(ctx, kvc); err != nil { + return fmt.Errorf("KV migration %s %dā†’%d failed: %w", bucket, m.FromVersion, m.ToVersion, err) + } + if err := kvutil.WriteVersion(ctx, kv, m.ToVersion); err != nil { + return fmt.Errorf("stamp version %d on %s: %w", m.ToVersion, bucket, err) + } + } + return nil +} diff --git a/spinifex/handlers/ec2/account/service_impl.go b/spinifex/handlers/ec2/account/service_impl.go index 2578f1a90..9d7c257bb 100644 --- a/spinifex/handlers/ec2/account/service_impl.go +++ b/spinifex/handlers/ec2/account/service_impl.go @@ -11,7 +11,7 @@ import ( "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/spinifex/spinifex/config" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" - "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" diff --git a/spinifex/handlers/ec2/eigw/service_impl.go b/spinifex/handlers/ec2/eigw/service_impl.go index d51de7202..0efabf2c0 100644 --- a/spinifex/handlers/ec2/eigw/service_impl.go +++ b/spinifex/handlers/ec2/eigw/service_impl.go @@ -14,9 +14,9 @@ import ( "github.com/mulgadc/spinifex/spinifex/config" "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters" - handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" - "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" + handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" diff --git a/spinifex/handlers/ec2/eip/service_impl.go b/spinifex/handlers/ec2/eip/service_impl.go index ea1c38b9d..a6025d832 100644 --- a/spinifex/handlers/ec2/eip/service_impl.go +++ b/spinifex/handlers/ec2/eip/service_impl.go @@ -13,9 +13,9 @@ import ( "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters" - handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" - "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" + handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" "github.com/mulgadc/spinifex/spinifex/network/topology" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" diff --git a/spinifex/handlers/ec2/igw/service_impl.go b/spinifex/handlers/ec2/igw/service_impl.go index dcc0a54ee..631e811ca 100644 --- a/spinifex/handlers/ec2/igw/service_impl.go +++ b/spinifex/handlers/ec2/igw/service_impl.go @@ -14,9 +14,9 @@ import ( "github.com/mulgadc/spinifex/spinifex/config" "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters" - handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" - "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" + handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" diff --git a/spinifex/handlers/ec2/launchtemplate/service_impl.go b/spinifex/handlers/ec2/launchtemplate/service_impl.go index b27491e8b..a450e300c 100644 --- a/spinifex/handlers/ec2/launchtemplate/service_impl.go +++ b/spinifex/handlers/ec2/launchtemplate/service_impl.go @@ -14,11 +14,11 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" - "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" "github.com/mulgadc/spinifex/spinifex/config" + "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" - "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" diff --git a/spinifex/handlers/ec2/natgw/service_impl.go b/spinifex/handlers/ec2/natgw/service_impl.go index 8bdb5979c..535921f8b 100644 --- a/spinifex/handlers/ec2/natgw/service_impl.go +++ b/spinifex/handlers/ec2/natgw/service_impl.go @@ -13,10 +13,10 @@ import ( "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters" + "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" + "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" handlers_ec2_eip "github.com/mulgadc/spinifex/spinifex/handlers/ec2/eip" handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" - "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" - "github.com/mulgadc/spinifex/spinifex/migrate" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" diff --git a/spinifex/handlers/ec2/placementgroup/service_impl.go b/spinifex/handlers/ec2/placementgroup/service_impl.go index 28e03aa68..8ad56fc9f 100644 --- a/spinifex/handlers/ec2/placementgroup/service_impl.go +++ b/spinifex/handlers/ec2/placementgroup/service_impl.go @@ -15,7 +15,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" - "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" diff --git a/spinifex/handlers/ec2/routetable/service_impl.go b/spinifex/handlers/ec2/routetable/service_impl.go index 7a3fe0b32..fa6d0fc53 100644 --- a/spinifex/handlers/ec2/routetable/service_impl.go +++ b/spinifex/handlers/ec2/routetable/service_impl.go @@ -14,11 +14,11 @@ import ( "github.com/mulgadc/spinifex/spinifex/config" "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters" + "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" + "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" handlers_ec2_igw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/igw" handlers_ec2_natgw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/natgw" handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" - "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" - "github.com/mulgadc/spinifex/spinifex/migrate" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" diff --git a/spinifex/handlers/ec2/snapshot/service_impl.go b/spinifex/handlers/ec2/snapshot/service_impl.go index a5dff94bb..668a72065 100644 --- a/spinifex/handlers/ec2/snapshot/service_impl.go +++ b/spinifex/handlers/ec2/snapshot/service_impl.go @@ -16,15 +16,15 @@ import ( "github.com/aws/aws-sdk-go/service/ec2" "github.com/aws/aws-sdk-go/service/s3" "github.com/mulgadc/bluebottle/pkg/safecast" - "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" "github.com/mulgadc/spinifex/spinifex/config" "github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata" - "github.com/mulgadc/spinifex/spinifex/providers/ebs" + "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters" - "github.com/mulgadc/spinifex/spinifex/handlers/ec2/volumestate" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore" - "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" + "github.com/mulgadc/spinifex/spinifex/handlers/ec2/volumestate" "github.com/mulgadc/spinifex/spinifex/objectstore" + "github.com/mulgadc/spinifex/spinifex/providers/ebs" "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" diff --git a/spinifex/handlers/ec2/spotinstance/service_impl.go b/spinifex/handlers/ec2/spotinstance/service_impl.go index 897f8e82c..1570cc5c6 100644 --- a/spinifex/handlers/ec2/spotinstance/service_impl.go +++ b/spinifex/handlers/ec2/spotinstance/service_impl.go @@ -15,7 +15,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" - "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" diff --git a/spinifex/handlers/ec2/vpc/ipam.go b/spinifex/handlers/ec2/vpc/ipam.go index 084b8d5af..9bb7b1bf7 100644 --- a/spinifex/handlers/ec2/vpc/ipam.go +++ b/spinifex/handlers/ec2/vpc/ipam.go @@ -8,7 +8,7 @@ import ( "net/netip" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore" - "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" "github.com/nats-io/nats.go/jetstream" ) diff --git a/spinifex/handlers/ec2/vpc/service_impl.go b/spinifex/handlers/ec2/vpc/service_impl.go index 208ad698a..36246a24b 100644 --- a/spinifex/handlers/ec2/vpc/service_impl.go +++ b/spinifex/handlers/ec2/vpc/service_impl.go @@ -17,7 +17,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" - "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" diff --git a/spinifex/handlers/ecs/store.go b/spinifex/handlers/ecs/store.go index 05f766f97..0e36a2eeb 100644 --- a/spinifex/handlers/ecs/store.go +++ b/spinifex/handlers/ecs/store.go @@ -6,9 +6,9 @@ import ( "fmt" "time" - "github.com/mulgadc/spinifex/spinifex/kvlease" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" - "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" + "github.com/mulgadc/spinifex/spinifex/kvlease" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" ) diff --git a/spinifex/handlers/eks/store.go b/spinifex/handlers/eks/store.go index c670c6ff9..ec5367560 100644 --- a/spinifex/handlers/eks/store.go +++ b/spinifex/handlers/eks/store.go @@ -9,10 +9,10 @@ import ( "strings" "time" - "github.com/mulgadc/spinifex/spinifex/kvlease" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" - "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" + "github.com/mulgadc/spinifex/spinifex/kvlease" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" ) diff --git a/spinifex/handlers/elbv2/store.go b/spinifex/handlers/elbv2/store.go index f32421bda..0cce10a37 100644 --- a/spinifex/handlers/elbv2/store.go +++ b/spinifex/handlers/elbv2/store.go @@ -12,7 +12,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" - "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" ) diff --git a/spinifex/handlers/iam/service_impl.go b/spinifex/handlers/iam/service_impl.go index ace4b3e9a..a6d2fe4f0 100644 --- a/spinifex/handlers/iam/service_impl.go +++ b/spinifex/handlers/iam/service_impl.go @@ -27,7 +27,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/foundation/aws/arn" "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" - "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" "github.com/mulgadc/spinifex/spinifex/utils" ) diff --git a/spinifex/handlers/rds/store.go b/spinifex/handlers/rds/store.go index 482896c55..b0043dd4c 100644 --- a/spinifex/handlers/rds/store.go +++ b/spinifex/handlers/rds/store.go @@ -11,10 +11,10 @@ import ( "strings" "time" - "github.com/mulgadc/spinifex/spinifex/kvlease" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" - "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" + "github.com/mulgadc/spinifex/spinifex/kvlease" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" ) diff --git a/spinifex/handlers/sts/session_credentials.go b/spinifex/handlers/sts/session_credentials.go index 7f74cf8de..a68fd3e16 100644 --- a/spinifex/handlers/sts/session_credentials.go +++ b/spinifex/handlers/sts/session_credentials.go @@ -11,10 +11,10 @@ import ( "strings" "time" - "github.com/mulgadc/spinifex/spinifex/foundation/telemetry" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" - "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" + "github.com/mulgadc/spinifex/spinifex/foundation/telemetry" "github.com/nats-io/nats.go/jetstream" ) diff --git a/spinifex/kvlease/bucket.go b/spinifex/kvlease/bucket.go index 073703be1..14d95e322 100644 --- a/spinifex/kvlease/bucket.go +++ b/spinifex/kvlease/bucket.go @@ -7,7 +7,7 @@ import ( "log/slog" "time" - "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" "github.com/mulgadc/spinifex/spinifex/foundation/telemetry" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" diff --git a/spinifex/migrate/migrate.go b/spinifex/migrate/migrate.go index 77b43ea38..e05776080 100644 --- a/spinifex/migrate/migrate.go +++ b/spinifex/migrate/migrate.go @@ -10,27 +10,11 @@ import ( "time" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil" + statemigrate "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" "github.com/mulgadc/spinifex/spinifex/objectstore" "github.com/nats-io/nats.go/jetstream" ) -// KVMigration represents a versioned transformation of KV bucket data. -type KVMigration struct { - FromVersion int - ToVersion int - Description string - Run func(ctx context.Context, kvc KVContext) error -} - -// KVContext provides KV migration functions access to the bucket being migrated. -// JetStream is non-nil only when the caller used RunKVWithJetStream ā€” used by -// migrations that need to read sibling buckets (e.g. owner-attribution lookups). -type KVContext struct { - KV jetstream.KeyValue - JetStream jetstream.JetStream - Logger *slog.Logger -} - // ConfigMigration represents a versioned transformation of on-disk config files. type ConfigMigration struct { FromVersion int @@ -79,16 +63,16 @@ type configTarget struct { reader ConfigVersionReader } -// Registry holds all registered migrations, keyed by target name. +// Registry holds configuration and object-store migrations, keyed by target. +// KV migrations are state primitives and live in foundation/state/migrate. type Registry struct { - kvMigrations map[string][]KVMigration configMigrations map[string][]ConfigMigration configTargets map[string]configTarget objectMigrations map[string][]ObjectMigration } -// DefaultRegistry is the global migration registry. Migrations self-register -// via init() functions so they are available before any service starts. +// DefaultRegistry is the global registry for configuration and object-store +// migrations. KV migrations use foundation/state/migrate.DefaultRegistry. var DefaultRegistry = NewRegistry() // spinifexTarget is registered even with no migrations against it, so @@ -102,21 +86,12 @@ func init() { // NewRegistry creates an empty migration registry. func NewRegistry() *Registry { return &Registry{ - kvMigrations: make(map[string][]KVMigration), configMigrations: make(map[string][]ConfigMigration), configTargets: make(map[string]configTarget), objectMigrations: make(map[string][]ObjectMigration), } } -// RegisterKV adds a KV bucket migration. Migrations are kept sorted by FromVersion. -func (r *Registry) RegisterKV(bucket string, m KVMigration) { - r.kvMigrations[bucket] = append(r.kvMigrations[bucket], m) - sort.Slice(r.kvMigrations[bucket], func(i, j int) bool { - return r.kvMigrations[bucket][i].FromVersion < r.kvMigrations[bucket][j].FromVersion - }) -} - // RegisterConfigTarget registers a config file target with its relative path // and version reader. Must be called before RegisterConfig for the same target. func (r *Registry) RegisterConfigTarget(name, relPath string, reader ConfigVersionReader) { @@ -153,7 +128,7 @@ func (r *Registry) RunObject(ctx context.Context, target string, objects objects } if current > targetVersion { - return SchemaAheadError{Bucket: target, Found: current, Understood: targetVersion} + return statemigrate.SchemaAheadError{Bucket: target, Found: current, Understood: targetVersion} } if current == targetVersion { return nil @@ -210,102 +185,6 @@ 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. -func (r *Registry) RunKVWithJetStream(ctx context.Context, bucket string, kv jetstream.KeyValue, js jetstream.JetStream, targetVersion int) error { - return r.runKV(ctx, bucket, kv, js, targetVersion) -} - -// RunKV applies pending KV migrations up to targetVersion. Stamps directly when -// no migrations are registered (fresh bucket). Errors if the chain is incomplete. -func (r *Registry) RunKV(ctx context.Context, bucket string, kv jetstream.KeyValue, targetVersion int) error { - return r.runKV(ctx, bucket, kv, nil, targetVersion) -} - -func (r *Registry) runKV(ctx context.Context, bucket string, kv jetstream.KeyValue, js jetstream.JetStream, targetVersion int) error { - current, err := kvutil.ReadVersion(ctx, kv) - if err != nil { - return fmt.Errorf("read version for %s: %w", bucket, err) - } - - if current > targetVersion { - return SchemaAheadError{Bucket: bucket, Found: current, Understood: targetVersion} - } - if current == targetVersion { - return nil - } - - all := r.kvMigrations[bucket] - - // Fresh bucket, no migrations: stamp directly (common first-init path). - if current == 0 && len(all) == 0 { - return kvutil.WriteVersion(ctx, kv, targetVersion) - } - - // Fresh bucket with migrations: no v0 schema by convention; start at chain bottom. - if current == 0 { - current = all[0].FromVersion // sorted ascending by FromVersion - } - - // Require a complete chain from current to target. - var pending []KVMigration - for _, m := range all { - if m.FromVersion >= current && m.ToVersion <= targetVersion { - pending = append(pending, m) - } - } - - if len(pending) == 0 { - return fmt.Errorf("no migrations registered for %s from version %d to %d", bucket, current, targetVersion) - } - - // Validate contiguous chain. - expected := current - for _, m := range pending { - if m.FromVersion != expected { - return fmt.Errorf("migration chain gap for %s: expected from %d, got from %d", bucket, expected, m.FromVersion) - } - expected = m.ToVersion - } - if expected != targetVersion { - return fmt.Errorf("migration chain for %s ends at version %d, target is %d", bucket, expected, targetVersion) - } - - logger := slog.Default() - for _, m := range pending { - logger.Info("Running KV migration", "bucket", bucket, "from", m.FromVersion, "to", m.ToVersion, "description", m.Description) - kvc := KVContext{KV: kv, JetStream: js, Logger: logger} - if err := m.Run(ctx, kvc); err != nil { - return fmt.Errorf("KV migration %s %dā†’%d failed: %w", bucket, m.FromVersion, m.ToVersion, err) - } - if err := kvutil.WriteVersion(ctx, kv, m.ToVersion); err != nil { - return fmt.Errorf("stamp version %d on %s: %w", m.ToVersion, bucket, err) - } - } - - return nil -} - // RunConfig executes all pending config migrations for a target. // Before each migration step, a timestamped backup is created. // Returns an error on failure (setup.sh should abort). diff --git a/spinifex/migrate/migrate_test.go b/spinifex/migrate/migrate_test.go index a78eabcf4..0d810ee80 100644 --- a/spinifex/migrate/migrate_test.go +++ b/spinifex/migrate/migrate_test.go @@ -9,6 +9,7 @@ import ( "testing" "github.com/mulgadc/spinifex/internal/testkit" + statemigrate "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -34,9 +35,9 @@ func createTestBucket(t *testing.T, nc *nats.Conn, name string) jetstream.KeyVal // --- Registry validation tests --- func TestRegistry_ValidatesChainNoGaps(t *testing.T) { - r := NewRegistry() - r.RegisterKV("test-bucket", KVMigration{FromVersion: 1, ToVersion: 2, Description: "first", Run: func(context.Context, KVContext) error { return nil }}) - r.RegisterKV("test-bucket", KVMigration{FromVersion: 3, ToVersion: 4, Description: "gap", Run: func(context.Context, KVContext) error { return nil }}) + r := statemigrate.NewRegistry() + r.RegisterKV("test-bucket", statemigrate.KVMigration{FromVersion: 1, ToVersion: 2, Description: "first", Run: func(context.Context, statemigrate.KVContext) error { return nil }}) + r.RegisterKV("test-bucket", statemigrate.KVMigration{FromVersion: 3, ToVersion: 4, Description: "gap", Run: func(context.Context, statemigrate.KVContext) error { return nil }}) _, nc := startTestNATS(t) kv := createTestBucket(t, nc, "test-bucket") @@ -50,9 +51,9 @@ func TestRegistry_ValidatesChainNoGaps(t *testing.T) { } func TestRegistry_RejectsDuplicateVersions(t *testing.T) { - r := NewRegistry() - r.RegisterKV("test-bucket", KVMigration{FromVersion: 1, ToVersion: 2, Description: "first", Run: func(context.Context, KVContext) error { return nil }}) - r.RegisterKV("test-bucket", KVMigration{FromVersion: 1, ToVersion: 2, Description: "duplicate", Run: func(context.Context, KVContext) error { return nil }}) + r := statemigrate.NewRegistry() + r.RegisterKV("test-bucket", statemigrate.KVMigration{FromVersion: 1, ToVersion: 2, Description: "first", Run: func(context.Context, statemigrate.KVContext) error { return nil }}) + r.RegisterKV("test-bucket", statemigrate.KVMigration{FromVersion: 1, ToVersion: 2, Description: "duplicate", Run: func(context.Context, statemigrate.KVContext) error { return nil }}) _, nc := startTestNATS(t) kv := createTestBucket(t, nc, "test-bucket") @@ -72,7 +73,7 @@ func TestRegistry_RejectsDuplicateVersions(t *testing.T) { // --- RunKV tests --- func TestRunKV_NoPendingMigrations_NoOp(t *testing.T) { - r := NewRegistry() + r := statemigrate.NewRegistry() _, nc := startTestNATS(t) kv := createTestBucket(t, nc, "test-bucket") @@ -87,10 +88,10 @@ func TestRunKV_NoPendingMigrations_NoOp(t *testing.T) { func TestRunKV_FreshBucket_WithMigrations(t *testing.T) { ran := false - r := NewRegistry() - r.RegisterKV("test-bucket", KVMigration{ + r := statemigrate.NewRegistry() + r.RegisterKV("test-bucket", statemigrate.KVMigration{ FromVersion: 1, ToVersion: 2, Description: "first kv migration", - Run: func(context.Context, KVContext) error { ran = true; return nil }, + Run: func(context.Context, statemigrate.KVContext) error { ran = true; return nil }, }) _, nc := startTestNATS(t) @@ -108,7 +109,7 @@ func TestRunKV_FreshBucket_WithMigrations(t *testing.T) { } func TestRunKV_FreshBucket_StampsVersion(t *testing.T) { - r := NewRegistry() + r := statemigrate.NewRegistry() _, nc := startTestNATS(t) kv := createTestBucket(t, nc, "test-bucket") @@ -124,14 +125,14 @@ func TestRunKV_FreshBucket_StampsVersion(t *testing.T) { func TestRunKV_ExecutesMigrationsInOrder(t *testing.T) { var order []int - r := NewRegistry() - r.RegisterKV("test-bucket", KVMigration{ + r := statemigrate.NewRegistry() + r.RegisterKV("test-bucket", statemigrate.KVMigration{ FromVersion: 1, ToVersion: 2, Description: "step 1", - Run: func(context.Context, KVContext) error { order = append(order, 1); return nil }, + Run: func(context.Context, statemigrate.KVContext) error { order = append(order, 1); return nil }, }) - r.RegisterKV("test-bucket", KVMigration{ + r.RegisterKV("test-bucket", statemigrate.KVMigration{ FromVersion: 2, ToVersion: 3, Description: "step 2", - Run: func(context.Context, KVContext) error { order = append(order, 2); return nil }, + Run: func(context.Context, statemigrate.KVContext) error { order = append(order, 2); return nil }, }) _, nc := startTestNATS(t) @@ -150,17 +151,17 @@ func TestRunKV_ExecutesMigrationsInOrder(t *testing.T) { } func TestRunKV_StampsAfterEachStep(t *testing.T) { - r := NewRegistry() - r.RegisterKV("test-bucket", KVMigration{ + r := statemigrate.NewRegistry() + r.RegisterKV("test-bucket", statemigrate.KVMigration{ FromVersion: 1, ToVersion: 2, Description: "step 1", - Run: func(context.Context, KVContext) error { + Run: func(context.Context, statemigrate.KVContext) error { // After this runs, version should be stamped to 2 by RunKV. return nil }, }) - r.RegisterKV("test-bucket", KVMigration{ + r.RegisterKV("test-bucket", statemigrate.KVMigration{ FromVersion: 2, ToVersion: 3, Description: "step 2 fails", - Run: func(context.Context, KVContext) error { return errors.New("boom") }, + Run: func(context.Context, statemigrate.KVContext) error { return errors.New("boom") }, }) _, nc := startTestNATS(t) @@ -179,10 +180,10 @@ func TestRunKV_StampsAfterEachStep(t *testing.T) { } func TestRunKV_StopsOnFailure_VersionNotBumped(t *testing.T) { - r := NewRegistry() - r.RegisterKV("test-bucket", KVMigration{ + r := statemigrate.NewRegistry() + r.RegisterKV("test-bucket", statemigrate.KVMigration{ FromVersion: 1, ToVersion: 2, Description: "fails", - Run: func(context.Context, KVContext) error { return errors.New("migration error") }, + Run: func(context.Context, statemigrate.KVContext) error { return errors.New("migration error") }, }) _, nc := startTestNATS(t) @@ -200,7 +201,7 @@ func TestRunKV_StopsOnFailure_VersionNotBumped(t *testing.T) { } func TestRunKV_RejectsMissingMigration(t *testing.T) { - r := NewRegistry() + r := statemigrate.NewRegistry() // No migrations registered, but bucket is at version 1 and target is 2. _, nc := startTestNATS(t) @@ -215,10 +216,10 @@ func TestRunKV_RejectsMissingMigration(t *testing.T) { func TestRunKV_Idempotent(t *testing.T) { runCount := 0 - r := NewRegistry() - r.RegisterKV("test-bucket", KVMigration{ + r := statemigrate.NewRegistry() + r.RegisterKV("test-bucket", statemigrate.KVMigration{ FromVersion: 1, ToVersion: 2, Description: "add field", - Run: func(ctx context.Context, kvc KVContext) error { + Run: func(ctx context.Context, kvc statemigrate.KVContext) error { runCount++ // Idempotent: write a key, re-running writes the same value. _, err := kvc.KV.PutString(ctx, "data.key1", `{"field":"value"}`) @@ -247,10 +248,10 @@ func TestRunKV_Idempotent(t *testing.T) { // 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{ + r := statemigrate.NewRegistry() + r.RegisterKV("test-bucket", statemigrate.KVMigration{ FromVersion: 1, ToVersion: 2, Description: "add field", - Run: func(ctx context.Context, kvc KVContext) error { return nil }, + Run: func(ctx context.Context, kvc statemigrate.KVContext) error { return nil }, }) _, nc := startTestNATS(t) @@ -260,7 +261,7 @@ func TestRunKV_SchemaAheadOfThisBuildIsRefused(t *testing.T) { err = r.RunKV(t.Context(), "test-bucket", kv, 2) - var ahead SchemaAheadError + var ahead statemigrate.SchemaAheadError require.ErrorAs(t, err, &ahead) assert.Equal(t, 5, ahead.Found) assert.Equal(t, 2, ahead.Understood) diff --git a/spinifex/network/external/static_pool.go b/spinifex/network/external/static_pool.go index b6d940df1..0c6145b60 100644 --- a/spinifex/network/external/static_pool.go +++ b/spinifex/network/external/static_pool.go @@ -10,7 +10,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" "github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore" - "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/foundation/state/migrate" "github.com/nats-io/nats.go/jetstream" )