fix: minor bug in registering callback

.github/workflows/ci.yml

@@ -2,9 +2,9 @@ name: CI
 
 on:
   push:
-    branches: [ main ]
+    branches: [main]
   pull_request:
-    branches: [ main ]
+    branches: [main]
 
 env:
   CARGO_TERM_COLOR: always
@@ -48,38 +48,6 @@ jobs:
       - name: Run cargo test with all features
         run: cargo test --all-features --verbose
 
-  features:
-    name: Feature Matrix Tests
-    runs-on: ubuntu-latest
-    strategy:
-      matrix:
-        features:
-          - ""
-          - "--features migrate"
-          - "--features async-std-comp"
-          - "--features async-std-comp-native-tls"
-          - "--features tokio-comp"
-          - "--features tokio-comp-native-tls"
-          - "--all-features"
-    steps:
-      - name: Checkout sources
-        uses: actions/checkout@v4
-
-      - name: Install Rust toolchain
-        uses: dtolnay/rust-toolchain@stable
-
-      - name: Cache cargo registry
-        uses: actions/cache@v4
-        with:
-          path: |
-            ~/.cargo/registry
-            ~/.cargo/git
-            target
-          key: ${{ runner.os }}-cargo-features-${{ hashFiles('**/Cargo.lock') }}
-
-      - name: Test with features
-        run: cargo test ${{ matrix.features }} --verbose
-
   fmt:
     name: Rustfmt
     runs-on: ubuntu-latest

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "apalis-sqlite"
-version = "1.0.0-alpha.5"
+version = "1.0.0-alpha.6"
 authors = ["Njuguna Mureithi <mureithinjuguna@gmail.com>"]
 readme = "README.md"
 edition = "2024"
@@ -27,8 +27,8 @@ features = ["chrono", "sqlite"]
 [dependencies]
 serde = { version = "1", features = ["derive"] }
 serde_json = { version = "1" }
-apalis-core = { version = "1.0.0-alpha.8", features = ["sleep"] }
-apalis-sql = { version = "1.0.0-alpha.8" }
+apalis-core = { version = "1.0.0-alpha.9", features = ["sleep"] }
+apalis-sql = { version = "1.0.0-alpha.9" }
 log = "0.4.21"
 futures = "0.3.30"
 tokio = { version = "1", features = ["rt", "net"], optional = true }
@@ -42,9 +42,11 @@ bytes = "1.1.0"
 
 [dev-dependencies]
 tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
-apalis-core = { version = "1.0.0-alpha.8", features = ["test-utils"] }
+apalis = { version = "1.0.0-alpha.9" }
 apalis-sqlite = { path = ".", features = ["migrate", "tokio-comp"] }
-apalis-workflow = { version = "0.1.0-alpha.5" }
+apalis-workflow = { version = "0.1.0-alpha.6" }
+
+futures-util = "0.3.31"
 
 [package.metadata.docs.rs]
 # defines the configuration attribute `docsrs`

README.md

@@ -105,7 +105,7 @@ async fn main() {
 ```rust,no_run
 #[tokio::main]
 async fn main() {
-    let workflow = WorkFlow::new("odd-numbers-workflow")
+    let workflow = Workflow::new("odd-numbers-workflow")
         .then(|a: usize| async move {
             Ok::<_, WorkflowError>((0..=a).collect::<Vec<_>>())
         })

examples/workflow.rs

@@ -0,0 +1,31 @@
+use apalis::prelude::*;
+use apalis_sqlite::SqliteStorage;
+use apalis_workflow::{TaskFlowSink, Workflow};
+use sqlx::SqlitePool;
+
+#[tokio::main]
+async fn main() {
+    let pool = SqlitePool::connect(":memory:").await.unwrap();
+    SqliteStorage::setup(&pool).await.unwrap();
+    let mut backend = SqliteStorage::new(&pool);
+    backend.push_start(42).await.unwrap();
+
+    async fn task1(task: u32) -> u32 {
+        task + 99
+    }
+    async fn task2(task: u32) -> u32 {
+        task + 1
+    }
+    async fn task3(task: u32, worker: WorkerContext) {
+        assert_eq!(task, 142);
+        worker.stop().unwrap();
+    }
+    let workflow = Workflow::new("test_workflow")
+        .then(task1)
+        .then(task2)
+        .then(task3);
+    let worker = WorkerBuilder::new("rango-tango")
+        .backend(backend)
+        .build(workflow);
+    worker.run().await.unwrap();
+}

queries/backend/register_worker.sql

@@ -1,19 +1,24 @@
 INSERT INTO
     Workers (id, worker_type, storage_name, layers, last_seen, started_at)
-SELECT
+VALUES (
     ?1,
     ?2,
     ?3,
     ?4,
     strftime('%s', 'now'),
     strftime('%s', 'now')
+)
+ON CONFLICT(id) DO UPDATE SET
+    worker_type = excluded.worker_type,
+    storage_name = excluded.storage_name,
+    layers = excluded.layers,
+    last_seen = excluded.last_seen,
+    started_at = excluded.started_at,
+    -- Force a constraint violation to throw an error
+    id = CASE
+        WHEN strftime('%s', 'now') - Workers.last_seen <= ?5
+        THEN NULL  -- This will fail if id has NOT NULL constraint
+        ELSE excluded.id
+    END
 WHERE
-    NOT EXISTS (
-        SELECT
-            1
-        FROM
-            Workers
-        WHERE
-            id = ?1
-            AND strftime('%s', 'now') - last_seen >= ?5
-    );
+    strftime('%s', 'now') - Workers.last_seen > ?5;