usbip: fix import-all, lease release broadcast, and Events contract

Drop the constructor guard so client import-all (Devices empty/omitted)
actually works, route ledger lease mutations through a new
mutateAndBroadcast helper so Unsubscribe-triggered release reaches all
subscribers, and tighten the Events contract: subscribe at Start so a
host failure aborts service start instead of silently disabling hotplug.

option/usbip.go

@@ -66,5 +66,7 @@ type USBIPServerServiceOptions struct {
 type USBIPClientServiceOptions struct {
 	ServerOptions
 	DialerOptions
+	// Devices selects which exported devices to import. Omit or leave
+	// empty to import every device the server exports.
 	Devices []USBIPDeviceMatch `json:"devices,omitempty"`
 }

service/usbip/client.go

@@ -26,7 +26,6 @@ type ClientService struct {
 	logger     log.ContextLogger
 	dialer     N.Dialer
 	serverAddr M.Socksaddr
-	matches    []option.USBIPDeviceMatch
 	host       ImportHost
 
 	assignment      *clientAssignment
@@ -43,9 +42,6 @@ type ClientService struct {
 }
 
 func NewClientService(ctx context.Context, logger log.ContextLogger, tag string, options option.USBIPClientServiceOptions) (adapter.Service, error) {
-	if len(options.Devices) == 0 {
-		return nil, E.New("devices: at least one match is required")
-	}
 	for i, m := range options.Devices {
 		if m.IsZero() {
 			return nil, E.New("devices[", i, "]: at least one of busid/vendor_id/product_id/serial is required")
@@ -73,7 +69,6 @@ func NewClientService(ctx context.Context, logger log.ContextLogger, tag string,
 		logger:     logger,
 		dialer:     outboundDialer,
 		serverAddr: options.ServerOptions.Build(),
-		matches:    options.Devices,
 		host:       host,
 		assignment: newClientAssignment(options.Devices),
 		allWorkers: make(map[string]context.CancelFunc),

service/usbip/export_ledger.go

@@ -192,13 +192,27 @@ func (l *exportLedger) TryReserveForImport(ctx context.Context, busid string) (E
 }
 
 func (l *exportLedger) ReleaseImport(ctx context.Context, busid string, removeExport bool) {
+	l.mutateAndBroadcast(ctx, func() bool {
+		delete(l.busy, busid)
+		if removeExport {
+			delete(l.exports, busid)
+		}
+		return true
+	})
+}
+
+// mutateAndBroadcast holds slow for the body, then broadcasts iff body
+// reported a change. This centralizes the "any reservedLocked-affecting
+// mutation must broadcast" invariant so future mutation sites cannot
+// silently skip it. body must NOT acquire fast (lock-ordering rule at
+// the top of this file).
+func (l *exportLedger) mutateAndBroadcast(ctx context.Context, body func() bool) {
 	l.slow.Lock()
-	delete(l.busy, busid)
-	if removeExport {
-		delete(l.exports, busid)
-	}
+	changed := body()
 	l.slow.Unlock()
-	l.BroadcastIfChanged(ctx)
+	if changed {
+		l.BroadcastIfChanged(ctx)
+	}
 }
 
 // IssueLease captures the current broadcast sequence as opaque metadata
@@ -301,96 +315,127 @@ func (l *exportLedger) ConsumeLeaseAndReserve(ctx context.Context, request Impor
 	var (
 		export   Export
 		identity ExportLeaseIdentity
+		phase1OK bool
+		reason   string
 	)
-	l.slow.Lock()
-	now := l.now()
-	l.cleanupExpiredLocked(now)
+	l.mutateAndBroadcast(ctx, func() bool {
+		now := l.now()
+		changed := l.cleanupExpiredLocked(now)
 
-	lease, found := l.leases[request.BusID]
-	if !found {
-		l.slow.Unlock()
-		return nil, false, "lease not found"
+		lease, found := l.leases[request.BusID]
+		if !found {
+			reason = "lease not found"
+			return changed
+		}
+		if lease.ID != request.LeaseID || lease.ClientNonce != request.ClientNonce {
+			reason = "lease mismatch"
+			return changed
+		}
+		if !now.Before(lease.Expires) {
+			delete(l.leases, request.BusID)
+			reason = "lease expired"
+			return true
+		}
+		current, stillExported := l.exports[request.BusID]
+		if !stillExported {
+			delete(l.leases, request.BusID)
+			reason = "unknown busid"
+			return true
+		}
+		identity = current.LeaseIdentity()
+		if identity != lease.Identity {
+			delete(l.leases, request.BusID)
+			reason = "lease stale"
+			return true
+		}
+		if l.busy[request.BusID] {
+			delete(l.leases, request.BusID)
+			reason = deviceStateBusy
+			return true
+		}
+		export = current
+		phase1OK = true
+		return changed
+	})
+	if !phase1OK {
+		return nil, false, reason
 	}
-	if lease.ID != request.LeaseID || lease.ClientNonce != request.ClientNonce {
-		l.slow.Unlock()
-		return nil, false, "lease mismatch"
-	}
-	if !now.Before(lease.Expires) {
-		delete(l.leases, request.BusID)
-		l.slow.Unlock()
-		return nil, false, "lease expired"
-	}
-	export, stillExported := l.exports[request.BusID]
-	if !stillExported {
-		delete(l.leases, request.BusID)
-		l.slow.Unlock()
-		return nil, false, "unknown busid"
-	}
-	identity = export.LeaseIdentity()
-	if identity != lease.Identity {
-		delete(l.leases, request.BusID)
-		l.slow.Unlock()
-		return nil, false, "lease stale"
-	}
-	if l.busy[request.BusID] {
-		delete(l.leases, request.BusID)
-		l.slow.Unlock()
-		return nil, false, deviceStateBusy
-	}
-	l.slow.Unlock()
 
 	leaseOK, leaseReason := export.LeaseCheck(ctx)
 	if !leaseOK {
-		l.slow.Lock()
-		currentLease, exists := l.leases[request.BusID]
-		if exists && currentLease.ID == request.LeaseID && currentLease.ClientNonce == request.ClientNonce {
+		l.mutateAndBroadcast(ctx, func() bool {
+			currentLease, exists := l.leases[request.BusID]
+			if !exists {
+				return false
+			}
+			if currentLease.ID != request.LeaseID || currentLease.ClientNonce != request.ClientNonce {
+				return false
+			}
 			delete(l.leases, request.BusID)
-		}
-		l.slow.Unlock()
+			return true
+		})
 		return nil, false, leaseReason
 	}
 
-	l.slow.Lock()
-	defer l.slow.Unlock()
+	var (
+		finalExport Export
+		finalOK     bool
+	)
+	l.mutateAndBroadcast(ctx, func() bool {
+		now := l.now()
+		changed := l.cleanupExpiredLocked(now)
 
-	now = l.now()
-	l.cleanupExpiredLocked(now)
-
-	lease, found = l.leases[request.BusID]
-	if !found {
-		return nil, false, "lease not found"
-	}
-	if lease.ID != request.LeaseID || lease.ClientNonce != request.ClientNonce {
-		return nil, false, "lease mismatch"
-	}
-	if !now.Before(lease.Expires) {
+		lease, found := l.leases[request.BusID]
+		if !found {
+			reason = "lease not found"
+			return changed
+		}
+		if lease.ID != request.LeaseID || lease.ClientNonce != request.ClientNonce {
+			reason = "lease mismatch"
+			return changed
+		}
+		if !now.Before(lease.Expires) {
+			delete(l.leases, request.BusID)
+			reason = "lease expired"
+			return true
+		}
+		current, stillExported := l.exports[request.BusID]
+		if !stillExported {
+			delete(l.leases, request.BusID)
+			reason = "unknown busid"
+			return true
+		}
+		if lease.Identity != identity || current.LeaseIdentity() != identity {
+			delete(l.leases, request.BusID)
+			reason = "lease stale"
+			return true
+		}
+		if l.busy[request.BusID] {
+			delete(l.leases, request.BusID)
+			reason = deviceStateBusy
+			return true
+		}
 		delete(l.leases, request.BusID)
-		return nil, false, "lease expired"
+		l.busy[request.BusID] = true
+		finalExport = current
+		finalOK = true
+		return true
+	})
+	if !finalOK {
+		return nil, false, reason
 	}
-	current, stillExported := l.exports[request.BusID]
-	if !stillExported {
-		delete(l.leases, request.BusID)
-		return nil, false, "unknown busid"
-	}
-	if lease.Identity != identity || current.LeaseIdentity() != identity {
-		delete(l.leases, request.BusID)
-		return nil, false, "lease stale"
-	}
-	if l.busy[request.BusID] {
-		delete(l.leases, request.BusID)
-		return nil, false, deviceStateBusy
-	}
-	delete(l.leases, request.BusID)
-	l.busy[request.BusID] = true
-	return current, true, ""
+	return finalExport, true, ""
 }
 
-func (l *exportLedger) cleanupExpiredLocked(now time.Time) {
+func (l *exportLedger) cleanupExpiredLocked(now time.Time) bool {
+	changed := false
 	for busid, lease := range l.leases {
 		if !now.Before(lease.Expires) {
 			delete(l.leases, busid)
+			changed = true
 		}
 	}
+	return changed
 }
 
 // Subscribe enqueues a freshly computed snapshot to extension-capable
@@ -445,18 +490,24 @@ func (l *exportLedger) Subscribe(ctx context.Context, conn net.Conn, capabilitie
 }
 
 // Unsubscribe leaves the subscriber's send channel for the GC to
-// reclaim; the transport read loop has already exited.
-func (l *exportLedger) Unsubscribe(sub *exportSubscriber) {
+// reclaim; the transport read loop has already exited. Any leases the
+// subscriber held are released and the resulting state change is
+// broadcast so remaining subscribers see the busid become available
+// again.
+func (l *exportLedger) Unsubscribe(ctx context.Context, sub *exportSubscriber) {
 	l.fast.Lock()
 	delete(l.subs, sub.id)
 	l.fast.Unlock()
-	l.slow.Lock()
-	for busid, lease := range l.leases {
-		if lease.SubscriberID == sub.id {
-			delete(l.leases, busid)
+	l.mutateAndBroadcast(ctx, func() bool {
+		released := false
+		for busid, lease := range l.leases {
+			if lease.SubscriberID == sub.id {
+				delete(l.leases, busid)
+				released = true
+			}
 		}
-	}
-	l.slow.Unlock()
+		return released
+	})
 }
 
 // CloseAllSubscribers returns the underlying connections so the caller

service/usbip/export_ledger_test.go

@@ -121,6 +121,61 @@ func TestConsumeLeaseAndReserveRejectsUnavailableExport(t *testing.T) {
 	}
 }
 
+func TestUnsubscribeBroadcastsLeaseRelease(t *testing.T) {
+	ctx := context.Background()
+	ledger := newExportLedger(nil, time.Second, func() time.Time { return time.Unix(0, 0) })
+	exp := &testExport{busid: "1-1", vendorID: 0x1111, productID: 0x0001}
+
+	ledger.ApplyHostSnapshot(map[string]Export{exp.busid: exp}, nil)
+	ledger.SeedBroadcastState(ctx)
+
+	holder, _ := ledger.Subscribe(ctx, nil, controlCapabilities)
+	drainSubscriber(holder)
+
+	response := ledger.IssueLease(ctx, holder.id, controlLeaseRequest{BusID: exp.busid, ClientNonce: 7})
+	if response.ErrorCode != "" {
+		t.Fatalf("IssueLease failed: %s", response.ErrorMessage)
+	}
+	// Pretend a topology event flushed the busy state into the broadcast
+	// baseline, matching the real-world scenario where hotplug churn keeps
+	// l.state roughly in sync with reality. Without this the diff machinery
+	// in BroadcastIfChanged has no busy→available transition to emit.
+	ledger.BroadcastIfChanged(ctx)
+	drainSubscriber(holder)
+
+	observer, _ := ledger.Subscribe(ctx, nil, controlCapabilities)
+	drainSubscriber(observer)
+
+	ledger.Unsubscribe(ctx, holder)
+
+	select {
+	case msg := <-observer.send:
+		if msg.Frame.Type != controlFrameDeviceDelta {
+			t.Fatalf("expected device delta after lease release, got frame type %d", msg.Frame.Type)
+		}
+		var delta controlDeviceDelta
+		err := unmarshalControlPayload(msg.Payload, &delta)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if len(delta.Updated) != 1 || delta.Updated[0].State != deviceStateAvailable {
+			t.Fatalf("expected one Updated entry flipped to available, got %#v", delta)
+		}
+	default:
+		t.Fatal("expected observer to receive a delta after holder's lease was released")
+	}
+}
+
+func drainSubscriber(sub *exportSubscriber) {
+	for {
+		select {
+		case <-sub.send:
+		default:
+			return
+		}
+	}
+}
+
 func TestConsumeLeaseAndReserveMarksBusyOnSuccess(t *testing.T) {
 	ctx := context.Background()
 	ledger := newExportLedger(nil, time.Second, func() time.Time { return time.Unix(0, 0) })
@@ -172,11 +227,15 @@ func (e *testExport) Snapshot(ctx context.Context, busy bool) ExportSnapshot {
 	if onSnapshot != nil {
 		onSnapshot()
 	}
+	state := deviceStateAvailable
+	if busy {
+		state = deviceStateBusy
+	}
 	return ExportSnapshot{
 		Entry: DeviceEntry{
 			Info: e.deviceInfo(),
 		},
-		State: deviceStateAvailable,
+		State: state,
 	}
 }
 

service/usbip/host.go

@@ -18,8 +18,10 @@ type ExportHost interface {
 	Close() error
 	Reconcile(ctx context.Context, isReserved func(busid string) bool) (snapshot map[string]Export, released []string, err error)
 	FinishImport(ctx context.Context, busid string) (released bool, err error)
-	// Events returning (nil, nil) means "no native event source; rely
-	// on polling".
+	// Events returns a coalescing channel that signals topology
+	// changes; the channel is closed when ctx is cancelled. A non-nil
+	// error means the host could not subscribe and the server must not
+	// continue.
 	Events(ctx context.Context) (<-chan struct{}, error)
 }
 

service/usbip/server.go

@@ -81,6 +81,11 @@ func (s *ServerService) Start(stage adapter.StartStage) error {
 	if err != nil {
 		return err
 	}
+	events, err := s.host.Events(s.ctx)
+	if err != nil {
+		_ = s.host.Close()
+		return E.Cause(err, "subscribe topology events")
+	}
 	err = s.reconcileAndBroadcast(false)
 	if err != nil {
 		_ = s.host.Close()
@@ -92,7 +97,7 @@ func (s *ServerService) Start(stage adapter.StartStage) error {
 		return err
 	}
 	go s.acceptLoop(tcpListener)
-	go s.eventLoop()
+	go s.eventLoop(events)
 	return nil
 }
 
@@ -126,15 +131,7 @@ func (s *ServerService) Close() error {
 	return err
 }
 
-func (s *ServerService) eventLoop() {
-	events, err := s.host.Events(s.ctx)
-	if err != nil {
-		s.logger.Warn("subscribe topology events: ", err)
-		return
-	}
-	if events == nil {
-		return
-	}
+func (s *ServerService) eventLoop(events <-chan struct{}) {
 	for {
 		select {
 		case <-s.ctx.Done():
@@ -144,7 +141,7 @@ func (s *ServerService) eventLoop() {
 				return
 			}
 		}
-		err = s.reconcileAndBroadcast(true)
+		err := s.reconcileAndBroadcast(true)
 		if err != nil {
 			s.logger.Warn("reconcile exports: ", err)
 		}
@@ -231,7 +228,7 @@ func (s *ServerService) handleControlConn(conn net.Conn) {
 	}
 	capabilities := hello.Capabilities & controlCapabilities
 	sub, seq := s.ledger.Subscribe(s.ctx, conn, capabilities)
-	defer s.ledger.Unsubscribe(sub)
+	defer s.ledger.Unsubscribe(s.ctx, sub)
 	err = writeControlMessage(conn, controlFrame{
 		Type:         controlFrameAck,
 		Version:      controlProtocolVersion,