feat(replay): replay-bypass side-effect guard — Node core (ADR-0027)

muhammetsafak · muhammetsafak · commit 406be2870c54 · 2026-06-21T07:04:03.000+03:00
A deliberate DLQ replay can tell its handler to skip external side-effects that
already fired (don't re-charge/re-email) while the idempotent core still runs &mdash;
mirroring the Go reference. New src/replay.ts (HEADER_REPLAY_BYPASS + isReplay +
bypassExternalEffects) + redrive bypass wiring. Rides the out-of-band HeaderCarrier
seam beside the frozen envelope (GR-1), trace_id preserved (GR-4). Opt-in. v1.5.0.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -9,7 +9,26 @@ The envelope wire format is versioned separately by `meta.schema_version`
 
 ## [Unreleased]
 
+## [1.5.0] - 2026-06-21
+
 ### Added
+- **Replay-bypass &mdash; an out-of-band side-effect guard for DLQ replay (ADR-0027).** A
+  deliberate `redrive` re-runs the handler and re-fires its external side-effects (a second
+  charge, a duplicate email); idempotency stops an *accidental* duplicate, not the *intended*
+  reprocess. This closes that gap.
+  - New `redrive(io, dlq, { bypass: true })` option stamps a `bq-replay-bypass` **transport
+    header** on each replayed message; the per-item result gains a `bypassed` flag. It takes
+    effect only when the `RedriveIO` implements the new optional `publishWithHeaders(queue,
+    body, headers)` capability &mdash; otherwise `bypass` is a no-op and the message is still
+    redriven (`bypassed: false`), exactly like the Go reference.
+  - New `isReplay(headers)` + `bypassExternalEffects(headers, effect)` consume-side guard
+    (plus the `HEADER_REPLAY_BYPASS` constant): a handler wraps its external, non-idempotent
+    side so a replay skips it while the idempotent core still runs.
+  - The marker rides **beside** the frozen envelope on the `HeaderCarrier` seam, never inside
+    it (`schema_version` stays **1**, GR-1; `trace_id` preserved, GR-4) &mdash; the same out-of-band
+    channel as the `traceparent` header. Fully opt-in and backward compatible: a header-less
+    message behaves exactly as before. Per-adapter transport wiring (the
+    `babelqueue-node-adapters` repo) is the documented follow-up, like ADR-0028's.
 - **OpenTelemetry v0.2 &mdash; W3C `traceparent` cross-hop span linkage (ADR-0028).** The
   `@babelqueue/core/otel` module now propagates the active span as a W3C `traceparent`
   (and `tracestate`) **transport header** so a consumer span is a true **child** of the
diff --git a/README.md b/README.md
@@ -101,6 +101,34 @@ const dlq = annotate(env, "failed", "orders", { attempts: 3, error: "boom" });
 `annotate` returns a copy &mdash; the original envelope is preserved unchanged inside
 the dead-lettered message, so any-language consumers can still read it.
 
+### Replay-bypass (optional)
+
+A deliberate replay off the DLQ ([`redrive`](src/redrive.ts)) re-runs the handler,
+re-firing its external side-effects &mdash; a second charge, a duplicate email. With
+`bypass`, `redrive` stamps a `bq-replay-bypass` **transport header** on each replayed
+message; a handler reads the delivered headers and skips the effects that already ran,
+while the idempotent core still runs (ADR-0027).
+
+```ts
+import { redrive, isReplay, bypassExternalEffects } from "@babelqueue/core";
+
+// PRODUCER: redrive with bypass (the IO must carry headers &mdash; publishWithHeaders).
+await redrive(io, "orders.dlq", { bypass: true });
+
+// CONSUMER: the adapter surfaces the delivered message's out-of-band headers.
+async function onMessage(env, headers) {
+  saveOrder(env);                                    // idempotent core &mdash; always runs
+  await bypassExternalEffects(headers, async () => { // skipped when isReplay(headers)
+    await sendConfirmationEmail(env);
+  });
+}
+```
+
+The marker rides **beside** the frozen envelope, never inside it (`schema_version`
+stays **1**, GR-1) &mdash; the same out-of-band `HeaderCarrier` seam as the `traceparent`
+header. It takes effect only when the `RedriveIO` implements `publishWithHeaders`;
+otherwise `bypass` is a no-op (`bypassed: false`) and the message is still redriven.
+
 ## What this core is (and isn't)
 
 It enforces the **contract**: the envelope shape, URN identity, trace propagation,
diff --git a/package-lock.json b/package-lock.json
diff --git a/package.json b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@babelqueue/core",
-  "version": "1.4.0",
+  "version": "1.5.0",
   "description": "Polyglot Queues, Simplified &mdash; the Node/TypeScript core: the canonical BabelQueue wire-envelope codec, contracts and dead-letter helpers.",
   "keywords": [
     "queue",
@@ -59,7 +59,7 @@
     "build": "tsup",
     "typecheck": "tsc --noEmit",
     "lint": "eslint src test",
-    "test": "node --import tsx --test test/codec.test.ts test/dead-letter.test.ts test/conformance.test.ts test/overhead.test.ts test/idempotency.test.ts test/schema.test.ts test/otel.test.ts test/redrive.test.ts",
+    "test": "node --import tsx --test test/codec.test.ts test/dead-letter.test.ts test/conformance.test.ts test/overhead.test.ts test/idempotency.test.ts test/schema.test.ts test/otel.test.ts test/redrive.test.ts test/replay.test.ts",
     "coverage": "c8 --check-coverage --lines 90 --functions 90 --branches 85 --reporter=text npm test",
     "prepublishOnly": "npm run build"
   },
diff --git a/src/index.ts b/src/index.ts
@@ -47,3 +47,5 @@ export type {
   RedriveOptions,
   RedriveResult,
 } from "./redrive.js";
+
+export { HEADER_REPLAY_BYPASS, isReplay, bypassExternalEffects } from "./replay.js";
diff --git a/src/redrive.ts b/src/redrive.ts
@@ -10,13 +10,19 @@
  * `attempts` reset to 0, while `job`, `trace_id`, `data` and `meta` are preserved verbatim, so
  * the replay is still fully traceable (same `trace_id`). The wire envelope is untouched (GR-1).
  *
- * Replay safety here is `dryRun` + `select` + redrive-to-`toQueue` (a sandbox). The
- * **Replay-Bypass** guard &mdash; a `bq-replay-bypass` transport header surfaced to handlers so a
- * replay can skip external side-effects &mdash; is a documented phase-two follow-up that touches the
- * runtime + every transport, like ADR-0025's `traceparent` follow-up.
+ * Replay safety here is `dryRun` + `select` + redrive-to-`toQueue` (a sandbox). On top of those,
+ * the **Replay-Bypass** guard (ADR-0027): with `opts.bypass`, a redriven message is stamped with
+ * the `bq-replay-bypass` transport header &mdash; surfaced to the handler via {@link isReplay} /
+ * {@link bypassExternalEffects} (see `./replay.js`) so a replay can skip external side-effects that
+ * already fired. It rides the out-of-band {@link HeaderCarrier}, never the frozen envelope (GR-1),
+ * and takes effect only when the {@link RedriveIO} can carry headers (its optional
+ * {@link RedriveIO.publishWithHeaders}); otherwise it is a no-op and the per-item `bypassed` flag
+ * stays `false`, exactly like the Go reference and the per-transport rollout of ADR-0028.
  */
 
 import { EnvelopeCodec, type Envelope } from "./codec.js";
+import type { HeaderCarrier } from "./contracts.js";
+import { HEADER_REPLAY_BYPASS } from "./replay.js";
 
 /** A message reserved from a queue, plus a way to acknowledge (remove) it. */
 export interface RedriveMessage {
@@ -30,6 +36,15 @@ export interface RedriveIO {
   pop(queue: string): Promise<RedriveMessage | null>;
   /** Append an already-encoded body to queue. */
   publish(queue: string, body: string): Promise<void>;
+  /**
+   * Optional capability &mdash; publish a body together with out-of-band transport `headers` (e.g. the
+   * `bq-replay-bypass` marker), for transports that carry per-message metadata on a native channel
+   * (AMQP headers, SQS `MessageAttributes`, a Redis transport-owned frame). The Node counterpart of
+   * Go's optional `HeaderPublisher`. A {@link RedriveIO} that omits it simply does not propagate
+   * headers &mdash; {@link redrive} falls back to plain {@link RedriveIO.publish} and `bypass` is a no-op
+   * (ADR-0027).
+   */
+  publishWithHeaders?(queue: string, body: string, headers: HeaderCarrier): Promise<void>;
 }
 
 /** Options for {@link redrive}. */
@@ -42,6 +57,13 @@ export interface RedriveOptions {
   dryRun?: boolean;
   /** Pick which messages to redrive (e.g. by reason or URN). Unselected messages are returned unchanged. */
   select?: (envelope: Envelope) => boolean;
+  /**
+   * Stamp the `bq-replay-bypass` transport header on each redriven message, so a handler can skip
+   * external side-effects that already ran (see `./replay.js`). Takes effect only when the
+   * {@link RedriveIO} implements {@link RedriveIO.publishWithHeaders}; otherwise it is a no-op and
+   * the per-item `bypassed` flag stays `false` (ADR-0027).
+   */
+  bypass?: boolean;
 }
 
 /** What happened to one message during a {@link redrive} run. */
@@ -55,6 +77,8 @@ export interface RedriveItem {
   to: string;
   /** True only when actually re-published to `to`. */
   redriven: boolean;
+  /** True when the `bq-replay-bypass` header was stamped on the redriven message (ADR-0027). */
+  bypassed: boolean;
 }
 
 /** Summary of a {@link redrive} run. */
@@ -120,7 +144,7 @@ export async function redrive(
       await io.publish(dlq, message.body); // restore the undecodable body; never drop it
       await message.ack();
       result.skipped++;
-      result.items.push({ messageId: "", traceId: "", urn: "", reason: "", from: dlq, to: "", redriven: false });
+      result.items.push({ messageId: "", traceId: "", urn: "", reason: "", from: dlq, to: "", redriven: false, bypassed: false });
       continue;
     }
 
@@ -132,6 +156,7 @@ export async function redrive(
       from: dlq,
       to: "",
       redriven: false,
+      bypassed: false,
     };
 
     if (opts.select && !opts.select(envelope)) {
@@ -153,18 +178,35 @@ export async function redrive(
       continue;
     }
 
+    const body = EnvelopeCodec.encode(resetForRedrive(envelope));
+    let bypassed: boolean;
     try {
-      await io.publish(target, EnvelopeCodec.encode(resetForRedrive(envelope)));
+      bypassed = await publishRedriven(io, target, body, opts.bypass ?? false);
     } catch (err) {
       await io.publish(dlq, message.body); // restore on a publish failure
       await message.ack();
       throw err;
     }
     await message.ack();
     item.redriven = true;
+    item.bypassed = bypassed;
     result.redriven++;
     result.items.push(item);
   }
 
   return result;
 }
+
+/**
+ * Re-publishes a reset `body` to `queue`. When `bypass` is set and the {@link RedriveIO} can carry
+ * headers ({@link RedriveIO.publishWithHeaders}), it stamps the `bq-replay-bypass` header and
+ * reports `true`; otherwise it publishes plainly and reports `false` (ADR-0027).
+ */
+async function publishRedriven(io: RedriveIO, queue: string, body: string, bypass: boolean): Promise<boolean> {
+  if (bypass && io.publishWithHeaders) {
+    await io.publishWithHeaders(queue, body, { [HEADER_REPLAY_BYPASS]: "1" });
+    return true;
+  }
+  await io.publish(queue, body);
+  return false;
+}
diff --git a/src/replay.ts b/src/replay.ts
@@ -0,0 +1,74 @@
+/**
+ * Replay-Bypass &mdash; a side-effect guard for DLQ replay (ADR-0027).
+ *
+ * A deliberate replay off the dead-letter queue ({@link redrive}) re-runs the handler, and the
+ * handler's external side-effects re-fire: a second charge, a duplicate email. {@link Wrap}
+ * (idempotency, ADR-0022) stops an *accidental* duplicate; it does not stop the *intended*
+ * reprocess from re-firing effects that already happened. This closes that gap.
+ *
+ * The marker that says "this is a replay, skip the external effects" rides **out of band** as the
+ * {@link HEADER_REPLAY_BYPASS} transport header on the {@link HeaderCarrier} &mdash; never in the frozen
+ * envelope (GR-1, `schema_version` stays 1). {@link redrive} (with `bypass`) stamps it when the
+ * transport can carry headers; a consume adapter, having reserved the message with its headers,
+ * passes them to the guard so the handler can skip effects that already fired:
+ *
+ * ```ts
+ * import { isReplay, bypassExternalEffects } from "@babelqueue/core";
+ *
+ * // CONSUMER: the adapter surfaces the delivered message's out-of-band headers.
+ * async function onMessage(env: Envelope, headers: HeaderCarrier): Promise<void> {
+ *   saveOrder(env);                                    // idempotent core &mdash; always runs
+ *   await bypassExternalEffects(headers, async () => { // external effect &mdash; skipped on replay
+ *     await sendConfirmationEmail(env);
+ *   });
+ * }
+ * ```
+ *
+ * The Node core is codec-only (no runtime / no transport), mirroring the Go reference
+ * `babelqueue-go/replay.go` and the Java `com.babelqueue.Replay`. Go/Java thread the flag through
+ * an ambient scope (a `context.Context` / a `ThreadLocal`); Node has no ambient request scope, so
+ * the delivered {@link HeaderCarrier} is passed explicitly &mdash; the same seam `otel.wrapHandler`
+ * already uses for delivered headers. A concrete broker transport carries the header once it
+ * implements the {@link RedriveHeaderIO} capability (the per-adapter rollout, like ADR-0028).
+ */
+
+import type { HeaderCarrier } from "./contracts.js";
+
+/**
+ * The out-of-band transport header {@link redrive} (with `bypass`) stamps on a replayed message,
+ * and that a consume adapter surfaces to the handler via {@link isReplay} /
+ * {@link bypassExternalEffects}. It rides on the {@link HeaderCarrier} beside the frozen envelope,
+ * never inside it (GR-1) &mdash; the same seam as the `traceparent` header (ADR-0028).
+ */
+export const HEADER_REPLAY_BYPASS = "bq-replay-bypass";
+
+/**
+ * Reports whether a delivered message was redriven with the replay-bypass marker &mdash; i.e. a
+ * deliberate replay whose external side-effects should be skipped. Reads the presence of
+ * {@link HEADER_REPLAY_BYPASS} on the message's out-of-band `headers` (nil/empty-safe: a
+ * header-less delivery is never a replay).
+ *
+ * @param headers the delivered message's out-of-band transport headers (may be null/undefined)
+ */
+export function isReplay(headers: HeaderCarrier | null | undefined): boolean {
+  return !!headers && headers[HEADER_REPLAY_BYPASS] !== undefined && headers[HEADER_REPLAY_BYPASS] !== "";
+}
+
+/**
+ * Runs `effect` unless the delivered message is a {@link isReplay replay}, in which case it is
+ * skipped and the returned promise resolves to `undefined`. Wrap the external, non-idempotent side
+ * of a handler &mdash; sending an email, charging a card, calling a third party &mdash; so a replay re-runs the
+ * idempotent core but does not re-fire effects that already happened.
+ *
+ * @param headers the delivered message's out-of-band transport headers (may be null/undefined)
+ * @param effect  the external side-effect to run only when this is not a replay
+ */
+export async function bypassExternalEffects<T>(
+  headers: HeaderCarrier | null | undefined,
+  effect: () => T | Promise<T>,
+): Promise<T | undefined> {
+  if (isReplay(headers)) {
+    return undefined;
+  }
+  return effect();
+}
diff --git a/test/conformance/manifest.json b/test/conformance/manifest.json
@@ -68,6 +68,80 @@
       "reason": "no 'job' or 'urn' &mdash; the message has no identity"
     }
   ],
+  "idempotency": {
+    "description": "Consumer idempotency conformance (ADR-0022). Broker-free: locks the seen-set dedupe contract every SDK's optional idempotency helper (Go idempotency.Wrap, PHP Idempotent::wrap, Python idempotency.wrap, Node Wrap, Java Idempotent.wrap) MUST honour over the frozen schema_version-1 envelope, independent of any transport. The helper does NOT touch the wire envelope &mdash; the cases above stay byte-identical; this block governs the consume-side decision (run the handler, or skip-and-ack) keyed on meta.id.",
+    "dedup_key": {
+      "description": "The dedupe key is meta.id verbatim &mdash; the canonical per-message identity (distinct from trace_id, which spans a causal chain). A consumer MUST key its seen-set on exactly fixtures/order-created.json's meta.id; two deliveries with the same meta.id are the SAME message (a redelivery), even byte-for-byte identical, and dedupe to one effect. trace_id is NOT the key: messages sharing a trace_id but with distinct meta.id are distinct effects.",
+      "envelope_file": "fixtures/order-created.json",
+      "key_field": "meta.id",
+      "expected_key": "f1e2d3c4-b5a6-4789-90ab-cdef01234567"
+    },
+    "sequences": {
+      "description": "Each sequence drives ONE handler wrapped by the SDK's idempotency helper, backed by ONE seen-set store, through an ordered list of deliveries. For each delivery the SDK MUST derive the effect: 'run' = the wrapped handler is invoked (the side-effect fires) and, on its outcome 'ok', the id is remembered; 'skip' = an already-remembered id is recognised, the handler is NOT invoked, and the delivery is acked (the runtime stops redelivering). 'outcome' is the handler's result for a 'run' delivery: 'ok' = returns/acks (remembered), 'throw' = raises/errors (the id is left UNMARKED so a later redelivery runs it again &mdash; retry/DLQ still apply). After replaying the whole sequence, total 'run'+'ok' invocations MUST equal expected_effects. This is at-least-once delivery collapsed to an exactly-once EFFECT by the consumer; it is seen-set post-success dedupe, NOT exactly-once delivery and NOT an in-flight concurrency lock.",
+      "cases": [
+        {
+          "name": "duplicate-delivery-runs-once",
+          "description": "The same meta.id delivered twice (a textbook at-least-once redelivery): first runs the effect, second is skipped. One effect.",
+          "deliveries": [
+            { "meta_id": "id-A", "expect_effect": "run",  "outcome": "ok" },
+            { "meta_id": "id-A", "expect_effect": "skip" }
+          ],
+          "expected_effects": 1
+        },
+        {
+          "name": "at-least-once-redelivery-is-noop",
+          "description": "Three redeliveries of one id (a flaky broker / worker-crash-before-ack storm): the effect fires once; every subsequent redelivery is a no-op skip.",
+          "deliveries": [
+            { "meta_id": "id-A", "expect_effect": "run",  "outcome": "ok" },
+            { "meta_id": "id-A", "expect_effect": "skip" },
+            { "meta_id": "id-A", "expect_effect": "skip" },
+            { "meta_id": "id-A", "expect_effect": "skip" }
+          ],
+          "expected_effects": 1
+        },
+        {
+          "name": "distinct-ids-each-run",
+          "description": "Different meta.ids are different messages &mdash; each runs once. Dedupe is per-id, never collapses unrelated messages.",
+          "deliveries": [
+            { "meta_id": "id-A", "expect_effect": "run", "outcome": "ok" },
+            { "meta_id": "id-B", "expect_effect": "run", "outcome": "ok" },
+            { "meta_id": "id-C", "expect_effect": "run", "outcome": "ok" }
+          ],
+          "expected_effects": 3
+        },
+        {
+          "name": "throw-leaves-id-unmarked",
+          "description": "A handler that throws does NOT remember the id, so the next redelivery runs it again (retry semantics survive the guard). The third delivery succeeds and is then deduped on a fourth.",
+          "deliveries": [
+            { "meta_id": "id-A", "expect_effect": "run",  "outcome": "throw" },
+            { "meta_id": "id-A", "expect_effect": "run",  "outcome": "throw" },
+            { "meta_id": "id-A", "expect_effect": "run",  "outcome": "ok" },
+            { "meta_id": "id-A", "expect_effect": "skip" }
+          ],
+          "expected_effects": 3
+        },
+        {
+          "name": "missing-id-fails-open",
+          "description": "A message with no usable meta.id (empty/absent) CANNOT be deduped, so the helper fails open: every delivery runs the handler. The guard never silently drops work it cannot identify.",
+          "deliveries": [
+            { "meta_id": "", "expect_effect": "run", "outcome": "ok" },
+            { "meta_id": "", "expect_effect": "run", "outcome": "ok" }
+          ],
+          "expected_effects": 2
+        },
+        {
+          "name": "forget-allows-rerun",
+          "description": "forget(id) evicts an id from the seen-set, so a later delivery of that id runs again (manual replay / TTL expiry). Locks the third Store method.",
+          "deliveries": [
+            { "meta_id": "id-A", "expect_effect": "run",  "outcome": "ok" },
+            { "meta_id": "id-A", "expect_effect": "skip" },
+            { "meta_id": "id-A", "forget_before": true, "expect_effect": "run", "outcome": "ok" }
+          ],
+          "expected_effects": 2
+        }
+      ]
+    }
+  },
   "sqs": {
     "description": "Amazon SQS binding conformance (broker-bindings.md &sect;3). Every SDK that ships an SQS transport must satisfy these. The envelope body stays byte-identical (the 'cases' above); these lock the native projection + reconciliation the binding adds. Per-message values reuse fixtures/order-created.json so the expected attributes are deterministic.",
     "attribute_projection": {
diff --git a/test/replay.test.ts b/test/replay.test.ts
