diff --git a/package/shared-native/ios/StreamMultipartUploadBodyStream.swift b/package/shared-native/ios/StreamMultipartUploadBodyStream.swift index f3b1376cd0..261775acc2 100644 --- a/package/shared-native/ios/StreamMultipartUploadBodyStream.swift +++ b/package/shared-native/ios/StreamMultipartUploadBodyStream.swift @@ -5,10 +5,48 @@ private enum StreamMultipartBodyElement { case file(URL) } +/// Carries a body-production failure out of band. +/// +/// A bound stream pair has no error channel: the only way the producer can signal failure is to +/// close the write end, which the server sees as a truncated body. Recording the real error here +/// lets the upload manager surface it instead of the generic transport error. +final class StreamMultipartBodyErrorBox: @unchecked Sendable { + private let lock = NSLock() + private var storedError: Error? + + var error: Error? { + lock.lock() + defer { lock.unlock() } + return storedError + } + + func record(_ error: Error) { + lock.lock() + defer { lock.unlock() } + if storedError == nil { + storedError = error + } + } +} + final class StreamMultipartUploadBodyStreamFactory { let boundary: String let contentLength: Int64? + /// Set when the body of the **most recent** attempt could not be produced in full. + /// + /// `URLSession` can ask for a fresh body stream (redirect, auth retry) via + /// `needNewBodyStream`. Each attempt therefore gets its own box and `makeStream()` installs it + /// as the current one, so a failure recorded by an abandoned attempt — e.g. its reader going + /// away because URLSession decided to retry — can never fail a later attempt that succeeds. + var bodyError: Error? { + boxLock.lock() + defer { boxLock.unlock() } + return currentErrorBox.error + } + + private let boxLock = NSLock() + private var currentErrorBox = StreamMultipartBodyErrorBox() private let elements: [StreamMultipartBodyElement] private init( @@ -63,7 +101,37 @@ final class StreamMultipartUploadBodyStreamFactory { } func makeStream() -> InputStream { - StreamMultipartSequentialInputStream(elements: elements) + var readStream: Unmanaged? + var writeStream: Unmanaged? + + CFStreamCreateBoundPair( + kCFAllocatorDefault, + &readStream, + &writeStream, + CFIndex(StreamMultipartBodyProducer.transferBufferSize) + ) + + // A fresh box per attempt; see `bodyError`. + let errorBox = StreamMultipartBodyErrorBox() + boxLock.lock() + currentErrorBox = errorBox + boxLock.unlock() + + guard + let input = readStream?.takeRetainedValue() as InputStream?, + let output = writeStream?.takeRetainedValue() + else { + errorBox.record(StreamMultipartUploadError.invalidRequest("Could not create a request body stream")) + return InputStream(data: Data()) + } + + StreamMultipartBodyProducer( + elements: elements, + output: output, + errorBox: errorBox + ).start() + + return input } private static func multipartTextData(boundary: String, part: StreamMultipartTextPart) -> Data { @@ -102,153 +170,263 @@ final class StreamMultipartUploadBodyStreamFactory { } } -private final class StreamMultipartSequentialInputStream: InputStream { +/// Feeds the write end of a `CFStreamCreateBoundPair` from the multipart element list. +/// +/// The body is handed to `URLSession` as the *read* end of a Core Foundation bound stream pair +/// rather than as a hand-rolled `InputStream` subclass. That matters: CFNetwork drives an HTTP/1.1 +/// request body through the `CFReadStream` client-callback machinery, and a plain `InputStream` +/// subclass cannot participate in it — it can never report end-of-stream. CFNetwork stops reading +/// as soon as `Content-Length` is satisfied, so with a subclass it never observed the end of the +/// body, never considered the request finished, and the task sat idle until it timed out (the +/// server had already answered 201). A real bound pair reports every event, and closing the write +/// end is what tells CFNetwork the body is complete. +/// +/// The write end is driven by **GCD** (`CFWriteStreamSetDispatchQueue`) rather than a run loop, so +/// this owns no thread and cannot outlive its work. +private final class StreamMultipartBodyProducer { + static let transferBufferSize = 64 * 1024 + + private enum Refill { + case filled + case drained + case failed(Error) + } + private let elements: [StreamMultipartBodyElement] + private let output: CFWriteStream + private let errorBox: StreamMultipartBodyErrorBox + private let queue: DispatchQueue + private let buffer = UnsafeMutablePointer.allocate( + capacity: StreamMultipartBodyProducer.transferBufferSize + ) + private var currentIndex = 0 private var currentStream: InputStream? - private weak var internalDelegate: StreamDelegate? - private var internalStatus: Stream.Status = .notOpen - private var internalError: Error? - private var scheduledRunLoops: [(runLoop: RunLoop, mode: RunLoop.Mode)] = [] - - init(elements: [StreamMultipartBodyElement]) { + private var bufferOffset = 0 + private var bufferLength = 0 + private var isFinished = false + /// Keeps the producer alive while the stream client holds an unretained pointer to it. + private var selfRetain: StreamMultipartBodyProducer? + + init( + elements: [StreamMultipartBodyElement], + output: CFWriteStream, + errorBox: StreamMultipartBodyErrorBox + ) { self.elements = elements - super.init(data: Data()) + self.output = output + self.errorBox = errorBox + queue = DispatchQueue( + label: "io.getstream.chat.multipart-upload-body", + qos: .userInitiated + ) } - override var delegate: StreamDelegate? { - get { - internalDelegate - } - set { - internalDelegate = newValue - currentStream?.delegate = newValue - } + deinit { + buffer.deallocate() } - override var hasBytesAvailable: Bool { - guard internalStatus != .closed, internalStatus != .error else { - return false - } + func start() { + selfRetain = self - if let currentStream, currentStream.hasBytesAvailable { - return true - } - - return currentIndex < elements.count - } + var context = CFStreamClientContext( + version: 0, + info: Unmanaged.passUnretained(self).toOpaque(), + retain: nil, + release: nil, + copyDescription: nil + ) - override var streamError: Error? { - internalError - } + let events: CFOptionFlags = CFStreamEventType.canAcceptBytes.rawValue + | CFStreamEventType.errorOccurred.rawValue + | CFStreamEventType.endEncountered.rawValue - override var streamStatus: Stream.Status { - internalStatus - } + let didSetClient = CFWriteStreamSetClient( + output, + events, + { _, event, info in + guard let info else { + return + } + Unmanaged.fromOpaque(info) + .takeUnretainedValue() + .handle(event) + }, + &context + ) - override func open() { - guard internalStatus == .notOpen else { + guard didSetClient else { + // No callbacks will ever arrive; finishing here is safe because the dispatch queue has not + // been attached yet, so nothing else can be running. + finish(error: writeStreamError() ?? StreamMultipartUploadError.invalidRequest( + "Could not observe the request body stream" + )) return } - internalStatus = .opening - advanceStreamIfNeeded() - if internalStatus == .error { + // Deliver client callbacks on our serial queue instead of scheduling on a run loop, so the + // producer needs no thread of its own and GCD owns its lifetime. + CFWriteStreamSetDispatchQueue(output, queue) + + guard CFWriteStreamOpen(output) else { + // The queue is attached now, so tear down on it to stay single-threaded. + queue.async { [self] in + finish(error: writeStreamError() ?? StreamMultipartUploadError.invalidRequest( + "Could not open the request body stream" + )) + } return } - internalStatus = currentStream == nil ? .atEnd : .open } - override func close() { - currentStream?.close() - currentStream = nil - internalStatus = .closed + /// The write end's own error, when Core Foundation has one to give. + private func writeStreamError() -> Error? { + CFWriteStreamCopyError(output) as Error? } - override func schedule(in aRunLoop: RunLoop, forMode mode: RunLoop.Mode) { - scheduledRunLoops.append((runLoop: aRunLoop, mode: mode)) - currentStream?.schedule(in: aRunLoop, forMode: mode) + // MARK: - Callbacks (always on `queue`) + + private func handle(_ event: CFStreamEventType) { + switch event { + case .canAcceptBytes: + pump() + case .errorOccurred: + // Usually the reader going away (a cancelled upload), but it can be a genuine write-side + // failure — record whatever CF gives us. Cancellation still wins in the manager, which + // checks `NSURLErrorCancelled` first. + finish(error: writeStreamError()) + case .endEncountered: + finish(error: nil) + default: + break + } } - override func remove(from aRunLoop: RunLoop, forMode mode: RunLoop.Mode) { - scheduledRunLoops.removeAll { $0.runLoop == aRunLoop && $0.mode == mode } - currentStream?.remove(from: aRunLoop, forMode: mode) - } + private func pump() { + while !isFinished, CFWriteStreamCanAcceptBytes(output) { + if bufferOffset >= bufferLength { + switch refill() { + case .filled: + break + case .drained: + finish(error: nil) + return + case .failed(let error): + finish(error: error) + return + } + } - override func read(_ buffer: UnsafeMutablePointer, maxLength len: Int) -> Int { - guard internalStatus != .closed else { - return 0 - } + let written = CFWriteStreamWrite( + output, + buffer + bufferOffset, + bufferLength - bufferOffset + ) + + if written > 0 { + bufferOffset += written + continue + } + + if written == 0 { + // Backpressure, NOT end of body: `CFWriteStreamCanAcceptBytes` may answer true without + // knowing, and the pair reports 0 when it is full. The unwritten remainder stays in + // `buffer` at `bufferOffset`, so the next `.canAcceptBytes` resumes exactly here. + // Closing the stream here would silently truncate the body. + return + } - if internalStatus == .notOpen { - open() + // A negative write is itself the failure signal — do not depend on CF having an error + // object, or the failure degrades into the clean stream close this refactor exists to + // disambiguate. + finish(error: writeStreamError() ?? StreamMultipartUploadError.invalidRequest( + "Could not write the request body stream" + )) + return } + } + /// Fills `buffer` from the next available element. + private func refill() -> Refill { while true { - guard let currentStream else { - if internalStatus == .error { - return -1 + if currentStream == nil { + guard currentIndex < elements.count else { + return .drained } - internalStatus = .atEnd - return 0 - } + let element = elements[currentIndex] + currentIndex += 1 + + switch element { + case .data(let data): + currentStream = InputStream(data: data) + case .file(let url): + guard let stream = InputStream(url: url) else { + return .failed(StreamMultipartUploadError.unreadableFile(url.path)) + } + currentStream = stream + } + + guard let stream = currentStream else { + return .drained + } - let bytesRead = currentStream.read(buffer, maxLength: len) + stream.open() - if bytesRead > 0 { - internalStatus = .open - return bytesRead + if stream.streamStatus == .error { + return .failed(stream.streamError ?? StreamMultipartUploadError.unreadableFile(elementPath())) + } } - if bytesRead < 0 { - internalError = currentStream.streamError - internalStatus = .error - return -1 + guard let stream = currentStream else { + return .drained } - currentStream.close() - self.currentStream = nil - advanceStreamIfNeeded() + let read = stream.read(buffer, maxLength: StreamMultipartBodyProducer.transferBufferSize) - if self.currentStream == nil { - internalStatus = .atEnd - return 0 + if read > 0 { + bufferOffset = 0 + bufferLength = read + return .filled + } + + let readError = read < 0 ? (stream.streamError ?? StreamMultipartUploadError.unreadableFile(elementPath())) : nil + stream.close() + currentStream = nil + + if let readError { + return .failed(readError) } } } - private func advanceStreamIfNeeded() { - guard currentStream == nil else { + private func elementPath() -> String { + guard currentIndex > 0, case .file(let url) = elements[currentIndex - 1] else { + return "" + } + return url.path + } + + private func finish(error: Error?) { + guard !isFinished else { return } - while currentIndex < elements.count { - let nextElement = elements[currentIndex] - currentIndex += 1 - - let nextStream: InputStream? - switch nextElement { - case .data(let data): - nextStream = InputStream(data: data) - case .file(let url): - nextStream = InputStream(url: url) - if nextStream == nil { - internalError = StreamMultipartUploadError.unreadableFile(url.path) - internalStatus = .error - return - } - } + isFinished = true - if let nextStream { - nextStream.delegate = internalDelegate - for scheduled in scheduledRunLoops { - nextStream.schedule(in: scheduled.runLoop, forMode: scheduled.mode) - } - nextStream.open() - currentStream = nextStream - return - } + if let error { + errorBox.record(error) } + + currentStream?.close() + currentStream = nil + + // Unregister before dropping the retain so no callback can arrive against a dead pointer. + CFWriteStreamSetClient(output, 0, nil, nil) + CFWriteStreamSetDispatchQueue(output, nil) + // Closing the write end is what surfaces end-of-stream on the read end. + CFWriteStreamClose(output) + + selfRetain = nil } } diff --git a/package/shared-native/ios/StreamMultipartUploadManager.swift b/package/shared-native/ios/StreamMultipartUploadManager.swift index 951c988ebb..7f2b9b062d 100644 --- a/package/shared-native/ios/StreamMultipartUploadManager.swift +++ b/package/shared-native/ios/StreamMultipartUploadManager.swift @@ -390,6 +390,11 @@ extension StreamMultipartUploadManager: URLSessionDataDelegate, URLSessionTaskDe if nsError.domain == NSURLErrorDomain, nsError.code == NSURLErrorCancelled { state.completion?(.failure(StreamMultipartUploadError.cancelled)) + } else if let bodyError = state.bodyFactory.bodyError { + // The request body could not be produced in full. A bound stream pair has no error + // channel — the reader only sees a truncated body — so prefer the recorded cause over + // the transport error it surfaces as. + state.completion?(.failure(bodyError)) } else { state.completion?(.failure(nsError)) } @@ -397,6 +402,16 @@ extension StreamMultipartUploadManager: URLSessionDataDelegate, URLSessionTaskDe return } + // The task completed without a transport error, but the body may still not have been produced + // in full. A bound stream pair has no error channel, so a producer failure closes the write end + // and the reader sees a clean EOF — with no `Content-Length` (chunked) that is a well-formed + // short body the server can happily accept. Never report a truncated upload as a success. + if let bodyError = state.bodyFactory.bodyError { + state.completion?(.failure(bodyError)) + state.completion = nil + return + } + guard let response = state.response else { state.completion?(.failure(StreamMultipartUploadError.missingHTTPResponse)) state.completion = nil