refactor: decompose internal god objects into focused, single-responsibility modules

src/scheduler/mod.rs

@@ -28,6 +28,7 @@ pub(crate) mod gate;
 pub mod progress;
 mod queries;
 mod run_loop;
+pub(crate) mod spawn;
 mod submit;
 #[cfg(test)]
 mod tests;

src/scheduler/run_loop.rs

@@ -10,8 +10,8 @@ use crate::task::IoBudget;
 
 use super::SchedulerEvent;
 
-use super::dispatch::{self, SpawnContext};
 use super::gate::GateContext;
+use super::spawn::{self, SpawnContext};
 use super::{Scheduler, ShutdownMode};
 
 impl Scheduler {
@@ -114,11 +114,11 @@ impl Scheduler {
 
         // Spawn the task — this inserts into the active map, builds the
         // context, emits Dispatched, and wires up completion handling.
-        dispatch::spawn_task(
+        spawn::spawn_task(
             task,
             executor,
             self.build_spawn_context().await,
-            dispatch::ExecutionPhase::Execute,
+            spawn::ExecutionPhase::Execute,
         )
         .await;
 
@@ -168,11 +168,11 @@ impl Scheduler {
         };
         let executor = Arc::clone(executor);
 
-        dispatch::spawn_task(
+        spawn::spawn_task(
             task,
             executor,
             self.build_spawn_context().await,
-            dispatch::ExecutionPhase::Finalize,
+            spawn::ExecutionPhase::Finalize,
         )
         .await;
 

src/scheduler/spawn.rs

@@ -0,0 +1,146 @@
+//! Task spawning — orchestrator and focused submodules.
+//!
+//! This module decomposes the former monolithic `spawn_task` function into
+//! focused, testable units:
+//!
+//! - [`context`] — `SpawnContext` and `TaskContext` construction
+//! - [`completion`] — success path (children check, completion, recurring)
+//! - [`failure`] — failure path (retry, dead-letter, fail-fast cascade)
+//! - [`parent`] — parent-child resolution after task completion
+
+mod completion;
+mod context;
+mod failure;
+mod parent;
+
+use std::sync::atomic::Ordering as AtomicOrdering;
+use std::sync::Arc;
+
+use crate::registry::ErasedExecutor;
+use crate::task::TaskRecord;
+
+use super::dispatch::ActiveTask;
+use super::SchedulerEvent;
+
+pub(crate) use context::SpawnContext;
+
+/// Whether to call `execute` or `finalize` on the executor.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum ExecutionPhase {
+    Execute,
+    Finalize,
+}
+
+/// Spawn a task executor and wire up completion/failure handling.
+///
+/// Inserts the task into the active map, starts a progress listener,
+/// and spawns the executor on a new tokio task. The actual success and
+/// failure handling is delegated to [`completion::handle_success`] and
+/// [`failure::handle_failure`].
+pub(crate) async fn spawn_task(
+    task: TaskRecord,
+    executor: Arc<dyn ErasedExecutor>,
+    ctx: SpawnContext,
+    phase: ExecutionPhase,
+) {
+    let prepared = context::build_task_context(&task, &ctx);
+
+    // Insert into active map before spawning to avoid races.
+    ctx.active.insert(
+        task.id,
+        ActiveTask {
+            record: task.clone(),
+            token: prepared.token.clone(),
+            reported_progress: None,
+            reported_at: None,
+            handle: None,
+            io: prepared.io.clone(),
+            started_at: std::time::Instant::now(),
+        },
+    );
+
+    // Increment the module running counter for this task.
+    if let Some(module_name) = task.module_name() {
+        if let Some(counter) = ctx.module_running.get(module_name) {
+            counter.fetch_add(1, AtomicOrdering::Relaxed);
+        }
+    }
+
+    // Emit dispatched event.
+    let _ = ctx
+        .event_tx
+        .send(SchedulerEvent::Dispatched(task.event_header()));
+
+    // Build deps for handlers (cloned from SpawnContext since they move into the spawned future).
+    let completion_deps = completion::CompletionDeps {
+        store: ctx.store.clone(),
+        active: ctx.active.clone(),
+        event_tx: ctx.event_tx.clone(),
+        work_notify: ctx.work_notify.clone(),
+        max_retries: ctx.max_retries,
+    };
+    let failure_deps = failure::FailureDeps {
+        store: ctx.store,
+        active: ctx.active.clone(),
+        event_tx: ctx.event_tx,
+        work_notify: ctx.work_notify,
+        max_retries: ctx.max_retries,
+        registry: ctx.registry,
+    };
+
+    let task_id_for_handle = task.id;
+    let active_for_handle = ctx.active;
+    let token_for_spawn = prepared.token.clone();
+    let module_running = ctx.module_running;
+    let io = prepared.io;
+
+    let handle = tokio::spawn(async move {
+        let task_id = task.id;
+
+        // Helper: decrement the module running counter when this task leaves "running".
+        let decrement_module = || {
+            if let Some(name) = task.module_name() {
+                if let Some(counter) = module_running.get(name) {
+                    counter.fetch_sub(1, AtomicOrdering::Relaxed);
+                }
+            }
+        };
+
+        let result = match phase {
+            ExecutionPhase::Execute => executor.execute_erased(&prepared.ctx).await,
+            ExecutionPhase::Finalize => executor.finalize_erased(&prepared.ctx).await,
+        };
+
+        // Read IO bytes from the context tracker.
+        let metrics = io.snapshot();
+
+        // Drop the context (and its progress reporter) — executor is done.
+        drop(prepared.ctx);
+
+        match result {
+            Ok(()) => {
+                completion::handle_success(
+                    &task,
+                    phase,
+                    &metrics,
+                    &completion_deps,
+                    decrement_module,
+                )
+                .await;
+            }
+            Err(te) => {
+                // If cancelled (preempted), the scheduler already paused it.
+                if token_for_spawn.is_cancelled() {
+                    decrement_module();
+                    failure_deps.active.remove(task_id);
+                    return;
+                }
+
+                failure::handle_failure(&task, te, &metrics, &failure_deps, decrement_module).await;
+            }
+        }
+    });
+
+    // Store the handle so shutdown can join it.
+    active_for_handle.set_handle(task_id_for_handle, handle);
+}

src/scheduler/spawn/completion.rs

@@ -0,0 +1,130 @@
+//! Success path: children check, completion, recurring re-enqueue, dependency resolution.
+
+use std::sync::Arc;
+
+use crate::store::TaskStore;
+use crate::task::{IoBudget, TaskRecord};
+
+use super::super::dispatch::ActiveTaskMap;
+use super::super::SchedulerEvent;
+use super::parent::handle_parent_resolution;
+use super::ExecutionPhase;
+
+/// Shared dependencies for the completion handler.
+pub(crate) struct CompletionDeps {
+    pub store: TaskStore,
+    pub active: ActiveTaskMap,
+    pub event_tx: tokio::sync::broadcast::Sender<SchedulerEvent>,
+    pub work_notify: Arc<tokio::sync::Notify>,
+    pub max_retries: i32,
+}
+
+/// Handle a successful task execution.
+///
+/// For the execute phase, checks if the task spawned children (transition to
+/// waiting). Otherwise records completion, resolves dependents, and handles
+/// recurring re-enqueue.
+pub(crate) async fn handle_success(
+    task: &TaskRecord,
+    phase: ExecutionPhase,
+    metrics: &IoBudget,
+    deps: &CompletionDeps,
+    decrement_module: impl FnOnce(),
+) {
+    let task_id = task.id;
+
+    // For the execute phase, check if the task spawned children.
+    // If so, transition to waiting instead of completing.
+    if phase == ExecutionPhase::Execute {
+        match deps.store.active_children_count(task_id).await {
+            Ok(count) if count > 0 => {
+                if let Err(e) = deps.store.set_waiting(task_id).await {
+                    tracing::error!(task_id, error = %e, "failed to set task to waiting");
+                }
+                decrement_module();
+                deps.active.remove(task_id);
+                let _ = deps.event_tx.send(SchedulerEvent::Waiting {
+                    task_id,
+                    children_count: count,
+                });
+                // Children may have completed before we set waiting.
+                // Re-check to avoid a missed finalization.
+                handle_parent_resolution(
+                    task_id,
+                    &deps.store,
+                    &deps.active,
+                    &deps.event_tx,
+                    deps.max_retries,
+                    &deps.work_notify,
+                )
+                .await;
+                // Wake the scheduler to dispatch children (or finalizer).
+                deps.work_notify.notify_one();
+                return;
+            }
+            Err(e) => {
+                tracing::error!(task_id, error = %e, "failed to check children count");
+                // Fall through to normal completion.
+            }
+            _ => {
+                // No children — complete normally.
+            }
+        }
+    }
+
+    match deps.store.complete_with_record(task, metrics).await {
+        Ok(recurring_info) => {
+            // Emit recurring event if this was a recurring task.
+            if task.recurring_interval_secs.is_some() {
+                let (next_run, exec_count) = match recurring_info {
+                    Some((next, count)) => (Some(next), count),
+                    None => (None, task.recurring_execution_count + 1),
+                };
+                let _ = deps.event_tx.send(SchedulerEvent::RecurringCompleted {
+                    header: task.event_header(),
+                    execution_count: exec_count,
+                    next_run,
+                });
+            }
+        }
+        Err(e) => {
+            tracing::error!(task_id, error = %e, "failed to record task completion");
+        }
+    }
+
+    // Remove from active tracking AFTER the store write completes.
+    decrement_module();
+    deps.active.remove(task_id);
+    let _ = deps
+        .event_tx
+        .send(SchedulerEvent::Completed(task.event_header()));
+
+    // Resolve dependency edges: unblock tasks waiting on this one.
+    match deps.store.resolve_dependents(task_id).await {
+        Ok(unblocked) => {
+            for uid in &unblocked {
+                let _ = deps
+                    .event_tx
+                    .send(SchedulerEvent::TaskUnblocked { task_id: *uid });
+            }
+        }
+        Err(e) => {
+            tracing::error!(task_id, error = %e, "failed to resolve dependents");
+        }
+    }
+
+    deps.work_notify.notify_one();
+
+    // If this was a child task, check if parent is ready.
+    if let Some(parent_id) = task.parent_id {
+        handle_parent_resolution(
+            parent_id,
+            &deps.store,
+            &deps.active,
+            &deps.event_tx,
+            deps.max_retries,
+            &deps.work_notify,
+        )
+        .await;
+    }
+}