Skip to main content

max / synckit

Hold unappliable remote changes instead of losing them at the cursor D4 of the 2026-07-27 backlog triage, steps 1-4. The apply path folded every unapplied row into one `skipped` counter, `pull_scope` then dropped that counter before it reached PullOutcome, and the cursor advanced regardless. The server never re-sends an entry, so the row was gone: device A had it, device B did not, and both reported a clean sync. This was the only genuine data-loss defect in the 425-row triage. Split the one counter into three outcomes that need opposite handling. Filtered is policy (exclude_where, DeleteMode::Ignore) and never surfaced. Rejected can never succeed (no object payload, no reconstructable key, no insertable columns) and is held for display only. Deferred can succeed later (unknown table, a constraint violation from a late parent) and is held and retried. New store::deferred holds the deferred and rejected entries in sync_deferred, keyed one row per (scope, table, row_id). The stored payload is the entry AS PULLED, not as the conflict layer resolved it, so a retry re-enters resolve_pull and is judged against current local state; replaying the resolved entry would decide a conflict against a database that has since moved and could clobber a newer local edit. Held entries go in front of the next page, so ordering and supersession fall out of the existing resolution path. Five attempts, then the entry is promoted to rejected: it stays visible but stops costing work, which bounds the held set when a parent genuinely never arrives. The FK half (the folded-in post-launch item) uses PRAGMA foreign_key_check over the touched tables that do not declare references_unsynced, run inside the apply transaction. The batch-wide foreign_keys=OFF is all-or-nothing and exists only for the relaxed tables; on every other table it let bad rows land silently and the divergence resurfaced later as orphans. A violating row belonging to this batch is now removed and deferred; one that matches no change in the batch is a pre-existing orphan and is logged, not deleted, since there is no pulled entry to hold for it. Note for anyone revisiting: defer_foreign_keys is settable inside a transaction but aborts the commit rather than skipping a row. Also fixes a related trap this work would otherwise have walked into: the pull recorded every resolved entry in the committed-HLC ledger, including ones the apply never wrote. A held entry would then be gated out of its own retry, since the gate drops anything not newer than the committed clock. Only applied and filtered rows advance the ledger now. The table is created from both migration_sql and ensure_scope_schema, so an existing install picks it up on the next connection open rather than waiting for its app to re-run a migration. Steps 5 (the GO surface) is not in this commit. synckit-client 415 lib + 111 integration tests pass, clippy clean; goingson, audiofiles-sync and balanced-breakfast-desktop all still type-check against the changed API.
Co-Authored-By
Claude Opus 5 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-27 22:24 UTC
Signed with PGP, not checked
Commit: 6467dd5cf4c041434534974e0400fa82e9ca8056
Parent: 08b1f86
6 files changed, +1018 insertions, -64 deletions
M synckit-client/src/store/apply.rs +285 -41
@@ -16,11 +16,12 @@
16 16 use std::collections::{HashMap, HashSet};
17 17
18 18 use rusqlite::types::ToSql;
19 - use rusqlite::{Connection, ErrorCode, Transaction};
19 + use rusqlite::{Connection, ErrorCode, OptionalExtension, Transaction};
20 20 use serde_json::Value;
21 21 use tracing::warn;
22 22
23 23 use super::db::with_applying_remote;
24 + use super::migrate::row_id_expr;
24 25 use super::schema::{DeleteMode, SyncMode, SyncSchema, SyncTable};
25 26 use crate::error::Result;
26 27 use crate::types::{ChangeEntry, ChangeOp};
@@ -28,18 +29,77 @@
28 29 /// A JSON null to borrow when a payload omits a whitelisted column.
29 30 static JSON_NULL: Value = Value::Null;
30 31
32 + /// A row the apply could not write, and why.
33 + ///
34 + /// Identified by `(table, row_id)`, which survives conflict resolution, so the
35 + /// caller can match it back to the entry as it was pulled and hold that verbatim.
36 + #[derive(Debug, Clone, PartialEq, Eq)]
37 + pub struct Unapplied {
38 + /// Table the change targeted.
39 + pub table: String,
40 + /// Wire row id of the change.
41 + pub row_id: String,
42 + /// Why it could not be written, for display and for the hold.
43 + pub cause: String,
44 + }
45 +
46 + impl Unapplied {
47 + fn new(table: &str, row_id: &str, cause: impl Into<String>) -> Self {
48 + Self {
49 + table: table.to_string(),
50 + row_id: row_id.to_string(),
51 + cause: cause.into(),
52 + }
53 + }
54 + }
55 +
31 56 /// Summary of an apply pass.
57 + ///
58 + /// The three not-applied outcomes are kept apart because they need opposite
59 + /// handling, and folding them into one number is what let a lost row look like a
60 + /// clean sync:
61 + ///
62 + /// - [`filtered`](Self::filtered) is policy, not failure. An `exclude_where`
63 + /// predicate or a `DeleteMode::Ignore` says the row does not belong on this
64 + /// device. Never surfaced, never held, never retried.
65 + /// - [`rejected`](Self::rejected) can never succeed: the same bytes will fail the
66 + /// same way. Held for display, not retried.
67 + /// - [`deferred`](Self::deferred) can succeed later: an unknown table clears on a
68 + /// client upgrade, a constraint violation clears when the missing parent lands.
69 + /// Held and retried.
32 70 #[derive(Debug, Default, PartialEq, Eq)]
33 71 pub struct ApplyOutcome {
34 72 /// Rows written (insert/update/delete/tombstone).
35 73 pub applied: usize,
36 - /// Rows intentionally dropped or unappliable (excluded, unknown table,
37 - /// missing key, or a skipped constraint violation).
38 - pub skipped: usize,
74 + /// Rows the schema deliberately dropped. Not a failure.
75 + pub filtered: usize,
76 + /// Rows that are permanently unappliable.
77 + pub rejected: Vec<Unapplied>,
78 + /// Rows that could not be applied yet, but might be later.
79 + pub deferred: Vec<Unapplied>,
39 80 /// Distinct tables an applied change touched, for selective UI invalidation.
40 81 pub changed_tables: HashSet<String>,
41 82 }
42 83
84 + impl ApplyOutcome {
85 + /// Rows that were not written, for whatever reason. The old single `skipped`
86 + /// counter, kept as a derived total for a caller that only wants "did
87 + /// everything land".
88 + pub fn unapplied(&self) -> usize {
89 + self.filtered + self.rejected.len() + self.deferred.len()
90 + }
91 + }
92 +
93 + /// What became of one row.
94 + enum RowOutcome {
95 + /// Written.
96 + Applied,
97 + /// Dropped on purpose, per the table's policy.
98 + Filtered,
99 + /// Unappliable, and retrying cannot change that.
100 + Rejected(&'static str),
101 + }
102 +
43 103 /// Apply resolved remote `changes` to `conn` in FK-safe order.
44 104 ///
45 105 /// Upserts run parents-first (schema declaration order); deletes run
@@ -64,8 +124,9 @@
64 124 conn.execute_batch("PRAGMA foreign_keys=OFF")?;
65 125 }
66 126
67 - let outcome =
68 - with_applying_remote(conn, |tx| apply_inner(tx, schema, &by_name, changes, scope));
127 + let outcome = with_applying_remote(conn, |tx| {
128 + apply_inner(tx, schema, &by_name, changes, scope, fk_off)
129 + });
69 130
70 131 if fk_off {
71 132 // Restore enforcement regardless of the apply result. A failed restore is
@@ -85,18 +146,24 @@
85 146 by_name: &HashMap<&str, &SyncTable>,
86 147 changes: &[ChangeEntry],
87 148 scope: &str,
149 + fk_relaxed: bool,
88 150 ) -> Result<ApplyOutcome> {
89 151 let mut out = ApplyOutcome::default();
90 152
91 - // Changes targeting a table the schema doesn't know are dropped (defensive,
92 - // a hostile or newer server could send one), not applied.
93 - let unknown = changes
153 + // Changes targeting a table the schema doesn't know are held, not applied: a
154 + // hostile or newer server could send one, but so could a server that is simply
155 + // ahead of this client, and that case clears on upgrade. Deferred, so the row
156 + // survives the upgrade rather than being lost at the cursor.
157 + for change in changes
94 158 .iter()
95 159 .filter(|c| !by_name.contains_key(c.table.as_str()))
96 - .count();
97 - if unknown > 0 {
98 - warn!("dropping {unknown} change(s) for unknown table(s)");
99 - out.skipped += unknown;
160 + {
161 + warn!(table = %change.table, "deferring change for unknown table");
162 + out.deferred.push(Unapplied::new(
163 + &change.table,
164 + &change.row_id,
165 + "table is not in this client's schema",
166 + ));
100 167 }
101 168
102 169 // Precompute NOT NULL columns for each Full-mode table once, not per row.
@@ -115,6 +182,7 @@
115 182 step(
116 183 &mut out,
117 184 table,
185 + change,
118 186 apply_upsert(tx, table, change, &not_null, scope),
119 187 )?;
120 188 }
@@ -126,43 +194,152 @@
126 194 .iter()
127 195 .filter(|c| c.table == table.name && matches!(c.op, ChangeOp::Delete))
128 196 {
129 - step(&mut out, table, apply_delete(tx, table, change))?;
197 + step(&mut out, table, change, apply_delete(tx, table, change))?;
130 198 }
131 199 }
132 200
201 + if fk_relaxed {
202 + fk_sweep(tx, by_name, changes, &mut out)?;
203 + }
204 +
133 205 Ok(out)
134 206 }
135 207
136 - /// Fold one row's result into the outcome. A constraint violation is skipped and
208 + /// Fold one row's result into the outcome. A constraint violation is deferred and
137 209 /// logged; any other SQLite error is returned so `?` rolls the whole batch back
138 210 /// (converting to `SyncKitError` at the caller).
139 211 fn step(
140 212 out: &mut ApplyOutcome,
141 213 table: &SyncTable,
142 - result: rusqlite::Result<bool>,
214 + change: &ChangeEntry,
215 + result: rusqlite::Result<RowOutcome>,
143 216 ) -> rusqlite::Result<()> {
144 217 match result {
145 - Ok(true) => {
218 + Ok(RowOutcome::Applied) => {
146 219 out.applied += 1;
147 220 out.changed_tables.insert(table.name.to_string());
148 221 Ok(())
149 222 }
150 - Ok(false) => {
151 - out.skipped += 1;
223 + Ok(RowOutcome::Filtered) => {
224 + out.filtered += 1;
225 + Ok(())
226 + }
227 + Ok(RowOutcome::Rejected(cause)) => {
228 + warn!(table = table.name, row_id = %change.row_id, "rejecting row: {cause}");
229 + out.rejected
230 + .push(Unapplied::new(table.name, &change.row_id, cause));
152 231 Ok(())
153 232 }
154 233 Err(e) if is_constraint_violation(&e) => {
155 234 warn!(
156 235 table = table.name,
157 - "skipping row on constraint violation: {e}"
236 + row_id = %change.row_id,
237 + "deferring row on constraint violation: {e}"
158 238 );
159 - out.skipped += 1;
239 + out.deferred.push(Unapplied::new(
240 + table.name,
241 + &change.row_id,
242 + format!("constraint violation: {e}"),
243 + ));
160 244 Ok(())
161 245 }
162 246 Err(e) => Err(e),
163 247 }
164 248 }
165 249
250 + // foreign keys
251 +
252 + /// Catch the violations the batch-wide `foreign_keys=OFF` let through.
253 + ///
254 + /// The pragma is all-or-nothing (SQLite has no per-table toggle) and it is only
255 + /// meant to cover tables that declare `references_unsynced`, whose parents may
256 + /// legitimately never arrive over sync. On every *other* table in the batch it
257 + /// silently lets bad rows land, and the divergence resurfaces later as orphans
258 + /// rather than as something anyone reported. That defeats the deferred counter,
259 + /// which exists to catch exactly this class.
260 + ///
261 + /// So: run `PRAGMA foreign_key_check` over the tables the batch touched that do
262 + /// *not* declare the relaxation, still inside the apply transaction. Any
263 + /// violating row that belongs to a change from this batch is removed and deferred,
264 + /// so it retries once its parent lands. Restricted to touched tables because the
265 + /// argument-less form scans the whole database.
266 + ///
267 + /// A violation that matches no change in this batch is a pre-existing orphan from
268 + /// an earlier relaxed apply. It is logged and left alone: there is no pulled entry
269 + /// to hold for it, so deleting it would destroy a row with no way to get it back.
270 + fn fk_sweep(
271 + tx: &Transaction<'_>,
272 + by_name: &HashMap<&str, &SyncTable>,
273 + changes: &[ChangeEntry],
274 + out: &mut ApplyOutcome,
275 + ) -> rusqlite::Result<()> {
276 + let touched: HashSet<&str> = changes
277 + .iter()
278 + .filter_map(|c| by_name.get(c.table.as_str()))
279 + .filter(|t| !t.references_unsynced)
280 + .map(|t| t.name)
281 + .collect();
282 +
283 + for name in touched {
284 + let table = by_name[name];
285 + // rowid is NULL for a WITHOUT ROWID table, which we cannot address here;
286 + // such a row is reported and left in place.
287 + let mut stmt = tx.prepare(&format!("PRAGMA foreign_key_check({name})"))?;
288 + let rowids: Vec<Option<i64>> = stmt
289 + .query_map([], |r| r.get::<_, Option<i64>>(1))?
290 + .collect::<rusqlite::Result<_>>()?;
291 + drop(stmt);
292 +
293 + for rowid in rowids {
294 + let Some(rowid) = rowid else {
295 + warn!(
296 + table = name,
297 + "foreign-key violation on a WITHOUT ROWID table, cannot isolate the row"
298 + );
299 + continue;
300 + };
301 + // Derive the wire row id the same way the triggers do, so it matches
302 + // the changes in this batch even for a hashed table.
303 + let row_id: Option<String> = tx
304 + .query_row(
305 + &format!(
306 + "SELECT {} FROM {name} WHERE rowid = ?1",
307 + row_id_expr(table, name)
308 + ),
309 + [rowid],
310 + |r| r.get(0),
311 + )
312 + .optional()?;
313 + let Some(row_id) = row_id else { continue };
314 +
315 + if !changes
316 + .iter()
317 + .any(|c| c.table == name && c.row_id == row_id)
318 + {
319 + warn!(
320 + table = name,
321 + "pre-existing foreign-key orphan left in place; no pulled entry to retry"
322 + );
323 + continue;
324 + }
325 +
326 + tx.execute(&format!("DELETE FROM {name} WHERE rowid = ?1"), [rowid])?;
327 + warn!(
328 + table = name,
329 + row_id = %row_id,
330 + "deferring row: foreign-key violation hidden by the batch-wide relaxation"
331 + );
332 + out.applied = out.applied.saturating_sub(1);
333 + out.deferred.push(Unapplied::new(
334 + name,
335 + &row_id,
336 + "foreign key violation: referenced parent row is missing",
337 + ));
338 + }
339 + }
340 + Ok(())
341 + }
342 +
166 343 /// Names of the NOT NULL columns of `table`, read from the live schema so the
167 344 /// list can never drift from the migrations.
168 345 fn not_null_columns(tx: &Transaction<'_>, table: &str) -> rusqlite::Result<HashSet<String>> {
@@ -217,12 +394,12 @@
217 394 change: &ChangeEntry,
218 395 not_null: &HashMap<&str, HashSet<String>>,
219 396 scope: &str,
220 - ) -> rusqlite::Result<bool> {
397 + ) -> rusqlite::Result<RowOutcome> {
221 398 let Some(data) = change.data.as_ref().filter(|v| v.is_object()) else {
222 - return Ok(false); // an upsert with no object payload cannot be applied
399 + return Ok(RowOutcome::Rejected("upsert carries no object payload"));
223 400 };
224 401 if is_excluded(tx, table, Some(data))? {
225 - return Ok(false);
402 + return Ok(RowOutcome::Filtered);
226 403 }
227 404 match &table.mode {
228 405 SyncMode::PartialUpdate { set } => apply_partial_update(tx, table, data, set),
@@ -241,7 +418,7 @@
241 418 data: &Value,
242 419 not_null: &HashSet<String>,
243 420 scope: &str,
244 - ) -> rusqlite::Result<bool> {
421 + ) -> rusqlite::Result<RowOutcome> {
245 422 // A remote null for a NOT NULL column is invalid input the changelog never
246 423 // emits; omit such columns so a new row takes the schema default and an
247 424 // existing row keeps its value. Nullable columns keep null so legitimate
@@ -264,7 +441,7 @@
264 441 }
265 442 }
266 443 if insert_cols.is_empty() {
267 - return Ok(false);
444 + return Ok(RowOutcome::Rejected("no insertable columns in the payload"));
268 445 }
269 446
270 447 // Group provenance: stamp a group-scoped table's `group_id` column from the
@@ -331,7 +508,7 @@
331 508 }
332 509
333 510 exec(tx, &sql, &params)?;
334 - Ok(true)
511 + Ok(RowOutcome::Applied)
335 512 }
336 513
337 514 fn apply_partial_update(
@@ -339,9 +516,9 @@
339 516 table: &SyncTable,
340 517 data: &Value,
341 518 set: &[&str],
342 - ) -> rusqlite::Result<bool> {
519 + ) -> rusqlite::Result<RowOutcome> {
343 520 let Some(pk) = pk_bindings(table, data, None) else {
344 - return Ok(false); // no primary key to target
521 + return Ok(RowOutcome::Rejected("no primary key to target"));
345 522 };
346 523 let set_present: Vec<&str> = set
347 524 .iter()
@@ -349,7 +526,7 @@
349 526 .filter(|c| data.get(*c).is_some())
350 527 .collect();
351 528 if set_present.is_empty() {
352 - return Ok(false);
529 + return Ok(RowOutcome::Rejected("no updatable columns in the payload"));
353 530 }
354 531
355 532 let mut params: Vec<Box<dyn ToSql>> = Vec::new();
@@ -377,7 +554,7 @@
377 554
378 555 let sql = format!("UPDATE {} SET {set_sql} WHERE {where_sql}", table.name);
379 556 exec(tx, &sql, &params)?;
380 - Ok(true)
557 + Ok(RowOutcome::Applied)
381 558 }
382 559
383 560 // delete
@@ -386,19 +563,21 @@
386 563 tx: &Transaction<'_>,
387 564 table: &SyncTable,
388 565 change: &ChangeEntry,
389 - ) -> rusqlite::Result<bool> {
566 + ) -> rusqlite::Result<RowOutcome> {
390 567 if matches!(table.deletes, DeleteMode::Ignore) {
391 - return Ok(false);
568 + return Ok(RowOutcome::Filtered);
392 569 }
393 570 if is_excluded(tx, table, change.data.as_ref())? {
394 - return Ok(false);
571 + return Ok(RowOutcome::Filtered);
395 572 }
396 573 let Some(pk) = pk_bindings(
397 574 table,
398 575 change.data.as_ref().unwrap_or(&JSON_NULL),
399 576 Some(&change.row_id),
400 577 ) else {
401 - return Ok(false); // can't reconstruct the key, skip
578 + return Ok(RowOutcome::Rejected(
579 + "delete key cannot be reconstructed from the payload or row id",
580 + ));
402 581 };
403 582
404 583 let mut params: Vec<Box<dyn ToSql>> = Vec::new();
@@ -426,7 +605,7 @@
426 605 DeleteMode::Ignore => unreachable!("returned above"),
427 606 };
428 607 exec(tx, &sql, &params)?;
429 - Ok(true)
608 + Ok(RowOutcome::Applied)
430 609 }
431 610
432 611 /// Reconstruct the primary-key bindings for a row.
@@ -684,7 +863,7 @@
684 863 .unwrap();
685 864 assert_eq!(name, None);
686 865 // A null for a NOT NULL column (child.parent_id) is omitted, so an insert
687 - // takes no value for it → constraint violation → skipped, not fatal.
866 + // takes no value for it → constraint violation → deferred, not fatal.
688 867 let o = apply(
689 868 &mut conn,
690 869 &[upsert(
@@ -694,7 +873,8 @@
694 873 )],
695 874 );
696 875 assert_eq!(o.applied, 0);
697 - assert_eq!(o.skipped, 1);
876 + assert_eq!(o.deferred.len(), 1);
877 + assert_eq!(o.deferred[0].row_id, "c1");
698 878 }
699 879
700 880 #[test]
@@ -915,7 +1095,7 @@
915 1095 ],
916 1096 );
917 1097 assert_eq!(o.applied, 1);
918 - assert_eq!(o.skipped, 1);
1098 + assert_eq!(o.deferred.len(), 1, "the poison row is held, not lost");
919 1099 assert_eq!(
920 1100 conn.query_row("SELECT COUNT(*) FROM parent", [], |r| r.get::<_, i64>(0))
921 1101 .unwrap(),
@@ -924,10 +1104,104 @@
924 1104 }
925 1105
926 1106 #[test]
927 - fn unknown_table_change_is_skipped() {
1107 + fn unknown_table_change_is_deferred_not_dropped() {
928 1108 let mut conn = db();
929 1109 let o = apply(&mut conn, &[upsert("nonexistent", "x", json!({"id":"x"}))]);
930 1110 assert_eq!(o.applied, 0);
931 - assert_eq!(o.skipped, 1);
1111 + // Deferred rather than rejected: the table may exist after a client
1112 + // upgrade, and then the held entry applies.
1113 + assert_eq!(o.deferred.len(), 1);
1114 + assert_eq!(o.deferred[0].table, "nonexistent");
1115 + assert!(o.rejected.is_empty());
1116 + }
1117 +
1118 + #[test]
1119 + fn an_excluded_row_is_filtered_not_held() {
1120 + let mut conn = db();
1121 + // cfg's include predicate is "key NOT LIKE 'sync_%'", so a sync_ key is
1122 + // excluded on import. That is policy, not failure, and must never reach
1123 + // the dead-letter.
1124 + let o = apply(
1125 + &mut conn,
1126 + &[upsert(
1127 + "cfg",
1128 + "sync_token",
1129 + json!({"key":"sync_token","value":"x"}),
1130 + )],
1131 + );
1132 + assert_eq!(o.applied, 0);
1133 + assert_eq!(o.filtered, 1);
1134 + assert!(o.deferred.is_empty());
1135 + assert!(o.rejected.is_empty());
1136 + }
1137 +
1138 + #[test]
1139 + fn a_payloadless_upsert_is_rejected_not_deferred() {
1140 + let mut conn = db();
1141 + let mut change = upsert("parent", "p1", json!({"id":"p1"}));
1142 + change.data = None;
1143 + let o = apply(&mut conn, &[change]);
1144 + assert_eq!(o.applied, 0);
1145 + assert_eq!(
1146 + o.rejected.len(),
1147 + 1,
1148 + "identical bytes would fail identically"
1149 + );
1150 + assert!(o.deferred.is_empty());
1151 + }
1152 +
1153 + #[test]
1154 + fn fk_sweep_catches_what_the_batch_wide_relaxation_hides() {
1155 + let mut conn = db();
1156 + // `reffer` declares references_unsynced, so the whole apply runs with
1157 + // foreign_keys=OFF. Without the sweep, the child row below lands with a
1158 + // missing parent and nothing is reported.
1159 + let o = apply(
1160 + &mut conn,
1161 + &[
1162 + upsert("reffer", "r1", json!({"id":"r1","ext_id":404})),
1163 + upsert(
1164 + "child",
1165 + "c1",
1166 + json!({"id":"c1","parent_id":"missing","note":"x"}),
1167 + ),
1168 + ],
1169 + );
1170 +
1171 + assert_eq!(
1172 + conn.query_row("SELECT COUNT(*) FROM child", [], |r| r.get::<_, i64>(0))
1173 + .unwrap(),
1174 + 0,
1175 + "the orphan is removed, not left to resurface later"
Lines truncated
M synckit-client/src/store/db.rs +4
@@ -125,6 +125,10 @@
125 125 INSERT OR IGNORE INTO sync_scope_cursor (scope, cursor)
126 126 SELECT '', CAST(value AS INTEGER) FROM sync_state WHERE key = 'pull_cursor';",
127 127 )?;
128 + // The dead-letter hold, created here as well as in `migration_sql` so an
129 + // install that predates it gets the table on the next connection open rather
130 + // than on the next time the app happens to re-run its migration.
131 + conn.execute_batch(super::deferred::DEFERRED_DDL)?;
128 132 Ok(())
129 133 }
130 134
M synckit-client/src/store/migrate.rs +1
@@ -82,6 +82,7 @@
82 82 /// ledger, an optional `row_id_salt`, and every table's triggers.
83 83 pub fn migration_sql(&self) -> String {
84 84 let mut out = String::from(BASE_TABLES);
85 + out.push_str(super::deferred::DEFERRED_DDL);
85 86 if self.any_hashed() {
86 87 out.push_str(SALT_SEED);
87 88 }
M synckit-client/src/store/mod.rs +3 -1
@@ -21,6 +21,7 @@
21 21 #[cfg(feature = "store")]
22 22 pub mod config;
23 23 pub mod db;
24 + pub mod deferred;
24 25 pub mod facade;
25 26 pub mod hlc;
26 27 pub mod migrate;
@@ -28,7 +29,7 @@
28 29 pub mod schema;
29 30 pub mod sync;
30 31
31 - pub use apply::{ApplyOutcome, apply_remote_changes};
32 + pub use apply::{ApplyOutcome, Unapplied, apply_remote_changes};
32 33 pub use blob::{
33 34 BlobOutcome, BlobPolicy, BlobRef, BlobTransport, download_blobs, download_one, sync_blobs,
34 35 upload_blobs,
@@ -38,6 +39,7 @@
38 39 DbSource, clear_applying_remote, device_id, get_scope_cursor, get_sync_state,
39 40 get_sync_state_or, set_device_id, set_scope_cursor, set_sync_state, with_applying_remote,
40 41 };
42 + pub use deferred::{HeldEntry, HoldCounts, HoldState, MAX_ATTEMPTS};
41 43 pub use facade::{SchedulerHandle, SyncConfig, SyncOutcome, SyncStore, SyncStoreBuilder};
42 44 pub use hlc::{
43 45 committed_hlc, load_clock, observe, record_committed, resolve_pull, set_committed,
M synckit-client/src/store/sync.rs +225 -22
@@ -12,7 +12,7 @@
12 12 //! behind [`SyncTransport`] so the loops test end-to-end against an in-memory
13 13 //! fake, and `SyncKitClient` satisfies it in production.
14 14
15 - use std::collections::HashSet;
15 + use std::collections::{HashMap, HashSet};
16 16 use std::future::Future;
17 17
18 18 use chrono::Utc;
@@ -22,7 +22,8 @@
22 22 use super::db::{
23 23 DbSource, device_id, get_scope_cursor, set_device_id, set_scope_cursor, set_sync_state,
24 24 };
25 - use super::hlc::{record_committed, resolve_pull, set_committed, stamp_pending};
25 + use super::deferred;
26 + use super::hlc::{resolve_pull, set_committed, stamp_pending};
26 27 use super::migrate::{json_object, row_id_expr};
27 28 use super::schema::{SyncMode, SyncSchema};
28 29 use crate::client::SyncKitClient;
@@ -246,10 +247,25 @@
246 247 }
247 248
248 249 /// Outcome of a pull pass.
250 + ///
251 + /// The unapplied counts are carried up deliberately. They used to stop at
252 + /// [`ApplyOutcome`] and never reach a caller, so an app could not have shown a
253 + /// failed row even if it wanted to; a divergence looked exactly like a clean
254 + /// sync. Anything counted in [`deferred`](Self::deferred) or
255 + /// [`rejected`](Self::rejected) is durably held (see
256 + /// [`deferred`](super::deferred)), so the numbers are a pointer into that table
257 + /// rather than the only trace of the failure.
249 258 #[derive(Debug, Default, PartialEq, Eq)]
250 259 pub struct PullOutcome {
251 260 /// Number of remote changes applied.
252 261 pub applied: u64,
262 + /// Changes the schema deliberately dropped (`exclude_where`,
263 + /// `DeleteMode::Ignore`). Not failures, not held.
264 + pub filtered: u64,
265 + /// Changes that can never be applied, held for display only.
266 + pub rejected: u64,
267 + /// Changes that could not be applied yet, held and retried on later pulls.
268 + pub deferred: u64,
253 269 /// Tables whose rows changed as a result of the pull.
254 270 pub changed_tables: HashSet<String>,
255 271 }
@@ -461,40 +477,43 @@
461 477 }
462 478 };
463 479
464 - if pulled.is_empty() {
465 - let db_c = db.clone();
466 - let sk = scope_key.clone();
467 - tokio::task::spawn_blocking(move || set_scope_cursor(&db_c.open()?, &sk, new_cursor))
468 - .await
469 - .map_err(|e| join_err(&e))??;
470 - break;
471 - }
472 -
480 + // An empty page still runs an apply when something is held, so a row
481 + // waiting on a parent that arrived through another path clears without
482 + // needing new remote traffic to carry it. It is still the last page.
483 + let is_empty_page = pulled.is_empty();
473 484 let db_apply = db.clone();
474 485 let schema = schema.clone();
475 486 let sk = scope_key.clone();
476 487 let outcome = tokio::task::spawn_blocking(move || {
477 - apply_pull(
478 - &mut db_apply.open()?,
479 - &schema,
480 - device_id,
481 - pulled,
482 - new_cursor,
483 - &sk,
484 - )
488 + let mut conn = db_apply.open()?;
489 + if pulled.is_empty() && !deferred::has_retryable(&conn, &sk)? {
490 + set_scope_cursor(&conn, &sk, new_cursor)?;
491 + return Ok(ApplyOutcome::default());
492 + }
493 + apply_pull(&mut conn, &schema, device_id, pulled, new_cursor, &sk)
485 494 })
486 495 .await
487 496 .map_err(|e| join_err(&e))??;
488 497
489 498 out.applied += outcome.applied as u64;
499 + out.filtered += outcome.filtered as u64;
500 + out.rejected += outcome.rejected.len() as u64;
501 + out.deferred += outcome.deferred.len() as u64;
490 502 out.changed_tables.extend(outcome.changed_tables);
491 - if !has_more {
503 + if is_empty_page || !has_more {
492 504 break;
493 505 }
494 506 }
495 507 Ok(out)
496 508 }
497 509
510 + /// Resolve and apply one page, retrying whatever the scope is holding first.
511 + ///
512 + /// Held entries go in front of the new page and back through `resolve_pull`, so
513 + /// the conflict layer judges them against the local state of *now* rather than
514 + /// replaying a decision made when they first failed. That ordering also means a
515 + /// newer change to the same row supersedes the held one, since resolution
516 + /// collapses a row to its highest HLC.
498 517 fn apply_pull(
499 518 conn: &mut Connection,
500 519 schema: &SyncSchema,
@@ -503,9 +522,40 @@
503 522 new_cursor: i64,
504 523 scope_key: &str,
505 524 ) -> Result<ApplyOutcome> {
506 - let resolved = resolve_pull(conn, schema, device_id, pulled, Utc::now())?;
525 + let retry = deferred::load_retryable(conn, scope_key)?;
526 + let retried: HashSet<deferred::RowKey> =
527 + retry.iter().map(|p| deferred::key_of(&p.entry)).collect();
528 +
529 + // Index the batch as pulled, before resolution rewrites it. What gets held
530 + // must be the wire entry, not the resolved one.
531 + let mut batch: HashMap<deferred::RowKey, PulledChange> = HashMap::new();
532 + for p in retry.iter().chain(pulled.iter()) {
533 + batch.insert(deferred::key_of(&p.entry), p.clone());
534 + }
535 +
536 + let mut all = retry;
537 + all.extend(pulled);
538 +
539 + let resolved = resolve_pull(conn, schema, device_id, all, Utc::now())?;
507 540 let outcome = apply_remote_changes(conn, schema, &resolved, scope_key)?;
508 - record_committed(conn, &resolved)?;
541 +
542 + // Only what actually landed advances the committed ledger. Recording an
543 + // unapplied row would gate its own retry out on the next pull, since the gate
544 + // drops any entry whose HLC is not newer than the committed one, and the row
545 + // would be lost exactly the way the hold exists to prevent.
546 + let unapplied: HashSet<deferred::RowKey> = outcome
547 + .deferred
548 + .iter()
549 + .chain(outcome.rejected.iter())
550 + .map(|u| (u.table.clone(), u.row_id.clone()))
551 + .collect();
552 + for entry in &resolved {
553 + if !unapplied.contains(&deferred::key_of(entry)) {
554 + set_committed(conn, &entry.table, &entry.row_id, &entry.hlc)?;
555 + }
556 + }
557 +
558 + deferred::settle(conn, scope_key, &outcome, &batch, &retried)?;
509 559 set_scope_cursor(conn, scope_key, new_cursor)?;
510 560 Ok(outcome)
511 561 }
@@ -745,6 +795,159 @@
745 795 .unwrap();
746 796 }
747 797
798 + /// A schema whose `child` table has a real foreign key to `parent`, so a
799 + /// child arriving first violates a constraint instead of quietly landing.
800 + fn fk_schema() -> SyncSchema {
801 + SyncSchema::new(vec![
802 + SyncTable::full("parent", &["id", "name"]),
803 + SyncTable::full("child", &["id", "parent_id"]),
804 + ])
805 + }
806 +
807 + fn fk_device(path: &std::path::Path, n: u128) -> (DbSource, DeviceId) {
808 + let db = DbSource::path(path);
809 + let conn = db.open().unwrap();
810 + conn.execute_batch(
811 + "CREATE TABLE parent (id TEXT PRIMARY KEY, name TEXT);
812 + CREATE TABLE child (id TEXT PRIMARY KEY, parent_id TEXT NOT NULL REFERENCES parent(id));",
813 + )
814 + .unwrap();
815 + conn.execute_batch(&fk_schema().migration_sql()).unwrap();
816 + (db, DeviceId::new(uuid::Uuid::from_u128(n)))
817 + }
818 +
819 + /// Put an entry on the fake server as if another device had pushed it.
820 + fn serve(server: &FakeServer, table: &str, row_id: &str, data: serde_json::Value) {
821 + server.log.lock().unwrap().push((
822 + DeviceId::new(uuid::Uuid::from_u128(0xAA)),
823 + ChangeEntry {
824 + table: table.into(),
825 + op: ChangeOp::Insert,
826 + row_id: row_id.into(),
827 + timestamp: Utc::now(),
828 + hlc: crate::types::hlc_legacy_floor(),
829 + data: Some(data),
830 + extra: serde_json::Map::default(),
831 + },
832 + ));
833 + }
834 +
835 + fn row_count(db: &DbSource, table: &str) -> i64 {
836 + db.open()
837 + .unwrap()
838 + .query_row(&format!("SELECT COUNT(*) FROM {table}"), [], |r| r.get(0))
839 + .unwrap()
840 + }
841 +
842 + #[tokio::test]
843 + async fn a_child_that_arrives_before_its_parent_is_held_and_lands_on_the_next_pull() {
844 + let dir = tempdir();
845 + let (db, node) = fk_device(&dir.join("fk.db"), 11);
846 + let server = FakeServer::default();
847 +
848 + // The child arrives alone. Its parent does not exist yet, so it cannot be
849 + // written; before the hold existed this row was gone for good, because the
850 + // cursor moved past it and the server never sends an entry twice.
851 + serve(
852 + &server,
853 + "child",
854 + "c1",
855 + serde_json::json!({"id":"c1","parent_id":"p1"}),
856 + );
857 + let out = pull_scope(&db, &server, &fk_schema(), node, SyncScope::Personal)
858 + .await
859 + .unwrap();
860 +
861 + assert_eq!(out.applied, 0);
862 + assert_eq!(out.deferred, 1);
863 + assert_eq!(row_count(&db, "child"), 0);
864 + {
865 + let conn = db.open().unwrap();
866 + assert_eq!(
867 + get_scope_cursor(&conn, "").unwrap(),
868 + 1,
869 + "the cursor still advances; the hold is what makes that safe"
870 + );
871 + assert_eq!(deferred::counts(&conn, "").unwrap().deferred, 1);
872 + }
873 +
874 + // The parent lands on the next pull, and the held child rides in with it.
875 + serve(
876 + &server,
877 + "parent",
878 + "p1",
879 + serde_json::json!({"id":"p1","name":"p"}),
880 + );
881 + let out = pull_scope(&db, &server, &fk_schema(), node, SyncScope::Personal)
882 + .await
883 + .unwrap();
884 +
885 + assert_eq!(out.applied, 2, "the new parent plus the retried child");
886 + assert_eq!(out.deferred, 0);
887 + assert_eq!(row_count(&db, "child"), 1);
888 + assert_eq!(
889 + deferred::counts(&db.open().unwrap(), "").unwrap().total(),
890 + 0,
891 + "a held entry that lands is cleared"
892 + );
893 + }
894 +
895 + #[tokio::test]
896 + async fn a_parent_that_never_arrives_stops_being_retried_at_the_cap() {
897 + let dir = tempdir();
898 + let (db, node) = fk_device(&dir.join("fk_cap.db"), 12);
899 + let server = FakeServer::default();
900 +
901 + serve(
902 + &server,
903 + "child",
904 + "c1",
905 + serde_json::json!({"id":"c1","parent_id":"nope"}),
906 + );
907 +
908 + // Each pull spends one attempt. The first holds it; MAX_ATTEMPTS more
909 + // exhaust it. An empty page still runs the retry, which is the point.
910 + for _ in 0..=deferred::MAX_ATTEMPTS {
911 + pull_scope(&db, &server, &fk_schema(), node, SyncScope::Personal)
912 + .await
913 + .unwrap();
914 + }
915 +
916 + let conn = db.open().unwrap();
917 + let counts = deferred::counts(&conn, "").unwrap();
918 + assert_eq!(counts.deferred, 0, "no longer retried");
919 + assert_eq!(counts.rejected, 1, "but still visible, not discarded");
920 + let listed = deferred::list(&conn, "").unwrap();
921 + assert_eq!(listed[0].row_id, "c1");
922 + assert_eq!(listed[0].attempts, deferred::MAX_ATTEMPTS);
923 + }
924 +
925 + #[tokio::test]
926 + async fn a_deferred_entry_is_not_recorded_as_committed() {
927 + let dir = tempdir();
928 + let (db, node) = fk_device(&dir.join("fk_gate.db"), 13);
929 + let server = FakeServer::default();
930 +
931 + serve(
932 + &server,
933 + "child",
934 + "c1",
935 + serde_json::json!({"id":"c1","parent_id":"p1"}),
936 + );
937 + pull_scope(&db, &server, &fk_schema(), node, SyncScope::Personal)
938 + .await
939 + .unwrap();
940 +
941 + // Recording an unapplied row's HLC would gate its own retry out on the
942 + // next pull, since the gate drops anything not newer than what is
943 + // committed, and the hold would be a queue that never drains.
944 + assert!(
945 + super::super::hlc::committed_hlc(&db.open().unwrap(), "child", "c1")
946 + .unwrap()
947 + .is_none()
948 + );
949 + }
950 +
748 951 fn stamp_at(db: &DbSource, node: DeviceId, now_ms: i64) {
749 952 stamp_pending(&db.open().unwrap(), node, now_ms).unwrap();
750 953 }
A synckit-client/src/store/deferred.rs +500
@@ -1,0 +1,539 @@
1 + //! The dead-letter hold for remote changes an apply pass could not land.
2 + //!
3 + //! Before this existed, `apply` folded every unapplied row into one `skipped`
4 + //! counter, `sync` dropped that counter on the floor, and the pull cursor
5 + //! advanced regardless. The server never sends an entry twice, so the row was
6 + //! gone: device A had it, device B did not, and both reported a clean sync.
7 + //!
8 + //! The hold makes the cursor safe to advance. Anything the apply could not write
9 + //! is stored here **as it came off the wire**, not as the conflict layer resolved
10 + //! it, so a retry re-enters [`resolve_pull`](super::hlc::resolve_pull) and is
11 + //! judged against the local state of the moment rather than replaying a decision
12 + //! made against a database that has since moved.
13 + //!
14 + //! Two states, and the difference is whether trying again could ever help:
15 + //!
16 + //! - **Deferred** is retryable: an unknown table (clears when the client is
17 + //! upgraded) or a constraint violation (clears when the missing parent lands).
18 + //! Retried automatically at the top of every pull, up to [`MAX_ATTEMPTS`].
19 + //! - **Rejected** is not: a payload with no object, no reconstructable primary
20 + //! key, no insertable columns. Identical bytes fail identically, so it is held
21 + //! for display only. A deferred entry that exhausts its attempts is promoted
22 + //! here rather than retried forever.
23 + //!
24 + //! Rows the schema *intends* to drop (an `exclude_where` predicate,
25 + //! `DeleteMode::Ignore`) are not failures and never reach this table.
26 +
27 + use std::collections::HashSet;
28 +
29 + use rusqlite::{Connection, OptionalExtension};
30 +
31 + use super::apply::{ApplyOutcome, Unapplied};
32 + use crate::error::Result;
33 + use crate::ids::DeviceId;
34 + use crate::types::{ChangeEntry, PulledChange};
35 +
36 + /// Retries before a deferred entry is promoted to rejected.
37 + ///
38 + /// A constraint violation whose parent genuinely never arrives would otherwise
39 + /// be retried on every pull for the life of the install, and the held set would
40 + /// grow without bound. Five is enough for the ordering cases this exists for (a
41 + /// late parent lands on the next pull, not the fifth) and short enough that a
42 + /// permanently-broken entry stops costing work.
43 + pub const MAX_ATTEMPTS: i64 = 5;
44 +
45 + /// DDL for the hold, shared by [`SyncSchema::migration_sql`](super::schema::SyncSchema::migration_sql)
46 + /// and the per-connection upgrade in [`db::ensure_scope_schema`](super::db::ensure_scope_schema),
47 + /// so an existing install gets the table on its next connection open rather than
48 + /// waiting for the app to re-run its migration.
49 + ///
50 + /// One row per (scope, table, row_id): a later change to a row that is already
51 + /// held replaces the payload and keeps the attempt count, so a row that keeps
52 + /// failing does not accumulate copies.
53 + pub(crate) const DEFERRED_DDL: &str = "\
54 + CREATE TABLE IF NOT EXISTS sync_deferred (
55 + scope TEXT NOT NULL DEFAULT '',
56 + table_name TEXT NOT NULL,
57 + row_id TEXT NOT NULL,
58 + cause TEXT NOT NULL,
59 + state TEXT NOT NULL DEFAULT 'deferred',
60 + attempts INTEGER NOT NULL DEFAULT 0,
61 + seq INTEGER NOT NULL DEFAULT 0,
62 + device_id TEXT NOT NULL DEFAULT '',
63 + entry TEXT NOT NULL,
64 + first_seen TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
65 + last_seen TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
66 + PRIMARY KEY (scope, table_name, row_id)
67 + );
68 + ";
69 +
70 + /// What a held entry is waiting on.
71 + #[derive(Debug, Clone, Copy, PartialEq, Eq)]
72 + pub enum HoldState {
73 + /// Retryable: retried at the top of each pull until [`MAX_ATTEMPTS`].
74 + Deferred,
75 + /// Unretryable, or out of attempts. Held for display only.
76 + Rejected,
77 + }
78 +
79 + impl HoldState {
80 + fn as_str(self) -> &'static str {
81 + match self {
82 + Self::Deferred => "deferred",
83 + Self::Rejected => "rejected",
84 + }
85 + }
86 + }
87 +
88 + /// One entry in the hold, as listed for a human.
89 + #[derive(Debug, Clone, PartialEq, Eq)]
90 + pub struct HeldEntry {
91 + /// Table the change targets.
92 + pub table: String,
93 + /// Wire row id of the change.
94 + pub row_id: String,
95 + /// Why the apply could not land it.
96 + pub cause: String,
97 + /// Whether a retry could still help.
98 + pub state: HoldState,
99 + /// Retries spent so far.
100 + pub attempts: i64,
101 + /// When the entry was first held (RFC 3339, UTC).
102 + pub first_seen: String,
103 + }
104 +
105 + /// How much is being held for a scope, for a sync-status surface.
106 + #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
107 + pub struct HoldCounts {
108 + /// Entries still awaiting an automatic retry.
109 + pub deferred: u64,
110 + /// Entries that will not be retried.
111 + pub rejected: u64,
112 + }
113 +
114 + impl HoldCounts {
115 + /// Total held entries, whatever their state.
116 + pub fn total(self) -> u64 {
117 + self.deferred + self.rejected
118 + }
119 + }
120 +
121 + /// Identity of a held row: `(table, row_id)`. Enough to match an apply outcome
122 + /// against the batch it came from, since a resolved entry keeps both.
123 + pub(crate) type RowKey = (String, String);
124 +
125 + pub(crate) fn key_of(entry: &ChangeEntry) -> RowKey {
126 + (entry.table.clone(), entry.row_id.clone())
127 + }
128 +
129 + fn key_of_unapplied(row: &Unapplied) -> RowKey {
130 + (row.table.clone(), row.row_id.clone())
131 + }
132 +
133 + /// Load the entries eligible for an automatic retry, as they were pulled.
134 + ///
135 + /// Returned in `seq` order so a held batch keeps its original server ordering
136 + /// relative to itself; the caller puts them in front of the newly pulled batch.
137 + pub fn load_retryable(conn: &Connection, scope: &str) -> Result<Vec<PulledChange>> {
138 + let mut stmt = conn.prepare(
139 + "SELECT entry, device_id, seq FROM sync_deferred \
140 + WHERE scope = ?1 AND state = 'deferred' AND attempts < ?2 \
141 + ORDER BY seq",
142 + )?;
143 + let rows = stmt.query_map(rusqlite::params![scope, MAX_ATTEMPTS], |r| {
144 + Ok((
145 + r.get::<_, String>(0)?,
146 + r.get::<_, String>(1)?,
147 + r.get::<_, i64>(2)?,
148 + ))
149 + })?;
150 +
151 + let mut out = Vec::new();
152 + for row in rows {
153 + let (entry_json, device, seq) = row?;
154 + // A payload that will not deserialize is a held entry we can never retry
155 + // (a downgrade, or a corrupted row). Skip it rather than failing the pull;
156 + // it stays in the table and stays visible.
157 + let Ok(entry) = serde_json::from_str::<ChangeEntry>(&entry_json) else {
158 + tracing::warn!("held entry could not be deserialized, skipping retry");
159 + continue;
160 + };
161 + // The originating device only feeds conflict detection ("did this come
162 + // from me"); an unparseable one degrades to nil, which is never a live
163 + // device, rather than failing the retry.
164 + let device_id =
165 + uuid::Uuid::parse_str(&device).map_or_else(|_| DeviceId::nil(), DeviceId::new);
166 + out.push(PulledChange {
167 + entry,
168 + device_id,
169 + seq,
170 + });
171 + }
172 + Ok(out)
173 + }
174 +
175 + /// List everything held for a scope, newest first, for a UI surface.
176 + pub fn list(conn: &Connection, scope: &str) -> Result<Vec<HeldEntry>> {
177 + let mut stmt = conn.prepare(
178 + "SELECT table_name, row_id, cause, state, attempts, first_seen \
179 + FROM sync_deferred WHERE scope = ?1 ORDER BY last_seen DESC",
180 + )?;
181 + let rows = stmt.query_map([scope], |r| {
182 + let state: String = r.get(3)?;
183 + Ok(HeldEntry {
184 + table: r.get(0)?,
185 + row_id: r.get(1)?,
186 + cause: r.get(2)?,
187 + state: if state == "rejected" {
188 + HoldState::Rejected
189 + } else {
190 + HoldState::Deferred
191 + },
192 + attempts: r.get(4)?,
193 + first_seen: r.get(5)?,
194 + })
195 + })?;
196 + let mut out = Vec::new();
197 + for row in rows {
198 + out.push(row?);
199 + }
200 + Ok(out)
201 + }
202 +
203 + /// Held counts for a scope.
204 + pub fn counts(conn: &Connection, scope: &str) -> Result<HoldCounts> {
205 + let mut out = HoldCounts::default();
206 + let mut stmt =
207 + conn.prepare("SELECT state, COUNT(*) FROM sync_deferred WHERE scope = ?1 GROUP BY state")?;
208 + let rows = stmt.query_map([scope], |r| {
209 + Ok((r.get::<_, String>(0)?, r.get::<_, i64>(1)?))
210 + })?;
211 + for row in rows {
212 + let (state, n) = row?;
213 + let n = u64::try_from(n).unwrap_or(0);
214 + match state.as_str() {
215 + "rejected" => out.rejected += n,
216 + _ => out.deferred += n,
217 + }
218 + }
219 + Ok(out)
220 + }
221 +
222 + /// Clear one held entry, by row identity. Returns whether a row was removed.
223 + ///
224 + /// The retry path calls this when an entry finally lands; a UI can call it to
225 + /// discard something the user has decided to abandon.
226 + pub fn clear(conn: &Connection, scope: &str, table: &str, row_id: &str) -> Result<bool> {
227 + let n = conn.execute(
228 + "DELETE FROM sync_deferred WHERE scope = ?1 AND table_name = ?2 AND row_id = ?3",
229 + rusqlite::params![scope, table, row_id],
230 + )?;
231 + Ok(n > 0)
232 + }
233 +
234 + /// Reset a rejected entry so the next pull retries it once more.
235 + ///
236 + /// This is the per-row Retry affordance: a rejected entry is not retried
237 + /// automatically, but a human who has fixed the cause (upgraded the client,
238 + /// restored the missing parent by hand) can put it back in the queue.
239 + pub fn requeue(conn: &Connection, scope: &str, table: &str, row_id: &str) -> Result<bool> {
240 + let n = conn.execute(
241 + "UPDATE sync_deferred SET state = 'deferred', attempts = 0 \
242 + WHERE scope = ?1 AND table_name = ?2 AND row_id = ?3",
243 + rusqlite::params![scope, table, row_id],
244 + )?;
245 + Ok(n > 0)
246 + }
247 +
248 + /// Reconcile the hold with the outcome of an apply pass.
249 + ///
250 + /// `batch` is every entry the pass was given, keyed by row, as pulled. `retried`
251 + /// is the subset that came out of the hold. For each row:
252 + ///
253 + /// - unapplied again, and it was retried: spend an attempt, promoting to rejected
254 + /// at [`MAX_ATTEMPTS`];
255 + /// - unapplied for the first time: hold it;
256 + /// - retried and no longer unapplied: it landed (or the schema filtered it), so
257 + /// clear it.
258 + pub(crate) fn settle(
259 + conn: &Connection,
260 + scope: &str,
261 + outcome: &ApplyOutcome,
262 + batch: &std::collections::HashMap<RowKey, PulledChange>,
263 + retried: &HashSet<RowKey>,
264 + ) -> Result<()> {
265 + let mut still_failing: HashSet<RowKey> = HashSet::new();
266 +
267 + for row in &outcome.deferred {
268 + let key = key_of_unapplied(row);
269 + still_failing.insert(key.clone());
270 + if retried.contains(&key) {
271 + spend_attempt(conn, scope, row)?;
272 + } else {
273 + hold(conn, scope, row, HoldState::Deferred, batch.get(&key))?;
274 + }
275 + }
276 +
277 + for row in &outcome.rejected {
278 + let key = key_of_unapplied(row);
279 + still_failing.insert(key.clone());
280 + // A reject never earns another automatic attempt, whether it is new or a
281 + // retry that failed the same way again.
282 + hold(conn, scope, row, HoldState::Rejected, batch.get(&key))?;
283 + }
284 +
285 + for key in retried {
286 + if !still_failing.contains(key) {
287 + clear(conn, scope, &key.0, &key.1)?;
288 + }
289 + }
290 + Ok(())
291 + }
292 +
293 + /// Insert or refresh a held row.
294 + fn hold(
295 + conn: &Connection,
296 + scope: &str,
297 + row: &Unapplied,
298 + state: HoldState,
299 + pulled: Option<&PulledChange>,
300 + ) -> Result<()> {
301 + let Some(pulled) = pulled else {
302 + // Every unapplied row is one the caller handed us, so this cannot happen
303 + // in the pull path. Refuse to hold a row with no payload rather than
304 + // write an entry no retry could ever use.
305 + tracing::warn!(
306 + table = %row.table,
307 + row_id = %row.row_id,
308 + "unapplied row has no pulled entry to hold; not recorded"
309 + );
310 + return Ok(());
311 + };
312 + let entry = serde_json::to_string(&pulled.entry)?;
313 + conn.execute(
314 + "INSERT INTO sync_deferred \
315 + (scope, table_name, row_id, cause, state, seq, device_id, entry) \
316 + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8) \
317 + ON CONFLICT(scope, table_name, row_id) DO UPDATE SET \
318 + cause = excluded.cause, \
319 + state = excluded.state, \
320 + seq = excluded.seq, \
321 + device_id = excluded.device_id, \
322 + entry = excluded.entry, \
323 + last_seen = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')",
324 + rusqlite::params![
325 + scope,
326 + row.table,
327 + row.row_id,
328 + row.cause,
329 + state.as_str(),
330 + pulled.seq,
331 + pulled.device_id.to_string(),
332 + entry,
333 + ],
334 + )?;
335 + Ok(())
336 + }
337 +
338 + /// Spend one of a held entry's attempts, promoting it to rejected at the cap.
339 + fn spend_attempt(conn: &Connection, scope: &str, row: &Unapplied) -> Result<()> {
340 + conn.execute(
341 + "UPDATE sync_deferred SET \
342 + attempts = attempts + 1, \
343 + cause = ?4, \
344 + state = CASE WHEN attempts + 1 >= ?5 THEN 'rejected' ELSE state END, \
345 + last_seen = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') \
346 + WHERE scope = ?1 AND table_name = ?2 AND row_id = ?3",
347 + rusqlite::params![scope, row.table, row.row_id, row.cause, MAX_ATTEMPTS],
348 + )?;
349 + Ok(())
350 + }
351 +
352 + /// Whether a scope holds anything at all, the cheap check the pull loop uses to
353 + /// decide if an empty batch is worth a retry pass.
354 + pub(crate) fn has_retryable(conn: &Connection, scope: &str) -> Result<bool> {
355 + let found = conn
356 + .query_row(
357 + "SELECT 1 FROM sync_deferred \
358 + WHERE scope = ?1 AND state = 'deferred' AND attempts < ?2 LIMIT 1",
359 + rusqlite::params![scope, MAX_ATTEMPTS],
360 + |_| Ok(()),
361 + )
362 + .optional()?;
363 + Ok(found.is_some())
364 + }
365 +
366 + #[cfg(test)]
367 + mod tests {
368 + use super::*;
369 + use crate::types::{ChangeOp, hlc_legacy_floor};
370 + use std::collections::HashMap;
371 +
372 + fn db() -> Connection {
373 + let conn = Connection::open_in_memory().unwrap();
374 + conn.execute_batch(DEFERRED_DDL).unwrap();
375 + conn
376 + }
377 +
378 + fn pulled(table: &str, row_id: &str, seq: i64) -> PulledChange {
379 + PulledChange {
380 + entry: ChangeEntry {
381 + table: table.into(),
382 + op: ChangeOp::Insert,
383 + row_id: row_id.into(),
384 + timestamp: chrono::Utc::now(),
385 + hlc: hlc_legacy_floor(),
386 + data: Some(serde_json::json!({"id": row_id})),
387 + extra: serde_json::Map::default(),
388 + },
389 + device_id: DeviceId::nil(),
390 + seq,
391 + }
392 + }
393 +
394 + fn unapplied(table: &str, row_id: &str) -> Unapplied {
395 + Unapplied {
396 + table: table.into(),
397 + row_id: row_id.into(),
398 + cause: "constraint violation".into(),
399 + }
400 + }
401 +
402 + fn batch(entries: &[PulledChange]) -> HashMap<RowKey, PulledChange> {
403 + entries
404 + .iter()
405 + .map(|p| (key_of(&p.entry), p.clone()))
406 + .collect()
407 + }
408 +
409 + fn deferred_outcome(rows: Vec<Unapplied>) -> ApplyOutcome {
410 + ApplyOutcome {
411 + deferred: rows,
412 + ..ApplyOutcome::default()
413 + }
414 + }
415 +
416 + #[test]
417 + fn a_new_deferred_row_is_held_with_its_pulled_entry() {
418 + let conn = db();
419 + let p = pulled("child", "c1", 7);
420 + settle(
421 + &conn,
422 + "",
423 + &deferred_outcome(vec![unapplied("child", "c1")]),
424 + &batch(&[p]),
425 + &HashSet::new(),
426 + )
427 + .unwrap();
428 +
429 + let held = load_retryable(&conn, "").unwrap();
430 + assert_eq!(held.len(), 1);
431 + assert_eq!(held[0].entry.row_id, "c1");
432 + assert_eq!(held[0].seq, 7, "the wire seq is preserved for ordering");
433 + assert_eq!(counts(&conn, "").unwrap().deferred, 1);
434 + }
435 +
436 + #[test]
437 + fn a_retry_that_lands_clears_the_hold() {
438 + let conn = db();
439 + let p = pulled("child", "c1", 7);
440 + let b = batch(&[p]);
441 + settle(
442 + &conn,
443 + "",
444 + &deferred_outcome(vec![unapplied("child", "c1")]),
445 + &b,
446 + &HashSet::new(),
447 + )
448 + .unwrap();
449 +
450 + let retried: HashSet<RowKey> = [("child".to_string(), "c1".to_string())]
451 + .into_iter()
452 + .collect();
453 + settle(&conn, "", &ApplyOutcome::default(), &b, &retried).unwrap();
454 +
455 + assert!(load_retryable(&conn, "").unwrap().is_empty());
456 + assert_eq!(counts(&conn, "").unwrap().total(), 0);
457 + }
458 +
459 + #[test]
460 + fn attempts_are_capped_and_the_entry_is_promoted_to_rejected() {
461 + let conn = db();
462 + let p = pulled("child", "c1", 7);
463 + let b = batch(&[p]);
464 + let outcome = deferred_outcome(vec![unapplied("child", "c1")]);
465 + settle(&conn, "", &outcome, &b, &HashSet::new()).unwrap();
466 +
467 + let retried: HashSet<RowKey> = [("child".to_string(), "c1".to_string())]
468 + .into_iter()
469 + .collect();
470 + for _ in 0..MAX_ATTEMPTS {
471 + settle(&conn, "", &outcome, &b, &retried).unwrap();
472 + }
473 +
474 + assert!(
475 + load_retryable(&conn, "").unwrap().is_empty(),
476 + "a capped entry is no longer retried"
477 + );
478 + let listed = list(&conn, "").unwrap();
479 + assert_eq!(listed.len(), 1, "but it stays visible");
480 + assert_eq!(listed[0].state, HoldState::Rejected);
481 + assert_eq!(listed[0].attempts, MAX_ATTEMPTS);
482 + assert!(!has_retryable(&conn, "").unwrap());
483 + }
484 +
485 + #[test]
486 + fn requeue_gives_a_rejected_entry_one_more_run() {
487 + let conn = db();
488 + let p = pulled("child", "c1", 7);
489 + settle(
490 + &conn,
491 + "",
492 + &ApplyOutcome {
493 + rejected: vec![unapplied("child", "c1")],
494 + ..ApplyOutcome::default()
495 + },
496 + &batch(&[p]),
497 + &HashSet::new(),
498 + )
499 + .unwrap();
500 + assert!(load_retryable(&conn, "").unwrap().is_empty());
Lines truncated