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 c28b6a80c..474507031 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 @@ -4526,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)" ); } @@ -4719,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 @@ -4756,12 +4759,227 @@ 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" ); } + /// 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_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("sweep_wave_plan(LEGACY_DELEGATES.len())")); + assert!( + 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: 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();")); + let task = &body[spawn..]; + assert!(task.contains("sweep_continuation(")); + // 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, !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(")); + } + + /// 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_eq!(sweep_start(false, 0, 5), SweepStart::Fresh, "first sweep"); + assert_eq!( + sweep_start(true, 0, 5), + SweepStart::Refuse, + "once per session" + ); + 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_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 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, false), + (true, false), + "a Retry already cleared it: no re-arm spent" + ); + } + + /// 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 + /// 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 + /// 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) ===== use freenet_scaffold::util::FastHash; @@ -10842,7 +11060,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 @@ -11167,8 +11386,14 @@ 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, 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, 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); @@ -11222,7 +11447,9 @@ 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`, 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 { @@ -11244,6 +11471,158 @@ 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. +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 +/// 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(()); + } + } + } +} + +/// 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() -> u64 { + let wave = NEXT_SWEEP_WAVE.fetch_add(1, Ordering::Relaxed); + SWEEP_WAVES + .lock() + .unwrap() + .insert(wave, SweepWave::default()); + wave +} + +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(wave: u64) -> Option> { + SWEEP_WAVES + .lock() + .unwrap() + .get_mut(&wave) + .and_then(SweepWave::seal) +} + +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 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]) { + 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) { + for wave in SWEEP_WAVES.lock().unwrap().values_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 @@ -11270,8 +11649,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 @@ -11287,11 +11669,31 @@ 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); + 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; } + 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)", @@ -11311,141 +11713,386 @@ 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 + // 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). + // + // 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 { + stop_sweep(epoch, false); + 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; + } + } - for (i, (key_bytes, code_hash_bytes)) in LEGACY_DELEGATES.iter().enumerate() { - 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()), - }; + // freenet/river#397 Codex review 9 (P1): if NO probe was actually dispatched + // (every send failed — connection down), the once-per-session guard was set + // 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). + // 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); + SWEEP_RUNNING_FOR.store(0, Ordering::Relaxed); + } + disarm_sweep_wave(wave); + return true; + } - 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 rest: Vec> = waves.collect(); + 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, !load_attempt_is_current(attempt)); + 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(); + 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, 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, 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; + } + 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); + stop_sweep(epoch, true); + 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); + stop_sweep(epoch, !load_attempt_is_current(attempt)); + }); - let app_msg = freenet_stdlib::prelude::ApplicationMessage::new(payload); + // 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 + // (freenet/river#763). Every merge this sweep's responses make is gated by + // the journal (`migration_journal::gate_sweep_import`). + true +} - 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), - ], - }); +/// 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_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. + SweepStart::Refuse + } else if !already { + SweepStart::Fresh + } else if running_for != 0 { + SweepStart::OverEarlier + } else { + SweepStart::Refuse + } +} - 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) - } - }; +/// 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() +} - match api_result { - Ok(_) => { - any_dispatched = true; - 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); - } - } - } +/// 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, 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 && guard_set) +} + +/// 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, 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 try_clear_guard && take_sweep_rearm() { + LEGACY_MIGRATION_ATTEMPTED.store(false, Ordering::Relaxed); + } +} - // 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}"); +/// 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> { + 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 list_app_msg = freenet_stdlib::prelude::ApplicationMessage::new(list_payload); - let list_delegate_request = DelegateOp(DelegateRequest::ApplicationMessages { + + 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(list_app_msg), - ], + inbound: vec![freenet_stdlib::prelude::InboundDelegateMsg::ApplicationMessage(app_msg)], }); - let list_result = { + + let api_result = { let mut web_api = WEB_API.write(); if let Some(api) = web_api.as_mut() { - api.send(list_delegate_request).await + api.send(delegate_request).await } else { Err(freenet_stdlib::client_api::Error::ConnectionClosed) } }; - match list_result { + + match api_result { Ok(_) => { - any_dispatched = true; - info!("Legacy ListRequest #{i} sent for per-room key discovery"); + 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) => { - // 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}"); + // 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); } } } - // freenet/river#397 Codex review 9 (P1): if NO probe was actually dispatched - // (every send failed — connection down), the once-per-session guard was set - // 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). - if !any_dispatched && load_attempt_is_current(attempt) { - LEGACY_MIGRATION_ATTEMPTED.store(false, Ordering::Relaxed); + // 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 +} - // 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 - // (freenet/river#763). Every merge this sweep's responses make is gated by - // the journal (`migration_journal::gate_sweep_import`). - true +/// 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 ac7c19877..c4110f101 100644 --- a/ui/src/components/app/freenet_api/delegate_migration.rs +++ b/ui/src/components/app/freenet_api/delegate_migration.rs @@ -1467,6 +1467,16 @@ 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>>, + /// 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> { @@ -1481,6 +1491,67 @@ 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(), + prefetched_journal: None, + } + } + + /// 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). + /// + /// 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 + /// 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 +1592,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 +1631,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 +1646,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 +1712,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; @@ -1891,12 +1976,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) } @@ -1955,7 +2045,22 @@ 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). 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 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 } @@ -2595,6 +2700,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 +2911,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 +2936,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 +5051,134 @@ 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" + ); + } + + /// `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] + 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. + #[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/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 a193b232f..dbb1a4bdc 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; @@ -157,6 +157,11 @@ impl ResponseHandler { let responding_delegate = key.clone(); if is_legacy_delegate { info!("Received response from LEGACY delegate - checking for migration data"); + 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 @@ -203,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). @@ -794,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); @@ -1407,7 +1437,7 @@ 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, + ); + } + drop(slot_fetch); + for (vk, response) in responses { + match response { Ok(ChatDelegateResponseMsg::GetResponse { value: Some(bytes), .. }) => { @@ -3445,7 +3512,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 +3846,76 @@ 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" + ); + // One generation's fan-out at a time, held until its responses are in + // (the queue budget assumes it), over deduplicated room keys. + let lock = region + .find("LEGACY_SLOT_FETCH.lock().await") + .expect("legacy fan-outs must take LEGACY_SLOT_FETCH"); + assert!(lock < enqueue_pos); + assert!(region.contains("seen.insert(vk.to_bytes())")); + let join_end = region.rfind("join_all(").unwrap(); + let drop_pos = region + .find("drop(slot_fetch);") + .expect("the lock is released"); + assert!( + drop_pos > 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 /// completion (in `hydrate_loaded_rooms`) must resolve `Loaded` only when it /// merged live rooms (`had_loaded_rooms`) — an empty legacy completion writes 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 118454bbe..638eb11f8 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; @@ -3970,10 +3972,11 @@ mod tests { // freenet/river#527 — identity conflicts resolve by DELEGATE GENERATION, // not by which probe answered first. // - // `fire_legacy_migration_request` probes every legacy generation at once - // and each hydrates independently, so for a long-time user whose rooms - // exist under many generations these merges arrive in a racy order — - // dispatched oldest-first, so the oldest copy usually lands first. + // `fire_legacy_migration_request` probes every legacy generation and each + // hydrates independently, so for a long-time user whose rooms exist under + // many generations these merges arrive in a racy order. (Dispatched + // oldest-first until freenet/river#809, newest-first in waves since; the + // ranking must hold for either order.) // ===================================================================== fn empty_rooms_for_merge() -> Rooms { @@ -4129,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. @@ -4174,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!( @@ -4694,9 +4697,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. @@ -10022,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]