From 7798401f06be61245fde6403890c85c5e7404a6e Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Thu, 6 Aug 2026 10:09:55 +0200 Subject: [PATCH] perf(apple): cache posters on disk and pool the mgmt connections Moving the management API onto Network.framework left one request per connection, so a library grid paid a TLS handshake per poster where the pooled URLSession had shared one. And the Apple client -- unlike Windows -- never cached art at all, so it re-fetched every poster on every visit. ArtCache: a size- and age-bounded blob cache in the CACHES directory (every byte is re-derivable from the host, so the system is welcome to evict it). Keyed by the SHA-256 of the absolute URL, so host-proxy paths and store CDN URLs share one cache without colliding. Reads touch the entry, so eviction is by last USE, not last write. Empty bodies and data: URLs are refused -- neither is worth a file. Defaults: 128 MB, 30 days. Connection pooling: MgmtConnectionPool keeps up to four keep-alive connections per host and makes further callers wait rather than opening more, which is the part that matters -- a grid can ask for dozens of posters at once. A connection the host dropped since we last used it is indistinguishable from a live one until we write, so a REUSED connection that fails is retried once on a fresh one; a fresh failure is a real failure. Keep-alive means a response can no longer be delimited by the peer hanging up, so HTTPResponseParser.messageLength finds the end from the framing itself -- Content-Length or the chunked terminal chunk plus trailers. Getting that wrong would truncate a response or bleed one into the next, silently, so it carries the bulk of the new tests. A connection with bytes left over after a response is dropped rather than reused: we never pipeline, so anything trailing means we are out of sync. LibraryView closes the loader's pooled connections on disappear instead of leaving sockets open on a screen the user has left. 16 new tests: message framing (both encodings, partial reads, back-to-back responses, close detection) and the cache (binary round trip, key separation, refusals, expiry, LRU eviction). --- .../PunktfunkKit/Connection/ArtCache.swift | 116 ++++++ .../Connection/HTTPResponse.swift | 134 +++++-- .../Connection/LibraryClient.swift | 20 + .../Connection/MgmtTransport.swift | 353 ++++++++++++++---- .../LibraryClientTests.swift | 132 +++++++ 5 files changed, 641 insertions(+), 114 deletions(-) create mode 100644 clients/apple/Sources/PunktfunkKit/Connection/ArtCache.swift diff --git a/clients/apple/Sources/PunktfunkKit/Connection/ArtCache.swift b/clients/apple/Sources/PunktfunkKit/Connection/ArtCache.swift new file mode 100644 index 00000000..5953be03 --- /dev/null +++ b/clients/apple/Sources/PunktfunkKit/Connection/ArtCache.swift @@ -0,0 +1,116 @@ +// On-disk cache for library cover art. +// +// Posters are the bulk of what the library screen transfers and they essentially never change, so +// re-fetching them on every visit is pure waste — the Windows client has cached them on disk for +// this reason and Apple did not. It matters more now that host art rides `MgmtTransport`: a cache +// hit costs no connection at all. +// +// Lives in the CACHES directory on purpose: every byte here is re-derivable from the host, so the +// system is welcome to evict it under storage pressure. Entries are keyed by the SHA-256 of the +// absolute URL, which covers both host-proxy paths and store CDN URLs without either colliding. +// +// Deliberately free of any Network.framework / PunktfunkCore dependency, so it can be unit-tested +// against a temporary directory. + +import CryptoKit +import Foundation + +/// A size- and age-bounded blob cache. An actor so disk work stays off whichever thread the +/// SwiftUI poster view happens to be on, and so pruning can never race a write. +actor ArtCache { + private let directory: URL + private let maxBytes: Int + private let maxAge: TimeInterval + private let fileManager = FileManager.default + + /// `directory` is created on demand. Defaults: 128 MB — a 200-title library of 600×900 + /// capsules lands far under that — and 30 days, which only matters for art a host later + /// replaces. + init(directory: URL, maxBytes: Int = 128 * 1024 * 1024, maxAge: TimeInterval = 30 * 24 * 3600) { + self.directory = directory + self.maxBytes = maxBytes + self.maxAge = maxAge + } + + /// The app's standard location, or nil if the caches directory is unavailable (in which case + /// callers simply run without a cache rather than failing). + static func standard() -> ArtCache? { + guard let caches = FileManager.default.urls( + for: .cachesDirectory, in: .userDomainMask).first + else { return nil } + return ArtCache(directory: caches.appendingPathComponent("PunktfunkArt", isDirectory: true)) + } + + func data(for url: URL) -> Data? { + let file = path(for: url) + guard let data = try? Data(contentsOf: file) else { return nil } + // Age out stale art rather than serving it forever. + if let modified = modificationDate(of: file), Date().timeIntervalSince(modified) > maxAge { + try? fileManager.removeItem(at: file) + return nil + } + // Touch, so eviction can order by last USE rather than last write. + try? fileManager.setAttributes([.modificationDate: Date()], ofItemAtPath: file.path) + return data + } + + func store(_ data: Data, for url: URL) { + // An empty body is not art, and a `data:` URL is already inline — caching either is a + // pure loss. + guard !data.isEmpty, url.scheme?.lowercased() != "data" else { return } + do { + try fileManager.createDirectory(at: directory, withIntermediateDirectories: true) + try data.write(to: path(for: url), options: .atomic) + } catch { + return // a cache that can't write is a slow cache, not a broken app + } + prune() + } + + /// Drop the oldest entries until the directory fits the budget. Also removes anything past + /// `maxAge` so a cache that is under budget still doesn't hoard stale art forever. + func prune() { + let keys: [URLResourceKey] = [.contentModificationDateKey, .fileSizeKey] + guard let entries = try? fileManager.contentsOfDirectory( + at: directory, includingPropertiesForKeys: keys, options: .skipsHiddenFiles) + else { return } + + var files: [(url: URL, date: Date, size: Int)] = [] + var total = 0 + let now = Date() + for entry in entries { + let values = try? entry.resourceValues(forKeys: Set(keys)) + let date = values?.contentModificationDate ?? .distantPast + let size = values?.fileSize ?? 0 + if now.timeIntervalSince(date) > maxAge { + try? fileManager.removeItem(at: entry) + continue + } + files.append((entry, date, size)) + total += size + } + guard total > maxBytes else { return } + + // Oldest first — `data(for:)` touches on read, so this is least-recently-USED. + for file in files.sorted(by: { $0.date < $1.date }) { + guard total > maxBytes else { break } + try? fileManager.removeItem(at: file.url) + total -= file.size + } + } + + /// Wipe the cache — for a "clear cached data" affordance, and for tests. + func clear() { + try? fileManager.removeItem(at: directory) + } + + private func path(for url: URL) -> URL { + let digest = SHA256.hash(data: Data(url.absoluteString.utf8)) + let name = digest.map { String(format: "%02x", $0) }.joined() + return directory.appendingPathComponent(name, isDirectory: false) + } + + private func modificationDate(of file: URL) -> Date? { + (try? file.resourceValues(forKeys: [.contentModificationDateKey]))?.contentModificationDate + } +} diff --git a/clients/apple/Sources/PunktfunkKit/Connection/HTTPResponse.swift b/clients/apple/Sources/PunktfunkKit/Connection/HTTPResponse.swift index 1c604cd7..ae61ea8f 100644 --- a/clients/apple/Sources/PunktfunkKit/Connection/HTTPResponse.swift +++ b/clients/apple/Sources/PunktfunkKit/Connection/HTTPResponse.swift @@ -3,10 +3,14 @@ // We speak HTTP ourselves because the management API has to be reached OUTSIDE the URL loading // system: App Transport Security applies to URLSession and cannot be relaxed for the arbitrary, // user-supplied addresses a punktfunk host lives at (see MgmtTransport for the full story). What -// we need is one unauthenticated-by-header GET per request, so this covers exactly that and -// nothing more — no redirects, no keep-alive, no request bodies, no content negotiation. +// we need is GETs against one host, so this covers exactly that and nothing more — no redirects, +// no request bodies, no content negotiation. // -// Deliberately free of any Network.framework / PunktfunkCore dependency: it is pure bytes-in, +// It does need to find where a response ENDS without waiting for the peer to hang up, because the +// connection is reused across a grid's worth of poster fetches (`messageLength`). Both framings +// hyper emits are handled: `Content-Length`, and `Transfer-Encoding: chunked` for the art proxy. +// +// Deliberately free of any Network.framework / PunktfunkCore dependency: pure bytes-in, // value-out, so it can be unit-tested (and typechecked) on its own. import Foundation @@ -19,6 +23,11 @@ struct HTTPResponse: Sendable { let body: Data func header(_ name: String) -> String? { headers[name.lowercased()] } + + /// Did the peer ask to close? HTTP/1.1 keeps the connection open unless it says otherwise. + var wantsClose: Bool { + header("connection")?.lowercased().contains("close") ?? false + } } enum HTTPParseError: Error, Sendable { @@ -32,14 +41,60 @@ enum HTTPParseError: Error, Sendable { } enum HTTPResponseParser { - /// Parse a complete response — everything read from the socket until EOF. + /// Byte length of the first complete response in `raw`, or nil when more bytes are needed. + /// + /// Also nil when the response carries no framing header at all, since then the body runs + /// until the peer closes and has no knowable length — a connection that answers that way + /// cannot be reused. + static func messageLength(in raw: Data) throws -> Int? { + let b = [UInt8](raw) + guard let head = try parseHead(b) else { return nil } + if head.headers["transfer-encoding"]?.lowercased().contains("chunked") == true { + return try chunkedEnd(b, from: head.bodyStart) + } + if let field = head.headers["content-length"] { + guard let length = Int(field.trimmingCharacters(in: .whitespaces)), length >= 0 else { + throw HTTPParseError.malformedHeader + } + let end = head.bodyStart + length + return b.count >= end ? end : nil + } + return nil // framed by connection close + } + + /// Parse one complete response. `raw` must hold exactly one message (use `messageLength` to + /// slice it) or, for a close-framed response, everything read up to EOF. static func parse(_ raw: Data) throws -> HTTPResponse { - let bytes = [UInt8](raw) - guard let headEnd = findHeaderEnd(bytes) else { throw HTTPParseError.truncated } - // Header field-values are ISO-8859-1 by the grammar; decoding that way never fails, which - // keeps a stray non-UTF-8 byte in some header from failing the whole response. - let head = String(decoding: bytes[0..= 0 else { + throw HTTPParseError.malformedHeader + } + guard rest.count >= length else { throw HTTPParseError.truncated } + body = rest.prefix(length) + } else { + body = rest // framed by connection close: what we read is what there is + } + return HTTPResponse(status: head.status, headers: head.headers, body: body) + } + + private struct Head { + let status: Int + let headers: [String: String] + /// Offset of the first body byte (just past the CRLFCRLF). + let bodyStart: Int + } + + /// Status line + header block, or nil if the block hasn't fully arrived. + private static func parseHead(_ b: [UInt8]) throws -> Head? { + guard let headEnd = findHeaderEnd(b) else { return nil } + let text = String(decoding: b[0..= 0 - else { throw HTTPParseError.malformedHeader } - guard rest.count >= length else { throw HTTPParseError.truncated } - body = rest.prefix(length) - } else { - // No framing header: the body runs to EOF, which is exactly what we read. - body = rest - } - return HTTPResponse(status: status, headers: headers, body: body) + return Head(status: status, headers: headers, bodyStart: headEnd + 4) } - /// Index of the CRLFCRLF that ends the header block. + /// Index just past the CRLFCRLF that ends the header block. private static func findHeaderEnd(_ b: [UInt8]) -> Int? { guard b.count >= 4 else { return nil } for i in 0...(b.count - 4) where b[i] == 0x0D && b[i + 1] == 0x0A @@ -90,6 +131,28 @@ enum HTTPResponseParser { return nil } + /// Offset just past a complete chunked body (terminal chunk plus any trailers), or nil if it + /// hasn't all arrived. + private static func chunkedEnd(_ b: [UInt8], from start: Int) throws -> Int? { + var i = start + while true { + guard let lineEnd = findCRLF(b, from: i) else { return nil } + guard let size = chunkSize(b, i, lineEnd) else { throw HTTPParseError.malformedChunk } + i = lineEnd + 2 + if size == 0 { + // Terminal chunk. Trailers (if any) run to the next empty line. + var j = i + while true { + guard let end = findCRLF(b, from: j) else { return nil } + if end == j { return j + 2 } + j = end + 2 + } + } + guard i + size + 2 <= b.count else { return nil } + i += size + 2 // payload plus its trailing CRLF + } + } + /// `Transfer-Encoding: chunked` decoding. hyper streams the art proxy this way, so this is a /// live path, not defensive dead code. static func decodeChunked(_ data: Data) throws -> Data { @@ -98,15 +161,9 @@ enum HTTPResponseParser { var out = Data() while true { guard let lineEnd = findCRLF(b, from: i) else { throw HTTPParseError.malformedChunk } - // "1a" or "1a;ext=value" — the size is hex, any chunk extension is ignored. - let sizeField = String(decoding: b[i..= 0 else { - throw HTTPParseError.malformedChunk - } + guard let size = chunkSize(b, i, lineEnd) else { throw HTTPParseError.malformedChunk } i = lineEnd + 2 - if size == 0 { return out } // terminal chunk; trailers (if any) are ignored + if size == 0 { return out } // terminal chunk; trailers are ignored guard i + size <= b.count else { throw HTTPParseError.malformedChunk } out.append(contentsOf: b[i..<(i + size)]) i += size @@ -117,6 +174,15 @@ enum HTTPResponseParser { } } + /// "1a" or "1a;ext=value" → 26. Nil if it isn't a hex size. + private static func chunkSize(_ b: [UInt8], _ from: Int, _ to: Int) -> Int? { + let field = String(decoding: b[from..= 0 else { return nil } + return size + } + private static func findCRLF(_ b: [UInt8], from: Int) -> Int? { guard from >= 0, b.count >= 2 else { return nil } var i = from diff --git a/clients/apple/Sources/PunktfunkKit/Connection/LibraryClient.swift b/clients/apple/Sources/PunktfunkKit/Connection/LibraryClient.swift index dc90c624..91e0035c 100644 --- a/clients/apple/Sources/PunktfunkKit/Connection/LibraryClient.swift +++ b/clients/apple/Sources/PunktfunkKit/Connection/LibraryClient.swift @@ -221,6 +221,10 @@ extension Artwork { /// Transport Security can stay ON app-wide. Every other origin keeps ordinary `URLSession` with /// full system trust evaluation and no client certificate, which is exactly what it should get. /// +/// Posters are cached on disk (`ArtCache`), so a second visit to a library costs no network at +/// all — and the connections behind a first visit are pooled and kept alive rather than paying a +/// TLS handshake per tile. +/// /// Built once per library screen and reused across a whole grid's worth of posters. public final class LibraryArtLoader: @unchecked Sendable { private let address: String @@ -230,6 +234,8 @@ public final class LibraryArtLoader: @unchecked Sendable { /// Third-party origins only. No delegate: these are ordinary public HTTPS URLs and get the /// system's normal certificate validation. private let cdn = URLSession(configuration: .default) + /// nil when the caches directory is unavailable — then we simply always fetch. + private let cache = ArtCache.standard() public init( address: String, @@ -245,6 +251,20 @@ public final class LibraryArtLoader: @unchecked Sendable { } public func data(for url: URL) async throws -> Data { + if let cache, let cached = await cache.data(for: url) { return cached } + let fetched = try await fetch(url) + if let cache { await cache.store(fetched, for: url) } + return fetched + } + + /// Release this host's pooled connections — call when the library screen goes away, so we + /// don't sit on open TLS sockets the user is finished with. + public func close() async { + await MgmtConnectionPool.shared.closeAll( + matching: "\(MgmtTransport.unbracketed(address)):\(port):") + } + + private func fetch(_ url: URL) async throws -> Data { guard isHostOrigin(url) else { return try await cdn.data(from: url).0 } var path = url.path.isEmpty ? "/" : url.path if let query = url.query { path += "?\(query)" } diff --git a/clients/apple/Sources/PunktfunkKit/Connection/MgmtTransport.swift b/clients/apple/Sources/PunktfunkKit/Connection/MgmtTransport.swift index eb7fdada..af931cc3 100644 --- a/clients/apple/Sources/PunktfunkKit/Connection/MgmtTransport.swift +++ b/clients/apple/Sources/PunktfunkKit/Connection/MgmtTransport.swift @@ -17,6 +17,11 @@ // user pinned during PIN pairing. That is a stronger check than CA trust here, not a weaker one, // and it is the same rule the QUIC stream plane has always applied via punktfunk-core — which is // precisely why streaming kept working over Tailscale while the library did not. +// +// Connections are POOLED and kept alive: a library screen fetches one JSON payload and then a +// poster per title, and giving each its own TLS handshake was pure latency. `MgmtConnectionPool` +// keeps a small number of connections per host, hands them out one request at a time, and makes +// callers wait rather than opening an unbounded number. import CryptoKit import Foundation @@ -30,6 +35,7 @@ enum MgmtTransportError: Error, Sendable { case connection(String) case timedOut case tooLarge + case invalidPort(UInt16) } enum MgmtTransport { @@ -39,6 +45,10 @@ enum MgmtTransport { /// `GET https://host:port/path`, authenticated by mTLS (`identity`) and pinned by /// `pinnedHostFingerprint` (nil = trust-on-first-use, matching the QUIC connect's semantics). + /// + /// Runs over a pooled keep-alive connection. A connection the host has since dropped is + /// indistinguishable from a live one until we write to it, so a REUSED connection that fails + /// is retried once on a fresh one; a fresh connection that fails is a real error. static func get( host: String, port: UInt16, @@ -48,85 +58,134 @@ enum MgmtTransport { timeout: TimeInterval = 15 ) async throws -> HTTPResponse { guard let nwPort = NWEndpoint.Port(rawValue: port) else { - throw MgmtTransportError.connection("invalid port \(port)") + throw MgmtTransportError.invalidPort(port) } - // One serial queue drives the connection, the verify block, and the timeout, so every - // touch of `Transfer` below is already serialized — no locking, and no chance of two - // callbacks racing to resume the continuation. - let queue = DispatchQueue(label: "io.unom.punktfunk.mgmt-transport") - let state = Transfer() - let options = tlsOptions(identity: identity, pin: pinnedHostFingerprint, - state: state, queue: queue) - let connection = NWConnection( - to: .hostPort(host: NWEndpoint.Host(unbracketed(host)), port: nwPort), - using: NWParameters(tls: options, tcp: NWProtocolTCP.Options())) - let request = requestBytes(host: host, port: port, path: path) + let pin = pinnedHostFingerprint + let key = "\(unbracketed(host)):\(port):\(pin.map(hex) ?? "tofu")" + var lastError: Error = MgmtTransportError.connection("no attempt made") - return try await withCheckedThrowingContinuation { continuation in - func finish(_ result: Result) { - guard !state.finished else { return } - state.finished = true - connection.cancel() - continuation.resume(with: result) + for attempt in 0..<2 { + let connection = await MgmtConnectionPool.shared.acquire(key: key) { + MgmtConnection(host: unbracketed(host), port: nwPort, identity: identity, pin: pin) } + let wasReused = connection.hasServedRequest + do { + let response = try await connection.perform(path: path, timeout: timeout) + await MgmtConnectionPool.shared.release(connection, key: key) + return response + } catch { + await MgmtConnectionPool.shared.release(connection, key: key) + lastError = error + // Only a reused connection earns a second try, and only once: retrying a fresh + // connection would just double every genuine failure's latency. + if !wasReused || attempt == 1 { throw error } + } + } + throw lastError + } - // A rejected pin surfaces as a generic handshake failure; `state.pinRejected` is how - // we recover what actually happened so the UI can say "re-pair" instead of "offline". - func fail(_ error: NWError) { - finish(.failure(state.pinRejected - ? MgmtTransportError.pinMismatch - : MgmtTransportError.connection(String(describing: error)))) - } + static func hex(_ data: Data) -> String { + data.map { String(format: "%02x", $0) }.joined() + } - func receiveLoop() { - connection.receive(minimumIncompleteLength: 1, maximumLength: 64 * 1024) { - chunk, _, isComplete, error in - if let chunk, !chunk.isEmpty { state.buffer.append(chunk) } - if state.buffer.count > maxResponseBytes { - finish(.failure(MgmtTransportError.tooLarge)) - return - } - if let error { fail(error); return } - guard isComplete else { receiveLoop(); return } - // `Connection: close` means EOF ends the response — parse what we have. - finish(Result { try HTTPResponseParser.parse(state.buffer) }) - } - } + /// Saved hosts store bare addresses, but a user who pasted a bracketed IPv6 literal shouldn't + /// get an unresolvable endpoint out of it. + static func unbracketed(_ host: String) -> String { + guard host.hasPrefix("["), host.hasSuffix("]"), host.count > 2 else { return host } + return String(host.dropFirst().dropLast()) + } +} - connection.stateUpdateHandler = { newState in - switch newState { - case .ready: - connection.send(content: request, completion: .contentProcessed { error in - if let error { fail(error); return } - receiveLoop() - }) - case .failed(let error): - fail(error) - case .cancelled: - // Only reachable via our own `finish`, which has already resumed; the guard - // in `finish` makes this a no-op rather than a double-resume crash. - finish(.failure(MgmtTransportError.connection("cancelled"))) - default: - break - } +/// A pool of keep-alive connections, at most `maxPerHost` per host. Callers past that wait for one +/// to come back rather than opening more — a library grid can ask for dozens of posters at once, +/// and answering that with dozens of TLS handshakes is what this exists to prevent. +actor MgmtConnectionPool { + static let shared = MgmtConnectionPool() + + private var available: [String: [MgmtConnection]] = [:] + /// Connections created and not yet closed, per host — the cap this pool enforces. + private var live: [String: Int] = [:] + private var waiters: [String: [CheckedContinuation]] = [:] + private let maxPerHost = 4 + + func acquire(key: String, make: () -> MgmtConnection) async -> MgmtConnection { + while true { + if var idle = available[key], let connection = idle.popLast() { + available[key] = idle + if connection.isHealthy { return connection } + connection.close() + live[key] = max(0, (live[key] ?? 1) - 1) + continue } - queue.asyncAfter(deadline: .now() + timeout) { - finish(.failure(MgmtTransportError.timedOut)) + if (live[key] ?? 0) < maxPerHost { + live[key] = (live[key] ?? 0) + 1 + return make() + } + await withCheckedContinuation { (continuation: CheckedContinuation) in + waiters[key, default: []].append(continuation) } - connection.start(queue: queue) } } - /// Mutable per-request state. Confined to the transport's serial queue — see `get`. - private final class Transfer: @unchecked Sendable { - var finished = false - var pinRejected = false - var buffer = Data() + /// Always call this, on success AND on failure: a connection that is never returned leaks a + /// slot, and enough leaked slots would hang every later request on the waiter queue. + func release(_ connection: MgmtConnection, key: String) { + if connection.isHealthy, (available[key]?.count ?? 0) < maxPerHost { + available[key, default: []].append(connection) + } else { + connection.close() + live[key] = max(0, (live[key] ?? 1) - 1) + } + if var queue = waiters[key], !queue.isEmpty { + let next = queue.removeFirst() + waiters[key] = queue + next.resume() + } } - private static func tlsOptions( - identity: SecIdentity, pin: Data?, state: Transfer, queue: DispatchQueue - ) -> NWProtocolTLS.Options { + /// Drop every idle connection for a host — used when a library screen goes away, so we don't + /// sit on sockets the user is done with. + func closeAll(matching prefix: String) { + for (key, connections) in available where key.hasPrefix(prefix) { + connections.forEach { $0.close() } + live[key] = max(0, (live[key] ?? 0) - connections.count) + available[key] = [] + } + } +} + +/// One TLS connection to a host, serving requests one at a time. The pool guarantees a single +/// caller at a time, so there is no request queueing here. +/// +/// Everything mutable is touched only on `queue`, which also runs the connection's callbacks, the +/// verify block and the timeout — so the state below needs no locking and two callbacks can never +/// race to resume the same continuation. +final class MgmtConnection: @unchecked Sendable { + private let queue = DispatchQueue(label: "io.unom.punktfunk.mgmt-connection") + private let connection: NWConnection + private let host: String + private let port: UInt16 + + private enum Phase { case idle, connecting, ready, dead } + private var phase: Phase = .idle + private var pending: CheckedContinuation? + private var pendingRequest: Data? + /// Bytes read past the end of the last response. Non-empty only if a host pipelines ahead of + /// us, which none do — but dropping them would silently corrupt the next read. + private var buffer = Data() + private var operation = 0 + private var pinRejected = false + private var servedRequest = false + + /// False once the connection has failed; the pool discards these instead of handing them out. + private(set) var isHealthy = true + /// Has this connection completed at least one request? Drives the retry-once rule in + /// `MgmtTransport.get` — only a connection the host may have dropped since is worth retrying. + var hasServedRequest: Bool { servedRequest } + + init(host: String, port: NWEndpoint.Port, identity: SecIdentity, pin: Data?) { + self.host = host + self.port = port.rawValue let options = NWProtocolTLS.Options() let sec = options.securityProtocolOptions sec_protocol_options_set_min_tls_protocol_version(sec, .TLSv12) @@ -135,6 +194,7 @@ enum MgmtTransport { if let secIdentity = sec_identity_create(identity) { sec_protocol_options_set_local_identity(sec, secIdentity) } + let rejected = RejectionFlag() // Replaces system trust evaluation wholesale, which is the point: the host is self-signed // and carries no SAN, so there is nothing for the system policy to succeed at. Pinning the // leaf's SHA-256 is the real check. @@ -143,7 +203,7 @@ enum MgmtTransport { guard let chain = SecTrustCopyCertificateChain(secTrust) as? [SecCertificate], let leaf = chain.first else { - state.pinRejected = true + rejected.value = true complete(false) return } @@ -153,32 +213,165 @@ enum MgmtTransport { } let fingerprint = Data(SHA256.hash(data: SecCertificateCopyData(leaf) as Data)) let matches = fingerprint == pin - if !matches { state.pinRejected = true } + if !matches { rejected.value = true } complete(matches) }, queue) - return options + self.connection = NWConnection( + to: .hostPort(host: NWEndpoint.Host(host), port: port), + using: NWParameters(tls: options, tcp: NWProtocolTCP.Options())) + self.rejection = rejected + self.connection.stateUpdateHandler = { [weak self] state in + self?.handle(state) + } } - private static func requestBytes(host: String, port: UInt16, path: String) -> Data { + /// Set from the verify block, read when mapping the resulting handshake failure. Its own + /// object because the block is built before `self` exists. + private let rejection: RejectionFlag + private final class RejectionFlag: @unchecked Sendable { var value = false } + + func perform(path: String, timeout: TimeInterval) async throws -> HTTPResponse { + try await withCheckedThrowingContinuation { continuation in + queue.async { + guard self.phase != .dead else { + continuation.resume(throwing: MgmtTransportError.connection("connection closed")) + return + } + self.operation += 1 + let op = self.operation + self.pending = continuation + self.pendingRequest = self.requestBytes(path: path) + self.buffer.removeAll(keepingCapacity: true) + self.queue.asyncAfter(deadline: .now() + timeout) { [weak self] in + guard let self, self.operation == op else { return } + self.finish(.failure(MgmtTransportError.timedOut)) + } + switch self.phase { + case .idle: + self.phase = .connecting + self.connection.start(queue: self.queue) + case .ready: + self.send() + case .connecting, .dead: + break // `.ready` (or a failure) will pick the pending request up + } + } + } + } + + func close() { + queue.async { + self.phase = .dead + self.isHealthy = false + self.connection.cancel() + } + } + + // MARK: - Queue-confined internals + + private func handle(_ state: NWConnection.State) { + switch state { + case .ready: + phase = .ready + if pendingRequest != nil { send() } + case .failed(let error): + phase = .dead + isHealthy = false + finish(.failure(mapped(error))) + case .cancelled: + phase = .dead + isHealthy = false + finish(.failure(MgmtTransportError.connection("cancelled"))) + default: + break + } + } + + private func send() { + guard let request = pendingRequest else { return } + pendingRequest = nil + connection.send(content: request, completion: .contentProcessed { [weak self] error in + guard let self else { return } + if let error { + self.isHealthy = false + self.finish(.failure(self.mapped(error))) + return + } + self.receive() + }) + } + + private func receive() { + connection.receive(minimumIncompleteLength: 1, maximumLength: 64 * 1024) { + [weak self] chunk, _, isComplete, error in + guard let self else { return } + if let chunk, !chunk.isEmpty { self.buffer.append(chunk) } + if self.buffer.count > MgmtTransport.maxResponseBytes { + self.isHealthy = false + self.finish(.failure(MgmtTransportError.tooLarge)) + return + } + if let error { + self.isHealthy = false + self.finish(.failure(self.mapped(error))) + return + } + do { + if let length = try HTTPResponseParser.messageLength(in: self.buffer) { + let message = self.buffer.prefix(length) + self.buffer = Data(self.buffer.dropFirst(length)) + let response = try HTTPResponseParser.parse(message) + // A response the peer means to be last leaves nothing reusable behind. Nor + // does a stream with bytes left over: we never pipeline, so anything trailing + // means we are out of sync, and reusing the connection would misread the next + // response rather than fail cleanly. + if response.wantsClose || !self.buffer.isEmpty { self.isHealthy = false } + self.servedRequest = true + self.finish(.success(response)) + return + } + if isComplete { + // No framing header: the body ran to EOF, so what we have is the whole thing + // and the connection is spent. + self.isHealthy = false + let response = try HTTPResponseParser.parse(self.buffer) + self.servedRequest = true + self.finish(.success(response)) + return + } + } catch { + self.isHealthy = false + self.finish(.failure(error)) + return + } + self.receive() + } + } + + private func finish(_ result: Result) { + guard let continuation = pending else { return } + pending = nil + operation += 1 // invalidate this operation's timeout + continuation.resume(with: result) + } + + /// A rejected pin surfaces as a generic handshake failure; the flag is how we recover what + /// actually happened, so the UI can say "re-pair" instead of "offline". + private func mapped(_ error: NWError) -> MgmtTransportError { + rejection.value ? .pinMismatch : .connection(String(describing: error)) + } + + private func requestBytes(path: String) -> Data { // An IPv6 literal is bracketed in the Host header (RFC 9110 §7.2); a name or IPv4 is not. - let bare = unbracketed(host) - let authority = bare.contains(":") ? "[\(bare)]:\(port)" : "\(bare):\(port)" + let authority = host.contains(":") ? "[\(host)]:\(port)" : "\(host):\(port)" let request = """ GET \(path) HTTP/1.1\r Host: \(authority)\r User-Agent: punktfunk-apple\r Accept: */*\r - Connection: close\r \r """ return Data(request.utf8) } - - /// Saved hosts store bare addresses, but a user who pasted a bracketed IPv6 literal shouldn't - /// get an unresolvable endpoint out of it. - private static func unbracketed(_ host: String) -> String { - guard host.hasPrefix("["), host.hasSuffix("]"), host.count > 2 else { return host } - return String(host.dropFirst().dropLast()) - } } diff --git a/clients/apple/Tests/PunktfunkKitTests/LibraryClientTests.swift b/clients/apple/Tests/PunktfunkKitTests/LibraryClientTests.swift index 84d789ab..cb71f517 100644 --- a/clients/apple/Tests/PunktfunkKitTests/LibraryClientTests.swift +++ b/clients/apple/Tests/PunktfunkKitTests/LibraryClientTests.swift @@ -165,6 +165,138 @@ final class LibraryClientTests: XCTestCase { XCTAssertEqual([UInt8](response.body), png) } + // MARK: - Message framing (keep-alive) + + // Connections are pooled and reused, so a response has to be delimited WITHOUT waiting for + // the peer to hang up. Getting this wrong either truncates a response or bleeds one response + // into the next, and both would be silent. + + func testMessageLengthDelimitsContentLengthFraming() throws { + let complete = raw("HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello") + XCTAssertEqual(try HTTPResponseParser.messageLength(in: complete), complete.count) + XCTAssertNil(try HTTPResponseParser.messageLength( + in: raw("HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhel"))) + } + + func testMessageLengthDelimitsChunkedFraming() throws { + let complete = raw("HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n0\r\n\r\n") + XCTAssertEqual(try HTTPResponseParser.messageLength(in: complete), complete.count) + // Mid-chunk, and terminal chunk without its closing blank line. + XCTAssertNil(try HTTPResponseParser.messageLength( + in: raw("HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhel"))) + XCTAssertNil(try HTTPResponseParser.messageLength( + in: raw("HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n0\r\n"))) + // Trailers belong to the message and must be consumed with it. + let trailered = raw("HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n" + + "3\r\nabc\r\n0\r\nX-Trailer: 1\r\n\r\n") + XCTAssertEqual(try HTTPResponseParser.messageLength(in: trailered), trailered.count) + } + + func testMessageLengthIsNilWithoutFraming() throws { + // No Content-Length and not chunked ⇒ the body runs to EOF and the connection can't be + // reused. Partial header blocks are likewise "not yet". + XCTAssertNil(try HTTPResponseParser.messageLength(in: raw("HTTP/1.1 200 OK\r\n\r\nto-eof"))) + XCTAssertNil(try HTTPResponseParser.messageLength(in: raw("HTTP/1.1 200 OK\r\nContent-Len"))) + } + + func testBackToBackResponsesSplitExactly() throws { + // The reuse case that matters: two responses arriving in one read. + let first = raw("HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello") + let second = raw("HTTP/1.1 200 OK\r\nContent-Length: 3\r\n\r\nbye") + let both = first + second + XCTAssertEqual(try HTTPResponseParser.messageLength(in: both), first.count) + XCTAssertEqual( + String(decoding: try HTTPResponseParser.parse(both.prefix(first.count)).body, + as: UTF8.self), "hello") + XCTAssertEqual( + String(decoding: try HTTPResponseParser.parse(both.dropFirst(first.count)).body, + as: UTF8.self), "bye") + } + + func testConnectionCloseIsDetected() throws { + let response = try HTTPResponseParser.parse( + raw("HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Length: 0\r\n\r\n")) + XCTAssertTrue(response.wantsClose) + // HTTP/1.1 keeps the connection open unless told otherwise. + XCTAssertFalse(try HTTPResponseParser.parse( + raw("HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")).wantsClose) + } + + // MARK: - Art cache + + private func temporaryCacheDirectory() -> URL { + URL(fileURLWithPath: NSTemporaryDirectory()) + .appendingPathComponent("pf-art-test-\(UUID().uuidString)", isDirectory: true) + } + + func testArtCacheRoundTripsBinaryAndSeparatesKeys() async throws { + let directory = temporaryCacheDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + let cache = ArtCache(directory: directory) + + let png = Data([0x89, 0x50, 0x4E, 0x47, 0x00, 0xFF, 0x0D, 0x0A]) + let portrait = URL(string: "https://100.64.1.2:47990/api/v1/library/art/steam:570/portrait")! + let header = URL(string: "https://100.64.1.2:47990/api/v1/library/art/steam:570/header")! + + var hit = await cache.data(for: portrait) + XCTAssertNil(hit, "cold cache must miss") + await cache.store(png, for: portrait) + hit = await cache.data(for: portrait) + XCTAssertEqual(hit, png, "posters are binary; the body must survive byte-for-byte") + // Sibling art of the same title must not collide. + let sibling = await cache.data(for: header) + XCTAssertNil(sibling) + } + + func testArtCacheRefusesEmptyAndInlineData() async throws { + let directory = temporaryCacheDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + let cache = ArtCache(directory: directory) + + let empty = URL(string: "https://cdn.example.com/empty.jpg")! + await cache.store(Data(), for: empty) + let emptyHit = await cache.data(for: empty) + XCTAssertNil(emptyHit, "an empty body is not art") + + let inline = URL(string: "data:image/png;base64,iVBORw0KGgo=")! + await cache.store(Data("x".utf8), for: inline) + let inlineHit = await cache.data(for: inline) + XCTAssertNil(inlineHit, "a data: URL is already inline — caching it is a pure loss") + } + + func testArtCacheAgesEntriesOut() async throws { + let directory = temporaryCacheDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + let cache = ArtCache(directory: directory, maxAge: 0.4) + let url = URL(string: "https://cdn.example.com/stale.jpg")! + + await cache.store(Data("stale".utf8), for: url) + let fresh = await cache.data(for: url) + XCTAssertNotNil(fresh) + try await Task.sleep(nanoseconds: 700_000_000) + let expired = await cache.data(for: url) + XCTAssertNil(expired) + } + + func testArtCacheEvictsLeastRecentlyUsedOverBudget() async throws { + let directory = temporaryCacheDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + // Budget holds three of these; the fourth store must evict. + let cache = ArtCache(directory: directory, maxBytes: 300) + let blob = Data(repeating: 0x41, count: 100) + var urls: [URL] = [] + for i in 0..<4 { + let url = URL(string: "https://cdn.example.com/blob\(i).jpg")! + urls.append(url) + await cache.store(blob, for: url) + try await Task.sleep(nanoseconds: 60_000_000) // distinct mtimes for LRU ordering + } + let evicted = await cache.data(for: urls[0]) + XCTAssertNil(evicted, "the oldest entry should have been evicted") + let newest = await cache.data(for: urls[3]) + XCTAssertEqual(newest, blob) + } + func testBaseURLBracketsIPv6Only() { XCTAssertEqual(LibraryClient.baseURL(address: "192.168.1.70", port: 47990), "https://192.168.1.70:47990")