diff --git a/CHANGELOG.md b/CHANGELOG.md index bf5cd12..91d9615 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,48 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- 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 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 + +- 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) +- 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. 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) + ## [0.1.32] - 2026-10-11 **Release:** diff --git a/docs/ORCHESTRATION-GUIDE.md b/docs/ORCHESTRATION-GUIDE.md index 9a89393..9458358 100644 --- a/docs/ORCHESTRATION-GUIDE.md +++ b/docs/ORCHESTRATION-GUIDE.md @@ -242,6 +242,106 @@ 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. + +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. + +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, +and failed abandons can still consume the poison-message attempt budget, so +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. 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. +`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 +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. +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. 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 @@ -2693,4 +2793,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/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/observability-guide.md b/docs/observability-guide.md index f371657..92c5b58 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 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** | @@ -265,6 +266,23 @@ All metrics follow Prometheus naming conventions with `_total` for counters and #### Activity Execution Metrics +**`duroxide_activity_admission_deferrals_total`** (Counter) +- **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 + +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. +Uncommitted deferrals do not increment the deferral counter and may consume a +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 - **Labels**: `activity_name`, `outcome` (success|app_error|infra_error|config_error), `retry_attempt` (0|1|2|3+) @@ -645,4 +663,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..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 @@ -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,28 @@ 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. +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`, 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 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 008f5c2..de476b2 100644 --- a/docs/sqlite-provider-design.md +++ b/docs/sqlite-provider-design.md @@ -101,6 +101,17 @@ 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 +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. Migration `20240113000000` creates the index for new and +existing file-backed databases; in-memory databases create it directly. + ## Implementation Plan ### Phase 1: Core Provider Structure 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/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/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 813631b..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 /// @@ -1727,6 +1728,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 4240f38..05c20cd 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() }; @@ -426,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)", ) @@ -505,9 +507,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. @@ -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 "#, ); @@ -2242,8 +2245,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 +4349,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; diff --git a/src/runtime/dispatchers/worker.rs b/src/runtime/dispatchers/worker.rs index 8bb1124..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; @@ -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(Duration), +} + +enum ActivityInvocation { + Completed(String), + Deferred(Duration), } /// Context for processing a single activity work item. @@ -482,18 +489,21 @@ async fn execute_activity(rt: &Arc, activities: &Arc activity_ctx, + Err(error) => { + 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; + } + }; let start_time = std::time::Instant::now(); @@ -525,13 +535,59 @@ 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 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)) + } + 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, @@ -543,7 +599,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!`. @@ -557,19 +613,58 @@ 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 { 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.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 + .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(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, } @@ -761,6 +856,9 @@ async fn handle_activity_outcome( Ok(()) => match outcome { ActivityOutcome::Success => rt.record_activity_success(), ActivityOutcome::AppError => rt.record_activity_app_error(), + ActivityOutcome::Deferred(_) => { + rt.record_activity_admission_deferral(&ctx.activity_name, ctx.tag.as_deref()) + } ActivityOutcome::Cancelled => {} }, Err(e) => { @@ -770,17 +868,44 @@ 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 - .history_store - .abandon_work_item(&ctx.lock_token, Some(Duration::from_millis(100)), false) - .await; - rt.record_activity_infra_error(); + let (delay, ignore_attempt) = match outcome { + ActivityOutcome::Deferred(delay) => (delay, true), + _ => (Duration::from_millis(100), false), + }; + if abandon_after_activity_failure(rt, ctx, delay, ignore_attempt).await + && matches!(outcome, ActivityOutcome::Deferred(_)) + { + rt.record_activity_admission_deferral(&ctx.activity_name, ctx.tag.as_deref()); + } } } } +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/src/runtime/mod.rs b/src/runtime/mod.rs index 268fda7..221f362 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, @@ -455,9 +457,62 @@ 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 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 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), +} + +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)) + } +} + /// 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. 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 + /// + /// 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; } @@ -658,6 +713,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..00def86 100644 --- a/src/runtime/observability.rs +++ b/src/runtime/observability.rs @@ -159,6 +159,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 +198,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 +460,23 @@ 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); + } + + /// 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); diff --git a/src/runtime/registry.rs b/src/runtime/registry.rs index 2d0f905..049ed19 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,43 @@ 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, +} + +#[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 +490,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, @@ -463,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 @@ -486,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 new file mode 100644 index 0000000..1438dbd --- /dev/null +++ b/tests/activity_admission_tests.rs @@ -0,0 +1,967 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::clone_on_ref_ptr)] + +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, AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +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 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 { + ctx.schedule_activity(activity, "payload").await + }) + .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; + 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.clone().shutdown(None).await; + assert!(admission_deferrals(&runtime) >= 20); + + 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); + // 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" + )); + 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_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 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!(admission_deferrals(&runtime), 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!(admission_deferrals(&runtime), 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 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!(admission_deferrals(&runtime), 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!(admission_deferrals(&runtime), 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, 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(read_failures); + let admissions = Arc::new(AtomicUsize::new(0)); + let callback_admissions = admissions.clone(); + 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); + assert_eq!(ctx.orchestration_name(), "Run"); + assert_eq!(ctx.orchestration_version(), "1.0.0"); + Ok(ActivityAdmission::admit(())) + }, + handler, + ) + } else { + builder.register("Limited", handler) + } + .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 read_failures == max_attempts { + assert!(matches!( + status, + OrchestrationStatus::Failed { + 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), usize::from(admission_controlled)); + assert_eq!(handler_calls.load(Ordering::SeqCst), 1); + } + 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, 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")) + ); + } +} + +#[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; + 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 (events, _guard) = common::tracing_capture::install_tracing_capture(); + 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.clone()); + 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:?}" + ); + 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(); + assert!(metrics.activity_app_errors >= 3); + assert_eq!(metrics.activity_infra_errors, 0); + assert_eq!(admission_deferrals(&runtime), 0); + assert!( + !events + .lock() + .unwrap() + .iter() + .any(|event| event.message == "Activity started") + ); +} + +#[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; +} diff --git a/tests/common/fault_injection.rs b/tests/common/fault_injection.rs index 683b1d6..89d6463 100644 --- a/tests/common/fault_injection.rs +++ b/tests/common/fault_injection.rs @@ -48,6 +48,11 @@ pub struct PoisonInjectingProvider { successful_ignored_abandons: AtomicU32, counted_abandon_delays: std::sync::Mutex>, orchestration_fetches: std::sync::Mutex>, + work_abandons: std::sync::Mutex, bool)>>, + work_abandon_failures: AtomicU32, + history_reads: AtomicU32, + execution_history_read_failures: AtomicU32, + hide_management: AtomicBool, } impl PoisonInjectingProvider { @@ -68,6 +73,11 @@ 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()), + work_abandon_failures: AtomicU32::new(0), + history_reads: AtomicU32::new(0), + execution_history_read_failures: AtomicU32::new(0), + hide_management: AtomicBool::new(false), } } @@ -169,6 +179,27 @@ 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() + } + + 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 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); + } } #[async_trait] @@ -341,6 +372,20 @@ impl Provider for PoisonInjectingProvider { delay: Option, ignore_attempt: bool, ) -> Result<(), ProviderError> { + self.work_abandons + .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 } @@ -370,10 +415,22 @@ 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); + 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 } @@ -389,8 +446,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/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( diff --git a/tests/observability_tests.rs b/tests/observability_tests.rs index 809a5a3..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 @@ -1491,6 +1620,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() { @@ -1514,6 +1688,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!(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); } /// Test: MetricsProvider poison message tracking 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}")) }; 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; 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