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

fix(persistence): assign event sequences atomically inside INSERT (#56)

Browse Source 
nextGlobalSequence()/nextSessionSequence() computed MAX+1 with two extra
SELECT round-trips, guarded only by an in-process mutex — two processes on the
same correx.db (CLI + server) could read the same MAX and collide on the unique
index. Compute both the global sequence and session_sequence as MAX+1 subqueries
inside the INSERT and RETURNING the assigned values. SQLite serializes writers,
so the subqueries are atomic across processes; the unique index is the backstop.
Also removes the two pre-SELECTs per append.
This commit is contained in:
kami
2026-07-12 13:19:50 +04:00
parent 759828c0f5
commit 65d5e9de83
1 changed files with 30 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/SqliteEventStore.kt
+29 -41
View File
@@ -76,16 +76,13 @@ class SqliteEventStore(
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(
val (globalSeqVal, sessionSeqVal) = insertAssigningSequences(event)
StoredEvent(
metadata = event.metadata,
sequence = globalSeqVal,
sessionSequence = sessionSeqVal,
payload = event.payload,
)
insert(s)
s
}
}
}
@@ -109,16 +106,13 @@ class SqliteEventStore(
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(
val (globalSeqVal, sessionSeqVal) = insertAssigningSequences(event)
StoredEvent(
metadata = event.metadata,
sequence = globalSeqVal,
sessionSequence = sessionSeqVal,
payload = event.payload,
)
insert(s)
s
}
}
}
@@ -240,29 +234,16 @@ class SqliteEventStore(
*/
private inline fun <T> withConnection(block: () -> T): T = synchronized(connection) { block() }
private fun nextGlobalSequence(): Long =
connection.prepareStatement(
"SELECT COALESCE(MAX(sequence), 0) FROM events"
).use { ps ->
ps.executeQuery().use { rs ->
rs.next()
rs.getLong(1) + 1
}
}
private fun nextSessionSequence(sessionId: SessionId): Long =
connection.prepareStatement(
"SELECT COALESCE(MAX(session_sequence), 0) FROM events WHERE session_id = ?"
).use { ps ->
ps.setString(1, sessionId.value)
ps.executeQuery().use { rs ->
rs.next()
rs.getLong(1) + 1
}
}
/**
* 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
* assigned values. SQLite serializes writers (one write transaction at a time), so the subqueries
* are atomic against other processes on the same DB — closing the CLI+server MAX+1 race that the
* old read-then-insert had (the unique index on `sequence` is the backstop). Within one appendAll
* transaction, each INSERT sees the prior rows, so a batch increments correctly.
*/
@Suppress("MagicNumber")
private fun insert(event: StoredEvent) {
private fun insertAssigningSequences(event: NewEvent): Pair<Long, Long> =
connection.prepareStatement(
"""
INSERT INTO events (
@@ -275,19 +256,26 @@ class SqliteEventStore(
causation_id,
correlation_id,
payload
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
) VALUES (
?, ?,
(SELECT COALESCE(MAX(sequence), 0) + 1 FROM events),
(SELECT COALESCE(MAX(session_sequence), 0) + 1 FROM events WHERE session_id = ?),
?, ?, ?, ?, ?
)
RETURNING sequence, session_sequence
""".trimIndent(),
).use { ps ->
ps.setString(1, event.metadata.eventId.value)
ps.setString(2, event.metadata.sessionId.value)
ps.setLong(3, event.sequence)
ps.setLong(4, event.sessionSequence)
ps.setString(5, event.metadata.timestamp.toString())
ps.setInt(6, event.metadata.schemaVersion)
ps.setString(7, event.metadata.causationId?.value)
ps.setString(8, event.metadata.correlationId?.value)
ps.setString(9, jsonSerializer.serialize(event.payload))
ps.executeUpdate()
ps.setString(3, event.metadata.sessionId.value)
ps.setString(4, event.metadata.timestamp.toString())
ps.setInt(5, event.metadata.schemaVersion)
ps.setString(6, event.metadata.causationId?.value)
ps.setString(7, event.metadata.correlationId?.value)
ps.setString(8, jsonSerializer.serialize(event.payload))
ps.executeQuery().use { rs ->
rs.next()
rs.getLong("sequence") to rs.getLong("session_sequence")
}
}