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

src/peer.ts

@@ -138,6 +138,13 @@ export interface PeerOptions {
    * owns it. Default: true.
    */
   validateCompletions?: boolean;
+  /**
+   * Called when the session cannot be continued after an event gap (the
+   * transport saw 410 and the automatic wmp.session.resume failed or had no
+   * resumption token). The application creates a new session
+   * (Transport Section 4.3.4).
+   */
+  onSessionLost?: (error: Error) => void;
   /** Called for every refused completion, with the violation (never a member value). */
   onRefusedCompletion?: (flowId: string, error: CompletionResultError) => void;
 }
@@ -160,6 +167,7 @@ export class Peer implements PeerContext {
   private authorize?: PeerOptions["authorize"];
   private validateCompletions: boolean;
   private onRefusedCompletion?: PeerOptions["onRefusedCompletion"];
+  private onSessionLost?: PeerOptions["onSessionLost"];
 
   constructor(transport: Transport, opts?: PeerOptions) {
     this.transport = transport;
@@ -169,11 +177,13 @@ export class Peer implements PeerContext {
     this.authorize = opts?.authorize;
     this.validateCompletions = opts?.validateCompletions ?? true;
     this.onRefusedCompletion = opts?.onRefusedCompletion;
+    this.onSessionLost = opts?.onSessionLost;
 
     // Wire up transport events.
     this.transport.on("message", (msg: Message) => this.dispatch(msg));
     this.transport.on("error", (err: Error) => this.onTransportError(err));
     this.transport.on("close", () => this.onTransportClose());
+    this.transport.on("gap", () => this.onTransportGap());
   }
 
   /** Register a profile. Call before starting to receive messages. */
@@ -201,7 +211,19 @@ export class Peer implements PeerContext {
 
   /** Drops the validation context of a flow that ended; the flow_id is free to start again. */
   private releaseFlow(flowId: unknown): void {
-    if (typeof flowId === "string") this.recordedFlows.delete(flowId);
+    if (typeof flowId !== "string") return;
+    this.recordedFlows.delete(flowId);
+    this.releaseGen(flowId);
+  }
+
+  /**
+   * Reclaims a flow id's generation once nothing needs it: the flow is not active
+   * and no cancellation for it is pending (a pending one compares against it).
+   * Without this every finished flow would leave an entry behind for the whole
+   * session. A missing entry reads as generation 0.
+   */
+  private releaseGen(flowId: string): void {
+    if (!this.recordedFlows.has(flowId) && !this.cancelling.has(flowId)) this.flowGen.delete(flowId);
   }
 
   async call<T = unknown>(method: string, params: unknown): Promise<T> {
@@ -289,23 +311,66 @@ export class Peer implements PeerContext {
     return result;
   }
 
-  /** Resume a previously established WMP session using a resumption token. */
+  /**
+   * Resume a previously established WMP session using a resumption token.
+   * Resumes run one at a time (explicit calls and gap recovery alike): the token
+   * is single use, so a second resume must wait for the first to store the
+   * rotated one and then present that.
+   */
   async resumeSession(opts?: {
     lastReceivedId?: string;
   }): Promise<SessionResumeResult> {
+    // The session (flow epoch) this resume is requested for is noted NOW, before
+    // it waits its turn: if that session is closed or replaced while it queues, it
+    // is dropped instead of resuming the session that replaced it.
+    return this.resumeSerialized(opts, this.flowEpoch);
+  }
+
+  /**
+   * `forEpoch`: recovery work that is only valid for the session it was queued
+   * for; if that session was closed or replaced (or the peer closed) while it
+   * waited its turn, it is dropped (StaleSessionError) instead of resuming the
+   * session that replaced it.
+   */
+  private async resumeSerialized(opts: { lastReceivedId?: string } | undefined, forEpoch: number): Promise<SessionResumeResult> {
+    const previous = this.resumeChain;
+    let release!: () => void;
+    this.resumeChain = new Promise<void>((r) => (release = r));
+    await previous;
+    try {
+      if (this.closed || this.flowEpoch !== forEpoch) throw new StaleSessionError();
+      return await this.resumeNow(opts);
+    } finally {
+      release();
+    }
+  }
+
+  private resumeChain: Promise<void> = Promise.resolve();
+
+  private async resumeNow(opts?: { lastReceivedId?: string }): Promise<SessionResumeResult> {
     if (!this.sessionId || !this.resumptionToken) {
       throw new Error("no session or resumption token to resume");
     }
     const params = {
       wmp: { version: this.negotiatedVersion ?? VERSION } as Metadata,
       session_id: this.sessionId,
       resumption_token: this.resumptionToken,
-      last_received_id: opts?.lastReceivedId,
+      // In the HTTP binding this is the last event ID (Transport Section 4.5).
+      last_received_id:
+        opts?.lastReceivedId ?? (this.transport as { lastEventId?: string }).lastEventId,
     };
+    const epoch = this.flowEpoch;
     const result = await this.call<SessionResumeResult>(
       Method.SessionResume,
       params,
     );
+    // The session this resume was for may have been closed or replaced while it
+    // was in flight: its result must not overwrite the session that replaced it.
+    if (this.flowEpoch !== epoch) throw new StaleSessionError();
+    // An unsuccessful resume changes nothing; the caller creates a new session.
+    if (result.resumed !== true) {
+      throw new WMPError(ErrorCode.SessionExpired, "Session could not be resumed");
+    }
     if (result.wmp?.session_id) {
       // Resuming the same session keeps its flow state; another session does not.
       if (result.wmp.session_id !== this.sessionId) this.resetFlows();
@@ -712,9 +777,46 @@ export class Peer implements PeerContext {
 
       case Method.FlowCancel:
         if (this.handler.onFlowCancel) {
-          const result = await this.handler.onFlowCancel(p);
-          this.endFlow(p?.flow_id, "cancelled");
-          return result;
+          const flowId: unknown = p?.flow_id;
+          if (typeof flowId === "string") {
+            // A cancellation already processed (a resume re-sends it with a fresh
+            // event ID) is acknowledged again without invoking the handler. Only a
+            // CANCELLED flow: one that ended by complete/error reaches the handler.
+            if (this.terminalSeen.get(flowId) === "cancelled") {
+              return { wmp: p.wmp, flow_id: flowId, status: "cancelled" };
+            }
+            // The flow this cancellation is for: the session epoch and the flow's
+            // generation (stable across the flow's completion, changed by a new flow
+            // under the same id or by a session reset).
+            const epoch = this.flowEpoch;
+            const gen = this.flowGen.get(flowId) ?? 0;
+            // A repeat for the same flow that arrives while the first is still being
+            // processed shares its outcome, even if the flow completed meanwhile. A
+            // replacement flow under the same id is a different generation.
+            const pending = this.cancelling.get(flowId);
+            if (pending && pending.epoch === epoch && pending.gen === gen) return pending.run;
+            const entry: { epoch: number; gen: number; run: Promise<unknown> } = { epoch, gen, run: undefined as unknown as Promise<unknown> };
+            entry.run = (async () => {
+              // Yield first: the entry is installed before the handler can throw
+              // (even synchronously), so its cleanup always sees it.
+              await Promise.resolve();
+              try {
+                const result = await this.handler.onFlowCancel!(p);
+                // Finalize only the flow that was cancelled: if another started under
+                // the id (or the session was reset) meanwhile, leave it alone.
+                if (this.flowEpoch === epoch && (this.flowGen.get(flowId) ?? 0) === gen) this.endFlow(flowId, "cancelled");
+                return result;
+              } finally {
+                if (this.cancelling.get(flowId) === entry) {
+                  this.cancelling.delete(flowId); // a failure leaves a retry possible
+                  this.releaseGen(flowId);
+                }
+              }
+            })();
+            this.cancelling.set(flowId, entry);
+            return entry.run;
+          }
+          return this.handler.onFlowCancel(p);
         }
         break;
 
@@ -768,13 +870,19 @@ export class Peer implements PeerContext {
   private endFlow(flowId: unknown, kind: "ended" | "cancelled" = "ended"): void {
     if (typeof flowId !== "string") return;
     const known = this.recordedFlows.delete(flowId);
-    if (known) this.markTerminal(flowId, kind);
+    this.releaseGen(flowId);
+    // A flow already in the history may be re-marked (a cancel of an ended flow
+    // upgrades its kind) without growing it.
+    if (known || this.terminalSeen.has(flowId)) this.markTerminal(flowId, kind);
   }
 
   /** A new or ended session: the old session's flow state does not carry over. */
   private resetFlows(): void {
+    this.flowEpoch++;
     this.recordedFlows.clear();
     this.terminalSeen.clear();
+    this.flowGen.clear();
+    this.cancelling.clear(); // pending cancellations belong to the old session
   }
 
   /**
@@ -803,12 +911,21 @@ export class Peer implements PeerContext {
     const q = flowType === "match" && typeof params === "object" && params !== null ? (params as { dcql_query?: unknown }).dcql_query : undefined;
     const rec = { type: flowType, queryIds: q === undefined ? undefined : dcqlQueryIds(q) };
     this.recordedFlows.set(flowId, rec);
+    const previousGen = this.flowGen.get(flowId);
+    const myGen = ++this.genCounter;
+    this.flowGen.set(flowId, myGen); // a new flow under this id (tentative until the start succeeds)
     return () => {
       // Only while this start's own record is still the flow's: if the
       // replacement already ended, or another flow took the id, the marker is theirs.
       if (this.recordedFlows.get(flowId) === rec) {
         this.recordedFlows.delete(flowId);
         if (hadTerminal !== undefined) this.terminalSeen.set(flowId, hadTerminal);
+        // The start failed: the flow generation it allocated goes with it, so work
+        // tied to the flow that was there before (a pending cancellation) still matches.
+        if (this.flowGen.get(flowId) === myGen) {
+          if (previousGen === undefined) this.flowGen.delete(flowId);
+          else this.flowGen.set(flowId, previousGen);
+        }
       }
     };
   }
@@ -850,6 +967,25 @@ export class Peer implements PeerContext {
     else this.handler.onFlowError?.(synthesized);
   }
 
+  /**
+   * Cancellations being processed, by flow_id, with the flow generation they
+   * belong to: a repeat for the SAME generation arriving meanwhile shares the
+   * outcome; a replacement flow under the same id never does.
+   */
+  private cancelling = new Map<string, { epoch: number; gen: number; run: Promise<unknown> }>();
+
+  /** Incremented whenever the flow state is reset (session closed or replaced). */
+  private flowEpoch = 0;
+
+  /**
+   * A per-flow-id generation that survives the flow's completion: it changes only
+   * when a new flow starts under the id (and with every reset), so work tied to
+   * "this flow" can tell a replacement from the same flow having ended.
+   */
+  private flowGen = new Map<string, number>();
+  /** Source of unique generation numbers (a rolled-back start never reuses one). */
+  private genCounter = 0;
+
   // Track flow_id → flow_type mapping for routing to the right FlowHandler.
   private activeFlows = new Map<string, string>();
 
@@ -886,6 +1022,42 @@ export class Peer implements PeerContext {
     // Could emit events here for reconnection logic.
   }
 
+  /**
+   * The transport lost events (410): resume the session, which re-sends the
+   * latest progress of active flows and the terminal notification of ended
+   * flows on the already open stream. Without a resumption token, or if the
+   * resume fails, the session is lost.
+   */
+  private onTransportGap(): void {
+    // Recoveries run one at a time: a second resume sent before the first one's
+    // response arrived would present the token the first already rotated. A gap
+    // that arrives while one is queued adds nothing (the queued one covers it).
+    // The session this gap belongs to: recovery work for a session that was closed
+    // or replaced meanwhile is discarded, never applied to the one that replaced it.
+    const epoch = this.flowEpoch;
+    if (this.recoveryQueuedEpoch === epoch) return; // coalesce only with a queued recovery of THIS session
+    this.recoveryQueuedEpoch = epoch;
+    this.recovery = this.recovery.then(async () => {
+      // A stale task must not clear the marker of a newer session's queued recovery.
+      if (this.recoveryQueuedEpoch === epoch) this.recoveryQueuedEpoch = undefined;
+      if (this.closed || this.flowEpoch !== epoch) return;
+      if (!this.sessionId || !this.resumptionToken) {
+        this.onSessionLost?.(new Error("event gap and no resumption token to resume with"));
+        return;
+      }
+      try {
+        await this.resumeSerialized(undefined, epoch);
+      } catch (err: unknown) {
+        // The session changed (or the peer closed) while resuming: nothing was lost.
+        if (err instanceof StaleSessionError || this.closed || this.flowEpoch !== epoch) return;
+        this.onSessionLost?.(err instanceof Error ? err : new Error(String(err)));
+      }
+    });
+  }
+
+  private recovery: Promise<void> = Promise.resolve();
+  private recoveryQueuedEpoch: number | undefined;
+
   private onTransportClose(): void {
     if (!this.closed) {
       // Transport closed unexpectedly — reject all pending.
@@ -902,6 +1074,18 @@ export class Peer implements PeerContext {
 // Error conversion
 // ---------------------------------------------------------------------------
 
+/** A resume result (or recovery) that belongs to a session that was closed or replaced meanwhile. */
+class StaleSessionError extends Error {
+  constructor() {
+    super("session changed while resuming");
+    this.name = "StaleSessionError";
+  }
+}
+
+function isFlowIdEnvelope(params: unknown): params is { flow_id: string } {
+  return typeof params === "object" && params !== null && typeof (params as { flow_id?: unknown }).flow_id === "string";
+}
+
 /** The shape a wmp.flow.complete must have before the early result check applies: flow_id and the wmp metadata object with its version. */
 function isFlowCompleteEnvelope(params: unknown): params is FlowCompleteParams {
   if (typeof params !== "object" || params === null) return false;
@@ -915,10 +1099,6 @@ function flowIdOf(params: unknown): unknown {
   return typeof params === "object" && params !== null ? (params as { flow_id?: unknown }).flow_id : undefined;
 }
 
-function isFlowIdEnvelope(params: unknown): params is { flow_id: string } {
-  return typeof params === "object" && params !== null && typeof (params as { flow_id?: unknown }).flow_id === "string";
-}
-
 function flowTypeOf(params: unknown): unknown {
   return typeof params === "object" && params !== null ? (params as { flow_type?: unknown }).flow_type : undefined;
 }