Fix deep workflow graph composition

CHANGELOG.md

@@ -6,6 +6,10 @@ Pre-1.0 note: while `pg_durable` is in major version `0`, minor releases may inc
 
 ## [0.2.6] - Unreleased
 
+### Fixed
+
+- **Deep workflow composition (#327):** workflow graphs deeper than serde_json's 127-level recursion limit no longer silently collapse into SQL text. Nested children are deserialized one graph level at a time, and `df.explain()` now enforces the configured graph-depth limit before traversal.
+
 ## [0.2.5] - 2026-07-30
 
 ### Added

Cargo.toml

@@ -30,7 +30,7 @@ pg_test = []
 [dependencies]
 pgrx = "=0.16.1"
 serde = { version = "1.0", features = ["derive"] }
-serde_json = { version = "1.0", features = ["preserve_order"] }
+serde_json = { version = "1.0", features = ["preserve_order", "raw_value"] }
 uuid = { version = "1.0", features = ["v4", "serde"] }
 percent-encoding = "2.3"
 

USER_GUIDE.md

@@ -309,6 +309,12 @@ SELECT df.start(
 );
 ```
 
+### Graph Depth
+
+A workflow graph can contain at most 256 nested composition levels. `df.start()` and
+`df.explain()` reject deeper graphs with a maximum-nesting-depth error. Split a larger workflow
+into smaller sub-workflows when it would exceed this limit.
+
 ### Variable Substitution
 
 Use `$name` to reference named results in subsequent steps:

docs/upgrade-testing.md

@@ -203,6 +203,14 @@ gate, so they never need to be added to the exclude list.
 Each schema-changing PR should add a section here documenting what changed,
 what the upgrade script handles, and any backward compatibility considerations.
 
+### v0.2.5 → v0.2.6
+
+#### Preserve parser resource errors in `df.ensure_durofut()`
+- **DDL change (function body only):** `df.ensure_durofut(text)` no longer catches `WHEN OTHERS` while deciding whether an operand is Durofut JSON or plain SQL. It still treats `invalid_text_representation` as plain SQL and re-raises its explicit unknown-node-type error, but PostgreSQL stack/resource errors now propagate instead of silently wrapping the serialized graph as a SQL node. The signature, volatility, search path, grants, and schema shape are unchanged.
+- **Upgrade script:** `sql/pg_durable--0.2.5--0.2.6.sql` uses `CREATE OR REPLACE FUNCTION` with the same body emitted for fresh installs from `src/lib.rs`. This keeps Scenario A snapshots identical without dropping the function or changing dependent operators.
+- **Scenario B1 considerations:** The new `.so` works against pre-0.2.6 schemas without runtime schema detection because no Rust SQL query or C symbol changed. Rust composers receive the opaque-child deserialization fix immediately from the new binary. Until `ALTER EXTENSION UPDATE` replaces the cataloged PL/pgSQL helper, the `?>` / `!>` operator path retains its older broad exception handler and can still misclassify a graph if PostgreSQL itself raises a stack/resource error while parsing it.
+- **Scenario B2 considerations:** No data migration and no durable-state or replay change. Existing serialized graphs retain the same wire format.
+
 ### v0.2.4 → v0.2.5
 
 #### Loop and sub-orchestration replay compatibility

sql/pg_durable--0.2.5--0.2.6.sql

@@ -6,5 +6,26 @@
 -- See docs/upgrade-testing.md for the upgrade-script and backward-compatibility
 -- requirements (Scenario A / B1 / B2).
 --
--- No schema changes yet for 0.2.6. Add DDL below as the 0.2.6 cycle lands
--- extension-schema changes.
+-- Preserve stack/resource errors while classifying plain SQL operands. A broad
+-- WHEN OTHERS handler silently wrapped over-depth Durofut JSON as SQL.
+CREATE OR REPLACE FUNCTION df.ensure_durofut(val text) RETURNS text AS $$
+DECLARE
+	node_type_val text;
+BEGIN
+	BEGIN
+		node_type_val := (val::jsonb)->>'node_type';
+		IF node_type_val IS NOT NULL THEN
+			IF node_type_val NOT IN ('SQL', 'THEN', 'IF', 'JOIN', 'LOOP', 'BREAK', 'RACE', 'SLEEP', 'WAIT_SCHEDULE', 'HTTP', 'HTTP_MULTIPART', 'SIGNAL') THEN
+				RAISE EXCEPTION 'Unknown node_type ''%''. Valid types: SQL, THEN, IF, JOIN, LOOP, BREAK, RACE, SLEEP, WAIT_SCHEDULE, HTTP, HTTP_MULTIPART, SIGNAL', node_type_val;
+			END IF;
+			RETURN val;
+		END IF;
+	EXCEPTION WHEN invalid_text_representation THEN
+		NULL;
+	WHEN raise_exception THEN
+		RAISE;
+	END;
+
+	RETURN df.sql(val);
+END;
+$$ LANGUAGE plpgsql IMMUTABLE SET search_path = pg_catalog, pg_temp;

src/dsl.rs

@@ -235,8 +235,8 @@ pub fn then_fn(a: &str, b: &str) -> String {
 
     Durofut {
         node_type: "THEN".to_string(),
-        left_node: Some(Box::new(a_fut)),
-        right_node: Some(Box::new(b_fut)),
+        left_node: Some(a_fut.into_raw()),
+        right_node: Some(b_fut.into_raw()),
         ..Default::default()
     }
     .to_json()
@@ -337,7 +337,7 @@ pub fn loop_fn(body: &str, condition: default!(Option<&str>, "NULL")) -> String
 
     Durofut {
         node_type: "LOOP".to_string(),
-        left_node: Some(Box::new(body_fut)),
+        left_node: Some(body_fut.into_raw()),
         query,
         ..Default::default()
     }
@@ -390,8 +390,8 @@ pub fn if_fn(condition: &str, then_branch: &str, else_branch: &str) -> String {
 
     Durofut {
         node_type: "IF".to_string(),
-        left_node: Some(Box::new(then_fut)),
-        right_node: Some(Box::new(else_fut)),
+        left_node: Some(then_fut.into_raw()),
+        right_node: Some(else_fut.into_raw()),
         query: Some(config.to_string()),
         ..Default::default()
     }
@@ -413,8 +413,8 @@ pub fn if_rows_fn(result_name: &str, then_branch: &str, else_branch: &str) -> St
 
     Durofut {
         node_type: "IF".to_string(),
-        left_node: Some(Box::new(then_fut)),
-        right_node: Some(Box::new(else_fut)),
+        left_node: Some(then_fut.into_raw()),
+        right_node: Some(else_fut.into_raw()),
         query: Some(config.to_string()),
         ..Default::default()
     }
@@ -430,8 +430,8 @@ pub fn join(a: &str, b: &str) -> String {
 
     Durofut {
         node_type: "JOIN".to_string(),
-        left_node: Some(Box::new(a_fut)),
-        right_node: Some(Box::new(b_fut)),
+        left_node: Some(a_fut.into_raw()),
+        right_node: Some(b_fut.into_raw()),
         ..Default::default()
     }
     .to_json()
@@ -451,8 +451,8 @@ pub fn join3(a: &str, b: &str, c: &str) -> String {
 
     Durofut {
         node_type: "JOIN".to_string(),
-        left_node: Some(Box::new(a_fut)),
-        right_node: Some(Box::new(b_fut)),
+        left_node: Some(a_fut.into_raw()),
+        right_node: Some(b_fut.into_raw()),
         query: Some(config.to_string()),
         ..Default::default()
     }
@@ -468,8 +468,8 @@ pub fn race(a: &str, b: &str) -> String {
 
     Durofut {
         node_type: "RACE".to_string(),
-        left_node: Some(Box::new(a_fut)),
-        right_node: Some(Box::new(b_fut)),
+        left_node: Some(a_fut.into_raw()),
+        right_node: Some(b_fut.into_raw()),
         ..Default::default()
     }
     .to_json()
@@ -1072,9 +1072,11 @@ fn start_in_caller_transaction(fut: &str, label: Option<&str>, database: Option<
             );
         }
         // Recursively insert children FIRST to get their IDs
-        let left_id = node.left_node.as_ref().map(|n| {
+        let left_id = node.left_node.as_ref().map(|raw| {
+            let child = Durofut::child_from_raw(raw)
+                .unwrap_or_else(|e| pgrx::error!("Invalid left child in graph: {}", e));
             insert_nodes(
-                n,
+                &child,
                 instance_id,
                 None,
                 current_user_oid,
@@ -1083,9 +1085,11 @@ fn start_in_caller_transaction(fut: &str, label: Option<&str>, database: Option<
                 node_count,
             )
         });
-        let right_id = node.right_node.as_ref().map(|n| {
+        let right_id = node.right_node.as_ref().map(|raw| {
+            let child = Durofut::child_from_raw(raw)
+                .unwrap_or_else(|e| pgrx::error!("Invalid right child in graph: {}", e));
             insert_nodes(
-                n,
+                &child,
                 instance_id,
                 None,
                 current_user_oid,

src/explain.rs

@@ -229,6 +229,9 @@ fn explain_expression(expr: &str) -> String {
 
     // First try to parse as Durofut JSON
     if let Ok(root) = Durofut::try_from_json(expr) {
+        if let Err(e) = root.validate_recursive() {
+            return format!("Invalid durable function graph: {e}");
+        }
         // Build in-memory node map from nested structure with generated IDs
         let mut nodes = HashMap::new();
         let mut id_counter = 0;
@@ -271,6 +274,9 @@ fn explain_expression(expr: &str) -> String {
         Ok(d) => d,
         Err(e) => return format!("Failed to parse Durofut JSON: {e}"),
     };
+    if let Err(e) = root.validate_recursive() {
+        return format!("Invalid durable function graph: {e}");
+    }
 
     // Build in-memory node map from nested structure with generated IDs
     let mut nodes = HashMap::new();
@@ -293,14 +299,16 @@ fn collect_nodes(
     let node_id = format!("N{}", id_counter);
 
     // Recursively collect children first to get their IDs
-    let left_id = node
-        .left_node
-        .as_ref()
-        .map(|n| collect_nodes(n, nodes, id_counter));
-    let right_id = node
-        .right_node
-        .as_ref()
-        .map(|n| collect_nodes(n, nodes, id_counter));
+    let left_id = node.left_node.as_ref().map(|raw| {
+        let child = crate::types::Durofut::child_from_raw(raw)
+            .expect("validated Durofut child must deserialize");
+        collect_nodes(&child, nodes, id_counter)
+    });
+    let right_id = node.right_node.as_ref().map(|raw| {
+        let child = crate::types::Durofut::child_from_raw(raw)
+            .expect("validated Durofut child must deserialize");
+        collect_nodes(&child, nodes, id_counter)
+    });
 
     // Process config JSON to collect embedded nodes and replace Durofuts with IDs
     let updated_query =