zzstoatzz.io
search for standard sites
pub-search.waow.tech
search
zig
blog
atproto
12 kB
279 lines
1//! Builder mode: offline snapshot build + verified R2 publish.
2//!
3//! Invariant #1 of docs/scaling-plan.md: background data movement never
4//! touches the serving box. This runs as a separate process (BUILDER_MODE=1
5//! on an ephemeral machine or a cron), builds a fresh replica from turso into
6//! a scratch path, refuses to publish unless sentinels pass, and uploads
7//! artifact + manifest to R2 — latest pointer last, so a reader can never see
8//! a pointer to an incomplete artifact (typeahead's channel discipline).
9//!
10//! Channels: staging by default; prod requires BUILDER_ALLOW_PROD=1 so a
11//! scratch run can never overwrite the production pointer.
12//!
13//! Env: TURSO_URL, TURSO_TOKEN, INDEX_R2_{ENDPOINT,BUCKET,ACCESS_KEY_ID,
14//! SECRET_ACCESS_KEY}; optional BUILD_DIR (default /tmp/leaflet-build),
15//! BUILDER_CHANNEL (staging|prod), BUILDER_VERSION (git sha), SKIP_UPLOAD=1
16//! (build + verify only).
17
18const std = @import("std");
19const Io = std.Io;
20const Allocator = std.mem.Allocator;
21const logfire = @import("logfire");
22const zqlite = @import("zqlite");
23const db = @import("db.zig");
24const LocalDb = @import("db/LocalDb.zig");
25const sync = @import("db/sync.zig");
26const r2 = @import("r2.zig");
27
28pub const MANIFEST_VERSION = 1;
29
30const Channel = enum { staging, prod };
31const DEFAULT_TIMEOUT_SECS: u64 = 45 * 60;
32
33fn getenv(name: [*:0]const u8) ?[]const u8 {
34 return if (std.c.getenv(name)) |v| std.mem.span(v) else null;
35}
36
37fn parseTimeoutSeconds(raw: ?[]const u8) u64 {
38 const value = std.fmt.parseInt(u64, raw orelse return DEFAULT_TIMEOUT_SECS, 10) catch
39 return DEFAULT_TIMEOUT_SECS;
40 return if (value == 0) DEFAULT_TIMEOUT_SECS else value;
41}
42
43fn timeoutWatchdog(io: Io, timeout_s: u64) void {
44 io.sleep(Io.Duration.fromSeconds(@intCast(timeout_s)), .awake) catch return;
45 logfire.err("builder: exceeded {d}s wall-clock timeout; terminating", .{timeout_s});
46 std.process.exit(1);
47}
48
49pub fn run(allocator: Allocator, io: Io) !void {
50 const started_s: i64 = @intCast(@divFloor(Io.Timestamp.now(io, .real).nanoseconds, std.time.ns_per_s));
51 const timeout_s = parseTimeoutSeconds(getenv("BUILDER_TIMEOUT_SECS"));
52 const watchdog = try std.Thread.spawn(.{}, timeoutWatchdog, .{ io, timeout_s });
53 watchdog.detach();
54 logfire.info("builder: watchdog armed ({d}s)", .{timeout_s});
55
56 const channel: Channel = blk: {
57 const raw = getenv("BUILDER_CHANNEL") orelse "staging";
58 if (std.mem.eql(u8, raw, "prod")) {
59 if (getenv("BUILDER_ALLOW_PROD") == null) {
60 logfire.err("builder: BUILDER_CHANNEL=prod requires BUILDER_ALLOW_PROD=1", .{});
61 return error.ProdNotArmed;
62 }
63 break :blk .prod;
64 }
65 break :blk .staging;
66 };
67
68 const build_dir = getenv("BUILD_DIR") orelse "/tmp/leaflet-build";
69 _ = std.c.mkdir(@ptrCast((std.fmt.allocPrintSentinel(allocator, "{s}", .{build_dir}, 0) catch return error.OutOfMemory).ptr), 0o755);
70
71 var id_buf: [48]u8 = undefined;
72 const nanos: u64 = @intCast(@mod(Io.Timestamp.now(io, .real).nanoseconds, 65536));
73 const build_id = try std.fmt.bufPrint(&id_buf, "b{d}-{x:0>4}", .{ started_s, nanos });
74
75 const db_path = try std.fmt.allocPrint(allocator, "{s}/replica.db", .{build_dir});
76 const manifest_path = try std.fmt.allocPrint(allocator, "{s}/manifest.json", .{build_dir});
77
78 // fresh scratch file every run — stale partial builds must not leak in
79 for ([_][]const u8{ "", "-wal", "-shm" }) |suffix| {
80 const path = try std.fmt.allocPrintSentinel(allocator, "{s}{s}", .{ db_path, suffix }, 0);
81 _ = std.c.unlink(path.ptr);
82 }
83
84 logfire.info("builder: starting build {s} (channel={s}) -> {s}", .{ build_id, @tagName(channel), db_path });
85
86 try db.initTurso(io);
87 const turso = db.getClient() orelse return error.TursoNotInitialized;
88
89 // watermark BEFORE paging: docs indexed mid-build land in the next
90 // snapshot; the overlay/promote side uses this as its freshness cutoff
91 const watermark = blk: {
92 var result = try turso.query("SELECT COALESCE(MAX(indexed_at), '') FROM documents", &.{});
93 defer result.deinit();
94 if (result.first()) |row| {
95 break :blk try allocator.dupe(u8, row.text(0));
96 }
97 break :blk try allocator.dupe(u8, "");
98 };
99
100 var local = LocalDb.init(allocator, io);
101 try local.openAt(db_path);
102
103 const counts = try sync.buildSnapshot(turso, &local, watermark);
104 logfire.info("builder: built {d} docs, {d} pubs, {d} tags, {d} popular, {d} recommends, {d} subscriptions", .{
105 counts.documents, counts.publications, counts.tags, counts.popular, counts.recommends, counts.subscriptions,
106 });
107
108 // ------------------------------------------------------------------
109 // verification gates: refuse to publish a snapshot that fails any
110 // ------------------------------------------------------------------
111 const conn = local.getConn() orelse return error.LocalNotOpen;
112
113 // sync_meta: a server adopting this snapshot (or offline catchup) resumes
114 // incrementally from the build start, not from epoch
115 {
116 var ts_buf: [20]u8 = undefined;
117 const ts_str = std.fmt.bufPrint(&ts_buf, "{d}", .{started_s}) catch "0";
118 try conn.exec("INSERT OR REPLACE INTO sync_meta (key, value) VALUES ('last_sync', ?)", .{ts_str});
119 }
120
121 // gate 1: row count vs turso, both sides pinned to the watermark + the
122 // same policy filters. The set is immutable below the watermark, so any
123 // mismatch beyond deletes-during-build is a real bug; keep a small
124 // tolerance for those deletes only.
125 {
126 logfire.info("builder: phase=verify_remote", .{});
127 var result = try turso.query(sync.BUILD_DOC_COUNT_SQL, &.{watermark});
128 defer result.deinit();
129 const expected: usize = if (result.first()) |row| @intCast(row.int(0)) else 0;
130 const tolerance = @max(expected / 1000, 10); // 0.1%, floor 10 — deletes mid-build only
131 const diff = if (expected > counts.documents) expected - counts.documents else counts.documents - expected;
132 if (expected == 0 or diff > tolerance) {
133 logfire.err("builder: GATE FAIL doc count: built {d}, turso expects {d} (±{d})", .{ counts.documents, expected, tolerance });
134 return error.DocCountGate;
135 }
136 }
137
138 // gate 2: FTS answers a query
139 {
140 const row = try conn.row("SELECT COUNT(*) FROM documents_fts WHERE documents_fts MATCH 'the'", .{});
141 if (row) |r| {
142 defer r.deinit();
143 if (r.int(0) == 0) {
144 logfire.err("builder: GATE FAIL fts sentinel returned 0 rows", .{});
145 return error.FtsSentinelGate;
146 }
147 } else return error.FtsSentinelGate;
148 }
149
150 // leaving WAL mode requires being the ONLY connection — close LocalDb's
151 // write conn + read pool first, then compact on a fresh solo connection
152 local.deinit();
153
154 {
155 logfire.info("builder: phase=compact", .{});
156 const path_z = try std.fmt.allocPrintSentinel(allocator, "{s}", .{db_path}, 0);
157 const solo = try zqlite.open(path_z.ptr, zqlite.OpenFlags.ReadWrite);
158 defer solo.close();
159
160 // compact to a single file: checkpoint WAL, drop to DELETE journal,
161 // VACUUM, then back to WAL (serving re-asserts WAL at open anyway)
162 try solo.exec("PRAGMA wal_checkpoint(TRUNCATE)", .{});
163 try solo.exec("PRAGMA journal_mode=DELETE", .{});
164 try solo.exec("VACUUM", .{});
165 try solo.exec("PRAGMA journal_mode=WAL", .{});
166 try solo.exec("PRAGMA wal_checkpoint(TRUNCATE)", .{});
167
168 // gate 3: page-structure integrity on the final file
169 const row = try solo.row("PRAGMA quick_check", .{});
170 if (row) |r| {
171 defer r.deinit();
172 if (!std.mem.eql(u8, r.text(0), "ok")) {
173 logfire.err("builder: GATE FAIL quick_check: {s}", .{r.text(0)});
174 return error.QuickCheckGate;
175 }
176 } else return error.QuickCheckGate;
177 }
178
179 // hash + size of the exact bytes that will be served
180 var sha_hex: [64]u8 = undefined;
181 const byte_size = try sha256File(allocator, io, db_path, &sha_hex);
182 logfire.info("builder: replica.db {d} bytes sha256={s}", .{ byte_size, sha_hex });
183
184 const prefix = switch (channel) {
185 .prod => "",
186 .staging => "staging/",
187 };
188 const snapshot_key = try std.fmt.allocPrint(allocator, "{s}builds/{s}/replica.db", .{ prefix, build_id });
189 const manifest_key = try std.fmt.allocPrint(allocator, "{s}builds/{s}/manifest.json", .{ prefix, build_id });
190 const latest_key: []const u8 = switch (channel) {
191 .prod => "latest.json",
192 .staging => "latest.staging.json",
193 };
194
195 const manifest = try std.fmt.allocPrint(allocator,
196 \\{{
197 \\ "manifest_version": {d},
198 \\ "schema_version": {d},
199 \\ "build_id": "{s}",
200 \\ "channel": "{s}",
201 \\ "snapshot_key": "{s}",
202 \\ "byte_size": {d},
203 \\ "sha256": "{s}",
204 \\ "source_watermark": "{s}",
205 \\ "doc_count": {d},
206 \\ "pub_count": {d},
207 \\ "tag_count": {d},
208 \\ "rec_count": {d},
209 \\ "created_at": {d},
210 \\ "builder_version": "{s}"
211 \\}}
212 \\
213 , .{
214 MANIFEST_VERSION, LocalDb.SCHEMA_VERSION, build_id, @tagName(channel),
215 snapshot_key, byte_size, &sha_hex, watermark,
216 counts.documents, counts.publications, counts.tags, counts.recommends,
217 started_s, getenv("BUILDER_VERSION") orelse "dev",
218 });
219
220 try writeFile(io, manifest_path, manifest);
221
222 if (getenv("SKIP_UPLOAD") != null) {
223 logfire.info("builder: SKIP_UPLOAD set — artifacts left at {s} ({s})", .{ build_dir, build_id });
224 std.debug.print("builder: done (no upload). replica={s} manifest={s}\n", .{ db_path, manifest_path });
225 return;
226 }
227
228 const cfg = try r2.configure(allocator, io);
229 logfire.info("builder: phase=upload", .{});
230 try r2.upload(allocator, io, cfg, db_path, snapshot_key);
231 try r2.upload(allocator, io, cfg, manifest_path, manifest_key);
232 // latest pointer LAST — readers can never see a pointer to a partial build
233 try r2.upload(allocator, io, cfg, manifest_path, latest_key);
234
235 const took_s: i64 = @intCast(@divFloor(Io.Timestamp.now(io, .real).nanoseconds, std.time.ns_per_s) - started_s);
236 logfire.info("builder: published {s} to {s} channel ({d} docs, {d}s)", .{ build_id, @tagName(channel), counts.documents, took_s });
237}
238
239test "builder timeout parser uses a safe nonzero default" {
240 try std.testing.expectEqual(DEFAULT_TIMEOUT_SECS, parseTimeoutSeconds(null));
241 try std.testing.expectEqual(DEFAULT_TIMEOUT_SECS, parseTimeoutSeconds(""));
242 try std.testing.expectEqual(DEFAULT_TIMEOUT_SECS, parseTimeoutSeconds("nope"));
243 try std.testing.expectEqual(DEFAULT_TIMEOUT_SECS, parseTimeoutSeconds("0"));
244 try std.testing.expectEqual(@as(u64, 90), parseTimeoutSeconds("90"));
245}
246
247fn writeFile(io: Io, path: []const u8, content: []const u8) !void {
248 const file = try Io.Dir.createFileAbsolute(io, path, .{ .truncate = true });
249 defer file.close(io);
250 var wbuf: [4096]u8 = undefined;
251 var fw = Io.File.Writer.init(file, io, &wbuf);
252 try fw.interface.writeAll(content);
253 try fw.interface.flush();
254}
255
256/// Streaming sha256 of a file; returns byte count, writes lowercase hex into out.
257fn sha256File(allocator: Allocator, io: Io, path: []const u8, out_hex: *[64]u8) !u64 {
258 _ = allocator;
259 const file = try Io.Dir.openFileAbsolute(io, path, .{});
260 defer file.close(io);
261
262 var hasher = std.crypto.hash.sha2.Sha256.init(.{});
263 var total: u64 = 0;
264 var buf: [256 * 1024]u8 = undefined;
265 while (true) {
266 const n = file.readStreaming(io, &.{&buf}) catch |err| switch (err) {
267 error.EndOfStream => break,
268 else => return err,
269 };
270 if (n == 0) break;
271 hasher.update(buf[0..n]);
272 total += n;
273 }
274 var digest: [32]u8 = undefined;
275 hasher.final(&digest);
276 const hex = std.fmt.bytesToHex(digest, .lower);
277 @memcpy(out_hex, &hex);
278 return total;
279}