diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt b/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt index df96c111..d3f048a6 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt @@ -65,6 +65,9 @@ import java.util.UUID private val log = LoggerFactory.getLogger(GlobalStreamHandler::class.java) private const val BUFFER_CAPACITY = 1024 + +/** Raised to tear down a WS connection whose forward buffer overflowed, so the client reconnects. */ +private class SlowClientException : RuntimeException("global WS stream buffer overflow") private const val RESOURCE_PUSH_INTERVAL_MS = 2500L private const val BYTES_PER_MB = 1024L * 1024L @@ -788,8 +791,17 @@ internal suspend fun streamGlobal( is SharedFlow -> source.onSubscription { subscribed.complete(Unit) } else -> source.onStart { subscribed.complete(Unit) } } + // trySend, never send: a blocking send here would back-pressure globalFlow (SUSPEND overflow), + // suspending append() and stalling the whole kernel on one wedged client. On overflow we instead + // drop the connection (close the buffer); the client reconnects and re-syncs via replaySnapshot. val subscription = launch { - signaled.collect { buffer.send(it) } + signaled.collect { + val result = buffer.trySend(it) + if (result.isFailure && !result.isClosed) { + log.warn("global WS stream buffer overflow (slow client); dropping connection to re-sync") + buffer.close(SlowClientException()) + } + } } try { diff --git a/infrastructure/persistence/build.gradle b/infrastructure/persistence/build.gradle index 14e1144c..0e0af0e6 100644 --- a/infrastructure/persistence/build.gradle +++ b/infrastructure/persistence/build.gradle @@ -11,6 +11,7 @@ dependencies { implementation(project(":core:artifacts")) implementation(project(":core:artifacts-store")) implementation "org.xerial:sqlite-jdbc" + implementation "org.slf4j:slf4j-api:2.0.16" testImplementation(testFixtures(project(":testing:contracts"))) testImplementation(project(":testing:fixtures")) testImplementation "org.junit.jupiter:junit-jupiter" diff --git a/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/InMemoryEventStore.kt b/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/InMemoryEventStore.kt index efdf655a..054b0c4c 100644 --- a/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/InMemoryEventStore.kt +++ b/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/InMemoryEventStore.kt @@ -9,9 +9,12 @@ import kotlinx.coroutines.channels.BufferOverflow import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.asSharedFlow +import org.slf4j.LoggerFactory import java.util.concurrent.* import java.util.concurrent.atomic.* +private val log = LoggerFactory.getLogger(InMemoryEventStore::class.java) + class InMemoryEventStore : EventStore { private val streams = ConcurrentHashMap>() private val sequences = ConcurrentHashMap() @@ -33,7 +36,7 @@ class InMemoryEventStore : EventStore { } stored = doAppend(event, stream) } - subscriptions[event.metadata.sessionId]?.tryEmit(stored) + subscriptions[event.metadata.sessionId]?.let { warnIfDropped(it.tryEmit(stored), stored) } globalFlow.emit(stored) return stored } @@ -47,7 +50,7 @@ class InMemoryEventStore : EventStore { stored = events.map { doAppend(it, stream) } } val flow = subscriptions[sessionId] - if (flow != null) stored.forEach { flow.tryEmit(it) } + if (flow != null) stored.forEach { warnIfDropped(flow.tryEmit(it), it) } stored.forEach { globalFlow.emit(it) } return stored } @@ -77,6 +80,19 @@ class InMemoryEventStore : EventStore { override fun allSessionIds(): Set = sequences.keys + /** A lagging per-session collector's buffer (extraBufferCapacity 64) is full; the event is durably + * stored but dropped from the live SharedFlow. Surface it rather than swallow the false return. */ + private fun warnIfDropped(emitted: Boolean, event: StoredEvent) { + if (!emitted) { + log.warn( + "dropped live subscription emit for session {} seq {} ({}): buffer full, collector lagging", + event.metadata.sessionId.value, + event.sequence, + event.payload::class.simpleName, + ) + } + } + private fun doAppend(event: NewEvent, stream: MutableList): StoredEvent { val sessionSeq = sequences.computeIfAbsent(event.metadata.sessionId) { AtomicLong(0) } .incrementAndGet() diff --git a/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/SqliteEventStore.kt b/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/SqliteEventStore.kt index 6b450326..7b6175ee 100644 --- a/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/SqliteEventStore.kt +++ b/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/SqliteEventStore.kt @@ -21,10 +21,13 @@ import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock import kotlinx.coroutines.withContext import kotlinx.datetime.Instant +import org.slf4j.LoggerFactory import java.sql.Connection import java.sql.ResultSet import java.util.concurrent.* +private val log = LoggerFactory.getLogger(SqliteEventStore::class.java) + class SqliteEventStore( private val connection: Connection, private val jsonSerializer: JsonEventSerializer = JsonEventSerializer(eventJson), @@ -89,7 +92,7 @@ class SqliteEventStore( } } val result = checkNotNull(stored) - subscriptions[event.metadata.sessionId]?.tryEmit(result) + subscriptions[event.metadata.sessionId]?.let { warnIfDropped(it.tryEmit(result), result) } globalFlow.emit(result) return result } @@ -120,7 +123,7 @@ class SqliteEventStore( } } val flow = subscriptions[sessionId] - if (flow != null) stored.forEach { flow.tryEmit(it) } + if (flow != null) stored.forEach { warnIfDropped(flow.tryEmit(it), it) } stored.forEach { globalFlow.emit(it) } return stored } @@ -234,6 +237,24 @@ class SqliteEventStore( */ private inline fun withConnection(block: () -> T): T = synchronized(connection) { block() } + /** + * A per-session subscription's buffer (extraBufferCapacity 64) is full and this event was dropped. + * The event is durably persisted (the drop is only on the live SharedFlow), but a lagging in-process + * collector — e.g. LiveArtifactRepository — silently diverges from the log. Surface it instead of + * swallowing the false return. ponytail: log-only; give that collector a bounded rebuild-on-lag if + * divergence ever bites in practice. + */ + private fun warnIfDropped(emitted: Boolean, event: StoredEvent) { + if (!emitted) { + log.warn( + "dropped live subscription emit for session {} seq {} ({}): buffer full, collector lagging", + event.metadata.sessionId.value, + event.sequence, + event.payload::class.simpleName, + ) + } + } + /** * Inserts [event], letting SQLite assign both the global `sequence` and the per-session * `session_sequence` as `MAX+1` subqueries evaluated inside the INSERT itself, and RETURNs the