Workflows¶

def validate(
    wf_graph: graph.Graph, settings: graph.Settings
) -> graph.ValidationResult:
    """Return the issues of a workflow version."""
    issues = _Issues()
    if len(wf_graph.nodes) > MAX_NODES:
        issues.add('error', 'too_many_nodes', f'More than {MAX_NODES} steps')
    if len(wf_graph.edges) > MAX_EDGES:
        issues.add(
            'error', 'too_many_edges', f'More than {MAX_EDGES} connections'
        )
    if len(wf_graph.model_dump_json().encode()) > MAX_GRAPH_BYTES:
        issues.add(
            'error',
            'graph_too_large',
            f'The graph is larger than {MAX_GRAPH_BYTES} bytes',
        )
    if not issues:
        _Validator(wf_graph, issues).run()
        _check_concurrency(wf_graph, settings, issues)
    return graph.ValidationResult(issues=issues)

def order(wf_graph: graph.Graph) -> list[str]:
    """Return the node ids in topological order.

    Ties keep the order of ``wf_graph.nodes``. Nodes on a cycle are not
    in the result; activation refuses a graph with a cycle.
    """
    pending = {n.id: 0 for n in wf_graph.nodes}
    for edge in wf_graph.edges:
        if edge.target in pending:
            pending[edge.target] += 1
    result = [n for n, count in pending.items() if count == 0]
    for node_id in result:
        for edge in wf_graph.edges:
            if edge.source == node_id and edge.target in pending:
                pending[edge.target] -= 1
                if pending[edge.target] == 0:
                    result.append(edge.target)
    return result

def advance(
    wf_graph: graph.Graph,
    status: dict[str, str],
    ports: dict[str, int | None],
) -> tuple[list[str], list[str]]:
    """Return the ready steps and the steps that become ``skipped``.

    ``status`` is changed in place for the skipped steps. A pending node
    with no incoming edge is a trigger that did not start the run; the
    caller marks those ``skipped`` when the run starts.
    """
    nodes = order(wf_graph)
    parents: dict[str, list[graph.Edge]] = {n: [] for n in nodes}
    for edge in wf_graph.edges:
        if edge.target in parents:
            parents[edge.target].append(edge)
    joins = {n.id: n.join for n in wf_graph.nodes}
    skipped: list[str] = []
    ready: list[str] = []
    # Topological order: a skip reaches every step after it in one pass.
    for node_id in nodes:
        if status.get(node_id) != 'pending' or not parents[node_id]:
            continue
        states = [_edge_state(e, status, ports) for e in parents[node_id]]
        if all(s == 'inactive' for s in states):
            status[node_id] = 'skipped'
            skipped.append(node_id)
        elif 'active' in states and (
            joins[node_id] == 'any' or 'waiting' not in states
        ):
            ready.append(node_id)
    return ready, skipped

async def append_event(
    conn: psycopg.AsyncConnection[typing.Any],
    run_id: uuid.UUID,
    event_type: str,
    payload: dict[str, typing.Any],
) -> None:
    """Write the next event of a run. Call it in the run's transaction.

    The ``UPDATE`` of ``last_seq`` locks the run row, so the sequence
    has no gaps.
    """
    cur = await conn.execute(
        'UPDATE workflow_runtime.runs SET last_seq = last_seq + 1'
        ' WHERE id = %s RETURNING last_seq',
        (run_id,),
    )
    row = await cur.fetchone()
    if row is None:
        raise LookupError(f'No run {run_id}')
    await conn.execute(
        'INSERT INTO workflow_runtime.run_events'
        ' (run_id, seq, event_id, type, payload)'
        ' VALUES (%s, %s, %s, %s, %s)',
        (run_id, row[0], uuid.uuid4(), event_type, pg_json.Jsonb(payload)),
    )

async def pause_bindings(
    conn: psycopg.AsyncConnection[typing.Any], workflow_id: uuid.UUID
) -> None:
    """Disable the trigger bindings of a workflow that pauses.

    Call it in the transaction that pauses the workflow.
    """
    await conn.execute(
        'UPDATE workflows.trigger_bindings SET enabled = FALSE,'
        " sync_state = 'pending', updated_at = NOW()"
        ' WHERE workflow_id = %s',
        (workflow_id,),
    )
    await conn.execute(f'NOTIFY {BINDINGS_CHANNEL}')