Feature request: keep producer subscription alive after source stream ends to serve late resume requests

Problem

When the source stream finishes, the producer immediately sets the sentinel to DONE and unsubscribes from the request channel. After that, any call to resumeStream returns null, even though the producer process is still alive and holds the complete chunk buffer in memory.

This creates a blind spot in any application that does post-stream work (database writes, analytics, cleanup) after the source stream ends but before the process exits. A client that disconnects and reconnects during that window gets null from resumeStream, even though the producer could serve the full replay.

Timeline of the problem

t0  Source stream starts. Producer subscribes to request channel.
    Chunks are buffered in memory and forwarded to any active listeners.

t1  Source stream ends.
    - isDone = true
    - sentinel = DONE          ← blocks all future resumeStream calls
    - unsubscribe(request)     ← kills the callback that could replay chunks
    - DONE_MESSAGE sent to any currently-connected listeners

t2  Producer continues with post-stream work (1-30 seconds).
    Process is alive. Redis clients are connected. chunks[] is in memory.

t3  Client reconnects and calls resumeStream.
    → resumeExistingStream checks sentinel → DONE → returns null.
    Client gets nothing despite the producer being fully capable of serving.

t4  Producer finishes post-stream work (e.g. database write commits).

t5  Producer process exits or Redis clients are closed.
    Now returning null is correct. the buffer is gone.

The gap between t1 and t5 is the problem window. The library treats the stream as dead at t1, but the producer is alive until t5.

What the client actually needs

The complete chunk buffer (including DONE_MESSAGE) that the producer already holds in memory at t3. The data is there. the library just blocks access to it.

Existing code that almost works

The subscribe callback in createNewResumableStream (runtime.ts:145-165) already handles the isDone case:

await ctx.subscriber.subscribe(
  `${ctx.keyPrefix}:request:${streamId}`,
  async (message: string) => {
    const parsedMessage = JSON.parse(message);
    listenerChannels.push(parsedMessage.listenerId);
    const chunksToSend = chunks.join("").slice(parsedMessage.skipCharacters || 0);

    const promises = [];
    promises.push(
      ctx.publisher.publish(`${ctx.keyPrefix}:chunk:${parsedMessage.listenerId}`, chunksToSend)
    );
    if (isDone) {
      // Late-arriving listener gets the full buffer PLUS the done signal
      promises.push(
        ctx.publisher.publish(`${ctx.keyPrefix}:chunk:${parsedMessage.listenerId}`, DONE_MESSAGE)
      );
    }
    await Promise.all(promises);
  }
);

When isDone === true, a resume request arriving via pub/sub receives the full chunk buffer replayed from memory, followed by DONE_MESSAGE. The client would see the complete stream as if it had been connected the whole time.

This logic is already written and correct. It just never runs after the source ends because of the two lines that fire simultaneously (runtime.ts:181-195):

if (done) {
  isDone = true;
  controller.close();

  const promises = [];
  // 1. Sentinel set to DONE. makes resumeExistingStream return null
  //    without ever trying the pub/sub path
  promises.push(
    ctx.publisher.set(`${ctx.keyPrefix}:sentinel:${streamId}`, DONE_VALUE, {
      EX: 24 * 60 * 60,
    })
  );
  // 2. Unsubscribe. removes the callback that already knows how to
  //    replay the buffer to late-arriving listeners
  promises.push(
    ctx.subscriber.unsubscribe(`${ctx.keyPrefix}:request:${streamId}`)
  );
  for (const listenerId of listenerChannels) {
    promises.push(
      ctx.publisher.publish(`${ctx.keyPrefix}:chunk:${listenerId}`, DONE_MESSAGE)
    );
  }
  await Promise.all(promises);
  streamDoneResolver?.();
}

1. publisher.set(sentinelKey, DONE_VALUE). causes resumeExistingStream (runtime.ts:117-124) to short-circuit to null at the sentinel check, before attempting the pub/sub path.
2. subscriber.unsubscribe(requestChannel). removes the callback entirely, so even if a resume request somehow bypassed the sentinel, no one would answer.

Proposed change

Defer the sentinel update and unsubscribe so the producer can continue serving resume requests from its in-memory chunk buffer for as long as the process is alive and the Redis clients are connected.

Option A: expose a cleanup() function to the caller

When the source stream ends, only set isDone = true (which already controls the DONE_MESSAGE logic in the callback). Do NOT set sentinel to DONE and do NOT unsubscribe.

Instead, return (or expose via the context) a cleanup() function that the caller invokes when the process is ready to shut down:

async function cleanup() {
  await publisher.set(sentinelKey, DONE_VALUE, { EX: 24 * 60 * 60 });
  await subscriber.unsubscribe(requestChannel);
}

The caller controls when the producer goes dark. The library already has a waitUntil mechanism that could be wired to this. streamDoneResolver could be deferred until cleanup() is called instead of resolving when the source ends.

Important: streamDoneResolver (line 196) currently resolves when the source ends, which signals waitUntil that the work is done. If cleanup is deferred, streamDoneResolver must also be deferred — otherwise serverless runtimes may kill the process before cleanup() runs, defeating the purpose.

Pros: caller has full control over the producer's lifetime. No new states.
Cons: requires the caller to call cleanup() explicitly. If they don't, the subscription leaks until Redis client close or TTL expiry (the sentinel TTL already acts as a safety net).

Option B: introduce an intermediate sentinel state

Instead of going directly from the initial value ("1") to "DONE", introduce a third state (e.g. "FLUSHED") that means "source has ended, but the producer may still be warm."

When the source stream ends:

• Set sentinel to "FLUSHED" (instead of "DONE").
• Keep the request channel subscription alive.

In resumeExistingStream:

• null → stream never existed → return undefined.
• "FLUSHED" → stream finished but producer may be warm → attempt pub/sub resume via the existing resumeStream function. If the 1-second ack timeout fires (producer is dead), fall back to returning null. Note: the timeout handler (runtime.ts:265-275) currently only special-cases DONE_VALUE to resolve null; a FLUSHED sentinel would fall through to the "Timeout waiting for ack" error. The timeout handler must be updated to treat FLUSHED the same as DONE (resolve null).
• "DONE" → producer is fully gone → return null immediately (current behavior).

The sentinel transitions to "DONE" when:

• The caller explicitly calls cleanup(), OR
• The sentinel's Redis TTL expires (safety net. existing 24h TTL works here).

Pros: backward-compatible for callers that don't know about the new state. Falls back cleanly via the ack timeout when the producer is dead. No new API surface required.
Cons: adds a small latency (the ack timeout duration) on the null path when the producer is truly gone and the sentinel is "FLUSHED".

Option C: always attempt pub/sub before returning null on DONE

The simplest option: when resumeExistingStream sees sentinel === DONE, instead of returning null immediately, attempt the pub/sub path with the existing 1-second timeout. If the producer is still alive and subscribed, it responds with buffered chunks. If not, the timeout fires and we return null as before.

This requires:

• NOT unsubscribing from the request channel when the source ends (the subscription stays alive until the Redis client closes).
• Changing resumeExistingStream to try pub/sub even when sentinel is DONE.

 async function resumeExistingStream(initPromise, ctx, streamId, skipCharacters) {
   await initPromise;
   const state = await ctx.publisher.get(`${ctx.keyPrefix}:sentinel:${streamId}`);
   if (!state) return undefined;
-  if (state === DONE_VALUE) return null;
+  // Don't short-circuit. the producer may still be alive with a warm buffer.
+  // resumeStream will timeout after 1s if nobody answers.
   return resumeStream(ctx, streamId, skipCharacters);
 }

Pros: smallest diff. No new states, no new API. Falls back via the existing ack timeout.
Cons: every DONE resume pays the 1-second timeout cost when the producer is truly dead. May not be acceptable if resume is called frequently (e.g. client retry loops).

Expected behavior after the change

t0  Source stream starts. Sentinel = "1". Producer subscribes to request channel.
t1  Source stream ends.
    - isDone = true
    - Sentinel stays "1" (or transitions to "FLUSHED" in Option B)
    - Request channel subscription stays alive
    - DONE_MESSAGE sent to any currently-connected listeners
t2  Producer does post-stream work. Process alive, Redis connected, chunks[] in memory.
t3  Client reconnects → resumeStream:
    - Sentinel is not DONE → pub/sub path
    - Publish request → producer callback fires
    - Full chunk buffer replayed + DONE_MESSAGE
    - Client receives complete stream ✓
t4  Producer calls cleanup() (or process exits, or Redis client closes):
    - Sentinel = DONE
    - Unsubscribe from request channel
t5  Late resume after t4:
    - Sentinel = DONE → null (Option A/B) or pub/sub timeout → null (Option C)
    - Correct. the buffer is gone

Summary

In any environment where the producer process outlives the HTTP response (serverless functions, long-running servers with deferred writes, background job processors), there is a window between "stream finished" and "process shutdown" where the producer holds authoritative data that no other system has yet. The current design discards access to that data the instant the source stream ends.

The practical consequence is a race condition: the client reconnects before the producer's database write commits, gets null from resumeStream, fetches from the database, and sees stale data. The data the client needs is sitting in the producer's memory. the library just won't serve it.

Keeping the producer's pub/sub subscription alive for the full lifetime of the process (not just the lifetime of the source stream) eliminates this race by serving data from the authoritative source (the producer's in-memory buffer) instead of the eventually-consistent replica (the database).

Start in src/runtime.ts, especially createNewResumableStream around lines 145-195 and resumeExistingStream around lines 110-125, then inspect the timeout handling around lines 265-275. Decide with maintainers which lifecycle option to implement, and verify that late resumes replay buffered chunks while resumes after producer cleanup still terminate correctly.