feat(server): event-store health probe

Third HealthProbe (after disk + llama): times a cheap lastGlobalSequence()
read as an event-store responsiveness heartbeat — DEGRADED on throw or when
latency exceeds eventStoreLatencyWarnMs (default 500). Read-only by design
(no probe writes into the source-of-truth log; Invariant #1). Plugs into the
existing HealthMonitor; no new events. Observability spec §4.

Also wraps a long line in LlamaServerHealthProbeTest to clear a detekt
MaxLineLength finding introduced in aaf896d.
This commit is contained in:
2026-06-14 18:40:29 +04:00
parent 0957f7c69e
commit 7a1362c973
5 changed files with 214 additions and 1 deletions
@@ -449,6 +449,10 @@ fun main() {
livenessTimeoutMs = hc.llamaLivenessTimeoutMs, livenessTimeoutMs = hc.llamaLivenessTimeoutMs,
tpsWarnBelow = hc.llamaTpsWarnBelow, tpsWarnBelow = hc.llamaTpsWarnBelow,
), ),
com.correx.apps.server.health.EventStoreHealthProbe(
eventStore = eventStore,
latencyWarnMs = hc.eventStoreLatencyWarnMs,
),
), ),
intervalMs = hc.intervalMs, intervalMs = hc.intervalMs,
initialStatuses = seeded, initialStatuses = seeded,
@@ -0,0 +1,68 @@
package com.correx.apps.server.health
import com.correx.core.events.events.HealthSubject
import com.correx.core.events.stores.EventStore
/**
* Health probe for the event store (observability-spec §4: EVENT_STORE subject).
*
* **Responsiveness** — every tick times a cheap [EventStore.lastGlobalSequence] read. The
* suspend store call dispatches its own I/O, so the probe times just the read (no extra
* dispatcher hop to inflate the metric). If the call throws (store unavailable) or takes
* longer than [latencyWarnMs], the observation is DEGRADED; otherwise HEALTHY with the
* measured latency as [HealthObservation.observedValue].
*
* **Why a read, not a write**: The event log is the single source of truth (Invariant #1).
* Writing probe events on every health tick would pollute the journal and grow it
* unbounded. A read gives an equally faithful heartbeat — if the store can't answer a
* sequence query it can't be trusted to record facts. [HealthMonitor] owns edge detection
* and records the observation as a [HealthDegradedEvent] / [HealthRestoredEvent]; this
* probe only observes and returns (Invariants #8/#9).
*
* [nanoTimeSource] is injectable for deterministic testing (fake clocks, no real sleeps).
*/
class EventStoreHealthProbe(
private val eventStore: EventStore,
private val latencyWarnMs: Long = 500,
private val nanoTimeSource: () -> Long = System::nanoTime,
) : HealthProbe {
override val subject: HealthSubject = HealthSubject.EVENT_STORE
override suspend fun observe(): HealthObservation {
val startNs = nanoTimeSource()
val result = runCatching { eventStore.lastGlobalSequence() }
val elapsedMs = (nanoTimeSource() - startNs) / NS_PER_MS
return when {
result.isFailure -> degraded(
metric = "read_latency_ms",
value = elapsedMs,
threshold = latencyWarnMs,
detail = "lastGlobalSequence() threw: ${result.exceptionOrNull()?.message}",
)
elapsedMs >= latencyWarnMs -> degraded(
metric = "read_latency_ms",
value = elapsedMs,
threshold = latencyWarnMs,
detail = "lastGlobalSequence() took ${elapsedMs}ms (warn ≥ ${latencyWarnMs}ms)",
)
else -> healthy(
metric = "read_latency_ms",
value = elapsedMs,
threshold = latencyWarnMs,
detail = "event store responsive: lastGlobalSequence() in ${elapsedMs}ms",
)
}
}
private fun degraded(metric: String, value: Long, threshold: Long, detail: String) =
HealthObservation(subject, HealthStatus.DEGRADED, metric, value, threshold, detail)
private fun healthy(metric: String, value: Long, threshold: Long, detail: String) =
HealthObservation(subject, HealthStatus.HEALTHY, metric, value, threshold, detail)
companion object {
private const val NS_PER_MS = 1_000_000L
}
}
@@ -0,0 +1,135 @@
package com.correx.apps.server.health
import com.correx.core.events.events.NewEvent
import com.correx.core.events.events.StoredEvent
import com.correx.core.events.stores.EventStore
import com.correx.core.events.types.SessionId
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.emptyFlow
import kotlinx.coroutines.runBlocking
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Assertions.assertTrue
import org.junit.jupiter.api.Test
class EventStoreHealthProbeTest {
// ---------------------------------------------------------------------------
// Fake store helpers
// ---------------------------------------------------------------------------
/** Minimal EventStore that returns a fixed global sequence or throws on demand. */
private class FakeEventStore(
private val globalSequence: Long = 0L,
private val throwOnRead: Boolean = false,
) : EventStore {
override suspend fun append(event: NewEvent): StoredEvent = error("unused")
override suspend fun appendAll(events: List<NewEvent>): List<StoredEvent> = error("unused")
override fun read(sessionId: SessionId): List<StoredEvent> = emptyList()
override fun readFrom(sessionId: SessionId, fromSequence: Long): List<StoredEvent> = emptyList()
override fun lastSequence(sessionId: SessionId): Long? = null
override fun subscribe(sessionId: SessionId): Flow<StoredEvent> = emptyFlow()
override fun subscribeAll(): Flow<StoredEvent> = emptyFlow()
override suspend fun lastGlobalSequence(): Long {
if (throwOnRead) error("store unavailable")
return globalSequence
}
override fun allEvents(): Sequence<StoredEvent> = emptySequence()
override fun allSessionIds(): Set<SessionId> = emptySet()
}
/**
* Fake nano-time source that reports [elapsedMs] milliseconds between the first and
* second call — deterministically, with no real sleep. The first call (start) returns 0;
* the second call (after the read) returns [elapsedMs] * NS_PER_MS.
*/
private fun fakeClockMs(elapsedMs: Long): () -> Long {
val ns = elapsedMs * 1_000_000L
var call = 0
return { if (call++ == 0) 0L else ns }
}
private fun probe(
store: EventStore = FakeEventStore(),
latencyWarnMs: Long = 500,
nanoTimeSource: () -> Long = fakeClockMs(0L),
) = EventStoreHealthProbe(
eventStore = store,
latencyWarnMs = latencyWarnMs,
nanoTimeSource = nanoTimeSource,
)
// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------
@Test
fun `healthy when read completes within threshold`(): Unit = runBlocking {
val obs = probe(
store = FakeEventStore(globalSequence = 42L),
latencyWarnMs = 500,
nanoTimeSource = fakeClockMs(elapsedMs = 10),
).observe()
assertEquals(HealthStatus.HEALTHY, obs.status)
assertEquals("read_latency_ms", obs.metric)
assertEquals(10L, obs.observedValue)
assertEquals(500L, obs.threshold)
assertTrue(obs.detail.contains("responsive"), "detail should mention 'responsive': ${obs.detail}")
}
@Test
fun `degraded when read throws`(): Unit = runBlocking {
val obs = probe(
store = FakeEventStore(throwOnRead = true),
latencyWarnMs = 500,
nanoTimeSource = fakeClockMs(elapsedMs = 1),
).observe()
assertEquals(HealthStatus.DEGRADED, obs.status)
assertEquals("read_latency_ms", obs.metric)
assertTrue(obs.detail.contains("threw"), "detail should mention 'threw': ${obs.detail}")
assertTrue(obs.detail.contains("store unavailable"), "detail should contain error message: ${obs.detail}")
}
@Test
fun `degraded when read exceeds latency threshold`(): Unit = runBlocking {
val obs = probe(
store = FakeEventStore(globalSequence = 0L),
latencyWarnMs = 500,
nanoTimeSource = fakeClockMs(elapsedMs = 600),
).observe()
assertEquals(HealthStatus.DEGRADED, obs.status)
assertEquals("read_latency_ms", obs.metric)
assertEquals(600L, obs.observedValue)
assertEquals(500L, obs.threshold)
assertTrue(obs.detail.contains("600ms"), "detail should contain elapsed ms: ${obs.detail}")
assertTrue(obs.detail.contains("500ms"), "detail should contain threshold: ${obs.detail}")
}
@Test
fun `healthy when read latency equals threshold minus one`(): Unit = runBlocking {
// Boundary: exactly one ms under the threshold is still HEALTHY
val obs = probe(
store = FakeEventStore(globalSequence = 1L),
latencyWarnMs = 500,
nanoTimeSource = fakeClockMs(elapsedMs = 499),
).observe()
assertEquals(HealthStatus.HEALTHY, obs.status)
assertEquals(499L, obs.observedValue)
}
@Test
fun `degraded when read latency equals threshold exactly`(): Unit = runBlocking {
// Boundary: exactly at the threshold trips DEGRADED (>= check)
val obs = probe(
store = FakeEventStore(globalSequence = 1L),
latencyWarnMs = 500,
nanoTimeSource = fakeClockMs(elapsedMs = 500),
).observe()
assertEquals(HealthStatus.DEGRADED, obs.status)
assertEquals(500L, obs.observedValue)
}
}
@@ -63,7 +63,8 @@ class LlamaServerHealthProbeTest {
private class ListEventStore(private val events: MutableList<StoredEvent> = mutableListOf()) : EventStore { private class ListEventStore(private val events: MutableList<StoredEvent> = mutableListOf()) : EventStore {
override suspend fun append(event: NewEvent): StoredEvent = error("unused") override suspend fun append(event: NewEvent): StoredEvent = error("unused")
override suspend fun appendAll(events: List<NewEvent>): List<StoredEvent> = error("unused") override suspend fun appendAll(events: List<NewEvent>): List<StoredEvent> = error("unused")
override fun read(sessionId: SessionId): List<StoredEvent> = events.filter { it.metadata.sessionId == sessionId } override fun read(sessionId: SessionId): List<StoredEvent> =
events.filter { it.metadata.sessionId == sessionId }
override fun readFrom(sessionId: SessionId, fromSequence: Long): List<StoredEvent> = override fun readFrom(sessionId: SessionId, fromSequence: Long): List<StoredEvent> =
events.filter { it.metadata.sessionId == sessionId && it.sequence >= fromSequence } events.filter { it.metadata.sessionId == sessionId && it.sequence >= fromSequence }
override fun lastSequence(sessionId: SessionId): Long? = override fun lastSequence(sessionId: SessionId): Long? =
@@ -29,6 +29,10 @@ data class CorrexConfig(
* timeout within this window is reported as DEGRADED (liveness). [llamaTpsWarnBelow] is the * timeout within this window is reported as DEGRADED (liveness). [llamaTpsWarnBelow] is the
* tokens/sec floor below which the probe reports degradation (KV-cache pressure, thermal throttle); * tokens/sec floor below which the probe reports degradation (KV-cache pressure, thermal throttle);
* 0.0 disables the degradation dimension (no baseline to compare against on a fresh install). * 0.0 disables the degradation dimension (no baseline to compare against on a fresh install).
*
* [eventStoreLatencyWarnMs] is the read-latency threshold above which the event-store probe reports
* DEGRADED. A single cheap read (lastGlobalSequence) is timed each tick; a slow or throwing read
* signals contention or unavailability. 500 ms suits typical SQLite on a homelab SSD.
*/ */
@Serializable @Serializable
data class HealthConfig( data class HealthConfig(
@@ -38,6 +42,7 @@ data class HealthConfig(
val diskWarnGrowthBytesPerMin: Long = 50_000_000, val diskWarnGrowthBytesPerMin: Long = 50_000_000,
val llamaLivenessTimeoutMs: Long = 5_000, val llamaLivenessTimeoutMs: Long = 5_000,
val llamaTpsWarnBelow: Double = 0.0, val llamaTpsWarnBelow: Double = 0.0,
val eventStoreLatencyWarnMs: Long = 500,
) )
/** /**