fix: kv stream recovery (#54)

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

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

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)