From 363d880a69ca8837e6a6eba32c38a41b9f50bd84 Mon Sep 17 00:00:00 2001 From: kami Date: Sun, 12 Jul 2026 12:52:07 +0400 Subject: [PATCH] fix(l3): persistent sidecar reader, kill-on-desync, wire persistPath save/load --- .../router/turbovec/TurboVecL3MemoryStore.kt | 78 ++++++++++++------- 1 file changed, 52 insertions(+), 26 deletions(-) diff --git a/infrastructure/router/turbovec/src/main/kotlin/com/correx/infrastructure/router/turbovec/TurboVecL3MemoryStore.kt b/infrastructure/router/turbovec/src/main/kotlin/com/correx/infrastructure/router/turbovec/TurboVecL3MemoryStore.kt index acf3d7d8..ab986077 100644 --- a/infrastructure/router/turbovec/src/main/kotlin/com/correx/infrastructure/router/turbovec/TurboVecL3MemoryStore.kt +++ b/infrastructure/router/turbovec/src/main/kotlin/com/correx/infrastructure/router/turbovec/TurboVecL3MemoryStore.kt @@ -8,8 +8,10 @@ import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock import kotlinx.coroutines.withTimeout import kotlinx.serialization.json.Json +import java.io.BufferedReader import java.nio.file.Paths import kotlin.io.path.createDirectories +import kotlin.io.path.exists /** * Adapter to TurboVec sidecar process for cross-session vector-based memory. @@ -24,6 +26,9 @@ class TurboVecL3MemoryStore(private val config: TurboVecSidecarConfig) : Rehydra private val json = Json { ignoreUnknownKeys = true } private val mutex = Mutex() private var process: Process? = null + // One reader owned by the store for the process's lifetime. A fresh bufferedReader() per request + // would strand read-ahead bytes in the discarded reader's buffer, desyncing the stream. + private var reader: BufferedReader? = null private val metadata = java.util.concurrent.ConcurrentHashMap() override suspend fun rehydrateMetadata(entries: List) { @@ -72,10 +77,12 @@ class TurboVecL3MemoryStore(private val config: TurboVecSidecarConfig) : Rehydra override suspend fun close() { mutex.withLock { process?.let { p -> - val shutdownRequest = SidecarRequest(op = "shutdown") - runCatching { sendRequest(shutdownRequest) } + // Persist vectors before exit so the next process can `load` them (only save() puts + // the quantized index on disk — init alone loses everything between runs). + config.persistPath?.let { runCatching { sendRequestLocked(SidecarRequest(op = "save", path = it.toString())) } } + runCatching { sendRequestLocked(SidecarRequest(op = "shutdown")) } try { - withTimeout(2000) { + withTimeout(SHUTDOWN_TIMEOUT_MS) { p.waitFor() } } catch (e: Exception) { @@ -85,27 +92,41 @@ class TurboVecL3MemoryStore(private val config: TurboVecSidecarConfig) : Rehydra throw e } } + reader = null + process = null } } } - private suspend fun sendRequest(request: SidecarRequest): SidecarResponse { - mutex.withLock { - ensureProcessStarted() - val process = process ?: throw RuntimeException("Process failed to start") + private suspend fun sendRequest(request: SidecarRequest): SidecarResponse = + mutex.withLock { sendRequestLocked(request) } - val requestLine = json.encodeToString(SidecarRequest.serializer(), request) + "\n" - process.outputStream.write(requestLine.toByteArray()) - process.outputStream.flush() + /** Caller must hold [mutex]. */ + private fun sendRequestLocked(request: SidecarRequest): SidecarResponse { + ensureProcessStarted() + val process = process ?: throw RuntimeException("Process failed to start") + val reader = reader ?: throw RuntimeException("Sidecar reader not initialized") - val responseLine = runCatching { - withTimeout(config.requestTimeoutMs) { - process.inputStream.bufferedReader().readLine() - } - }.getOrNull() ?: throw RuntimeException("No response from sidecar") + val requestLine = json.encodeToString(SidecarRequest.serializer(), request) + "\n" + process.outputStream.write(requestLine.toByteArray()) + process.outputStream.flush() - return json.decodeFromString(SidecarResponse.serializer(), responseLine) + // A blocking readLine can't be cancelled by a coroutine timeout, so a slow/late response + // would land in the NEXT request's read and desync the stream permanently. We don't wrap it + // in withTimeout at all: instead the null/failure path below tears the process down so the + // next request restarts from a clean, empty stream rather than reading a stale line. + val responseLine = runCatching { reader.readLine() }.getOrNull() + if (responseLine == null) { + killProcess() + throw RuntimeException("No response from sidecar (process reset)") } + return json.decodeFromString(SidecarResponse.serializer(), responseLine) + } + + private fun killProcess() { + runCatching { process?.destroyForcibly() } + reader = null + process = null } private fun ensureProcessStarted() { @@ -122,19 +143,24 @@ class TurboVecL3MemoryStore(private val config: TurboVecSidecarConfig) : Rehydra val started = pb.start() process = started + reader = started.inputStream.bufferedReader() - // The sidecar rejects add/search until the index is created. Send init once on startup; - // without this every store/query fails with "Index not initialized". - val initLine = json.encodeToString( - SidecarRequest.serializer(), - SidecarRequest(op = "init", dim = config.dim, bitWidth = config.bitWidth) - ) + "\n" - started.outputStream.write(initLine.toByteArray()) + // Reload a persisted index if one exists, else create a fresh one. Either way the sidecar + // rejects add/search until initialized; without this every store/query fails "not initialized". + val bootstrap = config.persistPath + ?.takeIf { it.exists() } + ?.let { SidecarRequest(op = "load", path = it.toString()) } + ?: SidecarRequest(op = "init", dim = config.dim, bitWidth = config.bitWidth) + + started.outputStream.write((json.encodeToString(SidecarRequest.serializer(), bootstrap) + "\n").toByteArray()) started.outputStream.flush() - val response = started.inputStream.bufferedReader().readLine() - ?.let { json.decodeFromString(SidecarResponse.serializer(), it) } + val response = reader!!.readLine()?.let { json.decodeFromString(SidecarResponse.serializer(), it) } if (response?.ok != true) { - throw RuntimeException("TurboVec init failed: ${response?.error ?: "no response"}") + throw RuntimeException("TurboVec ${bootstrap.op} failed: ${response?.error ?: "no response"}") } } + + private companion object { + const val SHUTDOWN_TIMEOUT_MS = 2000L + } }