zzstoatzz.io
search for standard sites
pub-search.waow.tech
search
zig
blog
atproto
chore(sync): delete in-place Turso→replica sync (footgun removal)
The serving box's replica is refreshed only by verified snapshot adoption at
boot (SYNC_DISABLE=1 has held in prod since the 2026-06-10 cutover). The
in-place incrementalSync/fullSync path was still compiled and wired into a
background thread + watchdog, neutralized only by an env flag — a footgun: drop
the flag and background data movement touches the serving box again (the
06-10 outage). Delete it so the serving artifact is immutable by construction.
- sync.zig: drop fullSync + incrementalSync (and the dead BATCH_SIZE); keep
buildSnapshot + insert helpers (the offline builder, used by builder.zig)
- db.zig: startSync now just marks the replica ready; remove syncLoop,
syncWatchdog, the heartbeat atomics, and the second Turso client
- off-box freshness (scripts/offline-replica-catchup) is a standalone python
reimplementation — unaffected
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
author
zzstoatzz
co-author
Claude Opus 4.8
date
(Jun 26, 2026, 3:38 PM -0500)
commit
94de2382
94de23823a0451dcf5ffc2c96948c95aad3a5b07
parent
8d9f1a54
8d9f1a545a7eb71608ae545d771207bbf3372871
+13
-520
+8
-99
backend/src/db.zig
+8
-99
backend/src/db.zig
···
4
4
5
5
const schema = @import("db/schema.zig");
6
6
const result = @import("db/result.zig");
7
-
const sync = @import("db/sync.zig");
8
7
9
8
// re-exports
10
9
pub const Client = @import("db/Client.zig");
···
16
15
// global state
17
16
const allocator = std.heap.smp_allocator;
18
17
var client: ?Client = null;
19
-
var sync_client: ?Client = null;
20
18
var local_db: ?LocalDb = null;
21
19
22
-
// sync liveness heartbeat (unix seconds). set at the top of each sync loop
23
-
// iteration; watchdog thread aborts the process if it goes stale so fly
24
-
// restarts us. zig 0.16 http.Client has no timeout, so a wedged HTTP call
25
-
// would otherwise silently kill sync forever.
26
-
var sync_heartbeat_s = std.atomic.Value(i64).init(0);
27
-
var sync_interval_secs_atomic = std.atomic.Value(u64).init(300);
28
-
29
20
/// Initialize Turso client only (fast, call synchronously at startup).
30
21
/// Schema migrations run separately via initSchema() in the background thread
31
22
/// so a slow/unreachable turso doesn't block the accept loop.
32
23
pub fn initTurso(io: Io) !void {
33
24
client = try Client.init(allocator, io);
34
-
sync_client = try Client.init(allocator, io);
35
25
}
36
26
37
27
/// Run schema migrations via zug. Call from background thread.
···
91
81
return null;
92
82
}
93
83
94
-
/// Start background sync thread (call from main after db.init)
84
+
/// Mark the local replica ready to serve. In-place Turso→replica sync was
85
+
/// deleted (2026-06-26): the serving box's replica is refreshed only by
86
+
/// verified snapshot adoption at boot, never by background data movement
87
+
/// (the 2026-06-10 invariant). Kept as a named call so main.zig's startup
88
+
/// sequence reads unchanged.
95
89
pub fn startSync(io: Io) void {
96
-
// hard rule (2026-06-10): background data movement must never interrupt
97
-
// serving. until the snapshot-swap replica lands, in-place sync can be
98
-
// disabled entirely — stale-but-fast beats fresh-but-down.
99
-
if (std.c.getenv("SYNC_DISABLE") != null) {
100
-
std.debug.print("sync: DISABLED via SYNC_DISABLE — serving existing replica as-is\n", .{});
101
-
if (getLocalDbRaw()) |l| l.setReady(true);
102
-
return;
103
-
}
104
-
const c = if (sync_client) |*sc| sc else {
105
-
std.debug.print("sync: no sync client, skipping\n", .{});
106
-
return;
107
-
};
108
-
const local = getLocalDbRaw() orelse {
109
-
std.debug.print("sync: no local db, skipping\n", .{});
110
-
return;
111
-
};
112
-
113
-
const thread = std.Thread.spawn(.{}, syncLoop, .{ c, local, io }) catch |err| {
114
-
std.debug.print("sync: failed to start thread: {}\n", .{err});
115
-
return;
116
-
};
117
-
thread.detach();
118
-
std.debug.print("sync: background thread started\n", .{});
119
-
120
-
const watchdog = std.Thread.spawn(.{}, syncWatchdog, .{io}) catch |err| {
121
-
std.debug.print("sync: failed to start watchdog: {}\n", .{err});
122
-
return;
123
-
};
124
-
watchdog.detach();
125
-
std.debug.print("sync: watchdog thread started\n", .{});
126
-
}
127
-
128
-
fn nowSeconds(io: Io) i64 {
129
-
return @intCast(@divFloor(Io.Timestamp.now(io, .real).nanoseconds, std.time.ns_per_s));
130
-
}
131
-
132
-
fn syncLoop(turso: *Client, local: *LocalDb, io: Io) void {
133
-
// get sync interval from env (default 5 minutes)
134
-
const interval_secs: u64 = blk: {
135
-
const env_val = if (std.c.getenv("SYNC_INTERVAL_SECS")) |p| std.mem.span(p) else "300";
136
-
break :blk std.fmt.parseInt(u64, env_val, 10) catch 300;
137
-
};
138
-
sync_interval_secs_atomic.store(interval_secs, .release);
139
-
140
-
// incremental sync on startup — gets new docs + cleans tombstoned deletions
141
-
// (falls back to full sync automatically if no last_sync exists, i.e., first boot)
142
-
sync_heartbeat_s.store(nowSeconds(io), .release);
143
-
sync.incrementalSync(turso, local) catch |err| {
144
-
std.debug.print("sync: initial sync failed: {}\n", .{err});
145
-
};
146
-
147
-
std.debug.print("sync: incremental sync every {d} seconds\n", .{interval_secs});
148
-
149
-
// periodic incremental sync
150
-
while (true) {
151
-
io.sleep(Io.Duration.fromSeconds(@intCast(interval_secs)), .awake) catch {};
152
-
// update heartbeat before each attempt; a hung HTTP call prevents the
153
-
// next iteration's update, which the watchdog will catch.
154
-
sync_heartbeat_s.store(nowSeconds(io), .release);
155
-
sync.incrementalSync(turso, local) catch |err| {
156
-
std.debug.print("sync: incremental sync failed: {}\n", .{err});
157
-
};
158
-
}
159
-
}
160
-
161
-
/// Watchdog: aborts the process if the sync loop hasn't updated its heartbeat
162
-
/// within ~3x the configured sync interval. fly auto-restarts the machine.
163
-
fn syncWatchdog(io: Io) void {
164
-
// wait for syncLoop to record its initial interval + heartbeat
165
-
io.sleep(Io.Duration.fromSeconds(30), .awake) catch {};
166
-
167
-
while (true) {
168
-
io.sleep(Io.Duration.fromSeconds(60), .awake) catch {};
169
-
170
-
const interval = sync_interval_secs_atomic.load(.acquire);
171
-
const heartbeat = sync_heartbeat_s.load(.acquire);
172
-
if (heartbeat == 0) continue; // not started yet
173
-
174
-
const now_s = nowSeconds(io);
175
-
const staleness_s: i64 = now_s - heartbeat;
176
-
const max_staleness_s: i64 = @intCast(interval * 3);
177
-
178
-
if (staleness_s > max_staleness_s) {
179
-
logfire.err("sync watchdog: heartbeat stale for {d}s (max {d}s) — aborting for restart", .{ staleness_s, max_staleness_s });
180
-
std.debug.print("sync watchdog: heartbeat stale for {d}s (max {d}s) — aborting for restart\n", .{ staleness_s, max_staleness_s });
181
-
std.process.exit(1);
182
-
}
183
-
}
90
+
_ = io;
91
+
if (getLocalDbRaw()) |l| l.setReady(true);
92
+
std.debug.print("sync: in-place sync removed — serving adopted snapshot as-is\n", .{});
184
93
}
+5
-421
backend/src/db/sync.zig
+5
-421
backend/src/db/sync.zig
···
1
-
//! Sync from Turso to local SQLite
2
-
//! Full sync on startup, incremental sync periodically
1
+
//! Offline snapshot builder: reads Turso and populates a FRESH local SQLite
2
+
//! file off-box (builder.zig). In-place serving-box sync was deleted — the
3
+
//! replica is refreshed only by verified snapshot adoption (see
4
+
//! docs/snapshot-pipeline.md). Background data movement never touches the
5
+
//! serving box (2026-06-10 invariant).
3
6
4
7
const std = @import("std");
5
8
const Io = std.Io;
···
11
14
const policy = @import("../policy.zig");
12
15
const pubkey = @import("../server/pubkey.zig");
13
16
14
-
const BATCH_SIZE = 500;
15
-
16
-
/// Full sync: fetch all data from Turso and populate local SQLite.
17
-
/// Uses INSERT OR REPLACE so existing data stays queryable during sync.
18
-
/// Only marks not-ready on first-ever sync (empty DB).
19
-
pub fn fullSync(turso: *Client, local: *LocalDb) !void {
20
-
std.debug.print("sync: starting full sync...\n", .{});
21
-
22
-
const conn = local.getConn() orelse return error.LocalNotOpen;
23
-
24
-
// check if we have existing data — if so, keep serving while syncing
25
-
const has_data = blk: {
26
-
local.lock();
27
-
defer local.unlock();
28
-
const row = conn.row("SELECT COUNT(*) FROM documents", .{}) catch break :blk false;
29
-
if (row) |r| {
30
-
defer r.deinit();
31
-
break :blk r.int(0) > 0;
32
-
}
33
-
break :blk false;
34
-
};
35
-
36
-
if (!has_data) {
37
-
// first-ever sync: nothing to serve, mark not-ready
38
-
local.setReady(false);
39
-
40
-
// clear tables for clean initial sync
41
-
local.lock();
42
-
defer local.unlock();
43
-
conn.exec("DELETE FROM documents_fts", .{}) catch {};
44
-
conn.exec("DELETE FROM documents", .{}) catch {};
45
-
conn.exec("DELETE FROM publications_fts", .{}) catch {};
46
-
conn.exec("DELETE FROM publications", .{}) catch {};
47
-
conn.exec("DELETE FROM document_tags", .{}) catch {};
48
-
} else {
49
-
// re-sync: keep serving existing data while we refresh in-place
50
-
// INSERT OR REPLACE will update rows; stale data is acceptable
51
-
local.setReady(true);
52
-
std.debug.print("sync: local has data, keeping ready during re-sync\n", .{});
53
-
}
54
-
55
-
// create temp table to track synced URIs (for stale-doc cleanup)
56
-
{
57
-
local.lock();
58
-
defer local.unlock();
59
-
conn.exec("DROP TABLE IF EXISTS _synced_uris", .{}) catch {};
60
-
conn.exec("CREATE TEMP TABLE _synced_uris (uri TEXT PRIMARY KEY)", .{}) catch {};
61
-
}
62
-
63
-
// sync documents in batches — fetch from Turso unlocked, write to local with brief lock
64
-
var doc_count: usize = 0;
65
-
var offset: usize = 0;
66
-
while (true) {
67
-
var offset_buf: [16]u8 = undefined;
68
-
const offset_str = std.fmt.bufPrint(&offset_buf, "{d}", .{offset}) catch break;
69
-
70
-
// fetch from Turso (no lock held — search can use local DB)
71
-
var result = turso.query(
72
-
\\SELECT uri, did, rkey, title, content, created_at, publication_uri,
73
-
\\ platform, source_collection, path, base_path, has_publication, indexed_at, embedded_at,
74
-
\\ COALESCE(cover_image, '') as cover_image, COALESCE(is_bridgyfed, 0) as is_bridgyfed,
75
-
\\ COALESCE(url_dead, 0) as url_dead
76
-
\\FROM documents
77
-
\\ORDER BY uri
78
-
\\LIMIT 500 OFFSET ?
79
-
, &.{offset_str}) catch |err| {
80
-
logfire.warn("sync: turso query failed: {}", .{err});
81
-
break;
82
-
};
83
-
defer result.deinit();
84
-
85
-
if (result.rows.len == 0) break;
86
-
87
-
// write batch to local (brief lock)
88
-
{
89
-
local.lock();
90
-
defer local.unlock();
91
-
conn.exec("BEGIN", .{}) catch {};
92
-
for (result.rows) |row| {
93
-
insertDocumentLocal(conn, row) catch |err| {
94
-
logfire.warn("sync: insert doc failed: {}", .{err});
95
-
};
96
-
conn.exec("INSERT OR IGNORE INTO _synced_uris (uri) VALUES (?)", .{row.text(0)}) catch {};
97
-
doc_count += 1;
98
-
}
99
-
conn.exec("COMMIT", .{}) catch {};
100
-
}
101
-
102
-
offset += result.rows.len;
103
-
if (offset % 1000 == 0) {
104
-
std.debug.print("sync: synced {d} documents...\n", .{offset});
105
-
}
106
-
}
107
-
108
-
// sync publications (fetch unlocked, write with brief lock)
109
-
var pub_count: usize = 0;
110
-
{
111
-
var pub_result = turso.query(
112
-
"SELECT uri, did, rkey, name, description, base_path, platform, indexed_at FROM publications",
113
-
&.{},
114
-
) catch |err| {
115
-
logfire.warn("sync: turso publications query failed: {}", .{err});
116
-
return;
117
-
};
118
-
defer pub_result.deinit();
119
-
120
-
local.lock();
121
-
defer local.unlock();
122
-
conn.exec("BEGIN", .{}) catch {};
123
-
for (pub_result.rows) |row| {
124
-
insertPublicationLocal(conn, row) catch |err| {
125
-
logfire.warn("sync: insert pub failed: {}", .{err});
126
-
};
127
-
pub_count += 1;
128
-
}
129
-
conn.exec("COMMIT", .{}) catch {};
130
-
}
131
-
132
-
// sync tags
133
-
var tag_count: usize = 0;
134
-
{
135
-
var tags_result = turso.query(
136
-
"SELECT document_uri, tag FROM document_tags",
137
-
&.{},
138
-
) catch |err| {
139
-
logfire.warn("sync: turso tags query failed: {}", .{err});
140
-
return;
141
-
};
142
-
defer tags_result.deinit();
143
-
144
-
local.lock();
145
-
defer local.unlock();
146
-
conn.exec("BEGIN", .{}) catch {};
147
-
for (tags_result.rows) |row| {
148
-
conn.exec(
149
-
"INSERT OR IGNORE INTO document_tags (document_uri, tag) VALUES (?, ?)",
150
-
.{ row.text(0), row.text(1) },
151
-
) catch {};
152
-
tag_count += 1;
153
-
}
154
-
conn.exec("COMMIT", .{}) catch {};
155
-
}
156
-
157
-
// sync popular searches
158
-
var popular_count: usize = 0;
159
-
{
160
-
var popular_result = turso.query(
161
-
"SELECT query, count FROM popular_searches",
162
-
&.{},
163
-
) catch |err| {
164
-
logfire.warn("sync: turso popular_searches query failed: {}", .{err});
165
-
return;
166
-
};
167
-
defer popular_result.deinit();
168
-
169
-
local.lock();
170
-
defer local.unlock();
171
-
conn.exec("DELETE FROM popular_searches", .{}) catch {};
172
-
conn.exec("BEGIN", .{}) catch {};
173
-
for (popular_result.rows) |row| {
174
-
conn.exec(
175
-
"INSERT OR REPLACE INTO popular_searches (query, count) VALUES (?, ?)",
176
-
.{ row.text(0), row.text(1) },
177
-
) catch {};
178
-
popular_count += 1;
179
-
}
180
-
conn.exec("COMMIT", .{}) catch {};
181
-
}
182
-
183
-
// clean up stale docs that were deleted from Turso (brief lock)
184
-
{
185
-
local.lock();
186
-
defer local.unlock();
187
-
conn.exec("DELETE FROM documents_fts WHERE uri NOT IN (SELECT uri FROM _synced_uris)", .{}) catch {};
188
-
conn.exec("DELETE FROM documents WHERE uri NOT IN (SELECT uri FROM _synced_uris)", .{}) catch {};
189
-
conn.exec("DELETE FROM document_tags WHERE document_uri NOT IN (SELECT uri FROM _synced_uris)", .{}) catch {};
190
-
conn.exec("DROP TABLE IF EXISTS _synced_uris", .{}) catch {};
191
-
}
192
-
193
-
// record sync time (brief lock)
194
-
{
195
-
local.lock();
196
-
defer local.unlock();
197
-
var ts_buf: [20]u8 = undefined;
198
-
const now_s: i64 = @intCast(@divFloor(Io.Timestamp.now(turso.io, .real).nanoseconds, std.time.ns_per_s));
199
-
const ts_str = std.fmt.bufPrint(&ts_buf, "{d}", .{now_s}) catch "0";
200
-
conn.exec(
201
-
"INSERT OR REPLACE INTO sync_meta (key, value) VALUES ('last_sync', ?)",
202
-
.{ts_str},
203
-
) catch {};
204
-
}
205
-
206
-
// checkpoint WAL to prevent unbounded growth
207
-
{
208
-
local.lock();
209
-
defer local.unlock();
210
-
conn.exec("PRAGMA wal_checkpoint(PASSIVE)", .{}) catch |err| {
211
-
logfire.warn("sync: wal checkpoint failed: {}", .{err});
212
-
};
213
-
}
214
-
215
-
local.setReady(true);
216
-
std.debug.print("sync: full sync complete - {d} docs, {d} pubs, {d} tags, {d} popular\n", .{ doc_count, pub_count, tag_count, popular_count });
217
-
}
218
-
219
17
pub const BuildCounts = struct {
220
18
documents: usize = 0,
221
19
publications: usize = 0,
···
422
220
}
423
221
424
222
return counts;
425
-
}
426
-
427
-
/// Incremental sync: fetch documents created since last sync
428
-
pub fn incrementalSync(turso: *Client, local: *LocalDb) !void {
429
-
const sync_span = logfire.span("sync.incremental", .{});
430
-
defer sync_span.end();
431
-
432
-
const conn = local.getConn() orelse {
433
-
sync_span.recordError(error.LocalNotOpen);
434
-
return error.LocalNotOpen;
435
-
};
436
-
437
-
// get last sync time
438
-
local.lock();
439
-
const last_sync_ts = blk: {
440
-
const row = conn.row(
441
-
"SELECT value FROM sync_meta WHERE key = 'last_sync'",
442
-
.{},
443
-
) catch {
444
-
local.unlock();
445
-
break :blk @as(i64, 0);
446
-
};
447
-
if (row) |r| {
448
-
defer r.deinit();
449
-
const val = r.text(0);
450
-
local.unlock();
451
-
// empty string (NULL) or invalid -> 0
452
-
break :blk if (val.len == 0) 0 else std.fmt.parseInt(i64, val, 10) catch 0;
453
-
}
454
-
local.unlock();
455
-
break :blk @as(i64, 0);
456
-
};
457
-
458
-
if (last_sync_ts == 0) {
459
-
// no previous sync, do full sync
460
-
std.debug.print("sync: no last_sync found, doing full sync\n", .{});
461
-
return fullSync(turso, local);
462
-
}
463
-
464
-
// local has data from a previous sync — mark ready immediately
465
-
local.setReady(true);
466
-
467
-
// convert timestamp to ISO date for query
468
-
// rough estimate: subtract 5 minutes buffer to catch any stragglers
469
-
const since_ts = last_sync_ts - 300;
470
-
const epoch_secs: u64 = @intCast(since_ts);
471
-
const epoch = std.time.epoch.EpochSeconds{ .secs = epoch_secs };
472
-
const day_secs = epoch.getDaySeconds();
473
-
const year_day = epoch.getEpochDay().calculateYearDay();
474
-
const month_day = year_day.calculateMonthDay();
475
-
476
-
var since_buf: [24]u8 = undefined;
477
-
const since_str = std.fmt.bufPrint(&since_buf, "{d:0>4}-{d:0>2}-{d:0>2}T{d:0>2}:{d:0>2}:{d:0>2}", .{
478
-
year_day.year,
479
-
@intFromEnum(month_day.month),
480
-
month_day.day_index + 1,
481
-
day_secs.getHoursIntoDay(),
482
-
day_secs.getMinutesIntoHour(),
483
-
day_secs.getSecondsIntoMinute(),
484
-
}) catch {
485
-
logfire.warn("sync: failed to format since date", .{});
486
-
return;
487
-
};
488
-
489
-
std.debug.print("sync: incremental sync since {s}\n", .{since_str});
490
-
sync_span.setAttribute("since", since_str);
491
-
492
-
// fetch new documents (use indexed_at, not created_at, because resynced
493
-
// documents can have old publication dates but recent insertion times).
494
-
//
495
-
// Paged via keyset (indexed_at, uri) — an unbounded since-window query
496
-
// grows without limit while sync keeps failing, until the response is
497
-
// structurally unfetchable (7h stall + memory churn, 2026-06-10). Pages
498
-
// are bounded regardless of how far behind we are, and each page holds
499
-
// the local write lock only briefly.
500
-
//
501
-
// PAGE_SIZE is set by payload, not row count: patent docs average ~24KB,
502
-
// and responses beyond tens of MB stall the backend's http path (turso
503
-
// itself serves them in ~1s — measured). 100 rows ≈ 2.5MB per page.
504
-
var new_docs: usize = 0;
505
-
{
506
-
var cursor_at_buf: [40]u8 = undefined;
507
-
var cursor_uri_buf: [256]u8 = undefined;
508
-
var cursor_at: []const u8 = std.fmt.bufPrint(&cursor_at_buf, "{s}", .{since_str}) catch since_str;
509
-
var cursor_uri: []const u8 = "";
510
-
511
-
var page_n: usize = 0;
512
-
while (true) {
513
-
page_n += 1;
514
-
std.debug.print("sync: fetching page {d} (cursor {s})\n", .{ page_n, cursor_at });
515
-
var result = turso.query(
516
-
\\SELECT uri, did, rkey, title, content, created_at, publication_uri,
517
-
\\ platform, source_collection, path, base_path, has_publication, indexed_at, embedded_at,
518
-
\\ COALESCE(cover_image, '') as cover_image, COALESCE(is_bridgyfed, 0) as is_bridgyfed,
519
-
\\ COALESCE(url_dead, 0) as url_dead
520
-
\\FROM documents
521
-
\\WHERE (indexed_at, uri) > (?, ?)
522
-
\\ORDER BY indexed_at, uri
523
-
\\LIMIT 100
524
-
, &.{ cursor_at, cursor_uri }) catch |err| {
525
-
logfire.warn("sync: incremental query failed (after {d} docs this cycle): {}", .{ new_docs, err });
526
-
sync_span.recordError(err);
527
-
return;
528
-
};
529
-
defer result.deinit();
530
-
531
-
std.debug.print("sync: page {d} -> {d} rows\n", .{ page_n, result.rows.len });
532
-
if (result.rows.len == 0) break;
533
-
534
-
{
535
-
local.lock();
536
-
defer local.unlock();
537
-
for (result.rows) |row| {
538
-
insertDocumentLocal(conn, row) catch {};
539
-
new_docs += 1;
540
-
}
541
-
}
542
-
543
-
// advance the keyset cursor past the last row of this page
544
-
const last = result.rows[result.rows.len - 1];
545
-
cursor_at = std.fmt.bufPrint(&cursor_at_buf, "{s}", .{last.text(12)}) catch break;
546
-
cursor_uri = std.fmt.bufPrint(&cursor_uri_buf, "{s}", .{last.text(0)}) catch break;
547
-
548
-
if (result.rows.len < 100) break;
549
-
}
550
-
551
-
// update sync time only after the full window landed
552
-
local.lock();
553
-
defer local.unlock();
554
-
var ts_buf: [20]u8 = undefined;
555
-
const inc_now_s: i64 = @intCast(@divFloor(Io.Timestamp.now(turso.io, .real).nanoseconds, std.time.ns_per_s));
556
-
const ts_str = std.fmt.bufPrint(&ts_buf, "{d}", .{inc_now_s}) catch "0";
557
-
conn.exec(
558
-
"INSERT OR REPLACE INTO sync_meta (key, value) VALUES ('last_sync', ?)",
559
-
.{ts_str},
560
-
) catch {};
561
-
}
562
-
563
-
// fetch new/updated publications (same since_str as documents).
564
-
// publications were silently missing from incremental sync before
565
-
// `publications.indexed_at` was added — they only synced on fullSync,
566
-
// so the local replica drifted by hundreds of rows over time.
567
-
var new_pubs: usize = 0;
568
-
pubs: {
569
-
var pub_result = turso.query(
570
-
\\SELECT uri, did, rkey, name, description, base_path, platform, indexed_at
571
-
\\FROM publications
572
-
\\WHERE indexed_at >= ?
573
-
\\ORDER BY indexed_at
574
-
, &.{since_str}) catch |err| {
575
-
logfire.warn("sync: incremental publications query failed: {}", .{err});
576
-
sync_span.recordError(err);
577
-
// fall through to tombstones — don't abort entire sync
578
-
break :pubs;
579
-
};
580
-
defer pub_result.deinit();
581
-
582
-
local.lock();
583
-
defer local.unlock();
584
-
585
-
for (pub_result.rows) |row| {
586
-
insertPublicationLocal(conn, row) catch {};
587
-
new_pubs += 1;
588
-
}
589
-
}
590
-
591
-
// sync deletions via tombstones (deleted_at is unix timestamp integer)
592
-
var deleted: usize = 0;
593
-
tombstone: {
594
-
var since_ts_buf: [20]u8 = undefined;
595
-
const since_ts_str = std.fmt.bufPrint(&since_ts_buf, "{d}", .{since_ts}) catch break :tombstone;
596
-
597
-
var tomb_result = turso.query(
598
-
"SELECT uri, record_type FROM tombstones WHERE deleted_at >= ?",
599
-
&.{since_ts_str},
600
-
) catch |err| {
601
-
logfire.warn("sync: tombstone query failed: {}", .{err});
602
-
sync_span.recordError(err);
603
-
break :tombstone;
604
-
};
605
-
defer tomb_result.deinit();
606
-
607
-
if (tomb_result.rows.len > 0) {
608
-
local.lock();
609
-
defer local.unlock();
610
-
for (tomb_result.rows) |row| {
611
-
const uri = row.text(0);
612
-
const record_type = row.text(1);
613
-
if (std.mem.eql(u8, record_type, "document")) {
614
-
// FTS keyed by documents.rowid — drop it before the row
615
-
conn.exec("DELETE FROM documents_fts WHERE rowid = (SELECT rowid FROM documents WHERE uri = ?)", .{uri}) catch {};
616
-
conn.exec("DELETE FROM documents WHERE uri = ?", .{uri}) catch {};
617
-
conn.exec("DELETE FROM document_tags WHERE document_uri = ?", .{uri}) catch {};
618
-
} else if (std.mem.eql(u8, record_type, "publication")) {
619
-
conn.exec("DELETE FROM publications WHERE uri = ?", .{uri}) catch {};
620
-
conn.exec("DELETE FROM publications_fts WHERE uri = ?", .{uri}) catch {};
621
-
}
622
-
deleted += 1;
623
-
}
624
-
}
625
-
}
626
-
627
-
// periodic WAL checkpoint to prevent unbounded growth
628
-
local.lock();
629
-
conn.exec("PRAGMA wal_checkpoint(PASSIVE)", .{}) catch {};
630
-
local.unlock();
631
-
632
-
sync_span.setAttribute("new_docs", @as(i64, @intCast(new_docs)));
633
-
sync_span.setAttribute("new_pubs", @as(i64, @intCast(new_pubs)));
634
-
sync_span.setAttribute("deleted", @as(i64, @intCast(deleted)));
635
-
636
-
if (new_docs > 0 or new_pubs > 0 or deleted > 0) {
637
-
std.debug.print("sync: incremental sync — {d} new docs, {d} new pubs, {d} tombstone deletions\n", .{ new_docs, new_pubs, deleted });
638
-
}
639
223
}
640
224
641
225
fn insertDocumentLocal(conn: zqlite.Conn, row: anytype) !void {