usbip: serialize darwin submit scheduling and commit reconcile state

service/usbip/host.go

@@ -8,9 +8,11 @@ import (
 )
 
 // ExportHost lifecycle: Start → Reconcile* → Close. Reconcile may be
-// called many times; FinishImport runs after each data session ends so
-// the platform can do post-import cleanup (Linux: write -1 to
-// usbip_sockfd; Darwin: release stale-marked captures).
+// called many times; it returns the committed post-reconcile export
+// state, and callers must apply snapshot/released even when err != nil.
+// FinishImport runs after each data session ends so the platform can do
+// post-import cleanup (Linux: write -1 to usbip_sockfd; Darwin:
+// release stale-marked captures).
 type ExportHost interface {
 	Start(ctx context.Context) error
 	Close() error

service/usbip/host_darwin.go

@@ -356,7 +356,8 @@ type darwinServerDataSession struct {
 	device      *darwinUSBHostDevice
 	writeAccess sync.Mutex
 	access      sync.Mutex
-	pending     map[uint32]darwinServerPendingSubmit
+	pending     map[uint32]darwinServerSubmitState
+	endpoints   map[uint8]*darwinServerEndpointState
 	wg          sync.WaitGroup
 
 	done     chan struct{}
@@ -370,20 +371,33 @@ type darwinServerDataSession struct {
 	closeErr    error
 }
 
-type darwinServerPendingSubmit struct {
+type darwinServerSubmitState struct {
+	command  SubmitCommand
 	endpoint uint8
+	started  bool
 	unlinked bool
 	drained  chan struct{}
 }
 
+type darwinServerEndpointState struct {
+	active uint32
+	queued []uint32
+}
+
+type darwinServerNextSubmit struct {
+	sequence uint32
+	command  SubmitCommand
+}
+
 func newDarwinServerDataSession(ctx context.Context, logger log.ContextLogger, conn net.Conn, device *darwinUSBHostDevice) *darwinServerDataSession {
 	return &darwinServerDataSession{
-		ctx:     ctx,
-		logger:  logger,
-		conn:    conn,
-		device:  device,
-		pending: make(map[uint32]darwinServerPendingSubmit),
-		done:    make(chan struct{}),
+		ctx:       ctx,
+		logger:    logger,
+		conn:      conn,
+		device:    device,
+		pending:   make(map[uint32]darwinServerSubmitState),
+		endpoints: make(map[uint8]*darwinServerEndpointState),
+		done:      make(chan struct{}),
 	}
 }
 
@@ -458,32 +472,23 @@ func (s *darwinServerDataSession) serve() error {
 			if err != nil {
 				return err
 			}
-			s.trackSubmit(command.Header.SeqNum, commandEndpoint(command))
-			s.wg.Add(1)
-			go func() {
-				defer s.wg.Done()
-				response := s.handleSubmit(command)
-				if !s.finishSubmit(command.Header.SeqNum) {
-					return
-				}
-				s.writeAccess.Lock()
-				err := WriteSubmitResponse(s.conn, response)
-				s.writeAccess.Unlock()
-				if err != nil {
-					_ = s.conn.Close()
-				}
-			}()
+			next, shouldStart := s.enqueueSubmit(command)
+			if shouldStart {
+				s.startSubmit(next)
+			}
 		case CmdUnlink:
 			command, err := ReadUnlinkCommandBody(s.conn, header)
 			if err != nil {
 				return err
 			}
 			status := int32(0)
-			endpoint, drained, found := s.markSubmitUnlinked(command.SeqNum)
+			endpoint, drained, shouldAbort, found := s.unlinkSubmit(command.SeqNum)
 			if found {
-				abortErr := s.device.abortEndpoint(endpoint)
-				if abortErr != nil {
-					s.logger.Debug("abort endpoint 0x", hex8(endpoint), ": ", abortErr)
+				if shouldAbort {
+					abortErr := s.device.abortEndpoint(endpoint)
+					if abortErr != nil {
+						s.logger.Debug("abort endpoint 0x", hex8(endpoint), ": ", abortErr)
+					}
 				}
 				<-drained
 				status = usbipStatusECONNRESET
@@ -503,6 +508,57 @@ func (s *darwinServerDataSession) serve() error {
 	}
 }
 
+func (s *darwinServerDataSession) enqueueSubmit(command SubmitCommand) (darwinServerNextSubmit, bool) {
+	endpoint := submitScheduleEndpoint(command)
+	sequence := command.Header.SeqNum
+
+	s.access.Lock()
+	defer s.access.Unlock()
+
+	state := darwinServerSubmitState{
+		command:  command,
+		endpoint: endpoint,
+	}
+	endpointState, found := s.endpoints[endpoint]
+	if !found {
+		endpointState = &darwinServerEndpointState{}
+		s.endpoints[endpoint] = endpointState
+	}
+	if endpointState.active == 0 {
+		state.started = true
+		s.pending[sequence] = state
+		endpointState.active = sequence
+		return darwinServerNextSubmit{
+			sequence: sequence,
+			command:  command,
+		}, true
+	}
+	s.pending[sequence] = state
+	endpointState.queued = append(endpointState.queued, sequence)
+	return darwinServerNextSubmit{}, false
+}
+
+func (s *darwinServerDataSession) startSubmit(next darwinServerNextSubmit) {
+	s.wg.Add(1)
+	go func() {
+		defer s.wg.Done()
+
+		response := s.handleSubmit(next.command)
+		shouldSend, followUp, hasFollowUp := s.finishSubmit(next.sequence)
+		if shouldSend {
+			s.writeAccess.Lock()
+			err := WriteSubmitResponse(s.conn, response)
+			s.writeAccess.Unlock()
+			if err != nil {
+				_ = s.conn.Close()
+			}
+		}
+		if hasFollowUp {
+			s.startSubmit(followUp)
+		}
+	}()
+}
+
 func (s *darwinServerDataSession) handleSubmit(command SubmitCommand) SubmitResponse {
 	response := SubmitResponse{
 		Header: DataHeader{
@@ -589,64 +645,128 @@ func packIsoInResponseBuffer(buffer []byte, packets []IsoPacketDescriptor) []byt
 	return packed
 }
 
-func (s *darwinServerDataSession) trackSubmit(seq uint32, endpoint uint8) {
-	s.access.Lock()
-	defer s.access.Unlock()
-	s.pending[seq] = darwinServerPendingSubmit{endpoint: endpoint}
-}
+func (s *darwinServerDataSession) unlinkSubmit(seq uint32) (uint8, <-chan struct{}, bool, bool) {
+	var drained chan struct{}
 
-func (s *darwinServerDataSession) markSubmitUnlinked(seq uint32) (uint8, <-chan struct{}, bool) {
-	s.access.Lock()
-	defer s.access.Unlock()
-	pending, found := s.pending[seq]
-	if !found {
-		return 0, nil, false
-	}
-	pending.unlinked = true
-	if pending.drained == nil {
-		pending.drained = make(chan struct{})
-	}
-	s.pending[seq] = pending
-	return pending.endpoint, pending.drained, true
-}
-
-func (s *darwinServerDataSession) finishSubmit(seq uint32) bool {
 	s.access.Lock()
 	pending, found := s.pending[seq]
 	if !found {
 		s.access.Unlock()
-		return true
+		return 0, nil, false, false
+	}
+	if pending.drained == nil {
+		pending.drained = make(chan struct{})
+	}
+	drained = pending.drained
+	if !pending.started {
+		endpointState := s.endpoints[pending.endpoint]
+		if endpointState != nil {
+			endpointState.queued = removeQueuedSequence(endpointState.queued, seq)
+			if endpointState.active == 0 && len(endpointState.queued) == 0 {
+				delete(s.endpoints, pending.endpoint)
+			}
+		}
+		delete(s.pending, seq)
+		s.access.Unlock()
+		close(drained)
+		return pending.endpoint, drained, false, true
+	}
+	shouldAbort := !pending.unlinked
+	pending.unlinked = true
+	s.pending[seq] = pending
+	s.access.Unlock()
+	return pending.endpoint, drained, shouldAbort, true
+}
+
+func (s *darwinServerDataSession) finishSubmit(seq uint32) (bool, darwinServerNextSubmit, bool) {
+	var drained chan struct{}
+	var followUp darwinServerNextSubmit
+	var hasFollowUp bool
+
+	s.access.Lock()
+	pending, found := s.pending[seq]
+	if !found {
+		s.access.Unlock()
+		return true, darwinServerNextSubmit{}, false
+	}
+	endpointState := s.endpoints[pending.endpoint]
+	if endpointState != nil && endpointState.active == seq {
+		endpointState.active = 0
 	}
 	delete(s.pending, seq)
-	drained := pending.drained
+	if endpointState != nil {
+		for len(endpointState.queued) > 0 {
+			nextSequence := endpointState.queued[0]
+			endpointState.queued = endpointState.queued[1:]
+			nextPending, nextFound := s.pending[nextSequence]
+			if !nextFound {
+				continue
+			}
+			nextPending.started = true
+			s.pending[nextSequence] = nextPending
+			endpointState.active = nextSequence
+			followUp = darwinServerNextSubmit{
+				sequence: nextSequence,
+				command:  nextPending.command,
+			}
+			hasFollowUp = true
+			break
+		}
+		if endpointState.active == 0 && len(endpointState.queued) == 0 {
+			delete(s.endpoints, pending.endpoint)
+		}
+	}
+	drained = pending.drained
 	unlinked := pending.unlinked
 	s.access.Unlock()
 	if drained != nil {
 		close(drained)
 	}
-	return !unlinked
+	return !unlinked, followUp, hasFollowUp
 }
 
 func (s *darwinServerDataSession) abortPendingSubmits() {
+	var (
+		activeEndpoints []uint8
+		drained         []chan struct{}
+	)
+
 	s.access.Lock()
 	seen := make(map[uint8]struct{})
 	for seq, pending := range s.pending {
+		if !pending.started {
+			delete(s.pending, seq)
+			if pending.drained != nil {
+				drained = append(drained, pending.drained)
+			}
+			continue
+		}
 		if !pending.unlinked {
 			seen[pending.endpoint] = struct{}{}
 		}
 		pending.unlinked = true
 		s.pending[seq] = pending
 	}
-	endpoints := make([]uint8, 0, len(seen))
-	for endpoint := range seen {
-		endpoints = append(endpoints, endpoint)
+	for endpoint := range s.endpoints {
+		endpointState := s.endpoints[endpoint]
+		if endpointState != nil {
+			endpointState.queued = nil
+		}
 	}
-	slices.Sort(endpoints)
 	s.access.Unlock()
+
+	for _, drainedChannel := range drained {
+		close(drainedChannel)
+	}
+	activeEndpoints = make([]uint8, 0, len(seen))
+	for endpoint := range seen {
+		activeEndpoints = append(activeEndpoints, endpoint)
+	}
+	slices.Sort(activeEndpoints)
 	if s.device == nil {
 		return
 	}
-	for _, endpoint := range endpoints {
+	for _, endpoint := range activeEndpoints {
 		err := s.device.abortEndpoint(endpoint)
 		if err != nil {
 			s.logger.Debug("abort endpoint 0x", hex8(endpoint), ": ", err)
@@ -654,6 +774,23 @@ func (s *darwinServerDataSession) abortPendingSubmits() {
 	}
 }
 
+func removeQueuedSequence(queue []uint32, sequence uint32) []uint32 {
+	for index, current := range queue {
+		if current != sequence {
+			continue
+		}
+		return append(queue[:index], queue[index+1:]...)
+	}
+	return queue
+}
+
+func submitScheduleEndpoint(command SubmitCommand) uint8 {
+	if command.Header.Endpoint == 0 {
+		return 0
+	}
+	return commandEndpoint(command)
+}
+
 func commandEndpoint(command SubmitCommand) uint8 {
 	endpoint := uint8(command.Header.Endpoint & 0x0f)
 	if command.Header.Direction == USBIPDirIn {

service/usbip/host_linux.go

@@ -324,41 +324,79 @@ func (h *linuxExportHost) Reconcile(ctx context.Context, isBusy func(busid strin
 	h.access.Unlock()
 
 	plan := classifyLinuxReconcile(current, desired, isBusy)
-	toAdd := make([]*linuxExport, 0, len(plan.toBind))
-	for _, exp := range plan.toRelease {
-		err := h.releaseExport(exp, false)
-		if err != nil {
-			h.logger.Warn("release ", exp.busid, ": ", err)
+	committed := make(map[string]*linuxExport, len(current)+len(plan.toBind))
+	maps.Copy(committed, current)
+	var reconcileErrors []error
+
+	for _, busid := range plan.toStale {
+		exp, found := committed[busid]
+		if found {
+			staleExport := cloneLinuxExport(exp)
+			staleExport.stale = true
+			committed[busid] = staleExport
 		}
 	}
 
+	for _, exp := range plan.toRelease {
+		releaseErr := h.releaseExport(exp, false)
+		if releaseErr != nil {
+			h.logger.Warn("release ", exp.busid, ": ", releaseErr)
+			reconcileErrors = append(reconcileErrors, E.Cause(releaseErr, "release ", exp.busid))
+		}
+		var desiredDevice *sysfsDevice
+		desiredEntry, found := desired[exp.busid]
+		if found {
+			desiredDevice = &desiredEntry
+		}
+		resolved, resolveErr := h.resolveCommittedRelease(exp, desiredDevice)
+		if resolveErr != nil {
+			reconcileErrors = append(reconcileErrors, resolveErr)
+		}
+		if resolved == nil {
+			delete(committed, exp.busid)
+			continue
+		}
+		committed[exp.busid] = resolved
+	}
+
 	for busid, device := range plan.toBind {
-		exp, bindErr := h.bindOne(&device)
-		if bindErr != nil {
-			return h.snapshotSelf(), nil, E.Cause(bindErr, "bind ", busid)
+		_, found := committed[busid]
+		if found {
+			continue
 		}
-		toAdd = append(toAdd, exp)
+		previousDriver, probeErr := currentDriver(busid)
+		if probeErr != nil {
+			reconcileErrors = append(reconcileErrors, E.Cause(probeErr, "probe driver before bind ", busid))
+		}
+		exp, bindErr := h.bindOne(&device)
+		if bindErr == nil {
+			committed[busid] = exp
+			continue
+		}
+		reconcileErrors = append(reconcileErrors, E.Cause(bindErr, "bind ", busid))
+		resolved, resolveErr := h.resolveCommittedBind(busid, &device, previousDriver)
+		if resolveErr != nil {
+			reconcileErrors = append(reconcileErrors, resolveErr)
+		}
+		if resolved != nil {
+			committed[busid] = resolved
+		}
+	}
+
+	released := make([]string, 0, len(plan.released))
+	for _, busid := range plan.released {
+		exp, found := committed[busid]
+		if found && !exp.stale {
+			continue
+		}
+		released = append(released, busid)
 	}
 
 	h.access.Lock()
-	for _, busid := range plan.toStale {
-		exp, ok := h.exports[busid]
-		if ok {
-			exp.stale = true
-		}
-	}
-	for _, exp := range plan.toRelease {
-		currentExport, ok := h.exports[exp.busid]
-		if ok && currentExport == exp {
-			delete(h.exports, exp.busid)
-		}
-	}
-	for _, exp := range toAdd {
-		h.exports[exp.busid] = exp
-	}
+	h.exports = committed
 	h.access.Unlock()
 
-	return h.snapshotSelf(), plan.released, nil
+	return snapshotLinuxExports(committed), released, E.Errors(reconcileErrors...)
 }
 
 func (h *linuxExportHost) FinishImport(ctx context.Context, busid string) (bool, error) {
@@ -389,8 +427,12 @@ func (h *linuxExportHost) FinishImport(ctx context.Context, busid string) (bool,
 func (h *linuxExportHost) snapshotSelf() map[string]Export {
 	h.access.Lock()
 	defer h.access.Unlock()
-	out := make(map[string]Export, len(h.exports))
-	for busid, exp := range h.exports {
+	return snapshotLinuxExports(h.exports)
+}
+
+func snapshotLinuxExports(exports map[string]*linuxExport) map[string]Export {
+	out := make(map[string]Export, len(exports))
+	for busid, exp := range exports {
 		if exp.stale {
 			continue
 		}
@@ -399,6 +441,67 @@ func (h *linuxExportHost) snapshotSelf() map[string]Export {
 	return out
 }
 
+func cloneLinuxExport(exp *linuxExport) *linuxExport {
+	if exp == nil {
+		return nil
+	}
+	clone := *exp
+	clone.descriptor.Interfaces = slices.Clone(exp.descriptor.Interfaces)
+	clone.identity.Interfaces = slices.Clone(exp.identity.Interfaces)
+	return &clone
+}
+
+func (h *linuxExportHost) resolveCommittedRelease(exp *linuxExport, desired *sysfsDevice) (*linuxExport, error) {
+	if desired != nil {
+		resolved, found, err := h.probeDesiredBoundExport(exp.busid, desired, exp.managed, exp.originalDriver)
+		if err != nil {
+			return exp, err
+		}
+		if found {
+			return resolved, nil
+		}
+	}
+	driver, err := currentDriver(exp.busid)
+	if err != nil {
+		return exp, E.Cause(err, "probe driver ", exp.busid)
+	}
+	if driver != "usbip-host" {
+		return nil, nil
+	}
+	return exp, nil
+}
+
+func (h *linuxExportHost) resolveCommittedBind(busid string, desired *sysfsDevice, originalDriver string) (*linuxExport, error) {
+	if desired == nil {
+		return nil, nil
+	}
+	resolved, found, err := h.probeDesiredBoundExport(busid, desired, true, originalDriver)
+	if err != nil {
+		return nil, err
+	}
+	if !found {
+		return nil, nil
+	}
+	return resolved, nil
+}
+
+func (h *linuxExportHost) probeDesiredBoundExport(busid string, desired *sysfsDevice, managed bool, originalDriver string) (*linuxExport, bool, error) {
+	driver, err := currentDriver(busid)
+	if err != nil {
+		return nil, false, E.Cause(err, "probe driver ", busid)
+	}
+	if driver != "usbip-host" {
+		return nil, false, nil
+	}
+	descriptor, err := readSysfsDevice(busid, filepath.Join(sysBusUSBDevices, busid))
+	if err != nil {
+		descriptor = *desired
+	} else if !newLinuxExportIdentity(descriptor).Equal(newLinuxExportIdentity(*desired)) {
+		return nil, false, nil
+	}
+	return h.newExport(descriptor, managed, originalDriver), true, nil
+}
+
 func (h *linuxExportHost) bindOne(d *sysfsDevice) (*linuxExport, error) {
 	var (
 		exp *linuxExport

service/usbip/server.go

@@ -168,16 +168,13 @@ func (s *ServerService) reconcileAndBroadcast(notify bool) error {
 		return nil
 	}
 	snapshot, released, err := s.host.Reconcile(s.ctx, s.ledger.IsBusy)
-	if err != nil {
-		return err
-	}
 	s.ledger.ApplyHostSnapshot(snapshot, released)
 	if notify {
 		s.ledger.BroadcastIfChanged(s.ctx)
 	} else {
 		s.ledger.SeedBroadcastState(s.ctx)
 	}
-	return nil
+	return err
 }
 
 func (s *ServerService) handleStandardConn(conn net.Conn, header OpHeader) {