store outputs in a content-addressed store

Cargo.toml

@@ -5,15 +5,18 @@ edition = "2018"
 
 [dependencies]
 anyhow = "1.0"
+blake3 = "1.3.1"
 byteorder = "1.4"
 clap = { version = "3.2.16", features = ["color", "suggestions", "unstable-v4", "env", "cargo", "derive"] }
 digest = "0.10"
 itertools = "0.10.3"
 libc = "0.2"
+log = { version = "0.4.17", features = ["max_level_trace", "release_max_level_info"] }
 notify = "4"
 roc_std = { path = "vendor/roc_std" }
 serde = { version = "1.0.143", features = ["derive"] }
 serde_json = "1.0.83"
+simple_logger = { version = "2.2.0", features = ["stderr"] }
 sled = "0.34"
 tempfile = "3.2"
 walkdir = "2.3"

src/cli.rs

@@ -1,5 +1,6 @@
 use crate::coordinator::Coordinator;
 use crate::glue;
+use crate::store::Store;
 use anyhow::{Context, Result};
 use clap::Parser;
 use core::mem::MaybeUninit;
@@ -14,20 +15,23 @@ pub struct Cli {
     use_fake_runner: bool,
 
     #[clap(long, default_value = ".rbt")]
-    isolator_root: PathBuf,
+    root_dir: PathBuf,
 }
 
 impl Cli {
     pub fn run(&self) -> Result<()> {
         let rbt = Self::load();
 
-        let mut coordinator = Coordinator::default();
+        let store = Store::new(self.root_dir.join("store")).context("could not open store")?;
+
+        let mut coordinator = Coordinator::new(self.root_dir.join("workspaces"), store);
         coordinator.add_target(rbt.f0.default);
 
         let runner: Box<dyn crate::coordinator::Runner> = if self.use_fake_runner {
+            log::info!("using fake runner");
             Box::new(crate::fake_runner::FakeRunner::default())
         } else {
-            Box::new(crate::runner::Runner::new(self.isolator_root.to_owned()))
+            Box::new(crate::runner::Runner)
         };
 
         while coordinator.has_outstanding_work() {

src/coordinator.rs

@@ -1,16 +1,33 @@
 use crate::glue;
 use crate::job::{self, Job};
+use crate::store::Store;
+use crate::workspace::Workspace;
 use anyhow::{Context, Result};
 use std::collections::{HashMap, HashSet};
+use std::path::PathBuf;
 
-#[derive(Debug, Default)]
+#[derive(Debug)]
 pub struct Coordinator {
+    workspace_root: PathBuf,
+    store: Store,
+
     jobs: HashMap<job::Id, Job>,
     blocked: HashMap<job::Id, HashSet<job::Id>>,
     ready: Vec<job::Id>,
 }
 
 impl Coordinator {
+    pub fn new(workspace_root: PathBuf, store: Store) -> Self {
+        Coordinator {
+            workspace_root,
+            store,
+
+            jobs: HashMap::default(),
+            blocked: HashMap::default(),
+            ready: Vec::default(),
+        }
+    }
+
     pub fn add_target(&mut self, top_job: glue::Job) {
         // Note: this data structure is going to grow the ability to refer to other
         // jobs as soon as it's possibly feasible. When that happens, a depth-first
@@ -25,18 +42,30 @@ impl Coordinator {
     }
 
     pub fn run_next<R: Runner>(&mut self, runner: &R) -> Result<()> {
-        let next = match self.ready.pop() {
+        let id = match self.ready.pop() {
             Some(id) => id,
             None => anyhow::bail!("no work ready to do"),
         };
 
-        runner
-            .run(
-                self.jobs
-                    .get(&next)
-                    .context("had a bad job ID in Coordinator.ready")?,
-            )
-            .context("could not run job")?;
+        let job = self
+            .jobs
+            .get(&id)
+            .context("had a bad job ID in Coordinator.ready")?;
+
+        log::debug!("preparing to run job {}", job.id);
+
+        if self.store.for_job(job).is_none() {
+            let workspace = Workspace::create(&self.workspace_root, job)
+                .with_context(|| format!("could not create workspace for job {}", job.id))?;
+
+            runner.run(job, &workspace).context("could not run job")?;
+
+            self.store
+                .store_from_workspace(job, workspace)
+                .context("could not store job output")?;
+        } else {
+            log::debug!("already had output of this job; skipping");
+        }
 
         // Now that we're done running the job, we update our bookkeeping to
         // figure out what running that job just unblocked.
@@ -46,13 +75,14 @@ impl Coordinator {
         let mut newly_unblocked = vec![]; // avoiding mutating both fields of self in the loop below
 
         self.blocked.retain(|blocked, blockers| {
-            let removed = blockers.remove(&next);
+            let removed = blockers.remove(&id);
             if !removed {
                 return false;
             }
 
             let no_blockers_remaining = blockers.is_empty();
             if no_blockers_remaining {
+                log::debug!("unblocked {}", blocked);
                 newly_unblocked.push(*blocked)
             }
             !no_blockers_remaining
@@ -64,11 +94,11 @@ impl Coordinator {
 }
 
 pub trait Runner {
-    fn run(&self, job: &Job) -> Result<()>;
+    fn run(&self, job: &Job, workspace: &Workspace) -> Result<()>;
 }
 
 impl Runner for Box<dyn Runner> {
-    fn run(&self, job: &Job) -> Result<()> {
-        self.as_ref().run(job)
+    fn run(&self, job: &Job, workspace: &Workspace) -> Result<()> {
+        self.as_ref().run(job, workspace)
     }
 }

src/fake_runner.rs

@@ -1,13 +1,14 @@
 use crate::coordinator::Runner;
 use crate::job::Job;
+use crate::workspace::Workspace;
 use anyhow::Result;
 
 #[derive(Debug, Default)]
 pub struct FakeRunner {}
 
 impl Runner for FakeRunner {
-    fn run(&self, job: &Job) -> Result<()> {
-        eprintln!("running job: {:#?}", job);
+    fn run(&self, job: &Job, _workspace: &Workspace) -> Result<()> {
+        log::info!("fake-running job {:?}", job);
 
         std::thread::sleep(std::time::Duration::from_millis(500));