Cluster sharding on Cloudflare Durable Objects

.changeset/cloudflare-alchemy-cluster.md

@@ -0,0 +1,22 @@
+---
+"@effect/platform-cloudflare": patch
+---
+
+Add Alchemy v2 deployment for the Cloudflare cluster to
+`@effect/platform-cloudflare`.
+
+`AlchemyCloudflareCluster.make` runs inside an Effect-native
+`Cloudflare.Worker` init program: it registers the four cluster Durable
+Object classes on the hosting Worker (Alchemy owns bindings, class exports,
+and SQLite migrations), builds the cluster layer together with the user's
+handler layer into the isolate-lifetime scope, and returns a handle with
+`provide`, `wake`, the four native namespace bindings, and the built
+`context`. The user never declares or re-exports Durable Object classes, and
+Cron Triggers stay user-declared via
+`Cloudflare.Workers.cron(expr, cluster.wake(name))`.
+
+The new `CloudflareDurableObjectPrograms` module exposes the class-independent
+programs behind the bundled Durable Object classes (entity, workflow, durable
+queue, singleton) so a framework that creates its own native classes can run
+the same behavior. The Wrangler path is unchanged and never imports `alchemy`;
+the `alchemy` peer dependency (`>=2.0.0-beta <3`) is optional.

.changeset/cloudflare-cluster.md

@@ -0,0 +1,20 @@
+---
+"effect": patch
+---
+
+Add the `@effect/platform-cloudflare` package, running Effect Cluster on
+Cloudflare Workers and Durable Objects.
+
+One entity instance is one Durable Object with its SQLite storage as the
+system of record. The package provides the four Durable Object classes
+(entity, workflow, durable queue, singleton), the length-prefixed entity name
+encoding, and `CloudflareCluster.layer`, which wires the cluster `Sharding`
+service, the `WorkflowEngine`, and the `PersistedQueueFactory` from the
+same-Worker namespace bindings. The `Entity`, `Workflow`, `Activity`,
+`DurableClock`, `DurableQueue`, `Singleton`, and `ClusterCron` user APIs are
+unchanged on this path; every `DurableClock` is durable through the object's
+alarm.
+
+Queue handlers receive one-based attempt metadata. Retried items wait for the
+configured retry delay while leased, before handler delivery. Global queue
+cleanup is explicitly unsupported; completed rows remain retained for deduplication.

.changeset/cloudflare-tell-no-pin.md

@@ -0,0 +1,5 @@
+---
+"@effect/platform-cloudflare": patch
+---
+
+Complete a persisted tell's invoke once the request is journaled and its handler forked, instead of pinning the caller until the handler finishes. Replayed tell rows no longer delay other callers' invoke results either; alarm re-arming and replay failure handling still await handler completion, and volatile tells keep their best-effort in-request execution.

.changeset/config.json

@@ -28,6 +28,7 @@
       "@effect/opentelemetry",
       "@effect/platform-browser",
       "@effect/platform-bun",
+      "@effect/platform-cloudflare",
       "@effect/platform-deno",
       "@effect/platform-node",
       "@effect/platform-node-shared",

.github/workflows/alchemy-canary.yml

@@ -0,0 +1,29 @@
+name: Alchemy Canary
+on:
+  workflow_dispatch:
+
+permissions: {}
+
+# Typechecks the @effect/platform-cloudflare Alchemy integration (module,
+# example, and typetests) against the latest alchemy@beta instead of the
+# pinned devDependency. Credential-free; run it manually when an alchemy
+# release may have moved the surface the integration depends on.
+jobs:
+  canary:
+    name: Canary
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+    timeout-minutes: 20
+    steps:
+      - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6
+      - name: Install dependencies
+        uses: ./.github/actions/setup
+      - name: Install latest alchemy@beta
+        run: |
+          pnpm --dir packages/platform/cloudflare add --save-dev --config.minimumReleaseAge=0 alchemy@beta
+          pnpm ls --dir packages/platform/cloudflare alchemy
+      - name: Typecheck module, example, and typetests
+        run: pnpm exec tsc -b packages/platform/cloudflare/tsconfig.alchemy.json
+      - name: Run typetests
+        run: pnpm test-types --target '>=5.9' AlchemyCloudflareCluster

.github/workflows/check.yml

@@ -50,6 +50,8 @@ jobs:
         with:
           deno-version: v2.9.4
       - run: pnpm check
+      - name: Typecheck Alchemy integration
+        run: pnpm exec tsc -b packages/platform/cloudflare/tsconfig.alchemy.json
       - run: deno check .
       - run: pnpm test-types --target '>=5.9'
 

deno.json

@@ -27,6 +27,7 @@
     "packages/opentelemetry/",
     "packages/platform/browser/",
     "packages/platform/bun/",
+    "packages/platform/cloudflare/",
     "packages/platform/node/",
     "packages/platform/node-shared/",
     "packages/tools/",

packages/effect/src/unstable/cluster/Entity.ts

@@ -716,6 +716,11 @@ export const keepAlive: (
   never,
   Sharding | CurrentAddress
 > = Effect.fnUntraced(function*(enabled: boolean) {
+  const ohandler = yield* Effect.serviceOption(KeepAliveHandler)
+  if (ohandler._tag === "Some") {
+    yield* ohandler.value(enabled)
+    return
+  }
   const olatch = yield* Effect.serviceOption(KeepAliveLatch)
   if (olatch._tag === "None") return
   if (!enabled) {
@@ -783,3 +788,21 @@ export const KeepAliveRpc = Rpc.make("Cluster/Entity/keepAlive")
 export class KeepAliveLatch extends Context.Service<KeepAliveLatch, Latch.Latch>()(
   "effect/cluster/Entity/KeepAliveLatch"
 ) {}
+
+/**
+ * Service tag for the runtime hook behind {@link keepAlive}.
+ *
+ * **Details**
+ *
+ * Runtimes that support pinning an entity in memory provide this service; the
+ * handler receives `true` while at least one keep-alive holder exists and
+ * `false` once the last holder is released. When the service is absent,
+ * `keepAlive` is a no-op.
+ *
+ * @category services
+ * @since 4.0.0
+ */
+export class KeepAliveHandler extends Context.Service<
+  KeepAliveHandler,
+  (enabled: boolean) => Effect.Effect<void>
+>()("effect/cluster/Entity/KeepAliveHandler") {}

packages/effect/src/unstable/workflow/DurableClock.ts

@@ -60,6 +60,24 @@ const InstanceTag = Context.Service<
   "effect/workflow/WorkflowEngine/WorkflowInstance" satisfies typeof WorkflowInstance.key
 )
 
+/**
+ * Context reference containing the default `inMemoryThreshold` used by
+ * {@link sleep} when the option is not passed.
+ *
+ * **Details**
+ *
+ * Workflow engines whose timers are always durable (for example the Cloudflare
+ * Durable Object engine) provide `Duration.zero` so every `sleep` without an
+ * explicit `inMemoryThreshold` schedules a durable clock.
+ *
+ * @category services
+ * @since 4.0.0
+ */
+export const InMemoryThreshold = Context.Reference<Duration.Duration>(
+  "effect/workflow/DurableClock/InMemoryThreshold",
+  { defaultValue: () => Duration.seconds(60) }
+)
+
 /**
  * Waits inside a workflow, using an in-memory activity for durations at or
  * below the threshold and scheduling a durable clock for longer durations.
@@ -95,7 +113,7 @@ export const sleep: (
 
   const inMemoryThreshold = options.inMemoryThreshold !== undefined
     ? Duration.fromInputUnsafe(options.inMemoryThreshold)
-    : defaultInMemoryThreshold
+    : yield* InMemoryThreshold
 
   if (Duration.isLessThanOrEqualTo(duration, inMemoryThreshold)) {
     return yield* Activity.make({
@@ -113,5 +131,3 @@ export const sleep: (
   })
   return yield* DurableDeferred.await(clock.deferred)
 })
-
-const defaultInMemoryThreshold = Duration.seconds(60)

packages/effect/test/unstable/persistence/PersistedQueueTest.ts

@@ -25,11 +25,16 @@ const advancePastTtl = Effect.gen(function*() {
   yield* Effect.sleep(1500).pipe(TestClock.withLive)
 })
 
+interface SuiteOptions {
+  readonly cleanup?: "supported" | "unsupported"
+}
+
 export const suiteWith = <R>(
   name: string,
   layer: Layer.Layer<PersistedQueue.PersistedQueueStore, unknown, R>,
   testApi: Vitest.MethodsNonLive<R>,
-  timeout: Duration.Input = "30 seconds"
+  timeout: Duration.Input = "30 seconds",
+  options: SuiteOptions = {}
 ) => {
   // Tests share and advance the same TestClock.
   const testOptions = { timeout: Duration.toMillis(timeout) }
@@ -242,82 +247,97 @@ export const suiteWith = <R>(
         yield* assertNotDelivered(fiber)
       }), testOptions)
 
-    it.effect("cleanup removes expired completed elements", () =>
-      Effect.gen(function*() {
-        const store = yield* PersistedQueue.PersistedQueueStore
-        const queue = yield* PersistedQueue.make({
-          name: "test-queue-cleanup",
-          schema: Item
-        })
-
-        yield* queue.offer({ n: 1n }, { id: "cleanup-id" })
-        yield* queue.take(Effect.succeed)
-
-        // within the ttl the dedupe entry survives, so re-offers are ignored
-        yield* store.cleanup({ timeToLive: Duration.days(30), failedTimeToLive: undefined })
-        yield* queue.offer({ n: 2n }, { id: "cleanup-id" })
-        const fiber = yield* queue.take(Effect.succeed).pipe(Effect.forkScoped)
-        yield* assertNotDelivered(fiber)
-
-        // after the ttl the completed element and its dedupe entry go away
-        yield* advancePastTtl
-        yield* store.cleanup({ timeToLive: Duration.seconds(1), failedTimeToLive: undefined })
-        yield* queue.offer({ n: 3n }, { id: "cleanup-id" })
-        yield* TestClock.adjust(1000)
-        assert.deepStrictEqual(yield* Fiber.join(fiber), { n: 3n })
-      }), testOptions)
-
-    it.effect("cleanup removes failed elements only with failedTimeToLive", () =>
-      Effect.gen(function*() {
-        const store = yield* PersistedQueue.PersistedQueueStore
-        const queue = yield* PersistedQueue.make({
-          name: "test-queue-cleanup-failed",
-          schema: Item,
-          maxAttempts: 1,
-          retrySchedule: Schedule.spaced(0)
-        })
+    if (options.cleanup === "unsupported") {
+      it.effect("cleanup explicitly fails as unsupported", () =>
+        Effect.gen(function*() {
+          const store = yield* PersistedQueue.PersistedQueueStore
+          for (const failedTimeToLive of [undefined, Duration.seconds(1)]) {
+            const error = yield* store.cleanup({
+              timeToLive: Duration.seconds(1),
+              failedTimeToLive
+            }).pipe(Effect.flip)
+            assert.instanceOf(error, PersistedQueue.PersistedQueueError)
+            assert.match(error.message, /unsupported|not supported/i)
+          }
+        }), testOptions)
+    } else {
+      it.effect("cleanup removes expired completed elements", () =>
+        Effect.gen(function*() {
+          const store = yield* PersistedQueue.PersistedQueueStore
+          const queue = yield* PersistedQueue.make({
+            name: "test-queue-cleanup",
+            schema: Item
+          })
 
-        yield* queue.offer({ n: 1n }, { id: "failed-cleanup-id" })
-        const error = yield* queue.take(() => Effect.fail("boom")).pipe(Effect.flip)
-        assert.strictEqual(error, "boom")
+          yield* queue.offer({ n: 1n }, { id: "cleanup-id" })
+          yield* queue.take(Effect.succeed)
 
-        // without failedTimeToLive the failed element is the dead-letter
-        // record and is kept, so its id stays deduplicated
-        yield* advancePastTtl
-        yield* store.cleanup({ timeToLive: Duration.seconds(1), failedTimeToLive: undefined })
-        yield* queue.offer({ n: 2n }, { id: "failed-cleanup-id" })
-        const fiber = yield* queue.take(Effect.succeed).pipe(Effect.forkScoped)
-        yield* assertNotDelivered(fiber)
+          // within the ttl the dedupe entry survives, so re-offers are ignored
+          yield* store.cleanup({ timeToLive: Duration.days(30), failedTimeToLive: undefined })
+          yield* queue.offer({ n: 2n }, { id: "cleanup-id" })
+          const fiber = yield* queue.take(Effect.succeed).pipe(Effect.forkScoped)
+          yield* assertNotDelivered(fiber)
 
-        // with failedTimeToLive the failed element and its dedupe entry go away
-        yield* store.cleanup({ timeToLive: Duration.days(30), failedTimeToLive: Duration.seconds(1) })
-        yield* queue.offer({ n: 3n }, { id: "failed-cleanup-id" })
-        yield* TestClock.adjust(1000)
-        assert.deepStrictEqual(yield* Fiber.join(fiber), { n: 3n })
-      }), testOptions)
+          // after the ttl the completed element and its dedupe entry go away
+          yield* advancePastTtl
+          yield* store.cleanup({ timeToLive: Duration.seconds(1), failedTimeToLive: undefined })
+          yield* queue.offer({ n: 3n }, { id: "cleanup-id" })
+          yield* TestClock.adjust(1000)
+          assert.deepStrictEqual(yield* Fiber.join(fiber), { n: 3n })
+        }), testOptions)
+
+      it.effect("cleanup removes failed elements only with failedTimeToLive", () =>
+        Effect.gen(function*() {
+          const store = yield* PersistedQueue.PersistedQueueStore
+          const queue = yield* PersistedQueue.make({
+            name: "test-queue-cleanup-failed",
+            schema: Item,
+            maxAttempts: 1,
+            retrySchedule: Schedule.spaced(0)
+          })
 
-    it.effect("cleanup keeps dedupe entries for unprocessed elements", () =>
-      Effect.gen(function*() {
-        const store = yield* PersistedQueue.PersistedQueueStore
-        const queue = yield* PersistedQueue.make({
-          name: "test-queue-cleanup-pending",
-          schema: Item
-        })
+          yield* queue.offer({ n: 1n }, { id: "failed-cleanup-id" })
+          const error = yield* queue.take(() => Effect.fail("boom")).pipe(Effect.flip)
+          assert.strictEqual(error, "boom")
+
+          // without failedTimeToLive the failed element is the dead-letter
+          // record and is kept, so its id stays deduplicated
+          yield* advancePastTtl
+          yield* store.cleanup({ timeToLive: Duration.seconds(1), failedTimeToLive: undefined })
+          yield* queue.offer({ n: 2n }, { id: "failed-cleanup-id" })
+          const fiber = yield* queue.take(Effect.succeed).pipe(Effect.forkScoped)
+          yield* assertNotDelivered(fiber)
+
+          // with failedTimeToLive the failed element and its dedupe entry go away
+          yield* store.cleanup({ timeToLive: Duration.days(30), failedTimeToLive: Duration.seconds(1) })
+          yield* queue.offer({ n: 3n }, { id: "failed-cleanup-id" })
+          yield* TestClock.adjust(1000)
+          assert.deepStrictEqual(yield* Fiber.join(fiber), { n: 3n })
+        }), testOptions)
+
+      it.effect("cleanup keeps dedupe entries for unprocessed elements", () =>
+        Effect.gen(function*() {
+          const store = yield* PersistedQueue.PersistedQueueStore
+          const queue = yield* PersistedQueue.make({
+            name: "test-queue-cleanup-pending",
+            schema: Item
+          })
 
-        yield* queue.offer({ n: 1n }, { id: "pending-cleanup-id" })
+          yield* queue.offer({ n: 1n }, { id: "pending-cleanup-id" })
 
-        // an element older than the ttl that was never processed keeps its
-        // dedupe entry, so the re-offer does not enqueue a duplicate
-        yield* advancePastTtl
-        yield* store.cleanup({ timeToLive: Duration.seconds(1), failedTimeToLive: undefined })
-        yield* queue.offer({ n: 2n }, { id: "pending-cleanup-id" })
+          // an element older than the ttl that was never processed keeps its
+          // dedupe entry, so the re-offer does not enqueue a duplicate
+          yield* advancePastTtl
+          yield* store.cleanup({ timeToLive: Duration.seconds(1), failedTimeToLive: undefined })
+          yield* queue.offer({ n: 2n }, { id: "pending-cleanup-id" })
 
-        const value = yield* queue.take(Effect.succeed)
-        assert.deepStrictEqual(value, { n: 1n })
+          const value = yield* queue.take(Effect.succeed)
+          assert.deepStrictEqual(value, { n: 1n })
 
-        const fiber = yield* queue.take(Effect.succeed).pipe(Effect.forkScoped)
-        yield* assertNotDelivered(fiber)
-      }), testOptions)
+          const fiber = yield* queue.take(Effect.succeed).pipe(Effect.forkScoped)
+          yield* assertNotDelivered(fiber)
+        }), testOptions)
+    }
 
     it.effect("processes concurrent elements exactly once with retries", () =>
       Effect.gen(function*() {
@@ -371,5 +391,8 @@ export const suiteWith = <R>(
   })
 }
 
-export const suite = (name: string, layer: Layer.Layer<PersistedQueue.PersistedQueueStore, unknown>) =>
-  suiteWith(name, layer, it)
+export const suite = (
+  name: string,
+  layer: Layer.Layer<PersistedQueue.PersistedQueueStore, unknown>,
+  options: SuiteOptions = {}
+) => suiteWith(name, layer, it, "30 seconds", options)

packages/platform/cloudflare/LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2023 Effectful Technologies Inc
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.