From 244679fe0477595ffbec3689e53c53ab70eef384 Mon Sep 17 00:00:00 2001 From: Samuel Xing Date: Mon, 3 Aug 2026 19:25:43 -0700 Subject: [PATCH] docs: HTTP-trigger example and exactly-once receiver recipe MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The two integration asks (HTTP frameworks, event-source receivers) both reduce to the same primitive — the workflow id is the idempotency key — so before any adapter crate exists, each gets its minimal honest form: - examples/http_trigger.rs: an axum handler that turns the dbos-idempotency-key header into the workflow id. Retried POSTs attach to the same run; /charges shows the count staying at 1. Requires the admin feature only because that is what pulls axum into this crate's graph — applications depend on axum directly. - messaging guide, new Event-source receivers section: consume from any at-least-once source, derive the id from the message coordinates (topic-partition-offset), start, then ack. The workflow row is durable before the source's ack, so redelivery lands on the existing run — at-least-once delivery + idempotent start = exactly-once execution. Dedicated integration crates (axum extractors, Kafka receivers) remain deliberately post-1.0, once the core API freezes. --- CHANGELOG.md | 13 ++++++ Cargo.toml | 5 +++ examples/http_trigger.rs | 97 ++++++++++++++++++++++++++++++++++++++++ src/lib.rs | 2 +- src/messaging.rs | 32 +++++++++++++ 5 files changed, 148 insertions(+), 1 deletion(-) create mode 100644 examples/http_trigger.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 8014914..a528c67 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,19 @@ project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] +### Documentation + +- HTTP triggering recipe: a runnable axum example + (`examples/http_trigger.rs`) showing the two-line integration — extract + the `dbos-idempotency-key` header, make it the workflow id — so retried + requests attach to the same run instead of repeating effects. The same + shape works in any framework; no adapter crate needed. +- Event-source receiver recipe: the `messaging` guide now documents the + exactly-once consumption pattern (message coordinates as the workflow + id; ack only after the start persisted), which turns any at-least-once + source — Kafka, SQS, `LISTEN` — into exactly-once workflow execution + without a broker integration crate. + ## [0.4.0] - 2026-08-01 The schema catches up to the reference SDKs (migrations 38-40) and the diff --git a/Cargo.toml b/Cargo.toml index 7b3791f..77e37ca 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -116,6 +116,11 @@ futures-util = { version = "0.3", default-features = false, features = ["std", " # the library too — making the `fail_point!` hooks active under test. fail = { version = "0.5", features = ["failpoints"] } +[[example]] +name = "http_trigger" +path = "examples/http_trigger.rs" +required-features = ["admin"] + [[example]] name = "order" path = "examples/order.rs" diff --git a/examples/http_trigger.rs b/examples/http_trigger.rs new file mode 100644 index 0000000..b364aa3 --- /dev/null +++ b/examples/http_trigger.rs @@ -0,0 +1,97 @@ +//! Trigger durable workflows over HTTP: an axum handler starts a workflow and +//! the DBOS idempotency-key header makes retried requests attach to the same +//! run instead of repeating its effects. +//! +//! The whole integration is two lines of glue — there is no framework +//! adapter to configure, because durability lives in the engine, not the +//! transport: +//! * read the `dbos-idempotency-key` header (the cross-SDK name; any +//! caller-supplied stable id works) and make it the workflow id; +//! * `engine.start(...)` — a repeated id attaches to the existing run, so a +//! client or proxy retrying the POST cannot double-charge. +//! +//! The same shape works in any HTTP framework: extract a stable id, pass it +//! as `WorkflowOptions::with_id`. Responses can be immediate (return the +//! workflow id, poll later — shown here) or synchronous (await the handle). +//! +//! ```text +//! cargo run --example http_trigger --features admin +//! # then, in another terminal (same key twice — one workflow): +//! curl -X POST localhost:8080/orders -H 'dbos-idempotency-key: order-1001' -d '1001' +//! curl -X POST localhost:8080/orders -H 'dbos-idempotency-key: order-1001' -d '1001' +//! ``` +//! +//! (Requires the `admin` feature only because that is what pulls axum into +//! this crate's dev graph — your application depends on axum directly.) + +use axum::extract::State; +use axum::http::HeaderMap; +use axum::routing::{get, post}; +use axum::Router; +use durare::{DurableContext, DurableEngine, Error, InMemoryProvider, Result, WorkflowOptions}; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::sync::Arc; + +/// Stand-in for a payment API that must never run twice for one order. +static CHARGES: AtomicU32 = AtomicU32::new(0); + +async fn process_order(ctx: DurableContext, order_id: String) -> Result { + let charge = ctx + .step("charge", || async { + CHARGES.fetch_add(1, Ordering::SeqCst); + Ok::<_, Error>(format!("ch_{order_id}")) + }) + .await?; + Ok(charge) +} + +async fn create_order( + State(engine): State>, + headers: HeaderMap, + body: String, +) -> std::result::Result { + // The caller's idempotency key becomes the workflow id: a retried request + // (same key) attaches to the same run — started exactly once. + let opts = match headers + .get("dbos-idempotency-key") + .and_then(|v| v.to_str().ok()) + { + Some(key) => WorkflowOptions::with_id(key), + None => WorkflowOptions::default(), // no key: every request is a fresh run + }; + let handle = engine + .start::("process_order", body, opts) + .await + .map_err(|e| (axum::http::StatusCode::BAD_REQUEST, e.to_string()))?; + let id = handle.id().to_string(); + // Respond immediately; the run continues durably. (Awaiting `handle` + // instead gives a synchronous response.) + Ok(id) +} + +async fn charges() -> String { + format!("charges: {}\n", CHARGES.load(Ordering::SeqCst)) +} + +#[tokio::main] +async fn main() -> Result<()> { + let mut engine = DurableEngine::new(Arc::new(InMemoryProvider::new())).await?; + engine.register("process_order", process_order); + let engine = Arc::new(engine); + engine.launch().await?; + + let app = Router::new() + .route("/orders", post(create_order)) + .route("/charges", get(charges)) + .with_state(engine); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:8080") + .await + .map_err(|e| Error::app(format!("bind failed: {e}")))?; + println!("POST an order: curl -X POST localhost:8080/orders -H 'dbos-idempotency-key: order-1001' -d '1001'"); + println!("check the count: curl localhost:8080/charges (stays 1 however often you retry)"); + axum::serve(listener, app) + .await + .map_err(|e| Error::app(format!("serve failed: {e}")))?; + Ok(()) +} diff --git a/src/lib.rs b/src/lib.rs index e624ddb..1ad96dd 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -121,7 +121,7 @@ //! [`messaging`], [`transactions`], [`observability`] (spans, probes, and //! metrics), [`operations`] (connections, pool sizing, and the resource //! model), and [`security`] (trust boundaries, exposure, and the SQL and -//! secret-handling invariants). Eleven runnable, end-to-end examples live in +//! secret-handling invariants). Twelve runnable, end-to-end examples live in //! [`examples/`](https://github.com/SamuelXing/durare/tree/main/examples). //! //! # Cargo features diff --git a/src/messaging.rs b/src/messaging.rs index a6f4b7c..937593b 100644 --- a/src/messaging.rs +++ b/src/messaging.rs @@ -89,6 +89,38 @@ //! offset-ordered, and tailable from outside with //! [`DurableEngine::read_stream_values`]. //! +//! # Event-source receivers (Kafka, SQS, and friends) +//! +//! Consuming an external event source and starting a workflow **exactly once +//! per message** needs no adapter crate — the workflow id *is* the dedup +//! mechanism. Make the id from the message's coordinates and start; a +//! redelivered message attaches to the existing run instead of starting a +//! second one: +//! +//! ```ignore +//! // In your consumer loop (any at-least-once source works the same way): +//! loop { +//! let msg = consumer.recv().await?; // Kafka, SQS, LISTEN, ... +//! let id = format!("kafka-{}-{}-{}", msg.topic(), msg.partition(), msg.offset()); +//! engine +//! .start::( +//! "handle_message", +//! decode(&msg)?, +//! WorkflowOptions::with_id(id), // coordinates as identity +//! ) +//! .await?; // duplicate ⇒ same run +//! consumer.commit(&msg).await?; // ack only after the start persisted +//! } +//! ``` +//! +//! The ordering is the whole trick: the workflow row is durable *before* the +//! source's ack, so a crash between the two redelivers the message — and the +//! redelivery lands on the already-persisted run. At-least-once delivery plus +//! an idempotent start composes to exactly-once workflow execution; the +//! handler itself then gets crash-safety from ordinary steps. Use a durable +//! queue ([`WorkflowOptions::queue`](crate::WorkflowOptions)) if consumption +//! should outpace execution. +//! //! [`send`]: DurableContext::send //! [`recv`]: DurableContext::recv //! [`set_event`]: DurableContext::set_event