crates/jet/src/swarm.rs at f3b5b86e23ef · andymccall/oxygenflow

# f3b5b86e23ef oxygenflow / crates / jet / src / swarm.rs

18.1 KiB · Rust · last changed Swarm: neighbouring chunks in runs, and a picker that does not rescan (closes #97) 9 days ago source blame history raw

//! The swarm (oxygenflow#47, docs/jetstream.md §8.1): runners fetch image
//! chunks from each other as well as from the hub.
//!
//! This module holds what both ends agree on and what can be tested
//! without a network: the have-map a peer offers ([`Bits`]), the ticket the
//! hub signs to admit a runner to its peers ([`Ticket`]), and the order in
//! which a fetcher asks its sources for chunks ([`Picker`]): rarest first,
//! the hub only for what no peer has, and an endgame at the close.

use oxy_core::hash::KeyId;
use oxy_core::identity::{verify_challenge, Identity, SecretKey, Signature};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::time::{Duration, Instant};

/// How long a ticket admits a runner to its peers.
pub const TICKET_LIFE_MS: u64 = 15 * 60 * 1000;
/// Peers make no progress for this long and the hub is asked for anything.
pub const STALL: Duration = Duration::from_secs(2);
/// The most neighbouring chunks a fetcher takes in order where rarity does
/// not decide (about 5.6 MB of an image's 88 KB chunks); a small image's
/// runs are shorter, so it still has a few hundred to spread runners over.
const RUN: usize = 64;
/// Most sources asked for the same chunk in the endgame.
const ENDGAME_COPIES: u8 = 2;

/// One bit per chunk of a manifest's needed list, in its order.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Bits {
    words: Vec<u64>,
    len: usize,
}

impl Bits {
    pub fn new(len: usize) -> Bits {
        Bits { words: vec![0; len.div_ceil(64)], len }
    }

    /// Every bit set: a peer that holds the whole image.
    pub fn full(len: usize) -> Bits {
        let mut b = Bits::new(len);
        (0..len).for_each(|i| b.set(i));
        b
    }

    pub fn len(&self) -> usize {
        self.len
    }

    pub fn is_empty(&self) -> bool {
        self.len == 0
    }

    pub fn get(&self, i: usize) -> bool {
        i < self.len && self.words[i / 64] & (1 << (i % 64)) != 0
    }

    pub fn set(&mut self, i: usize) {
        if i < self.len {
            self.words[i / 64] |= 1 << (i % 64);
        }
    }

    pub fn clear(&mut self, i: usize) {
        if i < self.len {
            self.words[i / 64] &= !(1 << (i % 64));
        }
    }

    pub fn count(&self) -> usize {
        self.words.iter().map(|w| w.count_ones() as usize).sum()
    }

    /// Little-endian bytes, as `GET …/have` sends them.
    pub fn to_bytes(&self) -> Vec<u8> {
        let mut out: Vec<u8> = self.words.iter().flat_map(|w| w.to_le_bytes()).collect();
        out.truncate(self.len.div_ceil(8));
        out
    }

    /// The bits of a `have` answer for a list of `len` chunks; bits past
    /// `len` are ignored, and a short answer means the rest are unset.
    pub fn from_bytes(len: usize, bytes: &[u8]) -> Bits {
        let mut b = Bits::new(len);
        for (i, byte) in bytes.iter().enumerate().take(len.div_ceil(8)) {
            b.words[i / 8] |= (*byte as u64) << (8 * (i % 8));
        }
        if let Some(last) = b.words.last_mut() {
            if !len.is_multiple_of(64) {
                *last &= (1u64 << (len % 64)) - 1;
            }
        }
        b
    }
}

/// What the hub signs to admit a runner to its peers.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct Ticket {
    /// The hub's key, hex: a peer accepts only its own hub's tickets.
    pub hub: String,
    pub runner: i64,
    pub name: String,
    pub site: String,
    pub expires_ms: u64,
}

impl Ticket {
    /// The ticket as text: its JSON and the hub's signature over it, hex,
    /// joined by a dot.
    pub fn seal(&self, key: &KeyId, secret: &SecretKey) -> String {
        let body = serde_json::to_vec(self).expect("a ticket serialises");
        let sig = secret.sign_challenge(key, &body).expect("a ticket is longer than 16 bytes");
        format!("{}.{}", hex::encode(&body), hex::encode(Signature::encode_list(&[sig])))
    }

    /// Read and check a ticket: signed by `hub`, not expired at `now_ms`.
    pub fn open(text: &str, hub: &Identity, now_ms: u64) -> Result<Ticket, String> {
        let (body, sig) = text.trim().split_once('.').ok_or("not a ticket")?;
        let body = hex::decode(body).map_err(|_| "not a ticket")?;
        let sigs = hex::decode(sig).ok().and_then(|b| Signature::decode_list(&b).ok()).ok_or("not a ticket's signature")?;
        let sig = sigs.first().ok_or("an unsigned ticket")?;
        verify_challenge(hub, sig, &body).map_err(|_| "the ticket is not signed by this runner's hub")?;
        let t: Ticket = serde_json::from_slice(&body).map_err(|_| "not a ticket")?;
        if t.hub != hub.hash().to_hex() {
            return Err("the ticket names another hub".into());
        }
        if t.expires_ms <= now_ms {
            return Err("the ticket has expired".into());
        }
        Ok(t)
    }
}

/// Where a chunk comes from: the hub, or a peer by its index.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum Source {
    Hub,
    Peer(usize),
}

/// What a source should do next.
#[derive(Debug, PartialEq, Eq)]
pub enum Pick {
    /// Ask for this chunk (an index into the needed list).
    Chunk(usize),
    /// Nothing for this source now; ask again shortly.
    Wait,
    /// Every chunk is here.
    Done,
}

/// The order a fetcher asks its sources for chunks (docs/jetstream.md
/// §8.1): the rarest chunk a peer holds first; the hub for chunks no peer
/// holds, or for anything once the peers stall; and in the endgame, a
/// second source for a chunk already asked for.
pub struct Picker {
    have: Bits,
    asked: Vec<u8>,
    /// The source first asked for each chunk in flight: the endgame asks
    /// a different one.
    first: Vec<Option<Source>>,
    /// Each peer's map: what it has, and what it is fetching now.
    peers: HashMap<usize, (Bits, Bits)>,
    /// Missing chunks, rarest first; rebuilt when the peers' maps change.
    order: Vec<usize>,
    /// Where each chunk sits in `order`.
    pos: Vec<usize>,
    /// No chunk before this place in `order` can be picked fresh: each is
    /// here or asked for. Moved back when one fails.
    skip: usize,
    /// How many peers hold each chunk, as of the last rebuild.
    holders: Vec<u16>,
    /// Whether some peer is fetching each chunk now: the hub is not asked
    /// for it, the peer will have it shortly.
    coming: Vec<bool>,
    progress: Instant,
    seed: u64,
}

impl Picker {
    /// A picker for `len` chunks, of which `have` are already here.
    pub fn new(len: usize, have: Bits, seed: u64) -> Picker {
        let mut p = Picker {
            have,
            asked: vec![0; len],
            first: vec![None; len],
            peers: HashMap::new(),
            order: Vec::new(),
            pos: vec![0; len],
            skip: 0,
            holders: vec![0; len],
            coming: vec![false; len],
            progress: Instant::now(),
            seed: seed | 1,
        };
        p.reorder();
        p
    }

    pub fn len(&self) -> usize {
        self.asked.len()
    }

    pub fn is_empty(&self) -> bool {
        self.asked.is_empty()
    }

    pub fn missing(&self) -> usize {
        self.len() - self.have.count()
    }

    pub fn have(&self) -> &Bits {
        &self.have
    }

    /// A peer's have-map, new or refreshed.
    pub fn peer(&mut self, peer: usize, bits: Bits) {
        let none = Bits::new(bits.len());
        self.peer_fetching(peer, bits, none);
    }

    /// A peer's have-map and the chunks it is fetching now.
    pub fn peer_fetching(&mut self, peer: usize, have: Bits, fetching: Bits) {
        self.peers.insert(peer, (have, fetching));
        self.reorder();
    }

    /// A peer that failed: it is asked for nothing more.
    pub fn drop_peer(&mut self, peer: usize) {
        self.peers.remove(&peer);
        self.reorder();
    }

    fn next_rand(&mut self) -> u64 {
        // xorshift64: enough to spread runners that start together.
        self.seed ^= self.seed << 13;
        self.seed ^= self.seed >> 7;
        self.seed ^= self.seed << 17;
        self.seed
    }

    /// Rarest first; among chunks as rare as each other, runs of up to
    /// [`RUN`] neighbours in order, the runs shuffled. A peer reads them out of a
    /// whole image, and on a spinning disk scattered reads cost a seek each
    /// (Earth sent 2.3 GiB in 20 minutes, picked one chunk at a time at
    /// random); runners that start together still spread over the runs.
    fn reorder(&mut self) {
        let run = (self.len() / 256).clamp(1, RUN);
        let shuffle: Vec<u64> = (0..self.len().div_ceil(run)).map(|_| self.next_rand()).collect();
        let mut keyed: Vec<(u16, u64, usize)> = Vec::new();
        for i in 0..self.len() {
            self.holders[i] = self.peers.values().filter(|(b, _)| b.get(i)).count().min(u16::MAX as usize) as u16;
            self.coming[i] = self.peers.values().any(|(_, f)| f.get(i));
            if !self.have.get(i) {
                keyed.push((self.holders[i], shuffle[i / run], i));
            }
        }
        keyed.sort_unstable();
        self.order = keyed.into_iter().map(|(_, _, i)| i).collect();
        for (at, &i) in self.order.iter().enumerate() {
            self.pos[i] = at;
        }
        self.skip = 0;
    }

    /// The next chunk for `src` to fetch.
    pub fn pick(&mut self, src: Source) -> Pick {
        if self.missing() == 0 {
            return Pick::Done;
        }
        let holds = |i: usize| match src {
            Source::Hub => true,
            Source::Peer(p) => self.peers.get(&p).is_some_and(|(b, _)| b.get(i)),
        };
        let stalled = self.progress.elapsed() >= STALL;
        // Not yet asked for: a peer takes the rarest it holds; the hub
        // takes what no peer holds, or anything once the peers stall.
        while self.skip < self.order.len() && (self.have.get(self.order[self.skip]) || self.asked[self.order[self.skip]] > 0) {
            self.skip += 1;
        }
        let fresh = self.order[self.skip..].iter().copied().find(|&i| {
            !self.have.get(i)
                && self.asked[i] == 0
                && holds(i)
                && (src != Source::Hub || stalled || self.peers.is_empty() || (self.holders[i] == 0 && !self.coming[i]))
        });
        if let Some(i) = fresh {
            self.asked[i] += 1;
            self.first[i] = Some(src);
            return Pick::Chunk(i);
        }
        // The endgame: everything missing is asked for already. The hub
        // joins it only once the peers stall.
        if src == Source::Hub && !stalled && !self.peers.is_empty() {
            return Pick::Wait;
        }
        let mut left = self.order.iter().copied().filter(|&i| !self.have.get(i));
        if left.clone().all(|i| self.asked[i] > 0) {
            if let Some(i) = left.find(|&i| self.asked[i] < ENDGAME_COPIES && self.first[i] != Some(src) && holds(i)) {
                self.asked[i] += 1;
                return Pick::Chunk(i);
            }
        }
        Pick::Wait
    }

    /// Chunk `i` arrived and was checked.
    pub fn got(&mut self, i: usize) {
        self.have.set(i);
        self.asked[i] = self.asked[i].saturating_sub(1);
        self.first[i] = None;
        self.progress = Instant::now();
    }

    /// Chunk `i` did not come: it can be asked for again.
    pub fn failed(&mut self, i: usize) {
        self.asked[i] = self.asked[i].saturating_sub(1);
        if self.asked[i] == 0 {
            self.first[i] = None;
            if !self.have.get(i) {
                self.skip = self.skip.min(self.pos[i]);
            }
        }
    }
}

#[cfg(test)]
mod tests {
    #[test]
    fn from_one_peer_neighbouring_chunks_come_in_order() {
        let n = RUN * 256;
        let mut p = Picker::new(n, Bits::new(n), 3);
        p.peer(0, Bits::full(n));
        let first: Vec<usize> = (0..RUN)
            .map(|_| match p.pick(Source::Peer(0)) {
                Pick::Chunk(i) => i,
                other => panic!("{other:?}"),
            })
            .collect();
        assert_eq!(first[0] % RUN, 0, "a run starts at its first chunk: {first:?}");
        assert!(first.windows(2).all(|w| w[1] == w[0] + 1), "one run, in order: {first:?}");
        // A failed chunk is asked for again before going further.
        p.failed(first[3]);
        assert_eq!(p.pick(Source::Peer(0)), Pick::Chunk(first[3]));
    }

    /// A fetch the size of the 14 GB Rust image (142,244 chunks) from one
    /// peer that holds them all: how long picking takes.
    #[test]
    #[ignore]
    fn picking_a_big_image_from_one_peer_is_quick() {
        let n = 142_244;
        let mut p = Picker::new(n, Bits::new(n), 7);
        p.peer(0, Bits::full(n));
        let started = std::time::Instant::now();
        let mut in_flight = std::collections::VecDeque::new();
        loop {
            match p.pick(Source::Peer(0)) {
                Pick::Chunk(i) => {
                    in_flight.push_back(i);
                    if in_flight.len() >= 4 {
                        p.got(in_flight.pop_front().unwrap());
                    }
                }
                Pick::Wait => p.got(in_flight.pop_front().unwrap()),
                Pick::Done => break,
            }
        }
        eprintln!("picked {n} chunks in {:?}", started.elapsed());
    }

    use super::*;

    #[test]
    fn bits_travel_as_bytes() {
        let mut b = Bits::new(70);
        [0, 9, 63, 64, 69].iter().for_each(|&i| b.set(i));
        let back = Bits::from_bytes(70, &b.to_bytes());
        assert_eq!(back, b);
        assert_eq!(b.to_bytes().len(), 9);
        assert_eq!(Bits::from_bytes(10, &[0xff, 0xff, 0xff]).count(), 10, "bits past the list are ignored");
        assert_eq!(Bits::from_bytes(10, &[]).count(), 0, "a short answer leaves the rest unset");
        assert_eq!(Bits::full(130).count(), 130);
    }

    #[test]
    fn a_ticket_opens_only_with_its_hub_and_in_time() {
        let sk = SecretKey::generate();
        let hub = sk.identity("hub", "hub@example", 1);
        let t = Ticket { hub: hub.hash().to_hex(), runner: 7, name: "earth-1".into(), site: "home".into(), expires_ms: 10_000 };
        let text = t.seal(&hub.hash(), &sk);
        assert_eq!(Ticket::open(&text, &hub, 9_999).unwrap(), t);
        assert!(Ticket::open(&text, &hub, 10_000).unwrap_err().contains("expired"));
        let other_sk = SecretKey::generate();
        let other = other_sk.identity("other", "o@example", 1);
        assert!(Ticket::open(&text, &other, 1).is_err(), "another hub's key");
        let mut forged = text.clone();
        forged.replace_range(2..4, "ff");
        assert!(Ticket::open(&forged, &hub, 1).is_err(), "a changed body");
        assert!(Ticket::open("nonsense", &hub, 1).is_err());
    }

    #[test]
    fn peers_are_asked_for_the_rarest_chunk_they_hold() {
        let mut p = Picker::new(4, Bits::new(4), 42);
        let mut a = Bits::new(4);
        [0, 1, 2].iter().for_each(|&i| a.set(i));
        let mut b = Bits::new(4);
        [0, 1].iter().for_each(|&i| b.set(i));
        p.peer(0, a);
        p.peer(1, b);
        // Chunk 2 only peer 0 holds; chunk 3 nobody does.
        assert_eq!(p.pick(Source::Peer(0)), Pick::Chunk(2));
        assert_eq!(p.pick(Source::Hub), Pick::Chunk(3), "the hub seeds what no peer holds");
        assert_eq!(p.pick(Source::Hub), Pick::Wait, "and leaves the rest to the peers");
        let first = p.pick(Source::Peer(1));
        assert!(matches!(first, Pick::Chunk(0) | Pick::Chunk(1)));
    }

    #[test]
    fn the_hub_is_not_asked_for_what_a_peer_is_fetching() {
        let mut p = Picker::new(2, Bits::new(2), 4);
        let mut fetching = Bits::new(2);
        fetching.set(0);
        p.peer_fetching(0, Bits::new(2), fetching);
        assert_eq!(p.pick(Source::Hub), Pick::Chunk(1));
        assert_eq!(p.pick(Source::Hub), Pick::Wait, "chunk 0 is coming from the peer");
        let mut have = Bits::new(2);
        have.set(0);
        p.peer(0, have);
        assert_eq!(p.pick(Source::Peer(0)), Pick::Chunk(0));
    }

    #[test]
    fn without_peers_the_hub_serves_everything() {
        let mut p = Picker::new(3, Bits::new(3), 1);
        let mut seen = Vec::new();
        while let Pick::Chunk(i) = p.pick(Source::Hub) {
            seen.push(i);
            p.got(i);
        }
        seen.sort();
        assert_eq!(seen, vec![0, 1, 2]);
        assert_eq!(p.pick(Source::Hub), Pick::Done);
    }

    #[test]
    fn the_endgame_never_asks_the_same_source_twice() {
        let mut p = Picker::new(1, Bits::new(1), 9);
        assert_eq!(p.pick(Source::Hub), Pick::Chunk(0));
        assert_eq!(p.pick(Source::Hub), Pick::Wait);
        p.peer(0, Bits::full(1));
        assert_eq!(p.pick(Source::Peer(0)), Pick::Chunk(0), "a peer may race the hub for it");
        assert_eq!(p.pick(Source::Hub), Pick::Wait, "but not the hub a peer, while the peers make progress");
    }

    #[test]
    fn the_endgame_asks_a_second_source_and_a_failure_frees_the_chunk() {
        let mut p = Picker::new(1, Bits::new(1), 3);
        p.peer(0, Bits::full(1));
        p.peer(1, Bits::full(1));
        assert_eq!(p.pick(Source::Peer(0)), Pick::Chunk(0));
        assert_eq!(p.pick(Source::Peer(1)), Pick::Chunk(0), "the endgame");
        assert_eq!(p.pick(Source::Peer(1)), Pick::Wait, "at most two copies");
        p.failed(0);
        p.failed(0);
        p.drop_peer(0);
        assert_eq!(p.pick(Source::Peer(1)), Pick::Chunk(0));
        p.got(0);
        assert_eq!(p.pick(Source::Peer(1)), Pick::Done);
    }

    #[test]
    fn chunks_already_here_are_never_asked_for() {
        let mut have = Bits::new(3);
        have.set(1);
        let mut p = Picker::new(3, have, 5);
        assert_eq!(p.missing(), 2);
        let mut seen = Vec::new();
        while let Pick::Chunk(i) = p.pick(Source::Hub) {
            seen.push(i);
            p.got(i);
        }
        assert!(!seen.contains(&1));
    }

    #[test]
    fn runners_starting_together_ask_the_hub_for_different_chunks() {
        let a: Vec<_> = {
            let mut p = Picker::new(64, Bits::new(64), 11);
            (0..8).filter_map(|_| if let Pick::Chunk(i) = p.pick(Source::Hub) { Some(i) } else { None }).collect()
        };
        let b: Vec<_> = {
            let mut p = Picker::new(64, Bits::new(64), 12);
            (0..8).filter_map(|_| if let Pick::Chunk(i) = p.pick(Source::Hub) { Some(i) } else { None }).collect()
        };
        assert_ne!(a, b);
    }
}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

1	//! The swarm (oxygenflow#47, docs/jetstream.md §8.1): runners fetch image
2	//! chunks from each other as well as from the hub.
3	//!
4	//! This module holds what both ends agree on and what can be tested
5	//! without a network: the have-map a peer offers ([`Bits`]), the ticket the
6	//! hub signs to admit a runner to its peers ([`Ticket`]), and the order in
7	//! which a fetcher asks its sources for chunks ([`Picker`]): rarest first,
8	//! the hub only for what no peer has, and an endgame at the close.
9
10	use oxy_core::hash::KeyId;
11	use oxy_core::identity::{verify_challenge, Identity, SecretKey, Signature};
12	use serde::{Deserialize, Serialize};
13	use std::collections::HashMap;
14	use std::time::{Duration, Instant};
15
16	/// How long a ticket admits a runner to its peers.
17	pub const TICKET_LIFE_MS: u64 = 15 * 60 * 1000;
18	/// Peers make no progress for this long and the hub is asked for anything.
19	pub const STALL: Duration = Duration::from_secs(2);
20	/// The most neighbouring chunks a fetcher takes in order where rarity does
21	/// not decide (about 5.6 MB of an image's 88 KB chunks); a small image's
22	/// runs are shorter, so it still has a few hundred to spread runners over.
23	const RUN: usize = 64;
24	/// Most sources asked for the same chunk in the endgame.
25	const ENDGAME_COPIES: u8 = 2;
26
27	/// One bit per chunk of a manifest's needed list, in its order.
28	#[derive(Clone, Debug, Default, PartialEq, Eq)]
29	pub struct Bits {
30	words: Vec<u64>,
31	len: usize,
32	}
33
34	impl Bits {
35	pub fn new(len: usize) -> Bits {
36	Bits { words: vec![0; len.div_ceil(64)], len }
37	}
38
39	/// Every bit set: a peer that holds the whole image.
40	pub fn full(len: usize) -> Bits {
41	let mut b = Bits::new(len);
42	(0..len).for_each(\|i\| b.set(i));
43	b
44	}
45
46	pub fn len(&self) -> usize {
47	self.len
48	}
49
50	pub fn is_empty(&self) -> bool {
51	self.len == 0
52	}
53
54	pub fn get(&self, i: usize) -> bool {
55	i < self.len && self.words[i / 64] & (1 << (i % 64)) != 0
56	}
57
58	pub fn set(&mut self, i: usize) {
59	if i < self.len {
60	self.words[i / 64] \|= 1 << (i % 64);
61	}
62	}
63
64	pub fn clear(&mut self, i: usize) {
65	if i < self.len {
66	self.words[i / 64] &= !(1 << (i % 64));
67	}
68	}
69
70	pub fn count(&self) -> usize {
71	self.words.iter().map(\|w\| w.count_ones() as usize).sum()
72	}
73
74	/// Little-endian bytes, as `GET …/have` sends them.
75	pub fn to_bytes(&self) -> Vec<u8> {
76	let mut out: Vec<u8> = self.words.iter().flat_map(\|w\| w.to_le_bytes()).collect();
77	out.truncate(self.len.div_ceil(8));
78	out
79	}
80
81	/// The bits of a `have` answer for a list of `len` chunks; bits past
82	/// `len` are ignored, and a short answer means the rest are unset.
83	pub fn from_bytes(len: usize, bytes: &[u8]) -> Bits {
84	let mut b = Bits::new(len);
85	for (i, byte) in bytes.iter().enumerate().take(len.div_ceil(8)) {
86	b.words[i / 8] \|= (byte as u64) << (8 (i % 8));
87	}
88	if let Some(last) = b.words.last_mut() {
89	if !len.is_multiple_of(64) {
90	*last &= (1u64 << (len % 64)) - 1;
91	}
92	}
93	b
94	}
95	}
96
97	/// What the hub signs to admit a runner to its peers.
98	#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
99	pub struct Ticket {
100	/// The hub's key, hex: a peer accepts only its own hub's tickets.
101	pub hub: String,
102	pub runner: i64,
103	pub name: String,
104	pub site: String,
105	pub expires_ms: u64,
106	}
107
108	impl Ticket {
109	/// The ticket as text: its JSON and the hub's signature over it, hex,
110	/// joined by a dot.
111	pub fn seal(&self, key: &KeyId, secret: &SecretKey) -> String {
112	let body = serde_json::to_vec(self).expect("a ticket serialises");
113	let sig = secret.sign_challenge(key, &body).expect("a ticket is longer than 16 bytes");
114	format!("{}.{}", hex::encode(&body), hex::encode(Signature::encode_list(&[sig])))
115	}
116
117	/// Read and check a ticket: signed by `hub`, not expired at `now_ms`.
118	pub fn open(text: &str, hub: &Identity, now_ms: u64) -> Result<Ticket, String> {
119	let (body, sig) = text.trim().split_once('.').ok_or("not a ticket")?;
120	let body = hex::decode(body).map_err(\|_\| "not a ticket")?;
121	let sigs = hex::decode(sig).ok().and_then(\|b\| Signature::decode_list(&b).ok()).ok_or("not a ticket's signature")?;
122	let sig = sigs.first().ok_or("an unsigned ticket")?;
123	verify_challenge(hub, sig, &body).map_err(\|_\| "the ticket is not signed by this runner's hub")?;
124	let t: Ticket = serde_json::from_slice(&body).map_err(\|_\| "not a ticket")?;
125	if t.hub != hub.hash().to_hex() {
126	return Err("the ticket names another hub".into());
127	}
128	if t.expires_ms <= now_ms {
129	return Err("the ticket has expired".into());
130	}
131	Ok(t)
132	}
133	}
134
135	/// Where a chunk comes from: the hub, or a peer by its index.
136	#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
137	pub enum Source {
138	Hub,
139	Peer(usize),
140	}
141
142	/// What a source should do next.
143	#[derive(Debug, PartialEq, Eq)]
144	pub enum Pick {
145	/// Ask for this chunk (an index into the needed list).
146	Chunk(usize),
147	/// Nothing for this source now; ask again shortly.
148	Wait,
149	/// Every chunk is here.
150	Done,
151	}
152
153	/// The order a fetcher asks its sources for chunks (docs/jetstream.md
154	/// §8.1): the rarest chunk a peer holds first; the hub for chunks no peer
155	/// holds, or for anything once the peers stall; and in the endgame, a
156	/// second source for a chunk already asked for.
157	pub struct Picker {
158	have: Bits,
159	asked: Vec<u8>,
160	/// The source first asked for each chunk in flight: the endgame asks
161	/// a different one.
162	first: Vec<Option<Source>>,
163	/// Each peer's map: what it has, and what it is fetching now.
164	peers: HashMap<usize, (Bits, Bits)>,
165	/// Missing chunks, rarest first; rebuilt when the peers' maps change.
166	order: Vec<usize>,
167	/// Where each chunk sits in `order`.
168	pos: Vec<usize>,
169	/// No chunk before this place in `order` can be picked fresh: each is
170	/// here or asked for. Moved back when one fails.
171	skip: usize,
172	/// How many peers hold each chunk, as of the last rebuild.
173	holders: Vec<u16>,
174	/// Whether some peer is fetching each chunk now: the hub is not asked
175	/// for it, the peer will have it shortly.
176	coming: Vec<bool>,
177	progress: Instant,
178	seed: u64,
179	}
180
181	impl Picker {
182	/// A picker for `len` chunks, of which `have` are already here.
183	pub fn new(len: usize, have: Bits, seed: u64) -> Picker {
184	let mut p = Picker {
185	have,
186	asked: vec![0; len],
187	first: vec![None; len],
188	peers: HashMap::new(),
189	order: Vec::new(),
190	pos: vec![0; len],
191	skip: 0,
192	holders: vec![0; len],
193	coming: vec![false; len],
194	progress: Instant::now(),
195	seed: seed \| 1,
196	};
197	p.reorder();
198	p
199	}
200
201	pub fn len(&self) -> usize {
202	self.asked.len()
203	}
204
205	pub fn is_empty(&self) -> bool {
206	self.asked.is_empty()
207	}
208
209	pub fn missing(&self) -> usize {
210	self.len() - self.have.count()
211	}
212
213	pub fn have(&self) -> &Bits {
214	&self.have
215	}
216
217	/// A peer's have-map, new or refreshed.
218	pub fn peer(&mut self, peer: usize, bits: Bits) {
219	let none = Bits::new(bits.len());
220	self.peer_fetching(peer, bits, none);
221	}
222
223	/// A peer's have-map and the chunks it is fetching now.
224	pub fn peer_fetching(&mut self, peer: usize, have: Bits, fetching: Bits) {
225	self.peers.insert(peer, (have, fetching));
226	self.reorder();
227	}
228
229	/// A peer that failed: it is asked for nothing more.
230	pub fn drop_peer(&mut self, peer: usize) {
231	self.peers.remove(&peer);
232	self.reorder();
233	}
234
235	fn next_rand(&mut self) -> u64 {
236	// xorshift64: enough to spread runners that start together.
237	self.seed ^= self.seed << 13;
238	self.seed ^= self.seed >> 7;
239	self.seed ^= self.seed << 17;
240	self.seed
241	}
242
243	/// Rarest first; among chunks as rare as each other, runs of up to
244	/// [`RUN`] neighbours in order, the runs shuffled. A peer reads them out of a
245	/// whole image, and on a spinning disk scattered reads cost a seek each
246	/// (Earth sent 2.3 GiB in 20 minutes, picked one chunk at a time at
247	/// random); runners that start together still spread over the runs.
248	fn reorder(&mut self) {
249	let run = (self.len() / 256).clamp(1, RUN);
250	let shuffle: Vec<u64> = (0..self.len().div_ceil(run)).map(\|_\| self.next_rand()).collect();
251	let mut keyed: Vec<(u16, u64, usize)> = Vec::new();
252	for i in 0..self.len() {
253	self.holders[i] = self.peers.values().filter(\|(b, _)\| b.get(i)).count().min(u16::MAX as usize) as u16;
254	self.coming[i] = self.peers.values().any(\|(_, f)\| f.get(i));
255	if !self.have.get(i) {
256	keyed.push((self.holders[i], shuffle[i / run], i));
257	}
258	}
259	keyed.sort_unstable();
260	self.order = keyed.into_iter().map(\|(_, _, i)\| i).collect();
261	for (at, &i) in self.order.iter().enumerate() {
262	self.pos[i] = at;
263	}
264	self.skip = 0;
265	}
266
267	/// The next chunk for `src` to fetch.
268	pub fn pick(&mut self, src: Source) -> Pick {
269	if self.missing() == 0 {
270	return Pick::Done;
271	}
272	let holds = \|i: usize\| match src {
273	Source::Hub => true,
274	Source::Peer(p) => self.peers.get(&p).is_some_and(\|(b, _)\| b.get(i)),
275	};
276	let stalled = self.progress.elapsed() >= STALL;
277	// Not yet asked for: a peer takes the rarest it holds; the hub
278	// takes what no peer holds, or anything once the peers stall.
279	while self.skip < self.order.len() && (self.have.get(self.order[self.skip]) \|\| self.asked[self.order[self.skip]] > 0) {
280	self.skip += 1;
281	}
282	let fresh = self.order[self.skip..].iter().copied().find(\|&i\| {
283	!self.have.get(i)
284	&& self.asked[i] == 0
285	&& holds(i)
286	&& (src != Source::Hub \|\| stalled \|\| self.peers.is_empty() \|\| (self.holders[i] == 0 && !self.coming[i]))
287	});
288	if let Some(i) = fresh {
289	self.asked[i] += 1;
290	self.first[i] = Some(src);
291	return Pick::Chunk(i);
292	}
293	// The endgame: everything missing is asked for already. The hub
294	// joins it only once the peers stall.
295	if src == Source::Hub && !stalled && !self.peers.is_empty() {
296	return Pick::Wait;
297	}
298	let mut left = self.order.iter().copied().filter(\|&i\| !self.have.get(i));
299	if left.clone().all(\|i\| self.asked[i] > 0) {
300	if let Some(i) = left.find(\|&i\| self.asked[i] < ENDGAME_COPIES && self.first[i] != Some(src) && holds(i)) {
301	self.asked[i] += 1;
302	return Pick::Chunk(i);
303	}
304	}
305	Pick::Wait
306	}
307
308	/// Chunk `i` arrived and was checked.
309	pub fn got(&mut self, i: usize) {
310	self.have.set(i);
311	self.asked[i] = self.asked[i].saturating_sub(1);
312	self.first[i] = None;
313	self.progress = Instant::now();
314	}
315
316	/// Chunk `i` did not come: it can be asked for again.
317	pub fn failed(&mut self, i: usize) {
318	self.asked[i] = self.asked[i].saturating_sub(1);
319	if self.asked[i] == 0 {
320	self.first[i] = None;
321	if !self.have.get(i) {
322	self.skip = self.skip.min(self.pos[i]);
323	}
324	}
325	}
326	}
327
328	#[cfg(test)]
329	mod tests {
330	#[test]
331	fn from_one_peer_neighbouring_chunks_come_in_order() {
332	let n = RUN * 256;
333	let mut p = Picker::new(n, Bits::new(n), 3);
334	p.peer(0, Bits::full(n));
335	let first: Vec<usize> = (0..RUN)
336	.map(\|_\| match p.pick(Source::Peer(0)) {
337	Pick::Chunk(i) => i,
338	other => panic!("{other:?}"),
339	})
340	.collect();
341	assert_eq!(first[0] % RUN, 0, "a run starts at its first chunk: {first:?}");
342	assert!(first.windows(2).all(\|w\| w[1] == w[0] + 1), "one run, in order: {first:?}");
343	// A failed chunk is asked for again before going further.
344	p.failed(first[3]);
345	assert_eq!(p.pick(Source::Peer(0)), Pick::Chunk(first[3]));
346	}
347
348	/// A fetch the size of the 14 GB Rust image (142,244 chunks) from one
349	/// peer that holds them all: how long picking takes.
350	#[test]
351	#[ignore]
352	fn picking_a_big_image_from_one_peer_is_quick() {
353	let n = 142_244;
354	let mut p = Picker::new(n, Bits::new(n), 7);
355	p.peer(0, Bits::full(n));
356	let started = std::time::Instant::now();
357	let mut in_flight = std::collections::VecDeque::new();
358	loop {
359	match p.pick(Source::Peer(0)) {
360	Pick::Chunk(i) => {
361	in_flight.push_back(i);
362	if in_flight.len() >= 4 {
363	p.got(in_flight.pop_front().unwrap());
364	}
365	}
366	Pick::Wait => p.got(in_flight.pop_front().unwrap()),
367	Pick::Done => break,
368	}
369	}
370	eprintln!("picked {n} chunks in {:?}", started.elapsed());
371	}
372
373	use super::*;
374
375	#[test]
376	fn bits_travel_as_bytes() {
377	let mut b = Bits::new(70);
378	[0, 9, 63, 64, 69].iter().for_each(\|&i\| b.set(i));
379	let back = Bits::from_bytes(70, &b.to_bytes());
380	assert_eq!(back, b);
381	assert_eq!(b.to_bytes().len(), 9);
382	assert_eq!(Bits::from_bytes(10, &[0xff, 0xff, 0xff]).count(), 10, "bits past the list are ignored");
383	assert_eq!(Bits::from_bytes(10, &[]).count(), 0, "a short answer leaves the rest unset");
384	assert_eq!(Bits::full(130).count(), 130);
385	}
386
387	#[test]
388	fn a_ticket_opens_only_with_its_hub_and_in_time() {
389	let sk = SecretKey::generate();
390	let hub = sk.identity("hub", "hub@example", 1);
391	let t = Ticket { hub: hub.hash().to_hex(), runner: 7, name: "earth-1".into(), site: "home".into(), expires_ms: 10_000 };
392	let text = t.seal(&hub.hash(), &sk);
393	assert_eq!(Ticket::open(&text, &hub, 9_999).unwrap(), t);
394	assert!(Ticket::open(&text, &hub, 10_000).unwrap_err().contains("expired"));
395	let other_sk = SecretKey::generate();
396	let other = other_sk.identity("other", "o@example", 1);
397	assert!(Ticket::open(&text, &other, 1).is_err(), "another hub's key");
398	let mut forged = text.clone();
399	forged.replace_range(2..4, "ff");
400	assert!(Ticket::open(&forged, &hub, 1).is_err(), "a changed body");
401	assert!(Ticket::open("nonsense", &hub, 1).is_err());
402	}
403
404	#[test]
405	fn peers_are_asked_for_the_rarest_chunk_they_hold() {
406	let mut p = Picker::new(4, Bits::new(4), 42);
407	let mut a = Bits::new(4);
408	[0, 1, 2].iter().for_each(\|&i\| a.set(i));
409	let mut b = Bits::new(4);
410	[0, 1].iter().for_each(\|&i\| b.set(i));
411	p.peer(0, a);
412	p.peer(1, b);
413	// Chunk 2 only peer 0 holds; chunk 3 nobody does.
414	assert_eq!(p.pick(Source::Peer(0)), Pick::Chunk(2));
415	assert_eq!(p.pick(Source::Hub), Pick::Chunk(3), "the hub seeds what no peer holds");
416	assert_eq!(p.pick(Source::Hub), Pick::Wait, "and leaves the rest to the peers");
417	let first = p.pick(Source::Peer(1));
418	assert!(matches!(first, Pick::Chunk(0) \| Pick::Chunk(1)));
419	}
420
421	#[test]
422	fn the_hub_is_not_asked_for_what_a_peer_is_fetching() {
423	let mut p = Picker::new(2, Bits::new(2), 4);
424	let mut fetching = Bits::new(2);
425	fetching.set(0);
426	p.peer_fetching(0, Bits::new(2), fetching);
427	assert_eq!(p.pick(Source::Hub), Pick::Chunk(1));
428	assert_eq!(p.pick(Source::Hub), Pick::Wait, "chunk 0 is coming from the peer");
429	let mut have = Bits::new(2);
430	have.set(0);
431	p.peer(0, have);
432	assert_eq!(p.pick(Source::Peer(0)), Pick::Chunk(0));
433	}
434
435	#[test]
436	fn without_peers_the_hub_serves_everything() {
437	let mut p = Picker::new(3, Bits::new(3), 1);
438	let mut seen = Vec::new();
439	while let Pick::Chunk(i) = p.pick(Source::Hub) {
440	seen.push(i);
441	p.got(i);
442	}
443	seen.sort();
444	assert_eq!(seen, vec![0, 1, 2]);
445	assert_eq!(p.pick(Source::Hub), Pick::Done);
446	}
447
448	#[test]
449	fn the_endgame_never_asks_the_same_source_twice() {
450	let mut p = Picker::new(1, Bits::new(1), 9);
451	assert_eq!(p.pick(Source::Hub), Pick::Chunk(0));
452	assert_eq!(p.pick(Source::Hub), Pick::Wait);
453	p.peer(0, Bits::full(1));
454	assert_eq!(p.pick(Source::Peer(0)), Pick::Chunk(0), "a peer may race the hub for it");
455	assert_eq!(p.pick(Source::Hub), Pick::Wait, "but not the hub a peer, while the peers make progress");
456	}
457
458	#[test]
459	fn the_endgame_asks_a_second_source_and_a_failure_frees_the_chunk() {
460	let mut p = Picker::new(1, Bits::new(1), 3);
461	p.peer(0, Bits::full(1));
462	p.peer(1, Bits::full(1));
463	assert_eq!(p.pick(Source::Peer(0)), Pick::Chunk(0));
464	assert_eq!(p.pick(Source::Peer(1)), Pick::Chunk(0), "the endgame");
465	assert_eq!(p.pick(Source::Peer(1)), Pick::Wait, "at most two copies");
466	p.failed(0);
467	p.failed(0);
468	p.drop_peer(0);
469	assert_eq!(p.pick(Source::Peer(1)), Pick::Chunk(0));
470	p.got(0);
471	assert_eq!(p.pick(Source::Peer(1)), Pick::Done);
472	}
473
474	#[test]
475	fn chunks_already_here_are_never_asked_for() {
476	let mut have = Bits::new(3);
477	have.set(1);
478	let mut p = Picker::new(3, have, 5);
479	assert_eq!(p.missing(), 2);
480	let mut seen = Vec::new();
481	while let Pick::Chunk(i) = p.pick(Source::Hub) {
482	seen.push(i);
483	p.got(i);
484	}
485	assert!(!seen.contains(&1));
486	}
487
488	#[test]
489	fn runners_starting_together_ask_the_hub_for_different_chunks() {
490	let a: Vec<_> = {
491	let mut p = Picker::new(64, Bits::new(64), 11);
492	(0..8).filter_map(\|_\| if let Pick::Chunk(i) = p.pick(Source::Hub) { Some(i) } else { None }).collect()
493	};
494	let b: Vec<_> = {
495	let mut p = Picker::new(64, Bits::new(64), 12);
496	(0..8).filter_map(\|_\| if let Pick::Chunk(i) = p.pick(Source::Hub) { Some(i) } else { None }).collect()
497	};
498	assert_ne!(a, b);
499	}
500	}

Keyboard