//! 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. 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, and only
//!   from a peer asked; of content, only with the size its reference names), then the
//!   chunks, a few in flight per peer 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 is asked of another peer. 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::{Result, anyhow, bail};
use automerge::{AutoCommit, ChangeHash};
use tokio::sync::{Semaphore, mpsc};
use tokio::time::Instant;
use tracing::{debug, info, warn};

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.
const WINDOW: usize = 4;
/// 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);
/// How long to wait for a chunk before asking another peer (raised by the peer's
/// round-trip time).
const CHUNK_TIMEOUT: Duration = Duration::from_secs(60);
/// A download gives up after this long without any chunk arriving.
const STALL_TIMEOUT: Duration = Duration::from_secs(180);
/// 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)>>>,
    /// Peers whose offered delta is being downloaded (`swarm::sync`).
    deltas: Mutex<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 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(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
    }

    /// Starts downloading a delta offered by `peer`; false if one from it is running.
    pub(crate) fn start_delta(&self, peer: &str) -> bool {
        self.deltas.lock_or_recover().insert(peer.to_owned())
    }

    pub(crate) fn end_delta(&self, peer: &str) {
        self.deltas.lock_or_recover().remove(peer);
    }

    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 || delta_bytes(&mut doc.blocking_write().doc, &spec))
                .await
                .ok()
                .flatten()
                .map(Arc::new)?
        }
        BlobId::Change(hash) => {
            let doc = Arc::clone(&state.doc);
            tokio::task::spawn_blocking(move || {
                // get_change_by_hash needs the document mutably.
                let mut replica = doc.blocking_write();
                replica
                    .doc
                    .get_change_by_hash(&hash)
                    .map(|change| change.raw_bytes().to_vec())
            })
            .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,
    });
    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(WINDOW * 2) {
        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 data = bytes::Bytes::copy_from_slice(&prepared.bytes[start..end]);
        reply(state, peer, id, BlobBody::Chunk { index, data }).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);
    }
}

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(),
        },
    });
    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,
    timeout: Duration,
}

/// 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.
    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(|(_, rtt)| 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.
    let mut manifest = None;
    let mut missing: HashSet<String> = HashSet::new();
    for peer in &candidates {
        request(state, peer, id, Vec::new(), delta).await;
        let deadline = Instant::now() + MANIFEST_TIMEOUT;
        loop {
            let answer = tokio::time::timeout_at(deadline, answers.recv()).await;
            match answer {
                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);
                        break;
                    }
                    manifest = Some(Manifest {
                        size,
                        chunk_size,
                        hashes,
                    });
                    break;
                }
                Ok(Some((from, BlobBody::Missing))) => {
                    missing.insert(from.clone());
                    if &from == peer {
                        break;
                    }
                }
                Ok(Some(_)) => {} // chunks of an earlier attempt
                Ok(None) | Err(_) => break,
            }
        }
        if manifest.is_some() {
            break;
        }
    }
    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();

    while done < total {
        // Fill up the sources, then their windows.
        while sources.len() < MAX_SOURCES {
            let Some(peer) = candidates.pop_front() else {
                break;
            };
            let timeout =
                CHUNK_TIMEOUT + rtts.get(&peer).copied().flatten().unwrap_or_default() * 4;
            sources.push(Source {
                peer,
                in_flight: HashMap::new(),
                window: WINDOW,
                bad: 0,
                timeout,
            });
        }
        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;
        }

        match tokio::time::timeout(Duration::from_millis(500), answers.recv()).await {
            Ok(Some((from, BlobBody::Chunk { index, data }))) => {
                let Some(source) = sources.iter_mut().find(|source| source.peer == from) else {
                    continue;
                };
                let sent_at = source.in_flight.remove(&index);
                let slot = index as usize;
                if slot >= total || chunks[slot].is_some() {
                    continue;
                }
                let expected = expected_len(&manifest, slot);
                if data.len() != expected
                    || *blake3::hash(&data).as_bytes() != manifest.hashes[slot]
                {
                    source.bad += 1;
                    warn!(peer = %from, blob = %id, index, "Bad blob chunk");
                    continue;
                }
                if let Some(sent_at) = sent_at {
                    debug!(peer = %from, blob = %id, index, ms = sent_at.elapsed().as_millis() as u64, "Blob chunk");
                }
                chunks[slot] = Some(data);
                done += 1;
                last_progress = Instant::now();
                // A peer that delivers gets its full window back.
                source.window = WINDOW;
            }
            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();
        for source in &mut sources {
            let timeout = 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 {
                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
        });
        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<automerge::Change> {
    let bytes = fetch(state, BlobId::Change(hash), peers, None, None).await?;
    let change = automerge::Change::from_bytes(bytes)
        .map_err(|error| anyhow!("downloaded change {hash} doesn't load: {error}"))?;
    if change.hash() != hash {
        bail!("downloaded change has hash {}, not {hash}", change.hash());
    }
    Ok(change)
}

/// 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
}

/// The bytes of delta `spec` from `doc`: the changes in the history of `spec.to` but
/// not of `spec.from`, each as Automerge encodes it, sorted by hash and concatenated
/// (`AutoCommit::load_incremental` reads that). The same on every node holding `to`,
/// whatever else it has. None if `doc` lacks part of `to`.
pub(crate) fn delta_bytes(doc: &mut AutoCommit, spec: &DeltaSpec) -> Option<Vec<u8>> {
    if spec
        .to
        .iter()
        .any(|head| doc.get_change_by_hash(head).is_none())
    {
        return None;
    }
    // Heads of `from` we don't have can't hide anything we have.
    let from: Vec<ChangeHash> = spec
        .from
        .iter()
        .copied()
        .filter(|head| doc.get_change_by_hash(head).is_some())
        .collect();
    let since: HashMap<ChangeHash, (Vec<ChangeHash>, Vec<u8>)> = doc
        .get_changes(&from)
        .iter()
        .map(|change| {
            (
                change.hash(),
                (change.deps().to_vec(), change.raw_bytes().to_vec()),
            )
        })
        .collect();
    // Of those, the ones `to` depends on: what we have beyond `to` stays out.
    let mut included: HashSet<ChangeHash> = HashSet::new();
    let mut pending: Vec<ChangeHash> = spec.to.clone();
    while let Some(hash) = pending.pop() {
        if let Some((deps, _)) = since.get(&hash)
            && included.insert(hash)
        {
            pending.extend(deps.iter().copied());
        }
    }
    let mut hashes: Vec<ChangeHash> = included.into_iter().collect();
    hashes.sort_unstable();
    let mut bytes = Vec::new();
    for hash in hashes {
        bytes.extend_from_slice(&since[&hash].1);
    }
    Some(bytes)
}

#[cfg(test)]
mod tests {
    use super::*;
    use automerge::{ActorId, ROOT, transaction::Transactable};

    fn random_write(rng: &mut crate::testrand::Rng, doc: &mut AutoCommit) {
        let key = format!("k{}", rng.below(6));
        doc.put(ROOT, key, rng.below(100) as i64).unwrap();
        doc.commit();
    }

    fn sorted(mut heads: Vec<ChangeHash>) -> Vec<ChangeHash> {
        heads.sort_unstable();
        heads
    }

    /// Over random histories: two nodes that reached the same heads through different
    /// merge orders, one with more on top, produce the same bytes for a delta to those
    /// heads; a document at `from` that loads them is at `to`.
    #[test]
    fn delta_bytes_are_the_same_everywhere_and_complete() {
        crate::testrand::for_seeds(200, |rng| {
            let actor = |i: u8| ActorId::from(&[i; 16][..]);
            let mut base = AutoCommit::new().with_actor(actor(0));
            for _ in 0..rng.between(0, 5) {
                random_write(rng, &mut base);
            }
            let from = base.get_heads();
            let mut writers: Vec<AutoCommit> = (1..=rng.between(1, 3))
                .map(|i| base.fork().with_actor(actor(i as u8)))
                .collect();
            for _ in 0..rng.between(1, 15) {
                let writer = rng.below(writers.len());
                random_write(rng, &mut writers[writer]);
            }
            let mut one = base.fork().with_actor(actor(10));
            let mut two = base.fork().with_actor(actor(11));
            for writer in writers.iter_mut() {
                one.merge(writer).unwrap();
            }
            for writer in writers.iter_mut().rev() {
                two.merge(writer).unwrap();
            }
            let to = one.get_heads();
            assert_eq!(sorted(to.clone()), sorted(two.get_heads()));
            // Beyond `to`: must stay out of the delta.
            random_write(rng, &mut two);

            let spec = DeltaSpec::new(from.clone(), to.clone());
            let bytes = delta_bytes(&mut one, &spec).unwrap();
            assert_eq!(delta_bytes(&mut two, &spec).unwrap(), bytes);

            let mut behind = base.fork();
            behind.load_incremental(&bytes).unwrap();
            assert_eq!(sorted(behind.get_heads()), sorted(to));
        });
    }

    #[test]
    fn no_delta_without_its_heads() {
        let mut doc = AutoCommit::new();
        doc.put(ROOT, "k", 1_i64).unwrap();
        doc.commit();
        let spec = DeltaSpec::new(Vec::new(), vec![ChangeHash([7; 32])]);
        assert!(delta_bytes(&mut doc, &spec).is_none());
    }

    #[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());
    }
}