usbip: drain darwin watcher callbacks and wait for active sessions on close

`darwinExportHost.Events` closed its producer channel after destroying
the IOKit notification port, but `IONotificationPortDestroy` does not
synchronously drain the dispatch queue: a callback block already
executing or enqueued could still call `signal()` on the closed
channel, or resolve the `cgo.Handle` after `ref.Delete()`. Retain the
serial dispatch queue inside the watcher and `dispatch_sync` an empty
block before releasing it, so no native callback is in flight when
the Go side proceeds.

`ServerService.Close` cancelled the context and immediately closed the
`ExportHost`. On Darwin that calls `IOUSBHostDevice destroy` for every
capture while `darwinServerDataSession.handleSubmit` goroutines may
still be inside `IOUSBHostPipe` requests on the same device. Track
active sessions, close them explicitly, and wait for the per-session
cleanup goroutines via a WaitGroup before tearing down the host.

service/usbip/server.go

@@ -31,6 +31,11 @@ type ServerService struct {
 	ledger   *exportLedger
 
 	reconcileAccess sync.Mutex
+
+	sessionsAccess sync.Mutex
+	sessions       map[DataSession]struct{}
+	sessionsClosed bool
+	sessionsWG     sync.WaitGroup
 }
 
 func NewServerService(ctx context.Context, logger log.ContextLogger, tag string, options option.USBIPServerServiceOptions) (adapter.Service, error) {
@@ -48,13 +53,14 @@ func NewServerService(ctx context.Context, logger log.ContextLogger, tag string,
 	}
 	ctx, cancel := context.WithCancel(ctx)
 	return &ServerService{
-		Adapter: boxService.NewAdapter(C.TypeUSBIPServer, tag),
-		ctx:     ctx,
-		cancel:  cancel,
-		logger:  logger,
-		matches: options.Devices,
-		host:    host,
-		ledger:  newExportLedger(logger, importLeaseTTL, time.Now),
+		Adapter:  boxService.NewAdapter(C.TypeUSBIPServer, tag),
+		ctx:      ctx,
+		cancel:   cancel,
+		logger:   logger,
+		matches:  options.Devices,
+		host:     host,
+		ledger:   newExportLedger(logger, importLeaseTTL, time.Now),
+		sessions: make(map[DataSession]struct{}),
 		listener: listener.New(listener.Options{
 			Context: ctx,
 			Logger:  logger,
@@ -91,10 +97,25 @@ func (s *ServerService) Close() error {
 	if s.cancel != nil {
 		s.cancel()
 	}
+
+	s.sessionsAccess.Lock()
+	s.sessionsClosed = true
+	sessions := make([]DataSession, 0, len(s.sessions))
+	for session := range s.sessions {
+		sessions = append(sessions, session)
+	}
+	s.sessionsAccess.Unlock()
+
 	for _, conn := range s.ledger.CloseAllSubscribers() {
 		_ = conn.Close()
 	}
 	err := common.Close(common.PtrOrNil(s.listener))
+
+	for _, session := range sessions {
+		_ = session.Close()
+	}
+	s.sessionsWG.Wait()
+
 	s.reconcileAccess.Lock()
 	defer s.reconcileAccess.Unlock()
 	_ = s.host.Close()
@@ -301,15 +322,34 @@ func (s *ServerService) handleImportBusID(conn net.Conn, busid string, extended
 		s.tearDownPreparedSession(busid, session)
 		return false
 	}
+
+	s.sessionsAccess.Lock()
+	if s.sessionsClosed {
+		s.sessionsAccess.Unlock()
+		s.tearDownPreparedSession(busid, session)
+		return false
+	}
+	s.sessions[session] = struct{}{}
+	s.sessionsWG.Add(1)
+	s.sessionsAccess.Unlock()
+
 	err = session.Start()
 	if err != nil {
+		s.sessionsAccess.Lock()
+		delete(s.sessions, session)
+		s.sessionsAccess.Unlock()
+		s.sessionsWG.Done()
 		s.logger.Warn("start data session ", busid, ": ", err)
 		s.tearDownPreparedSession(busid, session)
 		return false
 	}
 	s.logger.Info("attached ", busid, " to remote ", conn.RemoteAddr())
 	go func() {
+		defer s.sessionsWG.Done()
 		<-session.Done()
+		s.sessionsAccess.Lock()
+		delete(s.sessions, session)
+		s.sessionsAccess.Unlock()
 		released, err := s.host.FinishImport(s.ctx, busid)
 		if err != nil {
 			s.logger.Debug("finish import ", busid, ": ", err)

service/usbip/usbhost_darwin.m

@@ -49,6 +49,7 @@ struct box_usbhost_device_watcher {
 	IONotificationPortRef port;
 	io_iterator_t matched;
 	io_iterator_t terminated;
+	void *queue;
 	uintptr_t ref;
 };
 
@@ -391,6 +392,7 @@ box_usbhost_device_watcher_t *box_usbhost_device_watcher_create(uintptr_t ref, c
 			return NULL;
 		}
 		dispatch_queue_t queue = dispatch_queue_create("io.nekohasekai.sing-box.usbhost-watch", DISPATCH_QUEUE_SERIAL);
+		watcher->queue = (void *)CFBridgingRetain(queue);
 		IONotificationPortSetDispatchQueue(watcher->port, queue);
 
 		CFMutableDictionaryRef matching = box_usbhost_device_matching_dictionary();
@@ -440,6 +442,15 @@ void box_usbhost_device_watcher_destroy(box_usbhost_device_watcher_t *watcher) {
 		IONotificationPortDestroy(watcher->port);
 		watcher->port = NULL;
 	}
+	if (watcher->queue != NULL) {
+		// Drain pending IOKit notification callbacks before releasing the
+		// cgo.Handle: the serial queue guarantees any block already
+		// enqueued (or executing) finishes before this empty block runs.
+		dispatch_queue_t queue = (__bridge dispatch_queue_t)watcher->queue;
+		dispatch_sync(queue, ^{});
+		CFRelease(watcher->queue);
+		watcher->queue = NULL;
+	}
 	free(watcher);
 }