feat: implement sibling task spawning with automatic parent-ID inheritance

docs/quick-start.md

@@ -308,6 +308,56 @@ scheduler.submit(
 ).await?;
 ```
 
+## Sibling tasks
+
+When a child task needs to spawn peer tasks under the same parent (flat hierarchy), use `ctx.spawn_sibling_with()` instead of manually extracting and threading the parent ID. The new task's `parent_id` is set to the current task's `parent_id`, making it a peer under the same orchestrator.
+
+```rust
+impl TypedExecutor<ScanL1DirTask> for DirScanner {
+    async fn execute<'a>(
+        &'a self, task: ScanL1DirTask, ctx: DomainTaskContext<'a, Scanner>,
+    ) -> Result<(), TaskError> {
+        for subdir in list_subdirs(&task.prefix).await? {
+            ctx.spawn_sibling_with(ScanL1DirTask {
+                bucket: task.bucket.clone(),
+                prefix: subdir,
+            })
+            .key(&format!("{}:{}", task.bucket, subdir))
+            .await?;
+        }
+        Ok(())
+    }
+}
+```
+
+For high-fan-out patterns, use `spawn_siblings_with()` which routes through a single-transaction batch path:
+
+```rust
+let siblings: Vec<ScanL1DirTask> = subdirs.into_iter()
+    .map(|d| ScanL1DirTask { bucket: bucket.clone(), prefix: d })
+    .collect();
+ctx.spawn_siblings_with(siblings).await?;
+```
+
+If the current task has no parent (i.e. it's a root task), both methods return `StoreError::InvalidState` instead of silently creating a root task.
+
+For cross-domain siblings, use `.sibling_of(&ctx)` on the submit builder:
+
+```rust
+ctx.domain::<Analytics>()
+    .submit_with(ScanStartedEvent { .. })
+    .sibling_of(&ctx)?
+    .priority(Priority::HIGH)
+    .await?;
+```
+
+| Method | `parent_id` on new task |
+|---|---|
+| `submit_with(task)` | `None` (root) |
+| `submit_with(task).parent(id)` | Explicit ID |
+| `ctx.spawn_child_with(task)` | Current task's ID |
+| `ctx.spawn_sibling_with(task)` | Current task's `parent_id` |
+
 ## Sharing the scheduler
 
 A single `Scheduler` is `Clone` (via `Arc`) and can be shared across your entire application. Domains can carry their own scoped state, so library domains don't need to share a namespace with the host app's state.

examples/test_sibling.rs

@@ -0,0 +1,154 @@
+use serde::{Deserialize, Serialize};
+use std::time::Duration;
+use taskmill::store::TaskStore;
+use taskmill::{
+    Domain, DomainKey, DomainTaskContext, Scheduler, SchedulerEvent, TaskError, TaskSubmission,
+    TypedExecutor, TypedTask,
+};
+use tokio_util::sync::CancellationToken;
+
+struct TestDomain;
+impl DomainKey for TestDomain {
+    const NAME: &'static str = "test";
+}
+
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+struct OrcTask;
+impl TypedTask for OrcTask {
+    type Domain = TestDomain;
+    const TASK_TYPE: &'static str = "orc";
+}
+
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+struct ChildT;
+impl TypedTask for ChildT {
+    type Domain = TestDomain;
+    const TASK_TYPE: &'static str = "child";
+}
+
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+struct SibT;
+impl TypedTask for SibT {
+    type Domain = TestDomain;
+    const TASK_TYPE: &'static str = "sib";
+}
+
+struct OrcExec;
+impl TypedExecutor<OrcTask> for OrcExec {
+    async fn execute<'a>(
+        &'a self,
+        _: OrcTask,
+        ctx: DomainTaskContext<'a, TestDomain>,
+    ) -> Result<(), TaskError> {
+        ctx.spawn_child_with(ChildT)
+            .key("c0")
+            .await
+            .map_err(|e| TaskError::new(e.to_string()))?;
+        Ok(())
+    }
+}
+
+struct ChildExec;
+impl TypedExecutor<ChildT> for ChildExec {
+    async fn execute<'a>(
+        &'a self,
+        _: ChildT,
+        ctx: DomainTaskContext<'a, TestDomain>,
+    ) -> Result<(), TaskError> {
+        eprintln!(
+            "  ChildExec: spawning sibling (my parent_id={:?})",
+            ctx.record().parent_id
+        );
+        let outcome = ctx
+            .spawn_sibling_with(SibT)
+            .key("s0")
+            .await
+            .map_err(|e| TaskError::new(e.to_string()))?;
+        eprintln!("  ChildExec: sibling spawned, outcome={:?}", outcome);
+        Ok(())
+    }
+}
+
+struct SibExec;
+impl TypedExecutor<SibT> for SibExec {
+    async fn execute<'a>(
+        &'a self,
+        _: SibT,
+        ctx: DomainTaskContext<'a, TestDomain>,
+    ) -> Result<(), TaskError> {
+        eprintln!(
+            "  SibExec: running (parent_id={:?})",
+            ctx.record().parent_id
+        );
+        Ok(())
+    }
+}
+
+#[tokio::main]
+async fn main() {
+    let store = TaskStore::open_memory().await.unwrap();
+    let qs = store.clone();
+
+    let sched = Scheduler::builder()
+        .store(store)
+        .domain(
+            Domain::<TestDomain>::new()
+                .task::<OrcTask>(OrcExec)
+                .task::<ChildT>(ChildExec)
+                .task::<SibT>(SibExec),
+        )
+        .max_concurrency(4)
+        .poll_interval(Duration::from_millis(20))
+        .build()
+        .await
+        .unwrap();
+
+    let orc = sched
+        .submit(&TaskSubmission::new("test::orc").key("o0"))
+        .await
+        .unwrap();
+    eprintln!("Orchestrator submitted: {:?}", orc);
+
+    let mut rx = sched.subscribe();
+    let token = CancellationToken::new();
+    let sc = sched.clone();
+    let tc = token.clone();
+    let h = tokio::spawn(async move {
+        sc.run(tc).await;
+    });
+
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
+    while tokio::time::Instant::now() < deadline {
+        match tokio::time::timeout(Duration::from_millis(200), rx.recv()).await {
+            Ok(Ok(evt)) => {
+                if let Some(hdr) = evt.header() {
+                    eprintln!(
+                        "Event: {:?} type={} id={}",
+                        std::mem::discriminant(&evt),
+                        hdr.task_type,
+                        hdr.task_id
+                    );
+                } else {
+                    eprintln!("Event: {:?}", evt);
+                }
+                if matches!(&evt, SchedulerEvent::Completed(h) if h.task_type == "test::orc") {
+                    break;
+                }
+            }
+            _ => {}
+        }
+    }
+
+    // Check history
+    for id in 1..=10 {
+        if let Ok(Some(h)) = qs.history_by_id(id).await {
+            eprintln!(
+                "History id={}: type={} parent_id={:?}",
+                h.id, h.task_type, h.parent_id
+            );
+        }
+    }
+
+    token.cancel();
+    let _ = h.await;
+}

src/domain.rs

@@ -962,6 +962,29 @@ impl<D: DomainKey> DomainSubmitBuilder<D> {
         self.parent(ctx.record().id)
     }
 
+    /// Mark this task as a sibling of the task currently executing in the
+    /// given [`DomainTaskContext`] (shares the same parent).
+    ///
+    /// Returns `Err(StoreError::InvalidState)` if the context task has no
+    /// `parent_id`.
+    ///
+    /// ```ignore
+    /// ctx.domain::<Analytics>()
+    ///     .submit_with(ScanStartedEvent { .. })
+    ///     .sibling_of(&ctx)?
+    ///     .priority(Priority::HIGH)
+    ///     .await?;
+    /// ```
+    pub fn sibling_of<D2: DomainKey>(
+        self,
+        ctx: &DomainTaskContext<'_, D2>,
+    ) -> Result<Self, StoreError> {
+        let pid = ctx.record().parent_id.ok_or_else(|| {
+            StoreError::InvalidState("sibling_of called on a task with no parent_id".into())
+        })?;
+        Ok(self.parent(pid))
+    }
+
     /// Submit the task, returning the outcome.
     pub async fn submit(self) -> Result<SubmitOutcome, StoreError> {
         self.inner.submit().await

src/lib.rs

@@ -139,7 +139,7 @@
 //! [`SchedulerBuilder::group_minimum_slots`], or adjust at runtime via
 //! [`Scheduler::set_group_weight`] and [`Scheduler::set_group_minimum_slots`].
 //!
-//! ## Child tasks & two-phase execution
+//! ## Child tasks, sibling tasks & two-phase execution
 //!
 //! An executor can spawn child tasks via [`DomainTaskContext::spawn_child_with`]. When
 //! children exist, the parent enters a **waiting** state after its executor
@@ -151,6 +151,15 @@
 //! [`DomainSubmitBuilder::fail_fast(false)`](DomainSubmitBuilder::fail_fast)
 //! (or [`TaskSubmission::fail_fast`] for untyped submissions).
 //!
+//! A child executor can also spawn **sibling** tasks — peers under the same
+//! parent — via [`DomainTaskContext::spawn_sibling_with`]. This avoids manually
+//! extracting and threading the parent ID, and returns
+//! [`StoreError::InvalidState`] if the current task has no parent (instead of
+//! silently creating a root task). A batch variant
+//! [`DomainTaskContext::spawn_siblings_with`] uses a single-transaction path
+//! for high-fan-out patterns (e.g. BFS directory scans). For cross-domain
+//! siblings, use [`DomainSubmitBuilder::sibling_of`].
+//!
 //! ## Task TTL & automatic expiry
 //!
 //! Tasks can be given a time-to-live (TTL) so they expire automatically if they
@@ -839,7 +848,7 @@ pub use domain::{
     Domain, DomainHandle, DomainKey, DomainSubmitBuilder, TaskEvent, TaskTypeConfig,
     TaskTypeOptions, TypedEventStream, TypedExecutor,
 };
-pub use registry::{ChildSpawnBuilder, DomainTaskContext};
+pub use registry::{ChildSpawnBuilder, DomainTaskContext, SiblingSpawnBuilder};
 
 // ── Core re-exports ──────────────────────────────────────────────────
 pub use backpressure::{CompositePressure, PressureSource, ThrottlePolicy};

src/module.rs

@@ -20,7 +20,7 @@ use crate::task::retry::RetryPolicy;
 use crate::task::submit_builder::TypedTaskDefaults;
 use crate::task::{ModuleSubmitDefaults, SubmitBuilder};
 use crate::task::{
-    SubmitOutcome, TaskHistoryRecord, TaskRecord, TaskStatus, TaskSubmission, TypedTask,
+    SubmitOutcome, TaskHistoryRecord, TaskRecord, TaskStatus, TaskSubmission, TtlFrom, TypedTask,
 };
 
 /// Per-executor options for task type registration within a module.
@@ -487,6 +487,58 @@ impl ModuleHandle {
         self.submit(sub).with_typed_defaults(typed_defaults)
     }
 
+    /// Submit multiple tasks in a single transaction.
+    ///
+    /// Each submission gets the module prefix and defaults applied (the
+    /// `submit()` path — non-typed defaults fill gaps only). When all
+    /// submissions share the same `parent_id`, the parent record is fetched
+    /// once and used to inherit remaining TTL and tags for the batch.
+    ///
+    /// Used by `spawn_children_with` and `spawn_siblings_with` for
+    /// single-transaction efficiency.
+    pub(crate) async fn submit_batch(
+        &self,
+        submissions: Vec<TaskSubmission>,
+    ) -> Result<Vec<SubmitOutcome>, StoreError> {
+        let mut resolved = Vec::with_capacity(submissions.len());
+        for sub in submissions {
+            // Reuse SubmitBuilder::resolve() for prefix + defaults (submit() path).
+            let builder = SubmitBuilder::new(
+                sub,
+                self.scheduler.clone(),
+                self.name.as_ref(),
+                self.defaults.clone(),
+            );
+            let (_sched, r) = builder.resolve_only();
+            resolved.push(r);
+        }
+
+        // Inherit parent TTL and tags for all submissions sharing a parent_id.
+        // Fetch the parent record once and apply to all.
+        if let Some(pid) = resolved.first().and_then(|s| s.parent_id) {
+            if let Ok(Some(parent)) = self.scheduler.store().task_by_id(pid).await {
+                for s in &mut resolved {
+                    // TTL inheritance (only if no layer set a TTL)
+                    if s.ttl.is_none() {
+                        if let Some(remaining) = parent.remaining_ttl() {
+                            s.ttl = Some(remaining);
+                            s.ttl_from = TtlFrom::Submission;
+                        }
+                    }
+                    // Tag inheritance (parent fills gaps)
+                    for (k, v) in &parent.tags {
+                        s.tags.entry(k.clone()).or_insert_with(|| v.clone());
+                    }
+                }
+            }
+        }
+
+        self.scheduler
+            .submit_batch(&resolved)
+            .await
+            .map(|batch_outcome| batch_outcome.outcomes)
+    }
+
     // ── Single-task operations ────────────────────────────────────
 
     /// Cancel a task by ID.

src/registry/context.rs

@@ -314,12 +314,11 @@ impl TaskContext {
             return spawner.spawn_batch(&mut submissions).await;
         }
 
-        // Module-aware path: prepare each submission then submit via module handle.
-        let mut outcomes = Vec::with_capacity(submissions.len());
-        for sub in submissions {
-            let sub = spawner.prepare(sub);
-            outcomes.push(self.current_module().submit(sub).await?);
-        }
-        Ok(outcomes)
+        // Module-aware path: prepare each submission then batch-submit via module handle.
+        let prepared: Vec<_> = submissions
+            .into_iter()
+            .map(|sub| spawner.prepare(sub))
+            .collect();
+        self.current_module().submit_batch(prepared).await
     }
 }