fix(platform-cloudflare): reject ask deduplication to tells

Author: tim-smart

packages/platform/cloudflare/src/CloudflareCluster.ts

@@ -125,7 +125,7 @@ interface EntityStub {
   }) => Promise<{
     readonly requestId: string
     readonly replies: ReadonlyArray<string>
-    readonly error?: "MailboxFull" | "EncodedMessageTooLarge" | undefined
+    readonly error?: "MailboxFull" | "EncodedMessageTooLarge" | "AskDeduplicatedToTell" | undefined
   }>
   readonly acknowledge: (requestId: string, replyId: string) => Promise<ReadonlyArray<string>>
   readonly interrupt?: (storageRequestId: string, clientRequestId?: string) => Promise<void>
@@ -316,6 +316,12 @@ const make = Effect.fnUntraced(function*(options: LayerOptions) {
                         | MailboxFull
                         | PersistenceError
                     )
+                  } else if (result.error === "AskDeduplicatedToTell") {
+                    return Effect.fail(
+                      new PersistenceError({
+                        cause: new Error("Cannot deduplicate an ask onto a tell with the same PrimaryKey")
+                      }) as MailboxFull | PersistenceError
+                    )
                   }
                   entry.storageRequestId = result.requestId
                   if (replyHandler !== undefined && result.requestId !== clientRequestId) {

packages/platform/cloudflare/src/CloudflareDurableObjects.ts

@@ -80,7 +80,7 @@ interface ReplayMessage {
 interface InvokeResult {
   readonly requestId: string
   readonly replies: ReadonlyArray<string>
-  readonly error?: "MailboxFull" | "EncodedMessageTooLarge" | undefined
+  readonly error?: "MailboxFull" | "EncodedMessageTooLarge" | "AskDeduplicatedToTell" | undefined
 }
 
 interface InvokeOutcome {
@@ -214,6 +214,17 @@ export class ClusterEntity extends DurableObject<unknown> {
         return yield* Effect.die(error)
       }
       if (persisted._tag === "Duplicate") {
+        const original = loadMessage(storage.sql, persisted.originalId)
+        if (original === undefined) return yield* Effect.die("Duplicate mailbox row disappeared")
+        if (original.discard && !discard) {
+          return {
+            result: {
+              requestId: persisted.originalId,
+              replies: [],
+              error: "AskDeduplicatedToTell" as const
+            }
+          }
+        }
         const nextReply = loadNextReply(storage.sql, persisted.originalId)
         if (nextReply !== undefined) {
           this.#releaseTerminalSession(persisted.originalId, nextReply)
@@ -231,8 +242,6 @@ export class ClusterEntity extends DurableObject<unknown> {
             String(envelope.requestId)
           )
         }
-        const original = loadMessage(storage.sql, persisted.originalId)
-        if (original === undefined) return yield* Effect.die("Duplicate mailbox row disappeared")
         if (original.deliverAt !== undefined && original.deliverAt > Date.now()) {
           yield* this.#armEarliestAlarm()
           return this.#delayedOutcome(

packages/platform/cloudflare/src/internal/entityMailbox.ts

@@ -53,7 +53,7 @@ export const persistRequest = (
   }
 
   const existing = sql.exec(
-    `SELECT m.request_id, m.processed, m.reply_to, r.reply AS last_reply
+    `SELECT m.request_id, m.discard, m.processed, m.reply_to, r.reply AS last_reply
      FROM cluster_messages m
      LEFT JOIN cluster_replies r ON r.reply_id = m.last_reply_id
      WHERE m.request_id = ? OR (? IS NOT NULL AND m.message_id = ?)
@@ -63,7 +63,7 @@ export const persistRequest = (
     primaryKey
   ).toArray()[0]
   if (existing !== undefined) {
-    if (replyTo !== null && Number(existing.processed) === 0) {
+    if (replyTo !== null && Number(existing.discard) === 0 && Number(existing.processed) === 0) {
       const replyTos = decodeReplyTargets(existing.reply_to)
       if (!replyTos.includes(replyTo)) replyTos.push(replyTo)
       sql.exec(

packages/platform/cloudflare/test/CloudflareCluster.test.ts

@@ -430,6 +430,38 @@ describe("CloudflareCluster", () => {
       }).pipe(Effect.provide(CloudflareCluster.layer(options)))
     })
 
+    it.effect("surfaces ask-to-tell deduplication as a persistence failure", () => {
+      const stub = {
+        invoke() {
+          return Promise.resolve({
+            requestId: "original-tell",
+            replies: [],
+            error: "AskDeduplicatedToTell" as const
+          })
+        },
+        acknowledge() {
+          return Promise.resolve([])
+        }
+      }
+      const options: CloudflareCluster.LayerOptions = {
+        entities: [Scheduled],
+        entityNamespace: new FakeNamespace(stub) as any,
+        workflowNamespace: new FakeNamespace() as any,
+        queueNamespace: new FakeNamespace() as any,
+        singletonNamespace: new FakeNamespace() as any
+      }
+
+      return Effect.gen(function*() {
+        const makeClient = yield* Scheduled.client
+        const exit = yield* makeClient("one").Ask({ deliverAt: Date.now() + 60_000, id: "tell" }).pipe(
+          Effect.provideService(CurrentEntityName, "6:Callerone"),
+          Effect.exit
+        )
+        assert.isTrue(Exit.isFailure(exit))
+        assert.isFalse(yield* Effect.promise(() => deliverReply("original-tell", "unused")))
+      }).pipe(Effect.provide(CloudflareCluster.layer(options)))
+    })
+
     it.effect("does not retain reset targets for volatile requests", () => {
       let requestId = ""
       let resets = 0

packages/platform/cloudflare/test/CloudflareDurableObjects.test.ts

@@ -219,25 +219,54 @@ describe("CloudflareDurableObjects", () => {
       assert.deepStrictEqual(result, { firstStatus: "pending", secondStatus: "rejected" })
     }), 60_000)
 
-  it.effect("does not run a future row when an immediate request deduplicates to it", () =>
+  it.effect(
+    "rejects an ask deduplicated onto a future tell without hanging or running it early",
+    () =>
+      Effect.gen(function*() {
+        const miniflare = yield* makeMiniflare
+        const fetchJson = (path: string) =>
+          Effect.promise(() =>
+            Promise.race([
+              miniflare.dispatchFetch(`http://placeholder${path}`).then(async (response) => {
+                const body = await response.text()
+                assert.strictEqual(response.status, 200, body)
+                return JSON.parse(body)
+              }),
+              new Promise<never>((_, reject) => setTimeout(() => reject(new Error(`${path} did not return`)), 2_000))
+            ])
+          )
+        const id = "scheduled-dedup"
+        yield* fetchJson(
+          `/delayed?id=${id}&operationId=same&discard=true&deliverAt=${Date.now() + 60_000}`
+        )
+        const duplicate = yield* fetchJson(`/mailbox?id=${id}&tag=Add&operationId=same&discard=false`)
+        assert.strictEqual(duplicate.error, "AskDeduplicatedToTell")
+
+        const result = yield* fetchJson(`/mailbox?id=${id}&tag=Get`)
+        assert.deepStrictEqual(JSON.parse(result.replies[0]).exit, { _tag: "Success", value: 0 })
+      }),
+    60_000
+  )
+
+  it.effect("rejects an ask deduplicated onto a processed tell without hanging", () =>
     Effect.gen(function*() {
       const miniflare = yield* makeMiniflare
       const fetchJson = (path: string) =>
         Effect.promise(() =>
-          miniflare.dispatchFetch(`http://placeholder${path}`).then(async (response) => {
-            const body = await response.text()
-            assert.strictEqual(response.status, 200, body)
-            return JSON.parse(body)
-          })
+          Promise.race([
+            miniflare.dispatchFetch(`http://placeholder${path}`).then(async (response) => {
+              const body = await response.text()
+              assert.strictEqual(response.status, 200, body)
+              return JSON.parse(body)
+            }),
+            new Promise<never>((_, reject) => setTimeout(() => reject(new Error(`${path} did not return`)), 2_000))
+          ])
         )
-      const id = "scheduled-dedup"
-      yield* fetchJson(
-        `/delayed?id=${id}&operationId=same&discard=true&deliverAt=${Date.now() + 60_000}`
-      )
+      const id = "processed-tell-dedup"
       yield* fetchJson(`/mailbox?id=${id}&tag=Add&operationId=same`)
+      const duplicate = yield* fetchJson(`/mailbox?id=${id}&tag=Add&operationId=same&discard=false`)
 
-      const result = yield* fetchJson(`/mailbox?id=${id}&tag=Get`)
-      assert.deepStrictEqual(JSON.parse(result.replies[0]).exit, { _tag: "Success", value: 0 })
+      assert.strictEqual(duplicate.error, "AskDeduplicatedToTell")
     }), 60_000)
 
   it.effect("delivers a scheduled ask reply to the caller Durable Object", () =>

packages/platform/cloudflare/test/fixtures/worker.ts

@@ -204,6 +204,8 @@ export default {
       const tag = url.searchParams.get("tag") ?? "Get"
       const operationId = url.searchParams.get("operationId") ?? "operation"
       const requestId = crypto.randomUUID()
+      const discardParam = url.searchParams.get("discard")
+      const discard = discardParam === null ? tag === "Add" || tag === "AddVolatile" : discardParam === "true"
       try {
         const result = await stub.invoke(
           JSON.stringify({
@@ -218,7 +220,7 @@ export default {
             payload: tag === "Get" || tag === "Watch" ? null : { operationId },
             headers: {}
           }),
-          tag === "Add" || tag === "AddVolatile"
+          discard
         )
         return Response.json(result)
       } catch (error) {