Adjust to the latest service definition

Author: igalshilman

proto/services.proto

@@ -18,20 +18,6 @@ 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;
@@ -129,6 +115,9 @@ message SendResponse {
     // 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;
+
+    // This means the invocation is completed, and the result should be collected using GetResult
+    google.protobuf.Empty invocation_completed = 3;
   }
 }
 
@@ -150,6 +139,9 @@ message RecvResponse {
     // 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;
+
+    // This means the invocation is completed, and the result should be collected using GetResult
+    google.protobuf.Empty invocation_completed = 3;
   }
 }
 

src/connection/embedded_connection.ts

@@ -9,6 +9,12 @@ export class FencedOffError extends Error {
   }
 }
 
+export class InvocationAlreadyCompletedError extends Error {
+  constructor() {
+    super("Completed");
+  }
+}
+
 export class EmbeddedConnection implements Connection {
   private queue: Message[] = [];
   private flushing: Promise<void> = Promise.resolve();
@@ -57,11 +63,11 @@ export class EmbeddedConnection implements Connection {
       messages: buffer,
     });
 
-    if (!res.ok) {
-      throw new Error("Error connecting to restate");
-    }
     if (res.invalidStream !== undefined) {
       throw new FencedOffError();
     }
+    if (res.invocationCompleted !== undefined) {
+      throw new InvocationAlreadyCompletedError();
+    }
   }
 }

src/embedded/invocation.ts

@@ -16,8 +16,10 @@ import { InvocationBuilder } from "../invocation";
 import { HostedGrpcServiceMethod } from "../types/grpc";
 import { StateMachine } from "../state_machine";
 import { ProtocolMode } from "../generated/proto/discovery";
-import { RequestError } from "./http2_remote";
-import { EmbeddedConnection } from "../connection/embedded_connection";
+import {
+  EmbeddedConnection,
+  FencedOffError,
+} from "../connection/embedded_connection";
 
 export const doInvoke = async <I, O>(
   remote: RemoteContext,
@@ -78,16 +80,18 @@ export const doInvoke = async <I, O>(
         streamId,
       });
       if (recv.invalidStream !== undefined) {
-        throw new Error("Operation fenced off");
+        throw new FencedOffError();
+      }
+      if (recv.invocationCompleted !== undefined) {
+        break;
       }
       const buffer = recv.messages ?? Buffer.alloc(0);
       const messages = decodeMessagesBuffer(buffer);
       messages.forEach((m: Message) => stateMachine.handleMessage(m));
     }
   } catch (e) {
-    if (!(e instanceof RequestError) || !e.precondtionFailed()) {
-      stateMachine.handleStreamError(e as Error);
-    }
+    stateMachine.handleStreamError(e as Error);
+    throw e;
   }
 
   //