Skip to main content

max / synckit

25.1 KB · 723 lines History Blame Raw
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 /// Scope the change belongs to: `""` for personal, otherwise the group id.
92 pub scope: String,
93 /// Table the change targets.
94 pub table: String,
95 /// Wire row id of the change.
96 pub row_id: String,
97 /// Why the apply could not land it.
98 pub cause: String,
99 /// Whether a retry could still help.
100 pub state: HoldState,
101 /// Retries spent so far.
102 pub attempts: i64,
103 /// When the entry was first held (RFC 3339, UTC).
104 pub first_seen: String,
105 /// The held change's row payload, so a consumer can name the row in terms
106 /// its user recognises. `None` for a delete, or a payload that no longer
107 /// parses.
108 pub payload: Option<serde_json::Value>,
109 }
110
111 /// How much is being held for a scope, for a sync-status surface.
112 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
113 pub struct HoldCounts {
114 /// Entries still awaiting an automatic retry.
115 pub deferred: u64,
116 /// Entries that will not be retried.
117 pub rejected: u64,
118 }
119
120 impl HoldCounts {
121 /// Total held entries, whatever their state.
122 pub fn total(self) -> u64 {
123 self.deferred + self.rejected
124 }
125 }
126
127 /// Identity of a held row: `(table, row_id)`. Enough to match an apply outcome
128 /// against the batch it came from, since a resolved entry keeps both.
129 pub(crate) type RowKey = (String, String);
130
131 pub(crate) fn key_of(entry: &ChangeEntry) -> RowKey {
132 (entry.table.clone(), entry.row_id.clone())
133 }
134
135 fn key_of_unapplied(row: &Unapplied) -> RowKey {
136 (row.table.clone(), row.row_id.clone())
137 }
138
139 /// Load the entries eligible for an automatic retry, as they were pulled.
140 ///
141 /// Returned in `seq` order so a held batch keeps its original server ordering
142 /// relative to itself; the caller puts them in front of the newly pulled batch.
143 pub fn load_retryable(conn: &Connection, scope: &str) -> Result<Vec<PulledChange>> {
144 let mut stmt = conn.prepare(
145 "SELECT entry, device_id, seq FROM sync_deferred \
146 WHERE scope = ?1 AND state = 'deferred' AND attempts < ?2 \
147 ORDER BY seq",
148 )?;
149 let rows = stmt.query_map(rusqlite::params![scope, MAX_ATTEMPTS], |r| {
150 Ok((
151 r.get::<_, String>(0)?,
152 r.get::<_, String>(1)?,
153 r.get::<_, i64>(2)?,
154 ))
155 })?;
156
157 let mut out = Vec::new();
158 for row in rows {
159 let (entry_json, device, seq) = row?;
160 // A payload that will not deserialize is a held entry we can never retry
161 // (a downgrade, or a corrupted row). Skip it rather than failing the pull;
162 // it stays in the table and stays visible.
163 let Ok(entry) = serde_json::from_str::<ChangeEntry>(&entry_json) else {
164 tracing::warn!("held entry could not be deserialized, skipping retry");
165 continue;
166 };
167 // The originating device only feeds conflict detection ("did this come
168 // from me"); an unparseable one degrades to nil, which is never a live
169 // device, rather than failing the retry.
170 let device_id =
171 uuid::Uuid::parse_str(&device).map_or_else(|_| DeviceId::nil(), DeviceId::new);
172 out.push(PulledChange {
173 entry,
174 device_id,
175 seq,
176 // A held entry already passed the peer storage gate on the pull that
177 // held it, and the hold stores the change rather than the envelope, so
178 // there is no stamp to re-read and nothing left to re-check.
179 storage_version: None,
180 });
181 }
182 Ok(out)
183 }
184
185 const LIST_COLUMNS: &str =
186 "scope, table_name, row_id, cause, state, attempts, first_seen, entry FROM sync_deferred";
187
188 fn read_held(r: &rusqlite::Row<'_>) -> rusqlite::Result<HeldEntry> {
189 let state: String = r.get(4)?;
190 Ok(HeldEntry {
191 scope: r.get(0)?,
192 table: r.get(1)?,
193 row_id: r.get(2)?,
194 cause: r.get(3)?,
195 state: if state == "rejected" {
196 HoldState::Rejected
197 } else {
198 HoldState::Deferred
199 },
200 attempts: r.get(5)?,
201 first_seen: r.get(6)?,
202 // The payload is handed back so a consumer can label the row with
203 // something a person recognises (a task's title) instead of a wire row
204 // id. A payload that will not parse degrades to None, never an error.
205 payload: serde_json::from_str::<ChangeEntry>(&r.get::<_, String>(7)?)
206 .ok()
207 .and_then(|e| e.data),
208 })
209 }
210
211 /// List everything held for a scope, newest first, for a UI surface.
212 pub fn list(conn: &Connection, scope: &str) -> Result<Vec<HeldEntry>> {
213 let mut stmt = conn.prepare(&format!(
214 "SELECT {LIST_COLUMNS} WHERE scope = ?1 ORDER BY last_seen DESC"
215 ))?;
216 collect(stmt.query_map([scope], read_held)?)
217 }
218
219 /// List everything held across every scope, newest first.
220 ///
221 /// A device syncs its personal scope plus one per group, and a row held in a
222 /// group scope is just as lost as one held in the personal scope, so a status
223 /// surface that covered only personal would hide the failures this exists to
224 /// show.
225 pub fn list_all(conn: &Connection) -> Result<Vec<HeldEntry>> {
226 let mut stmt = conn.prepare(&format!("SELECT {LIST_COLUMNS} ORDER BY last_seen DESC"))?;
227 collect(stmt.query_map([], read_held)?)
228 }
229
230 fn collect<I>(rows: I) -> Result<Vec<HeldEntry>>
231 where
232 I: Iterator<Item = rusqlite::Result<HeldEntry>>,
233 {
234 let mut out = Vec::new();
235 for row in rows {
236 out.push(row?);
237 }
238 Ok(out)
239 }
240
241 fn read_tally(r: &rusqlite::Row<'_>) -> rusqlite::Result<(String, i64)> {
242 Ok((r.get(0)?, r.get(1)?))
243 }
244
245 /// Held counts for a scope.
246 pub fn counts(conn: &Connection, scope: &str) -> Result<HoldCounts> {
247 let mut stmt =
248 conn.prepare("SELECT state, COUNT(*) FROM sync_deferred WHERE scope = ?1 GROUP BY state")?;
249 tally(stmt.query_map([scope], read_tally)?)
250 }
251
252 /// Held counts across every scope. See [`list_all`].
253 pub fn counts_all(conn: &Connection) -> Result<HoldCounts> {
254 let mut stmt = conn.prepare("SELECT state, COUNT(*) FROM sync_deferred GROUP BY state")?;
255 tally(stmt.query_map([], read_tally)?)
256 }
257
258 fn tally<I>(rows: I) -> Result<HoldCounts>
259 where
260 I: Iterator<Item = rusqlite::Result<(String, i64)>>,
261 {
262 let mut out = HoldCounts::default();
263 for row in rows {
264 let (state, n) = row?;
265 let n = u64::try_from(n).unwrap_or(0);
266 match state.as_str() {
267 "rejected" => out.rejected += n,
268 _ => out.deferred += n,
269 }
270 }
271 Ok(out)
272 }
273
274 /// Clear one held entry, by row identity. Returns whether a row was removed.
275 ///
276 /// The retry path calls this when an entry finally lands; a UI can call it to
277 /// discard something the user has decided to abandon.
278 pub fn clear(conn: &Connection, scope: &str, table: &str, row_id: &str) -> Result<bool> {
279 let n = conn.execute(
280 "DELETE FROM sync_deferred WHERE scope = ?1 AND table_name = ?2 AND row_id = ?3",
281 rusqlite::params![scope, table, row_id],
282 )?;
283 Ok(n > 0)
284 }
285
286 /// Reset a rejected entry so the next pull retries it once more.
287 ///
288 /// This is the per-row Retry affordance: a rejected entry is not retried
289 /// automatically, but a human who has fixed the cause (upgraded the client,
290 /// restored the missing parent by hand) can put it back in the queue.
291 pub fn requeue(conn: &Connection, scope: &str, table: &str, row_id: &str) -> Result<bool> {
292 let n = conn.execute(
293 "UPDATE sync_deferred SET state = 'deferred', attempts = 0 \
294 WHERE scope = ?1 AND table_name = ?2 AND row_id = ?3",
295 rusqlite::params![scope, table, row_id],
296 )?;
297 Ok(n > 0)
298 }
299
300 /// Reconcile the hold with the outcome of an apply pass.
301 ///
302 /// `batch` is every entry the pass was given, keyed by row, as pulled. `retried`
303 /// is the subset that came out of the hold. For each row:
304 ///
305 /// - unapplied again, and it was retried: spend an attempt, promoting to rejected
306 /// at [`MAX_ATTEMPTS`];
307 /// - unapplied for the first time: hold it;
308 /// - retried and no longer unapplied: it landed (or the schema filtered it), so
309 /// clear it.
310 pub(crate) fn settle(
311 conn: &Connection,
312 scope: &str,
313 outcome: &ApplyOutcome,
314 batch: &std::collections::HashMap<RowKey, PulledChange>,
315 retried: &HashSet<RowKey>,
316 ) -> Result<()> {
317 let mut still_failing: HashSet<RowKey> = HashSet::new();
318
319 for row in &outcome.deferred {
320 let key = key_of_unapplied(row);
321 still_failing.insert(key.clone());
322 if retried.contains(&key) {
323 spend_attempt(conn, scope, row)?;
324 } else {
325 hold(conn, scope, row, HoldState::Deferred, batch.get(&key))?;
326 }
327 }
328
329 for row in &outcome.rejected {
330 let key = key_of_unapplied(row);
331 still_failing.insert(key.clone());
332 // A reject never earns another automatic attempt, whether it is new or a
333 // retry that failed the same way again.
334 hold(conn, scope, row, HoldState::Rejected, batch.get(&key))?;
335 }
336
337 for key in retried {
338 if !still_failing.contains(key) {
339 clear(conn, scope, &key.0, &key.1)?;
340 }
341 }
342 Ok(())
343 }
344
345 /// Insert or refresh a held row.
346 fn hold(
347 conn: &Connection,
348 scope: &str,
349 row: &Unapplied,
350 state: HoldState,
351 pulled: Option<&PulledChange>,
352 ) -> Result<()> {
353 let Some(pulled) = pulled else {
354 // Every unapplied row is one the caller handed us, so this cannot happen
355 // in the pull path. Refuse to hold a row with no payload rather than
356 // write an entry no retry could ever use.
357 tracing::warn!(
358 table = %row.table,
359 row_id = %row.row_id,
360 "unapplied row has no pulled entry to hold; not recorded"
361 );
362 return Ok(());
363 };
364 let entry = serde_json::to_string(&pulled.entry)?;
365 conn.execute(
366 "INSERT INTO sync_deferred \
367 (scope, table_name, row_id, cause, state, seq, device_id, entry) \
368 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8) \
369 ON CONFLICT(scope, table_name, row_id) DO UPDATE SET \
370 cause = excluded.cause, \
371 state = excluded.state, \
372 seq = excluded.seq, \
373 device_id = excluded.device_id, \
374 entry = excluded.entry, \
375 last_seen = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')",
376 rusqlite::params![
377 scope,
378 row.table,
379 row.row_id,
380 row.cause,
381 state.as_str(),
382 pulled.seq,
383 pulled.device_id.to_string(),
384 entry,
385 ],
386 )?;
387 Ok(())
388 }
389
390 /// Spend one of a held entry's attempts, promoting it to rejected at the cap.
391 fn spend_attempt(conn: &Connection, scope: &str, row: &Unapplied) -> Result<()> {
392 conn.execute(
393 "UPDATE sync_deferred SET \
394 attempts = attempts + 1, \
395 cause = ?4, \
396 state = CASE WHEN attempts + 1 >= ?5 THEN 'rejected' ELSE state END, \
397 last_seen = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') \
398 WHERE scope = ?1 AND table_name = ?2 AND row_id = ?3",
399 rusqlite::params![scope, row.table, row.row_id, row.cause, MAX_ATTEMPTS],
400 )?;
401 Ok(())
402 }
403
404 /// Whether a scope holds anything at all, the cheap check the pull loop uses to
405 /// decide if an empty batch is worth a retry pass.
406 pub(crate) fn has_retryable(conn: &Connection, scope: &str) -> Result<bool> {
407 let found = conn
408 .query_row(
409 "SELECT 1 FROM sync_deferred \
410 WHERE scope = ?1 AND state = 'deferred' AND attempts < ?2 LIMIT 1",
411 rusqlite::params![scope, MAX_ATTEMPTS],
412 |_| Ok(()),
413 )
414 .optional()?;
415 Ok(found.is_some())
416 }
417
418 #[cfg(test)]
419 mod tests {
420 use super::*;
421 use crate::types::{ChangeOp, hlc_legacy_floor};
422 use std::collections::HashMap;
423
424 fn db() -> Connection {
425 let conn = Connection::open_in_memory().unwrap();
426 conn.execute_batch(DEFERRED_DDL).unwrap();
427 conn
428 }
429
430 fn pulled(table: &str, row_id: &str, seq: i64) -> PulledChange {
431 PulledChange {
432 storage_version: None,
433 entry: ChangeEntry {
434 table: table.into(),
435 op: ChangeOp::Insert,
436 row_id: row_id.into(),
437 timestamp: chrono::Utc::now(),
438 hlc: hlc_legacy_floor(),
439 data: Some(serde_json::json!({"id": row_id})),
440 extra: serde_json::Map::default(),
441 },
442 device_id: DeviceId::nil(),
443 seq,
444 }
445 }
446
447 fn unapplied(table: &str, row_id: &str) -> Unapplied {
448 Unapplied {
449 table: table.into(),
450 row_id: row_id.into(),
451 cause: "constraint violation".into(),
452 }
453 }
454
455 fn batch(entries: &[PulledChange]) -> HashMap<RowKey, PulledChange> {
456 entries
457 .iter()
458 .map(|p| (key_of(&p.entry), p.clone()))
459 .collect()
460 }
461
462 fn deferred_outcome(rows: Vec<Unapplied>) -> ApplyOutcome {
463 ApplyOutcome {
464 deferred: rows,
465 ..ApplyOutcome::default()
466 }
467 }
468
469 #[test]
470 fn a_new_deferred_row_is_held_with_its_pulled_entry() {
471 let conn = db();
472 let p = pulled("child", "c1", 7);
473 settle(
474 &conn,
475 "",
476 &deferred_outcome(vec![unapplied("child", "c1")]),
477 &batch(&[p]),
478 &HashSet::new(),
479 )
480 .unwrap();
481
482 let held = load_retryable(&conn, "").unwrap();
483 assert_eq!(held.len(), 1);
484 assert_eq!(held[0].entry.row_id, "c1");
485 assert_eq!(held[0].seq, 7, "the wire seq is preserved for ordering");
486 assert_eq!(counts(&conn, "").unwrap().deferred, 1);
487 }
488
489 #[test]
490 fn a_retry_that_lands_clears_the_hold() {
491 let conn = db();
492 let p = pulled("child", "c1", 7);
493 let b = batch(&[p]);
494 settle(
495 &conn,
496 "",
497 &deferred_outcome(vec![unapplied("child", "c1")]),
498 &b,
499 &HashSet::new(),
500 )
501 .unwrap();
502
503 let retried: HashSet<RowKey> = [("child".to_string(), "c1".to_string())]
504 .into_iter()
505 .collect();
506 settle(&conn, "", &ApplyOutcome::default(), &b, &retried).unwrap();
507
508 assert!(load_retryable(&conn, "").unwrap().is_empty());
509 assert_eq!(counts(&conn, "").unwrap().total(), 0);
510 }
511
512 #[test]
513 fn attempts_are_capped_and_the_entry_is_promoted_to_rejected() {
514 let conn = db();
515 let p = pulled("child", "c1", 7);
516 let b = batch(&[p]);
517 let outcome = deferred_outcome(vec![unapplied("child", "c1")]);
518 settle(&conn, "", &outcome, &b, &HashSet::new()).unwrap();
519
520 let retried: HashSet<RowKey> = [("child".to_string(), "c1".to_string())]
521 .into_iter()
522 .collect();
523 for _ in 0..MAX_ATTEMPTS {
524 settle(&conn, "", &outcome, &b, &retried).unwrap();
525 }
526
527 assert!(
528 load_retryable(&conn, "").unwrap().is_empty(),
529 "a capped entry is no longer retried"
530 );
531 let listed = list(&conn, "").unwrap();
532 assert_eq!(listed.len(), 1, "but it stays visible");
533 assert_eq!(listed[0].state, HoldState::Rejected);
534 assert_eq!(listed[0].attempts, MAX_ATTEMPTS);
535 assert!(!has_retryable(&conn, "").unwrap());
536 }
537
538 #[test]
539 fn requeue_gives_a_rejected_entry_one_more_run() {
540 let conn = db();
541 let p = pulled("child", "c1", 7);
542 settle(
543 &conn,
544 "",
545 &ApplyOutcome {
546 rejected: vec![unapplied("child", "c1")],
547 ..ApplyOutcome::default()
548 },
549 &batch(&[p]),
550 &HashSet::new(),
551 )
552 .unwrap();
553 assert!(load_retryable(&conn, "").unwrap().is_empty());
554
555 assert!(requeue(&conn, "", "child", "c1").unwrap());
556 assert_eq!(load_retryable(&conn, "").unwrap().len(), 1);
557 }
558
559 #[test]
560 fn holds_are_partitioned_by_scope() {
561 let conn = db();
562 let p = pulled("child", "c1", 7);
563 let b = batch(&[p]);
564 let outcome = deferred_outcome(vec![unapplied("child", "c1")]);
565 settle(&conn, "", &outcome, &b, &HashSet::new()).unwrap();
566 settle(&conn, "group-a", &outcome, &b, &HashSet::new()).unwrap();
567
568 assert_eq!(counts(&conn, "").unwrap().deferred, 1);
569 assert_eq!(counts(&conn, "group-a").unwrap().deferred, 1);
570 assert!(clear(&conn, "", "child", "c1").unwrap());
571 assert_eq!(counts(&conn, "").unwrap().deferred, 0);
572 assert_eq!(
573 counts(&conn, "group-a").unwrap().deferred,
574 1,
575 "clearing one scope leaves the other alone"
576 );
577 }
578
579 /// An outcome holding both kinds at once, for the count arithmetic.
580 fn mixed_outcome(deferred: Vec<Unapplied>, rejected: Vec<Unapplied>) -> ApplyOutcome {
581 ApplyOutcome {
582 rejected,
583 deferred,
584 ..ApplyOutcome::default()
585 }
586 }
587
588 #[test]
589 fn hold_counts_total_adds_the_two_states() {
590 let conn = db();
591 // Two of one and three of the other: any arithmetic other than a sum
592 // lands somewhere else (a difference underflows, a product gives six).
593 let entries: Vec<PulledChange> = (0..5)
594 .map(|i| pulled("child", &format!("c{i}"), i))
595 .collect();
596 settle(
597 &conn,
598 "",
599 &mixed_outcome(
600 vec![unapplied("child", "c0"), unapplied("child", "c1")],
601 vec![
602 unapplied("child", "c2"),
603 unapplied("child", "c3"),
604 unapplied("child", "c4"),
605 ],
606 ),
607 &batch(&entries),
608 &HashSet::new(),
609 )
610 .unwrap();
611
612 let counts = counts(&conn, "").unwrap();
613 assert_eq!(counts.deferred, 2);
614 assert_eq!(counts.rejected, 3);
615 assert_eq!(counts.total(), 5);
616 assert_eq!(list(&conn, "").unwrap().len(), 5);
617 }
618
619 #[test]
620 fn the_all_scope_reads_see_every_scope_at_once() {
621 // A row held in a group scope is as lost as one held in the personal
622 // scope, so the status surface reads across both. Different counts per
623 // scope, so neither read can pass by looking at one of them twice.
624 let conn = db();
625 let personal = pulled("child", "p1", 1);
626 let group = [pulled("child", "g1", 2), pulled("child", "g2", 3)];
627 settle(
628 &conn,
629 "",
630 &deferred_outcome(vec![unapplied("child", "p1")]),
631 &batch(&[personal]),
632 &HashSet::new(),
633 )
634 .unwrap();
635 settle(
636 &conn,
637 "group-a",
638 &mixed_outcome(
639 vec![unapplied("child", "g1")],
640 vec![unapplied("child", "g2")],
641 ),
642 &batch(&group),
643 &HashSet::new(),
644 )
645 .unwrap();
646
647 let all = counts_all(&conn).unwrap();
648 assert_eq!(all.deferred, 2, "the personal one plus the group's");
649 assert_eq!(all.rejected, 1);
650 assert_eq!(all.total(), 3);
651
652 let listed = list_all(&conn).unwrap();
653 assert_eq!(listed.len(), 3);
654 let mut scopes: Vec<&str> = listed.iter().map(|e| e.scope.as_str()).collect();
655 scopes.sort_unstable();
656 scopes.dedup();
657 assert_eq!(scopes, ["", "group-a"], "both scopes are represented");
658
659 // The per-scope reads still see only their own, which is what makes the
660 // pair of reads worth having.
661 assert_eq!(counts(&conn, "").unwrap().total(), 1);
662 assert_eq!(list(&conn, "group-a").unwrap().len(), 2);
663 }
664
665 #[test]
666 fn clear_and_requeue_report_whether_the_row_was_there() {
667 // Both return the rows-affected of their statement, and a caller uses it
668 // to tell "retried" from "there was nothing to retry".
669 let conn = db();
670 settle(
671 &conn,
672 "",
673 &ApplyOutcome {
674 rejected: vec![unapplied("child", "c1")],
675 ..ApplyOutcome::default()
676 },
677 &batch(&[pulled("child", "c1", 7)]),
678 &HashSet::new(),
679 )
680 .unwrap();
681
682 assert!(
683 !requeue(&conn, "", "child", "absent").unwrap(),
684 "no such row id"
685 );
686 assert!(
687 !requeue(&conn, "", "other-scope", "c1").unwrap(),
688 "no such table"
689 );
690 assert!(
691 !clear(&conn, "elsewhere", "child", "c1").unwrap(),
692 "no such scope"
693 );
694 assert_eq!(
695 counts(&conn, "").unwrap().total(),
696 1,
697 "none of that touched the held row"
698 );
699
700 assert!(requeue(&conn, "", "child", "c1").unwrap());
701 assert!(clear(&conn, "", "child", "c1").unwrap());
702 assert!(
703 !clear(&conn, "", "child", "c1").unwrap(),
704 "the second clear finds nothing left"
705 );
706 assert_eq!(counts(&conn, "").unwrap().total(), 0);
707 }
708
709 #[test]
710 fn a_repeat_failure_replaces_the_payload_without_duplicating_the_row() {
711 let conn = db();
712 let first = pulled("child", "c1", 7);
713 let second = pulled("child", "c1", 9);
714 let outcome = deferred_outcome(vec![unapplied("child", "c1")]);
715 settle(&conn, "", &outcome, &batch(&[first]), &HashSet::new()).unwrap();
716 settle(&conn, "", &outcome, &batch(&[second]), &HashSet::new()).unwrap();
717
718 let held = load_retryable(&conn, "").unwrap();
719 assert_eq!(held.len(), 1);
720 assert_eq!(held[0].seq, 9, "the newer entry wins");
721 }
722 }
723