Support already completed journals in lambda handlers (#189)

Author: jackkleeman

src/connection/lambda_connection.ts

@@ -23,13 +23,12 @@ const RESOLVED: Promise<void> = Promise.resolve();
 export class LambdaConnection implements Connection {
   // Empty buffer to store journal output messages
   private outputBuffer: Buffer = Buffer.alloc(0);
-  private suspendedOrCompleted = false;
 
   // Callback to resolve the invocation promise of the Lambda handler when the response is ready
   private readonly completionPromise: Promise<Buffer>;
   private resolveOnCompleted!: (value: Buffer | PromiseLike<Buffer>) => void;
 
-  constructor() {
+  constructor(private suspendedOrCompleted = false) {
     // Promise that signals when the invocation is over, to then flush the messages
     this.completionPromise = new Promise<Buffer>((resolve) => {
       this.resolveOnCompleted = resolve;

src/server/restate_lambda_handler.ts

@@ -28,6 +28,7 @@ import { Message } from "../types/types";
 import { StateMachine } from "../state_machine";
 import { ensureError } from "../types/errors";
 import { KeyedRouter, UnKeyedRouter } from "../public_api";
+import {OUTPUT_STREAM_ENTRY_MESSAGE_TYPE} from "../types/protocol";
 
 /**
  * Creates an Restate entrypoint for services deployed on AWS Lambda and invoked
@@ -223,10 +224,11 @@ export class LambdaRestateServer extends BaseRestateServer {
       let decodedEntries: Message[] | null = decodeLambdaBody(event.body);
       const journalBuilder = new InvocationBuilder(method);
       decodedEntries.forEach((e: Message) => journalBuilder.handleMessage(e));
+      const alreadyCompleted = decodedEntries.find((e: Message) => e.messageType === OUTPUT_STREAM_ENTRY_MESSAGE_TYPE) !== undefined
       decodedEntries = null;
 
       // set up and invoke the state machine
-      const connection = new LambdaConnection();
+      const connection = new LambdaConnection(alreadyCompleted);
       const stateMachine = new StateMachine(
         connection,
         journalBuilder.build(),