From 7ff154bdab30cb72947b4de6ec81b638f9f42ef4 Mon Sep 17 00:00:00 2001 From: Brett Henderson Date: Tue, 9 Jun 2026 16:18:42 -0700 Subject: [PATCH 1/2] fix network audit findings: transfer wedges, cancellation, connection races - downloads: wedge-forever paths (inbound-F catch, missing stall timeout), watchdog token collisions, duplicate queueing, retry classifier misclassifying our own teardown errors - cancellation: cancelDownload/cancelUpload APIs wired from the UI; streaming loops check per chunk, completion can't stomp .cancelled - peer connections: double-resume crash on bind retry (generation- stamped attempts), out-of-order file bytes after stopReceiving, event-stream/Task leak per failed dial, per-IP slot accounting, pool key collisions, obfuscated->raw handoff residue - coordination: distributed parent death detection, pending-browse leaks, login timeout races, NAT external-port re-advertisement - protocol: decompression truncation, compress buffer headroom, zero-byte files, deleted broken fileSearchRoom (code 25) - uploads: duplicate QueueUpload dedup, PF short-read guard, slot double-count, speed limit setting wired, legacy direction=download TransferRequests answered, atomic share rescan --- .../Connections/FileConnectionAttempt.swift | 42 +- .../Network/Connections/PeerConnection.swift | 305 +++++++--- .../Connections/PeerConnectionPool.swift | 106 +++- .../Connections/ServerConnection.swift | 109 +++- .../Handlers/ServerMessageHandler.swift | 43 +- .../SeeleseekCore/Network/NetworkClient.swift | 240 ++++++-- .../Network/Protocol/Decompression.swift | 89 ++- .../Network/Protocol/MessageBuilder.swift | 79 +-- .../Network/Protocol/MessageParser.swift | 28 +- .../Network/Protocol/ObfuscationCodec.swift | 4 +- .../Network/Services/ListenerService.swift | 23 +- .../Network/Services/NATService.swift | 29 +- .../Storage/DownloadManager.swift | 531 +++++++++++++----- .../SeeleseekCore/Storage/ShareManager.swift | 76 ++- .../SeeleseekCore/Storage/UploadManager.swift | 434 +++++++++++--- seeleseek/App/AppState.swift | 23 +- .../Features/Settings/SettingsState.swift | 5 + .../Features/Transfers/TransferState.swift | 11 + .../DownloadRetryClassifierTests.swift | 16 +- seeleseekTests/DownloadRetryFlowTests.swift | 56 +- seeleseekTests/FailureTests.swift | 14 +- seeleseekTests/PeerConnectivityFixTests.swift | 9 +- .../ServerMessageRoundTripTests.swift | 13 - 23 files changed, 1741 insertions(+), 544 deletions(-) diff --git a/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Connections/FileConnectionAttempt.swift b/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Connections/FileConnectionAttempt.swift index c71b7a3..97846ff 100644 --- a/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Connections/FileConnectionAttempt.swift +++ b/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Connections/FileConnectionAttempt.swift @@ -18,33 +18,36 @@ extension UploadManager { let connection = NWConnection(to: endpoint, using: params) let hasResumed = Mutex(false) + let timeoutTask = Mutex?>(nil) + // First claimant wins; everyone else's events are ignored. + let claimResume: @Sendable () -> Bool = { + hasResumed.withLock { old in + guard !old else { return false } + old = true + return true + } + } return await withCheckedContinuation { continuation in connection.stateUpdateHandler = { state in switch state { case .ready: - guard hasResumed.withLock({ old in - guard !old else { return false } - old = true - return true - }) else { return } + guard claimResume() else { return } + timeoutTask.withLock { $0?.cancel() } continuation.resume(returning: .ready(connection)) case .failed(let error): - guard hasResumed.withLock({ old in - guard !old else { return false } - old = true - return true - }) else { return } + guard claimResume() else { return } + timeoutTask.withLock { $0?.cancel() } + // No caller uses the socket from failure outcomes; + // cancel it here so it doesn't leak. + connection.cancel() if PeerConnection.isBindFailure(error) { continuation.resume(returning: .bindFailed(connection)) } else { continuation.resume(returning: .failed(connection)) } case .cancelled: - guard hasResumed.withLock({ old in - guard !old else { return false } - old = true - return true - }) else { return } + guard claimResume() else { return } + timeoutTask.withLock { $0?.cancel() } continuation.resume(returning: .failed(connection)) default: break @@ -52,16 +55,13 @@ extension UploadManager { } connection.start(queue: .global(qos: .userInitiated)) - Task { + let task = Task { try? await Task.sleep(for: timeout) - guard hasResumed.withLock({ old in - guard !old else { return false } - old = true - return true - }) else { return } + guard claimResume() else { return } connection.cancel() continuation.resume(returning: .failed(connection)) } + timeoutTask.withLock { $0 = task } } } } diff --git a/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Connections/PeerConnection.swift b/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Connections/PeerConnection.swift index 093f40d..21946ac 100644 --- a/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Connections/PeerConnection.swift +++ b/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Connections/PeerConnection.swift @@ -128,7 +128,15 @@ public actor PeerConnection { private(set) var messagesReceived: UInt32 = 0 private(set) var messagesSent: UInt32 = 0 private(set) var connectedAt: Date? - private(set) var lastActivityAt: Date? + // Mutex-backed so the pool's MainActor cleanup timer can read it + // synchronously while raw file bytes are flowing. + private nonisolated let _lastActivityAt = Mutex(nil) + public nonisolated var lastActivityAt: Date? { + _lastActivityAt.withLock { $0 } + } + private nonisolated func touchLastActivity() { + _lastActivityAt.withLock { $0 = Date() } + } // MARK: - Initialization @@ -195,8 +203,23 @@ public actor PeerConnection { // Track if connect continuation has been resumed to prevent double-resume private var connectContinuationResumed = false + // Each dial/accept attempt gets a generation; state events from + // abandoned attempts (e.g. bind-retry's first socket) are dropped. + private var connectAttemptGeneration: UInt64 = 0 + // Generation that reached .ready; gates event-stream finish on teardown. + private var connectionEstablishedGeneration: UInt64? public func connect() async throws { + do { + try await dialWithBindRetry() + } catch { + // No usable connection: end the event stream so consumers exit. + eventContinuation.finish() + throw error + } + } + + private func dialWithBindRetry() async throws { guard case .disconnected = state else { return } // Validate port range (must be valid UInt16 and non-zero) @@ -230,6 +253,8 @@ public actor PeerConnection { private func performConnect(to endpoint: NWEndpoint, bindTo localPort: UInt16?) async throws { updateState(.connecting) connectContinuationResumed = false + connectAttemptGeneration += 1 + let generation = connectAttemptGeneration let params = Self.makeOutboundParameters(bindTo: localPort, remoteEndpoint: endpoint) let conn = NWConnection(to: endpoint, using: params) @@ -241,7 +266,7 @@ public actor PeerConnection { conn.stateUpdateHandler = { [weak self] newState in guard let self else { return } Task { - await self.handleConnectionState(newState, continuation: continuation) + await self.handleConnectionState(newState, generation: generation, continuation: continuation) } } conn.start(queue: .global(qos: .userInitiated)) @@ -288,6 +313,9 @@ public actor PeerConnection { guard let connection, isIncoming else { return } updateState(.connecting) + connectContinuationResumed = false + connectAttemptGeneration += 1 + let generation = connectAttemptGeneration return try await withCheckedThrowingContinuation { continuation in connection.stateUpdateHandler = { [weak self] newState in @@ -297,7 +325,7 @@ public actor PeerConnection { if case .ready = newState { await self.extractRemoteEndpointIfNeeded() } - await self.handleConnectionState(newState, continuation: continuation) + await self.handleConnectionState(newState, generation: generation, continuation: continuation) } } @@ -585,16 +613,37 @@ public actor PeerConnection { throw PeerError.notConnected } + // Serve from the file transfer buffer first (bytes the stopped + // message loop already pulled off the socket). + if fileTransferBuffer.count >= exactLength { + let data = fileTransferBuffer.prefix(exactLength) + fileTransferBuffer.removeFirst(exactLength) + return Data(data) + } + + // Let the message loop's in-flight receive land, then re-check. + await waitForMessageLoopReceiveToLand() + if fileTransferBuffer.count >= exactLength { + let data = fileTransferBuffer.prefix(exactLength) + fileTransferBuffer.removeFirst(exactLength) + return Data(data) + } + + let bufferedData = fileTransferBuffer + fileTransferBuffer.removeAll() + let neededFromNetwork = exactLength - bufferedData.count + return try await withCheckedThrowingContinuation { continuation in - connection.receive(minimumIncompleteLength: exactLength, maximumLength: exactLength) { [weak self] data, _, _, error in + connection.receive(minimumIncompleteLength: neededFromNetwork, maximumLength: neededFromNetwork) { [weak self] data, _, _, error in if let error { continuation.resume(throwing: error) - } else if let data { + } else if let data, data.count == neededFromNetwork { Task { await self?.recordReceived(data.count) } - continuation.resume(returning: data) + continuation.resume(returning: bufferedData + data) } else { + // Short delivery (connection closed mid-read) continuation.resume(throwing: PeerError.connectionClosed) } } @@ -640,14 +689,6 @@ public actor PeerConnection { return Data(data) } - // If we have some buffered data but not enough, we need to receive more - let neededFromNetwork = count - fileTransferBuffer.count - logger.debug("[\(self.peerInfo.username)] Waiting for \(neededFromNetwork) raw bytes from network (have \(self.fileTransferBuffer.count) buffered, need \(count) total, timeout: \(timeout)s)...") - - // Capture and clear buffer before entering non-isolated closure - let bufferedData = fileTransferBuffer - fileTransferBuffer.removeAll() - // Cancel the underlying NWConnection on timeout. Without this, the // timeout child throws but the receive continuation stays pending — // `withThrowingTaskGroup` then waits on the orphan child forever, so @@ -655,30 +696,11 @@ public actor PeerConnection { return try await withThrowingTaskGroup(of: Data.self) { group in group.addTask { [self] in try await withTaskCancellationHandler { - try await withCheckedThrowingContinuation { continuation in - connection.receive(minimumIncompleteLength: neededFromNetwork, maximumLength: neededFromNetwork) { [weak self] data, _, _, error in - if let error { - self?.logger.debug("[\(self?.peerInfo.username ?? "??")] receiveRawBytes error: \(error)") - continuation.resume(throwing: error) - } else if let data, data.count >= neededFromNetwork { - self?.logger.debug("[\(self?.peerInfo.username ?? "??")] Received \(data.count) raw bytes from network") - Task { - await self?.recordReceived(data.count) - } - // Combine buffered data with newly received data - if !bufferedData.isEmpty { - var combined = bufferedData - combined.append(data) - continuation.resume(returning: Data(combined.prefix(count))) - } else { - continuation.resume(returning: data) - } - } else { - self?.logger.debug("[\(self?.peerInfo.username ?? "??")] Received incomplete data: \(data?.count ?? 0)/\(neededFromNetwork)") - continuation.resume(throwing: PeerError.connectionClosed) - } - } - } + // Let the message loop's in-flight receive land first so + // we never have two competing receives on one socket; + // its bytes go to fileTransferBuffer and are re-checked. + await waitForMessageLoopReceiveToLand() + return try await receiveRawBytesFromSocket(count: count, timeout: timeout, connection: connection) } onCancel: { // Force the pending receive to fire its callback (with an // error) so the continuation resolves and this child task @@ -700,6 +722,50 @@ public actor PeerConnection { } } + /// Re-checks the buffer after the message loop quiesces, then issues the + /// actual socket receive for whatever is still missing. + private func receiveRawBytesFromSocket(count: Int, timeout: TimeInterval, connection: NWConnection) async throws -> Data { + if fileTransferBuffer.count >= count { + let data = fileTransferBuffer.prefix(count) + fileTransferBuffer.removeFirst(count) + logger.debug("[\(self.peerInfo.username)] Got \(count) raw bytes from file transfer buffer (post-quiesce)") + return Data(data) + } + + // If we have some buffered data but not enough, we need to receive more + let neededFromNetwork = count - fileTransferBuffer.count + logger.debug("[\(self.peerInfo.username)] Waiting for \(neededFromNetwork) raw bytes from network (have \(self.fileTransferBuffer.count) buffered, need \(count) total, timeout: \(timeout)s)...") + + // Capture and clear buffer before entering non-isolated closure + let bufferedData = fileTransferBuffer + fileTransferBuffer.removeAll() + + return try await withCheckedThrowingContinuation { continuation in + connection.receive(minimumIncompleteLength: neededFromNetwork, maximumLength: neededFromNetwork) { [weak self] data, _, _, error in + if let error { + self?.logger.debug("[\(self?.peerInfo.username ?? "??")] receiveRawBytes error: \(error)") + continuation.resume(throwing: error) + } else if let data, data.count >= neededFromNetwork { + self?.logger.debug("[\(self?.peerInfo.username ?? "??")] Received \(data.count) raw bytes from network") + Task { + await self?.recordReceived(data.count) + } + // Combine buffered data with newly received data + if !bufferedData.isEmpty { + var combined = bufferedData + combined.append(data) + continuation.resume(returning: Data(combined.prefix(count))) + } else { + continuation.resume(returning: data) + } + } else { + self?.logger.debug("[\(self?.peerInfo.username ?? "??")] Received incomplete data: \(data?.count ?? 0)/\(neededFromNetwork)") + continuation.resume(throwing: PeerError.connectionClosed) + } + } + } + } + /// Result type for file chunk reception - distinguishes between data, completion, and errors public enum FileChunkResult: Sendable { case data(Data) @@ -715,19 +781,19 @@ public actor PeerConnection { } // First, check if we have buffered data from when the receive loop was stopped - if !fileTransferBuffer.isEmpty { - let chunk: Data - if fileTransferBuffer.count <= maxLength { - chunk = fileTransferBuffer - fileTransferBuffer.removeAll() - } else { - chunk = fileTransferBuffer.prefix(maxLength) - fileTransferBuffer.removeFirst(maxLength) - } + if let chunk = takeFileChunkFromBuffer(maxLength: maxLength) { logger.debug("Using \(chunk.count) bytes from file transfer buffer") return .data(chunk) } + // Let the message loop's in-flight receive land, then re-check the + // buffer: its bytes must be consumed in order, before fresh receives. + await waitForMessageLoopReceiveToLand() + if let chunk = takeFileChunkFromBuffer(maxLength: maxLength) { + logger.debug("Using \(chunk.count) bytes from file transfer buffer (post-quiesce)") + return .data(chunk) + } + return try await withCheckedThrowingContinuation { continuation in // `minimumIncompleteLength: 1` blocks until at least one byte is // available (or the connection closes), so we never spin returning @@ -771,17 +837,69 @@ public actor PeerConnection { } } + /// Dequeue up to `maxLength` bytes from the file transfer buffer. + private func takeFileChunkFromBuffer(maxLength: Int) -> Data? { + guard !fileTransferBuffer.isEmpty else { return nil } + let chunk: Data + if fileTransferBuffer.count <= maxLength { + chunk = fileTransferBuffer + fileTransferBuffer.removeAll() + } else { + chunk = Data(fileTransferBuffer.prefix(maxLength)) + fileTransferBuffer.removeFirst(maxLength) + } + return chunk + } + // Flag to stop the receive loop for raw file transfers private var shouldStopReceiving = false // Buffer for file transfer data received after stopping message parsing private var fileTransferBuffer = Data() + // True while the message loop has an NWConnection.receive in flight. + // Raw readers must wait for it to land before issuing their own receive, + // or NWConnection delivers bytes to the loop first (out-of-order data). + private var messageLoopReceiveArmed = false + private var messageLoopIdleWaiters: [UUID: CheckedContinuation] = [:] + + /// Called on every non-rearm exit of the message loop's receive callback. + private func markMessageLoopReceiveLanded() { + messageLoopReceiveArmed = false + let waiters = messageLoopIdleWaiters + messageLoopIdleWaiters.removeAll() + for waiter in waiters.values { + waiter.resume() + } + } + + /// Suspends until the message loop's in-flight receive has landed. + private func waitForMessageLoopReceiveToLand() async { + guard messageLoopReceiveArmed else { return } + let id = UUID() + await withTaskCancellationHandler { + await withCheckedContinuation { (continuation: CheckedContinuation) in + if messageLoopReceiveArmed { + messageLoopIdleWaiters[id] = continuation + } else { + continuation.resume() + } + } + } onCancel: { + Task { await self.resumeIdleWaiter(id) } + } + } + + private func resumeIdleWaiter(_ id: UUID) { + messageLoopIdleWaiters.removeValue(forKey: id)?.resume() + } + /// Stop the normal receive loop so we can do raw file transfers public func stopReceiving() { shouldStopReceiving = true // Clear the message receive buffer - any pending data will go to file transfer buffer receiveBuffer.removeAll() + migrateObfuscatedTailToFileBuffer() logger.info("Stopping receive loop for file transfer") logger.debug("[\(self.peerInfo.username)] Stopped receive loop, cleared message buffer") } @@ -808,19 +926,25 @@ public actor PeerConnection { do { return try await withThrowingTaskGroup(of: Data.self) { group in group.addTask { - try await withCheckedThrowingContinuation { continuation in - // Use minimumIncompleteLength: 0 to return immediately with whatever is available - connection.receive(minimumIncompleteLength: 0, maximumLength: maxLength) { [weak self] data, _, isComplete, error in - if let error { - continuation.resume(throwing: error) - } else if let data, !data.isEmpty { - Task { await self?.recordReceived(data.count) } - continuation.resume(returning: data) - } else { - // No data available - continuation.resume(returning: Data()) + try await withTaskCancellationHandler { + try await withCheckedThrowingContinuation { continuation in + // Use minimumIncompleteLength: 0 to return immediately with whatever is available + connection.receive(minimumIncompleteLength: 0, maximumLength: maxLength) { [weak self] data, _, isComplete, error in + if let error { + continuation.resume(throwing: error) + } else if let data, !data.isEmpty { + Task { await self?.recordReceived(data.count) } + continuation.resume(returning: data) + } else { + // No data available + continuation.resume(returning: Data()) + } } } + } onCancel: { + // Resolve a pending receive so the timeout can return. + // Safe: only called after the transfer completed. + connection.cancel() } } @@ -842,12 +966,20 @@ public actor PeerConnection { // MARK: - Private Methods - private func handleConnectionState(_ state: NWConnection.State, continuation: CheckedContinuation?) { + private func handleConnectionState(_ state: NWConnection.State, generation: UInt64, continuation: CheckedContinuation?) { + // Drop events from abandoned attempts: a stale socket's late + // .cancelled/.failed must not resume the live attempt's continuation, + // reset the shared flag, cancel the live connection, or stomp state. + guard generation == connectAttemptGeneration else { + logger.debug("Ignoring stale connection state (gen \(generation) < \(self.connectAttemptGeneration)): \(String(describing: state))") + return + } switch state { case .ready: logger.info("Peer connected: \(self.peerInfo.username) at \(self.peerInfo.ip):\(self.peerInfo.port)") logger.info("Connected to peer \(self.peerInfo.username) at \(self.peerInfo.ip):\(self.peerInfo.port)") connectedAt = Date() + connectionEstablishedGeneration = generation updateState(.connected) // Only auto-start receiving if flag is set (for outgoing connections) // For incoming connections, we delay until callbacks are configured @@ -876,6 +1008,12 @@ public actor PeerConnection { connectContinuationResumed = true continuation?.resume(throwing: error) } + // Established connection died: end the event stream (after the + // .stateChanged yield above) so the pool's consume loop exits. + // Gated so a stale bind-retry socket can't finish the live stream. + if connectionEstablishedGeneration == generation { + eventContinuation.finish() + } case .waiting(let error): logger.debug("Peer connection waiting: \(self.peerInfo.username) at \(self.peerInfo.ip):\(self.peerInfo.port)") @@ -909,6 +1047,9 @@ public actor PeerConnection { connectContinuationResumed = true continuation?.resume(throwing: CancellationError()) } + if connectionEstablishedGeneration == generation { + eventContinuation.finish() + } case .setup: break @@ -930,12 +1071,20 @@ public actor PeerConnection { logger.debug("[\(self.peerInfo.username)] Resuming receive loop for P connection (browse)") shouldStopReceiving = false - // Move any data from file transfer buffer back to receive buffer + // Move buffered data back: decoded prefix to receiveBuffer, any + // undecoded obfuscated tail back to obfuscatedBuffer for the codec. if !fileTransferBuffer.isEmpty { - receiveBuffer.append(fileTransferBuffer) + let boundary = isObfuscated + ? min(fileBufferDecodedPrefixLength, fileTransferBuffer.count) + : fileTransferBuffer.count + receiveBuffer.append(fileTransferBuffer.prefix(boundary)) + if boundary < fileTransferBuffer.count { + obfuscatedBuffer.append(fileTransferBuffer.dropFirst(boundary)) + } fileTransferBuffer.removeAll() - logger.debug("[\(self.peerInfo.username)] Moved \(self.receiveBuffer.count) bytes back to receive buffer") + logger.debug("[\(self.peerInfo.username)] Restored \(boundary) decoded bytes to receive buffer, \(self.obfuscatedBuffer.count) undecoded to obfuscated buffer") } + fileBufferDecodedPrefixLength = 0 startReceiving() } @@ -943,10 +1092,13 @@ public actor PeerConnection { private func startReceiving() { guard let connection else { logger.warning("[\(self.peerInfo.username)] startReceiving called but no connection!") + // A rearm that finds no connection still has to release waiters. + markMessageLoopReceiveLanded() return } logger.debug("[\(self.peerInfo.username)] Starting receive loop...") + messageLoopReceiveArmed = true connection.receive(minimumIncompleteLength: 1, maximumLength: 262144) { [weak self] data, _, isComplete, error in guard let self else { @@ -967,6 +1119,7 @@ public actor PeerConnection { await self.appendToFileTransferBuffer(data) self.logger.debug("[\(username)] Receive loop stopped, stored \(data.count) bytes for file transfer") } + await self.markMessageLoopReceiveLanded() return // Don't continue receive loop } @@ -984,16 +1137,19 @@ public actor PeerConnection { // receives and cause a race condition. if await self.shouldStopReceiving { self.logger.debug("[\(username)] Receive loop stopped after processing (file transfer mode)") + await self.markMessageLoopReceiveLanded() return } if isComplete { self.logger.debug("[\(username)] Connection complete, disconnecting") await self.disconnect() + await self.markMessageLoopReceiveLanded() } else if error == nil { await self.startReceiving() } else { self.logger.debug("[\(username)] Not continuing receive due to error") + await self.markMessageLoopReceiveLanded() } } } @@ -1003,6 +1159,23 @@ public actor PeerConnection { fileTransferBuffer.append(data) } + /// On obfuscated connections, fileTransferBuffer can hold decoded plain + /// bytes (moved from receiveBuffer at a flip) followed by undecoded wire + /// bytes. This boundary lets `resumeReceivingForPeerConnection` split + /// them back instead of feeding cipher bytes to the plain parser. + private var fileBufferDecodedPrefixLength = 0 + + /// Call at every plain→file-mode flip, after moving receiveBuffer over: + /// records the decoded/undecoded boundary and migrates any undecoded + /// obfuscated tail so it isn't stranded in `obfuscatedBuffer`. + private func migrateObfuscatedTailToFileBuffer() { + fileBufferDecodedPrefixLength = fileTransferBuffer.count + guard isObfuscated, !obfuscatedBuffer.isEmpty else { return } + logger.debug("[\(self.peerInfo.username)] Moving \(self.obfuscatedBuffer.count) undecoded obfuscated bytes to file transfer buffer") + fileTransferBuffer.append(obfuscatedBuffer) + obfuscatedBuffer.removeAll() + } + // Track if we've completed handshake private var handshakeComplete = false private var peerHandshakeReceived = false // True when we receive peer's PeerInit @@ -1040,7 +1213,7 @@ public actor PeerConnection { private func handleReceivedData(_ data: Data) async { bytesReceived += UInt64(data.count) - lastActivityAt = Date() + touchLastActivity() if isObfuscated { obfuscatedBuffer.append(data) @@ -1193,6 +1366,7 @@ public actor PeerConnection { logger.debug("PierceFirewall: moved \(self.receiveBuffer.count) bytes to file transfer buffer") receiveBuffer.removeAll() } + migrateObfuscatedTailToFileBuffer() eventContinuation.yield(.pierceFirewall(token: token)) } @@ -1244,6 +1418,7 @@ public actor PeerConnection { logger.debug("F connection: moved \(self.receiveBuffer.count) bytes from receive buffer to file transfer buffer") receiveBuffer.removeAll() } + migrateObfuscatedTailToFileBuffer() logger.debug("F connection detected, yielding fileTransferConnection event") eventContinuation.yield(.fileTransferConnection(username: username, token: peerToken, connection: self)) @@ -1636,12 +1811,12 @@ public actor PeerConnection { private func recordSent(_ bytes: Int) { bytesSent += UInt64(bytes) messagesSent += 1 - lastActivityAt = Date() + touchLastActivity() } private func recordReceived(_ bytes: Int) { bytesReceived += UInt64(bytes) - lastActivityAt = Date() + touchLastActivity() } // MARK: - Zlib Decompression diff --git a/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Connections/PeerConnectionPool.swift b/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Connections/PeerConnectionPool.swift index 4ecd10b..d102e11 100644 --- a/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Connections/PeerConnectionPool.swift +++ b/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Connections/PeerConnectionPool.swift @@ -113,9 +113,17 @@ public final class PeerConnectionPool { // MARK: - Per-IP Connection Tracking private var connectionsPerIP: [String: Int] = [:] + /// Connection ids (incoming only) currently holding a per-IP slot. + /// Makes slot release idempotent across the many removal paths. + private var heldIPSlots: Set = [] // SECURITY: Track connection attempts per IP for rate limiting private var connectionAttempts: [String: [Date]] = [:] + /// Monotonic suffix for outgoing connection ids. Two token-0 direct + /// dials to the same user would otherwise share a key and the second + /// would overwrite the first's tracking. + @ObservationIgnored private var outgoingConnectionSequence: UInt64 = 0 + /// Token → expected username for an upcoming PierceFirewall. /// /// Populated by callers that send `ConnectToPeer` (browse, folder, @@ -285,25 +293,32 @@ public final class PeerConnectionPool { let connection = PeerConnection(peerInfo: peerInfo, token: token, isObfuscated: obfuscated) // Start consuming events BEFORE connecting to avoid missing early events - let outgoingId = "\(username)-\(token)" - consumeEvents(from: connection, username: username, connectionId: outgoingId, capturedIP: peerInfo.ip, isIncoming: false) + outgoingConnectionSequence += 1 + let connectionId = "\(username)-\(token)-\(outgoingConnectionSequence)" + consumeEvents(from: connection, username: username, connectionId: connectionId, capturedIP: peerInfo.ip, isIncoming: false) try await connection.connect() // For DIRECT connections, send PeerInit to identify ourselves // For INDIRECT connections (responding to ConnectToPeer), skip PeerInit - caller will send PierceFirewall - if !isIndirect { - if !ourUsername.isEmpty { - try await connection.sendPeerInit(username: ourUsername) - logger.debug("Sent PeerInit to \(username) as '\(self.ourUsername)'") + do { + if !isIndirect { + if !ourUsername.isEmpty { + try await connection.sendPeerInit(username: ourUsername) + logger.debug("Sent PeerInit to \(username) as '\(self.ourUsername)'") + } else { + logger.warning("ourUsername not set, skipping PeerInit") + } } else { - logger.warning("ourUsername not set, skipping PeerInit") + logger.debug("Indirect connection to \(username) - skipping PeerInit (will send PierceFirewall)") } - } else { - logger.debug("Indirect connection to \(username) - skipping PeerInit (will send PierceFirewall)") + } catch { + // Handshake failed after TCP connect: close the socket so it + // doesn't linger open and untracked. + await connection.disconnect() + throw error } - let connectionId = "\(username)-\(token)" let info = PeerConnectionInfo( id: connectionId, username: username, @@ -339,7 +354,26 @@ public final class PeerConnectionPool { isObfuscated: obfuscated ) - try await connection.accept() + // Race accept against a timeout. On timeout, cancel the NWConnection + // so the pending accept continuation resolves instead of hanging. + try await withThrowingTaskGroup(of: Void.self) { group in + group.addTask { + try await connection.accept() + } + group.addTask { + try await Task.sleep(for: .seconds(15)) + nwConnection.cancel() + throw PeerConnectionError.timeout + } + do { + _ = try await group.next() + } catch { + nwConnection.cancel() + group.cancelAll() + throw error + } + group.cancelAll() + } // We'll know the username after handshake // NOTE: Don't start receiving yet - caller must set up callbacks first, then call beginReceiving() @@ -368,6 +402,9 @@ public final class PeerConnectionPool { } let peerIP = Self.canonicalIP(from: nwConnection.endpoint) + // Generate the id before taking the per-IP slot so the catch below + // can release it idempotently. + let connectionId = "incoming-\(UUID().uuidString.prefix(8))" // Enforce per-IP connection limit to prevent single peer from exhausting resources if !peerIP.isEmpty { @@ -400,13 +437,12 @@ public final class PeerConnectionPool { // Increment per-IP counter connectionsPerIP[ip] = currentCount + 1 + heldIPSlots.insert(connectionId) } do { let connection = try await acceptIncoming(nwConnection, obfuscated: obfuscated) - let connectionId = "incoming-\(UUID().uuidString.prefix(8))" - let capturedIP = peerIP consumeEvents(from: connection, username: "unknown", connectionId: connectionId, capturedIP: capturedIP, isIncoming: true) @@ -434,6 +470,7 @@ public final class PeerConnectionPool { logger.info("Incoming connection accepted and callbacks configured") logger.info("Incoming connection stored: \(connectionId), receive loop started") } catch { + releaseIPSlot(connectionId: connectionId, ip: peerIP) // Inbound timeouts / refused / resets are normal on a // public-facing peer (dead peers, NAT tarpits). Debug only. logger.debug("Failed to handle incoming connection: \(error.localizedDescription)") @@ -449,6 +486,9 @@ public final class PeerConnectionPool { info.username == username ? key : nil } for key in keysToRemove { + if let info = connections[key] { + releaseIPSlot(connectionId: key, ip: info.ip) + } connections.removeValue(forKey: key) if let conn = activeConnections_.removeValue(forKey: key) { await conn.disconnect() @@ -510,6 +550,8 @@ public final class PeerConnectionPool { activeConnections_.removeAll() connections.removeAll() lastActivities.removeAll() + connectionsPerIP.removeAll() + heldIPSlots.removeAll() activeConnections = 0 } @@ -536,6 +578,9 @@ public final class PeerConnectionPool { return connection } else { logger.debug("Found stale connection for \(username) (key: \(key)), removing") + if let info = connections[key] { + releaseIPSlot(connectionId: key, ip: info.ip) + } activeConnections_.removeValue(forKey: key) connections.removeValue(forKey: key) } @@ -562,8 +607,8 @@ public final class PeerConnectionPool { /// Writes to the non-observable `lastActivities` dict so SwiftUI views /// observing `connections` don't invalidate on every peer message. /// `connectionId` is the dict key for both incoming and outgoing - /// connections (outgoing keys are `"\(username)-\(token)"`, set in - /// `connect(...)`), so this is a direct lookup — no prefix scan. + /// connections (outgoing keys are `"\(username)-\(token)-\(seq)"`, set + /// in `connect(...)`), so this is a direct lookup — no prefix scan. private func touchActivity(connectionId: String) { lastActivities[connectionId] = Date() } @@ -593,18 +638,25 @@ public final class PeerConnectionPool { var toRemove: [String] = [] for (id, info) in connections { + // Connection-level byte activity. Pool events only fire per + // decoded message, so a multi-MB frame accumulating (browse) + // bumps no event for minutes — but bytes ARE flowing. Read the + // connection's Mutex-backed timestamp to avoid reaping it. + let byteActivity = activeConnections_[id]?.lastActivityAt if let lastActivity = lastActivities[id] { if lastActivity <= idleCutoff { + if let byteActivity, byteActivity > idleCutoff { continue } toRemove.append(id) } } else if let connectedAt = info.connectedAt, connectedAt <= stuckHandshakeCutoff { + if let byteActivity, byteActivity > stuckHandshakeCutoff { continue } toRemove.append(id) } } for id in toRemove { if let info = connections[id] { - decrementIPCounter(for: info.ip) + releaseIPSlot(connectionId: id, ip: info.ip) } if let conn = activeConnections_[id] { Task { await conn.disconnect() } @@ -652,6 +704,14 @@ public final class PeerConnectionPool { } } + /// Idempotent release of an incoming connection's per-IP slot. Safe to + /// call from every removal path; only the first call decrements, and + /// outgoing ids (which never take slots) are no-ops. + private func releaseIPSlot(connectionId: String, ip: String) { + guard heldIPSlots.remove(connectionId) != nil else { return } + decrementIPCounter(for: ip) + } + /// Decrement per-IP connection counter (call when removing a connection) private func decrementIPCounter(for ip: String) { guard !ip.isEmpty else { return } @@ -761,9 +821,7 @@ public final class PeerConnectionPool { // (e.g. "bob-1" vs "bob-12"). connections[connectionId]?.state = state if case .disconnected = state { - if isIncoming { - decrementIPCounter(for: capturedIP) - } + releaseIPSlot(connectionId: connectionId, ip: capturedIP) connections.removeValue(forKey: connectionId) activeConnections_.removeValue(forKey: connectionId) activeConnections = connections.count @@ -775,9 +833,7 @@ public final class PeerConnectionPool { // Close connection after results received Task { await connection.disconnect() - if isIncoming { - self.decrementIPCounter(for: capturedIP) - } + self.releaseIPSlot(connectionId: connectionId, ip: capturedIP) self.connections.removeValue(forKey: connectionId) self.activeConnections_.removeValue(forKey: connectionId) self.activeConnections = self.connections.count @@ -808,7 +864,7 @@ public final class PeerConnectionPool { "Blocked inbound peer: \(discoveredUsername)", detail: capturedIP.isEmpty ? "matches block pattern" : "\(capturedIP) — matches block pattern" ) - decrementIPCounter(for: capturedIP) + releaseIPSlot(connectionId: connectionId, ip: capturedIP) connections.removeValue(forKey: connectionId) activeConnections_.removeValue(forKey: connectionId) activeConnections = connections.count @@ -845,7 +901,7 @@ public final class PeerConnectionPool { // insight into raw file bytes flowing outside its peer-message // framing. Same pattern as .pierceFirewall below; both events // transfer ownership of the connection from pool to consumer. - decrementIPCounter(for: capturedIP) + releaseIPSlot(connectionId: connectionId, ip: capturedIP) connections.removeValue(forKey: connectionId) activeConnections_.removeValue(forKey: connectionId) activeConnections = connections.count @@ -854,7 +910,7 @@ public final class PeerConnectionPool { case .pierceFirewall(let token): logger.info("PierceFirewall received: token=\(token)") incrementPierceFirewallCount() - decrementIPCounter(for: capturedIP) + releaseIPSlot(connectionId: connectionId, ip: capturedIP) // Stamp the username on the connection BEFORE we yield the // event or process any subsequent message on this connection. diff --git a/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Connections/ServerConnection.swift b/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Connections/ServerConnection.swift index 0956758..eac42ed 100644 --- a/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Connections/ServerConnection.swift +++ b/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Connections/ServerConnection.swift @@ -45,8 +45,21 @@ public actor ServerConnection { // Connection continuation - stored as property to ensure single-resume safety private var connectContinuation: CheckedContinuation? + // Bounds the connect attempt: NWConnection can sit in `.waiting` + // indefinitely (e.g. connection refused keeps retrying on path + // changes and never reaches `.failed`), which would otherwise hang + // the caller forever and leave `state` stuck at `.connecting`. + private var connectTimeoutTask: Task? + private static let connectTimeoutSeconds: Int = 15 + // Async stream for messages private var messageContinuation: AsyncStream.Continuation? + // Frames parsed before a consumer registers (the stream's continuation + // is handed over via an actor hop, so there's a window after `.ready` + // where messages would otherwise be dropped). Flushed on registration. + private var pendingMessages: [Data] = [] + private static let maxPendingMessages = 2048 + private var streamGeneration = 0 private let logger = Logger(subsystem: "com.seeleseek", category: "ServerConnection") @@ -70,17 +83,29 @@ public actor ServerConnection { Task { await self.setMessageContinuation(continuation) } - continuation.onTermination = { @Sendable _ in - Task { await self.clearContinuation() } - } } } private func setMessageContinuation(_ continuation: AsyncStream.Continuation) { + // A second consumer replaces the first; finish the old stream so its + // `for await` loop exits instead of silently going quiet. + messageContinuation?.finish() + streamGeneration += 1 + let generation = streamGeneration messageContinuation = continuation + continuation.onTermination = { @Sendable _ in + Task { await self.clearContinuation(ifGeneration: generation) } + } + // Flush frames that arrived before the consumer registered. + for frame in pendingMessages { + continuation.yield(frame) + } + pendingMessages.removeAll() } - private func clearContinuation() { + private func clearContinuation(ifGeneration generation: Int) { + // A stale stream's termination must not tear down a newer stream. + guard generation == streamGeneration else { return } messageContinuation = nil } @@ -115,6 +140,7 @@ public actor ServerConnection { } guard let nwPort = NWEndpoint.Port(rawValue: port) else { + updateState(.disconnected) throw ConnectionError.connectionFailed("Invalid port: \(port)") } let endpoint = NWEndpoint.hostPort( @@ -127,6 +153,11 @@ public actor ServerConnection { return try await withCheckedThrowingContinuation { continuation in self.connectContinuation = continuation + self.connectTimeoutTask = Task { [weak self] in + try? await Task.sleep(for: .seconds(Self.connectTimeoutSeconds)) + guard !Task.isCancelled else { return } + await self?.handleConnectTimeout() + } conn.stateUpdateHandler = { [weak self] newState in guard let self else { return } Task { @@ -137,18 +168,40 @@ public actor ServerConnection { } } + /// Resume the pending connect continuation exactly once and stop the + /// connect timeout. Safe to call from any path; no-ops when nothing + /// is pending. + private func resumeConnectContinuation(throwing error: Error?) { + connectTimeoutTask?.cancel() + connectTimeoutTask = nil + guard let continuation = connectContinuation else { return } + connectContinuation = nil + if let error { + continuation.resume(throwing: error) + } else { + continuation.resume() + } + } + + private func handleConnectTimeout() { + guard connectContinuation != nil else { return } + logger.error("Connect to \(self.host):\(self.port) timed out after \(Self.connectTimeoutSeconds)s") + resumeConnectContinuation(throwing: ConnectionError.timeout) + // Tear down so state returns to .disconnected and a future + // connect() isn't no-op'd by the .connecting guard. + disconnect() + } + public func disconnect() { connection?.cancel() connection = nil receiveBuffer = Data() // Resume any pending connect continuation before state change - if let continuation = connectContinuation { - connectContinuation = nil - continuation.resume(throwing: ConnectionError.notConnected) - } + resumeConnectContinuation(throwing: ConnectionError.notConnected) // Finish the async message stream so NetworkClient's `for await` loop exits messageContinuation?.finish() messageContinuation = nil + pendingMessages.removeAll() updateState(.disconnected) } @@ -219,20 +272,12 @@ public actor ServerConnection { case .ready: logger.info("Connected to \(self.host):\(self.port)") updateState(.connected) - // Resume continuation exactly once, then nil it out - if let continuation = connectContinuation { - connectContinuation = nil - continuation.resume() - } + resumeConnectContinuation(throwing: nil) Task { await startReceiving() } case .failed(let error): logger.error("Connection failed: \(error.localizedDescription)") - // Resume continuation exactly once if still connecting - if let continuation = connectContinuation { - connectContinuation = nil - continuation.resume(throwing: ConnectionError.connectionFailed(error.localizedDescription)) - } + resumeConnectContinuation(throwing: ConnectionError.connectionFailed(error.localizedDescription)) // Clean up and end the async stream so NetworkClient detects the loss disconnect() @@ -240,13 +285,23 @@ public actor ServerConnection { logger.info("Connection cancelled") updateState(.disconnected) // If cancelled during connect, resume with error - if let continuation = connectContinuation { - connectContinuation = nil - continuation.resume(throwing: ConnectionError.connectionFailed("Connection cancelled")) - } + resumeConnectContinuation(throwing: ConnectionError.connectionFailed("Connection cancelled")) case .waiting(let error): logger.warning("Connection waiting: \(error.localizedDescription)") + // TCP-level refusals/unreachability park NWConnection in .waiting + // (it retries on path changes and never reaches .failed). Treat + // the definitive POSIX codes as failures so connect() doesn't + // hang until the timeout: ENOMEM, ENETUNREACH, ENOTCONN, + // ETIMEDOUT, ECONNREFUSED, EHOSTUNREACH. + if case .posix(let posixError) = error { + let code = posixError.rawValue + if code == 12 || code == 51 || code == 57 || code == 60 || code == 61 || code == 65 { + logger.error("Server connection definitive failure: POSIX \(code)") + resumeConnectContinuation(throwing: ConnectionError.connectionFailed(error.localizedDescription)) + disconnect() + } + } default: break @@ -304,8 +359,14 @@ public actor ServerConnection { completeMessage.appendUInt32(frame.code) completeMessage.append(frame.payload) - // Yield to async stream - messageContinuation?.yield(completeMessage) + // Yield to async stream; if the consumer hasn't registered its + // continuation yet (actor hop in `messages`), buffer the frame + // so a fast login response isn't dropped. + if let messageContinuation { + messageContinuation.yield(completeMessage) + } else if pendingMessages.count < Self.maxPendingMessages { + pendingMessages.append(completeMessage) + } // Also call legacy handler if set await messageHandler?(frame.code, frame.payload) diff --git a/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Handlers/ServerMessageHandler.swift b/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Handlers/ServerMessageHandler.swift index 5d580fd..167ed67 100644 --- a/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Handlers/ServerMessageHandler.swift +++ b/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Handlers/ServerMessageHandler.swift @@ -346,7 +346,7 @@ public final class ServerMessageHandler { guard info.exists else { Task { @MainActor in - self.client?.handleUserStatusResponse(username: info.username, status: .offline, privileged: false) + self.client?.handleUserStatusResponse(username: info.username, status: .offline, privileged: nil) } return } @@ -358,7 +358,9 @@ public final class ServerMessageHandler { let dirs = info.dirs ?? 0 Task { @MainActor in - self.client?.handleUserStatusResponse(username: info.username, status: status, privileged: false) + // WatchUser replies don't carry privileged — nil keeps the + // last server-reported value instead of fabricating false. + self.client?.handleUserStatusResponse(username: info.username, status: status, privileged: nil) self.client?.dispatchUserStats(username: info.username, avgSpeed: avgSpeed, uploadNum: UInt64(uploadNum), files: files, dirs: dirs) } @@ -566,8 +568,18 @@ public final class ServerMessageHandler { return } + // Adoption runs in a task that can take up to 15s (3 × 5s connects); + // a second PossibleParents in that window must not start a second + // adoption loop (duplicate parents, inconsistent branch reports). + if isConnectingToParent { + logger.debug("Parent adoption already in progress, ignoring PossibleParents") + return + } + isConnectingToParent = true + // Try to connect to first few parents until one succeeds (limit to avoid resource exhaustion) Task { + defer { isConnectingToParent = false } let maxAttempts = min(3, parents.count) for i in 0..? private var pingTask: Task? private var loginContinuation: CheckedContinuation? + private var loginTimeoutTask: Task? + private var loginAttemptGeneration = 0 // MARK: - Auto-Reconnect private var reconnectTask: Task? @@ -530,6 +532,15 @@ public final class NetworkClient { logger.info("Starting connection to \(server):\(port) as \(username)") + // Let any in-flight teardown finish before starting the listener — + // otherwise the old teardown's listenerService.stop() can land + // after the new start() and silently kill the fresh listener while + // its port is still advertised to the server. (Safe re-entrancy: + // isConnecting is already true, so a concurrent connect() bails at + // the guard above.) + await teardownTask?.value + teardownTask = nil + do { // Step 1: Start listener for incoming peer connections listenerConsumerTask?.cancel() @@ -581,26 +592,48 @@ public final class NetworkClient { loginSuccess = try await withCheckedThrowingContinuation { (continuation: CheckedContinuation) in self.loginContinuation = continuation - // Timeout after 10 seconds so we don't wait forever - Task { + // Timeout after 10 seconds so we don't wait forever. + // Generation-stamped: an uncancelled timer from attempt N + // must never resume attempt N+1's continuation (quick + // disconnect/reconnect inside the 10s window). + self.loginAttemptGeneration += 1 + let generation = self.loginAttemptGeneration + self.loginTimeoutTask?.cancel() + self.loginTimeoutTask = Task { try? await Task.sleep(for: .seconds(10)) - if let pending = self.loginContinuation { - self.loginContinuation = nil - pending.resume(throwing: ServerConnection.ConnectionError.timeout) - } + guard !Task.isCancelled, + generation == self.loginAttemptGeneration, + let pending = self.loginContinuation else { return } + self.loginContinuation = nil + pending.resume(throwing: ServerConnection.ConnectionError.timeout) } } + loginTimeoutTask?.cancel() + loginTimeoutTask = nil } catch { - // Login timed out or failed — don't auto-reconnect on auth failure. + loginTimeoutTask?.cancel() + loginTimeoutTask = nil // Route through the full teardown so the listener, NAT, peer // pool, pending waiters, distributed state, and server socket // all get cleaned up — previously this path only stopped the // listener, leaving stale state for the next connect(). isConnecting = false connectionError = error.localizedDescription - shouldAutoReconnect = false + // Only a credential rejection should disable auto-reconnect; + // a timeout or socket error during the handshake is transient + // and the user asked for a persistent connection. + let wasEligibleForReconnect: Bool + if case ServerConnection.ConnectionError.loginFailed = error { + shouldAutoReconnect = false + wasEligibleForReconnect = false + } else { + wasEligibleForReconnect = shouldAutoReconnect + } performDisconnect() await teardownTask?.value + if wasEligibleForReconnect { + scheduleReconnect(reason: error.localizedDescription) + } return } @@ -616,6 +649,8 @@ public final class NetworkClient { obfuscatedPort: UInt32(obfuscatedPort) ) try await connection.send(portMessage) + lastAdvertisedListenPort = listenPort + lastAdvertisedObfuscatedPort = obfuscatedPort // Step 6: Set online status let statusMessage = MessageBuilder.setOnlineStatusMessage(status: .online) @@ -937,7 +972,53 @@ public final class NetworkClient { // MARK: - NAT Setup (Background) + /// NAT-PMP (and UPnP with a busy port) may grant a DIFFERENT external + /// port than requested. Peers dial the external port, so the server must + /// be re-told whenever the granted port differs from what login step 5 + /// advertised — otherwise peers dial a port the router doesn't forward. + private var externalListenPort: UInt16 = 0 // 0 = no mapping / same as listenPort + private var externalObfuscatedPort: UInt16 = 0 + private var lastAdvertisedListenPort: UInt16 = 0 + private var lastAdvertisedObfuscatedPort: UInt16 = 0 + + private func readvertiseListenPortsIfNeeded() async { + guard isConnected else { return } + let effectivePort = externalListenPort > 0 ? externalListenPort : listenPort + let effectiveObfuscated = externalObfuscatedPort > 0 ? externalObfuscatedPort : obfuscatedPort + guard effectivePort != lastAdvertisedListenPort + || effectiveObfuscated != lastAdvertisedObfuscatedPort else { return } + do { + let message = MessageBuilder.setListenPortMessage( + port: UInt32(effectivePort), + obfuscatedPort: UInt32(effectiveObfuscated) + ) + try await requireConnectedServerConnection().send(message) + lastAdvertisedListenPort = effectivePort + lastAdvertisedObfuscatedPort = effectiveObfuscated + logger.info("NAT: re-advertised external ports \(effectivePort) (obfuscated \(effectiveObfuscated)) to server") + } catch { + logger.warning("NAT: failed to re-advertise external port: \(error.localizedDescription)") + } + } + private func setupNATInBackground() async { + externalListenPort = 0 + externalObfuscatedPort = 0 + // Refresh can renumber a mapping (router reboot, lease churn) — + // push the new external port to the server when it does. + await natService.setOnExternalPortChanged { [weak self] internalPort, newExternalPort in + Task { @MainActor [weak self] in + guard let self else { return } + if internalPort == self.listenPort { + self.externalListenPort = newExternalPort + } else if internalPort == self.obfuscatedPort { + self.externalObfuscatedPort = newExternalPort + } else { + return + } + await self.readvertiseListenPortsIfNeeded() + } + } // Check if UPnP/NAT-PMP is enabled in settings let enableNAT = UserDefaults.standard.object(forKey: "settingsEnableUPnP") == nil ? true // Default to enabled @@ -965,6 +1046,7 @@ public final class NetworkClient { do { let mappedPort = try await natService.mapPort(listenPort) logger.info("NAT: Mapped port \(self.listenPort) -> \(mappedPort)") + externalListenPort = mappedPort } catch { logger.warning("NAT: Port mapping failed (will rely on server-mediated connections)") } @@ -977,11 +1059,16 @@ public final class NetworkClient { do { let mappedObfuscated = try await natService.mapPort(obfuscatedPort) logger.info("NAT: Mapped obfuscated port \(self.obfuscatedPort) -> \(mappedObfuscated)") + externalObfuscatedPort = mappedObfuscated } catch { // Silent failure for obfuscated port } } + // The router may have granted different external ports than the + // ones login advertised — tell the server about the real ones. + await readvertiseListenPortsIfNeeded() + // Discover external IP if let extIP = await natService.discoverExternalIP() { await MainActor.run { @@ -1396,19 +1483,18 @@ public final class NetworkClient { try? await Task.sleep(for: .seconds(timeout)) guard let self else { return } - // If still pending without a connection, mark as timed out + // If still pending without a connection, fail the waiter and + // drop the entry — nothing consumes it after this point, and + // leaving it in the dict leaked one entry per failed browse + // (and let a late PierceFirewall park an orphaned socket in it). if var pending = self.pendingBrowseStates[token] { if pending.receivedConnection == nil { logger.warning("Browse: Timeout waiting for PierceFirewall from \(pending.username) (token=\(token))") - pending.timedOut = true - self.pendingBrowseStates[token] = pending - - // If there's a continuation waiting, resume it with error if let continuation = pending.continuation { pending.continuation = nil - self.pendingBrowseStates[token] = pending continuation.resume(throwing: NetworkError.timeout) } + self.pendingBrowseStates.removeValue(forKey: token) } } // Clear any leftover username pre-registration. @@ -1442,34 +1528,54 @@ public final class NetworkClient { } } - // Wait for connection - return try await withCheckedThrowingContinuation { continuation in - if var state = pendingBrowseStates[token] { - // Check again if connection arrived while we were setting up - if let connection = state.receivedConnection { - pendingBrowseStates.removeValue(forKey: token) - continuation.resume(returning: connection) - return - } - if state.timedOut { - pendingBrowseStates.removeValue(forKey: token) + // Wait for connection. Cancellation-aware: callers race this waiter + // against a direct connect inside task groups — without the handler, + // group.cancelAll() can't unwind the waiter and the group blocks at + // scope exit until the registration timeout fires. + return try await withTaskCancellationHandler { + try await withCheckedThrowingContinuation { continuation in + if var state = pendingBrowseStates[token] { + // Check again if connection arrived while we were setting up + if let connection = state.receivedConnection { + pendingBrowseStates.removeValue(forKey: token) + continuation.resume(returning: connection) + return + } + if state.timedOut { + pendingBrowseStates.removeValue(forKey: token) + continuation.resume(throwing: NetworkError.timeout) + return + } + if let reason = state.failureReason { + pendingBrowseStates.removeValue(forKey: token) + continuation.resume(throwing: NetworkError.connectionFailed(reason)) + return + } + state.continuation = continuation + pendingBrowseStates[token] = state + } else { + // Token was already removed (cancelled or error) continuation.resume(throwing: NetworkError.timeout) - return } - if let reason = state.failureReason { - pendingBrowseStates.removeValue(forKey: token) - continuation.resume(throwing: NetworkError.connectionFailed(reason)) - return - } - state.continuation = continuation - pendingBrowseStates[token] = state - } else { - // Token was already removed (cancelled or error) - continuation.resume(throwing: NetworkError.timeout) + } + } onCancel: { + Task { @MainActor [weak self] in + self?.cancelWaitForPendingBrowse(token: token) } } } + /// Unblock a cancelled `waitForPendingBrowse` waiter. The entry itself + /// stays registered (minus the waiter) so a late PierceFirewall is still + /// matched and cleaned up rather than falling through unhandled. + private func cancelWaitForPendingBrowse(token: UInt32) { + guard var state = pendingBrowseStates[token], + let continuation = state.continuation else { return } + state.continuation = nil + pendingBrowseStates[token] = state + continuation.resume(throwing: CancellationError()) + } + /// Mark a pending browse as failed (e.g. server sent CantConnectToPeer). /// The waiter, if any, fails fast instead of sitting on the 30s timeout. public func failPendingBrowse(token: UInt32, reason: String) { @@ -1491,6 +1597,11 @@ public final class NetworkClient { if let state = pendingBrowseStates.removeValue(forKey: token) { state.timeoutTask?.cancel() // Don't resume continuation - caller will handle the success case + // A PierceFirewall that arrived but was never consumed (direct + // path won the race) is an orphaned live socket — close it. + if let connection = state.receivedConnection { + Task { await connection.disconnect() } + } } peerConnectionPool.clearExpectedPierceFirewallUsername(token: token) } @@ -1505,6 +1616,14 @@ public final class NetworkClient { /// it even if the pool's pre-registration map was cleared). public func handlePierceFirewallForBrowse(token: UInt32, connection: PeerConnection) async -> Bool { if var state = pendingBrowseStates[token] { + // A browse that already failed (CantConnectToPeer) has no waiter + // left; storing the connection would orphan a live socket. + if state.timedOut || state.failureReason != nil { + logger.debug("Browse: late PierceFirewall token=\(token) for dead browse; closing") + pendingBrowseStates.removeValue(forKey: token) + await connection.disconnect() + return true + } logger.debug("Browse: PierceFirewall token=\(token) matched pending browse for \(state.username)") // Stamp username synchronously — must complete BEFORE we resume @@ -1699,16 +1818,30 @@ public final class NetworkClient { } /// Handle status response - resumes every pending status check for this user - public func handleUserStatusResponse(username: String, status: UserStatus, privileged: Bool) { + /// Last privileged flag the server actually reported per user. WatchUser + /// replies don't carry privileged, so they pass nil and fall back to + /// this instead of fabricating `false` (which could clobber a concurrent + /// GetUserStatus waiter with wrong data). + private var lastKnownPrivileged: [String: Bool] = [:] + + public func handleUserStatusResponse(username: String, status: UserStatus, privileged: Bool?) { + let resolvedPrivileged: Bool + if let privileged { + lastKnownPrivileged[username] = privileged + resolvedPrivileged = privileged + } else { + resolvedPrivileged = lastKnownPrivileged[username] ?? false + } + if let waiters = pendingStatusRequests.removeValue(forKey: username) { for waiter in waiters { - waiter.continuation.resume(returning: (status: status, privileged: privileged)) + waiter.continuation.resume(returning: (status: status, privileged: resolvedPrivileged)) } } // Notify all registered status handlers for handler in userStatusHandlers { - handler(username, status, privileged) + handler(username, status, resolvedPrivileged) } } @@ -1758,14 +1891,6 @@ public final class NetworkClient { logger.info("Room search in \(room): \(query)") } - /// Legacy room search request (server code 25) - public func searchRoomLegacy(_ room: String, query: String, token: UInt32) async throws { - guard isConnected else { throw NetworkError.notConnected } - let message = MessageBuilder.fileSearchRoomMessage(room: room, token: token, query: query) - try await requireConnectedServerConnection().send(message) - logger.info("Legacy room search in \(room): \(query)") - } - /// Add a wishlist search (runs periodically) public func addWishlistSearch(query: String, token: UInt32) async throws { guard isConnected else { throw NetworkError.notConnected } @@ -2250,7 +2375,26 @@ public final class NetworkClient { registerPendingBrowse(token: token, username: username, timeout: 30) await sendConnectToPeer(token: token, username: username, connectionType: "P") - let (ip, port, obfuscatedPort) = try await getPeerAddress(for: username) + let ip: String + let port: Int + let obfuscatedPort: Int + do { + (ip, port, obfuscatedPort) = try await getPeerAddress(for: username) + } catch { + // Without this the registered browse entry (and its 30s timer) + // leaks for the session whenever the address lookup fails. + cancelPendingBrowse(token: token) + throw error + } + + // The server answers GetPeerAddress for an offline/unknown user with + // 0.0.0.0:0 — dialing port 0 throws immediately and the indirect + // wait can never succeed (the user isn't there to pierce). Fail fast + // instead of burning the full 30s window. + guard ip != "0.0.0.0", port > 0 || obfuscatedPort > 0 else { + cancelPendingBrowse(token: token) + throw NetworkError.connectionFailed("\(username) is offline (no address)") + } // Prefer the peer's obfuscated port whenever they advertise one. // Peers always advertise the plain port too, so falling back to plain diff --git a/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Protocol/Decompression.swift b/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Protocol/Decompression.swift index ba96ce1..e9bf55f 100644 --- a/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Protocol/Decompression.swift +++ b/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Protocol/Decompression.swift @@ -7,6 +7,8 @@ public enum DecompressionError: Error { case decompressionFailed case suspiciousCompressionRatio case decompressedSizeExceeded + case presetDictionaryUnsupported + case checksumMismatch } /// Standalone zlib/deflate decompression utility. @@ -28,15 +30,48 @@ public enum ZlibDecompression { let compressionMethod = cmf & 0x0F if compressionMethod == 8 { + // FDICT (FLG bit 5) means a 4-byte DICTID follows the header; + // we don't support preset dictionaries, and feeding the DICTID + // to the inflater as compressed data fails confusingly. + let flg = data[data.index(after: data.startIndex)] + guard flg & 0x20 == 0 else { + throw DecompressionError.presetDictionaryUnsupported + } // Standard zlib: strip 2-byte header and 4-byte Adler-32 footer let deflateData = Data(data.dropFirst(2).dropLast(4)) - return try decompressRawDeflate(deflateData) + let decompressed = try decompressRawDeflate(deflateData) + // Apple's buffer API can't signal a truncated DEFLATE stream — + // the Adler-32 trailer is the only integrity signal we have. + let expectedChecksum = data.suffix(4).reduce(UInt32(0)) { ($0 << 8) | UInt32($1) } + guard adler32(decompressed) == expectedChecksum else { + throw DecompressionError.checksumMismatch + } + return decompressed } else { // Not zlib format — try raw DEFLATE return try decompressRawDeflate(data) } } + /// RFC 1950 Adler-32, with the standard deferred-modulo chunking so the + /// hot loop is pure adds (5552 is the largest n with no UInt32 overflow). + nonisolated static func adler32(_ data: Data) -> UInt32 { + var a: UInt32 = 1 + var b: UInt32 = 0 + var index = data.startIndex + while index < data.endIndex { + let chunkEnd = data.index(index, offsetBy: 5552, limitedBy: data.endIndex) ?? data.endIndex + for byte in data[index.. Data { try data.withUnsafeBytes { sourceBuffer -> Data in @@ -45,44 +80,38 @@ public enum ZlibDecompression { throw DecompressionError.decompressionFailed } + // compression_decode_buffer can't distinguish "buffer too small" + // from success — it fills the buffer and returns destinationSize. + // Grow and retry until the output no longer fills the buffer; + // a result that still fills a max-size buffer is truncation, not + // success, so it must throw rather than return partial data. var destinationSize = min(max(sourceSize * 20, 65536), maxDecompressedSize) - var destinationBuffer = [UInt8](repeating: 0, count: destinationSize) - - var decodedSize = compression_decode_buffer( - &destinationBuffer, destinationSize, - baseAddress, sourceSize, - nil, COMPRESSION_ZLIB - ) - - // If output buffer was too small, retry with larger buffer (capped) - if decodedSize == 0 || decodedSize == destinationSize { - destinationSize = min(sourceSize * 100, maxDecompressedSize) - guard destinationSize <= maxDecompressedSize else { - throw DecompressionError.decompressedSizeExceeded - } - destinationBuffer = [UInt8](repeating: 0, count: destinationSize) - decodedSize = compression_decode_buffer( + while true { + var destinationBuffer = [UInt8](repeating: 0, count: destinationSize) + let decodedSize = compression_decode_buffer( &destinationBuffer, destinationSize, baseAddress, sourceSize, nil, COMPRESSION_ZLIB ) - } - guard decodedSize > 0 else { - throw DecompressionError.decompressionFailed - } + if decodedSize == 0 || decodedSize == destinationSize { + guard destinationSize < maxDecompressedSize else { + throw decodedSize == 0 + ? DecompressionError.decompressionFailed + : DecompressionError.decompressedSizeExceeded + } + destinationSize = min(destinationSize * 4, maxDecompressedSize) + continue + } - // Security: check compression ratio - let compressionRatio = decodedSize / max(sourceSize, 1) - if compressionRatio > maxCompressionRatio { - throw DecompressionError.suspiciousCompressionRatio - } + // Security: check compression ratio + let compressionRatio = decodedSize / max(sourceSize, 1) + if compressionRatio > maxCompressionRatio { + throw DecompressionError.suspiciousCompressionRatio + } - guard decodedSize <= maxDecompressedSize else { - throw DecompressionError.decompressedSizeExceeded + return Data(destinationBuffer.prefix(decodedSize)) } - - return Data(destinationBuffer.prefix(decodedSize)) } } } diff --git a/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Protocol/MessageBuilder.swift b/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Protocol/MessageBuilder.swift index 9357670..5fd2a7d 100644 --- a/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Protocol/MessageBuilder.swift +++ b/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Protocol/MessageBuilder.swift @@ -236,14 +236,9 @@ public enum MessageBuilder { // Private (buddy-only) section appendDirectories(privateFiles) - // Compress with zlib - guard let compressed = compressZlib(uncompressedPayload) else { - // Fallback: send uncompressed (not ideal but better than nothing) - var payload = Data() - payload.appendUInt32(UInt32(PeerMessageCode.sharesReply.rawValue)) - payload.append(uncompressedPayload) - return wrapMessage(payload) - } + // Compress with zlib (mandatory for code 5 — receivers inflate + // unconditionally, so an uncompressed fallback can never parse) + let compressed = compressZlib(uncompressedPayload) // Build final message var payload = Data() @@ -332,14 +327,8 @@ public enum MessageBuilder { // Private (buddy-only) results appendResults(privateResults) - // Compress with zlib - guard let compressed = compressZlib(uncompressedPayload) else { - // Fallback: send uncompressed (not ideal but better than nothing) - var payload = Data() - payload.appendUInt32(UInt32(PeerMessageCode.searchReply.rawValue)) - payload.append(uncompressedPayload) - return wrapMessage(payload) - } + // Compress with zlib (mandatory for code 9) + let compressed = compressZlib(uncompressedPayload) // Build final message var payload = Data() @@ -401,8 +390,8 @@ public enum MessageBuilder { } } - // Compress with zlib - let compressedPayload = compressZlib(uncompressedPayload) ?? uncompressedPayload + // Compress with zlib (mandatory for code 37) + let compressedPayload = compressZlib(uncompressedPayload) var payload = Data() payload.appendUInt32(UInt32(PeerMessageCode.folderContentsReply.rawValue)) @@ -411,14 +400,19 @@ public enum MessageBuilder { return wrapMessage(payload) } - /// Compress data using zlib - nonisolated private static func compressZlib(_ data: Data) -> Data? { + /// Compress data using zlib. Always produces a valid zlib stream — + /// codes 5/9/37 are mandatorily compressed on the wire, so there is + /// no legal uncompressed fallback. + nonisolated private static func compressZlib(_ data: Data) -> Data { var compressed = Data() // Add zlib header compressed.append(0x78) // CMF: compression method 8 (deflate), window size 7 compressed.append(0x9C) // FLG: default compression level - let bufferSize = max(65536, data.count) // At least 64KB, or input size + // Deflate can EXPAND incompressible input (~5 bytes per 16KB block); + // without headroom, encode returns 0 for >64KB payloads that don't + // compress below input size. + let bufferSize = data.count + data.count / 16 + 64 + 6 var compressedBuffer = [UInt8](repeating: 0, count: bufferSize) let compressedSize = data.withUnsafeBytes { sourceBuffer -> Int in @@ -435,8 +429,14 @@ public enum MessageBuilder { ) } - guard compressedSize > 0 else { return nil } - compressed.append(Data(compressedBuffer.prefix(compressedSize))) + if compressedSize > 0 { + compressed.append(Data(compressedBuffer.prefix(compressedSize))) + } else { + // Shouldn't be reachable with proper headroom (empty input is the + // one known case) — emit DEFLATE stored blocks so the output is + // still a stream the receiver can inflate. + compressed.append(storedDeflateBlocks(data)) + } // Add Adler-32 checksum let checksum = adler32(data) @@ -446,6 +446,29 @@ public enum MessageBuilder { return compressed } + /// RFC 1951 stored (uncompressed) blocks: 1-byte BFINAL/BTYPE=00 header, + /// LEN + NLEN (one's complement) little-endian, then raw bytes; max + /// 65535 bytes per block. + nonisolated private static func storedDeflateBlocks(_ data: Data) -> Data { + var out = Data() + var index = data.startIndex + repeat { + let chunkEnd = data.index(index, offsetBy: 65535, limitedBy: data.endIndex) ?? data.endIndex + let chunk = data[index..> 8)) + let nlen = ~len + out.append(UInt8(nlen & 0xFF)) + out.append(UInt8(nlen >> 8)) + out.append(chunk) + index = chunkEnd + } while index < data.endIndex + return out + } + /// Calculate Adler-32 checksum for zlib nonisolated private static func adler32(_ data: Data) -> UInt32 { var a: UInt32 = 1 @@ -665,16 +688,6 @@ public enum MessageBuilder { return wrapMessage(payload) } - /// Legacy room search code 25 (still used by some peers/servers) - public nonisolated static func fileSearchRoomMessage(room: String, token: UInt32, query: String) -> Data { - var payload = Data() - payload.appendUInt32(ServerMessageCode.fileSearchRoom.rawValue) - payload.appendString(room) - payload.appendUInt32(token) - payload.appendString(query) - return wrapMessage(payload) - } - /// Add a wishlist search (code 103) public nonisolated static func wishlistSearch(token: UInt32, query: String) -> Data { var payload = Data() diff --git a/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Protocol/MessageParser.swift b/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Protocol/MessageParser.swift index e0cb6c1..23ba436 100644 --- a/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Protocol/MessageParser.swift +++ b/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Protocol/MessageParser.swift @@ -161,6 +161,10 @@ public enum MessageParser { public let port: UInt32 public let token: UInt32 public let privileged: Bool + /// Trailing ConnectToPeer fields (0 when the server omits them): + /// the peer's obfuscation support and its obfuscated listen port. + public var obfuscationType: UInt32 = 0 + public var obfuscatedPort: UInt32 = 0 } public nonisolated static func parseConnectToPeer(_ payload: Data) -> PeerInfo? { @@ -182,6 +186,12 @@ public enum MessageParser { offset += 4 let privileged = payload.readBool(at: offset) ?? false + offset += 1 + + // Trailing fields (spec fields 7-8); older servers may omit them. + let obfuscationType = payload.readUInt32(at: offset) ?? 0 + offset += 4 + let obfuscatedPort = payload.readUInt32(at: offset) ?? 0 let ipString = formatLittleEndianIPv4(ip) @@ -191,7 +201,9 @@ public enum MessageParser { ip: ipString, port: port, token: token, - privileged: privileged + privileged: privileged, + obfuscationType: obfuscationType, + obfuscatedPort: obfuscatedPort ) } @@ -518,15 +530,11 @@ public enum MessageParser { } logger.debug("fileSize parsed: \(parsed)") - // A zero size for an upload-direction TransferRequest is - // either a parser misalignment or a peer protocol bug. Either - // way, accepting it would create a "transfer" that does - // nothing and immediately succeeds with 0/0 bytes — not - // recoverable into a valid download. Reject. - guard parsed > 0 else { - logger.warning("TransferRequest: fileSize is 0 for upload direction — rejecting as malformed") - logger.debug("Full payload hex dump: \(payload.map { String(format: "%02x", $0) }.joined(separator: " "))") - return nil + // Zero-byte files are legal shares (slskd/Seeker advertise + // size 0 for them); rejecting the message would stall a queued + // 0-byte download forever. + if parsed == 0 { + logger.debug("TransferRequest: fileSize is 0 (zero-byte file)") } fileSize = parsed } diff --git a/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Protocol/ObfuscationCodec.swift b/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Protocol/ObfuscationCodec.swift index f560b41..5124ce6 100644 --- a/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Protocol/ObfuscationCodec.swift +++ b/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Protocol/ObfuscationCodec.swift @@ -104,7 +104,9 @@ public enum ObfuscationCodec { /// allocate arbitrary memory while we wait for bytes that never arrive). public static func decodeMessage( from buffer: Data, - maxPayloadLength: Int = 16 * 1024 * 1024 + // Matches MessageParser's plain-frame limit — large share lists can + // exceed 16MB and must parse identically on obfuscated connections. + maxPayloadLength: Int = 100_000_000 ) throws -> (payload: Data, bytesConsumed: Int)? { guard buffer.count >= keyLength + 4 else { return nil } diff --git a/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Services/ListenerService.swift b/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Services/ListenerService.swift index 972b1bf..5ee3a1c 100644 --- a/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Services/ListenerService.swift +++ b/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Services/ListenerService.swift @@ -153,12 +153,21 @@ public actor ListenerService { }) else { return } continuation.resume() case .failed(let error): - guard hasResumed.withLock({ + let shouldResume = hasResumed.withLock({ guard !$0 else { return false } $0 = true return true - }) else { return } - continuation.resume(throwing: error) + }) + if shouldResume { + continuation.resume(throwing: error) + } else { + // Post-startup death (network change, interface drop): + // without this the listener stays non-nil and + // listeningPort keeps advertising a dead port. + Task { [weak self] in + await self?.handlePlainListenerFailure(error) + } + } case .cancelled: break default: @@ -238,10 +247,18 @@ public actor ListenerService { self.obfuscatedListener = newListener } + private func handlePlainListenerFailure(_ error: Error) { + logger.error("Listener failed after startup: \(error.localizedDescription)") + listener?.cancel() + listener = nil + listeningPort = 0 + } + private func handleObfuscatedListenerFailure(_ error: Error) { logger.error("Obfuscated listener failed: \(error.localizedDescription)") obfuscatedListener?.cancel() obfuscatedListener = nil + obfuscatedPort = 0 } private func handleNewConnection(_ connection: NWConnection, obfuscated: Bool) { diff --git a/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Services/NATService.swift b/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Services/NATService.swift index 518bddb..f302659 100644 --- a/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Services/NATService.swift +++ b/Packages/SeeleseekCore/Sources/SeeleseekCore/Network/Services/NATService.swift @@ -34,7 +34,9 @@ public actor NATService { private struct InternalMapping: Sendable { let internalPort: UInt16 - let externalPort: UInt16 + /// Router-assigned. Refresh can renumber it (router reboot, lease + /// churn), so it's mutable and renumbers fire `onExternalPortChanged`. + var externalPort: UInt16 let proto: String let method: MappingMethod /// Lease duration the router granted (seconds). `0` means permanent @@ -43,6 +45,14 @@ public actor NATService { var createdAt: Date } + /// Called with (internalPort, newExternalPort) when a refresh renumbers + /// a mapping — the owner re-advertises the new port to the server. + private var onExternalPortChanged: (@Sendable (UInt16, UInt16) -> Void)? + + public func setOnExternalPortChanged(_ handler: @escaping @Sendable (UInt16, UInt16) -> Void) { + onExternalPortChanged = handler + } + private struct UPnPGateway: Sendable { let ip: String let controlURL: String @@ -220,15 +230,16 @@ public actor NATService { for idx in indices { let mapping = mappedPorts[idx] do { + let refreshedPort: UInt16 switch mapping.method { case .upnp: - _ = try await mapPortUPnPWithFallback( + refreshedPort = try await mapPortUPnPWithFallback( mapping.internalPort, externalPort: mapping.externalPort, protocol: mapping.proto - ) + ).port case .natpmp: - _ = try await mapPortNATPMPRaw( + refreshedPort = try await mapPortNATPMPRaw( mapping.internalPort, externalPort: mapping.externalPort, protocol: mapping.proto, @@ -237,8 +248,16 @@ public actor NATService { } if idx < mappedPorts.count { mappedPorts[idx].createdAt = Date() + if refreshedPort != mapping.externalPort { + // Router renumbered the mapping — record it and let + // the owner re-advertise, or the recorded external + // port silently goes stale. + logger.warning("Mapping renumbered: \(mapping.proto) \(mapping.internalPort) external \(mapping.externalPort) -> \(refreshedPort)") + mappedPorts[idx].externalPort = refreshedPort + onExternalPortChanged?(mapping.internalPort, refreshedPort) + } } - logger.debug("Refreshed \(mapping.proto) \(mapping.internalPort)->\(mapping.externalPort)") + logger.debug("Refreshed \(mapping.proto) \(mapping.internalPort)->\(refreshedPort)") } catch { logger.warning("Mapping refresh failed for \(mapping.proto) \(mapping.externalPort): \(error.localizedDescription)") if case .upnp = mapping.method { diff --git a/Packages/SeeleseekCore/Sources/SeeleseekCore/Storage/DownloadManager.swift b/Packages/SeeleseekCore/Sources/SeeleseekCore/Storage/DownloadManager.swift index 9a51f94..9d37178 100644 --- a/Packages/SeeleseekCore/Sources/SeeleseekCore/Storage/DownloadManager.swift +++ b/Packages/SeeleseekCore/Sources/SeeleseekCore/Storage/DownloadManager.swift @@ -25,15 +25,37 @@ public final class DownloadManager { // Array-based to support multiple concurrent downloads from same user private var pendingFileTransfersByUser: [String: [PendingFileTransfer]] = [:] - /// Per-transferToken watchdog Task spawned in `handleTransferRequest`. - /// Sleeps 60 s and forces the row to fail if no F connection arrived. - /// Tracked so any path that consumes a pending entry — direct/indirect - /// success, F-fallback success, F-fallback failure, manual cancel — - /// can cancel the orphan before it wakes up and stomps a row that has - /// already moved on. `removePendingFileTransfer` is the single point - /// where this dict is cleared, so all callers get the cleanup for - /// free. - private var fileTransferWatchdogs: [UInt32: Task] = [:] + /// Per-(username, transferToken) watchdog Task spawned in + /// `handleTransferRequest`. Sleeps 60 s and forces the row to fail if no + /// F connection arrived. Tracked so any path that consumes a pending + /// entry — direct/indirect success, F-fallback success, F-fallback + /// failure, manual cancel — can cancel the orphan before it wakes up and + /// stomps a row that has already moved on. `removePendingFileTransfer` + /// is the single point where this dict is cleared, so all callers get + /// the cleanup for free. Keyed by `watchdogKey(username:transferToken:)` + /// rather than the bare token: tokens are peer-chosen and frequently + /// small, so two peers can collide on the same value — token-only keying + /// let one peer's registration cancel the other's watchdog. + private var fileTransferWatchdogs: [String: Task] = [:] + + private func watchdogKey(username: String, transferToken: UInt32) -> String { + "\(username)\u{0}\(transferToken)" + } + + /// Transfers the user has cancelled. The receive loops, completion + /// paths, salvage/resurrect paths, and retry machinery consult this so a + /// cancelled download stays terminal: nothing resurrects it except an + /// explicit (re)start of the same transferId, which clears the entry. + /// Partial files are kept on cancel so a later manual retry can resume. + private var cancelledTransferIds: Set = [] + + /// A transfer is "cancelled" if the user marked it so (in our set) or + /// the row already reads `.cancelled` (covers app-side cancels that + /// flipped the row before/without calling `cancelDownload`). + private func isCancelled(_ transferId: UUID) -> Bool { + cancelledTransferIds.contains(transferId) + || transferState?.getTransfer(id: transferId)?.status == .cancelled + } // MARK: - Post-Download Processing private var metadataReader: (any MetadataReading)? @@ -96,6 +118,7 @@ public final class DownloadManager { case timeout case incompleteTransfer(expected: UInt64, actual: UInt64) case verificationFailed + case cancelledByUser public var errorDescription: String? { switch self { @@ -107,6 +130,7 @@ public final class DownloadManager { case .incompleteTransfer(let expected, let actual): return "Incomplete transfer: received \(actual) of \(expected) bytes" case .verificationFailed: return "File verification failed" + case .cancelledByUser: return "Cancelled by user" } } } @@ -273,6 +297,47 @@ public final class DownloadManager { return } + // Duplicate detection. Two rows for the same (username, filename) + // mean two pendingDownloads entries and two F-connections appending + // to the same incomplete file — guaranteed corruption. If a live + // attempt already exists, no-op. If a prior attempt finished or was + // abandoned (.completed / .failed / .cancelled), reuse that row and + // re-drive it rather than spawning a parallel one. + let dupePending = pendingDownloads.values.contains { + $0.username == result.username && $0.filename == result.filename + } + if let existing = transferState.downloads.first(where: { + $0.direction == .download && + $0.username == result.username && + $0.filename == result.filename + }) { + switch existing.status { + case .queued, .waiting, .connecting, .transferring: + logger.info("Skipping duplicate download (already \(existing.status.rawValue)): \(result.filename)") + return + case .completed, .failed, .cancelled: + logger.info("Re-queuing previously \(existing.status.rawValue) download: \(result.filename)") + cancelledTransferIds.remove(existing.id) + transferState.updateTransfer(id: existing.id) { t in + t.status = .queued + t.error = nil + t.bytesTransferred = 0 + t.retryCount = 0 + t.nextRetryAt = nil + } + let id = existing.id + let user = existing.username + let file = existing.filename + let size = existing.size + Task { await self.startDownload(transferId: id, username: user, filename: file, size: size) } + return + } + } + if dupePending { + logger.info("Skipping duplicate download (pending in flight): \(result.filename)") + return + } + let transfer = Transfer( username: result.username, filename: result.filename, @@ -330,7 +395,7 @@ public final class DownloadManager { // Cancel any watchdogs tied to entries we're dropping so a // stale 60 s timer can't fire on the new attempt. for stale in entries where stale.transferId == transfer.id { - fileTransferWatchdogs.removeValue(forKey: stale.transferToken)?.cancel() + fileTransferWatchdogs.removeValue(forKey: watchdogKey(username: user, transferToken: stale.transferToken))?.cancel() } if kept.isEmpty { pendingFileTransfersByUser.removeValue(forKey: user) @@ -343,6 +408,10 @@ public final class DownloadManager { // call into us mid-flight; cancel it now since this fresh attempt // is the new source of truth. cancelRetry(transferId: transfer.id) + // An explicit (re)start is the user's intent to revive this row, so + // drop any prior cancellation marker — otherwise the receive loop + // and completion paths would immediately abort the new attempt. + cancelledTransferIds.remove(transfer.id) let token = UInt32.random(in: 0...UInt32.max) @@ -434,6 +503,10 @@ public final class DownloadManager { /// across startDownload/handlePeerAddress/queue paths. private func failPending(token: UInt32, reason: String) { guard let transferState, let pending = pendingDownloads[token] else { return } + if isCancelled(pending.transferId) { + pendingDownloads.removeValue(forKey: token) + return + } let currentRetryCount = transferState.getTransfer(id: pending.transferId)?.retryCount ?? 0 transferState.updateTransfer(id: pending.transferId) { t in t.status = .failed @@ -455,19 +528,19 @@ public final class DownloadManager { Self.matchPendingDownload(request: request, pending: pendingDownloads) } - /// Token → (username, filename). The earlier filename-only fallback - /// could misroute when two different peers happened to be sending the - /// same filename; it was only needed because `request.username` used - /// to arrive empty on reused connections. `handlePoolTransferRequest` - /// now normalizes the request with the connection's authoritative - /// `peerInfo.username` before calling this, so the fallback is gone. + /// Match by (username, filename). We deliberately do NOT try + /// `pending[request.token]` first: `request.token` is the peer's + /// upload ticket, while the `pending` dictionary is keyed by OUR + /// locally-generated random download token — they live in different + /// namespaces, so a hit there would be a coincidental collision, not a + /// real match. The earlier filename-only fallback is also gone: + /// `handlePoolTransferRequest` normalizes the request with the + /// connection's authoritative `peerInfo.username` before calling this. static func matchPendingDownload( request: TransferRequest, pending: [UInt32: PendingDownload] ) -> UInt32? { - if pending[request.token] != nil { - return request.token - } + guard !request.username.isEmpty else { return nil } return pending.first { (_, p) in p.username == request.username && p.filename == request.filename }?.key @@ -483,6 +556,24 @@ public final class DownloadManager { ? TransferRequest(direction: request.direction, token: request.token, filename: request.filename, size: request.size, username: peerUsername) : request + // direction == .download means the peer wants a file FROM us — the + // legacy (pre-QueueUpload) way of requesting an upload. Route it to + // the upload side; previously it was dropped unanswered and the + // requester hung until their timeout. + if normalizedRequest.direction == .download { + guard let uploadManager, !peerUsername.isEmpty else { + logger.info("Legacy download-direction TransferRequest dropped (no upload manager or username): \(request.filename)") + return + } + await uploadManager.handleDownloadTransferRequest( + username: peerUsername, + token: normalizedRequest.token, + filename: normalizedRequest.filename, + connection: connection + ) + return + } + if let token = matchPendingDownload(for: normalizedRequest) { logger.info("Pool TransferRequest matched pending download: user=\(peerUsername) file=\(request.filename)") await handleTransferRequest(token: token, request: normalizedRequest, connection: connection) @@ -549,6 +640,12 @@ public final class DownloadManager { private func handleTransferRequest(token: UInt32, request: TransferRequest, connection: PeerConnection) async { guard let transferState, let pending = pendingDownloads[token] else { return } + if isCancelled(pending.transferId) { + logger.info("Ignoring TransferRequest for cancelled transfer: \(pending.filename)") + pendingDownloads.removeValue(forKey: token) + return + } + let directionStr = request.direction == .upload ? "upload" : "download" logger.info("Transfer request received: direction=\(directionStr) size=\(request.size) from \(request.username)") @@ -583,10 +680,10 @@ public final class DownloadManager { let expectedSize = request.size > 0 ? request.size : pending.size let incompletePath = computeIncompletePath(for: pending.filename, username: pending.username) var resumeOffset: UInt64 = 0 - if FileManager.default.fileExists(atPath: incompletePath.path) { + if expectedSize > 0, FileManager.default.fileExists(atPath: incompletePath.path) { if let attrs = try? FileManager.default.attributesOfItem(atPath: incompletePath.path), let existingSize = attrs[.size] as? UInt64, - existingSize > 0 && (expectedSize == 0 || existingSize < expectedSize) { + existingSize > 0 && existingSize < expectedSize { resumeOffset = existingSize logger.info("Found incomplete file \(incompletePath.lastPathComponent), \(existingSize)/\(expectedSize) bytes, resuming from offset \(resumeOffset)") } @@ -602,6 +699,9 @@ public final class DownloadManager { offset: resumeOffset // Resume from partial file if exists ) pendingFileTransfersByUser[pending.username, default: []].append(pendingTransfer) + // Record the offset this attempt resumes from so a later + // UploadFailed can tell whether a resume was actually tried. + pendingDownloads[token]?.resumeOffset = resumeOffset logger.info("Registered pending file transfer for \(pending.username): transferToken=\(request.token)") // Earliest unambiguous signal the peer is responding. Drop @@ -628,8 +728,9 @@ public final class DownloadManager { // TransferRequest with a previously-seen token shouldn't happen // in practice, but Task.cancel is cheap and keeps invariants // tight). - fileTransferWatchdogs.removeValue(forKey: transferToken)?.cancel() - fileTransferWatchdogs[transferToken] = Task { [weak self] in + let wKey = watchdogKey(username: peerUsername, transferToken: transferToken) + fileTransferWatchdogs.removeValue(forKey: wKey)?.cancel() + fileTransferWatchdogs[wKey] = Task { [weak self] in guard let self else { return } // Wait 5 seconds for peer to connect to us try? await Task.sleep(for: .seconds(5)) @@ -744,6 +845,15 @@ public final class DownloadManager { let receivedToken = tokenData.readUInt32(at: 0) ?? 0 logger.debug("Received FileTransferInit: token=\(receivedToken) (expected=\(transferToken))") + // Validate the uploader's FileTransferInit token. A mismatch means + // this stream isn't the transfer we expect (stale/crossed wires); + // proceeding would append unknown bytes to the file. Bail and let + // the catch route through the normal failure/retry path. + guard receivedToken == transferToken else { + logger.warning("FileTransferInit token mismatch from \(username): got \(receivedToken), expected \(transferToken)") + throw DownloadError.connectionClosed + } + // Send FileOffset (offset - 8 bytes) var offsetData = Data() offsetData.appendUInt64(resumeOffset) @@ -770,6 +880,13 @@ public final class DownloadManager { transferId: pending.transferId, resumeOffset: resumeOffset ) + + if isCancelled(pending.transferId) { + logger.info("Download cancelled after receive; not finalizing \(pending.filename)") + pendingDownloads.removeValue(forKey: downloadToken) + connection.cancel() + return + } // `finalPath` may differ from `desiredFinalPath` if a file was // already at the destination — finalize will pick a non-colliding // suffix so we never silently clobber an existing local copy. @@ -798,11 +915,12 @@ public final class DownloadManager { applyFolderArtworkIfNeeded(for: finalPath) organizeCompletedDownload(currentPath: finalPath, soulseekFilename: pending.filename, username: username, transferId: pending.transferId) - // Record in statistics + // Record only bytes received THIS session against the session + // duration so resumed transfers don't report inflated totals/speed. statisticsState?.recordTransfer( filename: filename, username: username, - size: fileSize, + size: fileSize > resumeOffset ? fileSize - resumeOffset : fileSize, duration: duration, isDownload: true ) @@ -812,6 +930,11 @@ public final class DownloadManager { } catch { logger.error("Outgoing F connection failed: \(error.localizedDescription)") + if isCancelled(pending.transferId) { + _ = removePendingFileTransfer(username: username, transferToken: transferToken) + pendingDownloads.removeValue(forKey: downloadToken) + return + } // Explicit failure handling. Previously this branch deferred to // the 60s watchdog ("Don't mark as failed yet — the timeout // will handle that"), but the watchdog only fires while the @@ -915,7 +1038,7 @@ public final class DownloadManager { // group never returns. Calling connection.cancel() forces the // receive completion handler to fire (with error), the continuation // resumes, the child exits, and the timeout actually times out. - try await withThrowingTaskGroup(of: Data.self) { group in + try await withThrowingTaskGroup(of: Data?.self) { group in group.addTask { try await withTaskCancellationHandler { try await withCheckedThrowingContinuation { (continuation: CheckedContinuation) in @@ -935,15 +1058,24 @@ public final class DownloadManager { } group.addTask { - try await Task.sleep(for: .seconds(timeout)) - throw DownloadError.timeout + try? await Task.sleep(for: .seconds(timeout)) + return nil // timeout sentinel } - guard let result = try await group.next() else { + guard let first = try await group.next() else { throw DownloadError.timeout } + if let data = first { + group.cancelAll() + return data + } + // Timeout fired first; still surface a simultaneously-received + // result so the read is never discarded on the race. group.cancelAll() - return result + if let second = try? await group.next(), let data = second { + return data + } + throw DownloadError.timeout } } @@ -999,6 +1131,12 @@ public final class DownloadManager { // Receive data in chunks while bytesReceived < expectedSize { + if isCancelled(transferId) { + try? await fileIO.synchronize() + await fileIO.close() + connection.cancel() + throw DownloadError.cancelledByUser + } let (chunk, isComplete) = try await receiveChunkWithStatus(connection: connection) if chunk.isEmpty && isComplete { @@ -1013,9 +1151,10 @@ public final class DownloadManager { bytesReceived += UInt64(chunk.count) networkClient?.peerConnectionPool.recordBytesReceived(UInt64(chunk.count)) - // Update progress + // Update progress (speed from bytes received THIS session) let elapsed = Date().timeIntervalSince(startTime) - let speed = elapsed > 0 ? Int64(Double(bytesReceived) / elapsed) : 0 + let sessionBytes = bytesReceived > resumeOffset ? bytesReceived - resumeOffset : 0 + let speed = elapsed > 0 ? Int64(Double(sessionBytes) / elapsed) : 0 transferState?.updateTransfer(id: transferId) { t in t.bytesTransferred = bytesReceived @@ -1037,8 +1176,17 @@ public final class DownloadManager { let actualSize = attrs[.size] as? UInt64 ?? 0 logger.info("File verification: expected=\(expectedSize), received=\(bytesReceived), disk=\(actualSize)") - if expectedSize > 0 && actualSize >= expectedSize { - logger.info("Download complete: received \(actualSize) bytes (expected \(expectedSize))") + if expectedSize == 0 { + logger.info("Zero-byte file complete") + } else if actualSize == expectedSize { + logger.info("Download complete: received \(actualSize) bytes") + } else if actualSize > expectedSize { + // Peer ignored our offset and re-streamed from 0 — partial+full + // got appended. The file is corrupt; delete so the next attempt + // starts clean. + logger.error("Oversize transfer: \(actualSize) > \(expectedSize); deleting corrupt file") + try? FileManager.default.removeItem(at: destPath) + throw DownloadError.incompleteTransfer(expected: expectedSize, actual: actualSize) } else { logger.error("Incomplete transfer: \(actualSize)/\(expectedSize) bytes") throw DownloadError.incompleteTransfer(expected: expectedSize, actual: actualSize) @@ -1048,20 +1196,54 @@ public final class DownloadManager { logger.info("File transfer complete and verified: \(actualSize) bytes received") } - private func receiveChunkWithStatus(connection: NWConnection) async throws -> (Data, Bool) { - try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<(Data, Bool), Error>) in - // Use 1MB buffer for better throughput on file transfers - connection.receive(minimumIncompleteLength: 1, maximumLength: 1024 * 1024) { data, _, isComplete, error in - if let error { - continuation.resume(throwing: error) - } else if let data, !data.isEmpty { - continuation.resume(returning: (data, isComplete)) - } else if isComplete { - continuation.resume(returning: (Data(), true)) - } else { - continuation.resume(returning: (Data(), false)) + private func receiveChunkWithStatus(connection: NWConnection, timeout: TimeInterval = 30) async throws -> (Data, Bool) { + // Race the receive against a 30s stall timeout (same shape as the + // inbound path). Without it, a half-open peer suspends the receive + // continuation forever and the row sticks `.transferring`. On timeout + // we cancel the NWConnection so the receive callback fires and the + // child exits. A `nil` race result is the timeout sentinel; we prefer + // a real data result so a chunk that arrives simultaneously with the + // timeout is never discarded. + try await withThrowingTaskGroup(of: Optional<(Data, Bool)>.self) { group in + group.addTask { + try await withTaskCancellationHandler { + try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<(Data, Bool), Error>) in + connection.receive(minimumIncompleteLength: 1, maximumLength: 1024 * 1024) { data, _, isComplete, error in + if let error { + continuation.resume(throwing: error) + } else if let data, !data.isEmpty { + continuation.resume(returning: (data, isComplete)) + } else if isComplete { + continuation.resume(returning: (Data(), true)) + } else { + continuation.resume(returning: (Data(), false)) + } + } + } + } onCancel: { + connection.cancel() } } + group.addTask { + try? await Task.sleep(for: .seconds(timeout)) + return nil + } + + guard let first = try await group.next() else { + throw DownloadError.timeout + } + if let value = first { + group.cancelAll() + return value + } + // Timeout fired first. Cancel the receive child, but if it had + // already produced data simultaneously, return that instead of + // throwing it away. + group.cancelAll() + if let second = try? await group.next(), let value = second { + return value + } + throw DownloadError.timeout } } @@ -1575,13 +1757,14 @@ public final class DownloadManager { } else { pendingFileTransfersByUser[key] = entries } - fileTransferWatchdogs.removeValue(forKey: transferToken)?.cancel() + fileTransferWatchdogs.removeValue(forKey: watchdogKey(username: key, transferToken: transferToken))?.cancel() return removed } /// Handle F connection when multiple transfers are pending for same user. /// Receives FileTransferInit token first to match the right pending entry. private func handleFileTransferWithTokenMatch(entries: [PendingFileTransfer], username: String, connection: PeerConnection) async { + var matched: PendingFileTransfer? do { await connection.stopReceiving() try await Task.sleep(for: .milliseconds(50)) @@ -1605,54 +1788,79 @@ public final class DownloadManager { logger.info("F connection: received FileTransferInit token=\(receivedToken), matching against \(entries.count) pending entries") // Match by token - if let pending = removePendingFileTransfer(username: username, transferToken: receivedToken) { - // Send FileOffset and proceed - var offsetData = Data() - offsetData.appendUInt64(pending.offset) - try await connection.sendRaw(offsetData) - - let desiredFinalPath = computeDestPath(for: pending.filename, username: pending.username) - let incompletePath = computeIncompletePath(for: pending.filename, username: pending.username) - try await receiveFileDataFromPeer( - connection: connection, - destPath: incompletePath, - expectedSize: pending.size, - transferId: pending.transferId, - resumeOffset: pending.offset - ) - let finalPath = try finalizeCompletedDownload(from: incompletePath, to: desiredFinalPath) - - let duration = Date().timeIntervalSince(transferState?.getTransfer(id: pending.transferId)?.startTime ?? Date()) - cancelRetry(transferId: pending.transferId) - // Drop the corresponding pendingDownloads entry too — - // otherwise a late peer message (UploadFailed / - // UploadDenied) finds the stale entry by filename and - // tries to re-queue an already-finished transfer. The - // other completion paths already do this; the - // incoming-F path was the missing one. - pendingDownloads.removeValue(forKey: pending.downloadToken) - transferState?.updateTransfer(id: pending.transferId) { t in - t.status = .completed - t.bytesTransferred = pending.size - t.localPath = finalPath - t.error = nil - } - ActivityLogger.shared?.logDownloadCompleted(filename: finalPath.lastPathComponent) - applyFolderArtworkIfNeeded(for: finalPath) - organizeCompletedDownload(currentPath: finalPath, soulseekFilename: pending.filename, username: pending.username, transferId: pending.transferId) - statisticsState?.recordTransfer( - filename: finalPath.lastPathComponent, - username: pending.username, - size: pending.size, - duration: duration, - isDownload: true - ) - } else { + guard let pending = removePendingFileTransfer(username: username, transferToken: receivedToken) else { logger.warning("F connection token \(receivedToken) didn't match any pending transfer for \(username); closing stale connection") await connection.disconnect() + return + } + matched = pending + + if isCancelled(pending.transferId) { + logger.info("F connection for cancelled transfer \(pending.filename); closing") + pendingDownloads.removeValue(forKey: pending.downloadToken) + await connection.disconnect() + return } + + // Send FileOffset and proceed + var offsetData = Data() + offsetData.appendUInt64(pending.offset) + try await connection.sendRaw(offsetData) + + let desiredFinalPath = computeDestPath(for: pending.filename, username: pending.username) + let incompletePath = computeIncompletePath(for: pending.filename, username: pending.username) + try await receiveFileDataFromPeer( + connection: connection, + destPath: incompletePath, + expectedSize: pending.size, + transferId: pending.transferId, + resumeOffset: pending.offset + ) + + if isCancelled(pending.transferId) { + logger.info("Download cancelled after receive; not finalizing \(pending.filename)") + pendingDownloads.removeValue(forKey: pending.downloadToken) + await connection.disconnect() + return + } + + let finalPath = try finalizeCompletedDownload(from: incompletePath, to: desiredFinalPath) + + let duration = Date().timeIntervalSince(transferState?.getTransfer(id: pending.transferId)?.startTime ?? Date()) + cancelRetry(transferId: pending.transferId) + // Drop the pendingDownloads entry so a late UploadFailed/Denied + // can't re-queue an already-finished transfer. + pendingDownloads.removeValue(forKey: pending.downloadToken) + transferState?.updateTransfer(id: pending.transferId) { t in + t.status = .completed + t.bytesTransferred = pending.size + t.localPath = finalPath + t.error = nil + } + ActivityLogger.shared?.logDownloadCompleted(filename: finalPath.lastPathComponent) + applyFolderArtworkIfNeeded(for: finalPath) + organizeCompletedDownload(currentPath: finalPath, soulseekFilename: pending.filename, username: pending.username, transferId: pending.transferId) + // Record only this session's bytes (resumed portion was on disk). + statisticsState?.recordTransfer( + filename: finalPath.lastPathComponent, + username: pending.username, + size: pending.size > pending.offset ? pending.size - pending.offset : pending.size, + duration: duration, + isDownload: true + ) } catch { logger.error("Failed token-match F connection: \(error.localizedDescription)") + await connection.disconnect() + guard let matched else { return } + pendingDownloads.removeValue(forKey: matched.downloadToken) + if isCancelled(matched.transferId) { return } + failDownload( + transferId: matched.transferId, + username: matched.username, + filename: matched.filename, + size: matched.size, + reason: "F connection failed: \(error.localizedDescription)" + ) } } @@ -1739,6 +1947,12 @@ public final class DownloadManager { // Nicotine+ approach: receive until connection ACTUALLY closes, then verify byte count // Don't use artificial timeouts that could cut off slow transfers receiveLoop: while true { + if isCancelled(transferId) { + try? await fileIO.synchronize() + await fileIO.close() + await connection.disconnect() + throw DownloadError.cancelledByUser + } // Receive data - no artificial timeout that returns fake completion let chunkResult: PeerConnection.FileChunkResult do { @@ -1752,7 +1966,7 @@ public final class DownloadManager { // callback inside `receiveFileChunk` never fires, the child // task's continuation stays pending, and the task group waits // on the orphan forever — defeating the timeout entirely. - chunkResult = try await withThrowingTaskGroup(of: PeerConnection.FileChunkResult.self) { group in + chunkResult = try await withThrowingTaskGroup(of: PeerConnection.FileChunkResult?.self) { group in group.addTask { try await withTaskCancellationHandler { try await connection.receiveFileChunk() @@ -1765,14 +1979,24 @@ public final class DownloadManager { } } group.addTask { - try await Task.sleep(for: .seconds(30)) - throw DownloadError.timeout + try? await Task.sleep(for: .seconds(30)) + return nil // timeout sentinel } - guard let result = try await group.next() else { + guard let first = try await group.next() else { throw DownloadError.timeout } + if let chunk = first { + group.cancelAll() + return chunk + } + // Timeout fired first. Cancel the receive child but still + // surface any chunk it produced simultaneously, so a + // received chunk is never discarded on a timeout race. group.cancelAll() - return result + if let second = try? await group.next(), let chunk = second { + return chunk + } + throw DownloadError.timeout } } catch is DownloadError { // Timeout - but try to drain any remaining buffered data first @@ -1816,9 +2040,10 @@ public final class DownloadManager { networkClient?.peerConnectionPool.recordBytesReceived(UInt64(chunk.count)) lastDataTime = Date() // Reset timeout tracker - // Update progress periodically (not every chunk to reduce UI overhead) + // Update progress (speed from bytes received THIS session) let elapsed = Date().timeIntervalSince(startTime) - let speed = elapsed > 0 ? Int64(Double(bytesReceived) / elapsed) : 0 + let sessionBytes = bytesReceived > resumeOffset ? bytesReceived - resumeOffset : 0 + let speed = elapsed > 0 ? Int64(Double(sessionBytes) / elapsed) : 0 transferState?.updateTransfer(id: transferId) { t in t.bytesTransferred = bytesReceived @@ -1905,9 +2130,16 @@ public final class DownloadManager { let percentComplete = expectedSize > 0 ? Double(actualSize) / Double(expectedSize) * 100 : 100 logger.info("Verify: expected=\(expectedSize), received=\(bytesReceived), disk=\(actualSize) (\(String(format: "%.1f", percentComplete))%)") - // Like nicotine+: require actualSize >= expectedSize - if expectedSize > 0 && actualSize >= expectedSize { - logger.info("Download complete: received \(actualSize) bytes (expected \(expectedSize))") + if expectedSize == 0 { + logger.info("Zero-byte file complete") + } else if actualSize == expectedSize { + logger.info("Download complete: received \(actualSize) bytes") + } else if actualSize > expectedSize { + // Peer ignored our offset and re-streamed from 0 — partial+full + // got appended. Corrupt; delete so the next attempt starts clean. + logger.error("Oversize transfer: \(actualSize) > \(expectedSize); deleting corrupt file") + try? FileManager.default.removeItem(at: destPath) + throw DownloadError.incompleteTransfer(expected: expectedSize, actual: actualSize) } else { logger.error("Incomplete transfer: \(actualSize)/\(expectedSize) bytes (\(String(format: "%.1f", percentComplete))%)") throw DownloadError.incompleteTransfer(expected: expectedSize, actual: actualSize) @@ -2310,38 +2542,17 @@ public final class DownloadManager { } } - // Check if we attempted a resume - if so, delete partial and retry from scratch + // If THIS attempt resumed from a partial (offset > 0) and the peer + // rejected it, the peer likely can't resume — drop the partial so the + // retry starts clean. Otherwise keep the partial as groundwork. Either + // way the retry is routed through the normal failDownload machinery so + // it's counted, capped at maxRetries, and cancellable via cancelRetry. let incompletePath = computeIncompletePath(for: pending.filename, username: pending.username) - if FileManager.default.fileExists(atPath: incompletePath.path) { - if let attrs = try? FileManager.default.attributesOfItem(atPath: incompletePath.path), - let existingSize = attrs[.size] as? UInt64, - existingSize > 0 { - // We had a partial file - the peer might not support resume - // Delete partial and retry from scratch - logger.warning("Upload failed after resume attempt - deleting partial file and retrying from scratch") - try? FileManager.default.removeItem(at: incompletePath) - - // Mark for retry with status .queued - transferState?.updateTransfer(id: pending.transferId) { t in - t.status = .queued - t.bytesTransferred = 0 - t.error = nil - } - - // Schedule automatic retry - let transferId = pending.transferId - let username = pending.username - let filenameCopy = pending.filename - let size = pending.size - - pendingDownloads.removeValue(forKey: token) - - Task { - try? await Task.sleep(for: .seconds(2)) - self.logger.info("Retrying download from scratch: \(filenameCopy)") - await self.startDownload(transferId: transferId, username: username, filename: filenameCopy, size: size) - } - return + if pending.resumeOffset > 0, FileManager.default.fileExists(atPath: incompletePath.path) { + logger.warning("Upload failed after resume attempt - deleting partial to retry from scratch") + try? FileManager.default.removeItem(at: incompletePath) + transferState?.updateTransfer(id: pending.transferId) { t in + t.bytesTransferred = 0 } } @@ -2381,6 +2592,13 @@ public final class DownloadManager { reason: String, retryCount explicitRetryCount: Int? = nil ) { + // A cancelled transfer is terminal: don't overwrite .cancelled with + // .failed and don't schedule a retry. + if isCancelled(transferId) { + logger.info("Not failing cancelled transfer \(transferId): \(reason)") + return + } + let currentRetryCount = explicitRetryCount ?? transferState?.getTransfer(id: transferId)?.retryCount ?? 0 @@ -2421,11 +2639,14 @@ public final class DownloadManager { return false } - // Known terminal reasons — user action or peer-side decisions - // that re-asking won't change. `cancel` (bare stem) matches both - // "cancelled" and "canceled" spellings. Mirror UploadManager. + // Known terminal reasons — peer-side decisions that re-asking won't + // change. The bare "cancel" stem was removed: it also matched the + // client's own transient teardown strings ("Connection was + // cancelled", "Operation canceled"), wrongly marking them terminal. + // User-initiated cancels no longer flow through failDownload at all + // (see cancelDownload + the cancelled-transfer guards), so we only + // need genuine peer-denial reasons here. let terminalPatterns = [ - "cancel", "denied", "not shared", "not available", @@ -2523,6 +2744,8 @@ public final class DownloadManager { /// calling this; `retryFailedDownload` checks `.failed || .cancelled`. private func retryDownload(transferId: UUID, username: String, filename: String, size: UInt64, retryCount: Int) { logger.info("Retrying download: \(filename) (attempt \(retryCount))") + // Manual/automatic retry revives the row; drop any cancellation marker. + cancelledTransferIds.remove(transferId) // Update the existing transfer record. Reset to .queued so // `startDownload` (which sets it to .connecting) sees a clean slate. @@ -2592,6 +2815,44 @@ public final class DownloadManager { } } + /// Cancel an in-flight or queued download. Marks the transfer cancelled so + /// the receive loops abort at their next chunk check, completion paths + /// refuse to finalize it, the salvage/TransferRequest path won't resurrect + /// it, and failDownload treats it as terminal (no retry). Any partial file + /// is kept on disk so a later manual retry can resume from it. Idempotent. + public func cancelDownload(transferId: UUID) { + cancelledTransferIds.insert(transferId) + cancelRetry(transferId: transferId) + + // Drop our pending bookkeeping for this transfer. + let tokens = pendingDownloads.compactMap { $0.value.transferId == transferId ? $0.key : nil } + for token in tokens { + pendingDownloads.removeValue(forKey: token) + } + for (user, entries) in pendingFileTransfersByUser { + let toRemove = entries.filter { $0.transferId == transferId } + guard !toRemove.isEmpty else { continue } + for entry in toRemove { + fileTransferWatchdogs.removeValue(forKey: watchdogKey(username: user, transferToken: entry.transferToken))?.cancel() + } + let kept = entries.filter { $0.transferId != transferId } + if kept.isEmpty { + pendingFileTransfersByUser.removeValue(forKey: user) + } else { + pendingFileTransfersByUser[user] = kept + } + } + + transferState?.updateTransfer(id: transferId) { t in + t.status = .cancelled + t.error = nil + t.speed = 0 + t.queuePosition = nil + t.nextRetryAt = nil + } + logger.info("Cancelled download \(transferId)") + } + /// Rearm in-memory retry timers for any persisted `.failed` rows that /// were mid-backoff when the app last quit. Without this, a row that /// was scheduled to retry in 28 minutes but interrupted by a quit diff --git a/Packages/SeeleseekCore/Sources/SeeleseekCore/Storage/ShareManager.swift b/Packages/SeeleseekCore/Sources/SeeleseekCore/Storage/ShareManager.swift index ab192fe..00d8239 100644 --- a/Packages/SeeleseekCore/Sources/SeeleseekCore/Storage/ShareManager.swift +++ b/Packages/SeeleseekCore/Sources/SeeleseekCore/Storage/ShareManager.swift @@ -249,13 +249,25 @@ public final class ShareManager { isScanning = true scanProgress = 0 - fileIndex.removeAll() + // Build the new index aside and swap at the end. Clearing + // `fileIndex` up front left a minutes-wide window where every + // peer lookup missed and got a terminal "File not shared." + var newIndex: [IndexedFile] = [] for (index, folder) in sharedFolders.enumerated() { - await scanFolder(folder) + if let result = await scanFolderResult(folder) { + newIndex.append(contentsOf: result.indexed) + applyFolderStats(result) + logger.info("Scanned \(folder.displayName): \(result.fileCount) files") + } else { + logger.error("Failed to enumerate folder: \(folder.path)") + } scanProgress = Double(index + 1) / Double(sharedFolders.count) } + // Atomic swap — old index served lookups during the scan. + fileIndex = newIndex + lastScanDate = Date() isScanning = false // Only persist if the for-loop actually ran. With an empty @@ -286,7 +298,46 @@ public final class ShareManager { let totalSize: UInt64 } + /// Single-folder scan that mutates state directly (addFolder path). + /// `rescanAll` uses `scanFolderResult` + a deferred index swap instead. private func scanFolder(_ folder: SharedFolder) async { + guard let result = await scanFolderResult(folder) else { + logger.error("Failed to enumerate folder: \(folder.path)") + return + } + fileIndex.append(contentsOf: result.indexed) + applyFolderStats(result) + logger.info("Scanned \(folder.displayName): \(result.fileCount) files") + } + + /// Update the per-folder counters from a completed scan. + private func applyFolderStats(_ result: ScanResult) { + if let index = sharedFolders.firstIndex(where: { $0.id == result.folderID }) { + sharedFolders[index].fileCount = result.fileCount + sharedFolders[index].totalSize = result.totalSize + sharedFolders[index].lastScanned = Date() + } + } + + /// Disambiguate duplicate share-root display names so sharedPaths + /// stay unique across roots (e.g. two folders both named "Music" + /// become "Music" and "Music (2)"). + private func uniqueDisplayName(for folder: SharedFolder) -> String { + let base = folder.displayName + let sameName = sharedFolders.filter { $0.displayName == base } + guard sameName.count > 1, + let position = sameName.firstIndex(where: { $0.id == folder.id }), + position > 0 else { + return base + } + let unique = "\(base) (\(position + 1))" + logger.warning("Share root name collision for \(base) — using \(unique)") + return unique + } + + /// Walk a folder on a background task and return the indexed files + /// plus stats, without touching published state. + private func scanFolderResult(_ folder: SharedFolder) async -> ScanResult? { let folderURL = URL(fileURLWithPath: folder.path) // Restore bookmark access on the main actor before handing the URL @@ -309,9 +360,10 @@ public final class ShareManager { // references escape. let folderID = folder.id let folderVisibility = folder.visibility - let folderDisplayName = folder.displayName + // Suffix duplicate root names so sharedPaths are unique. + let folderDisplayName = uniqueDisplayName(for: folder) - let result: ScanResult? = await Task.detached(priority: .utility) { + return await Task.detached(priority: .utility) { let fileManager = FileManager.default guard let enumerator = fileManager.enumerator( at: folderURL, @@ -353,22 +405,6 @@ public final class ShareManager { return ScanResult(folderID: folderID, indexed: files, fileCount: count, totalSize: total) }.value - - guard let result else { - logger.error("Failed to enumerate folder: \(folder.path)") - return - } - - // Single publish step back on main: append the scanned batch and - // update folder stats in one @MainActor hop. - fileIndex.append(contentsOf: result.indexed) - if let index = sharedFolders.firstIndex(where: { $0.id == result.folderID }) { - sharedFolders[index].fileCount = result.fileCount - sharedFolders[index].totalSize = result.totalSize - sharedFolders[index].lastScanned = Date() - } - - logger.info("Scanned \(folder.displayName): \(result.fileCount) files") } // Static so the detached scan task can call it without a main-actor hop. diff --git a/Packages/SeeleseekCore/Sources/SeeleseekCore/Storage/UploadManager.swift b/Packages/SeeleseekCore/Sources/SeeleseekCore/Storage/UploadManager.swift index e8aa182..6ac917d 100644 --- a/Packages/SeeleseekCore/Sources/SeeleseekCore/Storage/UploadManager.swift +++ b/Packages/SeeleseekCore/Sources/SeeleseekCore/Storage/UploadManager.swift @@ -72,6 +72,17 @@ public final class UploadManager { /// makes the no-op safe; this just stops the wasted sleep. private var pendingRetries: [UUID: Task] = [:] + /// Transfer ids cancelled mid-stream. Checked per chunk by both + /// streaming loops; cleared on stream exit or re-drive. + private var cancelledTransferIds: Set = [] + /// Per-transfer teardown that drops the live connection so a wedged + /// send unblocks. Registered when a stream starts, removed on exit. + private var uploadTeardowns: [UUID: @Sendable () -> Void] = [:] + /// Reentrancy guard for `processQueue` — startUpload's failure paths + /// call back into processQueue, which could double-start queue items. + private var isProcessingQueue = false + private var needsQueuePass = false + /// Called to check if an upload should be allowed (checks blocklist + leech status) /// Set by AppState to delegate to SocialState public var uploadPermissionChecker: ((String) -> Bool)? @@ -227,6 +238,15 @@ public final class UploadManager { let position = getQueuePosition(for: filename, username: username) if position == 0 { + // Pending or active for this user+file: report front of queue. + if isInFlight(username: username, filename: filename) { + do { + try await connection.sendPlaceInQueue(filename: filename, place: 1) + } catch { + logger.debug("Failed to send PlaceInQueue: \(error.localizedDescription)") + } + return + } // Not in queue - maybe file doesn't exist or isn't shared logger.debug("File not in queue: \(filename)") // Could send UploadDenied here if file doesn't exist @@ -251,21 +271,48 @@ public final class UploadManager { } } + /// Distinct in-flight transfer ids. A transfer briefly appears in both + /// `pendingTransfers` and `activeUploads` (PierceFirewall window), so + /// summing the dict counts double-counts and starves slots. + private var inFlightTransferCount: Int { + var ids = Set(activeUploads.keys) + for pending in pendingTransfers.values { + ids.insert(pending.transferId) + } + return ids.count + } + + /// True if a transfer for this user+file is pending or actively streaming. + private func isInFlight(username: String, filename: String) -> Bool { + pendingTransfers.values.contains { $0.username == username && $0.filename == filename } + || activeUploads.values.contains { $0.username == username && $0.filename == filename } + } + /// Process the upload queue - start uploads if slots available private func processQueue() async { - let inFlightCount = activeUploads.count + pendingTransfers.count - guard inFlightCount < maxConcurrentUploads else { - // Still broadcast updated positions to queued peers - await broadcastQueuePositions() + // Reentrancy: a recursive call (startUpload failure path) just + // requests another pass instead of double-starting queue items. + if isProcessingQueue { + needsQueuePass = true return } - guard !uploadQueue.isEmpty else { return } + isProcessingQueue = true + defer { isProcessingQueue = false } - let availableSlots = maxConcurrentUploads - inFlightCount - let uploadsToStart = uploadQueue.prefix(availableSlots) + needsQueuePass = true + while needsQueuePass { + needsQueuePass = false + let inFlightCount = inFlightTransferCount + guard inFlightCount < maxConcurrentUploads, !uploadQueue.isEmpty else { break } - for upload in uploadsToStart { - await startUpload(upload) + let availableSlots = maxConcurrentUploads - inFlightCount + let uploadsToStart = Array(uploadQueue.prefix(availableSlots)) + + for upload in uploadsToStart { + // Re-validate: a nested pass may have taken it already. + guard uploadQueue.contains(where: { $0.id == upload.id }) else { continue } + await startUpload(upload) + } } // Broadcast updated positions to remaining queued peers @@ -296,30 +343,26 @@ public final class UploadManager { // MARK: - Upload Flow - /// Handle incoming QueueUpload request from a peer - private func handleQueueUpload(username: String, filename: String, connection: PeerConnection) async { - logger.info("QueueUpload from \(username): \(filename)") + /// Outcome of share-index + policy validation for an upload request. + private enum UploadRequestValidation { + case allowed(ShareManager.IndexedFile) + case denied(reason: String) + } + /// Shared validation for QueueUpload and legacy TransferRequest + /// (direction=download). Checks share index, buddy visibility, local + /// file existence, permission checker, and per-user queue limit. + private func validateUploadRequest(username: String, filename: String) -> UploadRequestValidation { guard let shareManager else { logger.error("ShareManager not configured") - do { - try await connection.sendUploadDenied(filename: filename, reason: "Server error") - } catch { - logger.error("Failed to send UploadDenied: \(error.localizedDescription)") - } - return + return .denied(reason: "Server error") } // Look up the file in our shares // The filename from SoulSeek uses backslashes as path separators guard let indexedFile = shareManager.fileIndex.first(where: { $0.sharedPath == filename }) else { logger.warning("File not found in shares: \(filename)") - do { - try await connection.sendUploadDenied(filename: filename, reason: "File not shared.") - } catch { - logger.error("Failed to send UploadDenied: \(error.localizedDescription)") - } - return + return .denied(reason: "File not shared.") } // Visibility gate. Buddy-only files must not be served to @@ -333,29 +376,19 @@ public final class UploadManager { if indexedFile.visibility == .buddies { let isBuddy = networkClient?.isBuddyChecker?(username) ?? false if !isBuddy { - logger.info("QueueUpload denied (buddy-only file, non-buddy requester): \(username) \(filename)") + logger.info("Upload denied (buddy-only file, non-buddy requester): \(username) \(filename)") ActivityLogger.shared?.logInfo( "Denied upload of buddy-only file to \(username)", detail: filename ) - do { - try await connection.sendUploadDenied(filename: filename, reason: "File not shared.") - } catch { - logger.error("Failed to send UploadDenied: \(error.localizedDescription)") - } - return + return .denied(reason: "File not shared.") } } // Check if file exists locally guard FileManager.default.fileExists(atPath: indexedFile.localPath) else { logger.warning("Local file missing: \(indexedFile.localPath)") - do { - try await connection.sendUploadDenied(filename: filename, reason: "File not shared.") - } catch { - logger.error("Failed to send UploadDenied: \(error.localizedDescription)") - } - return + return .denied(reason: "File not shared.") } // Check if upload is allowed (blocklist + leech detection) @@ -365,24 +398,34 @@ public final class UploadManager { "Denied upload request from \(username)", detail: filename ) - do { - try await connection.sendUploadDenied(filename: filename, reason: "File not shared.") - } catch { - logger.error("Failed to send UploadDenied: \(error.localizedDescription)") - } - return + return .denied(reason: "File not shared.") } // Check per-user queue limit (like nicotine+) let userQueueCount = uploadQueue.filter { $0.username == username }.count if userQueueCount >= maxQueuedPerUser { logger.warning("User \(username) has too many queued uploads (\(userQueueCount))") + return .denied(reason: "Too many files") + } + + return .allowed(indexedFile) + } + + /// Handle incoming QueueUpload request from a peer + private func handleQueueUpload(username: String, filename: String, connection: PeerConnection) async { + logger.info("QueueUpload from \(username): \(filename)") + + let indexedFile: ShareManager.IndexedFile + switch validateUploadRequest(username: username, filename: filename) { + case .denied(let reason): do { - try await connection.sendUploadDenied(filename: filename, reason: "Too many files") + try await connection.sendUploadDenied(filename: filename, reason: reason) } catch { logger.error("Failed to send UploadDenied: \(error.localizedDescription)") } return + case .allowed(let file): + indexedFile = file } // Check for duplicate (same user + same file) @@ -397,6 +440,13 @@ public final class UploadManager { return } + // Re-sent QueueUpload for a transfer already pending/active: + // acknowledge benignly, never create a duplicate row/token. + if isInFlight(username: username, filename: filename) { + logger.debug("QueueUpload ignored, transfer already in flight: \(filename)") + return + } + // Add to queue let queued = QueuedUpload( username: username, @@ -410,8 +460,7 @@ public final class UploadManager { logger.info("Added to upload queue: \(filename) for \(username), position: \(self.uploadQueue.count)") // If we have free slots, start immediately, otherwise send queue position - let inFlightCount = activeUploads.count + pendingTransfers.count - if inFlightCount < maxConcurrentUploads { + if inFlightTransferCount < maxConcurrentUploads { await startUpload(queued) } else { // Send queue position @@ -425,11 +474,77 @@ public final class UploadManager { } } + /// Handle a legacy TransferRequest with direction=download — older + /// clients request files this way instead of QueueUpload. Per + /// nicotine+ behavior: validate, reply TransferReply(allowed=false, + /// reason="Queued") on the peer's token, then enqueue normally. + public func handleDownloadTransferRequest(username: String, token: UInt32, filename: String, connection: PeerConnection) async { + logger.info("Legacy TransferRequest(download) from \(username): \(filename), token=\(token)") + + let indexedFile: ShareManager.IndexedFile + switch validateUploadRequest(username: username, filename: filename) { + case .denied(let reason): + do { + try await connection.sendTransferReply(token: token, allowed: false, reason: reason) + } catch { + logger.error("Failed to send TransferReply: \(error.localizedDescription)") + } + return + case .allowed(let file): + indexedFile = file + } + + // Ack on the peer's token; our own upload flow follows with a + // fresh TransferRequest when a slot frees up. + do { + try await connection.sendTransferReply(token: token, allowed: false, reason: "Queued") + } catch { + logger.error("Failed to send TransferReply: \(error.localizedDescription)") + } + + // Dedup against queue and in-flight transfers. + if uploadQueue.contains(where: { $0.username == username && $0.filename == filename }) + || isInFlight(username: username, filename: filename) { + logger.debug("TransferRequest(download) ignored, already queued/in flight: \(filename)") + return + } + + let queued = QueuedUpload( + username: username, + filename: filename, + localPath: indexedFile.localPath, + size: indexedFile.size, + queuedAt: Date() + ) + uploadQueue.append(queued) + await processQueue() + } + /// Start an upload - send TransferRequest to peer private func startUpload(_ upload: QueuedUpload) async { // Remove from queue uploadQueue.removeAll { $0.id == upload.id } + // Re-check permission — blocklist/leech status may have changed + // while the item sat in the queue. + if let checker = uploadPermissionChecker, !checker(upload.username) { + logger.info("Upload denied at start for \(upload.username): blocked or leech") + await sendUploadDeniedToPeer(username: upload.username, filename: upload.filename, reason: "File not shared.") + if let existing = upload.existingTransferId { + cancelRetry(transferId: existing) + transferState?.updateTransfer(id: existing) { t in + t.status = .failed + t.error = "Denied" + } + } + return + } + + // This transferId is being re-driven; drop any stale cancel flag. + if let existing = upload.existingTransferId { + cancelledTransferIds.remove(existing) + } + let token = UInt32.random(in: 0...UInt32.max) // Reuse the existing transfer record on retries; otherwise create @@ -501,6 +616,8 @@ public final class UploadManager { self.transferResponseTimeouts.removeValue(forKey: token) if let pending = self.pendingTransfers.removeValue(forKey: token) { self.failUpload(transferId: pending.transferId, error: "Timeout waiting for peer response") + // Refill the freed slot. + await self.processQueue() } } } catch { @@ -731,7 +848,14 @@ public final class UploadManager { // Also cancel the per-token PierceFirewall timeout if one was armed // by an earlier direct-failure pass; otherwise it would wake 30s // later and `failUpload` the row we just promoted to active. - pendingTransfers.removeValue(forKey: token) + // If PierceFirewall already consumed the pending entry, the live + // transfer is streaming on that connection — drop this duplicate + // direct connection without touching transfer state. + guard pendingTransfers.removeValue(forKey: token) != nil else { + logger.info("Late direct F connection for token=\(token) — PierceFirewall path owns the transfer, dropping") + connection.cancel() + return + } pierceFirewallTimeouts.removeValue(forKey: token)?.cancel() // Send PeerInit with type "F" and token 0 (always 0 for F connections per protocol) @@ -835,10 +959,21 @@ public final class UploadManager { let startTime = Date() let chunkSize = 65536 // 64KB chunks + // Teardown lets cancelUpload unblock a wedged send. + uploadTeardowns[transferId] = { connection.cancel() } + logger.info("Sending file data: \(filePath) from offset \(offset)") do { while bytesSent < totalSize { + // User cancelled — stop; cancelUpload already updated state. + if cancelledTransferIds.contains(transferId) { + cancelledTransferIds.remove(transferId) + uploadTeardowns.removeValue(forKey: transferId) + logger.info("Upload cancelled mid-stream: \(filePath)") + return + } + // Read chunk off MainActor guard let chunk = try await fileIO.read(upTo: chunkSize), !chunk.isEmpty else { break @@ -865,13 +1000,21 @@ public final class UploadManager { } } - // Respect speed limit if set - if let limit = uploadSpeedLimit, speed > limit { + // Respect speed limit if set (limit > 0 guards div-by-zero) + if let limit = uploadSpeedLimit, limit > 0, speed > limit { let delay = Double(chunk.count) / Double(limit) try? await Task.sleep(for: .milliseconds(Int(delay * 1000))) } } + // Re-check cancel before completing — don't record a + // cancelled transfer as completed. + if cancelledTransferIds.remove(transferId) != nil { + uploadTeardowns.removeValue(forKey: transferId) + logger.info("Upload cancelled at completion: \(filePath)") + return + } + // Short read (file changed under us, or `read(upTo:)` ended // early) — surface as failure rather than racing the peer's // UploadFailed with a bogus `.completed`. @@ -923,16 +1066,18 @@ public final class UploadManager { t.bytesTransferred = bytesSent } - // Record in statistics + // Record session delta only (matches PF path), not the + // resumed offset we never sent. statisticsState?.recordTransfer( filename: filename, username: uploadUsername, - size: bytesSent, + size: bytesSent - offset, duration: duration, isDownload: false ) } + uploadTeardowns.removeValue(forKey: transferId) activeUploads.removeValue(forKey: transferId) ActivityLogger.shared?.logUploadCompleted(filename: filename) @@ -942,13 +1087,18 @@ public final class UploadManager { } catch { logger.error("Upload failed: \(error.localizedDescription)") - await MainActor.run { [self] in - self.failUpload(transferId: transferId, error: error.localizedDescription) - } + let wasCancelled = cancelledTransferIds.remove(transferId) != nil + uploadTeardowns.removeValue(forKey: transferId) + + if !wasCancelled { + await MainActor.run { [self] in + self.failUpload(transferId: transferId, error: error.localizedDescription) + } - // Notify peer so they can re-queue - if let active = activeUploads[transferId] { - await sendUploadFailedToPeer(username: active.username, filename: active.filename) + // Notify peer so they can re-queue + if let active = activeUploads[transferId] { + await sendUploadFailedToPeer(username: active.username, filename: active.filename) + } } activeUploads.removeValue(forKey: transferId) @@ -1110,27 +1260,81 @@ public final class UploadManager { /// Summary string for upload slots (e.g. "2/3") public var slotsSummary: String { "\(activeUploads.count)/\(maxConcurrentUploads)" } - /// Cancel a queued upload - public func cancelQueuedUpload(_ id: UUID) { - uploadQueue.removeAll { $0.id == id } + /// Set upload speed limit in KB/s. <= 0 means unlimited. + public func setUploadSpeedLimit(kbPerSecond: Int) { + uploadSpeedLimit = kbPerSecond > 0 ? Int64(kbPerSecond) * 1024 : nil + logger.info("Upload speed limit set to \(kbPerSecond) KB/s") } - /// Cancel an active upload - public func cancelActiveUpload(_ transferId: UUID) async { - if let upload = activeUploads.removeValue(forKey: transferId) { - // "Cancelled" is a terminal reason in the retry classifier, so - // routing through `failUpload` correctly suppresses any retry. + /// Cancel a queued upload by queue-entry id. Notifies the peer and + /// resolves the transfer row (retry-driven entries carry one) so it + /// doesn't strand at `.queued`. + public func cancelQueuedUpload(_ id: UUID) async { + guard let queued = uploadQueue.first(where: { $0.id == id }) else { return } + uploadQueue.removeAll { $0.id == id } + await sendUploadDeniedToPeer(username: queued.username, filename: queued.filename, reason: "Cancelled") + if let transferId = queued.existingTransferId { cancelRetry(transferId: transferId) - // If we were waiting on the peer's PierceFirewall, drop the - // pending entry and cancel the 30 s watchdog so it can't fire - // later and overwrite the `.cancelled` status with `.failed`. - if let pendingToken = pendingTransfers.first(where: { $0.value.transferId == transferId })?.key { - pendingTransfers.removeValue(forKey: pendingToken) - pierceFirewallTimeouts.removeValue(forKey: pendingToken)?.cancel() + transferState?.updateTransfer(id: transferId) { t in + t.status = .cancelled + t.error = "Cancelled" } - failUpload(transferId: transferId, error: "Cancelled") - logger.info("Cancelled upload: \(upload.filename)") } + logger.info("Cancelled queued upload: \(queued.filename)") + } + + /// Cancel an upload wherever it is in the pipeline: queued, pending + /// (TransferRequest sent / awaiting PierceFirewall), or streaming. + public func cancelUpload(transferId: UUID) async { + cancelRetry(transferId: transferId) + + // Queued (retry-driven entries carry the transferId) + if let queued = uploadQueue.first(where: { $0.existingTransferId == transferId }) { + uploadQueue.removeAll { $0.id == queued.id } + await sendUploadDeniedToPeer(username: queued.username, filename: queued.filename, reason: "Cancelled") + transferState?.updateTransfer(id: transferId) { t in + t.status = .cancelled + t.error = "Cancelled" + } + logger.info("Cancelled queued upload: \(queued.filename)") + return + } + + // Pending — drop the entry and its watchdog timers. + if let token = pendingTransfers.first(where: { $0.value.transferId == transferId })?.key { + let pending = pendingTransfers.removeValue(forKey: token) + transferResponseTimeouts.removeValue(forKey: token)?.cancel() + pierceFirewallTimeouts.removeValue(forKey: token)?.cancel() + transferState?.updateTransfer(id: transferId) { t in + t.status = .cancelled + t.error = "Cancelled" + } + // A transfer can be pending AND active (PierceFirewall window); + // fall through to the active check below. + if activeUploads[transferId] == nil { + logger.info("Cancelled pending upload: \(pending?.filename ?? "?")") + await processQueue() + return + } + } + + // Active — flag for the streaming loop and drop the connection. + if let active = activeUploads.removeValue(forKey: transferId) { + cancelledTransferIds.insert(transferId) + uploadTeardowns.removeValue(forKey: transferId)?() + transferState?.updateTransfer(id: transferId) { t in + t.status = .cancelled + t.error = "Cancelled" + } + logger.info("Cancelled active upload: \(active.filename)") + await processQueue() + } + } + + /// Cancel an active upload. Kept for compatibility; routes through + /// `cancelUpload(transferId:)`. + public func cancelActiveUpload(_ transferId: UUID) async { + await cancelUpload(transferId: transferId) } // MARK: - PierceFirewall Handling @@ -1294,8 +1498,21 @@ public final class UploadManager { let startTime = Date() var lastProgressUpdate = Date() + // Teardown lets cancelUpload unblock a wedged send. + uploadTeardowns[transferId] = { + Task { await connection.disconnect() } + } + do { while bytesSent < totalSize { + // User cancelled — stop; cancelUpload already updated state. + if cancelledTransferIds.contains(transferId) { + cancelledTransferIds.remove(transferId) + uploadTeardowns.removeValue(forKey: transferId) + logger.info("Upload cancelled mid-stream: \(filePath)") + return + } + // Read chunk from file (off MainActor) guard let chunk = try await fileIO.read(upTo: chunkSize), !chunk.isEmpty else { break @@ -1326,6 +1543,30 @@ public final class UploadManager { t.speed = speed } } + + // Respect speed limit if set (limit > 0 guards div-by-zero) + if let limit = uploadSpeedLimit, limit > 0 { + let elapsed = Date().timeIntervalSince(startTime) + let speed = elapsed > 0 ? Int64(Double(bytesSent - offset) / elapsed) : 0 + if speed > limit { + let delay = Double(chunk.count) / Double(limit) + try? await Task.sleep(for: .milliseconds(Int(delay * 1000))) + } + } + } + + // Re-check cancel before completing — don't record a + // cancelled transfer as completed. + if cancelledTransferIds.remove(transferId) != nil { + uploadTeardowns.removeValue(forKey: transferId) + logger.info("Upload cancelled at completion: \(filePath)") + return + } + + // Short read — fail rather than report a bogus `.completed` + // (matches the direct path guard). + guard bytesSent >= totalSize else { + throw UploadError.connectionFailed } // Complete @@ -1344,6 +1585,7 @@ public final class UploadManager { t.error = nil } + uploadTeardowns.removeValue(forKey: transferId) activeUploads.removeValue(forKey: transferId) ActivityLogger.shared?.logUploadCompleted(filename: (filePath as NSString).lastPathComponent) @@ -1364,11 +1606,16 @@ public final class UploadManager { } catch { logger.error("Upload failed via PeerConnection: \(error.localizedDescription)") - failUpload(transferId: transferId, error: error.localizedDescription) + let wasCancelled = cancelledTransferIds.remove(transferId) != nil + uploadTeardowns.removeValue(forKey: transferId) + + if !wasCancelled { + failUpload(transferId: transferId, error: error.localizedDescription) - // Notify peer so they can re-queue - if let active = activeUploads[transferId] { - await sendUploadFailedToPeer(username: active.username, filename: active.filename) + // Notify peer so they can re-queue + if let active = activeUploads[transferId] { + await sendUploadFailedToPeer(username: active.username, filename: active.filename) + } } activeUploads.removeValue(forKey: transferId) @@ -1376,6 +1623,19 @@ public final class UploadManager { } } + /// Send UploadDenied to peer over a P connection (best effort) + private func sendUploadDeniedToPeer(username: String, filename: String, reason: String) async { + guard let pool = networkClient?.peerConnectionPool else { return } + if let pConn = await pool.getConnectionForUser(username) { + do { + try await pConn.sendUploadDenied(filename: filename, reason: reason) + logger.info("Sent UploadDenied (\(reason)) to \(username) for \(filename)") + } catch { + logger.debug("Could not send UploadDenied to \(username): \(error.localizedDescription)") + } + } + } + /// Send UploadFailed to peer over a P connection so they can re-queue the download private func sendUploadFailedToPeer(username: String, filename: String) async { guard let pool = networkClient?.peerConnectionPool else { return } @@ -1465,6 +1725,11 @@ public final class UploadManager { logger.warning("failUpload: no transfer for \(transferId)") return } + // Never overwrite a user-cancelled row with .failed / a retry. + guard !cancelledTransferIds.contains(transferId), transfer.status != .cancelled else { + logger.info("failUpload skipped, row is cancelled: \(transfer.filename)") + return + } let currentRetryCount = transfer.retryCount transferState?.updateTransfer(id: transferId) { t in t.status = .failed @@ -1574,6 +1839,21 @@ public final class UploadManager { return } + // Re-driving this transferId — drop any stale cancel flag. + cancelledTransferIds.remove(transferId) + + // Re-check permission; terminal fail if no longer allowed. + if let checker = uploadPermissionChecker, !checker(username) { + logger.info("Upload retry denied for \(username): blocked or leech") + transferState?.updateTransfer(id: transferId) { t in + t.status = .failed + t.error = "Denied" + t.retryCount = retryCount + t.nextRetryAt = nil + } + return + } + // Re-resolve the share's `localPath` (not on the Transfer record). // Same lookup `handleQueueUpload` uses on first request. If the // file has been removed from shares between attempts, drop the diff --git a/seeleseek/App/AppState.swift b/seeleseek/App/AppState.swift index 430f54a..793d49d 100644 --- a/seeleseek/App/AppState.swift +++ b/seeleseek/App/AppState.swift @@ -78,6 +78,13 @@ final class AppState { uploadManager?.setMaxConcurrentUploads(newValue) } + // Same wiring for the upload speed limit (KB/s, 0 = unlimited) — + // previously persisted and displayed but never applied. + uploadManager.setUploadSpeedLimit(kbPerSecond: settings.uploadSpeedLimit) + settings.onUploadSpeedLimitChange = { [weak uploadManager] newValue in + uploadManager?.setUploadSpeedLimit(kbPerSecond: newValue) + } + // Peer-status watcher — SocialState tracks live online/away/offline // state for any peer currently in the transfer list (not just // buddies), so rows can surface offline state even for strangers. @@ -95,6 +102,15 @@ final class AppState { self?.uploadManager.cancelRetry(transferId: transferId) } + // Cancel/remove must stop the actual network work, not just flip the + // row status — without this a "cancelled" transfer keeps streaming + // and the completion path flips it back to .completed. + transferState.onCancelRequested = { [weak self] transferId in + guard let self else { return } + self.downloadManager.cancelDownload(transferId: transferId) + Task { await self.uploadManager.cancelUpload(transferId: transferId) } + } + uploadManager.uploadPermissionChecker = { [weak self] username in guard let self else { return true } let patterns = self.settings.activeBlockedPatterns @@ -479,10 +495,11 @@ final class AppState { // Load settings from database await settings.loadFromDatabase() - // `loadFromDatabase` sets `isLoading = true`, so the `didSet` hook - // that normally pushes maxUploadSlots to UploadManager is suppressed. - // Re-apply after the load completes. + // `loadFromDatabase` sets `isLoading = true`, so the `didSet` hooks + // that normally push maxUploadSlots / uploadSpeedLimit to + // UploadManager are suppressed. Re-apply after the load completes. uploadManager.setMaxConcurrentUploads(settings.maxUploadSlots) + uploadManager.setUploadSpeedLimit(kbPerSecond: settings.uploadSpeedLimit) // Load resumable transfers await transferState.loadPersisted() diff --git a/seeleseek/Features/Settings/SettingsState.swift b/seeleseek/Features/Settings/SettingsState.swift index 62c3e06..dc5f354 100644 --- a/seeleseek/Features/Settings/SettingsState.swift +++ b/seeleseek/Features/Settings/SettingsState.swift @@ -187,8 +187,13 @@ final class SettingsState: DownloadSettingsProviding { didSet { guard !isLoading else { return } save() + onUploadSpeedLimitChange?(uploadSpeedLimit) } } + /// Live push (KB/s, 0 = unlimited) to UploadManager. Wired by AppState — + /// the setting was previously cosmetic: persisted and displayed, but + /// never assigned to the manager's limiter. + var onUploadSpeedLimitChange: ((Int) -> Void)? var downloadSpeedLimit: Int = 0 { didSet { guard !isLoading else { return } diff --git a/seeleseek/Features/Transfers/TransferState.swift b/seeleseek/Features/Transfers/TransferState.swift index 8e23a5f..086f101 100644 --- a/seeleseek/Features/Transfers/TransferState.swift +++ b/seeleseek/Features/Transfers/TransferState.swift @@ -447,10 +447,19 @@ final class TransferState: TransferTracking { /// the no-op safe; this just stops the wasted sleep. var onDownloadTerminated: ((UUID) -> Void)? + /// Invoked when the user cancels (or removes) a transfer that may have + /// live network activity. Set by AppState to the managers' cancel APIs + /// so the actual streaming/queue work stops — flipping the row status + /// alone leaves the bytes flowing. Deliberately NOT fired by + /// retryTransfer, which also fires onDownloadTerminated but wants the + /// transfer re-driven, not torn down. + var onCancelRequested: ((UUID) -> Void)? + func cancelTransfer(id: UUID) { updateTransfer(id: id) { transfer in transfer.status = .cancelled } + onCancelRequested?(id) onDownloadTerminated?(id) } @@ -470,6 +479,8 @@ final class TransferState: TransferTracking { uploads.removeAll { $0.id == id } speedHistoryStore.removeValue(forKey: id) reconcilePeerWatches() + // Removing an in-flight transfer must also stop its network work. + onCancelRequested?(id) onDownloadTerminated?(id) // Remove from database diff --git a/seeleseekTests/DownloadRetryClassifierTests.swift b/seeleseekTests/DownloadRetryClassifierTests.swift index 0ce5fd2..36ac816 100644 --- a/seeleseekTests/DownloadRetryClassifierTests.swift +++ b/seeleseekTests/DownloadRetryClassifierTests.swift @@ -21,18 +21,18 @@ struct DownloadRetryClassifierTests { #expect(DownloadManager.isRetriableError(message) == true) } - @Test("NWError canceled (American spelling) is treated as user cancel", + // Cancel-family strings are generated by the client's OWN connection + // teardown (NWError 89, "Connection was cancelled"), i.e. transient + // failures. User-initiated cancels never reach the classifier anymore — + // cancelDownload short-circuits failDownload — so these must retry. + @Test("NWError canceled strings are transient teardown, not user cancel", arguments: [ "Operation canceled", "The operation couldn’t be completed. (Network.NWError error 89 - Operation canceled)", + "Connection was cancelled", ]) - func canceledIsTerminal(message: String) { - #expect(DownloadManager.isRetriableError(message) == false) - } - - @Test("British 'cancelled' is still treated as terminal") - func cancelledIsTerminal() { - #expect(DownloadManager.isRetriableError("Cancelled by user") == false) + func canceledIsRetriable(message: String) { + #expect(DownloadManager.isRetriableError(message) == true) } @Test("Timed-out phrasing retries even without the 'timeout' token", diff --git a/seeleseekTests/DownloadRetryFlowTests.swift b/seeleseekTests/DownloadRetryFlowTests.swift index fed66c9..9dd82dc 100644 --- a/seeleseekTests/DownloadRetryFlowTests.swift +++ b/seeleseekTests/DownloadRetryFlowTests.swift @@ -142,8 +142,8 @@ struct DownloadRetryFlowTests { /// spawns a 2-second-delayed retry Task; we don't await it /// (networkClient is nil in tests, so it would log an error and /// return), and the assertions below run before it wakes. - @Test("UploadFailed after a partial file deletes it and re-queues from scratch") - func uploadFailedAfterPartialDeletesAndRequeues() throws { + @Test("UploadFailed without a resume attempt keeps the partial and uses counted retries") + func uploadFailedWithoutResumeKeepsPartial() throws { let manager = DownloadManager() let tracking = MockTransferTracking() let filename = "@@music\\Artist\\Album\\song.mp3" @@ -151,6 +151,7 @@ struct DownloadRetryFlowTests { transfer.bytesTransferred = 256 tracking.downloads.append(transfer) manager._setTransferStateForTest(tracking) + // resumeOffset stays 0: no resume was attempted this cycle. manager._seedPendingDownloadForTest(makePending(transfer), token: 10) let tempRoot = URL(fileURLWithPath: NSTemporaryDirectory()) @@ -167,15 +168,54 @@ struct DownloadRetryFlowTests { withIntermediateDirectories: true ) try Data(repeating: 0xAB, count: 256).write(to: partialPath) - #expect(FileManager.default.fileExists(atPath: partialPath.path)) manager.handleUploadFailed(username: "alice", filename: filename) let row = tracking.downloads.first - #expect(row?.status == .queued, "partial-file branch must re-queue from scratch") - #expect(row?.error == nil) - #expect(row?.bytesTransferred == 0, "delete-and-restart must reset bytes to zero") - #expect(!FileManager.default.fileExists(atPath: partialPath.path), "partial file must be deleted") - #expect(manager._pendingDownloadCount == 0, "old pending entry is consumed by re-queue") + #expect(row?.status == .failed, "must route through the counted retry machinery") + #expect(row?.error == "Retrying in 10s...") + #expect(row?.nextRetryAt != nil) + #expect(FileManager.default.fileExists(atPath: partialPath.path), + "no resume was attempted, so the partial is kept as groundwork") + #expect(manager._pendingDownloadCount == 0) + } + + @Test("UploadFailed after a resume attempt deletes the partial so the retry starts clean") + func uploadFailedAfterResumeDeletesPartial() throws { + let manager = DownloadManager() + let tracking = MockTransferTracking() + let filename = "@@music\\Artist\\Album\\song.mp3" + var transfer = makeTransfer(username: "alice", filename: filename, status: .connecting) + transfer.bytesTransferred = 256 + tracking.downloads.append(transfer) + manager._setTransferStateForTest(tracking) + var pending = makePending(transfer) + pending.resumeOffset = 256 // this attempt offered a resume offset + manager._seedPendingDownloadForTest(pending, token: 10) + + let tempRoot = URL(fileURLWithPath: NSTemporaryDirectory()) + .appendingPathComponent("seeleseek-tests-\(UUID().uuidString)", isDirectory: true) + try FileManager.default.createDirectory(at: tempRoot, withIntermediateDirectories: true) + defer { try? FileManager.default.removeItem(at: tempRoot) } + manager._setDownloadDirectoryOverrideForTest(tempRoot) + + let partialPath = tempRoot + .appendingPathComponent("Incomplete") + .appendingPathComponent(manager._incompleteBasenameForTest(soulseekPath: filename, username: "alice")) + try FileManager.default.createDirectory( + at: partialPath.deletingLastPathComponent(), + withIntermediateDirectories: true + ) + try Data(repeating: 0xAB, count: 256).write(to: partialPath) + + manager.handleUploadFailed(username: "alice", filename: filename) + + let row = tracking.downloads.first + #expect(row?.status == .failed, "retry is counted/capped, not a silent requeue") + #expect(row?.error == "Retrying in 10s...") + #expect(row?.bytesTransferred == 0, "delete-and-restart resets bytes") + #expect(!FileManager.default.fileExists(atPath: partialPath.path), + "peer rejected a resume, so the partial must be deleted") + #expect(manager._pendingDownloadCount == 0) } } diff --git a/seeleseekTests/FailureTests.swift b/seeleseekTests/FailureTests.swift index 59139ea..dd1cfd1 100644 --- a/seeleseekTests/FailureTests.swift +++ b/seeleseekTests/FailureTests.swift @@ -576,7 +576,7 @@ struct FailureTests { #expect(MessageParser.parseTransferRequest(payload) == nil) } - @Test("Transfer request upload direction with zero fileSize is rejected") + @Test("Transfer request upload direction with zero fileSize is accepted") func testTransferRequestUploadZeroFileSize() { var payload = Data() payload.appendUInt32(1) // upload @@ -584,10 +584,14 @@ struct FailureTests { payload.appendString("test.mp3") payload.appendUInt64(0) // explicit zero size - // Same rationale as above: a zero-size upload-direction request - // is either parser misalignment or a peer-side bug; accepting it - // would make a corrupt transfer look valid. - #expect(MessageParser.parseTransferRequest(payload) == nil) + // Zero-byte files are legal shares (slskd/Seeker advertise size 0 + // for them); rejecting the message stalled queued 0-byte downloads + // forever. Truncated payloads are still rejected (test above) — + // an explicit zero is distinguishable from missing bytes. + let result = MessageParser.parseTransferRequest(payload) + #expect(result != nil) + #expect(result?.direction == .upload) + #expect(result?.fileSize == 0) } @Test("Transfer request download direction succeeds without fileSize") diff --git a/seeleseekTests/PeerConnectivityFixTests.swift b/seeleseekTests/PeerConnectivityFixTests.swift index deb084c..6bfcf58 100644 --- a/seeleseekTests/PeerConnectivityFixTests.swift +++ b/seeleseekTests/PeerConnectivityFixTests.swift @@ -181,8 +181,11 @@ struct PeerConnectivityFixTests { // MARK: - Transfer routing - @Test("Token match takes precedence over filename") - func transferRoutingPrefersToken() { + @Test("Peer token never matches our local token namespace") + func transferRoutingIgnoresPeerToken() { + // The peer's TransferRequest ticket and our locally-generated + // pendingDownloads keys are unrelated namespaces — a numeric + // collision must not route the request to an unrelated download. let pending: [UInt32: DownloadManager.PendingDownload] = [ 42: .init(transferId: UUID(), username: "alice", filename: "song.flac", size: 1000), 7: .init(transferId: UUID(), username: "bob", filename: "other.flac", size: 1000) @@ -190,7 +193,7 @@ struct PeerConnectivityFixTests { let request = TransferRequest( direction: .upload, token: 42, filename: "ANY.flac", size: 1000, username: "zzz" ) - #expect(DownloadManager.matchPendingDownload(request: request, pending: pending) == 42) + #expect(DownloadManager.matchPendingDownload(request: request, pending: pending) == nil) } @Test("(username, filename) wins when token doesn't match") diff --git a/seeleseekTests/ServerMessageRoundTripTests.swift b/seeleseekTests/ServerMessageRoundTripTests.swift index ed9b13b..d15bc64 100644 --- a/seeleseekTests/ServerMessageRoundTripTests.swift +++ b/seeleseekTests/ServerMessageRoundTripTests.swift @@ -115,19 +115,6 @@ struct ServerMessageRoundTripTests { #expect(q == "iron maiden") } - @Test("fileSearchRoom message (legacy code 25)") - func testFileSearchRoom() { - let msg = MessageBuilder.fileSearchRoomMessage(room: "Jazz", token: 777, query: "miles") - let (code, off) = parseMessage(msg) - #expect(code == ServerMessageCode.fileSearchRoom.rawValue) - var o = off - let (room, rLen) = msg.readString(at: o)!; o += rLen - #expect(room == "Jazz") - #expect(msg.readUInt32(at: o) == 777); o += 4 - let (q, _) = msg.readString(at: o)! - #expect(q == "miles") - } - @Test("wishlistSearch message") func testWishlistSearch() { let msg = MessageBuilder.wishlistSearch(token: 1234, query: "rare vinyl") From 32edea04da96b232dc6c18aba68971941b86d1fa Mon Sep 17 00:00:00 2001 From: Brett Henderson Date: Tue, 9 Jun 2026 16:20:52 -0700 Subject: [PATCH 2/2] bump v --- seeleseek.xcodeproj/project.pbxproj | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/seeleseek.xcodeproj/project.pbxproj b/seeleseek.xcodeproj/project.pbxproj index 2eb666f..a06cd93 100644 --- a/seeleseek.xcodeproj/project.pbxproj +++ b/seeleseek.xcodeproj/project.pbxproj @@ -453,7 +453,7 @@ ASSETCATALOG_COMPILER_GLOBAL_ACCENT_COLOR_NAME = AccentColor; CODE_SIGN_ENTITLEMENTS = seeleseek/seeleseek.Debug.entitlements; CODE_SIGN_STYLE = Automatic; - CURRENT_PROJECT_VERSION = 14; + CURRENT_PROJECT_VERSION = 15; DEAD_CODE_STRIPPING = NO; DEVELOPMENT_TEAM = 94XUGF9CU7; ENABLE_APP_SANDBOX = NO; @@ -505,7 +505,7 @@ ASSETCATALOG_COMPILER_GLOBAL_ACCENT_COLOR_NAME = AccentColor; CODE_SIGN_ENTITLEMENTS = seeleseek/seeleseek.entitlements; CODE_SIGN_STYLE = Automatic; - CURRENT_PROJECT_VERSION = 14; + CURRENT_PROJECT_VERSION = 15; DEAD_CODE_STRIPPING = YES; DEVELOPMENT_TEAM = 94XUGF9CU7; ENABLE_APP_SANDBOX = NO;