Introduce JournalValueCodec API (#581)

* JournalValueCodec

* Introduce new interface JournalValueCodec to allow customize encoding of values after serialization and before deserialization. Expose it as experimental.
* Disable non-deterministic checks when the user provides a codec
* Use the codec in handler input/output, attach, get state, set state, call, send, run, awakeable, complete awakeable, get promise, peek promise, complete promise

* Add JournalValueCodec support to ingress client

---------

Co-authored-by: igalshilman <[email protected]>

Author: slinkydeveloper

packages/restate-sdk-clients/src/api.ts

@@ -7,6 +7,7 @@ import type {
   WorkflowDefinitionFrom,
   Serde,
   Duration,
+  JournalValueCodec,
 } from "@restatedev/restate-sdk-core";
 import { millisOrDurationToMillis } from "@restatedev/restate-sdk-core";
 
@@ -322,4 +323,11 @@ export type ConnectionOpts = {
    * Use this to attach authentication headers.
    */
   headers?: Record<string, string>;
+
+  /**
+   * Codec to use for input/outputs. Check {@link JournalValueCodec} for more details
+   *
+   * @experimental
+   */
+  journalValueCodec?: JournalValueCodec;
 };

packages/restate-sdk-clients/src/ingress.ts

@@ -18,6 +18,7 @@ import {
   type VirtualObjectDefinitionFrom,
   type Serde,
   serde,
+  type JournalValueCodec,
 } from "@restatedev/restate-sdk-core";
 import type {
   ConnectionOpts,
@@ -147,7 +148,8 @@ const doComponentInvocation = async <I, O>(
 
   const { body, contentType } = serializeBodyWithContentType(
     params.parameter,
-    inputSerde
+    inputSerde,
+    opts.journalValueCodec
   );
   //
   // headers
@@ -201,12 +203,15 @@ const doComponentInvocation = async <I, O>(
       `Request failed: ${httpResponse.status}\n${body}`
     );
   }
-  const responseBuf = await httpResponse.arrayBuffer();
+  const responseBuf = new Uint8Array(await httpResponse.arrayBuffer());
   if (!params.send) {
+    const decodedBuf = opts.journalValueCodec
+      ? await opts.journalValueCodec.decode(responseBuf)
+      : responseBuf;
     const outputSerde = params.opts?.opts.output ?? serde.json;
-    return outputSerde.deserialize(new Uint8Array(responseBuf)) as O;
+    return outputSerde.deserialize(decodedBuf) as O;
   }
-  const json = serde.json.deserialize(new Uint8Array(responseBuf)) as O;
+  const json = serde.json.deserialize(responseBuf) as O;
   return { ...json, attachable };
 };
 
@@ -252,8 +257,11 @@ const doWorkflowHandleCall = async <O>(
     signal,
   });
   if (httpResponse.ok) {
-    const responseBuf = await httpResponse.arrayBuffer();
-    return outputSerde.deserialize(new Uint8Array(responseBuf)) as O;
+    const responseBuf = new Uint8Array(await httpResponse.arrayBuffer());
+    const decodedBuf = opts.journalValueCodec
+      ? await opts.journalValueCodec.decode(responseBuf)
+      : responseBuf;
+    return outputSerde.deserialize(decodedBuf) as O;
   }
   const body = await httpResponse.text();
   throw new HttpCallError(
@@ -412,7 +420,8 @@ class HttpIngress implements Ingress {
     const url = `${this.opts.url}/restate/a/${id}/resolve`;
     const { body, contentType } = serializeBodyWithContentType(
       payload,
-      payloadSerde ?? serde.json
+      payloadSerde ?? serde.json,
+      this.opts.journalValueCodec
     );
     const headers = {
       ...(this.opts.headers ?? {}),
@@ -481,9 +490,11 @@ class HttpIngress implements Ingress {
       headers,
     });
     if (httpResponse.ok) {
-      const responseBuf = await httpResponse.arrayBuffer();
-      const ser = resultSerde ?? serde.json;
-      return ser.deserialize(new Uint8Array(responseBuf)) as T;
+      const responseBuf = new Uint8Array(await httpResponse.arrayBuffer());
+      const decodedBuf = this.opts.journalValueCodec
+        ? await this.opts.journalValueCodec.decode(responseBuf)
+        : responseBuf;
+      return (resultSerde ?? serde.json).deserialize(decodedBuf) as T;
     }
     const body = await httpResponse.text();
     throw new HttpCallError(
@@ -504,15 +515,19 @@ function computeDelayAsIso(opts: SendOpts): string {
 
 function serializeBodyWithContentType(
   body: unknown,
-  serde: Serde<unknown>
+  serde: Serde<unknown>,
+  journalValueCodec?: JournalValueCodec
 ): {
   body?: Uint8Array;
   contentType?: string;
 } {
   if (body === undefined) {
     return {};
   }
-  const buffer = serde.serialize(body);
+  let buffer = serde.serialize(body);
+  if (journalValueCodec) {
+    buffer = journalValueCodec.encode(buffer);
+  }
   return {
     body: buffer,
     contentType: serde.contentType,

packages/restate-sdk-clients/src/public_api.ts

@@ -21,6 +21,7 @@ export type {
   Workflow,
   VirtualObject,
   Duration,
+  JournalValueCodec,
 } from "@restatedev/restate-sdk-core";
 
 export { serde } from "@restatedev/restate-sdk-core";

packages/restate-sdk-cloudflare-workers/patches/vm/sdk_shared_core_wasm_bindings.d.ts

@@ -95,7 +95,7 @@ export class WasmResponseHead {
 }
 export class WasmVM {
   free(): void;
-  constructor(headers: WasmHeader[], log_level: LogLevel, logger_id: number);
+  constructor(headers: WasmHeader[], log_level: LogLevel, logger_id: number, disable_payload_checks: boolean);
   get_response_head(): WasmResponseHead;
   notify_input(buffer: Uint8Array): void;
   notify_input_closed(): void;

packages/restate-sdk-cloudflare-workers/patches/vm/sdk_shared_core_wasm_bindings_bg.js

@@ -550,11 +550,12 @@ export class WasmVM {
      * @param {WasmHeader[]} headers
      * @param {LogLevel} log_level
      * @param {number} logger_id
+     * @param {boolean} disable_payload_checks
      */
-    constructor(headers, log_level, logger_id) {
+    constructor(headers, log_level, logger_id, disable_payload_checks) {
         const ptr0 = passArrayJsValueToWasm0(headers, wasm.__wbindgen_malloc);
         const len0 = WASM_VECTOR_LEN;
-        const ret = wasm.wasmvm_new(ptr0, len0, log_level, logger_id);
+        const ret = wasm.wasmvm_new(ptr0, len0, log_level, logger_id, disable_payload_checks);
         if (ret[2]) {
             throw takeFromExternrefTable0(ret[1]);
         }

packages/restate-sdk-cloudflare-workers/patches/vm/sdk_shared_core_wasm_bindings_bg.wasm

@@ -14,7 +14,7 @@ export const __wbg_wasminput_free: (a: number, b: number) => void;
 export const __wbg_get_wasminput_headers: (a: number) => [number, number];
 export const __wbg_get_wasminput_input: (a: number) => any;
 export const __wbg_wasmvm_free: (a: number, b: number) => void;
-export const wasmvm_new: (a: number, b: number, c: number, d: number) => [number, number, number];
+export const wasmvm_new: (a: number, b: number, c: number, d: number, e: number) => [number, number, number];
 export const wasmvm_get_response_head: (a: number) => number;
 export const wasmvm_notify_input: (a: number, b: number, c: number) => void;
 export const wasmvm_notify_input_closed: (a: number) => void;

packages/restate-sdk-cloudflare-workers/patches/vm/sdk_shared_core_wasm_bindings_bg.wasm.d.ts

@@ -0,0 +1,46 @@
+/*
+ * Copyright (c) 2023-2025 - Restate Software, Inc., Restate GmbH
+ *
+ * This file is part of the Restate SDK for Node.js/TypeScript,
+ * which is released under the MIT license.
+ *
+ * You can find a copy of the license in file LICENSE in the root
+ * directory of this repository or package, or at
+ * https://github.com/restatedev/sdk-typescript/blob/main/LICENSE
+ */
+
+/**
+ * Journal values codec.
+ *
+ * This allows to transform journal values after being serialized, before writing them to the wire.
+ *
+ * Values that are passed through the codec:
+ *
+ * * Handlers input and success output
+ * * ctx.run success results
+ * * Awakeables/Promise success results
+ * * State values
+ *
+ * @experimental
+ */
+export type JournalValueCodec = {
+  /**
+   * Encodes the given buffer.
+   *
+   * This will be applied *after* serialization.
+   *
+   * @param buf The buffer to encode. Empty byte buffers should be appropriately handled as well.
+   * @returns The encoded buffer
+   */
+  encode(buf: Uint8Array): Uint8Array;
+
+  /**
+   * Decodes the given buffer.
+   *
+   * This will be applied *before* deserialization.
+   *
+   * @param buf The buffer to decode.
+   * @returns A promise that resolves to the decoded buffer.
+   */
+  decode(buf: Uint8Array): Promise<Uint8Array>;
+};

packages/restate-sdk-core/src/entry_codec.ts

@@ -38,3 +38,5 @@ export { serde } from "./serde_api.js";
 
 export type { Duration } from "./duration.js";
 export { durationToMillis, millisOrDurationToMillis } from "./duration.js";
+
+export type { JournalValueCodec } from "./entry_codec.js";

packages/restate-sdk-core/src/public_api.ts

@@ -63,6 +63,7 @@ export type {
   WorkflowHandler,
   WorkflowSharedHandler,
   Duration,
+  JournalValueCodec,
 } from "@restatedev/restate-sdk-core";
 export { serde } from "@restatedev/restate-sdk-core";