Improve port forwarding concurrency and throughput

Frameworks/TermindRemoteInfrastructure/SSH/PortForwardingController.swift

@@ -59,6 +59,7 @@ public final class PortForwardingConnection {
     fileprivate var session: (any SSHForwardingSessionProtocol)?
     fileprivate var task: Task<Void, Never>?
     fileprivate var connected = false
+    fileprivate var connectionCounts: [PortForwardingRule.ID: Int] = [:]
 
     fileprivate init(host: Host, route: [Host]) {
         hostID = host.id
@@ -247,10 +248,12 @@ public final class PortForwardingController {
             { [weak self, weak context] id, count in
                 Task { @MainActor in
                     guard let self, let context, self.isCurrent(context, attempt: attempt),
-                        self.generations[id] != nil, self.presentations[id] === context,
-                        case .running = self.states[id]
+                        self.generations[id] != nil, self.presentations[id] === context
                     else { return }
-                    self.states[id] = .running(activeConnections: count)
+                    context.connectionCounts[id] = count
+                    if case .running = self.states[id], self.states[id] != .running(activeConnections: count) {
+                        self.states[id] = .running(activeConnections: count)
+                    }
                 }
             },
             { [weak self, weak context] message in
@@ -366,6 +369,7 @@ public final class PortForwardingController {
 
     private func startRule(_ rule: PortForwardingRule, in context: PortForwardingConnection) {
         guard let token = generations[rule.id], let session = context.session else { return }
+        context.connectionCounts[rule.id] = 0
         states[rule.id] = .starting
         let previous = ruleTasks[rule.id]
         ruleTasks[rule.id] = Task { [weak self] in
@@ -374,7 +378,7 @@ public final class PortForwardingController {
             do {
                 try await session.start(rule)
                 guard generations[rule.id] == token else { return }
-                states[rule.id] = .running(activeConnections: 0)
+                states[rule.id] = .running(activeConnections: context.connectionCounts[rule.id] ?? 0)
                 context.journey.succeed()
             } catch {
                 guard generations[rule.id] == token else { return }

Frameworks/TermindRemoteInfrastructure/SSH/SSHBufferedTransport.swift

@@ -0,0 +1,86 @@
+import CSSH2
+import Darwin
+import Foundation
+
+/// Queue-confined encrypted output. libssh2's forwarded-tcpip confirmation
+/// retry loses its pending channel in the pinned library when send returns
+/// EAGAIN. Accept complete packets here and apply backpressure between library
+/// calls instead. The receive callback also stops packet processing at the
+/// watermark, bounding replies generated while libssh2 drains input.
+nonisolated final class SSHBufferedTransport {
+    private let output = SSHByteQueue()
+    private var disposed = false
+    private var readAllowance = SSHStreamPump.quantum
+    var exhaustedReadBudget: Bool { readAllowance == 0 }
+
+    func beginPass() { readAllowance = SSHStreamPump.quantum }
+    var isBackpressured: Bool { output.count >= SSHStreamPump.highWater }
+    var wantsWrite: Bool { !output.isEmpty }
+
+    init(session: OpaquePointer) {
+        libssh2_session_abstract(session)?.pointee = Unmanaged.passUnretained(self).toOpaque()
+        libssh2_session_callback_set2(
+            session, LIBSSH2_CALLBACK_SEND, unsafeBitCast(Self.send, to: (@convention(c) () -> Void).self))
+        libssh2_session_callback_set2(
+            session, LIBSSH2_CALLBACK_RECV, unsafeBitCast(Self.receive, to: (@convention(c) () -> Void).self))
+    }
+
+    func flush(socket: Int32) throws {
+        while !output.isEmpty {
+            let count = output.withFirstBytes { Darwin.send(socket, $0.baseAddress!, $0.count, MSG_DONTWAIT) }
+            if count > 0 {
+                output.consume(count)
+            } else if count < 0 && errno == EINTR {
+                continue
+            } else if count < 0 && (errno == EAGAIN || errno == EWOULDBLOCK) {
+                return
+            } else {
+                throw SSHStreamPump.StreamError.closed
+            }
+        }
+    }
+
+    /// Teardown cannot wait on a peer. Discard final protocol writes and return
+    /// EOF to reads, so libssh2 can release channels/listeners without EAGAIN.
+    func dispose(session: OpaquePointer) {
+        disposed = true
+        let result = libssh2_session_free(session)
+        assert(result == 0, "SSH session disposal must finish without socket I/O")
+    }
+
+    private static let send:
+        @convention(c) (
+            Int32, UnsafeRawPointer?, Int, Int32, UnsafeMutablePointer<UnsafeMutableRawPointer?>?
+        ) -> Int = { socket, bytes, count, flags, abstract in
+            guard let opaque = abstract?.pointee, let bytes else { return -Int(EINVAL) }
+            let transport = Unmanaged<SSHBufferedTransport>.fromOpaque(opaque).takeUnretainedValue()
+            if transport.disposed { return count }
+            var sent = 0
+            if transport.output.isEmpty {
+                repeat {
+                    sent = Darwin.send(socket, bytes, count, flags | MSG_DONTWAIT)
+                } while sent < 0 && errno == EINTR
+                if sent < 0 {
+                    guard errno == EAGAIN || errno == EWOULDBLOCK else { return -Int(errno) }
+                    sent = 0
+                }
+            }
+            if sent < count {
+                transport.output.append(UnsafeRawBufferPointer(start: bytes + sent, count: count - sent))
+            }
+            return count
+        }
+
+    private static let receive:
+        @convention(c) (
+            Int32, UnsafeMutableRawPointer?, Int, Int32, UnsafeMutablePointer<UnsafeMutableRawPointer?>?
+        ) -> Int = { socket, bytes, count, flags, abstract in
+            guard let opaque = abstract?.pointee else { return -Int(EINVAL) }
+            let transport = Unmanaged<SSHBufferedTransport>.fromOpaque(opaque).takeUnretainedValue()
+            if transport.disposed { return 0 }
+            if transport.isBackpressured || transport.exhaustedReadBudget { return -Int(EAGAIN) }
+            let received = Darwin.recv(socket, bytes, min(count, transport.readAllowance), flags | MSG_DONTWAIT)
+            if received > 0 { transport.readAllowance -= received }
+            return received < 0 ? -Int(errno) : received
+        }
+}

Frameworks/TermindRemoteInfrastructure/SSH/SSHForwardingConnector.swift

@@ -0,0 +1,172 @@
+import Darwin
+import Foundation
+
+/// Forwarded target connections must never block the queue that services SSH.
+/// Only DNS uses blocking workers (at most four); connect and its overall
+/// five-second deadline use readiness events on the owning session queue.
+nonisolated final class SSHForwardingConnector: @unchecked Sendable {
+    private struct Address: Sendable {
+        let family: Int32
+        let type: Int32
+        let proto: Int32
+        let bytes: Data
+    }
+
+    private static let resolver: OperationQueue = {
+        let queue = OperationQueue()
+        queue.name = "com.akinokaede.termind.forwarding-dns"
+        queue.maxConcurrentOperationCount = 4
+        return queue
+    }()
+
+    private let queue: DispatchQueue
+    private var completion: (@Sendable (Result<Int32, any Error>) -> Void)?
+    private var addresses: [Address] = []
+    private var socket: Int32 = -1
+    private var events: SSHSocketEvents?
+    private var deadline: (any DispatchSourceTimer)?
+    private var resolution: BlockOperation?
+    private var finished = false
+    private var cancelled = false
+
+    init(queue: DispatchQueue, completion: @escaping @Sendable (Result<Int32, any Error>) -> Void) {
+        self.queue = queue
+        self.completion = completion
+    }
+
+    func start(host: String, port: Int) {
+        let timer = DispatchSource.makeTimerSource(queue: queue)
+        timer.schedule(deadline: .now() + 5)
+        timer.setEventHandler { [weak self] in self?.finish(.failure(POSIXError(.ETIMEDOUT))) }
+        timer.resume()
+        deadline = timer
+        // Numeric targets do not need a resolver worker or a thread-pool hop.
+        let numeric = Self.resolve(host: host, port: port, flags: AI_NUMERICHOST)
+        if !numeric.isEmpty {
+            addresses = numeric
+            tryNext()
+            return
+        }
+        let operation = BlockOperation { [weak self] in
+            let resolved = Self.resolve(host: host, port: port, flags: 0)
+            guard let self else { return }
+            self.queue.async {
+                guard !self.finished else { return }
+                self.resolution = nil
+                self.addresses = resolved
+                self.tryNext()
+            }
+        }
+        resolution = operation
+        Self.resolver.addOperation(operation)
+    }
+
+    private static func resolve(host: String, port: Int, flags: Int32) -> [Address] {
+        var hints = addrinfo()
+        hints.ai_flags = flags
+        hints.ai_family = AF_UNSPEC
+        hints.ai_socktype = SOCK_STREAM
+        hints.ai_protocol = IPPROTO_TCP
+        var result: UnsafeMutablePointer<addrinfo>?
+        guard getaddrinfo(host, String(port), &hints, &result) == 0, let first = result else { return [] }
+        defer { freeaddrinfo(first) }
+        var addresses: [Address] = []
+        var cursor: UnsafeMutablePointer<addrinfo>? = first
+        while let info = cursor?.pointee {
+            addresses.append(
+                Address(
+                    family: info.ai_family, type: info.ai_socktype, proto: info.ai_protocol,
+                    bytes: Data(bytes: info.ai_addr, count: Int(info.ai_addrlen))))
+            cursor = info.ai_next
+        }
+        return addresses
+    }
+
+    func cancel() {
+        cancelled = true
+        completion = nil
+        finish(.failure(CancellationError()))
+    }
+
+    private func tryNext() {
+        guard !finished else { return }
+        guard !addresses.isEmpty else {
+            finish(.failure(POSIXError(.ECONNREFUSED)))
+            return
+        }
+        let address = addresses.removeFirst()
+        let candidate = Darwin.socket(address.family, address.type, address.proto)
+        guard candidate >= 0 else {
+            tryNext()
+            return
+        }
+        socket = candidate
+        var yes: Int32 = 1
+        _ = setsockopt(candidate, SOL_SOCKET, SO_NOSIGPIPE, &yes, socklen_t(MemoryLayout<Int32>.size))
+        let flags = fcntl(candidate, F_GETFL, 0)
+        guard flags >= 0, fcntl(candidate, F_SETFL, flags | O_NONBLOCK) == 0 else {
+            Darwin.close(candidate)
+            socket = -1
+            tryNext()
+            return
+        }
+        let result = address.bytes.withUnsafeBytes {
+            Darwin.connect(candidate, $0.baseAddress!.assumingMemoryBound(to: sockaddr.self), socklen_t($0.count))
+        }
+        if result == 0 {
+            finish(.success(candidate))
+            return
+        }
+        guard errno == EINPROGRESS else {
+            Darwin.close(candidate)
+            socket = -1
+            tryNext()
+            return
+        }
+        events = SSHSocketEvents(socket: candidate, queue: queue) { [weak self] in self?.connected() }
+        events?.update(read: false, write: true)
+    }
+
+    private func connected() {
+        guard !finished else { return }
+        var error: Int32 = 0
+        var size = socklen_t(MemoryLayout<Int32>.size)
+        if getsockopt(socket, SOL_SOCKET, SO_ERROR, &error, &size) == 0 && error == 0 {
+            finish(.success(socket))
+        } else {
+            let descriptor = socket
+            socket = -1
+            events?.cancel { [self] in
+                Darwin.close(descriptor)
+                tryNext()
+            }
+            events = nil
+        }
+    }
+
+    private func finish(_ result: Result<Int32, any Error>) {
+        guard !finished else { return }
+        finished = true
+        resolution?.cancel()
+        resolution = nil
+        deadline?.cancel()
+        deadline = nil
+        let descriptor = socket
+        socket = -1
+        let deliver: @Sendable () -> Void = { [self] in
+            if case .failure = result, descriptor >= 0 { Darwin.close(descriptor) }
+            if cancelled {
+                if case .success(let opened) = result { Darwin.close(opened) }
+            } else {
+                completion?(result)
+            }
+            completion = nil
+        }
+        if let events {
+            events.cancel(then: deliver)
+            self.events = nil
+        } else {
+            deliver()
+        }
+    }
+}

Frameworks/TermindRemoteInfrastructure/SSH/SSHForwardingSession.swift

@@ -52,33 +52,46 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable {
         precondition(libssh2_init(0) == 0, "libssh2 initialization failed")
     }()
 
-    private final class RuleContext {
+    private final class RuleContext: @unchecked Sendable {
         let rule: PortForwardingRule
         var socket: Int32 = -1
+        var events: SSHSocketEvents?
         var remoteListener: OpaquePointer?
         var flows: [UUID: Flow] = [:]
+        var start: CheckedContinuation<Void, any Error>?
+        var stopped = false
+        var stop: (@Sendable () -> Void)?
+        var reportedCount = -1
 
         init(rule: PortForwardingRule) { self.rule = rule }
     }
 
-    private final class Flow {
-        enum Phase { case greeting, request, forwarding }
+    private final class Flow: @unchecked Sendable {
+        enum Phase { case greeting, request, opening, connecting, forwarding }
         let id = UUID()
-        var socket: Int32
-        var channel: OpaquePointer?
+        let context: RuleContext
+        let stream: SSHStreamPump
         var phase: Phase
+        var target: (String, Int)?
         var handshake = Data()
-        var toChannel = Data()
-        var toSocket = Data()
-        var socketEOF = false
-        var channelEOF = false
-        var socketWriteClosed = false
-        var channelWriteClosed = false
+        var events: SSHSocketEvents?
+        var connector: SSHForwardingConnector?
+        var retired = false
 
-        init(socket: Int32, dynamic: Bool, channel: OpaquePointer? = nil) {
-            self.socket = socket
-            self.phase = dynamic ? .greeting : .forwarding
-            self.channel = channel
+        init(socket: Int32, context: RuleContext, channel: OpaquePointer? = nil) {
+            self.context = context
+            stream = SSHStreamPump(socket: socket, channel: channel)
+            phase = context.rule.kind == .dynamic ? .greeting : channel == nil ? .opening : .connecting
+            if let host = context.rule.targetHost, let port = context.rule.targetPort { target = (host, port) }
+        }
+    }
+
+    private final class RetiredListener {
+        let listener: OpaquePointer
+        let completion: () -> Void
+        init(_ listener: OpaquePointer, completion: @escaping () -> Void) {
+            self.listener = listener
+            self.completion = completion
         }
     }
 
@@ -89,9 +102,25 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable {
     private var routeTransport: SSHRoutedSocket?
     private var session: OpaquePointer?
     private var rules: [PortForwardingRule.ID: RuleContext] = [:]
+    private var flows: [UUID: Flow] = [:]
+    private var bufferedTransport: SSHBufferedTransport?
+    private var transportEvents: SSHSocketEvents?
     private var timer: (any DispatchSourceTimer)?
+    private var ready = SSHReadyQueue<UUID>()
+    private var pendingOpens = SSHReadyQueue<UUID>()
+    private var opening: Flow?
+    private var pendingListeners: [RuleContext] = []
+    private var listening: RuleContext?
+    private var retiredFlows: [Flow] = []
+    private var retiredListeners: [RetiredListener] = []
+    private var scratch = [UInt8](repeating: 0, count: 32 * 1024)
+    private var scheduled = false
+    private var generation = 0
+    private var controlDirections: Int32 = 0
+    private var keepaliveDue = true
+    private var nextKeepalive = DispatchTime.now()
 
-    public init(
+    init(
         connectionHandler: @escaping SSHForwardingSession.ConnectionHandler,
         failureHandler: @escaping SSHForwardingSession.FailureHandler
     ) {
@@ -100,7 +129,9 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable {
         self.failureHandler = failureHandler
     }
 
-    public func connect(_ request: SSHConnectionRequest, progress: SSHRouteProgressHandler?) async throws {
+    deinit { cleanup() }
+
+    func connect(_ request: SSHConnectionRequest, progress: SSHRouteProgressHandler?) async throws {
         await queue.perform { self.cleanup() }
         let opened = try await SSHSessionWorker.openAuthenticatedSession(
             using: request, timeout: 20, progress: progress)
@@ -109,9 +140,10 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable {
                 self.routeTransport = opened.transport
                 self.socket = opened.transport.socket
                 self.session = opened.session
+                self.bufferedTransport = SSHBufferedTransport(session: opened.session)
                 libssh2_session_set_blocking(opened.session, 0)
-                libssh2_keepalive_config(opened.session, 1, 20)
-                self.startTimer()
+                libssh2_keepalive_config(opened.session, 0, 20)
+                self.startEvents()
             }
             try Task.checkCancellation()
         } catch {
@@ -120,11 +152,11 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable {
         }
     }
 
-    public func start(_ rule: PortForwardingRule) async throws {
+    func start(_ rule: PortForwardingRule) async throws {
         try await withCheckedThrowingContinuation { continuation in
             queue.async { [self] in
                 do {
-                    guard let session else { throw TermindSSHError.disconnected }
+                    guard session != nil else { throw TermindSSHError.disconnected }
                     if rules[rule.id] != nil {
                         continuation.resume()
                         return
@@ -132,223 +164,497 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable {
                     let context = RuleContext(rule: rule)
                     switch rule.kind {
                     case .local, .dynamic:
-                        do {
-                            context.socket = try makeListener(
-                                host: rule.bindHost, port: rule.bindPort
-                            )
-                        } catch {
+                        do { context.socket = try makeListener(host: rule.bindHost, port: rule.bindPort) } catch {
                             throw TermindSSHError.route(
                                 String(
                                     localized: "Unable to listen on \(rule.bindAddress): \(error.localizedDescription)")
                             )
                         }
+                        context.events = SSHSocketEvents(socket: context.socket, queue: queue) {
+                            [weak self, weak context] in
+                            guard let self, let context, !context.stopped else { return }
+                            self.acceptLocal(context)
+                        }
+                        context.events?.update(read: true, write: false)
+                        rules[rule.id] = context
+                        report(context)
+                        continuation.resume()
                     case .remote:
-                        var boundPort: Int32 = 0
-                        let listener = rule.bindHost.withCString { host in
-                            libssh2_channel_forward_listen_ex(
-                                session, host, Int32(rule.bindPort), &boundPort, 64
-                            )
-                        }
-                        guard let listener else {
-                            throw TermindSSHError.route(
-                                SSHSessionWorker.sessionError(
-                                    session, prefix: String(localized: "Unable to start remote forwarding")
-                                ).localizedDescription
-                            )
-                        }
-                        context.remoteListener = listener
+                        context.start = continuation
+                        rules[rule.id] = context
+                        pendingListeners.append(context)
+                        schedulePump()
                     }
-                    rules[rule.id] = context
-                    connectionHandler(rule.id, 0)
-                    continuation.resume()
                 } catch { continuation.resume(throwing: error) }
             }
         }
     }
 
-    public func stop(_ id: PortForwardingRule.ID) async {
+    func stop(_ id: PortForwardingRule.ID) async {
         await withCheckedContinuation { continuation in
             queue.async { [self] in
-                if let context = rules.removeValue(forKey: id) { close(context) }
-                continuation.resume()
+                guard let context = rules.removeValue(forKey: id) else {
+                    continuation.resume()
+                    return
+                }
+                close(context) { continuation.resume() }
+                schedulePump()
             }
         }
     }
 
-    public func disconnect() async {
-        await withCheckedContinuation { continuation in
-            queue.async { [self] in
-                cleanup()
-                continuation.resume()
-            }
-        }
-    }
+    func disconnect() async { await queue.perform { self.cleanup() } }
 
-    private func startTimer() {
-        timer?.cancel()
+    private func startEvents() {
+        let token = generation
+        transportEvents = SSHSocketEvents(socket: socket, queue: queue) { [weak self] in
+            guard let self, self.generation == token else { return }
+            self.wakeAll()
+        }
+        transportEvents?.update(read: true, write: false)
         let source = DispatchSource.makeTimerSource(queue: queue)
-        source.schedule(deadline: .now(), repeating: .milliseconds(12), leeway: .milliseconds(4))
-        source.setEventHandler { [weak self] in self?.pump() }
+        // This timer services protocol deadlines and keepalive, never data throughput.
+        source.schedule(deadline: .now() + 1, repeating: .seconds(1), leeway: .milliseconds(100))
+        source.setEventHandler { [weak self] in
+            guard let self, self.generation == token else { return }
+            if DispatchTime.now() >= self.nextKeepalive { self.keepaliveDue = true }
+            if self.keepaliveDue || self.opening != nil || self.listening != nil
+                || !self.retiredFlows.isEmpty || !self.retiredListeners.isEmpty
+            {
+                self.wakeAll()
+            }
+        }
         timer = source
         source.resume()
+        keepaliveDue = true
+        schedulePump()
+    }
+
+    private func wakeAll() {
+        for id in flows.keys { ready.insert(id) }
+        schedulePump()
+    }
+
+    private func schedulePump() {
+        guard session != nil, !scheduled else { return }
+        scheduled = true
+        let token = generation
+        queue.async { [weak self] in
+            guard let self, self.generation == token else { return }
+            self.scheduled = false
+            self.pump()
+        }
     }
 
     private func pump() {
-        guard let session else { return }
-        for context in Array(rules.values) {
-            switch context.rule.kind {
-            case .local, .dynamic: acceptLocal(context, session: session)
-            case .remote: acceptRemote(context)
-            }
-            for flow in Array(context.flows.values) { pump(flow, in: context, session: session) }
-            connectionHandler(context.rule.id, context.flows.count)
+        guard let session, let bufferedTransport else { return }
+        do { try bufferedTransport.flush(socket: socket) } catch {
+            fail(session)
+            return
         }
-        var next = Int32(0)
-        let keepalive = libssh2_keepalive_send(session, &next)
-        if keepalive < 0, keepalive != LIBSSH2_ERROR_EAGAIN {
-            let message = SSHSessionWorker.sessionError(
-                session, prefix: String(localized: "The forwarding SSH connection closed")
-            ).localizedDescription
-            cleanup()
-            failureHandler(message)
+        if bufferedTransport.isBackpressured {
+            for flow in flows.values { flow.events?.update(read: false, write: false) }
+            transportEvents?.update(read: false, write: true)
+            return
         }
-    }
-
-    private func acceptLocal(_ context: RuleContext, session: OpaquePointer) {
-        while true {
-            let accepted = Darwin.accept(context.socket, nil, nil)
-            guard accepted >= 0 else { break }
-            makeNonblocking(accepted)
-            let flow = Flow(socket: accepted, dynamic: context.rule.kind == .dynamic)
-            if context.rule.kind == .local {
-                guard let targetHost = context.rule.targetHost, let targetPort = context.rule.targetPort else {
-                    Darwin.close(accepted)
-                    continue
-                }
-                flow.channel = targetHost.withCString { target in
-                    libssh2_channel_direct_tcpip_ex(session, target, Int32(targetPort), "127.0.0.1", 0)
-                }
-                guard flow.channel != nil else {
-                    Darwin.close(accepted)
-                    continue
-                }
-            }
-            context.flows[flow.id] = flow
-        }
-    }
-
-    private func acceptRemote(_ context: RuleContext) {
-        guard let listener = context.remoteListener,
-            let targetHost = context.rule.targetHost, let targetPort = context.rule.targetPort
-        else { return }
-        while let channel = libssh2_channel_forward_accept(listener) {
-            do {
-                let local = try SSHSocketConnector.open(
-                    host: targetHost, port: targetPort, timeout: 5
-                )
-                makeNonblocking(local)
-                let flow = Flow(socket: local, dynamic: false, channel: channel)
-                context.flows[flow.id] = flow
-            } catch {
-                _ = libssh2_channel_free(channel)
-            }
-        }
-    }
-
-    private func pump(_ flow: Flow, in context: RuleContext, session: OpaquePointer) {
-        let highWaterMark = 2 * 1024 * 1024
-        if !flow.socketEOF,
-            (flow.phase == .forwarding ? flow.toChannel.count : flow.handshake.count) < highWaterMark
+        bufferedTransport.beginPass()
+        let deadline = DispatchTime.now().uptimeNanoseconds + 2_000_000
+        controlDirections = 0
+        var progressed = advanceListeners(session)
+        if advanceOpen(session) { progressed = true }
+        while !bufferedTransport.isBackpressured, DispatchTime.now().uptimeNanoseconds < deadline, let id = ready.pop()
         {
-            var buffer = [UInt8](repeating: 0, count: 32 * 1024)
-            let count = Darwin.recv(flow.socket, &buffer, buffer.count, 0)
-            if count > 0 {
-                let data = Data(buffer.prefix(count))
-                if flow.phase == .forwarding {
-                    flow.toChannel.append(data)
-                } else {
-                    flow.handshake.append(data)
-                    if flow.handshake.count > 64 * 1024 {
-                        close(flow, in: context)
-                        return
-                    }
-                    handleSOCKSHandshake(flow, session: session)
-                }
-            } else if count == 0 {
-                flow.socketEOF = true
-            } else if errno != EAGAIN && errno != EWOULDBLOCK {
-                close(flow, in: context)
+            guard let flow = flows[id] else { continue }
+            if pump(flow, session: session) { progressed = true }
+        }
+        // Any channel read/write can consume packets for another channel,
+        // including forwarded-tcpip opens and control replies. Service those
+        // queues after stream I/O; waiting for another socket edge would stall.
+        if advanceOpen(session) { progressed = true }
+        if advanceListeners(session) { progressed = true }
+        for context in rules.values where context.remoteListener != nil {
+            if acceptRemote(context) { progressed = true }
+        }
+        for flow in flows.values where flow.phase == .forwarding {
+            guard let channel = flow.stream.channel else { continue }
+            if (flow.stream.wantsChannelRead && libssh2_poll_channel_read(channel, 0) == 1)
+                || (!flow.stream.channelEOF && libssh2_channel_eof(channel) == 1)
+            {
+                ready.insert(flow.id)
+            }
+        }
+        if reap(session) { progressed = true }
+        if keepaliveDue {
+            var next: Int32 = 0
+            let result = libssh2_keepalive_send(session, &next)
+            if result == LIBSSH2_ERROR_EAGAIN {
+                blocked(session)
+            } else if result < 0 {
+                fail(session)
                 return
-            }
-        }
-
-        if let channel = flow.channel {
-            while !flow.toChannel.isEmpty {
-                let written = flow.toChannel.withUnsafeBytes { bytes -> Int in
-                    guard let base = bytes.baseAddress?.assumingMemoryBound(to: CChar.self) else { return 0 }
-                    return libssh2_channel_write_ex(channel, 0, base, bytes.count)
-                }
-                if written > 0 {
-                    flow.toChannel.removeFirst(written)
-                } else if written == LIBSSH2_ERROR_EAGAIN {
-                    break
-                } else {
-                    close(flow, in: context)
-                    return
-                }
-            }
-            if flow.socketEOF, flow.toChannel.isEmpty, !flow.channelWriteClosed {
-                let result = libssh2_channel_send_eof(channel)
-                if result == 0 {
-                    flow.channelWriteClosed = true
-                } else if result != LIBSSH2_ERROR_EAGAIN {
-                    close(flow, in: context)
-                    return
-                }
-            }
-            if flow.toSocket.count < highWaterMark {
-                var buffer = [CChar](repeating: 0, count: 32 * 1024)
-                while flow.toSocket.count < highWaterMark {
-                    let count = libssh2_channel_read_ex(channel, 0, &buffer, buffer.count)
-                    if count > 0 {
-                        flow.toSocket.append(Data(bytes: buffer, count: count))
-                    } else if count == LIBSSH2_ERROR_EAGAIN || count == 0 {
-                        break
-                    } else {
-                        close(flow, in: context)
-                        return
-                    }
-                }
-            }
-            if libssh2_channel_eof(channel) == 1 { flow.channelEOF = true }
-        }
-
-        while !flow.toSocket.isEmpty {
-            let sent = flow.toSocket.withUnsafeBytes { bytes -> Int in
-                guard let base = bytes.baseAddress else { return 0 }
-                return Darwin.send(flow.socket, base, bytes.count, 0)
-            }
-            if sent > 0 {
-                flow.toSocket.removeFirst(sent)
-            } else if errno == EAGAIN || errno == EWOULDBLOCK {
-                break
             } else {
-                close(flow, in: context)
-                return
+                keepaliveDue = false
+                nextKeepalive = .now() + .seconds(max(1, Int(next)))
             }
         }
-        if flow.channelEOF, flow.toSocket.isEmpty, !flow.socketWriteClosed {
-            _ = Darwin.shutdown(flow.socket, SHUT_WR)
-            flow.socketWriteClosed = true
+        for context in rules.values { report(context) }
+        let directions = flows.values.reduce(controlDirections) { $0 | $1.stream.blockDirections }
+        var byte: UInt8 = 0
+        let available = Darwin.recv(socket, &byte, 1, MSG_PEEK | MSG_DONTWAIT)
+        if available == 0 || (available < 0 && errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR) {
+            fail(session)
+            return
         }
-        if flow.socketEOF,
-            flow.channel == nil || flow.channelEOF,
-            flow.toChannel.isEmpty, flow.toSocket.isEmpty
-        {
-            close(flow, in: context)
+        // With no channel/listener libssh2 has no public session-read operation.
+        // Park unsolicited bytes until the next operation can consume them;
+        // never repeatedly wake on the same unread packet.
+        let canRead =
+            available < 0
+            || flows.values.contains {
+                $0.phase == .forwarding && ($0.stream.wantsChannelRead || !$0.stream.toChannel.isEmpty)
+            } || rules.values.contains { $0.remoteListener != nil }
+            || opening != nil || listening != nil || !retiredFlows.isEmpty || !retiredListeners.isEmpty
+        do { try bufferedTransport.flush(socket: socket) } catch {
+            fail(session)
+            return
+        }
+        transportEvents?.update(
+            read: canRead && !bufferedTransport.isBackpressured,
+            write: bufferedTransport.wantsWrite || directions & LIBSSH2_SESSION_BLOCK_OUTBOUND != 0)
+        if progressed || bufferedTransport.exhaustedReadBudget { wakeAll() } else if !ready.isEmpty { schedulePump() }
+    }
+
+    private func blocked(_ session: OpaquePointer) {
+        controlDirections |= libssh2_session_block_directions(session)
+    }
+
+    private func advanceListeners(_ session: OpaquePointer) -> Bool {
+        if listening == nil {
+            while !pendingListeners.isEmpty {
+                let next = pendingListeners.removeFirst()
+                if !next.stopped {
+                    listening = next
+                    break
+                }
+            }
+        }
+        guard let context = listening else { return false }
+        var boundPort: Int32 = 0
+        let listener = context.rule.bindHost.withCString {
+            libssh2_channel_forward_listen_ex(session, $0, Int32(context.rule.bindPort), &boundPort, 128)
+        }
+        if let listener {
+            listening = nil
+            if context.stopped {
+                retiredListeners.append(RetiredListener(listener, completion: context.stop ?? {}))
+                context.stop = nil
+            } else {
+                context.remoteListener = listener
+                context.start?.resume()
+                context.start = nil
+                report(context)
+            }
+            return true
+        }
+        if libssh2_session_last_errno(session) == LIBSSH2_ERROR_EAGAIN {
+            blocked(session)
+            return false
+        }
+        listening = nil
+        if rules[context.rule.id] === context { rules[context.rule.id] = nil }
+        context.start?.resume(
+            throwing: SSHSessionWorker.sessionError(
+                session, prefix: String(localized: "Unable to start remote forwarding")))
+        context.start = nil
+        context.stop?()
+        context.stop = nil
+        return true
+    }
+
+    private func advanceOpen(_ session: OpaquePointer) -> Bool {
+        if opening == nil {
+            while let id = pendingOpens.pop() {
+                if let flow = flows[id], flow.phase == .opening {
+                    opening = flow
+                    break
+                }
+            }
+        }
+        guard let flow = opening, let (host, port) = flow.target else { return false }
+        let channel = host.withCString {
+            libssh2_channel_direct_tcpip_ex(session, $0, Int32(port), "127.0.0.1", 0)
+        }
+        guard let channel else {
+            if libssh2_session_last_errno(session) == LIBSSH2_ERROR_EAGAIN {
+                blocked(session)
+                return false
+            }
+            opening = nil
+            if !flow.retired {
+                if flow.context.rule.kind == .dynamic {
+                    flow.stream.toSocket.append(socksReply(0x05))
+                    flow.stream.socketEOF = true
+                    flow.phase = .request
+                    ready.insert(flow.id)
+                } else {
+                    close(flow)
+                }
+            }
+            return true
+        }
+        opening = nil
+        flow.stream.channel = channel
+        if flow.retired {
+            retiredFlows.append(flow)
+        } else {
+            if flow.context.rule.kind == .dynamic { flow.stream.toSocket.append(socksReply(0x00)) }
+            flow.phase = .forwarding
+            ready.insert(flow.id)
+        }
+        return true
+    }
+
+    private func acceptLocal(_ context: RuleContext) {
+        for _ in 0..<32 {
+            let descriptor = Darwin.accept(context.socket, nil, nil)
+            guard descriptor >= 0 else { break }
+            makeNonblocking(descriptor)
+            var yes: Int32 = 1
+            setsockopt(descriptor, SOL_SOCKET, SO_NOSIGPIPE, &yes, socklen_t(MemoryLayout<Int32>.size))
+            let flow = Flow(socket: descriptor, context: context)
+            register(flow)
+            watch(flow)
+            if flow.phase == .opening { pendingOpens.insert(flow.id) }
+        }
+        schedulePump()
+    }
+
+    private func acceptRemote(_ context: RuleContext) -> Bool {
+        guard let listener = context.remoteListener, let host = context.rule.targetHost,
+            let port = context.rule.targetPort
+        else { return false }
+        var progressed = false
+        for _ in 0..<32 {
+            guard let channel = libssh2_channel_forward_accept(listener) else {
+                if let session, libssh2_session_last_errno(session) == LIBSSH2_ERROR_EAGAIN { blocked(session) }
+                break
+            }
+            progressed = true
+            let flow = Flow(socket: -1, context: context, channel: channel)
+            register(flow)
+            let connector = SSHForwardingConnector(queue: queue) { [weak self, weak flow] result in
+                guard let self, let flow, !flow.retired else {
+                    if case .success(let socket) = result { Darwin.close(socket) }
+                    return
+                }
+                flow.connector = nil
+                switch result {
+                case .success(let socket):
+                    flow.stream.socket = socket
+                    flow.phase = .forwarding
+                    self.watch(flow)
+                    self.ready.insert(flow.id)
+                case .failure: self.close(flow)
+                }
+                self.schedulePump()
+            }
+            flow.connector = connector
+            connector.start(host: host, port: port)
+        }
+        return progressed
+    }
+
+    private func register(_ flow: Flow) {
+        flows[flow.id] = flow
+        flow.context.flows[flow.id] = flow
+        ready.insert(flow.id)
+    }
+
+    private func watch(_ flow: Flow) {
+        flow.events = SSHSocketEvents(socket: flow.stream.socket, queue: queue) { [weak self, weak flow] in
+            guard let self, let flow, !flow.retired else { return }
+            self.wakeAll()
+        }
+        updateInterest(flow)
+    }
+
+    private func updateInterest(_ flow: Flow) {
+        let reading: Bool
+        switch flow.phase {
+        case .forwarding: reading = flow.stream.wantsSocketRead
+        case .greeting, .request: reading = !flow.stream.socketEOF
+        case .opening, .connecting: reading = false
+        }
+        flow.events?.update(read: reading, write: flow.stream.socketWriteBlocked)
+    }
+
+    private func pump(_ flow: Flow, session: OpaquePointer) -> Bool {
+        guard !flow.retired else { return false }
+        var progressed = false
+        do {
+            switch flow.phase {
+            case .forwarding:
+                progressed = try flow.stream.pump(session: session, scratch: &scratch)
+                if flow.stream.finished {
+                    close(flow)
+                    return true
+                }
+            case .greeting, .request:
+                if !flow.stream.socketEOF {
+                    let count = Darwin.recv(flow.stream.socket, &scratch, scratch.count, 0)
+                    if count > 0 {
+                        flow.handshake.append(contentsOf: scratch.prefix(count))
+                        guard flow.handshake.count <= 64 * 1024 else {
+                            close(flow)
+                            return true
+                        }
+                        handleSOCKSHandshake(flow, session: session)
+                        progressed = true
+                    } else if count == 0 {
+                        flow.stream.socketEOF = true
+                        progressed = true
+                    } else if errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR {
+                        close(flow)
+                        return true
+                    }
+                }
+                if try flow.stream.flushSocket() { progressed = true }
+                if flow.stream.socketEOF && flow.stream.toSocket.isEmpty {
+                    close(flow)
+                    return true
+                }
+            case .opening:
+                if try flow.stream.flushSocket() { progressed = true }
+            case .connecting: break
+            }
+        } catch {
+            close(flow)
+            return true
+        }
+        updateInterest(flow)
+        return progressed
+    }
+
+    private func report(_ context: RuleContext) {
+        guard !context.stopped, context.start == nil, context.reportedCount != context.flows.count else { return }
+        context.reportedCount = context.flows.count
+        connectionHandler(context.rule.id, context.reportedCount)
+    }
+
+    private func close(_ flow: Flow) {
+        guard !flow.retired else { return }
+        flow.retired = true
+        flows[flow.id] = nil
+        flow.context.flows[flow.id] = nil
+        flow.connector?.cancel()
+        flow.connector = nil
+        let descriptor = flow.stream.socket
+        flow.stream.socket = -1
+        if let events = flow.events {
+            events.cancel { Darwin.close(descriptor) }
+            flow.events = nil
+        } else if descriptor >= 0 {
+            Darwin.close(descriptor)
+        }
+        if flow.stream.channel != nil { retiredFlows.append(flow) }
+    }
+
+    private func close(_ context: RuleContext, completion: @escaping @Sendable () -> Void) {
+        context.stopped = true
+        context.start?.resume(throwing: CancellationError())
+        context.start = nil
+        for flow in Array(context.flows.values) { close(flow) }
+        if context === listening {
+            // Drain the in-flight global request and cancel the resulting
+            // listener before allowing a replacement to bind the same port.
+            context.stop = completion
+            return
+        }
+        if let listener = context.remoteListener {
+            retiredListeners.append(RetiredListener(listener, completion: completion))
+            context.remoteListener = nil
+        } else if let events = context.events {
+            let descriptor = context.socket
+            context.socket = -1
+            context.events = nil
+            events.cancel {
+                Darwin.close(descriptor)
+                completion()
+            }
+        } else {
+            completion()
         }
     }
 
+    private func reap(_ session: OpaquePointer) -> Bool {
+        var progressed = false
+        retiredFlows.removeAll { flow in
+            guard flow.stream.finishPendingWrite(session: session), let channel = flow.stream.channel else {
+                blocked(session)
+                return false
+            }
+            flow.stream.toChannel.removeAll()
+            flow.stream.toSocket.removeAll()
+            let result = libssh2_channel_free(channel)
+            if result == LIBSSH2_ERROR_EAGAIN {
+                blocked(session)
+                return false
+            }
+            flow.stream.channel = nil
+            progressed = true
+            return true
+        }
+        retiredListeners.removeAll { entry in
+            let result = libssh2_channel_forward_cancel(entry.listener)
+            if result == LIBSSH2_ERROR_EAGAIN {
+                blocked(session)
+                return false
+            }
+            entry.completion()
+            progressed = true
+            return true
+        }
+        return progressed
+    }
+
+    private func fail(_ session: OpaquePointer) {
+        let message = SSHSessionWorker.sessionError(
+            session, prefix: String(localized: "The forwarding SSH connection closed")
+        ).localizedDescription
+        cleanup()
+        failureHandler(message)
+    }
+
+    private func cleanup() {
+        generation += 1
+        scheduled = false
+        timer?.cancel()
+        timer = nil
+        for context in rules.values { close(context, completion: {}) }
+        rules.removeAll()
+        // Shutdown makes disposal nonblocking even if a peer stopped reading.
+        if socket >= 0 { _ = Darwin.shutdown(socket, SHUT_RDWR) }
+        if let session { bufferedTransport?.dispose(session: session) }
+        bufferedTransport = nil
+        session = nil
+        for entry in retiredListeners { entry.completion() }
+        retiredListeners.removeAll()
+        retiredFlows.removeAll()
+        opening = nil
+        listening?.stop?()
+        listening?.stop = nil
+        listening = nil
+        pendingListeners.removeAll()
+        flows.removeAll()
+        ready = SSHReadyQueue()
+        pendingOpens = SSHReadyQueue()
+        if let transport = routeTransport {
+            if let events = transportEvents { events.cancel { transport.close() } } else { transport.close() }
+        }
+        transportEvents = nil
+        routeTransport = nil
+        socket = -1
+    }
     private func handleSOCKSHandshake(_ flow: Flow, session: OpaquePointer) {
         switch flow.phase {
         case .greeting:
@@ -358,19 +664,19 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable {
             guard flow.handshake[0] == 0x05,
                 flow.handshake[2..<(2 + methods)].contains(0x00)
             else {
-                flow.toSocket.append(contentsOf: [0x05, 0xff])
-                flow.socketEOF = true
+                flow.stream.toSocket.append(contentsOf: [0x05, 0xff])
+                flow.stream.socketEOF = true
                 return
             }
-            flow.handshake.removeFirst(2 + methods)
-            flow.toSocket.append(contentsOf: [0x05, 0x00])
+            flow.handshake = Data(flow.handshake.dropFirst(2 + methods))
+            flow.stream.toSocket.append(contentsOf: [0x05, 0x00])
             flow.phase = .request
             handleSOCKSHandshake(flow, session: session)
         case .request:
             guard flow.handshake.count >= 4 else { return }
             guard flow.handshake[0] == 0x05, flow.handshake[1] == 0x01 else {
-                flow.toSocket.append(socksReply(0x07))
-                flow.socketEOF = true
+                flow.stream.toSocket.append(socksReply(0x07))
+                flow.stream.socketEOF = true
                 return
             }
             let parsed: (String, Int, Int)?
@@ -404,27 +710,18 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable {
                     Int(flow.handshake[20]) << 8 | Int(flow.handshake[21]), 22
                 )
             default:
-                flow.toSocket.append(socksReply(0x08))
-                flow.socketEOF = true
+                flow.stream.toSocket.append(socksReply(0x08))
+                flow.stream.socketEOF = true
                 return
             }
             guard let (host, port, consumed) = parsed else { return }
-            flow.handshake.removeFirst(consumed)
-            flow.channel = host.withCString { target in
-                libssh2_channel_direct_tcpip_ex(session, target, Int32(port), "127.0.0.1", 0)
-            }
-            guard flow.channel != nil else {
-                flow.toSocket.append(socksReply(0x05))
-                flow.socketEOF = true
-                return
-            }
-            flow.toSocket.append(socksReply(0x00))
-            flow.phase = .forwarding
-            if !flow.handshake.isEmpty {
-                flow.toChannel.append(flow.handshake)
-                flow.handshake.removeAll()
-            }
-        case .forwarding: break
+            flow.target = (host, port)
+            flow.stream.toChannel.append(Data(flow.handshake.dropFirst(consumed)))
+            flow.handshake.removeAll()
+            flow.phase = .opening
+            pendingOpens.insert(flow.id)
+            schedulePump()
+        case .opening, .connecting, .forwarding: break
         }
     }
 
@@ -467,45 +764,4 @@ private nonisolated final class SSHForwardingWorker: @unchecked Sendable {
         if flags >= 0 { _ = fcntl(socket, F_SETFL, flags | O_NONBLOCK) }
     }
 
-    private func close(_ flow: Flow, in context: RuleContext) {
-        context.flows[flow.id] = nil
-        if flow.socket >= 0 {
-            Darwin.close(flow.socket)
-            flow.socket = -1
-        }
-        if let channel = flow.channel {
-            _ = libssh2_channel_close(channel)
-            _ = libssh2_channel_free(channel)
-            flow.channel = nil
-        }
-    }
-
-    private func close(_ context: RuleContext) {
-        for flow in Array(context.flows.values) { close(flow, in: context) }
-        if context.socket >= 0 {
-            Darwin.close(context.socket)
-            context.socket = -1
-        }
-        if let listener = context.remoteListener {
-            _ = libssh2_channel_forward_cancel(listener)
-            context.remoteListener = nil
-        }
-    }
-
-    private func cleanup() {
-        timer?.cancel()
-        timer = nil
-        for context in Array(rules.values) { close(context) }
-        rules.removeAll()
-        if let session {
-            _ = libssh2_session_disconnect_ex(
-                session, SSH_DISCONNECT_BY_APPLICATION, "Termind stopped forwarding", ""
-            )
-            _ = libssh2_session_free(session)
-            self.session = nil
-        }
-        socket = -1
-        routeTransport?.close()
-        routeTransport = nil
-    }
 }

Frameworks/TermindRemoteInfrastructure/SSH/SSHRoutedSocket.swift

@@ -420,19 +420,18 @@ private nonisolated final class SSHRouteOpeningState: @unchecked Sendable {
 /// to this queue.
 private nonisolated final class SSHChannelSocketBridge: @unchecked Sendable {
     public let clientSocket: Int32
-
     private let queue = DispatchQueue(label: "com.akinokaede.termind.ssh.route-bridge")
     private var transportSocket: Int32
-    private var pumpSocket: Int32
     private var session: OpaquePointer?
-    private var channel: OpaquePointer?
+    private let bufferedTransport: SSHBufferedTransport
+    private let stream: SSHStreamPump
+    private var transportEvents: SSHSocketEvents?
+    private var socketEvents: SSHSocketEvents?
     private var timer: (any DispatchSourceTimer)?
-    private var toChannel = Data()
-    private var toSocket = Data()
-    private var socketEOF = false
-    private var channelEOF = false
-    private var channelWriteClosed = false
+    private var scratch = [UInt8](repeating: 0, count: 32 * 1024)
     private var isClosed = false
+    private var scheduled = false
+    private var keepaliveDue = true
 
     public init(
         transportSocket: Int32, session: OpaquePointer, channel: OpaquePointer,
@@ -447,8 +446,8 @@ private nonisolated final class SSHChannelSocketBridge: @unchecked Sendable {
         }
         self.transportSocket = transportSocket
         self.session = session
-        self.channel = channel
-        pumpSocket = pair[0]
+        bufferedTransport = SSHBufferedTransport(session: session)
+        stream = SSHStreamPump(socket: pair[0], channel: channel)
         clientSocket = pair[1]
 
         var yes: Int32 = 1
@@ -467,97 +466,77 @@ private nonisolated final class SSHChannelSocketBridge: @unchecked Sendable {
             clientSocket, SOL_SOCKET, SO_SNDTIMEO, &interval,
             socklen_t(MemoryLayout<timeval>.size)
         )
-        let flags = fcntl(pumpSocket, F_GETFL, 0)
-        if flags >= 0 { _ = fcntl(pumpSocket, F_SETFL, flags | O_NONBLOCK) }
+        let flags = fcntl(stream.socket, F_GETFL, 0)
+        if flags >= 0 { _ = fcntl(stream.socket, F_SETFL, flags | O_NONBLOCK) }
         libssh2_session_set_blocking(session, 0)
 
-        let source = DispatchSource.makeTimerSource(queue: queue)
-        source.schedule(deadline: .now(), repeating: .milliseconds(2), leeway: .milliseconds(1))
-        source.setEventHandler { [weak self] in self?.pump() }
-        timer = source
-        source.resume()
+        libssh2_keepalive_config(session, 0, 20)
+        queue.async { [self] in startEvents() }
     }
 
-    public func close() {
-        queue.sync { closeOnQueue() }
+    private func startEvents() {
+        guard !isClosed else { return }
+        transportEvents = SSHSocketEvents(socket: transportSocket, queue: queue) { [weak self] in self?.schedule() }
+        socketEvents = SSHSocketEvents(socket: stream.socket, queue: queue) { [weak self] in self?.schedule() }
+        transportEvents?.update(read: true, write: false)
+        socketEvents?.update(read: true, write: false)
+        let source = DispatchSource.makeTimerSource(queue: queue)
+        source.schedule(deadline: .now() + 20, repeating: .seconds(20), leeway: .seconds(1))
+        source.setEventHandler { [weak self] in
+            self?.keepaliveDue = true
+            self?.schedule()
+        }
+        timer = source
+        source.resume()
+        schedule()
+    }
+
+    public func close() { queue.sync { closeOnQueue() } }
+
+    private func schedule() {
+        guard !isClosed, !scheduled else { return }
+        scheduled = true
+        queue.async { [weak self] in
+            guard let self, !self.isClosed else { return }
+            self.scheduled = false
+            self.pump()
+        }
     }
 
     private func pump() {
-        guard !isClosed, let session, let channel else { return }
-        let highWaterMark = 2 * 1024 * 1024
-
-        if !socketEOF, toChannel.count < highWaterMark {
-            var bytes = [UInt8](repeating: 0, count: 32 * 1024)
-            let count = Darwin.recv(pumpSocket, &bytes, bytes.count, 0)
-            if count > 0 {
-                toChannel.append(contentsOf: bytes.prefix(count))
-            } else if count == 0 {
-                socketEOF = true
-            } else if errno != EAGAIN && errno != EWOULDBLOCK {
+        guard !isClosed, let session else { return }
+        do {
+            try bufferedTransport.flush(socket: transportSocket)
+            if bufferedTransport.isBackpressured {
+                transportEvents?.update(read: false, write: true)
+                socketEvents?.update(read: false, write: stream.socketWriteBlocked)
+                return
+            }
+            bufferedTransport.beginPass()
+            let progressed = try stream.pump(session: session, scratch: &scratch)
+            if stream.finished {
                 closeOnQueue()
                 return
             }
-        }
-
-        while !toChannel.isEmpty {
-            let written = toChannel.withUnsafeBytes { bytes -> Int in
-                guard let base = bytes.baseAddress?.assumingMemoryBound(to: CChar.self) else { return 0 }
-                return libssh2_channel_write_ex(channel, 0, base, bytes.count)
-            }
-            if written > 0 {
-                toChannel.removeFirst(written)
-            } else if written == LIBSSH2_ERROR_EAGAIN {
-                break
-            } else {
-                closeOnQueue()
-                return
-            }
-        }
-        if socketEOF, toChannel.isEmpty, !channelWriteClosed {
-            let result = libssh2_channel_send_eof(channel)
-            if result == 0 {
-                channelWriteClosed = true
-            } else if result != LIBSSH2_ERROR_EAGAIN {
-                closeOnQueue()
-                return
-            }
-        }
-
-        if toSocket.count < highWaterMark {
-            var bytes = [CChar](repeating: 0, count: 32 * 1024)
-            while toSocket.count < highWaterMark {
-                let count = libssh2_channel_read_ex(channel, 0, &bytes, bytes.count)
-                if count > 0 {
-                    toSocket.append(Data(bytes: bytes, count: count))
-                } else if count == LIBSSH2_ERROR_EAGAIN || count == 0 {
-                    break
-                } else {
+            if keepaliveDue {
+                var next: Int32 = 0
+                let result = libssh2_keepalive_send(session, &next)
+                if result == LIBSSH2_ERROR_EAGAIN {
+                    stream.blocked(session)
+                } else if result < 0 {
                     closeOnQueue()
                     return
+                } else {
+                    keepaliveDue = false
                 }
             }
-        }
-        if libssh2_channel_eof(channel) == 1 { channelEOF = true }
-
-        while !toSocket.isEmpty {
-            let sent = toSocket.withUnsafeBytes { bytes -> Int in
-                guard let base = bytes.baseAddress else { return 0 }
-                return Darwin.send(pumpSocket, base, bytes.count, 0)
-            }
-            if sent > 0 {
-                toSocket.removeFirst(sent)
-            } else if errno == EAGAIN || errno == EWOULDBLOCK {
-                break
-            } else {
-                closeOnQueue()
-                return
-            }
-        }
-        if channelEOF && toSocket.isEmpty {
-            _ = Darwin.shutdown(pumpSocket, SHUT_WR)
-        }
-        var next = Int32(0)
-        _ = libssh2_keepalive_send(session, &next)
+            try bufferedTransport.flush(socket: transportSocket)
+            socketEvents?.update(read: stream.wantsSocketRead, write: stream.socketWriteBlocked)
+            transportEvents?.update(
+                read: !bufferedTransport.isBackpressured && (stream.wantsChannelRead || !stream.toChannel.isEmpty),
+                write: bufferedTransport.wantsWrite || stream.blockDirections & LIBSSH2_SESSION_BLOCK_OUTBOUND != 0)
+            if progressed || bufferedTransport.exhaustedReadBudget { schedule() }
+        } catch { closeOnQueue() }
     }
 
     private func closeOnQueue() {
@@ -565,25 +544,17 @@ private nonisolated final class SSHChannelSocketBridge: @unchecked Sendable {
         isClosed = true
         timer?.cancel()
         timer = nil
-        if pumpSocket >= 0 {
-            Darwin.close(pumpSocket)
-            pumpSocket = -1
-        }
-        if let channel {
-            _ = libssh2_channel_close(channel)
-            _ = libssh2_channel_free(channel)
-            self.channel = nil
-        }
-        if let session {
-            _ = libssh2_session_disconnect_ex(
-                session, SSH_DISCONNECT_BY_APPLICATION, "Termind closed the jump route", ""
-            )
-            _ = libssh2_session_free(session)
-            self.session = nil
-        }
-        if transportSocket >= 0 {
-            Darwin.close(transportSocket)
-            transportSocket = -1
-        }
+        let local = stream.socket
+        stream.socket = -1
+        if let socketEvents { socketEvents.cancel { Darwin.close(local) } } else { Darwin.close(local) }
+        socketEvents = nil
+        let remote = transportSocket
+        transportSocket = -1
+        _ = Darwin.shutdown(remote, SHUT_RDWR)
+        if let session { bufferedTransport.dispose(session: session) }
+        session = nil
+        stream.channel = nil
+        if let transportEvents { transportEvents.cancel { Darwin.close(remote) } } else { Darwin.close(remote) }
+        transportEvents = nil
     }
 }

Frameworks/TermindRemoteInfrastructure/SSH/SSHStreamPump.swift

@@ -0,0 +1,349 @@
+import CSSH2
+import Darwin
+import Foundation
+
+/// Queue-confined chunks keep libssh2's retry pointer and length stable even when
+/// new socket data arrives while an earlier write is blocked.
+nonisolated final class SSHByteQueue {
+    private final class Chunk {
+        let bytes: UnsafeMutableRawPointer
+        let size: Int
+        var offset = 0
+
+        init(_ source: UnsafeRawBufferPointer) {
+            size = source.count
+            bytes = .allocate(byteCount: size, alignment: 1)
+            bytes.copyMemory(from: source.baseAddress!, byteCount: size)
+        }
+
+        deinit { bytes.deallocate() }
+    }
+
+    private var chunks: [Chunk?] = []
+    private var head = 0
+    private(set) var count = 0
+    var isEmpty: Bool { count == 0 }
+
+    func append(_ data: Data) { data.withUnsafeBytes { append($0) } }
+    func append(contentsOf bytes: [UInt8]) { bytes.withUnsafeBytes { append($0) } }
+
+    func append(_ bytes: UnsafeRawBufferPointer) {
+        var offset = 0
+        while offset < bytes.count {
+            let end = min(offset + 32 * 1024, bytes.count)
+            chunks.append(Chunk(UnsafeRawBufferPointer(rebasing: bytes[offset..<end])))
+            offset = end
+        }
+        count += bytes.count
+    }
+
+    func withFirstBytes<T>(_ body: (UnsafeRawBufferPointer) -> T) -> T {
+        guard head < chunks.count, let chunk = chunks[head] else {
+            return body(UnsafeRawBufferPointer(start: nil, count: 0))
+        }
+        return body(UnsafeRawBufferPointer(start: chunk.bytes + chunk.offset, count: chunk.size - chunk.offset))
+    }
+
+    func removeAll() {
+        chunks.removeAll()
+        head = 0
+        count = 0
+    }
+
+    func consume(_ size: Int) {
+        precondition(size > 0 && size <= count)
+        let chunk = chunks[head]!
+        precondition(size <= chunk.size - chunk.offset)
+        chunk.offset += size
+        count -= size
+        if chunk.offset == chunk.size {
+            chunks[head] = nil
+            head += 1
+            if head == chunks.count {
+                chunks.removeAll(keepingCapacity: true)
+                head = 0
+            } else if head >= 64 && head * 2 >= chunks.count {
+                chunks.removeFirst(head)
+                head = 0
+            }
+        }
+    }
+}
+
+/// Sources, interest changes and their handlers all belong to the supplied queue.
+/// Closing the descriptor must wait for both cancellation handlers, including
+/// when a source was suspended for backpressure.
+nonisolated final class SSHSocketEvents {
+    private let read: any DispatchSourceRead
+    private let write: any DispatchSourceWrite
+    private let queue: DispatchQueue
+    private let cancelled = DispatchGroup()
+    private var reading = false
+    private var writing = false
+    private var stopped = false
+
+    init(socket: Int32, queue: DispatchQueue, ready: @escaping @Sendable () -> Void) {
+        self.queue = queue
+        read = DispatchSource.makeReadSource(fileDescriptor: socket, queue: queue)
+        write = DispatchSource.makeWriteSource(fileDescriptor: socket, queue: queue)
+        for source in [read as any DispatchSourceProtocol, write as any DispatchSourceProtocol] {
+            cancelled.enter()
+            source.setCancelHandler { [cancelled] in cancelled.leave() }
+            source.setEventHandler { [weak self] in
+                guard let self, !self.stopped else { return }
+                ready()
+            }
+        }
+    }
+
+    func update(read wantsRead: Bool, write wantsWrite: Bool) {
+        guard !stopped else { return }
+        if wantsRead != reading {
+            reading = wantsRead
+            if reading { read.resume() } else { read.suspend() }
+        }
+        if wantsWrite != writing {
+            writing = wantsWrite
+            if writing { write.resume() } else { write.suspend() }
+        }
+    }
+
+    func cancel(then completion: @escaping @Sendable () -> Void = {}) {
+        guard !stopped else { return }
+        stopped = true
+        read.cancel()
+        write.cancel()
+        if !reading { read.resume() }
+        if !writing { write.resume() }
+        cancelled.notify(queue: queue, execute: completion)
+    }
+}
+
+/// Deduplicated FIFO; a hot stream goes behind its peers rather than monopolizing
+/// the session queue. Also used for listener/channel work that must retain order.
+nonisolated struct SSHReadyQueue<Element: Hashable> {
+    private var items: [Element] = []
+    private var members: Set<Element> = []
+    private var head = 0
+    var isEmpty: Bool { members.isEmpty }
+
+    mutating func insert(_ item: Element) {
+        if members.insert(item).inserted { items.append(item) }
+    }
+
+    mutating func pop() -> Element? {
+        guard head < items.count else { return nil }
+        let item = items[head]
+        members.remove(item)
+        head += 1
+        if head == items.count {
+            items.removeAll(keepingCapacity: true)
+            head = 0
+        } else if head >= 64 && head * 2 >= items.count {
+            items.removeFirst(head)
+            head = 0
+        }
+        return item
+    }
+}
+
+/// Internal I/O seam: production uses libssh2; tests can force short writes,
+/// blocked windows and EOF independently of operating-system timing.
+nonisolated struct SSHChannelIO {
+    var read: (OpaquePointer, UnsafeMutableRawBufferPointer) -> Int = {
+        libssh2_channel_read_ex($0, 0, $1.baseAddress!.assumingMemoryBound(to: CChar.self), $1.count)
+    }
+    var write: (OpaquePointer, UnsafeRawBufferPointer) -> Int = {
+        libssh2_channel_write_ex($0, 0, $1.baseAddress!.assumingMemoryBound(to: CChar.self), $1.count)
+    }
+    var writeWindow: (OpaquePointer) -> UInt = { libssh2_channel_window_write_ex($0, nil) }
+    var eof: (OpaquePointer) -> Int32 = { libssh2_channel_eof($0) }
+    var sendEOF: (OpaquePointer) -> Int32 = { libssh2_channel_send_eof($0) }
+    var directions: (OpaquePointer) -> Int32 = { libssh2_session_block_directions($0) }
+}
+
+/// A socket/channel pair. Its owner serializes every call with all other calls
+/// on the same SSH session, and provides a reusable scratch buffer.
+nonisolated final class SSHStreamPump {
+    static let highWater = 2 * 1024 * 1024
+    static let lowWater = 1024 * 1024
+    static let quantum = 256 * 1024
+
+    private let io: SSHChannelIO
+    var socket: Int32
+    var channel: OpaquePointer?
+    let toChannel = SSHByteQueue()
+    let toSocket = SSHByteQueue()
+    var socketEOF = false
+    var channelEOF = false
+    var socketWriteClosed = false
+    var channelWriteClosed = false
+    private(set) var writePending = false
+    private(set) var blockDirections: Int32 = 0
+    private var socketReadPaused = false
+    private var channelReadPaused = false
+    private(set) var socketWriteBlocked = false
+
+    init(socket: Int32, channel: OpaquePointer? = nil, io: SSHChannelIO = SSHChannelIO()) {
+        self.io = io
+        self.socket = socket
+        self.channel = channel
+    }
+
+    var wantsSocketRead: Bool { !socketEOF && !socketReadPaused }
+    var wantsChannelRead: Bool { !channelEOF && !channelReadPaused }
+    var finished: Bool { socketEOF && channelEOF && toChannel.isEmpty && toSocket.isEmpty }
+
+    func blocked(_ session: OpaquePointer) {
+        blockDirections |= io.directions(session)
+    }
+
+    /// Returns whether bytes/EOF moved. Even a final successful read can have
+    /// buffered packets for another channel: the owner schedules another sweep.
+    func pump(session: OpaquePointer, scratch: inout [UInt8]) throws -> Bool {
+        blockDirections = 0
+        socketWriteBlocked = false
+        var progressed = false
+        var waitingForWindow = false
+        if socketReadPaused && toChannel.count <= Self.lowWater { socketReadPaused = false }
+        if channelReadPaused && toSocket.count <= Self.lowWater { channelReadPaused = false }
+        var remaining = Self.quantum
+        while wantsSocketRead && remaining > 0 {
+            let capacity = min(scratch.count, remaining, Self.highWater - toChannel.count)
+            if capacity == 0 {
+                socketReadPaused = true
+                break
+            }
+            let count = Darwin.recv(socket, &scratch, capacity, 0)
+            if count > 0 {
+                scratch.withUnsafeBytes { toChannel.append(UnsafeRawBufferPointer(rebasing: $0[..<count])) }
+                remaining -= count
+                progressed = true
+            } else if count == 0 {
+                socketEOF = true
+                progressed = true
+            } else if errno == EINTR {
+                continue
+            } else if errno == EAGAIN || errno == EWOULDBLOCK {
+                break
+            } else {
+                throw POSIXError(POSIXErrorCode(rawValue: errno) ?? .EIO)
+            }
+        }
+        if toChannel.count >= Self.highWater { socketReadPaused = true }
+        if let channel {
+            remaining = Self.quantum
+            while !toChannel.isEmpty && remaining > 0 {
+                let written = try write(session: session, channel: channel)
+                if written == 0 {
+                    waitingForWindow = io.writeWindow(channel) == 0
+                    break
+                }
+                remaining -= written
+                progressed = true
+            }
+            if socketEOF && toChannel.isEmpty && !channelWriteClosed {
+                let result = io.sendEOF(channel)
+                if result == 0 {
+                    channelWriteClosed = true
+                    progressed = true
+                } else if result == LIBSSH2_ERROR_EAGAIN {
+                    blocked(session)
+                } else {
+                    throw StreamError.closed
+                }
+            }
+            remaining = Self.quantum
+            while wantsChannelRead && remaining > 0 {
+                let capacity = min(scratch.count, remaining, Self.highWater - toSocket.count)
+                if capacity == 0 {
+                    channelReadPaused = true
+                    break
+                }
+                let count = scratch.withUnsafeMutableBytes {
+                    io.read(channel, UnsafeMutableRawBufferPointer(rebasing: $0[..<capacity]))
+                }
+                if count > 0 {
+                    scratch.withUnsafeBytes { toSocket.append(UnsafeRawBufferPointer(rebasing: $0[..<count])) }
+                    remaining -= count
+                    progressed = true
+                } else if count == LIBSSH2_ERROR_EAGAIN || count == 0 {
+                    blocked(session)
+                    break
+                } else {
+                    throw StreamError.closed
+                }
+            }
+            if !channelEOF && io.eof(channel) == 1 {
+                channelEOF = true
+                progressed = true
+            }
+        }
+        if toSocket.count >= Self.highWater { channelReadPaused = true }
+        if try flushSocket() { progressed = true }
+        if channelEOF && toSocket.isEmpty && !socketWriteClosed {
+            _ = Darwin.shutdown(socket, SHUT_WR)
+            socketWriteClosed = true
+            progressed = true
+        }
+        if waitingForWindow, let channel, io.writeWindow(channel) > 0 { progressed = true }
+        return progressed
+    }
+
+    func flushSocket() throws -> Bool {
+        var remaining = Self.quantum
+        var progressed = false
+        socketWriteBlocked = false
+        while !toSocket.isEmpty && remaining > 0 {
+            let sent = toSocket.withFirstBytes { Darwin.send(socket, $0.baseAddress!, $0.count, 0) }
+            if sent > 0 {
+                toSocket.consume(sent)
+                remaining -= sent
+                progressed = true
+            } else if sent < 0 && errno == EINTR {
+                continue
+            } else if sent < 0 && (errno == EAGAIN || errno == EWOULDBLOCK) {
+                socketWriteBlocked = true
+                break
+            } else {
+                throw StreamError.closed
+            }
+        }
+        return progressed
+    }
+
+    /// Finish an already-started write before freeing a retired channel. The SSH
+    /// transport may still own this channel's packet, so simply dropping it can
+    /// prevent every peer from writing.
+    func finishPendingWrite(session: OpaquePointer) -> Bool {
+        guard writePending, let channel else { return true }
+        // A zero-window write has not entered libssh2's packet-send state.
+        // Cancellation must not wait for a peer that will never read again.
+        if io.writeWindow(channel) == 0 {
+            writePending = false
+            return true
+        }
+        do { _ = try write(session: session, channel: channel) } catch { writePending = false }
+        return !writePending
+    }
+
+    private func write(session: OpaquePointer, channel: OpaquePointer) throws -> Int {
+        let written = toChannel.withFirstBytes {
+            io.write(channel, $0)
+        }
+        if written > 0 {
+            writePending = false
+            toChannel.consume(written)
+            return written
+        }
+        if written == LIBSSH2_ERROR_EAGAIN || written == 0 {
+            writePending = true
+            blocked(session)
+            return 0
+        }
+        writePending = false
+        throw StreamError.closed
+    }
+
+    enum StreamError: Error { case closed }
+}

Memory/ARCHITECTURE.md

@@ -254,6 +254,28 @@ transport failure fails all current rules without restarting them. Host route ed
 (including jump hosts) and rule replacement stop affected runs. Authentication's own
 Continue and Save may continue its explicitly requested opening. Connection and rule
 generations reject late callbacks, and stop/start serializes listener cleanup.
+Forwarding and jump-host bridges service sockets through DispatchSource readiness
+on their owning serial queues; every libssh2 session has exactly one queue owner.
+The shared stream pump retains stable write chunks across EAGAIN, uses a 256 KiB
+per-direction quantum, caps encrypted reads at 256 KiB per session pass (including
+reads performed internally by libssh2 writes), and pauses input at a 2 MiB high
+watermark until the queue falls below 1 MiB. A deduplicated FIFO and a 2 ms worker slice prevent a hot flow
+from monopolizing its peers. Half-close drains accepted bytes before propagating
+EOF. Remote target DNS uses at most four workers; nonblocking TCP connects and a
+five-second deadline never block SSH I/O. Channel opens and remote listener starts
+retain and retry the library's session-wide operation state.
+
+Authenticated forwarding/bridge sessions install an encrypted-output adapter using
+libssh2's public send/receive callbacks. It accepts complete encrypted packets and
+applies bounded backpressure between library calls, including receive-generated
+replies. This avoids the pinned libssh2 forwarded-tcpip confirmation retry losing
+its channel on socket EAGAIN. It is installed only after authentication has released
+the session abstract context. Teardown switches the adapter to discard writes and
+return EOF, allowing session disposal without waiting on the peer. Socket descriptors
+are closed only after dispatch-source cancellation handlers retire. Timers serve
+keepalive/protocol deadlines, never bulk-data polling; count notifications occur
+only on changes, including counts received while a rule is still starting.
+
 An Identity may contain only a username: preparation
 permits an empty credential list, SSH tries `none`, then advertised methods. Missing
 configured secrets and existing authentication failure/retry behavior stay unchanged.

Memory/PORT_FORWARDING_PERFORMANCE.md

@@ -0,0 +1,178 @@
+# Port forwarding concurrency and transfer evaluation
+
+Measured on 2026-09-19, Apple M3 Pro, 36 GiB RAM, macOS 26.6.2, Xcode 27.
+
+## Implementation
+
+Local, remote and SOCKS5 forwarding share an event-driven stream pump with jump
+bridges. Socket readiness replaces the old repeating data timer. Each SSH session
+still has one serial owner: SSH channels run concurrently without concurrently
+calling libssh2 on the same session. A ready FIFO, per-flow transfer quantum and
+session input budget bound each scheduling pass. Immutable chunks preserve the
+write address and length required by libssh2 across EAGAIN.
+
+Remote connects are nonblocking, DNS uses four resolver workers, and numeric
+addresses bypass DNS workers. Half-close drains data, high/low watermarks stop
+unbounded application queues, and source cancellation precedes descriptor close.
+Listener/channel creation retries EAGAIN; stopping an in-flight remote listener
+waits for its request and cancellation before permitting a replacement. Starting
+connection counts are retained instead of resetting the UI count to zero.
+
+The pinned libssh2 implementation clears its pending forwarded-tcpip channel when
+retrying a blocked open confirmation (`packet_queue_listener` in `packet.c`). This
+caused remote forwarding through jumps to stall even with the old bridge. The new
+adapter uses public send/receive callbacks to queue complete encrypted packets and
+apply backpressure outside the library. No dependency pins or private ABI changed.
+
+## Method
+
+The opt-in tests in `TFMTests/SSHForwardingWireTests.swift` use
+`Scripts/ssh-authentication-fixture.py` with Paramiko 4.0.0 and loopback echo targets.
+Every received payload and final EOF is checked. Metrics are printed only after all
+streams complete successfully. Dedicated client threads avoid exhausting the
+Dispatch worker pool used by the implementation.
+
+- Request/response: 1, 16, 64 streams; nine 64 KiB exchanges per stream.
+- Sustained duplex echo: 1, 16, 64 streams; 129 × 64 KiB per stream per direction,
+  with independent producers and consumers (516 MiB per direction at 64 streams).
+- Jumps: each forwarding kind, 16 streams, one and two SSH bridges.
+- Mixed: eight duplex bulk streams plus sixteen 64-byte request streams, 0/1/2 jumps.
+- Lifecycle: slow consumer, repeated disconnect under backpressure, independent
+  listeners, in-flight listener cancellation, port reuse, fragmented SOCKS handshake.
+- Idle: an established forwarding connection through two jumps, observed for five seconds.
+
+All throughput values are useful payload **per direction**, including connection
+setup in elapsed time. `ready_p95_ms` measures TCP connect, channel/target setup and
+first echoed probe; SOCKS negotiation is included. Request latency starts after
+that probe. CPU is user+system CPU of the entire test-host process, including test
+clients; fixture CPU is excluded. RSS is the process-lifetime high-water mark,
+including the app, test threads and earlier cases; it is not per-flow memory.
+
+Three fresh Release test-host runs were made sequentially without simultaneous
+builds or other test suites. Normal desktop background processes were not disabled.
+Raw successful measurements are in `PORT_FORWARDING_BENCHMARK.json`.
+
+## Results
+
+All table entries are medians of three runs; ranges show the three-run spread.
+
+| Kind | Streams | MiB/s/direction (range) | Ready p95 ms | CPU ms/total payload MiB |
+|---|---:|---:|---:|---:|
+| local | 1 | 67.3 (67.1–67.7) | 0.4 | 7.54 |
+| local | 16 | 68.5 (68.5–69.5) | 91.8 | 7.48 |
+| local | 64 | 66.6 (65.9–67.0) | 447.8 | 7.68 |
+| remote | 1 | 69.7 (69.3–70.2) | 0.6 | 7.46 |
+| remote | 16 | 69.8 (69.2–70.3) | 431.2 | 7.68 |
+| remote | 64 | 67.6 (67.6–68.0) | 1790.0 | 7.74 |
+| dynamic | 1 | 67.1 (66.2–67.2) | 0.6 | 7.56 |
+| dynamic | 16 | 70.3 (69.9–71.7) | 167.9 | 7.31 |
+| dynamic | 64 | 67.7 (65.2–67.8) | 653.7 | 7.59 |
+
+At 64 streams, aggregate throughput remains within about 5% of the single-stream
+rate, with all payloads intact. Concurrency improves service to many connections;
+it does not multiply a shared session’s bandwidth. Remote connection readiness
+under sustained load is noticeably slower (p95 about 1.79 s at 64 streams).
+
+| Mixed workload jumps | MiB/s/direction | Ready p95 ms | Small request p95 ms (range) |
+|---|---:|---:|---:|
+| 0 | 61.0 | 119.4 | 55.9 (55.3–56.8) |
+| 1 | 30.3 | 1699.4 | 56.1 (2.4–95.0) |
+| 2 | 28.9 | 1457.7 | 21.4 (3.4–122.0) |
+
+The mixed workload completes every stream, but new-channel admission remains a
+latency limitation under bulk traffic, especially through jumps. The large spread
+in post-setup latency reflects how much bulk traffic remains when small streams
+become ready. This is not evidence of consistently low interactive latency under
+saturation. Further optimization should profile admission and protocol queueing
+against an OpenSSH server before changing session-sharing/authentication policy.
+
+Across full runs, process peak RSS was 273.0, 275.3 and 272.6 MiB. The two-jump idle
+sample used 0.110–0.123 CPU seconds over five seconds (2.2–2.5% of one core for the
+entire app/test process). These figures include the host UI and test client overhead;
+they do not establish a transport-only CPU or allocation profile.
+
+Bridge comparison, 16 request/response streams, median of three runs on each implementation:
+
+| Kind | Jumps | Original bridge MiB/s | New bridge MiB/s | Ratio | CPU seconds before → after |
+|---|---:|---:|---:|---:|---:|
+| local | 1 | 3.72 | 38.57 | 10.4× | 1.813 → 0.338 |
+| local | 2 | 2.57 | 36.92 | 14.4× | 2.832 → 0.523 |
+| remote | 1 | 3.75 | 36.02 | 9.6× | 1.846 → 0.339 |
+| remote | 2 | 2.60 | 36.25 | 13.9× | 2.891 → 0.522 |
+| dynamic | 1 | 3.62 | 38.63 | 10.7× | 1.982 → 0.335 |
+| dynamic | 2 | 2.61 | 37.09 | 14.2× | 2.899 → 0.518 |
+
+The bridge comparison improves useful throughput by 9.6–14.4× for this workload.
+It also reduces CPU time to move the same payload. The original two-jump bridge
+idle samples consumed 0.485–0.531 CPU seconds per five seconds, versus 0.110–0.123
+with the new bridge. This comparison includes the app/test host in both cases.
+
+## Comparison limits
+
+The unmodified runtime at `5c88a2bf` produced no successful metric records with the
+same harness: direct channel opens were treated as failures on EAGAIN, remote
+listener creation returned “Would block”, and SOCKS runs could terminate the test
+host. A whole-implementation throughput speedup cannot be calculated from failed
+transfers. The separate bridge comparison retains the new forwarding worker and
+substitutes only the original `SSHRoutedSocket.swift`; it isolates the bridge
+implementation, not the complete old application and not the timer alone.
+
+This is an encrypted loopback/Paramiko benchmark, not a WAN or OpenSSH capacity
+claim. All channels share a serialized SSH session and server; increasing connection
+count does not promise proportional aggregate bandwidth. Mixed-load readiness and
+request latency must be considered together: a low post-setup request percentile
+can hide admission delayed by existing bulk traffic.
+
+Application plaintext buffering is bounded at 2 MiB per direction per flow, with
+resume below 1 MiB. Encrypted output is paused at 2 MiB per session, with bounded
+overshoot from the current pump/control operations. libssh2 windows, socket buffers,
+thread stacks and app memory are additional. These are per-flow/session bounds,
+not a fixed total-memory cap for an unlimited number of connections.
+
+## Reproduction
+
+Create a temporary virtualenv, install `paramiko==4.0.0`, then run:
+
+```sh
+python Scripts/ssh-authentication-fixture.py --metadata /tmp/ssh-fixture.json
+```
+
+In a second terminal:
+
+```sh
+TEST_RUNNER_TERMIND_SSH_FIXTURE=/tmp/ssh-fixture.json \
+TEST_RUNNER_TERMIND_FORWARDING_BENCHMARK=1 \
+xcodebuild -project Termind.xcodeproj -scheme TFM \
+  -destination 'platform=macOS,arch=arm64' \
+  -onlyUsePackageVersionsFromResolvedFile \
+  -only-testing:TFMTests/SSHForwardingWireTests test
+```
+
+That command uses Debug. Release measurements used an isolated worktree with
+`-configuration Release ENABLE_TESTABILITY=YES ONLY_ACTIVE_ARCH=YES`. The existing
+Release test host needed two harness-only adjustments, applied equally before and
+after: exclude FrameworkTests and unrelated Debug-only TFM tests from that host;
+link the existing ScrubberKit package product in TFM's Release target. No such
+project/dependency changes were made in the submitted tree. Subsequent repetitions
+used `test-without-building`, with the same environment variables. Do not benchmark
+alongside builds, simulator rendering tests, or other traffic generators.
+
+## Validation and outstanding suite failures
+
+- Final macOS targeted run: **56 tests passed**, covering forwarding, deterministic
+  pump/callback tests, authentication, controller journey and architecture boundaries.
+- Shared frameworks: **335 tests passed** on macOS; also **335 passed** in the iOS run.
+- Release harness: all three complete performance runs passed, with 27 successful
+  metric records each, plus lifecycle and idle cases.
+- Final iOS Simulator app build passed. Repository-wide strict Swift formatting,
+  Chinese localization validation and Python syntax checks passed.
+- The full macOS suite was attempted but did not finish cleanly: the StoreKit annual
+  product test hung and was interrupted. Timing-sensitive Agent/forwarding-rendering
+  failures passed in the focused rerun; three language variants of
+  `MetricsDetailToolbarTests.legacyCompactSearchOpensTheNativeFieldAndKeepsItsQueryWhenWidened`
+  still failed. No Metrics UI code was changed.
+- Full iOS app suite: 319 tests completed with nine issues in
+  `MobileSavedNameRenderingTests`, `MobileUploadBatchTests`, `MobileEditorToolbarTests`
+  and `MobileForwardingRenderingTests`. Its observer remained running after test
+  output and was interrupted. These failures were not repaired as part of this
+  transport change; the full repository suite is **not all green**.

Scripts/ssh-authentication-fixture.py

@@ -16,6 +16,7 @@ import select
 import signal
 import socket
 import threading
+import time
 
 import paramiko
 
@@ -101,6 +102,7 @@ class Files(paramiko.SFTPServerInterface):
 
 class Authentication(paramiko.ServerInterface):
     def __init__(self, client_key):
+        self.listeners = {}
         self.client_key = client_key
         self.user = ""
         self.factor = False
@@ -119,7 +121,7 @@ class Authentication(paramiko.ServerInterface):
 
     def check_auth_none(self, username):
         self.user = username
-        return paramiko.AUTH_SUCCESSFUL if username == "none" else paramiko.AUTH_FAILED
+        return paramiko.AUTH_SUCCESSFUL if username in ("none", "none-slow-forward") else paramiko.AUTH_FAILED
 
     def check_auth_password(self, username, password):
         if password != "fixture-password":
@@ -181,27 +183,99 @@ class Authentication(paramiko.ServerInterface):
 
     def check_channel_direct_tcpip_request(self, chanid, origin, destination):
         # Allow routing only back to this fixture, never to other services.
-        if destination != ("127.0.0.1", self.port):
+        if destination not in self.allowed_destinations:
             return paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED
         self.destinations[chanid] = destination
         return paramiko.OPEN_SUCCEEDED
 
+    def check_port_forward_request(self, address, port):
+        if self.user == "none-slow-forward":
+            time.sleep(0.1)
+        if address not in ("127.0.0.1", "::1"):
+            return False
+        listener = socket.socket(socket.AF_INET6 if ":" in address else socket.AF_INET)
+        try:
+            listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+            listener.bind((address, port))
+            listener.listen(128)
+            listener.settimeout(0.2)
+        except OSError as error:
+            print(f"Fixture bind {address}:{port} failed: {error}", flush=True)
+            listener.close()
+            return False
+        bound = listener.getsockname()[1]
+        self.listeners[(address, bound)] = listener
+
+        def accept():
+            while self.transport.is_active():
+                try:
+                    peer, origin = listener.accept()
+                    channel = self.transport.open_forwarded_tcpip_channel(origin[:2], (address, bound))
+                    threading.Thread(target=relay_peer, args=(channel, peer), daemon=True).start()
+                except socket.timeout:
+                    continue
+                except (OSError, EOFError, paramiko.SSHException):
+                    break
+
+        threading.Thread(target=accept, daemon=True).start()
+        return bound
+
+    def cancel_port_forward_request(self, address, port):
+        listener = self.listeners.pop((address, port), None)
+        if listener:
+            listener.close()
+
+
+def relay_peer(channel, peer):
+    with peer:
+        try:
+            readers = [channel, peer]
+            while readers:
+                ready, _, _ = select.select(readers, [], [], 1)
+                for source in ready:
+                    destination = peer if source is channel else channel
+                    data = source.recv(32768)
+                    if data:
+                        destination.sendall(data)
+                    else:
+                        readers.remove(source)
+                        if destination is channel:
+                            channel.shutdown_write()
+                        else:
+                            peer.shutdown(socket.SHUT_WR)
+        except (OSError, EOFError, paramiko.SSHException):
+            pass
+        finally:
+            try:
+                channel.close()
+            except (OSError, EOFError, paramiko.SSHException):
+                pass
+
+
+def echo_server(listener):
+    def echo(peer):
+        with peer:
+            try:
+                while data := peer.recv(65536):
+                    peer.sendall(data)
+            except OSError:
+                pass
+    while True:
+        try:
+            peer, _ = listener.accept()
+            threading.Thread(target=echo, args=(peer,), daemon=True).start()
+        except OSError:
+            return
+
 
 def relay(channel, destination):
-    with socket.create_connection(destination) as peer:
-        try:
-            while True:
-                ready, _, _ = select.select([channel, peer], [], [], 1)
-                for source in ready:
-                    data = source.recv(32768)
-                    if not data:
-                        return
-                    (peer if source is channel else channel).sendall(data)
-        finally:
-            channel.close()
+    try:
+        relay_peer(channel, socket.create_connection(destination))
+    except OSError:
+        channel.close()
 
 
-def serve(sock, host_key, client_key, port):
+def serve(sock, host_key, client_key, port, echo_port):
     transport = paramiko.Transport(sock)
     transport.banner_timeout = 120
     transport.auth_timeout = 120
@@ -209,6 +283,8 @@ def serve(sock, host_key, client_key, port):
     transport.set_subsystem_handler("sftp", paramiko.SFTPServer, Files)
     server = Authentication(client_key)
     server.port = port
+    server.transport = transport
+    server.allowed_destinations = {("127.0.0.1", port), ("127.0.0.1", echo_port), ("::1", echo_port)}
     channels = []
     try:
         transport.start_server(server=server)
@@ -222,6 +298,8 @@ def serve(sock, host_key, client_key, port):
     except (EOFError, OSError, paramiko.SSHException):
         pass
     finally:
+        for listener in server.listeners.values():
+            listener.close()
         transport.close()
 
 
@@ -236,13 +314,18 @@ def main():
     client_key.write_private_key(private)
     encrypted_private = io.StringIO()
     client_key.write_private_key(encrypted_private, password="fixture-passphrase")
-    with socket.socket() as listener:
+    with socket.socket() as listener, socket.socket() as echo_listener:
+        echo_listener.bind(("127.0.0.1", 0))
+        echo_listener.listen(128)
+        echo_port = echo_listener.getsockname()[1]
+        threading.Thread(target=echo_server, args=(echo_listener,), daemon=True).start()
         listener.bind(("127.0.0.1", 0))
-        listener.listen()
+        listener.listen(128)
         port = listener.getsockname()[1]
         fingerprint = base64.b64encode(hashlib.sha256(host_key.asbytes()).digest()).decode().rstrip("=")
         metadata = {
             "port": port,
+            "echoPort": echo_port,
             "privateKey": private.getvalue(),
             "encryptedPrivateKey": encrypted_private.getvalue(),
             "hostFingerprint": "SHA256:" + fingerprint,
@@ -255,7 +338,7 @@ def main():
         try:
             while True:
                 sock, _ = listener.accept()
-                threading.Thread(target=serve, args=(sock, host_key, client_key, port), daemon=True).start()
+                threading.Thread(target=serve, args=(sock, host_key, client_key, port, echo_port), daemon=True).start()
         except KeyboardInterrupt:
             pass
         finally:

TFMTests/PortForwardingJourneyTests.swift

@@ -10,6 +10,14 @@ import struct TermindDomain.Host
 
 @Suite @MainActor
 struct PortForwardingJourneyTests {
+    @Test func aCountReceivedWhileStartingSurvivesStartCompletion() async throws {
+        let fixture = Fixture(mode: .countDuringStart)
+        defer { fixture.controller.stopAll() }
+        let rule = fixture.rules[0]
+        fixture.controller.start(rule.id)
+        try await wait { fixture.controller.state(for: rule.id) == .running(activeConnections: 3) }
+    }
+
     @Test func windowsShareForwardingAuthenticationAndDeletingItsJumpHostCancelsIt() async throws {
         let jump = Host(name: "Jump", host: "jump.invalid")
         let host = Host(name: "Server", host: "server.invalid", jumpHostID: jump.id)
@@ -312,7 +320,7 @@ struct PortForwardingJourneyTests {
 }
 
 private actor ForwardingTestSession: SSHForwardingSessionProtocol {
-    enum Mode: Sendable { case immediate, keyboard, trust, failure, listenerFailure }
+    enum Mode: Sendable { case immediate, keyboard, trust, failure, listenerFailure, countDuringStart }
     let mode: Mode
     let count: SSHForwardingSession.ConnectionHandler
     let failure: SSHForwardingSession.FailureHandler
@@ -356,6 +364,10 @@ private actor ForwardingTestSession: SSHForwardingSessionProtocol {
     }
     func start(_ rule: PortForwardingRule) async throws {
         if mode == .listenerFailure && rule.bindPort == 9000 { throw TermindSSHError.route("Port in use") }
+        if mode == .countDuringStart {
+            count(rule.id, 3)
+            try await Task.sleep(for: .milliseconds(20))
+        }
         starts.append(rule.id)
         listening.insert(rule.id)
     }

TFMTests/SSHForwardingPumpTests.swift

@@ -0,0 +1,304 @@
+import CSSH2
+import Darwin
+import Foundation
+import Testing
+
+@testable import TermindRemoteInfrastructure
+
+@Suite
+struct SSHForwardingPumpTests {
+    @Test func blockedWriteKeepsItsPointerLengthAndBytesWhileMoreDataArrives() {
+        let queue = SSHByteQueue()
+        queue.append(Data(repeating: 0x31, count: 100))
+        let original = queue.withFirstBytes { (UInt(bitPattern: $0.baseAddress!), $0.count, Data($0)) }
+        for _ in 0..<100 { queue.append(Data(repeating: 0x72, count: 32768)) }
+        for _ in 0..<5 {
+            queue.withFirstBytes {
+                #expect(UInt(bitPattern: $0.baseAddress!) == original.0)
+                #expect($0.count == original.1)
+                #expect(Data($0) == original.2)
+            }
+        }
+        queue.consume(37)
+        queue.withFirstBytes {
+            #expect(UInt(bitPattern: $0.baseAddress!) == original.0 + 37)
+            #expect(Data($0) == Data(repeating: 0x31, count: 63))
+        }
+        queue.consume(63)
+        #expect(queue.withFirstBytes { Data($0) } == Data(repeating: 0x72, count: 32768))
+    }
+
+    @Test func chunkCompactionPreservesOrderAndReleasesConsumedBytes() {
+        let queue = SSHByteQueue()
+        for value in UInt8(0)..<200 { queue.append(contentsOf: [value]) }
+        for value in UInt8(0)..<200 {
+            #expect(queue.withFirstBytes { Array($0) } == [value])
+            queue.consume(1)
+            #expect(queue.count == 199 - Int(value))
+        }
+        #expect(queue.isEmpty)
+        queue.append(contentsOf: [255])
+        #expect(queue.withFirstBytes { Array($0) } == [255])
+    }
+
+    @Test func readyQueueDeduplicatesAndPutsHotStreamsBehindPeers() {
+        var ready = SSHReadyQueue<Int>()
+        for value in 0..<100 {
+            ready.insert(value)
+            ready.insert(value)
+        }
+        for value in 0..<100 {
+            #expect(ready.pop() == value)
+            ready.insert(value)
+        }
+        for value in 0..<100 { #expect(ready.pop() == value) }
+        #expect(ready.pop() == nil)
+    }
+
+    @Test func cancellingSuspendedSourcesClosesOnlyAfterBothHandlersRetire() async throws {
+        let descriptors = try pair()
+        defer { Darwin.close(descriptors[1]) }
+        await withCheckedContinuation { continuation in
+            let queue = DispatchQueue(label: "forwarding-source-test")
+            queue.async {
+                let events = SSHSocketEvents(socket: descriptors[0], queue: queue) {
+                    Issue.record("A cancelled source delivered readiness")
+                }
+                events.cancel {
+                    #expect(fcntl(descriptors[0], F_GETFD) >= 0)
+                    Darwin.close(descriptors[0])
+                    continuation.resume()
+                }
+            }
+        }
+    }
+
+    @Test func targetConnectionCanBeCancelledDuringDNSWithoutLateDelivery() async {
+        let queue = DispatchQueue(label: "forwarding-connect-test")
+        await withCheckedContinuation { continuation in
+            queue.async {
+                let connector = SSHForwardingConnector(queue: queue) { result in
+                    if case .success(let socket) = result { Darwin.close(socket) }
+                    Issue.record("Cancelled target delivered a result")
+                }
+                connector.start(host: "localhost", port: 1)
+                connector.cancel()
+                queue.asyncAfter(deadline: .now() + 0.1) {
+                    _ = connector
+                    continuation.resume()
+                }
+            }
+        }
+    }
+
+    @Test func repeatedEAGAINRetainsTheOriginalWriteAndPreservesLaterData() throws {
+        let sockets = try pair()
+        defer { for socket in sockets { Darwin.close(socket) } }
+        _ = fcntl(sockets[0], F_SETFL, O_NONBLOCK)
+        var attempts: [(UInt, Data)] = []
+        var delivered = Data()
+        var io = idleIO()
+        io.write = { _, bytes in
+            attempts.append((UInt(bitPattern: bytes.baseAddress!), Data(bytes)))
+            if attempts.count <= 2 { return Int(LIBSSH2_ERROR_EAGAIN) }
+            let count = min(7, bytes.count)
+            delivered.append(contentsOf: bytes.prefix(count))
+            return count
+        }
+        let handle = OpaquePointer(bitPattern: 1)!
+        let stream = SSHStreamPump(socket: sockets[0], channel: handle, io: io)
+        var scratch = [UInt8](repeating: 0, count: 32768)
+        let first = Data(repeating: 31, count: 20)
+        let second = Data(repeating: 72, count: 13)
+        first.withUnsafeBytes { _ = Darwin.send(sockets[1], $0.baseAddress!, $0.count, 0) }
+        _ = try stream.pump(session: handle, scratch: &scratch)
+        second.withUnsafeBytes { _ = Darwin.send(sockets[1], $0.baseAddress!, $0.count, 0) }
+        _ = try stream.pump(session: handle, scratch: &scratch)
+        #expect(attempts.count == 2)
+        #expect(attempts[0].0 == attempts[1].0)
+        #expect(attempts[0].1 == first && attempts[1].1 == first)
+        _ = try stream.pump(session: handle, scratch: &scratch)
+        #expect(delivered == first + second)
+        #expect(stream.toChannel.isEmpty)
+    }
+
+    @Test func backpressureParksSocketReadsAndResumesAfterDraining() throws {
+        let sockets = try pair()
+        defer { for socket in sockets { Darwin.close(socket) } }
+        for socket in sockets { _ = fcntl(socket, F_SETFL, O_NONBLOCK) }
+        var blocked = true
+        var io = idleIO()
+        io.write = { _, bytes in blocked ? Int(LIBSSH2_ERROR_EAGAIN) : bytes.count }
+        let handle = OpaquePointer(bitPattern: 1)!
+        let stream = SSHStreamPump(socket: sockets[0], channel: handle, io: io)
+        var scratch = [UInt8](repeating: 0, count: 32768)
+        stream.toChannel.append(Data(repeating: 42, count: SSHStreamPump.highWater))
+        _ = try stream.pump(session: handle, scratch: &scratch)
+        #expect(stream.toChannel.count == SSHStreamPump.highWater)
+        #expect(!stream.wantsSocketRead)
+        blocked = false
+        for _ in 0..<10 { _ = try stream.pump(session: handle, scratch: &scratch) }
+        #expect(stream.toChannel.isEmpty)
+        #expect(stream.wantsSocketRead)
+    }
+
+    @Test func halfCloseWaitsForPendingWritesAndStillDeliversTheResponse() throws {
+        let sockets = try pair()
+        defer { for socket in sockets { Darwin.close(socket) } }
+        _ = fcntl(sockets[0], F_SETFL, O_NONBLOCK)
+        var blocked = true
+        var sentEOF = 0
+        var sentResponse = false
+        var io = idleIO()
+        io.write = { _, bytes in blocked ? Int(LIBSSH2_ERROR_EAGAIN) : bytes.count }
+        io.sendEOF = { _ in
+            sentEOF += 1
+            return 0
+        }
+        io.read = { _, bytes in
+            guard !blocked, !sentResponse else { return Int(LIBSSH2_ERROR_EAGAIN) }
+            bytes[0] = 99
+            sentResponse = true
+            return 1
+        }
+        io.eof = { _ in sentResponse ? 1 : 0 }
+        let handle = OpaquePointer(bitPattern: 1)!
+        let stream = SSHStreamPump(socket: sockets[0], channel: handle, io: io)
+        stream.toChannel.append(contentsOf: [42])
+        _ = Darwin.shutdown(sockets[1], SHUT_WR)
+        var scratch = [UInt8](repeating: 0, count: 32768)
+        _ = try stream.pump(session: handle, scratch: &scratch)
+        #expect(sentEOF == 0)
+        blocked = false
+        _ = try stream.pump(session: handle, scratch: &scratch)
+        #expect(sentEOF == 1)
+        #expect(stream.finished)
+        var byte: UInt8 = 0
+        #expect(Darwin.recv(sockets[1], &byte, 1, 0) == 1)
+        #expect(byte == 99)
+        #expect(Darwin.recv(sockets[1], &byte, 1, 0) == 0)
+    }
+
+    @Test func aWindowUpdateWithoutPayloadSchedulesTheBlockedWriterAgain() throws {
+        let sockets = try pair()
+        defer { for socket in sockets { Darwin.close(socket) } }
+        _ = fcntl(sockets[0], F_SETFL, O_NONBLOCK)
+        var window: UInt = 0
+        var io = idleIO()
+        io.write = { _, bytes in window == 0 ? Int(LIBSSH2_ERROR_EAGAIN) : bytes.count }
+        io.writeWindow = { _ in window }
+        io.read = { _, _ in
+            window = 32768
+            return Int(LIBSSH2_ERROR_EAGAIN)
+        }
+        let handle = OpaquePointer(bitPattern: 1)!
+        let stream = SSHStreamPump(socket: sockets[0], channel: handle, io: io)
+        stream.toChannel.append(contentsOf: [42])
+        var scratch = [UInt8](repeating: 0, count: 32768)
+        #expect(try stream.pump(session: handle, scratch: &scratch))
+        #expect(stream.toChannel.count == 1)
+        #expect(try stream.pump(session: handle, scratch: &scratch))
+        #expect(stream.toChannel.isEmpty)
+    }
+
+    @Test func encryptedPacketsSurviveSocketBackpressureWithoutLibrarySendRetries() throws {
+        let sockets = try pair()
+        defer { for socket in sockets { Darwin.close(socket) } }
+        for socket in sockets { _ = fcntl(socket, F_SETFL, O_NONBLOCK) }
+        var capacity: Int32 = 4096
+        setsockopt(sockets[0], SOL_SOCKET, SO_SNDBUF, &capacity, socklen_t(MemoryLayout<Int32>.size))
+        let session = try #require(libssh2_session_init_ex(nil, nil, nil, nil))
+        let transport = SSHBufferedTransport(session: session)
+        defer { transport.dispose(session: session) }
+        let callback = try #require(libssh2_session_callback_set2(session, LIBSSH2_CALLBACK_SEND, nil))
+        libssh2_session_callback_set2(session, LIBSSH2_CALLBACK_SEND, callback)
+        typealias Send =
+            @convention(c) (
+                Int32, UnsafeRawPointer?, Int, Int32, UnsafeMutablePointer<UnsafeMutableRawPointer?>?
+            ) -> Int
+        let send = unsafeBitCast(callback, to: Send.self)
+        var expected = Data()
+        for index in 0..<100 where !transport.isBackpressured {
+            let packet = Data(repeating: UInt8(index), count: 32768)
+            expected.append(packet)
+            let result = packet.withUnsafeBytes {
+                send(sockets[0], $0.baseAddress!, $0.count, 0, libssh2_session_abstract(session))
+            }
+            #expect(result == packet.count)
+        }
+        #expect(transport.isBackpressured)
+        #expect(expected.count <= SSHStreamPump.highWater + 65536)
+        var actual = Data()
+        var scratch = [UInt8](repeating: 0, count: 32768)
+        for _ in 0..<1000 {
+            let count = Darwin.recv(sockets[1], &scratch, scratch.count, 0)
+            if count > 0 { actual.append(contentsOf: scratch.prefix(count)) }
+            try transport.flush(socket: sockets[0])
+            if actual.count == expected.count { break }
+        }
+        #expect(actual == expected)
+        #expect(!transport.wantsWrite)
+    }
+
+    @Test func cancellingAZeroWindowWriteDoesNotWaitForThePeerToRead() throws {
+        let sockets = try pair()
+        defer { for socket in sockets { Darwin.close(socket) } }
+        _ = fcntl(sockets[0], F_SETFL, O_NONBLOCK)
+        var io = idleIO()
+        var writes = 0
+        io.write = { _, _ in
+            writes += 1
+            return Int(LIBSSH2_ERROR_EAGAIN)
+        }
+        io.writeWindow = { _ in 0 }
+        let handle = OpaquePointer(bitPattern: 1)!
+        let stream = SSHStreamPump(socket: sockets[0], channel: handle, io: io)
+        stream.toChannel.append(Data(repeating: 42, count: 32768))
+        var scratch = [UInt8](repeating: 0, count: 32768)
+        _ = try stream.pump(session: handle, scratch: &scratch)
+        #expect(stream.writePending)
+        #expect(stream.finishPendingWrite(session: handle))
+        #expect(!stream.writePending)
+        #expect(writes == 1)
+    }
+
+    @Test func continuousEncryptedInputYieldsAtTheSessionReadBudget() throws {
+        let sockets = try pair()
+        defer { for socket in sockets { Darwin.close(socket) } }
+        for socket in sockets { _ = fcntl(socket, F_SETFL, O_NONBLOCK) }
+        let session = try #require(libssh2_session_init_ex(nil, nil, nil, nil))
+        let transport = SSHBufferedTransport(session: session)
+        defer { transport.dispose(session: session) }
+        let callback = try #require(libssh2_session_callback_set2(session, LIBSSH2_CALLBACK_RECV, nil))
+        libssh2_session_callback_set2(session, LIBSSH2_CALLBACK_RECV, callback)
+        typealias Receive =
+            @convention(c) (
+                Int32, UnsafeMutableRawPointer?, Int, Int32, UnsafeMutablePointer<UnsafeMutableRawPointer?>?
+            ) -> Int
+        let receive = unsafeBitCast(callback, to: Receive.self)
+        var packet = [UInt8](repeating: 42, count: 4096)
+        for _ in 0..<(SSHStreamPump.quantum / packet.count) {
+            try #require(Darwin.send(sockets[1], &packet, packet.count, 0) == packet.count)
+            try #require(
+                receive(sockets[0], &packet, packet.count, 0, libssh2_session_abstract(session)) == packet.count)
+        }
+        #expect(transport.exhaustedReadBudget)
+        try #require(Darwin.send(sockets[1], &packet, packet.count, 0) == packet.count)
+        #expect(receive(sockets[0], &packet, packet.count, 0, libssh2_session_abstract(session)) == -Int(EAGAIN))
+        transport.beginPass()
+        #expect(receive(sockets[0], &packet, packet.count, 0, libssh2_session_abstract(session)) == packet.count)
+    }
+
+    private func idleIO() -> SSHChannelIO {
+        SSHChannelIO(
+            read: { _, _ in Int(LIBSSH2_ERROR_EAGAIN) }, write: { _, bytes in bytes.count },
+            writeWindow: { _ in 32768 }, eof: { _ in 0 }, sendEOF: { _ in 0 },
+            directions: { _ in LIBSSH2_SESSION_BLOCK_INBOUND })
+    }
+
+    private func pair() throws -> [Int32] {
+        var sockets: [Int32] = [-1, -1]
+        guard Darwin.socketpair(AF_UNIX, SOCK_STREAM, 0, &sockets) == 0 else { throw POSIXError(.EIO) }
+        return sockets
+    }
+}

TFMTests/SSHForwardingWireTests.swift

@@ -0,0 +1,405 @@
+import Darwin
+import Foundation
+import TermindApplication
+import TermindDomain
+import TermindRemoteInfrastructure
+import Testing
+
+/// Uses only the loopback fixture from Scripts/ssh-authentication-fixture.py.
+@Suite(.enabled(if: ProcessInfo.processInfo.environment["TERMIND_SSH_FIXTURE"] != nil), .serialized)
+nonisolated struct SSHForwardingWireTests {
+    @Test(arguments: [PortForwardingRule.Kind.local, .remote, .dynamic], [1, 16, 64])
+    func concurrentStreamsKeepTheirPayloadsAndHalfClose(_ kind: PortForwardingRule.Kind, _ concurrency: Int)
+        async throws
+    {
+        try await exercise(kind: kind, concurrency: concurrency, jumps: 0)
+    }
+
+    @Test(arguments: [PortForwardingRule.Kind.local, .remote, .dynamic], [1, 2])
+    func forwardingCrossesJumpBridges(_ kind: PortForwardingRule.Kind, _ jumps: Int) async throws {
+        try await exercise(kind: kind, concurrency: 16, jumps: jumps)
+    }
+
+    @Test func stoppingOneRulePreservesItsPeerAndReleasesItsListener() async throws {
+        let fixture = try fixture()
+        let session = SSHForwardingSession(
+            connectionHandler: { _, _ in }, failureHandler: { Issue.record(Comment(rawValue: $0)) })
+        try await session.connect(request(fixture, jumps: 0))
+        do {
+            let first = try rule(.local, fixture: fixture)
+            let second = try rule(.local, fixture: fixture)
+            try await session.start(first)
+            try await session.start(second)
+            await session.stop(first.id)
+            try await session.start(first)
+            try await exchange(port: first.bindPort, target: nil, seed: 1, rounds: 2)
+            await session.stop(first.id)
+            try await exchange(port: second.bindPort, target: nil, seed: 2, rounds: 2)
+        } catch {
+            await session.disconnect()
+            throw error
+        }
+        await session.disconnect()
+    }
+
+    @Test(
+        .enabled(if: ProcessInfo.processInfo.environment["TERMIND_FORWARDING_BENCHMARK"] == "1"),
+        arguments: [PortForwardingRule.Kind.local, .remote, .dynamic], [1, 16, 64])
+    func sustainedDuplexTransfer(_ kind: PortForwardingRule.Kind, _ concurrency: Int) async throws {
+        try await exercise(kind: kind, concurrency: concurrency, jumps: 0, rounds: 128, streaming: true)
+    }
+
+    @Test(
+        .enabled(if: ProcessInfo.processInfo.environment["TERMIND_FORWARDING_BENCHMARK"] == "1"), arguments: [0, 1, 2])
+    func smallRequestsProgressBesideBulkStreams(_ jumps: Int) async throws {
+        try await exercise(kind: .local, concurrency: 24, jumps: jumps, rounds: 128, mixed: true)
+    }
+
+    @Test func slowConsumerPreservesPayloadWhileItsPeerProgresses() async throws {
+        let fixture = try fixture()
+        let session = SSHForwardingSession(connectionHandler: { _, _ in }, failureHandler: { _ in })
+        try await session.connect(request(fixture, jumps: 2))
+        do {
+            let rule = try rule(.local, fixture: fixture)
+            try await session.start(rule)
+            async let slow = exchange(
+                port: rule.bindPort, target: nil, seed: 41, rounds: 256, streaming: true, delayedRead: true)
+            async let small = exchange(port: rule.bindPort, target: nil, seed: 42, rounds: 128, payloadSize: 64)
+            let (bulk, peer) = try await (slow, small)
+            try #require(bulk.bytes == 257 * 65536)
+            try #require(peer.bytes == 129 * 64)
+        } catch {
+            await session.disconnect()
+            throw error
+        }
+        await session.disconnect()
+    }
+
+    @Test func disconnectUnderBackpressureReleasesTheListenerAcrossRepeatedStarts() async throws {
+        let fixture = try fixture()
+        let session = SSHForwardingSession(connectionHandler: { _, _ in }, failureHandler: { _ in })
+        let rule = try rule(.local, fixture: fixture)
+        for _ in 0..<5 {
+            try await session.connect(request(fixture, jumps: 1))
+            try await session.start(rule)
+            let socket = try SSHSocketConnector.open(host: "127.0.0.1", port: rule.bindPort, timeout: 5)
+            let writer = Task {
+                await withCheckedContinuation { continuation in
+                    Thread.detachNewThread {
+                        let payload = Data(repeating: 42, count: 65536)
+                        for _ in 0..<1024 {
+                            do { try send(socket, payload) } catch { break }
+                        }
+                        continuation.resume()
+                    }
+                }
+            }
+            try await Task.sleep(for: .milliseconds(200))
+            let began = ContinuousClock.now
+            await session.disconnect()
+            try #require(began.duration(to: .now) < .seconds(2))
+            _ = Darwin.shutdown(socket, SHUT_RDWR)
+            await writer.value
+            Darwin.close(socket)
+        }
+        try await session.connect(request(fixture, jumps: 0))
+        try await session.start(rule)
+        try await exchange(port: rule.bindPort, target: nil, seed: 1, rounds: 2)
+        await session.disconnect()
+    }
+
+    @Test func stoppingAnInFlightRemoteListenDrainsItBeforeRebinding() async throws {
+        let fixture = try fixture()
+        let session = SSHForwardingSession(connectionHandler: { _, _ in }, failureHandler: { _ in })
+        try await session.connect(request(fixture, jumps: 0, username: "none-slow-forward"))
+        do {
+            let rule = try rule(.remote, fixture: fixture)
+            let starting = Task { try await session.start(rule) }
+            try await Task.sleep(for: .milliseconds(30))
+            await session.stop(rule.id)
+            do {
+                try await starting.value
+                Issue.record("Slow listener should have been cancelled while starting")
+            } catch is CancellationError {}
+            try await session.start(rule)
+            try await exchange(port: rule.bindPort, target: nil, seed: 1, rounds: 2)
+        } catch {
+            await session.disconnect()
+            throw error
+        }
+        await session.disconnect()
+    }
+
+    @Test func multipleRemoteListenersCanStopAndRestartIndependently() async throws {
+        let fixture = try fixture()
+        let session = SSHForwardingSession(connectionHandler: { _, _ in }, failureHandler: { _ in })
+        try await session.connect(request(fixture, jumps: 1))
+        do {
+            let first = try rule(.remote, fixture: fixture)
+            let second = try rule(.remote, fixture: fixture)
+            async let startFirst: Void = session.start(first)
+            async let startSecond: Void = session.start(second)
+            _ = try await (startFirst, startSecond)
+            try await exchange(port: first.bindPort, target: nil, seed: 1, rounds: 2)
+            await session.stop(first.id)
+            try await exchange(port: second.bindPort, target: nil, seed: 2, rounds: 2)
+            try await session.start(first)
+            try await exchange(port: first.bindPort, target: nil, seed: 3, rounds: 2)
+        } catch {
+            await session.disconnect()
+            throw error
+        }
+        await session.disconnect()
+    }
+
+    @Test(.enabled(if: ProcessInfo.processInfo.environment["TERMIND_FORWARDING_BENCHMARK"] == "1"))
+    func idleForwardingCPU() async throws {
+        let fixture = try fixture()
+        let session = SSHForwardingSession(connectionHandler: { _, _ in }, failureHandler: { _ in })
+        try await session.connect(request(fixture, jumps: 2))
+        let rule = try rule(.local, fixture: fixture)
+        try await session.start(rule)
+        let socket = try SSHSocketConnector.open(host: "127.0.0.1", port: rule.bindPort, timeout: 5)
+        try send(socket, Data([42]))
+        try #require(try receive(socket, count: 1) == Data([42]))
+        try await Task.sleep(for: .milliseconds(300))
+        let before = usage()
+        let began = ContinuousClock.now
+        try await Task.sleep(for: .seconds(5))
+        let elapsed = seconds(began.duration(to: .now))
+        let after = usage()
+        print("FORWARDING_IDLE cpu_s=\(after.cpu - before.cpu) elapsed_s=\(elapsed) jumps=2")
+        Darwin.close(socket)
+        await session.disconnect()
+    }
+
+    private func exercise(
+        kind: PortForwardingRule.Kind, concurrency: Int, jumps: Int,
+        rounds: Int = 8, streaming: Bool = false, mixed: Bool = false
+    ) async throws {
+        let fixture = try fixture()
+        let session = SSHForwardingSession(
+            connectionHandler: { _, _ in }, failureHandler: { Issue.record(Comment(rawValue: $0)) })
+        try await session.connect(request(fixture, jumps: jumps))
+        do {
+            let rule = try rule(kind, fixture: fixture)
+            try await session.start(rule)
+            let start = ContinuousClock.now
+            let usageBefore = usage()
+            let results = try await withThrowingTaskGroup(of: Measurement.self) { group in
+                for index in 0..<concurrency {
+                    group.addTask {
+                        try await exchange(
+                            port: rule.bindPort, target: kind == .dynamic ? fixture.echoPort : nil,
+                            seed: UInt8(index), rounds: rounds,
+                            payloadSize: mixed && index >= 8 ? 64 : 65536,
+                            streaming: streaming || (mixed && index < 8))
+                    }
+                }
+                var results: [Measurement] = []
+                for try await result in group { results.append(result) }
+                return results
+            }
+            let elapsed = seconds(start.duration(to: .now))
+            let usageAfter = usage()
+            let payloadBytes = results.reduce(0) { $0 + $1.bytes }
+            let record: [String: Any] = [
+                "kind": kind.rawValue, "streams": concurrency, "jumps": jumps,
+                "workload": mixed ? "mixed" : streaming ? "duplex" : "request_response",
+                "successful_streams": results.count, "payload_bytes_per_direction": payloadBytes,
+                "elapsed_s": elapsed, "payload_mib_s_per_direction": Double(payloadBytes) / 1_048_576 / elapsed,
+                "ready_p95_ms": percentile(results.map(\.connect)) * 1000,
+                "request_p95_ms": percentile(results.filter { !$0.streaming }.flatMap(\.latencies)) * 1000,
+                "small_request_p95_ms": percentile(results.filter { $0.small }.flatMap(\.latencies)) * 1000,
+                "cpu_s": usageAfter.cpu - usageBefore.cpu,
+                "process_peak_rss_mib": Double(usageAfter.rss) / 1_048_576,
+            ]
+            print(
+                "FORWARDING_METRIC "
+                    + String(
+                        decoding: try JSONSerialization.data(
+                            withJSONObject: record, options: [.sortedKeys]), as: UTF8.self))
+            await session.stop(rule.id)
+        } catch {
+            await session.disconnect()
+            throw error
+        }
+        await session.disconnect()
+    }
+
+    private struct Fixture: Decodable {
+        let port: Int
+        let echoPort: Int
+    }
+
+    private func fixture() throws -> Fixture {
+        let path = try #require(ProcessInfo.processInfo.environment["TERMIND_SSH_FIXTURE"])
+        return try JSONDecoder().decode(Fixture.self, from: Data(contentsOf: URL(fileURLWithPath: path)))
+    }
+
+    private func request(_ fixture: Fixture, jumps: Int, username: String = "none") -> SSHConnectionRequest {
+        var configuration = SSHConnectionConfiguration(
+            hostID: UUID(), host: "127.0.0.1", port: fixture.port, username: username, credentials: [],
+            expectedHostKeys: [])
+        configuration.jumps = (0..<jumps).map { _ in
+            SSHRouteHopConfiguration(
+                id: UUID(), name: "Fixture", host: "127.0.0.1", port: fixture.port, username: "none",
+                credentials: [], expectedHostKeys: [])
+        }
+        return SSHConnectionRequest(configuration: configuration, authorizeHostKey: { _ in .accept })
+    }
+
+    private func rule(_ kind: PortForwardingRule.Kind, fixture: Fixture) throws -> PortForwardingRule {
+        let descriptor = Darwin.socket(AF_INET, SOCK_STREAM, 0)
+        guard descriptor >= 0 else { throw POSIXError(.EIO) }
+        defer { Darwin.close(descriptor) }
+        var address = sockaddr_in()
+        address.sin_len = UInt8(MemoryLayout<sockaddr_in>.size)
+        address.sin_family = sa_family_t(AF_INET)
+        address.sin_addr.s_addr = inet_addr("127.0.0.1")
+        let result = withUnsafePointer(to: &address) {
+            $0.withMemoryRebound(to: sockaddr.self, capacity: 1) {
+                Darwin.bind(descriptor, $0, socklen_t(MemoryLayout<sockaddr_in>.size))
+            }
+        }
+        guard result == 0 else { throw POSIXError(.EIO) }
+        var length = socklen_t(MemoryLayout<sockaddr_in>.size)
+        let bound = withUnsafeMutablePointer(to: &address) {
+            $0.withMemoryRebound(to: sockaddr.self, capacity: 1) { getsockname(descriptor, $0, &length) }
+        }
+        guard bound == 0 else { throw POSIXError(.EIO) }
+        return PortForwardingRule(
+            hostID: UUID(), kind: kind, bindHost: "127.0.0.1", bindPort: Int(UInt16(bigEndian: address.sin_port)),
+            targetHost: "127.0.0.1", targetPort: fixture.echoPort)
+    }
+
+    private struct Measurement: Sendable {
+        let connect: Double
+        let latencies: [Double]
+        let bytes: Int
+        let streaming: Bool
+        let small: Bool
+    }
+
+    private func seconds(_ duration: Duration) -> Double {
+        Double(duration.components.seconds) + Double(duration.components.attoseconds) / 1e18
+    }
+
+    private func percentile(_ values: [Double]) -> Double {
+        guard !values.isEmpty else { return 0 }
+        let sorted = values.sorted()
+        return sorted[min(sorted.count - 1, Int(ceil(Double(sorted.count) * 0.95)) - 1)]
+    }
+
+    private func usage() -> (cpu: Double, rss: Int) {
+        var value = rusage()
+        getrusage(RUSAGE_SELF, &value)
+        let cpu =
+            Double(value.ru_utime.tv_sec + value.ru_stime.tv_sec)
+            + Double(value.ru_utime.tv_usec + value.ru_stime.tv_usec) / 1e6
+        return (cpu, value.ru_maxrss)
+    }
+
+    @discardableResult private func exchange(
+        port: Int, target: Int?, seed: UInt8, rounds: Int, payloadSize: Int = 65536, streaming: Bool = false,
+        delayedRead: Bool = false
+    ) async throws -> Measurement {
+        // Dedicated test-client threads cannot exhaust the dispatch worker pool
+        // needed by the transport and its bounded DNS resolver.
+        try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<Measurement, any Error>) in
+            Thread.detachNewThread {
+                do {
+                    let began = ContinuousClock.now
+                    let socket = try SSHSocketConnector.open(host: "127.0.0.1", port: port, timeout: 15)
+                    defer { Darwin.close(socket) }
+                    if let target {
+                        // A fragmented greeting followed by CONNECT + payload in
+                        // one write exercises both parser boundaries.
+                        try send(socket, Data([5]))
+                        try send(socket, Data([1, 0]))
+                        try #require(try receive(socket, count: 2) == Data([5, 0]))
+                        let request = Data([5, 1, 0, 1, 127, 0, 0, 1, UInt8(target >> 8), UInt8(target & 255)])
+                        try send(socket, request + Data([seed]))
+                        try #require(try receive(socket, count: 10) == Data([5, 0, 0, 1, 0, 0, 0, 0, 0, 0]))
+                        try #require(try receive(socket, count: 1) == Data([seed]))
+                    }
+                    // Include channel negotiation and target connect for every
+                    // forwarding kind, rather than just the local TCP accept.
+                    try send(socket, Data([seed]))
+                    try #require(try receive(socket, count: 1) == Data([seed]))
+                    let connected = seconds(began.duration(to: .now))
+                    let payload = Data((0..<payloadSize).map { UInt8(truncatingIfNeeded: $0) &+ seed })
+                    var latencies: [Double] = []
+                    if streaming {
+                        // Independent producer and consumer exercise sustained
+                        // full-duplex forwarding beyond both SSH/socket windows.
+                        let producer = DispatchGroup()
+                        producer.enter()
+                        Thread.detachNewThread {
+                            defer { producer.leave() }
+                            do {
+                                for _ in 0...rounds { try send(socket, payload) }
+                            } catch { Issue.record(error) }
+                            _ = Darwin.shutdown(socket, SHUT_WR)
+                        }
+                        defer {
+                            _ = Darwin.shutdown(socket, SHUT_RDWR)
+                            producer.wait()
+                        }
+                        if delayedRead { Thread.sleep(forTimeInterval: 0.3) }
+                        for _ in 0...rounds {
+                            try #require(try receive(socket, count: payload.count) == payload)
+                        }
+                        var byte: UInt8 = 0
+                        try #require(Darwin.recv(socket, &byte, 1, 0) == 0)
+                    } else {
+                        for _ in 0..<rounds {
+                            let sent = ContinuousClock.now
+                            try send(socket, payload)
+                            try #require(try receive(socket, count: payload.count) == payload)
+                            latencies.append(seconds(sent.duration(to: .now)))
+                        }
+                        // The response is still pending when the client sends FIN.
+                        try send(socket, payload)
+                        _ = Darwin.shutdown(socket, SHUT_WR)
+                        try #require(try receive(socket, count: payload.count) == payload)
+                    }
+                    if !streaming {
+                        var byte: UInt8 = 0
+                        try #require(Darwin.recv(socket, &byte, 1, 0) == 0)
+                    }
+                    continuation.resume(
+                        returning: Measurement(
+                            connect: connected, latencies: latencies, bytes: (rounds + 1) * payloadSize,
+                            streaming: streaming, small: payloadSize == 64))
+                } catch { continuation.resume(throwing: error) }
+            }
+        }
+    }
+
+    private func send(_ socket: Int32, _ data: Data) throws {
+        try data.withUnsafeBytes { bytes in
+            var offset = 0
+            while offset < bytes.count {
+                let count = Darwin.send(socket, bytes.baseAddress! + offset, bytes.count - offset, 0)
+                guard count > 0 else { throw POSIXError(POSIXErrorCode(rawValue: errno) ?? .EIO) }
+                offset += count
+            }
+        }
+    }
+
+    private func receive(_ socket: Int32, count: Int) throws -> Data {
+        var data = Data(count: count)
+        try data.withUnsafeMutableBytes { bytes in
+            var offset = 0
+            while offset < count {
+                let read = Darwin.recv(socket, bytes.baseAddress! + offset, count - offset, 0)
+                guard read > 0 else {
+                    throw POSIXError(POSIXErrorCode(rawValue: errno) ?? .EIO)
+                }
+                offset += read
+            }
+        }
+        return data
+    }
+}