feat: extract helpers (#109)

scripts/check-coverage.sh

@@ -5,7 +5,7 @@ set -euo pipefail
 # Usage: scripts/check-coverage.sh <coverprofile> [quiet]
 
 PROFILE="${1:?Usage: check-coverage.sh <coverprofile> [quiet]}"
-MIN=60.0
+MIN=65.0
 QUIET="${2:-}"
 
 if [[ ! -s "$PROFILE" ]]; then

scripts/clone-deps.sh

@@ -3,7 +3,6 @@
 # Spinifex Development Dependencies Setup Script
 # This script clones Viperblock and Predastore repositories for cross-repo development
 
-# TODO: Improve, remove CLAUDE readme reference, simplify next-steps
 set -e
 
 SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"

scripts/coverage.sh

@@ -1,20 +0,0 @@
-#!/usr/bin/env bash
-set -euo pipefail
-
-# Run Go tests with coverage and report total code coverage.
-
-COVERPROFILE=$(mktemp /tmp/spinifex-coverage-XXXXXX.out)
-trap 'rm -f "$COVERPROFILE"' EXIT
-
-echo "Running tests with coverage..."
-echo ""
-LOG_IGNORE=1 go test -timeout 120s -coverprofile="$COVERPROFILE" -covermode=atomic ./spinifex/... 2>&1
-
-if [[ ! -s "$COVERPROFILE" ]]; then
-    echo "No coverage data collected."
-    exit 1
-fi
-
-echo ""
-echo "=== Total ==="
-go tool cover -func="$COVERPROFILE" | awk '/^total:/ { print $NF }'

scripts/diff-coverage.sh

@@ -12,7 +12,7 @@ set -euo pipefail
 
 PROFILE=""
 BASE_REF=""
-THRESHOLD=60
+THRESHOLD=65
 QUIET="${QUIET:-}"
 
 while [[ $# -gt 0 ]]; do

scripts/start-multi-nats.sh

@@ -1,3 +0,0 @@
-#!/bin/sh
-
-SPINIFEX_CONFIG_PATH=~/node1/config/nats/nats.conf ./bin/spx service nats start

spinifex/daemon/daemon.go

@@ -2215,74 +2215,18 @@ func (d *Daemon) StartInstance(instance *vm.VM) error {
 	consoleLogPath := filepath.Join(runtimeDir, fmt.Sprintf("console-%s.log", instance.ID))
 	serialSocket := filepath.Join(runtimeDir, fmt.Sprintf("serial-%s.sock", instance.ID))
 
-	instance.Config = vm.Config{
-		Name:           instance.ID,
-		PIDFile:        pidFile,
-		EnableKVM:      true, // If available, if kvm fails, will use cpu max
-		NoGraphic:      true,
-		MachineType:    "q35",
-		ConsoleLogPath: consoleLogPath,
-		SerialSocket:   serialSocket,
-		CPUType:        "host", // If available, if kvm fails, will use cpu max
-		Memory:         int(memoryMiB),
-		CPUCount:       vCPUs,
-		Architecture:   architecture,
-	}
+	instance.Config = buildBaseVMConfig(instance.ID, pidFile, consoleLogPath, serialSocket, architecture, vCPUs, int(memoryMiB))
 
-	// Add PCIe root ports for volume hotplug (Q35 requires explicit root ports).
-	// 11 ports for /dev/sd[f-p] hotplug slots, starting at chassis 1.
-	for i := 1; i <= 11; i++ {
-		instance.Config.Devices = append(instance.Config.Devices, vm.Device{
-			Value: fmt.Sprintf("pcie-root-port,id=hotplug%d,chassis=%d,slot=0", i, i),
-		})
-	}
-
-	// Loop through each volume in volumes
+	// Build QEMU drives from EBS volume requests.
 	instance.EBSRequests.Mu.Lock()
-
-	for _, v := range instance.EBSRequests.Requests {
-		drive := vm.Drive{}
-
-		// Use the NBDURI from mount response - contains socket path or TCP address
-		// NBDURI format: "nbd:unix:/path/to/socket.sock" or "nbd://host:port"
-		if v.NBDURI == "" {
-			slog.Error("NBDURI not set for volume", "volume", v.Name)
-			return fmt.Errorf("NBDURI not set for volume %s - was volume mounted?", v.Name)
-		}
-		drive.File = v.NBDURI
-		slog.Info("Using NBD URI for drive", "volume", v.Name, "uri", v.NBDURI)
-
-		if v.Boot {
-			drive.Format = "raw"
-			drive.If = "none"
-			drive.Media = "disk"
-			drive.ID = "os"
-			drive.Cache = "none"
-
-			iothreadID := "ioth-os"
-			instance.Config.IOThreads = append(instance.Config.IOThreads, vm.IOThread{ID: iothreadID})
-
-			instance.Config.Devices = append(instance.Config.Devices, vm.Device{
-				Value: fmt.Sprintf("virtio-blk-pci,drive=%s,iothread=%s,num-queues=%d,bootindex=1",
-					drive.ID, iothreadID, instance.Config.CPUCount),
-			})
-		}
-
-		if v.CloudInit {
-			drive.Format = "raw"
-			drive.If = "virtio"
-			drive.Media = "cdrom"
-			drive.ID = "cloudinit"
-		}
-
-		// TODO: Add EFI support
-		if v.EFI {
-			continue
-		}
-
-		instance.Config.Drives = append(instance.Config.Drives, drive)
-	}
+	drives, iothreads, devices, err := buildDrives(instance.EBSRequests.Requests, vCPUs)
 	instance.EBSRequests.Mu.Unlock()
+	if err != nil {
+		return err
+	}
+	instance.Config.Drives = append(instance.Config.Drives, drives...)
+	instance.Config.IOThreads = append(instance.Config.IOThreads, iothreads...)
+	instance.Config.Devices = append(instance.Config.Devices, devices...)
 
 	// VPC tap networking vs user-mode fallback
 	if instance.ENIId != "" && d.networkPlumber != nil {
@@ -2523,6 +2467,83 @@ func (d *Daemon) StartInstance(instance *vm.VM) error {
 	return nil
 }
 
+// buildBaseVMConfig creates a vm.Config with base QEMU settings and PCIe
+// hotplug root ports. Architecture, vCPU, and memory come from the caller
+// (resolved from instance type info).
+func buildBaseVMConfig(instanceID, pidFile, consoleLogPath, serialSocket, architecture string, vCPUs, memoryMiB int) vm.Config {
+	cfg := vm.Config{
+		Name:           instanceID,
+		PIDFile:        pidFile,
+		EnableKVM:      true,
+		NoGraphic:      true,
+		MachineType:    "q35",
+		ConsoleLogPath: consoleLogPath,
+		SerialSocket:   serialSocket,
+		CPUType:        "host",
+		Memory:         memoryMiB,
+		CPUCount:       vCPUs,
+		Architecture:   architecture,
+	}
+
+	// Add PCIe root ports for volume hotplug (Q35 requires explicit root ports).
+	// 11 ports for /dev/sd[f-p] hotplug slots, starting at chassis 1.
+	for i := 1; i <= 11; i++ {
+		cfg.Devices = append(cfg.Devices, vm.Device{
+			Value: fmt.Sprintf("pcie-root-port,id=hotplug%d,chassis=%d,slot=0", i, i),
+		})
+	}
+
+	return cfg
+}
+
+// buildDrives converts EBS volume requests into QEMU drive, iothread, and device
+// configurations. Returns an error if any non-EFI volume is missing its NBDURI.
+func buildDrives(requests []types.EBSRequest, cpuCount int) ([]vm.Drive, []vm.IOThread, []vm.Device, error) {
+	var drives []vm.Drive
+	var iothreads []vm.IOThread
+	var devices []vm.Device
+
+	for _, v := range requests {
+		// TODO: Add EFI support
+		if v.EFI {
+			continue
+		}
+
+		if v.NBDURI == "" {
+			return nil, nil, nil, fmt.Errorf("NBDURI not set for volume %s - was volume mounted?", v.Name)
+		}
+
+		drive := vm.Drive{File: v.NBDURI}
+
+		if v.Boot {
+			drive.Format = "raw"
+			drive.If = "none"
+			drive.Media = "disk"
+			drive.ID = "os"
+			drive.Cache = "none"
+
+			iothreadID := "ioth-os"
+			iothreads = append(iothreads, vm.IOThread{ID: iothreadID})
+			devices = append(devices, vm.Device{
+				Value: fmt.Sprintf("virtio-blk-pci,drive=%s,iothread=%s,num-queues=%d,bootindex=1",
+					drive.ID, iothreadID, cpuCount),
+			})
+		}
+
+		if v.CloudInit {
+			drive.Format = "raw"
+			drive.If = "virtio"
+			drive.Media = "cdrom"
+			drive.ID = "cloudinit"
+		}
+
+		slog.Info("Using NBD URI for drive", "volume", v.Name, "uri", v.NBDURI)
+		drives = append(drives, drive)
+	}
+
+	return drives, iothreads, devices, nil
+}
+
 // ebsTopic returns a node-specific EBS NATS topic, e.g. "ebs.node1.mount".
 // This ensures mount/unmount requests are routed to the viperblock instance
 // running on the same node as the daemon (NBD sockets are local).

spinifex/daemon/daemon_test.go

@@ -3838,3 +3838,191 @@ func TestDaemonReadyFlag(t *testing.T) {
 	daemon.ready.Store(true)
 	assert.True(t, daemon.ready.Load(), "daemon should be ready after setting flag")
 }
+
+func TestBuildBaseVMConfig(t *testing.T) {
+	tests := []struct {
+		name         string
+		instanceID   string
+		pidFile      string
+		consolePath  string
+		serialSocket string
+		architecture string
+		vCPUs        int
+		memoryMiB    int
+	}{
+		{
+			name:         "x86_64 instance",
+			instanceID:   "i-abc123",
+			pidFile:      "/tmp/qemu-i-abc123.pid",
+			consolePath:  "/run/console-i-abc123.log",
+			serialSocket: "/run/serial-i-abc123.sock",
+			architecture: "x86_64",
+			vCPUs:        4,
+			memoryMiB:    8192,
+		},
+		{
+			name:         "arm64 instance",
+			instanceID:   "i-def456",
+			pidFile:      "/tmp/qemu-i-def456.pid",
+			consolePath:  "/run/console-i-def456.log",
+			serialSocket: "/run/serial-i-def456.sock",
+			architecture: "arm64",
+			vCPUs:        2,
+			memoryMiB:    4096,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cfg := buildBaseVMConfig(tt.instanceID, tt.pidFile, tt.consolePath, tt.serialSocket, tt.architecture, tt.vCPUs, tt.memoryMiB)
+
+			assert.Equal(t, tt.instanceID, cfg.Name)
+			assert.Equal(t, tt.pidFile, cfg.PIDFile)
+			assert.Equal(t, tt.consolePath, cfg.ConsoleLogPath)
+			assert.Equal(t, tt.serialSocket, cfg.SerialSocket)
+			assert.Equal(t, tt.architecture, cfg.Architecture)
+			assert.Equal(t, tt.vCPUs, cfg.CPUCount)
+			assert.Equal(t, tt.memoryMiB, cfg.Memory)
+			assert.True(t, cfg.EnableKVM)
+			assert.True(t, cfg.NoGraphic)
+			assert.Equal(t, "q35", cfg.MachineType)
+			assert.Equal(t, "host", cfg.CPUType)
+
+			// 11 PCIe root ports for hotplug slots
+			require.Len(t, cfg.Devices, 11)
+			for i, dev := range cfg.Devices {
+				expected := fmt.Sprintf("pcie-root-port,id=hotplug%d,chassis=%d,slot=0", i+1, i+1)
+				assert.Equal(t, expected, dev.Value)
+			}
+		})
+	}
+}
+
+func TestBuildDrives(t *testing.T) {
+	tests := []struct {
+		name          string
+		requests      []types.EBSRequest
+		cpuCount      int
+		wantDrives    int
+		wantIOThreads int
+		wantDevices   int
+		wantErr       string
+	}{
+		{
+			name: "boot volume",
+			requests: []types.EBSRequest{
+				{Name: "vol-boot", NBDURI: "nbd:unix:/tmp/boot.sock", Boot: true},
+			},
+			cpuCount:      4,
+			wantDrives:    1,
+			wantIOThreads: 1,
+			wantDevices:   1,
+		},
+		{
+			name: "cloud-init volume",
+			requests: []types.EBSRequest{
+				{Name: "vol-ci", NBDURI: "nbd:unix:/tmp/ci.sock", CloudInit: true},
+			},
+			cpuCount:      2,
+			wantDrives:    1,
+			wantIOThreads: 0,
+			wantDevices:   0,
+		},
+		{
+			name: "EFI volume skipped",
+			requests: []types.EBSRequest{
+				{Name: "vol-efi", EFI: true},
+			},
+			cpuCount:      2,
+			wantDrives:    0,
+			wantIOThreads: 0,
+			wantDevices:   0,
+		},
+		{
+			name: "missing NBDURI returns error",
+			requests: []types.EBSRequest{
+				{Name: "vol-bad"},
+			},
+			cpuCount: 2,
+			wantErr:  "NBDURI not set for volume vol-bad",
+		},
+		{
+			name: "mixed boot + cloud-init + EFI",
+			requests: []types.EBSRequest{
+				{Name: "vol-boot", NBDURI: "nbd:unix:/tmp/boot.sock", Boot: true},
+				{Name: "vol-ci", NBDURI: "nbd:unix:/tmp/ci.sock", CloudInit: true},
+				{Name: "vol-efi", EFI: true},
+			},
+			cpuCount:      4,
+			wantDrives:    2, // boot + cloud-init (EFI skipped)
+			wantIOThreads: 1, // only boot gets iothread
+			wantDevices:   1, // only boot gets virtio-blk device
+		},
+		{
+			name:          "empty requests",
+			requests:      []types.EBSRequest{},
+			cpuCount:      2,
+			wantDrives:    0,
+			wantIOThreads: 0,
+			wantDevices:   0,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			drives, iothreads, devices, err := buildDrives(tt.requests, tt.cpuCount)
+
+			if tt.wantErr != "" {
+				require.Error(t, err)
+				assert.Contains(t, err.Error(), tt.wantErr)
+				return
+			}
+
+			require.NoError(t, err)
+			assert.Len(t, drives, tt.wantDrives)
+			assert.Len(t, iothreads, tt.wantIOThreads)
+			assert.Len(t, devices, tt.wantDevices)
+		})
+	}
+}
+
+func TestBuildDrives_BootVolume(t *testing.T) {
+	requests := []types.EBSRequest{
+		{Name: "vol-boot", NBDURI: "nbd:unix:/tmp/boot.sock", Boot: true},
+	}
+
+	drives, iothreads, devices, err := buildDrives(requests, 4)
+	require.NoError(t, err)
+
+	require.Len(t, drives, 1)
+	d := drives[0]
+	assert.Equal(t, "nbd:unix:/tmp/boot.sock", d.File)
+	assert.Equal(t, "raw", d.Format)
+	assert.Equal(t, "none", d.If)
+	assert.Equal(t, "disk", d.Media)
+	assert.Equal(t, "os", d.ID)
+	assert.Equal(t, "none", d.Cache)
+
+	require.Len(t, iothreads, 1)
+	assert.Equal(t, "ioth-os", iothreads[0].ID)
+
+	require.Len(t, devices, 1)
+	assert.Equal(t, "virtio-blk-pci,drive=os,iothread=ioth-os,num-queues=4,bootindex=1", devices[0].Value)
+}
+
+func TestBuildDrives_CloudInitVolume(t *testing.T) {
+	requests := []types.EBSRequest{
+		{Name: "vol-ci", NBDURI: "nbd:unix:/tmp/ci.sock", CloudInit: true},
+	}
+
+	drives, _, _, err := buildDrives(requests, 2)
+	require.NoError(t, err)
+
+	require.Len(t, drives, 1)
+	d := drives[0]
+	assert.Equal(t, "nbd:unix:/tmp/ci.sock", d.File)
+	assert.Equal(t, "raw", d.Format)
+	assert.Equal(t, "virtio", d.If)
+	assert.Equal(t, "cdrom", d.Media)
+	assert.Equal(t, "cloudinit", d.ID)
+}

spinifex/daemon/state_test.go

@@ -10,10 +10,9 @@ import (
 
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/spinifex/config"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/mulgadc/spinifex/spinifex/types"
 	"github.com/mulgadc/spinifex/spinifex/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"
 )
@@ -23,23 +22,7 @@ import (
 func createDaemonWithJetStream(t *testing.T) *Daemon {
 	t.Helper()
 
-	jsTmpDir := t.TempDir()
-
-	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()
-	require.True(t, ns.ReadyForConnections(5*time.Second), "NATS server failed to start")
-	t.Cleanup(func() { ns.Shutdown() })
-
-	natsURL := ns.ClientURL()
+	_, nc, _ := testutil.StartTestJetStream(t)
 
 	tmpDir := t.TempDir()
 
@@ -51,10 +34,6 @@ func createDaemonWithJetStream(t *testing.T) *Daemon {
 	require.NoError(t, err)
 	daemon.config = &config.Config{BaseDir: tmpDir}
 
-	nc, err := nats.Connect(natsURL)
-	require.NoError(t, err)
-	t.Cleanup(func() { nc.Close() })
-
 	daemon.natsConn = nc
 	daemon.jsManager, err = NewJetStreamManager(nc, 1)
 	require.NoError(t, err)

spinifex/gateway/ec2/instance/helpers_test.go

@@ -2,41 +2,14 @@ package gateway_ec2_instance
 
 import (
 	"testing"
-	"time"
 
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
-	"github.com/stretchr/testify/require"
 )
 
 // startTestNATSServer starts an embedded NATS server for testing
 func startTestNATSServer(t *testing.T) (*server.Server, *nats.Conn) {
 	t.Helper()
-
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: false,
-		NoLog:     true,
-		NoSigs:    true,
-	}
-
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err, "Failed to create NATS server")
-
-	go ns.Start()
-
-	if !ns.ReadyForConnections(5 * time.Second) {
-		t.Fatal("NATS server failed to start")
-	}
-
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err, "Failed to connect to NATS")
-
-	t.Cleanup(func() {
-		nc.Close()
-		ns.Shutdown()
-	})
-
-	return ns, nc
+	return testutil.StartTestNATS(t)
 }

spinifex/gateway/gateway_test.go

@@ -11,12 +11,11 @@ import (
 	"net/http/httptest"
 	"strings"
 	"testing"
-	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
 	handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/mulgadc/spinifex/spinifex/types"
-	"github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -264,25 +263,7 @@ func TestErrorHandler_ContentTypeXML(t *testing.T) {
 
 func startTestNATS(t *testing.T) *nats.Conn {
 	t.Helper()
-	ns, err := server.NewServer(&server.Options{
-		Host:   "127.0.0.1",
-		Port:   -1,
-		NoLog:  true,
-		NoSigs: true,
-	})
-	require.NoError(t, err)
-
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second), "NATS server failed to start")
-
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-
-	t.Cleanup(func() {
-		nc.Close()
-		ns.Shutdown()
-	})
-
+	_, nc := testutil.StartTestNATS(t)
 	return nc
 }
 

spinifex/gateway/spx/spx_test.go

@@ -3,8 +3,8 @@ package spx
 import (
 	"encoding/json"
 	"testing"
-	"time"
 
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/mulgadc/spinifex/spinifex/types"
 	"github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
@@ -52,23 +52,7 @@ func TestGetVersion_Dev(t *testing.T) {
 // startEmbeddedNATS starts an in-process NATS server for testing.
 func startEmbeddedNATS(t *testing.T) (*server.Server, *nats.Conn) {
 	t.Helper()
-	opts := &server.Options{
-		Host: "127.0.0.1",
-		Port: -1, // random port
-	}
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err)
-	go ns.Start()
-	if !ns.ReadyForConnections(2 * time.Second) {
-		t.Fatal("NATS server not ready")
-	}
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	t.Cleanup(func() {
-		nc.Close()
-		ns.Shutdown()
-	})
-	return ns, nc
+	return testutil.StartTestNATS(t)
 }
 
 func TestGetNodes_NoResponders(t *testing.T) {

spinifex/handlers/ec2/account/service_impl_test.go

@@ -2,12 +2,10 @@ package handlers_ec2_account
 
 import (
 	"testing"
-	"time"
 
 	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/mulgadc/spinifex/spinifex/utils"
-	"github.com/nats-io/nats-server/v2/server"
-	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -16,20 +14,7 @@ import (
 // the service against a NATS server without JetStream returns an error instead
 // of silently degrading.
 func TestNewAccountSettingsServiceImplWithNATS_NoJetStream(t *testing.T) {
-	ns, err := server.NewServer(&server.Options{
-		Host:   "127.0.0.1",
-		Port:   -1,
-		NoLog:  true,
-		NoSigs: true,
-	})
-	require.NoError(t, err)
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second))
-	t.Cleanup(func() { ns.Shutdown() })
-
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	t.Cleanup(func() { nc.Close() })
+	_, nc := testutil.StartTestNATS(t)
 
 	svc, err := NewAccountSettingsServiceImplWithNATS(nil, nc)
 	assert.Error(t, err)
@@ -40,23 +25,7 @@ var _ AccountSettingsService = (*AccountSettingsServiceImpl)(nil)
 
 func setupTestAccountService(t *testing.T) *AccountSettingsServiceImpl {
 	t.Helper()
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: true,
-		StoreDir:  t.TempDir(),
-		NoLog:     true,
-		NoSigs:    true,
-	}
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err)
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second))
-	t.Cleanup(func() { ns.Shutdown() })
-
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	t.Cleanup(func() { nc.Close() })
+	_, nc, _ := testutil.StartTestJetStream(t)
 
 	svc, err := NewAccountSettingsServiceImplWithNATS(nil, nc)
 	require.NoError(t, err)

spinifex/handlers/ec2/eigw/service_impl.go

@@ -240,13 +240,7 @@ func (s *EgressOnlyIGWServiceImpl) recordToEC2(record *EgressOnlyIGWRecord) *ec2
 		},
 	}
 
-	if len(record.Tags) > 0 {
-		tags := make([]*ec2.Tag, 0, len(record.Tags))
-		for k, v := range record.Tags {
-			tags = append(tags, &ec2.Tag{Key: aws.String(k), Value: aws.String(v)})
-		}
-		eigw.Tags = tags
-	}
+	eigw.Tags = utils.MapToEC2Tags(record.Tags)
 
 	return eigw
 }

spinifex/handlers/ec2/eigw/service_impl_test.go

@@ -2,14 +2,12 @@ package handlers_ec2_eigw
 
 import (
 	"testing"
-	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/mulgadc/spinifex/spinifex/utils"
-	"github.com/nats-io/nats-server/v2/server"
-	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -18,33 +16,14 @@ const testAccountID = "123456789012"
 
 func setupTestEIGWService(t *testing.T) *EgressOnlyIGWServiceImpl {
 	t.Helper()
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: true,
-		StoreDir:  t.TempDir(),
-		NoLog:     true,
-		NoSigs:    true,
-	}
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err)
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second))
-	t.Cleanup(func() { ns.Shutdown() })
-
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	t.Cleanup(func() { nc.Close() })
+	_, nc, js := testutil.StartTestJetStream(t)
 
 	// Create VPC KV bucket and register test VPCs so fail-closed ownership checks pass
-	js, err := nc.JetStream()
-	require.NoError(t, err)
-	vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1})
-	require.NoError(t, err)
+	vpcEntries := map[string][]byte{}
 	for _, vpcID := range []string{"vpc-test123", "vpc-tagged", "vpc-tagged2"} {
-		_, err = vpcKV.Put(utils.AccountKey(testAccountID, vpcID), []byte(`{"vpc_id":"`+vpcID+`","state":"available"}`))
-		require.NoError(t, err)
+		vpcEntries[utils.AccountKey(testAccountID, vpcID)] = []byte(`{"vpc_id":"` + vpcID + `","state":"available"}`)
 	}
+	testutil.SeedKV(t, js, handlers_ec2_vpc.KVBucketVPCs, vpcEntries)
 
 	svc, err := NewEgressOnlyIGWServiceImplWithNATS(nil, nc)
 	require.NoError(t, err)
@@ -293,32 +272,15 @@ func TestDescribeEgressOnlyInternetGateways_FilterByTag(t *testing.T) {
 // TestCreateEgressOnlyInternetGateway_CrossAccountVPCRejected tests that creating an EIGW in another account's VPC is rejected.
 func TestCreateEgressOnlyInternetGateway_CrossAccountVPCRejected(t *testing.T) {
 	// Set up with manual NATS to get VPC KV access
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: true,
-		StoreDir:  t.TempDir(),
-		NoLog:     true,
-		NoSigs:    true,
-	}
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err)
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second))
-	t.Cleanup(func() { ns.Shutdown() })
-
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	t.Cleanup(func() { nc.Close() })
+	_, nc, js := testutil.StartTestJetStream(t)
 
 	// Create VPC KV bucket with a VPC owned by testAccountID
-	js, err := nc.JetStream()
-	require.NoError(t, err)
-	vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1})
-	require.NoError(t, err)
-
 	vpcID := "vpc-alpha456"
-	_, err = vpcKV.Put(utils.AccountKey(testAccountID, vpcID), []byte(`{"vpc_id":"vpc-alpha456","state":"available"}`))
+	testutil.SeedKV(t, js, handlers_ec2_vpc.KVBucketVPCs, map[string][]byte{
+		utils.AccountKey(testAccountID, vpcID): []byte(`{"vpc_id":"vpc-alpha456","state":"available"}`),
+	})
+
+	var err error
 	require.NoError(t, err)
 
 	// Create EIGW service (VPC KV will be populated since we created it before EIGW init)

spinifex/handlers/ec2/eip/service_impl.go

@@ -569,26 +569,13 @@ func (s *EIPServiceImpl) findByAssociationID(accountID, associationID string) (*
 // publishNATEvent publishes a NAT lifecycle event to NATS for vpcd consumption.
 // This is fire-and-forget; errors are logged but do not fail the API response.
 func (s *EIPServiceImpl) publishNATEvent(topic, vpcID, externalIP, logicalIP, eniID, mac string) {
-	if s.natsConn == nil {
-		return
-	}
-
-	evt := natEvent{
+	utils.PublishEvent(s.natsConn, topic, natEvent{
 		VpcId:      vpcID,
 		ExternalIP: externalIP,
 		LogicalIP:  logicalIP,
 		PortName:   eniID,
 		MAC:        mac,
-	}
-
-	data, err := json.Marshal(evt)
-	if err != nil {
-		slog.Error("Failed to marshal NAT event", "topic", topic, "err", err)
-		return
-	}
-	if err := s.natsConn.Publish(topic, data); err != nil {
-		slog.Error("Failed to publish NAT event", "topic", topic, "err", err)
-	}
+	})
 }
 
 // eipRecordToEC2 converts an EIPRecord to an EC2 Address.
@@ -613,13 +600,7 @@ func (s *EIPServiceImpl) eipRecordToEC2(record *EIPRecord) *ec2.Address {
 		addr.PrivateIpAddress = aws.String(record.PrivateIp)
 	}
 
-	if len(record.Tags) > 0 {
-		tags := make([]*ec2.Tag, 0, len(record.Tags))
-		for k, v := range record.Tags {
-			tags = append(tags, &ec2.Tag{Key: aws.String(k), Value: aws.String(v)})
-		}
-		addr.Tags = tags
-	}
+	addr.Tags = utils.MapToEC2Tags(record.Tags)
 
 	return addr
 }

spinifex/handlers/ec2/eip/service_impl_test.go

@@ -3,13 +3,11 @@ package handlers_ec2_eip
 import (
 	"encoding/json"
 	"testing"
-	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
-	"github.com/nats-io/nats-server/v2/server"
-	"github.com/nats-io/nats.go"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -28,26 +26,7 @@ func testPool() handlers_ec2_vpc.ExternalPoolConfig {
 
 func setupTestEIP(t *testing.T) (*EIPServiceImpl, *handlers_ec2_vpc.ExternalIPAM) {
 	t.Helper()
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: true,
-		StoreDir:  t.TempDir(),
-		NoLog:     true,
-		NoSigs:    true,
-	}
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err)
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second))
-	t.Cleanup(func() { ns.Shutdown() })
-
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	t.Cleanup(func() { nc.Close() })
-
-	js, err := nc.JetStream()
-	require.NoError(t, err)
+	_, nc, js := testutil.StartTestJetStream(t)
 
 	pool := testPool()
 	ipam, err := handlers_ec2_vpc.NewExternalIPAM(js, []handlers_ec2_vpc.ExternalPoolConfig{pool})

spinifex/handlers/ec2/igw/service_impl.go

@@ -391,13 +391,7 @@ func (s *IGWServiceImpl) recordToEC2(record *IGWRecord) *ec2.InternetGateway {
 		}
 	}
 
-	if len(record.Tags) > 0 {
-		tags := make([]*ec2.Tag, 0, len(record.Tags))
-		for k, v := range record.Tags {
-			tags = append(tags, &ec2.Tag{Key: aws.String(k), Value: aws.String(v)})
-		}
-		igw.Tags = tags
-	}
+	igw.Tags = utils.MapToEC2Tags(record.Tags)
 
 	return igw
 }

spinifex/handlers/ec2/igw/service_impl_test.go

@@ -7,8 +7,8 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/mulgadc/spinifex/spinifex/utils"
-	"github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -16,33 +16,14 @@ import (
 
 func setupTestIGWService(t *testing.T) (*IGWServiceImpl, *nats.Conn) {
 	t.Helper()
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: true,
-		StoreDir:  t.TempDir(),
-		NoLog:     true,
-		NoSigs:    true,
-	}
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err)
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second))
-	t.Cleanup(func() { ns.Shutdown() })
-
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	t.Cleanup(func() { nc.Close() })
+	_, nc, js := testutil.StartTestJetStream(t)
 
 	// Create VPC KV bucket and register test VPCs so fail-closed ownership checks pass
-	js, err := nc.JetStream()
-	require.NoError(t, err)
-	vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1})
-	require.NoError(t, err)
+	vpcEntries := map[string][]byte{}
 	for _, vpcID := range []string{"vpc-test123", "vpc-other", "vpc-lifecycle", "vpc-event-test"} {
-		_, err = vpcKV.Put(utils.AccountKey(testAccountID, vpcID), []byte(`{"vpc_id":"`+vpcID+`","state":"available"}`))
-		require.NoError(t, err)
+		vpcEntries[utils.AccountKey(testAccountID, vpcID)] = []byte(`{"vpc_id":"` + vpcID + `","state":"available"}`)
 	}
+	testutil.SeedKV(t, js, handlers_ec2_vpc.KVBucketVPCs, vpcEntries)
 
 	svc, err := NewIGWServiceImplWithNATS(nil, nc)
 	require.NoError(t, err)

spinifex/handlers/ec2/instance/service_nats_test.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
@@ -18,29 +19,10 @@ import (
 
 // startTestNATSServer starts an embedded NATS server for testing
 func startTestNATSServer(t *testing.T) (*server.Server, string) {
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1, // Auto-allocate 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 goroutine
-	go ns.Start()
-
-	// Wait for server to be ready
-	if !ns.ReadyForConnections(5 * time.Second) {
-		t.Fatal("NATS server failed to start")
-	}
-
-	url := ns.ClientURL()
-	t.Logf("Test NATS server started at: %s", url)
-
-	return ns, url
+	t.Helper()
+	ns, nc := testutil.StartTestNATS(t)
+	_ = nc // connection managed by testutil cleanup
+	return ns, ns.ClientURL()
 }
 
 // createValidRunInstancesInput creates a valid RunInstancesInput for testing

spinifex/handlers/ec2/natgw/service_impl.go

@@ -287,20 +287,12 @@ func (s *NatGatewayServiceImpl) publishDeleteEventsForNatGateway(record *NatGate
 				continue
 			}
 
-			evt := natGatewayEvent{
+			utils.PublishEvent(s.natsConn, "vpc.delete-nat-gateway", natGatewayEvent{
 				VpcId:        record.VpcId,
 				NatGatewayId: record.NatGatewayId,
 				PublicIp:     record.PublicIp,
 				SubnetCidr:   subnet.CidrBlock,
-			}
-			eventData, err := json.Marshal(evt)
-			if err != nil {
-				slog.Warn("Failed to marshal NAT GW delete event", "err", err)
-				continue
-			}
-			if err := s.natsConn.Publish("vpc.delete-nat-gateway", eventData); err != nil {
-				slog.Warn("Failed to publish NAT GW delete event", "subnet", subnet.CidrBlock, "err", err)
-			}
+			})
 		}
 	}
 }
@@ -430,44 +422,22 @@ func natgwMatchesFilters(record *NatGatewayRecord, filters map[string][]string)
 // PublishAddEvent publishes a vpc.add-nat-gateway event for vpcd to create the SNAT rule.
 // Called by the route table service when CreateRoute targets a NAT GW.
 func (s *NatGatewayServiceImpl) PublishAddEvent(vpcId, natGatewayId, publicIp, subnetCidr string) {
-	if s.natsConn == nil {
-		return
-	}
-	evt := natGatewayEvent{
+	utils.PublishEvent(s.natsConn, "vpc.add-nat-gateway", natGatewayEvent{
 		VpcId:        vpcId,
 		NatGatewayId: natGatewayId,
 		PublicIp:     publicIp,
 		SubnetCidr:   subnetCidr,
-	}
-	eventData, err := json.Marshal(evt)
-	if err != nil {
-		slog.Error("Failed to marshal NAT GW add event", "err", err)
-		return
-	}
-	if err := s.natsConn.Publish("vpc.add-nat-gateway", eventData); err != nil {
-		slog.Warn("Failed to publish NAT GW add event", "err", err)
-	}
+	})
 }
 
 // PublishDeleteEvent publishes a vpc.delete-nat-gateway event for vpcd to remove the SNAT rule.
 func (s *NatGatewayServiceImpl) PublishDeleteEvent(vpcId, natGatewayId, publicIp, subnetCidr string) {
-	if s.natsConn == nil {
-		return
-	}
-	evt := natGatewayEvent{
+	utils.PublishEvent(s.natsConn, "vpc.delete-nat-gateway", natGatewayEvent{
 		VpcId:        vpcId,
 		NatGatewayId: natGatewayId,
 		PublicIp:     publicIp,
 		SubnetCidr:   subnetCidr,
-	}
-	eventData, err := json.Marshal(evt)
-	if err != nil {
-		slog.Error("Failed to marshal NAT GW delete event", "err", err)
-		return
-	}
-	if err := s.natsConn.Publish("vpc.delete-nat-gateway", eventData); err != nil {
-		slog.Warn("Failed to publish NAT GW delete event", "err", err)
-	}
+	})
 }
 
 // GetNatGateway retrieves a NAT Gateway record by ID
@@ -505,13 +475,7 @@ func recordToEC2(record *NatGatewayRecord) *ec2.NatGateway {
 		}
 	}
 
-	if len(record.Tags) > 0 {
-		tags := make([]*ec2.Tag, 0, len(record.Tags))
-		for k, v := range record.Tags {
-			tags = append(tags, &ec2.Tag{Key: aws.String(k), Value: aws.String(v)})
-		}
-		ngw.Tags = tags
-	}
+	ngw.Tags = utils.MapToEC2Tags(record.Tags)
 
 	return ngw
 }

spinifex/handlers/ec2/natgw/service_impl_test.go

@@ -2,16 +2,14 @@ package handlers_ec2_natgw
 
 import (
 	"testing"
-	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
 	handlers_ec2_eip "github.com/mulgadc/spinifex/spinifex/handlers/ec2/eip"
 	handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/mulgadc/spinifex/spinifex/utils"
-	"github.com/nats-io/nats-server/v2/server"
-	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -20,46 +18,23 @@ const testAccountID = "123456789012"
 
 func setupTestService(t *testing.T) *NatGatewayServiceImpl {
 	t.Helper()
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: true,
-		StoreDir:  t.TempDir(),
-		NoLog:     true,
-		NoSigs:    true,
-	}
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err)
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second))
-	t.Cleanup(func() { ns.Shutdown() })
-
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	t.Cleanup(func() { nc.Close() })
-
-	js, err := nc.JetStream()
-	require.NoError(t, err)
+	_, nc, js := testutil.StartTestJetStream(t)
 
 	// Seed VPC KV
-	vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1})
-	require.NoError(t, err)
-	_, err = vpcKV.Put(utils.AccountKey(testAccountID, "vpc-test1"), []byte(`{"vpc_id":"vpc-test1","cidr_block":"10.0.0.0/16","state":"available"}`))
-	require.NoError(t, err)
+	testutil.SeedKV(t, js, handlers_ec2_vpc.KVBucketVPCs, map[string][]byte{
+		utils.AccountKey(testAccountID, "vpc-test1"): []byte(`{"vpc_id":"vpc-test1","cidr_block":"10.0.0.0/16","state":"available"}`),
+	})
 
 	// Seed subnet KV (public subnet)
-	subnetKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketSubnets, History: 1})
-	require.NoError(t, err)
-	_, err = subnetKV.Put(utils.AccountKey(testAccountID, "subnet-pub1"), []byte(`{"subnet_id":"subnet-pub1","vpc_id":"vpc-test1","cidr_block":"10.0.1.0/24","state":"available","map_public_ip_on_launch":true}`))
-	require.NoError(t, err)
+	testutil.SeedKV(t, js, handlers_ec2_vpc.KVBucketSubnets, map[string][]byte{
+		utils.AccountKey(testAccountID, "subnet-pub1"): []byte(`{"subnet_id":"subnet-pub1","vpc_id":"vpc-test1","cidr_block":"10.0.1.0/24","state":"available","map_public_ip_on_launch":true}`),
+	})
 
 	// Seed EIP KV
-	eipKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_eip.KVBucketEIPs, History: 1})
-	require.NoError(t, err)
-	_, err = eipKV.Put(utils.AccountKey(testAccountID, "eipalloc-test1"), []byte(`{"allocation_id":"eipalloc-test1","public_ip":"203.0.113.50","state":"allocated"}`))
-	require.NoError(t, err)
-	_, err = eipKV.Put(utils.AccountKey(testAccountID, "eipalloc-used"), []byte(`{"allocation_id":"eipalloc-used","public_ip":"203.0.113.51","state":"associated","association_id":"eipassoc-xxx"}`))
-	require.NoError(t, err)
+	testutil.SeedKV(t, js, handlers_ec2_eip.KVBucketEIPs, map[string][]byte{
+		utils.AccountKey(testAccountID, "eipalloc-test1"): []byte(`{"allocation_id":"eipalloc-test1","public_ip":"203.0.113.50","state":"allocated"}`),
+		utils.AccountKey(testAccountID, "eipalloc-used"):  []byte(`{"allocation_id":"eipalloc-used","public_ip":"203.0.113.51","state":"associated","association_id":"eipassoc-xxx"}`),
+	})
 
 	svc, err := NewNatGatewayServiceImplWithNATS(nc)
 	require.NoError(t, err)

spinifex/handlers/ec2/placementgroup/service_impl_test.go

@@ -2,13 +2,11 @@ package handlers_ec2_placementgroup
 
 import (
 	"testing"
-	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
-	"github.com/nats-io/nats-server/v2/server"
-	"github.com/nats-io/nats.go"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -17,23 +15,7 @@ const testAccountID = "123456789012"
 
 func setupTestService(t *testing.T) *PlacementGroupServiceImpl {
 	t.Helper()
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: true,
-		StoreDir:  t.TempDir(),
-		NoLog:     true,
-		NoSigs:    true,
-	}
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err)
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second))
-	t.Cleanup(func() { ns.Shutdown() })
-
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	t.Cleanup(func() { nc.Close() })
+	_, nc, _ := testutil.StartTestJetStream(t)
 
 	svc, err := NewPlacementGroupServiceImplWithNATS(nil, nc)
 	require.NoError(t, err)

spinifex/handlers/ec2/routetable/service_impl.go

@@ -960,13 +960,7 @@ func recordToEC2(record *RouteTableRecord) *ec2.RouteTable {
 		rtb.Associations = append(rtb.Associations, assoc)
 	}
 
-	if len(record.Tags) > 0 {
-		tags := make([]*ec2.Tag, 0, len(record.Tags))
-		for k, v := range record.Tags {
-			tags = append(tags, &ec2.Tag{Key: aws.String(k), Value: aws.String(v)})
-		}
-		rtb.Tags = tags
-	}
+	rtb.Tags = utils.MapToEC2Tags(record.Tags)
 
 	return rtb
 }

spinifex/handlers/ec2/routetable/service_impl_test.go

@@ -2,16 +2,14 @@ package handlers_ec2_routetable
 
 import (
 	"testing"
-	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
 	handlers_ec2_igw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/igw"
 	handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/mulgadc/spinifex/spinifex/utils"
-	"github.com/nats-io/nats-server/v2/server"
-	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -20,44 +18,22 @@ const testAccountID = "123456789012"
 
 func setupTestService(t *testing.T) *RouteTableServiceImpl {
 	t.Helper()
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: true,
-		StoreDir:  t.TempDir(),
-		NoLog:     true,
-		NoSigs:    true,
-	}
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err)
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second))
-	t.Cleanup(func() { ns.Shutdown() })
-
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	t.Cleanup(func() { nc.Close() })
-
-	js, err := nc.JetStream()
-	require.NoError(t, err)
+	_, nc, js := testutil.StartTestJetStream(t)
 
 	// Seed VPC KV
-	vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1})
-	require.NoError(t, err)
-	_, err = vpcKV.Put(utils.AccountKey(testAccountID, "vpc-test1"), []byte(`{"vpc_id":"vpc-test1","cidr_block":"10.0.0.0/16","state":"available"}`))
-	require.NoError(t, err)
+	testutil.SeedKV(t, js, handlers_ec2_vpc.KVBucketVPCs, map[string][]byte{
+		utils.AccountKey(testAccountID, "vpc-test1"): []byte(`{"vpc_id":"vpc-test1","cidr_block":"10.0.0.0/16","state":"available"}`),
+	})
 
 	// Seed IGW KV (attached to vpc-test1)
-	igwKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_igw.KVBucketIGW, History: 1})
-	require.NoError(t, err)
-	_, err = igwKV.Put(utils.AccountKey(testAccountID, "igw-test1"), []byte(`{"internet_gateway_id":"igw-test1","vpc_id":"vpc-test1","state":"attached"}`))
-	require.NoError(t, err)
+	testutil.SeedKV(t, js, handlers_ec2_igw.KVBucketIGW, map[string][]byte{
+		utils.AccountKey(testAccountID, "igw-test1"): []byte(`{"internet_gateway_id":"igw-test1","vpc_id":"vpc-test1","state":"attached"}`),
+	})
 
 	// Seed subnet KV
-	subnetKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketSubnets, History: 1})
-	require.NoError(t, err)
-	_, err = subnetKV.Put(utils.AccountKey(testAccountID, "subnet-test1"), []byte(`{"subnet_id":"subnet-test1","vpc_id":"vpc-test1","cidr_block":"10.0.1.0/24","state":"available"}`))
-	require.NoError(t, err)
+	testutil.SeedKV(t, js, handlers_ec2_vpc.KVBucketSubnets, map[string][]byte{
+		utils.AccountKey(testAccountID, "subnet-test1"): []byte(`{"subnet_id":"subnet-test1","vpc_id":"vpc-test1","cidr_block":"10.0.1.0/24","state":"available"}`),
+	})
 
 	svc, err := NewRouteTableServiceImplWithNATS(nil, nc)
 	require.NoError(t, err)

spinifex/handlers/ec2/volume/service_impl.go

@@ -722,16 +722,7 @@ func (s *VolumeServiceImpl) getVolumeByID(volumeID string) (*volumeResult, error
 		}
 	}
 
-	if len(volMeta.Tags) > 0 {
-		tags := make([]*ec2.Tag, 0, len(volMeta.Tags))
-		for key, value := range volMeta.Tags {
-			tags = append(tags, &ec2.Tag{
-				Key:   aws.String(key),
-				Value: aws.String(value),
-			})
-		}
-		volume.Tags = tags
-	}
+	volume.Tags = utils.MapToEC2Tags(volMeta.Tags)
 
 	return &volumeResult{volume: volume, tenantID: volMeta.TenantID}, nil
 }

spinifex/handlers/ec2/vpc/eni.go

@@ -527,13 +527,7 @@ func (s *VPCServiceImpl) eniRecordToEC2(record *ENIRecord, accountID string) *ec
 		}
 	}
 
-	if len(record.Tags) > 0 {
-		tags := make([]*ec2.Tag, 0, len(record.Tags))
-		for k, v := range record.Tags {
-			tags = append(tags, &ec2.Tag{Key: aws.String(k), Value: aws.String(v)})
-		}
-		eni.TagSet = tags
-	}
+	eni.TagSet = utils.MapToEC2Tags(record.Tags)
 
 	return eni
 }
@@ -549,10 +543,7 @@ func generateENIMac(eniId string) string {
 
 // publishENIEvent publishes an ENI lifecycle event to NATS for vpcd consumption.
 func (s *VPCServiceImpl) publishENIEvent(topic, eniId, subnetId, vpcId, privateIP, macAddr string) {
-	if s.natsConn == nil {
-		return
-	}
-	evt := struct {
+	utils.PublishEvent(s.natsConn, topic, struct {
 		NetworkInterfaceId string `json:"network_interface_id"`
 		SubnetId           string `json:"subnet_id"`
 		VpcId              string `json:"vpc_id"`
@@ -564,14 +555,5 @@ func (s *VPCServiceImpl) publishENIEvent(topic, eniId, subnetId, vpcId, privateI
 		VpcId:              vpcId,
 		PrivateIpAddress:   privateIP,
 		MacAddress:         macAddr,
-	}
-
-	data, err := json.Marshal(evt)
-	if err != nil {
-		slog.Error("Failed to marshal ENI event", "topic", topic, "err", err)
-		return
-	}
-	if err := s.natsConn.Publish(topic, data); err != nil {
-		slog.Error("Failed to publish ENI event", "topic", topic, "err", err)
-	}
+	})
 }

spinifex/handlers/ec2/vpc/external_ipam_test.go

@@ -3,36 +3,15 @@ package handlers_ec2_vpc
 import (
 	"sync"
 	"testing"
-	"time"
 
-	"github.com/nats-io/nats-server/v2/server"
-	"github.com/nats-io/nats.go"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
 func setupTestExternalIPAM(t *testing.T, pools []ExternalPoolConfig) *ExternalIPAM {
 	t.Helper()
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: true,
-		StoreDir:  t.TempDir(),
-		NoLog:     true,
-		NoSigs:    true,
-	}
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err)
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second))
-	t.Cleanup(func() { ns.Shutdown() })
-
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	t.Cleanup(func() { nc.Close() })
-
-	js, err := nc.JetStream()
-	require.NoError(t, err)
+	_, _, js := testutil.StartTestJetStream(t)
 
 	ipam, err := NewExternalIPAM(js, pools)
 	require.NoError(t, err)
@@ -355,26 +334,7 @@ func TestExternalIPAM_RangeValidation(t *testing.T) {
 func TestExternalIPAM_InitFromConfig(t *testing.T) {
 	pool := testPool()
 	// Create IPAM twice — second init should be idempotent
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: true,
-		StoreDir:  t.TempDir(),
-		NoLog:     true,
-		NoSigs:    true,
-	}
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err)
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second))
-	t.Cleanup(func() { ns.Shutdown() })
-
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	t.Cleanup(func() { nc.Close() })
-
-	js, err := nc.JetStream()
-	require.NoError(t, err)
+	_, _, js := testutil.StartTestJetStream(t)
 
 	// First init
 	ipam1, err := NewExternalIPAM(js, []ExternalPoolConfig{pool})

spinifex/handlers/ec2/vpc/ipam_test.go

@@ -3,9 +3,8 @@ package handlers_ec2_vpc
 import (
 	"fmt"
 	"testing"
-	"time"
 
-	"github.com/nats-io/nats-server/v2/server"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -13,26 +12,7 @@ import (
 
 func setupTestIPAM(t *testing.T) *IPAM {
 	t.Helper()
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: true,
-		StoreDir:  t.TempDir(),
-		NoLog:     true,
-		NoSigs:    true,
-	}
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err)
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second))
-	t.Cleanup(func() { ns.Shutdown() })
-
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	t.Cleanup(func() { nc.Close() })
-
-	js, err := nc.JetStream()
-	require.NoError(t, err)
+	_, _, js := testutil.StartTestJetStream(t)
 
 	ipam, err := NewIPAM(js)
 	require.NoError(t, err)
@@ -186,26 +166,7 @@ func TestIPAM_LargerSubnet(t *testing.T) {
 }
 
 func TestIPAM_NewIPAMWithKV(t *testing.T) {
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: true,
-		StoreDir:  t.TempDir(),
-		NoLog:     true,
-		NoSigs:    true,
-	}
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err)
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second))
-	t.Cleanup(func() { ns.Shutdown() })
-
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	t.Cleanup(func() { nc.Close() })
-
-	js, err := nc.JetStream()
-	require.NoError(t, err)
+	_, _, js := testutil.StartTestJetStream(t)
 
 	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{
 		Bucket: "test-ipam-kv",

spinifex/handlers/ec2/vpc/security_group.go

@@ -494,13 +494,7 @@ func (s *VPCServiceImpl) sgRecordToEC2(record *SecurityGroupRecord, accountID st
 		IpPermissionsEgress: sgRulesToIpPermissions(record.EgressRules),
 	}
 
-	if len(record.Tags) > 0 {
-		tags := make([]*ec2.Tag, 0, len(record.Tags))
-		for k, v := range record.Tags {
-			tags = append(tags, &ec2.Tag{Key: aws.String(k), Value: aws.String(v)})
-		}
-		sg.Tags = tags
-	}
+	sg.Tags = utils.MapToEC2Tags(record.Tags)
 
 	return sg
 }
@@ -628,15 +622,5 @@ func sgRuleKey(r SGRule) string {
 
 // publishSGEvent publishes a security group lifecycle event to NATS for vpcd consumption.
 func (s *VPCServiceImpl) publishSGEvent(topic string, evt SGEvent) {
-	if s.natsConn == nil {
-		return
-	}
-	data, err := json.Marshal(evt)
-	if err != nil {
-		slog.Error("Failed to marshal SG event", "topic", topic, "err", err)
-		return
-	}
-	if err := s.natsConn.Publish(topic, data); err != nil {
-		slog.Error("Failed to publish SG event", "topic", topic, "err", err)
-	}
+	utils.PublishEvent(s.natsConn, topic, evt)
 }

spinifex/handlers/ec2/vpc/service_impl.go

@@ -713,13 +713,7 @@ func (s *VPCServiceImpl) vpcRecordToEC2(record *VPCRecord, accountID string) *ec
 		InstanceTenancy: aws.String("default"),
 	}
 
-	if len(record.Tags) > 0 {
-		tags := make([]*ec2.Tag, 0, len(record.Tags))
-		for k, v := range record.Tags {
-			tags = append(tags, &ec2.Tag{Key: aws.String(k), Value: aws.String(v)})
-		}
-		vpc.Tags = tags
-	}
+	vpc.Tags = utils.MapToEC2Tags(record.Tags)
 
 	return vpc
 }
@@ -737,13 +731,7 @@ func (s *VPCServiceImpl) subnetRecordToEC2(record *SubnetRecord, availableIPs in
 		MapPublicIpOnLaunch:     aws.Bool(record.MapPublicIpOnLaunch),
 	}
 
-	if len(record.Tags) > 0 {
-		tags := make([]*ec2.Tag, 0, len(record.Tags))
-		for k, v := range record.Tags {
-			tags = append(tags, &ec2.Tag{Key: aws.String(k), Value: aws.String(v)})
-		}
-		subnet.Tags = tags
-	}
+	subnet.Tags = utils.MapToEC2Tags(record.Tags)
 
 	return subnet
 }
@@ -1136,42 +1124,18 @@ func (s *VPCServiceImpl) GetSubnet(accountID, subnetId string) (*SubnetRecord, e
 // publishVPCEvent publishes a VPC lifecycle event to NATS for vpcd consumption.
 // This is fire-and-forget; errors are logged but do not fail the API response.
 func (s *VPCServiceImpl) publishVPCEvent(topic, vpcId, cidrBlock string, vni int64) {
-	if s.natsConn == nil {
-		return
-	}
-	evt := struct {
+	utils.PublishEvent(s.natsConn, topic, struct {
 		VpcId     string `json:"vpc_id"`
 		CidrBlock string `json:"cidr_block"`
 		VNI       int64  `json:"vni"`
-	}{VpcId: vpcId, CidrBlock: cidrBlock, VNI: vni}
-
-	data, err := json.Marshal(evt)
-	if err != nil {
-		slog.Error("Failed to marshal VPC event", "topic", topic, "err", err)
-		return
-	}
-	if err := s.natsConn.Publish(topic, data); err != nil {
-		slog.Error("Failed to publish VPC event", "topic", topic, "err", err)
-	}
+	}{VpcId: vpcId, CidrBlock: cidrBlock, VNI: vni})
 }
 
 // publishSubnetEvent publishes a subnet lifecycle event to NATS for vpcd consumption.
 func (s *VPCServiceImpl) publishSubnetEvent(topic, subnetId, vpcId, cidrBlock string) {
-	if s.natsConn == nil {
-		return
-	}
-	evt := struct {
+	utils.PublishEvent(s.natsConn, topic, struct {
 		SubnetId  string `json:"subnet_id"`
 		VpcId     string `json:"vpc_id"`
 		CidrBlock string `json:"cidr_block"`
-	}{SubnetId: subnetId, VpcId: vpcId, CidrBlock: cidrBlock}
-
-	data, err := json.Marshal(evt)
-	if err != nil {
-		slog.Error("Failed to marshal subnet event", "topic", topic, "err", err)
-		return
-	}
-	if err := s.natsConn.Publish(topic, data); err != nil {
-		slog.Error("Failed to publish subnet event", "topic", topic, "err", err)
-	}
+	}{SubnetId: subnetId, VpcId: vpcId, CidrBlock: cidrBlock})
 }

spinifex/handlers/ec2/vpc/service_impl_test.go

@@ -8,7 +8,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/mulgadc/spinifex/spinifex/config"
-	"github.com/nats-io/nats-server/v2/server"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -18,23 +18,7 @@ const testAccountID = "123456789012"
 
 func setupTestVPCServiceWithNC(t *testing.T) (*VPCServiceImpl, *nats.Conn) {
 	t.Helper()
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: true,
-		StoreDir:  t.TempDir(),
-		NoLog:     true,
-		NoSigs:    true,
-	}
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err)
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second))
-	t.Cleanup(func() { ns.Shutdown() })
-
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	t.Cleanup(func() { nc.Close() })
+	_, nc, _ := testutil.StartTestJetStream(t)
 
 	svc, err := NewVPCServiceImplWithNATS(nil, nc)
 	require.NoError(t, err)
@@ -795,23 +779,7 @@ func TestDeleteSubnet_PublishesEvent(t *testing.T) {
 
 func TestEnsureDefaultVPC_WithConfigAZ(t *testing.T) {
 	// Create a service with custom config that has AZ set
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: true,
-		StoreDir:  t.TempDir(),
-		NoLog:     true,
-		NoSigs:    true,
-	}
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err)
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second))
-	t.Cleanup(func() { ns.Shutdown() })
-
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	t.Cleanup(func() { nc.Close() })
+	_, nc, _ := testutil.StartTestJetStream(t)
 
 	cfg := &config.Config{AZ: "us-west-2b"}
 	svc, err := NewVPCServiceImplWithNATS(cfg, nc)

spinifex/handlers/elbv2/service_impl_health_test.go

@@ -2,14 +2,12 @@ package handlers_elbv2
 
 import (
 	"testing"
-	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/aws/aws-sdk-go/service/elbv2"
 	handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
-	"github.com/nats-io/nats-server/v2/server"
-	"github.com/nats-io/nats.go"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -18,23 +16,7 @@ import (
 // and a simulated instance ENI (attached to an instance ID).
 func setupTestServiceWithInstance(t *testing.T, instanceID, instanceIP string) (*ELBv2ServiceImpl, *handlers_ec2_vpc.VPCServiceImpl, string) {
 	t.Helper()
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: true,
-		StoreDir:  t.TempDir(),
-		NoLog:     true,
-		NoSigs:    true,
-	}
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err)
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second))
-	t.Cleanup(func() { ns.Shutdown() })
-
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	t.Cleanup(func() { nc.Close() })
+	_, nc, _ := testutil.StartTestJetStream(t)
 
 	vpcSvc, err := handlers_ec2_vpc.NewVPCServiceImplWithNATS(nil, nc)
 	require.NoError(t, err)

spinifex/handlers/elbv2/service_impl_test.go

@@ -2,35 +2,17 @@ package handlers_elbv2
 
 import (
 	"testing"
-	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/elbv2"
-	"github.com/nats-io/nats-server/v2/server"
-	"github.com/nats-io/nats.go"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
 func setupTestService(t *testing.T) *ELBv2ServiceImpl {
 	t.Helper()
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: true,
-		StoreDir:  t.TempDir(),
-		NoLog:     true,
-		NoSigs:    true,
-	}
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err)
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second))
-	t.Cleanup(func() { ns.Shutdown() })
-
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	t.Cleanup(func() { nc.Close() })
+	_, nc, _ := testutil.StartTestJetStream(t)
 
 	svc, err := NewELBv2ServiceImplWithNATS(nil, nc)
 	require.NoError(t, err)

spinifex/handlers/elbv2/service_impl_vpc_test.go

@@ -2,14 +2,12 @@ package handlers_elbv2
 
 import (
 	"testing"
-	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/aws/aws-sdk-go/service/elbv2"
 	handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
-	"github.com/nats-io/nats-server/v2/server"
-	"github.com/nats-io/nats.go"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -18,23 +16,7 @@ import (
 // with a pre-created VPC and subnet for ENI allocation testing.
 func setupTestServiceWithVPC(t *testing.T) (*ELBv2ServiceImpl, *handlers_ec2_vpc.VPCServiceImpl) {
 	t.Helper()
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: true,
-		StoreDir:  t.TempDir(),
-		NoLog:     true,
-		NoSigs:    true,
-	}
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err)
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second))
-	t.Cleanup(func() { ns.Shutdown() })
-
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	t.Cleanup(func() { nc.Close() })
+	_, nc, _ := testutil.StartTestJetStream(t)
 
 	// Create VPC service
 	vpcSvc, err := handlers_ec2_vpc.NewVPCServiceImplWithNATS(nil, nc)

spinifex/handlers/elbv2/service_nats_e2e_test.go

@@ -3,12 +3,11 @@ package handlers_elbv2
 import (
 	"encoding/json"
 	"testing"
-	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/elbv2"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/mulgadc/spinifex/spinifex/utils"
-	"github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -19,23 +18,7 @@ import (
 func setupNATSELBv2Test(t *testing.T) (ELBv2Service, *ELBv2ServiceImpl) {
 	t.Helper()
 
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: true,
-		StoreDir:  t.TempDir(),
-		NoLog:     true,
-		NoSigs:    true,
-	}
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err)
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second))
-	t.Cleanup(func() { ns.Shutdown() })
-
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	t.Cleanup(func() { nc.Close() })
+	_, nc, _ := testutil.StartTestJetStream(t)
 
 	backend, err := NewELBv2ServiceImplWithNATS(nil, nc)
 	require.NoError(t, err)

spinifex/handlers/elbv2/store_test.go

@@ -4,8 +4,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/nats-io/nats-server/v2/server"
-	"github.com/nats-io/nats.go"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -14,23 +13,7 @@ const testAccountID = "123456789012"
 
 func setupTestStore(t *testing.T) *Store {
 	t.Helper()
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: true,
-		StoreDir:  t.TempDir(),
-		NoLog:     true,
-		NoSigs:    true,
-	}
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err)
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second))
-	t.Cleanup(func() { ns.Shutdown() })
-
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	t.Cleanup(func() { nc.Close() })
+	_, nc, _ := testutil.StartTestJetStream(t)
 
 	store, err := NewStore(nc)
 	require.NoError(t, err)

spinifex/handlers/iam/service_impl_test.go

@@ -11,9 +11,8 @@ import (
 	"github.com/aws/aws-sdk-go/service/iam"
 	"github.com/mulgadc/spinifex/spinifex/admin"
 	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/mulgadc/spinifex/spinifex/utils"
-	"github.com/nats-io/nats-server/v2/server"
-	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -25,23 +24,7 @@ const testAccountID = utils.GlobalAccountID
 
 func setupTestIAMService(t *testing.T) *IAMServiceImpl {
 	t.Helper()
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: true,
-		StoreDir:  t.TempDir(),
-		NoLog:     true,
-		NoSigs:    true,
-	}
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err)
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second))
-	t.Cleanup(func() { ns.Shutdown() })
-
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	t.Cleanup(func() { nc.Close() })
+	_, nc, _ := testutil.StartTestJetStream(t)
 
 	masterKey, err := GenerateMasterKey()
 	require.NoError(t, err)

spinifex/services/awsgw/awsgw.go

@@ -65,14 +65,7 @@ func (svc *Service) Stop() (err error) {
 }
 
 func (svc *Service) Status() (string, error) {
-	pid, err := utils.ReadPidFileFrom(svc.Config.NodeBaseDir(), serviceName)
-	if err != nil {
-		if os.IsNotExist(err) {
-			return "stopped", nil
-		}
-		return "", fmt.Errorf("read pid file: %w", err)
-	}
-	return fmt.Sprintf("running (pid: %d)", pid), nil
+	return utils.ServiceStatus(svc.Config.NodeBaseDir(), serviceName)
 }
 
 func (svc *Service) Shutdown() (err error) {

spinifex/services/nats/nats.go

@@ -52,14 +52,7 @@ func (svc *Service) Stop() (err error) {
 }
 
 func (svc *Service) Status() (string, error) {
-	pid, err := utils.ReadPidFileFrom(svc.Config.DataDir, serviceName)
-	if err != nil {
-		if os.IsNotExist(err) {
-			return "stopped", nil
-		}
-		return "", fmt.Errorf("read pid file: %w", err)
-	}
-	return fmt.Sprintf("running (pid: %d)", pid), nil
+	return utils.ServiceStatus(svc.Config.DataDir, serviceName)
 }
 
 func (svc *Service) Shutdown() (err error) {

spinifex/services/predastore/predastore.go

@@ -106,14 +106,7 @@ func (svc *Service) Stop() error {
 
 // Status returns the status of the predastore service
 func (svc *Service) Status() (string, error) {
-	pid, err := utils.ReadPidFileFrom(svc.Config.BasePath, serviceName)
-	if err != nil {
-		if os.IsNotExist(err) {
-			return "stopped", nil
-		}
-		return "", fmt.Errorf("read pid file: %w", err)
-	}
-	return fmt.Sprintf("running (pid: %d)", pid), nil
+	return utils.ServiceStatus(svc.Config.BasePath, serviceName)
 }
 
 // Shutdown gracefully shuts down the predastore service

spinifex/services/spinifex/spinifex.go

@@ -49,14 +49,7 @@ func (svc *Service) Stop() (err error) {
 }
 
 func (svc *Service) Status() (string, error) {
-	pid, err := utils.ReadPidFileFrom(svc.Config.NodeBaseDir(), serviceName)
-	if err != nil {
-		if os.IsNotExist(err) {
-			return "stopped", nil
-		}
-		return "", fmt.Errorf("read pid file: %w", err)
-	}
-	return fmt.Sprintf("running (pid: %d)", pid), nil
+	return utils.ServiceStatus(svc.Config.NodeBaseDir(), serviceName)
 }
 
 func (svc *Service) Shutdown() (err error) {

spinifex/services/spinifexui/spinifexui.go

@@ -98,14 +98,7 @@ func (svc *Service) Stop() error {
 
 // Status returns the status of the spinifex-ui service
 func (svc *Service) Status() (string, error) {
-	pid, err := utils.ReadPidFile(serviceName)
-	if err != nil {
-		if os.IsNotExist(err) {
-			return "stopped", nil
-		}
-		return "", fmt.Errorf("read pid file: %w", err)
-	}
-	return fmt.Sprintf("running (pid: %d)", pid), nil
+	return utils.ServiceStatus("", serviceName)
 }
 
 // Shutdown gracefully shuts down the spinifex-ui service

spinifex/services/viperblockd/viperblockd.go

@@ -86,10 +86,7 @@ func makeSnapshotHandler(vb *viperblock.VB, volumeName string) nats.MsgHandler {
 		var snapRequest types.EBSSnapshotRequest
 		if err := json.Unmarshal(msg.Data, &snapRequest); err != nil {
 			slog.Error("Failed to unmarshal ebs.snapshot message", "volume", volumeName, "err", err)
-			errResp, _ := json.Marshal(types.EBSSnapshotResponse{Error: fmt.Sprintf("bad request: %v", err)})
-			if err := msg.Respond(errResp); err != nil {
-				slog.Error("Failed to respond to ebs.snapshot request", "err", err)
-			}
+			respondJSON(msg, types.EBSSnapshotResponse{Error: fmt.Sprintf("bad request: %v", err)})
 			return
 		}
 
@@ -105,18 +102,38 @@ func makeSnapshotHandler(vb *viperblock.VB, volumeName string) nats.MsgHandler {
 			slog.Info("ebs.snapshot: snapshot created", "volume", volumeName, "snapshotId", snapRequest.SnapshotID)
 		}
 
-		response, err := json.Marshal(snapResponse)
-		if err != nil {
-			slog.Error("Failed to marshal ebs.snapshot response", "err", err)
-			if err := msg.Respond([]byte(`{"Error":"internal marshal failure"}`)); err != nil {
-				slog.Error("Failed to respond to ebs.snapshot request", "err", err)
-			}
-			return
-		}
+		respondJSON(msg, snapResponse)
+	}
+}
 
-		if err := msg.Respond(response); err != nil {
-			slog.Error("Failed to respond to ebs.snapshot request", "err", err)
-		}
+// respondJSON marshals data and sends it as a NATS response. On marshal
+// failure a raw JSON error string is sent instead.
+func respondJSON(msg *nats.Msg, data any) {
+	response, err := json.Marshal(data)
+	if err != nil {
+		slog.Error("Failed to marshal response", "type", fmt.Sprintf("%T", data), "err", err)
+		_ = msg.Respond([]byte(`{"Error":"internal marshal failure"}`))
+		return
+	}
+	if err := msg.Respond(response); err != nil {
+		slog.Error("Failed to respond to NATS request", "err", err)
+	}
+}
+
+// respondAndPublish is like respondJSON but also publishes the marshaled
+// response to the given NATS topic (used for ebs.mount.response etc.).
+func respondAndPublish(msg *nats.Msg, nc *nats.Conn, topic string, data any) {
+	response, err := json.Marshal(data)
+	if err != nil {
+		slog.Error("Failed to marshal response", "type", fmt.Sprintf("%T", data), "err", err)
+		_ = msg.Respond([]byte(`{"Error":"internal marshal failure"}`))
+		return
+	}
+	if err := msg.Respond(response); err != nil {
+		slog.Error("Failed to respond to NATS request", "err", err)
+	}
+	if err := nc.Publish(topic, response); err != nil {
+		slog.Error("Failed to publish response", "topic", topic, "err", err)
 	}
 }
 
@@ -139,14 +156,7 @@ func (svc *Service) Stop() (err error) {
 }
 
 func (svc *Service) Status() (string, error) {
-	pid, err := utils.ReadPidFileFrom(svc.Config.BaseDir, serviceName)
-	if err != nil {
-		if os.IsNotExist(err) {
-			return "stopped", nil
-		}
-		return "", fmt.Errorf("read pid file: %w", err)
-	}
-	return fmt.Sprintf("running (pid: %d)", pid), nil
+	return utils.ServiceStatus(svc.Config.BaseDir, serviceName)
 }
 
 func (svc *Service) Shutdown() (err error) {
@@ -177,13 +187,9 @@ func launchService(cfg *Config) (err error) {
 		slog.Info("Received ebs.delete message", "data", string(msg.Data))
 
 		var ebsRequest types.EBSDeleteRequest
-		err := json.Unmarshal(msg.Data, &ebsRequest)
-		if err != nil {
+		if err := json.Unmarshal(msg.Data, &ebsRequest); err != nil {
 			slog.Error("Failed to unmarshal ebs.delete message", "err", err)
-			errResp, _ := json.Marshal(types.EBSDeleteResponse{Error: fmt.Sprintf("bad request: %v", err)})
-			if err := msg.Respond(errResp); err != nil {
-				slog.Error("Failed to respond to ebs.delete request", "err", err)
-			}
+			respondJSON(msg, types.EBSDeleteResponse{Error: fmt.Sprintf("bad request: %v", err)})
 			return
 		}
 
@@ -232,18 +238,7 @@ func launchService(cfg *Config) (err error) {
 			slog.Info("ebs.delete: volume not mounted (expected for available volumes)", "volume", ebsRequest.Volume)
 		}
 
-		respData, err := json.Marshal(response)
-		if err != nil {
-			slog.Error("Failed to marshal ebs.delete response", "err", err)
-			if err := msg.Respond([]byte(`{"Error":"internal marshal failure"}`)); err != nil {
-				slog.Error("Failed to respond to ebs.delete request", "err", err)
-			}
-			return
-		}
-
-		if err := msg.Respond(respData); err != nil {
-			slog.Error("Failed to respond to ebs.delete request", "err", err)
-		}
+		respondJSON(msg, response)
 	}); err != nil {
 		return fmt.Errorf("failed to subscribe to ebs.delete: %w", err)
 	}
@@ -264,18 +259,9 @@ func launchService(cfg *Config) (err error) {
 
 		// Parse the message
 		var ebsRequest types.EBSRequest
-		err := json.Unmarshal(msg.Data, &ebsRequest)
-		if err != nil {
+		if err := json.Unmarshal(msg.Data, &ebsRequest); err != nil {
 			slog.Error("Failed to unmarshal ebs.unmount message", "err", err)
-			errResp, marshalErr := json.Marshal(types.EBSUnMountResponse{Error: fmt.Sprintf("bad request: %v", err)})
-			if marshalErr != nil {
-				slog.Error("Failed to marshal ebs.unmount error response", "err", marshalErr)
-				_ = msg.Respond([]byte(`{"Error":"internal marshal failure"}`))
-				return
-			}
-			if err := msg.Respond(errResp); err != nil {
-				slog.Error("Failed to respond to ebs.unmount request", "err", err)
-			}
+			respondJSON(msg, types.EBSUnMountResponse{Error: fmt.Sprintf("bad request: %v", err)})
 			return
 		}
 
@@ -335,20 +321,7 @@ func launchService(cfg *Config) (err error) {
 			}
 		}
 
-		// Marshal the response
-		response, err := json.Marshal(ebsResponse)
-		if err != nil {
-			slog.Error("Failed to marshal response", "err", err)
-			_ = msg.Respond([]byte(`{"Error":"internal marshal failure"}`))
-			return
-		}
-
-		if err := msg.Respond(response); err != nil {
-			slog.Error("Failed to respond to ebs.unmount request", "err", err)
-		}
-		if err := nc.Publish("ebs.unmount.response", response); err != nil {
-			slog.Error("Failed to publish ebs.unmount.response", "err", err)
-		}
+		respondAndPublish(msg, nc, "ebs.unmount.response", ebsResponse)
 	}); err != nil {
 		return fmt.Errorf("failed to subscribe to %s: %w", unmountTopic, err)
 	}
@@ -359,15 +332,7 @@ func launchService(cfg *Config) (err error) {
 		var syncRequest types.EBSSyncRequest
 		if err := json.Unmarshal(msg.Data, &syncRequest); err != nil {
 			slog.Error("Failed to unmarshal ebs.sync message", "err", err)
-			errResp, marshalErr := json.Marshal(types.EBSSyncResponse{Error: fmt.Sprintf("bad request: %v", err)})
-			if marshalErr != nil {
-				slog.Error("Failed to marshal ebs.sync error response", "err", marshalErr)
-				_ = msg.Respond([]byte(`{"Error":"internal marshal failure"}`))
-				return
-			}
-			if err := msg.Respond(errResp); err != nil {
-				slog.Error("Failed to respond to ebs.sync request", "err", err)
-			}
+			respondJSON(msg, types.EBSSyncResponse{Error: fmt.Sprintf("bad request: %v", err)})
 			return
 		}
 
@@ -396,18 +361,7 @@ func launchService(cfg *Config) (err error) {
 				"volumeSize", foundVB.GetVolumeSize())
 		}
 
-		response, err := json.Marshal(syncResponse)
-		if err != nil {
-			slog.Error("Failed to marshal ebs.sync response", "err", err)
-			if err := msg.Respond([]byte(`{"Error":"internal marshal failure"}`)); err != nil {
-				slog.Error("Failed to respond to ebs.sync request", "err", err)
-			}
-			return
-		}
-
-		if err := msg.Respond(response); err != nil {
-			slog.Error("Failed to respond to ebs.sync request", "err", err)
-		}
+		respondJSON(msg, syncResponse)
 	}); err != nil {
 		return fmt.Errorf("failed to subscribe to ebs.sync: %w", err)
 	}
@@ -432,18 +386,9 @@ func launchService(cfg *Config) (err error) {
 
 		// Parse the message
 		var ebsRequest types.EBSRequest
-		err := json.Unmarshal(msg.Data, &ebsRequest)
-		if err != nil {
+		if err := json.Unmarshal(msg.Data, &ebsRequest); err != nil {
 			slog.Error("Failed to unmarshal ebs.mount message", "err", err)
-			errResp, marshalErr := json.Marshal(types.EBSMountResponse{Error: fmt.Sprintf("bad request: %v", err)})
-			if marshalErr != nil {
-				slog.Error("Failed to marshal ebs.mount error response", "err", marshalErr)
-				_ = msg.Respond([]byte(`{"Error":"internal marshal failure"}`))
-				return
-			}
-			if err := msg.Respond(errResp); err != nil {
-				slog.Error("Failed to respond to ebs.mount request", "err", err)
-			}
+			respondJSON(msg, types.EBSMountResponse{Error: fmt.Sprintf("bad request: %v", err)})
 			return
 		}
 
@@ -498,19 +443,7 @@ func launchService(cfg *Config) (err error) {
 
 		if err != nil {
 			ebsResponse.Error = fmt.Sprintf("Failed to connect to Viperblock store: %v", err)
-			// Marshal and send error response immediately
-			response, marshalErr := json.Marshal(ebsResponse)
-			if marshalErr != nil {
-				slog.Error("Failed to marshal ebs.mount error response", "err", marshalErr)
-				_ = msg.Respond([]byte(`{"Error":"internal marshal failure"}`))
-				return
-			}
-			if err := msg.Respond(response); err != nil {
-				slog.Error("Failed to respond to ebs.mount request", "err", err)
-			}
-			if err := nc.Publish("ebs.mount.response", response); err != nil {
-				slog.Error("Failed to publish ebs.mount.response", "err", err)
-			}
+			respondAndPublish(msg, nc, "ebs.mount.response", ebsResponse)
 			return
 		}
 
@@ -523,19 +456,7 @@ func launchService(cfg *Config) (err error) {
 
 		if err != nil {
 			ebsResponse.Error = err.Error()
-			// Marshal and send error response immediately
-			response, marshalErr := json.Marshal(ebsResponse)
-			if marshalErr != nil {
-				slog.Error("Failed to marshal ebs.mount error response", "err", marshalErr)
-				_ = msg.Respond([]byte(`{"Error":"internal marshal failure"}`))
-				return
-			}
-			if err := msg.Respond(response); err != nil {
-				slog.Error("Failed to respond to ebs.mount request", "err", err)
-			}
-			if err := nc.Publish("ebs.mount.response", response); err != nil {
-				slog.Error("Failed to publish ebs.mount.response", "err", err)
-			}
+			respondAndPublish(msg, nc, "ebs.mount.response", ebsResponse)
 			return
 		}
 
@@ -545,19 +466,7 @@ func launchService(cfg *Config) (err error) {
 
 		if err != nil {
 			ebsResponse.Error = err.Error()
-			// Marshal and send error response immediately
-			response, marshalErr := json.Marshal(ebsResponse)
-			if marshalErr != nil {
-				slog.Error("Failed to marshal ebs.mount error response", "err", marshalErr)
-				_ = msg.Respond([]byte(`{"Error":"internal marshal failure"}`))
-				return
-			}
-			if err := msg.Respond(response); err != nil {
-				slog.Error("Failed to respond to ebs.mount request", "err", err)
-			}
-			if err := nc.Publish("ebs.mount.response", response); err != nil {
-				slog.Error("Failed to publish ebs.mount.response", "err", err)
-			}
+			respondAndPublish(msg, nc, "ebs.mount.response", ebsResponse)
 			return
 		}
 
@@ -575,18 +484,7 @@ func launchService(cfg *Config) (err error) {
 			portStr, err := viperblock.FindFreePort()
 			if err != nil {
 				ebsResponse.Error = err.Error()
-				response, marshalErr := json.Marshal(ebsResponse)
-				if marshalErr != nil {
-					slog.Error("Failed to marshal ebs.mount error response", "err", marshalErr)
-					_ = msg.Respond([]byte(`{"Error":"internal marshal failure"}`))
-					return
-				}
-				if err := msg.Respond(response); err != nil {
-					slog.Error("Failed to respond to ebs.mount request", "err", err)
-				}
-				if err := nc.Publish("ebs.mount.response", response); err != nil {
-					slog.Error("Failed to publish ebs.mount.response", "err", err)
-				}
+				respondAndPublish(msg, nc, "ebs.mount.response", ebsResponse)
 				return
 			}
 
@@ -596,18 +494,7 @@ func launchService(cfg *Config) (err error) {
 			if err != nil {
 				slog.Error("Failed to convert port to int", "err", err)
 				ebsResponse.Error = fmt.Sprintf("failed to parse port: %v", err)
-				response, marshalErr := json.Marshal(ebsResponse)
-				if marshalErr != nil {
-					slog.Error("Failed to marshal ebs.mount error response", "err", marshalErr)
-					_ = msg.Respond([]byte(`{"Error":"internal marshal failure"}`))
-					return
-				}
-				if err := msg.Respond(response); err != nil {
-					slog.Error("Failed to respond to ebs.mount request", "err", err)
-				}
-				if err := nc.Publish("ebs.mount.response", response); err != nil {
-					slog.Error("Failed to publish ebs.mount.response", "err", err)
-				}
+				respondAndPublish(msg, nc, "ebs.mount.response", ebsResponse)
 				return
 			}
 
@@ -618,18 +505,7 @@ func launchService(cfg *Config) (err error) {
 			nbdSocket, err = utils.GenerateUniqueSocketFile(ebsRequest.Name)
 			if err != nil {
 				ebsResponse.Error = err.Error()
-				response, marshalErr := json.Marshal(ebsResponse)
-				if marshalErr != nil {
-					slog.Error("Failed to marshal ebs.mount error response", "err", marshalErr)
-					_ = msg.Respond([]byte(`{"Error":"internal marshal failure"}`))
-					return
-				}
-				if err := msg.Respond(response); err != nil {
-					slog.Error("Failed to respond to ebs.mount request", "err", err)
-				}
-				if err := nc.Publish("ebs.mount.response", response); err != nil {
-					slog.Error("Failed to publish ebs.mount.response", "err", err)
-				}
+				respondAndPublish(msg, nc, "ebs.mount.response", ebsResponse)
 				return
 			}
 
@@ -642,18 +518,7 @@ func launchService(cfg *Config) (err error) {
 		if err != nil {
 			slog.Error("Failed to generate nbdkit pid file", "err", err)
 			ebsResponse.Error = fmt.Sprintf("failed to generate pid file: %v", err)
-			response, marshalErr := json.Marshal(ebsResponse)
-			if marshalErr != nil {
-				slog.Error("Failed to marshal ebs.mount error response", "err", marshalErr)
-				_ = msg.Respond([]byte(`{"Error":"internal marshal failure"}`))
-				return
-			}
-			if err := msg.Respond(response); err != nil {
-				slog.Error("Failed to respond to ebs.mount request", "err", err)
-			}
-			if err := nc.Publish("ebs.mount.response", response); err != nil {
-				slog.Error("Failed to publish ebs.mount.response", "err", err)
-			}
+			respondAndPublish(msg, nc, "ebs.mount.response", ebsResponse)
 			return
 		}
 
@@ -716,19 +581,7 @@ func launchService(cfg *Config) (err error) {
 
 		if pid == 0 {
 			ebsResponse.Error = "Failed to start nbdkit"
-			// Marshal and send error response immediately
-			response, marshalErr := json.Marshal(ebsResponse)
-			if marshalErr != nil {
-				slog.Error("Failed to marshal ebs.mount error response", "err", marshalErr)
-				_ = msg.Respond([]byte(`{"Error":"internal marshal failure"}`))
-				return
-			}
-			if err := msg.Respond(response); err != nil {
-				slog.Error("Failed to respond to ebs.mount request", "err", err)
-			}
-			if err := nc.Publish("ebs.mount.response", response); err != nil {
-				slog.Error("Failed to publish ebs.mount.response", "err", err)
-			}
+			respondAndPublish(msg, nc, "ebs.mount.response", ebsResponse)
 			return
 		}
 
@@ -740,18 +593,7 @@ func launchService(cfg *Config) (err error) {
 		case exitErr := <-exitChan:
 			if exitErr != 0 {
 				ebsResponse.Error = fmt.Sprintf("nbdkit failed: %v", exitErr)
-				response, marshalErr := json.Marshal(ebsResponse)
-				if marshalErr != nil {
-					slog.Error("Failed to marshal ebs.mount error response", "err", marshalErr)
-					_ = msg.Respond([]byte(`{"Error":"internal marshal failure"}`))
-					return
-				}
-				if err := msg.Respond(response); err != nil {
-					slog.Error("Failed to respond to ebs.mount request", "err", err)
-				}
-				if err := nc.Publish("ebs.mount.response", response); err != nil {
-					slog.Error("Failed to publish ebs.mount.response", "err", err)
-				}
+				respondAndPublish(msg, nc, "ebs.mount.response", ebsResponse)
 				return
 			}
 		default:
@@ -780,25 +622,8 @@ func launchService(cfg *Config) (err error) {
 		})
 		cfg.mu.Unlock()
 
-		// Marshal the response
-		response, err := json.Marshal(ebsResponse)
-		if err != nil {
-			slog.Error("Failed to marshal response", "err", err)
-			_ = msg.Respond([]byte(`{"Error":"internal marshal failure"}`))
-			return
-		}
-
-		if err := msg.Respond(response); err != nil {
-			slog.Error("Failed to respond to ebs.mount request", "err", err)
-			return
-		}
-
-		if err := nc.Publish("ebs.mount.response", response); err != nil {
-			slog.Error("Failed to publish ebs.mount.response", "err", err)
-			return
-		}
-
-		slog.Debug("Sent ebs.mount response", "response", string(response))
+		respondAndPublish(msg, nc, "ebs.mount.response", ebsResponse)
+		slog.Debug("Sent ebs.mount response")
 	}); err != nil {
 		return fmt.Errorf("failed to subscribe to %s: %w", mountTopic, err)
 	}

spinifex/services/vpcd/integration_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/ec2"
 	handlers_ec2_igw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/igw"
 	handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
@@ -20,34 +21,7 @@ const testIGWAccountID = "123456789012"
 // startTestJetStreamNATS starts an embedded NATS server with JetStream for integration tests.
 func startTestJetStreamNATS(t *testing.T) (*server.Server, *nats.Conn) {
 	t.Helper()
-	opts := &server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: true,
-		StoreDir:  t.TempDir(),
-		NoLog:     true,
-		NoSigs:    true,
-	}
-	ns, err := server.NewServer(opts)
-	if err != nil {
-		t.Fatalf("start NATS server: %v", err)
-	}
-	go ns.Start()
-	if !ns.ReadyForConnections(5_000_000_000) {
-		t.Fatal("NATS server not ready")
-	}
-
-	nc, err := nats.Connect(ns.ClientURL())
-	if err != nil {
-		ns.Shutdown()
-		t.Fatalf("connect to NATS: %v", err)
-	}
-
-	t.Cleanup(func() {
-		nc.Close()
-		ns.Shutdown()
-	})
-
+	ns, nc, _ := testutil.StartTestJetStream(t)
 	return ns, nc
 }
 

spinifex/services/vpcd/topology_test.go

@@ -6,6 +6,7 @@ import (
 	"testing"
 
 	"github.com/mulgadc/spinifex/spinifex/services/vpcd/nbdb"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/mulgadc/spinifex/spinifex/types"
 	"github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
@@ -14,31 +15,7 @@ import (
 // startTestNATS starts an embedded NATS server for testing.
 func startTestNATS(t *testing.T) (*server.Server, *nats.Conn) {
 	t.Helper()
-	opts := &server.Options{
-		Host: "127.0.0.1",
-		Port: -1, // auto-assign
-	}
-	ns, err := server.NewServer(opts)
-	if err != nil {
-		t.Fatalf("start NATS server: %v", err)
-	}
-	go ns.Start()
-	if !ns.ReadyForConnections(5_000_000_000) { // 5s
-		t.Fatal("NATS server not ready")
-	}
-
-	nc, err := nats.Connect(ns.ClientURL())
-	if err != nil {
-		ns.Shutdown()
-		t.Fatalf("connect to NATS: %v", err)
-	}
-
-	t.Cleanup(func() {
-		nc.Close()
-		ns.Shutdown()
-	})
-
-	return ns, nc
+	return testutil.StartTestNATS(t)
 }
 
 func TestTopologyHandler_VPCCreate(t *testing.T) {

spinifex/services/vpcd/vpcd.go

@@ -127,14 +127,7 @@ func (svc *Service) Stop() error {
 
 // Status returns the vpcd service status.
 func (svc *Service) Status() (string, error) {
-	pid, err := utils.ReadPidFileFrom(svc.Config.BaseDir, serviceName)
-	if err != nil {
-		if os.IsNotExist(err) {
-			return "stopped", nil
-		}
-		return "", fmt.Errorf("read pid file: %w", err)
-	}
-	return fmt.Sprintf("running (pid: %d)", pid), nil
+	return utils.ServiceStatus(svc.Config.BaseDir, serviceName)
 }
 
 // Shutdown gracefully shuts down the vpcd service.

spinifex/testutil/nats.go

@@ -0,0 +1,73 @@
+package testutil
+
+import (
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats-server/v2/server"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/require"
+)
+
+// StartTestNATS starts an embedded NATS server (no JetStream) and returns
+// the server and a connected client. Both are cleaned up via t.Cleanup.
+func StartTestNATS(t *testing.T) (*server.Server, *nats.Conn) {
+	t.Helper()
+	opts := &server.Options{
+		Host:   "127.0.0.1",
+		Port:   -1,
+		NoLog:  true,
+		NoSigs: true,
+	}
+	ns, err := server.NewServer(opts)
+	require.NoError(t, err)
+	go ns.Start()
+	require.True(t, ns.ReadyForConnections(5*time.Second))
+	t.Cleanup(func() { ns.Shutdown() })
+
+	nc, err := nats.Connect(ns.ClientURL())
+	require.NoError(t, err)
+	t.Cleanup(func() { nc.Close() })
+	return ns, nc
+}
+
+// StartTestJetStream starts an embedded NATS server with JetStream enabled
+// and returns the server, a connected client, and a JetStream context.
+// All resources are cleaned up via t.Cleanup.
+func StartTestJetStream(t *testing.T) (*server.Server, *nats.Conn, nats.JetStreamContext) {
+	t.Helper()
+	opts := &server.Options{
+		Host:      "127.0.0.1",
+		Port:      -1,
+		JetStream: true,
+		StoreDir:  t.TempDir(),
+		NoLog:     true,
+		NoSigs:    true,
+	}
+	ns, err := server.NewServer(opts)
+	require.NoError(t, err)
+	go ns.Start()
+	require.True(t, ns.ReadyForConnections(5*time.Second))
+	t.Cleanup(func() { ns.Shutdown() })
+
+	nc, err := nats.Connect(ns.ClientURL())
+	require.NoError(t, err)
+	t.Cleanup(func() { nc.Close() })
+
+	js, err := nc.JetStream()
+	require.NoError(t, err)
+	return ns, nc, js
+}
+
+// SeedKV creates a KV bucket and populates it with the given entries.
+// Returns the KV handle for further use.
+func SeedKV(t *testing.T, js nats.JetStreamContext, bucket string, entries map[string][]byte) nats.KeyValue {
+	t.Helper()
+	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: bucket, History: 1})
+	require.NoError(t, err)
+	for k, v := range entries {
+		_, err := kv.Put(k, v)
+		require.NoError(t, err)
+	}
+	return kv
+}

spinifex/utils/account_test.go

@@ -2,9 +2,8 @@ package utils
 
 import (
 	"testing"
-	"time"
 
-	"github.com/nats-io/nats-server/v2/server"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -62,33 +61,14 @@ func TestGlobalAccountID(t *testing.T) {
 }
 
 // startJSNATSServer starts an embedded JetStream-enabled NATS server for testing.
-func startJSNATSServer(t *testing.T) *server.Server {
+func startJSNATSServer(t *testing.T) (*nats.Conn, nats.JetStreamContext) {
 	t.Helper()
-	tmpDir := t.TempDir()
-
-	ns, err := server.NewServer(&server.Options{
-		Host:      "127.0.0.1",
-		Port:      -1,
-		JetStream: true,
-		StoreDir:  tmpDir,
-		NoLog:     true,
-		NoSigs:    true,
-	})
-	require.NoError(t, err)
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second))
-	t.Cleanup(func() { ns.Shutdown() })
-	return ns
+	_, nc, js := testutil.StartTestJetStream(t)
+	return nc, js
 }
 
 func TestWriteVersion_WritesKey(t *testing.T) {
-	ns := startJSNATSServer(t)
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	defer nc.Close()
-
-	js, err := nc.JetStream()
-	require.NoError(t, err)
+	_, js := startJSNATSServer(t)
 
 	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "test-write-version"})
 	require.NoError(t, err)
@@ -102,13 +82,7 @@ func TestWriteVersion_WritesKey(t *testing.T) {
 }
 
 func TestReadVersion_ReturnsZeroWhenUnset(t *testing.T) {
-	ns := startJSNATSServer(t)
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	defer nc.Close()
-
-	js, err := nc.JetStream()
-	require.NoError(t, err)
+	_, js := startJSNATSServer(t)
 
 	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "test-read-unset"})
 	require.NoError(t, err)
@@ -119,13 +93,7 @@ func TestReadVersion_ReturnsZeroWhenUnset(t *testing.T) {
 }
 
 func TestWriteVersion_IdempotentOnSameVersion(t *testing.T) {
-	ns := startJSNATSServer(t)
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	defer nc.Close()
-
-	js, err := nc.JetStream()
-	require.NoError(t, err)
+	_, js := startJSNATSServer(t)
 
 	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "test-idempotent"})
 	require.NoError(t, err)
@@ -139,13 +107,7 @@ func TestWriteVersion_IdempotentOnSameVersion(t *testing.T) {
 }
 
 func TestWriteVersion_UpdatesOnHigherVersion(t *testing.T) {
-	ns := startJSNATSServer(t)
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	defer nc.Close()
-
-	js, err := nc.JetStream()
-	require.NoError(t, err)
+	_, js := startJSNATSServer(t)
 
 	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "test-upgrade"})
 	require.NoError(t, err)
@@ -159,13 +121,7 @@ func TestWriteVersion_UpdatesOnHigherVersion(t *testing.T) {
 }
 
 func TestWriteVersion_NoOpOnLowerVersion(t *testing.T) {
-	ns := startJSNATSServer(t)
-	nc, err := nats.Connect(ns.ClientURL())
-	require.NoError(t, err)
-	defer nc.Close()
-
-	js, err := nc.JetStream()
-	require.NoError(t, err)
+	_, js := startJSNATSServer(t)
 
 	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "test-downgrade"})
 	require.NoError(t, err)

spinifex/utils/nats.go

@@ -151,6 +151,22 @@ func NATSScatterGather[Out any](conn *nats.Conn, subject string, input any, time
 	return nil, fmt.Errorf("scatter-gather timeout: no responses received for %s", subject)
 }
 
+// PublishEvent marshals event as JSON and publishes it to the given NATS topic.
+// Errors are logged but not returned (fire-and-forget). A nil connection is a no-op.
+func PublishEvent(nc *nats.Conn, topic string, event any) {
+	if nc == nil {
+		return
+	}
+	data, err := json.Marshal(event)
+	if err != nil {
+		slog.Warn("Failed to marshal event", "topic", topic, "error", err)
+		return
+	}
+	if err := nc.Publish(topic, data); err != nil {
+		slog.Warn("Failed to publish event", "topic", topic, "error", err)
+	}
+}
+
 // AccountIDFromMsg extracts the caller's account ID from a NATS message header.
 // Returns the account ID, or empty string if the header is not set.
 func AccountIDFromMsg(msg *nats.Msg) string {

spinifex/utils/nats_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
@@ -13,21 +14,7 @@ import (
 
 func startTestNATSServer(t *testing.T) *server.Server {
 	t.Helper()
-
-	opts := &server.Options{
-		Host:   "127.0.0.1",
-		Port:   -1,
-		NoLog:  true,
-		NoSigs: true,
-	}
-
-	ns, err := server.NewServer(opts)
-	require.NoError(t, err)
-
-	go ns.Start()
-	require.True(t, ns.ReadyForConnections(5*time.Second), "NATS server failed to start")
-
-	t.Cleanup(func() { ns.Shutdown() })
+	ns, _ := testutil.StartTestNATS(t)
 	return ns
 }
 

spinifex/utils/pidfile.go

@@ -113,6 +113,19 @@ func RemovePidFileAt(dir string, name string) error {
 	return os.Remove(filepath.Join(dir, fmt.Sprintf("%s.pid", name)))
 }
 
+// ServiceStatus returns a human-readable status string for a service by
+// checking its PID file. If dir is empty, the default pidPath() is used.
+func ServiceStatus(dir, name string) (string, error) {
+	pid, err := ReadPidFileFrom(dir, name)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return "stopped", nil
+		}
+		return "", fmt.Errorf("read pid file: %w", err)
+	}
+	return fmt.Sprintf("running (pid: %d)", pid), nil
+}
+
 // StopProcessAt stops a process using a PID file in a specific directory.
 // If dir is empty, falls back to the default pidPath(). The PID file is
 // always removed, even if the process is already dead, to prevent stale

spinifex/utils/tags.go

@@ -1,9 +1,23 @@
 package utils
 
 import (
+	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
 )
 
+// MapToEC2Tags converts a map[string]string to a slice of EC2 Tag pointers.
+// Returns nil when the input map is empty.
+func MapToEC2Tags(m map[string]string) []*ec2.Tag {
+	if len(m) == 0 {
+		return nil
+	}
+	tags := make([]*ec2.Tag, 0, len(m))
+	for k, v := range m {
+		tags = append(tags, &ec2.Tag{Key: aws.String(k), Value: aws.String(v)})
+	}
+	return tags
+}
+
 // ExtractTags returns a map of tags from the TagSpecification that matches
 // the given resourceType. If no specification matches, the returned map is
 // empty (never nil).