Add missing sharedcore features

- add call/send with idempotency key
- add cancel with invocation id
- add call/send to return the invocation id
- add ctx.attach

Author: igalshilman

packages/restate-e2e-services/exclusions.yaml

@@ -1,26 +1,11 @@
 exclusions:
   "alwaysSuspending":
-    - "dev.restate.sdktesting.tests.Cancellation.cancelFromContext"
     - "dev.restate.sdktesting.tests.Combinators.awakeableOrTimeoutUsingAwaitAny"
-    - "dev.restate.sdktesting.tests.ServiceToServiceCommunication.callWithIdempotencyKey"
-    - "dev.restate.sdktesting.tests.ServiceToServiceCommunication.oneWayCallWithIdempotencyKey"
   "default":
-    - "dev.restate.sdktesting.tests.Cancellation.cancelFromContext"
     - "dev.restate.sdktesting.tests.Combinators.awakeableOrTimeoutUsingAwaitAny"
-    - "dev.restate.sdktesting.tests.ServiceToServiceCommunication.callWithIdempotencyKey"
-    - "dev.restate.sdktesting.tests.ServiceToServiceCommunication.oneWayCallWithIdempotencyKey"
   "singleThreadSinglePartition":
-    - "dev.restate.sdktesting.tests.Cancellation.cancelFromContext"
     - "dev.restate.sdktesting.tests.Combinators.awakeableOrTimeoutUsingAwaitAny"
-    - "dev.restate.sdktesting.tests.ServiceToServiceCommunication.callWithIdempotencyKey"
-    - "dev.restate.sdktesting.tests.ServiceToServiceCommunication.oneWayCallWithIdempotencyKey"
   "threeNodes":
-    - "dev.restate.sdktesting.tests.Cancellation.cancelFromContext"
     - "dev.restate.sdktesting.tests.Combinators.awakeableOrTimeoutUsingAwaitAny"
-    - "dev.restate.sdktesting.tests.ServiceToServiceCommunication.callWithIdempotencyKey"
-    - "dev.restate.sdktesting.tests.ServiceToServiceCommunication.oneWayCallWithIdempotencyKey"
   "threeNodesAlwaysSuspending":
-    - "dev.restate.sdktesting.tests.Cancellation.cancelFromContext"
     - "dev.restate.sdktesting.tests.Combinators.awakeableOrTimeoutUsingAwaitAny"
-    - "dev.restate.sdktesting.tests.ServiceToServiceCommunication.callWithIdempotencyKey"
-    - "dev.restate.sdktesting.tests.ServiceToServiceCommunication.oneWayCallWithIdempotencyKey"

packages/restate-e2e-services/src/proxy.ts

@@ -32,35 +32,28 @@ function rawCall(
   ctx: restate.Context,
   request: ProxyRequest
 ): Promise<Uint8Array> {
-  // TODO add idempotency key support here
-  if (request.idempotencyKey != undefined) {
-    throw new TerminalError("idempotency key not supported yet");
-  }
   return ctx.genericCall({
     service: request.serviceName,
     method: request.handlerName,
     key: request.virtualObjectKey,
     inputSerde: restate.serde.binary,
     outputSerde: restate.serde.binary,
     parameter: new Uint8Array(request.message),
+    idempotencyKey: request.idempotencyKey,
   });
 }
 
 function rawSend(ctx: restate.Context, request: ProxyRequest): Promise<string> {
-  // TODO add idempotency key support here
-  if (request.idempotencyKey != undefined) {
-    throw new TerminalError("idempotency key not supported yet");
-  }
-  ctx.genericSend({
+  const handle = ctx.genericSend({
     service: request.serviceName,
     method: request.handlerName,
     key: request.virtualObjectKey,
     inputSerde: restate.serde.binary,
     parameter: new Uint8Array(request.message),
     delay: request.delayMillis,
+    idempotencyKey: request.idempotencyKey,
   });
-  // TODO this should return the invocation id
-  return Promise.resolve("unknown");
+  return handle.invocationId;
 }
 
 const o = restate.service({

packages/restate-e2e-services/src/test_utils.ts

@@ -66,13 +66,11 @@ const o = restate.service({
     },
 
     async cancelInvocation(
-      // eslint-disable-next-line @typescript-eslint/no-unused-vars
       ctx: restate.Context,
-      // eslint-disable-next-line @typescript-eslint/no-unused-vars
       invocationId: string
     ): Promise<void> {
-      // TODO add cancel invocation here!
-      return Promise.reject(new TerminalError("cancel is not supported yet"));
+      const id = restate.InvocationIdParser.fromString(invocationId);
+      ctx.cancel(id);
     },
   },
 });

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

@@ -91,8 +91,10 @@ export class WasmVM {
   sys_clear_state(key: string): void;
   sys_clear_all_state(): void;
   sys_sleep(millis: bigint): number;
-  sys_call(service: string, handler: string, buffer: Uint8Array, key: string | null | undefined, headers: WasmHeader[]): WasmCallHandle;
-  sys_send(service: string, handler: string, buffer: Uint8Array, key: string | null | undefined, headers: WasmHeader[], delay?: bigint | null): WasmSendHandle;
+  sys_attach_invocation(invocation_id: string): number;
+  sys_get_invocation_output(invocation_id: string): number;
+  sys_call(service: string, handler: string, buffer: Uint8Array, key: string | null | undefined, headers: WasmHeader[], idempotency_key?: string | null): WasmCallHandle;
+  sys_send(service: string, handler: string, buffer: Uint8Array, key: string | null | undefined, headers: WasmHeader[], delay?: bigint | null, idempotency_key?: string | null): WasmSendHandle;
   sys_awakeable(): WasmAwakeable;
   sys_complete_awakeable_success(id: string, buffer: Uint8Array): void;
   sys_complete_awakeable_failure(id: string, value: WasmFailure): void;

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

@@ -691,15 +691,42 @@ export class WasmVM {
         }
         return ret[0] >>> 0;
     }
+    /**
+     * @param {string} invocation_id
+     * @returns {number}
+     */
+    sys_attach_invocation(invocation_id) {
+        const ptr0 = passStringToWasm0(invocation_id, wasm.__wbindgen_malloc, wasm.__wbindgen_realloc);
+        const len0 = WASM_VECTOR_LEN;
+        const ret = wasm.wasmvm_sys_attach_invocation(this.__wbg_ptr, ptr0, len0);
+        if (ret[2]) {
+            throw takeFromExternrefTable0(ret[1]);
+        }
+        return ret[0] >>> 0;
+    }
+    /**
+     * @param {string} invocation_id
+     * @returns {number}
+     */
+    sys_get_invocation_output(invocation_id) {
+        const ptr0 = passStringToWasm0(invocation_id, wasm.__wbindgen_malloc, wasm.__wbindgen_realloc);
+        const len0 = WASM_VECTOR_LEN;
+        const ret = wasm.wasmvm_sys_get_invocation_output(this.__wbg_ptr, ptr0, len0);
+        if (ret[2]) {
+            throw takeFromExternrefTable0(ret[1]);
+        }
+        return ret[0] >>> 0;
+    }
     /**
      * @param {string} service
      * @param {string} handler
      * @param {Uint8Array} buffer
      * @param {string | null | undefined} key
      * @param {WasmHeader[]} headers
+     * @param {string | null} [idempotency_key]
      * @returns {WasmCallHandle}
      */
-    sys_call(service, handler, buffer, key, headers) {
+    sys_call(service, handler, buffer, key, headers, idempotency_key) {
         const ptr0 = passStringToWasm0(service, wasm.__wbindgen_malloc, wasm.__wbindgen_realloc);
         const len0 = WASM_VECTOR_LEN;
         const ptr1 = passStringToWasm0(handler, wasm.__wbindgen_malloc, wasm.__wbindgen_realloc);
@@ -710,7 +737,9 @@ export class WasmVM {
         var len3 = WASM_VECTOR_LEN;
         const ptr4 = passArrayJsValueToWasm0(headers, wasm.__wbindgen_malloc);
         const len4 = WASM_VECTOR_LEN;
-        const ret = wasm.wasmvm_sys_call(this.__wbg_ptr, ptr0, len0, ptr1, len1, ptr2, len2, ptr3, len3, ptr4, len4);
+        var ptr5 = isLikeNone(idempotency_key) ? 0 : passStringToWasm0(idempotency_key, wasm.__wbindgen_malloc, wasm.__wbindgen_realloc);
+        var len5 = WASM_VECTOR_LEN;
+        const ret = wasm.wasmvm_sys_call(this.__wbg_ptr, ptr0, len0, ptr1, len1, ptr2, len2, ptr3, len3, ptr4, len4, ptr5, len5);
         if (ret[2]) {
             throw takeFromExternrefTable0(ret[1]);
         }
@@ -723,9 +752,10 @@ export class WasmVM {
      * @param {string | null | undefined} key
      * @param {WasmHeader[]} headers
      * @param {bigint | null} [delay]
+     * @param {string | null} [idempotency_key]
      * @returns {WasmSendHandle}
      */
-    sys_send(service, handler, buffer, key, headers, delay) {
+    sys_send(service, handler, buffer, key, headers, delay, idempotency_key) {
         const ptr0 = passStringToWasm0(service, wasm.__wbindgen_malloc, wasm.__wbindgen_realloc);
         const len0 = WASM_VECTOR_LEN;
         const ptr1 = passStringToWasm0(handler, wasm.__wbindgen_malloc, wasm.__wbindgen_realloc);
@@ -736,7 +766,9 @@ export class WasmVM {
         var len3 = WASM_VECTOR_LEN;
         const ptr4 = passArrayJsValueToWasm0(headers, wasm.__wbindgen_malloc);
         const len4 = WASM_VECTOR_LEN;
-        const ret = wasm.wasmvm_sys_send(this.__wbg_ptr, ptr0, len0, ptr1, len1, ptr2, len2, ptr3, len3, ptr4, len4, !isLikeNone(delay), isLikeNone(delay) ? BigInt(0) : delay);
+        var ptr5 = isLikeNone(idempotency_key) ? 0 : passStringToWasm0(idempotency_key, wasm.__wbindgen_malloc, wasm.__wbindgen_realloc);
+        var len5 = WASM_VECTOR_LEN;
+        const ret = wasm.wasmvm_sys_send(this.__wbg_ptr, ptr0, len0, ptr1, len1, ptr2, len2, ptr3, len3, ptr4, len4, !isLikeNone(delay), isLikeNone(delay) ? BigInt(0) : delay, ptr5, len5);
         if (ret[2]) {
             throw takeFromExternrefTable0(ret[1]);
         }

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

@@ -31,8 +31,10 @@ export const wasmvm_sys_set_state: (a: number, b: number, c: number, d: number,
 export const wasmvm_sys_clear_state: (a: number, b: number, c: number) => [number, number];
 export const wasmvm_sys_clear_all_state: (a: number) => [number, number];
 export const wasmvm_sys_sleep: (a: number, b: bigint) => [number, number, number];
-export const wasmvm_sys_call: (a: number, b: number, c: number, d: number, e: number, f: number, g: number, h: number, i: number, j: number, k: number) => [number, number, number];
-export const wasmvm_sys_send: (a: number, b: number, c: number, d: number, e: number, f: number, g: number, h: number, i: number, j: number, k: number, l: number, m: bigint) => [number, number, number];
+export const wasmvm_sys_attach_invocation: (a: number, b: number, c: number) => [number, number, number];
+export const wasmvm_sys_get_invocation_output: (a: number, b: number, c: number) => [number, number, number];
+export const wasmvm_sys_call: (a: number, b: number, c: number, d: number, e: number, f: number, g: number, h: number, i: number, j: number, k: number, l: number, m: number) => [number, number, number];
+export const wasmvm_sys_send: (a: number, b: number, c: number, d: number, e: number, f: number, g: number, h: number, i: number, j: number, k: number, l: number, m: bigint, n: number, o: number) => [number, number, number];
 export const wasmvm_sys_awakeable: (a: number) => [number, number, number];
 export const wasmvm_sys_complete_awakeable_success: (a: number, b: number, c: number, d: number, e: number) => [number, number];
 export const wasmvm_sys_complete_awakeable_failure: (a: number, b: number, c: number, d: any) => [number, number];

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

@@ -18,8 +18,11 @@ export type {
   Rand,
   GenericCall,
   GenericSend,
+  InvocationId,
+  InvocationHandle,
+  InvocationPromise,
 } from "./context.js";
-export { CombineablePromise } from "./context.js";
+export { InvocationIdParser, CombineablePromise } from "./context.js";
 
 export type { Serde } from "@restatedev/restate-sdk-core";
 export { serde } from "@restatedev/restate-sdk-core";

packages/restate-sdk/src/common_api.ts

@@ -231,6 +231,7 @@ export type GenericCall<REQ, RES> = {
   headers?: Record<string, string>;
   inputSerde?: Serde<REQ>;
   outputSerde?: Serde<RES>;
+  idempotencyKey?: string;
 };
 
 /**
@@ -246,6 +247,7 @@ export type GenericSend<REQ> = {
   headers?: Record<string, string>;
   inputSerde?: Serde<REQ>;
   delay?: number;
+  idempotencyKey?: string;
 };
 
 /**
@@ -537,15 +539,33 @@ export interface Context extends RestateContext {
 
   genericCall<REQ = Uint8Array, RES = Uint8Array>(
     call: GenericCall<REQ, RES>
-  ): Promise<RES>;
+  ): InvocationPromise<RES>;
 
-  genericSend<REQ = Uint8Array>(call: GenericSend<REQ>): void;
+  genericSend<REQ = Uint8Array>(call: GenericSend<REQ>): InvocationHandle;
 
   /**
    * Returns the raw request that triggered that handler.
    * Use that object to inspect the original request headers
    */
   request(): Request;
+
+  /**
+   * Cancel an invocation
+   *
+   * @param invocationId the invocation id to cancel
+   */
+  cancel(invocationId: InvocationId): void;
+
+  /**
+   * Attach to an invocation
+   *
+   * @param invocationId the invocation id to attach to
+   * @param serde the serde to use for the result, default to JSON serde.
+   */
+  attach<T>(
+    invocationId: InvocationId,
+    serde?: Serde<T>
+  ): CombineablePromise<T>;
 }
 
 /**
@@ -633,6 +653,34 @@ export type CombineablePromise<T> = Promise<T> & {
   orTimeout(millis: number): CombineablePromise<T>;
 };
 
+/**
+ * Represents an invocation id.
+ * @see {@link InvocationIdParser}
+ */
+export type InvocationId = string & { __brand: "InvocationId" };
+
+export const InvocationIdParser = {
+  /**
+   * Creates an invocation id from a string.
+   * @param id the string to use as invocation id.
+   */
+  fromString(id: string): InvocationId {
+    if (!id.startsWith("inv")) {
+      throw new Error(
+        `Expected invocation id to start with 'inv' but got ${id}`
+      );
+    }
+    return id as InvocationId;
+  },
+};
+
+export type InvocationHandle = {
+  // The invocation id of the call
+  readonly invocationId: Promise<InvocationId>;
+};
+
+export type InvocationPromise<T> = CombineablePromise<T> & InvocationHandle;
+
 export const CombineablePromise = {
   /**
    * Creates a Promise that is resolved with an array of results when all of the provided Promises