fix(webapp): stop idle chat sessions erroring on their next message

Sending a message to a chat session that had been idle for a while could
fail. The session's realtime streams are created empty and auto-deleted
once they have been empty for the basin's delete-on-empty age, and a
message that arrived during the brief window while a stream was being
deleted failed with a conflict.

The default delete-on-empty age
(`REALTIME_STREAMS_BASIN_DELETE_ON_EMPTY_MIN_AGE`) is raised from 1 hour
to 30 days, matching the stream retention age, so an idle session's
streams are no longer removed out from under it. This age is now also
applied when a basin is reconfigured, not only when it is first created.
As an extra guard, a stream append that races the deletion is retried
with bounded backoff and recreates the stream instead of surfacing an
error.

Mono-RevId: de345da6508411a22df4179671e7161161d8b26b

Author: ericallam

apps/webapp/app/env.server.ts

@@ -2523,7 +2523,7 @@ const EnvironmentSchema = z
     REALTIME_STREAMS_BASIN_NAME_ENV: z.string().default("dev"),
     REALTIME_STREAMS_BASIN_DEFAULT_RETENTION: durationString().default("30d"),
     REALTIME_STREAMS_BASIN_STORAGE_CLASS: z.enum(["express", "standard"]).default("express"),
-    REALTIME_STREAMS_BASIN_DELETE_ON_EMPTY_MIN_AGE: durationString().default("1h"),
+    REALTIME_STREAMS_BASIN_DELETE_ON_EMPTY_MIN_AGE: durationString().default("30d"),
     REALTIME_STREAMS_DEFAULT_VERSION: z.enum(["v1", "v2"]).default("v1"),
     WAIT_UNTIL_TIMEOUT_MS: z.coerce.number().int().default(600_000),
 

apps/webapp/app/services/realtime/s2realtimeStreams.server.ts

@@ -609,14 +609,6 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
 
   // ---------- Internals: S2 REST ----------
   private async s2Append(stream: string, body: S2AppendInput): Promise<S2AppendAck> {
-    // POST /v1/streams/{stream}/records (JSON).
-    //
-    // Retries transient failures (network errors and 5xx) up to 3 times with
-    // exponential backoff. Undici's "fetch failed" errors observed locally
-    // are pre-connection (DNS/TCP) so the request never reaches S2, making
-    // retry safe — the alternative is a 500 surfacing to the SDK transport,
-    // which then retries the whole `/in/append` round-trip and pollutes
-    // logs. 4xx are not retried (genuine client errors).
     const url = `${this.baseUrl}/streams/${encodeURIComponent(stream)}/records`;
     const init: RequestInit = {
       method: "POST",
@@ -629,8 +621,8 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
       body: JSON.stringify(body),
     };
 
-    const maxAttempts = 3;
-    const backoffsMs = [100, 250, 600];
+    const maxAttempts = 5;
+    const backoffsMs = [500, 1000, 1500, 2000];
     let lastError: unknown;
 
     for (let attempt = 0; attempt < maxAttempts; attempt++) {
@@ -647,15 +639,17 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
         if (res.ok) {
           return (await res.json()) as S2AppendAck;
         }
-        await cancelResponseBody(res);
-        const httpError = new Error(`S2 append failed: ${res.status} ${res.statusText}`);
-        if (res.status >= 400 && res.status < 500) {
-          // 4xx — caller-side problem (auth, malformed body, closed stream).
-          // Retrying won't help.
-          throw httpError;
+        if (res.status === 409) {
+          const text = await res.text().catch(() => "");
+          lastError = new Error(`S2 append failed: 409 ${res.statusText} ${text}`.trim());
+        } else {
+          await cancelResponseBody(res);
+          const httpError = new Error(`S2 append failed: ${res.status} ${res.statusText}`);
+          if (res.status >= 400 && res.status < 500) {
+            throw httpError;
+          }
+          lastError = httpError;
         }
-        // 5xx — retryable.
-        lastError = httpError;
       }
 
       const isLastAttempt = attempt === maxAttempts - 1;

apps/webapp/app/services/realtime/streamBasinProvisioner.server.ts

@@ -105,7 +105,11 @@ async function reconfigureBasinForOrg(orgId: string, retention: string): Promise
   });
   if (!org?.streamBasinName) return;
 
-  await s2ReconfigureBasin(org.streamBasinName, { accessToken, retentionPolicy: retention });
+  await s2ReconfigureBasin(org.streamBasinName, {
+    accessToken,
+    retentionPolicy: retention,
+    deleteOnEmptyMinAge: env.REALTIME_STREAMS_BASIN_DELETE_ON_EMPTY_MIN_AGE,
+  });
 
   logger.info("[streamBasinProvisioner] reconfigured basin retention", {
     orgId,
@@ -219,13 +223,15 @@ async function s2CreateBasin(name: string, opts: CreateBasinOptions): Promise<vo
 type ReconfigureBasinOptions = {
   accessToken: string;
   retentionPolicy: string;
+  deleteOnEmptyMinAge: string;
 };
 
 async function s2ReconfigureBasin(name: string, opts: ReconfigureBasinOptions): Promise<void> {
   const url = `${env.REALTIME_STREAMS_S2_ACCOUNT_URL}/basins/${encodeURIComponent(name)}`;
   const body = {
     default_stream_config: {
       retention_policy: { age: parseDuration(opts.retentionPolicy) },
+      delete_on_empty: { min_age_secs: parseDuration(opts.deleteOnEmptyMinAge) },
     },
   };