taquba-bulk 0.1.0

Bulk multi-step processing on the Taquba durable task queue: run one pipeline over many inputs in parallel, with per-item memoization and cost rollup.
Documentation

taquba-bulk

Bulk multi-step processing on top of taquba-workflow and the Taquba durable task queue.

Part of the Taquba ecosystem; see the workspace README for the queue core and the other crates that compose with this one.

taquba-bulk runs one Pipeline over many input items in parallel, inside a single process, with per-item memoization, retry, streamed output, and a rolled-up cost report. It is the per-batch orchestrator for workloads that fan out 10-1000x per run: bulk LLM jobs (classify, look up, draft, check, refine over thousands of tickets), document/OCR pipelines, data enrichment, parameter sweeps. The pipeline contract is workload agnostic.

Execution model: one item, one run, one step

Each input item becomes one taquba-workflow run whose single step invokes Pipeline::run. The pipeline's own logical steps live inside that method as BulkCtx::memoized calls. Taquba delivers at-least-once, so a step may run again if its lease expires before it acks; memoization makes that replay cheap, because each completed logical step returns its cached result instead of repeating a paid call. A pipeline error retries with backoff and then dead-letters the item (terminating it failed); the rest of the batch is unaffected.

Single process, remote work per step

The orchestrator is single-process by design: SlateDB allows one writer per store, so all producers and workers for a batch share one Arc<Queue>. That is not a throughput ceiling for bulk work. Each step's expensive operation is a call to a remote service (an LLM API, an OCR service), so the process is I/O-bound and one host sustains hundreds of concurrent items. The remote call runs elsewhere and its response is memoized on return.

Install

cargo add taquba-bulk taquba
cargo add tokio --features full

Quick start

use std::sync::Arc;
use serde::{Deserialize, Serialize};
use taquba::{Queue, object_store::memory::InMemory};
use taquba_bulk::{Bulk, BulkCtx, Pipeline, StepError};

#[derive(Serialize, Deserialize)]
struct Ticket { id: String, body: String }

#[derive(Serialize, Deserialize)]
struct Processed { id: String, classification: String }

struct TicketPipeline;

impl Pipeline for TicketPipeline {
    type Input = Ticket;
    type Output = Processed;
    type Error = StepError;

    async fn run(&self, ctx: &BulkCtx<Ticket>) -> Result<Processed, StepError> {
        let classification = ctx
            .memoized("classify", async {
                ctx.record_cost("llm_calls", 1.0);
                // one paid call, cached on retry
                Ok::<_, StepError>("billing".to_string())
            })
            .await?;
        Ok(Processed { id: ctx.input.id.clone(), classification })
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let store = Arc::new(InMemory::new());
    let queue = Arc::new(Queue::open(store.clone(), "db").await?);

    let bulk = Bulk::builder(queue, store, TicketPipeline)
        .key_fn(|t| t.id.clone())
        .max_concurrent(200)
        .build();

    let inputs = vec![
        Ticket { id: "t1".into(), body: "help".into() },
        Ticket { id: "t2".into(), body: "refund".into() },
    ];
    let report = bulk.run(inputs).await?;
    println!("{}/{} succeeded", report.succeeded, report.total);
    Ok(())
}

See examples/wordcount_bulk.rs for a runnable, network-free end-to-end run.

Cost tracking

Pipelines report arbitrary named metrics via BulkCtx::record_cost (token counts, paid-API units, compute-seconds, dollars). Per-item totals roll up into ProgressSnapshot::cost and BulkReport::cost, so the batch cost is visible live and in the final report.

Failure policy

Per-item failures are recorded, not fatal: each failed item is written to the output sink with its error and its run id is collected on BulkReport::failed_run_ids. Set BulkBuilder::fail_threshold to turn the whole run into an Error::FailureThresholdExceeded when the share of failures crosses a percentage, so a silent mass failure surfaces.

Replay

Because memo entries are retained, re-submitting a failed item's input with the same run id resumes from its last cached step rather than recomputing. BulkReport::failed_run_ids is the set to replay.

Input and output

Line-delimited JSON: read_jsonl decodes inputs and JsonlSink writes one result record per line. Both sides are traits (OutputSink), so other formats can be added without touching the runner. The default sink is NullSink, for pipelines whose results are side effects.

License

Licensed under either of

at your option.

Contribution

Unless you explicitly state otherwise, any contribution intentionally submitted for inclusion in the work by you, as defined in the Apache-2.0 license, shall be dual licensed as above, without any additional terms or conditions.