fix(client): advance download watermark atomically per batch

client/migrations/2026-04-14-000000_add_sync_state/down.sql

@@ -0,0 +1 @@
+DROP TABLE sync_state;

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
+);

client/src/registry.rs

@@ -88,3 +88,126 @@ pub fn latest_jid(conn: &mut Connection, namespace_id: i32) -> Result<i32> {
         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<i32> {
+    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::<i32>(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<usize> {
+    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);
+    }
+}

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,);

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<models::CreateForm> = Vec::new();
+    let mut delete_forms: Vec<models::DeleteForm> = 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())
 }