Concurrency

imbolc · Jun 28, 2024 · 179cc1b · 179cc1b
1 parent 4cf5154
commit 179cc1b
Show file tree

Hide file tree

Showing 6 changed files with 214 additions and 190 deletions.
diff --git a/.sqlx/query-2873b774933c51e429479514b18eeb761e1702532482e7b310413d48832d71d0.json b/.sqlx/query-2873b774933c51e429479514b18eeb761e1702532482e7b310413d48832d71d0.json
diff --git a/.sqlx/query-61eee517879d1017c23bf070468544070818bee22799cb20d7c4898223260b2f.json b/.sqlx/query-61eee517879d1017c23bf070468544070818bee22799cb20d7c4898223260b2f.json
diff --git a/Cargo.toml b/Cargo.toml
@@ -11,6 +11,7 @@ async-trait = "0.1"
 chrono = { version = "0.4", features = ["std", "serde"] }
 code-path = "0.3"
 displaydoc = "0.2"
+num_cpus = "1"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 source-chain = "0.1"

diff --git a/examples/counter.rs b/examples/counter.rs
@@ -58,14 +58,13 @@ impl Step<Count> for Proceed {
     const RETRY_DELAY: Duration = Duration::from_secs(1);
 
     async fn step(self, _db: &PgPool) -> StepResult<Count> {
-        // return Err(anyhow::anyhow!("bailing").into());
         let Self {
             up_to,
             mut cur,
             started_at,
         } = self;
+
         cur += 1;
-        // println!("1..{up_to}: {cur}");
         if cur < up_to {
             NextStep::now(Proceed {
                 up_to,

diff --git a/src/error.rs b/src/error.rs
@@ -14,13 +14,15 @@ pub enum Error {
     WaiterConnect(#[source] sqlx::Error),
     /// waiter can't start listening to tables changes
     WaiterListen(#[source] sqlx::Error),
+    /// unreachable: worker semaphore is closed
+    UnreachableWorkerSemaphoreClosed(#[source] tokio::sync::AcquireError),
 }
 
 /// The crate result
 pub type Result<T> = StdResult<T, Error>;
 
 /// Error of a task step
-pub type StepError = Box<dyn StdError + 'static>;
+pub type StepError = Box<dyn StdError + 'static + Send + Sync>;
 
 /// Result returning from task steps
 pub type StepResult<T> = StdResult<NextStep<T>, StepError>;