Support HTTP2 client

Author: igalshilman

examples/embedded_example.ts

@@ -26,14 +26,14 @@ app.post("/workflow", async (req: Request, res: Response) => {
     id,
     input: name,
     handler: async (ctx, name) => {
-      const p1 = await ctx.sideEffect(async () => `Hello ${name}!`);
-      const p2 = await ctx.sideEffect(async () => `Bonjour ${name}`);
-      const p3 = await ctx.sideEffect(async () => `Hi ${name}`);
-      // const p4 = await ctx
-      //   .rpc<{ greet: (name: string) => Promise<string> }>({ path: "greeter" })
+      const p1 = ctx.sideEffect(async () => `Hello ${name}!`);
+      const p2 = ctx.sideEffect(async () => `Bonjour ${name}`);
+      const p3 = ctx.sideEffect(async () => `Hi ${name}`);
+      //const p4 =  ctx
+      // .rpc<{ greet: (name: string) => Promise<string> }>({ path: "greeter" })
       //   .greet(name);
 
-      return p1 + p2 + p3; //+ p4;
+      return (await p1) + (await p2) + (await p3);
     },
   });
 

proto/services.proto

@@ -0,0 +1,182 @@
+// Copyright (c) 2023 - Restate Software, Inc., Restate GmbH
+//
+// This file is part of the Restate service protocol, 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/proto/blob/main/LICENSE
+
+syntax = "proto3";
+
+/*
+  This package contains internal service interfaces
+ */
+package dev.restate.internal;
+
+import "dev/restate/ext.proto";
+import "google/protobuf/empty.proto";
+import "google/protobuf/struct.proto";
+
+service Proxy {
+  // Proxy invocation through this service. This service is mostly used for proxying invocations through a specific partition processor, e.g. to reuse the deduplication id map.
+  rpc ProxyThrough(ProxyThroughRequest) returns (google.protobuf.Empty);
+}
+
+message ProxyThroughRequest {
+  string target_service = 1;
+  string target_method = 2;
+  bytes target_key = 3;
+  bytes target_invocation_uuid = 4;
+
+  bytes input = 5;
+}
+
+// RemoteContext service to implement the embedded handler API
+service RemoteContext {
+  option (dev.restate.ext.service_type) = KEYED;
+
+  // Start a new invocation, or resume a previously existing one.
+  //
+  // If another client is already executing this invocation, it will be fenced off and this client will take precedence.
+  //
+  // This method is not idempotent.
+  rpc Start(StartRequest) returns (StartResponse);
+
+  // Send new messages to append to the message stream.
+  //
+  // This method is not idempotent, and a request can fail for several reasons,
+  // including errors in sent messages, or some other transient error.
+  // The client should consider the stream in an unrecoverable error state and it can retry
+  // by creating a new stream through Start() with a different stream_id.
+  //
+  // Once the invocation is completed, subsequent Send fail.
+  rpc Send(SendRequest) returns (SendResponse);
+
+  // Receive new messages from the message stream.
+  //
+  // This method is not idempotent, and a request can fail for several reasons,
+  // including errors in sent messages, or some other transient error.
+  // The client should consider the stream in an unrecoverable error state and it can retry
+  // by creating a new stream through Start() with a different stream_id.
+  //
+  // If the invocation is completed, Recv returns a response with messages field empty.
+  rpc Recv(RecvRequest) returns (RecvResponse);
+
+  // Get the result of the invocation.
+  //
+  // In case another client is executing the invocation (through a sequence of Start/Send/Recv), this method will block
+  // until a response is computed.
+  // In case the response is already available, it will return immediately with the response.
+  // In case no client is executing the invocation, that is no client ever invoked Start() for the given operation_id,
+  // this method will return response.none.
+  //
+  // This method can be safely invoked by multiple clients and it's idempotent.
+  rpc GetResult(GetResultRequest) returns (GetResultResponse);
+
+  // Cleanup all the state of the invocation, excluding the user state.
+  //
+  // This is automatically executed when retention_period_sec is past, but it can be manually invoked before the expiry time elapsed.
+  rpc Cleanup(CleanupRequest) returns (google.protobuf.Empty);
+}
+
+message StartRequest {
+  // User provided operation id, this is used as idempotency key.
+  string operation_id = 1 [(dev.restate.ext.field) = KEY];
+
+  // Stream id to uniquely identify a open stream between client and Restate.
+  // There can be at most one open stream at the same time.
+  string stream_id = 2;
+
+  // Retention period for the response in seconds.
+  // After the invocation completes, the response will be persisted for the given duration.
+  // Afterwards, the system will cleanup the response and treats any subsequent invocation with same operation_id as new.
+  //
+  // If not set, 30 minutes will be used as retention period.
+  uint32 retention_period_sec = 3;
+
+  // Argument of the invocation
+  bytes argument = 4;
+}
+
+message StartResponse {
+  oneof invocation_status {
+    // Contains the concatenated first messages of the stream, encoded using the same framing used by service-protocol
+    bytes executing = 1;
+
+    // Contains the result of the invocation
+    GetResultResponse completed = 2;
+  }
+}
+
+message SendRequest {
+  // User provided operation id, this is used as idempotency key.
+  string operation_id = 1 [(dev.restate.ext.field) = KEY];
+
+  // Stream id to uniquely identify a open stream between client and Restate.
+  // There can be at most one open stream at the same time.
+  string stream_id = 2;
+
+  // Contains the concatenated messages of the stream, encoded using the same framing used by service-protocol
+  bytes messages = 3;
+}
+
+message SendResponse {
+  oneof response {
+    google.protobuf.Empty ok = 1;
+
+    // This means the provided stream_id is invalid, and it should not be reused,
+    // nor the client should create a new stream using Start().
+    // The client can instead read the invocation result using GetResult().
+    google.protobuf.Empty invalid_stream = 2;
+  }
+}
+
+message RecvRequest {
+  // User provided operation id, this is used as idempotency key.
+  string operation_id = 1 [(dev.restate.ext.field) = KEY];
+
+  // Stream id to uniquely identify a open stream between client and Restate.
+  // There can be at most one open stream at the same time.
+  string stream_id = 2;
+}
+
+message RecvResponse {
+  oneof response {
+    // Contains the concatenated messages of the stream, encoded using the same framing used by service-protocol
+    bytes messages = 1;
+
+    // This means the provided stream_id is invalid, and it should not be reused,
+    // nor the client should create a new stream using Start().
+    // The client can instead read the invocation result using GetResult().
+    google.protobuf.Empty invalid_stream = 2;
+  }
+}
+
+message GetResultRequest {
+  // User provided operation id, this is used as idempotency key.
+  string operation_id = 1 [(dev.restate.ext.field) = KEY];
+}
+
+message GetResultResponse {
+  message InvocationFailure {
+    uint32 code = 1;
+    string message = 2;
+  }
+
+  oneof response {
+    // See GetResult documentation
+    google.protobuf.Empty none = 1;
+    bytes success = 2;
+    InvocationFailure failure = 3;
+  }
+
+  // Timestamp of the response expiry time in RFC3339.
+  // Empty if response = none
+  string expiry_time = 15;
+}
+
+message CleanupRequest {
+  // User provided operation id, this is used as idempotency key.
+  string operation_id = 1 [(dev.restate.ext.field) = KEY];
+}

src/connection/embedded_connection.ts

@@ -0,0 +1,67 @@
+import { RemoteContext } from "../generated/proto/services";
+import { encodeMessages } from "../io/encoder";
+import { Message } from "../types/types";
+import { Connection } from "./connection";
+
+export class FencedOffError extends Error {
+  constructor() {
+    super("FencedOff");
+  }
+}
+
+export class EmbeddedConnection implements Connection {
+  private queue: Message[] = [];
+  private flushing: Promise<void> = Promise.resolve();
+
+  constructor(
+    private readonly operationId: string,
+    private readonly streamId: string,
+    private readonly remote: RemoteContext
+  ) {}
+
+  send(msg: Message): Promise<void> {
+    const len = this.queue.push(msg);
+    if (len === 1) {
+      // we are the first in line, therefore we schedule a flush,
+      // BUT we must wait for the previous flush to end.
+      //
+      this.flushing = this.flushing.then(() => this.scheduleFlush());
+    }
+    // tag along to the previously scheduled flush.
+    return this.flushing;
+  }
+
+  end(): Promise<void> {
+    return this.flush();
+  }
+
+  private scheduleFlush(): Promise<void> {
+    // schedule a flush at the end of the current event loop iteration.
+    return new Promise((resolve, reject) =>
+      setImmediate(() => {
+        this.flush().then(resolve).catch(reject);
+      })
+    );
+  }
+
+  private async flush(): Promise<void> {
+    if (this.queue.length === 0) {
+      return Promise.resolve();
+    }
+    const buffer = encodeMessages(this.queue) as Buffer;
+    this.queue = [];
+
+    const res = await this.remote.send({
+      operationId: this.operationId,
+      streamId: this.streamId,
+      messages: buffer,
+    });
+
+    if (!res.ok) {
+      throw new Error("Error connecting to restate");
+    }
+    if (res.invalidStream !== undefined) {
+      throw new FencedOffError();
+    }
+  }
+}

src/embedded/api.ts

@@ -0,0 +1,34 @@
+import { RpcContext } from "../restate_context";
+import { doInvoke } from "./invocation";
+import { wrapHandler } from "./handler";
+import crypto from "crypto";
+import { RemoteContext } from "../generated/proto/services";
+import { bufConnectRemoteContext } from "./http2_remote";
+
+export type RestateConnectionOptions = {
+  ingress: string;
+};
+
+export type RestateInvocationOptions<I, O> = {
+  id: string;
+  handler: (ctx: RpcContext, input: I) => Promise<O>;
+  input: I;
+  retain?: number;
+};
+
+export const connection = (opts: RestateConnectionOptions): RestateConnection =>
+  new RestateConnection(opts);
+
+export class RestateConnection {
+  private remote: RemoteContext;
+
+  constructor(readonly opts: RestateConnectionOptions) {
+    this.remote = bufConnectRemoteContext(opts.ingress);
+  }
+
+  public invoke<I, O>(opt: RestateInvocationOptions<I, O>): Promise<O> {
+    const method = wrapHandler(opt.handler);
+    const streamId = crypto.randomUUID();
+    return doInvoke<I, O>(this.remote, opt.id, streamId, method, opt.input);
+  }
+}