From 4771c7397f4687dcfe85c67044b1fac5f0c2bb42 Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Fri, 23 Oct 2026 17:57:56 +0000 Subject: [PATCH 01/11] Add nonblocking activity admission control Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- docs/ORCHESTRATION-GUIDE.md | 47 +++++- src/runtime/dispatchers/worker.rs | 66 +++++++-- src/runtime/mod.rs | 32 ++++ src/runtime/registry.rs | 46 +++++- tests/activity_admission_tests.rs | 235 ++++++++++++++++++++++++++++++ 5 files changed, 413 insertions(+), 13 deletions(-) create mode 100644 tests/activity_admission_tests.rs diff --git a/docs/ORCHESTRATION-GUIDE.md b/docs/ORCHESTRATION-GUIDE.md index 9a89393..efc8e57 100644 --- a/docs/ORCHESTRATION-GUIDE.md +++ b/docs/ORCHESTRATION-GUIDE.md @@ -242,6 +242,52 @@ activities.register("CallExternalAPI", |ctx: ActivityContext, request: String| a | `ctx.is_cancelled()` | Returns `true` if cancellation was requested | | `ctx.cancelled()` | Returns a `Future` that completes when cancelled (use in `select!`) | +### Activity Admission + +Use `ActivityRegistry::builder().register_with_admission(name, admission, handler)` +to impose resource limits without occupying worker slots while waiting. +The synchronous admission callback receives `&ActivityContext` and the activity's +input as `&str`. It must not block or perform I/O. + +```rust +use duroxide::runtime::{ActivityAdmission, registry::ActivityRegistry}; +use std::{sync::Arc, time::Duration}; +use tokio::sync::{Semaphore, TryAcquireError}; + +let capacity = Arc::new(Semaphore::new(4)); +let activities = ActivityRegistry::builder() + .register_with_admission( + "Limited", + move |_ctx, _input| match capacity.clone().try_acquire_owned() { + Ok(permit) => Ok(ActivityAdmission::admit(permit)), + Err(TryAcquireError::NoPermits) => { + Ok(ActivityAdmission::Defer(Duration::from_millis(100))) + } + Err(error) => Err(error.to_string()), + }, + |_ctx, input| async move { Ok(input) }, + ) + .build(); +``` + +`admit(guard)` retains a resource reservation until the handler returns or its +task is dropped, including error, panic, and cancellation. If several limits +apply, reserve all of them atomically or defer holding none. Detached work is +not protected after the handler finishes. + +`Defer(delay)` requires a positive delay. It leaves the original activity +pending in durable storage, frees the worker to fetch other work, and does not +record an activity result or consume the poison-message fetch-attempt budget. +It is not a retry of an executed activity. An admission error, panic, or invalid +delay fails the activity rather than invoking the handler. Storage errors during +deferral are reported as infrastructure errors. + +Admission runs outside orchestration replay; completed results are unaffected. +Each recovered execution attempt is admitted again. Resource counters belong +to the application and are not automatically coordinated across runtimes. +Keep total worker capacity above the admitted activity ceiling if other work +needs headroom. Admission does not guarantee fair scheduling between workflows. + ### The Replay Model @@ -2693,4 +2739,3 @@ async fn fast_updates(ctx: OrchestrationContext, items_json: String) -> Result<( --- **You now have everything needed to build production-grade durable workflows!** 🎉 - diff --git a/src/runtime/dispatchers/worker.rs b/src/runtime/dispatchers/worker.rs index 8bb1124..85d09e4 100644 --- a/src/runtime/dispatchers/worker.rs +++ b/src/runtime/dispatchers/worker.rs @@ -39,7 +39,7 @@ use tokio_util::sync::CancellationToken; use tokio_util::task::{AbortOnDropHandle, TaskTracker}; use tracing::{error, warn}; -use super::super::{Runtime, registry}; +use super::super::{ActivityAdmission, Runtime, registry}; // ============================================================================ // Types @@ -123,6 +123,13 @@ enum ActivityOutcome { AppError, /// Orchestration was cancelled, activity result dropped Cancelled, + /// Admission deferred without executing the handler + Deferred, +} + +enum ActivityInvocation { + Completed(String), + Deferred(Duration), } /// Context for processing a single activity work item. @@ -559,17 +566,41 @@ async fn run_activity_with_cancellation( let input = ctx.input.clone(); // Abort-on-drop so the invocation cannot outlive this task if the worker is // aborted mid-activity during shutdown. - let mut activity_handle = AbortOnDropHandle::new( - rt.task_tracker - .spawn(async move { handler.invoke(activity_ctx, input).await }), - ); + let mut activity_handle = AbortOnDropHandle::new(rt.task_tracker.spawn(async move { + let _admission_guard = match handler.try_admit(&activity_ctx, &input)? { + ActivityAdmission::Admit(guard) => guard, + ActivityAdmission::Defer(delay) if delay.is_zero() => { + return Err("Activity admission deferral requires a positive retry delay".to_string()); + } + ActivityAdmission::Defer(delay) => return Ok(ActivityInvocation::Deferred(delay)), + }; + handler + .invoke(activity_ctx, input) + .await + .map(ActivityInvocation::Completed) + })); tokio::select! { joined = &mut activity_handle => { // Keep the renewal guard until acknowledgement finishes, not just invocation. // Handle normal activity completion (success, error, or panic) match joined { - Ok(Ok(result)) => handle_activity_success(rt, ctx, result, start_time).await, + Ok(Ok(ActivityInvocation::Completed(result))) => { + handle_activity_success(rt, ctx, result, start_time).await + } + Ok(Ok(ActivityInvocation::Deferred(delay))) => { + tracing::debug!( + instance = %ctx.instance, + activity_name = %ctx.activity_name, + activity_id = ctx.activity_id, + retry_after_ms = delay.as_millis(), + "Activity admission deferred" + ); + let result = rt.history_store + .abandon_work_item(&ctx.lock_token, Some(delay), true) + .await; + (result, ActivityOutcome::Deferred) + } 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, } @@ -761,7 +792,7 @@ async fn handle_activity_outcome( Ok(()) => match outcome { ActivityOutcome::Success => rt.record_activity_success(), ActivityOutcome::AppError => rt.record_activity_app_error(), - ActivityOutcome::Cancelled => {} + ActivityOutcome::Cancelled | ActivityOutcome::Deferred => {} }, Err(e) => { warn!( @@ -770,12 +801,25 @@ async fn handle_activity_outcome( activity_id = ctx.activity_id, worker_id = %ctx.worker_id, error = %e, - "worker: atomic ack failed, abandoning work item" + outcome = ?outcome, + "worker: activity acknowledgment or deferral failed, abandoning work item" ); - let _ = rt + if let Err(error) = rt .history_store - .abandon_work_item(&ctx.lock_token, Some(Duration::from_millis(100)), false) - .await; + .abandon_work_item( + &ctx.lock_token, + Some(Duration::from_millis(100)), + matches!(outcome, ActivityOutcome::Deferred), + ) + .await + { + warn!( + instance = %ctx.instance, + activity_id = ctx.activity_id, + error = %error, + "Failed to abandon activity work item; waiting for lock expiry" + ); + } rt.record_activity_infra_error(); } } diff --git a/src/runtime/mod.rs b/src/runtime/mod.rs index 268fda7..c5f1efe 100644 --- a/src/runtime/mod.rs +++ b/src/runtime/mod.rs @@ -455,9 +455,41 @@ where } } +/// A nonblocking admission decision made before invoking an activity. +pub enum ActivityAdmission { + /// The guard is held until the activity handler returns or its task is dropped. + Admit(Box), + /// Leave the activity pending and retry admission after a positive delay. + /// + /// Deferral releases the worker slot without recording a history event or + /// consuming a fetch attempt. It does not retry an executed activity. + Defer(Duration), +} + +impl ActivityAdmission { + /// Admit an activity while retaining ownership of its resource reservation. + pub fn admit(guard: impl Send + 'static) -> Self { + Self::Admit(Box::new(guard)) + } +} + /// Trait implemented by activity handlers that can be invoked by the runtime. #[async_trait] pub trait ActivityHandler: Send + Sync { + /// Reserve resources without blocking or performing I/O. + /// + /// Called again on each deferred fetch, outside orchestration replay. An + /// error rejects the activity as an application failure. A returned guard + /// must release its reservation on drop, including cancellation and panic. + /// + /// # Errors + /// + /// Return an error when admission configuration is invalid or a required + /// resource is unavailable permanently, rather than temporarily at capacity. + fn try_admit(&self, _ctx: &crate::ActivityContext, _input: &str) -> Result { + Ok(ActivityAdmission::admit(())) + } + async fn invoke(&self, ctx: crate::ActivityContext, input: String) -> Result; } diff --git a/src/runtime/registry.rs b/src/runtime/registry.rs index 2d0f905..b7c25d5 100644 --- a/src/runtime/registry.rs +++ b/src/runtime/registry.rs @@ -13,7 +13,7 @@ #![allow(clippy::unwrap_used)] #![allow(clippy::clone_on_ref_ptr)] -use super::{ActivityHandler, FnActivity, FnOrchestration, OrchestrationHandler}; +use super::{ActivityAdmission, ActivityHandler, FnActivity, FnOrchestration, OrchestrationHandler}; use crate::_typed_codec::Codec; use crate::OrchestrationContext; use semver::Version; @@ -411,6 +411,27 @@ impl OrchestrationRegistryBuilder { // Activity Builder - Specialized Methods // ============================================================================ +struct AdmissionControlledActivity { + admission: A, + handler: F, +} + +#[async_trait::async_trait] +impl ActivityHandler for AdmissionControlledActivity +where + A: Fn(&crate::ActivityContext, &str) -> Result + Send + Sync, + F: Fn(crate::ActivityContext, String) -> Fut + Send + Sync, + Fut: std::future::Future> + Send, +{ + fn try_admit(&self, ctx: &crate::ActivityContext, input: &str) -> Result { + (self.admission)(ctx, input) + } + + async fn invoke(&self, ctx: crate::ActivityContext, input: String) -> Result { + (self.handler)(ctx, input).await + } +} + impl ActivityRegistryBuilder { /// Deprecated: use `ActivityRegistry::builder_from` instead pub fn from_registry(reg: &ActivityRegistry) -> Self { @@ -453,6 +474,29 @@ impl ActivityRegistryBuilder { self } + /// Register an activity with a nonblocking resource-admission callback. + /// + /// The callback must atomically reserve all required capacity or return + /// `Defer` holding none. Admission applies to each execution attempt, not + /// replayed activity results. No cross-runtime coordination is provided. + pub fn register_with_admission(mut self, name: impl Into, admission: A, f: F) -> Self + where + A: Fn(&crate::ActivityContext, &str) -> Result + Send + Sync + 'static, + F: Fn(crate::ActivityContext, String) -> Fut + Send + Sync + 'static, + Fut: std::future::Future> + Send + 'static, + { + let name = name.into(); + if self.check_reserved_prefix(&name) || self.check_duplicate(&name, &DEFAULT_VERSION, "activity") { + return self; + } + self.map.entry(name.clone()).or_default().insert( + DEFAULT_VERSION, + Arc::new(AdmissionControlledActivity { admission, handler: f }), + ); + self.policy.insert(name, VersionPolicy::Latest); + self + } + pub fn register_typed(mut self, name: impl Into, f: F) -> Self where In: serde::de::DeserializeOwned + Send + 'static, diff --git a/tests/activity_admission_tests.rs b/tests/activity_admission_tests.rs new file mode 100644 index 0000000..580688e --- /dev/null +++ b/tests/activity_admission_tests.rs @@ -0,0 +1,235 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::clone_on_ref_ptr)] + +mod common; + +use duroxide::runtime::registry::ActivityRegistry; +use duroxide::runtime::{ActivityAdmission, Runtime, RuntimeOptions}; +use duroxide::{ActivityContext, Client, EventKind, OrchestrationContext, OrchestrationRegistry, OrchestrationStatus}; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::{Semaphore, mpsc}; + +fn options() -> RuntimeOptions { + RuntimeOptions { + worker_concurrency: 1, + orchestration_concurrency: 1, + dispatcher_min_poll_interval: Duration::from_millis(5), + dispatcher_long_poll_timeout: Duration::from_millis(10), + worker_lock_timeout: Duration::from_secs(2), + activity_cancellation_grace_period: Duration::from_millis(10), + max_attempts: 1, + ..Default::default() + } +} + +fn orchestrations() -> OrchestrationRegistry { + OrchestrationRegistry::builder() + .register("Run", |ctx: OrchestrationContext, activity: String| async move { + ctx.schedule_activity(activity, "payload").await + }) + .build() +} + +#[tokio::test(flavor = "current_thread")] +async fn deferred_admission_frees_worker_without_consuming_attempts_and_survives_restart() { + let (store, _dir) = common::create_sqlite_store_disk().await; + let capacity = Arc::new(Semaphore::new(1)); + let held = capacity.clone().acquire_owned().await.unwrap(); + let (deferred, mut deferrals) = mpsc::unbounded_channel(); + let activities = ActivityRegistry::builder() + .register_with_admission( + "Limited", + move |ctx, input| { + assert_eq!(ctx.instance_id(), "limited"); + assert_eq!(input, "payload"); + match capacity.clone().try_acquire_owned() { + Ok(permit) => Ok(ActivityAdmission::admit(permit)), + Err(tokio::sync::TryAcquireError::NoPermits) => { + deferred.send(()).unwrap(); + Ok(ActivityAdmission::Defer(Duration::from_millis(10))) + } + Err(error) => Err(error.to_string()), + } + }, + |_ctx, input| async move { Ok(input) }, + ) + .register("Free", |_ctx: ActivityContext, input: String| async move { Ok(input) }) + .build(); + let runtime = Runtime::start_with_options(store.clone(), activities.clone(), orchestrations(), options()).await; + let client = Client::new(store.clone()); + client.start_orchestration("limited", "Run", "Limited").await.unwrap(); + tokio::time::timeout(Duration::from_secs(5), async { + for _ in 0..20 { + deferrals.recv().await.unwrap(); + } + }) + .await + .expect("admission must keep deferring beyond the poison attempt limit"); + + client.start_orchestration("free", "Run", "Free").await.unwrap(); + assert!(matches!( + client.wait_for_orchestration("free", Duration::from_secs(5)).await.unwrap(), + OrchestrationStatus::Completed { output, .. } if output == "payload" + )); + runtime.shutdown(None).await; + + let history = store.read("limited").await.unwrap(); + assert_eq!( + history + .iter() + .filter(|event| matches!(event.kind, EventKind::ActivityScheduled { .. })) + .count(), + 1 + ); + assert!(!history.iter().any(|event| matches!( + event.kind, + EventKind::ActivityCompleted { .. } | EventKind::ActivityFailed { .. } + ))); + + drop(held); + let runtime = Runtime::start_with_options(store.clone(), activities, orchestrations(), options()).await; + assert!(matches!( + client.wait_for_orchestration("limited", Duration::from_secs(5)).await.unwrap(), + OrchestrationStatus::Completed { output, .. } if output == "payload" + )); + runtime.shutdown(None).await; + let history = store.read("limited").await.unwrap(); + assert_eq!( + history + .iter() + .filter(|event| matches!(event.kind, EventKind::ActivityCompleted { .. })) + .count(), + 1 + ); +} + +#[tokio::test(flavor = "current_thread")] +async fn admission_guard_released_on_success_error_and_panic() { + let (store, _dir) = common::create_sqlite_store_disk().await; + let capacity = Arc::new(Semaphore::new(1)); + let admission_capacity = capacity.clone(); + let activities = ActivityRegistry::builder() + .register_with_admission( + "Limited", + move |_ctx, _input| { + admission_capacity + .clone() + .try_acquire_owned() + .map(ActivityAdmission::admit) + .map_err(|error| format!("unexpected retained reservation: {error}")) + }, + |ctx, _input| async move { + match ctx.instance_id() { + "success" => Ok("done".to_string()), + "error" => Err("expected error".to_string()), + "panic" => panic!("expected handler panic"), + other => panic!("unexpected test instance: {other}"), + } + }, + ) + .build(); + let runtime = Runtime::start_with_options(store.clone(), activities, orchestrations(), options()).await; + let client = Client::new(store); + for instance in ["success", "error", "panic"] { + client.start_orchestration(instance, "Run", "Limited").await.unwrap(); + let status = client + .wait_for_orchestration(instance, Duration::from_secs(5)) + .await + .unwrap(); + match (instance, status) { + ("success", OrchestrationStatus::Completed { output, .. }) => assert_eq!(output, "done"), + ("error", OrchestrationStatus::Failed { details, .. }) => { + assert!(details.display_message().contains("expected error")); + } + ("panic", OrchestrationStatus::Failed { details, .. }) => { + assert!(details.display_message().contains("expected handler panic")); + } + (instance, status) => panic!("{instance}: unexpected status {status:?}"), + } + assert_eq!(capacity.available_permits(), 1, "{instance} leaked its admission guard"); + } + runtime.shutdown(None).await; +} + +#[tokio::test(flavor = "current_thread")] +async fn rejected_panicking_and_invalid_admission_never_invokes_handler() { + let (store, _dir) = common::create_sqlite_store_disk().await; + let activities = ActivityRegistry::builder() + .register_with_admission( + "Limited", + |ctx, _input| match ctx.instance_id() { + "rejected" => Err("invalid admission configuration".to_string()), + "panic" => panic!("expected admission panic"), + "zero-delay" => Ok(ActivityAdmission::Defer(Duration::ZERO)), + other => panic!("unexpected test instance: {other}"), + }, + |_ctx, _input| async move { panic!("rejected handler must not execute") }, + ) + .build(); + let runtime = Runtime::start_with_options(store.clone(), activities, orchestrations(), options()).await; + let client = Client::new(store); + for (instance, message) in [ + ("rejected", "invalid admission configuration"), + ("panic", "expected admission panic"), + ("zero-delay", "positive retry delay"), + ] { + client.start_orchestration(instance, "Run", "Limited").await.unwrap(); + let status = client + .wait_for_orchestration(instance, Duration::from_secs(5)) + .await + .unwrap(); + assert!( + matches!(status, OrchestrationStatus::Failed { ref details, .. } if details.display_message().contains(message)), + "unexpected status: {status:?}" + ); + } + runtime.shutdown(None).await; +} + +#[tokio::test(flavor = "current_thread")] +async fn admission_guard_released_when_cancelled_handler_is_aborted() { + let (store, _dir) = common::create_sqlite_store_disk().await; + let capacity = Arc::new(Semaphore::new(1)); + let admission_capacity = capacity.clone(); + let (started, mut starts) = mpsc::unbounded_channel(); + let activities = ActivityRegistry::builder() + .register_with_admission( + "Limited", + move |_ctx, _input| { + admission_capacity + .clone() + .try_acquire_owned() + .map(ActivityAdmission::admit) + .map_err(|error| error.to_string()) + }, + move |_ctx, _input| { + let started = started.clone(); + async move { + started.send(()).unwrap(); + std::future::pending().await + } + }, + ) + .build(); + let runtime = Runtime::start_with_options(store.clone(), activities, orchestrations(), options()).await; + let client = Client::new(store); + client.start_orchestration("cancel", "Run", "Limited").await.unwrap(); + tokio::time::timeout(Duration::from_secs(5), starts.recv()) + .await + .expect("handler must start") + .unwrap(); + assert_eq!(capacity.available_permits(), 0); + client + .cancel_instance("cancel", "test admission cleanup") + .await + .unwrap(); + let permit = tokio::time::timeout(Duration::from_secs(5), capacity.acquire()) + .await + .expect("aborting the handler must release its admission guard") + .unwrap(); + drop(permit); + runtime.shutdown(None).await; +} From de0cafa4c074ef05ebd60d82b4c2e297552eebfb Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Thu, 29 Oct 2026 15:03:49 +0000 Subject: [PATCH 02/11] Address review feedback --- docs/ORCHESTRATION-GUIDE.md | 7 +++-- src/runtime/dispatchers/worker.rs | 5 +++- src/runtime/mod.rs | 1 + tests/activity_admission_tests.rs | 50 +++++++++++++++++++++++++++++-- tests/common/fault_injection.rs | 11 +++++++ 5 files changed, 68 insertions(+), 6 deletions(-) diff --git a/docs/ORCHESTRATION-GUIDE.md b/docs/ORCHESTRATION-GUIDE.md index efc8e57..0518dae 100644 --- a/docs/ORCHESTRATION-GUIDE.md +++ b/docs/ORCHESTRATION-GUIDE.md @@ -275,9 +275,10 @@ task is dropped, including error, panic, and cancellation. If several limits apply, reserve all of them atomically or defer holding none. Detached work is not protected after the handler finishes. -`Defer(delay)` requires a positive delay. It leaves the original activity -pending in durable storage, frees the worker to fetch other work, and does not -record an activity result or consume the poison-message fetch-attempt budget. +`Defer(delay)` requires a positive delay. Positive sub-millisecond delays are +rounded up to one millisecond. Deferral leaves the original activity pending in +durable storage, frees the worker to fetch other work, and does not record an +activity result or consume the poison-message fetch-attempt budget. It is not a retry of an executed activity. An admission error, panic, or invalid delay fails the activity rather than invoking the handler. Storage errors during deferral are reported as infrastructure errors. diff --git a/src/runtime/dispatchers/worker.rs b/src/runtime/dispatchers/worker.rs index 85d09e4..518695e 100644 --- a/src/runtime/dispatchers/worker.rs +++ b/src/runtime/dispatchers/worker.rs @@ -572,7 +572,10 @@ async fn run_activity_with_cancellation( ActivityAdmission::Defer(delay) if delay.is_zero() => { return Err("Activity admission deferral requires a positive retry delay".to_string()); } - ActivityAdmission::Defer(delay) => return Ok(ActivityInvocation::Deferred(delay)), + ActivityAdmission::Defer(delay) => { + // Keep positive delays nonzero in millisecond-resolution storage. + return Ok(ActivityInvocation::Deferred(delay.max(Duration::from_millis(1)))); + } }; handler .invoke(activity_ctx, input) diff --git a/src/runtime/mod.rs b/src/runtime/mod.rs index c5f1efe..d977503 100644 --- a/src/runtime/mod.rs +++ b/src/runtime/mod.rs @@ -461,6 +461,7 @@ pub enum ActivityAdmission { Admit(Box), /// Leave the activity pending and retry admission after a positive delay. /// + /// Positive sub-millisecond delays are rounded up to one millisecond. /// Deferral releases the worker slot without recording a history event or /// consuming a fetch attempt. It does not retry an executed activity. Defer(Duration), diff --git a/tests/activity_admission_tests.rs b/tests/activity_admission_tests.rs index 580688e..067322a 100644 --- a/tests/activity_admission_tests.rs +++ b/tests/activity_admission_tests.rs @@ -5,10 +5,12 @@ mod common; +use common::fault_injection::PoisonInjectingProvider; +use duroxide::providers::sqlite::SqliteProvider; use duroxide::runtime::registry::ActivityRegistry; use duroxide::runtime::{ActivityAdmission, Runtime, RuntimeOptions}; use duroxide::{ActivityContext, Client, EventKind, OrchestrationContext, OrchestrationRegistry, OrchestrationStatus}; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use std::time::Duration; use tokio::sync::{Semaphore, mpsc}; @@ -90,7 +92,12 @@ async fn deferred_admission_frees_worker_without_consuming_attempts_and_survives ))); drop(held); - let runtime = Runtime::start_with_options(store.clone(), activities, orchestrations(), options()).await; + // Shutdown can cancel a fetch after its attempt has already been committed. + let restart_options = RuntimeOptions { + max_attempts: 2, + ..options() + }; + let runtime = Runtime::start_with_options(store.clone(), activities, orchestrations(), restart_options).await; assert!(matches!( client.wait_for_orchestration("limited", Duration::from_secs(5)).await.unwrap(), OrchestrationStatus::Completed { output, .. } if output == "payload" @@ -106,6 +113,45 @@ async fn deferred_admission_frees_worker_without_consuming_attempts_and_survives ); } +#[tokio::test(flavor = "current_thread")] +async fn admission_deferral_delays_are_clamped_before_reaching_storage() { + let store = Arc::new(PoisonInjectingProvider::new(Arc::new( + SqliteProvider::new_in_memory().await.unwrap(), + ))); + let cases = [ + (Duration::from_nanos(1), Duration::from_millis(1)), + (Duration::from_micros(500), Duration::from_millis(1)), + (Duration::from_nanos(999_999), Duration::from_millis(1)), + (Duration::from_millis(1), Duration::from_millis(1)), + (Duration::from_micros(1500), Duration::from_micros(1500)), + (Duration::from_millis(10), Duration::from_millis(10)), + ]; + let delays = Mutex::new(cases.into_iter()); + let activities = ActivityRegistry::builder() + .register_with_admission( + "Limited", + move |_ctx, _input| { + Ok(match delays.lock().unwrap().next() { + Some((delay, _)) => ActivityAdmission::Defer(delay), + None => ActivityAdmission::admit(()), + }) + }, + |_ctx, input| async move { Ok(input) }, + ) + .build(); + let runtime = Runtime::start_with_options(store.clone(), activities, orchestrations(), options()).await; + let client = Client::new(store.clone()); + client.start_orchestration("clamped", "Run", "Limited").await.unwrap(); + let status = client.wait_for_orchestration("clamped", Duration::from_secs(5)).await; + runtime.shutdown(None).await; + + assert!(matches!( + status.unwrap(), + OrchestrationStatus::Completed { output, .. } if output == "payload" + )); + assert_eq!(store.work_abandons(), cases.map(|(_, expected)| (Some(expected), true))); +} + #[tokio::test(flavor = "current_thread")] async fn admission_guard_released_on_success_error_and_panic() { let (store, _dir) = common::create_sqlite_store_disk().await; diff --git a/tests/common/fault_injection.rs b/tests/common/fault_injection.rs index 683b1d6..6ff5c4c 100644 --- a/tests/common/fault_injection.rs +++ b/tests/common/fault_injection.rs @@ -48,6 +48,7 @@ pub struct PoisonInjectingProvider { successful_ignored_abandons: AtomicU32, counted_abandon_delays: std::sync::Mutex>, orchestration_fetches: std::sync::Mutex>, + work_abandons: std::sync::Mutex, bool)>>, } impl PoisonInjectingProvider { @@ -68,6 +69,7 @@ impl PoisonInjectingProvider { successful_ignored_abandons: AtomicU32::new(0), counted_abandon_delays: std::sync::Mutex::new(Vec::new()), orchestration_fetches: std::sync::Mutex::new(Vec::new()), + work_abandons: std::sync::Mutex::new(Vec::new()), } } @@ -169,6 +171,11 @@ impl PoisonInjectingProvider { pub fn counted_abandon_delays(&self) -> Vec { self.counted_abandon_delays.lock().expect("delays lock").clone() } + + /// (delay, ignore_attempt) for each work item abandonment. + pub fn work_abandons(&self) -> Vec<(Option, bool)> { + self.work_abandons.lock().expect("work abandons lock").clone() + } } #[async_trait] @@ -341,6 +348,10 @@ impl Provider for PoisonInjectingProvider { delay: Option, ignore_attempt: bool, ) -> Result<(), ProviderError> { + self.work_abandons + .lock() + .expect("work abandons lock") + .push((delay, ignore_attempt)); self.inner.abandon_work_item(token, delay, ignore_attempt).await } From 8d1d4f6a5d4d895af174634c8ed825f4098bb2fc Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Thu, 29 Oct 2026 15:43:13 +0000 Subject: [PATCH 03/11] Timestamp fixes --- docs/sqlite-provider-design.md | 5 +++ src/providers/sqlite.rs | 59 ++++++++++++++++++++++++++++++---- 2 files changed, 58 insertions(+), 6 deletions(-) diff --git a/docs/sqlite-provider-design.md b/docs/sqlite-provider-design.md index 008f5c2..2f0f59a 100644 --- a/docs/sqlite-provider-design.md +++ b/docs/sqlite-provider-design.md @@ -101,6 +101,11 @@ CREATE TABLE sessions ( 5. **Session table as a flow table** — routes session-bound activities to owning workers (analogous to network flow affinity) 5. **JSON storage** for flexible event/work item serialization +Delayed visibility is stored as signed 64-bit Unix timestamps in milliseconds. +If a delay would exceed that range, the visibility timestamp saturates at +`i64::MAX`. Oversized delays therefore remain in the future rather than wrapping +around and becoming immediately eligible for delivery. + ## Implementation Plan ### Phase 1: Core Provider Structure diff --git a/src/providers/sqlite.rs b/src/providers/sqlite.rs index 4240f38..c212950 100644 --- a/src/providers/sqlite.rs +++ b/src/providers/sqlite.rs @@ -98,8 +98,7 @@ impl SqliteProvider { // Calculate visible_at based on delay let visible_at = if let Some(delay) = delay { - let delay_ms = delay.as_millis().min(i64::MAX as u128) as i64; - Self::now_millis().saturating_add(delay_ms) + Self::timestamp_after(delay) } else { Self::now_millis() }; @@ -505,9 +504,10 @@ impl SqliteProvider { .as_millis() as i64 } - /// Get future timestamp in milliseconds + /// Get a future timestamp in milliseconds, saturating at `i64::MAX`. fn timestamp_after(duration: Duration) -> i64 { - Self::now_millis() + duration.as_millis() as i64 + let delay_ms = duration.as_millis().min(i64::MAX as u128) as i64; + Self::now_millis().saturating_add(delay_ms) } /// Build a SQL WHERE clause fragment for tag filtering. @@ -2242,8 +2242,7 @@ impl Provider for SqliteProvider { // Optionally delay only the messages this fetch locked. Rows that became visible // after the fetch (for example a continue-as-new start) keep their visibility. if let Some(delay) = delay { - let delay_ms = delay.as_millis().min(i64::MAX as u128) as i64; - let visible_at = Self::now_millis().saturating_add(delay_ms); + let visible_at = Self::timestamp_after(delay); sqlx::query("UPDATE orchestrator_queue SET visible_at = ? WHERE lock_token = ?") .bind(visible_at) .bind(lock_token) @@ -4347,6 +4346,54 @@ mod tests { assert_ne!(lock_token2, lock_token); // Different lock token } + #[tokio::test] + async fn test_abandon_work_item_with_oversized_delay() { + for delay in [ + Duration::MAX, + Duration::from_millis(i64::MAX as u64), + Duration::from_millis(i64::MAX as u64 + 1), + ] { + let store = create_test_store().await; + for instance in ["deferred", "ready"] { + store + .enqueue_for_worker(WorkItem::ActivityExecute { + instance: instance.to_string(), + execution_id: 1, + id: 1, + name: "TestActivity".to_string(), + input: String::new(), + session_id: None, + tag: None, + }) + .await + .unwrap(); + } + + let (item, lock_token, _) = store + .fetch_work_item(Duration::from_secs(30), Duration::ZERO, None, &TagFilter::default()) + .await + .unwrap() + .unwrap(); + assert!(matches!(item, WorkItem::ActivityExecute { instance, .. } if instance == "deferred")); + + store.abandon_work_item(&lock_token, Some(delay), true).await.unwrap(); + + let deferred_state: (i64, i64, Option) = + sqlx::query_as("SELECT visible_at, attempt_count, lock_token FROM worker_queue ORDER BY id LIMIT 1") + .fetch_one(&store.pool) + .await + .unwrap(); + assert_eq!(deferred_state, (i64::MAX, 0, None), "{delay:?}"); + + let (item, _, _) = store + .fetch_work_item(Duration::from_secs(30), Duration::ZERO, None, &TagFilter::default()) + .await + .unwrap() + .unwrap(); + assert!(matches!(item, WorkItem::ActivityExecute { instance, .. } if instance == "ready")); + } + } + #[tokio::test] async fn test_abandon_work_item_invalid_token() { let store = create_test_store().await; From 86c496955c96ce9d0e115c3681f95fb6c813ffdc Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Fri, 30 Oct 2026 12:13:51 +0000 Subject: [PATCH 04/11] Fixes from review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Deferred work now yields to already-ready work in both SQLite and PostgreSQL. - Activity metadata avoids full-history reads for current executions and remains execution-specific. - Admission delays are clamped to 1 ms–24 hours. - Failed deferrals retry with the same delay and attempt-restoration flag. - Corrected start logging, added a committed-deferral metric, and updated documentation/changelogs. - Added Debug and #[non_exhaustive]. Typed registration remains deferred as agreed. --- CHANGELOG.md | 24 ++ docs/ORCHESTRATION-GUIDE.md | 26 +- docs/observability-guide.md | 13 +- docs/provider-implementation-guide.md | 15 +- docs/provider-testing-guide.md | 7 +- docs/sqlite-provider-design.md | 13 +- src/provider_validation/queue_semantics.rs | 75 +++++- src/provider_validations.rs | 6 +- src/providers/mod.rs | 6 + src/providers/sqlite.rs | 7 +- src/runtime/dispatchers/worker.rs | 113 +++++--- src/runtime/mod.rs | 25 +- src/runtime/observability.rs | 15 ++ tests/activity_admission_tests.rs | 285 ++++++++++++++++++++- tests/common/fault_injection.rs | 37 ++- tests/observability_tests.rs | 7 + tests/sqlite_provider_validations.rs | 5 + 17 files changed, 622 insertions(+), 57 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bf5cd12..8bed747 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,30 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Opt-in activity resource admission through `register_with_admission`. + Admission can retain a guard for the handler's lifetime or durably defer + execution without occupying a worker slot or recording an activity result. + Positive retry delays are clamped to 1 ms–24 hours. Only committed deferrals + restore their fetch attempt; failed or interrupted deferrals may still use + the poison-message budget. Committed deferrals have a separate + `duroxide_activity_admission_deferrals_total` metric. (#74) + +### Fixed + +- SQLite worker selection now orders eligible items by visibility time, then + queue ID, so a deferred backlog cannot repeatedly overtake already-ready work. + Other providers must implement equivalent queue progress semantics; existing + stored histories and work-item formats are unchanged. SQLite workers must all + be upgraded to provide this behavior in a mixed-version deployment. (#74) +- Activity metadata uses the lightweight instance lookup for the current + execution instead of re-reading its full history on every attempt. Activities + from earlier executions read their own execution's metadata. Failed deferral + retries preserve the requested delay and attempt-restoration flag. (#74) +- Relative SQLite queue delays saturate instead of overflowing their timestamp + representation. (#74) + ## [0.1.32] - 2026-10-11 **Release:** diff --git a/docs/ORCHESTRATION-GUIDE.md b/docs/ORCHESTRATION-GUIDE.md index 0518dae..36e89e6 100644 --- a/docs/ORCHESTRATION-GUIDE.md +++ b/docs/ORCHESTRATION-GUIDE.md @@ -275,19 +275,29 @@ task is dropped, including error, panic, and cancellation. If several limits apply, reserve all of them atomically or defer holding none. Detached work is not protected after the handler finishes. -`Defer(delay)` requires a positive delay. Positive sub-millisecond delays are -rounded up to one millisecond. Deferral leaves the original activity pending in -durable storage, frees the worker to fetch other work, and does not record an -activity result or consume the poison-message fetch-attempt budget. -It is not a retry of an executed activity. An admission error, panic, or invalid -delay fails the activity rather than invoking the handler. Storage errors during -deferral are reported as infrastructure errors. +`Defer(delay)` requires a positive delay. The runtime clamps it to 1 ms through +24 hours (`ActivityAdmission::MAX_DEFER_DELAY`) before passing it to any provider. +Deferral leaves the original activity pending in durable storage, frees the +worker to fetch other work, and does not record an activity result. +A **committed** deferral restores its fetch attempt. Crashes, shutdown aborts, +and failed abandons can still consume the poison-message attempt budget, so +leave room for recovery attempts. Deferral is not a retry of an executed activity. + +An admission error, panic, or zero delay fails the activity rather than invoking +the handler and is reported as an application error. Storage errors during +deferral are infrastructure errors; an immediate fallback abandon preserves the +same delay and attempt-restoration flag. If that also fails, recovery waits for +the work lock to expire. Successfully committed deferrals increment +`duroxide_activity_admission_deferrals_total`, not the activity execution counter. +The "Activity started" log is emitted only after admission succeeds. Admission runs outside orchestration replay; completed results are unaffected. Each recovered execution attempt is admitted again. Resource counters belong to the application and are not automatically coordinated across runtimes. Keep total worker capacity above the admitted activity ceiling if other work -needs headroom. Admission does not guarantee fair scheduling between workflows. +needs headroom. Providers must allow already-ready work to progress past deferred +items, including after a deferral's delay expires. This is not a guarantee of +equal throughput or admission fairness between workflows. ### The Replay Model diff --git a/docs/observability-guide.md b/docs/observability-guide.md index f371657..20807b6 100644 --- a/docs/observability-guide.md +++ b/docs/observability-guide.md @@ -195,6 +195,7 @@ All metrics follow Prometheus naming conventions with `_total` for counters and | `duroxide_active_orchestrations` | Gauge | Orchestration | _(none)_ | **Current** active count | | **Activity Execution** | | `duroxide_activity_executions_total` | Counter | Activity | activity_name, outcome, retry_attempt | Track success/failure rates | +| `duroxide_activity_admission_deferrals_total` | Counter | Activity | activity_name, tag | Track committed resource-admission deferrals | | `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 | | **Sub-Orchestration** | @@ -265,6 +266,17 @@ All metrics follow Prometheus naming conventions with `_total` for counters and #### Activity Execution Metrics +**`duroxide_activity_admission_deferrals_total`** (Counter) +- **Description**: Admission deferrals successfully committed to storage, including a successful fallback after an abandon error +- **Labels**: `activity_name`, `tag` +- **Use**: Identify resource-constrained activities without counting deferrals as executions + +The "Activity started" log appears only after admission succeeds. Admission +rejections, panics, and zero retry delays count as `app_error`; storage failures +while deferring count as `infra_error`, even if the fallback abandon succeeds. +Uncommitted deferrals do not increment the deferral counter and may consume a +fetch attempt on recovery. + **`duroxide_activity_executions_total`** (Counter) - **Description**: Activity execution attempts - **Labels**: `activity_name`, `outcome` (success|app_error|infra_error|config_error), `retry_attempt` (0|1|2|3+) @@ -645,4 +657,3 @@ Import this dashboard into Grafana to get: - [Library Observability Guide](library-observability.md) - For library developers - [examples/with_observability.rs](../examples/with_observability.rs) - Working example - [examples/grafana-dashboard.json](../examples/grafana-dashboard.json) - Production dashboard - diff --git a/docs/provider-implementation-guide.md b/docs/provider-implementation-guide.md index ab39b4c..944b12d 100644 --- a/docs/provider-implementation-guide.md +++ b/docs/provider-implementation-guide.md @@ -1393,7 +1393,7 @@ async fn fetch_work_item( SELECT id FROM worker_queue WHERE visible_at <= ?3 AND (lock_token IS NULL OR locked_until <= ?3) - ORDER BY id + ORDER BY visible_at, id LIMIT 1 ) RETURNING id, work_item, attempt_count, instance_id, execution_id @@ -1442,6 +1442,19 @@ async fn get_execution_state(&self, instance_id: &str, execution_id: Option - `attempts`: How many times this item has been fetched (for poison detection) - `state`: Execution state (Running or Terminated for cancellation) +**Worker queue progress:** Repeatedly deferred activities must not overtake work +that was already ready, even after their delay expires. Order eligible rows by +`visible_at, id` (with a matching index), or use an equivalent requeue policy. +Apply the same ordering to session-aware and tag-filtered fetches. Admission +deferral restores an attempt only when `abandon_work_item(..., true)` commits; +failed or interrupted abandons can still consume attempts. + +The worker uses `ProviderAdmin::get_instance_info` for lightweight activity +metadata when it describes the activity's execution. Providers without that +capability, and activities from an earlier execution, fall back to +`read_with_execution`. Metadata must describe the requested execution, not a +newer continue-as-new execution. + ### ack_work_item() — Complete Implementation Delete from worker queue and optionally enqueue completion: diff --git a/docs/provider-testing-guide.md b/docs/provider-testing-guide.md index 28dd3f4..0e5ff2a 100644 --- a/docs/provider-testing-guide.md +++ b/docs/provider-testing-guide.md @@ -509,7 +509,7 @@ async fn test_my_provider_worker_queue_fifo_ordering() { ### What the Tests Validate -The validation test suite includes **169 individual test functions** organized into 18 categories: +The validation test suite includes the following categories: 1. **Atomicity Tests (4 tests)** - `test_atomicity_failure_rollback` - All-or-nothing commit semantics, rollback on failure @@ -559,8 +559,9 @@ The validation test suite includes **169 individual test functions** organized i - `test_continue_as_new_creates_new_execution` - ContinueAsNew creates new execution - `test_execution_history_persistence` - All executions' history persists independently -6. **Queue Semantics Tests (8 tests)** +6. **Queue Semantics Tests (9 tests)** - `test_worker_queue_fifo_ordering` - Worker items dequeued in FIFO order + - `test_worker_deferred_item_does_not_starve_ready_work` - Expired deferrals yield to already-ready work for session and non-session fetches, retaining attempt restoration - `test_worker_peek_lock_semantics` - Dequeue doesn't remove item until ack - `test_worker_ack_atomicity` - Ack_worker atomically removes item and enqueues completion - `test_timer_delayed_visibility` - TimerFired items only dequeued when visible @@ -1095,7 +1096,7 @@ This generates `stress-test-results.md` with: - **Test Implementation**: `src/provider_validation/` (individual test modules) - **Test API**: `src/provider_validations.rs` (test function exports) -- **Example Usage**: `tests/sqlite_provider_validations.rs` (complete example with all 202 tests) +- **Example Usage**: `tests/sqlite_provider_validations.rs` (complete provider validation suite) - **Test Specification**: See individual test function documentation - **Provider Guide**: `docs/provider-implementation-guide.md` - **Built-in Providers**: `src/providers/sqlite.rs` diff --git a/docs/sqlite-provider-design.md b/docs/sqlite-provider-design.md index 2f0f59a..f185057 100644 --- a/docs/sqlite-provider-design.md +++ b/docs/sqlite-provider-design.md @@ -101,10 +101,15 @@ CREATE TABLE sessions ( 5. **Session table as a flow table** — routes session-bound activities to owning workers (analogous to network flow affinity) 5. **JSON storage** for flexible event/work item serialization -Delayed visibility is stored as signed 64-bit Unix timestamps in milliseconds. -If a delay would exceed that range, the visibility timestamp saturates at -`i64::MAX`. Oversized delays therefore remain in the future rather than wrapping -around and becoming immediately eligible for delivery. +Visibility timestamps use signed 64-bit Unix milliseconds. Relative delays +passed to queue enqueue/abandon operations saturate at `i64::MAX` if their +future timestamp exceeds that range. This does not cover absolute +`TimerFired.fire_at_ms` timestamps. + +Eligible worker items are ordered by `visible_at, id`, supported by the +`idx_worker_ready_order` index. A deferred item goes behind already-ready work, +even if its delay expires before the backlog drains. Session and tag filtering +use the same ordering. ## Implementation Plan diff --git a/src/provider_validation/queue_semantics.rs b/src/provider_validation/queue_semantics.rs index b542f29..a76b8a8 100644 --- a/src/provider_validation/queue_semantics.rs +++ b/src/provider_validation/queue_semantics.rs @@ -3,7 +3,7 @@ use crate::provider_validation::{Event, EventKind, ExecutionMetadata, create_instance, start_item}; use crate::provider_validations::ProviderFactory; -use crate::providers::{TagFilter, WorkItem}; +use crate::providers::{SessionFetchConfig, TagFilter, WorkItem}; use std::time::Duration; /// Test 5.1: Worker Queue FIFO Ordering @@ -117,6 +117,79 @@ pub async fn test_worker_peek_lock_semantics(factory: &F) { tracing::info!("✓ Test passed: worker peek-lock semantics verified"); } +/// Deferred work must not overtake work that was already ready, even after its delay expires. +pub async fn test_worker_deferred_item_does_not_starve_ready_work(factory: &F) { + for session in [ + None, + Some(SessionFetchConfig { + owner_id: "progress-worker".to_string(), + lock_timeout: factory.lock_timeout(), + }), + ] { + let provider = factory.create_provider().await; + for id in [1, 2] { + provider + .enqueue_for_worker(WorkItem::ActivityExecute { + instance: "deferred-progress".to_string(), + execution_id: 1, + id, + name: "Activity".to_string(), + input: String::new(), + session_id: session.as_ref().map(|_| "progress-session".to_string()), + tag: None, + }) + .await + .unwrap(); + } + let (item, token, _) = provider + .fetch_work_item( + factory.lock_timeout(), + Duration::ZERO, + session.as_ref(), + &TagFilter::Any, + ) + .await + .unwrap() + .unwrap(); + assert!(matches!(item, WorkItem::ActivityExecute { id: 1, .. })); + provider + .abandon_work_item(&token, Some(Duration::from_millis(1)), true) + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(20)).await; + + let (item, token, _) = provider + .fetch_work_item( + factory.lock_timeout(), + Duration::ZERO, + session.as_ref(), + &TagFilter::Any, + ) + .await + .unwrap() + .unwrap(); + assert!( + matches!(item, WorkItem::ActivityExecute { id: 2, .. }), + "an expired deferral must not jump ahead of already-ready work" + ); + provider.ack_work_item(&token, None).await.unwrap(); + + let (item, token, attempts) = provider + .fetch_work_item( + factory.lock_timeout(), + Duration::ZERO, + session.as_ref(), + &TagFilter::Any, + ) + .await + .unwrap() + .unwrap(); + assert!(matches!(item, WorkItem::ActivityExecute { id: 1, .. })); + assert_eq!(attempts, 1, "the committed deferral must restore its attempt"); + provider.ack_work_item(&token, None).await.unwrap(); + } +} + /// Test 5.3: Worker Ack Atomicity /// Goal: Verify ack_work_item atomically removes item and enqueues completion. pub async fn test_worker_ack_atomicity(factory: &F) { diff --git a/src/provider_validations.rs b/src/provider_validations.rs index d937e66..8f594e4 100644 --- a/src/provider_validations.rs +++ b/src/provider_validations.rs @@ -111,6 +111,7 @@ pub use crate::provider_validation::race_replay; /// - `test_lost_lock_token_handling` - Verify locked items become available after expiration /// - `test_worker_item_immediate_visibility` - Verify newly enqueued items are immediately visible /// - `test_worker_delayed_visibility_skips_future_items` - Verify items with future visible_at are skipped +/// - `test_worker_deferred_item_does_not_starve_ready_work` - Verify expired deferrals yield to already-ready work /// - `test_orphan_queue_messages_dropped` - Verify QueueMessage for non-existent instance is dropped; kept for existing instance /// /// **Management Tests:** @@ -178,8 +179,9 @@ pub use crate::provider_validation::multi_execution::{ #[cfg(feature = "provider-test")] pub use crate::provider_validation::queue_semantics::{ test_lost_lock_token_handling, test_orphan_queue_messages_dropped, test_timer_delayed_visibility, - test_worker_ack_atomicity, test_worker_delayed_visibility_skips_future_items, - test_worker_item_immediate_visibility, test_worker_peek_lock_semantics, test_worker_queue_fifo_ordering, + test_worker_ack_atomicity, test_worker_deferred_item_does_not_starve_ready_work, + test_worker_delayed_visibility_skips_future_items, test_worker_item_immediate_visibility, + test_worker_peek_lock_semantics, test_worker_queue_fifo_ordering, }; #[cfg(feature = "provider-test")] diff --git a/src/providers/mod.rs b/src/providers/mod.rs index 813631b..ce74636 100644 --- a/src/providers/mod.rs +++ b/src/providers/mod.rs @@ -1727,6 +1727,12 @@ pub trait Provider: Any + Send + Sync { /// 2. Lock it with a unique token /// 3. Return item + token (item stays in queue until ack) /// + /// # Ordering + /// + /// Deferred items must not repeatedly overtake already-ready work after their + /// visibility delay expires. Order eligible items by `visible_at`, then queue ID, + /// or use an equivalent requeue policy. This also applies to session and tagged work. + /// /// # Parameters /// /// - `lock_timeout`: Duration to lock the item for processing. diff --git a/src/providers/sqlite.rs b/src/providers/sqlite.rs index c212950..05c20cd 100644 --- a/src/providers/sqlite.rs +++ b/src/providers/sqlite.rs @@ -425,6 +425,9 @@ impl SqliteProvider { sqlx::query("CREATE INDEX IF NOT EXISTS idx_worker_available ON worker_queue(lock_token, id)") .execute(pool) .await?; + sqlx::query("CREATE INDEX IF NOT EXISTS idx_worker_ready_order ON worker_queue(visible_at, id)") + .execute(pool) + .await?; sqlx::query( "CREATE INDEX IF NOT EXISTS idx_worker_identity ON worker_queue(instance_id, execution_id, activity_id)", ) @@ -1843,7 +1846,7 @@ impl Provider for SqliteProvider { OR s.session_id IS NULL ) AND ({tag_clause}) - ORDER BY q.id + ORDER BY q.visible_at, q.id LIMIT 1 "#, ); @@ -1864,7 +1867,7 @@ impl Provider for SqliteProvider { AND (q.lock_token IS NULL OR q.locked_until <= ?1) AND q.session_id IS NULL AND ({tag_clause}) - ORDER BY q.id + ORDER BY q.visible_at, q.id LIMIT 1 "#, ); diff --git a/src/runtime/dispatchers/worker.rs b/src/runtime/dispatchers/worker.rs index 518695e..e0a7ce8 100644 --- a/src/runtime/dispatchers/worker.rs +++ b/src/runtime/dispatchers/worker.rs @@ -124,7 +124,9 @@ enum ActivityOutcome { /// Orchestration was cancelled, activity result dropped Cancelled, /// Admission deferred without executing the handler - Deferred, + Deferred(Duration), + /// Activity setup failed before invocation + InfraError, } enum ActivityInvocation { @@ -489,18 +491,13 @@ async fn execute_activity(rt: &Arc, activities: &Arc activity_ctx, + Err(error) => { + handle_activity_outcome(rt, &ctx, Err(error), ActivityOutcome::InfraError).await; + return; + } + }; let start_time = std::time::Instant::now(); @@ -532,13 +529,53 @@ async fn build_activity_context( rt: &Arc, ctx: &ActivityWorkContext, cancellation_token: CancellationToken, -) -> crate::ActivityContext { - let descriptor = rt.get_orchestration_descriptor(&ctx.instance).await; - let (orch_name, orch_version) = descriptor - .map(|d| (d.name, d.version)) - .unwrap_or_else(|| ("unknown".to_string(), "unknown".to_string())); +) -> Result { + let metadata = if let Some(admin) = rt.history_store.as_management_capability() { + match admin.get_instance_info(&ctx.instance).await { + Ok(info) if info.current_execution_id == ctx.execution_id && info.orchestration_version != "unknown" => { + Some((info.orchestration_name, info.orchestration_version)) + } + Ok(_) => None, + Err(error) => { + warn!( + instance = %ctx.instance, + execution_id = ctx.execution_id, + %error, + "Failed to read activity metadata; falling back to execution history" + ); + None + } + } + } else { + None + }; + let (orch_name, orch_version) = match metadata { + Some(metadata) => metadata, + None => { + let history = rt + .history_store + .read_with_execution(&ctx.instance, ctx.execution_id) + .await?; + history + .iter() + .find_map(|event| match &event.kind { + crate::EventKind::OrchestrationStarted { name, version, .. } => { + Some((name.clone(), version.clone())) + } + _ => None, + }) + .unwrap_or_else(|| { + warn!( + instance = %ctx.instance, + execution_id = ctx.execution_id, + "Orchestration metadata missing for activity" + ); + ("unknown".to_string(), "unknown".to_string()) + }) + } + }; - crate::ActivityContext::new_with_cancellation( + Ok(crate::ActivityContext::new_with_cancellation( ctx.instance.clone(), ctx.execution_id, orch_name, @@ -550,7 +587,7 @@ async fn build_activity_context( ctx.tag.clone(), cancellation_token, Arc::clone(&rt.history_store), - ) + )) } /// Run an activity with cancellation support using `tokio::select!`. @@ -564,6 +601,7 @@ async fn run_activity_with_cancellation( start_time: std::time::Instant, ) -> (Result<(), crate::providers::ProviderError>, 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 // aborted mid-activity during shutdown. let mut activity_handle = AbortOnDropHandle::new(rt.task_tracker.spawn(async move { @@ -573,10 +611,21 @@ async fn run_activity_with_cancellation( return Err("Activity admission deferral requires a positive retry delay".to_string()); } ActivityAdmission::Defer(delay) => { - // Keep positive delays nonzero in millisecond-resolution storage. - return Ok(ActivityInvocation::Deferred(delay.max(Duration::from_millis(1)))); + return Ok(ActivityInvocation::Deferred( + delay.clamp(Duration::from_millis(1), ActivityAdmission::MAX_DEFER_DELAY), + )); } }; + tracing::debug!( + target: "duroxide::runtime", + instance_id = %activity_ctx.instance_id(), + execution_id = activity_ctx.execution_id(), + activity_name = %activity_ctx.activity_name(), + activity_id, + worker_id = %activity_ctx.worker_id(), + activity_tag = ?activity_ctx.tag(), + "Activity started" + ); handler .invoke(activity_ctx, input) .await @@ -602,7 +651,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) + (result, 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, @@ -795,7 +844,11 @@ async fn handle_activity_outcome( Ok(()) => match outcome { ActivityOutcome::Success => rt.record_activity_success(), ActivityOutcome::AppError => rt.record_activity_app_error(), - ActivityOutcome::Cancelled | ActivityOutcome::Deferred => {} + ActivityOutcome::Deferred(_) => { + rt.record_activity_admission_deferral(&ctx.activity_name, ctx.tag.as_deref()) + } + ActivityOutcome::InfraError => rt.record_activity_infra_error(), + ActivityOutcome::Cancelled => {} }, Err(e) => { warn!( @@ -807,13 +860,13 @@ async fn handle_activity_outcome( outcome = ?outcome, "worker: activity acknowledgment or deferral failed, abandoning work item" ); + let (delay, ignore_attempt) = match outcome { + ActivityOutcome::Deferred(delay) => (delay, true), + _ => (Duration::from_millis(100), false), + }; if let Err(error) = rt .history_store - .abandon_work_item( - &ctx.lock_token, - Some(Duration::from_millis(100)), - matches!(outcome, ActivityOutcome::Deferred), - ) + .abandon_work_item(&ctx.lock_token, Some(delay), ignore_attempt) .await { warn!( @@ -822,6 +875,8 @@ async fn handle_activity_outcome( error = %error, "Failed to abandon activity work item; waiting for lock expiry" ); + } else if matches!(outcome, ActivityOutcome::Deferred(_)) { + rt.record_activity_admission_deferral(&ctx.activity_name, ctx.tag.as_deref()); } rt.record_activity_infra_error(); } diff --git a/src/runtime/mod.rs b/src/runtime/mod.rs index d977503..75a1965 100644 --- a/src/runtime/mod.rs +++ b/src/runtime/mod.rs @@ -456,18 +456,32 @@ where } /// A nonblocking admission decision made before invoking an activity. +#[non_exhaustive] pub enum ActivityAdmission { /// The guard is held until the activity handler returns or its task is dropped. Admit(Box), /// Leave the activity pending and retry admission after a positive delay. /// - /// Positive sub-millisecond delays are rounded up to one millisecond. + /// Positive delays are clamped to one millisecond through [`Self::MAX_DEFER_DELAY`]. /// Deferral releases the worker slot without recording a history event or - /// consuming a fetch attempt. It does not retry an executed activity. + /// consuming a fetch attempt once the deferral commits. Interrupted or failed + /// deferrals can still consume attempts. It does not retry an executed activity. Defer(Duration), } +impl std::fmt::Debug for ActivityAdmission { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Admit(_) => f.debug_tuple("Admit").finish_non_exhaustive(), + Self::Defer(delay) => f.debug_tuple("Defer").field(delay).finish(), + } + } +} + impl ActivityAdmission { + /// Maximum admission retry delay: 24 hours, independent of the provider. + pub const MAX_DEFER_DELAY: Duration = Duration::from_secs(24 * 60 * 60); + /// Admit an activity while retaining ownership of its resource reservation. pub fn admit(guard: impl Send + 'static) -> Self { Self::Admit(Box::new(guard)) @@ -691,6 +705,13 @@ impl Runtime { } } + #[inline] + fn record_activity_admission_deferral(&self, activity_name: &str, tag: Option<&str>) { + if let Some(provider) = self.metrics_provider() { + provider.record_activity_admission_deferral(activity_name, tag); + } + } + #[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 dcea2f8..0d2d1c8 100644 --- a/src/runtime/observability.rs +++ b/src/runtime/observability.rs @@ -133,6 +133,7 @@ pub struct MetricsSnapshot { pub activity_infra_errors: u64, pub activity_config_errors: u64, pub activity_poison: u64, + pub activity_admission_deferrals: u64, pub orch_dispatcher_items_fetched: u64, pub worker_dispatcher_items_fetched: u64, pub orch_continue_as_new: u64, @@ -159,6 +160,7 @@ pub struct MetricsProvider { activity_infra_errors_atomic: AtomicU64, activity_config_errors_atomic: AtomicU64, activity_poison_atomic: AtomicU64, + activity_admission_deferrals_atomic: AtomicU64, // Dispatcher counters orch_dispatcher_items_fetched_atomic: AtomicU64, @@ -197,6 +199,7 @@ impl MetricsProvider { activity_infra_errors_atomic: AtomicU64::new(0), activity_config_errors_atomic: AtomicU64::new(0), activity_poison_atomic: AtomicU64::new(0), + activity_admission_deferrals_atomic: AtomicU64::new(0), orch_dispatcher_items_fetched_atomic: AtomicU64::new(0), worker_dispatcher_items_fetched_atomic: AtomicU64::new(0), orch_continue_as_new_atomic: AtomicU64::new(0), @@ -458,6 +461,17 @@ impl MetricsProvider { self.activity_config_errors_atomic.fetch_add(1, Ordering::Relaxed); } + #[inline] + pub fn record_activity_admission_deferral(&self, activity_name: &str, tag: Option<&str>) { + counter!( + "duroxide_activity_admission_deferrals_total", + "activity_name" => activity_name.to_string(), + "tag" => tag.unwrap_or("default").to_string(), + ) + .increment(1); + self.activity_admission_deferrals_atomic.fetch_add(1, Ordering::Relaxed); + } + #[inline] pub fn record_orchestration_poison(&self) { self.orch_poison_atomic.fetch_add(1, Ordering::Relaxed); @@ -625,6 +639,7 @@ impl MetricsProvider { activity_infra_errors: self.activity_infra_errors_atomic.load(Ordering::Relaxed), activity_config_errors: self.activity_config_errors_atomic.load(Ordering::Relaxed), activity_poison: self.activity_poison_atomic.load(Ordering::Relaxed), + activity_admission_deferrals: self.activity_admission_deferrals_atomic.load(Ordering::Relaxed), orch_dispatcher_items_fetched: self.orch_dispatcher_items_fetched_atomic.load(Ordering::Relaxed), worker_dispatcher_items_fetched: self.worker_dispatcher_items_fetched_atomic.load(Ordering::Relaxed), orch_continue_as_new: self.orch_continue_as_new_atomic.load(Ordering::Relaxed), diff --git a/tests/activity_admission_tests.rs b/tests/activity_admission_tests.rs index 067322a..d4aeb9e 100644 --- a/tests/activity_admission_tests.rs +++ b/tests/activity_admission_tests.rs @@ -7,9 +7,11 @@ mod common; use common::fault_injection::PoisonInjectingProvider; use duroxide::providers::sqlite::SqliteProvider; +use duroxide::providers::{Provider, WorkItem}; use duroxide::runtime::registry::ActivityRegistry; use duroxide::runtime::{ActivityAdmission, Runtime, RuntimeOptions}; use duroxide::{ActivityContext, Client, EventKind, OrchestrationContext, OrchestrationRegistry, OrchestrationStatus}; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; use tokio::sync::{Semaphore, mpsc}; @@ -76,7 +78,8 @@ async fn deferred_admission_frees_worker_without_consuming_attempts_and_survives client.wait_for_orchestration("free", Duration::from_secs(5)).await.unwrap(), OrchestrationStatus::Completed { output, .. } if output == "payload" )); - runtime.shutdown(None).await; + runtime.clone().shutdown(None).await; + assert!(runtime.metrics_snapshot().unwrap().activity_admission_deferrals >= 20); let history = store.read("limited").await.unwrap(); assert_eq!( @@ -152,6 +155,272 @@ async fn admission_deferral_delays_are_clamped_before_reaching_storage() { assert_eq!(store.work_abandons(), cases.map(|(_, expected)| (Some(expected), true))); } +#[tokio::test(flavor = "current_thread")] +async fn oversized_admission_delays_are_bounded_before_reaching_storage() { + let maximum = Duration::from_secs(24 * 60 * 60); + for delay in [maximum, maximum + Duration::from_nanos(1), Duration::MAX] { + let sqlite = Arc::new(SqliteProvider::new_in_memory().await.unwrap()); + let store = Arc::new(PoisonInjectingProvider::new(sqlite.clone())); + let (deferred, mut deferrals) = mpsc::unbounded_channel(); + let activities = ActivityRegistry::builder() + .register_with_admission( + "Limited", + move |_ctx, _input| { + deferred.send(()).unwrap(); + Ok(ActivityAdmission::Defer(delay)) + }, + |_ctx, _input| async move { panic!("deferred handler must not run") }, + ) + .build(); + let runtime = Runtime::start_with_options(store.clone(), activities, orchestrations(), options()).await; + Client::new(sqlite) + .start_orchestration("bounded", "Run", "Limited") + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(5), deferrals.recv()) + .await + .unwrap() + .unwrap(); + runtime.shutdown(None).await; + assert_eq!(store.work_abandons(), vec![(Some(maximum), true)], "{delay:?}"); + } +} + +#[tokio::test(flavor = "current_thread")] +async fn failed_deferral_retries_the_same_delay_and_restores_the_attempt() { + let (events, _guard) = common::tracing_capture::install_tracing_capture(); + let sqlite = Arc::new(SqliteProvider::new_in_memory().await.unwrap()); + let store = Arc::new(PoisonInjectingProvider::new(sqlite.clone())); + store.fail_work_abandons(1); + let first_attempt = AtomicBool::new(true); + let delay = Duration::from_millis(25); + let activities = ActivityRegistry::builder() + .register_with_admission( + "Limited", + move |_ctx, _input| { + Ok(if first_attempt.swap(false, Ordering::SeqCst) { + ActivityAdmission::Defer(delay) + } else { + ActivityAdmission::admit(()) + }) + }, + |_ctx, input| async move { Ok(input) }, + ) + .build(); + let runtime = Runtime::start_with_options(store.clone(), activities, orchestrations(), options()).await; + let client = Client::new(sqlite); + client.start_orchestration("retry", "Run", "Limited").await.unwrap(); + let status = client.wait_for_orchestration("retry", Duration::from_secs(5)).await; + runtime.clone().shutdown(None).await; + assert!(matches!( + status.unwrap(), + OrchestrationStatus::Completed { output, .. } if output == "payload" + )); + assert_eq!(store.work_abandons(), vec![(Some(delay), true); 2]); + let metrics = runtime.metrics_snapshot().unwrap(); + assert_eq!(metrics.activity_admission_deferrals, 1); + assert_eq!(metrics.activity_infra_errors, 1); + assert_eq!( + events + .lock() + .unwrap() + .iter() + .filter(|event| event.message == "Activity started") + .count(), + 1, + "only the admitted invocation should be logged as started" + ); +} + +#[tokio::test(flavor = "current_thread")] +async fn uncommitted_deferral_consumes_attempt_and_is_not_counted_as_deferred() { + let (events, _guard) = common::tracing_capture::install_tracing_capture(); + let sqlite = Arc::new(SqliteProvider::new_in_memory().await.unwrap()); + let store = Arc::new(PoisonInjectingProvider::new(sqlite.clone())); + store.fail_work_abandons(2); + let delay = Duration::from_secs(60); + let activities = ActivityRegistry::builder() + .register_with_admission( + "Limited", + move |_ctx, _input| Ok(ActivityAdmission::Defer(delay)), + |_ctx, _input| async move { panic!("deferred handler must not run") }, + ) + .build(); + let runtime = Runtime::start_with_options( + store.clone(), + activities, + orchestrations(), + RuntimeOptions { + worker_lock_timeout: Duration::from_millis(100), + ..options() + }, + ) + .await; + let client = Client::new(sqlite); + client + .start_orchestration("failed-deferral", "Run", "Limited") + .await + .unwrap(); + let status = client + .wait_for_orchestration("failed-deferral", Duration::from_secs(5)) + .await; + runtime.clone().shutdown(None).await; + assert!(matches!( + status.unwrap(), + OrchestrationStatus::Failed { + details: duroxide::ErrorDetails::Poison { attempt_count: 2, .. }, + .. + } + )); + assert_eq!(store.work_abandons(), vec![(Some(delay), true); 2]); + let metrics = runtime.metrics_snapshot().unwrap(); + assert_eq!(metrics.activity_admission_deferrals, 0); + assert_eq!(metrics.activity_infra_errors, 1); + assert_eq!(metrics.activity_app_errors, 0); + assert!( + !events + .lock() + .unwrap() + .iter() + .any(|event| event.message == "Activity started") + ); +} + +#[test] +fn admission_debug_does_not_require_a_debug_guard() { + struct Guard; + assert_eq!(format!("{:?}", ActivityAdmission::admit(Guard)), "Admit(..)"); + assert_eq!( + format!("{:?}", ActivityAdmission::Defer(Duration::from_secs(1))), + "Defer(1s)" + ); + assert_eq!(ActivityAdmission::MAX_DEFER_DELAY, Duration::from_secs(24 * 60 * 60)); +} + +#[tokio::test(flavor = "current_thread")] +async fn deferred_backlog_does_not_starve_ready_work_or_reread_history() { + for (backlog, workers) in [(300, 1), (1000, 4)] { + let dir = tempfile::tempdir().unwrap(); + let sqlite = Arc::new( + SqliteProvider::new( + &format!("sqlite:{}?mode=rwc", dir.path().join("test.db").display()), + None, + ) + .await + .unwrap(), + ); + let store = Arc::new(PoisonInjectingProvider::new(sqlite.clone())); + let (ready, mut ready_events) = mpsc::unbounded_channel(); + let activities = ActivityRegistry::builder() + .register_with_admission( + "Limited", + |ctx, _input| { + assert_eq!(ctx.orchestration_name(), "Backlog"); + assert_eq!(ctx.orchestration_version(), "1.0.0"); + Ok(ActivityAdmission::Defer(Duration::from_millis(100))) + }, + |_ctx, _input| async move { panic!("deferred handler must not run") }, + ) + .register("Free", move |_ctx: ActivityContext, input: String| { + let ready = ready.clone(); + async move { + ready.send(()).unwrap(); + Ok(input) + } + }) + .build(); + let orchestrations = OrchestrationRegistry::builder() + .register("Backlog", move |ctx: OrchestrationContext, _input: String| async move { + let mut pending = (0..backlog) + .map(|_| ctx.schedule_activity("Limited", "payload")) + .collect::>(); + pending.push(ctx.schedule_activity("Free", "payload")); + ctx.join(pending).await; + Ok(String::new()) + }) + .build(); + let runtime = Runtime::start_with_options( + store.clone(), + activities, + orchestrations, + RuntimeOptions { + worker_concurrency: workers, + ..options() + }, + ) + .await; + Client::new(sqlite) + .start_orchestration("backlog", "Backlog", "") + .await + .unwrap(); + let progress = tokio::time::timeout(Duration::from_secs(10), ready_events.recv()).await; + runtime.shutdown(None).await; + assert_eq!( + progress.unwrap(), + Some(()), + "{backlog} deferred activities, {workers} workers" + ); + assert_eq!( + store.history_reads(), + 0, + "current-execution metadata must not load history" + ); + } +} + +#[tokio::test(flavor = "current_thread")] +async fn admission_metadata_uses_the_activity_execution() { + for hide_management in [false, true] { + let sqlite = Arc::new(SqliteProvider::new_in_memory().await.unwrap()); + common::test_create_execution(sqlite.as_ref(), "metadata", "Old", "1.0.0", "", None, None) + .await + .unwrap(); + common::test_create_execution(sqlite.as_ref(), "metadata", "New", "2.0.0", "", None, None) + .await + .unwrap(); + let store = Arc::new(PoisonInjectingProvider::new(sqlite.clone())); + if hide_management { + store.hide_management_capability(); + } + let (checked, mut checks) = mpsc::unbounded_channel(); + let activities = ActivityRegistry::builder() + .register_with_admission( + "Limited", + move |ctx, _input| { + checked + .send(( + ctx.execution_id(), + ctx.orchestration_name().to_string(), + ctx.orchestration_version().to_string(), + )) + .unwrap(); + Ok(ActivityAdmission::Defer(Duration::from_secs(60))) + }, + |_ctx, _input| async move { panic!("deferred handler must not run") }, + ) + .build(); + sqlite + .enqueue_for_worker(WorkItem::ActivityExecute { + instance: "metadata".to_string(), + execution_id: 1, + id: 2, + name: "Limited".to_string(), + input: String::new(), + session_id: None, + tag: None, + }) + .await + .unwrap(); + let runtime = Runtime::start_with_options(store, activities, orchestrations(), options()).await; + let metadata = tokio::time::timeout(Duration::from_secs(5), checks.recv()) + .await + .unwrap() + .unwrap(); + runtime.shutdown(None).await; + assert_eq!(metadata, (1, "Old".to_string(), "1.0.0".to_string())); + } +} + #[tokio::test(flavor = "current_thread")] async fn admission_guard_released_on_success_error_and_panic() { let (store, _dir) = common::create_sqlite_store_disk().await; @@ -202,6 +471,7 @@ async fn admission_guard_released_on_success_error_and_panic() { #[tokio::test(flavor = "current_thread")] async fn rejected_panicking_and_invalid_admission_never_invokes_handler() { + let (events, _guard) = common::tracing_capture::install_tracing_capture(); let (store, _dir) = common::create_sqlite_store_disk().await; let activities = ActivityRegistry::builder() .register_with_admission( @@ -232,7 +502,18 @@ async fn rejected_panicking_and_invalid_admission_never_invokes_handler() { "unexpected status: {status:?}" ); } - runtime.shutdown(None).await; + runtime.clone().shutdown(None).await; + let metrics = runtime.metrics_snapshot().unwrap(); + assert!(metrics.activity_app_errors >= 3); + assert_eq!(metrics.activity_infra_errors, 0); + assert_eq!(metrics.activity_admission_deferrals, 0); + assert!( + !events + .lock() + .unwrap() + .iter() + .any(|event| event.message == "Activity started") + ); } #[tokio::test(flavor = "current_thread")] diff --git a/tests/common/fault_injection.rs b/tests/common/fault_injection.rs index 6ff5c4c..6f9c62b 100644 --- a/tests/common/fault_injection.rs +++ b/tests/common/fault_injection.rs @@ -49,6 +49,9 @@ pub struct PoisonInjectingProvider { counted_abandon_delays: std::sync::Mutex>, orchestration_fetches: std::sync::Mutex>, work_abandons: std::sync::Mutex, bool)>>, + work_abandon_failures: AtomicU32, + history_reads: AtomicU32, + hide_management: AtomicBool, } impl PoisonInjectingProvider { @@ -70,6 +73,9 @@ impl PoisonInjectingProvider { counted_abandon_delays: std::sync::Mutex::new(Vec::new()), orchestration_fetches: std::sync::Mutex::new(Vec::new()), work_abandons: std::sync::Mutex::new(Vec::new()), + work_abandon_failures: AtomicU32::new(0), + history_reads: AtomicU32::new(0), + hide_management: AtomicBool::new(false), } } @@ -176,6 +182,18 @@ impl PoisonInjectingProvider { pub fn work_abandons(&self) -> Vec<(Option, bool)> { self.work_abandons.lock().expect("work abandons lock").clone() } + + pub fn fail_work_abandons(&self, count: u32) { + self.work_abandon_failures.store(count, Ordering::SeqCst); + } + + pub fn history_reads(&self) -> u32 { + self.history_reads.load(Ordering::SeqCst) + } + + pub fn hide_management_capability(&self) { + self.hide_management.store(true, Ordering::SeqCst); + } } #[async_trait] @@ -352,6 +370,16 @@ impl Provider for PoisonInjectingProvider { .lock() .expect("work abandons lock") .push((delay, ignore_attempt)); + if self + .work_abandon_failures + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |remaining| remaining.checked_sub(1)) + .is_ok() + { + return Err(ProviderError::retryable( + "abandon_work_item", + "injected abandon failure", + )); + } self.inner.abandon_work_item(token, delay, ignore_attempt).await } @@ -381,10 +409,12 @@ impl Provider for PoisonInjectingProvider { } async fn read(&self, instance: &str) -> Result, ProviderError> { + self.history_reads.fetch_add(1, Ordering::SeqCst); self.inner.read(instance).await } async fn read_with_execution(&self, instance: &str, execution_id: u64) -> Result, ProviderError> { + self.history_reads.fetch_add(1, Ordering::SeqCst); self.inner.read_with_execution(instance, execution_id).await } @@ -400,8 +430,11 @@ impl Provider for PoisonInjectingProvider { } fn as_management_capability(&self) -> Option<&dyn ProviderAdmin> { - // Delegate to inner's management capability - self.inner.as_management_capability() + if self.hide_management.load(Ordering::SeqCst) { + None + } else { + self.inner.as_management_capability() + } } async fn get_custom_status( diff --git a/tests/observability_tests.rs b/tests/observability_tests.rs index 809a5a3..6f6c104 100644 --- a/tests/observability_tests.rs +++ b/tests/observability_tests.rs @@ -1514,6 +1514,13 @@ async fn test_metrics_activity_outcomes() { // Config error metrics.record_activity_execution("TestActivity", "config_error", 0.3, 0, Some("build")); assert_eq!(metrics.snapshot().activity_config_errors, 1); + + metrics.record_activity_admission_deferral("TestActivity", Some("build")); + let snapshot = metrics.snapshot(); + assert_eq!(snapshot.activity_admission_deferrals, 1); + assert_eq!(snapshot.activity_success, 1); + assert_eq!(snapshot.activity_app_errors, 1); + assert_eq!(snapshot.activity_infra_errors, 1); } /// Test: MetricsProvider poison message tracking diff --git a/tests/sqlite_provider_validations.rs b/tests/sqlite_provider_validations.rs index d810957..7c5c779 100644 --- a/tests/sqlite_provider_validations.rs +++ b/tests/sqlite_provider_validations.rs @@ -497,6 +497,11 @@ mod tests { test_worker_queue_fifo_ordering(&SqliteTestFactory).await; } + #[tokio::test] + async fn test_sqlite_worker_deferred_item_does_not_starve_ready_work() { + duroxide::provider_validations::test_worker_deferred_item_does_not_starve_ready_work(&SqliteTestFactory).await; + } + #[tokio::test] async fn test_sqlite_worker_peek_lock_semantics() { test_worker_peek_lock_semantics(&SqliteTestFactory).await; From aba85769c6d2c266b3989b784037188a9a4f8056 Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Fri, 30 Oct 2026 14:29:44 +0000 Subject: [PATCH 05/11] feat: add typed activity admission registration Add register_typed_with_admission with borrowed typed admission input, owned typed handler input, and the existing typed codec. Preserve deferral and guard-lifetime semantics, and validate decoding, encoding, and registry errors. Ref: #74 (typed admission) Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 3 +- docs/ORCHESTRATION-GUIDE.md | 8 ++ src/runtime/registry.rs | 53 ++++++++++- tests/activity_admission_tests.rs | 147 ++++++++++++++++++++++++++++++ tests/registry_tests.rs | 25 +++++ 5 files changed, 230 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8bed747..65b16ff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- Opt-in activity resource admission through `register_with_admission`. +- Opt-in activity resource admission through `register_with_admission` and + `register_typed_with_admission` for typed input and output. Admission can retain a guard for the handler's lifetime or durably defer execution without occupying a worker slot or recording an activity result. Positive retry delays are clamped to 1 ms–24 hours. Only committed deferrals diff --git a/docs/ORCHESTRATION-GUIDE.md b/docs/ORCHESTRATION-GUIDE.md index 36e89e6..2d38dee 100644 --- a/docs/ORCHESTRATION-GUIDE.md +++ b/docs/ORCHESTRATION-GUIDE.md @@ -275,6 +275,14 @@ task is dropped, including error, panic, and cancellation. If several limits apply, reserve all of them atomically or defer holding none. Detached work is not protected after the handler finishes. +For typed input and output, use +`register_typed_with_admission(name, admission, handler)`. The admission callback +receives `&In`; the handler receives an owned `In` and returns `Result`. +It uses the same codec as `register_typed`, including raw-string compatibility. +Invalid input fails before either callback runs. Input is decoded for each +admission check and again when invoking an admitted handler, so deserialization +must be side-effect-free. + `Defer(delay)` requires a positive delay. The runtime clamps it to 1 ms through 24 hours (`ActivityAdmission::MAX_DEFER_DELAY`) before passing it to any provider. Deferral leaves the original activity pending in durable storage, frees the diff --git a/src/runtime/registry.rs b/src/runtime/registry.rs index b7c25d5..049ed19 100644 --- a/src/runtime/registry.rs +++ b/src/runtime/registry.rs @@ -411,6 +411,22 @@ impl OrchestrationRegistryBuilder { // Activity Builder - Specialized Methods // ============================================================================ +async fn invoke_typed_activity( + handler: &F, + ctx: crate::ActivityContext, + input: String, +) -> Result +where + In: serde::de::DeserializeOwned + Send, + Out: serde::Serialize + Send, + F: Fn(crate::ActivityContext, In) -> Fut + Send + Sync, + Fut: std::future::Future> + Send, +{ + let input: In = crate::_typed_codec::Json::decode(&input)?; + let output: Out = handler(ctx, input).await?; + crate::_typed_codec::Json::encode(&output) +} + struct AdmissionControlledActivity { admission: A, handler: F, @@ -507,11 +523,7 @@ impl ActivityRegistryBuilder { let f_clone = std::sync::Arc::new(f); let wrapper = move |ctx: crate::ActivityContext, input_s: String| { let f_inner = f_clone.clone(); - async move { - let input: In = crate::_typed_codec::Json::decode(&input_s)?; - let out: Out = (f_inner)(ctx, input).await?; - crate::_typed_codec::Json::encode(&out) - } + async move { invoke_typed_activity::(f_inner.as_ref(), ctx, input_s).await } }; let name = name.into(); // Check reserved prefix first @@ -530,6 +542,37 @@ impl ActivityRegistryBuilder { self } + /// Register a typed activity with a nonblocking resource-admission callback. + /// + /// Admission receives a borrowed decoded input; the handler receives an owned + /// decoded input. Uses the same codec as [`Self::register_typed`], including + /// raw-string compatibility. Invalid input fails before either callback runs. + /// + /// Input is decoded for each admission check and again for an admitted handler, + /// so deserialization must be side-effect-free. Admission and guard lifetime + /// follow [`Self::register_with_admission`]. + pub fn register_typed_with_admission(self, name: impl Into, admission: A, f: F) -> Self + where + In: serde::de::DeserializeOwned + Send + 'static, + Out: serde::Serialize + Send + 'static, + A: Fn(&crate::ActivityContext, &In) -> Result + Send + Sync + 'static, + F: Fn(crate::ActivityContext, In) -> Fut + Send + Sync + 'static, + Fut: std::future::Future> + Send + 'static, + { + let handler = Arc::new(f); + self.register_with_admission( + name, + move |ctx, input| { + let input: In = crate::_typed_codec::Json::decode(input)?; + admission(ctx, &input) + }, + move |ctx, input| { + let handler = Arc::clone(&handler); + async move { invoke_typed_activity::(handler.as_ref(), ctx, input).await } + }, + ) + } + /// Register a built-in activity (bypasses reserved prefix check). /// Used internally to register system activities like new_guid and utc_now_ms. pub(crate) fn register_builtin(mut self, name: impl Into, f: F) -> Self diff --git a/tests/activity_admission_tests.rs b/tests/activity_admission_tests.rs index d4aeb9e..287038c 100644 --- a/tests/activity_admission_tests.rs +++ b/tests/activity_admission_tests.rs @@ -37,6 +37,153 @@ fn orchestrations() -> OrchestrationRegistry { .build() } +#[derive(serde::Serialize, serde::Deserialize)] +struct TypedInput { + value: u32, +} + +#[derive(serde::Serialize, serde::Deserialize)] +struct TypedOutput { + doubled: u32, +} + +#[tokio::test(flavor = "current_thread")] +async fn typed_admission_defers_then_invokes_with_decoded_input() { + let (store, _dir) = common::create_sqlite_store_disk().await; + let capacity = Arc::new(Semaphore::new(1)); + let held = capacity.clone().acquire_owned().await.unwrap(); + let admission_capacity = capacity.clone(); + let handler_capacity = capacity.clone(); + let (deferred, mut deferrals) = mpsc::unbounded_channel(); + let activities = ActivityRegistry::builder() + .register_typed_with_admission( + "Limited", + move |ctx, input: &TypedInput| { + assert_eq!(ctx.instance_id(), "typed"); + assert_eq!(input.value, 7); + match admission_capacity.clone().try_acquire_owned() { + Ok(permit) => Ok(ActivityAdmission::admit(permit)), + Err(tokio::sync::TryAcquireError::NoPermits) => { + deferred.send(()).unwrap(); + Ok(ActivityAdmission::Defer(Duration::from_millis(10))) + } + Err(error) => Err(error.to_string()), + } + }, + move |_ctx, input: TypedInput| { + let capacity = handler_capacity.clone(); + async move { + assert_eq!(capacity.available_permits(), 0); + Ok(TypedOutput { + doubled: input.value * 2, + }) + } + }, + ) + .build(); + let orchestrations = OrchestrationRegistry::builder() + .register("Run", |ctx: OrchestrationContext, _input: String| async move { + let output: TypedOutput = ctx.schedule_activity_typed("Limited", &TypedInput { value: 7 }).await?; + Ok(output.doubled.to_string()) + }) + .build(); + let runtime = Runtime::start_with_options(store.clone(), activities, orchestrations, options()).await; + let client = Client::new(store); + client.start_orchestration("typed", "Run", "").await.unwrap(); + tokio::time::timeout(Duration::from_secs(5), deferrals.recv()) + .await + .unwrap() + .unwrap(); + drop(held); + let status = client.wait_for_orchestration("typed", Duration::from_secs(5)).await; + runtime.shutdown(None).await; + assert!(matches!( + status.unwrap(), + OrchestrationStatus::Completed { output, .. } if output == "14" + )); + assert_eq!(capacity.available_permits(), 1); +} + +#[tokio::test(flavor = "current_thread")] +async fn typed_admission_invalid_input_skips_both_callbacks() { + let (store, _dir) = common::create_sqlite_store_disk().await; + let activities = ActivityRegistry::builder() + .register_typed_with_admission::( + "Limited", + |_ctx, _input| panic!("admission must not run for invalid input"), + |_ctx, _input| async move { panic!("handler must not run for invalid input") }, + ) + .build(); + let orchestrations = OrchestrationRegistry::builder() + .register("Run", |ctx: OrchestrationContext, input: String| async move { + ctx.schedule_activity("Limited", input).await + }) + .build(); + let runtime = Runtime::start_with_options(store.clone(), activities, orchestrations, options()).await; + let client = Client::new(store); + for (instance, input) in [("malformed", "{"), ("wrong-type", r#"{"value":"not a number"}"#)] { + client.start_orchestration(instance, "Run", input).await.unwrap(); + let status = client + .wait_for_orchestration(instance, Duration::from_secs(5)) + .await + .unwrap(); + assert!( + matches!(status, OrchestrationStatus::Failed { ref details, .. } if details.display_message().contains("invalid type")), + "unexpected status: {status:?}" + ); + } + runtime.shutdown(None).await; +} + +#[tokio::test(flavor = "current_thread")] +async fn typed_admission_preserves_raw_strings_and_releases_guard_on_encoding_error() { + let (store, _dir) = common::create_sqlite_store_disk().await; + let capacity = Arc::new(Semaphore::new(1)); + let admission_capacity = capacity.clone(); + let activities = ActivityRegistry::builder() + .register_typed_with_admission( + "String", + |_ctx, input: &String| { + assert_eq!(input, "payload"); + Ok(ActivityAdmission::admit(())) + }, + |_ctx, input: String| async move { Ok(input) }, + ) + .register_typed_with_admission( + "InvalidOutput", + move |_ctx, input: &String| { + assert_eq!(input, "payload"); + admission_capacity + .clone() + .try_acquire_owned() + .map(ActivityAdmission::admit) + .map_err(|error| error.to_string()) + }, + |_ctx, _input: String| async move { Ok(std::collections::BTreeMap::from([((1, 2), 3)])) }, + ) + .build(); + let runtime = Runtime::start_with_options(store.clone(), activities, orchestrations(), options()).await; + let client = Client::new(store); + client.start_orchestration("string", "Run", "String").await.unwrap(); + assert!(matches!( + client.wait_for_orchestration("string", Duration::from_secs(5)).await.unwrap(), + OrchestrationStatus::Completed { output, .. } if output == "payload" + )); + client + .start_orchestration("encoding-error", "Run", "InvalidOutput") + .await + .unwrap(); + let status = client + .wait_for_orchestration("encoding-error", Duration::from_secs(5)) + .await; + runtime.shutdown(None).await; + assert!(matches!( + status.unwrap(), + OrchestrationStatus::Failed { details, .. } if details.display_message().contains("key must be a string") + )); + assert_eq!(capacity.available_permits(), 1); +} + #[tokio::test(flavor = "current_thread")] async fn deferred_admission_frees_worker_without_consuming_attempts_and_survives_restart() { let (store, _dir) = common::create_sqlite_store_disk().await; diff --git a/tests/registry_tests.rs b/tests/registry_tests.rs index 7949743..e1675df 100644 --- a/tests/registry_tests.rs +++ b/tests/registry_tests.rs @@ -681,6 +681,31 @@ fn test_activity_register_typed() { assert_eq!(v, semver::Version::parse("1.0.0").unwrap()); } +#[test] +fn test_typed_admission_registration_validation() { + let admission = |_ctx: &ActivityContext, _input: &String| Ok(runtime::ActivityAdmission::admit(())); + let handler = |_ctx: ActivityContext, input: String| async move { Ok(input) }; + + let registry = ActivityRegistry::builder() + .register_typed_with_admission("typed-admission", admission, handler) + .build(); + assert_eq!( + registry.resolve_version("typed-admission"), + Some(semver::Version::new(1, 0, 0)) + ); + + let duplicate = ActivityRegistry::builder() + .register_typed("duplicate", handler) + .register_typed_with_admission("duplicate", admission, handler) + .build_result(); + assert!(duplicate.err().unwrap().contains("duplicate activity registration")); + + let reserved = ActivityRegistry::builder() + .register_typed_with_admission("__duroxide_syscall:typed", admission, handler) + .build_result(); + assert!(reserved.err().unwrap().contains("uses reserved prefix")); +} + #[test] fn test_register_all() { let handler = |_ctx: OrchestrationContext, input: String| async move { Ok(format!("processed: {input}")) }; From 74590e45e250c5ba1941d725840f8af637ea1577 Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Fri, 30 Oct 2026 15:11:06 +0000 Subject: [PATCH 06/11] Fix docs and add tests around resource-admission deferrals --- CHANGELOG.md | 5 +- docs/ORCHESTRATION-GUIDE.md | 12 +- docs/observability-guide.md | 7 +- docs/provider-implementation-guide.md | 9 +- src/runtime/mod.rs | 6 +- tests/activity_admission_tests.rs | 195 +++++++++++++++++++++++++- tests/common/fault_injection.rs | 16 +++ tests/common/long_polling.rs | 7 +- 8 files changed, 246 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 65b16ff..906fff0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,8 +15,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 execution without occupying a worker slot or recording an activity result. Positive retry delays are clamped to 1 ms–24 hours. Only committed deferrals restore their fetch attempt; failed or interrupted deferrals may still use - the poison-message budget. Committed deferrals have a separate - `duroxide_activity_admission_deferrals_total` metric. (#74) + the poison-message budget, even when the handler never starts. Uncommitted + deferral delays may be lost on lock expiry. Successfully observed deferrals + have a separate `duroxide_activity_admission_deferrals_total` metric. (#74) ### Fixed diff --git a/docs/ORCHESTRATION-GUIDE.md b/docs/ORCHESTRATION-GUIDE.md index 2d38dee..47ce680 100644 --- a/docs/ORCHESTRATION-GUIDE.md +++ b/docs/ORCHESTRATION-GUIDE.md @@ -289,14 +289,20 @@ Deferral leaves the original activity pending in durable storage, frees the worker to fetch other work, and does not record an activity result. A **committed** deferral restores its fetch attempt. Crashes, shutdown aborts, and failed abandons can still consume the poison-message attempt budget, so -leave room for recovery attempts. Deferral is not a retry of an executed activity. +leave room for recovery attempts. This is a fetch-attempt budget, not a count of +handler starts: an activity can become poison without ever invoking its handler. +Failures loading activity metadata before admission also consume attempts. +Deferral is not a retry of an executed activity. An admission error, panic, or zero delay fails the activity rather than invoking the handler and is reported as an application error. Storage errors during deferral are infrastructure errors; an immediate fallback abandon preserves the same delay and attempt-restoration flag. If that also fails, recovery waits for -the work lock to expire. Successfully committed deferrals increment -`duroxide_activity_admission_deferrals_total`, not the activity execution counter. +the work lock to expire, which can happen before the requested delay. A requested +delay is not a durable "not before" guarantee until its abandon commits. +Deferrals whose abandon returns success increment +`duroxide_activity_admission_deferrals_total`, not the activity execution counter; +a commit whose response is lost can restore the attempt without being counted. The "Activity started" log is emitted only after admission succeeds. Admission runs outside orchestration replay; completed results are unaffected. diff --git a/docs/observability-guide.md b/docs/observability-guide.md index 20807b6..cfae0ec 100644 --- a/docs/observability-guide.md +++ b/docs/observability-guide.md @@ -267,7 +267,7 @@ All metrics follow Prometheus naming conventions with `_total` for counters and #### Activity Execution Metrics **`duroxide_activity_admission_deferrals_total`** (Counter) -- **Description**: Admission deferrals successfully committed to storage, including a successful fallback after an abandon error +- **Description**: Admission deferrals whose abandon returned success, including a successful fallback after an abandon error - **Labels**: `activity_name`, `tag` - **Use**: Identify resource-constrained activities without counting deferrals as executions @@ -275,7 +275,10 @@ The "Activity started" log appears only after admission succeeds. Admission rejections, panics, and zero retry delays count as `app_error`; storage failures while deferring count as `infra_error`, even if the fallback abandon succeeds. Uncommitted deferrals do not increment the deferral counter and may consume a -fetch attempt on recovery. +fetch attempt on recovery. A deferral that commits before shutdown or a lost +response can restore its attempt without incrementing this counter. The counter +is not a durable audit of all storage commits. Activity metadata-read failures +before admission count as infrastructure errors, not deferrals. **`duroxide_activity_executions_total`** (Counter) - **Description**: Activity execution attempts diff --git a/docs/provider-implementation-guide.md b/docs/provider-implementation-guide.md index 944b12d..9e5c939 100644 --- a/docs/provider-implementation-guide.md +++ b/docs/provider-implementation-guide.md @@ -1448,12 +1448,19 @@ that was already ready, even after their delay expires. Order eligible rows by Apply the same ordering to session-aware and tag-filtered fetches. Admission deferral restores an attempt only when `abandon_work_item(..., true)` commits; failed or interrupted abandons can still consume attempts. +If no abandon commits, lock expiry can make the item eligible before its requested +retry delay. Conversely, if the commit succeeds but its response is lost, the +stored delay and restored attempt survive even though the runtime did not observe +success. Neither fetch counts nor a durable admission/start-intent write can prove +that a handler actually began executing across a process crash. The worker uses `ProviderAdmin::get_instance_info` for lightweight activity metadata when it describes the activity's execution. Providers without that capability, and activities from an earlier execution, fall back to `read_with_execution`. Metadata must describe the requested execution, not a -newer continue-as-new execution. +newer continue-as-new execution. If the execution-history read fails, the worker +abandons the fetch without restoring its attempt and does not invoke admission +or the handler. ### ack_work_item() — Complete Implementation diff --git a/src/runtime/mod.rs b/src/runtime/mod.rs index 75a1965..3461a18 100644 --- a/src/runtime/mod.rs +++ b/src/runtime/mod.rs @@ -247,6 +247,8 @@ pub struct RuntimeOptions { /// /// After this many fetch attempts, the runtime will immediately fail /// the orchestration/activity with a Poison error instead of processing. + /// Fetches can consume this budget before a handler starts, including metadata + /// read failures and shutdown during an uncommitted admission deferral. /// /// Default: 10 pub max_attempts: u32, @@ -465,7 +467,9 @@ pub enum ActivityAdmission { /// Positive delays are clamped to one millisecond through [`Self::MAX_DEFER_DELAY`]. /// Deferral releases the worker slot without recording a history event or /// consuming a fetch attempt once the deferral commits. Interrupted or failed - /// deferrals can still consume attempts. It does not retry an executed activity. + /// deferrals can still consume attempts and be redelivered on lock expiry + /// before the requested delay. An activity may become poison without starting + /// its handler. Deferral does not retry an executed activity. Defer(Duration), } diff --git a/tests/activity_admission_tests.rs b/tests/activity_admission_tests.rs index 287038c..5e496f6 100644 --- a/tests/activity_admission_tests.rs +++ b/tests/activity_admission_tests.rs @@ -6,12 +6,13 @@ mod common; use common::fault_injection::PoisonInjectingProvider; +use common::long_polling::{LongPollingSqliteProvider, ProviderGate, ProviderOperation}; use duroxide::providers::sqlite::SqliteProvider; use duroxide::providers::{Provider, WorkItem}; use duroxide::runtime::registry::ActivityRegistry; use duroxide::runtime::{ActivityAdmission, Runtime, RuntimeOptions}; use duroxide::{ActivityContext, Client, EventKind, OrchestrationContext, OrchestrationRegistry, OrchestrationStatus}; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; use tokio::sync::{Semaphore, mpsc}; @@ -444,6 +445,198 @@ fn admission_debug_does_not_require_a_debug_guard() { assert_eq!(ActivityAdmission::MAX_DEFER_DELAY, Duration::from_secs(24 * 60 * 60)); } +#[tokio::test(flavor = "current_thread")] +async fn shutdown_before_deferral_commit_can_poison_without_a_handler_start() { + let (store, _dir) = common::create_sqlite_store_disk().await; + let admissions = Arc::new(AtomicUsize::new(0)); + let callback_admissions = admissions.clone(); + let activities = ActivityRegistry::builder() + .register_with_admission( + "Limited", + move |_ctx, _input| { + callback_admissions.fetch_add(1, Ordering::SeqCst); + Ok(ActivityAdmission::Defer(Duration::from_secs(60))) + }, + |_ctx, _input| async move { panic!("no handler should start") }, + ) + .build(); + let opts = RuntimeOptions { + max_attempts: 2, + worker_lock_timeout: Duration::from_millis(100), + ..options() + }; + let client = Client::new(store.clone()); + client + .start_orchestration("stopped-deferral", "Run", "Limited") + .await + .unwrap(); + for attempt in 1..=2 { + let gate = Arc::new(ProviderGate::default()); + let provider = Arc::new( + LongPollingSqliteProvider::new(store.clone()).with_gate(ProviderOperation::BeforeWorkAbandon, gate.clone()), + ); + let runtime = Runtime::start_with_options(provider, activities.clone(), orchestrations(), opts.clone()).await; + tokio::time::timeout(Duration::from_secs(5), gate.wait_until_entered()) + .await + .expect("deferral must reach its uncommitted abandon"); + runtime.clone().shutdown(Some(0)).await; + assert_eq!(admissions.load(Ordering::SeqCst), attempt); + assert_eq!(runtime.metrics_snapshot().unwrap().activity_admission_deferrals, 0); + assert!( + !store + .read("stopped-deferral") + .await + .unwrap() + .iter() + .any(|event| matches!( + event.kind, + EventKind::ActivityCompleted { .. } | EventKind::ActivityFailed { .. } + )) + ); + } + + let runtime = Runtime::start_with_options(store, activities, orchestrations(), opts).await; + let status = client + .wait_for_orchestration("stopped-deferral", Duration::from_secs(5)) + .await; + runtime.shutdown(None).await; + assert!(matches!( + status.unwrap(), + OrchestrationStatus::Failed { + details: duroxide::ErrorDetails::Poison { + attempt_count: 3, + max_attempts: 2, + .. + }, + .. + } + )); + assert_eq!( + admissions.load(Ordering::SeqCst), + 2, + "poison is checked before admission" + ); +} + +#[tokio::test(flavor = "current_thread")] +async fn shutdown_after_deferral_commit_preserves_attempt_refund_without_a_response() { + let (store, _dir) = common::create_sqlite_store_disk().await; + let gate = Arc::new(ProviderGate::default()); + let provider = Arc::new( + LongPollingSqliteProvider::new(store.clone()).with_gate(ProviderOperation::AfterWorkAbandon, gate.clone()), + ); + let admitted = Arc::new(AtomicBool::new(false)); + let callback_admitted = admitted.clone(); + let activities = ActivityRegistry::builder() + .register_with_admission( + "Limited", + move |_ctx, _input| { + Ok(if callback_admitted.load(Ordering::SeqCst) { + ActivityAdmission::admit(()) + } else { + ActivityAdmission::Defer(Duration::from_millis(25)) + }) + }, + |_ctx, input| async move { Ok(input) }, + ) + .build(); + let runtime = Runtime::start_with_options(provider, activities.clone(), orchestrations(), options()).await; + let client = Client::new(store.clone()); + client + .start_orchestration("committed-deferral", "Run", "Limited") + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(5), gate.wait_until_entered()) + .await + .expect("deferral must commit before its response is blocked"); + runtime.clone().shutdown(Some(0)).await; + assert_eq!(runtime.metrics_snapshot().unwrap().activity_admission_deferrals, 0); + assert!( + !store + .read("committed-deferral") + .await + .unwrap() + .iter() + .any(|event| matches!( + event.kind, + EventKind::ActivityCompleted { .. } | EventKind::ActivityFailed { .. } + )) + ); + + admitted.store(true, Ordering::SeqCst); + let runtime = Runtime::start_with_options(store, activities, orchestrations(), options()).await; + let status = client + .wait_for_orchestration("committed-deferral", Duration::from_secs(5)) + .await; + runtime.shutdown(None).await; + assert!(matches!( + status.unwrap(), + OrchestrationStatus::Completed { output, .. } if output == "payload" + )); +} + +#[tokio::test(flavor = "current_thread")] +async fn metadata_read_failure_consumes_an_attempt_before_admission() { + for max_attempts in [1, 2] { + let sqlite = Arc::new(SqliteProvider::new_in_memory().await.unwrap()); + let store = Arc::new(PoisonInjectingProvider::new(sqlite.clone())); + store.hide_management_capability(); + store.fail_execution_history_reads(1); + let admissions = Arc::new(AtomicUsize::new(0)); + let callback_admissions = admissions.clone(); + let activities = ActivityRegistry::builder() + .register_with_admission( + "Limited", + move |ctx, _input| { + callback_admissions.fetch_add(1, Ordering::SeqCst); + assert_eq!(ctx.orchestration_name(), "Run"); + assert_eq!(ctx.orchestration_version(), "1.0.0"); + Ok(ActivityAdmission::admit(())) + }, + |_ctx, input| async move { Ok(input) }, + ) + .build(); + let runtime = Runtime::start_with_options( + store.clone(), + activities, + orchestrations(), + RuntimeOptions { + max_attempts, + ..options() + }, + ) + .await; + let client = Client::new(sqlite); + client + .start_orchestration("metadata-failure", "Run", "Limited") + .await + .unwrap(); + let status = client + .wait_for_orchestration("metadata-failure", Duration::from_secs(5)) + .await + .unwrap(); + runtime.clone().shutdown(None).await; + if max_attempts == 1 { + assert!(matches!( + status, + OrchestrationStatus::Failed { + details: duroxide::ErrorDetails::Poison { attempt_count: 2, .. }, + .. + } + )); + assert_eq!(admissions.load(Ordering::SeqCst), 0); + } else { + assert!(matches!(status, OrchestrationStatus::Completed { output, .. } if output == "payload")); + assert_eq!(admissions.load(Ordering::SeqCst), 1); + } + assert_eq!(store.work_abandons(), vec![(Some(Duration::from_millis(100)), false)]); + let metrics = runtime.metrics_snapshot().unwrap(); + assert_eq!(metrics.activity_infra_errors, 1); + assert_eq!(metrics.activity_app_errors, 0); + assert_eq!(metrics.activity_admission_deferrals, 0); + } +} + #[tokio::test(flavor = "current_thread")] async fn deferred_backlog_does_not_starve_ready_work_or_reread_history() { for (backlog, workers) in [(300, 1), (1000, 4)] { diff --git a/tests/common/fault_injection.rs b/tests/common/fault_injection.rs index 6f9c62b..89d6463 100644 --- a/tests/common/fault_injection.rs +++ b/tests/common/fault_injection.rs @@ -51,6 +51,7 @@ pub struct PoisonInjectingProvider { work_abandons: std::sync::Mutex, bool)>>, work_abandon_failures: AtomicU32, history_reads: AtomicU32, + execution_history_read_failures: AtomicU32, hide_management: AtomicBool, } @@ -75,6 +76,7 @@ impl PoisonInjectingProvider { work_abandons: std::sync::Mutex::new(Vec::new()), work_abandon_failures: AtomicU32::new(0), history_reads: AtomicU32::new(0), + execution_history_read_failures: AtomicU32::new(0), hide_management: AtomicBool::new(false), } } @@ -191,6 +193,10 @@ impl PoisonInjectingProvider { self.history_reads.load(Ordering::SeqCst) } + pub fn fail_execution_history_reads(&self, count: u32) { + self.execution_history_read_failures.store(count, Ordering::SeqCst); + } + pub fn hide_management_capability(&self) { self.hide_management.store(true, Ordering::SeqCst); } @@ -415,6 +421,16 @@ impl Provider for PoisonInjectingProvider { async fn read_with_execution(&self, instance: &str, execution_id: u64) -> Result, ProviderError> { self.history_reads.fetch_add(1, Ordering::SeqCst); + if self + .execution_history_read_failures + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |remaining| remaining.checked_sub(1)) + .is_ok() + { + return Err(ProviderError::retryable( + "read_with_execution", + "injected history read failure", + )); + } self.inner.read_with_execution(instance, execution_id).await } diff --git a/tests/common/long_polling.rs b/tests/common/long_polling.rs index 365b894..c0193ed 100644 --- a/tests/common/long_polling.rs +++ b/tests/common/long_polling.rs @@ -29,6 +29,8 @@ pub enum ProviderOperation { AfterOrchestrationAck, BeforeWorkAck, AfterWorkAck, + BeforeWorkAbandon, + AfterWorkAbandon, BeforeOrchestrationRenewal, BeforeWorkRenewal, } @@ -232,7 +234,10 @@ impl Provider for LongPollingSqliteProvider { delay: Option, ignore_attempt: bool, ) -> Result<(), ProviderError> { - self.inner.abandon_work_item(token, delay, ignore_attempt).await + self.wait_at(ProviderOperation::BeforeWorkAbandon).await; + self.inner.abandon_work_item(token, delay, ignore_attempt).await?; + self.wait_at(ProviderOperation::AfterWorkAbandon).await; + Ok(()) } async fn renew_session_lock( From 13673f80ed0e563716e9d3e115fbf74813b600ed Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Fri, 30 Oct 2026 15:23:28 +0000 Subject: [PATCH 07/11] fix: preserve MetricsSnapshot source compatibility Expose admission deferrals through a MetricsProvider accessor instead of adding a public snapshot field. Keep downstream struct literals and exhaustive patterns source-compatible, with an integration-test compile guard. Ref: #74 (74-N1) Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- docs/observability-guide.md | 3 ++ src/runtime/observability.rs | 8 ++++-- tests/activity_admission_tests.rs | 22 ++++++++++----- tests/observability_tests.rs | 47 ++++++++++++++++++++++++++++++- 4 files changed, 70 insertions(+), 10 deletions(-) diff --git a/docs/observability-guide.md b/docs/observability-guide.md index cfae0ec..e96ae88 100644 --- a/docs/observability-guide.md +++ b/docs/observability-guide.md @@ -271,6 +271,9 @@ All metrics follow Prometheus naming conventions with `_total` for counters and - **Labels**: `activity_name`, `tag` - **Use**: Identify resource-constrained activities without counting deferrals as executions +For an in-process count, use `MetricsProvider::activity_admission_deferrals()`. +The existing public `MetricsSnapshot` fields remain unchanged. + The "Activity started" log appears only after admission succeeds. Admission rejections, panics, and zero retry delays count as `app_error`; storage failures while deferring count as `infra_error`, even if the fallback abandon succeeds. diff --git a/src/runtime/observability.rs b/src/runtime/observability.rs index 0d2d1c8..00def86 100644 --- a/src/runtime/observability.rs +++ b/src/runtime/observability.rs @@ -133,7 +133,6 @@ pub struct MetricsSnapshot { pub activity_infra_errors: u64, pub activity_config_errors: u64, pub activity_poison: u64, - pub activity_admission_deferrals: u64, pub orch_dispatcher_items_fetched: u64, pub worker_dispatcher_items_fetched: u64, pub orch_continue_as_new: u64, @@ -472,6 +471,12 @@ impl MetricsProvider { self.activity_admission_deferrals_atomic.fetch_add(1, Ordering::Relaxed); } + /// Number of admission deferrals whose abandon returned success. + #[inline] + pub fn activity_admission_deferrals(&self) -> u64 { + self.activity_admission_deferrals_atomic.load(Ordering::Relaxed) + } + #[inline] pub fn record_orchestration_poison(&self) { self.orch_poison_atomic.fetch_add(1, Ordering::Relaxed); @@ -639,7 +644,6 @@ impl MetricsProvider { activity_infra_errors: self.activity_infra_errors_atomic.load(Ordering::Relaxed), activity_config_errors: self.activity_config_errors_atomic.load(Ordering::Relaxed), activity_poison: self.activity_poison_atomic.load(Ordering::Relaxed), - activity_admission_deferrals: self.activity_admission_deferrals_atomic.load(Ordering::Relaxed), orch_dispatcher_items_fetched: self.orch_dispatcher_items_fetched_atomic.load(Ordering::Relaxed), worker_dispatcher_items_fetched: self.worker_dispatcher_items_fetched_atomic.load(Ordering::Relaxed), orch_continue_as_new: self.orch_continue_as_new_atomic.load(Ordering::Relaxed), diff --git a/tests/activity_admission_tests.rs b/tests/activity_admission_tests.rs index 5e496f6..8baa300 100644 --- a/tests/activity_admission_tests.rs +++ b/tests/activity_admission_tests.rs @@ -30,6 +30,14 @@ fn options() -> RuntimeOptions { } } +fn admission_deferrals(runtime: &Runtime) -> u64 { + runtime + .observability_handle() + .unwrap() + .metrics_provider() + .activity_admission_deferrals() +} + fn orchestrations() -> OrchestrationRegistry { OrchestrationRegistry::builder() .register("Run", |ctx: OrchestrationContext, activity: String| async move { @@ -227,7 +235,7 @@ async fn deferred_admission_frees_worker_without_consuming_attempts_and_survives OrchestrationStatus::Completed { output, .. } if output == "payload" )); runtime.clone().shutdown(None).await; - assert!(runtime.metrics_snapshot().unwrap().activity_admission_deferrals >= 20); + assert!(admission_deferrals(&runtime) >= 20); let history = store.read("limited").await.unwrap(); assert_eq!( @@ -366,7 +374,7 @@ async fn failed_deferral_retries_the_same_delay_and_restores_the_attempt() { )); assert_eq!(store.work_abandons(), vec![(Some(delay), true); 2]); let metrics = runtime.metrics_snapshot().unwrap(); - assert_eq!(metrics.activity_admission_deferrals, 1); + assert_eq!(admission_deferrals(&runtime), 1); assert_eq!(metrics.activity_infra_errors, 1); assert_eq!( events @@ -422,7 +430,7 @@ async fn uncommitted_deferral_consumes_attempt_and_is_not_counted_as_deferred() )); assert_eq!(store.work_abandons(), vec![(Some(delay), true); 2]); let metrics = runtime.metrics_snapshot().unwrap(); - assert_eq!(metrics.activity_admission_deferrals, 0); + assert_eq!(admission_deferrals(&runtime), 0); assert_eq!(metrics.activity_infra_errors, 1); assert_eq!(metrics.activity_app_errors, 0); assert!( @@ -481,7 +489,7 @@ async fn shutdown_before_deferral_commit_can_poison_without_a_handler_start() { .expect("deferral must reach its uncommitted abandon"); runtime.clone().shutdown(Some(0)).await; assert_eq!(admissions.load(Ordering::SeqCst), attempt); - assert_eq!(runtime.metrics_snapshot().unwrap().activity_admission_deferrals, 0); + assert_eq!(admission_deferrals(&runtime), 0); assert!( !store .read("stopped-deferral") @@ -550,7 +558,7 @@ async fn shutdown_after_deferral_commit_preserves_attempt_refund_without_a_respo .await .expect("deferral must commit before its response is blocked"); runtime.clone().shutdown(Some(0)).await; - assert_eq!(runtime.metrics_snapshot().unwrap().activity_admission_deferrals, 0); + assert_eq!(admission_deferrals(&runtime), 0); assert!( !store .read("committed-deferral") @@ -633,7 +641,7 @@ async fn metadata_read_failure_consumes_an_attempt_before_admission() { let metrics = runtime.metrics_snapshot().unwrap(); assert_eq!(metrics.activity_infra_errors, 1); assert_eq!(metrics.activity_app_errors, 0); - assert_eq!(metrics.activity_admission_deferrals, 0); + assert_eq!(admission_deferrals(&runtime), 0); } } @@ -846,7 +854,7 @@ async fn rejected_panicking_and_invalid_admission_never_invokes_handler() { let metrics = runtime.metrics_snapshot().unwrap(); assert!(metrics.activity_app_errors >= 3); assert_eq!(metrics.activity_infra_errors, 0); - assert_eq!(metrics.activity_admission_deferrals, 0); + assert_eq!(admission_deferrals(&runtime), 0); assert!( !events .lock() diff --git a/tests/observability_tests.rs b/tests/observability_tests.rs index 6f6c104..8398acc 100644 --- a/tests/observability_tests.rs +++ b/tests/observability_tests.rs @@ -1491,6 +1491,51 @@ async fn test_metrics_outcome_matrix() { assert_eq!(metrics.snapshot().orch_continue_as_new, 1); } +#[test] +fn metrics_snapshot_preserves_public_struct_literals_and_patterns() { + use duroxide::runtime::observability::MetricsSnapshot; + + let snapshot = MetricsSnapshot { + orch_starts: 1, + orch_completions: 0, + orch_failures: 0, + orch_application_errors: 0, + orch_infrastructure_errors: 0, + orch_configuration_errors: 0, + orch_poison: 0, + activity_success: 0, + activity_app_errors: 0, + activity_infra_errors: 0, + activity_config_errors: 0, + activity_poison: 0, + orch_dispatcher_items_fetched: 0, + worker_dispatcher_items_fetched: 0, + orch_continue_as_new: 0, + suborchestration_calls: 0, + provider_errors: 0, + }; + let MetricsSnapshot { + orch_starts, + orch_completions: _, + orch_failures: _, + orch_application_errors: _, + orch_infrastructure_errors: _, + orch_configuration_errors: _, + orch_poison: _, + activity_success: _, + activity_app_errors: _, + activity_infra_errors: _, + activity_config_errors: _, + activity_poison: _, + orch_dispatcher_items_fetched: _, + worker_dispatcher_items_fetched: _, + orch_continue_as_new: _, + suborchestration_calls: _, + provider_errors: _, + } = snapshot; + assert_eq!(orch_starts, 1); +} + /// Test: MetricsProvider activity outcome types #[tokio::test] async fn test_metrics_activity_outcomes() { @@ -1517,7 +1562,7 @@ async fn test_metrics_activity_outcomes() { metrics.record_activity_admission_deferral("TestActivity", Some("build")); let snapshot = metrics.snapshot(); - assert_eq!(snapshot.activity_admission_deferrals, 1); + assert_eq!(metrics.activity_admission_deferrals(), 1); assert_eq!(snapshot.activity_success, 1); assert_eq!(snapshot.activity_app_errors, 1); assert_eq!(snapshot.activity_infra_errors, 1); From f8d4db7967576b9f0255ea165a13122243918e5b Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Fri, 30 Oct 2026 15:27:21 +0000 Subject: [PATCH 08/11] fix: migrate SQLite worker visibility index Install idx_worker_ready_order through the file-backed migration path as well as direct schema creation. Verify fresh databases and upgrades record the migration, create the expected index columns, and preserve queued work. Ref: #74 (74-N2) Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 2 + docs/sqlite-provider-design.md | 3 +- .../20240113000000_add_worker_ready_order.sql | 5 ++ tests/sqlite_tests.rs | 54 +++++++++++++++++++ 4 files changed, 63 insertions(+), 1 deletion(-) create mode 100644 migrations/20240113000000_add_worker_ready_order.sql diff --git a/CHANGELOG.md b/CHANGELOG.md index 906fff0..b1d0d78 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,6 +23,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - SQLite worker selection now orders eligible items by visibility time, then queue ID, so a deferred backlog cannot repeatedly overtake already-ready work. + Migration `20240113000000` adds the supporting `visible_at, id` index to new + and existing file-backed databases. Other providers must implement equivalent queue progress semantics; existing stored histories and work-item formats are unchanged. SQLite workers must all be upgraded to provide this behavior in a mixed-version deployment. (#74) diff --git a/docs/sqlite-provider-design.md b/docs/sqlite-provider-design.md index f185057..de476b2 100644 --- a/docs/sqlite-provider-design.md +++ b/docs/sqlite-provider-design.md @@ -109,7 +109,8 @@ future timestamp exceeds that range. This does not cover absolute Eligible worker items are ordered by `visible_at, id`, supported by the `idx_worker_ready_order` index. A deferred item goes behind already-ready work, even if its delay expires before the backlog drains. Session and tag filtering -use the same ordering. +use the same ordering. Migration `20240113000000` creates the index for new and +existing file-backed databases; in-memory databases create it directly. ## Implementation Plan diff --git a/migrations/20240113000000_add_worker_ready_order.sql b/migrations/20240113000000_add_worker_ready_order.sql new file mode 100644 index 0000000..803ed88 --- /dev/null +++ b/migrations/20240113000000_add_worker_ready_order.sql @@ -0,0 +1,5 @@ +-- Copyright (c) Microsoft Corporation. +-- Licensed under the MIT License. + +-- Support visibility-ordered worker fetches on file-backed databases. +CREATE INDEX IF NOT EXISTS idx_worker_ready_order ON worker_queue(visible_at, id); diff --git a/tests/sqlite_tests.rs b/tests/sqlite_tests.rs index 0fc11ff..4ddae11 100644 --- a/tests/sqlite_tests.rs +++ b/tests/sqlite_tests.rs @@ -38,6 +38,60 @@ async fn create_sqlite_store_named(name: &str) -> (Arc, TempDir, S // BASIC FUNCTIONALITY TESTS // ============================================================================ +#[tokio::test] +async fn test_worker_ready_index_on_file_backed_create_and_upgrade() { + for upgrade in [false, true] { + let (mut store, _dir, url) = create_sqlite_store_named("worker-ready-index").await; + let item = WorkItem::ActivityExecute { + instance: "index-migration".to_string(), + execution_id: 1, + id: 1, + name: "Activity".to_string(), + input: "preserved".to_string(), + session_id: None, + tag: None, + }; + store.enqueue_for_worker(item.clone()).await.unwrap(); + let pool = sqlx::SqlitePool::connect(&url).await.unwrap(); + if upgrade { + // This migration adds only the index, so removing both reproduces the previous schema. + sqlx::query("DROP INDEX IF EXISTS idx_worker_ready_order") + .execute(&pool) + .await + .unwrap(); + sqlx::query("DELETE FROM _sqlx_migrations WHERE version = 20240113000000") + .execute(&pool) + .await + .unwrap(); + drop(store); + store = Arc::new(SqliteProvider::new(&url, None).await.unwrap()); + } + + let columns: Vec = + sqlx::query_scalar("SELECT name FROM pragma_index_info('idx_worker_ready_order') ORDER BY seqno") + .fetch_all(&pool) + .await + .unwrap(); + assert_eq!(columns, ["visible_at", "id"], "upgrade={upgrade}"); + let applied: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM _sqlx_migrations WHERE version = 20240113000000 AND success = 1") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(applied, 1, "the migration must run, not the schema-creation fallback"); + + let (fetched, token, attempts) = store + .fetch_work_item(Duration::from_secs(30), Duration::ZERO, None, &TagFilter::DefaultOnly) + .await + .unwrap() + .unwrap(); + assert_eq!(fetched, item); + assert_eq!(attempts, 1); + store.ack_work_item(&token, None).await.unwrap(); + pool.close().await; + } +} + #[tokio::test] async fn test_sqlite_provider_basic() { // Create in-memory SQLite store From e3743b2714c7276f2a168a65a17f6e0448f642aa Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Fri, 30 Oct 2026 15:31:41 +0000 Subject: [PATCH 09/11] docs: define terminal admission failure semantics Keep Defer as the explicit signal for temporary resource shortages. Admission errors, panics, and zero delays intentionally produce non-retryable application failures rather than automatic worker retries, which could repeat invalid input or faulty configuration indefinitely. Verify the exact durable ActivityFailed classification for all three cases. Ref: #74 (74-S6) Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- docs/ORCHESTRATION-GUIDE.md | 18 +++++++++++++----- src/runtime/mod.rs | 10 +++++++--- tests/activity_admission_tests.rs | 19 ++++++++++++++++++- 3 files changed, 38 insertions(+), 9 deletions(-) diff --git a/docs/ORCHESTRATION-GUIDE.md b/docs/ORCHESTRATION-GUIDE.md index 47ce680..9d97144 100644 --- a/docs/ORCHESTRATION-GUIDE.md +++ b/docs/ORCHESTRATION-GUIDE.md @@ -295,11 +295,19 @@ Failures loading activity metadata before admission also consume attempts. Deferral is not a retry of an executed activity. An admission error, panic, or zero delay fails the activity rather than invoking -the handler and is reported as an application error. Storage errors during -deferral are infrastructure errors; an immediate fallback abandon preserves the -same delay and attempt-restoration flag. If that also fails, recovery waits for -the work lock to expire, which can happen before the requested delay. A requested -delay is not a durable "not before" guarantee until its abandon commits. +the handler. These deliberately use the same durable `ActivityFailed` result as +handler errors, with application classification and `retryable: false`. +`Defer` is the retry signal for temporary capacity shortages; `Err` means a +permanent rejection, while panic and zero delay indicate faulty admission logic. +Automatically retrying those failures as infrastructure backpressure could +repeat invalid input or broken configuration indefinitely. The orchestration +can catch the activity failure and choose its own recovery. + +Storage errors during deferral are infrastructure errors; an immediate fallback +abandon preserves the same delay and attempt-restoration flag. If that also +fails, recovery waits for the work lock to expire, which can happen before the +requested delay. A requested delay is not a durable "not before" guarantee until +its abandon commits. Deferrals whose abandon returns success increment `duroxide_activity_admission_deferrals_total`, not the activity execution counter; a commit whose response is lost can restore the attempt without being counted. diff --git a/src/runtime/mod.rs b/src/runtime/mod.rs index 3461a18..221f362 100644 --- a/src/runtime/mod.rs +++ b/src/runtime/mod.rs @@ -497,9 +497,13 @@ impl ActivityAdmission { pub trait ActivityHandler: Send + Sync { /// Reserve resources without blocking or performing I/O. /// - /// Called again on each deferred fetch, outside orchestration replay. An - /// error rejects the activity as an application failure. A returned guard - /// must release its reservation on drop, including cancellation and panic. + /// Called again on each deferred fetch, outside orchestration replay. Use + /// `Defer` for temporary capacity shortages. An error, panic, or zero-delay + /// decision produces a durable, non-retryable application failure without + /// invoking the handler; the worker does not retry faulty admission logic + /// as infrastructure backpressure. An orchestration can handle that failure. + /// A returned guard must release its reservation on drop, including + /// cancellation and panic. /// /// # Errors /// diff --git a/tests/activity_admission_tests.rs b/tests/activity_admission_tests.rs index 8baa300..0fc225c 100644 --- a/tests/activity_admission_tests.rs +++ b/tests/activity_admission_tests.rs @@ -834,7 +834,7 @@ async fn rejected_panicking_and_invalid_admission_never_invokes_handler() { ) .build(); let runtime = Runtime::start_with_options(store.clone(), activities, orchestrations(), options()).await; - let client = Client::new(store); + let client = Client::new(store.clone()); for (instance, message) in [ ("rejected", "invalid admission configuration"), ("panic", "expected admission panic"), @@ -849,6 +849,23 @@ async fn rejected_panicking_and_invalid_admission_never_invokes_handler() { matches!(status, OrchestrationStatus::Failed { ref details, .. } if details.display_message().contains(message)), "unexpected status: {status:?}" ); + let history = store.read(instance).await.unwrap(); + let failures: Vec<_> = history + .iter() + .filter_map(|event| match &event.kind { + EventKind::ActivityFailed { details, .. } => Some(details), + _ => None, + }) + .collect(); + assert_eq!(failures.len(), 1, "admission rejection is a terminal activity result"); + assert!(matches!( + failures[0], + duroxide::ErrorDetails::Application { + kind: duroxide::AppErrorKind::ActivityFailed, + retryable: false, + .. + } + )); } runtime.clone().shutdown(None).await; let metrics = runtime.metrics_snapshot().unwrap(); From 80483d1dfcbcf2bf9847e5a5f2b0a887ba6ce188 Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Fri, 30 Oct 2026 15:47:58 +0000 Subject: [PATCH 10/11] fix: complete activity metadata diagnostics and telemetry Instrument lightweight and execution-history metadata reads with shared latency and error metrics. Log metadata failures separately from acknowledgments and deferrals, and remove the unreachable successful infrastructure-error outcome. Cover ordinary and admission-controlled activities, transient recovery, persistent failures, actual metric labels and values. Document metadata failure behavior and per-attempt history fallback costs, add admission deferrals to both telemetry specifications, and correct the remaining visibility-order example. Ref: #74 (74-S4, 74-N3, 74-N4) Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 11 ++- docs/duroxide-telemetry-spec.md | 30 ++++++ docs/metrics-specification.md | 27 ++++++ docs/provider-implementation-guide.md | 12 ++- src/providers/instrumented.rs | 63 ++++++++----- src/providers/mod.rs | 9 +- src/runtime/dispatchers/worker.rs | 59 ++++++++---- tests/activity_admission_tests.rs | 64 ++++++++++--- tests/observability_tests.rs | 131 +++++++++++++++++++++++++- 9 files changed, 343 insertions(+), 63 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b1d0d78..a346448 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,8 +30,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 be upgraded to provide this behavior in a mixed-version deployment. (#74) - Activity metadata uses the lightweight instance lookup for the current execution instead of re-reading its full history on every attempt. Activities - from earlier executions read their own execution's metadata. Failed deferral - retries preserve the requested delay and attempt-restoration flag. (#74) + from earlier executions read their own execution's metadata. Both lookup paths + record provider-operation metrics. If the execution-history fallback fails, + processing now abandons the fetch with its attempt counted and reports an + infrastructure error instead of invoking the activity with `unknown` metadata. + This applies to ordinary and admission-controlled activities; repeated read + failures can exhaust the fetch budget before either handler runs. Metadata + failures are logged separately from acknowledgment and deferral failures. (#74) +- Failed deferral retries preserve the requested delay and attempt-restoration + flag. (#74) - Relative SQLite queue delays saturate instead of overflowing their timestamp representation. (#74) diff --git a/docs/duroxide-telemetry-spec.md b/docs/duroxide-telemetry-spec.md index 28755ab..7f15b9a 100644 --- a/docs/duroxide-telemetry-spec.md +++ b/docs/duroxide-telemetry-spec.md @@ -184,6 +184,31 @@ by (activity_name) --- +#### `duroxide_activity_admission_deferrals_total` (Counter) + +**Description:** Admission deferrals whose abandon returned success, including +a successful fallback after an abandon error. + +**Labels:** +- `activity_name` - Registered activity name +- `tag` - Routing tag, or `"default"` for untagged work + +**Semantics:** This counts observed successful deferrals, not handler executions. +Deferrals do not contribute activity execution counts or duration samples. +A commit with a lost response may not be counted. Abandon failures count as +`infra_error`; admission `Err`, panic, and zero delay produce terminal +`ActivityFailed` application results and count as `app_error`. + +The in-process accessor is `MetricsProvider::activity_admission_deferrals()`, +without adding fields to the existing public `MetricsSnapshot`. + +**Example:** +```promql +sum by (activity_name, tag) (rate(duroxide_activity_admission_deferrals_total[5m])) +``` + +--- + ### 1.3 Provider (Storage) Metrics #### `duroxide_provider_operation_duration_seconds` (Histogram) @@ -196,6 +221,11 @@ by (activity_name) **Use Case:** Database performance monitoring, identify slow queries +Activity metadata is included with `operation="get_instance_info"` for the +lightweight lookup and `operation="read_with_execution"` for the history +fallback, each with `status="success"` or `"error"`. Errors also increment +`duroxide_provider_errors_total` with the same operation label. + **Example:** ```promql # p95 database operation latency diff --git a/docs/metrics-specification.md b/docs/metrics-specification.md index 426fa46..99ad69d 100644 --- a/docs/metrics-specification.md +++ b/docs/metrics-specification.md @@ -28,6 +28,7 @@ This document provides a complete reference of all metrics emitted by duroxide v | `duroxide_worker_queue_depth` | Gauge | Runtime | _(none)_ | ✅ `test_queue_depth_gauges_tracking` | ✅ Code Review | Observable gauge with callback, initialized via `initialize_gauges()` | | **Activity Execution** | | `duroxide_activity_executions_total` | Counter | Activity | `activity_name`, `outcome`, `retry_attempt`, `tag` | ✅ `metrics_capture_activity_and_orchestration_outcomes` | ✅ Code Review | Tracks all outcomes including app_error | +| `duroxide_activity_admission_deferrals_total` | Counter | Activity | `activity_name`, `tag` | ✅ `admission_deferral_metric_has_activity_and_tag_labels` | ✅ Recorder test | Counts observed successful deferral abandons, not executions | | `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 | @@ -461,6 +462,29 @@ topk(5, --- +### 2.5 `duroxide_activity_admission_deferrals_total` (Counter) + +**Description:** Admission deferrals whose abandon operation returned success, +including a successful fallback after an abandon error. + +**Labels:** +- `activity_name` (string) - Activity name +- `tag` (string) - Activity routing tag, or `"default"` for untagged work + +**Semantics:** Deferrals do not count as activity executions or produce activity +duration samples. Failed abandons count as infrastructure errors; a storage +commit whose response is lost may not increment this counter. Admission `Err`, +panic, and zero delay instead produce terminal application failures (`app_error`). + +For in-process diagnostics, use `MetricsProvider::activity_admission_deferrals()`; +the public `MetricsSnapshot` shape is unchanged. + +```promql +sum by (activity_name, tag) (rate(duroxide_activity_admission_deferrals_total[5m])) +``` + +--- + ## 3. Sub-Orchestration Metrics > **Session Metrics (Future):** Session-specific metrics (session claims, migrations, @@ -531,6 +555,8 @@ histogram_quantile(0.95, - `"save_event"` - Save event to history - `"create_instance"` - Create new orchestration instance - `"query"` - Query operation + - `"get_instance_info"` - Lightweight activity metadata lookup + - `"read_with_execution"` - Execution-history metadata fallback - `status` (string) - Operation status: - `"success"` - Operation succeeded - `"error"` - Operation failed @@ -566,6 +592,7 @@ histogram_quantile(0.99, - `"deadlock"` - Database deadlock - `"corruption"` - Data corruption - `"constraint_violation"` - Database constraint violation + - `"other"` - Errors not classified as timeout, connection, or deadlock **Purpose:** Database health monitoring, alerting on infrastructure issues diff --git a/docs/provider-implementation-guide.md b/docs/provider-implementation-guide.md index 9e5c939..ed71f4a 100644 --- a/docs/provider-implementation-guide.md +++ b/docs/provider-implementation-guide.md @@ -748,7 +748,7 @@ async fn fetch_work_item( WHERE id = ( SELECT id FROM worker_queue WHERE visible_at <= ? AND (lock_token IS NULL OR locked_until <= ?) - ORDER BY id LIMIT 1 + ORDER BY visible_at, id LIMIT 1 ) RETURNING work_item, attempt_count, instance_id", lock_token, locked_until, now, now @@ -1457,10 +1457,12 @@ that a handler actually began executing across a process crash. The worker uses `ProviderAdmin::get_instance_info` for lightweight activity metadata when it describes the activity's execution. Providers without that capability, and activities from an earlier execution, fall back to -`read_with_execution`. Metadata must describe the requested execution, not a -newer continue-as-new execution. If the execution-history read fails, the worker -abandons the fetch without restoring its attempt and does not invoke admission -or the handler. +`read_with_execution`, loading the full execution history on every attempt, +including repeated admission deferrals. There is no descriptor cache. +Both lookup paths record provider-operation latency and errors. Metadata must +describe the requested execution, not a newer continue-as-new execution. If the +execution-history read fails, the worker abandons the fetch without restoring +its attempt and does not invoke admission or the handler. ### ack_work_item() — Complete Implementation diff --git a/src/providers/instrumented.rs b/src/providers/instrumented.rs index 459ad93..a0fb4a0 100644 --- a/src/providers/instrumented.rs +++ b/src/providers/instrumented.rs @@ -14,6 +14,40 @@ use super::{ use crate::Event; use crate::runtime::observability::MetricsProvider; +fn record_provider_error(metrics: &MetricsProvider, operation: &str, error: &ProviderError) { + let error_type = if error.message.contains("deadlock") { + "deadlock" + } else if error.message.contains("timeout") { + "timeout" + } else if error.message.contains("connection") { + "connection" + } else { + "other" + }; + metrics.record_provider_error(operation, error_type); +} + +// Management capabilities bypass the wrapper, so their callers share this instrumentation. +pub(crate) async fn instrument_provider_operation( + metrics: Option<&MetricsProvider>, + operation: &str, + future: impl std::future::Future>, +) -> Result { + let start = std::time::Instant::now(); + let result = future.await; + if let Some(metrics) = metrics { + metrics.record_provider_operation( + operation, + start.elapsed().as_secs_f64(), + if result.is_ok() { "success" } else { "error" }, + ); + if let Err(error) = &result { + record_provider_error(metrics, operation, error); + } + } + result +} + /// Wrapper that adds metrics instrumentation to any Provider implementation. /// /// This follows the decorator pattern to automatically record: @@ -61,16 +95,7 @@ impl InstrumentedProvider { #[inline] fn record_error(&self, operation: &str, error: &ProviderError) { if let Some(ref m) = self.metrics { - let error_type = if error.message.contains("deadlock") { - "deadlock" - } else if error.message.contains("timeout") { - "timeout" - } else if error.message.contains("connection") { - "connection" - } else { - "other" - }; - m.record_provider_error(operation, error_type); + record_provider_error(m, operation, error); } } } @@ -159,20 +184,16 @@ impl Provider for InstrumentedProvider { } async fn read(&self, instance: &str) -> Result, ProviderError> { - let start = std::time::Instant::now(); - let result = self.inner.read(instance).await; - let duration = start.elapsed(); - - self.record_operation("read", duration, if result.is_ok() { "success" } else { "error" }); - if let Err(ref e) = result { - self.record_error("read", e); - } - - result + instrument_provider_operation(self.metrics.as_deref(), "read", self.inner.read(instance)).await } async fn read_with_execution(&self, instance: &str, execution_id: u64) -> Result, ProviderError> { - self.inner.read_with_execution(instance, execution_id).await + instrument_provider_operation( + self.metrics.as_deref(), + "read_with_execution", + self.inner.read_with_execution(instance, execution_id), + ) + .await } async fn append_with_execution( diff --git a/src/providers/mod.rs b/src/providers/mod.rs index ce74636..9c4c2a1 100644 --- a/src/providers/mod.rs +++ b/src/providers/mod.rs @@ -1606,13 +1606,14 @@ pub trait Provider: Any + Send + Sync { /// The runtime hot path uses `fetch_orchestration_item()` which loads history internally. async fn read(&self, instance: &str) -> Result, ProviderError>; - // ===== History Reading and Testing Methods (NOT used by runtime hot path) ===== - // These methods are NOT called by the main runtime during orchestration execution. - // They are used by testing, validation suites, client APIs, and debugging tools. + // ===== History Reading and Testing Methods ===== + // Execution-history reads also support activity metadata fallback. /// Read the full event history for a specific execution within an instance. /// - /// **NOT called in runtime hot path.** Used by testing, validation suites, and debugging tools. + /// Also used for activity metadata when management metadata is unavailable or + /// describes a different execution. In that case the worker reads the full + /// history on every attempt, including repeated admission deferrals. /// /// # Parameters /// diff --git a/src/runtime/dispatchers/worker.rs b/src/runtime/dispatchers/worker.rs index e0a7ce8..d7f0141 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; +use crate::providers::{WorkItem, instrumented::instrument_provider_operation}; use std::sync::Arc; use std::time::Duration; use tokio::task::JoinHandle; @@ -125,8 +125,6 @@ enum ActivityOutcome { Cancelled, /// Admission deferred without executing the handler Deferred(Duration), - /// Activity setup failed before invocation - InfraError, } enum ActivityInvocation { @@ -494,7 +492,15 @@ async fn execute_activity(rt: &Arc, activities: &Arc activity_ctx, Err(error) => { - handle_activity_outcome(rt, &ctx, Err(error), ActivityOutcome::InfraError).await; + warn!( + instance = %ctx.instance, + execution_id = ctx.execution_id, + activity_id = ctx.activity_id, + worker_id = %ctx.worker_id, + %error, + "worker: activity metadata read failed, abandoning work item" + ); + abandon_after_activity_failure(rt, &ctx, Duration::from_millis(100), false).await; return; } }; @@ -531,7 +537,13 @@ async fn build_activity_context( cancellation_token: CancellationToken, ) -> Result { let metadata = if let Some(admin) = rt.history_store.as_management_capability() { - match admin.get_instance_info(&ctx.instance).await { + match instrument_provider_operation( + rt.metrics_provider(), + "get_instance_info", + admin.get_instance_info(&ctx.instance), + ) + .await + { Ok(info) if info.current_execution_id == ctx.execution_id && info.orchestration_version != "unknown" => { Some((info.orchestration_name, info.orchestration_version)) } @@ -847,7 +859,6 @@ async fn handle_activity_outcome( ActivityOutcome::Deferred(_) => { rt.record_activity_admission_deferral(&ctx.activity_name, ctx.tag.as_deref()) } - ActivityOutcome::InfraError => rt.record_activity_infra_error(), ActivityOutcome::Cancelled => {} }, Err(e) => { @@ -864,25 +875,37 @@ async fn handle_activity_outcome( ActivityOutcome::Deferred(delay) => (delay, true), _ => (Duration::from_millis(100), false), }; - if let Err(error) = rt - .history_store - .abandon_work_item(&ctx.lock_token, Some(delay), ignore_attempt) - .await + if abandon_after_activity_failure(rt, ctx, delay, ignore_attempt).await + && matches!(outcome, ActivityOutcome::Deferred(_)) { - warn!( - instance = %ctx.instance, - activity_id = ctx.activity_id, - error = %error, - "Failed to abandon activity work item; waiting for lock expiry" - ); - } else if matches!(outcome, ActivityOutcome::Deferred(_)) { rt.record_activity_admission_deferral(&ctx.activity_name, ctx.tag.as_deref()); } - rt.record_activity_infra_error(); } } } +async fn abandon_after_activity_failure( + rt: &Arc, + ctx: &ActivityWorkContext, + delay: Duration, + ignore_attempt: bool, +) -> bool { + let result = rt + .history_store + .abandon_work_item(&ctx.lock_token, Some(delay), ignore_attempt) + .await; + if let Err(error) = &result { + warn!( + instance = %ctx.instance, + activity_id = ctx.activity_id, + %error, + "Failed to abandon activity work item; waiting for lock expiry" + ); + } + rt.record_activity_infra_error(); + result.is_ok() +} + // ============================================================================ // Activity Manager (Lock Renewal + Cancellation Detection) // ============================================================================ diff --git a/tests/activity_admission_tests.rs b/tests/activity_admission_tests.rs index 0fc225c..1438dbd 100644 --- a/tests/activity_admission_tests.rs +++ b/tests/activity_admission_tests.rs @@ -585,15 +585,32 @@ async fn shutdown_after_deferral_commit_preserves_attempt_refund_without_a_respo #[tokio::test(flavor = "current_thread")] async fn metadata_read_failure_consumes_an_attempt_before_admission() { - for max_attempts in [1, 2] { + for (max_attempts, read_failures, admission_controlled) in [ + (1, 1, false), + (1, 1, true), + (2, 1, false), + (2, 1, true), + (2, 2, false), + (2, 2, true), + ] { + let (events, _guard) = common::tracing_capture::install_tracing_capture(); let sqlite = Arc::new(SqliteProvider::new_in_memory().await.unwrap()); let store = Arc::new(PoisonInjectingProvider::new(sqlite.clone())); store.hide_management_capability(); - store.fail_execution_history_reads(1); + store.fail_execution_history_reads(read_failures); let admissions = Arc::new(AtomicUsize::new(0)); let callback_admissions = admissions.clone(); - let activities = ActivityRegistry::builder() - .register_with_admission( + let handler_calls = Arc::new(AtomicUsize::new(0)); + let calls = handler_calls.clone(); + let handler = move |ctx: ActivityContext, input: String| { + calls.fetch_add(1, Ordering::SeqCst); + assert_eq!(ctx.orchestration_name(), "Run"); + assert_eq!(ctx.orchestration_version(), "1.0.0"); + async move { Ok(input) } + }; + let builder = ActivityRegistry::builder(); + let activities = if admission_controlled { + builder.register_with_admission( "Limited", move |ctx, _input| { callback_admissions.fetch_add(1, Ordering::SeqCst); @@ -601,9 +618,12 @@ async fn metadata_read_failure_consumes_an_attempt_before_admission() { assert_eq!(ctx.orchestration_version(), "1.0.0"); Ok(ActivityAdmission::admit(())) }, - |_ctx, input| async move { Ok(input) }, + handler, ) - .build(); + } else { + builder.register("Limited", handler) + } + .build(); let runtime = Runtime::start_with_options( store.clone(), activities, @@ -624,24 +644,44 @@ async fn metadata_read_failure_consumes_an_attempt_before_admission() { .await .unwrap(); runtime.clone().shutdown(None).await; - if max_attempts == 1 { + if read_failures == max_attempts { assert!(matches!( status, OrchestrationStatus::Failed { - details: duroxide::ErrorDetails::Poison { attempt_count: 2, .. }, + details: duroxide::ErrorDetails::Poison { attempt_count, .. }, .. - } + } if attempt_count == max_attempts + 1 )); assert_eq!(admissions.load(Ordering::SeqCst), 0); + assert_eq!(handler_calls.load(Ordering::SeqCst), 0); } else { assert!(matches!(status, OrchestrationStatus::Completed { output, .. } if output == "payload")); - assert_eq!(admissions.load(Ordering::SeqCst), 1); + assert_eq!(admissions.load(Ordering::SeqCst), usize::from(admission_controlled)); + assert_eq!(handler_calls.load(Ordering::SeqCst), 1); } - assert_eq!(store.work_abandons(), vec![(Some(Duration::from_millis(100)), false)]); + let expected_failures = usize::try_from(read_failures).unwrap(); + assert_eq!( + store.work_abandons(), + vec![(Some(Duration::from_millis(100)), false); expected_failures] + ); let metrics = runtime.metrics_snapshot().unwrap(); - assert_eq!(metrics.activity_infra_errors, 1); + assert_eq!(metrics.activity_infra_errors, u64::from(read_failures)); + assert_eq!(metrics.provider_errors, u64::from(read_failures)); assert_eq!(metrics.activity_app_errors, 0); assert_eq!(admission_deferrals(&runtime), 0); + let events = events.lock().unwrap(); + assert_eq!( + events + .iter() + .filter(|event| event.message == "worker: activity metadata read failed, abandoning work item") + .count(), + expected_failures + ); + assert!( + !events + .iter() + .any(|event| event.message.contains("activity acknowledgment or deferral failed")) + ); } } diff --git a/tests/observability_tests.rs b/tests/observability_tests.rs index 8398acc..199f062 100644 --- a/tests/observability_tests.rs +++ b/tests/observability_tests.rs @@ -8,7 +8,7 @@ mod common; -use common::fault_injection::FailingProvider; +use common::fault_injection::{FailingProvider, PoisonInjectingProvider}; use common::tracing_capture::CapturedEvent; use duroxide::providers::sqlite::SqliteProvider; use duroxide::providers::{Provider, WorkItem}; @@ -16,6 +16,7 @@ use duroxide::runtime; use duroxide::runtime::registry::ActivityRegistry; use duroxide::runtime::{LogFormat, ObservabilityConfig, RuntimeOptions, UnregisteredBackoffConfig}; use duroxide::{ActivityContext, Client, OrchestrationContext, OrchestrationRegistry, OrchestrationStatus}; +use metrics_util::debugging::{DebugValue, DebuggingRecorder}; use std::sync::Arc; use std::time::Duration; use tracing::Level; @@ -1079,6 +1080,134 @@ async fn test_versioned_orchestration_metrics() { // but not exposed in atomic counter snapshot } +#[tokio::test(flavor = "current_thread")] +async fn activity_metadata_provider_operations_are_instrumented() { + for history_fallback in [false, true] { + let recorder = DebuggingRecorder::new(); + let _recorder_guard = metrics::set_default_local_recorder(&recorder); + let sqlite = Arc::new(SqliteProvider::new_in_memory().await.unwrap()); + let provider = Arc::new(PoisonInjectingProvider::new(sqlite.clone())); + if history_fallback { + provider.hide_management_capability(); + provider.fail_execution_history_reads(1); + } + let activities = ActivityRegistry::builder() + .register("Activity", |ctx: ActivityContext, input: String| async move { + assert_eq!(ctx.orchestration_name(), "MetadataOrch"); + assert_eq!(ctx.orchestration_version(), "1.0.0"); + Ok(input) + }) + .build(); + let orchestrations = OrchestrationRegistry::builder() + .register("MetadataOrch", |ctx: OrchestrationContext, input: String| async move { + ctx.schedule_activity("Activity", input).await + }) + .build(); + let rt = runtime::Runtime::start_with_options( + provider, + activities, + orchestrations, + RuntimeOptions { + worker_concurrency: 1, + orchestration_concurrency: 1, + max_attempts: 2, + dispatcher_min_poll_interval: Duration::from_millis(5), + ..Default::default() + }, + ) + .await; + let client = Client::new(sqlite); + client + .start_orchestration("metadata-metrics", "MetadataOrch", "payload") + .await + .unwrap(); + let status = client + .wait_for_orchestration("metadata-metrics", Duration::from_secs(5)) + .await; + rt.clone().shutdown(None).await; + assert!(matches!( + status.unwrap(), + OrchestrationStatus::Completed { output, .. } if output == "payload" + )); + + let captured = recorder.snapshotter().snapshot().into_vec(); + let assert_duration = |operation: &str, status: &str| { + let (key, _, _, value) = captured + .iter() + .find(|(key, _, _, _)| { + key.key().name() == "duroxide_provider_operation_duration_seconds" + && key + .key() + .labels() + .any(|label| label.key() == "operation" && label.value() == operation) + && key + .key() + .labels() + .any(|label| label.key() == "status" && label.value() == status) + }) + .unwrap_or_else(|| panic!("missing duration metric for {operation}/{status}")); + assert_eq!(key.key().labels().count(), 2); + assert!(matches!(value, DebugValue::Histogram(samples) if samples.len() == 1)); + }; + if history_fallback { + assert_duration("read_with_execution", "error"); + assert_duration("read_with_execution", "success"); + let (key, _, _, value) = captured + .iter() + .find(|(key, _, _, _)| { + key.key().name() == "duroxide_provider_errors_total" + && key + .key() + .labels() + .any(|label| label.key() == "operation" && label.value() == "read_with_execution") + }) + .expect("metadata read failure must emit a provider error counter"); + assert!( + key.key() + .labels() + .any(|label| label.key() == "error_type" && label.value() == "other") + ); + assert_eq!(value, &DebugValue::Counter(1)); + assert_eq!(rt.metrics_snapshot().unwrap().provider_errors, 1); + } else { + assert_duration("get_instance_info", "success"); + assert_eq!(rt.metrics_snapshot().unwrap().provider_errors, 0); + } + } +} + +#[test] +fn admission_deferral_metric_has_activity_and_tag_labels() { + let recorder = DebuggingRecorder::new(); + let _recorder_guard = metrics::set_default_local_recorder(&recorder); + let metrics = runtime::observability::MetricsProvider::new(&ObservabilityConfig::default()).unwrap(); + metrics.record_activity_admission_deferral("Limited", Some("network")); + metrics.record_activity_admission_deferral("Limited", None); + + let captured = recorder.snapshotter().snapshot().into_vec(); + assert_eq!(captured.len(), 2); + for tag in ["network", "default"] { + let (key, _, _, value) = captured + .iter() + .find(|(key, _, _, _)| { + key.key().name() == "duroxide_activity_admission_deferrals_total" + && key + .key() + .labels() + .any(|label| label.key() == "tag" && label.value() == tag) + }) + .expect("missing admission deferral counter"); + assert_eq!(key.key().labels().count(), 2); + assert!( + key.key() + .labels() + .any(|label| label.key() == "activity_name" && label.value() == "Limited") + ); + assert_eq!(value, &DebugValue::Counter(1)); + } + assert_eq!(metrics.activity_admission_deferrals(), 2); +} + #[tokio::test(flavor = "current_thread")] async fn test_provider_metrics_recorded() { // Test that provider operation metrics are recorded via InstrumentedProvider From 112dcf02945829863f8514e963828d9af2e854b0 Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Fri, 30 Oct 2026 18:06:09 +0000 Subject: [PATCH 11/11] docs: clarify admission retries and typed registration Explain orchestration retry helpers rescheduling admission failures despite retryable: false, and show how to floor computed delays that reach zero. Add a compiled ten-line typed semaphore example, enumerate the public admission APIs and failure behavior in the changelog, and align the deferral metric wording with observed abandon success. Ref: #74 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 13 ++++++++++--- docs/ORCHESTRATION-GUIDE.md | 21 +++++++++++++++++++++ docs/observability-guide.md | 2 +- 3 files changed, 32 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a346448..91d9615 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,15 +9,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- Opt-in activity resource admission through `register_with_admission` and - `register_typed_with_admission` for typed input and output. +- Opt-in activity resource admission through `ActivityAdmission`, the default + `ActivityHandler::try_admit` method, and the `register_with_admission` and + `register_typed_with_admission` registration helpers. Admission can retain a guard for the handler's lifetime or durably defer execution without occupying a worker slot or recording an activity result. Positive retry delays are clamped to 1 ms–24 hours. Only committed deferrals restore their fetch attempt; failed or interrupted deferrals may still use the poison-message budget, even when the handler never starts. Uncommitted deferral delays may be lost on lock expiry. Successfully observed deferrals - have a separate `duroxide_activity_admission_deferrals_total` metric. (#74) + have a separate `duroxide_activity_admission_deferrals_total` metric and the + `MetricsProvider::activity_admission_deferrals()` accessor for in-process + counts without changing the public `MetricsSnapshot` shape. (#74) +- Admission errors, callback panics, and zero retry delays produce terminal + `ActivityFailed` application results with `retryable: false`, without invoking + the handler. Orchestration-level retry helpers still retry these errors + according to their policy. (#74) ### Fixed diff --git a/docs/ORCHESTRATION-GUIDE.md b/docs/ORCHESTRATION-GUIDE.md index 9d97144..9458358 100644 --- a/docs/ORCHESTRATION-GUIDE.md +++ b/docs/ORCHESTRATION-GUIDE.md @@ -283,8 +283,26 @@ Invalid input fails before either callback runs. Input is decoded for each admission check and again when invoking an admitted handler, so deserialization must be side-effect-free. +Using the imports above, the same semaphore limit can protect a typed activity: + +```rust +let capacity = Arc::new(Semaphore::new(4)); +let activities = ActivityRegistry::builder().register_typed_with_admission( + "Limited", + move |_ctx, _input: &u32| match capacity.clone().try_acquire_owned() { + Ok(permit) => Ok(ActivityAdmission::admit(permit)), + Err(TryAcquireError::NoPermits) => Ok(ActivityAdmission::Defer(Duration::from_millis(100))), + Err(error) => Err(error.to_string()), + }, + |_ctx, input: u32| async move { Ok(input) }, +).build(); +``` + `Defer(delay)` requires a positive delay. The runtime clamps it to 1 ms through 24 hours (`ActivityAdmission::MAX_DEFER_DELAY`) before passing it to any provider. +Computed delays such as `deadline.saturating_duration_since(now)` can become zero +and fail that activity attempt; use `.max(Duration::from_millis(1))` when zero +should mean retry soon. Deferral leaves the original activity pending in durable storage, frees the worker to fetch other work, and does not record an activity result. A **committed** deferral restores its fetch attempt. Crashes, shutdown aborts, @@ -302,6 +320,9 @@ permanent rejection, while panic and zero delay indicate faulty admission logic. Automatically retrying those failures as infrastructure backpressure could repeat invalid input or broken configuration indefinitely. The orchestration can catch the activity failure and choose its own recovery. +`schedule_activity_with_retry` receives only an error message, so it retries +admission failures regardless of `retryable: false`, scheduling fresh attempts +and rerunning admission up to the policy's attempt limit. Storage errors during deferral are infrastructure errors; an immediate fallback abandon preserves the same delay and attempt-restoration flag. If that also diff --git a/docs/observability-guide.md b/docs/observability-guide.md index e96ae88..92c5b58 100644 --- a/docs/observability-guide.md +++ b/docs/observability-guide.md @@ -195,7 +195,7 @@ All metrics follow Prometheus naming conventions with `_total` for counters and | `duroxide_active_orchestrations` | Gauge | Orchestration | _(none)_ | **Current** active count | | **Activity Execution** | | `duroxide_activity_executions_total` | Counter | Activity | activity_name, outcome, retry_attempt | Track success/failure rates | -| `duroxide_activity_admission_deferrals_total` | Counter | Activity | activity_name, tag | Track committed resource-admission deferrals | +| `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 | | **Sub-Orchestration** |