swarm.rs 96.2%

279 of 290 lines run, 77 of 88 functions · run #197 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);
/// 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>,
    /// 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(),
            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
    }

    fn reorder(&mut self) {
        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) {
                let r = self.next_rand();
                keyed.push((self.holders[i], r, i));
            }
        }
        keyed.sort_unstable();
        self.order = keyed.into_iter().map(|(_, _, i)| i).collect();
    }

    /// 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.
        let fresh = self.order.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;
        }
    }
}

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

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

Keyboard