feat(stream): bridge SessionManager termination to SSE transport cleanup

Sessions can terminate through paths that bypass the controller — user
session-limit eviction, grace-period expiry, idle cleanup — and the
previous wiring left transport-level state (active emitter, broadcaster
sender, heartbeat task) leaking on those paths. Inject the
SseStreamController via ObjectProvider into StreamComponentWiring and
invoke releaseTransportResources(sessionId) inside the
onSessionTerminated callback so every termination path releases the
same resources. The call is idempotent and safe to re-run when the
controller already released them via cleanupSession().

SseStreamController.cleanupSession is split: transport release moves to
the public releaseTransportResources(), and the controller-driven
session cleanup orchestrates subscription cleanup, transport release,
and SessionManager.terminateSession() in that order. setupCleanupCallbacks
gains an AtomicBoolean CAS gate because Spring fires both onTimeout and
onCompletion when an emitter times out, and without dedup the cleanup
work — including the log line and listener wake-ups — duplicated.

SessionManager.completeTermination now runs the onSessionTerminated
callback BEFORE clearing the session's subscriptions, so listeners can
iterate session.getSubscriptions() to release per-key state. The
callback is wrapped in try/catch so a misbehaving listener cannot leave
the session half-cleaned. Registry unregistration moves ahead of the
callback so concurrent lookups observe the termination immediately.
The user-limit branch in registerSession now evicts the oldest session
(FIFO) explicitly to admit the new connection, matching the behaviour
that was already implemented but only a comment-deep.

Author: thkwag

simplix-stream/src/main/java/dev/simplecore/simplix/stream/autoconfigure/SimpliXStreamCoreConfiguration.java

@@ -14,7 +14,9 @@
 import dev.simplecore.simplix.stream.infrastructure.local.LocalBroadcaster;
 import dev.simplecore.simplix.stream.infrastructure.local.LocalSessionRegistry;
 import dev.simplecore.simplix.stream.persistence.service.DbSessionRegistry;
+import dev.simplecore.simplix.stream.transport.sse.SseStreamController;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
@@ -161,18 +163,24 @@ public StreamComponentWiring streamComponentWiring(
             SessionManager sessionManager,
             StreamProperties properties,
             SimpliXStreamDataCollectorRegistry collectorRegistry,
+            ObjectProvider<SseStreamController> sseControllerProvider,
             @Autowired(required = false) EventStreamHandler eventStreamHandler,
             @Autowired(required = false) SimpliXStreamEventSourceRegistry eventSourceRegistry) {
         return new StreamComponentWiring(
                 subscriptionManager, schedulerManager, sessionManager, properties,
-                collectorRegistry, eventStreamHandler, eventSourceRegistry);
+                collectorRegistry, sseControllerProvider.getIfAvailable(),
+                eventStreamHandler, eventSourceRegistry);
     }
 
     /**
      * Wires stream components together after construction.
      * <p>
      * Routes subscriptions to either SchedulerManager (polling) or EventStreamHandler (event-based)
      * depending on whether a StreamEventSource is registered for the resource.
+     * Also bridges {@link SessionManager} termination events back to the SSE transport
+     * so non-controller-driven termination paths (user-limit eviction, grace expiry,
+     * idle cleanup) still release transport-level resources (active emitter, broadcaster
+     * sender, heartbeat task).
      */
     @Slf4j
     public static class StreamComponentWiring {
@@ -183,6 +191,7 @@ public StreamComponentWiring(
                 SessionManager sessionManager,
                 StreamProperties properties,
                 SimpliXStreamDataCollectorRegistry collectorRegistry,
+                SseStreamController sseStreamController,
                 EventStreamHandler eventStreamHandler,
                 SimpliXStreamEventSourceRegistry eventSourceRegistry) {
 
@@ -228,7 +237,9 @@ public StreamComponentWiring(
                 }
             });
 
-            // Wire session termination to subscription cleanup
+            // Wire session termination to subscription cleanup AND transport resource release.
+            // SessionManager invokes this callback BEFORE clearing the session's subscriptions,
+            // so iterating session.getSubscriptions() yields the live keys.
             sessionManager.setOnSessionTerminated(session -> {
                 for (var key : session.getSubscriptions()) {
                     String resource = key.getResource();
@@ -244,6 +255,13 @@ public StreamComponentWiring(
                 if (eventSourceEnabled) {
                     eventStreamHandler.removeSubscriberFromAll(session.getId());
                 }
+
+                // Release SSE transport resources for paths that bypass the controller
+                // (user-limit eviction, grace expiry, idle cleanup). Idempotent — safe to
+                // re-run when the controller already released them in cleanupSession().
+                if (sseStreamController != null) {
+                    sseStreamController.releaseTransportResources(session.getId());
+                }
             });
 
             if (eventSourceEnabled) {

simplix-stream/src/main/java/dev/simplecore/simplix/stream/core/session/SessionManager.java

@@ -84,8 +84,9 @@ public StreamSession createSession(String userId, TransportType transportType) {
             long currentCount = sessionRegistry.countByUserId(userId);
             if (currentCount >= maxPerUser) {
                 log.warn("User {} has reached session limit: {}/{}", userId, currentCount, maxPerUser);
-                // Option: terminate oldest session or reject new connection
-                // Currently rejecting - can be configured
+                // Policy: evict oldest session (FIFO) to admit the new connection.
+                // Alternative would be to reject the new connection — make this configurable
+                // via properties when both policies are needed.
                 terminateOldestSession(userId);
             }
         }
@@ -295,25 +296,32 @@ public Set<SubscriptionKey> terminateSession(String sessionId) {
      * Complete session termination after state is marked as terminated.
      * <p>
      * This method should be called after acquiring gracePeriodLock and marking
-     * the session as terminated.
+     * the session as terminated. The termination callback runs while the
+     * session's subscriptions are still populated so listeners can iterate
+     * them; the subscriptions are cleared afterwards.
      *
      * @param session the session to terminate
      * @return the cleared subscriptions
      */
     private Set<SubscriptionKey> completeTermination(StreamSession session) {
-        Set<SubscriptionKey> clearedSubscriptions = session.clearSubscriptions();
-
-        // Unregister from registry
+        // Unregister from registry first so concurrent lookups see termination
         sessionRegistry.unregister(session.getId());
 
-        log.info("Session terminated: {} (user={}, subscriptions={})",
-                session.getId(), session.getUserId(), clearedSubscriptions.size());
-
-        // Notify callback
+        // Notify callback BEFORE clearing subscriptions so listeners can read them
         if (onSessionTerminated != null) {
-            onSessionTerminated.accept(session);
+            try {
+                onSessionTerminated.accept(session);
+            } catch (RuntimeException e) {
+                log.warn("onSessionTerminated callback threw for session {}: {}",
+                        session.getId(), e.getMessage(), e);
+            }
         }
 
+        Set<SubscriptionKey> clearedSubscriptions = session.clearSubscriptions();
+
+        log.info("Session terminated: {} (user={}, subscriptions={})",
+                session.getId(), session.getUserId(), clearedSubscriptions.size());
+
         return clearedSubscriptions;
     }
 

simplix-stream/src/main/java/dev/simplecore/simplix/stream/transport/sse/SseStreamController.java

@@ -32,6 +32,7 @@
 import java.time.Duration;
 import java.util.*;
 import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * REST controller for SSE streaming connections.
@@ -325,7 +326,16 @@ private void sendReconnectionMessage(SseStreamSession sseSession, StreamSession
     }
 
     private void setupCleanupCallbacks(SseEmitter emitter, String sessionId) {
+        // Spring fires both onTimeout AND onCompletion when an SSE emitter times out
+        // (timeout callback first, then completion when the response is finalized).
+        // Without dedup the cleanup runs twice — markDisconnected is internally guarded
+        // but the log line and listener wake-ups still duplicate. CAS gate runs cleanup
+        // exactly once per emitter regardless of which callback fires first.
+        AtomicBoolean cleaned = new AtomicBoolean(false);
         Runnable cleanup = () -> {
+            if (!cleaned.compareAndSet(false, true)) {
+                return;
+            }
             log.debug("SSE connection closed, starting grace period for session: {}", sessionId);
             SseStreamSession sseSession = activeSessions.get(sessionId);
             if (sseSession != null) {
@@ -394,22 +404,44 @@ private void sendConnectionMessage(SseStreamSession sseSession, String sessionId
     }
 
     private void cleanupSession(String sessionId) {
-        SseStreamSession sseSession = activeSessions.remove(sessionId);
-        if (sseSession == null) {
+        if (!activeSessions.containsKey(sessionId)) {
             return; // Already cleaned up — idempotent guard
         }
 
-        cancelHeartbeat(sessionId);
-        sseSession.close();
-
-        broadcaster.unregisterSender(sessionId);
+        // Release transport resources first; per-key subscription cleanup runs through
+        // SubscriptionManager so scheduler/event listeners receive removal notifications.
         subscriptionManager.clearSubscriptions(sessionId);
+        releaseTransportResources(sessionId);
         sessionManager.terminateSession(sessionId);
+
+        log.info("Session cleaned up: {}", sessionId);
+    }
+
+    /**
+     * Release transport-level resources for a session.
+     * <p>
+     * Closes the SSE emitter, removes the session from the active map and the
+     * broadcaster's sender registry, cancels the heartbeat task, and removes the
+     * session from any event-source registries. All operations are idempotent so
+     * this method is safe to call multiple times for the same session ID.
+     * <p>
+     * Invoked from two paths: (a) controller-driven cleanup (explicit disconnect,
+     * heartbeat-detected invalid session), and (b) the wiring-config callback when
+     * SessionManager terminates a session without going through the controller
+     * (user-limit eviction, grace-period expiry).
+     *
+     * @param sessionId the session ID
+     */
+    public void releaseTransportResources(String sessionId) {
+        cancelHeartbeat(sessionId);
+        SseStreamSession sseSession = activeSessions.remove(sessionId);
+        if (sseSession != null) {
+            sseSession.close();
+        }
+        broadcaster.unregisterSender(sessionId);
         if (eventStreamHandler != null) {
             eventStreamHandler.removeSubscriberFromAll(sessionId);
         }
-
-        log.info("Session cleaned up: {}", sessionId);
     }
 
     private List<Subscription> convertToSubscriptions(SubscriptionRequest request, StreamSession session) {

simplix-stream/src/test/java/dev/simplecore/simplix/stream/autoconfigure/StreamComponentWiringTest.java

@@ -10,6 +10,7 @@
 import dev.simplecore.simplix.stream.core.subscription.SubscriptionManager;
 import dev.simplecore.simplix.stream.eventsource.EventStreamHandler;
 import dev.simplecore.simplix.stream.eventsource.SimpliXStreamEventSourceRegistry;
+import dev.simplecore.simplix.stream.transport.sse.SseStreamController;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Nested;
@@ -80,7 +81,7 @@ void shouldRouteToSchedulerManager() {
 
             new SimpliXStreamCoreConfiguration.StreamComponentWiring(
                     subscriptionManager, schedulerManager, sessionManager, properties,
-                    collectorRegistry, null, null);
+                    collectorRegistry, null, null, null);
 
             verify(subscriptionManager).setOnSubscriptionAdded(addCaptor.capture());
             verify(subscriptionManager).setOnSubscriptionRemoved(removeCaptor.capture());
@@ -102,7 +103,7 @@ void shouldUseCollectorInterval() {
 
             new SimpliXStreamCoreConfiguration.StreamComponentWiring(
                     subscriptionManager, schedulerManager, sessionManager, properties,
-                    collectorRegistry, null, null);
+                    collectorRegistry, null, null, null);
 
             verify(subscriptionManager).setOnSubscriptionAdded(addCaptor.capture());
 
@@ -125,7 +126,7 @@ void shouldRouteRemovalToScheduler() {
 
             new SimpliXStreamCoreConfiguration.StreamComponentWiring(
                     subscriptionManager, schedulerManager, sessionManager, properties,
-                    collectorRegistry, null, null);
+                    collectorRegistry, null, null, null);
 
             verify(subscriptionManager).setOnSubscriptionRemoved(removeCaptor.capture());
 
@@ -143,7 +144,7 @@ void shouldCleanUpOnSessionTermination() {
 
             new SimpliXStreamCoreConfiguration.StreamComponentWiring(
                     subscriptionManager, schedulerManager, sessionManager, properties,
-                    collectorRegistry, null, null);
+                    collectorRegistry, null, null, null);
 
             verify(sessionManager).setOnSessionTerminated(terminateCaptor.capture());
 
@@ -173,7 +174,7 @@ void shouldRouteToEventHandler() {
 
             new SimpliXStreamCoreConfiguration.StreamComponentWiring(
                     subscriptionManager, schedulerManager, sessionManager, properties,
-                    collectorRegistry, eventStreamHandler, eventSourceRegistry);
+                    collectorRegistry, null, eventStreamHandler, eventSourceRegistry);
 
             verify(subscriptionManager).setOnSubscriptionAdded(addCaptor.capture());
 
@@ -196,7 +197,7 @@ void shouldRouteRemovalToEventHandler() {
 
             new SimpliXStreamCoreConfiguration.StreamComponentWiring(
                     subscriptionManager, schedulerManager, sessionManager, properties,
-                    collectorRegistry, eventStreamHandler, eventSourceRegistry);
+                    collectorRegistry, null, eventStreamHandler, eventSourceRegistry);
 
             verify(subscriptionManager).setOnSubscriptionRemoved(removeCaptor.capture());
 
@@ -219,7 +220,7 @@ void shouldCleanUpFromEventHandlerOnTermination() {
 
             new SimpliXStreamCoreConfiguration.StreamComponentWiring(
                     subscriptionManager, schedulerManager, sessionManager, properties,
-                    collectorRegistry, eventStreamHandler, eventSourceRegistry);
+                    collectorRegistry, null, eventStreamHandler, eventSourceRegistry);
 
             verify(sessionManager).setOnSessionTerminated(terminateCaptor.capture());
 
@@ -234,6 +235,52 @@ void shouldCleanUpFromEventHandlerOnTermination() {
             verify(eventStreamHandler).removeSubscriberFromAll("session-1");
         }
 
+        @Test
+        @DisplayName("should release SSE transport resources on session termination when controller present")
+        void shouldReleaseSseTransportResourcesOnTermination() {
+            ArgumentCaptor<Consumer<StreamSession>> terminateCaptor =
+                    ArgumentCaptor.forClass(Consumer.class);
+            SseStreamController sseController = mock(SseStreamController.class);
+
+            when(eventSourceRegistry.getRegisteredResources()).thenReturn(Set.of("events"));
+
+            new SimpliXStreamCoreConfiguration.StreamComponentWiring(
+                    subscriptionManager, schedulerManager, sessionManager, properties,
+                    collectorRegistry, sseController, eventStreamHandler, eventSourceRegistry);
+
+            verify(sessionManager).setOnSessionTerminated(terminateCaptor.capture());
+
+            StreamSession session = mock(StreamSession.class);
+            when(session.getSubscriptions()).thenReturn(Set.of());
+            when(session.getId()).thenReturn("session-leak-fix");
+
+            terminateCaptor.getValue().accept(session);
+
+            verify(sseController).releaseTransportResources("session-leak-fix");
+        }
+
+        @Test
+        @DisplayName("should not fail termination callback when SSE controller is absent")
+        void shouldHandleAbsentSseControllerOnTermination() {
+            ArgumentCaptor<Consumer<StreamSession>> terminateCaptor =
+                    ArgumentCaptor.forClass(Consumer.class);
+
+            when(eventSourceRegistry.getRegisteredResources()).thenReturn(Set.of("events"));
+
+            new SimpliXStreamCoreConfiguration.StreamComponentWiring(
+                    subscriptionManager, schedulerManager, sessionManager, properties,
+                    collectorRegistry, null, eventStreamHandler, eventSourceRegistry);
+
+            verify(sessionManager).setOnSessionTerminated(terminateCaptor.capture());
+
+            StreamSession session = mock(StreamSession.class);
+            when(session.getSubscriptions()).thenReturn(Set.of());
+            when(session.getId()).thenReturn("session-no-sse");
+
+            // Must not throw — wiring should tolerate WebSocket-only or no-transport deployments
+            terminateCaptor.getValue().accept(session);
+        }
+
         @Test
         @DisplayName("should route non-event resources to scheduler even with event source enabled")
         void shouldRouteNonEventResourcesToScheduler() {
@@ -245,7 +292,7 @@ void shouldRouteNonEventResourcesToScheduler() {
 
             new SimpliXStreamCoreConfiguration.StreamComponentWiring(
                     subscriptionManager, schedulerManager, sessionManager, properties,
-                    collectorRegistry, eventStreamHandler, eventSourceRegistry);
+                    collectorRegistry, null, eventStreamHandler, eventSourceRegistry);
 
             verify(subscriptionManager).setOnSubscriptionAdded(addCaptor.capture());
 

simplix-stream/src/test/java/dev/simplecore/simplix/stream/core/session/SessionManagerTest.java

@@ -633,6 +633,47 @@ void shouldNotInvokeCallbackWhenNotSet() {
             // No callback set
             assertDoesNotThrow(() -> sessionManager.terminateSession(session.getId()));
         }
+
+        @Test
+        @DisplayName("should expose subscriptions to callback before clearing them")
+        void shouldExposeSubscriptionsToCallbackBeforeClearing() {
+            // Regression: callback used to fire AFTER clearSubscriptions(), so listeners
+            // iterating session.getSubscriptions() saw an empty set and could not perform
+            // per-subscription cleanup (e.g., removing scheduler subscribers).
+            StreamSession session = StreamSession.create("user123", TransportType.SSE);
+            SubscriptionKey k1 = SubscriptionKey.of("stock", Map.of("symbol", "AAPL"));
+            SubscriptionKey k2 = SubscriptionKey.of("news", Map.of("topic", "tech"));
+            session.addSubscription(k1);
+            session.addSubscription(k2);
+            when(sessionRegistry.findById(session.getId())).thenReturn(Optional.of(session));
+
+            AtomicReference<Set<SubscriptionKey>> snapshotInCallback = new AtomicReference<>();
+            sessionManager.setOnSessionTerminated(s ->
+                    snapshotInCallback.set(s.getSubscriptions()));
+
+            sessionManager.terminateSession(session.getId());
+
+            assertNotNull(snapshotInCallback.get());
+            assertEquals(Set.of(k1, k2), snapshotInCallback.get(),
+                    "callback must see subscriptions intact; reordering of clear/callback regressed");
+            // After termination the session itself is cleared
+            assertTrue(session.getSubscriptions().isEmpty());
+        }
+
+        @Test
+        @DisplayName("should still terminate when callback throws")
+        void shouldStillTerminateWhenCallbackThrows() {
+            StreamSession session = StreamSession.create("user123", TransportType.SSE);
+            when(sessionRegistry.findById(session.getId())).thenReturn(Optional.of(session));
+
+            sessionManager.setOnSessionTerminated(s -> {
+                throw new RuntimeException("listener boom");
+            });
+
+            assertDoesNotThrow(() -> sessionManager.terminateSession(session.getId()));
+            verify(sessionRegistry).unregister(session.getId());
+            assertEquals(SessionState.TERMINATED, session.getState());
+        }
     }
 
     @Nested