Skip to main content

max / synckit

6.2 KB · 161 lines History Blame Raw
1 //! Last-synced row snapshots: the base version a three-way field merge needs.
2 //!
3 //! [`resolve_field_merge`](crate::conflict::resolve_field_merge) has always been
4 //! able to merge two edits that touched different columns, and nothing could call
5 //! it, because a three-way merge needs the version both devices started from and
6 //! the engine kept no such version. This is that version.
7 //!
8 //! A snapshot is the wire payload of the last change for a row that **both
9 //! devices have seen**, which is exactly the two moments a row becomes common
10 //! ground: a remote change this device applied, and a local change the server
11 //! acknowledged. Written at both, so the base re-bases itself on every sync and
12 //! never needs a separate reconciliation pass.
13 //!
14 //! The payload is stored rather than the row read back, deliberately. A merge
15 //! compares two payloads against the base, and a row read carries columns the
16 //! wire never sends (`preserve_local` secrets, group provenance, anything outside
17 //! the manifest). Basing a merge on those would report a field as "changed" on
18 //! every device that has a different local secret. The payload is the only shape
19 //! all three sides share.
20 //!
21 //! Opt-in per table ([`SyncTable::field_merge`](super::schema::SyncTable::field_merge)):
22 //! a table nobody opted in stores nothing here and resolves conflicts exactly as
23 //! it did before. Storage roughly doubles for the tables that do opt in, which is
24 //! why it is not simply on for everything.
25
26 use rusqlite::{Connection, OptionalExtension};
27 use serde_json::Value;
28
29 use crate::error::Result;
30
31 /// DDL for the snapshot store, shared by
32 /// [`SyncSchema::migration_sql`](super::schema::SyncSchema::migration_sql) and the
33 /// per-connection upgrade in [`db::ensure_scope_schema`](super::db::ensure_scope_schema),
34 /// so an install that predates field merge gets the table on its next connection
35 /// open rather than waiting for the app to re-run its migration.
36 ///
37 /// One row per `(table_name, row_id)`. Not scoped: a row lives in one scope at a
38 /// time and moving it between scopes rewrites the row, so a per-scope base would
39 /// be a second copy of the same answer.
40 pub(crate) const SNAPSHOT_DDL: &str = "\
41 CREATE TABLE IF NOT EXISTS sync_row_snapshot (
42 table_name TEXT NOT NULL,
43 row_id TEXT NOT NULL,
44 payload TEXT NOT NULL,
45 updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
46 PRIMARY KEY (table_name, row_id)
47 ) WITHOUT ROWID;
48 ";
49
50 /// Record `payload` as the last-synced base for a row.
51 ///
52 /// Called on both sides of common ground: after a remote change is applied, and
53 /// after a local change is acknowledged by a push. A non-object payload is not
54 /// stored, since it could never serve as a merge base.
55 pub(crate) fn record(conn: &Connection, table: &str, row_id: &str, payload: &Value) -> Result<()> {
56 if !payload.is_object() {
57 return Ok(());
58 }
59 conn.execute(
60 "INSERT INTO sync_row_snapshot (table_name, row_id, payload) VALUES (?1, ?2, ?3) \
61 ON CONFLICT(table_name, row_id) DO UPDATE SET \
62 payload = excluded.payload, \
63 updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')",
64 rusqlite::params![table, row_id, serde_json::to_string(payload)?],
65 )?;
66 Ok(())
67 }
68
69 /// Drop a row's base, because the row is gone.
70 ///
71 /// A stale base outliving its row would be handed to the merge if the same key
72 /// were later recreated, and it would describe a version of a different row.
73 pub(crate) fn forget(conn: &Connection, table: &str, row_id: &str) -> Result<()> {
74 conn.execute(
75 "DELETE FROM sync_row_snapshot WHERE table_name = ?1 AND row_id = ?2",
76 rusqlite::params![table, row_id],
77 )?;
78 Ok(())
79 }
80
81 /// The base for a row, or [`Value::Null`] when there is none.
82 ///
83 /// Null rather than an error or an `Option` because that is what the merge takes:
84 /// [`resolve_field_merge`](crate::conflict::resolve_field_merge) treats a
85 /// non-object base as "no usable base" and falls back on its own. A read error is
86 /// logged and reported as absent, since a merge this device cannot base is a
87 /// merge that should not happen, not a sync that should fail.
88 pub(crate) fn load(conn: &Connection, table: &str, row_id: &str) -> Value {
89 let stored: Option<String> = match conn
90 .query_row(
91 "SELECT payload FROM sync_row_snapshot WHERE table_name = ?1 AND row_id = ?2",
92 rusqlite::params![table, row_id],
93 |r| r.get(0),
94 )
95 .optional()
96 {
97 Ok(v) => v,
98 Err(e) => {
99 tracing::warn!(
100 table,
101 row_id,
102 "snapshot lookup failed, treating as absent: {e}"
103 );
104 None
105 }
106 };
107 stored
108 .and_then(|s| serde_json::from_str(&s).ok())
109 .unwrap_or(Value::Null)
110 }
111
112 #[cfg(test)]
113 mod tests {
114 use super::*;
115 use serde_json::json;
116
117 fn conn() -> Connection {
118 let conn = Connection::open_in_memory().unwrap();
119 conn.execute_batch(SNAPSHOT_DDL).unwrap();
120 conn
121 }
122
123 #[test]
124 fn round_trips_and_overwrites() {
125 let c = conn();
126 record(&c, "note", "r1", &json!({"name": "one"})).unwrap();
127 assert_eq!(load(&c, "note", "r1"), json!({"name": "one"}));
128
129 record(&c, "note", "r1", &json!({"name": "two"})).unwrap();
130 assert_eq!(
131 load(&c, "note", "r1"),
132 json!({"name": "two"}),
133 "a later sync must re-base the row, not accumulate versions"
134 );
135 }
136
137 #[test]
138 fn absent_and_forgotten_rows_read_as_null() {
139 let c = conn();
140 assert_eq!(load(&c, "note", "missing"), Value::Null);
141
142 record(&c, "note", "r1", &json!({"name": "one"})).unwrap();
143 forget(&c, "note", "r1").unwrap();
144 assert_eq!(
145 load(&c, "note", "r1"),
146 Value::Null,
147 "a deleted row's base must not survive to be merged against"
148 );
149 }
150
151 /// A delete carries no object payload, and an entry with no payload has no
152 /// base to offer. Storing one would put a scalar where the merge expects an
153 /// object.
154 #[test]
155 fn a_non_object_payload_is_not_stored() {
156 let c = conn();
157 record(&c, "note", "r1", &json!("not an object")).unwrap();
158 assert_eq!(load(&c, "note", "r1"), Value::Null);
159 }
160 }
161