From b44ee294bc17743ed2c854a381a199ab17eab758 Mon Sep 17 00:00:00 2001 From: DJ Majumdar Date: Thu, 19 Mar 2026 18:53:55 -0700 Subject: [PATCH] refactor: migrate all tests/benches from TaskExecutor to TypedExecutor Eliminate all usage of the untyped `TaskExecutor` + `raw_executor` API from tests, benchmarks, and examples in preparation for removing the untyped public API. - Convert 47 `impl TaskExecutor for` to `impl TypedExecutor for` - Convert 87 `.raw_executor("name", exec)` to `.task::(exec)` - Add `define_task!` macro in common.rs for zero-boilerplate task types - Define ~35 TypedTask unit structs across test/bench files - Make shared test executors generic: `impl TypedExecutor` - Handle missing payloads in `TaskContext::payload()` by deserializing from JSON null (supports unit-struct tasks submitted without payload) No public API changes. All 276 tests pass. --- benches/dependencies.rs | 18 +- benches/groups.rs | 22 +- benches/history.rs | 18 +- benches/retry.rs | 26 +- benches/scheduler.rs | 37 +- benches/tags.rs | 17 +- examples/profile_dep_chain.rs | 21 +- src/module.rs | 23 +- src/registry/context.rs | 12 +- src/registry/mod.rs | 49 +- src/scheduler/tests.rs | 778 ++++++++++++++------------- tests/integration/common.rs | 101 +++- tests/integration/cross_module.rs | 179 +++--- tests/integration/dependencies.rs | 19 +- tests/integration/module_features.rs | 155 ++---- tests/integration/modules.rs | 84 ++- tests/integration/retry_policy.rs | 46 +- tests/integration/scheduler_core.rs | 117 ++-- 18 files changed, 960 insertions(+), 762 deletions(-) diff --git a/benches/dependencies.rs b/benches/dependencies.rs index 46fda7f..3869d82 100644 --- a/benches/dependencies.rs +++ b/benches/dependencies.rs @@ -7,9 +7,10 @@ use std::time::{Duration, Instant}; use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion}; #[cfg(feature = "profile")] use pprof::criterion::{Output, PProfProfiler}; +use serde::{Deserialize, Serialize}; use taskmill::{ - Domain, DomainKey, Scheduler, SchedulerEvent, TaskContext, TaskError, TaskExecutor, TaskStore, - TaskSubmission, + Domain, DomainKey, Scheduler, SchedulerEvent, TaskContext, TaskError, TaskStore, + TaskSubmission, TypedExecutor, TypedTask, }; use tokio::runtime::Runtime; use tokio_util::sync::CancellationToken; @@ -19,10 +20,17 @@ impl DomainKey for BenchDomain { const NAME: &'static str = "bench"; } +#[derive(Serialize, Deserialize)] +struct BenchTask; +impl TypedTask for BenchTask { + type Domain = BenchDomain; + const TASK_TYPE: &'static str = "test"; +} + struct NoopExecutor; -impl TaskExecutor for NoopExecutor { - async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for NoopExecutor { + async fn execute(&self, _payload: BenchTask, _ctx: &TaskContext) -> Result<(), TaskError> { Ok(()) } } @@ -30,7 +38,7 @@ impl TaskExecutor for NoopExecutor { async fn build_scheduler(max_concurrency: usize) -> Scheduler { Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("test", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .max_concurrency(max_concurrency) .poll_interval(Duration::from_millis(10)) .build() diff --git a/benches/groups.rs b/benches/groups.rs index fc9eaf1..1872157 100644 --- a/benches/groups.rs +++ b/benches/groups.rs @@ -5,9 +5,10 @@ use std::time::{Duration, Instant}; use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion}; +use serde::{Deserialize, Serialize}; use taskmill::{ - Domain, DomainKey, Scheduler, SchedulerEvent, TaskContext, TaskError, TaskExecutor, TaskStore, - TaskSubmission, + Domain, DomainKey, Scheduler, SchedulerEvent, TaskContext, TaskError, TaskStore, + TaskSubmission, TypedExecutor, TypedTask, }; use tokio::runtime::Runtime; use tokio_util::sync::CancellationToken; @@ -17,10 +18,17 @@ impl DomainKey for BenchDomain { const NAME: &'static str = "bench"; } +#[derive(Serialize, Deserialize)] +struct BenchTask; +impl TypedTask for BenchTask { + type Domain = BenchDomain; + const TASK_TYPE: &'static str = "test"; +} + struct NoopExecutor; -impl TaskExecutor for NoopExecutor { - async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for NoopExecutor { + async fn execute(&self, _payload: BenchTask, _ctx: &TaskContext) -> Result<(), TaskError> { Ok(()) } } @@ -56,7 +64,7 @@ fn bench_dispatch_no_groups(c: &mut Criterion) { for _ in 0..iters { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("test", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .max_concurrency(8) .poll_interval(Duration::from_millis(10)) .build() @@ -90,7 +98,7 @@ fn bench_dispatch_one_group(c: &mut Criterion) { for _ in 0..iters { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("test", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .max_concurrency(8) .group_concurrency("g0", 500) // high limit — no artificial throttling .poll_interval(Duration::from_millis(10)) @@ -138,7 +146,7 @@ fn bench_dispatch_group_scaling(c: &mut Criterion) { let mut builder = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("test", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .max_concurrency(8) .poll_interval(Duration::from_millis(10)); diff --git a/benches/history.rs b/benches/history.rs index 3500be0..dcd8917 100644 --- a/benches/history.rs +++ b/benches/history.rs @@ -5,9 +5,10 @@ use std::time::Duration; use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion}; +use serde::{Deserialize, Serialize}; use taskmill::{ - Domain, DomainKey, Scheduler, SchedulerEvent, TaskContext, TaskError, TaskExecutor, TaskStore, - TaskSubmission, + Domain, DomainKey, Scheduler, SchedulerEvent, TaskContext, TaskError, TaskStore, + TaskSubmission, TypedExecutor, TypedTask, }; use tokio::runtime::Runtime; use tokio_util::sync::CancellationToken; @@ -17,10 +18,17 @@ impl DomainKey for BenchDomain { const NAME: &'static str = "bench"; } +#[derive(Serialize, Deserialize)] +struct BenchTask; +impl TypedTask for BenchTask { + type Domain = BenchDomain; + const TASK_TYPE: &'static str = "test"; +} + struct NoopExecutor; -impl TaskExecutor for NoopExecutor { - async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for NoopExecutor { + async fn execute(&self, _payload: BenchTask, _ctx: &TaskContext) -> Result<(), TaskError> { Ok(()) } } @@ -29,7 +37,7 @@ impl TaskExecutor for NoopExecutor { async fn build_scheduler_with_history(n: usize) -> Scheduler { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("test", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .max_concurrency(32) .poll_interval(Duration::from_millis(10)) .build() diff --git a/benches/retry.rs b/benches/retry.rs index 240e521..d208069 100644 --- a/benches/retry.rs +++ b/benches/retry.rs @@ -5,9 +5,10 @@ use std::time::{Duration, Instant}; use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion}; +use serde::{Deserialize, Serialize}; use taskmill::{ BackoffStrategy, Domain, DomainKey, RetryPolicy, Scheduler, SchedulerEvent, TaskContext, - TaskError, TaskExecutor, TaskStore, TaskSubmission, + TaskError, TaskStore, TaskSubmission, TypedExecutor, TypedTask, }; use tokio::runtime::Runtime; use tokio_util::sync::CancellationToken; @@ -19,20 +20,29 @@ impl DomainKey for BenchDomain { const NAME: &'static str = "bench"; } +// ── Typed Tasks ──────────────────────────────────────────────────── + +#[derive(Serialize, Deserialize)] +struct FailTask; +impl TypedTask for FailTask { + type Domain = BenchDomain; + const TASK_TYPE: &'static str = "fail"; +} + // ── Executors ─────────────────────────────────────────────────────── struct FailPermanentExecutor; -impl TaskExecutor for FailPermanentExecutor { - async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for FailPermanentExecutor { + async fn execute(&self, _payload: FailTask, _ctx: &TaskContext) -> Result<(), TaskError> { Err(TaskError::permanent("bench: permanent failure")) } } struct FailRetryableExecutor; -impl TaskExecutor for FailRetryableExecutor { - async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for FailRetryableExecutor { + async fn execute(&self, _payload: FailTask, _ctx: &TaskContext) -> Result<(), TaskError> { Err(TaskError::retryable("bench: transient failure")) } } @@ -103,9 +113,7 @@ fn bench_dispatch_permanent_failure(c: &mut Criterion) { for _ in 0..iters { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain( - Domain::::new().raw_executor("fail", FailPermanentExecutor), - ) + .domain(Domain::::new().task::(FailPermanentExecutor)) .max_concurrency(8) .max_retries(0) .poll_interval(Duration::from_millis(10)) @@ -203,7 +211,7 @@ fn bench_dispatch_retryable_dead_letter(c: &mut Criterion) { .store(TaskStore::open_memory().await.unwrap()) .domain( Domain::::new() - .raw_executor("fail", FailRetryableExecutor) + .task::(FailRetryableExecutor) .default_retry(policy), ) .max_concurrency(8) diff --git a/benches/scheduler.rs b/benches/scheduler.rs index 658d6bd..8e6f3a4 100644 --- a/benches/scheduler.rs +++ b/benches/scheduler.rs @@ -5,9 +5,10 @@ use std::time::{Duration, Instant}; use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion}; +use serde::{Deserialize, Serialize}; use taskmill::{ - Domain, DomainKey, Priority, Scheduler, SchedulerEvent, TaskContext, TaskError, TaskExecutor, - TaskStore, TaskSubmission, + Domain, DomainKey, Priority, Scheduler, SchedulerEvent, TaskContext, TaskError, TaskStore, + TaskSubmission, TypedExecutor, TypedTask, }; use tokio::runtime::Runtime; use tokio_util::sync::CancellationToken; @@ -19,12 +20,28 @@ impl DomainKey for BenchDomain { const NAME: &'static str = "bench"; } +// ── Typed Tasks ──────────────────────────────────────────────────── + +#[derive(Serialize, Deserialize)] +struct BenchTask; +impl TypedTask for BenchTask { + type Domain = BenchDomain; + const TASK_TYPE: &'static str = "test"; +} + +#[derive(Serialize, Deserialize)] +struct ByteTestTask; +impl TypedTask for ByteTestTask { + type Domain = BenchDomain; + const TASK_TYPE: &'static str = "byte-test"; +} + // ── Test Executors ────────────────────────────────────────────────── struct NoopExecutor; -impl TaskExecutor for NoopExecutor { - async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for NoopExecutor { + async fn execute(&self, _payload: BenchTask, _ctx: &TaskContext) -> Result<(), TaskError> { Ok(()) } } @@ -35,8 +52,8 @@ struct ByteProgressExecutor { chunk_size: u64, } -impl TaskExecutor for ByteProgressExecutor { - async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for ByteProgressExecutor { + async fn execute(&self, _payload: ByteTestTask, ctx: &TaskContext) -> Result<(), TaskError> { ctx.set_bytes_total(self.total); let mut remaining = self.total; while remaining > 0 { @@ -53,7 +70,7 @@ impl TaskExecutor for ByteProgressExecutor { async fn build_scheduler(max_concurrency: usize) -> Scheduler { Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("test", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .max_concurrency(max_concurrency) .poll_interval(std::time::Duration::from_millis(10)) .build() @@ -360,8 +377,7 @@ fn bench_byte_progress_overhead(c: &mut Criterion) { for _ in 0..iters { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor( - "byte-test", + .domain(Domain::::new().task::( ByteProgressExecutor { total: 1_048_576, chunk_size: 1024, @@ -417,8 +433,7 @@ fn bench_byte_progress_snapshot(c: &mut Criterion) { for _ in 0..iters { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor( - "byte-test", + .domain(Domain::::new().task::( ByteProgressExecutor { total: 10_485_760, chunk_size: 65_536, diff --git a/benches/tags.rs b/benches/tags.rs index 6423e96..a98016a 100644 --- a/benches/tags.rs +++ b/benches/tags.rs @@ -5,8 +5,10 @@ use std::time::{Duration, Instant}; use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion}; +use serde::{Deserialize, Serialize}; use taskmill::{ - Domain, DomainKey, Scheduler, TaskContext, TaskError, TaskExecutor, TaskStore, TaskSubmission, + Domain, DomainKey, Scheduler, TaskContext, TaskError, TaskStore, TaskSubmission, TypedExecutor, + TypedTask, }; use tokio::runtime::Runtime; @@ -15,10 +17,17 @@ impl DomainKey for BenchDomain { const NAME: &'static str = "bench"; } +#[derive(Serialize, Deserialize)] +struct BenchTask; +impl TypedTask for BenchTask { + type Domain = BenchDomain; + const TASK_TYPE: &'static str = "test"; +} + struct NoopExecutor; -impl TaskExecutor for NoopExecutor { - async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for NoopExecutor { + async fn execute(&self, _payload: BenchTask, _ctx: &TaskContext) -> Result<(), TaskError> { Ok(()) } } @@ -57,7 +66,7 @@ fn bench_submit_with_tags(c: &mut Criterion) { for _ in 0..iters { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("test", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .max_concurrency(4) .poll_interval(Duration::from_millis(10)) .build() diff --git a/examples/profile_dep_chain.rs b/examples/profile_dep_chain.rs index 4558879..77ce17c 100644 --- a/examples/profile_dep_chain.rs +++ b/examples/profile_dep_chain.rs @@ -2,8 +2,10 @@ /// Run with: cargo run --release --example profile_dep_chain use std::time::{Duration, Instant}; +use serde::{Deserialize, Serialize}; use taskmill::{ - Domain, DomainKey, Scheduler, TaskContext, TaskError, TaskExecutor, TaskStore, TaskSubmission, + Domain, DomainKey, Scheduler, TaskContext, TaskError, TaskStore, TaskSubmission, TypedExecutor, + TypedTask, }; struct BenchDomain; @@ -11,9 +13,20 @@ impl DomainKey for BenchDomain { const NAME: &'static str = "bench"; } +#[derive(Serialize, Deserialize)] +struct BenchTask; +impl TypedTask for BenchTask { + type Domain = BenchDomain; + const TASK_TYPE: &'static str = "test"; +} + struct NoopExecutor; -impl TaskExecutor for NoopExecutor { - async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for NoopExecutor { + async fn execute<'a>( + &'a self, + _payload: BenchTask, + _ctx: &'a TaskContext, + ) -> Result<(), TaskError> { Ok(()) } } @@ -21,7 +34,7 @@ impl TaskExecutor for NoopExecutor { async fn build_scheduler() -> Scheduler { Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("test", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .max_concurrency(8) .poll_interval(Duration::from_millis(10)) .build() diff --git a/src/module.rs b/src/module.rs index f706019..30c0f20 100644 --- a/src/module.rs +++ b/src/module.rs @@ -891,11 +891,11 @@ impl ModuleHandle { #[cfg(test)] mod tests { - use std::sync::Arc; use std::time::Duration; + use crate::domain::{Domain, DomainKey, TypedExecutor}; use crate::priority::Priority; - use crate::registry::{TaskContext, TaskExecutor}; + use crate::registry::TaskContext; use crate::task::retry::{BackoffStrategy, RetryPolicy}; use crate::task::{TaskError, TypedTask}; @@ -903,8 +903,12 @@ mod tests { struct NoopExecutor; - impl TaskExecutor for NoopExecutor { - async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { + impl TypedExecutor for NoopExecutor { + async fn execute<'a>( + &'a self, + _payload: T, + _ctx: &'a TaskContext, + ) -> Result<(), TaskError> { Ok(()) } } @@ -914,19 +918,20 @@ mod tests { path: String, } - struct TestDomain; - impl crate::domain::DomainKey for TestDomain { - const NAME: &'static str = "test"; + struct MediaDomain; + impl DomainKey for MediaDomain { + const NAME: &'static str = "media"; } impl TypedTask for ThumbTask { - type Domain = TestDomain; + type Domain = MediaDomain; const TASK_TYPE: &'static str = "thumbnail"; } #[test] fn new_stores_name_and_typed_executor_reads_task_type() { - let module = Module::new("media").typed_executor::(Arc::new(NoopExecutor)); + let domain = Domain::::new().task::(NoopExecutor); + let module = domain.into_module(); assert_eq!(module.name(), "media"); assert_eq!(module.prefix(), "media::"); diff --git a/src/registry/context.rs b/src/registry/context.rs index c7951ca..94b8f6e 100644 --- a/src/registry/context.rs +++ b/src/registry/context.rs @@ -99,10 +99,14 @@ impl TaskContext { /// } /// ``` pub fn payload(&self) -> Result { - self.record - .deserialize_payload() - .map_err(TaskError::from)? - .ok_or_else(|| TaskError::new("missing payload")) + match self.record.deserialize_payload().map_err(TaskError::from)? { + Some(v) => Ok(v), + // No payload stored — try deserializing from JSON `null`. + // This succeeds for unit-struct typed tasks (e.g. `struct Noop;`) + // that are submitted via raw `TaskSubmission::new(...)`. + None => serde_json::from_value::(serde_json::Value::Null) + .map_err(|_| TaskError::new("missing payload")), + } } // ── Shared state ───────────────────────────────────────────────── diff --git a/src/registry/mod.rs b/src/registry/mod.rs index 5e3c1c2..2f94a5e 100644 --- a/src/registry/mod.rs +++ b/src/registry/mod.rs @@ -273,20 +273,52 @@ impl Default for TaskTypeRegistry { #[cfg(test)] mod tests { use super::*; + use crate::domain::{Domain, DomainKey, TypedExecutor}; use crate::task::retry::{BackoffStrategy, RetryPolicy}; + use crate::task::TypedTask; + + struct TestDomain; + impl DomainKey for TestDomain { + const NAME: &'static str = "test"; + } + + #[derive(serde::Serialize, serde::Deserialize)] + struct NoopTask; + impl TypedTask for NoopTask { + type Domain = TestDomain; + const TASK_TYPE: &'static str = "noop"; + } struct NoopExecutor; - impl TaskExecutor for NoopExecutor { - async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { + impl TypedExecutor for NoopExecutor { + async fn execute<'a>( + &'a self, + _payload: NoopTask, + _ctx: &'a TaskContext, + ) -> Result<(), TaskError> { Ok(()) } } + /// Helper: produce an erased executor from a `TypedExecutor` via + /// `Domain::task()`. This mirrors how the scheduler registers typed + /// executors internally. + fn erased_from_domain() -> Arc { + let domain = Domain::::new().task::(NoopExecutor); + domain + .into_module() + .executors + .into_iter() + .next() + .unwrap() + .executor + } + #[test] fn register_and_lookup() { let mut reg = TaskTypeRegistry::new(); - reg.register("test-type", Arc::new(NoopExecutor)); + reg.register_erased("test-type", erased_from_domain()); assert!(reg.get("test-type").is_some()); assert!(reg.get("unknown").is_none()); @@ -297,8 +329,8 @@ mod tests { #[should_panic(expected = "already registered")] fn duplicate_registration_panics() { let mut reg = TaskTypeRegistry::new(); - reg.register("dup", Arc::new(NoopExecutor)); - reg.register("dup", Arc::new(NoopExecutor)); + reg.register_erased("dup", erased_from_domain()); + reg.register_erased("dup", erased_from_domain()); } #[test] @@ -312,7 +344,7 @@ mod tests { }, max_retries: 5, }; - reg.register_with_retry_policy("api-call", Arc::new(NoopExecutor), policy); + reg.register_erased_with_retry_policy("api-call", erased_from_domain(), policy); assert!(reg.get("api-call").is_some()); let retrieved = reg.type_retry_policy("api-call").unwrap(); @@ -322,7 +354,7 @@ mod tests { #[test] fn type_retry_policy_returns_none_for_missing() { let mut reg = TaskTypeRegistry::new(); - reg.register("plain", Arc::new(NoopExecutor)); + reg.register_erased("plain", erased_from_domain()); assert!(reg.type_retry_policy("plain").is_none()); assert!(reg.type_retry_policy("nonexistent").is_none()); @@ -337,8 +369,7 @@ mod tests { }, max_retries: 7, }; - let executor: Arc = Arc::new(NoopExecutor); - reg.register_erased_with_retry_policy("erased-type", executor, policy); + reg.register_erased_with_retry_policy("erased-type", erased_from_domain(), policy); assert!(reg.get("erased-type").is_some()); let retrieved = reg.type_retry_policy("erased-type").unwrap(); diff --git a/src/scheduler/tests.rs b/src/scheduler/tests.rs index b1bfa0f..84f945b 100644 --- a/src/scheduler/tests.rs +++ b/src/scheduler/tests.rs @@ -1,20 +1,102 @@ use std::sync::Arc; +use serde::{Deserialize, Serialize}; use tokio::time::Duration; use tokio_util::sync::CancellationToken; -use crate::backpressure::{CompositePressure, ThrottlePolicy}; +use crate::domain::{Domain, DomainKey, TypedExecutor}; use crate::priority::Priority; -use crate::registry::{TaskContext, TaskExecutor, TaskTypeRegistry}; +use crate::registry::TaskContext; use crate::store::TaskStore; -use crate::task::{DuplicateStrategy, HistoryStatus, SubmitOutcome, TaskError, TaskSubmission}; +use crate::task::{ + DuplicateStrategy, HistoryStatus, SubmitOutcome, TaskError, TaskSubmission, TypedTask, +}; -use super::{Scheduler, SchedulerConfig, SchedulerEvent, TaskProgress}; +use super::{Scheduler, SchedulerEvent, TaskProgress}; + +// ── Domain & task type definitions ────────────────────────────────── + +struct TestDomain; +impl DomainKey for TestDomain { + const NAME: &'static str = "test"; +} + +struct ParentDomain; +impl DomainKey for ParentDomain { + const NAME: &'static str = "parent"; +} + +// ChildTask lives in ParentDomain so spawn_child (which prefixes with +// the owning module) produces "parent::child" — matching the registration. + +struct ByteTestDomain; +impl DomainKey for ByteTestDomain { + const NAME: &'static str = "byte-test"; +} + +struct AlphaDomain; +impl DomainKey for AlphaDomain { + const NAME: &'static str = "alpha"; +} + +struct BetaDomain; +impl DomainKey for BetaDomain { + const NAME: &'static str = "beta"; +} + +struct UnknownDomain; +impl DomainKey for UnknownDomain { + const NAME: &'static str = "unknown"; +} + +#[derive(Serialize, Deserialize)] +struct TestTask; +impl TypedTask for TestTask { + type Domain = TestDomain; + const TASK_TYPE: &'static str = "test"; +} + +#[derive(Serialize, Deserialize)] +struct ParentTask; +impl TypedTask for ParentTask { + type Domain = ParentDomain; + const TASK_TYPE: &'static str = "parent"; +} + +#[derive(Serialize, Deserialize)] +struct ChildTask; +impl TypedTask for ChildTask { + type Domain = ParentDomain; + const TASK_TYPE: &'static str = "child"; +} + +#[derive(Serialize, Deserialize)] +struct ByteTestTask; +impl TypedTask for ByteTestTask { + type Domain = ByteTestDomain; + const TASK_TYPE: &'static str = "byte-test"; +} + +#[derive(Serialize, Deserialize)] +struct AlphaTask; +impl TypedTask for AlphaTask { + type Domain = AlphaDomain; + const TASK_TYPE: &'static str = "alpha"; +} + +#[derive(Serialize, Deserialize)] +struct BetaTask; +impl TypedTask for BetaTask { + type Domain = BetaDomain; + const TASK_TYPE: &'static str = "beta"; +} + +// ── Executors ─────────────────────────────────────────────────────── struct InstantExecutor; -impl TaskExecutor for InstantExecutor { - async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for InstantExecutor { + async fn execute<'a>(&'a self, _payload: T, ctx: &'a TaskContext) -> Result<(), TaskError> { ctx.record_read_bytes(100); ctx.record_write_bytes(50); Ok(()) @@ -23,8 +105,8 @@ impl TaskExecutor for InstantExecutor { struct SlowExecutor; -impl TaskExecutor for SlowExecutor { - async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for SlowExecutor { + async fn execute<'a>(&'a self, _payload: T, ctx: &'a TaskContext) -> Result<(), TaskError> { tokio::select! { _ = ctx.token().cancelled() => { Err(TaskError::new("cancelled")) @@ -43,8 +125,12 @@ struct CancelHookExecutor { cancel_called: Arc, } -impl TaskExecutor for CancelHookExecutor { - async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for CancelHookExecutor { + async fn execute<'a>( + &'a self, + _payload: TestTask, + ctx: &'a TaskContext, + ) -> Result<(), TaskError> { tokio::select! { _ = ctx.token().cancelled() => { Err(TaskError::new("cancelled")) @@ -55,7 +141,11 @@ impl TaskExecutor for CancelHookExecutor { } } - async fn on_cancel<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { + async fn on_cancel<'a>( + &'a self, + _payload: TestTask, + _ctx: &'a TaskContext, + ) -> Result<(), TaskError> { self.cancel_called .store(true, std::sync::atomic::Ordering::SeqCst); Ok(()) @@ -65,38 +155,37 @@ impl TaskExecutor for CancelHookExecutor { #[allow(dead_code)] struct FailingExecutor; -impl TaskExecutor for FailingExecutor { - async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for FailingExecutor { + async fn execute<'a>( + &'a self, + _payload: TestTask, + _ctx: &'a TaskContext, + ) -> Result<(), TaskError> { Err(TaskError::retryable("boom")) } } -async fn setup(executor: Arc) -> Scheduler { - let store = TaskStore::open_memory().await.unwrap(); - let mut registry = TaskTypeRegistry::new(); - registry.register_erased("test", executor); - - Scheduler::new( - store, - SchedulerConfig::default(), - Arc::new(registry), - CompositePressure::new(), - ThrottlePolicy::default_three_tier(), - ) +async fn setup(executor: impl TypedExecutor) -> Scheduler { + Scheduler::builder() + .store(TaskStore::open_memory().await.unwrap()) + .domain(Domain::::new().task::(executor)) + .build() + .await + .unwrap() } -fn arc_erased(e: E) -> Arc { - Arc::new(e) as Arc +/// Helper: create a test-domain submission with a serialized `TestTask` payload. +fn test_sub(key: &str) -> TaskSubmission { + TaskSubmission::new("test::test") + .key(key) + .payload_json(&TestTask) } #[tokio::test] async fn dispatch_executes_task() { - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; - sched - .submit(&TaskSubmission::new("test").key("k1")) - .await - .unwrap(); + sched.submit(&test_sub("k1")).await.unwrap(); let dispatched = sched.try_dispatch().await.unwrap(); assert!(dispatched); @@ -105,7 +194,7 @@ async fn dispatch_executes_task() { tokio::time::sleep(Duration::from_millis(50)).await; // Task should be completed and in history. - let k1 = crate::task::generate_dedup_key("test", Some(b"k1")); + let k1 = crate::task::generate_dedup_key("test::test", Some(b"k1")); assert!(sched.store().task_by_key(&k1).await.unwrap().is_none()); let hist = sched.store().history_by_key(&k1).await.unwrap(); assert_eq!(hist.len(), 1); @@ -113,7 +202,7 @@ async fn dispatch_executes_task() { #[tokio::test] async fn dispatch_returns_false_when_empty() { - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; let dispatched = sched.try_dispatch().await.unwrap(); assert!(!dispatched); } @@ -121,15 +210,13 @@ async fn dispatch_returns_false_when_empty() { #[tokio::test] async fn unregistered_type_fails_task() { let store = TaskStore::open_memory().await.unwrap(); - let registry = TaskTypeRegistry::new(); // empty — no executors - - let sched = Scheduler::new( - store, - SchedulerConfig::default(), - Arc::new(registry), - CompositePressure::new(), - ThrottlePolicy::default_three_tier(), - ); + // Empty domain — no executors registered for "unknown::unknown". + let sched = Scheduler::builder() + .store(store) + .domain(Domain::::new()) + .build() + .await + .unwrap(); sched .submit(&TaskSubmission::new("unknown").key("k")) @@ -145,9 +232,9 @@ async fn unregistered_type_fails_task() { #[tokio::test] async fn dedup_via_scheduler() { - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; - let sub = TaskSubmission::new("test").key("dup"); + let sub = test_sub("dup"); let first = sched.submit(&sub).await.unwrap(); let second = sched.submit(&sub).await.unwrap(); @@ -157,7 +244,7 @@ async fn dedup_via_scheduler() { #[tokio::test] async fn set_max_concurrency_works() { - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; assert_eq!(sched.max_concurrency(), 4); sched.set_max_concurrency(8); assert_eq!(sched.max_concurrency(), 8); @@ -165,10 +252,10 @@ async fn set_max_concurrency_works() { #[tokio::test] async fn cancel_pending_task() { - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; let id = sched - .submit(&TaskSubmission::new("test").key("cancel-me")) + .submit(&test_sub("cancel-me")) .await .unwrap() .id() @@ -178,7 +265,7 @@ async fn cancel_pending_task() { assert!(cancelled); // Task should be gone. - let cancel_key = crate::task::generate_dedup_key("test", Some(b"cancel-me")); + let cancel_key = crate::task::generate_dedup_key("test::test", Some(b"cancel-me")); assert!(sched .store() .task_by_key(&cancel_key) @@ -189,10 +276,10 @@ async fn cancel_pending_task() { #[tokio::test] async fn cancel_running_task() { - let sched = setup(arc_erased(SlowExecutor)).await; + let sched = setup(SlowExecutor).await; let id = sched - .submit(&TaskSubmission::new("test").key("cancel-running")) + .submit(&test_sub("cancel-running")) .await .unwrap() .id() @@ -208,13 +295,10 @@ async fn cancel_running_task() { #[tokio::test] async fn event_emitted_on_complete() { - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; let mut rx = sched.subscribe(); - sched - .submit(&TaskSubmission::new("test").key("evt")) - .await - .unwrap(); + sched.submit(&test_sub("evt")).await.unwrap(); sched.try_dispatch().await.unwrap(); @@ -231,17 +315,14 @@ async fn event_emitted_on_complete() { #[tokio::test] async fn scheduler_is_clone() { - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; let sched2 = sched.clone(); // Both should share the same store. - sched - .submit(&TaskSubmission::new("test").key("shared")) - .await - .unwrap(); + sched.submit(&test_sub("shared")).await.unwrap(); // The clone can see the task. - let shared_key = crate::task::generate_dedup_key("test", Some(b"shared")); + let shared_key = crate::task::generate_dedup_key("test::test", Some(b"shared")); let task = sched2.store().task_by_key(&shared_key).await.unwrap(); assert!(task.is_some()); } @@ -255,13 +336,13 @@ async fn submit_typed_enqueues_task() { path: String, } - struct TestDomain; - impl crate::domain::DomainKey for TestDomain { + struct TestDomain2; + impl crate::domain::DomainKey for TestDomain2 { const NAME: &'static str = "test"; } impl crate::task::TypedTask for Thumb { - type Domain = TestDomain; + type Domain = TestDomain2; const TASK_TYPE: &'static str = "test"; fn config() -> crate::domain::TaskTypeConfig { @@ -269,7 +350,7 @@ async fn submit_typed_enqueues_task() { } } - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; let task = Thumb { path: "/a.jpg".into(), @@ -295,14 +376,11 @@ async fn submit_typed_enqueues_task() { #[tokio::test] async fn snapshot_returns_dashboard_state() { - let sched = setup(arc_erased(SlowExecutor)).await; + let sched = setup(SlowExecutor).await; // Submit two tasks. for key in &["snap-a", "snap-b"] { - sched - .submit(&TaskSubmission::new("test").key(*key)) - .await - .unwrap(); + sched.submit(&test_sub(key)).await.unwrap(); } // Dispatch one so it becomes running. @@ -322,14 +400,11 @@ async fn snapshot_returns_dashboard_state() { #[tokio::test] async fn pause_all_stops_dispatching() { - let sched = setup(arc_erased(SlowExecutor)).await; + let sched = setup(SlowExecutor).await; // Submit two tasks. for key in &["pa-1", "pa-2"] { - sched - .submit(&TaskSubmission::new("test").key(*key)) - .await - .unwrap(); + sched.submit(&test_sub(key)).await.unwrap(); } // Dispatch one so it's running. @@ -357,7 +432,7 @@ async fn pause_all_stops_dispatching() { #[tokio::test] async fn pause_resume_events_emitted() { - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; let mut rx = sched.subscribe(); sched.pause_all().await; @@ -379,8 +454,12 @@ async fn app_state_accessible_from_executor() { struct StateCheckExecutor; - impl TaskExecutor for StateCheckExecutor { - async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> { + impl TypedExecutor for StateCheckExecutor { + async fn execute<'a>( + &'a self, + _payload: TestTask, + ctx: &'a TaskContext, + ) -> Result<(), TaskError> { let state = ctx.state::().expect("state should be set"); state.flag.store(true, Ordering::SeqCst); Ok(()) @@ -391,16 +470,13 @@ async fn app_state_accessible_from_executor() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .module(crate::module::Module::new("test").executor("test", Arc::new(StateCheckExecutor))) + .domain(Domain::::new().task::(StateCheckExecutor)) .app_state(MyState { flag: flag.clone() }) .build() .await .unwrap(); - sched - .submit(&TaskSubmission::new("test::test").key("state-test")) - .await - .unwrap(); + sched.submit(&test_sub("state-test")).await.unwrap(); sched.try_dispatch().await.unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; @@ -410,14 +486,14 @@ async fn app_state_accessible_from_executor() { #[tokio::test] async fn task_lookup_pending() { - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; - sched - .submit(&TaskSubmission::new("test").key("lookup-1")) + sched.submit(&test_sub("lookup-1")).await.unwrap(); + + let result = sched + .task_lookup("test::test", Some(b"lookup-1")) .await .unwrap(); - - let result = sched.task_lookup("test", Some(b"lookup-1")).await.unwrap(); assert!(matches!( result, crate::task::TaskLookup::Active(ref r) if r.status == crate::task::TaskStatus::Pending @@ -426,18 +502,15 @@ async fn task_lookup_pending() { #[tokio::test] async fn task_lookup_completed() { - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; - sched - .submit(&TaskSubmission::new("test").key("lookup-done")) - .await - .unwrap(); + sched.submit(&test_sub("lookup-done")).await.unwrap(); sched.try_dispatch().await.unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; let result = sched - .task_lookup("test", Some(b"lookup-done")) + .task_lookup("test::test", Some(b"lookup-done")) .await .unwrap(); assert!(matches!(result, crate::task::TaskLookup::History(_))); @@ -445,9 +518,9 @@ async fn task_lookup_completed() { #[tokio::test] async fn task_lookup_not_found() { - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; let result = sched - .task_lookup("test", Some(b"does-not-exist")) + .task_lookup("test::test", Some(b"does-not-exist")) .await .unwrap(); assert!(matches!(result, crate::task::TaskLookup::NotFound)); @@ -472,7 +545,7 @@ async fn lookup_typed_works() { const TASK_TYPE: &'static str = "test"; } - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; let task = Thumb { path: "/a.jpg".into(), @@ -490,12 +563,17 @@ struct SpawningExecutor { num_children: usize, } -impl TaskExecutor for SpawningExecutor { - async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for SpawningExecutor { + async fn execute<'a>( + &'a self, + _payload: ParentTask, + ctx: &'a TaskContext, + ) -> Result<(), TaskError> { for i in 0..self.num_children { let sub = TaskSubmission::new("child") .key(format!("child-{i}")) - .priority(ctx.record().priority); + .priority(ctx.record().priority) + .payload_json(&ChildTask); ctx.spawn_child(sub).await?; } Ok(()) @@ -508,18 +586,27 @@ struct FinalizeTrackingExecutor { finalized: Arc, } -impl TaskExecutor for FinalizeTrackingExecutor { - async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for FinalizeTrackingExecutor { + async fn execute<'a>( + &'a self, + _payload: ParentTask, + ctx: &'a TaskContext, + ) -> Result<(), TaskError> { for i in 0..self.children { let sub = TaskSubmission::new("child") .key(format!("ft-child-{i}")) - .priority(ctx.record().priority); + .priority(ctx.record().priority) + .payload_json(&ChildTask); ctx.spawn_child(sub).await?; } Ok(()) } - async fn finalize<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { + async fn finalize<'a>( + &'a self, + _payload: ParentTask, + _ctx: &'a TaskContext, + ) -> Result<(), TaskError> { self.finalized .store(true, std::sync::atomic::Ordering::SeqCst); Ok(()) @@ -528,23 +615,26 @@ impl TaskExecutor for FinalizeTrackingExecutor { #[tokio::test] async fn parent_enters_waiting_when_children_spawned() { - let store = TaskStore::open_memory().await.unwrap(); - let mut registry = TaskTypeRegistry::new(); - registry.register_erased("parent", arc_erased(SpawningExecutor { num_children: 2 })); - registry.register_erased("child", arc_erased(InstantExecutor)); - - let sched = Scheduler::new( - store, - SchedulerConfig::default(), - Arc::new(registry), - CompositePressure::new(), - ThrottlePolicy::default_three_tier(), - ); + let sched = Scheduler::builder() + .store(TaskStore::open_memory().await.unwrap()) + .domain( + Domain::::new() + .task::(SpawningExecutor { num_children: 2 }) + .task::(InstantExecutor), + ) + .build() + .await + .unwrap(); + let mut rx = sched.subscribe(); // Submit parent task. sched - .submit(&TaskSubmission::new("parent").key("p1")) + .submit( + &TaskSubmission::new("parent::parent") + .key("p1") + .payload_json(&ParentTask), + ) .await .unwrap(); @@ -565,7 +655,7 @@ async fn parent_enters_waiting_when_children_spawned() { assert!(saw_waiting, "expected Waiting event for parent"); // Parent should be in waiting status in the store. - let parent_key = crate::task::generate_dedup_key("parent", Some(b"p1")); + let parent_key = crate::task::generate_dedup_key("parent::parent", Some(b"p1")); let parent = sched .store() .task_by_key(&parent_key) @@ -580,22 +670,25 @@ async fn parent_enters_waiting_when_children_spawned() { #[tokio::test] async fn parent_auto_completes_after_children_finish() { - let store = TaskStore::open_memory().await.unwrap(); - let mut registry = TaskTypeRegistry::new(); - registry.register_erased("parent", arc_erased(SpawningExecutor { num_children: 2 })); - registry.register_erased("child", arc_erased(InstantExecutor)); - - let sched = Scheduler::new( - store, - SchedulerConfig::default(), - Arc::new(registry), - CompositePressure::new(), - ThrottlePolicy::default_three_tier(), - ); + let sched = Scheduler::builder() + .store(TaskStore::open_memory().await.unwrap()) + .domain( + Domain::::new() + .task::(SpawningExecutor { num_children: 2 }) + .task::(InstantExecutor), + ) + .build() + .await + .unwrap(); + let mut rx = sched.subscribe(); sched - .submit(&TaskSubmission::new("parent").key("p-complete")) + .submit( + &TaskSubmission::new("parent::parent") + .key("p-complete") + .payload_json(&ParentTask), + ) .await .unwrap(); @@ -608,12 +701,12 @@ async fn parent_auto_completes_after_children_finish() { }); // Wait for parent Completed event. - let parent_key = crate::task::generate_dedup_key("parent", Some(b"p-complete")); + let parent_key = crate::task::generate_dedup_key("parent::parent", Some(b"p-complete")); let deadline = tokio::time::Instant::now() + Duration::from_secs(5); let mut parent_completed = false; while tokio::time::Instant::now() < deadline { match tokio::time::timeout(Duration::from_millis(200), rx.recv()).await { - Ok(Ok(SchedulerEvent::Completed(ref h))) if h.task_type == "parent" => { + Ok(Ok(SchedulerEvent::Completed(ref h))) if h.task_type == "parent::parent" => { parent_completed = true; break; } @@ -638,28 +731,28 @@ async fn parent_auto_completes_after_children_finish() { async fn finalize_called_after_children_complete() { let finalized = Arc::new(std::sync::atomic::AtomicBool::new(false)); - let store = TaskStore::open_memory().await.unwrap(); - let mut registry = TaskTypeRegistry::new(); - registry.register_erased( - "parent", - arc_erased(FinalizeTrackingExecutor { - children: 1, - finalized: finalized.clone(), - }), - ); - registry.register_erased("child", arc_erased(InstantExecutor)); - - let sched = Scheduler::new( - store, - SchedulerConfig::default(), - Arc::new(registry), - CompositePressure::new(), - ThrottlePolicy::default_three_tier(), - ); + let sched = Scheduler::builder() + .store(TaskStore::open_memory().await.unwrap()) + .domain( + Domain::::new() + .task::(FinalizeTrackingExecutor { + children: 1, + finalized: finalized.clone(), + }) + .task::(InstantExecutor), + ) + .build() + .await + .unwrap(); + let mut rx = sched.subscribe(); sched - .submit(&TaskSubmission::new("parent").key("p-finalize")) + .submit( + &TaskSubmission::new("parent::parent") + .key("p-finalize") + .payload_json(&ParentTask), + ) .await .unwrap(); @@ -674,7 +767,7 @@ async fn finalize_called_after_children_complete() { let deadline = tokio::time::Instant::now() + Duration::from_secs(5); while tokio::time::Instant::now() < deadline { match tokio::time::timeout(Duration::from_millis(100), rx.recv()).await { - Ok(Ok(SchedulerEvent::Completed(ref h))) if h.task_type == "parent" => { + Ok(Ok(SchedulerEvent::Completed(ref h))) if h.task_type == "parent::parent" => { break; } _ => {} @@ -692,21 +785,23 @@ async fn finalize_called_after_children_complete() { #[tokio::test] async fn cancel_parent_cascades_to_children() { - let store = TaskStore::open_memory().await.unwrap(); - let mut registry = TaskTypeRegistry::new(); - registry.register_erased("parent", arc_erased(SpawningExecutor { num_children: 3 })); - registry.register_erased("child", arc_erased(SlowExecutor)); - - let sched = Scheduler::new( - store, - SchedulerConfig::default(), - Arc::new(registry), - CompositePressure::new(), - ThrottlePolicy::default_three_tier(), - ); + let sched = Scheduler::builder() + .store(TaskStore::open_memory().await.unwrap()) + .domain( + Domain::::new() + .task::(SpawningExecutor { num_children: 3 }) + .task::(SlowExecutor), + ) + .build() + .await + .unwrap(); let parent_id = sched - .submit(&TaskSubmission::new("parent").key("p-cancel")) + .submit( + &TaskSubmission::new("parent::parent") + .key("p-cancel") + .payload_json(&ParentTask), + ) .await .unwrap() .id() @@ -728,17 +823,14 @@ async fn cancel_parent_cascades_to_children() { #[tokio::test] async fn no_children_completes_normally() { // Task without children should complete as before (backward compat). - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; - sched - .submit(&TaskSubmission::new("test").key("no-kids")) - .await - .unwrap(); + sched.submit(&test_sub("no-kids")).await.unwrap(); sched.try_dispatch().await.unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; - let key = crate::task::generate_dedup_key("test", Some(b"no-kids")); + let key = crate::task::generate_dedup_key("test::test", Some(b"no-kids")); let lookup = sched.store().task_lookup(&key).await.unwrap(); assert!(matches!(lookup, crate::task::TaskLookup::History(_))); } @@ -748,8 +840,12 @@ async fn no_children_completes_normally() { /// An executor that reports byte-level progress incrementally. struct ByteProgressExecutor; -impl TaskExecutor for ByteProgressExecutor { - async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for ByteProgressExecutor { + async fn execute<'a>( + &'a self, + _payload: ByteTestTask, + ctx: &'a TaskContext, + ) -> Result<(), TaskError> { ctx.set_bytes_total(1_048_576); for _ in 0..1024 { ctx.add_bytes(1024); @@ -759,31 +855,26 @@ impl TaskExecutor for ByteProgressExecutor { } } +/// Helper: create a byte-test submission with a serialized payload. +fn byte_test_sub(key: &str) -> TaskSubmission { + TaskSubmission::new("byte-test::byte-test") + .key(key) + .payload_json(&ByteTestTask) +} + #[tokio::test] async fn byte_progress_events_received() { - let store = TaskStore::open_memory().await.unwrap(); - let mut registry = TaskTypeRegistry::new(); - registry.register_erased("byte-test", arc_erased(ByteProgressExecutor)); - - let config = SchedulerConfig { - progress_interval: Some(Duration::from_millis(50)), - ..Default::default() - }; - - let sched = Scheduler::new( - store, - config, - Arc::new(registry), - CompositePressure::new(), - ThrottlePolicy::default_three_tier(), - ); + let sched = Scheduler::builder() + .store(TaskStore::open_memory().await.unwrap()) + .domain(Domain::::new().task::(ByteProgressExecutor)) + .progress_interval(Duration::from_millis(50)) + .build() + .await + .unwrap(); let mut progress_rx = sched.subscribe_progress(); - sched - .submit(&TaskSubmission::new("byte-test").key("bp1")) - .await - .unwrap(); + sched.submit(&byte_test_sub("bp1")).await.unwrap(); // Run the scheduler. let token = CancellationToken::new(); @@ -832,29 +923,17 @@ async fn byte_progress_events_received() { #[tokio::test] async fn lifecycle_events_not_polluted_by_byte_progress() { - let store = TaskStore::open_memory().await.unwrap(); - let mut registry = TaskTypeRegistry::new(); - registry.register_erased("byte-test", arc_erased(ByteProgressExecutor)); - - let config = SchedulerConfig { - progress_interval: Some(Duration::from_millis(50)), - ..Default::default() - }; - - let sched = Scheduler::new( - store, - config, - Arc::new(registry), - CompositePressure::new(), - ThrottlePolicy::default_three_tier(), - ); + let sched = Scheduler::builder() + .store(TaskStore::open_memory().await.unwrap()) + .domain(Domain::::new().task::(ByteProgressExecutor)) + .progress_interval(Duration::from_millis(50)) + .build() + .await + .unwrap(); let mut lifecycle_rx = sched.subscribe(); - sched - .submit(&TaskSubmission::new("byte-test").key("bp-lifecycle")) - .await - .unwrap(); + sched.submit(&byte_test_sub("bp-lifecycle")).await.unwrap(); let token = CancellationToken::new(); let sched_clone = sched.clone(); @@ -899,23 +978,15 @@ async fn lifecycle_events_not_polluted_by_byte_progress() { #[tokio::test] async fn byte_progress_in_snapshot() { - let store = TaskStore::open_memory().await.unwrap(); - let mut registry = TaskTypeRegistry::new(); - registry.register_erased("byte-test", arc_erased(ByteProgressExecutor)); - - let sched = Scheduler::new( - store, - SchedulerConfig::default(), - Arc::new(registry), - CompositePressure::new(), - ThrottlePolicy::default_three_tier(), - ); - - sched - .submit(&TaskSubmission::new("byte-test").key("bp-snap")) + let sched = Scheduler::builder() + .store(TaskStore::open_memory().await.unwrap()) + .domain(Domain::::new().task::(ByteProgressExecutor)) + .build() .await .unwrap(); + sched.submit(&byte_test_sub("bp-snap")).await.unwrap(); + sched.try_dispatch().await.unwrap(); // Let the executor run a bit so bytes are reported. tokio::time::sleep(Duration::from_millis(100)).await; @@ -931,11 +1002,15 @@ async fn byte_progress_in_snapshot() { #[tokio::test] async fn batch_submitted_event() { - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; let mut rx = sched.subscribe(); let subs: Vec<_> = (0..3) - .map(|i| TaskSubmission::new("test").key(format!("ev-{i}"))) + .map(|i| { + TaskSubmission::new("test::test") + .key(format!("ev-{i}")) + .payload_json(&TestTask) + }) .collect(); let outcome = sched.submit_batch(&subs).await.unwrap(); @@ -959,19 +1034,12 @@ async fn batch_submitted_event() { #[tokio::test] async fn batch_outcome_convenience_methods() { - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; // Submit one task first so re-submitting it produces a Duplicate. - sched - .submit(&TaskSubmission::new("test").key("existing")) - .await - .unwrap(); + sched.submit(&test_sub("existing")).await.unwrap(); - let subs = vec![ - TaskSubmission::new("test").key("new-1"), - TaskSubmission::new("test").key("existing"), - TaskSubmission::new("test").key("new-2"), - ]; + let subs = vec![test_sub("new-1"), test_sub("existing"), test_sub("new-2")]; let outcome = sched.submit_batch(&subs).await.unwrap(); assert_eq!(outcome.len(), 3); @@ -985,13 +1053,13 @@ async fn batch_outcome_convenience_methods() { async fn submit_built_applies_defaults() { use crate::task::BatchSubmission; - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; let batch = BatchSubmission::new() .default_group("g1") .default_priority(Priority::HIGH) - .task(TaskSubmission::new("test").key("built-1")) - .task(TaskSubmission::new("test").key("built-2")); + .task(test_sub("built-1")) + .task(test_sub("built-2")); let outcome = sched.submit_built(batch).await.unwrap(); assert_eq!(outcome.inserted().len(), 2); @@ -1001,10 +1069,10 @@ async fn submit_built_applies_defaults() { #[tokio::test] async fn cancel_pending_records_history() { - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; let id = sched - .submit(&TaskSubmission::new("test").key("cancel-hist")) + .submit(&test_sub("cancel-hist")) .await .unwrap() .id() @@ -1014,7 +1082,7 @@ async fn cancel_pending_records_history() { assert!(cancelled); // Task should be gone from active queue. - let key = crate::task::generate_dedup_key("test", Some(b"cancel-hist")); + let key = crate::task::generate_dedup_key("test::test", Some(b"cancel-hist")); assert!(sched.store().task_by_key(&key).await.unwrap().is_none()); // History should have a cancelled entry. @@ -1030,20 +1098,15 @@ async fn cancel_running_records_history_and_fires_hook() { cancel_called: cancel_called.clone(), }; - let store = TaskStore::open_memory().await.unwrap(); - let mut registry = TaskTypeRegistry::new(); - registry.register_erased("test", arc_erased(executor)); - - let sched = Scheduler::new( - store, - SchedulerConfig::default(), - Arc::new(registry), - CompositePressure::new(), - ThrottlePolicy::default_three_tier(), - ); + let sched = Scheduler::builder() + .store(TaskStore::open_memory().await.unwrap()) + .domain(Domain::::new().task::(executor)) + .build() + .await + .unwrap(); let id = sched - .submit(&TaskSubmission::new("test").key("cancel-running-hist")) + .submit(&test_sub("cancel-running-hist")) .await .unwrap() .id() @@ -1065,7 +1128,7 @@ async fn cancel_running_records_history_and_fires_hook() { ); // History should have a cancelled entry. - let key = crate::task::generate_dedup_key("test", Some(b"cancel-running-hist")); + let key = crate::task::generate_dedup_key("test::test", Some(b"cancel-running-hist")); let hist = sched.store().history_by_key(&key).await.unwrap(); assert_eq!(hist.len(), 1); assert_eq!(hist[0].status, crate::task::HistoryStatus::Cancelled); @@ -1073,21 +1136,23 @@ async fn cancel_running_records_history_and_fires_hook() { #[tokio::test] async fn cancel_parent_cascade_records_history() { - let store = TaskStore::open_memory().await.unwrap(); - let mut registry = TaskTypeRegistry::new(); - registry.register_erased("parent", arc_erased(SpawningExecutor { num_children: 2 })); - registry.register_erased("child", arc_erased(SlowExecutor)); - - let sched = Scheduler::new( - store, - SchedulerConfig::default(), - Arc::new(registry), - CompositePressure::new(), - ThrottlePolicy::default_three_tier(), - ); + let sched = Scheduler::builder() + .store(TaskStore::open_memory().await.unwrap()) + .domain( + Domain::::new() + .task::(SpawningExecutor { num_children: 2 }) + .task::(SlowExecutor), + ) + .build() + .await + .unwrap(); let parent_id = sched - .submit(&TaskSubmission::new("parent").key("p-cancel-hist")) + .submit( + &TaskSubmission::new("parent::parent") + .key("p-cancel-hist") + .payload_json(&ParentTask), + ) .await .unwrap() .id() @@ -1128,19 +1193,19 @@ async fn check_cancelled_returns_error() { #[tokio::test] async fn cancel_group_cancels_matching_tasks() { - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; // Submit tasks in different groups. sched - .submit(&TaskSubmission::new("test").key("g-a1").group("group-a")) + .submit(&test_sub("g-a1").group("group-a")) .await .unwrap(); sched - .submit(&TaskSubmission::new("test").key("g-a2").group("group-a")) + .submit(&test_sub("g-a2").group("group-a")) .await .unwrap(); sched - .submit(&TaskSubmission::new("test").key("g-b1").group("group-b")) + .submit(&test_sub("g-b1").group("group-b")) .await .unwrap(); @@ -1153,33 +1218,40 @@ async fn cancel_group_cancels_matching_tasks() { #[tokio::test] async fn cancel_type_cancels_matching_tasks() { - let store = TaskStore::open_memory().await.unwrap(); - let mut registry = TaskTypeRegistry::new(); - registry.register_erased("alpha", arc_erased(InstantExecutor)); - registry.register_erased("beta", arc_erased(InstantExecutor)); - - let sched = Scheduler::new( - store, - SchedulerConfig::default(), - Arc::new(registry), - CompositePressure::new(), - ThrottlePolicy::default_three_tier(), - ); + let sched = Scheduler::builder() + .store(TaskStore::open_memory().await.unwrap()) + .domain(Domain::::new().task::(InstantExecutor)) + .domain(Domain::::new().task::(InstantExecutor)) + .build() + .await + .unwrap(); sched - .submit(&TaskSubmission::new("alpha").key("a1")) + .submit( + &TaskSubmission::new("alpha::alpha") + .key("a1") + .payload_json(&AlphaTask), + ) .await .unwrap(); sched - .submit(&TaskSubmission::new("alpha").key("a2")) + .submit( + &TaskSubmission::new("alpha::alpha") + .key("a2") + .payload_json(&AlphaTask), + ) .await .unwrap(); sched - .submit(&TaskSubmission::new("beta").key("b1")) + .submit( + &TaskSubmission::new("beta::beta") + .key("b1") + .payload_json(&BetaTask), + ) .await .unwrap(); - let cancelled = sched.cancel_type("alpha").await.unwrap(); + let cancelled = sched.cancel_type("alpha::alpha").await.unwrap(); assert_eq!(cancelled.len(), 2); // beta task should still exist. @@ -1188,13 +1260,10 @@ async fn cancel_type_cancels_matching_tasks() { #[tokio::test] async fn cancel_where_filters_correctly() { - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; for i in 0..5 { - sched - .submit(&TaskSubmission::new("test").key(format!("cw-{i}"))) - .await - .unwrap(); + sched.submit(&test_sub(&format!("cw-{i}"))).await.unwrap(); } // Cancel only tasks whose key contains "cw-3" or "cw-4". @@ -1210,40 +1279,39 @@ async fn cancel_where_filters_correctly() { async fn on_cancel_hook_timeout_does_not_block() { struct SlowCancelExecutor; - impl TaskExecutor for SlowCancelExecutor { - async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> { + impl TypedExecutor for SlowCancelExecutor { + async fn execute<'a>( + &'a self, + _payload: TestTask, + ctx: &'a TaskContext, + ) -> Result<(), TaskError> { tokio::select! { _ = ctx.token().cancelled() => Err(TaskError::new("cancelled")), _ = tokio::time::sleep(Duration::from_secs(60)) => Ok(()), } } - async fn on_cancel<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { + async fn on_cancel<'a>( + &'a self, + _payload: TestTask, + _ctx: &'a TaskContext, + ) -> Result<(), TaskError> { // Simulate a very slow cancel hook. tokio::time::sleep(Duration::from_secs(60)).await; Ok(()) } } - let store = TaskStore::open_memory().await.unwrap(); - let mut registry = TaskTypeRegistry::new(); - registry.register_erased("test", arc_erased(SlowCancelExecutor)); - - let config = SchedulerConfig { - cancel_hook_timeout: Duration::from_millis(50), - ..Default::default() - }; - - let sched = Scheduler::new( - store, - config, - Arc::new(registry), - CompositePressure::new(), - ThrottlePolicy::default_three_tier(), - ); + let sched = Scheduler::builder() + .store(TaskStore::open_memory().await.unwrap()) + .domain(Domain::::new().task::(SlowCancelExecutor)) + .cancel_hook_timeout(Duration::from_millis(50)) + .build() + .await + .unwrap(); let id = sched - .submit(&TaskSubmission::new("test").key("timeout-hook")) + .submit(&test_sub("timeout-hook")) .await .unwrap() .id() @@ -1271,11 +1339,9 @@ async fn on_cancel_hook_timeout_does_not_block() { #[tokio::test] async fn reject_returns_rejected() { - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; - let sub = TaskSubmission::new("test") - .key("dup") - .on_duplicate(DuplicateStrategy::Reject); + let sub = test_sub("dup").on_duplicate(DuplicateStrategy::Reject); let first = sched.submit(&sub).await.unwrap(); assert!(first.is_inserted()); @@ -1285,10 +1351,10 @@ async fn reject_returns_rejected() { #[tokio::test] async fn supersede_pending_replaces_in_place() { - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; // Submit initial task. - let sub1 = TaskSubmission::new("test") + let sub1 = TaskSubmission::new("test::test") .key("replace-me") .priority(Priority::NORMAL) .payload_raw(b"old".to_vec()); @@ -1296,7 +1362,7 @@ async fn supersede_pending_replaces_in_place() { let first_id = first.id().unwrap(); // Supersede with new payload and higher priority. - let sub2 = TaskSubmission::new("test") + let sub2 = TaskSubmission::new("test::test") .key("replace-me") .priority(Priority::HIGH) .payload_raw(b"new".to_vec()) @@ -1326,10 +1392,10 @@ async fn supersede_running_cancels_and_inserts_new() { let executor = CancelHookExecutor { cancel_called: cancel_called.clone(), }; - let sched = setup(arc_erased(executor)).await; + let sched = setup(executor).await; // Submit and dispatch (now running). - let sub1 = TaskSubmission::new("test").key("running-sup"); + let sub1 = test_sub("running-sup"); sched.submit(&sub1).await.unwrap(); sched.try_dispatch().await.unwrap(); @@ -1337,7 +1403,7 @@ async fn supersede_running_cancels_and_inserts_new() { tokio::time::sleep(Duration::from_millis(20)).await; // Supersede the running task. - let sub2 = TaskSubmission::new("test") + let sub2 = TaskSubmission::new("test::test") .key("running-sup") .payload_raw(b"replacement".to_vec()) .on_duplicate(DuplicateStrategy::Supersede); @@ -1369,15 +1435,13 @@ async fn supersede_running_cancels_and_inserts_new() { #[tokio::test] async fn supersede_emits_event() { - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; let mut rx = sched.subscribe(); - let sub1 = TaskSubmission::new("test").key("evt"); + let sub1 = test_sub("evt"); sched.submit(&sub1).await.unwrap(); - let sub2 = TaskSubmission::new("test") - .key("evt") - .on_duplicate(DuplicateStrategy::Supersede); + let sub2 = test_sub("evt").on_duplicate(DuplicateStrategy::Supersede); sched.submit(&sub2).await.unwrap(); // Drain events and look for Superseded. @@ -1392,14 +1456,14 @@ async fn supersede_emits_event() { #[tokio::test] async fn supersede_in_batch() { - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; // Pre-submit a task. - let sub1 = TaskSubmission::new("test").key("batch-sup"); + let sub1 = test_sub("batch-sup"); sched.submit(&sub1).await.unwrap(); // Batch supersede it. - let sub2 = TaskSubmission::new("test") + let sub2 = TaskSubmission::new("test::test") .key("batch-sup") .payload_raw(b"batch-new".to_vec()) .on_duplicate(DuplicateStrategy::Supersede); @@ -1418,17 +1482,17 @@ async fn supersede_in_batch() { #[tokio::test] async fn chain_of_supersedes() { - let sched = setup(arc_erased(InstantExecutor)).await; + let sched = setup(InstantExecutor).await; // A supersedes nothing (fresh insert). - let sub_a = TaskSubmission::new("test") + let sub_a = TaskSubmission::new("test::test") .key("chain") .payload_raw(b"A".to_vec()); let out_a = sched.submit(&sub_a).await.unwrap(); assert!(matches!(out_a, SubmitOutcome::Inserted(_))); // B supersedes A. - let sub_b = TaskSubmission::new("test") + let sub_b = TaskSubmission::new("test::test") .key("chain") .payload_raw(b"B".to_vec()) .on_duplicate(DuplicateStrategy::Supersede); @@ -1436,7 +1500,7 @@ async fn chain_of_supersedes() { assert!(matches!(out_b, SubmitOutcome::Superseded { .. })); // C supersedes B. - let sub_c = TaskSubmission::new("test") + let sub_c = TaskSubmission::new("test::test") .key("chain") .payload_raw(b"C".to_vec()) .on_duplicate(DuplicateStrategy::Supersede); @@ -1461,33 +1525,18 @@ async fn chain_of_supersedes() { #[tokio::test] async fn retry_dead_letter_resubmits_with_reset_retry_count() { // Use max_retries=0 so a retryable failure immediately dead-letters. - let store = TaskStore::open_memory().await.unwrap(); - let mut registry = TaskTypeRegistry::new(); - registry.register_erased("test", arc_erased(FailingExecutor)); - let config = SchedulerConfig { - max_retries: 0, - ..SchedulerConfig::default() - }; - - let sched = Scheduler::new( - store, - config, - Arc::new(registry), - CompositePressure::new(), - ThrottlePolicy::default_three_tier(), - ); + let sched = Scheduler::builder() + .store(TaskStore::open_memory().await.unwrap()) + .domain(Domain::::new().task::(FailingExecutor)) + .max_retries(0) + .build() + .await + .unwrap(); let mut rx = sched.subscribe(); // Submit and dispatch — will fail with retryable error and dead-letter. - sched - .submit( - &TaskSubmission::new("test") - .key("dl-retry") - .payload_raw(b"payload".to_vec()), - ) - .await - .unwrap(); + sched.submit(&test_sub("dl-retry")).await.unwrap(); sched.try_dispatch().await.unwrap(); tokio::time::sleep(Duration::from_millis(100)).await; @@ -1519,8 +1568,7 @@ async fn retry_dead_letter_resubmits_with_reset_retry_count() { assert!(dl_after.is_empty()); // New task should be in the active queue with retry_count=0. - let key = crate::task::generate_dedup_key("test", Some(b"dl-retry")); + let key = crate::task::generate_dedup_key("test::test", Some(b"dl-retry")); let task = sched.store().task_by_key(&key).await.unwrap().unwrap(); assert_eq!(task.retry_count, 0); - assert_eq!(task.payload.as_deref(), Some(b"payload".as_slice())); } diff --git a/tests/integration/common.rs b/tests/integration/common.rs index e5fb7fc..d3cdec6 100644 --- a/tests/integration/common.rs +++ b/tests/integration/common.rs @@ -6,8 +6,10 @@ use std::sync::atomic::{AtomicBool, AtomicI32, AtomicUsize, Ordering}; use std::sync::Arc; use std::time::Duration; +use serde::{Deserialize, Serialize}; use taskmill::{ - DomainKey, PressureSource, SchedulerEvent, TaskContext, TaskError, TaskExecutor, TaskSubmission, + DomainKey, PressureSource, SchedulerEvent, TaskContext, TaskError, TaskSubmission, + TypedExecutor, TypedTask, }; // ── Domain Keys ──────────────────────────────────────────────────── @@ -57,13 +59,72 @@ impl DomainKey for GammaDomain { const NAME: &'static str = "gamma"; } +// ── Helper macro for defining no-payload TypedTask types ─────────── + +/// Define a zero-payload [`TypedTask`] type in one line. +/// +/// ```ignore +/// define_task!(TestTask, TestDomain, "test"); +/// ``` +macro_rules! define_task { + ($name:ident, $domain:ty, $task_type:expr) => { + #[derive(Debug, Clone, Serialize, Deserialize)] + pub struct $name; + impl TypedTask for $name { + type Domain = $domain; + const TASK_TYPE: &'static str = $task_type; + } + }; +} + +pub(crate) use define_task; + +// ── Shared TypedTask types ───────────────────────────────────────── + +// TestDomain tasks +define_task!(TestTask, TestDomain, "test"); +define_task!(SlowTask, TestDomain, "slow"); +define_task!(FastTask, TestDomain, "fast"); +define_task!(ParentTask, TestDomain, "parent"); +define_task!(ChildTask, TestDomain, "child"); +define_task!(WorkerTask, TestDomain, "worker"); +define_task!(SpawnerTask, TestDomain, "spawner"); +define_task!(ProbeTask, TestDomain, "probe"); +define_task!(StepTask, TestDomain, "step"); + +// MediaDomain tasks +define_task!(ThumbTask, MediaDomain, "thumb"); +define_task!(TranscodeTask, MediaDomain, "transcode"); +define_task!(MediaWorkTask, MediaDomain, "work"); +define_task!(MediaLeaderTask, MediaDomain, "leader"); +define_task!(MediaFollowerTask, MediaDomain, "follower"); +define_task!(MediaParentTask, MediaDomain, "parent"); +define_task!(MediaLayeredTask, MediaDomain, "layered"); + +// SyncDomain tasks +define_task!(PushTask, SyncDomain, "push"); +define_task!(SyncWorkTask, SyncDomain, "work"); + +// AnalyticsDomain tasks +define_task!(AnalyticsWorkTask, AnalyticsDomain, "work"); + +// Multi-domain "work" tasks +define_task!(AlphaWorkTask, AlphaDomain, "work"); +define_task!(BetaWorkTask, BetaDomain, "work"); +define_task!(GammaWorkTask, GammaDomain, "work"); + +// DomainA / DomainB tasks +define_task!(TriggerTask, DomainA, "trigger"); +define_task!(DomainATask, DomainA, "task"); +define_task!(DomainBTask, DomainB, "task"); + // ── Test Executors ────────────────────────────────────────────────── /// Completes immediately with no side effects. pub struct NoopExecutor; -impl TaskExecutor for NoopExecutor { - async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for NoopExecutor { + async fn execute<'a>(&'a self, _payload: T, _ctx: &'a TaskContext) -> Result<(), TaskError> { Ok(()) } } @@ -71,8 +132,8 @@ impl TaskExecutor for NoopExecutor { /// Sleeps for a configurable duration, respecting cancellation. pub struct DelayExecutor(pub Duration); -impl TaskExecutor for DelayExecutor { - async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for DelayExecutor { + async fn execute<'a>(&'a self, _payload: T, ctx: &'a TaskContext) -> Result<(), TaskError> { tokio::select! { _ = ctx.token().cancelled() => Err(TaskError::new("cancelled")), _ = tokio::time::sleep(self.0) => Ok(()), @@ -85,8 +146,8 @@ pub struct CountingExecutor { pub count: Arc, } -impl TaskExecutor for CountingExecutor { - async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for CountingExecutor { + async fn execute<'a>(&'a self, _payload: T, _ctx: &'a TaskContext) -> Result<(), TaskError> { self.count.fetch_add(1, Ordering::SeqCst); Ok(()) } @@ -98,8 +159,8 @@ pub struct FailNTimesExecutor { pub max_failures: i32, } -impl TaskExecutor for FailNTimesExecutor { - async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for FailNTimesExecutor { + async fn execute<'a>(&'a self, _payload: T, _ctx: &'a TaskContext) -> Result<(), TaskError> { let count = self.failures.fetch_add(1, Ordering::SeqCst); if count < self.max_failures { Err(TaskError::retryable("transient failure")) @@ -115,8 +176,8 @@ pub struct IoReportingExecutor { pub write: i64, } -impl TaskExecutor for IoReportingExecutor { - async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for IoReportingExecutor { + async fn execute<'a>(&'a self, _payload: T, ctx: &'a TaskContext) -> Result<(), TaskError> { ctx.record_read_bytes(self.read); ctx.record_write_bytes(self.write); Ok(()) @@ -130,8 +191,8 @@ pub struct ConcurrencyTrackingExecutor { pub delay: Duration, } -impl TaskExecutor for ConcurrencyTrackingExecutor { - async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for ConcurrencyTrackingExecutor { + async fn execute<'a>(&'a self, _payload: T, ctx: &'a TaskContext) -> Result<(), TaskError> { let prev = self.current.fetch_add(1, Ordering::SeqCst); self.max_seen.fetch_max(prev + 1, Ordering::SeqCst); tokio::select! { @@ -150,8 +211,8 @@ pub struct ChildSpawnerExecutor { pub fail_fast: bool, } -impl TaskExecutor for ChildSpawnerExecutor { - async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for ChildSpawnerExecutor { + async fn execute<'a>(&'a self, _payload: T, ctx: &'a TaskContext) -> Result<(), TaskError> { for i in 0..self.count { let sub = TaskSubmission::new(self.child_type) .key(format!("child-{i}")) @@ -169,8 +230,8 @@ pub struct FinalizeTracker { pub finalized: Arc, } -impl TaskExecutor for FinalizeTracker { - async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for FinalizeTracker { + async fn execute<'a>(&'a self, _payload: T, ctx: &'a TaskContext) -> Result<(), TaskError> { for i in 0..self.child_count { let sub = TaskSubmission::new("child") .key(format!("ft-child-{i}")) @@ -180,7 +241,7 @@ impl TaskExecutor for FinalizeTracker { Ok(()) } - async fn finalize<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { + async fn finalize<'a>(&'a self, _payload: T, _ctx: &'a TaskContext) -> Result<(), TaskError> { self.finalized.store(true, Ordering::SeqCst); Ok(()) } @@ -189,8 +250,8 @@ impl TaskExecutor for FinalizeTracker { /// Fails unconditionally with a non-retryable error. pub struct AlwaysFailExecutor; -impl TaskExecutor for AlwaysFailExecutor { - async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for AlwaysFailExecutor { + async fn execute<'a>(&'a self, _payload: T, _ctx: &'a TaskContext) -> Result<(), TaskError> { Err(TaskError::new("permanent failure")) } } diff --git a/tests/integration/cross_module.rs b/tests/integration/cross_module.rs index 850c265..ffc015d 100644 --- a/tests/integration/cross_module.rs +++ b/tests/integration/cross_module.rs @@ -5,14 +5,19 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::time::Duration; +use serde::{Deserialize, Serialize}; use taskmill::{ - Domain, DomainHandle, DomainKey, Scheduler, SchedulerEvent, TaskContext, TaskError, - TaskExecutor, TaskStore, TaskSubmission, + Domain, DomainHandle, DomainKey, Scheduler, SchedulerEvent, TaskContext, TaskError, TaskStore, + TaskSubmission, TypedExecutor, TypedTask, }; use tokio_util::sync::CancellationToken; use super::common::*; +// ── Extra task types not defined in common.rs ─────────────────────────── +define_task!(MediaThumbnailTask, MediaDomain, "thumbnail"); +define_task!(MediaChildTask, MediaDomain, "child"); + // ── Step 8: TaskContext domain access ───────────────────────────────────── /// Executor in module A that submits a task to module B via `ctx.domain::()`. @@ -20,8 +25,12 @@ struct CrossDomainSubmitter { submitted: Arc, } -impl TaskExecutor for CrossDomainSubmitter { - async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for CrossDomainSubmitter { + async fn execute<'a>( + &'a self, + _payload: TriggerTask, + ctx: &'a TaskContext, + ) -> Result<(), TaskError> { let b: DomainHandle = ctx.domain::(); b.submit_raw(TaskSubmission::new("task").key("cross-module-child")) .await @@ -40,16 +49,19 @@ async fn ctx_domain_submits_to_other_module_with_prefix_and_defaults() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor( - "trigger", - CrossDomainSubmitter { + .domain( + Domain::::new().task::(CrossDomainSubmitter { submitted: submitted_clone, - }, - )) - .domain(Domain::::new().raw_executor("task", { + }), + ) + .domain(Domain::::new().task::({ struct B(Arc); - impl TaskExecutor for B { - async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { + impl TypedExecutor for B { + async fn execute<'a>( + &'a self, + _payload: DomainBTask, + _ctx: &'a TaskContext, + ) -> Result<(), TaskError> { self.0.store(true, Ordering::SeqCst); Ok(()) } @@ -92,8 +104,12 @@ struct SameDomainSubmitter { submitted: Arc, } -impl TaskExecutor for SameDomainSubmitter { - async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for SameDomainSubmitter { + async fn execute<'a>( + &'a self, + _payload: MediaLeaderTask, + ctx: &'a TaskContext, + ) -> Result<(), TaskError> { let media: DomainHandle = ctx.domain::(); media .submit_raw(TaskSubmission::new("follower").key("same-module-follower")) @@ -115,17 +131,15 @@ async fn ctx_domain_self_submit_applies_owning_module_defaults() { .store(TaskStore::open_memory().await.unwrap()) .domain( Domain::::new() - .raw_executor( - "leader", - SameDomainSubmitter { - submitted: submitted_clone, - }, - ) - .raw_executor("follower", { + .task::(SameDomainSubmitter { + submitted: submitted_clone, + }) + .task::({ struct Follower(Arc); - impl TaskExecutor for Follower { + impl TypedExecutor for Follower { async fn execute<'a>( &'a self, + _payload: MediaFollowerTask, _ctx: &'a TaskContext, ) -> Result<(), TaskError> { self.0.store(true, Ordering::SeqCst); @@ -177,8 +191,12 @@ async fn ctx_try_domain_returns_none_for_unknown_domain() { } struct TryDomainExecutor(Arc>>); - impl TaskExecutor for TryDomainExecutor { - async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> { + impl TypedExecutor for TryDomainExecutor { + async fn execute<'a>( + &'a self, + _payload: ProbeTask, + ctx: &'a TaskContext, + ) -> Result<(), TaskError> { let found = ctx.try_domain::().is_some(); *self.0.lock().unwrap() = Some(found); Ok(()) @@ -187,7 +205,7 @@ async fn ctx_try_domain_returns_none_for_unknown_domain() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("probe", TryDomainExecutor(result_clone))) + .domain(Domain::::new().task::(TryDomainExecutor(result_clone))) .max_concurrency(2) .poll_interval(Duration::from_millis(20)) .build() @@ -223,8 +241,12 @@ async fn spawn_child_routes_through_owning_module() { let child_ran_clone = child_ran.clone(); struct SpawnChildExecutor; - impl TaskExecutor for SpawnChildExecutor { - async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> { + impl TypedExecutor for SpawnChildExecutor { + async fn execute<'a>( + &'a self, + _payload: SpawnerTask, + ctx: &'a TaskContext, + ) -> Result<(), TaskError> { ctx.spawn_child(TaskSubmission::new("worker").key("spawned-child")) .await?; Ok(()) @@ -232,8 +254,12 @@ async fn spawn_child_routes_through_owning_module() { } struct WorkerExecutor(Arc); - impl TaskExecutor for WorkerExecutor { - async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { + impl TypedExecutor for WorkerExecutor { + async fn execute<'a>( + &'a self, + _payload: WorkerTask, + _ctx: &'a TaskContext, + ) -> Result<(), TaskError> { self.0.store(true, Ordering::SeqCst); Ok(()) } @@ -243,8 +269,8 @@ async fn spawn_child_routes_through_owning_module() { .store(TaskStore::open_memory().await.unwrap()) .domain( Domain::::new() - .raw_executor("spawner", SpawnChildExecutor) - .raw_executor("worker", WorkerExecutor(child_ran_clone)), + .task::(SpawnChildExecutor) + .task::(WorkerExecutor(child_ran_clone)), ) .max_concurrency(4) .poll_interval(Duration::from_millis(20)) @@ -280,8 +306,12 @@ struct CrossDomainParentExec { child_submitted: Arc, } -impl TaskExecutor for CrossDomainParentExec { - async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for CrossDomainParentExec { + async fn execute<'a>( + &'a self, + _payload: MediaParentTask, + ctx: &'a TaskContext, + ) -> Result<(), TaskError> { let analytics: DomainHandle = ctx.domain::(); analytics .submit_raw(TaskSubmission::new("work").key("cross-child")) @@ -304,16 +334,19 @@ async fn cross_domain_parent_child_lifecycle() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor( - "parent", - CrossDomainParentExec { + .domain( + Domain::::new().task::(CrossDomainParentExec { child_submitted: child_submitted_clone, - }, - )) - .domain(Domain::::new().raw_executor("work", { + }), + ) + .domain(Domain::::new().task::({ struct AnalyticsExec(Arc); - impl TaskExecutor for AnalyticsExec { - async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { + impl TypedExecutor for AnalyticsExec { + async fn execute<'a>( + &'a self, + _payload: AnalyticsWorkTask, + _ctx: &'a TaskContext, + ) -> Result<(), TaskError> { self.0.store(true, Ordering::SeqCst); Ok(()) } @@ -371,13 +404,12 @@ async fn cross_domain_parent_child_lifecycle() { async fn cross_domain_failure_cascade() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor( - "parent", - CrossDomainParentExec { + .domain( + Domain::::new().task::(CrossDomainParentExec { child_submitted: Arc::new(AtomicBool::new(false)), - }, - )) - .domain(Domain::::new().raw_executor("work", AlwaysFailExecutor)) + }), + ) + .domain(Domain::::new().task::(AlwaysFailExecutor)) .max_concurrency(4) .max_retries(0) .poll_interval(Duration::from_millis(20)) @@ -427,9 +459,9 @@ async fn cross_domain_failure_cascade() { async fn scheduler_domains_returns_registered_modules() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("work", NoopExecutor)) - .domain(Domain::::new().raw_executor("work", NoopExecutor)) - .domain(Domain::::new().raw_executor("work", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .max_concurrency(4) .build() .await @@ -447,8 +479,29 @@ async fn scheduler_active_tasks_returns_tasks_from_all_modules() { let barrier_clone = barrier.clone(); struct BarrierExecutor(Arc); - impl TaskExecutor for BarrierExecutor { - async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> { + impl TypedExecutor for BarrierExecutor { + async fn execute<'a>( + &'a self, + _payload: AlphaWorkTask, + ctx: &'a TaskContext, + ) -> Result<(), TaskError> { + self.0.wait().await; + tokio::select! { + _ = ctx.token().cancelled() => {}, + _ = tokio::time::sleep(Duration::from_secs(5)) => {}, + } + Ok(()) + } + } + + // Need a separate type for BetaWorkTask since BarrierExecutor already impls for AlphaWorkTask + struct BarrierExecutorBeta(Arc); + impl TypedExecutor for BarrierExecutorBeta { + async fn execute<'a>( + &'a self, + _payload: BetaWorkTask, + ctx: &'a TaskContext, + ) -> Result<(), TaskError> { self.0.wait().await; tokio::select! { _ = ctx.token().cancelled() => {}, @@ -460,8 +513,12 @@ async fn scheduler_active_tasks_returns_tasks_from_all_modules() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("work", BarrierExecutor(barrier.clone()))) - .domain(Domain::::new().raw_executor("work", BarrierExecutor(barrier_clone))) + .domain( + Domain::::new().task::(BarrierExecutor(barrier.clone())), + ) + .domain( + Domain::::new().task::(BarrierExecutorBeta(barrier_clone)), + ) .max_concurrency(4) .poll_interval(Duration::from_millis(10)) .build() @@ -509,8 +566,8 @@ async fn scheduler_active_tasks_returns_tasks_from_all_modules() { async fn cross_domain_cancel_by_tag_via_domain_handles() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("work", NoopExecutor)) - .domain(Domain::::new().raw_executor("work", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .max_concurrency(8) .build() .await @@ -595,8 +652,8 @@ async fn parent_method_inherits_ttl_and_tags() { .store(TaskStore::open_memory().await.unwrap()) .domain( Domain::::new() - .raw_executor("parent", NoopExecutor) - .raw_executor("child", NoopExecutor), + .task::(NoopExecutor) + .task::(NoopExecutor), ) .max_concurrency(2) .build() @@ -671,7 +728,7 @@ async fn parent_method_inherits_ttl_and_tags() { async fn event_header_module_field_populated_from_task_type_prefix() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("thumbnail", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .max_concurrency(4) .build() .await @@ -719,8 +776,8 @@ async fn event_header_module_field_populated_from_task_type_prefix() { async fn module_receiver_events_match_module_field() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("thumbnail", NoopExecutor)) - .domain(Domain::::new().raw_executor("push", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .max_concurrency(8) .build() .await diff --git a/tests/integration/dependencies.rs b/tests/integration/dependencies.rs index 6dfb4bc..01c863a 100644 --- a/tests/integration/dependencies.rs +++ b/tests/integration/dependencies.rs @@ -4,16 +4,11 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use std::time::Duration; -use taskmill::{Domain, DomainKey, Scheduler, TaskStore, TaskSubmission}; +use taskmill::{Domain, Scheduler, TaskStore, TaskSubmission}; use tokio_util::sync::CancellationToken; use super::common::*; -struct TestDomain; -impl DomainKey for TestDomain { - const NAME: &'static str = "test"; -} - // ═══════════════════════════════════════════════════════════════════ // M. Task Dependencies // ═══════════════════════════════════════════════════════════════════ @@ -434,8 +429,7 @@ async fn dep_blocked_count_in_snapshot() { let sched = Scheduler::builder() .store(store) .domain( - Domain::::new() - .raw_executor("test", DelayExecutor(Duration::from_secs(60))), + Domain::::new().task::(DelayExecutor(Duration::from_secs(60))), ) .build() .await @@ -473,12 +467,11 @@ async fn dep_full_chain_with_scheduler() { let sched = Scheduler::builder() .store(store) - .domain(Domain::::new().raw_executor( - "step", - CountingExecutor { + .domain( + Domain::::new().task::(CountingExecutor { count: counter.clone(), - }, - )) + }), + ) .build() .await .unwrap(); diff --git a/tests/integration/module_features.rs b/tests/integration/module_features.rs index e0f1392..d607718 100644 --- a/tests/integration/module_features.rs +++ b/tests/integration/module_features.rs @@ -6,35 +6,13 @@ use std::sync::Arc; use std::time::Duration; use taskmill::{ - Domain, DomainKey, Priority, Scheduler, SchedulerEvent, TaskContext, TaskError, TaskExecutor, - TaskStore, TaskSubmission, TaskTypeConfig, + Domain, Priority, Scheduler, SchedulerEvent, TaskContext, TaskError, TaskStore, TaskSubmission, + TaskTypeConfig, TypedExecutor, TypedTask, }; use tokio_util::sync::CancellationToken; use super::common::*; -// ── Domain key structs ────────────────────────────────────────────── - -struct MediaDomain; -impl DomainKey for MediaDomain { - const NAME: &'static str = "media"; -} - -struct SyncDomain; -impl DomainKey for SyncDomain { - const NAME: &'static str = "sync"; -} - -struct ADomain; -impl DomainKey for ADomain { - const NAME: &'static str = "a"; -} - -struct BDomain; -impl DomainKey for BDomain { - const NAME: &'static str = "b"; -} - // ═══════════════════════════════════════════════════════════════════ // P. Default Layering (Step 5) // ═══════════════════════════════════════════════════════════════════ @@ -70,7 +48,7 @@ async fn submit_typed_five_layer_precedence_chain() { .default_ttl(std::time::Duration::from_secs(14400)) // layer 5 (not reached) .domain( Domain::::new() - .raw_executor("layered", NoopExecutor) + .task::(NoopExecutor) .default_priority(Priority::BACKGROUND) // layer 3: overrides TypedTask HIGH .default_group("module-group") // layer 3: overrides typed-group .default_ttl(std::time::Duration::from_secs(10800)) // layer 3: 3 h @@ -141,14 +119,11 @@ async fn module_cap_limits_concurrency_to_2() { .poll_interval(Duration::from_millis(20)) .domain( Domain::::new() - .raw_executor( - "work", - ConcurrencyTrackingExecutor { - current: current.clone(), - max_seen: max_seen.clone(), - delay: Duration::from_millis(100), - }, - ) + .task::(ConcurrencyTrackingExecutor { + current: current.clone(), + max_seen: max_seen.clone(), + delay: Duration::from_millis(100), + }) .max_concurrency(2), ) .build() @@ -204,14 +179,11 @@ async fn module_cap_and_group_cap_are_independent() { .group_concurrency("gpu", 2) // group cap = 2 .domain( Domain::::new() - .raw_executor( - "work", - ConcurrencyTrackingExecutor { - current: current.clone(), - max_seen: max_seen.clone(), - delay: Duration::from_millis(100), - }, - ) + .task::(ConcurrencyTrackingExecutor { + current: current.clone(), + max_seen: max_seen.clone(), + delay: Duration::from_millis(100), + }) .max_concurrency(4), // module cap = 4 ) .build() @@ -270,14 +242,11 @@ async fn ungrouped_task_respects_module_cap() { .poll_interval(Duration::from_millis(20)) .domain( Domain::::new() - .raw_executor( - "work", - ConcurrencyTrackingExecutor { - current: current.clone(), - max_seen: max_seen.clone(), - delay: Duration::from_millis(100), - }, - ) + .task::(ConcurrencyTrackingExecutor { + current: current.clone(), + max_seen: max_seen.clone(), + delay: Duration::from_millis(100), + }) .max_concurrency(3), ) .build() @@ -332,26 +301,20 @@ async fn global_cap_is_hard_ceiling_over_module_caps() { .poll_interval(Duration::from_millis(20)) .domain( Domain::::new() - .raw_executor( - "work", - ConcurrencyTrackingExecutor { - current: total_current.clone(), - max_seen: total_max.clone(), - delay: Duration::from_millis(100), - }, - ) + .task::(ConcurrencyTrackingExecutor { + current: total_current.clone(), + max_seen: total_max.clone(), + delay: Duration::from_millis(100), + }) .max_concurrency(3), ) .domain( Domain::::new() - .raw_executor( - "work", - ConcurrencyTrackingExecutor { - current: total_current.clone(), - max_seen: total_max.clone(), - delay: Duration::from_millis(100), - }, - ) + .task::(ConcurrencyTrackingExecutor { + current: total_current.clone(), + max_seen: total_max.clone(), + delay: Duration::from_millis(100), + }) .max_concurrency(3), ) .build() @@ -409,14 +372,11 @@ async fn set_max_concurrency_changes_dispatch_behavior() { .poll_interval(Duration::from_millis(20)) .domain( Domain::::new() - .raw_executor( - "work", - ConcurrencyTrackingExecutor { - current: current.clone(), - max_seen: max_seen.clone(), - delay: Duration::from_millis(100), - }, - ) + .task::(ConcurrencyTrackingExecutor { + current: current.clone(), + max_seen: max_seen.clone(), + delay: Duration::from_millis(100), + }) .max_concurrency(4), // initial cap — will be narrowed at runtime ) .build() @@ -482,8 +442,8 @@ async fn module_state_is_scoped_to_module() { saw_a: Arc, no_b: Arc, } - impl TaskExecutor for CheckerExec { - async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> { + impl TypedExecutor for CheckerExec { + async fn execute<'a>(&'a self, _payload: T, ctx: &'a TaskContext) -> Result<(), TaskError> { self.saw_a .store(ctx.state::().is_some(), Ordering::SeqCst); if ctx.state::().is_some() { @@ -497,19 +457,16 @@ async fn module_state_is_scoped_to_module() { .store(TaskStore::open_memory().await.unwrap()) .poll_interval(Duration::from_millis(20)) .domain( - Domain::::new() - .raw_executor( - "task", - CheckerExec { - saw_a: Arc::clone(&saw_a), - no_b: Arc::clone(&no_b), - }, - ) + Domain::::new() + .task::(CheckerExec { + saw_a: Arc::clone(&saw_a), + no_b: Arc::clone(&no_b), + }) .state(ConfigA("a-config".into())), ) .domain( - Domain::::new() - .raw_executor("task", NoopExecutor) + Domain::::new() + .task::(NoopExecutor) .state(ConfigB("b-config".into())), ) .build() @@ -517,7 +474,7 @@ async fn module_state_is_scoped_to_module() { .unwrap(); sched - .domain::() + .domain::() .submit_raw(TaskSubmission::new("task").key("t1")) .await .unwrap(); @@ -560,8 +517,8 @@ async fn global_state_accessible_from_all_modules() { let b_saw = Arc::new(AtomicBool::new(false)); struct GlobalChecker(Arc); - impl TaskExecutor for GlobalChecker { - async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> { + impl TypedExecutor for GlobalChecker { + async fn execute<'a>(&'a self, _payload: T, ctx: &'a TaskContext) -> Result<(), TaskError> { self.0 .store(ctx.state::().is_some(), Ordering::SeqCst); Ok(()) @@ -572,19 +529,19 @@ async fn global_state_accessible_from_all_modules() { .store(TaskStore::open_memory().await.unwrap()) .poll_interval(Duration::from_millis(20)) .app_state(SharedConfig("global".into())) - .domain(Domain::::new().raw_executor("task", GlobalChecker(Arc::clone(&a_saw)))) - .domain(Domain::::new().raw_executor("task", GlobalChecker(Arc::clone(&b_saw)))) + .domain(Domain::::new().task::(GlobalChecker(Arc::clone(&a_saw)))) + .domain(Domain::::new().task::(GlobalChecker(Arc::clone(&b_saw)))) .build() .await .unwrap(); sched - .domain::() + .domain::() .submit_raw(TaskSubmission::new("task").key("ta")) .await .unwrap(); sched - .domain::() + .domain::() .submit_raw(TaskSubmission::new("task").key("tb")) .await .unwrap(); @@ -625,8 +582,8 @@ async fn module_state_shadows_global_state() { let b_value = Arc::new(std::sync::Mutex::new(String::new())); struct ValueCapture(Arc>); - impl TaskExecutor for ValueCapture { - async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> { + impl TypedExecutor for ValueCapture { + async fn execute<'a>(&'a self, _payload: T, ctx: &'a TaskContext) -> Result<(), TaskError> { if let Some(cfg) = ctx.state::() { *self.0.lock().unwrap() = cfg.0.clone(); } @@ -639,22 +596,22 @@ async fn module_state_shadows_global_state() { .poll_interval(Duration::from_millis(20)) .app_state(Config("global".into())) .domain( - Domain::::new() - .raw_executor("task", ValueCapture(Arc::clone(&a_value))) + Domain::::new() + .task::(ValueCapture(Arc::clone(&a_value))) .state(Config("module-a".into())), ) - .domain(Domain::::new().raw_executor("task", ValueCapture(Arc::clone(&b_value)))) + .domain(Domain::::new().task::(ValueCapture(Arc::clone(&b_value)))) .build() .await .unwrap(); sched - .domain::() + .domain::() .submit_raw(TaskSubmission::new("task").key("ta")) .await .unwrap(); sched - .domain::() + .domain::() .submit_raw(TaskSubmission::new("task").key("tb")) .await .unwrap(); diff --git a/tests/integration/modules.rs b/tests/integration/modules.rs index 338c56f..96c0d84 100644 --- a/tests/integration/modules.rs +++ b/tests/integration/modules.rs @@ -4,30 +4,17 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use std::time::Duration; +use serde::{Deserialize, Serialize}; use taskmill::{ Domain, DomainHandle, DomainKey, Scheduler, SchedulerEvent, TaskStatus, TaskStore, - TaskSubmission, + TaskSubmission, TypedTask, }; use tokio_util::sync::CancellationToken; use super::common::*; -// ── Domain key structs ────────────────────────────────────────────── - -struct MediaDomain; -impl DomainKey for MediaDomain { - const NAME: &'static str = "media"; -} - -struct SyncDomain; -impl DomainKey for SyncDomain { - const NAME: &'static str = "sync"; -} - -struct AnalyticsDomain; -impl DomainKey for AnalyticsDomain { - const NAME: &'static str = "analytics"; -} +// ── Local task type for analytics::thumb (not in common.rs) ───────── +define_task!(AnalyticsThumbTask, AnalyticsDomain, "thumb"); // ═══════════════════════════════════════════════════════════════════ // N. Module Registration (Step 3) @@ -40,18 +27,16 @@ async fn two_modules_route_to_correct_executors() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor( - "thumb", - CountingExecutor { + .domain( + Domain::::new().task::(CountingExecutor { count: media_count.clone(), - }, - )) - .domain(Domain::::new().raw_executor( - "push", - CountingExecutor { + }), + ) + .domain( + Domain::::new().task::(CountingExecutor { count: sync_count.clone(), - }, - )) + }), + ) .max_concurrency(4) .build() .await @@ -101,8 +86,8 @@ async fn zero_modules_build_returns_error() { async fn duplicate_module_names_build_returns_error() { let result = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("thumb", NoopExecutor)) - .domain(Domain::::new().raw_executor("transcode", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .build() .await; @@ -123,8 +108,8 @@ async fn task_type_collision_across_modules_returns_error() { // Instead, verify that two distinct modules with distinct types succeed. let result = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("thumb", NoopExecutor)) - .domain(Domain::::new().raw_executor("thumb", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .build() .await; @@ -146,8 +131,8 @@ async fn two_module_scheduler() -> ( ) { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("thumb", NoopExecutor)) - .domain(Domain::::new().raw_executor("push", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .poll_interval(Duration::from_millis(20)) .max_concurrency(8) .build() @@ -284,12 +269,9 @@ async fn module_active_tasks_returns_only_own_module() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) .domain( - Domain::::new() - .raw_executor("thumb", DelayExecutor(Duration::from_secs(5))), - ) - .domain( - Domain::::new().raw_executor("push", DelayExecutor(Duration::from_secs(5))), + Domain::::new().task::(DelayExecutor(Duration::from_secs(5))), ) + .domain(Domain::::new().task::(DelayExecutor(Duration::from_secs(5)))) .poll_interval(Duration::from_millis(20)) .max_concurrency(8) .build() @@ -347,18 +329,16 @@ async fn module_subscribe_receives_only_own_events() { let count = Arc::new(AtomicUsize::new(0)); let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor( - "thumb", - CountingExecutor { + .domain( + Domain::::new().task::(CountingExecutor { count: count.clone(), - }, - )) - .domain(Domain::::new().raw_executor( - "push", - CountingExecutor { + }), + ) + .domain( + Domain::::new().task::(CountingExecutor { count: count.clone(), - }, - )) + }), + ) .poll_interval(Duration::from_millis(20)) .max_concurrency(8) .build() @@ -440,7 +420,7 @@ impl DomainKey for NonexistentDomain { async fn scheduler_module_nonexistent_panics() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("thumb", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .build() .await .unwrap(); @@ -452,7 +432,7 @@ async fn scheduler_module_nonexistent_panics() { async fn scheduler_try_module_nonexistent_returns_none() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("thumb", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .build() .await .unwrap(); @@ -491,8 +471,8 @@ async fn scheduler_task_returns_regardless_of_module() { async fn module_registry_stored_in_scheduler() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("thumb", NoopExecutor)) - .domain(Domain::::new().raw_executor("push", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .build() .await .unwrap(); diff --git a/tests/integration/retry_policy.rs b/tests/integration/retry_policy.rs index 4a2bf91..54bf142 100644 --- a/tests/integration/retry_policy.rs +++ b/tests/integration/retry_policy.rs @@ -4,17 +4,12 @@ use std::time::Duration; use serde::{Deserialize, Serialize}; use taskmill::{ - BackoffStrategy, Domain, DomainKey, RetryPolicy, Scheduler, SchedulerEvent, TaskContext, - TaskError, TaskExecutor, TaskStore, TaskSubmission, TaskTypeConfig, TypedExecutor, TypedTask, + BackoffStrategy, Domain, RetryPolicy, Scheduler, SchedulerEvent, TaskContext, TaskError, + TaskStore, TaskSubmission, TaskTypeConfig, TypedExecutor, TypedTask, }; use tokio_util::sync::CancellationToken; -// ── Domain Key ───────────────────────────────────────────────────── - -struct TestDomain; -impl DomainKey for TestDomain { - const NAME: &'static str = "test"; -} +use super::common::{define_task, TestDomain}; // ── Typed Tasks ──────────────────────────────────────────────────── @@ -83,6 +78,11 @@ impl TypedTask for RetryEventTask { } } +// ── Local task types for formerly-untyped executors ──────────────── + +define_task!(LegacyTask, TestDomain, "legacy"); +define_task!(RetryOverrideTask, TestDomain, "retry-override"); + // ── Typed Executors ──────────────────────────────────────────────── /// Always fails with a retryable error. Works as a TypedExecutor for any task type. @@ -94,20 +94,28 @@ impl TypedExecutor for AlwaysRetryableTypedExec { } } -/// Always fails with a retryable error (untyped, for raw_executor). +/// Always fails with a retryable error. struct AlwaysRetryableExecutor; -impl TaskExecutor for AlwaysRetryableExecutor { - async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for AlwaysRetryableExecutor { + async fn execute<'a>( + &'a self, + _payload: LegacyTask, + _ctx: &'a TaskContext, + ) -> Result<(), TaskError> { Err(TaskError::retryable("transient")) } } -/// Fails with a retryable error and requests a specific retry delay (untyped). +/// Fails with a retryable error and requests a specific retry delay. struct RetryAfterExecutor(Duration); -impl TaskExecutor for RetryAfterExecutor { - async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> { +impl TypedExecutor for RetryAfterExecutor { + async fn execute<'a>( + &'a self, + _payload: RetryOverrideTask, + _ctx: &'a TaskContext, + ) -> Result<(), TaskError> { Err(TaskError::retryable("rate limited").retry_after(self.0)) } } @@ -322,10 +330,10 @@ async fn failed_event_includes_retry_after_duration() { async fn failed_event_includes_executor_retry_after_override() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor( - "retry-override", - RetryAfterExecutor(Duration::from_secs(42)), - )) + .domain( + Domain::::new() + .task::(RetryAfterExecutor(Duration::from_secs(42))), + ) .max_retries(3) .max_concurrency(1) .poll_interval(Duration::from_millis(50)) @@ -379,7 +387,7 @@ async fn failed_event_includes_executor_retry_after_override() { async fn null_max_retries_uses_global_default() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("legacy", AlwaysRetryableExecutor)) + .domain(Domain::::new().task::(AlwaysRetryableExecutor)) .max_retries(2) .max_concurrency(1) .poll_interval(Duration::from_millis(50)) diff --git a/tests/integration/scheduler_core.rs b/tests/integration/scheduler_core.rs index 8bb9474..e005062 100644 --- a/tests/integration/scheduler_core.rs +++ b/tests/integration/scheduler_core.rs @@ -7,16 +7,14 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use std::time::Duration; -use taskmill::{Domain, DomainKey, Priority, Scheduler, SchedulerEvent, TaskStore, TaskSubmission}; +use serde::{Deserialize, Serialize}; +use taskmill::{Domain, Priority, Scheduler, SchedulerEvent, TaskStore, TaskSubmission, TypedTask}; use tokio_util::sync::CancellationToken; use super::common::*; -/// Domain key for all tests in this module. -pub struct TestDomain; -impl DomainKey for TestDomain { - const NAME: &'static str = "test"; -} +// Additional task type not in common.rs +define_task!(CountingTask, TestDomain, "counting"); // ═══════════════════════════════════════════════════════════════════ // A. Priority & Ordering @@ -26,7 +24,7 @@ impl DomainKey for TestDomain { async fn priority_ordering_dispatches_highest_first() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("test", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .max_concurrency(1) // dispatch one at a time .build() .await @@ -85,13 +83,12 @@ async fn priority_ordering_dispatches_highest_first() { async fn retryable_error_retries_then_succeeds() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor( - "test", - FailNTimesExecutor { + .domain( + Domain::::new().task::(FailNTimesExecutor { failures: std::sync::atomic::AtomicI32::new(0), max_failures: 2, - }, - )) + }), + ) .max_retries(3) .max_concurrency(1) .build() @@ -130,13 +127,12 @@ async fn retryable_error_retries_then_succeeds() { async fn retryable_error_exhausts_retries() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor( - "test", - FailNTimesExecutor { + .domain( + Domain::::new().task::(FailNTimesExecutor { failures: std::sync::atomic::AtomicI32::new(0), max_failures: 100, // will never succeed - }, - )) + }), + ) .max_retries(2) .max_concurrency(1) .build() @@ -183,8 +179,8 @@ async fn preemption_resumes_after_preemptor_completes() { .store(TaskStore::open_memory().await.unwrap()) .domain( Domain::::new() - .raw_executor("slow", DelayExecutor(Duration::from_secs(10))) - .raw_executor("fast", NoopExecutor), + .task::(DelayExecutor(Duration::from_secs(10))) + .task::(NoopExecutor), ) .max_concurrency(1) .preempt_priority(Priority::REALTIME) @@ -261,7 +257,7 @@ async fn backpressure_throttles_low_priority_tasks() { // Default three-tier policy: BACKGROUND throttled >50%. let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("test", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .pressure_source(Box::new(FixedPressure { value: 0.6, name: "test-pressure", @@ -305,7 +301,7 @@ async fn backpressure_throttles_low_priority_tasks() { async fn backpressure_blocks_normal_at_high_pressure() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("test", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .pressure_source(Box::new(FixedPressure { value: 0.8, name: "test-pressure", @@ -356,14 +352,13 @@ async fn group_concurrency_limits_dispatch() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor( - "test", - ConcurrencyTrackingExecutor { + .domain( + Domain::::new().task::(ConcurrencyTrackingExecutor { current: current.clone(), max_seen: max_seen.clone(), delay: Duration::from_millis(100), - }, - )) + }), + ) .max_concurrency(10) // high global limit .group_concurrency("s3-bucket", 2) // but group capped at 2 .poll_interval(Duration::from_millis(50)) @@ -424,12 +419,11 @@ async fn run_loop_processes_queue_to_completion() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor( - "test", - CountingExecutor { + .domain( + Domain::::new().task::(CountingExecutor { count: count.clone(), - }, - )) + }), + ) .max_concurrency(4) .poll_interval(Duration::from_millis(50)) .build() @@ -481,14 +475,13 @@ async fn concurrent_tasks_respect_max_concurrency() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor( - "test", - ConcurrencyTrackingExecutor { + .domain( + Domain::::new().task::(ConcurrencyTrackingExecutor { current: current.clone(), max_seen: max_seen.clone(), delay: Duration::from_millis(50), - }, - )) + }), + ) .max_concurrency(2) .poll_interval(Duration::from_millis(20)) .build() @@ -542,15 +535,12 @@ async fn fail_fast_cancels_siblings_on_child_failure() { .store(TaskStore::open_memory().await.unwrap()) .domain( Domain::::new() - .raw_executor( - "parent", - ChildSpawnerExecutor { - child_type: "child", - count: 3, - fail_fast: true, - }, - ) - .raw_executor("child", AlwaysFailExecutor), + .task::(ChildSpawnerExecutor { + child_type: "child", + count: 3, + fail_fast: true, + }) + .task::(AlwaysFailExecutor), ) .max_concurrency(4) .max_retries(0) // no retries so failures are permanent @@ -603,14 +593,11 @@ async fn non_fail_fast_waits_for_all_children() { .store(TaskStore::open_memory().await.unwrap()) .domain( Domain::::new() - .raw_executor( - "parent", - FinalizeTracker { - child_count: 2, - finalized: finalized.clone(), - }, - ) - .raw_executor("child", NoopExecutor), + .task::(FinalizeTracker { + child_count: 2, + finalized: finalized.clone(), + }) + .task::(NoopExecutor), ) .max_concurrency(4) .poll_interval(Duration::from_millis(50)) @@ -703,7 +690,7 @@ async fn running_tasks_reset_to_pending_on_restart() { async fn submit_batch_enqueues_all_tasks() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("test", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .build() .await .unwrap(); @@ -731,13 +718,12 @@ async fn submit_batch_enqueues_all_tasks() { async fn io_metrics_recorded_in_history() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor( - "test", - IoReportingExecutor { + .domain( + Domain::::new().task::(IoReportingExecutor { read: 4096, write: 1024, - }, - )) + }), + ) .build() .await .unwrap(); @@ -767,7 +753,7 @@ async fn io_metrics_recorded_in_history() { async fn snapshot_reflects_pressure_breakdown() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("test", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .pressure_source(Box::new(FixedPressure { value: 0.42, name: "api-load", @@ -1004,12 +990,11 @@ async fn delayed_task_full_scheduler_lifecycle() { let count = Arc::new(AtomicUsize::new(0)); let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor( - "counting", - CountingExecutor { + .domain( + Domain::::new().task::(CountingExecutor { count: count.clone(), - }, - )) + }), + ) .poll_interval(Duration::from_millis(50)) .build() .await @@ -1035,7 +1020,7 @@ async fn delayed_task_full_scheduler_lifecycle() { async fn recurring_task_snapshot_includes_schedules() { let sched = Scheduler::builder() .store(TaskStore::open_memory().await.unwrap()) - .domain(Domain::::new().raw_executor("test", NoopExecutor)) + .domain(Domain::::new().task::(NoopExecutor)) .build() .await .unwrap();