git recover: rescue an import that stopped before it finished 54e2e6309aa3

    /// Recover an import that stopped before it moved the bookmark: its

    /// unfinished pack indexed (`Store::adopt_unfinished_packs`), then every

    /// change in the store that carries a `Git-Commit:` trailer, its heads

    /// (those no other such change depends on) made the bookmark's heads,

    /// so `oxy git fetch` carries on from there. Reads every object once.

    /// Returns how many imported changes it found and the heads.

    pub fn recover_git_import(&self, progress: &dyn Fn(&str)) -> Result<(usize, usize)> {

        // A stopped import's objects are in a pack it never indexed.

        // OXY_RECOVER_MIN_AGE (seconds) overrides the five minutes it must

        // have been left alone, for testing.

        let min_age = std::env::var("OXY_RECOVER_MIN_AGE")

            .ok()

            .and_then(|v| v.parse().ok())

            .unwrap_or(300);

        self.store

            .adopt_unfinished_packs(std::time::Duration::from_secs(min_age), progress)?;

        let all = self

            .store

            .list()

            .map_err(|e| RepoError::Msg(format!("listing the store: {e}")))?;

        let total = all.len();

        let mut found: HashMap<ChangeHash, Vec<ChangeHash>> = HashMap::new();

        for (i, r) in all.iter().enumerate() {

            if i.is_multiple_of(100_000) {

                progress(&format!(

                    "{i} of {total} objects read, {} imported changes",

                    found.len()

                ));

            }

            let Ok(meta) = self.store.get_change_meta(r) else {

                continue;

            };

            if meta.message.lines().any(|l| l.starts_with("Git-Commit: ")) {

                found.insert(*r, meta.deps.clone());

            }

        }

        if found.is_empty() {

            return fail("no imported Git commit's change in this store");

        }

        let depended: HashSet<ChangeHash> = found.values().flatten().copied().collect();

        let mut heads: Vec<ChangeHash> = found

            .keys()

            .filter(|h| !depended.contains(h))

            .copied()

            .collect();

        // The bookmark's present heads stay: a recovery only adds.

        heads.extend(self.bookmark_heads().unwrap_or_default());

        let st = self.load_state_history(&heads)?;

        let mut heads = st.maximal(&heads);

        heads.sort();

        heads.dedup();

        let (op_ref, op) = self.head_op()?;

        let mut view = op.view.clone();

        let state_ref = self.store.put_state(&StateObject::new(heads.clone()))?;

        view.bookmarks

            .insert(self.config.bookmark.clone(), state_ref);

        self.record_op(&[op_ref], "oxy git recover", view)?;

        Ok((found.len(), heads.len()))

    }

    /// Finish packs a writer left unfinished: a batch writes its objects to

    /// `.tmp-ID.pack` and only indexes it at the end, so a process that

    /// stopped part way (an import killed after hours) leaves every object

    /// in a pack nothing can read. Each such pack untouched for `min_age`

    /// is read through: every whole entry checked against its reference,

    /// a torn last one cut off, then indexed as a pack of the store.

    /// Returns the packs and objects adopted.

    pub fn adopt_unfinished_packs(

        &self,

        min_age: std::time::Duration,

        progress: &dyn Fn(&str),

    ) -> Result<(usize, usize)> {

        use std::io::{BufReader, Read};

        let mut packs = 0;

        let mut objects = 0;

        let Ok(dir) = fs::read_dir(&self.scratch) else {

            return Ok((0, 0));

        };

        for entry in dir.flatten() {

            let name = entry.file_name().to_string_lossy().to_string();

            let Some(id) = name

                .strip_prefix(".tmp-")

                .and_then(|n| n.strip_suffix(".pack"))

            else {

                continue;

            };

            let path = entry.path();

            let meta = fs::metadata(&path)?;

            let age = meta

                .modified()

                .ok()

                .and_then(|m| m.elapsed().ok())

                .unwrap_or_default();

            if age < min_age {

                progress(&format!(

                    "{name}: written to {}s ago, left alone",

                    age.as_secs()

                ));

                continue;

            }

            let mut f = BufReader::with_capacity(1 << 20, fs::File::open(&path)?);

            let mut magic = [0u8; 8];

            if f.read_exact(&mut magic).is_err() || &magic != PACK_MAGIC {

                progress(&format!("{name}: not a pack, left alone"));

                continue;

            }

            let mut offset = PACK_MAGIC.len() as u64;

            let mut index: Vec<(ObjectRef, PackEntry)> = Vec::new();

            loop {

                let mut head = [0u8; 42];

                if f.read_exact(&mut head).is_err() {

                    break;

                }

                let Some(ty) = ObjectType::ALL.get(head[0] as usize).copied() else {

                    break;

                };

                let flags = head[1];

                let stored_len = u32::from_le_bytes(head[2..6].try_into().unwrap());

                let raw_len = u32::from_le_bytes(head[6..10].try_into().unwrap());

                let r = ObjectRef(head[10..42].try_into().unwrap());

                let mut stored = vec![0u8; stored_len as usize];

                if flags & FLAG_DICT != 0 || f.read_exact(&mut stored).is_err() {

                    break;

                }

                let e = PackEntry {

                    offset: offset + 42,

                    stored_len,

                    raw_len,

                    ty,

                    flags,

                };

                if decode_entry(&r, e, stored, None).is_err() {

                    break;

                }

                index.push((r, e));

                offset += 42 + stored_len as u64;

                if index.len().is_multiple_of(200_000) {

                    progress(&format!(

                        "{name}: {} objects checked, {} MiB",

                        index.len(),

                        offset >> 20

                    ));

                }

            }

            if index.is_empty() {

                progress(&format!("{name}: no whole object, left alone"));

                continue;

            }

            if offset < meta.len() {

                progress(&format!(

                    "{name}: {} torn bytes at the end cut off",

                    meta.len() - offset

                ));

                fs::OpenOptions::new()

                    .write(true)

                    .open(&path)?

                    .set_len(offset)?;

            }

            index.sort_by_key(|(r, _)| *r);

            index.dedup_by_key(|(r, _)| *r);

            objects += index.len();

            packs += 1;

            progress(&format!("{name}: {} objects indexed", index.len()));

            self.finish_pack_files(id, &path, &index)?;

        }

        Ok((packs, objects))

    }

                .get(r)

                .unwrap_or_else(|e| panic!("{}: {e}", r.short(12)));

        }

    }

    #[test]

    fn a_pack_a_stopped_batch_left_is_adopted_and_its_torn_end_cut() {

        // oxygen#79: a batch never finished (a process killed part way)

        // leaves its objects in an unindexed .tmp pack.

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

        let mut refs = Vec::new();

        {

            let store = Store::open(dir.path()).unwrap();

            let batch = store.begin_batch().unwrap();

            for i in 0..300u32 {

                let body = format!("object {i} {}", "y".repeat(20_000 + i as usize)).into_bytes();

                refs.push(store.put(ObjectType::Blob, &body).unwrap());

            }

            std::mem::forget(batch);

        }

        let tmp: Vec<_> = fs::read_dir(dir.path())

            .unwrap()

            .flatten()

            .filter(|e| e.file_name().to_string_lossy().starts_with(".tmp-"))

            .map(|e| e.path())

            .collect();

        assert_eq!(tmp.len(), 1, "one unfinished pack");

        // A torn last entry: a header and part of its body.

        let mut f = fs::OpenOptions::new().append(true).open(&tmp[0]).unwrap();

        f.write_all(&[0, 1, 0xff, 0, 0, 0, 9, 9, 0, 0]).unwrap();

        f.write_all(&[7; 40]).unwrap();

        drop(f);

        let store = Store::open(dir.path()).unwrap();

        assert!(store.get(&refs[0]).is_err(), "unreadable before");

        let young = store

            .adopt_unfinished_packs(std::time::Duration::from_secs(3600), &|_| {})

            .unwrap();

        assert_eq!(young, (0, 0), "a pack written to lately is left alone");

        let (packs, objects) = store

            .adopt_unfinished_packs(std::time::Duration::ZERO, &|_| {})

            .unwrap();

        assert_eq!((packs, objects), (1, 300));

        for r in &refs {

            store

        let again = Store::open(dir.path()).unwrap();

        assert!(again.get(&refs[299]).is_ok(), "indexed for good");

A long import moves its bookmark to the commits done so far every 2000

commits, so if it stops, `oxy git fetch GIT_DIR --branch BRANCH` in the new

repository carries on from there. One stopped before its first checkpoint

(or by an older oxy) is recovered first with `oxy git recover`, which

indexes the objects it had written and points the bookmark at them.

    /// Recover an import that stopped before it moved the bookmark: point the bookmark at every imported commit's change in the store, so oxy git fetch carries on from there

    Recover,

            cmd: GitCmd::Recover,

        } => sync::git_recover(),

        Cmd::Git {

pub fn git_recover() -> Result<i32> {

    let repo = open()?;

    let sp = ui::spinner("Reading the store…");

    let (changes, heads) = repo.recover_git_import(&|m| sp.set(m))?;

    sp.done(&format!(

        "{} {} from imported Git commits: {} now has {}",

        ui::done("Recovered"),

        plural(changes, "change"),

        ui::bookmark(&repo.config.bookmark),

        plural(heads, "head")

    ));

    println!(

        "  {} {}",

        ui::note(),

        ui::dim("carry on with oxy git fetch GIT_DIR --branch BRANCH")

    );

    Ok(0)

}