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. */
@@ -289,23 +299,48 @@ 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> {
+    const previous = this.resumeChain;
+    let release!: () => void;
+    this.resumeChain = new Promise<void>((r) => (release = r));
+    await previous;
+    try {
+      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 result = await this.call<SessionResumeResult>(
       Method.SessionResume,
       params,
     );
+    // 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();
@@ -695,9 +730,42 @@ 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 generation this cancellation is for: the session epoch and the
+            // flow's record (if any).
+            const gen = this.flowEpoch;
+            const rec = this.recordedFlows.get(flowId);
+            // A repeat for the same generation that arrives while the first is
+            // still being processed shares its outcome. A replacement flow under
+            // the same id is a different generation and gets its own run.
+            const pending = this.cancelling.get(flowId);
+            if (pending && pending.gen === gen && pending.rec === rec) return pending.run;
+            const entry: { gen: number; rec: object | undefined; run: Promise<unknown> } = { gen, rec, 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 generation that was cancelled: if that flow
+                // ended and another started under the id meanwhile, leave it alone.
+                if (this.flowEpoch === gen && this.recordedFlows.get(flowId) === rec) this.endFlow(flowId, "cancelled");
+                return result;
+              } finally {
+                if (this.cancelling.get(flowId) === entry) this.cancelling.delete(flowId); // a failure leaves a retry possible
+              }
+            })();
+            this.cancelling.set(flowId, entry);
+            return entry.run;
+          }
+          return this.handler.onFlowCancel(p);
         }
         break;
 
@@ -751,13 +819,17 @@ 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);
+    // 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.cancelling.clear(); // pending cancellations belong to the old session
   }
 
   /**
@@ -833,6 +905,16 @@ 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, { gen: number; rec: object | undefined; run: Promise<unknown> }>();
+
+  /** Incremented whenever the flow state is reset (session closed or replaced). */
+  private flowEpoch = 0;
+
   // Track flow_id → flow_type mapping for routing to the right FlowHandler.
   private activeFlows = new Map<string, string>();
 
@@ -869,6 +951,35 @@ 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).
+    if (this.recoveryQueued) return;
+    this.recoveryQueued = true;
+    this.recovery = this.recovery.then(async () => {
+      this.recoveryQueued = false;
+      if (!this.sessionId || !this.resumptionToken) {
+        this.onSessionLost?.(new Error("event gap and no resumption token to resume with"));
+        return;
+      }
+      try {
+        await this.resumeSession();
+      } catch (err: unknown) {
+        this.onSessionLost?.(err instanceof Error ? err : new Error(String(err)));
+      }
+    });
+  }
+
+  private recovery: Promise<void> = Promise.resolve();
+  private recoveryQueued = false;
+
   private onTransportClose(): void {
     if (!this.closed) {
       // Transport closed unexpectedly — reject all pending.
@@ -885,6 +996,10 @@ export class Peer implements PeerContext {
 // Error conversion
 // ---------------------------------------------------------------------------
 
+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;
@@ -898,10 +1013,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;
 }