Something went wrong. Try again.
My version of https://github.com/vercel-labs/fx
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002const std = @import("std");const io_mod = @import("../shared/io.zig");const debug_trace = @import("../shared/debug_trace.zig");const profile_paths = @import("../shared/profile_paths.zig");const generation_fact_codec = @import("generation_fact_codec.zig");const usage_report = @import("usage_report.zig");
const Allocator = std.mem.Allocator;const usage_file = profile_paths.usage_file_name;const usage_lock_file = "usage.lock";const lock_deadline_ms: u64 = 2000;const max_file_bytes: usize = 32 * 1024 * 1024;const max_record_bytes: usize = 16 * 1024;const max_records: usize = 200_000;const compaction_threshold_bytes: u64 = 8 * 1024 * 1024;const retention_ms: i64 = std.time.ms_per_day * 35;const private_dir_permissions = std.Io.Dir.Permissions.fromMode(0o700);const private_file_permissions = std.Io.File.Permissions.fromMode(0o600);
pub const AppendOutcome = enum { appended, duplicate, conflict,};
pub const Loaded = struct { coverage_started_at_ms: ?i64, facts: []usage_report.GenerationFact, pending: []usage_report.PendingMarker, incidents: []usage_report.Incident, record_count: usize,
pub fn deinit(self: *Loaded, alloc: Allocator) void { for (self.facts) |*fact| fact.deinit(alloc); alloc.free(self.facts); for (self.pending) |*marker| marker.deinit(alloc); alloc.free(self.pending); alloc.free(self.incidents); self.* = undefined; }};
const ParsedRecord = union(enum) { coverage: i64, generation: usage_report.GenerationFact, pending: usage_report.PendingMarker, incident: usage_report.Incident,
fn deinit(self: *ParsedRecord, alloc: Allocator) void { switch (self.*) { .generation => |*fact| fact.deinit(alloc), .pending => |*marker| marker.deinit(alloc), .coverage, .incident => {}, } self.* = undefined; }};
const AppendDecision = struct { outcome: AppendOutcome, write: bool,};
const VariantIndexes = struct { first: usize, second: ?usize = null,};
/// Hard bound on incident records kept in the ledger file. Incidents mark/// coverage gaps; an unbounded flood of them adds parse cost to every append/// without adding signal.const max_file_incidents: usize = 4096;
/// Suffix bytes ending at the indexed boundary, used to detect a foreign/// replace (compaction or tail repair from another process) that kept or grew/// the file length.const tail_sample_bytes: usize = 32;
/// One durable parse of the usage ledger, kept across appends so steady-state/// publication reads only the bytes appended since the last boundary instead/// of re-parsing the whole file.const RecordIndex = struct { coverage_started_at_ms: ?i64 = null, facts: std.ArrayList(usage_report.GenerationFact) = .empty, pending: std.ArrayList(usage_report.PendingMarker) = .empty, incidents: std.ArrayList(usage_report.Incident) = .empty, fact_indexes: std.StringHashMapUnmanaged(VariantIndexes) = .empty, pending_indexes: std.StringHashMapUnmanaged(VariantIndexes) = .empty, record_count: usize = 0, /// Offset just past the last absorbed byte. Appends are whole lines, so /// this always points at the start of a line. boundary: u64 = 0, tail_sample: [tail_sample_bytes]u8 = undefined, tail_sample_len: usize = 0, /// Times a tail absorb extended the index; tests use it to prove the /// incremental path ran instead of a full re-parse. incremental_absorbs: usize = 0,
fn deinit(self: *RecordIndex, alloc: Allocator) void { for (self.facts.items) |*fact| fact.deinit(alloc); self.facts.deinit(alloc); for (self.pending.items) |*marker| marker.deinit(alloc); self.pending.deinit(alloc); self.incidents.deinit(alloc); self.fact_indexes.deinit(alloc); self.pending_indexes.deinit(alloc); self.* = undefined; }
/// Moves the record lists into a caller-owned Loaded for one-shot reads. fn toLoaded(self: *RecordIndex, alloc: Allocator) !Loaded { var built: Loaded = .{ .coverage_started_at_ms = null, .facts = &.{}, .pending = &.{}, .incidents = &.{}, .record_count = 0, }; errdefer built.deinit(alloc); built.coverage_started_at_ms = self.coverage_started_at_ms; built.facts = try self.facts.toOwnedSlice(alloc); built.pending = try self.pending.toOwnedSlice(alloc); built.incidents = try self.incidents.toOwnedSlice(alloc); built.record_count = self.record_count; return built; }
fn captureTailSample(self: *RecordIndex, bytes: []const u8) void { const n = @min(tail_sample_bytes, bytes.len); @memcpy(self.tail_sample[0..n], bytes[bytes.len - n ..]); self.tail_sample_len = n; }};
fn absorbRecord(index: *RecordIndex, alloc: Allocator, record: ParsedRecord) !void { switch (record) { .coverage => |started_at_ms| { if (index.coverage_started_at_ms) |existing| { if (existing != started_at_ms) return error.InvalidUsageStore; } else { index.coverage_started_at_ms = started_at_ms; } }, .generation => |fact| { if (index.coverage_started_at_ms == null) return error.InvalidUsageStore; if (index.fact_indexes.getPtr(fact.id)) |indexes| { if (usage_report.GenerationFact.eql( index.facts.items[indexes.first], fact, ) or (indexes.second != null and usage_report.GenerationFact.eql( index.facts.items[indexes.second.?], fact, ))) { return; } if (indexes.second == null) { var owned = try fact.dupe(alloc); index.facts.append(alloc, owned) catch |err| { owned.deinit(alloc); return err; }; indexes.second = index.facts.items.len - 1; } } else { var owned = try fact.dupe(alloc); index.facts.append(alloc, owned) catch |err| { owned.deinit(alloc); return err; }; try index.fact_indexes.put( alloc, index.facts.items[index.facts.items.len - 1].id, .{ .first = index.facts.items.len - 1 }, ); } }, .pending => |marker| { if (index.coverage_started_at_ms == null) return error.InvalidUsageStore; if (index.pending_indexes.getPtr(marker.id)) |indexes| { if (usage_report.PendingMarker.eql( index.pending.items[indexes.first], marker, ) or (indexes.second != null and usage_report.PendingMarker.eql( index.pending.items[indexes.second.?], marker, ))) { return; } if (indexes.second == null) { var owned = try marker.dupe(alloc); index.pending.append(alloc, owned) catch |err| { owned.deinit(alloc); return err; }; indexes.second = index.pending.items.len - 1; try index.incidents.append(alloc, .{ .occurred_at_ms = marker.observed_at_ms, .completeness = .incomplete, }); } } else { var owned = try marker.dupe(alloc); index.pending.append(alloc, owned) catch |err| { owned.deinit(alloc); return err; }; try index.pending_indexes.put( alloc, index.pending.items[index.pending.items.len - 1].id, .{ .first = index.pending.items.len - 1 }, ); } }, .incident => |incident| try index.incidents.append(alloc, incident), }}
fn absorbBytes(index: *RecordIndex, alloc: Allocator, bytes: []const u8) !void { var lines = std.mem.splitScalar(u8, bytes, '\n'); while (lines.next()) |line| { if (line.len == 0) continue; index.record_count += 1; if (index.record_count > max_records or line.len > max_record_bytes) { return error.UsageCapacityExceeded; } var record = try parseRecord(alloc, line); defer record.deinit(alloc); try absorbRecord(index, alloc, record); }}
fn verifyTailSample(index: *const RecordIndex, file: std.Io.File) bool { if (index.tail_sample_len == 0) return false; var buf: [tail_sample_bytes]u8 = undefined; const start = index.boundary - index.tail_sample_len; const count = file.readPositionalAll(io_mod.getIo(), buf[0..index.tail_sample_len], start) catch return false; if (count != index.tail_sample_len) return false; return std.mem.eql(u8, buf[0..index.tail_sample_len], index.tail_sample[0..index.tail_sample_len]);}
const TailBoundary = struct { length: u64, incomplete: bool,};
pub const Store = struct { home_path: []u8, durable_home: ?io_mod.VerifiedDir = null, lock_ops: io_mod.LockOps = .{}, index: ?RecordIndex = null, index_alloc: ?Allocator = null,
pub fn initFromHome(alloc: Allocator, home_path: []const u8) !Store { const owned_home = try alloc.dupe(u8, home_path); errdefer alloc.free(owned_home); return .{ .home_path = owned_home, .durable_home = try openExistingDurableHome(home_path), }; }
pub fn deinit(self: *Store, alloc: Allocator) void { self.invalidateIndex(); if (self.durable_home) |*dir| dir.close(); alloc.free(self.home_path); self.* = undefined; }
fn appendFact( self: *Store, alloc: Allocator, fact: usage_report.GenerationFact, ) !AppendOutcome { return self.appendEvent(alloc, .{ .generation = fact }); }
pub fn appendEvent( self: *Store, alloc: Allocator, event: usage_report.ProfileEvent, ) !AppendOutcome { try validateEvent(event); const event_line = try serializeEvent(alloc, event); defer alloc.free(event_line); if (event_line.len > max_record_bytes) return error.UsageRecordTooLarge;
try self.ensureWritable(); var lock = self.acquireLock() catch |err| switch (err) { error.LockBusy => return error.UsageLockBusy, error.LockUnsupported => return error.UsageLockUnsupported, else => return err, }; defer lock.release();
var file: ?std.Io.File = (try self.openUsage(true, true)).?; defer if (file) |open_file| open_file.close(io_mod.getIo()); const tail = try inspectTail(file.?); const has_incomplete_tail = tail.incomplete; var boundary = tail.length; const index = try self.ensureIndex(alloc, file.?, boundary);
const decision = classifyEvent(index, event);
const now_ms = @max(io_mod.milliTimestamp(), 0); var append_bytes: std.Io.Writer.Allocating = .init(alloc); defer append_bytes.deinit(); var append_record_count: usize = 0; if (index.coverage_started_at_ms == null) { try writeCoverage(&append_bytes.writer, now_ms); append_record_count += 1; } if (has_incomplete_tail) { try writeIncident(&append_bytes.writer, .{ .occurred_at_ms = now_ms, .completeness = .incomplete, }); append_record_count += 1; } if (decision.write) { try append_bytes.writer.writeAll(event_line); append_record_count += 1; } if (append_bytes.written().len == 0) return decision.outcome;
var next_length = std.math.add( u64, boundary, std.math.cast(u64, append_bytes.written().len) orelse return error.UsageCapacityExceeded, ) catch return error.UsageCapacityExceeded; var compacted_before_append = false; var append_committed = false; var base_record_count = index.record_count; if (shouldCompactBeforeAppend( next_length, base_record_count, append_record_count, index, now_ms, )) { compacted_before_append = true; base_record_count = retainedRecordCount(index, now_ms); if (has_incomplete_tail) { var replacement: std.Io.Writer.Allocating = .init(alloc); defer replacement.deinit(); try writeRetainedRecords(&replacement.writer, index, now_ms); try replacement.writer.writeAll(append_bytes.written()); next_length = std.math.cast(u64, replacement.written().len) orelse return error.UsageCapacityExceeded; if (next_length > max_file_bytes) return error.UsageCapacityExceeded; _ = try checkedRecordCount(base_record_count, append_record_count); try self.replaceUsageLocked(alloc, replacement.written()); file.?.close(io_mod.getIo()); file = null; append_committed = true; } else { file.?.close(io_mod.getIo()); file = null; try self.compactLocked(alloc, now_ms); file = (try self.openUsage(true, false)) orelse return error.UsageReadFailed; boundary = try file.?.length(io_mod.getIo()); next_length = std.math.add( u64, boundary, std.math.cast(u64, append_bytes.written().len) orelse return error.UsageCapacityExceeded, ) catch return error.UsageCapacityExceeded; } } if (next_length > max_file_bytes) return error.UsageCapacityExceeded; _ = try checkedRecordCount(base_record_count, append_record_count); if (!append_committed) { if (has_incomplete_tail) { var replacement: std.Io.Writer.Allocating = .init(alloc); defer replacement.deinit(); try copyFilePrefix( &replacement.writer, file.?, boundary, ); try replacement.writer.writeAll(append_bytes.written()); try self.replaceUsageLocked(alloc, replacement.written()); file.?.close(io_mod.getIo()); file = null; } else { try file.?.writePositionalAll(io_mod.getIo(), append_bytes.written(), boundary); file.?.sync(io_mod.getIo()) catch return error.UsageWriteFailed; } }
const compact_after = shouldCompactAfterAppend( next_length, index, eventTimestamp(event), now_ms, compacted_before_append, ); self.noteAppendedRecords(alloc, append_bytes.written(), next_length, compacted_before_append); if (compact_after) { if (file) |open_file| { open_file.close(io_mod.getIo()); file = null; } try self.compactLocked(alloc, now_ms); self.invalidateIndex(); } return decision.outcome; }
/// Rebuilds or incrementally extends the resident parse of the ledger so an /// append pays for new bytes only. The store lock is already held by the /// caller; other processes may have appended since our last boundary. fn ensureIndex( self: *Store, alloc: Allocator, file: std.Io.File, boundary: u64, ) !*RecordIndex { if (self.index_alloc) |index_alloc| { if (index_alloc.ptr != alloc.ptr or index_alloc.vtable != alloc.vtable) { self.invalidateIndex(); } } if (self.index) |*index| { if (boundary == index.boundary and verifyTailSample(index, file)) return index; if (boundary > index.boundary and self.tryAbsorbTail(alloc, index, file, boundary)) { return index; } self.invalidateIndex(); } if (boundary > max_file_bytes) return error.UsageCapacityExceeded; var fresh: RecordIndex = .{}; errdefer fresh.deinit(alloc); if (boundary > 0) { const byte_len = std.math.cast(usize, boundary) orelse return error.UsageCapacityExceeded; const bytes = try alloc.alloc(u8, byte_len); defer alloc.free(bytes); const read_count = try file.readPositionalAll(io_mod.getIo(), bytes, 0); if (read_count != byte_len) return error.UsageReadFailed; if (bytes[bytes.len - 1] != '\n') return error.UsageStoreIncomplete; try absorbBytes(&fresh, alloc, bytes); fresh.boundary = boundary; fresh.captureTailSample(bytes); } self.index = fresh; self.index_alloc = alloc; return &self.index.?; }
fn tryAbsorbTail( self: *Store, alloc: Allocator, index: *RecordIndex, file: std.Io.File, boundary: u64, ) bool { _ = self; if (boundary - index.boundary > max_file_bytes) return false; if (!verifyTailSample(index, file)) return false; const start = index.boundary; const tail_len: usize = @intCast(boundary - start); const bytes = alloc.alloc(u8, tail_len) catch return false; defer alloc.free(bytes); const read_count = file.readPositionalAll(io_mod.getIo(), bytes, start) catch return false; if (read_count != tail_len) return false; if (bytes.len == 0 or bytes[bytes.len - 1] != '\n') return false; absorbBytes(index, alloc, bytes) catch return false; index.boundary = boundary; index.captureTailSample(bytes); index.incremental_absorbs += 1; return true; }
/// After a committed append the index tracks the file by absorbing the /// lines we wrote; a compaction replaced the content instead. fn noteAppendedRecords( self: *Store, alloc: Allocator, written: []const u8, final_length: u64, content_replaced: bool, ) void { if (content_replaced) { self.invalidateIndex(); return; } const index = &(self.index orelse return); absorbBytes(index, alloc, written) catch { self.invalidateIndex(); return; }; index.boundary = final_length; index.captureTailSample(written); }
fn invalidateIndex(self: *Store) void { if (self.index) |*index| { if (self.index_alloc) |index_alloc| index.deinit(index_alloc); } self.index = null; self.index_alloc = null; }
/// Loads one stable read-only boundary without creating profile state. pub fn load(self: *Store, alloc: Allocator) !Loaded { if (self.durable_home == null) { self.durable_home = try openExistingDurableHome(self.home_path); } try self.validateReadable(); while (true) { const existing_lock = self.acquireExistingLock() catch |err| switch (err) { error.LockBusy => return error.UsageLockBusy, error.LockUnsupported => return error.UsageLockUnsupported, else => return err, }; if (existing_lock) |held| { var lock = held; defer lock.release(); return self.loadUnlocked(alloc); }
var loaded = try self.loadUnlocked(alloc); if (!try self.lockFileExists()) return loaded; loaded.deinit(alloc); } }
fn loadUnlocked(self: *Store, alloc: Allocator) !Loaded { var file = (try self.openUsage(false, false)) orelse return .{ .coverage_started_at_ms = null, .facts = try alloc.alloc(usage_report.GenerationFact, 0), .pending = try alloc.alloc(usage_report.PendingMarker, 0), .incidents = try alloc.alloc(usage_report.Incident, 0), .record_count = 0, }; defer file.close(io_mod.getIo()); const boundary = try file.length(io_mod.getIo()); return self.loadFromFile(alloc, file, boundary); }
fn validateReadable(self: *Store) !void { const durable_home = self.durable_home orelse return; const stat = try durable_home.dir.stat(io_mod.getIo()); if (stat.kind != .directory) return error.DurablePathUnsafe; if (stat.permissions.toMode() & 0o777 != 0o700) { return error.PrivateStatePermissionsUnsupported; } }
fn ensureWritable(self: *Store) !void { if (self.durable_home == null) { const zio = io_mod.getIo(); var home = io_mod.VerifiedDir{ .dir = std.Io.Dir.openDirAbsolute(zio, self.home_path, .{ .iterate = true, }) catch return error.DurableLayoutFailed, }; defer home.close(); self.durable_home = io_mod.openOrCreateVerifiedPrivateDir( &home, profile_paths.root_dir_name, ) catch |err| switch (err) { error.PrivateStatePermissionsUnsupported, error.DurablePathUnsafe => return err, else => return error.DurableLayoutFailed, }; } self.durable_home.?.dir.setPermissions( io_mod.getIo(), private_dir_permissions, ) catch return error.PrivateStatePermissionsUnsupported; const stat = try self.durable_home.?.dir.stat(io_mod.getIo()); if (stat.kind != .directory) return error.DurablePathUnsafe; if (stat.permissions.toMode() & 0o777 != 0o700) { return error.PrivateStatePermissionsUnsupported; } }
fn acquireLock(self: *Store) !io_mod.TimedAdvisoryLock { return io_mod.acquireTimedAdvisoryLockWithOps( &self.durable_home.?, usage_lock_file, lock_deadline_ms, self.lock_ops, ); }
fn acquireExistingLock(self: *Store) !?io_mod.TimedAdvisoryLock { const durable_home = self.durable_home orelse return null; const zio = io_mod.getIo(); const file = durable_home.dir.openFile(zio, usage_lock_file, .{ .mode = .read_write, .allow_directory = false, .follow_symlinks = false, .resolve_beneath = true, }) catch |err| switch (err) { error.FileNotFound => return null, error.SymLinkLoop, error.IsDir, error.NotDir => return error.DurablePathUnsafe, else => return err, }; errdefer file.close(zio); const stat = try file.stat(zio); if (stat.kind != .file or stat.nlink != 1) { return error.DurablePathUnsafe; } if (stat.permissions.toMode() & 0o777 != 0o600) { return error.PrivateStatePermissionsUnsupported; }
const started = self.lock_ops.now_ms(self.lock_ops.ctx); const deadline = started + @as(i64, @intCast(lock_deadline_ms)); while (true) { const locked = self.lock_ops.try_lock( self.lock_ops.ctx, file, ) catch |err| switch (err) { error.FileLocksUnsupported => return error.LockUnsupported, else => return err, }; if (locked) return .{ .file = file }; if (self.lock_ops.now_ms(self.lock_ops.ctx) >= deadline) { return error.LockBusy; } self.lock_ops.sleep_ms( self.lock_ops.ctx, @min(@as(u64, 10), lock_deadline_ms), ); } }
fn lockFileExists(self: *Store) !bool { const durable_home = self.durable_home orelse return false; const stat = durable_home.dir.statFile(io_mod.getIo(), usage_lock_file, .{ .follow_symlinks = false, }) catch |err| switch (err) { error.FileNotFound => return false, error.SymLinkLoop, error.NotDir => return error.DurablePathUnsafe, else => return err, }; if (stat.kind != .file or stat.nlink != 1) { return error.DurablePathUnsafe; } if (stat.permissions.toMode() & 0o777 != 0o600) { return error.PrivateStatePermissionsUnsupported; } return true; }
fn openUsage( self: *Store, writable: bool, create: bool, ) !?std.Io.File { if (self.durable_home == null) return null; const zio = io_mod.getIo(); var created = false; var file = openExistingUsageFile( self.durable_home.?.dir, if (writable) .read_write else .read_only, ) catch |err| switch (err) { error.FileNotFound => create_file: { if (!create) return null; const new_file = self.durable_home.?.dir.createFile(zio, usage_file, .{ .read = true, .truncate = false, .exclusive = true, .permissions = private_file_permissions, .resolve_beneath = true, }) catch |create_err| switch (create_err) { error.PathAlreadyExists => return self.openUsage(writable, false), else => return create_err, }; created = true; break :create_file new_file; }, error.SymLinkLoop, error.IsDir, error.NotDir => return error.DurablePathUnsafe, else => return err, }; errdefer file.close(zio);
const initial = try file.stat(zio); try io_mod.verifyOpenedRegularFile( initial, if (writable) .read_write else .read_only, ); if (writable) { file.setPermissions(zio, private_file_permissions) catch return error.PrivateStatePermissionsUnsupported; } const verified = if (writable) try file.stat(zio) else initial; if (verified.permissions.toMode() & 0o777 != 0o600) { return error.PrivateStatePermissionsUnsupported; } if (created) { io_mod.syncVerifiedDir(self.durable_home.?.dir) catch return error.DurableLayoutFailed; } return file; }
fn loadFromFile( self: *Store, alloc: Allocator, file: std.Io.File, boundary: u64, ) !Loaded { _ = self; if (boundary > max_file_bytes) return error.UsageCapacityExceeded; if (boundary == 0) { return .{ .coverage_started_at_ms = null, .facts = try alloc.alloc(usage_report.GenerationFact, 0), .pending = try alloc.alloc(usage_report.PendingMarker, 0), .incidents = try alloc.alloc(usage_report.Incident, 0), .record_count = 0, }; } const byte_len = std.math.cast(usize, boundary) orelse return error.UsageCapacityExceeded; const bytes = try alloc.alloc(u8, byte_len); defer alloc.free(bytes); const read_count = try file.readPositionalAll(io_mod.getIo(), bytes, 0); if (read_count != byte_len) return error.UsageReadFailed; if (bytes[bytes.len - 1] != '\n') return error.UsageStoreIncomplete;
var index: RecordIndex = .{}; defer index.deinit(alloc); try absorbBytes(&index, alloc, bytes); return index.toLoaded(alloc); }
fn compactLocked(self: *Store, alloc: Allocator, now_ms: i64) !void { var file = (try self.openUsage(false, false)) orelse return; defer file.close(io_mod.getIo()); const boundary = try file.length(io_mod.getIo()); const index = try self.ensureIndex(alloc, file, boundary);
var replacement: std.Io.Writer.Allocating = .init(alloc); defer replacement.deinit(); try writeRetainedRecords(&replacement.writer, index, now_ms); try self.replaceUsageLocked(alloc, replacement.written()); }
fn replaceUsageLocked( self: *Store, alloc: Allocator, contents: []const u8, ) !void { io_mod.durableReplaceVerified( alloc, &self.durable_home.?, usage_file, contents, ) catch |err| switch (err) { error.DurableReplacePostRenameFailed => return error.UsageCommitIndeterminate, error.DurableReplacePreRenameFailed => return error.UsageWriteFailed, else => return err, }; }};
fn openExistingUsageFile( dir: std.Io.Dir, mode: std.Io.Dir.OpenFileOptions.Mode,) !std.Io.File { return io_mod.openExistingRegularFile(dir, usage_file, mode) catch |err| switch (err) { error.FileControlFailed => error.UsageReadFailed, else => err, };}
fn openExistingDurableHome(home_path: []const u8) !?io_mod.VerifiedDir { const zio = io_mod.getIo(); var home = try std.Io.Dir.openDirAbsolute(zio, home_path, .{ .iterate = true }); defer home.close(zio);
if (home.openDir(zio, profile_paths.root_dir_name, .{ .iterate = true, .follow_symlinks = false, })) |dir| { errdefer dir.close(zio); const stat = try dir.stat(zio); if (stat.kind != .directory) return error.DurablePathUnsafe; return .{ .dir = dir }; } else |err| switch (err) { error.FileNotFound => return null, error.NotDir, error.SymLinkLoop => return error.DurablePathUnsafe, else => return err, }}
fn retentionCutoff(now_ms: i64) i64 { return std.math.sub(i64, now_ms, retention_ms) catch 0;}
fn hasExpiredRecords(index: *const RecordIndex, now_ms: i64) bool { const cutoff_ms = retentionCutoff(now_ms); for (index.facts.items) |fact| { if (fact.created_at_ms < cutoff_ms) return true; } for (index.pending.items) |marker| { if (marker.observed_at_ms < cutoff_ms or pendingResolved(index, marker.id)) { return true; } } for (index.incidents.items) |incident| { if (incident.occurred_at_ms < cutoff_ms) return true; } return false;}
/// Age-only expiry for the after-append check. A resolved pending marker is/// redundant, but it is a ~100 byte line; rewriting the whole ledger to drop it/// right after every turn would cost far more than keeping it until genuinely/// aged records accumulate.fn hasAgedRecords(index: *const RecordIndex, now_ms: i64) bool { const cutoff_ms = retentionCutoff(now_ms); for (index.facts.items) |fact| { if (fact.created_at_ms < cutoff_ms) return true; } for (index.pending.items) |marker| { if (marker.observed_at_ms < cutoff_ms) return true; } for (index.incidents.items) |incident| { if (incident.occurred_at_ms < cutoff_ms) return true; } return false;}
fn pendingResolved(index: *const RecordIndex, id: []const u8) bool { return index.fact_indexes.contains(id);}
fn checkedRecordCount(current: usize, additional: usize) !usize { const next = std.math.add(usize, current, additional) catch return error.UsageCapacityExceeded; if (next > max_records) return error.UsageCapacityExceeded; return next;}
fn exceedsRecordCapacity(current: usize, additional: usize) bool { _ = checkedRecordCount(current, additional) catch return true; return false;}
fn retainedRecordCount(index: *const RecordIndex, now_ms: i64) usize { const cutoff_ms = retentionCutoff(now_ms); var count: usize = @intFromBool(index.coverage_started_at_ms != null); for (index.facts.items) |fact| { if (fact.created_at_ms >= cutoff_ms) count += 1; } for (index.pending.items) |marker| { if (marker.observed_at_ms >= cutoff_ms and !pendingResolved(index, marker.id)) { count += 1; } } for (index.incidents.items) |incident| { if (incident.occurred_at_ms >= cutoff_ms) count += 1; } return count;}
fn shouldCompactBeforeAppend( next_length: u64, record_count: usize, append_record_count: usize, index: *const RecordIndex, now_ms: i64,) bool { return (next_length > max_file_bytes or exceedsRecordCapacity(record_count, append_record_count)) and hasExpiredRecords(index, now_ms);}
fn shouldCompactAfterAppend( next_length: u64, index: *const RecordIndex, appended_at_ms: i64, now_ms: i64, compacted_before_append: bool,) bool { return !compacted_before_append and next_length > compaction_threshold_bytes and (hasAgedRecords(index, now_ms) or appended_at_ms < retentionCutoff(now_ms));}
fn inspectTail(file: std.Io.File) !TailBoundary { const length = try file.length(io_mod.getIo()); if (length == 0) return .{ .length = 0, .incomplete = false }; var last: [1]u8 = undefined; const count = try file.readPositionalAll(io_mod.getIo(), &last, length - 1); if (count == 1 and last[0] == '\n') { return .{ .length = length, .incomplete = false }; }
var cursor = length; var buffer: [8192]u8 = undefined; while (cursor > 0) { const start = cursor - @min(cursor, buffer.len); const read_len: usize = @intCast(cursor - start); const read_count = try file.readPositionalAll( io_mod.getIo(), buffer[0..read_len], start, ); if (read_count != read_len) return error.UsageWriteFailed; if (std.mem.lastIndexOfScalar(u8, buffer[0..read_count], '\n')) |newline| { return .{ .length = start + newline + 1, .incomplete = true, }; } cursor = start; } return .{ .length = 0, .incomplete = true };}
fn copyFilePrefix( writer: *std.Io.Writer, file: std.Io.File, boundary: u64,) !void { var offset: u64 = 0; var buffer: [8192]u8 = undefined; while (offset < boundary) { const remaining = boundary - offset; const read_len: usize = @intCast(@min(remaining, buffer.len)); const read_count = try file.readPositionalAll( io_mod.getIo(), buffer[0..read_len], offset, ); if (read_count != read_len) return error.UsageReadFailed; try writer.writeAll(buffer[0..read_count]); offset += read_count; }}
fn writeRetainedRecords( writer: *std.Io.Writer, index: *const RecordIndex, now_ms: i64,) !void { const cutoff_ms = retentionCutoff(now_ms); if (index.coverage_started_at_ms) |started_at_ms| { try writeCoverage(writer, started_at_ms); } for (index.facts.items) |fact| { if (fact.created_at_ms < cutoff_ms) continue; try writeGeneration(writer, fact); } for (index.pending.items) |marker| { if (marker.observed_at_ms < cutoff_ms or pendingResolved(index, marker.id)) { continue; } try writePending(writer, marker); } for (index.incidents.items) |incident| { if (incident.occurred_at_ms < cutoff_ms) continue; try writeIncident(writer, incident); }}
fn validateEvent(event: usage_report.ProfileEvent) !void { switch (event) { .generation => |fact| try usage_report.validateFact(fact), .pending => |marker| try usage_report.validatePendingMarker(marker), .incident => |incident| try usage_report.validateIncident(incident), }}
fn classifyEvent( index: *const RecordIndex, event: usage_report.ProfileEvent,) AppendDecision { return switch (event) { .generation => |fact| classifyGeneration(index, fact), .pending => |marker| classifyPending(index, marker), .incident => |incident| blk: { if (index.incidents.items.len >= max_file_incidents) { debug_trace.logf( "session", "usage profile incident dropped reason=cap limit={d}", .{max_file_incidents}, ); break :blk .{ .outcome = .duplicate, .write = false }; } for (index.incidents.items) |existing| { if (existing.occurred_at_ms == incident.occurred_at_ms and existing.completeness == incident.completeness) { break :blk .{ .outcome = .duplicate, .write = false }; } } break :blk .{ .outcome = .appended, .write = true }; }, };}
fn classifyGeneration( index: *const RecordIndex, fact: usage_report.GenerationFact,) AppendDecision { const indexes = index.fact_indexes.get(fact.id) orelse return .{ .outcome = .appended, .write = true }; if (usage_report.GenerationFact.eql(index.facts.items[indexes.first], fact)) { return .{ .outcome = .duplicate, .write = false }; } if (indexes.second) |second| { if (usage_report.GenerationFact.eql(index.facts.items[second], fact)) { return .{ .outcome = .duplicate, .write = false }; } return .{ .outcome = .conflict, .write = false }; } return .{ .outcome = .conflict, .write = true };}
fn classifyPending( index: *const RecordIndex, marker: usage_report.PendingMarker,) AppendDecision { const indexes = index.pending_indexes.get(marker.id) orelse return .{ .outcome = .appended, .write = true }; if (usage_report.PendingMarker.eql(index.pending.items[indexes.first], marker)) { return .{ .outcome = .duplicate, .write = false }; } if (indexes.second) |second| { if (usage_report.PendingMarker.eql(index.pending.items[second], marker)) { return .{ .outcome = .duplicate, .write = false }; } return .{ .outcome = .conflict, .write = false }; } return .{ .outcome = .conflict, .write = true };}
fn eventTimestamp(event: usage_report.ProfileEvent) i64 { return switch (event) { .generation => |fact| fact.created_at_ms, .pending => |marker| marker.observed_at_ms, .incident => |incident| incident.occurred_at_ms, };}
fn serializeEvent( alloc: Allocator, event: usage_report.ProfileEvent,) ![]u8 { var writer: std.Io.Writer.Allocating = .init(alloc); errdefer writer.deinit(); switch (event) { .generation => |fact| try writeGeneration(&writer.writer, fact), .pending => |marker| try writePending(&writer.writer, marker), .incident => |incident| try writeIncident(&writer.writer, incident), } return writer.toOwnedSlice();}
fn writeCoverage(writer: *std.Io.Writer, started_at_ms: i64) !void { try writer.print( "{{\"schema_version\":1,\"kind\":\"coverage\",\"started_at_ms\":{d}}}\n", .{started_at_ms}, );}
fn writeGeneration( writer: *std.Io.Writer, fact: usage_report.GenerationFact,) !void { try writer.writeAll( "{\"schema_version\":1,\"kind\":\"generation\",\"fact\":", ); try generation_fact_codec.write(writer, fact); try writer.writeAll("}\n");}
fn writePending( writer: *std.Io.Writer, marker: usage_report.PendingMarker,) !void { try writer.writeAll("{\"schema_version\":1,\"kind\":\"pending\",\"id\":"); try std.json.Stringify.value(marker.id, .{}, writer); try writer.print(",\"observed_at_ms\":{d}}}\n", .{marker.observed_at_ms});}
fn writeIncident( writer: *std.Io.Writer, incident: usage_report.Incident,) !void { try writer.print( "{{\"schema_version\":1,\"kind\":\"incident\",\"occurred_at_ms\":{d},\"completeness\":", .{incident.occurred_at_ms}, ); try std.json.Stringify.value(@tagName(incident.completeness), .{}, writer); try writer.writeAll("}\n");}
fn parseRecord(alloc: Allocator, line: []const u8) !ParsedRecord { var parsed = std.json.parseFromSlice( std.json.Value, alloc, line, .{ .allocate = .alloc_always, .parse_numbers = false }, ) catch |err| switch (err) { error.OutOfMemory => return error.OutOfMemory, else => return error.InvalidUsageStore, }; defer parsed.deinit(); if (parsed.value != .object) return error.InvalidUsageStore; const schema = try parseU64(parsed.value.object.get("schema_version")); const kind_value = parsed.value.object.get("kind") orelse return error.InvalidUsageStore; if (schema != 1 or kind_value != .string) return error.InvalidUsageStore;
if (std.mem.eql(u8, kind_value.string, "coverage")) { if (parsed.value.object.count() != 3) return error.InvalidUsageStore; return .{ .coverage = try parseI64(parsed.value.object.get("started_at_ms")) }; } if (std.mem.eql(u8, kind_value.string, "pending")) { if (parsed.value.object.count() != 4) return error.InvalidUsageStore; const id_value = parsed.value.object.get("id") orelse return error.InvalidUsageStore; if (id_value != .string) return error.InvalidUsageStore; const id = try alloc.dupe(u8, id_value.string); errdefer alloc.free(id); const marker = usage_report.PendingMarker{ .id = id, .observed_at_ms = try parseI64( parsed.value.object.get("observed_at_ms"), ), }; usage_report.validatePendingMarker(marker) catch return error.InvalidUsageStore; return .{ .pending = marker }; } if (std.mem.eql(u8, kind_value.string, "incident")) { if (parsed.value.object.count() != 4) return error.InvalidUsageStore; const completeness_value = parsed.value.object.get("completeness") orelse return error.InvalidUsageStore; if (completeness_value != .string) return error.InvalidUsageStore; const completeness = std.meta.stringToEnum( usage_report.Completeness, completeness_value.string, ) orelse return error.InvalidUsageStore; if (completeness != .pending and completeness != .incomplete) { return error.InvalidUsageStore; } return .{ .incident = .{ .occurred_at_ms = try parseI64(parsed.value.object.get("occurred_at_ms")), .completeness = completeness, } }; } if (!std.mem.eql(u8, kind_value.string, "generation") or parsed.value.object.count() != 3) { return error.InvalidUsageStore; } const fact_value = parsed.value.object.get("fact") orelse return error.InvalidUsageStore; return .{ .generation = generation_fact_codec.parse( alloc, fact_value, ) catch |err| switch (err) { error.OutOfMemory => return error.OutOfMemory, error.InvalidGenerationFact => return error.InvalidUsageStore, } };}
fn parseU64(value: ?std.json.Value) !u64 { const actual = value orelse return error.InvalidUsageStore; return switch (actual) { .integer => |number| if (number >= 0) std.math.cast(u64, number) orelse error.InvalidUsageStore else error.InvalidUsageStore, .number_string => |text| std.fmt.parseInt(u64, text, 10) catch error.InvalidUsageStore, else => error.InvalidUsageStore, };}
fn parseI64(value: ?std.json.Value) !i64 { const number = try parseU64(value); return std.math.cast(i64, number) orelse error.InvalidUsageStore;}
fn makeTestFact( alloc: Allocator, id: []const u8, created_at_ms: i64, input_tokens: u64,) !usage_report.GenerationFact { const id_copy = try alloc.dupe(u8, id); errdefer alloc.free(id_copy); return .{ .id = id_copy, .created_at_ms = created_at_ms, .model = try alloc.dupe(u8, "provider/model"), .input_tokens = input_tokens, .output_tokens = 2, .cache_read_tokens = 1, .cache_write_tokens = 0, .reasoning_tokens = 1, .total_cost = 0.25, };}
test "profile usage store is read-only until first fact and dedupes exact replay" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const home = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "."); defer alloc.free(home); var store = try Store.initFromHome(alloc, home); defer store.deinit(alloc);
var empty = try store.load(alloc); defer empty.deinit(alloc); try std.testing.expect(empty.coverage_started_at_ms == null);
var fact = try makeTestFact( alloc, "gen_01ARZ3NDEKTSV4RRFFQ69G5FAV", 1000, 10, ); defer fact.deinit(alloc); try std.testing.expectEqual(AppendOutcome.appended, try store.appendFact(alloc, fact)); try std.testing.expectEqual(AppendOutcome.duplicate, try store.appendFact(alloc, fact));
var loaded = try store.load(alloc); defer loaded.deinit(alloc); try std.testing.expect(loaded.coverage_started_at_ms.? > 0); try std.testing.expectEqual(@as(usize, 1), loaded.facts.len); try std.testing.expectEqual(@as(usize, 0), loaded.incidents.len);}
test "profile usage store discovers profile state created after initialization" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const home = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "."); defer alloc.free(home); var observer = try Store.initFromHome(alloc, home); defer observer.deinit(alloc); var writer = try Store.initFromHome(alloc, home); defer writer.deinit(alloc);
var fact = try makeTestFact( alloc, "gen_01ARZ3NDEKTSV4RRFFQ69G5FAV", 1000, 10, ); defer fact.deinit(alloc); try std.testing.expectEqual( AppendOutcome.appended, try writer.appendFact(alloc, fact), );
var loaded = try observer.load(alloc); defer loaded.deinit(alloc); try std.testing.expect(loaded.coverage_started_at_ms != null); try std.testing.expectEqual(@as(usize, 1), loaded.facts.len); try std.testing.expect(usage_report.GenerationFact.eql( fact, loaded.facts[0], ));}
test "profile usage readers wait for the cross-process writer boundary" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const home = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "."); defer alloc.free(home); var writer = try Store.initFromHome(alloc, home); defer writer.deinit(alloc); var reader = try Store.initFromHome(alloc, home); defer reader.deinit(alloc); var fact = try makeTestFact( alloc, "gen_01ARZ3NDEKTSV4RRFFQ69G5FAV", 1000, 10, ); defer fact.deinit(alloc); try std.testing.expectEqual( AppendOutcome.appended, try writer.appendFact(alloc, fact), );
const Worker = struct { store: *Store, started: std.atomic.Value(bool) = .init(false), done: std.atomic.Value(bool) = .init(false), fact_count: usize = 0, failure: ?anyerror = null,
fn run(self: *@This()) void { self.started.store(true, .seq_cst); var loaded = self.store.load(std.heap.c_allocator) catch |err| { self.failure = err; self.done.store(true, .seq_cst); return; }; defer loaded.deinit(std.heap.c_allocator); self.fact_count = loaded.facts.len; self.done.store(true, .seq_cst); } }; var worker = Worker{ .store = &reader }; var lock = try writer.acquireLock(); const thread = try std.Thread.spawn(.{}, Worker.run, .{&worker}); while (!worker.started.load(.seq_cst)) std.Thread.yield() catch {}; for (0..100) |_| std.Thread.yield() catch {}; const blocked_at_boundary = !worker.done.load(.seq_cst); lock.release(); thread.join();
try std.testing.expect(blocked_at_boundary); try std.testing.expect(worker.failure == null); try std.testing.expectEqual(@as(usize, 1), worker.fact_count);}
test "profile usage store preserves conflicting duplicates as incomplete evidence" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const home = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "."); defer alloc.free(home); var store = try Store.initFromHome(alloc, home); defer store.deinit(alloc);
var first = try makeTestFact( alloc, "gen_01ARZ3NDEKTSV4RRFFQ69G5FAV", 1000, 10, ); defer first.deinit(alloc); var conflict = try makeTestFact( alloc, "gen_01ARZ3NDEKTSV4RRFFQ69G5FAV", 1000, 11, ); defer conflict.deinit(alloc); try std.testing.expectEqual(AppendOutcome.appended, try store.appendFact(alloc, first)); try std.testing.expectEqual(AppendOutcome.conflict, try store.appendFact(alloc, conflict)); var file = (try store.openUsage(false, false)).?; const conflicted_length = try file.length(io_mod.getIo()); file.close(io_mod.getIo()); try std.testing.expectEqual(AppendOutcome.duplicate, try store.appendFact(alloc, conflict)); file = (try store.openUsage(false, false)).?; defer file.close(io_mod.getIo()); try std.testing.expectEqual(conflicted_length, try file.length(io_mod.getIo()));
var loaded = try store.load(alloc); defer loaded.deinit(alloc); try std.testing.expectEqual(@as(usize, 2), loaded.facts.len); try std.testing.expectEqual(@as(usize, 0), loaded.incidents.len);
var report = try usage_report.buildRollingSnapshot( alloc, .hours_24, 2000, 0, loaded.facts, loaded.incidents, ); defer report.deinit(alloc); try std.testing.expectEqual( usage_report.Completeness.incomplete, report.completeness, );}
test "profile usage store repairs an incomplete tail and records the gap" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const home = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "."); defer alloc.free(home); var store = try Store.initFromHome(alloc, home); defer store.deinit(alloc);
var first = try makeTestFact( alloc, "gen_01ARZ3NDEKTSV4RRFFQ69G5FAV", 1000, 10, ); defer first.deinit(alloc); try std.testing.expectEqual( AppendOutcome.appended, try store.appendFact(alloc, first), );
var file = (try store.openUsage(true, false)).?; const length = try file.length(io_mod.getIo()); try file.writePositionalAll(io_mod.getIo(), "{\"schema_version\":1", length); try file.sync(io_mod.getIo()); file.close(io_mod.getIo()); try std.testing.expectError(error.UsageStoreIncomplete, store.load(alloc));
var second = try makeTestFact( alloc, "gen_01ARZ3NDEKTSV4RRFFQ69G5FAW", 2000, 20, ); defer second.deinit(alloc); try std.testing.expectEqual( AppendOutcome.appended, try store.appendFact(alloc, second), ); var loaded = try store.load(alloc); defer loaded.deinit(alloc); try std.testing.expectEqual(@as(usize, 2), loaded.facts.len); try std.testing.expectEqual(@as(usize, 1), loaded.incidents.len); try std.testing.expectEqual( usage_report.Completeness.incomplete, loaded.incidents[0].completeness, );}
test "profile usage store records a repaired tail when the replayed fact is duplicate" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const home = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "."); defer alloc.free(home); var store = try Store.initFromHome(alloc, home); defer store.deinit(alloc);
var fact = try makeTestFact( alloc, "gen_01ARZ3NDEKTSV4RRFFQ69G5FAV", 1000, 10, ); defer fact.deinit(alloc); try std.testing.expectEqual( AppendOutcome.appended, try store.appendFact(alloc, fact), );
var file = (try store.openUsage(true, false)).?; const length = try file.length(io_mod.getIo()); try file.writePositionalAll(io_mod.getIo(), "{\"schema_version\":1", length); try file.sync(io_mod.getIo()); file.close(io_mod.getIo());
try std.testing.expectEqual( AppendOutcome.duplicate, try store.appendFact(alloc, fact), ); var loaded = try store.load(alloc); defer loaded.deinit(alloc); try std.testing.expectEqual(@as(usize, 1), loaded.facts.len); try std.testing.expectEqual(@as(usize, 1), loaded.incidents.len); try std.testing.expectEqual( usage_report.Completeness.incomplete, loaded.incidents[0].completeness, );}
test "profile usage store leaves an incomplete tail intact when repair exceeds record capacity" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); try tmp.dir.createDir( io_mod.getIo(), ".fx", std.Io.Dir.Permissions.fromMode(0o700), ); var profile = try tmp.dir.openDir(io_mod.getIo(), ".fx", .{ .iterate = true }); defer profile.close(io_mod.getIo()); profile.setPermissions(io_mod.getIo(), .fromMode(0o700)) catch return error.SkipZigTest;
var contents: std.Io.Writer.Allocating = .init(alloc); defer contents.deinit(); for (0..max_records) |_| try writeCoverage(&contents.writer, 1); try contents.writer.writeAll("{\"schema_version\":1"); var file = try profile.createFile(io_mod.getIo(), usage_file, .{ .permissions = private_file_permissions, }); try file.writeStreamingAll(io_mod.getIo(), contents.written()); const original_length = try file.length(io_mod.getIo()); file.close(io_mod.getIo()); var lock = try profile.createFile(io_mod.getIo(), usage_lock_file, .{ .permissions = private_file_permissions, }); lock.close(io_mod.getIo());
const home = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "."); defer alloc.free(home); var store = try Store.initFromHome(alloc, home); defer store.deinit(alloc); var fact = try makeTestFact( alloc, "gen_01ARZ3NDEKTSV4RRFFQ69G5FAV", 2, 1, ); defer fact.deinit(alloc); try std.testing.expectError( error.UsageCapacityExceeded, store.appendFact(alloc, fact), );
var unchanged = (try store.openUsage(false, false)).?; defer unchanged.close(io_mod.getIo()); try std.testing.expectEqual( original_length, try unchanged.length(io_mod.getIo()), ); try std.testing.expectError(error.UsageStoreIncomplete, store.load(alloc));}
test "profile usage store repairs an existing profile directory to private mode" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); try tmp.dir.createDir( io_mod.getIo(), ".fx", std.Io.File.Permissions.fromMode(0o755), ); var profile = try tmp.dir.openDir(io_mod.getIo(), ".fx", .{ .iterate = true }); defer profile.close(io_mod.getIo()); profile.setPermissions(io_mod.getIo(), .fromMode(0o755)) catch return error.SkipZigTest;
const home = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "."); defer alloc.free(home); var store = try Store.initFromHome(alloc, home); defer store.deinit(alloc); var fact = try makeTestFact( alloc, "gen_01ARZ3NDEKTSV4RRFFQ69G5FAV", 1000, 10, ); defer fact.deinit(alloc); try std.testing.expectEqual( AppendOutcome.appended, try store.appendFact(alloc, fact), );
const stat = try profile.stat(io_mod.getIo()); try std.testing.expectEqual(@as(u32, 0o700), stat.permissions.toMode() & 0o777);}
test "profile usage reads reject an unsafe profile directory without repairing it" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); try tmp.dir.createDir( io_mod.getIo(), ".fx", std.Io.File.Permissions.fromMode(0o755), ); var profile = try tmp.dir.openDir(io_mod.getIo(), ".fx", .{ .iterate = true }); defer profile.close(io_mod.getIo()); profile.setPermissions(io_mod.getIo(), .fromMode(0o755)) catch return error.SkipZigTest;
var contents: std.Io.Writer.Allocating = .init(alloc); defer contents.deinit(); try writeCoverage(&contents.writer, 1); try writeGeneration(&contents.writer, .{ .id = @constCast("gen_01ARZ3NDEKTSV4RRFFQ69G5FAV"), .created_at_ms = 1, .model = @constCast("provider/model"), .input_tokens = 1, .output_tokens = 1, .cache_read_tokens = 0, .cache_write_tokens = 0, .reasoning_tokens = null, .total_cost = 0.001, }); var file = try profile.createFile(io_mod.getIo(), usage_file, .{ .permissions = private_file_permissions, }); try file.writeStreamingAll(io_mod.getIo(), contents.written()); file.close(io_mod.getIo());
const home = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "."); defer alloc.free(home); var store = try Store.initFromHome(alloc, home); defer store.deinit(alloc); try std.testing.expectError( error.PrivateStatePermissionsUnsupported, store.load(alloc), );
const stat = try profile.stat(io_mod.getIo()); try std.testing.expectEqual(@as(u32, 0o755), stat.permissions.toMode() & 0o777);}
test "profile usage store decodes a large ledger with stable id indexing" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); try tmp.dir.createDir( io_mod.getIo(), ".fx", std.Io.File.Permissions.fromMode(0o700), ); var profile = try tmp.dir.openDir(io_mod.getIo(), ".fx", .{ .iterate = true }); defer profile.close(io_mod.getIo()); profile.setPermissions(io_mod.getIo(), .fromMode(0o700)) catch return error.SkipZigTest;
const record_count: usize = 4096; const now_ms = @max(io_mod.milliTimestamp(), 1); var contents: std.Io.Writer.Allocating = .init(alloc); defer contents.deinit(); try writeCoverage(&contents.writer, now_ms - 1); const alphabet = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"; for (0..record_count) |index| { var id = "gen_00000000000000000000000000".*; var value = index; for (0..3) |digit| { id[id.len - 1 - digit] = alphabet[value & 31]; value >>= 5; } try writeGeneration(&contents.writer, .{ .id = &id, .created_at_ms = now_ms - 1, .model = @constCast("provider/model"), .input_tokens = 1, .output_tokens = 1, .cache_read_tokens = 0, .cache_write_tokens = 0, .reasoning_tokens = null, .total_cost = 0.001, }); } var file = try profile.createFile(io_mod.getIo(), usage_file, .{ .permissions = private_file_permissions, }); try file.writeStreamingAll(io_mod.getIo(), contents.written()); file.close(io_mod.getIo()); var lock = try profile.createFile(io_mod.getIo(), usage_lock_file, .{ .permissions = private_file_permissions, }); lock.close(io_mod.getIo());
const home = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "."); defer alloc.free(home); var store = try Store.initFromHome(alloc, home); defer store.deinit(alloc); var loaded = try store.load(alloc); defer loaded.deinit(alloc); try std.testing.expectEqual(record_count, loaded.facts.len);}
test "profile usage compaction eligibility requires expired records" { const alloc = std.testing.allocator; const now_ms = std.time.ms_per_day * 100; var recent_fact = usage_report.GenerationFact{ .id = @constCast("gen_01ARZ3NDEKTSV4RRFFQ69G5FAV"), .created_at_ms = now_ms - std.time.ms_per_day, .model = @constCast("provider/model"), .input_tokens = 1, .output_tokens = 1, .cache_read_tokens = 0, .cache_write_tokens = 0, .reasoning_tokens = null, .total_cost = 0.001, }; var loaded = Loaded{ .coverage_started_at_ms = 0, .facts = (&recent_fact)[0..1], .pending = &.{}, .incidents = &.{}, .record_count = 2, }; var index = try testIndexFromLoaded(alloc, loaded); defer index.deinit(alloc); try std.testing.expect(!hasExpiredRecords(&index, now_ms)); try std.testing.expect(!shouldCompactAfterAppend( compaction_threshold_bytes + 1, &index, recent_fact.created_at_ms, now_ms, false, )); try std.testing.expect(!shouldCompactBeforeAppend( max_file_bytes + 1, index.record_count, 1, &index, now_ms, )); // A resolved pending marker alone must not trigger an after-append rewrite. var resolved_marker = usage_report.PendingMarker{ .id = @constCast("gen_01ARZ3NDEKTSV4RRFFQ69G5FAV"), .observed_at_ms = now_ms - 1000, }; loaded.pending = (&resolved_marker)[0..1]; loaded.record_count += 1; index.deinit(alloc); index = try testIndexFromLoaded(alloc, loaded); try std.testing.expect(!shouldCompactAfterAppend( compaction_threshold_bytes + 1, &index, now_ms, now_ms, false, )); loaded.pending = &.{}; loaded.facts[0].created_at_ms = now_ms - std.time.ms_per_day * 36; index.deinit(alloc); index = try testIndexFromLoaded(alloc, loaded); try std.testing.expect(hasExpiredRecords(&index, now_ms)); try std.testing.expect(shouldCompactAfterAppend( compaction_threshold_bytes + 1, &index, now_ms - 1, now_ms, false, )); try std.testing.expect(shouldCompactBeforeAppend( max_file_bytes + 1, index.record_count, 1, &index, now_ms, )); try std.testing.expect(shouldCompactBeforeAppend( 0, max_records, 1, &index, now_ms, ));}
fn testIndexFromLoaded(alloc: Allocator, loaded: Loaded) !RecordIndex { var index: RecordIndex = .{}; errdefer index.deinit(alloc); index.coverage_started_at_ms = loaded.coverage_started_at_ms; index.record_count = loaded.record_count; for (loaded.facts) |fact| try absorbRecord(&index, alloc, .{ .generation = fact }); for (loaded.pending) |marker| try absorbRecord(&index, alloc, .{ .pending = marker }); for (loaded.incidents) |incident| try absorbRecord(&index, alloc, .{ .incident = incident }); return index;}
test "profile usage record capacity is checked before append" { try std.testing.expectEqual( max_records, try checkedRecordCount(max_records - 2, 2), ); try std.testing.expectError( error.UsageCapacityExceeded, checkedRecordCount(max_records, 1), ); try std.testing.expectError( error.UsageCapacityExceeded, checkedRecordCount(max_records - 1, 2), );}
test "concurrent profile usage writers serialize without losing facts" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const home = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "."); defer alloc.free(home); var first_store = try Store.initFromHome(alloc, home); defer first_store.deinit(alloc); var second_store = try Store.initFromHome(alloc, home); defer second_store.deinit(alloc);
const Worker = struct { store: *Store, fact: usage_report.GenerationFact, outcome: ?AppendOutcome = null, failure: ?anyerror = null,
fn run(self: *@This()) void { self.outcome = self.store.appendFact( std.heap.c_allocator, self.fact, ) catch |err| { self.failure = err; return; }; } }; var first = Worker{ .store = &first_store, .fact = .{ .id = @constCast("gen_01ARZ3NDEKTSV4RRFFQ69G5FAV"), .created_at_ms = 1000, .model = @constCast("provider/a"), .input_tokens = 10, .output_tokens = 2, .cache_read_tokens = 0, .cache_write_tokens = 0, .reasoning_tokens = null, .total_cost = 0.1, }, }; var second = Worker{ .store = &second_store, .fact = .{ .id = @constCast("gen_01ARZ3NDEKTSV4RRFFQ69G5FAW"), .created_at_ms = 1001, .model = @constCast("provider/b"), .input_tokens = 20, .output_tokens = 4, .cache_read_tokens = 1, .cache_write_tokens = 0, .reasoning_tokens = 1, .total_cost = 0.2, }, }; const first_thread = try std.Thread.spawn(.{}, Worker.run, .{&first}); const second_thread = try std.Thread.spawn(.{}, Worker.run, .{&second}); first_thread.join(); second_thread.join();
try std.testing.expect(first.failure == null); try std.testing.expect(second.failure == null); try std.testing.expectEqual(AppendOutcome.appended, first.outcome.?); try std.testing.expectEqual(AppendOutcome.appended, second.outcome.?); var loaded = try first_store.load(alloc); defer loaded.deinit(alloc); try std.testing.expectEqual(@as(usize, 2), loaded.facts.len);}
test "profile usage store refuses a symlinked ledger leaf" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const home = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "."); defer alloc.free(home); var store = try Store.initFromHome(alloc, home); defer store.deinit(alloc);
var first = try makeTestFact( alloc, "gen_01ARZ3NDEKTSV4RRFFQ69G5FAV", 1000, 10, ); defer first.deinit(alloc); try std.testing.expectEqual( AppendOutcome.appended, try store.appendFact(alloc, first), ); try tmp.dir.deleteFile(io_mod.getIo(), ".fx/usage.jsonl"); var outside = try tmp.dir.createFile(io_mod.getIo(), "outside-usage", .{}); try outside.writeStreamingAll(io_mod.getIo(), "outside"); outside.close(io_mod.getIo()); tmp.dir.symLink( io_mod.getIo(), "../outside-usage", ".fx/usage.jsonl", .{ .is_directory = false }, ) catch |err| switch (err) { error.AccessDenied => return error.SkipZigTest, else => return err, };
try std.testing.expectError(error.DurablePathUnsafe, store.load(alloc)); try std.testing.expectError( error.DurablePathUnsafe, store.appendFact(alloc, first), ); const outside_bytes = try tmp.dir.readFileAlloc( io_mod.getIo(), "outside-usage", alloc, .limited(32), ); defer alloc.free(outside_bytes); try std.testing.expectEqualStrings("outside", outside_bytes);}
test "profile usage store merges foreign appends incrementally" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const home = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "."); defer alloc.free(home); var first = try Store.initFromHome(alloc, home); defer first.deinit(alloc); var second = try Store.initFromHome(alloc, home); defer second.deinit(alloc);
var fact_one = try makeTestFact(alloc, "gen_01ARZ3NDEKTSV4RRFFQ69G5FAV", 1000, 10); defer fact_one.deinit(alloc); var fact_two = try makeTestFact(alloc, "gen_01ARZ3NDEKTSV4RRFFQ69G5FAW", 2000, 20); defer fact_two.deinit(alloc); var fact_three = try makeTestFact(alloc, "gen_01ARZ3NDEKTSV4RRFFQ69G5FAX", 3000, 30); defer fact_three.deinit(alloc);
try std.testing.expectEqual(AppendOutcome.appended, try first.appendFact(alloc, fact_one)); try std.testing.expectEqual(AppendOutcome.appended, try second.appendFact(alloc, fact_two)); // The first store's index is stale here; the duplicate verdict must come // from the incrementally absorbed tail. try std.testing.expectEqual(AppendOutcome.duplicate, try first.appendFact(alloc, fact_one)); try std.testing.expectEqual(@as(usize, 1), first.index.?.incremental_absorbs); try std.testing.expectEqual(AppendOutcome.appended, try first.appendFact(alloc, fact_three));
var loaded = try first.load(alloc); defer loaded.deinit(alloc); try std.testing.expectEqual(@as(usize, 3), loaded.facts.len); var ids = std.StringHashMap(void).init(alloc); defer ids.deinit(); for (loaded.facts) |fact| try ids.put(fact.id, {}); try std.testing.expect(ids.contains("gen_01ARZ3NDEKTSV4RRFFQ69G5FAV")); try std.testing.expect(ids.contains("gen_01ARZ3NDEKTSV4RRFFQ69G5FAW")); try std.testing.expect(ids.contains("gen_01ARZ3NDEKTSV4RRFFQ69G5FAX"));}
test "profile usage store caps incident records in the ledger file" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const home = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "."); defer alloc.free(home); var store = try Store.initFromHome(alloc, home); defer store.deinit(alloc);
var fact = try makeTestFact(alloc, "gen_01ARZ3NDEKTSV4RRFFQ69G5FAV", 1000, 10); defer fact.deinit(alloc); try std.testing.expectEqual(AppendOutcome.appended, try store.appendFact(alloc, fact));
var appended: usize = 0; for (0..max_file_incidents + 4) |i| { const outcome = try store.appendEvent(alloc, .{ .incident = .{ .occurred_at_ms = @intCast(10_000 + i), .completeness = .incomplete, } }); if (outcome == .appended) appended += 1; } try std.testing.expectEqual(max_file_incidents, appended);
var loaded = try store.load(alloc); defer loaded.deinit(alloc); try std.testing.expectEqual(max_file_incidents, loaded.incidents.len);}
test "profile usage store fully re-parses after a foreign replace fools the length check" { const alloc = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); const home = try io_mod.dirRealpathAlloc(alloc, tmp.dir, "."); defer alloc.free(home); var first = try Store.initFromHome(alloc, home); defer first.deinit(alloc);
var fact_one = try makeTestFact(alloc, "gen_01ARZ3NDEKTSV4RRFFQ69G5FAV", 1000, 10); defer fact_one.deinit(alloc); var fact_two = try makeTestFact(alloc, "gen_01ARZ3NDEKTSV4RRFFQ69G5FAW", 2000, 20); defer fact_two.deinit(alloc);
try std.testing.expectEqual(AppendOutcome.appended, try first.appendFact(alloc, fact_one)); const absorbs_before = first.index.?.incremental_absorbs;
// A foreign process compacts and then appends, growing the file while // replacing the bytes the resident index parsed. var profile = try tmp.dir.openDir(io_mod.getIo(), ".fx", .{ .iterate = true }); defer profile.close(io_mod.getIo()); var contents: std.Io.Writer.Allocating = .init(alloc); defer contents.deinit(); try writeCoverage(&contents.writer, 1); try writeGeneration(&contents.writer, fact_two); try writeGeneration(&contents.writer, fact_two); var file = try profile.createFile(io_mod.getIo(), usage_file, .{ .permissions = private_file_permissions, .truncate = true, }); try file.writeStreamingAll(io_mod.getIo(), contents.written()); file.close(io_mod.getIo());
try std.testing.expectEqual(AppendOutcome.appended, try first.appendFact(alloc, fact_one)); try std.testing.expectEqual(absorbs_before, first.index.?.incremental_absorbs);
var loaded = try first.load(alloc); defer loaded.deinit(alloc); try std.testing.expectEqual(@as(usize, 2), loaded.facts.len);}