fix(server): transactional sync push; single-use refresh rotation #136

Merged
funman300 merged 1 commits from fix/sync-auth-races into master 2026-07-07 03:04:41 +00:00
2 changed files with 44 additions and 31 deletions
Showing only changes of commit 8eb316751d - Show all commits
+12 -15
View File
@@ -283,23 +283,20 @@ pub async fn refresh(
// Tokens without jti predate rotation — require re-login. // Tokens without jti predate rotation — require re-login.
let jti = claims.jti.ok_or(AppError::Unauthorized)?; let jti = claims.jti.ok_or(AppError::Unauthorized)?;
// Verify this jti is still live (not yet consumed or from a deleted account). // Consume the old token before issuing new ones, gating on the DELETE
// SQLite TEXT columns are always nullable in sqlx; flatten the double-Option. // actually removing a row. rows_affected == 0 covers both "jti never
let exists: Option<String> = // existed / account deleted" and "a concurrent refresh already consumed
sqlx::query_scalar!("SELECT jti FROM refresh_tokens WHERE jti = ?", jti) // it" — the previous SELECT-then-DELETE let two concurrent refreshes
.fetch_optional(&state.pool) // both pass the check and both mint fresh token pairs. The DELETE is
.await? // the mutex: whoever removes the row wins; everyone else gets 401.
.flatten(); // If the insert below fails, the user loses this session (must
// re-login) — safe by design.
if exists.is_none() { let deleted = sqlx::query!("DELETE FROM refresh_tokens WHERE jti = ?", jti)
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)
.execute(&state.pool) .execute(&state.pool)
.await?; .await?;
if deleted.rows_affected() != 1 {
return Err(AppError::Unauthorized);
}
let new_access = make_access_token(&claims.sub, &state.jwt_secret)?; let new_access = make_access_token(&claims.sub, &state.jwt_secret)?;
let (new_refresh, new_jti) = make_refresh_token(&claims.sub, &state.jwt_secret)?; let (new_refresh, new_jti) = make_refresh_token(&claims.sub, &state.jwt_secret)?;
+32 -16
View File
@@ -5,7 +5,6 @@
use axum::{Json, extract::State}; use axum::{Json, extract::State};
use chrono::Utc; use chrono::Utc;
use sqlx::SqlitePool;
use uuid::Uuid; use uuid::Uuid;
use solitaire_sync::{ use solitaire_sync::{
@@ -26,13 +25,19 @@ struct SyncRow {
/// Load the stored `SyncPayload` for `user_id` from the database. /// Load the stored `SyncPayload` for `user_id` from the database.
/// Returns `None` if this user has not pushed any data yet. /// 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!( let row = sqlx::query_as!(
SyncRow, SyncRow,
"SELECT stats_json, achievements_json, progress_json FROM sync_state WHERE user_id = ?", "SELECT stats_json, achievements_json, progress_json FROM sync_state WHERE user_id = ?",
user_id user_id
) )
.fetch_optional(pool) .fetch_optional(exec)
.await?; .await?;
Ok(row) 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. /// Persist a `SyncPayload` for `user_id` using an upsert.
async fn store_payload( async fn store_payload(
pool: &SqlitePool, exec: impl sqlx::Executor<'_, Database = sqlx::Sqlite>,
user_id: &str, user_id: &str,
payload: &SyncPayload, payload: &SyncPayload,
) -> Result<(), AppError> { ) -> Result<(), AppError> {
@@ -92,7 +97,7 @@ async fn store_payload(
progress_json, progress_json,
now now
) )
.execute(pool) .execute(exec)
.await?; .await?;
Ok(()) Ok(())
@@ -159,12 +164,20 @@ pub async fn push(
return Err(AppError::BadRequest("user_id mismatch".into())); 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)?, Some(row) => row_to_payload(&row, &user.user_id)?,
None => { None => {
// First push — nothing to merge against; store directly. // First push — nothing to merge against; store directly.
store_payload(&state.pool, &user.user_id, &client_payload).await?; store_payload(&mut *tx, &user.user_id, &client_payload).await?;
update_leaderboard_if_opted_in(&state.pool, &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 { return Ok(Json(SyncResponse {
merged: client_payload, merged: client_payload,
server_time: Utc::now(), server_time: Utc::now(),
@@ -175,8 +188,9 @@ pub async fn push(
let (merged, conflicts) = merge(&client_payload, &server_payload); let (merged, conflicts) = merge(&client_payload, &server_payload);
store_payload(&state.pool, &user.user_id, &merged).await?; store_payload(&mut *tx, &user.user_id, &merged).await?;
update_leaderboard_if_opted_in(&state.pool, &user.user_id, &merged).await?; update_leaderboard_if_opted_in(&mut tx, &user.user_id, &merged).await?;
tx.commit().await?;
Ok(Json(SyncResponse { Ok(Json(SyncResponse {
merged, merged,
@@ -188,16 +202,18 @@ pub async fn push(
/// If the user is opted in to the leaderboard, update their row with the /// 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`. /// 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 /// Runs on the caller's transaction connection, so the opt-in check and the
/// conditional UPDATE (WHERE EXISTS subquery) to avoid a TOCTOU race where /// update are atomic with the surrounding push — an opt-out between the check
/// the user opts out between the check and the write. /// 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( async fn update_leaderboard_if_opted_in(
pool: &SqlitePool, conn: &mut sqlx::SqliteConnection,
user_id: &str, user_id: &str,
payload: &SyncPayload, payload: &SyncPayload,
) -> Result<(), AppError> { ) -> Result<(), AppError> {
let opted_in = sqlx::query!("SELECT leaderboard_opt_in FROM users WHERE id = ?", user_id) let opted_in = sqlx::query!("SELECT leaderboard_opt_in FROM users WHERE id = ?", user_id)
.fetch_optional(pool) .fetch_optional(&mut *conn)
.await? .await?
.map(|r| r.leaderboard_opt_in) .map(|r| r.leaderboard_opt_in)
.unwrap_or(0); .unwrap_or(0);
@@ -231,7 +247,7 @@ async fn update_leaderboard_if_opted_in(
now, now,
user_id user_id
) )
.execute(pool) .execute(&mut *conn)
.await?; .await?;
Ok(()) Ok(())