Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948//! Streaming parser for Jetstream V2 operator timestamp imports.//!//! The source is an uncompressed RFC 4180 CSV so later phases can seek back//! to an accepted row by byte offset. Parsing is forward-only and bounded by//! `max_row_bytes`; strings in `Row` and `Reject` borrow parser scratch and are//! valid only for the duration of their callback.
const std = @import("std");const zat = @import("zat");
pub const max_row_bytes = 64 * 1024;pub const default_reject_sample_limit: usize = 100;pub const max_diagnostic_value_len: usize = 256;
pub const Scope = enum(u8) { all_versions = 0, specific_version = 1,};
/// Atmos' CID representation is deliberately fixed to the formats accepted/// by upstream: CIDv1, dag-cbor/raw, SHA-256. This also gives the rule store a/// stable allocation-free key representation.pub const Cid = struct { codec: u8, hash: [32]u8,
pub fn eql(a: Cid, b: Cid) bool { return a.codec == b.codec and std.mem.eql(u8, &a.hash, &b.hash); }};
pub const Row = struct { offset: u64, did: []const u8, collection: []const u8, rkey: []const u8, scope: Scope, cid: ?Cid, timestamp_micros: i64,};
pub const RejectReason = enum { malformed_csv, missing_field, bad_uri, uri_authority_not_did, uri_missing_collection_or_rkey, bad_timestamp, non_positive_timestamp, unknown_scope, specific_version_missing_cid, bad_cid, row_too_long,};
pub const Reject = struct { row_number: u64, offset: u64, reason: RejectReason, field: []const u8, value: []const u8,};
pub const OwnedReject = struct { row_number: u64, offset: u64, reason: RejectReason, field: []u8, value: []u8,};
pub const Stats = struct { rows_total: u64 = 0, rows_valid: u64 = 0, rows_rejected: u64 = 0, rejects_by_reason: [@typeInfo(RejectReason).@"enum".field_names.len]u64 = @splat(0), reject_sample: std.ArrayList(OwnedReject) = .empty,
pub fn rejected(self: *const Stats, reason: RejectReason) u64 { return self.rejects_by_reason[@backingInt(reason)]; }
pub fn deinit(self: *Stats, allocator: std.mem.Allocator) void { for (self.reject_sample.items) |item| { allocator.free(item.field); allocator.free(item.value); } self.reject_sample.deinit(allocator); self.* = undefined; }};
pub const Options = struct { context: ?*anyopaque, on_row: *const fn (?*anyopaque, Row) anyerror!void, on_reject: ?*const fn (?*anyopaque, Reject) void = null, /// Zero selects the upstream default; `no_reject_sample` disables it. reject_sample_limit: usize = 0,
pub const no_reject_sample = std.math.maxInt(usize);};
const Columns = struct { uri: u3, timestamp: u3, scope: ?u3, cid: ?u3, count: u3,};
const Record = struct { bytes: []const u8, offset: u64, footprint: u64, too_long: bool,};
const FieldSet = struct { values: [5][]const u8 = @splat(""), count: usize = 0,};
/// Positioned Phase-C reader for offsets produced by the streaming parser./// It owns one file handle and private scratch, so separate instances can be/// used concurrently without sharing a seek cursor. Every read independently/// verifies the row boundary, 64 KiB contract, CSV shape, and field semantics.pub const RowReader = struct { io: std.Io, file: std.Io.File, size: u64, columns: Columns, raw: [max_row_bytes]u8 = undefined, decoded: [max_row_bytes]u8 = undefined,
pub fn open(io: std.Io, path: []const u8) !RowReader { const file = try std.Io.Dir.cwd().openFile(io, path, .{}); errdefer file.close(io); const stat = try file.stat(io); var self: RowReader = .{ .io = io, .file = file, .size = stat.size, .columns = undefined, }; const count = try file.readPositionalAll(io, &self.raw, 0); var reader: std.Io.Reader = .fixed(self.raw[0..count]); var position: u64 = 0; const record = (try readRecord(&reader, &self.raw, &position)) orelse return error.InvalidHeader; if (record.too_long or record.footprint >= max_row_bytes) return error.InvalidHeader; self.columns = parseHeader(decodeRecord(record.bytes, &self.decoded) catch return error.InvalidHeader) catch return error.InvalidHeader; return self; }
pub fn close(self: *RowReader) void { self.file.close(self.io); self.* = undefined; }
pub fn read(self: *RowReader, offset: u64) !Row { if (offset == 0 or offset >= self.size) return error.CorruptOffset; var previous: [1]u8 = undefined; if (try self.file.readPositionalAll(self.io, &previous, offset - 1) != 1 or previous[0] != '\n') return error.CorruptOffset;
const span: usize = @intCast(@min(self.size - offset, max_row_bytes)); const count = try self.file.readPositionalAll(self.io, self.raw[0..span], offset); var reader: std.Io.Reader = .fixed(self.raw[0..count]); var position = offset; const record = (try readRecord(&reader, &self.raw, &position)) orelse return error.CorruptOffset; if (record.too_long or record.footprint >= max_row_bytes) return error.CorruptOffset; const fields = decodeRecord(record.bytes, &self.decoded) catch return error.CorruptOffset; if (fields.count != self.columns.count) return error.CorruptOffset; return switch (validate(self.columns, fields, offset)) { .row => |row| row, .reject => error.CorruptOffset, }; }};
/// Parse a timestamp import in one forward-only pass. Only structural header/// errors, quote corruption, read failures, allocation failures, and the row/// callback abort the import. Ordinary data errors are counted and skipped.pub fn parse( allocator: std.mem.Allocator, reader: *std.Io.Reader, opts: Options,) !Stats { var stats: Stats = .{}; parseInto(allocator, reader, opts, &stats) catch |err| { stats.deinit(allocator); return err; }; return stats;}
/// Variant used by the durable importer so partial counts remain available/// when a structural error terminates the file. `stats` is initialized before/// any read and always belongs to the caller, on success or error.pub fn parseInto( allocator: std.mem.Allocator, reader: *std.Io.Reader, opts: Options, stats: *Stats,) !void { stats.* = .{};
var raw: [max_row_bytes]u8 = undefined; var decoded: [max_row_bytes]u8 = undefined; var input_offset: u64 = 0;
const header_record = (try readRecord(reader, &raw, &input_offset)) orelse return error.InvalidHeader; if (header_record.too_long) return error.InvalidHeader; const header_fields = decodeRecord(header_record.bytes, &decoded) catch return error.InvalidHeader; const columns = parseHeader(header_fields) catch return error.InvalidHeader;
var row_number: u64 = 0; while (try readRecord(reader, &raw, &input_offset)) |record| { row_number += 1; if (record.too_long or record.footprint >= max_row_bytes) { try noteReject(allocator, stats, opts, .{ .row_number = row_number, .offset = record.offset, .reason = .row_too_long, .field = "", .value = "row reaches or exceeds the 64 KiB limit", }); continue; }
const fields = decodeRecord(record.bytes, &decoded) catch |err| switch (err) { error.MalformedCsv => { try noteReject(allocator, stats, opts, .{ .row_number = row_number, .offset = record.offset, .reason = .malformed_csv, .field = "", .value = record.bytes, }); continue; }, }; if (fields.count != columns.count) { try noteReject(allocator, stats, opts, .{ .row_number = row_number, .offset = record.offset, .reason = .missing_field, .field = "", .value = record.bytes, }); continue; }
const result = validate(columns, fields, record.offset); switch (result) { .reject => |reject| try noteReject(allocator, stats, opts, .{ .row_number = row_number, .offset = record.offset, .reason = reject.reason, .field = reject.field, .value = reject.value, }), .row => |row| { stats.rows_total += 1; stats.rows_valid += 1; try opts.on_row(opts.context, row); }, } }}
const ValidationReject = struct { reason: RejectReason, field: []const u8, value: []const u8,};
const Validation = union(enum) { row: Row, reject: ValidationReject,};
fn validate(columns: Columns, fields: FieldSet, offset: u64) Validation { const uri_text = trim(fields.values[columns.uri]); if (uri_text.len == 0) return rejected(.missing_field, "uri", ""); const timestamp_text = trim(fields.values[columns.timestamp]); if (timestamp_text.len == 0) return rejected(.missing_field, "timestamp", "");
const uri = zat.AtUri.parse(uri_text) orelse return rejected(.bad_uri, "uri", uri_text); const did = zat.Did.parse(uri.authority()) orelse return rejected(.uri_authority_not_did, "uri", uri_text); const collection = uri.collection() orelse return rejected(.uri_missing_collection_or_rkey, "uri", uri_text); const rkey = uri.rkey() orelse return rejected(.uri_missing_collection_or_rkey, "uri", uri_text);
const timestamp_micros = parseRfc3339Micros(timestamp_text) orelse return rejected(.bad_timestamp, "timestamp", timestamp_text); if (timestamp_micros <= 0) return rejected(.non_positive_timestamp, "timestamp", timestamp_text);
var scope: Scope = .all_versions; if (columns.scope) |index| { const value = trim(fields.values[index]); if (value.len != 0) { if (std.mem.eql(u8, value, "all_versions")) { scope = .all_versions; } else if (std.mem.eql(u8, value, "specific_version")) { scope = .specific_version; } else { return rejected(.unknown_scope, "scope", value); } } }
var cid: ?Cid = null; if (scope == .specific_version) { const cid_text = if (columns.cid) |index| trim(fields.values[index]) else ""; if (cid_text.len == 0) return rejected(.specific_version_missing_cid, "cid", ""); cid = parseCid(cid_text) orelse return rejected(.bad_cid, "cid", cid_text); }
return .{ .row = .{ .offset = offset, .did = did.str(), .collection = collection, .rkey = rkey, .scope = scope, .cid = cid, .timestamp_micros = timestamp_micros, } };}
fn rejected(reason: RejectReason, field: []const u8, value: []const u8) Validation { return .{ .reject = .{ .reason = reason, .field = field, .value = value } };}
fn noteReject( allocator: std.mem.Allocator, stats: *Stats, opts: Options, reject: Reject,) !void { var diagnostic_buffer: [max_diagnostic_value_len + 3]u8 = undefined; const value = truncateDiagnostic(reject.value, &diagnostic_buffer); const bounded: Reject = .{ .row_number = reject.row_number, .offset = reject.offset, .reason = reject.reason, .field = reject.field, .value = value, }; stats.rows_total += 1; stats.rows_rejected += 1; stats.rejects_by_reason[@backingInt(reject.reason)] += 1; if (opts.on_reject) |callback| callback(opts.context, bounded);
const limit = if (opts.reject_sample_limit == 0) default_reject_sample_limit else if (opts.reject_sample_limit == Options.no_reject_sample) 0 else opts.reject_sample_limit; if (stats.reject_sample.items.len >= limit) return;
const field = try allocator.dupe(u8, bounded.field); errdefer allocator.free(field); const owned_value = try allocator.dupe(u8, bounded.value); errdefer allocator.free(owned_value); try stats.reject_sample.append(allocator, .{ .row_number = bounded.row_number, .offset = bounded.offset, .reason = bounded.reason, .field = field, .value = owned_value, });}
fn parseHeader(fields: FieldSet) !Columns { if (fields.count == 0 or fields.count > 4) return error.InvalidHeader; var uri: ?u3 = null; var timestamp: ?u3 = null; var scope: ?u3 = null; var cid: ?u3 = null;
for (fields.values[0..fields.count], 0..) |raw, index| { const name = trim(raw); const slot: *?u3 = if (std.ascii.eqlIgnoreCase(name, "uri")) &uri else if (std.ascii.eqlIgnoreCase(name, "timestamp")) ×tamp else if (std.ascii.eqlIgnoreCase(name, "scope")) &scope else if (std.ascii.eqlIgnoreCase(name, "cid")) &cid else return error.InvalidHeader; if (slot.* != null) return error.InvalidHeader; slot.* = @intCast(index); } return .{ .uri = uri orelse return error.InvalidHeader, .timestamp = timestamp orelse return error.InvalidHeader, .scope = scope, .cid = cid, .count = @intCast(fields.count), };}
/// Read one RFC 4180 logical record. Blank physical lines are skipped exactly/// like Go's encoding/csv; the returned offset remains before skipped lines so/// seeking there and decoding again still yields the same row.fn readRecord(reader: *std.Io.Reader, storage: []u8, input_offset: *u64) !?Record { const State = enum { field_start, unquoted, quoted, after_quote, after_quote_cr }; const offset = input_offset.*; var state: State = .field_start; var length: usize = 0; var footprint: u64 = 0; var too_long = false;
while (true) { const byte = reader.takeByte() catch |err| switch (err) { error.EndOfStream => { if (state == .quoted) return error.UnbalancedQuote; if (state == .after_quote_cr) return error.UnbalancedQuote; if (footprint == 0) return null; if (isBlankRecord(storage[0..length])) return null; return .{ .bytes = storage[0..length], .offset = offset, .footprint = footprint, .too_long = too_long, }; }, error.ReadFailed => return error.ReadFailed, }; input_offset.* += 1; footprint += 1; if (length < storage.len) { storage[length] = byte; length += 1; } else { too_long = true; }
switch (state) { .field_start => switch (byte) { '"' => state = .quoted, ',' => {}, '\n' => { if (isBlankRecord(storage[0..length])) { length = 0; state = .field_start; continue; } return .{ .bytes = storage[0..length], .offset = offset, .footprint = footprint, .too_long = too_long }; }, else => state = .unquoted, }, .unquoted => switch (byte) { '"' => return error.UnbalancedQuote, ',' => state = .field_start, '\n' => { if (isBlankRecord(storage[0..length])) { length = 0; state = .field_start; continue; } return .{ .bytes = storage[0..length], .offset = offset, .footprint = footprint, .too_long = too_long }; }, else => {}, }, .quoted => { if (byte == '"') state = .after_quote; }, .after_quote => switch (byte) { '"' => state = .quoted, ',' => state = .field_start, '\r' => state = .after_quote_cr, '\n' => return .{ .bytes = storage[0..length], .offset = offset, .footprint = footprint, .too_long = too_long }, else => return error.UnbalancedQuote, }, .after_quote_cr => if (byte == '\n') return .{ .bytes = storage[0..length], .offset = offset, .footprint = footprint, .too_long = too_long } else return error.UnbalancedQuote, } }}
fn isBlankRecord(bytes: []const u8) bool { return bytes.len == 0 or std.mem.eql(u8, bytes, "\n") or std.mem.eql(u8, bytes, "\r\n");}
/// Decode a structurally checked record into at most five fields. Five is/// enough to prove an invalid field count because the schema has at most four.fn decodeRecord(raw_record: []const u8, storage: []u8) error{MalformedCsv}!FieldSet { var raw = raw_record; if (raw.len > 0 and raw[raw.len - 1] == '\n') raw = raw[0 .. raw.len - 1]; if (raw.len > 0 and raw[raw.len - 1] == '\r') raw = raw[0 .. raw.len - 1];
var result: FieldSet = .{}; var in_pos: usize = 0; var out_pos: usize = 0; while (true) { const field_start = out_pos; if (in_pos < raw.len and raw[in_pos] == '"') { in_pos += 1; while (true) { if (in_pos >= raw.len) return error.MalformedCsv; const byte = raw[in_pos]; if (byte == '"') { if (in_pos + 1 < raw.len and raw[in_pos + 1] == '"') { storage[out_pos] = '"'; out_pos += 1; in_pos += 2; continue; } in_pos += 1; break; } // encoding/csv normalizes CRLF within quoted fields to LF. if (byte == '\r' and in_pos + 1 < raw.len and raw[in_pos + 1] == '\n') { in_pos += 1; continue; } storage[out_pos] = byte; out_pos += 1; in_pos += 1; } if (in_pos < raw.len and raw[in_pos] != ',') return error.MalformedCsv; } else { while (in_pos < raw.len and raw[in_pos] != ',') : (in_pos += 1) { if (raw[in_pos] == '"') return error.MalformedCsv; storage[out_pos] = raw[in_pos]; out_pos += 1; } }
if (result.count < result.values.len) result.values[result.count] = storage[field_start..out_pos]; result.count += 1; if (in_pos >= raw.len) break; in_pos += 1; // comma } return result;}
fn parseCid(text: []const u8) ?Cid { if (text.len < 2 or text[0] != 'b') return null; var scratch: [128]u8 = undefined; var fba = std.heap.FixedBufferAllocator.init(&scratch); const raw = zat.multibase.base32lower.decode(fba.allocator(), text[1..]) catch return null; // Atmos accepts only minimally encoded CIDv1, dag-cbor/raw, SHA-256/32. if (raw.len != 36 or raw[0] != 1 or (raw[1] != 0x71 and raw[1] != 0x55) or raw[2] != 0x12 or raw[3] != 0x20) return null; var hash: [32]u8 = undefined; @memcpy(&hash, raw[4..36]); return .{ .codec = raw[1], .hash = hash };}
/// Go time.Parse(time.RFC3339) semantics needed by the import format: fixed/// four-digit civil date, seconds, optional 1-9 digit fraction, and Z or a/// numeric offset. The result truncates nanoseconds to Unix microseconds.fn parseRfc3339Micros(text: []const u8) ?i64 { if (text.len < 20) return null; if (text[4] != '-' or text[7] != '-' or text[10] != 'T' or text[13] != ':' or text[16] != ':') return null; const year = parseDigits(text[0..4]) orelse return null; const month = parseDigits(text[5..7]) orelse return null; const day = parseDigits(text[8..10]) orelse return null; const hour = parseDigits(text[11..13]) orelse return null; const minute = parseDigits(text[14..16]) orelse return null; const second = parseDigits(text[17..19]) orelse return null; if (month < 1 or month > 12 or day < 1 or day > daysInMonth(year, month) or hour > 23 or minute > 59 or second > 59) return null;
var pos: usize = 19; var micros_fraction: i64 = 0; if (pos < text.len and (text[pos] == '.' or text[pos] == ',')) { pos += 1; const start = pos; var digits: usize = 0; while (pos < text.len and std.ascii.isDigit(text[pos])) : (pos += 1) { digits += 1; if (digits <= 6) micros_fraction = micros_fraction * 10 + (text[pos] - '0'); } if (pos == start) return null; while (digits < 6) : (digits += 1) micros_fraction *= 10; }
var offset_seconds: i64 = 0; if (pos < text.len and text[pos] == 'Z') { pos += 1; } else { if (pos + 6 != text.len or (text[pos] != '+' and text[pos] != '-') or text[pos + 3] != ':') return null; const zone_hour = parseDigits(text[pos + 1 .. pos + 3]) orelse return null; const zone_minute = parseDigits(text[pos + 4 .. pos + 6]) orelse return null; if (zone_hour > 24 or zone_minute > 60) return null; offset_seconds = @as(i64, zone_hour) * 3600 + @as(i64, zone_minute) * 60; if (text[pos] == '-') offset_seconds = -offset_seconds; pos += 6; } if (pos != text.len) return null;
const days = daysFromCivil(@intCast(year), @intCast(month), @intCast(day)); const seconds = days * std.time.s_per_day + @as(i64, hour) * 3600 + @as(i64, minute) * 60 + @as(i64, second) - offset_seconds; const whole_micros = std.math.mul(i64, seconds, std.time.us_per_s) catch return null; return std.math.add(i64, whole_micros, micros_fraction) catch return null;}
fn parseDigits(text: []const u8) ?u16 { if (text.len == 0) return null; var value: u16 = 0; for (text) |byte| { if (!std.ascii.isDigit(byte)) return null; value = value * 10 + (byte - '0'); } return value;}
fn daysInMonth(year: u16, month: u16) u16 { return switch (month) { 1, 3, 5, 7, 8, 10, 12 => 31, 4, 6, 9, 11 => 30, 2 => if (@mod(year, 4) == 0 and (@mod(year, 100) != 0 or @mod(year, 400) == 0)) 29 else 28, else => 0, };}
// Howard Hinnant's civil-date transform, relative to 1970-01-01.fn daysFromCivil(year_in: i64, month_in: i64, day: i64) i64 { var year = year_in; if (month_in <= 2) year -= 1; const era = @divFloor(year, 400); const yoe = year - era * 400; const month = month_in + (if (month_in > 2) @as(i64, -3) else 9); const doy = @divFloor(153 * month + 2, 5) + day - 1; const doe = yoe * 365 + @divFloor(yoe, 4) - @divFloor(yoe, 100) + doy; return era * 146097 + doe - 719468;}
fn trim(value: []const u8) []const u8 { return std.mem.trim(u8, value, " \t\r\n\x0b\x0c");}
fn truncateDiagnostic(value: []const u8, buffer: *[max_diagnostic_value_len + 3]u8) []const u8 { if (value.len <= max_diagnostic_value_len) return value; @memcpy(buffer[0..max_diagnostic_value_len], value[0..max_diagnostic_value_len]); @memcpy(buffer[max_diagnostic_value_len..], "…"); return buffer;}
const Collector = struct { const TestRow = struct { offset: u64, did: []u8, collection: []u8, rkey: []u8, scope: Scope, cid: ?Cid, timestamp_micros: i64, };
rows: std.ArrayList(TestRow) = .empty, rejects_seen: usize = 0,
fn onRow(raw: ?*anyopaque, row: Row) !void { const self: *Collector = @ptrCast(@alignCast(raw.?)); const allocator = std.testing.allocator; const did = try allocator.dupe(u8, row.did); errdefer allocator.free(did); const collection = try allocator.dupe(u8, row.collection); errdefer allocator.free(collection); const rkey = try allocator.dupe(u8, row.rkey); errdefer allocator.free(rkey); try self.rows.append(allocator, .{ .offset = row.offset, .did = did, .collection = collection, .rkey = rkey, .scope = row.scope, .cid = row.cid, .timestamp_micros = row.timestamp_micros, }); }
fn onReject(raw: ?*anyopaque, _: Reject) void { const self: *Collector = @ptrCast(@alignCast(raw.?)); self.rejects_seen += 1; }
fn deinit(self: *Collector) void { for (self.rows.items) |row| { std.testing.allocator.free(row.did); std.testing.allocator.free(row.collection); std.testing.allocator.free(row.rkey); } self.rows.deinit(std.testing.allocator); }};
fn collect(source: []const u8, collector: *Collector, sample_limit: usize) !Stats { var reader: std.Io.Reader = .fixed(source); return parse(std.testing.allocator, &reader, .{ .context = collector, .on_row = Collector.onRow, .on_reject = Collector.onReject, .reject_sample_limit = sample_limit, });}
test "valid rows preserve offsets, defaults, quoted fields, and exact CID" { const testing = std.testing; const valid_cid = "bafyreigbtj4x7ip5legnfznufuopl4sg4knzc2cof6duas4b3q2fy6swua"; const row_one = "at://did:plc:alice/app.bsky.feed.post/r1,2022-01-02T03:04:05Z,,\r\n"; const source = "uri,timestamp,scope,cid\r\n" ++ row_one ++ "\"at://did:plc:bob/app.bsky.feed.like/r2\",2023-06-07T10:09:10+02:00,specific_version," ++ valid_cid ++ "\r\n"; var collector: Collector = .{}; defer collector.deinit(); var stats = try collect(source, &collector, 0); defer stats.deinit(testing.allocator);
try testing.expectEqual(@as(u64, 2), stats.rows_valid); try testing.expectEqual(@as(usize, 2), collector.rows.items.len); const first = collector.rows.items[0]; try testing.expectEqualStrings("did:plc:alice", first.did); try testing.expectEqual(@as(i64, 1_641_092_645_000_000), first.timestamp_micros); try testing.expectEqual("uri,timestamp,scope,cid\r\n".len, first.offset); try testing.expect(std.mem.startsWith(u8, source[first.offset..], row_one)); try testing.expectEqual(Scope.all_versions, first.scope); try testing.expectEqual(null, first.cid); const second = collector.rows.items[1]; try testing.expectEqual(@as(i64, 1_686_125_350_000_000), second.timestamp_micros); try testing.expectEqual(Scope.specific_version, second.scope); try testing.expectEqual(@as(u8, 0x71), second.cid.?.codec);}
test "bad data rows are exhaustively rejected without aborting later rows" { const testing = std.testing; const source = "uri,timestamp,scope,cid\n" ++ "at://did:plc:ok/app.bsky.feed.post/r1,2022-01-02T03:04:05Z,,\n" ++ "not-a-uri,2022-01-02T03:04:05Z,,\n" ++ "at://alice.example/app.bsky.feed.post/r,2022-01-02T03:04:05Z,,\n" ++ "at://did:plc:pathless,2022-01-02T03:04:05Z,,\n" ++ "at://did:plc:time/app.bsky.feed.post/r,not-a-time,,\n" ++ "at://did:plc:epoch/app.bsky.feed.post/r,1970-01-01T00:00:00Z,,\n" ++ "at://did:plc:scope/app.bsky.feed.post/r,2022-01-02T03:04:05Z,bogus,\n" ++ ",2022-01-02T03:04:05Z,,\n" ++ "at://did:plc:missing/app.bsky.feed.post/r,,,\n" ++ "at://did:plc:cid/app.bsky.feed.post/r,2022-01-02T03:04:05Z,specific_version,\n" ++ "at://did:plc:badcid/app.bsky.feed.post/r,2022-01-02T03:04:05Z,specific_version,nope\n" ++ "at://did:plc:fields/app.bsky.feed.post/r,2022-01-02T03:04:05Z\n" ++ "at://did:plc:later/app.bsky.feed.post/r2,2023-01-02T03:04:05Z,,\n"; var collector: Collector = .{}; defer collector.deinit(); var stats = try collect(source, &collector, 4); defer stats.deinit(testing.allocator);
try testing.expectEqual(@as(u64, 13), stats.rows_total); try testing.expectEqual(@as(u64, 2), stats.rows_valid); try testing.expectEqual(@as(u64, 11), stats.rows_rejected); try testing.expectEqual(@as(usize, 11), collector.rejects_seen); try testing.expectEqual(@as(usize, 4), stats.reject_sample.items.len); try testing.expectEqual(@as(u64, 3), stats.rejected(.missing_field)); try testing.expectEqual(@as(u64, 1), stats.rejected(.bad_uri)); try testing.expectEqual(@as(u64, 1), stats.rejected(.uri_authority_not_did)); try testing.expectEqual(@as(u64, 1), stats.rejected(.uri_missing_collection_or_rkey)); try testing.expectEqual(@as(u64, 1), stats.rejected(.bad_timestamp)); try testing.expectEqual(@as(u64, 1), stats.rejected(.non_positive_timestamp)); try testing.expectEqual(@as(u64, 1), stats.rejected(.unknown_scope)); try testing.expectEqual(@as(u64, 1), stats.rejected(.specific_version_missing_cid)); try testing.expectEqual(@as(u64, 1), stats.rejected(.bad_cid)); try testing.expectEqualStrings("did:plc:later", collector.rows.items[1].did);}
test "header is strict, order independent, and case insensitive" { const testing = std.testing; const invalid = [_][]const u8{ "", "timestamp,scope\n", "uri,scope\n", "uri,uri,timestamp\n", "uri,timestamp,timestmp\n", }; for (invalid) |source| { var reader: std.Io.Reader = .fixed(source); var collector: Collector = .{}; defer collector.deinit(); try testing.expectError(error.InvalidHeader, parse(testing.allocator, &reader, .{ .context = &collector, .on_row = Collector.onRow, })); }
var collector: Collector = .{}; defer collector.deinit(); var stats = try collect( " Timestamp , URI , CID , Scope \n" ++ "2022-01-02T03:04:05Z,at://did:plc:a/app.bsky.feed.post/r,,all_versions\n", &collector, Options.no_reject_sample, ); defer stats.deinit(testing.allocator); try testing.expectEqual(@as(u64, 1), stats.rows_valid);}
test "quoted multiline rows and escaped quotes are RFC 4180 decoded" { const testing = std.testing; // A newline inside the invalid timestamp remains one logical rejected row; // the escaped quote is decoded and diagnosed, and the next row survives. const source = "uri,timestamp\n" ++ "at://did:plc:a/app.bsky.feed.post/r,\"bad\r\n\"\"timestamp\"\n" ++ "at://did:plc:b/app.bsky.feed.post/r,2022-01-02T03:04:05.123456789Z\n"; var collector: Collector = .{}; defer collector.deinit(); var stats = try collect(source, &collector, 0); defer stats.deinit(testing.allocator); try testing.expectEqual(@as(u64, 1), stats.rows_rejected); try testing.expectEqual(@as(u64, 1), stats.rows_valid); try testing.expectEqual(@as(i64, 1_641_092_645_123_456), collector.rows.items[0].timestamp_micros);}
test "unbalanced or bare quote fails loudly after prior rows" { const testing = std.testing; const source = "uri,timestamp\n" ++ "at://did:plc:a/app.bsky.feed.post/r,2022-01-02T03:04:05Z\n" ++ "at://did:plc:b/app.bsky.feed.post/\"r,2022-01-02T03:04:05Z\n" ++ "at://did:plc:c/app.bsky.feed.post/r,2022-01-02T03:04:05Z\n"; var reader: std.Io.Reader = .fixed(source); var collector: Collector = .{}; defer collector.deinit(); try testing.expectError(error.UnbalancedQuote, parse(testing.allocator, &reader, .{ .context = &collector, .on_row = Collector.onRow, })); try testing.expectEqual(@as(usize, 1), collector.rows.items.len);}
test "parseInto preserves partial statistics on fatal structure error" { const testing = std.testing; const source = "uri,timestamp\n" ++ "at://did:plc:a/app.bsky.feed.post/r,2022-01-02T03:04:05Z\n" ++ "\"unterminated\n"; var reader: std.Io.Reader = .fixed(source); var collector: Collector = .{}; defer collector.deinit(); var stats: Stats = .{}; defer stats.deinit(testing.allocator); try testing.expectError(error.UnbalancedQuote, parseInto(testing.allocator, &reader, .{ .context = &collector, .on_row = Collector.onRow, }, &stats)); try testing.expectEqual(@as(u64, 1), stats.rows_total); try testing.expectEqual(@as(u64, 1), stats.rows_valid);}
test "row footprint is bounded and blank lines remain part of seek offset" { const testing = std.testing; var source: std.ArrayList(u8) = .empty; defer source.deinit(testing.allocator); try source.appendSlice(testing.allocator, "uri,timestamp\n\n\r\n"); const expected_offset = "uri,timestamp\n".len; try source.appendSlice(testing.allocator, "at://did:plc:huge/app.bsky.feed.post/"); try source.appendNTimes(testing.allocator, 'x', max_row_bytes); try source.appendSlice(testing.allocator, ",2022-01-02T03:04:05Z\n"); try source.appendSlice(testing.allocator, "at://did:plc:ok/app.bsky.feed.post/r,2022-01-02T03:04:05Z\n");
var collector: Collector = .{}; defer collector.deinit(); var stats = try collect(source.items, &collector, 0); defer stats.deinit(testing.allocator); try testing.expectEqual(@as(u64, 1), stats.rejected(.row_too_long)); try testing.expectEqual(@as(u64, 1), stats.rows_valid); try testing.expectEqual(@as(u64, 2), stats.rows_total); try testing.expectEqual(@as(u64, expected_offset), stats.reject_sample.items[0].offset);}
test "RFC3339 conversion matches Go edge semantics" { const testing = std.testing; try testing.expectEqual(@as(?i64, 1_641_092_645_123_456), parseRfc3339Micros("2022-01-02T03:04:05.1234567890Z")); try testing.expectEqual(@as(?i64, 1_641_092_645_100_000), parseRfc3339Micros("2022-01-02T03:04:05,1Z")); try testing.expectEqual(@as(?i64, 1_641_006_245_000_000), parseRfc3339Micros("2022-01-02T03:04:05+24:00")); try testing.expectEqual(@as(?i64, 1_641_089_045_000_000), parseRfc3339Micros("2022-01-02T03:04:05+00:60")); try testing.expectEqual(@as(?i64, null), parseRfc3339Micros("2022-01-02t03:04:05Z")); try testing.expectEqual(@as(?i64, null), parseRfc3339Micros("2022-01-02T03:04:05z")); try testing.expectEqual(@as(?i64, null), parseRfc3339Micros("2022-01-02T03:04:60Z"));}
test "reject sample and callback are independently bounded" { const testing = std.testing; var source: std.ArrayList(u8) = .empty; defer source.deinit(testing.allocator); try source.appendSlice(testing.allocator, "uri,timestamp\n"); for (0..150) |_| try source.appendSlice(testing.allocator, "bad,bad\n"); var collector: Collector = .{}; defer collector.deinit(); var stats = try collect(source.items, &collector, 10); defer stats.deinit(testing.allocator); try testing.expectEqual(@as(usize, 150), collector.rejects_seen); try testing.expectEqual(@as(usize, 10), stats.reject_sample.items.len);}
test "streaming parse performs no heap allocation when reject sampling is disabled" { const testing = std.testing; const Counter = struct { count: usize = 0, fn onRow(raw: ?*anyopaque, _: Row) !void { const self: *@This() = @ptrCast(@alignCast(raw.?)); self.count += 1; } };
var source: std.ArrayList(u8) = .empty; defer source.deinit(testing.allocator); try source.appendSlice(testing.allocator, "uri,timestamp\n"); for (0..10_000) |index| { var row_buffer: [128]u8 = undefined; const row = try std.fmt.bufPrint( &row_buffer, "at://did:plc:bounded/app.bsky.feed.post/{d},2022-01-02T03:04:05Z\n", .{index}, ); try source.appendSlice(testing.allocator, row); }
var reader: std.Io.Reader = .fixed(source.items); var counter: Counter = .{}; var stats = try parse(testing.failing_allocator, &reader, .{ .context = &counter, .on_row = Counter.onRow, .reject_sample_limit = Options.no_reject_sample, }); defer stats.deinit(testing.failing_allocator); try testing.expectEqual(@as(usize, 10_000), counter.count); try testing.expectEqual(@as(u64, 10_000), stats.rows_valid);}