refactor: extract KV migration engine to state foundation

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"
 )

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"

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
+}

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"

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"

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"

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"

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"

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"

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"

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"

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"

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"

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"
 )
 

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"

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"
 )

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"
 )

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"
 )

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"
 )
 

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"
 )

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"
 )
 

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"

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).

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)

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"
 )