Implement protocol to notice server rollback

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

sync-server/src/app_state/database/mutations.rs

@@ -10,14 +10,15 @@ impl Database {
         request_id: uuid::Uuid,
         fingerprint: &[u8],
     ) -> Result<VaultUpdateId> {
-        Ok(
+        Ok(sqlx::query(
-            sqlx::query("INSERT INTO events(request_id, request_fingerprint) VALUES (?, ?)")
+            "INSERT INTO events(request_id, request_fingerprint, event_token) VALUES (?, ?, ?)",
-                .bind(request_id.to_string())
-                .bind(fingerprint)
-                .execute(&mut **tx)
-                .await?
-                .last_insert_rowid(),
         )
+        .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(

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<String> {
+        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<bool> {
+        if checkpoint == EMPTY_HISTORY_CHECKPOINT {
+            return Ok(true);
+        }
+
+        let Some((id, token)) = checkpoint.split_once(':') else {
+            return Ok(false);
+        };
+
+        let Ok(id) = id.parse::<i64>() 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)
+    }
+}

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

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<AppState> {
             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))
 }
 

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;

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<AppState>,
+    Path(path_params): Path<HashMap<String, String>>,
+    req: Request,
+    next: Next,
+) -> Result<Response, SyncServerError> {
+    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)
+}