Flow Execution

Flow Execution

How ION Flow executes workflows - Complete documentation of the flow execution process, from loading to completion.

---

Overview

Flow execution in ION Flow follows a pointer-based model similar to database cursors in PHP 5 with MySQL. This design allows for:

• Stateless execution between calls
• Persistent state in SQLite
• External orchestration of the execution loop
• Pause/resume capabilities

---

Execution Lifecycle

1. Load Flow

Initialize the flow workspace by creating a SQLite database:

func LoadFlow(flowId string, flow models.Flow) error

What happens:

1. Creates a new SQLite database at {DB_PATH}/{flowId}.db
2. Creates tables: nodes, edges, queue, stack, outputs
3. Inserts all nodes and edges from the flow definition
4. Returns success or error

Example:

import "gitlab.com/altacrest/flow_binaries/core/actions/flow"

flowData := models.Flow{
    Nodes: []models.Node{
        {ID: "n_1", Type: "trigger", Data: map[string]any{}},
        {ID: "n_2", Type: "mapper", Data: map[string]any{"params": {...}}},
    },
    Edges: []models.Edge{
        {ID: "e_1", Source: "n_1", Target: "n_2", SourceHandle: "output", TargetHandle: "input"},
    },
}

err := flow.LoadFlow("flow-123", flowData)

---

2. Run Node

Start execution from a specific node:

func RunNode(flowId string, nodeId string, targetId string) (*models.Response, error)

Parameters:

• flowId - The flow workspace identifier
• nodeId - The starting node ID
• targetId - The output handle to use (e.g., “output”, “init”)

What happens:

1. Opens the SQLite database
2. Retrieves the node from the database
3. Evaluates expression references in node data (replaces {{...}})
4. Updates node result cache
5. Creates initial queue entry
6. Updates node status to “processing”
7. Returns response with queue and node info

Response Structure:

type Response struct {
    Params  interface{} // Processed node data
    Queue   *Queue      // Current queue entry
    Node    *Node       // Node being processed
    Message *string     // Optional message
}

---

3. Push Node

Push execution results and continue to connected nodes:

func PushNode(flowId string, ots []models.Ot, message string, status string, clientPath string) (*models.Result, error)

Parameters:

• flowId - The flow workspace identifier
• ots - Output targets (where data flows next)
• message - Status message
• status - Execution status (“success”, “error”)
• clientPath - Client-specific path (for logging)

What happens:

1. Pops the current queue entry
2. Gets the target node
3. Executes node logic based on type
4. Finds connected edges for each output target
5. Creates new queue entries for connected nodes
6. Updates node status
7. Returns result with new queues

Result Structure:

type Result struct {
    Status  string    // "success" or "error"
    Message string    // Status message
    Queue   *Queue    // Original queue entry
    Queues  []Queue   // New queue entries created
    Edges   []string  // Edge IDs traversed
    Node    *Node     // Processed node
    Outputs []Ot      // Output targets
}

---

4. Continue Node

Get the next node from the queue:

func ContinueNode(flowId string) (*models.Response, error)

What happens:

1. Gets the next queue entry (FIFO)
2. Pushes current entry to stack (for reference tracking)
3. Retrieves the target node
4. Evaluates expression references in node data
5. Returns response with next node to process

Returns nil when:

• Queue is empty (flow completed)
• No more nodes to process

---

Complete Execution Loop

package main

import (
    "log"
    "gitlab.com/altacrest/flow_binaries/core/actions/flow"
    "gitlab.com/altacrest/flow_binaries/core/models"
    "gitlab.com/altacrest/flow_binaries/core/models/status"
)

func ExecuteFlow(flowId string, flowData models.Flow, startNodeId string) error {
    // Step 1: Load the flow
    if err := flow.LoadFlow(flowId, flowData); err != nil {
        return err
    }

    // Step 2: Run the initial node
    response, err := flow.RunNode(flowId, startNodeId, "output")
    if err != nil {
        return err
    }

    // Step 3: Execution loop
    for response != nil && response.Queue != nil {
        // Process the current node (your logic here)
        outputs := processNode(response.Node, response.Params)

        // Push results
        result, err := flow.PushNode(
            flowId,
            outputs,
            "Node processed",
            status.Success,
            "",
        )
        if err != nil {
            return err
        }

        log.Printf("Processed node %s, created %d new queue entries",
            response.Node.ID, len(result.Queues))

        // Continue to next node
        response, err = flow.ContinueNode(flowId)
        if err != nil {
            return err
        }
    }

    log.Println("Flow execution completed")
    return nil
}

func processNode(node *models.Node, params interface{}) []models.Ot {
    // Your node processing logic
    return []models.Ot{
        {TargetHandle: helpers.String("output"), Data: params},
    }
}

---

Node Type Execution

Each node type has its own Execute function with the signature:

func Execute(node models.Node, queue models.Queue) ([]models.Ot, string, string)

Returns:

1. []models.Ot - Output targets with data
2. string - Message
3. string - Status (“success”, “error”)

Node Types and Outputs

Node Type	Possible Outputs
Condition	then, false, error
Switch	output_0, output_1, …, default, error
Mapper	output, error
Iterator	output (multiple), error
Timer	output, error
Store	output, error

---

Expression Evaluation

Before processing, node data is evaluated for expressions:

// Original node data
{
    "url": "https://api.example.com/users/{{$1.user_id}}",
    "headers": {
        "Authorization": "Bearer {{$2.token}}"
    }
}

// After evaluation (referencing previous nodes)
{
    "url": "https://api.example.com/users/123",
    "headers": {
        "Authorization": "Bearer abc123xyz"
    }
}

Reference Patterns

Pattern	Description
{{$1}}	Data from node at stack position 1
{{$1.field}}	Specific field from node data
{{n_1.response}}	Named node reference
{{toBase64(value)}}	Built-in function

---

Error Handling

Flow-Level Errors

response, err := flow.RunNode(flowId, nodeId, targetId)
if err != nil {
    // Handle flow-level error
    log.Printf("Flow error: %v", err)
}

Node-Level Errors

Node errors are returned via the error output target:

// Node execution returns error
outputs := []models.Ot{
    {TargetHandle: helpers.String("error"), Data: "Invalid input"},
}
return outputs, "Validation failed", status.Error

---

State Persistence

All execution state is stored in SQLite:

Database Schema

-- Nodes table
CREATE TABLE nodes (
    id TEXT PRIMARY KEY,
    type TEXT,
    data TEXT,
    data_type TEXT,
    status TEXT DEFAULT 'pending',
    updated_at TEXT,
    cache TEXT,
    cache_type TEXT,
    result TEXT,
    result_type TEXT
);

-- Edges table
CREATE TABLE edges (
    id TEXT PRIMARY KEY,
    source TEXT,
    target TEXT,
    source_handler_id TEXT,
    target_handler_id TEXT
);

-- Queue table (FIFO)
CREATE TABLE queue (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    source TEXT,
    target TEXT,
    source_handle TEXT,
    target_handle TEXT,
    data TEXT,
    data_type TEXT,
    edge_id TEXT,
    created_at TEXT,
    status TEXT DEFAULT 'pending'
);

-- Stack table (execution history)
CREATE TABLE stack (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    source TEXT,
    target TEXT,
    source_handle TEXT,
    target_handle TEXT,
    data TEXT,
    data_type TEXT,
    edge_id TEXT,
    created_at TEXT,
    status TEXT
);

---

Related Documentation

• Execution Model - Queue/Stack architecture
• Models - Data structures
• Actions - Node implementations
• Expression Library - Expression evaluation