From ac30565f08d63ac4392827e3e7ee65748f44742f Mon Sep 17 00:00:00 2001 From: Alexey Dubovskoy Date: Tue, 14 Apr 2026 20:53:26 +0100 Subject: [PATCH] fix(client): advance download watermark atomically per batch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The incremental download watermark was derived from `max(file_records.jid)`, which advances every time a single file is saved. Combined with the server's `list(jid)` filter — which hides any path whose latest commit id is <= `jid` — this permanently skips files if a batch is interrupted partway. If the client saved a high-jid file (say id=500) and then crashed before saving a lower-jid file from the same batch (id=300), the next poll requested `list(500)`, got nothing for id=300, and the file was invisible forever. Move the watermark into a dedicated `sync_state` table and advance it only once per batch, inside a DB transaction that also writes all `file_records` rows. Either everything lands or nothing does; a partial crash leaves the watermark pointing at the previous safe value so the whole batch is re-requested on the next poll (chunks already on disk are reused via the local cache). Fresh installs and upgrades start with watermark=0, triggering one full re-sync. That also heals any gaps that accumulated under the old mechanism. --- .../2026-04-14-000000_add_sync_state/down.sql | 1 + .../2026-04-14-000000_add_sync_state/up.sql | 8 ++ client/src/registry.rs | 123 ++++++++++++++++++ client/src/schema.rs | 9 ++ client/src/syncer.rs | 45 +++++-- 5 files changed, 177 insertions(+), 9 deletions(-) create mode 100644 client/migrations/2026-04-14-000000_add_sync_state/down.sql create mode 100644 client/migrations/2026-04-14-000000_add_sync_state/up.sql diff --git a/client/migrations/2026-04-14-000000_add_sync_state/down.sql b/client/migrations/2026-04-14-000000_add_sync_state/down.sql new file mode 100644 index 0000000..39d875c --- /dev/null +++ b/client/migrations/2026-04-14-000000_add_sync_state/down.sql @@ -0,0 +1 @@ +DROP TABLE sync_state; diff --git a/client/migrations/2026-04-14-000000_add_sync_state/up.sql b/client/migrations/2026-04-14-000000_add_sync_state/up.sql new file mode 100644 index 0000000..e3b7159 --- /dev/null +++ b/client/migrations/2026-04-14-000000_add_sync_state/up.sql @@ -0,0 +1,8 @@ +-- Per-namespace download watermark. Separated from file_records so it can +-- be advanced atomically after a full download batch completes, instead of +-- being derived from max(file_records.jid) which drifts forward whenever +-- any single file is saved. See fix/download-watermark for context. +CREATE TABLE sync_state ( + namespace_id INTEGER PRIMARY KEY NOT NULL, + download_watermark INTEGER NOT NULL DEFAULT 0 +); diff --git a/client/src/registry.rs b/client/src/registry.rs index a86d688..f054e6c 100644 --- a/client/src/registry.rs +++ b/client/src/registry.rs @@ -88,3 +88,126 @@ pub fn latest_jid(conn: &mut Connection, namespace_id: i32) -> Result { Err(e) => Err(e), } } + +/// Read the incremental-download watermark for a namespace. +/// +/// The watermark is advanced atomically only after a full download batch +/// completes (see `syncer::check_download_once`). This prevents the bug +/// that arises when a per-file advancing watermark (e.g. `max(jid)` across +/// `file_records`) drifts past lower-jid files that the client hasn't yet +/// persisted — the server's `list(jid)` filter then hides those files +/// forever. +/// +/// Returns 0 if no row exists for `namespace_id`, which triggers a full +/// initial sync — the safe default on first use and after a schema upgrade. +pub fn get_download_watermark(conn: &mut Connection, namespace_id: i32) -> Result { + trace!("get_download_watermark ns={}", namespace_id); + + let r = sync_state::table + .filter(sync_state::namespace_id.eq(namespace_id)) + .select(sync_state::download_watermark) + .first::(conn) + .optional()?; + + Ok(r.unwrap_or(0)) +} + +/// Advance the incremental-download watermark for a namespace. +/// +/// Callers must only invoke this after every file in the current batch has +/// been persisted to the local registry; otherwise a subsequent incremental +/// request will skip still-missing files (see `get_download_watermark`). +pub fn set_download_watermark( + conn: &mut Connection, + namespace_id: i32, + value: i32, +) -> Result { + trace!("set_download_watermark ns={} value={}", namespace_id, value); + + // Upsert: SQLite ON CONFLICT on the PRIMARY KEY updates the row in place. + insert_into(sync_state::table) + .values(( + sync_state::namespace_id.eq(namespace_id), + sync_state::download_watermark.eq(value), + )) + .on_conflict(sync_state::namespace_id) + .do_update() + .set(sync_state::download_watermark.eq(value)) + .execute(conn) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::connection::{get_connection, get_connection_pool}; + use tempfile::TempDir; + + fn setup() -> (TempDir, crate::connection::ConnectionPool) { + let tmp = TempDir::new().unwrap(); + let db_path = tmp.path().join("test.db"); + let pool = get_connection_pool(db_path.to_str().unwrap()).unwrap(); + (tmp, pool) + } + + #[test] + fn given_no_row_when_get_download_watermark_then_returns_zero() { + // Given + let (_tmp, pool) = setup(); + let mut conn = get_connection(&pool).unwrap(); + + // When + let watermark = get_download_watermark(&mut conn, 1).unwrap(); + + // Then + assert_eq!(watermark, 0); + } + + #[test] + fn given_set_watermark_when_get_download_watermark_then_returns_stored_value() { + // Given + let (_tmp, pool) = setup(); + let mut conn = get_connection(&pool).unwrap(); + set_download_watermark(&mut conn, 1, 42).unwrap(); + + // When + let watermark = get_download_watermark(&mut conn, 1).unwrap(); + + // Then + assert_eq!(watermark, 42); + } + + #[test] + fn given_existing_value_when_set_download_watermark_then_overwrites_without_duplicating() { + // Given + let (_tmp, pool) = setup(); + let mut conn = get_connection(&pool).unwrap(); + set_download_watermark(&mut conn, 1, 100).unwrap(); + + // When + set_download_watermark(&mut conn, 1, 200).unwrap(); + + // Then + assert_eq!(get_download_watermark(&mut conn, 1).unwrap(), 200); + let row_count: i64 = sync_state::table + .filter(sync_state::namespace_id.eq(1)) + .count() + .get_result(&mut *conn) + .unwrap(); + assert_eq!(row_count, 1); + } + + #[test] + fn given_different_namespaces_when_set_download_watermark_then_values_isolated() { + // Given + let (_tmp, pool) = setup(); + let mut conn = get_connection(&pool).unwrap(); + + // When + set_download_watermark(&mut conn, 1, 100).unwrap(); + set_download_watermark(&mut conn, 2, 200).unwrap(); + + // Then + assert_eq!(get_download_watermark(&mut conn, 1).unwrap(), 100); + assert_eq!(get_download_watermark(&mut conn, 2).unwrap(), 200); + } +} diff --git a/client/src/schema.rs b/client/src/schema.rs index 24392c4..899bd2b 100644 --- a/client/src/schema.rs +++ b/client/src/schema.rs @@ -11,3 +11,12 @@ diesel::table! { namespace_id -> Integer, } } + +diesel::table! { + sync_state (namespace_id) { + namespace_id -> Integer, + download_watermark -> Integer, + } +} + +diesel::allow_tables_to_appear_in_same_query!(file_records, sync_state,); diff --git a/client/src/syncer.rs b/client/src/syncer.rs index b551317..916c624 100644 --- a/client/src/syncer.rs +++ b/client/src/syncer.rs @@ -254,7 +254,13 @@ pub async fn check_download_once( let conn = &mut get_connection(pool)?; - let latest_local = registry::latest_jid(conn, namespace_id).unwrap_or(0); + // The download watermark is read from a dedicated `sync_state` row and + // only advanced after every file in the batch has been persisted. Using + // `max(file_records.jid)` here would advance per-file and — because the + // server's `list(jid)` filter hides any path whose latest commit id is + // at or below `jid` — permanently skip files if a prior batch was + // interrupted partway (e.g. process killed, network drop). + let latest_local = registry::get_download_watermark(conn, namespace_id).unwrap_or(0); let to_download = remote.list(latest_local).await?; // TODO maybe should limit one download at a time and use batches // it can also overflow in-memory cache @@ -300,32 +306,53 @@ pub async fn check_download_once( } } + // Apply per-file disk changes first. Any failure here bails before we + // touch the DB, so the watermark stays put and the next poll retries + // the whole batch (chunks already written are reused via the cache). + let mut create_forms: Vec = Vec::new(); + let mut delete_forms: Vec = Vec::new(); + for d in &to_download { - trace!("udpating downloaded files {:?}", d); + trace!("updating downloaded files {:?}", d); let mut chunker = chunker.lock().await; if d.deleted { - let form = build_delete_form(&d.path, storage_path, d.id, namespace_id); - // TODO atomic? - registry::delete(conn, &vec![form])?; if chunker.exists(&d.path) { chunker.delete(&d.path).await?; } + delete_forms.push(build_delete_form(&d.path, storage_path, d.id, namespace_id)); } else { let chunks: Vec<&str> = d.chunk_ids.split(',').collect(); - // TODO atomic? store in tmp first and then move? - // TODO should be after we create record in db if let Err(e) = chunker.save(&d.path, chunks).await { error!("{:?}", e); return Err(e); } - let form = build_file_record(&d.path, storage_path, d.id, namespace_id)?; - registry::create(conn, &vec![form])?; + create_forms.push(build_file_record(&d.path, storage_path, d.id, namespace_id)?); } } + // Atomically persist all file_record writes for this batch together + // with the advanced watermark. Either all land or none: a partial + // crash after this point still leaves a consistent `(file_records, + // sync_state)` pair, and a crash before this point leaves the + // watermark untouched so the next poll re-requests the same batch. + if let Some(max_id) = to_download.iter().map(|d| d.id).max() { + use diesel::connection::Connection as _; + + conn.transaction::<_, diesel::result::Error, _>(|tx_conn| { + if !create_forms.is_empty() { + registry::create(tx_conn, &create_forms)?; + } + if !delete_forms.is_empty() { + registry::delete(tx_conn, &delete_forms)?; + } + registry::set_download_watermark(tx_conn, namespace_id, max_id)?; + Ok(()) + })?; + } + Ok(!to_download.is_empty()) }