Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720//! Durable timestamp-import intent and append-time stamping.//!//! This is a separate RocksDB instance because the rule map is large, cold,//! and rebuildable. CSV rows are sorted into external SST files and ingested//! in source order, preserving upstream's scalable last-write-wins behavior.
const std = @import("std");const rdb = @import("rocksdb_raw");const segment = @import("../storage/segment.zig");const parse_mod = @import("parse.zig");
const Allocator = std.mem.Allocator;const Io = std.Io;const log = std.log.scoped(.stream_timestamp_rules);
pub const subdir = "import-rules";pub const default_sst_chunk_entries: usize = 1_000_000;
const key_all_versions: u8 = 'a';const key_specific: u8 = 's';const key_has_specific: u8 = 'p';const key_collection: u8 = 'c';const key_sep: u8 = 0;const marker_value: [8]u8 = .{ 1, 0, 0, 0, 0, 0, 0, 0 };
pub const ImportResult = struct { parse: parse_mod.Stats = .{}, rules: u64 = 0, sstables: usize = 0, bytes_ingested: u64 = 0,
pub fn deinit(self: *ImportResult, allocator: Allocator) void { self.parse.deinit(allocator); self.* = undefined; }};
pub const ImportConfig = struct { csv_path: []const u8, scratch_dir: []const u8, chunk_entries: usize = 0, cancelled: ?*const std.atomic.Value(bool) = null,};
pub const Store = struct { allocator: Allocator, io: Io, db: *rdb.rocksdb_t, options: *rdb.rocksdb_options_t, block_options: *rdb.rocksdb_block_based_table_options_t, read_options: *rdb.rocksdb_readoptions_t, collections: std.StringHashMapUnmanaged(void) = .empty, collections_lock: Io.RwLock = .init,
pub fn open(allocator: Allocator, io: Io, data_dir: []const u8) !Store { if (data_dir.len == 0) return error.MissingDataDir; const path = try std.fs.path.join(allocator, &.{ data_dir, subdir }); defer allocator.free(path); const path_z = try allocator.dupeSentinel(u8, path, 0); defer allocator.free(path_z);
const options = rdb.rocksdb_options_create() orelse return error.OutOfMemory; errdefer rdb.rocksdb_options_destroy(options); rdb.rocksdb_options_set_create_if_missing(options, 1); const block_options = rdb.rocksdb_block_based_options_create() orelse return error.OutOfMemory; errdefer rdb.rocksdb_block_based_options_destroy(block_options); const bloom = rdb.rocksdb_filterpolicy_create_bloom_full(10) orelse return error.OutOfMemory; rdb.rocksdb_block_based_options_set_filter_policy(block_options, bloom); rdb.rocksdb_options_set_block_based_table_factory(options, block_options);
var err_ptr: ?[*:0]u8 = null; const db = rdb.rocksdb_open(options, path_z.ptr, errorOut(&err_ptr)); try checkError(&err_ptr, "open timestamp rule database"); const live_db = db orelse return error.RocksDBOpen; errdefer rdb.rocksdb_close(live_db); const read_options = rdb.rocksdb_readoptions_create() orelse return error.OutOfMemory; errdefer rdb.rocksdb_readoptions_destroy(read_options);
var store: Store = .{ .allocator = allocator, .io = io, .db = live_db, .options = options, .block_options = block_options, .read_options = read_options, }; try store.reloadCollections(); log.info("timestamp rule store opened at {s}", .{path}); return store; }
pub fn deinit(self: *Store) void { self.collections_lock.lockUncancelable(self.io); self.freeCollections(&self.collections); self.collections_lock.unlock(self.io); rdb.rocksdb_readoptions_destroy(self.read_options); rdb.rocksdb_close(self.db); rdb.rocksdb_options_destroy(self.options); // The filter policy was transferred to the block options. rdb.rocksdb_block_based_options_destroy(self.block_options); self.* = undefined; }
pub fn hasRules(self: *Store) bool { self.collections_lock.lockSharedUncancelable(self.io); defer self.collections_lock.unlockShared(self.io); return self.collections.count() != 0; }
pub fn reloadCollections(self: *Store) !void { var next: std.StringHashMapUnmanaged(void) = .empty; errdefer self.freeCollections(&next);
const iterator = rdb.rocksdb_create_iterator(self.db, self.read_options) orelse return error.OutOfMemory; defer rdb.rocksdb_iter_destroy(iterator); const prefix = [_]u8{key_collection}; rdb.rocksdb_iter_seek(iterator, &prefix, prefix.len); while (rdb.rocksdb_iter_valid(iterator) != 0) { var key_len: usize = 0; const key_ptr = rdb.rocksdb_iter_key(iterator, &key_len); if (key_ptr == null or key_len == 0) return error.CorruptRuleCollection; const key = key_ptr[0..key_len]; if (key[0] != key_collection) break; if (key.len == 1) return error.CorruptRuleCollection; const owned = try self.allocator.dupe(u8, key[1..]); errdefer self.allocator.free(owned); const gop = try next.getOrPut(self.allocator, owned); if (gop.found_existing) self.allocator.free(owned); rdb.rocksdb_iter_next(iterator); } var err_ptr: ?[*:0]u8 = null; rdb.rocksdb_iter_get_error(iterator, errorOut(&err_ptr)); try checkError(&err_ptr, "iterate timestamp rule collections");
self.collections_lock.lockUncancelable(self.io); defer self.collections_lock.unlock(self.io); self.freeCollections(&self.collections); self.collections = next; }
/// Apply a rule to a materialization row. Any RocksDB failure is fatal to /// the caller; silently falling back to witnessed time would lose import /// intent and diverge from upstream. pub fn stamp(self: *Store, event: *segment.Event) !void { if (!isMaterialization(event.kind)) return; self.collections_lock.lockSharedUncancelable(self.io); const imported = self.collections.contains(event.collection); self.collections_lock.unlockShared(self.io); if (!imported) return;
var timestamp: i64 = 0; const marker_key = try pathKey(self.allocator, key_has_specific, event.did, event.collection, event.rkey); defer self.allocator.free(marker_key); if (try self.has(marker_key)) { const specific_key = try specificKeyForPayload(self.allocator, event.did, event.collection, event.rkey, event.payload); defer self.allocator.free(specific_key); timestamp = (try self.getTimestamp(specific_key)) orelse 0; } if (timestamp == 0) { const all_key = try pathKey(self.allocator, key_all_versions, event.did, event.collection, event.rkey); defer self.allocator.free(all_key); timestamp = (try self.getTimestamp(all_key)) orelse 0; } if (timestamp != 0) event.indexed_at = timestamp; }
pub fn stampOpaque(raw: *anyopaque, event: *segment.Event) anyerror!void { const self: *Store = @ptrCast(@alignCast(raw)); try self.stamp(event); }
/// Stream, validate, sort, and externally ingest a complete rule CSV. /// `result` always remains initialized and must be deinitialized, including /// when this function returns an error. pub fn importCsv(self: *Store, cfg: ImportConfig, result: *ImportResult) !void { result.* = .{}; if (cfg.csv_path.len == 0) return error.MissingCsvPath; if (cfg.scratch_dir.len == 0) return error.MissingScratchDir; try Io.Dir.cwd().createDirPath(self.io, cfg.scratch_dir);
var file = try Io.Dir.cwd().openFile(self.io, cfg.csv_path, .{}); defer file.close(self.io); var read_buffer: [64 * 1024]u8 = undefined; var reader = file.reader(self.io, &read_buffer); var builder: Builder = .{ .allocator = self.allocator, .store = self, .scratch_dir = cfg.scratch_dir, .chunk_entries = if (cfg.chunk_entries == 0) default_sst_chunk_entries else cfg.chunk_entries, .cancelled = cfg.cancelled, }; defer builder.deinit();
parse_mod.parseInto(self.allocator, &reader.interface, .{ .context = &builder, .on_row = Builder.onRow, }, &result.parse) catch |err| { result.rules = builder.rules; result.sstables = builder.paths.items.len; result.bytes_ingested = builder.bytes_written; return err; }; try builder.flush(); result.rules = builder.rules; result.sstables = builder.paths.items.len; result.bytes_ingested = builder.bytes_written; try builder.ingest(); try self.reloadCollections(); }
fn has(self: *Store, key: []const u8) !bool { var value_len: usize = 0; var err_ptr: ?[*:0]u8 = null; const value = rdb.rocksdb_get( self.db, self.read_options, key.ptr, key.len, &value_len, errorOut(&err_ptr), ); try checkError(&err_ptr, "read timestamp rule marker"); if (value) |ptr| { rdb.rocksdb_free(ptr); return true; } return false; }
fn getTimestamp(self: *Store, key: []const u8) !?i64 { var value_len: usize = 0; var err_ptr: ?[*:0]u8 = null; const value = rdb.rocksdb_get( self.db, self.read_options, key.ptr, key.len, &value_len, errorOut(&err_ptr), ); try checkError(&err_ptr, "read timestamp rule"); const ptr = value orelse return null; defer rdb.rocksdb_free(ptr); if (value_len != 8) return error.CorruptRuleTimestamp; return std.mem.readInt(i64, ptr[0..8], .little); }
fn freeCollections(self: *Store, collections: *std.StringHashMapUnmanaged(void)) void { var iterator = collections.keyIterator(); while (iterator.next()) |key| self.allocator.free(key.*); collections.deinit(self.allocator); collections.* = .empty; }};
const RuleEntry = struct { key: []u8, value: [8]u8, ordinal: u64,};
const Builder = struct { allocator: Allocator, store: *Store, scratch_dir: []const u8, chunk_entries: usize, cancelled: ?*const std.atomic.Value(bool), entries: std.ArrayList(RuleEntry) = .empty, paths: std.ArrayList([]u8) = .empty, rules: u64 = 0, next_ordinal: u64 = 0, bytes_written: u64 = 0,
fn deinit(self: *Builder) void { self.clearEntries(); self.entries.deinit(self.allocator); for (self.paths.items) |path| { Io.Dir.cwd().deleteFile(self.store.io, path) catch {}; self.allocator.free(path); } self.paths.deinit(self.allocator); }
fn onRow(raw: ?*anyopaque, row: parse_mod.Row) !void { const self: *Builder = @ptrCast(@alignCast(raw.?)); try self.add(row); }
fn add(self: *Builder, row: parse_mod.Row) !void { if (self.cancelled) |flag| if (flag.load(.acquire)) return error.ImportCancelled; try self.appendEntry(try collectionKey(self.allocator, row.collection), marker_value); switch (row.scope) { .all_versions => try self.appendEntry( try pathKey(self.allocator, key_all_versions, row.did, row.collection, row.rkey), timestampValue(row.timestamp_micros), ), .specific_version => { try self.appendEntry( try pathKey(self.allocator, key_has_specific, row.did, row.collection, row.rkey), marker_value, ); try self.appendEntry( try specificKey(self.allocator, row.did, row.collection, row.rkey, row.cid.?), timestampValue(row.timestamp_micros), ); }, } self.rules += 1; if (self.entries.items.len >= self.chunk_entries) try self.flush(); }
fn appendEntry(self: *Builder, key: []u8, value: [8]u8) !void { errdefer self.allocator.free(key); try self.entries.append(self.allocator, .{ .key = key, .value = value, .ordinal = self.next_ordinal, }); self.next_ordinal += 1; }
fn flush(self: *Builder) !void { if (self.entries.items.len == 0) return; std.mem.sort(RuleEntry, self.entries.items, {}, struct { fn less(_: void, a: RuleEntry, b: RuleEntry) bool { const order = std.mem.order(u8, a.key, b.key); return order == .lt or (order == .eq and a.ordinal < b.ordinal); } }.less);
const path = try std.fmt.allocPrint( self.allocator, "{s}/rules_{d:0>6}.sst", .{ self.scratch_dir, self.paths.items.len }, ); errdefer self.allocator.free(path); errdefer Io.Dir.cwd().deleteFile(self.store.io, path) catch {}; const path_z = try self.allocator.dupeSentinel(u8, path, 0); defer self.allocator.free(path_z);
const env_options = rdb.rocksdb_envoptions_create() orelse return error.OutOfMemory; defer rdb.rocksdb_envoptions_destroy(env_options); const writer = rdb.rocksdb_sstfilewriter_create(env_options, self.store.options) orelse return error.OutOfMemory; defer rdb.rocksdb_sstfilewriter_destroy(writer); var err_ptr: ?[*:0]u8 = null; rdb.rocksdb_sstfilewriter_open(writer, path_z.ptr, errorOut(&err_ptr)); try checkError(&err_ptr, "open timestamp rule SST");
var index: usize = 0; while (index < self.entries.items.len) { var chosen = index; var next = index + 1; while (next < self.entries.items.len and std.mem.eql(u8, self.entries.items[index].key, self.entries.items[next].key)) : (next += 1) { chosen = next; } const entry = self.entries.items[chosen]; rdb.rocksdb_sstfilewriter_put( writer, entry.key.ptr, entry.key.len, &entry.value, entry.value.len, errorOut(&err_ptr), ); try checkError(&err_ptr, "write timestamp rule SST"); index = next; } rdb.rocksdb_sstfilewriter_finish(writer, errorOut(&err_ptr)); try checkError(&err_ptr, "finish timestamp rule SST"); var file_size: u64 = 0; rdb.rocksdb_sstfilewriter_file_size(writer, &file_size); self.bytes_written += file_size; try self.paths.append(self.allocator, path); self.clearEntries(); }
fn ingest(self: *Builder) !void { const options = rdb.rocksdb_ingestexternalfileoptions_create() orelse return error.OutOfMemory; defer rdb.rocksdb_ingestexternalfileoptions_destroy(options); rdb.rocksdb_ingestexternalfileoptions_set_move_files(options, 1); rdb.rocksdb_ingestexternalfileoptions_set_allow_global_seqno(options, 1); rdb.rocksdb_ingestexternalfileoptions_set_allow_blocking_flush(options, 1); for (self.paths.items) |path| { const path_z = try self.allocator.dupeSentinel(u8, path, 0); defer self.allocator.free(path_z); const paths = [_][*c]const u8{path_z.ptr}; var err_ptr: ?[*:0]u8 = null; rdb.rocksdb_ingest_external_file( self.store.db, @ptrCast(&paths), paths.len, options, errorOut(&err_ptr), ); try checkError(&err_ptr, "ingest timestamp rule SST"); } }
fn clearEntries(self: *Builder) void { for (self.entries.items) |entry| self.allocator.free(entry.key); self.entries.clearRetainingCapacity(); }};
fn timestampValue(timestamp: i64) [8]u8 { var value: [8]u8 = undefined; std.mem.writeInt(i64, &value, timestamp, .little); return value;}
fn collectionKey(allocator: Allocator, collection: []const u8) ![]u8 { const key = try allocator.alloc(u8, 1 + collection.len); key[0] = key_collection; @memcpy(key[1..], collection); return key;}
fn pathKey(allocator: Allocator, prefix: u8, did: []const u8, collection: []const u8, rkey: []const u8) ![]u8 { const key = try allocator.alloc(u8, 1 + did.len + 1 + collection.len + 1 + rkey.len); var at: usize = 0; key[at] = prefix; at += 1; @memcpy(key[at..][0..did.len], did); at += did.len; key[at] = key_sep; at += 1; @memcpy(key[at..][0..collection.len], collection); at += collection.len; key[at] = key_sep; at += 1; @memcpy(key[at..][0..rkey.len], rkey); return key;}
fn specificKey(allocator: Allocator, did: []const u8, collection: []const u8, rkey: []const u8, cid: parse_mod.Cid) ![]u8 { const path = try pathKey(allocator, key_specific, did, collection, rkey); defer allocator.free(path); const key = try allocator.alloc(u8, path.len + 1 + 36); @memcpy(key[0..path.len], path); key[path.len] = key_sep; key[path.len + 1] = 1; key[path.len + 2] = cid.codec; key[path.len + 3] = 0x12; key[path.len + 4] = 0x20; @memcpy(key[path.len + 5 ..], &cid.hash); return key;}
fn specificKeyForPayload(allocator: Allocator, did: []const u8, collection: []const u8, rkey: []const u8, payload: []const u8) ![]u8 { var hash: [32]u8 = undefined; std.crypto.hash.sha2.Sha256.hash(payload, &hash, .{}); return specificKey(allocator, did, collection, rkey, .{ .codec = 0x71, .hash = hash });}
fn isMaterialization(kind: segment.Kind) bool { return switch (kind) { .create, .update, .create_resync => true, else => false, };}
fn errorOut(pointer: *?[*:0]u8) [*c][*c]u8 { return @ptrCast(pointer);}
fn checkError(pointer: *?[*:0]u8, operation: []const u8) !void { const message = pointer.* orelse return; pointer.* = null; defer rdb.rocksdb_free(message); log.err("{s}: {s}", .{ operation, std.mem.span(message) }); return error.RocksDB;}
const rules_did = "did:plc:alice";const rules_collection = "app.bsky.feed.post";const rules_rkey = "r1";
fn importFixture(store: *Store, tmp: *std.testing.TmpDir, rows: []const u8, chunk_entries: usize) !ImportResult { const testing = std.testing; try tmp.dir.writeFile(testing.io, .{ .sub_path = "rules.csv", .data = rows }); try tmp.dir.createDirPath(testing.io, "scratch"); var root_buffer: [std.fs.max_path_bytes]u8 = undefined; const root_len = try tmp.dir.realPath(testing.io, &root_buffer); const root = root_buffer[0..root_len]; const csv = try std.fs.path.join(testing.allocator, &.{ root, "rules.csv" }); defer testing.allocator.free(csv); const scratch = try std.fs.path.join(testing.allocator, &.{ root, "scratch" }); defer testing.allocator.free(scratch); var result: ImportResult = .{}; try store.importCsv(.{ .csv_path = csv, .scratch_dir = scratch, .chunk_entries = chunk_entries, }, &result); return result;}
fn openFixture(tmp: *std.testing.TmpDir) !Store { var root_buffer: [std.fs.max_path_bytes]u8 = undefined; const root_len = try tmp.dir.realPath(std.testing.io, &root_buffer); return Store.open(std.testing.allocator, std.testing.io, root_buffer[0..root_len]);}
fn fixtureEvent(kind: segment.Kind, collection: []const u8, payload: []const u8) segment.Event { return .{ .seq = 1, .witnessed_at = 10, .indexed_at = 0, .kind = kind, .did = rules_did, .collection = collection, .rkey = rules_rkey, .rev = "3lrev", .payload = payload, };}
test "all-versions rule stamps materializations only" { const testing = std.testing; var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); var store = try openFixture(&tmp); defer store.deinit(); var result = try importFixture(&store, &tmp, "uri,timestamp,scope,cid\n" ++ "at://" ++ rules_did ++ "/" ++ rules_collection ++ "/" ++ rules_rkey ++ ",2021-12-20T11:33:20Z,all_versions,\n", 0); defer result.deinit(testing.allocator); try testing.expect(store.hasRules()); try testing.expectEqual(@as(u64, 1), result.rules); try testing.expect(result.sstables >= 1); try testing.expect(result.bytes_ingested > 0);
var create = fixtureEvent(.create, rules_collection, &.{0xa0}); try store.stamp(&create); try testing.expectEqual(@as(i64, 1_640_000_000_000_000), create.indexed_at); var delete = fixtureEvent(.delete, rules_collection, ""); try store.stamp(&delete); try testing.expectEqual(@as(i64, 0), delete.indexed_at); var other = fixtureEvent(.create, "app.bsky.feed.like", &.{0xa0}); try store.stamp(&other); try testing.expectEqual(@as(i64, 0), other.indexed_at);}
test "specific version wins and misses fall back to all versions" { const testing = std.testing; var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); var store = try openFixture(&tmp); defer store.deinit(); const payload_v1 = "payload-v1"; const parsed_cid = try @import("zat").cbor.Cid.forDagCbor(testing.allocator, payload_v1); defer testing.allocator.free(@constCast(parsed_cid.raw)); const cid = try @import("../serve/wire.zig").cidString(testing.allocator, parsed_cid); defer testing.allocator.free(cid); const csv = try std.fmt.allocPrint(testing.allocator, "uri,timestamp,scope,cid\n" ++ "at://{s}/{s}/{s},2021-12-20T11:33:20Z,all_versions,\n" ++ "at://{s}/{s}/{s},2022-01-02T03:04:05Z,specific_version,{s}\n", .{ rules_did, rules_collection, rules_rkey, rules_did, rules_collection, rules_rkey, cid }); defer testing.allocator.free(csv); var result = try importFixture(&store, &tmp, csv, 0); defer result.deinit(testing.allocator);
var specific = fixtureEvent(.create, rules_collection, payload_v1); try store.stamp(&specific); try testing.expectEqual(@as(i64, 1_641_092_645_000_000), specific.indexed_at); var fallback = fixtureEvent(.update, rules_collection, "payload-v2"); try store.stamp(&fallback); try testing.expectEqual(@as(i64, 1_640_000_000_000_000), fallback.indexed_at);}
test "duplicate key is last-write-wins within and across SST chunks" { const testing = std.testing; var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); var store = try openFixture(&tmp); defer store.deinit(); var result = try importFixture(&store, &tmp, "uri,timestamp,scope,cid\n" ++ "at://" ++ rules_did ++ "/" ++ rules_collection ++ "/" ++ rules_rkey ++ ",2021-12-20T11:33:20Z,all_versions,\n" ++ "at://" ++ rules_did ++ "/" ++ rules_collection ++ "/" ++ rules_rkey ++ ",2022-01-02T03:04:05Z,all_versions,\n" ++ "at://" ++ rules_did ++ "/" ++ rules_collection ++ "/" ++ rules_rkey ++ ",2023-03-04T05:06:07Z,all_versions,\n", 2); defer result.deinit(testing.allocator); try testing.expectEqual(@as(u64, 3), result.rules); try testing.expect(result.sstables >= 2); var event = fixtureEvent(.create, rules_collection, &.{0xa0}); try store.stamp(&event); try testing.expectEqual(@as(i64, 1_677_906_367_000_000), event.indexed_at);}
test "rules and resident collection filter survive reopen" { const testing = std.testing; var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); { var store = try openFixture(&tmp); defer store.deinit(); var result = try importFixture(&store, &tmp, "uri,timestamp\n" ++ "at://" ++ rules_did ++ "/" ++ rules_collection ++ "/" ++ rules_rkey ++ ",2021-12-20T11:33:20Z\n", 0); defer result.deinit(testing.allocator); } var reopened = try openFixture(&tmp); defer reopened.deinit(); try testing.expect(reopened.hasRules()); var event = fixtureEvent(.create, rules_collection, &.{0xa0}); try reopened.stamp(&event); try testing.expectEqual(@as(i64, 1_640_000_000_000_000), event.indexed_at);}
test "fatal parse does not ingest completed scratch chunks" { const testing = std.testing; var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); var store = try openFixture(&tmp); defer store.deinit(); try tmp.dir.writeFile(testing.io, .{ .sub_path = "bad.csv", .data = "uri,timestamp\n" ++ "at://" ++ rules_did ++ "/" ++ rules_collection ++ "/" ++ rules_rkey ++ ",2021-12-20T11:33:20Z\n" ++ "\"unterminated\n", }); try tmp.dir.createDirPath(testing.io, "bad-scratch"); var root_buffer: [std.fs.max_path_bytes]u8 = undefined; const root_len = try tmp.dir.realPath(testing.io, &root_buffer); const csv = try std.fs.path.join(testing.allocator, &.{ root_buffer[0..root_len], "bad.csv" }); defer testing.allocator.free(csv); const scratch = try std.fs.path.join(testing.allocator, &.{ root_buffer[0..root_len], "bad-scratch" }); defer testing.allocator.free(scratch); var result: ImportResult = .{}; defer result.deinit(testing.allocator); try testing.expectError(error.UnbalancedQuote, store.importCsv(.{ .csv_path = csv, .scratch_dir = scratch, .chunk_entries = 2, }, &result)); try testing.expectEqual(@as(u64, 1), result.parse.rows_valid); try testing.expectEqual(@as(u64, 1), result.rules); try testing.expectEqual(@as(usize, 1), result.sstables); try testing.expect(!store.hasRules()); var event = fixtureEvent(.create, rules_collection, &.{0xa0}); try store.stamp(&event); try testing.expectEqual(@as(i64, 0), event.indexed_at);
var scratch_dir = try Io.Dir.cwd().openDir(testing.io, scratch, .{ .iterate = true }); defer scratch_dir.close(testing.io); var iterator = scratch_dir.iterate(); try testing.expectEqual(null, try iterator.next(testing.io));}
test "rule ingestion observes cancellation without publishing partial rules" { const testing = std.testing; var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); var store = try openFixture(&tmp); defer store.deinit(); try tmp.dir.writeFile(testing.io, .{ .sub_path = "cancel.csv", .data = "uri,timestamp\n" ++ "at://" ++ rules_did ++ "/" ++ rules_collection ++ "/" ++ rules_rkey ++ ",2021-12-20T11:33:20Z\n", }); var root_buffer: [std.fs.max_path_bytes]u8 = undefined; const root_len = try tmp.dir.realPath(testing.io, &root_buffer); const csv = try std.fs.path.join(testing.allocator, &.{ root_buffer[0..root_len], "cancel.csv" }); defer testing.allocator.free(csv); const scratch = try std.fs.path.join(testing.allocator, &.{ root_buffer[0..root_len], "cancel-scratch" }); defer testing.allocator.free(scratch); var cancelled: std.atomic.Value(bool) = .init(true); var result: ImportResult = .{}; defer result.deinit(testing.allocator); try testing.expectError(error.ImportCancelled, store.importCsv(.{ .csv_path = csv, .scratch_dir = scratch, .cancelled = &cancelled, }, &result)); try testing.expectEqual(@as(u64, 1), result.parse.rows_total); try testing.expectEqual(@as(u64, 1), result.parse.rows_valid); try testing.expect(!store.hasRules());}
test "archive transaction stamps caller, writer memory, and durable JSS row identically" { const testing = std.testing; const archive_mod = @import("../storage/archive.zig"); const writer_mod = @import("../storage/segment_writer.zig"); var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); var root_buffer: [std.fs.max_path_bytes]u8 = undefined; const root_len = try tmp.dir.realPath(io, &root_buffer); const root = root_buffer[0..root_len];
var store = try Store.open(testing.allocator, io, root); defer store.deinit(); var result = try importFixture(&store, &tmp, "uri,timestamp\n" ++ "at://" ++ rules_did ++ "/" ++ rules_collection ++ "/" ++ rules_rkey ++ ",2021-12-20T11:33:20Z\n", 0); defer result.deinit(testing.allocator);
var archive = try archive_mod.Archive.init(testing.allocator, io, root); defer archive.deinit(); archive.stamper_ctx = &store; archive.stamper = Store.stampOpaque; archive.max_events_per_block = 10; archive.writer.max_events_per_block = 10; var rows = [_]segment.Event{fixtureEvent(.create, rules_collection, &.{0xa0})}; _ = try archive.appendBatch(&rows, 1); const wanted: i64 = 1_640_000_000_000_000; try testing.expectEqual(wanted, rows[0].indexed_at); try testing.expectEqual(@as(u64, 1), rows[0].seq); try testing.expectEqual(wanted, archive.writer.pending.items[0].indexed_at); try archive.flushBlock();
var iterator: writer_mod.ActiveIterator = .{ .bytes = archive.writer.bytes() }; var block = (try iterator.next(testing.allocator)).?; defer block.deinit(testing.allocator); try testing.expectEqual(@as(usize, 1), block.events.len); try testing.expectEqual(wanted, block.events[0].indexed_at); try testing.expectEqual(rows[0].displayTime(), block.events[0].displayTime());}