diff --git a/Package.resolved b/Package.resolved index 7d8394f..d4414c1 100644 --- a/Package.resolved +++ b/Package.resolved @@ -1,5 +1,5 @@ { - "originHash" : "1b4e137ea3dcd0d4e006082f78d81f891f628f0196b8b7cb39bc1205ee66007f", + "originHash" : "651d260da7e20e6ca89d05327ce8dc35c2397a5027fbf040bdf598ca59c18570", "pins" : [ { "identity" : "pureyaml", @@ -7,7 +7,7 @@ "location" : "https://codeberg.org/NodetteHQ/PureYAML.git", "state" : { "branch" : "main", - "revision" : "c13b7c97fd6cd7e3ccfbb5f301322fa340a4d184" + "revision" : "2fcf54684e4216a1459cb8bd7bb14dac05b856ca" } }, { diff --git a/Package.swift b/Package.swift index e66d18c..4d80bdb 100644 --- a/Package.swift +++ b/Package.swift @@ -10,11 +10,11 @@ let package = Package( products: [ .library( name: "PureYAMLStreaming", - targets: ["PureYAMLStreaming"], + targets: ["PureYAMLStreaming"] ), .executable( name: "pureyaml-streaming-wasm-smoke", - targets: ["PureYAMLStreamingWasmSmoke"], + targets: ["PureYAMLStreamingWasmSmoke"] ), ], dependencies: [ @@ -24,7 +24,7 @@ let package = Package( targets: [ .target( name: "PureYAMLStreaming", - dependencies: ["PureYAML"], + dependencies: ["PureYAML"] ), .executableTarget( name: "PureYAMLStreamingWasmSmoke", @@ -32,11 +32,11 @@ let package = Package( "PureYAMLStreaming", "PureYAML", .product(name: "WASIHTTPClient", package: "SwiftWASIHTTPClient"), - ], + ] ), .testTarget( name: "PureYAMLStreamingTests", - dependencies: ["PureYAMLStreaming"], + dependencies: ["PureYAMLStreaming"] ), - ], + ] ) diff --git a/README.md b/README.md index 669a37e..b2ad599 100644 --- a/README.md +++ b/README.md @@ -17,14 +17,21 @@ adapters around that stable core. Prototype. -The current implementation streams YAML input in chunks, detects -document-boundary candidates, and emits one parsed `PureYAML.Stream.Document` at -a time. That gives bounded memory for multi-document streams where each document -fits in memory. +The current implementation streams YAML input in chunks, detects document +boundaries with PureYAML's resumable token scanner +(`PureYAML.Streaming.Scanner`), and emits one parsed +`PureYAML.Stream.Document` at a time. That gives bounded memory for +multi-document streams where each document fits in memory. -It is not yet a token-level YAML scanner. A future milestone will replace the -document-boundary scanner with a resumable token scanner that can feed -event-level consumers directly. +Boundary detection is grammar-driven, not line-based: a `---` counts as a +boundary only where the token scanner emits a `documentStart` token, a +directive starts a new document only after an explicit `...`, and a chunk +boundary may fall anywhere, including mid-token. Splitting is lossless; the +emitted document sources concatenate back to the exact input. + +A future milestone will surface token-level events to event-level consumers +directly; the public `StreamingToken` currently carries whole document +sources only. ## Memory Contract diff --git a/Sources/PureYAMLStreaming/ChunkedUTF8Reader.swift b/Sources/PureYAMLStreaming/ChunkedUTF8Reader.swift index 2fbe9e1..4c0be19 100644 --- a/Sources/PureYAMLStreaming/ChunkedUTF8Reader.swift +++ b/Sources/PureYAMLStreaming/ChunkedUTF8Reader.swift @@ -1,8 +1,8 @@ import Foundation -extension PureYAMLStreaming { +public extension PureYAMLStreaming { /// Bounded UTF-8 chunk reader with byte-accurate mark tracking. - public struct ChunkedUTF8Reader: Sendable { + struct ChunkedUTF8Reader: Sendable { private var source: Source private let chunkSize: Int private var pendingBytes: [UInt8] = [] @@ -13,7 +13,7 @@ extension PureYAMLStreaming { guard chunkSize > 0 else { throw StreamingError.invalidChunkSize(chunkSize) } - self.source = .file(try FileHandle(forReadingFrom: fileURL)) + source = try .file(FileHandle(forReadingFrom: fileURL)) self.chunkSize = chunkSize } @@ -21,7 +21,7 @@ extension PureYAMLStreaming { guard chunkSize > 0 else { throw StreamingError.invalidChunkSize(chunkSize) } - self.source = .file(fileHandle) + source = .file(fileHandle) self.chunkSize = chunkSize } @@ -29,7 +29,7 @@ extension PureYAMLStreaming { guard chunkSize > 0 else { throw StreamingError.invalidChunkSize(chunkSize) } - self.source = .data(data, offset: 0) + source = .data(data, offset: 0) self.chunkSize = chunkSize } @@ -58,7 +58,7 @@ extension PureYAMLStreaming { throw StreamingError.invalidUTF8( line: markStorage.line, column: markStorage.column, - index: markStorage.index, + index: markStorage.index ) } @@ -70,7 +70,7 @@ extension PureYAMLStreaming { throw StreamingError.invalidUTF8( line: markStorage.line, column: markStorage.column, - index: markStorage.index, + index: markStorage.index ) } advanceMark(over: chunk) @@ -98,7 +98,7 @@ extension PureYAMLStreaming { throw StreamingError.invalidUTF8( line: markStorage.line, column: markStorage.column, - index: markStorage.index, + index: markStorage.index ) } pendingBytes.removeAll(keepingCapacity: true) diff --git a/Sources/PureYAMLStreaming/Parser.swift b/Sources/PureYAMLStreaming/Parser.swift index ade84bb..bb3b8dd 100644 --- a/Sources/PureYAMLStreaming/Parser.swift +++ b/Sources/PureYAMLStreaming/Parser.swift @@ -1,15 +1,15 @@ import Foundation import PureYAML -extension PureYAMLStreaming { +public extension PureYAMLStreaming { /// Chunked YAML stream parser that emits complete documents as they finish. - public struct Parser: Sendable { + struct Parser: Sendable { public var chunkSize: Int public var pureParser: PureYAML.Parsing.Parser public init( chunkSize: Int = 64 * 1024, - pureParser: PureYAML.Parsing.Parser = .init(), + pureParser: PureYAML.Parsing.Parser = .init() ) { self.chunkSize = chunkSize self.pureParser = pureParser @@ -22,7 +22,7 @@ extension PureYAMLStreaming { /// documents unless the caller does. public func parseDocuments( from url: URL, - _ body: (PureYAML.Stream.Document) throws -> Void, + _ body: (PureYAML.Stream.Document) throws -> Void ) throws { var reader = try ChunkedUTF8Reader(fileURL: url, chunkSize: chunkSize) try parseDocuments(from: &reader, body) @@ -31,7 +31,7 @@ extension PureYAMLStreaming { /// Parses documents from UTF-8 data and calls `body` once per completed document. public func parseDocuments( data: Data, - _ body: (PureYAML.Stream.Document) throws -> Void, + _ body: (PureYAML.Stream.Document) throws -> Void ) throws { var reader = try ChunkedUTF8Reader(data: data, chunkSize: chunkSize) try parseDocuments(from: &reader, body) @@ -59,7 +59,7 @@ extension PureYAMLStreaming { private func parseDocuments( from reader: inout ChunkedUTF8Reader, - _ body: (PureYAML.Stream.Document) throws -> Void, + _ body: (PureYAML.Stream.Document) throws -> Void ) throws { var scanner = ResumableDocumentScanner() var nextIndex = 0 @@ -71,7 +71,7 @@ extension PureYAMLStreaming { } } - if let source = try scanner.finish(), !source.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty { + for source in try scanner.finish() where !source.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty { try emit(source, index: &nextIndex, body) } } @@ -79,7 +79,7 @@ extension PureYAMLStreaming { private func emit( _ source: String, index: inout Int, - _ body: (PureYAML.Stream.Document) throws -> Void, + _ body: (PureYAML.Stream.Document) throws -> Void ) throws { let documents = try pureParser.parseStream(source) for document in documents { diff --git a/Sources/PureYAMLStreaming/ResumableDocumentScanner.swift b/Sources/PureYAMLStreaming/ResumableDocumentScanner.swift index e6cb18a..cbe2770 100644 --- a/Sources/PureYAMLStreaming/ResumableDocumentScanner.swift +++ b/Sources/PureYAMLStreaming/ResumableDocumentScanner.swift @@ -1,86 +1,182 @@ -import Foundation +import PureYAML -extension PureYAMLStreaming { - /// Resumable document-boundary scanner. +public extension PureYAMLStreaming { + /// Resumable document-boundary scanner driven by `PureYAML.Streaming.Scanner`. /// - /// This is intentionally a document-token scanner, not yet a YAML grammar - /// token scanner. It proves chunked input and document-at-a-time parsing - /// while the lower-level token scanner remains incubating. - public struct ResumableDocumentScanner: Sendable { - private var pendingLine = "" - private var currentDocument = "" - private var hasCurrentDocumentContent = false - private var line = 1 + /// Chunks are tokenized by PureYAML's resumable token scanner (proven + /// byte-for-byte identical to batch scanning across the yaml-test-suite + /// corpus), and the buffered source text is cut immediately before each + /// `documentStart` token, or before the first directive token that follows + /// an explicit `...` document end. Unlike the previous line-based + /// heuristic, a `---` that the grammar does not treat as a document start + /// (for example inside a quoted scalar) can never misfire as a boundary, + /// directives stay attached to the document they introduce, and a chunk + /// boundary may fall anywhere, including mid-token. + /// + /// Guarantees: + /// - Lossless: concatenating every emitted segment (from `consume(_:)` and + /// `finish()`, in order) reproduces the fed input exactly. + /// - Boundary-sound: no YAML document ever spans two segments, so parsing + /// the segments one at a time yields the same documents as batch parsing + /// the whole stream. + /// - Bounded: after each cut the token scanner is restarted on the + /// remaining buffer (a document boundary fully resets scanner state), so + /// retained state is bounded by the current document, not the stream. + struct ResumableDocumentScanner: Sendable { + private var scanner = PureYAML.Streaming.Scanner() + private var buffer: [Unicode.Scalar] = [] + private var scannerFedCount = 0 + private var segmentHasContent = false + private var inDirectivePreamble = false + private var afterDocumentEnd = false + private var isFinal = false public init() {} + /// Appends a chunk of YAML source and returns every document source + /// completed by it. A chunk may end anywhere, including mid-token. public mutating func consume(_ chunk: String) throws -> [String] { - var documents: [String] = [] - pendingLine.append(chunk) - - while let newlineRange = pendingLine.rangeOfCharacter(from: .newlines) { - let lineText = String(pendingLine[.. [StreamingToken] { try consume(chunk).map(StreamingToken.documentSource) } - public mutating func finish() throws -> String? { - if !pendingLine.isEmpty { - var ignored: [String] = [] - try consumeLine(pendingLine, documents: &ignored) - pendingLine.removeAll(keepingCapacity: true) + /// Marks the end of input and returns the remaining document sources: + /// any boundary that only end-of-input could confirm (for example a + /// trailing `---` line), followed by the final buffered document. + public mutating func finish() throws -> [String] { + isFinal = true + feedScanner() + scanner.finish() + var documents = try drainTokens() + if !buffer.isEmpty { + documents.append(String(String.UnicodeScalarView(buffer))) } - defer { - currentDocument.removeAll(keepingCapacity: true) - hasCurrentDocumentContent = false - } - return currentDocument.isEmpty ? nil : currentDocument + self = ResumableDocumentScanner() + return documents } - private mutating func consumeLine( - _ sourceLine: String, - documents: inout [String], - ) throws { - if isDocumentStart(sourceLine), hasCurrentDocumentContent { - documents.append(currentDocument) - currentDocument = sourceLine - hasCurrentDocumentContent = true + /// Feeds the token scanner every buffered scalar it has not seen yet, + /// except a trailing CR while more input may still arrive. Whether + /// that CR is a lone break or the first half of a CR LF pair is + /// undecidable until the next scalar (or end of input) is known, and + /// PureYAML's persisted plain-scalar suspension counts a pair split + /// exactly there as two line breaks, shifting every later token mark + /// down a line (observed at PureYAML commit 2fcf546). Withholding the + /// CR keeps every break the scanner sees whole, so the token line + /// numbers this type cuts by always match batch scanning. + private mutating func feedScanner() { + var target = buffer.count + if !isFinal, buffer.last == "\r" { + target -= 1 + } + guard target > scannerFedCount else { return } - - currentDocument.append(sourceLine) - if hasSemanticContent(sourceLine) { - hasCurrentDocumentContent = true - } + scanner.feed(String(String.UnicodeScalarView(buffer[scannerFedCount ..< target]))) + scannerFedCount = target } - private func isDocumentStart(_ sourceLine: String) -> Bool { - let trimmedNewline = sourceLine.trimmingCharacters(in: .newlines) - guard trimmedNewline.hasPrefix("---") else { - return false - } - let markerEnd = trimmedNewline.index(trimmedNewline.startIndex, offsetBy: 3) - if markerEnd != trimmedNewline.endIndex { - let next = trimmedNewline[markerEnd] - guard next == " " || next == "\t" || next == "#" else { - return false + private mutating func drainTokens() throws -> [String] { + var documents: [String] = [] + drain: while true { + switch try scanner.next() { + case let .token(token): + handle(token, documents: &documents) + case .needMoreInput, .done: + break drain } } - return trimmedNewline.prefix(3) == "---" + return documents } - private func hasSemanticContent(_ sourceLine: String) -> Bool { - let trimmed = sourceLine.trimmingCharacters(in: .whitespacesAndNewlines) - return !trimmed.isEmpty && !trimmed.hasPrefix("#") + private mutating func handle(_ token: PureYAML.Streaming.Token, documents: inout [String]) { + switch token.kind { + case .streamStart, .streamEnd: + return + case .documentStart: + if segmentHasContent, !inDirectivePreamble { + // The restarted scanner re-emits this token from the new + // buffer start; that re-emission sets the segment flags. + documents.append(cutSegment(beforeLine: token.range.start.line)) + return + } + inDirectivePreamble = false + afterDocumentEnd = false + segmentHasContent = true + case .versionDirective, .tagDirective, .reservedDirective: + // A directive is a legal document boundary only after an + // explicit `...`. Without one, batch parsing rejects the + // stream, so the directive must stay in the current segment + // for the per-segment parse to reject it the same way. + if segmentHasContent, !inDirectivePreamble, afterDocumentEnd { + documents.append(cutSegment(beforeLine: token.range.start.line)) + return + } + inDirectivePreamble = true + afterDocumentEnd = false + segmentHasContent = true + case .documentEnd: + inDirectivePreamble = false + afterDocumentEnd = true + segmentHasContent = true + default: + inDirectivePreamble = false + afterDocumentEnd = false + segmentHasContent = true + } + } + + /// Cuts the buffer at the start of `line` (0-based, relative to the + /// current scanner instance, whose input always begins at `buffer[0]`), + /// emits the prefix, and restarts the token scanner on the remainder so + /// that no state older than the current document is retained. + /// + /// Boundary tokens (`---` and directives) always start at column 0, so + /// the cut offset is that line's first scalar. Line counting mirrors + /// the scanner's reader exactly: CR LF is one break, and CR, LF, NEL, + /// LINE SEPARATOR, and PARAGRAPH SEPARATOR each end a line. + private mutating func cutSegment(beforeLine line: Int) -> String { + var offset = 0 + var linesSeen = 0 + while linesSeen < line, offset < buffer.count { + if buffer[offset] == "\r", offset + 1 < buffer.count, buffer[offset + 1] == "\n" { + offset += 2 + linesSeen += 1 + } else if Self.isBreak(buffer[offset]) { + offset += 1 + linesSeen += 1 + } else { + offset += 1 + } + } + + let segment = String(String.UnicodeScalarView(buffer[.. Bool { + switch scalar { + case "\r", "\n", "\u{85}", "\u{2028}", "\u{2029}": true + default: false + } } } } diff --git a/Sources/PureYAMLStreaming/StreamingError.swift b/Sources/PureYAMLStreaming/StreamingError.swift index 6f2cc30..831d0d5 100644 --- a/Sources/PureYAMLStreaming/StreamingError.swift +++ b/Sources/PureYAMLStreaming/StreamingError.swift @@ -1,8 +1,8 @@ import PureYAML -extension PureYAMLStreaming { +public extension PureYAMLStreaming { /// Errors from the chunked streaming layer before or around PureYAML parsing. - public enum StreamingError: Error, Equatable, Sendable, CustomStringConvertible { + enum StreamingError: Error, Equatable, Sendable, CustomStringConvertible { case invalidUTF8(line: Int, column: Int, index: Int) case invalidChunkSize(Int) case nestedDocumentStart(line: Int, column: Int) diff --git a/Sources/PureYAMLStreaming/StreamingMark.swift b/Sources/PureYAMLStreaming/StreamingMark.swift index c8eeaf5..ec8c2c3 100644 --- a/Sources/PureYAMLStreaming/StreamingMark.swift +++ b/Sources/PureYAMLStreaming/StreamingMark.swift @@ -1,6 +1,6 @@ -extension PureYAMLStreaming { +public extension PureYAMLStreaming { /// Source position at a streaming boundary. - public struct StreamingMark: Equatable, Sendable { + struct StreamingMark: Equatable, Sendable { public var line: Int public var column: Int public var index: Int diff --git a/Sources/PureYAMLStreaming/StreamingToken.swift b/Sources/PureYAMLStreaming/StreamingToken.swift index 92dc555..bd659ea 100644 --- a/Sources/PureYAMLStreaming/StreamingToken.swift +++ b/Sources/PureYAMLStreaming/StreamingToken.swift @@ -1,6 +1,6 @@ -extension PureYAMLStreaming { +public extension PureYAMLStreaming { /// Tokens emitted by the current streaming scanner milestone. - public enum StreamingToken: Equatable, Sendable { + enum StreamingToken: Equatable, Sendable { case documentSource(String) } } diff --git a/Sources/PureYAMLStreamingWasmSmoke/main.swift b/Sources/PureYAMLStreamingWasmSmoke/main.swift index 92691ba..73becff 100644 --- a/Sources/PureYAMLStreamingWasmSmoke/main.swift +++ b/Sources/PureYAMLStreamingWasmSmoke/main.swift @@ -50,7 +50,7 @@ let fallbackYAML = Data( enabled: true nested: index: 2 - """.utf8, + """.utf8 ) do { @@ -91,7 +91,7 @@ do { statusCode = nil stats = try parseInstrumentedDocuments(data: fallbackYAML, chunkSize: chunkSize) } - let parseMilliseconds = Date().timeIntervalSince(parseStart) * 1_000 + let parseMilliseconds = Date().timeIntervalSince(parseStart) * 1000 let ok = stats.documentCount > 0 let result = SmokeResult( ok: ok, @@ -108,9 +108,9 @@ do { firstName: stats.firstName, lastName: stats.lastName, parseMilliseconds: parseMilliseconds, - message: ok ? "PureYAMLStreaming WASM parse passed" : "PureYAMLStreaming WASM parse found no documents", + message: ok ? "PureYAMLStreaming WASM parse passed" : "PureYAMLStreaming WASM parse found no documents" ) - print(try renderJSON(result)) + try print(renderJSON(result)) } catch { let result = SmokeResult( ok: false, @@ -127,7 +127,7 @@ do { firstName: nil, lastName: nil, parseMilliseconds: 0, - message: String(describing: error), + message: String(describing: error) ) print((try? renderJSON(result)) ?? #"{"ok":false,"message":"failed to render error"}"#) } @@ -161,7 +161,7 @@ func parseInstrumentedDocuments(reader: inout PureYAMLStreaming.ChunkedUTF8Reade } } - if let source = try scanner.finish(), !source.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty { + for source in try scanner.finish() where !source.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty { stats.scannerDocumentSourceCount += 1 try emitDocuments(from: source, parser: pureParser, stats: &stats) } @@ -172,7 +172,7 @@ func parseInstrumentedDocuments(reader: inout PureYAMLStreaming.ChunkedUTF8Reade func emitDocuments( from source: String, parser: PureYAML.Parsing.Parser, - stats: inout ParseStats, + stats: inout ParseStats ) throws { let documents = try parser.parseStream(source) for document in documents { @@ -194,8 +194,8 @@ func stringValue(_ key: String, in value: PureYAML.Model.Value) -> String? { return string } -func renderJSON(_ value: Value) throws -> String { +func renderJSON(_ value: some Encodable) throws -> String { let encoder = JSONEncoder() encoder.outputFormatting = [.prettyPrinted, .sortedKeys] - return String(decoding: try encoder.encode(value), as: UTF8.self) + return try String(decoding: encoder.encode(value), as: UTF8.self) } diff --git a/Tests/PureYAMLStreamingTests/DocumentScannerDifferentialTests.swift b/Tests/PureYAMLStreamingTests/DocumentScannerDifferentialTests.swift new file mode 100644 index 0000000..8b7766d --- /dev/null +++ b/Tests/PureYAMLStreamingTests/DocumentScannerDifferentialTests.swift @@ -0,0 +1,99 @@ +import Foundation +import Testing +@testable import PureYAMLStreaming + +/// Differential proof that the token-driven `ResumableDocumentScanner` emits +/// byte-identical document splits to the retired line-based heuristic on every +/// input where that heuristic was correct: the existing fixture corpus, +/// generated multi-document streams, and handwritten benign shapes. Inputs +/// where the heuristic was wrong (directives, grammar-internal `---` lines, +/// input ending on a `---` line without a newline) are covered by +/// `DocumentScannerTests` instead, which pins the corrected behavior. +@Suite("Document scanner differential") +struct DocumentScannerDifferentialTests { + /// Streams the line-based heuristic split correctly: newline-terminated, + /// directive-free, and with every column-zero `---` a real boundary. + static let benignInputs: [String] = [ + "---\na: 1\n---\nb: 2\n", + "name: only\n", + "# note\n---\na: 1\n", + "---\na: 1\n# trailing\n---\nb: 2\n", + "a: 1\n--- # comment\nb: 2\n", + "a: 1\n...\nb: 2\n", + "---\na: 1\n...\n---\nb: 2\n", + "---\n---\n", + "a: 1\r\n---\r\nb: 2\r\n", + "# only comments\n\n", + "", + ] + + @Test("matches the line-based reference on benign streams", arguments: benignInputs) + func matchesReferenceOnBenignStreams(input: String) throws { + for chunkSize in [1, 2, 3, 7, 64, max(input.unicodeScalars.count, 1)] { + let rewired = try rewiredSegments(input, chunkSize: chunkSize) + let reference = try referenceSegments(input, chunkSize: chunkSize) + #expect(rewired == reference, "chunk size \(chunkSize) diverged for: \(input.debugDescription)") + #expect(rewired.joined() == input) + } + } + + @Test("matches the line-based reference on a generated multi-document stream") + func matchesReferenceOnGeneratedStream() throws { + var input = "" + for index in 0 ..< 300 { + input += """ + --- + name: doc-\(index) + enabled: \(index.isMultiple(of: 2) ? "true" : "false") + nested: + index: \(index) + tags: + - real + - streaming + - chunk-\(index % 17) + + """ + } + + for chunkSize in [7, 31, 4096] { + let rewired = try rewiredSegments(input, chunkSize: chunkSize) + let reference = try referenceSegments(input, chunkSize: chunkSize) + #expect(rewired.count == 300) + #expect(rewired == reference, "chunk size \(chunkSize) diverged on the generated stream") + #expect(rewired.joined() == input) + } + } + + @Test("matches the line-based reference across the Geekbench corpus") + func matchesReferenceAcrossCorpus() throws { + let fixtureRoot = URL(fileURLWithPath: #filePath) + .deletingLastPathComponent() + .deletingLastPathComponent() + .deletingLastPathComponent() + .deletingLastPathComponent() + .appendingPathComponent("PureYAMLGeekbench") + .appendingPathComponent("Fixtures") + .appendingPathComponent("real-yaml") + guard FileManager.default.fileExists(atPath: fixtureRoot.path) else { + return + } + + let files = try FileManager.default.contentsOfDirectory( + at: fixtureRoot, + includingPropertiesForKeys: nil + ).filter { url in + url.pathExtension == "yaml" || url.pathExtension == "yml" + } + #expect(files.count >= 100) + + for file in files { + let input = try String(decoding: Data(contentsOf: file), as: UTF8.self) + for chunkSize in [31, 4097] { + let rewired = try rewiredSegments(input, chunkSize: chunkSize) + let reference = try referenceSegments(input, chunkSize: chunkSize) + #expect(rewired == reference, "chunk size \(chunkSize) diverged for \(file.lastPathComponent)") + #expect(rewired.joined() == input, "segments lost bytes for \(file.lastPathComponent)") + } + } + } +} diff --git a/Tests/PureYAMLStreamingTests/DocumentScannerTestSupport.swift b/Tests/PureYAMLStreamingTests/DocumentScannerTestSupport.swift new file mode 100644 index 0000000..a3c6c92 --- /dev/null +++ b/Tests/PureYAMLStreamingTests/DocumentScannerTestSupport.swift @@ -0,0 +1,54 @@ +import Foundation +import PureYAML +@testable import PureYAMLStreaming + +/// The integer stored under `key` when `value` is a mapping, or nil. +func intValue(_ key: String, in value: PureYAML.Model.Value) -> Int? { + guard case let .mapping(mapping) = value, + case let .int(number)? = mapping[key] + else { + return nil + } + return number +} + +/// Splits `input` into chunks of `size` Unicode scalars, so a chunk boundary +/// can land anywhere: mid-line, mid-token, and mid-grapheme. +func scalarChunks(_ input: String, size: Int) -> [String] { + precondition(size > 0) + let scalars = Array(input.unicodeScalars) + var chunks: [String] = [] + var index = 0 + while index < scalars.count { + let end = min(index + size, scalars.count) + chunks.append(String(String.UnicodeScalarView(scalars[index ..< end]))) + index = end + } + return chunks +} + +/// All document sources the rewired scanner emits for `input` fed in +/// `chunkSize`-scalar chunks, in emission order. +func rewiredSegments(_ input: String, chunkSize: Int) throws -> [String] { + var scanner = PureYAMLStreaming.ResumableDocumentScanner() + var segments: [String] = [] + for chunk in scalarChunks(input, size: chunkSize) { + segments += try scanner.consume(chunk) + } + segments += try scanner.finish() + return segments +} + +/// All document sources the frozen line-based reference emits for `input` fed +/// in `chunkSize`-scalar chunks, in emission order. +func referenceSegments(_ input: String, chunkSize: Int) throws -> [String] { + var scanner = LineBasedReferenceScanner() + var segments: [String] = [] + for chunk in scalarChunks(input, size: chunkSize) { + segments += try scanner.consume(chunk) + } + if let final = try scanner.finish() { + segments.append(final) + } + return segments +} diff --git a/Tests/PureYAMLStreamingTests/DocumentScannerTests.swift b/Tests/PureYAMLStreamingTests/DocumentScannerTests.swift new file mode 100644 index 0000000..400dab1 --- /dev/null +++ b/Tests/PureYAMLStreamingTests/DocumentScannerTests.swift @@ -0,0 +1,162 @@ +import Foundation +import PureYAML +import Testing +@testable import PureYAMLStreaming + +@Suite("Resumable document scanner") +struct DocumentScannerTests { + /// Grammar-valid streams the scanner must split without loss for any + /// chunking. Each exercises a distinct boundary shape: explicit markers, + /// no markers, comments around markers, marker with trailing comment, + /// block scalar bodies, explicit document ends, empty documents, CRLF, + /// a non-newline break character, comment-only input, and empty input. + static let chunkingInputs: [String] = [ + "---\na: 1\n---\nb: 2\n", + "name: only\n", + "# note\n---\na: 1\n", + "---\na: 1\n# trailing\n---\nb: 2\n", + "a: 1\n--- # comment\nb: 2\n", + "a: |\n ---\n not a boundary\n---\nb: 2\n", + "a: 1\n...\nb: 2\n", + "---\na: 1\n...\n---\nb: 2\n", + "---\n---\n", + "a: 1\r\n---\r\nb: 2\r\n", + "a: \"x\u{2028} y\"\n---\nb: 2\n", + "# only comments\n\n", + "", + ] + + @Test("splits explicit documents at --- markers") + func splitsExplicitDocuments() throws { + let segments = try rewiredSegments("---\na: 1\n---\nb: 2\n", chunkSize: 1024) + #expect(segments == ["---\na: 1\n", "---\nb: 2\n"]) + } + + @Test("emits identical lossless segments for any chunking", arguments: chunkingInputs) + func chunkingDoesNotChangeSegments(input: String) throws { + let wholeFeed = try rewiredSegments(input, chunkSize: max(input.unicodeScalars.count, 1)) + #expect(wholeFeed.joined() == input) + for chunkSize in [1, 2, 3, 7, 64] { + let segments = try rewiredSegments(input, chunkSize: chunkSize) + #expect(segments == wholeFeed, "chunk size \(chunkSize) changed the segments for: \(input.debugDescription)") + } + } + + @Test("attaches leading directives to their document") + func attachesLeadingDirectives() throws { + let input = "%YAML 1.2\n---\na: 1\n" + #expect(try rewiredSegments(input, chunkSize: 3) == [input]) + + let documents = try PureYAMLStreaming.Parser(chunkSize: 3).collectDocuments(data: Data(input.utf8)) + #expect(documents.count == 1) + #expect(intValue("a", in: documents[0].value) == 1) + } + + @Test("splits before directives that follow an explicit document end") + func splitsBeforeDirectivesAfterDocumentEnd() throws { + let input = "---\na: 1\n...\n%YAML 1.2\n---\nb: 2\n" + let segments = try rewiredSegments(input, chunkSize: 5) + #expect(segments == ["---\na: 1\n...\n", "%YAML 1.2\n---\nb: 2\n"]) + + let documents = try PureYAMLStreaming.Parser(chunkSize: 5).collectDocuments(data: Data(input.utf8)) + #expect(documents.count == 2) + } + + @Test("keeps a directive without a preceding document end in the erroneous segment") + func keepsInvalidDirectiveInSegment() throws { + // Batch parsing rejects a directive that follows content without an + // explicit `...`, so the splitter must not carve out a boundary that + // would make both halves individually valid. + let input = "a: 1\n%YAML 1.2\n---\nb: 2\n" + #expect(throws: (any Error).self) { + try PureYAML.Parsing.Parser().parseStream(input) + } + + let segments = try rewiredSegments(input, chunkSize: 4) + #expect(segments == [input]) + #expect(throws: (any Error).self) { + try PureYAML.Parsing.Parser().parseStream(segments[0]) + } + } + + @Test("rejects a grammar-internal --- like batch parsing instead of splitting") + func rejectsDocumentMarkerInsideQuotedScalar() throws { + // The old line-based heuristic cut this quoted scalar in half. The + // grammar rejects a document indicator inside a quoted scalar, so the + // token-driven scanner must surface the same rejection batch parsing + // does, not fabricate a boundary. + let input = "a: \"x\n--- y\"\nb: 2\n" + #expect(throws: (any Error).self) { + try PureYAML.Parsing.Parser().parseStream(input) + } + #expect(throws: (any Error).self) { + _ = try rewiredSegments(input, chunkSize: 4) + } + } + + @Test("emits a boundary confirmable only at end of input") + func emitsBoundaryAtEndOfInput() throws { + // A trailing `---` line without a newline is a document start only if + // no fourth `-` follows, which end of input alone can confirm. The old + // scanner dropped the first document here. + var scanner = PureYAMLStreaming.ResumableDocumentScanner() + let consumed = try scanner.consume("a: 1\n---") + let finished = try scanner.finish() + #expect(consumed + finished == ["a: 1\n", "---"]) + + let documents = try PureYAMLStreaming.Parser(chunkSize: 3).collectDocuments(data: Data("a: 1\n---".utf8)) + #expect(documents.count == 2) + #expect(intValue("a", in: documents[0].value) == 1) + #expect(documents[1].value == .null) + } + + @Test("preserves CRLF line endings across boundaries") + func preservesCRLF() throws { + let segments = try rewiredSegments("a: 1\r\n---\r\nb: 2\r\n", chunkSize: 2) + #expect(segments == ["a: 1\r\n", "---\r\nb: 2\r\n"]) + } + + @Test("counts non-newline break characters like the token scanner") + func countsUnicodeBreaks() throws { + // U+2028 is a line break to the scanner's reader, so the boundary + // line's index must account for it when mapping back to text. The + // continuation after the break is indented, as the grammar requires + // for a multi-line quoted scalar. + let segments = try rewiredSegments("a: \"x\u{2028} y\"\n---\nb: 2\n", chunkSize: 3) + #expect(segments == ["a: \"x\u{2028} y\"\n", "---\nb: 2\n"]) + } + + @Test("lumps a bare document after an explicit end into one segment") + func lumpsBareDocumentAfterEnd() throws { + let input = "a: 1\n...\nb: 2\n" + #expect(try rewiredSegments(input, chunkSize: 4) == [input]) + + let documents = try PureYAMLStreaming.Parser(chunkSize: 4).collectDocuments(data: Data(input.utf8)) + #expect(documents.count == 2) + } + + @Test("returns nothing for empty input") + func returnsNothingForEmptyInput() throws { + var scanner = PureYAMLStreaming.ResumableDocumentScanner() + #expect(try scanner.consume("") == []) + #expect(try scanner.finish() == []) + } + + @Test("returns the raw remainder for comment-only input") + func returnsRemainderForCommentOnlyInput() throws { + let input = "# only\n\n" + #expect(try rewiredSegments(input, chunkSize: 2) == [input]) + + let documents = try PureYAMLStreaming.Parser(chunkSize: 2).collectDocuments(data: Data(input.utf8)) + #expect(documents.isEmpty) + } + + @Test("wraps document sources in streaming tokens") + func wrapsDocumentSourcesInStreamingTokens() throws { + var scanner = PureYAMLStreaming.ResumableDocumentScanner() + var tokens: [PureYAMLStreaming.StreamingToken] = [] + tokens += try scanner.consumeToken("---\na: 1\n---\n") + tokens += try scanner.finish().map(PureYAMLStreaming.StreamingToken.documentSource) + #expect(tokens == [.documentSource("---\na: 1\n"), .documentSource("---\n")]) + } +} diff --git a/Tests/PureYAMLStreamingTests/LineBasedReferenceScanner.swift b/Tests/PureYAMLStreamingTests/LineBasedReferenceScanner.swift new file mode 100644 index 0000000..5ab1358 --- /dev/null +++ b/Tests/PureYAMLStreamingTests/LineBasedReferenceScanner.swift @@ -0,0 +1,83 @@ +import Foundation + +/// The line-based document-boundary scanner this package shipped before the +/// token-driven rewiring, kept verbatim (minus the namespace) as a +/// differential reference. `DocumentScannerDifferentialTests` proves the +/// rewired scanner emits byte-identical segment lists on every input where +/// this heuristic was correct, so the reference must stay frozen; do not fix +/// its known defects (misfiring on grammar-internal `---` lines, splitting +/// directives away from their document, and dropping the previous document +/// when the input ends on a `---` line without a trailing newline). +struct LineBasedReferenceScanner { + private var pendingLine = "" + private var currentDocument = "" + private var hasCurrentDocumentContent = false + private var line = 1 + + init() {} + + mutating func consume(_ chunk: String) throws -> [String] { + var documents: [String] = [] + pendingLine.append(chunk) + + while let newlineRange = pendingLine.rangeOfCharacter(from: .newlines) { + let lineText = String(pendingLine[.. String? { + if !pendingLine.isEmpty { + var ignored: [String] = [] + try consumeLine(pendingLine, documents: &ignored) + pendingLine.removeAll(keepingCapacity: true) + } + defer { + currentDocument.removeAll(keepingCapacity: true) + hasCurrentDocumentContent = false + } + return currentDocument.isEmpty ? nil : currentDocument + } + + private mutating func consumeLine( + _ sourceLine: String, + documents: inout [String] + ) throws { + if isDocumentStart(sourceLine), hasCurrentDocumentContent { + documents.append(currentDocument) + currentDocument = sourceLine + hasCurrentDocumentContent = true + return + } + + currentDocument.append(sourceLine) + if hasSemanticContent(sourceLine) { + hasCurrentDocumentContent = true + } + } + + private func isDocumentStart(_ sourceLine: String) -> Bool { + let trimmedNewline = sourceLine.trimmingCharacters(in: .newlines) + guard trimmedNewline.hasPrefix("---") else { + return false + } + let markerEnd = trimmedNewline.index(trimmedNewline.startIndex, offsetBy: 3) + if markerEnd != trimmedNewline.endIndex { + let next = trimmedNewline[markerEnd] + guard next == " " || next == "\t" || next == "#" else { + return false + } + } + return trimmedNewline.prefix(3) == "---" + } + + private func hasSemanticContent(_ sourceLine: String) -> Bool { + let trimmed = sourceLine.trimmingCharacters(in: .whitespacesAndNewlines) + return !trimmed.isEmpty && !trimmed.hasPrefix("#") + } +} diff --git a/Tests/PureYAMLStreamingTests/StreamingTests.swift b/Tests/PureYAMLStreamingTests/StreamingTests.swift index 7fa5406..7fe6686 100644 --- a/Tests/PureYAMLStreamingTests/StreamingTests.swift +++ b/Tests/PureYAMLStreamingTests/StreamingTests.swift @@ -1,7 +1,7 @@ import Foundation import PureYAML -@testable import PureYAMLStreaming import Testing +@testable import PureYAMLStreaming @Suite("PureYAML Streaming") struct StreamingTests { @@ -9,7 +9,7 @@ struct StreamingTests { func chunkedReaderPreservesSplitUTF8() throws { var reader = try PureYAMLStreaming.ChunkedUTF8Reader( data: Data("aé\nb".utf8), - chunkSize: 2, + chunkSize: 2 ) var chunks: [String] = [] @@ -28,9 +28,7 @@ struct StreamingTests { documents += try scanner.consume("---\na: 1\n--") documents += try scanner.consume("-\nb: 2\n") - if let final = try scanner.finish() { - documents.append(final) - } + documents += try scanner.finish() #expect(documents == ["---\na: 1\n", "---\nb: 2\n"]) } @@ -43,7 +41,7 @@ struct StreamingTests { name: one --- name: two - """.utf8, + """.utf8 ) let parser = PureYAMLStreaming.Parser(chunkSize: 5) let documents = try parser.collectDocuments(data: yaml) @@ -64,7 +62,7 @@ struct StreamingTests { let files = try FileManager.default.contentsOfDirectory( at: fixtureRoot, - includingPropertiesForKeys: nil, + includingPropertiesForKeys: nil ).filter { url in url.pathExtension == "yaml" || url.pathExtension == "yml" } @@ -88,7 +86,7 @@ struct StreamingTests { func parserStreamsGeneratedLargeFileWithoutCollectingDocuments() throws { let fileURL = FileManager.default.temporaryDirectory .appendingPathComponent("pureyaml-stream-\(UUID().uuidString).yaml") - try writeGeneratedStream(to: fileURL, documentCount: 5_000) + try writeGeneratedStream(to: fileURL, documentCount: 5000) defer { try? FileManager.default.removeItem(at: fileURL) } @@ -108,7 +106,7 @@ struct StreamingTests { count += 1 } - #expect(count == 5_000) + #expect(count == 5000) #expect(firstName == "doc-0") #expect(lastName == "doc-4999") } diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index c35098e..369463b 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -4,8 +4,10 @@ PureYAMLStreaming separates three concerns: 1. `ChunkedUTF8Reader` pulls bounded byte chunks from a file or data source and tracks byte offsets plus line and column marks. -2. `ResumableDocumentScanner` consumes chunks, detects top-level YAML document - boundaries, and emits one document source at a time. +2. `ResumableDocumentScanner` consumes chunks, tokenizes them with + `PureYAML.Streaming.Scanner`, and emits one document source at a time, cut + immediately before each `documentStart` token or before the first directive + token that follows an explicit `...` document end. 3. `Parser` parses each emitted document through PureYAML and hands completed documents to the caller. @@ -17,22 +19,38 @@ flowchart TD pureParser --> documentSink["Document Sink"] ``` -## Why Document Streaming First +## Token-Driven Boundary Detection -PureYAML's internal parser already avoids retaining the full token and event -arrays, but the public boundary is still a complete `String`. A fully resumable -YAML token scanner needs careful handling for quoted scalars, block scalars, -comments, indentation, CRLF, Unicode, tags, anchors, and merge keys. +PureYAML ships a resumable token scanner as public API +(`PureYAML.Streaming.Scanner`), proven byte-for-byte identical to batch +scanning across the yaml-test-suite corpus at multiple chunk sizes. +`ResumableDocumentScanner` drives it token by token and cuts the buffered +source text at grammar-confirmed boundaries, so a `---` inside a quoted +scalar can never misfire as a document split and directives stay attached to +the document they introduce. Chunk boundaries may fall anywhere, including +mid-token; the token scanner suspends and resumes across them. -Document streaming is the safe first milestone: it proves file/chunk IO, -document-at-a-time memory behavior, and API ergonomics without forking the YAML -grammar implementation. +Two implementation details keep the contract tight: + +- After each cut the token scanner is restarted on the remaining buffer. A + document boundary fully resets scanner state, and the restart bounds + retained state by the current document rather than the whole stream (the + scanner's reader keeps everything it was ever fed). +- A trailing CR is withheld from the token scanner while more input may still + arrive, so it never sees a CR LF pair split at a suspension point. Split + exactly there, the persisted plain-scalar suspension counts the pair as two + line breaks and every later token mark drifts down a line (observed at + PureYAML commit `2fcf546`). Withholding the CR keeps the token line numbers + the cuts rely on identical to batch scanning. + +Parsing stays document-at-a-time: each emitted segment goes through the +stable batch `PureYAML` entry point, which keeps the parser semantics +identical to non-streaming use. ## Future Event Streaming -The next scanner milestone should replace `ResumableDocumentScanner` with a -token-level scanner that can suspend and resume inside any YAML token. That -scanner can then feed a public event sink directly. +The next milestone can feed token-level events to a public event sink +directly instead of emitting whole document sources. The intended final shape is: