How to Implement Robust Error Handling and Recovery Within ApraPipes Pipelines

ApraPipes provides a layered error handling mechanism that combines static validation to catch configuration errors before execution and runtime policies that determine whether to restart modules, skip frames, or halt the entire pipeline when exceptions occur.

Implementing robust error handling and recovery within ApraPipes pipelines requires understanding both the declarative configuration options available in the framework and the runtime behavior defined in the core execution loop. The apra-labs/aprapipes repository provides three distinct recovery strategies—restart_module, stop_pipeline, and skip—that you can configure via the PipelineSettings structure to match your application's resilience requirements.

Configuring Error Handling Policies in PipelineDescription.h

The foundation of error handling in ApraPipes rests in the PipelineSettings structure defined in base/include/declarative/PipelineDescription.h. This struct exposes an on_error field that accepts string values determining runtime behavior when a module throws an exception.

// base/include/declarative/PipelineDescription.h
// https://github.com/apra-labs/aprapipes/blob/main/base/include/declarative/PipelineDescription.h#L76
struct PipelineSettings {
    std::string name;
    std::string version = "1.0";
    std::string description;
    int queue_size = 10;
    std::string on_error = "restart_module";  // "stop_pipeline" | "skip"
    bool auto_start = false;
};

You can specify the policy programmatically or through declarative configuration files. The JsonParser.cpp file handles both camelCase (onError) and snake_case (on_error) keys for flexibility:

// base/src/declarative/JsonParser.cpp
// https://github.com/apra-labs/aprapipes/blob/main/base/src/declarative/JsonParser.cpp#L124
if (settings.contains("onError") && settings["onError"].is_string()) {
    desc.settings.on_error = settings["onError"].get<std::string>();
}
if (settings.contains("on_error") && settings["on_error"].is_string()) {
    desc.settings.on_error = settings["on_error"].get<std::string>();
}

Understanding the Three Recovery Strategies

ApraPipes implements three distinct behaviors when a module throws a std::runtime_error during frame processing. Each strategy serves different operational requirements, from strict reliability to maximum throughput.

Restart Module

The default restart_module policy resets the internal state of the offending module and continues processing subsequent frames. This approach works well for transient errors such as temporary resource exhaustion or network hiccups that clear upon reinitialization.

When this policy activates, the runtime calls module->reset() before resuming the loop, clearing any corrupted internal buffers or connections.

Stop Pipeline

The stop_pipeline policy implements fail-fast behavior. When any module throws an exception, the entire pipeline terminates immediately with status PL_TERMINATED. This strategy suits critical applications where processing partial results or corrupted frames is unacceptable.

Skip Frame

The skip policy drops the current frame without resetting the module state. The pipeline continues with the next available frame. This approach maximizes throughput for loss-tolerant applications such as real-time video analytics where occasional frame drops are preferable to processing delays.

Static Validation with PipelineValidator

Before runtime execution begins, the PipelineValidator in base/src/declarative/PipelineValidator.cpp analyzes the pipeline description to catch configuration errors. This static analysis prevents runtime failures caused by missing modules, invalid property types, or disconnected graphs.

When validation fails, the system emits Issue::error objects that the CLI surfaces as JSON output:

// base/src/declarative/PipelineValidator.cpp
// https://github.com/apra-labs/aprapipes/blob/main/base/src/declarative/PipelineValidator.cpp#L31
result.issues.push_back(Issue::error(
    Issue::UNKNOWN_MODULE,
    location,
    "Unknown module type '" + module.module_type + "'",
    suggestion));

Running validation before execution ensures that only well-formed pipelines reach the runtime error handling mechanisms:

./aprapipes_cli validate pipeline.json

Runtime Error Propagation in PipeLine.cpp

The core error handling logic resides in base/src/PipeLine.cpp, where the main processing loop wraps module execution in try-catch blocks. When a module throws std::runtime_error or derived exceptions, the catch block consults desc.settings.on_error to determine the recovery action.

The runtime behavior follows this pattern:

// Conceptual extraction from base/src/PipeLine.cpp
try {
    module->process(frame);
} catch (const std::exception& e) {
    const auto& policy = pipelineDesc.settings.on_error;
    if (policy == "restart_module") {
        module->reset();
        LOG_WARN << "Module restarted after error: " << e.what();
    } else if (policy == "stop_pipeline") {
        LOG_ERROR << "Stopping pipeline: " << e.what();
        pipelineStatus = PL_TERMINATED;
        break;
    } else if (policy == "skip") {
        LOG_WARN << "Skipping frame due to error: " << e.what();
        continue;
    }
}

This centralized handling ensures consistent error recovery across all pipeline configurations without requiring individual modules to implement recovery logic.

Practical Implementation Examples

Example 1: Fail-Fast Configuration for Critical Processing

For applications requiring strict reliability, configure the pipeline to terminate on any error:

{
  "settings": {
    "name": "medical-imaging-processor",
    "on_error": "stop_pipeline",
    "queue_size": 20
  },
  "modules": {
    "source": { "type": "dicom_reader", "props": { "path": "/data/study" } },
    "processor": { "type": "image_enhancer" },
    "sink": { "type": "file_writer", "props": { "format": "png" } }
  },
  "connections": [
    { "from": "source.output", "to": "processor.input" },
    { "from": "processor.output", "to": "sink.input" }
  ]
}

Execute with validation:

./aprapipes_cli validate medical_pipeline.json
./aprapipes_cli run medical_pipeline.json

Example 2: Resilient Video Streaming with Module Restart

For video processing where transient decoder errors occur, use the default restart policy:

{
  "settings": {
    "name": "rtsp-stream-processor",
    "on_error": "restart_module",
    "auto_start": true
  },
  "modules": {
    "source": { "type": "rtsp_client", "props": { "url": "rtsp://camera1" } },
    "decoder": { "type": "h264_decoder" },
    "analytics": { "type": "motion_detector" }
  },
  "connections": [
    { "from": "source.output", "to": "decoder.input" },
    { "from": "decoder.output", "to": "analytics.input" }
  ]
}

Example 3: High-Throughput Frame Skipping

For analytics pipelines where occasional frame loss is acceptable:

{
  "settings": {
    "name": "high-throughput-analytics",
    "on_error": "skip",
    "queue_size": 50
  }
}

Extending Error Handling for Custom Recovery

The centralized error handling architecture in PipeLine.cpp allows you to implement custom recovery strategies without modifying individual modules. To add a new policy:

1. Update the schema in PipelineDescription.h to document the new on_error value
2. Extend the parser in JsonParser.cpp to recognize the new policy string
3. Implement the handler in PipeLine.cpp within the main catch block to execute your custom logic

Because modules throw standard std::runtime_error objects, you can also implement fine-grained recovery by catching specific exception types in your custom handler before falling back to the general policy.

Summary

• ApraPipes provides three built-in error recovery policies: restart_module (default), stop_pipeline, and skip, configurable via the on_error field in PipelineSettings.
• Static validation through PipelineValidator.cpp catches configuration errors before runtime, emitting structured Issue::error objects that prevent malformed pipelines from executing.
• Runtime handling in PipeLine.cpp wraps module processing in try-catch blocks, consulting the on_error policy to determine whether to reset modules, halt execution, or drop individual frames.
• Declarative configuration supports both JSON and TOML formats, with JsonParser.cpp accepting both camelCase and snake_case keys for maximum flexibility.

Frequently Asked Questions

What happens if I don't specify an on_error policy in my pipeline configuration?

If you omit the on_error field, ApraPipes defaults to restart_module as defined in base/include/declarative/PipelineDescription.h. This means any module that throws a std::runtime_error will be automatically reset via module->reset() and processing will continue with the next frame, providing resilience against transient failures without requiring explicit configuration.

How does ApraPipes validate pipeline configurations before execution?

The PipelineValidator class in base/src/declarative/PipelineValidator.cpp performs static analysis on the pipeline description before any modules are instantiated. It checks for unknown module types, invalid property values, disconnected graphs, and other structural issues, emitting Issue::error objects for each problem found. The CLI surfaces these as JSON responses with "success": false, allowing you to fix configuration errors before runtime.

Can I implement custom error recovery logic beyond the three built-in policies?

Yes, the centralized error handling in base/src/PipeLine.cpp allows for custom recovery strategies. You can extend the system by adding new string values to the on_error field in PipelineDescription.h, updating JsonParser.cpp to recognize the new value, and implementing the corresponding logic in the catch block of the main processing loop. Because modules use standard std::exception types, you can also catch specific exception subclasses before applying the general policy for fine-grained control.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}