From 76a1ddae381028db07bf22c5616db0ae011029eb Mon Sep 17 00:00:00 2001 From: Pino de Candia Date: Tue, 28 Jul 2026 01:46:36 +0000 Subject: [PATCH 1/6] Retry retryable ack failures instead of abandoning the work item The worker dispatcher abandoned an activity work item on any ack_work_item error, which redelivers the item and re-runs the handler. The dispatcher poll loop already branches on ProviderError::is_retryable() and backs off before retrying fetch_work_item; the ack path ignored the classification entirely. Add ack_work_item_with_retry: up to 5 attempts with exponential backoff (100ms..1s, ~1.5s total) for retryable errors, then fall back to the existing abandon path. Permanent errors still abandon immediately, so poison handling and the existing failure tests are unaffected. This is reachable from any provider that reports transient failures. It is easiest to observe with SqliteProvider::new_in_memory, which uses sqlite::memory:?cache=shared. Shared-cache mode uses table-level locks and returns SQLITE_LOCKED on a read-to-write upgrade collision without invoking the busy handler, so PRAGMA busy_timeout does not cover it. Also relax test_multi_worker_heterogeneous_config, which asserted `a + b == 3`. Activities are documented as at-least-once, so any redelivery makes equality wrong; assert `a + b >= 3` instead. Fixes #41 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/runtime/dispatchers/worker.rs | 97 ++++++++++++++++++++++--------- tests/common/fault_injection.rs | 27 +++++++++ tests/session_e2e_tests.rs | 4 ++ tests/worker_reliability_test.rs | 84 ++++++++++++++++++++++++++ 4 files changed, 183 insertions(+), 29 deletions(-) diff --git a/src/runtime/dispatchers/worker.rs b/src/runtime/dispatchers/worker.rs index d7f0141..6a04ae5 100644 --- a/src/runtime/dispatchers/worker.rs +++ b/src/runtime/dispatchers/worker.rs @@ -703,7 +703,7 @@ async fn run_activity_with_cancellation( let duration_seconds = start_time.elapsed().as_secs_f64(); rt.record_activity_execution(&ctx.activity_name, "cancelled", duration_seconds, 0, ctx.tag.as_deref()); - let result = rt.history_store.ack_work_item(&ctx.lock_token, None).await; + let result = ack_work_item_with_retry(rt, ctx, None).await; if let Err(e) = &result { tracing::warn!( target: "duroxide::runtime", @@ -722,6 +722,47 @@ async fn run_activity_with_cancellation( // Activity Completion Handlers // ============================================================================ +/// Maximum number of attempts when acking an activity completion. +const MAX_ACK_ATTEMPTS: u32 = 5; + +/// Ack a work item, retrying transient provider failures with exponential backoff. +/// +/// Activities have at-least-once semantics, so an ack failure is not a correctness +/// violation — but it is expensive: the caller abandons the work item, the activity is +/// redelivered, and the handler runs again. Retryable provider errors (database lock +/// contention, transient connection failures) usually clear within milliseconds, so +/// retrying here avoids most redundant re-executions. This mirrors the retry handling +/// the dispatcher loop already applies to `fetch_work_item`. +async fn ack_work_item_with_retry( + rt: &Arc, + ctx: &ActivityWorkContext, + item: Option, +) -> Result<(), crate::providers::ProviderError> { + let mut attempt = 1; + loop { + match rt.history_store.ack_work_item(&ctx.lock_token, item.clone()).await { + Ok(()) => return Ok(()), + Err(e) if e.is_retryable() && attempt < MAX_ACK_ATTEMPTS => { + let backoff_ms = (50 * 2_u64.pow(attempt)).min(1000); + warn!( + instance = %ctx.instance, + execution_id = ctx.execution_id, + activity_id = ctx.activity_id, + worker_id = %ctx.worker_id, + attempt = attempt, + max_attempts = MAX_ACK_ATTEMPTS, + error = %e, + "worker: ack failed (retryable), backing off {}ms", + backoff_ms + ); + tokio::time::sleep(Duration::from_millis(backoff_ms)).await; + attempt += 1; + } + Err(e) => return Err(e), + } + } +} + /// Handle successful activity completion. async fn handle_activity_success( rt: &Arc, @@ -748,18 +789,17 @@ async fn handle_activity_success( rt.record_activity_execution(&ctx.activity_name, "success", duration_seconds, 0, ctx.tag.as_deref()); - let ack_result = rt - .history_store - .ack_work_item( - &ctx.lock_token, - Some(WorkItem::ActivityCompleted { - instance: ctx.instance.clone(), - execution_id: ctx.execution_id, - id: ctx.activity_id, - result, - }), - ) - .await; + let ack_result = ack_work_item_with_retry( + rt, + ctx, + Some(WorkItem::ActivityCompleted { + instance: ctx.instance.clone(), + execution_id: ctx.execution_id, + id: ctx.activity_id, + result, + }), + ) + .await; (ack_result, ActivityOutcome::Success) } @@ -790,22 +830,21 @@ async fn handle_activity_error( rt.record_activity_execution(&ctx.activity_name, "app_error", duration_seconds, 0, ctx.tag.as_deref()); - let ack_result = rt - .history_store - .ack_work_item( - &ctx.lock_token, - Some(WorkItem::ActivityFailed { - instance: ctx.instance.clone(), - execution_id: ctx.execution_id, - id: ctx.activity_id, - details: crate::ErrorDetails::Application { - kind: crate::AppErrorKind::ActivityFailed, - message: error, - retryable: false, - }, - }), - ) - .await; + let ack_result = ack_work_item_with_retry( + rt, + ctx, + Some(WorkItem::ActivityFailed { + instance: ctx.instance.clone(), + execution_id: ctx.execution_id, + id: ctx.activity_id, + details: crate::ErrorDetails::Application { + kind: crate::AppErrorKind::ActivityFailed, + message: error, + retryable: false, + }, + }), + ) + .await; (ack_result, ActivityOutcome::AppError) } diff --git a/tests/common/fault_injection.rs b/tests/common/fault_injection.rs index 89d6463..5b8980e 100644 --- a/tests/common/fault_injection.rs +++ b/tests/common/fault_injection.rs @@ -646,6 +646,8 @@ impl Provider for FilterBypassProvider { pub struct FailingProvider { inner: Arc, fail_next_ack_work_item: AtomicBool, + /// Number of remaining `ack_work_item` calls to fail with a *retryable* error + retryable_ack_work_item_failures: AtomicU32, fail_next_ack_orchestration_item: AtomicBool, fail_next_fetch_orchestration_item: AtomicBool, fail_next_fetch_work_item: AtomicBool, @@ -660,6 +662,7 @@ impl FailingProvider { Self { inner, fail_next_ack_work_item: AtomicBool::new(false), + retryable_ack_work_item_failures: AtomicU32::new(0), fail_next_ack_orchestration_item: AtomicBool::new(false), fail_next_fetch_orchestration_item: AtomicBool::new(false), fail_next_fetch_work_item: AtomicBool::new(false), @@ -672,6 +675,19 @@ impl FailingProvider { self.fail_next_ack_work_item.store(true, Ordering::SeqCst); } + /// Fail the next `count` `ack_work_item` calls with a *retryable* error. + /// + /// Unlike `fail_next_ack_work_item`, the underlying ack is not performed, so the + /// work item stays locked and the caller is expected to retry in place. + pub fn fail_next_ack_work_item_retryable(&self, count: u32) { + self.retryable_ack_work_item_failures.store(count, Ordering::SeqCst); + } + + /// Number of injected retryable ack failures that have not been consumed yet. + pub fn remaining_retryable_ack_failures(&self) -> u32 { + self.retryable_ack_work_item_failures.load(Ordering::SeqCst) + } + pub fn fail_next_ack_orchestration_item(&self) { self.fail_next_ack_orchestration_item.store(true, Ordering::SeqCst); } @@ -798,6 +814,17 @@ impl Provider for FailingProvider { } async fn ack_work_item(&self, token: &str, completion: Option) -> Result<(), ProviderError> { + if self + .retryable_ack_work_item_failures + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| n.checked_sub(1).filter(|_| n > 0)) + .is_ok() + { + return Err(ProviderError::retryable( + "ack_work_item", + "simulated transient lock contention", + )); + } + if self.fail_next_ack_work_item.swap(false, Ordering::SeqCst) { // If ack_then_fail is set, do the actual ack first if self.ack_then_fail.load(Ordering::SeqCst) { diff --git a/tests/session_e2e_tests.rs b/tests/session_e2e_tests.rs index 6933296..e979de9 100644 --- a/tests/session_e2e_tests.rs +++ b/tests/session_e2e_tests.rs @@ -1583,6 +1583,10 @@ async fn test_multi_worker_heterogeneous_config() { // A should hold at most 1 session concurrently. Worker B picks up the rest // while A is busy. A may process more than 1 *total* (serially), but B // should handle at least 1. + // + // Activities have at-least-once semantics: an ack failure or lock expiry + // redelivers the work item, so the counters can exceed the number of + // scheduled activities. Assert a lower bound, not equality. let a = worker_a_sessions.load(Ordering::SeqCst); let b = worker_b_sessions.load(Ordering::SeqCst); assert!(a + b >= 3, "Total should be at least 3, got A={a} B={b}"); diff --git a/tests/worker_reliability_test.rs b/tests/worker_reliability_test.rs index b04dc90..ee3f497 100644 --- a/tests/worker_reliability_test.rs +++ b/tests/worker_reliability_test.rs @@ -411,3 +411,87 @@ async fn worker_abandon_on_ack_failure_enables_retry() { rt.shutdown(None).await; } + +/// Test that transient (retryable) ack failures are retried in place instead of +/// abandoning the work item. +/// +/// Abandoning on a retryable ack failure redelivers the work item and re-executes the +/// activity handler. Activities are only guaranteed at-least-once, but re-running a +/// handler because the database was briefly contended is wasteful and surprising, so the +/// worker retries the ack before falling back to abandon. +#[tokio::test] +async fn worker_retries_retryable_ack_failure_without_reexecuting_activity() { + use common::fault_injection::FailingProvider; + use duroxide::providers::sqlite::SqliteProvider; + use duroxide::runtime::RuntimeOptions; + use std::sync::atomic::{AtomicU32, Ordering}; + use std::time::Duration; + + let sqlite = Arc::new(SqliteProvider::new_in_memory().await.unwrap()); + let failing_provider = Arc::new(FailingProvider::new(sqlite)); + + let execution_count = Arc::new(AtomicU32::new(0)); + let exec_count_clone = execution_count.clone(); + + let activities = runtime::registry::ActivityRegistry::builder() + .register("CountingActivity", move |_ctx: ActivityContext, _input: String| { + let count = exec_count_clone.clone(); + async move { + count.fetch_add(1, Ordering::SeqCst); + Ok("done".to_string()) + } + }) + .build(); + + let orchestrations = runtime::registry::OrchestrationRegistry::builder() + .register("RetryAckOrch", |ctx: OrchestrationContext, _input: String| async move { + ctx.schedule_activity("CountingActivity", "{}").await + }) + .build(); + + let options = RuntimeOptions { + // Long lock timeout so any retry must come from the in-place ack retry, + // not from lock expiry redelivery. + worker_lock_timeout: Duration::from_secs(30), + ..Default::default() + }; + + let provider: Arc = failing_provider.clone(); + let rt = runtime::Runtime::start_with_options(provider.clone(), activities, orchestrations, options).await; + let client = Client::new(provider.clone()); + + // Fail the first 3 acks with a retryable error. The worker allows 5 attempts, + // so the 4th attempt should succeed. + failing_provider.fail_next_ack_work_item_retryable(3); + + client + .start_orchestration("retry-ack-test", "RetryAckOrch", "") + .await + .unwrap(); + + let status = client + .wait_for_orchestration("retry-ack-test", Duration::from_secs(10)) + .await + .unwrap(); + + match status { + runtime::OrchestrationStatus::Completed { output, .. } => { + assert_eq!(output, "done"); + } + other => panic!("Unexpected status: {other:?}"), + } + + assert_eq!( + failing_provider.remaining_retryable_ack_failures(), + 0, + "All injected retryable ack failures should have been exercised" + ); + + assert_eq!( + execution_count.load(Ordering::SeqCst), + 1, + "Activity must not be re-executed when the ack failure was retryable" + ); + + rt.shutdown(None).await; +} From 2ffe1da4a8c4f2fcc29ef0961c5226d90673d845 Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Tue, 13 Oct 2026 16:00:43 +0000 Subject: [PATCH 2/6] Harden implementation --- docs/provider-implementation-guide.md | 18 +- tests/common/fault_injection.rs | 56 +++- tests/worker_reliability_test.rs | 389 +++++++++++++++++--------- 3 files changed, 321 insertions(+), 142 deletions(-) diff --git a/docs/provider-implementation-guide.md b/docs/provider-implementation-guide.md index ed71f4a..a86e32c 100644 --- a/docs/provider-implementation-guide.md +++ b/docs/provider-implementation-guide.md @@ -904,7 +904,7 @@ Provider methods return `Result<..., ProviderError>`. The runtime uses error cla // Retryable: Transient failures, runtime will retry with backoff ProviderError::retryable("operation", "Database busy") -// Permanent: Unrecoverable, runtime will fail the orchestration +// Permanent: Retrying the operation will not recover it ProviderError::permanent("operation", "Duplicate event detected") ``` @@ -914,12 +914,26 @@ ProviderError::permanent("operation", "Duplicate event detected") - Network failure - Temporary resource exhaustion -**Permanent errors** (runtime will fail): +**Permanent errors** (do not retry): - Duplicate events - Missing lock token (already processed or cancelled) - Data corruption - Invalid input +For acknowledgements of activity success, application errors, or cancellation, +the runtime retries retryable errors in place up to five total attempts before +abandoning the work item. Permanent errors go directly to abandonment. The retry +waits are 100, 200, 400, and 800 milliseconds: 1.5 seconds of cumulative backoff, +excluding time spent in provider calls. Normal activity completion retains lock +renewal while acknowledgement is pending. Activities still have at-least-once +execution semantics; exhausted retries or lost locks can lead to re-execution. + +A retryable error does not necessarily mean the transaction rolled back: a +connection failure can leave the commit outcome unknown. If an acknowledgement +already committed, a repeated call must reject the missing lock token with a +permanent error without enqueueing another completion. Preserve the atomic +delete-and-enqueue operation and lock validation described below. + ### Atomicity Requirements Several operations must be atomic (all-or-nothing): diff --git a/tests/common/fault_injection.rs b/tests/common/fault_injection.rs index 5b8980e..ec36b7f 100644 --- a/tests/common/fault_injection.rs +++ b/tests/common/fault_injection.rs @@ -641,6 +641,17 @@ impl Provider for FilterBypassProvider { } } +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum WorkItemCall { + Ack { + token: String, + completion: Option, + }, + Abandon { + token: String, + }, +} + /// A provider wrapper that can inject transient failures for testing /// error handling and retry logic. pub struct FailingProvider { @@ -648,6 +659,7 @@ pub struct FailingProvider { fail_next_ack_work_item: AtomicBool, /// Number of remaining `ack_work_item` calls to fail with a *retryable* error retryable_ack_work_item_failures: AtomicU32, + work_item_calls: std::sync::Mutex>, fail_next_ack_orchestration_item: AtomicBool, fail_next_fetch_orchestration_item: AtomicBool, fail_next_fetch_work_item: AtomicBool, @@ -663,6 +675,7 @@ impl FailingProvider { inner, fail_next_ack_work_item: AtomicBool::new(false), retryable_ack_work_item_failures: AtomicU32::new(0), + work_item_calls: std::sync::Mutex::new(Vec::new()), fail_next_ack_orchestration_item: AtomicBool::new(false), fail_next_fetch_orchestration_item: AtomicBool::new(false), fail_next_fetch_work_item: AtomicBool::new(false), @@ -677,8 +690,7 @@ impl FailingProvider { /// Fail the next `count` `ack_work_item` calls with a *retryable* error. /// - /// Unlike `fail_next_ack_work_item`, the underlying ack is not performed, so the - /// work item stays locked and the caller is expected to retry in place. + /// The underlying ack is not performed unless `set_ack_then_fail(true)` is set. pub fn fail_next_ack_work_item_retryable(&self, count: u32) { self.retryable_ack_work_item_failures.store(count, Ordering::SeqCst); } @@ -688,6 +700,11 @@ impl FailingProvider { self.retryable_ack_work_item_failures.load(Ordering::SeqCst) } + /// Return work-item acknowledgement and abandonment calls in invocation order. + pub fn work_item_calls(&self) -> Vec { + self.work_item_calls.lock().unwrap().clone() + } + pub fn fail_next_ack_orchestration_item(&self) { self.fail_next_ack_orchestration_item.store(true, Ordering::SeqCst); } @@ -814,26 +831,36 @@ impl Provider for FailingProvider { } async fn ack_work_item(&self, token: &str, completion: Option) -> Result<(), ProviderError> { - if self + self.work_item_calls.lock().unwrap().push(WorkItemCall::Ack { + token: token.to_string(), + completion: completion.clone(), + }); + + let failure = if self .retryable_ack_work_item_failures - .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| n.checked_sub(1).filter(|_| n > 0)) + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| { + n.checked_sub(1).filter(|_| n > 0) + }) .is_ok() { - return Err(ProviderError::retryable( + Some(ProviderError::retryable( "ack_work_item", "simulated transient lock contention", - )); - } + )) + } else if self.fail_next_ack_work_item.swap(false, Ordering::SeqCst) { + Some(ProviderError::permanent( + "ack_work_item", + "simulated infrastructure failure", + )) + } else { + None + }; - if self.fail_next_ack_work_item.swap(false, Ordering::SeqCst) { - // If ack_then_fail is set, do the actual ack first + if let Some(error) = failure { if self.ack_then_fail.load(Ordering::SeqCst) { self.inner.ack_work_item(token, completion).await?; } - Err(ProviderError::permanent( - "ack_work_item", - "simulated infrastructure failure", - )) + Err(error) } else { self.inner.ack_work_item(token, completion).await } @@ -849,6 +876,9 @@ impl Provider for FailingProvider { delay: Option, ignore_attempt: bool, ) -> Result<(), ProviderError> { + self.work_item_calls.lock().unwrap().push(WorkItemCall::Abandon { + token: token.to_string(), + }); self.inner.abandon_work_item(token, delay, ignore_attempt).await } diff --git a/tests/worker_reliability_test.rs b/tests/worker_reliability_test.rs index ee3f497..e6f5bab 100644 --- a/tests/worker_reliability_test.rs +++ b/tests/worker_reliability_test.rs @@ -5,11 +5,16 @@ #![allow(clippy::clone_on_ref_ptr)] #![allow(clippy::expect_used)] +use common::fault_injection::{FailingProvider, WorkItemCall}; use duroxide::EventKind; +use duroxide::providers::sqlite::SqliteProvider; +use duroxide::providers::{Provider, WorkItem}; use duroxide::runtime; use duroxide::runtime::registry::{ActivityRegistry, OrchestrationRegistry}; use duroxide::*; use std::sync::Arc; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::time::Duration; mod common; @@ -334,164 +339,294 @@ async fn multiple_activities_reliability_after_crash() { } } -/// Test that work items are abandoned and retried when ack_work_item fails. -/// -/// This verifies the worker dispatcher calls abandon_work_item when ack fails, -/// making the work item available for retry without waiting for lock expiration. -#[tokio::test] -async fn worker_abandon_on_ack_failure_enables_retry() { - use common::fault_injection::FailingProvider; - use duroxide::providers::sqlite::SqliteProvider; - use duroxide::runtime::RuntimeOptions; - use std::sync::atomic::{AtomicU32, Ordering}; - use std::time::Duration; +struct AckTest { + provider: Arc, + client: Client, + rt: Arc, + _db: tempfile::NamedTempFile, +} - let sqlite = Arc::new(SqliteProvider::new_in_memory().await.unwrap()); - let failing_provider = Arc::new(FailingProvider::new(sqlite)); +impl AckTest { + async fn new(activities: ActivityRegistry) -> Self { + // WAL keeps exact call counts independent of shared-cache deadlocks. + let db = tempfile::NamedTempFile::new().unwrap(); + let sqlite = Arc::new( + SqliteProvider::new(&format!("sqlite:{}", db.path().display()), None) + .await + .unwrap(), + ); + let provider = Arc::new(FailingProvider::new(sqlite)); + let orchestrations = OrchestrationRegistry::builder() + .register("AckOrch", |ctx: OrchestrationContext, _input: String| async move { + ctx.schedule_activity("CountingActivity", "{}").await + }) + .build(); + let options = runtime::RuntimeOptions { + worker_concurrency: 1, + orchestration_concurrency: 1, + worker_lock_timeout: Duration::from_secs(2), + worker_lock_renewal_buffer: Duration::from_millis(500), + activity_cancellation_grace_period: Duration::ZERO, + ..Default::default() + }; + let rt = runtime::Runtime::start_with_options(provider.clone(), activities, orchestrations, options).await; + let client = Client::new(provider.clone()); + + Self { + provider, + client, + rt, + _db: db, + } + } - // Track how many times the activity executes - let execution_count = Arc::new(AtomicU32::new(0)); - let exec_count_clone = execution_count.clone(); + async fn start(&self) { + self.client + .start_orchestration("ack-test", "AckOrch", "") + .await + .unwrap(); + } + + async fn finish(&self) -> runtime::OrchestrationStatus { + let status = self + .client + .wait_for_orchestration("ack-test", Duration::from_secs(10)) + .await + .unwrap(); - let activities = runtime::registry::ActivityRegistry::builder() + // A commit can become visible before the worker finishes retrying its acknowledgement. + Arc::clone(&self.rt).shutdown(Some(10_000)).await; + status + } +} + +fn counting_activity(result: Result) -> (ActivityRegistry, Arc) { + let execution_count = Arc::new(AtomicU32::new(0)); + let counter = execution_count.clone(); + let activities = ActivityRegistry::builder() .register("CountingActivity", move |_ctx: ActivityContext, _input: String| { - let count = exec_count_clone.clone(); + let counter = counter.clone(); + let result = result.clone(); async move { - count.fetch_add(1, Ordering::SeqCst); - Ok("done".to_string()) + counter.fetch_add(1, Ordering::SeqCst); + result } }) .build(); + (activities, execution_count) +} - let orchestrations = runtime::registry::OrchestrationRegistry::builder() - .register("AckFailOrch", |ctx: OrchestrationContext, _input: String| async move { - ctx.schedule_activity("CountingActivity", "{}").await - }) - .build(); - - // Use short lock timeout - if abandon doesn't work, we'd have to wait for this to expire - let options = RuntimeOptions { - worker_lock_timeout: Duration::from_secs(30), // Long timeout - retry should happen via abandon, not expiry - ..Default::default() +#[tokio::test] +async fn worker_abandon_on_ack_failure_enables_retry() { + let (activities, execution_count) = counting_activity(Ok("done".to_string())); + let test = AckTest::new(activities).await; + test.provider.fail_next_ack_work_item(); + test.start().await; + + assert!(matches!( + test.finish().await, + runtime::OrchestrationStatus::Completed { output, .. } if output == "done" + )); + assert_eq!(execution_count.load(Ordering::SeqCst), 2); + + let calls = test.provider.work_item_calls(); + let [ + WorkItemCall::Ack { token, completion }, + WorkItemCall::Abandon { token: abandoned }, + WorkItemCall::Ack { + token: redelivered, + completion: retried_completion, + }, + ] = calls.as_slice() + else { + panic!("Permanent failure must abandon without retrying the ack: {calls:?}"); }; - - let provider: Arc = failing_provider.clone(); - let rt = runtime::Runtime::start_with_options(provider.clone(), activities, orchestrations, options).await; - let client = Client::new(provider.clone()); - - // Make the first ack_work_item fail (but the ack actually succeeds internally) - // This simulates a transient failure where the ack completed but we got an error - failing_provider.set_ack_then_fail(true); - failing_provider.fail_next_ack_work_item(); - - client - .start_orchestration("ack-fail-test", "AckFailOrch", "") - .await - .unwrap(); - - // Wait for completion - let status = client - .wait_for_orchestration("ack-fail-test", Duration::from_secs(5)) - .await - .unwrap(); - - match status { - runtime::OrchestrationStatus::Completed { output, .. } => { - assert_eq!(output, "done"); - } - other => panic!("Unexpected status: {other:?}"), - } - - // Activity should have executed at least once - // (may execute twice if retry happened - that's fine for this test) - assert!( - execution_count.load(Ordering::SeqCst) >= 1, - "Activity should execute at least once" - ); - - rt.shutdown(None).await; + assert_eq!(token, abandoned); + assert_ne!(token, redelivered); + assert_eq!(completion, retried_completion); } -/// Test that transient (retryable) ack failures are retried in place instead of -/// abandoning the work item. -/// -/// Abandoning on a retryable ack failure redelivers the work item and re-executes the -/// activity handler. Activities are only guaranteed at-least-once, but re-running a -/// handler because the database was briefly contended is wasteful and surprising, so the -/// worker retries the ack before falling back to abandon. #[tokio::test] async fn worker_retries_retryable_ack_failure_without_reexecuting_activity() { - use common::fault_injection::FailingProvider; - use duroxide::providers::sqlite::SqliteProvider; - use duroxide::runtime::RuntimeOptions; - use std::sync::atomic::{AtomicU32, Ordering}; - use std::time::Duration; + let (activities, execution_count) = counting_activity(Ok("done".to_string())); + let test = AckTest::new(activities).await; + test.provider.fail_next_ack_work_item_retryable(4); + test.start().await; + + assert!(matches!( + test.finish().await, + runtime::OrchestrationStatus::Completed { output, .. } if output == "done" + )); + assert_eq!(test.provider.remaining_retryable_ack_failures(), 0); + assert_eq!(execution_count.load(Ordering::SeqCst), 1); + + let calls = test.provider.work_item_calls(); + assert_eq!(calls.len(), 5, "The fifth and final attempt must succeed"); + assert!(matches!( + &calls[0], + WorkItemCall::Ack { + completion: Some(WorkItem::ActivityCompleted { result, .. }), + .. + } if result == "done" + )); + assert_eq!(calls, vec![calls[0].clone(); 5]); +} - let sqlite = Arc::new(SqliteProvider::new_in_memory().await.unwrap()); - let failing_provider = Arc::new(FailingProvider::new(sqlite)); +#[tokio::test] +async fn worker_abandons_after_retryable_ack_attempts_are_exhausted() { + let (activities, execution_count) = counting_activity(Ok("done".to_string())); + let test = AckTest::new(activities).await; + test.provider.fail_next_ack_work_item_retryable(5); + test.start().await; + + assert!(matches!( + test.finish().await, + runtime::OrchestrationStatus::Completed { output, .. } if output == "done" + )); + assert_eq!(test.provider.remaining_retryable_ack_failures(), 0); + assert_eq!(execution_count.load(Ordering::SeqCst), 2); + + let calls = test.provider.work_item_calls(); + assert_eq!(calls.len(), 7, "Five attempts, abandon, then a successful redelivery"); + let WorkItemCall::Ack { token, completion } = &calls[0] else { + panic!("Expected an acknowledgement: {calls:?}"); + }; + assert!(calls[..5].iter().all(|call| call == &calls[0]), "{calls:?}"); + assert_eq!(calls[5], WorkItemCall::Abandon { token: token.clone() }); + let WorkItemCall::Ack { + token: redelivered, + completion: retried_completion, + } = &calls[6] + else { + panic!("Expected acknowledgement after redelivery: {calls:?}"); + }; + assert_ne!(token, redelivered); + assert_eq!(completion, retried_completion); +} - let execution_count = Arc::new(AtomicU32::new(0)); - let exec_count_clone = execution_count.clone(); +#[tokio::test] +async fn worker_retries_application_error_ack_without_reexecuting_activity() { + let (activities, execution_count) = counting_activity(Err("activity error".to_string())); + let test = AckTest::new(activities).await; + test.provider.fail_next_ack_work_item_retryable(3); + test.start().await; + + assert!(matches!( + test.finish().await, + runtime::OrchestrationStatus::Failed { .. } + )); + assert_eq!(test.provider.remaining_retryable_ack_failures(), 0); + assert_eq!(execution_count.load(Ordering::SeqCst), 1); + + let calls = test.provider.work_item_calls(); + assert_eq!(calls.len(), 4); + assert!(matches!( + &calls[0], + WorkItemCall::Ack { + completion: Some(WorkItem::ActivityFailed { + details: ErrorDetails::Application { + kind: AppErrorKind::ActivityFailed, + message, + retryable: false, + }, + .. + }), + .. + } if message == "activity error" + )); + assert_eq!(calls, vec![calls[0].clone(); 4]); +} - let activities = runtime::registry::ActivityRegistry::builder() +#[tokio::test] +async fn worker_retries_cancelled_activity_ack_without_completion() { + let entered = Arc::new(tokio::sync::Notify::new()); + let activity_entered = entered.clone(); + let execution_count = Arc::new(AtomicU32::new(0)); + let counter = execution_count.clone(); + let activities = ActivityRegistry::builder() .register("CountingActivity", move |_ctx: ActivityContext, _input: String| { - let count = exec_count_clone.clone(); + let entered = activity_entered.clone(); + let counter = counter.clone(); async move { - count.fetch_add(1, Ordering::SeqCst); - Ok("done".to_string()) + counter.fetch_add(1, Ordering::SeqCst); + entered.notify_one(); + std::future::pending::>().await } }) .build(); - - let orchestrations = runtime::registry::OrchestrationRegistry::builder() - .register("RetryAckOrch", |ctx: OrchestrationContext, _input: String| async move { - ctx.schedule_activity("CountingActivity", "{}").await - }) - .build(); - - let options = RuntimeOptions { - // Long lock timeout so any retry must come from the in-place ack retry, - // not from lock expiry redelivery. - worker_lock_timeout: Duration::from_secs(30), - ..Default::default() - }; - - let provider: Arc = failing_provider.clone(); - let rt = runtime::Runtime::start_with_options(provider.clone(), activities, orchestrations, options).await; - let client = Client::new(provider.clone()); - - // Fail the first 3 acks with a retryable error. The worker allows 5 attempts, - // so the 4th attempt should succeed. - failing_provider.fail_next_ack_work_item_retryable(3); - - client - .start_orchestration("retry-ack-test", "RetryAckOrch", "") + let test = AckTest::new(activities).await; + test.start().await; + tokio::time::timeout(Duration::from_secs(5), entered.notified()) .await .unwrap(); - - let status = client - .wait_for_orchestration("retry-ack-test", Duration::from_secs(10)) + test.provider.fail_next_ack_work_item_retryable(3); + test.client + .cancel_instance("ack-test", "test cancellation") .await .unwrap(); - match status { - runtime::OrchestrationStatus::Completed { output, .. } => { - assert_eq!(output, "done"); + assert!(matches!( + test.finish().await, + runtime::OrchestrationStatus::Failed { + details: ErrorDetails::Application { + kind: AppErrorKind::Cancelled { .. }, + .. + }, + .. } - other => panic!("Unexpected status: {other:?}"), - } + )); + assert_eq!(test.provider.remaining_retryable_ack_failures(), 0); + assert_eq!(execution_count.load(Ordering::SeqCst), 1); + + let calls = test.provider.work_item_calls(); + assert_eq!(calls.len(), 5); + let WorkItemCall::Ack { + token, + completion: None, + } = &calls[0] + else { + panic!("Cancellation must acknowledge without a completion: {calls:?}"); + }; + assert!(calls[..4].iter().all(|call| call == &calls[0]), "{calls:?}"); + // Cancellation deleted the queue entry, so the final ack fails permanently. + assert_eq!(calls[4], WorkItemCall::Abandon { token: token.clone() }); +} - assert_eq!( - failing_provider.remaining_retryable_ack_failures(), - 0, - "All injected retryable ack failures should have been exercised" - ); +#[tokio::test] +async fn worker_retryable_ack_error_after_commit_does_not_duplicate_completion() { + let (activities, execution_count) = counting_activity(Ok("done".to_string())); + let test = AckTest::new(activities).await; + test.provider.set_ack_then_fail(true); + test.provider.fail_next_ack_work_item_retryable(1); + test.start().await; + + assert!(matches!( + test.finish().await, + runtime::OrchestrationStatus::Completed { output, .. } if output == "done" + )); + assert_eq!(test.provider.remaining_retryable_ack_failures(), 0); + assert_eq!(execution_count.load(Ordering::SeqCst), 1); + + let calls = test.provider.work_item_calls(); + assert_eq!(calls.len(), 3); + let WorkItemCall::Ack { + token, + completion: Some(WorkItem::ActivityCompleted { result, .. }), + } = &calls[0] + else { + panic!("Expected an activity completion: {calls:?}"); + }; + assert_eq!(result, "done"); + assert_eq!(calls[0], calls[1]); + assert_eq!(calls[2], WorkItemCall::Abandon { token: token.clone() }); + let history = test.provider.read("ack-test").await.unwrap(); assert_eq!( - execution_count.load(Ordering::SeqCst), + history + .iter() + .filter(|event| matches!(event.kind, EventKind::ActivityCompleted { .. })) + .count(), 1, - "Activity must not be re-executed when the ack failure was retryable" + "An ambiguous commit must not produce duplicate activity completions" ); - - rt.shutdown(None).await; } From 45dfd5f68ae77dce24e7c7e338e35599d19b11f2 Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Fri, 30 Oct 2026 12:08:42 +0000 Subject: [PATCH 3/6] Implement fixes from review --- docs/provider-implementation-guide.md | 25 +++++- docs/provider-testing-guide.md | 4 +- src/provider_validation/atomicity.rs | 67 +++++++++++++- src/provider_validations.rs | 5 +- src/providers/mod.rs | 3 + src/runtime/dispatchers/worker.rs | 73 +++++++++------ src/runtime/mod.rs | 5 ++ tests/common/fault_injection.rs | 12 +++ tests/sqlite_provider_validations.rs | 8 +- tests/worker_reliability_test.rs | 124 ++++++++++++++++++++++---- 10 files changed, 274 insertions(+), 52 deletions(-) diff --git a/docs/provider-implementation-guide.md b/docs/provider-implementation-guide.md index a86e32c..d6e4d1b 100644 --- a/docs/provider-implementation-guide.md +++ b/docs/provider-implementation-guide.md @@ -920,20 +920,39 @@ ProviderError::permanent("operation", "Duplicate event detected") - Data corruption - Invalid input -For acknowledgements of activity success, application errors, or cancellation, +For acknowledgements of activity success or application errors, the runtime retries retryable errors in place up to five total attempts before -abandoning the work item. Permanent errors go directly to abandonment. The retry -waits are 100, 200, 400, and 800 milliseconds: 1.5 seconds of cumulative backoff, +abandoning the work item. A permanent error on the first attempt goes directly +to abandonment. The retry waits are 100, 200, 400, and 800 milliseconds: 1.5 seconds of cumulative backoff, excluding time spent in provider calls. Normal activity completion retains lock renewal while acknowledgement is pending. Activities still have at-least-once execution semantics; exhausted retries or lost locks can lead to re-execution. +Cancellation acknowledgements (`ack_work_item(token, None)`) are attempted only +once. If that attempt fails, the worker falls back to abandonment rather than +retrying the acknowledgement. + A retryable error does not necessarily mean the transaction rolled back: a connection failure can leave the commit outcome unknown. If an acknowledgement already committed, a repeated call must reject the missing lock token with a permanent error without enqueueing another completion. Preserve the atomic delete-and-enqueue operation and lock validation described below. +A permanent error after a retryable acknowledgement error is reported as an +**uncertain outcome**, not as a confirmed activity infrastructure failure. The +runtime cannot distinguish an earlier commit from lock loss or another permanent +error. It does not attempt another abandonment or count an activity infrastructure +error in this case; any uncommitted work remains eligible for redelivery after +its lock expires. + +Acknowledgement retries do not extend the shutdown grace period. The default +`shutdown(None)` grace is 1 second, shorter than the full 1.5-second backoff budget, +so shutdown may abort a retrying acknowledgement, including during a backoff wait. +Shutdown remains a bounded drain: retries continue within the grace period and +are aborted when it expires. Choose a longer grace period to allow more time for +retries and provider calls. Uncommitted work is recovered through lock expiry; +an acknowledgement that committed before cancellation remains committed. + ### Atomicity Requirements Several operations must be atomic (all-or-nothing): diff --git a/docs/provider-testing-guide.md b/docs/provider-testing-guide.md index 0e5ff2a..0659c78 100644 --- a/docs/provider-testing-guide.md +++ b/docs/provider-testing-guide.md @@ -422,6 +422,7 @@ use duroxide::providers::Provider; use duroxide::provider_validations::{ ProviderFactory, // Atomicity tests + test_ack_work_item_duplicate_rejected, test_atomicity_failure_rollback, test_multi_operation_atomic_ack, // Locking tests @@ -511,11 +512,12 @@ async fn test_my_provider_worker_queue_fifo_ordering() { The validation test suite includes the following categories: -1. **Atomicity Tests (4 tests)** +1. **Atomicity Tests (5 tests)** - `test_atomicity_failure_rollback` - All-or-nothing commit semantics, rollback on failure - `test_multi_operation_atomic_ack` - Complex ack succeeds atomically - `test_lock_released_only_on_successful_ack` - Lock only released on success - `test_concurrent_ack_prevention` - Only one ack succeeds with same token + - `test_ack_work_item_duplicate_rejected` - Repeating `ack_work_item(Some(...))` with the same token fails non-retryably and leaves exactly one completion queued, without relying on replay deduplication 2. **Error Handling Tests (5 tests)** - `test_invalid_lock_token_on_ack` - Invalid lock token rejection diff --git a/src/provider_validation/atomicity.rs b/src/provider_validation/atomicity.rs index ab8ec7b..cf7fc8c 100644 --- a/src/provider_validation/atomicity.rs +++ b/src/provider_validation/atomicity.rs @@ -1,7 +1,7 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. -use crate::provider_validation::{Event, EventKind, ExecutionMetadata, start_item}; +use crate::provider_validation::{Event, EventKind, ExecutionMetadata, create_instance, start_item}; use crate::provider_validations::ProviderFactory; use crate::providers::{TagFilter, WorkItem}; use std::sync::Arc; @@ -463,3 +463,68 @@ pub async fn test_concurrent_ack_prevention(factory: &F) { assert_eq!(failures.len(), 1, "Exactly one ack should fail"); tracing::info!("✓ Test passed: concurrent ack prevention verified"); } + +/// Test 2.5: Duplicate Worker Ack +/// Goal: Reject a repeated completion ack without enqueueing a duplicate completion. +pub async fn test_ack_work_item_duplicate_rejected(factory: &F) { + let provider = factory.create_provider().await; + let instance = "instance-duplicate-ack"; + create_instance(provider.as_ref(), instance).await.unwrap(); + + provider + .enqueue_for_worker(WorkItem::ActivityExecute { + instance: instance.to_string(), + execution_id: 1, + id: 2, + name: "TestActivity".to_string(), + input: "{}".to_string(), + session_id: None, + tag: None, + }) + .await + .unwrap(); + let (_, token, _) = provider + .fetch_work_item(Duration::from_secs(30), Duration::ZERO, None, &TagFilter::default()) + .await + .unwrap() + .unwrap(); + let completion = WorkItem::ActivityCompleted { + instance: instance.to_string(), + execution_id: 1, + id: 2, + result: "done".to_string(), + }; + provider.ack_work_item(&token, Some(completion.clone())).await.unwrap(); + let error = provider + .ack_work_item(&token, Some(completion.clone())) + .await + .expect_err("A second ack with the same token must fail"); + assert!(!error.is_retryable(), "A repeated ack must fail permanently: {error}"); + + let (item, token, _) = provider + .fetch_orchestration_item(Duration::from_secs(30), Duration::ZERO, None) + .await + .unwrap() + .unwrap(); + assert_eq!(item.messages, vec![completion], "Exactly one completion must be queued"); + provider + .ack_orchestration_item(&token, 1, vec![], vec![], vec![], ExecutionMetadata::default(), vec![]) + .await + .unwrap(); + assert!( + provider + .fetch_orchestration_item(Duration::from_secs(30), Duration::ZERO, None) + .await + .unwrap() + .is_none(), + "No duplicate completion may remain outside the first batch" + ); + assert!( + provider + .fetch_work_item(Duration::from_secs(30), Duration::ZERO, None, &TagFilter::default()) + .await + .unwrap() + .is_none(), + "The acknowledged worker item must be removed" + ); +} diff --git a/src/provider_validations.rs b/src/provider_validations.rs index 8f594e4..e9bed32 100644 --- a/src/provider_validations.rs +++ b/src/provider_validations.rs @@ -62,6 +62,7 @@ pub use crate::provider_validation::race_replay; /// - `test_multi_operation_atomic_ack` - Verify complex ack succeeds atomically /// - `test_lock_released_only_on_successful_ack` - Verify lock only released on success /// - `test_concurrent_ack_prevention` - Verify only one ack succeeds with same token +/// - `test_ack_work_item_duplicate_rejected` - Verify a repeated worker ack cannot enqueue another completion /// /// **Error Handling Tests:** /// - `test_invalid_lock_token_on_ack` - Verify invalid lock tokens are rejected @@ -141,8 +142,8 @@ pub use crate::provider_validation::instance_creation::{ #[cfg(feature = "provider-test")] pub use crate::provider_validation::atomicity::{ - test_atomicity_failure_rollback, test_concurrent_ack_prevention, test_lock_released_only_on_successful_ack, - test_multi_operation_atomic_ack, + test_ack_work_item_duplicate_rejected, test_atomicity_failure_rollback, test_concurrent_ack_prevention, + test_lock_released_only_on_successful_ack, test_multi_operation_atomic_ack, }; #[cfg(feature = "provider-test")] diff --git a/src/providers/mod.rs b/src/providers/mod.rs index 9c4c2a1..eab9446 100644 --- a/src/providers/mod.rs +++ b/src/providers/mod.rs @@ -1811,6 +1811,9 @@ pub trait Provider: Any + Send + Sync { /// - The activity was cancelled (the backing entry was removed/invalidated by orchestration) /// - The lock was stolen by another worker (lock expired and work-item was re-fetched/completed) /// - The lock expired (worker took too long without renewing) + /// - An earlier acknowledgement committed, but its response was lost + /// + /// Repeating an acknowledgement with the same token MUST NOT enqueue another completion. /// /// The worker dispatcher uses this error to detect cancellation races and log appropriately. /// diff --git a/src/runtime/dispatchers/worker.rs b/src/runtime/dispatchers/worker.rs index 6a04ae5..e7794c6 100644 --- a/src/runtime/dispatchers/worker.rs +++ b/src/runtime/dispatchers/worker.rs @@ -31,7 +31,7 @@ //! 3. The worker dispatcher waits up to the grace period for the activity to complete //! 4. If the activity doesn't complete, it's aborted and the work item is dropped -use crate::providers::{WorkItem, instrumented::instrument_provider_operation}; +use crate::providers::{ProviderError, WorkItem, instrumented::instrument_provider_operation}; use std::sync::Arc; use std::time::Duration; use tokio::task::JoinHandle; @@ -132,6 +132,11 @@ enum ActivityInvocation { Deferred(Duration), } +enum ActivityAckError { + Failed(ProviderError), + Uncertain(ProviderError), +} + /// Context for processing a single activity work item. /// /// Groups together all the data needed to execute an activity, @@ -611,7 +616,7 @@ async fn run_activity_with_cancellation( cancellation_token: CancellationToken, _manager_handle: AbortOnDropHandle<()>, start_time: std::time::Instant, -) -> (Result<(), crate::providers::ProviderError>, ActivityOutcome) { +) -> (Result<(), ActivityAckError>, ActivityOutcome) { let input = ctx.input.clone(); let activity_id = ctx.activity_id; // Abort-on-drop so the invocation cannot outlive this task if the worker is @@ -663,7 +668,7 @@ async fn run_activity_with_cancellation( let result = rt.history_store .abandon_work_item(&ctx.lock_token, Some(delay), true) .await; - (result, ActivityOutcome::Deferred(delay)) + (result.map_err(ActivityAckError::Failed), ActivityOutcome::Deferred(delay)) } Ok(Err(error)) => handle_activity_error(rt, ctx, error, start_time).await, Err(join_error) => handle_activity_error(rt, ctx, join_error.to_string(), start_time).await, @@ -699,11 +704,11 @@ async fn run_activity_with_cancellation( let _ = activity_handle.await; } - // Record metrics and ack (drop result since orchestration is terminal) let duration_seconds = start_time.elapsed().as_secs_f64(); rt.record_activity_execution(&ctx.activity_name, "cancelled", duration_seconds, 0, ctx.tag.as_deref()); - let result = ack_work_item_with_retry(rt, ctx, None).await; + // Renewal errors can falsely signal cancellation; retrying ack(None) could delete live work. + let result = rt.history_store.ack_work_item(&ctx.lock_token, None).await; if let Err(e) = &result { tracing::warn!( target: "duroxide::runtime", @@ -713,7 +718,7 @@ async fn run_activity_with_cancellation( "Failed to ack cancelled activity work item" ); } - (result, ActivityOutcome::Cancelled) + (result.map_err(ActivityAckError::Failed), ActivityOutcome::Cancelled) } } } @@ -725,25 +730,29 @@ async fn run_activity_with_cancellation( /// Maximum number of attempts when acking an activity completion. const MAX_ACK_ATTEMPTS: u32 = 5; -/// Ack a work item, retrying transient provider failures with exponential backoff. +/// Ack an activity completion, retrying transient provider failures with exponential backoff. /// /// Activities have at-least-once semantics, so an ack failure is not a correctness /// violation — but it is expensive: the caller abandons the work item, the activity is /// redelivered, and the handler runs again. Retryable provider errors (database lock /// contention, transient connection failures) usually clear within milliseconds, so -/// retrying here avoids most redundant re-executions. This mirrors the retry handling -/// the dispatcher loop already applies to `fetch_work_item`. -async fn ack_work_item_with_retry( +/// retrying here avoids most redundant re-executions. Unlike fetch retries, these +/// retries are bounded and retain the activity's lock renewal. +async fn ack_activity_completion_with_retry( rt: &Arc, ctx: &ActivityWorkContext, - item: Option, -) -> Result<(), crate::providers::ProviderError> { + completion: WorkItem, +) -> Result<(), ActivityAckError> { let mut attempt = 1; loop { - match rt.history_store.ack_work_item(&ctx.lock_token, item.clone()).await { + match rt + .history_store + .ack_work_item(&ctx.lock_token, Some(completion.clone())) + .await + { Ok(()) => return Ok(()), Err(e) if e.is_retryable() && attempt < MAX_ACK_ATTEMPTS => { - let backoff_ms = (50 * 2_u64.pow(attempt)).min(1000); + let backoff_ms = 50 * 2_u64.pow(attempt); warn!( instance = %ctx.instance, execution_id = ctx.execution_id, @@ -758,7 +767,10 @@ async fn ack_work_item_with_retry( tokio::time::sleep(Duration::from_millis(backoff_ms)).await; attempt += 1; } - Err(e) => return Err(e), + // A missing token after a transient error can mean the earlier ack committed. + // ProviderError cannot distinguish that from lock loss or other permanent errors. + Err(e) if !e.is_retryable() && attempt > 1 => return Err(ActivityAckError::Uncertain(e)), + Err(e) => return Err(ActivityAckError::Failed(e)), } } } @@ -769,7 +781,7 @@ async fn handle_activity_success( ctx: &ActivityWorkContext, result: String, start_time: std::time::Instant, -) -> (Result<(), crate::providers::ProviderError>, ActivityOutcome) { +) -> (Result<(), ActivityAckError>, ActivityOutcome) { let duration_ms = start_time.elapsed().as_millis() as u64; let duration_seconds = duration_ms as f64 / 1000.0; @@ -789,15 +801,15 @@ async fn handle_activity_success( rt.record_activity_execution(&ctx.activity_name, "success", duration_seconds, 0, ctx.tag.as_deref()); - let ack_result = ack_work_item_with_retry( + let ack_result = ack_activity_completion_with_retry( rt, ctx, - Some(WorkItem::ActivityCompleted { + WorkItem::ActivityCompleted { instance: ctx.instance.clone(), execution_id: ctx.execution_id, id: ctx.activity_id, result, - }), + }, ) .await; @@ -810,7 +822,7 @@ async fn handle_activity_error( ctx: &ActivityWorkContext, error: String, start_time: std::time::Instant, -) -> (Result<(), crate::providers::ProviderError>, ActivityOutcome) { +) -> (Result<(), ActivityAckError>, ActivityOutcome) { let duration_ms = start_time.elapsed().as_millis() as u64; let duration_seconds = duration_ms as f64 / 1000.0; @@ -830,10 +842,10 @@ async fn handle_activity_error( rt.record_activity_execution(&ctx.activity_name, "app_error", duration_seconds, 0, ctx.tag.as_deref()); - let ack_result = ack_work_item_with_retry( + let ack_result = ack_activity_completion_with_retry( rt, ctx, - Some(WorkItem::ActivityFailed { + WorkItem::ActivityFailed { instance: ctx.instance.clone(), execution_id: ctx.execution_id, id: ctx.activity_id, @@ -842,7 +854,7 @@ async fn handle_activity_error( message: error, retryable: false, }, - }), + }, ) .await; @@ -888,7 +900,7 @@ async fn abandon_unregistered_activity(rt: &Arc, ctx: &ActivityWorkCont async fn handle_activity_outcome( rt: &Arc, ctx: &ActivityWorkContext, - ack_result: Result<(), crate::providers::ProviderError>, + ack_result: Result<(), ActivityAckError>, outcome: ActivityOutcome, ) { match ack_result { @@ -900,7 +912,18 @@ async fn handle_activity_outcome( } ActivityOutcome::Cancelled => {} }, - Err(e) => { + Err(ActivityAckError::Uncertain(e)) => { + warn!( + instance = %ctx.instance, + execution_id = ctx.execution_id, + activity_id = ctx.activity_id, + worker_id = %ctx.worker_id, + ack_outcome = "uncertain", + error = %e, + "worker: acknowledgement outcome uncertain after retry; any uncommitted work will be redelivered after lock expiry" + ); + } + Err(ActivityAckError::Failed(e)) => { warn!( instance = %ctx.instance, execution_id = ctx.execution_id, diff --git a/src/runtime/mod.rs b/src/runtime/mod.rs index 221f362..dc3d767 100644 --- a/src/runtime/mod.rs +++ b/src/runtime/mod.rs @@ -1110,6 +1110,11 @@ impl Runtime { /// - `Some(0)`: Abort immediately, then wait for cancellation cleanup /// - `Some(ms)`: Allow up to the given milliseconds before aborting stragglers /// + /// Activity acknowledgement retries can spend up to 1.5 seconds in backoff, + /// plus time in provider calls. The default 1-second grace can interrupt them; + /// retries do not extend the grace period. Unacknowledged work becomes available + /// for redelivery after its lock expires. + /// /// This is not a hard return deadline. Tokio cancellation requires tasks to yield; /// blocking or non-yielding user code can delay shutdown beyond the grace period. /// Tasks or threads spawned by application code are the application's responsibility. diff --git a/tests/common/fault_injection.rs b/tests/common/fault_injection.rs index ec36b7f..5f8d508 100644 --- a/tests/common/fault_injection.rs +++ b/tests/common/fault_injection.rs @@ -663,6 +663,7 @@ pub struct FailingProvider { fail_next_ack_orchestration_item: AtomicBool, fail_next_fetch_orchestration_item: AtomicBool, fail_next_fetch_work_item: AtomicBool, + fail_next_renew_work_item_lock: AtomicBool, /// If true, allow failure event commits to succeed even when failing acks allow_failure_commits: AtomicBool, /// If true, do the actual ack before returning error (for testing post-ack failures) @@ -679,6 +680,7 @@ impl FailingProvider { fail_next_ack_orchestration_item: AtomicBool::new(false), fail_next_fetch_orchestration_item: AtomicBool::new(false), fail_next_fetch_work_item: AtomicBool::new(false), + fail_next_renew_work_item_lock: AtomicBool::new(false), allow_failure_commits: AtomicBool::new(false), ack_then_fail: AtomicBool::new(false), } @@ -717,6 +719,10 @@ impl FailingProvider { self.fail_next_fetch_work_item.store(true, Ordering::SeqCst); } + pub fn fail_next_renew_work_item_lock(&self) { + self.fail_next_renew_work_item_lock.store(true, Ordering::SeqCst); + } + /// When set, failure event commits (OrchestrationFailed) will succeed /// even when ack_orchestration_item is set to fail. pub fn set_allow_failure_commits(&self, allow: bool) { @@ -867,6 +873,12 @@ impl Provider for FailingProvider { } async fn renew_work_item_lock(&self, token: &str, extension: Duration) -> Result<(), ProviderError> { + if self.fail_next_renew_work_item_lock.swap(false, Ordering::SeqCst) { + return Err(ProviderError::retryable( + "renew_work_item_lock", + "simulated transient renewal failure", + )); + } self.inner.renew_work_item_lock(token, extension).await } diff --git a/tests/sqlite_provider_validations.rs b/tests/sqlite_provider_validations.rs index 7c5c779..665cd15 100644 --- a/tests/sqlite_provider_validations.rs +++ b/tests/sqlite_provider_validations.rs @@ -91,10 +91,11 @@ mod tests { test_abandon_work_item_releases_lock, test_abandon_work_item_with_delay, test_ack_only_affects_locked_messages, + // Atomicity tests + test_ack_work_item_duplicate_rejected, // Cancellation tests test_ack_work_item_fails_when_entry_deleted, test_ack_work_item_none_deletes_without_enqueue, - // Atomicity tests test_atomicity_failure_rollback, test_batch_cancellation_deletes_multiple_activities, test_cancelled_activities_deleted_from_worker_queue, @@ -352,6 +353,11 @@ mod tests { test_concurrent_ack_prevention(&SqliteTestFactory).await; } + #[tokio::test] + async fn test_sqlite_ack_work_item_duplicate_rejected() { + test_ack_work_item_duplicate_rejected(&SqliteTestFactory).await; + } + // Error handling tests #[tokio::test] async fn test_sqlite_invalid_lock_token_on_ack() { diff --git a/tests/worker_reliability_test.rs b/tests/worker_reliability_test.rs index e6f5bab..b7c2fba 100644 --- a/tests/worker_reliability_test.rs +++ b/tests/worker_reliability_test.rs @@ -7,8 +7,8 @@ use common::fault_injection::{FailingProvider, WorkItemCall}; use duroxide::EventKind; +use duroxide::providers::WorkItem; use duroxide::providers::sqlite::SqliteProvider; -use duroxide::providers::{Provider, WorkItem}; use duroxide::runtime; use duroxide::runtime::registry::{ActivityRegistry, OrchestrationRegistry}; use duroxide::*; @@ -538,7 +538,7 @@ async fn worker_retries_application_error_ack_without_reexecuting_activity() { } #[tokio::test] -async fn worker_retries_cancelled_activity_ack_without_completion() { +async fn worker_does_not_retry_cancelled_activity_ack() { let entered = Arc::new(tokio::sync::Notify::new()); let activity_entered = entered.clone(); let execution_count = Arc::new(AtomicU32::new(0)); @@ -559,7 +559,7 @@ async fn worker_retries_cancelled_activity_ack_without_completion() { tokio::time::timeout(Duration::from_secs(5), entered.notified()) .await .unwrap(); - test.provider.fail_next_ack_work_item_retryable(3); + test.provider.fail_next_ack_work_item_retryable(1); test.client .cancel_instance("ack-test", "test cancellation") .await @@ -579,7 +579,7 @@ async fn worker_retries_cancelled_activity_ack_without_completion() { assert_eq!(execution_count.load(Ordering::SeqCst), 1); let calls = test.provider.work_item_calls(); - assert_eq!(calls.len(), 5); + assert_eq!(calls.len(), 2); let WorkItemCall::Ack { token, completion: None, @@ -587,13 +587,59 @@ async fn worker_retries_cancelled_activity_ack_without_completion() { else { panic!("Cancellation must acknowledge without a completion: {calls:?}"); }; - assert!(calls[..4].iter().all(|call| call == &calls[0]), "{calls:?}"); - // Cancellation deleted the queue entry, so the final ack fails permanently. - assert_eq!(calls[4], WorkItemCall::Abandon { token: token.clone() }); + assert_eq!(calls[1], WorkItemCall::Abandon { token: token.clone() }); } -#[tokio::test] -async fn worker_retryable_ack_error_after_commit_does_not_duplicate_completion() { +#[tokio::test(flavor = "current_thread")] +async fn worker_redelivers_after_transient_renewal_and_cancellation_ack_failures() { + let execution_count = Arc::new(AtomicU32::new(0)); + let counter = execution_count.clone(); + let activities = ActivityRegistry::builder() + .register("CountingActivity", move |_ctx: ActivityContext, _input: String| { + let counter = counter.clone(); + async move { + if counter.fetch_add(1, Ordering::SeqCst) == 0 { + std::future::pending::<()>().await; + } + Ok("done".to_string()) + } + }) + .build(); + let test = AckTest::new(activities).await; + test.provider.fail_next_renew_work_item_lock(); + test.provider.fail_next_ack_work_item_retryable(1); + test.start().await; + + assert!(matches!( + test.finish().await, + runtime::OrchestrationStatus::Completed { output, .. } if output == "done" + )); + assert_eq!(test.provider.remaining_retryable_ack_failures(), 0); + assert_eq!(execution_count.load(Ordering::SeqCst), 2); + + let calls = test.provider.work_item_calls(); + let [ + WorkItemCall::Ack { + token, + completion: None, + }, + WorkItemCall::Abandon { token: abandoned }, + WorkItemCall::Ack { + token: redelivered, + completion: Some(WorkItem::ActivityCompleted { result, .. }), + }, + ] = calls.as_slice() + else { + panic!("A failed cancellation ack must allow the live activity to be redelivered: {calls:?}"); + }; + assert_eq!(token, abandoned); + assert_ne!(token, redelivered); + assert_eq!(result, "done"); +} + +#[tokio::test(flavor = "current_thread")] +async fn worker_retryable_ack_error_after_commit_is_reported_as_uncertain() { + let (captured, _guard) = common::tracing_capture::install_tracing_capture(); let (activities, execution_count) = counting_activity(Ok("done".to_string())); let test = AckTest::new(activities).await; test.provider.set_ack_then_fail(true); @@ -608,25 +654,65 @@ async fn worker_retryable_ack_error_after_commit_does_not_duplicate_completion() assert_eq!(execution_count.load(Ordering::SeqCst), 1); let calls = test.provider.work_item_calls(); - assert_eq!(calls.len(), 3); + assert_eq!(calls.len(), 2); let WorkItemCall::Ack { - token, completion: Some(WorkItem::ActivityCompleted { result, .. }), + .. } = &calls[0] else { panic!("Expected an activity completion: {calls:?}"); }; assert_eq!(result, "done"); assert_eq!(calls[0], calls[1]); - assert_eq!(calls[2], WorkItemCall::Abandon { token: token.clone() }); + assert_eq!(test.rt.metrics_snapshot().unwrap().activity_infra_errors, 0); - let history = test.provider.read("ack-test").await.unwrap(); - assert_eq!( - history + let events = captured.lock().unwrap(); + assert!( + events .iter() - .filter(|event| matches!(event.kind, EventKind::ActivityCompleted { .. })) - .count(), - 1, - "An ambiguous commit must not produce duplicate activity completions" + .any(|event| event.field("ack_outcome").as_deref() == Some("uncertain")), + "The runtime must report an uncertain acknowledgement outcome" + ); + assert!( + !events + .iter() + .any(|event| event.message.contains("activity acknowledgment or deferral failed")), + "An uncertain acknowledgement must not be reported as a confirmed failure" + ); +} + +#[tokio::test] +async fn worker_uncertain_ack_before_commit_recovers_after_lock_expiry() { + let (activities, execution_count) = counting_activity(Ok("done".to_string())); + let test = AckTest::new(activities).await; + test.provider.fail_next_ack_work_item_retryable(1); + test.provider.fail_next_ack_work_item(); + test.start().await; + + assert!(matches!( + test.finish().await, + runtime::OrchestrationStatus::Completed { output, .. } if output == "done" + )); + assert_eq!(execution_count.load(Ordering::SeqCst), 2); + assert_eq!(test.rt.metrics_snapshot().unwrap().activity_infra_errors, 0); + + let calls = test.provider.work_item_calls(); + assert_eq!( + calls.len(), + 3, + "Two uncertain ack attempts, then redelivery without abandon" ); + let WorkItemCall::Ack { token, completion } = &calls[0] else { + panic!("Expected an acknowledgement: {calls:?}"); + }; + assert_eq!(calls[0], calls[1]); + let WorkItemCall::Ack { + token: redelivered, + completion: retried_completion, + } = &calls[2] + else { + panic!("Expected acknowledgement after lock expiry: {calls:?}"); + }; + assert_ne!(token, redelivered); + assert_eq!(completion, retried_completion); } From 03bf08041fb82cdaf4987238ac9b72f6b11056a5 Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Fri, 30 Oct 2026 15:41:55 +0000 Subject: [PATCH 4/6] Addressed remaining review nits - Softened the uncertain-ack log so it no longer promises redelivery. - Added duroxide_activity_ack_uncertain_total, counted once per uncertain outcome without inflating failure counters. - Updated metric documentation and added an Unreleased entry in CHANGELOG.md. --- CHANGELOG.md | 9 ++++++++ docs/metrics-specification.md | 17 ++++++++++++++- docs/observability-guide.md | 6 ++++++ docs/provider-implementation-guide.md | 8 ++++--- src/runtime/dispatchers/worker.rs | 3 ++- src/runtime/mod.rs | 7 ++++++ src/runtime/observability.rs | 10 +++++++++ tests/observability_tests.rs | 31 +++++++++++++++++++++++++++ tests/worker_reliability_test.rs | 13 +++++++++-- 9 files changed, 97 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 91d9615..2bff1e2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -49,6 +49,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Relative SQLite queue delays saturate instead of overflowing their timestamp representation. (#74) +- **Activity acknowledgement retries** - Retry transient provider errors when + acknowledging activity success or application errors, up to five attempts with + lock renewal retained. This avoids unnecessary activity re-execution. + Cancellation acknowledgements remain single-attempt, and retries do not extend + the shutdown grace period. A permanent error after a transient acknowledgement + error is reported as an uncertain outcome rather than a confirmed infrastructure + failure; neither commit nor redelivery is guaranteed. Uncertain outcomes are + counted separately in `duroxide_activity_ack_uncertain_total`. (#72) + ## [0.1.32] - 2026-10-11 **Release:** diff --git a/docs/metrics-specification.md b/docs/metrics-specification.md index 99ad69d..b7dd897 100644 --- a/docs/metrics-specification.md +++ b/docs/metrics-specification.md @@ -32,6 +32,7 @@ This document provides a complete reference of all metrics emitted by duroxide v | `duroxide_activity_duration_seconds` | Histogram | Activity | `activity_name`, `outcome`, `tag` | ✅ `test_activity_duration_tracking` | ✅ Code Review | Buckets: 0.01-300s | | `duroxide_activity_infrastructure_errors_total` | Counter | Activity | `activity_name` | ✅ `metrics_capture_activity_and_orchestration_outcomes` | ✅ Code Review | Infra-specific errors | | `duroxide_activity_configuration_errors_total` | Counter | Activity | `activity_name` | ✅ `test_error_classification_metrics` | ✅ Code Review | Config-specific errors | +| `duroxide_activity_ack_uncertain_total` | Counter | Activity | _(none)_ | ✅ `test_metrics_activity_ack_uncertain` | ✅ Emitted | Counts uncertain acknowledgement outcomes, not confirmed failures | | **Poison Message** | | `duroxide_orchestration_poison_total` | Counter | Poison | _(none)_ | ✅ `test_poison_message_metrics` | ✅ Code Review | Orchestrations exceeding max_attempts | | `duroxide_activity_poison_total` | Counter | Poison | _(none)_ | ✅ `test_poison_message_metrics` | ✅ Code Review | Activities exceeding max_attempts | @@ -485,6 +486,21 @@ sum by (activity_name, tag) (rate(duroxide_activity_admission_deferrals_total[5m --- +### 2.6 `duroxide_activity_ack_uncertain_total` (Counter) + +**Description:** Activity completion acknowledgements ending in a permanent error +after a retryable error. Incremented once per uncertain outcome, not per retry. + +**Labels:** _(none)_ + +**Purpose:** Track uncertain acknowledgements separately from confirmed activity +infrastructure failures. An earlier attempt may have committed or the lock may +have been lost; redelivery is not guaranteed. This counter does not increment +`duroxide_activity_executions_total` or `duroxide_activity_infrastructure_errors_total`. +The runtime metrics snapshot exposes it as `activity_ack_uncertain`. + +--- + ## 3. Sub-Orchestration Metrics > **Session Metrics (Future):** Session-specific metrics (session claims, migrations, @@ -935,4 +951,3 @@ RuntimeOptions { --- **Document Status:** This specification reflects the **current implementation** as of 2025-12-13. Client metrics are defined but not yet instrumented. All implemented metrics have test coverage in `tests/observability_tests.rs`. Poison message handling was added in v1.1. - diff --git a/docs/observability-guide.md b/docs/observability-guide.md index 92c5b58..f437e93 100644 --- a/docs/observability-guide.md +++ b/docs/observability-guide.md @@ -198,6 +198,7 @@ All metrics follow Prometheus naming conventions with `_total` for counters and | `duroxide_activity_admission_deferrals_total` | Counter | Activity | activity_name, tag | Track admission deferrals whose abandon returned success | | `duroxide_activity_duration_seconds` | Histogram | Activity | activity_name, outcome | Activity latency p95/p99 | | `duroxide_activity_errors_total` | Counter | Activity | activity_name, error_type, retryable | Detailed error tracking | +| `duroxide_activity_ack_uncertain_total` | Counter | Activity | _(none)_ | Track uncertain acknowledgement outcomes separately from failures | | **Sub-Orchestration** | | `duroxide_suborchestration_calls_total` | Counter | Sub-Orch | parent_orchestration, child_orchestration, outcome | Track sub-orch calls | | `duroxide_suborchestration_duration_seconds` | Histogram | Sub-Orch | parent_orchestration, child_orchestration, outcome | Sub-orch latency | @@ -299,6 +300,11 @@ before admission count as infrastructure errors, not deferrals. - **Labels**: `activity_name`, `error_type` (app_error|infrastructure_error|config_error), `retryable` (true|false) - **Use**: Distinguish transient vs permanent errors +**`duroxide_activity_ack_uncertain_total`** (Counter) +- **Description**: Activity completion acknowledgements ending in a permanent error after a retryable error +- **Labels**: _(none)_ +- **Use**: Track uncertainty separately from confirmed infrastructure failures. Counted once per uncertain outcome, not per retry. An earlier attempt may have committed or the lock may have been lost; redelivery is not guaranteed. + #### Sub-Orchestration Metrics **`duroxide_suborchestration_calls_total`** (Counter) diff --git a/docs/provider-implementation-guide.md b/docs/provider-implementation-guide.md index d6e4d1b..dfdbf6e 100644 --- a/docs/provider-implementation-guide.md +++ b/docs/provider-implementation-guide.md @@ -941,9 +941,11 @@ delete-and-enqueue operation and lock validation described below. A permanent error after a retryable acknowledgement error is reported as an **uncertain outcome**, not as a confirmed activity infrastructure failure. The runtime cannot distinguish an earlier commit from lock loss or another permanent -error. It does not attempt another abandonment or count an activity infrastructure -error in this case; any uncommitted work remains eligible for redelivery after -its lock expires. +error. It increments `duroxide_activity_ack_uncertain_total` once for this outcome, +without attempting abandonment or counting an activity infrastructure error. +Redelivery is not guaranteed: the work item may already be gone. If uncommitted +work remains in the queue, it can become eligible for redelivery after its lock +expires. Acknowledgement retries do not extend the shutdown grace period. The default `shutdown(None)` grace is 1 second, shorter than the full 1.5-second backoff budget, diff --git a/src/runtime/dispatchers/worker.rs b/src/runtime/dispatchers/worker.rs index e7794c6..33b703e 100644 --- a/src/runtime/dispatchers/worker.rs +++ b/src/runtime/dispatchers/worker.rs @@ -920,8 +920,9 @@ async fn handle_activity_outcome( worker_id = %ctx.worker_id, ack_outcome = "uncertain", error = %e, - "worker: acknowledgement outcome uncertain after retry; any uncommitted work will be redelivered after lock expiry" + "worker: acknowledgement outcome uncertain after retry; an earlier attempt may have committed or the lock may have been lost" ); + rt.record_activity_ack_uncertain(); } Err(ActivityAckError::Failed(e)) => { warn!( diff --git a/src/runtime/mod.rs b/src/runtime/mod.rs index dc3d767..bc7ebb2 100644 --- a/src/runtime/mod.rs +++ b/src/runtime/mod.rs @@ -720,6 +720,13 @@ impl Runtime { } } + #[inline] + fn record_activity_ack_uncertain(&self) { + if let Some(provider) = self.metrics_provider() { + provider.record_activity_ack_uncertain(); + } + } + #[inline] fn record_orchestration_poison(&self) { if let Some(provider) = self.metrics_provider() { diff --git a/src/runtime/observability.rs b/src/runtime/observability.rs index 00def86..7f98bc0 100644 --- a/src/runtime/observability.rs +++ b/src/runtime/observability.rs @@ -131,6 +131,7 @@ pub struct MetricsSnapshot { pub activity_success: u64, pub activity_app_errors: u64, pub activity_infra_errors: u64, + pub activity_ack_uncertain: u64, pub activity_config_errors: u64, pub activity_poison: u64, pub orch_dispatcher_items_fetched: u64, @@ -157,6 +158,7 @@ pub struct MetricsProvider { activity_success_atomic: AtomicU64, activity_app_errors_atomic: AtomicU64, activity_infra_errors_atomic: AtomicU64, + activity_ack_uncertain_atomic: AtomicU64, activity_config_errors_atomic: AtomicU64, activity_poison_atomic: AtomicU64, activity_admission_deferrals_atomic: AtomicU64, @@ -196,6 +198,7 @@ impl MetricsProvider { activity_success_atomic: AtomicU64::new(0), activity_app_errors_atomic: AtomicU64::new(0), activity_infra_errors_atomic: AtomicU64::new(0), + activity_ack_uncertain_atomic: AtomicU64::new(0), activity_config_errors_atomic: AtomicU64::new(0), activity_poison_atomic: AtomicU64::new(0), activity_admission_deferrals_atomic: AtomicU64::new(0), @@ -453,6 +456,12 @@ impl MetricsProvider { self.activity_infra_errors_atomic.fetch_add(1, Ordering::Relaxed); } + #[inline] + pub fn record_activity_ack_uncertain(&self) { + counter!("duroxide_activity_ack_uncertain_total").increment(1); + self.activity_ack_uncertain_atomic.fetch_add(1, Ordering::Relaxed); + } + #[inline] pub fn record_activity_config_error(&self) { counter!("duroxide_activity_executions_total", "outcome" => "config_error").increment(1); @@ -642,6 +651,7 @@ impl MetricsProvider { activity_success: self.activity_success_atomic.load(Ordering::Relaxed), activity_app_errors: self.activity_app_errors_atomic.load(Ordering::Relaxed), activity_infra_errors: self.activity_infra_errors_atomic.load(Ordering::Relaxed), + activity_ack_uncertain: self.activity_ack_uncertain_atomic.load(Ordering::Relaxed), activity_config_errors: self.activity_config_errors_atomic.load(Ordering::Relaxed), activity_poison: self.activity_poison_atomic.load(Ordering::Relaxed), orch_dispatcher_items_fetched: self.orch_dispatcher_items_fetched_atomic.load(Ordering::Relaxed), diff --git a/tests/observability_tests.rs b/tests/observability_tests.rs index 199f062..9b04bbd 100644 --- a/tests/observability_tests.rs +++ b/tests/observability_tests.rs @@ -1697,6 +1697,37 @@ async fn test_metrics_activity_outcomes() { assert_eq!(snapshot.activity_infra_errors, 1); } +#[test] +fn test_metrics_activity_ack_uncertain() { + use duroxide::runtime::observability::{MetricsProvider, ObservabilityConfig}; + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; + + let metrics = MetricsProvider::new(&ObservabilityConfig::default()).unwrap(); + let recorder = DebuggingRecorder::new(); + let mut expected = metrics.snapshot(); + assert_eq!(expected.activity_ack_uncertain, 0); + + metrics::with_local_recorder(&recorder, || { + metrics.record_activity_ack_uncertain(); + metrics.record_activity_ack_uncertain(); + }); + + expected.activity_ack_uncertain = 2; + assert_eq!( + metrics.snapshot(), + expected, + "Other activity outcome counters must not change" + ); + + let recorded = recorder.snapshotter().snapshot().into_vec(); + let [(key, _, _, value)] = recorded.as_slice() else { + panic!("Only the dedicated uncertainty counter should be emitted: {recorded:?}"); + }; + assert_eq!(key.key().name(), "duroxide_activity_ack_uncertain_total"); + assert_eq!(key.key().labels().count(), 0); + assert_eq!(value, &DebugValue::Counter(2)); +} + /// Test: MetricsProvider poison message tracking #[tokio::test] async fn test_metrics_poison_messages() { diff --git a/tests/worker_reliability_test.rs b/tests/worker_reliability_test.rs index b7c2fba..f5b1ff1 100644 --- a/tests/worker_reliability_test.rs +++ b/tests/worker_reliability_test.rs @@ -470,6 +470,7 @@ async fn worker_retries_retryable_ack_failure_without_reexecuting_activity() { } if result == "done" )); assert_eq!(calls, vec![calls[0].clone(); 5]); + assert_eq!(test.rt.metrics_snapshot().unwrap().activity_ack_uncertain, 0); } #[tokio::test] @@ -664,7 +665,9 @@ async fn worker_retryable_ack_error_after_commit_is_reported_as_uncertain() { }; assert_eq!(result, "done"); assert_eq!(calls[0], calls[1]); - assert_eq!(test.rt.metrics_snapshot().unwrap().activity_infra_errors, 0); + let metrics = test.rt.metrics_snapshot().unwrap(); + assert_eq!(metrics.activity_infra_errors, 0); + assert_eq!(metrics.activity_ack_uncertain, 1); let events = captured.lock().unwrap(); assert!( @@ -679,6 +682,10 @@ async fn worker_retryable_ack_error_after_commit_is_reported_as_uncertain() { .any(|event| event.message.contains("activity acknowledgment or deferral failed")), "An uncertain acknowledgement must not be reported as a confirmed failure" ); + assert!( + !events.iter().any(|event| event.message.contains("will be redelivered")), + "An uncertain acknowledgement must not promise redelivery when the row may be gone" + ); } #[tokio::test] @@ -694,7 +701,9 @@ async fn worker_uncertain_ack_before_commit_recovers_after_lock_expiry() { runtime::OrchestrationStatus::Completed { output, .. } if output == "done" )); assert_eq!(execution_count.load(Ordering::SeqCst), 2); - assert_eq!(test.rt.metrics_snapshot().unwrap().activity_infra_errors, 0); + let metrics = test.rt.metrics_snapshot().unwrap(); + assert_eq!(metrics.activity_infra_errors, 0); + assert_eq!(metrics.activity_ack_uncertain, 1); let calls = test.provider.work_item_calls(); assert_eq!( From 71032ce03d997910d7582978edebc96dd8bd540a Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Fri, 30 Oct 2026 18:00:40 +0000 Subject: [PATCH 5/6] Fix review N1/N2 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Restored MetricsSnapshot’s original fields and added MetricsProvider::activity_ack_uncertain(). Updated callers and documentation. A new exhaustive struct-literal test reproduced E0063 before the fix and now passes. - Added the missing telemetry specification entry and assertions that both confirmed-failure paths leave uncertainty at zero. --- docs/duroxide-telemetry-spec.md | 16 ++++++++++++++++ docs/metrics-specification.md | 4 +++- src/runtime/observability.rs | 8 ++++++-- tests/observability_tests.rs | 6 +++--- tests/worker_reliability_test.rs | 20 +++++++++++++------- 5 files changed, 41 insertions(+), 13 deletions(-) diff --git a/docs/duroxide-telemetry-spec.md b/docs/duroxide-telemetry-spec.md index 7f15b9a..e6ba14e 100644 --- a/docs/duroxide-telemetry-spec.md +++ b/docs/duroxide-telemetry-spec.md @@ -209,6 +209,22 @@ sum by (activity_name, tag) (rate(duroxide_activity_admission_deferrals_total[5m --- +#### `duroxide_activity_ack_uncertain_total` (Counter) + +**Description:** Activity completion acknowledgements ending in a permanent error +after a retryable error. Incremented once per uncertain outcome, not per retry. + +**Labels:** _(none)_ + +**Use Case:** Track uncertain acknowledgements separately from confirmed +infrastructure failures. An earlier attempt may have committed or the lock may +have been lost; redelivery is not guaranteed. Successful retries and confirmed +acknowledgement failures do not increment this counter. + +Read the in-process count through `MetricsProvider::activity_ack_uncertain()`. + +--- + ### 1.3 Provider (Storage) Metrics #### `duroxide_provider_operation_duration_seconds` (Histogram) diff --git a/docs/metrics-specification.md b/docs/metrics-specification.md index b7dd897..5e6ad51 100644 --- a/docs/metrics-specification.md +++ b/docs/metrics-specification.md @@ -497,7 +497,8 @@ after a retryable error. Incremented once per uncertain outcome, not per retry. infrastructure failures. An earlier attempt may have committed or the lock may have been lost; redelivery is not guaranteed. This counter does not increment `duroxide_activity_executions_total` or `duroxide_activity_infrastructure_errors_total`. -The runtime metrics snapshot exposes it as `activity_ack_uncertain`. +Read the in-process count through `MetricsProvider::activity_ack_uncertain()`; +the public fields of `MetricsSnapshot` are unchanged. --- @@ -951,3 +952,4 @@ RuntimeOptions { --- **Document Status:** This specification reflects the **current implementation** as of 2025-12-13. Client metrics are defined but not yet instrumented. All implemented metrics have test coverage in `tests/observability_tests.rs`. Poison message handling was added in v1.1. + diff --git a/src/runtime/observability.rs b/src/runtime/observability.rs index 7f98bc0..47682eb 100644 --- a/src/runtime/observability.rs +++ b/src/runtime/observability.rs @@ -131,7 +131,6 @@ pub struct MetricsSnapshot { pub activity_success: u64, pub activity_app_errors: u64, pub activity_infra_errors: u64, - pub activity_ack_uncertain: u64, pub activity_config_errors: u64, pub activity_poison: u64, pub orch_dispatcher_items_fetched: u64, @@ -462,6 +461,12 @@ impl MetricsProvider { self.activity_ack_uncertain_atomic.fetch_add(1, Ordering::Relaxed); } + /// Number of activity completion acknowledgements with an uncertain outcome. + #[inline] + pub fn activity_ack_uncertain(&self) -> u64 { + self.activity_ack_uncertain_atomic.load(Ordering::Relaxed) + } + #[inline] pub fn record_activity_config_error(&self) { counter!("duroxide_activity_executions_total", "outcome" => "config_error").increment(1); @@ -651,7 +656,6 @@ impl MetricsProvider { activity_success: self.activity_success_atomic.load(Ordering::Relaxed), activity_app_errors: self.activity_app_errors_atomic.load(Ordering::Relaxed), activity_infra_errors: self.activity_infra_errors_atomic.load(Ordering::Relaxed), - activity_ack_uncertain: self.activity_ack_uncertain_atomic.load(Ordering::Relaxed), activity_config_errors: self.activity_config_errors_atomic.load(Ordering::Relaxed), activity_poison: self.activity_poison_atomic.load(Ordering::Relaxed), orch_dispatcher_items_fetched: self.orch_dispatcher_items_fetched_atomic.load(Ordering::Relaxed), diff --git a/tests/observability_tests.rs b/tests/observability_tests.rs index 9b04bbd..5f3cd89 100644 --- a/tests/observability_tests.rs +++ b/tests/observability_tests.rs @@ -1704,15 +1704,15 @@ fn test_metrics_activity_ack_uncertain() { let metrics = MetricsProvider::new(&ObservabilityConfig::default()).unwrap(); let recorder = DebuggingRecorder::new(); - let mut expected = metrics.snapshot(); - assert_eq!(expected.activity_ack_uncertain, 0); + let expected = metrics.snapshot(); + assert_eq!(metrics.activity_ack_uncertain(), 0); metrics::with_local_recorder(&recorder, || { metrics.record_activity_ack_uncertain(); metrics.record_activity_ack_uncertain(); }); - expected.activity_ack_uncertain = 2; + assert_eq!(metrics.activity_ack_uncertain(), 2); assert_eq!( metrics.snapshot(), expected, diff --git a/tests/worker_reliability_test.rs b/tests/worker_reliability_test.rs index f5b1ff1..c08f403 100644 --- a/tests/worker_reliability_test.rs +++ b/tests/worker_reliability_test.rs @@ -398,6 +398,10 @@ impl AckTest { Arc::clone(&self.rt).shutdown(Some(10_000)).await; status } + + fn metrics_provider(&self) -> &runtime::observability::MetricsProvider { + self.rt.observability_handle().unwrap().metrics_provider() + } } fn counting_activity(result: Result) -> (ActivityRegistry, Arc) { @@ -444,6 +448,8 @@ async fn worker_abandon_on_ack_failure_enables_retry() { assert_eq!(token, abandoned); assert_ne!(token, redelivered); assert_eq!(completion, retried_completion); + assert_eq!(test.rt.metrics_snapshot().unwrap().activity_infra_errors, 1); + assert_eq!(test.metrics_provider().activity_ack_uncertain(), 0); } #[tokio::test] @@ -470,7 +476,7 @@ async fn worker_retries_retryable_ack_failure_without_reexecuting_activity() { } if result == "done" )); assert_eq!(calls, vec![calls[0].clone(); 5]); - assert_eq!(test.rt.metrics_snapshot().unwrap().activity_ack_uncertain, 0); + assert_eq!(test.metrics_provider().activity_ack_uncertain(), 0); } #[tokio::test] @@ -503,6 +509,8 @@ async fn worker_abandons_after_retryable_ack_attempts_are_exhausted() { }; assert_ne!(token, redelivered); assert_eq!(completion, retried_completion); + assert_eq!(test.rt.metrics_snapshot().unwrap().activity_infra_errors, 1); + assert_eq!(test.metrics_provider().activity_ack_uncertain(), 0); } #[tokio::test] @@ -665,9 +673,8 @@ async fn worker_retryable_ack_error_after_commit_is_reported_as_uncertain() { }; assert_eq!(result, "done"); assert_eq!(calls[0], calls[1]); - let metrics = test.rt.metrics_snapshot().unwrap(); - assert_eq!(metrics.activity_infra_errors, 0); - assert_eq!(metrics.activity_ack_uncertain, 1); + assert_eq!(test.rt.metrics_snapshot().unwrap().activity_infra_errors, 0); + assert_eq!(test.metrics_provider().activity_ack_uncertain(), 1); let events = captured.lock().unwrap(); assert!( @@ -701,9 +708,8 @@ async fn worker_uncertain_ack_before_commit_recovers_after_lock_expiry() { runtime::OrchestrationStatus::Completed { output, .. } if output == "done" )); assert_eq!(execution_count.load(Ordering::SeqCst), 2); - let metrics = test.rt.metrics_snapshot().unwrap(); - assert_eq!(metrics.activity_infra_errors, 0); - assert_eq!(metrics.activity_ack_uncertain, 1); + assert_eq!(test.rt.metrics_snapshot().unwrap().activity_infra_errors, 0); + assert_eq!(test.metrics_provider().activity_ack_uncertain(), 1); let calls = test.provider.work_item_calls(); assert_eq!( From 1e883a3dbd6752c2076b7985d0f2757fbb7ffa98 Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Tue, 3 Nov 2026 15:31:54 +0000 Subject: [PATCH 6/6] Add provider notes for 0.2.0 --- CHANGELOG.md | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2bff1e2..f08b4cc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -58,6 +58,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 failure; neither commit nor redelivery is guaranteed. Uncertain outcomes are counted separately in `duroxide_activity_ack_uncertain_total`. (#72) +### Provider notes for 0.2.0 + +- **Duplicate activity acknowledgements** - Completion acknowledgements can be + retried after the transaction committed but its response was lost. Providers + must atomically remove the locked worker item and enqueue its completion. + After that commit, repeating `ack_work_item(token, Some(completion))` with the + same token must return a non-retryable error without enqueueing another + completion. See the [provider contract](docs/provider-implementation-guide.md#error-classification). (#72) +- After upgrading the Duroxide dependency, register + [`duroxide::provider_validations::test_ack_work_item_duplicate_rejected`](src/provider_validation/atomicity.rs) + in your provider's validation suite. It acknowledges the same token twice and + verifies that the second call fails non-retryably, exactly one completion is + queued, and the worker item is removed. SQLite already runs this validation; + external providers must add their own wrapper. (#72) + ## [0.1.32] - 2026-10-11 **Release:**