kami/correx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(persistence): serialize SqliteEventStore reads with append transaction (#52)

Browse Source 
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.
This commit is contained in:
kami
2026-07-12 13:17:38 +04:00
parent bb73bc9371
commit 759828c0f5
1 changed files with 58 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/SqliteEventStore.kt
+58 -38
View File
@@ -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<StoredEvent> =
override fun read(sessionId: SessionId): List<StoredEvent> = withConnection {
connection.prepareStatement(
"""
SELECT * FROM events
@@ -144,8 +148,9 @@ class SqliteEventStore(
}
}
}
}
override fun readFrom(sessionId: SessionId, fromSequence: Long): List<StoredEvent> =
override fun readFrom(sessionId: SessionId, fromSequence: Long): List<StoredEvent> = 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<StoredEvent> =
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<StoredEvent> =
override fun allEvents(): Sequence<StoredEvent> = withConnection {
connection.prepareStatement(
"""
SELECT * FROM events
@@ -208,8 +217,9 @@ class SqliteEventStore(
}
}
}.asSequence()
}
override fun allSessionIds(): Set<SessionId> =
override fun allSessionIds(): Set<SessionId> = withConnection {
mutableSetOf<SessionId>().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 <T> withConnection(block: () -> T): T = synchronized(connection) { block() }
private fun nextGlobalSequence(): Long =
connection.prepareStatement(
"SELECT COALESCE(MAX(sequence), 0) FROM events"