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

fix: harden event store, serialization, and grant engine; wire router chat

Browse Source 
Event log integrity (sole source of truth):
- SqliteEventStore: serialize append/appendAll under a Mutex, enable WAL +
  busy_timeout + synchronous=NORMAL, roll back on any Throwable. Replaces the
  app-computed `SELECT MAX(seq)+1` read-modify-write that dropped events under
  concurrent appends (race was invisible to CI: only the in-memory store and
  single-threaded :memory: tests were ever exercised).
- Serialization: encodeDefaults + ignoreUnknownKeys, explicit @SerialName on all
  46 EventPayload subclasses and event-referenced enum constants, @Serializable on
  RiskLevel/RiskAction. Guards the append-only log against silent corruption from
  future class renames, field removals, or default-value changes.

Approval/grant security (Invariant: approvals cannot widen authority):
- Tier authorization is now a ceiling (request.tier.level <= max granted) instead
  of set membership; SESSION grants bind to a specific toolName instead of matching
  everything; handleCreateGrant rejects empty/over-tier/scopeless grants.

Router chat wiring (Phase 0-2): ChatInput round-trip, StartChatSession from IDLE,
router-panel layout, workflows behind Ctrl+W.

Tests: SqliteEventStore concurrency, serialization round-trip + discriminator
stability + unknown-key tolerance, adversarial grant scope/tier; updated existing
approval and reducer tests for the new grant semantics and StartChatSession effect.
This commit is contained in:
kami
2026-05-28 22:39:15 +04:00
parent 57d2237ba0
commit cdee5f2245
43 changed files with 690 additions and 223 deletions
Download Patch File Download Diff File Expand all files Collapse all files
infrastructure/persistence/build.gradle
+1
View File
@@ -12,6 +12,7 @@ dependencies {
implementation(project(":core:artifacts-store"))
implementation "org.xerial:sqlite-jdbc"
testImplementation(testFixtures(project(":testing:contracts")))
testImplementation(project(":testing:fixtures"))
testImplementation "org.junit.jupiter:junit-jupiter"
}
infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/SqliteEventStore.kt
+41 -31
View File
@@ -17,6 +17,8 @@ import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.asSharedFlow
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlinx.coroutines.withContext
import kotlinx.datetime.Instant
import java.sql.Connection
@@ -28,6 +30,7 @@ class SqliteEventStore(
private val jsonSerializer: JsonEventSerializer = JsonEventSerializer(eventJson),
private val artifactStore: ArtifactStore,
) : EventStore {
private val appendMutex = Mutex()
private val subscriptions = ConcurrentHashMap<SessionId, MutableSharedFlow<StoredEvent>>()
private val globalFlow = MutableSharedFlow<StoredEvent>(
replay = 0,
@@ -37,6 +40,9 @@ class SqliteEventStore(
init {
connection.createStatement().use { stmt ->
stmt.execute("PRAGMA journal_mode=WAL;")
stmt.execute("PRAGMA busy_timeout=5000;")
stmt.execute("PRAGMA synchronous=NORMAL;")
stmt.execute(
"""
CREATE TABLE IF NOT EXISTS events (
@@ -63,21 +69,23 @@ class SqliteEventStore(
override suspend fun append(event: NewEvent): StoredEvent {
var stored: StoredEvent? = null
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
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
}
}
}
}
@@ -91,22 +99,24 @@ class SqliteEventStore(
if (events.isEmpty()) return emptyList()
val sessionId = events.first().metadata.sessionId
var stored: List<StoredEvent> = emptyList()
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
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
}
}
}
}
infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/util/JDBCHelper.kt
+1 -2
View File
@@ -1,7 +1,6 @@
package com.correx.infrastructure.persistence.util
import java.sql.Connection
import java.sql.SQLException
object JDBCHelper {
inline fun <T> Connection.transaction(block: () -> T): T {
@@ -13,7 +12,7 @@ object JDBCHelper {
commit()
return result
} catch (e: SQLException) {
} catch (e: Throwable) {
rollback()
throw e
} finally {
infrastructure/persistence/src/test/kotlin/com/correx/infrastructure/persistence/SqliteEventStoreConcurrencyTest.kt
+53
View File
@@ -0,0 +1,53 @@
package com.correx.infrastructure.persistence
import com.correx.core.events.types.EventId
import com.correx.core.events.types.SessionId
import com.correx.testing.contracts.fixtures.artifactstore.NoopArtifactStore
import com.correx.testing.fixtures.EventFixtures
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.runBlocking
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Assertions.assertTrue
import org.junit.jupiter.api.Test
import org.junit.jupiter.api.io.TempDir
import java.nio.file.Path
import java.sql.DriverManager
class SqliteEventStoreConcurrencyTest {
@Test
fun `concurrent appends to same session produce no dropped or duplicate sequence numbers`(
@TempDir tempDir: Path,
): Unit = runBlocking {
val dbFile = tempDir.resolve("events.db").toAbsolutePath().toString()
val conn = DriverManager.getConnection("jdbc:sqlite:$dbFile")
val store = SqliteEventStore(conn, artifactStore = NoopArtifactStore())
val sessionId = SessionId("concurrent-session")
val n = 50
(1..n)
.map { i ->
async(Dispatchers.IO) {
store.append(EventFixtures.newEvent(EventId("e-$i"), sessionId))
}
}
.awaitAll()
val events = store.read(sessionId)
assertEquals(n, events.size, "Expected exactly $n events but got ${events.size}")
val sessionSeqs = events.map { it.sessionSequence }.sorted()
assertEquals((1L..n.toLong()).toList(), sessionSeqs, "Session sequences must be exactly 1..$n with no gaps or duplicates")
val globalSeqs = events.map { it.sequence }
assertEquals(globalSeqs.size, globalSeqs.toSet().size, "Global sequence values must all be unique")
globalSeqs.forEach { seq ->
assertTrue(seq >= 1L, "Global sequence $seq must be >= 1")
}
}
}