perf(ui): restore rooms in ~4 s, not 12-18 s, on the first load after a re-key

.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

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<Vec<u8>, Option<Vec<u8>>>,
+    /// 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<Option<Vec<u8>>>,
 }
 
 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<Vec<u8>> = 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<u8>) -> Result<Option<Vec<u8>>, 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<u8>, 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<C: RiverDelegateChannel, S: RiverImportSeam> JournaledSuccessorIo
     /// mismatched reply): `None` would make a later store start the journal
     /// afresh over a stored one.
     async fn load_journal(&mut self) -> Result<Option<Vec<u8>>, 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<C: RiverDelegateChannel>(
     successor: DelegateKey,
     lineage: &[DelegateLineageEntry],
 ) -> Result<JournalStatus, JournalError> {
+    // 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<usize>,
         /// `(target delegate key bytes, "list" | "get:<key>" | "store:<key>")`
         pub(crate) log: RefCell<Vec<(Vec<u8>, String)>>,
+        /// Sequential round trips: each `request` outside a batch, and each
+        /// `request_all` batch, counts one (freenet/river#809).
+        pub(crate) round_trips: Cell<usize>,
+        batching: Cell<bool>,
     }
 
     impl MockChannel {
@@ -2802,6 +2911,9 @@ pub(crate) mod tests_support {
             target: &DelegateKey,
             request: ChatDelegateRequestMsg,
         ) -> Result<Option<ChatDelegateResponseMsg>, 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<Result<Option<ChatDelegateResponseMsg>, 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<DelegateLineageEntry>, 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<u32>) -> SweepWrite {
         block_on(note_sweep_import_with(
             channel,

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() {