pr feedback

Signed-off-by: Nik Nasr <nik@restate.dev>

Author: nikrooz

packages/libs/restate-sdk/src/endpoint/handlers/fetch.ts

@@ -48,9 +48,8 @@ export function fetcher(handler: RestateHandler) {
           abortSignal: event.signal,
         })
         .catch((e) => {
-          // wrapResponseWithSafety guarantees writeHead is called and
-          // closes the output stream; anything reaching here is a
-          // post-commit error.
+          // Responses handle their own errors before rejecting; anything
+          // reaching here is an unexpected failure — just log.
           const error = ensureError(e);
           const logger =
             tryCreateContextualLogger(

packages/libs/restate-sdk/src/endpoint/handlers/generic.ts

@@ -72,16 +72,6 @@ const HOOK_CONTEXT_IS_PROCESSING_SYMBOL = Symbol.for(
   "@restatedev/restate-sdk/hooks.isProcessing"
 );
 
-/**
- * Builds the Restate handler for an endpoint.
- *
- * Responses returned by `handle()` are wrapped via
- * {@link wrapResponseWithSafety}, which enforces exactly-once `writeHead`,
- * emits a 500 fallback if `process()` fails before `writeHead` or resolves
- * without committing a head, and closes the output stream on every path.
- * Adapters (node, fetch, lambda) can therefore treat transport failures as
- * logging concerns.
- */
 export function createRestateHandler(
   endpoint: Endpoint,
   protocolMode: ProtocolMode,
@@ -94,78 +84,6 @@ export function createRestateHandler(
   );
 }
 
-/**
- * Safety net around a `RestateResponse` returned by `_handle()`, applied
- * in `RestateHandlerImpl.handle()` before the response reaches any adapter.
- *
- * Guarantees to adapters:
- *   - `writeHead` is called exactly once on the success path.
- *   - If `process()` rejects before `writeHead`, a 500 `errorResponse` is
- *     written instead (including input drain).
- *   - If `process()` resolves without calling `writeHead`, same recovery.
- *   - Post-writeHead errors propagate to the adapter for logging — the head
- *     is already committed to the transport and can't be changed.
- *
- * Mirrors the existing pattern where `handle()` wraps user errors via
- * `errorResponse`; this extends that safety net to the `process()` phase.
- */
-function wrapResponseWithSafety(inner: RestateResponse): RestateResponse {
-  return {
-    async process({ inputReader, outputWriter, writeHead, abortSignal }) {
-      let committed = false;
-      const safeWriteHead = (
-        statusCode: number,
-        headers: ResponseHeaders
-      ): void => {
-        if (committed) {
-          throw new Error("writeHead() called more than once");
-        }
-        committed = true;
-        writeHead(statusCode, headers);
-      };
-
-      try {
-        try {
-          await inner.process({
-            inputReader,
-            outputWriter,
-            writeHead: safeWriteHead,
-            abortSignal,
-          });
-        } catch (e) {
-          if (committed) {
-            // Head already on the wire — nothing structural the adapter can do.
-            throw e;
-          }
-          await errorResponse(500, ensureError(e).message).process({
-            inputReader,
-            outputWriter,
-            writeHead,
-            abortSignal,
-          });
-          return;
-        }
-
-        if (!committed) {
-          await errorResponse(
-            500,
-            "Handler did not produce a response head"
-          ).process({
-            inputReader,
-            outputWriter,
-            writeHead,
-            abortSignal,
-          });
-        }
-      } finally {
-        // Ensure the output stream closes on every path. If it was already
-        // closed (the usual case), close() rejects — which we swallow.
-        await outputWriter.close().catch(() => {});
-      }
-    },
-  };
-}
-
 /**
  * This is the RestateHandler implementation
  */
@@ -204,7 +122,7 @@ class RestateHandlerImpl implements RestateHandler {
     context?: AdditionalContext
   ): RestateResponse {
     try {
-      return wrapResponseWithSafety(this._handle(request, context));
+      return this._handle(request, context);
     } catch (e) {
       const error = ensureError(e);
       (
@@ -216,11 +134,9 @@ class RestateHandlerImpl implements RestateHandler {
       ).error(
         "Error while handling request: " + (error.stack ?? error.message)
       );
-      return wrapResponseWithSafety(
-        errorResponse(
-          error instanceof RestateError ? error.code : 500,
-          error.message
-        )
+      return errorResponse(
+        error instanceof RestateError ? error.code : 500,
+        error.message
       );
     }
   }

packages/libs/restate-sdk/src/endpoint/handlers/lambda.ts

@@ -155,8 +155,8 @@ export class LambdaHandler {
         };
       }
 
-      // wrapResponseWithSafety guarantees writeHead was called, so this
-      // resolves synchronously on the next microtask.
+      // Responses always call writeHead, so this resolves on the next
+      // microtask.
       const head = await headPromise;
 
       const responseBodyBuffer = Buffer.concat(chunks);

packages/libs/restate-sdk/src/endpoint/handlers/types.ts

@@ -36,13 +36,8 @@ export interface RestateResponse {
    *
    * Implementations own the order of things: they must call {@link writeHead}
    * exactly once before the first {@link outputWriter.write}, read any
-   * {@link inputReader} content they need, write the body, and finally call
-   * {@link outputWriter.close}.
-   *
-   * {@link RestateHandler.handle} returns responses wrapped in a safety layer
-   * that emits a 500 fallback if `process()` rejects before {@link writeHead}
-   * or resolves without committing a head, and closes the output stream on
-   * every path. Adapters can treat transport failures as logging concerns.
+   * {@link inputReader} content they need, write the body, and call
+   * {@link outputWriter.close} on every exit path (including error paths).
    */
   process(value: {
     inputReader: InputReader;

packages/libs/restate-sdk/src/endpoint/handlers/utils.ts

@@ -12,7 +12,6 @@ import type {
 import { createLogger, Logger } from "../../logging/logger.js";
 import { parseUrlComponents } from "../components.js";
 import { X_RESTATE_SERVER } from "../../user_agent.js";
-import { CompletablePromise } from "../../utils/completable_promise.js";
 
 export function tryCreateContextualLogger(
   loggerTransport: LoggerTransport,
@@ -94,24 +93,21 @@ export function emptyInputReader(): InputReader {
 
 /**
  * Bundles a `writeHead` callback with a Promise that resolves once the head
- * is committed. Used by adapters (fetch, lambda) that need to observe head
- * commit from outside the `process()` call.
- *
- * `writeHead` is expected to be called exactly once — enforced by the
- * safety layer in {@link RestateHandler.handle}, so there is no guard here.
+ * is committed. Used by adapters (fetch, lambda) that need to observe the
+ * head commit from outside the `process()` call.
  */
 export function captureHead(): {
   writeHead: (statusCode: number, headers: ResponseHeaders) => void;
   head: Promise<{ statusCode: number; headers: ResponseHeaders }>;
 } {
-  const ready = new CompletablePromise<{
+  const { promise, resolve } = Promise.withResolvers<{
     statusCode: number;
     headers: ResponseHeaders;
   }>();
   return {
     writeHead: (statusCode, headers) => {
-      ready.resolve({ statusCode, headers: { ...headers } });
+      resolve({ statusCode, headers: { ...headers } });
     },
-    head: ready.promise,
+    head: promise,
   };
 }

packages/libs/restate-sdk/src/endpoint/node_endpoint.ts

@@ -209,8 +209,8 @@ function nodeHandlerImpl(
         abortSignal: abortController.signal,
       })
       .catch((e) => {
-        // wrapResponseWithSafety guarantees writeHead is called; anything
-        // reaching here is a post-commit error. Nothing to do but log.
+        // Responses handle their own errors before rejecting; anything
+        // reaching here is an unexpected failure — just log.
         const error = ensureError(e);
         const logger =
           tryCreateContextualLogger(