This document provides a detailed technical walkthrough of how pg_durable builds function graphs (Phase 1: DSL) and executes them durably (Phase 2: Orchestration).
- Overview
- PostgreSQL Extension Architecture
- Phase 1: Function Graph Construction
- Phase 2: Orchestration Execution
- Data Flow Diagram
- Key Files Reference
pg_durable executes durable SQL functions in two distinct phases:
┌──────────────────────────────────────────────────────────────────────────────┐
│ USER SESSION │
│ │
│ Phase 1: Graph Construction (synchronous, in user transaction) │
│ ┌────────────────────────────────────────────────────────────────────────┐ │
│ │ SELECT df.start( │ │
│ │ 'SELECT 1' |=> 'a' ~> 'SELECT $a + 1' │ │
│ │ ); │ │
│ │ │ │
│ │ 1. Operators (~>, |=>) call DSL functions (df.seq, df.as) │ │
│ │ 2. Each function creates a node in df.nodes │ │
│ │ 3. df.start() links nodes, creates instance, enqueues to duroxide │ │
│ │ 4. Returns instance_id immediately (e.g., "a1b2c3d4") │ │
│ └────────────────────────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────────────────────┘
│
│ instance_id enqueued
▼
┌──────────────────────────────────────────────────────────────────────────────┐
│ BACKGROUND WORKER │
│ │
│ Phase 2: Graph Execution (async, durable via duroxide) │
│ ┌────────────────────────────────────────────────────────────────────────┐ │
│ │ 1. Duroxide dispatcher picks up orchestration │ │
│ │ 2. LoadFunctionGraph activity loads nodes from df.nodes │ │
│ │ 3. ExecuteFunctionGraph orchestration walks the graph │ │
│ │ 4. Each SQL node → ExecuteSQL activity (checkpointed) │ │
│ │ 5. Results stored, status updated to 'completed' │ │
│ └────────────────────────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────────────────────┘
Important: pg_durable is a PostgreSQL extension built with pgrx. Everything runs inside the PostgreSQL server process — there are no external services, daemons, or network calls to external orchestrators.
┌─────────────────────────────────────────────────────────────────────────────────────┐
│ POSTGRESQL SERVER │
│ │
│ ┌───────────────────────────────────────────────────────────────────────────────┐ │
│ │ MAIN POSTGRES PROCESS │ │
│ │ │ │
│ │ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ │
│ │ │ Backend Process │ │ Backend Process │ │ Backend Process │ ... │ │
│ │ │ (user session) │ │ (user session) │ │ (user session) │ │ │
│ │ │ │ │ │ │ │ │ │
│ │ │ • Runs SQL │ │ • Runs SQL │ │ • Runs SQL │ │ │
│ │ │ • Calls df.*() │ │ • Calls df.*() │ │ • Calls df.*() │ │ │
│ │ │ • Builds graph │ │ • Builds graph │ │ • Builds graph │ │ │
│ │ │ • Uses SPI │ │ • Uses SPI │ │ • Uses SPI │ │ │
│ │ └────────┬────────┘ └────────┬────────┘ └────────┬────────┘ │ │
│ │ │ │ │ │ │
│ │ │ Enqueue via duroxide.start_orchestration() │ │
│ │ │ │ │ │ │
│ │ ▼ ▼ ▼ │ │
│ │ ┌───────────────────────────────────────────────────────────────────────┐ │ │
│ │ │ SHARED POSTGRESQL TABLES │ │ │
│ │ │ │ │ │
│ │ │ df.instances df.nodes df.vars duroxide.* │ │ │
│ │ │ (instances) (graph) (config) (orchestration state) │ │ │
│ │ └───────────────────────────────────────────────────────────────────────┘ │ │
│ │ ▲ ▲ ▲ │ │
│ │ │ │ │ │ │
│ │ │ Poll & execute via sqlx │ │
│ │ │ │ │ │ │
│ └────────────┼────────────────────┼────────────────────┼────────────────────────┘ │
│ │ │ │ │
│ ┌────────────┴────────────────────┴────────────────────┴────────────────────────┐ │
│ │ BACKGROUND WORKER PROCESS │ │
│ │ (pg_durable_worker) │ │
│ │ │ │
│ │ Registered via BackgroundWorkerBuilder in _PG_init() │ │
│ │ Started automatically when PostgreSQL starts │ │
│ │ │ │
│ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │
│ │ │ DUROXIDE RUNTIME │ │ │
│ │ │ │ │ │
│ │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────────┐ │ │ │
│ │ │ │ Orchestration│ │ Activity │ │ PostgresProvider │ │ │ │
│ │ │ │ Dispatcher │ │ Dispatcher │ │ (duroxide-pg) │ │ │ │
│ │ │ │ │ │ │ │ │ │ │ │
│ │ │ │ Polls for │ │ Polls for │ │ • Connects via sqlx │ │ │ │
│ │ │ │ orchestration│ │ activity │ │ • Stores state in │ │ │ │
│ │ │ │ work items │ │ work items │ │ duroxide.* tables │ │ │ │
│ │ │ └──────────────┘ └──────────────┘ └──────────────────────────┘ │ │ │
│ │ │ │ │ │
│ │ │ ┌─────────────────────────────────────────────────────────────┐ │ │ │
│ │ │ │ REGISTERED COMPONENTS │ │ │ │
│ │ │ │ │ │ │ │
│ │ │ │ Orchestrations: Activities: │ │ │ │
│ │ │ │ • execute-function-graph • load-function-graph │ │ │ │
│ │ │ │ • execute-subtree • execute-sql │ │ │ │
│ │ │ │ • execute-http │ │ │ │
│ │ │ │ • update-instance-status │ │ │ │
│ │ │ │ • update-node-status │ │ │ │
│ │ │ └─────────────────────────────────────────────────────────────┘ │ │ │
│ │ └─────────────────────────────────────────────────────────────────────┘ │ │
│ └───────────────────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────────────┘
-
No External Services: Unlike systems like Temporal or Azure Durable Functions, pg_durable requires no external infrastructure. Everything is self-contained within PostgreSQL.
-
Two Execution Contexts:
- Backend Processes (user sessions): Execute DSL functions synchronously via pgrx's SPI (Server Programming Interface). This is Phase 1 - graph construction.
- Background Worker: A single persistent worker process (registered via
shared_preload_libraries) that runs the duroxide runtime. This is Phase 2 - durable execution.
-
Communication via Tables: The two contexts communicate through PostgreSQL tables:
df.nodes,df.instances,df.vars: Application-level state (function graphs, instances, config)duroxide.*: Orchestration runtime state (work queues, checkpoints, history)
-
Background Worker Registration:
// src/worker.rs - Called during extension load
pub fn register_background_worker() {
BackgroundWorkerBuilder::new("pg_durable_worker")
.set_function("background_worker_main")
.set_library("pg_durable")
.set_restart_time(Some(Duration::from_secs(1)))
.enable_spi_access()
.load();
}
// src/lib.rs - Extension initialization
#[pg_guard]
pub extern "C-unwind" fn _PG_init() {
worker::register_background_worker();
}- Why This Matters:
- Deployment: Just install the extension. No separate services to manage.
- Durability: State is stored in PostgreSQL tables with full ACID guarantees.
- Failover: If PostgreSQL fails over, the new primary picks up where the old one left off.
- Backup: Regular PostgreSQL backups include all orchestration state.
- Security: Uses PostgreSQL's authentication and authorization.
A Durofut represents an abstract function graph, sub-graph or leaf node. It's serialized as JSON and passed between DSL functions.
// src/types.rs
pub struct Durofut {
pub node_type: String, // SQL, THEN, IF, JOIN, LOOP, etc.
pub left_node: Option<Box<Durofut>>, // Embedded left child
pub right_node: Option<Box<Durofut>>, // Embedded right child
pub query: Option<String>, // SQL query or config JSON
pub result_name: Option<String>, // Named result (from |=> operator)
}When serialized to JSON:
{
"node_type": "THEN",
"left_node": {
"node_type": "SQL",
"query": "SELECT 1"
},
"right_node": {
"node_type": "SQL",
"query": "SELECT 2"
}
}df.start(<function>) adds a new row to df.instances, then iterates the nodes in the function graph bottom up, persisting each one to the df.nodes table along with the instance ID, a new ID for the node, and the IDs of its child nodes, if any.
CREATE TABLE df.nodes (
id VARCHAR(8) PRIMARY KEY,
instance_id VARCHAR(8), -- Set by df.start()
node_type TEXT NOT NULL, -- SQL, THEN, IF, JOIN, LOOP, etc.
query TEXT, -- SQL query or config JSON
result_name TEXT, -- Named result for $variable substitution
left_node VARCHAR(8), -- Left child ID
right_node VARCHAR(8), -- Right child ID
status TEXT DEFAULT 'pending',
result JSONB,
created_at TIMESTAMPTZ DEFAULT now()
);Each DSL function (df.sql, df.sleep, df.join, etc.) creates a Durofut and returns its JSON representation. All graph construction is stateless.
// src/dsl.rs
#[pg_extern(schema = "df")]
pub fn sql(query: &str) -> String {
Durofut {
node_type: "SQL".to_string(),
query: Some(query.to_string()),
..Default::default()
}
.to_json()
}#[pg_extern(name = "seq", schema = "df")]
pub fn then_fn(a: &str, b: &str) -> String {
let a_fut = Durofut::ensure(a); // Auto-wrap plain SQL if needed
let b_fut = Durofut::ensure(b);
Durofut {
node_type: "THEN".to_string(),
left_node: Some(Box::new(a_fut)), // Embed first step
right_node: Some(Box::new(b_fut)), // Embed second step
..Default::default()
}
.to_json()
}The Durofut::ensure() function detects whether a string is already a Durofut JSON or plain SQL:
// src/types.rs
impl Durofut {
pub fn ensure(s: &str) -> Self {
if Self::is_durofut(s) {
Self::from_json(s) // Already a Durofut
} else {
// Plain SQL string
Durofut {
node_type: "SQL".to_string(),
query: Some(s.to_string()),
..Default::default()
}
}
}
pub fn is_durofut(s: &str) -> bool {
// Check if valid JSON with a recognized node_type
serde_json::from_str::<Durofut>(s)
.map(|d| VALID_NODE_TYPES.contains(&d.node_type.as_str()))
.unwrap_or(false)
}
}This allows users to write 'SELECT 1' ~> 'SELECT 2' instead of df.sql('SELECT 1') ~> df.sql('SELECT 2').
Operators are syntactic sugar that call DSL functions:
-- src/lib.rs (extension_sql!)
-- Sequence: a ~> b calls df.seq(a, b)
CREATE OPERATOR ~> (
FUNCTION = df.seq,
LEFTARG = text,
RIGHTARG = text
);
-- Name result: a |=> 'name' calls df.as_op(a, name)
CREATE OPERATOR |=> (
FUNCTION = df.as_op,
LEFTARG = text,
RIGHTARG = text
);
-- Parallel join: a & b calls df.join(a, b)
CREATE OPERATOR & (
FUNCTION = df.join,
LEFTARG = text,
RIGHTARG = text
);
-- Conditional: cond ?> then !> else
CREATE OPERATOR ?> (FUNCTION = df.if_then_op, ...);
CREATE OPERATOR !> (FUNCTION = df.if_else_op, ...);
-- Loop prefix: @> body calls df.loop(body)
CREATE OPERATOR @> (FUNCTION = df.loop_prefix_op, RIGHTARG = text);When df.start() is called, it validates the complete graph and recursively inserts all nodes.
Opaque children are parsed one level at a time, and config children are materialized through the
same helper used by df.explain():
fn insert_nodes(node: &Durofut, instance_id: &str) -> Result<String, String> {
let left_id = insert_optional_child(node.left_node.as_deref(), instance_id)?;
let right_id = insert_optional_child(node.right_node.as_deref(), instance_id)?;
// condition_node and extra_nodes are first-class Durofut children. The
// persisted query keeps the worker-facing child-ID representation.
let query = node.transform_config_children(|child| insert_nodes(child, instance_id))?;
insert_node_row(node, query, left_id, right_id, instance_id)
}Variables set via df.setvar() are captured at df.start() time:
// Capture vars from df.vars table
let vars: HashMap<String, String> = Spi::connect(|client| {
let mut vars = HashMap::new();
for row in client.select("SELECT name, value FROM df.vars", None, &[]) {
vars.insert(row.get("name"), row.get("value"));
}
vars
});
// Pass to orchestration
let input = FunctionInput {
instance_id: instance_id.clone(),
label: label.map(|s| s.to_string()),
vars, // Captured snapshot - immutable during execution
};pg_durable uses duroxide for durable execution. Key concepts:
- Orchestrations: Deterministic functions that make scheduling decisions
- Activities: Non-deterministic I/O operations (SQL queries, HTTP calls)
- Replay: On restart, orchestrations replay to reconstruct state
// src/registry.rs - Register orchestrations and activities
pub fn register_orchestrations(registry: &mut OrchestrationRegistry) {
registry.register(
execute_function_graph::NAME,
execute_function_graph::execute,
);
registry.register(
execute_function_graph::SUBTREE_NAME,
execute_function_graph::execute_subtree,
);
}
pub fn register_activities(registry: &mut ActivityRegistry<PgPool>) {
registry.register(load_function_graph::NAME, load_function_graph::execute);
registry.register(execute_sql::NAME, execute_sql::execute);
registry.register(execute_http::NAME, execute_http::execute);
// ...
}The LoadFunctionGraph activity loads the graph from PostgreSQL:
// src/activities/load_function_graph.rs
pub async fn execute(
ctx: ActivityContext,
pool: Arc<PgPool>,
instance_id: String,
) -> Result<String, String> {
// Get root node ID
let root_node_id: String = sqlx::query_scalar(
"SELECT root_node FROM df.instances WHERE id = $1"
).bind(&instance_id).fetch_one(&pool).await?;
// Load all nodes for this instance
let rows = sqlx::query(
"SELECT id, node_type, query, result_name, left_node, right_node
FROM df.nodes WHERE instance_id = $1"
).bind(&instance_id).fetch_all(&pool).await?;
// Build FunctionGraph
let mut nodes = BTreeMap::new(); // BTreeMap for deterministic order
for row in rows {
let node = FunctionNode {
id: row.get("id"),
node_type: row.get("node_type"),
query: row.get("query"),
result_name: row.get("result_name"),
left_node: row.get("left_node"),
right_node: row.get("right_node"),
};
nodes.insert(node.id.clone(), node);
}
let graph = FunctionGraph { instance_id, root_node_id, nodes };
Ok(serde_json::to_string(&graph)?)
}Internal node handlers return NodeResult, a Result whose error arm is a typed
NodeError rather than a plain String. This lets df.break() propagate through the
compound nodes (THEN, IF, JOIN, RACE) automatically via the ? operator, instead
of every handler having to recognise an in-band JSON break sentinel:
// src/orchestrations/execute_function_graph.rs
pub enum NodeError {
/// df.break() fired. Carries the break value. Caught only by execute_loop_node.
Break(String),
/// A genuine failure. Surfaces as a failed instance.
Failure(String),
}
pub type NodeResult = Result<String, NodeError>;
// Any `?` on an existing Result<_, String> auto-converts the error to Failure, so
// activity calls and helpers need no per-call changes.
impl From<String> for NodeError {
fn from(e: String) -> Self {
NodeError::Failure(e)
}
}execute_loop_node is the only handler that catches NodeError::Break (turning it into the
loop's Ok result); NodeError::Failure keeps propagating. The orchestration boundary
functions (execute / execute_subtree) still return Result<String, String> because they
are registered with duroxide:
execute: an uncaught top-levelBreakbecomes a clearErr("df.break() was called outside of a loop"), so the instance fails instead of completing with a sentinel value.execute_subtree(used by JOIN/RACE branches): aBreakis carried out-of-band in the subtree envelope'scontrolfield and re-raised asNodeError::Breakbyparse_subtree_envelopein the parent orchestration.
The orchestration walks the graph recursively:
// src/orchestrations/execute_function_graph.rs
pub async fn execute(ctx: OrchestrationContext, input_json: String) -> Result<String, String> {
let input: FunctionInput = serde_json::from_str(&input_json)?;
// Load graph via activity (checkpointed)
let graph_json = ctx
.schedule_activity(load_function_graph::NAME, input.instance_id.clone())
.into_activity()
.await?;
let graph: FunctionGraph = serde_json::from_str(&graph_json)?;
let mut results: HashMap<String, String> = HashMap::new();
// Execute starting from root node
let exec_ctx = ExecutionContext {
vars: input.vars.clone(),
label: input.label.clone(),
};
let result = execute_function_node_with_vars(
&ctx, &graph, &graph.root_node_id, &mut results, &exec_ctx
).await?;
// Update status to completed
ctx.schedule_activity(update_instance_status::NAME, ...).await;
Ok(result)
}
async fn execute_function_node_with_vars(
ctx: &OrchestrationContext,
graph: &FunctionGraph,
node_id: &str,
results: &mut HashMap<String, String>,
exec_ctx: &ExecutionContext,
) -> NodeResult {
let node = graph.nodes.get(node_id).ok_or("Node not found")?;
ctx.trace_info(format!("Executing node {} (type: {})", node_id, node.node_type));
let result = match node.node_type.as_str() {
"SQL" => execute_sql_node(ctx, node, results, exec_ctx).await?,
"THEN" => execute_then_node(ctx, graph, node, results, exec_ctx).await?,
"IF" => execute_if_node(ctx, graph, node, node_id, results, exec_ctx).await?,
"JOIN" => execute_join_node(ctx, graph, node, node_id, results, exec_ctx).await?,
"RACE" => execute_race_node(ctx, graph, node, node_id, results, exec_ctx).await?,
"LOOP" => execute_loop_node(ctx, graph, node, results, exec_ctx).await?,
"SLEEP" => execute_sleep_node(ctx, node).await?,
"HTTP" => execute_http_node(ctx, node, results, exec_ctx).await?,
"SIGNAL" => execute_signal_node(ctx, node).await?,
"BREAK" => execute_break_node(ctx, node, node_id).await?,
other => return Err(NodeError::Failure(format!("Unknown node type: {other}"))),
};
// Store named results for $variable substitution
if let Some(ref name) = node.result_name {
results.insert(name.clone(), result.clone());
}
Ok(result)
}async fn execute_sql_node(
ctx: &OrchestrationContext,
node: &FunctionNode,
results: &HashMap<String, String>,
exec_ctx: &ExecutionContext,
) -> Result<String, String> {
let query = node.query.as_ref().ok_or("SQL node has no query")?;
// Substitute variables: $name, {var}, {sys_instance_id}
let sys_vars = SystemVars {
instance_id: exec_ctx.instance_id.clone(),
label: exec_ctx.label.clone(),
};
let substituted = substitute_all(query, results, &exec_ctx.vars, &sys_vars);
ctx.trace_info(format!("Executing SQL: {}", substituted));
// Schedule activity (checkpointed by duroxide)
ctx.schedule_activity(execute_sql::NAME, substituted)
.into_activity()
.await
}async fn execute_then_node(
ctx: &OrchestrationContext,
graph: &FunctionGraph,
node: &FunctionNode,
results: &mut HashMap<String, String>,
exec_ctx: &ExecutionContext,
) -> Result<String, String> {
// Execute left (first step)
let left_id = node.left_node.as_ref().ok_or("THEN missing left")?;
let _ = execute_function_node_with_vars(ctx, graph, left_id, results, exec_ctx).await?;
// Execute right (second step)
let right_id = node.right_node.as_ref().ok_or("THEN missing right")?;
execute_function_node_with_vars(ctx, graph, right_id, results, exec_ctx).await
}Three types of variables are substituted:
- Result variables (
$name): From|=>operator, stores previous step results - User variables (
{name}): Fromdf.setvar(), captured at start - System variables (
{sys_instance_id},{sys_label}): Runtime metadata
// src/types.rs
pub fn substitute_all(
query: &str,
results: &HashMap<String, String>,
vars: &HashMap<String, String>,
sys_vars: &SystemVars,
) -> String {
let mut result = query.to_string();
// 1. System vars: {sys_*}
result = result.replace("{sys_instance_id}", &sys_vars.instance_id);
result = result.replace("{sys_label}", sys_vars.label.as_deref().unwrap_or(""));
// 2. User vars: {name}
for (name, value) in vars {
result = result.replace(&format!("{{{}}}", name), value);
}
// 3. Result vars: $name (with smart extraction from SQL results)
for (name, value) in results {
let pattern = format!("${}", name);
if result.contains(&pattern) {
// Extract first column of first row from SQL result JSON
let replacement = extract_value_for_substitution(value);
result = result.replace(&pattern, &replacement);
}
}
result
}For IF, LOOP(body, condition), and conditional operators:
// src/types.rs
pub fn evaluate_condition(result: &str) -> Result<bool, String> {
if let Ok(json) = serde_json::from_str::<serde_json::Value>(result) {
// Extract first column of first row
if let Some(rows) = json.get("rows").and_then(|r| r.as_array()) {
if let Some(first_row) = rows.first() {
if let Some(obj) = first_row.as_object() {
if let Some((_, value)) = obj.iter().next() {
return Ok(is_truthy(value));
}
}
}
}
return Ok(is_truthy(&json));
}
// Fallback for plain strings
let lower = result.to_lowercase();
Ok(matches!(lower.as_str(), "true" | "t" | "yes" | "1"))
}
pub fn is_truthy(value: &serde_json::Value) -> bool {
match value {
Value::Bool(b) => *b,
Value::Number(n) => n.as_i64().map(|i| i != 0).unwrap_or(false),
Value::String(s) => matches!(s.to_lowercase().as_str(), "true" | "t" | "yes" | "1"),
Value::Array(a) => !a.is_empty(),
Value::Object(o) => !o.is_empty(),
Value::Null => false,
}
}JOIN and RACE use duroxide's sub-orchestration support:
async fn execute_join_node(...) -> Result<String, String> {
let left_id = node.left_node.as_ref().ok_or("JOIN missing left")?;
let right_id = node.right_node.as_ref().ok_or("JOIN missing right")?;
// Create sub-orchestration inputs
let left_input = create_subtree_input(graph, left_id, results);
let right_input = create_subtree_input(graph, right_id, results);
// Schedule parallel sub-orchestrations
let left_handle = ctx.schedule_orchestration(SUBTREE_NAME, &left_id, left_input);
let right_handle = ctx.schedule_orchestration(SUBTREE_NAME, &right_id, right_input);
// Wait for all to complete (duroxide handles parallelism)
let (left_result, right_result) = tokio::join!(
left_handle.into_orchestration(),
right_handle.into_orchestration()
);
// Combine results
let results = vec![left_result?, right_result?];
Ok(serde_json::to_string(&results)?)
}For RACE, duroxide's select is used to return the first completed result.
Loops use duroxide's continue_as_new to avoid unbounded history growth. Their execution context is determined by graph position, but only one rule is needed:
- A loop that is the root of the current orchestration's node tree runs inline, and its
continue_as_newstarts that orchestration's next generation. This applies both to a loop at the function graph's root (hosted byexecute_function_graph) and to a loop at the root of a subtree (hosted byexecute_subtree). It is safe because there is no upstream prefix to re-execute: re-entering from the root lands back on the same loop node. - Any deeper loop is spawned as an
execute_subtreechild rooted at the loop node, which then runs it inline per the rule above. Itscontinue_as_newadvances only that child, preserving prefix and suffix work in the waiting parent. A loop used directly as a JOIN or RACE branch needs no special case — every branch is anexecute_subtreechild.
execute_subtree is therefore structurally identical to execute_function_graph: both root an execution context at their own node and host an inline root loop. They differ only in the input envelope they re-enter with on continue_as_new (FunctionInput vs SubtreeInput), in the fact that only the root orchestration touches instance-level status, and in where their graph comes from — execute_function_graph loads it from df.nodes on its first generation, while execute_subtree receives it inline from its parent. The graph is loaded exactly once per instance and then carried inline through every child input and every continue_as_new generation, so submitted_by is fixed for the instance's lifetime and a post-start df.nodes tamper is never read. Role deletion and privilege revocation are still enforced per node execution, by connecting as submitted_by for SQL and by re-checking EXECUTE privilege per HTTP request.
Both paths call run_loop_iteration, which executes the body, catches NodeError::Break, evaluates the optional post-body condition, and propagates NodeError::Failure. A child stamps its LOOP node running on each generation and completed or failed on exit; because continue_as_new returns a future that never resolves, a continuing generation never stamps a terminal status. If a live loop loses a RACE, the parent records the loop node as terminal failed with a cancellation reason because duroxide cancellation stops the child before it can run its own terminal stamp.
Node status stamps contain the full composed orchestration lineage: {root_instance}::{generation}::{child_node}::{generation}.... Read-time inference and the write fence walk that lineage so stale writes and superseded nested branches are evaluated at every ancestor generation.
User Session Background Worker
───────────── ─────────────────
SELECT df.start(
'SELECT 1' |=> 'a'
~> 'SELECT $a + 1'
);
│
├─► df.sql('SELECT 1')
│ └─► INSERT INTO df.nodes (id='abc', type='SQL', query='SELECT 1')
│ └─► Returns: {"node_id":"abc","node_type":"SQL",...}
│
├─► df.as_op(..., 'a')
│ └─► UPDATE df.nodes SET result_name='a' WHERE id='abc'
│ └─► Returns: {"node_id":"abc","result_name":"a",...}
│
├─► df.sql('SELECT $a + 1')
│ └─► INSERT INTO df.nodes (id='def', type='SQL', query='SELECT $a + 1')
│
├─► df.seq(abc, def)
│ └─► INSERT INTO df.nodes (id='ghi', type='THEN', left='abc', right='def')
│
└─► df.start(ghi, NULL)
├─► INSERT INTO df.instances (id='xyz', root_node='ghi')
├─► UPDATE df.nodes SET instance_id='xyz' WHERE id IN ('abc','def','ghi')
├─► Capture vars from df.vars
└─► duroxide.start_orchestration('xyz', input)
│
│ ┌─────────────────────────┐
└────────────────────────────────────►│ Duroxide Dispatcher │
│ │
│ Picks up orchestration │
│ instance 'xyz' │
└───────────┬─────────────┘
│
▼
┌─────────────────────────┐
│ execute_function_graph │
│ │
│ 1. LoadFunctionGraph │
│ (activity) │
│ │
│ 2. Execute THEN node │
│ → Execute SQL 'abc' │
│ (activity) │
│ → Store result 'a' │
│ → Execute SQL 'def' │
│ with $a substituted│
│ │
│ 3. Update status │
└─────────────────────────┘
| File | Purpose |
|---|---|
src/types.rs |
Core types: Durofut, FunctionNode, FunctionGraph, variable substitution |
src/dsl.rs |
DSL functions: df.sql, df.join, df.if, df.loop, etc. |
src/lib.rs |
Schema setup, SQL operators (~>, ` |
src/client.rs |
Duroxide client for df.start(), df.signal(), df.cancel() |
src/worker.rs |
Background worker setup and duroxide runtime initialization |
src/registry.rs |
Orchestration and activity registration |
src/orchestrations/execute_function_graph.rs |
Main orchestration: graph walking, node execution |
src/activities/load_function_graph.rs |
Load graph from df.nodes |
src/activities/execute_sql.rs |
Execute SQL via sqlx |
src/activities/execute_http.rs |
Execute HTTP requests via reqwest |
-
Phase 1 (DSL): User calls DSL functions via SQL. Each function creates a node in
df.nodes. Operators chain nodes into a graph.df.start()links all nodes to an instance and enqueues to duroxide. -
Phase 2 (Execution): Background worker's duroxide runtime picks up the orchestration.
LoadFunctionGraphactivity loads the graph. Orchestration walks the graph, scheduling activities for each step. Results flow between nodes via$variablesubstitution. Loops usecontinue_as_newfor durability.
The key insight is that graph construction is synchronous (in user transaction) while execution is asynchronous and durable (in background worker via duroxide replay).