Skip to content

fix(ui): defer signal mutations via setTimeout to prevent lost subscriptions #172

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sanity merged 2 commits into from
Mar 15, 2026
+134 −53
Merged

fix(ui): defer signal mutations via setTimeout to prevent lost subscriptions #172

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
af14b73
fix(ui): defer signal mutations via setTimeout to prevent lost subscr…
sanity Mar 15, 2026
a1989eb
fix: set sync baseline to incoming contract state, not post-merge state
sanity Mar 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 11 additions & 1 deletion AGENTS.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -334,10 +334,20 @@ These rules prevent re-entrant borrow crashes.
// WRONG — panics if signal is being written
let rooms = ROOMS.read();

// RIGHT — returns Err instead of panicking, still registers Dioxus subscriptions
// RIGHT — returns Err instead of panicking
let Ok(rooms) = ROOMS.try_read() else { return; };
```

**IMPORTANT:** In Dioxus 0.7.x, `try_read()` does NOT register signal subscriptions
when it returns `Err`. The subscription is registered only on the success path
(after the borrow succeeds). This means a `use_memo` that hits `try_read() -> Err`
will NOT be notified of future signal changes — it permanently stops re-evaluating.

To mitigate: ensure signal mutations happen in clean execution contexts (via
`setTimeout(0)` deferral) so `try_read()` never encounters a concurrent borrow.
Also, memos that read multiple signals (e.g., `CURRENT_ROOM.read()` + `ROOMS.try_read()`)
get a backup subscription from the non-try signal.

### Never call `spawn_local` inside a polled future

Use `safe_spawn_local()` (in `util.rs`) which defers via `setTimeout(0)`:
Expand Down
175 changes: 123 additions & 52 deletions ui/src/components/app/freenet_api/room_synchronizer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,13 +42,45 @@ pub struct RoomSynchronizer {
}

impl RoomSynchronizer {
/// Applies a delta update to a room's state.
///
/// Like update_room_state, deferred via setTimeout(0) on WASM to prevent
/// re-entrant signal borrow issues. See update_room_state docs for details.
pub(crate) fn apply_delta(&self, owner_vk: &VerifyingKey, delta: ChatRoomStateV1Delta) {
let owner_vk = *owner_vk;

#[cfg(target_arch = "wasm32")]
{
use wasm_bindgen::prelude::*;
let cb = Closure::once_into_js(move || {
Self::apply_delta_inner(owner_vk, delta);
});
web_sys::window()
.expect("no window")
.set_timeout_with_callback(&cb.into())
.ok();
}

#[cfg(not(target_arch = "wasm32"))]
Self::apply_delta_inner(owner_vk, delta);
}

/// Inner implementation of apply_delta, runs in a clean execution context on WASM.
fn apply_delta_inner(owner_vk: VerifyingKey, delta: ChatRoomStateV1Delta) {
// Extract new messages for notifications before entering the mutable borrow
let new_messages = delta.recent_messages.clone();

// Will be populated inside with_mut if new messages need notification
let mut pending_notification: Option<(
Vec<_>,
MemberId,
MemberInfoV1,
HashMap<u32, [u8; 32]>,
)> = None;

ROOMS.with_mut(|rooms| {
if let Some(room_data) = rooms.map.get_mut(owner_vk) {
let params = ChatRoomParametersV1 { owner: *owner_vk };
if let Some(room_data) = rooms.map.get_mut(&owner_vk) {
let params = ChatRoomParametersV1 { owner: owner_vk };

// Log the delta being applied, especially any member_info with versions
if let Some(member_info) = &delta.member_info {
Expand Down Expand Up @@ -130,38 +162,24 @@ impl RoomSynchronizer {
// Keep cached self membership data up to date
room_data.capture_self_membership_data(&params);

// Update the last synced state
SYNC_INFO
.write()
.update_last_synced_state(owner_vk, &room_data.room_state);
// NOTE: We do not update last_synced_state in the delta path.
// We only have a delta (not the full contract state), so we can't
// set the baseline to the contract's actual state. The full-state path
// (update_room_state_inner) handles baseline updates correctly.
// This may cause one redundant UPDATE on the next sync cycle, but
// it's harmless since the contract will see it as a no-op merge.

// Notify about new messages from other users
// Store notification data for AFTER with_mut completes
// (notify_new_messages calls ROOMS.read() internally, causing deadlock if called here)
if let Some(messages) = new_messages {
// Record receive timestamps for propagation delay tracking
let msg_ids: Vec<_> = messages.iter().map(|m| m.id()).collect();
record_receive_times(&msg_ids);

// Use updated member_info (after delta applied) so new sender nicknames are included
let updated_member_info = room_data.room_state.member_info.clone();
notify_new_messages(
owner_vk,
&messages,
self_member_id,
&updated_member_info,
&room_secrets,
);

// If user is viewing this room with tab visible, mark as read immediately
let is_visible = *DOCUMENT_VISIBLE.read();
let is_current_room = CURRENT_ROOM.read().owner_key == Some(*owner_vk);
if is_visible && is_current_room {
mark_current_room_as_read();
}
pending_notification = Some((messages, self_member_id, updated_member_info, room_secrets));
}

// Update document title (may show unread count)
update_document_title();

// Persist to delegate so state survives refresh
wasm_bindgen_futures::spawn_local(async {
if let Err(e) = save_rooms_to_delegate().await {
Expand All @@ -175,10 +193,29 @@ impl RoomSynchronizer {
}
} else {
warn!("Room not found in rooms map for apply_delta, ignoring delta");
// For now, we'll just ignore deltas for rooms we don't have
// The room should be created through a GET response, not a delta
}
});

// Update document title after ROOMS.with_mut completes (update_document_title calls ROOMS.read())
update_document_title();

// Now safe to call notify_new_messages (it calls ROOMS.read() internally)
if let Some((messages, self_member_id, member_info, room_secrets)) = pending_notification {
notify_new_messages(
&owner_vk,
&messages,
self_member_id,
&member_info,
&room_secrets,
);

// If user is viewing this room with tab visible, mark as read
let is_visible = *DOCUMENT_VISIBLE.read();
let is_current_room = CURRENT_ROOM.read().owner_key == Some(owner_vk);
if is_visible && is_current_room {
mark_current_room_as_read();
}
}
}
}

Expand Down Expand Up @@ -618,12 +655,44 @@ impl RoomSynchronizer {
Ok(())
}

/// Updates the room state and last_sync_state, should be called after state update received from network
/// Updates the room state and last_sync_state, should be called after state update received from network.
///
/// IMPORTANT: On WASM targets, the actual state mutation is deferred via setTimeout(0).
/// This prevents re-entrant signal borrow panics: Dioxus fires subscriber notifications
/// synchronously during Drop of the write guard, which causes `try_read()` in `use_memo`
/// closures to fail. When `try_read()` fails, the memo doesn't subscribe to ROOMS and
/// permanently stops re-evaluating — causing "messages not visible until you post" bugs.
/// setTimeout(0) breaks out of the WASM call stack, ensuring the write happens in a
/// clean execution context where no signal borrows are active.
pub(crate) fn update_room_state(&self, room_owner_vk: &VerifyingKey, state: &ChatRoomStateV1) {
let room_owner_vk = *room_owner_vk;
let state = state.clone();

#[cfg(target_arch = "wasm32")]
{
use wasm_bindgen::prelude::*;
let cb = Closure::once_into_js(move || {
Self::update_room_state_inner(room_owner_vk, state);
});
web_sys::window()
.expect("no window")
.set_timeout_with_callback(&cb.into())
.ok();
}

#[cfg(not(target_arch = "wasm32"))]
Self::update_room_state_inner(room_owner_vk, state);
}

/// Inner implementation of update_room_state, runs in a clean execution context on WASM.
fn update_room_state_inner(room_owner_vk: VerifyingKey, state: ChatRoomStateV1) {
// Capture data needed for notifications BEFORE the mutable borrow
let (old_message_ids, self_member_id, member_info_clone, room_secrets) = {
let rooms = ROOMS.read();
if let Some(room_data) = rooms.map.get(room_owner_vk) {
let Ok(rooms) = ROOMS.try_read() else {
warn!("update_room_state: ROOMS is currently borrowed, skipping update");
return;
};
if let Some(room_data) = rooms.map.get(&room_owner_vk) {
let old_ids: std::collections::HashSet<_> = room_data
.room_state
.recent_messages
Expand All @@ -634,7 +703,7 @@ impl RoomSynchronizer {
info!(
"update_room_state: Captured {} old message IDs for room {:?}",
old_ids.len(),
MemberId::from(*room_owner_vk)
MemberId::from(room_owner_vk)
);
let self_id = MemberId::from(&room_data.self_sk.verifying_key());
let member_info = room_data.room_state.member_info.clone();
Expand All @@ -643,7 +712,7 @@ impl RoomSynchronizer {
} else {
info!(
"update_room_state: Room {:?} not found in ROOMS when capturing old IDs",
MemberId::from(*room_owner_vk)
MemberId::from(room_owner_vk)
);
(None, None, None, HashMap::new())
}
Expand All @@ -653,17 +722,17 @@ impl RoomSynchronizer {
info!(
"update_room_state: Incoming state has {} messages for room {:?}",
state.recent_messages.messages.len(),
MemberId::from(*room_owner_vk)
MemberId::from(room_owner_vk)
);

// Will be populated inside with_mut if new messages are detected
let mut pending_notification: Option<(Vec<_>, MemberId)> = None;
// Updated member_info captured after state merge (so new sender nicknames are included)
let mut updated_member_info: Option<MemberInfoV1> = None;
let room_owner_copy = *room_owner_vk;
let room_owner_copy = room_owner_vk;

ROOMS.with_mut(|rooms| {
if let Some(room_data) = rooms.map.get_mut(room_owner_vk) {
if let Some(room_data) = rooms.map.get_mut(&room_owner_vk) {
// Log member info versions before merge
info!(
"Before merge - Local member info versions ({} items):",
Expand Down Expand Up @@ -695,9 +764,9 @@ impl RoomSynchronizer {
match room_data.room_state.merge(
&room_data.room_state.clone(),
&ChatRoomParametersV1 {
owner: *room_owner_vk,
owner: room_owner_vk,
},
state,
&state,
) {
Ok(_) => {
// For private rooms, rebuild actions_state with decrypted content
Expand Down Expand Up @@ -748,22 +817,26 @@ impl RoomSynchronizer {
}

// Keep cached self membership data up to date
let params = ChatRoomParametersV1 { owner: *room_owner_vk };
let params = ChatRoomParametersV1 { owner: room_owner_vk };
room_data.capture_self_membership_data(&params);

// Make sure the room is registered in SYNC_INFO
// Make sure the room is registered in SYNC_INFO and update the
// baseline to the INCOMING contract state (not the post-merge state).
// The incoming state represents what the contract currently has.
// If we used the post-merge state (which includes any pending local
// changes), needs_to_send_update() would see states_match==true and
// skip sending the user's pending changes. By using the incoming state,
// local changes remain as a detectable diff above the baseline.
SYNC_INFO.with_mut(|sync_info| {
sync_info.register_new_room(*room_owner_vk);
// We use the post-merged state to avoid some edge cases
sync_info
.update_last_synced_state(room_owner_vk, &room_data.room_state);
sync_info.register_new_room(room_owner_vk);
sync_info.update_last_synced_state(&room_owner_vk, &state);
});

// Check if initial sync was already complete before this update
let was_sync_complete = INITIAL_SYNC_COMPLETE.read().contains(room_owner_vk);
let was_sync_complete = INITIAL_SYNC_COMPLETE.read().contains(&room_owner_vk);

// Mark initial sync complete for this room (enables notifications)
mark_initial_sync_complete(room_owner_vk);
mark_initial_sync_complete(&room_owner_vk);

// Detect new messages - store for notification AFTER with_mut completes
// (notify_new_messages calls ROOMS.read() internally, causing deadlock if called here)
Expand All @@ -783,7 +856,7 @@ impl RoomSynchronizer {
info!(
"Detected {} new messages in state update for room {:?}",
new_messages.len(),
MemberId::from(*room_owner_vk)
MemberId::from(room_owner_vk)
);

// Only record receive times after initial sync — during
Expand All @@ -800,7 +873,7 @@ impl RoomSynchronizer {
} else {
info!(
"No new messages detected for room {:?} (old_ids: {}, post-merge: {})",
MemberId::from(*room_owner_vk),
MemberId::from(room_owner_vk),
old_ids.len(),
room_data.room_state.recent_messages.messages.len()
);
Expand All @@ -824,14 +897,12 @@ impl RoomSynchronizer {
// Instead, we should request the full state with a GET reques
// This is handled by registering the room in SYNC_INFO which will trigger a GET request in the next sync cycle

// Register the room in SYNC_INFO to trigger a GET reques
// Register the room in SYNC_INFO to trigger a GET request
SYNC_INFO.with_mut(|sync_info| {
sync_info.register_new_room(*room_owner_vk);
// Store the state temporarily so it can be merged when we get the full room data
sync_info.update_last_synced_state(room_owner_vk, state);
sync_info.register_new_room(room_owner_vk);
});

info!("Registered room {:?} for GET request after receiving update without existing room data", MemberId::from(*room_owner_vk));
info!("Registered room {:?} for GET request after receiving update without existing room data", MemberId::from(room_owner_vk));
}
});

Expand Down
Loading