usbip: fix darwin session close-ordering and lease/import reservation race

2026-05-15 16:31:21 +08:00
parent 37065daccf
commit cc83b9185a
3 changed files with 89 additions and 36 deletions
@@ -144,17 +144,20 @@ func (l *exportLedger) BroadcastIfChanged(ctx context.Context) bool {

 // TryReserveForImport runs Export.LeaseCheck outside the slow lock; the
 // busy mark is inserted only after a second availability re-check
-// confirms no goroutine raced in. The caller must pair every success
-// with a later ReleaseImport.
+// confirms no goroutine raced in. An outstanding lease counts as busy so
+// legacy OP_REQ_IMPORT cannot steal a slot a control client has already
+// reserved via IssueLease. The caller must pair every success with a
+// later ReleaseImport.
 func (l *exportLedger) TryReserveForImport(ctx context.Context, busid string) (Export, bool, string) {
 	l.slow.Lock()
 	export, found := l.exports[busid]
 	busy := l.busy[busid]
+	_, leased := l.leases[busid]
 	l.slow.Unlock()
 	if !found {
 		return nil, false, "unknown busid"
 	}
-	if busy {
+	if busy || leased {
 		return nil, false, deviceStateBusy
 	}
 	leaseOK, leaseReason := export.LeaseCheck(ctx)
@@ -170,6 +173,9 @@ func (l *exportLedger) TryReserveForImport(ctx context.Context, busid string) (E
 	if l.busy[busid] {
 		return nil, false, deviceStateBusy
 	}
+	if _, leasedNow := l.leases[busid]; leasedNow {
+		return nil, false, deviceStateBusy
+	}
 	l.busy[busid] = true
 	return export, true, ""
 }
@@ -185,7 +191,8 @@ func (l *exportLedger) ReleaseImport(ctx context.Context, busid string, removeEx
 }

 // IssueLease captures the seq generation at entry so a subsequent
-// ConsumeLease can reject stale leases issued before a topology change.
+// ConsumeLeaseAndReserve can reject stale leases issued before a
+// topology change.
 func (l *exportLedger) IssueLease(ctx context.Context, subID uint64, request controlLeaseRequest) controlLeaseResponse {
 	response := controlLeaseResponse{
 		BusID:       request.BusID,
@@ -268,33 +275,60 @@ func (l *exportLedger) IssueLease(ctx context.Context, subID uint64, request con
 	return response
 }

-// ConsumeLease has consume-on-read semantics: the entry is removed
-// regardless of outcome, except on mismatched nonce — which preserves
-// the lease for the legitimate holder.
-func (l *exportLedger) ConsumeLease(request ImportExtRequest) bool {
+// ConsumeLeaseAndReserve atomically validates a lease, removes its entry,
+// and marks the busid busy under a single slow-lock critical section so
+// no concurrent import can observe the consume-before-reserve gap. It
+// keeps the consume-on-read semantics of the old ConsumeLease: the entry
+// is removed on every outcome except a nonce/ID mismatch, which preserves
+// the lease for the legitimate holder. LeaseCheck is not re-run — the
+// lease itself attests that LeaseCheck passed at issue time, and the
+// generation equality test below confirms the export has not been
+// reconciled since. The caller must pair every success with a later
+// ReleaseImport.
+func (l *exportLedger) ConsumeLeaseAndReserve(request ImportExtRequest) (Export, bool, string) {
 	l.slow.Lock()
 	now := l.now()
 	l.cleanupExpiredLocked(now)
 	lease, found := l.leases[request.BusID]
 	if !found {
 		l.slow.Unlock()
-		return false
+		return nil, false, "lease not found"
 	}
 	if lease.ID != request.LeaseID || lease.ClientNonce != request.ClientNonce {
 		l.slow.Unlock()
-		return false
+		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"
+	}
+	if l.busy[request.BusID] {
+		delete(l.leases, request.BusID)
+		l.slow.Unlock()
+		return nil, false, deviceStateBusy
 	}
 	delete(l.leases, request.BusID)
-	leaseExpiry := lease.Expires
+	l.busy[request.BusID] = true
 	leaseGeneration := lease.Generation
 	l.slow.Unlock()
-	if !now.Before(leaseExpiry) {
-		return false
-	}
+
 	l.fast.Lock()
 	currentGeneration := l.seq
 	l.fast.Unlock()
-	return leaseGeneration == currentGeneration
+	if leaseGeneration != currentGeneration {
+		l.slow.Lock()
+		delete(l.busy, request.BusID)
+		l.slow.Unlock()
+		return nil, false, "lease stale"
+	}
+	return export, true, ""
 }

 func (l *exportLedger) cleanupExpiredLocked(now time.Time) {
@@ -355,12 +355,15 @@ type darwinServerDataSession struct {
 	pending     map[uint32]darwinServerPendingSubmit
 	wg          sync.WaitGroup

-	done      chan struct{}
-	doneOnce  sync.Once
-	runErr    error
-	startOnce sync.Once
-	closeOnce sync.Once
-	closeErr  error
+	done     chan struct{}
+	doneOnce sync.Once
+	runErr   error
+
+	stateAccess sync.Mutex
+	started     bool
+	closed      bool
+	closeOnce   sync.Once
+	closeErr    error
 }

 type darwinServerPendingSubmit struct {
@@ -388,9 +391,13 @@ func (s *darwinServerDataSession) Err() error {
 }

 func (s *darwinServerDataSession) Start() error {
-	s.startOnce.Do(func() {
-		go s.run()
-	})
+	s.stateAccess.Lock()
+	defer s.stateAccess.Unlock()
+	if s.started || s.closed {
+		return nil
+	}
+	s.started = true
+	go s.run()
 	return nil
 }

@@ -398,8 +405,15 @@ func (s *darwinServerDataSession) Close() error {
 	s.closeOnce.Do(func() {
 		s.closeErr = common.Close(s.conn)
 	})
-	s.markDone(nil)
-	<-s.done
+	s.stateAccess.Lock()
+	started := s.started
+	s.closed = true
+	s.stateAccess.Unlock()
+	if started {
+		<-s.done
+	} else {
+		s.markDone(nil)
+	}
 	return s.closeErr
 }

@@ -200,7 +200,7 @@ func (s *ServerService) handleStandardConn(conn net.Conn, header OpHeader) {
 			s.logger.Debug("read import body: ", err)
 			break
 		}
-		closeConn = !s.handleImportBusID(conn, busid, false)
+		closeConn = !s.handleImportBusID(conn, busid)
 	case OpReqImportExt:
 		closeConn = !s.handleImportExt(conn)
 	default:
@@ -282,25 +282,30 @@ func (s *ServerService) handleImportExt(conn net.Conn) bool {
 		s.logger.Debug("read import-ext body: ", err)
 		return false
 	}
-	if !s.ledger.ConsumeLease(request) {
-		s.logger.Info("import-ext rejected (invalid lease): ", request.BusID)
+	export, ok, reason := s.ledger.ConsumeLeaseAndReserve(request)
+	if !ok {
+		s.logger.Info("import-ext rejected (", request.BusID, ": ", reason, ")")
 		_ = WriteOpRepImport(conn, OpRepImportExt, OpStatusError, nil)
 		return false
 	}
-	return s.handleImportBusID(conn, request.BusID, true)
+	return s.handleImportReserved(conn, request.BusID, export, true)
 }

-func (s *ServerService) handleImportBusID(conn net.Conn, busid string, extended bool) bool {
-	opCode := uint16(OpRepImport)
-	if extended {
-		opCode = OpRepImportExt
-	}
+func (s *ServerService) handleImportBusID(conn net.Conn, busid string) bool {
 	export, ok, reason := s.ledger.TryReserveForImport(s.ctx, busid)
 	if !ok {
 		s.logger.Info("import rejected (", busid, ": ", reason, ")")
-		_ = WriteOpRepImport(conn, opCode, OpStatusError, nil)
+		_ = WriteOpRepImport(conn, OpRepImport, OpStatusError, nil)
 		return false
 	}
+	return s.handleImportReserved(conn, busid, export, false)
+}
+
+func (s *ServerService) handleImportReserved(conn net.Conn, busid string, export Export, extended bool) bool {
+	opCode := uint16(OpRepImport)
+	if extended {
+		opCode = OpRepImportExt
+	}
 	info, err := export.DeviceInfo(s.ctx)
 	if err != nil {
 		s.ledger.ReleaseImport(s.ctx, busid, false)