feat: add nats jetstream support

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "openai-api-dispatch"
-version = "0.1.1"
+version = "0.2.0"
 authors = ["Victor Lopez <vhrlopes@gmail.com>"]
 description = "OpenAI-compatible chat requests through memory or NATS queues."
 edition = "2024"
@@ -38,9 +38,20 @@ uuid = { version = "1.25.0", default-features = false, features = ["serde"] }
 tokio = { version = "1.53.1", features = ["full"] }
 
 [features]
-default = ["memory-queue", "nats-queue", "std"]
+default = ["jetstream-queue", "memory-queue", "nats-queue", "std"]
 memory-queue = ["spin"]
 nats-queue = ["async-nats", "std", "tokio-stream"]
+jetstream-queue = [
+  "async-nats",
+  "std",
+  "tokio-stream",
+  "async-nats/jetstream",
+  "async-nats/kv",
+  "async-nats/server_2_10",
+  "tokio/macros",
+  "tokio/rt",
+  "tokio/time",
+]
 std = [
   "anyhow/default",
   "async-openai",

README.md

@@ -4,11 +4,11 @@
 [![Documentation](https://docs.rs/openai-api-dispatch/badge.svg)](https://docs.rs/openai-api-dispatch/)
 [![License](https://img.shields.io/crates/l/openai-api-dispatch.svg)](#license)
 
-OpenAI-compatible chat requests through memory or NATS queues.
+OpenAI-compatible chat requests through memory, Core NATS, or durable JetStream queues.
 
 Producers submit typed tasks, workers call the configured API, and responses return through the queue.
 
-## NATS quick start
+## Core NATS quick start
 
 Run a NATS server, then configure the model and OpenAI-compatible endpoint:
 
@@ -41,7 +41,6 @@ async fn main() -> anyhow::Result<()> {
         .build_chat()?;
 
     let response = task.send_and_wait(&producer, Some(30)).await?;
-    anyhow::ensure!(response.success, "{}", response.contents);
     println!("{}", response.contents);
 
     Ok(())
@@ -52,24 +51,54 @@ async fn main() -> anyhow::Result<()> {
 
 Workers subscribe to `openai-api-queue/<model>` by default. Override the prefix with `OPENAI_API_NATS_PREFIX`. In production, run workers and producers as separate processes.
 
+## Durable JetStream queues
+
+The `jetstream-queue` feature is enabled by default. Use
+`queue::jetstream::JetStreamProducer` to submit tasks, and combine
+`queue::jetstream::JetStreamWorker` with an executor through `worker::Worker::new`.
+The `queue::jetstream` module documents all connection, provisioning, and worker
+environment settings.
+
+JetStream stores tasks in a work-queue stream and keeps status and responses in
+a KV bucket. Workers save a response before acknowledging a successful task or
+terminating redelivery of a failed task. Defaults are three total deliveries per
+message, a 30-second acknowledgment wait, and a one-hour maximum age for task
+messages and KV entries. KV updates start a new entry age. Use fresh task IDs for
+new work: an existing KV record suppresses publication of that ID.
+
+Unacknowledged tasks can be redelivered while an active worker is available and
+delivery and retention limits permit it. The worker does not renew acknowledgment
+deadlines or impose an execution deadline. API calls can repeat if execution
+outlasts the acknowledgment wait or a worker crashes before acknowledgment.
+Reaching the server's delivery limit does not itself produce an error response,
+so callers should bound their response wait.
+
+Producer and worker constructors open existing resources or create missing ones.
+Existing resource configurations are not updated. Set
+`OPENAI_API_NATS_STREAM_FORBID_CREATE` and `OPENAI_API_NATS_STORE_FORBID_CREATE` to
+require an existing stream and KV bucket; a worker can still create its durable
+consumer. Both flags are enabled by any value, including `0` or `false`.
+
 ## Configuration and behavior
 
 Environment settings are read when constructing producers, workers, and executors.
+The NATS connection and routing notes below describe Core NATS; JetStream uses
+the resource and delivery settings documented in `queue::jetstream`.
 
 | Variable | Default |
 | --- | --- |
 | `OPENAI_API_NATS_URL` | `nats://localhost:4222` |
 | `OPENAI_API_NATS_WORKERS_GROUP` | `task_workers` |
 | `OPENAI_API_NATS_PREFIX` | `openai-api-queue/` |
 | `OPENAI_API_URL` | `http://127.0.0.1:8000/v1` |
-| `OPENAI_API_DEFAULT_MODEL` | Unset; required by NATS workers |
+| `OPENAI_API_DEFAULT_MODEL` | Unset; required by Core NATS workers |
 
-- A task's explicit model overrides the default and selects its NATS subject.
-- Constructors do not wait for server confirmation of subscriptions, so startup can race with publishing.
+- A task's explicit model overrides the executor default and selects its Core NATS subject. JetStream uses one configured subject for all task models.
+- Core NATS constructors do not wait for server confirmation of subscriptions, so startup can race with publishing.
 - Each worker handles one task at a time.
 - Queue/API errors stop its loop without an error reply; supervise spawned workers.
-- `send_and_wait` limits only the reply wait, not submission, and expiry does not cancel execution.
-- Check `response.success` even when the call returns `Ok`.
+- `send_and_wait` returns a `ValidatedResponse` after checking the response's success flag; an unsuccessful response returns `Err`. Its optional timeout limits only the reply wait, not submission, and expiry does not cancel execution.
+- Direct calls to `Executor::execute` or `QueueProducer::receive_response` return raw responses; check their `success` flag even when the call returns `Ok`.
 - Only non-streaming chat is implemented.
 - The current prompt is sent as a user message; system entries in input history are ignored (use `with_system`).
 - Schemas request strict JSON output without local validation.
@@ -78,17 +107,22 @@ Environment settings are read when constructing producers, workers, and executor
 
 ## Features
 
-Default features are `std`, `memory-queue`, and `nats-queue`; NATS implies `std`. The API executor requires `std`.
+Default features are `std`, `memory-queue`, `nats-queue`, and `jetstream-queue`.
+Both NATS backends imply `std` and Tokio support. The API executor requires `std`.
 
 #### no_std
 
-For memory queues, use the feature `memory-queue`. An allocator and pointer/32-bit atomics are required.
+Disable default features and enable `memory-queue` to use the polling memory
+backend without `std`. An allocator and pointer/32-bit atomics are required.
 
-This backend polls once, has no timeout support, and evicts old items when bounded; the `std` backend uses bounded Tokio channels and backpressure. 
+This backend polls once, has no timeout support, and evicts old items when bounded; the `std` backend uses bounded Tokio channels and backpressure.
 
 ## Development
 
 Run `cargo test` for local tests. `just check` also requires `cargo-hack` and the `thumbv7em-none-eabi` target. Live integration tests are ignored by default: run `just check-nats MODEL [URL]` or `just check-openai MODEL [URL]` against configured servers.
+Run `just check-jetstream [MODEL] [URL]` against a JetStream-enabled NATS server.
+Each invocation uses a separate stream and KV bucket with memory storage, and
+each task uses a fresh ID so retained results do not suppress test submissions.
 
 ## License
 

justfile

@@ -7,18 +7,27 @@ check-format:
   cargo fmt --all -- --check
 
 check-clippy:
-  cargo clippy --workspace --all-targets -- -D warnings
+  cargo clippy --workspace --all-targets --all-features -- -D warnings
 
 check-test:
   cargo hack test --feature-powerset
 
 check-doc:
-  RUSTDOCFLAGS="-D warnings" cargo doc --no-deps
+  RUSTDOCFLAGS="-D warnings" cargo doc --all-features --no-deps
 
 check-nats model="test" url="nats://127.0.0.1:4222":
   OPENAI_API_DEFAULT_MODEL="{{ model }}" OPENAI_API_NATS_URL="{{ url }}" \
     cargo test --features nats-queue nats -- --ignored
 
+check-jetstream model="test" url="nats://127.0.0.1:4222":
+  test_namespace="openai_api_dispatch_tests_{{ uuid() }}"; \
+    OPENAI_API_DEFAULT_MODEL="{{ model }}" \
+    OPENAI_API_NATS_URL="{{ url }}" \
+    OPENAI_API_NATS_STREAM="$test_namespace" \
+    OPENAI_API_NATS_BUCKET="$test_namespace" \
+    OPENAI_API_NATS_MEMORY_STORAGE=1 \
+    cargo test --features jetstream-queue jetstream -- --ignored
+
 check-openai model="test" url="http://127.0.0.1:8000/v1":
   OPENAI_API_DEFAULT_MODEL="{{ model }}" OPENAI_API_URL="{{ url }}" \
     cargo test openai -- --ignored
@@ -31,5 +40,5 @@ check-features:
   RUSTFLAGS="-D warnings" cargo hack check \
     --target thumbv7em-none-eabi \
     --feature-powerset \
-    --exclude-features std,default,nats-queue \
+    --exclude-features std,default,nats-queue,jetstream-queue \
     --no-dev-deps

src/executor/openai/mod.rs

@@ -30,14 +30,14 @@ mod tests;
 /// an unsuccessful [`Response`]. Reported usage is total tokens, or zero if absent.
 ///
 /// Sends the optional system instruction, prior user/assistant turns, and the
-/// current prompt as a user message, in that order.
+/// current prompt as a user message, in that order. Successful responses contain
+/// those user/assistant turns with the new assistant reply appended; the current
+/// prompt appears once and system turns are omitted from returned history.
 ///
 /// # Current limitations
 ///
 /// - System entries in input history are discarded; use [`TaskChat::system`]
 ///   for a persistent system instruction.
-/// - Returned history currently duplicates the latest user prompt: it is kept
-///   from the request messages and appended again before the assistant reply.
 /// - Schemas request strict JSON output named `response`; returned text is not
 ///   locally parsed or validated against the schema.
 /// - `max_tokens` is cast from `u64` to `u32` without range checking. Payload

src/lib.rs

@@ -1,5 +1,5 @@
 #![cfg_attr(not(feature = "std"), no_std)]
-#![warn(missing_docs)]
+#![deny(missing_docs)]
 #![doc = include_str!("../README.md")]
 
 extern crate alloc;

src/queue/jetstream/mod.rs

@@ -0,0 +1,80 @@
+//! Durable task delivery through NATS JetStream, with task status and responses
+//! retained in a key-value (KV) bucket.
+//!
+//! [crate::queue::jetstream::JetStreamProducer] publishes JSON tasks to a work-queue stream and watches
+//! their KV records for responses. [crate::queue::jetstream::JetStreamWorker] shares a durable pull
+//! consumer with other workers in its group, waits for tasks, and records results
+//! before acknowledging deliveries. This module requires the `jetstream-queue`
+//! feature, a Tokio runtime, and a server with JetStream enabled.
+//!
+//! # Environment configuration
+//!
+//! [crate::queue::jetstream::JetStreamProducer::from_env_or_default] and
+//! [crate::queue::jetstream::JetStreamWorker::from_env_or_default] read the following variables during
+//! construction. Existing instances do not track subsequent environment changes.
+//! Defaults apply when a variable is unset or contains non-Unicode data. Invalid
+//! numeric text returns an error when the corresponding setting is parsed;
+//! resource names and numeric limits are also subject to NATS validation.
+//!
+//! ## Connection and resource names
+//!
+//! | Variable | Default | Behavior |
+//! | --- | --- | --- |
+//! | `OPENAI_API_NATS_URL` | `nats://localhost:4222` | NATS connection URL used by both producers and workers. |
+//! | `OPENAI_API_NATS_STREAM` | `OPENAI_API_DISPATCH` | Task stream name, also used as the prefix of the task subject. |
+//! | `OPENAI_API_NATS_SUBJECT` | `dispatch.task` | Subject suffix. The full subject is `<stream>.<subject>`, with a literal dot inserted between the values. |
+//! | `OPENAI_API_NATS_BUCKET` | `openai_api_dispatch_task` | KV bucket containing task records keyed by the decimal task ID. |
+//!
+//! The default task subject is `OPENAI_API_DISPATCH.dispatch.task`. Producers
+//! publish to this subject and workers filter on it; task models do not select
+//! subjects in this backend. Producers and workers must agree on the stream,
+//! subject, and bucket. The bucket name is independent of the stream name, so use
+//! distinct names for both when isolating queues. An existing task record causes
+//! the producer to reuse that record instead of publishing the task again.
+//!
+//! ## Stream and KV provisioning
+//!
+//! Both constructors reuse existing resources and create missing ones by default.
+//! Creation settings below do not update existing streams or buckets.
+//!
+//! | Variable | Default | Behavior |
+//! | --- | --- | --- |
+//! | `OPENAI_API_NATS_MEMORY_STORAGE` | Unset (file storage) | Presence selects memory storage for newly created task streams and KV buckets. |
+//! | `OPENAI_API_NATS_TTL_SECS` | `3600` | Maximum age in seconds (`u64`) of task stream messages and KV entries. Each KV update has its own age; this is not a response-wait timeout. |
+//! | `OPENAI_API_NATS_DUPLICATE_WINDOW_SECS` | `600` | Stream deduplication window in seconds (`u64`) for message IDs, which the producer sets to task IDs. KV record deduplication is separate. |
+//! | `OPENAI_API_NATS_REPLICAS` | `1` | Replica count (`usize`) for the task stream. KV bucket replication uses the client's default configuration. |
+//! | `OPENAI_API_NATS_MAX_BYTES` | `0` | Total KV bucket size limit in bytes (`i64`), passed to NATS through the KV configuration. Applies only when creating a bucket. |
+//! | `OPENAI_API_NATS_STREAM_FORBID_CREATE` | Unset | Presence requires the task stream to exist; fetching it must succeed. Stream creation settings are skipped. |
+//! | `OPENAI_API_NATS_STORE_FORBID_CREATE` | Unset | Presence requires the KV bucket to be accessible; a failed lookup returns an error instead of attempting creation. |
+//!
+//! The three presence flags (`MEMORY_STORAGE`, `STREAM_FORBID_CREATE`, and
+//! `STORE_FORBID_CREATE`, each prefixed with `OPENAI_API_NATS_`) accept any Unicode
+//! value, including an empty string, `0`, or `false`, as enabled. Unset a flag to
+//! disable it. The creation restrictions apply to the stream and bucket only;
+//! workers still create their durable consumer if it is missing.
+//!
+//! ## Worker settings
+//!
+//! These variables are read by [crate::queue::jetstream::JetStreamWorker::from_env_or_default]. Existing
+//! durable consumers are reused without updating their server configuration.
+//!
+//! | Variable | Default | Behavior |
+//! | --- | --- | --- |
+//! | `OPENAI_API_NATS_CONSUMER_GROUP` | `openai_api_dispatch_workers` | Durable pull consumer name. Workers using the same stream and consumer share task deliveries. |
+//! | `OPENAI_API_NATS_ACK_WAIT` | `30` | Time in seconds (`u64`) the consumer waits for an acknowledgment before a delivery becomes eligible for redelivery. |
+//! | `OPENAI_API_NATS_MAX_DELIVER` | `3` | Consumer delivery limit per message (`i64`), including the initial delivery. Positive values also enable the worker's local delivery-count check. |
+//! | `OPENAI_API_NATS_MAX_ACK_PENDING` | `100` | Maximum number of outstanding, unacknowledged messages (`i64`) across the entire consumer, shared by all its workers. |
+//! | `OPENAI_API_DEFAULT_MODEL` | Unset | Optional fallback model when the worker constructs a delivery-limit error response for a task without an explicit model. |
+//!
+//! The default model is not required to construct a JetStream worker and does not
+//! affect its subscription. Executing tasks may impose separate model requirements.
+//!
+mod producer;
+mod types;
+mod worker;
+
+#[cfg(test)]
+mod tests;
+
+pub use producer::JetStreamProducer;
+pub use worker::JetStreamWorker;