From 8a4534faa855776ca599c10a1182fb1856db8f4a Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Thu, 9 Jul 2026 21:09:28 +1000 Subject: [PATCH] fix(ebs): repair QMP reconnect ordering + benign DEVICE_DELETED timeout; VolumeInUse mapping (#488) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 4 +- .../daemon/daemon_handlers_volume_test.go | 6 +- spinifex/vm/lifecycle.go | 15 ++- spinifex/vm/lifecycle_test.go | 92 +++++++++++++++++++ spinifex/vm/volumes.go | 26 ++++-- spinifex/vm/volumes_test.go | 43 +++++++-- 6 files changed, 161 insertions(+), 25 deletions(-) diff --git a/spinifex/daemon/daemon_handlers_volume.go b/spinifex/daemon/daemon_handlers_volume.go index 0ee731bd3..bb98f0b18 100644 --- a/spinifex/daemon/daemon_handlers_volume.go +++ b/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 } diff --git a/spinifex/daemon/daemon_handlers_volume_test.go b/spinifex/daemon/daemon_handlers_volume_test.go index 9068baf01..18ca0b88b 100644 --- a/spinifex/daemon/daemon_handlers_volume_test.go +++ b/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) } diff --git a/spinifex/vm/lifecycle.go b/spinifex/vm/lifecycle.go index b0f31b2a7..feb6b9b75 100644 --- a/spinifex/vm/lifecycle.go +++ b/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 diff --git a/spinifex/vm/lifecycle_test.go b/spinifex/vm/lifecycle_test.go index b0fbd2a58..5a20ff2d9 100644 --- a/spinifex/vm/lifecycle_test.go +++ b/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) { diff --git a/spinifex/vm/volumes.go b/spinifex/vm/volumes.go index 434d6b656..713aa6378 100644 --- a/spinifex/vm/volumes.go +++ b/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" { diff --git a/spinifex/vm/volumes_test.go b/spinifex/vm/volumes_test.go index ba9f9232f..d7812d2b9 100644 --- a/spinifex/vm/volumes_test.go +++ b/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) {