crates/jet/src/swarm.rs at v0.5.2 · andymccall/oxygenflow

◆ v0.5.2 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

b14aeaae Andy McCall 23 days ago ↶	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);
069e16fb Andy McCall 9 days ago ↶	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;
b14aeaae Andy McCall 23 days ago ↶	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>,
069e16fb Andy McCall 9 days ago ↶	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,
b14aeaae Andy McCall 23 days ago ↶	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(),
069e16fb Andy McCall 9 days ago ↶	190	pos: vec![0; len],
069e16fb Andy McCall 9 days ago ↶	191	skip: 0,
b14aeaae Andy McCall 23 days ago ↶	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
069e16fb Andy McCall 9 days ago ↶	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.
b14aeaae Andy McCall 23 days ago ↶	248	fn reorder(&mut self) {
069e16fb Andy McCall 9 days ago ↶	249	let run = (self.len() / 256).clamp(1, RUN);
069e16fb Andy McCall 9 days ago ↶	250	let shuffle: Vec<u64> = (0..self.len().div_ceil(run)).map(\|_\| self.next_rand()).collect();
b14aeaae Andy McCall 23 days ago ↶	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) {
069e16fb Andy McCall 9 days ago ↶	256	keyed.push((self.holders[i], shuffle[i / run], i));
b14aeaae Andy McCall 23 days ago ↶	257	}
	258	}
	259	keyed.sort_unstable();
	260	self.order = keyed.into_iter().map(\|(_, _, i)\| i).collect();
069e16fb Andy McCall 9 days ago ↶	261	for (at, &i) in self.order.iter().enumerate() {
	262	self.pos[i] = at;
	263	}
	264	self.skip = 0;
b14aeaae Andy McCall 23 days ago ↶	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.
069e16fb Andy McCall 9 days ago ↶	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\| {
b14aeaae Andy McCall 23 days ago ↶	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;
069e16fb Andy McCall 9 days ago ↶	321	if !self.have.get(i) {
	322	self.skip = self.skip.min(self.pos[i]);
	323	}
b14aeaae Andy McCall 23 days ago ↶	324	}
	325	}
	326	}
	327
	328	#[cfg(test)]
	329	mod tests {
069e16fb Andy McCall 9 days ago ↶	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
b14aeaae Andy McCall 23 days ago ↶	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