power loss

sync-server/README.md

@@ -1,22 +1,23 @@
 # Sync server
 
-## API v4
+## API v5
 
 The server is a compare-and-swap store. Content versions do not contain paths or
 deletions. A separately versioned file manifest owns the namespace and membership.
 All routes below are under `/vaults/:vault_id` and require bearer authentication
 (except the WebSocket, which authenticates its first message).
 
-| Endpoint | Purpose |
-| --- | --- |
-| `PUT /documents/:document_id` | CAS an immutable content version. |
-| `GET /documents/:document_id` | Current content version, including base64 bytes. |
-| `GET /documents/:document_id/versions/:version_id/content` | Immutable raw bytes. |
-| `GET /file_manifest` | Current file manifest. |
-| `PUT /file_manifest` | CAS the complete file manifest. |
-| `GET /vault_snapshot` | File manifest, referenced content heads, and event watermark from one read transaction. |
-| `GET /events_since?after=N` | Every event after N, in order, with the current watermark. |
-| `GET /ws` | Ordered event delivery and ephemeral cursor positions. |
+| Endpoint                                                   | Purpose                                                                                 |
+| ---------------------------------------------------------- | --------------------------------------------------------------------------------------- |
+| `PUT /documents/:document_id`                              | CAS an immutable content version.                                                       |
+| `GET /documents/:document_id/metadata`                     | Current version metadata and size without content bytes.                                |
+| `GET /documents/:document_id`                              | Current content version, including base64 bytes.                                        |
+| `GET /documents/:document_id/versions/:version_id/content` | Immutable raw bytes.                                                                    |
+| `GET /file-manifest`                                       | Current file manifest.                                                                  |
+| `PUT /file-manifest`                                       | CAS the complete file manifest.                                                         |
+| `GET /vault-snapshot`                                      | File manifest, referenced content heads, and event watermark from one read transaction. |
+| `GET /events-since?after=N`                                | Every event after N, in order, with the current watermark.                              |
+| `GET /ws`                                                  | Ordered event delivery and ephemeral cursor positions.                                  |
 
 A content push includes a `Device-Id` header and this JSON body:
 
@@ -54,13 +55,14 @@ UUID-to-path mapping in `file_manifest_entries`. Each version has a header row
 even when its entries are empty. Entries are unique by document ID and exact path
 within a version; portable path validation also rejects aliases and directory
 conflicts. Reads reconstruct the complete map, and vault snapshots join these
-rows to the latest content versions. The version, entries, and complete file
-manifest event in `events.event_json` commit atomically. Event replay and retry
-acknowledgements continue to use `event_json`.
+rows to the latest content versions. Event replay and retry acknowledgements are
+also reconstructed from normalized content/manifest rows, without storing a
+second complete map in the event table.
 
 Both pushes return a tagged `Accepted` or `StaleBase` response. An accepted file
 manifest response returns its assigned file manifest ID. A stale content response
-includes bytes; a stale file manifest response includes the complete map.
+includes metadata and byte length only; clients fetch immutable bytes after applying
+their size and exclusion policy. A stale file manifest response includes the complete map.
 Accepted requests store their unique request ID and fingerprint in the event row,
 atomically with the version. Retrying the exact request reconstructs the original
 `Accepted` response from that event before checking the current head, even after
@@ -70,8 +72,8 @@ identity is not part of the fingerprint.
 
 File manifest validation rejects duplicate UUID keys (including aliases), missing
 content, path aliases, inconsistent directory spelling, and file/ancestor
-conflicts. Paths use `/`, NFC normalization, at most 240 UTF-8 bytes, and portable
-Windows filename restrictions. Alias comparison is NFC → Unicode uppercase →
+conflicts. Paths use `/`, NFC normalization, and portable Windows filename
+restrictions without imposing a universal total-path byte limit. Alias comparison is NFC → Unicode uppercase →
 NFC. `.vault-link-sync` and its aliases are reserved. The server never sanitizes
 or allocates a replacement name.
 
@@ -87,11 +89,19 @@ determine event ordering.
 The WebSocket handshake retains `lastSeenVaultUpdateId`. The server sends
 `{type: "vaultEvents", headEventId, events}` in increasing order, including events
 originated by that client. Each connection drains the database through one sender.
-In-memory notifications only wake it; polling covers missed wakes and cancellation
-after commit. Reconnect with the last durably processed event ID. The HTTP replay
-endpoint uses the same history, so losing the final notification is recoverable.
-A fresh client reads `/vault_snapshot`, applies it durably, then replays after its
-watermark.
+In-memory notifications only wake it; polling covers missed wakes and requests
+interrupted after commit. Reconnect with the last durably processed event ID. The
+HTTP replay endpoint uses the same history, so losing the final notification is
+recoverable.
+A fresh client reads `/vault-snapshot`, applies it durably, then replays after its
+watermark. Successful HTTP responses include `X-Vault-Link-History`, containing
+an event ID and a random event-incarnation token. Clients persist and echo this
+checkpoint on later HTTP requests. The server returns HTTP 409 with
+`X-Vault-Link-History-Mismatch: 1` before executing a request whose checkpoint is
+absent from the current database. Restored backups retain their original prefix;
+new events get new tokens even when numeric IDs or request UUIDs are reused.
+Restore databases while the server is stopped. Live replacement of open SQLite
+files is not supported.
 Cursor notifications are ephemeral and do not consume event IDs.
 Cursor expiry and disconnect broadcast the remaining clients, including an empty
 list when the last client leaves.
@@ -116,8 +126,20 @@ flushes database directory entries; the filesystem must support those durability
 operations. Content, file manifests, and events (including request fingerprints)
 are retained indefinitely.
 
-API v4 requires fresh vault databases and upgraded clients. Existing database
-migration checksums intentionally reject earlier schemas; there is no automatic
-erasure or migration. Use a new database directory and preserve old data.
+Deploy API v5 clients and server together. The v4 database schema migrates by
+adding event-incarnation tokens without deleting content or receipts. Schemas
+from before v4 still require a separate database directory and explicit recovery.
+Vault databases open and migrate independently on first access; one corrupt or
+incompatible vault does not prevent healthy vaults from serving requests.
+
+Diffs are limited to 10,000 items and reconstructed in one pass over Unicode
+scalars on a blocking worker, before taking the SQLite writer transaction. The
+transaction rechecks request identity and the parent version before committing.
+Clients send a snapshot when an edit exceeds the diff-operation budget. Decoded
+content is bounded by `max_body_size_mb` as well as the encoded request limit.
+
+A missing configuration file is initialized with durably stored credentials.
+Existing configurations must explicitly supply `users.user_configs`; omitted
+credentials are rejected instead of generating an unpersisted token at startup.
 Base64 increases snapshot body size by about one third; request limits apply to
 the encoded JSON body.

sync-server/src/app_state/cursors.rs

@@ -2,6 +2,7 @@ use core::time::Duration;
 use std::{collections::HashMap, sync::Arc};
 
 use tokio::sync::Mutex;
+use uuid::Uuid;
 
 use super::{
     database::models::{DeviceId, VaultId},
@@ -19,6 +20,7 @@ pub struct Cursors {
     config: DatabaseConfig,
     broadcasts: Broadcasts,
     vault_to_cursors: Arc<Mutex<HashMap<VaultId, Vec<ClientCursorsWithTimeToLive>>>>,
+    sessions: Arc<Mutex<HashMap<VaultId, HashMap<DeviceId, Uuid>>>>,
 }
 
 impl Cursors {
@@ -27,16 +29,39 @@ impl Cursors {
             config: config.clone(),
             broadcasts: broadcasts.clone(),
             vault_to_cursors: Arc::new(Mutex::new(HashMap::new())),
+            sessions: Arc::default(),
         }
     }
 
+    pub async fn register_connection(&self, vault_id: &str, device_id: &str) -> Uuid {
+        let mut sessions = self.sessions.lock().await;
+        let session = Uuid::new_v4();
+        sessions
+            .entry(vault_id.to_owned())
+            .or_default()
+            .insert(device_id.to_owned(), session);
+        self.remove_device_cursors(vault_id, device_id).await;
+        session
+    }
+
     pub async fn update_cursors(
         &self,
         vault_id: VaultId,
         user_name: String,
         device_id: &DeviceId,
+        session: Uuid,
         document_to_cursors: Vec<DocumentWithCursors>,
     ) {
+        // A reconnect can overlap an old connection's buffered messages and
+        // cleanup. Only the most recently authenticated session owns presence.
+        let sessions = self.sessions.lock().await;
+        if sessions
+            .get(&vault_id)
+            .and_then(|devices| devices.get(device_id))
+            != Some(&session)
+        {
+            return;
+        }
         let mut vault_to_cursors = self.vault_to_cursors.lock().await;
 
         let all_device_cursors = vault_to_cursors
@@ -100,7 +125,22 @@ impl Cursors {
             .await;
     }
 
-    pub async fn remove_cursors_of_device(&self, vault_id: &str, device_id: &str) {
+    pub async fn remove_cursors_of_device(&self, vault_id: &str, device_id: &str, session: Uuid) {
+        let mut sessions = self.sessions.lock().await;
+        let Some(devices) = sessions.get_mut(vault_id) else {
+            return;
+        };
+        if devices.get(device_id) != Some(&session) {
+            return;
+        }
+        devices.remove(device_id);
+        if devices.is_empty() {
+            sessions.remove(vault_id);
+        }
+        self.remove_device_cursors(vault_id, device_id).await;
+    }
+
+    async fn remove_device_cursors(&self, vault_id: &str, device_id: &str) {
         let mut vault_to_cursors = self.vault_to_cursors.lock().await;
 
         if let Some(cursors) = vault_to_cursors.get_mut(vault_id) {
@@ -147,32 +187,42 @@ mod tests {
     async fn disconnect_broadcasts_remaining_clients_and_the_final_empty_state() {
         let broadcasts = Broadcasts::new(&ServerConfig::default());
         let cursors = Cursors::new(&DatabaseConfig::default(), &broadcasts);
+        let mut sessions = HashMap::new();
         for device in ["one", "two"] {
+            let session = cursors.register_connection("vault", device).await;
+            sessions.insert(device, session);
             cursors
                 .update_cursors(
                     "vault".to_owned(),
                     "user".to_owned(),
                     &device.to_owned(),
+                    session,
                     vec![],
                 )
                 .await;
         }
         let mut receiver = broadcasts.get_receiver("vault".to_owned()).await;
         let mut other = broadcasts.get_receiver("other".to_owned()).await;
-        cursors.remove_cursors_of_device("vault", "one").await;
+        cursors
+            .remove_cursors_of_device("vault", "one", sessions["one"])
+            .await;
         let Notification::Cursors(update) = receiver.try_recv().unwrap() else {
             panic!("expected cursors")
         };
         assert_eq!(update.clients.len(), 1);
         assert_eq!(update.clients[0].device_id, "two");
 
-        cursors.remove_cursors_of_device("vault", "two").await;
+        cursors
+            .remove_cursors_of_device("vault", "two", sessions["two"])
+            .await;
         let Notification::Cursors(update) = receiver.try_recv().unwrap() else {
             panic!("expected cursors")
         };
         assert!(update.clients.is_empty());
         assert!(other.try_recv().is_err());
-        cursors.remove_cursors_of_device("vault", "two").await;
+        cursors
+            .remove_cursors_of_device("vault", "two", sessions["two"])
+            .await;
         assert!(receiver.try_recv().is_err());
     }
 
@@ -182,11 +232,13 @@ mod tests {
         let config = DatabaseConfig::default();
         let cursors = Cursors::new(&config, &broadcasts);
         for (vault, device) in [("vault", "expired"), ("vault", "live"), ("other", "live")] {
+            let session = cursors.register_connection(vault, device).await;
             cursors
                 .update_cursors(
                     vault.to_owned(),
                     "user".to_owned(),
                     &device.to_owned(),
+                    session,
                     vec![],
                 )
                 .await;
@@ -229,4 +281,39 @@ mod tests {
         assert!(receiver.try_recv().is_err());
         assert!(other.try_recv().is_err());
     }
+    #[tokio::test]
+    async fn replacement_session_fences_old_updates_and_disconnects() {
+        let broadcasts = Broadcasts::new(&ServerConfig::default());
+        let cursors = Cursors::new(&DatabaseConfig::default(), &broadcasts);
+        let vault = "vault".to_owned();
+        let device = "device".to_owned();
+        let old = cursors.register_connection(&vault, &device).await;
+        cursors
+            .update_cursors(vault.clone(), "old".to_owned(), &device, old, vec![])
+            .await;
+        let new = cursors.register_connection(&vault, &device).await;
+        cursors
+            .update_cursors(vault.clone(), "new".to_owned(), &device, new, vec![])
+            .await;
+        let mut receiver = broadcasts.get_receiver(vault.clone()).await;
+        cursors
+            .update_cursors(
+                vault.clone(),
+                "late old update".to_owned(),
+                &device,
+                old,
+                vec![],
+            )
+            .await;
+        cursors.remove_cursors_of_device(&vault, &device, old).await;
+        assert!(receiver.try_recv().is_err());
+        assert_eq!(cursors.get_cursors(&vault).await[0].user_name, "new");
+        cursors.remove_cursors_of_device(&vault, &device, new).await;
+        let Notification::Cursors(update) = receiver.try_recv().unwrap() else {
+            panic!("expected cursors")
+        };
+        assert!(update.clients.is_empty());
+        assert!(cursors.get_cursors(&vault).await.is_empty());
+        assert!(cursors.sessions.lock().await.is_empty());
+    }
 }

sync-server/src/app_state/database.rs

@@ -2,7 +2,7 @@ use core::time::Duration;
 use std::{
     collections::HashMap,
     path::{Path, PathBuf},
-    sync::Arc,
+    sync::{Arc, Weak},
 };
 
 use anyhow::{Context as _, Result};
@@ -48,6 +48,7 @@ impl std::fmt::Debug for PoolWithTimestamp {
 pub struct Database {
     config: DatabaseConfig,
     connection_pools: Arc<Mutex<HashMap<VaultId, PoolWithTimestamp>>>,
+    opening_pools: Arc<Mutex<HashMap<VaultId, Weak<Mutex<()>>>>>,
 }
 
 pub type Transaction<'a> = sqlx::Transaction<'a, Sqlite>;
@@ -66,25 +67,16 @@ impl Database {
                 )
             })?;
 
-        info!("Applying pending database migrations");
+        // SQLite flushes its own files, not ancestors created by this process.
+        Self::flush_directory_ancestors(&vault_directory).await?;
 
-        // Filenames encode only a digest, so their vault names cannot be
-        // recovered here. Check migrations now and open pools lazily by name.
-        let mut entries = tokio::fs::read_dir(&vault_directory).await?;
-        while let Some(entry) = entries.next_entry().await? {
-            if entry.file_type().await?.is_file()
-                && entry.path().extension().is_some_and(|ext| ext == "sqlite")
-            {
-                Self::open_database(config, &entry.path())
-                    .await?
-                    .close()
-                    .await;
-            }
-        }
+        // Open and migrate each vault lazily under its own lock. A corrupt or
+        // incompatible database must fail that vault, not global startup.
 
         let database = Self {
             config: config.clone(),
             connection_pools: Arc::default(),
+            opening_pools: Arc::default(),
         };
 
         // Start background task to cleanup idle connection pools
@@ -127,9 +119,32 @@ impl Database {
 
         Self::run_migrations(&pool).await?;
 
+        if let Some(directory) = file_name.parent() {
+            Self::flush_directory_ancestors(directory).await?;
+        }
+
         Ok(pool)
     }
 
+    async fn flush_directory_ancestors(directory: &Path) -> Result<()> {
+        #[cfg(unix)]
+        {
+            let directory = tokio::fs::canonicalize(directory).await?;
+            tokio::task::spawn_blocking(move || -> Result<()> {
+                for ancestor in directory.ancestors() {
+                    std::fs::File::open(ancestor)?.sync_all().with_context(|| {
+                        format!("Cannot flush directory {}", ancestor.display())
+                    })?;
+                }
+                Ok(())
+            })
+            .await??;
+        }
+        #[cfg(not(unix))]
+        let _ = directory;
+        Ok(())
+    }
+
     async fn run_migrations(pool: &Pool<Sqlite>) -> Result<()> {
         sqlx::migrate!("src/app_state/database/migrations")
             .run(pool)
@@ -140,28 +155,49 @@ impl Database {
     async fn get_connection_pool(&self, vault: &VaultId) -> Result<Pool<Sqlite>> {
         let vault = normalize_string(vault);
         validate_vault_id(&vault)?;
-        let mut pools = self.connection_pools.lock().await;
-
-        if !pools.contains_key(&vault) {
-            let file_name = Self::database_path(&self.config, &vault);
-            let pool = Self::open_database(&self.config, &file_name).await?;
-            pools.insert(
-                vault.clone(),
-                PoolWithTimestamp {
-                    pool,
-                    last_accessed: Instant::now(),
-                },
-            );
+        if let Some(pool) = self.existing_pool(&vault).await {
+            return Ok(pool);
         }
 
-        let pool_with_timestamp = pools
-            .get_mut(&vault)
-            .expect("Pool was just inserted or already exists");
+        // Serialize initialization per vault. SQLite lock waits, migrations and
+        // directory flushes must never hold the registry for unrelated vaults.
+        // Weak entries also allow cancelled/failed opens to be retried safely.
+        let opening = {
+            let mut openings = self.opening_pools.lock().await;
+            openings.retain(|_, opening| opening.strong_count() > 0);
+            if let Some(opening) = openings.get(&vault).and_then(Weak::upgrade) {
+                opening
+            } else {
+                let opening = Arc::new(Mutex::new(()));
+                openings.insert(vault.clone(), Arc::downgrade(&opening));
+                opening
+            }
+        };
+        let _opening = opening.lock().await;
+        if let Some(pool) = self.existing_pool(&vault).await {
+            return Ok(pool);
+        }
+        let file_name = Self::database_path(&self.config, &vault);
+        let pool = Self::open_database(&self.config, &file_name).await?;
+        self.connection_pools.lock().await.insert(
+            vault,
+            PoolWithTimestamp {
+                pool: pool.clone(),
+                last_accessed: Instant::now(),
+            },
+        );
+        Ok(pool)
+    }
 
-        // Update last accessed time
-        pool_with_timestamp.last_accessed = Instant::now();
-
-        Ok(pool_with_timestamp.pool.clone())
+    async fn existing_pool(&self, vault: &VaultId) -> Option<Pool<Sqlite>> {
+        self.connection_pools
+            .lock()
+            .await
+            .get_mut(vault)
+            .map(|entry| {
+                entry.last_accessed = Instant::now();
+                entry.pool.clone()
+            })
     }
 
     /// Attempting to write from this transaction might result in a
@@ -196,16 +232,22 @@ impl Database {
             .iter()
             .filter(|(_, pool_with_timestamp)| {
                 now.duration_since(pool_with_timestamp.last_accessed) > idle_timeout
+                    && pool_with_timestamp.pool.num_idle()
+                        == pool_with_timestamp.pool.size() as usize
             })
             .map(|(vault_id, _)| vault_id.clone())
             .collect();
 
-        // Close and remove idle pools
-        for vault_id in &vaults_to_remove {
-            if let Some(pool_with_timestamp) = pools.remove(vault_id) {
-                info!("Closing idle database connection pool for vault `{vault_id}`");
-                pool_with_timestamp.pool.close().await;
-            }
+        let closing: Vec<_> = vaults_to_remove
+            .into_iter()
+            .filter_map(|vault| pools.remove(&vault).map(|entry| (vault, entry.pool)))
+            .collect();
+        drop(pools);
+        // A checkout racing cleanup can delay close. It must never retain the
+        // mutex used to access every other vault in the process.
+        for (vault, pool) in closing {
+            info!("Closing idle database connection pool for vault `{vault}`");
+            pool.close().await;
         }
     }
 

sync-server/src/app_state/database/migrations/20260920000000_history_checkpoint.sql

@@ -0,0 +1,4 @@
+-- Tokens identify an event's incarnation, including when a restored backup
+-- reuses its numeric ID or a client resubmits the same request UUID.
+ALTER TABLE events ADD COLUMN event_token TEXT;
+UPDATE events SET event_token = lower(hex(randomblob(16)));

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

@@ -123,5 +123,9 @@ pub struct VaultSnapshot {
 pub struct EventBatch {
     #[ts(as = "f64")]
     pub head_event_id: VaultUpdateId,
+    /// Last event in this page; `head_event_id` is the vault's current head.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    #[ts(optional, as = "Option<f64>")]
+    pub end_event_id: Option<VaultUpdateId>,
     pub events: Vec<EventRecord>,
 }

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

@@ -10,14 +10,15 @@ impl Database {
         request_id: uuid::Uuid,
         fingerprint: &[u8],
     ) -> Result<VaultUpdateId> {
-        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(

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

@@ -8,6 +8,22 @@ use super::{
 use anyhow::{Context as _, Result, ensure};
 
 impl Database {
+    pub async fn missing_document(
+        tx: &mut Transaction<'_>,
+        ids: &[DocumentId],
+    ) -> Result<Option<String>> {
+        // A single covering-index lookup per requested identity; never load a
+        // content blob or decode a version while validating manifest references.
+        Ok(sqlx::query_scalar(
+            "SELECT value FROM json_each(?) AS requested
+             WHERE NOT EXISTS (SELECT 1 FROM documents WHERE document_id = requested.value)
+             LIMIT 1",
+        )
+        .bind(serde_json::to_string(ids)?)
+        .fetch_optional(&mut **tx)
+        .await?)
+    }
+
     pub async fn get_request_event(
         tx: &mut Transaction<'_>,
         request_id: uuid::Uuid,
@@ -168,24 +184,72 @@ impl Database {
         );
 
         let rows: Vec<(VaultUpdateId, String)> = sqlx::query_as(
-            "SELECT event_id, request_id FROM events WHERE event_id > ? ORDER BY event_id",
+            "SELECT event_id, request_id FROM events WHERE event_id > ? ORDER BY event_id LIMIT 64",
         )
         .bind(after)
         .fetch_all(&mut *tx)
         .await?;
 
         let mut events = Vec::with_capacity(rows.len());
+        let mut bytes = 0;
+        let mut end_event_id = after;
         for (event_id, request_id) in rows {
-            events.push(EventRecord {
+            let event = EventRecord {
                 event_id,
                 request_id: request_id.parse()?,
                 event: Self::event_by_id(&mut tx, event_id).await?,
-            });
+            };
+            let size = serde_json::to_vec(&event)?.len();
+            // One manifest is atomic and may exceed this budget. Always allow
+            // one event so even such a page advances; never accumulate history.
+            if !events.is_empty() && bytes + size > 1024 * 1024 {
+                break;
+            }
+            bytes += size;
+            end_event_id = event_id;
+            events.push(event);
         }
 
         Ok(EventBatch {
             head_event_id,
+            end_event_id: Some(end_event_id),
             events,
         })
     }
 }
+
+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(
+            || "0:empty".to_owned(),
+            |(id, token)| format!("{id}:{token}"),
+        ))
+    }
+
+    pub async fn contains_checkpoint(&self, vault: &VaultId, checkpoint: &str) -> Result<bool> {
+        if checkpoint == "0:empty" {
+            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

@@ -94,6 +94,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).await.unwrap().events.len(),
+        1
+    );
+}
+
 #[tokio::test]
 async fn snapshot_returns_latest_referenced_metadata_and_byte_lengths() {
     let directory = tempfile::tempdir().unwrap();
@@ -134,3 +160,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).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)
+            .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).await.unwrap();
+    assert!(!batch.events.is_empty());
+    assert!(batch.events.len() < 8);
+    assert!(serde_json::to_vec(&batch).unwrap().len() < 1024 * 1024 + 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/app_state/websocket/broadcasts.rs

@@ -12,23 +12,39 @@ pub enum Notification {
 #[derive(Debug, Clone)]
 pub struct Broadcasts {
     capacity: usize,
+    admission: Arc<std::sync::Mutex<HashMap<VaultId, std::sync::Weak<tokio::sync::Semaphore>>>>,
     senders: Arc<Mutex<HashMap<VaultId, broadcast::Sender<Notification>>>>,
 }
 
 impl Broadcasts {
     pub fn new(config: &ServerConfig) -> Self {
         Self {
-            capacity: config.max_clients_per_vault.max(1),
+            capacity: config.max_clients_per_vault,
+            admission: Arc::default(),
             senders: Arc::default(),
         }
     }
 
+    pub fn try_admit(&self, vault: &VaultId) -> Option<tokio::sync::OwnedSemaphorePermit> {
+        let mut limits = self.admission.lock().unwrap();
+        limits.retain(|_, limit| limit.strong_count() > 0);
+        let limit = limits
+            .get(vault)
+            .and_then(std::sync::Weak::upgrade)
+            .unwrap_or_else(|| {
+                let limit = Arc::new(tokio::sync::Semaphore::new(self.capacity));
+                limits.insert(vault.clone(), Arc::downgrade(&limit));
+                limit
+            });
+        limit.try_acquire_owned().ok()
+    }
+
     async fn sender(&self, vault: VaultId) -> broadcast::Sender<Notification> {
         self.senders
             .lock()
             .await
             .entry(vault)
-            .or_insert_with(|| broadcast::channel(self.capacity).0)
+            .or_insert_with(|| broadcast::channel(self.capacity.max(1)).0)
             .clone()
     }
 

sync-server/src/app_state/websocket/utils.rs

@@ -49,9 +49,13 @@ pub async fn send_update_over_websocket(
         .context("Failed to serialize update")
         .map_err(server_error)?;
 
-    sender
-        .send(Message::Text(serialized_update))
-        .await
-        .context("Failed to send message over websocket")
-        .map_err(server_error)
+    tokio::time::timeout(
+        std::time::Duration::from_secs(10),
+        sender.send(Message::Text(serialized_update)),
+    )
+    .await
+    .context("WebSocket send deadline expired")
+    .map_err(server_error)?
+    .context("Failed to send message over websocket")
+    .map_err(server_error)
 }

sync-server/src/config.rs

@@ -20,7 +20,6 @@ pub struct Config {
     pub database: DatabaseConfig,
     #[serde(default)]
     pub server: ServerConfig,
-    #[serde(default)]
     pub users: UserConfig,
     #[serde(default)]
     pub logging: LoggingConfig,
@@ -28,23 +27,52 @@ pub struct Config {
 
 impl Config {
     pub async fn read_or_create(path: &Path) -> Result<Self> {
-        let config = if path.exists() {
-            info!(
-                "Loading configuration from `{}`",
-                path.canonicalize().unwrap().display()
-            );
-            Self::load_from_file(path).await?
-        } else {
-            Self::default()
-        };
-
-        config.write(path).await?;
-        info!(
-            "Updated configuration at `{}`",
-            path.canonicalize().unwrap().display()
-        );
-
-        Ok(config)
+        match Self::load_from_file(path).await {
+            Ok(config) => return Ok(config),
+            Err(error)
+                if error
+                    .downcast_ref::<std::io::Error>()
+                    .is_some_and(|error| error.kind() == std::io::ErrorKind::NotFound) => {}
+            Err(error) => return Err(error),
+        }
+        let contents = serde_yaml::to_string(&Self::default())?;
+        let destination = path.to_owned();
+        tokio::task::spawn_blocking(move || -> Result<()> {
+            use std::io::Write;
+            let path = destination;
+            let parent = path
+                .parent()
+                .filter(|p| !p.as_os_str().is_empty())
+                .unwrap_or(Path::new("."));
+            let temporary = parent.join(format!(".vault-link-config-{}.tmp", uuid::Uuid::new_v4()));
+            let result = (|| -> Result<()> {
+                let mut options = std::fs::OpenOptions::new();
+                options.write(true).create_new(true);
+                #[cfg(unix)]
+                {
+                    use std::os::unix::fs::OpenOptionsExt;
+                    options.mode(0o600);
+                }
+                let mut file = options.open(&temporary)?;
+                file.write_all(contents.as_bytes())?;
+                file.sync_all()?;
+                // Publish a complete, durable file without replacing a config
+                // another process may have created while we generated defaults.
+                match std::fs::hard_link(&temporary, &path) {
+                    Ok(()) => {}
+                    Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
+                    Err(error) => return Err(error.into()),
+                }
+                #[cfg(unix)]
+                std::fs::File::open(parent)?.sync_all()?;
+                Ok(())
+            })();
+            let _ = std::fs::remove_file(&temporary);
+            result
+        })
+        .await??;
+        info!("Loading configuration from `{}`", path.display());
+        Self::load_from_file(path).await
     }
 
     pub async fn load_from_file(path: &Path) -> Result<Self> {
@@ -57,12 +85,54 @@ impl Config {
 
         serde_yaml::from_str(&contents).context("Failed to parse configuration")
     }
+}
 
-    pub async fn write(&self, path: &Path) -> Result<()> {
-        let contents = serde_yaml::to_string(&self).context("Failed to serialize configuration")?;
+#[cfg(test)]
+mod tests {
+    use super::*;
 
-        fs::write(path, contents)
-            .await
-            .context("Failed to write configuration to disk")
+    #[tokio::test]
+    async fn startup_preserves_existing_configuration_bytes_and_permissions() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("config.yml");
+        let original = b"# administrator's comments and formatting\nserver: {port: 9123}\nusers: {user_configs: []}\n";
+        fs::write(&path, original).await.unwrap();
+        let mut permissions = fs::metadata(&path).await.unwrap().permissions();
+        permissions.set_readonly(true);
+        fs::set_permissions(&path, permissions).await.unwrap();
+        let config = Config::read_or_create(&path).await.unwrap();
+        assert_eq!(config.server.port, 9123);
+        assert_eq!(fs::read(&path).await.unwrap(), original);
+    }
+
+    #[tokio::test]
+    async fn existing_configuration_cannot_generate_unpersisted_credentials() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("config.yml");
+        for contents in ["server: {port: 9123}\n", "users: {}\n"] {
+            fs::write(&path, contents).await.unwrap();
+            for _ in 0..2 {
+                assert!(
+                    Config::read_or_create(&path).await.is_err(),
+                    "Existing credentials must be explicit; random defaults are only for initialization"
+                );
+            }
+            assert_eq!(fs::read_to_string(&path).await.unwrap(), contents);
+        }
+    }
+
+    #[tokio::test]
+    async fn concurrent_initialization_never_overwrites_configuration() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("config.yml");
+        let results =
+            futures::future::join_all((0..16).map(|_| Config::read_or_create(&path))).await;
+        let config = Config::load_from_file(&path).await.unwrap();
+        for result in results {
+            assert_eq!(
+                serde_yaml::to_string(&result.unwrap()).unwrap(),
+                serde_yaml::to_string(&config).unwrap()
+            );
+        }
     }
 }

sync-server/src/config/server_config.rs

@@ -85,8 +85,9 @@ mod tests {
         let path = directory.path().join("config.yml");
         let fresh = Config::read_or_create(&path).await.unwrap();
         let reread = Config::load_from_file(&path).await.unwrap();
-        let omitted: Config = serde_yaml::from_str("{}").unwrap();
-        let explicit: Config = serde_yaml::from_str("server: {}").unwrap();
+        let omitted: Config = serde_yaml::from_str("users: {user_configs: []}").unwrap();
+        let explicit: Config =
+            serde_yaml::from_str("server: {}\nusers: {user_configs: []}").unwrap();
         for config in [fresh, reread, omitted, explicit] {
             assert_eq!(config.server.host, DEFAULT_HOST);
             assert_eq!(config.server.port, DEFAULT_PORT);

sync-server/src/config/user_config.rs

@@ -9,7 +9,7 @@ use crate::{
 
 #[derive(Debug, Deserialize, Serialize, Clone)]
 pub struct UserConfig {
-    #[serde(default = "default_users", deserialize_with = "validate_users")]
+    #[serde(deserialize_with = "validate_users")]
     pub user_configs: Vec<User>,
 }
 

sync-server/src/server.rs

@@ -63,6 +63,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,6 +118,10 @@ 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_version::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),

sync-server/src/server/auth.rs

@@ -2,8 +2,9 @@ use std::collections::HashMap;
 
 use axum::{
     extract::{Path, Request, State},
+    http::{HeaderValue, StatusCode},
     middleware::Next,
-    response::Response,
+    response::{IntoResponse, Response},
 };
 use axum_extra::{
     TypedHeader,
@@ -14,7 +15,9 @@ use log::info;
 use crate::{
     app_state::{AppState, database::models::VaultId},
     config::user_config::{AllowListedVaults, User, VaultAccess},
-    errors::{SyncServerError, client_error, permission_denied_error, unauthenticated_error},
+    errors::{
+        SyncServerError, client_error, permission_denied_error, server_error, unauthenticated_error,
+    },
     utils::normalize_vault_id::{normalize_string, validate_vault_id},
 };
 
@@ -34,9 +37,42 @@ pub async fn auth_middleware(
 
     let user = auth(&state, token, &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);
+        }
+    }
+
     req.extensions_mut().insert(user);
 
-    Ok(next.run(req).await)
+    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)
 }
 
 pub fn auth(state: &AppState, token: &str, vault_id: &VaultId) -> Result<User, SyncServerError> {

sync-server/src/server/endpoints/fetch_latest_document_version.rs

@@ -9,7 +9,7 @@ use super::VaultPath;
 use crate::{
     app_state::{
         AppState,
-        database::models::{DocumentId, DocumentVersion},
+        database::models::{DocumentId, DocumentVersion, DocumentVersionWithoutContent},
     },
     errors::{SyncServerError, not_found_error, server_error},
 };
@@ -37,3 +37,21 @@ pub async fn fetch_latest_document_version(
 
     Ok(Json(latest_version.into()))
 }
+
+pub async fn fetch_latest_document_metadata(
+    Path((VaultPath(vault_id), document_id)): Path<(VaultPath, DocumentId)>,
+    State(state): State<AppState>,
+) -> Result<Json<DocumentVersionWithoutContent>, SyncServerError> {
+    let mut tx = state
+        .database
+        .create_readonly_transaction(&vault_id)
+        .await
+        .map_err(server_error)?;
+    let metadata = sqlx::query_as::<_, DocumentVersionWithoutContent>(
+        "SELECT vault_update_id, document_id, updated_date, user_id, device_id, length(content) AS content_size
+         FROM documents WHERE document_id = ? ORDER BY vault_update_id DESC LIMIT 1"
+    ).bind(document_id.hyphenated()).fetch_optional(&mut *tx).await
+        .map_err(|error| server_error(error.into()))?
+        .ok_or_else(|| not_found_error(anyhow!("Document does not exist")))?;
+    Ok(Json(metadata))
+}

sync-server/src/server/endpoints/put_file_content.rs

@@ -23,7 +23,7 @@ use axum::{
 use axum_extra::TypedHeader;
 use base64::{Engine as _, engine::general_purpose::STANDARD};
 use log::debug;
-use reconcile_text::{BuiltinTokenizer, EditedText, NumberOrText};
+use reconcile_text::NumberOrText;
 use sha2::{Digest, Sha256};
 
 #[axum::debug_handler]
@@ -41,6 +41,30 @@ pub async fn put_file_content(
             .map_err(|error| server_error(error.into()))?,
     );
 
+    // Build bytes against an immutable parent off the async executor and before
+    // taking the writer lock. The transaction below rechecks CAS and receipts.
+    let initial_parent = state
+        .database
+        .get_latest_document_version(&vault_id, &document_id, None)
+        .await
+        .map_err(server_error)?;
+    let prepared = if initial_parent.as_ref().map(|p| p.vault_update_id) == push.parent_version_id {
+        let limit = state
+            .config
+            .server
+            .max_body_size_mb
+            .saturating_mul(1024 * 1024);
+        Some(
+            tokio::task::spawn_blocking(move || {
+                decode_content(initial_parent, push.content, limit)
+            })
+            .await
+            .map_err(|error| server_error(error.into()))?,
+        )
+    } else {
+        None
+    };
+
     let mut tx = state
         .database
         .create_write_transaction(&vault_id)
@@ -78,33 +102,13 @@ pub async fn put_file_content(
         )));
     }
 
-    let content = match push.content {
-        PushContent::Snapshot(bytes) => STANDARD
-            .decode(bytes)
-            .map_err(|error| client_error(error.into()))?,
-
-        PushContent::Diff(diff) => {
-            if diff
-                .iter()
-                .any(|n| matches!(n, NumberOrText::Number(i64::MIN)))
-            {
-                return Err(client_error(anyhow!("Invalid diff length")));
-            }
-
-            let parent = parent
-                .as_ref()
-                .ok_or_else(|| client_error(anyhow!("Diff requires a parent")))?;
-
-            let text =
-                str::from_utf8(&parent.content).map_err(|error| client_error(error.into()))?;
-
-            EditedText::from_diff(text, diff, &*BuiltinTokenizer::Word)
-                .map_err(|error| client_error(error.into()))?
-                .apply()
-                .text()
-                .into_bytes()
-        }
-    };
+    let content = prepared
+        .ok_or_else(|| {
+            server_error(anyhow!(
+                "Document changed while preparing content; retry the request"
+            ))
+        })?
+        .map_err(client_error)?;
 
     let version = StoredDocumentVersion {
         vault_update_id: Database::allocate_event(&mut tx, push.request_id, &fingerprint)
@@ -132,8 +136,65 @@ pub async fn put_file_content(
     Ok(Json(response))
 }
 
+const MAX_DIFF_ITEMS: usize = 10_000;
+
+fn decode_content(
+    parent: Option<StoredDocumentVersion>,
+    content: PushContent,
+    limit: usize,
+) -> anyhow::Result<Vec<u8>> {
+    let bytes = match content {
+        PushContent::Snapshot(encoded) => STANDARD.decode(encoded)?,
+        PushContent::Diff(diff) => {
+            anyhow::ensure!(
+                diff.len() <= MAX_DIFF_ITEMS,
+                "Diff exceeds operation budget; send a snapshot"
+            );
+            let parent = parent.ok_or_else(|| anyhow!("Diff requires a parent"))?;
+            let text = str::from_utf8(&parent.content)?;
+            apply_diff(text, diff, limit)?
+        }
+    };
+    anyhow::ensure!(bytes.len() <= limit, "Content exceeds size limit");
+    Ok(bytes)
+}
+
+/// Each Unicode scalar in the parent is visited at most once. No tokenization
+/// or repeated scans of a prefix are needed to apply the wire diff.
+fn apply_diff(text: &str, diff: Vec<NumberOrText>, limit: usize) -> anyhow::Result<Vec<u8>> {
+    let mut source = text.chars();
+    let mut result = String::new();
+    for item in diff {
+        match item {
+            NumberOrText::Text(insert) => {
+                anyhow::ensure!(
+                    insert.len() <= limit.saturating_sub(result.len()),
+                    "Content exceeds size limit"
+                );
+                result.push_str(&insert);
+            }
+            NumberOrText::Number(length) => {
+                for _ in 0..length.unsigned_abs() {
+                    let ch = source
+                        .next()
+                        .ok_or_else(|| anyhow!("Diff exceeds parent length"))?;
+                    if length > 0 {
+                        anyhow::ensure!(
+                            ch.len_utf8() <= limit.saturating_sub(result.len()),
+                            "Content exceeds size limit"
+                        );
+                        result.push(ch);
+                    }
+                }
+            }
+        }
+    }
+    Ok(result.into_bytes())
+}
+
 #[cfg(test)]
 mod tests {
+    use reconcile_text::EditedText;
     use std::path::PathBuf;
 
     use super::*;
@@ -278,6 +339,29 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn fragmented_diff_budget_rejects_without_committing_an_event() {
+        let vault = TestVault::new().await;
+        let document = uuid::Uuid::new_v4();
+        let base = accepted(
+            vault
+                .send(snapshot(document, None, &"a".repeat(1024 * 1024)))
+                .await,
+        );
+        let mut push = snapshot(document, Some(base.vault_update_id), "");
+        let mut fragments = vec![NumberOrText::Number(0); 10_001];
+        fragments.push(NumberOrText::Number(1024 * 1024));
+        push.content = PushContent::Diff(fragments);
+        assert!(matches!(
+            push.send(&vault.state, &vault.vault).await,
+            Err(SyncServerError::ClientError(_))
+        ));
+        assert_eq!(
+            vault.latest(document).await.vault_update_id,
+            base.vault_update_id
+        );
+    }
+
     #[tokio::test]
     async fn json_snapshot_preserves_bytes_and_rejects_invalid_base64_without_a_version() {
         let vault = TestVault::new().await;
@@ -363,11 +447,8 @@ mod tests {
             outcomes => panic!("Expected one accepted push and one stale base: {outcomes:?}"),
         };
         let latest = vault.latest(document).await;
-        assert_eq!(winner.vault_update_id, rejected.metadata.vault_update_id);
-        assert_eq!(
-            STANDARD.decode(rejected.content_base64).unwrap(),
-            latest.content
-        );
+        assert_eq!(winner.vault_update_id, rejected.vault_update_id);
+        assert_eq!(rejected.content_size, latest.content.len());
         assert!(latest.content == b"left" || latest.content == b"right");
         let original = vault
             .state
@@ -577,7 +658,7 @@ mod tests {
         assert!(matches!(
             vault.send(foreign_base).await,
             DocumentUpdateResponse::StaleBase(version)
-                if version.metadata.vault_update_id == updated.vault_update_id
+                if version.vault_update_id == updated.vault_update_id
         ));
 
         let binary_document = uuid::Uuid::new_v4();

sync-server/src/server/endpoints/put_file_manifest.rs

@@ -62,18 +62,14 @@ pub async fn put_file_manifest(
         return Ok(Json(FileManifestUpdateResponse::StaleBase(latest)));
     }
 
-    for id in push.entries.keys() {
-        if state
-            .database
-            .get_latest_document_version(&vault_id, id, Some(&mut tx))
+    if let Some(id) =
+        Database::missing_document(&mut tx, &push.entries.keys().copied().collect::<Vec<_>>())
             .await
             .map_err(server_error)?
-            .is_none()
-        {
-            return Err(client_error(anyhow!(
-                "File manifest references missing content: {id}"
-            )));
-        }
+    {
+        return Err(client_error(anyhow!(
+            "File manifest references missing content: {id}"
+        )));
     }
 
     let file_manifest = FileManifest {

sync-server/src/server/endpoints/websocket.rs

@@ -15,7 +15,8 @@ use axum::{
         Path, State,
         ws::{Message, WebSocket, WebSocketUpgrade},
     },
-    response::Response,
+    http::StatusCode,
+    response::{IntoResponse, Response},
 };
 use futures::StreamExt;
 use log::{debug, info};
@@ -32,11 +33,22 @@ pub async fn websocket_handler(
 ) -> Result<Response, SyncServerError> {
     debug!("Upgrading WebSocket connection for vault `{vault_id}`");
 
-    Ok(ws.on_upgrade(move |socket| async move {
-        if let Err(error) = websocket(state, socket, vault_id).await {
-            debug!("WebSocket disconnected: {error}");
-        }
-    }))
+    let Some(permit) = state.broadcasts.try_admit(&vault_id) else {
+        return Ok((
+            StatusCode::TOO_MANY_REQUESTS,
+            "Vault connection limit reached",
+        )
+            .into_response());
+    };
+    Ok(ws
+        .max_message_size(1024 * 1024)
+        .max_frame_size(1024 * 1024)
+        .on_upgrade(move |socket| async move {
+            let _permit = permit;
+            if let Err(error) = websocket(state, socket, vault_id).await {
+                debug!("WebSocket disconnected: {error}");
+            }
+        }))
 }
 
 async fn websocket(
@@ -48,10 +60,15 @@ async fn websocket(
     let authenticated = get_authenticated_handshake(
         &state,
         &vault,
-        receiver.next().await.transpose().unwrap_or_default(),
+        tokio::time::timeout(Duration::from_secs(5), receiver.next())
+            .await
+            .map_err(|_| client_error(anyhow::anyhow!("WebSocket handshake deadline expired")))?
+            .transpose()
+            .unwrap_or_default(),
     )?;
 
     let device = authenticated.handshake.device_id;
+    let session = state.cursors.register_connection(&vault, &device).await;
     let mut after = authenticated
         .handshake
         .last_seen_vault_update_id
@@ -69,7 +86,7 @@ async fn websocket(
             let batch = state.database.events_after(&vault, after).await.map_err(server_error)?;
 
             if !batch.events.is_empty() {
-                let head = batch.head_event_id;
+                let head = batch.end_event_id.unwrap_or(batch.head_event_id);
                 send_update_over_websocket(&WebSocketServerMessage::VaultEvents(batch), &mut sender).await?;
                 after = head;
             }
@@ -92,7 +109,7 @@ async fn websocket(
                         match message {
                             WebSocketClientMessage::Handshake(_) => return Err(client_error(anyhow::anyhow!("Unexpected handshake"))),
                             WebSocketClientMessage::CursorPositions(positions) => {
-                                state.cursors.update_cursors(vault.clone(), authenticated.user.name.clone(), &device, positions.documents_with_cursors).await;
+                                state.cursors.update_cursors(vault.clone(), authenticated.user.name.clone(), &device, session, positions.documents_with_cursors).await;
                                 let clients = state.cursors.get_cursors(&vault).await.into_iter().filter(|c| c.device_id != device).collect();
                                 send_update_over_websocket(&WebSocketServerMessage::CursorPositions(CursorPositionFromServer { clients }), &mut sender).await?;
                             }
@@ -109,7 +126,7 @@ async fn websocket(
 
     state
         .cursors
-        .remove_cursors_of_device(&vault, &device)
+        .remove_cursors_of_device(&vault, &device, session)
         .await;
 
     result

sync-server/src/server/responses.rs

@@ -2,7 +2,7 @@ use serde::{self, Deserialize, Serialize};
 use ts_rs::TS;
 
 use crate::app_state::database::models::{
-    DocumentVersion, DocumentVersionWithoutContent, FileManifest, VaultUpdateId,
+    DocumentVersionWithoutContent, FileManifest, VaultUpdateId,
 };
 
 /// Response to a ping request.
@@ -34,7 +34,7 @@ pub enum DocumentUpdateResponse {
     Accepted(DocumentVersionWithoutContent),
 
     /// Nothing was written. The client must incorporate this version before retrying.
-    StaleBase(DocumentVersion),
+    StaleBase(DocumentVersionWithoutContent),
 }
 
 #[derive(TS, Debug, Clone, Serialize, Deserialize)]