Add the journal: following the table's best routes at one's own pace

Whatever acts on the best routes (export to peers, the forwarding table,
the status page) reads their changes from a per-shard journal through a
table::Reader. Nothing uses it yet.

- The journal records that a prefix's best route changed, not what to. A
  reader that comes to an entry is given the route as it is then. A prefix
  that changes a thousand times between reads is one event, and the
  journal never holds more live entries than the table has prefixes.
- Each row keeps the sequence number of its latest change; an entry with
  another number is dead and is dropped when the journal is tidied.
- Entries go once every reader has passed them. A reader that does not
  read cannot hold them back for ever: past a limit the oldest half is
  dropped, and that reader is given the whole table again (Reset, every
  best route, Synced). A new reader starts the same way.
- With no reader nothing is recorded.
- Reader::changed waits for there to be something to read.

Tested against a mirror kept from the events: after any interleaving of
announcements, withdrawals, session flaps, reads with small budgets and
readers coming and going, with journals small enough to be tidied and
overrun constantly, a reader that has caught up holds exactly the table's
best routes. Also with writers on four threads and a reader on a fifth.
The property test found one fault (a pass that resumed part-way claimed
to have caught up) and catches three more put in on purpose.

On the AMS-IX dump, 12 threads: loading takes 2.29 s with a follower
reading throughout against 2.17 s with none, the journal peaking at
170,000 entries for 1.39 million prefixes. With a follower that does not
read, the journal ends at 1.74 million entries, and the follower then
catches up in 0.08 s.

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

crates/bench/benches/real_table.rs

@@ -10,14 +10,14 @@ use std::collections::HashSet;
 use std::net::{Ipv4Addr, SocketAddr};
 use std::path::PathBuf;
 use std::sync::Arc;
-use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::time::{Duration, Instant, SystemTime};
 
 use bgp_bench::memory::Region;
 use bgp_bench::mrt::{Dump, parse_attributes};
 use bgp_packet::message::{NlriNextHop, PathAttribute};
 use bgp_primitives::ip_prefix::IpPrefix;
-use bgp_rib::table::{PeerInfo, RibTable};
+use bgp_rib::table::{Event, PeerIndex, PeerInfo, RibTable};
 use bgp_rib::{AdjRibIn, PeerId, ReceivedUpdate, RibEntry};
 
 /// Bytes allocated and not freed since `region` began; zero when memory
@@ -115,6 +115,7 @@ fn main() {
     let threads = std::thread::available_parallelism().map_or(1, |n| n.get());
     sharded_table(&dump, &peers, &parsed, 1);
     sharded_table(&dump, &peers, &parsed, threads);
+    followed_table(&dump, &peers, &parsed, threads);
 }
 
 /// What the table looks like, as far as it decides what a RIB costs.
@@ -288,25 +289,7 @@ fn current_table(
 /// chosen best route.
 fn sharded_table(dump: &Dump, peers: &[usize], parsed: &[Option<Attributes>], threads: usize) {
     println!("\n== The sharded table, {threads} thread(s)");
-    // A peer's routes grouped as UPDATEs: the attribute set and the
+    let updates = grouped(dump, peers, parsed);
-    // prefixes announced with it.
-    let updates: Vec<Vec<(u32, Vec<IpPrefix>)>> = peers
-        .iter()
-        .map(|&peer| {
-            let mut routes = dump.routes[peer].clone();
-            routes.sort_by_key(|(_, set)| *set);
-            routes
-                .chunk_by(|a, b| a.1 == b.1)
-                .filter(|group| parsed[group[0].1 as usize].is_some())
-                .map(|group| {
-                    (
-                        group[0].1,
-                        group.iter().map(|(prefix, _)| *prefix).collect(),
-                    )
-                })
-                .collect()
-        })
-        .collect();
     let routes: usize = updates
         .iter()
         .flatten()
@@ -314,49 +297,10 @@ fn sharded_table(dump: &Dump, peers: &[usize], parsed: &[Option<Attributes>], th
         .sum();
 
     let region = Region::begin();
-    let table = RibTable::new(threads * 16);
+    let table = Arc::new(RibTable::new(threads * 16));
-    let indices: Vec<_> = peers
+    let indices = add_peers(&table, dump, peers);
-        .iter()
-        .map(|&peer| {
-            table
-                .add_peer(PeerInfo {
-                    asn: dump.peers[peer].asn,
-                    router_id: Ipv4Addr::from(peer as u32 + 1),
-                    address: dump.peers[peer].address,
-                    ibgp: false,
-                })
-                .expect("room for the peer")
-        })
-        .collect();
-
-    let next = AtomicUsize::new(0);
-    let best_changes = AtomicUsize::new(0);
     let started = Instant::now();
-    std::thread::scope(|scope| {
+    let best_changes = load(&table, dump, &updates, &indices, threads);
-        for _ in 0..threads {
-            scope.spawn(|| {
-                let mut changes = 0;
-                loop {
-                    let at = next.fetch_add(1, Ordering::Relaxed);
-                    let Some(updates) = updates.get(at) else {
-                        break;
-                    };
-                    for (set, prefixes) in updates {
-                        // From the bytes, as a dump has them and as a
-                        // session could hand them over without parsing.
-                        let Some(attrs) = table
-                            .attrs()
-                            .intern_wire(&dump.attribute_sets[*set as usize])
-                        else {
-                            continue;
-                        };
-                        changes += table.announce(indices[at], &attrs, prefixes).best_changed;
-                    }
-                }
-                best_changes.fetch_add(changes, Ordering::Relaxed);
-            });
-        }
-    });
     let elapsed = started.elapsed();
     assert_eq!(table.len(), routes);
     let held = retained(&region);
@@ -374,7 +318,7 @@ fn sharded_table(dump: &Dump, peers: &[usize], parsed: &[Option<Attributes>], th
         "    {} prefixes, {} attribute sets, best route changed {} times",
         table.prefix_count(),
         table.attrs().len(),
-        best_changes.load(Ordering::Relaxed)
+        best_changes
     );
 
     let started = Instant::now();
@@ -425,3 +369,183 @@ fn sharded_table(dump: &Dump, peers: &[usize], parsed: &[Option<Attributes>], th
         format!("{swept} dropped"),
     );
 }
+
+/// A peer's routes grouped as UPDATEs: the attribute set and the prefixes
+/// announced with it.
+type Updates = Vec<Vec<(u32, Vec<IpPrefix>)>>;
+
+fn grouped(dump: &Dump, peers: &[usize], parsed: &[Option<Attributes>]) -> Updates {
+    peers
+        .iter()
+        .map(|&peer| {
+            let mut routes = dump.routes[peer].clone();
+            routes.sort_by_key(|(_, set)| *set);
+            routes
+                .chunk_by(|a, b| a.1 == b.1)
+                .filter(|group| parsed[group[0].1 as usize].is_some())
+                .map(|group| {
+                    (
+                        group[0].1,
+                        group.iter().map(|(prefix, _)| *prefix).collect(),
+                    )
+                })
+                .collect()
+        })
+        .collect()
+}
+
+fn add_peers(table: &RibTable, dump: &Dump, peers: &[usize]) -> Vec<PeerIndex> {
+    peers
+        .iter()
+        .map(|&peer| {
+            table
+                .add_peer(PeerInfo {
+                    asn: dump.peers[peer].asn,
+                    router_id: Ipv4Addr::from(peer as u32 + 1),
+                    address: dump.peers[peer].address,
+                    ibgp: false,
+                })
+                .expect("room for the peer")
+        })
+        .collect()
+}
+
+/// Announces everything in `updates` on `threads` threads, each taking
+/// whole peers. Returns how many times a best route changed.
+fn load(
+    table: &RibTable,
+    dump: &Dump,
+    updates: &Updates,
+    indices: &[PeerIndex],
+    threads: usize,
+) -> usize {
+    let next = AtomicUsize::new(0);
+    let best_changes = AtomicUsize::new(0);
+    std::thread::scope(|scope| {
+        for _ in 0..threads {
+            scope.spawn(|| {
+                let mut changes = 0;
+                loop {
+                    let at = next.fetch_add(1, Ordering::Relaxed);
+                    let Some(updates) = updates.get(at) else {
+                        break;
+                    };
+                    for (set, prefixes) in updates {
+                        // From the bytes, as a dump has them and as a
+                        // session could hand them over without parsing.
+                        let Some(attrs) = table
+                            .attrs()
+                            .intern_wire(&dump.attribute_sets[*set as usize])
+                        else {
+                            continue;
+                        };
+                        changes += table.announce(indices[at], &attrs, prefixes).best_changed;
+                    }
+                }
+                best_changes.fetch_add(changes, Ordering::Relaxed);
+            });
+        }
+    });
+    best_changes.load(Ordering::Relaxed)
+}
+
+/// The same load with someone following the best routes through the
+/// journal: once reading as fast as it can while the table fills, once not
+/// reading at all until the table is full.
+fn followed_table(dump: &Dump, peers: &[usize], parsed: &[Option<Attributes>], threads: usize) {
+    println!("\n== The sharded table with a follower, {threads} thread(s)");
+    let updates = grouped(dump, peers, parsed);
+    let routes: usize = updates
+        .iter()
+        .flatten()
+        .map(|(_, prefixes)| prefixes.len())
+        .sum();
+    let count = |counts: &mut (usize, usize, usize), event: Event<'_>| match event {
+        Event::Best { attrs, .. } => {
+            // Look at the route, as a follower would.
+            counts.0 += usize::from(attrs.key().as_path_len > 0);
+        }
+        Event::Withdrawn { .. } => counts.1 += 1,
+        Event::Reset | Event::Synced => counts.2 += 1,
+    };
+
+    // Reading while the table fills.
+    let table = Arc::new(RibTable::new(threads * 16));
+    let indices = add_peers(&table, dump, peers);
+    let done = AtomicBool::new(false);
+    let started = Instant::now();
+    let (best_changes, counts, longest, after_load) = std::thread::scope(|scope| {
+        let follower = scope.spawn(|| {
+            let mut reader = table.reader();
+            let mut counts = (0, 0, 0);
+            let mut longest = 0;
+            while !done.load(Ordering::Acquire) {
+                reader.poll(4096, |event| count(&mut counts, event));
+                longest = longest.max(table.journal_len());
+            }
+            let finishing = Instant::now();
+            while !reader
+                .poll(4096, |event| count(&mut counts, event))
+                .caught_up
+            {}
+            (counts, longest, finishing.elapsed())
+        });
+        let best_changes = load(&table, dump, &updates, &indices, threads);
+        done.store(true, Ordering::Release);
+        let (counts, longest, after_load) = follower.join().expect("the follower runs");
+        (best_changes, counts, longest, after_load)
+    });
+    let elapsed = started.elapsed();
+    assert_eq!(table.len(), routes);
+    row(
+        "load, follower reading throughout",
+        elapsed,
+        format!(
+            "{} routes; caught up {:.3} s after the last route",
+            per_second(routes, elapsed),
+            after_load.as_secs_f64()
+        ),
+    );
+    println!(
+        "    best route changed {best_changes} times; the follower was given {} events for it; \n    journal at most {longest} entries for {} prefixes",
+        counts.0 + counts.1,
+        table.prefix_count()
+    );
+    drop(table);
+
+    // Not reading: the journal has to stay small on its own account.
+    let table = Arc::new(RibTable::new(threads * 16));
+    let indices = add_peers(&table, dump, peers);
+    let mut reader = table.reader();
+    let mut counts = (0, 0, 0);
+    reader.poll(usize::MAX, |event| count(&mut counts, event));
+    let started = Instant::now();
+    load(&table, dump, &updates, &indices, threads);
+    let elapsed = started.elapsed();
+    row(
+        "load, follower not reading",
+        elapsed,
+        format!(
+            "{} routes; journal holds {} entries for {} prefixes",
+            per_second(routes, elapsed),
+            table.journal_len(),
+            table.prefix_count()
+        ),
+    );
+    let started = Instant::now();
+    let mut counts = (0, 0, 0);
+    while !reader
+        .poll(4096, |event| count(&mut counts, event))
+        .caught_up
+    {}
+    let elapsed = started.elapsed();
+    row(
+        "the follower then catches up",
+        elapsed,
+        format!(
+            "{} events, {} of them a reset",
+            counts.0 + counts.1,
+            counts.2
+        ),
+    );
+}

crates/rib/proptest-regressions/table/journal_tests.txt

@@ -0,0 +1,7 @@
+# Seeds for failure cases proptest has generated in the past. It is
+# automatically read and these particular cases re-run before any
+# novel cases are generated.
+#
+# It is recommended to check this file in to source control so that
+# everyone who runs the test benefits from these saved cases.
+cc 7107cf5580fe7bb95e138f1fead386d89273532f83aa1d11808aa9586aa9da63 # shrinks to ops = [Withdraw { peer: 0, prefix: 1 }, Withdraw { peer: 0, prefix: 1 }, Poll { reader: 2, budget: 2 }, Withdraw { peer: 0, prefix: 10 }, Poll { reader: 2, budget: 4 }, Announce { peer: 2, prefix: 7, route: Route { local_pref: None, path: [1, 2], origin: 0, med: Some(0) } }, Announce { peer: 3, prefix: 6, route: Route { local_pref: None, path: [1], origin: 0, med: None } }, Announce { peer: 3, prefix: 6, route: Route { local_pref: None, path: [1, 1], origin: 0, med: None } }, Withdraw { peer: 1, prefix: 5 }, Withdraw { peer: 0, prefix: 11 }, Withdraw { peer: 3, prefix: 6 }, Announce { peer: 0, prefix: 3, route: Route { local_pref: None, path: [1, 2], origin: 1, med: None } }, Poll { reader: 2, budget: 3 }, Withdraw { peer: 0, prefix: 8 }, Flap { peer: 3 }, Withdraw { peer: 1, prefix: 1 }, Announce { peer: 2, prefix: 5, route: Route { local_pref: None, path: [1], origin: 1, med: Some(0) } }, Announce { peer: 0, prefix: 1, route: Route { local_pref: None, path: [1, 1], origin: 1, med: None } }, Withdraw { peer: 0, prefix: 1 }, Withdraw { peer: 2, prefix: 8 }, Poll { reader: 1, budget: 2 }, Flap { peer: 1 }, Poll { reader: 2, budget: 2 }, Announce { peer: 0, prefix: 9, route: Route { local_pref: None, path: [1], origin: 0, med: Some(1) } }, Announce { peer: 2, prefix: 2, route: Route { local_pref: Some(100), path: [2], origin: 0, med: Some(1) } }, Poll { reader: 1, budget: 3 }, Withdraw { peer: 3, prefix: 1 }, Announce { peer: 1, prefix: 4, route: Route { local_pref: None, path: [2], origin: 1, med: None } }, Announce { peer: 3, prefix: 7, route: Route { local_pref: None, path: [2], origin: 0, med: None } }, Announce { peer: 1, prefix: 11, route: Route { local_pref: None, path: [1], origin: 1, med: Some(1) } }, Withdraw { peer: 0, prefix: 6 }, Withdraw { peer: 0, prefix: 4 }, Announce { peer: 3, prefix: 10, route: Route { local_pref: Some(100), path: [2, 2], origin: 1, med: None } }, Announce { peer: 1, prefix: 2, route: Route { local_pref: None, path: [2], origin: 0, med: Some(2) } }, Announce { peer: 1, prefix: 3, route: Route { local_pref: None, path: [1], origin: 1, med: None } }, Withdraw { peer: 0, prefix: 2 }, Announce { peer: 1, prefix: 4, route: Route { local_pref: None, path: [1, 1], origin: 0, med: None } }, Poll { reader: 1, budget: 2 }, Poll { reader: 1, budget: 4 }, Poll { reader: 2, budget: 1 }, Poll { reader: 1, budget: 3 }, Poll { reader: 1, budget: 5 }, Poll { reader: 1, budget: 1 }, Withdraw { peer: 3, prefix: 5 }, Withdraw { peer: 3, prefix: 9 }, Withdraw { peer: 2, prefix: 10 }], shards = 3

crates/rib/src/table/journal.rs

@@ -0,0 +1,554 @@
+//! The journal: how the rest of the router follows the best routes.
+//!
+//! Whatever acts on the table's choices (the peers that are sent the best
+//! routes, the forwarding table, the status page) has to learn when a best
+//! route changes, at its own pace, without the table waiting for it. Each
+//! shard keeps a journal for that, and each follower reads it through a
+//! [`Reader`].
+//!
+//! # What the journal holds
+//!
+//! It records *that* a prefix's best route changed, not what it changed to.
+//! A reader that comes to an entry looks the prefix up and is given the best
+//! route as it is at that moment. A follower does not want the history; it
+//! wants to end up with what the table has now.
+//!
+//! This is what keeps the journal small. If a prefix changes again before a
+//! reader has got to its entry, the older entry no longer says anything: the
+//! newer one will deliver the current route. Each row remembers the
+//! sequence number of its latest change, an entry with any other number is
+//! dead, and dead entries are thrown out when the journal is tidied. So a
+//! prefix that flaps a thousand times costs a reader one event, and the
+//! journal never has more live entries than the table has prefixes.
+//!
+//! # Readers that fall behind
+//!
+//! Entries are dropped once every reader has passed them. A reader that
+//! stops reading would hold them all back, so there is a limit: when the
+//! journal is too long even after tidying, its oldest half is dropped
+//! regardless. A reader that was behind that point is told so the next time
+//! it reads ([`Event::Reset`]) and is given the whole table again. Reading
+//! too slowly costs the slow reader a resynchronisation; it costs the
+//! table, and the other readers, nothing.
+//!
+//! A new reader starts the same way: `Reset`, every best route, `Synced`,
+//! and changes from then on.
+//!
+//! # Order
+//!
+//! For any one prefix, a reader is given its states in the order they came
+//! about, and always, in the end, the latest. Between prefixes there is no
+//! order worth relying on.
+
+use std::collections::VecDeque;
+use std::hash::BuildHasher;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+
+use bgp_primitives::ip_prefix::IpPrefix;
+use foldhash::fast::RandomState;
+use hashbrown::HashMap;
+
+use super::{AttrSet, PeerIndex, RibTable, Shard, read, write};
+
+/// A cursor that holds nothing back: a reader slot that is free, or a
+/// reader that has not read this shard yet.
+const NO_CURSOR: u64 = u64::MAX;
+
+/// When a journal is tidied, and when it is too long.
+#[derive(Debug, Clone, Copy)]
+pub(super) struct Limits {
+    /// Below this many entries a journal is not worth tidying.
+    pub(super) tidy_at: usize,
+    /// Room allowed beyond the number of prefixes before the journal is
+    /// thought too long.
+    pub(super) slack: usize,
+}
+
+impl Default for Limits {
+    fn default() -> Self {
+        Self {
+            tidy_at: 1024,
+            slack: 64,
+        }
+    }
+}
+
+struct Entry {
+    seq: u64,
+    prefix: IpPrefix,
+}
+
+/// One shard's journal.
+pub(super) struct Journal {
+    /// The sequence number of the latest change. Counts every change of a
+    /// best route in the shard, whether or not anyone was reading.
+    seq: u64,
+    /// Entries up to and including this number were dropped without regard
+    /// to the readers. A reader whose cursor is below it missed something.
+    floor: u64,
+    /// In order of sequence number. Not every number is here: dead entries
+    /// are removed, and nothing is recorded while there are no readers.
+    entries: VecDeque<Entry>,
+    /// Prefixes that have no route any more and whose disappearance is
+    /// still to be read, with the sequence number of the entry that says
+    /// so. A row keeps this number itself; a prefix without a row cannot.
+    gone: HashMap<IpPrefix, u64, RandomState>,
+    /// Where each reader has read to, by reader slot. Written by the
+    /// readers themselves while they hold the shard for reading.
+    cursors: Vec<AtomicU64>,
+    /// How many reader slots are in use.
+    readers: usize,
+    limits: Limits,
+}
+
+impl Journal {
+    pub(super) fn new(limits: Limits) -> Self {
+        Self {
+            limits,
+            seq: 0,
+            floor: 0,
+            entries: VecDeque::new(),
+            gone: HashMap::default(),
+            cursors: Vec::new(),
+            readers: 0,
+        }
+    }
+
+    /// Notes that the best route of `prefix`, which has a row, changed.
+    /// Returns the sequence number for the row to remember.
+    pub(super) fn best_changed(&mut self, prefix: IpPrefix) -> u64 {
+        self.seq += 1;
+        if self.readers > 0 {
+            self.entries.push_back(Entry {
+                seq: self.seq,
+                prefix,
+            });
+        }
+        self.seq
+    }
+
+    /// As [`Journal::best_changed`], for a prefix whose row was just made:
+    /// if it had gone and is back, the note that it had gone is void.
+    pub(super) fn appeared(&mut self, prefix: IpPrefix) -> u64 {
+        if !self.gone.is_empty() {
+            self.gone.remove(&prefix);
+        }
+        self.best_changed(prefix)
+    }
+
+    /// Notes that `prefix` lost its last route and has no row any more.
+    pub(super) fn disappeared(&mut self, prefix: IpPrefix) {
+        self.seq += 1;
+        if self.readers > 0 {
+            self.entries.push_back(Entry {
+                seq: self.seq,
+                prefix,
+            });
+            self.gone.insert(prefix, self.seq);
+        }
+    }
+
+    fn drop_entry(gone: &mut HashMap<IpPrefix, u64, RandomState>, entry: &Entry) {
+        if gone.get(&entry.prefix) == Some(&entry.seq) {
+            gone.remove(&entry.prefix);
+        }
+    }
+
+    pub(super) fn len(&self) -> usize {
+        self.entries.len()
+    }
+}
+
+impl Shard {
+    /// Whether `entry` still says something: it is the latest change of
+    /// its prefix.
+    fn is_live(&self, entry: &Entry) -> bool {
+        let hash = self.hasher.hash_one(entry.prefix);
+        match self.rows.find(hash, |row| row.prefix == entry.prefix) {
+            Some(row) => row.changed == entry.seq,
+            None => self.journal.gone.get(&entry.prefix) == Some(&entry.seq),
+        }
+    }
+
+    /// Keeps the journal from growing. Called after every change to the
+    /// shard; does nothing until there is something worth doing.
+    pub(super) fn tidy_journal(&mut self) {
+        if self.journal.entries.len() <= self.journal.limits.tidy_at {
+            return;
+        }
+        // What every reader has passed can go.
+        let journal = &mut self.journal;
+        let passed = journal
+            .cursors
+            .iter()
+            .map(|cursor| cursor.load(Ordering::Acquire))
+            .min()
+            .unwrap_or(NO_CURSOR);
+        while let Some(front) = journal.entries.front() {
+            if front.seq > passed {
+                break;
+            }
+            Journal::drop_entry(&mut journal.gone, front);
+            journal.entries.pop_front();
+        }
+
+        let limit = 2 * (self.rows.len() + self.journal.limits.slack);
+        if self.journal.entries.len() <= limit {
+            return;
+        }
+        // Dead entries: their prefix has changed again since. This looks
+        // every entry up, but only once the journal has doubled, so the
+        // cost per change stays small.
+        let mut entries = std::mem::take(&mut self.journal.entries);
+        entries.retain(|entry| self.is_live(entry));
+        self.journal.entries = entries;
+
+        if self.journal.entries.len() <= limit {
+            return;
+        }
+        // Still too long: some reader is not reading, and prefixes keep
+        // coming and going. It loses its place.
+        let journal = &mut self.journal;
+        for entry in journal.entries.drain(..limit / 2) {
+            Journal::drop_entry(&mut journal.gone, &entry);
+            journal.floor = entry.seq;
+        }
+    }
+}
+
+/// What a [`Reader`] is given.
+#[derive(Debug, Clone, Copy)]
+pub enum Event<'a> {
+    /// Forget what you know. The whole table follows, as [`Event::Best`]
+    /// for every prefix, and then [`Event::Synced`]. The first thing a new
+    /// reader is given, and what a reader that fell too far behind is
+    /// given in place of what it missed.
+    Reset,
+    /// This is now the best route for `prefix`.
+    Best {
+        prefix: IpPrefix,
+        peer: PeerIndex,
+        attrs: &'a Arc<AttrSet>,
+    },
+    /// `prefix` has no route any more.
+    Withdrawn { prefix: IpPrefix },
+    /// The table that began with [`Event::Reset`] is complete: a prefix
+    /// not heard of since the reset has no route.
+    Synced,
+}
+
+/// What a call to [`Reader::poll`] got done.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct Progress {
+    /// How many events were given out.
+    pub events: usize,
+    /// Whether the reader has now been given everything there was when the
+    /// call began. If not, call again.
+    pub caught_up: bool,
+}
+
+enum Phase {
+    /// Giving out the whole table; `next` is the shard to do next.
+    Resync { next: usize, announced: bool },
+    /// Following the journals; `next` is the shard to read next.
+    Follow { next: usize },
+}
+
+/// Follows the best routes of a table. Made by [`RibTable::reader`].
+///
+/// A reader holds its place in every shard's journal. Dropping it gives
+/// the place up, so that the journals need not keep entries for it.
+pub struct Reader {
+    table: Arc<RibTable>,
+    slot: usize,
+    /// Where this reader has read to in each shard; the copy in the shard
+    /// is for the shard's benefit.
+    cursors: Vec<u64>,
+    phase: Phase,
+    /// The table's version when the reader was last found caught up.
+    seen: u64,
+    caught_up: bool,
+}
+
+impl RibTable {
+    /// A new follower of this table's best routes. It is given the whole
+    /// table first and every change after.
+    pub fn reader(self: &Arc<Self>) -> Reader {
+        let mut slots = self
+            .reader_slots
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let slot = match slots.iter().position(|used| !used) {
+            Some(free) => free,
+            None => {
+                slots.push(false);
+                slots.len() - 1
+            }
+        };
+        slots[slot] = true;
+        for shard in &self.shards {
+            let journal = &mut write(shard).journal;
+            while journal.cursors.len() <= slot {
+                journal.cursors.push(AtomicU64::new(NO_CURSOR));
+            }
+            journal.cursors[slot].store(NO_CURSOR, Ordering::Release);
+            journal.readers += 1;
+        }
+        self.readers.fetch_add(1, Ordering::AcqRel);
+        Reader {
+            table: Arc::clone(self),
+            slot,
+            cursors: vec![NO_CURSOR; self.shards.len()],
+            phase: Phase::Resync {
+                next: 0,
+                announced: false,
+            },
+            seen: 0,
+            caught_up: false,
+        }
+    }
+
+    /// How many entries the journals hold, over all shards. For tests and
+    /// for watching: it stays bounded whatever the readers do.
+    pub fn journal_len(&self) -> usize {
+        self.shards
+            .iter()
+            .map(|shard| read(shard).journal.len())
+            .sum()
+    }
+
+    /// Tells readers that are waiting that there is something to read.
+    pub(super) fn wake_readers(&self) {
+        if self.readers.load(Ordering::Acquire) > 0 {
+            self.version.fetch_add(1, Ordering::AcqRel);
+            self.wake.notify_waiters();
+        }
+    }
+}
+
+impl Drop for Reader {
+    fn drop(&mut self) {
+        for shard in &self.table.shards {
+            let journal = &mut write(shard).journal;
+            journal.cursors[self.slot].store(NO_CURSOR, Ordering::Release);
+            journal.readers -= 1;
+            if journal.readers == 0 {
+                // Nobody left to read it.
+                journal.entries.clear();
+                journal.gone.clear();
+            }
+        }
+        self.table.readers.fetch_sub(1, Ordering::AcqRel);
+        let mut slots = self
+            .table
+            .reader_slots
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        slots[self.slot] = false;
+    }
+}
+
+enum Drained {
+    /// Entries were dropped that this reader had not read.
+    Lagged,
+    /// `events` were given out; `more` if the budget ran out first.
+    Read { events: usize, more: bool },
+}
+
+impl Reader {
+    /// Gives `visit` what has changed since the last call, in up to about
+    /// `budget` events (a shard's part of a resynchronisation is not cut
+    /// short, so it can be more).
+    ///
+    /// `visit` is called with a shard of the table held for reading. It
+    /// should be quick, and it must not change the table.
+    pub fn poll(&mut self, budget: usize, mut visit: impl FnMut(Event<'_>)) -> Progress {
+        let shards = self.table.shards.len();
+        let mut events = 0;
+        loop {
+            match self.phase {
+                Phase::Resync { next, announced } => {
+                    if !announced {
+                        self.forget_place();
+                        visit(Event::Reset);
+                        events += 1;
+                    }
+                    if next == shards {
+                        visit(Event::Synced);
+                        events += 1;
+                        self.phase = Phase::Follow { next: 0 };
+                        continue;
+                    }
+                    events += self.scan(next, &mut visit);
+                    self.phase = Phase::Resync {
+                        next: next + 1,
+                        announced: true,
+                    };
+                    if events >= budget {
+                        return self.progress(events, false);
+                    }
+                }
+                Phase::Follow { next } => {
+                    // Read before the pass: a change made during it shows as
+                    // a newer version, and `changed` then returns at once.
+                    let version = if next == 0 {
+                        self.table.version.load(Ordering::Acquire)
+                    } else {
+                        self.seen
+                    };
+                    self.seen = version;
+                    let mut shard = next;
+                    let mut lagged = false;
+                    while shard < shards {
+                        if events >= budget {
+                            self.phase = Phase::Follow { next: shard };
+                            return self.progress(events, false);
+                        }
+                        match self.drain(shard, budget - events, &mut visit) {
+                            Drained::Lagged => {
+                                lagged = true;
+                                break;
+                            }
+                            Drained::Read { events: read, more } => {
+                                events += read;
+                                if more {
+                                    self.phase = Phase::Follow { next: shard };
+                                    return self.progress(events, false);
+                                }
+                            }
+                        }
+                        shard += 1;
+                    }
+                    if lagged {
+                        self.phase = Phase::Resync {
+                            next: 0,
+                            announced: false,
+                        };
+                        continue;
+                    }
+                    self.phase = Phase::Follow { next: 0 };
+                    if next != 0 {
+                        // This pass took up where an earlier call stopped.
+                        // The shards before that point have not been looked
+                        // at in this call, so go round again from the first.
+                        continue;
+                    }
+                    return self.progress(events, true);
+                }
+            }
+        }
+    }
+
+    fn progress(&mut self, events: usize, caught_up: bool) -> Progress {
+        self.caught_up = caught_up;
+        Progress { events, caught_up }
+    }
+
+    /// Waits until there may be something to read. Returns at once if the
+    /// last [`Reader::poll`] did not catch up.
+    pub async fn changed(&mut self) {
+        loop {
+            // Registered before the check, so that a change between the
+            // check and the wait is not missed.
+            let woken = self.table.wake.notified();
+            tokio::pin!(woken);
+            woken.as_mut().enable();
+            if !self.caught_up || self.table.version.load(Ordering::Acquire) != self.seen {
+                return;
+            }
+            woken.await;
+        }
+    }
+
+    /// Gives up this reader's place in every journal, before a
+    /// resynchronisation takes a new one.
+    fn forget_place(&mut self) {
+        for (index, shard) in self.table.shards.iter().enumerate() {
+            read(shard).journal.cursors[self.slot].store(NO_CURSOR, Ordering::Release);
+            self.cursors[index] = NO_CURSOR;
+        }
+    }
+
+    fn set_cursor(&mut self, index: usize, shard: &Shard, cursor: u64) {
+        self.cursors[index] = cursor;
+        shard.journal.cursors[self.slot].store(cursor, Ordering::Release);
+    }
+
+    /// Gives out every best route in a shard and takes a place at the end
+    /// of its journal, in one hold of the shard: what changes afterwards is
+    /// in the journal after that place.
+    fn scan(&mut self, index: usize, visit: &mut impl FnMut(Event<'_>)) -> usize {
+        let table = Arc::clone(&self.table);
+        let shard = read(&table.shards[index]);
+        for row in &shard.rows {
+            let best = &row.paths[usize::from(row.best)];
+            visit(Event::Best {
+                prefix: row.prefix,
+                peer: best.peer(),
+                attrs: &best.attrs,
+            });
+        }
+        self.set_cursor(index, &shard, shard.journal.seq);
+        shard.rows.len()
+    }
+
+    /// Reads a shard's journal from this reader's place, for at most `max`
+    /// events.
+    fn drain(&mut self, index: usize, max: usize, visit: &mut impl FnMut(Event<'_>)) -> Drained {
+        let table = Arc::clone(&self.table);
+        let shard = read(&table.shards[index]);
+        let journal = &shard.journal;
+        let mut cursor = self.cursors[index];
+        if cursor < journal.floor {
+            return Drained::Lagged;
+        }
+        if cursor == journal.seq {
+            return Drained::Read {
+                events: 0,
+                more: false,
+            };
+        }
+
+        let start = journal.entries.partition_point(|entry| entry.seq <= cursor);
+        let mut events = 0;
+        let mut more = false;
+        for entry in journal.entries.range(start..) {
+            if events == max {
+                more = true;
+                break;
+            }
+            cursor = entry.seq;
+            let hash = shard.hasher.hash_one(entry.prefix);
+            match shard.rows.find(hash, |row| row.prefix == entry.prefix) {
+                Some(row) if row.changed == entry.seq => {
+                    let best = &row.paths[usize::from(row.best)];
+                    visit(Event::Best {
+                        prefix: row.prefix,
+                        peer: best.peer(),
+                        attrs: &best.attrs,
+                    });
+                    events += 1;
+                }
+                None if journal.gone.get(&entry.prefix) == Some(&entry.seq) => {
+                    visit(Event::Withdrawn {
+                        prefix: entry.prefix,
+                    });
+                    events += 1;
+                }
+                // The prefix has changed again; a later entry has it.
+                _ => {}
+            }
+        }
+        if !more {
+            // Past the last entry there is nothing up to the latest change.
+            cursor = journal.seq;
+        }
+        self.set_cursor(index, &shard, cursor);
+        Drained::Read { events, more }
+    }
+}
+
+#[cfg(test)]
+#[path = "journal_tests.rs"]
+mod tests;

crates/rib/src/table/journal_tests.rs

@@ -0,0 +1,466 @@
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+
+use bgp_primitives::constants::AddressFamilyId;
+use bgp_primitives::ip_prefix::IpPrefix;
+use proptest::prelude::*;
+
+use super::super::tests::{Fixture, Route, external, internal, prefix, route};
+use super::super::{AttrSet, PeerIndex, PeerInfo, RibTable};
+use super::{Event, Limits, Reader};
+
+/// What a follower would keep: its own copy of the best routes, brought
+/// up to date from the events it is given.
+#[derive(Default)]
+struct Mirror {
+    best: HashMap<IpPrefix, (PeerIndex, Arc<AttrSet>)>,
+    /// What was held before a reset and has not been heard of since.
+    stale: HashMap<IpPrefix, (PeerIndex, Arc<AttrSet>)>,
+    resets: usize,
+    synced: usize,
+    best_events: usize,
+    withdrawn_events: usize,
+}
+
+impl Mirror {
+    fn apply(&mut self, event: Event<'_>) {
+        match event {
+            Event::Reset => {
+                self.resets += 1;
+                self.stale = std::mem::take(&mut self.best);
+            }
+            Event::Best {
+                prefix,
+                peer,
+                attrs,
+            } => {
+                self.best_events += 1;
+                self.stale.remove(&prefix);
+                self.best.insert(prefix, (peer, Arc::clone(attrs)));
+            }
+            Event::Withdrawn { prefix } => {
+                self.withdrawn_events += 1;
+                self.best.remove(&prefix);
+            }
+            Event::Synced => {
+                self.synced += 1;
+                self.stale.clear();
+            }
+        }
+    }
+
+    /// Reads until there is nothing more, in steps of `budget`.
+    fn catch_up(&mut self, reader: &mut Reader, budget: usize) {
+        while !reader.poll(budget, |event| self.apply(event)).caught_up {}
+    }
+
+    #[track_caller]
+    fn assert_matches(&self, table: &RibTable) {
+        let mut expected = HashMap::new();
+        table.for_each_best(|prefix, peer, attrs| {
+            expected.insert(*prefix, (peer, Arc::clone(attrs)));
+        });
+        assert_eq!(
+            self.best.len(),
+            expected.len(),
+            "the mirror has {} prefixes, the table {}",
+            self.best.len(),
+            expected.len()
+        );
+        for (prefix, (peer, attrs)) in &expected {
+            let Some((mirrored_peer, mirrored_attrs)) = self.best.get(prefix) else {
+                panic!("{prefix} is in the table and not in the mirror");
+            };
+            assert_eq!(mirrored_peer, peer, "{prefix}: from the wrong peer");
+            assert!(
+                Arc::ptr_eq(mirrored_attrs, attrs),
+                "{prefix}: wrong attributes"
+            );
+        }
+    }
+}
+
+fn nth_prefix(n: u32) -> IpPrefix {
+    IpPrefix::new(AddressFamilyId::Ipv4, (n << 8).to_be_bytes(), 24).unwrap()
+}
+
+#[test]
+fn a_new_reader_is_given_the_table_and_then_the_changes() {
+    let fixture = Fixture::new(&[external(1), external(2)]);
+    let (a, b) = (prefix("10.0.0.0/8"), prefix("2001:db8::/32"));
+    fixture.announce(0, &a, &Route::via(&[1, 2]));
+    fixture.announce(0, &b, &Route::via(&[1, 2]));
+
+    let mut reader = fixture.table.reader();
+    let mut mirror = Mirror::default();
+    let mut events = Vec::new();
+    let progress = reader.poll(100, |event| {
+        events.push(
+            format!("{event:?}")
+                .split([' ', '{'])
+                .next()
+                .unwrap()
+                .to_string(),
+        );
+        mirror.apply(event);
+    });
+    assert!(progress.caught_up);
+    assert_eq!(events, ["Reset", "Best", "Best", "Synced"]);
+    mirror.assert_matches(&fixture.table);
+
+    // Nothing has happened: nothing to read.
+    assert_eq!(reader.poll(100, |_| panic!("no event expected")).events, 0);
+
+    // A change that does not change the best route is not an event.
+    fixture.announce(1, &a, &Route::via(&[3, 4, 5]));
+    assert_eq!(reader.poll(100, |_| panic!("no event expected")).events, 0);
+
+    fixture.announce(1, &a, &Route::via(&[3]));
+    fixture.withdraw(0, &b);
+    mirror.catch_up(&mut reader, 100);
+    assert_eq!((mirror.best_events, mirror.withdrawn_events), (3, 1));
+    assert_eq!((mirror.resets, mirror.synced), (1, 1));
+    mirror.assert_matches(&fixture.table);
+}
+
+/// A follower wants what the table has now, not how it got there.
+#[test]
+fn many_changes_to_a_prefix_between_reads_are_one_event() {
+    let fixture = Fixture::new(&[external(1), external(2)]);
+    let p = prefix("10.0.0.0/8");
+    let mut reader = fixture.table.reader();
+    let mut mirror = Mirror::default();
+    mirror.catch_up(&mut reader, 100);
+
+    for round in 0..50 {
+        fixture.announce(0, &p, &Route::via(&[1, 2, 3]));
+        fixture.announce(1, &p, &Route::via(&[4, 5]));
+        fixture.withdraw(1, &p);
+        if round % 2 == 0 {
+            fixture.withdraw(0, &p);
+        }
+    }
+    // It ended with peer 0's route in place.
+    mirror.catch_up(&mut reader, 100);
+    assert_eq!((mirror.best_events, mirror.withdrawn_events), (1, 0));
+    mirror.assert_matches(&fixture.table);
+
+    // And a prefix that came and went again is, at most, a withdrawal.
+    let q = prefix("10.1.0.0/16");
+    for _ in 0..50 {
+        fixture.announce(0, &q, &Route::via(&[1]));
+        fixture.withdraw(0, &q);
+    }
+    mirror.catch_up(&mut reader, 100);
+    assert_eq!((mirror.best_events, mirror.withdrawn_events), (1, 1));
+    mirror.assert_matches(&fixture.table);
+}
+
+#[test]
+fn reading_in_small_steps_gives_the_same_as_reading_at_once() {
+    let fixture = Fixture::new(&[external(1)]);
+    for n in 0..500 {
+        fixture.announce(0, &nth_prefix(n), &Route::via(&[1, n % 7 + 1]));
+    }
+    let mut reader = fixture.table.reader();
+    let mut mirror = Mirror::default();
+    mirror.catch_up(&mut reader, 7);
+    mirror.assert_matches(&fixture.table);
+
+    for n in 0..500 {
+        if n % 3 == 0 {
+            fixture.withdraw(0, &nth_prefix(n));
+        } else {
+            fixture.announce(0, &nth_prefix(n), &Route::via(&[2, n % 5 + 1]));
+        }
+    }
+    let mut steps = 0;
+    while !reader.poll(7, |event| mirror.apply(event)).caught_up {
+        steps += 1;
+    }
+    assert!(steps > 50, "the budget was kept to: {steps} steps");
+    mirror.assert_matches(&fixture.table);
+    assert_eq!(mirror.resets, 1);
+}
+
+#[test]
+fn with_no_reader_nothing_is_kept() {
+    let fixture = Fixture::new(&[external(1)]);
+    for n in 0..5000 {
+        fixture.announce(0, &nth_prefix(n), &Route::via(&[1]));
+        fixture.withdraw(0, &nth_prefix(n));
+    }
+    assert_eq!(fixture.table.journal_len(), 0);
+
+    // And what a reader was holding back goes when the reader does.
+    let reader = fixture.table.reader();
+    for n in 0..100 {
+        fixture.announce(0, &nth_prefix(n), &Route::via(&[1]));
+    }
+    assert_eq!(fixture.table.journal_len(), 100);
+    drop(reader);
+    assert_eq!(fixture.table.journal_len(), 0);
+}
+
+/// A reader that stops reading must not make the table keep a growing
+/// journal for it. It is the one that pays: it is given the table again.
+#[test]
+fn a_reader_that_stops_reading_is_reset_and_holds_nothing_up() {
+    // One shard, so that the journal's length is that shard's.
+    let fixture = Fixture::on(RibTable::new(1), &[external(1)]);
+    fixture.announce(0, &prefix("192.0.2.0/24"), &Route::via(&[1]));
+
+    let mut stuck = fixture.table.reader();
+    let mut stuck_mirror = Mirror::default();
+    stuck_mirror.catch_up(&mut stuck, 100);
+    let mut diligent = fixture.table.reader();
+    let mut diligent_mirror = Mirror::default();
+
+    // Prefixes come and go, each a different one, so no entry is made
+    // dead by a later one for the same prefix.
+    let limits = Limits::default();
+    let mut longest = 0;
+    for n in 0..20_000 {
+        fixture.announce(0, &nth_prefix(n), &Route::via(&[1]));
+        fixture.withdraw(0, &nth_prefix(n));
+        longest = longest.max(fixture.table.journal_len());
+        if n % 100 == 0 {
+            diligent_mirror.catch_up(&mut diligent, 1000);
+        }
+    }
+    assert!(
+        longest <= limits.tidy_at.max(2 * (1 + limits.slack)) + 2,
+        "the journal reached {longest} entries"
+    );
+
+    // The one that kept up never noticed.
+    diligent_mirror.catch_up(&mut diligent, 1000);
+    assert_eq!(
+        diligent_mirror.resets, 1,
+        "only the reset every reader starts with"
+    );
+    diligent_mirror.assert_matches(&fixture.table);
+
+    // The one that did not is told, and ends up right all the same.
+    stuck_mirror.catch_up(&mut stuck, 1000);
+    assert_eq!(stuck_mirror.resets, 2);
+    assert_eq!(stuck_mirror.synced, 2);
+    stuck_mirror.assert_matches(&fixture.table);
+}
+
+#[tokio::test]
+async fn a_waiting_reader_is_woken_by_a_change_of_best_route() {
+    let fixture = Fixture::new(&[external(1), external(2)]);
+    let p = prefix("10.0.0.0/8");
+    let mut reader = fixture.table.reader();
+    let mut mirror = Mirror::default();
+
+    // Not caught up yet: no waiting.
+    reader.changed().await;
+    mirror.catch_up(&mut reader, 100);
+
+    let waiting = tokio::time::timeout(std::time::Duration::from_millis(50), reader.changed());
+    assert!(waiting.await.is_err(), "nothing changed, so it waits");
+
+    let table = Arc::clone(&fixture.table);
+    let attrs = table
+        .attrs()
+        .intern(&Route::via(&[1]).attributes(), None)
+        .unwrap();
+    let peer = fixture.peers[0];
+    let writer = tokio::spawn(async move {
+        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+        table.announce(peer, &attrs, &[p]);
+    });
+    tokio::time::timeout(std::time::Duration::from_secs(5), reader.changed())
+        .await
+        .expect("woken by the announcement");
+    writer.await.unwrap();
+    mirror.catch_up(&mut reader, 100);
+    mirror.assert_matches(&fixture.table);
+
+    // A change made before the wait began is not missed either.
+    fixture.withdraw(0, &p);
+    tokio::time::timeout(std::time::Duration::from_secs(5), reader.changed())
+        .await
+        .expect("the change was already there");
+}
+
+/// Writers on several threads, a reader on another, reading while they
+/// write. When the writers are done and the reader has caught up, it has
+/// what the table has.
+#[test]
+fn a_reader_keeps_up_with_writers_on_other_threads() {
+    const WRITERS: usize = 4;
+    const PREFIXES: u32 = 300;
+    let infos: Vec<PeerInfo> = (0..WRITERS).map(|i| external(i as u8 + 1)).collect();
+    // Small journals, so that tidying and falling behind happen too.
+    let fixture = Fixture::on(
+        RibTable::with_journal_limits(
+            8,
+            Limits {
+                tidy_at: 16,
+                slack: 4,
+            },
+        ),
+        &infos,
+    );
+    let table = &fixture.table;
+    let done = AtomicBool::new(false);
+    let mut mirror = Mirror::default();
+
+    std::thread::scope(|scope| {
+        let writers: Vec<_> = (0..WRITERS)
+            .map(|writer| {
+                let peer = fixture.peers[writer];
+                scope.spawn(move || {
+                    // A simple generator; each writer its own sequence.
+                    let mut state = 0x9e37_79b9_7f4a_7c15u64.wrapping_mul(writer as u64 + 1);
+                    let mut next = || {
+                        state ^= state << 13;
+                        state ^= state >> 7;
+                        state ^= state << 17;
+                        state
+                    };
+                    for _ in 0..30_000 {
+                        let p = nth_prefix((next() % u64::from(PREFIXES)) as u32);
+                        if next() % 3 == 0 {
+                            table.withdraw(peer, &[p]);
+                        } else {
+                            let route = Route {
+                                med: Some((next() % 3) as u32),
+                                ..Route::via(
+                                    &[(next() % 3) as u32 + 1, 9][..(next() % 2 + 1) as usize],
+                                )
+                            };
+                            let attrs = table.attrs().intern(&route.attributes(), None).unwrap();
+                            table.announce(peer, &attrs, &[p]);
+                        }
+                    }
+                })
+            })
+            .collect();
+
+        let mirror = &mut mirror;
+        let done = &done;
+        let reading = scope.spawn(move || {
+            let mut reader = table.reader();
+            while !done.load(Ordering::Acquire) {
+                reader.poll(50, |event| mirror.apply(event));
+            }
+            mirror.catch_up(&mut reader, 50);
+        });
+        for writer in writers {
+            writer.join().unwrap();
+        }
+        done.store(true, Ordering::Release);
+        reading.join().unwrap();
+    });
+
+    mirror.assert_matches(table);
+    assert!(mirror.best_events > 0);
+}
+
+// ---------------------------------------------------------------------------
+// Readers against the table, under any interleaving
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone)]
+enum Op {
+    Announce {
+        peer: usize,
+        prefix: usize,
+        route: Route,
+    },
+    Withdraw {
+        peer: usize,
+        prefix: usize,
+    },
+    Flap {
+        peer: usize,
+    },
+    Poll {
+        reader: usize,
+        budget: usize,
+    },
+    /// The reader is dropped and a new one takes its place.
+    Replace {
+        reader: usize,
+    },
+}
+
+const PEERS: usize = 4;
+const PREFIXES: usize = 12;
+const READERS: usize = 3;
+
+fn op() -> impl Strategy<Value = Op> {
+    prop_oneof![
+        8 => (0..PEERS, 0..PREFIXES, route())
+            .prop_map(|(peer, prefix, route)| Op::Announce { peer, prefix, route }),
+        5 => (0..PEERS, 0..PREFIXES).prop_map(|(peer, prefix)| Op::Withdraw { peer, prefix }),
+        1 => (0..PEERS).prop_map(|peer| Op::Flap { peer }),
+        // Reader 0 is polled rarely: it is the one that falls behind.
+        4 => (1..READERS, 1usize..6).prop_map(|(reader, budget)| Op::Poll { reader, budget }),
+        1 => (0..READERS, 1usize..6).prop_map(|(reader, budget)| Op::Poll { reader, budget }),
+        1 => (1..READERS).prop_map(|reader| Op::Replace { reader }),
+    ]
+}
+
+proptest! {
+    #![proptest_config(ProptestConfig::with_cases(1500))]
+
+    /// However changes and reads are interleaved, and however small the
+    /// journals: a reader that has caught up holds exactly the table's best
+    /// routes, and every reader gets there in the end.
+    #[test]
+    fn a_reader_that_has_caught_up_has_what_the_table_has(
+        ops in proptest::collection::vec(op(), 1..150),
+        shards in 1usize..4,
+    ) {
+        let infos: Vec<PeerInfo> = (0..PEERS)
+            .map(|i| if i == 0 { internal(9) } else { external(i as u8) })
+            .collect();
+        // Journals that are tidied, and found too long, almost at once.
+        let limits = Limits { tidy_at: 3, slack: 1 };
+        let mut fixture = Fixture::on(RibTable::with_journal_limits(shards, limits), &infos);
+        let prefixes: Vec<IpPrefix> = (0..PREFIXES as u32).map(nth_prefix).collect();
+        let mut readers: Vec<(Reader, Mirror)> = (0..READERS)
+            .map(|_| (fixture.table.reader(), Mirror::default()))
+            .collect();
+
+        for op in ops {
+            match op {
+                Op::Announce { peer, prefix, route } => {
+                    fixture.announce(peer, &prefixes[prefix], &route);
+                }
+                Op::Withdraw { peer, prefix } => {
+                    fixture.withdraw(peer, &prefixes[prefix]);
+                }
+                Op::Flap { peer } => {
+                    fixture.table.remove_peer(fixture.peers[peer]);
+                    fixture.peers[peer] = fixture.table.add_peer(infos[peer]).unwrap();
+                }
+                Op::Poll { reader, budget } => {
+                    let (reader, mirror) = &mut readers[reader];
+                    let progress = reader.poll(budget, |event| mirror.apply(event));
+                    if progress.caught_up {
+                        mirror.assert_matches(&fixture.table);
+                    }
+                }
+                Op::Replace { reader } => {
+                    readers[reader] = (fixture.table.reader(), Mirror::default());
+                }
+            }
+            // Never much longer than the table, whatever the readers do.
+            let bound = shards * (2 * (PREFIXES + limits.slack) + limits.tidy_at + PREFIXES);
+            prop_assert!(fixture.table.journal_len() <= bound);
+        }
+
+        for (reader, mirror) in &mut readers {
+            mirror.catch_up(reader, 5);
+            mirror.assert_matches(&fixture.table);
+            prop_assert_eq!(mirror.resets, mirror.synced);
+        }
+    }
+}

crates/rib/src/table/mod.rs

@@ -21,22 +21,28 @@
 //!   is slowed down, which is the back-pressure that is wanted.
 //! - **Readers do not stop the table.** Reading takes a shard's lock for
 //!   reading, one shard at a time, and looks at the rows in place.
+//! - **Followers go at their own pace.** What acts on the best routes reads
+//!   the changes from a [`journal`], and a slow reader holds nothing up.
 
 pub mod attrs;
+mod journal;
 mod select;
 #[cfg(test)]
 mod tests;
 
 use std::hash::BuildHasher;
 use std::net::{IpAddr, Ipv4Addr};
+use std::sync::atomic::{AtomicU64, AtomicUsize};
 use std::sync::{Arc, Mutex, PoisonError, RwLock, RwLockReadGuard, RwLockWriteGuard};
 
 use bgp_primitives::ip_prefix::IpPrefix;
 use foldhash::fast::RandomState;
 use hashbrown::HashTable;
 use hashbrown::hash_table::Entry;
+use tokio::sync::Notify;
 
 pub use attrs::{AttrSet, AttrStore, SelectionKey};
+pub use journal::{Event, Progress, Reader};
 
 /// What route selection needs to know about a peer.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -118,6 +124,9 @@ struct Row {
     /// Index of the best route in `paths`. A row always has at least one
     /// route, and so always a best one.
     best: u16,
+    /// The journal's sequence number for the latest change of this row's
+    /// best route; see [`journal`].
+    changed: u64,
     /// The routes, one from each peer that has announced the prefix, in one
     /// run of memory: finding a peer's route and choosing the best are
     /// scans of it.
@@ -139,6 +148,8 @@ struct Shard {
     hasher: RandomState,
     /// Reused between selections.
     scratch: select::Scratch,
+    /// Which prefixes' best routes have changed, for those who follow.
+    journal: journal::Journal,
 }
 
 /// The routing table. Shared between threads by reference; every method
@@ -150,6 +161,13 @@ pub struct RibTable {
     /// Which peer numbers are in use. Changing the peer list also takes
     /// every shard's lock in turn; this serialises that.
     peers: Mutex<Vec<Option<PeerInfo>>>,
+    /// Which reader slots are in use; see [`journal`].
+    reader_slots: Mutex<Vec<bool>>,
+    /// How many readers there are, to skip waking them when there are none.
+    readers: AtomicUsize,
+    /// Goes up whenever a best route changes while there are readers.
+    version: AtomicU64,
+    wake: Notify,
 }
 
 /// No more peers can be added: every [`PeerIndex`] is in use.
@@ -161,6 +179,12 @@ impl RibTable {
     /// A table in `shards` shards. More shards let more sessions write at
     /// once; a few times the number of cores is plenty.
     pub fn new(shards: usize) -> Self {
+        Self::with_journal_limits(shards, journal::Limits::default())
+    }
+
+    /// A table whose journals are tidied, and found too long, at the given
+    /// sizes. Tests use small ones to get there quickly.
+    fn with_journal_limits(shards: usize, limits: journal::Limits) -> Self {
         let hasher = RandomState::default();
         Self {
             shards: (0..shards.max(1))
@@ -171,12 +195,17 @@ impl RibTable {
                         peers: Vec::new(),
                         hasher: hasher.clone(),
                         scratch: select::Scratch::default(),
+                        journal: journal::Journal::new(limits),
                     })
                 })
                 .collect(),
             hasher,
             attrs: AttrStore::new(),
             peers: Mutex::default(),
+            reader_slots: Mutex::default(),
+            readers: AtomicUsize::new(0),
+            version: AtomicU64::new(0),
+            wake: Notify::new(),
         }
     }
 
@@ -192,7 +221,7 @@ impl RibTable {
         let index = match peers.iter().position(Option::is_none) {
             Some(free) => free,
             None => {
-                // u16::MAX is kept back: see `select::NONE`.
+                // A peer number shares a word with other things; see `Path`.
                 if peers.len() >= usize::from(u16::MAX) {
                     return Err(TooManyPeers);
                 }
@@ -220,6 +249,7 @@ impl RibTable {
         for shard in &self.shards {
             let mut shard = write(shard);
             outcome += shard.remove_peer(peer);
+            shard.tidy_journal();
             if let Some(slot) = shard.peers.get_mut(peer.as_usize()) {
                 *slot = None;
             }
@@ -227,6 +257,9 @@ impl RibTable {
         if let Some(slot) = peers.get_mut(peer.as_usize()) {
             *slot = None;
         }
+        if outcome.best_changed > 0 {
+            self.wake_readers();
+        }
         outcome
     }
 
@@ -252,6 +285,9 @@ impl RibTable {
                 }
             }
         });
+        if outcome.best_changed > 0 {
+            self.wake_readers();
+        }
         outcome
     }
 
@@ -266,6 +302,9 @@ impl RibTable {
                 }
             }
         });
+        if outcome.best_changed > 0 {
+            self.wake_readers();
+        }
         outcome
     }
 
@@ -296,6 +335,7 @@ impl RibTable {
         for run in placed.chunk_by(|a, b| shard_of(a.0) == shard_of(b.0)) {
             let mut shard = write(&self.shards[shard_of(run[0].0) as usize]);
             apply(&mut shard, run);
+            shard.tidy_journal();
         }
     }
 
@@ -421,6 +461,7 @@ impl Shard {
                 vacant.insert(Row {
                     prefix,
                     best: 0,
+                    changed: self.journal.appeared(prefix),
                     paths: vec![path],
                 });
                 self.routes += 1;
@@ -456,6 +497,9 @@ impl Shard {
                 } else {
                     select::choose(row, &self.peers, &mut self.scratch)
                 };
+                if best_changed {
+                    row.changed = self.journal.best_changed(prefix);
+                }
                 Outcome {
                     added: 1,
                     best_changed: best_changed.into(),
@@ -484,6 +528,9 @@ impl Shard {
                 } else {
                     select::choose(row, &self.peers, &mut self.scratch)
                 };
+                if best_changed {
+                    row.changed = self.journal.best_changed(prefix);
+                }
                 Outcome {
                     replaced: 1,
                     best_changed: best_changed.into(),
@@ -504,6 +551,7 @@ impl Shard {
         self.routes -= 1;
         if row.paths.len() == 1 {
             entry.remove();
+            self.journal.disappeared(*prefix);
             return Outcome {
                 removed: 1,
                 best_changed: 1,
@@ -511,6 +559,9 @@ impl Shard {
             };
         }
         let best_changed = Self::remove_path(row, at, &self.peers, &mut self.scratch);
+        if best_changed {
+            row.changed = self.journal.best_changed(*prefix);
+        }
         Outcome {
             removed: 1,
             best_changed: best_changed.into(),
@@ -551,6 +602,7 @@ impl Shard {
         let mut outcome = Outcome::default();
         let peers = &self.peers;
         let scratch = &mut self.scratch;
+        let journal = &mut self.journal;
         self.rows.retain(|row| {
             let Some(at) = row.position(peer) else {
                 return true;
@@ -558,10 +610,12 @@ impl Shard {
             outcome.removed += 1;
             if row.paths.len() == 1 {
                 outcome.best_changed += 1;
+                journal.disappeared(row.prefix);
                 return false;
             }
             if Self::remove_path(row, at, peers, scratch) {
                 outcome.best_changed += 1;
+                row.changed = journal.best_changed(row.prefix);
             }
             true
         });

crates/rib/src/table/tests.rs

@@ -11,21 +11,21 @@ use proptest::prelude::*;
 
 use super::*;
 
-fn prefix(s: &str) -> IpPrefix {
+pub(super) fn prefix(s: &str) -> IpPrefix {
     IpPrefix::try_from(s).unwrap()
 }
 
 /// What a test says about a route; everything selection looks at.
 #[derive(Debug, Clone, PartialEq, Eq)]
-struct Route {
+pub(super) struct Route {
-    local_pref: Option<u32>,
+    pub(super) local_pref: Option<u32>,
-    path: Vec<u32>,
+    pub(super) path: Vec<u32>,
-    origin: u8,
+    pub(super) origin: u8,
-    med: Option<u32>,
+    pub(super) med: Option<u32>,
 }
 
 impl Route {
-    fn via(path: &[u32]) -> Self {
+    pub(super) fn via(path: &[u32]) -> Self {
         Self {
             local_pref: None,
             path: path.to_vec(),
@@ -34,7 +34,7 @@ impl Route {
         }
     }
 
-    fn attributes(&self) -> Vec<PathAttribute> {
+    pub(super) fn attributes(&self) -> Vec<PathAttribute> {
         let mut attributes = vec![
             PathAttribute::Origin(match self.origin {
                 0 => OriginPathAttribute::IGP,
@@ -56,7 +56,7 @@ impl Route {
     }
 }
 
-fn external(number: u8) -> PeerInfo {
+pub(super) fn external(number: u8) -> PeerInfo {
     PeerInfo {
         asn: 65000 + u32::from(number),
         router_id: Ipv4Addr::new(10, 0, 0, number),
@@ -65,7 +65,7 @@ fn external(number: u8) -> PeerInfo {
     }
 }
 
-fn internal(number: u8) -> PeerInfo {
+pub(super) fn internal(number: u8) -> PeerInfo {
     PeerInfo {
         asn: 64999,
         ibgp: true,
@@ -75,20 +75,24 @@ fn internal(number: u8) -> PeerInfo {
 
 /// A table and the peers added to it, for tests that speak of peers by the
 /// order they were added in.
-struct Fixture {
+pub(super) struct Fixture {
-    table: RibTable,
+    pub(super) table: Arc<RibTable>,
-    peers: Vec<PeerIndex>,
+    pub(super) peers: Vec<PeerIndex>,
 }
 
 impl Fixture {
-    fn new(peers: &[PeerInfo]) -> Self {
+    pub(super) fn new(peers: &[PeerInfo]) -> Self {
         // More than one shard, so that a test's prefixes are spread.
-        let table = RibTable::new(4);
+        Self::on(RibTable::new(4), peers)
+    }
+
+    pub(super) fn on(table: RibTable, peers: &[PeerInfo]) -> Self {
+        let table = Arc::new(table);
         let peers = peers.iter().map(|p| table.add_peer(*p).unwrap()).collect();
         Self { table, peers }
     }
 
-    fn announce(&self, peer: usize, prefix: &IpPrefix, route: &Route) -> Outcome {
+    pub(super) fn announce(&self, peer: usize, prefix: &IpPrefix, route: &Route) -> Outcome {
         let attrs = self
             .table
             .attrs()
@@ -97,7 +101,7 @@ impl Fixture {
         self.table.announce(self.peers[peer], &attrs, &[*prefix])
     }
 
-    fn withdraw(&self, peer: usize, prefix: &IpPrefix) -> Outcome {
+    pub(super) fn withdraw(&self, peer: usize, prefix: &IpPrefix) -> Outcome {
         self.table.withdraw(self.peers[peer], &[*prefix])
     }
 
@@ -478,7 +482,7 @@ const PEERS: usize = 6;
 // Few prefixes, so that each gathers routes from most peers.
 const PREFIXES: usize = 2;
 
-fn route() -> impl Strategy<Value = Route> {
+pub(super) fn route() -> impl Strategy<Value = Route> {
     // Few values of everything, so that ties at every step are common:
     // ties are where the shortcuts could be wrong.
     (