feat(server): add grouped provider fallback routing (#2170)
This commit is contained in:
@@ -151,6 +151,174 @@ describe('configKVService', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it('llm router config should preserve explicit LLM and TTS provider groups', async () => {
|
||||
await service.set('LLM_ROUTER_CONFIG', {
|
||||
llm: {
|
||||
models: {
|
||||
'step-3.5-flash': {
|
||||
upstreams: [
|
||||
{
|
||||
id: 'plan',
|
||||
baseURL: 'https://api.stepfun.com/step_plan/v1',
|
||||
keys: [{ id: 'plan-key', ciphertext: 'plan-ciphertext' }],
|
||||
headerTemplate: 'Bearer {KEY}',
|
||||
},
|
||||
{
|
||||
id: 'paygo',
|
||||
baseURL: 'https://api.stepfun.com/v1',
|
||||
keys: [{ id: 'paygo-key', ciphertext: 'paygo-ciphertext' }],
|
||||
headerTemplate: 'Bearer {KEY}',
|
||||
},
|
||||
],
|
||||
routing: {
|
||||
groups: [
|
||||
{
|
||||
id: 'plan',
|
||||
upstreamIds: ['plan'],
|
||||
retryOn: { httpCodes: [402, 429, 500, 502, 503, 504], onTimeout: true },
|
||||
continueOn: { httpCodes: [402], onTimeout: false },
|
||||
},
|
||||
{
|
||||
id: 'paygo',
|
||||
upstreamIds: ['paygo'],
|
||||
retryOn: { httpCodes: [429, 500, 502, 503, 504], onTimeout: true },
|
||||
},
|
||||
],
|
||||
},
|
||||
fallbackTriggers: {
|
||||
httpCodes: [401, 402, 403, 429, 500, 502, 503, 504],
|
||||
onTimeout: true,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
tts: {
|
||||
models: {
|
||||
'stepfun/stepaudio-2.5-tts': {
|
||||
provider: 'stepfun',
|
||||
upstreams: [
|
||||
{
|
||||
id: 'plan',
|
||||
baseURL: 'https://api.stepfun.com',
|
||||
keys: [{ id: 'plan-key', ciphertext: 'plan-ciphertext' }],
|
||||
adapterParams: { endpointProfile: 'step-plan' },
|
||||
maxConcurrency: 1,
|
||||
},
|
||||
{
|
||||
id: 'paygo',
|
||||
baseURL: 'https://api.stepfun.com',
|
||||
keys: [{ id: 'paygo-key', ciphertext: 'paygo-ciphertext' }],
|
||||
adapterParams: { endpointProfile: 'default' },
|
||||
},
|
||||
],
|
||||
routing: {
|
||||
groups: [
|
||||
{
|
||||
id: 'plan',
|
||||
upstreamIds: ['plan'],
|
||||
strategy: 'least-inflight',
|
||||
retryOn: { httpCodes: [402, 429, 500, 502, 503, 504], onTimeout: true },
|
||||
continueOn: { httpCodes: [402], onTimeout: false },
|
||||
},
|
||||
{
|
||||
id: 'paygo',
|
||||
upstreamIds: ['paygo'],
|
||||
strategy: 'ordered',
|
||||
retryOn: { httpCodes: [429, 500, 502, 503, 504], onTimeout: true },
|
||||
},
|
||||
],
|
||||
},
|
||||
fallbackTriggers: {
|
||||
httpCodes: [401, 402, 429, 500, 502, 503, 504],
|
||||
onTimeout: true,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
defaults: {
|
||||
perAttemptTimeoutMs: 30000,
|
||||
fullChainTimeoutMs: 60000,
|
||||
fallbackHttpCodes: [401, 402, 403, 429, 500, 502, 503, 504],
|
||||
},
|
||||
})
|
||||
|
||||
const value = await service.getOrThrow('LLM_ROUTER_CONFIG')
|
||||
const model = value.tts.models['stepfun/stepaudio-2.5-tts']
|
||||
|
||||
expect(value.llm.models['step-3.5-flash'].routing?.groups.map(group => group.id)).toEqual(['plan', 'paygo'])
|
||||
expect(model.routing?.groups.map(group => group.id)).toEqual(['plan', 'paygo'])
|
||||
expect(model.routing?.groups[0].continueOn).toEqual({
|
||||
httpCodes: [402],
|
||||
onTimeout: false,
|
||||
})
|
||||
})
|
||||
|
||||
it('rejects a TTS provider group that references an unknown upstream', async () => {
|
||||
redis._store.set(configRedisKey('LLM_ROUTER_CONFIG'), JSON.stringify({
|
||||
llm: { models: {} },
|
||||
tts: {
|
||||
models: {
|
||||
tts: {
|
||||
provider: 'stepfun',
|
||||
upstreams: [{
|
||||
id: 'plan',
|
||||
baseURL: 'https://api.stepfun.com',
|
||||
keys: [{ id: 'plan-key', ciphertext: 'ciphertext' }],
|
||||
}],
|
||||
routing: {
|
||||
groups: [{
|
||||
id: 'plan',
|
||||
upstreamIds: ['missing'],
|
||||
strategy: 'ordered',
|
||||
retryOn: { httpCodes: [402], onTimeout: false },
|
||||
}],
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}))
|
||||
|
||||
await expect(service.getOptional('LLM_ROUTER_CONFIG'))
|
||||
.rejects
|
||||
.toMatchObject({
|
||||
statusCode: 503,
|
||||
errorCode: 'CONFIG_INVALID',
|
||||
})
|
||||
})
|
||||
|
||||
it('rejects least-inflight routing without an explicit concurrency cap', async () => {
|
||||
redis._store.set(configRedisKey('LLM_ROUTER_CONFIG'), JSON.stringify({
|
||||
llm: { models: {} },
|
||||
tts: {
|
||||
models: {
|
||||
tts: {
|
||||
provider: 'stepfun',
|
||||
upstreams: [{
|
||||
id: 'plan',
|
||||
baseURL: 'https://api.stepfun.com',
|
||||
keys: [{ id: 'plan-key', ciphertext: 'ciphertext' }],
|
||||
}],
|
||||
routing: {
|
||||
groups: [{
|
||||
id: 'plan',
|
||||
upstreamIds: ['plan'],
|
||||
strategy: 'least-inflight',
|
||||
retryOn: { httpCodes: [402], onTimeout: false },
|
||||
}],
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}))
|
||||
|
||||
await expect(service.getOptional('LLM_ROUTER_CONFIG'))
|
||||
.rejects
|
||||
.toMatchObject({
|
||||
statusCode: 503,
|
||||
errorCode: 'CONFIG_INVALID',
|
||||
})
|
||||
})
|
||||
|
||||
it('set should store string values as JSON strings', async () => {
|
||||
await service.set('STRIPE_FLUX_PRODUCT_ID', 'prod_abc123')
|
||||
|
||||
|
||||
@@ -9,8 +9,9 @@ import { configRedisKey } from '../../utils/redis-keys'
|
||||
|
||||
/**
|
||||
* LLM/TTS router config tree. Single composite entry under configKV holds the
|
||||
* entire routing surface: per-model upstream list, per-upstream key array
|
||||
* (envelope-encrypted ciphertexts), fallback triggers, default timeouts.
|
||||
* entire routing surface: per-model upstream list, optional candidate groups,
|
||||
* per-upstream key array (envelope-encrypted ciphertexts), transition policies,
|
||||
* and default timeouts.
|
||||
*
|
||||
* Schema enforces:
|
||||
* - key entry id must not contain `|` — the envelope-crypto AAD uses `|` as
|
||||
@@ -30,6 +31,18 @@ export const fallbackTriggersSchema = optional(
|
||||
{ httpCodes: [401, 402, 403, 429, 500, 502, 503, 504], onTimeout: true },
|
||||
)
|
||||
|
||||
/**
|
||||
* Explicit allow-list for one routing transition.
|
||||
*
|
||||
* Unlike {@link fallbackTriggersSchema}, this contract has no permissive
|
||||
* defaults: an omitted status or timeout never authorizes a transition across
|
||||
* a configured routing boundary.
|
||||
*/
|
||||
export const routeFailureTriggersSchema = object({
|
||||
httpCodes: optional(array(number()), []),
|
||||
onTimeout: optional(boolean(), false),
|
||||
})
|
||||
|
||||
export const keyEntrySchema = object({
|
||||
id: pipe(
|
||||
string(),
|
||||
@@ -40,6 +53,11 @@ export const keyEntrySchema = object({
|
||||
})
|
||||
|
||||
export const llmUpstreamSchema = object({
|
||||
id: optional(pipe(
|
||||
string(),
|
||||
nonEmpty('llm.upstreams[].id must not be empty'),
|
||||
regex(/^[^|]+$/, 'llm.upstreams[].id must not contain "|"'),
|
||||
)),
|
||||
baseURL: pipe(string(), nonEmpty('llm.upstreams[].baseURL must not be empty')),
|
||||
overrideModel: optional(string()),
|
||||
keys: pipe(array(keyEntrySchema), check(v => v.length >= 1, 'llm.upstreams[].keys must contain at least 1 entry')),
|
||||
@@ -47,15 +65,58 @@ export const llmUpstreamSchema = object({
|
||||
timeoutMs: optional(number()),
|
||||
})
|
||||
|
||||
export const llmModelSchema = object({
|
||||
upstreams: pipe(array(llmUpstreamSchema), check(v => v.length >= 1, 'llm.models[].upstreams must contain at least 1 entry')),
|
||||
fallbackTriggers: fallbackTriggersSchema,
|
||||
export const llmRoutingGroupSchema = object({
|
||||
id: pipe(string(), nonEmpty('llm.routing.groups[].id must not be empty')),
|
||||
upstreamIds: pipe(
|
||||
array(pipe(string(), nonEmpty('llm.routing.groups[].upstreamIds[] must not be empty'))),
|
||||
check(v => v.length >= 1, 'llm.routing.groups[].upstreamIds must contain at least 1 entry'),
|
||||
check(v => new Set(v).size === v.length, 'llm.routing.groups[].upstreamIds must be unique'),
|
||||
),
|
||||
retryOn: routeFailureTriggersSchema,
|
||||
continueOn: optional(routeFailureTriggersSchema),
|
||||
})
|
||||
|
||||
export const llmRoutingSchema = object({
|
||||
groups: pipe(
|
||||
array(llmRoutingGroupSchema),
|
||||
check(v => v.length >= 1, 'llm.routing.groups must contain at least 1 entry'),
|
||||
check(v => new Set(v.map(group => group.id)).size === v.length, 'llm.routing.groups[].id must be unique'),
|
||||
),
|
||||
})
|
||||
|
||||
export const llmModelSchema = pipe(
|
||||
object({
|
||||
upstreams: pipe(array(llmUpstreamSchema), check(v => v.length >= 1, 'llm.models[].upstreams must contain at least 1 entry')),
|
||||
routing: optional(llmRoutingSchema),
|
||||
fallbackTriggers: fallbackTriggersSchema,
|
||||
}),
|
||||
check((model) => {
|
||||
if (model.routing == null)
|
||||
return true
|
||||
const upstreamIds = model.upstreams.map(upstream => upstream.id)
|
||||
return upstreamIds.every(id => id != null)
|
||||
&& new Set(upstreamIds).size === upstreamIds.length
|
||||
}, 'llm.models[].upstreams must have unique ids when routing is configured'),
|
||||
check((model) => {
|
||||
if (model.routing == null)
|
||||
return true
|
||||
const upstreamIds = new Set(model.upstreams.map(upstream => upstream.id))
|
||||
const referencedIds = model.routing.groups.flatMap(group => group.upstreamIds)
|
||||
return referencedIds.length === upstreamIds.size
|
||||
&& new Set(referencedIds).size === referencedIds.length
|
||||
&& referencedIds.every(id => upstreamIds.has(id))
|
||||
}, 'llm.routing.groups must reference every upstream id exactly once'),
|
||||
)
|
||||
|
||||
const ttsProviderSchema = picklist(['azure', 'dashscope-cosyvoice', 'stepfun', 'volcengine'])
|
||||
const asrProviderSchema = picklist(['aliyun-nls'])
|
||||
|
||||
export const ttsUpstreamSchema = object({
|
||||
id: optional(pipe(
|
||||
string(),
|
||||
nonEmpty('tts.upstreams[].id must not be empty'),
|
||||
regex(/^[^|]+$/, 'tts.upstreams[].id must not contain "|"'),
|
||||
)),
|
||||
baseURL: pipe(string(), nonEmpty('tts.upstreams[].baseURL must not be empty')),
|
||||
keys: pipe(array(keyEntrySchema), check(v => v.length >= 1, 'tts.upstreams[].keys must contain at least 1 entry')),
|
||||
adapterParams: optional(record(string(), any()), {}),
|
||||
@@ -68,6 +129,26 @@ export const ttsUpstreamSchema = object({
|
||||
maxConcurrency: optional(pipe(number(), check(v => v >= 1, 'tts.upstreams[].maxConcurrency must be >= 1 when set'))),
|
||||
})
|
||||
|
||||
export const ttsRoutingGroupSchema = object({
|
||||
id: pipe(string(), nonEmpty('tts.routing.groups[].id must not be empty')),
|
||||
upstreamIds: pipe(
|
||||
array(pipe(string(), nonEmpty('tts.routing.groups[].upstreamIds[] must not be empty'))),
|
||||
check(v => v.length >= 1, 'tts.routing.groups[].upstreamIds must contain at least 1 entry'),
|
||||
check(v => new Set(v).size === v.length, 'tts.routing.groups[].upstreamIds must be unique'),
|
||||
),
|
||||
strategy: optional(picklist(['ordered', 'least-inflight']), 'ordered'),
|
||||
retryOn: routeFailureTriggersSchema,
|
||||
continueOn: optional(routeFailureTriggersSchema),
|
||||
})
|
||||
|
||||
export const ttsRoutingSchema = object({
|
||||
groups: pipe(
|
||||
array(ttsRoutingGroupSchema),
|
||||
check(v => v.length >= 1, 'tts.routing.groups must contain at least 1 entry'),
|
||||
check(v => new Set(v.map(group => group.id)).size === v.length, 'tts.routing.groups[].id must be unique'),
|
||||
),
|
||||
})
|
||||
|
||||
export const streamingTtsUpstreamSchema = object({
|
||||
baseURL: pipe(string(), nonEmpty('UNSPEECH_UPSTREAM.streaming.baseURL must not be empty')),
|
||||
keys: pipe(array(keyEntrySchema), check(v => v.length >= 1, 'UNSPEECH_UPSTREAM.streaming.keys must contain at least 1 entry')),
|
||||
@@ -88,11 +169,39 @@ export const unspeechUpstreamSchema = object({
|
||||
streaming: optional(streamingTtsUpstreamSchema),
|
||||
})
|
||||
|
||||
export const ttsModelSchema = object({
|
||||
provider: ttsProviderSchema,
|
||||
upstreams: pipe(array(ttsUpstreamSchema), check(v => v.length >= 1, 'tts.models[].upstreams must contain at least 1 entry')),
|
||||
fallbackTriggers: fallbackTriggersSchema,
|
||||
})
|
||||
export const ttsModelSchema = pipe(
|
||||
object({
|
||||
provider: ttsProviderSchema,
|
||||
upstreams: pipe(array(ttsUpstreamSchema), check(v => v.length >= 1, 'tts.models[].upstreams must contain at least 1 entry')),
|
||||
routing: optional(ttsRoutingSchema),
|
||||
fallbackTriggers: fallbackTriggersSchema,
|
||||
}),
|
||||
check((model) => {
|
||||
if (model.routing == null)
|
||||
return true
|
||||
const upstreamIds = model.upstreams.map(upstream => upstream.id)
|
||||
return upstreamIds.every(id => id != null)
|
||||
&& new Set(upstreamIds).size === upstreamIds.length
|
||||
}, 'tts.models[].upstreams must have unique ids when routing is configured'),
|
||||
check((model) => {
|
||||
if (model.routing == null)
|
||||
return true
|
||||
const upstreamIds = new Set(model.upstreams.map(upstream => upstream.id))
|
||||
const referencedIds = model.routing.groups.flatMap(group => group.upstreamIds)
|
||||
return referencedIds.length === upstreamIds.size
|
||||
&& new Set(referencedIds).size === referencedIds.length
|
||||
&& referencedIds.every(id => upstreamIds.has(id))
|
||||
}, 'tts.routing.groups must reference every upstream id exactly once'),
|
||||
check((model) => {
|
||||
if (model.routing == null)
|
||||
return true
|
||||
const upstreamById = new Map(model.upstreams.map(upstream => [upstream.id, upstream]))
|
||||
return model.routing.groups.every(group =>
|
||||
group.strategy !== 'least-inflight'
|
||||
|| group.upstreamIds.every(id => upstreamById.get(id)?.maxConcurrency != null),
|
||||
)
|
||||
}, 'tts.routing least-inflight groups require maxConcurrency on every upstream'),
|
||||
)
|
||||
|
||||
export const asrUpstreamSchema = object({
|
||||
keys: pipe(array(keyEntrySchema), check(v => v.length >= 1, 'asr.upstreams[].keys must contain at least 1 entry')),
|
||||
|
||||
@@ -298,7 +298,7 @@ describe('azureAdapter.send', () => {
|
||||
})
|
||||
|
||||
describe('stepfunAdapter', () => {
|
||||
it('lists StepFun voices through unspeech provider=stepfun', async () => {
|
||||
it('uses unspeech as the StepFun voice-catalog source', async () => {
|
||||
const adapter = getAdapter('stepfun')
|
||||
const fetchImpl = vi.fn(async () => new Response(JSON.stringify({
|
||||
voices: [{
|
||||
@@ -323,6 +323,7 @@ describe('stepfunAdapter', () => {
|
||||
}),
|
||||
]),
|
||||
)
|
||||
expect(fetchImpl).toHaveBeenCalledTimes(1)
|
||||
const [calledUrl] = (fetchImpl as unknown as { mock: { calls: [string, RequestInit][] } }).mock.calls[0]
|
||||
expect(calledUrl).toBe('http://unspeech.local/api/voices?provider=stepfun')
|
||||
})
|
||||
@@ -348,7 +349,7 @@ describe('stepfunAdapter', () => {
|
||||
},
|
||||
{
|
||||
keyPlaintext: Buffer.from('step-key', 'utf8'),
|
||||
baseURL: 'https://api.stepfun.com/v1/audio/speech',
|
||||
baseURL: 'https://api.stepfun.com',
|
||||
unspeechBaseURL: 'http://unspeech.local:5933',
|
||||
adapterParams: { model: 'stepaudio-2.5-tts' },
|
||||
fetchImpl,
|
||||
@@ -379,6 +380,57 @@ describe('stepfunAdapter', () => {
|
||||
expect(result.body).toBeInstanceOf(ArrayBuffer)
|
||||
})
|
||||
|
||||
it('passes the Step Plan endpoint profile to unspeech', async () => {
|
||||
const adapter = getAdapter('stepfun')
|
||||
const fetchImpl = vi.fn(async () => new Response(new Uint8Array([1, 2, 3]), {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'audio/mpeg' },
|
||||
})) as unknown as typeof fetch
|
||||
|
||||
const result = await adapter.send(
|
||||
{
|
||||
text: '你好',
|
||||
voice: 'cixingnansheng',
|
||||
responseFormat: 'mp3',
|
||||
speed: 1.1,
|
||||
extraOptions: {
|
||||
instruction: '温柔、克制',
|
||||
},
|
||||
},
|
||||
{
|
||||
keyPlaintext: Buffer.from('step-plan-key', 'utf8'),
|
||||
baseURL: 'https://api.stepfun.com',
|
||||
unspeechBaseURL: 'http://unspeech.local:5933',
|
||||
adapterParams: {
|
||||
endpointProfile: 'step-plan',
|
||||
model: 'stepaudio-2.5-tts',
|
||||
},
|
||||
fetchImpl,
|
||||
},
|
||||
)
|
||||
|
||||
const [calledURL, init] = (fetchImpl as unknown as { mock: { calls: [string, RequestInit][] } }).mock.calls[0]
|
||||
expect(calledURL).toBe('http://unspeech.local:5933/v1/audio/speech')
|
||||
expect(init.method).toBe('POST')
|
||||
expect(init.headers).toMatchObject({
|
||||
'Authorization': 'Bearer step-plan-key',
|
||||
'Content-Type': 'application/json',
|
||||
})
|
||||
expect(JSON.parse(init.body as string)).toEqual({
|
||||
model: 'stepfun/stepaudio-2.5-tts',
|
||||
input: '你好',
|
||||
voice: 'cixingnansheng',
|
||||
response_format: 'mp3',
|
||||
speed: 1.1,
|
||||
extra_body: {
|
||||
endpoint_profile: 'step-plan',
|
||||
instruction: '温柔、克制',
|
||||
},
|
||||
})
|
||||
expect(result.contentType).toBe('audio/mpeg')
|
||||
expect(result.body).toBeInstanceOf(ArrayBuffer)
|
||||
})
|
||||
|
||||
it('passes voice_label through to unspeech for provider-level validation', async () => {
|
||||
const adapter = getAdapter('stepfun')
|
||||
const fetchImpl = vi.fn(async () => new Response(new Uint8Array([1]), {
|
||||
@@ -395,7 +447,7 @@ describe('stepfunAdapter', () => {
|
||||
},
|
||||
{
|
||||
keyPlaintext: Buffer.from('step-key', 'utf8'),
|
||||
baseURL: 'https://api.stepfun.com/v1/audio/speech',
|
||||
baseURL: 'https://api.stepfun.com',
|
||||
unspeechBaseURL: 'http://unspeech.local',
|
||||
adapterParams: { model: 'stepaudio-2.5-tts' },
|
||||
fetchImpl,
|
||||
@@ -415,13 +467,35 @@ describe('stepfunAdapter', () => {
|
||||
{ text: 'hi', voice: 'cixingnansheng' },
|
||||
{
|
||||
keyPlaintext: Buffer.from('bad-key', 'utf8'),
|
||||
baseURL: 'https://api.stepfun.com/v1/audio/speech',
|
||||
baseURL: 'https://api.stepfun.com',
|
||||
unspeechBaseURL: 'http://unspeech.local',
|
||||
adapterParams: { model: 'stepaudio-2.5-tts' },
|
||||
fetchImpl,
|
||||
},
|
||||
)).rejects.toMatchObject({ status: 401 })
|
||||
})
|
||||
|
||||
it('preserves an unspeech request abort for router timeout classification', async () => {
|
||||
const adapter = getAdapter('stepfun')
|
||||
const abortController = new AbortController()
|
||||
const abortError = new Error('attempt-timeout')
|
||||
abortController.abort(abortError)
|
||||
const fetchImpl = vi.fn(async (_input: string | URL | Request, init?: RequestInit) => {
|
||||
throw init?.signal?.reason ?? new Error('aborted')
|
||||
}) as unknown as typeof fetch
|
||||
|
||||
await expect(adapter.send(
|
||||
{ text: 'hi', voice: 'cixingnansheng' },
|
||||
{
|
||||
keyPlaintext: Buffer.from('step-key', 'utf8'),
|
||||
baseURL: 'https://api.stepfun.com',
|
||||
unspeechBaseURL: 'http://unspeech.local',
|
||||
adapterParams: { model: 'stepaudio-2.5-tts' },
|
||||
fetchImpl,
|
||||
abortSignal: abortController.signal,
|
||||
},
|
||||
)).rejects.toBe(abortError)
|
||||
})
|
||||
})
|
||||
|
||||
describe('volcengineAdapter.send', () => {
|
||||
|
||||
@@ -15,8 +15,8 @@ const STEPFUN_DEFAULT_VOICE = 'cixingnansheng'
|
||||
* StepFun TTS adapter.
|
||||
*
|
||||
* Use when:
|
||||
* - Routing hosted speech synthesis to StepFun through unspeech's
|
||||
* OpenAI-compatible `stepfun/*` backend.
|
||||
* - Routing speech synthesis to StepFun through unspeech's OpenAI-compatible
|
||||
* `stepfun/*` backend.
|
||||
*
|
||||
* Expects:
|
||||
* - `ctx.unspeechBaseURL` points at an unspeech deployment that includes the
|
||||
@@ -24,6 +24,8 @@ const STEPFUN_DEFAULT_VOICE = 'cixingnansheng'
|
||||
* - `ctx.keyPlaintext` is the StepFun API key.
|
||||
* - `ctx.adapterParams.model` optionally selects `stepaudio-2.5-tts`,
|
||||
* `step-tts-2`, or `step-tts-mini`.
|
||||
* - `ctx.adapterParams.endpointProfile` optionally selects a provider-owned
|
||||
* endpoint profile such as `step-plan`; AIRI never owns the endpoint URL.
|
||||
*
|
||||
* Returns:
|
||||
* - {@link TtsResult} with the upstream audio body and content type.
|
||||
@@ -41,6 +43,7 @@ export const stepfunAdapter: TtsAdapter = {
|
||||
const responseFormat = input.responseFormat ?? (typeof ctx.adapterParams.responseFormat === 'string' && ctx.adapterParams.responseFormat
|
||||
? ctx.adapterParams.responseFormat
|
||||
: STEPFUN_DEFAULT_FORMAT)
|
||||
const extraBody = buildExtraBody(input, ctx)
|
||||
|
||||
return sendSpeechViaUnSpeech({
|
||||
ctx,
|
||||
@@ -49,7 +52,7 @@ export const stepfunAdapter: TtsAdapter = {
|
||||
voice,
|
||||
speed: input.speed,
|
||||
responseFormat,
|
||||
extraBody: buildExtraBody(input, ctx),
|
||||
extraBody,
|
||||
fallbackContentType: audioMimeFromFormat(responseFormat),
|
||||
providerLabel: 'stepfun',
|
||||
})
|
||||
@@ -68,6 +71,9 @@ function buildExtraBody(input: TtsInput, ctx: TtsAdapterContext): Record<string,
|
||||
const extraOptions = input.extraOptions ?? {}
|
||||
const body: Record<string, unknown> = {}
|
||||
|
||||
if (typeof ctx.adapterParams.endpointProfile === 'string' && ctx.adapterParams.endpointProfile)
|
||||
body.endpoint_profile = ctx.adapterParams.endpointProfile
|
||||
|
||||
if (typeof extraOptions.volume === 'number' && Number.isFinite(extraOptions.volume))
|
||||
body.volume = extraOptions.volume
|
||||
else if (typeof ctx.adapterParams.volume === 'number' && Number.isFinite(ctx.adapterParams.volume))
|
||||
|
||||
@@ -42,16 +42,11 @@ export interface TtsAdapterContext {
|
||||
/**
|
||||
* Per-upstream baseURL from `LLM_ROUTER_CONFIG.tts.upstreams[i].baseURL`.
|
||||
*
|
||||
* Historically the upstream provider URL (e.g.
|
||||
* `https://eastasia.tts.speech.microsoft.com/cognitiveservices/v1`). After
|
||||
* the Phase-B unspeech migration, adapters no longer call upstreams
|
||||
* directly — every `send()` forwards through unspeech REST — so this field
|
||||
* is informational only and adapters MAY ignore it. Kept on the context so
|
||||
* existing operator configs continue to validate (the schema requires a
|
||||
* non-empty string).
|
||||
* Adapters forward through unspeech and may use this as provider metadata.
|
||||
* Provider endpoint selection belongs to unspeech, not this URL.
|
||||
*/
|
||||
baseURL: string
|
||||
/** unspeech REST base URL (no trailing slash) — adapters POST to `<this>/v1/audio/speech`. */
|
||||
/** unspeech REST base URL (no trailing slash). */
|
||||
unspeechBaseURL: string
|
||||
/** Free-form adapter-specific params from `tts.upstreams[i].adapterParams` (e.g. Volcengine `appid` / `cluster`). */
|
||||
adapterParams: Record<string, unknown>
|
||||
@@ -90,12 +85,10 @@ export type TtsAdapterId = 'azure' | 'dashscope-cosyvoice' | 'stepfun' | 'volcen
|
||||
* `keyPlaintext` and `region` are mandatory for live providers (Azure) that
|
||||
* proxy through unspeech and call the upstream provider with a subscription
|
||||
* key; the router decrypts the envelope key and forwards `adapterParams.region`
|
||||
* verbatim. Providers with static, credential-less catalogs (DashScope
|
||||
* cosyvoice, Volcengine) ignore both fields.
|
||||
* verbatim. Unspeech-backed static catalogs ignore both fields.
|
||||
*
|
||||
* `unspeechBaseURL` is `UNSPEECH_UPSTREAM.restBaseURL` resolved by the
|
||||
* router. Passing it through the context keeps adapters free of configKV
|
||||
* coupling — they receive a fully-resolved URL string.
|
||||
* `unspeechBaseURL` is `UNSPEECH_UPSTREAM.restBaseURL` resolved by the router.
|
||||
* Passing it through the context keeps adapters free of configKV coupling.
|
||||
*/
|
||||
export interface TtsVoiceCatalogContext {
|
||||
/** Decrypted upstream credential (live providers only). */
|
||||
@@ -139,10 +132,9 @@ export interface TtsAdapter {
|
||||
/**
|
||||
* Returns the voice catalog for the provider.
|
||||
*
|
||||
* Live providers (Azure) call upstream via unspeech using the supplied
|
||||
* region + plaintext key. Static providers (dashscope-cosyvoice, volcengine)
|
||||
* return their compiled-in JSON and ignore the context fields. Adapters
|
||||
* MUST throw on upstream failure — no empty-array fallback.
|
||||
* Live providers (Azure) call upstream through unspeech using the supplied
|
||||
* region + plaintext key. Static provider catalogs are also owned and served
|
||||
* by unspeech. Adapters MUST throw on upstream failure — no empty fallback.
|
||||
*/
|
||||
getVoiceCatalog: (ctx: TtsVoiceCatalogContext) => Promise<Voice[]>
|
||||
}
|
||||
|
||||
@@ -66,6 +66,11 @@ export async function sendSpeechViaUnSpeech(options: SendSpeechOptions): Promise
|
||||
}
|
||||
}
|
||||
catch (error) {
|
||||
// Keep abort identity intact so the router can apply `onTimeout`
|
||||
// independently from HTTP 500 fallback policy.
|
||||
if (ctx.abortSignal?.aborted)
|
||||
throw error
|
||||
|
||||
if (error instanceof UnSpeechAPIError) {
|
||||
const err = new Error(`${providerLabel} tts upstream ${error.status}: ${error.responseBody.slice(0, 256)}`) as Error & { status?: number }
|
||||
err.status = error.status
|
||||
|
||||
@@ -391,7 +391,7 @@ export function buildStepfunSlice(input: StepfunSliceInput, envelope: EnvelopeCr
|
||||
model: {
|
||||
provider: 'stepfun',
|
||||
upstreams: [{
|
||||
baseURL: 'https://api.stepfun.com/v1/audio/speech',
|
||||
baseURL: 'https://api.stepfun.com',
|
||||
keys: [{ id: keyEntryId, ciphertext }],
|
||||
adapterParams: {
|
||||
model: input.upstreamModel ?? 'stepaudio-2.5-tts',
|
||||
@@ -604,7 +604,7 @@ function buildStepfunSlicePreservingKey(input: StepfunSliceInput, envelope: Enve
|
||||
model: {
|
||||
provider: 'stepfun',
|
||||
upstreams: [{
|
||||
baseURL: 'https://api.stepfun.com/v1/audio/speech',
|
||||
baseURL: 'https://api.stepfun.com',
|
||||
keys: [key],
|
||||
adapterParams: {
|
||||
model: input.upstreamModel ?? 'stepaudio-2.5-tts',
|
||||
@@ -715,6 +715,11 @@ export function buildSlice(
|
||||
* - The next config tree, ready to feed `configKV.set('LLM_ROUTER_CONFIG', ...)`.
|
||||
* `defaults` is preserved verbatim when merging — the admin endpoint does
|
||||
* not currently re-tune timeouts via this path.
|
||||
*
|
||||
* Throws:
|
||||
* - When merge mode targets a grouped LLM/TTS model. The legacy slice contract
|
||||
* cannot identify one upstream or represent routing groups, so replacing that
|
||||
* model would silently erase its routing policy.
|
||||
*/
|
||||
export function buildNextRouterConfig(
|
||||
mode: 'merge' | 'reset',
|
||||
@@ -729,12 +734,25 @@ export function buildNextRouterConfig(
|
||||
= mode === 'merge' && existing?.asr?.models ? { ...existing.asr.models } : {}
|
||||
|
||||
for (const slice of slices) {
|
||||
if (slice.surface === 'llm')
|
||||
if (slice.surface === 'llm') {
|
||||
if (mode === 'merge' && llmModels[slice.modelName]?.routing != null) {
|
||||
throw new Error(
|
||||
`Legacy admin config cannot update grouped llm model ${slice.modelName}; use a group-aware router config update`,
|
||||
)
|
||||
}
|
||||
llmModels[slice.modelName] = slice.model
|
||||
else if (slice.surface === 'tts')
|
||||
}
|
||||
else if (slice.surface === 'tts') {
|
||||
if (mode === 'merge' && ttsModels[slice.modelName]?.routing != null) {
|
||||
throw new Error(
|
||||
`Legacy admin config cannot update grouped tts model ${slice.modelName}; use a group-aware router config update`,
|
||||
)
|
||||
}
|
||||
ttsModels[slice.modelName] = slice.model
|
||||
else
|
||||
}
|
||||
else {
|
||||
asrModels[slice.modelName] = slice.model
|
||||
}
|
||||
}
|
||||
|
||||
// Defaults live alongside the models but aren't editable through this
|
||||
|
||||
+65
-1
@@ -271,7 +271,7 @@ describe('buildStepfunSlice', () => {
|
||||
|
||||
expect(built.kind).toBe('stepfun')
|
||||
expect(built.model.provider).toBe('stepfun')
|
||||
expect(built.model.upstreams[0].baseURL).toBe('https://api.stepfun.com/v1/audio/speech')
|
||||
expect(built.model.upstreams[0].baseURL).toBe('https://api.stepfun.com')
|
||||
expect(built.model.upstreams[0].adapterParams).toEqual({
|
||||
model: 'stepaudio-2.5-tts',
|
||||
defaultVoice: 'cixingnansheng',
|
||||
@@ -582,6 +582,70 @@ describe('createAdminRouterConfigService', () => {
|
||||
expect(Object.keys(written.tts.models)).toEqual(['microsoft/v1'])
|
||||
})
|
||||
|
||||
it('rejects legacy merge updates that would replace a grouped TTS model', async () => {
|
||||
const modelName = 'stepfun/stepaudio-2.5-tts'
|
||||
const existingConfig = {
|
||||
llm: { models: {} },
|
||||
tts: {
|
||||
models: {
|
||||
[modelName]: {
|
||||
provider: 'stepfun' as const,
|
||||
upstreams: [
|
||||
{
|
||||
id: 'plan',
|
||||
baseURL: 'https://api.stepfun.com',
|
||||
keys: [{ id: 'plan-key', ciphertext: 'plan-ciphertext' }],
|
||||
adapterParams: { endpointProfile: 'step-plan', model: 'stepaudio-2.5-tts' },
|
||||
},
|
||||
{
|
||||
id: 'paygo',
|
||||
baseURL: 'https://api.stepfun.com',
|
||||
keys: [{ id: 'paygo-key', ciphertext: 'paygo-ciphertext' }],
|
||||
adapterParams: { endpointProfile: 'default', model: 'stepaudio-2.5-tts' },
|
||||
},
|
||||
],
|
||||
routing: {
|
||||
groups: [
|
||||
{
|
||||
id: 'plan',
|
||||
upstreamIds: ['plan'],
|
||||
strategy: 'ordered' as const,
|
||||
retryOn: { httpCodes: [402], onTimeout: false },
|
||||
continueOn: { httpCodes: [402], onTimeout: false },
|
||||
},
|
||||
{
|
||||
id: 'paygo',
|
||||
upstreamIds: ['paygo'],
|
||||
strategy: 'ordered' as const,
|
||||
retryOn: { httpCodes: [429, 500], onTimeout: true },
|
||||
},
|
||||
],
|
||||
},
|
||||
fallbackTriggers: DEFAULT_FALLBACK_TRIGGERS,
|
||||
},
|
||||
},
|
||||
},
|
||||
defaults: { perAttemptTimeoutMs: 30000, fullChainTimeoutMs: 60000, fallbackHttpCodes: [500] },
|
||||
}
|
||||
kv.store.set('LLM_ROUTER_CONFIG', existingConfig)
|
||||
|
||||
const service = createAdminRouterConfigService({ configKV: kv.service, envelope, redis })
|
||||
|
||||
await expect(service.apply({
|
||||
mode: 'merge',
|
||||
dryRun: false,
|
||||
slices: [{
|
||||
kind: 'stepfun',
|
||||
modelName,
|
||||
upstreamModel: 'stepaudio-2.5-tts',
|
||||
plaintextKey: 'rotated-key',
|
||||
}],
|
||||
})).rejects.toThrow(/cannot update grouped tts model/i)
|
||||
|
||||
expect(kv.store.get('LLM_ROUTER_CONFIG')).toBe(existingConfig)
|
||||
expect(captured).toEqual([])
|
||||
})
|
||||
|
||||
it('current returns editable slices from configKV without exposing raw ciphertext', async () => {
|
||||
kv.store.set('LLM_ROUTER_CONFIG', {
|
||||
llm: {
|
||||
|
||||
@@ -8,7 +8,7 @@ import type { EnvelopeCrypto } from '../../../utils/envelope-crypto'
|
||||
import type { ConfigKVService } from '../../adapters/config-kv'
|
||||
import type { TtsAdapterId, TtsInput } from '../../adapters/tts/types'
|
||||
import type { ConcurrencyLedger } from './concurrency-ledger'
|
||||
import type { LlmRouteContext, LlmRouteRequest, LlmUpstream, TtsUpstream } from './types'
|
||||
import type { LlmRouteContext, LlmRouteRequest, LlmRoutingGroup, LlmUpstream, RouteFailureTriggers, TtsRoutingGroup, TtsUpstream } from './types'
|
||||
|
||||
import { Buffer as NodeBuffer } from 'node:buffer'
|
||||
|
||||
@@ -95,13 +95,34 @@ function deriveProviderTag(baseURL: string): string {
|
||||
/**
|
||||
* Identity of the pool (concurrency pool) one TTS upstream belongs to. One
|
||||
* upstream == one app_id, so the Volcengine `adapterParams.appid` is the pool
|
||||
* key when present; the baseURL is a stable fallback for providers without an
|
||||
* app_id concept. Two upstreams sharing an app_id would (correctly) share one
|
||||
* concurrency budget, though thetypical config gives each app_id its own upstream.
|
||||
* key when present. Other providers use a model-scoped upstream id, then a
|
||||
* model-scoped baseURL for configs without ids. This prevents model-local ids
|
||||
* such as `plan` from sharing Redis counters across unrelated models. Two
|
||||
* upstreams sharing an app_id correctly share one global concurrency budget.
|
||||
*/
|
||||
function ttsPoolId(upstream: TtsUpstream): string {
|
||||
function ttsPoolId(upstream: TtsUpstream, modelName: string): string {
|
||||
const appid = upstream.adapterParams?.appid
|
||||
return typeof appid === 'string' && appid.length > 0 ? appid : upstream.baseURL
|
||||
if (typeof appid === 'string' && appid.length > 0)
|
||||
return appid
|
||||
return `model:${JSON.stringify([
|
||||
modelName,
|
||||
upstream.id == null ? 'baseURL' : 'id',
|
||||
upstream.id ?? upstream.baseURL,
|
||||
])}`
|
||||
}
|
||||
|
||||
function failuresMatch(
|
||||
statuses: ReadonlyArray<number | 'timeout'>,
|
||||
triggers: RouteFailureTriggers | undefined,
|
||||
): boolean {
|
||||
if (statuses.length === 0 || triggers == null)
|
||||
return false
|
||||
|
||||
return statuses.every((status) => {
|
||||
if (status === 'timeout')
|
||||
return triggers.onTimeout
|
||||
return triggers.httpCodes.includes(status)
|
||||
})
|
||||
}
|
||||
|
||||
export interface CreateLlmRouterServiceOptions {
|
||||
@@ -311,10 +332,10 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
|
||||
status_code: status,
|
||||
})
|
||||
if (!fallbackHttpCodes.includes(status)) {
|
||||
// Status not in the fallback whitelist — surface as the last
|
||||
// status and stop walking this upstream. We still let the outer
|
||||
// loop try the next upstream (KTD-13: cross-upstream fallback
|
||||
// happens in the same request, regardless of per-status policy).
|
||||
// Status not in the key-level fallback whitelist — surface as the
|
||||
// last status and stop walking this upstream. A configured provider
|
||||
// group decides separately whether another account may be attempted;
|
||||
// models without groups preserve the historical KTD-13 behavior.
|
||||
attemptIndex += 1
|
||||
break
|
||||
}
|
||||
@@ -367,14 +388,14 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
|
||||
throw new Error(`Expected llm model slice for ${req.modelName}, got ${slice.kind}`)
|
||||
}
|
||||
|
||||
const llmModel = slice.model
|
||||
const defaults = slice.defaults ?? { perAttemptTimeoutMs: 30000, fullChainTimeoutMs: 60000, fallbackHttpCodes: [401, 402, 403, 429, 500, 502, 503, 504] }
|
||||
const fallbackHttpCodes = slice.model.fallbackTriggers?.httpCodes ?? defaults.fallbackHttpCodes ?? [401, 402, 403, 429, 500, 502, 503, 504]
|
||||
const fallbackHttpCodes = llmModel.fallbackTriggers?.httpCodes ?? defaults.fallbackHttpCodes ?? [401, 402, 403, 429, 500, 502, 503, 504]
|
||||
|
||||
const allFailures: Array<{ provider: string, keyId: string, status: number | 'timeout', bodySnippet?: string, errorMessage?: string }> = []
|
||||
let triedUpstreams = 0
|
||||
|
||||
for (let i = 0; i < slice.model.upstreams.length; i += 1) {
|
||||
const upstream = slice.model.upstreams[i]
|
||||
async function attemptUpstream(upstream: LlmUpstream, index: number) {
|
||||
const provider = deriveProviderTag(upstream.baseURL)
|
||||
triedUpstreams += 1
|
||||
// Surface the current upstream so the caller can label success metrics
|
||||
@@ -387,7 +408,7 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
|
||||
|
||||
const result = await dispatchOneUpstream(
|
||||
upstream,
|
||||
i,
|
||||
index,
|
||||
req,
|
||||
perAttemptTimeoutMs,
|
||||
fallbackHttpCodes,
|
||||
@@ -397,14 +418,70 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
|
||||
if (result.kind === 'ok') {
|
||||
if (ctx)
|
||||
ctx.upstreamModel = result.upstreamModel
|
||||
return result.response
|
||||
return { kind: 'ok' as const, response: result.response }
|
||||
}
|
||||
|
||||
// This upstream exhausted; record and continue.
|
||||
options.gatewayMetrics?.keyExhaustedCount.add(1, { provider })
|
||||
return {
|
||||
kind: 'exhausted' as const,
|
||||
statuses: result.failures.map(failure => failure.status),
|
||||
}
|
||||
}
|
||||
|
||||
// FULL exhaustion: every upstream's every key failed.
|
||||
async function routeGroup(group: LlmRoutingGroup): Promise<
|
||||
| { kind: 'ok', response: Response }
|
||||
| { kind: 'exhausted', statuses: Array<number | 'timeout'>, transitionBlocked: boolean }
|
||||
> {
|
||||
const statuses: Array<number | 'timeout'> = []
|
||||
for (let groupCandidateIndex = 0; groupCandidateIndex < group.upstreamIds.length; groupCandidateIndex += 1) {
|
||||
const upstreamId = group.upstreamIds[groupCandidateIndex]
|
||||
const index = llmModel.upstreams.findIndex(upstream => upstream.id === upstreamId)
|
||||
if (index === -1) {
|
||||
throw new Error(
|
||||
`LLM routing group ${group.id} references unknown upstream ${upstreamId} for model ${req.modelName}`,
|
||||
)
|
||||
}
|
||||
|
||||
const result = await attemptUpstream(llmModel.upstreams[index], index)
|
||||
if (result.kind === 'ok')
|
||||
return result
|
||||
statuses.push(...result.statuses)
|
||||
const hasNextCandidate = groupCandidateIndex < group.upstreamIds.length - 1
|
||||
if (hasNextCandidate && !failuresMatch(result.statuses, group.retryOn))
|
||||
return { kind: 'exhausted', statuses, transitionBlocked: true }
|
||||
}
|
||||
|
||||
return { kind: 'exhausted', statuses, transitionBlocked: false }
|
||||
}
|
||||
|
||||
if (llmModel.routing != null) {
|
||||
for (let groupIndex = 0; groupIndex < llmModel.routing.groups.length; groupIndex += 1) {
|
||||
const group = llmModel.routing.groups[groupIndex]
|
||||
const result = await routeGroup(group)
|
||||
if (result.kind === 'ok')
|
||||
return result.response
|
||||
|
||||
const hasNextGroup = groupIndex < llmModel.routing.groups.length - 1
|
||||
if (
|
||||
result.transitionBlocked
|
||||
|| !hasNextGroup
|
||||
|| !failuresMatch(result.statuses, group.continueOn)
|
||||
) {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
else {
|
||||
for (let index = 0; index < llmModel.upstreams.length; index += 1) {
|
||||
const result = await attemptUpstream(llmModel.upstreams[index], index)
|
||||
if (result.kind === 'ok')
|
||||
return result.response
|
||||
}
|
||||
}
|
||||
|
||||
// Terminal exhaustion: every transition allowed by the active provider
|
||||
// route has failed. A policy boundary may intentionally leave later
|
||||
// upstreams untouched.
|
||||
const lastFailure = allFailures.at(-1)
|
||||
if (lastFailure == null) {
|
||||
// Should not happen: schema guarantees ≥1 upstream and ≥1 key. Treat
|
||||
@@ -412,9 +489,9 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
|
||||
throw new Error(`Router exhausted with no recorded failures for model ${req.modelName}`)
|
||||
}
|
||||
|
||||
// Same-status exhaustion: every recorded failure shares the same
|
||||
// status (or 'timeout'). Strong signal of an account-level / shared-
|
||||
// backend cap that ordinary fallback cannot recover from.
|
||||
// Same-status exhaustion: every recorded failure shares one status (or
|
||||
// timeout). This is a strong signal of a shared upstream constraint that
|
||||
// ordinary candidate fallback cannot recover from.
|
||||
const distinctStatuses = new Set(allFailures.map(f => f.status))
|
||||
if (distinctStatuses.size === 1) {
|
||||
const status = allFailures[0].status
|
||||
@@ -553,8 +630,9 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
|
||||
status_code: rawStatus,
|
||||
})
|
||||
if (!fallbackHttpCodes.includes(rawStatus)) {
|
||||
// Same policy as chat: non-fallback status stops this upstream
|
||||
// but the outer loop still tries the next upstream.
|
||||
// Same key-level policy as chat: stop rotating credentials in this
|
||||
// candidate. The enclosing group separately decides whether another
|
||||
// candidate may be tried.
|
||||
attemptIndex += 1
|
||||
break
|
||||
}
|
||||
@@ -572,14 +650,14 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
|
||||
}
|
||||
|
||||
/**
|
||||
* Capacity-aware layer over {@link dispatchOneTtsUpstream}: spreads one TTS
|
||||
* request across the model's pool (one app_id per upstream) by least-loaded
|
||||
* ordering, gating each dispatch on an atomic concurrency-slot acquire.
|
||||
* Capacity-aware layer over {@link dispatchOneTtsUpstream}: gates each capped
|
||||
* dispatch on an atomic concurrency-slot acquire. It can preserve configured
|
||||
* order or rank equivalent accounts by current in-flight usage.
|
||||
*
|
||||
* Returns:
|
||||
* - the 2xx `Response` on success,
|
||||
* - `null` when every dispatched upstream exhausted (caller maps the recorded
|
||||
* failures to an upstream error via the shared exhaustion path),
|
||||
* - an `ok` result with the 2xx `Response`,
|
||||
* - an `exhausted` result with every attempted status and whether a retry
|
||||
* policy blocked the remaining peers,
|
||||
* - throws 503 `TTS_POOL_SATURATED` when every pool was at capacity or in a
|
||||
* 429 cool-down so nothing was dispatched - fail-fast with context, never a
|
||||
* silent stall (origin R3).
|
||||
@@ -589,9 +667,14 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
|
||||
modelName: string,
|
||||
attemptUpstream: (upstream: TtsUpstream, index: number) => Promise<
|
||||
| { kind: 'ok', response: Response }
|
||||
| { kind: 'exhausted', sawTooManyRequests: boolean }
|
||||
| { kind: 'exhausted', sawTooManyRequests: boolean, statuses: Array<number | 'timeout'> }
|
||||
>,
|
||||
): Promise<Response | null> {
|
||||
retryOn?: RouteFailureTriggers,
|
||||
strategy: 'least-inflight' | 'ordered' = 'least-inflight',
|
||||
): Promise<
|
||||
| { kind: 'ok', response: Response }
|
||||
| { kind: 'exhausted', statuses: Array<number | 'timeout'>, transitionBlocked: boolean }
|
||||
> {
|
||||
async function markSaturated(upstream: TtsUpstream, poolId: string): Promise<void> {
|
||||
await ledger.markSaturated(poolId, ttsPoolSaturationTtlSeconds)
|
||||
options.gatewayMetrics?.poolSaturationMarked.add(1, {
|
||||
@@ -600,12 +683,11 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
|
||||
})
|
||||
}
|
||||
|
||||
// Best-effort pre-read: order pools least-loaded-first (spreads load) and
|
||||
// drop pools already full or in a saturation cool-down. tryAcquire below is
|
||||
// the authoritative gate against the cross-replica race — ordering only
|
||||
// decides *preference*, not correctness.
|
||||
const ranked = (await Promise.all(upstreams.map(async (upstream, index) => {
|
||||
const poolId = ttsPoolId(upstream)
|
||||
// Best-effort pre-read drops pools already full or in a saturation
|
||||
// cool-down. tryAcquire below remains the authoritative gate against the
|
||||
// cross-replica race.
|
||||
const candidates = await Promise.all(upstreams.map(async (upstream, index) => {
|
||||
const poolId = ttsPoolId(upstream, modelName)
|
||||
const maxConcurrency = typeof upstream.maxConcurrency === 'number' ? upstream.maxConcurrency : null
|
||||
const saturated = await ledger.isSaturated(poolId)
|
||||
if (saturated) {
|
||||
@@ -614,30 +696,39 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
|
||||
index,
|
||||
poolId,
|
||||
maxConcurrency,
|
||||
remaining: maxConcurrency == null ? Number.POSITIVE_INFINITY : 0,
|
||||
inflight: Number.POSITIVE_INFINITY,
|
||||
eligible: false,
|
||||
}
|
||||
}
|
||||
if (maxConcurrency == null)
|
||||
return { upstream, index, poolId, maxConcurrency, remaining: Number.POSITIVE_INFINITY, eligible: true }
|
||||
return { upstream, index, poolId, maxConcurrency, inflight: 0, eligible: true }
|
||||
|
||||
const inflight = await ledger.currentInflight(poolId)
|
||||
const remaining = maxConcurrency - inflight
|
||||
return { upstream, index, poolId, maxConcurrency, remaining, eligible: remaining > 0 }
|
||||
})))
|
||||
.filter(c => c.eligible)
|
||||
.sort((a, b) => b.remaining - a.remaining)
|
||||
return { upstream, index, poolId, maxConcurrency, inflight, eligible: inflight < maxConcurrency }
|
||||
}))
|
||||
const eligible = candidates.filter(c => c.eligible)
|
||||
const ranked = strategy === 'least-inflight'
|
||||
? eligible.sort((a, b) => a.inflight - b.inflight)
|
||||
: eligible
|
||||
|
||||
let dispatchedAny = false
|
||||
for (const { upstream, index, poolId, maxConcurrency } of ranked) {
|
||||
let attemptedPools = 0
|
||||
const statuses: Array<number | 'timeout'> = []
|
||||
for (let rankedIndex = 0; rankedIndex < ranked.length; rankedIndex += 1) {
|
||||
const { upstream, index, poolId, maxConcurrency } = ranked[rankedIndex]
|
||||
const hasNextCandidate = rankedIndex < ranked.length - 1
|
||||
if (maxConcurrency == null) {
|
||||
// Unlimited pool — dispatch without occupying a slot.
|
||||
dispatchedAny = true
|
||||
attemptedPools += 1
|
||||
const result = await attemptUpstream(upstream, index)
|
||||
if (result.kind === 'ok')
|
||||
return result.response
|
||||
return result
|
||||
statuses.push(...result.statuses)
|
||||
if (result.sawTooManyRequests)
|
||||
await markSaturated(upstream, poolId)
|
||||
if (hasNextCandidate && retryOn != null && !failuresMatch(result.statuses, retryOn))
|
||||
return { kind: 'exhausted', statuses, transitionBlocked: true }
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -652,12 +743,16 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
|
||||
}
|
||||
|
||||
dispatchedAny = true
|
||||
attemptedPools += 1
|
||||
try {
|
||||
const result = await attemptUpstream(upstream, index)
|
||||
if (result.kind === 'ok')
|
||||
return result.response
|
||||
return result
|
||||
statuses.push(...result.statuses)
|
||||
if (result.sawTooManyRequests)
|
||||
await markSaturated(upstream, poolId)
|
||||
if (hasNextCandidate && retryOn != null && !failuresMatch(result.statuses, retryOn))
|
||||
return { kind: 'exhausted', statuses, transitionBlocked: true }
|
||||
}
|
||||
finally {
|
||||
await ledger.release(poolId)
|
||||
@@ -672,7 +767,15 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
|
||||
)
|
||||
}
|
||||
|
||||
return null
|
||||
// Advancing past a group requires evidence from every candidate. A pool
|
||||
// skipped because it was full, circuit-broken, or lost the acquire race has
|
||||
// not produced a failure status, so it must block the transition even when
|
||||
// every dispatched pool returned an allowed status.
|
||||
return {
|
||||
kind: 'exhausted',
|
||||
statuses,
|
||||
transitionBlocked: attemptedPools !== upstreams.length,
|
||||
}
|
||||
}
|
||||
|
||||
async function routeTts(req: { modelName: string, input: TtsInput, abortSignal?: AbortSignal }, ctx?: LlmRouteContext): Promise<Response> {
|
||||
@@ -694,9 +797,6 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
|
||||
const defaults = slice.defaults ?? { perAttemptTimeoutMs: 30000, fullChainTimeoutMs: 60000, fallbackHttpCodes: [401, 402, 403, 429, 500, 502, 503, 504] }
|
||||
const fallbackHttpCodes = ttsModel.fallbackTriggers?.httpCodes ?? defaults.fallbackHttpCodes ?? [401, 402, 403, 429, 500, 502, 503, 504]
|
||||
|
||||
// Adapters POST to unspeech `/v1/audio/speech`; resolve the base URL once
|
||||
// per request rather than per upstream attempt so a single configKV miss
|
||||
// surfaces as a clean 503 before any key rotation happens.
|
||||
const unspeechBaseURL = (await options.configKV.getOrThrow('UNSPEECH_UPSTREAM')).restBaseURL
|
||||
|
||||
const allFailures: Array<{ provider: string, keyId: string, status: number | 'timeout', errorMessage?: string }> = []
|
||||
@@ -712,7 +812,7 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
|
||||
// so the caller can circuit-break thatpool.
|
||||
async function attemptUpstream(upstream: TtsUpstream, index: number): Promise<
|
||||
| { kind: 'ok', response: Response }
|
||||
| { kind: 'exhausted', sawTooManyRequests: boolean }
|
||||
| { kind: 'exhausted', sawTooManyRequests: boolean, statuses: Array<number | 'timeout'> }
|
||||
> {
|
||||
const providerTag = deriveProviderTag(upstream.baseURL)
|
||||
triedUpstreams += 1
|
||||
@@ -741,25 +841,93 @@ export function createLlmRouterService(options: CreateLlmRouterServiceOptions) {
|
||||
}
|
||||
|
||||
options.gatewayMetrics?.keyExhaustedCount.add(1, { provider: providerTag })
|
||||
return { kind: 'exhausted', sawTooManyRequests: result.failures.some(f => f.status === 429) }
|
||||
return {
|
||||
kind: 'exhausted',
|
||||
sawTooManyRequests: result.failures.some(f => f.status === 429),
|
||||
statuses: result.failures.map(failure => failure.status),
|
||||
}
|
||||
}
|
||||
|
||||
// A model "uses the pool" when any upstream declares a concurrency cap. Models
|
||||
// without one keep the original fixed-order fallback and make zero Redis
|
||||
// calls — no behavior change for existing single-app configs.
|
||||
const poolingEnabled = ttsModel.upstreams.some(u => typeof u.maxConcurrency === 'number')
|
||||
async function routeGroup(group: TtsRoutingGroup): Promise<
|
||||
| { kind: 'ok', response: Response }
|
||||
| { kind: 'exhausted', statuses: Array<number | 'timeout'>, transitionBlocked: boolean }
|
||||
> {
|
||||
const indexedUpstreams = group.upstreamIds.map((upstreamId) => {
|
||||
const index = ttsModel.upstreams.findIndex(upstream => upstream.id === upstreamId)
|
||||
if (index === -1) {
|
||||
throw new Error(
|
||||
`TTS routing group ${group.id} references unknown upstream ${upstreamId} for model ${req.modelName}`,
|
||||
)
|
||||
}
|
||||
return { upstream: ttsModel.upstreams[index], index }
|
||||
})
|
||||
|
||||
if (!poolingEnabled) {
|
||||
for (let i = 0; i < ttsModel.upstreams.length; i += 1) {
|
||||
const result = await attemptUpstream(ttsModel.upstreams[i], i)
|
||||
const capacityManaged = indexedUpstreams.some(({ upstream }) => upstream.maxConcurrency != null)
|
||||
if (group.strategy === 'least-inflight' || capacityManaged) {
|
||||
const indexByUpstream = new Map(indexedUpstreams.map(({ upstream, index }) => [upstream, index]))
|
||||
return routeTtsAcrossPools(
|
||||
indexedUpstreams.map(({ upstream }) => upstream),
|
||||
req.modelName,
|
||||
(upstream) => {
|
||||
const index = indexByUpstream.get(upstream)
|
||||
if (index == null)
|
||||
throw new Error(`TTS routing lost upstream index for model ${req.modelName}`)
|
||||
return attemptUpstream(upstream, index)
|
||||
},
|
||||
group.retryOn,
|
||||
group.strategy,
|
||||
)
|
||||
}
|
||||
|
||||
const statuses: Array<number | 'timeout'> = []
|
||||
for (let groupCandidateIndex = 0; groupCandidateIndex < indexedUpstreams.length; groupCandidateIndex += 1) {
|
||||
const { upstream, index } = indexedUpstreams[groupCandidateIndex]
|
||||
const result = await attemptUpstream(upstream, index)
|
||||
if (result.kind === 'ok')
|
||||
return result
|
||||
statuses.push(...result.statuses)
|
||||
const hasNextCandidate = groupCandidateIndex < indexedUpstreams.length - 1
|
||||
if (hasNextCandidate && !failuresMatch(result.statuses, group.retryOn))
|
||||
return { kind: 'exhausted', statuses, transitionBlocked: true }
|
||||
}
|
||||
|
||||
return { kind: 'exhausted', statuses, transitionBlocked: false }
|
||||
}
|
||||
|
||||
if (ttsModel.routing != null) {
|
||||
for (let groupIndex = 0; groupIndex < ttsModel.routing.groups.length; groupIndex += 1) {
|
||||
const group = ttsModel.routing.groups[groupIndex]
|
||||
const result = await routeGroup(group)
|
||||
if (result.kind === 'ok')
|
||||
return result.response
|
||||
|
||||
const hasNextGroup = groupIndex < ttsModel.routing.groups.length - 1
|
||||
if (
|
||||
result.transitionBlocked
|
||||
|| !hasNextGroup
|
||||
|| !failuresMatch(result.statuses, group.continueOn)
|
||||
) {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
else {
|
||||
const served = await routeTtsAcrossPools(ttsModel.upstreams, req.modelName, attemptUpstream)
|
||||
if (served != null)
|
||||
return served
|
||||
// Models without an explicit provider route preserve the established
|
||||
// behavior: fixed order unless any upstream declares a concurrency cap.
|
||||
const poolingEnabled = ttsModel.upstreams.some(u => typeof u.maxConcurrency === 'number')
|
||||
|
||||
if (!poolingEnabled) {
|
||||
for (let i = 0; i < ttsModel.upstreams.length; i += 1) {
|
||||
const result = await attemptUpstream(ttsModel.upstreams[i], i)
|
||||
if (result.kind === 'ok')
|
||||
return result.response
|
||||
}
|
||||
}
|
||||
else {
|
||||
const result = await routeTtsAcrossPools(ttsModel.upstreams, req.modelName, attemptUpstream)
|
||||
if (result.kind === 'ok')
|
||||
return result.response
|
||||
}
|
||||
}
|
||||
|
||||
const lastFailure = allFailures.at(-1)
|
||||
|
||||
@@ -77,7 +77,7 @@ function makeLedger(overrides: Partial<ConcurrencyLedger> = {}): ConcurrencyLedg
|
||||
function makeConfigKV(config: RouterConfig | null): ConfigKVService {
|
||||
return {
|
||||
getOptional: vi.fn(async (key: string) => (key === 'LLM_ROUTER_CONFIG' ? config : null)),
|
||||
// routeTts reads UNSPEECH_UPSTREAM once per request via getOrThrow.
|
||||
// routeTts resolves UNSPEECH_UPSTREAM lazily when the chosen adapter needs it.
|
||||
// LLM-side tests never invoke routeTts so the value is irrelevant; TTS
|
||||
// tests need a populated restBaseURL.
|
||||
getOrThrow: vi.fn(async (key: string) => {
|
||||
@@ -691,6 +691,121 @@ describe('createLlmRouterService', () => {
|
||||
expect((configKV.getOptional as ReturnType<typeof vi.fn>).mock.calls.length).toBe(2)
|
||||
})
|
||||
|
||||
describe('route LLM provider groups', () => {
|
||||
function makeGroupedLlmRouter(fetchImpl: typeof fetch) {
|
||||
const { config, crypto } = makeConfig({
|
||||
upstreams: [
|
||||
{ baseURL: 'https://api.stepfun.com/step_plan/v1', keyIds: ['plan-a'] },
|
||||
{ baseURL: 'https://api.stepfun.com/step_plan/v1', keyIds: ['plan-b'] },
|
||||
{ baseURL: 'https://api.stepfun.com/v1', keyIds: ['paygo'] },
|
||||
],
|
||||
})
|
||||
const model = config.llm.models['openai/gpt-5-mini']
|
||||
Object.assign(model.upstreams[0], { id: 'plan-a' })
|
||||
Object.assign(model.upstreams[1], { id: 'plan-b' })
|
||||
Object.assign(model.upstreams[2], { id: 'paygo' })
|
||||
Object.assign(model, {
|
||||
routing: {
|
||||
groups: [
|
||||
{
|
||||
id: 'plan',
|
||||
upstreamIds: ['plan-a', 'plan-b'],
|
||||
retryOn: {
|
||||
httpCodes: [402, 429, 500, 502, 503, 504],
|
||||
onTimeout: true,
|
||||
},
|
||||
continueOn: {
|
||||
httpCodes: [402],
|
||||
onTimeout: false,
|
||||
},
|
||||
},
|
||||
{
|
||||
id: 'paygo',
|
||||
upstreamIds: ['paygo'],
|
||||
retryOn: {
|
||||
httpCodes: [429, 500, 502, 503, 504],
|
||||
onTimeout: true,
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
})
|
||||
|
||||
return createLlmRouterService({
|
||||
configKV: makeConfigKV(config),
|
||||
envelopeCrypto: crypto,
|
||||
gatewayMetrics: makeMetrics(),
|
||||
fetchImpl,
|
||||
redis: makeRedisStub(),
|
||||
concurrencyLedger: makeLedger(),
|
||||
})
|
||||
}
|
||||
|
||||
it('uses the ordinary LLM API only after every Plan account returns 402', async () => {
|
||||
const calledURLs: string[] = []
|
||||
const fetchImpl = vi.fn(async (input: string | URL | Request) => {
|
||||
const url = String(input)
|
||||
calledURLs.push(url)
|
||||
if (url.includes('/step_plan/'))
|
||||
return failResponse(402, { error: { code: 'quota_exceeded' } })
|
||||
return happyResponse({ id: 'completion' })
|
||||
}) as unknown as typeof fetch
|
||||
|
||||
const router = makeGroupedLlmRouter(fetchImpl)
|
||||
const response = await router.route({
|
||||
modelName: 'openai/gpt-5-mini',
|
||||
body: { messages: [] },
|
||||
})
|
||||
|
||||
expect(response.status).toBe(200)
|
||||
expect(calledURLs).toEqual([
|
||||
'https://api.stepfun.com/step_plan/v1/chat/completions',
|
||||
'https://api.stepfun.com/step_plan/v1/chat/completions',
|
||||
'https://api.stepfun.com/v1/chat/completions',
|
||||
])
|
||||
})
|
||||
|
||||
it('does not spend ordinary LLM API balance when the Plan group is rate-limited', async () => {
|
||||
const calledURLs: string[] = []
|
||||
const fetchImpl = vi.fn(async (input: string | URL | Request) => {
|
||||
calledURLs.push(String(input))
|
||||
return failResponse(429)
|
||||
}) as unknown as typeof fetch
|
||||
|
||||
const router = makeGroupedLlmRouter(fetchImpl)
|
||||
|
||||
await expect(router.route({
|
||||
modelName: 'openai/gpt-5-mini',
|
||||
body: { messages: [] },
|
||||
})).rejects.toBeInstanceOf(ApiError)
|
||||
|
||||
expect(calledURLs).toEqual([
|
||||
'https://api.stepfun.com/step_plan/v1/chat/completions',
|
||||
'https://api.stepfun.com/step_plan/v1/chat/completions',
|
||||
])
|
||||
expect(calledURLs).not.toContain('https://api.stepfun.com/v1/chat/completions')
|
||||
})
|
||||
|
||||
it('stops the LLM provider route immediately on Plan authentication failure', async () => {
|
||||
const calledURLs: string[] = []
|
||||
const fetchImpl = vi.fn(async (input: string | URL | Request) => {
|
||||
calledURLs.push(String(input))
|
||||
return failResponse(401)
|
||||
}) as unknown as typeof fetch
|
||||
|
||||
const router = makeGroupedLlmRouter(fetchImpl)
|
||||
|
||||
await expect(router.route({
|
||||
modelName: 'openai/gpt-5-mini',
|
||||
body: { messages: [] },
|
||||
})).rejects.toBeInstanceOf(ApiError)
|
||||
|
||||
expect(calledURLs).toEqual([
|
||||
'https://api.stepfun.com/step_plan/v1/chat/completions',
|
||||
])
|
||||
})
|
||||
})
|
||||
|
||||
// --- routeTts adapter error contract -------------------------------------
|
||||
//
|
||||
// ROOT CAUSE:
|
||||
@@ -899,9 +1014,305 @@ describe('createLlmRouterService', () => {
|
||||
})
|
||||
})
|
||||
|
||||
describe('routeTts provider groups', () => {
|
||||
function endpointProfileFrom(init?: RequestInit): string {
|
||||
const body = JSON.parse(String(init?.body)) as {
|
||||
extra_body?: { endpoint_profile?: string }
|
||||
}
|
||||
return body.extra_body?.endpoint_profile ?? 'default'
|
||||
}
|
||||
|
||||
function makeGroupedStepfunConfig(): { config: RouterConfig, crypto: ReturnType<typeof createEnvelopeCrypto> } {
|
||||
const crypto = createEnvelopeCrypto({ masterKey: freshMasterKey() })
|
||||
const modelName = 'stepfun/stepaudio-2.5-tts'
|
||||
const upstreams = [
|
||||
{
|
||||
id: 'plan-a',
|
||||
baseURL: 'https://api.stepfun.com',
|
||||
keyId: 'plan-key-a',
|
||||
endpointProfile: 'step-plan',
|
||||
},
|
||||
{
|
||||
id: 'plan-b',
|
||||
baseURL: 'https://api.stepfun.com',
|
||||
keyId: 'plan-key-b',
|
||||
endpointProfile: 'step-plan',
|
||||
},
|
||||
{
|
||||
id: 'paygo',
|
||||
baseURL: 'https://api.stepfun.com',
|
||||
keyId: 'paygo-key',
|
||||
endpointProfile: 'default',
|
||||
},
|
||||
].map(upstream => ({
|
||||
id: upstream.id,
|
||||
baseURL: upstream.baseURL,
|
||||
keys: [{
|
||||
id: upstream.keyId,
|
||||
ciphertext: crypto.encryptKey(`sk-${upstream.keyId}`, {
|
||||
modelName,
|
||||
keyEntryId: upstream.keyId,
|
||||
}),
|
||||
}],
|
||||
adapterParams: {
|
||||
endpointProfile: upstream.endpointProfile,
|
||||
model: 'stepaudio-2.5-tts',
|
||||
},
|
||||
}))
|
||||
|
||||
const config = {
|
||||
llm: { models: {} },
|
||||
tts: {
|
||||
models: {
|
||||
[modelName]: {
|
||||
provider: 'stepfun',
|
||||
upstreams,
|
||||
routing: {
|
||||
groups: [
|
||||
{
|
||||
id: 'plan',
|
||||
upstreamIds: ['plan-a', 'plan-b'],
|
||||
strategy: 'ordered',
|
||||
retryOn: {
|
||||
httpCodes: [402, 429, 500, 502, 503, 504],
|
||||
onTimeout: true,
|
||||
},
|
||||
continueOn: {
|
||||
httpCodes: [402],
|
||||
onTimeout: false,
|
||||
},
|
||||
},
|
||||
{
|
||||
id: 'paygo',
|
||||
upstreamIds: ['paygo'],
|
||||
strategy: 'ordered',
|
||||
retryOn: {
|
||||
httpCodes: [429, 500, 502, 503, 504],
|
||||
onTimeout: true,
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
fallbackTriggers: {
|
||||
httpCodes: [401, 402, 429, 500, 502, 503, 504],
|
||||
onTimeout: true,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
defaults: {
|
||||
perAttemptTimeoutMs: 5000,
|
||||
fullChainTimeoutMs: 10000,
|
||||
fallbackHttpCodes: [401, 402, 429, 500, 502, 503, 504],
|
||||
},
|
||||
} as unknown as RouterConfig
|
||||
|
||||
return { config, crypto }
|
||||
}
|
||||
|
||||
function makeGroupedStepfunRouter(
|
||||
fetchImpl: typeof fetch,
|
||||
): ReturnType<typeof createLlmRouterService> {
|
||||
const { config, crypto } = makeGroupedStepfunConfig()
|
||||
return createLlmRouterService({
|
||||
configKV: makeConfigKV(config),
|
||||
envelopeCrypto: crypto,
|
||||
gatewayMetrics: makeMetrics(),
|
||||
fetchImpl,
|
||||
redis: makeRedisStub(),
|
||||
concurrencyLedger: makeLedger(),
|
||||
})
|
||||
}
|
||||
|
||||
it('uses pay-as-you-go only after every Plan account reports quota exhaustion', async () => {
|
||||
const calledProfiles: string[] = []
|
||||
const fetchImpl = vi.fn(async (input: string | URL | Request, init?: RequestInit) => {
|
||||
expect(String(input)).toBe('http://unspeech.local:5933/v1/audio/speech')
|
||||
const profile = endpointProfileFrom(init)
|
||||
calledProfiles.push(profile)
|
||||
if (profile === 'step-plan')
|
||||
return failResponse(402, { error: { code: 'quota_exceeded' } })
|
||||
return new Response(new Uint8Array([0x01]), {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'audio/mpeg' },
|
||||
})
|
||||
}) as unknown as typeof fetch
|
||||
|
||||
const router = makeGroupedStepfunRouter(fetchImpl)
|
||||
const response = await router.routeTts({
|
||||
modelName: 'stepfun/stepaudio-2.5-tts',
|
||||
input: { text: '你好' },
|
||||
})
|
||||
|
||||
expect(response.status).toBe(200)
|
||||
expect(calledProfiles).toEqual(['step-plan', 'step-plan', 'default'])
|
||||
})
|
||||
|
||||
it('stays inside the Plan group when a Plan account succeeds', async () => {
|
||||
const calledProfiles: string[] = []
|
||||
const fetchImpl = vi.fn(async (input: string | URL | Request, init?: RequestInit) => {
|
||||
expect(String(input)).toBe('http://unspeech.local:5933/v1/audio/speech')
|
||||
calledProfiles.push(endpointProfileFrom(init))
|
||||
return new Response(new Uint8Array([0x01]), {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'audio/mpeg' },
|
||||
})
|
||||
}) as unknown as typeof fetch
|
||||
|
||||
const router = makeGroupedStepfunRouter(fetchImpl)
|
||||
const response = await router.routeTts({
|
||||
modelName: 'stepfun/stepaudio-2.5-tts',
|
||||
input: { text: '你好' },
|
||||
})
|
||||
|
||||
expect(response.status).toBe(200)
|
||||
expect(calledProfiles).toEqual(['step-plan'])
|
||||
})
|
||||
|
||||
it('requires UNSPEECH_UPSTREAM for an endpoint-profile request', async () => {
|
||||
const { config, crypto } = makeGroupedStepfunConfig()
|
||||
const configKV = makeConfigKV(config)
|
||||
const getOrThrow = vi.fn(async () => {
|
||||
throw new Error('UNSPEECH_UPSTREAM not configured')
|
||||
})
|
||||
Object.assign(configKV, { getOrThrow })
|
||||
const fetchImpl = vi.fn(async () => new Response(new Uint8Array([0x01]), {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'audio/mpeg' },
|
||||
})) as unknown as typeof fetch
|
||||
const router = createLlmRouterService({
|
||||
configKV,
|
||||
envelopeCrypto: crypto,
|
||||
gatewayMetrics: makeMetrics(),
|
||||
fetchImpl,
|
||||
redis: makeRedisStub(),
|
||||
concurrencyLedger: makeLedger(),
|
||||
})
|
||||
|
||||
await expect(router.routeTts({
|
||||
modelName: 'stepfun/stepaudio-2.5-tts',
|
||||
input: { text: '你好' },
|
||||
})).rejects.toThrow('UNSPEECH_UPSTREAM not configured')
|
||||
|
||||
expect(fetchImpl).not.toHaveBeenCalled()
|
||||
expect(getOrThrow).toHaveBeenCalledWith('UNSPEECH_UPSTREAM')
|
||||
})
|
||||
|
||||
it('requires UNSPEECH_UPSTREAM for the StepFun voice catalog', async () => {
|
||||
const { config, crypto } = makeGroupedStepfunConfig()
|
||||
const configKV = makeConfigKV(config)
|
||||
const getOrThrow = vi.fn(async () => {
|
||||
throw new Error('UNSPEECH_UPSTREAM not configured')
|
||||
})
|
||||
Object.assign(configKV, { getOrThrow })
|
||||
const fetchImpl = vi.fn() as unknown as typeof fetch
|
||||
const router = createLlmRouterService({
|
||||
configKV,
|
||||
envelopeCrypto: crypto,
|
||||
gatewayMetrics: makeMetrics(),
|
||||
fetchImpl,
|
||||
redis: makeRedisStub(),
|
||||
concurrencyLedger: makeLedger(),
|
||||
})
|
||||
|
||||
await expect(
|
||||
router.listTtsVoices('stepfun/stepaudio-2.5-tts'),
|
||||
).rejects.toThrow('UNSPEECH_UPSTREAM not configured')
|
||||
|
||||
expect(fetchImpl).not.toHaveBeenCalled()
|
||||
expect(getOrThrow).toHaveBeenCalledWith('UNSPEECH_UPSTREAM')
|
||||
})
|
||||
|
||||
it('keeps a Step Plan attempt timeout distinct from HTTP 500 at the paid boundary', async () => {
|
||||
const { config, crypto } = makeGroupedStepfunConfig()
|
||||
config.defaults!.perAttemptTimeoutMs = 10
|
||||
const model = config.tts.models['stepfun/stepaudio-2.5-tts']
|
||||
model.routing!.groups[0].continueOn = {
|
||||
httpCodes: [500],
|
||||
onTimeout: false,
|
||||
}
|
||||
const calledProfiles: string[] = []
|
||||
const fetchImpl = vi.fn((input: string | URL | Request, init?: RequestInit) => {
|
||||
expect(String(input)).toBe('http://unspeech.local:5933/v1/audio/speech')
|
||||
const profile = endpointProfileFrom(init)
|
||||
calledProfiles.push(profile)
|
||||
if (profile !== 'step-plan') {
|
||||
return Promise.resolve(new Response(new Uint8Array([0x01]), {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'audio/mpeg' },
|
||||
}))
|
||||
}
|
||||
|
||||
return new Promise<Response>((_, reject) => {
|
||||
const rejectAbort = () => reject(init?.signal?.reason ?? new Error('aborted'))
|
||||
if (init?.signal?.aborted)
|
||||
rejectAbort()
|
||||
else
|
||||
init?.signal?.addEventListener('abort', rejectAbort, { once: true })
|
||||
})
|
||||
}) as unknown as typeof fetch
|
||||
const router = createLlmRouterService({
|
||||
configKV: makeConfigKV(config),
|
||||
envelopeCrypto: crypto,
|
||||
gatewayMetrics: makeMetrics(),
|
||||
fetchImpl,
|
||||
redis: makeRedisStub(),
|
||||
concurrencyLedger: makeLedger(),
|
||||
})
|
||||
|
||||
await expect(router.routeTts({
|
||||
modelName: 'stepfun/stepaudio-2.5-tts',
|
||||
input: { text: '你好' },
|
||||
})).rejects.toMatchObject({
|
||||
statusCode: 504,
|
||||
details: expect.objectContaining({ lastStatusCode: 'timeout' }),
|
||||
})
|
||||
|
||||
expect(calledProfiles).toEqual(['step-plan', 'step-plan'])
|
||||
expect(calledProfiles).not.toContain('default')
|
||||
})
|
||||
|
||||
it('does not cross the paid boundary when the Plan group is rate-limited', async () => {
|
||||
const calledProfiles: string[] = []
|
||||
const fetchImpl = vi.fn(async (input: string | URL | Request, init?: RequestInit) => {
|
||||
expect(String(input)).toBe('http://unspeech.local:5933/v1/audio/speech')
|
||||
calledProfiles.push(endpointProfileFrom(init))
|
||||
return failResponse(429)
|
||||
}) as unknown as typeof fetch
|
||||
|
||||
const router = makeGroupedStepfunRouter(fetchImpl)
|
||||
|
||||
await expect(router.routeTts({
|
||||
modelName: 'stepfun/stepaudio-2.5-tts',
|
||||
input: { text: '你好' },
|
||||
})).rejects.toBeInstanceOf(ApiError)
|
||||
|
||||
expect(calledProfiles).toEqual(['step-plan', 'step-plan'])
|
||||
expect(calledProfiles).not.toContain('default')
|
||||
})
|
||||
|
||||
it('stops the provider route immediately on a Plan authentication failure', async () => {
|
||||
const calledProfiles: string[] = []
|
||||
const fetchImpl = vi.fn(async (input: string | URL | Request, init?: RequestInit) => {
|
||||
expect(String(input)).toBe('http://unspeech.local:5933/v1/audio/speech')
|
||||
calledProfiles.push(endpointProfileFrom(init))
|
||||
return failResponse(401)
|
||||
}) as unknown as typeof fetch
|
||||
|
||||
const router = makeGroupedStepfunRouter(fetchImpl)
|
||||
|
||||
await expect(router.routeTts({
|
||||
modelName: 'stepfun/stepaudio-2.5-tts',
|
||||
input: { text: '你好' },
|
||||
})).rejects.toBeInstanceOf(ApiError)
|
||||
|
||||
expect(calledProfiles).toEqual(['step-plan'])
|
||||
})
|
||||
})
|
||||
|
||||
describe('routeTtspool capacity-aware routing', () => {
|
||||
// One app_id == one upstream (Volcengine `adapterParams.appid`), each capped
|
||||
// at `maxConcurrency`. The router spreads load least-loaded-first across pools
|
||||
// at `maxConcurrency`. The router spreads load least-inflight-first across pools
|
||||
// and circuit-breaks a pool on 429 (app_id concurrency exceeded upstream-side).
|
||||
function makePoolConfig(
|
||||
upstreams: Array<{ baseURL: string, appid: string, maxConcurrency?: number }>,
|
||||
@@ -934,7 +1345,7 @@ describe('createLlmRouterService', () => {
|
||||
return { config, crypto }
|
||||
}
|
||||
|
||||
// Stateful in-memory ledger so least-loaded ordering and capacity gating are
|
||||
// Stateful in-memory ledger so least-inflight ordering and capacity gating are
|
||||
// observable. `seed` pre-loads inflight counts to drive deterministic ranking.
|
||||
function makeStatefulLedger(seed: Record<string, number> = {}, saturatedSeed: string[] = []) {
|
||||
const inflight = new Map<string, number>(Object.entries(seed))
|
||||
@@ -974,7 +1385,7 @@ describe('createLlmRouterService', () => {
|
||||
})
|
||||
}
|
||||
|
||||
it('routes to the least-loadedpool (covers AE1 — load spread, not first-fill)', async () => {
|
||||
it('routes to the least-inflight pool (covers AE1 — load spread, not first-fill)', async () => {
|
||||
// @example two app_ids cap 10, seeded 8 vs 2 in-flight -> the new request
|
||||
// goes to the freer pool (app-2), not the config-first pool (app-1).
|
||||
const { config, crypto } = makePoolConfig([
|
||||
@@ -992,6 +1403,208 @@ describe('createLlmRouterService', () => {
|
||||
expect(tryAcquire.mock.calls[0][0]).toBe('app-2')
|
||||
})
|
||||
|
||||
it('ranks least-inflight accounts by current usage when concurrency caps differ', async () => {
|
||||
const { config, crypto } = makePoolConfig([
|
||||
{ baseURL: 'https://up-a.example', appid: 'app-1', maxConcurrency: 100 },
|
||||
{ baseURL: 'https://up-b.example', appid: 'app-2', maxConcurrency: 10 },
|
||||
])
|
||||
const { ledger, tryAcquire } = makeStatefulLedger({ 'app-1': 50, 'app-2': 0 })
|
||||
const fetchImpl = vi.fn(async () => happyResponse({ ok: 1 })) as unknown as typeof fetch
|
||||
|
||||
const router = makePoolRouter(config, crypto, ledger, fetchImpl)
|
||||
const response = await router.routeTts({ modelName: 'tts-pool', input: { text: 'hi' } })
|
||||
|
||||
expect(response.status).toBe(200)
|
||||
expect(tryAcquire).toHaveBeenCalledTimes(1)
|
||||
expect(tryAcquire).toHaveBeenCalledWith('app-2', 10)
|
||||
})
|
||||
|
||||
it('namespaces a non-appid pool by model and upstream id', async () => {
|
||||
const crypto = createEnvelopeCrypto({ masterKey: freshMasterKey() })
|
||||
const modelName = 'stepfun/stepaudio-2.5-tts'
|
||||
const keyEntryId = 'plan-key'
|
||||
const config = {
|
||||
llm: { models: {} },
|
||||
tts: {
|
||||
models: {
|
||||
[modelName]: {
|
||||
provider: 'stepfun',
|
||||
upstreams: [{
|
||||
id: 'plan',
|
||||
baseURL: 'https://api.stepfun.com',
|
||||
keys: [{
|
||||
id: keyEntryId,
|
||||
ciphertext: crypto.encryptKey('sk-plan', { modelName, keyEntryId }),
|
||||
}],
|
||||
adapterParams: {
|
||||
endpointProfile: 'step-plan',
|
||||
model: 'stepaudio-2.5-tts',
|
||||
},
|
||||
maxConcurrency: 1,
|
||||
}],
|
||||
routing: {
|
||||
groups: [{
|
||||
id: 'plan',
|
||||
upstreamIds: ['plan'],
|
||||
strategy: 'least-inflight',
|
||||
retryOn: { httpCodes: [402, 429, 500, 502, 503, 504], onTimeout: true },
|
||||
}],
|
||||
},
|
||||
fallbackTriggers: { httpCodes: [402, 429, 500, 502, 503, 504], onTimeout: true },
|
||||
},
|
||||
},
|
||||
},
|
||||
defaults: {
|
||||
perAttemptTimeoutMs: 5000,
|
||||
fullChainTimeoutMs: 10000,
|
||||
fallbackHttpCodes: [402, 429, 500, 502, 503, 504],
|
||||
},
|
||||
} as RouterConfig
|
||||
const { ledger, tryAcquire } = makeStatefulLedger()
|
||||
const fetchImpl = vi.fn(async () => new Response(new Uint8Array([0x01]), {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'audio/mpeg' },
|
||||
})) as unknown as typeof fetch
|
||||
|
||||
const router = makePoolRouter(config, crypto, ledger, fetchImpl)
|
||||
const response = await router.routeTts({ modelName, input: { text: 'hi' } })
|
||||
|
||||
expect(response.status).toBe(200)
|
||||
expect(tryAcquire).toHaveBeenCalledWith(
|
||||
'model:["stepfun/stepaudio-2.5-tts","id","plan"]',
|
||||
1,
|
||||
)
|
||||
})
|
||||
|
||||
it('enforces maxConcurrency for an ordered provider group', async () => {
|
||||
const { config, crypto } = makePoolConfig([
|
||||
{ baseURL: 'https://up-a.example', appid: 'app-1', maxConcurrency: 10 },
|
||||
])
|
||||
const model = config.tts.models['tts-pool']
|
||||
Object.assign(model.upstreams[0], { id: 'primary' })
|
||||
Object.assign(model, {
|
||||
routing: {
|
||||
groups: [{
|
||||
id: 'primary',
|
||||
upstreamIds: ['primary'],
|
||||
strategy: 'ordered',
|
||||
retryOn: { httpCodes: [429, 500, 502, 503, 504], onTimeout: true },
|
||||
}],
|
||||
},
|
||||
})
|
||||
const { ledger, tryAcquire, release } = makeStatefulLedger()
|
||||
const fetchImpl = vi.fn(async () => happyResponse({ ok: 1 })) as unknown as typeof fetch
|
||||
|
||||
const router = makePoolRouter(config, crypto, ledger, fetchImpl)
|
||||
const response = await router.routeTts({ modelName: 'tts-pool', input: { text: 'hi' } })
|
||||
|
||||
expect(response.status).toBe(200)
|
||||
expect(tryAcquire).toHaveBeenCalledWith('app-1', 10)
|
||||
expect(release).toHaveBeenCalledWith('app-1')
|
||||
})
|
||||
|
||||
it('keeps least-inflight selection inside the active group before considering pay-as-you-go', async () => {
|
||||
const { config, crypto } = makePoolConfig([
|
||||
{ baseURL: 'https://plan-a.example', appid: 'plan-a', maxConcurrency: 10 },
|
||||
{ baseURL: 'https://plan-b.example', appid: 'plan-b', maxConcurrency: 10 },
|
||||
{ baseURL: 'https://paygo.example', appid: 'paygo' },
|
||||
])
|
||||
const model = config.tts.models['tts-pool']
|
||||
Object.assign(model.upstreams[0], { id: 'plan-a' })
|
||||
Object.assign(model.upstreams[1], { id: 'plan-b' })
|
||||
Object.assign(model.upstreams[2], { id: 'paygo' })
|
||||
Object.assign(model, {
|
||||
routing: {
|
||||
groups: [
|
||||
{
|
||||
id: 'plan',
|
||||
upstreamIds: ['plan-a', 'plan-b'],
|
||||
strategy: 'least-inflight',
|
||||
retryOn: { httpCodes: [429, 500, 502, 503, 504], onTimeout: true },
|
||||
continueOn: { httpCodes: [402], onTimeout: false },
|
||||
},
|
||||
{
|
||||
id: 'paygo',
|
||||
upstreamIds: ['paygo'],
|
||||
strategy: 'ordered',
|
||||
retryOn: { httpCodes: [429, 500, 502, 503, 504], onTimeout: true },
|
||||
},
|
||||
],
|
||||
},
|
||||
})
|
||||
const { ledger, tryAcquire } = makeStatefulLedger({ 'plan-a': 8, 'plan-b': 2 })
|
||||
const selectedAppIds: string[] = []
|
||||
const fetchImpl = vi.fn(async (_input: string | URL | Request, init?: RequestInit) => {
|
||||
const body = JSON.parse(String(init?.body)) as { extra_body?: { app?: { appid?: string } } }
|
||||
selectedAppIds.push(body.extra_body?.app?.appid ?? 'unknown')
|
||||
return new Response(new Uint8Array([0x01]), {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'audio/mpeg' },
|
||||
})
|
||||
}) as unknown as typeof fetch
|
||||
|
||||
const router = makePoolRouter(config, crypto, ledger, fetchImpl)
|
||||
const response = await router.routeTts({ modelName: 'tts-pool', input: { text: 'hi' } })
|
||||
|
||||
expect(response.status).toBe(200)
|
||||
expect(selectedAppIds).toEqual(['plan-b'])
|
||||
expect(tryAcquire).toHaveBeenCalledTimes(1)
|
||||
expect(tryAcquire).toHaveBeenCalledWith('plan-b', 10)
|
||||
})
|
||||
|
||||
it('does not cross groups when a Plan account was skipped at its concurrency limit', async () => {
|
||||
const { config, crypto } = makePoolConfig([
|
||||
{ baseURL: 'https://plan-a.example', appid: 'plan-a', maxConcurrency: 10 },
|
||||
{ baseURL: 'https://plan-b.example', appid: 'plan-b', maxConcurrency: 10 },
|
||||
{ baseURL: 'https://paygo.example', appid: 'paygo' },
|
||||
])
|
||||
const model = config.tts.models['tts-pool']
|
||||
Object.assign(model.upstreams[0], { id: 'plan-a' })
|
||||
Object.assign(model.upstreams[1], { id: 'plan-b' })
|
||||
Object.assign(model.upstreams[2], { id: 'paygo' })
|
||||
Object.assign(model, {
|
||||
routing: {
|
||||
groups: [
|
||||
{
|
||||
id: 'plan',
|
||||
upstreamIds: ['plan-a', 'plan-b'],
|
||||
strategy: 'least-inflight',
|
||||
retryOn: { httpCodes: [402, 429, 500, 502, 503, 504], onTimeout: true },
|
||||
continueOn: { httpCodes: [402], onTimeout: false },
|
||||
},
|
||||
{
|
||||
id: 'paygo',
|
||||
upstreamIds: ['paygo'],
|
||||
strategy: 'ordered',
|
||||
retryOn: { httpCodes: [429, 500, 502, 503, 504], onTimeout: true },
|
||||
},
|
||||
],
|
||||
},
|
||||
})
|
||||
const { ledger } = makeStatefulLedger({ 'plan-a': 10, 'plan-b': 0 })
|
||||
const selectedAppIds: string[] = []
|
||||
const fetchImpl = vi.fn(async (_input: string | URL | Request, init?: RequestInit) => {
|
||||
const body = JSON.parse(String(init?.body)) as { extra_body?: { app?: { appid?: string } } }
|
||||
const appid = body.extra_body?.app?.appid ?? 'unknown'
|
||||
selectedAppIds.push(appid)
|
||||
if (appid === 'plan-b')
|
||||
return failResponse(402, { error: { code: 'quota_exceeded' } })
|
||||
return new Response(new Uint8Array([0x01]), {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'audio/mpeg' },
|
||||
})
|
||||
}) as unknown as typeof fetch
|
||||
|
||||
const router = makePoolRouter(config, crypto, ledger, fetchImpl)
|
||||
|
||||
await expect(router.routeTts({
|
||||
modelName: 'tts-pool',
|
||||
input: { text: 'hi' },
|
||||
})).rejects.toBeInstanceOf(ApiError)
|
||||
|
||||
expect(selectedAppIds).toEqual(['plan-b'])
|
||||
})
|
||||
|
||||
it('skips a fullpool and dispatches to one with capacity', async () => {
|
||||
// @example app-1 at cap (10/10) -> filtered out; app-2 (0/10) serves.
|
||||
const { config, crypto } = makePoolConfig([
|
||||
|
||||
@@ -14,8 +14,13 @@ import type {
|
||||
llmModelSchema,
|
||||
llmRouterConfigSchema,
|
||||
llmRouterDefaultsSchema,
|
||||
llmRoutingGroupSchema,
|
||||
llmRoutingSchema,
|
||||
llmUpstreamSchema,
|
||||
routeFailureTriggersSchema,
|
||||
ttsModelSchema,
|
||||
ttsRoutingGroupSchema,
|
||||
ttsRoutingSchema,
|
||||
ttsUpstreamSchema,
|
||||
} from '../../adapters/config-kv'
|
||||
|
||||
@@ -31,25 +36,45 @@ export type RouterConfig = InferOutput<typeof llmRouterConfigSchema>
|
||||
export type RouterDefaults = InferOutput<typeof llmRouterDefaultsSchema>
|
||||
|
||||
/**
|
||||
* LLM upstream — one provider endpoint with its ordered key list.
|
||||
* LLM upstream — one candidate endpoint with its ordered key list.
|
||||
*/
|
||||
export type LlmUpstream = InferOutput<typeof llmUpstreamSchema>
|
||||
|
||||
/**
|
||||
* LLM model entry — ordered list of upstreams to try in fallback order.
|
||||
* LLM model entry — upstream candidates plus an optional grouped route.
|
||||
*/
|
||||
export type LlmModel = InferOutput<typeof llmModelSchema>
|
||||
|
||||
/**
|
||||
* TTS upstream — one provider endpoint with adapter params + key list.
|
||||
* LLM route composed from ordered candidate groups.
|
||||
*/
|
||||
export type LlmRouting = InferOutput<typeof llmRoutingSchema>
|
||||
|
||||
/**
|
||||
* One ordered group of interchangeable LLM candidates.
|
||||
*/
|
||||
export type LlmRoutingGroup = InferOutput<typeof llmRoutingGroupSchema>
|
||||
|
||||
/**
|
||||
* TTS upstream — one candidate endpoint with adapter params + key list.
|
||||
*/
|
||||
export type TtsUpstream = InferOutput<typeof ttsUpstreamSchema>
|
||||
|
||||
/**
|
||||
* TTS model entry — provider tag + ordered upstreams.
|
||||
* TTS model entry — provider tag, upstream candidates, and optional grouped route.
|
||||
*/
|
||||
export type TtsModel = InferOutput<typeof ttsModelSchema>
|
||||
|
||||
/**
|
||||
* TTS route composed from ordered candidate groups.
|
||||
*/
|
||||
export type TtsRouting = InferOutput<typeof ttsRoutingSchema>
|
||||
|
||||
/**
|
||||
* One group of interchangeable TTS candidates.
|
||||
*/
|
||||
export type TtsRoutingGroup = InferOutput<typeof ttsRoutingGroupSchema>
|
||||
|
||||
/**
|
||||
* ASR model entry — provider tag + ordered upstreams for realtime transcription.
|
||||
*/
|
||||
@@ -66,6 +91,11 @@ export type AsrUpstream = InferOutput<typeof asrUpstreamSchema>
|
||||
*/
|
||||
export type FallbackTriggers = InferOutput<typeof fallbackTriggersSchema>
|
||||
|
||||
/**
|
||||
* Failure allow-list that authorizes a routing transition.
|
||||
*/
|
||||
export type RouteFailureTriggers = InferOutput<typeof routeFailureTriggersSchema>
|
||||
|
||||
/**
|
||||
* One entry in `upstream.keys`: stable id + at-rest envelope ciphertext.
|
||||
* The plaintext key is only produced lazily by the key-rotator at call time.
|
||||
|
||||
@@ -46,9 +46,10 @@ export function lockRedisKey(domain: string, ...identifiers: RedisKeyPart[]): st
|
||||
|
||||
/**
|
||||
* In-flight request counter for one TTSpool (per app_id concurrency pool).
|
||||
* `poolId` is the upstream's `adapterParams.appid` (or baseURL fallback). The
|
||||
* counter is INCR'd on slot acquire and DECR'd on release; a short TTL bounds
|
||||
* leakage if a replica crashes between acquire and release.
|
||||
* `poolId` is the upstream's global `adapterParams.appid` or a model-scoped
|
||||
* upstream identity for providers without app ids. The counter is INCR'd on
|
||||
* slot acquire and DECR'd on release; a short TTL bounds leakage if a replica
|
||||
* crashes between acquire and release.
|
||||
*/
|
||||
export function ttsPoolInflightRedisKey(poolId: string): string {
|
||||
return redisKeyFrom('tts', 'pool', 'inflight', poolId)
|
||||
|
||||
Reference in New Issue
Block a user