From 759828c0f5e81ba0b4939bbe44bc208f18d68fbd Mon Sep 17 00:00:00 2001 From: kami Date: Sun, 12 Jul 2026 13:17:38 +0400 Subject: [PATCH] fix(persistence): serialize SqliteEventStore reads with append transaction (#52) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reads (read/readFrom/lastSequence/allEvents/allSessionIds/lastGlobalSequence) shared the single JDBC Connection with append's transaction unsynchronized. The coroutine appendMutex only excludes append-vs-append; a read on the caller thread could hit the connection mid-transaction (setAutoCommit(false)..commit) and corrupt tx state — SQLite JDBC is not thread-safe. Guard every connection access with a shared JVM monitor (synchronized(connection)) via withConnection, so reads and the IO-thread append transaction are mutually exclusive. --- .../persistence/SqliteEventStore.kt | 96 +++++++++++-------- 1 file changed, 58 insertions(+), 38 deletions(-) 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 9a9167cf..41fd2157 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 @@ -72,19 +72,21 @@ class SqliteEventStore( appendMutex.withLock { artifactStore.flushBefore { withContext(Dispatchers.IO) { - stored = connection.transaction { - val existing = findByEventId(event.metadata.eventId) - check(existing == null) { "duplicate event_id: ${event.metadata.eventId}" } - val globalSeqVal = nextGlobalSequence() - val sessionSeqVal = nextSessionSequence(event.metadata.sessionId) - val s = StoredEvent( - metadata = event.metadata, - sequence = globalSeqVal, - sessionSequence = sessionSeqVal, - payload = event.payload, - ) - insert(s) - s + stored = withConnection { + connection.transaction { + val existing = findByEventId(event.metadata.eventId) + check(existing == null) { "duplicate event_id: ${event.metadata.eventId}" } + val globalSeqVal = nextGlobalSequence() + val sessionSeqVal = nextSessionSequence(event.metadata.sessionId) + val s = StoredEvent( + metadata = event.metadata, + sequence = globalSeqVal, + sessionSequence = sessionSeqVal, + payload = event.payload, + ) + insert(s) + s + } } } } @@ -102,20 +104,22 @@ class SqliteEventStore( appendMutex.withLock { artifactStore.flushBefore { withContext(Dispatchers.IO) { - stored = connection.transaction { - events.map { event -> - val existing = findByEventId(event.metadata.eventId) - check(existing == null) { "duplicate event_id: ${event.metadata.eventId}" } - val globalSeqVal = nextGlobalSequence() - val sessionSeqVal = nextSessionSequence(sessionId) - val s = StoredEvent( - metadata = event.metadata, - sequence = globalSeqVal, - sessionSequence = sessionSeqVal, - payload = event.payload, - ) - insert(s) - s + stored = withConnection { + connection.transaction { + events.map { event -> + val existing = findByEventId(event.metadata.eventId) + check(existing == null) { "duplicate event_id: ${event.metadata.eventId}" } + val globalSeqVal = nextGlobalSequence() + val sessionSeqVal = nextSessionSequence(sessionId) + val s = StoredEvent( + metadata = event.metadata, + sequence = globalSeqVal, + sessionSequence = sessionSeqVal, + payload = event.payload, + ) + insert(s) + s + } } } } @@ -127,7 +131,7 @@ class SqliteEventStore( return stored } - override fun read(sessionId: SessionId): List = + override fun read(sessionId: SessionId): List = withConnection { connection.prepareStatement( """ SELECT * FROM events @@ -144,8 +148,9 @@ class SqliteEventStore( } } } + } - override fun readFrom(sessionId: SessionId, fromSequence: Long): List = + override fun readFrom(sessionId: SessionId, fromSequence: Long): List = withConnection { connection.prepareStatement( """ SELECT * FROM events @@ -164,8 +169,9 @@ class SqliteEventStore( } } } + } - override fun lastSequence(sessionId: SessionId): Long? = + override fun lastSequence(sessionId: SessionId): Long? = withConnection { connection.prepareStatement( "SELECT MAX(session_sequence) FROM events WHERE session_id = ?" ).use { ps -> @@ -175,6 +181,7 @@ class SqliteEventStore( else null } } + } override fun subscribe(sessionId: SessionId): Flow = subscriptions.computeIfAbsent(sessionId) { MutableSharedFlow(replay = 0, extraBufferCapacity = 64) } @@ -183,17 +190,19 @@ class SqliteEventStore( override suspend fun lastGlobalSequence(): Long = withContext(Dispatchers.IO) { - connection.prepareStatement( - "SELECT COALESCE(MAX(sequence), 0) FROM events" - ).use { ps -> - ps.executeQuery().use { rs -> - rs.next() - rs.getLong(1) + withConnection { + connection.prepareStatement( + "SELECT COALESCE(MAX(sequence), 0) FROM events" + ).use { ps -> + ps.executeQuery().use { rs -> + rs.next() + rs.getLong(1) + } } } } - override fun allEvents(): Sequence = + override fun allEvents(): Sequence = withConnection { connection.prepareStatement( """ SELECT * FROM events @@ -208,8 +217,9 @@ class SqliteEventStore( } } }.asSequence() + } - override fun allSessionIds(): Set = + override fun allSessionIds(): Set = withConnection { mutableSetOf().apply { connection.prepareStatement("SELECT DISTINCT session_id FROM events").use { ps -> ps.executeQuery().use { rs -> @@ -217,9 +227,19 @@ class SqliteEventStore( } } } + } // ---------- helpers ---------- + /** + * Serializes all JDBC access to the shared [connection] on a single JVM monitor. Reads run on the + * caller thread while append's transaction runs on Dispatchers.IO; the coroutine [appendMutex] only + * excludes append-vs-append, so without this a read could hit the connection mid-transaction + * (SQLite JDBC is not thread-safe → corrupted tx state). ponytail: store-wide lock, one session at a + * time is fine here; shard per-connection only if read throughput ever matters. + */ + private inline fun withConnection(block: () -> T): T = synchronized(connection) { block() } + private fun nextGlobalSequence(): Long = connection.prepareStatement( "SELECT COALESCE(MAX(sequence), 0) FROM events"