diff --git a/Justfile b/Justfile index c0abc7a5bf..f1ae4b83f8 100644 --- a/Justfile +++ b/Justfile @@ -543,7 +543,7 @@ test-unit: # never connects. # `^config::` is anchored so it does not also select nip_fi_config. cargo nextest run -p buzz-relay --lib --bin buzz-relay \ - -E 'test(/^api::admin::/) + test(/^push_runtime::/) + test(/^operator_listener::tests::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^handlers::system_channels::tests::/) + test(/^nip11::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_config::tests::/) + test(/^nip_fi_resolver::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::profiles::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(/^startup_steps::tests::/) + test(/^mesh_boot::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::neither_a_confirmed_inactive_community_nor_a_failed_lookup_admits_the_socket) + test(=state::tests::periodic_revalidation_disconnects_inside_the_fenced_callback) + test(=state::tests::bare_community_disconnect_fails_closed_when_the_fence_is_unavailable) + test(=connection::tests::send_loop_sends_policy_close_when_community_is_archived) + test(=state::tests::community_disconnect_then_nip_fi_keeps_community_deleted_reason) + test(=state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame) + test(=handlers::req::tests::timed_out_historical_read_deregisters_before_closed) + test(=handlers::req::tests::superseded_timeout_leaves_replacement_intact) + test(=handlers::req::tests::search_claim_retires_live_and_yields_to_replacement) + test(=handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner) + test(=handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim) + test(=handlers::req::tests::revoke_then_replacement_keeps_replacement_whole) + test(=handlers::req::tests::claims_after_connection_cleanup_are_refused) + test(=handlers::req::tests::dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::count_fallback_covers_filters_that_can_match_private_deletions) + test(=handlers::req::tests::group_state_t_and_creator_tags_reach_sql_only_for_group_state_kinds) + test(=api::bridge::postgres_tests::bridge_search_mode_extension_accepts_prefix_snake_or_camel_case) + test(=api::bridge::postgres_tests::bridge_search_sort_extension_selects_newest_only_when_named) + test(=api::invites::postgres_tests::restriction_tests::postgres_tests::destructive_fixture_rejects_shared_database_urls) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))' + -E 'test(/^api::admin::/) + test(/^push_runtime::/) + test(/^operator_listener::tests::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^handlers::coalesced_refresh::tests::/) + test(/^handlers::system_channels::tests::/) + test(/^nip11::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_config::tests::/) + test(/^nip_fi_resolver::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::profiles::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(/^startup_steps::tests::/) + test(/^mesh_boot::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::neither_a_confirmed_inactive_community_nor_a_failed_lookup_admits_the_socket) + test(=state::tests::periodic_revalidation_disconnects_inside_the_fenced_callback) + test(=state::tests::bare_community_disconnect_fails_closed_when_the_fence_is_unavailable) + test(=connection::tests::send_loop_sends_policy_close_when_community_is_archived) + test(=state::tests::community_disconnect_then_nip_fi_keeps_community_deleted_reason) + test(=state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame) + test(=handlers::req::tests::timed_out_historical_read_deregisters_before_closed) + test(=handlers::req::tests::superseded_timeout_leaves_replacement_intact) + test(=handlers::req::tests::search_claim_retires_live_and_yields_to_replacement) + test(=handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner) + test(=handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim) + test(=handlers::req::tests::revoke_then_replacement_keeps_replacement_whole) + test(=handlers::req::tests::claims_after_connection_cleanup_are_refused) + test(=handlers::req::tests::dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::count_fallback_covers_filters_that_can_match_private_deletions) + test(=handlers::req::tests::group_state_t_and_creator_tags_reach_sql_only_for_group_state_kinds) + test(=api::bridge::postgres_tests::bridge_search_mode_extension_accepts_prefix_snake_or_camel_case) + test(=api::bridge::postgres_tests::bridge_search_sort_extension_selects_newest_only_when_named) + test(=api::invites::postgres_tests::restriction_tests::postgres_tests::destructive_fixture_rejects_shared_database_urls) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))' # Note on audio::join::tests scope: the full suite is infra-free (no DB, # no Redis). The infra-free audio/FI regression witnesses — bootstrap # ordering barrier, CommitConfirmed arm, pending-close invisibility, diff --git a/crates/buzz-db/src/store/thread.rs b/crates/buzz-db/src/store/thread.rs index 0e4c9d9375..7c617aba6a 100644 --- a/crates/buzz-db/src/store/thread.rs +++ b/crates/buzz-db/src/store/thread.rs @@ -18,6 +18,60 @@ use crate::{ ReadSessionInner, RouteDecision, RoutePredicate, }; +/// [`THREAD_PARTICIPANT_SCAN_LIMIT`] as a literal, so SQL can `concat!` it. +macro_rules! thread_participant_scan_limit { + () => { + 1000 + }; +} + +/// How many of a thread's newest replies a participant summary reads. +/// +/// Participants are the thread's most recently active distinct authors, so +/// the newest replies decide them. Aggregating the whole thread instead made +/// every summary cost O(thread), recomputed after every reply, so one very +/// large thread could saturate the writer. An author silent for this many +/// replies drops out of the facepile, which shows at most 10. +pub const THREAD_PARTICIPANT_SCAN_LIMIT: i64 = thread_participant_scan_limit!(); + +/// Correlated `ARRAY(...)` of up to 10 participant pubkeys, most recent +/// first, for the thread rooted at `root.event_id` in community `$1`. +/// +/// Walks `idx_thread_metadata_window` newest-first and stops after +/// [`THREAD_PARTICIPANT_SCAN_LIMIT`] rows, so cost is bounded by the window, +/// not the thread. The one argument is an extra (possibly empty) predicate on +/// the window's `tm` rows. A macro so callers stay `&'static str` via +/// `concat!`. +macro_rules! recent_thread_participants_sql { + ($extra:literal) => { + concat!( + r#"ARRAY( + SELECT e.pubkey + FROM ( + SELECT tm.event_created_at, tm.event_id + FROM thread_metadata tm + WHERE tm.community_id = $1 + AND tm.root_event_id = root.event_id "#, + $extra, + r#" + ORDER BY tm.event_created_at DESC, tm.event_id ASC + LIMIT "#, + thread_participant_scan_limit!(), + r#" + ) recent + JOIN events e + ON e.community_id = $1 + AND e.created_at = recent.event_created_at + AND e.id = recent.event_id + WHERE e.deleted_at IS NULL + GROUP BY e.pubkey + ORDER BY MAX(e.created_at) DESC, e.pubkey ASC + LIMIT 10 + )"# + ) + }; +} + // -- Structs ------------------------------------------------------------------ /// A single reply within a thread, joined with event content. @@ -56,7 +110,9 @@ pub struct ThreadSummary { pub descendant_count: i32, /// Timestamp of the most recent reply in the thread. pub last_reply_at: Option>, - /// Compressed public keys of all participants who have replied. + /// Up to 10 distinct reply authors, most recently active first, drawn + /// from the newest [`THREAD_PARTICIPANT_SCAN_LIMIT`] replies. Deleted + /// replies count toward that window but contribute no author. pub participants: Vec>, } @@ -467,63 +523,29 @@ pub async fn get_thread_summary( crate::observability::WriterOperation::EventWrite, ) .await?; - let row = sqlx::query( - r#" - SELECT reply_count, descendant_count, last_reply_at - FROM thread_metadata - WHERE community_id = $1 AND event_id = $2 + let row = sqlx::query(concat!( + "SELECT root.reply_count, root.descendant_count, root.last_reply_at, ", + recent_thread_participants_sql!(""), + r#" AS participants + FROM thread_metadata root + WHERE root.community_id = $1 AND root.event_id = $2 LIMIT 1 "#, - ) + )) .bind(community_id.as_uuid()) .bind(event_id) .fetch_optional(&mut *connection) .await?; - let row = match row { - Some(r) => r, - None => return Ok(None), - }; - - let reply_count: i32 = row.try_get("reply_count")?; - let descendant_count: i32 = row.try_get("descendant_count")?; - let last_reply_at: Option> = row.try_get("last_reply_at")?; - - // Collect distinct participant pubkeys from the thread, most recent first. - let participant_rows = sqlx::query( - r#" - SELECT pubkey FROM ( - SELECT DISTINCT e.pubkey, MAX(e.created_at) AS last_seen - FROM thread_metadata tm - JOIN events e - ON e.community_id = tm.community_id - AND e.created_at = tm.event_created_at - AND e.id = tm.event_id - WHERE tm.community_id = $1 - AND tm.root_event_id = $2 - AND e.deleted_at IS NULL - GROUP BY e.pubkey - ) sub - ORDER BY last_seen DESC - LIMIT 10 - "#, - ) - .bind(community_id.as_uuid()) - .bind(event_id) - .fetch_all(&mut *connection) - .await?; - - let participants: Vec> = participant_rows - .into_iter() - .map(|r| r.try_get::, _>("pubkey")) - .collect::>()?; - - Ok(Some(ThreadSummary { - reply_count, - descendant_count, - last_reply_at, - participants, - })) + row.map(|row| { + Ok(ThreadSummary { + reply_count: row.try_get("reply_count")?, + descendant_count: row.try_get("descendant_count")?, + last_reply_at: row.try_get("last_reply_at")?, + participants: row.try_get("participants")?, + }) + }) + .transpose() } /// Fetch one channel window: top-level rows (depth = 0, missing metadata, or @@ -692,39 +714,21 @@ pub(crate) async fn get_channel_window_on( } // Batch participants for every row with thread activity — one query for - // the whole window instead of a per-root fan-out. Same shape and 10-cap - // as get_thread_summary. + // the whole window instead of a per-root fan-out, each root bounded the + // same way as get_thread_summary. let roots: Vec> = rows .iter() .filter(|r| r.thread_summary.is_some()) .map(|r| r.stored_event.event.id.as_bytes().to_vec()) .collect(); if !roots.is_empty() { - let participant_rows = sqlx::query( - r#" - SELECT root_event_id, pubkey FROM ( - SELECT - tm.root_event_id, - e.pubkey, - MAX(e.created_at) AS last_seen, - ROW_NUMBER() OVER ( - PARTITION BY tm.root_event_id - ORDER BY MAX(e.created_at) DESC - ) AS rn - FROM thread_metadata tm - JOIN events e - ON e.community_id = tm.community_id - AND e.created_at = tm.event_created_at - AND e.id = tm.event_id - WHERE tm.community_id = $1 - AND tm.root_event_id = ANY($2) - AND e.deleted_at IS NULL - GROUP BY tm.root_event_id, e.pubkey - ) sub - WHERE rn <= 10 - ORDER BY root_event_id, rn + let participant_rows = sqlx::query(concat!( + "SELECT root.event_id, ", + recent_thread_participants_sql!(""), + r#" AS participants + FROM unnest($2::bytea[]) AS root(event_id) "#, - ) + )) .bind(community_id.as_uuid()) .bind(&roots) .fetch_all(&mut *conn) @@ -733,9 +737,7 @@ pub(crate) async fn get_channel_window_on( let mut by_root: std::collections::HashMap, Vec>> = std::collections::HashMap::new(); for row in participant_rows { - let root: Vec = row.try_get("root_event_id")?; - let pubkey: Vec = row.try_get("pubkey")?; - by_root.entry(root).or_default().push(pubkey); + by_root.insert(row.try_get("event_id")?, row.try_get("participants")?); } for row in &mut rows { if let Some(summary) = &mut row.thread_summary { @@ -982,18 +984,10 @@ impl Db { crate::observability::WriterOperation::SubscriptionHistory, ) .await?; - let rows = sqlx::query( - r#" - SELECT root.event_id, root.reply_count, root.descendant_count, root.last_reply_at, - ARRAY( - SELECT e.pubkey - FROM thread_metadata tm - JOIN events e ON e.community_id = tm.community_id - AND e.created_at = tm.event_created_at AND e.id = tm.event_id - WHERE tm.community_id = root.community_id AND tm.root_event_id = root.event_id - AND tm.channel_id = root.channel_id AND e.deleted_at IS NULL - GROUP BY e.pubkey ORDER BY MAX(e.created_at) DESC LIMIT 10 - ) AS participants + let rows = sqlx::query(concat!( + "SELECT root.event_id, root.reply_count, root.descendant_count, root.last_reply_at, ", + recent_thread_participants_sql!("AND tm.channel_id = root.channel_id"), + r#" AS participants FROM thread_metadata root WHERE root.community_id = $1 AND root.channel_id = $2 AND root.event_id IN ( @@ -1003,7 +997,7 @@ impl Db { AND target.event_id = ANY($3) AND target.parent_event_id IS NOT NULL ) "#, - ) + )) .bind(community_id.as_uuid()) .bind(channel_id) .bind(target_ids) @@ -2320,4 +2314,101 @@ mod postgres_tests { "reply-less row carries no summary" ); } + + /// Every participant reader (live summary, channel window, deleted-reply + /// root resolution) aggregates only the newest + /// [`THREAD_PARTICIPANT_SCAN_LIMIT`] replies, most recent author first, + /// skipping deleted replies. An author who last posted before that window + /// drops out; a whole-thread aggregate would still return them. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn participant_summaries_read_only_the_newest_replies() { + let pool = setup_pool().await; + let db = Db::from_pool(pool.clone()); + let early = Keys::generate(); + let steady = Keys::generate(); + let deleted_author = Keys::generate(); + let latest = Keys::generate(); + let (channel, community) = create_test_channel( + &pool, + &format!("participant-window-{}", Uuid::new_v4()), + ChannelType::Stream, + ChannelVisibility::Open, + None, + steady.public_key().to_bytes().as_slice(), + None, + ) + .await + .expect("create channel"); + + let root = make_stream_event(&steady, "root"); + insert_root(&pool, community, channel.id, &root).await; + let base = root.created_at.as_secs(); + let reply_at = |keys: &Keys, offset: u64| { + EventBuilder::new(Kind::Custom(9), format!("reply-{offset}")) + .custom_created_at(nostr::Timestamp::from(base + offset)) + .sign_with_keys(keys) + .expect("sign reply") + }; + + // `early` posts first, then `steady` fills the whole window after it. + let early_reply = reply_at(&early, 1); + insert_reply(&pool, community, channel.id, &root, &early_reply, false).await; + let window = THREAD_PARTICIPANT_SCAN_LIMIT as u64; + for offset in 2..window { + let reply = reply_at(&steady, offset); + insert_reply(&pool, community, channel.id, &root, &reply, false).await; + } + let deleted_reply = reply_at(&deleted_author, window); + insert_reply(&pool, community, channel.id, &root, &deleted_reply, false).await; + let latest_reply = reply_at(&latest, window + 1); + insert_reply(&pool, community, channel.id, &root, &latest_reply, false).await; + db.soft_delete_event_and_update_thread( + community, + deleted_reply.id.as_bytes(), + Some(root.id.as_bytes()), + Some(root.id.as_bytes()), + ) + .await + .expect("delete reply"); + + let expected = vec![ + latest.public_key().to_bytes().to_vec(), + steady.public_key().to_bytes().to_vec(), + ]; + + let summary = get_thread_summary(&pool, community, root.id.as_bytes()) + .await + .expect("thread summary") + .expect("root has thread metadata"); + assert_eq!(summary.participants, expected, "live summary"); + + let window = get_channel_window(&pool, community, channel.id, 50, None, None) + .await + .expect("fetch window"); + let row = window + .rows + .iter() + .find(|r| r.stored_event.event.id == root.id) + .expect("root row present"); + assert_eq!( + row.thread_summary + .as_ref() + .expect("root has a summary") + .participants, + expected, + "channel window" + ); + + let resolved = db + .resolve_thread_root_summaries( + community, + channel.id, + &[deleted_reply.id.as_bytes().to_vec()], + ) + .await + .expect("resolve root"); + assert_eq!(resolved.len(), 1); + assert_eq!(resolved[0].1.participants, expected, "resolved root"); + } } diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index cc11238d4f..fcf4e6ddb1 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -409,7 +409,6 @@ fn p0_pool_acquisitions_use_typed_operation_pairs_without_other() { .0; assert!(thread_summary.contains("WriterOperation::EventWrite")); assert!(thread_summary.contains("fetch_optional(&mut *connection)")); - assert!(thread_summary.contains("fetch_all(&mut *connection)")); assert!(!thread_summary.contains("fetch_optional(pool)")); assert!(!thread_summary.contains("fetch_all(pool)")); diff --git a/crates/buzz-relay/src/handlers/coalesced_refresh.rs b/crates/buzz-relay/src/handlers/coalesced_refresh.rs new file mode 100644 index 0000000000..0110af2fa0 --- /dev/null +++ b/crates/buzz-relay/src/handlers/coalesced_refresh.rs @@ -0,0 +1,287 @@ +//! Per-key single-flight refresh with a minimum interval. +//! +//! A refresh for a key runs at once when none is in flight. Requests that +//! arrive while one is running or cooling down collapse into one trailing +//! rerun, so a key refreshes at most once per interval per process and the +//! last request is always covered by a refresh that started after it. Each +//! run is bounded by a timeout, so a hung refresh can't wedge its key. + +use std::future::Future; +use std::hash::Hash; +use std::sync::Arc; +use std::time::Duration; + +use dashmap::mapref::entry::Entry; +use dashmap::DashMap; + +/// Coalesces refresh requests per key. The map holds only keys with a +/// refresh in flight; the value is "another request arrived since it began". +pub struct CoalescedRefresh { + pending: DashMap, + min_interval: Duration, + run_timeout: Duration, +} + +impl CoalescedRefresh +where + K: Eq + Hash + Clone + Send + Sync + 'static, +{ + /// A coalescer that refreshes each key at most once per `min_interval`, + /// abandoning any single run that takes longer than `run_timeout`. + pub fn new(min_interval: Duration, run_timeout: Duration) -> Self { + Self { + pending: DashMap::new(), + min_interval, + run_timeout, + } + } + + /// Request a refresh of `key`. Runs `refresh` on a spawned task now when + /// `key` is idle; otherwise marks the in-flight refresh to run once more + /// after its interval elapses. + pub fn request(self: &Arc, key: K, refresh: F) + where + F: Fn() -> Fut + Send + 'static, + Fut: Future + Send, + { + match self.pending.entry(key.clone()) { + Entry::Occupied(mut rerun) => { + *rerun.get_mut() = true; + return; + } + Entry::Vacant(slot) => { + slot.insert(false); + } + } + // Frees the key if the task panics or is cancelled, including before + // its first poll, so a failed refresh can't block the key forever. + let mut release = Release { + owner: Arc::clone(self), + key: Some(key.clone()), + }; + tokio::spawn(async move { + let this = Arc::clone(&release.owner); + loop { + if tokio::time::timeout(this.run_timeout, refresh()) + .await + .is_err() + { + tracing::warn!( + timeout_ms = this.run_timeout.as_millis() as u64, + "coalesced refresh timed out" + ); + } + tokio::time::sleep(this.min_interval).await; + if !this.rerun_or_release(&key) { + break; + } + } + // Finished normally: rerun_or_release already freed the key. + release.key.take(); + }); + } + + /// Clear the rerun flag and report it, or free the key when no request + /// arrived. Atomic with [`Self::request`]'s entry check, so a request + /// either sets the flag before this reads it or finds the key free. + fn rerun_or_release(&self, key: &K) -> bool { + match self.pending.entry(key.clone()) { + Entry::Occupied(mut entry) if *entry.get() => { + *entry.get_mut() = false; + true + } + Entry::Occupied(entry) => { + entry.remove(); + false + } + Entry::Vacant(_) => false, + } + } +} + +struct Release { + owner: Arc>, + key: Option, +} + +impl Drop for Release { + fn drop(&mut self) { + if let Some(key) = self.key.take() { + self.owner.pending.remove(&key); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + + const INTERVAL: Duration = Duration::from_secs(1); + const TIMEOUT: Duration = Duration::from_secs(5); + + fn counting(runs: &Arc) -> impl Fn() -> std::future::Ready<()> + Send + 'static { + let runs = Arc::clone(runs); + move || { + runs.fetch_add(1, Ordering::SeqCst); + std::future::ready(()) + } + } + + /// Let spawned tasks run up to their next timer. + async fn settle() { + for _ in 0..10 { + tokio::task::yield_now().await; + } + } + + #[tokio::test(start_paused = true)] + async fn burst_runs_once_now_and_once_trailing_then_frees_the_key() { + let refresh = Arc::new(CoalescedRefresh::new(INTERVAL, TIMEOUT)); + let runs = Arc::new(AtomicUsize::new(0)); + + for _ in 0..100 { + refresh.request("root", counting(&runs)); + } + settle().await; + assert_eq!(runs.load(Ordering::SeqCst), 1, "first request runs at once"); + + tokio::time::advance(INTERVAL).await; + settle().await; + assert_eq!( + runs.load(Ordering::SeqCst), + 2, + "the burst gets one trailing run" + ); + + tokio::time::advance(INTERVAL).await; + settle().await; + assert_eq!(runs.load(Ordering::SeqCst), 2, "no request, no rerun"); + assert!(refresh.pending.is_empty(), "idle key is released"); + + refresh.request("root", counting(&runs)); + settle().await; + assert_eq!( + runs.load(Ordering::SeqCst), + 3, + "a later request runs at once" + ); + } + + #[tokio::test(start_paused = true)] + async fn request_during_a_run_gets_a_trailing_run_that_sees_its_state() { + let refresh = Arc::new(CoalescedRefresh::new(INTERVAL, TIMEOUT)); + let state = Arc::new(AtomicUsize::new(1)); + let seen = Arc::new(std::sync::Mutex::new(Vec::new())); + let gate = Arc::new(tokio::sync::Semaphore::new(0)); + // Records the state it reads, then holds until the gate opens. + let reader = { + let (state, seen, gate) = (Arc::clone(&state), Arc::clone(&seen), Arc::clone(&gate)); + move || { + let (state, seen, gate) = + (Arc::clone(&state), Arc::clone(&seen), Arc::clone(&gate)); + async move { + seen.lock().unwrap().push(state.load(Ordering::SeqCst)); + let _permit = gate.acquire().await; + } + } + }; + + refresh.request("root", reader.clone()); + settle().await; + assert_eq!(*seen.lock().unwrap(), [1], "the first run read state 1"); + + state.store(2, Ordering::SeqCst); + refresh.request("root", reader.clone()); + settle().await; + assert_eq!( + *seen.lock().unwrap(), + [1], + "no second run while one is in flight" + ); + + gate.add_permits(1); + settle().await; + tokio::time::advance(INTERVAL).await; + settle().await; + assert_eq!( + *seen.lock().unwrap(), + [1, 2], + "the trailing run reads the state written mid-run" + ); + + tokio::time::advance(INTERVAL).await; + settle().await; + assert_eq!(*seen.lock().unwrap(), [1, 2], "no further run"); + assert!(refresh.pending.is_empty(), "idle key is released"); + } + + #[test] + fn a_task_dropped_before_its_first_poll_releases_the_key() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_time() + .build() + .unwrap(); + let refresh = Arc::new(CoalescedRefresh::new(INTERVAL, TIMEOUT)); + let runs = Arc::new(AtomicUsize::new(0)); + // `block_on` returns without yielding, so the spawned task is never + // polled; dropping the runtime drops it unpolled. + runtime.block_on(async { refresh.request("root", counting(&runs)) }); + assert!(!refresh.pending.is_empty(), "the queued task holds the key"); + + drop(runtime); + assert_eq!(runs.load(Ordering::SeqCst), 0, "the task never ran"); + assert!( + refresh.pending.is_empty(), + "an unpolled task must not wedge the key" + ); + } + + #[tokio::test(start_paused = true)] + async fn keys_refresh_independently() { + let refresh = Arc::new(CoalescedRefresh::new(INTERVAL, TIMEOUT)); + let runs = Arc::new(AtomicUsize::new(0)); + + refresh.request("a", counting(&runs)); + refresh.request("b", counting(&runs)); + settle().await; + assert_eq!(runs.load(Ordering::SeqCst), 2); + } + + #[tokio::test(start_paused = true)] + async fn panicking_refresh_releases_the_key() { + let refresh = Arc::new(CoalescedRefresh::new(INTERVAL, TIMEOUT)); + refresh.request("root", || async { panic!("refresh failed") }); + settle().await; + assert!(refresh.pending.is_empty(), "panic must not wedge the key"); + + let runs = Arc::new(AtomicUsize::new(0)); + refresh.request("root", counting(&runs)); + settle().await; + assert_eq!(runs.load(Ordering::SeqCst), 1); + } + + #[tokio::test(start_paused = true)] + async fn hung_refresh_times_out_and_releases_the_key() { + let refresh = Arc::new(CoalescedRefresh::new(INTERVAL, TIMEOUT)); + refresh.request("root", std::future::pending::<()>); + settle().await; + assert!(!refresh.pending.is_empty(), "the hung run holds the key"); + + // Step one timer at a time: the cooldown sleep is armed only after + // the timeout fires, so a single jump would overshoot it. + tokio::time::advance(TIMEOUT).await; + settle().await; + tokio::time::advance(INTERVAL).await; + settle().await; + assert!( + refresh.pending.is_empty(), + "a hung run must not wedge the key" + ); + + let runs = Arc::new(AtomicUsize::new(0)); + refresh.request("root", counting(&runs)); + settle().await; + assert_eq!(runs.load(Ordering::SeqCst), 1); + } +} diff --git a/crates/buzz-relay/src/handlers/mod.rs b/crates/buzz-relay/src/handlers/mod.rs index 3315ea8eb6..cd61066890 100644 --- a/crates/buzz-relay/src/handlers/mod.rs +++ b/crates/buzz-relay/src/handlers/mod.rs @@ -6,6 +6,8 @@ pub mod auth; pub mod channel_authz; /// Subscription close (CLOSE) handler. pub mod close; +/// Per-key single-flight refresh with a minimum interval. +pub mod coalesced_refresh; /// Command executor — transactional processing for command kinds. pub mod command_executor; /// Relay-operator community provisioning HTTP support. diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 101ac4bc24..add1ddcd9b 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -981,6 +981,11 @@ async fn fan_out_system_message( /// incremented, so the emitted summary is exact even under concurrent /// replies/deletes (newest `created_at` wins client-side). /// +/// Coalesced per root through [`AppState::thread_summary_refresh`]: a quiet +/// thread emits at once, while a busy one emits at most once per +/// [`crate::state::THREAD_SUMMARY_REFRESH_INTERVAL`] per pod, with a trailing +/// emit that reads state committed after the last mutation in the burst. +/// /// Spawned: runs after the triggering write committed and must not add /// latency to the ingest acknowledgement, mirroring /// `dispatch_persistent_event`. @@ -990,88 +995,99 @@ pub fn emit_live_thread_summary( channel_id: Uuid, root_id: Vec, ) { + let key = (tenant.community(), root_id.clone()); let tenant = tenant.clone(); - let state = Arc::clone(state); - tokio::spawn(async move { - let summary = match state - .db - .get_thread_summary(tenant.community(), &root_id) - .await - { - Ok(Some(summary)) => summary, - // Root has no thread row — nothing to summarize (e.g., the root - // itself was just deleted). - Ok(None) => return, - Err(e) => { - warn!( - root = %hex::encode(&root_id), - "live thread summary lookup failed: {e}" - ); - return; - } - }; - - let root_hex = hex::encode(&root_id); - // Same tags/content shape as the channel-window page overlay in - // `api/bridge.rs` — one contract, two delivery doors. - let content = serde_json::json!({ - "reply_count": summary.reply_count, - "descendant_count": summary.descendant_count, - "last_reply_at": summary.last_reply_at.map(|t| t.timestamp()), - "participants": summary.participants.iter().map(hex::encode).collect::>(), - }); - let tags = [ - Tag::parse(["e", &root_hex]), - Tag::parse(["d", &root_hex]), - Tag::parse(["h", &channel_id.to_string()]), - ]; - let mut parsed = Vec::with_capacity(tags.len()); - for tag in tags { - match tag { - Ok(tag) => parsed.push(tag), - Err(e) => { - warn!(root = %root_hex, "live thread summary tag failed: {e}"); - return; - } - } - } - let event = match EventBuilder::new( - Kind::Custom(KIND_THREAD_SUMMARY as u16), - content.to_string(), + let task_state = Arc::clone(state); + state.thread_summary_refresh.request(key, move || { + publish_live_thread_summary( + tenant.clone(), + Arc::clone(&task_state), + channel_id, + root_id.clone(), ) - .tags(parsed) - .sign_with_keys(&state.relay_keypair) - { - Ok(event) => event, - Err(e) => { - warn!(root = %root_hex, "live thread summary sign failed: {e}"); - return; - } - }; - - // Redis before local fan-out so subscribers on other relay pods - // receive it too, matching `dispatch_persistent_event`. - state.mark_local_event(tenant.community(), &event.id); - if let Err(e) = state - .pubsub - .publish_event(&tenant, EventTopic::Channel(channel_id), &event) - .await - { - state - .local_event_ids - .invalidate(&(tenant.community(), event.id.to_bytes())); - warn!(root = %root_hex, "live thread summary Redis publish failed: {e}"); - } - let stored = StoredEvent::new(event, Some(channel_id)); - crate::handlers::event::fan_out_event_to_local_subscribers( - &state, - tenant.community(), - &stored, - ) - .await; }); } +async fn publish_live_thread_summary( + tenant: TenantContext, + state: Arc, + channel_id: Uuid, + root_id: Vec, +) { + let summary = match state + .db + .get_thread_summary(tenant.community(), &root_id) + .await + { + Ok(Some(summary)) => summary, + // Root has no thread row — nothing to summarize (e.g., the root + // itself was just deleted). + Ok(None) => return, + Err(e) => { + warn!( + root = %hex::encode(&root_id), + "live thread summary lookup failed: {e}" + ); + return; + } + }; + + let root_hex = hex::encode(&root_id); + // Same tags/content shape as the channel-window page overlay in + // `api/bridge.rs` — one contract, two delivery doors. + let content = serde_json::json!({ + "reply_count": summary.reply_count, + "descendant_count": summary.descendant_count, + "last_reply_at": summary.last_reply_at.map(|t| t.timestamp()), + "participants": summary.participants.iter().map(hex::encode).collect::>(), + }); + let tags = [ + Tag::parse(["e", &root_hex]), + Tag::parse(["d", &root_hex]), + Tag::parse(["h", &channel_id.to_string()]), + ]; + let mut parsed = Vec::with_capacity(tags.len()); + for tag in tags { + match tag { + Ok(tag) => parsed.push(tag), + Err(e) => { + warn!(root = %root_hex, "live thread summary tag failed: {e}"); + return; + } + } + } + let event = match EventBuilder::new( + Kind::Custom(KIND_THREAD_SUMMARY as u16), + content.to_string(), + ) + .tags(parsed) + .sign_with_keys(&state.relay_keypair) + { + Ok(event) => event, + Err(e) => { + warn!(root = %root_hex, "live thread summary sign failed: {e}"); + return; + } + }; + + // Redis before local fan-out so subscribers on other relay pods + // receive it too, matching `dispatch_persistent_event`. + state.mark_local_event(tenant.community(), &event.id); + if let Err(e) = state + .pubsub + .publish_event(&tenant, EventTopic::Channel(channel_id), &event) + .await + { + state + .local_event_ids + .invalidate(&(tenant.community(), event.id.to_bytes())); + warn!(root = %root_hex, "live thread summary Redis publish failed: {e}"); + } + let stored = StoredEvent::new(event, Some(channel_id)); + crate::handlers::event::fan_out_event_to_local_subscribers(&state, tenant.community(), &stored) + .await; +} + /// Emit a relay-signed membership notification event stored globally (channel_id = None). /// /// kind:44100 = member added, kind:44101 = member removed. Every notice diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 6898f40fe4..4fe9e7b1e7 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -1199,6 +1199,14 @@ impl Default for ConnectionManager { } } +/// Minimum spacing between live thread-summary emits for one thread root on +/// one relay pod. See [`AppState::thread_summary_refresh`]. +pub const THREAD_SUMMARY_REFRESH_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1); + +/// Upper bound on one live thread-summary refresh (DB read, Redis publish, +/// local fan-out). A hung refresh is abandoned so its root keeps emitting. +pub const THREAD_SUMMARY_REFRESH_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); + /// Shared application state, cloned cheaply via inner `Arc` fields. #[derive(Clone)] pub struct AppState { @@ -1335,6 +1343,11 @@ pub struct AppState { Arc>>, /// Current in-flight media uploads per (community, uploader pubkey). pub media_uploads_in_flight: Arc>, + /// Per-(community, thread root) coalescing for live `kind:39005` thread + /// summaries, so a busy thread recomputes its summary at most once per + /// [`THREAD_SUMMARY_REFRESH_INTERVAL`] instead of once per reply. + pub thread_summary_refresh: + Arc)>>, /// Cache for observer agent-owner authorization (kind 24200). /// Key: (community_id, agent_pubkey_bytes, owner_pubkey_bytes). Value: is_owner. /// `agent_owner_pubkey` is immutable inside one community, so a long TTL @@ -1609,6 +1622,12 @@ impl AppState { .build(), ), media_uploads_in_flight: Arc::new(DashMap::new()), + thread_summary_refresh: Arc::new( + crate::handlers::coalesced_refresh::CoalescedRefresh::new( + THREAD_SUMMARY_REFRESH_INTERVAL, + THREAD_SUMMARY_REFRESH_TIMEOUT, + ), + ), observer_owner_cache: Arc::new( moka::sync::Cache::builder() .max_capacity(100_000) diff --git a/docs/bridge-channel-window.md b/docs/bridge-channel-window.md index aec9d19542..1f2470866a 100644 --- a/docs/bridge-channel-window.md +++ b/docs/bridge-channel-window.md @@ -99,7 +99,7 @@ Clients **partition by kind before any cursor math**: - tags: `["e", ]`, `["d", ]`, `["h", ]` - content: `{"reply_count":n,"descendant_count":n,"last_reply_at":ts|null,"participants":["",...]}` - (participants: up to 10, most recent first) + (participants: up to 10 from the newest 1000 replies, deleted ones included, most recent first) - Signed by the relay keypair. Synthesized at query time, **never stored**. Clients treat it as replace-by-target metadata keyed by the `e`/`d` tag (the `d` tag gives parameterized-replaceable semantics natively); it is diff --git a/docs/nips/NIP-CW.md b/docs/nips/NIP-CW.md index 9e6ea14e1b..9016df4e52 100644 --- a/docs/nips/NIP-CW.md +++ b/docs/nips/NIP-CW.md @@ -139,7 +139,7 @@ One per returned row with replies. Tag cardinality is exact: one `e`, one `d`, o - `reply_count` — direct replies to the row. `descendant_count` — all events in the row's thread subtree. - `last_reply_at` — unix seconds of the newest descendant, or `null`. -- `participants` — up to 10 distinct author pubkeys from the thread, most recent first. +- `participants` — up to 10 distinct author pubkeys from the thread's newest 1000 replies, deleted replies included, most recent first. Deleted replies contribute no author, and an author silent for 1000 replies drops out. - The `e` and `d` tags both carry the row's event id: `e` for reference-following, `d` for replaceable addressing. ### `kind:39006` — window bounds diff --git a/mobile/lib/features/channels/channels_provider.dart b/mobile/lib/features/channels/channels_provider.dart index d1672ddb0e..c5f7babe51 100644 --- a/mobile/lib/features/channels/channels_provider.dart +++ b/mobile/lib/features/channels/channels_provider.dart @@ -78,11 +78,6 @@ class ChannelsNotifier extends AsyncNotifier> { List _directoryMetas = const []; Set _hiddenDmIds = const {}; - /// Channels opened from their own metadata before this list had them, as a - /// fast DM open does. The list takes precedence once it has them; a DM's - /// participants are fixed, so the opened copy stays valid until then. - Map _openedAhead = const {}; - /// Fences directory responses to the relay and identity that requested them. late final _ChannelRefreshCoordinator _refreshCoordinator = _ChannelRefreshCoordinator.forRef(ref); @@ -104,14 +99,17 @@ class ChannelsNotifier extends AsyncNotifier> { Set get hiddenDmIds => Set.unmodifiable(_hiddenDmIds); /// Keeps [channel], opened before this list has it, findable by - /// [channelById] while the list reloads. - void rememberOpened(Channel channel) => - _openedAhead = {..._openedAhead, channel.id: channel}; + /// [channelById] and [channelByIdProvider] until the list has it. + void rememberOpened(Channel channel) { + // The open's background refresh may have listed it already. + if (state.value?.any((listed) => listed.id == channel.id) ?? false) return; + ref.read(_openedAheadProvider.notifier).remember(channel); + } /// [channelId] from the list, else as it was opened ahead of the list. Channel? channelById(String channelId) => state.value?.where((channel) => channel.id == channelId).firstOrNull ?? - _openedAhead[channelId]; + ref.read(_openedAheadProvider)[channelId]; bool get hasLoaded => _hasLoaded; @@ -176,7 +174,6 @@ class ChannelsNotifier extends AsyncNotifier> { _memberSnapshotsByChannelId = const {}; _directoryMetas = const []; _hiddenDmIds = const {}; - _openedAhead = const {}; // Retire any in-flight directory request: its response describes the // previous relay or identity and must not reach this scope's state. _refreshCoordinator.retireInFlight(); @@ -185,6 +182,11 @@ class ChannelsNotifier extends AsyncNotifier> { (_hasLoaded ? state.value : null) ?? _restoreSnapshot(pubkey, relayBaseUrl); if (snapshot != null) _hasCachedSnapshot = true; + listenSelf((_, next) { + final listed = next.value; + if (listed == null || !ref.exists(_openedAheadProvider)) return; + ref.read(_openedAheadProvider.notifier).dropListed(listed); + }); _saveSnapshot = pubkey == null ? null : _captureSnapshotWriter(pubkey, relayBaseUrl); @@ -1115,3 +1117,46 @@ class ChannelsNotifier extends AsyncNotifier> { final channelsProvider = AsyncNotifierProvider>( ChannelsNotifier.new, ); + +/// The channel with this id from the list, else as it was opened ahead of the +/// list (see [ChannelsNotifier.rememberOpened]). +final channelByIdProvider = Provider.autoDispose.family( + (ref, channelId) => + ref.watch( + channelsProvider.select( + (channels) => channels.value + ?.where((channel) => channel.id == channelId) + .firstOrNull, + ), + ) ?? + ref.watch(_openedAheadProvider.select((opened) => opened[channelId])), +); + +/// Channels opened from their own metadata before the list had them, as a +/// fast DM open does. Scoped to one relay and identity. Each is dropped once +/// the list has it, so a DM later hidden or left is not found through it. +final _openedAheadProvider = + NotifierProvider<_OpenedAheadNotifier, Map>( + _OpenedAheadNotifier.new, + ); + +class _OpenedAheadNotifier extends Notifier> { + @override + Map build() { + ref.watch(relayConfigProvider.select((config) => config.baseUrl)); + ref.watch(myPubkeyProvider); + return const {}; + } + + void remember(Channel channel) => state = {...state, channel.id: channel}; + + void dropListed(List listed) { + if (state.isEmpty) return; + final ids = {for (final channel in listed) channel.id}; + if (!state.keys.any(ids.contains)) return; + state = { + for (final entry in state.entries) + if (!ids.contains(entry.key)) entry.key: entry.value, + }; + } +} diff --git a/mobile/lib/features/channels/compose_bar/compose_bar_widget.dart b/mobile/lib/features/channels/compose_bar/compose_bar_widget.dart index 591bcdd2e1..a950b7ceff 100644 --- a/mobile/lib/features/channels/compose_bar/compose_bar_widget.dart +++ b/mobile/lib/features/channels/compose_bar/compose_bar_widget.dart @@ -246,8 +246,7 @@ class ComposeBar extends HookConsumerWidget { final currentPubkey = ref.watch(currentPubkeyProvider); final userCache = ref.watch(userCacheProvider); final isDmChannel = - channelsAsync.asData?.value.any((c) => c.id == channelId && c.isDm) ?? - false; + ref.watch(channelByIdProvider(channelId))?.isDm ?? false; // Preload profiles for channel members, mentionable agents, and their // owners so @mention suggestions show names ("managed by …" included). diff --git a/mobile/lib/features/channels/mentions/mention_candidates_provider.dart b/mobile/lib/features/channels/mentions/mention_candidates_provider.dart index a5589d08a4..87273980e6 100644 --- a/mobile/lib/features/channels/mentions/mention_candidates_provider.dart +++ b/mobile/lib/features/channels/mentions/mention_candidates_provider.dart @@ -162,9 +162,7 @@ final mentionCandidatesProvider = Provider.family final channels = channelsAsync.asData?.value ?? const []; final userCache = ref.watch(userCacheProvider); final currentPubkey = ref.watch(currentPubkeyProvider); - final channel = channels - .where((candidate) => candidate.id == args.channelId) - .firstOrNull; + final channel = ref.watch(channelByIdProvider(args.channelId)); // Archived channels take no mentions. if (channel?.isArchived == true) return const []; final directory = channel?.mentionsOutsidePeople ?? false; @@ -260,9 +258,7 @@ IdentityNames mentionPickerNames( /// section 2). Archived channels never search. final mentionSearchFailedProvider = Provider.family .autoDispose((ref, args) { - final channel = (ref.watch(channelsProvider).asData?.value ?? const []) - .where((candidate) => candidate.id == args.channelId) - .firstOrNull; + final channel = ref.watch(channelByIdProvider(args.channelId)); if (channel == null || channel.isArchived || !channel.mentionsOutsidePeople) { diff --git a/mobile/lib/features/channels/message_actions.dart b/mobile/lib/features/channels/message_actions.dart index a0af9e27fe..a6e5180bb6 100644 --- a/mobile/lib/features/channels/message_actions.dart +++ b/mobile/lib/features/channels/message_actions.dart @@ -457,8 +457,7 @@ bool messageActionShowsUnread( required TimelineMessage message, }) { final readerPubkey = ref.read(myPubkeyProvider); - final channels = ref.read(channelsProvider).asData?.value; - final channel = channels?.where((c) => c.id == channelId).firstOrNull; + final channel = ref.read(channelsProvider.notifier).channelById(channelId); final channelCatchUp = readerPubkey != null && channel != null && diff --git a/mobile/lib/features/channels/thread_detail_page.dart b/mobile/lib/features/channels/thread_detail_page.dart index d217640c61..8b36284940 100644 --- a/mobile/lib/features/channels/thread_detail_page.dart +++ b/mobile/lib/features/channels/thread_detail_page.dart @@ -745,12 +745,7 @@ class ThreadDetailPage extends HookConsumerWidget { visible.add(replies[replyIndex]); } final isDm = - ref - .read(channelsProvider) - .value - ?.where((candidate) => candidate.id == channelId) - .firstOrNull - ?.isDm ?? + ref.read(channelsProvider.notifier).channelById(channelId)?.isDm ?? false; final marks = readingMarks( readState: readState, @@ -900,16 +895,7 @@ class ThreadDetailPage extends HookConsumerWidget { return null; }, [settledImeBottomInset.value]); - final channel = - ref.watch( - channelsProvider.select( - (channels) => channels.value - ?.where((candidate) => candidate.id == channelId) - .firstOrNull, - ), - ) ?? - // A DM opened ahead of a reloading list. - ref.read(channelsProvider.notifier).channelById(channelId); + final channel = ref.watch(channelByIdProvider(channelId)); final channelNamesMap = ref.watch(channelIdsByNameProvider); return FrostedScaffold( diff --git a/mobile/test/features/channels/channel_detail_page_test.dart b/mobile/test/features/channels/channel_detail_page_test.dart index ebe243714f..55f02451a2 100644 --- a/mobile/test/features/channels/channel_detail_page_test.dart +++ b/mobile/test/features/channels/channel_detail_page_test.dart @@ -10799,12 +10799,12 @@ void main() { final head = formatTimeline([ _textMsg(id: threadIdThreadDashRoot, pubkey: 'pat', content: 'Hi'), ]).single; + final channels = _FakeChannelsNotifier(const []); await tester.pumpWidget( _buildTestable( messages: [], sender: sender, - // The DM was just opened, so the reloading list does not have it yet. - channelsNotifier: _FakeChannelsNotifier(const [])..rememberOpened(dm), + channelsNotifier: channels, home: ThreadDetailPage( threadHead: head, allMessages: [head], @@ -10815,6 +10815,8 @@ void main() { ), ), ); + // The DM was just opened, so the reloading list does not have it yet. + channels.rememberOpened(dm); await tester.pumpAndSettle(); await tester .widget(find.byType(ComposeBar)) diff --git a/mobile/test/features/channels/channels_page_test.dart b/mobile/test/features/channels/channels_page_test.dart index 804a92e0eb..382ba5e12c 100644 --- a/mobile/test/features/channels/channels_page_test.dart +++ b/mobile/test/features/channels/channels_page_test.dart @@ -3123,6 +3123,7 @@ void main() { addTearDown(tester.view.resetPhysicalSize); addTearDown(tester.view.resetDevicePixelRatio); + final waits = []; const directoryUsers = [ DirectoryUser( pubkey: 'alice', @@ -3148,6 +3149,9 @@ void main() { ) .toList(); }), + channelActionsProvider.overrideWith( + (ref) => _PendingDmActions(waits), + ), ], ), ); @@ -3281,6 +3285,11 @@ void main() { ); expect(searchField.textInputAction, TextInputAction.done); expect(searchField.onSubmitted, isNotNull); + + // Done opens the DM at once instead of waiting on the full channel list. + await tester.testTextInput.receiveAction(TextInputAction.done); + await tester.pump(); + expect(waits, [false]); }); testWidgets('wraps many recipients without overflowing above the keyboard', ( @@ -4042,3 +4051,21 @@ class _CompletedPairingNotifier extends PairingNotifier { @override PairingState build() => const PairingState(status: PairingStatus.success); } + +class _PendingDmActions implements ChannelActions { + _PendingDmActions(this.waits); + + final List waits; + + @override + Future openDm({ + required List pubkeys, + bool waitForChannelList = true, + }) { + waits.add(waitForChannelList); + return Completer().future; + } + + @override + dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation); +} diff --git a/mobile/test/features/channels/channels_provider_test.dart b/mobile/test/features/channels/channels_provider_test.dart index 9be1a620df..77afe3fea1 100644 --- a/mobile/test/features/channels/channels_provider_test.dart +++ b/mobile/test/features/channels/channels_provider_test.dart @@ -2320,6 +2320,83 @@ void main() { ); }); + test('a DM opened ahead of the list is not found after it is left', () async { + final session = _FakeRelaySession( + memberships: [_membership(_channelA, myPk)], + metadata: [_meta(id: _channelA, name: 'Alice', channelType: 'dm')], + ); + final container = _buildContainer(session: session); + addTearDown(container.dispose); + final notifier = container.read(channelsProvider.notifier); + final dm = (await container.read(channelsProvider.future)).single; + + // A fast open finds the DM from its own metadata before the list has it. + session.memberships = []; + await notifier.refresh(); + notifier.rememberOpened(dm); + expect(notifier.channelById(_channelA)?.id, _channelA); + + // Once listed, the list owns it, so leaving drops it everywhere. + session.memberships = [_membership(_channelA, myPk)]; + await notifier.refresh(); + session.memberships = []; + await notifier.refresh(); + expect(container.read(channelsProvider).value, isEmpty); + expect(notifier.channelById(_channelA), isNull); + expect(container.read(channelByIdProvider(_channelA)), isNull); + + // The open's background refresh can list the DM before the open + // remembers it. The list still owns it. + session.memberships = [_membership(_channelA, myPk)]; + await notifier.refresh(); + notifier.rememberOpened(dm); + session.memberships = []; + await notifier.refresh(); + expect(notifier.channelById(_channelA), isNull); + }); + + test('a DM opened ahead of the list stays in its community', () async { + final session = _FakeRelaySession( + memberships: [_membership(_channelA, myPk)], + metadata: [_meta(id: _channelA, name: 'Alice', channelType: 'dm')], + ); + final container = _buildContainer(session: session); + addTearDown(container.dispose); + final notifier = container.read(channelsProvider.notifier); + final dm = (await container.read(channelsProvider.future)).single; + session.memberships = []; + await notifier.refresh(); + notifier.rememberOpened(dm); + final found = []; + final lookup = container.listen( + channelByIdProvider(_channelA), + (_, next) => found.add(next?.id), + fireImmediately: true, + ); + addTearDown(lookup.close); + expect(found, [_channelA]); + + // Neither community lists the DM, yet the open lookup lets it go. + container + .read(relayConfigProvider.notifier) + .update(baseUrl: 'https://community-b.example'); + await container.read(channelsProvider.future); + await _settle(); + expect(lookup.read(), isNull); + expect( + container.read(channelsProvider.notifier).channelById(_channelA), + isNull, + ); + + // Likewise across an identity switch. + notifier.rememberOpened(dm); + expect(lookup.read()?.id, _channelA); + container.read(_testPubkeyProvider.notifier).set('other'); + await container.read(channelsProvider.future); + await _settle(); + expect(lookup.read(), isNull); + }); + test( 'archived kind:39000 metadata sets Channel.isArchived (covers TTL auto-archive)', () async { diff --git a/mobile/test/features/channels/compose_bar_test.dart b/mobile/test/features/channels/compose_bar_test.dart index 63c77d5ddc..09f48a5869 100644 --- a/mobile/test/features/channels/compose_bar_test.dart +++ b/mobile/test/features/channels/compose_bar_test.dart @@ -4724,6 +4724,7 @@ void main() { final pat = '1' * 64; List? sentRecipients; List>? sentTags; + var searches = 0; await tester.pumpWidget( _buildComposeBar( uploadService: _testUploadService(signer.nsec), @@ -4744,6 +4745,11 @@ void main() { channelType: 'dm', participantPubkeys: [signer.public, pat], ), + searchPeople: (query) async { + if (query != 'Mary J') return const []; + if (++searches == 1) throw StateError('relay down'); + return [maryJane]; + }, onSend: (_, recipients, {mediaTags = const >[]}) async { sentRecipients = recipients; @@ -4752,6 +4758,21 @@ void main() { ), ); await _expandComposer(tester); + // The picker searches the directory for people outside the DM, as it + // does once the list has the DM, and keeps a failed search's retry. + await tester.enterText(find.byType(TextField), '@Mary J'); + await tester.pump(); + await tester.pump(mentionSearchDebounce * 2); + expect( + find.byKey(const ValueKey('mention-search-error')), + findsOneWidget, + ); + await tester.tap(find.text('Retry')); + await tester.pump(); + await tester.pump(mentionSearchDebounce * 2); + expect(find.text('Mary Jane'), findsOneWidget); + expect(find.text('not in DM'), findsOneWidget); + await tester.enterText(find.byType(TextField), '@Pat'); await tester.pump(); await tester.pumpAndSettle(); diff --git a/mobile/test/features/profile/user_profile_sheet_message_test.dart b/mobile/test/features/profile/user_profile_sheet_message_test.dart index 1ad80bd592..ebdae37a70 100644 --- a/mobile/test/features/profile/user_profile_sheet_message_test.dart +++ b/mobile/test/features/profile/user_profile_sheet_message_test.dart @@ -31,6 +31,7 @@ void main() { tester.view.devicePixelRatio = 1; addTearDown(tester.view.reset); final opened = >[]; + final waits = []; SharedPreferences.setMockInitialValues({}); final prefs = await SharedPreferences.getInstance(); await tester.pumpWidget( @@ -42,7 +43,7 @@ void main() { }), ), channelActionsProvider.overrideWith( - (ref) => _FakeDmActions(dm, opened), + (ref) => _FakeDmActions(dm, opened, waits), ), savedPrefsProvider.overrideWithValue(prefs), // Solo organization is still loading. @@ -70,6 +71,8 @@ void main() { expect(opened, [ [peer], ]); + // The page opens at once instead of waiting on the full channel list. + expect(waits, [false]); expect( tester.widget(find.byType(ChannelDetailPage)).channel, dm, @@ -92,10 +95,11 @@ class _UserCache extends UserCacheNotifier { } class _FakeDmActions implements ChannelActions { - _FakeDmActions(this.dm, this.opened); + _FakeDmActions(this.dm, this.opened, this.waits); final Channel dm; final List> opened; + final List waits; @override Future openDm({ @@ -103,6 +107,7 @@ class _FakeDmActions implements ChannelActions { bool waitForChannelList = true, }) async { opened.add(pubkeys); + waits.add(waitForChannelList); return dm; } diff --git a/mobile/test/features/search/search_page_test.dart b/mobile/test/features/search/search_page_test.dart index 11ee833fcb..4f4aebd03d 100644 --- a/mobile/test/features/search/search_page_test.dart +++ b/mobile/test/features/search/search_page_test.dart @@ -131,6 +131,7 @@ void main() { isMember: true, ); final opened = >[]; + final waits = []; SharedPreferences.setMockInitialValues({}); final prefs = await SharedPreferences.getInstance(); await tester.pumpWidget( @@ -152,7 +153,7 @@ void main() { profileProvider.overrideWith(() => _FakeProfileNotifier()), channelsProvider.overrideWith(() => _FakeChannelsNotifier()), channelActionsProvider.overrideWith( - (ref) => _FakeDmActions(dm, opened), + (ref) => _FakeDmActions(dm, opened, waits), ), savedPrefsProvider.overrideWithValue(prefs), // Solo organization is still loading. @@ -169,6 +170,8 @@ void main() { expect(opened, [ ['maya'], ]); + // The page opens at once instead of waiting on the full channel list. + expect(waits, [false]); final page = tester.widget( find.byType(ChannelDetailPage), ); @@ -999,10 +1002,11 @@ class _FakeChannelsNotifier extends ChannelsNotifier { } class _FakeDmActions implements ChannelActions { - _FakeDmActions(this.dm, this.opened); + _FakeDmActions(this.dm, this.opened, this.waits); final Channel dm; final List> opened; + final List waits; @override Future openDm({ @@ -1010,6 +1014,7 @@ class _FakeDmActions implements ChannelActions { bool waitForChannelList = true, }) async { opened.add(pubkeys); + waits.add(waitForChannelList); return dm; } diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index f833ad93b5..77319c0795 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -206,6 +206,9 @@ run_unit_tests() { run_test_step "buzz-relay side-effects helper tests" \ cargo test -p buzz-relay --lib handlers::side_effects::tests:: -- --nocapture + run_test_step "buzz-relay coalesced refresh tests" \ + cargo test -p buzz-relay --lib handlers::coalesced_refresh::tests:: -- --nocapture + run_test_step "buzz-relay system channel scope tests" \ cargo test -p buzz-relay --lib handlers::system_channels::tests:: -- --nocapture