feat: implement consecutive algorithm.

This is the default negotiation algorithm.

Author: Byron

Cargo.lock

@@ -17,6 +17,8 @@ gix-hash = { version = "^0.11.1", path = "../gix-hash" }
 gix-object = { version = "^0.29.2", path = "../gix-object" }
 gix-commitgraph = { version = "^0.15.0", path = "../gix-commitgraph" }
 gix-revision = { version = "^0.14.0", path = "../gix-revision" }
+thiserror = "1.0.40"
+bitflags = "2"
 
 [dev-dependencies]
 gix-testtools = { path = "../tests/tools" }

gix-negotiate/Cargo.toml

@@ -1,2 +1,183 @@
-// TODO: make this the actual bitflags
-pub(crate) type Flags = u32;
+use crate::{Error, Negotiator};
+use gix_hash::ObjectId;
+use gix_revision::graph::CommitterTimestamp;
+bitflags::bitflags! {
+    /// Whether something can be read or written.
+    #[derive(Debug, Default, Copy, Clone)]
+    pub struct Flags: u8 {
+        /// The revision is known to be in common with the remote
+        const COMMON = 1 << 0;
+        /// The revision is common and was set by merit of a remote tracking ref (e.g. `refs/heads/origin/main`).
+        const COMMON_REF = 1 << 1;
+        /// The revision was processed by us and used to avoid processing it again.
+        const SEEN = 1 << 2;
+        /// The revision was popped off our primary priority queue, used to avoid double-counting of `non_common_revs`
+        const POPPED = 1 << 3;
+    }
+}
+
+pub(crate) struct Algorithm<'find> {
+    graph: gix_revision::Graph<'find, Flags>,
+    revs: gix_revision::PriorityQueue<CommitterTimestamp, ObjectId>,
+    non_common_revs: usize,
+}
+
+impl<'a> Algorithm<'a> {
+    pub fn new(graph: gix_revision::Graph<'a, Flags>) -> Self {
+        Self {
+            graph,
+            revs: gix_revision::PriorityQueue::new(),
+            non_common_revs: 0,
+        }
+    }
+
+    /// Add `id` to our priority queue and *add* `flags` to it.
+    fn add_to_queue(&mut self, id: ObjectId, flags: Flags) -> Result<(), Error> {
+        let mut is_common = false;
+        let commit = self.graph.try_lookup_and_insert(id, |current| {
+            *current |= flags;
+            is_common = current.contains(Flags::COMMON);
+        })?;
+        if let Some(timestamp) = commit.map(|c| c.committer_timestamp()).transpose()? {
+            self.revs.insert(timestamp, id);
+            if !is_common {
+                self.non_common_revs += 1;
+            }
+        }
+        Ok(())
+    }
+
+    fn mark_common(&mut self, id: ObjectId, mode: Mark, ancestors: Ancestors) -> Result<(), Error> {
+        let mut is_common = false;
+        if let Some(commit) = self
+            .graph
+            .try_lookup_and_insert(id, |current| is_common = current.contains(Flags::COMMON))?
+            .filter(|_| !is_common)
+        {
+            let mut queue =
+                gix_revision::PriorityQueue::from_iter(Some((commit.committer_timestamp()?, (id, 0_usize))));
+            if let Mark::ThisCommitAndAncestors = mode {
+                let current = self.graph.get_mut(&id).expect("just inserted");
+                *current |= Flags::COMMON;
+                if current.contains(Flags::SEEN) && !current.contains(Flags::POPPED) {
+                    self.non_common_revs -= 1;
+                }
+            }
+            let mut parents = Vec::with_capacity(2);
+            while let Some((id, generation)) = queue.pop() {
+                if self.graph.get(&id).map_or(true, |d| !d.contains(Flags::SEEN)) {
+                    self.add_to_queue(id, Flags::SEEN)?;
+                } else if matches!(ancestors, Ancestors::AllUnseen) || generation == 0 {
+                    if let Some(commit) = self.graph.try_lookup_and_insert(id, |_| {})? {
+                        collect_parents(commit.iter_parents(), &mut parents)?;
+                        for parent_id in parents.drain(..) {
+                            let mut prev_flags = Flags::default();
+                            if let Some(parent) = self
+                                .graph
+                                .try_lookup_and_insert(parent_id, |d| {
+                                    prev_flags = *d;
+                                    *d |= Flags::COMMON;
+                                })?
+                                .filter(|_| !prev_flags.contains(Flags::COMMON))
+                            {
+                                if prev_flags.contains(Flags::SEEN) && !prev_flags.contains(Flags::POPPED) {
+                                    self.non_common_revs -= 1;
+                                }
+                                queue.insert(parent.committer_timestamp()?, (parent_id, generation + 1))
+                            }
+                        }
+                    }
+                }
+            }
+        }
+        Ok(())
+    }
+}
+
+fn collect_parents(parents: gix_revision::graph::commit::Parents<'_>, out: &mut Vec<ObjectId>) -> Result<(), Error> {
+    out.clear();
+    for parent in parents {
+        out.push(parent.map_err(|err| match err {
+            gix_revision::graph::commit::iter_parents::Error::DecodeCommit(err) => Error::DecodeCommit(err),
+            gix_revision::graph::commit::iter_parents::Error::DecodeCommitGraph(err) => Error::DecodeCommitInGraph(err),
+        })?);
+    }
+    Ok(())
+}
+
+impl<'a> Negotiator for Algorithm<'a> {
+    fn known_common(&mut self, id: ObjectId) -> Result<(), Error> {
+        if self.graph.get(&id).map_or(true, |d| !d.contains(Flags::SEEN)) {
+            self.add_to_queue(id, Flags::COMMON_REF | Flags::SEEN)?;
+            self.mark_common(id, Mark::AncestorsOnly, Ancestors::DirectUnseen)?;
+        }
+        Ok(())
+    }
+
+    fn add_tip(&mut self, id: ObjectId) -> Result<(), Error> {
+        self.add_to_queue(id, Flags::SEEN)
+    }
+
+    fn next_have(&mut self) -> Option<Result<ObjectId, Error>> {
+        let mut parents = Vec::new();
+        loop {
+            let id = self.revs.pop().filter(|_| self.non_common_revs != 0)?;
+            let flags = self.graph.get_mut(&id).expect("it was added to the graph by now");
+            *flags |= Flags::POPPED;
+
+            if !flags.contains(Flags::COMMON) {
+                self.non_common_revs -= 1;
+            }
+
+            let (res, mark) = if flags.contains(Flags::COMMON) {
+                (None, Flags::COMMON | Flags::SEEN)
+            } else if flags.contains(Flags::COMMON_REF) {
+                (Some(id), Flags::COMMON | Flags::SEEN)
+            } else {
+                (Some(id), Flags::SEEN)
+            };
+
+            let commit = match self.graph.try_lookup(&id) {
+                Ok(c) => c.expect("it was found before, must still be there"),
+                Err(err) => return Some(Err(err.into())),
+            };
+            if let Err(err) = collect_parents(commit.iter_parents(), &mut parents) {
+                return Some(Err(err));
+            }
+            for parent_id in parents.drain(..) {
+                if self.graph.get(&parent_id).map_or(true, |d| !d.contains(Flags::SEEN)) {
+                    if let Err(err) = self.add_to_queue(parent_id, mark) {
+                        return Some(Err(err));
+                    }
+                }
+                if mark.contains(Flags::COMMON) {
+                    if let Err(err) = self.mark_common(parent_id, Mark::AncestorsOnly, Ancestors::AllUnseen) {
+                        return Some(Err(err));
+                    }
+                }
+            }
+
+            if let Some(id) = res {
+                return Some(Ok(id));
+            }
+        }
+    }
+
+    fn in_common_with_remote(&mut self, id: ObjectId) -> Result<bool, Error> {
+        let known_to_be_common = self.graph.get(&id).map_or(false, |d| d.contains(Flags::COMMON));
+        self.mark_common(id, Mark::ThisCommitAndAncestors, Ancestors::DirectUnseen)?;
+        Ok(known_to_be_common)
+    }
+}
+
+enum Mark {
+    AncestorsOnly,
+    ThisCommitAndAncestors,
+}
+
+enum Ancestors {
+    /// Traverse only the parents of a commit.
+    DirectUnseen,
+    /// Traverse all ancestors that weren't yet seen.
+    AllUnseen,
+}

gix-negotiate/src/consecutive.rs

@@ -47,7 +47,7 @@ impl Algorithm {
         self,
         find: Find,
         cache: impl Into<Option<gix_commitgraph::Graph>>,
-    ) -> Box<dyn Negotiator>
+    ) -> Box<dyn Negotiator + 'find>
     where
         Find:
             for<'a> FnMut(&gix_hash::oid, &'a mut Vec<u8>) -> Result<Option<gix_object::CommitRefIter<'a>>, E> + 'find,
@@ -56,8 +56,8 @@ impl Algorithm {
         match &self {
             Algorithm::Noop => Box::new(noop::Noop) as Box<dyn Negotiator>,
             Algorithm::Consecutive => {
-                let _graph = gix_revision::Graph::<'_, consecutive::Flags>::new(find, cache);
-                todo!()
+                let graph = gix_revision::Graph::<'_, consecutive::Flags>::new(find, cache);
+                Box::new(consecutive::Algorithm::new(graph))
             }
             Algorithm::Skipping => todo!(),
         }
@@ -69,18 +69,32 @@ pub trait Negotiator {
     /// Mark `id` as common between the remote and us.
     ///
     /// These ids are typically the local tips of remote tracking branches.
-    fn known_common(&mut self, id: &gix_hash::oid);
+    fn known_common(&mut self, id: gix_hash::ObjectId) -> Result<(), Error>;
 
     /// Add `id` as starting point of a traversal across commits that aren't necessarily common between the remote and us.
     ///
     /// These tips are usually the commits of local references whose tips should lead to objects that we have in common with the remote.
-    fn add_tip(&mut self, id: &gix_hash::oid);
+    fn add_tip(&mut self, id: gix_hash::ObjectId) -> Result<(), Error>;
 
     /// Produce the next id of an object that we want the server to know we have. It's an object we don't know we have in common or not.
     ///
     /// Returns `None` if we have exhausted all options, which might mean we have traversed the entire commit graph.
-    fn next_have(&mut self) -> Option<gix_hash::ObjectId>;
+    fn next_have(&mut self) -> Option<Result<gix_hash::ObjectId, Error>>;
 
     /// Mark `id` as being common with the remote (as informed by the remote itself) and return `true` if we knew it was common already.
-    fn in_common_with_remote(&mut self, id: &gix_hash::oid) -> bool;
+    ///
+    /// We can assume to have already seen `id` as we were the one to inform the remote in a prior `have`.
+    fn in_common_with_remote(&mut self, id: gix_hash::ObjectId) -> Result<bool, Error>;
+}
+
+/// An error that happened during any of the methods on a [`Negotiator`].
+#[derive(Debug, thiserror::Error)]
+#[allow(missing_docs)]
+pub enum Error {
+    #[error(transparent)]
+    DecodeCommit(#[from] gix_object::decode::Error),
+    #[error(transparent)]
+    DecodeCommitInGraph(#[from] gix_commitgraph::file::commit::Error),
+    #[error(transparent)]
+    LookupCommitInGraph(#[from] gix_revision::graph::lookup::Error),
 }

gix-negotiate/src/lib.rs

@@ -1,18 +1,22 @@
-use crate::Negotiator;
-use gix_hash::{oid, ObjectId};
+use crate::{Error, Negotiator};
+use gix_hash::ObjectId;
 
 pub(crate) struct Noop;
 
 impl Negotiator for Noop {
-    fn known_common(&mut self, _id: &oid) {}
+    fn known_common(&mut self, _id: ObjectId) -> Result<(), Error> {
+        Ok(())
+    }
 
-    fn add_tip(&mut self, _id: &oid) {}
+    fn add_tip(&mut self, _id: ObjectId) -> Result<(), Error> {
+        Ok(())
+    }
 
-    fn next_have(&mut self) -> Option<ObjectId> {
+    fn next_have(&mut self) -> Option<Result<ObjectId, Error>> {
         None
     }
 
-    fn in_common_with_remote(&mut self, _id: &oid) -> bool {
-        false
+    fn in_common_with_remote(&mut self, _id: ObjectId) -> Result<bool, Error> {
+        Ok(false)
     }
 }

gix-negotiate/src/noop.rs

@@ -6,12 +6,16 @@ use gix_odb::Find;
 #[test]
 fn run() -> crate::Result {
     let root = gix_testtools::scripted_fixture_read_only("make_repos.sh")?;
-    for case in ["no_parents"] {
+    for case in [
+        "no_parents",
+        "clock_skew",
+        "two_colliding_skips", /* "multi_round" */
+    ] {
         let base = root.join(case);
 
         for (algo_name, algo) in [
             ("noop", Algorithm::Noop),
-            // ("consecutive", Algorithm::Consecutive),
+            ("consecutive", Algorithm::Consecutive),
             // ("skipping", Algorithm::Skipping),
         ] {
             let buf = std::fs::read(base.join(format!("baseline.{algo_name}")))?;
@@ -31,25 +35,26 @@ fn run() -> crate::Result {
                     cache,
                 );
                 for tip in &tips {
-                    negotiator.add_tip(tip);
+                    negotiator.add_tip(*tip)?;
                 }
                 for Round { haves, common } in ParseRounds::new(buf.lines()) {
                     for have in haves {
                         let actual = negotiator.next_have().unwrap_or_else(|| {
-                            panic!(
-                                "{algo_name}: one have per baseline: {have} missing or in wrong order, left: {:?}",
-                                std::iter::from_fn(|| negotiator.next_have()).collect::<Vec<_>>()
-                            )
-                        });
-                        assert_eq!(actual, have, "{algo_name}: order and commit matches exactly");
+                            panic!("{algo_name}: one have per baseline: {have} missing or in wrong order")
+                        })?;
+                        assert_eq!(
+                            actual,
+                            have,
+                            "{algo_name}: order and commit matches exactly, left: {:?}",
+                            std::iter::from_fn(|| negotiator.next_have()).collect::<Vec<_>>()
+                        );
                     }
                     for common_revision in common {
-                        negotiator.in_common_with_remote(&common_revision);
+                        negotiator.in_common_with_remote(common_revision)?;
                     }
                 }
-                assert_eq!(
-                    negotiator.next_have(),
-                    None,
+                assert!(
+                    negotiator.next_have().is_none(),
                     "{algo_name}: negotiator should be depleted after all recorded baseline rounds"
                 );
             }