andras/vault-link
1
0
Fork 0
Code Issues Pull requests 18 Projects Releases 55 Packages Wiki Activity Actions 1

no remote path chacnge

Browse source
This commit is contained in:
Andras Schmelczer 2026-04-24 21:33:00 +01:00
parent 19d5dc1999
commit 17a1f4d060
16 changed files with 93 additions and 314 deletions
Download patch file Download diff file Expand all files Collapse all files

70
sync-server/src/app_state/database.rs
View file

@ -20,24 +20,6 @@ pub mod models;
#[error("Database is busy")]
pub struct WriteBusyError;
/// Tells [`Database::insert_document_version`] which WebSocket events the
/// just-committed version should produce. The caller is the only party
/// with enough context to decide this (the DB layer has no access to
/// "what the client sent" or "what the prior version looked like").
#[derive(Debug, Clone, Copy, Default)]
pub struct InsertBroadcast {
/// Emit a `VaultUpdate` (filtered from the origin device). Set when
/// the stored bytes differ from the prior version's bytes — i.e.
/// peers need to pull new content.
pub content_changed: bool,
/// Emit a `PathChange` (delivered to every client, origin included).
/// Set when the stored path differs from the prior stored path *or*
/// from the path the origin client sent — i.e. someone needs to
/// reconcile a dedupe, rename, or first-rename-wins outcome.
pub path_changed: bool,
}
use sqlx::{
Pool, Sqlite, pool::PoolConnection, sqlite::SqliteConnection, sqlite::SqlitePoolOptions,
};
@ -47,10 +29,7 @@ use uuid::fmt::Hyphenated;
use super::websocket::{
broadcasts::Broadcasts,
models::{
WebSocketServerMessage, WebSocketServerMessageWithOrigin, WebSocketVaultPathChange,
WebSocketVaultUpdate,
},
models::{WebSocketServerMessage, WebSocketServerMessageWithOrigin, WebSocketVaultUpdate},
};
use crate::config::database_config::DatabaseConfig;
use crate::consts::IDLE_POOL_TIMEOUT;
@ -693,7 +672,6 @@ impl Database {
vault_id: &VaultId,
version: &StoredDocumentVersion,
mut transaction: WriteTransaction,
broadcast: InsertBroadcast,
) -> Result<()> {
let document_id = version.document_id.as_hyphenated();
let query = sqlx::query!(
@ -739,39 +717,19 @@ impl Database {
.await
.context("Failed to commit transaction")?;
if broadcast.content_changed {
// Content events are filtered out for the origin device — the
// origin already has the content (or learns about the merge
// via the HTTP response).
self.broadcasts.send_document_update(
vault_id.clone(),
WebSocketServerMessageWithOrigin::with_origin(
version.device_id.clone(),
WebSocketServerMessage::VaultUpdate(WebSocketVaultUpdate {
documents: vec![version.clone().into()],
is_initial_sync: false,
}),
),
);
}
if broadcast.path_changed {
// Path change events intentionally carry no origin so *every*
// connected client (including the one that made the write)
// receives them. The create/update HTTP response no longer
// carries `relative_path`, so the origin device relies on this
// event to learn the server-canonical path.
self.broadcasts.send_document_update(
vault_id.clone(),
WebSocketServerMessageWithOrigin::new(WebSocketServerMessage::PathChange(
WebSocketVaultPathChange {
vault_update_id: version.vault_update_id,
document_id: version.document_id,
relative_path: version.relative_path.clone(),
},
)),
);
}
// The broadcast is delivered to every connected client except the
// author — the send task filters on `origin_device_id` (see
// `websocket.rs`). The origin already has authoritative state
// from the HTTP response that triggered this write.
self.broadcasts.send_document_update(
vault_id.clone(),
WebSocketServerMessageWithOrigin::with_origin(
version.device_id.clone(),
WebSocketServerMessage::VaultUpdate(WebSocketVaultUpdate {
document: version.clone().into(),
}),
),
);
Ok(())
}

25
sync-server/src/app_state/websocket/models.rs
View file

@ -58,23 +58,15 @@ pub struct CursorPositionFromServer {
pub clients: Vec<ClientCursors>,
}
// Clients only get notified of other clients' updates through WebSocketVaultUpdate.
// One committed version, broadcast to every connected client *except*
// the device that authored it — that device already has the new state
// via its HTTP response. The server also emits these one-at-a-time to
// catch up a freshly-connected client on versions committed while it
// was offline, in ascending `vault_update_id` order.
#[derive(TS, Serialize, Clone, Debug)]
#[serde(rename_all = "camelCase")]
pub struct WebSocketVaultUpdate {
pub documents: Vec<DocumentVersionWithoutContent>,
pub is_initial_sync: bool,
}
// Clients get notified of both their own and other clients' path changes through WebSocketVaultPathChange.
// This is becuase we must absolutely order path updates as they may all depend on all previous updates.
#[derive(TS, Serialize, Clone, Debug)]
#[serde(rename_all = "camelCase")]
pub struct WebSocketVaultPathChange {
#[ts(type = "number")]
pub vault_update_id: VaultUpdateId,
pub document_id: DocumentId,
pub relative_path: String,
pub document: DocumentVersionWithoutContent,
}
#[derive(TS, Deserialize, Clone, Debug)]
@ -90,10 +82,13 @@ pub enum WebSocketClientMessage {
#[ts(export)]
pub enum WebSocketServerMessage {
VaultUpdate(WebSocketVaultUpdate),
PathChange(WebSocketVaultPathChange),
CursorPositions(CursorPositionFromServer),
}
/// Broadcast envelope carrying the message plus the device that produced
/// it. The per-recipient send task compares `origin_device_id` against
/// its own device id to fill in `originates_from_self` before the message
/// is serialized on the wire.
#[derive(Clone, Debug)]
pub struct WebSocketServerMessageWithOrigin {
pub origin_device_id: Option<DeviceId>,

19
sync-server/src/server/create_document.rs
View file

@ -11,10 +11,7 @@ use super::{device_id_header::DeviceIdHeader, requests::CreateDocumentVersion};
use crate::{
app_state::{
AppState,
database::{
InsertBroadcast,
models::{StoredDocumentVersion, VaultId},
},
database::models::{StoredDocumentVersion, VaultId},
},
config::user_config::User,
errors::{SyncServerError, client_error, server_error, write_transaction_error},
@ -128,8 +125,6 @@ pub async fn create_document(
);
}
let path_changed = deduped_path != sanitized_relative_path;
let new_vault_update_id = last_update_id + 1;
let new_version = StoredDocumentVersion {
vault_update_id: new_vault_update_id,
@ -146,17 +141,7 @@ pub async fn create_document(
state
.database
.insert_document_version(
&vault_id,
&new_version,
transaction,
InsertBroadcast {
// A brand-new document is always a content change for peers.
content_changed: true,
// Origin needs to know if the server deduped its requested path.
path_changed,
},
)
.insert_document_version(&vault_id, &new_version, transaction)
.await
.map_err(server_error)?;

17
sync-server/src/server/delete_document.rs
View file

@ -11,10 +11,7 @@ use super::device_id_header::DeviceIdHeader;
use crate::{
app_state::{
AppState,
database::{
InsertBroadcast,
models::{DocumentId, DocumentVersionWithoutContent, StoredDocumentVersion, VaultId},
},
database::models::{DocumentId, DocumentVersionWithoutContent, StoredDocumentVersion, VaultId},
},
config::user_config::User,
errors::{SyncServerError, not_found_error, server_error, write_transaction_error},
@ -101,17 +98,7 @@ pub async fn delete_document(
state
.database
.insert_document_version(
&vault_id,
&new_version,
transaction,
InsertBroadcast {
// Deletion is a content change peers must learn about.
content_changed: true,
// Delete never renames.
path_changed: false,
},
)
.insert_document_version(&vault_id, &new_version, transaction)
.await
.map_err(server_error)?;

20
sync-server/src/server/update_document.rs
View file

@ -17,7 +17,7 @@ use crate::{
app_state::{
AppState,
database::{
InsertBroadcast, WriteTransaction,
WriteTransaction,
models::{DocumentId, StoredDocumentVersion, VaultId, VaultUpdateId},
},
},
@ -292,14 +292,6 @@ pub async fn update_document(
latest_version.relative_path.clone()
};
let content_changed = merged_content != latest_version.content;
// Stored path differs from either the prior stored path (peers need
// to learn about the rename) or from the path the origin sent
// (origin needs to learn if its rename was deduped or rejected by
// first-rename-wins).
let path_changed = new_relative_path != latest_version.relative_path
|| new_relative_path != sanitized_relative_path;
let new_version = StoredDocumentVersion {
document_id,
vault_update_id: last_update_id + 1,
@ -315,15 +307,7 @@ pub async fn update_document(
state
.database
.insert_document_version(
&vault_id,
&new_version,
transaction,
InsertBroadcast {
content_changed,
path_changed,
},
)
.insert_document_version(&vault_id, &new_version, transaction)
.await
.map_err(server_error)?;

29
sync-server/src/server/websocket.rs
View file

@ -134,19 +134,21 @@ async fn websocket(
}
};
send_update_over_websocket(
&WebSocketServerMessage::VaultUpdate(WebSocketVaultUpdate {
documents: get_unseen_documents(
&state,
&vault_id,
authed_handshake.handshake.last_seen_vault_update_id,
)
.await?,
is_initial_sync: true,
}),
&mut sender,
// Catch-up on versions committed while this client was offline,
// streamed one-at-a-time in ascending `vault_update_id` order
let unseen_documents = get_unseen_documents(
&state,
&vault_id,
authed_handshake.handshake.last_seen_vault_update_id,
)
.await?;
for document in unseen_documents {
send_update_over_websocket(
&WebSocketServerMessage::VaultUpdate(WebSocketVaultUpdate { document }),
&mut sender,
)
.await?;
}
send_update_over_websocket(
&WebSocketServerMessage::CursorPositions(CursorPositionFromServer {
@ -161,6 +163,8 @@ async fn websocket(
loop {
match broadcast_receiver.recv().await {
Ok(update) => {
// Drop messages this device authored because the HTTP
// response already carried authoritative state back.
if Some(&device_id) == update.origin_device_id.as_ref() {
continue;
}
@ -174,8 +178,7 @@ async fn websocket(
.filter(|client| client.device_id != device_id)
.collect(),
}),
WebSocketServerMessage::VaultUpdate(_)
| WebSocketServerMessage::PathChange(_) => update.message,
WebSocketServerMessage::VaultUpdate(_) => update.message,
};
send_update_over_websocket(&message, &mut sender).await?;