Jetstream: runners share images peer to peer b14aeaae683a

//! The swarm (oxygenflow#47, docs/jetstream.md §8.1): a runner with a

//! `[runner.swarm]` section serves the image chunks it holds to its peers,

//! and fetches an image it lacks from its peers and the hub together.

//!

//! The server is a small HTTP/1.1 endpoint on the swarm address: a peer

//! shows the ticket its hub signed and gets a session, then asks which

//! chunks of an image this runner has and for chunks by hash. A chunk is

//! served from the chunk cache, or read out of a whole image by its

//! manifest. Every chunk is checked by the side that asked for it, so a

//! peer can only fail to help.

use crate::home::{Home, RunnerConfig};

use crate::images::{cache_path, chunk_cache, manifest_path, save_manifest, write_image, Client, Moved};

use anyhow::{bail, Context, Result};

use flow_core::api::{SwarmAddr, SwarmPeer};

use jet::images::{is_hash, Manifest};

use jet::swarm::{Bits, Pick, Picker, Source, Ticket};

use oxy_core::identity::Identity;

use std::collections::HashMap;

use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write};

use std::net::{TcpListener, TcpStream};

use std::path::{Path, PathBuf};

use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};

use std::sync::{Arc, Mutex};

use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

/// Connections served at once; more are told 503 and try elsewhere.

const MAX_CONNS: usize = 16;

/// Requests in flight to one peer, and to the hub.

const PEER_WORKERS: usize = 4;

const HUB_WORKERS: usize = 8;

/// How often a fetcher asks its peers what they have.

const REFRESH: Duration = Duration::from_millis(500);

/// Hub failures a fetch survives before it gives up.

const HUB_FAILURES: usize = 5;

/// The largest request body the server reads: a ticket.

const MAX_BODY: usize = 64 * 1024;

/// Names the part files of chunks two sources fetch at once apart.

static PART: AtomicU64 = AtomicU64::new(0);

fn now_ms() -> u64 {

    SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0)

}

/// An image this runner offers: which of its needed chunks are here

/// (`None`: all of them, the image is whole) and which it is fetching now.

struct Offer {

    bits: Option<Bits>,

    fetching: Bits,

}

struct State {

    root: PathBuf,

    site: String,

    client: Client,

    /// The hub's identity, which signs tickets; fetched when first needed.

    hub: Mutex<Option<Identity>>,

    offers: Mutex<HashMap<String, Offer>>,

    /// Where a chunk lies in a whole image: the image, offset and length.

    located: Mutex<HashMap<String, (PathBuf, u64, u32)>>,

    /// Sessions: token to (peer name, expiry).

    sessions: Mutex<HashMap<String, (String, u64)>>,

    conns: AtomicUsize,

    /// The upload allowance: bytes in hand (negative: owed) and when it

    /// was last topped up.

    bucket: Mutex<(f64, Instant)>,

    rate: f64,

}

/// This runner's place in the swarm.

pub struct Swarm {

    state: Arc<State>,

    pub addr: SwarmAddr,

    peer: ureq::Agent,

}

impl Swarm {

    /// Start serving peers when the runner is configured for it.

    pub fn start(home: &Home, cfg: &RunnerConfig) -> Result<Option<Swarm>> {

        let Some(sc) = &cfg.swarm else { return Ok(None) };

        let listener = TcpListener::bind(&sc.listen).with_context(|| format!("swarm: listen on {}", sc.listen))?;

        let client = Client::new(&cfg.hub, &cfg.token, cfg.ca.as_deref())?;

        let hub = client.hub_identity().map_err(|e| eprintln!("swarm: the hub's key: {e:#}; trying again when a peer comes")).ok();

        let rate = sc.upload_mib_s.unwrap_or(50).max(1) as f64 * (1 << 20) as f64;

        let state = Arc::new(State {

            root: home.root.clone(),

            site: sc.site.clone(),

            client,

            hub: Mutex::new(hub),

            offers: Mutex::new(HashMap::new()),

            located: Mutex::new(HashMap::new()),

            sessions: Mutex::new(HashMap::new()),

            conns: AtomicUsize::new(0),

            bucket: Mutex::new((rate, Instant::now())),

            rate,

        });

        // Whole images: offered from their manifests; one without a

        // manifest (imported here, not fetched) is scanned for one, in the

        // background and one at a time.

        let mut unscanned = Vec::new();

        for h in home.images() {

            match std::fs::read(manifest_path(&home.root, &h)).ok().and_then(|b| serde_json::from_slice::<Manifest>(&b).ok()) {

                Some(m) if m.hash == h => offer_whole(&state, &m, &home.rootfs(&h)),

                _ => unscanned.push((h.clone(), home.rootfs(&h), home.image_arch(&h))),

            }

        }

        let scanning = state.clone();

        std::thread::spawn(move || {

            for (h, path, arch) in unscanned {

                match jet::images::scan_file(&path, arch, |_, _| Ok(())) {

                    Ok(m) if m.hash == h => {

                        save_manifest(&scanning.root, &m).ok();

                        offer_whole(&scanning, &m, &path);

                    }

                    Ok(_) => eprintln!("swarm: {} does not hold the image its name says; not offering it", path.display()),

                    Err(e) => eprintln!("swarm: scan {}: {e}", path.display()),

                }

            }

        });

        let serving = state.clone();

        std::thread::spawn(move || accept(listener, serving));

        let addr = SwarmAddr { addr: sc.advertise.clone().unwrap_or_else(|| sc.listen.clone()), site: sc.site.clone() };

        eprintln!("swarm: serving image chunks on {} (site {:?}, up to {} MiB/s)", sc.listen, sc.site, sc.upload_mib_s.unwrap_or(50));

        let peer = ureq::AgentBuilder::new().timeout_connect(Duration::from_secs(2)).timeout(Duration::from_secs(20)).build();

        Ok(Some(Swarm { state, addr, peer }))

    }

    /// Fetch image `hash` into `home` from the peers and the hub (§8.1):

    /// the rarest chunks from peers first, the hub for what no peer has.

    /// `Ok(None)` when the hub does not hold it.

    pub fn fetch(&self, home: &Home, hash: &str) -> Result<Option<Moved>> {

        let start = Instant::now();

        let client = &self.state.client;

        let Some(m) = client.manifest(hash)? else { return Ok(None) };

        if m.hash != hash {

            bail!("the hub's manifest for {} names {}", &hash[..12], &m.hash[..12]);

        }

        save_manifest(&home.root, &m)?;

        let needed: Vec<String> = m.needed().into_iter().map(String::from).collect();

        let cache = chunk_cache(home);

        let mut have = Bits::new(needed.len());

        for (i, h) in needed.iter().enumerate() {

            if cache_path(&cache, h).is_file() {

                have.set(i);

            }

        }

        self.state.offers.lock().unwrap().insert(hash.into(), Offer { bits: Some(have.clone()), fetching: Bits::new(needed.len()) });

        let announced = client

            .announce(hash, "fetching")

            .map_err(|e| eprintln!("swarm: announce {}: {e:#}; fetching from the hub alone", &hash[..12]))

            .ok();

        let seed = now_ms() ^ (std::process::id() as u64) << 32;

        let picker = Mutex::new(Picker::new(needed.len(), have, seed));

        let peers: Mutex<Vec<Arc<Peer>>> = Mutex::new(Vec::new());

        if let Some(a) = &announced {

            self.join(&a.peers, &a.ticket, hash, needed.len(), &peers, &picker);

        }

        let (bytes, peer_bytes, got) = (AtomicU64::new(0), AtomicU64::new(0), AtomicUsize::new(0));

        let (failed, hub_failures) = (Mutex::new(None::<anyhow::Error>), AtomicUsize::new(0));

        {

            let (picker, peers, needed, cache) = (&picker, &peers, &needed, &cache);

            let (bytes, peer_bytes, got, failed, hub_failures) = (&bytes, &peer_bytes, &got, &failed, &hub_failures);

            let (state, m) = (&self.state, &m);

            let worker = move |src: Source| loop {

                if failed.lock().unwrap().is_some() {

                    break;

                }

                let peer = match src {

                    Source::Peer(p) => Some(peers.lock().unwrap()[p].clone()),

                    Source::Hub => None,

                };

                if peer.as_ref().is_some_and(|p| !p.alive.load(Ordering::Relaxed)) {

                    break;

                }

                // Taken apart from the match, so the lock is not held

                // while waiting.

                let pick = picker.lock().unwrap().pick(src);

                let i = match pick {

                    Pick::Done => break,

                    Pick::Wait => {

                        std::thread::sleep(Duration::from_millis(20));

                        continue;

                    }

                    Pick::Chunk(i) => i,

                };

                if let Some(o) = state.offers.lock().unwrap().get_mut(&m.hash) {

                    o.fetching.set(i);

                }

                let h = &needed[i];

                let fetched = match &peer {

                    Some(p) => self.peer_chunk(p, h),

                    None => client.chunk(h),

                };

                let kept = fetched.and_then(|b| {

                    if blake3::hash(&b).to_hex().as_str() != h {

                        bail!("chunk {} came back with other bytes", &h[..12]);

                    }

                    let p = cache_path(cache, h);

                    std::fs::create_dir_all(p.parent().expect("a chunk directory"))?;

                    let tmp = p.with_extension(format!("part{}", PART.fetch_add(1, Ordering::Relaxed)));

                    std::fs::write(&tmp, &b)?;

                    std::fs::rename(&tmp, &p)?;

                    Ok(b.len() as u64)

                });

                match kept {

                    Ok(n) => {

                        if let Some(Offer { bits: Some(b), fetching }) = state.offers.lock().unwrap().get_mut(&m.hash) {

                            b.set(i);

                            fetching.clear(i);

                        }

                        picker.lock().unwrap().got(i);

                        bytes.fetch_add(n, Ordering::Relaxed);

                        got.fetch_add(1, Ordering::Relaxed);

                        if peer.is_some() {

                            peer_bytes.fetch_add(n, Ordering::Relaxed);

                        }

                    }

                    Err(e) => {

                        picker.lock().unwrap().failed(i);

                        if let Some(o) = state.offers.lock().unwrap().get_mut(&m.hash) {

                            o.fetching.clear(i);

                        }

                        match (&peer, src) {

                            (Some(p), Source::Peer(idx)) => {

                                eprintln!("swarm: peer {} failed ({e:#}); carrying on without it", p.name);

                                p.alive.store(false, Ordering::Relaxed);

                                picker.lock().unwrap().drop_peer(idx);

                                break;

                            }

                            _ => {

                                if hub_failures.fetch_add(1, Ordering::Relaxed) + 1 >= HUB_FAILURES {

                                    *failed.lock().unwrap() = Some(e);

                                    break;

                                }

                                std::thread::sleep(Duration::from_millis(500));

                            }

                        }

                    }

                }

            };

            std::thread::scope(|s| {

                for _ in 0..HUB_WORKERS {

                    s.spawn(move || worker(Source::Hub));

                }

                let mut spawned = 0;

                let mut spawn_new = || {

                    let n = peers.lock().unwrap().len();

                    for p in spawned..n {

                        for _ in 0..PEER_WORKERS {

                            s.spawn(move || worker(Source::Peer(p)));

                        }

                    }

                    spawned = n;

                };

                spawn_new();

                // Keep the peers' maps fresh, and find new peers while

                // fetching, until every chunk is here.

                let mut asked_hub = Instant::now();

                let interval = Duration::from_secs(announced.as_ref().map(|a| a.interval_s).unwrap_or(30).max(5));

                let mut last = Instant::now();

                while picker.lock().unwrap().missing() > 0 && failed.lock().unwrap().is_none() {

                    std::thread::sleep(Duration::from_millis(100));

                    if last.elapsed() < REFRESH {

                        continue;

                    }

                    last = Instant::now();

                    let now: Vec<(usize, Arc<Peer>)> =

                        peers.lock().unwrap().iter().cloned().enumerate().filter(|(_, p)| p.alive.load(Ordering::Relaxed)).collect();

                    for (idx, p) in now {

                        match self.peer_have(&p, hash, needed.len()) {

                            Ok((bits, fetching)) => picker.lock().unwrap().peer_fetching(idx, bits, fetching),

                            Err(_) => {

                                p.alive.store(false, Ordering::Relaxed);

                                picker.lock().unwrap().drop_peer(idx);

                            }

                        }

                    }

                    if announced.is_some() && asked_hub.elapsed() >= interval {

                        asked_hub = Instant::now();

                        if let Ok(a) = client.announce(hash, "fetching") {

                            self.join(&a.peers, &a.ticket, hash, needed.len(), peers, picker);

                            spawn_new();

                        }

                    }

                }

            });

        }

        if let Some(e) = failed.into_inner().unwrap() {

            client.announce(hash, "stopped").ok();

            return Err(e);

        }

        write_image(home, &m)?;

        offer_whole(&self.state, &m, &home.rootfs(hash));

        client.announce(hash, "done").ok();

        Ok(Some(Moved {

            chunks: got.into_inner(),

            bytes: bytes.into_inner(),

            peer_bytes: peer_bytes.into_inner(),

            elapsed: start.elapsed(),

        }))

    }

    /// Admit this runner to the peers it does not know yet: a session from

    /// each, and what each has.

    fn join(&self, found: &[SwarmPeer], ticket: &str, hash: &str, len: usize, peers: &Mutex<Vec<Arc<Peer>>>, picker: &Mutex<Picker>) {

        for sp in found {

            if peers.lock().unwrap().iter().any(|p| p.runner == sp.runner) {

                continue;

            }

            let joined = self.hello(&sp.addr, ticket).and_then(|session| {

                let p = Peer {

                    runner: sp.runner,

                    name: sp.name.clone(),

                    base: format!("http://{}/swarm/v1", sp.addr),

                    session,

                    alive: AtomicBool::new(true),

                };

                let (bits, fetching) = self.peer_have(&p, hash, len)?;

                Ok((p, bits, fetching))

            });

            match joined {

                Ok((p, bits, fetching)) => {

                    let mut list = peers.lock().unwrap();

                    picker.lock().unwrap().peer_fetching(list.len(), bits, fetching);

                    list.push(Arc::new(p));

                }

                Err(e) => eprintln!("swarm: peer {} ({}): {e:#}", sp.name, sp.addr),

            }

        }

    }

    fn hello(&self, addr: &str, ticket: &str) -> Result<String> {

        let url = format!("http://{addr}/swarm/v1/hello");

        let r: serde_json::Value = self.peer.post(&url).send_string(ticket).map_err(|e| peer_error(&url, e))?.into_json()?;

        r["session"].as_str().map(String::from).context("the peer gave no session")

    }

    /// What a peer has of image `hash`, and what it is fetching now:

    /// nothing yet when it has not begun (the tracker may name it a moment

    /// before it starts).

    fn peer_have(&self, p: &Peer, hash: &str, len: usize) -> Result<(Bits, Bits)> {

        let url = format!("{}/images/{hash}/have", p.base);

        let r = match self.peer.get(&url).set("Authorization", &format!("Bearer {}", p.session)).call() {

            Ok(r) => r,

            Err(ureq::Error::Status(404, _)) => return Ok((Bits::new(len), Bits::new(len))),

            Err(e) => return Err(peer_error(&url, e)),

        };

        let mut body = Vec::new();

        r.into_reader().take(len as u64 / 4 + 64).read_to_end(&mut body)?;

        if body == b"complete" {

            return Ok((Bits::full(len), Bits::new(len)));

        }

        let (have, fetching) = body.split_at(len.div_ceil(8).min(body.len()));

        Ok((Bits::from_bytes(len, have), Bits::from_bytes(len, fetching)))

    }

    fn peer_chunk(&self, p: &Peer, hash: &str) -> Result<Vec<u8>> {

        let url = format!("{}/chunks/{hash}", p.base);

        let r = self.peer.get(&url).set("Authorization", &format!("Bearer {}", p.session)).call().map_err(|e| peer_error(&url, e))?;

        let mut out = Vec::new();

        r.into_reader().take(oxy_core::chunk::MAX_SIZE as u64 + 1).read_to_end(&mut out)?;

        Ok(out)

    }

}

/// A peer this fetch asks for chunks.

struct Peer {

    runner: i64,

    name: String,

    base: String,

    session: String,

    alive: AtomicBool,

}

fn peer_error(url: &str, e: ureq::Error) -> anyhow::Error {

    match e {

        ureq::Error::Status(code, r) => anyhow::anyhow!("{url}: HTTP {code}: {}", r.into_string().unwrap_or_default().trim()),

        e => anyhow::anyhow!("{url}: {e}"),

    }

}

/// Offer a whole image: every chunk of it can be read out of `path`.

fn offer_whole(state: &State, m: &Manifest, path: &Path) {

    let mut located = state.located.lock().unwrap();

    let mut at = 0u64;

    for c in &m.chunks {

        if !c.is_zeros() {

            located.entry(c.hash().to_string()).or_insert_with(|| (path.to_path_buf(), at, c.1));

        }

        at += c.1 as u64;

    }

    state.offers.lock().unwrap().insert(m.hash.clone(), Offer { bits: None, fetching: Bits::new(0) });

}

fn accept(listener: TcpListener, state: Arc<State>) {

    for conn in listener.incoming() {

        let Ok(mut conn) = conn else { continue };

        if state.conns.load(Ordering::Relaxed) >= MAX_CONNS {

            respond(&mut conn, 503, "text/plain", b"busy: ask another peer").ok();

            continue;

        }

        state.conns.fetch_add(1, Ordering::Relaxed);

        let state = state.clone();

        std::thread::spawn(move || {

            serve_conn(conn, &state).ok();

            state.conns.fetch_sub(1, Ordering::Relaxed);

        });

    }

}

fn respond(conn: &mut TcpStream, code: u16, kind: &str, body: &[u8]) -> std::io::Result<()> {

    let reason = match code {

        200 => "OK",

        400 => "Bad Request",

        401 => "Unauthorized",

        403 => "Forbidden",

        404 => "Not Found",

        503 => "Service Unavailable",

        _ => "Error",

    };

    write!(conn, "HTTP/1.1 {code} {reason}\r\nContent-Type: {kind}\r\nContent-Length: {}\r\n\r\n", body.len())?;

    conn.write_all(body)?;

    conn.flush()

}

/// One connection: requests one after another while the peer keeps it.

fn serve_conn(conn: TcpStream, state: &State) -> std::io::Result<()> {

    conn.set_read_timeout(Some(Duration::from_secs(30)))?;

    let mut out = conn.try_clone()?;

    let mut r = BufReader::new(conn);

    loop {

        let mut line = String::new();

        if r.read_line(&mut line)? == 0 {

            return Ok(());

        }

        let mut parts = line.split_whitespace();

        let (method, path) = (parts.next().unwrap_or("").to_string(), parts.next().unwrap_or("").to_string());

        let (mut length, mut auth, mut close) = (0usize, String::new(), false);

        loop {

            let mut h = String::new();

            if r.read_line(&mut h)? == 0 {

                return Ok(());

            }

            let h = h.trim_end();

            if h.is_empty() {

                break;

            }

            if let Some((k, v)) = h.split_once(':') {

                match k.trim().to_ascii_lowercase().as_str() {

                    "content-length" => length = v.trim().parse().unwrap_or(usize::MAX),

                    "authorization" => auth = v.trim().to_string(),

                    "connection" => close = v.trim().eq_ignore_ascii_case("close"),

                    _ => {}

                }

            }

        }

        if length > MAX_BODY {

            return respond(&mut out, 400, "text/plain", b"too large");

        }

        let mut body = vec![0; length];

        r.read_exact(&mut body)?;

        let (code, kind, reply) = route(state, &method, &path, &auth, &body);

        if code == 200 && kind == "application/octet-stream" {

            throttle(state, reply.len());

        }

        respond(&mut out, code, kind, &reply)?;

        if close {

            return Ok(());

        }

    }

}

fn route(state: &State, method: &str, path: &str, auth: &str, body: &[u8]) -> (u16, &'static str, Vec<u8>) {

    let text = |code, s: &str| (code, "text/plain", s.as_bytes().to_vec());

    if method == "POST" && path == "/swarm/v1/hello" {

        return match hello(state, &String::from_utf8_lossy(body)) {

            Ok(session) => (200, "application/json", serde_json::json!({ "session": session }).to_string().into_bytes()),

            Err(why) => text(403, &why),

        };

    }

    let session = auth.strip_prefix("Bearer ").unwrap_or("");

    let admitted = {

        let mut sessions = state.sessions.lock().unwrap();

        let now = now_ms();

        sessions.retain(|_, (_, until)| *until > now);

        sessions.contains_key(session)

    };

    if !admitted {

        return text(401, "show a ticket at /swarm/v1/hello first");

    }

    let rest = path.strip_prefix("/swarm/v1/").unwrap_or("");

    match (method, rest.split('/').collect::<Vec<_>>().as_slice()) {

        ("GET", ["images", image, "have"]) => match state.offers.lock().unwrap().get(*image) {

            Some(Offer { bits: None, .. }) => (200, "text/plain", b"complete".to_vec()),

            Some(Offer { bits: Some(b), fetching }) => (200, "application/x-have", [b.to_bytes(), fetching.to_bytes()].concat()),

            None => text(404, "this runner does not hold that image"),

        },

        ("GET", ["chunks", chunk]) if is_hash(chunk) => match read_chunk(state, chunk) {

            Some(b) => (200, "application/octet-stream", b),

            None => text(404, "no such chunk here"),

        },

        _ => text(404, "not a swarm path"),

    }

}

/// Admit a peer that shows a ticket from this runner's hub, for this

/// site: a session until the ticket expires.

fn hello(state: &State, ticket: &str) -> std::result::Result<String, String> {

    let hub = {

        let mut hub = state.hub.lock().unwrap();

        if hub.is_none() {

            *hub = state.client.hub_identity().ok();

        }

        hub.clone().ok_or("this runner cannot reach its hub to check tickets")?

    };

    let t = Ticket::open(ticket, &hub, now_ms())?;

    if t.site != state.site {

        return Err(format!("this runner is in site {:?}, the ticket's runner in {:?}", state.site, t.site));

    }

    let session = random_token().map_err(|e| format!("make a session: {e}"))?;

    state.sessions.lock().unwrap().insert(session.clone(), (t.name, t.expires_ms));

    Ok(session)

}

fn random_token() -> std::io::Result<String> {

    let mut b = [0u8; 32];

    std::fs::File::open("/dev/urandom")?.read_exact(&mut b)?;

    Ok(hex::encode(b))

}

fn read_chunk(state: &State, hash: &str) -> Option<Vec<u8>> {

    if let Ok(b) = std::fs::read(cache_path(&state.root.join("chunks"), hash)) {

        return Some(b);

    }

    let (path, at, len) = state.located.lock().unwrap().get(hash).cloned()?;

    let mut f = std::fs::File::open(path).ok()?;

    f.seek(SeekFrom::Start(at)).ok()?;

    let mut b = vec![0; len as usize];

    f.read_exact(&mut b).ok()?;

    Some(b)

}

/// Hold back until sending `n` more bytes keeps within the upload rate.

fn throttle(state: &State, n: usize) {

    let wait = {

        let mut b = state.bucket.lock().unwrap();

        let now = Instant::now();

        b.0 = (b.0 + now.duration_since(b.1).as_secs_f64() * state.rate).min(state.rate);

        b.1 = now;

        b.0 -= n as f64;

        if b.0 < 0.0 {

            -b.0 / state.rate

        } else {

            0.0

        }

    };

    if wait > 0.0 {

        std::thread::sleep(Duration::from_secs_f64(wait));

    }

}

#[cfg(test)]

mod tests {

    use super::*;

    use oxy_core::identity::SecretKey;

    /// A hub that holds one image and runs the tracker: it serves the

    /// manifest and chunks (counting the chunks), and answers announces

    /// with every other runner as a peer and a ticket for the caller,

    /// known by its token.

    struct FakeHub {

        url: String,

        chunks_served: Arc<AtomicUsize>,

        runners: Arc<Mutex<Vec<(String, SwarmPeer)>>>,

    }

    fn fake_hub(m: Manifest, image: Vec<u8>, site: &str) -> FakeHub {

        slow_hub(m, image, site, Duration::ZERO)

    }

    /// A hub that takes `delay` over each chunk.

    fn slow_hub(m: Manifest, image: Vec<u8>, site: &str, delay: Duration) -> FakeHub {

        let sk = SecretKey::generate();

        let id = sk.identity("hub", "hub@example", 1);

        let key = id.hash();

        let mut chunks = HashMap::new();

        let mut at = 0usize;

        for c in &m.chunks {

            chunks.insert(c.hash().to_string(), image[at..at + c.len()].to_vec());

            at += c.len();

        }

        let listener = TcpListener::bind("127.0.0.1:0").unwrap();

        let url = format!("http://{}", listener.local_addr().unwrap());

        let served = Arc::new(AtomicUsize::new(0));

        let runners: Arc<Mutex<Vec<(String, SwarmPeer)>>> = Arc::new(Mutex::new(Vec::new()));

        let (count, known, site) = (served.clone(), runners.clone(), site.to_string());

        let shared = Arc::new((m, chunks, id, key, sk));

        std::thread::spawn(move || {

            for conn in listener.incoming() {

                let Ok(conn) = conn else { continue };

                let (shared, count, known, site) = (shared.clone(), count.clone(), known.clone(), site.clone());

                std::thread::spawn(move || {

                    let (m, chunks, id, key, sk) = &*shared;

                    let mut out = conn.try_clone().unwrap();

                    let mut r = BufReader::new(conn);

                    loop {

                        let mut line = String::new();

                        if r.read_line(&mut line).unwrap_or(0) == 0 {

                            return;

                        }

                        let path = line.split_whitespace().nth(1).unwrap_or("").to_string();

                        let (mut len, mut token) = (0, String::new());

                        loop {

                            let mut h = String::new();

                            r.read_line(&mut h).unwrap();

                            let h = h.trim_end().to_string();

                            if h.is_empty() {

                                break;

                            }

                            let lower = h.to_ascii_lowercase();

                            if let Some(v) = lower.strip_prefix("content-length:") {

                                len = v.trim().parse().unwrap();

                            }

                            if let Some(v) = h.strip_prefix("Authorization: Bearer ") {

                                token = v.to_string();

                            }

                        }

                        let mut body = vec![0; len];

                        r.read_exact(&mut body).unwrap();

                        let (code, reply): (u16, Vec<u8>) = if path == "/.well-known/oxyhub-key" {

                            (200, id.encode())

                        } else if path == format!("/api/v1/flow/images/{}", m.hash) {

                            (200, serde_json::to_vec(m).unwrap())

                        } else if let Some(c) = path.strip_prefix("/api/v1/flow/images/chunks/") {

                            count.fetch_add(1, Ordering::Relaxed);

                            std::thread::sleep(delay);

                            (200, chunks[c].clone())

                        } else if path == "/api/v1/flow/swarm/announce" {

                            let runners = known.lock().unwrap();

                            let me = runners.iter().find(|(t, _)| *t == token).map(|(_, p)| p.clone()).unwrap();

                            let ticket = Ticket {

                                hub: key.to_hex(),

                                runner: me.runner,

                                name: me.name.clone(),

                                site: site.clone(),

                                expires_ms: now_ms() + 60_000,

                            };

                            let peers: Vec<SwarmPeer> = runners.iter().filter(|(t, _)| *t != token).map(|(_, p)| p.clone()).collect();

                            let a = flow_core::api::Announced { ticket: ticket.seal(key, sk), peers, interval_s: 5 };

                            (200, serde_json::to_vec(&a).unwrap())

                        } else {

                            (404, b"no".to_vec())

                        };

                        let head = format!("HTTP/1.1 {code} X\r\nContent-Length: {}\r\n\r\n", reply.len());

                        if out.write_all(head.as_bytes()).and_then(|_| out.write_all(&reply)).is_err() {

                            return;

                        }

                    }

                });

            }

        });

        FakeHub { url, chunks_served: served, runners }

    }

    fn free_port() -> u16 {

        TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port()

    }

    /// A runner home in `dir` with a swarm on `port`, known to `hub` by

    /// `token` as runner `id`.

    fn runner(dir: &Path, hub: &FakeHub, id: i64, token: &str, port: u16, site: &str) -> (Home, Swarm) {

        let toml = format!(

            "hub = \"{}\"\ntoken = \"{token}\"\nname = \"r{id}\"\n[swarm]\nlisten = \"127.0.0.1:{port}\"\nsite = \"{site}\"\n",

            hub.url

        );

        std::fs::write(dir.join(crate::home::RUNNER_TOML), toml).unwrap();

        let home = Home::locate(Some(dir)).unwrap();

        let cfg = home.config.runner.clone().unwrap();

        let swarm = Swarm::start(&home, &cfg).unwrap().unwrap();

        hub.runners

            .lock()

            .unwrap()

            .push((token.into(), SwarmPeer { runner: id, name: format!("r{id}"), addr: format!("127.0.0.1:{port}"), complete: true }));

        (home, swarm)

    }

    fn image() -> Vec<u8> {

        // 3 MiB of varied bytes, with a run of zeros the swarm never moves.

        let mut x = 0x9e3779b97f4a7c15u64;

        let mut b: Vec<u8> = (0..3 << 20)

            .map(|_| {

                x ^= x << 13;

                x ^= x >> 7;

                x ^= x << 17;

                x as u8

            })

            .collect();

        b[1 << 20..(1 << 20) + (300 << 10)].fill(0);

        b

    }

    #[test]

    fn a_runner_fetches_an_image_from_its_peer_and_not_the_hub() {

        let bytes = image();

        let m = jet::images::scan(&bytes[..], Some("x86_64".into()), |_, _| Ok(())).unwrap();

        let hub = fake_hub(m.clone(), bytes.clone(), "lab");

        let (a_dir, b_dir) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());

        // Runner A holds the image, imported rather than fetched: its swarm

        // scans it for a manifest before offering it.

        std::fs::create_dir_all(a_dir.path().join("cas")).unwrap();

        std::fs::write(a_dir.path().join("cas").join(format!("{}.ext4", m.hash)), &bytes).unwrap();

        let (_a_home, a) = runner(a_dir.path(), &hub, 1, "ta", free_port(), "lab");

        let deadline = Instant::now() + Duration::from_secs(20);

        while !a.state.offers.lock().unwrap().contains_key(&m.hash) {

            assert!(Instant::now() < deadline, "runner A never offered its image");

            std::thread::sleep(Duration::from_millis(20));

        }

        let (b_home, b) = runner(b_dir.path(), &hub, 2, "tb", free_port(), "lab");

        let moved = b.fetch(&b_home, &m.hash).unwrap().unwrap();

        assert_eq!(std::fs::read(b_home.rootfs(&m.hash)).unwrap(), bytes, "the image came back whole");

        assert_eq!(moved.peer_bytes, moved.bytes, "every chunk came from the peer");

        assert_eq!(hub.chunks_served.load(Ordering::Relaxed), 0, "the hub sent no chunks");

        assert_eq!(moved.chunks, m.needed().len());

        // B now offers the whole image too, read out of its rootfs.

        assert!(matches!(b.state.offers.lock().unwrap().get(&m.hash), Some(Offer { bits: None, .. })));

    }

    #[test]

    fn runners_starting_together_share_what_each_has_fetched() {

        let bytes = image();

        let m = jet::images::scan(&bytes[..], None, |_, _| Ok(())).unwrap();

        let needed = m.needed().len();

        // A slow hub, as a remote one is: the runners have time to swap.

        let hub = slow_hub(m.clone(), bytes.clone(), "lab", Duration::from_millis(400));

        let (b_dir, c_dir) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());

        let (b_home, b) = runner(b_dir.path(), &hub, 2, "tb", free_port(), "lab");

        let (c_home, c) = runner(c_dir.path(), &hub, 3, "tc", free_port(), "lab");

        let (mb, mc) = std::thread::scope(|s| {

            let fb = s.spawn(|| b.fetch(&b_home, &m.hash).unwrap().unwrap());

            let fc = s.spawn(|| c.fetch(&c_home, &m.hash).unwrap().unwrap());

            (fb.join().unwrap(), fc.join().unwrap())

        });

        assert_eq!(std::fs::read(b_home.rootfs(&m.hash)).unwrap(), bytes);

        assert_eq!(std::fs::read(c_home.rootfs(&m.hash)).unwrap(), bytes);

        let served = hub.chunks_served.load(Ordering::Relaxed);

        assert!(served < 2 * needed, "the hub sent {served} chunks for two runners needing {needed} each");

        assert!(mb.peer_bytes + mc.peer_bytes > 0, "they shared");

    }

    #[test]

    fn without_peers_the_hub_sends_everything_and_a_ticket_from_another_site_is_refused() {

        let bytes = image();

        let m = jet::images::scan(&bytes[..], None, |_, _| Ok(())).unwrap();

        let hub = fake_hub(m.clone(), bytes.clone(), "lab");

        let dir = tempfile::tempdir().unwrap();

        let (home, s) = runner(dir.path(), &hub, 1, "t1", free_port(), "lab");

        let moved = s.fetch(&home, &m.hash).unwrap().unwrap();

        assert_eq!(moved.peer_bytes, 0);

        assert_eq!(hub.chunks_served.load(Ordering::Relaxed), m.needed().len());

        assert_eq!(std::fs::read(home.rootfs(&m.hash)).unwrap(), bytes);

        // A runner in another site shows a ticket for "lab": refused.

        let other = tempfile::tempdir().unwrap();

        let port = free_port();

        let (_h, o) = runner(other.path(), &hub, 3, "t3", port, "elsewhere");

        let ticket = s.state.client.announce(&m.hash, "fetching").unwrap().ticket;

        let err = o.hello(&format!("127.0.0.1:{port}"), &ticket).unwrap_err();

        assert!(format!("{err:#}").contains("HTTP 403"), "{err:#}");

        // And no session, no chunks.

        let url = format!("http://127.0.0.1:{port}/swarm/v1/chunks/{}", m.chunks[0].hash());

        assert!(matches!(o.peer.get(&url).call(), Err(ureq::Error::Status(401, _))));

    }

    #[test]

    fn the_upload_rate_holds_back_a_burst() {

        let state = State {

            root: PathBuf::new(),

            site: String::new(),

            client: Client::new("http://127.0.0.1:1", "t", None).unwrap(),

            hub: Mutex::new(None),

            offers: Mutex::new(HashMap::new()),

            located: Mutex::new(HashMap::new()),

            sessions: Mutex::new(HashMap::new()),

            conns: AtomicUsize::new(0),

            bucket: Mutex::new((1_000_000.0, Instant::now())),

            rate: 1_000_000.0,

        };

        let start = Instant::now();

        throttle(&state, 1_000_000);

        assert!(start.elapsed() < Duration::from_millis(100), "the first second's worth goes at once");

        throttle(&state, 300_000);

        assert!(start.elapsed() >= Duration::from_millis(250), "then it waits for the rate");

    }

}

//! 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);

    }

}

- **Runners share images (#47, docs/jetstream.md §8.1).** A runner with a `[runner.swarm]` section serves the image chunks it holds to other runners in its site, and fetches an image it lacks from them and the hub together, rarest chunks first. The hub is asked only for chunks no peer has or is fetching, so runners starting together split its work. Peers admit each other with a ticket the hub signs, and every chunk is checked against its hash. Uploads are capped (`upload_mib_s`, default 50) at 16 connections. Without the section a runner fetches from the hub alone, as before.

    /// Where this runner serves image chunks to its peers (docs/jetstream.md

    /// §8.1). Absent: it is not in a swarm.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub swarm: Option<SwarmAddr>,

}

/// A runner's place in the image swarm: the address peers reach it on and

/// the site it is in; peers are only ever in the same site.

#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]

pub struct SwarmAddr {

    pub addr: String,

    #[serde(default)]

    pub site: String,

}

/// `POST /flow/swarm/announce`: a runner starting, continuing or ending a

/// fetch of `image`.

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]

pub struct Announce {

    pub image: String,

    /// `fetching`, `done` or `stopped`.

    pub state: String,

}

/// The hub's answer to an announce: the peers to ask, and the ticket that

/// admits this runner to them.

#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]

pub struct Announced {

    pub ticket: String,

    pub peers: Vec<SwarmPeer>,

    /// Announce again after this many seconds while still fetching.

    pub interval_s: u64,

}

/// A peer in the swarm for one image.

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]

pub struct SwarmPeer {

    pub runner: i64,

    pub name: String,

    pub addr: String,

    /// It holds the whole image (rather than fetching it now).

    pub complete: bool,

    /// Sharing images with other runners (docs/jetstream.md §8.1): absent,

    /// the runner neither serves nor asks peers.

    #[serde(default)]

    pub swarm: Option<SwarmConfig>,

}

/// `[runner.swarm]` in flow.toml, or `[swarm]` in runner.toml.

#[derive(Clone, Debug, Deserialize)]

pub struct SwarmConfig {

    /// Where peers reach this runner, e.g. `0.0.0.0:7401`.

    pub listen: String,

    /// The address peers are given (default: `listen`).

    #[serde(default)]

    pub advertise: Option<String>,

    /// Peers are only ever in the same site.

    #[serde(default)]

    pub site: String,

    /// The most this runner sends to peers, MiB a second (default 50).

    #[serde(default)]

    pub upload_mib_s: Option<u32>,