usbip: harden darwin response and stale-state handling

service/usbip/client_darwin.go

@@ -155,16 +155,22 @@ func (c *darwinVirtualController) readLoop() {
 		}
 		switch header.Command {
 		case RetSubmit:
-			c.pendingAccess.Lock()
-			pending, ok := c.pending[header.SeqNum]
-			c.pendingAccess.Unlock()
-			payloadDirection := header.Direction
-			if ok {
-				payloadDirection = pending.direction
+			pending, ok := c.pendingSubmit(header.SeqNum)
+			if !ok {
+				err = E.New("unexpected RET_SUBMIT seq ", header.SeqNum)
+				c.logger.Debug(err)
+				c.runErr = err
+				c.requestClose()
+				c.failPending()
+				return
 			}
-			response, err := ReadSubmitResponseBody(c.conn, header, payloadDirection)
+			response, err := ReadSubmitResponseBody(c.conn, header, pending.direction)
 			if err != nil {
 				c.logger.Debug("read RET_SUBMIT: ", err)
+				if !E.IsClosedOrCanceled(err) {
+					c.runErr = err
+				}
+				c.requestClose()
 				c.failPending()
 				return
 			}
@@ -173,11 +179,18 @@ func (c *darwinVirtualController) readLoop() {
 			_, err := ReadUnlinkResponseBody(c.conn, header)
 			if err != nil {
 				c.logger.Debug("read RET_UNLINK: ", err)
+				if !E.IsClosedOrCanceled(err) {
+					c.runErr = err
+				}
+				c.requestClose()
 				c.failPending()
 				return
 			}
 		default:
-			c.logger.Debug(fmt.Sprintf("unexpected USB/IP response 0x%08x", header.Command))
+			err = E.New(fmt.Sprintf("unexpected USB/IP response 0x%08x", header.Command))
+			c.logger.Debug(err)
+			c.runErr = err
+			c.requestClose()
 			c.failPending()
 			return
 		}
@@ -588,15 +601,11 @@ func (c *darwinVirtualController) sendSubmit(command SubmitCommand) (SubmitRespo
 	c.pendingAccess.Lock()
 	c.pending[seq] = darwinPendingSubmit{direction: command.Header.Direction, reply: reply}
 	c.pendingAccess.Unlock()
-	defer func() {
-		c.pendingAccess.Lock()
-		delete(c.pending, seq)
-		c.pendingAccess.Unlock()
-	}()
 	c.writeAccess.Lock()
 	err := WriteSubmitCommand(c.conn, command)
 	c.writeAccess.Unlock()
 	if err != nil {
+		c.removePendingSubmit(seq)
 		return SubmitResponse{}, err
 	}
 	select {
@@ -610,6 +619,19 @@ func (c *darwinVirtualController) sendSubmit(command SubmitCommand) (SubmitRespo
 	}
 }
 
+func (c *darwinVirtualController) pendingSubmit(seq uint32) (darwinPendingSubmit, bool) {
+	c.pendingAccess.Lock()
+	defer c.pendingAccess.Unlock()
+	pending, ok := c.pending[seq]
+	return pending, ok
+}
+
+func (c *darwinVirtualController) removePendingSubmit(seq uint32) {
+	c.pendingAccess.Lock()
+	delete(c.pending, seq)
+	c.pendingAccess.Unlock()
+}
+
 func (c *darwinVirtualController) deliverSubmit(response SubmitResponse) {
 	c.pendingAccess.Lock()
 	pending, ok := c.pending[response.Header.SeqNum]

service/usbip/darwin_integration_test.go

@@ -196,6 +196,82 @@ func TestDarwinUSBHostDeviceWatcherSmoke(t *testing.T) {
 	watcher.Close()
 }
 
+func TestDarwinControllerReadsINRetSubmitUsingPendingDirection(t *testing.T) {
+	t.Parallel()
+
+	clientConn, serverConn := net.Pipe()
+	defer serverConn.Close()
+
+	controller := newDarwinVirtualController(context.Background(), newTestLogger(t), clientConn, darwinFakeDeviceEntry().Info)
+	go controller.readLoop()
+	defer func() {
+		_ = controller.Close()
+		<-controller.Done()
+	}()
+
+	reply := make(chan SubmitResponse, 1)
+	controller.pendingAccess.Lock()
+	controller.pending[1] = darwinPendingSubmit{
+		direction: USBIPDirIn,
+		reply:     reply,
+	}
+	controller.pendingAccess.Unlock()
+
+	entry := darwinFakeDeviceEntry()
+	expectedPayload := []byte{0xde, 0xad, 0xbe, 0xef}
+	err := WriteSubmitResponse(serverConn, SubmitResponse{
+		Header: DataHeader{
+			Command:   RetSubmit,
+			SeqNum:    1,
+			DevID:     entry.Info.DevID(),
+			Direction: USBIPDirIn,
+			Endpoint:  1,
+		},
+		ActualLength:    int32(len(expectedPayload)),
+		NumberOfPackets: nonIsoPacketCount,
+		Buffer:          expectedPayload,
+	})
+	require.NoError(t, err)
+
+	select {
+	case response, ok := <-reply:
+		require.True(t, ok)
+		require.Equal(t, expectedPayload, response.Buffer)
+		require.Zero(t, response.Header.Direction)
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for RET_SUBMIT")
+	}
+
+	controller.pendingAccess.Lock()
+	_, found := controller.pending[1]
+	controller.pendingAccess.Unlock()
+	require.False(t, found)
+}
+
+func TestDarwinControllerRejectsUnknownRetSubmitImmediately(t *testing.T) {
+	t.Parallel()
+
+	clientConn, serverConn := net.Pipe()
+	defer serverConn.Close()
+
+	controller := newDarwinVirtualController(context.Background(), newTestLogger(t), clientConn, darwinFakeDeviceEntry().Info)
+	go controller.readLoop()
+
+	err := writeDataHeader(serverConn, DataHeader{
+		Command: RetSubmit,
+		SeqNum:  99,
+	})
+	require.NoError(t, err)
+
+	select {
+	case <-controller.Done():
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for controller to reject unexpected RET_SUBMIT")
+	}
+
+	require.ErrorContains(t, controller.Err(), "unexpected RET_SUBMIT seq 99")
+}
+
 func startDarwinFakeUSBIPServer(t *testing.T) *darwinFakeUSBIPServer {
 	t.Helper()
 

service/usbip/data_protocol.go

@@ -112,7 +112,10 @@ func ReadSubmitCommandBody(r io.Reader, header DataHeader) (SubmitCommand, error
 	return command, nil
 }
 
-func ReadSubmitResponseBody(r io.Reader, header DataHeader, payloadDirection uint32) (SubmitResponse, error) {
+// ReadSubmitResponseBody decodes a RET_SUBMIT body. requestDirection must be
+// the original CMD_SUBMIT direction, because the USB/IP response header zeroes
+// direction on the wire.
+func ReadSubmitResponseBody(r io.Reader, header DataHeader, requestDirection uint32) (SubmitResponse, error) {
 	var raw [28]byte
 	if _, err := io.ReadFull(r, raw[:]); err != nil {
 		return SubmitResponse{}, err
@@ -127,7 +130,7 @@ func ReadSubmitResponseBody(r io.Reader, header DataHeader, payloadDirection uin
 	}
 	copy(response.Setup[:], raw[20:28])
 	bufferLength := max(response.ActualLength, 0)
-	buffer, isoPackets, err := readUSBIPPayload(r, payloadDirection, bufferLength, response.NumberOfPackets, false)
+	buffer, isoPackets, err := readUSBIPPayload(r, requestDirection, bufferLength, response.NumberOfPackets, false)
 	if err != nil {
 		return SubmitResponse{}, err
 	}

service/usbip/export_ledger.go

@@ -516,7 +516,7 @@ func (l *exportLedger) snapshotDeviceState(ctx context.Context) []DeviceInfoV2 {
 	out := make([]DeviceInfoV2, 0, len(entries))
 	for _, e := range entries {
 		snapshot := e.export.Snapshot(ctx, e.busy)
-		if snapshot.State == deviceStateUnavailable && snapshot.Entry.Info.IDVendor == 0 {
+		if snapshot.State == deviceStateUnavailable && snapshot.Entry.Info.BusIDString() == "" {
 			continue
 		}
 		out = append(out, deviceInfoV2FromEntry(snapshot.Entry, snapshot.Backend, snapshot.StableID, snapshot.State, snapshot.RawStatus, snapshot.StatusReason))

service/usbip/export_ledger_darwin_test.go

@@ -0,0 +1,56 @@
+//go:build darwin && cgo
+
+package usbip
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestDarwinStaleExportBroadcastsUnavailableUpdate(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	ledger := newExportLedger(nil, time.Second, func() time.Time { return time.Unix(0, 0) })
+	entry := darwinFakeDeviceEntry()
+	export := &darwinExport{
+		busid:      entry.Info.BusIDString(),
+		registryID: 0x1234,
+		entry:      entry,
+	}
+
+	ledger.ApplyHostSnapshot(map[string]Export{export.busid: export}, nil)
+	ledger.SeedBroadcastState(ctx)
+
+	sub, _ := ledger.Subscribe(ctx, nil, controlCapabilities)
+	select {
+	case <-sub.send:
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for initial snapshot")
+	}
+
+	export.stale = true
+	export.pendingRegistryID = 0x5678
+
+	if !ledger.BroadcastIfChanged(ctx) {
+		t.Fatal("expected stale darwin export to broadcast an update")
+	}
+
+	select {
+	case message := <-sub.send:
+		require.Equal(t, controlFrameDeviceDelta, message.Frame.Type)
+
+		var delta controlDeviceDelta
+		require.NoError(t, unmarshalControlPayload(message.Payload, &delta))
+		require.Empty(t, delta.Removed)
+		require.Len(t, delta.Updated, 1)
+		require.Equal(t, export.busid, delta.Updated[0].BusID)
+		require.Equal(t, deviceStateUnavailable, delta.Updated[0].State)
+		require.Equal(t, "device replaced", delta.Updated[0].StatusReason)
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for unavailable update")
+	}
+}

service/usbip/host.go

@@ -42,8 +42,10 @@ type Export interface {
 }
 
 // ExportSnapshot Backend and StableID fields are populated
-// unconditionally so the caller can build an unavailable record even
-// when Entry could not be read.
+// unconditionally. Unavailable snapshots should keep a cached Entry,
+// including BusID, so the caller can broadcast a state transition
+// instead of removing the device outright; snapshots without a BusID
+// are treated as non-broadcastable.
 type ExportSnapshot struct {
 	Entry        DeviceEntry
 	Backend      string

service/usbip/host_darwin.go

@@ -286,13 +286,22 @@ func (e *darwinExport) LeaseIdentity() ExportLeaseIdentity {
 	return ExportLeaseIdentity(fmt.Sprintf("darwin:%016x", e.registryID))
 }
 
+func (e *darwinExport) staleReason() string {
+	if e.pendingRegistryID != 0 {
+		return "device replaced"
+	}
+	return "capture released"
+}
+
 func (e *darwinExport) Snapshot(ctx context.Context, busy bool) ExportSnapshot {
 	stableID := fmt.Sprintf("darwin-registry:%016x", e.registryID)
 	if e.stale {
 		return ExportSnapshot{
-			Backend:  backendIDDarwinIOKit,
-			StableID: stableID,
-			State:    deviceStateUnavailable,
+			Entry:        e.entry,
+			Backend:      backendIDDarwinIOKit,
+			StableID:     stableID,
+			State:        deviceStateUnavailable,
+			StatusReason: e.staleReason(),
 		}
 	}
 	state := deviceStateAvailable
@@ -309,7 +318,7 @@ func (e *darwinExport) Snapshot(ctx context.Context, busy bool) ExportSnapshot {
 
 func (e *darwinExport) LeaseCheck(ctx context.Context) (bool, string) {
 	if e.stale {
-		return false, "capture released"
+		return false, e.staleReason()
 	}
 	return true, ""
 }

service/usbip/usbhost_darwin.go

@@ -548,6 +548,9 @@ func darwinIOReturnToUSBIPStatus(status int32) int32 {
 	}
 }
 
+// darwinUSBIPStatusToCIStatus maps USB/IP transfer completion status to the
+// corresponding IOUSBHostCI completion status. This is only used for real
+// transfer completion, not for EndpointPause-driven state machine events.
 func darwinUSBIPStatusToCIStatus(status int32) int {
 	if status == 0 {
 		return int(C.IOUSBHostCIMessageStatusSuccess)
@@ -555,6 +558,8 @@ func darwinUSBIPStatusToCIStatus(status int32) int {
 	switch -status {
 	case int32(unix.EPIPE):
 		return int(C.IOUSBHostCIMessageStatusStallError)
+	case int32(unix.ENODEV), int32(unix.ECONNRESET):
+		return int(C.IOUSBHostCIMessageStatusOffline)
 	case int32(unix.ETIMEDOUT):
 		return int(C.IOUSBHostCIMessageStatusTimeout)
 	case int32(unix.ENOMEM):
@@ -563,8 +568,8 @@ func darwinUSBIPStatusToCIStatus(status int32) int {
 		return int(C.IOUSBHostCIMessageStatusBadArgument)
 	case int32(unix.EPERM):
 		return int(C.IOUSBHostCIMessageStatusNotPermitted)
-	case int32(unix.ECONNRESET):
-		return int(C.IOUSBHostCIMessageStatusEndpointStopped)
+	case int32(unix.EOVERFLOW):
+		return int(C.IOUSBHostCIMessageStatusOverrunError)
 	default:
 		return int(C.IOUSBHostCIMessageStatusError)
 	}