fix: evict idle HTTP sessions instead of leaking them

CLAUDE.md

@@ -53,6 +53,7 @@ Digital Samba Embedded API MCP Server - a Model Context Protocol server for Digi
 - `DIGITAL_SAMBA_DEVELOPER_KEY`, `DIGITAL_SAMBA_API_URL`, `OAUTH_API_URL`
 - `OAUTH_CLIENT_ID`, `OAUTH_CLIENT_SECRET`, `OAUTH_AUTHORIZE_URL`, `OAUTH_TOKEN_URL`, `OAUTH_REDIRECT_URI`, `OAUTH_ISSUER`
 - `REDIS_URL`
+- `SESSION_IDLE_TIMEOUT_MS` (default 30min; 0 disables), `SESSION_SWEEP_INTERVAL_MS` (default 5min) — idle HTTP session eviction
 
 ## Architecture
 
@@ -64,6 +65,7 @@ src/
 ├── oauth.ts              # OAuth 2.0 / PKCE / DCR implementation
 ├── session-store.ts      # Redis-backed session store (memory fallback)
 ├── auth.ts               # AsyncLocalStorage API-key context
+├── session-registry.ts   # HTTP session tracking + idle sweep (clients rarely send DELETE)
 ├── cache.ts              # Simple memory cache
 ├── logger.ts             # Console logger (writes to stderr in stdio mode)
 ├── errors.ts             # Error type definitions

src/session-registry.ts

@@ -0,0 +1,135 @@
+/**
+ * Tracking of active HTTP transport sessions.
+ *
+ * Most MCP clients never send `DELETE /mcp`, and a transport's `onclose` only
+ * fires on an explicit close, so a client that simply goes away would otherwise
+ * leave its entry — and the Server instance behind it — in memory forever.
+ * This registry adds last-seen tracking and an idle sweep on top of the map.
+ *
+ * Sessions holding an open SSE stream are never swept, however long they have
+ * been quiet: streaming is activity.
+ *
+ * @module session-registry
+ */
+
+import logger from "./logger.js";
+
+/** The subset of a transport this registry needs. */
+export interface ClosableTransport {
+  close(): Promise<void> | void;
+}
+
+export interface SessionRegistryOptions {
+  /** Evict sessions idle for longer than this. 0 disables sweeping. */
+  idleTimeoutMs?: number;
+}
+
+export class SessionRegistry<T extends ClosableTransport> {
+  private readonly transports = new Map<string, T>();
+  private readonly lastActivity = new Map<string, number>();
+  private readonly openStreams = new Map<string, number>();
+
+  readonly idleTimeoutMs: number;
+
+  /** Cumulative count of swept sessions, surfaced via /health. */
+  sweptCount = 0;
+
+  constructor(options: SessionRegistryOptions = {}) {
+    this.idleTimeoutMs = options.idleTimeoutMs ?? 30 * 60 * 1000;
+  }
+
+  get size(): number {
+    return this.transports.size;
+  }
+
+  /** Number of sessions currently holding at least one open stream. */
+  get streamingCount(): number {
+    return this.openStreams.size;
+  }
+
+  has(sessionId: string): boolean {
+    return this.transports.has(sessionId);
+  }
+
+  get(sessionId: string): T | undefined {
+    return this.transports.get(sessionId);
+  }
+
+  values(): IterableIterator<T> {
+    return this.transports.values();
+  }
+
+  entries(): IterableIterator<[string, T]> {
+    return this.transports.entries();
+  }
+
+  add(sessionId: string, transport: T, now: number = Date.now()): void {
+    this.transports.set(sessionId, transport);
+    this.lastActivity.set(sessionId, now);
+  }
+
+  /** Record traffic on a session, resetting its idle clock. */
+  touch(sessionId: string, now: number = Date.now()): void {
+    if (this.transports.has(sessionId)) {
+      this.lastActivity.set(sessionId, now);
+    }
+  }
+
+  /** Drop a session and all its bookkeeping. Does not close the transport. */
+  remove(sessionId: string): void {
+    this.transports.delete(sessionId);
+    this.lastActivity.delete(sessionId);
+    this.openStreams.delete(sessionId);
+  }
+
+  /** Mark an SSE stream as opened; the session can't be swept while open. */
+  openStream(sessionId: string, now: number = Date.now()): void {
+    this.openStreams.set(sessionId, (this.openStreams.get(sessionId) ?? 0) + 1);
+    this.touch(sessionId, now);
+  }
+
+  /** Mark an SSE stream as closed; the idle clock restarts from now. */
+  closeStream(sessionId: string, now: number = Date.now()): void {
+    const remaining = (this.openStreams.get(sessionId) ?? 1) - 1;
+    if (remaining > 0) {
+      this.openStreams.set(sessionId, remaining);
+    } else {
+      this.openStreams.delete(sessionId);
+    }
+    this.touch(sessionId, now);
+  }
+
+  /**
+   * Close and evict every session idle beyond the timeout.
+   *
+   * @returns the number of sessions swept
+   */
+  async sweep(now: number = Date.now()): Promise<number> {
+    if (this.idleTimeoutMs <= 0) return 0;
+
+    let swept = 0;
+    for (const [id, transport] of [...this.transports]) {
+      if ((this.openStreams.get(id) ?? 0) > 0) continue; // streaming = active
+
+      const seen = this.lastActivity.get(id) ?? 0;
+      if (now - seen < this.idleTimeoutMs) continue;
+
+      // Remove first: close() triggers onclose, and a throwing close() must not
+      // leave the entry behind — that would defeat the whole point of the sweep.
+      this.remove(id);
+      try {
+        await transport.close();
+      } catch (err: any) {
+        logger.warn(`Error closing idle session ${id}: ${err?.message}`);
+      }
+      swept++;
+      logger.info(`Swept idle session: ${id}`);
+    }
+
+    if (swept > 0) {
+      this.sweptCount += swept;
+      logger.info(`Swept ${swept} idle session(s), ${this.size} remaining`);
+    }
+    return swept;
+  }
+}

src/transports/http.ts

@@ -24,6 +24,7 @@ import {
 } from "../server.js";
 import logger from "../logger.js";
 import apiKeyContext from "../auth.js";
+import { SessionRegistry } from "../session-registry.js";
 import {
   loadOAuthConfig,
   generateState,
@@ -55,8 +56,18 @@ export interface HttpTransportConfig {
   requireAuth?: boolean;
 }
 
-// Active transport sessions
-const transports: Map<string, StreamableHTTPServerTransport> = new Map();
+// 0 disables the idle sweep entirely.
+const SESSION_IDLE_TIMEOUT_MS = Number(
+  process.env.SESSION_IDLE_TIMEOUT_MS ?? 30 * 60 * 1000,
+);
+const SESSION_SWEEP_INTERVAL_MS = Number(
+  process.env.SESSION_SWEEP_INTERVAL_MS ?? 5 * 60 * 1000,
+);
+
+// Active transport sessions, with idle tracking (see session-registry.ts).
+const transports = new SessionRegistry<StreamableHTTPServerTransport>({
+  idleTimeoutMs: SESSION_IDLE_TIMEOUT_MS,
+});
 
 /**
  * Authentication middleware
@@ -212,6 +223,8 @@ export async function startHttpServer(
       commit: GIT_COMMIT,
       transport: "http",
       activeSessions: transports.size,
+      streamingSessions: transports.streamingCount,
+      sweptSessions: transports.sweptCount,
       oauthSessions: await getActiveSessionCount(),
       registeredClients: await getRegisteredClientCount(),
     });
@@ -563,6 +576,7 @@ export async function startHttpServer(
       if (mcpSessionId && transports.has(mcpSessionId)) {
         // Reuse existing session
         transport = transports.get(mcpSessionId)!;
+        transports.touch(mcpSessionId);
         logger.debug(`Reusing session: ${mcpSessionId}`);
       } else if (!mcpSessionId && isInitializeRequest(req.body)) {
         // New session initialization
@@ -571,18 +585,18 @@ export async function startHttpServer(
         transport = new StreamableHTTPServerTransport({
           sessionIdGenerator: () => randomUUID(),
           onsessioninitialized: (id) => {
-            transports.set(id, transport);
+            transports.add(id, transport);
             logger.info(`Session initialized: ${id}`);
           },
           onsessionclosed: (id) => {
-            transports.delete(id);
+            transports.remove(id);
             logger.info(`Session closed: ${id}`);
           },
         });
 
         transport.onclose = () => {
           if (transport.sessionId) {
-            transports.delete(transport.sessionId);
+            transports.remove(transport.sessionId);
             logger.debug(
               `Transport closed, session cleaned up: ${transport.sessionId}`,
             );
@@ -667,6 +681,14 @@ export async function startHttpServer(
     }
 
     const transport = transports.get(sessionId)!;
+
+    // An SSE stream can stay open for hours with no other traffic. Count it as
+    // in-flight for the whole time it is open so the idle sweep can't evict a
+    // session that is actively streaming.
+    transports.openStream(sessionId);
+    // Start the idle clock from when the stream ends, not when it began.
+    res.on("close", () => transports.closeStream(sessionId));
+
     const apiKey = (req as any).apiKey;
     if (apiKey) {
       await apiKeyContext.run(apiKey, async () => {
@@ -694,6 +716,7 @@ export async function startHttpServer(
     }
 
     const transport = transports.get(sessionId)!;
+    transports.touch(sessionId);
     const apiKey = (req as any).apiKey;
     if (apiKey) {
       await apiKeyContext.run(apiKey, async () => {
@@ -1333,6 +1356,20 @@ export async function startHttpServer(
   });
   app.delete("/", handleMcpDelete);
 
+  // Evict sessions abandoned without a DELETE (most clients never send one).
+  const sweepTimer =
+    SESSION_IDLE_TIMEOUT_MS > 0
+      ? setInterval(() => {
+          transports
+            .sweep()
+            .catch((err) =>
+              logger.error(`Idle session sweep failed: ${err?.message}`),
+            );
+        }, SESSION_SWEEP_INTERVAL_MS)
+      : null;
+  // Don't hold the event loop open on account of the sweep.
+  sweepTimer?.unref();
+
   // Start listening
   app.listen(port, host, () => {
     logger.info("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
@@ -1342,13 +1379,21 @@ export async function startHttpServer(
     logger.info(`Listening: http://${host}:${port}`);
     logger.info(`MCP Endpoint: http://${host}:${port}/mcp`);
     logger.info(`Auth: ${requireAuth ? "Required" : "Optional"}`);
+    logger.info(
+      `Idle session sweep: ${
+        SESSION_IDLE_TIMEOUT_MS > 0
+          ? `every ${SESSION_SWEEP_INTERVAL_MS / 1000}s, timeout ${SESSION_IDLE_TIMEOUT_MS / 60000}min`
+          : "disabled"
+      }`,
+    );
     logger.info("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
   });
 
   // Graceful shutdown
   process.on("SIGINT", async () => {
     logger.info("Shutting down HTTP server...");
-    for (const [id, transport] of transports) {
+    if (sweepTimer) clearInterval(sweepTimer);
+    for (const [id, transport] of transports.entries()) {
       logger.debug(`Closing session: ${id}`);
       await transport.close();
     }
@@ -1357,6 +1402,7 @@ export async function startHttpServer(
 
   process.on("SIGTERM", async () => {
     logger.info("Received SIGTERM, shutting down...");
+    if (sweepTimer) clearInterval(sweepTimer);
     for (const transport of transports.values()) {
       await transport.close();
     }