fix(ebs): repair QMP reconnect ordering + benign DEVICE_DELETED timeout; VolumeInUse mapping (#488)

PR #487 turned attach/detach flaky: reconnectQMP dialed a fresh QMP
client before closing the old connection. QEMU's monitor is a
single-client `server,nowait` listener, so while the old conn holds the
slot the redial gets no greeting, blocks ~30s, and fails, leaving
q.Dead permanently true and every later attach/detach ServerInternal.
Fix the ordering (close, then dial via dialQMPWithRetry to absorb the
brief post-close ECONNREFUSED window).

waitForDeviceDeletedEvent marked q.Dead on any decode error, including
a benign read-deadline timeout — routine, since DEVICE_DELETED is
frequently pre-swallowed by device_del's own reply loop. The
underlying connection survives a timeout, but encoding/json.Decoder
permanently caches read errors on itself, so on a timeout the Decoder
is swapped for a fresh one over the same conn instead of forcing a
full QMP reconnect; q.Dead is now only set on a genuine decode error.

Also: an AttachVolume re-attach to the same instance with a different
device now returns VolumeInUse (matching AWS) instead of
InvalidParameterValue.

spinifex/daemon/daemon_handlers_volume.go

@@ -66,7 +66,9 @@ func (d *Daemon) handleAttachVolume(ctx context.Context, msg *nats.Msg, command
 					"volumeId", volumeID, "instanceId", command.ID,
 					"requestedDevice", command.AttachVolumeData.Device,
 					"attachedDevice", volCfg.VolumeMetadata.DeviceName)
-				respondWithError(msg, attachDetachErrorCode(vm.ErrVolumeDeviceMismatch))
+				// AWS returns VolumeInUse (not InvalidParameterValue) when a
+				// re-attach targets a device other than the one already in use.
+				respondWithError(msg, awserrors.ErrorVolumeInUse)
 				return
 			}
 

spinifex/daemon/daemon_handlers_volume_test.go

@@ -259,8 +259,8 @@ func TestAttachVolume_InUseDifferentInstance(t *testing.T) {
 
 // TestAttachVolume_IdempotentSameInstance_DeviceMismatch verifies that a
 // same-instance re-attach requesting a DIFFERENT device than the one
-// already attached is treated as a real CSI conflict (InvalidParameterValue
-// via vm.ErrVolumeDeviceMismatch), not silently echoed back or accepted.
+// already attached is treated as a real CSI conflict — AWS returns
+// VolumeInUse for this case — not silently echoed back or accepted.
 func TestAttachVolume_IdempotentSameInstance_DeviceMismatch(t *testing.T) {
 	daemon, store := createFullTestDaemonWithStore(t, sharedNATSURL)
 
@@ -312,5 +312,5 @@ func TestAttachVolume_IdempotentSameInstance_DeviceMismatch(t *testing.T) {
 		5*time.Second,
 	)
 	require.NoError(t, err)
-	assert.Contains(t, string(resp.Data), awserrors.ErrorInvalidParameterValue)
+	assert.Contains(t, string(resp.Data), awserrors.ErrorVolumeInUse)
 }

spinifex/vm/lifecycle.go

@@ -870,7 +870,17 @@ func reconnectQMP(q *qmp.QMPClient, instanceID string) error {
 	if q.Path == "" {
 		return fmt.Errorf("QMP client has no socket path")
 	}
-	fresh, err := qmp.NewQMPClient(q.Path)
+
+	// QEMU's monitor is a single-client `server,nowait` listener: the old
+	// connection must be closed before redialing, or it still holds the
+	// only client slot and the fresh dial gets no greeting, blocking until
+	// NewQMPClient's read deadline expires. dialQMPWithRetry absorbs the
+	// brief post-close ECONNREFUSED window while the listener re-arms.
+	if q.Conn != nil {
+		_ = q.Conn.Close()
+	}
+
+	fresh, err := dialQMPWithRetry(q.Path)
 	if err != nil {
 		return fmt.Errorf("redial QMP socket %s: %w", q.Path, err)
 	}
@@ -903,9 +913,6 @@ func reconnectQMP(q *qmp.QMPClient, instanceID string) error {
 	}
 	_ = fresh.Conn.SetReadDeadline(time.Time{})
 
-	if q.Conn != nil {
-		_ = q.Conn.Close()
-	}
 	q.Conn = fresh.Conn
 	q.Decoder = fresh.Decoder
 	q.Encoder = fresh.Encoder

spinifex/vm/lifecycle_test.go

@@ -1007,6 +1007,98 @@ func TestSendQMPCommand_NoPathCannotReconnect(t *testing.T) {
 	assert.Contains(t, err.Error(), "no socket path")
 }
 
+// startSingleClientQMPListener models a real `-qmp unix:...,server,nowait`
+// QEMU monitor: single-threaded, serving exactly one connection at a time.
+// Unlike startWorkingQMPListener (which spawns a goroutine per accepted
+// connection and greets all of them concurrently, hiding the
+// reconnect-ordering bug), Accept is only called again once the current
+// connection's serve loop returns — so a second connection dialed while the
+// first is still open sits unaccepted (and ungreeted) in the kernel backlog,
+// exactly like a real single-client monitor, with no lock/goroutine race.
+func startSingleClientQMPListener(t *testing.T) (sockPath string, stop func()) {
+	t.Helper()
+	sockPath = filepath.Join(t.TempDir(), "qmp.sock")
+	ln, err := net.Listen("unix", sockPath)
+	require.NoError(t, err)
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for {
+			conn, err := ln.Accept()
+			if err != nil {
+				return
+			}
+			serveSingleClientQMPConn(conn)
+		}
+	}()
+
+	stopped := false
+	stop = func() {
+		if stopped {
+			return
+		}
+		stopped = true
+		_ = ln.Close()
+		<-done
+	}
+	t.Cleanup(stop)
+	return sockPath, stop
+}
+
+// serveSingleClientQMPConn greets conn, then echoes {"return":{}} for every
+// command line until the connection errors or closes. It blocks for the
+// life of the connection, which is what makes startSingleClientQMPListener's
+// accept loop strictly serial.
+func serveSingleClientQMPConn(conn net.Conn) {
+	defer func() { _ = conn.Close() }()
+	greeting := `{"QMP":{"version":{"qemu":{"major":8,"minor":0}},"capabilities":[]}}` + "\n"
+	if _, err := conn.Write([]byte(greeting)); err != nil {
+		return
+	}
+	dec := json.NewDecoder(conn)
+	for {
+		var cmd map[string]any
+		if err := dec.Decode(&cmd); err != nil {
+			return
+		}
+		if _, err := conn.Write([]byte(`{"return":{}}` + "\n")); err != nil {
+			return
+		}
+	}
+}
+
+// TestSendQMPCommand_ReconnectsAgainstSingleClientServer is the regression
+// test for the reconnectQMP ordering bug (PR #487): against a real QEMU
+// `server,nowait` QMP monitor, only one client may hold a connection at a
+// time. A wedged client's reconnect must close its own stale connection
+// BEFORE dialing fresh, freeing the slot — dialing fresh first (the pre-fix
+// ordering) leaves the old connection occupying the only slot, so the new
+// dial gets no greeting and reconnect fails, leaving q.Dead permanently
+// true and every later attach/detach on the VM erroring.
+func TestSendQMPCommand_ReconnectsAgainstSingleClientServer(t *testing.T) {
+	sockPath, stop := startSingleClientQMPListener(t)
+	defer stop()
+
+	client, err := qmp.NewQMPClient(sockPath)
+	require.NoError(t, err)
+	defer func() { _ = client.Conn.Close() }()
+
+	// The old connection is still open (not yet closed) when Dead is set —
+	// exactly the state a wedged decode leaves behind, and exactly what the
+	// pre-fix bare-dial-before-close ordering could not recover from against
+	// a single-client monitor.
+	client.Dead = true
+
+	resp, err := sendQMPCommand(t.Context(), client, qmp.QMPCommand{Execute: "query-status"}, "i-single-client")
+	require.NoError(t, err, "reconnect must close the stale connection before redialing so the single-client monitor's slot is free")
+	require.NotNil(t, resp)
+	assert.False(t, client.Dead, "reconnect must clear Dead so later commands reuse the fresh stream")
+
+	_, err = sendQMPCommand(t.Context(), client, qmp.QMPCommand{Execute: "query-status"}, "i-single-client")
+	require.NoError(t, err, "a second command must succeed over the reconnected stream")
+}
+
 // TestRemoveStaleQMPSocket covers the SIGKILL-leftover unlink: a stale socket
 // inode is removed, and a missing file is a no-op (not an error).
 func TestRemoveStaleQMPSocket(t *testing.T) {

spinifex/vm/volumes.go

@@ -2,9 +2,11 @@ package vm
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"log/slog"
+	"net"
 	"strconv"
 	"strings"
 	"time"
@@ -428,13 +430,15 @@ func (m *Manager) DetachVolume(ctx context.Context, id, volumeID, device string,
 // while it runs) rather than going through sendQMPCommand, since the signal
 // we need is an async event, not a command reply.
 //
-// A timeout is treated the same as any other decode failure: q.Decoder is a
-// single shared *json.Decoder, and once Decode returns an error — including a
-// read-deadline timeout — that Decoder instance is permanently unusable, so
-// q.Dead is set exactly as sendQMPCommand does. The next QMP command
-// transparently redials via reconnectQMP. This is non-fatal either way:
-// DetachVolume falls back to the bounded blockdev-del retry, so a missed
-// event never blocks the caller forever.
+// A read-deadline timeout is benign (message boundary, connection stays
+// healthy once the deferred deadline reset below runs), so it must not force
+// the costly full QMP reconnect that q.Dead triggers. encoding/json.Decoder
+// caches read errors on itself permanently, though, so the *json.Decoder is
+// swapped in place instead — cheap, and safe since nothing was read off the
+// wire for the pending value. Any other decode error (EOF, malformed
+// message) leaves the stream position genuinely unknown, so q.Dead is set
+// exactly as sendQMPCommand does. Either way DetachVolume tolerates a miss
+// via the bounded blockdev-del retry, so this never blocks the caller.
 func waitForDeviceDeletedEvent(ctx context.Context, q *qmp.QMPClient, deviceID string, timeout time.Duration, instanceID string) error {
 	if q == nil || q.Conn == nil || q.Decoder == nil {
 		return fmt.Errorf("QMP client is not initialized")
@@ -461,7 +465,13 @@ func waitForDeviceDeletedEvent(ctx context.Context, q *qmp.QMPClient, deviceID s
 			} `json:"data"`
 		}
 		if err := q.Decoder.Decode(&msg); err != nil {
-			q.Dead = true
+			var netErr net.Error
+			if errors.As(err, &netErr) && netErr.Timeout() {
+				// Replace the poisoned Decoder but keep the healthy conn.
+				q.Decoder = json.NewDecoder(q.Conn)
+			} else {
+				q.Dead = true
+			}
 			return fmt.Errorf("decode error waiting for DEVICE_DELETED: %w", err)
 		}
 		if msg.Event != "DEVICE_DELETED" {

spinifex/vm/volumes_test.go

@@ -92,8 +92,8 @@ func (s *mockQMPServer) pushEvent(ev map[string]any) error {
 //
 // The listener keeps accepting connections for the life of the test (each
 // served on its own goroutine, greeting first) so a client-side reconnect —
-// e.g. reconnectQMP redialing after waitForDeviceDeletedEvent marks the
-// client Dead on timeout — is served exactly like the first connection.
+// e.g. reconnectQMP redialing after a genuine decode error marks the client
+// Dead — is served exactly like the first connection.
 func newMockQMPClientWithEvents(t *testing.T, responder func(qmp.QMPCommand, *mockQMPServer) map[string]any) (*qmp.QMPClient, func()) {
 	t.Helper()
 
@@ -1033,6 +1033,16 @@ func TestTryBlockdevDel(t *testing.T) {
 // pushed independently of any command/response exchange.
 func newMockQMPServerConn(t *testing.T) (*qmp.QMPClient, *json.Encoder, func()) {
 	t.Helper()
+	client, enc, _, cancel := newMockQMPServerConnRaw(t)
+	return client, enc, cancel
+}
+
+// newMockQMPServerConnRaw is newMockQMPServerConn plus the raw server-side
+// net.Conn, needed by tests that close the server side independently (e.g. to
+// force a genuine EOF decode error rather than a benign read-deadline
+// timeout).
+func newMockQMPServerConnRaw(t *testing.T) (*qmp.QMPClient, *json.Encoder, net.Conn, func()) {
+	t.Helper()
 
 	sockPath := filepath.Join(t.TempDir(), "qmp.sock")
 	ln, err := net.Listen("unix", sockPath)
@@ -1067,7 +1077,7 @@ func newMockQMPServerConn(t *testing.T) (*qmp.QMPClient, *json.Encoder, func())
 		_ = clientConn.Close()
 		_ = serverConn.Close()
 	}
-	return client, enc, cancel
+	return client, enc, serverConn, cancel
 }
 
 // TestWaitForDeviceDeletedEvent covers waitForDeviceDeletedEvent directly:
@@ -1110,18 +1120,33 @@ func TestWaitForDeviceDeletedEvent(t *testing.T) {
 		require.NoError(t, <-errCh)
 	})
 
-	t.Run("no event before timeout returns error and marks the client dead", func(t *testing.T) {
+	t.Run("no event before timeout returns error but leaves the client alive", func(t *testing.T) {
 		qmpClient, _, cancel := newMockQMPServerConn(t)
 		defer cancel()
 
 		err := waitForDeviceDeletedEvent(t.Context(), qmpClient, "vdisk-vol-1", 50*time.Millisecond, "i-1")
 		require.Error(t, err)
-		// q.Decoder is a single shared *json.Decoder: once Decode returns any
-		// error (including a deadline timeout) that instance is permanently
-		// unusable, so Dead must be set — exactly like sendQMPCommand — so the
-		// next QMP command redials via reconnectQMP instead of reusing it.
+		// A read-deadline timeout fires at a message boundary — nothing was
+		// partially consumed — and the deferred SetReadDeadline reset restores
+		// the connection, so this is benign: DEVICE_DELETED is frequently
+		// pre-swallowed by device_del's own reply loop, and poisoning the
+		// connection on every such miss was the wedge regression.
+		assert.False(t, qmpClient.Dead,
+			"a benign read-deadline timeout must not mark the client dead")
+	})
+
+	t.Run("genuine decode error marks the client dead", func(t *testing.T) {
+		qmpClient, _, serverConn, cancel := newMockQMPServerConnRaw(t)
+		defer cancel()
+
+		// Close only the server side: the client's next Decode fails with EOF,
+		// not a deadline timeout, so the stream position is genuinely unknown.
+		_ = serverConn.Close()
+
+		err := waitForDeviceDeletedEvent(t.Context(), qmpClient, "vdisk-vol-1", 5*time.Second, "i-1")
+		require.Error(t, err)
 		assert.True(t, qmpClient.Dead,
-			"a decode timeout must mark the client dead so the next command reconnects")
+			"a non-timeout decode error (EOF/connection close) must mark the client dead")
 	})
 
 	t.Run("nil client returns error", func(t *testing.T) {