NodetteHQ/PureYAMLStreaming
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

fix: drive document boundary scanning off PureYAML's token scanner

Browse Source 
Replace the line-based --- heuristic with a scan over PureYAML's
resumable token scanner, so a --- inside a quoted scalar can no longer
misfire as a document boundary and directives stay attached to the
document they introduce. Add a differential suite that checks the new
scanner against the old line-based reference across generated streams
and the full Geekbench corpus.
This commit is contained in:
Mihaela Mihaljevic 2026-07-26 20:59:39 +02:00
parent cca20d26f7
commit b4253fd289
16 changed files with 645 additions and 128 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
Package.resolved
View File

@ -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"
}
},
{

12
Package.swift
View File

@ -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"]
),
],
]
)

21
README.md
View File

@ -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

16
Sources/PureYAMLStreaming/ChunkedUTF8Reader.swift
View File

@ -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)

16
Sources/PureYAMLStreaming/Parser.swift
View File

@ -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 {

216
Sources/PureYAMLStreaming/ResumableDocumentScanner.swift
View File

@ -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[..<newlineRange.lowerBound])
let newline = String(pendingLine[newlineRange])
pendingLine.removeSubrange(..<newlineRange.upperBound)
try consumeLine(lineText + newline, documents: &documents)
line += 1
}
return documents
buffer.append(contentsOf: chunk.unicodeScalars)
feedScanner()
return try drainTokens()
}
/// Like `consume(_:)`, but wraps each completed document source in a
/// ``StreamingToken``.
public mutating func consumeToken(_ chunk: String) throws -> [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[..<offset]))
buffer.removeFirst(offset)
segmentHasContent = false
inDirectivePreamble = false
afterDocumentEnd = false
scanner = PureYAML.Streaming.Scanner()
scannerFedCount = 0
feedScanner()
if isFinal {
scanner.finish()
}
return segment
}
/// The scanner reader's break set (libyaml's `IS_BREAK`).
private static func isBreak(_ scalar: Unicode.Scalar) -> Bool {
switch scalar {
case "\r", "\n", "\u{85}", "\u{2028}", "\u{2029}": true
default: false
}
}
}
}

4
Sources/PureYAMLStreaming/StreamingError.swift
View File

@ -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)

4
Sources/PureYAMLStreaming/StreamingMark.swift
View File

@ -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

4
Sources/PureYAMLStreaming/StreamingToken.swift
View File

@ -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)
}
}

18
Sources/PureYAMLStreamingWasmSmoke/main.swift
View File

@ -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: Encodable>(_ 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)
}

99
Tests/PureYAMLStreamingTests/DocumentScannerDifferentialTests.swift Normal file
View File

@ -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)")
}
}
}
}

54
Tests/PureYAMLStreamingTests/DocumentScannerTestSupport.swift Normal file
View File

@ -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
}

162
Tests/PureYAMLStreamingTests/DocumentScannerTests.swift Normal file
View File

@ -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")])
}
}

83
Tests/PureYAMLStreamingTests/LineBasedReferenceScanner.swift Normal file
View File

@ -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[..<newlineRange.lowerBound])
let newline = String(pendingLine[newlineRange])
pendingLine.removeSubrange(..<newlineRange.upperBound)
try consumeLine(lineText + newline, documents: &documents)
line += 1
}
return documents
}
mutating func finish() throws -> 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("#")
}
}

16
Tests/PureYAMLStreamingTests/StreamingTests.swift
View File

@ -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")
}

44
docs/ARCHITECTURE.md
View File

@ -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: