Rust SDK

The reference implementation. Lives at workers/rust/. Every other SDK mirrors its API shape.

Install

The crates are not yet on crates.io — depend on them via path until v0.1 is published.

# Cargo.toml
[dependencies]
conduit-worker = { path = "../conduit/workers/rust/crates/conduit-worker" }
tokio = { version = "1", features = ["full"] }
async-trait = "0.1"   # only if you implement Handler manually

The crate re-exports conduit-worker-macros::handler, so you do not depend on the macro crate directly.

Quick start with #[handler]

The fastest path: an attribute macro on a free async fn.

use std::sync::Arc;
use conduit_worker::{
    handler, Client, ClientConfig, ExternalTask, HandlerError, HandlerResult,
    Runner, RunnerConfig, Variable,
};

#[handler(topic = "http.call")]
async fn http_call(task: &ExternalTask) -> Result<HandlerResult, HandlerError> {
    let url = task
        .variable("url")
        .and_then(|v| v.as_str())
        .ok_or("missing variable: url")?;
    let body = reqwest::get(url).await.map_err(|e| e.to_string())?
        .text().await.map_err(|e| e.to_string())?;

    Ok(HandlerResult::complete(vec![
        Variable::string("response_body", body),
    ]))
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = Client::new(ClientConfig::new("http://localhost:8080"))?;
    let runner = Runner::new(
        client,
        Arc::new(HttpCallHandler),                   // generated by the macro
        RunnerConfig::new("http-worker-1"),
    );
    runner.run().await;
    Ok(())
}

The macro generates:

• A unit struct named after the function in PascalCase plus the suffix Handler. Here, async fn http_call produces pub struct HttpCallHandler;.
• An impl conduit_worker::Handler for HttpCallHandler whose topic() returns the literal you passed and whose handle() delegates to your function.

When to implement Handler directly

The macro is sugar — for stateful handlers (a connection pool, a config struct, a Kafka producer) implement the trait yourself. This is what the reference http-worker binary does.

use async_trait::async_trait;
use conduit_worker::{ExternalTask, Handler, HandlerError, HandlerResult};

pub struct PaymentsHandler {
    api_base: String,
    http: reqwest::Client,
}

#[async_trait]
impl Handler for PaymentsHandler {
    fn topic(&self) -> &str { "payments.charge" }

    async fn handle(&self, task: &ExternalTask) -> Result<HandlerResult, HandlerError> {
        let resp = self
            .http
            .post(format!("{}/charges", self.api_base))
            .json(&serde_json::json!({ "task_id": task.id }))
            .send()
            .await
            .map_err(|e| HandlerError::new(e.to_string()))?;

        if resp.status() == 402 {
            return Ok(HandlerResult::bpmn_error("PAYMENT_DECLINED", "card declined"));
        }
        Ok(HandlerResult::ok())
    }
}

RunnerConfig

let config = RunnerConfig::new("http-worker-1");   // worker_id (lock owner)
// fields, all overridable:
//   max_jobs_per_fetch:  10                 (1..=100)
//   lock_duration_secs:  60
//   idle_backoff:        Duration::from_millis(500)
//   error_backoff:       Duration::from_secs(5)

Spawn one Runner per topic. For multi-topic fleets, clone the Client and spawn each runner on its own Tokio task:

let client = Client::new(ClientConfig::new("http://localhost:8080"))?;
let r1 = Runner::new(client.clone(), Arc::new(HttpCallHandler), RunnerConfig::new("w-1"));
let r2 = Runner::new(client.clone(), Arc::new(PolicyCheckHandler), RunnerConfig::new("w-1"));
tokio::join!(r1.run(), r2.run());

Reporting outcomes

Every handler returns one of:

Convenience constructors:

HandlerResult::ok()                              // complete, no variables
HandlerResult::complete(vec![Variable::long("status", 201)])
HandlerResult::bpmn_error("DENIED", "policy violation")
HandlerError::new("timeout connecting to upstream")

Variables

Variable::string("name", "alice")
Variable::long("count", 42)
Variable::double("ratio", 0.75)
Variable::boolean("approved", true)
Variable::json("payload", serde_json::json!({ "x": 1 }))
Variable::null("tombstone")

Read variables off the incoming task:

let order_id: &str = task.variable("order_id").and_then(|v| v.as_str()).unwrap_or("");
let all: HashMap<String, serde_json::Value> = task.variable_map();

Building and running

cd workers/rust
cargo test --workspace
cargo run -p http-worker -- --config worker.yaml   # reference http.call binary

See the http-worker source for a full multi-topic worker driven by a YAML config (URL templates, header secrets, response transforms).

Idempotency

The Rust SDK does not protect side effects automatically. Make handlers idempotent under retry — strategies are documented in Idempotency.