feat: Phase 12b + Phase 16 — KV reconcile on startup + NAT mode support

Phase 12b: vpcd now reconciles OVN topology from NATS KV on startup (Pass 2). After subscribing to topics, it reads all VPCs, subnets, IGW attachments, and ENIs from KV and ensures matching OVN objects exist. Handles reboots, OVN DB loss, and missed events. All operations are idempotent. Phase 16 prep: Add --gateway-ip flag to admin init for NAT mode (specifies the OVN gateway router's external IP without DHCP). Update reset-dev-env.sh to support EXTERNAL_MODE=nat env var for NAT/DHCP mode testing alongside the default pool mode.
2026-10-15 09:01:31 +00:00 · 2026-03-24 15:03:04 +10:00
parent 88baf5e202
commit 216ca9e710
5 changed files with 431 additions and 13 deletions
@@ -221,6 +221,7 @@ func init() {
 	adminInitCmd.Flags().String("external-iface", "", "WAN NIC for br-external (auto-detected from default route)")
 	adminInitCmd.Flags().String("external-pool", "", "External IP pool range as start-end (e.g., 192.168.1.150-192.168.1.250)")
 	adminInitCmd.Flags().String("external-gateway", "", "WAN gateway IP (auto-detected from default route)")
+	adminInitCmd.Flags().String("gateway-ip", "", "OVN gateway router's external IP for SNAT (default: pool range_start for pool mode, required for nat mode without DHCP)")
 	adminInitCmd.Flags().Int("external-prefix-len", 24, "External pool subnet prefix length (auto-detected)")
 	adminInitCmd.Flags().Bool("no-external", false, "Disable external networking (overlay-only, no internet for VMs)")

@@ -543,6 +544,7 @@ func runAdminInit(cmd *cobra.Command, args []string) {
 	externalPool, _ := cmd.Flags().GetString("external-pool")
 	externalGateway, _ := cmd.Flags().GetString("external-gateway")
 	externalPrefixLen, _ := cmd.Flags().GetInt("external-prefix-len")
+	gatewayIP, _ := cmd.Flags().GetString("gateway-ip")
 	noExternal, _ := cmd.Flags().GetBool("no-external")

 	// Auto-detect network topology
@@ -625,6 +627,10 @@ func runAdminInit(cmd *cobra.Command, args []string) {
 		fmt.Fprintf(os.Stderr, "❌ Error: --external-gateway is not a valid IP: %s\n", externalGateway)
 		os.Exit(1)
 	}
+	if gatewayIP != "" && net.ParseIP(gatewayIP) == nil {
+		fmt.Fprintf(os.Stderr, "❌ Error: --gateway-ip is not a valid IP: %s\n", gatewayIP)
+		os.Exit(1)
+	}

 	// Detect DNS servers from the host for VM DHCP
 	var dnsServers []string
@@ -635,9 +641,9 @@ func runAdminInit(cmd *cobra.Command, args []string) {
 		}
 	}

-	// For nat mode, we need the gateway IP. The DHCP flag tells setup-ovn.sh
-	// to obtain one via DHCP on the macvlan/bridge interface.
-	useExternalDHCP := externalMode == "nat" && poolStart == ""
+	// For nat mode, we need the gateway IP. Either provided via --gateway-ip
+	// or obtained via DHCP on the macvlan/bridge interface.
+	useExternalDHCP := externalMode == "nat" && gatewayIP == "" && poolStart == ""

 	// Validate IP address format
 	if net.ParseIP(bindIP) == nil {
@@ -844,6 +850,7 @@ func runAdminInit(cmd *cobra.Command, args []string) {
 		PoolStart:     poolStart,
 		PoolEnd:       poolEnd,
 		PoolGateway:    externalGateway,
+		PoolGatewayIP:  gatewayIP,
 		PoolPrefixLen:  externalPrefixLen,
 		PoolDNSServers: dnsServers,

@@ -858,10 +865,13 @@ func runAdminInit(cmd *cobra.Command, args []string) {
 	// Print external networking summary
 	if externalMode != "" {
 		fmt.Printf("\n📡 External networking: %s\n", externalMode)
-		fmt.Printf("  WAN interface: %s (macvlan)\n", externalIface)
+		fmt.Printf("  WAN interface: %s\n", externalIface)
 		if externalMode == "pool" {
 			fmt.Printf("  IP pool:       %s - %s\n", poolStart, poolEnd)
 			fmt.Printf("  ⚠️  Ensure %s-%s is excluded from your router's DHCP range.\n", poolStart, poolEnd)
+		} else if gatewayIP != "" {
+			fmt.Printf("  Gateway IP:    %s (static)\n", gatewayIP)
+			fmt.Println("  VMs get:       outbound internet via SNAT (no per-VM public IPs)")
 		} else if useExternalDHCP {
 			fmt.Println("  Gateway IP:    DHCP (obtained during OVN setup)")
 			fmt.Println("  VMs get:       outbound internet via SNAT")
@@ -159,19 +159,31 @@ fi

 echo "Initializing platform"
 ADMIN_INIT_ARGS="--region $REGION --az ${REGION}a --node node1 --nodes 1"
+
+# External networking mode: set EXTERNAL_MODE=nat for NAT/DHCP mode (outbound-only),
+# or leave unset / EXTERNAL_MODE=pool for pool mode (per-VM public IPs).
+EXTERNAL_MODE="${EXTERNAL_MODE:-pool}"
+
 if [ -n "$WAN_IFACE" ] && [ -n "$WAN_GW" ]; then
-    # Suggest pool range at high end of WAN subnet to avoid DHCP conflicts
    WAN_IP=$(ip -4 -o addr show "$WAN_IFACE" 2>/dev/null | awk '{print $4}' | cut -d/ -f1 | head -1)
    WAN_PREFIX=$(ip -4 -o addr show "$WAN_IFACE" 2>/dev/null | awk '{print $4}' | cut -d/ -f2 | head -1)
    if [ -z "$WAN_PREFIX" ]; then WAN_PREFIX=24; fi

-    # Use high range of subnet: .200-.250 for /24, avoid common DHCP ranges
-    IFS='.' read -r o1 o2 o3 o4 <<< "$WAN_GW"
-    POOL_START="${o1}.${o2}.${o3}.200"
-    POOL_END="${o1}.${o2}.${o3}.250"
-
-    echo "  External pool: $POOL_START - $POOL_END (gateway: $WAN_GW)"
-    ADMIN_INIT_ARGS="$ADMIN_INIT_ARGS --external-mode=pool --external-pool=${POOL_START}-${POOL_END} --external-gateway=${WAN_GW} --external-prefix-len=${WAN_PREFIX}"
+    if [ "$EXTERNAL_MODE" = "nat" ]; then
+        # NAT mode: outbound-only via shared SNAT. Use a single IP from the
+        # high end of the WAN subnet as the gateway IP. No per-VM public IPs.
+        IFS='.' read -r o1 o2 o3 o4 <<< "$WAN_GW"
+        GATEWAY_IP="${GATEWAY_IP:-${o1}.${o2}.${o3}.200}"
+        echo "  External mode: nat (gateway IP: $GATEWAY_IP, WAN gateway: $WAN_GW)"
+        ADMIN_INIT_ARGS="$ADMIN_INIT_ARGS --external-mode=nat --gateway-ip=${GATEWAY_IP} --external-gateway=${WAN_GW} --external-prefix-len=${WAN_PREFIX}"
+    else
+        # Pool mode: per-VM public IPs from a static range
+        IFS='.' read -r o1 o2 o3 o4 <<< "$WAN_GW"
+        POOL_START="${o1}.${o2}.${o3}.200"
+        POOL_END="${o1}.${o2}.${o3}.250"
+        echo "  External pool: $POOL_START - $POOL_END (gateway: $WAN_GW)"
+        ADMIN_INIT_ARGS="$ADMIN_INIT_ARGS --external-mode=pool --external-pool=${POOL_START}-${POOL_END} --external-gateway=${WAN_GW} --external-prefix-len=${WAN_PREFIX}"
+    fi
 fi
 ./bin/spx admin init $ADMIN_INIT_ARGS

@@ -2,14 +2,23 @@ package vpcd

 import (
 	"context"
+	"encoding/json"
+	"errors"
 	"log/slog"
+
+	handlers_ec2_igw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/igw"
+	handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
+	"github.com/mulgadc/spinifex/spinifex/services/vpcd/nbdb"
+	"github.com/mulgadc/spinifex/spinifex/utils"
+	"github.com/nats-io/nats.go"
 )

 // ReconcileResult tracks what was created during reconciliation.
 type ReconcileResult struct {
-	RoutersCreated int
+	RoutersCreated  int
 	SwitchesCreated int
 	IGWsCreated     int
+	PortsCreated    int
 }

 // Reconcile ensures OVN topology matches the expected state from the bootstrap config.
@@ -81,3 +90,197 @@ func Reconcile(ctx context.Context, topo *TopologyHandler, bootstrap *BootstrapV

 	return result
 }
+
+// ReconcileFromKV reads all VPCs, subnets, IGW attachments, and ENIs from NATS KV
+// and ensures OVN topology matches. This is Pass 2 from the Phase 12 plan — it
+// handles reboots, OVN DB loss, and any missed events. All operations are idempotent.
+func ReconcileFromKV(ctx context.Context, nc *nats.Conn, topo *TopologyHandler) ReconcileResult {
+	var result ReconcileResult
+
+	js, err := nc.JetStream()
+	if err != nil {
+		slog.Warn("vpcd reconcile-kv: failed to get JetStream context", "err", err)
+		return result
+	}
+
+	// 1. Reconcile VPCs: ensure each VPC has a logical router
+	vpcKV, err := js.KeyValue(handlers_ec2_vpc.KVBucketVPCs)
+	if err != nil {
+		slog.Debug("vpcd reconcile-kv: VPC KV bucket not available (first boot before daemon?)", "err", err)
+		return result
+	}
+
+	// Build a map of VPC ID → CIDR for subnet/IGW reconciliation
+	type vpcInfo struct {
+		VpcId     string
+		CidrBlock string
+	}
+	vpcMap := make(map[string]vpcInfo)
+
+	vpcKeys, err := vpcKV.Keys()
+	if err != nil && !errors.Is(err, nats.ErrNoKeysFound) {
+		slog.Warn("vpcd reconcile-kv: failed to list VPC keys", "err", err)
+	}
+	for _, key := range vpcKeys {
+		if key == utils.VersionKey {
+			continue
+		}
+		entry, err := vpcKV.Get(key)
+		if err != nil {
+			continue
+		}
+		var rec handlers_ec2_vpc.VPCRecord
+		if err := json.Unmarshal(entry.Value(), &rec); err != nil {
+			slog.Warn("vpcd reconcile-kv: failed to unmarshal VPC record", "key", key, "err", err)
+			continue
+		}
+		vpcMap[rec.VpcId] = vpcInfo{VpcId: rec.VpcId, CidrBlock: rec.CidrBlock}
+
+		routerName := "vpc-" + rec.VpcId
+		if _, err := topo.ovn.GetLogicalRouter(ctx, routerName); err != nil {
+			slog.Info("vpcd reconcile-kv: creating VPC router", "router", routerName)
+			if err := topo.reconcileVPC(ctx, rec.VpcId, rec.CidrBlock); err != nil {
+				slog.Error("vpcd reconcile-kv: failed to create VPC router", "err", err)
+			} else {
+				result.RoutersCreated++
+			}
+		}
+	}
+
+	// 2. Reconcile subnets: ensure each subnet has a logical switch + router port + DHCP
+	subnetKV, err := js.KeyValue(handlers_ec2_vpc.KVBucketSubnets)
+	if err != nil {
+		slog.Debug("vpcd reconcile-kv: subnet KV bucket not available", "err", err)
+	} else {
+		subnetKeys, err := subnetKV.Keys()
+		if err != nil && !errors.Is(err, nats.ErrNoKeysFound) {
+			slog.Warn("vpcd reconcile-kv: failed to list subnet keys", "err", err)
+		}
+		for _, key := range subnetKeys {
+			if key == utils.VersionKey {
+				continue
+			}
+			entry, err := subnetKV.Get(key)
+			if err != nil {
+				continue
+			}
+			var rec handlers_ec2_vpc.SubnetRecord
+			if err := json.Unmarshal(entry.Value(), &rec); err != nil {
+				slog.Warn("vpcd reconcile-kv: failed to unmarshal subnet record", "key", key, "err", err)
+				continue
+			}
+
+			switchName := "subnet-" + rec.SubnetId
+			if _, err := topo.ovn.GetLogicalSwitch(ctx, switchName); err != nil {
+				slog.Info("vpcd reconcile-kv: creating subnet topology", "switch", switchName)
+				if err := topo.reconcileSubnet(ctx, rec.SubnetId, rec.VpcId, rec.CidrBlock); err != nil {
+					slog.Error("vpcd reconcile-kv: failed to create subnet topology", "err", err)
+				} else {
+					result.SwitchesCreated++
+				}
+			}
+		}
+	}
+
+	// 3. Reconcile IGW attachments: ensure attached IGWs have external switch + SNAT
+	igwKV, err := js.KeyValue(handlers_ec2_igw.KVBucketIGW)
+	if err != nil {
+		slog.Debug("vpcd reconcile-kv: IGW KV bucket not available", "err", err)
+	} else {
+		igwKeys, err := igwKV.Keys()
+		if err != nil && !errors.Is(err, nats.ErrNoKeysFound) {
+			slog.Warn("vpcd reconcile-kv: failed to list IGW keys", "err", err)
+		}
+		for _, key := range igwKeys {
+			if key == utils.VersionKey {
+				continue
+			}
+			entry, err := igwKV.Get(key)
+			if err != nil {
+				continue
+			}
+			var rec handlers_ec2_igw.IGWRecord
+			if err := json.Unmarshal(entry.Value(), &rec); err != nil {
+				slog.Warn("vpcd reconcile-kv: failed to unmarshal IGW record", "key", key, "err", err)
+				continue
+			}
+			// Only reconcile attached IGWs
+			if rec.VpcId == "" || rec.State != "attached" {
+				continue
+			}
+
+			extSwitchName := "ext-" + rec.VpcId
+			if _, err := topo.ovn.GetLogicalSwitch(ctx, extSwitchName); err != nil {
+				slog.Info("vpcd reconcile-kv: creating IGW topology", "switch", extSwitchName, "igw_id", rec.InternetGatewayId)
+				if err := topo.reconcileIGW(ctx, rec.VpcId, rec.InternetGatewayId); err != nil {
+					slog.Error("vpcd reconcile-kv: failed to create IGW topology", "err", err)
+				} else {
+					result.IGWsCreated++
+				}
+			}
+		}
+	}
+
+	// 4. Reconcile ENI ports: ensure each ENI has a logical switch port
+	eniKV, err := js.KeyValue(handlers_ec2_vpc.KVBucketENIs)
+	if err != nil {
+		slog.Debug("vpcd reconcile-kv: ENI KV bucket not available", "err", err)
+	} else {
+		eniKeys, err := eniKV.Keys()
+		if err != nil && !errors.Is(err, nats.ErrNoKeysFound) {
+			slog.Warn("vpcd reconcile-kv: failed to list ENI keys", "err", err)
+		}
+		for _, key := range eniKeys {
+			if key == utils.VersionKey {
+				continue
+			}
+			entry, err := eniKV.Get(key)
+			if err != nil {
+				continue
+			}
+			var rec handlers_ec2_vpc.ENIRecord
+			if err := json.Unmarshal(entry.Value(), &rec); err != nil {
+				slog.Warn("vpcd reconcile-kv: failed to unmarshal ENI record", "key", key, "err", err)
+				continue
+			}
+
+			portName := "port-" + rec.NetworkInterfaceId
+			if _, err := topo.ovn.GetLogicalSwitchPort(ctx, portName); err != nil {
+				switchName := "subnet-" + rec.SubnetId
+				addrStr := rec.MacAddress + " " + rec.PrivateIpAddress
+				lsp := &nbdb.LogicalSwitchPort{
+					Name:         portName,
+					Addresses:    []string{addrStr},
+					PortSecurity: []string{addrStr},
+					ExternalIDs: map[string]string{
+						"spinifex:eni_id":    rec.NetworkInterfaceId,
+						"spinifex:subnet_id": rec.SubnetId,
+						"spinifex:vpc_id":    rec.VpcId,
+					},
+				}
+
+				// Attach DHCP options if available
+				dhcpOpts, dhcpErr := topo.ovn.FindDHCPOptionsByExternalID(ctx, "spinifex:subnet_id", rec.SubnetId)
+				if dhcpErr == nil {
+					lsp.DHCPv4Options = &dhcpOpts.UUID
+				}
+
+				slog.Info("vpcd reconcile-kv: creating ENI port", "port", portName, "switch", switchName)
+				if err := topo.ovn.CreateLogicalSwitchPort(ctx, switchName, lsp); err != nil {
+					slog.Error("vpcd reconcile-kv: failed to create ENI port", "port", portName, "err", err)
+				} else {
+					result.PortsCreated++
+				}
+			}
+		}
+	}
+
+	slog.Info("vpcd reconcile-kv: complete",
+		"routers_created", result.RoutersCreated,
+		"switches_created", result.SwitchesCreated,
+		"igws_created", result.IGWsCreated,
+		"ports_created", result.PortsCreated,
+	)
+
+	return result
+}
@@ -2,8 +2,14 @@ package vpcd

 import (
 	"context"
+	"encoding/json"
 	"testing"
+	"time"

+	handlers_ec2_igw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/igw"
+	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/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -147,3 +153,186 @@ func TestReconcile_PartialState(t *testing.T) {
 	assert.Equal(t, 1, result.SwitchesCreated)
 	assert.Equal(t, 1, result.IGWsCreated)
 }
+
+// --- ReconcileFromKV tests ---
+
+// seedKVBuckets creates NATS KV buckets and populates them with test data.
+func seedKVBuckets(t *testing.T, nc *nats.Conn, vpcs []handlers_ec2_vpc.VPCRecord, subnets []handlers_ec2_vpc.SubnetRecord, igws []handlers_ec2_igw.IGWRecord, enis []handlers_ec2_vpc.ENIRecord) {
+	t.Helper()
+	js, err := nc.JetStream()
+	require.NoError(t, err)
+
+	vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1})
+	require.NoError(t, err)
+	for _, v := range vpcs {
+		data, _ := json.Marshal(v)
+		_, err := vpcKV.Put(utils.AccountKey("000000000001", v.VpcId), data)
+		require.NoError(t, err)
+	}
+
+	subnetKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketSubnets, History: 1})
+	require.NoError(t, err)
+	for _, s := range subnets {
+		data, _ := json.Marshal(s)
+		_, err := subnetKV.Put(utils.AccountKey("000000000001", s.SubnetId), data)
+		require.NoError(t, err)
+	}
+
+	igwKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_igw.KVBucketIGW, History: 1})
+	require.NoError(t, err)
+	for _, i := range igws {
+		data, _ := json.Marshal(i)
+		_, err := igwKV.Put(utils.AccountKey("000000000001", i.InternetGatewayId), data)
+		require.NoError(t, err)
+	}
+
+	eniKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketENIs, History: 1})
+	require.NoError(t, err)
+	for _, e := range enis {
+		data, _ := json.Marshal(e)
+		_, err := eniKV.Put(utils.AccountKey("000000000001", e.NetworkInterfaceId), data)
+		require.NoError(t, err)
+	}
+}
+
+func TestReconcileFromKV_CreatesFullTopology(t *testing.T) {
+	_, nc := startTestJetStreamNATS(t)
+	ovn := NewMockOVNClient()
+	_ = ovn.Connect(context.Background())
+	ctx := context.Background()
+
+	topo := NewTopologyHandler(ovn,
+		WithExternalNetwork("pool", []ExternalPoolConfig{{
+			Name:       "wan",
+			RangeStart: "10.0.0.200",
+			RangeEnd:   "10.0.0.250",
+			Gateway:    "10.0.0.1",
+			GatewayIP:  "10.0.0.200",
+			PrefixLen:  24,
+		}}),
+		WithChassisNames([]string{"chassis-node1"}),
+	)
+
+	seedKVBuckets(t, nc,
+		[]handlers_ec2_vpc.VPCRecord{{
+			VpcId: "vpc-kv1", CidrBlock: "10.100.0.0/16", State: "available",
+			CreatedAt: time.Now(),
+		}},
+		[]handlers_ec2_vpc.SubnetRecord{{
+			SubnetId: "subnet-kv1", VpcId: "vpc-kv1", CidrBlock: "10.100.1.0/24",
+			State: "available", CreatedAt: time.Now(),
+		}},
+		[]handlers_ec2_igw.IGWRecord{{
+			InternetGatewayId: "igw-kv1", VpcId: "vpc-kv1", State: "attached",
+			CreatedAt: time.Now(),
+		}},
+		[]handlers_ec2_vpc.ENIRecord{{
+			NetworkInterfaceId: "eni-kv1", SubnetId: "subnet-kv1", VpcId: "vpc-kv1",
+			PrivateIpAddress: "10.100.1.10", MacAddress: "02:00:00:aa:bb:cc",
+			Status: "in-use", CreatedAt: time.Now(),
+		}},
+	)
+
+	result := ReconcileFromKV(ctx, nc, topo)
+	assert.Equal(t, 1, result.RoutersCreated)
+	assert.Equal(t, 1, result.SwitchesCreated)
+	assert.Equal(t, 1, result.IGWsCreated)
+	assert.Equal(t, 1, result.PortsCreated)
+
+	// Verify OVN objects
+	_, err := ovn.GetLogicalRouter(ctx, "vpc-vpc-kv1")
+	require.NoError(t, err)
+
+	_, err = ovn.GetLogicalSwitch(ctx, "subnet-subnet-kv1")
+	require.NoError(t, err)
+
+	_, err = ovn.GetLogicalSwitch(ctx, "ext-vpc-kv1")
+	require.NoError(t, err)
+
+	lsp, err := ovn.GetLogicalSwitchPort(ctx, "port-eni-kv1")
+	require.NoError(t, err)
+	assert.Contains(t, lsp.Addresses[0], "10.100.1.10")
+}
+
+func TestReconcileFromKV_Idempotent(t *testing.T) {
+	_, nc := startTestJetStreamNATS(t)
+	ovn := NewMockOVNClient()
+	_ = ovn.Connect(context.Background())
+	ctx := context.Background()
+
+	topo := NewTopologyHandler(ovn,
+		WithExternalNetwork("pool", []ExternalPoolConfig{{
+			Name: "wan", Gateway: "10.0.0.1", GatewayIP: "10.0.0.200", PrefixLen: 24,
+		}}),
+		WithChassisNames([]string{"chassis-node1"}),
+	)
+
+	seedKVBuckets(t, nc,
+		[]handlers_ec2_vpc.VPCRecord{{
+			VpcId: "vpc-idem2", CidrBlock: "10.200.0.0/16", State: "available",
+			CreatedAt: time.Now(),
+		}},
+		[]handlers_ec2_vpc.SubnetRecord{{
+			SubnetId: "subnet-idem2", VpcId: "vpc-idem2", CidrBlock: "10.200.1.0/24",
+			State: "available", CreatedAt: time.Now(),
+		}},
+		nil, nil,
+	)
+
+	r1 := ReconcileFromKV(ctx, nc, topo)
+	assert.Equal(t, 1, r1.RoutersCreated)
+	assert.Equal(t, 1, r1.SwitchesCreated)
+
+	// Second run: everything exists
+	r2 := ReconcileFromKV(ctx, nc, topo)
+	assert.Equal(t, 0, r2.RoutersCreated)
+	assert.Equal(t, 0, r2.SwitchesCreated)
+}
+
+func TestReconcileFromKV_SkipsDetachedIGW(t *testing.T) {
+	_, nc := startTestJetStreamNATS(t)
+	ovn := NewMockOVNClient()
+	_ = ovn.Connect(context.Background())
+	ctx := context.Background()
+
+	topo := NewTopologyHandler(ovn,
+		WithExternalNetwork("pool", []ExternalPoolConfig{{
+			Name: "wan", Gateway: "10.0.0.1", GatewayIP: "10.0.0.200", PrefixLen: 24,
+		}}),
+	)
+
+	seedKVBuckets(t, nc,
+		[]handlers_ec2_vpc.VPCRecord{{
+			VpcId: "vpc-det", CidrBlock: "10.50.0.0/16", State: "available",
+			CreatedAt: time.Now(),
+		}},
+		nil,
+		[]handlers_ec2_igw.IGWRecord{{
+			InternetGatewayId: "igw-det", VpcId: "", State: "available",
+			CreatedAt: time.Now(),
+		}},
+		nil,
+	)
+
+	result := ReconcileFromKV(ctx, nc, topo)
+	assert.Equal(t, 1, result.RoutersCreated) // VPC router created
+	assert.Equal(t, 0, result.IGWsCreated)    // Detached IGW skipped
+
+	_, err := ovn.GetLogicalSwitch(ctx, "ext-vpc-det")
+	assert.Error(t, err) // External switch should NOT exist
+}
+
+func TestReconcileFromKV_NoBuckets(t *testing.T) {
+	_, nc := startTestJetStreamNATS(t)
+	ovn := NewMockOVNClient()
+	_ = ovn.Connect(context.Background())
+
+	topo := NewTopologyHandler(ovn)
+
+	// No KV buckets seeded — should handle gracefully
+	result := ReconcileFromKV(context.Background(), nc, topo)
+	assert.Equal(t, 0, result.RoutersCreated)
+	assert.Equal(t, 0, result.SwitchesCreated)
+	assert.Equal(t, 0, result.IGWsCreated)
+	assert.Equal(t, 0, result.PortsCreated)
+}
@@ -277,6 +277,10 @@ func launchService(cfg *Config) error {
 		}
 	}()

+	// Pass 2: Reconcile from NATS KV (handles reboots, OVN DB loss, missed events).
+	// Runs after subscribing so new events are not missed during reconciliation.
+	ReconcileFromKV(ctx, nc, topo)
+
 	slog.Info("vpcd service started, waiting for VPC lifecycle events", "subscriptions", len(subs))

 	// Wait for shutdown signal