usbip: make all-devices worker bookkeeping atomic

The desired set (assignment lock) and the worker map (workerAccess)
were updated in separate critical sections, so a concurrent resync
could observe one but not the other: the same busid ended up with two
workers (leaking the first cancel), and stop decisions ran against
half-applied state. allWorkers is now the single source of truth,
diffed, started, and stopped in one workerAccess section; workers
remove themselves on exit with an identity check and restart in place
if their busid became desired again meanwhile.

The per-busid active flag also became a count: during matched-mode
handoff two workers briefly reference one busid, and a failing worker's
deactivation must not erase the mark of the genuinely attached one
(which made ApplyAll detach live devices, oscillating).

service/usbip/client.go

@@ -27,15 +27,23 @@ type ClientService struct {
 	serverAddr M.Socksaddr
 	host       ImportHost
 
+	// workerAccess guards assignedWorkers and allWorkers. All-devices
+	// worker bookkeeping (diffing the desired set, starting, stopping,
+	// self-removal) happens in single workerAccess critical sections;
+	// assignment.access may be taken nested inside, never the reverse.
 	assignment      *clientAssignment
 	workerAccess    sync.Mutex
 	assignedWorkers []*clientAssignedWorker
-	allWorkers      map[string]context.CancelFunc
+	allWorkers      map[string]*clientRemoteWorker
 
 	remoteAccess  sync.Mutex
 	remoteDevices map[string]ControlDeviceInfo
 }
 
+type clientRemoteWorker struct {
+	cancel context.CancelFunc
+}
+
 func NewClientService(ctx context.Context, logger log.ContextLogger, tag string, options option.USBIPClientServiceOptions) (adapter.Service, error) {
 	for i, m := range options.Devices {
 		if m.IsZero() {
@@ -66,7 +74,7 @@ func NewClientService(ctx context.Context, logger log.ContextLogger, tag string,
 		serverAddr: options.ServerOptions.Build(),
 		host:       host,
 		assignment: newClientAssignment(options.Devices),
-		allWorkers: make(map[string]context.CancelFunc),
+		allWorkers: make(map[string]*clientRemoteWorker),
 	}, nil
 }
 

service/usbip/client_assignment.go

@@ -17,9 +17,12 @@ type clientAssignment struct {
 	matchedKnownKeys map[string]DeviceKey
 
 	allDesired map[string]struct{}
-	registered map[string]struct{}
 
-	activeBusIDs map[string]struct{}
+	// activeBusIDs counts attach attempts per busid. A count, not a
+	// flag: during matched-mode handoff two workers briefly reference
+	// the same busid, and a failing worker's deactivation must not
+	// erase the mark of the worker that is genuinely attached.
+	activeBusIDs map[string]int
 }
 
 func newClientAssignment(matches []option.USBIPDeviceMatch) *clientAssignment {
@@ -42,8 +45,7 @@ func newClientAssignment(matches []option.USBIPDeviceMatch) *clientAssignment {
 	return &clientAssignment{
 		targets:      targets,
 		allDesired:   make(map[string]struct{}),
-		registered:   make(map[string]struct{}),
-		activeBusIDs: make(map[string]struct{}),
+		activeBusIDs: make(map[string]int),
 	}
 }
 
@@ -66,10 +68,21 @@ func (a *clientAssignment) SetActive(busid string, active bool) {
 	a.access.Lock()
 	defer a.access.Unlock()
 	if active {
-		a.activeBusIDs[busid] = struct{}{}
-	} else {
-		delete(a.activeBusIDs, busid)
+		a.activeBusIDs[busid]++
+		return
 	}
+	count := a.activeBusIDs[busid]
+	if count <= 1 {
+		delete(a.activeBusIDs, busid)
+	} else {
+		a.activeBusIDs[busid] = count - 1
+	}
+}
+
+func (a *clientAssignment) IsActive(busid string) bool {
+	a.access.Lock()
+	defer a.access.Unlock()
+	return a.activeBusIDs[busid] > 0
 }
 
 func (a *clientAssignment) ApplyMatched(entries []DeviceEntry, knownKeys map[string]DeviceKey) (next []string, previous []string) {
@@ -90,44 +103,15 @@ func (a *clientAssignment) ApplyMatched(entries []DeviceEntry, knownKeys map[str
 	return nextAssigned, prev
 }
 
-func (a *clientAssignment) ApplyAll(entries []DeviceEntry) (start []string, stop []string) {
-	desired := make(map[string]struct{}, len(entries))
-	for i := range entries {
-		busid := entries[i].Info.BusIDString()
-		if busid == "" {
-			continue
-		}
-		desired[busid] = struct{}{}
-	}
+func (a *clientAssignment) SetAllDesired(desired map[string]struct{}) {
 	a.access.Lock()
-	defer a.access.Unlock()
 	a.allDesired = desired
-	for busid := range a.registered {
-		if _, ok := desired[busid]; ok {
-			continue
-		}
-		if _, active := a.activeBusIDs[busid]; active {
-			continue
-		}
-		stop = append(stop, busid)
-		delete(a.registered, busid)
-	}
-	for busid := range desired {
-		if _, ok := a.registered[busid]; ok {
-			continue
-		}
-		start = append(start, busid)
-		a.registered[busid] = struct{}{}
-	}
-	return start, stop
+	a.access.Unlock()
 }
 
 func (a *clientAssignment) IsRetryDesired(busid string) bool {
 	a.access.Lock()
 	defer a.access.Unlock()
-	if _, registered := a.registered[busid]; !registered {
-		return false
-	}
 	_, desired := a.allDesired[busid]
 	return desired
 }
@@ -276,7 +260,7 @@ func (a *clientAssignment) activeCurrentAssignmentsLocked(current []string, know
 		if _, ok := knownKeys[busid]; !ok {
 			continue
 		}
-		if _, active := a.activeBusIDs[busid]; !active {
+		if a.activeBusIDs[busid] == 0 {
 			continue
 		}
 		if activeCurrent == nil {

service/usbip/client_shared.go

@@ -292,28 +292,52 @@ func (c *ClientService) applyRemoteDeviceState(devices []ControlDeviceInfo) {
 	c.applyMatchedExportsWithRetained(availableEntries, knownKeys)
 }
 
+// applyRemoteExports reconciles the all-devices worker set against a
+// devlist/snapshot in one workerAccess critical section. allWorkers is
+// the single source of truth for running workers; diffing and mutating
+// it under separate locks let a concurrent sync interleave and start a
+// second worker for the same busid.
 func (c *ClientService) applyRemoteExports(entries []DeviceEntry) {
-	start, stop := c.assignment.ApplyAll(entries)
-
-	c.workerAccess.Lock()
-	stopCancels := make([]context.CancelFunc, 0, len(stop))
-	for _, busid := range stop {
-		cancel, ok := c.allWorkers[busid]
-		if !ok {
+	desired := make(map[string]struct{}, len(entries))
+	for i := range entries {
+		busid := entries[i].Info.BusIDString()
+		if busid == "" {
 			continue
 		}
-		stopCancels = append(stopCancels, cancel)
+		desired[busid] = struct{}{}
+	}
+
+	c.workerAccess.Lock()
+	c.assignment.SetAllDesired(desired)
+	var stopCancels []context.CancelFunc
+	for busid, worker := range c.allWorkers {
+		if _, wanted := desired[busid]; wanted {
+			continue
+		}
+		// A busy device is omitted from devlists; never stop the worker
+		// that is the reason it is busy.
+		if c.assignment.IsActive(busid) {
+			continue
+		}
+		stopCancels = append(stopCancels, worker.cancel)
 		delete(c.allWorkers, busid)
 	}
+	var start []string
+	for busid := range desired {
+		if _, exists := c.allWorkers[busid]; exists {
+			continue
+		}
+		start = append(start, busid)
+	}
+	slices.Sort(start)
+	for _, busid := range start {
+		c.startRemoteBusIDWorkerLocked(busid)
+	}
 	c.workerAccess.Unlock()
 
 	for _, cancel := range stopCancels {
 		cancel()
 	}
-	slices.Sort(start)
-	for _, busid := range start {
-		c.startRemoteBusIDWorker(busid, busid)
-	}
 }
 
 func (c *ClientService) applyMatchedExportsWithRetained(entries []DeviceEntry, knownKeys map[string]DeviceKey) {
@@ -391,29 +415,37 @@ func (w *clientAssignedWorker) setDesiredBusID(busid string) {
 	w.updates <- busid
 }
 
-func (c *ClientService) startRemoteBusIDWorker(busid, description string) {
+// startRemoteBusIDWorkerLocked must run under workerAccess. The worker
+// removes itself from allWorkers when it exits on its own (export
+// disappeared); the identity check keeps a stop-and-restart for the
+// same busid from deleting its successor. If the busid became desired
+// again while the worker was deciding to exit, restart it immediately —
+// the next snapshot may be far away.
+func (c *ClientService) startRemoteBusIDWorkerLocked(busid string) {
 	runCtx, cancel := context.WithCancel(c.ctx)
-
-	c.workerAccess.Lock()
-	c.allWorkers[busid] = cancel
-	c.workerAccess.Unlock()
-
+	worker := &clientRemoteWorker{cancel: cancel}
+	c.allWorkers[busid] = worker
 	go func() {
-		c.runBusIDLoop(runCtx, busid, description)
+		c.runBusIDLoop(runCtx, busid, busid)
+		cancel()
+		c.workerAccess.Lock()
+		if c.allWorkers[busid] == worker {
+			delete(c.allWorkers, busid)
+			if c.ctx.Err() == nil && c.assignment.IsRetryDesired(busid) {
+				c.startRemoteBusIDWorkerLocked(busid)
+			}
+		}
+		c.workerAccess.Unlock()
 	}()
 }
 
 func (c *ClientService) stopAllWorkers() {
-	c.assignment.access.Lock()
-	c.assignment.registered = make(map[string]struct{})
-	c.assignment.access.Unlock()
-
 	c.workerAccess.Lock()
 	cancels := make([]context.CancelFunc, 0, len(c.allWorkers))
-	for _, cancel := range c.allWorkers {
-		cancels = append(cancels, cancel)
+	for _, worker := range c.allWorkers {
+		cancels = append(cancels, worker.cancel)
 	}
-	c.allWorkers = make(map[string]context.CancelFunc)
+	c.allWorkers = make(map[string]*clientRemoteWorker)
 	c.workerAccess.Unlock()
 
 	for _, cancel := range cancels {