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:
@@ -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
|
||||||
|
}
|
||||||
|
}
|
||||||
+135
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
+2
-1
@@ -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,
|
||||||
)
|
)
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|||||||
Reference in New Issue
Block a user