diff --git a/CHANGELOG.md b/CHANGELOG.md index 91d9615..f08b4cc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -49,6 +49,30 @@ 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) + +### 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:** 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 99ad69d..5e6ad51 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,22 @@ 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`. +Read the in-process count through `MetricsProvider::activity_ack_uncertain()`; +the public fields of `MetricsSnapshot` are unchanged. + +--- + ## 3. Sub-Orchestration Metrics > **Session Metrics (Future):** Session-specific metrics (session claims, migrations, 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 ed71f4a..dfdbf6e 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,47 @@ 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 or application errors, +the runtime retries retryable errors in place up to five total attempts before +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 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, +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 d7f0141..33b703e 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,10 +704,10 @@ 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()); + // 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!( @@ -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) } } } @@ -722,13 +727,61 @@ 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 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. 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, + completion: WorkItem, +) -> Result<(), ActivityAckError> { + let mut attempt = 1; + loop { + 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); + 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; + } + // 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)), + } + } +} + /// Handle successful activity completion. async fn handle_activity_success( rt: &Arc, 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; @@ -748,18 +801,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_activity_completion_with_retry( + rt, + ctx, + WorkItem::ActivityCompleted { + instance: ctx.instance.clone(), + execution_id: ctx.execution_id, + id: ctx.activity_id, + result, + }, + ) + .await; (ack_result, ActivityOutcome::Success) } @@ -770,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; @@ -790,22 +842,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_activity_completion_with_retry( + rt, + ctx, + 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) } @@ -849,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 { @@ -861,7 +912,19 @@ 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; an earlier attempt may have committed or the lock may have been lost" + ); + rt.record_activity_ack_uncertain(); + } + 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..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() { @@ -1110,6 +1117,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/src/runtime/observability.rs b/src/runtime/observability.rs index 00def86..47682eb 100644 --- a/src/runtime/observability.rs +++ b/src/runtime/observability.rs @@ -157,6 +157,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 +197,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 +455,18 @@ 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); + } + + /// 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); diff --git a/tests/common/fault_injection.rs b/tests/common/fault_injection.rs index 89d6463..5f8d508 100644 --- a/tests/common/fault_injection.rs +++ b/tests/common/fault_injection.rs @@ -641,14 +641,29 @@ 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 { 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, + work_item_calls: std::sync::Mutex>, 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) @@ -660,9 +675,12 @@ impl FailingProvider { Self { 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), + fail_next_renew_work_item_lock: AtomicBool::new(false), allow_failure_commits: AtomicBool::new(false), ack_then_fail: AtomicBool::new(false), } @@ -672,6 +690,23 @@ impl FailingProvider { self.fail_next_ack_work_item.store(true, Ordering::SeqCst); } + /// Fail the next `count` `ack_work_item` calls with a *retryable* error. + /// + /// 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); + } + + /// 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) + } + + /// 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); } @@ -684,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) { @@ -798,21 +837,48 @@ impl Provider for FailingProvider { } async fn ack_work_item(&self, token: &str, completion: Option) -> Result<(), ProviderError> { - 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) { - self.inner.ack_work_item(token, completion).await?; - } - Err(ProviderError::permanent( + 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) + }) + .is_ok() + { + 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 let Some(error) = failure { + if self.ack_then_fail.load(Ordering::SeqCst) { + self.inner.ack_work_item(token, completion).await?; + } + Err(error) } else { self.inner.ack_work_item(token, completion).await } } 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 } @@ -822,6 +888,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/observability_tests.rs b/tests/observability_tests.rs index 199f062..5f3cd89 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 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(); + }); + + assert_eq!(metrics.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/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/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 b04dc90..c08f403 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::WorkItem; +use duroxide::providers::sqlite::SqliteProvider; 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,80 +339,395 @@ 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 metrics_provider(&self) -> &runtime::observability::MetricsProvider { + self.rt.observability_handle().unwrap().metrics_provider() + } +} + +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:?}"); }; + 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] +async fn worker_retries_retryable_ack_failure_without_reexecuting_activity() { + 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]); + assert_eq!(test.metrics_provider().activity_ack_uncertain(), 0); +} - 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()); +#[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); + assert_eq!(test.rt.metrics_snapshot().unwrap().activity_infra_errors, 1); + assert_eq!(test.metrics_provider().activity_ack_uncertain(), 0); +} - // 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(); +#[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]); +} - client - .start_orchestration("ack-fail-test", "AckFailOrch", "") +#[tokio::test] +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)); + let counter = execution_count.clone(); + let activities = ActivityRegistry::builder() + .register("CountingActivity", move |_ctx: ActivityContext, _input: String| { + let entered = activity_entered.clone(); + let counter = counter.clone(); + async move { + counter.fetch_add(1, Ordering::SeqCst); + entered.notify_one(); + std::future::pending::>().await + } + }) + .build(); + let test = AckTest::new(activities).await; + test.start().await; + tokio::time::timeout(Duration::from_secs(5), entered.notified()) .await .unwrap(); - - // Wait for completion - let status = client - .wait_for_orchestration("ack-fail-test", Duration::from_secs(5)) + test.provider.fail_next_ack_work_item_retryable(1); + 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(), 2); + let WorkItemCall::Ack { + token, + completion: None, + } = &calls[0] + else { + panic!("Cancellation must acknowledge without a completion: {calls:?}"); + }; + assert_eq!(calls[1], WorkItemCall::Abandon { token: token.clone() }); +} + +#[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"); +} - // Activity should have executed at least once - // (may execute twice if retry happened - that's fine for this test) +#[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); + 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(), 2); + let WorkItemCall::Ack { + 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!(test.rt.metrics_snapshot().unwrap().activity_infra_errors, 0); + assert_eq!(test.metrics_provider().activity_ack_uncertain(), 1); + + let events = captured.lock().unwrap(); + assert!( + events + .iter() + .any(|event| event.field("ack_outcome").as_deref() == Some("uncertain")), + "The runtime must report an uncertain acknowledgement outcome" + ); assert!( - execution_count.load(Ordering::SeqCst) >= 1, - "Activity should execute at least once" + !events + .iter() + .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" ); +} - rt.shutdown(None).await; +#[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); + assert_eq!(test.metrics_provider().activity_ack_uncertain(), 1); + + 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); }