Skip to main content

max / goingson

13.1 KB · 335 lines History Blame Raw
1 //! Pull remote changes and apply them to the local database.
2
3 use chrono::Utc;
4 use goingson_core::CoreError;
5 use sqlx::{SqliteConnection, SqlitePool};
6 use synckit_client::{
7 ChangeEntry, ChangeOp, DeviceId, Hlc, Resolution, SyncKitClient,
8 detect_conflicts, resolve_lww,
9 };
10 use tracing::{debug, warn};
11 use uuid::Uuid;
12
13 use super::state::{get_sync_state, set_sync_state};
14 use super::hlc::{assign_pending_hlcs, load_committed_hlcs, observe_remote, record_committed_hlc};
15 use super::{UPSERT_ORDER, DELETE_ORDER};
16 use super::apply::{apply_upsert, apply_delete};
17
18 #[tracing::instrument(skip_all)]
19 pub async fn pull_changes(
20 pool: &SqlitePool,
21 client: &SyncKitClient,
22 device_id: Uuid,
23 ) -> Result<PullOutcome, CoreError> {
24 let cursor_str = get_sync_state(pool, "pull_cursor").await?;
25 let mut cursor: i64 = cursor_str.parse().unwrap_or(0);
26 let mut total_applied: i64 = 0;
27 let mut changed_tables: std::collections::HashSet<String> = std::collections::HashSet::new();
28
29 // Read local pending changes once (unpushed changelog entries). This also
30 // stamps them with their HLC so conflict resolution can compare clocks.
31 let local_pending = read_local_pending(pool, device_id).await?;
32
33 loop {
34 let (changes, new_cursor, has_more) = client
35 .pull_rich(DeviceId::new(device_id), cursor)
36 .await
37 .map_err(|e| CoreError::sync(format!("pull failed: {}", e)))?;
38
39 if changes.is_empty() {
40 set_sync_state(pool, "pull_cursor", &new_cursor.to_string()).await?;
41 break;
42 }
43
44 // Advance our clock past everything in this batch, so future local changes
45 // causally follow the remote ones. Observing the batch maximum suffices.
46 if let Some(max_remote) = changes.iter().map(|p| p.entry.hlc).max() {
47 observe_remote(pool, max_remote, device_id).await?;
48 }
49
50 // Detect conflicts between remote changes and local pending
51 let (clean, conflicts) = detect_conflicts(changes, &local_pending, DeviceId::new(device_id));
52
53 // Apply non-conflicting changes — but HLC-gate them first. "Clean" only
54 // means no local *pending* edit contests the row; an older remote edit for
55 // an already-committed row also lands here, and applying it blindly would
56 // clobber the newer local value. Pre-fetch the rows' committed clocks, then
57 // let SyncKit's CleanChanges gate drop anything older-or-equal.
58 let clean_count = clean.len() as i64;
59 if !clean.is_empty() {
60 let keys: Vec<(String, String)> = clean
61 .row_keys()
62 .map(|(t, r)| (t.to_string(), r.to_string()))
63 .collect();
64 let committed = load_committed_hlcs(pool, &keys).await?;
65 let clean_entries: Vec<ChangeEntry> =
66 clean.gated(|t, r| committed.get(&(t.to_string(), r.to_string())).copied());
67 if !clean_entries.is_empty() {
68 changed_tables.extend(clean_entries.iter().map(|e| e.table.clone()));
69 apply_remote_changes(pool, clean_entries).await?;
70 }
71 }
72
73 // Resolve conflicts with LWW
74 let mut resolved_count: i64 = 0;
75 if !conflicts.is_empty() {
76 let conflict_count = conflicts.len();
77 let mut resolved_entries: Vec<ChangeEntry> = Vec::new();
78
79 for pair in &conflicts {
80 match resolve_lww(&pair.local, &pair.remote) {
81 Resolution::KeepRemote => {
82 resolved_entries.push(pair.remote.entry.clone());
83 }
84 Resolution::KeepLocal => {
85 // Skip — local version will push on next cycle
86 }
87 Resolution::Merged(data) => {
88 // Apply merged data using the remote entry as template
89 let mut merged = pair.remote.entry.clone();
90 merged.data = Some(data);
91 resolved_entries.push(merged);
92 }
93 Resolution::Skip => {
94 // Skip both
95 }
96 }
97 }
98
99 resolved_count = resolved_entries.len() as i64;
100 if !resolved_entries.is_empty() {
101 changed_tables.extend(resolved_entries.iter().map(|e| e.table.clone()));
102 apply_remote_changes(pool, resolved_entries).await?;
103 }
104
105 debug!(
106 "Resolved {} conflicts ({} applied as remote wins)",
107 conflict_count,
108 conflict_count - conflicts.iter().filter(|p| {
109 matches!(resolve_lww(&p.local, &p.remote), Resolution::KeepLocal | Resolution::Skip)
110 }).count(),
111 );
112 }
113
114 total_applied += clean_count + resolved_count;
115
116 // Save cursor after each batch (crash-safe)
117 set_sync_state(pool, "pull_cursor", &new_cursor.to_string()).await?;
118 cursor = new_cursor;
119
120 if !has_more {
121 break;
122 }
123 }
124
125 if total_applied > 0 {
126 debug!("Pulled and applied {} remote changes", total_applied);
127 }
128 Ok(PullOutcome { applied: total_applied, changed_tables })
129 }
130
131 /// Result of a pull: how many changes landed and which tables they touched.
132 ///
133 /// The table set drives selective cache invalidation in the UI so a pull that
134 /// only touched, say, `tasks` doesn't force the compose screen to re-hydrate
135 /// every contact.
136 pub struct PullOutcome {
137 pub applied: i64,
138 pub changed_tables: std::collections::HashSet<String>,
139 }
140
141 /// Read unpushed changelog entries as ChangeEntry values for conflict detection.
142 ///
143 /// Stamps any not-yet-stamped rows with their HLC first, so the clocks used here
144 /// match the ones the push path will send.
145 /// A pending sync_changelog row: (table_name, op, row_id, timestamp, data, hlc_wall, hlc_counter).
146 type PendingRow = (String, String, String, String, Option<String>, Option<i64>, Option<i64>);
147
148 async fn read_local_pending(pool: &SqlitePool, device_id: Uuid) -> Result<Vec<ChangeEntry>, CoreError> {
149 assign_pending_hlcs(pool, device_id).await?;
150
151 let rows: Vec<PendingRow> = sqlx::query_as(
152 "SELECT table_name, op, row_id, timestamp, data, hlc_wall, hlc_counter \
153 FROM sync_changelog WHERE pushed = 0 ORDER BY id ASC"
154 )
155 .fetch_all(pool)
156 .await
157 .map_err(CoreError::database)?;
158
159 let entries: Vec<ChangeEntry> = rows
160 .into_iter()
161 .filter_map(|(table, op, row_id, timestamp, data, hlc_wall, hlc_counter)| {
162 let ts = chrono::DateTime::parse_from_rfc3339(&timestamp)
163 .map(|dt| dt.with_timezone(&Utc))
164 .unwrap_or(chrono::DateTime::UNIX_EPOCH);
165
166 let change_op = ChangeOp::from_str_opt(&op)?;
167 let json_data = data.and_then(|d| serde_json::from_str(&d).ok());
168
169 let hlc = match (hlc_wall, hlc_counter) {
170 (Some(wall), Some(counter)) => Hlc { wall_ms: wall, counter: counter as u32, node: DeviceId::new(device_id) },
171 _ => Hlc::from_legacy(ts.timestamp_millis(), DeviceId::new(device_id)),
172 };
173
174 Some(ChangeEntry {
175 table,
176 op: change_op,
177 row_id,
178 timestamp: ts,
179 hlc,
180 data: json_data,
181 extra: Default::default(),
182 })
183 })
184 .collect();
185
186 Ok(entries)
187 }
188
189 /// Apply remote changes to local DB with triggers suppressed and FK enforcement off.
190 ///
191 /// FK enforcement is disabled so that tasks with `source_email_id` pointing to
192 /// emails not yet fetched locally can be inserted without error. Uses a dedicated
193 /// connection (same pattern as `crates/db-sqlite/src/migrations.rs`).
194 ///
195 /// The `applying_remote` flag is set inside a transaction on the dedicated connection.
196 /// In WAL mode, uncommitted changes are only visible to the writing connection, so
197 /// other connections (handling user edits) never see the flag and their triggers fire
198 /// normally. The flag is reset to '0' before commit so it's never globally visible as '1'.
199 ///
200 /// Uses `detach()` to prevent returning the connection to the pool with FK OFF
201 /// if pragma restoration fails. The pool will create a fresh connection (with
202 /// FK ON via connect options) to replace it.
203 pub(crate) async fn apply_remote_changes(pool: &SqlitePool, changes: Vec<ChangeEntry>) -> Result<(), CoreError> {
204 // Acquire a dedicated connection for FK pragma and transaction
205 let mut conn = pool.acquire().await.map_err(CoreError::database)?;
206
207 // FK pragma must be set outside a transaction (SQLite requirement)
208 sqlx::query("PRAGMA foreign_keys = OFF")
209 .execute(&mut *conn)
210 .await
211 .map_err(CoreError::database)?;
212
213 // Use a transaction so applying_remote is only visible to this connection (WAL isolation).
214 // Other connections see the committed value ('0') and their triggers fire normally.
215 sqlx::query("BEGIN IMMEDIATE")
216 .execute(&mut *conn)
217 .await
218 .map_err(CoreError::database)?;
219
220 sqlx::query("UPDATE sync_state SET value = '1' WHERE key = 'applying_remote'")
221 .execute(&mut *conn)
222 .await
223 .map_err(CoreError::database)?;
224
225 let result = apply_changes_inner(&mut conn, changes).await;
226
227 // Reset flag before commit so it's never globally visible as '1'
228 let _ = sqlx::query("UPDATE sync_state SET value = '0' WHERE key = 'applying_remote'")
229 .execute(&mut *conn)
230 .await;
231
232 match result {
233 Ok(skipped) => {
234 if let Err(e) = sqlx::query("COMMIT").execute(&mut *conn).await {
235 let _ = sqlx::query("ROLLBACK").execute(&mut *conn).await;
236 let _ = sqlx::query("PRAGMA foreign_keys = ON").execute(&mut *conn).await;
237 return Err(CoreError::database(e));
238 }
239 if skipped > 0 {
240 warn!(
241 skipped,
242 "Skipped un-appliable remote change entries; the rest of the batch \
243 was applied and the pull cursor advanced (sync not wedged)."
244 );
245 }
246 }
247 Err(ref _e) => {
248 let _ = sqlx::query("ROLLBACK").execute(&mut *conn).await;
249 }
250 }
251
252 // Always restore FK enforcement. If this fails, detach the connection
253 // so it doesn't return to the pool with FK OFF.
254 if let Err(e) = sqlx::query("PRAGMA foreign_keys = ON")
255 .execute(&mut *conn)
256 .await
257 {
258 conn.detach();
259 return Err(CoreError::database(e));
260 }
261
262 result.map(|_| ())
263 }
264
265 /// Apply a batch of remote changes within an open transaction.
266 ///
267 /// Per-entry failures are skipped and logged rather than aborting the whole batch:
268 /// before, one un-appliable remote row (e.g. a payload missing a NOT NULL column)
269 /// rolled back the transaction, `apply_remote_changes` returned Err, and the pull
270 /// cursor never advanced -- so every subsequent pull re-fetched the same poisoned
271 /// batch and failed identically, wedging sync permanently (ultra-fuzz Run #27
272 /// Data S-1). A failed statement does not poison the SQLite transaction, so the
273 /// good rows still commit and the cursor advances. Returns the count of skipped
274 /// entries (the caller logs a summary); only a catastrophic error returns `Err`.
275 pub(crate) async fn apply_changes_inner(
276 conn: &mut SqliteConnection,
277 changes: Vec<ChangeEntry>,
278 ) -> Result<usize, CoreError> {
279 // Separate upserts from deletes
280 let mut upserts: Vec<&ChangeEntry> = Vec::new();
281 let mut deletes: Vec<&ChangeEntry> = Vec::new();
282
283 for change in &changes {
284 match change.op {
285 ChangeOp::Insert | ChangeOp::Update => upserts.push(change),
286 ChangeOp::Delete => deletes.push(change),
287 }
288 }
289
290 let mut skipped = 0usize;
291
292 // Apply upserts in parent-first FK order
293 for table in UPSERT_ORDER {
294 for change in &upserts {
295 if change.table != *table {
296 continue;
297 }
298 let Some(ref data) = change.data else {
299 continue;
300 };
301 match apply_upsert(&mut *conn, table, &change.row_id, data).await {
302 Ok(()) => {
303 record_committed_hlc(&mut *conn, table, &change.row_id, change.hlc).await?;
304 }
305 Err(e) => {
306 warn!(table = *table, row_id = %change.row_id, error = %e,
307 "Skipping un-appliable remote upsert");
308 skipped += 1;
309 }
310 }
311 }
312 }
313
314 // Apply deletes in child-first FK order
315 for table in DELETE_ORDER {
316 for change in &deletes {
317 if change.table != *table {
318 continue;
319 }
320 match apply_delete(&mut *conn, table, &change.row_id).await {
321 Ok(()) => {
322 record_committed_hlc(&mut *conn, table, &change.row_id, change.hlc).await?;
323 }
324 Err(e) => {
325 warn!(table = *table, row_id = %change.row_id, error = %e,
326 "Skipping un-appliable remote delete");
327 skipped += 1;
328 }
329 }
330 }
331 }
332
333 Ok(skipped)
334 }
335