Mutation cursor ref reconciliation

cli/Cargo.toml

@@ -65,6 +65,7 @@ sha2 = "0.11"
 
 [dev-dependencies]
 quint-connect = "0.1.2"
+tempfile = "3"
 
 [lints]
 workspace = true

cli/src/services/db/mod.rs

@@ -486,6 +486,37 @@ fn resolve_query_retry_policy<M: DbSpec>() -> RetryPolicy {
     QUERY_RETRY_POLICY
 }
 
+#[cfg(test)]
+thread_local! {
+    static READ_STATEMENTS_ISSUED: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
+}
+
+/// Record that one [`TursoDb`] read statement was issued on this thread.
+#[cfg(test)]
+fn note_read_statement_issued() {
+    READ_STATEMENTS_ISSUED.with(|count| count.set(count.get() + 1));
+}
+
+/// Run `body`, returning its result together with the number of [`TursoDb`]
+/// read statements ([`TursoDb::query`], [`TursoDb::query_values`],
+/// [`TursoDb::query_map`]) it issued on the current thread.
+///
+/// Each read method bumps the counter once in its synchronous prelude, before
+/// the retry wrapper, so a transient retry never inflates the count and the
+/// number reflects *logical* read statements, not connection round-trips.
+/// Lets a deterministic single-threaded test assert that an operation which
+/// must observe one coherent database snapshot — for example
+/// `MutationTraceStore::load_all_tree_roots`, a single `UNION` statement —
+/// issues exactly one, and fail if it is ever reimplemented as several
+/// independent `SELECT`s unioned in Rust. Not shared across threads.
+#[cfg(test)]
+pub(crate) fn count_read_statements<T>(body: impl FnOnce() -> T) -> (T, usize) {
+    READ_STATEMENTS_ISSUED.with(|count| count.set(0));
+    let result = body();
+    let issued = READ_STATEMENTS_ISSUED.with(std::cell::Cell::get);
+    (result, issued)
+}
+
 /// Generic Turso database adapter.
 ///
 /// Wraps a Turso connection with a tokio current-thread runtime so callers can
@@ -645,6 +676,9 @@ impl<M: DbSpec> TursoDb<M> {
         })?;
         let operation_name = format!("query {} database", M::db_name());
 
+        #[cfg(test)]
+        note_read_statement_issued();
+
         run_with_retry_sync(
             resolve_query_retry_policy::<M>(),
             &operation_name,
@@ -673,6 +707,9 @@ impl<M: DbSpec> TursoDb<M> {
         })?;
         let operation_name = format!("query and fetch {} database values", M::db_name());
 
+        #[cfg(test)]
+        note_read_statement_issued();
+
         run_with_retry_sync(
             resolve_query_retry_policy::<M>(),
             &operation_name,
@@ -810,6 +847,9 @@ impl<M: DbSpec> TursoDb<M> {
         })?;
         let operation_name = format!("query and fetch {} database rows", M::db_name());
 
+        #[cfg(test)]
+        note_read_statement_issued();
+
         let rows = run_with_retry_sync(
             resolve_query_retry_policy::<M>(),
             &operation_name,

cli/src/services/mutation_trace/runtime/coordinator.rs

@@ -179,19 +179,28 @@ pub fn coordinate<P>(
 where
     P: FnOnce() -> anyhow::Result<RepositoryAgentTraceDb>,
 {
-    coordinate_inner(repository_root, boundary, open_db, || {}, |_attempt| Ok(()))
+    coordinate_inner(
+        repository_root,
+        boundary,
+        open_db,
+        || {},
+        |_attempt| {},
+        |_attempt| Ok(()),
+    )
 }
 
-fn coordinate_inner<P, F, R>(
+pub(super) fn coordinate_inner<P, F, L, R>(
     repository_root: &Path,
     boundary: &RuntimeBoundary,
     open_db: P,
     on_lock_contention: F,
+    after_load: L,
     after_recovery: R,
 ) -> Result<CoordinateOutcome, CoordinateError>
 where
     P: FnOnce() -> anyhow::Result<RepositoryAgentTraceDb>,
     F: FnOnce(),
+    L: FnMut(u32),
     R: FnMut(u32) -> Result<()>,
 {
     let git_dir = resolve_git_dir(repository_root).map_err(CoordinateError::Other)?;
@@ -220,6 +229,7 @@ where
         boundary,
         open_db,
         inherited_external_taint,
+        after_load,
         after_recovery,
     )?;
 
@@ -232,16 +242,18 @@ where
     }
 }
 
-fn coordinate_protected<P, R>(
+fn coordinate_protected<P, L, R>(
     repository_root: &Path,
     git_dir: &Path,
     boundary: &RuntimeBoundary,
     open_db: P,
     inherited_external_taint: bool,
+    after_load: L,
     after_recovery: R,
 ) -> Result<CoordinateOutcome, CoordinateError>
 where
     P: FnOnce() -> anyhow::Result<RepositoryAgentTraceDb>,
+    L: FnMut(u32),
     R: FnMut(u32) -> Result<()>,
 {
     let checkout_id = get_or_create_checkout_id(git_dir).map_err(CoordinateError::Other)?;
@@ -257,7 +269,7 @@ where
         &worktree_id,
         boundary,
         inherited_external_taint,
-        |_attempt| {},
+        after_load,
         after_recovery,
     )
 }
@@ -1510,6 +1522,7 @@ mod tests {
                         .send(())
                         .expect("contention signal channel should still be open");
                 },
+                |_attempt| {},
                 |_attempt| Ok(()),
             );
             result_tx
@@ -1719,6 +1732,7 @@ mod tests {
                             .send(())
                             .expect("contention signal channel should still be open");
                     },
+                    |_attempt| {},
                     |_attempt| Ok(()),
                 )
             })
@@ -2085,6 +2099,7 @@ mod tests {
             },
             ok_db,
             || {},
+            |_attempt| {},
             |_attempt| {
                 anyhow::bail!("injected failure after recovery, before the boundary commits")
             },