feat: add inactivity timeout

Author: tilyupo

src/index.ts

@@ -296,31 +296,58 @@ class ReplaneRemoteStorage implements ReplaneStorage {
   private async *startReplicationStreamImpl(
     options: StartReplicationStreamReplaneStorageOptions
   ): AsyncIterable<ReplicationStreamRecord> {
-    const rawEvents = fetchSse({
-      fetchFn: options.fetchFn,
-      headers: {
-        Authorization: this.getAuthHeader(options),
-        "Content-Type": "application/json",
-      },
-      body: JSON.stringify(options.getBody()),
-      timeoutMs: options.requestTimeoutMs,
-      method: "POST",
-      signal: options.signal,
-      url: this.getApiEndpoint(`/sdk/v1/replication/stream`, options),
-      onConnect: options.onConnect,
-    });
+    // Create an abort controller for inactivity timeout
+    const inactivityAbortController = new AbortController();
+    const { signal: combinedSignal, cleanUpSignals } = options.signal
+      ? combineAbortSignals([options.signal, inactivityAbortController.signal])
+      : { signal: inactivityAbortController.signal, cleanUpSignals: () => {} };
+
+    let inactivityTimer: ReturnType<typeof setTimeout> | null = null;
+
+    const resetInactivityTimer = () => {
+      if (inactivityTimer) clearTimeout(inactivityTimer);
+      inactivityTimer = setTimeout(() => {
+        inactivityAbortController.abort();
+      }, options.inactivityTimeoutMs);
+    };
+
+    try {
+      const rawEvents = fetchSse({
+        fetchFn: options.fetchFn,
+        headers: {
+          Authorization: this.getAuthHeader(options),
+          "Content-Type": "application/json",
+        },
+        body: JSON.stringify(options.getBody()),
+        timeoutMs: options.requestTimeoutMs,
+        method: "POST",
+        signal: combinedSignal,
+        url: this.getApiEndpoint(`/sdk/v1/replication/stream`, options),
+        onConnect: () => {
+          resetInactivityTimer();
+          options.onConnect?.();
+        },
+      });
+
+      for await (const sseEvent of rawEvents) {
+        resetInactivityTimer();
+
+        if (sseEvent.type === "ping") continue;
 
-    for await (const rawEvent of rawEvents) {
-      const event = JSON.parse(rawEvent);
-      if (
-        typeof event === "object" &&
-        event !== null &&
-        "type" in event &&
-        typeof event.type === "string" &&
-        (SUPPORTED_REPLICATION_STREAM_RECORD_TYPES as unknown as string[]).includes(event.type)
-      ) {
-        yield event as ReplicationStreamRecord;
+        const event = JSON.parse(sseEvent.payload);
+        if (
+          typeof event === "object" &&
+          event !== null &&
+          "type" in event &&
+          typeof event.type === "string" &&
+          (SUPPORTED_REPLICATION_STREAM_RECORD_TYPES as unknown as string[]).includes(event.type)
+        ) {
+          yield event as ReplicationStreamRecord;
+        }
       }
+    } finally {
+      if (inactivityTimer) clearTimeout(inactivityTimer);
+      cleanUpSignals();
     }
   }
 
@@ -374,6 +401,12 @@ export interface ReplaneClientOptions<T extends Configs> {
    * @default 200
    */
   retryDelayMs?: number;
+  /**
+   * Timeout in ms for SSE connection inactivity.
+   * If no events (including pings) are received within this time, the connection will be re-established.
+   * @default 60000
+   */
+  inactivityTimeoutMs?: number;
   /**
    * Optional logger (defaults to console).
    */
@@ -428,6 +461,7 @@ interface ReplaneFinalOptions {
   fetchFn: typeof fetch;
   requestTimeoutMs: number;
   initializationTimeoutMs: number;
+  inactivityTimeoutMs: number;
   sdkKey: string;
   logger: ReplaneLogger;
   retryDelayMs: number;
@@ -725,6 +759,7 @@ function toFinalOptions<T extends Configs>(defaults: ReplaneClientOptions<T>): R
       globalThis.fetch.bind(globalThis),
     requestTimeoutMs: defaults.requestTimeoutMs ?? 2000,
     initializationTimeoutMs: defaults.initializationTimeoutMs ?? 5000,
+    inactivityTimeoutMs: defaults.inactivityTimeoutMs ?? 60_000,
     logger: defaults.logger ?? console,
     retryDelayMs: defaults.retryDelayMs ?? 200,
     context: {
@@ -775,6 +810,8 @@ async function fetchWithTimeout(
 
 const SSE_DATA_PREFIX = "data:";
 
+type SseEvent = { type: "ping" } | { type: "data"; payload: string };
+
 async function* fetchSse(params: {
   fetchFn: typeof fetch;
   url: string;
@@ -784,7 +821,7 @@ async function* fetchSse(params: {
   method?: string;
   signal?: AbortSignal;
   onConnect?: () => void;
-}) {
+}): AsyncGenerator<SseEvent> {
   const abortController = new AbortController();
   const { signal, cleanUpSignals } = params.signal
     ? combineAbortSignals([params.signal, abortController.signal])
@@ -841,10 +878,15 @@ async function* fetchSse(params: {
         for (const frame of frames) {
           // Parse lines inside a single SSE event frame
           const dataLines: string[] = [];
+          let isPing = false;
 
           for (const rawLine of frame.split(/\r?\n/)) {
             if (!rawLine) continue;
-            if (rawLine.startsWith(":")) continue; // comment/keepalive
+            if (rawLine.startsWith(":")) {
+              // comment/keepalive - treat as ping
+              isPing = true;
+              continue;
+            }
 
             if (rawLine.startsWith(SSE_DATA_PREFIX)) {
               // Keep leading space after "data:" if present per spec
@@ -855,7 +897,9 @@ async function* fetchSse(params: {
 
           if (dataLines.length) {
             const payload = dataLines.join("\n");
-            yield payload;
+            yield { type: "data", payload };
+          } else if (isPing) {
+            yield { type: "ping" };
           }
         }
       }

tests/index.spec.ts

@@ -1981,6 +1981,126 @@ describe("createReplaneClient", () => {
       expect(client.get("config1")).toBe("initial");
     });
   });
+
+  describe("inactivity timeout", () => {
+    async function waitFor(
+      condition: () => boolean,
+      timeoutMs: number = 1000,
+      intervalMs: number = 10
+    ): Promise<void> {
+      const start = Date.now();
+      while (!condition()) {
+        if (Date.now() - start > timeoutMs) {
+          throw new Error("waitFor timed out");
+        }
+        await new Promise((resolve) => setTimeout(resolve, intervalMs));
+      }
+    }
+
+    it("should reconnect when no events are received within inactivity timeout", async () => {
+      const inactivityTimeoutMs = 50;
+      clientPromise = createClient({ inactivityTimeoutMs });
+
+      // First connection
+      const connection1 = await mockServer.acceptConnection();
+      await connection1.push({
+        type: "init",
+        configs: [{ name: "config1", overrides: [], version: 1, value: "initial" }],
+      });
+
+      const client = await clientPromise;
+      await sync();
+      expect(client.get("config1")).toBe("initial");
+
+      // Wait for connection to be aborted due to inactivity
+      await waitFor(() => connection1.aborted, inactivityTimeoutMs + 100);
+      expect(connection1.aborted).toBe(true);
+
+      // Client should reconnect - accept the new connection
+      const connection2 = await mockServer.acceptConnection();
+      await connection2.push({
+        type: "init",
+        configs: [{ name: "config1", overrides: [], version: 2, value: "reconnected" }],
+      });
+      await sync();
+
+      expect(client.get("config1")).toBe("reconnected");
+
+      client.close();
+    });
+
+    it("should reset inactivity timer when ping is received", async () => {
+      const inactivityTimeoutMs = 100;
+      clientPromise = createClient({ inactivityTimeoutMs });
+
+      const connection = await mockServer.acceptConnection();
+      await connection.push({
+        type: "init",
+        configs: [{ name: "config1", overrides: [], version: 1, value: "initial" }],
+      });
+
+      const client = await clientPromise;
+      await sync();
+      expect(client.get("config1")).toBe("initial");
+
+      // Wait for some time, but less than inactivity timeout
+      await new Promise((resolve) => setTimeout(resolve, inactivityTimeoutMs / 2));
+      expect(connection.aborted).toBe(false);
+
+      // Send a ping to reset the timer
+      await connection.ping();
+      await sync();
+
+      // Wait again, but less than inactivity timeout from the ping
+      await new Promise((resolve) => setTimeout(resolve, inactivityTimeoutMs / 2));
+      expect(connection.aborted).toBe(false);
+
+      // Now wait past the inactivity timeout from the ping
+      await waitFor(() => connection.aborted, inactivityTimeoutMs + 50);
+      expect(connection.aborted).toBe(true);
+
+      client.close();
+    });
+
+    it("should reset inactivity timer when data event is received", async () => {
+      const inactivityTimeoutMs = 100;
+      clientPromise = createClient({ inactivityTimeoutMs });
+
+      const connection = await mockServer.acceptConnection();
+      await connection.push({
+        type: "init",
+        configs: [{ name: "config1", overrides: [], version: 1, value: "initial" }],
+      });
+
+      const client = await clientPromise;
+      await sync();
+
+      // Wait for some time, but less than inactivity timeout
+      await new Promise((resolve) => setTimeout(resolve, inactivityTimeoutMs / 2));
+      expect(connection.aborted).toBe(false);
+
+      // Send a config change to reset the timer
+      await connection.push({
+        type: "config_change",
+        name: "config1",
+        overrides: [],
+        version: 2,
+        value: "updated",
+      });
+      await sync();
+      expect(client.get("config1")).toBe("updated");
+
+      // Wait again, but less than inactivity timeout from the event
+      await new Promise((resolve) => setTimeout(resolve, inactivityTimeoutMs / 2));
+      expect(connection.aborted).toBe(false);
+
+      // Now wait past the inactivity timeout from the event
+      await waitFor(() => connection.aborted, inactivityTimeoutMs + 50);
+      expect(connection.aborted).toBe(true);
+
+      client.close();
+    });
+  });
 });
 
 describe("createInMemoryReplaneClient", () => {

tests/utils.ts

@@ -20,7 +20,7 @@ export interface ReplaneServerMockHandler {
   startReplicationStream: (
     body: StartReplicationStreamBody,
     signal?: AbortSignal
-  ) => AsyncIterable<ReplicationStreamRecord>;
+  ) => AsyncIterable<MockConnectionEvent>;
 }
 
 export function createReplaneServerMock(handler: ReplaneServerMockHandler) {
@@ -59,8 +59,11 @@ export function createReplaneServerMock(handler: ReplaneServerMockHandler) {
 
           for await (const event of replicationStream) {
             if (signal?.aborted) break;
-            controller.enqueue({ type: "data", data: JSON.stringify(event) });
-            controller.enqueue({ type: "ping" });
+            if (event.type === "ping") {
+              controller.enqueue({ type: "ping" });
+            } else {
+              controller.enqueue({ type: "data", data: JSON.stringify(event.record) });
+            }
           }
 
           if (!signal?.aborted) {
@@ -122,7 +125,9 @@ export class MockReplaneServerController {
         const connection = new MockReplaneServerConnection(body, signal);
         self.knownConnections.add(connection);
         await self.connections.push(connection);
-        yield* connection.events;
+        for await (const event of connection.events) {
+          yield event;
+        }
       },
     });
   }
@@ -140,8 +145,12 @@ export class MockReplaneServerController {
   }
 }
 
+type MockConnectionEvent =
+  | { type: "data"; record: ReplicationStreamRecord }
+  | { type: "ping" };
+
 export class MockReplaneServerConnection {
-  private readonly _events = new Channel<ReplicationStreamRecord>();
+  private readonly _events = new Channel<MockConnectionEvent>();
   private _closed = false;
 
   constructor(
@@ -171,7 +180,7 @@ export class MockReplaneServerConnection {
     return this._closed;
   }
 
-  get events(): AsyncIterable<ReplicationStreamRecord> {
+  get events(): AsyncIterable<MockConnectionEvent> {
     return this._events;
   }
 
@@ -181,7 +190,12 @@ export class MockReplaneServerConnection {
 
   async push(event: ReplicationStreamRecord) {
     if (this._closed) return;
-    await this._events.push(event);
+    await this._events.push({ type: "data", record: event });
+  }
+
+  async ping() {
+    if (this._closed) return;
+    await this._events.push({ type: "ping" });
   }
 
   async throw(error: Error) {