Introduce session sweeping

docs/server.md

@@ -161,7 +161,10 @@ The SDK includes several built-in transport provider implementations:
 
     - Efficient bidirectional HTTP communication
     - Session management for multiple client connections
-    - Configurable keep-alive intervals
+    - Keep-alive pings on sessions with an open stream, enabled by default every 30 minutes
+      (`keepAliveInterval`, `null` to disable)
+    - Eviction of idle sessions — no open stream and no request for a full interval — every
+      30 minutes by default (`sessionSweepInterval`, `null` to keep sessions until deleted)
     - Security validation support
     - Graceful shutdown support
 

mcp-core/src/main/java/io/modelcontextprotocol/util/KeepAliveScheduler.java

@@ -1,11 +1,12 @@
 /**
- * Copyright 2025 - 2025 the original author or authors.
+ * Copyright 2025 - 2026 the original author or authors.
  */
 
 package io.modelcontextprotocol.util;
 
 import java.time.Duration;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
 import java.util.function.Supplier;
 
 import org.slf4j.Logger;
@@ -28,6 +29,7 @@
  * The pings are sent to all active mcp sessions at regular intervals.
  *
  * @author Christian Tzolov
+ * @author Daniel Garnier-Moiroux
  */
 public class KeepAliveScheduler {
 
@@ -57,20 +59,25 @@ public class KeepAliveScheduler {
 	/** Supplier for reactive McpSession instances */
 	private final Supplier<Flux<McpSession>> mcpSessions;
 
+	/** Invoked with the session whose keep-alive ping went unanswered */
+	private final Consumer<McpSession> onPingFailure;
+
 	/**
 	 * Creates a KeepAliveScheduler with a custom scheduler, initial delay, interval and a
 	 * supplier for McpSession instances.
 	 * @param scheduler The scheduler to use for executing keepAlive calls
 	 * @param initialDelay Initial delay before the first keepAlive call
 	 * @param interval Interval between subsequent keepAlive calls
 	 * @param mcpSessions Supplier for McpSession instances
+	 * @param onPingFailure Callback invoked with the session whose ping went unanswered
 	 */
 	KeepAliveScheduler(Scheduler scheduler, Duration initialDelay, Duration interval,
-			Supplier<Flux<McpSession>> mcpSessions) {
+			Supplier<Flux<McpSession>> mcpSessions, Consumer<McpSession> onPingFailure) {
 		this.scheduler = scheduler;
 		this.initialDelay = initialDelay;
 		this.interval = interval;
 		this.mcpSessions = mcpSessions;
+		this.onPingFailure = onPingFailure;
 	}
 
 	/**
@@ -92,8 +99,14 @@ public Disposable start() {
 				.doOnNext(tick -> {
 					this.mcpSessions.get()
 						.flatMap(session -> session.sendRequest(McpSchema.METHOD_PING, null, OBJECT_TYPE_REF)
-							.doOnError(e -> logger.warn("Failed to send keep-alive ping to session {}: {}", session,
-									e.getMessage()))
+							// A ping has to be answered before the next one is due. The
+							// request timeout of the session is unrelated to keeping the
+							// connection alive, and is measured in hours by default.
+							.timeout(this.interval)
+							.doOnError(e -> {
+								logger.warn("Keep-alive ping to session {} failed: {}", session, e.getMessage());
+								this.onPingFailure.accept(session);
+							})
 							.onErrorComplete())
 						.subscribe();
 				})
@@ -148,12 +161,15 @@ public static class Builder {
 
 		private Scheduler scheduler = Schedulers.boundedElastic();
 
-		private Duration initialDelay = Duration.ofSeconds(0);
+		private Duration initialDelay = Duration.ofSeconds(30);
 
 		private Duration interval = Duration.ofSeconds(30);
 
 		private Supplier<Flux<McpSession>> mcpSessions;
 
+		private Consumer<McpSession> onPingFailure = session -> {
+		};
+
 		/**
 		 * Creates a new Builder instance with a supplier for McpSession instances.
 		 * @param mcpSessions The supplier for McpSession instances
@@ -204,12 +220,28 @@ public Builder interval(Duration interval) {
 			return this;
 		}
 
+		/**
+		 * Sets the callback invoked when a session does not answer a keep-alive ping
+		 * within the keep-alive interval. An unanswered ping means the connection the
+		 * ping was written to is dead, which the operating system does not necessarily
+		 * report: writing to a connection whose peer is gone keeps succeeding until it
+		 * resets. It does not mean the session itself is over, as the client is free to
+		 * reconnect to it.
+		 * @param onPingFailure The callback receiving the unresponsive session
+		 * @return This builder instance for method chaining
+		 */
+		public Builder onPingFailure(Consumer<McpSession> onPingFailure) {
+			Assert.notNull(onPingFailure, "onPingFailure must not be null");
+			this.onPingFailure = onPingFailure;
+			return this;
+		}
+
 		/**
 		 * Builds and returns a new KeepAliveScheduler instance.
 		 * @return A new KeepAliveScheduler configured with the builder's settings
 		 */
 		public KeepAliveScheduler build() {
-			return new KeepAliveScheduler(scheduler, initialDelay, interval, mcpSessions);
+			return new KeepAliveScheduler(scheduler, initialDelay, interval, mcpSessions, onPingFailure);
 		}
 
 	}

mcp-core/src/test/java/io/modelcontextprotocol/spec/McpStreamableServerSessionTests.java

@@ -0,0 +1,264 @@
+/*
+ * Copyright 2026-2026 the original author or authors.
+ */
+
+package io.modelcontextprotocol.spec;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import io.modelcontextprotocol.json.TypeRef;
+import io.modelcontextprotocol.server.McpRequestHandler;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Mono;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link McpStreamableServerSession}.
+ */
+class McpStreamableServerSessionTests {
+
+	private static final Duration TIMEOUT = Duration.ofSeconds(5);
+
+	private McpStreamableServerSession session() {
+		return session(Map.of());
+	}
+
+	private McpStreamableServerSession session(Map<String, McpRequestHandler<?>> requestHandlers) {
+		return new McpStreamableServerSession("session-1", McpSchema.ClientCapabilities.builder().build(),
+				new McpSchema.Implementation("test-client", "1.0.0"), TIMEOUT, requestHandlers, Map.of());
+	}
+
+	@Test
+	void replacedListeningStreamHasItsConnectionReleased() {
+		var session = session();
+		var firstTransport = new RecordingTransport();
+
+		session.listeningStream(firstTransport);
+		assertThat(firstTransport.closed).isFalse();
+
+		session.listeningStream(new RecordingTransport());
+		assertThat(firstTransport.closed).isTrue();
+	}
+
+	@Test
+	void replacingListeningStreamKeepsItsPendingRequestsResolvable() {
+		var session = session();
+		var firstTransport = new RecordingTransport();
+		session.listeningStream(firstTransport);
+
+		// Server-initiated requests (sampling, elicitation, roots/list) are sent on the
+		// listening SSE stream, but the client answers them with a separate HTTP POST
+		// which outlives that stream.
+		var pending = session.sendRequest("sampling/createMessage", null, new TypeRef<String>() {
+		}).toFuture();
+		assertThat(firstTransport.sent).hasSize(1);
+		var requestId = ((McpSchema.JSONRPCRequest) firstTransport.sent.peek()).id();
+
+		// The client reconnects with a Last-Event-ID header, replacing the listening
+		// stream. The request sent on the replaced stream must stay pending.
+		session.listeningStream(new RecordingTransport());
+		assertThat(pending).isNotDone();
+
+		session.accept(new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, requestId, "response-value", null))
+			.block(TIMEOUT);
+
+		assertThat(pending).succeedsWithin(TIMEOUT).isEqualTo("response-value");
+	}
+
+	@Test
+	void closingTheSessionReleasesTheConnectionOfItsStreams() {
+		var session = session();
+		var transport = new RecordingTransport();
+		session.listeningStream(transport);
+
+		session.close();
+
+		assertThat(transport.closed).isTrue();
+	}
+
+	@Test
+	void closingTheSessionFailsThePendingRequestsOfReleasedStreams() {
+		var session = session();
+		session.listeningStream(new RecordingTransport());
+
+		var pending = session.sendRequest("sampling/createMessage", null, new TypeRef<String>() {
+		}).toFuture();
+
+		// The stream the request was sent on is released, which leaves the request
+		// pending: a reconnecting client can still answer it over a separate POST
+		session.releaseListeningStream();
+		assertThat(pending).isNotDone();
+
+		// The client cannot respond on a closed session, the pending request must be
+		// failed.
+		session.close();
+
+		assertThat(pending).failsWithin(TIMEOUT).withThrowableThat().havingCause().withMessage("Session closed");
+	}
+
+	@Test
+	void closingTheSessionGracefullyFailsThePendingRequestsOfReleasedStreams() {
+		var session = session();
+		session.listeningStream(new RecordingTransport());
+
+		var pending = session.sendRequest("sampling/createMessage", null, new TypeRef<String>() {
+		}).toFuture();
+
+		session.releaseListeningStream();
+		assertThat(pending).isNotDone();
+
+		session.closeGracefully().block(TIMEOUT);
+
+		assertThat(pending).failsWithin(TIMEOUT).withThrowableThat().havingCause().withMessage("Session closed");
+	}
+
+	@Test
+	void closingAStreamFailsOnlyItsOwnPendingRequests() {
+		var session = session();
+		var listeningTransport = new RecordingTransport();
+		session.listeningStream(listeningTransport);
+
+		var onListeningStream = session.sendRequest("sampling/createMessage", null, new TypeRef<String>() {
+		}).toFuture();
+
+		// A POST response stream carries its own server-initiated requests, which can be
+		// closed independently
+		var responseTransport = new RecordingTransport();
+		var responseStream = session.new McpStreamableServerSessionStream(responseTransport);
+		var onResponseStream = responseStream.sendRequest("elicitation/create", null, new TypeRef<String>() {
+		}).toFuture();
+
+		responseStream.close();
+
+		assertThat(onResponseStream).failsWithin(TIMEOUT)
+			.withThrowableThat()
+			.havingCause()
+			.withMessage("Stream closed");
+		assertThat(onListeningStream).isNotDone();
+
+		var requestId = ((McpSchema.JSONRPCRequest) listeningTransport.sent.peek()).id();
+		session.accept(new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, requestId, "response-value", null))
+			.block(TIMEOUT);
+		assertThat(onListeningStream).succeedsWithin(TIMEOUT).isEqualTo("response-value");
+	}
+
+	@Test
+	void endOfTheConnectionCarryingAResponseStreamDetachesItFromTheSession() {
+		// A request whose handler never completes, as seen when a client gives up and
+		// disconnects while the server is still working on its tool call
+		var session = session(Map.of("tools/call", (exchange, params) -> Mono.never()));
+		var transport = new RecordingTransport();
+
+		// The caller owns the stream, so it can detach it from the session once the
+		// container tells it the connection carrying it is gone
+		var stream = session.responseStream(transport);
+		stream.handle(new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION, "tools/call", "request-1", null))
+			.subscribe();
+		assertThat(session.hasOpenStream()).isTrue();
+
+		stream.releaseTransport();
+
+		// The session must stop believing it holds a live connection: hasOpenStream() is
+		// what tells the session sweeper that a client is still around, so a stream which
+		// outlives its connection makes the session impossible to reclaim
+		assertThat(session.hasOpenStream()).isFalse();
+		assertThat(transport.closed).isTrue();
+	}
+
+	@Test
+	void responseStreamIsDetachedFromTheSessionOnceItsRequestIsAnswered() {
+		var session = session(Map.of("tools/call", (exchange, params) -> Mono.just("result")));
+		var transport = new RecordingTransport();
+
+		var stream = session.responseStream(transport);
+		assertThat(session.hasOpenStream()).isTrue();
+
+		stream.handle(new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION, "tools/call", "request-1", null))
+			.block(TIMEOUT);
+
+		assertThat(session.hasOpenStream()).isFalse();
+	}
+
+	@Test
+	void responseStreamIsDetachedFromTheSessionWhenItsResponseCannotBeSent() {
+		var session = session(Map.of("tools/call", (exchange, params) -> Mono.just("result")));
+
+		var stream = session.responseStream(new FailingTransport());
+		var handling = stream
+			.handle(new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION, "tools/call", "request-1", null))
+			.toFuture();
+
+		// The connection which was to carry the response failed. The caller gets to see
+		// it, and the stream must not be left attached to the session.
+		assertThat(handling).failsWithin(TIMEOUT).withThrowableThat().havingCause().withMessage("connection gone");
+		assertThat(session.hasOpenStream()).isFalse();
+	}
+
+	@Test
+	void abandonedResponseStreamIsDetachedFromTheSession() {
+		var session = session(Map.of("tools/call", (exchange, params) -> Mono.never()));
+
+		var stream = session.responseStream(new RecordingTransport());
+		var subscription = stream
+			.handle(new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION, "tools/call", "request-1", null))
+			.subscribe();
+		assertThat(session.hasOpenStream()).isTrue();
+
+		// The caller gives up on a request which would never terminate on its own, so
+		// nothing sends the response the stream was created to carry
+		subscription.dispose();
+
+		assertThat(session.hasOpenStream()).isFalse();
+	}
+
+	/**
+	 * A transport whose connection is gone, so that nothing can be written to it.
+	 */
+	static class FailingTransport extends RecordingTransport {
+
+		@Override
+		public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message, String messageId) {
+			return Mono.error(new RuntimeException("connection gone"));
+		}
+
+	}
+
+	static class RecordingTransport implements McpStreamableServerTransport {
+
+		final Queue<McpSchema.JSONRPCMessage> sent = new ConcurrentLinkedQueue<>();
+
+		volatile boolean closed;
+
+		@Override
+		public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message, String messageId) {
+			return Mono.fromRunnable(() -> this.sent.add(message));
+		}
+
+		@Override
+		public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message) {
+			return sendMessage(message, null);
+		}
+
+		@Override
+		public Mono<Void> closeGracefully() {
+			return Mono.fromRunnable(() -> this.closed = true);
+		}
+
+		@Override
+		public void close() {
+			this.closed = true;
+		}
+
+		@SuppressWarnings("unchecked")
+		@Override
+		public <T> T unmarshalFrom(Object data, TypeRef<T> typeRef) {
+			return (T) data;
+		}
+
+	}
+
+}