diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 0000000..abc83aa --- /dev/null +++ b/.github/dependabot.yml @@ -0,0 +1,12 @@ +version: 2 +updates: + - package-ecosystem: cargo + directory: / + schedule: + interval: weekly + open-pull-requests-limit: 10 + - package-ecosystem: github-actions + directory: / + schedule: + interval: weekly + open-pull-requests-limit: 10 diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 045a26a..b5b0803 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -5,14 +5,43 @@ on: branches: [main] pull_request: +permissions: + contents: read + +env: + CARGO_TERM_COLOR: always + jobs: - rust: + lint-test: + name: Lint, test, and package runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - uses: dtolnay/rust-toolchain@stable with: components: clippy, rustfmt - - run: cargo fmt --check - - run: cargo clippy --all-targets -- -D warnings - - run: cargo test + - uses: Swatinem/rust-cache@v2 + - run: cargo fmt --all --check + - run: cargo clippy --all-targets --all-features -- -D warnings + - run: cargo test --all-targets --all-features + - run: RUSTDOCFLAGS="-D warnings" cargo doc --no-deps --all-features + - run: cargo package + + msrv: + name: MSRV + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: dtolnay/rust-toolchain@1.85.0 + - uses: Swatinem/rust-cache@v2 + - run: cargo check --all-targets --all-features + + dependencies: + name: Dependency policy + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: EmbarkStudios/cargo-deny-action@v2 + with: + command: check + command-arguments: advisories bans licenses sources diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml new file mode 100644 index 0000000..a85cf9b --- /dev/null +++ b/.github/workflows/release.yml @@ -0,0 +1,31 @@ +name: Release + +on: + push: + branches: [main] + workflow_dispatch: + +permissions: + contents: write + pull-requests: write + +env: + CARGO_TERM_COLOR: always + +jobs: + release-plz: + name: Release-plz + runs-on: ubuntu-latest + concurrency: + group: release-plz-${{ github.ref }} + cancel-in-progress: true + steps: + - uses: actions/checkout@v4 + with: + fetch-depth: 0 + - uses: dtolnay/rust-toolchain@stable + - uses: Swatinem/rust-cache@v2 + - uses: release-plz/action@v0.5 + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + CARGO_REGISTRY_TOKEN: ${{ secrets.CARGO_REGISTRY_TOKEN }} diff --git a/Cargo.toml b/Cargo.toml index 9ed8efb..c79cfa0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,7 +1,8 @@ [package] name = "absurd-rust-sdk" version = "0.1.0" -edition = "2021" +edition = "2024" +rust-version = "1.85" authors = ["Manuel Schiller"] description = "Community Rust SDK for Absurd, a Postgres-native durable workflow system" license = "Apache-2.0" @@ -9,23 +10,41 @@ repository = "https://github.com/manuschillerdev/absurd-rust-sdk" readme = "README.md" keywords = ["absurd", "postgres", "workflow", "durable", "queue"] categories = ["asynchronous", "database"] +include = ["Cargo.toml", "LICENSE", "README.md", "examples/**", "src/**", "tests/**"] [lib] name = "absurd_rust_sdk" path = "src/lib.rs" [dependencies] -chrono = { version = "0.4", features = ["serde"] } -deadpool-postgres = { version = "0.14", features = ["rt_tokio_1"] } +chrono = { version = "0.4", default-features = false, features = ["clock", "serde", "std"] } +deadpool-postgres = { version = "0.14", default-features = false, features = ["rt_tokio_1"] } hostname = "0.4" -serde = { version = "1", features = ["derive"] } -serde_json = "1" -thiserror = "2" -tokio = { version = "1", features = ["macros", "rt-multi-thread", "signal", "sync", "time"] } -tokio-postgres = { version = "0.7", features = ["runtime", "with-chrono-0_4", "with-serde_json-1", "with-uuid-1"] } +serde = { version = "1.0", features = ["derive"] } +serde_json = "1.0" +thiserror = "2.0" +tokio = { version = "1.21", features = ["macros", "rt-multi-thread", "signal", "sync", "time"] } +tokio-postgres = { version = "0.7", default-features = false, features = ["runtime", "with-chrono-0_4", "with-serde_json-1", "with-uuid-1"] } tracing = "0.1" -uuid = { version = "1", features = ["serde", "v4"] } +uuid = { version = "1.0", default-features = false, features = ["serde", "std", "v4"] } [dev-dependencies] -tokio = { version = "1", features = ["macros", "rt-multi-thread", "time"] } +tokio = { version = "1.21", features = ["macros", "rt-multi-thread", "time"] } tracing-subscriber = { version = "0.3", features = ["env-filter"] } + +[lints.rust] +future_incompatible = { level = "deny", priority = -1 } +nonstandard_style = { level = "deny", priority = -1 } +rust_2018_idioms = { level = "deny", priority = -1 } +unsafe_code = "forbid" + +[lints.clippy] +all = { level = "deny", priority = -1 } +cargo = { level = "deny", priority = -1 } +dbg_macro = "deny" +expect_used = "deny" +multiple_crate_versions = "allow" +panic = "deny" +todo = "deny" +unimplemented = "deny" +unwrap_used = "deny" diff --git a/README.md b/README.md index 0d685d4..67e694c 100644 --- a/README.md +++ b/README.md @@ -115,6 +115,21 @@ Next hardening targets: Environment resolution uses `ABSURD_DATABASE_URL`, then `DATABASE_URL`, then `PGDATABASE`, then `postgresql://localhost/absurd`. +## Development + +The crate targets Rust 1.85+ and edition 2024. + +```sh +cargo fmt --all --check +cargo clippy --all-targets --all-features -- -D warnings +cargo test --all-targets --all-features +RUSTDOCFLAGS="-D warnings" cargo doc --no-deps --all-features +cargo deny check advisories bans licenses sources +cargo package +``` + +`Cargo.lock` is intentionally not committed because this is a library crate; CI resolves the current compatible dependency graph and Dependabot tracks manifest/action updates. + ## License Apache-2.0 diff --git a/deny.toml b/deny.toml new file mode 100644 index 0000000..5fd7458 --- /dev/null +++ b/deny.toml @@ -0,0 +1,24 @@ +[advisories] +ignore = [] + +[licenses] +allow = [ + "Apache-2.0", + "Apache-2.0 WITH LLVM-exception", + "BSL-1.0", + "MIT", + "Unicode-3.0", + "Zlib", +] +confidence-threshold = 0.8 + +[bans] +highlight = "all" +multiple-versions = "warn" +wildcards = "deny" + +[sources] +allow-git = [] +allow-registry = ["https://github.com/rust-lang/crates.io-index"] +unknown-git = "deny" +unknown-registry = "deny" diff --git a/rust-toolchain.toml b/rust-toolchain.toml new file mode 100644 index 0000000..92a57c5 --- /dev/null +++ b/rust-toolchain.toml @@ -0,0 +1,4 @@ +[toolchain] +channel = "stable" +components = ["clippy", "rustfmt"] +profile = "minimal" diff --git a/rustfmt.toml b/rustfmt.toml new file mode 100644 index 0000000..b1b3c34 --- /dev/null +++ b/rustfmt.toml @@ -0,0 +1,4 @@ +edition = "2024" +newline_style = "Unix" +use_field_init_shorthand = true +use_try_shorthand = true diff --git a/src/client.rs b/src/client.rs index 4fd3fa9..75ec4ae 100644 --- a/src/client.rs +++ b/src/client.rs @@ -1,14 +1,14 @@ use crate::context::TaskContext; -use crate::error::{map_database_error, Error, Result}; +use crate::error::{Error, Result, map_database_error}; use crate::executor::execute_claimed_catching; use crate::task::Task; use crate::types::{ - duration_seconds_ceil, validate_queue_name, CancellationPolicy, ClaimedTask, CleanupResult, - Json, SpawnOptions, SpawnResult, TaskOptions, WorkBatchOptions, WorkerOptions, + CancellationPolicy, ClaimedTask, CleanupResult, Json, SpawnOptions, SpawnResult, TaskOptions, + WorkBatchOptions, WorkerOptions, duration_seconds_ceil, validate_queue_name, }; use crate::worker::Worker; use deadpool_postgres::{Config as PoolConfig, Pool, Runtime}; -use serde::{de::DeserializeOwned, Serialize}; +use serde::{Serialize, de::DeserializeOwned}; use serde_json::Value; use std::collections::HashMap; use std::future::Future; diff --git a/src/context.rs b/src/context.rs index f9c592b..a1d8c41 100644 --- a/src/context.rs +++ b/src/context.rs @@ -1,8 +1,8 @@ -use crate::error::{map_database_error, Error, Result}; -use crate::types::{duration_seconds_ceil, ClaimedTask, Json}; +use crate::error::{Error, Result, map_database_error}; +use crate::types::{ClaimedTask, Json, duration_seconds_ceil}; use chrono::{DateTime, Utc}; use deadpool_postgres::Pool; -use serde::{de::DeserializeOwned, Serialize}; +use serde::{Serialize, de::DeserializeOwned}; use serde_json::{Map, Value}; use std::collections::{HashMap, HashSet}; use std::future::Future; diff --git a/src/executor.rs b/src/executor.rs index f329874..30057f4 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -1,6 +1,6 @@ use crate::client::{RegisteredTask, TaskHandler}; use crate::context::TaskContext; -use crate::error::{map_database_error, Error, Result}; +use crate::error::{Error, Result, map_database_error}; use crate::types::{ClaimedTask, Json, UnknownTaskPolicy}; use deadpool_postgres::Pool; use serde_json::json; diff --git a/src/types.rs b/src/types.rs index fe69770..30038cd 100644 --- a/src/types.rs +++ b/src/types.rs @@ -1,5 +1,5 @@ use crate::error::{Error, Result}; -use serde_json::{json, Map, Value}; +use serde_json::{Map, Value, json}; use std::time::Duration; use uuid::Uuid; diff --git a/src/worker.rs b/src/worker.rs index c14304d..1958caf 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -1,4 +1,4 @@ -use crate::client::{claim_tasks, RegisteredTask}; +use crate::client::{RegisteredTask, claim_tasks}; use crate::error::{Error, Result}; use crate::executor::execute_claimed_catching; use crate::types::WorkerOptions; diff --git a/tests/integration.rs b/tests/integration.rs index d5b15c8..6ee2f73 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -1,5 +1,13 @@ -use absurd_rust_sdk::{AwaitEventOptions, Client, Error, Result, Task, WorkBatchOptions}; +use absurd_rust_sdk::{ + AwaitEventOptions, Client, Error, Result, RetryStrategy, SpawnOptions, Task, WorkBatchOptions, +}; +use chrono::{DateTime, Duration as ChronoDuration, Utc}; use serde::{Deserialize, Serialize}; +use serde_json::{Value, json}; +use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, +}; use std::time::Duration; #[derive(Clone, Debug, Deserialize, Serialize)] @@ -12,18 +20,60 @@ struct Output { doubled: i64, } +#[derive(Debug)] +struct TaskRow { + state: String, + attempts: i32, + completed_payload: Option, + cancelled_at: Option>, +} + +#[derive(Debug)] +struct RunRow { + state: String, + wake_event: Option, + available_at: Option>, + failure_reason: Option, +} + fn database_url() -> String { std::env::var("ABSURD_DATABASE_URL") .or_else(|_| std::env::var("DATABASE_URL")) .unwrap_or_else(|_| "postgresql://localhost/absurd_test".to_string()) } +fn random_queue() -> String { + format!("rs_{}", uuid::Uuid::new_v4().simple()) +} + +async fn test_client() -> Result<(String, Client)> { + let queue = random_queue(); + let client = Client::connect_queue(database_url(), &queue).await?; + client.create_queue().await?; + Ok((queue, client)) +} + #[tokio::test] #[ignore = "requires a Postgres database initialized with Absurd SQL"] -async fn basic_typed_task_round_trip() -> Result<()> { - let queue = format!("rust_sdk_{}", uuid::Uuid::new_v4().simple()); +async fn queue_create_list_drop_round_trip() -> Result<()> { + let queue = random_queue(); let client = Client::connect_queue(database_url(), &queue).await?; + client.create_queue().await?; + assert!(client.list_queues().await?.contains(&queue)); + assert_eq!(queue_table_count(&queue).await?, 5); + + client.drop_queue().await?; + assert!(!client.list_queues().await?.contains(&queue)); + assert_eq!(queue_table_count(&queue).await?, 0); + + Ok(()) +} + +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn basic_typed_task_round_trip() -> Result<()> { + let (queue, client) = test_client().await?; let task = Task::::new("double").queue(&queue); client.register(&task, |params, mut ctx| async move { @@ -43,12 +93,309 @@ async fn basic_typed_task_round_trip() -> Result<()> { .await?; assert_eq!(worked, 1); - let payload = fetch_task_payload(&queue, spawned.task_id).await?; - assert_eq!(payload.state, "completed"); + let task = fetch_task(&queue, spawned.task_id).await?; + assert_eq!(task.state, "completed"); + assert_eq!(task.completed_payload, Some(json!({ "doubled": 42 }))); + + client.drop_queue().await?; + Ok(()) +} + +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn step_checkpoint_is_reused_after_retry() -> Result<()> { + let (queue, client) = test_client().await?; + let executions = Arc::new(AtomicUsize::new(0)); + let attempts = Arc::new(AtomicUsize::new(0)); + + let task = Task::<(), Value>::new("cached-step") + .queue(&queue) + .default_max_attempts(2); + client.register(&task, { + let executions = Arc::clone(&executions); + let attempts = Arc::clone(&attempts); + move |(), mut ctx| { + let executions = Arc::clone(&executions); + let attempts = Arc::clone(&attempts); + async move { + let attempt = attempts.fetch_add(1, Ordering::SeqCst) + 1; + let value: i64 = ctx + .step("expensive", || { + let executions = Arc::clone(&executions); + async move { + executions.fetch_add(1, Ordering::SeqCst); + Ok(42) + } + }) + .await?; + + if attempt == 1 { + return Err(Error::message("retry after checkpoint")); + } + + Ok(json!({ + "value": value, + "executions": executions.load(Ordering::SeqCst), + })) + } + } + })?; + + let spawned = client.spawn(&task, (), Default::default()).await?; + + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 1); + assert_eq!(executions.load(Ordering::SeqCst), 1); + + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 1); + assert_eq!(executions.load(Ordering::SeqCst), 1); + assert_eq!(attempts.load(Ordering::SeqCst), 2); + + let task = fetch_task(&queue, spawned.task_id).await?; + assert_eq!(task.state, "completed"); + assert_eq!(task.attempts, 2); + assert_eq!( + task.completed_payload, + Some(json!({ "value": 42, "executions": 1 })) + ); + + client.drop_queue().await?; + Ok(()) +} + +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn failed_step_is_not_checkpointed_and_reexecutes() -> Result<()> { + let (queue, client) = test_client().await?; + let invocations = Arc::new(AtomicUsize::new(0)); + + let task = Task::<(), Value>::new("fragile-step") + .queue(&queue) + .default_max_attempts(2); + client.register(&task, { + let invocations = Arc::clone(&invocations); + move |(), mut ctx| { + let invocations = Arc::clone(&invocations); + async move { + let result: String = ctx + .step("fragile", || { + let invocations = Arc::clone(&invocations); + async move { + let invocation = invocations.fetch_add(1, Ordering::SeqCst) + 1; + if invocation == 1 { + Err(Error::message("step failed before checkpoint")) + } else { + Ok("success".to_string()) + } + } + }) + .await?; + Ok(json!({ "result": result })) + } + } + })?; + + let spawned = client.spawn(&task, (), Default::default()).await?; + + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 1); + assert_eq!(invocations.load(Ordering::SeqCst), 1); + + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 1); + assert_eq!(invocations.load(Ordering::SeqCst), 2); + + let task = fetch_task(&queue, spawned.task_id).await?; + assert_eq!(task.state, "completed"); + assert_eq!(task.completed_payload, Some(json!({ "result": "success" }))); + + client.drop_queue().await?; + Ok(()) +} + +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn repeated_step_names_are_numbered() -> Result<()> { + let (queue, client) = test_client().await?; + + let task = Task::<(), Value>::new("loop-steps").queue(&queue); + client.register(&task, |(), mut ctx| async move { + let mut results = Vec::new(); + for i in 0_i64..3 { + let result: i64 = ctx.step("loop-step", || async move { Ok(i * 10) }).await?; + results.push(result); + } + Ok(json!({ "results": results })) + })?; + + let spawned = client.spawn(&task, (), Default::default()).await?; + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 1); + + let task = fetch_task(&queue, spawned.task_id).await?; + assert_eq!(task.state, "completed"); + assert_eq!( + task.completed_payload, + Some(json!({ "results": [0, 10, 20] })) + ); + assert_eq!( + fetch_checkpoints(&queue, spawned.task_id).await?, + vec![ + ("loop-step".to_string(), json!(0)), + ("loop-step#2".to_string(), json!(10)), + ("loop-step#3".to_string(), json!(20)), + ] + ); + + client.drop_queue().await?; + Ok(()) +} + +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn sleep_until_suspends_then_resumes_from_checkpoint() -> Result<()> { + let (queue, client) = test_client().await?; + let executions = Arc::new(AtomicUsize::new(0)); + + let task = Task::<(), Value>::new("sleepy").queue(&queue); + client.register(&task, { + let executions = Arc::clone(&executions); + move |(), mut ctx| { + let executions = Arc::clone(&executions); + async move { + let execution = executions.fetch_add(1, Ordering::SeqCst) + 1; + let wake_at = Utc::now() + ChronoDuration::seconds(1); + ctx.sleep_until("pause", wake_at).await?; + Ok(json!({ "executions": execution })) + } + } + })?; + + let spawned = client.spawn(&task, (), Default::default()).await?; + + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 1); + + let task = fetch_task(&queue, spawned.task_id).await?; + let run = fetch_run(&queue, spawned.run_id).await?; + assert_eq!(task.state, "sleeping"); + assert_eq!(run.state, "sleeping"); + assert!(run.wake_event.is_none()); + + let checkpoints = fetch_checkpoints(&queue, spawned.task_id).await?; + assert_eq!(checkpoints.len(), 1); + assert_eq!(checkpoints[0].0, "pause"); + let checkpoint_wake: DateTime = serde_json::from_value(checkpoints[0].1.clone())?; + assert_eq!(run.available_at, Some(checkpoint_wake)); + + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 0); + + tokio::time::sleep(Duration::from_millis(1_100)).await; + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 1); + + let task = fetch_task(&queue, spawned.task_id).await?; + assert_eq!(task.state, "completed"); + assert_eq!(task.completed_payload, Some(json!({ "executions": 2 }))); + assert_eq!(executions.load(Ordering::SeqCst), 2); + + client.drop_queue().await?; + Ok(()) +} + +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn pre_emitted_event_is_available_to_late_waiter() -> Result<()> { + let (queue, client) = test_client().await?; + let event_name = format!("pre_emitted_{queue}"); + let payload = json!({ "data": "ready" }); + + client.emit_event(&event_name, &payload, None).await?; + + let task = Task::<(), Value>::new("late-waiter").queue(&queue); + client.register(&task, { + let event_name = event_name.clone(); + move |(), mut ctx| { + let event_name = event_name.clone(); + async move { + let received: Value = ctx.await_event(&event_name).await?; + Ok(json!({ "received": received })) + } + } + })?; + + let spawned = client.spawn(&task, (), Default::default()).await?; + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 1); + + let task = fetch_task(&queue, spawned.task_id).await?; + assert_eq!(task.state, "completed"); + assert_eq!(task.completed_payload, Some(json!({ "received": payload }))); + + client.drop_queue().await?; + Ok(()) +} + +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn emitted_event_wakes_all_waiters() -> Result<()> { + let (queue, client) = test_client().await?; + let event_name = format!("broadcast_{queue}"); + + let task = Task::::new("multi-waiter").queue(&queue); + client.register(&task, { + let event_name = event_name.clone(); + move |params, mut ctx| { + let event_name = event_name.clone(); + async move { + let payload: Value = ctx.await_event(&event_name).await?; + Ok(json!({ + "task_num": params["task_num"], + "received": payload, + })) + } + } + })?; + + let spawned = [ + client + .spawn(&task, json!({ "task_num": 1 }), Default::default()) + .await?, + client + .spawn(&task, json!({ "task_num": 2 }), Default::default()) + .await?, + client + .spawn(&task, json!({ "task_num": 3 }), Default::default()) + .await?, + ]; + assert_eq!( - payload.completed_payload, - serde_json::json!({ "doubled": 42 }) + client + .work_batch(WorkBatchOptions::new().batch_size(10)) + .await?, + 3 ); + for item in &spawned { + let task = fetch_task(&queue, item.task_id).await?; + let run = fetch_run(&queue, item.run_id).await?; + assert_eq!(task.state, "sleeping"); + assert_eq!(run.state, "sleeping"); + assert_eq!(run.wake_event.as_deref(), Some(event_name.as_str())); + } + + let payload = json!({ "data": "broadcast" }); + client.emit_event(&event_name, &payload, None).await?; + + assert_eq!( + client + .work_batch(WorkBatchOptions::new().batch_size(10)) + .await?, + 3 + ); + for (index, item) in spawned.iter().enumerate() { + let task = fetch_task(&queue, item.task_id).await?; + assert_eq!(task.state, "completed"); + assert_eq!( + task.completed_payload, + Some(json!({ + "task_num": index + 1, + "received": payload, + })) + ); + } client.drop_queue().await?; Ok(()) @@ -57,27 +404,25 @@ async fn basic_typed_task_round_trip() -> Result<()> { #[tokio::test] #[ignore = "requires a Postgres database initialized with Absurd SQL"] async fn event_timeout_can_be_caught_without_recreating_wait() -> Result<()> { - let queue = format!("rust_sdk_{}", uuid::Uuid::new_v4().simple()); - let client = Client::connect_queue(database_url(), &queue).await?; - client.create_queue().await?; + let (queue, client) = test_client().await?; - let task = Task::<(), serde_json::Value>::new("timeout-flow").queue(&queue); + let task = Task::<(), Value>::new("timeout-flow").queue(&queue); client.register(&task, |(), mut ctx| async move { let options = AwaitEventOptions::new() .step_name("wait") .timeout(Duration::ZERO); match ctx - .await_event_with_options::("never", options.clone()) + .await_event_with_options::("never", options.clone()) .await { Err(Error::EventTimeout { .. }) => {} - other => return other.map(|_| serde_json::json!({ "unexpected": true })), + other => return other.map(|_| json!({ "unexpected": true })), } - let payload: serde_json::Value = ctx.await_event_with_options("never", options).await?; + let payload: Value = ctx.await_event_with_options("never", options).await?; - Ok(serde_json::json!({ + Ok(json!({ "timed_out": true, "second_payload": payload, })) @@ -90,31 +435,274 @@ async fn event_timeout_can_be_caught_without_recreating_wait() -> Result<()> { let second = client.work_batch(WorkBatchOptions::new()).await?; assert_eq!(second, 1); - let payload = fetch_task_payload(&queue, spawned.task_id).await?; - assert_eq!(payload.state, "completed"); + let task = fetch_task(&queue, spawned.task_id).await?; + assert_eq!(task.state, "completed"); assert_eq!( - payload.completed_payload, - serde_json::json!({ "timed_out": true, "second_payload": null }) + task.completed_payload, + Some(json!({ "timed_out": true, "second_payload": null })) ); client.drop_queue().await?; Ok(()) } -struct TaskPayload { - state: String, - completed_payload: serde_json::Value, +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn idempotent_spawn_creates_one_task_and_executes_once() -> Result<()> { + let (queue, client) = test_client().await?; + let executions = Arc::new(AtomicUsize::new(0)); + + let task = Task::<(), Value>::new("idempotent-task").queue(&queue); + client.register(&task, { + let executions = Arc::clone(&executions); + move |(), _ctx| { + let executions = Arc::clone(&executions); + async move { + executions.fetch_add(1, Ordering::SeqCst); + Ok(json!({ "done": true })) + } + } + })?; + + let options = SpawnOptions::new().idempotency_key("daily-report:2025-01-15"); + let first = client.spawn(&task, (), options.clone()).await?; + let second = client.spawn(&task, (), options.clone()).await?; + let third = client.spawn(&task, (), options).await?; + + assert!(first.created); + assert!(!second.created); + assert!(!third.created); + assert_eq!(first.task_id, second.task_id); + assert_eq!(first.task_id, third.task_id); + assert_eq!(count_tasks(&queue).await?, 1); + + assert_eq!( + client + .work_batch(WorkBatchOptions::new().batch_size(10)) + .await?, + 1 + ); + assert_eq!(executions.load(Ordering::SeqCst), 1); + + let task = fetch_task(&queue, first.task_id).await?; + assert_eq!(task.state, "completed"); + assert_eq!(task.completed_payload, Some(json!({ "done": true }))); + + client.drop_queue().await?; + Ok(()) +} + +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn failed_task_retries_immediately_until_success() -> Result<()> { + let (queue, client) = test_client().await?; + let attempts = Arc::new(AtomicUsize::new(0)); + + let task = Task::<(), Value>::new("retry-once") + .queue(&queue) + .default_max_attempts(2); + client.register(&task, { + let attempts = Arc::clone(&attempts); + move |(), _ctx| { + let attempts = Arc::clone(&attempts); + async move { + let attempt = attempts.fetch_add(1, Ordering::SeqCst) + 1; + if attempt == 1 { + Err(Error::message("first attempt failed")) + } else { + Ok(json!({ "attempts": attempt })) + } + } + } + })?; + + let spawned = client.spawn(&task, (), Default::default()).await?; + + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 1); + let task = fetch_task(&queue, spawned.task_id).await?; + assert_eq!(task.state, "pending"); + assert_eq!(task.attempts, 2); + + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 1); + let task = fetch_task(&queue, spawned.task_id).await?; + assert_eq!(task.state, "completed"); + assert_eq!(task.attempts, 2); + assert_eq!(task.completed_payload, Some(json!({ "attempts": 2 }))); + + client.drop_queue().await?; + Ok(()) } -async fn fetch_task_payload(queue: &str, task_id: uuid::Uuid) -> Result { +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn fixed_retry_strategy_delays_next_attempt() -> Result<()> { + let (queue, client) = test_client().await?; + let attempts = Arc::new(AtomicUsize::new(0)); + + let task = Task::<(), Value>::new("fixed-retry") + .queue(&queue) + .default_max_attempts(2); + client.register(&task, { + let attempts = Arc::clone(&attempts); + move |(), _ctx| { + let attempts = Arc::clone(&attempts); + async move { + let attempt = attempts.fetch_add(1, Ordering::SeqCst) + 1; + if attempt == 1 { + Err(Error::message("first attempt failed")) + } else { + Ok(json!({ "attempts": attempt })) + } + } + } + })?; + + let spawned = client + .spawn( + &task, + (), + SpawnOptions::new().retry_strategy(RetryStrategy::fixed(Duration::from_secs(1))), + ) + .await?; + + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 1); + let task = fetch_task(&queue, spawned.task_id).await?; + assert_eq!(task.state, "sleeping"); + assert_eq!(task.attempts, 2); + assert_eq!(attempts.load(Ordering::SeqCst), 1); + + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 0); + assert_eq!(attempts.load(Ordering::SeqCst), 1); + + tokio::time::sleep(Duration::from_millis(1_100)).await; + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 1); + + let task = fetch_task(&queue, spawned.task_id).await?; + assert_eq!(task.state, "completed"); + assert_eq!(task.attempts, 2); + assert_eq!(task.completed_payload, Some(json!({ "attempts": 2 }))); + assert_eq!(attempts.load(Ordering::SeqCst), 2); + + client.drop_queue().await?; + Ok(()) +} + +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn unknown_task_is_deferred_by_default() -> Result<()> { + let (queue, client) = test_client().await?; + + let spawned = client + .spawn_named( + "ghost-task", + json!({ "value": 1 }), + SpawnOptions::new().queue(&queue).max_attempts(1), + ) + .await?; + + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 1); + + let task = fetch_task(&queue, spawned.task_id).await?; + assert_eq!(task.state, "sleeping"); + assert_eq!(task.attempts, 1); + + let run = fetch_run(&queue, spawned.run_id).await?; + assert_eq!(run.state, "sleeping"); + assert!(run.failure_reason.is_none()); + assert!(run.available_at.is_some()); + + client.drop_queue().await?; + Ok(()) +} + +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn manual_cancel_pending_task_prevents_claim() -> Result<()> { + let (queue, client) = test_client().await?; + + let task = Task::<(), Value>::new("pending-cancel").queue(&queue); + client.register(&task, |(), _ctx| async move { Ok(json!({ "ok": true })) })?; + + let spawned = client.spawn(&task, (), Default::default()).await?; + client.cancel_task(spawned.task_id, None).await?; + + let task = fetch_task(&queue, spawned.task_id).await?; + assert_eq!(task.state, "cancelled"); + assert!(task.cancelled_at.is_some()); + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 0); + + client.drop_queue().await?; + Ok(()) +} + +async fn pg_client() -> Result { let (pg, connection) = tokio_postgres::connect(&database_url(), tokio_postgres::NoTls).await?; tokio::spawn(async move { let _ = connection.await; }); - let query = format!("SELECT state, completed_payload FROM absurd.t_{queue} WHERE task_id = $1"); + Ok(pg) +} + +async fn fetch_task(queue: &str, task_id: uuid::Uuid) -> Result { + let pg = pg_client().await?; + let query = format!( + "SELECT state, attempts, completed_payload, cancelled_at FROM absurd.t_{queue} WHERE task_id = $1" + ); let row = pg.query_one(&query, &[&task_id]).await?; - Ok(TaskPayload { + Ok(TaskRow { state: row.get(0), - completed_payload: row.get(1), + attempts: row.get(1), + completed_payload: row.get(2), + cancelled_at: row.get(3), }) } + +async fn fetch_run(queue: &str, run_id: uuid::Uuid) -> Result { + let pg = pg_client().await?; + let query = format!( + "SELECT state, wake_event, nullif(available_at, 'infinity'::timestamptz), failure_reason FROM absurd.r_{queue} WHERE run_id = $1" + ); + let row = pg.query_one(&query, &[&run_id]).await?; + Ok(RunRow { + state: row.get(0), + wake_event: row.get(1), + available_at: row.get(2), + failure_reason: row.get(3), + }) +} + +async fn fetch_checkpoints(queue: &str, task_id: uuid::Uuid) -> Result> { + let pg = pg_client().await?; + let query = format!( + "SELECT checkpoint_name, state FROM absurd.c_{queue} WHERE task_id = $1 ORDER BY checkpoint_name" + ); + let rows = pg.query(&query, &[&task_id]).await?; + Ok(rows + .into_iter() + .map(|row| (row.get(0), row.get(1))) + .collect()) +} + +async fn count_tasks(queue: &str) -> Result { + let pg = pg_client().await?; + let query = format!("SELECT count(*) FROM absurd.t_{queue}"); + Ok(pg.query_one(&query, &[]).await?.get(0)) +} + +async fn queue_table_count(queue: &str) -> Result { + let pg = pg_client().await?; + let table_names = vec![ + format!("c_{queue}"), + format!("e_{queue}"), + format!("r_{queue}"), + format!("t_{queue}"), + format!("w_{queue}"), + ]; + Ok(pg + .query_one( + "SELECT count(*) FROM pg_tables WHERE schemaname = 'absurd' AND tablename = ANY($1::text[])", + &[&table_names], + ) + .await? + .get(0)) +}