usbip: cover import-all retry refresh

service/usbip/client_linux.go

@@ -68,6 +68,7 @@ type ClientService struct {
 	assigned        []string
 	assignedWorkers []*clientAssignedWorker
 	allWorkers      map[string]*clientBusIDWorker
+	allDesired      map[string]struct{}
 
 	attachMu sync.Mutex // serializes vhci port pick + attach
 	wg       sync.WaitGroup
@@ -106,6 +107,7 @@ func NewClientService(ctx context.Context, logger log.ContextLogger, tag string,
 		matches:      options.Devices,
 		ops:          systemUSBIPOps,
 		allWorkers:   make(map[string]*clientBusIDWorker),
+		allDesired:   make(map[string]struct{}),
 		ports:        make(map[int]struct{}),
 		activeBusIDs: make(map[string]struct{}),
 	}, nil
@@ -289,7 +291,11 @@ func (c *ClientService) controlPingLoop(conn net.Conn, done <-chan struct{}) {
 }
 
 func (c *ClientService) syncRemoteState() error {
-	entries, err := c.fetchDevList(c.ctx)
+	return c.syncRemoteStateContext(c.ctx)
+}
+
+func (c *ClientService) syncRemoteStateContext(ctx context.Context) error {
+	entries, err := c.fetchDevList(ctx)
 	if err != nil {
 		return err
 	}
@@ -316,6 +322,7 @@ func (c *ClientService) applyRemoteExports(entries []DeviceEntry) {
 	}
 
 	c.stateMu.Lock()
+	c.allDesired = desired
 	stopWorkers := make([]*clientBusIDWorker, 0)
 	for busid, worker := range c.allWorkers {
 		if _, ok := desired[busid]; ok {
@@ -519,6 +526,10 @@ func (c *ClientService) runBusIDLoop(ctx context.Context, busid, description str
 		if err := ctx.Err(); err != nil {
 			return
 		}
+		if !c.shouldRetryBusID(ctx, busid) {
+			c.logger.Info("remote export ", busid, " disappeared; stopping import worker")
+			return
+		}
 		c.logger.Info("vhci port ", port, " released; reattaching ", busid)
 		if !sleepCtx(ctx, clientReconnectDelay) {
 			return
@@ -595,6 +606,9 @@ func (c *ClientService) watchPort(ctx context.Context, port int, busid string) {
 		case <-settleDeadline.C:
 			if !seenUsed {
 				c.logger.Warn("vhci port ", port, " never reached used state; reattaching ", busid)
+				if err := c.ops.vhciDetach(port); err != nil {
+					c.logger.Warn("detach port ", port, " (", busid, "): ", err)
+				}
 				return
 			}
 		case <-ticker.C:
@@ -668,6 +682,29 @@ func (c *ClientService) isBusIDActive(busid string) bool {
 	return exists
 }
 
+func (c *ClientService) shouldRetryBusID(ctx context.Context, busid string) bool {
+	if len(c.matches) != 0 {
+		return true
+	}
+	if err := c.syncRemoteStateContext(ctx); err != nil {
+		c.logger.Warn("refresh remote exports after releasing ", busid, ": ", err)
+		return true
+	}
+	return c.isBusIDRetryDesired(busid)
+}
+
+func (c *ClientService) isBusIDRetryDesired(busid string) bool {
+	c.stateMu.Lock()
+	defer c.stateMu.Unlock()
+	if _, registered := c.allWorkers[busid]; !registered {
+		return false
+	}
+	if _, desired := c.allDesired[busid]; desired {
+		return true
+	}
+	return false
+}
+
 func isBusIDOnlyMatch(m option.USBIPDeviceMatch) bool {
 	return m.BusID != "" && m.VendorID == 0 && m.ProductID == 0 && m.Serial == ""
 }

service/usbip/linux_test.go

@@ -35,6 +35,18 @@ func (testDialer) ListenPacket(context.Context, M.Socksaddr) (net.PacketConn, er
 	return nil, errors.New("unused")
 }
 
+type failingDialer struct {
+	err error
+}
+
+func (d failingDialer) DialContext(context.Context, string, M.Socksaddr) (net.Conn, error) {
+	return nil, d.err
+}
+
+func (d failingDialer) ListenPacket(context.Context, M.Socksaddr) (net.PacketConn, error) {
+	return nil, errors.New("unused")
+}
+
 type testDeviceStore struct {
 	mu       sync.Mutex
 	devices  map[string]sysfsDevice
@@ -415,6 +427,63 @@ func TestClientApplyRemoteExportsKeepsActiveBusIDWorker(t *testing.T) {
 	require.NotContains(t, client.allWorkers, "1-1")
 }
 
+func TestClientShouldRetryBusIDRefreshesImportAllState(t *testing.T) {
+	t.Parallel()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	server := &ServerService{
+		ctx:         ctx,
+		cancel:      cancel,
+		logger:      newTestLogger(),
+		exports:     make(map[string]serverExport),
+		controlSubs: make(map[uint64]*serverControlConn),
+		ops:         newTestUSBIPOps(t),
+	}
+	serverAddr, closeServer := startDispatchServer(t, server)
+	defer closeServer()
+
+	canceled := false
+	client := &ClientService{
+		ctx:          context.Background(),
+		logger:       newTestLogger(),
+		dialer:       testDialer{},
+		serverAddr:   serverAddr,
+		allWorkers:   map[string]*clientBusIDWorker{"1-1": {cancel: func() { canceled = true }}},
+		allDesired:   map[string]struct{}{"1-1": {}},
+		activeBusIDs: make(map[string]struct{}),
+		ops:          newTestUSBIPOps(t),
+	}
+
+	require.False(t, client.shouldRetryBusID(context.Background(), "1-1"))
+	require.True(t, canceled)
+	require.NotContains(t, client.allWorkers, "1-1")
+	require.Empty(t, client.allDesired)
+}
+
+func TestClientShouldRetryBusIDKeepsRetryOnRefreshFailure(t *testing.T) {
+	t.Parallel()
+
+	expectedErr := errors.New("devlist unavailable")
+	canceled := false
+	client := &ClientService{
+		ctx:          context.Background(),
+		logger:       newTestLogger(),
+		dialer:       failingDialer{err: expectedErr},
+		serverAddr:   M.ParseSocksaddrHostPort("127.0.0.1", 3240),
+		allWorkers:   map[string]*clientBusIDWorker{"1-1": {cancel: func() { canceled = true }}},
+		allDesired:   map[string]struct{}{"1-1": {}},
+		activeBusIDs: make(map[string]struct{}),
+		ops:          newTestUSBIPOps(t),
+	}
+
+	require.True(t, client.shouldRetryBusID(context.Background(), "1-1"))
+	require.False(t, canceled)
+	require.Contains(t, client.allWorkers, "1-1")
+	require.Contains(t, client.allDesired, "1-1")
+}
+
 func TestAssignMatchedBusIDs(t *testing.T) {
 	t.Parallel()