Sig.Engine (Sig v0.1.0)

Copy Markdown View Source

The execution engine behind Sig.

It resolves which nodes can run, executes them with retries, stores their results and moves executions through their states. Most applications should use the functions in Sig, which delegate here; this module is useful when you need lower-level control, such as running a single node.

Execution states

pending -> running -> completed
                   -> failed
                   -> suspended -> running (resume)
                   -> stopped

Node states

Sig.Schemas.NodeExecution records go through pending, running, retrying and finally completed or failed.

Retries

A failed node is retried until it reaches its max_retries attempts, sleeping backoff_ms * 2^(attempt - 1) milliseconds (capped at 30 seconds) between attempts. When the last attempt fails, the execution is marked as failed.

Idempotency

Before running a node, the engine checks whether it already completed in this execution (in node_results or in its NodeExecution record). If so, the stored result is reused and the node is not run again.

Summary

Functions

execute_node(execution_id, node_name)

Runs a single node of an execution, with retries.

execute_sync(execution_id)

Runs an execution in the calling process until it completes, fails, is suspended or is stopped.

resume_execution(execution_id)

Moves a suspended execution back to running, without dispatching any nodes.

resume_workflow(workflow_id)

See Sig.resume_workflow/1.

stop_execution(execution_id)

See Sig.stop_execution/1.

stop_workflow(workflow_id)

See Sig.stop_workflow/1.

suspend_execution(execution_id)

See Sig.suspend_execution/1.

suspend_workflow(workflow_id)

See Sig.suspend_workflow/1.

Functions

execute_node(execution_id, node_name)

@spec execute_node(String.t(), String.t() | atom()) ::
  {:ok, term()} | {:error, term()}

Runs a single node of an execution, with retries.

Used by Sig.BullMQ.Worker to process one job. It does not dispatch downstream nodes. If the node already completed, its stored result is returned without running it again.

Errors

  • {:error, :execution_not_found}
  • {:error, :workflow_suspended} / {:error, :workflow_stopped}
  • {:error, :execution_suspended} / {:error, :execution_stopped}
  • {:error, {:execution_already_finished, status}}
  • {:error, {:node_not_found, node_name}}
  • {:error, reason} - the node failed on its last attempt.

execute_sync(execution_id)

@spec execute_sync(String.t()) ::
  {:ok, Sig.Schemas.WorkflowExecution.t()} | {:error, term()}

Runs an execution in the calling process until it completes, fails, is suspended or is stopped.

Nodes run one at a time, in dependency order. Returns the execution in its final state.

resume_execution(execution_id)

@spec resume_execution(String.t()) ::
  {:ok, Sig.Schemas.WorkflowExecution.t()} | {:error, term()}

Moves a suspended execution back to running, without dispatching any nodes.

Sig.resume_execution/2 calls this and then starts the execution.

resume_workflow(workflow_id)

@spec resume_workflow(String.t()) ::
  {:ok, Sig.Schemas.Workflow.t()} | {:error, term()}

See Sig.resume_workflow/1.

stop_execution(execution_id)

@spec stop_execution(String.t()) ::
  {:ok, Sig.Schemas.WorkflowExecution.t()} | {:error, term()}

See Sig.stop_execution/1.

stop_workflow(workflow_id)

@spec stop_workflow(String.t()) :: {:ok, Sig.Schemas.Workflow.t()} | {:error, term()}

See Sig.stop_workflow/1.

suspend_execution(execution_id)

@spec suspend_execution(String.t()) ::
  {:ok, Sig.Schemas.WorkflowExecution.t()} | {:error, term()}

See Sig.suspend_execution/1.

suspend_workflow(workflow_id)

@spec suspend_workflow(String.t()) ::
  {:ok, Sig.Schemas.Workflow.t()} | {:error, term()}

See Sig.suspend_workflow/1.