J3: HttpSseTransport event IDs, replay dedupe, 410 gap -> resume

src/transport.ts

@@ -17,6 +17,12 @@ export interface TransportEvents {
   error: (err: Error) => void;
   close: () => void;
   open: () => void;
+  /**
+   * The server answered a reconnect with 410 (Transport Section 4.3.4): events
+   * were lost. The transport has already discarded its cursor and opened a new
+   * stream with none; the application (Peer) now resumes the session.
+   */
+  gap: () => void;
 }
 
 export type TransportEventName = keyof TransportEvents;
@@ -160,8 +166,13 @@ export class WebSocketTransport extends Emitter implements Transport {
 export interface HttpSseTransportOptions {
   /** Authorization header value (e.g., "Bearer <token>"). */
   authorization?: string;
-  /** Last-Event-ID for SSE replay. */
+  /**
+   * Initial SSE cursor (the ID of the last event already processed). It is
+   * sent on the first connectSSE() and replaces itself as events arrive.
+   */
   lastEventId?: string;
+  /** How many processed event IDs to remember for replay dedupe. Default: 4096. */
+  dedupeWindow?: number;
   /** Custom fetch implementation (default: global fetch). */
   fetch?: typeof globalThis.fetch;
   /** Custom EventSource constructor (for polyfills that support headers). */
@@ -183,6 +194,12 @@ export class HttpSseTransport extends Emitter implements Transport {
   private fetchFn: typeof globalThis.fetch;
   private EventSourceCtor: typeof EventSource;
   private closed = false;
+  private cursor?: string;
+  private seenIds = new Set<string>();
+  private dedupeWindow: number;
+  private pendingGap = false;
+  /** The event stream was open when the session ended: reopen it for the next session once bound. */
+  private reopenOnBind = false;
 
   constructor(
     rpcUrl: string,
@@ -195,6 +212,17 @@ export class HttpSseTransport extends Emitter implements Transport {
     this.authorization = opts?.authorization;
     this.fetchFn = opts?.fetch ?? globalThis.fetch.bind(globalThis);
     this.EventSourceCtor = opts?.EventSource ?? globalThis.EventSource;
+    this.cursor = opts?.lastEventId || undefined;
+    this.dedupeWindow = opts?.dedupeWindow ?? 4096;
+  }
+
+  /**
+   * The ID of the last event this transport delivered (the SSE cursor), or
+   * undefined when none was received or the cursor was discarded after a gap.
+   * Event IDs are opaque: store and send them back, never parse or compare.
+   */
+  get lastEventId(): string | undefined {
+    return this.cursor;
   }
 
   /**
@@ -211,75 +239,197 @@ export class HttpSseTransport extends Emitter implements Transport {
    * - `session_id` query parameter on SSE connections
    */
   setSessionId(sessionId: string): void {
+    let reopen = this.reopenOnBind;
+    if (this.sessionId !== sessionId) {
+      const hadSession = this.sessionId !== undefined;
+      // A stream is tied to the session it was opened for (its events are ignored
+      // once that is no longer the bound one), so a stream that is open (also one
+      // opened before the first binding) is closed and reopened for the new session.
+      if (this.eventSource !== null) {
+        reopen = true;
+        this.eventSource.close();
+        this.eventSource = null;
+      }
+      if (hadSession) {
+        // Event IDs are unique only within a session: another session's cursor
+        // would draw a 410 and its seen IDs could swallow this session's events.
+        // (The first binding keeps a cursor the transport was seeded with.)
+        this.cursor = undefined;
+        this.seenIds.clear();
+        this.pendingGap = false;
+      }
+    }
     this.sessionId = sessionId;
+    if (reopen && !this.closed && this.eventSource === null) {
+      this.reopenOnBind = false;
+      this.openStream(this.cursor);
+    }
   }
 
   /**
-   * Connect the SSE stream. Call this after constructing the transport.
-   * Separated from the constructor so callers can register listeners first.
+   * Forget the bound session and its replay state and close the event stream: the
+   * session ended. A later session starts clean (no stale Wmp-Session-Id, cursor
+   * or seen ids).
    */
-  connectSSE(lastEventId?: string): void {
+  clearSession(): void {
+    // If a stream was open it is reopened when the next session is bound.
+    this.reopenOnBind = this.reopenOnBind || this.eventSource !== null;
+    this.sessionId = undefined;
+    this.cursor = undefined;
+    this.seenIds.clear();
+    this.pendingGap = false;
     this.eventSource?.close();
     this.eventSource = null;
+  }
 
-    this.setupEventSource(lastEventId);
+  /**
+   * Connect the SSE stream. Call this after constructing the transport.
+   * Separated from the constructor so callers can register listeners first.
+   * Resumes from the transport's own cursor unless `lastEventId` is given.
+   */
+  connectSSE(lastEventId?: string): void {
+    if (lastEventId !== undefined) this.cursor = lastEventId || undefined;
+    this.openStream(this.cursor);
   }
 
   /**
-   * Reconnect the SSE stream with the current authorization/session state.
-   * Useful after calling setAuthorization.
+   * Reconnect the SSE stream with the current authorization/session state
+   * and the recorded cursor. Useful after calling setAuthorization.
    */
   reconnectSSE(lastEventId?: string): void {
     this.connectSSE(lastEventId);
   }
 
-  private setupEventSource(lastEventId?: string): void {
-    // Build the SSE URL with session_id (required for tenant routing)
-    // and optional lastEventId for event replay.
-    let url = this.eventsUrl;
-    if (this.sessionId) {
-      const sep = url.includes("?") ? "&" : "?";
-      url += `${sep}session_id=${encodeURIComponent(this.sessionId)}`;
-    }
-    if (lastEventId) {
-      const sep = url.includes("?") ? "&" : "?";
-      url += `${sep}lastEventId=${encodeURIComponent(lastEventId)}`;
+  private streamHeaders(cursor?: string): Record<string, string> {
+    const h: Record<string, string> = {};
+    if (this.authorization) h["Authorization"] = this.authorization;
+    if (cursor) h["Last-Event-ID"] = cursor;
+    return h;
+  }
+
+  private streamUrl(cursor?: string): string {
+    // session_id (required for tenant routing) and the cursor. The cursor goes
+    // as last_event_id (Transport Section 4.3.3) and, for servers that predate
+    // that name, as lastEventId. Only the query string is touched: a relative
+    // URL ("events", "/x", "//host/x") must resolve exactly as it was supplied.
+    const hashAt = this.eventsUrl.indexOf("#");
+    const hash = hashAt < 0 ? "" : this.eventsUrl.slice(hashAt);
+    const noHash = hashAt < 0 ? this.eventsUrl : this.eventsUrl.slice(0, hashAt);
+    const qAt = noHash.indexOf("?");
+    const base = qAt < 0 ? noHash : noHash.slice(0, qAt);
+    const params = new URLSearchParams(qAt < 0 ? "" : noHash.slice(qAt + 1));
+    if (this.sessionId) params.set("session_id", this.sessionId);
+    // The cursor names are reserved: whatever the supplied URL carried, only the
+    // current cursor (none after a 410) goes out, or the recovery request would
+    // draw the same 410 again.
+    params.delete("last_event_id");
+    params.delete("lastEventId");
+    if (cursor) {
+      params.set("last_event_id", cursor);
+      params.set("lastEventId", cursor);
     }
+    const query = params.toString();
+    return base + (query ? `?${query}` : "") + hash;
+  }
 
-    this.eventSource = new this.EventSourceCtor(url);
+  private openStream(cursor: string | undefined): void {
+    this.eventSource?.close();
+    this.eventSource = null;
 
-    this.eventSource.onopen = () => this.emit("open");
-    this.eventSource.onerror = () => {
-      if (!this.closed) {
-        this.emit("error", new Error("SSE connection error"));
+    const headers = this.streamHeaders(cursor);
+    // Polyfills that can set request headers read `headers` from the init
+    // object; a native EventSource ignores it.
+    const init = Object.keys(headers).length > 0 ? ({ headers } as EventSourceInit) : undefined;
+    const es = new this.EventSourceCtor(this.streamUrl(cursor), init);
+    this.eventSource = es;
+    // The session this stream belongs to: once it is no longer the bound one (or
+    // the stream was replaced or closed) nothing it delivers may touch the
+    // cursor, the seen ids or the application.
+    const session = this.sessionId;
+
+    es.onopen = () => {
+      if (this.eventSource !== es) return;
+      this.emit("open");
+      if (this.pendingGap) {
+        this.pendingGap = false;
+        this.emit("gap");
       }
     };
-
-    // WMP events use the "wmp" event type per the spec.
-    this.eventSource.addEventListener("wmp", (ev) => {
-      try {
-        const msg = decodeMessage((ev as MessageEvent).data, { maxSize: 4 * 1024 * 1024 });
-        this.emit("message", msg);
-      } catch (err) {
-        this.emit(
-          "error",
-          err instanceof Error ? err : new Error(String(err)),
-        );
+    es.onerror = () => {
+      if (this.closed || this.eventSource !== es) return;
+      // A closed stream with a cursor may be the 410 of a gap; the EventSource
+      // API does not expose the status, so ask.
+      if (es.readyState === 2 /* CLOSED */ && this.cursor) {
+        void this.checkGap(es, this.cursor);
+        return;
       }
-    });
+      this.emit("error", new Error("SSE connection error"));
+    };
 
+    // WMP events use the "wmp" event type per the spec.
+    es.addEventListener("wmp", (ev) => this.onSseEvent(ev as MessageEvent, es, session));
     // Also handle default "message" events for compatibility.
-    this.eventSource.onmessage = (ev) => {
-      try {
-        const msg = decodeMessage(ev.data, { maxSize: 4 * 1024 * 1024 });
-        this.emit("message", msg);
-      } catch (err) {
-        this.emit(
-          "error",
-          err instanceof Error ? err : new Error(String(err)),
-        );
-      }
-    };
+    es.onmessage = (ev) => this.onSseEvent(ev, es, session);
+  }
+
+  /**
+   * Handles one SSE event: an event whose ID was already processed (a replay
+   * after a reconnect) is dropped; otherwise it is delivered and then its ID
+   * becomes the cursor (Transport Section 4.3.2).
+   */
+  private onSseEvent(ev: MessageEvent, source: EventSource, session: string | undefined): void {
+    const current = () => this.eventSource === source && this.sessionId === session;
+    if (!current()) return; // a stale stream or session
+    const id = typeof ev.lastEventId === "string" && ev.lastEventId !== "" ? ev.lastEventId : undefined;
+    if (id !== undefined && this.seenIds.has(id)) return;
+    try {
+      this.emit("message", decodeMessage(ev.data, { maxSize: 4 * 1024 * 1024 }));
+    } catch (err) {
+      this.emit("error", err instanceof Error ? err : new Error(String(err)));
+    } finally {
+      // Handling the message may have ended the session (a session.close event):
+      // its id must not become the next session's cursor.
+      if (id !== undefined && current()) this.recordEventId(id);
+    }
+  }
+
+  private recordEventId(id: string): void {
+    this.cursor = id;
+    this.seenIds.add(id);
+    if (this.seenIds.size > this.dedupeWindow) {
+      // Sets iterate in insertion order: drop the oldest.
+      this.seenIds.delete(this.seenIds.values().next().value as string);
+    }
+  }
+
+  /** Probes the stream URL for a 410 after the EventSource gave up with `cursor`. */
+  private async checkGap(failed: EventSource, cursor: string): Promise<void> {
+    // The session the stream (and so the probe) belongs to: a 410 for a session
+    // that is no longer the bound one says nothing about the current session.
+    const session = this.sessionId;
+    let status: number | undefined;
+    try {
+      const abort = new AbortController();
+      const resp = await this.fetchFn(this.streamUrl(cursor), {
+        headers: { ...this.streamHeaders(cursor), Accept: "text/event-stream" },
+        signal: abort.signal,
+      });
+      status = resp.status;
+      abort.abort(); // only the status matters; do not hold the stream open
+    } catch {
+      status = undefined;
+    }
+    if (this.closed || this.eventSource !== failed || this.sessionId !== session) return;
+    if (status === 410) {
+      // Discard the cursor (never reconnect with it again) and open a stream
+      // with none; "gap" is emitted once it is open, so the resume's re-sent
+      // notifications have a connection to arrive on.
+      this.cursor = undefined;
+      this.pendingGap = true;
+      this.openStream(undefined);
+      return;
+    }
+    this.emit("error", new Error("SSE connection error"));
   }
 
   async send(msg: Message): Promise<void> {
@@ -293,8 +443,22 @@ export class HttpSseTransport extends Emitter implements Transport {
     if (this.authorization) {
       headers["Authorization"] = this.authorization;
     }
-    if (this.sessionId) {
-      headers["Wmp-Session-Id"] = this.sessionId;
+    // A session.create must not carry a session id (Transport 4.3.1: the server
+    // assigns it and refuses a request that names one). The binding itself stays,
+    // in case the create fails and the old session is still the one in use.
+    const method = (msg as { method?: unknown }).method;
+    let routeId: string | undefined = this.sessionId;
+    if (method === "wmp.session.create") {
+      routeId = undefined;
+    } else if (method === "wmp.session.resume") {
+      // A resume is addressed to the session it asks for, which may not be the one
+      // currently bound (the binding changes only if the resume succeeds): header
+      // and body must name the same session.
+      const asked = (msg as { params?: { session_id?: unknown } }).params?.session_id;
+      if (typeof asked === "string" && asked !== "") routeId = asked;
+    }
+    if (routeId) {
+      headers["Wmp-Session-Id"] = routeId;
     }
 
     const resp = await this.fetchFn(this.rpcUrl, {

test/mock-transport.ts

@@ -30,6 +30,11 @@ export class MockTransport implements Transport {
     this.listeners.get(event)?.delete(listener as (...args: unknown[]) => void);
   }
 
+  /** Fires a transport event other than "message" (for example "gap"). */
+  emitEvent(event: TransportEventName): void {
+    this.listeners.get(event)?.forEach((fn) => fn());
+  }
+
   receive(msg: Message): void {
     this.listeners.get("message")?.forEach((fn) => fn(msg));
   }