Add the ability to spawn tasks by name (#22)

* wip

* made durable generic over a context type

* renamed app context to state

* made TaskContext generic over app state

* give callers the ability to spawn pre-registered tasks by name

* added a test that covers spawning by name from context

* addressed PR comment

Author: virajmehta

src/client.rs

@@ -282,14 +282,26 @@ where
             .await
     }
 
-    /// Spawn a task by name (dynamic version for unregistered tasks)
+    /// Spawn a task by name (dynamic version).
+    ///
+    /// The task must be registered before spawning.
     pub async fn spawn_by_name(
         &self,
         task_name: &str,
         params: JsonValue,
         options: SpawnOptions,
     ) -> anyhow::Result<SpawnResult> {
-        self.spawn_by_name_with(&self.pool, task_name, params, options)
+        // Validate that the task is registered
+        {
+            let registry = self.registry.read().await;
+            anyhow::ensure!(
+                registry.contains_key(task_name),
+                "Unknown task: {}. Task must be registered before spawning.",
+                task_name
+            );
+        }
+
+        self.spawn_by_name_internal(&self.pool, task_name, params, options)
             .await
     }
 
@@ -335,18 +347,46 @@ where
         T: Task<State>,
         E: Executor<'e, Database = Postgres>,
     {
-        self.spawn_by_name_with(executor, T::NAME, serde_json::to_value(&params)?, options)
+        // Type-safe spawn uses T::NAME which is already registered
+        self.spawn_by_name_internal(executor, T::NAME, serde_json::to_value(&params)?, options)
             .await
     }
 
     /// Spawn a task by name using a custom executor.
+    ///
+    /// The task must be registered before spawning.
     pub async fn spawn_by_name_with<'e, E>(
         &self,
         executor: E,
         task_name: &str,
         params: JsonValue,
         options: SpawnOptions,
     ) -> anyhow::Result<SpawnResult>
+    where
+        E: Executor<'e, Database = Postgres>,
+    {
+        // Validate that the task is registered
+        {
+            let registry = self.registry.read().await;
+            anyhow::ensure!(
+                registry.contains_key(task_name),
+                "Unknown task: {}. Task must be registered before spawning.",
+                task_name
+            );
+        }
+
+        self.spawn_by_name_internal(executor, task_name, params, options)
+            .await
+    }
+
+    /// Internal spawn implementation without registry validation.
+    async fn spawn_by_name_internal<'e, E>(
+        &self,
+        executor: E,
+        task_name: &str,
+        params: JsonValue,
+        options: SpawnOptions,
+    ) -> anyhow::Result<SpawnResult>
     where
         E: Executor<'e, Database = Postgres>,
     {

src/context.rs

@@ -4,11 +4,13 @@ use serde_json::Value as JsonValue;
 use sqlx::PgPool;
 use std::collections::HashMap;
 use std::marker::PhantomData;
+use std::sync::Arc;
 use std::time::Duration;
+use tokio::sync::RwLock;
 use uuid::Uuid;
 
 use crate::error::{ControlFlow, TaskError, TaskResult};
-use crate::task::Task;
+use crate::task::{Task, TaskRegistry};
 use crate::types::{
     AwaitEventResult, CheckpointRow, ChildCompletePayload, ChildStatus, ClaimedTask, SpawnOptions,
     SpawnResultRow, TaskHandle,
@@ -67,6 +69,9 @@ where
     /// Notifies the worker when the lease is extended via step() or heartbeat().
     lease_extender: LeaseExtender,
 
+    /// Task registry for validating spawn_by_name calls.
+    registry: Arc<RwLock<TaskRegistry<State>>>,
+
     /// Phantom data to carry the State type parameter.
     _state: PhantomData<State>,
 }
@@ -93,6 +98,7 @@ where
         task: ClaimedTask,
         claim_timeout: u64,
         lease_extender: LeaseExtender,
+        registry: Arc<RwLock<TaskRegistry<State>>>,
     ) -> Result<Self, sqlx::Error> {
         // Load all checkpoints for this task into cache
         let checkpoints: Vec<CheckpointRow> = sqlx::query_as(
@@ -120,6 +126,7 @@ where
             checkpoint_cache: cache,
             step_counters: HashMap::new(),
             lease_extender,
+            registry,
             _state: PhantomData,
         })
     }
@@ -456,6 +463,57 @@ where
     where
         T: Task<State>,
     {
+        let params_json = serde_json::to_value(&params)?;
+        self.spawn_by_name(name, T::NAME, params_json, options)
+            .await
+    }
+
+    /// Spawn a subtask by task name (dynamic version).
+    ///
+    /// This is similar to [`spawn`](Self::spawn) but works with task names
+    /// instead of requiring a concrete type. Useful for dynamic task invocation
+    /// where the task type isn't known at compile time.
+    ///
+    /// The spawn is checkpointed - if this task retries after spawning, the
+    /// same subtask ID is returned without spawning a duplicate.
+    ///
+    /// # Arguments
+    ///
+    /// * `name` - Unique name for this spawn operation (used for checkpointing)
+    /// * `task_name` - The registered name of the task to spawn
+    /// * `params` - JSON parameters to pass to the task
+    /// * `options` - Spawn options (max_attempts, priority, etc.)
+    ///
+    /// # Returns
+    ///
+    /// A [`TaskHandle`] that can be passed to [`join`](Self::join) to wait for
+    /// the result. The output type `T` must match the actual task's output type.
+    ///
+    /// # Errors
+    ///
+    /// * `TaskError::Failed` - If the task name is not registered in the registry
+    ///
+    /// # Example
+    ///
+    /// ```ignore
+    /// // Spawn a task by name
+    /// let handle: TaskHandle<ProcessResult> = ctx.spawn_by_name(
+    ///     "process-item",
+    ///     "process-item-task",
+    ///     serde_json::json!({ "item_id": 123 }),
+    ///     Default::default(),
+    /// ).await?;
+    ///
+    /// // Wait for result
+    /// let result: ProcessResult = ctx.join("process-item", handle).await?;
+    /// ```
+    pub async fn spawn_by_name<T: DeserializeOwned>(
+        &mut self,
+        name: &str,
+        task_name: &str,
+        params: JsonValue,
+        options: SpawnOptions,
+    ) -> TaskResult<TaskHandle<T>> {
         validate_user_name(name)?;
         let checkpoint_name = self.get_checkpoint_name(&format!("$spawn:{name}"));
 
@@ -465,8 +523,18 @@ where
             return Ok(TaskHandle::new(task_id));
         }
 
+        // Validate that the task is registered
+        {
+            let registry = self.registry.read().await;
+            if !registry.contains_key(task_name) {
+                return Err(TaskError::Failed(anyhow::anyhow!(
+                    "Unknown task: {}. Task must be registered before spawning.",
+                    task_name
+                )));
+            }
+        }
+
         // Build options JSON, merging user options with parent_task_id
-        let params_json = serde_json::to_value(&params)?;
         #[derive(Serialize)]
         struct SubtaskOptions<'a> {
             parent_task_id: Uuid,
@@ -482,8 +550,8 @@ where
             "SELECT task_id, run_id, attempt FROM durable.spawn_task($1, $2, $3, $4)",
         )
         .bind(&self.queue_name)
-        .bind(T::NAME)
-        .bind(&params_json)
+        .bind(task_name)
+        .bind(&params)
         .bind(&options_json)
         .fetch_one(&self.pool)
         .await?;

src/worker.rs

@@ -247,6 +247,7 @@ impl Worker {
             task.clone(),
             claim_timeout,
             lease_extender,
+            registry.clone(),
         )
         .await
         {

tests/common/tasks.rs

@@ -1241,3 +1241,51 @@ impl Task<()> for SpawnThenFailTask {
         }))
     }
 }
+
+// ============================================================================
+// SpawnByNameTask - Tests spawn_by_name on TaskContext
+// ============================================================================
+
+/// Parent task that spawns a child using spawn_by_name (dynamic version)
+#[allow(dead_code)]
+pub struct SpawnByNameTask;
+
+#[allow(dead_code)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct SpawnByNameParams {
+    pub child_value: i32,
+}
+
+#[allow(dead_code)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct SpawnByNameOutput {
+    pub child_result: i32,
+}
+
+#[async_trait]
+impl Task<()> for SpawnByNameTask {
+    const NAME: &'static str = "spawn-by-name";
+    type Params = SpawnByNameParams;
+    type Output = SpawnByNameOutput;
+
+    async fn run(
+        params: Self::Params,
+        mut ctx: TaskContext,
+        _state: (),
+    ) -> TaskResult<Self::Output> {
+        // Spawn child task using spawn_by_name (dynamic version)
+        let handle: TaskHandle<i32> = ctx
+            .spawn_by_name(
+                "child",
+                "double", // task name string instead of type
+                serde_json::json!({ "value": params.child_value }),
+                Default::default(),
+            )
+            .await?;
+
+        // Join and get result
+        let child_result: i32 = ctx.join("child", handle).await?;
+
+        Ok(SpawnByNameOutput { child_result })
+    }
+}

tests/execution_test.rs

@@ -349,32 +349,24 @@ async fn test_unregistered_task_fails(pool: PgPool) -> sqlx::Result<()> {
     client.create_queue(None).await.unwrap();
     // Note: We don't register any task handler
 
-    // Spawn a task by name
+    // Spawn a task by name - should fail at spawn time because task is not registered
     let spawn_result = client
         .spawn_by_name(
             "unregistered-task",
             serde_json::json!({}),
             Default::default(),
         )
-        .await
-        .expect("Failed to spawn task");
-
-    let worker = client
-        .start_worker(WorkerOptions {
-            poll_interval: 0.05,
-            claim_timeout: 30,
-            ..Default::default()
-        })
         .await;
 
-    tokio::time::sleep(Duration::from_millis(500)).await;
-    worker.shutdown().await;
-
-    // Task should have failed because handler is not registered
-    let state = get_task_state(&pool, "exec_unreg", spawn_result.task_id).await;
-    assert_eq!(
-        state, "failed",
-        "Task with unregistered handler should fail"
+    assert!(
+        spawn_result.is_err(),
+        "Spawning an unregistered task should fail"
+    );
+    let err = spawn_result.unwrap_err();
+    assert!(
+        err.to_string().contains("Unknown task"),
+        "Error should mention 'Unknown task': {}",
+        err
     );
 
     Ok(())

tests/fanout_test.rs

@@ -4,8 +4,9 @@ mod common;
 
 use common::tasks::{
     DoubleTask, FailingChildTask, MultiSpawnOutput, MultiSpawnParams, MultiSpawnTask,
-    SingleSpawnOutput, SingleSpawnParams, SingleSpawnTask, SlowChildTask, SpawnFailingChildTask,
-    SpawnSlowChildParams, SpawnSlowChildTask,
+    SingleSpawnOutput, SingleSpawnParams, SingleSpawnTask, SlowChildTask, SpawnByNameOutput,
+    SpawnByNameParams, SpawnByNameTask, SpawnFailingChildTask, SpawnSlowChildParams,
+    SpawnSlowChildTask,
 };
 use durable::{Durable, MIGRATOR, WorkerOptions};
 use sqlx::{AssertSqlSafe, PgPool};
@@ -348,3 +349,53 @@ async fn test_cascade_cancel_when_parent_cancelled(pool: PgPool) -> sqlx::Result
 
     Ok(())
 }
+
+// ============================================================================
+// spawn_by_name Tests
+// ============================================================================
+
+#[sqlx::test(migrator = "MIGRATOR")]
+async fn test_spawn_by_name_from_task_context(pool: PgPool) -> sqlx::Result<()> {
+    let client = create_client(pool.clone(), "fanout_by_name").await;
+    client.create_queue(None).await.unwrap();
+    client.register::<SpawnByNameTask>().await;
+    client.register::<DoubleTask>().await;
+
+    // Spawn parent task that will use spawn_by_name internally
+    let spawn_result = client
+        .spawn::<SpawnByNameTask>(SpawnByNameParams { child_value: 21 })
+        .await
+        .expect("Failed to spawn task");
+
+    // Start worker with concurrency to handle both parent and child
+    let worker = client
+        .start_worker(WorkerOptions {
+            poll_interval: 0.05,
+            claim_timeout: 30,
+            concurrency: 2,
+            ..Default::default()
+        })
+        .await;
+
+    // Wait for tasks to complete
+    tokio::time::sleep(Duration::from_millis(2000)).await;
+    worker.shutdown().await;
+
+    // Verify parent task completed
+    let state = get_task_state(&pool, "fanout_by_name", spawn_result.task_id).await;
+    assert_eq!(state, "completed", "Parent task should be completed");
+
+    // Verify result
+    let result = get_task_result(&pool, "fanout_by_name", spawn_result.task_id)
+        .await
+        .expect("Task should have a result");
+
+    let output: SpawnByNameOutput =
+        serde_json::from_value(result).expect("Failed to deserialize result");
+    assert_eq!(
+        output.child_result, 42,
+        "Child should have doubled 21 to 42 (spawned via spawn_by_name)"
+    );
+
+    Ok(())
+}