perf(concord): Volley Sync — boot paint for every community in seconds, not minutes

Boot-to-messages on a 72-channel account: 95-112s of serial per-channel
chains down to ~8s total, with the hottest channels painting in 1-4s.
Chat planes are flat, linear, non-authoritative data — they paint from
stored state immediately; the control plane verifies behind the paint
and retro-hide revokes anything a ban invalidates.

- Volley Sync (community/v2/volley.rs): every v2 channel's latest page
  in mass-batched multi-filter REQs on the shared warm client, most
  recently active channels fired first. One plane per channel at the
  MAX HELD epoch (the community row's epoch fields can lag a rotation);
  per-filter since = the chat's last held message. Per-relay-set
  pipelines gate on their own first live socket. Auth-gating relays
  refuse batches by protocol, so a bounded per-plane authed fallback
  covers exactly the channels whose live relays all gate (learned
  durably from NIP-42 challenges via the auth_gate KV), plus a recency
  confirmation for active channels when a flaky open relay may hold
  holes. Ingest reuses the backfill pipeline (v2_ingest_chat_page).
- v1 hot lane: the 12 most recently active v1 channels run their full
  probe-cheapened chain 12-wide, ahead of the sweep; the sweep skips
  the lane's successes.
- Evidence doctrine enforced in transport: the until→Full floor is
  gone — chat reads ride their declared tier. Every completeness-
  sensitive conclusion requests Full at its own call site (v1
  history-start latch, join-verify genesis anchor, refound compaction,
  guestbook folds); tolerant control reads declare Quorum explicitly.
- Boot order: chat paint first (volley ∥ control probe, then hot
  lane), consensus follows and the verification sweep behind it — boot
  exists to deliver message data. A reconnect landing mid-sweep queues
  one follow-up run instead of being dropped.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: JSKitty

crates/vector-core/src/community/send.rs

@@ -215,10 +215,11 @@ pub async fn fetch_channel_page<T: Transport + ?Sized>(
         // z_tags, above), and back-pagination passes `None` here.
         since,
         limit: Some(limit),
-        // Latest pages are positive-data reads. Older pages (`until` set) are
-        // force-promoted to Full by the transport — the history-start latch
-        // needs the completest union the reachable relays allow.
-        evidence: Evidence::Fast,
+        // Latest pages are positive-data reads and ride Fast. Older pages
+        // request Full HERE (no transport floor does it anymore): this is the
+        // one fetch whose short result latches "history starts here", and an
+        // absence verdict trusts only the completest reachable union.
+        evidence: if until.is_some() { Evidence::Full } else { Evidence::Fast },
         ..Default::default()
     };
     transport.fetch(&query, &community.relays).await

crates/vector-core/src/community/transport.rs

@@ -411,15 +411,13 @@ fn breaker_record_at(generation: u64, url: &str, success: bool, full_budget: boo
     })
 }
 
-/// `until` forces Full — a back-page verdict (the history-start latch) trusts
-/// "nothing older than the cursor" only against the completest union the
-/// reachable relay set allows. A floor in the transport, not trust in callers.
+/// Callers own their evidence tier. The chat plane is flat, linear data — an
+/// event exists or it doesn't — so no transport floor promotes its reads.
+/// Every site that draws a completeness-sensitive conclusion from an `until`
+/// walk (the v1 history-start latch, join-verify's genesis anchor, refound
+/// compaction, guestbook folds) REQUESTS Full explicitly at its own Query.
 fn effective_evidence(query: &Query) -> Evidence {
-    if query.until.is_some() {
-        Evidence::Full
-    } else {
-        query.evidence
-    }
+    query.evidence
 }
 
 // ── Plane connection pool (fetch_plane) ─────────────────────────────────────
@@ -783,7 +781,7 @@ impl LiveTransport {
     /// the pool doesn't hold yet is added idempotently (mirrors what the realtime subscription does), then
     /// `connect()` kicks it without disturbing the already-connected majority. Never shut this client down:
     /// it is shared. Errors only if there is no client yet or every relay url was invalid.
-    async fn warm_client(relays: &[String], connect_timeout: std::time::Duration) -> Result<Client, String> {
+    pub(crate) async fn warm_client(relays: &[String], connect_timeout: std::time::Duration) -> Result<Client, String> {
         if relays.is_empty() {
             return Err("community has no relays configured".to_string());
         }
@@ -1915,12 +1913,12 @@ mod tests {
     // ── The evidence floor ───────────────────────────────────────────────────
 
     #[test]
-    fn until_forces_full_evidence_and_default_is_quorum() {
+    fn declared_evidence_stands_and_default_is_quorum() {
         assert_eq!(Query::default().evidence, Evidence::Quorum, "unclassified sites get Quorum");
         assert_eq!(
             effective_evidence(&Query { until: Some(1), evidence: Evidence::Fast, ..Default::default() }),
-            Evidence::Full,
-            "a back-page can never ride Fast — the history-start latch needs the full union"
+            Evidence::Fast,
+            "chat pagination rides its declared tier — absence verdicts request Full themselves"
         );
         assert_eq!(
             effective_evidence(&Query { evidence: Evidence::Fast, ..Default::default() }),

crates/vector-core/src/community/v2/mod.rs

@@ -25,6 +25,7 @@ pub mod inbound;
 pub mod invite;
 pub mod list;
 pub mod realtime;
+pub mod volley;
 pub mod rekey;
 pub mod roles;
 pub mod service;

crates/vector-core/src/community/v2/service.rs

@@ -535,7 +535,7 @@ pub async fn fetch_channel<T: Transport + ?Sized>(
     channel_id: &ChannelId,
     limit: usize,
 ) -> Result<Vec<FetchedEvent>, String> {
-    fetch_channel_history(transport, community, channel_id, limit, 1, |_| true).await
+    fetch_channel_history(transport, community, channel_id, limit, 1, None, crate::community::transport::Evidence::Quorum, |_| true).await
 }
 
 /// Walk a channel's history newest-first (CORD-03 §3 "clients load a Channel
@@ -557,6 +557,8 @@ pub async fn fetch_channel_history<T: Transport + ?Sized>(
     channel_id: &ChannelId,
     page: usize,
     max_pages: usize,
+    since: Option<u64>,
+    evidence: crate::community::transport::Evidence,
     mut keep_paging: impl FnMut(&[FetchedEvent]) -> bool,
 ) -> Result<Vec<FetchedEvent>, String> {
     // Guards the opportunistic scrub-key heals below — the fetch loop straddles
@@ -609,8 +611,10 @@ pub async fn fetch_channel_history<T: Transport + ?Sized>(
             let q = Query {
                 kinds: vec![stream::KIND_WRAP],
                 authors: vec![plane.pk_hex()],
+                since,
                 until,
                 limit: Some(page),
+                evidence,
                 ..Default::default()
             };
             if let Ok(evs) = transport.fetch_plane(plane.keys(), &q, &community.relays).await {
@@ -1049,11 +1053,15 @@ pub async fn community_is_public<T: Transport + ?Sized>(transport: &T, community
     let mut oldest: Option<u64> = None;
     let mut until: Option<u64> = None;
     for page in 0..COMPACT_MAX_PAGES {
+        // Quorum, DECLARED (the until→Full transport floor is gone): these
+        // control reads tolerate a partial union — their fold semantics are
+        // fail-safe on gaps (seeded banlists, withheld roster cache).
         let query = Query {
             kinds: vec![stream::KIND_WRAP],
             authors: vec![control.pk_hex()],
             until,
             limit: Some(FOLLOW_PAGE),
+            evidence: crate::community::transport::Evidence::Quorum,
             ..Default::default()
         };
         let Ok(wraps) = transport.fetch(&query, &community.relays).await else { return true };
@@ -1282,10 +1290,11 @@ async fn verify_owner_root_and_reconcile<T: Transport + ?Sized>(
     // T's genesis onto a fake root to MITM another T-joiner — is closed only by
     // binding the root into community_id (protocol, deferred).
     //
-    // Seed `until` with a FAR-FUTURE constant (NOT now-based): `until.is_some()` takes
-    // the transport's AUTHORITATIVE drain-ALL-relays path (an open `until` returns only
-    // a fast relay's partial window and misses a genesis on a lagging relay — routine
-    // over Tor), while a constant beyond any real created_at clips NOTHING — so neither
+    // Seed `until` with a FAR-FUTURE constant (NOT now-based), and request
+    // Evidence::Full EXPLICITLY below: this walk draws an ABSENCE verdict (no
+    // owner-signed genesis ⇒ reject), which trusts only the completest union —
+    // an open partial window misses a genesis on a lagging relay (routine over
+    // Tor). A constant beyond any real created_at clips NOTHING — so neither
     // a clock-skewed future-dated genesis nor a >1h-slow-clock joiner is excluded (a
     // now-based bound could clip either). Break on an EMPTY page (a short page is a
     // relay cap). A forged root walks to exhaustion and rejects; a flood/deep plane
@@ -1323,6 +1332,7 @@ async fn verify_owner_root_and_reconcile<T: Transport + ?Sized>(
                 authors: vec![control_pk.clone()],
                 until,
                 limit: Some(PAGE),
+                evidence: crate::community::transport::Evidence::Full,
                 ..Default::default()
             };
             let wraps = transport.fetch(&query, &community.relays).await?;
@@ -1720,11 +1730,15 @@ pub async fn fetch_authority<T: Transport + ?Sized>(transport: &T, community: &C
     // silently un-ban on withheld data.
     let mut a = fold_authority(community, &[], &floors);
     for _ in 0..FOLLOW_MAX_PAGES {
+        // Quorum, DECLARED (the until→Full transport floor is gone): these
+        // control reads tolerate a partial union — their fold semantics are
+        // fail-safe on gaps (seeded banlists, withheld roster cache).
         let query = Query {
             kinds: vec![stream::KIND_WRAP],
             authors: vec![control.pk_hex()],
             until,
             limit: Some(FOLLOW_PAGE),
+            evidence: crate::community::transport::Evidence::Quorum,
             ..Default::default()
         };
         let Ok(wraps) = transport.fetch(&query, &community.relays).await else { break };
@@ -1785,7 +1799,9 @@ async fn fetch_guestbook_events<T: Transport + ?Sized>(
     let mut until: Option<u64> = None;
     let mut oldest: Option<u64> = None;
     for _ in 0..GB_MAX_PAGES {
-        let query = Query { kinds: vec![stream::KIND_WRAP], authors: vec![gb_group.pk_hex()], until, limit: Some(GB_PAGE), ..Default::default() };
+        // Full: this set becomes the refound's recipient list — a member's
+        // Join visible only on a minority relay must not be severed.
+        let query = Query { kinds: vec![stream::KIND_WRAP], authors: vec![gb_group.pk_hex()], until, limit: Some(GB_PAGE), evidence: crate::community::transport::Evidence::Full, ..Default::default() };
         let wraps = transport.fetch(&query, &community.relays).await?;
         let mut fresh = 0usize;
         for wrap in &wraps {
@@ -2155,11 +2171,15 @@ pub async fn refound_community<T: Transport + ?Sized>(transport: &T, community:
     // coverage test, so stopping there could compact it away.
     let mut truncated = false;
     for page in 0..COMPACT_MAX_PAGES {
+        // Full: compaction re-wraps the head set it can SEE — a control
+        // edition (a ban head) reachable only on a minority relay must not be
+        // compacted away by a partial union.
         let query = Query {
             kinds: vec![stream::KIND_WRAP],
             authors: vec![current_control.pk_hex()],
             until,
             limit: Some(FOLLOW_PAGE),
+            evidence: crate::community::transport::Evidence::Full,
             ..Default::default()
         };
         let wraps = transport.fetch(&query, &community.relays).await?;
@@ -2444,7 +2464,7 @@ pub async fn refound_at_birth<T: Transport + ?Sized>(
     // Exhaustion, not coverage — see the sibling read in `refound_community`.
     let mut truncated = false;
     for page in 0..COMPACT_MAX_PAGES {
-        let query = Query { kinds: vec![stream::KIND_WRAP], authors: vec![current_control.pk_hex()], until, limit: Some(FOLLOW_PAGE), ..Default::default() };
+        let query = Query { kinds: vec![stream::KIND_WRAP], authors: vec![current_control.pk_hex()], until, limit: Some(FOLLOW_PAGE), evidence: crate::community::transport::Evidence::Full, ..Default::default() };
         let wraps = transport.fetch(&query, &community.relays).await?;
         let mut fresh = 0usize;
         for w in &wraps {
@@ -3541,11 +3561,15 @@ pub async fn follow_control<T: Transport + ?Sized>(
     // device's baseline is the one step that outlives the round.
     let mut truncated = true;
     for _ in 0..FOLLOW_MAX_PAGES {
+        // Quorum, DECLARED (the until→Full transport floor is gone): these
+        // control reads tolerate a partial union — their fold semantics are
+        // fail-safe on gaps (seeded banlists, withheld roster cache).
         let query = Query {
             kinds: vec![stream::KIND_WRAP],
             authors: vec![control.pk_hex()],
             until,
             limit: Some(FOLLOW_PAGE),
+            evidence: crate::community::transport::Evidence::Quorum,
             ..Default::default()
         };
         let wraps = transport.fetch(&query, &community.relays).await?;
@@ -8139,7 +8163,7 @@ mod tests {
         let general = community.channels[0].id;
         flood_general(&relay, &community, &owner, 120, 10_000).await;
 
-        let all = fetch_channel_history(&relay, &community, &general, 50, 8, |_| true).await.unwrap();
+        let all = fetch_channel_history(&relay, &community, &general, 50, 8, None, crate::community::transport::Evidence::Quorum, |_| true).await.unwrap();
         assert_eq!(all.len(), 120, "the walk pages the whole burst");
         // Oldest→newest, no duplicates.
         let contents: Vec<String> = all.iter().map(|f| f.event.opened().rumor.content.clone()).collect();
@@ -8164,7 +8188,7 @@ mod tests {
 
         // The caller says "I hold everything" after the first page — no deeper fetch.
         let mut pages = 0usize;
-        let got = fetch_channel_history(&relay, &community, &general, 50, 8, |_| {
+        let got = fetch_channel_history(&relay, &community, &general, 50, 8, None, crate::community::transport::Evidence::Quorum, |_| {
             pages += 1;
             false
         })
@@ -8190,7 +8214,7 @@ mod tests {
             let (wrap, _) = chat::seal_chat_rumor(&rumor, &group, &owner, Timestamp::from_secs(5_000), false).unwrap();
             relay.publish(&wrap, &community.relays).await.unwrap();
         }
-        let got = fetch_channel_history(&relay, &community, &general, 25, 8, |_| true).await.unwrap();
+        let got = fetch_channel_history(&relay, &community, &general, 25, 8, None, crate::community::transport::Evidence::Quorum, |_| true).await.unwrap();
         assert!(got.len() >= 25, "at least the relay page is read");
         assert!(got.len() <= 60, "sane bound");
         // Termination is the assertion: reaching here means the wall didn't loop.

crates/vector-core/src/community/v2/streamauth.rs

@@ -205,6 +205,15 @@ pub fn ensure_responder(client: &Client) {
             if let ClientNotification::Message { relay_url, message } = n {
                 if let nostr_sdk::prelude::RelayMessage::Auth { challenge } = *message {
                     let challenge = challenge.into_owned();
+                    // Durable capability fact: this relay gates reads behind
+                    // NIP-42 (the boot volley routes fallbacks by it). Write
+                    // once — a challenge-spamming relay must not drive DB
+                    // writes at frame rate on the notification loop.
+                    let gate_key =
+                        format!("auth_gate:{}", relay_url.as_str().trim_end_matches('/'));
+                    if crate::db::get_sql_setting(gate_key.clone()).ok().flatten().is_none() {
+                        let _ = crate::db::set_sql_setting(gate_key, "1".to_string());
+                    }
                     let fresh_connection = remember_challenge(&relay_url, &challenge);
                     if !is_empty() {
                         authenticate_streams(&client, &relay_url, &challenge).await;