zzstoatzz.io
This repository has no description
1//! the database is a signed atproto repo.
2//!
3//! every orchestration write becomes a record in an MST, sealed by a signed
4//! v3 commit, and the whole thing persists as a single CAR file on disk.
5//! anyone holding the CAR can verify the full history offline against the
6//! workspace DID — `zat.verifyCommitCar` is the proof.
7//!
8//! entity records (flow, flowRun, taskRun) are keyed by their prefect uuid;
9//! state transitions are append-only records keyed by TID, so a collection
10//! scan of `io.prefect.flowRun.state` IS the chronological state history.
11//!
12//! all methods assume the caller holds the app-level write lock; this type
13//! adds no locking of its own. memory is arena-style: the repo owns an arena
14//! that lives as long as the process (orchestration history is append-only,
15//! so unbounded growth is the semantics, not a leak).
16
17const std = @import("std");
18const zat = @import("zat");
19
20pub const Record = struct {
21 key: []const u8, // "collection/rkey"
22 cbor_data: []const u8,
23};
24
25pub const Repo = struct {
26 arena_state: std.heap.ArenaAllocator,
27 child_allocator: std.mem.Allocator,
28 allocator: std.mem.Allocator, // arena handle
29 io: std.Io,
30 keypair: zat.Keypair,
31 did: []const u8,
32 tree: zat.mst.Mst,
33 blocks: std.StringHashMapUnmanaged([]const u8), // record cid raw -> cbor block
34 record_count: usize = 0,
35 commit_count: usize = 0,
36 last_rev_us: u64 = 0,
37 data_dir: []const u8,
38 latest_car: ?[]const u8 = null,
39 latest_car_owned: bool = false,
40
41 pub fn boot(child_allocator: std.mem.Allocator, io: std.Io, data_dir: []const u8) !*Repo {
42 // the arena must live at its final address before anything captures
43 // a handle to it (Mst stores the allocator), so create the struct
44 // first and build fields in place.
45 const self = try child_allocator.create(Repo);
46 errdefer child_allocator.destroy(self);
47 self.arena_state = std.heap.ArenaAllocator.init(child_allocator);
48 errdefer self.arena_state.deinit();
49 self.child_allocator = child_allocator;
50 const arena = self.arena_state.allocator();
51
52 std.Io.Dir.createDirPath(.cwd(), io, data_dir) catch {};
53
54 const keypair = try loadOrCreateKey(io, arena, data_dir);
55
56 self.allocator = arena;
57 self.io = io;
58 self.keypair = keypair;
59 self.did = try keypair.did(arena);
60 self.tree = zat.mst.Mst.init(arena);
61 self.blocks = .empty;
62 self.record_count = 0;
63 self.commit_count = 0;
64 self.last_rev_us = 0;
65 self.data_dir = try arena.dupe(u8, data_dir);
66 self.latest_car = null;
67 self.latest_car_owned = false;
68
69 try self.loadCarIfPresent();
70 return self;
71 }
72
73 pub fn deinit(self: *Repo, child_allocator: std.mem.Allocator) void {
74 if (self.latest_car) |old| {
75 if (self.latest_car_owned) child_allocator.free(old);
76 }
77 self.arena_state.deinit();
78 child_allocator.destroy(self);
79 }
80
81 fn loadOrCreateKey(io: std.Io, arena: std.mem.Allocator, data_dir: []const u8) !zat.Keypair {
82 var path_buf: [512]u8 = undefined;
83 const key_path = try std.fmt.bufPrint(&path_buf, "{s}/signing.key", .{data_dir});
84
85 var secret: [32]u8 = undefined;
86 if (std.Io.Dir.readFile(.cwd(), io, key_path, &secret)) |bytes| {
87 if (bytes.len == 32) return zat.Keypair.fromSecretKey(.p256, secret);
88 } else |_| {}
89
90 try io.randomSecure(&secret);
91 const keypair = try zat.Keypair.fromSecretKey(.p256, secret);
92 try std.Io.Dir.writeFile(.cwd(), io, .{ .sub_path = key_path, .data = &secret });
93 _ = arena;
94 return keypair;
95 }
96
97 fn carPath(self: *Repo, buf: []u8) ![]const u8 {
98 return std.fmt.bufPrint(buf, "{s}/repo.car", .{self.data_dir});
99 }
100
101 fn loadCarIfPresent(self: *Repo) !void {
102 var path_buf: [512]u8 = undefined;
103 const path = try self.carPath(&path_buf);
104
105 const car_bytes = std.Io.Dir.readFileAlloc(.cwd(), self.io, path, self.allocator, .limited(1 << 30)) catch return;
106
107 const loaded = try zat.loadCommitFromCAR(self.allocator, car_bytes);
108 self.tree = try zat.mst.Mst.loadFromBlocks(self.allocator, loaded.repo_car, loaded.commit.data_cid);
109
110 if (zat.Tid.parse(loaded.commit.rev)) |tid| self.last_rev_us = tid.timestamp();
111
112 // index record blocks by walking the tree
113 const Ctx = struct {
114 repo: *Repo,
115 repo_car: zat.car.Car,
116 fn onEntry(ctx: *anyopaque, entry: zat.mst.WalkEntry) anyerror!void {
117 const c: *@This() = @ptrCast(@alignCast(ctx));
118 const data = zat.car.findBlock(c.repo_car, entry.value.raw) orelse return error.MissingRecordBlock;
119 const cid_key = try c.repo.allocator.dupe(u8, entry.value.raw);
120 try c.repo.blocks.put(c.repo.allocator, cid_key, try c.repo.allocator.dupe(u8, data));
121 c.repo.record_count += 1;
122 }
123 };
124 var ctx = Ctx{ .repo = self, .repo_car = loaded.repo_car };
125 try self.tree.walk(.{ .ctx = &ctx, .entryFn = Ctx.onEntry });
126 self.latest_car = car_bytes;
127 }
128
129 /// put a record (dag-cbor bytes) at collection/rkey. caller must commit().
130 pub fn putRecord(self: *Repo, collection: []const u8, rkey: []const u8, cbor_data: []const u8) !zat.cbor.Cid {
131 const data = try self.allocator.dupe(u8, cbor_data);
132 const cid = try zat.cbor.Cid.forDagCbor(self.allocator, data);
133 const key = try std.fmt.allocPrint(self.allocator, "{s}/{s}", .{ collection, rkey });
134 try self.tree.put(key, cid);
135 try self.blocks.put(self.allocator, cid.raw, data);
136 self.record_count += 1;
137 return cid;
138 }
139
140 /// sign a commit over the current tree and persist the full repo as a
141 /// CAR file (tmp + atomic rename). returns the commit's rev.
142 ///
143 /// commit-transient memory (the serialized CAR, the block list, the
144 /// commit object) lives in a per-commit scratch arena — only the rev
145 /// survives into the repo arena, and `latest_car` is owned by the child
146 /// allocator and freed on replacement. without this, retaining one full
147 /// CAR per commit makes memory quadratic in commit count.
148 pub fn commit(self: *Repo) ![]const u8 {
149 const root = try self.tree.rootCid();
150
151 var scratch_state = std.heap.ArenaAllocator.init(self.child_allocator);
152 defer scratch_state.deinit();
153 const scratch = scratch_state.allocator();
154
155 const now_us: u64 = @intCast(@max(0, nowMicros(self.io)));
156 const rev_us = @max(now_us, self.last_rev_us + 1);
157 self.last_rev_us = rev_us;
158 var tid = zat.Tid.fromTimestamp(rev_us, 0);
159 const rev = try self.allocator.dupe(u8, tid.str());
160
161 const unsigned_entries = [_]zat.cbor.Value.MapEntry{
162 .{ .key = "did", .value = .{ .text = self.did } },
163 .{ .key = "rev", .value = .{ .text = rev } },
164 .{ .key = "data", .value = .{ .cid = root } },
165 .{ .key = "prev", .value = .null },
166 .{ .key = "version", .value = .{ .unsigned = 3 } },
167 };
168 const unsigned_bytes = try zat.cbor.encodeAlloc(scratch, .{ .map = &unsigned_entries });
169 const sig = try self.keypair.sign(unsigned_bytes);
170
171 const signed_entries = unsigned_entries ++ [_]zat.cbor.Value.MapEntry{
172 .{ .key = "sig", .value = .{ .bytes = &sig.bytes } },
173 };
174 const signed_bytes = try zat.cbor.encodeAlloc(scratch, .{ .map = &signed_entries });
175 const commit_cid = try zat.cbor.Cid.forDagCbor(scratch, signed_bytes);
176
177 var car_blocks: std.ArrayList(zat.car.Block) = .empty;
178 try car_blocks.append(scratch, .{ .cid_raw = commit_cid.raw, .data = signed_bytes });
179 try self.tree.collectBlocksInto(scratch, &car_blocks);
180 var it = self.blocks.iterator();
181 while (it.next()) |e| {
182 try car_blocks.append(scratch, .{ .cid_raw = e.key_ptr.*, .data = e.value_ptr.* });
183 }
184
185 const car_bytes = try zat.car.writeAlloc(self.child_allocator, .{
186 .roots = &.{commit_cid},
187 .blocks = car_blocks.items,
188 });
189 errdefer self.child_allocator.free(car_bytes);
190
191 var path_buf: [512]u8 = undefined;
192 const path = try self.carPath(&path_buf);
193 var tmp_buf: [512]u8 = undefined;
194 const tmp_path = try std.fmt.bufPrint(&tmp_buf, "{s}.tmp", .{path});
195 try std.Io.Dir.writeFile(.cwd(), self.io, .{ .sub_path = tmp_path, .data = car_bytes });
196 try std.Io.Dir.rename(.cwd(), tmp_path, .cwd(), path, self.io);
197
198 if (self.latest_car) |old| {
199 if (self.latest_car_owned) self.child_allocator.free(old);
200 }
201 self.latest_car = car_bytes;
202 self.latest_car_owned = true;
203 self.commit_count += 1;
204 return rev;
205 }
206
207 pub fn getRecord(self: *Repo, collection: []const u8, rkey: []const u8) ?[]const u8 {
208 var key_buf: [256]u8 = undefined;
209 const key = std.fmt.bufPrint(&key_buf, "{s}/{s}", .{ collection, rkey }) catch return null;
210 const cid = self.tree.get(key) orelse return null;
211 return self.blocks.get(cid.raw);
212 }
213
214 /// collect (key, cbor) pairs for every record in a collection, in rkey
215 /// order (the MST walk is ordered, and TID rkeys sort chronologically).
216 pub fn listRecords(self: *Repo, out_alloc: std.mem.Allocator, collection: []const u8) ![]Record {
217 const Ctx = struct {
218 repo: *Repo,
219 out_alloc: std.mem.Allocator,
220 prefix: []const u8,
221 list: std.ArrayList(Record) = .empty,
222 fn onEntry(ctx: *anyopaque, entry: zat.mst.WalkEntry) anyerror!void {
223 const c: *@This() = @ptrCast(@alignCast(ctx));
224 if (!std.mem.startsWith(u8, entry.key, c.prefix)) return;
225 const data = c.repo.blocks.get(entry.value.raw) orelse return;
226 try c.list.append(c.out_alloc, .{
227 .key = try c.out_alloc.dupe(u8, entry.key),
228 .cbor_data = data,
229 });
230 }
231 };
232 var prefix_buf: [256]u8 = undefined;
233 const prefix = try std.fmt.bufPrint(&prefix_buf, "{s}/", .{collection});
234 var ctx = Ctx{ .repo = self, .out_alloc = out_alloc, .prefix = prefix };
235 try self.tree.walk(.{ .ctx = &ctx, .entryFn = Ctx.onEntry });
236 return ctx.list.items;
237 }
238
239 fn nowMicros(io: std.Io) i64 {
240 const ns = std.Io.Timestamp.now(io, .real).nanoseconds;
241 return @intCast(@divFloor(ns, std.time.ns_per_us));
242 }
243};
244
245test "repo: put, commit, reload from CAR, verify" {
246 var threaded = std.Io.Threaded.init(std.testing.allocator, .{});
247 defer threaded.deinit();
248 const io = threaded.io();
249
250 const data_dir = ".zig-cache/test-repo-data";
251 std.Io.Dir.deleteTree(.cwd(), io, data_dir) catch {};
252 defer std.Io.Dir.deleteTree(.cwd(), io, data_dir) catch {};
253
254 const did: []const u8 = blk: {
255 var repo = try Repo.boot(std.testing.allocator, io, data_dir);
256 defer repo.deinit(std.testing.allocator);
257
258 const record = [_]zat.cbor.Value.MapEntry{
259 .{ .key = "$type", .value = .{ .text = "io.prefect.flow" } },
260 .{ .key = "name", .value = .{ .text = "hello" } },
261 };
262 const cbor_data = try zat.cbor.encodeAlloc(repo.allocator, .{ .map = &record });
263 _ = try repo.putRecord("io.prefect.flow", "a-b-c", cbor_data);
264 _ = try repo.commit();
265
266 // verify the persisted CAR against the signing key
267 const pubkey = try repo.keypair.publicKey();
268 const result = try zat.verifyCommitCar(repo.allocator, repo.latest_car.?, .{
269 .key_type = .p256,
270 .raw = &pubkey,
271 }, .{ .expected_did = repo.did });
272 try std.testing.expectEqual(@as(usize, 1), result.record_count);
273 break :blk try std.testing.allocator.dupe(u8, repo.did);
274 };
275 defer std.testing.allocator.free(did);
276
277 // reboot from disk: same identity, record still there
278 var repo2 = try Repo.boot(std.testing.allocator, io, data_dir);
279 defer repo2.deinit(std.testing.allocator);
280 try std.testing.expectEqualStrings(did, repo2.did);
281 try std.testing.expect(repo2.getRecord("io.prefect.flow", "a-b-c") != null);
282 try std.testing.expectEqual(@as(usize, 1), repo2.record_count);
283}