From 55486490aaeea3d40b68d20793520acaf1f27ebe Mon Sep 17 00:00:00 2001 From: Julian Sommer <115516918+juliansommer@users.noreply.github.com> Date: Wed, 25 Feb 2026 11:41:28 +1100 Subject: [PATCH] fix: kv stream recovery (#54) --- hive/daemon/jetstream.go | 165 +++++++++++++++++- hive/daemon/jetstream_test.go | 309 ++++++++++++++++++++++++++++++++++ tests/e2e/TEST_COVERAGE.md | 77 ++++++++- 3 files changed, 546 insertions(+), 5 deletions(-) diff --git a/hive/daemon/jetstream.go b/hive/daemon/jetstream.go index 1915dc5d9..32eb707c5 100644 --- a/hive/daemon/jetstream.go +++ b/hive/daemon/jetstream.go @@ -5,6 +5,7 @@ import ( "errors" "log/slog" "strings" + "sync" "time" "github.com/mulgadc/hive/hive/vm" @@ -28,6 +29,7 @@ type JetStreamManager struct { kv nats.KeyValue // hive-instance-state clusterKV nats.KeyValue // hive-cluster-state replicas int + kvMu sync.Mutex // protects kv during recovery } // NewJetStreamManager creates a new JetStreamManager from a NATS connection. @@ -99,6 +101,62 @@ func (m *JetStreamManager) InitClusterStateBucket() error { return nil } +// isStreamUnavailable checks if an error indicates the underlying JetStream stream +// was lost or is unreachable. This can happen during NATS cluster formation when +// streams created with low replication are disrupted by node join/catchup operations. +// Different KV operations surface different errors when the stream is gone: +// - Get/Keys → ErrNoResponders ("no responders available for request") +// - Put/Delete → ErrNoStreamResponse ("no response from stream") +// - Direct stream queries → ErrStreamNotFound ("stream not found") +func isStreamUnavailable(err error) bool { + if err == nil { + return false + } + if errors.Is(err, nats.ErrStreamNotFound) || + errors.Is(err, nats.ErrNoStreamResponse) || + errors.Is(err, nats.ErrNoResponders) { + return true + } + return strings.Contains(err.Error(), "stream not found") +} + +// recoverKVBucket attempts to reconnect to or re-create the instance-state KV bucket +// after the underlying JetStream stream was lost during cluster formation. +// Returns the recovered KV handle directly so callers avoid a racy re-read of m.kv. +func (m *JetStreamManager) recoverKVBucket() (nats.KeyValue, error) { + m.kvMu.Lock() + defer m.kvMu.Unlock() + + // Try to reconnect to existing bucket first (another goroutine may have recovered it) + kv, err := m.js.KeyValue(InstanceStateBucket) + if err == nil { + m.kv = kv + slog.Info("Reconnected to KV bucket", "bucket", InstanceStateBucket) + return kv, nil + } + + if !errors.Is(err, nats.ErrBucketNotFound) && !isStreamUnavailable(err) { + return nil, err + } + + // Bucket truly doesn't exist — recreate it + slog.Warn("KV bucket stream lost, recreating", "bucket", InstanceStateBucket, "replicas", m.replicas) + kv, err = m.js.CreateKeyValue(&nats.KeyValueConfig{ + Bucket: InstanceStateBucket, + Description: "Hive instance state storage", + History: 1, + Replicas: m.replicas, + }) + if err != nil { + slog.Error("Failed to recreate KV bucket", "bucket", InstanceStateBucket, "err", err) + return nil, err + } + + m.kv = kv + slog.Info("KV bucket recreated successfully", "bucket", InstanceStateBucket) + return kv, nil +} + // Heartbeat represents a daemon's periodic health status published to cluster KV. type Heartbeat struct { Node string `json:"node"` @@ -274,6 +332,18 @@ func (m *JetStreamManager) WriteState(nodeID string, instances *vm.Instances) er key := InstanceStatePrefix + nodeID _, err = m.kv.Put(key, jsonData) if err != nil { + if isStreamUnavailable(err) { + slog.Warn("KV stream unavailable, attempting recovery", "operation", "WriteState", "key", key, "err", err) + kv, recoverErr := m.recoverKVBucket() + if recoverErr != nil { + return err + } + if _, retryErr := kv.Put(key, jsonData); retryErr != nil { + return retryErr + } + slog.Debug("Wrote state to JetStream KV (after recovery)", "key", key, "instances", len(instances.VMS)) + return nil + } return err } @@ -291,11 +361,28 @@ func (m *JetStreamManager) LoadState(nodeID string) (*vm.Instances, error) { entry, err := m.kv.Get(key) if err != nil { if errors.Is(err, nats.ErrKeyNotFound) { - // Key doesn't exist, return empty state slog.Debug("No existing state in JetStream KV, returning empty state", "key", key) return &vm.Instances{VMS: make(map[string]*vm.VM)}, nil } - return nil, err + if isStreamUnavailable(err) { + slog.Warn("KV stream unavailable, attempting recovery", "operation", "LoadState", "key", key, "err", err) + kv, recoverErr := m.recoverKVBucket() + if recoverErr != nil { + return nil, err + } + // Retry the read — if we reconnected, data may still exist + entry, err = kv.Get(key) + if err != nil { + if errors.Is(err, nats.ErrKeyNotFound) { + slog.Warn("No state found after KV recovery", "key", key) + return &vm.Instances{VMS: make(map[string]*vm.VM)}, nil + } + return nil, err + } + // Fall through to unmarshal below + } else { + return nil, err + } } var instances vm.Instances @@ -321,6 +408,18 @@ func (m *JetStreamManager) DeleteState(nodeID string) error { key := InstanceStatePrefix + nodeID err := m.kv.Delete(key) if err != nil && !errors.Is(err, nats.ErrKeyNotFound) { + if isStreamUnavailable(err) { + slog.Warn("KV stream unavailable, attempting recovery", "operation", "DeleteState", "key", key, "err", err) + kv, recoverErr := m.recoverKVBucket() + if recoverErr != nil { + return err + } + // Retry — if we reconnected the key may still exist + if retryErr := kv.Delete(key); retryErr != nil && !errors.Is(retryErr, nats.ErrKeyNotFound) { + return retryErr + } + return nil + } return err } @@ -393,6 +492,18 @@ func (m *JetStreamManager) WriteStoppedInstance(instanceID string, instance *vm. key := StoppedInstancePrefix + instanceID _, err = m.kv.Put(key, jsonData) if err != nil { + if isStreamUnavailable(err) { + slog.Warn("KV stream unavailable, attempting recovery", "operation", "WriteStoppedInstance", "key", key, "err", err) + kv, recoverErr := m.recoverKVBucket() + if recoverErr != nil { + return err + } + if _, retryErr := kv.Put(key, jsonData); retryErr != nil { + return retryErr + } + slog.Debug("Wrote stopped instance to JetStream KV (after recovery)", "key", key, "instanceId", instanceID) + return nil + } return err } @@ -413,7 +524,24 @@ func (m *JetStreamManager) LoadStoppedInstance(instanceID string) (*vm.VM, error if errors.Is(err, nats.ErrKeyNotFound) { return nil, nil } - return nil, err + if isStreamUnavailable(err) { + slog.Warn("KV stream unavailable, attempting recovery", "operation", "LoadStoppedInstance", "key", key, "err", err) + kv, recoverErr := m.recoverKVBucket() + if recoverErr != nil { + return nil, err + } + // Retry — if we reconnected, data may still exist + entry, err = kv.Get(key) + if err != nil { + if errors.Is(err, nats.ErrKeyNotFound) { + return nil, nil + } + return nil, err + } + // Fall through to unmarshal below + } else { + return nil, err + } } var instance vm.VM @@ -434,6 +562,18 @@ func (m *JetStreamManager) DeleteStoppedInstance(instanceID string) error { key := StoppedInstancePrefix + instanceID err := m.kv.Delete(key) if err != nil && !errors.Is(err, nats.ErrKeyNotFound) { + if isStreamUnavailable(err) { + slog.Warn("KV stream unavailable, attempting recovery", "operation", "DeleteStoppedInstance", "key", key, "err", err) + kv, recoverErr := m.recoverKVBucket() + if recoverErr != nil { + return err + } + // Retry — if we reconnected the key may still exist + if retryErr := kv.Delete(key); retryErr != nil && !errors.Is(retryErr, nats.ErrKeyNotFound) { + return retryErr + } + return nil + } return err } @@ -452,7 +592,24 @@ func (m *JetStreamManager) ListStoppedInstances() ([]*vm.VM, error) { if errors.Is(err, nats.ErrNoKeysFound) { return nil, nil } - return nil, err + if isStreamUnavailable(err) { + slog.Warn("KV stream unavailable, attempting recovery", "operation", "ListStoppedInstances", "err", err) + kv, recoverErr := m.recoverKVBucket() + if recoverErr != nil { + return nil, err + } + // Retry — if we reconnected, data may still exist + keys, err = kv.Keys() + if err != nil { + if errors.Is(err, nats.ErrNoKeysFound) { + return nil, nil + } + return nil, err + } + // Fall through to iterate keys below + } else { + return nil, err + } } var instances []*vm.VM diff --git a/hive/daemon/jetstream_test.go b/hive/daemon/jetstream_test.go index b0e399718..4ebfbebcf 100644 --- a/hive/daemon/jetstream_test.go +++ b/hive/daemon/jetstream_test.go @@ -2,6 +2,7 @@ package daemon import ( "encoding/json" + "errors" "testing" "github.com/mulgadc/hive/hive/vm" @@ -653,3 +654,311 @@ func TestJetStreamManager_WriteServiceManifest_ClusterKVNotInitialized(t *testin assert.Error(t, err) assert.Contains(t, err.Error(), "cluster state KV not initialized") } + +// --- KV bucket recovery tests --- +// These tests simulate the stream being deleted after initialization (which happens +// during NATS cluster formation when catchup resets corrupt a stream) and verify +// that operations recover by re-creating the bucket. + +// deleteInstanceStateBucket deletes the underlying JetStream stream for the instance-state KV bucket. +func deleteInstanceStateBucket(t *testing.T, nc *nats.Conn) { + t.Helper() + js, err := nc.JetStream() + require.NoError(t, err) + err = js.DeleteStream("KV_" + InstanceStateBucket) + require.NoError(t, err) +} + +func TestJetStreamManager_WriteState_RecoverAfterStreamLost(t *testing.T) { + nc, err := nats.Connect(sharedJSNATSURL) + require.NoError(t, err) + defer nc.Close() + + jsm, err := NewJetStreamManager(nc, 1) + require.NoError(t, err) + err = jsm.InitKVBucket() + require.NoError(t, err) + + // Delete the underlying stream to simulate cluster formation loss + deleteInstanceStateBucket(t, nc) + + // WriteState should recover by recreating the bucket + testInstances := &vm.Instances{ + VMS: map[string]*vm.VM{ + "i-recover-001": {ID: "i-recover-001", Status: vm.StateRunning}, + }, + } + err = jsm.WriteState("recovery-node", testInstances) + require.NoError(t, err, "WriteState should recover after stream loss") + + // Verify data was written + loaded, err := jsm.LoadState("recovery-node") + require.NoError(t, err) + assert.Len(t, loaded.VMS, 1) + assert.NotNil(t, loaded.VMS["i-recover-001"]) +} + +func TestJetStreamManager_LoadState_RecoverAfterStreamLost(t *testing.T) { + nc, err := nats.Connect(sharedJSNATSURL) + require.NoError(t, err) + defer nc.Close() + + jsm, err := NewJetStreamManager(nc, 1) + require.NoError(t, err) + err = jsm.InitKVBucket() + require.NoError(t, err) + + deleteInstanceStateBucket(t, nc) + + // LoadState should recover and return empty state + loaded, err := jsm.LoadState("any-node") + require.NoError(t, err, "LoadState should recover after stream loss") + assert.Empty(t, loaded.VMS) +} + +func TestJetStreamManager_DeleteState_RecoverAfterStreamLost(t *testing.T) { + nc, err := nats.Connect(sharedJSNATSURL) + require.NoError(t, err) + defer nc.Close() + + jsm, err := NewJetStreamManager(nc, 1) + require.NoError(t, err) + err = jsm.InitKVBucket() + require.NoError(t, err) + + deleteInstanceStateBucket(t, nc) + + // DeleteState should recover (nothing to delete in fresh bucket) + err = jsm.DeleteState("any-node") + require.NoError(t, err, "DeleteState should recover after stream loss") +} + +func TestJetStreamManager_WriteStoppedInstance_RecoverAfterStreamLost(t *testing.T) { + nc, err := nats.Connect(sharedJSNATSURL) + require.NoError(t, err) + defer nc.Close() + + jsm, err := NewJetStreamManager(nc, 1) + require.NoError(t, err) + err = jsm.InitKVBucket() + require.NoError(t, err) + + deleteInstanceStateBucket(t, nc) + + // WriteStoppedInstance should recover + testVM := &vm.VM{ID: "i-stopped-recover", Status: vm.StateStopped, InstanceType: "t3.micro"} + err = jsm.WriteStoppedInstance(testVM.ID, testVM) + require.NoError(t, err, "WriteStoppedInstance should recover after stream loss") + + // Verify data was written + loaded, err := jsm.LoadStoppedInstance(testVM.ID) + require.NoError(t, err) + require.NotNil(t, loaded) + assert.Equal(t, "i-stopped-recover", loaded.ID) + + _ = jsm.DeleteStoppedInstance(testVM.ID) +} + +func TestJetStreamManager_LoadStoppedInstance_RecoverAfterStreamLost(t *testing.T) { + nc, err := nats.Connect(sharedJSNATSURL) + require.NoError(t, err) + defer nc.Close() + + jsm, err := NewJetStreamManager(nc, 1) + require.NoError(t, err) + err = jsm.InitKVBucket() + require.NoError(t, err) + + deleteInstanceStateBucket(t, nc) + + // LoadStoppedInstance should recover and return nil (not found) + loaded, err := jsm.LoadStoppedInstance("i-nonexistent") + require.NoError(t, err, "LoadStoppedInstance should recover after stream loss") + assert.Nil(t, loaded) +} + +func TestJetStreamManager_DeleteStoppedInstance_RecoverAfterStreamLost(t *testing.T) { + nc, err := nats.Connect(sharedJSNATSURL) + require.NoError(t, err) + defer nc.Close() + + jsm, err := NewJetStreamManager(nc, 1) + require.NoError(t, err) + err = jsm.InitKVBucket() + require.NoError(t, err) + + deleteInstanceStateBucket(t, nc) + + err = jsm.DeleteStoppedInstance("i-any") + require.NoError(t, err, "DeleteStoppedInstance should recover after stream loss") +} + +func TestJetStreamManager_ListStoppedInstances_RecoverAfterStreamLost(t *testing.T) { + nc, err := nats.Connect(sharedJSNATSURL) + require.NoError(t, err) + defer nc.Close() + + jsm, err := NewJetStreamManager(nc, 1) + require.NoError(t, err) + err = jsm.InitKVBucket() + require.NoError(t, err) + + deleteInstanceStateBucket(t, nc) + + // ListStoppedInstances should recover and return empty + instances, err := jsm.ListStoppedInstances() + require.NoError(t, err, "ListStoppedInstances should recover after stream loss") + assert.Empty(t, instances) +} + +// --- KV bucket recovery failure tests --- +// These tests verify that when recovery itself fails (e.g., NATS is unreachable), +// errors are properly propagated rather than silently returning empty state. + +// swapToNonJSContext replaces the JetStreamManager's JS context with one from the +// non-JetStream NATS server. The original kv handle still targets the JS server +// (where the stream was deleted), so operations fail with stream-unavailable errors. +// Recovery then also fails because the non-JS server has no JetStream support. +func swapToNonJSContext(t *testing.T, jsm *JetStreamManager) { + t.Helper() + nc, err := nats.Connect(sharedNATSURL) + require.NoError(t, err) + t.Cleanup(func() { nc.Close() }) + js, err := nc.JetStream() + require.NoError(t, err) + jsm.js = js +} + +func TestJetStreamManager_WriteState_RecoveryFailure(t *testing.T) { + nc, err := nats.Connect(sharedJSNATSURL) + require.NoError(t, err) + defer nc.Close() + + jsm, err := NewJetStreamManager(nc, 1) + require.NoError(t, err) + err = jsm.InitKVBucket() + require.NoError(t, err) + + deleteInstanceStateBucket(t, nc) + swapToNonJSContext(t, jsm) + + testInstances := &vm.Instances{VMS: map[string]*vm.VM{"i-fail": {ID: "i-fail"}}} + err = jsm.WriteState("fail-node", testInstances) + assert.Error(t, err, "WriteState should return error when recovery fails") +} + +func TestJetStreamManager_LoadState_RecoveryFailure(t *testing.T) { + nc, err := nats.Connect(sharedJSNATSURL) + require.NoError(t, err) + defer nc.Close() + + jsm, err := NewJetStreamManager(nc, 1) + require.NoError(t, err) + err = jsm.InitKVBucket() + require.NoError(t, err) + + deleteInstanceStateBucket(t, nc) + swapToNonJSContext(t, jsm) + + loaded, err := jsm.LoadState("fail-node") + assert.Error(t, err, "LoadState should return error when recovery fails") + assert.Nil(t, loaded) +} + +func TestJetStreamManager_DeleteState_RecoveryFailure(t *testing.T) { + nc, err := nats.Connect(sharedJSNATSURL) + require.NoError(t, err) + defer nc.Close() + + jsm, err := NewJetStreamManager(nc, 1) + require.NoError(t, err) + err = jsm.InitKVBucket() + require.NoError(t, err) + + deleteInstanceStateBucket(t, nc) + swapToNonJSContext(t, jsm) + + err = jsm.DeleteState("fail-node") + assert.Error(t, err, "DeleteState should return error when recovery fails") +} + +func TestJetStreamManager_WriteStoppedInstance_RecoveryFailure(t *testing.T) { + nc, err := nats.Connect(sharedJSNATSURL) + require.NoError(t, err) + defer nc.Close() + + jsm, err := NewJetStreamManager(nc, 1) + require.NoError(t, err) + err = jsm.InitKVBucket() + require.NoError(t, err) + + deleteInstanceStateBucket(t, nc) + swapToNonJSContext(t, jsm) + + testVM := &vm.VM{ID: "i-fail", Status: vm.StateStopped} + err = jsm.WriteStoppedInstance(testVM.ID, testVM) + assert.Error(t, err, "WriteStoppedInstance should return error when recovery fails") +} + +func TestJetStreamManager_LoadStoppedInstance_RecoveryFailure(t *testing.T) { + nc, err := nats.Connect(sharedJSNATSURL) + require.NoError(t, err) + defer nc.Close() + + jsm, err := NewJetStreamManager(nc, 1) + require.NoError(t, err) + err = jsm.InitKVBucket() + require.NoError(t, err) + + deleteInstanceStateBucket(t, nc) + swapToNonJSContext(t, jsm) + + loaded, err := jsm.LoadStoppedInstance("i-fail") + assert.Error(t, err, "LoadStoppedInstance should return error when recovery fails") + assert.Nil(t, loaded) +} + +func TestJetStreamManager_DeleteStoppedInstance_RecoveryFailure(t *testing.T) { + nc, err := nats.Connect(sharedJSNATSURL) + require.NoError(t, err) + defer nc.Close() + + jsm, err := NewJetStreamManager(nc, 1) + require.NoError(t, err) + err = jsm.InitKVBucket() + require.NoError(t, err) + + deleteInstanceStateBucket(t, nc) + swapToNonJSContext(t, jsm) + + err = jsm.DeleteStoppedInstance("i-fail") + assert.Error(t, err, "DeleteStoppedInstance should return error when recovery fails") +} + +func TestJetStreamManager_ListStoppedInstances_RecoveryFailure(t *testing.T) { + nc, err := nats.Connect(sharedJSNATSURL) + require.NoError(t, err) + defer nc.Close() + + jsm, err := NewJetStreamManager(nc, 1) + require.NoError(t, err) + err = jsm.InitKVBucket() + require.NoError(t, err) + + deleteInstanceStateBucket(t, nc) + swapToNonJSContext(t, jsm) + + instances, err := jsm.ListStoppedInstances() + assert.Error(t, err, "ListStoppedInstances should return error when recovery fails") + assert.Nil(t, instances) +} + +func TestIsStreamUnavailable(t *testing.T) { + assert.False(t, isStreamUnavailable(nil)) + assert.True(t, isStreamUnavailable(nats.ErrStreamNotFound)) + assert.True(t, isStreamUnavailable(nats.ErrNoStreamResponse)) + assert.True(t, isStreamUnavailable(nats.ErrNoResponders)) + assert.True(t, isStreamUnavailable(errors.New("nats: stream not found"))) + assert.False(t, isStreamUnavailable(errors.New("some other error"))) + assert.False(t, isStreamUnavailable(nats.ErrKeyNotFound)) +} diff --git a/tests/e2e/TEST_COVERAGE.md b/tests/e2e/TEST_COVERAGE.md index 7434eabf3..eb4ffa1c4 100644 --- a/tests/e2e/TEST_COVERAGE.md +++ b/tests/e2e/TEST_COVERAGE.md @@ -9,6 +9,11 @@ - Start all services (`start-dev.sh`) - Wait for AWS gateway on `localhost:9999` +### Phase 1b: Cluster Stats CLI +- `hive get nodes` — verify node shows Ready +- `hive top nodes` — verify CPU/MEM resource stats +- `hive get vms` — verify "No VMs found" before any launches + ### Phase 2: Discovery & Metadata - `describe-regions` - `describe-availability-zones` (verify zone name and state) @@ -34,12 +39,21 @@ - `run-instances` (launch VM with key pair) - `describe-instances` (poll pending -> running) +### Phase 5a-pre: Cluster Stats CLI (with running VM) +- `hive get vms` — verify running instance appears in output + ### Phase 5a: Instance Metadata Validation - `describe-instances` — verify InstanceType matches requested type - Verify KeyName matches requested key - Verify ImageId matches requested AMI - Verify at least 1 BlockDeviceMapping present +### Phase 5a-ii: SSH Connectivity & Volume Verification +- SSH into instance via QEMU hostfwd port +- Verify SSH connectivity (`id` command returns ec2-user) +- Verify root volume size from inside VM matches API-reported size (`lsblk` vs `describe-volumes`) +- Verify VM hostname + ### Phase 5a-iii: Console Output & Serial Console Enforcement - `get-console-output` when serial console disabled (expect `SerialConsoleSessionUnavailable`) - `enable-serial-console-access` @@ -141,7 +155,10 @@ - `delete-snapshot` (CreateImage backing snapshot, so DeleteOnTermination is not blocked) - `terminate-instances` (poll -> terminated) -### Phase 9a: Volume Cleanup Verification +### Phase 9a: SSH Unreachable Verification +- Verify SSH connection is refused/unreachable after termination + +### Phase 9b: Volume Cleanup Verification - `describe-volumes` on root volume after termination - Verify root volume is deleted (DeleteOnTermination) @@ -261,6 +278,22 @@ - Verify crash loop is detected and restarts stop after max attempts (3 in 10-min window) - Verify instance reaches error state (won't restart further) +### Phase 5c: VPC Networking +- Step 1: `create-vpc` (10.100.0.0/16) + `create-subnet` (10.100.1.0/24) +- Step 2: `run-instances` x3 with `--subnet-id` (launch into VPC subnet) +- Poll all VPC instances to running state +- Step 3: Verify `PrivateIpAddress` in `describe-instances` for each instance +- Verify `SubnetId` and `VpcId` match requested values +- Verify at least 1 `NetworkInterface` per instance +- Verify all IPs are unique and in subnet range (10.100.1.x) +- Step 4: SSH + ping connectivity (skipped in CI — OVN DHCP wait too slow) +- Step 5: Stop/start IP persistence + - `stop-instances` (all VPC instances, poll -> stopped) + - Verify `PrivateIpAddress` persists in stopped state + - `start-instances` (all VPC instances, poll -> running) + - Verify `PrivateIpAddress` identical after restart +- Step 6: Cleanup — terminate VPC instances, delete subnet, delete VPC + ### Phase 6: Cluster Shutdown + Restart #### Test 6a: Dry-Run Shutdown @@ -285,3 +318,45 @@ ### Cleanup - Coordinated cluster shutdown (with fallback to per-node PID stops) - Remove simulated IPs + +--- + +## VPC (`run-vpc-e2e.sh`) + +Standalone VPC networking test suite. Runs against a running Hive endpoint (configurable via `ENDPOINT` env var). OVN topology tests are skipped when OVN is unavailable. + +### Phase 1: VPC CRUD +- `create-vpc` (10.99.0.0/16, verify VpcId) +- `describe-vpcs` (by ID, verify exactly 1 returned) + +### Phase 2: Subnet CRUD +- `create-subnet` (10.99.1.0/24 in VPC, verify SubnetId) +- `describe-subnets` (by ID, verify exactly 1 returned) + +### Phase 3: Internet Gateway CRUD +- `create-internet-gateway` (verify InternetGatewayId) +- `describe-internet-gateways` (by ID, verify exactly 1 returned) + +### Phase 4: Internet Gateway Attach / Detach +- `attach-internet-gateway` (IGW to VPC) +- `describe-internet-gateways` (verify attachment VpcId) +- `delete-internet-gateway` on attached IGW (expect `DependencyViolation` or rejection) +- `detach-internet-gateway` (IGW from VPC) +- `describe-internet-gateways` (verify no attachments) + +### Phase 5: OVN Topology Verification (requires OVN) +- Re-attach IGW for topology inspection +- Verify OVN logical switch exists for subnet +- Verify OVN logical router exists for VPC +- Verify OVN external switch exists (IGW attached) +- Verify SNAT rule on VPC router +- Verify default route (0.0.0.0/0) on VPC router +- Dump full OVN NB DB topology for debugging +- Detach IGW +- Skipped when `ovn-nbctl` is not available + +### Phase 6: Cleanup +- `delete-internet-gateway` +- `delete-subnet` +- `delete-vpc` +- Verify OVN cleanup (no VPC routers remaining, when OVN available)