From bb41d63286e1ca4871e7572dacd80cea1c560958 Mon Sep 17 00:00:00 2001 From: Andras Schmelczer Date: Tue, 22 Sep 2026 20:40:53 +0100 Subject: [PATCH] Implement protocol to notice server rollback --- .../migrations/20241207143519_bootstrap.sql | 1 + .../src/app_state/database/mutations.rs | 15 +- sync-server/src/app_state/database/queries.rs | 44 +++ sync-server/src/app_state/database/tests.rs | 270 ++++++++++++++++++ sync-server/src/server.rs | 17 +- sync-server/src/server/endpoints.rs | 1 + sync-server/src/server/history.rs | 62 ++++ 7 files changed, 402 insertions(+), 8 deletions(-) create mode 100644 sync-server/src/server/history.rs diff --git a/sync-server/src/app_state/database/migrations/20241207143519_bootstrap.sql b/sync-server/src/app_state/database/migrations/20241207143519_bootstrap.sql index e5a86edc..847ce041 100644 --- a/sync-server/src/app_state/database/migrations/20241207143519_bootstrap.sql +++ b/sync-server/src/app_state/database/migrations/20241207143519_bootstrap.sql @@ -1,5 +1,6 @@ CREATE TABLE events ( event_id INTEGER PRIMARY KEY AUTOINCREMENT CHECK(event_id <= 9007199254740991), -- JavaScript’s Number.MAX_SAFE_INTEGER + event_token TEXT, request_id TEXT NOT NULL UNIQUE, request_fingerprint BLOB NOT NULL ); diff --git a/sync-server/src/app_state/database/mutations.rs b/sync-server/src/app_state/database/mutations.rs index b0af9e4a..6585d102 100644 --- a/sync-server/src/app_state/database/mutations.rs +++ b/sync-server/src/app_state/database/mutations.rs @@ -10,14 +10,15 @@ impl Database { request_id: uuid::Uuid, fingerprint: &[u8], ) -> Result { - Ok( - sqlx::query("INSERT INTO events(request_id, request_fingerprint) VALUES (?, ?)") - .bind(request_id.to_string()) - .bind(fingerprint) - .execute(&mut **tx) - .await? - .last_insert_rowid(), + Ok(sqlx::query( + "INSERT INTO events(request_id, request_fingerprint, event_token) VALUES (?, ?, ?)", ) + .bind(request_id.to_string()) + .bind(fingerprint) + .bind(uuid::Uuid::new_v4().simple().to_string()) + .execute(&mut **tx) + .await? + .last_insert_rowid()) } pub async fn insert_document_version( diff --git a/sync-server/src/app_state/database/queries.rs b/sync-server/src/app_state/database/queries.rs index f345a2bc..162b0fe6 100644 --- a/sync-server/src/app_state/database/queries.rs +++ b/sync-server/src/app_state/database/queries.rs @@ -9,6 +9,8 @@ use anyhow::{Context as _, Result, ensure}; use crate::consts::MAX_EVENTS_PAGE_BYTES; +const EMPTY_HISTORY_CHECKPOINT: &str = "0:empty"; + impl Database { pub async fn get_missing_document( tx: &mut Transaction<'_>, @@ -236,3 +238,45 @@ impl Database { }) } } + +impl Database { + pub async fn history_checkpoint(&self, vault: &VaultId) -> Result { + let mut tx = self.create_readonly_transaction(vault).await?; + + let row: Option<(i64, String)> = sqlx::query_as( + "SELECT event_id, event_token FROM events ORDER BY event_id DESC LIMIT 1", + ) + .fetch_optional(&mut *tx) + .await?; + + Ok(row.map_or_else( + || EMPTY_HISTORY_CHECKPOINT.to_owned(), + |(id, token)| format!("{id}:{token}"), + )) + } + + pub async fn contains_checkpoint(&self, vault: &VaultId, checkpoint: &str) -> Result { + if checkpoint == EMPTY_HISTORY_CHECKPOINT { + return Ok(true); + } + + let Some((id, token)) = checkpoint.split_once(':') else { + return Ok(false); + }; + + let Ok(id) = id.parse::() else { + return Ok(false); + }; + + let mut tx = self.create_readonly_transaction(vault).await?; + let exists: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM events WHERE event_id = ? AND event_token = ?)", + ) + .bind(id) + .bind(token) + .fetch_one(&mut *tx) + .await?; + + Ok(exists) + } +} diff --git a/sync-server/src/app_state/database/tests.rs b/sync-server/src/app_state/database/tests.rs index 9b5b1e62..4ca365b4 100644 --- a/sync-server/src/app_state/database/tests.rs +++ b/sync-server/src/app_state/database/tests.rs @@ -1,4 +1,5 @@ use super::*; +use crate::consts::{DEFAULT_EVENTS_PAGE_SIZE, MAX_EVENTS_PAGE_BYTES}; use models::{DocumentId, FileManifest, StoredDocumentVersion}; fn config(directory: &tempfile::TempDir) -> DatabaseConfig { @@ -94,6 +95,32 @@ async fn rejects_invalid_vault_names_at_the_database_boundary() { assert!(database.connection_pools.lock().await.is_empty()); } +#[tokio::test] +async fn migrations_can_be_reapplied_without_losing_requests_or_content() { + let directory = tempfile::tempdir().unwrap(); + let config = config(&directory); + let database = Database::try_new(&config).await.unwrap(); + let vault = "migrations".to_owned(); + let version = insert_content(&database, &vault, uuid::Uuid::new_v4(), b"only copy").await; + let pool = database.get_connection_pool(&vault).await.unwrap(); + Database::run_migrations(&pool).await.unwrap(); + Database::run_migrations(&pool).await.unwrap(); + let reopened = Database::try_new(&config).await.unwrap(); + assert_eq!( + reopened + .get_document_version(&vault, version.vault_update_id, None) + .await + .unwrap() + .unwrap() + .content, + b"only copy" + ); + assert_eq!( + reopened.events_after(&vault, 0, DEFAULT_EVENTS_PAGE_SIZE).await.unwrap().events.len(), + 1 + ); +} + #[tokio::test] async fn snapshot_returns_latest_referenced_metadata_and_byte_lengths() { let directory = tempfile::tempdir().unwrap(); @@ -134,3 +161,246 @@ async fn snapshot_returns_latest_referenced_metadata_and_byte_lengths() { ); } } + +#[tokio::test] +async fn replay_pages_are_bounded_and_cover_every_event_in_order() { + let directory = tempfile::tempdir().unwrap(); + let database = Database::try_new(&config(&directory)).await.unwrap(); + let vault = "pages".to_owned(); + let id = uuid::Uuid::new_v4(); + for _ in 0..150 { + insert_content(&database, &vault, id, b"edit").await; + } + let mut after = 0; + while after < 150 { + let batch = database.events_after(&vault, after, DEFAULT_EVENTS_PAGE_SIZE).await.unwrap(); + assert!(batch.events.len() <= 64, "catchup must bound each page"); + assert!(!batch.events.is_empty()); + assert_eq!(batch.head_event_id, 150); + for event in batch.events { + assert_eq!(event.event_id, after + 1); + after = event.event_id; + } + } + assert!( + database + .events_after(&vault, after, DEFAULT_EVENTS_PAGE_SIZE) + .await + .unwrap() + .events + .is_empty() + ); +} + +#[tokio::test] +async fn idle_cleanup_does_not_wait_for_an_active_vault() { + let directory = tempfile::tempdir().unwrap(); + let database = Database::try_new(&config(&directory)).await.unwrap(); + database.vault_snapshot(&"other".to_owned()).await.unwrap(); + let vault = "slow".to_owned(); + let transaction = database.create_write_transaction(&vault).await.unwrap(); + database + .connection_pools + .lock() + .await + .get_mut(&vault) + .unwrap() + .last_accessed = Instant::now() - Duration::from_secs(301); + tokio::time::timeout(Duration::from_millis(200), database.cleanup_idle_pools()) + .await + .expect("cleanup must leave an active pool alone"); + let other = tokio::time::timeout( + Duration::from_millis(200), + database.vault_snapshot(&"other".to_owned()), + ) + .await + .expect("other vaults must remain accessible") + .unwrap(); + assert_eq!(other.head_event_id, 0); + assert!( + !database + .get_connection_pool(&vault) + .await + .unwrap() + .is_closed() + ); + drop(transaction); +} + +#[tokio::test] +async fn manifest_existence_checks_are_independent_of_payload_decoding() { + let directory = tempfile::tempdir().unwrap(); + let database = Database::try_new(&config(&directory)).await.unwrap(); + let vault = "existence".to_owned(); + let id = uuid::Uuid::new_v4(); + insert_content(&database, &vault, id, b"content").await; + let mut tx = database.create_write_transaction(&vault).await.unwrap(); + // Make full-row deserialization fail: checking a foreign identity must not + // decode unrelated version fields or read its potentially enormous blob. + sqlx::query("UPDATE documents SET updated_date = 'unreadable timestamp'") + .execute(&mut *tx) + .await + .unwrap(); + assert_eq!( + Database::missing_document(&mut tx, &[id]).await.unwrap(), + None + ); + let missing = uuid::Uuid::new_v4(); + assert_eq!( + Database::missing_document(&mut tx, &[id, missing]) + .await + .unwrap(), + Some(missing.to_string()) + ); + assert_eq!( + Database::missing_document(&mut tx, &[]).await.unwrap(), + None + ); +} + +#[tokio::test] +async fn replay_byte_budget_prevents_accumulating_large_historical_manifests() { + let directory = tempfile::tempdir().unwrap(); + let database = Database::try_new(&config(&directory)).await.unwrap(); + let vault = "large-manifests".to_owned(); + let mut tx = database.create_write_transaction(&vault).await.unwrap(); + // Each event fits independently, but fewer than 64 fit the response budget. + let entries: std::collections::BTreeMap<_, _> = (0..1500) + .map(|index| { + ( + uuid::Uuid::new_v4(), + format!("{index}-{}.md", "x".repeat(180)), + ) + }) + .collect(); + for _ in 0..8 { + let file_manifest_id = Database::allocate_event(&mut tx, uuid::Uuid::new_v4(), b"test") + .await + .unwrap(); + Database::insert_file_manifest( + &mut tx, + &FileManifest { + file_manifest_id, + entries: entries.clone(), + }, + ) + .await + .unwrap(); + } + tx.commit().await.unwrap(); + let batch = database.events_after(&vault, 0, DEFAULT_EVENTS_PAGE_SIZE).await.unwrap(); + assert!(!batch.events.is_empty()); + assert!(batch.events.len() < 8); + assert!(serde_json::to_vec(&batch).unwrap().len() < MAX_EVENTS_PAGE_BYTES + 1024); + assert_eq!( + batch.end_event_id, + batch.events.last().map(|event| event.event_id) + ); +} + +#[tokio::test] +async fn opening_a_locked_vault_does_not_block_unrelated_vaults() { + let directory = tempfile::tempdir().unwrap(); + let config = config(&directory); + let database = Database::try_new(&config).await.unwrap(); + database.vault_snapshot(&"other".to_owned()).await.unwrap(); + let vault = "locked-opening".to_owned(); + // A second process holds the writer lock while this process lazily checks + // migrations. Even already-open, unrelated vaults must remain available. + let external = Database::open_database(&config, &Database::database_path(&config, &vault)) + .await + .unwrap(); + let transaction = external.begin_with("BEGIN IMMEDIATE").await.unwrap(); + let opening_database = database.clone(); + let opening = tokio::spawn(async move { opening_database.get_connection_pool(&vault).await }); + tokio::time::sleep(Duration::from_millis(100)).await; + let opening_is_blocked = !opening.is_finished(); + let other = tokio::time::timeout( + Duration::from_millis(300), + database.vault_snapshot(&"other".to_owned()), + ) + .await; + transaction.rollback().await.unwrap(); + tokio::time::timeout(Duration::from_secs(5), opening) + .await + .unwrap() + .unwrap() + .unwrap(); + external.close().await; + assert!( + opening_is_blocked, + "the test must exercise an in-progress database open" + ); + assert_eq!( + other + .expect("a blocked migration must not lock every vault") + .unwrap() + .head_event_id, + 0 + ); +} + +#[tokio::test] +async fn concurrent_vault_opens_share_a_pool_and_cancelled_opens_can_retry() { + let directory = tempfile::tempdir().unwrap(); + let mut config = config(&directory); + config.max_connections_per_vault = 1; + let database = Database::try_new(&config).await.unwrap(); + let vault = "cancelled-opening".to_owned(); + let external = Database::open_database(&config, &Database::database_path(&config, &vault)) + .await + .unwrap(); + let transaction = external.begin_with("BEGIN IMMEDIATE").await.unwrap(); + let opener = database.clone(); + let name = vault.clone(); + let cancelled = tokio::spawn(async move { opener.get_connection_pool(&name).await }); + tokio::time::sleep(Duration::from_millis(50)).await; + assert!(!cancelled.is_finished()); + cancelled.abort(); + assert!(cancelled.await.unwrap_err().is_cancelled()); + transaction.rollback().await.unwrap(); + let mut tasks = Vec::new(); + for _ in 0..8 { + let database = database.clone(); + let vault = vault.clone(); + tasks.push(tokio::spawn(async move { + database.get_connection_pool(&vault).await.unwrap() + })); + } + let pools = tokio::time::timeout(Duration::from_secs(5), futures::future::join_all(tasks)) + .await + .unwrap(); + let pools: Vec<_> = pools.into_iter().map(Result::unwrap).collect(); + let checked_out = pools[0].acquire().await.unwrap(); + for pool in &pools[1..] { + assert!( + tokio::time::timeout(Duration::from_millis(25), pool.acquire()) + .await + .is_err(), + "all callers must share the single configured connection" + ); + } + drop(checked_out); + external.close().await; +} + +#[tokio::test] +async fn one_corrupt_vault_does_not_prevent_startup_or_healthy_vault_access() { + let directory = tempfile::tempdir().unwrap(); + let config = config(&directory); + let bad = Database::database_path(&config, "broken"); + tokio::fs::create_dir_all(bad.parent().unwrap()) + .await + .unwrap(); + tokio::fs::write(&bad, b"not a SQLite database") + .await + .unwrap(); + let database = Database::try_new(&config).await.unwrap(); + assert!(database.vault_snapshot(&"broken".to_owned()).await.is_err()); + let saved = insert_content(&database, "healthy", uuid::Uuid::new_v4(), b"safe").await; + assert_eq!(saved.content, b"safe"); + assert_eq!( + tokio::fs::read(bad).await.unwrap(), + b"not a SQLite database" + ); +} diff --git a/sync-server/src/server.rs b/sync-server/src/server.rs index 9eb66172..0a6145c6 100644 --- a/sync-server/src/server.rs +++ b/sync-server/src/server.rs @@ -1,7 +1,8 @@ pub mod auth; mod device_id_header; mod endpoints; -pub(crate) mod requests; +pub(crate) mod history; +mod requests; pub(crate) mod responses; #[cfg(test)] @@ -17,6 +18,7 @@ use axum::{ routing::{IntoMakeService, get}, }; use device_id_header::DEVICE_ID_HEADER_NAME; +use history::history_middleware; use log::info; use tokio::signal; use tower_http::{ @@ -63,6 +65,11 @@ pub async fn create_server(config: Config) -> Result<()> { http::header::CONTENT_TYPE, http::header::AUTHORIZATION, DEVICE_ID_HEADER_NAME.clone(), + http::header::HeaderName::from_static("x-vault-link-history"), + ]) + .expose_headers([ + http::header::HeaderName::from_static("x-vault-link-history"), + http::header::HeaderName::from_static("x-vault-link-history-mismatch"), ]) .allow_methods([Method::GET, Method::PUT]), ) @@ -113,10 +120,18 @@ fn get_authed_routes(app_state: AppState) -> Router { get(endpoints::fetch_latest_document_version::fetch_latest_document_version) .put(endpoints::put_file_content::put_file_content), ) + .route( + "/vaults/:vault_id/documents/:document_id/metadata", + get(endpoints::fetch_latest_document_metadata::fetch_latest_document_metadata), + ) .route( "/vaults/:vault_id/documents/:document_id/versions/:vault_update_id/content", get(endpoints::fetch_document_version_content::fetch_document_version_content), ) + .layer(middleware::from_fn_with_state( + app_state.clone(), + history_middleware, + )) .layer(middleware::from_fn_with_state(app_state, auth_middleware)) } diff --git a/sync-server/src/server/endpoints.rs b/sync-server/src/server/endpoints.rs index 23323a69..70724dc0 100644 --- a/sync-server/src/server/endpoints.rs +++ b/sync-server/src/server/endpoints.rs @@ -1,5 +1,6 @@ pub mod events; pub mod fetch_document_version_content; +pub mod fetch_latest_document_metadata; pub mod fetch_latest_document_version; pub mod get_file_manifest; pub mod index; diff --git a/sync-server/src/server/history.rs b/sync-server/src/server/history.rs new file mode 100644 index 00000000..167c5f1f --- /dev/null +++ b/sync-server/src/server/history.rs @@ -0,0 +1,62 @@ +use std::collections::HashMap; + +use axum::{ + extract::{Path, Request, State}, + http::{HeaderValue, StatusCode}, + middleware::Next, + response::{IntoResponse, Response}, +}; + +use crate::{ + app_state::{AppState, database::models::VaultId}, + errors::{SyncServerError, client_error, server_error}, + utils::normalize_vault_id::normalize_string, +}; + +pub async fn history_middleware( + State(state): State, + Path(path_params): Path>, + req: Request, + next: Next, +) -> Result { + let vault_id: VaultId = normalize_string( + path_params + .get("vault_id") + .ok_or_else(|| client_error(anyhow::anyhow!("Missing vault_id")))?, + ); + + // A restore happens while the server is stopped. Validate before executing + // any old request, including retries whose numeric parent IDs were reused. + if let Some(checkpoint) = req.headers().get("x-vault-link-history") { + let checkpoint = checkpoint + .to_str() + .map_err(|error| client_error(error.into()))?; + if !state + .database + .contains_checkpoint(&vault_id, checkpoint) + .await + .map_err(server_error)? + { + let mut response = (StatusCode::CONFLICT, "Server history changed").into_response(); + response.headers_mut().insert( + "x-vault-link-history-mismatch", + HeaderValue::from_static("1"), + ); + return Ok(response); + } + } + + let mut response = next.run(req).await; + if response.status().is_success() { + let checkpoint = state + .database + .history_checkpoint(&vault_id) + .await + .map_err(server_error)?; + response.headers_mut().insert( + "x-vault-link-history", + HeaderValue::from_str(&checkpoint).map_err(|error| server_error(error.into()))?, + ); + } + Ok(response) +} \ No newline at end of file