Fix: poll DHT in background when worker runs up a workflow + dual-sta…

…ck webserver (#590) Completes: - Closes #278. - Closes #589. Includes: * We no longer block DHT polling of receipts, and, instead, purposefully race execution and DHT finding/polling. * This moves us toward the goal of resolving outside-of-workflow awaited (CIDs), but we'll need another background process to re-run a workflow once known-awaits are resolved (separate ticket). The next set of work will also validate that those outside-of-workflow promises exist, i.e. currently running in another workflow or have already run. * v4/v6 host address settings for the Json-RPC web server so that the web server can handle dual v4/v6 requests on the same port. * Better / ordered error handling around CIDs that fail to resolve. * Workflow status field in the DB (+ migration) (SQLite enum, with special patching). * Workflow retries field in the DB (in same migration). * A new poller implementation that's general purpose. * Removes `async_trait` except for wasmtime impls (due to 1.73).
ipvm-wg · Feb 29, 2024 · 2d77b0a · 2d77b0a
1 parent 913278b
commit 2d77b0a
Show file tree

Hide file tree

Showing 43 changed files with 1,080 additions and 485 deletions.
diff --git a/.envrc b/.envrc
@@ -1,4 +1,4 @@
 use_flake
 
 export RUST_LOG=homestar=debug,homestar_runtime=debug,homestar_wasm=debug,libp2p=info,libp2p_gossipsub::behaviour=debug,tarpc=info,tower_http=debug,jsonrpsee_server=debug,moka=debug
-export RUST_BACKTRACE=full
+export RUST_BACKTRACE=1
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
@@ -80,7 +80,7 @@ repos:
       - id: trailing-whitespace
         exclude: bindings\.rs$
       - id: end-of-file-fixer
-        exclude: \.(txt|json)$
+        exclude: \.(txt|json|patch)$
       - id: check-yaml
       - id: check-json
       - id: check-added-large-files

diff --git a/Cargo.lock b/Cargo.lock
diff --git a/diesel.toml b/diesel.toml
@@ -3,6 +3,7 @@
 
 [print_schema]
 file = "homestar-runtime/src/db/schema.rs"
+patch_file = "homestar-runtime/src/db/schema.patch"
 
 [migrations_directory]
 dir = "homestar-runtime/migrations"
diff --git a/flake.lock b/flake.lock
diff --git a/flake.nix b/flake.nix
@@ -47,12 +47,23 @@
         pkgs = import nixpkgs {inherit system overlays;};
         unstable = import nixos-unstable {inherit system overlays;};
 
-        rust-toolchain = fenix.packages.${system}.fromToolchainFile {
+        file-toolchain = fenix.packages.${system}.fromToolchainFile {
           file = ./rust-toolchain.toml;
           # sha256 = pkgs.lib.fakeSha256;
           sha256 = "sha256-e4mlaJehWBymYxJGgnbuCObVlqMlQSilZ8FljG9zPHY=";
         };
 
+        default-toolchain = fenix.packages.${system}.complete.withComponents [
+          "cargo"
+          "clippy"
+          "llvm-tools-preview"
+          "rustfmt"
+          "rust-src"
+          "rust-std"
+        ];
+
+        rust-toolchain = fenix.packages.${system}.combine [file-toolchain default-toolchain];
+
         rustPlatform = pkgs.makeRustPlatform {
           cargo = rust-toolchain;
           rustc = rust-toolchain;

diff --git a/homestar-invocation/src/receipt.rs b/homestar-invocation/src/receipt.rs
@@ -41,7 +41,7 @@ pub struct Receipt<T> {
 }
 
 impl<T> Receipt<T> {
-    ///
+    /// Create a new [Receipt].
     pub fn new(
         ran: Pointer,
         result: task::Result<T>,

diff --git a/homestar-invocation/src/task/instruction/input.rs b/homestar-invocation/src/task/instruction/input.rs
@@ -81,8 +81,15 @@ where
         F: Fn(Cid) -> BoxFuture<'a, Result<task::Result<T>, ResolveError>> + Clone + Send + Sync,
         Ipld: From<T>,
     {
-        let inputs = resolve_args(self.0, lookup_fn);
-        Ok(Args(inputs.await))
+        let inputs = resolve_args(self.0, lookup_fn).await;
+        for input in inputs.iter() {
+            if let Input::Deferred(awaiting) = input {
+                return Err(ResolveError::UnresolvedCid(
+                    awaiting.instruction_cid().to_string(),
+                ));
+            }
+        }
+        Ok(Args(inputs))
     }
 }
 

diff --git a/homestar-runtime/Cargo.toml b/homestar-runtime/Cargo.toml
@@ -61,6 +61,7 @@ diesel = { version = "2.1", default-features = false, features = [
   "with-deprecated",
   "chrono",
 ] }
+diesel-derive-enum = { version = "2.1", features = ["sqlite"] }
 diesel_migrations = "2.1"
 dotenvy = "0.15"
 dyn-clone = "1.0"
@@ -79,6 +80,7 @@ homestar-workspace-hack = { workspace = true }
 http = "0.2"
 http-serde = "1.1"
 humantime = { workspace = true }
+hyper = { version = "0.14", default-features = false }
 indexmap = { workspace = true }
 ipfs-api = { version = "0.17", optional = true }
 ipfs-api-backend-hyper = { version = "0.6", default-features = false, features = [
@@ -147,7 +149,7 @@ serde_with = { version = "3.5", default-features = false, features = [
 ] }
 stream-cancel = "0.8"
 sysinfo = { version = "0.29", default-features = false, optional = true }
-tabled = { version = "0.14", default-features = false, features = [
+tabled = { version = "0.15", default-features = false, features = [
   "derive",
   "macros",
 ] }

diff --git a/homestar-runtime/config/defaults.toml b/homestar-runtime/config/defaults.toml
@@ -55,6 +55,7 @@ mesh_n = 2
 mesh_outbound_min = 1
 
 [node.network.libp2p.dht]
+enable_resolve_receipts_in_background = true
 p2p_receipt_timeout = 500
 p2p_workflow_info_timeout = 500
 p2p_provider_timeout = 10_000
@@ -74,7 +75,8 @@ max_connections = 10
 server_timeout = 120
 
 [node.network.webserver]
-host = "127.0.0.1"
+v4_host = "127.0.0.1"
+v6_host = "[::1]"
 port = 1337
 timeout = 120
 websocket_capacity = 2048

diff --git a/homestar-runtime/migrations/.keep b/homestar-runtime/migrations/.keep
diff --git a/homestar-runtime/migrations/2024-02-13-141620_add_workflows_status/down.sql b/homestar-runtime/migrations/2024-02-13-141620_add_workflows_status/down.sql
@@ -0,0 +1,2 @@
+ALTER TABLE workflows DROP COLUMN status;
+ALTER TABLE workflows DROP COLUMN retries;
diff --git a/homestar-runtime/migrations/2024-02-13-141620_add_workflows_status/up.sql b/homestar-runtime/migrations/2024-02-13-141620_add_workflows_status/up.sql
@@ -0,0 +1,4 @@
+ALTER TABLE workflows ADD COLUMN status TEXT CHECK(
+    status IN ('pending', 'completed', 'running', 'stuck')) NOT NULL DEFAULT
+            'pending';
+ALTER TABLE workflows ADD COLUMN retries INTEGER NOT NULL DEFAULT 0;
diff --git a/homestar-runtime/src/db.rs b/homestar-runtime/src/db.rs
@@ -23,6 +23,7 @@ use tokio::fs;
 use tracing::info;
 
 #[allow(missing_docs, unused_imports)]
+#[rustfmt::skip]
 pub mod schema;
 pub(crate) mod utils;
 
@@ -248,6 +249,20 @@ pub trait Database: Send + Sync + Clone {
         }
     }
 
+    /// Update workflow status given a Cid to the workflow.
+    fn set_workflow_status(
+        workflow_cid: Cid,
+        status: workflow::Status,
+        conn: &mut Connection,
+    ) -> Result<(), diesel::result::Error> {
+        diesel::update(schema::workflows::dsl::workflows)
+            .filter(schema::workflows::cid.eq(Pointer::new(workflow_cid)))
+            .set(schema::workflows::status.eq(status))
+            .execute(conn)?;
+
+        Ok(())
+    }
+
     /// Store workflow Cid and [Receipt] Cid in the database for inner join.
     fn store_workflow_receipt(
         workflow_cid: Cid,

diff --git a/homestar-runtime/src/db/schema.patch b/homestar-runtime/src/db/schema.patch
@@ -0,0 +1,9 @@
+@@ -25,7 +25,7 @@ diesel::table! {
+         cid -> Text,
+         name -> Nullable<Text>,
+         num_tasks -> Integer,
+         resources -> Binary,
+         created_at -> Timestamp,
+         completed_at -> Nullable<Timestamp>,
++        status -> crate::workflow::StatusMapping,
+-        status -> Text,
diff --git a/homestar-runtime/src/db/schema.rs b/homestar-runtime/src/db/schema.rs
@@ -21,6 +21,8 @@ diesel::table! {
         resources -> Binary,
         created_at -> Timestamp,
         completed_at -> Nullable<Timestamp>,
+        status -> crate::workflow::StatusMapping,
+        retries -> Integer,
     }
 }
 
@@ -34,4 +36,8 @@ diesel::table! {
 diesel::joinable!(workflows_receipts -> receipts (receipt_cid));
 diesel::joinable!(workflows_receipts -> workflows (workflow_cid));
 
-diesel::allow_tables_to_appear_in_same_query!(receipts, workflows, workflows_receipts,);
+diesel::allow_tables_to_appear_in_same_query!(
+    receipts,
+    workflows,
+    workflows_receipts,
+);
diff --git a/homestar-runtime/src/event_handler.rs b/homestar-runtime/src/event_handler.rs
@@ -11,7 +11,6 @@ use crate::{
     settings,
 };
 use anyhow::Result;
-use async_trait::async_trait;
 use fnv::FnvHashMap;
 use libp2p::{
     core::ConnectedPoint, futures::StreamExt, kad::QueryId, rendezvous::Cookie,
@@ -48,7 +47,6 @@ struct Bootstrap {
 }
 
 /// Handler trait for [EventHandler] events.
-#[async_trait]
 pub(crate) trait Handler<DB>
 where
     DB: Database,

diff --git a/homestar-runtime/src/event_handler/event.rs b/homestar-runtime/src/event_handler/event.rs
@@ -21,7 +21,6 @@ use crate::{
     workflow, Db, Receipt,
 };
 use anyhow::Result;
-use async_trait::async_trait;
 #[cfg(feature = "websocket-notify")]
 use homestar_invocation::Pointer;
 use homestar_invocation::Receipt as InvocationReceipt;
@@ -701,7 +700,6 @@ impl PeerRequest {
     }
 }
 
-#[async_trait]
 impl<DB> Handler<DB> for Event
 where
     DB: Database,

diff --git a/homestar-runtime/src/event_handler/notification.rs b/homestar-runtime/src/event_handler/notification.rs
@@ -40,6 +40,7 @@ pub(crate) fn emit_receipt(
             subject = "notification.receipt",
             category = "notification",
             cid = receipt_cid.to_string(),
+            instruction_cid = receipt.instruction().cid().to_string(),
             "emitting receipt to WebSocket"
         );
         if let Some(ipld) = metadata {