diff --git a/hive/daemon/state_test.go b/hive/daemon/state_test.go index c406d9f15..a3d46ab66 100644 --- a/hive/daemon/state_test.go +++ b/hive/daemon/state_test.go @@ -61,6 +61,7 @@ func createDaemonWithJetStream(t *testing.T) *Daemon { daemon.jsManager, err = NewJetStreamManager(nc, 1) require.NoError(t, err) require.NoError(t, daemon.jsManager.InitKVBucket()) + require.NoError(t, daemon.jsManager.InitClusterStateBucket()) return daemon } @@ -588,6 +589,15 @@ func TestTransitionState_MultipleInstancesIndependent(t *testing.T) { // running, isInstanceProcessRunning returns false for all instances, which // exercises the recovery state resolution logic. +// simulateCleanRestore writes a clean shutdown marker then calls restoreInstances. +// Without the marker, restoreInstances sleeps 3s waiting for stale QEMU PIDs — +// unnecessary in tests since no QEMU process ever runs. +func simulateCleanRestore(t *testing.T, daemon *Daemon) { + t.Helper() + require.NoError(t, daemon.jsManager.WriteShutdownMarker(daemon.node)) + daemon.restoreInstances() +} + // TestRestoreInstances_StoppingFinalizedToStopped verifies that an instance // stuck in StateStopping when the daemon died gets finalized to StateStopped // and migrated to shared KV. @@ -605,7 +615,7 @@ func TestRestoreInstances_StoppingFinalizedToStopped(t *testing.T) { daemon.Instances.VMS = make(map[string]*vm.VM) // restoreInstances loads from JetStream and resolves transitional states. - daemon.restoreInstances() + simulateCleanRestore(t, daemon) // Stopped instances are migrated to shared KV and removed from local map _, ok := daemon.Instances.VMS["i-restore-stop"] @@ -630,7 +640,7 @@ func TestRestoreInstances_ShuttingDownFinalizedToTerminated(t *testing.T) { require.NoError(t, daemon.WriteState()) daemon.Instances.VMS = make(map[string]*vm.VM) - daemon.restoreInstances() + simulateCleanRestore(t, daemon) instance, ok := daemon.Instances.VMS["i-restore-shut"] require.True(t, ok, "instance should be loaded from state") @@ -650,7 +660,7 @@ func TestRestoreInstances_TerminatedSkipped(t *testing.T) { require.NoError(t, daemon.WriteState()) daemon.Instances.VMS = make(map[string]*vm.VM) - daemon.restoreInstances() + simulateCleanRestore(t, daemon) instance, ok := daemon.Instances.VMS["i-restore-term"] require.True(t, ok, "terminated instance should still be loaded") @@ -673,7 +683,7 @@ func TestRestoreInstances_UserStoppedMigratedToSharedKV(t *testing.T) { require.NoError(t, daemon.WriteState()) daemon.Instances.VMS = make(map[string]*vm.VM) - daemon.restoreInstances() + simulateCleanRestore(t, daemon) // Stopped instances should be migrated to shared KV _, ok := daemon.Instances.VMS["i-restore-userstop"] @@ -700,7 +710,7 @@ func TestRestoreInstances_RunningResetToPending(t *testing.T) { require.NoError(t, daemon.WriteState()) daemon.Instances.VMS = make(map[string]*vm.VM) - daemon.restoreInstances() + simulateCleanRestore(t, daemon) instance, ok := daemon.Instances.VMS["i-restore-run"] require.True(t, ok, "running instance should still be loaded") @@ -732,7 +742,7 @@ func TestRestoreInstances_MixedStates(t *testing.T) { require.NoError(t, daemon.WriteState()) daemon.Instances.VMS = make(map[string]*vm.VM) - daemon.restoreInstances() + simulateCleanRestore(t, daemon) // Stopped instances are migrated to shared KV and removed from local map assert.Nil(t, daemon.Instances.VMS["i-mix-stopping"], @@ -771,7 +781,7 @@ func TestRestoreInstances_StatePersistsAfterRecovery(t *testing.T) { // First restart: finalize transitional states. daemon.Instances.VMS = make(map[string]*vm.VM) - daemon.restoreInstances() + simulateCleanRestore(t, daemon) // stopping→stopped migrates to shared KV assert.Nil(t, daemon.Instances.VMS["i-persist-stop"], @@ -786,7 +796,7 @@ func TestRestoreInstances_StatePersistsAfterRecovery(t *testing.T) { // Second restart: stopped instance is in shared KV (not per-node state), // so local map won't have it. Terminated instance persists in per-node state. daemon.Instances.VMS = make(map[string]*vm.VM) - daemon.restoreInstances() + simulateCleanRestore(t, daemon) // Stopped instance should still be in shared KV stoppedInst2, err := daemon.jsManager.LoadStoppedInstance("i-persist-stop") @@ -844,7 +854,7 @@ func TestRestoreInstances_StoppedInstanceMigratedToSharedKV(t *testing.T) { // Simulate daemon restart: clear in-memory state and restore from JetStream. daemon.Instances.VMS = make(map[string]*vm.VM) - daemon.restoreInstances() + simulateCleanRestore(t, daemon) // Instance should not be in local map (migrated to shared KV). _, ok := daemon.Instances.VMS[instanceID] @@ -876,7 +886,7 @@ func TestRestoreInstances_UnknownTypeMigratedToStoppedWithReason(t *testing.T) { // Simulate daemon restart daemon.Instances.VMS = make(map[string]*vm.VM) - daemon.restoreInstances() + simulateCleanRestore(t, daemon) // Instance should not be in local map (migrated to shared KV) _, ok := daemon.Instances.VMS[instanceID] diff --git a/hive/services/viperblockd/viperblockd_handlers_test.go b/hive/services/viperblockd/viperblockd_handlers_test.go index ccacb3269..68113e9a4 100644 --- a/hive/services/viperblockd/viperblockd_handlers_test.go +++ b/hive/services/viperblockd/viperblockd_handlers_test.go @@ -1,6 +1,6 @@ package viperblockd -// Tests for NATS message handlers: ebs.delete, ebs.sync, ebs.unmount (socket cleanup) +// Tests for NATS message handlers: ebs.delete, ebs.sync, ebs.unmount, ebs.snapshot // These extend the existing integration tests to cover untested handler paths. import ( @@ -11,11 +11,32 @@ import ( "time" "github.com/mulgadc/hive/hive/config" + "github.com/mulgadc/viperblock/viperblock" + "github.com/mulgadc/viperblock/viperblock/backends/file" "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) +// createTestVBWithState creates a real *viperblock.VB with file backend for handler tests. +func createTestVBWithState(t *testing.T, volumeName string) *viperblock.VB { + t.Helper() + baseDir := t.TempDir() + fileCfg := file.FileConfig{VolumeName: volumeName, BaseDir: baseDir} + vbCfg := &viperblock.VB{ + VolumeName: volumeName, + VolumeSize: 1, // minimal + BaseDir: baseDir, + WALSyncInterval: -1, // disable background goroutine + Cache: viperblock.Cache{Config: viperblock.CacheConfig{Size: 0}}, + } + vb, err := viperblock.New(vbCfg, "file", fileCfg) + require.NoError(t, err) + require.NoError(t, vb.Backend.Init()) + require.NoError(t, vb.SaveState()) + return vb +} + // --- ebs.delete handler tests --- func TestIntegration_EBSDeleteMountedVolume(t *testing.T) { @@ -299,6 +320,318 @@ func TestIntegration_EBSDeleteRemovesSocket(t *testing.T) { assert.False(t, fileExistsCheck(socketPath)) } +// --- ebs.snapshot handler tests --- + +func TestIntegration_SnapshotHandler_Success(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + vb := createTestVBWithState(t, "vol-snap-ok") + + snapSub, err := nc.Subscribe("ebs.snapshot.vol-snap-ok", makeSnapshotHandler(vb, "vol-snap-ok")) + require.NoError(t, err) + defer snapSub.Unsubscribe() + nc.Flush() + + reqData, _ := json.Marshal(config.EBSSnapshotRequest{Volume: "vol-snap-ok", SnapshotID: "snap-001"}) + msg, err := nc.Request("ebs.snapshot.vol-snap-ok", reqData, 3*time.Second) + require.NoError(t, err) + + var resp config.EBSSnapshotResponse + require.NoError(t, json.Unmarshal(msg.Data, &resp)) + assert.True(t, resp.Success) + assert.Equal(t, "snap-001", resp.SnapshotID) + assert.Empty(t, resp.Error) +} + +func TestIntegration_SnapshotHandler_InvalidJSON(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + vb := createTestVBWithState(t, "vol-snap-badjson") + + snapSub, err := nc.Subscribe("ebs.snapshot.vol-snap-badjson", makeSnapshotHandler(vb, "vol-snap-badjson")) + require.NoError(t, err) + defer snapSub.Unsubscribe() + nc.Flush() + + msg, err := nc.Request("ebs.snapshot.vol-snap-badjson", []byte("not json {{{"), 3*time.Second) + require.NoError(t, err) + + var resp config.EBSSnapshotResponse + require.NoError(t, json.Unmarshal(msg.Data, &resp)) + assert.Contains(t, resp.Error, "bad request:") +} + +func TestIntegration_SnapshotHandler_CreateSnapshotFailure(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + vb := createTestVBWithState(t, "vol-snap-fail") + + // Make the base directory read-only so WriteTo fails with permission denied + require.NoError(t, os.Chmod(vb.BaseDir, 0555)) + t.Cleanup(func() { os.Chmod(vb.BaseDir, 0755) }) + + snapSub, err := nc.Subscribe("ebs.snapshot.vol-snap-fail", makeSnapshotHandler(vb, "vol-snap-fail")) + require.NoError(t, err) + defer snapSub.Unsubscribe() + nc.Flush() + + reqData, _ := json.Marshal(config.EBSSnapshotRequest{Volume: "vol-snap-fail", SnapshotID: "snap-fail-001"}) + msg, err := nc.Request("ebs.snapshot.vol-snap-fail", reqData, 3*time.Second) + require.NoError(t, err) + + var resp config.EBSSnapshotResponse + require.NoError(t, json.Unmarshal(msg.Data, &resp)) + assert.False(t, resp.Success) + assert.Contains(t, resp.Error, "snapshot failed:") +} + +// --- ebs.sync with VB instance --- + +func TestIntegration_EBSSyncWithVBInstance(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + vb := createTestVBWithState(t, "vol-sync-vb") + + cfg := setupTestConfig(t, natsURL) + cfg.MountedVolumes = []MountedVolume{ + {Name: "vol-sync-vb", VB: vb}, + } + + go func() { launchService(cfg) }() + time.Sleep(500 * time.Millisecond) + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + reqData, _ := json.Marshal(config.EBSSyncRequest{Volume: "vol-sync-vb"}) + msg, err := nc.Request("ebs.sync", reqData, 3*time.Second) + require.NoError(t, err) + + var resp config.EBSSyncResponse + require.NoError(t, json.Unmarshal(msg.Data, &resp)) + assert.True(t, resp.Synced) + assert.Empty(t, resp.Error) +} + +// --- ebs.unmount invalid JSON --- + +func TestIntegration_EBSUnmountInvalidJSON(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + cfg := setupTestConfig(t, natsURL) + + go func() { launchService(cfg) }() + time.Sleep(500 * time.Millisecond) + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + // Current code silently returns without responding on unmarshal failure + _, err = nc.Request("ebs.test-node.unmount", []byte("not json {{{"), 1*time.Second) + assert.Error(t, err, "expected timeout because handler silently drops invalid JSON") +} + +// --- ebs.delete with VB instance (StopWALSyncer path) --- + +func TestIntegration_EBSDeleteWithVBInstance(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + vb := createTestVBWithState(t, "vol-del-vb") + + tmpDir := t.TempDir() + socketPath := filepath.Join(tmpDir, "vol-del-vb.sock") + require.NoError(t, os.WriteFile(socketPath, []byte("fake"), 0600)) + + snapSub, err := nc.Subscribe("ebs.snapshot.vol-del-vb", func(msg *nats.Msg) {}) + require.NoError(t, err) + + cfg := setupTestConfig(t, natsURL) + cfg.MountedVolumes = []MountedVolume{ + { + Name: "vol-del-vb", + Socket: socketPath, + PID: 99999, + VB: vb, + SnapshotSub: snapSub, + }, + } + + go func() { launchService(cfg) }() + time.Sleep(500 * time.Millisecond) + + reqData, _ := json.Marshal(config.EBSDeleteRequest{Volume: "vol-del-vb"}) + msg, err := nc.Request("ebs.delete", reqData, 3*time.Second) + require.NoError(t, err) + + var resp config.EBSDeleteResponse + require.NoError(t, json.Unmarshal(msg.Data, &resp)) + assert.True(t, resp.Success) + assert.Empty(t, resp.Error) + + // Verify full cleanup + assert.False(t, fileExistsCheck(socketPath)) + assert.False(t, snapSub.IsValid()) +} + +// --- ebs.unmount with VB instance + SnapshotSub --- + +func TestIntegration_EBSUnmountWithVBInstance(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + vb := createTestVBWithState(t, "vol-unmount-vb") + + tmpDir := t.TempDir() + socketPath := filepath.Join(tmpDir, "vol-unmount-vb.sock") + require.NoError(t, os.WriteFile(socketPath, []byte("fake"), 0600)) + + snapSub, err := nc.Subscribe("ebs.snapshot.vol-unmount-vb", func(msg *nats.Msg) {}) + require.NoError(t, err) + + cfg := setupTestConfig(t, natsURL) + cfg.MountedVolumes = []MountedVolume{ + { + Name: "vol-unmount-vb", + Socket: socketPath, + PID: 99999, + VB: vb, + SnapshotSub: snapSub, + }, + } + + go func() { launchService(cfg) }() + time.Sleep(500 * time.Millisecond) + + reqData, _ := json.Marshal(config.EBSRequest{Name: "vol-unmount-vb"}) + msg, err := nc.Request("ebs.test-node.unmount", reqData, 3*time.Second) + require.NoError(t, err) + + var resp config.EBSUnMountResponse + require.NoError(t, json.Unmarshal(msg.Data, &resp)) + assert.Equal(t, "vol-unmount-vb", resp.Volume) + assert.False(t, resp.Mounted) + assert.Empty(t, resp.Error) + + // Verify cleanup + assert.False(t, fileExistsCheck(socketPath)) + assert.False(t, snapSub.IsValid()) +} + +// --- ebs.unmount dual-publish verification --- + +func TestIntegration_EBSUnmountDualPublish(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + cfg := setupTestConfig(t, natsURL) + cfg.MountedVolumes = []MountedVolume{ + {Name: "vol-dual-pub", PID: 99999}, + } + + go func() { launchService(cfg) }() + time.Sleep(500 * time.Millisecond) + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + // Subscribe to the broadcast response topic before sending unmount + broadcastCh := make(chan *nats.Msg, 1) + _, err = nc.Subscribe("ebs.unmount.response", func(msg *nats.Msg) { + broadcastCh <- msg + }) + require.NoError(t, err) + nc.Flush() + + reqData, _ := json.Marshal(config.EBSRequest{Name: "vol-dual-pub"}) + msg, err := nc.Request("ebs.test-node.unmount", reqData, 3*time.Second) + require.NoError(t, err) + + // Verify direct reply + var directResp config.EBSUnMountResponse + require.NoError(t, json.Unmarshal(msg.Data, &directResp)) + assert.Equal(t, "vol-dual-pub", directResp.Volume) + assert.False(t, directResp.Mounted) + + // Verify broadcast response received + select { + case broadcastMsg := <-broadcastCh: + var broadcastResp config.EBSUnMountResponse + require.NoError(t, json.Unmarshal(broadcastMsg.Data, &broadcastResp)) + assert.Equal(t, "vol-dual-pub", broadcastResp.Volume) + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for broadcast on ebs.unmount.response") + } +} + // fileExistsCheck is a helper to check if a file exists on disk. func fileExistsCheck(path string) bool { _, err := os.Stat(path) diff --git a/hive/services/viperblockd/viperblockd_integration_test.go b/hive/services/viperblockd/viperblockd_integration_test.go index 4de0d353f..30917ce46 100644 --- a/hive/services/viperblockd/viperblockd_integration_test.go +++ b/hive/services/viperblockd/viperblockd_integration_test.go @@ -475,14 +475,83 @@ func TestIntegration_ServiceGracefulShutdown(t *testing.T) { t.Log("Service is running and responsive") } -// TestIntegration_NATSReconnection tests behavior when NATS connection is lost -func TestIntegration_NATSReconnection(t *testing.T) { +// TestIntegration_GenericTopicRouting tests that NodeName="" uses generic ebs.unmount with queue group +func TestIntegration_GenericTopicRouting(t *testing.T) { + t.Parallel() if testing.Short() { t.Skip("Skipping integration test in short mode") } - // This test would require stopping and restarting NATS server - // which is complex with the current architecture - // Marked as TODO for future enhancement - t.Skip("TODO: Implement NATS reconnection test") + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + cfg := setupTestConfig(t, natsURL) + cfg.NodeName = "" // Single-node / generic mode + cfg.MountedVolumes = []MountedVolume{ + {Name: "vol-generic", PID: 99999}, + } + + go func() { launchService(cfg) }() + time.Sleep(500 * time.Millisecond) + + nc, err := nats.Connect(natsURL) + assert.NoError(t, err) + defer nc.Close() + + // Send to generic topic (not ebs.{node}.unmount) + request := config.EBSRequest{Name: "vol-generic"} + requestData, _ := json.Marshal(request) + + msg, err := nc.Request("ebs.unmount", requestData, 3*time.Second) + assert.NoError(t, err) + + var response config.EBSUnMountResponse + assert.NoError(t, json.Unmarshal(msg.Data, &response)) + assert.Equal(t, "vol-generic", response.Volume) + assert.False(t, response.Mounted) + assert.Empty(t, response.Error) +} + +// TestIntegration_MountAuxiliaryVolumeSuffix tests that volumes with -cloudinit suffix +// reach the mount handler and trigger the auxiliary volume code path +func TestIntegration_MountAuxiliaryVolumeSuffix(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + ns, natsURL := setupEmbeddedNATS(t) + defer ns.Shutdown() + + cfg := setupTestConfig(t, natsURL) + + go func() { launchService(cfg) }() + time.Sleep(500 * time.Millisecond) + + nc, err := nats.Connect(natsURL) + assert.NoError(t, err) + defer nc.Close() + + nc.Flush() + + // Send mount request for auxiliary volume (will fail at S3 backend, but validates routing) + request := config.EBSRequest{ + Name: "vol-test-cloudinit", + VolType: "gp3", + } + + requestData, _ := json.Marshal(request) + + msg, err := nc.Request("ebs.test-node.mount", requestData, 5*time.Second) + if err != nil { + // Timeout is acceptable — service attempted to process but backend failed + t.Logf("Request error (expected with mocked S3 backend): %v", err) + return + } + + var response config.EBSMountResponse + assert.NoError(t, json.Unmarshal(msg.Data, &response)) + // Expect an error because S3 backend is mocked, but the request was processed + assert.NotEmpty(t, response.Error) + t.Logf("Auxiliary volume mount processed with expected error: %s", response.Error) }