@effect/cluster: resume lost when DurableDeferred.done lands before the Suspended run reply is persisted — workflow stays Suspended forever

[IgnisDa]

Summary

With ClusterWorkflowEngine, if a DurableQueue worker's DurableDeferred.done arrives before the awaiting workflow's Suspended run reply has been persisted, the resume is silently dropped: the entity's deferred handler calls resume(), which looks up the run request's reply, finds no Suspended reply yet, and returns without doing anything. The deferred exit is persisted, but nothing ever wakes the workflow — it stays Suspended forever even though deferredResult for that deferred already returns an exit.

Reproducible with any trivially fast DurableQueue job; frequency scales with how quickly workers complete relative to the suspend persist.

Environment

• @effect/cluster 0.59.0
• @effect/workflow 0.18.2
• effect 3.21.4, @effect/sql-pg 0.52.1 (PostgreSQL)
• SingleRunner.layer({ runnerStorage: "sql" })

Root cause

dist/esm/ClusterWorkflowEngine.js — the deferred handler resumes via resume():

deferred: Effect.fnUntraced(function* (request) {
  yield* ensureSuccess(resume(workflow, executionId));
  return request.payload.exit;
}),

and resume() is a no-op unless a persisted Suspended reply for the run request already exists:

const resume = Effect.fnUntraced(function* (workflow, executionId) {
  const maybeReply = yield* requestReply({ ..., tag: "run", id: "" });
  const maybeSuspended = Option.filter(maybeReply, (reply) =>
    reply.exit._tag === "Success" && reply.exit.value._tag === "Suspended");
  if (Option.isNone(maybeSuspended)) return;   // ← resume dropped here
  yield* sharding.reset(Snowflake.Snowflake(maybeSuspended.value.requestId));
  yield* sharding.pollStorage;
});

Race sequence:

1. Workflow runs DurableQueue.process: offers the job, deferredResult finds nothing, the run handler returns Workflow.Suspended — the Suspended reply persist is now in flight in the RPC/sharding layer.
2. The worker completes (fast job) and DurableDeferred.done sends the deferred RPC to the workflow entity. The handler's resume() reads storage, sees no Suspended run reply yet, returns.
3. The deferred exit is persisted as the deferred request's reply (so a repeat done is deduped and won't help).
4. The Suspended run reply finishes persisting. No pending signal remains — nothing resets the run request, and the execution is stuck Suspended indefinitely.

An external engine.resume(workflow, executionId) issued later does un-stick it (the Suspended reply now exists, so resume() resets and replays; the replay's deferredResult finds the exit and completes immediately) — confirming the state is merely a lost wakeup, not corruption.

Reproduction sketch

const FastQueue = DurableQueue.make({
  name: "fast",
  payload: Schema.Struct({ n: Schema.Number }),
  idempotencyKey: ({ n }) => String(n),
  success: Schema.Void,
})
const FastWorker = DurableQueue.worker(FastQueue, () => Effect.void) // completes instantly

const AwaitFast = Workflow.make({
  name: "AwaitFast",
  payload: Schema.Struct({ n: Schema.Number }),
  idempotencyKey: ({ n }) => String(n),
})
const AwaitFastLayer = AwaitFast.toLayer(({ n }) => DurableQueue.process(FastQueue, { n }))

// fire many executions, then poll each until non-Suspended with a deadline:
yield* Effect.forEach(Array.range(0, 100), (n) => AwaitFast.execute({ n }, { discard: true }))

Some fraction of executions hit the window in step 2 and never leave Suspended; in our e2e suite (real Postgres, in-process worker) this reproduced regularly enough to make any test awaiting a fast queue job flaky.

Suggested fix directions

• In the deferred handler, don't treat "no Suspended reply yet" as terminal when the run request exists and is still unreplied — retry the lookup briefly, or record that a resume is owed and replay it once the Suspended reply lands.
• Alternatively, after the engine persists a Suspended run reply, check whether any deferred replies for that execution arrived meanwhile and self-resume.

Relation to #6294

Same failure family as #6294 (resume signals lost between the deferred/child-completion path and the suspended run), but a distinct code path: #6294 is the missing pollStorage in sendResumeParent's reset branch (delayed resume, self-heals on the storage poll timer); this one loses the wakeup entirely and never self-heals. Found while stabilizing the same test suite, together with the message_id overflow reported in #6317.

Workaround we're using

Result pollers go through a wrapper that "nudges" a resume when an execution stays Suspended across polls:

const result = yield* engine.poll(workflow, executionId)
if (result?._tag === "Suspended" && suspendedLongerThan(executionId, "2 seconds")) {
  yield* engine.resume(workflow, executionId)
}

This is safe for genuinely in-flight jobs — resume only resets an execution whose persisted reply is Suspended, replay is cheap (journaled activities, deduped queue offers), and a resolved deferred completes the replay immediately — and it reliably un-sticks the lost-wakeup cases.