1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
use crate::error::Result;
use super::StateStore;
impl StateStore {
/// Record that `cdc.initial: snapshot`'s backfill for `(export_name,
/// table_name)` completed, on run `run_id`. Idempotent (upsert on the
/// `(export_name, table_name)` key), so a retried run never double-inserts.
///
/// This is the durable, cleanup-proof twin of the GCS `snapshot/_SUCCESS`
/// marker: once here, `cleanup_source: true` may wipe the bucket without the
/// next run mistaking the table for un-snapshotted and re-snapshotting it.
pub fn mark_snapshot_done(
&self,
export_name: &str,
table_name: &str,
run_id: &str,
) -> Result<()> {
let now = chrono::Utc::now().to_rfc3339();
// ON CONFLICT DO UPDATE upsert (SQLite ≥ 3.24 + Postgres) — the codebase's
// canonical upsert, replacing the old SQLite `INSERT OR REPLACE` arm.
self.execute(
"INSERT INTO cdc_snapshot (export_name, table_name, run_id, completed_at)
VALUES (?1, ?2, ?3, ?4)
ON CONFLICT (export_name, table_name) DO UPDATE SET
run_id = excluded.run_id,
completed_at = excluded.completed_at",
&[
export_name.into(),
table_name.into(),
run_id.into(),
now.into(),
],
)?;
Ok(())
}
/// Whether `(export_name, table_name)`'s initial snapshot has completed per
/// the state DB — the authoritative, GCS-independent signal.
pub fn snapshot_done(&self, export_name: &str, table_name: &str) -> Result<bool> {
Ok(self
.query_opt(
"SELECT COUNT(*) FROM cdc_snapshot WHERE export_name = ?1 AND table_name = ?2",
&[export_name.into(), table_name.into()],
|r| r.i64(0),
)?
.unwrap_or(0)
> 0)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn snapshot_done_is_false_until_marked_then_true() {
let s = StateStore::open_in_memory().unwrap();
assert!(!s.snapshot_done("customers", "customers").unwrap());
s.mark_snapshot_done("customers", "customers", "run_1")
.unwrap();
assert!(s.snapshot_done("customers", "customers").unwrap());
// Scoped per (export, table): a sibling table is still un-snapshotted.
assert!(!s.snapshot_done("customers", "orders").unwrap());
}
#[test]
fn mark_snapshot_done_is_idempotent() {
let s = StateStore::open_in_memory().unwrap();
s.mark_snapshot_done("e", "t", "run_1").unwrap();
s.mark_snapshot_done("e", "t", "run_2").unwrap(); // replay/re-record
assert!(s.snapshot_done("e", "t").unwrap());
}
}