From 0c38ee80468a8971c094aa1343663671ffe346a6 Mon Sep 17 00:00:00 2001 From: Ian Clarke Date: Sat, 17 Oct 2026 11:31:56 -0500 Subject: [PATCH 1/9] perf(ui): cut the round trips that held the first load after a re-key (#809) Measured in a rehearsal (pre-#703 UI -> main, isolated gateway seeded with production's 70 delegate registrations, 60 ms each way): rooms appeared at 16.1 s. The console showed where the time went: - the migration journal's `begin` read every predecessor's done and in-progress markers one round trip at a time (136 marker reads, ~8 s), and the sweep's merge waits for `begin`; - the sweep asked the legacy generations oldest first, so the newest one (which held the rooms) sat on the node behind every older one; - a legacy generation's room slots were fetched one round trip per room. `begin` now prefetches the markers in concurrent waves of 16 (definitive answers only; silence is re-read live and still refuses the start), the sweep asks newest first, and legacy slots are fetched in waves like the current delegate's per-room load. Claude-Session: https://claude.ai/code/session_01Mynjd42gMDNsNSxwbJWKu3 --- ui/src/components/app/chat_delegate.rs | 19 +- .../app/freenet_api/delegate_migration.rs | 172 +++++++++++++++++- .../app/freenet_api/response_handler.rs | 108 +++++++++-- 3 files changed, 276 insertions(+), 23 deletions(-) diff --git a/ui/src/components/app/chat_delegate.rs b/ui/src/components/app/chat_delegate.rs index c28b6a80c..68408c1ef 100644 --- a/ui/src/components/app/chat_delegate.rs +++ b/ui/src/components/app/chat_delegate.rs @@ -4762,6 +4762,17 @@ mod tests { ); } + /// freenet/river#809: the sweep asks the NEWEST legacy generation first, so + /// its answer is not queued on the node behind every older generation. + #[test] + fn legacy_sweep_asks_the_newest_generation_first() { + let src = include_str!("chat_delegate.rs"); + let body = &src[src + .rfind("pub(crate) async fn fire_legacy_migration_request() -> bool {") + .expect("fire_legacy_migration_request must exist")..]; + assert!(body.contains("in LEGACY_DELEGATES.iter().enumerate().rev() {")); + } + // ===== resolve_hidden_thread_hydration tests (#261 Codex P3) ===== use freenet_scaffold::util::FastHash; @@ -11315,7 +11326,13 @@ pub(crate) async fn fire_legacy_migration_request() -> bool { let attempt = LOAD_ATTEMPT_GEN.load(Ordering::Relaxed); let mut any_dispatched = false; - for (i, (key_bytes, code_hash_bytes)) in LEGACY_DELEGATES.iter().enumerate() { + // Newest generation first (freenet/river#809). The node runs delegate + // requests one at a time, in arrival order, so oldest-first put the + // generation most likely to hold the user's rooms behind every older one, + // each paying a module load on a cold node (0.4 s each in the rehearsal). + // Arrival order decides nothing else: every merge is ranked by source + // generation, never by arrival (freenet/river#527). + for (i, (key_bytes, code_hash_bytes)) in LEGACY_DELEGATES.iter().enumerate().rev() { let legacy_code_hash = CodeHash::new(*code_hash_bytes); let legacy_delegate_key = DelegateKey::new(*key_bytes, legacy_code_hash); diff --git a/ui/src/components/app/freenet_api/delegate_migration.rs b/ui/src/components/app/freenet_api/delegate_migration.rs index ac7c19877..39751c210 100644 --- a/ui/src/components/app/freenet_api/delegate_migration.rs +++ b/ui/src/components/app/freenet_api/delegate_migration.rs @@ -1467,6 +1467,11 @@ pub(crate) struct RiverSuccessorIo<'a, C: RiverDelegateChannel, S: RiverImportSe /// merged none cannot have resurrected anything, and re-merging would only /// dirty the rooms and cost a redundant save. rooms_merged: bool, + /// Marker values read ahead by [`Self::prefetch_markers`], by storage key. + /// Each is used by at most one [`Self::get_marker`] and is dropped when the + /// key is written, so only a read that would otherwise have gone to the + /// node a moment later is answered from here (freenet/river#809). + prefetched_markers: HashMap, Option>>, } impl<'a, C: RiverDelegateChannel, S: RiverImportSeam> RiverSuccessorIo<'a, C, S> { @@ -1481,6 +1486,60 @@ impl<'a, C: RiverDelegateChannel, S: RiverImportSeam> RiverSuccessorIo<'a, C, S> pending_provenance: Vec::new(), walk_orders: Vec::new(), rooms_merged: false, + prefetched_markers: HashMap::new(), + } + } + + /// Read the done and in-progress markers of every predecessor in + /// `lineage` concurrently, in waves of [`MARKER_PREFETCH_WAVE`], so that + /// `begin` does not read them one round trip at a time (freenet/river#809: + /// two reads per generation, 66 round trips for 33 generations, which held + /// the first load after a re-key for about 8 s at 120 ms RTT). + /// + /// Only definitive answers are kept: a marker that is silent or errs here + /// is read again, one at a time, when `begin` asks for it, and fails there + /// exactly as it would have without the prefetch. Silence is still never + /// read as absence. + pub(crate) async fn prefetch_markers(&mut self, lineage: &[DelegateLineageEntry]) { + let keys: Vec> = lineage + .iter() + .flat_map(|entry| { + let predecessor = + DelegateKey::new(entry.delegate_key, CodeHash::new(entry.code_hash)); + [ + pred_done_marker_key(&predecessor), + pred_wip_marker_key(&predecessor), + ] + }) + .collect(); + for wave in keys.chunks(MARKER_PREFETCH_WAVE) { + // `request_all`, NOT a `join_all` of `request`: see `prewarm`. + let outcomes = self + .channel + .request_all( + wave.iter() + .map(|key| { + ( + self.successor.clone(), + ChatDelegateRequestMsg::GetRequest { + key: ChatDelegateKey::new(key.clone()), + }, + ) + }) + .collect(), + ) + .await; + for (key, outcome) in wave.iter().zip(outcomes) { + if let Ok(Some(ChatDelegateResponseMsg::GetResponse { + key: reply_key, + value, + })) = outcome + { + if reply_key.as_bytes() == key.as_slice() { + self.prefetched_markers.insert(key.clone(), value); + } + } + } } } @@ -1521,6 +1580,9 @@ impl<'a, C: RiverDelegateChannel, S: RiverImportSeam> RiverSuccessorIo<'a, C, S> /// guessing `None` would re-import a possibly-done predecessor, and /// guessing `Done` would silently skip one. async fn get_marker(&mut self, storage_key: Vec) -> Result>, String> { + if let Some(value) = self.prefetched_markers.remove(&storage_key) { + return Ok(value); + } let request = ChatDelegateRequestMsg::GetRequest { key: ChatDelegateKey::new(storage_key.clone()), }; @@ -1557,6 +1619,7 @@ impl<'a, C: RiverDelegateChannel, S: RiverImportSeam> RiverSuccessorIo<'a, C, S> /// is durable when this returns (the crate's record_marker and /// store_journal contracts — never batched with items). async fn put_marker(&mut self, storage_key: Vec, value: &[u8]) -> Result<(), String> { + self.prefetched_markers.remove(&storage_key); put_current(self.channel, &self.successor, storage_key, value).await } @@ -1571,6 +1634,7 @@ impl<'a, C: RiverDelegateChannel, S: RiverImportSeam> RiverSuccessorIo<'a, C, S> saw_data: bool, ) -> Result<(), String> { let key = ChatDelegateKey::new(pred_wip_marker_key(predecessor)); + self.prefetched_markers.remove(key.as_bytes()); let (mut current, mut generation) = match self .channel .request( @@ -1636,6 +1700,15 @@ impl<'a, C: RiverDelegateChannel, S: RiverImportSeam> RiverSuccessorIo<'a, C, S> } } +/// How many marker reads [`RiverSuccessorIo::prefetch_markers`] keeps in +/// flight at once. Every delegate request on a node shares ONE admission queue +/// per priority tier, capped at 100 (`MAX_QUEUED_PER_CONTRACT`, freenet-core +/// `fair_queue.rs`), and the sweep's legacy burst lands in it at the same time; +/// a wave of 16 keeps well under the cap, as the room load's waves do +/// (`ROOM_LOAD_CONCURRENCY`). +pub(crate) const MARKER_PREFETCH_WAVE: usize = 16; +const _: () = assert!(MARKER_PREFETCH_WAVE > 0); + /// Compare-and-swap attempts for one in-progress marker write. Each conflict /// means another session wrote it in between; a handful is plenty. const WIP_MARKER_CAS_ATTEMPTS: u32 = 5; @@ -1956,6 +2029,7 @@ pub(crate) async fn begin_with( lineage: &[DelegateLineageEntry], ) -> Result { let mut writer = RiverSuccessorIo::new(channel, NoImports, successor); + writer.prefetch_markers(lineage).await; begin_migration_journal(&mut writer, lineage, &river_secret_policy()).await } @@ -2595,6 +2669,10 @@ pub(crate) mod tests_support { pub(crate) max_in_flight: Cell, /// `(target delegate key bytes, "list" | "get:" | "store:")` pub(crate) log: RefCell, String)>>, + /// Sequential round trips: each `request` outside a batch, and each + /// `request_all` batch, counts one (freenet/river#809). + pub(crate) round_trips: Cell, + batching: Cell, } impl MockChannel { @@ -2802,6 +2880,9 @@ pub(crate) mod tests_support { target: &DelegateKey, request: ChatDelegateRequestMsg, ) -> Result, String> { + if !self.batching.get() { + self.round_trips.set(self.round_trips.get() + 1); + } self.in_flight.set(self.in_flight.get() + 1); self.max_in_flight .set(self.max_in_flight.get().max(self.in_flight.get())); @@ -2824,12 +2905,16 @@ pub(crate) mod tests_support { &self, requests: Vec<(DelegateKey, ChatDelegateRequestMsg)>, ) -> Vec, String>> { - futures::future::join_all( + self.round_trips.set(self.round_trips.get() + 1); + self.batching.set(true); + let out = futures::future::join_all( requests .iter() .map(|(target, request)| self.request(target, request.clone())), ) - .await + .await; + self.batching.set(false); + out } } @@ -4935,6 +5020,89 @@ mod tests { ); } + /// A lineage of `n` generations, every predecessor installed and empty. + fn lineage_n(n: u8) -> (Vec, MockChannel) { + let table: Vec<([u8; 32], [u8; 32])> = + (0..n).map(|g| ([20 + g; 32], [120 + g; 32])).collect(); + let mut channel = + MockChannel::default().with_delegate(&successor_key(), DelegateSim::default()); + for (k, ch) in &table { + channel = channel.with_delegate( + &DelegateKey::new(*k, CodeHash::new(*ch)), + DelegateSim::default(), + ); + } + (river_delegate_lineage(&table), channel) + } + + /// freenet/river#809: `begin` read every predecessor's markers one round + /// trip at a time (done, then in-progress), so on the first load after a + /// re-key the sweep's merge waited 2 x 33 round trips for it: about 8 s of + /// the 16 s the rehearsal measured before any room showed. The marker reads + /// now go out in bounded concurrent waves, so `begin` costs a handful of + /// round trips whatever the lineage length. + #[test] + fn begin_reads_markers_in_waves_not_one_round_trip_each() { + let rounds = |n: u8| { + let (lineage, channel) = lineage_n(n); + block_on(begin_with(&channel, successor_key(), &lineage)).unwrap(); + channel.round_trips.get() + }; + let small = rounds(3); + let large = rounds(33); + let extra_waves = (2 * 33usize).div_ceil(MARKER_PREFETCH_WAVE) + - (2 * 3usize).div_ceil(MARKER_PREFETCH_WAVE); + assert!( + large <= small + extra_waves, + "begin over 33 generations took {large} round trips against {small} over 3: \ + its marker reads must not grow one round trip per generation" + ); + } + + /// The prefetch never turns silence into absence (constraint 1): a marker + /// the prefetch could not read is read again live, and a marker that stays + /// silent still refuses `begin`, exactly as without the prefetch. + #[test] + fn begin_still_refuses_on_a_marker_that_stays_silent() { + let (lineage, channel) = lineage_n(3); + let silent = pred_done_marker_key(&DelegateKey::new( + lineage[1].delegate_key, + CodeHash::new(lineage[1].code_hash), + )); + channel + .delegates + .borrow_mut() + .get_mut(successor_key().bytes()) + .unwrap() + .silent_get_keys + .push(silent.clone()); + assert!(block_on(begin_with(&channel, successor_key(), &lineage)).is_err()); + let reads = channel + .log + .borrow() + .iter() + .filter(|(_, what)| *what == format!("get:{}", String::from_utf8_lossy(&silent))) + .count(); + assert_eq!(reads, 2, "read once by the prefetch, then again live"); + } + + /// A prefetched marker is used at most once and is dropped by a write, so + /// the in-progress marker's read-back (the crate's marker contract) sees + /// what was just recorded, never the value read before it. + #[test] + fn a_prefetched_marker_does_not_outlive_a_write() { + let channel = channel_with_gen0(DelegateSim::default()); + let lineage = lineage3(); + let mut io = RiverSuccessorIo::new(&channel, NoImports, successor_key()); + block_on(io.prefetch_markers(&lineage)); + let pred = pred_key(0); + block_on(io.record_marker(&pred, MigrationMarker::InProgress { saw_data: true })).unwrap(); + assert_eq!( + block_on(io.read_marker(&pred)).unwrap(), + Some(MigrationMarker::InProgress { saw_data: true }) + ); + } + fn note(channel: &MockChannel, gen: usize, items: Option) -> SweepWrite { block_on(note_sweep_import_with( channel, diff --git a/ui/src/components/app/freenet_api/response_handler.rs b/ui/src/components/app/freenet_api/response_handler.rs index a193b232f..75d56ff5f 100644 --- a/ui/src/components/app/freenet_api/response_handler.rs +++ b/ui/src/components/app/freenet_api/response_handler.rs @@ -19,16 +19,16 @@ use crate::components::app::chat_delegate::{ complete_pending_public_key_request, complete_pending_request, complete_pending_sign_request, complete_pending_signing_key_request, current_delegate_source_rank, current_load_attempt, decide_legacy_migration_action, decide_per_room_load_action, enqueue_delegate_request, - fire_legacy_migration_request, hydrate_hidden_dm_threads, hydrate_outbound_dms_cache, - is_legacy_delegate_key, legacy_scoped_correlation, load_state_after_probe_legacy, - mark_fetch_failure_if_current, mark_outbound_dms_hydrated, mark_room_set_complete, - note_current_list_response, note_delegate_response_for_register_ack, parse_room_storage_key, - per_room_terminal, prune_outbound_dms_for_purges, request_outbound_dms_once, - response_correlation_base, room_storage_key, save_outbound_dms_to_delegate, - save_rooms_to_delegate, seed_saved_slots_from_load, send_delegate_request, - send_delegate_request_to, set_load_state_if_current, source_rank_for_delegate_key, - LegacyMigrationAction, LoadWorkerGuard, PendingDelegateRequest, RoomsLoadState, - OUTBOUND_DMS_STORAGE_KEY, ROOMS_META_KEY, ROOMS_STORAGE_KEY, + enqueue_delegate_request_to, fire_legacy_migration_request, hydrate_hidden_dm_threads, + hydrate_outbound_dms_cache, is_legacy_delegate_key, legacy_scoped_correlation, + load_state_after_probe_legacy, mark_fetch_failure_if_current, mark_outbound_dms_hydrated, + mark_room_set_complete, note_current_list_response, note_delegate_response_for_register_ack, + parse_room_storage_key, per_room_terminal, prune_outbound_dms_for_purges, + request_outbound_dms_once, response_correlation_base, room_storage_key, + save_outbound_dms_to_delegate, save_rooms_to_delegate, seed_saved_slots_from_load, + send_delegate_request, send_delegate_request_to, set_load_state_if_current, + source_rank_for_delegate_key, LegacyMigrationAction, LoadWorkerGuard, PendingDelegateRequest, + RoomsLoadState, OUTBOUND_DMS_STORAGE_KEY, ROOMS_META_KEY, ROOMS_STORAGE_KEY, }; use crate::components::app::notifications::mark_initial_sync_complete; use crate::components::app::sync_info::SYNC_INFO; @@ -1429,15 +1429,41 @@ async fn migrate_legacy_per_room(legacy_key: DelegateKey, keys: Vec, + )> = Vec::with_capacity(room_vks.len()); + for wave in room_vks.chunks(ROOM_LOAD_CONCURRENCY) { + let mut pending = Vec::with_capacity(wave.len()); + for vk in wave { + match enqueue_delegate_request_to( + legacy_key.clone(), + ChatDelegateRequestMsg::GetRequest { + key: ChatDelegateKey::new(room_storage_key(vk)), + }, + ) + .await + { + Ok(p) => pending.push((*vk, p)), + Err(e) => responses.push((*vk, Err(e))), + } + } + responses.extend( + futures::future::join_all( + pending + .into_iter() + .map(|(vk, p)| async move { (vk, await_delegate_response(p).await) }), + ) + .await, + ); + } + for (vk, response) in responses { + match response { Ok(ChatDelegateResponseMsg::GetResponse { value: Some(bytes), .. }) => { @@ -3445,7 +3471,7 @@ mod tests { .find("worker.set_load_state(RoomsLoadState::Migrating)") .expect("migrate_legacy_per_room must announce Migrating for a non-empty index"); let loop_pos = body - .find("for vk in room_vks") + .find("room_vks.chunks(ROOM_LOAD_CONCURRENCY)") .expect("migrate_legacy_per_room must have a per-slot fetch loop"); assert!( migrating_pos < loop_pos, @@ -3779,6 +3805,48 @@ mod tests { ); } + /// freenet/river#809: a legacy generation's room slots are fetched in + /// concurrent waves too, not one round trip per room. The first load after + /// a re-key reads every room through this path, so a serial loop cost N + /// round trips before any room showed. Same borrow-safety shape as + /// `per_room_load_fans_out_concurrently`. + #[test] + fn legacy_per_room_migration_fans_out_concurrently() { + let src = include_str!("response_handler.rs"); + let production = src + .split("mod tests {") + .next() + .expect("production code before `mod tests`"); + let start = production + .find("Migrating {} per-room slot(s) from legacy delegate") + .expect("legacy per-room fetch marker must exist"); + let end = production[start..] + .find("let meta = if has_meta") + .map(|i| start + i) + .expect("the legacy per-room loop must be followed by the meta load"); + let region = &production[start..end]; + + let enqueue_pos = region + .find("enqueue_delegate_request_to(") + .expect("legacy slot GETs must be enqueued, scoped to the legacy delegate"); + let join_pos = region + .find("join_all(") + .expect("legacy slot responses must be awaited concurrently via join_all"); + assert!(region.contains("await_delegate_response(")); + assert!( + enqueue_pos < join_pos && !region[join_pos..].contains("enqueue_delegate_request"), + "enqueues (the WEB_API borrow) must stay outside the join_all fan-out" + ); + assert!( + !region.contains("send_delegate_request_to("), + "legacy slots must not be fetched with the serial send+await" + ); + assert!( + region.contains("chunks(ROOM_LOAD_CONCURRENCY)"), + "the legacy fan-out must be bounded by ROOM_LOAD_CONCURRENCY" + ); + } + /// freenet/river#397 Codex review 4 (concurrency P2): the legacy re-save /// completion (in `hydrate_loaded_rooms`) must resolve `Loaded` only when it /// merged live rooms (`had_loaded_rooms`) — an empty legacy completion writes From 70498e6f4a877924e2788a63a8e06ba596b4d15b Mon Sep 17 00:00:00 2001 From: Ian Clarke Date: Sat, 17 Oct 2026 12:04:37 -0500 Subject: [PATCH 2/9] perf(ui): pace the legacy sweep, and prefetch markers only for a fresh journal (#809) Re-measuring the first fix made the first load slower (26 s): every delegate request on a node shares one admission queue capped at 100, the sweep sent 99 requests at once, and the extra concurrent reads pushed past the cap. The node refuses those with an error naming no delegate, so the refused room reads waited out their 10 s timeouts. The sweep now goes newest first in waves of 4 generations (12 requests); a wave is done when every request in it has a reply or an error naming its delegate, or after 2 s. Silence still concludes nothing. The marker prefetch now runs only when no journal is stored yet. With a journal stored, `begin` is off the visible path and the prefetch only competed with the room load (steady-state loads ~0.18 s slower). Claude-Session: https://claude.ai/code/session_01Mynjd42gMDNsNSxwbJWKu3 --- ui/src/components/app/chat_delegate.rs | 393 +++++++++++++----- .../app/freenet_api/delegate_migration.rs | 44 +- .../app/freenet_api/freenet_synchronizer.rs | 3 + .../app/freenet_api/response_handler.rs | 1 + 4 files changed, 340 insertions(+), 101 deletions(-) diff --git a/ui/src/components/app/chat_delegate.rs b/ui/src/components/app/chat_delegate.rs index 68408c1ef..f144377b2 100644 --- a/ui/src/components/app/chat_delegate.rs +++ b/ui/src/components/app/chat_delegate.rs @@ -4765,12 +4765,58 @@ mod tests { /// freenet/river#809: the sweep asks the NEWEST legacy generation first, so /// its answer is not queued on the node behind every older generation. #[test] - fn legacy_sweep_asks_the_newest_generation_first() { + fn legacy_sweep_asks_newest_first_in_bounded_waves() { let src = include_str!("chat_delegate.rs"); let body = &src[src .rfind("pub(crate) async fn fire_legacy_migration_request() -> bool {") .expect("fire_legacy_migration_request must exist")..]; - assert!(body.contains("in LEGACY_DELEGATES.iter().enumerate().rev() {")); + assert!(body.contains("(0..LEGACY_DELEGATES.len()).rev().collect()")); + assert!(body.contains("newest_first.chunks(SWEEP_WAVE_GENERATIONS)")); + assert!( + body.matches("expect_sweep_answer(key_bytes);").count() == 2, + "every sent request (the two GETs and the ListRequest) must be expected" + ); + assert!(body.contains("seal_sweep_wave()")); + // Three requests a generation: a wave stays far below the node's + // shared 100-request delegate queue. + assert!(3 * SWEEP_WAVE_GENERATIONS <= 16); + } + + /// A sweep wave completes once every request it sent is answered, by a + /// reply or by an error naming its delegate; an error naming no delegate + /// (the node's queue refusal) answers nothing, so the wave falls back to + /// its wait bound. + #[test] + fn a_sweep_wave_completes_when_every_request_is_answered() { + let a = [1u8; 32]; + let b = [2u8; 32]; + let mut wave = SweepWave::default(); + for _ in 0..3 { + wave.expect(&a); + } + wave.expect(&b); + let mut done = wave.seal().expect("requests are outstanding"); + wave.answered(&a); + wave.errored(&format!( + "missing delegate {}", + bs58::encode(a).into_string() + )); + wave.errored("no response received from handler"); + wave.answered(&[9u8; 32]); + assert!( + matches!(done.try_recv(), Ok(None)), + "a and b still outstanding" + ); + wave.errored(&format!( + "delegate {} is rate limited after repeated failures; retry in 80 ms", + bs58::encode(a).into_string() + )); + assert!(matches!(done.try_recv(), Ok(None)), "b still outstanding"); + wave.answered(&b); + assert!(matches!(done.try_recv(), Ok(Some(())))); + + let mut empty = SweepWave::default(); + assert!(empty.seal().is_none(), "nothing sent: nothing to wait for"); } // ===== resolve_hidden_thread_hydration tests (#261 Codex P3) ===== @@ -11255,6 +11301,132 @@ pub(crate) fn retry_rooms_load() { /// legacy delegate request, creating a tight error spam loop. static LEGACY_MIGRATION_ATTEMPTED: AtomicBool = AtomicBool::new(false); +/// Legacy generations the startup sweep asks at once (freenet/river#809). +/// +/// Every delegate request on a node, from every client and for every +/// delegate, waits in ONE admission queue per priority tier, capped at 100 +/// (`MAX_QUEUED_PER_CONTRACT`, freenet-core `fair_queue.rs`), and one past the +/// cap is refused with an error that names no delegate, so its waiter sits +/// out its whole timeout. The sweep sends three requests per generation, so +/// all 33 at once is 99: in the rehearsal that alone, plus the load's other +/// requests, got the newest generation's room reads refused and held the room +/// list for 26 s. Four generations is 12 requests a wave. +const SWEEP_WAVE_GENERATIONS: usize = 4; +const _: () = assert!(SWEEP_WAVE_GENERATIONS > 0); + +/// The longest a sweep wave waits for its answers before the next one goes +/// out. A warm node answers a wave in one round trip; a cold one loads each +/// delegate module first (about 0.4 s each in the rehearsal). +const SWEEP_WAVE_WAIT_MS: u64 = 2_000; + +/// The requests of the sweep wave in flight that have no answer yet: per +/// legacy delegate, how many were sent and not answered, with the key's +/// base58 form (what the node's error texts name). The wave is complete once +/// sealed with nothing outstanding. +#[derive(Default)] +pub(crate) struct SweepWave { + outstanding: HashMap<[u8; 32], (u32, String)>, + sealed: bool, + done: Option>, +} + +impl SweepWave { + pub(crate) fn expect(&mut self, key: &[u8; 32]) { + self.outstanding + .entry(*key) + .or_insert_with(|| (0, bs58::encode(key).into_string())) + .0 += 1; + } + + /// One request to `key` was answered: a reply, or an error naming it. A + /// key outside the wave (a late answer to an earlier wave, or another + /// path's request to the same generation) is ignored; one counted early + /// only lets the next wave go out sooner. + pub(crate) fn answered(&mut self, key: &[u8]) { + if let Ok(key) = <[u8; 32]>::try_from(key) { + if let Some((n, _)) = self.outstanding.get_mut(&key) { + *n = n.saturating_sub(1); + if *n == 0 { + self.outstanding.remove(&key); + } + } + } + self.finish_if_complete(); + } + + /// An error from the node: it answers the request to the wave key its + /// text names (typed `Missing`, the rate limiter's refusal, core's "not + /// found in store"). An error naming no key answers nothing. + pub(crate) fn errored(&mut self, text: &str) { + let named = self + .outstanding + .iter() + .find(|(_, (_, b58))| text.contains(b58.as_str())) + .map(|(k, _)| *k); + if let Some(key) = named { + self.answered(&key); + } + } + + /// Every request is sent: from now on the wave completes when nothing is + /// outstanding. `None` when it already is. + pub(crate) fn seal(&mut self) -> Option> { + self.sealed = true; + if self.outstanding.is_empty() { + return None; + } + let (tx, rx) = oneshot::channel(); + self.done = Some(tx); + Some(rx) + } + + fn finish_if_complete(&mut self) { + if self.sealed && self.outstanding.is_empty() { + if let Some(done) = self.done.take() { + let _ = done.send(()); + } + } + } +} + +static SWEEP_WAVE: LazyLock>> = LazyLock::new(|| Mutex::new(None)); + +fn arm_sweep_wave() { + *SWEEP_WAVE.lock().unwrap() = Some(SweepWave::default()); +} + +fn expect_sweep_answer(key: &[u8; 32]) { + if let Some(wave) = SWEEP_WAVE.lock().unwrap().as_mut() { + wave.expect(key); + } +} + +fn seal_sweep_wave() -> Option> { + SWEEP_WAVE + .lock() + .unwrap() + .as_mut() + .and_then(SweepWave::seal) +} + +fn disarm_sweep_wave() { + *SWEEP_WAVE.lock().unwrap() = None; +} + +/// A legacy delegate replied (called for every legacy `DelegateResponse`). +pub(crate) fn note_sweep_answer(key: &[u8]) { + if let Some(wave) = SWEEP_WAVE.lock().unwrap().as_mut() { + wave.answered(key); + } +} + +/// The node answered some request with an error (called for every one). +pub(crate) fn note_sweep_error(text: &str) { + if let Some(wave) = SWEEP_WAVE.lock().unwrap().as_mut() { + wave.errored(text); + } +} + /// Per-session guard so interrupted-migration recovery re-fires the legacy /// probe at most ONCE per session. Without it, every WebSocket reconnect that /// re-runs the per-room load would reset [`LEGACY_MIGRATION_ATTEMPTED`] and @@ -11326,126 +11498,147 @@ pub(crate) async fn fire_legacy_migration_request() -> bool { let attempt = LOAD_ATTEMPT_GEN.load(Ordering::Relaxed); let mut any_dispatched = false; - // Newest generation first (freenet/river#809). The node runs delegate - // requests one at a time, in arrival order, so oldest-first put the - // generation most likely to hold the user's rooms behind every older one, - // each paying a module load on a cold node (0.4 s each in the rehearsal). - // Arrival order decides nothing else: every merge is ranked by source - // generation, never by arrival (freenet/river#527). - for (i, (key_bytes, code_hash_bytes)) in LEGACY_DELEGATES.iter().enumerate().rev() { - let legacy_code_hash = CodeHash::new(*code_hash_bytes); - let legacy_delegate_key = DelegateKey::new(*key_bytes, legacy_code_hash); - - // Send a GetRequest for each storage key we want to migrate. - // rooms_data is the original and gates the migration; outbound_dms - // (#256) is mirrored so a delegate rebuild doesn't orphan the - // sender's own DM plaintext. - let storage_keys: [&[u8]; 2] = [ROOMS_STORAGE_KEY, OUTBOUND_DMS_STORAGE_KEY]; - for storage_key in storage_keys { - let request = ChatDelegateRequestMsg::GetRequest { - key: ChatDelegateKey::new(storage_key.to_vec()), - }; + // Newest generation first, in waves (freenet/river#809). The node runs + // delegate requests one at a time, in arrival order, so oldest-first put + // the generation most likely to hold the user's rooms behind every older + // one. And all of them at once (three requests each, 99 for 33 + // generations) nearly fills the ONE admission queue every delegate request + // on the node shares (see `SWEEP_WAVE_GENERATIONS`); a request the node + // turns away there comes back with no key, so its waiter sits out its + // whole timeout. Arrival order decides nothing else: every merge is ranked + // by source generation, never by arrival (freenet/river#527). + let newest_first: Vec = (0..LEGACY_DELEGATES.len()).rev().collect(); + for wave in newest_first.chunks(SWEEP_WAVE_GENERATIONS) { + arm_sweep_wave(); + for &i in wave { + let (key_bytes, code_hash_bytes) = &LEGACY_DELEGATES[i]; + let legacy_code_hash = CodeHash::new(*code_hash_bytes); + let legacy_delegate_key = DelegateKey::new(*key_bytes, legacy_code_hash); + + // Send a GetRequest for each storage key we want to migrate. + // rooms_data is the original and gates the migration; outbound_dms + // (#256) is mirrored so a delegate rebuild doesn't orphan the + // sender's own DM plaintext. + let storage_keys: [&[u8]; 2] = [ROOMS_STORAGE_KEY, OUTBOUND_DMS_STORAGE_KEY]; + for storage_key in storage_keys { + let request = ChatDelegateRequestMsg::GetRequest { + key: ChatDelegateKey::new(storage_key.to_vec()), + }; - let mut payload = Vec::new(); - if let Err(e) = ciborium::ser::into_writer(&request, &mut payload) { - error!( - "Failed to serialize legacy migration request #{} for key {:?}: {}", - i, - String::from_utf8_lossy(storage_key), - e - ); - continue; - } + let mut payload = Vec::new(); + if let Err(e) = ciborium::ser::into_writer(&request, &mut payload) { + error!( + "Failed to serialize legacy migration request #{} for key {:?}: {}", + i, + String::from_utf8_lossy(storage_key), + e + ); + continue; + } + + let app_msg = freenet_stdlib::prelude::ApplicationMessage::new(payload); + + let delegate_request = DelegateOp(DelegateRequest::ApplicationMessages { + key: legacy_delegate_key.clone(), + params: Parameters::from(Vec::::new()), + inbound: vec![ + freenet_stdlib::prelude::InboundDelegateMsg::ApplicationMessage(app_msg), + ], + }); + + let api_result = { + let mut web_api = WEB_API.write(); + if let Some(api) = web_api.as_mut() { + api.send(delegate_request).await + } else { + Err(freenet_stdlib::client_api::Error::ConnectionClosed) + } + }; - let app_msg = freenet_stdlib::prelude::ApplicationMessage::new(payload); + match api_result { + Ok(_) => { + any_dispatched = true; + expect_sweep_answer(key_bytes); + info!( + "Legacy migration request #{} sent for key {:?}", + i, + String::from_utf8_lossy(storage_key) + ); + } + Err(e) => { + // A send that RETURNS Err is a transport failure (connection + // closed), NOT "delegate not present". Mark it (P1) — but only + // for the current attempt (P2#1), so a stale probe can't + // pollute a reconnect's fresh load. + info!( + "Legacy migration request #{} for key {:?} failed to send \ + (transport): {}", + i, + String::from_utf8_lossy(storage_key), + e + ); + mark_fetch_failure_if_current(attempt); + } + } + } - let delegate_request = DelegateOp(DelegateRequest::ApplicationMessages { + // Per-room keys (room:, rooms_meta) are DYNAMIC, so the fixed + // GetRequests above can't discover them. Probe the legacy delegate's key + // index with a ListRequest; the ListResponse (is_legacy) is routed in + // response_handler.rs to a per-room migration that reconstructs those + // slots from THIS legacy delegate and re-saves them to the current one + // (freenet/river#345 / #65). Without this, the NEXT delegate-WASM bump + // after per-room storage ships would strand every room under the old key. + // Fire-and-forget (not awaited): a ListResponse only arrives if the + // legacy delegate is present, so an absent delegate costs nothing rather + // than a 10s per-delegate timeout for fresh users. + let list_request = ChatDelegateRequestMsg::ListRequest; + let mut list_payload = Vec::new(); + if let Err(e) = ciborium::ser::into_writer(&list_request, &mut list_payload) { + error!("Failed to serialize legacy ListRequest #{i}: {e}"); + continue; + } + let list_app_msg = freenet_stdlib::prelude::ApplicationMessage::new(list_payload); + let list_delegate_request = DelegateOp(DelegateRequest::ApplicationMessages { key: legacy_delegate_key.clone(), params: Parameters::from(Vec::::new()), inbound: vec![ - freenet_stdlib::prelude::InboundDelegateMsg::ApplicationMessage(app_msg), + freenet_stdlib::prelude::InboundDelegateMsg::ApplicationMessage(list_app_msg), ], }); - - let api_result = { + let list_result = { let mut web_api = WEB_API.write(); if let Some(api) = web_api.as_mut() { - api.send(delegate_request).await + api.send(list_delegate_request).await } else { Err(freenet_stdlib::client_api::Error::ConnectionClosed) } }; - - match api_result { + match list_result { Ok(_) => { any_dispatched = true; - info!( - "Legacy migration request #{} sent for key {:?}", - i, - String::from_utf8_lossy(storage_key) - ); + expect_sweep_answer(key_bytes); + info!("Legacy ListRequest #{i} sent for per-room key discovery"); } Err(e) => { - // A send that RETURNS Err is a transport failure (connection - // closed), NOT "delegate not present". Mark it (P1) — but only - // for the current attempt (P2#1), so a stale probe can't - // pollute a reconnect's fresh load. - info!( - "Legacy migration request #{} for key {:?} failed to send \ - (transport): {}", - i, - String::from_utf8_lossy(storage_key), - e - ); + // Transport failure on the ListRequest send → mark it (P1), + // current-attempt only (P2#1) via the uniform helper (review 11). + info!("Legacy ListRequest #{i} failed to send (transport): {e}"); mark_fetch_failure_if_current(attempt); } } } - - // Per-room keys (room:, rooms_meta) are DYNAMIC, so the fixed - // GetRequests above can't discover them. Probe the legacy delegate's key - // index with a ListRequest; the ListResponse (is_legacy) is routed in - // response_handler.rs to a per-room migration that reconstructs those - // slots from THIS legacy delegate and re-saves them to the current one - // (freenet/river#345 / #65). Without this, the NEXT delegate-WASM bump - // after per-room storage ships would strand every room under the old key. - // Fire-and-forget (not awaited): a ListResponse only arrives if the - // legacy delegate is present, so an absent delegate costs nothing rather - // than a 10s per-delegate timeout for fresh users. - let list_request = ChatDelegateRequestMsg::ListRequest; - let mut list_payload = Vec::new(); - if let Err(e) = ciborium::ser::into_writer(&list_request, &mut list_payload) { - error!("Failed to serialize legacy ListRequest #{i}: {e}"); - continue; - } - let list_app_msg = freenet_stdlib::prelude::ApplicationMessage::new(list_payload); - let list_delegate_request = DelegateOp(DelegateRequest::ApplicationMessages { - key: legacy_delegate_key.clone(), - params: Parameters::from(Vec::::new()), - inbound: vec![ - freenet_stdlib::prelude::InboundDelegateMsg::ApplicationMessage(list_app_msg), - ], - }); - let list_result = { - let mut web_api = WEB_API.write(); - if let Some(api) = web_api.as_mut() { - api.send(list_delegate_request).await - } else { - Err(freenet_stdlib::client_api::Error::ConnectionClosed) - } - }; - match list_result { - Ok(_) => { - any_dispatched = true; - info!("Legacy ListRequest #{i} sent for per-room key discovery"); - } - Err(e) => { - // Transport failure on the ListRequest send → mark it (P1), - // current-attempt only (P2#1) via the uniform helper (review 11). - info!("Legacy ListRequest #{i} failed to send (transport): {e}"); - mark_fetch_failure_if_current(attempt); - } - } + // The next wave goes out once every request in this one is answered, + // or after `SWEEP_WAVE_WAIT_MS` if some stay silent. Silence concludes + // nothing here: the sweep never treated a missing answer as absence, + // and the journaled walk asks again. + if let Some(answered) = seal_sweep_wave() { + let timeout = Box::pin(crate::util::sleep(std::time::Duration::from_millis( + SWEEP_WAVE_WAIT_MS, + ))); + let _ = select(answered, timeout).await; + } + disarm_sweep_wave(); } // freenet/river#397 Codex review 9 (P1): if NO probe was actually dispatched diff --git a/ui/src/components/app/freenet_api/delegate_migration.rs b/ui/src/components/app/freenet_api/delegate_migration.rs index 39751c210..548ed5db7 100644 --- a/ui/src/components/app/freenet_api/delegate_migration.rs +++ b/ui/src/components/app/freenet_api/delegate_migration.rs @@ -1496,6 +1496,8 @@ impl<'a, C: RiverDelegateChannel, S: RiverImportSeam> RiverSuccessorIo<'a, C, S> /// two reads per generation, 66 round trips for 33 generations, which held /// the first load after a re-key for about 8 s at 120 ms RTT). /// + /// Called only before the first `begin` on a successor ([`begin_with`]). + /// /// Only definitive answers are kept: a marker that is silent or errs here /// is read again, one at a time, when `begin` asks for it, and fails there /// exactly as it would have without the prefetch. Silence is still never @@ -2028,8 +2030,21 @@ pub(crate) async fn begin_with( successor: DelegateKey, lineage: &[DelegateLineageEntry], ) -> Result { + // Only a journal not yet stored (the first load after a re-key) reads + // every marker, and that `begin` holds the sweep's first merge, so only it + // is worth the prefetch. With a journal stored, `begin` runs off the + // visible path and reads few markers; the prefetch there only competes + // with the room load (about 0.2 s later in the rehearsal). A journal read + // that is silent or errs skips the prefetch; `begin` meets the same + // failure itself. + let fresh = matches!( + get_current(channel, &successor, MIGRATION_JOURNAL_KEY.to_vec()).await, + Ok(None) + ); let mut writer = RiverSuccessorIo::new(channel, NoImports, successor); - writer.prefetch_markers(lineage).await; + if fresh { + writer.prefetch_markers(lineage).await; + } begin_migration_journal(&mut writer, lineage, &river_secret_policy()).await } @@ -5059,6 +5074,33 @@ mod tests { ); } + /// With a journal stored, `begin` is off the visible path and reads few + /// markers, so it does not prefetch them all. + #[test] + fn begin_prefetches_only_before_the_journal_exists() { + let (lineage, channel) = lineage_n(5); + block_on(begin_with(&channel, successor_key(), &lineage)).unwrap(); + let marker_reads = |c: &MockChannel| { + c.log + .borrow() + .iter() + .filter(|(_, what)| what.starts_with("get:__migrate_pred_")) + .count() + }; + assert_eq!( + marker_reads(&channel), + 10, + "first begin: every marker, prefetched" + ); + channel.reset_counters(); + block_on(begin_with(&channel, successor_key(), &lineage)).unwrap(); + assert_eq!( + marker_reads(&channel), + 0, + "nothing resolved yet, so nothing to re-read" + ); + } + /// The prefetch never turns silence into absence (constraint 1): a marker /// the prefetch could not read is read again live, and a marker that stays /// silent still refuses `begin`, exactly as without the prefetch. diff --git a/ui/src/components/app/freenet_api/freenet_synchronizer.rs b/ui/src/components/app/freenet_api/freenet_synchronizer.rs index 860fc020c..c443c56f8 100644 --- a/ui/src/components/app/freenet_api/freenet_synchronizer.rs +++ b/ui/src/components/app/freenet_api/freenet_synchronizer.rs @@ -860,6 +860,9 @@ impl FreenetSynchronizer { } Err(e) => { error!("Received error in API response: {}", e); + crate::components::app::chat_delegate::note_sweep_error( + &e.to_string(), + ); let current = current_delegate_key(); if let Some(key) = e.register_failed_key() { diff --git a/ui/src/components/app/freenet_api/response_handler.rs b/ui/src/components/app/freenet_api/response_handler.rs index 75d56ff5f..3f435a4de 100644 --- a/ui/src/components/app/freenet_api/response_handler.rs +++ b/ui/src/components/app/freenet_api/response_handler.rs @@ -157,6 +157,7 @@ impl ResponseHandler { let responding_delegate = key.clone(); if is_legacy_delegate { info!("Received response from LEGACY delegate - checking for migration data"); + crate::components::app::chat_delegate::note_sweep_answer(key.bytes()); } // Log a short key fingerprint, not the full `DelegateKey` Debug: that From fc2d5ab7193a0f977dbcd3c1dd059dead2c72a25 Mon Sep 17 00:00:00 2001 From: Ian Clarke Date: Sat, 17 Oct 2026 12:30:13 -0500 Subject: [PATCH 3/9] fix(ui): address #810 review: sweep returns after its first wave, bounded fan-outs (#809) - The loader wrote `Loading` after the awaited sweep, overwriting the `Loaded` its first wave's merge had already written (identity import then stayed off until the 60 s backstop). The sweep now sends its first wave and returns, as the whole sweep used to; later waves run from a background task that holds a load worker. - Sweep waves are tracked per sweep (a Retry or recovery re-fire no longer overwrites another sweep's wave), and only the sweep's own replies (`rooms_data`/`outbound_dms` GETs, `ListRequest`, or an empty reply) count toward one; per-room slot replies no longer end a wave early. - Legacy slot fan-outs run one generation at a time and drop duplicate keys; a compile-time check keeps one client's first-load burst (sweep wave + slot wave + marker wave) at half the node's shared delegate queue. - A transport failure mid-sweep lets the next load attempt sweep again. - `begin_with`'s journal read is handed to `begin`'s first journal load: no extra round trip, and `begin` sees the journal the markers were read with. - Tests: the wave plan, sweep-reply classification, and the node's real typed `Missing` and rate-limit error texts answering a wave. Stale comments on oldest-first dispatch updated. Claude-Session: https://claude.ai/code/session_01Mynjd42gMDNsNSxwbJWKu3 --- ui/src/components/app/chat_delegate.rs | 491 ++++++++++++------ .../app/freenet_api/delegate_migration.rs | 42 +- .../app/freenet_api/response_handler.rs | 44 +- ui/src/room_data.rs | 16 +- 4 files changed, 405 insertions(+), 188 deletions(-) diff --git a/ui/src/components/app/chat_delegate.rs b/ui/src/components/app/chat_delegate.rs index f144377b2..2545b6c68 100644 --- a/ui/src/components/app/chat_delegate.rs +++ b/ui/src/components/app/chat_delegate.rs @@ -296,15 +296,17 @@ pub static LOAD_WORKERS_CHANGED: GlobalSignal = Global::new(|| 0); // DELEGATE-GENERATION AUTHORITY (freenet/river#527) // ============================================================================= // -// `fire_legacy_migration_request` probes EVERY legacy delegate generation at -// once, and each one that still holds data hydrates independently. Their -// responses race, so something has to decide which copy of a room is -// authoritative when two generations disagree about the user's identity for it. +// `fire_legacy_migration_request` probes EVERY legacy delegate generation (in +// waves since freenet/river#809), and each one that still holds data hydrates +// independently. Their responses race, so something has to decide which copy +// of a room is authoritative when two generations disagree about the user's +// identity for it. // // Before #527 nothing did: `Rooms::merge` kept whichever copy landed FIRST and -// silently skipped the rest. Probes are dispatched oldest-generation-first (the -// `LEGACY_DELEGATES` order), so the oldest surviving copy usually answered -// first — rolling a long-time user back to an identity from months earlier. +// silently skipped the rest. Probes were then dispatched oldest-generation-first +// (the `LEGACY_DELEGATES` order; newest-first since #809), so the oldest +// surviving copy usually answered first — rolling a long-time user back to an +// identity from months earlier. // // The rank below is that missing order. `LEGACY_DELEGATES` is generated from // `legacy_delegates.toml` in age order (V1 oldest first), so an entry's INDEX @@ -4770,16 +4772,84 @@ mod tests { let body = &src[src .rfind("pub(crate) async fn fire_legacy_migration_request() -> bool {") .expect("fire_legacy_migration_request must exist")..]; - assert!(body.contains("(0..LEGACY_DELEGATES.len()).rev().collect()")); - assert!(body.contains("newest_first.chunks(SWEEP_WAVE_GENERATIONS)")); + assert!(body.contains("sweep_wave_plan(LEGACY_DELEGATES.len())")); assert!( - body.matches("expect_sweep_answer(key_bytes);").count() == 2, + body.matches("expect_sweep_answer(wave, key_bytes);") + .count() + == 2, "every sent request (the two GETs and the ListRequest) must be expected" ); - assert!(body.contains("seal_sweep_wave()")); - // Three requests a generation: a wave stays far below the node's - // shared 100-request delegate queue. - assert!(3 * SWEEP_WAVE_GENERATIONS <= 16); + assert!(body.contains("await_sweep_wave(wave).await")); + } + + /// The plan: newest first, every generation exactly once, no wave larger + /// than `SWEEP_WAVE_GENERATIONS`. + #[test] + fn the_sweep_plan_is_newest_first_bounded_and_complete() { + for n in [0usize, 1, 3, 4, 5, 33, 64] { + let plan = sweep_wave_plan(n); + let flat: Vec = plan.iter().flatten().copied().collect(); + assert_eq!(flat, (0..n).rev().collect::>(), "n = {n}"); + assert!(plan + .iter() + .all(|w| !w.is_empty() && w.len() <= SWEEP_WAVE_GENERATIONS)); + } + } + + /// Only the sweep's own requests pace it: a per-room slot or `rooms_meta` + /// reply from the same generation must not end its wave. + #[test] + fn only_sweep_replies_count_toward_a_wave() { + let get = |k: &[u8]| ChatDelegateResponseMsg::GetResponse { + key: ChatDelegateKey::new(k.to_vec()), + value: None, + }; + assert!(is_sweep_reply(&get(ROOMS_STORAGE_KEY))); + assert!(is_sweep_reply(&get(OUTBOUND_DMS_STORAGE_KEY))); + assert!(is_sweep_reply(&ChatDelegateResponseMsg::ListResponse { + keys: vec![] + })); + assert!(!is_sweep_reply(&get(ROOMS_META_KEY))); + assert!(!is_sweep_reply(&get(&room_storage_key( + &SigningKey::from_bytes(&[3u8; 32]).verifying_key() + )))); + } + + /// The node's real error texts name the delegate in the form the wave + /// matches: typed `Missing` as the synchronizer stringifies it. + #[test] + fn a_typed_missing_error_answers_its_sweep_request() { + use freenet_stdlib::client_api::{ClientError, DelegateError, ErrorKind, RequestError}; + let key_bytes = [7u8; 32]; + let key = DelegateKey::new(key_bytes, CodeHash::new([8u8; 32])); + let err: ClientError = ErrorKind::RequestError(RequestError::DelegateError( + DelegateError::Missing(key.clone()), + )) + .into(); + let text = + crate::components::app::freenet_api::error::SynchronizerError::from_api_error(&err) + .to_string(); + let mut wave = SweepWave::default(); + wave.expect(&key_bytes); + wave.expect(&key_bytes); + let mut done = wave.seal().unwrap(); + wave.errored(&text); + // The rate limiter's refusal of the next request to the same key, as + // freenet-core words it (`client_events/websocket.rs`). + let refused: ClientError = + ErrorKind::RequestError(RequestError::DelegateError(DelegateError::ExecutionError( + format!("delegate {key} is rate limited after repeated failures; retry in 80 ms") + .into(), + ))) + .into(); + assert!(matches!(done.try_recv(), Ok(None)), "error text: {text}"); + wave.errored( + &crate::components::app::freenet_api::error::SynchronizerError::from_api_error( + &refused, + ) + .to_string(), + ); + assert!(matches!(done.try_recv(), Ok(Some(())))); } /// A sweep wave completes once every request it sent is answered, by a @@ -11311,7 +11381,7 @@ static LEGACY_MIGRATION_ATTEMPTED: AtomicBool = AtomicBool::new(false); /// all 33 at once is 99: in the rehearsal that alone, plus the load's other /// requests, got the newest generation's room reads refused and held the room /// list for 26 s. Four generations is 12 requests a wave. -const SWEEP_WAVE_GENERATIONS: usize = 4; +pub(crate) const SWEEP_WAVE_GENERATIONS: usize = 4; const _: () = assert!(SWEEP_WAVE_GENERATIONS > 0); /// The longest a sweep wave waits for its answers before the next one goes @@ -11389,40 +11459,66 @@ impl SweepWave { } } -static SWEEP_WAVE: LazyLock>> = LazyLock::new(|| Mutex::new(None)); +/// The sweep waves in flight, by token. Usually one; two when a recovery or a +/// Retry starts a sweep while an earlier one is still pacing, and each then +/// keeps its own wave. An answer counts toward every wave expecting its key. +static SWEEP_WAVES: LazyLock>> = + LazyLock::new(|| Mutex::new(HashMap::new())); +static NEXT_SWEEP_WAVE: AtomicU64 = AtomicU64::new(0); -fn arm_sweep_wave() { - *SWEEP_WAVE.lock().unwrap() = Some(SweepWave::default()); +fn arm_sweep_wave() -> u64 { + let wave = NEXT_SWEEP_WAVE.fetch_add(1, Ordering::Relaxed); + SWEEP_WAVES + .lock() + .unwrap() + .insert(wave, SweepWave::default()); + wave } -fn expect_sweep_answer(key: &[u8; 32]) { - if let Some(wave) = SWEEP_WAVE.lock().unwrap().as_mut() { - wave.expect(key); +fn expect_sweep_answer(wave: u64, key: &[u8; 32]) { + if let Some(w) = SWEEP_WAVES.lock().unwrap().get_mut(&wave) { + w.expect(key); } } -fn seal_sweep_wave() -> Option> { - SWEEP_WAVE +fn seal_sweep_wave(wave: u64) -> Option> { + SWEEP_WAVES .lock() .unwrap() - .as_mut() + .get_mut(&wave) .and_then(SweepWave::seal) } -fn disarm_sweep_wave() { - *SWEEP_WAVE.lock().unwrap() = None; +fn disarm_sweep_wave(wave: u64) { + SWEEP_WAVES.lock().unwrap().remove(&wave); +} + +/// Whether a legacy delegate's reply answers a sweep request: the +/// `rooms_data` or `outbound_dms` GET, or the `ListRequest`. A per-room slot +/// or `rooms_meta` reply (the migration that follows a listing) is not one, +/// and counting it would end a wave while its own requests are still queued. +pub(crate) fn is_sweep_reply(response: &ChatDelegateResponseMsg) -> bool { + match response { + ChatDelegateResponseMsg::ListResponse { .. } => true, + ChatDelegateResponseMsg::GetResponse { key, .. } => { + key.as_bytes() == ROOMS_STORAGE_KEY || key.as_bytes() == OUTBOUND_DMS_STORAGE_KEY + } + _ => false, + } } -/// A legacy delegate replied (called for every legacy `DelegateResponse`). +/// A legacy delegate answered a sweep request (see [`is_sweep_reply`]), or +/// replied with nothing at all (core 0.2.136 and earlier, for a delegate it +/// does not have). pub(crate) fn note_sweep_answer(key: &[u8]) { - if let Some(wave) = SWEEP_WAVE.lock().unwrap().as_mut() { + for wave in SWEEP_WAVES.lock().unwrap().values_mut() { wave.answered(key); } } /// The node answered some request with an error (called for every one). pub(crate) fn note_sweep_error(text: &str) { - if let Some(wave) = SWEEP_WAVE.lock().unwrap().as_mut() { + for wave in SWEEP_WAVES.lock().unwrap().values_mut() { wave.errored(text); } } @@ -11453,8 +11549,11 @@ pub(crate) fn arm_legacy_migration_recovery() -> bool { true } -/// Fire requests to load rooms from all known legacy delegates (fire and forget). -/// If any legacy delegate has room data, the response handler will migrate it. +/// Fire requests to load rooms from all known legacy delegates, newest +/// generation first, in waves (freenet/river#809). Only the first wave is sent +/// before this returns; the rest follow from a background task, each once the +/// one before is answered. The requests are fire-and-forget: if any legacy +/// delegate has room data, the response handler will migrate it. /// /// **Must only be called once the current delegate has confirmed it has no /// rooms_data** (see freenet/river#253). Firing this while the current delegate @@ -11496,8 +11595,6 @@ pub(crate) async fn fire_legacy_migration_request() -> bool { // still-current, so a STALE probe's late send-failure can't pollute the new // attempt's `SAW_FETCH_FAILURE` into a false LoadFailed. let attempt = LOAD_ATTEMPT_GEN.load(Ordering::Relaxed); - let mut any_dispatched = false; - // Newest generation first, in waves (freenet/river#809). The node runs // delegate requests one at a time, in arrival order, so oldest-first put // the generation most likely to hold the user's rooms behind every older @@ -11507,138 +11604,22 @@ pub(crate) async fn fire_legacy_migration_request() -> bool { // turns away there comes back with no key, so its waiter sits out its // whole timeout. Arrival order decides nothing else: every merge is ranked // by source generation, never by arrival (freenet/river#527). - let newest_first: Vec = (0..LEGACY_DELEGATES.len()).rev().collect(); - for wave in newest_first.chunks(SWEEP_WAVE_GENERATIONS) { - arm_sweep_wave(); - for &i in wave { - let (key_bytes, code_hash_bytes) = &LEGACY_DELEGATES[i]; - let legacy_code_hash = CodeHash::new(*code_hash_bytes); - let legacy_delegate_key = DelegateKey::new(*key_bytes, legacy_code_hash); - - // Send a GetRequest for each storage key we want to migrate. - // rooms_data is the original and gates the migration; outbound_dms - // (#256) is mirrored so a delegate rebuild doesn't orphan the - // sender's own DM plaintext. - let storage_keys: [&[u8]; 2] = [ROOMS_STORAGE_KEY, OUTBOUND_DMS_STORAGE_KEY]; - for storage_key in storage_keys { - let request = ChatDelegateRequestMsg::GetRequest { - key: ChatDelegateKey::new(storage_key.to_vec()), - }; - - let mut payload = Vec::new(); - if let Err(e) = ciborium::ser::into_writer(&request, &mut payload) { - error!( - "Failed to serialize legacy migration request #{} for key {:?}: {}", - i, - String::from_utf8_lossy(storage_key), - e - ); - continue; - } - - let app_msg = freenet_stdlib::prelude::ApplicationMessage::new(payload); - - let delegate_request = DelegateOp(DelegateRequest::ApplicationMessages { - key: legacy_delegate_key.clone(), - params: Parameters::from(Vec::::new()), - inbound: vec![ - freenet_stdlib::prelude::InboundDelegateMsg::ApplicationMessage(app_msg), - ], - }); - - let api_result = { - let mut web_api = WEB_API.write(); - if let Some(api) = web_api.as_mut() { - api.send(delegate_request).await - } else { - Err(freenet_stdlib::client_api::Error::ConnectionClosed) - } - }; - - match api_result { - Ok(_) => { - any_dispatched = true; - expect_sweep_answer(key_bytes); - info!( - "Legacy migration request #{} sent for key {:?}", - i, - String::from_utf8_lossy(storage_key) - ); - } - Err(e) => { - // A send that RETURNS Err is a transport failure (connection - // closed), NOT "delegate not present". Mark it (P1) — but only - // for the current attempt (P2#1), so a stale probe can't - // pollute a reconnect's fresh load. - info!( - "Legacy migration request #{} for key {:?} failed to send \ - (transport): {}", - i, - String::from_utf8_lossy(storage_key), - e - ); - mark_fetch_failure_if_current(attempt); - } - } - } - - // Per-room keys (room:, rooms_meta) are DYNAMIC, so the fixed - // GetRequests above can't discover them. Probe the legacy delegate's key - // index with a ListRequest; the ListResponse (is_legacy) is routed in - // response_handler.rs to a per-room migration that reconstructs those - // slots from THIS legacy delegate and re-saves them to the current one - // (freenet/river#345 / #65). Without this, the NEXT delegate-WASM bump - // after per-room storage ships would strand every room under the old key. - // Fire-and-forget (not awaited): a ListResponse only arrives if the - // legacy delegate is present, so an absent delegate costs nothing rather - // than a 10s per-delegate timeout for fresh users. - let list_request = ChatDelegateRequestMsg::ListRequest; - let mut list_payload = Vec::new(); - if let Err(e) = ciborium::ser::into_writer(&list_request, &mut list_payload) { - error!("Failed to serialize legacy ListRequest #{i}: {e}"); - continue; - } - let list_app_msg = freenet_stdlib::prelude::ApplicationMessage::new(list_payload); - let list_delegate_request = DelegateOp(DelegateRequest::ApplicationMessages { - key: legacy_delegate_key.clone(), - params: Parameters::from(Vec::::new()), - inbound: vec![ - freenet_stdlib::prelude::InboundDelegateMsg::ApplicationMessage(list_app_msg), - ], - }); - let list_result = { - let mut web_api = WEB_API.write(); - if let Some(api) = web_api.as_mut() { - api.send(list_delegate_request).await - } else { - Err(freenet_stdlib::client_api::Error::ConnectionClosed) - } - }; - match list_result { - Ok(_) => { - any_dispatched = true; - expect_sweep_answer(key_bytes); - info!("Legacy ListRequest #{i} sent for per-room key discovery"); - } - Err(e) => { - // Transport failure on the ListRequest send → mark it (P1), - // current-attempt only (P2#1) via the uniform helper (review 11). - info!("Legacy ListRequest #{i} failed to send (transport): {e}"); - mark_fetch_failure_if_current(attempt); - } - } - } - // The next wave goes out once every request in this one is answered, - // or after `SWEEP_WAVE_WAIT_MS` if some stay silent. Silence concludes - // nothing here: the sweep never treated a missing answer as absence, - // and the journaled walk asks again. - if let Some(answered) = seal_sweep_wave() { - let timeout = Box::pin(crate::util::sleep(std::time::Duration::from_millis( - SWEEP_WAVE_WAIT_MS, - ))); - let _ = select(answered, timeout).await; + // + // Only the first wave is sent before returning, as the whole sweep used to + // be, so the caller's load-state write still lands before any reply. The + // rest go out from a background task holding a load worker, so the load + // stays `Loading` (not "No rooms yet") while older generations may still + // answer. + let mut waves = sweep_wave_plan(LEGACY_DELEGATES.len()).into_iter(); + let Some(first) = waves.next() else { + return true; + }; + let wave = arm_sweep_wave(); + let mut any_dispatched = false; + for &i in &first { + if send_sweep_generation(i, attempt, wave).await.dispatched { + any_dispatched = true; } - disarm_sweep_wave(); } // freenet/river#397 Codex review 9 (P1): if NO probe was actually dispatched @@ -11650,8 +11631,34 @@ pub(crate) async fn fire_legacy_migration_request() -> bool { // the new attempt (whose own probe already ran / set it). if !any_dispatched && load_attempt_is_current(attempt) { LEGACY_MIGRATION_ATTEMPTED.store(false, Ordering::Relaxed); + disarm_sweep_wave(wave); + return true; } + let rest: Vec> = waves.collect(); + let worker = LoadWorkerGuard::new(); + crate::util::safe_spawn_local(async move { + let _worker = worker; + let mut wave = wave; + for generations in rest { + await_sweep_wave(wave).await; + disarm_sweep_wave(wave); + wave = arm_sweep_wave(); + let mut failed = false; + for &i in &generations { + failed |= send_sweep_generation(i, attempt, wave).await.failed; + } + if failed { + // The connection dropped mid-sweep: let the next load attempt + // sweep again rather than skip the generations not yet asked + // (the journaled walk asks them too). + LEGACY_MIGRATION_ATTEMPTED.store(false, Ordering::Relaxed); + break; + } + } + disarm_sweep_wave(wave); + }); + // The `freenet-migrate` walk is no longer armed from here: it runs on EVERY // load while the migration journal is incomplete // (`migration_journal::start_journal_load`), whatever the load plan @@ -11659,3 +11666,159 @@ pub(crate) async fn fire_legacy_migration_request() -> bool { // the journal (`migration_journal::gate_sweep_import`). true } + +/// The sweep's waves: generation indexes, newest first, at most +/// [`SWEEP_WAVE_GENERATIONS`] per wave, every generation exactly once. Pure. +pub(crate) fn sweep_wave_plan(generations: usize) -> Vec> { + let newest_first: Vec = (0..generations).rev().collect(); + newest_first + .chunks(SWEEP_WAVE_GENERATIONS) + .map(<[usize]>::to_vec) + .collect() +} + +/// What sending one generation's sweep requests did. +#[derive(Default, Clone, Copy)] +struct SweepSent { + /// At least one request went out. + dispatched: bool, + /// At least one send failed in transport (the connection is down). + failed: bool, +} + +/// Send legacy generation `i` its three sweep requests (the `rooms_data` and +/// `outbound_dms` GETs and the `ListRequest`), each expected by sweep wave +/// `wave`. Fire-and-forget: replies are routed by the response handler. +async fn send_sweep_generation(i: usize, attempt: u32, wave: u64) -> SweepSent { + let mut sent = SweepSent::default(); + let (key_bytes, code_hash_bytes) = &LEGACY_DELEGATES[i]; + let legacy_code_hash = CodeHash::new(*code_hash_bytes); + let legacy_delegate_key = DelegateKey::new(*key_bytes, legacy_code_hash); + + // Send a GetRequest for each storage key we want to migrate. + // rooms_data is the original and gates the migration; outbound_dms + // (#256) is mirrored so a delegate rebuild doesn't orphan the + // sender's own DM plaintext. + let storage_keys: [&[u8]; 2] = [ROOMS_STORAGE_KEY, OUTBOUND_DMS_STORAGE_KEY]; + for storage_key in storage_keys { + let request = ChatDelegateRequestMsg::GetRequest { + key: ChatDelegateKey::new(storage_key.to_vec()), + }; + + let mut payload = Vec::new(); + if let Err(e) = ciborium::ser::into_writer(&request, &mut payload) { + error!( + "Failed to serialize legacy migration request #{} for key {:?}: {}", + i, + String::from_utf8_lossy(storage_key), + e + ); + continue; + } + + let app_msg = freenet_stdlib::prelude::ApplicationMessage::new(payload); + + let delegate_request = DelegateOp(DelegateRequest::ApplicationMessages { + key: legacy_delegate_key.clone(), + params: Parameters::from(Vec::::new()), + inbound: vec![freenet_stdlib::prelude::InboundDelegateMsg::ApplicationMessage(app_msg)], + }); + + let api_result = { + let mut web_api = WEB_API.write(); + if let Some(api) = web_api.as_mut() { + api.send(delegate_request).await + } else { + Err(freenet_stdlib::client_api::Error::ConnectionClosed) + } + }; + + match api_result { + Ok(_) => { + sent.dispatched = true; + expect_sweep_answer(wave, key_bytes); + info!( + "Legacy migration request #{} sent for key {:?}", + i, + String::from_utf8_lossy(storage_key) + ); + } + Err(e) => { + // A send that RETURNS Err is a transport failure (connection + // closed), NOT "delegate not present". Mark it (P1) — but only + // for the current attempt (P2#1), so a stale probe can't + // pollute a reconnect's fresh load. + info!( + "Legacy migration request #{} for key {:?} failed to send \ + (transport): {}", + i, + String::from_utf8_lossy(storage_key), + e + ); + sent.failed = true; + mark_fetch_failure_if_current(attempt); + } + } + } + + // Per-room keys (room:, rooms_meta) are DYNAMIC, so the fixed + // GetRequests above can't discover them. Probe the legacy delegate's key + // index with a ListRequest; the ListResponse (is_legacy) is routed in + // response_handler.rs to a per-room migration that reconstructs those + // slots from THIS legacy delegate and re-saves them to the current one + // (freenet/river#345 / #65). Without this, the NEXT delegate-WASM bump + // after per-room storage ships would strand every room under the old key. + // Fire-and-forget (no waiter): a ListResponse only arrives if the legacy + // delegate is present, so an absent delegate costs nothing rather than a + // 10s per-delegate timeout for fresh users. Its reply or error only paces + // the next sweep wave. + let list_request = ChatDelegateRequestMsg::ListRequest; + let mut list_payload = Vec::new(); + if let Err(e) = ciborium::ser::into_writer(&list_request, &mut list_payload) { + error!("Failed to serialize legacy ListRequest #{i}: {e}"); + return sent; + } + let list_app_msg = freenet_stdlib::prelude::ApplicationMessage::new(list_payload); + let list_delegate_request = DelegateOp(DelegateRequest::ApplicationMessages { + key: legacy_delegate_key.clone(), + params: Parameters::from(Vec::::new()), + inbound: vec![ + freenet_stdlib::prelude::InboundDelegateMsg::ApplicationMessage(list_app_msg), + ], + }); + let list_result = { + let mut web_api = WEB_API.write(); + if let Some(api) = web_api.as_mut() { + api.send(list_delegate_request).await + } else { + Err(freenet_stdlib::client_api::Error::ConnectionClosed) + } + }; + match list_result { + Ok(_) => { + sent.dispatched = true; + expect_sweep_answer(wave, key_bytes); + info!("Legacy ListRequest #{i} sent for per-room key discovery"); + } + Err(e) => { + // Transport failure on the ListRequest send → mark it (P1), + // current-attempt only (P2#1) via the uniform helper (review 11). + info!("Legacy ListRequest #{i} failed to send (transport): {e}"); + sent.failed = true; + mark_fetch_failure_if_current(attempt); + } + } + sent +} + +/// Wait until sweep wave `wave` is answered, or [`SWEEP_WAVE_WAIT_MS`]. Silence +/// concludes nothing here: the sweep never treated a missing answer as +/// absence, and the journaled walk asks again. +async fn await_sweep_wave(wave: u64) { + if let Some(answered) = seal_sweep_wave(wave) { + let timeout = Box::pin(crate::util::sleep(std::time::Duration::from_millis( + SWEEP_WAVE_WAIT_MS, + ))); + let _ = select(answered, timeout).await; + } +} diff --git a/ui/src/components/app/freenet_api/delegate_migration.rs b/ui/src/components/app/freenet_api/delegate_migration.rs index 548ed5db7..b2803c7d8 100644 --- a/ui/src/components/app/freenet_api/delegate_migration.rs +++ b/ui/src/components/app/freenet_api/delegate_migration.rs @@ -1472,6 +1472,11 @@ pub(crate) struct RiverSuccessorIo<'a, C: RiverDelegateChannel, S: RiverImportSe /// key is written, so only a read that would otherwise have gone to the /// node a moment later is answered from here (freenet/river#809). prefetched_markers: HashMap, Option>>, + /// The journal as [`begin_with`] read it to decide on the prefetch, used + /// by the first [`JournaledSuccessorIo::load_journal`] instead of reading + /// it again, so the crate's `begin` sees the same journal the markers + /// were read against. + prefetched_journal: Option>>, } impl<'a, C: RiverDelegateChannel, S: RiverImportSeam> RiverSuccessorIo<'a, C, S> { @@ -1487,6 +1492,7 @@ impl<'a, C: RiverDelegateChannel, S: RiverImportSeam> RiverSuccessorIo<'a, C, S> walk_orders: Vec::new(), rooms_merged: false, prefetched_markers: HashMap::new(), + prefetched_journal: None, } } @@ -1966,12 +1972,17 @@ impl JournaledSuccessorIo /// mismatched reply): `None` would make a later store start the journal /// afresh over a stored one. async fn load_journal(&mut self) -> Result>, Self::Error> { - let loaded = get_current( - self.channel, - &self.successor, - MIGRATION_JOURNAL_KEY.to_vec(), - ) - .await?; + let loaded = match self.prefetched_journal.take() { + Some(loaded) => loaded, + None => { + get_current( + self.channel, + &self.successor, + MIGRATION_JOURNAL_KEY.to_vec(), + ) + .await? + } + }; self.journal = loaded.clone(); Ok(loaded) } @@ -2034,16 +2045,17 @@ pub(crate) async fn begin_with( // every marker, and that `begin` holds the sweep's first merge, so only it // is worth the prefetch. With a journal stored, `begin` runs off the // visible path and reads few markers; the prefetch there only competes - // with the room load (about 0.2 s later in the rehearsal). A journal read - // that is silent or errs skips the prefetch; `begin` meets the same - // failure itself. - let fresh = matches!( - get_current(channel, &successor, MIGRATION_JOURNAL_KEY.to_vec()).await, - Ok(None) - ); + // with the room load (about 0.2 s later in the rehearsal). The journal read + // here is handed to `begin`'s own first journal load, so it costs no extra + // round trip. A read that is silent or errs skips the prefetch and is not + // handed on; `begin` reads again and meets the failure itself. + let journal = get_current(channel, &successor, MIGRATION_JOURNAL_KEY.to_vec()).await; let mut writer = RiverSuccessorIo::new(channel, NoImports, successor); - if fresh { - writer.prefetch_markers(lineage).await; + if let Ok(journal) = journal { + if journal.is_none() { + writer.prefetch_markers(lineage).await; + } + writer.prefetched_journal = Some(journal); } begin_migration_journal(&mut writer, lineage, &river_secret_policy()).await } diff --git a/ui/src/components/app/freenet_api/response_handler.rs b/ui/src/components/app/freenet_api/response_handler.rs index 3f435a4de..b88331493 100644 --- a/ui/src/components/app/freenet_api/response_handler.rs +++ b/ui/src/components/app/freenet_api/response_handler.rs @@ -157,7 +157,11 @@ impl ResponseHandler { let responding_delegate = key.clone(); if is_legacy_delegate { info!("Received response from LEGACY delegate - checking for migration data"); - crate::components::app::chat_delegate::note_sweep_answer(key.bytes()); + if values.is_empty() { + // Core 0.2.136 and earlier answer a delegate it does + // not have with an empty reply: still an answer. + crate::components::app::chat_delegate::note_sweep_answer(key.bytes()); + } } // Log a short key fingerprint, not the full `DelegateKey` Debug: that @@ -204,6 +208,15 @@ impl ResponseHandler { ); if let Ok(response) = deserialization_result { + if is_legacy_delegate + && crate::components::app::chat_delegate::is_sweep_reply( + &response, + ) + { + crate::components::app::chat_delegate::note_sweep_answer( + responding_delegate.bytes(), + ); + } // Summary, NOT the full Debug. This single call site was // 98.5% of all console bytes River emits during startup: // 39 lines averaging ~383 KB (14.93 MB of 15.15 MB total). @@ -795,6 +808,22 @@ impl ResponseHandler { /// single wave. Waves run sequentially, so each wave also gets a fresh response /// timeout — a slow first-time delegate WASM compile can't time out later rooms. const ROOM_LOAD_CONCURRENCY: usize = 16; + +/// One legacy generation's per-room slot fan-out at a time (freenet/river#809). +static LEGACY_SLOT_FETCH: std::sync::LazyLock> = + std::sync::LazyLock::new(|| futures::lock::Mutex::new(())); + +// What one client can have in the node's shared delegate admission queue +// (capped at 100 for every client and delegate together, freenet-core +// `fair_queue.rs`) on the first load after a re-key: a sweep wave, one slot +// fan-out, and a marker-prefetch wave. Kept to half the cap, leaving room for +// other clients on a shared node (freenet/river#809). +const _: () = assert!( + 3 * crate::components::app::chat_delegate::SWEEP_WAVE_GENERATIONS + + ROOM_LOAD_CONCURRENCY + + super::delegate_migration::MARKER_PREFETCH_WAVE + <= 50 +); // `slice::chunks(0)` panics; pin the wave size as a valid chunk at compile time. const _: () = assert!(ROOM_LOAD_CONCURRENCY > 0); @@ -1434,7 +1463,17 @@ async fn migrate_legacy_per_room(legacy_key: DelegateKey, keys: Vec, @@ -1463,6 +1502,7 @@ async fn migrate_legacy_per_room(legacy_key: DelegateKey, keys: Vec Rooms { @@ -4694,9 +4695,10 @@ mod tests { /// freenet/river#590, PROBE-B — the interleaving that probe order makes the /// LIKELY one, and that a "is it in the map right now" guard cannot see. /// - /// `fire_legacy_migration_request` dispatches `LEGACY_DELEGATES.iter()` in - /// file order, oldest generation first, so the OLDEST generation usually - /// answers while the map is still empty. Its stale tombstone lands + /// `fire_legacy_migration_request` dispatched `LEGACY_DELEGATES.iter()` in + /// file order, oldest generation first, until freenet/river#809 (newest + /// first since), so the OLDEST generation usually answered while the map + /// was still empty. The interleaving stays possible on any order. Its stale tombstone lands /// unopposed; the newest generation's `Present` then arrives and is skipped /// by the tombstone check; and the re-save writes `room:R = Tombstone` to the /// current delegate, taking `self_sk` with it. From 7da0a3d678eb3515dc7ba00a0c52f20e78449c04 Mon Sep 17 00:00:00 2001 From: Ian Clarke Date: Sat, 17 Oct 2026 12:39:33 -0500 Subject: [PATCH 4/9] test(ui): pin that the sweep awaits no wave before returning (#809) Claude-Session: https://claude.ai/code/session_01Mynjd42gMDNsNSxwbJWKu3 --- ui/src/components/app/chat_delegate.rs | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/ui/src/components/app/chat_delegate.rs b/ui/src/components/app/chat_delegate.rs index 2545b6c68..6bc652e9e 100644 --- a/ui/src/components/app/chat_delegate.rs +++ b/ui/src/components/app/chat_delegate.rs @@ -4780,6 +4780,13 @@ mod tests { "every sent request (the two GETs and the ListRequest) must be expected" ); assert!(body.contains("await_sweep_wave(wave).await")); + // The caller writes the load state right after this returns, so no + // wave may be awaited before it does (#810 review): otherwise that + // write lands after the first wave's merge and undoes its `Loaded`. + let spawn = body + .find("crate::util::safe_spawn_local(async move {") + .expect("later waves run from a background task"); + assert!(!body[..spawn].contains("await_sweep_wave(")); } /// The plan: newest first, every generation exactly once, no wave larger From 415e0423a77f924846c7b4f4ec4681a000cba0d1 Mon Sep 17 00:00:00 2001 From: Ian Clarke Date: Sat, 17 Oct 2026 12:50:20 -0500 Subject: [PATCH 5/9] fix(ui): a pacing sweep yields to a newer sweep or load attempt (#809) Round-2 review of #810: the background waves captured their load attempt once and kept pacing after a reconnect or Retry, alongside the journaled walk or a second sweep, and a stale sweep could clear the once-per-session guard under a newer one. - Each sweep has an epoch. Before each later wave it stops if a newer sweep started (leaving the guard to it) or if its load attempt changed (clearing the guard so the new attempt sweeps from the start). Only the newest sweep clears the guard on a transport failure. - A new attempt may start a sweep while an earlier attempt's sweep is still pacing, never over its own. - No background task when the first wave is the whole sweep. - Tests: the stop/start decisions, per-sweep wave isolation, the journal handoff being one-shot (exactly two journal reads in a fresh `begin`), the pacing hooks being wired, and the slot fan-out lock and dedup. - Comments that still described oldest-first dispatch updated. Claude-Session: https://claude.ai/code/session_01Mynjd42gMDNsNSxwbJWKu3 --- ui/src/components/app/chat_delegate.rs | 146 ++++++++++++++++-- .../app/freenet_api/delegate_migration.rs | 22 +++ .../app/freenet_api/response_handler.rs | 30 +++- ui/src/components/direct_messages.rs | 7 +- ui/src/room_data.rs | 16 +- 5 files changed, 201 insertions(+), 20 deletions(-) diff --git a/ui/src/components/app/chat_delegate.rs b/ui/src/components/app/chat_delegate.rs index 6bc652e9e..095090e1f 100644 --- a/ui/src/components/app/chat_delegate.rs +++ b/ui/src/components/app/chat_delegate.rs @@ -4774,19 +4774,69 @@ mod tests { .expect("fire_legacy_migration_request must exist")..]; assert!(body.contains("sweep_wave_plan(LEGACY_DELEGATES.len())")); assert!( - body.matches("expect_sweep_answer(wave, key_bytes);") + body[body.find("async fn send_sweep_generation(").unwrap()..] + .matches("expect_sweep_answer(wave, key_bytes);") .count() == 2, "every sent request (the two GETs and the ListRequest) must be expected" ); assert!(body.contains("await_sweep_wave(wave).await")); // The caller writes the load state right after this returns, so no - // wave may be awaited before it does (#810 review): otherwise that + // wave may be awaited before it does: otherwise that // write lands after the first wave's merge and undoes its `Loaded`. let spawn = body .find("crate::util::safe_spawn_local(async move {") .expect("later waves run from a background task"); assert!(!body[..spawn].contains("await_sweep_wave(")); + // The worker is taken before returning, outside the task, so the idle + // timer cannot arm before the task first runs. + assert!(body[..spawn].contains("let worker = LoadWorkerGuard::new();")); + assert!(body[spawn..].contains("sweep_continuation(")); + } + + /// A pacing sweep stops for a newer sweep (leaving the guard alone) or a + /// new load attempt (clearing it), and a new attempt may start over a + /// sweep still pacing for an earlier one, but not over its own. + #[test] + fn a_pacing_sweep_yields_to_newer_sweeps_and_attempts() { + assert_eq!(sweep_continuation(true, true), SweepNext::Continue); + assert_eq!(sweep_continuation(false, true), SweepNext::Superseded); + assert_eq!(sweep_continuation(false, false), SweepNext::Superseded); + assert_eq!(sweep_continuation(true, false), SweepNext::Rearm); + + assert!(sweep_may_start(false, 0, 5), "first sweep of the session"); + assert!(!sweep_may_start(true, 0, 5), "once per session"); + assert!( + !sweep_may_start(true, 6, 5), + "this attempt's own sweep is pacing" + ); + assert!( + sweep_may_start(true, 5, 5), + "an earlier attempt's sweep is pacing" + ); + } + + /// Waves of overlapping sweeps are kept apart: an answer counts toward + /// every wave expecting its key, and disarming one leaves the other. + #[test] + fn sweep_waves_are_kept_per_sweep() { + let key = [42u8; 32]; + let a = arm_sweep_wave(); + let b = arm_sweep_wave(); + expect_sweep_answer(a, &key); + expect_sweep_answer(b, &key); + let mut done_a = seal_sweep_wave(a).unwrap(); + let mut done_b = seal_sweep_wave(b).unwrap(); + disarm_sweep_wave(a); + note_sweep_answer(&key); + assert!( + done_a.try_recv().is_err(), + "a disarmed wave's waiter is dropped" + ); + assert!(matches!(done_b.try_recv(), Ok(Some(())))); + disarm_sweep_wave(b); + expect_sweep_answer(b, &key); + assert!(seal_sweep_wave(b).is_none(), "a disarmed token is a no-op"); } /// The plan: newest first, every generation exactly once, no wave larger @@ -10976,7 +11026,8 @@ pub(crate) static LOAD_ACTIVITY_GEN: AtomicU32 = AtomicU32::new(0); /// shell is already interactive at ~240 ms). It was 4_000 ms; recalibrated /// toward the actual legacy-probe response burst (~72 fire-and-forget probes over /// 24 delegate generations, measured ~130 ms spread, all in by ~490 ms on a fast -/// link). +/// link). Since freenet/river#809 the sweep sends in waves and holds a load +/// worker while the later ones pace, so this timer arms only after the last. /// /// The FLOOR is set by the one cohort that can flash: a returning user whose /// CURRENT delegate is empty but a LEGACY delegate still has data (the first load @@ -11576,11 +11627,20 @@ pub(crate) fn arm_legacy_migration_recovery() -> bool { /// (nothing more is coming) — see [`load_state_after_probe_legacy`]. pub(crate) async fn fire_legacy_migration_request() -> bool { // Only attempt migration once per session to avoid error spam on reconnect. - // If the legacy delegates aren't installed, retrying won't help. - if LEGACY_MIGRATION_ATTEMPTED.swap(true, Ordering::Relaxed) { + // If the legacy delegates aren't installed, retrying won't help. The one + // exception: a sweep still pacing its waves for an EARLIER load attempt + // (freenet/river#809). It stops at its next wave, so this attempt starts a + // full sweep of its own rather than skip the generations it had not asked. + let already = LEGACY_MIGRATION_ATTEMPTED.swap(true, Ordering::Relaxed); + if !sweep_may_start( + already, + SWEEP_RUNNING_FOR.load(Ordering::Relaxed), + LOAD_ATTEMPT_GEN.load(Ordering::Relaxed), + ) { info!("Legacy migration already attempted this session, skipping"); return false; } + let epoch = SWEEP_EPOCH.fetch_add(1, Ordering::Relaxed) + 1; info!( "Attempting to migrate data from {} legacy delegate(s)", @@ -11643,13 +11703,37 @@ pub(crate) async fn fire_legacy_migration_request() -> bool { } let rest: Vec> = waves.collect(); + if rest.is_empty() { + disarm_sweep_wave(wave); + return true; + } + // Taken here, before returning, so the load's idle timer cannot arm + // between this return and the task's first poll. let worker = LoadWorkerGuard::new(); + SWEEP_RUNNING_FOR.store(attempt.wrapping_add(1), Ordering::Relaxed); crate::util::safe_spawn_local(async move { let _worker = worker; let mut wave = wave; for generations in rest { await_sweep_wave(wave).await; disarm_sweep_wave(wave); + match sweep_continuation( + SWEEP_EPOCH.load(Ordering::Relaxed) == epoch, + load_attempt_is_current(attempt), + ) { + SweepNext::Continue => {} + SweepNext::Superseded => return, + SweepNext::Rearm => { + // A reconnect or Retry began a new load attempt. Its + // requests go out on a new connection, and this sweep's + // last wave may have gone down with the old one; the new + // attempt sweeps from the start instead (the journaled + // walk asks the generations too). + LEGACY_MIGRATION_ATTEMPTED.store(false, Ordering::Relaxed); + SWEEP_RUNNING_FOR.store(0, Ordering::Relaxed); + return; + } + } wave = arm_sweep_wave(); let mut failed = false; for &i in &generations { @@ -11657,13 +11741,19 @@ pub(crate) async fn fire_legacy_migration_request() -> bool { } if failed { // The connection dropped mid-sweep: let the next load attempt - // sweep again rather than skip the generations not yet asked - // (the journaled walk asks them too). - LEGACY_MIGRATION_ATTEMPTED.store(false, Ordering::Relaxed); - break; + // sweep again rather than skip the generations not yet asked. + disarm_sweep_wave(wave); + if SWEEP_EPOCH.load(Ordering::Relaxed) == epoch { + LEGACY_MIGRATION_ATTEMPTED.store(false, Ordering::Relaxed); + SWEEP_RUNNING_FOR.store(0, Ordering::Relaxed); + } + return; } } disarm_sweep_wave(wave); + if SWEEP_EPOCH.load(Ordering::Relaxed) == epoch { + SWEEP_RUNNING_FOR.store(0, Ordering::Relaxed); + } }); // The `freenet-migrate` walk is no longer armed from here: it runs on EVERY @@ -11674,6 +11764,44 @@ pub(crate) async fn fire_legacy_migration_request() -> bool { true } +/// Bumped by every sweep that starts; a pacing sweep whose epoch is no longer +/// the latest stops at its next wave (freenet/river#809). +static SWEEP_EPOCH: AtomicU64 = AtomicU64::new(0); +/// `attempt + 1` of the load attempt whose sweep is still pacing its later +/// waves; 0 when none is. +static SWEEP_RUNNING_FOR: AtomicU32 = AtomicU32::new(0); + +/// Whether a sweep may start now. `already`: the once-per-session guard was +/// set; `running_for`: [`SWEEP_RUNNING_FOR`]. A sweep still pacing for an +/// earlier attempt does not count as this session's sweep: it stops at its +/// next wave, and this attempt sweeps from the start. Pure. +pub(crate) fn sweep_may_start(already: bool, running_for: u32, attempt: u32) -> bool { + !already || (running_for != 0 && running_for != attempt.wrapping_add(1)) +} + +/// What a pacing sweep does before its next wave. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum SweepNext { + /// Send it. + Continue, + /// A newer sweep started, and it asks every generation: stop, and leave + /// the once-per-session guard to it. + Superseded, + /// The load attempt changed and no newer sweep has started: stop, and + /// clear the guard so the new attempt sweeps. + Rearm, +} + +/// Decide [`SweepNext`] from whether this sweep is still the latest and its +/// load attempt still current. Pure. +pub(crate) fn sweep_continuation(latest: bool, attempt_current: bool) -> SweepNext { + match (latest, attempt_current) { + (false, _) => SweepNext::Superseded, + (true, false) => SweepNext::Rearm, + (true, true) => SweepNext::Continue, + } +} + /// The sweep's waves: generation indexes, newest first, at most /// [`SWEEP_WAVE_GENERATIONS`] per wave, every generation exactly once. Pure. pub(crate) fn sweep_wave_plan(generations: usize) -> Vec> { diff --git a/ui/src/components/app/freenet_api/delegate_migration.rs b/ui/src/components/app/freenet_api/delegate_migration.rs index b2803c7d8..c4110f101 100644 --- a/ui/src/components/app/freenet_api/delegate_migration.rs +++ b/ui/src/components/app/freenet_api/delegate_migration.rs @@ -1503,6 +1503,10 @@ impl<'a, C: RiverDelegateChannel, S: RiverImportSeam> RiverSuccessorIo<'a, C, S> /// the first load after a re-key for about 8 s at 120 ms RTT). /// /// Called only before the first `begin` on a successor ([`begin_with`]). + /// Another session can write a marker between this read and its use; + /// markers only move forward, so a value read early can only under-report + /// (an import re-run, or an in-progress flag the journal merge restores), + /// never seal anything. /// /// Only definitive answers are kept: a marker that is silent or errs here /// is read again, one at a time, when `begin` asks for it, and fails there @@ -5086,6 +5090,24 @@ mod tests { ); } + /// `begin_with`'s journal read is handed to `begin`'s first journal load, + /// once: a fresh `begin` reads the journal twice in all (that read, and + /// the store's read-merge-write, which must read fresh to merge another + /// tab's journal stored in between). + #[test] + fn begin_reads_the_journal_once_for_its_start_and_fresh_for_its_store() { + let (lineage, channel) = lineage_n(3); + block_on(begin_with(&channel, successor_key(), &lineage)).unwrap(); + let journal_get = format!("get:{}", String::from_utf8_lossy(MIGRATION_JOURNAL_KEY)); + let reads = channel + .log + .borrow() + .iter() + .filter(|(_, what)| *what == journal_get) + .count(); + assert_eq!(reads, 2); + } + /// With a journal stored, `begin` is off the visible path and reads few /// markers, so it does not prefetch them all. #[test] diff --git a/ui/src/components/app/freenet_api/response_handler.rs b/ui/src/components/app/freenet_api/response_handler.rs index b88331493..dbb1a4bdc 100644 --- a/ui/src/components/app/freenet_api/response_handler.rs +++ b/ui/src/components/app/freenet_api/response_handler.rs @@ -1437,7 +1437,7 @@ async fn migrate_legacy_per_room(legacy_key: DelegateKey, keys: Vec join_end, + "released only after every wave is awaited" + ); + } + + /// The sweep's pacing hears about replies and errors: without these hooks + /// every wave would sit out its full wait (freenet/river#809). + #[test] + fn sweep_pacing_hooks_are_wired() { + let handler = include_str!("response_handler.rs"); + let production = handler.split("mod tests {").next().unwrap(); + assert!(production.contains("chat_delegate::note_sweep_answer(key.bytes())")); + assert!(production.contains("chat_delegate::is_sweep_reply(")); + let sync = include_str!("freenet_synchronizer.rs"); + let sync_prod = sync.split("mod tests {").next().unwrap(); + assert!(sync_prod.contains("chat_delegate::note_sweep_error(")); } /// freenet/river#397 Codex review 4 (concurrency P2): the legacy re-save diff --git a/ui/src/components/direct_messages.rs b/ui/src/components/direct_messages.rs index 180652c13..10d7c0bd8 100644 --- a/ui/src/components/direct_messages.rs +++ b/ui/src/components/direct_messages.rs @@ -773,9 +773,10 @@ pub(crate) struct DmCutoffPlan { /// /// **Advance (OlderSnapshot only).** [`seed_dm_last_seen_if_needed`] treats /// every inbound DM in the FIRST non-empty `ROOMS` as already seen, once per -/// session. During a migration that first observation is usually the OLDEST -/// legacy generation's snapshot (probes go out oldest-first), so DMs only a -/// newer generation held all showed as unread. A copy merged as +/// session. During a migration that first observation was usually the OLDEST +/// legacy generation's snapshot (probes went out oldest-first until +/// freenet/river#809), so DMs only a newer generation held all showed as +/// unread. Any generation can still answer first. A copy merged as /// `OlderSnapshot` (a legacy generation, the crate walk's import, the current /// delegate's pre-per-room `rooms_data` blob) was written by an EARLIER /// session, so its DMs that landed count as seen: the seed's own diff --git a/ui/src/room_data.rs b/ui/src/room_data.rs index 87dd5a02e..cfd14cafc 100644 --- a/ui/src/room_data.rs +++ b/ui/src/room_data.rs @@ -2747,8 +2747,9 @@ impl Rooms { // against the combined set, which lost that ordering entirely: an ancient // "you left this room" deleted a room you had since rejoined, and the // migration's re-save wrote the tombstone to the current delegate, taking - // `self_sk` with it. Probes dispatch oldest-first, so the oldest - // generation usually answers while the map is still empty — which is why + // `self_sk` with it. Probes dispatched oldest-first (until + // freenet/river#809), so the oldest generation usually answered while the + // map was still empty, and any generation still can — which is why // guarding only "is it in the map right now" is not enough. // // An `Authoritative` source (the current delegate, or a deliberate @@ -2824,10 +2825,11 @@ impl Rooms { for (vk, mut room_data) in other.map { // Honour tombstones — never re-add a room the user has left — UNLESS // this observation comes from a strictly NEWER source than the one - // that tombstoned it (freenet/river#590). Probes are dispatched - // oldest-first, so without this a stale "you left" from the oldest - // generation suppresses the newest generation's `Present` and the - // re-save then tombstones the room on the current delegate. + // that tombstoned it (freenet/river#590). Generations answer in any + // order, so without this a stale "you left" from an older + // generation that answers first suppresses a newer generation's + // `Present` and the re-save then tombstones the room on the + // current delegate. if self.removed_rooms.contains(&vk) { if !ranks.presence_survives_tombstone(&vk.to_bytes(), incoming_rank) { continue; @@ -4130,7 +4132,7 @@ mod tests { // ===================================================================== // freenet/river#801: a room's read marker across a legacy migration. // - // The fan-out delivers one copy per delegate generation, oldest first, + // The fan-out delivers one copy per delegate generation, in any order, // and an old generation's frozen snapshot carries a weeks-old marker. The // merged room must keep the marker furthest along in the MERGED buffer, // whichever copy arrived first and whichever generation it came from. From 5e9544bdc22f15e2c67d5d36d5bcdb7f90e74d4c Mon Sep 17 00:00:00 2001 From: Ian Clarke Date: Sat, 17 Oct 2026 13:08:31 -0500 Subject: [PATCH 6/9] fix(ui): cap sweep re-arms and close the sweep-lifecycle gaps from round 3 (#809) Round-3 review of #810 (no Must Fix; no data-loss path): - A reconnect during a sweep's pacing cleared the once-per-session guard with no limit; now at most MAX_SWEEP_REARMS (2) times a session. - A recovery or Retry that cleared the guard could start a second sweep over the same attempt's pacing one; `sweep_may_start` now refuses that. - The running marker is set before the first send and only the latest sweep clears it, so an attempt that begins during the first wave starts its own. - A sweep superseded mid-wave sends no more of that wave; one whose attempt changed during the first wave spawns no task. - All stop paths go through `stop_sweep` (tested pure effects); the guard docs now say the guard limits re-probes and #253 is enforced by callers. Claude-Session: https://claude.ai/code/session_01Mynjd42gMDNsNSxwbJWKu3 --- ui/src/components/app/chat_delegate.rs | 118 ++++++++++++++++++++----- 1 file changed, 96 insertions(+), 22 deletions(-) diff --git a/ui/src/components/app/chat_delegate.rs b/ui/src/components/app/chat_delegate.rs index 095090e1f..9df91bce2 100644 --- a/ui/src/components/app/chat_delegate.rs +++ b/ui/src/components/app/chat_delegate.rs @@ -4528,7 +4528,7 @@ mod tests { } assert!( !body.contains("LEGACY_MIGRATION_ATTEMPTED"), - "begin_load_attempt must NOT touch LEGACY_MIGRATION_ATTEMPTED (retry-only; #253 gate)" + "begin_load_attempt must NOT touch LEGACY_MIGRATION_ATTEMPTED (a plain reconnect must not re-probe)" ); } @@ -4791,7 +4791,16 @@ mod tests { // The worker is taken before returning, outside the task, so the idle // timer cannot arm before the task first runs. assert!(body[..spawn].contains("let worker = LoadWorkerGuard::new();")); - assert!(body[spawn..].contains("sweep_continuation(")); + let task = &body[spawn..]; + assert!(task.contains("sweep_continuation(")); + // Every stop goes through `stop_sweep` (latest-only, capped re-arm); + // nothing in the task writes the guard or the marker directly. + assert!(task.contains("SweepNext::Superseded => return,")); + assert!(task.contains("stop_sweep(epoch, true);")); + assert!(task.contains("stop_sweep(epoch, false);")); + let task_end = task.find("\n });").expect("task end"); + assert!(!task[..task_end].contains("LEGACY_MIGRATION_ATTEMPTED.store(")); + assert!(!task[..task_end].contains("SWEEP_RUNNING_FOR.store(")); } /// A pacing sweep stops for a newer sweep (leaving the guard alone) or a @@ -4810,10 +4819,25 @@ mod tests { !sweep_may_start(true, 6, 5), "this attempt's own sweep is pacing" ); + assert!( + !sweep_may_start(false, 6, 5), + "not even when a recovery or Retry cleared the guard" + ); assert!( sweep_may_start(true, 5, 5), "an earlier attempt's sweep is pacing" ); + + // Stopping: only the latest sweep touches the shared state, and + // re-arming the guard is capped per session. + assert_eq!(sweep_stop_effects(false, true, 0), (false, false)); + assert_eq!(sweep_stop_effects(true, false, 0), (true, false)); + assert_eq!(sweep_stop_effects(true, true, 0), (true, true)); + assert_eq!( + sweep_stop_effects(true, true, MAX_SWEEP_REARMS), + (true, false), + "a run of reconnects cannot re-probe forever" + ); } /// Waves of overlapping sweeps are kept apart: an answer counts toward @@ -11352,8 +11376,11 @@ pub(crate) fn room_set_complete_core(latch: u32, attempt: u32, current_attempt: /// guard), so overlapping passes are harmless. It also zeroes the pending-worker /// count, clears `SAW_FETCH_FAILURE`, returns the rail to `Loading`, and arms a /// fresh gen-guarded hard-max backstop. Does NOT touch `LEGACY_MIGRATION_ATTEMPTED` -/// — a plain reconnect must not force a legacy re-probe (preserves the #253 gate); -/// only `retry_rooms_load` re-enables it. +/// — a plain reconnect must not force a legacy re-probe (re-probe spam). The +/// guard only limits repeat probes; #253 is enforced by the callers, which +/// fire only over an empty current delegate. Besides `retry_rooms_load`, a +/// sweep that a reconnect interrupts mid-pacing re-enables it, at most +/// `MAX_SWEEP_REARMS` times a session (freenet/river#809). pub(crate) fn begin_load_attempt() { LOAD_ATTEMPT_GEN.fetch_add(1, Ordering::Relaxed); PENDING_LOADS.store(0, Ordering::Relaxed); @@ -11407,7 +11434,8 @@ fn schedule_rooms_load_backstop() { pub(crate) fn retry_rooms_load() { // Retry-only (deliberately NOT in `begin_load_attempt`): re-enable the legacy // probe so a legacy source that failed THIS session is retried (P1b). A plain - // reconnect must not force this (re-probe spam on every reconnect). Set + // reconnect must not force this (re-probe spam on every reconnect), except + // one that interrupts a pacing sweep (`stop_sweep`, capped). Set // BEFORE the re-setup so `fire_legacy_migration_request` sees it cleared. LEGACY_MIGRATION_ATTEMPTED.store(false, Ordering::Relaxed); crate::util::safe_spawn_local(async { @@ -11641,6 +11669,10 @@ pub(crate) async fn fire_legacy_migration_request() -> bool { return false; } let epoch = SWEEP_EPOCH.fetch_add(1, Ordering::Relaxed) + 1; + let attempt = LOAD_ATTEMPT_GEN.load(Ordering::Relaxed); + // Marked before the first send, so an attempt that begins meanwhile sees + // this sweep as an earlier attempt's and starts its own. + SWEEP_RUNNING_FOR.store(attempt.wrapping_add(1), Ordering::Relaxed); info!( "Attempting to migrate data from {} legacy delegate(s)", @@ -11660,8 +11692,8 @@ pub(crate) async fn fire_legacy_migration_request() -> bool { // `begin_load_attempt` can advance the attempt mid-flight. Capture the attempt // and gate the global writes (`mark_fetch_failure`, the guard reset) on // still-current, so a STALE probe's late send-failure can't pollute the new - // attempt's `SAW_FETCH_FAILURE` into a false LoadFailed. - let attempt = LOAD_ATTEMPT_GEN.load(Ordering::Relaxed); + // attempt's `SAW_FETCH_FAILURE` into a false LoadFailed. (`attempt` is + // captured above, with the running marker.) // Newest generation first, in waves (freenet/river#809). The node runs // delegate requests one at a time, in arrival order, so oldest-first put // the generation most likely to hold the user's rooms behind every older @@ -11679,6 +11711,7 @@ pub(crate) async fn fire_legacy_migration_request() -> bool { // answer. let mut waves = sweep_wave_plan(LEGACY_DELEGATES.len()).into_iter(); let Some(first) = waves.next() else { + stop_sweep(epoch, false); return true; }; let wave = arm_sweep_wave(); @@ -11699,18 +11732,22 @@ pub(crate) async fn fire_legacy_migration_request() -> bool { if !any_dispatched && load_attempt_is_current(attempt) { LEGACY_MIGRATION_ATTEMPTED.store(false, Ordering::Relaxed); disarm_sweep_wave(wave); + stop_sweep(epoch, false); return true; } let rest: Vec> = waves.collect(); - if rest.is_empty() { + if rest.is_empty() || !load_attempt_is_current(attempt) { + // Done; or a reconnect began a new attempt during the first wave, so + // that attempt sweeps instead (no task: its worker would hold the NEW + // attempt's load open). disarm_sweep_wave(wave); + stop_sweep(epoch, !rest.is_empty()); return true; } // Taken here, before returning, so the load's idle timer cannot arm // between this return and the task's first poll. let worker = LoadWorkerGuard::new(); - SWEEP_RUNNING_FOR.store(attempt.wrapping_add(1), Ordering::Relaxed); crate::util::safe_spawn_local(async move { let _worker = worker; let mut wave = wave; @@ -11726,34 +11763,36 @@ pub(crate) async fn fire_legacy_migration_request() -> bool { SweepNext::Rearm => { // A reconnect or Retry began a new load attempt. Its // requests go out on a new connection, and this sweep's - // last wave may have gone down with the old one; the new - // attempt sweeps from the start instead (the journaled - // walk asks the generations too). - LEGACY_MIGRATION_ATTEMPTED.store(false, Ordering::Relaxed); - SWEEP_RUNNING_FOR.store(0, Ordering::Relaxed); + // last wave may have gone down with the old one, so the + // new attempt may sweep again if its load reaches the + // sweep (the journaled walk asks the generations too). + stop_sweep(epoch, true); return; } } wave = arm_sweep_wave(); let mut failed = false; for &i in &generations { + if sweep_continuation( + SWEEP_EPOCH.load(Ordering::Relaxed) == epoch, + load_attempt_is_current(attempt), + ) != SweepNext::Continue + { + // Superseded mid-wave: send no more of it. + break; + } failed |= send_sweep_generation(i, attempt, wave).await.failed; } if failed { // The connection dropped mid-sweep: let the next load attempt // sweep again rather than skip the generations not yet asked. disarm_sweep_wave(wave); - if SWEEP_EPOCH.load(Ordering::Relaxed) == epoch { - LEGACY_MIGRATION_ATTEMPTED.store(false, Ordering::Relaxed); - SWEEP_RUNNING_FOR.store(0, Ordering::Relaxed); - } + stop_sweep(epoch, true); return; } } disarm_sweep_wave(wave); - if SWEEP_EPOCH.load(Ordering::Relaxed) == epoch { - SWEEP_RUNNING_FOR.store(0, Ordering::Relaxed); - } + stop_sweep(epoch, false); }); // The `freenet-migrate` walk is no longer armed from here: it runs on EVERY @@ -11776,7 +11815,42 @@ static SWEEP_RUNNING_FOR: AtomicU32 = AtomicU32::new(0); /// earlier attempt does not count as this session's sweep: it stops at its /// next wave, and this attempt sweeps from the start. Pure. pub(crate) fn sweep_may_start(already: bool, running_for: u32, attempt: u32) -> bool { - !already || (running_for != 0 && running_for != attempt.wrapping_add(1)) + if running_for == attempt.wrapping_add(1) { + // This attempt's own sweep is pacing: it asks every generation, even + // if a recovery or Retry cleared the guard meanwhile. + return false; + } + !already || running_for != 0 +} + +/// How many times per session a stopped sweep may clear the once-per-session +/// guard so a later load attempt sweeps again. Bounds the re-probes a run of +/// reconnects during the sweep's pacing could otherwise cause. +const MAX_SWEEP_REARMS: u32 = 2; +static SWEEP_REARMS: AtomicU32 = AtomicU32::new(0); + +/// What stopping a sweep does to the shared state: `(clear the running +/// marker, clear the once-per-session guard)`. Only the latest sweep touches +/// either; a re-arm clears the guard while re-arms are left. Pure. +pub(crate) fn sweep_stop_effects(latest: bool, rearm: bool, rearms_used: u32) -> (bool, bool) { + if !latest { + return (false, false); + } + (true, rearm && rearms_used < MAX_SWEEP_REARMS) +} + +/// Stop sweep `epoch`: apply [`sweep_stop_effects`]. +fn stop_sweep(epoch: u64, rearm: bool) { + let latest = SWEEP_EPOCH.load(Ordering::Relaxed) == epoch; + let (clear_running, clear_guard) = + sweep_stop_effects(latest, rearm, SWEEP_REARMS.load(Ordering::Relaxed)); + if clear_running { + SWEEP_RUNNING_FOR.store(0, Ordering::Relaxed); + } + if clear_guard { + SWEEP_REARMS.fetch_add(1, Ordering::Relaxed); + LEGACY_MIGRATION_ATTEMPTED.store(false, Ordering::Relaxed); + } } /// What a pacing sweep does before its next wave. From 16888b5f97aeff8564cb0d3a586bc0b784347c43 Mon Sep 17 00:00:00 2001 From: Ian Clarke Date: Sat, 17 Oct 2026 13:29:34 -0500 Subject: [PATCH 7/9] fix(ui): one re-sweep budget for every path, and re-arm on a last-wave reconnect (#809) Round-4 review of #810 (no Must Fix; no data-loss path): - A new attempt starting over an earlier attempt's pacing sweep now spends a re-arm from the same per-session budget (MAX_SWEEP_REARMS) as a stopped sweep's re-arm, so a reconnect that wins the race with the old sweep no longer bypasses the cap (`sweep_start`, `take_sweep_rearm`). - A re-arm is spent only when the guard is still set: a Retry that already cleared it costs nothing. - An attempt change during the last wave, or during the first when no task is spawned, re-arms; before, the last wave had no next wave to notice it. - The all-sends-failed first wave still clears the guard without a re-arm (the probe never left), but only from the latest sweep. - The guard doc names the #352 recovery as the one caller that fires over a non-empty current delegate, and why that is safe. Claude-Session: https://claude.ai/code/session_01Mynjd42gMDNsNSxwbJWKu3 --- ui/src/components/app/chat_delegate.rs | 149 +++++++++++++++++-------- 1 file changed, 105 insertions(+), 44 deletions(-) diff --git a/ui/src/components/app/chat_delegate.rs b/ui/src/components/app/chat_delegate.rs index 9df91bce2..16bcf4c96 100644 --- a/ui/src/components/app/chat_delegate.rs +++ b/ui/src/components/app/chat_delegate.rs @@ -4793,11 +4793,13 @@ mod tests { assert!(body[..spawn].contains("let worker = LoadWorkerGuard::new();")); let task = &body[spawn..]; assert!(task.contains("sweep_continuation(")); - // Every stop goes through `stop_sweep` (latest-only, capped re-arm); - // nothing in the task writes the guard or the marker directly. + // Every stop in the task goes through `stop_sweep` (latest-only, + // capped re-arm); nothing in it writes the guard or the marker + // directly. The task's end re-arms on an attempt change too. assert!(task.contains("SweepNext::Superseded => return,")); assert!(task.contains("stop_sweep(epoch, true);")); - assert!(task.contains("stop_sweep(epoch, false);")); + assert!(task.contains("stop_sweep(epoch, !load_attempt_is_current(attempt));")); + assert!(body.contains("SweepStart::OverEarlier => take_sweep_rearm(),")); let task_end = task.find("\n });").expect("task end"); assert!(!task[..task_end].contains("LEGACY_MIGRATION_ATTEMPTED.store(")); assert!(!task[..task_end].contains("SWEEP_RUNNING_FOR.store(")); @@ -4813,30 +4815,37 @@ mod tests { assert_eq!(sweep_continuation(false, false), SweepNext::Superseded); assert_eq!(sweep_continuation(true, false), SweepNext::Rearm); - assert!(sweep_may_start(false, 0, 5), "first sweep of the session"); - assert!(!sweep_may_start(true, 0, 5), "once per session"); - assert!( - !sweep_may_start(true, 6, 5), - "this attempt's own sweep is pacing" + assert_eq!(sweep_start(false, 0, 5), SweepStart::Fresh, "first sweep"); + assert_eq!( + sweep_start(true, 0, 5), + SweepStart::Refuse, + "once per session" ); - assert!( - !sweep_may_start(false, 6, 5), + assert_eq!( + sweep_start(true, 6, 5), + SweepStart::Refuse, + "own sweep pacing" + ); + assert_eq!( + sweep_start(false, 6, 5), + SweepStart::Refuse, "not even when a recovery or Retry cleared the guard" ); - assert!( - sweep_may_start(true, 5, 5), - "an earlier attempt's sweep is pacing" + assert_eq!( + sweep_start(true, 5, 5), + SweepStart::OverEarlier, + "an earlier attempt's sweep is pacing: a re-sweep, from the budget" ); - // Stopping: only the latest sweep touches the shared state, and - // re-arming the guard is capped per session. - assert_eq!(sweep_stop_effects(false, true, 0), (false, false)); - assert_eq!(sweep_stop_effects(true, false, 0), (true, false)); - assert_eq!(sweep_stop_effects(true, true, 0), (true, true)); + // Stopping: only the latest sweep touches the shared state, and only + // a re-arm of a guard still set tries to clear it. + assert_eq!(sweep_stop_effects(false, true, true), (false, false)); + assert_eq!(sweep_stop_effects(true, false, true), (true, false)); + assert_eq!(sweep_stop_effects(true, true, true), (true, true)); assert_eq!( - sweep_stop_effects(true, true, MAX_SWEEP_REARMS), + sweep_stop_effects(true, true, false), (true, false), - "a run of reconnects cannot re-probe forever" + "a Retry already cleared it: no re-arm spent" ); } @@ -11378,9 +11387,12 @@ pub(crate) fn room_set_complete_core(latch: u32, attempt: u32, current_attempt: /// fresh gen-guarded hard-max backstop. Does NOT touch `LEGACY_MIGRATION_ATTEMPTED` /// — a plain reconnect must not force a legacy re-probe (re-probe spam). The /// guard only limits repeat probes; #253 is enforced by the callers, which -/// fire only over an empty current delegate. Besides `retry_rooms_load`, a -/// sweep that a reconnect interrupts mid-pacing re-enables it, at most -/// `MAX_SWEEP_REARMS` times a session (freenet/river#809). +/// fire only over an empty current delegate, except the interrupted-migration +/// recovery (#352), which fires only after a per-room load read every slot, +/// and saves room by room through a CAS read-merge-write. Besides +/// `retry_rooms_load` and that recovery, a sweep that a reconnect interrupts +/// mid-pacing re-enables it, at most `MAX_SWEEP_REARMS` times a session +/// (freenet/river#809). pub(crate) fn begin_load_attempt() { LOAD_ATTEMPT_GEN.fetch_add(1, Ordering::Relaxed); PENDING_LOADS.store(0, Ordering::Relaxed); @@ -11660,11 +11672,18 @@ pub(crate) async fn fire_legacy_migration_request() -> bool { // (freenet/river#809). It stops at its next wave, so this attempt starts a // full sweep of its own rather than skip the generations it had not asked. let already = LEGACY_MIGRATION_ATTEMPTED.swap(true, Ordering::Relaxed); - if !sweep_may_start( + let start = match sweep_start( already, SWEEP_RUNNING_FOR.load(Ordering::Relaxed), LOAD_ATTEMPT_GEN.load(Ordering::Relaxed), ) { + // Over an earlier attempt's pacing sweep: a re-sweep, from the same + // per-session budget as a stopped sweep's re-arm. + SweepStart::OverEarlier => take_sweep_rearm(), + SweepStart::Fresh => true, + SweepStart::Refuse => false, + }; + if !start { info!("Legacy migration already attempted this session, skipping"); return false; } @@ -11729,10 +11748,14 @@ pub(crate) async fn fire_legacy_migration_request() -> bool { // legacy delegate simply gets no response — do NOT re-probe-spam that case). // Gated on still-current (P2#1): a stale probe must not reopen the guard for // the new attempt (whose own probe already ran / set it). + // The probe never went out, so this is not a re-probe and costs no + // re-arm; only the latest sweep clears the guard (freenet/river#809). if !any_dispatched && load_attempt_is_current(attempt) { - LEGACY_MIGRATION_ATTEMPTED.store(false, Ordering::Relaxed); + if SWEEP_EPOCH.load(Ordering::Relaxed) == epoch { + LEGACY_MIGRATION_ATTEMPTED.store(false, Ordering::Relaxed); + SWEEP_RUNNING_FOR.store(0, Ordering::Relaxed); + } disarm_sweep_wave(wave); - stop_sweep(epoch, false); return true; } @@ -11742,7 +11765,7 @@ pub(crate) async fn fire_legacy_migration_request() -> bool { // that attempt sweeps instead (no task: its worker would hold the NEW // attempt's load open). disarm_sweep_wave(wave); - stop_sweep(epoch, !rest.is_empty()); + stop_sweep(epoch, !load_attempt_is_current(attempt)); return true; } // Taken here, before returning, so the load's idle timer cannot arm @@ -11778,7 +11801,9 @@ pub(crate) async fn fire_legacy_migration_request() -> bool { load_attempt_is_current(attempt), ) != SweepNext::Continue { - // Superseded mid-wave: send no more of it. + // Superseded, or the attempt changed, mid-wave: send no + // more of it (the stop after the loop or at the next + // wave decides the rest). break; } failed |= send_sweep_generation(i, attempt, wave).await.failed; @@ -11792,7 +11817,9 @@ pub(crate) async fn fire_legacy_migration_request() -> bool { } } disarm_sweep_wave(wave); - stop_sweep(epoch, false); + // An attempt change during the LAST wave has no next wave to notice + // it, so the end re-arms too in that case. + stop_sweep(epoch, !load_attempt_is_current(attempt)); }); // The `freenet-migrate` walk is no longer armed from here: it runs on EVERY @@ -11814,41 +11841,75 @@ static SWEEP_RUNNING_FOR: AtomicU32 = AtomicU32::new(0); /// set; `running_for`: [`SWEEP_RUNNING_FOR`]. A sweep still pacing for an /// earlier attempt does not count as this session's sweep: it stops at its /// next wave, and this attempt sweeps from the start. Pure. -pub(crate) fn sweep_may_start(already: bool, running_for: u32, attempt: u32) -> bool { +pub(crate) fn sweep_start(already: bool, running_for: u32, attempt: u32) -> SweepStart { if running_for == attempt.wrapping_add(1) { // This attempt's own sweep is pacing: it asks every generation, even // if a recovery or Retry cleared the guard meanwhile. - return false; + SweepStart::Refuse + } else if !already { + SweepStart::Fresh + } else if running_for != 0 { + SweepStart::OverEarlier + } else { + SweepStart::Refuse } - !already || running_for != 0 } -/// How many times per session a stopped sweep may clear the once-per-session -/// guard so a later load attempt sweeps again. Bounds the re-probes a run of -/// reconnects during the sweep's pacing could otherwise cause. +/// Whether a sweep may start, from [`sweep_start`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum SweepStart { + /// The guard was clear: the session's sweep, or one a Retry or recovery + /// allowed. + Fresh, + /// An earlier attempt's sweep is still pacing: it stops at its next + /// wave, and this one starts over, if a re-arm is left. + OverEarlier, + /// Already swept this session, or this attempt's own sweep is pacing. + Refuse, +} + +/// Take one of this session's [`MAX_SWEEP_REARMS`]; false when none is left. +fn take_sweep_rearm() -> bool { + SWEEP_REARMS + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |used| { + (used < MAX_SWEEP_REARMS).then_some(used + 1) + }) + .is_ok() +} + +/// How many re-sweeps a session allows: a stopped sweep clearing the +/// once-per-session guard, or a new attempt starting over an earlier +/// attempt's pacing sweep, each take one. Bounds the re-probes a run of +/// reconnects during the sweep's pacing (or a socket the sweep keeps +/// dropping) could otherwise cause. A Retry and the interrupted-migration +/// recovery clear the guard themselves and take none. const MAX_SWEEP_REARMS: u32 = 2; static SWEEP_REARMS: AtomicU32 = AtomicU32::new(0); /// What stopping a sweep does to the shared state: `(clear the running -/// marker, clear the once-per-session guard)`. Only the latest sweep touches -/// either; a re-arm clears the guard while re-arms are left. Pure. -pub(crate) fn sweep_stop_effects(latest: bool, rearm: bool, rearms_used: u32) -> (bool, bool) { +/// marker, try to clear the once-per-session guard)`. Only the latest sweep +/// touches either, and only a re-arm of a guard that is still set tries +/// (a Retry may have cleared it already; that must not spend a re-arm). Pure. +pub(crate) fn sweep_stop_effects(latest: bool, rearm: bool, guard_set: bool) -> (bool, bool) { if !latest { return (false, false); } - (true, rearm && rearms_used < MAX_SWEEP_REARMS) + (true, rearm && guard_set) } -/// Stop sweep `epoch`: apply [`sweep_stop_effects`]. +/// Stop sweep `epoch`: apply [`sweep_stop_effects`], the guard clear only +/// while a re-arm is left. fn stop_sweep(epoch: u64, rearm: bool) { let latest = SWEEP_EPOCH.load(Ordering::Relaxed) == epoch; - let (clear_running, clear_guard) = - sweep_stop_effects(latest, rearm, SWEEP_REARMS.load(Ordering::Relaxed)); + let (clear_running, try_clear_guard) = sweep_stop_effects( + latest, + rearm, + LEGACY_MIGRATION_ATTEMPTED.load(Ordering::Relaxed), + ); if clear_running { SWEEP_RUNNING_FOR.store(0, Ordering::Relaxed); } - if clear_guard { - SWEEP_REARMS.fetch_add(1, Ordering::Relaxed); + if try_clear_guard && take_sweep_rearm() { LEGACY_MIGRATION_ATTEMPTED.store(false, Ordering::Relaxed); } } From f4e1da668b48dcf3ae5cb6c753985273493b2011 Mon Sep 17 00:00:00 2001 From: Ian Clarke Date: Sat, 17 Oct 2026 13:34:07 -0500 Subject: [PATCH 8/9] fix(ui): await the sweep's last wave before stopping; an unsent wave never spends a re-arm (#809) Round-5 review of #810 (no Must Fix): - The last-wave re-arm covered only the wave's sends: the task stopped straight after them. It now awaits the last wave like the others, so an attempt change while its replies are out re-arms. - A first wave that went out nowhere (every send failed) clears the guard without spending a re-arm even when a reconnect began a new attempt meanwhile; the latest-sweep check replaces the still-current one, since a newer sweep is that attempt's own probe. - Docs name both users of the re-sweep budget. Claude-Session: https://claude.ai/code/session_01Mynjd42gMDNsNSxwbJWKu3 --- ui/src/components/app/chat_delegate.rs | 23 ++++++++++++++--------- 1 file changed, 14 insertions(+), 9 deletions(-) diff --git a/ui/src/components/app/chat_delegate.rs b/ui/src/components/app/chat_delegate.rs index 16bcf4c96..8d44e08e6 100644 --- a/ui/src/components/app/chat_delegate.rs +++ b/ui/src/components/app/chat_delegate.rs @@ -4758,9 +4758,9 @@ mod tests { .expect("must reset the guard when no probe dispatched"); let reset_block = &body[reset_pos..(reset_pos + 200).min(body.len())]; assert!( - reset_block.contains("load_attempt_is_current(attempt)") + reset_block.contains("SWEEP_EPOCH.load(Ordering::Relaxed) == epoch") && reset_block.contains("LEGACY_MIGRATION_ATTEMPTED.store(false"), - "all-sends-failed (still current) must reset LEGACY_MIGRATION_ATTEMPTED so a reconnect re-probes" + "all-sends-failed (latest sweep) must reset LEGACY_MIGRATION_ATTEMPTED so a reconnect re-probes" ); } @@ -11391,8 +11391,8 @@ pub(crate) fn room_set_complete_core(latch: u32, attempt: u32, current_attempt: /// recovery (#352), which fires only after a per-room load read every slot, /// and saves room by room through a CAS read-merge-write. Besides /// `retry_rooms_load` and that recovery, a sweep that a reconnect interrupts -/// mid-pacing re-enables it, at most `MAX_SWEEP_REARMS` times a session -/// (freenet/river#809). +/// mid-pacing re-enables it, or the new attempt starts over it; the two +/// share `MAX_SWEEP_REARMS` per session (freenet/river#809). pub(crate) fn begin_load_attempt() { LOAD_ATTEMPT_GEN.fetch_add(1, Ordering::Relaxed); PENDING_LOADS.store(0, Ordering::Relaxed); @@ -11447,7 +11447,8 @@ pub(crate) fn retry_rooms_load() { // Retry-only (deliberately NOT in `begin_load_attempt`): re-enable the legacy // probe so a legacy source that failed THIS session is retried (P1b). A plain // reconnect must not force this (re-probe spam on every reconnect), except - // one that interrupts a pacing sweep (`stop_sweep`, capped). Set + // one that interrupts a pacing sweep (`stop_sweep`, or a new attempt + // starting over it; both capped by `MAX_SWEEP_REARMS`). Set // BEFORE the re-setup so `fire_legacy_migration_request` sees it cleared. LEGACY_MIGRATION_ATTEMPTED.store(false, Ordering::Relaxed); crate::util::safe_spawn_local(async { @@ -11749,8 +11750,10 @@ pub(crate) async fn fire_legacy_migration_request() -> bool { // Gated on still-current (P2#1): a stale probe must not reopen the guard for // the new attempt (whose own probe already ran / set it). // The probe never went out, so this is not a re-probe and costs no - // re-arm; only the latest sweep clears the guard (freenet/river#809). - if !any_dispatched && load_attempt_is_current(attempt) { + // re-arm, whether or not a reconnect began a new attempt meanwhile. The + // still-current check above is now the latest-sweep check: a newer sweep + // (that attempt's own probe) owns the guard (freenet/river#809). + if !any_dispatched { if SWEEP_EPOCH.load(Ordering::Relaxed) == epoch { LEGACY_MIGRATION_ATTEMPTED.store(false, Ordering::Relaxed); SWEEP_RUNNING_FOR.store(0, Ordering::Relaxed); @@ -11816,9 +11819,11 @@ pub(crate) async fn fire_legacy_migration_request() -> bool { return; } } + // The last wave has no next wave to notice an attempt change, so it + // is awaited here too, and the end re-arms if the attempt changed + // while it was out (its replies may have gone down with the socket). + await_sweep_wave(wave).await; disarm_sweep_wave(wave); - // An attempt change during the LAST wave has no next wave to notice - // it, so the end re-arms too in that case. stop_sweep(epoch, !load_attempt_is_current(attempt)); }); From 7bd702f61a762031383efb072a1f54f5a51bb662 Mon Sep 17 00:00:00 2001 From: Ian Clarke Date: Sat, 17 Oct 2026 13:37:42 -0500 Subject: [PATCH 9/9] docs(ui): stale sweep comments after round 6; river-publish describes the paced sweep (#809) Comment and doc text only: the all-sends-failed guard reset is gated on the latest sweep (not still-current), two room_data.rs test texts no longer call oldest-first the fan-out's usual order, and river-publish.md step 4 describes newest-first waves. Claude-Session: https://claude.ai/code/session_01Mynjd42gMDNsNSxwbJWKu3 --- .claude/rules/river-publish.md | 9 ++++++++- ui/src/components/app/chat_delegate.rs | 17 ++++++++--------- ui/src/room_data.rs | 6 +++--- 3 files changed, 19 insertions(+), 13 deletions(-) diff --git a/.claude/rules/river-publish.md b/.claude/rules/river-publish.md index f878f580b..30fe3cf2a 100644 --- a/.claude/rules/river-publish.md +++ b/.claude/rules/river-publish.md @@ -284,7 +284,14 @@ where stale legacy data overwrites newer state on the current delegate gateway's sandboxed iframe, so in production neither ever persisted, and an interrupted migration stranded rooms for good (#760). The journal, stored in the current delegate, is the only authority. -4. `fire_legacy_migration_request` probes each legacy delegate TWO ways: +4. `fire_legacy_migration_request` probes each legacy delegate TWO ways, + newest generation first, in waves of 4 generations (freenet/river#809: + every delegate request on a node shares one admission queue capped at 100, + and a request refused there comes back naming no delegate). The first wave + goes out before it returns; the rest follow from a background task that + holds a load worker, each once the previous wave is answered or after 2 s. + A pacing sweep stops when a newer sweep starts or the load attempt + changes, and may be re-armed at most twice a session: (a) fixed `GetRequest` for `[ROOMS_STORAGE_KEY, OUTBOUND_DMS_STORAGE_KEY]` (the pre-#345 single-blob format + DM cache), and (b) a `ListRequest` to discover the legacy delegate's **dynamic per-room keys** — without (b), the diff --git a/ui/src/components/app/chat_delegate.rs b/ui/src/components/app/chat_delegate.rs index 8d44e08e6..474507031 100644 --- a/ui/src/components/app/chat_delegate.rs +++ b/ui/src/components/app/chat_delegate.rs @@ -4721,9 +4721,10 @@ mod tests { /// (all sends failed / connection down) the once-per-session /// `LEGACY_MIGRATION_ATTEMPTED` guard is reset so a reconnect re-probes. A send /// that SUCCEEDS sets `any_dispatched` and does NOT mark (the genuine new-user - /// case). Review 10 (P2#1): the probe captures the attempt and gates BOTH the - /// mark and the guard reset on still-current, so a stale probe (superseded by - /// a reconnect's begin_load_attempt) can't pollute the new attempt. + /// case). Review 10 (P2#1): the probe captures the attempt and gates the mark + /// on still-current, and the guard reset on being the latest sweep + /// (freenet/river#809), so a stale probe (superseded by a reconnect's + /// begin_load_attempt) can't pollute the new attempt. #[test] fn legacy_probe_transport_failure_marks_and_resets_guard() { // `fire_legacy_migration_request` is the last fn in the file, defined @@ -11747,12 +11748,10 @@ pub(crate) async fn fire_legacy_migration_request() -> bool { // for a probe that never left. Reset it so a reconnect re-probes. Keep it set // when at least one probe dispatched (a genuine new user whose node has no // legacy delegate simply gets no response — do NOT re-probe-spam that case). - // Gated on still-current (P2#1): a stale probe must not reopen the guard for - // the new attempt (whose own probe already ran / set it). - // The probe never went out, so this is not a re-probe and costs no - // re-arm, whether or not a reconnect began a new attempt meanwhile. The - // still-current check above is now the latest-sweep check: a newer sweep - // (that attempt's own probe) owns the guard (freenet/river#809). + // A stale probe must not reopen the guard for a new attempt whose own probe + // already ran (P2#1): the reset is gated on this being the latest sweep, + // since a new attempt's probe starts a newer one (freenet/river#809). It + // costs no re-arm: the probe never went out, so this is not a re-probe. if !any_dispatched { if SWEEP_EPOCH.load(Ordering::Relaxed) == epoch { LEGACY_MIGRATION_ATTEMPTED.store(false, Ordering::Relaxed); diff --git a/ui/src/room_data.rs b/ui/src/room_data.rs index cfd14cafc..638eb11f8 100644 --- a/ui/src/room_data.rs +++ b/ui/src/room_data.rs @@ -4177,7 +4177,7 @@ mod tests { assert_eq!( merged_marker(vk, (stale.clone(), 3), (fresh.clone(), 20)), Some(ids[2].clone()), - "oldest generation first (the fan-out's usual order): its stale \ + "oldest generation first (one possible order): its stale \ marker must not stick" ); assert_eq!( @@ -10026,8 +10026,8 @@ mod tests { assert_eq!(owed_rekey_merge(&placeholder), None); } - /// The legacy fan-out after a delegate re-key answers oldest generation - /// first. The marker a room keeps must come from the NEWEST delegate + /// The legacy fan-out after a delegate re-key answers in any order. The + /// marker a room keeps must come from the NEWEST delegate /// generation's copy (it knows the key the device last saved under), not /// from whichever copy got there first (freenet/river#769). #[test]