Prevent session streams from hitting runtime limits

.changeset/calm-streams-reconnect.md

@@ -0,0 +1,5 @@
+---
+"eve": patch
+---
+
+Close long-held session stream responses before common serverless runtime limits so reconnecting clients can continue from their event cursor.

docs/guides/client/streaming.mdx

@@ -128,7 +128,7 @@ The stream can emit an interim `session.waiting` while the authorization callbac
 
 ## Reconnection
 
-HTTP connections can end before a run does. The client reconnects from the number of events already consumed, so long turns continue without replaying events. By default, a turn response keeps reconnecting until it reaches a turn boundary or is aborted, including while the turn is paused for authorization. A manually opened `session.stream()` eventually stops after repeated empty streams when it can no longer make progress.
+HTTP connections can end before a run does. eve periodically closes long-held stream responses before common serverless runtime limits, and the client reconnects from the number of events already consumed. Long turns therefore continue without replaying events or leaving the server function open until the platform kills it. By default, a turn response keeps reconnecting until it reaches a turn boundary or is aborted, including while the turn is paused for authorization. A manually opened `session.stream()` eventually stops after repeated empty streams when it can no longer make progress.
 
 If your consumer persists events, key on `event.meta.id`. It is stable across reconnects and rewinds, so an overlapping replay is safe to ingest twice. See [the event envelope](../../concepts/sessions-runs-and-streaming#the-event-envelope).
 

packages/eve/src/public/channels/eve.test.ts

@@ -327,14 +327,17 @@ function createEveCompactHandler(input: EveChannelInput) {
 }
 
 /** Creates a GET handler test harness for the durable session stream route. */
-function createEveStreamHandler(input: EveChannelInput) {
+function createEveStreamHandler(
+  input: EveChannelInput,
+  options: { readonly eventStream?: ReadableStream<unknown> } = {},
+) {
   const channel = eveChannel(input);
   const streamRoute = channel.routes.find(
     (route) => route.method === "GET" && route.path === "/eve/v1/session/:sessionId/stream",
   );
   if (!streamRoute) throw new Error("No session stream GET route found");
 
-  const getEventStream = vi.fn().mockResolvedValue(new ReadableStream());
+  const getEventStream = vi.fn().mockResolvedValue(options.eventStream ?? new ReadableStream());
   const getStreamTailIndex = vi.fn().mockResolvedValue(-1);
   const session = createMockSession({
     getEventStream,
@@ -533,6 +536,30 @@ describe("eveChannel — stream cursor", () => {
     expect(new TextDecoder().decode(firstChunk.value)).toBe("\n");
   });
 
+  it("ends a long-held response cleanly so the client can reconnect", async () => {
+    vi.useFakeTimers();
+    const cancel = vi.fn();
+    const eventStream = new ReadableStream({ cancel });
+
+    try {
+      const handler = createEveStreamHandler({ auth: none() }, { eventStream });
+      const response = await handler.fetch(
+        "https://eve.test/eve/v1/session/test-session-id/stream",
+      );
+      const reader = response.body!.getReader();
+
+      expect(new TextDecoder().decode((await reader.read()).value)).toBe("\n");
+      const nextChunk = reader.read();
+
+      await vi.advanceTimersByTimeAsync(4 * 60 * 1000);
+
+      await expect(nextChunk).resolves.toEqual({ done: true, value: undefined });
+      expect(cancel).toHaveBeenCalledWith("eve session stream lifetime elapsed");
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
   it("forwards negative tail-relative start indices", async () => {
     const handler = createEveStreamHandler({ auth: none() });
 

packages/eve/src/public/channels/eve.ts

@@ -70,6 +70,11 @@ import { parseJsonObject, type JsonObject } from "#shared/json.js";
 
 const log = createLogger("eve.channel");
 
+// Keep a margin below the five-minute function limit used by common serverless
+// deployments. The eve client treats a clean EOF as a reconnect boundary and
+// resumes from its current event index.
+const SESSION_STREAM_MAX_LIFETIME_MS = 4 * 60 * 1000;
+
 /**
  * Event-handler channel context exposed by `eveChannel({ events })`. The default eve HTTP channel
  * has no platform-specific state, so handlers receive optional continuation routing here and the
@@ -1068,9 +1073,12 @@ async function createSessionStreamResponse(request: Request, session: Session):
     if (tailIndex !== undefined) {
       headers.set(EVE_STREAM_TAIL_INDEX_HEADER, String(tailIndex));
     }
-    return new Response(serializeAsNdjson(events), {
-      headers,
-    });
+    return new Response(
+      closeStreamAfter(serializeAsNdjson(events), SESSION_STREAM_MAX_LIFETIME_MS),
+      {
+        headers,
+      },
+    );
   } catch {
     return Response.json({ error: "Session not found.", ok: false }, { status: 404 });
   }
@@ -1365,3 +1373,44 @@ function serializeAsNdjson(events: ReadableStream<unknown>): ReadableStream<Uint
     }),
   );
 }
+
+function closeStreamAfter<T>(stream: ReadableStream<T>, durationMs: number): ReadableStream<T> {
+  const reader = stream.getReader();
+  let finished = false;
+  let timer: ReturnType<typeof setTimeout> | undefined;
+
+  return new ReadableStream<T>({
+    start(controller) {
+      timer = setTimeout(() => {
+        if (finished) return;
+        finished = true;
+        controller.close();
+        void reader.cancel("eve session stream lifetime elapsed").catch(() => undefined);
+      }, durationMs);
+    },
+    async pull(controller) {
+      try {
+        const result = await reader.read();
+        if (finished) return;
+        if (result.done) {
+          finished = true;
+          if (timer !== undefined) clearTimeout(timer);
+          controller.close();
+          return;
+        }
+        controller.enqueue(result.value);
+      } catch (error) {
+        if (finished) return;
+        finished = true;
+        if (timer !== undefined) clearTimeout(timer);
+        controller.error(error);
+      }
+    },
+    async cancel(reason) {
+      if (finished) return;
+      finished = true;
+      if (timer !== undefined) clearTimeout(timer);
+      await reader.cancel(reason);
+    },
+  });
+}