Give each connection its own sequence-number space

USB/IP numbers requests per connection, and the Linux client opens one
connection per attached device, so every client starts again at 1. Request
tracking used a single process-wide table keyed on the bare sequence number,
so a second client's first request looked like a duplicate of the first
client's and was refused. Attaching two devices from one machine was enough:
each attach is its own connection, each starts at seqnum 1.

The refusal was silent. The duplicate check threw from outside the block
that turns a rejected request into a RET_SUBMIT carrying the error, so the
exception reached ServerCoordinator, which logged it and sent nothing. The
client sat waiting for a reply that was never coming, rather than being told
its request had been refused.

Requests are now keyed by devid and sequence number together. A connection
carries exactly one imported device, so devid distinguishes clients. The
claim also moved inside the error-response block, so a genuine duplicate —
same device, same number, still outstanding — is answered rather than met
with silence.

Both faults predate this branch: the previous code applied the identical
guard to the identical process-wide dictionary from the identical position
relative to the error handler.

Found by running two clients against two devices at once, which had never
been tried. Five of ten concurrent transfers failed before the change and
none after; eighteen transfers across three devices now pass. The FTDI and
the CP2102N were the pair, which is also the first exercise of the
IOUserSerial ownership path since the classifier changed.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017rmyikdjWveP99ZUCDLY89

Author: beriberikix

Sources/USBIPDCore/Protocol/USBSubmitProcessor.swift

@@ -25,16 +25,29 @@ public class USBSubmitProcessor {
         let device: USBDevice?
     }
 
-    private var activeURBs: [UInt32: TrackedURB] = [:]
+    /// Identifies a request. A sequence number alone does not.
+    ///
+    /// USB/IP numbers requests per connection, and the Linux client opens one connection
+    /// per attached device, so every client starts again at 1. Tracking by bare sequence
+    /// number put all of them in one namespace: a second client's first request looked
+    /// like a duplicate of the first client's, and attaching two devices from a single
+    /// machine was enough to trigger it. Devid distinguishes them, since a connection
+    /// carries exactly one imported device.
+    private struct URBKey: Hashable {
+        let devid: UInt32
+        let seqnum: UInt32
+    }
+
+    private var activeURBs: [URBKey: TrackedURB] = [:]
 
-    /// Sequence numbers accepted and not yet answered, claimed before the URB exists so
-    /// that an UNLINK arriving during setup has something to find.
-    private var reservedSeqnums: Set<UInt32> = []
+    /// Requests accepted and not yet answered, claimed before the URB exists so that an
+    /// UNLINK arriving during setup has something to find.
+    private var reserved: Set<URBKey> = []
 
-    /// Sequence numbers the client withdrew. A transfer already in IOKit's hands may
-    /// still complete after the abort, and its result must be dropped rather than sent:
-    /// the client has been told the request is dead and is not expecting a reply.
-    private var cancelledSeqnums: Set<UInt32> = []
+    /// Requests the client withdrew. A transfer already in IOKit's hands may still
+    /// complete after the abort, and its result must be dropped rather than sent: the
+    /// client has been told the request is dead and is not expecting a reply.
+    private var cancelled: Set<URBKey> = []
     private let urbQueue = DispatchQueue(label: "com.usbipd.mac.urb", attributes: .concurrent)
 
     /// Device communicator for executing USB transfers
@@ -106,6 +119,20 @@ public class USBSubmitProcessor {
         do {
             try validateSubmitRequest(request)
             try await checkConcurrentRequestLimit(devid: request.devid)
+
+            // Claim the request before doing anything that can block.
+            //
+            // Registration used to happen after the device had been resolved, which
+            // means after an IOKit enumeration. A client that submits a read and cancels
+            // it a millisecond later — probe-rs drains the IN endpoint exactly that way
+            // before its first command — sent the UNLINK into that window, found no such
+            // URB, and was told the request had already completed. It had not: it went
+            // on to run, and later returned data for a request the client had abandoned.
+            //
+            // This sits inside the block above so that a rejected claim is answered.
+            // Thrown from outside it, a duplicate reached ServerCoordinator, which logs
+            // and sends nothing, and the client waited for a reply that never came.
+            try await reserveSequenceNumber(devid: request.devid, seqnum: request.seqnum)
         } catch {
             logger.warning("Rejected SUBMIT request", context: [
                 "seqnum": String(request.seqnum),
@@ -126,19 +153,6 @@ public class USBSubmitProcessor {
             )
         }
 
-        // Claim the sequence number before doing anything that can block.
-        //
-        // Registration used to happen after the device had been resolved, which means
-        // after an IOKit enumeration. A client that submits a read and cancels it a
-        // millisecond later — probe-rs drains the IN endpoint exactly that way before
-        // its first command — sent the UNLINK into that window, found no such URB, and
-        // was told the request had already completed. It had not: it went on to run,
-        // and later returned data for a request the client had abandoned.
-        //
-        // Only the sequence number is known this early, which is all the cancel path
-        // needs to reserve the slot; the full URB replaces this entry below.
-        try await reserveSequenceNumber(request.seqnum)
-
         // Prefer the device's own description of the endpoint. CMD_SUBMIT has no
         // transfer-type field, so anything derived from the request alone is a guess —
         // and the interval-based guess below misreads every bulk endpoint that
@@ -194,17 +208,17 @@ public class USBSubmitProcessor {
             // The data is dropped with it. That is the honest outcome: it belongs to a
             // transfer the client cancelled, and handing it to the next read would put
             // one command's answer in front of another's.
-            if await wasCancelled(request.seqnum) {
+            if await wasCancelled(devid: request.devid, seqnum: request.seqnum) {
                 logger.info("Discarding result of a cancelled request", context: [
                     "seqnum": String(request.seqnum),
                     "actualLength": String(result.actualLength)
                 ])
-                await removeActiveURB(request.seqnum)
+                await removeActiveURB(devid: request.devid, seqnum: request.seqnum)
                 return Data()
             }
 
             // Remove URB from tracking
-            await removeActiveURB(request.seqnum)
+            await removeActiveURB(devid: request.devid, seqnum: request.seqnum)
 
             return try USBIPMessageEncoder.encodeUSBSubmitResponse(
                 seqnum: response.seqnum,
@@ -221,11 +235,11 @@ public class USBSubmitProcessor {
         } catch {
             // Aborting a pipe makes the transfer in flight fail, so a cancelled request
             // usually arrives here rather than above. It is silent for the same reason.
-            if await wasCancelled(request.seqnum) {
+            if await wasCancelled(devid: request.devid, seqnum: request.seqnum) {
                 logger.info("Cancelled request ended without a reply", context: [
                     "seqnum": String(request.seqnum)
                 ])
-                await removeActiveURB(request.seqnum)
+                await removeActiveURB(devid: request.devid, seqnum: request.seqnum)
                 return Data()
             }
 
@@ -238,7 +252,7 @@ public class USBSubmitProcessor {
             let errorResponse = createErrorResponse(from: request, error: error)
 
             // Remove URB from tracking
-            await removeActiveURB(request.seqnum)
+            await removeActiveURB(devid: request.devid, seqnum: request.seqnum)
 
             return try USBIPMessageEncoder.encodeUSBSubmitResponse(
                 seqnum: errorResponse.seqnum,
@@ -324,12 +338,13 @@ public class USBSubmitProcessor {
     /// carries the duplicate check that used to live in `addActiveURB`, which is the
     /// right place for it: a repeat of a sequence number is a client error whether or
     /// not the first one has finished being prepared.
-    private func reserveSequenceNumber(_ seqnum: UInt32) async throws {
+    private func reserveSequenceNumber(devid: UInt32, seqnum: UInt32) async throws {
         try urbQueue.sync(flags: .barrier) {
-            guard !reservedSeqnums.contains(seqnum) else {
+            let key = URBKey(devid: devid, seqnum: seqnum)
+            guard !reserved.contains(key) else {
                 throw USBRequestError.duplicateRequest
             }
-            reservedSeqnums.insert(seqnum)
+            reserved.insert(key)
         }
     }
 
@@ -338,34 +353,37 @@ public class USBSubmitProcessor {
         // Barrier: this mutates shared state on a concurrent queue. Plain .sync
         // lets writers run simultaneously and corrupts the dictionary.
         urbQueue.sync(flags: .barrier) {
-            activeURBs[urb.seqnum] = TrackedURB(urb: urb, status: .pending, device: device)
+            activeURBs[URBKey(devid: urb.devid, seqnum: urb.seqnum)] =
+                TrackedURB(urb: urb, status: .pending, device: device)
         }
     }
 
     /// Whether the client withdrew this request while it was running.
-    func wasCancelled(_ seqnum: UInt32) async -> Bool {
-        return urbQueue.sync { cancelledSeqnums.contains(seqnum) }
+    func wasCancelled(devid: UInt32, seqnum: UInt32) async -> Bool {
+        return urbQueue.sync { cancelled.contains(URBKey(devid: devid, seqnum: seqnum)) }
     }
 
     /// Remove URB from active tracking
-    private func removeActiveURB(_ seqnum: UInt32) async {
+    private func removeActiveURB(devid: UInt32, seqnum: UInt32) async {
         // Barrier: this mutates shared state on a concurrent queue. Plain .sync
         // lets writers run simultaneously and corrupts the dictionary.
         urbQueue.sync(flags: .barrier) {
-            activeURBs.removeValue(forKey: seqnum)
-            reservedSeqnums.remove(seqnum)
-            cancelledSeqnums.remove(seqnum)
+            let key = URBKey(devid: devid, seqnum: seqnum)
+            activeURBs.removeValue(forKey: key)
+            reserved.remove(key)
+            cancelled.remove(key)
         }
     }
 
     /// Update URB status
-    private func updateURBStatus(_ seqnum: UInt32, status: URBStatus) async {
+    private func updateURBStatus(devid: UInt32, seqnum: UInt32, status: URBStatus) async {
         // Barrier: this mutates shared state on a concurrent queue. Plain .sync
         // lets writers run simultaneously and corrupts the dictionary.
         urbQueue.sync(flags: .barrier) {
-            if var entry = activeURBs[seqnum] {
+            let key = URBKey(devid: devid, seqnum: seqnum)
+            if var entry = activeURBs[key] {
                 entry.status = status
-                activeURBs[seqnum] = entry
+                activeURBs[key] = entry
             }
         }
     }
@@ -377,7 +395,7 @@ public class USBSubmitProcessor {
         }
 
         // Update URB status
-        await updateURBStatus(urb.seqnum, status: .inProgress)
+        await updateURBStatus(devid: urb.devid, seqnum: urb.seqnum, status: .inProgress)
 
         // Execute based on transfer type
         switch urb.transferType {
@@ -567,23 +585,27 @@ public class USBSubmitProcessor {
         return urbQueue.sync { activeURBs.count }
     }
 
-    /// Cancel URB by sequence number (for UNLINK support)
-    public func cancelURB(_ seqnum: UInt32) async -> Bool {
+    /// Cancel a request (for UNLINK support).
+    ///
+    /// Takes the devid as well as the sequence number: numbering restarts per
+    /// connection, so a sequence number on its own could name another client's request.
+    public func cancelURB(devid: UInt32, seqnum: UInt32) async -> Bool {
         // Barrier: this mutates shared state on a concurrent queue. Plain .sync
         // lets writers run simultaneously and corrupts the dictionary.
         let claimed: CancelClaim = urbQueue.sync(flags: .barrier) {
             // A reservation is enough. Requiring an entry in activeURBs meant a request
             // still being set up could not be cancelled, and the client was told its URB
             // had already completed when in fact it had not started.
-            guard reservedSeqnums.contains(seqnum) else {
+            let key = URBKey(devid: devid, seqnum: seqnum)
+            guard reserved.contains(key) else {
                 return CancelClaim(reserved: false, urb: nil, device: nil)
             }
-            cancelledSeqnums.insert(seqnum)
-            guard var entry = activeURBs[seqnum] else {
+            cancelled.insert(key)
+            guard var entry = activeURBs[key] else {
                 return CancelClaim(reserved: true, urb: nil, device: nil)
             }
             entry.status = .cancelled
-            activeURBs[seqnum] = entry
+            activeURBs[key] = entry
             return CancelClaim(reserved: true, urb: entry.urb, device: entry.device)
         }
 

Sources/USBIPDCore/Protocol/USBUnlinkProcessor.swift

@@ -165,7 +165,7 @@ public class USBUnlinkProcessor {
         ])
 
         // Attempt to cancel the URB at the protocol level first
-        let urbCancelSuccess = await processor.cancelURB(request.unlinkSeqnum)
+        let urbCancelSuccess = await processor.cancelURB(devid: request.devid, seqnum: request.unlinkSeqnum)
 
         // Aborting the pipe belongs to the submit processor, which is the only place
         // that knows what a sequence number was submitted against.

Tests/USBIPDCoreTests/Protocol/USBRequestProcessorTests.swift

@@ -432,7 +432,7 @@ final class USBRequestProcessorTests: XCTestCase {
         var didCancel = false
         for _ in 0..<100 where !didCancel {
             try await Task.sleep(nanoseconds: 5_000_000)
-            didCancel = await submitProcessor.cancelURB(seqnum)
+            didCancel = await submitProcessor.cancelURB(devid: USBRequestProcessorTests.testDevid, seqnum: seqnum)
         }
         XCTAssertTrue(didCancel, "a request that is still running should be cancellable")
 
@@ -448,7 +448,7 @@ final class USBRequestProcessorTests: XCTestCase {
     }
 
     func testUnknownSequenceNumberIsNotCancellable() async throws {
-        let cancelled = await submitProcessor.cancelURB(999_999)
+        let cancelled = await submitProcessor.cancelURB(devid: USBRequestProcessorTests.testDevid, seqnum: 999_999)
         XCTAssertFalse(cancelled, "a sequence number never submitted should not be cancellable")
     }
 
@@ -704,21 +704,85 @@ final class USBRequestProcessorTests: XCTestCase {
         XCTAssertEqual(finalResponse.status, 0)
         XCTAssertEqual(finalResponse.actualLength, 16)
     }
+
+    // MARK: - Two clients do not share a sequence-number space
+
+    /// USB/IP numbers requests per connection, and the Linux client opens one connection
+    /// per attached device — so every client starts again at 1. Tracking by bare
+    /// sequence number put them all in one namespace, and a second client's first
+    /// request was rejected as a duplicate of the first client's. Attaching two devices
+    /// from a single machine was enough to trigger it: five of ten concurrent transfers
+    /// failed against real hardware.
+    func testTwoDevicesMayUseTheSameSequenceNumber() async throws {
+        // Two attached devices, so both devids resolve to something real.
+        let other = USBDevice(
+            busID: "2", deviceID: "3",
+            vendorID: 0x4321, productID: 0x8765,
+            deviceClass: 0x00, deviceSubClass: 0x00, deviceProtocol: 0x00,
+            speed: .high,
+            manufacturerString: nil, productString: nil, serialNumberString: nil
+        )
+        let processor = USBSubmitProcessor(
+            deviceCommunicator: mockDeviceCommunicator,
+            deviceDiscovery: StubDeviceDiscovery(devices: [testDevice, other])
+        )
+
+        let first = USBRequestProcessorTests.testDevid
+        let second = (2 << 16) | UInt32(0x3)
+
+        let firstRequest = try createUSBSubmitRequestData(seqnum: 1, devid: first, endpoint: 0x00)
+        let secondRequest = try createUSBSubmitRequestData(seqnum: 1, devid: second, endpoint: 0x00)
+
+        let firstResponse = try USBIPSubmitResponse.decode(
+            from: try await processor.processSubmitRequest(firstRequest))
+        let secondResponse = try USBIPSubmitResponse.decode(
+            from: try await processor.processSubmitRequest(secondRequest))
+
+        XCTAssertEqual(firstResponse.status, 0)
+        XCTAssertEqual(secondResponse.status, 0, "a second device reusing seqnum 1 must not look like a duplicate")
+    }
+
+    /// A genuine duplicate — same device, same sequence number, still outstanding — is
+    /// still refused. And it is refused with a reply: the rejection used to be thrown
+    /// past the error-response path, so ServerCoordinator logged it, sent nothing, and
+    /// the client waited forever for an answer that was never coming.
+    func testDuplicateOnOneDeviceIsRefusedWithAReply() async throws {
+        mockDeviceCommunicator.setOperationLatency(300)
+
+        let requestData = try createUSBSubmitRequestData(seqnum: 77, endpoint: 0x00)
+        let inFlight = Task { try await self.submitProcessor.processSubmitRequest(requestData) }
+
+        // Let the first request claim the sequence number before repeating it.
+        try await Task.sleep(nanoseconds: 50_000_000)
+
+        let duplicate = try await submitProcessor.processSubmitRequest(requestData)
+        XCTAssertFalse(duplicate.isEmpty, "a duplicate must be answered, not met with silence")
+
+        let response = try USBIPSubmitResponse.decode(from: duplicate)
+        XCTAssertEqual(response.seqnum, 77)
+        XCTAssertNotEqual(response.status, 0, "a duplicate must report an error")
+
+        _ = try await inFlight.value
+    }
 }
 
-/// Returns a single device for any lookup, so submitted URBs resolve to test hardware.
 private final class StubDeviceDiscovery: DeviceDiscovery {
-    private let device: USBDevice
+    private let devices: [USBDevice]
+    private var device: USBDevice { devices[0] }
 
     var onDeviceConnected: ((USBDevice) -> Void)?
     var onDeviceDisconnected: ((USBDevice) -> Void)?
 
     init(device: USBDevice) {
-        self.device = device
+        self.devices = [device]
+    }
+
+    init(devices: [USBDevice]) {
+        self.devices = devices
     }
 
     func discoverDevices() throws -> [USBDevice] {
-        return [device]
+        return devices
     }
 
     func getDevice(busID: String, deviceID: String) throws -> USBDevice? {