use dbos::sysdb::postgres::{Config, PostgresSystemDatabase, Settings};
use dbos::sysdb::retry::RetryPolicy;
use dbos::sysdb::types::{
Applications, AwaitedOutcome, Change, Debounce, DebounceRequest, EncodedValue, Fork,
ForkOptions, ForkPoint, GetEventCaller, InitWorkflowCaller, Message, NewQueue, NewSchedule,
NewWorkflow, OnExistingQueue, Outcome, OutcomeWrite, QueueRecord, QueueUpdate, RateLimit,
RenameBatching, RenameFrom, ScheduleFilter, ScheduleStatus, ScheduleUpdate, StepTiming,
Submission, Timestamp, WorkflowDelay, WorkflowFilter, WorkflowRecord, WorkflowStatus,
WrittenBy,
};
use dbos::sysdb::{BackendErrorKind, Error, INTERNAL_QUEUE, SystemDatabase};
use dbos_test_support as support;
use dbos_test_support::{Backend, test_database};
fn workflow(id: &str) -> NewWorkflow<'_> {
NewWorkflow {
name: Some("checkout"),
input: Some(r#"{"positionalArgs":[1],"namedArgs":{}}"#),
serialization: Some("portable_json"),
executor_id: Some("local"),
application_version: Some("v1"),
..NewWorkflow::new(id)
}
}
async fn sysdb() -> (PostgresSystemDatabase, support::TestDatabase) {
let db = test_database().await;
let pool = db.pool().await;
(
PostgresSystemDatabase::from_pool(pool, &Settings::default()),
db,
)
}
#[tokio::test]
async fn a_workflow_round_trips() {
let (sys, _db) = sysdb().await;
let written = sys
.init_workflow(&workflow("wf-1"), None, Submission::Fresh, None)
.await
.expect("insert failed");
assert_eq!(written.status, WorkflowStatus::Pending);
assert_eq!(
written.recovery_attempts, 1,
"starting a workflow counts as an attempt; enqueueing would not",
);
assert!(written.should_execute);
let read = sys
.get_workflow("wf-1")
.await
.expect("read failed")
.expect("the workflow should exist");
assert_eq!(read.status, WorkflowStatus::Pending);
assert_eq!(read.name.as_deref(), Some("checkout"));
assert_eq!(read.serialization.as_deref(), Some("portable_json"));
}
#[tokio::test]
async fn payloads_are_stored_verbatim() {
let (sys, _db) = sysdb().await;
let mut r = workflow("wf-opaque");
r.input = Some("not json at all \u{1F600} '\"; --");
sys.init_workflow(&r, None, Submission::Fresh, None)
.await
.expect("insert failed");
let read = sys.get_workflow("wf-opaque").await.unwrap().unwrap();
assert_eq!(read.input.as_deref(), r.input);
}
#[tokio::test]
async fn a_missing_workflow_reads_as_none() {
let (sys, _db) = sysdb().await;
assert!(sys.get_workflow("nobody-here").await.unwrap().is_none());
}
#[tokio::test]
async fn resubmitting_reconciles_and_a_different_function_is_rejected() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-dup"), None, Submission::Fresh, None)
.await
.unwrap();
let mut different = workflow("wf-dup");
different.name = Some("a-different-function");
let err = sys
.init_workflow(&different, None, Submission::Fresh, None)
.await
.expect_err("a different function under the same id should be rejected");
assert!(
matches!(err, Error::ConflictingWorkflow { .. }),
"expected a conflict, got {err:?}",
);
let again = sys
.init_workflow(&workflow("wf-dup"), None, Submission::Fresh, None)
.await
.expect("re-submitting the same workflow should succeed");
assert_eq!(again.status, WorkflowStatus::Pending);
let read = sys.get_workflow("wf-dup").await.unwrap().unwrap();
assert_eq!(read.application_version.as_deref(), Some("v1"));
}
#[tokio::test]
async fn only_recoveries_and_dequeues_count_as_attempts() {
let (sys, _db) = sysdb().await;
let r = workflow("wf-attempts");
let first = sys
.init_workflow(&r, None, Submission::Fresh, None)
.await
.unwrap();
assert_eq!(first.recovery_attempts, 1);
let plain = sys
.init_workflow(&r, None, Submission::Fresh, None)
.await
.unwrap();
assert_eq!(
plain.recovery_attempts, 1,
"a re-submission is not an attempt"
);
let recovered = sys
.init_workflow(&r, None, Submission::Recovery, None)
.await
.unwrap();
assert_eq!(recovered.recovery_attempts, 2, "a recovery is an attempt");
}
#[tokio::test]
async fn the_initial_status_is_derived_from_the_queue_and_delay() {
let (sys, _db) = sysdb().await;
let cases = [
(None, None, WorkflowStatus::Pending),
(Some("orders"), None, WorkflowStatus::Enqueued),
(
Some("orders"),
Some(std::time::Duration::from_secs(5)),
WorkflowStatus::Delayed,
),
(
None,
Some(std::time::Duration::from_secs(5)),
WorkflowStatus::Pending,
),
];
for (index, (queue, delay, expected)) in cases.into_iter().enumerate() {
let id = format!("wf-status-{index}");
let wf = NewWorkflow {
queue_name: queue,
delay,
..workflow(&id)
};
assert_eq!(
wf.initial_status(),
expected,
"case {index} derived wrongly"
);
let init = sys
.init_workflow(&wf, None, Submission::Fresh, None)
.await
.unwrap();
assert_eq!(init.status, expected, "case {index} stored wrongly");
}
}
#[tokio::test]
async fn queued_workflows_do_not_accrue_attempts() {
let (sys, _db) = sysdb().await;
let mut r = workflow("wf-queued");
r.queue_name = Some("orders");
let first = sys
.init_workflow(&r, None, Submission::Fresh, None)
.await
.unwrap();
assert_eq!(first.recovery_attempts, 0, "enqueueing is not an attempt");
let again = sys
.init_workflow(&r, None, Submission::Recovery, None)
.await
.unwrap();
assert_eq!(
again.recovery_attempts, 0,
"a queued row stays at zero even when forced",
);
}
#[tokio::test]
async fn exceeding_the_recovery_limit_parks_the_workflow() {
let (sys, _db) = sysdb().await;
let r = workflow("wf-dlq");
sys.init_workflow(&r, None, Submission::Fresh, None)
.await
.unwrap();
let mut last = Ok(());
for _ in 0..5 {
last = sys
.init_workflow(&r, Some(2), Submission::Recovery, None)
.await
.map(|_| ());
if last.is_err() {
break;
}
}
let err = last.expect_err("the workflow should have been parked");
assert!(
matches!(err, Error::MaxRecoveryAttemptsExceeded { limit: 2, .. }),
"expected a dead-letter error, got {err:?}",
);
let read = sys.get_workflow("wf-dlq").await.unwrap().unwrap();
assert_eq!(read.status, WorkflowStatus::MaxRecoveryAttemptsExceeded);
assert_eq!(read.queue_name, None, "parking clears the queue assignment");
}
#[tokio::test]
async fn the_dequeue_claim_counts_an_attempt() {
let (sys, _db) = sysdb().await;
sys.upsert_queue(&NewQueue::new("counted"), OnExistingQueue::Update)
.await
.unwrap();
let enqueued = sys
.init_workflow(
&NewWorkflow {
queue_name: Some("counted"),
..workflow("wf-counted")
},
None,
Submission::Fresh,
None,
)
.await
.expect("enqueue failed");
assert_eq!(
enqueued.recovery_attempts, 0,
"enqueueing is not an attempt: nothing has run yet",
);
let queue = sys.get_queue("counted").await.unwrap().unwrap();
let claimed = sys
.start_queued_workflows(&queue, "local", "v1", None, 0, 0)
.await
.unwrap();
assert_eq!(claimed, ["wf-counted"]);
let row = sys.get_workflow("wf-counted").await.unwrap().unwrap();
assert_eq!(row.status, WorkflowStatus::Pending);
assert_eq!(
row.recovery_attempts, 1,
"the claim that starts the workflow is the attempt",
);
}
#[tokio::test]
async fn a_claim_that_wins_nothing_counts_nothing() {
let (sys, _db) = sysdb().await;
sys.upsert_queue(&NewQueue::new("raced"), OnExistingQueue::Update)
.await
.unwrap();
sys.init_workflow(
&NewWorkflow {
queue_name: Some("raced"),
..workflow("wf-raced")
},
None,
Submission::Fresh,
None,
)
.await
.unwrap();
let queue = sys.get_queue("raced").await.unwrap().unwrap();
let winner = sys
.start_queued_workflows(&queue, "winner", "v1", None, 0, 0)
.await
.unwrap();
assert_eq!(winner, ["wf-raced"]);
let loser = sys
.start_queued_workflows(&queue, "loser", "v1", None, 0, 0)
.await
.unwrap();
assert!(loser.is_empty(), "the second claim should have won nothing");
let row = sys.get_workflow("wf-raced").await.unwrap().unwrap();
assert_eq!(
row.recovery_attempts, 1,
"a claim that ran nothing must not spend the budget",
);
}
#[tokio::test]
async fn repeated_recoveries_through_the_queue_park_the_workflow() {
let (sys, _db) = sysdb().await;
sys.upsert_queue(&NewQueue::new("parking"), OnExistingQueue::Update)
.await
.unwrap();
let queued = NewWorkflow {
queue_name: Some("parking"),
..workflow("wf-parked")
};
sys.init_workflow(&queued, None, Submission::Fresh, None)
.await
.unwrap();
let queue = sys.get_queue("parking").await.unwrap().unwrap();
let mut last = Ok(());
for _ in 0..5 {
let claimed = sys
.start_queued_workflows(&queue, "local", "v1", None, 0, 0)
.await
.unwrap();
if claimed.is_empty() {
break;
}
last = sys
.init_workflow(&queued, Some(1), Submission::Dequeue, None)
.await
.map(|_| ());
if last.is_err() {
break;
}
sys.reenqueue_for_recovery(&["local"], "v1", INTERNAL_QUEUE)
.await
.unwrap();
}
let err = last.expect_err("the workflow should have been parked");
assert!(
matches!(err, Error::MaxRecoveryAttemptsExceeded { limit: 1, .. }),
"expected a dead-letter error, got {err:?}",
);
let read = sys.get_workflow("wf-parked").await.unwrap().unwrap();
assert_eq!(read.status, WorkflowStatus::MaxRecoveryAttemptsExceeded);
}
#[tokio::test]
async fn a_second_owner_records_but_does_not_execute() {
let (sys, _db) = sysdb().await;
let r = workflow("wf-owned");
let first = sys
.init_workflow(&r, None, Submission::Fresh, None)
.await
.unwrap();
assert!(first.should_execute);
let second = sys
.init_workflow(&r, None, Submission::Fresh, None)
.await
.unwrap();
assert!(
!second.should_execute,
"another owner holds this workflow, so running it would be a second execution",
);
let recovering = sys
.init_workflow(&r, None, Submission::Recovery, None)
.await
.unwrap();
assert!(recovering.should_execute, "a recovery may claim the row");
}
#[tokio::test]
async fn the_stored_serialization_is_reported_back() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-fmt"), None, Submission::Fresh, None)
.await
.unwrap();
let mut later = workflow("wf-fmt");
later.serialization = Some("rust_serde");
let outcome = sys
.init_workflow(&later, None, Submission::Fresh, None)
.await
.unwrap();
assert_eq!(
outcome.serialization.as_deref(),
Some("portable_json"),
"the stored format wins; the payloads already there are in it",
);
}
#[tokio::test]
async fn an_owner_is_generated_when_none_is_supplied() {
let (sys, _db) = sysdb().await;
let r = workflow("wf-auto-owner");
let first = sys
.init_workflow(&r, None, Submission::Fresh, None)
.await
.unwrap();
assert!(first.should_execute, "the creator owns what it created");
let second = sys
.init_workflow(&r, None, Submission::Fresh, None)
.await
.unwrap();
assert!(
!second.should_execute,
"a fresh attempt with a new identity must not claim a workflow someone else holds",
);
}
#[tokio::test]
async fn every_settable_field_round_trips() {
let (sys, _db) = sysdb().await;
let written = NewWorkflow {
class_name: Some("Checkout"),
config_name: Some("primary"),
queue_name: Some("orders"),
deduplication_id: Some("dedup-key"),
priority: 7,
queue_partition_key: Some("eu-west"),
delay: Some(std::time::Duration::from_secs(60)),
is_debounced: true,
debounce_deadline: Some(Timestamp::from_epoch_ms(1_700_000_020_000)),
timeout: Some(std::time::Duration::from_secs(30)),
deadline: Some(Timestamp::from_epoch_ms(1_700_000_030_000)),
application_id: Some("app-1"),
authenticated_user: Some("alice"),
authenticated_roles: vec!["admin", "auditor"],
assumed_role: Some("admin"),
schedule_name: Some("nightly"),
attributes: Some(r#"{"tenant": "acme"}"#),
..workflow("wf-all-fields")
};
sys.init_workflow(&workflow("wf-parent"), None, Submission::Fresh, None)
.await
.expect("the parent could not be created");
let before = Timestamp::now();
let init = sys
.init_workflow(
&written,
None,
Submission::Fresh,
Some(InitWorkflowCaller {
parent_workflow_id: "wf-parent",
step_id: 0,
step_name: "checkout",
started_at: Timestamp::now(),
}),
)
.await
.expect("insert failed");
assert_eq!(init.status, WorkflowStatus::Delayed);
assert_eq!(init.deadline, written.deadline);
let read = sys.get_workflow("wf-all-fields").await.unwrap().unwrap();
assert_eq!(read.workflow_id, "wf-all-fields");
assert_eq!(read.name.as_deref(), written.name);
assert_eq!(read.class_name.as_deref(), written.class_name);
assert_eq!(read.config_name.as_deref(), written.config_name);
assert_eq!(read.input.as_deref(), written.input);
assert_eq!(read.serialization.as_deref(), written.serialization);
assert_eq!(read.queue_name.as_deref(), written.queue_name);
assert_eq!(read.deduplication_id.as_deref(), written.deduplication_id);
assert_eq!(read.priority, written.priority);
assert_eq!(
read.queue_partition_key.as_deref(),
written.queue_partition_key
);
assert!(read.is_debounced);
assert_eq!(read.debounce_deadline, written.debounce_deadline);
assert_eq!(read.timeout, written.timeout);
assert_eq!(read.deadline, written.deadline);
assert_eq!(read.executor_id.as_deref(), written.executor_id);
assert_eq!(
read.application_version.as_deref(),
written.application_version
);
assert_eq!(read.application_id.as_deref(), written.application_id);
assert_eq!(
read.authenticated_user.as_deref(),
written.authenticated_user
);
assert_eq!(read.authenticated_roles, written.authenticated_roles);
assert_eq!(read.assumed_role.as_deref(), written.assumed_role);
assert_eq!(read.parent_workflow_id.as_deref(), Some("wf-parent"));
assert_eq!(read.schedule_name.as_deref(), written.schedule_name);
assert!(
read.attributes
.as_deref()
.unwrap_or_default()
.contains("acme"),
"attributes should survive, got {:?}",
read.attributes,
);
let delay_until = read.delay_until.expect("a delay should have been stamped");
let offset = delay_until.as_epoch_ms() - before.as_epoch_ms();
assert!(
(60_000..70_000).contains(&offset),
"delay_until should be ~60s from now, was {offset}ms",
);
assert_eq!(read.output, None);
assert_eq!(read.error, None);
assert_eq!(read.started_at, None);
assert_eq!(read.completed_at, None);
assert_eq!(read.forked_from, None);
assert!(!read.was_forked_from);
assert!(!read.rate_limited);
assert!(
read.owner_xid.is_some(),
"an owner should have been stamped"
);
assert!(read.created_at.as_epoch_ms() >= before.as_epoch_ms());
assert_eq!(read.created_at, read.updated_at);
}
#[tokio::test]
async fn only_the_first_outcome_is_recorded() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-race"), None, Submission::Fresh, None)
.await
.unwrap();
let first = sys
.record_workflow_outcome("wf-race", Outcome::Output(Some("\"winner\"")))
.await
.expect("the first write should succeed");
assert_eq!(first, OutcomeWrite::Recorded);
let second = sys
.record_workflow_outcome("wf-race", Outcome::Error("\"loser\""))
.await
.expect("the second write should not error");
assert_eq!(
second,
OutcomeWrite::AlreadyFinished,
"the loser should learn it lost",
);
let read = sys.get_workflow("wf-race").await.unwrap().unwrap();
assert_eq!(read.status, WorkflowStatus::Success);
assert_eq!(read.output.as_deref(), Some("\"winner\""));
assert_eq!(read.error, None, "the loser's error must not have landed");
assert!(read.completed_at.is_some(), "completion should be stamped");
}
#[tokio::test]
async fn recording_an_outcome_for_a_missing_workflow_is_not_an_error() {
let (sys, _db) = sysdb().await;
let result = sys
.record_workflow_outcome("never-existed", Outcome::Output(None))
.await
.expect("a missing row should not be an error");
assert_eq!(result, OutcomeWrite::AlreadyFinished);
}
const BRISK_POLL: std::time::Duration = std::time::Duration::from_millis(50);
const BRIEFLY: std::time::Duration = std::time::Duration::from_millis(300);
#[tokio::test]
async fn an_await_returns_a_recorded_outcome() {
let (sys, _db) = sysdb().await;
for id in ["wf-ok", "wf-bad"] {
sys.init_workflow(&workflow(id), None, Submission::Fresh, None)
.await
.unwrap();
}
sys.record_workflow_outcome("wf-ok", Outcome::Output(Some("\"done\"")))
.await
.unwrap();
assert_eq!(
sys.await_workflow_result("wf-ok", BRISK_POLL, false)
.await
.unwrap(),
AwaitedOutcome::Succeeded {
output: Some("\"done\"".to_owned()),
serialization: Some("portable_json".to_owned()),
}
);
sys.record_workflow_outcome("wf-bad", Outcome::Error("\"boom\""))
.await
.unwrap();
assert_eq!(
sys.await_workflow_result("wf-bad", BRISK_POLL, false)
.await
.unwrap(),
AwaitedOutcome::Failed {
error: "\"boom\"".to_owned(),
serialization: Some("portable_json".to_owned()),
}
);
}
#[tokio::test]
async fn an_await_reports_a_void_return_as_a_success() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-void"), None, Submission::Fresh, None)
.await
.unwrap();
sys.record_workflow_outcome("wf-void", Outcome::Output(None))
.await
.unwrap();
assert!(matches!(
sys.await_workflow_result("wf-void", BRISK_POLL, false)
.await
.unwrap(),
AwaitedOutcome::Succeeded { output: None, .. }
));
}
#[tokio::test]
async fn an_await_waits_for_a_workflow_that_has_not_finished() {
let (sys, db) = sysdb().await;
sys.init_workflow(&workflow("wf-slow"), None, Submission::Fresh, None)
.await
.unwrap();
let finisher = PostgresSystemDatabase::from_pool(db.pool().await, &Settings::default());
let finishing = tokio::spawn(async move {
tokio::time::sleep(std::time::Duration::from_millis(400)).await;
finisher
.record_workflow_outcome("wf-slow", Outcome::Output(Some("\"late\"")))
.await
.unwrap();
});
let started = std::time::Instant::now();
let settled = sys
.await_workflow_result("wf-slow", BRISK_POLL, false)
.await
.unwrap();
let waited = started.elapsed();
finishing.await.unwrap();
assert_eq!(
settled,
AwaitedOutcome::Succeeded {
output: Some("\"late\"".to_owned()),
serialization: Some("portable_json".to_owned()),
}
);
assert!(
waited >= BRIEFLY,
"it cannot have read an outcome that had not been recorded"
);
assert!(
waited < std::time::Duration::from_secs(5),
"waited {waited:?}"
);
}
#[tokio::test]
async fn a_waiter_that_never_finishes_does_not_starve_the_others() {
let db = test_database().await;
let sys = std::sync::Arc::new(PostgresSystemDatabase::from_pool(
db.pool().await,
&Settings {
polling_concurrency: Some(1),
..Settings::default()
},
));
const OTHERS: usize = 4;
let ids: Vec<String> = (0..OTHERS).map(|i| format!("wf-capped-{i}")).collect();
for id in std::iter::once(&"wf-never".to_owned()).chain(ids.iter()) {
sys.init_workflow(&workflow(id), None, Submission::Fresh, None)
.await
.unwrap();
}
let forever = {
let sys = std::sync::Arc::clone(&sys);
tokio::spawn(async move {
sys.await_workflow_result("wf-never", BRISK_POLL, false)
.await
})
};
tokio::time::sleep(BRIEFLY).await;
let waiting: Vec<_> = ids
.iter()
.map(|id| {
let (sys, id) = (std::sync::Arc::clone(&sys), id.clone());
tokio::spawn(async move { sys.await_workflow_result(&id, BRISK_POLL, false).await })
})
.collect();
tokio::time::sleep(BRIEFLY).await;
let finisher = PostgresSystemDatabase::from_pool(db.pool().await, &Settings::default());
for id in &ids {
finisher
.record_workflow_outcome(id, Outcome::Output(Some("\"done\"")))
.await
.unwrap();
}
for (id, handle) in ids.iter().zip(waiting) {
let settled = tokio::time::timeout(std::time::Duration::from_secs(20), handle)
.await
.unwrap_or_else(|_| panic!("{id} starved: a permit is being held across a wait"))
.unwrap()
.unwrap();
assert_eq!(
settled,
AwaitedOutcome::Succeeded {
output: Some("\"done\"".to_owned()),
serialization: Some("portable_json".to_owned()),
}
);
}
assert!(!forever.is_finished());
forever.abort();
}
#[tokio::test]
async fn a_wait_on_a_closed_handle_reports_rather_than_hanging() {
let db = test_database().await;
let sys = std::sync::Arc::new(PostgresSystemDatabase::from_pool(
db.pool().await,
&Settings {
polling_concurrency: Some(1),
..Settings::default()
},
));
sys.init_workflow(&workflow("wf-closing"), None, Submission::Fresh, None)
.await
.unwrap();
let waiting = {
let sys = std::sync::Arc::clone(&sys);
tokio::spawn(async move {
sys.await_workflow_result("wf-closing", BRISK_POLL, false)
.await
.map(|_| ())
})
};
tokio::time::sleep(BRIEFLY).await;
sys.close().await;
let ended = tokio::time::timeout(std::time::Duration::from_secs(20), waiting)
.await
.expect("closing the handle left a waiter parked")
.unwrap();
assert!(
ended.is_err(),
"a wait on a closed handle must report rather than succeed"
);
}
#[tokio::test]
async fn an_await_reports_a_cancelled_workflow_as_cancelled() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-doomed"), None, Submission::Fresh, None)
.await
.unwrap();
sys.cancel_workflows(&["wf-doomed"], false, None)
.await
.unwrap();
assert_eq!(
sys.await_workflow_result("wf-doomed", BRISK_POLL, false)
.await
.unwrap(),
AwaitedOutcome::Cancelled
);
}
#[tokio::test]
async fn an_await_reports_a_parked_workflow_rather_than_waiting_for_it() {
let (sys, _db) = sysdb().await;
let parked = workflow("wf-parked");
sys.init_workflow(&parked, None, Submission::Fresh, None)
.await
.unwrap();
for _ in 0..5 {
if sys
.init_workflow(&parked, Some(2), Submission::Recovery, None)
.await
.is_err()
{
break;
}
}
let settled = sys
.await_workflow_result("wf-parked", BRISK_POLL, false)
.await
.unwrap();
let AwaitedOutcome::Parked { recovery_attempts } = settled else {
panic!("expected a parked workflow, got {settled:?}");
};
assert!(
recovery_attempts > 2,
"the count that passed the limit, got {recovery_attempts}"
);
}
#[tokio::test]
async fn an_absent_row_is_waited_through_unless_the_caller_says_otherwise() {
let (sys, db) = sysdb().await;
let err = sys
.await_workflow_result("never-existed", BRISK_POLL, true)
.await
.expect_err("a caller that has seen the row is told it is gone");
assert!(
matches!(err, Error::NonExistentWorkflow { ref workflow_ids } if workflow_ids == &["never-existed"]),
"got {err:?}"
);
let inserting = tokio::spawn({
let sys = PostgresSystemDatabase::from_pool(db.pool().await, &Settings::default());
async move {
tokio::time::sleep(BRIEFLY).await;
sys.init_workflow(&workflow("wf-later"), None, Submission::Fresh, None)
.await
.unwrap();
sys.record_workflow_outcome("wf-later", Outcome::Output(Some("\"arrived\"")))
.await
.unwrap();
}
});
let settled = sys
.await_workflow_result("wf-later", BRISK_POLL, false)
.await
.expect("the row appeared while this waited");
inserting.await.unwrap();
assert_eq!(
settled,
AwaitedOutcome::Succeeded {
output: Some("\"arrived\"".to_owned()),
serialization: Some("portable_json".to_owned()),
}
);
}
#[tokio::test]
async fn the_trait_is_object_safe() {
let (sys, _db) = sysdb().await;
let dynamic: &dyn SystemDatabase = &sys;
dynamic
.init_workflow(&workflow("wf-dyn"), None, Submission::Fresh, None)
.await
.unwrap();
assert!(dynamic.get_workflow("wf-dyn").await.unwrap().is_some());
}
#[tokio::test]
async fn a_rejected_statement_is_not_retried() {
let db = test_database().await;
let pool = db.pool().await;
let sys = PostgresSystemDatabase::from_pool(
pool,
&Settings {
schema: "no_such_schema",
..Settings::default()
},
);
let result = tokio::time::timeout(
std::time::Duration::from_millis(500),
sys.get_workflow("wf-1"),
)
.await
.expect("a permanent failure was retried instead of being returned");
match result {
Err(Error::Backend(e)) => {
assert_eq!(e.kind, BackendErrorKind::Permanent);
assert_eq!(
e.sqlstate.as_deref(),
Some("42P01"),
"the SQLSTATE should survive onto the error, got {e:?}",
);
}
other => panic!("expected a permanent backend error, got {other:?}"),
}
}
#[tokio::test]
async fn an_unreachable_database_is_a_connection_failure() {
let db = test_database().await;
let unreachable = sqlx::pool::PoolOptions::new()
.max_connections(1)
.acquire_timeout(std::time::Duration::from_secs(1))
.connect_lazy_with(db.options().port(1));
let sys = PostgresSystemDatabase::from_pool(
unreachable,
&Settings {
retry: RetryPolicy {
retry_connection_errors: false,
..RetryPolicy::default()
},
..Settings::default()
},
);
let result = tokio::time::timeout(std::time::Duration::from_secs(10), sys.get_workflow("wf-1"))
.await
.expect("the opt-out should have returned rather than retried");
match result {
Err(Error::Backend(e)) => assert_eq!(
e.kind,
BackendErrorKind::Connection,
"an unreachable database may yet answer, so it is worth waiting for, got {e:?}",
),
other => panic!("expected a connection error, got {other:?}"),
}
}
#[tokio::test]
async fn a_killed_connection_is_waited_out() {
const TAG: &str = "chaos-waited-out";
let db = test_database().await;
let pool = db
.pool_options()
.test_before_acquire(false)
.connect_with(db.options().application_name(TAG))
.await
.expect("failed to connect");
let sys = PostgresSystemDatabase::from_pool(
pool,
&Settings {
retry: RetryPolicy {
initial_backoff: std::time::Duration::from_millis(50),
max_backoff: std::time::Duration::from_millis(200),
..RetryPolicy::default()
},
..Settings::default()
},
);
let wf = workflow("wf-chaos");
sys.init_workflow(&wf, None, Submission::Fresh, None)
.await
.expect("insert failed");
db.kill_connections(TAG).await;
let read = tokio::time::timeout(
std::time::Duration::from_secs(20),
sys.get_workflow("wf-chaos"),
)
.await
.expect("the retry never recovered")
.expect("the read should have succeeded on a later attempt");
assert_eq!(
read.expect("the workflow should still exist").workflow_id,
"wf-chaos",
"the row is durable; only the connection was lost",
);
}
#[tokio::test]
async fn the_opt_out_surfaces_a_killed_connection() {
const TAG: &str = "chaos-opt-out";
let db = test_database().await;
let pool = db
.pool_options()
.test_before_acquire(false)
.connect_with(db.options().application_name(TAG))
.await
.expect("failed to connect");
let sys = PostgresSystemDatabase::from_pool(
pool,
&Settings {
retry: RetryPolicy {
retry_connection_errors: false,
..RetryPolicy::default()
},
..Settings::default()
},
);
let wf = workflow("wf-chaos-optout");
sys.init_workflow(&wf, None, Submission::Fresh, None)
.await
.expect("insert failed");
db.kill_connections(TAG).await;
match sys.get_workflow("wf-chaos-optout").await {
Err(Error::Backend(e)) => assert_eq!(
e.kind,
BackendErrorKind::Connection,
"a terminated backend should classify as a connection failure, got {e:?}",
),
other => panic!("the connection kill did not land; got {other:?}"),
}
}
async fn seeded() -> (PostgresSystemDatabase, support::TestDatabase) {
let (sys, db) = sysdb().await;
let seeds = [
NewWorkflow {
name: Some("checkout"),
class_name: Some("Checkout"),
config_name: Some("primary"),
application_version: Some("v1"),
executor_id: Some("alpha"),
authenticated_user: Some("alice"),
attributes: Some(r#"{"tenant": "acme", "tier": "gold"}"#),
..NewWorkflow::new("wf-a")
},
NewWorkflow {
name: Some("refund"),
application_version: Some("v2"),
executor_id: Some("beta"),
authenticated_user: Some("bob"),
queue_name: Some("orders"),
deduplication_id: Some("dedup-b"),
schedule_name: Some("nightly"),
attributes: Some(r#"{"tenant": "globex"}"#),
..NewWorkflow::new("wf-b")
},
NewWorkflow {
name: Some("checkout"),
queue_name: Some("orders"),
delay: Some(std::time::Duration::from_secs(300)),
is_debounced: true,
deduplication_id: Some("dedup-c"),
..NewWorkflow::new("wf-c")
},
NewWorkflow {
name: Some("audit"),
queue_name: Some("reports"),
..NewWorkflow::new("wf-d")
},
NewWorkflow {
name: Some("audit"),
..NewWorkflow::new("other-100%-done")
},
];
let children_of_a = [("wf-b", 0), ("wf-d", 1)];
for wf in &seeds {
let caller = children_of_a
.iter()
.find(|(id, _)| *id == wf.workflow_id)
.map(|(_, step_id)| InitWorkflowCaller {
parent_workflow_id: "wf-a",
step_id: *step_id,
step_name: wf.name.unwrap_or_default(),
started_at: Timestamp::now(),
});
sys.init_workflow(wf, None, Submission::Fresh, caller)
.await
.expect("seed insert failed");
tokio::time::sleep(std::time::Duration::from_millis(2)).await;
}
let mut conn = db.admin_connection().await;
for (sql, id) in [
(
"UPDATE dbos.workflow_status SET forked_from = 'wf-a', was_forked_from = false WHERE workflow_uuid = $1",
"wf-c",
),
(
"UPDATE dbos.workflow_status SET was_forked_from = true WHERE workflow_uuid = $1",
"wf-a",
),
(
"UPDATE dbos.workflow_status SET completed_at = 5000, started_at_epoch_ms = 4000 WHERE workflow_uuid = $1",
"wf-b",
),
(
"UPDATE dbos.workflow_status SET completed_at = 9000, started_at_epoch_ms = 8000 WHERE workflow_uuid = $1",
"wf-d",
),
] {
sqlx::query(sqlx::AssertSqlSafe(sql))
.bind(id)
.execute(&mut conn)
.await
.expect("seed update failed");
}
(sys, db)
}
async fn ids(
sys: &PostgresSystemDatabase,
filter: &dbos::sysdb::types::WorkflowFilter<'_>,
) -> Vec<String> {
sys.list_workflows(filter, None)
.await
.expect("list failed")
.into_iter()
.map(|r| r.workflow_id)
.collect()
}
fn pending<'a>(executor_id: &'a str, application_version: &'a str) -> WorkflowFilter<'a> {
WorkflowFilter {
status: vec![WorkflowStatus::Pending],
executor_ids: vec![executor_id],
application_versions: vec![application_version],
load_input: false,
load_output: false,
..WorkflowFilter::default()
}
}
#[tokio::test]
async fn every_filter_narrows() {
use dbos::sysdb::types::WorkflowFilter as F;
let (sys, _db) = seeded().await;
let cases: Vec<(&str, F, &[&str])> = vec![
(
"no filter",
F::default(),
&["wf-a", "wf-b", "wf-c", "wf-d", "other-100%-done"],
),
(
"workflow_ids",
F {
workflow_ids: vec!["wf-a", "wf-d"],
..F::default()
},
&["wf-a", "wf-d"],
),
(
"workflow_id_prefixes",
F {
workflow_id_prefixes: vec!["wf-"],
..F::default()
},
&["wf-a", "wf-b", "wf-c", "wf-d"],
),
(
"prefix with a wildcard character",
F {
workflow_id_prefixes: vec!["other-100%"],
..F::default()
},
&["other-100%-done"],
),
(
"names",
F {
names: vec!["checkout"],
..F::default()
},
&["wf-a", "wf-c"],
),
(
"class_names",
F {
class_names: vec!["Checkout"],
..F::default()
},
&["wf-a"],
),
(
"config_names",
F {
config_names: vec!["primary"],
..F::default()
},
&["wf-a"],
),
(
"status",
F {
status: vec![WorkflowStatus::Delayed],
..F::default()
},
&["wf-c"],
),
(
"application_versions",
F {
application_versions: vec!["v2"],
..F::default()
},
&["wf-b"],
),
(
"executor_ids",
F {
executor_ids: vec!["alpha"],
..F::default()
},
&["wf-a"],
),
(
"authenticated_users",
F {
authenticated_users: vec!["bob"],
..F::default()
},
&["wf-b"],
),
(
"queue_names",
F {
queue_names: vec!["orders"],
..F::default()
},
&["wf-b", "wf-c"],
),
(
"queues_only",
F {
queues_only: true,
..F::default()
},
&["wf-b", "wf-c", "wf-d"],
),
(
"schedule_names",
F {
schedule_names: vec!["nightly"],
..F::default()
},
&["wf-b"],
),
(
"deduplication_ids",
F {
deduplication_ids: vec!["dedup-c"],
..F::default()
},
&["wf-c"],
),
(
"is_debounced",
F {
is_debounced: Some(true),
..F::default()
},
&["wf-c"],
),
(
"parent_workflow_ids",
F {
parent_workflow_ids: vec!["wf-a"],
..F::default()
},
&["wf-b", "wf-d"],
),
(
"has_parent",
F {
has_parent: Some(true),
..F::default()
},
&["wf-b", "wf-d"],
),
(
"has_parent = false",
F {
has_parent: Some(false),
..F::default()
},
&["wf-a", "wf-c", "other-100%-done"],
),
(
"forked_from",
F {
forked_from: vec!["wf-a"],
..F::default()
},
&["wf-c"],
),
(
"was_forked_from",
F {
was_forked_from: Some(true),
..F::default()
},
&["wf-a"],
),
(
"completed_after",
F {
completed_after: Some(Timestamp::from_epoch_ms(6000)),
..F::default()
},
&["wf-d"],
),
(
"completed_before",
F {
completed_before: Some(Timestamp::from_epoch_ms(6000)),
..F::default()
},
&["wf-b"],
),
(
"started_after",
F {
started_after: Some(Timestamp::from_epoch_ms(6000)),
..F::default()
},
&["wf-d"],
),
(
"started_before",
F {
started_before: Some(Timestamp::from_epoch_ms(6000)),
..F::default()
},
&["wf-b"],
),
(
"attributes",
F {
attributes: Some(r#"{"tenant": "acme"}"#),
..F::default()
},
&["wf-a"],
),
(
"combined filters are ANDed",
F {
names: vec!["checkout"],
queues_only: true,
..F::default()
},
&["wf-c"],
),
];
for (label, filter, expected) in cases {
let mut got = ids(&sys, &filter).await;
got.sort();
let mut want: Vec<String> = expected.iter().map(|s| (*s).to_owned()).collect();
want.sort();
assert_eq!(got, want, "filter `{label}` selected the wrong workflows");
}
}
#[tokio::test]
async fn results_are_ordered_and_pageable() {
use dbos::sysdb::types::WorkflowFilter as F;
let (sys, _db) = seeded().await;
let only_wf = F {
workflow_id_prefixes: vec!["wf-"],
..F::default()
};
let ascending = ids(&sys, &only_wf).await;
assert_eq!(ascending, ["wf-a", "wf-b", "wf-c", "wf-d"]);
let descending = ids(
&sys,
&F {
sort_desc: true,
..only_wf.clone()
},
)
.await;
assert_eq!(descending, ["wf-d", "wf-c", "wf-b", "wf-a"]);
let page = ids(
&sys,
&F {
limit: Some(2),
offset: Some(1),
..only_wf.clone()
},
)
.await;
assert_eq!(
page,
["wf-b", "wf-c"],
"limit and offset should page in sort order"
);
let rest = ids(
&sys,
&F {
offset: Some(3),
..only_wf
},
)
.await;
assert_eq!(rest, ["wf-d"]);
}
#[tokio::test]
async fn payloads_can_be_left_unloaded() {
use dbos::sysdb::types::WorkflowFilter as F;
let (sys, _db) = sysdb().await;
let wf = NewWorkflow {
input: Some(r#"{"positionalArgs":[1]}"#),
..NewWorkflow::new("wf-payload")
};
sys.init_workflow(&wf, None, Submission::Fresh, None)
.await
.unwrap();
sys.record_workflow_outcome("wf-payload", Outcome::Output(Some("42")))
.await
.unwrap();
let loaded = &sys.list_workflows(&F::default(), None).await.unwrap()[0];
assert_eq!(loaded.input.as_deref(), Some(r#"{"positionalArgs":[1]}"#));
assert_eq!(loaded.output.as_deref(), Some("42"));
let bare = &sys
.list_workflows(
&F {
load_input: false,
load_output: false,
..F::default()
},
None,
)
.await
.unwrap()[0];
assert_eq!(bare.workflow_id, "wf-payload", "the row is still returned");
assert_eq!(bare.input, None, "input was not asked for");
assert_eq!(bare.output, None, "output was not asked for");
assert_eq!(
bare.status,
WorkflowStatus::Success,
"declining payloads must not affect any other column",
);
}
#[tokio::test]
async fn cancelling_clears_the_queue_and_the_deduplication_key() {
let (sys, _db) = sysdb().await;
let wf = NewWorkflow {
queue_name: Some("orders"),
deduplication_id: Some("dedup-1"),
..NewWorkflow::new("wf-cancel")
};
sys.init_workflow(&wf, None, Submission::Fresh, None)
.await
.unwrap();
let cancelled = sys
.cancel_workflows(&["wf-cancel"], false, None)
.await
.unwrap();
assert_eq!(cancelled, ["wf-cancel"]);
let read = sys.get_workflow("wf-cancel").await.unwrap().unwrap();
assert_eq!(read.status, WorkflowStatus::Cancelled);
assert_eq!(
read.queue_name, None,
"a cancelled workflow must not dequeue"
);
assert_eq!(
read.deduplication_id, None,
"the key must be released, or it blocks a later workflow forever",
);
assert!(read.completed_at.is_some());
let holders = sys
.list_workflows(
&dbos::sysdb::types::WorkflowFilter {
deduplication_ids: vec!["dedup-1"],
..Default::default()
},
None,
)
.await
.unwrap();
assert!(
holders.is_empty(),
"the deduplication key should be held by nobody",
);
}
#[tokio::test]
async fn cancelling_twice_does_not_move_the_cancellation() {
let (sys, _db) = sysdb().await;
sys.init_workflow(
&workflow("wf-twice-cancelled"),
None,
Submission::Fresh,
None,
)
.await
.unwrap();
let first = sys
.cancel_workflows(&["wf-twice-cancelled"], false, None)
.await
.unwrap();
assert_eq!(first, ["wf-twice-cancelled"]);
let after_first = sys
.get_workflow("wf-twice-cancelled")
.await
.unwrap()
.unwrap();
let second = sys
.cancel_workflows(&["wf-twice-cancelled"], false, None)
.await
.unwrap();
assert!(
second.is_empty(),
"nothing moved the second time, and the caller is told so",
);
let after_second = sys
.get_workflow("wf-twice-cancelled")
.await
.unwrap()
.unwrap();
assert_eq!(
after_second.completed_at, after_first.completed_at,
"the moment of cancellation must not drift on a repeat call",
);
assert_eq!(after_second.updated_at, after_first.updated_at);
assert_eq!(after_second.status, WorkflowStatus::Cancelled);
}
#[tokio::test]
async fn cancelling_a_finished_workflow_does_not_overwrite_it() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-done"), None, Submission::Fresh, None)
.await
.unwrap();
sys.record_workflow_outcome("wf-done", Outcome::Output(Some("42")))
.await
.unwrap();
let cancelled = sys
.cancel_workflows(&["wf-done"], false, None)
.await
.unwrap();
assert!(
cancelled.is_empty(),
"nothing moved, and the caller is told so",
);
let read = sys.get_workflow("wf-done").await.unwrap().unwrap();
assert_eq!(read.status, WorkflowStatus::Success);
assert_eq!(read.output.as_deref(), Some("42"), "the result survives");
}
#[tokio::test]
async fn cancelling_children_descends_the_whole_tree() {
let (sys, _db) = sysdb().await;
for (id, parent) in [
("wf-root", None),
("wf-child", Some("wf-root")),
("wf-grandchild", Some("wf-child")),
("wf-unrelated", None),
] {
let caller = parent.map(|parent_workflow_id| InitWorkflowCaller {
parent_workflow_id,
step_id: 0,
step_name: "spawn",
started_at: Timestamp::now(),
});
sys.init_workflow(&NewWorkflow::new(id), None, Submission::Fresh, caller)
.await
.unwrap();
}
let mut cancelled = sys
.cancel_workflows(&["wf-root"], true, None)
.await
.unwrap();
cancelled.sort();
assert_eq!(cancelled, ["wf-child", "wf-grandchild", "wf-root"]);
let untouched = sys.get_workflow("wf-unrelated").await.unwrap().unwrap();
assert_eq!(
untouched.status,
WorkflowStatus::Pending,
"the cascade must follow parentage, not cancel everything",
);
sys.init_workflow(&workflow("wf-root2"), None, Submission::Fresh, None)
.await
.unwrap();
sys.init_workflow(
&NewWorkflow::new("wf-child2"),
None,
Submission::Fresh,
Some(InitWorkflowCaller {
parent_workflow_id: "wf-root2",
step_id: 0,
step_name: "spawn",
started_at: Timestamp::now(),
}),
)
.await
.unwrap();
let shallow = sys
.cancel_workflows(&["wf-root2"], false, None)
.await
.unwrap();
assert_eq!(shallow, ["wf-root2"]);
}
#[tokio::test]
async fn resuming_clears_the_attempt_count_and_the_deadline() {
let (sys, _db) = sysdb().await;
let wf = NewWorkflow {
deadline: Some(Timestamp::from_epoch_ms(1_000)),
..workflow("wf-resume")
};
for _ in 0..2 {
sys.init_workflow(&wf, None, Submission::Recovery, None)
.await
.unwrap();
}
let resumed = sys
.resume_workflows(&["wf-resume"], None, None)
.await
.unwrap();
assert_eq!(resumed, ["wf-resume"]);
let read = sys.get_workflow("wf-resume").await.unwrap().unwrap();
assert_eq!(read.status, WorkflowStatus::Enqueued);
assert_eq!(
read.queue_name.as_deref(),
Some(dbos::sysdb::INTERNAL_QUEUE),
"no queue named means the internal one",
);
assert_eq!(read.recovery_attempts, 0);
assert_eq!(
read.deadline, None,
"a deadline set before parking would fail the fresh attempt at once",
);
assert_eq!(read.completed_at, None);
sys.init_workflow(&workflow("wf-resume2"), None, Submission::Fresh, None)
.await
.unwrap();
sys.resume_workflows(&["wf-resume2"], Some("orders"), None)
.await
.unwrap();
let read = sys.get_workflow("wf-resume2").await.unwrap().unwrap();
assert_eq!(read.queue_name.as_deref(), Some("orders"));
}
#[tokio::test]
async fn resuming_a_missing_workflow_is_an_error_but_cancelling_one_is_not() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-real"), None, Submission::Fresh, None)
.await
.unwrap();
match sys
.resume_workflows(&["wf-real", "wf-ghost"], None, None)
.await
{
Err(Error::NonExistentWorkflow { workflow_ids }) => {
assert_eq!(workflow_ids, ["wf-ghost"], "only the missing id is named");
}
other => panic!("expected a non-existent-workflow error, got {other:?}"),
}
let read = sys.get_workflow("wf-real").await.unwrap().unwrap();
assert_eq!(
read.status,
WorkflowStatus::Pending,
"the batch is rejected before anything moves",
);
let cancelled = sys
.cancel_workflows(&["wf-ghost"], false, None)
.await
.expect("cancelling a missing workflow is a no-op, not an error");
assert!(cancelled.is_empty());
}
#[tokio::test]
async fn attributes_are_replaced_not_merged() {
let (sys, _db) = sysdb().await;
let wf = NewWorkflow {
attributes: Some(r#"{"tenant": "acme", "tier": "gold"}"#),
..NewWorkflow::new("wf-attrs")
};
sys.init_workflow(&wf, None, Submission::Fresh, None)
.await
.unwrap();
sys.update_workflow_attributes("wf-attrs", Some(r#"{"tier": "silver"}"#), None)
.await
.unwrap();
let read = sys.get_workflow("wf-attrs").await.unwrap().unwrap();
let attributes = read.attributes.unwrap();
assert!(attributes.contains("silver"));
assert!(
!attributes.contains("acme"),
"a replacement drops keys the new value omits, got {attributes}",
);
sys.update_workflow_attributes("wf-attrs", None, None)
.await
.unwrap();
let read = sys.get_workflow("wf-attrs").await.unwrap().unwrap();
assert_eq!(read.attributes, None);
}
#[tokio::test]
async fn empty_and_zero_values_are_rejected() {
let (sys, _db) = sysdb().await;
let cases: Vec<(&str, NewWorkflow)> = vec![
("workflow_id", NewWorkflow::new("")),
(
"queue_name",
NewWorkflow {
queue_name: Some(""),
..NewWorkflow::new("wf-bad")
},
),
(
"name",
NewWorkflow {
name: Some(""),
..NewWorkflow::new("wf-bad")
},
),
(
"deduplication_id",
NewWorkflow {
deduplication_id: Some(""),
..NewWorkflow::new("wf-bad")
},
),
(
"delay",
NewWorkflow {
delay: Some(std::time::Duration::ZERO),
..NewWorkflow::new("wf-bad")
},
),
(
"timeout",
NewWorkflow {
timeout: Some(std::time::Duration::ZERO),
..NewWorkflow::new("wf-bad")
},
),
];
for (expected_field, wf) in cases {
match sys.init_workflow(&wf, None, Submission::Fresh, None).await {
Err(Error::InvalidInput { field, .. }) => assert_eq!(
field, expected_field,
"the wrong field was blamed for {expected_field}",
),
other => panic!("expected {expected_field} to be rejected, got {other:?}"),
}
}
let all = sys
.list_workflows(&dbos::sysdb::types::WorkflowFilter::default(), None)
.await
.unwrap();
assert!(all.is_empty(), "a rejected workflow must leave no row");
}
#[tokio::test]
async fn empty_auth_fields_are_normalised_to_null() {
let (sys, _db) = sysdb().await;
let wf = NewWorkflow {
authenticated_user: Some(""),
assumed_role: Some(""),
..NewWorkflow::new("wf-auth")
};
sys.init_workflow(&wf, None, Submission::Fresh, None)
.await
.expect("empty auth fields are normalised, not rejected");
let read = sys.get_workflow("wf-auth").await.unwrap().unwrap();
assert_eq!(read.authenticated_user, None);
assert_eq!(read.assumed_role, None);
}
#[tokio::test]
async fn a_step_result_round_trips() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-steps"), None, Submission::Fresh, None)
.await
.unwrap();
assert_eq!(
sys.check_step("wf-steps", 0, "charge").await.unwrap(),
None,
"a step that has not run has no result",
);
let timing = StepTiming {
started_at: Timestamp::from_epoch_ms(1_000),
completed_at: Timestamp::from_epoch_ms(2_000),
};
sys.record_step(
"wf-steps",
0,
"charge",
Outcome::Output(Some(r#"{"ok":true}"#)),
Some("portable_json"),
Some(timing),
)
.await
.unwrap();
let read = sys
.check_step("wf-steps", 0, "charge")
.await
.unwrap()
.expect("the step should be recorded");
assert_eq!(read.output.as_deref(), Some(r#"{"ok":true}"#));
assert_eq!(read.error, None);
assert_eq!(read.serialization.as_deref(), Some("portable_json"));
assert_eq!(read.started_at, Some(Timestamp::from_epoch_ms(1_000)));
assert_eq!(read.completed_at, Some(Timestamp::from_epoch_ms(2_000)));
assert_eq!(sys.check_step("wf-steps", 1, "refund").await.unwrap(), None,);
}
#[tokio::test]
async fn a_step_recorded_twice_distinguishes_a_retry_from_a_rival() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-twice"), None, Submission::Fresh, None)
.await
.unwrap();
let timing = StepTiming {
started_at: Timestamp::from_epoch_ms(4_000),
completed_at: Timestamp::from_epoch_ms(5_000),
};
let first = Outcome::Output(Some("first"));
sys.record_step("wf-twice", 0, "charge", first, None, Some(timing))
.await
.unwrap();
sys.record_step("wf-twice", 0, "charge", first, None, Some(timing))
.await
.expect("a caller's own retry must succeed");
let rival_timing = StepTiming {
started_at: Timestamp::from_epoch_ms(4_000),
completed_at: Timestamp::from_epoch_ms(6_000),
};
match sys
.record_step(
"wf-twice",
0,
"charge",
Outcome::Output(Some("second")),
None,
Some(rival_timing),
)
.await
{
Err(Error::StepAlreadyRecorded { step_id, .. }) => assert_eq!(step_id, 0),
other => panic!("expected the rival to be rejected, got {other:?}"),
}
let read = sys
.check_step("wf-twice", 0, "charge")
.await
.unwrap()
.unwrap();
assert_eq!(read.output.as_deref(), Some("first"));
}
#[tokio::test]
async fn a_step_recorded_under_another_name_is_rejected() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-drift"), None, Submission::Fresh, None)
.await
.unwrap();
sys.record_step("wf-drift", 0, "charge", Outcome::Output(None), None, None)
.await
.unwrap();
match sys.check_step("wf-drift", 0, "refund").await {
Err(Error::UnexpectedStep {
expected, recorded, ..
}) => {
assert_eq!(expected, "refund");
assert_eq!(recorded, "charge");
}
other => panic!("expected a step-drift error, got {other:?}"),
}
}
#[tokio::test]
async fn a_cancelled_workflow_refuses_to_replay_steps() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-stopped"), None, Submission::Fresh, None)
.await
.unwrap();
sys.cancel_workflows(&["wf-stopped"], false, None)
.await
.unwrap();
match sys.check_step("wf-stopped", 0, "charge").await {
Err(Error::WorkflowCancelled { workflow_id }) => assert_eq!(workflow_id, "wf-stopped"),
other => panic!("expected a cancellation error, got {other:?}"),
}
match sys.check_step("wf-ghost", 0, "charge").await {
Err(Error::NonExistentWorkflow { workflow_ids }) => assert_eq!(workflow_ids, ["wf-ghost"]),
other => panic!("expected a non-existent-workflow error, got {other:?}"),
}
}
#[tokio::test]
async fn steps_are_listed_in_execution_order() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-list"), None, Submission::Fresh, None)
.await
.unwrap();
for id in [2, 0, 1] {
let (name, output) = (format!("step-{id}"), format!("out-{id}"));
sys.record_step(
"wf-list",
id,
&name,
Outcome::Output(Some(&output)),
None,
None,
)
.await
.unwrap();
}
let steps = sys
.list_workflow_steps("wf-list", true, None, None, None)
.await
.unwrap();
let ids: Vec<i32> = steps.iter().map(|s| s.step_id).collect();
assert_eq!(ids, [0, 1, 2], "ordered by position, not by when recorded");
assert_eq!(steps[1].step_name, "step-1");
assert_eq!(steps[1].output.as_deref(), Some("out-1"));
let bare = sys
.list_workflow_steps("wf-list", false, None, None, None)
.await
.unwrap();
assert_eq!(bare.len(), 3, "declining payloads returns the same rows");
assert_eq!(bare[1].output, None);
assert_eq!(bare[1].step_name, "step-1", "other columns still load");
let page = sys
.list_workflow_steps("wf-list", true, Some(1), Some(1), None)
.await
.unwrap();
assert_eq!(page.len(), 1);
assert_eq!(page[0].step_id, 1);
}
#[tokio::test]
async fn a_child_and_its_start_record_are_one_write() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-parent"), None, Submission::Fresh, None)
.await
.unwrap();
let start = InitWorkflowCaller {
parent_workflow_id: "wf-parent",
step_id: 0,
step_name: "checkout",
started_at: Timestamp::now(),
};
sys.init_workflow(&workflow("wf-kid"), None, Submission::Fresh, Some(start))
.await
.expect("the child was not created");
assert_eq!(
sys.get_workflow("wf-kid")
.await
.unwrap()
.expect("the child's own row is missing")
.parent_workflow_id
.as_deref(),
Some("wf-parent"),
);
let step = sys
.check_step("wf-parent", 0, "checkout")
.await
.unwrap()
.expect("the start is recorded against the parent's step");
assert_eq!(step.child_workflow_id.as_deref(), Some("wf-kid"));
let clash = InitWorkflowCaller {
parent_workflow_id: "wf-parent",
step_id: 0,
step_name: "checkout",
started_at: Timestamp::now(),
};
match sys
.init_workflow(
&workflow("wf-second-kid"),
None,
Submission::Fresh,
Some(clash),
)
.await
{
Err(Error::StepAlreadyRecorded { step_id, .. }) => assert_eq!(step_id, 0),
other => panic!("expected the clashing start to be refused, got {other:?}"),
}
assert!(
sys.get_workflow("wf-second-kid").await.unwrap().is_none(),
"the refused start left a child workflow behind, which is the state the transaction exists \
to make unreachable",
);
assert_eq!(
sys.check_step("wf-parent", 0, "checkout")
.await
.unwrap()
.and_then(|step| step.child_workflow_id),
Some("wf-kid".to_owned()),
);
}
#[tokio::test]
async fn a_child_workflow_is_recorded_against_its_step() {
let (sys, _db) = sysdb().await;
for id in ["wf-parent", "wf-kid"] {
sys.init_workflow(&workflow(id), None, Submission::Fresh, None)
.await
.unwrap();
}
assert_eq!(
sys.check_step("wf-parent", 0, "run_child").await.unwrap(),
None,
"nothing has started a child at this position",
);
sys.record_child_workflow("wf-parent", "wf-kid", 0, "run_child", None)
.await
.unwrap();
let step = sys
.check_step("wf-parent", 0, "run_child")
.await
.unwrap()
.expect("the launch is recorded as a step");
assert_eq!(step.child_workflow_id.as_deref(), Some("wf-kid"));
assert_eq!(step.output, None);
assert_eq!(
step.error, None,
"the launch carries no result; the child's outcome lives on the child's row",
);
sys.record_child_workflow("wf-parent", "wf-kid", 0, "run_child", None)
.await
.expect("re-recording the same child is idempotent");
match sys
.record_child_workflow("wf-parent", "wf-other", 0, "run_child", None)
.await
{
Err(Error::StepAlreadyRecorded { step_id, .. }) => assert_eq!(step_id, 0),
other => panic!("expected a conflicting child to be rejected, got {other:?}"),
}
match sys
.record_child_workflow("wf-parent", "", 2, "run_child", None)
.await
{
Err(Error::InvalidInput { field, .. }) => assert_eq!(field, "child_workflow_id"),
other => panic!("expected an empty child id to be rejected, got {other:?}"),
}
sys.record_step("wf-parent", 1, "charge", Outcome::Output(None), None, None)
.await
.unwrap();
let step = sys
.check_step("wf-parent", 1, "charge")
.await
.unwrap()
.unwrap();
assert_eq!(step.child_workflow_id, None);
}
#[tokio::test]
async fn recording_a_step_restamps_the_executor() {
let db = test_database().await;
let pool = db.pool().await;
let original = PostgresSystemDatabase::from_pool(pool.clone(), &Settings::default());
let wf = NewWorkflow {
executor_id: Some("executor-a"),
..workflow("wf-takeover")
};
original
.init_workflow(&wf, None, Submission::Fresh, None)
.await
.unwrap();
let recovering = PostgresSystemDatabase::from_pool(
pool,
&Settings {
executor_id: Some("executor-b"),
..Settings::default()
},
);
recovering
.record_step(
"wf-takeover",
0,
"charge",
Outcome::Output(None),
None,
None,
)
.await
.unwrap();
let read = recovering
.get_workflow("wf-takeover")
.await
.unwrap()
.unwrap();
assert_eq!(
read.executor_id.as_deref(),
Some("executor-b"),
"running a step is what makes an executor the owner",
);
}
#[tokio::test]
async fn a_step_may_be_recorded_without_timings() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-timing"), None, Submission::Fresh, None)
.await
.unwrap();
sys.record_step("wf-timing", 0, "charge", Outcome::Output(None), None, None)
.await
.expect("a step without timings is legal");
let read = sys
.check_step("wf-timing", 0, "charge")
.await
.unwrap()
.unwrap();
assert_eq!(read.started_at, None);
assert_eq!(read.completed_at, None);
let timing = StepTiming {
started_at: Timestamp::from_epoch_ms(1_000),
completed_at: Timestamp::from_epoch_ms(2_000),
};
sys.record_step(
"wf-timing",
1,
"refund",
Outcome::Output(None),
None,
Some(timing),
)
.await
.unwrap();
let read = sys
.check_step("wf-timing", 1, "refund")
.await
.unwrap()
.unwrap();
assert_eq!(read.started_at, Some(Timestamp::from_epoch_ms(1_000)));
assert_eq!(read.completed_at, Some(Timestamp::from_epoch_ms(2_000)));
}
#[tokio::test]
async fn an_untimed_step_cannot_detect_a_rival() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-untimed"), None, Submission::Fresh, None)
.await
.unwrap();
let first = Outcome::Output(Some("first"));
sys.record_step("wf-untimed", 0, "charge", first, None, None)
.await
.unwrap();
let rival = Outcome::Output(Some("second"));
sys.record_step("wf-untimed", 0, "charge", rival, None, None)
.await
.expect("with no completion time there is nothing to compare against");
let read = sys
.check_step("wf-untimed", 0, "charge")
.await
.unwrap()
.unwrap();
assert_eq!(
read.output.as_deref(),
Some("first"),
"the duplicate is ignored, not applied",
);
}
#[tokio::test]
async fn a_step_records_either_an_output_or_an_error() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-outcome"), None, Submission::Fresh, None)
.await
.unwrap();
let failed = Outcome::Error(r#"{"type":"ValueError"}"#);
sys.record_step("wf-outcome", 0, "charge", failed, None, None)
.await
.unwrap();
let read = sys
.check_step("wf-outcome", 0, "charge")
.await
.unwrap()
.unwrap();
assert_eq!(read.output, None);
assert_eq!(read.error.as_deref(), Some(r#"{"type":"ValueError"}"#));
sys.record_step("wf-outcome", 1, "notify", Outcome::Output(None), None, None)
.await
.unwrap();
let read = sys
.check_step("wf-outcome", 1, "notify")
.await
.unwrap()
.unwrap();
assert_eq!(read.output, None);
assert_eq!(read.error, None, "a void return is a success, not an error");
}
#[tokio::test]
async fn losing_the_checkpoint_does_not_claim_the_workflow() {
let db = test_database().await;
let pool = db.pool().await;
let winner = PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
executor_id: Some("executor-a"),
..Settings::default()
},
);
let loser = PostgresSystemDatabase::from_pool(
pool,
&Settings {
executor_id: Some("executor-b"),
..Settings::default()
},
);
winner
.init_workflow(&workflow("wf-race-step"), None, Submission::Fresh, None)
.await
.unwrap();
let winning = StepTiming {
started_at: Timestamp::from_epoch_ms(1_000),
completed_at: Timestamp::from_epoch_ms(2_000),
};
winner
.record_step(
"wf-race-step",
0,
"charge",
Outcome::Output(Some("first")),
None,
Some(winning),
)
.await
.unwrap();
let read = winner.get_workflow("wf-race-step").await.unwrap().unwrap();
assert_eq!(read.executor_id.as_deref(), Some("executor-a"));
let losing = StepTiming {
started_at: Timestamp::from_epoch_ms(1_000),
completed_at: Timestamp::from_epoch_ms(3_000),
};
let result = loser
.record_step(
"wf-race-step",
0,
"charge",
Outcome::Output(Some("second")),
None,
Some(losing),
)
.await;
assert!(
matches!(result, Err(Error::StepAlreadyRecorded { .. })),
"expected the loser to be rejected, got {result:?}",
);
let read = winner.get_workflow("wf-race-step").await.unwrap().unwrap();
assert_eq!(
read.executor_id.as_deref(),
Some("executor-a"),
"the loser must not take ownership of a workflow it is not advancing",
);
}
#[tokio::test]
async fn workflow_children_reach_the_whole_tree() {
let (sys, _db) = sysdb().await;
for (id, parent) in [
("wf-root", None),
("wf-child", Some("wf-root")),
("wf-grandchild", Some("wf-child")),
("wf-stranger", None),
] {
let caller = parent.map(|parent_workflow_id| InitWorkflowCaller {
parent_workflow_id,
step_id: 0,
step_name: "spawn",
started_at: Timestamp::now(),
});
sys.init_workflow(&NewWorkflow::new(id), None, Submission::Fresh, caller)
.await
.unwrap();
}
let mut children = sys.get_workflow_children("wf-root").await.unwrap();
children.sort();
assert_eq!(children, ["wf-child", "wf-grandchild"]);
assert!(
sys.get_workflow_children("wf-stranger")
.await
.unwrap()
.is_empty(),
);
}
#[tokio::test]
async fn deleting_a_workflow_cascades_to_its_rows() {
let (sys, _db) = sysdb().await;
for (id, parent) in [("wf-gone", None), ("wf-gone-kid", Some("wf-gone"))] {
let caller = parent.map(|parent_workflow_id| InitWorkflowCaller {
parent_workflow_id,
step_id: 0,
step_name: "spawn",
started_at: Timestamp::now(),
});
sys.init_workflow(&NewWorkflow::new(id), None, Submission::Fresh, caller)
.await
.unwrap();
}
sys.record_step("wf-gone", 1, "charge", Outcome::Output(None), None, None)
.await
.unwrap();
let deleted = sys
.delete_workflows(&["wf-gone"], false, None)
.await
.unwrap();
assert_eq!(deleted, 1);
assert!(sys.get_workflow("wf-gone").await.unwrap().is_none());
assert!(sys.get_workflow("wf-gone-kid").await.unwrap().is_some());
let steps = sys
.list_workflow_steps("wf-gone", true, None, None, None)
.await
.unwrap();
assert!(steps.is_empty(), "operation_outputs should cascade");
for (id, parent) in [
("wf-p", None),
("wf-c", Some("wf-p")),
("wf-g", Some("wf-c")),
] {
let caller = parent.map(|parent_workflow_id| InitWorkflowCaller {
parent_workflow_id,
step_id: 0,
step_name: "spawn",
started_at: Timestamp::now(),
});
sys.init_workflow(&NewWorkflow::new(id), None, Submission::Fresh, caller)
.await
.unwrap();
}
let deleted = sys.delete_workflows(&["wf-p"], true, None).await.unwrap();
assert_eq!(deleted, 3, "the whole tree, at every depth");
}
#[tokio::test]
async fn pending_workflows_are_scoped_by_executor_and_version() {
let (sys, _db) = sysdb().await;
for (id, executor, version) in [
("wf-mine", "alpha", "v1"),
("wf-theirs", "beta", "v1"),
("wf-old-code", "alpha", "v0"),
] {
let wf = NewWorkflow {
executor_id: Some(executor),
application_version: Some(version),
..NewWorkflow::new(id)
};
sys.init_workflow(&wf, None, Submission::Fresh, None)
.await
.unwrap();
}
let done = NewWorkflow {
executor_id: Some("alpha"),
application_version: Some("v1"),
..NewWorkflow::new("wf-finished")
};
sys.init_workflow(&done, None, Submission::Fresh, None)
.await
.unwrap();
sys.record_workflow_outcome("wf-finished", Outcome::Output(None))
.await
.unwrap();
let found = ids(&sys, &pending("alpha", "v1")).await;
assert_eq!(
found,
["wf-mine"],
"another executor's work, another version's work, and finished work are all excluded",
);
}
#[tokio::test]
async fn reenqueue_for_recovery_returns_workflows_to_their_own_queues() {
let (sys, _db) = sysdb().await;
sys.upsert_queue(&NewQueue::new("orders"), OnExistingQueue::Update)
.await
.unwrap();
for (id, queue) in [("wf-was-queued", Some("orders")), ("wf-never-queued", None)] {
let wf = NewWorkflow {
executor_id: Some("alpha"),
application_version: Some("v1"),
queue_name: queue,
..NewWorkflow::new(id)
};
sys.init_workflow(&wf, None, Submission::Fresh, None)
.await
.unwrap();
}
let queue = sys.get_queue("orders").await.unwrap().unwrap();
let claimed = sys
.start_queued_workflows(&queue, "alpha", "v1", None, 0, 0)
.await
.unwrap();
assert_eq!(claimed, ["wf-was-queued"]);
let mut moved = sys
.reenqueue_for_recovery(&["alpha"], "v1", INTERNAL_QUEUE)
.await
.unwrap();
moved.sort();
assert_eq!(moved, ["wf-never-queued", "wf-was-queued"]);
let was_queued = sys.get_workflow("wf-was-queued").await.unwrap().unwrap();
assert_eq!(was_queued.status, WorkflowStatus::Enqueued);
assert_eq!(
was_queued.queue_name.as_deref(),
Some("orders"),
"a workflow that came off a queue goes back to that queue"
);
assert!(
was_queued.started_at.is_none(),
"the dead run's start time would make the queue wait look like execution"
);
let never_queued = sys.get_workflow("wf-never-queued").await.unwrap().unwrap();
assert_eq!(never_queued.status, WorkflowStatus::Enqueued);
assert_eq!(
never_queued.queue_name.as_deref(),
Some(INTERNAL_QUEUE),
"a workflow that was never queued lands on the recovery queue"
);
}
#[tokio::test]
async fn reenqueue_for_recovery_leaves_a_workflow_a_live_executor_took() {
let (sys, _db) = sysdb().await;
let wf = NewWorkflow {
executor_id: Some("dead"),
application_version: Some("v1"),
..NewWorkflow::new("wf-taken-over")
};
sys.init_workflow(&wf, None, Submission::Fresh, None)
.await
.unwrap();
assert_eq!(
sys.reenqueue_for_recovery(&["dead"], "v1", INTERNAL_QUEUE)
.await
.unwrap(),
["wf-taken-over"],
);
let internal = NewQueue::new(INTERNAL_QUEUE);
sys.upsert_queue(&internal, OnExistingQueue::Update)
.await
.unwrap();
let queue = sys.get_queue(INTERNAL_QUEUE).await.unwrap().unwrap();
assert_eq!(
sys.start_queued_workflows(&queue, "live", "v1", None, 0, 0)
.await
.unwrap(),
["wf-taken-over"],
);
assert!(
sys.reenqueue_for_recovery(&["dead"], "v1", INTERNAL_QUEUE)
.await
.unwrap()
.is_empty(),
"a repeat sweep tore a running workflow off its executor"
);
assert_eq!(
sys.get_workflow("wf-taken-over")
.await
.unwrap()
.unwrap()
.status,
WorkflowStatus::Pending,
"the workflow the live executor is running was disturbed"
);
}
#[tokio::test]
async fn reenqueue_for_recovery_is_scoped_by_executor_and_version() {
let (sys, _db) = sysdb().await;
for (id, executor, version) in [
("wf-mine", "alpha", "v1"),
("wf-theirs", "beta", "v1"),
("wf-old-code", "alpha", "v0"),
] {
let wf = NewWorkflow {
executor_id: Some(executor),
application_version: Some(version),
..NewWorkflow::new(id)
};
sys.init_workflow(&wf, None, Submission::Fresh, None)
.await
.unwrap();
}
let done = NewWorkflow {
executor_id: Some("alpha"),
application_version: Some("v1"),
..NewWorkflow::new("wf-finished")
};
sys.init_workflow(&done, None, Submission::Fresh, None)
.await
.unwrap();
sys.record_workflow_outcome("wf-finished", Outcome::Output(None))
.await
.unwrap();
assert_eq!(
sys.reenqueue_for_recovery(&["alpha"], "v1", INTERNAL_QUEUE)
.await
.unwrap(),
["wf-mine"],
);
assert!(
sys.reenqueue_for_recovery(&[], "v1", INTERNAL_QUEUE)
.await
.unwrap()
.is_empty(),
"naming no executors moved something",
);
}
#[tokio::test]
async fn a_delay_can_be_moved_only_while_the_workflow_is_delayed() {
let (sys, _db) = sysdb().await;
let wf = NewWorkflow {
queue_name: Some("orders"),
delay: Some(std::time::Duration::from_secs(3600)),
..NewWorkflow::new("wf-delayed")
};
sys.init_workflow(&wf, None, Submission::Fresh, None)
.await
.unwrap();
assert_eq!(
sys.get_workflow("wf-delayed")
.await
.unwrap()
.unwrap()
.status,
WorkflowStatus::Delayed,
);
sys.set_workflow_delay(
"wf-delayed",
WorkflowDelay::Until(Timestamp::from_epoch_ms(9_000_000)),
None,
)
.await
.unwrap();
let read = sys.get_workflow("wf-delayed").await.unwrap().unwrap();
assert_eq!(read.delay_until, Some(Timestamp::from_epoch_ms(9_000_000)));
let before = Timestamp::now();
sys.set_workflow_delay(
"wf-delayed",
WorkflowDelay::For(std::time::Duration::from_secs(60)),
None,
)
.await
.unwrap();
let read = sys.get_workflow("wf-delayed").await.unwrap().unwrap();
let offset = read.delay_until.unwrap().as_epoch_ms() - before.as_epoch_ms();
assert!(
(60_000..70_000).contains(&offset),
"expected ~60s from now, got {offset}ms",
);
sys.init_workflow(&workflow("wf-running"), None, Submission::Fresh, None)
.await
.unwrap();
sys.set_workflow_delay(
"wf-running",
WorkflowDelay::Until(Timestamp::from_epoch_ms(9_000_000)),
None,
)
.await
.unwrap();
let read = sys.get_workflow("wf-running").await.unwrap().unwrap();
assert_eq!(read.delay_until, None, "a PENDING workflow is untouched");
assert_eq!(read.status, WorkflowStatus::Pending);
}
#[tokio::test]
async fn releasing_a_delayed_workflow_clears_only_the_debounce_key() {
let (sys, db) = sysdb().await;
for (id, debounced) in [("wf-debounced", true), ("wf-plain-dedup", false)] {
let key = format!("key-{id}");
let wf = NewWorkflow {
queue_name: Some("orders"),
delay: Some(std::time::Duration::from_secs(3600)),
deduplication_id: Some(&key),
is_debounced: debounced,
..NewWorkflow::new(id)
};
sys.init_workflow(&wf, None, Submission::Fresh, None)
.await
.unwrap();
}
let held = NewWorkflow {
queue_name: Some("orders"),
delay: Some(std::time::Duration::from_secs(3600)),
..NewWorkflow::new("wf-still-held")
};
sys.init_workflow(&held, None, Submission::Fresh, None)
.await
.unwrap();
let mut conn = db.admin_connection().await;
sqlx::query(sqlx::AssertSqlSafe(
"UPDATE dbos.workflow_status SET delay_until_epoch_ms = 1 \
WHERE workflow_uuid IN ('wf-debounced', 'wf-plain-dedup')",
))
.execute(&mut conn)
.await
.unwrap();
let moved = sys.transition_delayed_workflows().await.unwrap();
assert_eq!(moved, 2, "only the two whose delay expired");
let debounced = sys.get_workflow("wf-debounced").await.unwrap().unwrap();
assert_eq!(debounced.status, WorkflowStatus::Enqueued);
assert_eq!(
debounced.deduplication_id, None,
"a debounce key is released with the workflow",
);
let plain = sys.get_workflow("wf-plain-dedup").await.unwrap().unwrap();
assert_eq!(plain.status, WorkflowStatus::Enqueued);
assert_eq!(
plain.deduplication_id.as_deref(),
Some("key-wf-plain-dedup"),
"an ordinary deduplication id is not a debounce key and must survive",
);
let held = sys.get_workflow("wf-still-held").await.unwrap().unwrap();
assert_eq!(held.status, WorkflowStatus::Delayed);
}
#[tokio::test]
async fn a_queued_workflow_can_be_returned_to_its_queue() {
let (sys, db) = sysdb().await;
let queued = NewWorkflow {
queue_name: Some("orders"),
..NewWorkflow::new("wf-claimed")
};
sys.init_workflow(&queued, None, Submission::Fresh, None)
.await
.unwrap();
let mut conn = db.admin_connection().await;
sqlx::query(sqlx::AssertSqlSafe(
"UPDATE dbos.workflow_status SET status = 'PENDING', started_at_epoch_ms = 1000 \
WHERE workflow_uuid = 'wf-claimed'",
))
.execute(&mut conn)
.await
.unwrap();
assert!(sys.clear_queue_assignment("wf-claimed").await.unwrap());
let read = sys.get_workflow("wf-claimed").await.unwrap().unwrap();
assert_eq!(read.status, WorkflowStatus::Enqueued);
assert_eq!(read.started_at, None, "the claim's start time is cleared");
sys.init_workflow(&workflow("wf-direct"), None, Submission::Fresh, None)
.await
.unwrap();
assert!(!sys.clear_queue_assignment("wf-direct").await.unwrap());
}
#[tokio::test]
async fn a_held_deduplication_key_is_reported_as_such() {
let (sys, _db) = sysdb().await;
let holder = NewWorkflow {
queue_name: Some("orders"),
deduplication_id: Some("only-once"),
..NewWorkflow::new("wf-holder")
};
sys.init_workflow(&holder, None, Submission::Fresh, None)
.await
.unwrap();
let rival = NewWorkflow {
queue_name: Some("orders"),
deduplication_id: Some("only-once"),
..NewWorkflow::new("wf-rival")
};
match sys
.init_workflow(&rival, None, Submission::Fresh, None)
.await
{
Err(Error::QueueDeduplicated {
workflow_id,
queue_name,
deduplication_id,
}) => {
assert_eq!(workflow_id, "wf-rival");
assert_eq!(queue_name, "orders");
assert_eq!(deduplication_id, "only-once");
}
other => panic!("expected a deduplication error, got {other:?}"),
}
let elsewhere = NewWorkflow {
queue_name: Some("reports"),
deduplication_id: Some("only-once"),
..NewWorkflow::new("wf-elsewhere")
};
sys.init_workflow(&elsewhere, None, Submission::Fresh, None)
.await
.expect("a deduplication key is scoped to its queue");
sys.init_workflow(&holder, None, Submission::Fresh, None)
.await
.expect("the holder may re-submit itself");
}
#[tokio::test]
async fn authenticated_roles_are_encoded_by_this_layer() {
let (sys, db) = sysdb().await;
let wf = NewWorkflow {
authenticated_roles: vec!["admin", "auditor \"quoted\""],
..NewWorkflow::new("wf-roles")
};
sys.init_workflow(&wf, None, Submission::Fresh, None)
.await
.unwrap();
let read = sys.get_workflow("wf-roles").await.unwrap().unwrap();
assert_eq!(read.authenticated_roles, ["admin", "auditor \"quoted\""]);
let mut conn = db.admin_connection().await;
let stored: Option<String> = sqlx::query_scalar(sqlx::AssertSqlSafe(
"SELECT authenticated_roles FROM dbos.workflow_status WHERE workflow_uuid = 'wf-roles'",
))
.fetch_one(&mut conn)
.await
.unwrap();
assert_eq!(
stored.as_deref(),
Some(r#"["admin","auditor \"quoted\""]"#),
"quotes must be escaped, which is why this is not string concatenation",
);
sys.init_workflow(
&NewWorkflow::new("wf-no-roles"),
None,
Submission::Fresh,
None,
)
.await
.unwrap();
let stored: Option<String> = sqlx::query_scalar(sqlx::AssertSqlSafe(
"SELECT authenticated_roles FROM dbos.workflow_status WHERE workflow_uuid = 'wf-no-roles'",
))
.fetch_one(&mut conn)
.await
.unwrap();
assert_eq!(stored, None);
assert!(
sys.get_workflow("wf-no-roles")
.await
.unwrap()
.unwrap()
.authenticated_roles
.is_empty(),
"a NULL column reads back as no roles",
);
}
#[tokio::test]
async fn a_duplicate_submission_does_not_steal_the_executor_stamp() {
let (sys, _db) = sysdb().await;
let running = NewWorkflow {
executor_id: Some("executor-a"),
..workflow("wf-owned-elsewhere")
};
sys.init_workflow(&running, None, Submission::Fresh, None)
.await
.unwrap();
let duplicate = NewWorkflow {
executor_id: Some("executor-b"),
..workflow("wf-owned-elsewhere")
};
let result = sys
.init_workflow(&duplicate, None, Submission::Fresh, None)
.await
.unwrap();
assert!(
!result.should_execute,
"another owner holds this workflow, so running it would be a second execution",
);
let read = sys
.get_workflow("wf-owned-elsewhere")
.await
.unwrap()
.unwrap();
assert_eq!(
read.executor_id.as_deref(),
Some("executor-a"),
"the stamp must stay with the executor that is actually running it",
);
assert_eq!(
ids(&sys, &pending("executor-a", "v1")).await,
["wf-owned-elsewhere"],
);
assert!(ids(&sys, &pending("executor-b", "v1")).await.is_empty());
let recovering = NewWorkflow {
executor_id: Some("executor-b"),
..workflow("wf-owned-elsewhere")
};
let result = sys
.init_workflow(&recovering, None, Submission::Recovery, None)
.await
.unwrap();
assert!(result.should_execute);
let read = sys
.get_workflow("wf-owned-elsewhere")
.await
.unwrap()
.unwrap();
assert_eq!(read.executor_id.as_deref(), Some("executor-b"));
}
#[tokio::test]
async fn attributes_must_be_a_json_object() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-attr-ok"), None, Submission::Fresh, None)
.await
.unwrap();
for bad in [r#"[1,2,3]"#, "42", r#""a string""#, "not json at all"] {
let wf = NewWorkflow {
attributes: Some(bad),
..NewWorkflow::new("wf-attr-bad")
};
match sys.init_workflow(&wf, None, Submission::Fresh, None).await {
Err(Error::InvalidInput { field, .. }) => assert_eq!(field, "attributes"),
other => panic!("expected {bad} to be rejected at creation, got {other:?}"),
}
match sys
.update_workflow_attributes("wf-attr-ok", Some(bad), None)
.await
{
Err(Error::InvalidInput { field, .. }) => assert_eq!(field, "attributes"),
other => panic!("expected {bad} to be rejected on update, got {other:?}"),
}
}
sys.update_workflow_attributes("wf-attr-ok", Some(r#"{"tenant":"acme"}"#), None)
.await
.unwrap();
sys.update_workflow_attributes("wf-attr-ok", None, None)
.await
.unwrap();
assert!(sys.get_workflow("wf-attr-bad").await.unwrap().is_none());
}
#[tokio::test]
async fn application_versions_are_registered_once_and_ordered_by_timestamp() {
let (sys, _db) = sysdb().await;
assert_eq!(
sys.get_latest_application_version(None).await.unwrap(),
None,
"an empty registry is what a fresh database looks like, not an error",
);
sys.create_application_version("v1", None).await.unwrap();
sys.create_application_version("v2", None).await.unwrap();
sys.create_application_version("v1", None).await.unwrap();
let all = sys.list_application_versions().await.unwrap();
assert_eq!(all.len(), 2);
let latest = sys
.get_latest_application_version(None)
.await
.unwrap()
.unwrap();
assert_eq!(latest.version_name, "v2", "highest timestamp wins");
let promoted = Timestamp::from_epoch_ms(latest.version_timestamp.as_epoch_ms() + 60_000);
sys.update_application_version_timestamp("v1", promoted, None)
.await
.unwrap();
let latest = sys
.get_latest_application_version(None)
.await
.unwrap()
.unwrap();
assert_eq!(latest.version_name, "v1");
assert_eq!(latest.version_timestamp, promoted);
let all = sys.list_application_versions().await.unwrap();
assert_eq!(
all.iter()
.map(|v| v.version_name.as_str())
.collect::<Vec<_>>(),
["v1", "v2"],
"the list is latest-first, on the same ordering",
);
let v1 = &all[0];
assert_ne!(v1.version_id, v1.version_name);
sys.create_application_version("v1", None).await.unwrap();
let again = sys
.get_latest_application_version(None)
.await
.unwrap()
.unwrap();
assert_eq!(
again.version_id, v1.version_id,
"re-registering must not mint a new id or reset the promotion",
);
assert_eq!(again.version_timestamp, promoted);
}
#[tokio::test]
async fn the_bulk_readers_return_notifications_events_and_streams() {
let (sys, db) = sysdb().await;
sys.init_workflow(&workflow("wf-bulk"), None, Submission::Fresh, None)
.await
.unwrap();
let mut conn = db.admin_connection().await;
for sql in [
"INSERT INTO dbos.notifications (message_uuid, destination_uuid, topic, message, \
serialization, created_at_epoch_ms, consumed) \
VALUES ('msg-2', 'wf-bulk', 'orders', '\"second\"', 'portable_json', 2000, false), \
('msg-1', 'wf-bulk', NULL, '\"first\"', NULL, 1000, true)",
"INSERT INTO dbos.workflow_events (workflow_uuid, key, value, serialization) \
VALUES ('wf-bulk', 'progress', '50', 'portable_json'), ('wf-bulk', 'answer', '42', NULL)",
"INSERT INTO dbos.streams (workflow_uuid, key, \"offset\", value, serialization, \
function_id) \
VALUES ('wf-bulk', 'log', 1, '\"b\"', NULL, 7), \
('wf-bulk', 'log', 0, '\"a\"', 'portable_json', 3), \
('wf-bulk', 'audit', 0, '\"x\"', NULL, 3)",
] {
sqlx::query(sqlx::AssertSqlSafe(sql))
.execute(&mut conn)
.await
.unwrap();
}
let notifications = sys.get_all_notifications("wf-bulk").await.unwrap();
assert_eq!(notifications.len(), 2);
assert_eq!(notifications[0].message_uuid, "msg-1");
assert_eq!(notifications[0].message, "\"first\"");
assert_eq!(notifications[0].topic, None, "the default topic is NULL");
assert!(
notifications[0].consumed,
"receiving marks rather than deletes, so this must still be visible",
);
assert_eq!(notifications[1].topic.as_deref(), Some("orders"));
assert_eq!(
notifications[1].serialization.as_deref(),
Some("portable_json")
);
let events = sys.get_all_events("wf-bulk").await.unwrap();
assert_eq!(
events.iter().map(|e| e.key.as_str()).collect::<Vec<_>>(),
["answer", "progress"],
);
assert_eq!(events[0].value, "42");
assert_eq!(events[0].serialization, None);
let entries = sys.get_all_stream_entries("wf-bulk").await.unwrap();
assert_eq!(
entries
.iter()
.map(|e| (e.key.as_str(), e.offset))
.collect::<Vec<_>>(),
[("audit", 0), ("log", 0), ("log", 1)],
);
assert_eq!(entries[1].value, "\"a\"");
assert_eq!(
entries.iter().map(|e| e.step_id).collect::<Vec<_>>(),
[3, 3, 7],
);
sys.init_workflow(&workflow("wf-quiet"), None, Submission::Fresh, None)
.await
.unwrap();
assert!(
sys.get_all_notifications("wf-quiet")
.await
.unwrap()
.is_empty()
);
assert!(sys.get_all_events("wf-quiet").await.unwrap().is_empty());
assert!(
sys.get_all_stream_entries("wf-quiet")
.await
.unwrap()
.is_empty()
);
}
#[tokio::test]
async fn set_event_publishes_a_value_and_its_history() {
let (sys, db) = sysdb().await;
sys.init_workflow(&workflow("wf-publisher"), None, Submission::Fresh, None)
.await
.unwrap();
sys.set_event("wf-publisher", 0, "progress", "50", Some("portable_json"))
.await
.unwrap();
let events = sys.get_all_events("wf-publisher").await.unwrap();
assert_eq!(events.len(), 1);
assert_eq!(events[0].key, "progress");
assert_eq!(events[0].value, "50");
assert_eq!(events[0].serialization.as_deref(), Some("portable_json"));
let step = sys
.check_step("wf-publisher", 0, "DBOS.setEvent")
.await
.unwrap()
.expect("setting an event is a step");
assert_eq!(step.step_name, "DBOS.setEvent");
sys.set_event("wf-publisher", 1, "progress", "100", None)
.await
.unwrap();
let events = sys.get_all_events("wf-publisher").await.unwrap();
assert_eq!(events.len(), 1, "the current value is one row per key");
assert_eq!(events[0].value, "100");
let mut conn = db.admin_connection().await;
let history: Vec<(i32, String)> = sqlx::query_as(sqlx::AssertSqlSafe(
"SELECT function_id, value FROM dbos.workflow_events_history \
WHERE workflow_uuid = 'wf-publisher' AND key = 'progress' ORDER BY function_id",
))
.fetch_all(&mut conn)
.await
.unwrap();
assert_eq!(
history,
[(0, "50".to_owned()), (1, "100".to_owned())],
"history keeps a row per step, which is what a fork copies forward",
);
}
#[tokio::test]
async fn replaying_set_event_does_not_republish() {
let (sys, db) = sysdb().await;
sys.init_workflow(&workflow("wf-replay"), None, Submission::Fresh, None)
.await
.unwrap();
sys.set_event("wf-replay", 0, "answer", "42", None)
.await
.unwrap();
sys.set_event("wf-replay", 0, "answer", "99", None)
.await
.expect("a replay is skipped, not an error");
let events = sys.get_all_events("wf-replay").await.unwrap();
assert_eq!(events[0].value, "42", "the replay must not overwrite");
let mut conn = db.admin_connection().await;
let rows: (i64,) = sqlx::query_as(sqlx::AssertSqlSafe(
"SELECT count(*) FROM dbos.workflow_events_history WHERE workflow_uuid = 'wf-replay'",
))
.fetch_one(&mut conn)
.await
.unwrap();
assert_eq!(rows.0, 1, "the replay added no history row either");
}
const RECHECK: std::time::Duration = std::time::Duration::from_secs(1);
async fn publisher_and_reader(sys: &PostgresSystemDatabase, publisher: &str, reader: &str) {
for id in [publisher, reader] {
sys.init_workflow(&workflow(id), None, Submission::Fresh, None)
.await
.unwrap();
}
}
#[tokio::test]
async fn an_event_already_published_returns_at_once() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-publisher"), None, Submission::Fresh, None)
.await
.unwrap();
sys.set_event("wf-publisher", 0, "progress", "50", Some("portable_json"))
.await
.unwrap();
let began = Timestamp::now();
let found = sys
.get_event("wf-publisher", "progress", RECHECK * 30, None)
.await
.unwrap();
assert_eq!(
found,
Some(EncodedValue {
value: "50".to_owned(),
serialization: Some("portable_json".to_owned()),
})
);
assert!(
Timestamp::now().duration_since(began).unwrap() < RECHECK,
"a value already there should not cost an interval",
);
}
#[tokio::test]
async fn a_value_published_during_the_wait_still_arrives() {
let db = test_database().await;
let sys = std::sync::Arc::new(PostgresSystemDatabase::from_pool(
db.pool().await,
&Settings::default(),
));
let publisher = PostgresSystemDatabase::from_pool(db.pool().await, &Settings::default());
publisher_and_reader(&sys, "wf-publisher", "wf-reader").await;
let read = sys.get_event("wf-publisher", "progress", RECHECK * 600, None);
let mut read = std::pin::pin!(read);
tokio::select! {
answered = &mut read => {
panic!("the read answered before anything was published: {answered:?}")
}
() = tokio::time::sleep(RECHECK * 2) => {}
}
let mut probe = db.admin_connection().await;
let present: (i64,) = sqlx::query_as(sqlx::AssertSqlSafe(
"SELECT count(*) FROM dbos.workflow_events \
WHERE workflow_uuid = 'wf-publisher' AND key = 'progress'",
))
.fetch_one(&mut probe)
.await
.unwrap();
assert_eq!(
present.0, 0,
"the value was already published, so the first look could have answered the read",
);
let began = Timestamp::now();
publisher
.set_event(
"wf-publisher",
0,
"progress",
"\"done\"",
Some("portable_json"),
)
.await
.unwrap();
let found = tokio::time::timeout(RECHECK * 10, read)
.await
.expect("the wait loop never delivered a value that was published during it")
.unwrap();
assert!(
Timestamp::now().duration_since(began).unwrap() < RECHECK * 5,
"delivered, but far slower than the interval it is supposed to run at",
);
assert_eq!(
found,
Some(EncodedValue {
value: "\"done\"".to_owned(),
serialization: Some("portable_json".to_owned()),
})
);
}
#[tokio::test]
async fn a_read_that_finds_nothing_reports_absence() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-publisher"), None, Submission::Fresh, None)
.await
.unwrap();
let found = sys
.get_event("wf-publisher", "never-set", BRIEFLY, None)
.await
.unwrap();
assert_eq!(found, None);
let found = sys
.get_event("wf-nobody", "progress", BRIEFLY, None)
.await
.unwrap();
assert_eq!(found, None);
}
#[tokio::test]
async fn a_read_records_the_publishers_own_payload() {
let (sys, _db) = sysdb().await;
publisher_and_reader(&sys, "wf-publisher", "wf-reader").await;
sys.set_event("wf-publisher", 0, "progress", "50", Some("pickle"))
.await
.unwrap();
let caller = GetEventCaller {
workflow_id: "wf-reader",
step_id: 0,
timeout_step_id: 1,
};
sys.get_event("wf-publisher", "progress", RECHECK * 30, Some(caller))
.await
.unwrap();
let step = sys
.check_step("wf-reader", 0, "DBOS.getEvent")
.await
.unwrap()
.expect("reading an event inside a workflow is a step");
assert_eq!(step.output.as_deref(), Some("50"));
assert_eq!(
step.serialization.as_deref(),
Some("pickle"),
"the publisher's format travels with the value, not this layer's",
);
}
#[tokio::test]
async fn a_replayed_read_returns_what_the_first_run_saw() {
let (sys, _db) = sysdb().await;
publisher_and_reader(&sys, "wf-publisher", "wf-reader").await;
sys.set_event("wf-publisher", 0, "progress", "50", None)
.await
.unwrap();
let caller = GetEventCaller {
workflow_id: "wf-reader",
step_id: 0,
timeout_step_id: 1,
};
let first = sys
.get_event("wf-publisher", "progress", RECHECK * 30, Some(caller))
.await
.unwrap();
assert_eq!(first.as_ref().map(|v| v.value.as_str()), Some("50"));
sys.set_event("wf-publisher", 1, "progress", "100", None)
.await
.unwrap();
let replayed = sys
.get_event("wf-publisher", "progress", RECHECK * 30, Some(caller))
.await
.unwrap();
assert_eq!(
replayed, first,
"a replay must not see a value that arrived after the decision it fed",
);
}
#[tokio::test]
async fn a_replayed_timeout_stays_a_timeout() {
let (sys, _db) = sysdb().await;
publisher_and_reader(&sys, "wf-publisher", "wf-reader").await;
let caller = GetEventCaller {
workflow_id: "wf-reader",
step_id: 0,
timeout_step_id: 1,
};
assert_eq!(
sys.get_event("wf-publisher", "progress", BRIEFLY, Some(caller))
.await
.unwrap(),
None,
);
let began = Timestamp::now();
assert_eq!(
tokio::time::timeout(
RECHECK * 5,
sys.get_event("wf-publisher", "progress", RECHECK * 30, Some(caller)),
)
.await
.expect("a replay waited instead of returning the answer it had already recorded")
.unwrap(),
None,
);
assert!(
Timestamp::now().duration_since(began).unwrap() < RECHECK,
"a replay must not wait at all",
);
sys.set_event("wf-publisher", 0, "progress", "50", None)
.await
.unwrap();
assert_eq!(
sys.get_event("wf-publisher", "progress", RECHECK * 30, Some(caller))
.await
.unwrap(),
None,
"the recorded absence is the answer, not a fresh look",
);
}
#[tokio::test]
async fn a_replayed_read_does_not_depend_on_the_row_still_existing() {
let (sys, db) = sysdb().await;
publisher_and_reader(&sys, "wf-publisher", "wf-reader").await;
sys.set_event("wf-publisher", 0, "progress", "50", None)
.await
.unwrap();
let caller = GetEventCaller {
workflow_id: "wf-reader",
step_id: 0,
timeout_step_id: 1,
};
let timeout = std::time::Duration::from_secs(600);
let first = sys
.get_event("wf-publisher", "progress", timeout, Some(caller))
.await
.unwrap();
assert_eq!(first.as_ref().map(|v| v.value.as_str()), Some("50"));
let mut conn = db.admin_connection().await;
sqlx::query(sqlx::AssertSqlSafe(
"DELETE FROM dbos.workflow_events WHERE workflow_uuid = 'wf-publisher'",
))
.execute(&mut conn)
.await
.unwrap();
let replayed = tokio::time::timeout(
RECHECK * 10,
sys.get_event("wf-publisher", "progress", timeout, Some(caller)),
)
.await
.expect("a replay went looking for a row it had already read")
.unwrap();
assert_eq!(replayed, first);
}
#[tokio::test]
async fn a_read_adopts_a_rivals_answer_rather_than_failing() {
let db = test_database().await;
let sys = std::sync::Arc::new(PostgresSystemDatabase::from_pool(
db.pool().await,
&Settings::default(),
));
publisher_and_reader(&sys, "wf-publisher", "wf-reader").await;
let reading = {
let sys = std::sync::Arc::clone(&sys);
tokio::spawn(async move {
sys.get_event(
"wf-publisher",
"progress",
RECHECK * 30,
Some(GetEventCaller {
workflow_id: "wf-reader",
step_id: 0,
timeout_step_id: 1,
}),
)
.await
})
};
tokio::time::sleep(BRIEFLY).await;
sys.record_step(
"wf-reader",
0,
"DBOS.getEvent",
Outcome::Output(Some("\"rival\"")),
Some("portable_json"),
None,
)
.await
.unwrap();
sys.set_event("wf-publisher", 0, "progress", "\"mine\"", None)
.await
.unwrap();
let found = tokio::time::timeout(RECHECK * 20, reading)
.await
.expect("the read never finished")
.unwrap()
.expect("losing the race is not a failure");
assert_eq!(
found,
Some(EncodedValue {
value: "\"rival\"".to_owned(),
serialization: Some("portable_json".to_owned()),
}),
"the recorded step is the workflow's answer, whoever wrote it",
);
}
#[tokio::test]
async fn a_read_records_a_deadline_it_may_abandon_rather_than_a_sleep() {
let (sys, _db) = sysdb().await;
publisher_and_reader(&sys, "wf-publisher", "wf-reader").await;
sys.set_event("wf-publisher", 0, "progress", "50", None)
.await
.unwrap();
let timeout = std::time::Duration::from_secs(600);
sys.get_event(
"wf-publisher",
"progress",
timeout,
Some(GetEventCaller {
workflow_id: "wf-reader",
step_id: 0,
timeout_step_id: 1,
}),
)
.await
.unwrap();
let sleep = sys
.check_step("wf-reader", 1, "DBOS.sleep")
.await
.unwrap()
.expect("the deadline is checkpointed even when the value is already there");
let started = sleep.started_at.unwrap();
let completed = sleep.completed_at.unwrap();
let deadline: i64 = sleep.output.unwrap().parse().unwrap();
assert!(
deadline - started.as_epoch_ms() >= timeout.as_millis() as i64,
"the recorded deadline is still ten minutes out; only the completion differs",
);
assert!(
completed.duration_since(started).unwrap() < RECHECK,
"a deadline is stamped complete now, not at the deadline: \
started {started:?}, completed {completed:?}",
);
}
#[tokio::test]
async fn a_recovered_read_resumes_the_original_deadline() {
let (sys, _db) = sysdb().await;
publisher_and_reader(&sys, "wf-publisher", "wf-reader").await;
sys.record_sleep("wf-reader", 1, std::time::Duration::ZERO)
.await
.unwrap();
let began = Timestamp::now();
let found = tokio::time::timeout(
RECHECK * 10,
sys.get_event(
"wf-publisher",
"progress",
std::time::Duration::from_secs(600),
Some(GetEventCaller {
workflow_id: "wf-reader",
step_id: 0,
timeout_step_id: 1,
}),
),
)
.await
.expect("the recovered deadline had already passed; the timeout was restarted instead")
.unwrap();
assert_eq!(found, None);
assert!(
Timestamp::now().duration_since(began).unwrap() < RECHECK * 5,
"the recovered deadline had already passed; the timeout was restarted instead",
);
}
#[tokio::test]
async fn a_cancelled_reader_stops_rather_than_waiting() {
let (sys, _db) = sysdb().await;
publisher_and_reader(&sys, "wf-publisher", "wf-reader").await;
sys.cancel_workflows(&["wf-reader"], false, None)
.await
.unwrap();
let err = tokio::time::timeout(
RECHECK * 10,
sys.get_event(
"wf-publisher",
"progress",
std::time::Duration::from_secs(600),
Some(GetEventCaller {
workflow_id: "wf-reader",
step_id: 0,
timeout_step_id: 1,
}),
),
)
.await
.expect("a cancelled reader waited out its timeout instead of stopping")
.expect_err("a cancelled workflow finds out at its next step boundary");
assert!(
matches!(err, Error::WorkflowCancelled { ref workflow_id } if workflow_id == "wf-reader"),
"unexpected error: {err:?}",
);
}
#[tokio::test]
async fn a_capped_read_does_not_hold_its_permit_across_the_wait() {
let db = test_database().await;
let sys = std::sync::Arc::new(PostgresSystemDatabase::from_pool(
db.pool().await,
&Settings {
polling_concurrency: Some(1),
..Settings::default()
},
));
sys.init_workflow(&workflow("wf-publisher"), None, Submission::Fresh, None)
.await
.unwrap();
sys.set_event("wf-publisher", 0, "published", "50", None)
.await
.unwrap();
let forever = {
let sys = std::sync::Arc::clone(&sys);
tokio::spawn(async move {
sys.get_event("wf-publisher", "never-published", RECHECK * 60, None)
.await
})
};
tokio::time::sleep(BRIEFLY).await;
let found = tokio::time::timeout(
RECHECK * 10,
sys.get_event("wf-publisher", "published", RECHECK * 30, None),
)
.await
.expect("starved: a polling permit is being held across a wait")
.unwrap();
assert_eq!(found.map(|v| v.value), Some("50".to_owned()));
assert!(!forever.is_finished());
forever.abort();
}
async fn send_to_receiver(sys: &PostgresSystemDatabase, topic: Option<&str>, message: &str) {
sys.send_message(
&Message {
destination_id: "wf-receiver",
topic,
message,
idempotency_key: None,
},
Some("portable_json"),
None,
false,
)
.await
.unwrap();
}
#[tokio::test]
async fn a_message_already_waiting_is_taken_at_once() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-receiver"), None, Submission::Fresh, None)
.await
.unwrap();
send_to_receiver(&sys, Some("orders"), "\"one\"").await;
let began = Timestamp::now();
let taken = sys
.recv("wf-receiver", 0, 1, Some("orders"), RECHECK * 600)
.await
.unwrap();
assert_eq!(
taken,
Some(EncodedValue {
value: "\"one\"".to_owned(),
serialization: Some("portable_json".to_owned()),
})
);
assert!(
Timestamp::now().duration_since(began).unwrap() < RECHECK,
"a message already there should not cost an interval",
);
let notifications = sys.get_all_notifications("wf-receiver").await.unwrap();
assert_eq!(notifications.len(), 1);
assert!(notifications[0].consumed, "receiving consumes the message");
let step = sys
.check_step("wf-receiver", 0, "DBOS.recv")
.await
.unwrap()
.expect("receiving is a step");
assert_eq!(step.output.as_deref(), Some("\"one\""));
assert_eq!(
step.serialization.as_deref(),
Some("portable_json"),
"the sender's format travels with the message",
);
}
#[tokio::test]
async fn a_message_sent_during_the_wait_still_arrives() {
let db = test_database().await;
let sys = std::sync::Arc::new(PostgresSystemDatabase::from_pool(
db.pool().await,
&Settings::default(),
));
sys.init_workflow(&workflow("wf-receiver"), None, Submission::Fresh, None)
.await
.unwrap();
let sending = {
let sys = std::sync::Arc::clone(&sys);
tokio::spawn(async move {
tokio::time::sleep(BRIEFLY).await;
send_to_receiver(&sys, None, "\"late\"").await;
})
};
let began = Timestamp::now();
let taken = tokio::time::timeout(
RECHECK * 10,
sys.recv("wf-receiver", 0, 1, None, RECHECK * 600),
)
.await
.expect("the wait loop never delivered a message that was sent during it")
.unwrap();
assert!(
Timestamp::now().duration_since(began).unwrap() < RECHECK * 5,
"delivered, but far slower than the interval it is supposed to run at",
);
sending.await.unwrap();
assert_eq!(taken.map(|m| m.value), Some("\"late\"".to_owned()));
}
#[tokio::test]
async fn a_recv_that_finds_nothing_reports_absence() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-receiver"), None, Submission::Fresh, None)
.await
.unwrap();
assert_eq!(
sys.recv("wf-receiver", 0, 1, Some("orders"), BRIEFLY)
.await
.unwrap(),
None,
);
let step = sys
.check_step("wf-receiver", 0, "DBOS.recv")
.await
.unwrap()
.expect("a timeout is a result, and results are recorded");
assert_eq!(step.output, None, "nothing received is a NULL output");
send_to_receiver(&sys, None, "\"untopicked\"").await;
assert_eq!(
sys.recv("wf-receiver", 2, 3, Some("orders"), BRIEFLY)
.await
.unwrap(),
None,
"a message on the default topic is not a message on `orders`",
);
}
#[tokio::test]
async fn messages_are_taken_oldest_first() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-receiver"), None, Submission::Fresh, None)
.await
.unwrap();
for message in ["\"one\"", "\"two\"", "\"three\""] {
send_to_receiver(&sys, Some("orders"), message).await;
}
let mut taken = Vec::new();
for step_id in [0, 2, 4] {
taken.push(
sys.recv(
"wf-receiver",
step_id,
step_id + 1,
Some("orders"),
RECHECK * 600,
)
.await
.unwrap()
.map(|m| m.value),
);
}
assert_eq!(
taken,
[
Some("\"one\"".to_owned()),
Some("\"two\"".to_owned()),
Some("\"three\"".to_owned()),
],
"FIFO, and no message twice",
);
}
#[tokio::test]
async fn a_replayed_recv_does_not_take_a_second_message() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-receiver"), None, Submission::Fresh, None)
.await
.unwrap();
send_to_receiver(&sys, Some("orders"), "\"one\"").await;
send_to_receiver(&sys, Some("orders"), "\"two\"").await;
let first = sys
.recv("wf-receiver", 0, 1, Some("orders"), RECHECK * 600)
.await
.unwrap();
let replayed = sys
.recv("wf-receiver", 0, 1, Some("orders"), RECHECK * 600)
.await
.unwrap();
assert_eq!(replayed, first, "a replay returns, it does not receive");
let consumed = sys
.get_all_notifications("wf-receiver")
.await
.unwrap()
.iter()
.filter(|n| n.consumed)
.count();
assert_eq!(consumed, 1, "the replay must not have taken the second one");
}
#[tokio::test]
async fn a_replayed_recv_does_not_wait_for_a_message_it_already_has() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-receiver"), None, Submission::Fresh, None)
.await
.unwrap();
send_to_receiver(&sys, Some("orders"), "\"one\"").await;
let timeout = RECHECK * 600;
let first = sys
.recv("wf-receiver", 0, 1, Some("orders"), timeout)
.await
.unwrap();
assert_eq!(first.as_ref().map(|m| m.value.as_str()), Some("\"one\""));
let began = Timestamp::now();
let replayed = tokio::time::timeout(
RECHECK * 10,
sys.recv("wf-receiver", 0, 1, Some("orders"), timeout),
)
.await
.expect("a replay waited for a message it had already received")
.unwrap();
assert_eq!(replayed, first);
assert!(
Timestamp::now().duration_since(began).unwrap() < RECHECK,
"a replay must not wait at all",
);
}
#[tokio::test]
async fn a_recv_defers_to_a_rival_that_recorded_first() {
let db = test_database().await;
let sys = std::sync::Arc::new(PostgresSystemDatabase::from_pool(
db.pool().await,
&Settings::default(),
));
sys.init_workflow(&workflow("wf-receiver"), None, Submission::Fresh, None)
.await
.unwrap();
let receiving = {
let sys = std::sync::Arc::clone(&sys);
tokio::spawn(async move {
sys.recv("wf-receiver", 0, 1, Some("orders"), RECHECK * 600)
.await
})
};
tokio::time::sleep(BRIEFLY).await;
sys.record_step(
"wf-receiver",
0,
"DBOS.recv",
Outcome::Output(Some("\"rival\"")),
Some("portable_json"),
None,
)
.await
.unwrap();
send_to_receiver(&sys, Some("orders"), "\"mine\"").await;
let taken = tokio::time::timeout(RECHECK * 20, receiving)
.await
.expect("the receive never finished")
.unwrap()
.expect("losing to a rival is not a failure");
assert_eq!(
taken.map(|m| m.value),
Some("\"rival\"".to_owned()),
"the recorded step is the workflow's answer, whoever wrote it",
);
let notifications = sys.get_all_notifications("wf-receiver").await.unwrap();
assert!(
notifications.iter().all(|n| !n.consumed),
"the message must still be on the queue: nothing recorded having taken it",
);
}
#[tokio::test]
async fn a_second_receiver_on_one_topic_is_refused() {
let db = test_database().await;
let sys = std::sync::Arc::new(PostgresSystemDatabase::from_pool(
db.pool().await,
&Settings::default(),
));
sys.init_workflow(&workflow("wf-receiver"), None, Submission::Fresh, None)
.await
.unwrap();
let waiting = {
let sys = std::sync::Arc::clone(&sys);
tokio::spawn(async move {
sys.recv("wf-receiver", 0, 1, Some("orders"), RECHECK * 600)
.await
})
};
tokio::time::sleep(BRIEFLY).await;
let err = tokio::time::timeout(
RECHECK * 10,
sys.recv("wf-receiver", 2, 3, Some("orders"), RECHECK * 600),
)
.await
.expect("a second receiver was admitted and left to wait rather than refused")
.expect_err("a second receiver on one topic is a bug in the caller, and is told so");
assert!(
matches!(
err,
Error::ConcurrentRecv { ref workflow_id, ref topic }
if workflow_id == "wf-receiver" && topic.as_deref() == Some("orders")
),
"unexpected error: {err:?}",
);
send_to_receiver(&sys, Some("refunds"), "\"other\"").await;
assert!(
sys.recv("wf-receiver", 4, 5, Some("refunds"), RECHECK * 600)
.await
.unwrap()
.is_some(),
);
waiting.abort();
}
#[tokio::test]
async fn two_receivers_cannot_take_the_same_message() {
let db = test_database().await;
let first = std::sync::Arc::new(PostgresSystemDatabase::from_pool(
db.pool().await,
&Settings::default(),
));
let second = std::sync::Arc::new(PostgresSystemDatabase::from_pool(
db.pool().await,
&Settings::default(),
));
first
.init_workflow(&workflow("wf-receiver"), None, Submission::Fresh, None)
.await
.unwrap();
send_to_receiver(&first, Some("orders"), "\"one\"").await;
send_to_receiver(&first, Some("orders"), "\"two\"").await;
let racing: Vec<_> = [first.clone(), second.clone()]
.into_iter()
.map(|sys| {
tokio::spawn(async move {
sys.recv("wf-receiver", 0, 1, Some("orders"), RECHECK * 600)
.await
})
})
.collect();
let mut answers = Vec::new();
for handle in racing {
answers.push(
tokio::time::timeout(RECHECK * 20, handle)
.await
.expect("a racing receiver never finished")
.unwrap(),
);
}
let (delivered, refused): (Vec<_>, Vec<_>) = answers.iter().partition(|answer| answer.is_ok());
assert!(
refused.len() <= 1
&& refused
.iter()
.all(|answer| matches!(answer, Err(Error::StepAlreadyRecorded { step_id: 0, .. }))),
"at most one execution may lose the race, and only on its step write: {answers:?}",
);
let mut received = delivered.iter().map(|answer| answer.as_ref().unwrap());
let first_answer = received.next().expect("nobody received the message");
assert!(
first_answer.is_some(),
"a message was waiting, so the execution that recorded the step must hold it",
);
assert!(
received.all(|answer| answer == first_answer),
"an execution that adopted the recorded step must report what that step holds: {answers:?}",
);
let notifications = first.get_all_notifications("wf-receiver").await.unwrap();
assert_eq!(
notifications.iter().filter(|n| n.consumed).count(),
1,
"one message was received, so exactly one may be consumed: {notifications:?}",
);
}
#[tokio::test]
async fn a_cancelled_receiver_stops_rather_than_waiting() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-receiver"), None, Submission::Fresh, None)
.await
.unwrap();
sys.cancel_workflows(&["wf-receiver"], false, None)
.await
.unwrap();
let err = tokio::time::timeout(
RECHECK * 10,
sys.recv("wf-receiver", 0, 1, Some("orders"), RECHECK * 600),
)
.await
.expect("a cancelled receiver waited out its timeout instead of stopping")
.expect_err("a cancelled workflow finds out at its next step boundary");
assert!(
matches!(err, Error::WorkflowCancelled { ref workflow_id } if workflow_id == "wf-receiver"),
"unexpected error: {err:?}",
);
}
#[tokio::test]
async fn a_recovered_recv_resumes_the_original_deadline() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-receiver"), None, Submission::Fresh, None)
.await
.unwrap();
sys.record_sleep("wf-receiver", 1, std::time::Duration::ZERO)
.await
.unwrap();
let began = Timestamp::now();
let taken = tokio::time::timeout(
RECHECK * 10,
sys.recv("wf-receiver", 0, 1, Some("orders"), RECHECK * 600),
)
.await
.expect("the recovered deadline had already passed; the timeout was restarted instead")
.unwrap();
assert_eq!(taken, None);
assert!(
Timestamp::now().duration_since(began).unwrap() < RECHECK * 5,
"the recovered deadline had already passed; the timeout was restarted instead",
);
}
#[tokio::test]
async fn a_capped_recv_does_not_hold_its_permit_across_the_wait() {
let db = test_database().await;
let sys = std::sync::Arc::new(PostgresSystemDatabase::from_pool(
db.pool().await,
&Settings {
polling_concurrency: Some(1),
..Settings::default()
},
));
for id in ["wf-receiver", "wf-other"] {
sys.init_workflow(&workflow(id), None, Submission::Fresh, None)
.await
.unwrap();
}
sys.send_message(
&Message {
destination_id: "wf-other",
topic: None,
message: "\"here\"",
idempotency_key: None,
},
Some("portable_json"),
None,
false,
)
.await
.unwrap();
let forever = {
let sys = std::sync::Arc::clone(&sys);
tokio::spawn(async move {
sys.recv("wf-receiver", 0, 1, Some("never-sent"), RECHECK * 600)
.await
})
};
tokio::time::sleep(BRIEFLY).await;
let taken = tokio::time::timeout(
RECHECK * 10,
sys.recv("wf-other", 0, 1, None, RECHECK * 600),
)
.await
.expect("starved: a polling permit is being held across a wait")
.unwrap();
assert_eq!(taken.map(|m| m.value), Some("\"here\"".to_owned()));
assert!(!forever.is_finished());
forever.abort();
}
fn skip_without_listen_notify(db: &support::TestDatabase) -> bool {
let skipping = matches!(db.backend(), Backend::Cockroach);
if skipping {
eprintln!(
"skipped on CockroachDB: it has no LISTEN/NOTIFY, so there is no listener to test"
);
}
skipping
}
async fn listening(pool: sqlx::PgPool) -> PostgresSystemDatabase {
let sys = PostgresSystemDatabase::from_pool(pool, &Settings::default());
assert!(
sys.start_notifications().await,
"PostgreSQL supports LISTEN/NOTIFY, so one should have started",
);
for _ in 0..100 {
if sys.is_delivering() {
return sys;
}
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
}
panic!("the listener never proved it delivers");
}
#[tokio::test]
async fn a_listener_delivers_a_message_sooner_than_the_interval_allows() {
let db = test_database().await;
if skip_without_listen_notify(&db) {
return;
}
let sys = std::sync::Arc::new(listening(db.pool().await).await);
sys.init_workflow(&workflow("wf-receiver"), None, Submission::Fresh, None)
.await
.unwrap();
let receiving = {
let sys = std::sync::Arc::clone(&sys);
tokio::spawn(async move {
sys.recv("wf-receiver", 0, 1, Some("orders"), RECHECK * 600)
.await
})
};
tokio::time::sleep(BRIEFLY).await;
let sent = Timestamp::now();
send_to_receiver(&sys, Some("orders"), "\"pushed\"").await;
let taken = tokio::time::timeout(RECHECK * 20, receiving)
.await
.expect("the message never arrived")
.unwrap()
.unwrap();
assert_eq!(taken.map(|m| m.value), Some("\"pushed\"".to_owned()));
let waited = Timestamp::now().duration_since(sent).unwrap();
assert!(
waited < RECHECK,
"took {waited:?} after the send, which is the re-query rather than the notification",
);
}
#[tokio::test]
async fn a_message_that_notified_nobody_is_found_on_reconnect() {
const TAG: &str = "listener-outage";
let db = test_database().await;
if skip_without_listen_notify(&db) {
return;
}
let pool = db
.pool_options()
.connect_with(db.options().application_name(TAG))
.await
.expect("failed to connect");
let sys = std::sync::Arc::new(listening(pool).await);
sys.init_workflow(&workflow("wf-receiver"), None, Submission::Fresh, None)
.await
.unwrap();
let receiving = {
let sys = std::sync::Arc::clone(&sys);
tokio::spawn(async move {
sys.recv("wf-receiver", 0, 1, Some("orders"), RECHECK * 600)
.await
})
};
tokio::time::sleep(BRIEFLY).await;
let mut conn = db.admin_connection().await;
sqlx::query(sqlx::AssertSqlSafe(
"SET session_replication_role = replica",
))
.execute(&mut conn)
.await
.unwrap();
sqlx::query(sqlx::AssertSqlSafe(
"INSERT INTO dbos.notifications (destination_uuid, topic, message, serialization) \
VALUES ('wf-receiver', 'orders', '\"unannounced\"', 'portable_json')",
))
.execute(&mut conn)
.await
.unwrap();
db.kill_connections(TAG).await;
let taken = tokio::time::timeout(RECHECK * 30, receiving)
.await
.expect("the message was never found: nothing woke the waiter after the gap")
.unwrap()
.unwrap();
assert_eq!(taken.map(|m| m.value), Some("\"unannounced\"".to_owned()));
}
#[tokio::test]
async fn closing_the_handle_stops_the_listener() {
let db = test_database().await;
if skip_without_listen_notify(&db) {
return;
}
let sys = listening(db.pool().await).await;
assert!(sys.is_delivering());
tokio::time::timeout(RECHECK * 10, sys.close())
.await
.expect("closing hung: the listener never released its connection");
for _ in 0..100 {
if !sys.is_delivering() {
return;
}
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
}
panic!("the listener outlived the handle that owns it");
}
#[tokio::test]
async fn cockroach_gets_no_listener_and_reports_it() {
let db = test_database().await;
let sys = PostgresSystemDatabase::from_pool(db.pool().await, &Settings::default());
let started = sys.start_notifications().await;
match db.backend() {
Backend::Cockroach => {
assert!(!started, "CockroachDB has no LISTEN/NOTIFY to start one on");
assert!(!sys.is_delivering());
}
Backend::Postgres => assert!(started),
}
}
#[tokio::test]
async fn an_event_set_elsewhere_arrives_sooner_than_the_interval_allows() {
let db = test_database().await;
if skip_without_listen_notify(&db) {
return;
}
let reader = std::sync::Arc::new(listening(db.pool().await).await);
let writer = PostgresSystemDatabase::from_pool(db.pool().await, &Settings::default());
assert!(writer.start_notifications().await);
assert!(
writer.is_pushing(),
"a writer that pushes nothing wakes nobody"
);
writer
.init_workflow(&workflow("wf-publisher"), None, Submission::Fresh, None)
.await
.unwrap();
let reading = {
let reader = std::sync::Arc::clone(&reader);
tokio::spawn(async move {
reader
.get_event("wf-publisher", "ready", RECHECK * 600, None)
.await
})
};
tokio::time::sleep(BRIEFLY).await;
let published = Timestamp::now();
writer
.set_event("wf-publisher", 0, "ready", "\"go\"", Some("portable_json"))
.await
.unwrap();
let found = tokio::time::timeout(RECHECK * 20, reading)
.await
.expect("the event never arrived")
.unwrap()
.unwrap();
assert_eq!(found.map(|v| v.value), Some("\"go\"".to_owned()));
let waited = Timestamp::now().duration_since(published).unwrap();
assert!(
waited < RECHECK,
"took {waited:?} after the write, which is a re-query rather than the push",
);
}
#[tokio::test]
async fn a_stream_write_is_pushed_where_the_trigger_used_to_publish() {
let db = test_database().await;
if skip_without_listen_notify(&db) {
return;
}
let mut wire = sqlx::postgres::PgListener::connect_with(&db.pool().await)
.await
.expect("failed to listen");
wire.listen("dbos_streams_channel").await.unwrap();
let sys = PostgresSystemDatabase::from_pool(db.pool().await, &Settings::default());
assert!(sys.start_notifications().await);
sys.init_workflow(&workflow("wf-producer"), None, Submission::Fresh, None)
.await
.unwrap();
sys.write_stream(
"wf-producer",
0,
"progress",
"\"a\"",
Some("portable_json"),
WrittenBy::Workflow,
)
.await
.unwrap();
let notification = tokio::time::timeout(RECHECK * 5, wire.recv())
.await
.expect("nothing was pushed for the write")
.unwrap();
assert_eq!(notification.payload(), "wf-producer::progress");
}
#[tokio::test]
async fn a_close_pushes_what_the_window_was_still_holding() {
let db = test_database().await;
if skip_without_listen_notify(&db) {
return;
}
let mut wire = sqlx::postgres::PgListener::connect_with(&db.pool().await)
.await
.expect("failed to listen");
wire.listen("dbos_workflow_events_channel").await.unwrap();
let sys = PostgresSystemDatabase::from_pool(
db.pool().await,
&Settings {
notification_coalesce: Some(RECHECK * 5),
..Settings::default()
},
);
assert!(sys.start_notifications().await);
sys.init_workflow(&workflow("wf-publisher"), None, Submission::Fresh, None)
.await
.unwrap();
sys.set_event("wf-publisher", 0, "ready", "\"go\"", Some("portable_json"))
.await
.unwrap();
sys.close().await;
let notification = tokio::time::timeout(RECHECK * 3, wire.recv())
.await
.expect("the close dropped what the window was holding")
.unwrap();
assert_eq!(notification.payload(), "wf-publisher::ready");
}
#[tokio::test]
async fn a_write_wakes_a_waiter_in_its_own_process_with_nothing_pushing() {
let (sys, _db) = sysdb().await;
let sys = std::sync::Arc::new(sys);
assert!(
!sys.is_pushing(),
"no listener was started, so nothing pushes"
);
sys.init_workflow(&workflow("wf-publisher"), None, Submission::Fresh, None)
.await
.unwrap();
let reading = {
let sys = std::sync::Arc::clone(&sys);
tokio::spawn(async move {
sys.get_event("wf-publisher", "ready", RECHECK * 600, None)
.await
})
};
tokio::time::sleep(BRIEFLY).await;
let published = Timestamp::now();
sys.set_event("wf-publisher", 0, "ready", "\"go\"", Some("portable_json"))
.await
.unwrap();
let found = tokio::time::timeout(RECHECK * 20, reading)
.await
.expect("the event never arrived")
.unwrap()
.unwrap();
assert_eq!(found.map(|v| v.value), Some("\"go\"".to_owned()));
let waited = Timestamp::now().duration_since(published).unwrap();
assert!(
waited < RECHECK / 2,
"took {waited:?} after the write, which is the re-query rather than the wake",
);
}
#[tokio::test]
async fn a_replayed_sleep_keeps_its_original_wake_time() {
let (sys, db) = sysdb().await;
sys.init_workflow(&workflow("wf-sleeper"), None, Submission::Fresh, None)
.await
.unwrap();
let before = Timestamp::now();
let wake = sys
.record_sleep("wf-sleeper", 0, std::time::Duration::from_secs(3600))
.await
.unwrap();
let offset = wake.as_epoch_ms() - before.as_epoch_ms();
assert!(
(3_600_000..3_610_000).contains(&offset),
"expected ~1h from now, got {offset}ms",
);
let replayed = sys
.record_sleep("wf-sleeper", 0, std::time::Duration::from_secs(3600))
.await
.unwrap();
assert_eq!(replayed, wake, "a replay must not restart the clock");
let mut conn = db.admin_connection().await;
let (output, serialization): (Option<String>, Option<String>) =
sqlx::query_as(sqlx::AssertSqlSafe(
"SELECT output, serialization FROM dbos.operation_outputs \
WHERE workflow_uuid = 'wf-sleeper' AND function_id = 0",
))
.fetch_one(&mut conn)
.await
.unwrap();
assert_eq!(
output.as_deref(),
Some(wake.as_epoch_ms().to_string().as_str())
);
assert_eq!(serialization.as_deref(), Some("portable_json"));
}
#[tokio::test]
async fn a_sleep_records_its_duration_as_the_sleep() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-waiter"), None, Submission::Fresh, None)
.await
.unwrap();
let before = Timestamp::now();
let wake = sys
.record_sleep("wf-waiter", 0, std::time::Duration::from_secs(60))
.await
.unwrap();
let step = sys
.check_step("wf-waiter", 0, "DBOS.sleep")
.await
.unwrap()
.unwrap();
assert_eq!(step.completed_at, Some(wake));
assert!(
step.completed_at.unwrap().as_epoch_ms() > before.as_epoch_ms(),
"the completion is stamped ahead of now, which is the point",
);
let started = step.started_at.expect("a sleep records when it began");
assert!(started.as_epoch_ms() >= before.as_epoch_ms());
assert_eq!(
wake.as_epoch_ms() - started.as_epoch_ms(),
60_000,
"the recorded span is exactly the sleep",
);
}
#[tokio::test]
async fn operations_after_close_fail_rather_than_hang() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-closed"), None, Submission::Fresh, None)
.await
.unwrap();
sys.close().await;
let answered = tokio::time::timeout(
std::time::Duration::from_secs(5),
sys.get_workflow("wf-closed"),
)
.await
.expect("a closed pool must answer, not block");
match answered {
Err(Error::Backend(e)) => assert_eq!(
e.kind,
BackendErrorKind::Permanent,
"a closed pool cannot reopen, so waiting for it is waiting forever"
),
other => panic!("expected a backend failure, got {other:?}"),
}
}
#[tokio::test]
async fn connecting_without_migrating_refuses_a_schema_that_is_not_ready() {
let db = test_database().await;
let config = Config {
url: db.url(),
migrate: false,
settings: Settings {
schema: "never_migrated",
..Settings::default()
},
..Config::new(db.url())
};
let message = match PostgresSystemDatabase::connect(&config).await {
Err(e) => e.to_string(),
Ok(_) => panic!("an unmigrated schema should not connect"),
};
assert!(
message.contains("migration 0") && message.contains("migration enabled"),
"the message should say how far the database got and what to do, got: {message}"
);
let config = Config {
migrate: false,
..Config::new(db.url())
};
assert!(
PostgresSystemDatabase::connect(&config).await.is_ok(),
"a migrated schema should connect without migrating"
);
}
#[tokio::test]
async fn a_fork_carries_the_steps_below_its_start_step() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-src"), None, Submission::Fresh, None)
.await
.unwrap();
for step_id in 0..4 {
sys.record_step(
"wf-src",
step_id,
&format!("step_{step_id}"),
Outcome::Output(Some(&format!("\"out{step_id}\""))),
None,
None,
)
.await
.unwrap();
}
let forks = [Fork {
source_id: "wf-src",
forked_id: Some("wf-fork"),
start_step: 2,
}];
let ids = sys
.fork_workflows(&forks, &ForkOptions::default(), None)
.await
.unwrap();
assert_eq!(ids, ["wf-fork"]);
let fork = sys.get_workflow("wf-fork").await.unwrap().unwrap();
assert_eq!(fork.status, WorkflowStatus::Enqueued);
assert_eq!(fork.queue_name.as_deref(), Some(INTERNAL_QUEUE));
assert_eq!(fork.forked_from.as_deref(), Some("wf-src"));
let source = sys.get_workflow("wf-src").await.unwrap().unwrap();
assert_eq!(fork.name, source.name);
assert_eq!(fork.input, source.input);
assert_eq!(fork.application_version, source.application_version);
assert!(
source.was_forked_from,
"the source is marked as forked from"
);
let steps = sys
.list_workflow_steps("wf-fork", true, None, None, None)
.await
.unwrap();
assert_eq!(
steps
.iter()
.map(|s| (s.step_id, s.step_name.as_str(), s.output.as_deref()))
.collect::<Vec<_>>(),
[
(0, "step_0", Some("\"out0\"")),
(1, "step_1", Some("\"out1\"")),
],
);
}
#[tokio::test]
async fn a_fork_records_its_step_and_replays_from_it() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-forked"), None, Submission::Fresh, None)
.await
.unwrap();
sys.init_workflow(&workflow("wf-operator"), None, Submission::Fresh, None)
.await
.unwrap();
let caller = Some(("wf-operator", 0));
let first = sys
.fork_workflows(&[Fork::new("wf-forked")], &ForkOptions::default(), caller)
.await
.unwrap();
assert_eq!(first.len(), 1);
let steps = sys
.list_workflow_steps("wf-operator", true, None, None, None)
.await
.unwrap();
assert_eq!(
steps
.iter()
.map(|s| (s.step_id, s.step_name.as_str()))
.collect::<Vec<_>>(),
[(0, "DBOS.forkWorkflow")],
);
let replayed = sys
.fork_workflows(&[Fork::new("wf-forked")], &ForkOptions::default(), caller)
.await
.unwrap();
assert_eq!(
replayed, first,
"the replay forked again instead of replaying its checkpoint"
);
let forks: Vec<String> = sys
.list_workflows(&WorkflowFilter::default(), None)
.await
.unwrap()
.into_iter()
.filter(|w| w.forked_from.as_deref() == Some("wf-forked"))
.map(|w| w.workflow_id)
.collect();
assert_eq!(forks, first, "the source was forked twice");
}
#[tokio::test]
async fn a_cancel_records_its_step_and_replays_from_it() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-target"), None, Submission::Fresh, None)
.await
.unwrap();
sys.init_workflow(&workflow("wf-operator"), None, Submission::Fresh, None)
.await
.unwrap();
let caller = Some(("wf-operator", 0));
let first = sys
.cancel_workflows(&["wf-target"], false, caller)
.await
.unwrap();
assert_eq!(first, ["wf-target"]);
let steps = sys
.list_workflow_steps("wf-operator", true, None, None, None)
.await
.unwrap();
assert_eq!(
steps
.iter()
.map(|s| (s.step_id, s.step_name.as_str()))
.collect::<Vec<_>>(),
[(0, "DBOS.cancelWorkflow")],
);
let fresh = sys
.cancel_workflows(&["wf-target"], false, None)
.await
.unwrap();
assert!(fresh.is_empty(), "a second cancel moved something");
let replayed = sys
.cancel_workflows(&["wf-target"], false, caller)
.await
.unwrap();
assert_eq!(
replayed, first,
"the replay cancelled again instead of replaying its checkpoint"
);
}
#[tokio::test]
async fn a_listing_replays_the_rows_it_recorded() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-first"), None, Submission::Fresh, None)
.await
.unwrap();
sys.init_workflow(&workflow("wf-operator"), None, Submission::Fresh, None)
.await
.unwrap();
let caller = Some(("wf-operator", 0));
let filter = WorkflowFilter {
workflow_ids: vec!["wf-first", "wf-later"],
..WorkflowFilter::default()
};
let first = sys.list_workflows(&filter, caller).await.unwrap();
assert_eq!(first.len(), 1, "only one of the two exists yet");
sys.init_workflow(&workflow("wf-later"), None, Submission::Fresh, None)
.await
.unwrap();
let fresh = sys.list_workflows(&filter, None).await.unwrap();
assert_eq!(fresh.len(), 2, "the new workflow is really there");
let replayed = sys.list_workflows(&filter, caller).await.unwrap();
assert_eq!(
replayed, first,
"the replay read the table again instead of its checkpoint"
);
}
#[tokio::test]
async fn a_replayed_fork_from_does_not_resolve_the_fork_point_again() {
let (sys, db) = sysdb().await;
sys.init_workflow(&workflow("wf-resolved"), None, Submission::Fresh, None)
.await
.unwrap();
sys.init_workflow(&workflow("wf-operator"), None, Submission::Fresh, None)
.await
.unwrap();
sys.record_step(
"wf-resolved",
0,
"step_0",
Outcome::Output(Some("\"out\"")),
None,
None,
)
.await
.unwrap();
let caller = Some(("wf-operator", 0));
let first = sys
.fork_from(
&["wf-resolved"],
ForkPoint::LastStep,
&ForkOptions::default(),
caller,
)
.await
.unwrap();
let mut conn = db.admin_connection().await;
sqlx::query(sqlx::AssertSqlSafe(
"DELETE FROM dbos.operation_outputs WHERE workflow_uuid = 'wf-resolved'",
))
.execute(&mut conn)
.await
.unwrap();
let fresh = sys
.fork_from(
&["wf-resolved"],
ForkPoint::LastStep,
&ForkOptions::default(),
None,
)
.await;
assert!(
matches!(fresh, Err(Error::NoForkPoint { .. })),
"expected the lookup to be impossible now, got {fresh:?}"
);
let replayed = sys
.fork_from(
&["wf-resolved"],
ForkPoint::LastStep,
&ForkOptions::default(),
caller,
)
.await
.unwrap();
assert_eq!(
replayed, first,
"the replay looked the fork point up again instead of replaying its checkpoint"
);
}
#[tokio::test]
async fn a_fork_sees_the_event_values_as_of_its_start_step() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-events"), None, Submission::Fresh, None)
.await
.unwrap();
for (step_id, value) in [(0, "\"first\""), (1, "\"second\""), (2, "\"third\"")] {
sys.set_event("wf-events", step_id, "progress", value, None)
.await
.unwrap();
}
let forks = [Fork {
source_id: "wf-events",
forked_id: Some("wf-events-fork"),
start_step: 2,
}];
sys.fork_workflows(&forks, &ForkOptions::default(), None)
.await
.unwrap();
let events = sys.get_all_events("wf-events-fork").await.unwrap();
assert_eq!(events.len(), 1, "one key, at its value as of step 2");
assert_eq!(events[0].key, "progress");
assert_eq!(
events[0].value, "\"second\"",
"the value set at step 2 is the fork's future, not its past"
);
let source_events = sys.get_all_events("wf-events").await.unwrap();
assert_eq!(source_events[0].value, "\"third\"");
}
#[tokio::test]
async fn forking_a_missing_workflow_writes_nothing() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-present"), None, Submission::Fresh, None)
.await
.unwrap();
let forks = [
Fork {
source_id: "wf-present",
forked_id: Some("wf-would-be"),
start_step: 1,
},
Fork {
source_id: "wf-absent",
forked_id: Some("wf-never"),
start_step: 1,
},
];
match sys
.fork_workflows(&forks, &ForkOptions::default(), None)
.await
{
Err(Error::NonExistentWorkflow { workflow_ids }) => {
assert_eq!(workflow_ids, ["wf-absent"], "names which one is missing")
}
other => panic!("expected a missing-workflow error, got {other:?}"),
}
assert!(
sys.get_workflow("wf-would-be").await.unwrap().is_none(),
"the fork that could have been made must not have been"
);
}
#[tokio::test]
async fn a_fork_can_have_its_id_generated_and_its_placement_chosen() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-opts"), None, Submission::Fresh, None)
.await
.unwrap();
let ids = sys
.fork_workflows(
&[Fork::new("wf-opts")],
&ForkOptions {
application_version: Some("v2"),
queue_name: Some("reprocess"),
queue_partition_key: Some("eu"),
timeout: Some(std::time::Duration::from_secs(30)),
replacement_children: &[],
},
None,
)
.await
.unwrap();
assert_eq!(ids.len(), 1);
assert!(!ids[0].is_empty(), "an id should have been generated");
let fork = sys.get_workflow(&ids[0]).await.unwrap().unwrap();
assert_eq!(fork.application_version.as_deref(), Some("v2"));
assert_eq!(fork.queue_name.as_deref(), Some("reprocess"));
assert_eq!(fork.queue_partition_key.as_deref(), Some("eu"));
assert_eq!(fork.timeout, Some(std::time::Duration::from_secs(30)));
}
#[tokio::test]
async fn a_fork_option_that_is_empty_rather_than_absent_is_refused() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-blank"), None, Submission::Fresh, None)
.await
.unwrap();
for (field, options) in [
(
"application_version",
ForkOptions {
application_version: Some(""),
..ForkOptions::default()
},
),
(
"queue_name",
ForkOptions {
queue_name: Some(""),
..ForkOptions::default()
},
),
(
"timeout",
ForkOptions {
timeout: Some(std::time::Duration::ZERO),
..ForkOptions::default()
},
),
] {
let result = sys
.fork_workflows(&[Fork::new("wf-blank")], &options, None)
.await;
match result {
Err(Error::InvalidInput { field: f, .. }) => assert_eq!(f, field),
other => panic!("expected {field} to be refused, got {other:?}"),
}
}
let result = sys
.fork_workflows(
&[Fork {
source_id: "wf-blank",
forked_id: Some(""),
start_step: 0,
}],
&ForkOptions::default(),
None,
)
.await;
assert!(matches!(
result,
Err(Error::InvalidInput { ref field, .. }) if field == "forked_id"
));
}
#[tokio::test]
async fn replacing_one_child_twice_is_refused() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-dup"), None, Submission::Fresh, None)
.await
.unwrap();
sys.record_child_workflow("wf-dup", "child", 0, "spawn", None)
.await
.unwrap();
let result = sys
.fork_workflows(
&[Fork {
source_id: "wf-dup",
forked_id: Some("wf-dup-fork"),
start_step: 1,
}],
&ForkOptions {
replacement_children: &[("child", "fork-a"), ("child", "fork-b")],
..ForkOptions::default()
},
None,
)
.await;
match result {
Err(Error::InvalidInput { field, detail }) => {
assert_eq!(field, "replacement_children");
assert!(detail.contains("child"), "should name it, got: {detail}");
}
other => panic!("expected the duplicate to be refused, got {other:?}"),
}
assert!(sys.get_workflow("wf-dup-fork").await.unwrap().is_none());
}
#[tokio::test]
async fn a_fork_timeout_that_cannot_be_stored_is_refused() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-huge"), None, Submission::Fresh, None)
.await
.unwrap();
let result = sys
.fork_workflows(
&[Fork::new("wf-huge")],
&ForkOptions {
timeout: Some(std::time::Duration::MAX),
..ForkOptions::default()
},
None,
)
.await;
assert!(
matches!(
result,
Err(Error::InvalidInput { ref field, .. }) if field == "timeout"
),
"expected the timeout to be refused, got {result:?}"
);
}
#[tokio::test]
async fn a_fork_can_rewrite_the_children_it_replays() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-parent"), None, Submission::Fresh, None)
.await
.unwrap();
sys.record_child_workflow("wf-parent", "child-original", 0, "spawn", None)
.await
.unwrap();
sys.record_step(
"wf-parent",
1,
"charge",
Outcome::Output(Some("\"ok\"")),
None,
None,
)
.await
.unwrap();
sys.fork_workflows(
&[Fork {
source_id: "wf-parent",
forked_id: Some("wf-parent-fork"),
start_step: 2,
}],
&ForkOptions {
replacement_children: &[("child-original", "child-forked")],
..ForkOptions::default()
},
None,
)
.await
.unwrap();
let steps = sys
.list_workflow_steps("wf-parent-fork", true, None, None, None)
.await
.unwrap();
assert_eq!(steps.len(), 2, "both steps came across");
assert_eq!(
steps[0].child_workflow_id.as_deref(),
Some("child-forked"),
"the replayed step should point at the fork's own child"
);
assert_eq!(
steps[1].child_workflow_id, None,
"a step that spawned nothing must not acquire a child from the map"
);
assert_eq!(steps[1].output.as_deref(), Some("\"ok\""));
}
async fn workflow_with_a_failure(sys: &PostgresSystemDatabase, id: &str) {
sys.init_workflow(&workflow(id), None, Submission::Fresh, None)
.await
.unwrap();
sys.record_step(
id,
0,
"validate",
Outcome::Output(Some("\"ok\"")),
None,
None,
)
.await
.unwrap();
sys.record_step(id, 1, "charge", Outcome::Error("card declined"), None, None)
.await
.unwrap();
sys.record_step(
id,
2,
"notify",
Outcome::Output(Some("\"sent\"")),
None,
None,
)
.await
.unwrap();
}
#[tokio::test]
async fn forking_from_the_failure_restarts_at_the_failed_step() {
let (sys, _db) = sysdb().await;
workflow_with_a_failure(&sys, "wf-failed").await;
let ids = sys
.fork_from(
&["wf-failed"],
ForkPoint::LastFailure,
&ForkOptions::default(),
None,
)
.await
.unwrap();
let steps = sys
.list_workflow_steps(&ids[0], true, None, None, None)
.await
.unwrap();
assert_eq!(
steps
.iter()
.map(|s| (s.step_id, s.step_name.as_str()))
.collect::<Vec<_>>(),
[(0, "validate")],
);
}
#[tokio::test]
async fn forking_from_the_failure_of_a_workflow_that_never_failed_uses_its_last_step() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-clean"), None, Submission::Fresh, None)
.await
.unwrap();
for (step_id, name) in [(0, "one"), (1, "two")] {
sys.record_step(
"wf-clean",
step_id,
name,
Outcome::Output(Some("\"ok\"")),
None,
None,
)
.await
.unwrap();
}
let ids = sys
.fork_from(
&["wf-clean"],
ForkPoint::LastFailure,
&ForkOptions::default(),
None,
)
.await
.unwrap();
let steps = sys
.list_workflow_steps(&ids[0], true, None, None, None)
.await
.unwrap();
assert_eq!(steps.len(), 1, "resumes at step 1, so only step 0 replays");
assert_eq!(steps[0].step_name, "one");
}
#[tokio::test]
async fn each_fork_point_resolves_to_its_own_step() {
let (sys, _db) = sysdb().await;
workflow_with_a_failure(&sys, "wf-points").await;
for (point, expected_replayed) in [
(ForkPoint::LastStep, 2), (ForkPoint::StepNamed("charge"), 1), (ForkPoint::Step(0), 0), ] {
let ids = sys
.fork_from(&["wf-points"], point, &ForkOptions::default(), None)
.await
.unwrap();
let steps = sys
.list_workflow_steps(&ids[0], true, None, None, None)
.await
.unwrap();
assert_eq!(
steps.len(),
expected_replayed,
"{point:?} should leave {expected_replayed} steps to replay"
);
}
}
#[tokio::test]
async fn forking_from_a_point_that_does_not_exist_is_refused() {
let (sys, _db) = sysdb().await;
workflow_with_a_failure(&sys, "wf-has-steps").await;
sys.init_workflow(&workflow("wf-no-steps"), None, Submission::Fresh, None)
.await
.unwrap();
match sys
.fork_from(
&["wf-has-steps", "wf-no-steps"],
ForkPoint::LastStep,
&ForkOptions::default(),
None,
)
.await
{
Err(Error::NoForkPoint {
workflow_ids,
step_name,
}) => {
assert_eq!(workflow_ids, ["wf-no-steps"]);
assert_eq!(step_name, None);
}
other => panic!("expected a missing fork point, got {other:?}"),
}
match sys
.fork_from(
&["wf-has-steps"],
ForkPoint::StepNamed("refund"),
&ForkOptions::default(),
None,
)
.await
{
Err(Error::NoForkPoint {
workflow_ids,
step_name,
}) => {
assert_eq!(workflow_ids, ["wf-has-steps"]);
assert_eq!(step_name.as_deref(), Some("refund"));
}
other => panic!("expected a missing step name, got {other:?}"),
}
let forks = sys
.list_workflows(
&dbos::sysdb::types::WorkflowFilter {
forked_from: vec!["wf-has-steps"],
..Default::default()
},
None,
)
.await
.unwrap();
assert!(forks.is_empty(), "a refused batch forks nothing");
}
#[tokio::test]
async fn a_batch_resolves_each_workflow_separately_and_keeps_the_order() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-alpha"), None, Submission::Fresh, None)
.await
.unwrap();
for (step_id, outcome) in [
(0, Outcome::Output(Some("\"ok\""))),
(1, Outcome::Error("boom")),
(2, Outcome::Output(Some("\"ok\""))),
] {
sys.record_step("wf-alpha", step_id, "s", outcome, None, None)
.await
.unwrap();
}
sys.init_workflow(&workflow("wf-beta"), None, Submission::Fresh, None)
.await
.unwrap();
for (step_id, outcome) in [
(0, Outcome::Output(Some("\"ok\""))),
(1, Outcome::Output(Some("\"ok\""))),
(2, Outcome::Output(Some("\"ok\""))),
(3, Outcome::Error("boom")),
] {
sys.record_step("wf-beta", step_id, "s", outcome, None, None)
.await
.unwrap();
}
let ids = sys
.fork_from(
&["wf-alpha", "wf-beta"],
ForkPoint::LastFailure,
&ForkOptions::default(),
None,
)
.await
.unwrap();
assert_eq!(ids.len(), 2);
let alpha = sys.get_workflow(&ids[0]).await.unwrap().unwrap();
let beta = sys.get_workflow(&ids[1]).await.unwrap().unwrap();
assert_eq!(alpha.forked_from.as_deref(), Some("wf-alpha"));
assert_eq!(beta.forked_from.as_deref(), Some("wf-beta"));
let alpha_steps = sys
.list_workflow_steps(&ids[0], true, None, None, None)
.await
.unwrap();
let beta_steps = sys
.list_workflow_steps(&ids[1], true, None, None, None)
.await
.unwrap();
assert_eq!(alpha_steps.len(), 1, "wf-alpha resumes at its own step 1");
assert_eq!(beta_steps.len(), 3, "wf-beta resumes at its own step 3");
}
#[tokio::test]
async fn a_message_is_delivered_to_its_destination() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-dest"), None, Submission::Fresh, None)
.await
.unwrap();
sys.send_messages(
&[
Message {
destination_id: "wf-dest",
topic: Some("approvals"),
message: "\"yes\"",
idempotency_key: None,
},
Message {
destination_id: "wf-dest",
topic: None,
message: "\"untopicked\"",
idempotency_key: None,
},
],
Some("portable_json"),
None,
false,
)
.await
.unwrap();
let sent = sys.get_all_notifications("wf-dest").await.unwrap();
assert_eq!(sent.len(), 2);
let topic_of = |message: &str| {
sent.iter()
.find(|n| n.message == message)
.unwrap_or_else(|| panic!("{message} was not delivered"))
.topic
.clone()
};
assert_eq!(topic_of("\"yes\"").as_deref(), Some("approvals"));
assert_eq!(
topic_of("\"untopicked\"").as_deref(),
Some(dbos::sysdb::NULL_TOPIC)
);
}
#[tokio::test]
async fn a_keyed_message_is_delivered_once_however_often_it_is_sent() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-once"), None, Submission::Fresh, None)
.await
.unwrap();
let message = Message {
destination_id: "wf-once",
topic: Some("t"),
message: "\"payload\"",
idempotency_key: Some("order-42"),
};
for _ in 0..3 {
sys.send_message(&message, None, None, false).await.unwrap();
}
assert_eq!(sys.get_all_notifications("wf-once").await.unwrap().len(), 1);
}
#[tokio::test]
async fn two_messages_under_one_key_are_refused() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-clash"), None, Submission::Fresh, None)
.await
.unwrap();
let result = sys
.send_messages(
&[
Message {
destination_id: "wf-clash",
topic: None,
message: "\"a\"",
idempotency_key: Some("same"),
},
Message {
destination_id: "wf-clash",
topic: None,
message: "\"b\"",
idempotency_key: Some("same"),
},
],
None,
None,
false,
)
.await;
match result {
Err(Error::InvalidInput { field, detail }) => {
assert_eq!(field, "idempotency_key");
assert!(detail.contains("same"), "should name it, got: {detail}");
}
other => panic!("expected the clash to be refused, got {other:?}"),
}
assert!(
sys.get_all_notifications("wf-clash")
.await
.unwrap()
.is_empty()
);
}
#[tokio::test]
async fn a_replayed_send_does_not_send_again() {
let (sys, _db) = sysdb().await;
for id in ["wf-sender", "wf-receiver"] {
sys.init_workflow(&workflow(id), None, Submission::Fresh, None)
.await
.unwrap();
}
let caller = ("wf-sender", 0);
let message = Message {
destination_id: "wf-receiver",
topic: None,
message: "\"hello\"",
idempotency_key: None,
};
sys.send_message(&message, None, Some(caller), false)
.await
.unwrap();
sys.send_message(&message, None, Some(caller), false)
.await
.unwrap();
assert_eq!(
sys.get_all_notifications("wf-receiver")
.await
.unwrap()
.len(),
1,
"the replay should have found the step recorded and sent nothing"
);
let steps = sys
.list_workflow_steps("wf-sender", true, None, None, None)
.await
.unwrap();
assert_eq!(steps.len(), 1);
assert_eq!(steps[0].step_name, "DBOS.send");
}
#[tokio::test]
async fn sending_to_a_missing_workflow_is_refused() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-real"), None, Submission::Fresh, None)
.await
.unwrap();
let result = sys
.send_messages(
&[
Message {
destination_id: "wf-real",
topic: None,
message: "\"a\"",
idempotency_key: None,
},
Message {
destination_id: "wf-ghost",
topic: None,
message: "\"b\"",
idempotency_key: None,
},
],
None,
None,
false,
)
.await;
assert!(matches!(result, Err(Error::NonExistentWorkflow { .. })));
assert!(
sys.get_all_notifications("wf-real")
.await
.unwrap()
.is_empty(),
"the deliverable message must not have been delivered either"
);
}
#[tokio::test]
async fn a_message_can_follow_a_workflow_to_its_forks() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-root"), None, Submission::Fresh, None)
.await
.unwrap();
let child = sys
.fork_workflows(&[Fork::new("wf-root")], &ForkOptions::default(), None)
.await
.unwrap()
.remove(0);
let grandchild = sys
.fork_workflows(&[Fork::new(&child)], &ForkOptions::default(), None)
.await
.unwrap()
.remove(0);
sys.send_message(
&Message {
destination_id: "wf-root",
topic: Some("t"),
message: "\"broadcast\"",
idempotency_key: Some("key"),
},
None,
None,
true,
)
.await
.unwrap();
for id in ["wf-root", &child, &grandchild] {
assert_eq!(
sys.get_all_notifications(id).await.unwrap().len(),
1,
"{id} should have received the broadcast"
);
}
sys.send_message(
&Message {
destination_id: "wf-root",
topic: Some("t"),
message: "\"direct\"",
idempotency_key: Some("key2"),
},
None,
None,
false,
)
.await
.unwrap();
assert_eq!(sys.get_all_notifications("wf-root").await.unwrap().len(), 2);
assert_eq!(sys.get_all_notifications(&child).await.unwrap().len(), 1);
sys.send_message(
&Message {
destination_id: "wf-root",
topic: Some("t"),
message: "\"unkeyed\"",
idempotency_key: None,
},
None,
None,
true,
)
.await
.unwrap();
for (id, expected) in [
("wf-root", 3),
(child.as_str(), 2),
(grandchild.as_str(), 2),
] {
assert_eq!(
sys.get_all_notifications(id).await.unwrap().len(),
expected,
"{id} should have received the unkeyed broadcast"
);
}
}
#[tokio::test]
async fn sending_no_messages_still_records_the_step() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-empty"), None, Submission::Fresh, None)
.await
.unwrap();
sys.send_messages(&[], None, Some(("wf-empty", 0)), false)
.await
.unwrap();
let steps = sys
.list_workflow_steps("wf-empty", true, None, None, None)
.await
.unwrap();
assert_eq!(steps.len(), 1, "the step id must not be left unoccupied");
assert_eq!(steps[0].step_id, 0);
assert_eq!(steps[0].step_name, "DBOS.sendBulk");
sys.send_messages(&[], None, Some(("wf-empty", 0)), false)
.await
.unwrap();
assert_eq!(
sys.list_workflow_steps("wf-empty", true, None, None, None)
.await
.unwrap()
.len(),
1
);
sys.send_messages(&[], None, None, false).await.unwrap();
assert_eq!(
sys.list_workflow_steps("wf-empty", true, None, None, None)
.await
.unwrap()
.len(),
1
);
}
#[tokio::test]
async fn the_recorded_step_name_follows_the_surface_not_the_size() {
let (sys, _db) = sysdb().await;
for id in ["wf-namer", "wf-a", "wf-b"] {
sys.init_workflow(&workflow(id), None, Submission::Fresh, None)
.await
.unwrap();
}
let to = |destination_id| Message {
destination_id,
topic: None,
message: "\"m\"",
idempotency_key: None,
};
sys.send_message(&to("wf-a"), None, Some(("wf-namer", 0)), false)
.await
.unwrap();
sys.send_messages(&[to("wf-b")], None, Some(("wf-namer", 1)), false)
.await
.unwrap();
sys.send_messages(
&[to("wf-a"), to("wf-b")],
None,
Some(("wf-namer", 2)),
false,
)
.await
.unwrap();
let steps = sys
.list_workflow_steps("wf-namer", true, None, None, None)
.await
.unwrap();
assert_eq!(
steps
.iter()
.map(|s| (s.step_id, s.step_name.as_str()))
.collect::<Vec<_>>(),
[(0, "DBOS.send"), (1, "DBOS.sendBulk"), (2, "DBOS.sendBulk"),],
"the name is the surface the caller reached for, and a batch of one is still a batch",
);
}
#[tokio::test]
async fn an_unprotected_send_outside_a_workflow_delivers_every_time() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-unprotected"), None, Submission::Fresh, None)
.await
.unwrap();
let message = Message {
destination_id: "wf-unprotected",
topic: None,
message: "\"again\"",
idempotency_key: None,
};
for _ in 0..3 {
sys.send_message(&message, None, None, false).await.unwrap();
}
assert_eq!(
sys.get_all_notifications("wf-unprotected")
.await
.unwrap()
.len(),
3,
"no step and no key means no idempotency"
);
}
#[tokio::test]
async fn a_replay_that_changed_its_send_surface_is_refused() {
let (sys, _db) = sysdb().await;
for id in ["wf-varying", "wf-x", "wf-y"] {
sys.init_workflow(&workflow(id), None, Submission::Fresh, None)
.await
.unwrap();
}
let to = |destination_id| Message {
destination_id,
topic: None,
message: "\"m\"",
idempotency_key: None,
};
sys.send_message(&to("wf-x"), None, Some(("wf-varying", 0)), false)
.await
.unwrap();
let result = sys
.send_messages(
&[to("wf-x"), to("wf-y")],
None,
Some(("wf-varying", 0)),
false,
)
.await;
match result {
Err(Error::UnexpectedStep {
step_id,
expected,
recorded,
..
}) => {
assert_eq!(step_id, 0);
assert_eq!(expected, "DBOS.sendBulk");
assert_eq!(recorded, "DBOS.send");
}
other => panic!("expected the changed batch to be caught, got {other:?}"),
}
assert_eq!(sys.get_all_notifications("wf-x").await.unwrap().len(), 1);
assert!(sys.get_all_notifications("wf-y").await.unwrap().is_empty());
}
#[tokio::test]
async fn a_stream_offset_reads_back_with_its_producers_status() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-stream"), None, Submission::Fresh, None)
.await
.unwrap();
sys.write_stream(
"wf-stream",
0,
"progress",
"\"a\"",
Some("portable_json"),
WrittenBy::Workflow,
)
.await
.unwrap();
let read = sys
.read_stream_value("wf-stream", "progress", 0)
.await
.unwrap();
assert_eq!(read.status, WorkflowStatus::Pending);
assert_eq!(
read.value,
Some(EncodedValue {
value: "\"a\"".to_owned(),
serialization: Some("portable_json".to_owned()),
})
);
}
#[tokio::test]
async fn an_empty_offset_still_reports_whether_the_producer_is_running() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-stream"), None, Submission::Fresh, None)
.await
.unwrap();
for (key, offset) in [("progress", 0), ("never-written", 0), ("progress", 7)] {
let read = sys
.read_stream_value("wf-stream", key, offset)
.await
.unwrap();
assert_eq!(read.value, None, "key {key} offset {offset}");
assert_eq!(read.status, WorkflowStatus::Pending);
}
sys.record_workflow_outcome("wf-stream", Outcome::Output(Some("\"done\"")))
.await
.unwrap();
let read = sys
.read_stream_value("wf-stream", "progress", 0)
.await
.unwrap();
assert_eq!(read.status, WorkflowStatus::Success);
assert_eq!(read.value, None);
}
#[tokio::test]
async fn reading_a_stream_of_a_missing_workflow_is_refused() {
let (sys, _db) = sysdb().await;
let err = sys
.read_stream_value("wf-nobody", "progress", 0)
.await
.expect_err("there is no workflow to wait on");
assert!(
matches!(err, Error::NonExistentWorkflow { ref workflow_ids }
if workflow_ids == &["wf-nobody".to_owned()]),
"unexpected error: {err:?}",
);
}
#[tokio::test]
async fn a_closed_stream_reports_its_sentinel_like_any_other_value() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-stream"), None, Submission::Fresh, None)
.await
.unwrap();
sys.write_stream(
"wf-stream",
0,
"progress",
"\"a\"",
Some("portable_json"),
WrittenBy::Workflow,
)
.await
.unwrap();
sys.close_stream("wf-stream", 1, "progress").await.unwrap();
assert_eq!(
sys.read_stream_value("wf-stream", "progress", 1)
.await
.unwrap()
.value
.map(|v| v.value),
Some(dbos::sysdb::STREAM_CLOSED.to_owned()),
"the sentinel is a value at an offset like any other",
);
assert_eq!(
sys.read_stream_value("wf-stream", "progress", 2)
.await
.unwrap()
.value,
None,
);
}
#[tokio::test]
async fn the_offsets_of_a_stream_read_back_as_the_stream() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-stream"), None, Submission::Fresh, None)
.await
.unwrap();
for (step_id, value) in [(0, "\"a\""), (1, "\"b\""), (2, "\"c\"")] {
sys.write_stream(
"wf-stream",
step_id,
"progress",
value,
Some("portable_json"),
WrittenBy::Workflow,
)
.await
.unwrap();
}
sys.close_stream("wf-stream", 3, "progress").await.unwrap();
let mut read = Vec::new();
for offset in 0..10 {
let at = sys
.read_stream_value("wf-stream", "progress", offset)
.await
.unwrap();
match at.value {
Some(v) if v.value == dbos::sysdb::STREAM_CLOSED => break,
Some(v) => read.push(v.value),
None => break,
}
}
assert_eq!(read, ["\"a\"", "\"b\"", "\"c\""]);
}
#[tokio::test]
async fn capped_stream_reads_do_not_block_each_other() {
let db = test_database().await;
let sys = std::sync::Arc::new(PostgresSystemDatabase::from_pool(
db.pool().await,
&Settings {
polling_concurrency: Some(1),
..Settings::default()
},
));
sys.init_workflow(&workflow("wf-stream"), None, Submission::Fresh, None)
.await
.unwrap();
sys.write_stream(
"wf-stream",
0,
"progress",
"\"a\"",
None,
WrittenBy::Workflow,
)
.await
.unwrap();
let readers: Vec<_> = (0..8)
.map(|_| {
let sys = std::sync::Arc::clone(&sys);
tokio::spawn(async move { sys.read_stream_value("wf-stream", "progress", 0).await })
})
.collect();
for reader in readers {
let read = tokio::time::timeout(RECHECK * 20, reader)
.await
.expect("a reader starved behind the polling cap")
.unwrap()
.unwrap();
assert_eq!(read.value.map(|v| v.value), Some("\"a\"".to_owned()));
}
}
#[tokio::test]
async fn stream_writes_are_appended_in_order() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-stream"), None, Submission::Fresh, None)
.await
.unwrap();
for (step_id, value) in [(0, "\"a\""), (1, "\"b\""), (2, "\"c\"")] {
sys.write_stream(
"wf-stream",
step_id,
"progress",
value,
Some("portable_json"),
WrittenBy::Workflow,
)
.await
.unwrap();
}
let entries = sys.get_all_stream_entries("wf-stream").await.unwrap();
assert_eq!(
entries
.iter()
.map(|e| (e.offset, e.value.as_str()))
.collect::<Vec<_>>(),
[(0, "\"a\""), (1, "\"b\""), (2, "\"c\"")],
);
sys.write_stream("wf-stream", 3, "other", "\"z\"", None, WrittenBy::Workflow)
.await
.unwrap();
let other = sys.get_all_stream_entries("wf-stream").await.unwrap();
let zero_offsets = other.iter().filter(|e| e.offset == 0).count();
assert_eq!(zero_offsets, 2, "each key numbers its own entries");
}
#[tokio::test]
async fn a_replayed_stream_write_does_not_append_again() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-replay"), None, Submission::Fresh, None)
.await
.unwrap();
for _ in 0..3 {
sys.write_stream("wf-replay", 0, "k", "\"once\"", None, WrittenBy::Workflow)
.await
.unwrap();
}
assert_eq!(
sys.get_all_stream_entries("wf-replay").await.unwrap().len(),
1
);
let steps = sys
.list_workflow_steps("wf-replay", true, None, None, None)
.await
.unwrap();
assert_eq!(steps.len(), 1);
assert_eq!(steps[0].step_name, "DBOS.writeStream");
}
#[tokio::test]
async fn a_write_from_inside_a_step_records_nothing_of_its_own() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-instep"), None, Submission::Fresh, None)
.await
.unwrap();
for _ in 0..3 {
sys.write_stream("wf-instep", 0, "k", "\"each\"", None, WrittenBy::Step)
.await
.unwrap();
}
assert_eq!(
sys.get_all_stream_entries("wf-instep").await.unwrap().len(),
3
);
assert!(
sys.list_workflow_steps("wf-instep", true, None, None, None)
.await
.unwrap()
.is_empty(),
"the enclosing step is the durable unit, not this write"
);
}
#[tokio::test]
async fn closing_a_stream_appends_the_sentinel() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-close"), None, Submission::Fresh, None)
.await
.unwrap();
sys.write_stream("wf-close", 0, "k", "\"value\"", None, WrittenBy::Workflow)
.await
.unwrap();
sys.close_stream("wf-close", 1, "k").await.unwrap();
let entries = sys.get_all_stream_entries("wf-close").await.unwrap();
assert_eq!(entries.len(), 2);
assert_eq!(entries[1].offset, 1);
assert_eq!(
entries[1].value,
dbos::sysdb::STREAM_CLOSED,
"closing is an ordinary append of the sentinel"
);
let steps = sys
.list_workflow_steps("wf-close", true, None, None, None)
.await
.unwrap();
assert_eq!(
steps
.iter()
.map(|s| s.step_name.as_str())
.collect::<Vec<_>>(),
["DBOS.writeStream", "DBOS.closeStream"],
);
sys.close_stream("wf-close", 1, "k").await.unwrap();
assert_eq!(
sys.get_all_stream_entries("wf-close").await.unwrap().len(),
2
);
}
#[tokio::test]
async fn writing_a_stream_for_a_missing_workflow_is_refused() {
let (sys, _db) = sysdb().await;
let result = sys
.write_stream("wf-nowhere", 0, "k", "\"v\"", None, WrittenBy::Step)
.await;
assert!(matches!(result, Err(Error::NonExistentWorkflow { .. })));
}
#[tokio::test]
async fn a_named_handle_stamps_its_application_on_what_it_writes() {
let db = test_database().await;
let pool = db.pool().await;
let alpha = PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some("alpha"),
..Settings::default()
},
);
let anonymous = PostgresSystemDatabase::from_pool(pool.clone(), &Settings::default());
for (sys, id) in [(&alpha, "wf-alpha"), (&anonymous, "wf-nobody")] {
sys.init_workflow(&workflow(id), None, Submission::Fresh, None)
.await
.unwrap();
sys.record_step(
id,
0,
"a_step",
Outcome::Output(Some("\"out\"")),
None,
None,
)
.await
.unwrap();
sys.record_child_workflow(id, &format!("{id}-child"), 1, "a_child", None)
.await
.unwrap();
}
for (id, expected) in [("wf-alpha", Some("alpha")), ("wf-nobody", None)] {
let owner: Option<String> = sqlx::query_scalar(
r#"SELECT "application_name" FROM "dbos"."workflow_status" WHERE "workflow_uuid" = $1"#,
)
.bind(id)
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(owner.as_deref(), expected, "{id}: workflow owner");
let owners: Vec<Option<String>> = sqlx::query_scalar(
r#"SELECT "application_name" FROM "dbos"."operation_outputs"
WHERE "workflow_uuid" = $1 ORDER BY "function_id""#,
)
.bind(id)
.fetch_all(&pool)
.await
.unwrap();
assert_eq!(
owners.iter().map(|o| o.as_deref()).collect::<Vec<_>>(),
[expected, expected],
"{id}: step owners",
);
}
}
#[tokio::test]
async fn a_resubmission_does_not_re_own_a_claimed_workflow() {
let db = test_database().await;
let pool = db.pool().await;
let alpha = PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some("alpha"),
..Settings::default()
},
);
let beta = PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some("beta"),
..Settings::default()
},
);
alpha
.init_workflow(&workflow("wf-contested"), None, Submission::Fresh, None)
.await
.unwrap();
beta.init_workflow(&workflow("wf-contested"), None, Submission::Fresh, None)
.await
.unwrap();
let owner: Option<String> = sqlx::query_scalar(
r#"SELECT "application_name" FROM "dbos"."workflow_status" WHERE "workflow_uuid" = $1"#,
)
.bind("wf-contested")
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(
owner.as_deref(),
Some("alpha"),
"the first writer keeps the workflow",
);
}
#[tokio::test]
async fn a_fork_inherits_its_sources_application() {
let db = test_database().await;
let pool = db.pool().await;
let alpha = PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some("alpha"),
..Settings::default()
},
);
let beta = PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some("beta"),
..Settings::default()
},
);
let anonymous = PostgresSystemDatabase::from_pool(pool.clone(), &Settings::default());
for (sys, id) in [(&alpha, "wf-owned"), (&anonymous, "wf-unowned")] {
sys.init_workflow(&workflow(id), None, Submission::Fresh, None)
.await
.unwrap();
sys.record_step(
id,
0,
"a_step",
Outcome::Output(Some("\"out\"")),
None,
None,
)
.await
.unwrap();
}
beta.fork_workflows(
&[
Fork {
source_id: "wf-owned",
forked_id: Some("fork-of-owned"),
start_step: 1,
},
Fork {
source_id: "wf-unowned",
forked_id: Some("fork-of-unowned"),
start_step: 1,
},
],
&ForkOptions::default(),
None,
)
.await
.unwrap();
for (fork_id, expected) in [
("fork-of-owned", "alpha"), ("fork-of-unowned", "beta"), ] {
let owner: Option<String> = sqlx::query_scalar(
r#"SELECT "application_name" FROM "dbos"."workflow_status" WHERE "workflow_uuid" = $1"#,
)
.bind(fork_id)
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(
owner.as_deref(),
Some(expected),
"{fork_id}: workflow owner"
);
let step_owner: Option<String> = sqlx::query_scalar(
r#"SELECT "application_name" FROM "dbos"."operation_outputs"
WHERE "workflow_uuid" = $1"#,
)
.bind(fork_id)
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(
step_owner.as_deref(),
Some(expected),
"{fork_id}: step owner"
);
}
}
#[tokio::test]
async fn recovery_does_not_reach_across_applications() {
let db = test_database().await;
let pool = db.pool().await;
let named = |name: &'static str| {
PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some(name),
..Settings::default()
},
)
};
let (alpha, beta) = (named("alpha"), named("beta"));
let anonymous = PostgresSystemDatabase::from_pool(pool.clone(), &Settings::default());
for (sys, id) in [
(&alpha, "wf-alpha"),
(&beta, "wf-beta"),
(&anonymous, "wf-nobody"),
] {
sys.init_workflow(&workflow(id), None, Submission::Fresh, None)
.await
.unwrap();
}
let mut found = ids(&alpha, &pending("local", "v1")).await;
found.sort();
assert_eq!(
found,
["wf-alpha", "wf-nobody"],
"alpha recovers its own and the unclaimed, never beta's",
);
let mut all = ids(&anonymous, &pending("local", "v1")).await;
all.sort();
assert_eq!(all, ["wf-alpha", "wf-beta", "wf-nobody"]);
}
#[tokio::test]
async fn listing_scopes_to_the_caller_unless_it_names_ids_or_applications() {
let db = test_database().await;
let pool = db.pool().await;
let named = |name: &'static str| {
PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some(name),
..Settings::default()
},
)
};
let (alpha, beta) = (named("alpha"), named("beta"));
let anonymous = PostgresSystemDatabase::from_pool(pool.clone(), &Settings::default());
for (sys, id) in [
(&alpha, "wf-alpha"),
(&beta, "wf-beta"),
(&anonymous, "wf-nobody"),
] {
sys.init_workflow(&workflow(id), None, Submission::Fresh, None)
.await
.unwrap();
}
let ids = |found: Vec<WorkflowRecord>| {
let mut ids: Vec<String> = found.into_iter().map(|w| w.workflow_id).collect();
ids.sort();
ids
};
assert_eq!(
ids(alpha
.list_workflows(&WorkflowFilter::default(), None)
.await
.unwrap()),
["wf-alpha", "wf-nobody"],
);
assert_eq!(
ids(alpha
.list_workflows(
&WorkflowFilter {
applications: Applications::Any,
..WorkflowFilter::default()
},
None
)
.await
.unwrap()),
["wf-alpha", "wf-beta", "wf-nobody"],
);
assert_eq!(
ids(alpha
.list_workflows(
&WorkflowFilter {
applications: Applications::Named(vec!["beta"]),
..WorkflowFilter::default()
},
None
)
.await
.unwrap()),
["wf-beta", "wf-nobody"],
);
assert_eq!(
ids(alpha
.list_workflows(
&WorkflowFilter {
workflow_ids: vec!["wf-beta"],
..WorkflowFilter::default()
},
None
)
.await
.unwrap()),
["wf-beta"],
);
assert_eq!(
ids(alpha
.list_workflows(
&WorkflowFilter {
workflow_id_prefixes: vec!["wf-"],
..WorkflowFilter::default()
},
None
)
.await
.unwrap()),
["wf-alpha", "wf-nobody"],
);
assert_eq!(
ids(anonymous
.list_workflows(&WorkflowFilter::default(), None)
.await
.unwrap()),
["wf-alpha", "wf-beta", "wf-nobody"],
);
}
#[tokio::test]
async fn a_workflow_reports_the_application_that_owns_it() {
let db = test_database().await;
let pool = db.pool().await;
let alpha = PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some("alpha"),
..Settings::default()
},
);
let anonymous = PostgresSystemDatabase::from_pool(pool.clone(), &Settings::default());
alpha
.init_workflow(&workflow("wf-owned"), None, Submission::Fresh, None)
.await
.unwrap();
anonymous
.init_workflow(&workflow("wf-unowned"), None, Submission::Fresh, None)
.await
.unwrap();
let owned = alpha.get_workflow("wf-owned").await.unwrap().unwrap();
assert_eq!(owned.application_name.as_deref(), Some("alpha"));
let unowned = alpha.get_workflow("wf-unowned").await.unwrap().unwrap();
assert_eq!(
unowned.application_name, None,
"unclaimed reads back as unclaimed rather than as this handle's own",
);
}
#[tokio::test]
async fn releasing_delayed_workflows_stays_within_an_application() {
let db = test_database().await;
let pool = db.pool().await;
let named = |name: &'static str| {
PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some(name),
..Settings::default()
},
)
};
let (alpha, beta) = (named("alpha"), named("beta"));
for (sys, id) in [(&alpha, "wf-alpha"), (&beta, "wf-beta")] {
let wf = NewWorkflow {
queue_name: Some("orders"),
delay: Some(std::time::Duration::from_secs(3600)),
..NewWorkflow::new(id)
};
sys.init_workflow(&wf, None, Submission::Fresh, None)
.await
.unwrap();
sys.set_workflow_delay(id, WorkflowDelay::Until(Timestamp::from_epoch_ms(1)), None)
.await
.unwrap();
}
assert_eq!(
alpha.transition_delayed_workflows().await.unwrap(),
1,
"alpha releases only its own",
);
assert_eq!(
beta.get_workflow("wf-beta").await.unwrap().unwrap().status,
WorkflowStatus::Delayed,
"beta's workflow is untouched",
);
assert_eq!(beta.transition_delayed_workflows().await.unwrap(), 1);
}
#[tokio::test]
async fn a_peers_deploy_does_not_become_this_applications_latest() {
let db = test_database().await;
let pool = db.pool().await;
let named = |name: &'static str| {
PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some(name),
..Settings::default()
},
)
};
let (alpha, beta) = (named("alpha"), named("beta"));
alpha
.create_application_version("alpha-v1", None)
.await
.unwrap();
beta.create_application_version("beta-v1", None)
.await
.unwrap();
beta.create_application_version("beta-v2", None)
.await
.unwrap();
let latest = alpha
.get_latest_application_version(None)
.await
.unwrap()
.expect("alpha has a version");
assert_eq!(latest.version_name, "alpha-v1");
assert_eq!(latest.application_name.as_deref(), Some("alpha"));
let names: Vec<String> = alpha
.list_application_versions()
.await
.unwrap()
.into_iter()
.map(|v| v.version_name)
.collect();
assert_eq!(names, ["alpha-v1"]);
}
#[tokio::test]
async fn registering_a_peers_version_name_is_refused() {
let db = test_database().await;
let pool = db.pool().await;
let named = |name: &'static str| {
PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some(name),
..Settings::default()
},
)
};
let (alpha, beta) = (named("alpha"), named("beta"));
alpha
.create_application_version("v1.0.0", None)
.await
.unwrap();
let result = beta.create_application_version("v1.0.0", None).await;
match result {
Err(Error::RegisteredByAnother {
ref kind,
ref name,
ref holder,
ref claimant,
}) => {
assert_eq!(kind.as_ref(), "Application version");
assert_eq!(name, "v1.0.0");
assert_eq!(holder, "alpha");
assert_eq!(claimant.as_deref(), Some("beta"));
let text = result.as_ref().unwrap_err().to_string();
assert!(text.contains("already registered"), "got {text}");
assert!(text.contains("was renamed"), "got {text}");
}
other => panic!("expected a registration conflict, got {other:?}"),
}
let before = alpha
.get_latest_application_version(None)
.await
.unwrap()
.unwrap()
.version_timestamp;
let promote = beta
.update_application_version_timestamp("v1.0.0", Timestamp::from_epoch_ms(9_000_000), None)
.await;
assert!(matches!(promote, Err(Error::RegisteredByAnother { .. })));
assert_eq!(
alpha
.get_latest_application_version(None)
.await
.unwrap()
.unwrap()
.version_timestamp,
before,
);
alpha
.create_application_version("v1.0.0", None)
.await
.unwrap();
}
#[tokio::test]
async fn an_unclaimed_version_is_claimed_where_it_stands() {
let db = test_database().await;
let pool = db.pool().await;
let anonymous = PostgresSystemDatabase::from_pool(pool.clone(), &Settings::default());
let alpha = PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some("alpha"),
..Settings::default()
},
);
anonymous
.create_application_version("v1", None)
.await
.unwrap();
anonymous
.update_application_version_timestamp("v1", Timestamp::from_epoch_ms(5_000_000), None)
.await
.unwrap();
let unclaimed = anonymous
.get_latest_application_version(None)
.await
.unwrap()
.unwrap();
assert_eq!(unclaimed.application_name, None);
alpha.create_application_version("v1", None).await.unwrap();
let claimed = alpha
.get_latest_application_version(None)
.await
.unwrap()
.unwrap();
assert_eq!(claimed.application_name.as_deref(), Some("alpha"));
assert_eq!(
claimed.version_id, unclaimed.version_id,
"claimed in place rather than recreated",
);
assert_eq!(
claimed.version_timestamp, unclaimed.version_timestamp,
"the timestamp that decides which version is current is untouched",
);
anonymous
.create_application_version("v1", None)
.await
.unwrap();
assert_eq!(
alpha
.get_latest_application_version(None)
.await
.unwrap()
.unwrap()
.application_name
.as_deref(),
Some("alpha"),
);
}
#[tokio::test]
async fn a_named_target_overrides_the_handles_own_application() {
let db = test_database().await;
let pool = db.pool().await;
let alpha = PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some("alpha"),
..Settings::default()
},
);
let client = PostgresSystemDatabase::from_pool(pool.clone(), &Settings::default());
client
.create_application_version("beta-v1", Some("beta"))
.await
.unwrap();
alpha
.create_application_version("beta-v2", Some("beta"))
.await
.unwrap();
assert!(
alpha
.get_latest_application_version(None)
.await
.unwrap()
.is_none(),
"alpha registered nothing for itself",
);
let latest = alpha
.get_latest_application_version(Some("beta"))
.await
.unwrap()
.expect("beta has versions");
assert_eq!(latest.application_name.as_deref(), Some("beta"));
alpha
.update_application_version_timestamp(
"beta-v1",
Timestamp::from_epoch_ms(99_000_000_000_000),
Some("beta"),
)
.await
.unwrap();
assert_eq!(
alpha
.get_latest_application_version(Some("beta"))
.await
.unwrap()
.unwrap()
.version_name,
"beta-v1",
"an older version promoted past a newer one is what a rollback looks like",
);
alpha
.create_application_version("shared", None)
.await
.unwrap();
let stolen = client
.create_application_version("shared", Some("beta"))
.await;
assert!(matches!(stolen, Err(Error::RegisteredByAnother { .. })));
}
#[tokio::test]
async fn a_workflow_can_be_enqueued_for_another_application() {
let db = test_database().await;
let pool = db.pool().await;
let alpha = PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some("alpha"),
..Settings::default()
},
);
let mut for_beta = workflow("wf-for-beta");
for_beta.application_name = Some("beta");
alpha
.init_workflow(&for_beta, None, Submission::Fresh, None)
.await
.unwrap();
alpha
.init_workflow(&workflow("wf-for-alpha"), None, Submission::Fresh, None)
.await
.unwrap();
for (id, expected) in [("wf-for-beta", "beta"), ("wf-for-alpha", "alpha")] {
assert_eq!(
alpha
.get_workflow(id)
.await
.unwrap()
.unwrap()
.application_name
.as_deref(),
Some(expected),
);
}
assert_eq!(ids(&alpha, &pending("local", "v1")).await, ["wf-for-alpha"]);
}
#[tokio::test]
async fn two_applications_cannot_both_claim_one_version() {
let db = test_database().await;
let pool = db.pool().await;
let named = |name: &'static str| {
PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some(name),
..Settings::default()
},
)
};
let (alpha, beta) = (named("alpha"), named("beta"));
let anonymous = PostgresSystemDatabase::from_pool(pool.clone(), &Settings::default());
anonymous
.create_application_version("contested", None)
.await
.unwrap();
let (a, b) = tokio::join!(
alpha.create_application_version("contested", None),
beta.create_application_version("contested", None),
);
assert_eq!(
[a.is_ok(), b.is_ok()].iter().filter(|ok| **ok).count(),
1,
"exactly one application may hold a version name; got alpha={a:?} beta={b:?}",
);
let loser = if a.is_err() { &a } else { &b };
assert!(
matches!(loser, Err(Error::RegisteredByAnother { .. })),
"the loser should be told why, got {loser:?}",
);
let owner: Option<String> = sqlx::query_scalar(
r#"SELECT "application_name" FROM "dbos"."application_versions" WHERE "version_name" = $1"#,
)
.bind("contested")
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(
owner.as_deref(),
Some(if a.is_ok() { "alpha" } else { "beta" })
);
}
async fn register_queue_and_schedule(pool: &sqlx::PgPool, suffix: &str, owner: Option<&str>) {
sqlx::query(r#"INSERT INTO "dbos"."queues" ("name", "application_name") VALUES ($1, $2)"#)
.bind(format!("queue-{suffix}"))
.bind(owner)
.execute(pool)
.await
.unwrap();
sqlx::query(
r#"INSERT INTO "dbos"."workflow_schedules"
("schedule_id", "schedule_name", "workflow_name", "schedule", "context", "application_name")
VALUES ($1, $2, 'nightly', '0 0 * * *', '{}', $3)"#,
)
.bind(format!("sched-id-{suffix}"))
.bind(format!("sched-{suffix}"))
.bind(owner)
.execute(pool)
.await
.unwrap();
}
async fn owner_of(pool: &sqlx::PgPool, table: &str, key_column: &str, key: &str) -> Option<String> {
sqlx::query_scalar(sqlx::AssertSqlSafe(format!(
r#"SELECT "application_name" FROM "dbos"."{table}" WHERE "{key_column}" = $1"#
)))
.bind(key)
.fetch_one(pool)
.await
.unwrap()
}
#[tokio::test]
async fn a_rename_moves_every_kind_of_row() {
let db = test_database().await;
let pool = db.pool().await;
let alpha = PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some("alpha"),
..Settings::default()
},
);
register_queue_and_schedule(&pool, "a", Some("alpha")).await;
alpha.create_application_version("v1", None).await.unwrap();
alpha
.init_workflow(&workflow("wf-running"), None, Submission::Fresh, None)
.await
.unwrap();
alpha
.init_workflow(&workflow("wf-done"), None, Submission::Fresh, None)
.await
.unwrap();
alpha
.record_step(
"wf-done",
0,
"a_step",
Outcome::Output(Some("\"x\"")),
None,
None,
)
.await
.unwrap();
alpha
.record_workflow_outcome("wf-done", Outcome::Output(Some("\"done\"")))
.await
.unwrap();
let counts = alpha
.rename_application(
RenameFrom::Application("alpha"),
"alpha-renamed",
RenameBatching::default(),
)
.await
.unwrap();
assert_eq!(
counts,
dbos::sysdb::types::ApplicationRowCounts {
queues: 1,
schedules: 1,
versions: 1,
workflows: 2, steps: 1,
},
);
for (table, key_column, key) in [
("queues", "name", "queue-a"),
("workflow_schedules", "schedule_name", "sched-a"),
("application_versions", "version_name", "v1"),
("workflow_status", "workflow_uuid", "wf-running"),
("workflow_status", "workflow_uuid", "wf-done"),
("operation_outputs", "workflow_uuid", "wf-done"),
] {
assert_eq!(
owner_of(&pool, table, key_column, key).await.as_deref(),
Some("alpha-renamed"),
"{table}.{key} did not move",
);
}
}
#[tokio::test]
async fn a_rename_takes_unclaimed_rows_only_when_asked() {
let db = test_database().await;
let pool = db.pool().await;
let named = |name: &'static str| {
PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some(name),
..Settings::default()
},
)
};
let alpha = named("alpha");
let anonymous = PostgresSystemDatabase::from_pool(pool.clone(), &Settings::default());
alpha
.init_workflow(&workflow("wf-alpha"), None, Submission::Fresh, None)
.await
.unwrap();
anonymous
.init_workflow(&workflow("wf-nobody"), None, Submission::Fresh, None)
.await
.unwrap();
let counts = alpha
.rename_application(
RenameFrom::Application("alpha"),
"alpha-two",
RenameBatching::Unbatched,
)
.await
.unwrap();
assert_eq!(counts.workflows, 1);
assert_eq!(
owner_of(&pool, "workflow_status", "workflow_uuid", "wf-nobody").await,
None,
"an unclaimed row is not swept up by a plain rename",
);
let counts = alpha
.rename_application(
RenameFrom::Unclaimed,
"alpha-two",
RenameBatching::Unbatched,
)
.await
.unwrap();
assert_eq!(counts.workflows, 1);
assert_eq!(
owner_of(&pool, "workflow_status", "workflow_uuid", "wf-nobody")
.await
.as_deref(),
Some("alpha-two"),
);
}
#[tokio::test]
async fn a_batched_rename_loses_no_rows() {
let db = test_database().await;
let pool = db.pool().await;
let alpha = PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some("alpha"),
..Settings::default()
},
);
for i in 0..7 {
let id = format!("wf-{i:02}");
alpha
.init_workflow(&workflow(&id), None, Submission::Fresh, None)
.await
.unwrap();
for step in 0..2 {
alpha
.record_step(
&id,
step,
"a_step",
Outcome::Output(Some("\"x\"")),
None,
None,
)
.await
.unwrap();
}
alpha
.record_workflow_outcome(&id, Outcome::Output(Some("\"done\"")))
.await
.unwrap();
}
let counts = alpha
.rename_application(
RenameFrom::Application("alpha"),
"alpha-batched",
RenameBatching::Batched(2),
)
.await
.unwrap();
assert_eq!(counts.workflows, 7, "every workflow moved");
assert_eq!(counts.steps, 14, "every step moved");
let stragglers: i64 = sqlx::query_scalar(
r#"SELECT count(*) FROM "dbos"."workflow_status" WHERE "application_name" = 'alpha'"#,
)
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(stragglers, 0, "no workflow was left behind by the batching");
}
#[tokio::test]
async fn a_rename_refuses_a_bad_target_name() {
let (sys, _db) = sysdb().await;
for bad in ["ab", "Has-Capitals", "has spaces", &"x".repeat(257)] {
let result = sys
.rename_application(
RenameFrom::Application("alpha"),
bad,
RenameBatching::Unbatched,
)
.await;
assert!(
matches!(
result,
Err(Error::InvalidInput { ref field, .. }) if field == "new_name"
),
"{bad:?} should be rejected, got {result:?}",
);
}
let result = sys
.rename_application(
RenameFrom::Application("alpha"),
"alpha",
RenameBatching::Unbatched,
)
.await;
assert!(matches!(
result,
Err(Error::InvalidInput { ref field, .. }) if field == "new_name"
));
}
#[tokio::test]
async fn a_queue_round_trips_through_the_registry() {
let (sys, _db) = sysdb().await;
let queue = NewQueue {
concurrency: Some(8),
worker_concurrency: Some(2),
rate_limit: Some(RateLimit {
limit: 100,
period: std::time::Duration::from_millis(1_500),
}),
priority_enabled: true,
partition_queue: true,
polling_interval: std::time::Duration::from_millis(250),
..NewQueue::new("orders")
};
assert!(
sys.upsert_queue(&queue, OnExistingQueue::Update)
.await
.unwrap(),
"the first registration creates the row",
);
let read = sys.get_queue("orders").await.unwrap().expect("registered");
assert_eq!(read.name, "orders");
assert_eq!(read.concurrency, Some(8));
assert_eq!(read.worker_concurrency, Some(2));
assert_eq!(
read.rate_limit,
Some(RateLimit {
limit: 100,
period: std::time::Duration::from_millis(1_500),
}),
);
assert!(read.priority_enabled);
assert!(read.partition_queue);
assert_eq!(read.polling_interval, std::time::Duration::from_millis(250));
assert!(sys.get_queue("no-such-queue").await.unwrap().is_none());
}
#[tokio::test]
async fn re_registering_a_queue_reports_it_existed() {
let (sys, _db) = sysdb().await;
let first = NewQueue {
concurrency: Some(1),
..NewQueue::new("orders")
};
assert!(
sys.upsert_queue(&first, OnExistingQueue::Update)
.await
.unwrap()
);
let second = NewQueue {
concurrency: Some(9),
..NewQueue::new("orders")
};
assert!(
!sys.upsert_queue(&second, OnExistingQueue::Update)
.await
.unwrap(),
"the second registration finds the row already there",
);
assert_eq!(
sys.get_queue("orders").await.unwrap().unwrap().concurrency,
Some(9),
"Update overwrites the stored limits",
);
let third = NewQueue {
concurrency: Some(3),
..NewQueue::new("orders")
};
assert!(
!sys.upsert_queue(&third, OnExistingQueue::Leave)
.await
.unwrap()
);
assert_eq!(
sys.get_queue("orders").await.unwrap().unwrap().concurrency,
Some(9),
"Leave keeps what is stored",
);
}
#[tokio::test]
async fn a_peers_queue_name_is_refused() {
let db = test_database().await;
let pool = db.pool().await;
let named = |name: &'static str| {
PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some(name),
..Settings::default()
},
)
};
let (alpha, beta) = (named("alpha"), named("beta"));
alpha
.upsert_queue(&NewQueue::new("orders"), OnExistingQueue::Update)
.await
.unwrap();
for mode in [OnExistingQueue::Update, OnExistingQueue::Leave] {
let result = beta.upsert_queue(&NewQueue::new("orders"), mode).await;
match result {
Err(Error::RegisteredByAnother {
kind,
ref holder,
ref claimant,
..
}) => {
assert_eq!(kind, "Queue");
assert_eq!(holder, "alpha");
assert_eq!(claimant.as_deref(), Some("beta"));
}
other => panic!("{mode:?} should be refused, got {other:?}"),
}
}
assert_eq!(
sys_owner(&pool, "orders").await.as_deref(),
Some("alpha"),
"a refused registration changes nothing",
);
alpha
.upsert_queue(&NewQueue::new("orders"), OnExistingQueue::Update)
.await
.unwrap();
let anonymous = PostgresSystemDatabase::from_pool(pool.clone(), &Settings::default());
anonymous
.upsert_queue(&NewQueue::new("orders"), OnExistingQueue::Update)
.await
.unwrap();
assert_eq!(sys_owner(&pool, "orders").await.as_deref(), Some("alpha"));
}
async fn sys_owner(pool: &sqlx::PgPool, queue: &str) -> Option<String> {
sqlx::query_scalar(r#"SELECT "application_name" FROM "dbos"."queues" WHERE "name" = $1"#)
.bind(queue)
.fetch_one(pool)
.await
.unwrap()
}
#[tokio::test]
async fn listing_queues_scopes_but_reading_one_does_not() {
let db = test_database().await;
let pool = db.pool().await;
let named = |name: &'static str| {
PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some(name),
..Settings::default()
},
)
};
let (alpha, beta) = (named("alpha"), named("beta"));
let anonymous = PostgresSystemDatabase::from_pool(pool.clone(), &Settings::default());
for (sys, name) in [
(&alpha, "alpha-queue"),
(&beta, "beta-queue"),
(&anonymous, "nobodys-queue"),
] {
sys.upsert_queue(&NewQueue::new(name), OnExistingQueue::Update)
.await
.unwrap();
}
let names = |queues: Vec<dbos::sysdb::types::QueueRecord>| {
queues.into_iter().map(|q| q.name).collect::<Vec<_>>()
};
assert_eq!(
names(alpha.list_queues(&Applications::Unset).await.unwrap()),
["alpha-queue", "nobodys-queue"],
"a search defaults to this application's own plus the unclaimed",
);
assert_eq!(
names(alpha.list_queues(&Applications::Any).await.unwrap()),
["alpha-queue", "beta-queue", "nobodys-queue"],
);
assert_eq!(
names(
alpha
.list_queues(&Applications::Named(vec!["beta"]))
.await
.unwrap()
),
["beta-queue", "nobodys-queue"],
);
assert_eq!(
alpha.get_queue("beta-queue").await.unwrap().unwrap().name,
"beta-queue",
);
alpha.delete_queue("alpha-queue").await.unwrap();
assert!(alpha.get_queue("alpha-queue").await.unwrap().is_none());
alpha.delete_queue("alpha-queue").await.unwrap();
}
#[tokio::test]
async fn half_a_rate_limit_reads_as_none() {
let (sys, db) = sysdb().await;
let pool = db.pool().await;
sqlx::query(
r#"INSERT INTO "dbos"."queues" ("name", "rate_limit_max") VALUES ('half-limit', 10)"#,
)
.execute(&pool)
.await
.unwrap();
let queue = sys.get_queue("half-limit").await.unwrap().expect("exists");
assert_eq!(queue.rate_limit, None);
}
fn skip_dequeue_completeness(db: &support::TestDatabase) -> bool {
let skipping = matches!(db.backend(), Backend::Cockroach);
if skipping {
eprintln!("skipped on CockroachDB: SKIP LOCKED may under-deliver; see UPSTREAM item 7");
}
skipping
}
async fn enqueue(
sys: &PostgresSystemDatabase,
id: &str,
queue: &str,
priority: i32,
timeout: Option<std::time::Duration>,
) {
let wf = NewWorkflow {
queue_name: Some(queue),
priority,
timeout,
application_version: Some("v1"),
..NewWorkflow::new(id)
};
sys.init_workflow(&wf, None, Submission::Fresh, None)
.await
.unwrap();
}
#[tokio::test]
async fn a_dequeue_starts_workflows_in_order() {
let (sys, db) = sysdb().await;
if skip_dequeue_completeness(&db) {
return;
}
let queue = NewQueue::new("orders");
sys.upsert_queue(&queue, OnExistingQueue::Update)
.await
.unwrap();
let registered = sys.get_queue("orders").await.unwrap().unwrap();
enqueue(&sys, "wf-low", "orders", 10, None).await;
enqueue(&sys, "wf-high", "orders", 1, None).await;
enqueue(&sys, "wf-also-high", "orders", 1, None).await;
let started = sys
.start_queued_workflows(®istered, "exec-1", "v1", None, 0, 0)
.await
.unwrap();
assert_eq!(
started,
["wf-high", "wf-also-high", "wf-low"],
"priority first, then age within a priority",
);
let read = sys.get_workflow("wf-high").await.unwrap().unwrap();
assert_eq!(read.status, WorkflowStatus::Pending);
assert_eq!(read.executor_id.as_deref(), Some("exec-1"));
assert!(read.started_at.is_some());
assert!(
sys.start_queued_workflows(®istered, "exec-1", "v1", None, 0, 0)
.await
.unwrap()
.is_empty()
);
}
#[tokio::test]
async fn a_dequeue_sets_the_deadline_from_the_timeout() {
let (sys, db) = sysdb().await;
if skip_dequeue_completeness(&db) {
return;
}
sys.upsert_queue(&NewQueue::new("orders"), OnExistingQueue::Update)
.await
.unwrap();
let registered = sys.get_queue("orders").await.unwrap().unwrap();
enqueue(
&sys,
"wf-timed",
"orders",
0,
Some(std::time::Duration::from_secs(60)),
)
.await;
enqueue(&sys, "wf-untimed", "orders", 0, None).await;
let before = Timestamp::now();
sys.start_queued_workflows(®istered, "exec-1", "v1", None, 0, 0)
.await
.unwrap();
let timed = sys.get_workflow("wf-timed").await.unwrap().unwrap();
let deadline = timed
.deadline
.expect("a timeout becomes a deadline on dequeue");
let offset = deadline.as_epoch_ms() - before.as_epoch_ms();
assert!(
(60_000..=65_000).contains(&offset),
"the deadline is the start plus the timeout, got {offset}ms",
);
assert_eq!(
sys.get_workflow("wf-untimed")
.await
.unwrap()
.unwrap()
.deadline,
None,
"a workflow with no timeout gets no deadline",
);
}
#[tokio::test]
async fn worker_concurrency_bounds_a_dequeue() {
let (sys, db) = sysdb().await;
if skip_dequeue_completeness(&db) {
return;
}
let queue = NewQueue {
worker_concurrency: Some(2),
..NewQueue::new("orders")
};
sys.upsert_queue(&queue, OnExistingQueue::Update)
.await
.unwrap();
let registered = sys.get_queue("orders").await.unwrap().unwrap();
for i in 0..5 {
enqueue(&sys, &format!("wf-{i}"), "orders", 0, None).await;
}
assert_eq!(
sys.start_queued_workflows(®istered, "exec-1", "v1", None, 0, 0)
.await
.unwrap()
.len(),
2,
"a free worker takes its whole allowance",
);
assert_eq!(
sys.start_queued_workflows(®istered, "exec-1", "v1", None, 1, 0)
.await
.unwrap()
.len(),
1,
"one already running leaves room for one",
);
assert!(
sys.start_queued_workflows(®istered, "exec-1", "v1", None, 2, 0)
.await
.unwrap()
.is_empty(),
"a full worker takes nothing",
);
}
#[tokio::test]
async fn concurrency_counts_across_executors() {
let (sys, db) = sysdb().await;
if skip_dequeue_completeness(&db) {
return;
}
let queue = NewQueue {
concurrency: Some(2),
..NewQueue::new("orders")
};
sys.upsert_queue(&queue, OnExistingQueue::Update)
.await
.unwrap();
let registered = sys.get_queue("orders").await.unwrap().unwrap();
for i in 0..4 {
enqueue(&sys, &format!("wf-{i}"), "orders", 0, None).await;
}
assert_eq!(
sys.start_queued_workflows(®istered, "exec-other", "v1", None, 0, 0)
.await
.unwrap()
.len(),
2,
);
assert!(
sys.start_queued_workflows(®istered, "exec-1", "v1", None, 0, 0)
.await
.unwrap()
.is_empty(),
"the limit is global, so a second executor sees none free",
);
}
#[tokio::test]
async fn a_rate_limit_bounds_starts_per_window() {
let (sys, db) = sysdb().await;
if skip_dequeue_completeness(&db) {
return;
}
let queue = NewQueue {
rate_limit: Some(RateLimit {
limit: 2,
period: std::time::Duration::from_millis(400),
}),
..NewQueue::new("orders")
};
sys.upsert_queue(&queue, OnExistingQueue::Update)
.await
.unwrap();
let registered = sys.get_queue("orders").await.unwrap().unwrap();
for i in 0..5 {
enqueue(&sys, &format!("wf-{i}"), "orders", 0, None).await;
}
assert_eq!(
sys.start_queued_workflows(®istered, "exec-1", "v1", None, 0, 0)
.await
.unwrap()
.len(),
2,
"the window allows two",
);
assert!(
sys.start_queued_workflows(®istered, "exec-1", "v1", None, 0, 0)
.await
.unwrap()
.is_empty(),
"and no more until it passes",
);
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
assert_eq!(
sys.start_queued_workflows(®istered, "exec-1", "v1", None, 0, 0)
.await
.unwrap()
.len(),
2,
"a fresh window allows two more",
);
}
#[tokio::test]
async fn a_partition_rate_limit_bounds_each_key_separately() {
let (sys, db) = sysdb().await;
if skip_dequeue_completeness(&db) {
return;
}
let queue = NewQueue {
partition_rate_limit: Some(RateLimit {
limit: 2,
period: std::time::Duration::from_millis(400),
}),
..NewQueue::new("orders")
};
sys.upsert_queue(&queue, OnExistingQueue::Update)
.await
.unwrap();
let registered = sys.get_queue("orders").await.unwrap().unwrap();
for partition in ["a", "b"] {
for i in 0..3 {
enqueue_partitioned(&sys, &format!("wf-{partition}-{i}"), "orders", partition).await;
}
}
let dequeue = async |partition: &str| {
sys.start_queued_workflows(®istered, "exec-1", "v1", Some(partition), 0, 0)
.await
.unwrap()
.len()
};
assert_eq!(dequeue("a").await, 2, "the window allows two");
assert_eq!(dequeue("a").await, 0, "and no more until it passes");
assert_eq!(dequeue("b").await, 2, "another key has its own window");
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
assert_eq!(dequeue("a").await, 1, "a fresh window allows the last one");
}
#[tokio::test]
async fn a_dequeue_claims_what_it_starts() {
let db = test_database().await;
if skip_dequeue_completeness(&db) {
return;
}
let pool = db.pool().await;
let named = |name: &'static str| {
PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some(name),
..Settings::default()
},
)
};
let (alpha, beta) = (named("alpha"), named("beta"));
let anonymous = PostgresSystemDatabase::from_pool(pool.clone(), &Settings::default());
alpha
.upsert_queue(&NewQueue::new("orders"), OnExistingQueue::Update)
.await
.unwrap();
let registered = alpha.get_queue("orders").await.unwrap().unwrap();
enqueue(&alpha, "wf-alpha", "orders", 0, None).await;
enqueue(&beta, "wf-beta", "orders", 0, None).await;
enqueue(&anonymous, "wf-nobody", "orders", 0, None).await;
let mut started = alpha
.start_queued_workflows(®istered, "exec-1", "v1", None, 0, 0)
.await
.unwrap();
started.sort();
assert_eq!(
started,
["wf-alpha", "wf-nobody"],
"alpha takes its own and the unclaimed, never beta's",
);
assert_eq!(
alpha
.get_workflow("wf-nobody")
.await
.unwrap()
.unwrap()
.application_name
.as_deref(),
Some("alpha"),
"starting an unclaimed workflow claims it",
);
assert_eq!(
alpha.get_workflow("wf-beta").await.unwrap().unwrap().status,
WorkflowStatus::Enqueued,
"beta's workflow is untouched",
);
}
#[tokio::test]
async fn unversioned_work_goes_to_the_latest_version() {
let (sys, db) = sysdb().await;
if skip_dequeue_completeness(&db) {
return;
}
sys.upsert_queue(&NewQueue::new("orders"), OnExistingQueue::Update)
.await
.unwrap();
let registered = sys.get_queue("orders").await.unwrap().unwrap();
sys.create_application_version("v2", None).await.unwrap();
let unversioned = NewWorkflow {
queue_name: Some("orders"),
..NewWorkflow::new("wf-unversioned")
};
sys.init_workflow(&unversioned, None, Submission::Fresh, None)
.await
.unwrap();
assert!(
sys.start_queued_workflows(®istered, "exec-old", "v1", None, 0, 0)
.await
.unwrap()
.is_empty(),
"only the latest version adopts unversioned work",
);
assert_eq!(
sys.start_queued_workflows(®istered, "exec-new", "v2", None, 0, 0)
.await
.unwrap(),
["wf-unversioned"],
);
}
#[tokio::test]
async fn an_empty_partition_key_is_refused() {
let (sys, db) = sysdb().await;
if skip_dequeue_completeness(&db) {
return;
}
sys.upsert_queue(&NewQueue::new("orders"), OnExistingQueue::Update)
.await
.unwrap();
let registered = sys.get_queue("orders").await.unwrap().unwrap();
let result = sys
.start_queued_workflows(®istered, "exec-1", "v1", Some(""), 0, 0)
.await;
assert!(
matches!(
result,
Err(Error::InvalidInput { ref field, .. }) if field == "partition_key"
),
"got {result:?}",
);
enqueue(&sys, "wf-1", "orders", 0, None).await;
assert_eq!(
sys.start_queued_workflows(®istered, "exec-1", "v1", None, 0, 0)
.await
.unwrap(),
["wf-1"],
);
}
#[tokio::test]
async fn an_unrepresentable_period_is_malformed() {
let (sys, db) = sysdb().await;
let pool = db.pool().await;
sqlx::query(
r#"INSERT INTO "dbos"."queues" ("name", "polling_interval_sec") VALUES ('huge', 1e300)"#,
)
.execute(&pool)
.await
.unwrap();
match sys.get_queue("huge").await {
Err(Error::Malformed(message)) => {
assert!(message.contains("polling_interval_sec"), "got {message}");
}
other => panic!("expected a malformed row, got {other:?}"),
}
}
async fn partitioned_queue(sys: &PostgresSystemDatabase, name: &'static str) -> QueueRecord {
let queue = NewQueue {
concurrency: Some(1),
partition_queue: true,
..NewQueue::new(name)
};
sys.upsert_queue(&queue, OnExistingQueue::Update)
.await
.unwrap();
sys.get_queue(name).await.unwrap().unwrap()
}
async fn enqueue_partitioned(sys: &PostgresSystemDatabase, id: &str, queue: &str, partition: &str) {
let wf = NewWorkflow {
queue_name: Some(queue),
queue_partition_key: Some(partition),
application_version: Some("v1"),
..NewWorkflow::new(id)
};
sys.init_workflow(&wf, None, Submission::Fresh, None)
.await
.unwrap();
}
#[tokio::test]
async fn partition_concurrency_bounds_a_dequeue() {
let (sys, db) = sysdb().await;
if skip_dequeue_completeness(&db) {
return;
}
let queue = NewQueue {
partition_concurrency: Some(2),
..NewQueue::new("orders")
};
sys.upsert_queue(&queue, OnExistingQueue::Update)
.await
.unwrap();
let registered = sys.get_queue("orders").await.unwrap().unwrap();
for partition in ["alpha", "beta"] {
for n in 0..3 {
enqueue_partitioned(&sys, &format!("wf-{partition}-{n}"), "orders", partition).await;
}
}
let alpha = sys
.start_queued_workflows(®istered, "exec-1", "v1", Some("alpha"), 0, 0)
.await
.unwrap();
assert_eq!(alpha.len(), 2, "the partition's limit, not the queue's");
let beta = sys
.start_queued_workflows(®istered, "exec-1", "v1", Some("beta"), 0, 0)
.await
.unwrap();
assert_eq!(
beta.len(),
2,
"a full partition does not spend another partition's budget"
);
assert!(
sys.start_queued_workflows(®istered, "exec-1", "v1", Some("alpha"), 0, 0)
.await
.unwrap()
.is_empty(),
);
}
#[tokio::test]
async fn the_queue_wide_limit_bounds_the_sum_of_the_partitions() {
let (sys, db) = sysdb().await;
if skip_dequeue_completeness(&db) {
return;
}
let queue = NewQueue {
concurrency: Some(3),
partition_concurrency: Some(2),
..NewQueue::new("orders")
};
sys.upsert_queue(&queue, OnExistingQueue::Update)
.await
.unwrap();
let registered = sys.get_queue("orders").await.unwrap().unwrap();
for partition in ["alpha", "beta"] {
for n in 0..3 {
enqueue_partitioned(&sys, &format!("wf-{partition}-{n}"), "orders", partition).await;
}
}
let alpha = sys
.start_queued_workflows(®istered, "exec-1", "v1", Some("alpha"), 0, 0)
.await
.unwrap();
assert_eq!(alpha.len(), 2, "its own per-partition allowance");
let beta = sys
.start_queued_workflows(®istered, "exec-1", "v1", Some("beta"), 0, 0)
.await
.unwrap();
assert_eq!(
beta.len(),
1,
"the queue-wide limit counts every partition's PENDING rows, not just this key's"
);
assert!(
sys.start_queued_workflows(®istered, "exec-1", "v1", Some("beta"), 0, 0)
.await
.unwrap()
.is_empty(),
"the queue is at its limit whichever partition asks"
);
}
#[tokio::test]
async fn both_worker_limits_are_counted_at_their_own_scope() {
let (sys, db) = sysdb().await;
if skip_dequeue_completeness(&db) {
return;
}
let queue = NewQueue {
worker_concurrency: Some(3),
partition_worker_concurrency: Some(2),
..NewQueue::new("orders")
};
sys.upsert_queue(&queue, OnExistingQueue::Update)
.await
.unwrap();
let registered = sys.get_queue("orders").await.unwrap().unwrap();
for n in 0..4 {
enqueue_partitioned(&sys, &format!("wf-{n}"), "orders", "alpha").await;
}
let claimed = sys
.start_queued_workflows(®istered, "exec-1", "v1", Some("alpha"), 0, 0)
.await
.unwrap();
assert_eq!(claimed.len(), 2, "bounded by partition_worker_concurrency");
let claimed = sys
.start_queued_workflows(®istered, "exec-1", "v1", Some("alpha"), 0, 1)
.await
.unwrap();
assert_eq!(claimed.len(), 1, "the partition-local count is subtracted");
assert!(
sys.start_queued_workflows(®istered, "exec-1", "v1", Some("alpha"), 3, 0)
.await
.unwrap()
.is_empty(),
"the queue-wide worker limit still binds"
);
}
#[tokio::test]
async fn a_bounded_sweep_does_not_always_take_the_lowest_keys() {
let (sys, db) = sysdb().await;
if skip_dequeue_completeness(&db) {
return;
}
let queue = partitioned_queue(&sys, "orders").await;
for n in 0..12 {
enqueue_partitioned(&sys, &format!("wf-{n:02}"), "orders", &format!("p-{n:02}")).await;
}
let mut first_of_each = Vec::new();
for _ in 0..6 {
let claimed = sys
.start_queued_partitioned_workflows(&queue, "exec-1", "v1", Some(1))
.await
.unwrap();
assert_eq!(claimed.len(), 1, "the budget bounds the sweep to one head");
first_of_each.push(claimed[0].clone());
}
let in_key_order: Vec<String> = (0..6).map(|n| format!("wf-{n:02}")).collect();
assert_ne!(
first_of_each, in_key_order,
"a bounded sweep took the partitions in key order, which starves the tail"
);
assert_eq!(
first_of_each
.iter()
.collect::<std::collections::HashSet<_>>()
.len(),
6,
"each sweep claimed a partition the previous ones had not"
);
}
#[tokio::test]
async fn queue_partitions_are_the_keys_with_work() {
let (sys, _db) = sysdb().await;
partitioned_queue(&sys, "orders").await;
enqueue_partitioned(&sys, "wf-a1", "orders", "alpha").await;
enqueue_partitioned(&sys, "wf-a2", "orders", "alpha").await;
enqueue_partitioned(&sys, "wf-b1", "orders", "beta").await;
enqueue(&sys, "wf-none", "orders", 0, None).await;
assert_eq!(
sys.get_queue_partitions("orders").await.unwrap(),
["alpha", "beta"],
"each key once, in order, and no null key",
);
assert!(sys.get_queue_partitions("other").await.unwrap().is_empty());
}
#[tokio::test]
async fn a_sweep_takes_one_head_per_partition() {
let (sys, db) = sysdb().await;
if skip_dequeue_completeness(&db) {
return;
}
let queue = partitioned_queue(&sys, "orders").await;
for partition in ["alpha", "beta"] {
for n in 0..2 {
enqueue_partitioned(&sys, &format!("wf-{partition}-{n}"), "orders", partition).await;
}
}
let mut started = sys
.start_queued_partitioned_workflows(&queue, "exec-1", "v1", None)
.await
.unwrap();
started.sort();
assert_eq!(
started,
["wf-alpha-0", "wf-beta-0"],
"the head of each partition, never two from one",
);
assert!(
sys.start_queued_partitioned_workflows(&queue, "exec-1", "v1", None)
.await
.unwrap()
.is_empty(),
"a partition whose head is running is not eligible",
);
sys.record_workflow_outcome("wf-alpha-0", Outcome::Output(Some("\"done\"")))
.await
.unwrap();
assert_eq!(
sys.start_queued_partitioned_workflows(&queue, "exec-1", "v1", None)
.await
.unwrap(),
["wf-alpha-1"],
);
}
#[tokio::test]
async fn a_sweep_refuses_an_unsuitable_queue() {
let (sys, _db) = sysdb().await;
let unsuitable = [
NewQueue {
concurrency: Some(1),
..NewQueue::new("plain")
},
NewQueue {
concurrency: Some(2),
partition_queue: true,
..NewQueue::new("wide")
},
NewQueue {
concurrency: Some(1),
partition_queue: true,
rate_limit: Some(RateLimit {
limit: 5,
period: std::time::Duration::from_secs(1),
}),
..NewQueue::new("limited")
},
NewQueue {
partition_concurrency: Some(2),
..NewQueue::new("wide-partition")
},
NewQueue {
concurrency: Some(4),
partition_concurrency: Some(1),
..NewQueue::new("also-global")
},
NewQueue {
partition_concurrency: Some(1),
partition_rate_limit: Some(RateLimit {
limit: 5,
period: std::time::Duration::from_secs(1),
}),
..NewQueue::new("partition-limited")
},
];
for queue in unsuitable {
sys.upsert_queue(&queue, OnExistingQueue::Update)
.await
.unwrap();
let registered = sys.get_queue(queue.name).await.unwrap().unwrap();
let result = sys
.start_queued_partitioned_workflows(®istered, "exec-1", "v1", None)
.await;
assert!(
matches!(
result,
Err(Error::InvalidInput { ref field, .. }) if field == "queue"
),
"{} should be refused, got {result:?}",
queue.name,
);
}
}
#[tokio::test]
async fn a_sweep_stays_within_an_application() {
let db = test_database().await;
if skip_dequeue_completeness(&db) {
return;
}
let pool = db.pool().await;
let alpha = PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some("alpha"),
..Settings::default()
},
);
let beta = PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some("beta"),
..Settings::default()
},
);
let queue = partitioned_queue(&alpha, "orders").await;
enqueue_partitioned(&alpha, "wf-alpha", "orders", "p1").await;
enqueue_partitioned(&beta, "wf-beta", "orders", "p2").await;
assert_eq!(
alpha.get_queue_partitions("orders").await.unwrap(),
["p1"],
"beta's partition is not alpha's to poll",
);
assert_eq!(
alpha
.start_queued_partitioned_workflows(&queue, "exec-1", "v1", None)
.await
.unwrap(),
["wf-alpha"],
);
assert_eq!(
alpha.get_workflow("wf-beta").await.unwrap().unwrap().status,
WorkflowStatus::Enqueued,
);
}
#[tokio::test]
async fn an_update_changes_only_what_it_names() {
let (sys, _db) = sysdb().await;
let queue = NewQueue {
concurrency: Some(4),
worker_concurrency: Some(2),
rate_limit: Some(RateLimit {
limit: 10,
period: std::time::Duration::from_secs(1),
}),
priority_enabled: true,
polling_interval: std::time::Duration::from_millis(500),
..NewQueue::new("orders")
};
sys.upsert_queue(&queue, OnExistingQueue::Update)
.await
.unwrap();
sys.update_queue(
"orders",
&QueueUpdate {
concurrency: Change::Set(Some(9)),
worker_concurrency: Change::Set(None),
..QueueUpdate::default()
},
&|_, _| Ok(()),
)
.await
.unwrap();
let read = sys.get_queue("orders").await.unwrap().unwrap();
assert_eq!(read.concurrency, Some(9), "named and set");
assert_eq!(read.worker_concurrency, None, "named and cleared");
assert_eq!(
read.rate_limit,
Some(RateLimit {
limit: 10,
period: std::time::Duration::from_secs(1),
}),
"not named, so untouched",
);
assert!(read.priority_enabled, "not named, so untouched");
assert_eq!(read.polling_interval, std::time::Duration::from_millis(500));
}
#[tokio::test]
async fn an_update_moves_a_rate_limit_whole() {
let (sys, db) = sysdb().await;
let pool = db.pool().await;
let queue = NewQueue {
rate_limit: Some(RateLimit {
limit: 10,
period: std::time::Duration::from_secs(1),
}),
..NewQueue::new("orders")
};
sys.upsert_queue(&queue, OnExistingQueue::Update)
.await
.unwrap();
sys.update_queue(
"orders",
&QueueUpdate {
rate_limit: Change::Set(Some(RateLimit {
limit: 3,
period: std::time::Duration::from_millis(250),
})),
..QueueUpdate::default()
},
&|_, _| Ok(()),
)
.await
.unwrap();
assert_eq!(
sys.get_queue("orders").await.unwrap().unwrap().rate_limit,
Some(RateLimit {
limit: 3,
period: std::time::Duration::from_millis(250),
}),
);
sys.update_queue(
"orders",
&QueueUpdate {
rate_limit: Change::Set(None),
..QueueUpdate::default()
},
&|_, _| Ok(()),
)
.await
.unwrap();
assert_eq!(
sys.get_queue("orders").await.unwrap().unwrap().rate_limit,
None
);
let columns: (Option<i32>, Option<f64>) = sqlx::query_as(
r#"SELECT "rate_limit_max", "rate_limit_period_sec" FROM "dbos"."queues" WHERE "name" = $1"#,
)
.bind("orders")
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(columns, (None, None), "both columns cleared, not one");
}
#[tokio::test]
async fn an_empty_update_is_a_no_op() {
let (sys, db) = sysdb().await;
let pool = db.pool().await;
sys.upsert_queue(&NewQueue::new("orders"), OnExistingQueue::Update)
.await
.unwrap();
let before: i64 =
sqlx::query_scalar(r#"SELECT "updated_at" FROM "dbos"."queues" WHERE "name" = $1"#)
.bind("orders")
.fetch_one(&pool)
.await
.unwrap();
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
sys.update_queue("orders", &QueueUpdate::default(), &|_, _| Ok(()))
.await
.unwrap();
let after: i64 =
sqlx::query_scalar(r#"SELECT "updated_at" FROM "dbos"."queues" WHERE "name" = $1"#)
.bind("orders")
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(before, after, "an empty update records no write");
}
#[tokio::test]
async fn a_refused_update_writes_nothing() {
let (sys, db) = sysdb().await;
let pool = db.pool().await;
sys.upsert_queue(&NewQueue::new("orders"), OnExistingQueue::Update)
.await
.unwrap();
let before: i64 =
sqlx::query_scalar(r#"SELECT "updated_at" FROM "dbos"."queues" WHERE "name" = $1"#)
.bind("orders")
.fetch_one(&pool)
.await
.unwrap();
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
let error = sys
.update_queue(
"orders",
&QueueUpdate {
concurrency: Change::Set(Some(4)),
..QueueUpdate::default()
},
&|_, merged| {
assert_eq!(
merged.concurrency,
Some(4),
"the merged row is what is judged"
);
Err(Error::InvalidInput {
field: "concurrency".into(),
detail: "refused by the caller".to_owned(),
})
},
)
.await
.unwrap_err();
assert!(
matches!(&error, Error::InvalidInput { field, .. } if field == "concurrency"),
"the caller's own refusal comes back, got {error:?}"
);
let read = sys.get_queue("orders").await.unwrap().unwrap();
assert_eq!(read.concurrency, None, "the refused write did not land");
let after: i64 =
sqlx::query_scalar(r#"SELECT "updated_at" FROM "dbos"."queues" WHERE "name" = $1"#)
.bind("orders")
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(before, after, "not even `updated_at` moved");
}
#[tokio::test]
async fn updating_a_queue_that_is_not_registered_is_refused() {
let (sys, _db) = sysdb().await;
let error = sys
.update_queue(
"orders",
&QueueUpdate {
concurrency: Change::Set(Some(4)),
..QueueUpdate::default()
},
&|_, _| Ok(()),
)
.await
.unwrap_err();
assert!(
matches!(&error, Error::NotRegistered { kind, name } if kind == "Queue" && name == "orders"),
"got {error:?}"
);
}
#[tokio::test]
async fn the_deduplication_key_holder_is_read_by_queue_and_key() {
let (sys, _db) = sysdb().await;
let wf = NewWorkflow {
queue_name: Some("orders"),
deduplication_id: Some("cart-1"),
application_version: Some("v1"),
..NewWorkflow::new("wf-holder")
};
sys.init_workflow(&wf, None, Submission::Fresh, None)
.await
.unwrap();
assert_eq!(
sys.get_deduplication_key_holder("orders", "cart-1")
.await
.unwrap()
.as_deref(),
Some("wf-holder")
);
assert_eq!(
sys.get_deduplication_key_holder("shipping", "cart-1")
.await
.unwrap(),
None
);
assert_eq!(
sys.get_deduplication_key_holder("orders", "cart-2")
.await
.unwrap(),
None
);
}
#[tokio::test]
async fn finishing_releases_the_deduplication_key() {
let (sys, db) = sysdb().await;
if skip_dequeue_completeness(&db) {
return;
}
sys.upsert_queue(&NewQueue::new("orders"), OnExistingQueue::Update)
.await
.unwrap();
let held = NewWorkflow {
queue_name: Some("orders"),
deduplication_id: Some("cart-1"),
application_version: Some("v1"),
..NewWorkflow::new("wf-first")
};
sys.init_workflow(&held, None, Submission::Fresh, None)
.await
.unwrap();
let registered = sys.get_queue("orders").await.unwrap().unwrap();
assert_eq!(
sys.start_queued_workflows(®istered, "exec-1", "v1", None, 0, 0)
.await
.unwrap(),
["wf-first"],
);
assert_eq!(
sys.record_workflow_outcome("wf-first", Outcome::Output(Some("\"done\"")))
.await
.unwrap(),
OutcomeWrite::Recorded,
);
assert_eq!(
sys.get_deduplication_key_holder("orders", "cart-1")
.await
.unwrap(),
None,
"a finished workflow no longer holds its key"
);
let next = NewWorkflow {
queue_name: Some("orders"),
deduplication_id: Some("cart-1"),
application_version: Some("v1"),
..NewWorkflow::new("wf-next")
};
sys.init_workflow(&next, None, Submission::Fresh, None)
.await
.unwrap();
assert_eq!(
sys.get_deduplication_key_holder("orders", "cart-1")
.await
.unwrap()
.as_deref(),
Some("wf-next")
);
}
async fn schedule_names(sys: &PostgresSystemDatabase, filter: &ScheduleFilter<'_>) -> Vec<String> {
sys.list_schedules(filter, None)
.await
.unwrap()
.into_iter()
.map(|s| s.schedule_name)
.collect()
}
#[tokio::test]
async fn a_schedule_round_trips() {
let (sys, _db) = sysdb().await;
let schedule = NewSchedule {
schedule_id: Some("sch-1"),
workflow_class_name: Some("Reports"),
context: r#"{"tenant":"acme"}"#,
last_fired_at: Some(Timestamp::from_epoch_ms(1_786_492_800_000)),
automatic_backfill: true,
cron_timezone: Some("Europe/London"),
queue_name: Some("reports"),
..NewSchedule::new("nightly", "generate_report", "0 0 * * *")
};
sys.create_schedule(&schedule, None).await.unwrap();
let stored = sys.get_schedule("nightly", None).await.unwrap().unwrap();
assert_eq!(stored.schedule_id, "sch-1");
assert_eq!(stored.schedule_name, "nightly");
assert_eq!(stored.workflow_name, "generate_report");
assert_eq!(stored.workflow_class_name.as_deref(), Some("Reports"));
assert_eq!(stored.schedule, "0 0 * * *");
assert_eq!(stored.status, ScheduleStatus::Active);
assert_eq!(stored.context, r#"{"tenant":"acme"}"#);
assert_eq!(
stored.last_fired_at,
Some(Timestamp::from_epoch_ms(1_786_492_800_000)),
"the instant survives the column's ISO-8601 encoding"
);
assert!(stored.automatic_backfill);
assert_eq!(stored.cron_timezone.as_deref(), Some("Europe/London"));
assert_eq!(stored.queue_name.as_deref(), Some("reports"));
assert_eq!(sys.get_schedule("no-such", None).await.unwrap(), None);
}
#[tokio::test]
async fn a_peers_last_fired_at_spelling_is_read_as_an_instant() {
let db = test_database().await;
let pool = db.pool().await;
let sys = PostgresSystemDatabase::from_pool(pool.clone(), &Settings::default());
sys.create_schedule(
&NewSchedule::new("nightly", "generate_report", "0 0 * * *"),
None,
)
.await
.unwrap();
for stored in [
"2026-08-12T00:00:00.000Z",
"2026-08-12T00:00:00Z",
"2026-08-12T00:00:00+00:00",
"2026-08-12T00:00:00.000000000Z",
"2026-08-12T01:30:00+01:30",
] {
sqlx::query(
r#"UPDATE "dbos"."workflow_schedules" SET "last_fired_at" = $1
WHERE "schedule_name" = $2"#,
)
.bind(stored)
.bind("nightly")
.execute(&pool)
.await
.unwrap();
assert_eq!(
sys.get_schedule("nightly", None)
.await
.unwrap()
.unwrap()
.last_fired_at,
Some(Timestamp::from_epoch_ms(1_786_492_800_000)),
"{stored:?} is the same instant"
);
}
sqlx::query(
r#"UPDATE "dbos"."workflow_schedules" SET "last_fired_at" = 'yesterday'
WHERE "schedule_name" = $1"#,
)
.bind("nightly")
.execute(&pool)
.await
.unwrap();
assert!(matches!(
sys.get_schedule("nightly", None).await,
Err(Error::Malformed(_))
));
}
#[tokio::test]
async fn a_schedule_keeps_its_identity_across_a_re_apply() {
let (sys, _db) = sysdb().await;
sys.create_schedule(
&NewSchedule::new("nightly", "generate_report", "0 0 * * *"),
None,
)
.await
.unwrap();
let first = sys.get_schedule("nightly", None).await.unwrap().unwrap();
assert!(!first.schedule_id.is_empty(), "an id is generated");
sys.set_schedule_status("nightly", ScheduleStatus::Paused, None)
.await
.unwrap();
sys.update_schedule_last_fired_at("nightly", Timestamp::from_epoch_ms(1_786_492_800_000))
.await
.unwrap();
sys.upsert_schedule(
&NewSchedule {
workflow_class_name: Some("Reports"),
..NewSchedule::new("nightly", "generate_report", "*/5 * * * *")
},
None,
)
.await
.unwrap();
let after = sys.get_schedule("nightly", None).await.unwrap().unwrap();
assert_eq!(after.schedule_id, first.schedule_id, "the identity is kept");
assert_eq!(after.schedule, "*/5 * * * *", "the definition moved");
assert_eq!(after.workflow_class_name.as_deref(), Some("Reports"));
assert_eq!(
after.status,
ScheduleStatus::Paused,
"a redeployment does not resume a paused schedule"
);
assert_eq!(
after.last_fired_at,
Some(Timestamp::from_epoch_ms(1_786_492_800_000)),
"nor forget where it had got to"
);
}
#[tokio::test]
async fn creating_a_schedule_twice_is_refused() {
let (sys, _db) = sysdb().await;
let schedule = NewSchedule::new("nightly", "generate_report", "0 0 * * *");
sys.create_schedule(&schedule, None).await.unwrap();
assert!(matches!(
sys.create_schedule(&schedule, None).await,
Err(Error::AlreadyRegistered { ref kind, ref name, .. }) if kind == "Schedule" && name == "nightly"
));
let taken = sys.get_schedule("nightly", None).await.unwrap().unwrap();
assert!(matches!(
sys.create_schedule(
&NewSchedule {
schedule_id: Some(&taken.schedule_id),
..NewSchedule::new("hourly", "sweep", "0 * * * *")
},
None
)
.await,
Err(Error::AlreadyRegistered { ref kind, .. }) if kind == "Schedule id"
));
sys.upsert_schedule(&schedule, None).await.unwrap();
assert_eq!(
sys.list_schedules(&ScheduleFilter::default(), None)
.await
.unwrap()
.len(),
1
);
}
#[tokio::test]
async fn a_creation_losing_to_a_peer_reports_what_it_can_see() {
let db = test_database().await;
let pool = db.pool().await;
let alpha = PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some("alpha"),
..Settings::default()
},
);
let mut beta_tx = pool.begin().await.unwrap();
sqlx::query(
r#"INSERT INTO "dbos"."workflow_schedules"
(schedule_id, schedule_name, workflow_name, schedule, status, context, application_name)
VALUES ($1, $2, $3, $4, 'ACTIVE', 'null', 'beta')"#,
)
.bind("sch-beta")
.bind("nightly")
.bind("generate_report")
.bind("0 0 * * *")
.execute(&mut *beta_tx)
.await
.unwrap();
let racing = tokio::spawn(async move {
alpha
.create_schedule(
&NewSchedule::new("nightly", "generate_report", "0 0 * * *"),
None,
)
.await
});
tokio::time::sleep(std::time::Duration::from_millis(300)).await;
let started_waiting = std::time::Instant::now();
beta_tx.commit().await.unwrap();
assert!(
started_waiting.elapsed() < std::time::Duration::from_millis(300),
"beta's commit should be what releases alpha, not the other way round"
);
let outcome = racing.await.unwrap();
assert!(
matches!(
outcome,
Err(Error::AlreadyRegistered { ref kind, .. } | Error::RegisteredByAnother { ref kind, .. })
if kind == "Schedule"
),
"a peer that commits mid-call is a collision either way, got {outcome:?}"
);
let alpha = PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some("alpha"),
..Settings::default()
},
);
assert!(matches!(
alpha
.create_schedule(
&NewSchedule::new("nightly", "generate_report", "0 0 * * *"),
None
)
.await,
Err(Error::RegisteredByAnother { ref kind, ref holder, .. }) if kind == "Schedule" && holder == "beta"
));
}
#[tokio::test]
async fn a_schedule_name_held_by_another_application_is_refused() {
let db = test_database().await;
let pool = db.pool().await;
let alpha = PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some("alpha"),
..Settings::default()
},
);
let beta = PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some("beta"),
..Settings::default()
},
);
let schedule = NewSchedule::new("nightly", "generate_report", "0 0 * * *");
alpha.create_schedule(&schedule, None).await.unwrap();
assert_eq!(
alpha
.get_schedule("nightly", None)
.await
.unwrap()
.unwrap()
.application_name
.as_deref(),
Some("alpha")
);
for result in [
beta.create_schedule(&schedule, None).await,
beta.upsert_schedule(&schedule, None).await,
] {
assert!(matches!(
result,
Err(Error::RegisteredByAnother { ref kind, ref holder, .. }) if kind == "Schedule" && holder == "alpha"
));
}
let anonymous = PostgresSystemDatabase::from_pool(pool.clone(), &Settings::default());
anonymous
.create_schedule(&NewSchedule::new("hourly", "sweep", "0 * * * *"), None)
.await
.unwrap();
beta.upsert_schedule(&NewSchedule::new("hourly", "sweep", "0 * * * *"), None)
.await
.unwrap();
assert_eq!(
beta.get_schedule("hourly", None)
.await
.unwrap()
.unwrap()
.application_name
.as_deref(),
Some("beta")
);
}
#[tokio::test]
async fn applying_schedules_is_all_or_nothing() {
let db = test_database().await;
let pool = db.pool().await;
let alpha = PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some("alpha"),
..Settings::default()
},
);
let beta = PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some("beta"),
..Settings::default()
},
);
alpha
.create_schedule(
&NewSchedule::new("nightly", "generate_report", "0 0 * * *"),
None,
)
.await
.unwrap();
let result = beta
.apply_schedules(&[
NewSchedule::new("hourly", "sweep", "0 * * * *"),
NewSchedule::new("nightly", "generate_report", "0 0 * * *"),
])
.await;
assert!(matches!(
result,
Err(Error::RegisteredByAnother { ref holder, .. }) if holder == "alpha"
));
assert_eq!(
beta.get_schedule("hourly", None).await.unwrap(),
None,
"the first schedule rolled back with the second"
);
let declared = [
NewSchedule::new("hourly", "sweep", "0 * * * *"),
NewSchedule::new("weekly", "archive", "0 0 * * 0"),
];
beta.apply_schedules(&declared).await.unwrap();
beta.apply_schedules(&declared).await.unwrap();
assert_eq!(
schedule_names(&beta, &ScheduleFilter::default()).await,
["hourly", "weekly"]
);
assert!(
beta.apply_schedules(&[]).await.is_ok(),
"applying nothing is not an error"
);
let paused = NewSchedule {
status: ScheduleStatus::Paused,
last_fired_at: Some(Timestamp::from_epoch_ms(1_786_492_800_000)),
..NewSchedule::new("paused-decl", "run", "0 0 * * *")
};
let declared_paused = [paused];
beta.apply_schedules(&declared_paused).await.unwrap();
let seeded = beta
.get_schedule("paused-decl", None)
.await
.unwrap()
.unwrap();
assert_eq!(seeded.status, ScheduleStatus::Paused);
assert_eq!(
seeded.last_fired_at,
Some(Timestamp::from_epoch_ms(1_786_492_800_000))
);
beta.set_schedule_status("paused-decl", ScheduleStatus::Active, None)
.await
.unwrap();
beta.apply_schedules(&declared_paused).await.unwrap();
assert_eq!(
beta.get_schedule("paused-decl", None)
.await
.unwrap()
.unwrap()
.status,
ScheduleStatus::Active,
"re-applying does not pause a schedule an operator resumed"
);
}
#[tokio::test]
async fn schedules_are_listed_by_status_workflow_and_prefix() {
let (sys, _db) = sysdb().await;
for (name, workflow_name) in [
("report-nightly", "generate_report"),
("report-weekly", "generate_report"),
("sweep-hourly", "sweep"),
("100%-odd", "sweep"),
] {
sys.create_schedule(&NewSchedule::new(name, workflow_name, "0 0 * * *"), None)
.await
.unwrap();
}
sys.set_schedule_status("report-weekly", ScheduleStatus::Paused, None)
.await
.unwrap();
assert_eq!(
schedule_names(&sys, &ScheduleFilter::default()).await,
[
"100%-odd",
"report-nightly",
"report-weekly",
"sweep-hourly"
],
"an empty filter returns the table, ordered by name"
);
assert_eq!(
schedule_names(
&sys,
&ScheduleFilter {
statuses: vec![ScheduleStatus::Paused],
..Default::default()
}
)
.await,
["report-weekly"]
);
assert_eq!(
schedule_names(
&sys,
&ScheduleFilter {
workflow_names: vec!["sweep"],
..Default::default()
}
)
.await,
["100%-odd", "sweep-hourly"]
);
assert_eq!(
schedule_names(
&sys,
&ScheduleFilter {
schedule_name_prefixes: vec!["report-", "sweep-"],
..Default::default()
}
)
.await,
["report-nightly", "report-weekly", "sweep-hourly"],
"prefixes are an OR, not an AND"
);
assert_eq!(
schedule_names(
&sys,
&ScheduleFilter {
schedule_name_prefixes: vec!["100%"],
..Default::default()
}
)
.await,
["100%-odd"]
);
assert_eq!(
schedule_names(
&sys,
&ScheduleFilter {
statuses: vec![ScheduleStatus::Active],
workflow_names: vec!["generate_report"],
..Default::default()
}
)
.await,
["report-nightly"]
);
}
#[tokio::test]
async fn a_schedule_listing_defaults_to_its_own_application() {
let db = test_database().await;
let pool = db.pool().await;
let alpha = PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some("alpha"),
..Settings::default()
},
);
let beta = PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some("beta"),
..Settings::default()
},
);
let anonymous = PostgresSystemDatabase::from_pool(pool.clone(), &Settings::default());
alpha
.create_schedule(&NewSchedule::new("alpha-job", "run", "0 0 * * *"), None)
.await
.unwrap();
beta.create_schedule(&NewSchedule::new("beta-job", "run", "0 0 * * *"), None)
.await
.unwrap();
anonymous
.create_schedule(&NewSchedule::new("nobody-job", "run", "0 0 * * *"), None)
.await
.unwrap();
let scoped = |applications| ScheduleFilter {
applications,
..Default::default()
};
assert_eq!(
schedule_names(&alpha, &scoped(Applications::Unset)).await,
["alpha-job", "nobody-job"],
"its own plus the unclaimed"
);
assert_eq!(
schedule_names(&alpha, &scoped(Applications::Any)).await,
["alpha-job", "beta-job", "nobody-job"]
);
assert_eq!(
schedule_names(&alpha, &scoped(Applications::Named(vec!["beta"]))).await,
["beta-job", "nobody-job"]
);
assert_eq!(
schedule_names(&anonymous, &scoped(Applications::Unset)).await,
["alpha-job", "beta-job", "nobody-job"],
"an unnamed handle has no application to narrow to"
);
assert!(
alpha
.get_schedule("beta-job", None)
.await
.unwrap()
.is_some()
);
}
#[tokio::test]
async fn updating_a_schedule_touches_only_its_definition() {
let (sys, _db) = sysdb().await;
sys.create_schedule(
&NewSchedule {
cron_timezone: Some("Europe/London"),
queue_name: Some("reports"),
..NewSchedule::new("nightly", "generate_report", "0 0 * * *")
},
None,
)
.await
.unwrap();
sys.set_schedule_status("nightly", ScheduleStatus::Paused, None)
.await
.unwrap();
sys.update_schedule_last_fired_at("nightly", Timestamp::from_epoch_ms(1_786_492_800_000))
.await
.unwrap();
let before = sys.get_schedule("nightly", None).await.unwrap().unwrap();
sys.update_schedule(
"nightly",
&ScheduleUpdate {
schedule: Change::Set("*/5 * * * *"),
automatic_backfill: Change::Set(true),
cron_timezone: Change::Set(None),
queue_name: Change::Set(None),
..Default::default()
},
None,
)
.await
.unwrap();
let after = sys.get_schedule("nightly", None).await.unwrap().unwrap();
assert_eq!(after.schedule, "*/5 * * * *");
assert!(after.automatic_backfill);
assert_eq!(after.cron_timezone, None);
assert_eq!(after.queue_name, None);
assert_eq!(
after.context, before.context,
"unnamed fields are untouched"
);
assert_eq!(after.schedule_id, before.schedule_id);
assert_eq!(after.status, ScheduleStatus::Paused);
assert_eq!(after.last_fired_at, before.last_fired_at);
}
#[tokio::test]
async fn addressing_a_missing_schedule_is_refused() {
let (sys, _db) = sysdb().await;
let missing = |result: Result<(), Error>| matches!(result, Err(Error::NotRegistered { ref kind, ref name, .. }) if kind == "Schedule" && name == "ghost");
assert!(missing(
sys.update_schedule(
"ghost",
&ScheduleUpdate {
schedule: Change::Set("0 0 * * *"),
..Default::default()
},
None,
)
.await
));
assert!(missing(
sys.update_schedule("ghost", &ScheduleUpdate::default(), None)
.await
));
assert!(missing(
sys.set_schedule_status("ghost", ScheduleStatus::Paused, None)
.await
));
sys.update_schedule_last_fired_at("ghost", Timestamp::from_epoch_ms(1_786_492_800_000))
.await
.unwrap();
sys.delete_schedule("ghost", None).await.unwrap();
sys.create_schedule(
&NewSchedule::new("nightly", "generate_report", "0 0 * * *"),
None,
)
.await
.unwrap();
let before = sys.get_schedule("nightly", None).await.unwrap().unwrap();
sys.update_schedule("nightly", &ScheduleUpdate::default(), None)
.await
.unwrap();
assert_eq!(
sys.get_schedule("nightly", None).await.unwrap().unwrap(),
before
);
}
#[tokio::test]
async fn a_schedule_pauses_resumes_and_deletes() {
let (sys, _db) = sysdb().await;
sys.create_schedule(
&NewSchedule::new("nightly", "generate_report", "0 0 * * *"),
None,
)
.await
.unwrap();
sys.update_schedule_last_fired_at("nightly", Timestamp::from_epoch_ms(1_786_492_800_000))
.await
.unwrap();
sys.set_schedule_status("nightly", ScheduleStatus::Paused, None)
.await
.unwrap();
let paused = sys.get_schedule("nightly", None).await.unwrap().unwrap();
assert_eq!(paused.status, ScheduleStatus::Paused);
assert_eq!(
paused.last_fired_at,
Some(Timestamp::from_epoch_ms(1_786_492_800_000)),
"pausing does not forget where it had got to"
);
sys.set_schedule_status("nightly", ScheduleStatus::Active, None)
.await
.unwrap();
assert_eq!(
sys.get_schedule("nightly", None)
.await
.unwrap()
.unwrap()
.status,
ScheduleStatus::Active
);
sys.delete_schedule("nightly", None).await.unwrap();
assert_eq!(sys.get_schedule("nightly", None).await.unwrap(), None);
}
#[tokio::test]
async fn a_schedule_step_replays_rather_than_repeating() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-caller"), None, Submission::Fresh, None)
.await
.unwrap();
sys.create_schedule(
&NewSchedule::new("nightly", "generate_report", "0 0 * * *"),
Some(("wf-caller", 0)),
)
.await
.unwrap();
let recorded = sys
.check_step("wf-caller", 0, "DBOS.createSchedule")
.await
.unwrap()
.unwrap();
assert_eq!(
recorded.step_name, "DBOS.createSchedule",
"the write and its checkpoint committed together"
);
assert_eq!(recorded.output.as_deref(), Some("null"));
assert_eq!(recorded.error, None);
sys.create_schedule(
&NewSchedule::new("nightly", "generate_report", "0 0 * * *"),
Some(("wf-caller", 0)),
)
.await
.unwrap();
let first = sys
.get_schedule("nightly", Some(("wf-caller", 1)))
.await
.unwrap()
.unwrap();
sys.update_schedule(
"nightly",
&ScheduleUpdate {
schedule: Change::Set("*/5 * * * *"),
..Default::default()
},
None,
)
.await
.unwrap();
assert_eq!(
sys.get_schedule("nightly", Some(("wf-caller", 1)))
.await
.unwrap()
.unwrap(),
first,
"the replay is the recorded schedule, not the changed one"
);
assert_eq!(
sys.get_schedule("nightly", None)
.await
.unwrap()
.unwrap()
.schedule,
"*/5 * * * *",
"and the change did land"
);
assert_eq!(
sys.get_schedule("no-such", Some(("wf-caller", 4)))
.await
.unwrap(),
None
);
sys.create_schedule(&NewSchedule::new("no-such", "run", "0 0 * * *"), None)
.await
.unwrap();
assert_eq!(
sys.get_schedule("no-such", Some(("wf-caller", 4)))
.await
.unwrap(),
None,
"the replay is the recorded absence, not the schedule that now exists"
);
sys.delete_schedule("no-such", None).await.unwrap();
let listed = sys
.list_schedules(&ScheduleFilter::default(), Some(("wf-caller", 2)))
.await
.unwrap();
sys.delete_schedule("nightly", None).await.unwrap();
assert_eq!(
sys.list_schedules(&ScheduleFilter::default(), Some(("wf-caller", 2)))
.await
.unwrap(),
listed
);
assert!(
sys.list_schedules(&ScheduleFilter::default(), Some(("wf-caller", 3)))
.await
.unwrap()
.is_empty(),
"a fresh step sees the deletion"
);
}
#[tokio::test]
async fn pausing_and_resuming_are_distinct_steps() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-caller"), None, Submission::Fresh, None)
.await
.unwrap();
sys.create_schedule(
&NewSchedule::new("nightly", "generate_report", "0 0 * * *"),
None,
)
.await
.unwrap();
sys.set_schedule_status("nightly", ScheduleStatus::Paused, Some(("wf-caller", 0)))
.await
.unwrap();
sys.set_schedule_status("nightly", ScheduleStatus::Active, Some(("wf-caller", 1)))
.await
.unwrap();
assert_eq!(
sys.check_step("wf-caller", 0, "DBOS.pauseSchedule")
.await
.unwrap()
.map(|s| s.step_name),
Some("DBOS.pauseSchedule".to_owned())
);
assert_eq!(
sys.check_step("wf-caller", 1, "DBOS.resumeSchedule")
.await
.unwrap()
.map(|s| s.step_name),
Some("DBOS.resumeSchedule".to_owned())
);
}
#[tokio::test]
async fn every_schedule_write_replays_from_its_checkpoint() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-caller"), None, Submission::Fresh, None)
.await
.unwrap();
let schedule = |cron| NewSchedule::new("nightly", "generate_report", cron);
let cron = async |sys: &PostgresSystemDatabase| {
sys.get_schedule("nightly", None)
.await
.unwrap()
.map(|s| s.schedule)
};
sys.upsert_schedule(&schedule("0 0 * * *"), Some(("wf-caller", 0)))
.await
.unwrap();
sys.update_schedule(
"nightly",
&ScheduleUpdate {
schedule: Change::Set("*/5 * * * *"),
..Default::default()
},
None,
)
.await
.unwrap();
sys.upsert_schedule(&schedule("0 0 * * *"), Some(("wf-caller", 0)))
.await
.unwrap();
assert_eq!(
cron(&sys).await.as_deref(),
Some("*/5 * * * *"),
"the replayed upsert wrote nothing"
);
sys.update_schedule(
"nightly",
&ScheduleUpdate {
schedule: Change::Set("*/10 * * * *"),
..Default::default()
},
Some(("wf-caller", 1)),
)
.await
.unwrap();
assert_eq!(cron(&sys).await.as_deref(), Some("*/10 * * * *"));
sys.delete_schedule("nightly", None).await.unwrap();
sys.update_schedule(
"nightly",
&ScheduleUpdate {
schedule: Change::Set("*/10 * * * *"),
..Default::default()
},
Some(("wf-caller", 1)),
)
.await
.unwrap();
sys.create_schedule(&schedule("0 0 * * *"), None)
.await
.unwrap();
sys.set_schedule_status("nightly", ScheduleStatus::Paused, Some(("wf-caller", 2)))
.await
.unwrap();
sys.set_schedule_status("nightly", ScheduleStatus::Active, None)
.await
.unwrap();
sys.set_schedule_status("nightly", ScheduleStatus::Paused, Some(("wf-caller", 2)))
.await
.unwrap();
assert_eq!(
sys.get_schedule("nightly", None)
.await
.unwrap()
.unwrap()
.status,
ScheduleStatus::Active,
"the replayed pause left the resumed schedule alone"
);
sys.delete_schedule("nightly", Some(("wf-caller", 3)))
.await
.unwrap();
sys.create_schedule(&schedule("0 0 * * *"), None)
.await
.unwrap();
sys.delete_schedule("nightly", Some(("wf-caller", 3)))
.await
.unwrap();
assert!(
cron(&sys).await.is_some(),
"the replayed delete removed nothing"
);
let steps = sys
.list_workflow_steps("wf-caller", true, None, None, None)
.await
.unwrap();
let names: Vec<&str> = steps.iter().map(|s| s.step_name.as_str()).collect();
assert_eq!(
names,
[
"DBOS.upsertSchedule",
"DBOS.updateSchedule",
"DBOS.pauseSchedule",
"DBOS.deleteSchedule",
]
);
assert!(steps.iter().all(|s| s.output.as_deref() == Some("null")));
}
#[tokio::test]
async fn a_failed_schedule_step_records_nothing() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-caller"), None, Submission::Fresh, None)
.await
.unwrap();
sys.create_schedule(
&NewSchedule::new("nightly", "generate_report", "0 0 * * *"),
None,
)
.await
.unwrap();
assert!(matches!(
sys.create_schedule(
&NewSchedule::new("nightly", "generate_report", "0 0 * * *"),
Some(("wf-caller", 0))
)
.await,
Err(Error::AlreadyRegistered { .. })
));
assert_eq!(
sys.check_step("wf-caller", 0, "DBOS.createSchedule")
.await
.unwrap(),
None,
"a write that failed is not a step a replay can adopt"
);
sys.delete_schedule("nightly", None).await.unwrap();
sys.create_schedule(
&NewSchedule::new("nightly", "generate_report", "0 0 * * *"),
Some(("wf-caller", 0)),
)
.await
.unwrap();
assert!(sys.get_schedule("nightly", None).await.unwrap().is_some());
}
async fn enqueue_debounced(
sys: &PostgresSystemDatabase,
id: &str,
name: &str,
key: &str,
delay: std::time::Duration,
deadline: Option<Timestamp>,
) {
let wf = NewWorkflow {
name: Some(name),
queue_name: Some("orders"),
deduplication_id: Some(key),
is_debounced: true,
delay: Some(delay),
debounce_deadline: deadline,
input: Some("\"first\""),
application_version: Some("v1"),
..NewWorkflow::new(id)
};
sys.init_workflow(&wf, None, Submission::Fresh, None)
.await
.unwrap();
}
fn bounce<'a>(name: &'a str, key: &'a str, delay_until: Timestamp) -> DebounceRequest<'a> {
DebounceRequest {
workflow_name: name,
class_name: None,
config_name: None,
queue_name: "orders",
deduplication_id: key,
delay_until,
inputs: Some("\"later\""),
serialization: None,
application_name: None,
}
}
async fn enqueue_debounced_instance(
sys: &PostgresSystemDatabase,
id: &str,
class_name: &str,
config_name: &str,
key: &str,
) {
let wf = NewWorkflow {
name: Some("checkout"),
class_name: Some(class_name),
config_name: Some(config_name),
queue_name: Some("orders"),
deduplication_id: Some(key),
is_debounced: true,
delay: Some(std::time::Duration::from_secs(3600)),
input: Some("\"first\""),
application_version: Some("v1"),
..NewWorkflow::new(id)
};
sys.init_workflow(&wf, None, Submission::Fresh, None)
.await
.unwrap();
}
#[tokio::test]
async fn a_bounce_matches_the_class_and_configured_instance() {
let (sys, _db) = sysdb().await;
enqueue_debounced_instance(&sys, "wf-east", "Checkout", "east", "cart-1").await;
let west = DebounceRequest {
class_name: Some("Checkout"),
config_name: Some("west"),
..bounce("checkout", "cart-1", Timestamp::from_epoch_ms(9_000_000))
};
match sys.debounce_delayed_workflow(&west, None).await.unwrap() {
Debounce::Held(holder) => {
assert_eq!(holder.workflow_id, "wf-east");
assert_eq!(holder.class_name.as_deref(), Some("Checkout"));
assert_eq!(holder.config_name.as_deref(), Some("east"));
}
other => panic!("expected the east instance reported as the holder, got {other:?}"),
}
assert!(matches!(
sys.debounce_delayed_workflow(
&bounce("checkout", "cart-1", Timestamp::from_epoch_ms(9_000_000)),
None
)
.await
.unwrap(),
Debounce::Held(_)
));
let east = DebounceRequest {
class_name: Some("Checkout"),
config_name: Some("east"),
..bounce("checkout", "cart-1", Timestamp::from_epoch_ms(9_000_000))
};
assert_eq!(
sys.debounce_delayed_workflow(&east, None).await.unwrap(),
Debounce::Bounced {
workflow_id: "wf-east".to_owned()
}
);
}
#[tokio::test]
async fn a_bounce_refuses_an_empty_class_or_instance() {
let (sys, _db) = sysdb().await;
let request = DebounceRequest {
class_name: Some(""),
..bounce("checkout", "cart-1", Timestamp::from_epoch_ms(9_000_000))
};
assert!(matches!(
sys.debounce_delayed_workflow(&request, None).await,
Err(Error::InvalidInput { ref field, .. }) if field == "class_name"
));
}
#[tokio::test]
async fn a_bounce_extends_the_delay_and_replaces_the_inputs() {
let (sys, _db) = sysdb().await;
enqueue_debounced(
&sys,
"wf-debounced",
"checkout",
"cart-1",
std::time::Duration::from_secs(60),
None,
)
.await;
let pushed_to = Timestamp::from_epoch_ms(9_000_000_000);
assert_eq!(
sys.debounce_delayed_workflow(&bounce("checkout", "cart-1", pushed_to), None)
.await
.unwrap(),
Debounce::Bounced {
workflow_id: "wf-debounced".to_owned()
},
);
let read = sys.get_workflow("wf-debounced").await.unwrap().unwrap();
assert_eq!(read.delay_until, Some(pushed_to));
assert_eq!(read.input.as_deref(), Some("\"later\""));
assert_eq!(read.status, WorkflowStatus::Delayed);
}
#[tokio::test]
async fn a_bounce_cannot_push_past_the_debounce_deadline() {
let (sys, _db) = sysdb().await;
let deadline = Timestamp::from_epoch_ms(5_000_000_000);
enqueue_debounced(
&sys,
"wf-capped",
"checkout",
"cart-1",
std::time::Duration::from_secs(60),
Some(deadline),
)
.await;
sys.debounce_delayed_workflow(
&bounce(
"checkout",
"cart-1",
Timestamp::from_epoch_ms(9_000_000_000),
),
None,
)
.await
.unwrap();
assert_eq!(
sys.get_workflow("wf-capped")
.await
.unwrap()
.unwrap()
.delay_until,
Some(deadline),
"the deadline caps the push, so the workflow still runs",
);
}
#[tokio::test]
async fn a_key_collision_reports_the_holder() {
let (sys, _db) = sysdb().await;
enqueue_debounced(
&sys,
"wf-held",
"checkout",
"cart-1",
std::time::Duration::from_secs(60),
None,
)
.await;
let result = sys
.debounce_delayed_workflow(
&bounce("refund", "cart-1", Timestamp::from_epoch_ms(9_000_000)),
None,
)
.await
.unwrap();
match result {
Debounce::Held(holder) => {
assert_eq!(holder.workflow_id, "wf-held");
assert!(holder.is_debounced);
assert_eq!(holder.workflow_name.as_deref(), Some("checkout"));
}
other => panic!("expected the holder to be reported, got {other:?}"),
}
assert_eq!(
sys.get_workflow("wf-held")
.await
.unwrap()
.unwrap()
.input
.as_deref(),
Some("\"first\""),
);
}
#[tokio::test]
async fn an_unheld_key_is_reported_as_unheld() {
let (sys, _db) = sysdb().await;
sys.upsert_queue(&NewQueue::new("orders"), OnExistingQueue::Update)
.await
.unwrap();
assert_eq!(
sys.debounce_delayed_workflow(
&bounce(
"checkout",
"never-used",
Timestamp::from_epoch_ms(9_000_000)
),
None
)
.await
.unwrap(),
Debounce::Unheld,
);
}
#[tokio::test]
async fn a_bounce_stays_within_an_application() {
let db = test_database().await;
let pool = db.pool().await;
let named = |name: &'static str| {
PostgresSystemDatabase::from_pool(
pool.clone(),
&Settings {
application_name: Some(name),
..Settings::default()
},
)
};
let (alpha, beta) = (named("alpha"), named("beta"));
let anonymous = PostgresSystemDatabase::from_pool(pool.clone(), &Settings::default());
enqueue_debounced(
&beta,
"wf-beta",
"checkout",
"beta-key",
std::time::Duration::from_secs(60),
None,
)
.await;
enqueue_debounced(
&anonymous,
"wf-nobody",
"checkout",
"free-key",
std::time::Duration::from_secs(60),
None,
)
.await;
let result = alpha
.debounce_delayed_workflow(
&bounce("checkout", "beta-key", Timestamp::from_epoch_ms(9_000_000)),
None,
)
.await
.unwrap();
match result {
Debounce::Held(holder) => {
assert_eq!(holder.application_name.as_deref(), Some("beta"));
}
other => panic!("expected beta reported as the holder, got {other:?}"),
}
assert_eq!(
alpha
.debounce_delayed_workflow(
&bounce("checkout", "free-key", Timestamp::from_epoch_ms(9_000_000)),
None
)
.await
.unwrap(),
Debounce::Bounced {
workflow_id: "wf-nobody".to_owned()
},
);
assert_eq!(
alpha
.get_workflow("wf-nobody")
.await
.unwrap()
.unwrap()
.application_name
.as_deref(),
Some("alpha"),
"bouncing an unclaimed workflow claims it, as its dequeue would",
);
}
#[tokio::test]
async fn a_bounce_inside_a_workflow_is_a_step_and_replays() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-caller"), None, Submission::Fresh, None)
.await
.unwrap();
enqueue_debounced(
&sys,
"wf-debounced",
"checkout",
"cart-1",
std::time::Duration::from_secs(60),
None,
)
.await;
let pushed_to = Timestamp::from_epoch_ms(9_000_000_000);
let first = sys
.debounce_delayed_workflow(
&bounce("checkout", "cart-1", pushed_to),
Some(("wf-caller", 0)),
)
.await
.unwrap();
assert_eq!(
first,
Debounce::Bounced {
workflow_id: "wf-debounced".to_owned()
},
);
let steps = sys
.list_workflow_steps("wf-caller", true, None, None, None)
.await
.unwrap();
assert_eq!(
steps
.iter()
.map(|s| s.step_name.as_str())
.collect::<Vec<_>>(),
["DBOS.debounceDelayedWorkflow"],
);
sys.set_workflow_delay(
"wf-debounced",
WorkflowDelay::Until(Timestamp::from_epoch_ms(1)),
None,
)
.await
.unwrap();
let replayed = sys
.debounce_delayed_workflow(
&bounce(
"checkout",
"cart-1",
Timestamp::from_epoch_ms(7_000_000_000),
),
Some(("wf-caller", 0)),
)
.await
.unwrap();
assert_eq!(replayed, first, "a replay reports the first run's decision");
assert_eq!(
sys.get_workflow("wf-debounced")
.await
.unwrap()
.unwrap()
.delay_until,
Some(Timestamp::from_epoch_ms(1)),
"and does not bounce again",
);
}
#[tokio::test]
async fn a_replayed_bounce_reports_the_original_holder() {
let (sys, _db) = sysdb().await;
sys.init_workflow(&workflow("wf-caller"), None, Submission::Fresh, None)
.await
.unwrap();
enqueue_debounced(
&sys,
"wf-held",
"checkout",
"cart-1",
std::time::Duration::from_secs(60),
None,
)
.await;
let first = sys
.debounce_delayed_workflow(
&bounce("refund", "cart-1", Timestamp::from_epoch_ms(9_000_000)),
Some(("wf-caller", 0)),
)
.await
.unwrap();
assert!(matches!(first, Debounce::Held(_)), "got {first:?}");
sys.record_workflow_outcome("wf-held", Outcome::Output(Some("\"done\"")))
.await
.unwrap();
let replayed = sys
.debounce_delayed_workflow(
&bounce("refund", "cart-1", Timestamp::from_epoch_ms(9_000_000)),
Some(("wf-caller", 0)),
)
.await
.unwrap();
assert_eq!(
replayed, first,
"the recorded holder is reported, not one re-read after it changed",
);
}