From 45a6b1c115cc2efa25562d64c571200d8b66f077 Mon Sep 17 00:00:00 2001 From: Andras Schmelczer Date: Sat, 19 Sep 2026 22:10:38 +0100 Subject: [PATCH] Don't duplicate events --- .../src/app_state/database/mutations.rs | 11 +-- sync-server/src/app_state/database/queries.rs | 81 ++++++++++++++----- sync-server/src/app_state/database/tests.rs | 26 +----- .../src/server/endpoints/put_file_content.rs | 18 +---- .../src/server/endpoints/put_file_manifest.rs | 12 +-- sync-server/src/server/endpoints/utils.rs | 11 +-- sync-server/src/utils/portable_path.rs | 16 +--- 7 files changed, 75 insertions(+), 100 deletions(-) diff --git a/sync-server/src/app_state/database/mutations.rs b/sync-server/src/app_state/database/mutations.rs index 09f7a73b..5d7b52b4 100644 --- a/sync-server/src/app_state/database/mutations.rs +++ b/sync-server/src/app_state/database/mutations.rs @@ -1,6 +1,6 @@ use super::{ Database, Transaction, - models::{EventRecord, FileManifest, StoredDocumentVersion, VaultUpdateId}, + models::{FileManifest, StoredDocumentVersion, VaultUpdateId}, }; use anyhow::Result; @@ -59,13 +59,4 @@ impl Database { Ok(()) } - - pub async fn write_event(tx: &mut Transaction<'_>, event: &EventRecord) -> Result<()> { - sqlx::query("UPDATE events SET event_json = ? WHERE event_id = ?") - .bind(serde_json::to_string(event)?) - .bind(event.event_id) - .execute(&mut **tx) - .await?; - Ok(()) - } } diff --git a/sync-server/src/app_state/database/queries.rs b/sync-server/src/app_state/database/queries.rs index 2134a58a..5ba1a97d 100644 --- a/sync-server/src/app_state/database/queries.rs +++ b/sync-server/src/app_state/database/queries.rs @@ -1,23 +1,28 @@ use super::{ Database, Transaction, models::{ - DocumentId, DocumentVersionWithoutContent, EventBatch, FileManifest, StoredDocumentVersion, - VaultId, VaultSnapshot, VaultUpdateId, + DocumentId, DocumentVersionWithoutContent, EventBatch, EventRecord, FileManifest, + StoredDocumentVersion, VaultEvent, VaultId, VaultSnapshot, VaultUpdateId, }, }; -use anyhow::{Result, ensure}; +use anyhow::{Context as _, Result, ensure}; impl Database { pub async fn get_request_event( tx: &mut Transaction<'_>, request_id: uuid::Uuid, - ) -> Result, String)>> { - Ok(sqlx::query_as( - "SELECT request_fingerprint, event_json FROM events WHERE request_id = ?", - ) - .bind(request_id.to_string()) - .fetch_optional(&mut **tx) - .await?) + ) -> Result, VaultEvent)>> { + let row: Option<(Vec, VaultUpdateId)> = + sqlx::query_as("SELECT request_fingerprint, event_id FROM events WHERE request_id = ?") + .bind(request_id.to_string()) + .fetch_optional(&mut **tx) + .await?; + match row { + Some((fingerprint, event_id)) => { + Ok(Some((fingerprint, Self::event_by_id(tx, event_id).await?))) + } + None => Ok(None), + } } pub async fn get_latest_document_version( @@ -70,6 +75,24 @@ impl Database { return Ok(FileManifest::default()); }; + Self::file_manifest_by_id(tx, file_manifest_id) + .await? + .context("Latest file manifest is missing") + } + + async fn file_manifest_by_id( + tx: &mut Transaction<'_>, + file_manifest_id: VaultUpdateId, + ) -> Result> { + let exists = sqlx::query_scalar::<_, VaultUpdateId>( + "SELECT file_manifest_id FROM file_manifests WHERE file_manifest_id = ?", + ) + .bind(file_manifest_id) + .fetch_optional(&mut **tx) + .await?; + if exists.is_none() { + return Ok(None); + } let entries = sqlx::query_as::<_, (uuid::fmt::Hyphenated, String)>( "SELECT document_id, path FROM file_manifest_entries WHERE file_manifest_id = ?", ) @@ -80,10 +103,28 @@ impl Database { .map(|(document_id, path)| (document_id.into_uuid(), path)) .collect(); - Ok(FileManifest { + Ok(Some(FileManifest { file_manifest_id, entries, - }) + })) + } + + async fn event_by_id(tx: &mut Transaction<'_>, event_id: VaultUpdateId) -> Result { + let document = sqlx::query_as::<_, DocumentVersionWithoutContent>( + "SELECT vault_update_id, document_id, updated_date, user_id, device_id, + length(content) AS content_size + FROM documents WHERE vault_update_id = ?", + ) + .bind(event_id) + .fetch_optional(&mut **tx) + .await?; + if let Some(document) = document { + return Ok(VaultEvent::Content { document }); + } + let file_manifest = Self::file_manifest_by_id(tx, event_id) + .await? + .context("Event has no normalized content or file manifest")?; + Ok(VaultEvent::FileManifest { file_manifest }) } pub async fn latest_event_id(tx: &mut Transaction<'_>) -> Result { @@ -126,17 +167,21 @@ impl Database { "Event cursor is outside this vault's history" ); - let rows: Vec = sqlx::query_scalar( - "SELECT event_json FROM events WHERE event_id > ? ORDER BY event_id", + let rows: Vec<(VaultUpdateId, String)> = sqlx::query_as( + "SELECT event_id, request_id FROM events WHERE event_id > ? ORDER BY event_id", ) .bind(after) .fetch_all(&mut *tx) .await?; - let events = rows - .iter() - .map(|json| serde_json::from_str(json)) - .collect::>()?; + let mut events = Vec::with_capacity(rows.len()); + for (event_id, request_id) in rows { + events.push(EventRecord { + event_id, + request_id: request_id.parse()?, + event: Self::event_by_id(&mut tx, event_id).await?, + }); + } Ok(EventBatch { head_event_id, diff --git a/sync-server/src/app_state/database/tests.rs b/sync-server/src/app_state/database/tests.rs index 3181384a..9b5b1e62 100644 --- a/sync-server/src/app_state/database/tests.rs +++ b/sync-server/src/app_state/database/tests.rs @@ -1,5 +1,5 @@ use super::*; -use models::{DocumentId, EventRecord, FileManifest, StoredDocumentVersion, VaultEvent}; +use models::{DocumentId, FileManifest, StoredDocumentVersion}; fn config(directory: &tempfile::TempDir) -> DatabaseConfig { DatabaseConfig { @@ -32,18 +32,6 @@ async fn insert_content( Database::insert_document_version(&mut tx, &version) .await .unwrap(); - Database::write_event( - &mut tx, - &EventRecord { - event_id: version.vault_update_id, - request_id, - event: VaultEvent::Content { - document: (&version).into(), - }, - }, - ) - .await - .unwrap(); tx.commit().await.unwrap(); version } @@ -133,18 +121,6 @@ async fn snapshot_returns_latest_referenced_metadata_and_byte_lengths() { Database::insert_file_manifest(&mut tx, &manifest) .await .unwrap(); - Database::write_event( - &mut tx, - &EventRecord { - event_id: manifest.file_manifest_id, - request_id, - event: VaultEvent::FileManifest { - file_manifest: manifest.clone(), - }, - }, - ) - .await - .unwrap(); tx.commit().await.unwrap(); let snapshot = database.vault_snapshot(&vault).await.unwrap(); diff --git a/sync-server/src/server/endpoints/put_file_content.rs b/sync-server/src/server/endpoints/put_file_content.rs index ae04756f..4af4504c 100644 --- a/sync-server/src/server/endpoints/put_file_content.rs +++ b/sync-server/src/server/endpoints/put_file_content.rs @@ -4,7 +4,7 @@ use crate::{ AppState, database::{ Database, - models::{DocumentId, EventRecord, StoredDocumentVersion, VaultEvent}, + models::{DocumentId, StoredDocumentVersion, VaultEvent}, }, }, config::user_config::User, @@ -123,18 +123,6 @@ pub async fn put_file_content( let response = DocumentUpdateResponse::Accepted((&version).into()); - let event = EventRecord { - event_id: version.vault_update_id, - request_id: push.request_id, - event: VaultEvent::Content { - document: version.into(), - }, - }; - - Database::write_event(&mut tx, &event) - .await - .map_err(server_error)?; - tx.commit() .await .map_err(|error| server_error(error.into()))?; @@ -494,7 +482,7 @@ mod tests { } #[tokio::test] - async fn failed_event_commit_rolls_back_content_and_allows_retrying_the_request() { + async fn failed_normalized_event_write_rolls_back_and_allows_retrying_the_request() { let vault = TestVault::new().await; let mut tx = vault .state @@ -502,7 +490,7 @@ mod tests { .create_write_transaction(&vault.vault) .await .unwrap(); - sqlx::query("CREATE TRIGGER reject_event BEFORE UPDATE ON events BEGIN SELECT RAISE(ABORT, 'simulated event write failure'); END") + sqlx::query("CREATE TRIGGER reject_event BEFORE INSERT ON documents BEGIN SELECT RAISE(ABORT, 'simulated event write failure'); END") .execute(&mut *tx).await.unwrap(); tx.commit().await.unwrap(); diff --git a/sync-server/src/server/endpoints/put_file_manifest.rs b/sync-server/src/server/endpoints/put_file_manifest.rs index 945d4dcb..98b559d2 100644 --- a/sync-server/src/server/endpoints/put_file_manifest.rs +++ b/sync-server/src/server/endpoints/put_file_manifest.rs @@ -12,7 +12,7 @@ use crate::{ AppState, database::{ Database, - models::{EventRecord, FileManifest, VaultEvent}, + models::{FileManifest, VaultEvent}, }, }, errors::{SyncServerError, client_error, server_error}, @@ -91,16 +91,6 @@ pub async fn put_file_manifest( file_manifest_id: file_manifest.file_manifest_id, }; - let event = EventRecord { - event_id: file_manifest.file_manifest_id, - request_id: push.request_id, - event: VaultEvent::FileManifest { file_manifest }, - }; - - Database::write_event(&mut tx, &event) - .await - .map_err(server_error)?; - tx.commit() .await .map_err(|error| server_error(error.into()))?; diff --git a/sync-server/src/server/endpoints/utils.rs b/sync-server/src/server/endpoints/utils.rs index 75e08989..c0df4216 100644 --- a/sync-server/src/server/endpoints/utils.rs +++ b/sync-server/src/server/endpoints/utils.rs @@ -1,10 +1,7 @@ use anyhow::anyhow; use crate::{ - app_state::database::{ - Database, Transaction, - models::{EventRecord, VaultEvent}, - }, + app_state::database::{Database, Transaction, models::VaultEvent}, errors::{SyncServerError, client_error, server_error}, }; @@ -13,7 +10,7 @@ pub(super) async fn find_already_processed_event( request_id: uuid::Uuid, fingerprint: &[u8], ) -> Result, SyncServerError> { - let Some((stored_fingerprint, event_json)) = Database::get_request_event(tx, request_id) + let Some((stored_fingerprint, event)) = Database::get_request_event(tx, request_id) .await .map_err(server_error)? else { @@ -26,7 +23,5 @@ pub(super) async fn find_already_processed_event( ))); } - serde_json::from_str::(&event_json) - .map(|record| Some(record.event)) - .map_err(|error| server_error(error.into())) + Ok(Some(event)) } diff --git a/sync-server/src/utils/portable_path.rs b/sync-server/src/utils/portable_path.rs index 34551abf..7c170848 100644 --- a/sync-server/src/utils/portable_path.rs +++ b/sync-server/src/utils/portable_path.rs @@ -19,10 +19,7 @@ pub fn validate_file_manifest(entries: &FileManifestEntries) -> Result<()> { let mut nodes: BTreeMap = BTreeMap::new(); for path in entries.values() { - ensure!( - !path.is_empty() && path.len() <= 240, - "Path must contain 1–240 UTF-8 bytes" - ); + ensure!(!path.is_empty(), "Path must not be empty"); ensure!( path.nfc().collect::() == *path, @@ -136,16 +133,9 @@ mod tests { } #[test] - fn enforces_the_path_length_limit_in_utf8_bytes() { - for path in ["a".repeat(240), "é".repeat(120)] { + fn does_not_impose_a_cross_platform_total_path_limit() { + for path in ["a".repeat(300), "é".repeat(200)] { validate_file_manifest(&manifest(&[&path])).unwrap(); - - let too_long = format!("{path}a"); - - assert!( - validate_file_manifest(&manifest(&[&too_long])).is_err(), - "accepted a path over 240 bytes: {too_long:?}" - ); } }