From a362fecc2389955f942c9581bdfeba379ab115b3 Mon Sep 17 00:00:00 2001 From: Will Pfleger Date: Tue, 18 Aug 2026 18:35:50 -0400 Subject: [PATCH] perf(desktop): parallelize relay agent directory rebuild (#6258) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The shared-agent directory rebuild resolves runtime directories, owner profiles, and managed policies for every candidate agent via dozens of exact-author query batches. Issuing those batches serially made a ~100-agent rebuild take 6.5–8.4s, which dominated @mention autocomplete latency. #6224 already scoped the *send-path* revalidation (`revalidate_relay_agents`) to just the mentioned pubkeys, so the send stall is fixed. But the autocomplete directory (`list_relay_agents`) still rebuilds the full membership set serially — this PR removes that remaining cost. ## Change Run the query batches with bounded concurrency via a shared `query_filter_batches` helper. Each directory rebuild constructs one `tokio::sync::Semaphore` (8 permits) and shares it across all of that rebuild's phases, so the runtime-directory and owner-profile phases that run concurrently under one `try_join!` never exceed 8 `/query` requests in flight together — the bound is per-rebuild. The policy phase reuses the same budget. Same events, keyed by pubkey downstream so ordering is irrelevant. Both `list_relay_agents` (autocomplete) and `revalidate_relay_agents` (scoped send-path check from #6224) funnel through `list_relay_agents_for_selection`, so the helper is a no-op for the tiny 1–3-mention revalidation set and only the full autocomplete rebuild sees the win — the scoped send path is untouched. ## Measurement Live on the production relay, warm connection, full `list_relay_agents`: | | Full-directory rebuild | |---|---| | Serial (before) | 6.5–8.4s | | Bounded-concurrency (after) | 2.0–3.4s | Returned pubkey set is byte-identical pre/post. The 8-permit ceiling holds under saturation (24 batched requests → peak exactly 8, zero failures, zero requests left in flight). Signed-off-by: Will Pfleger Co-authored-by: Duncan --- .../agent_discovery/relay_directory.rs | 84 +++++++++++++------ 1 file changed, 60 insertions(+), 24 deletions(-) diff --git a/desktop/src-tauri/src/commands/agent_discovery/relay_directory.rs b/desktop/src-tauri/src/commands/agent_discovery/relay_directory.rs index d00969bf19..976519a076 100644 --- a/desktop/src-tauri/src/commands/agent_discovery/relay_directory.rs +++ b/desktop/src-tauri/src/commands/agent_discovery/relay_directory.rs @@ -9,6 +9,41 @@ use crate::{ const RELAY_DIRECTORY_PAGE_SIZE: usize = 500; const RELAY_FILTER_BATCH_SIZE: usize = 10; +/// Per-rebuild ceiling on directory-rebuild `/query` requests in flight at once. +/// The rebuild fans dozens of exact-author batches across the relay; issuing +/// them serially dominated agent-mention send latency (~6 s for ~100 +/// candidates). A bounded window collapses that to a few round trips while +/// keeping the request rate well under the relay's admission gate, which +/// back-pressures any 429 anyway. Each rebuild builds one semaphore and shares +/// it across every phase, so a single rebuild's runtime-directory and +/// owner-profile phases — which run concurrently under one `try_join!` — never +/// exceed it together. (Overlapping rebuilds each hold their own budget.) +const RELAY_DIRECTORY_MAX_CONCURRENCY: usize = 8; + +/// Run one `query_relay` request per `RELAY_FILTER_BATCH_SIZE` chunk of +/// `filters`, each acquiring a permit from `semaphore` so the total in-flight +/// request count stays within the shared ceiling even when several batch sets +/// run concurrently. Returned events are concatenated; order is unspecified — +/// every caller keys the events by pubkey downstream, so ordering is irrelevant. +async fn query_filter_batches( + state: &AppState, + semaphore: &tokio::sync::Semaphore, + filters: &[serde_json::Value], + error_label: &str, +) -> Result, String> { + let pages = futures_util::future::try_join_all(filters.chunks(RELAY_FILTER_BATCH_SIZE).map( + |batch| async move { + let _permit = semaphore.acquire().await.map_err(|error| { + format!("{error_label}: directory concurrency semaphore closed: {error}") + })?; + query_relay(state, batch) + .await + .map_err(|error| format!("{error_label}: {error}")) + }, + )) + .await?; + Ok(pages.into_iter().flatten().collect()) +} fn exact_author_filters(pubkeys: &[String], kind: u16) -> Vec { pubkeys @@ -129,24 +164,26 @@ async fn list_relay_agents_for_selection( return Ok(Vec::new()); } - let mut directory_events = Vec::new(); - let mut profile_events = Vec::new(); let directory_filters = exact_author_filters(&candidate_pubkeys, 10100); let profile_filters = exact_author_filters(&candidate_pubkeys, 0); - for filter_offset in (0..candidate_pubkeys.len()).step_by(RELAY_FILTER_BATCH_SIZE) { - let filter_end = (filter_offset + RELAY_FILTER_BATCH_SIZE).min(candidate_pubkeys.len()); - let (directory, profiles) = tokio::join!( - query_relay(state, &directory_filters[filter_offset..filter_end]), - query_relay(state, &profile_filters[filter_offset..filter_end]), - ); - directory_events.extend( - directory - .map_err(|error| format!("relay agent runtime-directory query failed: {error}"))?, - ); - profile_events.extend( - profiles.map_err(|error| format!("relay agent owner-profile query failed: {error}"))?, - ); - } + // One semaphore per rebuild caps `/query` requests across this rebuild's + // phases, so its runtime-directory and owner-profile phases below stay + // within the ceiling even though `try_join!` runs them concurrently. + let semaphore = tokio::sync::Semaphore::new(RELAY_DIRECTORY_MAX_CONCURRENCY); + let (directory_events, profile_events) = tokio::try_join!( + query_filter_batches( + state, + &semaphore, + &directory_filters, + "relay agent runtime-directory query failed", + ), + query_filter_batches( + state, + &semaphore, + &profile_filters, + "relay agent owner-profile query failed", + ), + )?; // Only the agent's signed NIP-OA profile can name the owner coordinate to // query. Each exact `(owner, d=agent)` filter returns at most one current @@ -161,14 +198,13 @@ async fn list_relay_agents_for_selection( retain_verified_owner(&mut verified_owners, &viewer_pubkey); } let managed_filters = managed_policy_filters(&candidate_pubkeys, &verified_owners); - let mut managed_agent_events = Vec::new(); - for filters in managed_filters.chunks(RELAY_FILTER_BATCH_SIZE) { - managed_agent_events.extend( - query_relay(state, filters) - .await - .map_err(|error| format!("relay agent managed-policy query failed: {error}"))?, - ); - } + let managed_agent_events = query_filter_batches( + state, + &semaphore, + &managed_filters, + "relay agent managed-policy query failed", + ) + .await?; let mut agents = nostr_convert::relay_agents_from_directory_events( &directory_events,