feat: фоновый пинг апстримов в backoff-откате (probe_interval) — мгновенное возвращение в пул

src/commonMain/kotlin/pw/binom/llmproxy/HealthProber.kt

@@ -0,0 +1,133 @@
+package pw.binom.llmproxy
+
+import io.ktor.client.HttpClient
+import io.ktor.client.call.body
+import io.ktor.client.request.headers
+import io.ktor.client.request.preparePost
+import io.ktor.client.request.setBody
+import io.ktor.utils.io.ByteReadChannel
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.currentCoroutineContext
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.isActive
+import kotlinx.coroutines.launch
+import kotlinx.serialization.json.JsonArray
+import kotlinx.serialization.json.JsonObject
+import kotlinx.serialization.json.JsonPrimitive
+import kotlin.time.Duration
+
+/**
+ * Фоновый «пингер» апстримов, которые в данный момент в backoff-откате.
+ *
+ * Пока апстрим молчит (в откате), HealthProber раз в [interval] шлёт ему
+ * минимальный chat.completion (`2+2=?`, `max_tokens=4`) и слушает первый байт
+ * ответа. Как только апстрим ответил — вызывается [BackoffRegistry.recordSuccess],
+ * и он возвращается в общий пул сразу, не дожидаясь истечения backoff-интервала.
+ * Это ускоряет восстановление после долгого простоя провайдера.
+ *
+ * Пинг НЕ занимает слот конкурентности ([UpstreamCounter]) — он идёт в обход,
+ * чтобы не отжимать слот у живых запросов. Ошибки пинга НЕ считаются в backoff
+ * (иначе каждая попытка удваивала бы интервал отката); считается только успех.
+ */
+class HealthProber(
+    private val providersById: Map<String, ProviderConf>,
+    private val upstreams: List<UpstreamConf>,
+    private val http: HttpClient,
+    private val backoff: BackoffRegistry,
+    private val metrics: MetricsRegistry,
+    private val scope: CoroutineScope,
+) {
+    private val jobs = mutableMapOf<String, Job>()
+
+    /** Запустить пинг-корутины для всех апстримов с включённым пингом. */
+    fun start() {
+        for (up in upstreams) {
+            val interval = effectiveProbeInterval(up, providersById[up.provider])
+            if (interval != null) {
+                jobs[up.id] = scope.launch { probeLoop(up, interval) }
+            }
+        }
+    }
+
+    /** Цикл пинга одного апстрима: пингуем только пока он в откате. */
+    internal suspend fun probeLoop(up: UpstreamConf, interval: Duration, pingImpl: suspend (UpstreamConf) -> Boolean = ::ping) {
+        while (currentCoroutineContext().isActive) {
+            if (backoff.isCooling(up)) {
+                val healthy = pingImpl(up)
+                metrics.recordProbe(up, if (healthy) "ok" else "fail")
+                if (healthy) {
+                    backoff.recordSuccess(up)
+                    log.info { "[llm-proxy] probe upstream=${up.id} ответил — снят с backoff, вернулся в пул" }
+                } else {
+                    log.info { "[llm-proxy] probe upstream=${up.id} молчит — остаюсь в backoff" }
+                }
+            }
+            delay(interval)
+        }
+    }
+
+    /** Одна пинг-попытка: минимальный stream-запрос, слушаем первый байт. */
+    private suspend fun ping(up: UpstreamConf): Boolean {
+        val provider = providersById[up.provider] ?: return false
+        val url = provider.url.trimEnd('/') + "/chat/completions"
+        val providerKey = resolveEnv(provider.key)
+        val body = buildProbeBody(up, provider)
+        val firstByteTimeout = effectiveFirstByteTimeout(up, provider) ?: DEFAULT_FIRST_BYTE_TIMEOUT
+        return try {
+            http.preparePost(url) {
+                headers {
+                    if (providerKey.isNotEmpty()) {
+                        set("Authorization", "Bearer $providerKey")
+                    }
+                    set("Content-Type", "application/json")
+                }
+                setBody(body.toString())
+            }.execute { resp ->
+                if (resp.status.value !in 200..299) {
+                    return@execute false
+                }
+                val ch = resp.body<ByteReadChannel>()
+                val buf = ByteArray(64)
+                readAvailableWithTimeout(ch, buf, firstByteTimeout.inWholeMilliseconds) > 0
+            }
+        } catch (e: CancellationException) {
+            throw e
+        } catch (e: Exception) {
+            log.info { "[llm-proxy] probe upstream=${up.id} ошибка: ${e.message}" }
+            false
+        }
+    }
+}
+
+/**
+ * Собрать тело пинг-запроса: минимальный chat.completion, на который любой
+ * живой апстрим ответит сразу. Применяются патчи провайдера и апстрима (в том
+ * же порядке, что и для реальных запросов), чтобы пинг валидировал реальный
+ * путь запроса, но без клиентских данных и модели.
+ */
+internal fun buildProbeBody(up: UpstreamConf, provider: ProviderConf?): JsonObject {
+    val base = JsonObject(
+        mapOf(
+            "model" to JsonPrimitive(up.model),
+            "messages" to JsonArray(
+                listOf(
+                    JsonObject(
+                        mapOf(
+                            "role" to JsonPrimitive("user"),
+                            "content" to JsonPrimitive("2+2=?"),
+                        ),
+                    ),
+                ),
+            ),
+            "max_tokens" to JsonPrimitive(4),
+            "stream" to JsonPrimitive(true),
+        ),
+    )
+    var acc = base
+    listOf(provider?.patch, up.patch).filterNotNull().forEach { patch ->
+        acc = merge(acc, patch)
+    }
+    return acc
+}

src/commonMain/kotlin/pw/binom/llmproxy/Main.kt

@@ -28,6 +28,9 @@ import io.ktor.utils.io.ByteReadChannel
 import io.ktor.utils.io.ByteWriteChannel
 import io.ktor.utils.io.writeFully
 import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.async
 import kotlinx.coroutines.cancelAndJoin
 import kotlinx.coroutines.coroutineScope
@@ -115,6 +118,13 @@ fun main() {
             config.providers.filter { it.first_byte_timeout != null }.map { p -> "${p.id}=${p.first_byte_timeout?.toIsoString()}" }.joinToString(", ") +
             " default=${DEFAULT_FIRST_BYTE_TIMEOUT.toIsoString()}"
     }
+    log.info {
+        "[llm-proxy] probe_interval: upstreams=" +
+            config.upstreams.filter { it.probe_interval != null }.map { up -> "${up.id}=${up.probe_interval?.toIsoString()}" }.joinToString(", ") +
+            " providers=" +
+            config.providers.filter { it.probe_interval != null }.map { p -> "${p.id}=${p.probe_interval?.toIsoString()}" }.joinToString(", ") +
+            " default=${DEFAULT_PROBE_INTERVAL.toIsoString()}"
+    }
     log.info {
         "[llm-proxy] server: bind host=${config.server.host} port=${config.server.port}"
     }
@@ -124,6 +134,8 @@ fun main() {
 
     val http = createHttpClient()
     val sessions = SessionRegistry()
+    val prober = HealthProber(providersById, config.upstreams, http, backoff, metrics, CoroutineScope(SupervisorJob() + Dispatchers.Default))
+    prober.start()
     startServer(config.server.host, config.server.port) {
         proxyModule(config, providersById, upstreamsById, active, sessions, http, backoff, metrics)
     }
@@ -527,6 +539,17 @@ internal fun effectiveFirstByteTimeout(up: UpstreamConf, provider: ProviderConf?
     return if (raw <= Duration.ZERO) null else raw
 }
 
+/**
+ * Эффективный интервал фонового пинга апстрима в backoff-откате: значение у
+ * апстрима, если задано; иначе у провайдера; иначе [DEFAULT_PROBE_INTERVAL].
+ * Возвращает null, если явно отключено (`Duration.ZERO` — `PT0S`): тогда
+ * HealthProber этот апстрим не пингует (ждём естественного истечения бэкоффа).
+ */
+internal fun effectiveProbeInterval(up: UpstreamConf, provider: ProviderConf?): Duration? {
+    val raw = up.probe_interval ?: provider?.probe_interval ?: DEFAULT_PROBE_INTERVAL
+    return if (raw <= Duration.ZERO) null else raw
+}
+
 /**
  * Сигнализирует, что апстрим не прислал ни одного байта тела стрим-ответа за
  * [timeoutMs] мс. Обрабатывается как обычный сбой (фейловер, backoff,
@@ -1083,6 +1106,7 @@ data class ProviderConf(
     val reasoning_empty_ok: Boolean = false,
     val backoff: Duration? = null,
     val first_byte_timeout: Duration? = null,
+    val probe_interval: Duration? = null,
 )
 
 data class UpstreamConf(
@@ -1094,6 +1118,7 @@ data class UpstreamConf(
     val think_tags: String? = null,
     val backoff: Duration? = null,
     val first_byte_timeout: Duration? = null,
+    val probe_interval: Duration? = null,
 )
 
 data class ModelConf(
@@ -1153,6 +1178,7 @@ internal fun parseConfig(root: YamlElement): Config {
             reasoning_empty_ok = m.strOrNull("reasoning_empty_ok")?.toBooleanStrictOrNull() ?: false,
             backoff = m.durationOrNull("backoff"),
             first_byte_timeout = m.durationOrNull("first_byte_timeout"),
+            probe_interval = m.durationOrNull("probe_interval"),
         )
     }
 
@@ -1167,6 +1193,7 @@ internal fun parseConfig(root: YamlElement): Config {
             think_tags = m.strOrNull("think_tags"),
             backoff = m.durationOrNull("backoff"),
             first_byte_timeout = m.durationOrNull("first_byte_timeout"),
+            probe_interval = m.durationOrNull("probe_interval"),
         )
     }
 

src/commonMain/kotlin/pw/binom/llmproxy/Metrics.kt

@@ -10,6 +10,8 @@ import kotlinx.coroutines.sync.withLock
  * Метрики (всё в памяти, сброс при рестарте):
  * - `llm_proxy_requests_total{model,upstream,provider,result}` — счётчик
  *   chat-запросов; `result`: `ok | 4xx | 429 | 402 | 5xx | net_err | timeout | cancelled`;
+ * - `llm_proxy_probe_pings_total{upstream,provider,result}` — счётчик фоновых
+ *   пингов апстримов в backoff-откате (HealthProber); `result`: `ok | fail`;
  * - `llm_proxy_upstream_inflight{upstream,provider}` — сейчас в работе (слоты);
  * - `llm_proxy_upstream_fail_streak{upstream,provider}` — текущая серия сбоев
  *   (счётчик backoff);
@@ -19,6 +21,7 @@ import kotlinx.coroutines.sync.withLock
 class MetricsRegistry {
     private val lock = Mutex()
     private val requests = HashMap<RequestKey, Long>()
+    private val probes = HashMap<ProbeKey, Long>()
 
     private data class RequestKey(
         val model: String,
@@ -27,6 +30,12 @@ class MetricsRegistry {
         val result: String,
     )
 
+    private data class ProbeKey(
+        val upstream: String,
+        val provider: String,
+        val result: String,
+    )
+
     /** Учёт chat-запроса по исходу (значения `result` — в описании класса). */
     suspend fun record(model: String, up: UpstreamConf, result: String) {
         lock.withLock {
@@ -35,6 +44,14 @@ class MetricsRegistry {
         }
     }
 
+    /** Учёт пинга апстрима в backoff-откате (result: `ok`/`fail`). */
+    suspend fun recordProbe(up: UpstreamConf, result: String) {
+        lock.withLock {
+            val key = ProbeKey(up.id, up.provider, result)
+            probes[key] = (probes[key] ?: 0L) + 1
+        }
+    }
+
     /** Текстовая выгрузка в формате Prometheus на текущий момент. */
     suspend fun render(
         config: Config,
@@ -56,6 +73,19 @@ class MetricsRegistry {
                     )
                 }
         }
+        sb.append("# HELP llm_proxy_probe_pings_total Фоновые пинги апстримов в backoff-откате\n")
+        sb.append("# TYPE llm_proxy_probe_pings_total counter\n")
+        lock.withLock {
+            probes.entries
+                .sortedWith(compareBy({ it.key.upstream }, { it.key.provider }, { it.key.result }))
+                .forEach { (k, n) ->
+                    sb.append(
+                        "llm_proxy_probe_pings_total{upstream=\"" + esc(k.upstream) +
+                            "\",provider=\"" + esc(k.provider) +
+                            "\",result=\"" + esc(k.result) + "\"} $n\n",
+                    )
+                }
+        }
         sb.append("# HELP llm_proxy_upstream_inflight Занято слотов апстрима сейчас\n")
         sb.append("# TYPE llm_proxy_upstream_inflight gauge\n")
         for (up in config.upstreams) {

src/commonMain/kotlin/pw/binom/llmproxy/Timeouts.kt

@@ -15,3 +15,12 @@ const val UPSTREAM_REQUEST_TIMEOUT_MS: Long = 5 * 60_000L
  * Отключается явным `first_byte_timeout: PT0S` на апстриме или провайдере.
  */
 val DEFAULT_FIRST_BYTE_TIMEOUT: Duration = 30.seconds
+
+/**
+ * Default интервал фоновой проверки («пинга») апстрима, который сейчас в
+ * backoff-откате: пока апстрим молчит, HealthProber шлёт ему минимальный
+ * запрос (2+2=?, max_tokens=4) каждые [DEFAULT_PROBE_INTERVAL], и как только
+ * он отвечает — снимает его с отката раньше, чем истечёт интервал бэкоффа.
+ * Отключается явным `probe_interval: PT0S` на апстриме или провайдере.
+ */
+val DEFAULT_PROBE_INTERVAL: Duration = 1.seconds