[core] Detect wedged waits instead of wake-looping on them forever

.changeset/wait-wedge-detection.md

@@ -0,0 +1,5 @@
+---
+'@workflow/core': patch
+---
+
+Detect wedged waits (a wait event write that conflicts while its event-log row is never readable) instead of silently wake-looping forever: warn within a tunable threshold (`WORKFLOW_WAIT_WEDGE_FAIL_AFTER_SECONDS`, default 10 minutes), then fail the run as `CORRUPTED_EVENT_LOG`.

docs/content/docs/v5/configuration/runtime-tuning.mdx

@@ -275,6 +275,13 @@ These variables are primarily for tests, debugging, or unusual deployments.
 - Default: `2`
 - Clock-skew tolerance for wait continuations that arrive near their target time.
 
+### `WORKFLOW_WAIT_WEDGE_FAIL_AFTER_SECONDS`
+
+- Default: `600` (10 minutes)
+- How long the runtime tolerates a wedged wait before failing the run as [`CORRUPTED_EVENT_LOG`](/docs/errors/corrupted-event-log).
+- A wait is wedged when the World rejects its `wait_created` or `wait_completed` event write as a duplicate while no such event can be read back from the event log — the entity state and the log disagree, which can happen if a backend commits the wait entity but loses the paired event row. Without this limit the run would silently wake-loop on the wait forever.
+- The tolerance is measured from the wait's target time for a wedged completion, and from the instant the wait was first scheduled (embedded in its replay-stable ID) for a wedged creation. Within the threshold the runtime logs a warning (and reports `workflow.wait.wedge_suspected` on the invocation span) but keeps retrying, so eventually consistent reads have time to converge; a wedged creation is additionally re-verified against a fresh event-log read before the run is failed. The generous default means read staleness cannot plausibly trigger a failure.
+
 ### `WORKFLOW_DEFERRED_CHECK_DELAY_MS`
 
 - Default: `100`

packages/core/src/runtime.ts

@@ -112,6 +112,10 @@ import { runStepSingleFlight } from './runtime/step-single-flight.js';
 import { handleSuspension } from './runtime/suspension-handler.js';
 import { useQuickJSVm } from './runtime/vm-mode.js';
 import { getWaitContinuationDispatch } from './runtime/wait-continuation.js';
+import {
+  classifyWaitWedgeObservation,
+  waitWedgeErrorMessage,
+} from './runtime/wait-wedge.js';
 import {
   getWorld,
   getWorldHandlers,
@@ -2768,6 +2772,17 @@ export function workflowEntrypoint(
                           },
                         }));
 
+                      // Completions the World refused as duplicates (409).
+                      // Usually a concurrent handler's write, whose row the
+                      // fetch below observes — but when no reload can produce
+                      // the row, the wait is wedged (entity committed, event
+                      // row lost) and this map feeds the detection after the
+                      // fetch. Keyed by correlationId, valued by the wait's
+                      // resumeAt for the stateless time-based escalation.
+                      const conflictedWaitCompletions = new Map<
+                        string,
+                        number
+                      >();
                       for (const waitEvent of waitsToComplete) {
                         try {
                           const created = await createEvent(waitEvent, {
@@ -2795,6 +2810,10 @@ export function workflowEntrypoint(
                                 correlationId: waitEvent.correlationId,
                               }
                             );
+                            conflictedWaitCompletions.set(
+                              waitEvent.correlationId,
+                              (waitEvent.eventData.resumeAt as Date).getTime()
+                            );
                             continue;
                           }
                           throw err;
@@ -2866,6 +2885,81 @@ export function workflowEntrypoint(
                         }
                       }
 
+                      // Wait-wedge detection. A completion the World refused
+                      // as a duplicate normally shows up in the fetch above (a
+                      // concurrent handler wrote it). When it does NOT — the
+                      // World attests the event exists while no read can
+                      // produce it — the wait is wedged between its entity
+                      // write and its event-log row, and without intervention
+                      // the run wake-loops on it forever (see
+                      // runtime/wait-wedge.ts). Warn while the contradiction
+                      // is young enough to be read staleness or an in-flight
+                      // insert; past the threshold, fail the run as
+                      // CORRUPTED_EVENT_LOG (the throw lands in the terminal
+                      // catch below, like the slot-gap check's). Skipped when
+                      // the run already recorded a terminal event — the check
+                      // right below consumes the delivery in that case.
+                      if (
+                        conflictedWaitCompletions.size > 0 &&
+                        !hasRecordedTerminalRunEvent(eventLog.events, runId)
+                      ) {
+                        const readableWaitCompletions = new Set(
+                          eventLog.events
+                            .filter((e) => e.eventType === 'wait_completed')
+                            .map((e) => e.correlationId)
+                        );
+                        let wedgesSuspected = 0;
+                        for (const [
+                          correlationId,
+                          resumeAtMs,
+                        ] of conflictedWaitCompletions) {
+                          if (readableWaitCompletions.has(correlationId)) {
+                            // The benign race: the conflicting write is in the
+                            // log. Nothing to do, and nothing to log.
+                            continue;
+                          }
+                          wedgesSuspected++;
+                          const observation = classifyWaitWedgeObservation({
+                            resumeAtMs,
+                            nowMs: now,
+                          });
+                          const message = waitWedgeErrorMessage({
+                            runId,
+                            correlationId,
+                            eventType: 'wait_completed',
+                            anchor: 'resumeAt',
+                            anchorMs: resumeAtMs,
+                            nowMs: now,
+                          });
+                          if (observation === 'fail') {
+                            span?.setAttributes(
+                              Attribute.WorkflowWaitWedgeSuspected(
+                                wedgesSuspected
+                              )
+                            );
+                            throw new CorruptedEventLogError(message);
+                          }
+                          runtimeLogger.warn(
+                            'Wait completion conflicted but no wait_completed row is readable; suspecting a wedged wait',
+                            {
+                              workflowRunId: runId,
+                              correlationId,
+                              resumeAt: new Date(resumeAtMs).toISOString(),
+                              secondsPastResume: Math.round(
+                                (now - resumeAtMs) / 1000
+                              ),
+                            }
+                          );
+                        }
+                        if (wedgesSuspected > 0) {
+                          span?.setAttributes(
+                            Attribute.WorkflowWaitWedgeSuspected(
+                              wedgesSuspected
+                            )
+                          );
+                        }
+                      }
+
                       // A replay reads the log as the complete record of what
                       // has happened, so a position nothing occupies is
                       // indistinguishable from an event that never occurred and
@@ -3154,7 +3248,16 @@ export function workflowEntrypoint(
                             // hand it back to the queue to spin again.
                             throw escalation.error;
                           }
-                          if (!FatalError.is(suspensionError)) {
+                          if (
+                            !FatalError.is(suspensionError) &&
+                            // A wedged wait detected while committing this
+                            // suspension's wait_created (see
+                            // runtime/wait-wedge.ts): redelivery would replay
+                            // into the same conflict forever, so it takes the
+                            // terminal path below and fails the run as
+                            // CORRUPTED_EVENT_LOG.
+                            !CorruptedEventLogError.is(suspensionError)
+                          ) {
                             // Transient failures propagate to the queue
                             // handler so the message is redelivered.
                             throw suspensionError;

packages/core/src/runtime/suspension-handler.ts

@@ -1,5 +1,6 @@
 import type { Span } from '@opentelemetry/api';
 import {
+  CorruptedEventLogError,
   EntityConflictError,
   FatalError,
   HookNotFoundError,
@@ -49,6 +50,12 @@ import {
   stepDispatchIdempotencyKey,
 } from './helpers.js';
 import { ReplayRecoveryReporter } from './replay-recovery-reporter.js';
+import {
+  decodeWaitScheduledAtMs,
+  getWaitWedgeFailAfterSeconds,
+  isWaitCreatedRowReadable,
+  waitWedgeErrorMessage,
+} from './wait-wedge.js';
 
 export interface SuspensionHandlerParams {
   suspension: WorkflowSuspension;
@@ -886,6 +893,64 @@ export async function handleSuspension({
             await createGuarded(waitEvent, { requestId });
           } catch (err) {
             if (EntityConflictError.is(err)) {
+              // This create only runs because the replayed log has no
+              // wait_created row for this wait (hasCreatedEvent was false),
+              // so a conflict means the wait entity exists while its row was
+              // not readable. Usually that is the concurrent-suspension race
+              // (another handler's write landed after this replay's
+              // snapshot) and stays silent, as always. But a wait whose
+              // entity committed WITHOUT its event row conflicts here on
+              // every replay forever — the elapsed-wait pass can only
+              // complete waits whose wait_created row it can see, so the run
+              // wake-loops on it. The wait's correlation id is replay-stable
+              // (seeded RNG + replay clock), so the ULID timestamp inside it
+              // is the one anchor the loop cannot reset: once the same wait
+              // id has been conflicting for longer than the threshold,
+              // verify with a fresh log read and fail the run as
+              // CORRUPTED_EVENT_LOG rather than loop (see
+              // runtime/wait-wedge.ts; the caller routes this error to its
+              // terminal path instead of redelivering). resumeAt cannot be
+              // the anchor here: an uncreated wait recomputes it from the
+              // live clock on every replay.
+              const nowMs = Date.now();
+              const scheduledAtMs = decodeWaitScheduledAtMs(
+                queueItem.correlationId
+              );
+              const thresholdMs = getWaitWedgeFailAfterSeconds() * 1000;
+              const suspectWedge =
+                scheduledAtMs !== undefined &&
+                nowMs - scheduledAtMs > thresholdMs &&
+                !(await isWaitCreatedRowReadable(
+                  world,
+                  runId,
+                  queueItem.correlationId
+                ));
+              if (suspectWedge) {
+                span?.setAttributes(Attribute.WorkflowWaitWedgeSuspected(1));
+                runtimeLogger.error(
+                  'Wait creation has been conflicting past the wedge threshold and no wait_created row is readable; failing the run',
+                  {
+                    workflowRunId: runId,
+                    correlationId: queueItem.correlationId,
+                    scheduledAt: new Date(scheduledAtMs).toISOString(),
+                    secondsSinceScheduled: Math.round(
+                      (nowMs - scheduledAtMs) / 1000
+                    ),
+                    message: err.message,
+                  }
+                );
+                throw new CorruptedEventLogError(
+                  waitWedgeErrorMessage({
+                    runId,
+                    correlationId: queueItem.correlationId,
+                    eventType: 'wait_created',
+                    anchor: 'scheduledAt',
+                    anchorMs: scheduledAtMs,
+                    nowMs,
+                  }),
+                  { cause: err }
+                );
+              }
               runtimeLogger.info('Wait already exists, continuing', {
                 workflowRunId: runId,
                 correlationId: queueItem.correlationId,