Skip to content

Task Queue & Step Runner

Jump to bottom
Chazona Baum edited this page Jun 24, 2026 · 1 revision

Relevant source files

The Task Queue and Step Runner form the execution engine of Lodestar's job-discovery pipeline. While the pipeline defines the high-level logic (e.g., "scrape careers page"), the Task Queue provides durable persistence to ensure runs survive app restarts, and the Step Runner handles the telemetry projection of individual execution stages into the user's vault.

Durable SQLite Queue (SqliteQueue)

Lodestar uses a local SQLite database to manage a durable task queue. This implementation ensures that long-running pipeline operations are not lost if the application crashes or is closed mid-run.

Task States & Lifecycle

Tasks within the SqliteQueue transition through several states defined in src-tauri/src/pipeline/queue.rs:

State Description
pending The initial state for a NewTask. Ready to be picked up by the runner src-tauri/src/pipeline/queue.rs#81
claimed A task currently being processed. Atomically transitioned from pending via claim_nextsrc-tauri/src/pipeline/queue.rs#132
done Successfully completed tasks src-tauri/src/pipeline/queue.rs#142
dead Tasks that have exceeded MAX_ATTEMPTS or encountered a terminal failure src-tauri/src/pipeline/queue.rs#151

Key Queue Operations

  • Atomicity (claim_next): To prevent race conditions, claim_next queries for the oldest pending task and immediately updates its state to claimed while incrementing the attempts counter within a single database lock src-tauri/src/pipeline/queue.rs#106-138
  • Exponential Backoff: Failed tasks are rescheduled with a delay calculated by backoff_delay(attempt), which caps at 5 minutes (300 seconds) src-tauri/src/pipeline/queue.rs#61-63
  • Crash Recovery: Upon opening the database, the queue automatically resets any tasks stuck in the claimed state back to pending. This ensures that if the worker dies mid-step, the run resumes on next launch src-tauri/src/pipeline/queue.rs#84
  • Run Abort (discard_run_tasks): Outstanding tasks (pending or claimed) for a specific run_id can be deleted to halt a pipeline run without losing the history of completed (done) or failed (dead) steps src-tauri/src/pipeline/queue.rs#178-187

Data Flow: Task Queue Entities

The following diagram maps the Natural Language concepts of the queue to the specific Rust structs and SQLite schema.

Task Queue Entity Mapping

flowchart TD
    subgraph subGraph1 ["Code Entity Space (src-tauri/src/pipeline/queue.rs)"]
        NewTask["struct NewTask { run_id, stage, payload }"]
        QueuedTask["struct QueuedTask { id, attempts, ... }"]
        SqliteQueue["struct SqliteQueue { conn: Mutex }"]
        SQL["CREATE TABLE tasks (state, last_error, ...)"]
    end
    subgraph subGraph0 ["Natural Language Space"]
        A["A New Work Item"]
        B["An Active Task"]
        C["Durable Storage"]
    end
    A --> NewTask
    B --> QueuedTask
    C --> SqliteQueue
    SqliteQueue --> SQL
    NewTask --> SQL
    SQL --> QueuedTask
Loading

Sources: src-tauri/src/pipeline/queue.rs#22-40src-tauri/src/pipeline/queue.rs#65-87

Step Runner (runner.rs)

The Step Runner executes individual pipeline stages and projects telemetry back into the Obsidian vault. It acts as the bridge between the execution clients (Scrapers, LLMs) and the durable Markdown records in the checks/ directory.

Telemetry Projection

Every execution stage results in a Step record. These records are appended to the Markdown frontmatter of a "Check" note in the vault via record_stepsrc-tauri/src/pipeline/runner.rs#113-135

The runner uses a Builder Pattern for outcomes:

Scrape Step Execution

The run_scrape_step function is a primary entry point for the runner. It orchestrates the following:

  1. Timestamping: Captures started_at using now_iso()src-tauri/src/pipeline/runner.rs#151
  2. Execution: Calls the Scraper::fetch method src-tauri/src/pipeline/runner.rs#152
  3. Recording: Automatically converts ScrapeResult credits into a Step record in the vault src-tauri/src/pipeline/runner.rs#155-159
  4. Error Handling: If the scrape fails, it records the failure before propagating the ScrapeErrorsrc-tauri/src/pipeline/runner.rs#168-172

Markdown Check Integration

The src-tauri/src/check.rs module handles the actual serialization of these steps into Markdown.

Step to Markdown Projection

flowchart LR
    subgraph subGraph2 ["Vault Structure"]
        StepsList["steps: #91; { stage, status, cost } #93;"]
    end
    subgraph subGraph1 ["Persistence (check.rs)"]
        RS["record_step()"]
        AS["append_step()"]
        CheckNote["Check Markdown Note"]
    end
    subgraph subGraph0 ["Execution (runner.rs)"]
        SO["StepOutcome"]
        SI["StepIdentity"]
    end
    SO --> RS
    SI --> RS
    RS --> AS
    AS --> CheckNote
    CheckNote --> StepsList
Loading

Sources: src-tauri/src/pipeline/runner.rs#113-135src-tauri/src/check.rs#151-158src-tauri/src/check.rs#98-135

Step Schema Reference

The Step struct is the unit of telemetry stored in the vault src-tauri/src/check.rs#12-41:

Field Description
stage The name of the pipeline stage (e.g., jd-scrape).
class The type of work (e.g., scrape, llm, script).
status ok, failed, or warning.
cost ScrapingBee credits or OpenRouter micro-USD.
cache_read_tokens Telemetry showing prompt cache hits (OpenRouter).
warnings A list of non-fatal issues encountered during the step.

Sources: src-tauri/src/pipeline/runner.rs#1-172src-tauri/src/pipeline/queue.rs#1-188src-tauri/src/check.rs#1-158

Clone this wiki locally