Orchestrator Messaging Blueprint (RabbitMQ + CEL + 512KB)¶

This document is the implementation blueprint for ADR 0006.

Scope¶

Open-source, local Kubernetes deployment.
Durable asynchronous workflow execution.
Two-track message handling: small vs large payloads.

Component Topology¶

orchestrator-api
validates incoming run command
determines track (small or large)
stores large payload by reference when needed (current implementation uses storage-service artifacts API)
writes accepted and queued state to PostgreSQL
publishes command to RabbitMQ
orchestrator-worker-small
consumes orchestrator.run.small
performs bounded in-flow transforms and CEL routing
orchestrator-worker-large
consumes orchestrator.run.large
resolves payload_ref
executes service-call-only transform path
postgres
run state, step state, idempotency receipts, event log
rabbitmq
command routing, retries, DLQ
minio
payload offload for large messages

Current implementation detail:

API endpoint: POST /run/{workflow_name}/enqueue
small track: payload is queued inline
large track: payload is uploaded first to storage artifacts, then queued as payload_ref
MinIO remains the target object-storage backend in deployment design; storage-service acts as the offload adapter in the current code path.

Queue consumers (step 4):

Process entrypoint: python -m app.queue_worker
Deployments:
orchestrator-worker-small consumes orchestrator.run.small
orchestrator-worker-large consumes orchestrator.run.large
On message:
validate envelope
resolve payload (payload_inline or storage artifact download + sha256 check)
execute workflow via orchestrator runtime (trigger_type=queue)
ACK on success; failed messages are rejected to retry/DLQ topology

PostgreSQL run-state (enabled when ORCH_MESSAGE_DATABASE_DSN is set):

Phase	`workflow_runs.status`	Events
enqueue accepted	`accepted`	`run.accepted`
after RabbitMQ publish	`queued`	`envelope.published`
worker picked up	`running`	`run.dequeued`
success	`succeeded` + `message_receipts` row	`run.succeeded`
failure	`failed`	`run.failed`

Duplicate enqueue with the same X-Idempotency-Key after a succeeded run returns accepted: false and the prior request_id.

Queue/Exchange Layout¶

Exchange: orchestrator.commands (type: topic)
Exchange: orchestrator.retry (type: topic)

Queues:

orchestrator.run.small
orchestrator.run.large
orchestrator.retry.small
orchestrator.retry.large
orchestrator.dlq

Routing keys:

run.small
run.large
retry.small
retry.large
dead

Retry pattern:

worker NACK/route to retry exchange with incremented attempt header
retry queue TTL expires and dead-letters back to run queue
max attempts exceeded -> route to DLQ

Message Envelope¶

{
  "message_id": "uuid",
  "correlation_id": "uuid-or-external-id",
  "workflow_name": "string",
  "track": "small",
  "headers": {
    "content_type": "application/json",
    "source": "gateway",
    "tenant": "default"
  },
  "idempotency_key": "string",
  "payload_inline": {},
  "payload_ref": null,
  "created_at": "2026-06-02T10:00:00Z"
}

Rules:

exactly one of payload_inline, payload_ref must be present
track=small requires payload_inline
track=large requires payload_ref

payload_ref shape:

{
  "bucket": "workflow-payloads",
  "key": "runs/<run-id>/input.bin",
  "version": "optional-version-id",
  "sha256": "hex"
}

PostgreSQL Schema (minimum)¶

`workflow_runs`¶

run_id (PK)
message_id (unique)
workflow_name
track
status
idempotency_key
payload_ref_json
created_at
updated_at

`workflow_steps`¶

run_id (FK)
step_id
attempt
status
started_at
finished_at
error_code
error_message

`message_receipts`¶

idempotency_key (PK)
run_id
final_status
result_ref_json
created_at

`workflow_events`¶

event_id (PK)
run_id (FK)
event_type
event_payload_json
created_at

CEL Routing Contract¶

Applicable only to small track.

Example route rule:

routes:
  - name: high_value
    when: "has(payload.amount) && payload.amount > 10000"
    target: "workflow_high_value"
  - name: default
    when: "true"
    target: "workflow_standard"

Guardrails:

expression size limit
evaluation timeout
denylist for unsafe dynamic functions

Memory and Performance Guardrails¶

ORCH_MESSAGE_TRACK_THRESHOLD_BYTES=524288
ORCH_ASYNC_QUEUE_ENABLED=true
ORCH_MESSAGE_STORAGE_UPLOAD_TIMEOUT_SECONDS=10
ORCH_WORKER_PREFETCH_SMALL=20
ORCH_WORKER_PREFETCH_LARGE=2
ORCH_MAX_INLINE_TRACE_BYTES to avoid payload logging blowups
separate worker deployments for small and large tracks

Local Kubernetes Rollout Plan¶

Deploy RabbitMQ, PostgreSQL, MinIO.
Deploy orchestrator-api with track split + publish only.
Deploy orchestrator-worker-small and validate end-to-end small flow.
Deploy orchestrator-worker-large with MinIO payload resolution.
Enable retry queues and DLQ alarms/metrics.
Add dashboards for run lifecycle and queue depth.

Operational Metrics¶

queue depth per track
dequeue latency and run completion time
retry rate and DLQ rate
worker memory RSS by track
idempotency hit ratio