swarm.rs 91.4%

308 of 337 lines run, 82 of 94 functions · run #210 on main · run missed

//! 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	76×	pub fn new(len: usize) -> Bits {
36	76×	Bits { words: vec![0; len.div_ceil(64)], len }
37	76×	}
38
39		/// Every bit set: a peer that holds the whole image.
40	7×	pub fn full(len: usize) -> Bits {
41	7×	let mut b = Bits::new(len);
42	16593×	(0..len).for_each(\|i\| b.set(i));
43	7×	b
44	7×	}
45
46	7×	pub fn len(&self) -> usize {
47	7×	self.len
48	7×	}
49
50	0×	pub fn is_empty(&self) -> bool {
51	0×	self.len == 0
52	0×	}
53
54	101892×	pub fn get(&self, i: usize) -> bool {
55	101892×	i < self.len && self.words[i / 64] & (1 << (i % 64)) != 0
56	101892×	}
57
58	17296×	pub fn set(&mut self, i: usize) {
59	17296×	if i < self.len {
60	17296×	self.words[i / 64] \|= 1 << (i % 64);
61	17296×	}
62	17296×	}
63
64	228×	pub fn clear(&mut self, i: usize) {
65	228×	if i < self.len {
66	228×	self.words[i / 64] &= !(1 << (i % 64));
67	228×	}
68	228×	}
69
70	1441×	pub fn count(&self) -> usize {
71	18018×	self.words.iter().map(\|w\| w.count_ones() as usize).sum()
72	1441×	}
73
74		/// Little-endian bytes, as `GET …/have` sends them.
75	20×	pub fn to_bytes(&self) -> Vec<u8> {
76	22×	let mut out: Vec<u8> = self.words.iter().flat_map(\|w\| w.to_le_bytes()).collect();
77	20×	out.truncate(self.len.div_ceil(8));
78	20×	out
79	20×	}
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	21×	pub fn from_bytes(len: usize, bytes: &[u8]) -> Bits {
84	21×	let mut b = Bits::new(len);
85	101×	for (i, byte) in bytes.iter().enumerate().take(len.div_ceil(8)) {
86	101×	b.words[i / 8] \|= (byte as u64) << (8 (i % 8));
87	101×	}
88	21×	if let Some(last) = b.words.last_mut() {
89	21×	if !len.is_multiple_of(64) {
90	21×	*last &= (1u64 << (len % 64)) - 1;
91	21×	}
92	0×	}
93	21×	b
94	21×	}
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	18×	pub fn seal(&self, key: &KeyId, secret: &SecretKey) -> String {
112	18×	let body = serde_json::to_vec(self).expect("a ticket serialises");
113	18×	let sig = secret.sign_challenge(key, &body).expect("a ticket is longer than 16 bytes");
114	18×	format!("{}.{}", hex::encode(&body), hex::encode(Signature::encode_list(&[sig])))
115	18×	}
116
117		/// Read and check a ticket: signed by `hub`, not expired at `now_ms`.
118	11×	pub fn open(text: &str, hub: &Identity, now_ms: u64) -> Result<Ticket, String> {
119	11×	let (body, sig) = text.trim().split_once('.').ok_or("not a ticket")?;
120	10×	let body = hex::decode(body).map_err(\|_\| "not a ticket")?;
121	10×	let sigs = hex::decode(sig).ok().and_then(\|b\| Signature::decode_list(&b).ok()).ok_or("not a ticket's signature")?;
122	10×	let sig = sigs.first().ok_or("an unsigned ticket")?;
123	10×	verify_challenge(hub, sig, &body).map_err(\|_\| "the ticket is not signed by this runner's hub")?;
124	8×	let t: Ticket = serde_json::from_slice(&body).map_err(\|_\| "not a ticket")?;
125	8×	if t.hub != hub.hash().to_hex() {
126	0×	return Err("the ticket names another hub".into());
127	8×	}
128	8×	if t.expires_ms <= now_ms {
129	1×	return Err("the ticket has expired".into());
130	7×	}
131	7×	Ok(t)
132	11×	}
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	16×	pub fn new(len: usize, have: Bits, seed: u64) -> Picker {
184	16×	let mut p = Picker {
185	16×	have,
186	16×	asked: vec![0; len],
187	16×	first: vec![None; len],
188	16×	peers: HashMap::new(),
189	16×	order: Vec::new(),
190	16×	pos: vec![0; len],
191	16×	skip: 0,
192	16×	holders: vec![0; len],
193	16×	coming: vec![false; len],
194	16×	progress: Instant::now(),
195	16×	seed: seed \| 1,
196	16×	};
197	16×	p.reorder();
198	16×	p
199	16×	}
200
201	1546×	pub fn len(&self) -> usize {
202	1546×	self.asked.len()
203	1546×	}
204
205	0×	pub fn is_empty(&self) -> bool {
206	0×	self.asked.is_empty()
207	0×	}
208
209	1438×	pub fn missing(&self) -> usize {
210	1438×	self.len() - self.have.count()
211	1438×	}
212
213	0×	pub fn have(&self) -> &Bits {
214	0×	&self.have
215	0×	}
216
217		/// A peer's have-map, new or refreshed.
218	7×	pub fn peer(&mut self, peer: usize, bits: Bits) {
219	7×	let none = Bits::new(bits.len());
220	7×	self.peer_fetching(peer, bits, none);
221	7×	}
222
223		/// A peer's have-map and the chunks it is fetching now.
224	19×	pub fn peer_fetching(&mut self, peer: usize, have: Bits, fetching: Bits) {
225	19×	self.peers.insert(peer, (have, fetching));
226	19×	self.reorder();
227	19×	}
228
229		/// A peer that failed: it is asked for nothing more.
230	1×	pub fn drop_peer(&mut self, peer: usize) {
231	1×	self.peers.remove(&peer);
232	1×	self.reorder();
233	1×	}
234
235	1354×	fn next_rand(&mut self) -> u64 {
236		// xorshift64: enough to spread runners that start together.
237	1354×	self.seed ^= self.seed << 13;
238	1354×	self.seed ^= self.seed >> 7;
239	1354×	self.seed ^= self.seed << 17;
240	1354×	self.seed
241	1354×	}
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	36×	fn reorder(&mut self) {
249	36×	let run = (self.len() / 256).clamp(1, RUN);
250	1354×	let shuffle: Vec<u64> = (0..self.len().div_ceil(run)).map(\|_\| self.next_rand()).collect();
251	36×	let mut keyed: Vec<(u16, u64, usize)> = Vec::new();
252	33610×	for i in 0..self.len() {
253	33610×	self.holders[i] = self.peers.values().filter(\|(b, _)\| b.get(i)).count().min(u16::MAX as usize) as u16;
254	33610×	self.coming[i] = self.peers.values().any(\|(_, f)\| f.get(i));
255	33610×	if !self.have.get(i) {
256	33472×	keyed.push((self.holders[i], shuffle[i / run], i));
257	33472×	}
258		}
259	36×	keyed.sort_unstable();
260	36×	self.order = keyed.into_iter().map(\|(_, _, i)\| i).collect();
261	33472×	for (at, &i) in self.order.iter().enumerate() {
262	33472×	self.pos[i] = at;
263	33472×	}
264	36×	self.skip = 0;
265	36×	}
266
267		/// The next chunk for `src` to fetch.
268	1294×	pub fn pick(&mut self, src: Source) -> Pick {
269	1294×	if self.missing() == 0 {
270	64×	return Pick::Done;
271	1230×	}
272	9479×	let holds = \|i: usize\| match src {
273	759×	Source::Hub => true,
274	8720×	Source::Peer(p) => self.peers.get(&p).is_some_and(\|(b, _)\| b.get(i)),
275	9479×	};
276	1230×	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	1501×	while self.skip < self.order.len() && (self.have.get(self.order[self.skip]) \|\| self.asked[self.order[self.skip]] > 0) {
280	271×	self.skip += 1;
281	271×	}
282	19213×	let fresh = self.order[self.skip..].iter().copied().find(\|&i\| {
283	19213×	!self.have.get(i)
284	12422×	&& self.asked[i] == 0
285	9477×	&& holds(i)
286	933×	&& (src != Source::Hub \|\| stalled \|\| self.peers.is_empty() \|\| (self.holders[i] == 0 && !self.coming[i]))
287	19213×	});
288	1230×	if let Some(i) = fresh {
289	322×	self.asked[i] += 1;
290	322×	self.first[i] = Some(src);
291	322×	return Pick::Chunk(i);
292	908×	}
293		// The endgame: everything missing is asked for already. The hub
294		// joins it only once the peers stall.
295	908×	if src == Source::Hub && !stalled && !self.peers.is_empty() {
296	265×	return Pick::Wait;
297	643×	}
298	5168×	let mut left = self.order.iter().copied().filter(\|&i\| !self.have.get(i));
299	1959×	if left.clone().all(\|i\| self.asked[i] > 0) {
300	110×	if let Some(i) = left.find(\|&i\| self.asked[i] < ENDGAME_COPIES && self.first[i] != Some(src) && holds(i)) {
301	2×	self.asked[i] += 1;
302	2×	return Pick::Chunk(i);
303	41×	}
304	600×	}
305	641×	Pick::Wait
306	1294×	}
307
308		/// Chunk `i` arrived and was checked.
309	234×	pub fn got(&mut self, i: usize) {
310	234×	self.have.set(i);
311	234×	self.asked[i] = self.asked[i].saturating_sub(1);
312	234×	self.first[i] = None;
313	234×	self.progress = Instant::now();
314	234×	}
315
316		/// Chunk `i` did not come: it can be asked for again.
317	3×	pub fn failed(&mut self, i: usize) {
318	3×	self.asked[i] = self.asked[i].saturating_sub(1);
319	3×	if self.asked[i] == 0 {
320	2×	self.first[i] = None;
321	2×	if !self.have.get(i) {
322	2×	self.skip = self.skip.min(self.pos[i]);
323	2×	}
324	1×	}
325	3×	}
326		}
327
328		#[cfg(test)]
329		mod tests {
330		#[test]
331	1×	fn from_one_peer_neighbouring_chunks_come_in_order() {
332	1×	let n = RUN * 256;
333	1×	let mut p = Picker::new(n, Bits::new(n), 3);
334	1×	p.peer(0, Bits::full(n));
335	1×	let first: Vec<usize> = (0..RUN)
336	64×	.map(\|_\| match p.pick(Source::Peer(0)) {
337	64×	Pick::Chunk(i) => i,
338	0×	other => panic!("{other:?}"),
339	64×	})
340	1×	.collect();
341	1×	assert_eq!(first[0] % RUN, 0, "a run starts at its first chunk: {first:?}");
342	63×	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	1×	p.failed(first[3]);
345	1×	assert_eq!(p.pick(Source::Peer(0)), Pick::Chunk(first[3]));
346	1×	}
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	0×	fn picking_a_big_image_from_one_peer_is_quick() {
353	0×	let n = 142_244;
354	0×	let mut p = Picker::new(n, Bits::new(n), 7);
355	0×	p.peer(0, Bits::full(n));
356	0×	let started = std::time::Instant::now();
357	0×	let mut in_flight = std::collections::VecDeque::new();
358		loop {
359	0×	match p.pick(Source::Peer(0)) {
360	0×	Pick::Chunk(i) => {
361	0×	in_flight.push_back(i);
362	0×	if in_flight.len() >= 4 {
363	0×	p.got(in_flight.pop_front().unwrap());
364	0×	}
365		}
366	0×	Pick::Wait => p.got(in_flight.pop_front().unwrap()),
367	0×	Pick::Done => break,
368		}
369		}
370	0×	eprintln!("picked {n} chunks in {:?}", started.elapsed());
371	0×	}
372
373		use super::*;
374
375		#[test]
376	1×	fn bits_travel_as_bytes() {
377	1×	let mut b = Bits::new(70);
378	5×	[0, 9, 63, 64, 69].iter().for_each(\|&i\| b.set(i));
379	1×	let back = Bits::from_bytes(70, &b.to_bytes());
380	1×	assert_eq!(back, b);
381	1×	assert_eq!(b.to_bytes().len(), 9);
382	1×	assert_eq!(Bits::from_bytes(10, &[0xff, 0xff, 0xff]).count(), 10, "bits past the list are ignored");
383	1×	assert_eq!(Bits::from_bytes(10, &[]).count(), 0, "a short answer leaves the rest unset");
384	1×	assert_eq!(Bits::full(130).count(), 130);
385	1×	}
386
387		#[test]
388	1×	fn a_ticket_opens_only_with_its_hub_and_in_time() {
389	1×	let sk = SecretKey::generate();
390	1×	let hub = sk.identity("hub", "hub@example", 1);
391	1×	let t = Ticket { hub: hub.hash().to_hex(), runner: 7, name: "earth-1".into(), site: "home".into(), expires_ms: 10_000 };
392	1×	let text = t.seal(&hub.hash(), &sk);
393	1×	assert_eq!(Ticket::open(&text, &hub, 9_999).unwrap(), t);
394	1×	assert!(Ticket::open(&text, &hub, 10_000).unwrap_err().contains("expired"));
395	1×	let other_sk = SecretKey::generate();
396	1×	let other = other_sk.identity("other", "o@example", 1);
397	1×	assert!(Ticket::open(&text, &other, 1).is_err(), "another hub's key");
398	1×	let mut forged = text.clone();
399	1×	forged.replace_range(2..4, "ff");
400	1×	assert!(Ticket::open(&forged, &hub, 1).is_err(), "a changed body");
401	1×	assert!(Ticket::open("nonsense", &hub, 1).is_err());
402	1×	}
403
404		#[test]
405	1×	fn peers_are_asked_for_the_rarest_chunk_they_hold() {
406	1×	let mut p = Picker::new(4, Bits::new(4), 42);
407	1×	let mut a = Bits::new(4);
408	3×	[0, 1, 2].iter().for_each(\|&i\| a.set(i));
409	1×	let mut b = Bits::new(4);
410	2×	[0, 1].iter().for_each(\|&i\| b.set(i));
411	1×	p.peer(0, a);
412	1×	p.peer(1, b);
413		// Chunk 2 only peer 0 holds; chunk 3 nobody does.
414	1×	assert_eq!(p.pick(Source::Peer(0)), Pick::Chunk(2));
415	1×	assert_eq!(p.pick(Source::Hub), Pick::Chunk(3), "the hub seeds what no peer holds");
416	1×	assert_eq!(p.pick(Source::Hub), Pick::Wait, "and leaves the rest to the peers");
417	1×	let first = p.pick(Source::Peer(1));
418	1×	assert!(matches!(first, Pick::Chunk(0) \| Pick::Chunk(1)));
419	1×	}
420
421		#[test]
422	1×	fn the_hub_is_not_asked_for_what_a_peer_is_fetching() {
423	1×	let mut p = Picker::new(2, Bits::new(2), 4);
424	1×	let mut fetching = Bits::new(2);
425	1×	fetching.set(0);
426	1×	p.peer_fetching(0, Bits::new(2), fetching);
427	1×	assert_eq!(p.pick(Source::Hub), Pick::Chunk(1));
428	1×	assert_eq!(p.pick(Source::Hub), Pick::Wait, "chunk 0 is coming from the peer");
429	1×	let mut have = Bits::new(2);
430	1×	have.set(0);
431	1×	p.peer(0, have);
432	1×	assert_eq!(p.pick(Source::Peer(0)), Pick::Chunk(0));
433	1×	}
434
435		#[test]
436	1×	fn without_peers_the_hub_serves_everything() {
437	1×	let mut p = Picker::new(3, Bits::new(3), 1);
438	1×	let mut seen = Vec::new();
439	4×	while let Pick::Chunk(i) = p.pick(Source::Hub) {
440	3×	seen.push(i);
441	3×	p.got(i);
442	3×	}
443	1×	seen.sort();
444	1×	assert_eq!(seen, vec![0, 1, 2]);
445	1×	assert_eq!(p.pick(Source::Hub), Pick::Done);
446	1×	}
447
448		#[test]
449	1×	fn the_endgame_never_asks_the_same_source_twice() {
450	1×	let mut p = Picker::new(1, Bits::new(1), 9);
451	1×	assert_eq!(p.pick(Source::Hub), Pick::Chunk(0));
452	1×	assert_eq!(p.pick(Source::Hub), Pick::Wait);
453	1×	p.peer(0, Bits::full(1));
454	1×	assert_eq!(p.pick(Source::Peer(0)), Pick::Chunk(0), "a peer may race the hub for it");
455	1×	assert_eq!(p.pick(Source::Hub), Pick::Wait, "but not the hub a peer, while the peers make progress");
456	1×	}
457
458		#[test]
459	1×	fn the_endgame_asks_a_second_source_and_a_failure_frees_the_chunk() {
460	1×	let mut p = Picker::new(1, Bits::new(1), 3);
461	1×	p.peer(0, Bits::full(1));
462	1×	p.peer(1, Bits::full(1));
463	1×	assert_eq!(p.pick(Source::Peer(0)), Pick::Chunk(0));
464	1×	assert_eq!(p.pick(Source::Peer(1)), Pick::Chunk(0), "the endgame");
465	1×	assert_eq!(p.pick(Source::Peer(1)), Pick::Wait, "at most two copies");
466	1×	p.failed(0);
467	1×	p.failed(0);
468	1×	p.drop_peer(0);
469	1×	assert_eq!(p.pick(Source::Peer(1)), Pick::Chunk(0));
470	1×	p.got(0);
471	1×	assert_eq!(p.pick(Source::Peer(1)), Pick::Done);
472	1×	}
473
474		#[test]
475	1×	fn chunks_already_here_are_never_asked_for() {
476	1×	let mut have = Bits::new(3);
477	1×	have.set(1);
478	1×	let mut p = Picker::new(3, have, 5);
479	1×	assert_eq!(p.missing(), 2);
480	1×	let mut seen = Vec::new();
481	3×	while let Pick::Chunk(i) = p.pick(Source::Hub) {
482	2×	seen.push(i);
483	2×	p.got(i);
484	2×	}
485	1×	assert!(!seen.contains(&1));
486	1×	}
487
488		#[test]
489	1×	fn runners_starting_together_ask_the_hub_for_different_chunks() {
490	1×	let a: Vec<_> = {
491	1×	let mut p = Picker::new(64, Bits::new(64), 11);
492	8×	(0..8).filter_map(\|_\| if let Pick::Chunk(i) = p.pick(Source::Hub) { Some(i) } else { None }).collect()
493		};
494	1×	let b: Vec<_> = {
495	1×	let mut p = Picker::new(64, Bits::new(64), 12);
496	8×	(0..8).filter_map(\|_\| if let Pick::Chunk(i) = p.pick(Source::Hub) { Some(i) } else { None }).collect()
497		};
498	1×	assert_ne!(a, b);
499	1×	}
500		}

Coverage / crates/jet/src/swarm.rs 91.4%

Keyboard