Reaktor Bus

Data-layer alignment (11 September 2026). This module participates in the unified data infrastructure through the same Service contracts and main graph identities. Delivery/session mechanisms retain their own guarantees; operation, publication, commit, projection and device-delivery receipts are distinct. The generalized Service interfaces are proposed, not implied to ship with this module.

Reaktor Bus is the durable async delivery layer for Reaktor.

It is not a generic "messages happen somewhere" abstraction. It is the framework contract for events, jobs, fanout, retries, dead letters, replay, outbox relay, idempotent consumers, and graph-visible delivery policy.

Today the codebase has the ingredients:

• ServiceTransport.PUBSUB and ServiceTransport.QUEUE exist in reaktor-service as endpoint identities.
• reaktor-google has a real GCP Pub/Sub adapter with topics, subscriptions, publish, pull, acknowledge, ordering keys, push endpoints, and delivery attempts.
• BestBuds has a product-specific Pub/Sub hello path between reaktorServer and a Cloudflare Worker.
• reaktor-cloudflare has Worker service bindings, Durable Object wrappers, Hono routing, and service-account helpers.

What does not exist yet is the generic reaktor-bus module that makes those pieces one implementation model.

This document defines that module.

1. Operating Model

HTTP and service bindings are request/response. They are the wrong shape for work that should survive process failure, fan out to multiple consumers, or retry after the user request has returned.

When a user uploads media, the request should commit the authoritative state and hand off the rest:

• scan or moderate content;
• transcode and generate thumbnails;
• update search;
• notify devices;
• write analytics;
• schedule cleanup;
• retry failed downstream work.

The bus is that handoff layer.

Reaktor's rule is:

Durable async delivery is a typed graph edge, not hidden plumbing.

Every bus route must be inspectable in the graph: producer, topic or queue, envelope kind, schema version, ordering policy, retry policy, DLQ, claim-check store, tenant scope, owner, and consumer handlers.

2. Boundaries

The bus is deliberately narrow.

The terms "bus", "queue", "pubsub", "mesh", and "actor mailbox" should not be used interchangeably. They have different failure modes and different correctness contracts.

3. Queue vs Pub/Sub

Use Pub/Sub for facts. Use Queues for work.

Pub/Sub

Pub/Sub is for events that have happened and may be observed by many independent consumers.

Message.Sent
  -> update room projection
  -> enqueue notification fanout
  -> update analytics
  -> feed moderation model

Characteristics:

• one publication can reach many subscriptions;
• each subscription has its own delivery, retry, filter, and DLQ policy;
• subscriptions are independent;
• ordering is per ordering key, not global;
• consumers must be idempotent because delivery is at least once;
• replay is feasible when the provider/subscription retains messages or an outbox/event store is present.

GCP Pub/Sub is the first Reaktor Pub/Sub substrate.

Queue

A queue is for one unit of work claimed by one consumer group.

Notification.Deliver(deviceToken=abc)
  -> one queue message
  -> one delivery worker claims it
  -> success acks it
  -> transient failure retries it
  -> poison failure dead-letters it

Characteristics:

• competing consumers divide work;
• retry and DLQ are per job;
• batching is normal;
• payloads should be small;
• a queue is not an event log;
• ordering is provider-specific and often weaker than the name suggests.

Cloudflare Queues is the first Reaktor Queue substrate.

4. Provider Contract Matrix

The generic bus API must expose provider differences instead of smoothing them into lies.

5. Bus Envelope

The bus envelope is the operational contract.

It carries enough data for schema evolution, tracing, deduplication, tenant scope, ordering, replay, graph projection, and provider mapping.

@Serializable
data class BusEnvelope<T>(
    val id: String,
    val kind: String,
    val schemaVersion: Int,
    val payload: T,

    val subject: String? = null,
    val source: String? = null,
    val tenantId: String? = null,

    val correlationId: String? = null,
    val causationId: String? = null,
    val traceparent: String? = null,

    val idempotencyKey: String? = null,
    val orderingKey: String? = null,
    val partitionKey: String? = null,

    val createdAtEpochMillis: Long,
    val deadlineEpochMillis: Long? = null,

    val contentType: String = "application/json",
    val claimCheck: ClaimCheckRef? = null,
    val attributes: Map<String, String> = emptyMap(),
)

@Serializable
data class ClaimCheckRef(
    val store: String,
    val key: String,
    val sha256: String,
    val contentType: String,
    val sizeBytes: Long,
)

Required fields:

• id: unique delivery identity, stable across retry and replay unless replay explicitly creates a new attempt.
• kind: semantic message type, for example Message.Sent or Notification.Deliver.
• schemaVersion: payload schema version.
• payload: typed payload or a lightweight descriptor when claimCheck carries the large body.
• createdAtEpochMillis: publication time at the producer boundary.

Strongly recommended fields:

• tenantId: tenant/app scope restored by the consumer before work.
• traceparent: W3C trace context for cross-runtime correlation.
• idempotencyKey: stable key for the business effect, often different from id.
• orderingKey: provider ordering key when supported.
• source: graph node id of the producer.
• subject: resource identity affected by the message.

Provider message ids, ack ids, receive counts, and queue names are delivery metadata. They belong in BusDeliveryContext, not in the logical envelope.

data class BusDeliveryContext(
    val provider: BusProvider,
    val routeId: String,
    val topicOrQueue: String,
    val providerMessageId: String?,
    val providerAckId: String?,
    val attempt: Int?,
    val receivedAtEpochMillis: Long,
)

6. Bus API

The core API should stay small.

interface ReaktorBus {
    suspend fun <T> publish(
        kind: String,
        payload: T,
        options: PublishOptions = PublishOptions(),
    ): PublishResult

    fun <T> handle(
        kind: String,
        handler: suspend (BusEnvelope<T>, BusDeliveryContext) -> ConsumeResult,
    )
}

data class PublishOptions(
    val subject: String? = null,
    val source: String? = null,
    val tenantId: String? = null,
    val correlationId: String? = null,
    val causationId: String? = null,
    val idempotencyKey: String? = null,
    val orderingKey: String? = null,
    val delaySeconds: Int? = null,
    val deadlineEpochMillis: Long? = null,
)

sealed interface ConsumeResult {
    data object Ack : ConsumeResult
    data class Retry(val reason: String, val delaySeconds: Int? = null) : ConsumeResult
    data class DeadLetter(val reason: String) : ConsumeResult
}

Consumer outcome must be explicit. A handler that throws can map to Retry, but a handler that intentionally rejects a poison message should return DeadLetter.

7. Route Policy

Routing is graph-authored data.

bus.route(
    id = "chat-message-sent",
    kind = "Message.Sent",
    transport = ServiceTransport.PUBSUB,
    topic = "chat-events",
    orderingKey = "{tenantId}:{roomId}",
    schema = "MessageSentV3",
)

bus.route(
    id = "notification-deliver",
    kind = "Notification.Deliver",
    transport = ServiceTransport.QUEUE,
    queue = "notification-delivery",
    maxAttempts = 5,
    deadLetter = "notification-delivery-dlq",
    claimCheckThresholdBytes = 96 * 1024,
)

Every route declares:

• route id;
• message kind;
• owner service/team;
• transport;
• topic or queue;
• schema and current version;
• producer auth boundary;
• consumer auth boundary;
• retry policy;
• max attempts;
• DLQ target;
• retention expectation;
• ordering key expression;
• idempotency key expression;
• claim-check policy;
• redaction policy for logs and DLQ inspection;
• graph node ids for producer, broker resource, consumer, outbox, and DLQ.

Unknown kind must fail before sending. A bus that silently publishes unknown message types is not debuggable.

8. GCP Pub/Sub Binding

GCP Pub/Sub is Reaktor's durable event fanout substrate.

Use it for:

• domain facts consumed by multiple services;
• cross-service event fanout;
• JVM-to-Worker and Worker-to-JVM async handoff;
• durable actor fallback wires where ordering key maps to actor identity;
• analytics/change feeds;
• long-running server-side effects that are not tied to a single edge queue.

Do not use it for:

• direct browser realtime;
• low-latency peer media/data;
• large binary payload transport;
• single-step edge-local work where Cloudflare Queues is enough.

Existing Reaktor Surface

reaktor-google already defines:

• PubSubTopic(projectId, topicId);
• PubSubSubscription(projectId, subscriptionId);
• PubSubMessage(data, attributes, orderingKey);
• PulledPubSubMessage(ackId, messageId, data, attributes, orderingKey, publishTime, deliveryAttempt);
• PubSubSubscriptionOptions(ackDeadlineSeconds, pushEndpoint);
• PubSubAdapter.ensureTopic;
• PubSubAdapter.ensureSubscription;
• PubSubAdapter.publish;
• PubSubAdapter.pull;
• PubSubAdapter.acknowledge.

The bus layer should wrap this adapter instead of replacing it.

Provider Facts To Encode

Important GCP Pub/Sub facts the Reaktor adapter must reflect:

• Pub/Sub is asynchronous, service-to-service messaging. It is not an end-user realtime transport.
• A publish request is limited to 10 MB or 1000 messages.
• One message's data field is limited to 10 MB.
• A message can have up to 100 attributes.
• Unacknowledged messages are retained for 7 days by default.
• Subscriptions expire after 31 days of inactivity by default.
• Retry policy always exists. The default is immediate redelivery; a subscription can choose exponential backoff.
• Exponential retry backoff minimum and maximum values are in the 0 to 600 second range.
• Dead-letter policy is configured on a subscription.
• Dead-letter max delivery attempts is best effort and ranges from 5 to 100.
• Ordering is by ordering key. Messages with the same ordering key are delivered serially to a subscriber.
• A hot ordering key is limited by subscriber processing speed.
• Push subscriptions with ordering enabled deliver only one outstanding message per ordering key.
• Exactly-once delivery can prevent redelivery after successful acknowledgement, but consumers still need idempotency for business effects.
• A Pub/Sub topic can enforce a schema and reject invalid publishes.

Envelope Mapping

Map envelope fields to Pub/Sub as follows:

Use reserved rk_ attributes for framework metadata. User attributes stay in the envelope's attributes map and must not overwrite rk_ keys.

Consumer Outcome Mapping

For pull subscriptions:

For push subscriptions:

Push endpoints must verify the push request and service identity. A public Worker endpoint that accepts Pub/Sub pushes without verification is not a bus boundary; it is an unauthenticated webhook.

Pub/Sub Route Defaults

Recommended defaults:

9. Cloudflare Queues Binding

Cloudflare Queues is Reaktor's edge work-queue substrate.

Use it for:

• Worker-native background jobs;
• notification delivery;
• media cleanup;
• cache invalidation;
• webhook retries;
• bounded fanout jobs after a Pub/Sub event;
• edge-to-edge or edge-to-origin work handoff;
• pull-based integration with non-Worker consumers.

Do not use it for:

• retained domain event history;
• fanout to many independent subscribers;
• ordering-sensitive event streams;
• direct browser realtime;
• large payload movement.

Provider contract and current limits

The Cloudflare adapter remains proposed. Its producer uses send/sendBatch; a Worker consumer receives batches through queue(batch, env, ctx) and explicitly acknowledges or retries individual messages. Queue delivery is at least once and ordering is not guaranteed. Broker acceptance and consumer completion are different receipts. Delivery guarantees and ordering define the provider boundary.

Declare batch size/wait, retry classification, delays, concurrency, retention and DLQ policy in the admitted environment profile. Consult current limits rather than copying a capacity table into several Reaktor pages. A queue has one active consumer configuration; independent fanout belongs to separate routes/subscriptions. Pull consumers can execute outside Workers with their own authority and visibility/ack protocol.

The Queues and Workflows chapter owns stable effect identity, atomic consumer dedup, claim-check references, source positions and authorized DLQ redrive. Its handoff contract defines when a Queue item may be acknowledged after starting a durable multi-step run. Queues do not replace the domain authority, retained log or device scheduler.

Reaktor Cloudflare Surface

reaktor-cloudflare should add a typed queue binding:

class QueueBinding internal constructor(name: String) : Binding<WorkerQueue>(name, "Queue")

class WorkerQueue internal constructor(...) {
    suspend fun <T> send(
        envelope: BusEnvelope<T>,
        options: QueueSendOptions = QueueSendOptions(),
    ): QueueSendResult

    suspend fun <T> sendBatch(
        envelopes: List<BusEnvelope<T>>,
        options: QueueBatchOptions = QueueBatchOptions(),
    ): QueueBatchSendResult
}

data class QueueSendOptions(
    val delaySeconds: Int? = null,
    val contentType: QueueContentType = QueueContentType.Json,
)

The Worker consumer wrapper should decode batches into typed bus deliveries:

suspend fun <T> consumeQueue(
    batch: WorkerQueueBatch,
    route: BusRoute<T>,
    handler: suspend (BusEnvelope<T>, BusDeliveryContext) -> ConsumeResult,
)

Consumer Outcome Mapping

Per-message ack and retry should be preferred for mixed-success batches.

Cloudflare Queue Route Defaults

Recommended defaults:

10. Reaktor Realtime Is Mesh, Not Bus

Cloudflare Realtime and TURN are not bus providers.

They belong to Reaktor Mesh:

• TURN issues short-lived ICE credentials for WebRTC connectivity.
• Cloudflare Realtime SFU hosts sessions and tracks for low-latency audio, video, and data.
• DataChannels carry live peer/session data, not durable work.

The bus may record durable effects caused by a mesh session, for example:

• Call.Started;
• Presence.Changed;
• Collaboration.PatchCommitted;
• Session.RecordingRequested;
• Mesh.TrackFailed.

But the bus does not carry raw audio, video, WebRTC signaling state, or every cursor movement. If a realtime message must survive disconnect and replay, mirror it as a bus event or persist it through a service before acknowledging it to the user.

11. Claim-Check Policy

Providers have hard payload limits, and production systems should not push large blobs through brokers anyway.

Reaktor uses claim-check for large or sensitive payloads:

1. Producer stores the large payload in object storage.
2. Producer computes hash, size, content type, and retention metadata.
3. Bus envelope carries claimCheck, not the payload body.
4. Consumer resolves the object through an authorized adapter.
5. Consumer verifies hash and size before processing.

Default thresholds:

Supported stores should include R2, GCS, and local test storage. Claim-check references must include tenant/app scope and expire according to route retention.

12. Idempotency And Dedup

The bus does not promise exactly-once delivery.

It promises a practical, auditable result:

at-least-once delivery
+ idempotent consumer
+ transactional outbox
+ dedup record
= effectively-once business effect

Each route declares its idempotency strategy:

Consumer flow:

1. Decode and validate envelope.
2. Resolve tenant/auth context.
3. Check dedup store.
4. Run handler transaction.
5. Record dedup success in the same transaction when possible.
6. Ack provider message only after the effect is committed.

Dedup stores:

13. Outbox

Domain code should not publish directly after a database write unless losing the event is acceptable.

Use a transactional outbox:

1. Domain transaction writes authoritative state.
2. The same transaction writes an outbox row with the bus envelope.
3. A relay publishes the row to Pub/Sub or Queue.
4. Relay marks the row published.
5. Consumer dedups by envelope id or business key.

This solves the dual-write problem:

• state changed but event was never published;
• event published but state rolled back;
• relay crashed after publish but before marking the row published.

Runtime-specific shape:

Outbox rows must include:

• envelope id;
• kind;
• schema version;
• tenant id;
• route id;
• serialized envelope;
• status;
• attempt count;
• next attempt time;
• last error;
• created time;
• published time;
• trace id.

14. DLQ And Replay

DLQ is not a graveyard. It is an operations surface.

Every durable route must declare:

• whether DLQ is required;
• max attempts before DLQ;
• DLQ topic or queue;
• who owns the DLQ;
• which fields are redacted;
• retention;
• inspection permissions;
• replay command shape;
• whether replay preserves original envelope id;
• whether replay preserves original idempotency key;
• how replay is audited.

Replay is a typed control-plane command, not a provider-console poke:

data class ReplayBusMessage(
    val dlqRouteId: String,
    val messageId: String,
    val replayMode: ReplayMode,
    val reason: String,
)

enum class ReplayMode {
    PreserveEnvelopeId,
    NewEnvelopeIdPreserveIdempotencyKey,
    NewEnvelopeAndIdempotencyKey,
}

Default replay mode should preserve the idempotency key. Creating a new business effect must be explicit.

15. Schema And Compatibility

Each bus route is a public contract between producers and consumers.

Rules:

• kind names the semantic event/job, not the provider resource.
• schemaVersion changes when the payload contract changes.
• Consumers may accept multiple schema versions during rollout.
• Producers publish one current version unless a route explicitly supports dual-write.
• Breaking changes require a new kind or a migration window.
• Provider schema enforcement is a useful second line of defense for Pub/Sub, not the only validation.

Recommended naming:

16. Graph Projection

The bus must be visible as graph structure.

Edges should show:

• publish;
• subscribe;
• enqueue;
• consume;
• retry;
• dead-letter;
• replay;
• claim-check read/write;
• outbox relay;
• dedup check/write;
• tenant/auth boundary;
• trace propagation.

The workbench should flag:

• a durable route with no DLQ;
• a consumer with no idempotency strategy;
• a Pub/Sub route with ordering requested but no ordering key;
• a queue route that assumes strict ordering;
• a payload that can exceed provider limits without claim-check;
• a route crossing tenant or residency boundaries without policy;
• an unowned topic, queue, or subscription;
• a DLQ with no replay path.

17. Observability

Every publish and consume operation emits telemetry.

Minimum spans:

• reaktor.bus.publish;
• reaktor.bus.consume;
• reaktor.bus.outbox.relay;
• reaktor.bus.dlq.write;
• reaktor.bus.replay;
• reaktor.bus.claim_check.write;
• reaktor.bus.claim_check.read.

Minimum attributes:

• rk.bus.route_id;
• rk.bus.kind;
• rk.bus.schema_version;
• rk.bus.provider;
• rk.bus.topic;
• rk.bus.queue;
• rk.bus.tenant_id;
• rk.bus.envelope_id;
• rk.bus.idempotency_key;
• rk.bus.ordering_key;
• rk.graph.producer_node_id;
• rk.graph.consumer_node_id;
• rk.bus.attempt;
• rk.bus.result.

Cloudflare Queue metrics should surface backlog count, backlog bytes, and oldest message age. Pub/Sub metrics should surface publish errors, subscription backlog, oldest unacked age, delivery attempt distribution, and DLQ volume.

18. Module Plan

Build the bus in small layers:

1. reaktor-bus-core: envelope, route model, publish options, consume result, errors, validation.
2. reaktor-bus-test: in-memory bus, deterministic retry/replay fixtures, contract tests.
3. reaktor-bus-pubsub: adapter over reaktor-google.
4. reaktor-cloudflare queue bindings: QueueBinding, WorkerQueue, batch/message wrappers.
5. reaktor-bus-cloudflare: Cloudflare Queue producer and consumer helpers.
6. reaktor-bus-outbox: outbox contract, relay state machine, dedup contract.
7. reaktor-bus-outbox-jvm: Postgres/Cockroach implementation.
8. reaktor-bus-outbox-cloudflare: D1/Durable Object implementation.
9. reaktor-bus-graph: graph projection visitor.
10. Reaktor Desktop/control-plane DLQ inspection and replay commands.

ServiceTransport.PUBSUB and ServiceTransport.QUEUE should remain endpoint identities in reaktor-service. Dispatch belongs in reaktor-bus.

19. Acceptance Tests

A real implementation is not done until these pass:

• unknown kind fails before provider send;
• publishing a known Pub/Sub route chooses the expected topic;
• publishing a known Queue route chooses the expected queue;
• Pub/Sub envelope maps reserved rk_ attributes correctly;
• Cloudflare Queue send rejects or claim-checks payloads above threshold;
• Pub/Sub ordering key is generated from route expression;
• queue route with strict ordering requested fails validation;
• duplicate idempotencyKey is skipped safely;
• handler Ack acks the provider message only after commit;
• handler Retry does not ack the provider message;
• handler DeadLetter writes to DLQ before acking when using explicit DLQ mode;
• outbox row and domain write commit atomically;
• relay crash after publish does not duplicate the business effect;
• DLQ replay is a typed audited command;
• graph export shows route, topic/queue, consumer, outbox, dedup, claim-check, and DLQ nodes;
• provider limits are unit-tested from constants so docs and implementation cannot drift silently.

20. Rough Edges

21. Provider References

Use current provider docs while implementing adapters:

• GCP Pub/Sub overview
• GCP Pub/Sub retry policy
• GCP Pub/Sub dead-letter topics
• GCP Pub/Sub ordering
• GCP Pub/Sub exactly-once delivery
• GCP Pub/Sub schemas
• GCP Pub/Sub quotas and limits
• Cloudflare Queues JavaScript APIs
• Cloudflare Queues batching and retries
• Cloudflare Queues dead-letter queues
• Cloudflare Queues pull consumers
• Cloudflare Queues limits

Related Docs

• Service Layer - Operation contracts and transport identities.
• Reaktor Mesh - Live peer/session/media/data transport over WebRTC, TURN, SFU, and DataChannels.
• Work - Background task scheduling and handoff boundaries.
• Server - Production backend patterns that use the bus.
• Notifications - Product-specific queue/outbox/DLQ design that should eventually reuse the generic bus.
• Unified Graph - How queues, topics, services, and cloud resources become graph nodes.