This commit is contained in:
Mateusz Gruszczyński committed 2026-10-14 13:38:59 +02:00
1 parent df403c0aa7
commit 9e133628e5
17 files changed
+270 -34

No files matched your search

+18 -11
View File
@@ -863,6 +863,7 @@ async fn handle_socket(
access_refresh.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
access_refresh.tick().await;
let (mut sender, mut receiver) = socket.split();
let mut last_broadcast_access_check = Instant::now() - Duration::from_secs(2);
if send_split(
&mut sender,
&ServerMessage::Diagnostics {
@@ -1053,17 +1054,23 @@ async fn handle_socket(
}
continue;
}
let (read_allowed, _) = current_resource_access(
&state,
"workspace",
&workspace_slug,
&external_tokens,
session_token.as_deref(),
password_ok,
).await;
if !read_allowed {
let _=send_split(&mut sender,&ServerMessage::Error{message:"Access expired or revoked".into()}).await;
break;
// Broadcasts can arrive hundreds of times a second per connection.
// Keep access checks off the per-message hot path; the separate
// access_refresh ticker also enforces revocation while idle.
if last_broadcast_access_check.elapsed() >= Duration::from_secs(1) {
let (read_allowed, _) = current_resource_access(
&state,
"workspace",
&workspace_slug,
&external_tokens,
session_token.as_deref(),
password_ok,
).await;
if !read_allowed {
let _=send_split(&mut sender,&ServerMessage::Error{message:"Access expired or revoked".into()}).await;
break;
}
last_broadcast_access_check = Instant::now();
}
match update {
Ok(RoomEvent::Document(update))=>if send_split(&mut sender,&ServerMessage::Document{base_revision_id:update.base_revision_id,revision_id:update.revision_id,updated_at:update.updated_at,author:update.author,client_id:update.client_id,update_id:update.update_id,operation:update.operation,owner_replacements:update.owner_replacements}).await.is_err(){break;},
+18 -11
View File
@@ -356,6 +356,7 @@ async fn handle_pad_socket(
.await;
let _ = channel.send(RoomEvent::Presence(users));
let mut last_chat = Instant::now() - Duration::from_secs(1);
let mut last_broadcast_access_check = Instant::now() - Duration::from_secs(2);
let mut access_refresh = tokio::time::interval(Duration::from_secs(10));
access_refresh.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
access_refresh.tick().await;
@@ -552,17 +553,23 @@ async fn handle_pad_socket(
}
continue;
}
let (read_allowed, _) = current_resource_access(
&state,
"pad",
&slug,
&external_tokens,
session_token.as_deref(),
password_ok,
).await;
if !read_allowed {
let _=send_pad_split(&mut sender,&PadServerMessage::Error{message:"Access expired or revoked".into()}).await;
break;
// Broadcasts can arrive hundreds of times a second per connection.
// Keep access checks off the per-message hot path; the separate
// access_refresh ticker also enforces revocation while idle.
if last_broadcast_access_check.elapsed() >= Duration::from_secs(1) {
let (read_allowed, _) = current_resource_access(
&state,
"pad",
&slug,
&external_tokens,
session_token.as_deref(),
password_ok,
).await;
if !read_allowed {
let _=send_pad_split(&mut sender,&PadServerMessage::Error{message:"Access expired or revoked".into()}).await;
break;
}
last_broadcast_access_check = Instant::now();
}
match update {
Ok(RoomEvent::Document(u))=>if send_pad_split(&mut sender,&PadServerMessage::Document{base_revision_id:u.base_revision_id,revision_id:u.revision_id,updated_at:u.updated_at,author:u.author,client_id:u.client_id,update_id:u.update_id,operation:u.operation,owner_replacements:u.owner_replacements}).await.is_err(){break;},