Final wireup with API definition for all and race (more to follow)

src/promise_combinator_tracker.ts

@@ -32,7 +32,7 @@ export function newJournalEntryPromiseId(entryIndex: number): PromiseId {
  */
 function prepareReplayedPromiseCombinator(
   combinatorIndex: number,
-  combinatorConstructor: (promises: Promise<any>[]) => Promise<any>,
+  combinatorConstructor: (promises: PromiseLike<any>[]) => Promise<any>,
   promises: Array<{ id: PromiseId; promise: Promise<any> }>,
   readReplayOrder: (combinatorIndex: number) => PromiseId[]
 ): WrappedPromise<any> {
@@ -133,7 +133,7 @@ export class PromiseCombinatorTracker {
   ) {}
 
   public createCombinatorInReplayMode(
-    combinatorConstructor: (promises: Promise<any>[]) => Promise<any>,
+    combinatorConstructor: (promises: PromiseLike<any>[]) => Promise<any>,
     promises: Array<{ id: PromiseId; promise: Promise<any> }>
   ): WrappedPromise<any> {
     const combinatorIndex = this.nextCombinatorIndex;

src/restate_context.ts

@@ -189,6 +189,34 @@ export interface RestateBaseContext {
    * await ctx.sleep(1000);
    */
   sleep(millis: number): CombineablePromise<void>;
+
+  // --- Promise combinators
+
+  /**
+   * Creates a Promise that is resolved with an array of results when all of the provided Promises
+   * resolve, or rejected when any Promise is rejected.
+   *
+   * See {@link Promise.all} for more details.
+   *
+   * @param values An iterable of Promises.
+   * @returns A new Promise.
+   */
+  all<T extends readonly CombineablePromise<unknown>[] | []>(
+    values: T
+  ): Promise<{ -readonly [P in keyof T]: Awaited<T[P]> }>;
+
+  /**
+   * Creates a Promise that is resolved or rejected when any of the provided Promises are resolved
+   * or rejected.
+   *
+   * See {@link Promise.race} for more details.
+   *
+   * @param values An iterable of Promises.
+   * @returns A new Promise.
+   */
+  race<T extends readonly CombineablePromise<unknown>[] | []>(
+    values: T
+  ): Promise<Awaited<T[number]>>;
 }
 
 export interface Rand {

src/restate_context_impl.ts

@@ -62,6 +62,10 @@ import { Client, SendClient } from "./types/router";
 import { RpcRequest, RpcResponse } from "./generated/proto/dynrpc";
 import { requestFromArgs } from "./utils/assumpsions";
 import { RandImpl } from "./utils/rand";
+import {
+  newJournalEntryPromiseId,
+  PromiseId,
+} from "./promise_combinator_tracker";
 
 export enum CallContexType {
   None,
@@ -225,6 +229,10 @@ export class RestateGrpcContextImpl implements RestateGrpcContext {
     );
   }
 
+    rpcGateway(): RpcGateway {
+        return new RpcContextImpl(this);
+    }
+
   // DON'T make this function async!!!
   // The reason is that we want the erros thrown by the initial checks to be propagated in the caller context,
   // and not in the promise context. To understand the semantic difference, make this function async and run the
@@ -344,6 +352,8 @@ export class RestateGrpcContextImpl implements RestateGrpcContext {
     );
   }
 
+  // -- Awakeables
+
   public awakeable<T>(): { id: string; promise: CombineablePromise<T> } {
     this.checkState("awakeable");
 
@@ -399,6 +409,44 @@ export class RestateGrpcContextImpl implements RestateGrpcContext {
     );
   }
 
+  // -- Combinators
+
+  all<T extends readonly CombineablePromise<unknown>[] | []>(
+    values: T
+  ): Promise<{ -readonly [P in keyof T]: Awaited<T[P]> }> {
+    return this.stateMachine.createCombinator(
+      Promise.all.bind(Promise),
+      this.extractPromisesWithIds(values)
+    );
+  }
+
+  race<T extends readonly CombineablePromise<unknown>[] | []>(
+    values: T
+  ): Promise<Awaited<T[number]>> {
+    return this.stateMachine.createCombinator(
+      Promise.race.bind(Promise),
+      this.extractPromisesWithIds(values)
+    );
+  }
+
+  private extractPromisesWithIds(
+    promises: Iterable<CombineablePromise<any>>
+  ): Array<{ id: PromiseId; promise: Promise<any> }> {
+    const outPromises = [];
+
+    for (const promise of promises) {
+      const index = (promise as InternalCombineablePromise<any>).journalIndex;
+      outPromises.push({
+        id: newJournalEntryPromiseId(index),
+        promise: promise,
+      });
+    }
+
+    return outPromises;
+  }
+
+  // -- Various private methods
+
   private isInSideEffect(): boolean {
     const context = RestateGrpcContextImpl.callContext.getStore();
     return context?.type === CallContexType.SideEffect;
@@ -460,10 +508,6 @@ export class RestateGrpcContextImpl implements RestateGrpcContext {
       },
     }) as InternalCombineablePromise<T>;
   }
-
-  rpcGateway(): RpcGateway {
-    return new RpcContextImpl(this);
-  }
 }
 
 async function executeWithRetries<T>(
@@ -626,6 +670,18 @@ export class RpcContextImpl implements RpcContext {
     return this.ctx.sleep(millis);
   }
 
+  all<T extends readonly CombineablePromise<unknown>[] | []>(
+    values: T
+  ): Promise<{ -readonly [P in keyof T]: Awaited<T[P]> }> {
+    return this.ctx.all(values);
+  }
+
+  race<T extends readonly CombineablePromise<unknown>[] | []>(
+    values: T
+  ): Promise<Awaited<T[number]>> {
+    return this.ctx.race(values);
+  }
+
   grpcChannel(): RestateGrpcChannel {
     return this.ctx;
   }

src/state_machine.ts

@@ -194,16 +194,30 @@ export class StateMachine<I, O> implements RestateStreamConsumer {
 
   // -- Methods related to combinators to wire up promise combinator API with PromiseCombinatorTracker
 
-  public createCombinator() {
-    // TODO create combinator
-    //  if replay { combinator in replay mode } else { combinator in processing mode }
+  public createCombinator(
+    combinatorConstructor: (promises: PromiseLike<any>[]) => Promise<any>,
+    promises: Array<{ id: PromiseId; promise: Promise<any> }>
+  ) {
+    if (this.stateMachineClosed) {
+      return WRAPPED_PROMISE_PENDING as WrappedPromise<any>;
+    }
 
-    // We need to wrap deeply again to schedule suspension here!
-    return wrapDeeply(Promise.resolve("TODO"), () => {
-      if (this.journal.isUnResolved(0 /* TODO */)) {
-        this.scheduleSuspension();
-      }
-    });
+    if (this.journal.isProcessing()) {
+      return wrapDeeply(
+        this.promiseCombinatorTracker.createCombinatorInProcessingMode(
+          combinatorConstructor,
+          promises
+        ),
+        () => {
+          this.scheduleSuspension();
+        }
+      );
+    } else {
+      return this.promiseCombinatorTracker.createCombinatorInReplayMode(
+        combinatorConstructor,
+        promises
+      );
+    }
   }
 
   readCombinatorOrderEntry(combinatorId: number): PromiseId[] {