Skip to main content

max / goingson

Stop one bad remote row wedging sync; fix parser edge cases Data & Sync axis remediation (ultra-fuzz Run #27): - SERIOUS: apply_changes_inner now skips-and-logs a per-entry failure instead of aborting the whole batch. Before, one un-appliable remote row (e.g. a payload missing a NOT NULL column) rolled back the transaction, the pull cursor never advanced, and every later pull re-failed identically -- sync wedged forever. A failed statement does not poison the SQLite transaction, so the good rows commit and the cursor advances. - blob_sync: replace &hash[..8] (panics on a short/multibyte synced hash) with a panic-safe short() helper. - iCal: floating and unresolved-TZID times are interpreted as local wall-clock per RFC 5545, not stamped UTC (which shifted events by the viewer's offset). - vCard: CATEGORIES splits on unescaped commas only, so an escaped "\," is one tag, not two. Tests: poison-batch skip-and-apply-rest, vCard escaped-comma category. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-18 23:39 UTC
Commit: 9a1dec46fab6cee3a0e8b6443add297376c5adb7
Parent: 79ecfb5
5 files changed, +172 insertions, -33 deletions
M src-tauri/src/external_sync/ical.rs +19 -5
@@ -3,7 +3,7 @@
3 3 //! Uses the `ical` crate for robust VEVENT parsing, then maps properties
4 4 //! to GO's Event model.
5 5
6 - use chrono::{DateTime, NaiveDate, NaiveDateTime, TimeZone, Utc};
6 + use chrono::{DateTime, Local, NaiveDate, NaiveDateTime, TimeZone, Utc};
7 7 use chrono_tz::Tz;
8 8 use goingson_core::Recurrence;
9 9 use ical::parser::ical::component::IcalEvent;
@@ -135,8 +135,10 @@ fn parse_datetime_property(properties: &[Property], name: &str) -> Option<DateTi
135 135 return Some(local_dt.with_timezone(&Utc));
136 136 }
137 137 }
138 - // Fall back to treating as UTC if timezone can't be resolved
139 - return Some(Utc.from_utc_datetime(&ndt));
138 + // Timezone name couldn't be resolved: interpret the wall-clock time in
139 + // the machine's local zone rather than UTC (closer to the author's
140 + // intent than a blind UTC stamp).
141 + return naive_local_to_utc(&ndt);
140 142 }
141 143 }
142 144
@@ -145,8 +147,20 @@ fn parse_datetime_property(properties: &[Property], name: &str) -> Option<DateTi
145 147 return parse_ical_datetime(clean).map(|ndt| Utc.from_utc_datetime(&ndt));
146 148 }
147 149
148 - // Floating time (no timezone) → treat as UTC
149 - parse_ical_datetime(value).map(|ndt| Utc.from_utc_datetime(&ndt))
150 + // Floating time (no timezone) → interpret as local wall-clock per RFC 5545,
151 + // not UTC (a blind UTC stamp shifts the event by the viewer's offset).
152 + parse_ical_datetime(value).and_then(|ndt| naive_local_to_utc(&ndt))
153 + }
154 +
155 + /// Convert a naive (timezone-less) datetime to UTC by interpreting it in the
156 + /// machine's local timezone -- the RFC 5545 meaning of a floating time. Handles
157 + /// DST gaps by falling back from `.earliest()` to `.latest()`.
158 + fn naive_local_to_utc(ndt: &NaiveDateTime) -> Option<DateTime<Utc>> {
159 + Local
160 + .from_local_datetime(ndt)
161 + .earliest()
162 + .or_else(|| Local.from_local_datetime(ndt).latest())
163 + .map(|dt| dt.with_timezone(&Utc))
150 164 }
151 165
152 166 /// Parse an iCalendar datetime string (YYYYMMDDTHHMMSS).
M src-tauri/src/external_sync/vcard.rs +44 -1
@@ -187,7 +187,10 @@ fn parse_single_vcard(lines: &[&str]) -> Option<ParsedVCard> {
187 187 }
188 188 }
189 189 "CATEGORIES" => {
190 - for cat in value.split(',') {
190 + // Split on unescaped commas only: a category containing an escaped
191 + // `\,` is one tag, not two (ultra-fuzz Run #27 Data minor). Unescape
192 + // each segment afterwards.
193 + for cat in split_unescaped_commas(&value) {
191 194 let cat = decode_value(cat.trim(), &params);
192 195 if !cat.is_empty() {
193 196 tags.push(cat);
@@ -313,6 +316,32 @@ fn parse_property_line(line: &str) -> (String, Vec<String>, String) {
313 316 }
314 317
315 318 /// Decode a value, handling quoted-printable encoding if indicated by params.
319 + /// Split a vCard list value on commas that are not backslash-escaped, preserving
320 + /// the escape sequences within each segment so `decode_value` can unescape them.
321 + fn split_unescaped_commas(s: &str) -> Vec<String> {
322 + let mut out = Vec::new();
323 + let mut cur = String::new();
324 + let mut escaped = false;
325 + for c in s.chars() {
326 + if escaped {
327 + cur.push('\\');
328 + cur.push(c);
329 + escaped = false;
330 + } else if c == '\\' {
331 + escaped = true;
332 + } else if c == ',' {
333 + out.push(std::mem::take(&mut cur));
334 + } else {
335 + cur.push(c);
336 + }
337 + }
338 + if escaped {
339 + cur.push('\\');
340 + }
341 + out.push(cur);
342 + out
343 + }
344 +
316 345 fn decode_value(value: &str, params: &[String]) -> String {
317 346 let is_qp = params.iter().any(|p| {
318 347 let upper = p.to_uppercase();
@@ -534,6 +563,20 @@ END:VCARD\r\n";
534 563 }
535 564
536 565 #[test]
566 + fn test_categories_escaped_comma_is_one_tag() {
567 + // An escaped comma inside a category must not split into two tags.
568 + let vcf = "\
569 + BEGIN:VCARD\r\n\
570 + VERSION:3.0\r\n\
571 + FN:Test\r\n\
572 + CATEGORIES:Smith\\, Jones & Co,VIP\r\n\
573 + END:VCARD\r\n";
574 +
575 + let cards = parse_vcf(vcf).unwrap();
576 + assert_eq!(cards[0].tags, vec!["Smith, Jones & Co", "VIP"]);
577 + }
578 +
579 + #[test]
537 580 fn test_line_folding() {
538 581 // In vCard, line folding splits content and prepends a single space/tab to continuation.
539 582 // The fold indicator (leading space) is stripped; the space in "continues " is content.
M src-tauri/src/sync_service/blob_sync.rs +18 -11
@@ -10,6 +10,13 @@ use tracing::{debug, info, warn};
10 10 use crate::commands::attachment::blob_path;
11 11 use crate::state::DESKTOP_USER_ID;
12 12
13 + /// First 8 chars of a blob hash for log lines. Panic-safe: a `blob_hash` can arrive
14 + /// via sync and need not be a clean 64-hex string, so a raw `&hash[..8]` could panic
15 + /// on a short or multibyte value (ultra-fuzz Run #27 Data minor).
16 + fn short(hash: &str) -> &str {
17 + hash.get(..8).unwrap_or(hash)
18 + }
19 +
13 20 /// Upload local blobs that haven't been synced to the server yet.
14 21 ///
15 22 /// Queries all distinct blob hashes from attachments, checks which ones have local
@@ -39,13 +46,13 @@ pub async fn upload_pending_blobs(
39 46 let upload_resp = match client.blob_upload_url(hash, *size).await {
40 47 Ok(r) => r,
41 48 Err(e) => {
42 - warn!("Failed to get upload URL for blob {}: {}", &hash[..8], e);
49 + warn!("Failed to get upload URL for blob {}: {}", short(hash), e);
43 50 continue;
44 51 }
45 52 };
46 53
47 54 if upload_resp.already_exists {
48 - debug!("Blob {} already on server, skipping", &hash[..8]);
55 + debug!("Blob {} already on server, skipping", short(hash));
49 56 continue;
50 57 }
51 58
@@ -53,24 +60,24 @@ pub async fn upload_pending_blobs(
53 60 let data = match tokio::fs::read(&path).await {
54 61 Ok(d) => d,
55 62 Err(e) => {
56 - warn!("Failed to read blob {}: {}", &hash[..8], e);
63 + warn!("Failed to read blob {}: {}", short(hash), e);
57 64 continue;
58 65 }
59 66 };
60 67
61 68 if let Err(e) = client.blob_upload(&upload_resp.upload_url, data).await {
62 - warn!("Failed to upload blob {}: {}", &hash[..8], e);
69 + warn!("Failed to upload blob {}: {}", short(hash), e);
63 70 continue;
64 71 }
65 72
66 73 // Confirm upload
67 74 if let Err(e) = client.blob_confirm(hash, *size).await {
68 - warn!("Failed to confirm blob {}: {}", &hash[..8], e);
75 + warn!("Failed to confirm blob {}: {}", short(hash), e);
69 76 continue;
70 77 }
71 78
72 79 uploaded += 1;
73 - debug!("Uploaded blob {}", &hash[..8]);
80 + debug!("Uploaded blob {}", short(hash));
74 81 }
75 82
76 83 if uploaded > 0 {
@@ -113,7 +120,7 @@ pub async fn download_missing_blobs(
113 120 let download_url = match client.blob_download_url(hash).await {
114 121 Ok(url) => url,
115 122 Err(e) => {
116 - warn!("Failed to get download URL for blob {}: {}", &hash[..8], e);
123 + warn!("Failed to get download URL for blob {}: {}", short(hash), e);
117 124 continue;
118 125 }
119 126 };
@@ -122,7 +129,7 @@ pub async fn download_missing_blobs(
122 129 let data = match client.blob_download(&download_url).await {
123 130 Ok(d) => d,
124 131 Err(e) => {
125 - warn!("Failed to download blob {}: {}", &hash[..8], e);
132 + warn!("Failed to download blob {}: {}", short(hash), e);
126 133 continue;
127 134 }
128 135 };
@@ -130,17 +137,17 @@ pub async fn download_missing_blobs(
130 137 // Write to disk atomically (tmp + rename) to prevent corrupt partial files
131 138 let tmp_path = path.with_extension("tmp");
132 139 if let Err(e) = tokio::fs::write(&tmp_path, &data).await {
133 - warn!("Failed to write blob {}: {}", &hash[..8], e);
140 + warn!("Failed to write blob {}: {}", short(hash), e);
134 141 continue;
135 142 }
136 143 if let Err(e) = tokio::fs::rename(&tmp_path, &path).await {
137 - warn!("Failed to rename blob {}: {}", &hash[..8], e);
144 + warn!("Failed to rename blob {}: {}", short(hash), e);
138 145 let _ = tokio::fs::remove_file(&tmp_path).await;
139 146 continue;
140 147 }
141 148
142 149 downloaded += 1;
143 - debug!("Downloaded blob {}", &hash[..8]);
150 + debug!("Downloaded blob {}", short(hash));
144 151 }
145 152
146 153 if downloaded > 0 {
M src-tauri/src/sync_service/pull.rs +49 -16
@@ -7,7 +7,7 @@ use synckit_client::{
7 7 ChangeEntry, ChangeOp, Resolution, SyncKitClient,
8 8 detect_conflicts, resolve_lww,
9 9 };
10 - use tracing::debug;
10 + use tracing::{debug, warn};
11 11 use uuid::Uuid;
12 12
13 13 use super::state::{get_sync_state, set_sync_state};
@@ -193,14 +193,24 @@ pub(crate) async fn apply_remote_changes(pool: &SqlitePool, changes: Vec<ChangeE
193 193 .execute(&mut *conn)
194 194 .await;
195 195
196 - if result.is_ok() {
197 - if let Err(e) = sqlx::query("COMMIT").execute(&mut *conn).await {
196 + match result {
197 + Ok(skipped) => {
198 + if let Err(e) = sqlx::query("COMMIT").execute(&mut *conn).await {
199 + let _ = sqlx::query("ROLLBACK").execute(&mut *conn).await;
200 + let _ = sqlx::query("PRAGMA foreign_keys = ON").execute(&mut *conn).await;
201 + return Err(CoreError::database(e));
202 + }
203 + if skipped > 0 {
204 + warn!(
205 + skipped,
206 + "Skipped un-appliable remote change entries; the rest of the batch \
207 + was applied and the pull cursor advanced (sync not wedged)."
208 + );
209 + }
210 + }
211 + Err(ref _e) => {
198 212 let _ = sqlx::query("ROLLBACK").execute(&mut *conn).await;
199 - let _ = sqlx::query("PRAGMA foreign_keys = ON").execute(&mut *conn).await;
200 - return Err(CoreError::database(e));
201 213 }
202 - } else {
203 - let _ = sqlx::query("ROLLBACK").execute(&mut *conn).await;
204 214 }
205 215
206 216 // Always restore FK enforcement. If this fails, detach the connection
@@ -213,13 +223,23 @@ pub(crate) async fn apply_remote_changes(pool: &SqlitePool, changes: Vec<ChangeE
213 223 return Err(CoreError::database(e));
214 224 }
215 225
216 - result
226 + result.map(|_| ())
217 227 }
218 228
229 + /// Apply a batch of remote changes within an open transaction.
230 + ///
231 + /// Per-entry failures are skipped and logged rather than aborting the whole batch:
232 + /// before, one un-appliable remote row (e.g. a payload missing a NOT NULL column)
233 + /// rolled back the transaction, `apply_remote_changes` returned Err, and the pull
234 + /// cursor never advanced -- so every subsequent pull re-fetched the same poisoned
235 + /// batch and failed identically, wedging sync permanently (ultra-fuzz Run #27
236 + /// Data S-1). A failed statement does not poison the SQLite transaction, so the
237 + /// good rows still commit and the cursor advances. Returns the count of skipped
238 + /// entries (the caller logs a summary); only a catastrophic error returns `Err`.
219 239 pub(crate) async fn apply_changes_inner(
220 240 conn: &mut SqliteConnection,
221 241 changes: Vec<ChangeEntry>,
222 - ) -> Result<(), CoreError> {
242 + ) -> Result<usize, CoreError> {
223 243 // Separate upserts from deletes
224 244 let mut upserts: Vec<&ChangeEntry> = Vec::new();
225 245 let mut deletes: Vec<&ChangeEntry> = Vec::new();
@@ -231,13 +251,21 @@ pub(crate) async fn apply_changes_inner(
231 251 }
232 252 }
233 253
254 + let mut skipped = 0usize;
255 +
234 256 // Apply upserts in parent-first FK order
235 257 for table in UPSERT_ORDER {
236 258 for change in &upserts {
237 - if change.table == *table {
238 - if let Some(ref data) = change.data {
239 - apply_upsert(&mut *conn, table, &change.row_id, data).await?;
240 - }
259 + if change.table != *table {
260 + continue;
261 + }
262 + let Some(ref data) = change.data else {
263 + continue;
264 + };
265 + if let Err(e) = apply_upsert(&mut *conn, table, &change.row_id, data).await {
266 + warn!(table = *table, row_id = %change.row_id, error = %e,
267 + "Skipping un-appliable remote upsert");
268 + skipped += 1;
241 269 }
242 270 }
243 271 }
@@ -245,11 +273,16 @@ pub(crate) async fn apply_changes_inner(
245 273 // Apply deletes in child-first FK order
246 274 for table in DELETE_ORDER {
247 275 for change in &deletes {
248 - if change.table == *table {
249 - apply_delete(&mut *conn, table, &change.row_id).await?;
276 + if change.table != *table {
277 + continue;
278 + }
279 + if let Err(e) = apply_delete(&mut *conn, table, &change.row_id).await {
280 + warn!(table = *table, row_id = %change.row_id, error = %e,
281 + "Skipping un-appliable remote delete");
282 + skipped += 1;
250 283 }
251 284 }
252 285 }
253 286
254 - Ok(())
287 + Ok(skipped)
255 288 }
M src-tauri/src/sync_service/tests.rs +42
@@ -554,6 +554,48 @@ use sqlx::SqlitePool;
554 554 assert_eq!(count.0, 0);
555 555 }
556 556
557 + #[tokio::test]
558 + async fn apply_changes_inner_skips_bad_entry_and_applies_the_rest() {
559 + // One un-appliable remote row must not abort the whole batch (ultra-fuzz
560 + // Run #27 Data S-1: a single bad row used to wedge the pull cursor forever).
561 + let pool = setup_test_db().await;
562 + let user_id = create_test_user(&pool).await;
563 + let now = now_sql();
564 + let good_id = uuid::Uuid::new_v4().to_string();
565 + let bad_id = uuid::Uuid::new_v4().to_string();
566 +
567 + let changes = vec![
568 + // Malformed: `name` is NOT NULL but omitted -> bound NULL -> fails.
569 + change("projects", synckit_client::ChangeOp::Insert, &bad_id, Some(json!({
570 + "id": bad_id, "description": "", "project_type": "Job",
571 + "status": "Active", "created_at": now, "user_id": user_id,
572 + }))),
573 + // Well-formed: must still be applied.
574 + change("projects", synckit_client::ChangeOp::Insert, &good_id, Some(json!({
575 + "id": good_id, "name": "Good project", "description": "",
576 + "project_type": "Job", "status": "Active",
577 + "created_at": now, "user_id": user_id,
578 + }))),
579 + ];
580 +
581 + let mut conn = pool.acquire().await.unwrap();
582 + let skipped = pull::apply_changes_inner(&mut conn, changes).await.unwrap();
583 + assert_eq!(skipped, 1, "the malformed row is skipped, not fatal");
584 +
585 + let good: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM projects WHERE id = ?")
586 + .bind(&good_id)
587 + .fetch_one(&pool)
588 + .await
589 + .unwrap();
590 + assert_eq!(good.0, 1, "well-formed row in the same batch still applied");
591 + let bad: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM projects WHERE id = ?")
592 + .bind(&bad_id)
593 + .fetch_one(&pool)
594 + .await
595 + .unwrap();
596 + assert_eq!(bad.0, 0, "malformed row not applied");
597 + }
598 +
557 599 // -- Trigger suppression --
558 600
559 601 #[tokio::test]