feat(land): add targets, policy and the landing queue

AGENTS.md

@@ -27,7 +27,7 @@ to a hot path (hashing, storage, snapshots, transfer) should come with a
 benchmark, and its numbers in the pull request.
 
 - `just bench`: the divan micro-benchmarks (encoding, store writes,
-  snapshots, input digests and promotion).
+  snapshots, input digests and promotion, landing).
 - `just bench-git`: times origin against git on a 20,000-file tree with
   hyperfine, and rewrites `docs/benchmarks.md`.
 

README.md

@@ -23,6 +23,7 @@ origin evolog <change>              # every revision of a change
 origin edit <change>                # switch the working copy to another change
 origin abandon --reason "Too slow"  # give a change up, keeping it for others to see
 origin converge <change>            # end a change's divergence
+origin resolve <path>...            # accept conflicted files as they are
 origin index rebuild                # rebuild the record index from the store
 
 origin annotate @ ci.test --pass --inputs src Cargo.toml   # a statement about a revision
@@ -31,15 +32,37 @@ origin annotate @ review --withdraw -m "Spoke too soon"    # take your statement
 origin annotate --on <id> flaky --neutral -m "Timeout on CI"  # a statement about a statement
 origin statements [<rev>]           # the statements about a revision, and those about them
 origin promote [<rev>]              # carry results over from revisions with the same inputs
+
+origin land [<change>] [--onto <target>]   # land a change and the ones below it
+origin land --from <target> --onto <target> # land a target's tip on another
+origin land --status                # the landings waiting for evidence or a conflict fix
+origin targets                      # every target, its tip, authority and requirements
+origin target new <name> --from <target>
+origin new [description] --on <target>     # start a change on a target's tip
 ```
 
+Landing rebases a change onto the target's tip, file by file, checks the
+statements `policy.toml` requires (promoting results whose inputs didn't
+change), and appends an acceptance signed by your node, which must be the
+target's authority. A landing that conflicts or lacks evidence waits: issue
+the statements, or a new revision, and run `origin land` again. Policy is a
+target too: edit it with `origin new --on policy`, then `origin land`.
+
 Changes are named by a unique prefix of their id, or `@` for the working
 copy's. A revision is `@`, a change (for its head), or a unique prefix of
 the revision's id. Your name and node key live in `$ORIGIN_CONFIG_DIR`, or
 else `origin` in `$XDG_CONFIG_HOME` or `~/.config`: set `user = "<name>"`
 in its `config.toml` if `$USER` isn't the name you want. See
 [the design](docs/design.md) for the model.
 
+`converge` merges the files of a change's heads. A text file two heads
+changed holds both versions between conflict markers (`<<<<<<<`, `=======`,
+`>>>>>>>`), and a binary one keeps one side. A `core.conflict` statement
+records which files conflict and what the merge wrote there, and the change
+doesn't land while they're unchanged: edit them, or accept a file as it is
+with `origin resolve <path>`. `status` lists those left, and every new
+revision, `describe` included, keeps them.
+
 ## Development
 
 ```sh

apps/origin-src/Cargo.toml

@@ -37,6 +37,10 @@ tempfile.workspace = true
 name = "annotations"
 harness = false
 
+[[bench]]
+name = "land"
+harness = false
+
 [[bench]]
 name = "snapshot"
 harness = false

apps/origin-src/benches/annotations.rs

@@ -180,12 +180,22 @@ fn promotions_cached(bencher: Bencher<'_, '_>) {
     let (_dir, store, target) = promotion_store();
     let issuer: Principal = "node:queue".parse().unwrap();
     let mut annotations = Annotations::open(&store).unwrap();
-    assert_eq!(annotations.promotions(target, &issuer).unwrap().len(), 9);
+    assert_eq!(
+        annotations
+            .promotions(RecordId::Revision(target), &issuer)
+            .unwrap()
+            .len(),
+        9
+    );
     let index = Index::open(&store).unwrap();
     bencher
         .counter(ItemsCount::new(STATEMENTS))
         .with_inputs(|| Annotations::new(&store, index.clone()))
-        .bench_local_refs(|annotations| annotations.promotions(target, &issuer).unwrap());
+        .bench_local_refs(|annotations| {
+            annotations
+                .promotions(RecordId::Revision(target), &issuer)
+                .unwrap()
+        });
 }
 
 /// The first search, which computes and remembers the digest of each of
@@ -202,5 +212,9 @@ fn promotions_uncached(bencher: Bencher<'_, '_>) {
             fs::remove_file(&log).unwrap();
             Annotations::open(&store).unwrap()
         })
-        .bench_local_refs(|annotations| annotations.promotions(target, &issuer).unwrap());
+        .bench_local_refs(|annotations| {
+            annotations
+                .promotions(RecordId::Revision(target), &issuer)
+                .unwrap()
+        });
 }

apps/origin-src/benches/land.rs

@@ -0,0 +1,235 @@
+//! The landing queue: the latency of one landing, made on the target's tip
+//! or rebased onto it, with and without a requirement, of a target landing
+//! another after long histories, and the throughput of several proposers
+//! landing different changes on one target at once. Run on a real disk:
+//! syncs are most of the cost.
+
+use std::thread;
+
+use divan::Bencher;
+use divan::counter::ItemsCount;
+use origin_platform::Native;
+use origin_shared::{
+    AcceptanceId, Base, ChangeId, EncodedObject, EntryKind, EntryName, Kind, NodeKey,
+    PrincipalName, Record, RecordId, Revision, Statement, TargetName, Tree, TreeEntry, Verdict,
+};
+use origin_src::annotations::Annotations;
+use origin_src::queue::{Landing, Queue};
+use origin_src::store::Store;
+
+fn main() {
+    divan::main();
+}
+
+/// A repository whose `policy` and `main` the node `queue` orders.
+struct Repo {
+    _dir: tempfile::TempDir,
+    store: Store,
+    key: NodeKey,
+}
+
+impl Repo {
+    /// A repository where landing on `main` requires `require`, a TOML
+    /// array.
+    fn new(require: &str) -> Self {
+        let dir = tempfile::tempdir().unwrap();
+        let store = Store::new(dir.path().join(".origin"), Native::default());
+        store.init().unwrap();
+        let repo = Self {
+            _dir: dir,
+            store,
+            key: NodeKey::from_bytes(&[7; 32]),
+        };
+        let policy = format!(
+            "[nodes]\nqueue = \"{}\"\n[groups]\nrunners = [\"node:ci\"]\n\
+             [targets.policy]\nauthority = \"node:queue\"\n\
+             [targets.feature]\nauthority = \"node:queue\"\n\
+             [targets.main]\nauthority = \"node:queue\"\nrequire = {require}\n",
+            repo.key.public()
+        );
+        repo.queue().init(&policy).unwrap();
+        repo
+    }
+
+    fn queue(&self) -> Queue<'_> {
+        Queue::new(&self.store, &self.key, PrincipalName::new("queue").unwrap())
+    }
+
+    fn tip(&self) -> AcceptanceId {
+        self.tip_of(&TargetName::MAIN)
+    }
+
+    fn tip_of(&self, target: &TargetName) -> AcceptanceId {
+        self.store.head(target).unwrap().unwrap().acceptance
+    }
+
+    /// Writes a change made on `base`, a tip of `main`, adding the file
+    /// `file-<n>`.
+    fn change(&self, n: usize, base: AcceptanceId) -> ChangeId {
+        let mut tree: Tree = self
+            .store
+            .get(self.store.get(base).unwrap().body().state)
+            .unwrap();
+        let content = EncodedObject::blob(format!("fn f{n}() {{}}\n").as_bytes());
+        tree.insert(
+            EntryName::new(format!("file-{n}")).unwrap(),
+            TreeEntry {
+                kind: EntryKind::File,
+                id: content.id(),
+            },
+        );
+        let tree = tree.encode();
+        let mut id = [0; 16];
+        id[..8].copy_from_slice(&(n as u64).to_le_bytes());
+        let change = ChangeId::from_bytes(id);
+        let revision = Revision {
+            change,
+            predecessors: Vec::new(),
+            base: Base::Acceptance(base),
+            deps: Vec::new(),
+            snapshot: tree.id().into(),
+            description: EncodedObject::blob(b"").id().into(),
+            author: "user:bench".parse().unwrap(),
+        }
+        .encode();
+        let description = EncodedObject::blob(b"");
+        self.store.write_all(&[content, tree, description]).unwrap();
+        self.store.write_all(&[revision]).unwrap();
+        change
+    }
+
+    fn land(&self, change: ChangeId) {
+        self.land_on(change, &TargetName::MAIN);
+    }
+
+    fn land_on(&self, change: ChangeId, onto: &TargetName) {
+        let landings = self.queue().land_change(change, onto).unwrap();
+        assert!(
+            matches!(landings[..], [Landing::Landed { .. }]),
+            "{landings:?}"
+        );
+    }
+}
+
+/// One landing of a change made on the tip: no rebase, no requirement.
+#[divan::bench(sample_count = 50)]
+fn land_on_tip(bencher: Bencher<'_, '_>) {
+    let repo = Repo::new("[]");
+    let mut n = 0;
+    bencher
+        .with_inputs(|| {
+            n += 1;
+            repo.change(n, repo.tip())
+        })
+        .bench_local_values(|change| repo.land(change));
+}
+
+/// One landing of a change made on an older tip, which the queue rebases
+/// into a new revision.
+#[divan::bench(sample_count = 50)]
+fn land_rebased(bencher: Bencher<'_, '_>) {
+    let repo = Repo::new("[]");
+    let first = repo.tip();
+    let mut n = 0;
+    bencher
+        .with_inputs(|| {
+            n += 1;
+            repo.change(n, first)
+        })
+        .bench_local_values(|change| repo.land(change));
+}
+
+/// One landing of a change made on the tip, with the `ci.test` result
+/// `main` requires.
+#[divan::bench(sample_count = 50)]
+fn land_with_requirement(bencher: Bencher<'_, '_>) {
+    let repo = Repo::new(r#"[{ kind = "ci.test", verdict = "pass", issuer = "group:runners" }]"#);
+    let mut n = 0;
+    bencher
+        .with_inputs(|| {
+            n += 1;
+            let change = repo.change(n, repo.tip());
+            let annotations = Annotations::open(&repo.store).unwrap();
+            let head = annotations.index().heads(change)[0];
+            let result = Statement {
+                subject: RecordId::Revision(head),
+                kind: Kind::new("ci.test").unwrap(),
+                verdict: Some(Verdict::Pass),
+                issuer: "node:ci".parse().unwrap(),
+                inputs: None,
+                payload: None,
+                supersedes: None,
+                promoted_from: None,
+            };
+            annotations.issue(&result, None).unwrap();
+            change
+        })
+        .bench_local_values(|change| repo.land(change));
+}
+
+/// How many changes each target lands before they merge.
+const HISTORY: usize = 1_000;
+
+/// Landing the tip of `feature` on `main` once each landed `HISTORY`
+/// changes since `feature` forked, after `feature` lands one more: the
+/// merge base reads both histories whole.
+#[divan::bench(sample_count = 20, sample_size = 1)]
+fn land_target_after_long_histories(bencher: Bencher<'_, '_>) {
+    let repo = Repo::new("[]");
+    let feature = TargetName::new("feature").unwrap();
+    repo.queue()
+        .create_target(&feature, &TargetName::MAIN)
+        .unwrap();
+    for n in 0..HISTORY {
+        repo.land(repo.change(n, repo.tip()));
+        let change = repo.change(HISTORY + n, repo.tip_of(&feature));
+        repo.land_on(change, &feature);
+    }
+    let mut n = 2 * HISTORY;
+    bencher
+        .with_inputs(|| {
+            n += 1;
+            let change = repo.change(n, repo.tip_of(&feature));
+            repo.land_on(change, &feature);
+        })
+        .bench_local_values(|()| {
+            let landing = repo
+                .queue()
+                .land_target(&feature, &TargetName::MAIN)
+                .unwrap();
+            assert!(matches!(landing, Landing::Landed { .. }), "{landing:?}");
+        });
+}
+
+/// How many changes each proposer lands.
+const EACH: usize = 8;
+
+/// `proposers` threads landing `EACH` changes each on `main` at once, all
+/// made on its first tip, so all but the first are rebased. Threads lock
+/// the queue as processes do: the lock belongs to the open file.
+#[divan::bench(args = [1, 2, 4, 8], sample_count = 5, sample_size = 1)]
+fn concurrent_proposers(bencher: Bencher<'_, '_>, proposers: usize) {
+    bencher
+        .counter(ItemsCount::new(proposers * EACH))
+        .with_inputs(|| {
+            let repo = Repo::new("[]");
+            let first = repo.tip();
+            let changes: Vec<_> = (0..proposers * EACH)
+                .map(|n| repo.change(n, first))
+                .collect();
+            (repo, changes)
+        })
+        .bench_local_values(|(repo, changes)| {
+            thread::scope(|scope| {
+                for changes in changes.chunks(EACH) {
+                    let repo = &repo;
+                    scope.spawn(move || {
+                        for &change in changes {
+                            repo.land(change);
+                        }
+                    });
+                }
+            });
+            (repo, changes)
+        });
+}