Refactor worker

src/agent/mod.rs

@@ -1,8 +1,6 @@
 mod api;
-mod results;
 
-use crate::agent::api::AgentApi;
-use crate::agent::results::ResultsUploader;
+pub use crate::agent::api::AgentApi;
 use crate::config::Config;
 use crate::crates::Crate;
 use crate::db::{Database, QueryUtils};
@@ -149,7 +147,6 @@ fn run_heartbeat(url: &str, token: &str) {
 fn run_experiment(
     agent: &Agent,
     workspace: &Workspace,
-    db: &ResultsUploader,
     threads_count: usize,
     past_experiment: &mut Option<String>,
 ) -> Result<(), (Option<Box<Experiment>>, Error)> {
@@ -173,9 +170,14 @@ fn run_experiment(
         }
     }
 
-    crate::runner::run_ex(&ex, workspace, db, threads_count, &agent.config, &|| {
-        agent.next_crate(&ex.name)
-    })
+    crate::runner::run_ex(
+        &ex,
+        workspace,
+        &agent.api,
+        threads_count,
+        &agent.config,
+        &|| agent.next_crate(&ex.name),
+    )
     .map_err(|err| (Some(Box::new(ex)), err))?;
     Ok(())
 }
@@ -188,15 +190,14 @@ pub fn run(
     workspace: &Workspace,
 ) -> Fallible<()> {
     let agent = Agent::new(url, token, caps)?;
-    let db = results::ResultsUploader::new(&agent.api);
 
     run_heartbeat(url, token);
     health_thread();
 
     let mut past_experiment = None;
     loop {
         if let Err((ex, err)) =
-            run_experiment(&agent, workspace, &db, threads_count, &mut past_experiment)
+            run_experiment(&agent, workspace, threads_count, &mut past_experiment)
         {
             utils::report_failure(&err);
             if let Some(ex) = ex {

src/results/db.rs

@@ -233,6 +233,24 @@ impl<'a> WriteResults for DatabaseDB<'a> {
     }
 }
 
+impl crate::runner::RecordProgress for DatabaseDB<'_> {
+    fn record_progress(
+        &self,
+        ex: &Experiment,
+        krate: &Crate,
+        toolchain: &Toolchain,
+        log: &[u8],
+        result: &TestResult,
+        version: Option<(&Crate, &Crate)>,
+    ) -> Fallible<()> {
+        self.store_result(ex, krate, toolchain, result, log, EncodingType::Plain)?;
+        if let Some((old, new)) = version {
+            self.update_crate_version(ex, old, new)?;
+        }
+        Ok(())
+    }
+}
+
 impl<'a> DeleteResults for DatabaseDB<'a> {
     fn delete_all_results(&self, ex: &Experiment) -> Fallible<()> {
         self.db

src/runner/mod.rs

@@ -7,11 +7,12 @@ use crate::config::Config;
 use crate::crates::Crate;
 use crate::experiments::{Experiment, Mode};
 use crate::prelude::*;
-use crate::results::{TestResult, WriteResults};
+use crate::results::TestResult;
 use crate::runner::worker::{DiskSpaceWatcher, Worker};
 use rustwide::Workspace;
 use std::thread::scope;
 use std::time::Duration;
+pub use worker::RecordProgress;
 
 const DISK_SPACE_WATCHER_INTERVAL: Duration = Duration::from_secs(30);
 const DISK_SPACE_WATCHER_THRESHOLD: f32 = 0.80;
@@ -20,10 +21,10 @@ const DISK_SPACE_WATCHER_THRESHOLD: f32 = 0.80;
 #[error("overridden task result to {0}")]
 pub struct OverrideResult(TestResult);
 
-pub fn run_ex<DB: WriteResults + Sync>(
+pub fn run_ex(
     ex: &Experiment,
     workspace: &Workspace,
-    db: &DB,
+    api: &dyn RecordProgress,
     threads_count: usize,
     config: &Config,
     next_crate: &(dyn Fn() -> Fallible<Option<Crate>> + Send + Sync),
@@ -82,7 +83,16 @@ pub fn run_ex<DB: WriteResults + Sync>(
     info!("running tasks in {} threads...", threads_count);
 
     let workers = (0..threads_count)
-        .map(|i| Worker::new(format!("worker-{i}"), workspace, ex, config, db, next_crate))
+        .map(|i| {
+            Worker::new(
+                format!("worker-{i}"),
+                workspace,
+                ex,
+                config,
+                api,
+                next_crate,
+            )
+        })
         .collect::<Vec<_>>();
 
     let disk_watcher = DiskSpaceWatcher::new(