diff --git a/cmd/spinifex/cmd/admin.go b/cmd/spinifex/cmd/admin.go index e342fcc8a..7128d78d0 100644 --- a/cmd/spinifex/cmd/admin.go +++ b/cmd/spinifex/cmd/admin.go @@ -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") diff --git a/scripts/reset-dev-env.sh b/scripts/reset-dev-env.sh index 8443734bf..7dfff0bfc 100755 --- a/scripts/reset-dev-env.sh +++ b/scripts/reset-dev-env.sh @@ -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 diff --git a/spinifex/services/vpcd/reconcile.go b/spinifex/services/vpcd/reconcile.go index 3cc20252f..2342b03e0 100644 --- a/spinifex/services/vpcd/reconcile.go +++ b/spinifex/services/vpcd/reconcile.go @@ -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 +} diff --git a/spinifex/services/vpcd/reconcile_test.go b/spinifex/services/vpcd/reconcile_test.go index 1b0d6d6e0..2dd9fa71d 100644 --- a/spinifex/services/vpcd/reconcile_test.go +++ b/spinifex/services/vpcd/reconcile_test.go @@ -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) +} diff --git a/spinifex/services/vpcd/vpcd.go b/spinifex/services/vpcd/vpcd.go index e3410ace7..54b3af8ed 100644 --- a/spinifex/services/vpcd/vpcd.go +++ b/spinifex/services/vpcd/vpcd.go @@ -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