responses-api-store

gRPC service for managing OpenAI Responses API compatible request state: storing response objects, tracking background jobs, and distributing work to background workers.

Designed as a shared dependency for Kubernetes-deployed services such as beranekio/duihua-ai-services.

Features

• Persist Responses API objects and materialized conversation input in Valkey/Redis
• Enqueue and claim background=true jobs via a Redis stream consumer group
• Reconcile stale queued responses to failed
• Tombstone in-flight background deletions instead of hard-deleting active jobs
• Rust and Go client SDKs generated from the same protobuf contract

Architecture

flowchart LR
  Gateway[Responses API gateway] -->|gRPC| Store[responses-api-store]
  Worker[Background worker] -->|gRPC| Store
  Store --> Valkey[(Valkey / Redis)]

Loading

Gateway services call the store when creating, retrieving, deleting, or enqueueing responses. Background workers claim jobs from the store, execute upstream inference, and write completed responses back.

gRPC API

Protobuf definition: proto/responsesapistore/v1/store.proto

RPC	Purpose
StoreResponse	Persist a stored response record
GetResponse	Load a record, optionally reconciling stale queued jobs
UpdateResponse	Replace a stored record
DeleteResponse	Delete or tombstone a record
EnqueueBackgroundJob	Store a queued record and publish it to the background stream
ClaimBackgroundJobs	Claim jobs for worker processing (XREADGROUP + XAUTOCLAIM); may return pending_jobs (and legacy pending_stream_ids) when record load fails transiently
AcknowledgeBackgroundJob	Acknowledge successful job processing
EnsureConsumerGroup	Bootstrap the Redis stream consumer group
GetBackgroundQueueStats	Return pending, in-progress, and workload counts for KEDA
ClaimBackgroundResponse	Atomically claim a queued background response (queued → in_progress) and return upstream payload
CompleteBackgroundResponse	Atomically persist a completed response when in_progress
FailBackgroundResponse	Atomically mark an in_progress (or claimable) response as failed
ReconcileStaleResponse	Mark stale queued responses as failed
GenerateResponseId	Allocate a new resp_* identifier
Health	Report service and Redis connectivity

Error semantics

Missing or deleted responses: GetResponse returns gRPC NOT_FOUND with message response not found: <id>. The server does not return a successful RPC with an empty record. Rust clients should use responses_api_store_client::is_not_found to detect absent IDs from either ClientError::NotFound or ClientError::Rpc with code NOT_FOUND. The Go client provides client.IsNotFound(err).

Storage failures: Valkey/Redis errors map to gRPC UNAVAILABLE with messages like storage unavailable (timeout): .... The (timeout), (connection_refused), (busy), or (other) prefix distinguishes common failure modes. Background workers should retry ClaimBackgroundJobs on UNAVAILABLE. Blocking XREADGROUP calls open a dedicated Redis connection per request (not the shared command pool) with a client response timeout of block_ms + 500ms, so values above the default 500ms redis-rs timeout work as intended without wedging server tasks indefinitely.

Background transition errors: ClaimBackgroundResponse, CompleteBackgroundResponse, and FailBackgroundResponse return gRPC NOT_FOUND when the stored record is missing or expired (same response not found: <id> message as GetResponse). Use responses_api_store_client::is_not_found or client.IsNotFound(err) for that case—for example when a record TTLs out between ClaimBackgroundJobs and ClaimBackgroundResponse. These RPCs return gRPC FAILED_PRECONDITION when the record exists but is terminal (cancelled/deleted/completed/failed) or in the wrong state (for example completing while still queued). Use responses_api_store_client::is_failed_precondition or client.IsFailedPrecondition(err) for lifecycle conflicts. ClaimBackgroundResponse keeps pending_upstream_request and upstream_authorization on the stored record until completion or failure so stream/autoclaim retries can re-hydrate claim payload after worker crashes.

Repository layout

proto/                         # Canonical protobuf definitions
crates/
  proto/                       # Generated Rust protobuf + tonic stubs
  core/                        # Valkey storage and queue implementation
  server/                      # gRPC server binary
  client/                      # Rust client SDK
sdk/go/                        # Go protobuf stubs and client SDK
charts/responses-api-store/    # Helm subchart

Configuration

Environment variable	Default	Description
GRPC_LISTEN_ADDR	0.0.0.0:50051	gRPC bind address
GRPC_MAX_MESSAGE_BYTES	67108864 (64 MiB)	Max gRPC send/recv message size
RESPONSE_ID_STORE_URL	redis://valkey:6379	Valkey/Redis URL
RESPONSE_ID_STORE_KEY_PREFIX	responses-api-store:responses	Key prefix for stored responses
RESPONSE_ID_STORE_TTL_SECONDS	86400	Default TTL for stored responses
BACKGROUND_QUEUE_STREAM_KEY	responses-api-store:background	Redis stream key for background jobs
BACKGROUND_QUEUE_STREAM_MAXLEN	10000	Approximate max stream length (XADD MAXLEN ~); 0 disables trimming
BACKGROUND_RESPONSE_STALE_SECONDS	3600	Stale threshold for queued background responses
METRICS_HTTP_ENABLED	true	Enable HTTP metrics listener for KEDA/Prometheus autoscaling
METRICS_HTTP_LISTEN_ADDR	0.0.0.0:8080	HTTP bind address for background queue metrics

Background queue metrics (KEDA)

The service exposes store-agnostic queue depth for autoscaling without Valkey credentials or Redis Streams semantics.

Requirements: stats prefer the Redis Streams consumer-group lag field (Redis 7.0+). When lag is missing or NULL (for example after stream trimming), the server falls back to paginated XRANGE counts. in_progress excludes PEL entries tombstoned by MAXLEN trimming (stream ID older than the stream tail) without mutating the consumer group, so workers can still acknowledge in-flight jobs. Lag fallback needs Redis/Valkey 6.2+ for exclusive stream ranges; older servers use ID increment instead. Check Health (redis_version, background_queue_lag_supported) before enabling metrics-based autoscaling.

On cold start, stats auto-create a consumer group only when the stream exists but has no groups yet (for example jobs enqueued before the first worker starts). A mistyped consumer_group while other groups already exist returns not-found rather than creating an orphan group.

gRPC: GetBackgroundQueueStats returns:

Field	Meaning
pending	Jobs waiting to be claimed by ClaimBackgroundJobs
in_progress	Jobs claimed but not yet AcknowledgeBackgroundJob
workload	Recommended scale signal (pending + in_progress)

HTTP (when METRICS_HTTP_ENABLED=true):

GET /metrics/background-queue?consumer_group=<name>

Example response:

{
  "consumer_group": "duihua-background",
  "pending": 3,
  "in_progress": 2,
  "workload": 5
}

Prometheus text is also available at GET /metrics?consumer_group=<name>.

KEDA metrics-api scaler: point at the chart Service metrics port (default 8080), set valueLocation: workload, and tune targetValue to your desired jobs-per-replica (for example targetValue: "1" scales up when workload >= 1). Use minReplicaCount to keep a minimum number of workers running. Use activationTargetValue as the threshold that activates the scaler when scaling from zero.

CI

GitHub Actions workflow .github/workflows/validate.yml runs on pushes and pull requests to main:

• Rust formatting, clippy, and unit tests
• Go tests and verification that generated protobuf stubs match proto/
• Helm chart lint and template rendering
• Dockerfile lint (hadolint) and image build
• Integration smoke test against a Valkey service container
• Helm chart smoke test on kind (chart install + Rust smoke example)

Run the same checks locally:

make ci

Local development

Run the server

# Start Valkey locally
docker run --rm -p 6379:6379 valkey/valkey:9.1.0-alpine

# Run the gRPC server
cargo run -p responses-api-store-server

Rust client

responses-api-store-client = { path = "../responses-api-store/crates/client" }

use responses_api_store_client::Client;

let mut client = Client::connect("http://127.0.0.1:50051").await?;
let health = client.health().await?;

Go client

import "github.com/beranekio/responses-api-store/sdk/go/client"

cli, err := client.Dial(ctx, "127.0.0.1:50051")

Regenerate Go stubs after proto changes:

./scripts/generate-go.sh

Docker

docker build -t responses-api-store:local .
docker run --rm -p 50051:50051 \
  -e RESPONSE_ID_STORE_URL=redis://host.docker.internal:6379 \
  responses-api-store:local

Helm

The chart can run standalone or as a subchart. By default it deploys an embedded Valkey instance for local development and smoke testing.

Bundled Valkey is ephemeral by default: the subchart disables RDB/AOF persistence and does not mount a PVC, so pod restarts or reschedules permanently lose stored responses, stream entries, and consumer-group state. For production, disable bundled Valkey and use external Redis/Valkey with persistence. For single-node installs that should survive restarts, enable optional bundled persistence:

valkey:
  persistence:
    enabled: true
    size: 1Gi

gRPC port alignment: set grpc.port (default 50051) as the canonical listener port. The chart derives containerPort, Service port, and GRPC_LISTEN_ADDR (0.0.0.0:<port>) from it unless grpc.listenAddr or service.port is set explicitly. When grpc.listenAddr includes a port, that port is also used for containerPort and the Service.

Readiness: the chart runs /responses-api-store-probe, which calls the Health RPC and fails when redis_ok is false. Liveness remains a TCP check on the gRPC port.

Metrics: when metrics.enabled is true (default), the chart exposes port metrics.port (default 8080) on the Service for KEDA/Prometheus queue-depth scraping. Set metrics.enabled: false to disable the HTTP listener.

Install from the repository

For local development, smoke tests, or when iterating on chart templates, install directly from this repository:

helm install responses-api-store ./charts/responses-api-store

Install from GHCR

Successful pushes to main also publish an OCI Helm chart to GHCR (see .github/workflows/publish.yml). This is often more convenient than vendoring the chart path: each published version is tied to a validated main commit and pins the server image to the matching ghcr.io/beranekio/responses-api-store:<git-sha> tag.

Charts are tagged with semver prereleases of the form 0.0.0-<git-sha> rather than independent release versions for now. Install a specific commit:

helm install responses-api-store \
  oci://ghcr.io/beranekio/charts/responses-api-store \
  --version 0.0.0-<git-sha>

To track the current main head without looking up a SHA manually:

SHA="$(git ls-remote https://github.com/beranekio/responses-api-store.git refs/heads/main | cut -f1)"
helm install responses-api-store \
  oci://ghcr.io/beranekio/charts/responses-api-store \
  --version "0.0.0-${SHA}"

Parent charts can depend on the OCI chart instead of a local path:

# Chart.yaml
dependencies:
  - name: responses-api-store
    version: "0.0.0-<git-sha>"
    repository: oci://ghcr.io/beranekio/charts

Disable bundled Valkey and point at an external Redis-compatible store:

valkey:
  enabled: false
redis:
  url: redis://shared-valkey:6379

For parent charts such as duihua-ai-services, add a dependency and set gateway/worker environment variables to the release gRPC endpoint:

# values.yaml excerpt
responsesApiStore:
  enabled: true
  endpoint: responses-api-store:50051

Stored record shape

The service stores the same record shape used by duihua-ai-services:

{
  "upstream": "http://inference:8000/v1",
  "response": { "id": "resp_...", "status": "queued", "background": true },
  "input": [{ "role": "user", "content": "hello" }],
  "pending_upstream_request": { "model": "demo", "input": "hello", "store": false },
  "upstream_authorization": "Bearer ...",
  "enqueued_at": 1746500000
}

License

MIT