Merge pull request 'fix(server): transactional sync push; single-use refresh rotation' (#136) from fix/sync-auth-races into master

Two concurrency fixes from the 2026-07-06 review (findings M1, M2):

sync push (M1): the load→merge→store cycle ran as three separate DB
operations. Two devices pushing concurrently both read the same stored
payload, merged independently, and the second store overwrote the first
merge — the server visibly regressed until the losing device pushed
again. The whole cycle (including the leaderboard update) now runs in
one transaction; SQLite serialises the writers. The leaderboard
helper's stale docstring (claiming a single conditional UPDATE that was
actually two statements) is corrected — the transaction now provides
the atomicity it described.

refresh rotation (M2): SELECT-then-DELETE let two concurrent refreshes
with the same token both pass the liveness check and both mint fresh
token pairs. The SELECT is gone; rotation now gates on the DELETE's
rows_affected — whoever removes the jti row wins, everyone else gets
401. Sequential reuse was already covered by
consumed_refresh_token_is_rejected, which still passes unchanged.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

solitaire_server/src/auth.rs

@@ -283,23 +283,20 @@ pub async fn refresh(
     // Tokens without jti predate rotation — require re-login.
     let jti = claims.jti.ok_or(AppError::Unauthorized)?;
 
-    // Verify this jti is still live (not yet consumed or from a deleted account).
-    // SQLite TEXT columns are always nullable in sqlx; flatten the double-Option.
-    let exists: Option<String> =
-        sqlx::query_scalar!("SELECT jti FROM refresh_tokens WHERE jti = ?", jti)
-            .fetch_optional(&state.pool)
-            .await?
-            .flatten();
-
-    if exists.is_none() {
-        return Err(AppError::Unauthorized);
-    }
-
-    // Consume the old token before issuing new ones. If the insert below
-    // fails, the user loses this session (must re-login) — safe by design.
-    sqlx::query!("DELETE FROM refresh_tokens WHERE jti = ?", jti)
+    // Consume the old token before issuing new ones, gating on the DELETE
+    // actually removing a row. rows_affected == 0 covers both "jti never
+    // existed / account deleted" and "a concurrent refresh already consumed
+    // it" — the previous SELECT-then-DELETE let two concurrent refreshes
+    // both pass the check and both mint fresh token pairs. The DELETE is
+    // the mutex: whoever removes the row wins; everyone else gets 401.
+    // If the insert below fails, the user loses this session (must
+    // re-login) — safe by design.
+    let deleted = sqlx::query!("DELETE FROM refresh_tokens WHERE jti = ?", jti)
         .execute(&state.pool)
         .await?;
+    if deleted.rows_affected() != 1 {
+        return Err(AppError::Unauthorized);
+    }
 
     let new_access = make_access_token(&claims.sub, &state.jwt_secret)?;
     let (new_refresh, new_jti) = make_refresh_token(&claims.sub, &state.jwt_secret)?;

solitaire_server/src/sync.rs

@@ -5,7 +5,6 @@
 
 use axum::{Json, extract::State};
 use chrono::Utc;
-use sqlx::SqlitePool;
 use uuid::Uuid;
 
 use solitaire_sync::{
@@ -26,13 +25,19 @@ struct SyncRow {
 
 /// Load the stored `SyncPayload` for `user_id` from the database.
 /// Returns `None` if this user has not pushed any data yet.
-async fn load_sync_row(pool: &SqlitePool, user_id: &str) -> Result<Option<SyncRow>, AppError> {
+///
+/// Executor-generic so `push` can run it inside its transaction while
+/// `pull` keeps passing the pool directly.
+async fn load_sync_row(
+    exec: impl sqlx::Executor<'_, Database = sqlx::Sqlite>,
+    user_id: &str,
+) -> Result<Option<SyncRow>, AppError> {
     let row = sqlx::query_as!(
         SyncRow,
         "SELECT stats_json, achievements_json, progress_json FROM sync_state WHERE user_id = ?",
         user_id
     )
-    .fetch_optional(pool)
+    .fetch_optional(exec)
     .await?;
     Ok(row)
 }
@@ -69,7 +74,7 @@ fn row_to_payload(row: &SyncRow, user_id: &str) -> Result<SyncPayload, AppError>
 
 /// Persist a `SyncPayload` for `user_id` using an upsert.
 async fn store_payload(
-    pool: &SqlitePool,
+    exec: impl sqlx::Executor<'_, Database = sqlx::Sqlite>,
     user_id: &str,
     payload: &SyncPayload,
 ) -> Result<(), AppError> {
@@ -92,7 +97,7 @@ async fn store_payload(
         progress_json,
         now
     )
-    .execute(pool)
+    .execute(exec)
     .await?;
 
     Ok(())
@@ -159,12 +164,20 @@ pub async fn push(
         return Err(AppError::BadRequest("user_id mismatch".into()));
     }
 
-    let server_payload = match load_sync_row(&state.pool, &user.user_id).await? {
+    // The whole read-merge-write cycle runs in ONE transaction. Without it,
+    // two devices pushing concurrently both read the same stored payload,
+    // merge independently, and the second store overwrites the first merge —
+    // the server visibly regresses until the losing device pushes again.
+    // SQLite serialises writers, so the second transaction simply waits.
+    let mut tx = state.pool.begin().await?;
+
+    let server_payload = match load_sync_row(&mut *tx, &user.user_id).await? {
         Some(row) => row_to_payload(&row, &user.user_id)?,
         None => {
             // First push — nothing to merge against; store directly.
-            store_payload(&state.pool, &user.user_id, &client_payload).await?;
-            update_leaderboard_if_opted_in(&state.pool, &user.user_id, &client_payload).await?;
+            store_payload(&mut *tx, &user.user_id, &client_payload).await?;
+            update_leaderboard_if_opted_in(&mut tx, &user.user_id, &client_payload).await?;
+            tx.commit().await?;
             return Ok(Json(SyncResponse {
                 merged: client_payload,
                 server_time: Utc::now(),
@@ -175,8 +188,9 @@ pub async fn push(
 
     let (merged, conflicts) = merge(&client_payload, &server_payload);
 
-    store_payload(&state.pool, &user.user_id, &merged).await?;
-    update_leaderboard_if_opted_in(&state.pool, &user.user_id, &merged).await?;
+    store_payload(&mut *tx, &user.user_id, &merged).await?;
+    update_leaderboard_if_opted_in(&mut tx, &user.user_id, &merged).await?;
+    tx.commit().await?;
 
     Ok(Json(SyncResponse {
         merged,
@@ -188,16 +202,18 @@ pub async fn push(
 /// If the user is opted in to the leaderboard, update their row with the
 /// better of the stored and incoming `best_single_score` / `fastest_win_seconds`.
 ///
-/// The opt-in check and the update are performed atomically in a single
-/// conditional UPDATE (WHERE EXISTS subquery) to avoid a TOCTOU race where
-/// the user opts out between the check and the write.
+/// Runs on the caller's transaction connection, so the opt-in check and the
+/// update are atomic with the surrounding push — an opt-out between the check
+/// and the write can no longer interleave. (An earlier doc comment claimed
+/// this was a single conditional UPDATE; it has always been two statements —
+/// the enclosing transaction is what actually provides the atomicity.)
 async fn update_leaderboard_if_opted_in(
-    pool: &SqlitePool,
+    conn: &mut sqlx::SqliteConnection,
     user_id: &str,
     payload: &SyncPayload,
 ) -> Result<(), AppError> {
     let opted_in = sqlx::query!("SELECT leaderboard_opt_in FROM users WHERE id = ?", user_id)
-        .fetch_optional(pool)
+        .fetch_optional(&mut *conn)
         .await?
         .map(|r| r.leaderboard_opt_in)
         .unwrap_or(0);
@@ -231,7 +247,7 @@ async fn update_leaderboard_if_opted_in(
         now,
         user_id
     )
-    .execute(pool)
+    .execute(&mut *conn)
     .await?;
 
     Ok(())