chore: daemon tests

hive/daemon/daemon.go

@@ -104,6 +104,9 @@ type Daemon struct {
 	// JetStream manager for KV state storage (nil if JetStream disabled)
 	jsManager *JetStreamManager
 
+	// Delay after QMP device_del before blockdev-del (default 1s, 0 in tests)
+	detachDelay time.Duration
+
 	mu sync.Mutex
 }
 
@@ -425,6 +428,7 @@ func NewDaemon(cfg *config.ClusterConfig) *Daemon {
 		Instances:         vm.Instances{VMS: make(map[string]*vm.VM)},
 		natsSubscriptions: make(map[string]*nats.Subscription),
 		startTime:         time.Now(),
+		detachDelay:       1 * time.Second,
 	}
 }
 

hive/daemon/daemon_handlers.go

@@ -392,7 +392,7 @@ func (d *Daemon) handleDetachVolume(msg *nats.Msg, command qmp.Command, instance
 	}
 
 	// Brief pause for guest to acknowledge PCI removal
-	time.Sleep(1 * time.Second)
+	time.Sleep(d.detachDelay)
 
 	// Phase 2: QMP blockdev-del (remove block node)
 	_, blockdevErr := d.SendQMPCommand(instance.QMPClient, qmp.QMPCommand{

hive/daemon/daemon_handlers_test.go

@@ -0,0 +1,657 @@
+package daemon
+
+import (
+	"encoding/json"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/hive/hive/config"
+	handlers_ec2_account "github.com/mulgadc/hive/hive/handlers/ec2/account"
+	handlers_ec2_eigw "github.com/mulgadc/hive/hive/handlers/ec2/eigw"
+	handlers_ec2_image "github.com/mulgadc/hive/hive/handlers/ec2/image"
+	handlers_ec2_key "github.com/mulgadc/hive/hive/handlers/ec2/key"
+	handlers_ec2_snapshot "github.com/mulgadc/hive/hive/handlers/ec2/snapshot"
+	handlers_ec2_tags "github.com/mulgadc/hive/hive/handlers/ec2/tags"
+	handlers_ec2_volume "github.com/mulgadc/hive/hive/handlers/ec2/volume"
+	"github.com/mulgadc/hive/hive/objectstore"
+	"github.com/mulgadc/hive/hive/qmp"
+	"github.com/mulgadc/hive/hive/vm"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// createFullTestDaemon creates a test daemon with ALL services initialized (including
+// key, image, snapshot, tags, eigw, account) using in-memory object stores.
+func createFullTestDaemon(t *testing.T, natsURL string) *Daemon {
+	daemon := createTestDaemon(t, natsURL)
+
+	memStore := objectstore.NewMemoryObjectStore()
+	cfg := daemon.config
+
+	daemon.keyService = handlers_ec2_key.NewKeyServiceImplWithStore(memStore, cfg.Predastore.Bucket, "123456789")
+	daemon.imageService = handlers_ec2_image.NewImageServiceImplWithStore(memStore, cfg.Predastore.Bucket, "123456789")
+	daemon.volumeService = handlers_ec2_volume.NewVolumeServiceImplWithStore(cfg, memStore, daemon.natsConn)
+	daemon.snapshotService = handlers_ec2_snapshot.NewSnapshotServiceImplWithStore(cfg, memStore, daemon.natsConn)
+	daemon.tagsService = handlers_ec2_tags.NewTagsServiceImplWithStore(cfg, memStore)
+	daemon.eigwService = handlers_ec2_eigw.NewEgressOnlyIGWServiceImpl(cfg)
+	daemon.accountService = handlers_ec2_account.NewAccountSettingsServiceImpl(cfg)
+
+	return daemon
+}
+
+// createFullTestDaemonWithJetStream creates a test daemon with JetStream KV enabled,
+// needed for tests involving state transitions (TransitionState calls WriteState).
+func createFullTestDaemonWithJetStream(t *testing.T, natsURL string) *Daemon {
+	daemon := createFullTestDaemon(t, natsURL)
+
+	var err error
+	daemon.jsManager, err = NewJetStreamManager(daemon.natsConn, 1)
+	require.NoError(t, err)
+	err = daemon.jsManager.InitKVBucket()
+	require.NoError(t, err)
+
+	return daemon
+}
+
+// --- handleNATSRequest generic tests ---
+
+type testInput struct {
+	Name string `json:"name"`
+}
+
+type testOutput struct {
+	Greeting string `json:"greeting"`
+}
+
+func TestHandleNATSRequest_ValidRequest(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	nc, err := nats.Connect(natsURL)
+	require.NoError(t, err)
+	defer nc.Close()
+
+	serviceFn := func(in *testInput) (*testOutput, error) {
+		return &testOutput{Greeting: "hello " + in.Name}, nil
+	}
+
+	sub, err := nc.Subscribe("test.greet", func(msg *nats.Msg) {
+		handleNATSRequest(msg, serviceFn)
+	})
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	reqData, _ := json.Marshal(testInput{Name: "world"})
+	reply, err := nc.Request("test.greet", reqData, 5*time.Second)
+	require.NoError(t, err)
+
+	var out testOutput
+	err = json.Unmarshal(reply.Data, &out)
+	require.NoError(t, err)
+	assert.Equal(t, "hello world", out.Greeting)
+}
+
+func TestHandleNATSRequest_MalformedJSON(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	nc, err := nats.Connect(natsURL)
+	require.NoError(t, err)
+	defer nc.Close()
+
+	serviceFn := func(in *testInput) (*testOutput, error) {
+		return &testOutput{Greeting: "hello"}, nil
+	}
+
+	sub, err := nc.Subscribe("test.malformed", func(msg *nats.Msg) {
+		handleNATSRequest(msg, serviceFn)
+	})
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	reply, err := nc.Request("test.malformed", []byte(`{not valid json}`), 5*time.Second)
+	require.NoError(t, err)
+
+	var errResp map[string]any
+	err = json.Unmarshal(reply.Data, &errResp)
+	require.NoError(t, err)
+	assert.Contains(t, errResp, "Code", "Should contain error code")
+}
+
+func TestHandleNATSRequest_ServiceError(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	nc, err := nats.Connect(natsURL)
+	require.NoError(t, err)
+	defer nc.Close()
+
+	serviceFn := func(in *testInput) (*testOutput, error) {
+		return nil, fmt.Errorf("something went wrong")
+	}
+
+	sub, err := nc.Subscribe("test.err", func(msg *nats.Msg) {
+		handleNATSRequest(msg, serviceFn)
+	})
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	reqData, _ := json.Marshal(testInput{Name: "world"})
+	reply, err := nc.Request("test.err", reqData, 5*time.Second)
+	require.NoError(t, err)
+
+	var errResp map[string]any
+	err = json.Unmarshal(reply.Data, &errResp)
+	require.NoError(t, err)
+	assert.Contains(t, errResp, "Code", "Should contain error code")
+}
+
+// --- Handler wrapper tests (representative set via NATS round-trip) ---
+
+func TestHandleEC2CreateKeyPair_RoundTrip(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createFullTestDaemon(t, natsURL)
+
+	sub, err := daemon.natsConn.QueueSubscribe("ec2.CreateKeyPair", "hive-workers", daemon.handleEC2CreateKeyPair)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	input := &ec2.CreateKeyPairInput{
+		KeyName: aws.String("test-key-001"),
+	}
+	reqData, _ := json.Marshal(input)
+	reply, err := daemon.natsConn.Request("ec2.CreateKeyPair", reqData, 5*time.Second)
+	require.NoError(t, err)
+	require.NotNil(t, reply)
+
+	var output ec2.CreateKeyPairOutput
+	err = json.Unmarshal(reply.Data, &output)
+	require.NoError(t, err)
+	assert.Equal(t, "test-key-001", *output.KeyName)
+	assert.NotEmpty(t, *output.KeyFingerprint)
+	assert.NotEmpty(t, *output.KeyMaterial)
+}
+
+func TestHandleEC2CreateTags_RoundTrip(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createFullTestDaemon(t, natsURL)
+
+	sub, err := daemon.natsConn.QueueSubscribe("ec2.CreateTags", "hive-workers", daemon.handleEC2CreateTags)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	input := &ec2.CreateTagsInput{
+		Resources: []*string{aws.String("i-12345678")},
+		Tags: []*ec2.Tag{
+			{Key: aws.String("Name"), Value: aws.String("test-instance")},
+		},
+	}
+	reqData, _ := json.Marshal(input)
+	reply, err := daemon.natsConn.Request("ec2.CreateTags", reqData, 5*time.Second)
+	require.NoError(t, err)
+	require.NotNil(t, reply)
+
+	var output ec2.CreateTagsOutput
+	err = json.Unmarshal(reply.Data, &output)
+	require.NoError(t, err)
+}
+
+func TestHandleEC2DescribeImages_RoundTrip(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createFullTestDaemon(t, natsURL)
+
+	sub, err := daemon.natsConn.QueueSubscribe("ec2.DescribeImages", "hive-workers", daemon.handleEC2DescribeImages)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	input := &ec2.DescribeImagesInput{}
+	reqData, _ := json.Marshal(input)
+	reply, err := daemon.natsConn.Request("ec2.DescribeImages", reqData, 5*time.Second)
+	require.NoError(t, err)
+	require.NotNil(t, reply)
+
+	var output ec2.DescribeImagesOutput
+	err = json.Unmarshal(reply.Data, &output)
+	require.NoError(t, err)
+}
+
+func TestHandleEC2DescribeVolumes_RoundTrip(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createFullTestDaemon(t, natsURL)
+
+	sub, err := daemon.natsConn.QueueSubscribe("ec2.DescribeVolumes", "hive-workers", daemon.handleEC2DescribeVolumes)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	input := &ec2.DescribeVolumesInput{}
+	reqData, _ := json.Marshal(input)
+	reply, err := daemon.natsConn.Request("ec2.DescribeVolumes", reqData, 5*time.Second)
+	require.NoError(t, err)
+	require.NotNil(t, reply)
+
+	var output ec2.DescribeVolumesOutput
+	err = json.Unmarshal(reply.Data, &output)
+	require.NoError(t, err)
+}
+
+func TestHandleEC2DescribeKeyPairs_RoundTrip(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createFullTestDaemon(t, natsURL)
+
+	sub, err := daemon.natsConn.QueueSubscribe("ec2.DescribeKeyPairs", "hive-workers", daemon.handleEC2DescribeKeyPairs)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	input := &ec2.DescribeKeyPairsInput{}
+	reqData, _ := json.Marshal(input)
+	reply, err := daemon.natsConn.Request("ec2.DescribeKeyPairs", reqData, 5*time.Second)
+	require.NoError(t, err)
+	require.NotNil(t, reply)
+
+	var output ec2.DescribeKeyPairsOutput
+	err = json.Unmarshal(reply.Data, &output)
+	require.NoError(t, err)
+}
+
+// --- handleHealthCheck tests ---
+
+func TestHandleHealthCheck(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createTestDaemon(t, natsURL)
+
+	topic := fmt.Sprintf("hive.admin.%s.health", daemon.node)
+	sub, err := daemon.natsConn.Subscribe(topic, daemon.handleHealthCheck)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	reply, err := daemon.natsConn.Request(topic, nil, 5*time.Second)
+	require.NoError(t, err)
+	require.NotNil(t, reply)
+
+	var resp config.NodeHealthResponse
+	err = json.Unmarshal(reply.Data, &resp)
+	require.NoError(t, err)
+
+	assert.Equal(t, daemon.node, resp.Node)
+	assert.Equal(t, "running", resp.Status)
+	assert.NotEmpty(t, resp.ConfigHash)
+	assert.GreaterOrEqual(t, resp.Uptime, int64(0))
+}
+
+// --- handleNodeDiscover tests ---
+
+func TestHandleNodeDiscover(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createTestDaemon(t, natsURL)
+
+	sub, err := daemon.natsConn.Subscribe("hive.nodes.discover", daemon.handleNodeDiscover)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	reply, err := daemon.natsConn.Request("hive.nodes.discover", nil, 5*time.Second)
+	require.NoError(t, err)
+	require.NotNil(t, reply)
+
+	var resp NodeDiscoverResponse
+	err = json.Unmarshal(reply.Data, &resp)
+	require.NoError(t, err)
+
+	assert.Equal(t, daemon.node, resp.Node)
+}
+
+// --- handleStopOrTerminateInstance tests (JetStream required for TransitionState) ---
+
+func TestHandleEC2Events_StopInstance(t *testing.T) {
+	natsURL := sharedJSNATSURL
+
+	daemon := createFullTestDaemonWithJetStream(t, natsURL)
+
+	instanceID := "i-test-stop-001"
+	daemon.Instances.VMS[instanceID] = &vm.VM{
+		ID:           instanceID,
+		InstanceType: getTestInstanceType(),
+		Status:       vm.StateRunning,
+		Instance:     &ec2.Instance{},
+		QMPClient:    &qmp.QMPClient{},
+	}
+
+	sub, err := daemon.natsConn.QueueSubscribe(
+		fmt.Sprintf("ec2.cmd.%s", instanceID),
+		"hive-events",
+		daemon.handleEC2Events,
+	)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	cmd := qmp.Command{
+		ID:         instanceID,
+		Attributes: qmp.Attributes{StopInstance: true},
+	}
+	cmdData, _ := json.Marshal(cmd)
+
+	reply, err := daemon.natsConn.Request(
+		fmt.Sprintf("ec2.cmd.%s", instanceID),
+		cmdData,
+		5*time.Second,
+	)
+	require.NoError(t, err)
+	require.NotNil(t, reply)
+
+	// Should get immediate {} response
+	assert.Equal(t, `{}`, string(reply.Data))
+
+	// State should transition to stopping
+	daemon.Instances.Mu.Lock()
+	status := daemon.Instances.VMS[instanceID].Status
+	daemon.Instances.Mu.Unlock()
+	assert.Equal(t, vm.StateStopping, status)
+}
+
+func TestHandleEC2Events_TerminateInstance(t *testing.T) {
+	natsURL := sharedJSNATSURL
+
+	daemon := createFullTestDaemonWithJetStream(t, natsURL)
+
+	instanceID := "i-test-term-001"
+	daemon.Instances.VMS[instanceID] = &vm.VM{
+		ID:           instanceID,
+		InstanceType: getTestInstanceType(),
+		Status:       vm.StateRunning,
+		Instance:     &ec2.Instance{},
+		QMPClient:    &qmp.QMPClient{},
+	}
+
+	sub, err := daemon.natsConn.QueueSubscribe(
+		fmt.Sprintf("ec2.cmd.%s", instanceID),
+		"hive-events",
+		daemon.handleEC2Events,
+	)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	cmd := qmp.Command{
+		ID:         instanceID,
+		Attributes: qmp.Attributes{TerminateInstance: true},
+	}
+	cmdData, _ := json.Marshal(cmd)
+
+	reply, err := daemon.natsConn.Request(
+		fmt.Sprintf("ec2.cmd.%s", instanceID),
+		cmdData,
+		5*time.Second,
+	)
+	require.NoError(t, err)
+	require.NotNil(t, reply)
+
+	assert.Equal(t, `{}`, string(reply.Data))
+
+	daemon.Instances.Mu.Lock()
+	status := daemon.Instances.VMS[instanceID].Status
+	daemon.Instances.Mu.Unlock()
+	assert.Equal(t, vm.StateShuttingDown, status)
+}
+
+func TestHandleEC2Events_InstanceNotFound(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createTestDaemon(t, natsURL)
+
+	sub, err := daemon.natsConn.QueueSubscribe("ec2.cmd.i-nonexistent", "hive-events", daemon.handleEC2Events)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	cmd := qmp.Command{
+		ID:         "i-nonexistent",
+		Attributes: qmp.Attributes{StopInstance: true},
+	}
+	cmdData, _ := json.Marshal(cmd)
+
+	reply, err := daemon.natsConn.Request("ec2.cmd.i-nonexistent", cmdData, 5*time.Second)
+	require.NoError(t, err)
+
+	var errResp map[string]any
+	err = json.Unmarshal(reply.Data, &errResp)
+	require.NoError(t, err)
+	assert.Contains(t, errResp, "Code")
+}
+
+func TestHandleEC2Events_MalformedJSON(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createTestDaemon(t, natsURL)
+
+	sub, err := daemon.natsConn.QueueSubscribe("ec2.cmd.test", "hive-events", daemon.handleEC2Events)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	reply, err := daemon.natsConn.Request("ec2.cmd.test", []byte(`{bad json}`), 5*time.Second)
+	require.NoError(t, err)
+
+	var errResp map[string]any
+	err = json.Unmarshal(reply.Data, &errResp)
+	require.NoError(t, err)
+	assert.Contains(t, errResp, "Code")
+}
+
+// --- respondWithVolumeAttachment tests ---
+
+func TestRespondWithVolumeAttachment(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createTestDaemon(t, natsURL)
+
+	sub, err := daemon.natsConn.Subscribe("test.volume.attach", func(msg *nats.Msg) {
+		respondWithError := func(errCode string) {
+			msg.Respond([]byte(fmt.Sprintf(`{"Code":"%s"}`, errCode)))
+		}
+		daemon.respondWithVolumeAttachment(msg, respondWithError, "vol-123", "i-456", "/dev/sdf", "attached")
+	})
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	reply, err := daemon.natsConn.Request("test.volume.attach", nil, 5*time.Second)
+	require.NoError(t, err)
+
+	var attachment ec2.VolumeAttachment
+	err = json.Unmarshal(reply.Data, &attachment)
+	require.NoError(t, err)
+
+	assert.Equal(t, "vol-123", *attachment.VolumeId)
+	assert.Equal(t, "i-456", *attachment.InstanceId)
+	assert.Equal(t, "/dev/sdf", *attachment.Device)
+	assert.Equal(t, "attached", *attachment.State)
+	assert.NotNil(t, attachment.AttachTime)
+	assert.Equal(t, false, *attachment.DeleteOnTermination)
+}
+
+// --- handleEC2ModifyVolume tests ---
+
+func TestHandleEC2ModifyVolume_MalformedInput(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createFullTestDaemon(t, natsURL)
+
+	sub, err := daemon.natsConn.QueueSubscribe("ec2.ModifyVolume", "hive-workers", daemon.handleEC2ModifyVolume)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	reply, err := daemon.natsConn.Request("ec2.ModifyVolume", []byte(`{bad}`), 5*time.Second)
+	require.NoError(t, err)
+
+	var errResp map[string]any
+	err = json.Unmarshal(reply.Data, &errResp)
+	require.NoError(t, err)
+	assert.Contains(t, errResp, "Code")
+}
+
+func TestHandleEC2ModifyVolume_VolumeNotFound(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createFullTestDaemon(t, natsURL)
+
+	sub, err := daemon.natsConn.QueueSubscribe("ec2.ModifyVolume", "hive-workers", daemon.handleEC2ModifyVolume)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	input := &ec2.ModifyVolumeInput{
+		VolumeId: aws.String("vol-nonexistent"),
+		Size:     aws.Int64(16),
+	}
+	reqData, _ := json.Marshal(input)
+	reply, err := daemon.natsConn.Request("ec2.ModifyVolume", reqData, 5*time.Second)
+	require.NoError(t, err)
+
+	// Should return an error since the volume doesn't exist
+	var errResp map[string]any
+	err = json.Unmarshal(reply.Data, &errResp)
+	require.NoError(t, err)
+	assert.Contains(t, errResp, "Code")
+}
+
+// --- Account settings handler tests ---
+
+func TestHandleEC2GetEbsEncryptionByDefault(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createFullTestDaemon(t, natsURL)
+
+	sub, err := daemon.natsConn.QueueSubscribe("ec2.GetEbsEncryptionByDefault", "hive-workers", daemon.handleEC2GetEbsEncryptionByDefault)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	input := &ec2.GetEbsEncryptionByDefaultInput{}
+	reqData, _ := json.Marshal(input)
+	reply, err := daemon.natsConn.Request("ec2.GetEbsEncryptionByDefault", reqData, 5*time.Second)
+	require.NoError(t, err)
+
+	var output ec2.GetEbsEncryptionByDefaultOutput
+	err = json.Unmarshal(reply.Data, &output)
+	require.NoError(t, err)
+	assert.NotNil(t, output.EbsEncryptionByDefault)
+}
+
+func TestHandleEC2GetSerialConsoleAccessStatus(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createFullTestDaemon(t, natsURL)
+
+	sub, err := daemon.natsConn.QueueSubscribe("ec2.GetSerialConsoleAccessStatus", "hive-workers", daemon.handleEC2GetSerialConsoleAccessStatus)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	input := &ec2.GetSerialConsoleAccessStatusInput{}
+	reqData, _ := json.Marshal(input)
+	reply, err := daemon.natsConn.Request("ec2.GetSerialConsoleAccessStatus", reqData, 5*time.Second)
+	require.NoError(t, err)
+
+	var output ec2.GetSerialConsoleAccessStatusOutput
+	err = json.Unmarshal(reply.Data, &output)
+	require.NoError(t, err)
+	assert.NotNil(t, output.SerialConsoleAccessEnabled)
+}
+
+func TestHandleEC2EnableEbsEncryptionByDefault(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createFullTestDaemon(t, natsURL)
+
+	sub, err := daemon.natsConn.QueueSubscribe("ec2.EnableEbsEncryptionByDefault", "hive-workers", daemon.handleEC2EnableEbsEncryptionByDefault)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	input := &ec2.EnableEbsEncryptionByDefaultInput{}
+	reqData, _ := json.Marshal(input)
+	reply, err := daemon.natsConn.Request("ec2.EnableEbsEncryptionByDefault", reqData, 5*time.Second)
+	require.NoError(t, err)
+
+	var output ec2.EnableEbsEncryptionByDefaultOutput
+	err = json.Unmarshal(reply.Data, &output)
+	require.NoError(t, err)
+	assert.NotNil(t, output.EbsEncryptionByDefault)
+	assert.True(t, *output.EbsEncryptionByDefault)
+}
+
+func TestHandleEC2DisableEbsEncryptionByDefault(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createFullTestDaemon(t, natsURL)
+
+	sub, err := daemon.natsConn.QueueSubscribe("ec2.DisableEbsEncryptionByDefault", "hive-workers", daemon.handleEC2DisableEbsEncryptionByDefault)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	input := &ec2.DisableEbsEncryptionByDefaultInput{}
+	reqData, _ := json.Marshal(input)
+	reply, err := daemon.natsConn.Request("ec2.DisableEbsEncryptionByDefault", reqData, 5*time.Second)
+	require.NoError(t, err)
+
+	var output ec2.DisableEbsEncryptionByDefaultOutput
+	err = json.Unmarshal(reply.Data, &output)
+	require.NoError(t, err)
+	assert.NotNil(t, output.EbsEncryptionByDefault)
+	assert.False(t, *output.EbsEncryptionByDefault)
+}
+
+func TestHandleEC2EnableSerialConsoleAccess(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createFullTestDaemon(t, natsURL)
+
+	sub, err := daemon.natsConn.QueueSubscribe("ec2.EnableSerialConsoleAccess", "hive-workers", daemon.handleEC2EnableSerialConsoleAccess)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	input := &ec2.EnableSerialConsoleAccessInput{}
+	reqData, _ := json.Marshal(input)
+	reply, err := daemon.natsConn.Request("ec2.EnableSerialConsoleAccess", reqData, 5*time.Second)
+	require.NoError(t, err)
+
+	var output ec2.EnableSerialConsoleAccessOutput
+	err = json.Unmarshal(reply.Data, &output)
+	require.NoError(t, err)
+	assert.NotNil(t, output.SerialConsoleAccessEnabled)
+	assert.True(t, *output.SerialConsoleAccessEnabled)
+}
+
+func TestHandleEC2DisableSerialConsoleAccess(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createFullTestDaemon(t, natsURL)
+
+	sub, err := daemon.natsConn.QueueSubscribe("ec2.DisableSerialConsoleAccess", "hive-workers", daemon.handleEC2DisableSerialConsoleAccess)
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	input := &ec2.DisableSerialConsoleAccessInput{}
+	reqData, _ := json.Marshal(input)
+	reply, err := daemon.natsConn.Request("ec2.DisableSerialConsoleAccess", reqData, 5*time.Second)
+	require.NoError(t, err)
+
+	var output ec2.DisableSerialConsoleAccessOutput
+	err = json.Unmarshal(reply.Data, &output)
+	require.NoError(t, err)
+	assert.NotNil(t, output.SerialConsoleAccessEnabled)
+	assert.False(t, *output.SerialConsoleAccessEnabled)
+}
+
+// --- SetConfigPath test ---
+
+func TestSetConfigPath(t *testing.T) {
+	clusterCfg := &config.ClusterConfig{
+		Node:  "node-1",
+		Nodes: map[string]config.Config{"node-1": {BaseDir: "/tmp"}},
+	}
+	daemon := NewDaemon(clusterCfg)
+
+	assert.Empty(t, daemon.configPath)
+	daemon.SetConfigPath("/etc/hive/config.toml")
+	assert.Equal(t, "/etc/hive/config.toml", daemon.configPath)
+}

hive/daemon/daemon_test.go

@@ -20,42 +20,12 @@ import (
 	"github.com/mulgadc/hive/hive/objectstore"
 	"github.com/mulgadc/hive/hive/qmp"
 	"github.com/mulgadc/hive/hive/vm"
-	"github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
 	"github.com/pelletier/go-toml/v2"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
-// startTestNATSServer starts an embedded NATS server for testing
-// Using port -1 allows NATS to automatically allocate an available port
-func startTestNATSServer(t *testing.T) (*server.Server, string) {
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1, // Let NATS auto-allocate an available port
-		JetStream: false,
-		NoLog:     true,
-		NoSigs:    true,
-	}
-
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err, "Failed to create NATS server")
-
-	// Start server in a goroutine
-	go ns.Start()
-
-	// Wait for server to be ready
-	if !ns.ReadyForConnections(5 * time.Second) {
-		t.Fatal("NATS server failed to start")
-	}
-
-	// Get the actual URL that was assigned
-	url := ns.ClientURL()
-	t.Logf("Test NATS server started at: %s", url)
-
-	return ns, url
-}
-
 // createTestDaemon creates a test daemon instance with minimal configuration
 func createTestDaemon(t *testing.T, natsURL string) *Daemon {
 	// Create a temporary directory for test data
@@ -102,6 +72,7 @@ func createTestDaemon(t *testing.T, natsURL string) *Daemon {
 	require.NoError(t, err, "Failed to connect to NATS")
 
 	daemon.natsConn = nc
+	daemon.detachDelay = 0 // Skip sleep in tests
 
 	// Initialize services (needed for handler tests)
 	daemon.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(cfg, daemon.resourceMgr.instanceTypes, nc, &daemon.Instances, objectstore.NewMemoryObjectStore())
@@ -236,8 +207,7 @@ func TestHandleEC2RunInstances_MessageParsing(t *testing.T) {
 			}
 
 			// Start test NATS server
-			ns, natsURL := startTestNATSServer(t)
-			defer ns.Shutdown()
+			natsURL := sharedNATSURL
 
 			// Create test daemon
 			daemon := createTestDaemon(t, natsURL)
@@ -318,8 +288,7 @@ func TestHandleEC2RunInstances_ResourceManagement(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			ns, natsURL := startTestNATSServer(t)
-			defer ns.Shutdown()
+			natsURL := sharedNATSURL
 
 			daemon := createTestDaemon(t, natsURL)
 
@@ -609,8 +578,7 @@ func TestGetAvailableInstanceTypeInfos_ResourceFiltering(t *testing.T) {
 
 // TestHandleEC2DescribeInstanceTypes tests the DescribeInstanceTypes handler
 func TestHandleEC2DescribeInstanceTypes(t *testing.T) {
-	ns, natsURL := startTestNATSServer(t)
-	defer ns.Shutdown()
+	natsURL := sharedNATSURL
 
 	daemon := createTestDaemon(t, natsURL)
 
@@ -916,27 +884,7 @@ func TestHandleEC2DescribeInstanceTypes(t *testing.T) {
 
 // TestDaemon_BootAllocation verifies that resources are correctly reconstructed on startup
 func TestDaemon_BootAllocation(t *testing.T) {
-	// Start NATS server with JetStream
-	jsTmpDir, err := os.MkdirTemp("", "nats-js-boot-test-*")
-	require.NoError(t, err)
-	defer os.RemoveAll(jsTmpDir)
-
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: true,
-		StoreDir:  jsTmpDir,
-		NoLog:     true,
-		NoSigs:    true,
-	}
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err)
-	go ns.Start()
-	if !ns.ReadyForConnections(5 * time.Second) {
-		t.Fatal("NATS server with JetStream failed to start")
-	}
-	defer ns.Shutdown()
-	natsURL := ns.ClientURL()
+	natsURL := sharedJSNATSURL
 
 	// Create daemon temp directory
 	tmpDir, err := os.MkdirTemp("", "hive-daemon-boot-test-*")
@@ -1193,8 +1141,7 @@ func TestCanAllocate_CountEdgeCases(t *testing.T) {
 
 // TestDescribeInstances_ReservationGrouping tests that instances are grouped by reservation ID
 func TestDescribeInstances_ReservationGrouping(t *testing.T) {
-	ns, natsURL := startTestNATSServer(t)
-	defer ns.Shutdown()
+	natsURL := sharedNATSURL
 
 	daemon := createTestDaemon(t, natsURL)
 
@@ -1382,8 +1329,7 @@ func TestDescribeInstances_ReservationGrouping(t *testing.T) {
 
 // TestRunInstances_CountValidation tests MinCount/MaxCount validation scenarios
 func TestRunInstances_CountValidation(t *testing.T) {
-	ns, natsURL := startTestNATSServer(t)
-	defer ns.Shutdown()
+	natsURL := sharedNATSURL
 
 	daemon := createTestDaemon(t, natsURL)
 
@@ -1621,8 +1567,7 @@ func TestGenerateVolumes_DeleteOnTermination_FromBlockDeviceMapping(t *testing.T
 // correctly handles DeleteOnTermination for each volume type.
 // Uses embedded NATS with mock subscribers to verify NATS messages.
 func TestStopInstance_DeleteOnTermination_VolumeDeletion(t *testing.T) {
-	ns, natsURL := startTestNATSServer(t)
-	defer ns.Shutdown()
+	natsURL := sharedNATSURL
 
 	daemon := createTestDaemon(t, natsURL)
 
@@ -1720,8 +1665,7 @@ func TestStopInstance_DeleteOnTermination_VolumeDeletion(t *testing.T) {
 // TestStopInstance_DeleteOnTermination_False_SkipsVolumeDeletion verifies that
 // volumes with DeleteOnTermination=false are NOT deleted during termination.
 func TestStopInstance_DeleteOnTermination_False_SkipsVolumeDeletion(t *testing.T) {
-	ns, natsURL := startTestNATSServer(t)
-	defer ns.Shutdown()
+	natsURL := sharedNATSURL
 
 	daemon := createTestDaemon(t, natsURL)
 
@@ -1819,8 +1763,7 @@ func TestStopInstance_DeleteOnTermination_False_SkipsVolumeDeletion(t *testing.T
 // TestStopInstance_NoDelete_OnStop verifies that volumes are NOT deleted during
 // a regular stop (non-termination), regardless of DeleteOnTermination flag.
 func TestStopInstance_NoDelete_OnStop(t *testing.T) {
-	ns, natsURL := startTestNATSServer(t)
-	defer ns.Shutdown()
+	natsURL := sharedNATSURL
 
 	daemon := createTestDaemon(t, natsURL)
 
@@ -1976,8 +1919,7 @@ func TestNextAvailableDevice(t *testing.T) {
 
 // TestHandleEC2Events_AttachVolume tests the attach-volume handler in handleEC2Events
 func TestHandleEC2Events_AttachVolume(t *testing.T) {
-	ns, natsURL := startTestNATSServer(t)
-	defer ns.Shutdown()
+	natsURL := sharedNATSURL
 
 	daemon := createTestDaemon(t, natsURL)
 
@@ -2106,8 +2048,7 @@ func TestEBSRequest_JSON_Serialization(t *testing.T) {
 
 // TestHandleEC2Events_DetachVolume tests the detach-volume handler in handleEC2Events
 func TestHandleEC2Events_DetachVolume(t *testing.T) {
-	ns, natsURL := startTestNATSServer(t)
-	defer ns.Shutdown()
+	natsURL := sharedNATSURL
 
 	daemon := createTestDaemon(t, natsURL)
 
@@ -2433,8 +2374,7 @@ func newMockQMPClient(t *testing.T, responseFunc func(cmd qmp.QMPCommand) map[st
 // TestDetachVolume_SuccessPath tests the full happy-path detach including QMP commands
 // and state cleanup using a mock QMP server.
 func TestDetachVolume_SuccessPath(t *testing.T) {
-	ns, natsURL := startTestNATSServer(t)
-	defer ns.Shutdown()
+	natsURL := sharedNATSURL
 
 	daemon := createTestDaemon(t, natsURL)
 
@@ -2575,8 +2515,7 @@ func TestDetachVolume_SuccessPath(t *testing.T) {
 
 // TestDetachVolume_ForceFlag tests that force=true continues past device_del failure
 func TestDetachVolume_ForceFlag(t *testing.T) {
-	ns, natsURL := startTestNATSServer(t)
-	defer ns.Shutdown()
+	natsURL := sharedNATSURL
 
 	daemon := createTestDaemon(t, natsURL)
 
@@ -2692,8 +2631,7 @@ func TestDetachVolume_ForceFlag(t *testing.T) {
 // TestDetachVolume_BlockdevDelFailure tests that when blockdev-del fails,
 // state is left intact to prevent double-attach and VM crashes.
 func TestDetachVolume_BlockdevDelFailure(t *testing.T) {
-	ns, natsURL := startTestNATSServer(t)
-	defer ns.Shutdown()
+	natsURL := sharedNATSURL
 
 	daemon := createTestDaemon(t, natsURL)
 
@@ -2805,8 +2743,7 @@ func TestDetachVolume_BlockdevDelFailure(t *testing.T) {
 
 // TestDetachVolume_SuccessWithDeviceMatch tests detach with correct --device cross-check
 func TestDetachVolume_SuccessWithDeviceMatch(t *testing.T) {
-	ns, natsURL := startTestNATSServer(t)
-	defer ns.Shutdown()
+	natsURL := sharedNATSURL
 
 	daemon := createTestDaemon(t, natsURL)
 
@@ -2890,8 +2827,7 @@ func TestDetachVolume_SuccessWithDeviceMatch(t *testing.T) {
 // that already has a stale EBSRequest entry (e.g. from a stop/start cycle)
 // replaces it rather than creating a duplicate.
 func TestAttachVolume_ReplacesStaleEBSRequest(t *testing.T) {
-	ns, natsURL := startTestNATSServer(t)
-	defer ns.Shutdown()
+	natsURL := sharedNATSURL
 
 	daemon := createTestDaemon(t, natsURL)
 
@@ -3334,3 +3270,155 @@ func TestNewDaemon_WalDirPreservedIfSet(t *testing.T) {
 	d := NewDaemon(cfg)
 	assert.Equal(t, "/fast-ssd/wal", d.config.WalDir)
 }
+
+// TestMarkInstanceFailed verifies that markInstanceFailed sets the StateReason and
+// transitions the instance to StateShuttingDown.
+func TestMarkInstanceFailed(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createTestDaemon(t, natsURL)
+
+	instanceID := "i-test-mark-failed"
+	ec2Instance := &ec2.Instance{}
+	ec2Instance.SetInstanceId(instanceID)
+
+	instance := &vm.VM{
+		ID:           instanceID,
+		InstanceType: getTestInstanceType(),
+		Status:       vm.StatePending,
+		Instance:     ec2Instance,
+		QMPClient:    &qmp.QMPClient{},
+	}
+	daemon.Instances.VMS[instanceID] = instance
+
+	daemon.markInstanceFailed(instance, "volume_preparation_failed")
+
+	daemon.Instances.Mu.Lock()
+	defer daemon.Instances.Mu.Unlock()
+
+	// Verify state transitioned to shutting-down
+	assert.Equal(t, vm.StateShuttingDown, instance.Status)
+
+	// Verify StateReason was set
+	require.NotNil(t, instance.Instance.StateReason)
+	assert.Equal(t, "Server.InternalError", *instance.Instance.StateReason.Code)
+	assert.Equal(t, "volume_preparation_failed", *instance.Instance.StateReason.Message)
+}
+
+// TestMarkInstanceFailed_NilInstance verifies that markInstanceFailed handles
+// a VM with no ec2.Instance (Instance == nil) gracefully.
+func TestMarkInstanceFailed_NilInstance(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createTestDaemon(t, natsURL)
+
+	instanceID := "i-test-mark-failed-nil"
+	instance := &vm.VM{
+		ID:           instanceID,
+		InstanceType: getTestInstanceType(),
+		Status:       vm.StatePending,
+		Instance:     nil, // no ec2.Instance
+		QMPClient:    &qmp.QMPClient{},
+	}
+	daemon.Instances.VMS[instanceID] = instance
+
+	// Should not panic
+	daemon.markInstanceFailed(instance, "test_failure")
+
+	daemon.Instances.Mu.Lock()
+	defer daemon.Instances.Mu.Unlock()
+	assert.Equal(t, vm.StateShuttingDown, instance.Status)
+}
+
+// TestRollbackEBSMount_Success verifies that rollbackEBSMount sends an ebs.unmount
+// request and handles a successful unmount response.
+func TestRollbackEBSMount_Success(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createTestDaemon(t, natsURL)
+
+	unmountCalled := make(chan string, 1)
+
+	// Mock ebs.unmount subscriber that returns success
+	sub, err := daemon.natsConn.Subscribe("ebs.unmount", func(msg *nats.Msg) {
+		var req config.EBSRequest
+		json.Unmarshal(msg.Data, &req)
+		unmountCalled <- req.Name
+		resp := config.EBSUnMountResponse{Volume: req.Name, Mounted: false}
+		data, _ := json.Marshal(resp)
+		msg.Respond(data)
+	})
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	ebsReq := config.EBSRequest{
+		Name:       "vol-rollback-test",
+		DeviceName: "/dev/sdf",
+	}
+
+	daemon.rollbackEBSMount(ebsReq)
+
+	select {
+	case volName := <-unmountCalled:
+		assert.Equal(t, "vol-rollback-test", volName)
+	case <-time.After(2 * time.Second):
+		t.Fatal("ebs.unmount was not called")
+	}
+}
+
+// TestRollbackEBSMount_UnmountError verifies that rollbackEBSMount handles
+// an error response from ebs.unmount gracefully (no panic, just logs).
+func TestRollbackEBSMount_UnmountError(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createTestDaemon(t, natsURL)
+
+	// Mock ebs.unmount subscriber that returns an error
+	sub, err := daemon.natsConn.Subscribe("ebs.unmount", func(msg *nats.Msg) {
+		resp := config.EBSUnMountResponse{Error: "unmount failed: device busy"}
+		data, _ := json.Marshal(resp)
+		msg.Respond(data)
+	})
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	ebsReq := config.EBSRequest{Name: "vol-rollback-err"}
+
+	// Should not panic — errors are logged but not propagated
+	daemon.rollbackEBSMount(ebsReq)
+}
+
+// TestRollbackEBSMount_StillMounted verifies that rollbackEBSMount handles
+// the case where the unmount response says the volume is still mounted.
+func TestRollbackEBSMount_StillMounted(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createTestDaemon(t, natsURL)
+
+	sub, err := daemon.natsConn.Subscribe("ebs.unmount", func(msg *nats.Msg) {
+		resp := config.EBSUnMountResponse{Mounted: true} // still mounted
+		data, _ := json.Marshal(resp)
+		msg.Respond(data)
+	})
+	require.NoError(t, err)
+	defer sub.Unsubscribe()
+
+	ebsReq := config.EBSRequest{Name: "vol-still-mounted"}
+
+	// Should not panic
+	daemon.rollbackEBSMount(ebsReq)
+}
+
+// TestRollbackEBSMount_NATSTimeout verifies that rollbackEBSMount handles
+// NATS request timeout gracefully (no subscriber on ebs.unmount).
+func TestRollbackEBSMount_NATSTimeout(t *testing.T) {
+	natsURL := sharedNATSURL
+
+	daemon := createTestDaemon(t, natsURL)
+
+	// No ebs.unmount subscriber — will timeout
+	ebsReq := config.EBSRequest{Name: "vol-timeout"}
+
+	// Should not panic, just log the timeout
+	daemon.rollbackEBSMount(ebsReq)
+}

hive/daemon/jetstream_test.go

@@ -1,58 +1,17 @@
 package daemon
 
 import (
-	"os"
 	"testing"
-	"time"
 
 	"github.com/mulgadc/hive/hive/vm"
-	"github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
-// startTestNATSServerWithJetStream starts an embedded NATS server with JetStream enabled for testing
-func startTestNATSServerWithJetStream(t *testing.T) (*server.Server, string) {
-	// Create a temporary directory for JetStream storage
-	tmpDir, err := os.MkdirTemp("", "nats-js-test-*")
-	require.NoError(t, err, "Failed to create temp directory for JetStream")
-
-	t.Cleanup(func() {
-		os.RemoveAll(tmpDir)
-	})
-
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1, // Let NATS auto-allocate an available port
-		JetStream: true,
-		StoreDir:  tmpDir,
-		NoLog:     true,
-		NoSigs:    true,
-	}
-
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err, "Failed to create NATS server with JetStream")
-
-	// Start server in a goroutine
-	go ns.Start()
-
-	// Wait for server to be ready
-	if !ns.ReadyForConnections(5 * time.Second) {
-		t.Fatal("NATS server with JetStream failed to start")
-	}
-
-	// Get the actual URL that was assigned
-	url := ns.ClientURL()
-	t.Logf("Test NATS server with JetStream started at: %s", url)
-
-	return ns, url
-}
-
 // TestJetStreamManager_WriteAndLoadState tests round-trip write and load of instance state
 func TestJetStreamManager_WriteAndLoadState(t *testing.T) {
-	ns, natsURL := startTestNATSServerWithJetStream(t)
-	defer ns.Shutdown()
+	natsURL := sharedJSNATSURL
 
 	// Connect to NATS
 	nc, err := nats.Connect(natsURL)
@@ -105,8 +64,7 @@ func TestJetStreamManager_WriteAndLoadState(t *testing.T) {
 
 // TestJetStreamManager_LoadState_KeyNotFound tests that LoadState returns empty state when key doesn't exist
 func TestJetStreamManager_LoadState_KeyNotFound(t *testing.T) {
-	ns, natsURL := startTestNATSServerWithJetStream(t)
-	defer ns.Shutdown()
+	natsURL := sharedJSNATSURL
 
 	nc, err := nats.Connect(natsURL)
 	require.NoError(t, err)
@@ -127,8 +85,7 @@ func TestJetStreamManager_LoadState_KeyNotFound(t *testing.T) {
 
 // TestJetStreamManager_BucketCreation tests that InitKVBucket creates the bucket when it doesn't exist
 func TestJetStreamManager_BucketCreation(t *testing.T) {
-	ns, natsURL := startTestNATSServerWithJetStream(t)
-	defer ns.Shutdown()
+	natsURL := sharedJSNATSURL
 
 	nc, err := nats.Connect(natsURL)
 	require.NoError(t, err)
@@ -147,8 +104,7 @@ func TestJetStreamManager_BucketCreation(t *testing.T) {
 
 // TestJetStreamManager_BucketReconnection tests that InitKVBucket connects to existing bucket
 func TestJetStreamManager_BucketReconnection(t *testing.T) {
-	ns, natsURL := startTestNATSServerWithJetStream(t)
-	defer ns.Shutdown()
+	natsURL := sharedJSNATSURL
 
 	// First connection - create the bucket
 	nc1, err := nats.Connect(natsURL)
@@ -195,8 +151,7 @@ func TestJetStreamManager_BucketReconnection(t *testing.T) {
 
 // TestJetStreamManager_DeleteState tests deleting state from the KV store
 func TestJetStreamManager_DeleteState(t *testing.T) {
-	ns, natsURL := startTestNATSServerWithJetStream(t)
-	defer ns.Shutdown()
+	natsURL := sharedJSNATSURL
 
 	nc, err := nats.Connect(natsURL)
 	require.NoError(t, err)
@@ -238,8 +193,7 @@ func TestJetStreamManager_DeleteState(t *testing.T) {
 
 // TestJetStreamManager_DeleteState_NonExistent tests deleting state that doesn't exist
 func TestJetStreamManager_DeleteState_NonExistent(t *testing.T) {
-	ns, natsURL := startTestNATSServerWithJetStream(t)
-	defer ns.Shutdown()
+	natsURL := sharedJSNATSURL
 
 	nc, err := nats.Connect(natsURL)
 	require.NoError(t, err)
@@ -258,8 +212,7 @@ func TestJetStreamManager_DeleteState_NonExistent(t *testing.T) {
 
 // TestJetStreamManager_WriteState_UpdateExisting tests that writing state updates existing entry
 func TestJetStreamManager_WriteState_UpdateExisting(t *testing.T) {
-	ns, natsURL := startTestNATSServerWithJetStream(t)
-	defer ns.Shutdown()
+	natsURL := sharedJSNATSURL
 
 	nc, err := nats.Connect(natsURL)
 	require.NoError(t, err)
@@ -311,8 +264,7 @@ func TestJetStreamManager_WriteState_UpdateExisting(t *testing.T) {
 
 // TestJetStreamManager_MultipleNodes tests storing state for multiple nodes
 func TestJetStreamManager_MultipleNodes(t *testing.T) {
-	ns, natsURL := startTestNATSServerWithJetStream(t)
-	defer ns.Shutdown()
+	natsURL := sharedJSNATSURL
 
 	nc, err := nats.Connect(natsURL)
 	require.NoError(t, err)
@@ -363,8 +315,7 @@ func TestJetStreamManager_MultipleNodes(t *testing.T) {
 
 // TestJetStreamManager_KVNotInitialized tests error handling when KV is not initialized
 func TestJetStreamManager_KVNotInitialized(t *testing.T) {
-	ns, natsURL := startTestNATSServerWithJetStream(t)
-	defer ns.Shutdown()
+	natsURL := sharedJSNATSURL
 
 	nc, err := nats.Connect(natsURL)
 	require.NoError(t, err)
@@ -387,8 +338,7 @@ func TestJetStreamManager_KVNotInitialized(t *testing.T) {
 
 // TestJetStreamManager_UpdateReplicas tests updating replica count for the KV bucket
 func TestJetStreamManager_UpdateReplicas(t *testing.T) {
-	ns, natsURL := startTestNATSServerWithJetStream(t)
-	defer ns.Shutdown()
+	natsURL := sharedJSNATSURL
 
 	nc, err := nats.Connect(natsURL)
 	require.NoError(t, err)

hive/daemon/main_test.go

@@ -0,0 +1,73 @@
+package daemon
+
+import (
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats-server/v2/server"
+)
+
+// Shared NATS servers for all daemon tests — started once in TestMain.
+var (
+	sharedNATSServer   *server.Server
+	sharedNATSURL      string
+	sharedJSServer     *server.Server
+	sharedJSNATSURL    string
+)
+
+func TestMain(m *testing.M) {
+	// Start a plain NATS server (no JetStream)
+	ns, err := server.NewServer(&server.Options{
+		Host:   "127.0.0.1",
+		Port:   -1,
+		NoLog:  true,
+		NoSigs: true,
+	})
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Failed to create shared NATS server: %v\n", err)
+		os.Exit(1)
+	}
+	go ns.Start()
+	if !ns.ReadyForConnections(5 * time.Second) {
+		fmt.Fprintln(os.Stderr, "Shared NATS server failed to start")
+		os.Exit(1)
+	}
+	sharedNATSServer = ns
+	sharedNATSURL = ns.ClientURL()
+
+	// Start a JetStream-enabled NATS server
+	jsTmpDir, err := os.MkdirTemp("", "nats-js-shared-*")
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Failed to create JetStream temp dir: %v\n", err)
+		os.Exit(1)
+	}
+	jsNS, err := server.NewServer(&server.Options{
+		Host:      "127.0.0.1",
+		Port:      -1,
+		JetStream: true,
+		StoreDir:  jsTmpDir,
+		NoLog:     true,
+		NoSigs:    true,
+	})
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Failed to create shared JetStream NATS server: %v\n", err)
+		os.Exit(1)
+	}
+	go jsNS.Start()
+	if !jsNS.ReadyForConnections(5 * time.Second) {
+		fmt.Fprintln(os.Stderr, "Shared JetStream NATS server failed to start")
+		os.Exit(1)
+	}
+	sharedJSServer = jsNS
+	sharedJSNATSURL = jsNS.ClientURL()
+
+	code := m.Run()
+
+	ns.Shutdown()
+	jsNS.Shutdown()
+	os.RemoveAll(jsTmpDir)
+
+	os.Exit(code)
+}