diff --git a/docs/parity/absurd-sql-0.4.0-test-parity.md b/docs/parity/absurd-sql-0.4.0-test-parity.md new file mode 100644 index 0000000..1e3e2e6 --- /dev/null +++ b/docs/parity/absurd-sql-0.4.0-test-parity.md @@ -0,0 +1,111 @@ +# Absurd SQL 0.4.0 test parity + +## Purpose + +This document tracks high-signal test parity for the Rust SDK against the upstream Absurd `0.4.0` Python, TypeScript, and Go SDK suites. + +Coverage parity is not the goal. The goal is confidence in public Rust SDK behavior, durable workflow semantics, and intentional Rust deviations documented in `docs/parity/absurd-sql-0.4.0.md`. + +## Target + +- Absurd SQL tag: `0.4.0` +- Absurd SQL commit: `05282a40c8dddc378acdc6933adc4c221583808a` +- Rust crate version: `0.1.0` +- Rust test suites: + - unit tests in `src/**` + - ignored database integration suite in `tests/integration.rs` + +## Scope + +Included: + +- public Rust SDK behavior +- database-backed durable semantics surfaced through the SDK +- failure, cancellation, retry, worker, event, checkpoint, and task-result invariants +- Rust-specific diagnostics and intentional API decisions + +Excluded: + +- duplicate sync/async coverage from Python and TypeScript +- language-specific decorator, contextvar, or global-context behavior +- Go driver matrix tests +- raw claim/execute APIs that Rust intentionally keeps internal +- `absurdctl`, cron, SQL migration, partition maintenance, and storage-engine tests owned by upstream Absurd +- exact traceback-string parity, which Rust intentionally does not implement + +## Test parity matrix + +| Area | Rust status | Existing Rust evidence | High-signal action | +|---|---|---|---| +| Absurd SQL `0.4.0` schema target | Covered | CI installs upstream SQL; DB integration suite runs against it | None | +| Connection defaults | Partial | `Client::from_env*` exists; no direct Rust test for env precedence | Add unit tests for explicit URL, `ABSURD_DATABASE_URL`, `DATABASE_URL`, `PGDATABASE`, and localhost fallback | +| Queue create, list, drop | Covered | `queue_create_list_drop_round_trip` | None | +| Partitioned queue creation | Covered | `partitioned_queue_create_round_trip` | Optional partitioned idempotency smoke; not required now | +| Queue policy get/set | Covered | `queue_policy_round_trip`; queue policy wire-key unit tests | None | +| Typed task registration and spawn | Covered | `basic_typed_task_round_trip` | None | +| Spawn options | Partial | `spawn_options_use_sql_wire_keys`; header and idempotency integration tests | Add max-attempt override/default persistence; reject unregistered spawn without queue; reject registered queue mismatch | +| Spawn headers | Covered | `spawn_headers_are_available_to_task_context`; hook tests | None | +| Idempotent spawn | Covered | `idempotent_spawn_creates_one_task_and_executes_once` | Optional different-key and queue-scope cases; not required now | +| Retry without explicit strategy | Covered | `failed_task_retries_immediately_until_success` | None | +| Fixed retry strategy | Covered | `fixed_retry_strategy_delays_next_attempt` | None | +| Exponential retry strategy | Missing | wire shape exists through `RetryStrategy`, but no DB behavior test | Add DB integration test using `absurd.fake_now` to validate increasing delay and max cap | +| Explicit none retry strategy | Missing | `RetryStrategy::none()` exists; no DB behavior test | Add DB integration test that `kind: none` is persisted and requeues immediately when attempts remain | +| Cancellation by max delay | Missing | wire serialization through `CancellationPolicy`; no DB behavior test | Add DB integration test where a delayed first claim cancels the task | +| Cancellation by max duration | Missing | wire serialization through `CancellationPolicy`; no DB behavior test | Add DB integration test where retry after elapsed duration cancels the task | +| Default task cancellation | Missing | `task_builder_sets_default_cancellation` only checks builder state | Add DB integration test proving task-level default cancellation is applied when spawn options omit cancellation | +| Worker batch execution | Partial | `basic_typed_task_round_trip`; event batch tests | Add mixed success/failure batch test proving failure of one task does not prevent later claimed tasks from running | +| Continuous worker lifecycle | Partial | `worker_close_waits_for_running_tasks` | Add concurrency-limit test for `start_worker` | +| Worker error callback | Covered | `worker_on_error_reports_claim_failures`; `worker_on_error_reports_execution_failures`; unit callback tests | None | +| Durable `TaskContext::step` checkpoint | Covered | `step_checkpoint_is_reused_after_retry`; `failed_step_is_not_checkpointed_and_reexecutes` | Add multistep retry test where only unfinished steps re-execute | +| Decomposed steps | Partial | `decomposed_step_handle_reuses_completed_state_after_retry`; `failed_decomposed_step_is_not_checkpointed_and_reexecutes` | Add repeated `begin_step` / `complete_step` name numbering test | +| Sleep / scheduled resume | Covered | `sleep_until_suspends_then_resumes_from_checkpoint` | None | +| Event await/emit | Partial | `pre_emitted_event_is_available_to_late_waiter`; `emitted_event_wakes_all_waiters`; `event_timeout_can_be_caught_without_recreating_wait` | Add first-write-wins `emit_event` test | +| Task result fetch/await | Partial | terminal fetch/await, failed/cancelled/missing, timeout, same-queue rejection, completed cross-queue await | Add cross-queue pending child wait; add checkpoint-survives-parent-retry-after-child-cleanup test | +| Manual retry task | Covered | default, max-attempt override, spawn-new, missing-task, and non-failed-task tests | None | +| Manual cancel task | Partial | pending cancel and terminal race tests | Add running cancel, sleeping cancel, completed/failed no-op, and missing-task error tests | +| Cleanup tasks/events | Missing | No Rust integration test found; parity doc currently claims coverage | Add `Client::cleanup` / `cleanup_with_limit` integration test for terminal tasks and emitted events | +| Hooks | Covered | before-spawn inject/preserve, wrap execution, and context access tests | None | +| Failure payload shape | Covered | user error, invalid headers, task-result failure payload assertions | None | +| Panic handling | Covered | `panic_failure_payload_records_diagnostic_shape` | None | +| Raw task claim API | Not applicable | intentionally internal in Rust | Do not add public parity tests | +| Sync client API | Not applicable | Rust SDK is intentionally async-only | Do not add | +| Global/decorator task context | Not applicable | Rust uses explicit `TaskContext` | Do not add | +| Logger injection | Not applicable | Rust uses `tracing` and `WorkerOptions::on_error` | Do not add | +| Failure traceback strings | Not applicable | Rust asserts diagnostic payload with `traceback: null` | Do not chase upstream traceback-string tests | + +## Recommended implementation backlog + +### Required high-signal additions + +1. `cleanup_removes_terminal_tasks_and_events_by_ttl` +2. `cancellation_policy_cancels_by_max_delay` +3. `cancellation_policy_cancels_by_max_duration` +4. `default_task_cancellation_is_applied` +5. `exponential_retry_strategy_delays_attempts` +6. `retry_strategy_none_requeues_immediately` +7. `spawn_resolution_applies_defaults_overrides_and_rejects_queue_mismatches` +8. `work_batch_handles_mixed_success_and_failure` +9. `worker_respects_concurrency_limit` +10. `event_emit_is_first_write_wins` +11. `context_await_task_result_waits_for_pending_child_in_other_queue` +12. `context_await_task_result_checkpoint_survives_parent_retry_after_child_cleanup` +13. `manual_cancel_running_sleeping_terminal_and_missing_cases` +14. `multi_step_only_reexecutes_uncompleted_steps` +15. `repeated_decomposed_step_names_are_numbered` +16. `connection_defaults_resolve_env_precedence` + +### Test design notes + +- Prefer one focused integration test per durable invariant. +- Use `SET absurd.fake_now` where possible instead of real sleeps for retry and cancellation timing. +- Keep optional permutation tests out unless they catch a distinct class of regression. +- Keep language-specific upstream tests out of the Rust suite unless Rust exposes an equivalent public surface. + +## Validation commands + +```sh +cargo fmt --all --check +cargo clippy --all-targets --all-features -- -D warnings +cargo test --all-targets --all-features +ABSURD_DATABASE_URL=postgresql://postgres:postgres@localhost:5432/absurd_test cargo test-db +``` diff --git a/src/client.rs b/src/client.rs index bfbcbbe..c806975 100644 --- a/src/client.rs +++ b/src/client.rs @@ -716,20 +716,27 @@ pub(crate) async fn claim_tasks( } fn resolve_database_url(explicit: Option<&str>) -> String { + resolve_database_url_with(explicit, |key| std::env::var(key).ok()) +} + +fn resolve_database_url_with(explicit: Option<&str>, mut env: F) -> String +where + F: FnMut(&str) -> Option, +{ if let Some(explicit) = explicit.filter(|value| !value.trim().is_empty()) { return explicit.to_string(); } - if let Ok(url) = std::env::var("ABSURD_DATABASE_URL") { + if let Some(url) = env("ABSURD_DATABASE_URL") { if !url.trim().is_empty() { return url; } } - if let Ok(url) = std::env::var("DATABASE_URL") { + if let Some(url) = env("DATABASE_URL") { if !url.trim().is_empty() { return url; } } - if let Ok(pgdatabase) = std::env::var("PGDATABASE") { + if let Some(pgdatabase) = env("PGDATABASE") { if !pgdatabase.trim().is_empty() { if pgdatabase.contains("://") || pgdatabase.contains('=') { return pgdatabase; @@ -739,3 +746,66 @@ fn resolve_database_url(explicit: Option<&str>) -> String { } "postgresql://localhost/absurd".to_string() } + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::HashMap; + + #[test] + fn connection_defaults_resolve_env_precedence() { + let env = HashMap::from([ + ("ABSURD_DATABASE_URL", "postgresql://localhost/absurd_env"), + ("DATABASE_URL", "postgresql://localhost/database_env"), + ("PGDATABASE", "pg_database_name"), + ]); + + assert_eq!( + resolve_database_url_with(Some("postgresql://explicit/database"), |key| env + .get(key) + .map(ToString::to_string)), + "postgresql://explicit/database" + ); + assert_eq!( + resolve_database_url_with(Some(" "), |key| env.get(key).map(ToString::to_string)), + "postgresql://localhost/absurd_env" + ); + assert_eq!( + resolve_database_url_with(None, |key| env.get(key).map(ToString::to_string)), + "postgresql://localhost/absurd_env" + ); + + let env_without_absurd = HashMap::from([ + ("DATABASE_URL", "postgresql://localhost/database_env"), + ("PGDATABASE", "pg_database_name"), + ]); + assert_eq!( + resolve_database_url_with(None, |key| env_without_absurd + .get(key) + .map(ToString::to_string)), + "postgresql://localhost/database_env" + ); + + let env_with_plain_pgdatabase = HashMap::from([("PGDATABASE", "pg_database_name")]); + assert_eq!( + resolve_database_url_with(None, |key| env_with_plain_pgdatabase + .get(key) + .map(ToString::to_string)), + "dbname=pg_database_name" + ); + + let env_with_pgdatabase_url = HashMap::from([("PGDATABASE", "postgresql://localhost/pg")]); + assert_eq!( + resolve_database_url_with(None, |key| env_with_pgdatabase_url + .get(key) + .map(ToString::to_string)), + "postgresql://localhost/pg" + ); + + let empty_env = HashMap::<&str, &str>::new(); + assert_eq!( + resolve_database_url_with(None, |key| empty_env.get(key).map(ToString::to_string)), + "postgresql://localhost/absurd" + ); + } +} diff --git a/tests/integration.rs b/tests/integration.rs index d6717aa..a911785 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -4,10 +4,10 @@ //! with Absurd SQL. Run them locally with `ABSURD_DATABASE_URL=... cargo test-db`. use absurd_rust_sdk::{ - AwaitEventOptions, AwaitTaskResultOptions, Client, CreateQueueOptions, Error, Hooks, - QueueDetachMode, QueuePolicyOptions, QueueStorageMode, Result, RetryStrategy, RetryTaskOptions, - SpawnOptions, Task, TaskResultSnapshot, UnknownTaskPolicy, WorkBatchOptions, WorkerErrorKind, - WorkerOptions, + AwaitEventOptions, AwaitTaskResultOptions, CancellationPolicy, Client, CreateQueueOptions, + Error, Hooks, QueueDetachMode, QueuePolicyOptions, QueueStorageMode, Result, RetryStrategy, + RetryTaskOptions, SpawnOptions, Task, TaskResultSnapshot, UnknownTaskPolicy, WorkBatchOptions, + WorkerErrorKind, WorkerOptions, }; use chrono::{DateTime, Duration as ChronoDuration, Utc}; use serde::{Deserialize, Serialize}; @@ -15,7 +15,7 @@ use serde_json::{Map, Value, json}; use std::collections::HashMap; use std::sync::{ Arc, - atomic::{AtomicUsize, Ordering}, + atomic::{AtomicBool, AtomicUsize, Ordering}, }; use std::time::Duration; use tokio::sync::Notify; @@ -47,6 +47,12 @@ struct RunRow { failure_reason: Option, } +#[derive(Debug)] +struct TaskSpawnMetadata { + retry_strategy: Option, + cancellation: Option, +} + fn database_url() -> String { std::env::var("ABSURD_DATABASE_URL") .or_else(|_| std::env::var("DATABASE_URL")) @@ -64,6 +70,52 @@ async fn test_client() -> Result<(String, Client)> { Ok((queue, client)) } +async fn test_client_with_single_connection_pool() +-> Result<(String, Client, deadpool_postgres::Pool)> { + let queue = random_queue(); + let (client, pool) = connect_queue_with_single_connection_pool(&queue).await?; + client.create_queue().await?; + Ok((queue, client, pool)) +} + +async fn connect_queue_with_single_connection_pool( + queue: &str, +) -> Result<(Client, deadpool_postgres::Pool)> { + let mut config = deadpool_postgres::Config::new(); + config.url = Some(database_url()); + config.pool = Some(deadpool_postgres::PoolConfig::new(1)); + let pool = config + .create_pool( + Some(deadpool_postgres::Runtime::Tokio1), + tokio_postgres::NoTls, + ) + .map_err(|err| Error::Config(format!("failed to create Postgres pool: {err}")))?; + + let connection = pool.get().await?; + drop(connection); + + Ok((Client::from_pool(pool.clone(), queue)?, pool)) +} + +fn utc(timestamp: &str) -> Result> { + Ok(DateTime::parse_from_rfc3339(timestamp) + .map_err(|err| Error::message(format!("invalid timestamp {timestamp:?}: {err}")))? + .with_timezone(&Utc)) +} + +async fn set_fake_now( + pool: &deadpool_postgres::Pool, + timestamp: Option>, +) -> Result<()> { + let pg = pool.get().await?; + let value = timestamp + .map(|timestamp| timestamp.to_rfc3339()) + .unwrap_or_default(); + pg.execute("SELECT set_config('absurd.fake_now', $1, false)", &[&value]) + .await?; + Ok(()) +} + #[tokio::test] #[ignore = "requires a Postgres database initialized with Absurd SQL"] async fn queue_create_list_drop_round_trip() -> Result<()> { @@ -198,6 +250,121 @@ async fn basic_typed_task_round_trip() -> Result<()> { Ok(()) } +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn spawn_resolution_applies_defaults_overrides_and_rejects_queue_mismatches() -> Result<()> { + let queue = random_queue(); + let other_queue = random_queue(); + let client = Client::connect_queue(database_url(), &queue) + .await? + .default_max_attempts(4); + client.create_queue().await?; + client.create_queue_in(&other_queue).await?; + + let client_default_task = Task::<(), Value>::new("client-default-attempts").queue(&queue); + client.register(&client_default_task, |(), _ctx| async move { + Ok(json!({ "ok": true })) + })?; + let client_default = client + .spawn(&client_default_task, (), SpawnOptions::new()) + .await?; + assert_eq!( + fetch_task(&queue, client_default.task_id) + .await? + .max_attempts, + Some(4) + ); + + let task_default = Task::<(), Value>::new("task-default-attempts") + .queue(&queue) + .default_max_attempts(2); + client.register(&task_default, |(), _ctx| async move { + Ok(json!({ "ok": true })) + })?; + let task_default_spawned = client.spawn(&task_default, (), SpawnOptions::new()).await?; + assert_eq!( + fetch_task(&queue, task_default_spawned.task_id) + .await? + .max_attempts, + Some(2) + ); + + let overridden = client + .spawn(&task_default, (), SpawnOptions::new().max_attempts(3)) + .await?; + assert_eq!( + fetch_task(&queue, overridden.task_id).await?.max_attempts, + Some(3) + ); + + let missing_queue = client + .spawn_named("unregistered-without-queue", (), SpawnOptions::new()) + .await; + assert!( + matches!(missing_queue, Err(Error::Config(message)) if message.contains("is not registered")) + ); + + let mismatched_queue = client + .spawn( + &task_default, + (), + SpawnOptions::new().queue(other_queue.clone()), + ) + .await; + assert!( + matches!(mismatched_queue, Err(Error::Config(message)) if message.contains("spawn requested queue")) + ); + + client.drop_queue().await?; + client.drop_queue_in(&other_queue).await?; + Ok(()) +} + +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn work_batch_handles_mixed_success_and_failure() -> Result<()> { + let (queue, client) = test_client().await?; + + let ok_a = Task::<(), Value>::new("mixed-ok-a").queue(&queue); + let ok_b = Task::<(), Value>::new("mixed-ok-b").queue(&queue); + let fail = Task::<(), Value>::new("mixed-fail") + .queue(&queue) + .default_max_attempts(1); + + client.register(&ok_a, |(), _ctx| async move { Ok(json!({ "ok": "a" })) })?; + client.register(&ok_b, |(), _ctx| async move { Ok(json!({ "ok": "b" })) })?; + client.register(&fail, |(), _ctx| async move { + Err::(Error::message("mixed batch failure")) + })?; + + let spawned_ok_a = client.spawn(&ok_a, (), SpawnOptions::new()).await?; + let spawned_fail = client.spawn(&fail, (), SpawnOptions::new()).await?; + let spawned_ok_b = client.spawn(&ok_b, (), SpawnOptions::new()).await?; + + assert_eq!( + client + .work_batch(WorkBatchOptions::new().batch_size(10)) + .await?, + 3 + ); + + assert_eq!( + fetch_task(&queue, spawned_ok_a.task_id).await?.state, + "completed" + ); + assert_eq!( + fetch_task(&queue, spawned_fail.task_id).await?.state, + "failed" + ); + assert_eq!( + fetch_task(&queue, spawned_ok_b.task_id).await?.state, + "completed" + ); + + client.drop_queue().await?; + Ok(()) +} + #[tokio::test] #[ignore = "requires a Postgres database initialized with Absurd SQL"] async fn worker_close_waits_for_running_tasks() -> Result<()> { @@ -236,7 +403,7 @@ async fn worker_close_waits_for_running_tasks() -> Result<()> { .map_err(|_| Error::message("worker did not start task"))?; let close = tokio::spawn(worker.close()); - tokio::time::sleep(Duration::from_millis(100)).await; + tokio::task::yield_now().await; assert!(!close.is_finished()); release.notify_one(); @@ -252,6 +419,89 @@ async fn worker_close_waits_for_running_tasks() -> Result<()> { Ok(()) } +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn worker_respects_concurrency_limit() -> Result<()> { + let (queue, client) = test_client().await?; + let task = Task::::new("concurrency-limited").queue(&queue); + let started = Arc::new(AtomicUsize::new(0)); + let release = Arc::new(AtomicBool::new(false)); + let release_notify = Arc::new(Notify::new()); + let (started_tx, mut started_rx) = tokio::sync::mpsc::unbounded_channel::(); + + client.register(&task, { + let started = Arc::clone(&started); + let release = Arc::clone(&release); + let release_notify = Arc::clone(&release_notify); + move |index, _ctx| { + let started = Arc::clone(&started); + let release = Arc::clone(&release); + let release_notify = Arc::clone(&release_notify); + let started_tx = started_tx.clone(); + async move { + started.fetch_add(1, Ordering::SeqCst); + let _ = started_tx.send(index); + while !release.load(Ordering::SeqCst) { + release_notify.notified().await; + } + Ok(json!({ "index": index })) + } + } + })?; + + let spawned = [ + client.spawn(&task, 1, SpawnOptions::new()).await?, + client.spawn(&task, 2, SpawnOptions::new()).await?, + client.spawn(&task, 3, SpawnOptions::new()).await?, + ]; + + let worker = client.start_worker( + WorkerOptions::new() + .concurrency(2) + .batch_size(3) + .poll_interval(Duration::from_millis(25)) + .fatal_on_lease_timeout(false), + )?; + + let mut started_indices = Vec::new(); + for _ in 0..2 { + let index = tokio::time::timeout(Duration::from_secs(5), started_rx.recv()) + .await + .map_err(|_| Error::message("worker did not start the first two tasks"))? + .ok_or_else(|| Error::message("started channel closed"))?; + started_indices.push(index); + } + + assert_eq!(started.load(Ordering::SeqCst), 2); + for (offset, item) in spawned.iter().enumerate() { + let index = i64::try_from(offset + 1) + .map_err(|err| Error::message(format!("invalid task index: {err}")))?; + let task = fetch_task(&queue, item.task_id).await?; + if started_indices.contains(&index) { + assert_eq!(task.state, "running"); + } else { + assert_eq!(task.state, "pending"); + } + } + + release.store(true, Ordering::SeqCst); + release_notify.notify_waiters(); + + tokio::time::timeout(Duration::from_secs(5), started_rx.recv()) + .await + .map_err(|_| Error::message("worker did not start the third task after capacity freed"))? + .ok_or_else(|| Error::message("started channel closed"))?; + + worker.close().await?; + + for item in spawned { + assert_eq!(fetch_task(&queue, item.task_id).await?.state, "completed"); + } + + client.drop_queue().await?; + Ok(()) +} + #[tokio::test] #[ignore = "requires a Postgres database initialized with Absurd SQL"] async fn worker_on_error_reports_claim_failures() -> Result<()> { @@ -691,6 +941,177 @@ async fn context_can_await_completed_task_result_from_another_queue() -> Result< Ok(()) } +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn context_await_task_result_waits_for_pending_child_in_other_queue() -> Result<()> { + let parent_queue = random_queue(); + let child_queue = random_queue(); + let parent = Client::connect_queue(database_url(), &parent_queue).await?; + let child = Client::connect_queue(database_url(), &child_queue).await?; + parent.create_queue().await?; + child.create_queue().await?; + + let child_task = Task::<(), Value>::new("pending-child-result").queue(&child_queue); + child.register(&child_task, |(), _ctx| async move { + Ok(json!({ "child": "eventually done" })) + })?; + + let (waiting_tx, mut waiting_rx) = tokio::sync::mpsc::unbounded_channel::<()>(); + let parent_task = + Task::::new("parent-waits-for-pending-child").queue(&parent_queue); + parent.register(&parent_task, { + let child_queue = child_queue.clone(); + move |child_id, mut ctx| { + let child_queue = child_queue.clone(); + let waiting_tx = waiting_tx.clone(); + async move { + let _ = waiting_tx.send(()); + let snapshot = ctx + .await_task_result( + child_id, + AwaitTaskResultOptions::new() + .queue(child_queue) + .timeout(Duration::from_secs(5)), + ) + .await?; + Ok(json!({ "snapshot": snapshot })) + } + } + })?; + + let child_spawned = child.spawn(&child_task, (), SpawnOptions::new()).await?; + let parent_spawned = parent + .spawn(&parent_task, child_spawned.task_id, SpawnOptions::new()) + .await?; + + let parent_worker = { + let parent = parent.clone(); + tokio::spawn(async move { parent.work_batch(WorkBatchOptions::new()).await }) + }; + + tokio::time::timeout(Duration::from_secs(5), waiting_rx.recv()) + .await + .map_err(|_| Error::message("parent did not begin waiting for child result"))? + .ok_or_else(|| Error::message("parent wait channel closed"))?; + assert_eq!(child.work_batch(WorkBatchOptions::new()).await?, 1); + assert_eq!( + parent_worker.await.map_err(Error::Join)??, + 1, + "parent task should complete after child result becomes available" + ); + + let task = fetch_task(&parent_queue, parent_spawned.task_id).await?; + assert_eq!(task.state, "completed"); + assert_eq!( + task.completed_payload, + Some(json!({ + "snapshot": { + "state": "completed", + "result": { "child": "eventually done" }, + } + })) + ); + + parent.drop_queue().await?; + child.drop_queue().await?; + Ok(()) +} + +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn context_await_task_result_checkpoint_survives_parent_retry_after_child_cleanup() +-> Result<()> { + let parent_queue = random_queue(); + let child_queue = random_queue(); + let parent = Client::connect_queue(database_url(), &parent_queue).await?; + let (child, child_pool) = connect_queue_with_single_connection_pool(&child_queue).await?; + parent.create_queue().await?; + child.create_queue().await?; + + let base = utc("2024-05-01T08:00:00Z")?; + set_fake_now(&child_pool, Some(base)).await?; + + let child_task = Task::<(), Value>::new("cleanup-child-result").queue(&child_queue); + child.register(&child_task, |(), _ctx| async move { + Ok(json!({ "child": "cached" })) + })?; + let child_spawned = child.spawn(&child_task, (), SpawnOptions::new()).await?; + assert_eq!(child.work_batch(WorkBatchOptions::new()).await?, 1); + + let attempts = Arc::new(AtomicUsize::new(0)); + let parent_task = Task::::new("parent-caches-child-result") + .queue(&parent_queue) + .default_max_attempts(2); + parent.register(&parent_task, { + let child_queue = child_queue.clone(); + let attempts = Arc::clone(&attempts); + move |child_id, mut ctx| { + let child_queue = child_queue.clone(); + let attempts = Arc::clone(&attempts); + async move { + let snapshot = ctx + .await_task_result( + child_id, + AwaitTaskResultOptions::new() + .queue(child_queue) + .timeout(Duration::from_secs(1)), + ) + .await?; + let attempt = attempts.fetch_add(1, Ordering::SeqCst) + 1; + if attempt == 1 { + return Err(Error::message("retry after child result checkpoint")); + } + Ok(json!({ "snapshot": snapshot, "attempt": attempt })) + } + } + })?; + + let parent_spawned = parent + .spawn(&parent_task, child_spawned.task_id, SpawnOptions::new()) + .await?; + assert_eq!(parent.work_batch(WorkBatchOptions::new()).await?, 1); + assert_eq!( + fetch_task(&parent_queue, parent_spawned.task_id) + .await? + .state, + "pending" + ); + assert_eq!( + fetch_checkpoints(&parent_queue, parent_spawned.task_id) + .await? + .len(), + 1 + ); + + set_fake_now(&child_pool, Some(base + ChronoDuration::hours(2))).await?; + let cleanup = child + .cleanup_with_limit(Duration::from_secs(3600), 10, None) + .await?; + assert_eq!(cleanup.tasks_deleted, 1); + assert_eq!( + child.fetch_task_result(child_spawned.task_id, None).await?, + None + ); + + assert_eq!(parent.work_batch(WorkBatchOptions::new()).await?, 1); + let task = fetch_task(&parent_queue, parent_spawned.task_id).await?; + assert_eq!(task.state, "completed"); + assert_eq!( + task.completed_payload, + Some(json!({ + "snapshot": { + "state": "completed", + "result": { "child": "cached" }, + }, + "attempt": 2, + })) + ); + + parent.drop_queue().await?; + child.drop_queue().await?; + Ok(()) +} + #[tokio::test] #[ignore = "requires a Postgres database initialized with Absurd SQL"] async fn decomposed_step_handle_reuses_completed_state_after_retry() -> Result<()> { @@ -785,6 +1206,44 @@ async fn failed_decomposed_step_is_not_checkpointed_and_reexecutes() -> Result<( Ok(()) } +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn repeated_decomposed_step_names_are_numbered() -> Result<()> { + let (queue, client) = test_client().await?; + + let task = Task::<(), Value>::new("manual-loop-steps").queue(&queue); + client.register(&task, |(), mut ctx| async move { + let mut results = Vec::new(); + for i in 0_i64..3 { + let handle = ctx.begin_step::("manual-loop").await?; + let value = ctx.complete_step(handle, i * 5).await?; + results.push(value); + } + Ok(json!({ "results": results })) + })?; + + let spawned = client.spawn(&task, (), SpawnOptions::new()).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, 5, 10] })) + ); + assert_eq!( + fetch_checkpoints(&queue, spawned.task_id).await?, + vec![ + ("manual-loop".to_string(), json!(0)), + ("manual-loop#2".to_string(), json!(5)), + ("manual-loop#3".to_string(), json!(10)), + ] + ); + + 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<()> { @@ -895,6 +1354,99 @@ async fn failed_step_is_not_checkpointed_and_reexecutes() -> Result<()> { Ok(()) } +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn multi_step_only_reexecutes_uncompleted_steps() -> Result<()> { + let (queue, client) = test_client().await?; + let first_calls = Arc::new(AtomicUsize::new(0)); + let second_calls = Arc::new(AtomicUsize::new(0)); + let third_calls = Arc::new(AtomicUsize::new(0)); + + let task = Task::<(), Value>::new("multi-step-retry") + .queue(&queue) + .default_max_attempts(2); + client.register(&task, { + let first_calls = Arc::clone(&first_calls); + let second_calls = Arc::clone(&second_calls); + let third_calls = Arc::clone(&third_calls); + move |(), mut ctx| { + let first_calls = Arc::clone(&first_calls); + let second_calls = Arc::clone(&second_calls); + let third_calls = Arc::clone(&third_calls); + async move { + let first: i64 = ctx + .step("first", || { + let first_calls = Arc::clone(&first_calls); + async move { + first_calls.fetch_add(1, Ordering::SeqCst); + Ok(1) + } + }) + .await?; + let second: i64 = ctx + .step("second", || { + let second_calls = Arc::clone(&second_calls); + async move { + second_calls.fetch_add(1, Ordering::SeqCst); + Ok(2) + } + }) + .await?; + let third: i64 = ctx + .step("third", || { + let third_calls = Arc::clone(&third_calls); + async move { + let call = third_calls.fetch_add(1, Ordering::SeqCst) + 1; + if call == 1 { + Err(Error::message("third step fails once")) + } else { + Ok(3) + } + } + }) + .await?; + Ok(json!({ + "sum": first + second + third, + "first_calls": first_calls.load(Ordering::SeqCst), + "second_calls": second_calls.load(Ordering::SeqCst), + "third_calls": third_calls.load(Ordering::SeqCst), + })) + } + } + })?; + + let spawned = client.spawn(&task, (), SpawnOptions::new()).await?; + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 1); + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 1); + + assert_eq!(first_calls.load(Ordering::SeqCst), 1); + assert_eq!(second_calls.load(Ordering::SeqCst), 1); + assert_eq!(third_calls.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!({ + "sum": 6, + "first_calls": 1, + "second_calls": 1, + "third_calls": 2, + })) + ); + assert_eq!( + fetch_checkpoints(&queue, spawned.task_id).await?, + vec![ + ("first".to_string(), json!(1)), + ("second".to_string(), json!(2)), + ("third".to_string(), json!(3)), + ] + ); + + client.drop_queue().await?; + Ok(()) +} + #[tokio::test] #[ignore = "requires a Postgres database initialized with Absurd SQL"] async fn repeated_step_names_are_numbered() -> Result<()> { @@ -971,13 +1523,45 @@ async fn sleep_until_suspends_then_resumes_from_checkpoint() -> Result<()> { assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 0); - tokio::time::sleep(Duration::from_millis(1_100)).await; + 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!({ "executions": 2 }))); - assert_eq!(executions.load(Ordering::SeqCst), 2); + assert_eq!(task.completed_payload, Some(json!({ "received": payload }))); client.drop_queue().await?; Ok(()) @@ -985,14 +1569,18 @@ async fn sleep_until_suspends_then_resumes_from_checkpoint() -> Result<()> { #[tokio::test] #[ignore = "requires a Postgres database initialized with Absurd SQL"] -async fn pre_emitted_event_is_available_to_late_waiter() -> Result<()> { +async fn event_emit_is_first_write_wins() -> Result<()> { let (queue, client) = test_client().await?; - let event_name = format!("pre_emitted_{queue}"); - let payload = json!({ "data": "ready" }); + let event_name = format!("first_write_{queue}"); - client.emit_event(&event_name, &payload, None).await?; + client + .emit_event(&event_name, &json!({ "version": 1 }), None) + .await?; + client + .emit_event(&event_name, &json!({ "version": 2 }), None) + .await?; - let task = Task::<(), Value>::new("late-waiter").queue(&queue); + let task = Task::<(), Value>::new("first-write-waiter").queue(&queue); client.register(&task, { let event_name = event_name.clone(); move |(), mut ctx| { @@ -1004,12 +1592,15 @@ async fn pre_emitted_event_is_available_to_late_waiter() -> Result<()> { } })?; - let spawned = client.spawn(&task, (), Default::default()).await?; + let spawned = client.spawn(&task, (), SpawnOptions::new()).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 }))); + assert_eq!( + task.completed_payload, + Some(json!({ "received": { "version": 1 } })) + ); client.drop_queue().await?; Ok(()) @@ -1132,6 +1723,48 @@ async fn event_timeout_can_be_caught_without_recreating_wait() -> Result<()> { Ok(()) } +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn cleanup_removes_terminal_tasks_and_events_by_ttl() -> Result<()> { + let (queue, client, pool) = test_client_with_single_connection_pool().await?; + let base = utc("2024-03-01T08:00:00Z")?; + set_fake_now(&pool, Some(base)).await?; + + let task = Task::<(), Value>::new("cleanup-target").queue(&queue); + client.register( + &task, + |(), _ctx| async move { Ok(json!({ "status": "done" })) }, + )?; + + let spawned = client.spawn(&task, (), SpawnOptions::new()).await?; + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 1); + client + .emit_event("cleanup-event", &json!({ "kind": "notify" }), None) + .await?; + + set_fake_now(&pool, Some(base + ChronoDuration::minutes(30))).await?; + let before_ttl = client.cleanup(Duration::from_secs(3600), None).await?; + assert_eq!(before_ttl.tasks_deleted, 0); + assert_eq!(before_ttl.events_deleted, 0); + assert_eq!( + fetch_task(&queue, spawned.task_id).await?.state, + "completed" + ); + assert_eq!(count_events(&queue).await?, 1); + + set_fake_now(&pool, Some(base + ChronoDuration::hours(2))).await?; + let cleanup = client + .cleanup_with_limit(Duration::from_secs(3600), 10, None) + .await?; + assert_eq!(cleanup.tasks_deleted, 1); + assert_eq!(cleanup.events_deleted, 1); + assert_eq!(count_tasks(&queue).await?, 0); + assert_eq!(count_events(&queue).await?, 0); + + client.drop_queue().await?; + Ok(()) +} + #[tokio::test] #[ignore = "requires a Postgres database initialized with Absurd SQL"] async fn idempotent_spawn_creates_one_task_and_executes_once() -> Result<()> { @@ -1415,7 +2048,9 @@ async fn retry_task_can_spawn_new_task_from_original_inputs() -> 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 (queue, client, pool) = test_client_with_single_connection_pool().await?; + let base = utc("2024-05-01T11:00:00Z")?; + set_fake_now(&pool, Some(base)).await?; let attempts = Arc::new(AtomicUsize::new(0)); let task = Task::<(), Value>::new("fixed-retry") @@ -1453,7 +2088,7 @@ async fn fixed_retry_strategy_delays_next_attempt() -> Result<()> { 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; + set_fake_now(&pool, Some(base + ChronoDuration::seconds(1))).await?; assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 1); let task = fetch_task(&queue, spawned.task_id).await?; @@ -1466,6 +2101,218 @@ async fn fixed_retry_strategy_delays_next_attempt() -> Result<()> { Ok(()) } +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn retry_strategy_none_requeues_immediately() -> Result<()> { + let (queue, client, pool) = test_client_with_single_connection_pool().await?; + let base = utc("2024-05-01T12:00:00Z")?; + set_fake_now(&pool, Some(base)).await?; + let attempts = Arc::new(AtomicUsize::new(0)); + + let task = Task::<(), Value>::new("none-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("retry immediately")) + } else { + Ok(json!({ "attempts": attempt })) + } + } + } + })?; + + let spawned = client + .spawn( + &task, + (), + SpawnOptions::new().retry_strategy(RetryStrategy::none()), + ) + .await?; + let metadata = fetch_task_spawn_metadata(&queue, spawned.task_id).await?; + assert_eq!(metadata.retry_strategy, Some(json!({ "kind": "none" }))); + + 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.completed_payload, Some(json!({ "attempts": 2 }))); + + client.drop_queue().await?; + Ok(()) +} + +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn exponential_retry_strategy_delays_attempts() -> Result<()> { + let (queue, client, pool) = test_client_with_single_connection_pool().await?; + let base = utc("2024-05-01T10:00:00Z")?; + set_fake_now(&pool, Some(base)).await?; + let attempts = Arc::new(AtomicUsize::new(0)); + + let task = Task::<(), Value>::new("exponential-retry") + .queue(&queue) + .default_max_attempts(3); + 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 < 3 { + Err(Error::message(format!("fail-{attempt}"))) + } else { + Ok(json!({ "attempts": attempt })) + } + } + } + })?; + + let spawned = client + .spawn( + &task, + (), + SpawnOptions::new().retry_strategy( + RetryStrategy::exponential(Duration::from_secs(1), 3.0) + .with_max(Duration::from_secs(2)), + ), + ) + .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); + + set_fake_now(&pool, Some(base + ChronoDuration::seconds(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, 3); + assert_eq!(attempts.load(Ordering::SeqCst), 2); + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 0); + + set_fake_now(&pool, Some(base + ChronoDuration::seconds(3))).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, 3); + assert_eq!(task.completed_payload, Some(json!({ "attempts": 3 }))); + + client.drop_queue().await?; + Ok(()) +} + +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn cancellation_policy_cancels_by_max_delay() -> Result<()> { + let (queue, client, pool) = test_client_with_single_connection_pool().await?; + let base = utc("2024-05-01T08:00:00Z")?; + set_fake_now(&pool, Some(base)).await?; + + let task = Task::<(), Value>::new("max-delay-cancel").queue(&queue); + client.register(&task, |(), _ctx| async move { + Ok(json!({ "unexpected": true })) + })?; + + let spawned = client + .spawn( + &task, + (), + SpawnOptions::new() + .cancellation(CancellationPolicy::new().max_delay(Duration::from_secs(60))), + ) + .await?; + + set_fake_now(&pool, Some(base + ChronoDuration::seconds(61))).await?; + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 0); + let task = fetch_task(&queue, spawned.task_id).await?; + assert_eq!(task.state, "cancelled"); + assert!(task.cancelled_at.is_some()); + + client.drop_queue().await?; + Ok(()) +} + +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn cancellation_policy_cancels_by_max_duration() -> Result<()> { + let (queue, client, pool) = test_client_with_single_connection_pool().await?; + let base = utc("2024-05-01T09:00:00Z")?; + set_fake_now(&pool, Some(base)).await?; + + let task = Task::<(), Value>::new("max-duration-cancel") + .queue(&queue) + .default_max_attempts(3); + client.register(&task, |(), _ctx| async move { + Err::(Error::message("fail until duration cancellation")) + })?; + + let spawned = client + .spawn( + &task, + (), + SpawnOptions::new() + .retry_strategy(RetryStrategy::fixed(Duration::from_secs(30))) + .cancellation(CancellationPolicy::new().max_duration(Duration::from_secs(90))), + ) + .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!(count_runs(&queue, spawned.task_id).await?, 2); + + set_fake_now(&pool, Some(base + ChronoDuration::seconds(91))).await?; + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 0); + let task = fetch_task(&queue, spawned.task_id).await?; + assert_eq!(task.state, "cancelled"); + assert!(task.cancelled_at.is_some()); + assert_eq!(count_runs(&queue, spawned.task_id).await?, 2); + + client.drop_queue().await?; + Ok(()) +} + +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn default_task_cancellation_is_applied() -> Result<()> { + let (queue, client, pool) = test_client_with_single_connection_pool().await?; + let base = utc("2024-05-01T08:30:00Z")?; + set_fake_now(&pool, Some(base)).await?; + + let task = Task::<(), Value>::new("default-cancel") + .queue(&queue) + .default_cancellation(CancellationPolicy::new().max_delay(Duration::from_secs(60))); + client.register(&task, |(), _ctx| async move { + Ok(json!({ "unexpected": true })) + })?; + + let spawned = client.spawn(&task, (), SpawnOptions::new()).await?; + let metadata = fetch_task_spawn_metadata(&queue, spawned.task_id).await?; + assert_eq!(metadata.cancellation, Some(json!({ "max_delay": 60 }))); + + set_fake_now(&pool, Some(base + ChronoDuration::seconds(61))).await?; + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 0); + let task = fetch_task(&queue, spawned.task_id).await?; + assert_eq!(task.state, "cancelled"); + assert!(task.cancelled_at.is_some()); + + 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<()> { @@ -1585,6 +2432,109 @@ async fn manual_cancel_pending_task_prevents_claim() -> Result<()> { Ok(()) } +#[tokio::test] +#[ignore = "requires a Postgres database initialized with Absurd SQL"] +async fn manual_cancel_running_sleeping_terminal_and_missing_cases() -> Result<()> { + let (queue, client) = test_client().await?; + + let running_started = Arc::new(Notify::new()); + let running_release = Arc::new(Notify::new()); + let running_task = Task::<(), Value>::new("running-cancel").queue(&queue); + client.register(&running_task, { + let running_started = Arc::clone(&running_started); + let running_release = Arc::clone(&running_release); + move |(), _ctx| { + let running_started = Arc::clone(&running_started); + let running_release = Arc::clone(&running_release); + async move { + running_started.notify_one(); + running_release.notified().await; + Ok(json!({ "would_complete": true })) + } + } + })?; + + let running = client.spawn(&running_task, (), SpawnOptions::new()).await?; + let worker = client.start_worker( + WorkerOptions::new() + .poll_interval(Duration::from_millis(25)) + .fatal_on_lease_timeout(false), + )?; + tokio::time::timeout(Duration::from_secs(5), running_started.notified()) + .await + .map_err(|_| Error::message("running cancel task did not start"))?; + client.cancel_task(running.task_id, None).await?; + running_release.notify_one(); + worker.close().await?; + let running_row = fetch_task(&queue, running.task_id).await?; + assert_eq!(running_row.state, "cancelled"); + assert!(running_row.cancelled_at.is_some()); + + let sleeping_event = format!("sleeping_cancel_{queue}"); + let sleeping_task = Task::<(), Value>::new("sleeping-cancel").queue(&queue); + client.register(&sleeping_task, { + let sleeping_event = sleeping_event.clone(); + move |(), mut ctx| { + let sleeping_event = sleeping_event.clone(); + async move { + let payload: Value = ctx.await_event(&sleeping_event).await?; + Ok(json!({ "payload": payload })) + } + } + })?; + let sleeping = client + .spawn(&sleeping_task, (), SpawnOptions::new()) + .await?; + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 1); + assert_eq!( + fetch_task(&queue, sleeping.task_id).await?.state, + "sleeping" + ); + client.cancel_task(sleeping.task_id, None).await?; + let sleeping_row = fetch_task(&queue, sleeping.task_id).await?; + assert_eq!(sleeping_row.state, "cancelled"); + assert!(sleeping_row.cancelled_at.is_some()); + client + .emit_event(&sleeping_event, &json!({ "ignored": true }), None) + .await?; + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 0); + + let completed_task = Task::<(), Value>::new("completed-cancel-noop").queue(&queue); + client.register(&completed_task, |(), _ctx| async move { + Ok(json!({ "done": true })) + })?; + let completed = client + .spawn(&completed_task, (), SpawnOptions::new()) + .await?; + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 1); + client.cancel_task(completed.task_id, None).await?; + let completed_row = fetch_task(&queue, completed.task_id).await?; + assert_eq!(completed_row.state, "completed"); + assert!(completed_row.cancelled_at.is_none()); + + let failed_task = Task::<(), Value>::new("failed-cancel-noop") + .queue(&queue) + .default_max_attempts(1); + client.register(&failed_task, |(), _ctx| async move { + Err::(Error::message("terminal failure")) + })?; + let failed = client.spawn(&failed_task, (), SpawnOptions::new()).await?; + assert_eq!(client.work_batch(WorkBatchOptions::new()).await?, 1); + client.cancel_task(failed.task_id, None).await?; + let failed_row = fetch_task(&queue, failed.task_id).await?; + assert_eq!(failed_row.state, "failed"); + assert!(failed_row.cancelled_at.is_none()); + + let missing_task_id = uuid::Uuid::new_v4(); + let missing = client.cancel_task(missing_task_id, None).await; + assert!( + matches!(missing, Err(Error::Database(err)) if format!("{err:?}").contains("not found")) + ); + + client.drop_queue().await?; + Ok(()) +} + #[tokio::test] #[ignore = "requires a Postgres database initialized with Absurd SQL"] async fn cancelled_terminal_race_does_not_record_task_failure() -> Result<()> { @@ -1687,6 +2637,17 @@ async fn fetch_task(queue: &str, task_id: uuid::Uuid) -> Result { }) } +async fn fetch_task_spawn_metadata(queue: &str, task_id: uuid::Uuid) -> Result { + let pg = pg_client().await?; + let query = + format!("SELECT retry_strategy, cancellation FROM absurd.t_{queue} WHERE task_id = $1"); + let row = pg.query_one(&query, &[&task_id]).await?; + Ok(TaskSpawnMetadata { + retry_strategy: row.get(0), + cancellation: row.get(1), + }) +} + async fn fetch_run(queue: &str, run_id: uuid::Uuid) -> Result { let pg = pg_client().await?; let query = format!( @@ -1726,6 +2687,12 @@ async fn count_tasks(queue: &str) -> Result { Ok(pg.query_one(&query, &[]).await?.get(0)) } +async fn count_events(queue: &str) -> Result { + let pg = pg_client().await?; + let query = format!("SELECT count(*) FROM absurd.e_{queue}"); + Ok(pg.query_one(&query, &[]).await?.get(0)) +} + async fn count_runs(queue: &str, task_id: uuid::Uuid) -> Result { let pg = pg_client().await?; let query = format!("SELECT count(*) FROM absurd.r_{queue} WHERE task_id = $1");