Bacalhau Documentation

Bacalhau is a distributed compute orchestration framework designed for 'Compute Over Data' (CoD). Instead of moving large datasets across networks to a central compute location, Bacalhau executes jobs close to where the data resides. This approach reduces latency, minimizes ingress/egress costs, and supports data sovereignty by allowing sensitive data to be processed within its original security boundaries.

Core Architecture Concepts

• Orchestrator-Compute Model: A dedicated orchestrator manages job scheduling, while compute nodes execute the actual tasks.
• Single Binary Simplicity: The Bacalhau binary can function as a client, an orchestrator, or a compute node.
• Modular Execution: Supports multiple execution engines like Docker and WebAssembly (WASM).
• Flexible Storage: Integrates with various storage providers including S3, HTTP/HTTPS, IPFS, and local storage.
• Job Types: Supports batch, ops, daemon, and service job types.
• Submission Methods: Jobs can be submitted declaratively using YAML files or imperatively via CLI arguments.
• Publisher Support: Results can be published to local volumes, S3, or other storage backends.

The lifecycle of an execution involves a coordinated flow between the endpoint, scheduler, and compute proxy to synchronize desired and observed states.

State Mutation Responsibilities

To maintain consistency, different components are responsible for specific parts of the execution state:

• Updating Observed State:
  • The Endpoint: Receives user requests and callbacks from compute nodes. It updates the execution's observed state and enqueues an evaluation to notify the scheduler.
  • The Compute Proxy: Receives a plan from the orchestrator and updates the execution's observed state after successfully notifying the compute node of the change.
• Updating Desired State:
  • The Scheduler: Polls evaluations to create new executions or update the desired state of existing ones. It does not modify the observed state directly; instead, it relies on the StateUpdater (a type of Planner) to apply these changes to the job store.

The NCL protocol decouples event processing from message delivery using a local event watcher system. Each node maintains an ordered Event Store (a ledger of all local events with unique monotonic sequence numbers).

The Workflow:

1. Event Store: Maintains the ordered sequence of all local events.
2. Event Watcher: Watches the store for new entries, filters relevant events, and supports resuming from a checkpoint.
3. Message Dispatcher: Creates messages from events, manages reliable delivery, and tracks publish acknowledgments.

Sequence Tracking: To ensure reliability during network partitions or restarts, both sides track progress independently:

• Orchestrator: Tracks which messages each compute node has processed based on heartbeat reports. For new nodes, it starts from the latest sequence; for reconnecting nodes, it uses the last known processed sequence.
• Compute Node: Tracks which messages it has processed from the orchestrator and reports this progress via periodic heartbeats. It maintains local checkpoints for recovery.

The NCL is an internal library used by Bacalhau to facilitate reliable, scalable, and efficient communication between the orchestrator and compute nodes. It uses NATS as the underlying messaging system and supports an event-driven architecture through two primary patterns:

1. Publish-Subscribe: For one-to-many message delivery (e.g., status updates, heartbeats).
2. Request-Response: For direct communication requiring a reply (e.g., job assignments).

The library is composed of four key functional components:

• Publisher: Handles message delivery, including synchronous publishing and request-response.
• Subscriber: Manages message consumption, filtering, and automatic retries/acknowledgments.
• Responder: Specifically handles the request-response pattern by routing requests to registered handlers.
• Encoder: Ensures consistent serialization, decoding, and metadata enrichment across all components.

The following sequence describes the interaction between the Orchestrator and the internal components of a Compute Node during the lifecycle of an execution request:

1. Bidding Phase:

   • Orchestrator sends Request Bid to the Endpoint.
   • Endpoint creates an entry in the Execution Store.
   • Execution Store notifies the Execution Watcher of the new execution.
   • Execution Watcher triggers the Bidder.
   • Bidder updates the Execution Store with the result (Accepted/Rejected).
   • Execution Store notifies the Callback Forwarder, which sends the result back to the Orchestrator.

2. Execution Phase (if Bid Accepted):

   • Orchestrator sends Accept Bid to the Endpoint.
   • Endpoint updates the Execution Store.
   • Execution Store notifies the Execution Watcher of acceptance.
   • Execution Watcher triggers the Executor to Start Execution.
   • Executor updates the Execution Store with Completed or Failed status.
   • Execution Store notifies the Callback Forwarder to send the final status to the Orchestrator.

3. Cancellation Phase:

   • Orchestrator sends Cancel Execution to the Endpoint.
   • Endpoint updates the Execution Store to a Canceled state.
   • Execution Store notifies the Execution Watcher.
   • Execution Watcher triggers the Executor to Cancel Execution.

sequenceDiagram
    actor OrchestratorNode as Orchestrator Node
    box Compute Node
        participant Endpoint
        participant CallbackForwarder as Callback Forwarder
        participant ExecutionStore as Execution Store
        participant ExecutionWatcher as Execution Watcher
        participant Bidder
        participant Executor
    end

    OrchestratorNode ->> Endpoint: Request Bid
    Endpoint ->> ExecutionStore: Create Execution Entry
    ExecutionStore ->> ExecutionWatcher: Notify: Execution Created
    ExecutionWatcher ->> Bidder: Initiate Bidding
    Bidder ->> ExecutionStore: Update Bid Result (Accepted/Rejected)
    ExecutionStore ->> CallbackForwarder: Notify: Bid Result
    CallbackForwarder ->> OrchestratorNode: Send Bid Result Notification

    alt Bid Accepted
        OrchestratorNode ->> Endpoint: Accept Bid
        Endpoint ->> ExecutionStore: Update Execution State
        ExecutionStore ->> ExecutionWatcher: Notify: Execution Accepted
        ExecutionWatcher ->> Executor: Start Execution
        Executor ->> ExecutionStore: Update Status (Completed/Failed)
        ExecutionStore ->> CallbackForwarder: Notify: Completion Status
        CallbackForwarder ->> OrchestratorNode: Send Execution Status Update
    else Bid Rejected
        OrchestratorNode ->> Endpoint: Reject Bid
        Endpoint ->> ExecutionStore: Update Bid Rejected State
    else Cancel
        OrchestratorNode ->> Endpoint: Cancel Execution
        Endpoint ->> ExecutionStore: Update Canceled State
        ExecutionStore ->> ExecutionWatcher: Notify: Execution Canceled
        ExecutionWatcher ->> Executor: Cancel Execution
    end

A Bacalhau compute node processes distributed execution requests through a multi-stage lifecycle involving bidding, execution, and reporting. The flow follows these three primary stages:

1. Bid evaluation and response: The node receives a request, evaluates its own resource availability via the Bidder, and responds to the Orchestrator.
2. Execution initialization and monitoring: Once a bid is accepted, the Executor performs the work while the Execution Watcher and Execution Store monitor the lifecycle.
3. Result reporting and status updates: The node uses a Callback Forwarder to asynchronously send completion, failure, or cancellation updates back to the Orchestrator.

Component Roles

External Component

• Orchestrator Node: The entity that initiates requests, evaluates bids, makes final decisions, and receives status updates.

Internal Components

• Endpoint: The entry point that handles incoming NATS messages, validates requests, and starts workflows.
• Bidder: Evaluates resource constraints and determines if a bid should be accepted or rejected.
• Executor: The component responsible for the actual workload execution, resource allocation, and cleanup.
• Execution Store: The source of truth for execution state, metadata, and history. It uses the /pkg/lib/watcher library to event state changes.
• Execution Watcher: Monitors the Execution Store for state transitions and triggers actions (like starting or canceling an execution).
• Callback Forwarder: Manages reliable, asynchronous communication back to the Orchestrator, including retry logic for notifications.