path: root/src/swarm/blob.rs

//! Blobs: content-addressed bytes fetched in chunks from several peers at once.
//!
//! Large transfers can't go in one message (it must arrive within the transport's
//! delivery limit, five minutes for Mycelium), and a blob every node holds identically
//! can come from many of them. A blob is an Automerge change by its hash (e.g. an
//! epoch's snapshot change, which every node on that epoch holds byte for byte), or
//! content by its BLAKE3 hash (e.g. the new document of a `replace` or `import`
//! remake), or the changes between two sets of heads (a delta: what a peer lacks, when
//! one sync message would be too large; `swarm::sync`).
//!
//! * Serving: a request without chunks gets the manifest (size, chunk size, a BLAKE3
//!   hash per chunk); one with chunks gets each chunk as a message of its own, deflated
//!   where that shrinks it (the hash is of the chunk, so any source's matches). A node
//!   that doesn't hold the blob answers `Missing`. The bytes of the blob served last
//!   are kept, so serving a download doesn't extract them for every chunk. Serving runs
//!   in tasks of its own, at most [`MAX_ANSWERING`] requests and [`MAX_SERVING`] chunk
//!   sends at a time.
//! * Fetching: the manifest from the first peer that has it (closest first, asked of
//!   [`MANIFESTS_AT_ONCE`] at a time, and only from a peer asked; of content, only with
//!   the size its reference names), then the
//!   chunks, a few in flight per peer at first, more as they come (up to
//!   [`MAX_WINDOW`]), from up to [`MAX_SOURCES`] peers. Every chunk is
//!   checked against the manifest; a peer that sends a bad one is dropped, a chunk
//!   that doesn't arrive in time (three times as long as its chunks took lately, where
//!   another peer could send it) is asked of another peer, as are all of a peer's once
//!   we no longer reach it (`swarm::reach`). The last chunks, once all are asked for,
//!   are asked of a second source too where one has room, if they are a second late
//!   ([`DOUBLE_AFTER`]). The caller checks the whole
//!   (for a change: its hash).
//!
//! The chunk size is the serving node's (`--blob-chunk`); the manifest carries it.

use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::{Arc, Mutex, MutexGuard};
use std::time::Duration;

use anyhow::{Context, Result, bail};
use tokio::sync::{Semaphore, mpsc};
use tokio::time::Instant;
use tracing::{debug, info, warn};

use crate::document::{ChangeHash, changes};
use crate::locks::MutexExt;
use crate::state::AppState;
use crate::transport::{BlobBody, BlobId, BlobMessage, DeltaSpec, ParsedMessage};

/// Chunk sends in progress at once, over all downloads served.
const MAX_SERVING: usize = 8;
/// Peers a download fetches chunks from at once.
const MAX_SOURCES: usize = 4;
/// Chunks in flight per peer at first, and after a chunk timed out (from 1).
const WINDOW: usize = 4;
/// At most this many in flight per peer: each chunk delivered allows one more, so a
/// link with a long round trip isn't left idle (4 chunks of 64 KiB per 400 ms round
/// trip are about 640 KiB/s; 32 about 5 MiB/s).
const MAX_WINDOW: usize = 32;
/// Largest blob a download of unknown size accepts (a change: an epoch's snapshot is as
/// large as the document). Content has a known size, and only that is accepted.
const MAX_SIZE: u64 = 4 << 30;
/// Smallest and largest chunk a download accepts (`--blob-chunk`'s range). The lower
/// bound keeps a manifest's chunk list, and the chunk map built from it, small.
const MIN_CHUNK: u32 = 1 << 10;
const MAX_CHUNK: u32 = 4 << 20;
/// Requests being answered at once; more are dropped (the requester asks again, or
/// another peer). Bounds the tasks anyone can make a node spawn.
const MAX_ANSWERING: usize = 32;
/// At most one warning about dropped requests per this interval: a few in a burst are
/// normal (many nodes fetching a new epoch's snapshot), a steady stream is not.
const DROPPED_LOG_INTERVAL: Duration = Duration::from_secs(60);
/// How long to wait for a manifest from one peer.
const MANIFEST_TIMEOUT: Duration = Duration::from_secs(15);
/// Peers asked for the manifest at once.
const MANIFESTS_AT_ONCE: usize = 3;
/// How long to wait for a chunk before asking another peer (raised by the peer's
/// round-trip time), until chunks from it or another source show how long they take:
/// then three times as long as they took lately (`Source::timeout_now`), but at least
/// `CHUNK_TIMEOUT_MIN`.
const CHUNK_TIMEOUT: Duration = Duration::from_secs(60);
const CHUNK_TIMEOUT_MIN: Duration = Duration::from_secs(5);
/// A download gives up after this long without any chunk arriving.
const STALL_TIMEOUT: Duration = Duration::from_secs(180);
/// At the end of a download, a chunk in flight this long is asked of a second source
/// with room too: what comes quickly isn't fetched twice.
const DOUBLE_AFTER: Duration = Duration::from_secs(1);
/// Bad chunks after which a peer is no longer asked.
const MAX_BAD: u32 = 2;

/// Per node: what is being served and fetched.
pub(crate) struct Blobs {
    chunk_size: u32,
    serving: Arc<Semaphore>,
    /// Requests being answered (`MAX_ANSWERING`).
    answering: Arc<Semaphore>,
    /// Requests dropped since the last warning about them, and when that was.
    dropped: Mutex<(u64, Option<Instant>)>,
    /// The blob served last.
    cache: Mutex<Option<(BlobId, Arc<Prepared>)>>,
    /// Answers for downloads in progress, by blob.
    fetches: Mutex<HashMap<BlobId, mpsc::UnboundedSender<(String, BlobBody)>>>,
    /// The peer whose offered delta is being downloaded (`swarm::sync`), and the peers
    /// whose offers came meanwhile (`start_delta`).
    deltas: Mutex<(Option<String>, HashSet<String>)>,
}

/// A sync message larger than this many chunks is sent as a delta blob instead
/// (`swarm::sync`): below it, a blob costs more round trips than it saves; above it, a
/// lost message costs too much to send again whole, and a slow link may not deliver it
/// within the transport's delivery limit.
const DELTA_THRESHOLD_CHUNKS: usize = 16;
/// The longest a sync message should take whole at the throughput measured to its peer
/// (`delta_threshold_for`): well within what a transport allows a message before its send
/// fails (10 s, plus its size at 1 MiB/s, on libp2p).
const WHOLE_WITHIN: Duration = Duration::from_secs(5);

/// The largest sync message accepted: the threshold of a sender with the largest chunks
/// it may have configured. Anything larger comes from a node that doesn't keep to it.
pub(crate) const MAX_SYNC_MESSAGE: usize = DELTA_THRESHOLD_CHUNKS * MAX_CHUNK as usize;

struct Prepared {
    bytes: Arc<Vec<u8>>,
    hashes: Vec<[u8; 32]>,
}

impl Blobs {
    pub(crate) fn new(chunk_size: u32) -> Self {
        Self {
            chunk_size: chunk_size.max(1),
            serving: Arc::new(Semaphore::new(MAX_SERVING)),
            answering: Arc::new(Semaphore::new(MAX_ANSWERING)),
            dropped: Mutex::new((0, None)),
            cache: Mutex::new(None),
            fetches: Mutex::new(HashMap::new()),
            deltas: Mutex::new((None, HashSet::new())),
        }
    }

    /// The largest sync message sent as it is (`DELTA_THRESHOLD_CHUNKS`).
    pub(crate) fn delta_threshold(&self) -> usize {
        DELTA_THRESHOLD_CHUNKS * self.chunk_size as usize
    }

    /// [`delta_threshold`](Self::delta_threshold) for one peer: lower where a message
    /// that large wouldn't make it whole. `rate`: the measured throughput to it, bytes
    /// per second (`swarm::throughput`), at which a message whole should take at most
    /// `WHOLE_WITHIN`; `failed`: the largest message to it that didn't make it lately
    /// (`PeerState::failed_whole`), of which messages half as large go as deltas, if it
    /// was larger than two chunks (smaller ones are lost for other reasons than their
    /// size). Never below a chunk: a delta then is no smaller per message.
    pub(crate) fn delta_threshold_for(&self, rate: Option<f64>, failed: Option<usize>) -> usize {
        let chunk = self.chunk_size as usize;
        let mut threshold = self.delta_threshold();
        if let Some(rate) = rate.filter(|rate| rate.is_finite() && *rate > 0.0) {
            threshold = threshold.min((rate * WHOLE_WITHIN.as_secs_f64()) as usize);
        }
        if let Some(failed) = failed.filter(|failed| *failed > 2 * chunk) {
            threshold = threshold.min(failed / 2);
        }
        threshold.max(chunk)
    }

    /// Starts downloading a delta offered by `peer`; false if one is downloading (from it
    /// or another), and `peer`'s offer waits for the end (`end_delta`). One at a time:
    /// while a large one comes slowly, the changes in it are what other neighbours
    /// announce and offer too (it lacks them), and they would come once per offer.
    pub(crate) fn start_delta(&self, peer: &str) -> bool {
        let mut deltas = self.deltas.lock_or_recover();
        if deltas.0.is_some() {
            deltas.1.insert(peer.to_owned());
            return false;
        }
        deltas.0 = Some(peer.to_owned());
        true
    }

    /// The delta from `peer` is in (or failed): the peers whose offers waited, to sync
    /// with now, which then send what is still missing (`peer` among them if it offered
    /// again meanwhile).
    pub(crate) fn end_delta(&self, peer: &str) -> Vec<String> {
        let mut deltas = self.deltas.lock_or_recover();
        if deltas.0.as_deref() == Some(peer) {
            deltas.0 = None;
        }
        deltas.1.drain().collect()
    }

    fn fetches(
        &self,
    ) -> MutexGuard<'_, HashMap<BlobId, mpsc::UnboundedSender<(String, BlobBody)>>> {
        self.fetches.lock_or_recover()
    }

    /// Records a dropped request. Returns how many were dropped since the last warning
    /// (this one included) if one is due now.
    fn note_dropped(&self) -> Option<u64> {
        let mut dropped = self.dropped.lock_or_recover();
        dropped.0 += 1;
        let due = dropped
            .1
            .is_none_or(|last| last.elapsed() >= DROPPED_LOG_INTERVAL);
        if !due {
            return None;
        }
        dropped.1 = Some(Instant::now());
        Some(std::mem::take(&mut dropped.0))
    }

    fn cached(&self, id: BlobId) -> Option<Arc<Prepared>> {
        self.cache
            .lock_or_recover()
            .as_ref()
            .filter(|(cached, _)| *cached == id)
            .map(|(_, prepared)| Arc::clone(prepared))
    }

    fn remember(&self, id: BlobId, prepared: Arc<Prepared>) {
        *self.cache.lock_or_recover() = Some((id, prepared));
    }
}

fn chunk_hashes(bytes: &[u8], chunk_size: u32) -> Vec<[u8; 32]> {
    bytes
        .chunks(chunk_size as usize)
        .map(|chunk| *blake3::hash(chunk).as_bytes())
        .collect()
}

fn chunk_count(size: u64, chunk_size: u32) -> u64 {
    size.div_ceil(u64::from(chunk_size))
}

/// A blob message from `peer`.
pub(crate) async fn handle(state: &AppState, peer: &str, message: BlobMessage) {
    match message.body {
        BlobBody::Request { chunks, delta } => {
            let Ok(permit) = Arc::clone(&state.blobs.answering).try_acquire_owned() else {
                state.stats.blob_request_dropped();
                debug!(peer = %peer, blob = %message.id, "Answering too many blob requests; dropping one");
                if let Some(dropped) = state.blobs.note_dropped() {
                    warn!(
                        dropped,
                        limit = MAX_ANSWERING,
                        "Dropped blob requests: answering too many at once (requesters ask again or elsewhere)"
                    );
                }
                return;
            };
            let state = state.clone();
            let peer = peer.to_owned();
            // Off the inbound loop: answering may send many large messages.
            tokio::spawn(async move {
                serve(&state, &peer, message.id, chunks, delta).await;
                drop(permit);
            });
        }
        body => {
            let sender = state.blobs.fetches().get(&message.id).cloned();
            match sender {
                Some(sender) => {
                    let _ = sender.send((peer.to_owned(), body));
                }
                None => {
                    debug!(peer = %peer, blob = %message.id, "Blob answer for no download in progress")
                }
            }
        }
    }
}

async fn prepare(state: &AppState, id: BlobId, delta: Option<DeltaSpec>) -> Option<Arc<Prepared>> {
    if let Some(prepared) = state.blobs.cached(id) {
        return Some(prepared);
    }
    let bytes = match id {
        BlobId::Delta(delta_id) => {
            // The spec must be the one the ID names.
            let spec = delta.filter(|spec| spec.id() == delta_id)?;
            let doc = Arc::clone(&state.doc);
            tokio::task::spawn_blocking(move || {
                changes::delta_bytes(&doc.blocking_read().doc, &spec.from, &spec.to)
            })
            .await
            .ok()
            .flatten()
            .map(Arc::new)?
        }
        BlobId::Change(hash) => {
            let doc = Arc::clone(&state.doc);
            tokio::task::spawn_blocking(move || {
                changes::change_bytes(&doc.blocking_read().doc, &hash)
            })
            .await
            .ok()
            .flatten()
            .map(Arc::new)?
        }
        BlobId::Content(hash) => state.remake.content(&hash)?,
    };
    let chunk_size = state.blobs.chunk_size;
    let prepared = tokio::task::spawn_blocking(move || {
        let hashes = chunk_hashes(&bytes, chunk_size);
        Prepared { bytes, hashes }
    })
    .await
    .ok()?;
    let prepared = Arc::new(prepared);
    state.blobs.remember(id, Arc::clone(&prepared));
    Some(prepared)
}

async fn reply(state: &AppState, peer: &str, id: BlobId, body: BlobBody) {
    let (epoch, epoch_seq) = state.epoch_stamp();
    let message = ParsedMessage::Blob(BlobMessage {
        epoch,
        epoch_seq,
        id,
        body,
        sent: state.clock.now_millis(),
    });
    if let Err(error) = state.transport.send_message(peer, &message).await {
        debug!(peer = %peer, blob = %id, error = %format!("{error:#}"), "Blob answer failed");
    }
}

async fn serve(
    state: &AppState,
    peer: &str,
    id: BlobId,
    chunks: Vec<u32>,
    delta: Option<DeltaSpec>,
) {
    let Some(prepared) = prepare(state, id, delta).await else {
        return reply(state, peer, id, BlobBody::Missing).await;
    };
    let chunk_size = state.blobs.chunk_size;
    if chunks.is_empty() {
        let body = BlobBody::Manifest {
            size: prepared.bytes.len() as u64,
            chunk_size,
            hashes: prepared.hashes.clone(),
        };
        return reply(state, peer, id, body).await;
    }
    for index in chunks.into_iter().take(MAX_WINDOW) {
        let start = index as usize * chunk_size as usize;
        if start >= prepared.bytes.len() {
            continue;
        }
        let end = (start + chunk_size as usize).min(prepared.bytes.len());
        let Ok(_permit) = state.blobs.serving.acquire().await else {
            return;
        };
        let chunk = &prepared.bytes[start..end];
        // Deflated where it shrinks (`compress`): changes are stored uncompressed.
        let body = match crate::compress::deflate(chunk) {
            Some(deflated) => {
                state.stats.chunk_served(chunk.len(), deflated.len());
                BlobBody::DeflatedChunk {
                    index,
                    data: deflated.into(),
                }
            }
            None => {
                state.stats.chunk_served(chunk.len(), chunk.len());
                BlobBody::Chunk {
                    index,
                    data: bytes::Bytes::copy_from_slice(chunk),
                }
            }
        };
        reply(state, peer, id, body).await;
    }
}

// --- fetching ----------------------------------------------------------------------

/// Removes the download's answer channel when it ends, however it ends.
struct Registration<'a> {
    state: &'a AppState,
    id: BlobId,
}

impl Drop for Registration<'_> {
    fn drop(&mut self) {
        self.state.blobs.fetches().remove(&self.id);
    }
}

/// Whether we no longer reach `peer_state`'s peer, as `swarm::reach` judges it.
fn unreached(peer_state: &crate::state::PeerState, now: Instant) -> bool {
    peer_state.liveness.is_dormant()
        || peer_state.is_unreachable()
        || peer_state.answer_overdue(now)
}

async fn request(
    state: &AppState,
    peer: &str,
    id: BlobId,
    chunks: Vec<u32>,
    delta: Option<&DeltaSpec>,
) {
    let (epoch, epoch_seq) = state.epoch_stamp();
    let message = ParsedMessage::Blob(BlobMessage {
        epoch,
        epoch_seq,
        id,
        body: BlobBody::Request {
            chunks,
            delta: delta.cloned(),
        },
        sent: state.clock.now_millis(),
    });
    if let Err(error) = state.transport.send_message(peer, &message).await {
        debug!(peer = %peer, blob = %id, error = %format!("{error:#}"), "Blob request failed");
    }
}

struct Manifest {
    size: u64,
    chunk_size: u32,
    hashes: Vec<[u8; 32]>,
}

/// One peer during a download.
struct Source {
    peer: String,
    in_flight: HashMap<u32, Instant>,
    window: usize,
    bad: u32,
    /// The longest a chunk is waited for (`CHUNK_TIMEOUT`, raised by the round trip).
    timeout: Duration,
    rtt: Duration,
    /// How long its chunks took lately, from the request (smoothed), once one came.
    took: Option<Duration>,
}

impl Source {
    /// How long a chunk from it is waited for now: three times as long as its chunks
    /// took lately (they wait in its queue too, as long as the window is), within
    /// `CHUNK_TIMEOUT_MIN` (and two round trips) and `timeout`. So a source that stops
    /// sending hands its chunks on within seconds where they came quickly, and a slow
    /// link isn't given up on for being slow. Before its first chunk came, by `fastest`,
    /// how long the fastest other source's took (chunks are alike in size): a source
    /// that never sends doesn't hold its window for `timeout`. Only where another peer
    /// could take them (`fetch`).
    fn timeout_now(&self, fastest: Option<Duration>) -> Duration {
        match self.took.or(fastest) {
            None => self.timeout,
            Some(took) => (took * 3).clamp(CHUNK_TIMEOUT_MIN + self.rtt * 2, self.timeout),
        }
    }

    /// A chunk asked for `took` ago came.
    fn came(&mut self, took: Duration) {
        self.took = Some(match self.took {
            None => took,
            Some(before) => (before * 7 + took) / 8,
        });
    }
}

/// Downloads blob `id` from `peers` (all thought to hold it): of `expected_size` bytes
/// if known, otherwise at most `MAX_SIZE`.
pub(crate) async fn fetch(
    state: &AppState,
    id: BlobId,
    peers: Vec<String>,
    expected_size: Option<u64>,
    delta: Option<&DeltaSpec>,
) -> Result<Vec<u8>> {
    let (sender, mut answers) = mpsc::unbounded_channel();
    if state.blobs.fetches().insert(id, sender).is_some() {
        bail!("{id} is being downloaded already");
    }
    let _registration = Registration { state, id };

    // Closest first; peers without a round-trip time last. For a delta, the first is
    // the one that offered it, which holds it: asked first however far.
    let offerer = delta.and_then(|_| peers.first().cloned());
    let now = Instant::now();
    let mut peers: Vec<(String, Option<Duration>)> = {
        let rtt = state.lock_rtt();
        peers
            .into_iter()
            .map(|peer| {
                let estimate = rtt.effective(&peer, now, state.timing.rtt_fresh_for());
                (peer, estimate)
            })
            .collect()
    };
    peers.sort_by_key(|(peer, rtt)| (offerer.as_ref() != Some(peer), rtt.unwrap_or(Duration::MAX)));
    let rtts: HashMap<String, Option<Duration>> = peers.iter().cloned().collect();
    let mut candidates: VecDeque<String> = peers.into_iter().map(|(peer, _)| peer).collect();
    if candidates.is_empty() {
        bail!("no peer to download {id} from");
    }

    // The manifest, from the first peer that has the blob: asked of the closest few at
    // once (`MANIFESTS_AT_ONCE`), and of the next as one answers it hasn't the blob or
    // doesn't answer in time, rather than of one at a time, each waited for.
    let mut manifest = None;
    let mut missing: HashSet<String> = HashSet::new();
    let mut asked: HashMap<String, Instant> = HashMap::new();
    let mut next = candidates.iter().cloned().collect::<VecDeque<_>>();
    loop {
        while asked.len() < MANIFESTS_AT_ONCE
            && let Some(peer) = next.pop_front()
        {
            request(state, &peer, id, Vec::new(), delta).await;
            asked.insert(peer, Instant::now() + MANIFEST_TIMEOUT);
        }
        let Some(deadline) = asked.values().min().copied() else {
            break;
        };
        match tokio::time::timeout_at(deadline, answers.recv()).await {
            Ok(Some((
                from,
                BlobBody::Manifest {
                    size,
                    chunk_size,
                    hashes,
                },
            ))) if candidates.contains(&from) => {
                let size_fits = match expected_size {
                    Some(expected) => size == expected,
                    None => size <= MAX_SIZE,
                };
                if !size_fits
                    || !(MIN_CHUNK..=MAX_CHUNK).contains(&chunk_size)
                    || hashes.len() as u64 != chunk_count(size, chunk_size)
                {
                    warn!(peer = %from, blob = %id, size, chunk_size, "Unusable blob manifest; asking another peer");
                    missing.insert(from.clone());
                    asked.remove(&from);
                    continue;
                }
                manifest = Some(Manifest {
                    size,
                    chunk_size,
                    hashes,
                });
                break;
            }
            Ok(Some((from, BlobBody::Missing))) => {
                missing.insert(from.clone());
                asked.remove(&from);
            }
            Ok(Some(_)) => {} // chunks of an earlier attempt
            Ok(None) => break,
            // Those asked too long ago make way for the next.
            Err(_) => {
                let now = Instant::now();
                asked.retain(|_, deadline| now < *deadline);
            }
        }
    }
    let Some(manifest) = manifest else {
        bail!("no peer sent a manifest for {id}");
    };
    candidates.retain(|peer| !missing.contains(peer));

    let total = manifest.hashes.len();
    info!(blob = %id, bytes = manifest.size, chunks = total, peers = candidates.len(), "Downloading blob");
    let started = Instant::now();
    let mut chunks: Vec<Option<bytes::Bytes>> = vec![None; total];
    let mut done = 0usize;
    let mut sources: Vec<Source> = Vec::new();
    let mut last_progress = Instant::now();
    // Chunks asked of a second source at the end.
    let mut doubled: HashSet<u32> = HashSet::new();

    while done < total {
        // Fill up the sources, then their windows.
        while sources.len() < MAX_SOURCES {
            let Some(peer) = candidates.pop_front() else {
                break;
            };
            let rtt = rtts.get(&peer).copied().flatten().unwrap_or_default();
            sources.push(Source {
                peer,
                in_flight: HashMap::new(),
                window: WINDOW,
                bad: 0,
                timeout: CHUNK_TIMEOUT + rtt * 4,
                rtt,
                took: None,
            });
        }
        if sources.is_empty() {
            bail!("no peer left to download {id} from ({done} of {total} chunks)");
        }
        let now = Instant::now();
        let assigned: HashSet<u32> = sources
            .iter()
            .flat_map(|source| source.in_flight.keys().copied())
            .collect();
        let mut wanted = (0..total as u32)
            .filter(|index| chunks[*index as usize].is_none() && !assigned.contains(index));
        for source in &mut sources {
            let free = source.window.saturating_sub(source.in_flight.len());
            let batch: Vec<u32> = wanted.by_ref().take(free).collect();
            if batch.is_empty() {
                continue;
            }
            for index in &batch {
                source.in_flight.insert(*index, now);
            }
            request(state, &source.peer, id, batch, delta).await;
        }
        // The last chunks: all are asked for, and a source has room. A chunk another one
        // has had in flight for `DOUBLE_AFTER` is asked of it too, once (`doubled`): the
        // download ends with the first of the two, not the slower.
        if wanted.next().is_none() && sources.len() > 1 {
            for at in 0..sources.len() {
                let free = sources[at]
                    .window
                    .saturating_sub(sources[at].in_flight.len());
                if free == 0 {
                    continue;
                }
                let mut others: Vec<(Instant, u32)> = sources
                    .iter()
                    .enumerate()
                    .filter(|(other, _)| *other != at)
                    .flat_map(|(_, other)| {
                        other.in_flight.iter().map(|(index, sent)| (*sent, *index))
                    })
                    .filter(|(sent, index)| {
                        now.saturating_duration_since(*sent) >= DOUBLE_AFTER
                            && !doubled.contains(index)
                            && !sources[at].in_flight.contains_key(index)
                    })
                    .collect();
                others.sort_unstable();
                let batch: Vec<u32> = others
                    .into_iter()
                    .take(free)
                    .map(|(_, index)| index)
                    .collect();
                if batch.is_empty() {
                    continue;
                }
                for index in &batch {
                    doubled.insert(*index);
                    sources[at].in_flight.insert(*index, now);
                }
                state.stats.chunks_doubled(batch.len());
                request(state, &sources[at].peer, id, batch, delta).await;
            }
        }

        let answer = tokio::time::timeout(Duration::from_millis(500), answers.recv()).await;
        // A deflated chunk inflated, to no more than a chunk; one that doesn't is bad.
        let answer = match answer {
            Ok(Some((from, BlobBody::DeflatedChunk { index, data }))) => {
                let data = crate::compress::inflate(&data, manifest.chunk_size as usize)
                    .map(bytes::Bytes::from)
                    .unwrap_or_default();
                Ok(Some((from, BlobBody::Chunk { index, data })))
            }
            other => other,
        };
        match answer {
            Ok(Some((from, BlobBody::Chunk { index, data }))) => {
                // From a source, or from a peer that was one (dropped meanwhile, as no
                // longer reached, say, while its chunks were on their way): its hash
                // tells whether it is good, not who sent it.
                let at = sources.iter().position(|source| source.peer == from);
                let sent_at = at.and_then(|at| sources[at].in_flight.remove(&index));
                let slot = index as usize;
                if slot >= total {
                    continue;
                }
                if chunks[slot].is_some() {
                    state.stats.chunk_twice();
                    continue;
                }
                let expected = expected_len(&manifest, slot);
                if data.len() != expected
                    || *blake3::hash(&data).as_bytes() != manifest.hashes[slot]
                {
                    if let Some(at) = at {
                        sources[at].bad += 1;
                    }
                    warn!(peer = %from, blob = %id, index, "Bad blob chunk");
                    continue;
                }
                if let Some(at) = at {
                    let source = &mut sources[at];
                    if let Some(sent_at) = sent_at {
                        debug!(peer = %from, blob = %id, index, ms = sent_at.elapsed().as_millis() as u64, "Blob chunk");
                        source.came(sent_at.elapsed());
                    }
                    // A peer that delivers may have one more in flight.
                    source.window = (source.window + 1).min(MAX_WINDOW);
                }
                if at.is_none() {
                    state.stats.chunk_from_dropped();
                }
                chunks[slot] = Some(data);
                done += 1;
                last_progress = Instant::now();
                // Asked of two at the end, or come from a peer no longer a source: no
                // other copy is awaited (nor counted against its source as late).
                if at.is_none() || doubled.contains(&index) {
                    for other in &mut sources {
                        other.in_flight.remove(&index);
                    }
                }
            }
            Ok(Some((from, BlobBody::Missing))) => {
                sources.retain(|source| source.peer != from);
            }
            Ok(Some(_)) => {}
            Ok(None) => bail!("download of {id} cancelled"),
            Err(_) => {}
        }

        // Chunks that took too long go to another peer; the slow peer gets a smaller
        // window. Peers that sent bad chunks are dropped.
        let now = Instant::now();
        // The quick timeout only where another peer could take the chunks: with one
        // source, asking it again would only bring them twice over a slow link.
        let others = sources.len() > 1 || !candidates.is_empty();
        let fastest = sources.iter().filter_map(|source| source.took).min();
        for source in &mut sources {
            let timeout = if others {
                source.timeout_now(fastest)
            } else {
                source.timeout
            };
            let before = source.in_flight.len();
            source
                .in_flight
                .retain(|_, sent| now.saturating_duration_since(*sent) < timeout);
            if source.in_flight.len() < before {
                state
                    .stats
                    .chunks_timed_out(before - source.in_flight.len());
                source.window = 1;
                debug!(peer = %source.peer, blob = %id, "Blob chunks timed out; asking others");
            }
        }
        sources.retain(|source| {
            let keep = source.bad < MAX_BAD;
            if !keep {
                warn!(peer = %source.peer, blob = %id, "Dropping a peer that sent bad blob chunks");
            }
            keep
        });
        // A source we no longer reach (dormant, unreachable, or what we sent it went
        // unanswered: `swarm::reach`) hands its chunks to the others at once, rather than
        // after its chunks' timeout; but only where there are others.
        if sources.len() > 1 || !candidates.is_empty() {
            let mut gone = Vec::new();
            for source in &sources {
                if let Some(peer_state) = state.get_peer_state(&source.peer).await
                    && unreached(&*peer_state.lock().await, now)
                {
                    gone.push(source.peer.clone());
                }
            }
            if !gone.is_empty() {
                debug!(blob = %id, peers = ?gone, "Blob sources no longer reached; asking others");
                sources.retain(|source| !gone.contains(&source.peer));
            }
        }
        if last_progress.elapsed() > STALL_TIMEOUT {
            bail!(
                "download of {id} stalled ({done} of {total} chunks, nothing for {}s)",
                STALL_TIMEOUT.as_secs()
            );
        }
    }

    let mut bytes = Vec::with_capacity(manifest.size as usize);
    for chunk in chunks.into_iter().flatten() {
        bytes.extend_from_slice(&chunk);
    }
    info!(
        blob = %id,
        bytes = bytes.len(),
        secs = started.elapsed().as_secs_f32(),
        "Downloaded blob"
    );
    Ok(bytes)
}

fn expected_len(manifest: &Manifest, slot: usize) -> usize {
    let chunk_size = manifest.chunk_size as u64;
    let start = slot as u64 * chunk_size;
    (manifest.size - start).min(chunk_size) as usize
}

/// Downloads change `hash` from `peers` and checks it.
/// Downloads the content with BLAKE3 hash `hash` from `peers` and checks it.
pub(crate) async fn fetch_content(
    state: &AppState,
    hash: [u8; 32],
    size: u64,
    peers: Vec<String>,
) -> Result<Vec<u8>> {
    let id = BlobId::Content(hash);
    let bytes = fetch(state, id, peers, Some(size), None).await?;
    if *blake3::hash(&bytes).as_bytes() != hash {
        bail!("downloaded {id} doesn't match its hash");
    }
    Ok(bytes)
}

pub(crate) async fn fetch_change(
    state: &AppState,
    hash: ChangeHash,
    peers: Vec<String>,
) -> Result<crate::document::Change> {
    let bytes = fetch(state, BlobId::Change(hash), peers, None, None).await?;
    changes::change_from_bytes(bytes, hash).context("downloaded")
}

/// Downloads the changes of `spec` from `peers` (any that hold its `to`). The caller
/// loads them and checks that it then has `to`.
pub(crate) async fn fetch_delta(
    state: &AppState,
    spec: &DeltaSpec,
    peers: Vec<String>,
) -> Result<Vec<u8>> {
    fetch(state, BlobId::Delta(spec.id()), peers, None, Some(spec)).await
}

#[cfg(test)]
mod tests {
    use super::*;

    /// A source's chunk timeout follows how long its chunks took, within bounds: before
    /// its first, how long the fastest other source's took; the long one until a chunk
    /// came from any.
    #[test]
    fn chunk_timeouts_follow_the_source() {
        let mut source = Source {
            peer: "p".into(),
            in_flight: HashMap::new(),
            window: WINDOW,
            bad: 0,
            timeout: CHUNK_TIMEOUT,
            rtt: Duration::from_millis(100),
            took: None,
        };
        assert_eq!(source.timeout_now(None), CHUNK_TIMEOUT);
        assert_eq!(
            source.timeout_now(Some(Duration::from_secs(2))),
            Duration::from_secs(6),
            "by the fastest other source's chunks"
        );
        source.came(Duration::from_secs(4));
        assert_eq!(source.timeout_now(None), Duration::from_secs(12));
        assert_eq!(
            source.timeout_now(Some(Duration::from_secs(1))),
            Duration::from_secs(12),
            "its own once it has some"
        );
        source.came(Duration::from_millis(100));
        assert!(source.timeout_now(None) < Duration::from_secs(12));
        for _ in 0..50 {
            source.came(Duration::from_millis(100));
        }
        assert_eq!(
            source.timeout_now(None),
            CHUNK_TIMEOUT_MIN + Duration::from_millis(200)
        );
        source.came(Duration::from_secs(600));
        assert_eq!(
            source.timeout_now(None),
            CHUNK_TIMEOUT,
            "at most the long one"
        );
    }

    /// One delta at a time: offers meanwhile wait, and are named when it ends.
    #[test]
    fn one_delta_at_a_time() {
        let blobs = Blobs::new(64 * 1024);
        assert!(blobs.start_delta("b"));
        assert!(!blobs.start_delta("c"), "c's offer waits");
        assert!(!blobs.start_delta("b"), "b's again too");
        let mut waited = blobs.end_delta("b");
        waited.sort();
        assert_eq!(waited, ["b", "c"]);
        assert!(blobs.start_delta("c"), "the next may start");
        assert!(blobs.end_delta("c").is_empty());
    }

    #[test]
    fn chunking() {
        let bytes: Vec<u8> = (0..1000u32).map(|n| n as u8).collect();
        let hashes = chunk_hashes(&bytes, 300);
        assert_eq!(hashes.len(), 4);
        assert_eq!(chunk_count(1000, 300), 4);
        assert_eq!(chunk_count(900, 300), 3);
        let manifest = Manifest {
            size: 1000,
            chunk_size: 300,
            hashes,
        };
        assert_eq!(expected_len(&manifest, 0), 300);
        assert_eq!(expected_len(&manifest, 3), 100);
        assert_eq!(manifest.hashes[3], *blake3::hash(&bytes[900..]).as_bytes());
    }
}