fix(chat): preserve recovered append sequence

Author: gtremper

apps/webapp/app/routes/realtime.v1.sessions.$session.$io.append.ts

@@ -181,21 +181,24 @@ const { action, loader } = createActionApiRoute(
           error: recoveryError,
         });
       } else if (recoveredSeq !== undefined) {
-        const recovered = await commitSessionStreamPart(
+        await commitSessionStreamPart(
           authentication.environment.id,
           addressingKey,
           params.io,
           partId,
           pendingClaim.claimValue,
           recoveredSeq
         );
-        if (recovered) {
-          appendSeq = recoveredSeq;
-          claim = { status: "committed", seq: recoveredSeq };
-        }
+        // The S2 record is durable even if Redis expired or evicted the
+        // pending claim before the best-effort sequence write-back.
+        appendSeq = recoveredSeq;
+        claim = { status: "committed", seq: recoveredSeq };
       }
 
-      if (claim.status === "pending") {
+      // `.out` data may have been trimmed after it was accepted, so a
+      // successful search with no record preserves the prior no-op response.
+      // A failed search proves nothing and remains retryable for both channels.
+      if (claim.status === "pending" && (params.io === "in" || recoveryError)) {
         return json(
           { ok: false, error: "This append is still in progress, please retry." },
           { status: 409, headers: { "Retry-After": "1" } }

apps/webapp/app/services/sessionStreamWaitpointCache.server.ts

@@ -163,10 +163,14 @@ const COMMIT_APPEND_DEDUPE_SCRIPT = `
   if current == ARGV[2] then
     return 1
   end
-  if current ~= ARGV[1] then
+  if current and current ~= ARGV[1] then
     return 0
   end
+  local restoredMissingClaim = not current
   redis.call("SET", KEYS[1], ARGV[2], "EX", ARGV[3])
+  if restoredMissingClaim then
+    return 2
+  end
   return 1
 `;
 
@@ -276,7 +280,7 @@ function parsePendingClaimedAt(value: string): number | undefined {
   return Number.isSafeInteger(claimedAt) && claimedAt >= 0 ? claimedAt : undefined;
 }
 
-/** Replace the exact pending claim with the committed S2 sequence. */
+/** Persist a committed S2 sequence if the expected claim still owns the key or the key vanished. */
 export async function commitSessionStreamPart(
   environmentId: string,
   addressingKey: string,
@@ -296,7 +300,7 @@ export async function commitSessionStreamPart(
       `${APPEND_DEDUPE_SEQUENCE_PREFIX}${seq}`,
       String(APPEND_DEDUPE_TTL_SECONDS)
     );
-    return result === 1;
+    return result === 1 || result === 2;
   } catch (error) {
     logger.error("Failed to commit session stream append part", {
       environmentId,

apps/webapp/app/v3/webhookEngine.server.ts

@@ -234,17 +234,17 @@ function createWebhookEngine() {
             pendingClaim.claimedAt
           );
           if (recoveredSeq !== undefined) {
-            const recovered = await commitSessionStreamPart(
+            await commitSessionStreamPart(
               environment.id,
               addressingKey,
               "in",
               deliveryId,
               pendingClaim.claimValue,
               recoveredSeq
             );
-            if (recovered) {
-              claim = { status: "committed", seq: recoveredSeq };
-            }
+            // The S2 record is durable even if Redis expired or evicted the
+            // pending claim before the best-effort sequence write-back.
+            claim = { status: "committed", seq: recoveredSeq };
           }
 
           if (claim.status === "pending") {

apps/webapp/test/helpers/sessionStream.ts

@@ -122,8 +122,9 @@ export async function appendInput(opts: {
   body: string;
   partId?: string;
   origin?: string;
+  io?: "out" | "in";
 }): Promise<{ status: number; acao: string | null; json: unknown }> {
-  const url = `${sessionChannelUrl(opts.baseUrl, opts.addressingKey, "in")}/append`;
+  const url = `${sessionChannelUrl(opts.baseUrl, opts.addressingKey, opts.io ?? "in")}/append`;
   const res = await fetch(url, {
     method: "POST",
     headers: {

apps/webapp/test/session-stream.e2e.test.ts

@@ -82,6 +82,7 @@ async function setupSession() {
   return {
     addressingKey,
     environmentId: environment.id,
+    apiKey,
     token,
     producer,
     inProducer,
@@ -478,7 +479,115 @@ describe("session stream e2e", () => {
     }
   });
 
-  it("E16 subscribe with an invalid token is rejected", async () => {
+  it("E16 in/append recovery survives an expired Redis claim", async () => {
+    const { addressingKey, environmentId, token, inProducer, baseUrl } = await setupSession();
+    const redis = new Redis({ ...server.redis, keyPrefix: "tr:" });
+    const monitorSource = new Redis(server.redis);
+    let monitor: Redis | undefined;
+
+    try {
+      const payload = JSON.stringify({ kind: "message", text: "expired during recovery" });
+      const partId = `expired-${randomBytes(6).toString("hex")}`;
+      const claimKey = `ssa:${encodeURIComponent(environmentId)}:${encodeURIComponent(
+        addressingKey
+      )}:in:${encodeURIComponent(partId)}`;
+      const waitpointKey = `ssw:${environmentId}:${addressingKey}:in`;
+
+      await redis.set(claimKey, `pending:${Date.now()}:crashed-owner`, "EX", 5 * 60);
+      await redis.sadd(waitpointKey, "waitpoint_expired_claim_test");
+      const originalSeq = await inProducer.appendData(payload, partId);
+
+      monitor = await monitorSource.monitor();
+      const claimRead = new Promise<void>((resolve, reject) => {
+        const timeout = setTimeout(
+          () => reject(new Error("Timed out waiting for claim read")),
+          5_000
+        );
+        const onMonitor = (_time: string, args: string[]) => {
+          if (args[0]?.toLowerCase() !== "get" || args[1] !== `tr:${claimKey}`) return;
+          monitor!.off("monitor", onMonitor);
+          void redis.del(claimKey).then(
+            () => {
+              clearTimeout(timeout);
+              resolve();
+            },
+            (error) => {
+              clearTimeout(timeout);
+              reject(error);
+            }
+          );
+        };
+        monitor.on("monitor", onMonitor);
+      });
+
+      const retryPromise = appendInput({ baseUrl, addressingKey, token, partId, body: payload });
+      await claimRead;
+      const retry = await retryPromise;
+
+      expect(retry.status).toBe(200);
+      expect(retry.json).toEqual({ ok: true, seq: originalSeq });
+      expect(await redis.get(claimKey)).toBe(`seq:${originalSeq}`);
+      expect(await redis.exists(waitpointKey)).toBe(0);
+
+      const committedRetry = await appendInput({
+        baseUrl,
+        addressingKey,
+        token,
+        partId,
+        body: payload,
+      });
+      expect(committedRetry.status).toBe(200);
+      expect(committedRetry.json).toEqual({ ok: true, seq: originalSeq });
+
+      const { parts } = await collectSessionOut({
+        baseUrl,
+        addressingKey,
+        token,
+        io: "in",
+        timeoutInSeconds: 1,
+        maxMs: 5_000,
+      });
+      expect(parts.filter((part) => part.chunk != null)).toHaveLength(1);
+    } finally {
+      monitor?.disconnect();
+      await monitorSource.quit();
+      await redis.quit();
+    }
+  });
+
+  it("E17 out/append keeps a missing recovered record as a successful no-op", async () => {
+    const { addressingKey, environmentId, apiKey, baseUrl } = await setupSession();
+    const redis = new Redis({ ...server.redis, keyPrefix: "tr:" });
+
+    try {
+      const payload = JSON.stringify({ kind: "message", text: "already trimmed" });
+      const partId = `trimmed-${randomBytes(6).toString("hex")}`;
+      const claimKey = `ssa:${encodeURIComponent(environmentId)}:${encodeURIComponent(
+        addressingKey
+      )}:out:${encodeURIComponent(partId)}`;
+      const waitpointKey = `ssw:${environmentId}:${addressingKey}:out`;
+
+      await redis.set(claimKey, `pending:${Date.now()}:crashed-owner`, "EX", 5 * 60);
+      await redis.sadd(waitpointKey, "waitpoint_trimmed_claim_test");
+
+      const retry = await appendInput({
+        baseUrl,
+        addressingKey,
+        token: apiKey,
+        partId,
+        body: payload,
+        io: "out",
+      });
+
+      expect(retry.status).toBe(200);
+      expect(retry.json).toEqual({ ok: true });
+      expect(await redis.exists(waitpointKey)).toBe(0);
+    } finally {
+      await redis.quit();
+    }
+  });
+
+  it("E18 subscribe with an invalid token is rejected", async () => {
     const { addressingKey, baseUrl } = await setupSession();
 
     const { status } = await openChannelRaw({