Decode replay payloads synchronously

Author: NathanColosimo

.changeset/decode-replay-payloads-synchronously.md

@@ -0,0 +1,5 @@
+---
+"@workflow/core": patch
+---
+
+Decode replay payloads synchronously with Node AES-GCM and zstd when the runtime supports it.

packages/core/src/encryption.test.ts

@@ -1,6 +1,12 @@
 import { RuntimeDecryptionError } from '@workflow/errors';
 import { describe, expect, it } from 'vitest';
-import { type CryptoKey, decrypt, encrypt, importKey } from './encryption.js';
+import {
+  type CryptoKey,
+  decrypt,
+  decryptSync,
+  encrypt,
+  importKey,
+} from './encryption.js';
 
 const RAW_KEY = new Uint8Array(32).fill(7);
 const OTHER_RAW_KEY = new Uint8Array(32).fill(8);
@@ -27,6 +33,16 @@ describe('encryption', () => {
       const decoded = await decrypt(key, ciphertext);
       expect(new TextDecoder().decode(decoded)).toBe('hello, workflow');
     });
+
+    it('decryptSync() returns the plaintext without a promise', async () => {
+      const key = await getKey();
+      const plaintext = new TextEncoder().encode('synchronous replay');
+      const ciphertext = await encrypt(key, plaintext);
+
+      const decoded = decryptSync(key, ciphertext);
+      expect(decoded).toBeInstanceOf(Uint8Array);
+      expect(new TextDecoder().decode(decoded)).toBe('synchronous replay');
+    });
   });
 
   describe('importKey', () => {
@@ -92,6 +108,19 @@ describe('encryption', () => {
       expect(cause?.name).toBe('OperationError');
     });
 
+    it('keeps RuntimeDecryptionError on synchronous auth failure', async () => {
+      const key = await getKey();
+      const ciphertext = await encrypt(
+        key,
+        new TextEncoder().encode('tamper me')
+      );
+      ciphertext[ciphertext.length - 1] ^= 0xff;
+
+      expect(() => decryptSync(key, ciphertext)).toThrowError(
+        RuntimeDecryptionError
+      );
+    });
+
     it('does not record a formatPrefix at the low-level layer', async () => {
       // This function only ever sees the stripped AES payload
       // (`[nonce][ciphertext+tag]`), never the outer `encr` envelope marker.

packages/core/src/encryption.ts

@@ -23,6 +23,46 @@ import { RuntimeDecryptionError, WorkflowRuntimeError } from '@workflow/errors';
 // so consumers can reference it without adding `dom` lib.
 export type CryptoKey = import('node:crypto').webcrypto.CryptoKey;
 
+/**
+ * Raw key material retained alongside keys imported by this module.
+ *
+ * Node's synchronous cipher API cannot consume a Web Crypto `CryptoKey`, and
+ * our keys are deliberately non-extractable. Keeping the original bytes in a
+ * WeakMap gives the Node replay path access to the same key without making it
+ * extractable or extending its lifetime beyond the `CryptoKey`. Browser/edge
+ * callers continue to use Web Crypto and never consult this map.
+ */
+const importedKeyMaterial = new WeakMap<CryptoKey, Uint8Array>();
+
+interface NodeDecipher {
+  setAAD(data: Uint8Array): NodeDecipher;
+  setAuthTag(tag: Uint8Array): NodeDecipher;
+  update(data: Uint8Array): Uint8Array;
+  final(): Uint8Array;
+}
+
+interface NodeCrypto {
+  createDecipheriv(
+    algorithm: string,
+    key: Uint8Array,
+    iv: Uint8Array,
+    options?: { authTagLength?: number }
+  ): NodeDecipher;
+}
+
+/** Resolve node:crypto without a static import, preserving browser bundles. */
+function getNodeCrypto(): NodeCrypto | undefined {
+  try {
+    return (
+      globalThis as {
+        process?: { getBuiltinModule?: (id: string) => NodeCrypto };
+      }
+    ).process?.getBuiltinModule?.('node:crypto');
+  } catch {
+    return undefined;
+  }
+}
+
 /** AES-GCM nonce length in bytes. */
 export const NONCE_LENGTH = 12;
 /** AES-GCM authentication tag length in bits. */
@@ -55,7 +95,7 @@ export async function importKey(
       `Encryption key must be exactly ${KEY_LENGTH} bytes, got ${raw.byteLength}`
     );
   }
-  return globalThis.crypto.subtle.importKey(
+  const key = await globalThis.crypto.subtle.importKey(
     'raw',
     raw,
     'AES-GCM',
@@ -65,6 +105,82 @@ export async function importKey(
     // a strict subset of `KeyUsage[]`, so this cast is sound.
     usages as ('encrypt' | 'decrypt')[]
   );
+  // Copy the caller's bytes: a caller may reuse/mutate its input buffer after
+  // importKey(), while a CryptoKey's material is immutable.
+  importedKeyMaterial.set(key, raw.slice());
+  return key;
+}
+
+/**
+ * Decrypt AES-256-GCM synchronously when running on Node and the key was
+ * imported by this module.
+ *
+ * Returns `undefined` when the portable Web Crypto fallback is required (for
+ * example in a browser, or for an externally-created CryptoKey). Authentication
+ * failures throw the same RuntimeDecryptionError shape as {@link decrypt}.
+ */
+export function decryptSync(
+  key: CryptoKey,
+  data: Uint8Array,
+  aad?: Uint8Array
+): Uint8Array | undefined {
+  const material = importedKeyMaterial.get(key);
+  const nodeCrypto = getNodeCrypto();
+  if (!material || !nodeCrypto) return undefined;
+  if (!key.usages.includes('decrypt')) {
+    throw new RuntimeDecryptionError(
+      'AES-256-GCM decryption failed: CryptoKey does not support decrypt',
+      {
+        context: { operation: 'decrypt', byteLength: data.byteLength },
+      }
+    );
+  }
+
+  const minLength = NONCE_LENGTH + TAG_BYTES;
+  if (data.byteLength < minLength) {
+    throw new RuntimeDecryptionError(
+      `Encrypted data too short: expected at least ${minLength} bytes, got ${data.byteLength}`,
+      {
+        context: {
+          operation: 'decrypt',
+          byteLength: data.byteLength,
+        },
+      }
+    );
+  }
+
+  const nonce = data.subarray(0, NONCE_LENGTH);
+  const ciphertextEnd = data.byteLength - TAG_BYTES;
+  const ciphertext = data.subarray(NONCE_LENGTH, ciphertextEnd);
+  const authTag = data.subarray(ciphertextEnd);
+  try {
+    const decipher = nodeCrypto.createDecipheriv(
+      'aes-256-gcm',
+      material,
+      nonce,
+      { authTagLength: TAG_BYTES }
+    );
+    if (aad) decipher.setAAD(aad);
+    decipher.setAuthTag(authTag);
+    const head = decipher.update(ciphertext);
+    const tail = decipher.final();
+    if (tail.byteLength === 0) return head;
+    const plaintext = new Uint8Array(head.byteLength + tail.byteLength);
+    plaintext.set(head, 0);
+    plaintext.set(tail, head.byteLength);
+    return plaintext;
+  } catch (cause) {
+    throw new RuntimeDecryptionError(
+      `AES-256-GCM decryption failed: ${cause instanceof Error ? cause.message : String(cause)}`,
+      {
+        cause,
+        context: {
+          operation: 'decrypt',
+          byteLength: data.byteLength,
+        },
+      }
+    );
+  }
 }
 
 /**

packages/core/src/serialization.ts

@@ -38,10 +38,12 @@ import {
   type CompressionStats,
   compress,
   decompress,
+  decompressReplayPayload,
 } from './serialization/compression.js';
 import {
   aesKeyOf,
   decrypt,
+  decryptReplayPayload,
   deriveRunPayloadKeys,
   type EncryptionKeyParam,
   encrypt,
@@ -3386,8 +3388,8 @@ export interface PreparedReplayPayload {
 
 /**
  * Swappable implementation of the host-side preparation stage. Supporting
- * both direct and promised results lets a future synchronous Node decryptor use
- * the same cache contract as today's asynchronous Web Crypto implementation.
+ * both direct and promised results covers synchronous Node AES/zstd as well as
+ * the portable Web Crypto and sealed-envelope fallbacks.
  */
 export type ReplayPayloadPreparer = (
   value: unknown,
@@ -3398,18 +3400,49 @@ export type ReplayPayloadPreparer = (
  * Decrypt and decompress persisted data without parsing it into JavaScript.
  * Legacy non-binary values pass through unchanged for their consumer to revive.
  */
-export const prepareReplayPayload: ReplayPayloadPreparer = async (
-  value,
-  key
-) => {
+function prepareReplayPayloadWithStats(
+  value: unknown,
+  key: PayloadKey | undefined,
+  compressionStats?: CompressionStats
+): PreparedReplayPayload | Promise<PreparedReplayPayload> {
+  const finish = (prepared: unknown): PreparedReplayPayload => ({
+    data: prepared,
+  });
+  const decompressPrepared = (
+    decrypted: unknown
+  ): PreparedReplayPayload | Promise<PreparedReplayPayload> => {
+    const prepared = decompressReplayPayload(decrypted, compressionStats);
+    return prepared instanceof Promise
+      ? prepared.then(finish)
+      : finish(prepared);
+  };
+
+  const decrypted = decryptReplayPayload(value, key);
+  return decrypted instanceof Promise
+    ? decrypted.then(decompressPrepared)
+    : decompressPrepared(decrypted);
+}
+
+// Replay preparation is event-at-a-time and may run inside the response
+// decoder. Per-payload compression attributes would add a detached O(N)
+// microtask tail and repeatedly overwrite one span, so the replay fast path
+// records only its aggregate preparation span.
+export const prepareReplayPayload: ReplayPayloadPreparer = (value, key) =>
+  prepareReplayPayloadWithStats(value, key);
+
+async function prepareReplayPayloadWithTelemetry(
+  value: unknown,
+  key: PayloadKey | undefined
+): Promise<PreparedReplayPayload> {
   const compressionStats: CompressionStats = {};
-  const prepared = await decompress(
-    await decrypt(value, key),
+  const prepared = await prepareReplayPayloadWithStats(
+    value,
+    key,
     compressionStats
   );
   await recordCompression(compressionStats, 'deserialize');
-  return { data: prepared };
-};
+  return prepared;
+}
 
 /**
  * Parse a prepared workflow argument or successful step/hook payload using the
@@ -3536,7 +3569,7 @@ export async function hydrateWorkflowArguments(
   prepared?: PreparedReplayPayload
 ): Promise<any> {
   return deserializePreparedReplayPayload(
-    prepared ?? (await prepareReplayPayload(value, key)),
+    prepared ?? (await prepareReplayPayloadWithTelemetry(value, key)),
     global,
     extraRevivers
   );
@@ -3833,7 +3866,7 @@ export async function hydrateStepError(
   prepared?: PreparedReplayPayload
 ): Promise<unknown> {
   return deserializePreparedStepError(
-    prepared ?? (await prepareReplayPayload(value, key)),
+    prepared ?? (await prepareReplayPayloadWithTelemetry(value, key)),
     global,
     extraRevivers
   );
@@ -3958,7 +3991,7 @@ export async function hydrateStepReturnValue(
   prepared?: PreparedReplayPayload
 ): Promise<any> {
   return deserializePreparedReplayPayload(
-    prepared ?? (await prepareReplayPayload(value, key)),
+    prepared ?? (await prepareReplayPayloadWithTelemetry(value, key)),
     global,
     extraRevivers
   );

packages/core/src/serialization/compression.ts

@@ -309,10 +309,10 @@ export async function compress(
  * Non-compressed data (including non-binary legacy data) is returned
  * unchanged, so this is safe to apply unconditionally on read paths.
  */
-export async function decompress(
+export function decompressReplayPayload(
   data: Uint8Array | unknown,
   stats?: CompressionStats
-): Promise<Uint8Array | unknown> {
+): Uint8Array | unknown | Promise<Uint8Array> {
   if (!(data instanceof Uint8Array)) return data;
   const prefix = peekFormatPrefix(data);
 
@@ -332,15 +332,24 @@ export async function decompress(
       );
     }
     const { payload } = decodeFormatPrefix(data);
-    const inflated = await gunzipBytes(payload);
-    recordStats(stats, 'gzip', inflated.length, data.length);
-    return inflated;
+    return gunzipBytes(payload).then((inflated) => {
+      recordStats(stats, 'gzip', inflated.length, data.length);
+      return inflated;
+    });
   }
 
   recordStats(stats, 'none', data.length, data.length);
   return data;
 }
 
+/** Portable always-Promise facade retained for existing callers. */
+export async function decompress(
+  data: Uint8Array | unknown,
+  stats?: CompressionStats
+): Promise<Uint8Array | unknown> {
+  return decompressReplayPayload(data, stats);
+}
+
 /**
  * Check if data is compressed (has a 'zstd' or 'gzip' format prefix).
  */