amadan

rastrillo / native Public

Support bounded native editor streams and document responses

Paul Campbell pushed by paul@keymail.dev 6ddea4d0c161d9347f426ad3e487ce31a4012a54 parent 33ffe99
7 files changed, +439 −6
  • README.md +17 −0
  • Sources/RastrilloApple/NativeHTTP.swift +54 −6
  • Sources/RastrilloApple/NativeResponse.swift +9 −0
  • Sources/RastrilloApple/NativeStreamLines.swift +42 −0
  • Tests/RastrilloAppleTests/HTTPStreamCancellationTests.swift +177 −0
  • Tests/RastrilloAppleTests/HTTPTests.swift +104 −0
  • Tests/RastrilloAppleTests/StreamLinesTests.swift +36 −0

CI cancelled — superseded by a newer push — run details

diff --git a/README.md b/README.md
index 7eaf6e6..dcb6beb 100644
--- a/README.md
+++ b/README.md
@@ -54,6 +54,23 @@ when it removes duplication: see [docs/go-mobile.md](docs/go-mobile.md).
## Validate
+Apple editor clients use `RastrilloApple.NativeHTTP` for ephemeral, origin-bound
+requests. `responseDetails` returns body, status and case-insensitive headers for
+protocol cursors. `response` and `request` retain the 4 MiB default body bound;
+an explicit `maxBytes` allows larger snapshots/exports up to 256 MiB. Oversized
+responses fail rather than returning truncated content.
+
+`stream(path:onOpen:onLine:)` requires a 200 `text/event-stream` response and supplies
+UTF-8 lines, including empty event delimiters. It accepts CR, LF and CRLF,
+limits each line to 256 KiB and each event to 1 MiB, and honors task cancellation.
+The optional `onOpen` callback runs after response validation and completes before
+any lines are delivered. A server that subscribes before flushing response headers
+allows this callback to fetch an initial snapshot without missing intervening edits.
+The consumer parses events, revalidates permissions through its server, and
+refetches/reconnects after EOF. An unfinished event at EOF must not be applied.
+Streams use the same no-redirect session and origin-bound credentials as ordinary
+requests. No response, stream or export is automatically persisted.
+
`make ci` runs the Swift tests and compiles both companion targets. It
requires macOS, Xcode and XcodeGen. Builds are unsigned; this gate does not
claim physical-device, signing or store-distribution validation.
diff --git a/Sources/RastrilloApple/NativeHTTP.swift b/Sources/RastrilloApple/NativeHTTP.swift
index 9298843..f8df215 100644
--- a/Sources/RastrilloApple/NativeHTTP.swift
+++ b/Sources/RastrilloApple/NativeHTTP.swift
@@ -45,8 +45,7 @@ public final class NativeHTTP: @unchecked Sendable {
return clean.url ?? url
}
- public func response(_ path: String, method: String = "GET", body: Data? = nil,
- contentType: String = "application/json") async throws -> (Data, Int) {
+ private func makeRequest(_ path: String, method: String, body: Data?, contentType: String) throws -> URLRequest {
guard path.hasPrefix("/"), !path.hasPrefix("//"),
let url = URL(string: path, relativeTo: origin)?.absoluteURL,
url.scheme == origin.scheme, url.host == origin.host, url.port == origin.port else {
@@ -59,19 +58,68 @@ public final class NativeHTTP: @unchecked Sendable {
request.setValue("application/json", forHTTPHeaderField: "Accept")
if body != nil { request.setValue(contentType, forHTTPHeaderField: "Content-Type") }
if let credential { request.setValue("\(credential.name)=\(credential.value)", forHTTPHeaderField: "Cookie") }
+ return request
+ }
+
+ public func response(_ path: String, method: String = "GET", body: Data? = nil,
+ contentType: String = "application/json", maxBytes: Int = 4 * 1024 * 1024) async throws -> (Data, Int) {
+ let result = try await responseDetails(path, method: method, body: body, contentType: contentType, maxBytes: maxBytes)
+ return (result.data, result.status)
+ }
+
+ /// Exposes protocol cursors (for example Docs-Seq) without sharing a cookie jar.
+ public func responseDetails(_ path: String, method: String = "GET", body: Data? = nil,
+ contentType: String = "application/json", maxBytes: Int = 4 * 1024 * 1024) async throws -> NativeResponse {
+ guard maxBytes > 0, maxBytes <= 256 * 1024 * 1024 else {
+ throw NativeClientError.message("Invalid response size limit.")
+ }
+ let request = try makeRequest(path, method: method, body: body, contentType: contentType)
let (bytes, response) = try await session.bytes(for: request)
+ defer { bytes.task.cancel() }
guard let response = response as? HTTPURLResponse else { throw NativeClientError.message("Invalid server response.") }
var data = Data()
for try await byte in bytes {
- guard data.count < 4 * 1024 * 1024 else { throw NativeClientError.message("The server response is too large.") }
+ try Task.checkCancellation()
+ guard data.count < maxBytes else { throw NativeClientError.message("The server response is too large.") }
data.append(byte)
}
- return (data, response.statusCode)
+ var headers: [String: String] = [:]
+ for (key, value) in response.allHeaderFields {
+ if let key = key as? String, let value = value as? String { headers[key.lowercased()] = value }
+ }
+ return NativeResponse(data: data, status: response.statusCode, headers: headers)
+ }
+
+ /// Delivers bounded UTF-8 SSE lines, including blank event delimiters. Callers
+ /// own event parsing and reconnect cursors; a clean EOF is never a saved edit.
+ public func stream(_ path: String, onOpen: @Sendable () async throws -> Void = {},
+ onLine: @Sendable (String) async throws -> Void) async throws {
+ var request = try makeRequest(path, method: "GET", body: nil, contentType: "application/json")
+ request.setValue("text/event-stream", forHTTPHeaderField: "Accept")
+ let (bytes, response) = try await session.bytes(for: request)
+ defer { bytes.task.cancel() }
+ guard let response = response as? HTTPURLResponse,
+ response.statusCode == 200,
+ response.value(forHTTPHeaderField: "Content-Type")?.split(separator: ";").first?.trimmingCharacters(in: .whitespaces).lowercased() == "text/event-stream" else {
+ throw NativeClientError.message("The live connection was refused. Reconnect to refresh your session.")
+ }
+ // A server that subscribes before flushing headers lets this callback fetch
+ // a snapshot without a subscribe-after-snapshot gap. Lines remain buffered
+ // until the callback completes; consumers reconcile their version cursors.
+ try Task.checkCancellation()
+ try await onOpen()
+ var parser = NativeStreamLines()
+ for try await byte in bytes {
+ try Task.checkCancellation()
+ if let line = try parser.append(byte) { try await onLine(line) }
+ }
+ // An unterminated event is not delivered at EOF; the next snapshot is the
+ // authority after reconnection, rather than an incomplete network frame.
}
public func request<T: Decodable>(_ path: String, method: String = "GET", body: Data? = nil,
- contentType: String = "application/json") async throws -> T {
- let (data, status) = try await response(path, method: method, body: body, contentType: contentType)
+ contentType: String = "application/json", maxBytes: Int = 4 * 1024 * 1024) async throws -> T {
+ let (data, status) = try await response(path, method: method, body: body, contentType: contentType, maxBytes: maxBytes)
guard (200..<300).contains(status) else {
if status == 401 || status == 303 { throw NativeClientError.message("Your session has ended. Sign out and sign in again.") }
if status == 404 { throw NativeClientError.message("This server does not support this native feature yet.") }
diff --git a/Sources/RastrilloApple/NativeResponse.swift b/Sources/RastrilloApple/NativeResponse.swift
new file mode 100644
index 0000000..4b174cd
--- /dev/null
+++ b/Sources/RastrilloApple/NativeResponse.swift
@@ -0,0 +1,9 @@
+import Foundation
+
+public struct NativeResponse: Sendable {
+ public let data: Data
+ public let status: Int
+ public let headers: [String: String]
+
+ public func header(_ name: String) -> String? { headers[name.lowercased()] }
+}
diff --git a/Sources/RastrilloApple/NativeStreamLines.swift b/Sources/RastrilloApple/NativeStreamLines.swift
new file mode 100644
index 0000000..9b9217a
--- /dev/null
+++ b/Sources/RastrilloApple/NativeStreamLines.swift
@@ -0,0 +1,42 @@
+import Foundation
+
+/// Bound before decoding: AsyncBytes.lines can allocate an unbounded line.
+struct NativeStreamLines {
+ private var buffer = Data()
+ private var eventBytes = 0
+ private var afterCR = false
+ private var firstLine = true
+ let lineLimit: Int
+ let eventLimit: Int
+
+ init(lineLimit: Int = 256 * 1024, eventLimit: Int = 1024 * 1024) {
+ self.lineLimit = lineLimit
+ self.eventLimit = eventLimit
+ }
+
+ mutating func append(_ byte: UInt8) throws -> String? {
+ if afterCR {
+ afterCR = false
+ if byte == 10 { return nil }
+ }
+ if byte == 10 || byte == 13 {
+ afterCR = byte == 13
+ guard var line = String(data: buffer, encoding: .utf8) else {
+ throw NativeClientError.message("The live connection returned invalid text.")
+ }
+ if firstLine {
+ firstLine = false
+ if line.hasPrefix("\u{FEFF}") { line.removeFirst() }
+ }
+ buffer.removeAll(keepingCapacity: true)
+ if line.isEmpty { eventBytes = 0 }
+ return line
+ }
+ guard buffer.count < lineLimit, eventBytes < eventLimit else {
+ throw NativeClientError.message("The live update is too large. Reconnect to refresh it.")
+ }
+ buffer.append(byte)
+ eventBytes += 1
+ return nil
+ }
+}
diff --git a/Tests/RastrilloAppleTests/HTTPStreamCancellationTests.swift b/Tests/RastrilloAppleTests/HTTPStreamCancellationTests.swift
new file mode 100644
index 0000000..fc1cbe8
--- /dev/null
+++ b/Tests/RastrilloAppleTests/HTTPStreamCancellationTests.swift
@@ -0,0 +1,177 @@
+#if os(macOS)
+import Foundation
+import XCTest
+@testable import RastrilloApple
+
+private actor OpenEvents {
+ var values: [String] = []
+ func append(_ value: String) { values.append(value) }
+}
+
+/// A live socket must close when its consumer rejects or cancels the response;
+/// returning a Swift error alone can leave URLSession downloading indefinitely.
+final class HTTPStreamCancellationTests: XCTestCase {
+ private struct Fixture {
+ let process: Process
+ let client: NativeHTTP
+ let directory: URL
+ var closed: URL { directory.appendingPathComponent("closed") }
+ func stop() {
+ if process.isRunning { process.terminate() }
+ try? FileManager.default.removeItem(at: directory)
+ }
+ }
+
+ private func fixture(contentType: String = "text/event-stream", prefix: String = ": ready\n\n") throws -> Fixture {
+ let directory = FileManager.default.temporaryDirectory.appendingPathComponent(UUID().uuidString)
+ try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true)
+ let process = Process()
+ process.executableURL = URL(fileURLWithPath: "/usr/bin/python3")
+ process.arguments = ["-u", "-c", #"""
+import http.server, pathlib, socket, sys
+marker, content_type, prefix = sys.argv[1:]
+class Handler(http.server.BaseHTTPRequestHandler):
+ def log_message(self, *args): pass
+ def do_GET(self):
+ self.send_response(200)
+ self.send_header('Content-Type', content_type)
+ self.end_headers()
+ self.wfile.write(prefix.encode('utf-8'))
+ self.wfile.flush()
+ # Keep the response open without more data. A read blocks in the client
+ # until cancellation; the server sees EOF only when the socket closes.
+ self.connection.settimeout(4)
+ try:
+ if self.connection.recv(1) == b'':
+ pathlib.Path(marker).write_text('closed')
+ except (ConnectionResetError, BrokenPipeError):
+ pathlib.Path(marker).write_text('closed')
+ except socket.timeout:
+ pass
+server = http.server.HTTPServer(('127.0.0.1', 0), Handler)
+print(server.server_port, flush=True)
+server.handle_request()
+"""#, directory.appendingPathComponent("closed").path, contentType, prefix]
+ let output = Pipe()
+ process.standardOutput = output
+ do { try process.run() } catch {
+ try? FileManager.default.removeItem(at: directory)
+ throw error
+ }
+ var port = Data()
+ while port.count < 12 {
+ let byte = output.fileHandleForReading.readData(ofLength: 1)
+ if byte.isEmpty || byte == Data([10]) { break }
+ port.append(byte)
+ }
+ guard let number = String(data: port, encoding: .utf8).flatMap(Int.init), number > 0,
+ let origin = URL(string: "http://127.0.0.1:\(number)") else {
+ if process.isRunning { process.terminate() }
+ try? FileManager.default.removeItem(at: directory)
+ throw NativeClientError.message("The isolated cancellation fixture could not start.")
+ }
+ return Fixture(process: process, client: NativeHTTP(origin: origin), directory: directory)
+ }
+
+ private func assertClosed(_ fixture: Fixture, file: StaticString = #filePath, line: UInt = #line) async throws {
+ for _ in 0..<60 {
+ if FileManager.default.fileExists(atPath: fixture.closed.path) { return }
+ try await Task.sleep(nanoseconds: 25_000_000)
+ }
+ XCTFail("The rejected or cancelled response left its network socket open", file: file, line: line)
+ }
+
+ func testRejectedContentTypeClosesLiveSocket() async throws {
+ // Clear Foundation's small text-response sniffing buffer before waiting.
+ let server = try fixture(contentType: "text/plain", prefix: String(repeating: "x", count: 4096))
+ defer { server.stop() }
+ do {
+ try await server.client.stream("/live", onOpen: { XCTFail("A refused response invoked onOpen") }) { _ in
+ XCTFail("Rejected content reached the consumer")
+ }
+ XCTFail("Accepted a response with the wrong content type")
+ } catch { XCTAssertTrue(error.localizedDescription.contains("refused")) }
+ try await assertClosed(server)
+ }
+
+ func testResponseSizeFailureClosesLiveSocket() async throws {
+ let server = try fixture(contentType: "application/octet-stream", prefix: "12345")
+ defer { server.stop() }
+ do {
+ _ = try await server.client.responseDetails("/live", maxBytes: 4)
+ XCTFail("Accepted an oversized response")
+ } catch { XCTAssertTrue(error.localizedDescription.contains("too large")) }
+ try await assertClosed(server)
+ }
+
+ func testThrowingLineConsumerClosesLiveSocket() async throws {
+ enum ConsumerFailure: Error { case stopped }
+ let server = try fixture()
+ defer { server.stop() }
+ do {
+ try await server.client.stream("/live") { _ in throw ConsumerFailure.stopped }
+ XCTFail("A consumer failure was swallowed")
+ } catch ConsumerFailure.stopped {} catch { XCTFail("Unexpected error: \(error)") }
+ try await assertClosed(server)
+ }
+
+ func testOpenCallbackCompletesBeforeFirstLine() async throws {
+ enum Stop: Error { case finished }
+ let server = try fixture()
+ defer { server.stop() }
+ let events = OpenEvents()
+ do {
+ try await server.client.stream("/live", onOpen: {
+ // Suspending proves the transport awaits completion rather than merely
+ // scheduling the snapshot callback concurrently with incoming lines.
+ try await Task.sleep(nanoseconds: 25_000_000)
+ await events.append("opened")
+ }) { line in
+ await events.append(line)
+ throw Stop.finished
+ }
+ XCTFail("The line callback failure was swallowed")
+ } catch Stop.finished {} catch { XCTFail("Unexpected error: \(error)") }
+ let received = await events.values
+ XCTAssertEqual(received, ["opened", ": ready"])
+ try await assertClosed(server)
+ }
+
+ func testThrowingOpenCallbackClosesWithoutDeliveringLines() async throws {
+ enum Stop: Error { case finished }
+ let server = try fixture()
+ defer { server.stop() }
+ do {
+ try await server.client.stream("/live", onOpen: { throw Stop.finished }) { _ in
+ XCTFail("Lines arrived after snapshot initialization failed")
+ }
+ XCTFail("The open callback failure was swallowed")
+ } catch Stop.finished {} catch { XCTFail("Unexpected error: \(error)") }
+ try await assertClosed(server)
+ }
+
+ func testCancellationWhileAwaitingBytesClosesLiveSocket() async throws {
+ let server = try fixture()
+ defer { server.stop() }
+ let ready = expectation(description: "received stream prefix")
+ let finished = expectation(description: "cancelled stream returned")
+ let task = Task {
+ do {
+ try await server.client.stream("/live") { line in
+ if line == ": ready" { ready.fulfill() }
+ }
+ XCTFail("A cancelled stream completed successfully")
+ } catch {
+ XCTAssertTrue(Task.isCancelled)
+ }
+ finished.fulfill()
+ }
+ await fulfillment(of: [ready], timeout: 2)
+ // Give the iterator an opportunity to suspend waiting for the next byte.
+ try await Task.sleep(nanoseconds: 50_000_000)
+ task.cancel()
+ await fulfillment(of: [finished], timeout: 2)
+ try await assertClosed(server)
+ }
+}
+#endif
diff --git a/Tests/RastrilloAppleTests/HTTPTests.swift b/Tests/RastrilloAppleTests/HTTPTests.swift
new file mode 100644
index 0000000..2264385
--- /dev/null
+++ b/Tests/RastrilloAppleTests/HTTPTests.swift
@@ -0,0 +1,104 @@
+#if os(macOS)
+import Foundation
+import Darwin
+import XCTest
+@testable import RastrilloApple
+
+private actor ReceivedLines {
+ var values: [String] = []
+ func append(_ line: String) { values.append(line) }
+}
+
+final class HTTPTests: XCTestCase {
+ private func stop(_ process: Process) {
+ process.terminate()
+ let deadline = Date().addingTimeInterval(0.5)
+ while process.isRunning && Date() < deadline { Thread.sleep(forTimeInterval: 0.01) }
+ if process.isRunning { kill(process.processIdentifier, SIGKILL) }
+ }
+
+ private func fixture() throws -> (Process, NativeHTTP) {
+ let process = Process()
+ process.executableURL = URL(fileURLWithPath: "/usr/bin/python3")
+ process.arguments = ["-u", "-c", #"""
+import http.server
+class Handler(http.server.BaseHTTPRequestHandler):
+ def log_message(self, *args): pass
+ def do_GET(self):
+ if self.path == '/redirect':
+ self.send_response(302)
+ self.send_header('Location', '/unexpected-follow')
+ self.end_headers()
+ return
+ if self.path == '/stream':
+ body = b'event: update\r\ndata: 42\r\n\r\n'
+ content_type = 'text/event-stream; charset=utf-8'
+ else:
+ body = b'12345'
+ content_type = 'application/octet-stream'
+ self.send_response(200)
+ self.send_header('Content-Type', content_type)
+ self.send_header('Content-Length', str(len(body)))
+ self.send_header('Docs-Seq', '42')
+ self.send_header('Seen-Origin', self.headers.get('Origin', 'missing'))
+ self.send_header('Seen-Cookie', self.headers.get('Cookie', 'missing'))
+ self.end_headers()
+ self.wfile.write(body)
+server = http.server.HTTPServer(('127.0.0.1', 0), Handler)
+print(server.server_port, flush=True)
+server.serve_forever()
+"""#]
+ let output = Pipe()
+ process.standardOutput = output
+ try process.run()
+ var port = Data()
+ while port.count < 12 {
+ let byte = output.fileHandleForReading.readData(ofLength: 1)
+ if byte.isEmpty || byte == Data([10]) { break }
+ port.append(byte)
+ }
+ guard let portText = String(data: port, encoding: .utf8), let number = Int(portText), number > 0,
+ let origin = URL(string: "http://127.0.0.1:\(number)") else {
+ process.terminate()
+ throw NativeClientError.message("The isolated HTTP fixture could not start.")
+ }
+ return (process, NativeHTTP(origin: origin, credential: NativeSession(name: "rastrillo_session", value: "fixture")))
+ }
+
+ func testResponseLimitHeadersAndOrigin() async throws {
+ let (server, client) = try fixture()
+ defer { stop(server) }
+ let response = try await client.responseDetails("/bytes", maxBytes: 5)
+ XCTAssertEqual(response.data, Data("12345".utf8))
+ XCTAssertEqual(response.header("Docs-Seq"), "42")
+ XCTAssertEqual(response.header("Seen-Origin"), client.origin.absoluteString)
+ XCTAssertEqual(response.header("Seen-Cookie"), "rastrillo_session=fixture")
+ do {
+ _ = try await client.response("/bytes", maxBytes: 4)
+ XCTFail("A truncated response must fail, never succeed with a prefix")
+ } catch { XCTAssertTrue(error.localizedDescription.contains("too large")) }
+ }
+
+ func testRedirectIsReturnedWithoutFollowing() async throws {
+ let (server, client) = try fixture()
+ defer { stop(server) }
+ let response = try await client.responseDetails("/redirect")
+ XCTAssertEqual(response.status, 302)
+ XCTAssertTrue(response.data.isEmpty)
+ }
+
+ func testLiveTransportAndWrongContentType() async throws {
+ let (server, client) = try fixture()
+ defer { stop(server) }
+ let lines = ReceivedLines()
+ try await client.stream("/stream") { line in await lines.append(line) }
+ let received = await lines.values
+ XCTAssertEqual(received, ["event: update", "data: 42", ""])
+ do {
+ try await client.stream("/bytes") { _ in XCTFail("Non-SSE content reached the live editor") }
+ XCTFail("Non-SSE response was accepted")
+ } catch { XCTAssertTrue(error.localizedDescription.contains("refused")) }
+ }
+}
+
+#endif
diff --git a/Tests/RastrilloAppleTests/StreamLinesTests.swift b/Tests/RastrilloAppleTests/StreamLinesTests.swift
new file mode 100644
index 0000000..0794e4e
--- /dev/null
+++ b/Tests/RastrilloAppleTests/StreamLinesTests.swift
@@ -0,0 +1,36 @@
+import XCTest
+@testable import RastrilloApple
+
+final class StreamLinesTests: XCTestCase {
+ func testUTF8AndAllSSELineEndings() throws {
+ var parser = NativeStreamLines()
+ let input = "\u{FEFF}event: update\r\ndata: café 👩🏽‍💻\r\r: ping\n\n"
+ let lines = try input.utf8.compactMap { try parser.append($0) }
+ XCTAssertEqual(lines, ["event: update", "data: café 👩🏽‍💻", "", ": ping", ""])
+ }
+
+ func testInvalidUTF8DoesNotReachConsumer() throws {
+ var parser = NativeStreamLines()
+ XCTAssertNil(try parser.append(0xff))
+ XCTAssertThrowsError(try parser.append(10))
+ }
+
+ func testLineLimitRejectsBeforeUnboundedAllocation() throws {
+ var parser = NativeStreamLines(lineLimit: 4)
+ for byte in "data".utf8 { XCTAssertNil(try parser.append(byte)) }
+ XCTAssertThrowsError(try parser.append(58))
+ }
+
+ func testEventLimitSpansLinesAndResetsAtDelimiter() throws {
+ var parser = NativeStreamLines(lineLimit: 8, eventLimit: 8)
+ _ = try "1234\n5678\n".utf8.compactMap { try parser.append($0) }
+ XCTAssertThrowsError(try parser.append(57))
+ XCTAssertEqual(try parser.append(10), "")
+ XCTAssertNil(try parser.append(57))
+ }
+
+ func testUnterminatedLineIsNotDelivered() throws {
+ var parser = NativeStreamLines()
+ XCTAssertEqual(try "event: update\ndata: partial".utf8.compactMap { try parser.append($0) }, ["event: update"])
+ }
+}