feat: экспоненциальный backoff (providers[].backoff / upstreams[].backoff)
Повторяющиеся сбои апстрима (5xx/429/402 и сетевые ошибки) увязываются экспоненциальным откатом: первая ошибка — отдых 1s, далее 2s, 4s, … до потолка из конфига (ISO-8601: P1M, PT15M…); успешный запрос сбрасывает счётчик. Приоритет: upstreams[].backoff (свой счётчик на модель) над providers[].backoff (общий счётчик на все модели провайдера). - Backoff.kt: BackoffGuard (Mutex/@Volatile, kotlinx-datetime) + BackoffRegistry (скоупы провайдер/модель); - pickFreeUpstream пропускает апстримы в откате (лог «~Ns — пропускаю»); - все апстримы модели в откате → 503 «all upstreams cooling, retry in Ns» + заголовок Retry-After; - некорректный ISO-8601 в backoff — warn + игнор поля; - BackoffTest (8 тестов) + обновлённые тесты pickFreeUpstream; - CONFIG.md/README/TESTING.md. Проверено: jvmTest, compileKotlinLinuxX64, linuxX64Test.
This commit is contained in:
@@ -0,0 +1,111 @@
|
||||
package pw.binom.llmproxy
|
||||
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.datetime.Clock
|
||||
import kotlinx.datetime.Instant
|
||||
import kotlin.concurrent.Volatile
|
||||
import kotlin.time.Duration
|
||||
import kotlin.time.Duration.Companion.seconds
|
||||
|
||||
/**
|
||||
* Экспоненциальный бэкофф для одной области отдыха: первая ошибка — отдых
|
||||
* [base] (1 с), каждая следующая удваивает интервал вплоть до [cap].
|
||||
* Успешный запрос сбрасывает счётчик и отдых.
|
||||
*/
|
||||
class BackoffGuard(val cap: Duration, val base: Duration = 1.seconds) {
|
||||
private val lock = Mutex()
|
||||
@Volatile
|
||||
private var failures: Int = 0
|
||||
@Volatile
|
||||
private var interval: Duration = Duration.ZERO
|
||||
@Volatile
|
||||
private var coolingUntil: Instant = Instant.DISTANT_PAST
|
||||
|
||||
fun isCoolingAt(now: Instant): Boolean = now < coolingUntil
|
||||
|
||||
/** Сколько осталось до выхода из отдыха (0 — не в откате). */
|
||||
fun remainingAt(now: Instant): Duration =
|
||||
if (coolingUntil > now) coolingUntil - now else Duration.ZERO
|
||||
|
||||
/** Учёт ошибки; возвращает интервал, на который область уходит в отдых. */
|
||||
suspend fun recordFailure(now: Instant): Duration {
|
||||
lock.lock()
|
||||
try {
|
||||
failures++
|
||||
interval = if (failures == 1) base else interval * 2
|
||||
if (interval > cap) interval = cap
|
||||
coolingUntil = now + interval
|
||||
} finally {
|
||||
lock.unlock()
|
||||
}
|
||||
return interval
|
||||
}
|
||||
|
||||
/** Успех: сброс счётчика и отдыха. */
|
||||
suspend fun recordSuccess() {
|
||||
lock.lock()
|
||||
try {
|
||||
failures = 0
|
||||
interval = Duration.ZERO
|
||||
coolingUntil = Instant.DISTANT_PAST
|
||||
} finally {
|
||||
lock.unlock()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Реестр областей бэкофф-отдыха. Область апстрима:
|
||||
* - у модели (записи upstreams) задан `backoff` — свой счётчик и потолок (приоритет);
|
||||
* - иначе у провайдера задан `backoff` — общий для всех моделей провайдера счётчик;
|
||||
* - ни там, ни там — бэкофф для этого апстрима выключен.
|
||||
*/
|
||||
class BackoffRegistry(
|
||||
private val providers: Map<String, ProviderConf>,
|
||||
private val upstreams: Map<String, UpstreamConf>,
|
||||
) {
|
||||
private val upstreamGuards = mutableMapOf<String, BackoffGuard>()
|
||||
private val providerGuards = mutableMapOf<String, BackoffGuard>()
|
||||
private val lock = Mutex()
|
||||
|
||||
private fun guardForLocked(up: UpstreamConf): BackoffGuard? {
|
||||
up.backoff?.let { cap ->
|
||||
return upstreamGuards.getOrPut(up.id) { BackoffGuard(cap) }
|
||||
}
|
||||
providers[up.provider]?.backoff?.let { cap ->
|
||||
return providerGuards.getOrPut(up.provider) { BackoffGuard(cap) }
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
private suspend fun <T> locked(block: () -> T?): T? {
|
||||
lock.lock()
|
||||
return try {
|
||||
block()
|
||||
} finally {
|
||||
lock.unlock()
|
||||
}
|
||||
}
|
||||
|
||||
suspend fun isCooling(up: UpstreamConf): Boolean {
|
||||
val g = locked { guardForLocked(up) } ?: return false
|
||||
return g.isCoolingAt(Clock.System.now())
|
||||
}
|
||||
|
||||
/** Секунд до выхода апстрима из отдыха (0 — не в откате). */
|
||||
suspend fun remainingSeconds(up: UpstreamConf): Long {
|
||||
val g = locked { guardForLocked(up) } ?: return 0L
|
||||
return g.remainingAt(Clock.System.now()).inWholeSeconds
|
||||
}
|
||||
|
||||
/** Учёт ошибки; возвращает интервал отдыха, или null, если бэкофф для апстрима выключен. */
|
||||
suspend fun recordFailure(up: UpstreamConf): Duration? {
|
||||
val g = locked { guardForLocked(up) } ?: return null
|
||||
return g.recordFailure(Clock.System.now())
|
||||
}
|
||||
|
||||
/** Успех: сброс области апстрима. */
|
||||
suspend fun recordSuccess(up: UpstreamConf) {
|
||||
locked { guardForLocked(up) }?.recordSuccess()
|
||||
}
|
||||
}
|
||||
@@ -46,6 +46,7 @@ import net.mamoe.yamlkt.YamlList
|
||||
import net.mamoe.yamlkt.YamlLiteral
|
||||
import net.mamoe.yamlkt.YamlMap
|
||||
import kotlin.concurrent.Volatile
|
||||
import kotlin.time.Duration
|
||||
import kotlin.time.Duration.Companion.seconds
|
||||
import kotlin.time.TimeSource
|
||||
|
||||
@@ -86,11 +87,18 @@ fun main() {
|
||||
val active = config.upstreams.associate { up ->
|
||||
up.id to UpstreamCounter(effectiveConcurrencyLimit(up, providersById[up.provider]))
|
||||
}
|
||||
val backoff = BackoffRegistry(providersById, upstreamsById)
|
||||
|
||||
log.info {
|
||||
"[llm-proxy] загружено: providers=${config.providers.size}, " +
|
||||
"upstreams=${config.upstreams.size}, models=${config.models.size} (config=$path)"
|
||||
}
|
||||
log.info {
|
||||
"[llm-proxy] backoff: upstreams=" +
|
||||
config.upstreams.filter { it.backoff != null }.map { up -> "${up.id}=${up.backoff?.toIsoString()}" }.joinToString(", ") +
|
||||
" providers=" +
|
||||
config.providers.filter { it.backoff != null }.map { p -> "${p.id}=${p.backoff?.toIsoString()}" }.joinToString(", ")
|
||||
}
|
||||
log.info {
|
||||
"[llm-proxy] server: bind host=${config.server.host} port=${config.server.port}"
|
||||
}
|
||||
@@ -101,7 +109,7 @@ fun main() {
|
||||
val http = createHttpClient()
|
||||
val sessions = SessionRegistry()
|
||||
startServer(config.server.host, config.server.port) {
|
||||
proxyModule(config, providersById, upstreamsById, active, sessions, http)
|
||||
proxyModule(config, providersById, upstreamsById, active, sessions, http, backoff)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -114,10 +122,11 @@ fun Application.proxyModule(
|
||||
active: Map<String, UpstreamCounter>,
|
||||
sessions: SessionRegistry,
|
||||
http: HttpClient,
|
||||
backoff: BackoffRegistry,
|
||||
) {
|
||||
routing {
|
||||
post("/v1/chat/completions") {
|
||||
handleChat(call, config, providersById, upstreamsById, active, sessions, http)
|
||||
handleChat(call, config, providersById, upstreamsById, active, sessions, http, backoff)
|
||||
}
|
||||
get("/v1/models") {
|
||||
handleModels(call, config)
|
||||
@@ -133,6 +142,7 @@ private suspend fun handleChat(
|
||||
active: Map<String, UpstreamCounter>,
|
||||
sessions: SessionRegistry,
|
||||
http: HttpClient,
|
||||
backoff: BackoffRegistry,
|
||||
) {
|
||||
log.info { "[llm-proxy] chat ${call.request.httpMethod.value} ${call.request.uri} headers: ${formatHeadersForLog(call.request.headers)}" }
|
||||
val raw = call.receiveText()
|
||||
@@ -165,7 +175,7 @@ private suspend fun handleChat(
|
||||
var anyClaimed = false
|
||||
|
||||
while (true) {
|
||||
val up = pickFreeUpstream(pool, active, failed) ?: break
|
||||
val up = pickFreeUpstream(pool, active, failed, backoff, modelName) ?: break
|
||||
anyClaimed = true
|
||||
val start = TimeSource.Monotonic.markNow()
|
||||
try {
|
||||
@@ -230,7 +240,11 @@ private suspend fun handleChat(
|
||||
}.execute { resp ->
|
||||
upstreamStatus = resp.status.value
|
||||
if (upstreamStatus >= 500 || upstreamStatus == 429 || upstreamStatus == 402) {
|
||||
log.warn { "[llm-proxy] model=$modelName upstream=${up.id} вернул $upstreamStatus → фейловер" }
|
||||
val rest = backoff.recordFailure(up)
|
||||
log.warn {
|
||||
"[llm-proxy] model=$modelName upstream=${up.id} вернул $upstreamStatus → фейловер" +
|
||||
(rest?.let { " (backoff: отдых ${it.toIsoString()})" } ?: "")
|
||||
}
|
||||
failed.add(up.id)
|
||||
failover = true
|
||||
return@execute
|
||||
@@ -245,6 +259,7 @@ private suspend fun handleChat(
|
||||
return@execute
|
||||
}
|
||||
responded = true
|
||||
backoff.recordSuccess(up)
|
||||
if (clientWantsStream) {
|
||||
val ct = resp.headers["Content-Type"] ?: "text/event-stream"
|
||||
val status = HttpStatusCode.fromValue(upstreamStatus)
|
||||
@@ -294,7 +309,11 @@ private suspend fun handleChat(
|
||||
log.info { "[llm-proxy] chat model=$modelName upstream=${up.id} ОТМЕНЕНО клиентом за ${start.elapsedNow().inWholeMilliseconds}ms" }
|
||||
throw e
|
||||
} catch (e: Exception) {
|
||||
log.error { "[llm-proxy] chat model=$modelName upstream=${up.id} ОШИБКА: ${e::class.simpleName}: ${e.message} за ${start.elapsedNow().inWholeMilliseconds}ms → фейловер" }
|
||||
val rest = backoff.recordFailure(up)
|
||||
log.error {
|
||||
"[llm-proxy] chat model=$modelName upstream=${up.id} ОШИБКА: ${e::class.simpleName}: ${e.message} за ${start.elapsedNow().inWholeMilliseconds}ms → фейловер" +
|
||||
(rest?.let { " (backoff: отдых ${it.toIsoString()})" } ?: "")
|
||||
}
|
||||
failed.add(up.id)
|
||||
continue
|
||||
} finally {
|
||||
@@ -305,7 +324,17 @@ private suspend fun handleChat(
|
||||
if (anyClaimed) {
|
||||
call.respondText(errorJson("all upstreams failed"), ContentType.Application.Json, HttpStatusCode.ServiceUnavailable)
|
||||
} else {
|
||||
call.respondText(errorJson("all upstreams busy"), ContentType.Application.Json, HttpStatusCode.ServiceUnavailable)
|
||||
val retryIn = pool.map { backoff.remainingSeconds(it) }.filter { it > 0 }.minOrNull()
|
||||
if (retryIn != null) {
|
||||
call.response.headers.append("Retry-After", retryIn.toString())
|
||||
call.respondText(
|
||||
errorJson("all upstreams cooling, retry in ${retryIn}s"),
|
||||
ContentType.Application.Json,
|
||||
HttpStatusCode.ServiceUnavailable,
|
||||
)
|
||||
} else {
|
||||
call.respondText(errorJson("all upstreams busy"), ContentType.Application.Json, HttpStatusCode.ServiceUnavailable)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -489,15 +518,28 @@ class UpstreamCounter(private val limit: Int, initial: Int = 0) {
|
||||
|
||||
/**
|
||||
* Выбор апстрима для попытки: первый по порядку (приоритету) апстрим из `pool`,
|
||||
* у которого свободен слот и который ещё не в `excluded` (не упал ранее).
|
||||
* Сразу занимает слот (через [tryClaim]). Если свободных нет — возвращает null.
|
||||
* у которого свободен слот, который ещё не в `excluded` (не упал ранее) и
|
||||
* не в backoff-откате. Сразу занимает слот (через [tryClaim]).
|
||||
* Если свободных нет — возвращает null.
|
||||
*/
|
||||
internal fun pickFreeUpstream(
|
||||
internal suspend fun pickFreeUpstream(
|
||||
pool: List<UpstreamConf>,
|
||||
active: Map<String, UpstreamCounter>,
|
||||
excluded: Set<String>,
|
||||
): UpstreamConf? =
|
||||
pool.firstOrNull { up -> up.id !in excluded && tryClaim(up, active) }
|
||||
backoff: BackoffRegistry,
|
||||
modelName: String = "?",
|
||||
): UpstreamConf? {
|
||||
for (up in pool) {
|
||||
if (up.id in excluded) continue
|
||||
val rest = backoff.remainingSeconds(up)
|
||||
if (rest > 0) {
|
||||
log.info { "[llm-proxy] model=$modelName upstream=${up.id} в backoff-откате (~${rest}c) — пропускаю" }
|
||||
continue
|
||||
}
|
||||
if (tryClaim(up, active)) return up
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
private suspend fun handleModels(call: ApplicationCall, config: Config) {
|
||||
log.info { "[llm-proxy] models ${call.request.httpMethod.value} ${call.request.uri} headers: ${formatHeadersForLog(call.request.headers)}" }
|
||||
@@ -903,6 +945,7 @@ data class ProviderConf(
|
||||
val think_tags: String? = null,
|
||||
val reasoning_field: String? = null,
|
||||
val reasoning_empty_ok: Boolean = false,
|
||||
val backoff: Duration? = null,
|
||||
)
|
||||
|
||||
data class UpstreamConf(
|
||||
@@ -912,6 +955,7 @@ data class UpstreamConf(
|
||||
val max_concurrency: Int? = null,
|
||||
val patch: JsonObject? = null,
|
||||
val think_tags: String? = null,
|
||||
val backoff: Duration? = null,
|
||||
)
|
||||
|
||||
data class ModelConf(
|
||||
@@ -969,6 +1013,7 @@ internal fun parseConfig(root: YamlElement): Config {
|
||||
think_tags = m.strOrNull("think_tags"),
|
||||
reasoning_field = m.strOrNull("reasoning_field"),
|
||||
reasoning_empty_ok = m.strOrNull("reasoning_empty_ok")?.toBooleanStrictOrNull() ?: false,
|
||||
backoff = m.durationOrNull("backoff"),
|
||||
)
|
||||
}
|
||||
|
||||
@@ -981,6 +1026,7 @@ internal fun parseConfig(root: YamlElement): Config {
|
||||
max_concurrency = m.strOrNull("max_concurrency")?.toIntOrNull(),
|
||||
patch = m.yamlMapOrNull("patch")?.let { yamlToJson(it) as JsonObject },
|
||||
think_tags = m.strOrNull("think_tags"),
|
||||
backoff = m.durationOrNull("backoff"),
|
||||
)
|
||||
}
|
||||
|
||||
@@ -1006,5 +1052,14 @@ internal fun Map<String, YamlElement>.str(key: String): String =
|
||||
internal fun Map<String, YamlElement>.strOrNull(key: String): String? =
|
||||
(this[key] as? YamlLiteral)?.content
|
||||
|
||||
/** ISO-8601-длительность (PT30M, P1M, PT1H15M…); некорректное значение — warn + null. */
|
||||
internal fun Map<String, YamlElement>.durationOrNull(key: String): Duration? =
|
||||
strOrNull(key)?.let { raw ->
|
||||
runCatching { Duration.parse(raw) }.getOrElse {
|
||||
log.warn { "[llm-proxy] config: поле '$key' — некорректная ISO-8601-длительность '$raw', проигнорировано" }
|
||||
null
|
||||
}
|
||||
}
|
||||
|
||||
internal fun Map<String, YamlElement>.yamlMapOrNull(key: String): YamlMap? =
|
||||
this[key] as? YamlMap
|
||||
|
||||
Reference in New Issue
Block a user