fix(kernel): stop compaction self-deadlocking long runs on re-entrant artifact Mutex

JournalCompactionService and TierContextSummarizer wrapped their event emit
in artifactStore.flushBefore { }. But emit -> SqliteEventStore.append already
calls flushBefore internally, re-acquiring CasArtifactStore's non-reentrant
Mutex -> the coroutine parks forever (no CPU, no thread, no exception, no
terminal event). Only fires once the journal crosses the compaction threshold,
i.e. exactly on long runs.

Emit directly; append()'s own flushBefore still fsyncs artifacts before the
referencing event is persisted, so durability ordering is preserved. Adds a
regression test with a lock-holding fake store (the prior fake took no lock,
which is why the deadlock escaped tests).

Vikunja #244.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LG7sGEbVJQncHJtsbPFJZm
This commit is contained in:
2026-07-17 16:17:56 +04:00
parent 9e62097c97
commit e25e9e46fd
3 changed files with 51 additions and 21 deletions
@@ -43,16 +43,18 @@ class JournalCompactionService(
} }
val summaryArtifactId = artifactStore.put(summaryText.toByteArray(Charsets.UTF_8)) val summaryArtifactId = artifactStore.put(summaryText.toByteArray(Charsets.UTF_8))
artifactStore.flushBefore { // Emit directly, NOT inside flushBefore { }: append() already fsyncs artifacts before it
emit( // persists the event, so wrapping the emit here re-acquires the (non-reentrant) artifact
JournalCompactedEvent( // Mutex that flushBefore already holds → self-deadlock. Only surfaces on long runs, which
sessionId = sessionId, // are the ones that cross the compaction threshold.
coversThroughSequence = throughSequence, emit(
summaryArtifactId = summaryArtifactId, JournalCompactedEvent(
lowSalienceOmittedCount = lowCount, sessionId = sessionId,
), coversThroughSequence = throughSequence,
) summaryArtifactId = summaryArtifactId,
} lowSalienceOmittedCount = lowCount,
),
)
return true return true
} }
} }
@@ -48,17 +48,18 @@ class TierContextSummarizer(
} }
val summaryText = summarize(prompt) val summaryText = summarize(prompt)
val summaryArtifactId = artifactStore.put(summaryText.toByteArray(Charsets.UTF_8)) val summaryArtifactId = artifactStore.put(summaryText.toByteArray(Charsets.UTF_8))
artifactStore.flushBefore { // Emit directly, NOT inside flushBefore { }: append() already fsyncs artifacts before it
emit( // persists the event, so wrapping the emit here re-acquires the (non-reentrant) artifact
ContextSummarizedEvent( // Mutex that flushBefore already holds → self-deadlock (same bug as JournalCompactionService).
sessionId = sessionId, emit(
coversThroughSequence = coversThroughSequence, ContextSummarizedEvent(
summaryArtifactId = summaryArtifactId, sessionId = sessionId,
tier = tier, coversThroughSequence = coversThroughSequence,
entriesOmittedCount = freeformEntries.size, summaryArtifactId = summaryArtifactId,
), tier = tier,
) entriesOmittedCount = freeformEntries.size,
} ),
)
return true return true
} }
} }
@@ -9,6 +9,9 @@ import com.correx.core.journal.model.DecisionKind
import com.correx.core.journal.model.DecisionRecord import com.correx.core.journal.model.DecisionRecord
import com.correx.core.utils.TypeId import com.correx.core.utils.TypeId
import kotlinx.coroutines.runBlocking import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlinx.coroutines.withTimeout
import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Assertions.assertFalse import org.junit.jupiter.api.Assertions.assertFalse
import org.junit.jupiter.api.Assertions.assertTrue import org.junit.jupiter.api.Assertions.assertTrue
@@ -70,6 +73,30 @@ class JournalCompactionServiceTest {
assertEquals(fixedArtifactId, event.summaryArtifactId) assertEquals(fixedArtifactId, event.summaryArtifactId)
} }
// Regression: append() flushes artifacts inside the SAME non-reentrant Mutex that flushBefore
// holds. If compaction wraps its emit in flushBefore, the emit->append->flushBefore re-acquires
// that Mutex and self-deadlocks. This store mirrors production locking so the deadlock is real;
// the fake above cannot catch it because its flushBefore takes no lock.
private fun reentrantHostileArtifactStore(): ArtifactStore = object : ArtifactStore {
val mutex = Mutex()
override suspend fun put(bytes: ByteArray) = fixedArtifactId
override suspend fun get(id: TypeId): ByteArray? = null
override suspend fun flushBefore(commit: suspend () -> Unit) = mutex.withLock { commit() }
}
@Test
fun `does not deadlock when emit re-enters the artifact store (as append does)`(): Unit = runBlocking {
val store = reentrantHostileArtifactStore()
val svc = JournalCompactionService(store, { "summary" }, tokenThreshold = { 100 })
val state = stateWithRecords(makeRecord(1, DecisionKind.INTENT))
// emit simulates SqliteEventStore.append: it flushes artifacts inside the store mutex.
withTimeout(2000) {
svc.compactIfNeeded(SessionId("s1"), state, renderedTokenEstimate = 500) {
store.flushBefore { /* persist event row */ }
}
}
}
@Test @Test
fun `only HIGH-salience records are included in the summarize prompt`(): Unit = runBlocking { fun `only HIGH-salience records are included in the summarize prompt`(): Unit = runBlocking {
val capturedPrompts = mutableListOf<String>() val capturedPrompts = mutableListOf<String>()