andras/vault-link
andras/vault-link
Watch
1
0
Fork
You've already forked vault-link
 0
Code Issues Pull requests 18 Projects Releases 55 Packages Wiki Activity Actions 31

Review pass

Browse source
This commit is contained in:
Andras Schmelczer 2026-09-19 19:51:08 +01:00
commit 99fa298b92
1 changed files with 621 additions and 94 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -39,7 +39,9 @@ impl Cursors {
) {
let mut vault_to_cursors = self.vault_to_cursors.lock().await;
let all_device_cursors = vault_to_cursors.entry(vault_id).or_insert_with(Vec::new);
let all_device_cursors = vault_to_cursors
.entry(vault_id.clone())
.or_insert_with(Vec::new);
all_device_cursors.retain(|c| &c.client_cursors.device_id != device_id);
all_device_cursors.push(ClientCursorsWithTimeToLive::new(ClientCursors {
@ -48,8 +50,7 @@ impl Cursors {
documents_with_cursors: document_to_cursors,
}));
drop(vault_to_cursors); // Explicitly drop the lock before broadcasting to avoid deadlock
self.broadcast_cursors().await;
self.broadcast_cursors(&vault_id, all_device_cursors).await;
}
pub async fn get_cursors(&self, vault_id: &VaultId) -> Vec<ClientCursors> {
@ -78,31 +79,42 @@ impl Cursors {
async fn remove_expired_cursors(&self) {
let mut vault_to_cursors = self.vault_to_cursors.lock().await;
for (_vault_id, cursors) in vault_to_cursors.iter_mut() {
for (vault_id, cursors) in vault_to_cursors.iter_mut() {
let previous_len = cursors.len();
cursors.retain(|cursor| !cursor.is_expired(self.config.cursor_timeout));
if cursors.len() != previous_len {
self.broadcast_cursors(vault_id, cursors).await;
}
}
vault_to_cursors.retain(|_, cursors| !cursors.is_empty());
}
async fn broadcast_cursors(&self) {
let vault_to_cursors = self.vault_to_cursors.lock().await;
for (vault_id, cursors) in vault_to_cursors.iter() {
self.broadcasts
.send(
vault_id.clone(),
Notification::Cursors(CursorPositionFromServer {
clients: cursors.iter().map(|c| c.client_cursors.clone()).collect(),
}),
)
.await;
}
async fn broadcast_cursors(&self, vault_id: &str, cursors: &[ClientCursorsWithTimeToLive]) {
self.broadcasts
.send(
vault_id.to_owned(),
Notification::Cursors(CursorPositionFromServer {
clients: cursors.iter().map(|c| c.client_cursors.clone()).collect(),
}),
)
.await;
}
pub async fn remove_cursors_of_device(&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) {
let previous_len = cursors.len();
cursors.retain(|c| c.client_cursors.device_id != device_id);
if cursors.len() != previous_len {
self.broadcast_cursors(vault_id, cursors).await;
}
if cursors.is_empty() {
vault_to_cursors.remove(vault_id);
}
}
}
}
@ -125,3 +137,96 @@ impl ClientCursorsWithTimeToLive {
self.last_updated.elapsed() > ttl
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::config::server_config::ServerConfig;
#[tokio::test]
async fn disconnect_broadcasts_remaining_clients_and_the_final_empty_state() {
let broadcasts = Broadcasts::new(&ServerConfig::default());
let cursors = Cursors::new(&DatabaseConfig::default(), &broadcasts);
for device in ["one", "two"] {
cursors
.update_cursors(
"vault".to_owned(),
"user".to_owned(),
&device.to_owned(),
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;
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;
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;
assert!(receiver.try_recv().is_err());
}
#[tokio::test]
async fn expiration_broadcasts_only_changed_vaults_including_empty_ones() {
let broadcasts = Broadcasts::new(&ServerConfig::default());
let config = DatabaseConfig::default();
let cursors = Cursors::new(&config, &broadcasts);
for (vault, device) in [("vault", "expired"), ("vault", "live"), ("other", "live")] {
cursors
.update_cursors(
vault.to_owned(),
"user".to_owned(),
&device.to_owned(),
vec![],
)
.await;
}
cursors
.vault_to_cursors
.lock()
.await
.get_mut("vault")
.unwrap()[0]
.last_updated = std::time::Instant::now()
.checked_sub(config.cursor_timeout + Duration::from_secs(1))
.unwrap();
let mut receiver = broadcasts.get_receiver("vault".to_owned()).await;
let mut other = broadcasts.get_receiver("other".to_owned()).await;
cursors.remove_expired_cursors().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, "live");
assert!(other.try_recv().is_err());
cursors
.vault_to_cursors
.lock()
.await
.get_mut("vault")
.unwrap()[0]
.last_updated = std::time::Instant::now()
.checked_sub(config.cursor_timeout + Duration::from_secs(1))
.unwrap();
cursors.remove_expired_cursors().await;
let Notification::Cursors(update) = receiver.try_recv().unwrap() else {
panic!("expected cursors")
};
assert!(update.clients.is_empty());
assert!(cursors.get_cursors(&"vault".to_owned()).await.is_empty());
cursors.remove_expired_cursors().await;
assert!(receiver.try_recv().is_err());
assert!(other.try_recv().is_err());
}
}