Skip to content

feat(cloudflare): add native Cloudflare Queues effect worker and push/pull consumers - #43

Closed
dylnslck wants to merge 48 commits into
mainfrom
feat/queue-worker
Closed

dylnslck wants to merge 48 commits into
mainfrom
feat/queue-worker

Conversation

@dylnslck

Copy link
Copy Markdown
Contributor

Problem and change

Connects Looms Durable Objects to external workers or background fleets over Cloudflare Queues:

  1. Acknowledged dispatch: dispatchSemantics: 'acknowledged' on EffectWorker ensures thrown queue sends (such as 429 rate limits or network errors) fail with standard retryable backoff rather than parking as ambiguous operator dispatches.
  2. Single retry authority: Consumers acknowledge the queue message immediately once the started callback is accepted, allowing Looms heartbeats and start-to-close deadlines to govern long executions without queue visibility/timeout limits.
  3. Producer & large inputs: cloudflareQueueWorker routes effect types to queue bindings, serializing versioned QueueEffectMessage records with input offload hooks for inputs exceeding ~100 KB.
  4. Fairness guard: maxInFlightWorkerEffects caps concurrent in-flight worker dispatches per run, preventing a single fanout from saturating a shared queue.
  5. In-flight cancellation: In-flight attempts learn of cancellation from heartbeat responses and abort the task signal.

Review notes

Includes an in-memory queue simulation for unit testing duplicate delivery, producer 429 backoff, consumer crash recovery, stale attempt rejection, and cancellation mid-flight.

Checked with bun x tsc -p packages/{runtime,cloudflare}/tsconfig.build.json --noEmit, bun x oxlint --deny-warnings on changed sources, and bun test src in packages/cloudflare and packages/runtime. bun run verify was not run.

Existing contributor checklist

  • bun run verify passes
  • Changeset added if @swirls/looms behavior changes (bun run changeset)
  • Docs updated for API / architecture changes

Made with Cursor

dylnslck and others added 30 commits October 23, 2026 10:13
BREAKING CHANGE: POST /runs/:runId/effects/:effectId/:attempt/:action returns { status, cancelRequested } instead of the run state. createLooms().workerCallback and the HTTP client return the same shape. Heartbeats may carry details that the next attempt's task includes. failEffect and worker fail callbacks can mark an attempt non-retryable or set retryAfterMs.

Co-authored-by: Cursor <cursoragent@cursor.com>
Attributes are immutable labels on the start event, run state, and summary, and authorization can see them. Secrets stay in a side store for handlers and in-process dispatch, and are kept out of events, snapshots, and HTTP responses.

Co-authored-by: Cursor <cursoragent@cursor.com>
…b store

Strings over 64 KiB, and any event payload that would still exceed 1 MiB, are stored by SHA-256. Reads and snapshots resolve the references, and trim deletes a body once nothing retained points at it.

Co-authored-by: Cursor <cursoragent@cursor.com>
Start, spawn, and recovery can load a definition that was not registered on the module, pin that version on the run, and fail a missing start before any event is written.

BREAKING CHANGE: Evaluator definitions require version, and lookup takes the name and version. defineEvaluator still defaults the version to v1.

Co-authored-by: Cursor <cursoragent@cursor.com>
WebSocket subscribers replay the retained log and then receive appends without a subscription held in memory, so an idle socket can sleep. SSE uses the same hello and resync frames.

Co-authored-by: Cursor <cursoragent@cursor.com>
Spawn, map, fanout, and loop can start a child run that signals the parent when it finishes. Cancelling the parent either stops those children immediately or waits until they finish cancelling.

BREAKING CHANGE: RunLauncher implementations must provide signal and cancel.

Co-authored-by: Cursor <cursoragent@cursor.com>
EventSource and browser WebSockets cannot send an Authorization header. A short-lived HMAC grant in the query string or WebSocket subprotocol lets one principal read one run.

Co-authored-by: Cursor <cursoragent@cursor.com>
…/pull consumers

Add cloudflareQueueWorker producer implementing EffectWorker with dispatchSemantics: 'acknowledged', queue push/pull consumers with single-retry authority and mid-flight cancellation abort, and maxInFlightWorkerEffects fairness guard per run.

Co-authored-by: Cursor <cursoragent@cursor.com>
…tasks, and child runs

Store W3C traceparent and tracestate on RunStartIdentity, pass child trace contexts to in-process effect handlers (EffectContext) and external worker tasks (EffectWorkerTask), link child runs spawned from effects with child span IDs, and export trace context parsing/injection utilities from @looms/otel.

Co-authored-by: Cursor <cursoragent@cursor.com>
The fold assigns traceparent and tracestate after building the start identity. The runtime no longer copies trace context into the requested identity, because the retry check never compares it.

Co-authored-by: Cursor <cursoragent@cursor.com>
…full fold

Run attributes are immutable and recorded on runtime.run.started, so authorization reads seq 1 and only falls back to loading the run when the log is trimmed. Store failures now fail the request instead of authorizing without attributes. Also fixes the start-body attribute decode, which did not type-check.

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
… runs

Dropping a reference used to list and read every run's manifest, so each trim cost grew with the number of runs in the bucket. Each run now writes refs/<digest>/<run> markers, and collecting a body lists only that digest's markers.

The old scheme also lost bodies when two Durable Objects shared a bucket. retain now writes markers before checking that each new body exists and fails the append if one was deleted. collect lists markers again after deleting and restores the body if a reference appeared. BlobBytes gains exists, and R2BucketLike gains head, so the check does not download the body.

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
…text

A transport that persists tasks cannot carry secrets, so its worker needs another way to read them. POST /runs/:runId/effects/:effectId/:attempt/secrets returns the secret context only while that attempt is started and unsettled, and stale or cancelled otherwise. It uses worker access, so it stays closed until the host configures worker auth, and responses are not cacheable. client.workerSecrets wraps it.

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
The merge takes feat/child-runs as is. The next commit reimplements trace propagation on top of it.

Co-authored-by: Cursor <cursoragent@cursor.com>
POST /runs reads the traceparent and tracestate headers, and start, startRun, and client.startRun accept trace. A valid version-00 context is recorded on runtime.run.started and startIdentity.trace. Invalid values are dropped rather than stored.

The context reaches effect handlers as ctx.trace, worker tasks as task.trace, and launched and child runs unless the launch sets its own. It is passed unchanged: Looms records no spans, so minting span ids would point at parents that never exist. The object's keys are the W3C header names, so it works directly as an OpenTelemetry carrier.

The local executor now takes one RunContext (correlation id, attributes, secrets, trace) in place of separate positional arguments.

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
The read grant now goes on the WebSocket subscription URL as well as the SSE one, through the shared events query.

Co-authored-by: Cursor <cursoragent@cursor.com>
The Durable Object accepts run WebSockets outside the fetch handler, so it called authorize without the attributes every other run route passes. An authorizer that checks the tenant saw none. runRouteInfo now builds RouteInfo for both paths from the same start-event read.

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
A browser that offers a grant in Sec-WebSocket-Protocol fails the handshake unless the response selects it. grantSubprotocol returns the offered v1. value so the Durable Object and other hosts can echo it.

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
The retry is scheduled from the wake's clock, which is read before the event timestamp, so comparing against the event could fail by a millisecond. The assertion also no longer skips when the payload is missing.

Co-authored-by: Cursor <cursoragent@cursor.com>
dylnslck and others added 18 commits October 23, 2026 15:04
The producer now sends trace context and attempt deadlines, and offloads inputs over 100 KB to the blob store as inputRef. Consumers use the worker callback client, read secrets through the secrets callback, ack after started, and report complete, fail, or cancelled once. A pull consumer with a Cloudflare HTTP source batches acks, backs off when idle, reports the backlog, and drains on stop. The in-memory queue models leases, retries, and dead letters.

In the runtime, an acknowledged dispatch rejection goes through the retry policy. Creating a runtime fails when an acknowledged worker claims an effect without a schedule-to-start timeout. The in-flight cap counts only outstanding worker attempts.

Co-authored-by: Cursor <cursoragent@cursor.com>
@dylnslck

Copy link
Copy Markdown
Contributor Author

Superseded by squash-merge of the stack tip (#46), which already includes this change set after #34 landed on main.

@dylnslck dylnslck closed this Oct 24, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant