Move changes as stored in a fetch on one machine 2c75672fb2fe

    /// Changes as stored, with their signatures, in dependency order, for a

    /// fetch between repositories on one machine: their text blobs and

    /// chunks are among `objects`. Never on the wire.

    pub raw_changes: Vec<(Vec<u8>, Vec<Signature>)>,

        self.repo.bundle_raw_for(changes, max_blob)

        let mut changes = Vec::with_capacity(b.raw_changes.len() + b.changes.len());

        // Changes as stored are already their encoding, their texts among

        // the objects.

        for (bytes, s) in &b.raw_changes {

            changes.push(Value::Array(vec![

                Value::Bytes(bytes.clone()),

                sigs_value(s),

            ]));

        }

            raw_changes: vec![],

        self.bundle_of(changes, max_blob, false)

    }

    /// Like [`Repo::bundle_for_filtered`] for a fetch between repositories

    /// on one machine: changes as they are stored ([`Bundle::raw_changes`]),

    /// with their text blobs and chunks among the objects, so neither side

    /// decodes or encodes their text.

    pub fn bundle_raw_for(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {

        self.bundle_of(changes, max_blob, true)

    }

    fn bundle_of(

        &self,

        changes: &[ChangeHash],

        max_blob: Option<u64>,

        raw: bool,

    ) -> Result<Bundle> {

            // Raw: the stored bytes, and a skeleton for the rest.

            let (c, encoded) = if raw {

                let bytes = self.store.get_typed(h, ObjectType::Change)?;

                let (skel, texts) = Change::decode_skeleton(&bytes, *h)?;

                for t in texts {

                    if let crate::objects::TextRef::Blob(r, _) = t {

                        let blob = Blob::decode(&self.store.get_typed(&r, ObjectType::Blob)?)?;

                        objects.insert(r);

                        objects.extend(blob.chunks);

                    }

                }

                (skel, Some(bytes))

            } else {

                (self.store.get_change(h)?, None)

            };

            }

            match encoded {

                Some(bytes) => bundle.raw_changes.push((bytes, sigs)),

                None => {

                    bundle.changes.push((c, sigs));

                    bundle.hashes.push(*h);

                }

        for (ty, _) in &bundle.objects {

            if !matches!(

                ty,

                    | ObjectType::Checkpoint

            ) {

                return fail(format!("bundle may not carry {} objects", ty.name()));

        // All at once: in a large batch they are compressed on every core.

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

            .objects

            .iter()

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

            .collect();

        self.store.put_many(&objects)?;

        let raw_hashes: Vec<ChangeHash> = bundle

            .raw_changes

            .iter()

            .map(|(b, _)| hash_object(ObjectType::Change, b))

            .collect();

        let incoming: BTreeSet<ChangeHash> = hashes.iter().chain(&raw_hashes).copied().collect();

        // Changes as stored: checked, then stored as they are, their text

        // blobs having come with the objects.

        for ((bytes, sigs), h) in bundle.raw_changes.iter().zip(raw_hashes) {

            let (skel, texts) = Change::decode_skeleton(bytes, h)?;

            for d in &skel.deps {

                if !self.store.has(d) && !incoming.contains(d) && !shallow && !boundary.contains(d)

                {

                    return fail(format!(

                        "change {} depends on {}, which is neither stored nor in the bundle",

                        h.short(12),

                        d.short(12)

                    ));

                }

            }

            for t in texts {

                if let crate::objects::TextRef::Blob(r, _) = t {

                    if !self.store.has(&r) {

                        return fail(format!(

                            "change {} came without its text blob {}",

                            h.short(12),

                            r.short(12)

                        ));

                    }

                }

            }

            self.store.put(ObjectType::Change, bytes)?;

            for sig in sigs {

                self.store.put_signature(&h, sig)?;

            }

            imported.push(h);

        }

                bundle.hashes.extend(more.hashes);

/// A fetch on one machine moves changes as stored, text blobs among the

/// objects; a change whose text blob did not come is refused, not stored.

#[test]

fn a_raw_bundle_without_a_changes_text_is_refused() {

    hermetic();

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

    let root = dir.path();

    let a = Repo::init(&root.join("a")).unwrap();

    let big: String = (0..3000).map(|i| format!("line {i}\n")).collect();

    write(&root.join("a"), "big.txt", big.as_bytes());

    let h = a.commit("big", &[]).unwrap();

    let bundle = a.bundle_raw_for(&[h], None).unwrap();

    assert_eq!(bundle.raw_changes.len(), 1);

    assert!(bundle.changes.is_empty());

    let b = Repo::init(&root.join("b")).unwrap();

    let mut without = bundle.clone();

    without.objects.retain(|(ty, _)| *ty != ObjectType::Blob);

    let e = b.import_bundle(&without, false).unwrap_err();

    assert!(e.to_string().contains("came without its text blob"), "{e}");

    assert!(!b.store().has(&h));

    assert_eq!(b.import_bundle(&bundle, false).unwrap(), vec![h]);

    assert_eq!(b.store().get_change(&h).unwrap().message, "big");

}

| Changes moved as stored (load average 5.2) | 48.2 s | 3.9 GB |

A fetch between repositories on one machine now moves changes as they

are stored, their text blobs and chunks among the objects, instead of

decoding every change's text on one side and chunking, hashing and

compressing it again on the other; the import stores all the objects at

once, compressed on every core, and refuses a change whose text blob did

not come.