feat(platform-cloudflare): add DurableQueue Durable Object

One queue name is one Durable Object: items, attempt counts, and in-flight
leases live on the object's SQLite storage behind its single alarm, which
acts as a watchdog redelivering items whose worker died. CloudflareCluster
now also provides PersistedQueueFactory, so the DurableQueue user API works
on the Cloudflare path out of the box.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: 2 people

.changeset/cloudflare-persisted-queue.md

@@ -0,0 +1,11 @@
+---
+"@effect/platform-cloudflare": minor
+---
+
+Add `CloudflarePersistedQueue`, running persisted queues on the dedicated
+queue Durable Object class. One queue name is one Durable Object: items,
+attempt counts, and in-flight leases live on the object's SQLite storage
+behind its single alarm, which acts as a watchdog redelivering items whose
+worker died before completing them. `CloudflareCluster.layer` now also
+provides the `PersistedQueueFactory` service, so the `DurableQueue` user API
+works on the Cloudflare path out of the box.

packages/platform/cloudflare/src/CloudflareCluster.ts

@@ -26,10 +26,12 @@ import * as Envelope from "effect/unstable/cluster/Envelope"
 import * as RunnerAddress from "effect/unstable/cluster/RunnerAddress"
 import * as ShardId from "effect/unstable/cluster/ShardId"
 import { Sharding } from "effect/unstable/cluster/Sharding"
+import type { PersistedQueueFactory } from "effect/unstable/persistence/PersistedQueue"
 import type * as Rpc from "effect/unstable/rpc/Rpc"
 import * as RpcClient from "effect/unstable/rpc/RpcClient"
 import * as RpcSchema from "effect/unstable/rpc/RpcSchema"
 import type { WorkflowEngine } from "effect/unstable/workflow/WorkflowEngine"
+import * as CloudflarePersistedQueue from "./CloudflarePersistedQueue.ts"
 import * as CloudflareWorkflowEngine from "./CloudflareWorkflowEngine.ts"
 import * as Internal from "./internal/clusterName.ts"
 import { registerEntity as registerEntityHandler, unregisterEntity } from "./internal/entityRegistry.ts"
@@ -473,7 +475,9 @@ const make = Effect.fnUntraced(function*(options: LayerOptions) {
  *
  * Provides the cluster `Sharding` service on top of the four same-Worker
  * Durable Object namespace bindings, plus the `WorkflowEngine` backed by the
- * workflow class. `Entity.client` resolves an entity to its Durable Object by
+ * workflow class and the `PersistedQueueFactory` backed by the queue class, so
+ * the `DurableQueue` user API works out of the box. `Entity.client` resolves
+ * an entity to its Durable Object by
  * encoding `(type, id)` with {@link encodeName} and calling `getByName`; an
  * unknown entity type or a bad encode fails at the Worker before any Durable
  * Object is contacted. Entity handlers registered with `Entity.toLayer` are
@@ -484,8 +488,9 @@ const make = Effect.fnUntraced(function*(options: LayerOptions) {
  * @category layers
  * @since 4.0.0
  */
-export const layer = (options: LayerOptions): Layer.Layer<Sharding | WorkflowEngine> =>
-  Layer.merge(
+export const layer = (options: LayerOptions): Layer.Layer<Sharding | WorkflowEngine | PersistedQueueFactory> =>
+  Layer.mergeAll(
     Layer.effect(Sharding)(make(options)),
-    CloudflareWorkflowEngine.layer({ workflowNamespace: options.workflowNamespace })
+    CloudflareWorkflowEngine.layer({ workflowNamespace: options.workflowNamespace }),
+    CloudflarePersistedQueue.layer({ queueNamespace: options.queueNamespace })
   )

packages/platform/cloudflare/src/CloudflareDurableObjects.ts

@@ -45,6 +45,8 @@ import { deliverReply as deliverEntityReply } from "./internal/entityReply.ts"
 import { makeEntityRuntime } from "./internal/entityRuntime.ts"
 import { armAlarm, earliestDeliverAt, ensureEntityStorage } from "./internal/entityStorage.ts"
 import { decodeReplyFor, decodeRequest, encodeReplyFor } from "./internal/entityWire.ts"
+import { makeQueueRuntime } from "./internal/queueRuntime.ts"
+import { earliestLeaseExpiry, ensureQueueStorage } from "./internal/queueStorage.ts"
 import type { WorkflowRunOptions, WorkflowStub } from "./internal/workflowRegistry.ts"
 import { makeWorkflowRuntime } from "./internal/workflowRuntime.ts"
 import { earliestClockWakeUp, ensureWorkflowStorage, loadExecution } from "./internal/workflowStorage.ts"
@@ -59,6 +61,14 @@ type EntityRuntime = Effect.Success<ReturnType<typeof makeEntityRuntime>>
 
 type WorkflowRuntime = ReturnType<typeof makeWorkflowRuntime>
 
+type QueueRuntime = ReturnType<typeof makeQueueRuntime>
+
+interface QueueItem {
+  readonly id: string
+  readonly element: string
+  readonly attempts: number
+}
+
 interface ReplySession {
   readonly replies: Array<string>
   readonly takers: Array<{
@@ -743,13 +753,86 @@ export class ClusterWorkflow extends DurableObject<unknown> {
 }
 
 /**
- * The durable queue class. Placeholder for `DurableQueue`; one object per
- * queue name. It only reserves the binding for now.
+ * The durable queue class behind the `PersistedQueue` implementation used by
+ * `DurableQueue`. One instance holds one named queue.
+ *
+ * **Details**
+ *
+ * The constructor stays cheap: it opens SQLite, ensures the queue table, and
+ * re-arms the single alarm from the earliest pending lease expiry. Items are
+ * leased to takers for a bounded time; the alarm watchdog expires overdue
+ * leases so an item whose worker died is redelivered.
  *
  * @category durable objects
  * @since 4.0.0
  */
 export class ClusterDurableQueue extends DurableObject<unknown> {
+  readonly #state: DurableObjectState
+  #runtime: QueueRuntime | undefined
+
+  constructor(ctx: DurableObjectState, env: unknown) {
+    super(ctx, env)
+    this.#state = ctx
+    if (ctx.id.name !== undefined && decodeName(ctx.id.name) === undefined) {
+      throw new Error("ClusterDurableQueue requires a canonical queue Durable Object name")
+    }
+    ensureQueueStorage(ctx.storage.sql)
+    const expiry = earliestLeaseExpiry(ctx.storage.sql)
+    if (expiry !== undefined) {
+      void ctx.blockConcurrencyWhile(() => Effect.runPromise(armAlarm(ctx.storage, expiry)))
+    }
+  }
+
+  #getRuntime(): QueueRuntime {
+    if (this.#runtime === undefined) {
+      this.#runtime = makeQueueRuntime({
+        sql: this.#state.storage.sql,
+        alarm: this.#state.storage,
+        now: () => Date.now()
+      })
+    }
+    return this.#runtime
+  }
+
+  /** @internal Same-Worker RPC transport used by `CloudflarePersistedQueue.layer`. */
+  offer(id: string, element: string): Promise<void> {
+    return this.#getRuntime().offer(id, element)
+  }
+
+  /** @internal Waits until an item is available, then leases it to the caller. */
+  take(takerId: string, maxAttempts: number, leaseMillis: number): Promise<QueueItem> {
+    return this.#getRuntime().take(takerId, maxAttempts, leaseMillis)
+  }
+
+  /** @internal Cancels a waiting take, releasing an item already leased to it. */
+  cancelTake(takerId: string): Promise<void> {
+    return this.#getRuntime().cancelTake(takerId)
+  }
+
+  /** @internal */
+  complete(id: string): Promise<void> {
+    return this.#getRuntime().complete(id)
+  }
+
+  /** @internal Records a failed attempt and requeues the item. */
+  fail(id: string, lastFailure: string): Promise<void> {
+    return this.#getRuntime().fail(id, lastFailure)
+  }
+
+  /** @internal Requeues the item without counting an attempt. */
+  release(id: string): Promise<void> {
+    return this.#getRuntime().release(id)
+  }
+
+  /** @internal Extends the lease of an item still being processed. */
+  extend(id: string, leaseMillis: number): Promise<void> {
+    return this.#getRuntime().extend(id, leaseMillis)
+  }
+
+  override alarm(): Promise<void> {
+    return this.#getRuntime().runAlarm()
+  }
+
   override fetch: () => never = notExposed("ClusterDurableQueue")
 }
 

packages/platform/cloudflare/src/CloudflarePersistedQueue.ts

@@ -0,0 +1,145 @@
+/**
+ * Runs persisted queues on the dedicated queue Durable Object class.
+ *
+ * On this path one queue name is one Durable Object: the Worker encodes the
+ * queue name into a Durable Object name with the same length-prefix scheme as
+ * entities and resolves the object through the queue namespace binding. Items,
+ * their attempt counts, and their in-flight leases live on the object's SQLite
+ * storage behind its single alarm; a take with no available item waits inside
+ * the object until an offer, a retry, or an expired lease produces one.
+ *
+ * Delivery is at-least-once: a taken item is leased for a bounded time and the
+ * lease is refreshed while the handler runs, so an item whose worker died is
+ * redelivered once the alarm watchdog expires the lease.
+ *
+ * This backs the `DurableQueue` user API; `CloudflareCluster.layer` already
+ * includes this layer.
+ *
+ * @since 4.0.0
+ */
+import * as Cause from "effect/Cause"
+import * as Effect from "effect/Effect"
+import * as Exit from "effect/Exit"
+import * as Layer from "effect/Layer"
+import * as Schedule from "effect/Schedule"
+import * as PersistedQueue from "effect/unstable/persistence/PersistedQueue"
+import { encodeName } from "./internal/clusterName.ts"
+import type { QueueItem } from "./internal/queueStorage.ts"
+
+/**
+ * The queue Durable Object namespace binding the store is built from.
+ *
+ * @category layers
+ * @since 4.0.0
+ */
+export interface LayerOptions {
+  readonly queueNamespace: DurableObjectNamespace
+}
+
+interface QueueStub {
+  readonly offer: (id: string, element: string) => Promise<void>
+  readonly take: (takerId: string, maxAttempts: number, leaseMillis: number) => Promise<QueueItem>
+  readonly cancelTake: (takerId: string) => Promise<void>
+  readonly complete: (id: string) => Promise<void>
+  readonly fail: (id: string, lastFailure: string) => Promise<void>
+  readonly release: (id: string) => Promise<void>
+  readonly extend: (id: string, leaseMillis: number) => Promise<void>
+}
+
+const leaseMillis = 120_000
+const leaseRefreshMillis = 30_000
+
+const finalize = (run: () => Promise<void>): Effect.Effect<void> =>
+  Effect.promise(run).pipe(
+    Effect.sandbox,
+    Effect.retry({ times: 5, schedule: Schedule.exponential(100, 1.5) }),
+    Effect.orDie
+  )
+
+/**
+ * Creates the `PersistedQueueStore` backed by the queue Durable Object
+ * namespace binding.
+ *
+ * @category constructors
+ * @since 4.0.0
+ */
+export const make = (options: LayerOptions): PersistedQueue.PersistedQueueStore["Service"] => {
+  const stubFor = (name: string): QueueStub =>
+    options.queueNamespace.getByName(encodeName("PersistedQueue", name)) as unknown as QueueStub
+
+  return PersistedQueue.PersistedQueueStore.of({
+    offer: ({ element, id, name }) =>
+      Effect.tryPromise({
+        try: () => stubFor(name).offer(id, JSON.stringify(element)),
+        catch: (cause) =>
+          new PersistedQueue.PersistedQueueError({
+            message: "Failed to offer element to persisted queue",
+            cause
+          })
+      }),
+    take: ({ maxAttempts, name }) =>
+      // Uninterruptible outside `restore` so the release finalizer is always
+      // registered once an item is leased; an interrupt while still waiting
+      // cancels the take by taker id, releasing an item that was already
+      // leased to it on the object.
+      Effect.uninterruptibleMask((restore) =>
+        Effect.gen(function*() {
+          const takerId = crypto.randomUUID()
+          // A broken take RPC means the object was evicted while this taker
+          // waited; retrying re-enters the queue with nothing lost.
+          const item = yield* restore(
+            Effect.promise(() => stubFor(name).take(takerId, maxAttempts, leaseMillis)).pipe(
+              Effect.sandbox,
+              Effect.tapCause((cause) => Effect.logWarning("PersistedQueue take failed, retrying", cause)),
+              Effect.retry(Schedule.spaced(500)),
+              Effect.orDie
+            )
+          ).pipe(
+            Effect.onInterrupt(() =>
+              Effect.promise(() => stubFor(name).cancelTake(takerId)).pipe(
+                Effect.sandbox,
+                Effect.retry({ times: 5, schedule: Schedule.exponential(100, 1.5) }),
+                Effect.ignore
+              )
+            )
+          )
+          yield* Effect.addFinalizer(Exit.match({
+            onFailure: (cause) =>
+              Cause.hasInterruptsOnly(cause)
+                ? finalize(() => stubFor(name).release(item.id))
+                : finalize(() => stubFor(name).fail(item.id, Cause.pretty(cause))),
+            onSuccess: () => finalize(() => stubFor(name).complete(item.id))
+          }))
+          yield* Effect.promise(() => stubFor(name).extend(item.id, leaseMillis)).pipe(
+            Effect.sandbox,
+            Effect.ignore,
+            Effect.schedule(Schedule.spaced(leaseRefreshMillis)),
+            Effect.forkScoped,
+            Effect.interruptible
+          )
+          return {
+            id: item.id,
+            attempts: item.attempts,
+            element: JSON.parse(item.element)
+          }
+        })
+      )
+  })
+}
+
+/**
+ * Layer that provides the `PersistedQueueFactory` backed by the queue Durable
+ * Object namespace binding.
+ *
+ * **Details**
+ *
+ * `CloudflareCluster.layer` already includes this layer; use it directly only
+ * when persisted queues are needed without the rest of the cluster.
+ *
+ * @category layers
+ * @since 4.0.0
+ */
+export const layer = (options: LayerOptions): Layer.Layer<PersistedQueue.PersistedQueueFactory> =>
+  PersistedQueue.layer.pipe(
+    Layer.provide(Layer.succeed(PersistedQueue.PersistedQueueStore)(make(options)))
+  )

packages/platform/cloudflare/src/index.ts

@@ -14,6 +14,11 @@ export * as CloudflareCluster from "./CloudflareCluster.ts"
  */
 export * as CloudflareDurableObjects from "./CloudflareDurableObjects.ts"
 
+/**
+ * @since 4.0.0
+ */
+export * as CloudflarePersistedQueue from "./CloudflarePersistedQueue.ts"
+
 /**
  * @since 4.0.0
  */