feat(ipam): Purpose field on every IPAM record (Slice B)

Part of bead mulga-siv-125.3.3. Adds a Purpose enum (handlers/ec2/vpc/purpose.go) tagging every IP allocation with its semantic role (eni-primary, eni-public, eip, igw-lrp, natgw-external, …) so multi-VPC clusters can reclaim/audit by (owner, purpose). External IPAM (spinifex-external-ipam, v1→v2): - ExternalIPAllocation.Type → .Purpose; legacy values mapped via LegacyExternalTypeToPurpose (gateway→igw-lrp, auto_assign→eni-public, elastic_ip→eip). Unknown legacy values warn and become "unknown". - AllocateIP / AllocateFromPool / allocateFromPool param rename allocType→purpose. All call sites pass Purpose* constants. - migrate/002_external_ipam_purpose.go: idempotent CAS-retry migration with v1/v2-union record shape so partial migrations resume cleanly. Internal IPAM (spinifex-vpc-ipam, v1→v2): - IPAMRecord.Allocated []string → []IPEntry{IP, Purpose, OwnerID}. - AllocateIP gains purpose+ownerID args; ReleaseIP unchanged (still keyed on IP). - AllocatedIPs returns []IPEntry. - eni.go reorders eniId generation before AllocateIP so the owning resource ID lands on the new entry. - migrate/003_ipam_purpose.go: cross-bucket backfill that reads the ENI bucket via the new RunKVWithJetStream path. Orphaned IPs (no owning ENI) warn and become Purpose=unknown. Migrate framework: - KVContext gains an optional JetStream field; new Registry.RunKVWithJetStream method threads it through. RunKV signature unchanged — all other callers are unaffected. Tests: - ipam_test.go: AllocateIP signature updates; AllocatedIPs now asserts IP / Purpose / OwnerID on the returned entries. - external_ipam_test.go: replaces "auto_assign"/"elastic_ip" literals with PurposeENIPublic/PurposeEIP and asserts alloc.Purpose. - migrate/002 + 003 tests: schema-detect, idempotent rerun, no-JetStream fallback, empty bucket.
2026-10-15 17:11:30 +00:00 · 2026-05-22 13:53:01 +10:00
parent 5e3f1090a8
commit 0bb0fcc532
14 changed files with 761 additions and 80 deletions
@@ -10,6 +10,7 @@ import (

 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
+	handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
 	handlers_elbv2 "github.com/mulgadc/spinifex/spinifex/handlers/elbv2"
 	"github.com/mulgadc/spinifex/spinifex/tags"
 	"github.com/mulgadc/spinifex/spinifex/utils"
@@ -209,7 +210,7 @@ func (d *Daemon) LaunchSystemInstance(input *handlers_elbv2.SystemInstanceInput)
 				region = d.config.Region
 				az = d.config.AZ
 			}
-			allocatedIP, poolName, allocErr := d.externalIPAM.AllocateIP(region, az, "auto_assign", "", instance.ENIId, instance.ID)
+			allocatedIP, poolName, allocErr := d.externalIPAM.AllocateIP(region, az, handlers_ec2_vpc.PurposeENIPublic, "", instance.ENIId, instance.ID)
 			if allocErr != nil {
 				slog.Error("LaunchSystemInstance: failed to allocate public IP for internet-facing ALB", "instanceId", instance.ID, "err", allocErr)
 				d.cleanupFailedSystemInstance(instance, instanceType)
@@ -73,7 +73,7 @@ func (s *EIPServiceImpl) AllocateAddress(input *ec2.AllocateAddressInput, accoun
 	if input.PublicIpv4Pool != nil && *input.PublicIpv4Pool != "" {
 		// Allocate from a specific named pool.
 		poolName = *input.PublicIpv4Pool
-		publicIP, err = s.externalIPAM.AllocateFromPool(poolName, "elastic_ip", allocID, "", "")
+		publicIP, err = s.externalIPAM.AllocateFromPool(poolName, handlers_ec2_vpc.PurposeEIP, allocID, "", "")
 		if err != nil {
 			slog.Error("AllocateAddress: IPAM pool allocation failed", "pool", poolName, "err", err)
 			return nil, errors.New(awserrors.ErrorInsufficientAddressCapacity)
@@ -82,7 +82,7 @@ func (s *EIPServiceImpl) AllocateAddress(input *ec2.AllocateAddressInput, accoun
 		// Allocate from the best pool matching region/AZ (empty strings = global fallback).
 		region := ""
 		az := ""
-		publicIP, poolName, err = s.externalIPAM.AllocateIP(region, az, "elastic_ip", allocID, "", "")
+		publicIP, poolName, err = s.externalIPAM.AllocateIP(region, az, handlers_ec2_vpc.PurposeEIP, allocID, "", "")
 		if err != nil {
 			slog.Error("AllocateAddress: IPAM allocation failed", "err", err)
 			return nil, errors.New(awserrors.ErrorInsufficientAddressCapacity)
@@ -22,6 +22,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
 	"github.com/mulgadc/spinifex/spinifex/config"
 	"github.com/mulgadc/spinifex/spinifex/filterutil"
+	handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
 	"github.com/mulgadc/spinifex/spinifex/instancetypes"
 	"github.com/mulgadc/spinifex/spinifex/objectstore"
 	spxtypes "github.com/mulgadc/spinifex/spinifex/types"
@@ -689,7 +690,7 @@ func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput,
 				if wantPublic {
 					region := s.config.Region
 					az := s.config.AZ
-					publicIP, poolName, allocErr := s.ipAllocator.AllocateIP(region, az, "auto_assign", "", *eni.NetworkInterfaceId, instance.ID)
+					publicIP, poolName, allocErr := s.ipAllocator.AllocateIP(region, az, handlers_ec2_vpc.PurposeENIPublic, "", *eni.NetworkInterfaceId, instance.ID)
 					if allocErr != nil {
 						slog.Warn("PrepareRunInstances: failed to allocate public IP", "instanceId", instance.ID, "err", allocErr)
 					} else {
@@ -78,21 +78,21 @@ func (s *VPCServiceImpl) CreateNetworkInterface(input *ec2.CreateNetworkInterfac
 		return nil, err
 	}

+	eniId := utils.GenerateResourceID("eni")
+
 	// Allocate IP from subnet
 	var privateIP string
 	if input.PrivateIpAddress != nil && *input.PrivateIpAddress != "" {
 		// TODO: validate the requested IP is in the subnet range and not already allocated
 		privateIP = *input.PrivateIpAddress
 	} else {
-		ip, err := s.ipam.AllocateIP(subnetId, subnet.CidrBlock)
+		ip, err := s.ipam.AllocateIP(subnetId, subnet.CidrBlock, PurposeENIPrimary, eniId)
 		if err != nil {
 			return nil, errors.New(awserrors.ErrorServerInternal)
 		}
 		privateIP = ip
 	}

-	eniId := utils.GenerateResourceID("eni")
-
 	// Generate a deterministic MAC address
 	macAddr := generateENIMac(eniId)

@@ -15,12 +15,13 @@ import (

 const (
 	KVBucketExternalIPAM        = "spinifex-external-ipam"
-	KVBucketExternalIPAMVersion = 1
+	KVBucketExternalIPAMVersion = 2
 )

-// ExternalIPAllocation describes how an external IP is being used.
+// ExternalIPAllocation describes how an external IP is being used. Purpose
+// values come from the Purpose enum in purpose.go.
 type ExternalIPAllocation struct {
-	Type         string `json:"type"`                    // "gateway", "auto_assign", "elastic_ip"
+	Purpose      string `json:"purpose"`                 // Purpose enum: igw-lrp, eni-public, eip, natgw-external
 	AllocationID string `json:"allocation_id,omitempty"` // For elastic IPs
 	Association  string `json:"association,omitempty"`   // ENI ID for elastic IPs
 	ENIId        string `json:"eni_id,omitempty"`        // ENI owning this IP
@@ -156,7 +157,7 @@ func (m *ExternalIPAM) initPool(pool ExternalPoolConfig) error {
 		GwLrpRangeStart: pool.GwLrpRangeStart,
 		GwLrpRangeEnd:   pool.GwLrpRangeEnd,
 		Allocated: map[string]ExternalIPAllocation{
-			gwIP: {Type: "gateway", Note: "OVN router SNAT address"},
+			gwIP: {Purpose: PurposeIGWLRP, Note: "OVN router SNAT address"},
 		},
 	}

@@ -179,13 +180,14 @@ func (m *ExternalIPAM) initPool(pool ExternalPoolConfig) error {

 // AllocateIP allocates the next available external IP from the best pool
 // matching the given region/AZ. Returns the allocated IP and pool name.
-// allocID is the EIP allocation ID (for elastic IPs); pass "" for other types.
-func (m *ExternalIPAM) AllocateIP(region, az, allocType, allocID, eniID, instanceID string) (string, string, error) {
+// purpose is one of the Purpose* constants in purpose.go; allocID is the
+// EIP allocation ID (for PurposeEIP) and "" otherwise.
+func (m *ExternalIPAM) AllocateIP(region, az, purpose, allocID, eniID, instanceID string) (string, string, error) {
 	pool := m.findPool(region, az)
 	if pool == nil {
 		return "", "", fmt.Errorf("InsufficientAddressCapacity: no external pool available for region=%q az=%q", region, az)
 	}
-	ip, err := m.allocateFromPool(pool.Name, allocType, allocID, eniID, instanceID)
+	ip, err := m.allocateFromPool(pool.Name, purpose, allocID, eniID, instanceID)
 	if err != nil {
 		return "", "", err
 	}
@@ -193,11 +195,11 @@ func (m *ExternalIPAM) AllocateIP(region, az, allocType, allocID, eniID, instanc
 }

 // AllocateFromPool allocates an IP from a specific named pool.
-func (m *ExternalIPAM) AllocateFromPool(poolName, allocType, allocID, eniID, instanceID string) (string, error) {
-	return m.allocateFromPool(poolName, allocType, allocID, eniID, instanceID)
+func (m *ExternalIPAM) AllocateFromPool(poolName, purpose, allocID, eniID, instanceID string) (string, error) {
+	return m.allocateFromPool(poolName, purpose, allocID, eniID, instanceID)
 }

-func (m *ExternalIPAM) allocateFromPool(poolName, allocType, allocID, eniID, instanceID string) (string, error) {
+func (m *ExternalIPAM) allocateFromPool(poolName, purpose, allocID, eniID, instanceID string) (string, error) {
 	for attempt := range 5 {
 		record, revision, err := m.getRecord(poolName)
 		if err != nil {
@@ -210,7 +212,7 @@ func (m *ExternalIPAM) allocateFromPool(poolName, allocType, allocID, eniID, ins
 		}

 		record.Allocated[ip] = ExternalIPAllocation{
-			Type:         allocType,
+			Purpose:      purpose,
 			AllocationID: allocID,
 			ENIId:        eniID,
 			InstanceId:   instanceID,
@@ -226,7 +228,7 @@ func (m *ExternalIPAM) allocateFromPool(poolName, allocType, allocID, eniID, ins
 			continue
 		}

-		slog.Info("external IPAM allocated IP", "pool", poolName, "ip", ip, "type", allocType)
+		slog.Info("external IPAM allocated IP", "pool", poolName, "ip", ip, "purpose", purpose)
 		return ip, nil
 	}

@@ -245,7 +247,7 @@ func (m *ExternalIPAM) ReleaseIP(poolName, ip string) error {
 		if !ok {
 			return fmt.Errorf("IP %s not allocated in pool %s", ip, poolName)
 		}
-		if alloc.Type == "gateway" {
+		if alloc.Purpose == PurposeIGWLRP {
 			return fmt.Errorf("cannot release gateway IP %s in pool %s", ip, poolName)
 		}

@@ -33,16 +33,16 @@ func TestExternalIPAM_AllocateSequential(t *testing.T) {
 	ipam := setupTestExternalIPAM(t, []ExternalPoolConfig{pool})

 	// .150 is reserved for gateway, so first allocable is .151
-	ip1, poolName, err := ipam.AllocateIP("", "", "auto_assign", "", "eni-1", "i-1")
+	ip1, poolName, err := ipam.AllocateIP("", "", PurposeENIPublic, "", "eni-1", "i-1")
 	require.NoError(t, err)
 	assert.Equal(t, "192.168.1.151", ip1)
 	assert.Equal(t, "wan", poolName)

-	ip2, _, err := ipam.AllocateIP("", "", "auto_assign", "", "eni-2", "i-2")
+	ip2, _, err := ipam.AllocateIP("", "", PurposeENIPublic, "", "eni-2", "i-2")
 	require.NoError(t, err)
 	assert.Equal(t, "192.168.1.152", ip2)

-	ip3, _, err := ipam.AllocateIP("", "", "auto_assign", "", "eni-3", "i-3")
+	ip3, _, err := ipam.AllocateIP("", "", PurposeENIPublic, "", "eni-3", "i-3")
 	require.NoError(t, err)
 	assert.Equal(t, "192.168.1.153", ip3)
 }
@@ -57,7 +57,7 @@ func TestExternalIPAM_GatewayIPReserved(t *testing.T) {
 	// Gateway IP (.150) should be reserved
 	alloc, ok := record.Allocated["192.168.1.150"]
 	assert.True(t, ok, "gateway IP should be in allocated map")
-	assert.Equal(t, "gateway", alloc.Type)
+	assert.Equal(t, PurposeIGWLRP, alloc.Purpose)

 	// Cannot release gateway IP
 	err = ipam.ReleaseIP("wan", "192.168.1.150")
@@ -79,7 +79,7 @@ func TestExternalIPAM_ExplicitGatewayIP(t *testing.T) {
 	assert.False(t, ok)

 	// First allocable is .150 since .155 is the gateway
-	ip1, _, err := ipam.AllocateIP("", "", "auto_assign", "", "eni-1", "i-1")
+	ip1, _, err := ipam.AllocateIP("", "", PurposeENIPublic, "", "eni-1", "i-1")
 	require.NoError(t, err)
 	assert.Equal(t, "192.168.1.150", ip1)
 }
@@ -88,11 +88,11 @@ func TestExternalIPAM_Release(t *testing.T) {
 	pool := testPool()
 	ipam := setupTestExternalIPAM(t, []ExternalPoolConfig{pool})

-	ip1, _, err := ipam.AllocateIP("", "", "auto_assign", "", "eni-1", "i-1")
+	ip1, _, err := ipam.AllocateIP("", "", PurposeENIPublic, "", "eni-1", "i-1")
 	require.NoError(t, err)
 	assert.Equal(t, "192.168.1.151", ip1)

-	ip2, _, err := ipam.AllocateIP("", "", "auto_assign", "", "eni-2", "i-2")
+	ip2, _, err := ipam.AllocateIP("", "", PurposeENIPublic, "", "eni-2", "i-2")
 	require.NoError(t, err)
 	assert.Equal(t, "192.168.1.152", ip2)

@@ -101,7 +101,7 @@ func TestExternalIPAM_Release(t *testing.T) {
 	require.NoError(t, err)

 	// Next allocation reuses released IP
-	ip3, _, err := ipam.AllocateIP("", "", "auto_assign", "", "eni-3", "i-3")
+	ip3, _, err := ipam.AllocateIP("", "", PurposeENIPublic, "", "eni-3", "i-3")
 	require.NoError(t, err)
 	assert.Equal(t, "192.168.1.151", ip3)
 }
@@ -117,16 +117,16 @@ func TestExternalIPAM_Exhaustion(t *testing.T) {
 	ipam := setupTestExternalIPAM(t, []ExternalPoolConfig{pool})

 	// .1 reserved for gateway, .2 and .3 allocable
-	ip1, _, err := ipam.AllocateIP("", "", "auto_assign", "", "eni-1", "i-1")
+	ip1, _, err := ipam.AllocateIP("", "", PurposeENIPublic, "", "eni-1", "i-1")
 	require.NoError(t, err)
 	assert.Equal(t, "10.0.0.2", ip1)

-	ip2, _, err := ipam.AllocateIP("", "", "auto_assign", "", "eni-2", "i-2")
+	ip2, _, err := ipam.AllocateIP("", "", PurposeENIPublic, "", "eni-2", "i-2")
 	require.NoError(t, err)
 	assert.Equal(t, "10.0.0.3", ip2)

 	// Pool exhausted
-	_, _, err = ipam.AllocateIP("", "", "auto_assign", "", "eni-3", "i-3")
+	_, _, err = ipam.AllocateIP("", "", PurposeENIPublic, "", "eni-3", "i-3")
 	assert.ErrorContains(t, err, "InsufficientAddressCapacity")
 }

@@ -143,7 +143,7 @@ func TestExternalIPAM_CASConflict(t *testing.T) {
 		wg.Add(1)
 		go func(idx int) {
 			defer wg.Done()
-			ip, _, err := ipam.AllocateIP("", "", "auto_assign", "", "eni-"+itoa(idx), "i-"+itoa(idx))
+			ip, _, err := ipam.AllocateIP("", "", PurposeENIPublic, "", "eni-"+itoa(idx), "i-"+itoa(idx))
 			results[idx] = ip
 			errs[idx] = err
 		}(i)
@@ -187,13 +187,13 @@ func TestExternalIPAM_MultiPool(t *testing.T) {
 	ipam := setupTestExternalIPAM(t, pools)

 	// Allocate from US pool
-	ip1, poolName1, err := ipam.AllocateIP("us-east-1", "us-east-1a", "auto_assign", "", "eni-1", "i-1")
+	ip1, poolName1, err := ipam.AllocateIP("us-east-1", "us-east-1a", PurposeENIPublic, "", "eni-1", "i-1")
 	require.NoError(t, err)
 	assert.Equal(t, "203.0.113.3", ip1) // .2 reserved for gateway
 	assert.Equal(t, "us-east", poolName1)

 	// Allocate from EU pool
-	ip2, poolName2, err := ipam.AllocateIP("eu-west-1", "eu-west-1a", "auto_assign", "", "eni-2", "i-2")
+	ip2, poolName2, err := ipam.AllocateIP("eu-west-1", "eu-west-1a", PurposeENIPublic, "", "eni-2", "i-2")
 	require.NoError(t, err)
 	assert.Equal(t, "198.51.100.3", ip2)
 	assert.Equal(t, "eu-west", poolName2)
@@ -229,19 +229,19 @@ func TestExternalIPAM_PoolFallback(t *testing.T) {
 	ipam := setupTestExternalIPAM(t, pools)

 	// AZ match: us-east-1a → az-pool
-	ip1, pool1, err := ipam.AllocateIP("us-east-1", "us-east-1a", "auto_assign", "", "eni-1", "i-1")
+	ip1, pool1, err := ipam.AllocateIP("us-east-1", "us-east-1a", PurposeENIPublic, "", "eni-1", "i-1")
 	require.NoError(t, err)
 	assert.Equal(t, "az-pool", pool1)
 	assert.Equal(t, "10.0.0.2", ip1) // .1 reserved for gw

 	// Region match (different AZ): us-east-1b → region-pool
-	ip2, pool2, err := ipam.AllocateIP("us-east-1", "us-east-1b", "auto_assign", "", "eni-2", "i-2")
+	ip2, pool2, err := ipam.AllocateIP("us-east-1", "us-east-1b", PurposeENIPublic, "", "eni-2", "i-2")
 	require.NoError(t, err)
 	assert.Equal(t, "region-pool", pool2)
 	assert.Equal(t, "10.0.1.2", ip2)

 	// No match: eu-west-1 → global-pool
-	ip3, pool3, err := ipam.AllocateIP("eu-west-1", "eu-west-1a", "auto_assign", "", "eni-3", "i-3")
+	ip3, pool3, err := ipam.AllocateIP("eu-west-1", "eu-west-1a", PurposeENIPublic, "", "eni-3", "i-3")
 	require.NoError(t, err)
 	assert.Equal(t, "global-pool", pool3)
 	assert.Equal(t, "10.0.2.2", ip3)
@@ -267,7 +267,7 @@ func TestExternalIPAM_SpecificPool(t *testing.T) {
 	ipam := setupTestExternalIPAM(t, pools)

 	// Allocate specifically from pool-b
-	ip, err := ipam.AllocateFromPool("pool-b", "elastic_ip", "eipalloc-test-b", "", "")
+	ip, err := ipam.AllocateFromPool("pool-b", PurposeEIP, "eipalloc-test-b", "", "")
 	require.NoError(t, err)
 	assert.Equal(t, "10.0.1.2", ip) // .1 reserved for gw
 }
@@ -341,7 +341,7 @@ func TestExternalIPAM_InitFromConfig(t *testing.T) {
 	require.NoError(t, err)

 	// Allocate an IP
-	ip, _, err := ipam1.AllocateIP("", "", "auto_assign", "", "eni-1", "i-1")
+	ip, _, err := ipam1.AllocateIP("", "", PurposeENIPublic, "", "eni-1", "i-1")
 	require.NoError(t, err)
 	assert.Equal(t, "192.168.1.151", ip)

@@ -376,7 +376,7 @@ func TestExternalIPAM_NoPoolAvailable(t *testing.T) {
 	ipam := setupTestExternalIPAM(t, []ExternalPoolConfig{pool})

 	// No pool matches eu-west
-	_, _, err := ipam.AllocateIP("eu-west-1", "eu-west-1a", "auto_assign", "", "eni-1", "i-1")
+	_, _, err := ipam.AllocateIP("eu-west-1", "eu-west-1a", PurposeENIPublic, "", "eni-1", "i-1")
 	assert.ErrorContains(t, err, "InsufficientAddressCapacity")
 }

@@ -388,7 +388,7 @@ func TestExternalIPAM_FindPoolByName_NotFound(t *testing.T) {
 	// has no KV record, so getRecord fails.
 	kv := ipam.kv
 	ipam2 := NewExternalIPAMWithKV(kv, []ExternalPoolConfig{pool})
-	_, err := ipam2.AllocateFromPool("nonexistent", "auto_assign", "", "eni-1", "i-1")
+	_, err := ipam2.AllocateFromPool("nonexistent", PurposeENIPublic, "", "eni-1", "i-1")
 	assert.Error(t, err)
 }

@@ -15,14 +15,22 @@ import (

 const (
 	KVBucketIPAM        = "spinifex-vpc-ipam"
-	KVBucketIPAMVersion = 1
+	KVBucketIPAMVersion = 2
 )

+// IPEntry tags one IP allocation with its Purpose + owner (eni-, eipalloc-,
+// etc) so multi-VPC clusters can reclaim/audit by (owner, purpose).
+type IPEntry struct {
+	IP      string `json:"ip"`
+	Purpose string `json:"purpose"`            // one of the Purpose* constants
+	OwnerID string `json:"owner_id,omitempty"` // ENI / EIP / NATGW / IGW resource ID
+}
+
 // IPAMRecord tracks allocated IPs for a subnet.
 type IPAMRecord struct {
-	SubnetId  string   `json:"subnet_id"`
-	CidrBlock string   `json:"cidr_block"`
-	Allocated []string `json:"allocated"` // List of allocated IPs
+	SubnetId  string    `json:"subnet_id"`
+	CidrBlock string    `json:"cidr_block"`
+	Allocated []IPEntry `json:"allocated"`
 }

 // IPAM manages IP address allocation for VPC subnets using NATS KV with CAS.
@@ -36,7 +44,7 @@ func NewIPAM(js nats.JetStreamContext) (*IPAM, error) {
 	if err != nil {
 		return nil, fmt.Errorf("failed to create IPAM KV bucket: %w", err)
 	}
-	if err := migrate.DefaultRegistry.RunKV(KVBucketIPAM, kv, KVBucketIPAMVersion); err != nil {
+	if err := migrate.DefaultRegistry.RunKVWithJetStream(KVBucketIPAM, kv, js, KVBucketIPAMVersion); err != nil {
 		return nil, fmt.Errorf("migrate %s: %w", KVBucketIPAM, err)
 	}
 	return &IPAM{kv: kv}, nil
@@ -50,8 +58,10 @@ func NewIPAMWithKV(kv nats.KeyValue) *IPAM {
 // AllocateIP allocates an IP address from the given subnet.
 // It reserves the first 4 and last IP per AWS convention:
 // .0=network, .1=gateway, .2=DNS, .3=reserved, .255=broadcast (for /24).
-// Uses CAS for conflict-free allocation across nodes.
-func (m *IPAM) AllocateIP(subnetId, cidrBlock string) (string, error) {
+// Uses CAS for conflict-free allocation across nodes. purpose is one of the
+// Purpose* constants in purpose.go; ownerID is the ENI / instance / EIP
+// resource ID that owns this allocation.
+func (m *IPAM) AllocateIP(subnetId, cidrBlock, purpose, ownerID string) (string, error) {
 	for attempt := range 5 {
 		record, revision, err := m.getRecord(subnetId)
 		if err != nil && !errors.Is(err, nats.ErrKeyNotFound) {
@@ -71,7 +81,7 @@ func (m *IPAM) AllocateIP(subnetId, cidrBlock string) (string, error) {
 			return "", err
 		}

-		record.Allocated = append(record.Allocated, ip)
+		record.Allocated = append(record.Allocated, IPEntry{IP: ip, Purpose: purpose, OwnerID: ownerID})

 		// CAS write
 		data, err := json.Marshal(record)
@@ -90,7 +100,7 @@ func (m *IPAM) AllocateIP(subnetId, cidrBlock string) (string, error) {
 			continue // CAS conflict, retry
 		}

-		slog.Info("IPAM allocated IP", "subnet", subnetId, "ip", ip)
+		slog.Info("IPAM allocated IP", "subnet", subnetId, "ip", ip, "purpose", purpose, "owner", ownerID)
 		return ip, nil
 	}

@@ -109,8 +119,8 @@ func (m *IPAM) ReleaseIP(subnetId, ip string) error {
 		}

 		found := false
-		for i, allocated := range record.Allocated {
-			if allocated == ip {
+		for i, entry := range record.Allocated {
+			if entry.IP == ip {
 				record.Allocated = append(record.Allocated[:i], record.Allocated[i+1:]...)
 				found = true
 				break
@@ -137,8 +147,8 @@ func (m *IPAM) ReleaseIP(subnetId, ip string) error {
 	return fmt.Errorf("IPAM release failed after CAS retries for subnet %s", subnetId)
 }

-// AllocatedIPs returns the list of allocated IPs for a subnet.
-func (m *IPAM) AllocatedIPs(subnetId string) ([]string, error) {
+// AllocatedIPs returns the list of allocated IP entries for a subnet.
+func (m *IPAM) AllocatedIPs(subnetId string) ([]IPEntry, error) {
 	record, _, err := m.getRecord(subnetId)
 	if err != nil {
 		if errors.Is(err, nats.ErrKeyNotFound) {
@@ -177,8 +187,8 @@ func (m *IPAM) nextAvailableIP(record *IPAMRecord) (string, error) {
 	}

 	allocated := make(map[string]bool, len(record.Allocated))
-	for _, ip := range record.Allocated {
-		allocated[ip] = true
+	for _, entry := range record.Allocated {
+		allocated[entry.IP] = true
 	}

 	ones, bits := ipNet.Mask.Size()
@@ -22,7 +22,7 @@ func setupTestIPAM(t *testing.T) *IPAM {
 func TestIPAM_AllocateFirst(t *testing.T) {
 	ipam := setupTestIPAM(t)

-	ip, err := ipam.AllocateIP("subnet-1", "10.0.1.0/24")
+	ip, err := ipam.AllocateIP("subnet-1", "10.0.1.0/24", PurposeENIPrimary, "eni-1")
 	require.NoError(t, err)
 	// First allocable IP is .4 (skip .0 network, .1 gateway, .2 DNS, .3 reserved)
 	assert.Equal(t, "10.0.1.4", ip)
@@ -31,15 +31,15 @@ func TestIPAM_AllocateFirst(t *testing.T) {
 func TestIPAM_AllocateSequential(t *testing.T) {
 	ipam := setupTestIPAM(t)

-	ip1, err := ipam.AllocateIP("subnet-seq", "10.0.2.0/24")
+	ip1, err := ipam.AllocateIP("subnet-seq", "10.0.2.0/24", PurposeENIPrimary, "eni-seq")
 	require.NoError(t, err)
 	assert.Equal(t, "10.0.2.4", ip1)

-	ip2, err := ipam.AllocateIP("subnet-seq", "10.0.2.0/24")
+	ip2, err := ipam.AllocateIP("subnet-seq", "10.0.2.0/24", PurposeENIPrimary, "eni-seq")
 	require.NoError(t, err)
 	assert.Equal(t, "10.0.2.5", ip2)

-	ip3, err := ipam.AllocateIP("subnet-seq", "10.0.2.0/24")
+	ip3, err := ipam.AllocateIP("subnet-seq", "10.0.2.0/24", PurposeENIPrimary, "eni-seq")
 	require.NoError(t, err)
 	assert.Equal(t, "10.0.2.6", ip3)
 }
@@ -47,11 +47,11 @@ func TestIPAM_AllocateSequential(t *testing.T) {
 func TestIPAM_Release(t *testing.T) {
 	ipam := setupTestIPAM(t)

-	ip1, err := ipam.AllocateIP("subnet-rel", "10.0.3.0/24")
+	ip1, err := ipam.AllocateIP("subnet-rel", "10.0.3.0/24", PurposeENIPrimary, "eni-rel")
 	require.NoError(t, err)
 	assert.Equal(t, "10.0.3.4", ip1)

-	ip2, err := ipam.AllocateIP("subnet-rel", "10.0.3.0/24")
+	ip2, err := ipam.AllocateIP("subnet-rel", "10.0.3.0/24", PurposeENIPrimary, "eni-rel")
 	require.NoError(t, err)
 	assert.Equal(t, "10.0.3.5", ip2)

@@ -60,7 +60,7 @@ func TestIPAM_Release(t *testing.T) {
 	require.NoError(t, err)

 	// Next allocation should reuse the released IP
-	ip3, err := ipam.AllocateIP("subnet-rel", "10.0.3.0/24")
+	ip3, err := ipam.AllocateIP("subnet-rel", "10.0.3.0/24", PurposeENIPrimary, "eni-rel")
 	require.NoError(t, err)
 	assert.Equal(t, "10.0.3.4", ip3)
 }
@@ -68,7 +68,7 @@ func TestIPAM_Release(t *testing.T) {
 func TestIPAM_ReleaseNotAllocated(t *testing.T) {
 	ipam := setupTestIPAM(t)

-	_, err := ipam.AllocateIP("subnet-rna", "10.0.4.0/24")
+	_, err := ipam.AllocateIP("subnet-rna", "10.0.4.0/24", PurposeENIPrimary, "eni-rna")
 	require.NoError(t, err)

 	err = ipam.ReleaseIP("subnet-rna", "10.0.4.99")
@@ -91,7 +91,7 @@ func TestIPAM_Exhaustion(t *testing.T) {

 	var allocated []string
 	for i := range 11 {
-		ip, err := ipam.AllocateIP(subnetId, cidr)
+		ip, err := ipam.AllocateIP(subnetId, cidr, PurposeENIPrimary, "eni-test")
 		require.NoError(t, err, "allocation %d should succeed", i)
 		allocated = append(allocated, ip)
 	}
@@ -101,7 +101,7 @@ func TestIPAM_Exhaustion(t *testing.T) {
 	assert.Equal(t, "10.0.5.14", allocated[len(allocated)-1])

 	// Next allocation should fail — subnet exhausted
-	_, err := ipam.AllocateIP(subnetId, cidr)
+	_, err := ipam.AllocateIP(subnetId, cidr, PurposeENIPrimary, "eni-test")
 	assert.ErrorContains(t, err, "exhausted")
 }

@@ -113,7 +113,7 @@ func TestIPAM_ReservedIPs(t *testing.T) {

 	// First 4 allocations should be .4, .5, .6, .7 (skipping .0-.3)
 	for i := 4; i <= 7; i++ {
-		ip, err := ipam.AllocateIP(subnetId, cidr)
+		ip, err := ipam.AllocateIP(subnetId, cidr, PurposeENIPrimary, "eni-test")
 		require.NoError(t, err)
 		expected := "10.0.6." + itoa(i)
 		assert.Equal(t, expected, ip)
@@ -129,29 +129,31 @@ func TestIPAM_AllocatedIPs(t *testing.T) {
 	assert.Nil(t, ips)

 	// Allocate some IPs
-	_, _ = ipam.AllocateIP("subnet-list", "10.0.7.0/24")
-	_, _ = ipam.AllocateIP("subnet-list", "10.0.7.0/24")
+	_, _ = ipam.AllocateIP("subnet-list", "10.0.7.0/24", PurposeENIPrimary, "eni-list")
+	_, _ = ipam.AllocateIP("subnet-list", "10.0.7.0/24", PurposeENIPrimary, "eni-list")

 	ips, err = ipam.AllocatedIPs("subnet-list")
 	require.NoError(t, err)
 	assert.Len(t, ips, 2)
-	assert.Equal(t, "10.0.7.4", ips[0])
-	assert.Equal(t, "10.0.7.5", ips[1])
+	assert.Equal(t, "10.0.7.4", ips[0].IP)
+	assert.Equal(t, PurposeENIPrimary, ips[0].Purpose)
+	assert.Equal(t, "eni-list", ips[0].OwnerID)
+	assert.Equal(t, "10.0.7.5", ips[1].IP)
 }

 func TestIPAM_MultipleSubnets(t *testing.T) {
 	ipam := setupTestIPAM(t)

-	ip1, err := ipam.AllocateIP("subnet-a", "10.0.10.0/24")
+	ip1, err := ipam.AllocateIP("subnet-a", "10.0.10.0/24", PurposeENIPrimary, "eni-a")
 	require.NoError(t, err)
 	assert.Equal(t, "10.0.10.4", ip1)

-	ip2, err := ipam.AllocateIP("subnet-b", "10.0.20.0/24")
+	ip2, err := ipam.AllocateIP("subnet-b", "10.0.20.0/24", PurposeENIPrimary, "eni-b")
 	require.NoError(t, err)
 	assert.Equal(t, "10.0.20.4", ip2)

 	// Each subnet tracks independently
-	ip3, err := ipam.AllocateIP("subnet-a", "10.0.10.0/24")
+	ip3, err := ipam.AllocateIP("subnet-a", "10.0.10.0/24", PurposeENIPrimary, "eni-a")
 	require.NoError(t, err)
 	assert.Equal(t, "10.0.10.5", ip3)
 }
@@ -160,7 +162,7 @@ func TestIPAM_LargerSubnet(t *testing.T) {
 	ipam := setupTestIPAM(t)

 	// /20 subnet — first allocable is still .4
-	ip, err := ipam.AllocateIP("subnet-big", "172.16.0.0/20")
+	ip, err := ipam.AllocateIP("subnet-big", "172.16.0.0/20", PurposeENIPrimary, "eni-big")
 	require.NoError(t, err)
 	assert.Equal(t, "172.16.0.4", ip)
 }
@@ -177,14 +179,14 @@ func TestIPAM_NewIPAMWithKV(t *testing.T) {
 	require.NotNil(t, ipam)

 	// Should work the same as regular IPAM
-	ip, err := ipam.AllocateIP("subnet-kv", "10.0.1.0/24")
+	ip, err := ipam.AllocateIP("subnet-kv", "10.0.1.0/24", PurposeENIPrimary, "eni-kv")
 	require.NoError(t, err)
 	assert.Equal(t, "10.0.1.4", ip)
 }

 func TestIPAM_InvalidCIDR(t *testing.T) {
 	ipam := setupTestIPAM(t)
-	_, err := ipam.AllocateIP("subnet-bad", "not-a-cidr")
+	_, err := ipam.AllocateIP("subnet-bad", "not-a-cidr", PurposeENIPrimary, "eni-bad")
 	assert.Error(t, err)
 	assert.Contains(t, err.Error(), "parse CIDR")
 }
@@ -0,0 +1,31 @@
+package handlers_ec2_vpc
+
+// Purpose tags every IP allocation with its semantic role so multi-VPC
+// clusters can reclaim / audit by (owner, purpose). The constants match the
+// JSON values written to KV records (mulga-siv-125.3.3 Slice B).
+const (
+	PurposeENIPrimary    = "eni-primary"
+	PurposeENISecondary  = "eni-secondary"
+	PurposeENIPublic     = "eni-public" // auto-assigned public IP
+	PurposeEIP           = "eip"
+	PurposeNATGWExternal = "natgw-external"
+	PurposeIGWLRP        = "igw-lrp"
+	// Future-reserved values; defined now so call sites and migrations can
+	// reference them without a follow-up rename.
+	PurposeEdgeLRExternal = "edge-lr-external" // 125.3.6
+	PurposeNodeMgmt       = "node-mgmt"        // currently .env
+	PurposeSubnetRouter   = "subnet-router"    // .1
+	PurposeSubnetDNS      = "subnet-dns"       // .2
+	PurposeSubnetReserved = "subnet-reserved"  // .0, .3, .last
+	// PurposeUnknown tags records the migration could not classify (e.g.
+	// orphan IPs with no owning ENIRecord). Surfaces in logs for triage.
+	PurposeUnknown = "unknown"
+)
+
+// LegacyExternalTypeToPurpose maps the old ExternalIPAllocation.Type values
+// onto the new Purpose enum. Used by the v1→v2 external-IPAM migration.
+var LegacyExternalTypeToPurpose = map[string]string{
+	"gateway":     PurposeIGWLRP,
+	"auto_assign": PurposeENIPublic,
+	"elastic_ip":  PurposeEIP,
+}
@@ -0,0 +1,134 @@
+package migrate
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"github.com/mulgadc/spinifex/spinifex/utils"
+	"github.com/nats-io/nats.go"
+)
+
+// extIPAMBucket mirrors handlers/ec2/vpc.KVBucketExternalIPAM. Duplicated to
+// keep the migrate package free of any handler dependency; if the bucket name
+// changes, both constants must move together (mulga-siv-125.3.3 Slice B).
+const extIPAMBucket = "spinifex-external-ipam"
+
+const extIPAMMaxRetries = 3
+
+// extLegacyTypeToPurpose maps the v1 ExternalIPAllocation.Type values onto
+// the v2 Purpose enum. Frozen copy of vpc.LegacyExternalTypeToPurpose to
+// keep this migration self-contained.
+var extLegacyTypeToPurpose = map[string]string{
+	"gateway":     "igw-lrp",
+	"auto_assign": "eni-public",
+	"elastic_ip":  "eip",
+}
+
+// extIPAllocV1V2 is a union of the v1 and v2 ExternalIPAllocation schemas.
+// The migration reads the legacy Type field, writes the new Purpose field,
+// and clears Type so the on-disk record matches the v2 struct exactly.
+type extIPAllocV1V2 struct {
+	Type         string `json:"type,omitempty"`
+	Purpose      string `json:"purpose,omitempty"`
+	AllocationID string `json:"allocation_id,omitempty"`
+	Association  string `json:"association,omitempty"`
+	ENIId        string `json:"eni_id,omitempty"`
+	InstanceId   string `json:"instance_id,omitempty"`
+	Note         string `json:"note,omitempty"`
+}
+
+type extIPAMRecordV1V2 struct {
+	PoolName        string                    `json:"pool_name"`
+	RangeStart      string                    `json:"range_start"`
+	RangeEnd        string                    `json:"range_end"`
+	Gateway         string                    `json:"gateway"`
+	GatewayIP       string                    `json:"gateway_ip"`
+	PrefixLen       int                       `json:"prefix_len"`
+	Region          string                    `json:"region,omitempty"`
+	AZ              string                    `json:"az,omitempty"`
+	GwLrpRangeStart string                    `json:"gw_lrp_range_start,omitempty"`
+	GwLrpRangeEnd   string                    `json:"gw_lrp_range_end,omitempty"`
+	Allocated       map[string]extIPAllocV1V2 `json:"allocated"`
+}
+
+func init() {
+	DefaultRegistry.RegisterKV(extIPAMBucket, KVMigration{
+		FromVersion: 1,
+		ToVersion:   2,
+		Description: "rename ExternalIPAllocation.Type → .Purpose with enum mapping",
+		Run: func(ctx KVContext) error {
+			keys, err := ctx.KV.Keys()
+			if err != nil {
+				if errors.Is(err, nats.ErrNoKeysFound) {
+					return nil
+				}
+				return fmt.Errorf("list keys: %w", err)
+			}
+
+			for _, key := range keys {
+				if key == utils.VersionKey {
+					continue
+				}
+				if err := renameExternalIPAMType(ctx, key); err != nil {
+					return err
+				}
+			}
+			return nil
+		},
+	})
+}
+
+func renameExternalIPAMType(ctx KVContext, key string) error {
+	var lastErr error
+	for attempt := range extIPAMMaxRetries {
+		entry, err := ctx.KV.Get(key)
+		if err != nil {
+			if errors.Is(err, nats.ErrKeyNotFound) {
+				return nil
+			}
+			return fmt.Errorf("get %s: %w", key, err)
+		}
+
+		var record extIPAMRecordV1V2
+		if err := json.Unmarshal(entry.Value(), &record); err != nil {
+			return fmt.Errorf("unmarshal %s: %w", key, err)
+		}
+
+		changed := false
+		for ip, alloc := range record.Allocated {
+			if alloc.Purpose != "" {
+				continue
+			}
+			purpose := extLegacyTypeToPurpose[alloc.Type]
+			if purpose == "" {
+				ctx.Logger.Warn("external IPAM v1→v2 migration: unknown legacy type, tagging unknown",
+					"pool", key, "ip", ip, "legacy_type", alloc.Type)
+				purpose = "unknown"
+			}
+			alloc.Purpose = purpose
+			alloc.Type = ""
+			record.Allocated[ip] = alloc
+			changed = true
+		}
+
+		if !changed {
+			return nil
+		}
+
+		data, err := json.Marshal(record)
+		if err != nil {
+			return fmt.Errorf("marshal %s: %w", key, err)
+		}
+		if _, err := ctx.KV.Update(key, data, entry.Revision()); err != nil {
+			if errors.Is(err, nats.ErrKeyExists) {
+				ctx.Logger.Warn("external IPAM migration CAS conflict, retrying", "key", key, "attempt", attempt+1)
+				lastErr = err
+				continue
+			}
+			return fmt.Errorf("update %s: %w", key, err)
+		}
+		return nil
+	}
+	return fmt.Errorf("rename Type→Purpose for %s: exceeded %d CAS retries: %w", key, extIPAMMaxRetries, lastErr)
+}
@@ -0,0 +1,103 @@
+package migrate
+
+import (
+	"encoding/json"
+	"log/slog"
+	"testing"
+
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func extIPAMMigrationKV(t *testing.T) nats.KeyValue {
+	t.Helper()
+	_, nc := startTestNATS(t)
+	return createTestBucket(t, nc, "ext-ipam-migration-test")
+}
+
+func runExtIPAMMigration(t *testing.T, kv nats.KeyValue) {
+	t.Helper()
+	for _, m := range DefaultRegistry.kvMigrations[extIPAMBucket] {
+		if m.FromVersion == 1 && m.ToVersion == 2 {
+			require.NoError(t, m.Run(KVContext{KV: kv, Logger: slog.Default()}))
+			return
+		}
+	}
+	t.Fatalf("v1→v2 migration not registered for %s", extIPAMBucket)
+}
+
+func TestExtIPAMMigration_RenamesType(t *testing.T) {
+	kv := extIPAMMigrationKV(t)
+
+	record := extIPAMRecordV1V2{
+		PoolName:   "wan",
+		RangeStart: "192.168.1.10",
+		RangeEnd:   "192.168.1.250",
+		Gateway:    "192.168.1.1",
+		PrefixLen:  24,
+		Allocated: map[string]extIPAllocV1V2{
+			"192.168.1.10": {Type: "gateway", Note: "OVN router SNAT address"},
+			"192.168.1.11": {Type: "auto_assign", ENIId: "eni-1", InstanceId: "i-1"},
+			"192.168.1.12": {Type: "elastic_ip", AllocationID: "eipalloc-1"},
+			"192.168.1.13": {Type: "junk-legacy", ENIId: "eni-zombie"},
+		},
+	}
+	data, err := json.Marshal(record)
+	require.NoError(t, err)
+	_, err = kv.Put("wan", data)
+	require.NoError(t, err)
+
+	runExtIPAMMigration(t, kv)
+
+	entry, err := kv.Get("wan")
+	require.NoError(t, err)
+	var got extIPAMRecordV1V2
+	require.NoError(t, json.Unmarshal(entry.Value(), &got))
+
+	assert.Equal(t, "igw-lrp", got.Allocated["192.168.1.10"].Purpose)
+	assert.Equal(t, "eni-public", got.Allocated["192.168.1.11"].Purpose)
+	assert.Equal(t, "eip", got.Allocated["192.168.1.12"].Purpose)
+	assert.Equal(t, "unknown", got.Allocated["192.168.1.13"].Purpose, "unrecognized legacy types tagged unknown")
+
+	for ip, alloc := range got.Allocated {
+		assert.Empty(t, alloc.Type, "legacy Type field cleared for %s", ip)
+	}
+
+	assert.Equal(t, "eni-1", got.Allocated["192.168.1.11"].ENIId, "non-Type fields preserved")
+	assert.Equal(t, "eipalloc-1", got.Allocated["192.168.1.12"].AllocationID)
+}
+
+func TestExtIPAMMigration_IdempotentOnAlreadyMigrated(t *testing.T) {
+	kv := extIPAMMigrationKV(t)
+
+	record := extIPAMRecordV1V2{
+		PoolName: "wan",
+		Allocated: map[string]extIPAllocV1V2{
+			"192.168.1.10": {Purpose: "igw-lrp"},
+			"192.168.1.11": {Purpose: "eni-public", ENIId: "eni-1"},
+		},
+	}
+	data, err := json.Marshal(record)
+	require.NoError(t, err)
+	_, err = kv.Put("wan", data)
+	require.NoError(t, err)
+	rev := getKVRevision(t, kv, "wan")
+
+	runExtIPAMMigration(t, kv)
+
+	newRev := getKVRevision(t, kv, "wan")
+	assert.Equal(t, rev, newRev, "no write when all records already have Purpose")
+}
+
+func TestExtIPAMMigration_EmptyBucket(t *testing.T) {
+	kv := extIPAMMigrationKV(t)
+	runExtIPAMMigration(t, kv)
+}
+
+func getKVRevision(t *testing.T, kv nats.KeyValue, key string) uint64 {
+	t.Helper()
+	entry, err := kv.Get(key)
+	require.NoError(t, err)
+	return entry.Revision()
+}
@@ -0,0 +1,225 @@
+package migrate
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"github.com/mulgadc/spinifex/spinifex/utils"
+	"github.com/nats-io/nats.go"
+)
+
+// ipamBucket / eniBucket mirror handlers/ec2/vpc.KVBucketIPAM and
+// .KVBucketENIs. Duplicated here so the migrate package stays handler-free
+// (mulga-siv-125.3.3 Slice B).
+const (
+	ipamBucket = "spinifex-vpc-ipam"
+	eniBucket  = "spinifex-vpc-enis"
+)
+
+const ipamMigrationMaxRetries = 3
+
+// ipEntryV2 mirrors handlers/ec2/vpc.IPEntry (frozen snapshot).
+type ipEntryV2 struct {
+	IP      string `json:"ip"`
+	Purpose string `json:"purpose"`
+	OwnerID string `json:"owner_id,omitempty"`
+}
+
+// ipamRecordV1V2 is a union of v1 ([]string) and v2 ([]IPEntry) schemas.
+// Allocated is left as raw JSON so we can detect which schema is on-disk
+// and re-marshal as IPEntry.
+type ipamRecordV1V2 struct {
+	SubnetId  string          `json:"subnet_id"`
+	CidrBlock string          `json:"cidr_block"`
+	Allocated json.RawMessage `json:"allocated"`
+}
+
+// ipamRecordV2 is the post-migration shape.
+type ipamRecordV2 struct {
+	SubnetId  string      `json:"subnet_id"`
+	CidrBlock string      `json:"cidr_block"`
+	Allocated []ipEntryV2 `json:"allocated"`
+}
+
+// eniRecordSnapshot is the minimum ENIRecord shape needed to attribute IPAM
+// entries to their owning ENI. Frozen copy keeps the migration self-contained.
+type eniRecordSnapshot struct {
+	NetworkInterfaceId string `json:"network_interface_id"`
+	SubnetId           string `json:"subnet_id"`
+	PrivateIpAddress   string `json:"private_ip_address"`
+}
+
+func init() {
+	DefaultRegistry.RegisterKV(ipamBucket, KVMigration{
+		FromVersion: 1,
+		ToVersion:   2,
+		Description: "convert subnet IPAM Allocated []string → []IPEntry; backfill Purpose+OwnerID from ENIs",
+		Run: func(ctx KVContext) error {
+			ownerIndex, err := buildIPAMOwnerIndex(ctx.JetStream)
+			if err != nil {
+				return fmt.Errorf("build owner index: %w", err)
+			}
+
+			keys, err := ctx.KV.Keys()
+			if err != nil {
+				if errors.Is(err, nats.ErrNoKeysFound) {
+					return nil
+				}
+				return fmt.Errorf("list keys: %w", err)
+			}
+
+			for _, key := range keys {
+				if key == utils.VersionKey {
+					continue
+				}
+				if err := convertIPAMRecord(ctx, key, ownerIndex); err != nil {
+					return err
+				}
+			}
+			return nil
+		},
+	})
+}
+
+// buildIPAMOwnerIndex returns map[subnetID]map[ip]eniID by walking the ENI
+// bucket. Returns an empty index if the ENI bucket is missing
+// (fresh-install case) so the migration still runs cleanly.
+func buildIPAMOwnerIndex(js nats.JetStreamContext) (map[string]map[string]string, error) {
+	if js == nil {
+		// Called via the bare RunKV path (e.g. in tests with no JS). Skip
+		// owner attribution — all entries become Purpose=unknown so the
+		// schema-only conversion still succeeds.
+		return map[string]map[string]string{}, nil
+	}
+	eniKV, err := js.KeyValue(eniBucket)
+	if err != nil {
+		if errors.Is(err, nats.ErrBucketNotFound) {
+			return map[string]map[string]string{}, nil
+		}
+		return nil, fmt.Errorf("open ENI bucket: %w", err)
+	}
+
+	keys, err := eniKV.Keys()
+	if err != nil {
+		if errors.Is(err, nats.ErrNoKeysFound) {
+			return map[string]map[string]string{}, nil
+		}
+		return nil, fmt.Errorf("list ENI keys: %w", err)
+	}
+
+	index := make(map[string]map[string]string)
+	for _, key := range keys {
+		if key == utils.VersionKey {
+			continue
+		}
+		entry, err := eniKV.Get(key)
+		if err != nil {
+			if errors.Is(err, nats.ErrKeyNotFound) {
+				continue
+			}
+			return nil, fmt.Errorf("get ENI %s: %w", key, err)
+		}
+		var eni eniRecordSnapshot
+		if err := json.Unmarshal(entry.Value(), &eni); err != nil {
+			continue
+		}
+		if eni.SubnetId == "" || eni.PrivateIpAddress == "" {
+			continue
+		}
+		if _, ok := index[eni.SubnetId]; !ok {
+			index[eni.SubnetId] = make(map[string]string)
+		}
+		index[eni.SubnetId][eni.PrivateIpAddress] = eni.NetworkInterfaceId
+	}
+	return index, nil
+}
+
+func convertIPAMRecord(ctx KVContext, key string, ownerIndex map[string]map[string]string) error {
+	var lastErr error
+	for attempt := range ipamMigrationMaxRetries {
+		entry, err := ctx.KV.Get(key)
+		if err != nil {
+			if errors.Is(err, nats.ErrKeyNotFound) {
+				return nil
+			}
+			return fmt.Errorf("get %s: %w", key, err)
+		}
+
+		var raw ipamRecordV1V2
+		if err := json.Unmarshal(entry.Value(), &raw); err != nil {
+			return fmt.Errorf("unmarshal %s: %w", key, err)
+		}
+
+		// Detect schema: v2 records start with `[{`, v1 with `["`.
+		// Idempotent skip when already migrated.
+		if v2Entries, ok := tryDecodeV2(raw.Allocated); ok {
+			_ = v2Entries
+			return nil
+		}
+
+		var legacy []string
+		if len(raw.Allocated) > 0 {
+			if err := json.Unmarshal(raw.Allocated, &legacy); err != nil {
+				return fmt.Errorf("unmarshal legacy []string for %s: %w", key, err)
+			}
+		}
+
+		ownerByIP := ownerIndex[raw.SubnetId]
+		converted := make([]ipEntryV2, 0, len(legacy))
+		for _, ip := range legacy {
+			ownerID, found := ownerByIP[ip]
+			purpose := "eni-primary"
+			if !found {
+				ownerID = ""
+				purpose = "unknown"
+				ctx.Logger.Warn("internal IPAM v1→v2 migration: no owning ENI, tagging unknown",
+					"subnet", raw.SubnetId, "ip", ip)
+			}
+			converted = append(converted, ipEntryV2{IP: ip, Purpose: purpose, OwnerID: ownerID})
+		}
+
+		newRecord := ipamRecordV2{
+			SubnetId:  raw.SubnetId,
+			CidrBlock: raw.CidrBlock,
+			Allocated: converted,
+		}
+		data, err := json.Marshal(newRecord)
+		if err != nil {
+			return fmt.Errorf("marshal %s: %w", key, err)
+		}
+		if _, err := ctx.KV.Update(key, data, entry.Revision()); err != nil {
+			if errors.Is(err, nats.ErrKeyExists) {
+				ctx.Logger.Warn("internal IPAM migration CAS conflict, retrying", "key", key, "attempt", attempt+1)
+				lastErr = err
+				continue
+			}
+			return fmt.Errorf("update %s: %w", key, err)
+		}
+		return nil
+	}
+	return fmt.Errorf("convert IPAM record %s: exceeded %d CAS retries: %w", key, ipamMigrationMaxRetries, lastErr)
+}
+
+// tryDecodeV2 returns true when raw decodes as []ipEntryV2 with at least one
+// entry whose IP field is set — the cheapest reliable v2 detector. Empty
+// arrays decode as either schema; we treat them as already migrated.
+func tryDecodeV2(raw json.RawMessage) ([]ipEntryV2, bool) {
+	if len(raw) == 0 {
+		return nil, true
+	}
+	var entries []ipEntryV2
+	if err := json.Unmarshal(raw, &entries); err != nil {
+		return nil, false
+	}
+	for _, e := range entries {
+		if e.IP != "" {
+			return entries, true
+		}
+	}
+	// Zero-length array — already v2-shaped.
+	if len(entries) == 0 {
+		return entries, true
+	}
+	return nil, false
+}
@@ -0,0 +1,158 @@
+package migrate
+
+import (
+	"encoding/json"
+	"log/slog"
+	"testing"
+
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func setupIPAMMigrationKV(t *testing.T) (nats.KeyValue, nats.JetStreamContext) {
+	t.Helper()
+	_, nc := startTestNATS(t)
+	js, err := nc.JetStream()
+	require.NoError(t, err)
+	kv := createTestBucket(t, nc, ipamBucket)
+	return kv, js
+}
+
+func runIPAMMigration(t *testing.T, kv nats.KeyValue, js nats.JetStreamContext) {
+	t.Helper()
+	for _, m := range DefaultRegistry.kvMigrations[ipamBucket] {
+		if m.FromVersion == 1 && m.ToVersion == 2 {
+			require.NoError(t, m.Run(KVContext{KV: kv, JetStream: js, Logger: slog.Default()}))
+			return
+		}
+	}
+	t.Fatalf("v1→v2 migration not registered for %s", ipamBucket)
+}
+
+func seedENIBucket(t *testing.T, js nats.JetStreamContext, enis []eniRecordSnapshot) {
+	t.Helper()
+	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: eniBucket, History: 1})
+	require.NoError(t, err)
+	for _, eni := range enis {
+		data, err := json.Marshal(eni)
+		require.NoError(t, err)
+		_, err = kv.Put(eni.NetworkInterfaceId, data)
+		require.NoError(t, err)
+	}
+}
+
+func TestIPAMMigration_ConvertsV1ToV2WithOwnerBackfill(t *testing.T) {
+	kv, js := setupIPAMMigrationKV(t)
+	seedENIBucket(t, js, []eniRecordSnapshot{
+		{NetworkInterfaceId: "eni-aaa", SubnetId: "subnet-1", PrivateIpAddress: "10.0.1.4"},
+		{NetworkInterfaceId: "eni-bbb", SubnetId: "subnet-1", PrivateIpAddress: "10.0.1.5"},
+		{NetworkInterfaceId: "eni-ccc", SubnetId: "subnet-2", PrivateIpAddress: "10.0.2.4"},
+	})
+
+	type legacyRecord struct {
+		SubnetId  string   `json:"subnet_id"`
+		CidrBlock string   `json:"cidr_block"`
+		Allocated []string `json:"allocated"`
+	}
+	subnet1 := legacyRecord{
+		SubnetId:  "subnet-1",
+		CidrBlock: "10.0.1.0/24",
+		Allocated: []string{"10.0.1.4", "10.0.1.5", "10.0.1.99"}, // .99 has no owning ENI
+	}
+	data, err := json.Marshal(subnet1)
+	require.NoError(t, err)
+	_, err = kv.Put("subnet-1", data)
+	require.NoError(t, err)
+
+	subnet2 := legacyRecord{
+		SubnetId:  "subnet-2",
+		CidrBlock: "10.0.2.0/24",
+		Allocated: []string{"10.0.2.4"},
+	}
+	data, err = json.Marshal(subnet2)
+	require.NoError(t, err)
+	_, err = kv.Put("subnet-2", data)
+	require.NoError(t, err)
+
+	runIPAMMigration(t, kv, js)
+
+	entry, err := kv.Get("subnet-1")
+	require.NoError(t, err)
+	var got ipamRecordV2
+	require.NoError(t, json.Unmarshal(entry.Value(), &got))
+
+	require.Len(t, got.Allocated, 3)
+	assert.Equal(t, "10.0.1.4", got.Allocated[0].IP)
+	assert.Equal(t, "eni-primary", got.Allocated[0].Purpose)
+	assert.Equal(t, "eni-aaa", got.Allocated[0].OwnerID)
+	assert.Equal(t, "10.0.1.5", got.Allocated[1].IP)
+	assert.Equal(t, "eni-bbb", got.Allocated[1].OwnerID)
+	assert.Equal(t, "10.0.1.99", got.Allocated[2].IP)
+	assert.Equal(t, "unknown", got.Allocated[2].Purpose, "orphan IP tagged unknown")
+	assert.Empty(t, got.Allocated[2].OwnerID)
+
+	entry, err = kv.Get("subnet-2")
+	require.NoError(t, err)
+	require.NoError(t, json.Unmarshal(entry.Value(), &got))
+	require.Len(t, got.Allocated, 1)
+	assert.Equal(t, "eni-ccc", got.Allocated[0].OwnerID)
+}
+
+func TestIPAMMigration_IdempotentOnAlreadyV2(t *testing.T) {
+	kv, js := setupIPAMMigrationKV(t)
+
+	v2 := ipamRecordV2{
+		SubnetId:  "subnet-1",
+		CidrBlock: "10.0.1.0/24",
+		Allocated: []ipEntryV2{
+			{IP: "10.0.1.4", Purpose: "eni-primary", OwnerID: "eni-aaa"},
+		},
+	}
+	data, err := json.Marshal(v2)
+	require.NoError(t, err)
+	_, err = kv.Put("subnet-1", data)
+	require.NoError(t, err)
+	rev := getKVRevision(t, kv, "subnet-1")
+
+	runIPAMMigration(t, kv, js)
+
+	newRev := getKVRevision(t, kv, "subnet-1")
+	assert.Equal(t, rev, newRev, "no write when record is already v2")
+}
+
+func TestIPAMMigration_NoJetStream_TagsUnknown(t *testing.T) {
+	_, nc := startTestNATS(t)
+	kv := createTestBucket(t, nc, ipamBucket)
+
+	type legacyRecord struct {
+		SubnetId  string   `json:"subnet_id"`
+		CidrBlock string   `json:"cidr_block"`
+		Allocated []string `json:"allocated"`
+	}
+	rec := legacyRecord{
+		SubnetId:  "subnet-1",
+		CidrBlock: "10.0.1.0/24",
+		Allocated: []string{"10.0.1.4"},
+	}
+	data, err := json.Marshal(rec)
+	require.NoError(t, err)
+	_, err = kv.Put("subnet-1", data)
+	require.NoError(t, err)
+
+	// Run migration with nil JetStream — schema conversion still runs but
+	// every entry becomes Purpose=unknown.
+	runIPAMMigration(t, kv, nil)
+
+	entry, err := kv.Get("subnet-1")
+	require.NoError(t, err)
+	var got ipamRecordV2
+	require.NoError(t, json.Unmarshal(entry.Value(), &got))
+	require.Len(t, got.Allocated, 1)
+	assert.Equal(t, "unknown", got.Allocated[0].Purpose)
+}
+
+func TestIPAMMigration_EmptyBucket(t *testing.T) {
+	kv, js := setupIPAMMigrationKV(t)
+	runIPAMMigration(t, kv, js)
+}
@@ -21,9 +21,12 @@ type KVMigration struct {
 }

 // 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     nats.KeyValue
-	Logger *slog.Logger
+	KV        nats.KeyValue
+	JetStream nats.JetStreamContext
+	Logger    *slog.Logger
 }

 // ConfigMigration represents a versioned transformation of on-disk config files.
@@ -97,6 +100,13 @@ func (r *Registry) RegisterConfig(target string, m ConfigMigration) {
 	})
 }

+// 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(bucket string, kv nats.KeyValue, js nats.JetStreamContext, targetVersion int) error {
+	return r.runKV(bucket, kv, js, targetVersion)
+}
+
 // RunKV executes all pending migrations for a KV bucket up to targetVersion.
 // It reads the current version via utils.ReadVersion, runs each migration in
 // order, and stamps the version via utils.WriteVersion after each step.
@@ -109,6 +119,10 @@ func (r *Registry) RegisterConfig(target string, m ConfigMigration) {
 // an error — migration authors must register migrations before bumping version
 // constants.
 func (r *Registry) RunKV(bucket string, kv nats.KeyValue, targetVersion int) error {
+	return r.runKV(bucket, kv, nil, targetVersion)
+}
+
+func (r *Registry) runKV(bucket string, kv nats.KeyValue, js nats.JetStreamContext, targetVersion int) error {
 	current, err := utils.ReadVersion(kv)
 	if err != nil {
 		return fmt.Errorf("read version for %s: %w", bucket, err)
@@ -168,7 +182,7 @@ func (r *Registry) RunKV(bucket string, kv nats.KeyValue, targetVersion int) err
 	logger := slog.Default()
 	for _, m := range pending {
 		logger.Info("Running KV migration", "bucket", bucket, "from", m.FromVersion, "to", m.ToVersion, "description", m.Description)
-		ctx := KVContext{KV: kv, Logger: logger}
+		ctx := KVContext{KV: kv, JetStream: js, Logger: logger}
 		if err := m.Run(ctx); err != nil {
 			return fmt.Errorf("KV migration %s %d→%d failed: %w", bucket, m.FromVersion, m.ToVersion, err)
 		}