A shallow clone fetches its checkpoint in parts (refs #137) 80b069facc69

- **A shallow clone fetches its checkpoint in parts (#137).** The checkpoint of a big state came as one answer, the whole state's graph blobs in one body (about 2 GB for the VS Code history), built and held whole on both sides. A server now answers `checkpoint-index` (the checkpoint and its history, no graphs), and the client fetches the graph blobs it lacks with `objects`, 2,048 at a time; `objects` reads them in pack order. An older server answers 404 and gets the one-piece request as before.

    /// A checkpoint of exactly `heads` without its graph blobs (spec §16.1

    /// `checkpoint-index`, oxygen#137), which the client then asks for in

    /// parts with `objects`. A remote that cannot answer it sends the

    /// whole checkpoint, which does as well.

    fn checkpoint_index(&self, heads: &[ChangeHash]) -> Result<Bundle> {

        self.checkpoint(heads)

    }

    }

    fn checkpoint_index(&self, heads: &[ChangeHash]) -> Result<Bundle> {

        self.repo.checkpoint_index_for(heads)

        wire::decode_bundle(&body)

    }

    fn checkpoint_index(&self, heads: &[ChangeHash]) -> Result<Bundle> {

        let body = {

            let _t = crate::timing::phase("remote: checkpoint index");

            match self.call(true, "checkpoint-index", Some(&wire::encode_hashes(heads))) {

                Ok(b) => b,

                // A server from before checkpoint-index: the whole answer.

                Err(RepoError::Msg(m)) if m.contains("HTTP 404") => return self.checkpoint(heads),

                Err(e) => return Err(e),

            }

        };

    /// The checkpoint of exactly `heads`: the bookmark's latest when it

    /// is that state, else one made now (kept as the bookmark's when the

    /// bookmark is at `heads`).

    fn checkpoint_ref_for(&self, heads: &[ChangeHash]) -> Result<ObjectRef> {

        let _t = crate::timing::phase("checkpoint: find or make");

        Ok(match self.latest_checkpoint(&bookmark)? {

        })

    }

    /// A checkpoint of exactly `heads` without its graph blobs (spec

    /// §16.1 `checkpoint-index`, oxygen#137): the checkpoint object, its

    /// payload blob and the blob's chunks. The client asks for the graph

    /// blobs the payload names in parts, with `objects`, so neither side

    /// holds a big state's whole answer at once.

    pub fn checkpoint_index_for(&self, heads: &[ChangeHash]) -> Result<Bundle> {

        let r = self.checkpoint_ref_for(heads)?;

        let chunks = Blob::decode(&self.store.get_typed(&cp.payload, ObjectType::Blob)?)?.chunks;

        for got in self.store.get_many(&chunks) {

            bundle.objects.push(got?);

        bundle.objects.push((ObjectType::Checkpoint, cp.encode()));

        Ok(bundle)

    }

    /// A checkpoint of exactly `heads` with its graph blobs and their

    /// chunks, in one answer (spec §16.1 `checkpoint`): what a client

    /// older than `checkpoint-index` asks for.

    pub fn checkpoint_bundle_for(&self, heads: &[ChangeHash]) -> Result<Bundle> {

        let mut bundle = self.checkpoint_index_for(heads)?;

        let _t = crate::timing::phase("checkpoint: gather its objects");

        let cp = match bundle.objects.last() {

            Some((ObjectType::Checkpoint, body)) => Checkpoint::decode(body)?,

            _ => return fail("a checkpoint index without its checkpoint"),

        };

        let last = bundle.objects.pop();

        bundle.objects.extend(last);

    fn checkpoint_index(&self, heads: &[ChangeHash]) -> Result<Bundle> {

        self.with_mirror(|m| m.checkpoint_index_for(heads))

    }

/// Graph blobs asked for in one `objects` request when a shallow clone

/// fetches its checkpoint in parts (#137): their chunk lists and chunks,

/// a few MiB a request rather than one answer of gigabytes.

const CHECKPOINT_PART: usize = 2048;

        // In two reads in pack order (#137): a checkpoint's graph blobs come

        // this way in parts, tens of thousands for a big state.

        let mut found = Vec::with_capacity(refs.len());

        for got in self.store.get_many(refs) {

            match got {

                Ok(x) => found.push(x),

                Err(StoreError::NotFound(_)) => {}

            }

        }

        let mut chunks = Vec::new();

        for (ty, body) in &found {

            if *ty == ObjectType::Blob {

                chunks.extend(Blob::decode(body)?.chunks);

            }

        }

        let mut out = Vec::with_capacity(found.len() + chunks.len());

        for got in self.store.get_many(&chunks) {

            match got {

                Ok(x) => out.push(x),

                Err(StoreError::NotFound(_)) => {}

                Err(e) => return Err(e.into()),

        out.extend(found);

        // Base: the remote's checkpoint of exactly these heads, without its

        // graph blobs, which come in parts below (#137).

            remote.checkpoint_index(&r.heads)?

        // The graph blobs the payload names, with their chunks, a part at

        // a time (#137): one answer for a big state was gigabytes, held

        // whole on both sides. A remote that sent the whole checkpoint has

        // left none to ask for.

        let cp = crate::objects::Checkpoint::decode(

            &self.store.get_typed(&cp_ref, ObjectType::Checkpoint)?,

        )?;

        let graphs: Vec<ObjectRef> = self

            .checkpoint_graph_blobs(&cp)?

            .into_iter()

            .filter(|b| !self.store.has(b))

            .collect();

        {

            let _t = crate::timing::phase("shallow: the checkpoint's graphs, in parts");

            for part in graphs.chunks(CHECKPOINT_PART) {

                let objects = remote.objects(part)?;

                let objects: Vec<(ObjectType, &[u8])> = objects

                    .iter()

                    .filter(|(ty, _)| matches!(ty, ObjectType::Blob | ObjectType::Chunk))

                    .map(|(ty, body)| (*ty, body.as_slice()))

                    .collect();

                let batch = self.store.begin_batch()?;

                let stored = self.store.put_many(&objects);

                batch.finish()?;

                stored?;

            }

            if let Some(b) = graphs.iter().find(|b| !self.store.has(b)) {

                return fail(format!(

                    "{remote_name} did not send graph blob {}",

                    b.short(12)

                ));

            }

        }

    CheckpointIndex,

            "checkpoint-index" => Endpoint::CheckpointIndex,

            Endpoint::CheckpointIndex => "checkpoint-index",

            Endpoint::CheckpointIndex => {

                let heads = wire::decode_hashes(&body)?;

                let lock = self.lock_for(name);

                let _guard = lock.lock().unwrap();

                ok(wire::encode_bundle(&r.checkpoint_index_for(&heads)?))

            }

| `checkpoint` request | `{0: [heads]}`; response a `Bundle` of objects: the checkpoint of exactly those heads, its payload blob, and the blob's chunks, with the payload's graph blobs and their chunks (§11) |

| `checkpoint-index` request | as `checkpoint`, without the graph blobs: the client asks for those it lacks with `objects`, in parts (oxygen#137), so neither side holds a big state's whole answer; a server without it answers 404 and the client falls back to `checkpoint` |