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

looks ok

Browse source
This commit is contained in:
Andras Schmelczer 2026-04-21 22:35:30 +01:00
parent 5ee9db0007
commit 6a8c7635f1
8 changed files with 122 additions and 78 deletions
Download patch file Download diff file Expand all files Collapse all files

18
sync-server/src/app_state/cursors.rs
View file

@ -117,16 +117,14 @@ impl Cursors {
.unwrap_or_default()
};
self.broadcasts
.send_document_update(
vault_id.clone(),
WebSocketServerMessageWithOrigin::new(WebSocketServerMessage::CursorPositions(
CursorPositionFromServer {
clients: client_cursors,
},
)),
)
.await;
self.broadcasts.send_document_update(
vault_id.clone(),
WebSocketServerMessageWithOrigin::new(WebSocketServerMessage::CursorPositions(
CursorPositionFromServer {
clients: client_cursors,
},
)),
);
}
pub async fn remove_cursors_of_device(&self, vault_id: &VaultId, device_id: &DeviceId) {

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

@ -739,22 +739,23 @@ impl Database {
.await
.context("Failed to commit transaction")?;
// Both sends are synchronous: there's no `.await` between the
// `commit()` above and function return, so a task cancellation
// can't drop the broadcast and leave peers permanently behind.
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,
}),
),
)
.await;
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 {
@ -763,18 +764,19 @@ impl Database {
// 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(),
},
)),
)
.await;
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(),
updated_date: version.updated_date,
user_id: version.user_id.clone(),
device_id: version.device_id.clone(),
},
)),
);
}
Ok(())

35
sync-server/src/app_state/websocket/broadcasts.rs
View file

@ -1,4 +1,7 @@
use std::{collections::HashMap, sync::Arc};
use std::{
collections::HashMap,
sync::{Arc, Mutex as StdMutex},
};
use log::{debug, warn};
use tokio::sync::{Mutex, broadcast};
@ -9,7 +12,12 @@ use crate::{app_state::database::models::VaultId, config::server_config::ServerC
#[derive(Debug, Clone)]
pub struct Broadcasts {
broadcast_channel_capacity: usize,
tx: Arc<Mutex<HashMap<VaultId, broadcast::Sender<WebSocketServerMessageWithOrigin>>>>,
// `tx` uses a blocking std::sync::Mutex because the critical section is
// a HashMap lookup plus a synchronous `broadcast::Sender::send`. Making
// this non-async lets `send_document_update` run without an `.await`,
// so an axum handler that is cancelled between `transaction.commit()`
// and the broadcast can never drop the notification mid-flight.
tx: Arc<StdMutex<HashMap<VaultId, broadcast::Sender<WebSocketServerMessageWithOrigin>>>>,
send_locks: Arc<Mutex<HashMap<VaultId, Arc<tokio::sync::Mutex<()>>>>>,
}
@ -19,7 +27,7 @@ impl Broadcasts {
pub fn new(server_config: &ServerConfig) -> Self {
Self {
broadcast_channel_capacity: server_config.broadcast_channel_capacity,
tx: Arc::new(Mutex::new(HashMap::new())),
tx: Arc::new(StdMutex::new(HashMap::new())),
send_locks: Arc::new(Mutex::new(HashMap::new())),
}
}
@ -42,19 +50,25 @@ impl Broadcasts {
tx_map.retain(|_, sender| sender.receiver_count() > 0);
}
pub async fn get_receiver(
pub fn get_receiver(
&self,
vault: VaultId,
max_clients: usize,
) -> Result<broadcast::Receiver<WebSocketServerMessageWithOrigin>, crate::errors::SyncServerError>
{
let mut tx_map = self.tx.lock().await;
let mut tx_map = self
.tx
.lock()
.expect("broadcasts.tx mutex poisoned — a previous holder panicked");
Self::prune_inactive_vaults(&mut tx_map);
let sender = tx_map
.entry(vault)
.or_insert_with(|| broadcast::channel(self.broadcast_channel_capacity).0);
// Hold the lock across the count check *and* the subscribe so the
// `max_clients` cap is atomic: two concurrent callers can't both
// observe `receiver_count() < max_clients` and both subscribe.
if sender.receiver_count() >= max_clients {
return Err(crate::errors::client_error(anyhow::anyhow!(
"Vault has reached the maximum number of clients ({max_clients})"
@ -65,13 +79,18 @@ impl Broadcasts {
}
/// Notify all clients (who are subscribed to the vault) about an update.
/// We only log failures and don't propagate them.
pub async fn send_document_update(
/// Synchronous: safe to invoke from a handler between `commit()` and
/// function return without worrying about task cancellation dropping
/// the broadcast mid-flight. Failures are logged, never propagated.
pub fn send_document_update(
&self,
vault: VaultId,
document: WebSocketServerMessageWithOrigin,
) {
let mut tx_map = self.tx.lock().await;
let mut tx_map = self
.tx
.lock()
.expect("broadcasts.tx mutex poisoned — a previous holder panicked");
Self::prune_inactive_vaults(&mut tx_map);
let sender = tx_map