radicle-job

The genesis of the `Job` collaborative object

.envrc

@@ -0,0 +1 @@
+use flake

.gitignore

@@ -0,0 +1 @@
+/target

Cargo.toml

@@ -0,0 +1,27 @@
+[package]
+name = "radicle-job"
+description = "Radicle Job Collaborative Object"
+license = "MIT OR Apache-2.0"
+version = "0.1.0"
+authors = ["Fintan Halpenny <fintan.halpenny@gmail.com>"]
+edition = "2021"
+
+[dependencies]
+indexmap = { version = "2.7.1", features = ["serde"] }
+nonempty = "0.11.0"
+once_cell = "1.20.3"
+qcheck = "1.0.0"
+serde = { version = "1.0", features = ["derive"] }
+thiserror = "2.0.11"
+url = { version = "2.5.4", features = ["serde"] }
+uuid = { version = "1.13.1", features = ["serde", "v4"] }
+
+# TODO: we depend on a commit that improves the `Transaction::initial` method
+[dependencies.radicle]
+git = "https://seed.radicle.xyz/z3gqcJUoA1n9HaHKufZs5FCSGazv5.git"
+rev= "refs/namespaces/z6MkireRatUThvd3qzfKht1S44wpm4FEWSSa4PRMTSQZ3voM/refs/heads/cob-transaction"
+
+[dev-dependencies.radicle]
+git = "https://seed.radicle.xyz/z3gqcJUoA1n9HaHKufZs5FCSGazv5.git"
+rev= "refs/namespaces/z6MkireRatUThvd3qzfKht1S44wpm4FEWSSa4PRMTSQZ3voM/refs/heads/cob-transaction"
+features = ["test"]

flake.lock

@@ -0,0 +1,98 @@
+{
+  "nodes": {
+    "crane": {
+      "locked": {
+        "lastModified": 1742394900,
+        "narHash": "sha256-vVOAp9ahvnU+fQoKd4SEXB2JG2wbENkpqcwlkIXgUC0=",
+        "owner": "ipetkov",
+        "repo": "crane",
+        "rev": "70947c1908108c0c551ddfd73d4f750ff2ea67cd",
+        "type": "github"
+      },
+      "original": {
+        "owner": "ipetkov",
+        "repo": "crane",
+        "type": "github"
+      }
+    },
+    "flake-utils": {
+      "inputs": {
+        "systems": "systems"
+      },
+      "locked": {
+        "lastModified": 1731533236,
+        "narHash": "sha256-l0KFg5HjrsfsO/JpG+r7fRrqm12kzFHyUHqHCVpMMbI=",
+        "owner": "numtide",
+        "repo": "flake-utils",
+        "rev": "11707dc2f618dd54ca8739b309ec4fc024de578b",
+        "type": "github"
+      },
+      "original": {
+        "owner": "numtide",
+        "repo": "flake-utils",
+        "type": "github"
+      }
+    },
+    "nixpkgs": {
+      "locked": {
+        "lastModified": 1742998488,
+        "narHash": "sha256-r7dqSLNkzg1rOPsq4Tuc6p+UtAGqNRMma4riupsuZv4=",
+        "owner": "NixOS",
+        "repo": "nixpkgs",
+        "rev": "8e0b1e9c8c4dfaef0ff85e09c91294ba4afa4168",
+        "type": "github"
+      },
+      "original": {
+        "owner": "NixOS",
+        "ref": "release-24.11",
+        "repo": "nixpkgs",
+        "type": "github"
+      }
+    },
+    "root": {
+      "inputs": {
+        "crane": "crane",
+        "flake-utils": "flake-utils",
+        "nixpkgs": "nixpkgs",
+        "rust-overlay": "rust-overlay"
+      }
+    },
+    "rust-overlay": {
+      "inputs": {
+        "nixpkgs": [
+          "nixpkgs"
+        ]
+      },
+      "locked": {
+        "lastModified": 1742956365,
+        "narHash": "sha256-Slrqmt6kJ/M7Z/ce4ebQWsz2aeEodrX56CsupOEPoz0=",
+        "owner": "oxalica",
+        "repo": "rust-overlay",
+        "rev": "a0e3395c63cdbc9c1ec17915f8328c077c79c4a1",
+        "type": "github"
+      },
+      "original": {
+        "owner": "oxalica",
+        "repo": "rust-overlay",
+        "type": "github"
+      }
+    },
+    "systems": {
+      "locked": {
+        "lastModified": 1681028828,
+        "narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
+        "owner": "nix-systems",
+        "repo": "default",
+        "rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
+        "type": "github"
+      },
+      "original": {
+        "owner": "nix-systems",
+        "repo": "default",
+        "type": "github"
+      }
+    }
+  },
+  "root": "root",
+  "version": 7
+}

flake.nix

@@ -0,0 +1,108 @@
+{
+  description = "Radicle Job Collaborative Object";
+
+  inputs = {
+    nixpkgs.url = "github:NixOS/nixpkgs/release-24.11";
+
+    crane.url = "github:ipetkov/crane";
+
+    flake-utils.url = "github:numtide/flake-utils";
+
+    rust-overlay = {
+      url = "github:oxalica/rust-overlay";
+      inputs.nixpkgs.follows = "nixpkgs";
+    };
+  };
+
+  outputs = {
+    self,
+    nixpkgs,
+    crane,
+    flake-utils,
+    rust-overlay,
+    ...
+  }:
+    flake-utils.lib.eachDefaultSystem (system: let
+      lib = nixpkgs.lib;
+      pkgs = import nixpkgs {
+        inherit system;
+        overlays = [(import rust-overlay)];
+      };
+
+      rustToolChain = pkgs.rust-bin.fromRustupToolchainFile ./rust-toolchain.toml;
+      craneLib = (crane.mkLib pkgs).overrideToolchain rustToolChain;
+
+      src = lib.cleanSourceWith {
+        src = ./.;
+      };
+
+      basicArgs = {
+        inherit src;
+        pname = "radicle-job";
+        strictDeps = true;
+      };
+
+      # Build *just* the cargo dependencies, so we can reuse
+      # all of that work (e.g. via cachix) when running in CI
+      cargoArtifacts = craneLib.buildDepsOnly basicArgs;
+
+      # Common arguments can be set here to avoid repeating them later
+      commonArgs =
+        basicArgs
+        // {
+          inherit cargoArtifacts;
+
+          nativeBuildInputs = with pkgs; [
+            # Add additional build inputs here
+          ];
+          buildInputs = lib.optionals pkgs.stdenv.buildPlatform.isDarwin (with pkgs; [
+            darwin.apple_sdk.frameworks.Security
+            pkgs.libiconv
+          ]);
+        };
+    in {
+      formatter = pkgs.alejandra;
+
+      checks = {
+        inherit (self.packages.${system}) radicle-job;
+
+        clippy = craneLib.cargoClippy (commonArgs
+          // {
+            cargoClippyExtraArgs = "--all-targets -- --deny warnings";
+          });
+
+        doc = craneLib.cargoDoc commonArgs;
+        fmt = craneLib.cargoFmt basicArgs;
+        # Run tests with cargo-nextest
+        nextest = craneLib.cargoNextest (commonArgs
+          // {
+            partitions = 1;
+            partitionType = "count";
+            # Ensure dev is used since we rely on env variables being
+            # set in tests.
+            env.CARGO_PROFILE = "dev";
+          });
+      };
+
+      packages.radicle-job = craneLib.buildPackage (commonArgs
+        // {
+          src = craneLib.cleanCargoSource ./.;
+          strictDeps = true;
+          doCheck = false;
+        });
+      packages.default = self.packages.${system}.radicle-job;
+
+      devShells.default = craneLib.devShell {
+        # Inherit inputs from checks.
+        checks = self.checks.${system};
+
+        # Extra inputs can be added here; cargo and rustc are provided by default
+        # from the toolchain that was specified earlier.
+        packages = with pkgs; [
+          cargo-watch
+          ripgrep
+          rust-analyzer
+        ];
+      };
+    });
+}

rust-toolchain.toml

@@ -0,0 +1,4 @@
+[toolchain]
+channel = "1.84"
+profile = "default"
+components = [ "rust-src" ]

src/error.rs

@@ -0,0 +1,22 @@
+use radicle::{cob, git};
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum Build {
+    #[error("initial action of job must request an OID")]
+    Initial,
+    #[error("missing commit for job run {oid}: {err}")]
+    MissingCommit {
+        oid: git::Oid,
+        #[source]
+        err: git::Error,
+    },
+}
+
+#[derive(Debug, Error)]
+pub enum Apply {
+    #[error(transparent)]
+    Build(#[from] Build),
+    #[error(transparent)]
+    Op(#[from] cob::op::OpEncodingError),
+}

src/lib.rs

@@ -0,0 +1,807 @@
+use std::collections::HashMap;
+use std::ops::{Deref, DerefMut};
+use std::str::FromStr;
+
+use indexmap::IndexMap;
+use once_cell::sync::Lazy;
+use radicle::cob::store::Cob;
+use radicle::cob::{self, store, EntryId, Evaluate, ObjectId, Op, TypeName};
+use radicle::crypto::Signer;
+use radicle::node::NodeId;
+use radicle::prelude::ReadRepository;
+use radicle::storage::{RepositoryError, SignRepository, WriteRepository};
+use radicle::{cob::store::CobAction, git::Oid};
+use serde::{Deserialize, Serialize};
+use url::Url;
+use uuid::Uuid;
+
+pub mod error;
+
+/// Type name of a patch.
+pub static TYPENAME: Lazy<TypeName> =
+    Lazy::new(|| FromStr::from_str("xyz.radworks.job").expect("type name is valid"));
+
+/// A `Job` describes a generic task run for a given commit [`Oid`] by a set of
+/// nodes.
+///
+/// A node may run the task many times, which are accumulated in its [`Runs`]
+/// set. Each [`Run`] is identified by a [`Uuid`].
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Job {
+    oid: Oid,
+    runs: HashMap<NodeId, Runs>,
+}
+
+/// A set of [`Run`]s identified by a [`Uuid`].
+///
+/// Iteration over this set is guaranteed to come in insertion order.
+#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Runs(IndexMap<Uuid, Run>);
+
+impl Runs {
+    /// Insert a new [`Run`], identified by the given [`Uuid`].
+    pub fn insert(&mut self, uuid: Uuid, run: Run) -> Option<Run> {
+        self.0.insert(uuid, run)
+    }
+
+    /// Check that the set of runs contains the given [`Uuid`].
+    pub fn contains_key(&self, uuid: &Uuid) -> bool {
+        self.0.contains_key(uuid)
+    }
+
+    /// Get the [`Run`] identifier by the given [`Uuid`].
+    ///
+    /// Return `None` if the `Uuid` does not exist.`
+    pub fn get(&self, uuid: &Uuid) -> Option<&Run> {
+        self.0.get(uuid)
+    }
+
+    /// Get the `nth` [`Run`] of the set.
+    pub fn get_index(&self, nth: usize) -> Option<(&Uuid, &Run)> {
+        self.0.get_index(nth)
+    }
+
+    /// Get the latest [`Run`] and its corresponding [`Uuid`].
+    pub fn latest(&self) -> Option<(&Uuid, &Run)> {
+        self.0.iter().next_back()
+    }
+
+    /// Get all [`Run`]s that have started [`Status`].
+    pub fn started(&self) -> Runs {
+        self.iter()
+            .filter_map(|(uuid, run)| run.is_started().then_some((*uuid, run.clone())))
+            .collect()
+    }
+
+    /// Get all [`Run`]s that have finished [`Status`].
+    pub fn finished(&self) -> Runs {
+        self.iter()
+            .filter_map(|(uuid, run)| run.is_finished().then_some((*uuid, run.clone())))
+            .collect()
+    }
+
+    /// Get all [`Run`]s that have finished [`Status`] and have the succeeded
+    /// [`Reason`].
+    pub fn succeeded(&self) -> Runs {
+        self.iter()
+            .filter_map(|(uuid, run)| run.succeeded().then_some((*uuid, run.clone())))
+            .collect()
+    }
+
+    /// Get all [`Run`]s that have finished [`Status`] and have the failed
+    /// [`Reason`].
+    pub fn failed(&self) -> Runs {
+        self.iter()
+            .filter_map(|(uuid, run)| run.failed().then_some((*uuid, run.clone())))
+            .collect()
+    }
+
+    /// Partition all [`Run`]s into started, succeeded, and failed.
+    pub fn partition(&self) -> (Runs, Runs, Runs) {
+        let mut started = IndexMap::new();
+        let mut succeeded = IndexMap::new();
+        let mut failed = IndexMap::new();
+
+        for (uuid, run) in self.0.iter() {
+            match run.status {
+                Status::Started => started.insert(*uuid, run.clone()),
+                Status::Finished(Reason::Succeeded) => succeeded.insert(*uuid, run.clone()),
+                Status::Finished(Reason::Failed) => failed.insert(*uuid, run.clone()),
+            };
+        }
+        (Runs(started), Runs(succeeded), Runs(failed))
+    }
+
+    /// Check is the set of [`Runs`] empty.
+    pub fn is_empty(&self) -> bool {
+        self.0.is_empty()
+    }
+
+    /// Get the number of [`Runs`] in the set.
+    pub fn len(&self) -> usize {
+        self.0.len()
+    }
+
+    /// Iterate over the [`Run`]s and their corresponding [`Uuid`]s.
+    ///
+    /// The order of the iteration is guaranteed to be insertion order.
+    pub fn iter(&self) -> impl Iterator<Item = (&Uuid, &Run)> {
+        self.0.iter()
+    }
+}
+
+impl FromIterator<(Uuid, Run)> for Runs {
+    fn from_iter<T: IntoIterator<Item = (Uuid, Run)>>(iter: T) -> Self {
+        Self(iter.into_iter().collect())
+    }
+}
+
+impl<'a> IntoIterator for &'a Runs {
+    type Item = (&'a Uuid, &'a Run);
+    type IntoIter = indexmap::map::Iter<'a, Uuid, Run>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        self.0.iter()
+    }
+}
+
+impl IntoIterator for Runs {
+    type Item = (Uuid, Run);
+    type IntoIter = indexmap::map::IntoIter<Uuid, Run>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        self.0.into_iter()
+    }
+}
+
+/// The collaborative object actions that are used for Radicle COB operations.
+#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
+pub enum Action {
+    /// Request a [`Job`] be created for the given [`Oid`].
+    ///
+    /// Note this is expected only once for initializing the [`Job`]. Every
+    /// other `Request` for the same `Oid` will be ignored.
+    Request {
+        /// The commit the [`Job`] corresponds to.
+        oid: Oid,
+    },
+    /// Notify that the node has started a [`Run`].
+    Run {
+        /// The [`Uuid`] that identifies this particular [`Run`] of the node.
+        uuid: Uuid,
+        /// The [`Url`] where the node will log any information or data.
+        log: Url,
+    },
+    /// Notify that the node has finished a [`Run`].
+    Finished {
+        /// The [`Uuid`] that identifies the [`Run`] that is finished.
+        uuid: Uuid,
+        /// The [`Reason`] which the node finished with.
+        reason: Reason,
+    },
+}
+
+impl CobAction for Action {
+    fn parents(&self) -> Vec<radicle::git::Oid> {
+        match self {
+            Action::Request { oid } => vec![*oid],
+            _ => Vec::new(),
+        }
+    }
+}
+
+impl Job {
+    /// Construct a new [`Job`].
+    fn new(oid: Oid) -> Self {
+        Self {
+            oid,
+            runs: HashMap::new(),
+        }
+    }
+
+    /// Get the [`Oid`] this [`Job`] is running tasks for.
+    pub fn oid(&self) -> &Oid {
+        &self.oid
+    }
+
+    /// Get all the nodes that have started, but not finished, [`Runs`].
+    pub fn started(&self) -> HashMap<NodeId, Runs> {
+        self.filter_map_by(|runs| runs.started())
+    }
+
+    /// Get all the nodes that have started, and finished, [`Runs`].
+    pub fn finished(&self) -> HashMap<NodeId, Runs> {
+        self.filter_map_by(|runs| runs.finished())
+    }
+
+    /// Get all the nodes that have succeeded [`Runs`].
+    pub fn succeeded(&self) -> HashMap<NodeId, Runs> {
+        self.filter_map_by(|runs| runs.succeeded())
+    }
+
+    /// Get all the nodes that have failed [`Runs`].
+    pub fn failed(&self) -> HashMap<NodeId, Runs> {
+        self.filter_map_by(|runs| runs.failed())
+    }
+
+    /// Get all nodes' started, succeeded, and failed runs – respectively.
+    pub fn partition(&self) -> HashMap<NodeId, (Runs, Runs, Runs)> {
+        self.runs
+            .iter()
+            .map(|(node, runs)| (*node, runs.partition()))
+            .collect()
+    }
+
+    /// Get the latest [`Run`] of the given [`NodeId`].
+    pub fn latest_of(&self, node: &NodeId) -> Option<(&Uuid, &Run)> {
+        self.runs
+            .get(node)
+            .and_then(|runs| runs.0.iter().next_back())
+    }
+
+    /// Get the latest [`Run`]s of all [`NodeId`]s.
+    pub fn latest(&self) -> impl Iterator<Item = (&NodeId, &Uuid, &Run)> + '_ {
+        self.runs
+            .iter()
+            .filter_map(|(node, runs)| runs.latest().map(|(uuid, run)| (node, uuid, run)))
+    }
+
+    /// Get the raw `HashMap` of the node runs.
+    pub fn runs(&self) -> &HashMap<NodeId, Runs> {
+        &self.runs
+    }
+
+    /// Get the [`Runs`] of a given [`NodeId`].
+    pub fn runs_of(&self, node: &NodeId) -> Option<&Runs> {
+        self.runs.get(node)
+    }
+
+    fn filter_map_by<P>(&self, p: P) -> HashMap<NodeId, Runs>
+    where
+        P: Fn(&Runs) -> Runs,
+    {
+        self.runs
+            .iter()
+            .filter_map(|(node, runs)| {
+                let runs = p(runs);
+                (!runs.is_empty()).then_some((*node, runs))
+            })
+            .collect()
+    }
+
+    fn insert(&mut self, node: NodeId, uuid: Uuid, run: Run) -> bool {
+        let runs = self.runs.entry(node).or_default();
+        if runs.contains_key(&uuid) {
+            false
+        } else {
+            runs.insert(uuid, run);
+            true
+        }
+    }
+
+    fn update(&mut self, node: NodeId, uuid: Uuid, reason: Reason) -> bool {
+        let Some(runs) = self.runs.get_mut(&node) else {
+            return false;
+        };
+        let mut updated = false;
+        runs.0.entry(uuid).and_modify(|run| {
+            updated = true;
+            *run = run.clone().finish(reason);
+        });
+        updated
+    }
+
+    fn action(&mut self, node: NodeId, action: Action) -> Result<(), error::Build> {
+        match action {
+            // Cannot request for another `oid`, so we ignore any superfluous
+            // request actions
+            Action::Request { .. } => Ok(()),
+            Action::Run { uuid, log } => {
+                self.insert(node, uuid, Run::new(log));
+                Ok(())
+            }
+            Action::Finished { uuid, reason } => {
+                self.update(node, uuid, reason);
+                Ok(())
+            }
+        }
+    }
+}
+
+impl store::Cob for Job {
+    type Action = Action;
+    type Error = error::Build;
+
+    fn type_name() -> &'static TypeName {
+        &TYPENAME
+    }
+
+    fn from_root<R: ReadRepository>(op: Op<Self::Action>, repo: &R) -> Result<Self, Self::Error> {
+        let mut actions = op.actions.into_iter();
+        let Some(Action::Request { oid }) = actions.next() else {
+            return Err(error::Build::Initial);
+        };
+        repo.commit(oid)
+            .map_err(|err| error::Build::MissingCommit { oid, err })?;
+        let mut runs = Self::new(oid);
+        for action in actions {
+            runs.action(op.author, action)?;
+        }
+        Ok(runs)
+    }
+
+    fn op<'a, R: ReadRepository, I: IntoIterator<Item = &'a radicle::cob::Entry>>(
+        &mut self,
+        op: Op<Self::Action>,
+        _concurrent: I,
+        _repo: &R,
+    ) -> Result<(), Self::Error> {
+        for action in op.actions {
+            self.action(op.author, action)?;
+        }
+        Ok(())
+    }
+}
+
+impl<R: ReadRepository> Evaluate<R> for Job {
+    type Error = error::Apply;
+
+    fn init(entry: &radicle::cob::Entry, store: &R) -> Result<Self, Self::Error> {
+        let op = Op::try_from(entry)?;
+        let object = Job::from_root(op, store)?;
+        Ok(object)
+    }
+
+    fn apply<'a, I: Iterator<Item = (&'a Oid, &'a radicle::cob::Entry)>>(
+        &mut self,
+        entry: &radicle::cob::Entry,
+        concurrent: I,
+        store: &R,
+    ) -> Result<(), Self::Error> {
+        let op = Op::try_from(entry)?;
+        self.op(op, concurrent.map(|(_, e)| e), store)
+            .map_err(error::Apply::from)
+    }
+}
+
+/// A `Run` represents a task run for a [`Job`].
+#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
+pub struct Run {
+    /// The status of the run.
+    ///
+    /// A run can be in one of three states:
+    ///   - Started
+    ///   - Succeeded – implies that it has finished
+    ///   - Failed – implies that it has finished
+    status: Status,
+    /// The [`Url`] of the [`Run`] where information is logged by the node.
+    log: Url,
+}
+
+impl Run {
+    fn new(log: Url) -> Self {
+        Self {
+            status: Status::Started,
+            log,
+        }
+    }
+
+    fn finish(self, reason: Reason) -> Self {
+        Self {
+            status: Status::Finished(reason),
+            log: self.log,
+        }
+    }
+
+    pub fn status(&self) -> &Status {
+        &self.status
+    }
+
+    pub fn is_started(&self) -> bool {
+        match self.status {
+            Status::Started => true,
+            Status::Finished(_) => false,
+        }
+    }
+
+    pub fn is_finished(&self) -> bool {
+        !self.is_started()
+    }
+
+    pub fn succeeded(&self) -> bool {
+        match self.status {
+            Status::Started => false,
+            Status::Finished(Reason::Failed) => false,
+            Status::Finished(Reason::Succeeded) => true,
+        }
+    }
+
+    pub fn failed(&self) -> bool {
+        match self.status {
+            Status::Started => false,
+            Status::Finished(Reason::Failed) => true,
+            Status::Finished(Reason::Succeeded) => false,
+        }
+    }
+}
+
+/// The status of a [`Run`].
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
+pub enum Status {
+    /// The [`Run`] has started.
+    Started,
+    /// The [`Run`] has finished with the given [`Reason`].
+    Finished(Reason),
+}
+
+/// The reason for a [`Status`] to have finished.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
+pub enum Reason {
+    Failed,
+    Succeeded,
+}
+
+pub struct Jobs<'a, R> {
+    raw: store::Store<'a, Job, R>,
+}
+
+impl<'a, R> Deref for Jobs<'a, R> {
+    type Target = store::Store<'a, Job, R>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.raw
+    }
+}
+
+impl<'a, R> Jobs<'a, R>
+where
+    R: ReadRepository + cob::Store,
+{
+    /// Open a jobs store.
+    pub fn open(repository: &'a R) -> Result<Self, RepositoryError> {
+        let identity = repository.identity_head()?;
+        let raw = store::Store::open(repository)?.identity(identity);
+
+        Ok(Self { raw })
+    }
+
+    /// Return the number of [`Job`]s in the store.
+    pub fn counts(&self) -> Result<usize, store::Error> {
+        Ok(self.all()?.count())
+    }
+
+    /// Get a [`Job`].
+    pub fn get(&self, id: &ObjectId) -> Result<Option<Job>, store::Error> {
+        self.raw.get(id)
+    }
+}
+
+impl<'a, R> Jobs<'a, R>
+where
+    R: ReadRepository + SignRepository + cob::Store,
+{
+    /// Get a [`JobMut`].
+    pub fn get_mut<'g, C>(&'g mut self, id: &ObjectId) -> Result<JobMut<'a, 'g, R>, store::Error> {
+        let job = self
+            .raw
+            .get(id)?
+            .ok_or_else(move || store::Error::NotFound(TYPENAME.clone(), *id))?;
+
+        Ok(JobMut {
+            id: *id,
+            job,
+            store: self,
+        })
+    }
+
+    pub fn create<'g, G>(
+        &'g mut self,
+        oid: Oid,
+        signer: &G,
+    ) -> Result<JobMut<'a, 'g, R>, store::Error>
+    where
+        G: Signer,
+    {
+        let (id, job) = store::Transaction::initial::<_, _, Transaction<R>>(
+            "Request job",
+            &mut self.raw,
+            signer,
+            |tx, _| {
+                tx.request(oid)?;
+                Ok(())
+            },
+        )?;
+
+        Ok(JobMut {
+            id,
+            job,
+            store: self,
+        })
+    }
+}
+
+pub struct JobMut<'a, 'g, R> {
+    pub id: ObjectId,
+
+    job: Job,
+    store: &'g mut Jobs<'a, R>,
+}
+
+impl<R> Deref for JobMut<'_, '_, R> {
+    type Target = Job;
+
+    fn deref(&self) -> &Self::Target {
+        &self.job
+    }
+}
+
+impl<'a, 'g, R> JobMut<'a, 'g, R>
+where
+    R: WriteRepository + cob::Store,
+{
+    pub fn new(id: ObjectId, job: Job, store: &'g mut Jobs<'a, R>) -> Self {
+        Self { id, job, store }
+    }
+
+    pub fn id(&self) -> &ObjectId {
+        &self.id
+    }
+
+    /// Reload the patch data from storage.
+    pub fn reload(&mut self) -> Result<(), store::Error> {
+        self.job = self
+            .store
+            .get(&self.id)?
+            .ok_or_else(|| store::Error::NotFound(TYPENAME.clone(), self.id))?;
+
+        Ok(())
+    }
+
+    /// Start a new [`Run`] for the node, where the run is identified by the
+    /// given [`Uuid`].
+    pub fn run<G>(&mut self, uuid: Uuid, log: Url, signer: &G) -> Result<EntryId, store::Error>
+    where
+        G: Signer,
+    {
+        self.transaction("Run node job", signer, |tx| tx.run(uuid, log))
+    }
+
+    /// Finish a [`Run`], identified by the given [`Uuid`], for the node, with
+    /// the provided [`Reason`].
+    pub fn finish<G>(
+        &mut self,
+        uuid: Uuid,
+        reason: Reason,
+        signer: &G,
+    ) -> Result<EntryId, store::Error>
+    where
+        G: Signer,
+    {
+        self.transaction("Finished node job", signer, |tx| tx.finish(uuid, reason))
+    }
+
+    pub fn transaction<G, F>(
+        &mut self,
+        message: &str,
+        signer: &G,
+        operations: F,
+    ) -> Result<EntryId, store::Error>
+    where
+        G: Signer,
+        F: FnOnce(&mut Transaction<R>) -> Result<(), store::Error>,
+    {
+        let mut tx = Transaction::default();
+        operations(&mut tx)?;
+
+        let (job, commit) = tx.0.commit(message, self.id, &mut self.store.raw, signer)?;
+        self.job = job;
+
+        Ok(commit)
+    }
+}
+
+pub struct Transaction<R: ReadRepository>(store::Transaction<Job, R>);
+
+impl<R> From<store::Transaction<Job, R>> for Transaction<R>
+where
+    R: ReadRepository,
+{
+    fn from(tx: store::Transaction<Job, R>) -> Self {
+        Self(tx)
+    }
+}
+
+impl<R> From<Transaction<R>> for store::Transaction<Job, R>
+where
+    R: ReadRepository,
+{
+    fn from(Transaction(tx): Transaction<R>) -> Self {
+        tx
+    }
+}
+
+impl<R> Default for Transaction<R>
+where
+    R: ReadRepository,
+{
+    fn default() -> Self {
+        Self(Default::default())
+    }
+}
+
+impl<R> Deref for Transaction<R>
+where
+    R: ReadRepository,
+{
+    type Target = store::Transaction<Job, R>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl<R> DerefMut for Transaction<R>
+where
+    R: ReadRepository,
+{
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.0
+    }
+}
+
+impl<R> Transaction<R>
+where
+    R: ReadRepository,
+{
+    pub fn request(&mut self, oid: Oid) -> Result<(), store::Error> {
+        self.0.push(Action::Request { oid })
+    }
+
+    pub fn run(&mut self, uuid: Uuid, log: Url) -> Result<(), store::Error> {
+        self.0.push(Action::Run { uuid, log })
+    }
+
+    pub fn finish(&mut self, uuid: Uuid, reason: Reason) -> Result<(), store::Error> {
+        self.0.push(Action::Finished { uuid, reason })
+    }
+}
+
+#[cfg(test)]
+#[allow(clippy::unwrap_used)]
+mod test {
+    use radicle::crypto::Signer;
+    use radicle::git::{raw::Repository, Oid};
+    use radicle::test;
+    use url::Url;
+    use uuid::Uuid;
+
+    use crate::{Jobs, Reason, Run, Runs, Status};
+
+    fn node_run() -> (Uuid, Url) {
+        let uuid = Uuid::new_v4();
+        let log = Url::parse(&format!("https://example.com/ci/logs?run={}", uuid)).unwrap();
+        (uuid, log)
+    }
+
+    fn commit(repo: &Repository) -> Oid {
+        let tree = {
+            let tree = repo.treebuilder(None).unwrap();
+            let oid = tree.write().unwrap();
+            repo.find_tree(oid).unwrap()
+        };
+
+        let author = repo.signature().unwrap();
+        repo.commit(None, &author, &author, "Test Commit", &tree, &[])
+            .unwrap()
+            .into()
+    }
+
+    #[test]
+    fn e2e() {
+        let test::setup::NodeWithRepo {
+            node: alice, repo, ..
+        } = test::setup::NodeWithRepo::default();
+        let oid = commit(&repo.backend);
+        let mut jobs = Jobs::open(&*repo).unwrap();
+
+        let test::setup::NodeWithRepo { node: bob, .. } = test::setup::NodeWithRepo::default();
+        let mut job = jobs.create(oid, &alice.signer).unwrap();
+
+        let (alice_uuid, alice_log) = node_run();
+        job.run(alice_uuid, alice_log.clone(), &alice.signer)
+            .unwrap();
+
+        let (bob_uuid, bob_log) = node_run();
+        job.run(bob_uuid, bob_log.clone(), &bob.signer).unwrap();
+
+        let alice_runs = job.runs_of(alice.signer.public_key());
+        assert!(alice_runs.is_some());
+        assert_eq!(
+            *alice_runs.unwrap(),
+            [(alice_uuid, Run::new(alice_log))]
+                .into_iter()
+                .collect::<Runs>()
+        );
+
+        let bob_runs = job.runs_of(bob.signer.public_key());
+        assert!(bob_runs.is_some());
+        assert_eq!(
+            *bob_runs.unwrap(),
+            [(bob_uuid, Run::new(bob_log))]
+                .into_iter()
+                .collect::<Runs>()
+        );
+
+        job.finish(alice_uuid, Reason::Succeeded, &alice.signer)
+            .unwrap();
+
+        let finished = job.finished();
+        assert!(finished.contains_key(alice.signer.public_key()));
+        assert!(!finished.contains_key(bob.signer.public_key()));
+
+        job.finish(bob_uuid, Reason::Failed, &bob.signer).unwrap();
+
+        let succeeded = job.succeeded();
+        assert!(succeeded.contains_key(alice.signer.public_key()));
+        assert!(!succeeded.contains_key(bob.signer.public_key()));
+        let failed = job.failed();
+        assert!(!failed.contains_key(alice.signer.public_key()));
+        assert!(failed.contains_key(bob.signer.public_key()));
+        let started = job.started();
+        assert!(started.is_empty());
+    }
+
+    #[test]
+    fn missing_commit() {
+        let test::setup::NodeWithRepo {
+            node: alice, repo, ..
+        } = test::setup::NodeWithRepo::default();
+        let mut jobs = Jobs::open(&*repo).unwrap();
+        let oid = test::arbitrary::oid();
+        let job = jobs.create(oid, &alice.signer);
+        assert!(job.is_err())
+    }
+
+    #[test]
+    fn idempotent_create() {
+        let test::setup::NodeWithRepo {
+            node: alice, repo, ..
+        } = test::setup::NodeWithRepo::default();
+        let oid = commit(&repo.backend);
+        let mut jobs = Jobs::open(&*repo).unwrap();
+        let job1 = {
+            let job1 = jobs.create(oid, &alice.signer).unwrap();
+            job1.id
+        };
+        let job2 = {
+            let job2 = jobs.create(oid, &alice.signer).unwrap();
+            job2.id
+        };
+
+        assert_eq!(job1, job2);
+        assert_eq!(jobs.get(&job1).unwrap(), jobs.get(&job2).unwrap());
+    }
+
+    #[test]
+    fn runs_insertion_order_iteration() {
+        let mut runs = Runs::default();
+        let uuids = (0..10).map(|_| Uuid::new_v4()).collect::<Vec<_>>();
+        for uuid in &uuids {
+            runs.insert(
+                *uuid,
+                Run {
+                    status: Status::Started,
+                    log: Url::parse("https://example.com/ci/logs").unwrap(),
+                },
+            );
+        }
+
+        assert_eq!(
+            uuids,
+            runs.iter()
+                .map(|(uuid, _)| uuid)
+                .copied()
+                .collect::<Vec<_>>()
+        )
+    }
+}