Use new reconcile api

This commit is contained in:
Andras Schmelczer 2026-09-22 21:20:42 +01:00
commit 49065f5b6d
6 changed files with 127 additions and 86 deletions

41
sync-server/Cargo.lock generated
View file

@ -1582,12 +1582,12 @@ dependencies = [
[[package]] [[package]]
name = "reconcile-text" name = "reconcile-text"
version = "0.8.0" version = "0.12.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "599cf9539996a2a19e501110404c59ba62f4974009f8fb864a8b7151c15ee5a5" checksum = "8eb67d18cd23948a32b940e137128552105970913d5747daa5860a58819a7453"
dependencies = [ dependencies = [
"serde", "serde",
"thiserror 2.0.17", "thiserror 2.0.20",
] ]
[[package]] [[package]]
@ -1907,7 +1907,7 @@ dependencies = [
"serde_json", "serde_json",
"sha2", "sha2",
"smallvec", "smallvec",
"thiserror 2.0.17", "thiserror 2.0.20",
"tokio", "tokio",
"tokio-stream", "tokio-stream",
"tracing", "tracing",
@ -1991,7 +1991,7 @@ dependencies = [
"smallvec", "smallvec",
"sqlx-core", "sqlx-core",
"stringprep", "stringprep",
"thiserror 2.0.17", "thiserror 2.0.20",
"tracing", "tracing",
"uuid", "uuid",
"whoami", "whoami",
@ -2030,7 +2030,7 @@ dependencies = [
"smallvec", "smallvec",
"sqlx-core", "sqlx-core",
"stringprep", "stringprep",
"thiserror 2.0.17", "thiserror 2.0.20",
"tracing", "tracing",
"uuid", "uuid",
"whoami", "whoami",
@ -2056,7 +2056,7 @@ dependencies = [
"serde", "serde",
"serde_urlencoded", "serde_urlencoded",
"sqlx-core", "sqlx-core",
"thiserror 2.0.17", "thiserror 2.0.20",
"tracing", "tracing",
"url", "url",
"uuid", "uuid",
@ -2112,6 +2112,17 @@ dependencies = [
"unicode-ident", "unicode-ident",
] ]
[[package]]
name = "syn"
version = "3.0.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8593e8e72159ed2257d083c7a454a85cbf854f37a0966d8d483aff8c8a3ebcee"
dependencies = [
"proc-macro2",
"quote",
"unicode-ident",
]
[[package]] [[package]]
name = "sync_server" name = "sync_server"
version = "0.14.0" version = "0.14.0"
@ -2136,7 +2147,7 @@ dependencies = [
"sha2", "sha2",
"sqlx", "sqlx",
"tempfile", "tempfile",
"thiserror 2.0.17", "thiserror 2.0.20",
"tokio", "tokio",
"tower", "tower",
"tower-http", "tower-http",
@ -2197,11 +2208,11 @@ dependencies = [
[[package]] [[package]]
name = "thiserror" name = "thiserror"
version = "2.0.17" version = "2.0.20"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f63587ca0f12b72a0600bcba1d40081f830876000bb46dd2337a3051618f4fc8" checksum = "ec86235f5fcc2a73650310756d2ac5b138a5780bbbdfae3eeccec992c435ba4f"
dependencies = [ dependencies = [
"thiserror-impl 2.0.17", "thiserror-impl 2.0.20",
] ]
[[package]] [[package]]
@ -2217,13 +2228,13 @@ dependencies = [
[[package]] [[package]]
name = "thiserror-impl" name = "thiserror-impl"
version = "2.0.17" version = "2.0.20"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3ff15c8ecd7de3849db632e14d18d2571fa09dfc5ed93479bc4485c7a517c913" checksum = "bc04cd3e1236dd4a98afca4569f2deb3f120e5422a4023be2cb683f8486292af"
dependencies = [ dependencies = [
"proc-macro2", "proc-macro2",
"quote", "quote",
"syn 2.0.90", "syn 3.0.6",
] ]
[[package]] [[package]]
@ -2428,7 +2439,7 @@ checksum = "e640d9b0964e9d39df633548591090ab92f7a4567bc31d3891af23471a3365c6"
dependencies = [ dependencies = [
"chrono", "chrono",
"lazy_static", "lazy_static",
"thiserror 2.0.17", "thiserror 2.0.20",
"ts-rs-macros", "ts-rs-macros",
"uuid", "uuid",
] ]

View file

@ -34,7 +34,7 @@ sha2 = "0.10"
unicode-normalization = "0.1" unicode-normalization = "0.1"
ts-rs = { version = "10.1", features = ["uuid-impl", "chrono-impl", "no-serde-warnings"] } ts-rs = { version = "10.1", features = ["uuid-impl", "chrono-impl", "no-serde-warnings"] }
base64 = "0.22.1" base64 = "0.22.1"
reconcile-text = { version = "0.8.0", features = ["serde"] } reconcile-text = { version = "0.12.1", features = ["serde"] }
[dev-dependencies] [dev-dependencies]
tempfile = "3" tempfile = "3"

View file

@ -1,6 +1,9 @@
# Sync server # Sync server
## API v4
- ban using unwraps
## API v5
The server is a compare-and-swap store. Content versions do not contain paths or 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. deletions. A separately versioned file manifest owns the namespace and membership.
@ -8,14 +11,15 @@ All routes below are under `/vaults/:vault_id` and require bearer authentication
(except the WebSocket, which authenticates its first message). (except the WebSocket, which authenticates its first message).
| Endpoint | Purpose | | Endpoint | Purpose |
| --- | --- | | ---------------------------------------------------------- | --------------------------------------------------------------------------------------- |
| `PUT /documents/:document_id` | CAS an immutable content version. | | `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` | Current content version, including base64 bytes. |
| `GET /documents/:document_id/versions/:version_id/content` | Immutable raw bytes. | | `GET /documents/:document_id/versions/:version_id/content` | Immutable raw bytes. |
| `GET /file_manifest` | Current file manifest. | | `GET /file-manifest` | Current file manifest. |
| `PUT /file_manifest` | CAS the complete 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 /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 /events-since?after=N` | Every event after N, in order, with the current watermark. |
| `GET /ws` | Ordered event delivery and ephemeral cursor positions. | | `GET /ws` | Ordered event delivery and ephemeral cursor positions. |
A content push includes a `Device-Id` header and this JSON body: A content push includes a `Device-Id` header and this JSON body:
@ -54,13 +58,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 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 within a version; portable path validation also rejects aliases and directory
conflicts. Reads reconstruct the complete map, and vault snapshots join these conflicts. Reads reconstruct the complete map, and vault snapshots join these
rows to the latest content versions. The version, entries, and complete file rows to the latest content versions. Event replay and retry acknowledgements are
manifest event in `events.event_json` commit atomically. Event replay and retry also reconstructed from normalized content/manifest rows, without storing a
acknowledgements continue to use `event_json`. second complete map in the event table.
Both pushes return a tagged `Accepted` or `StaleBase` response. An accepted file Both pushes return a tagged `Accepted` or `StaleBase` response. An accepted file
manifest response returns its assigned file manifest ID. A stale content response 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, Accepted requests store their unique request ID and fingerprint in the event row,
atomically with the version. Retrying the exact request reconstructs the original atomically with the version. Retrying the exact request reconstructs the original
`Accepted` response from that event before checking the current head, even after `Accepted` response from that event before checking the current head, even after
@ -70,8 +75,8 @@ identity is not part of the fingerprint.
File manifest validation rejects duplicate UUID keys (including aliases), missing File manifest validation rejects duplicate UUID keys (including aliases), missing
content, path aliases, inconsistent directory spelling, and file/ancestor content, path aliases, inconsistent directory spelling, and file/ancestor
conflicts. Paths use `/`, NFC normalization, at most 240 UTF-8 bytes, and portable conflicts. Paths use `/`, NFC normalization, and portable Windows filename
Windows filename restrictions. Alias comparison is NFC → Unicode uppercase → 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 NFC. `.vault-link-sync` and its aliases are reserved. The server never sanitizes
or allocates a replacement name. or allocates a replacement name.
@ -84,14 +89,20 @@ content/file-manifest version IDs are the event IDs that created them. Rejected
requests and retries of accepted requests append no events. Timestamps never requests and retries of accepted requests append no events. Timestamps never
determine event ordering. determine event ordering.
The WebSocket handshake retains `lastSeenVaultUpdateId`. The server sends WebSocket messages are `{type: "vaultChanged"}` hints. Clients fetch ordered
`{type: "vaultEvents", headEventId, events}` in increasing order, including events history through HTTP from their last incorporated event ID. The server compares
originated by that client. Each connection drains the database through one sender. its current history checkpoint after broadcasts and periodically, so missed wakes
In-memory notifications only wake it; polling covers missed wakes and cancellation and requests interrupted after commit still trigger catchup. Reconnecting sends a
after commit. Reconnect with the last durably processed event ID. The HTTP replay fresh hint without replaying events over the socket. Cursor messages are unchanged.
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
A fresh client reads `/vault_snapshot`, applies it durably, then replays after its watermark. Successful HTTP responses include `X-Vault-Link-History`, containing
watermark. 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 notifications are ephemeral and do not consume event IDs.
Cursor expiry and disconnect broadcast the remaining clients, including an empty Cursor expiry and disconnect broadcast the remaining clients, including an empty
list when the last client leaves. list when the last client leaves.
@ -116,8 +127,20 @@ flushes database directory entries; the filesystem must support those durability
operations. Content, file manifests, and events (including request fingerprints) operations. Content, file manifests, and events (including request fingerprints)
are retained indefinitely. are retained indefinitely.
API v4 requires fresh vault databases and upgraded clients. Existing database Deploy API v5 clients and server together. The v4 database schema migrates by
migration checksums intentionally reject earlier schemas; there is no automatic adding event-incarnation tokens without deleting content or receipts. Schemas
erasure or migration. Use a new database directory and preserve old data. 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 Base64 increases snapshot body size by about one third; request limits apply to
the encoded JSON body. the encoded JSON body.

View file

@ -1,5 +1,5 @@
[toolchain] [toolchain]
channel = "1.89.0" channel = "1.94.0"
targets = [ targets = [
"x86_64-unknown-linux-gnu", "x86_64-unknown-linux-gnu",
"x86_64-unknown-linux-musl", "x86_64-unknown-linux-musl",

View file

@ -242,18 +242,18 @@ async fn manifest_existence_checks_are_independent_of_payload_decoding() {
.await .await
.unwrap(); .unwrap();
assert_eq!( assert_eq!(
Database::missing_document(&mut tx, &[id]).await.unwrap(), Database::get_missing_document(&mut tx, &[id]).await.unwrap(),
None None
); );
let missing = uuid::Uuid::new_v4(); let missing = uuid::Uuid::new_v4();
assert_eq!( assert_eq!(
Database::missing_document(&mut tx, &[id, missing]) Database::get_missing_document(&mut tx, &[id, missing])
.await .await
.unwrap(), .unwrap(),
Some(missing.to_string()) Some(missing.to_string())
); );
assert_eq!( assert_eq!(
Database::missing_document(&mut tx, &[]).await.unwrap(), Database::get_missing_document(&mut tx, &[]).await.unwrap(),
None None
); );
} }

View file

@ -78,33 +78,15 @@ pub async fn put_file_content(
))); )));
} }
let content = match push.content { let (tx, content) = tokio::task::spawn_blocking(move || {
PushContent::Snapshot(bytes) => STANDARD let content = decode_content(parent, push.content);
.decode(bytes) (tx, content)
.map_err(|error| client_error(error.into()))?, })
.await
.map_err(|error| server_error(error.into()))?;
PushContent::Diff(diff) => { let mut tx = tx;
if diff let content = content.map_err(client_error)?;
.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 version = StoredDocumentVersion { let version = StoredDocumentVersion {
vault_update_id: Database::allocate_event(&mut tx, push.request_id, &fingerprint) vault_update_id: Database::allocate_event(&mut tx, push.request_id, &fingerprint)
@ -132,8 +114,32 @@ pub async fn put_file_content(
Ok(Json(response)) Ok(Json(response))
} }
fn decode_content(
parent: Option<StoredDocumentVersion>,
content: PushContent,
) -> anyhow::Result<Vec<u8>> {
let bytes = match content {
PushContent::Snapshot(encoded) => STANDARD.decode(encoded)?,
PushContent::Diff(diff) => {
let parent = parent.ok_or_else(|| anyhow!("Diff requires a parent"))?;
let text = str::from_utf8(&parent.content)?;
anyhow::ensure!(
!diff
.iter()
.any(|item| matches!(item, NumberOrText::Number(i64::MIN))),
"Diff contains an out-of-range length"
);
let edited = EditedText::from_diff(text, diff, &*BuiltinTokenizer::Character)?;
edited.apply().text().into_bytes()
}
};
Ok(bytes)
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use reconcile_text::EditedText;
use std::path::PathBuf; use std::path::PathBuf;
use super::*; use super::*;
@ -148,6 +154,7 @@ mod tests {
Config, database_config::DatabaseConfig, server_config::ServerConfig, Config, database_config::DatabaseConfig, server_config::ServerConfig,
user_config::VaultAccess, user_config::VaultAccess,
}, },
consts::DEFAULT_EVENTS_PAGE_SIZE,
}; };
async fn send_document( async fn send_document(
@ -363,11 +370,8 @@ mod tests {
outcomes => panic!("Expected one accepted push and one stale base: {outcomes:?}"), outcomes => panic!("Expected one accepted push and one stale base: {outcomes:?}"),
}; };
let latest = vault.latest(document).await; let latest = vault.latest(document).await;
assert_eq!(winner.vault_update_id, rejected.metadata.vault_update_id); assert_eq!(winner.vault_update_id, rejected.vault_update_id);
assert_eq!( assert_eq!(rejected.content_size, latest.content.len());
STANDARD.decode(rejected.content_base64).unwrap(),
latest.content
);
assert!(latest.content == b"left" || latest.content == b"right"); assert!(latest.content == b"left" || latest.content == b"right");
let original = vault let original = vault
.state .state
@ -446,7 +450,7 @@ mod tests {
} }
let events = reopened let events = reopened
.database .database
.events_after(&vault.vault, 0) .events_after(&vault.vault, 0, DEFAULT_EVENTS_PAGE_SIZE)
.await .await
.unwrap(); .unwrap();
assert_eq!(events.head_event_id, latest.vault_update_id); assert_eq!(events.head_event_id, latest.vault_update_id);
@ -473,7 +477,7 @@ mod tests {
let events = vault let events = vault
.state .state
.database .database
.events_after(&vault.vault, 0) .events_after(&vault.vault, 0, DEFAULT_EVENTS_PAGE_SIZE)
.await .await
.unwrap(); .unwrap();
assert_eq!(events.head_event_id, 1); assert_eq!(events.head_event_id, 1);
@ -510,7 +514,7 @@ mod tests {
let events = vault let events = vault
.state .state
.database .database
.events_after(&vault.vault, 0) .events_after(&vault.vault, 0, DEFAULT_EVENTS_PAGE_SIZE)
.await .await
.unwrap(); .unwrap();
assert_eq!(events.head_event_id, 0); assert_eq!(events.head_event_id, 0);
@ -534,7 +538,7 @@ mod tests {
let events = vault let events = vault
.state .state
.database .database
.events_after(&vault.vault, 0) .events_after(&vault.vault, 0, DEFAULT_EVENTS_PAGE_SIZE)
.await .await
.unwrap(); .unwrap();
assert_eq!(events.events.len(), 1); assert_eq!(events.events.len(), 1);
@ -549,8 +553,11 @@ mod tests {
let desired = "Hi beautiful world 🌍"; let desired = "Hi beautiful world 🌍";
let base = accepted(vault.send(snapshot(document, None, original)).await); let base = accepted(vault.send(snapshot(document, None, original)).await);
let mut update = snapshot(document, Some(base.vault_update_id), ""); let mut update = snapshot(document, Some(base.vault_update_id), "");
update.content = update.content = PushContent::Diff(
PushContent::Diff(EditedText::from_strings(original, &desired.into()).to_diff()); EditedText::from_strings(original, &desired.into())
.to_diff()
.unwrap(),
);
let updated = accepted(vault.send(update).await); let updated = accepted(vault.send(update).await);
assert_eq!(vault.latest(document).await.content, desired.as_bytes()); assert_eq!(vault.latest(document).await.content, desired.as_bytes());
@ -577,7 +584,7 @@ mod tests {
assert!(matches!( assert!(matches!(
vault.send(foreign_base).await, vault.send(foreign_base).await,
DocumentUpdateResponse::StaleBase(version) 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(); let binary_document = uuid::Uuid::new_v4();