openvibers/openvibe.ai Commits
Provider circuit breaker ignores caller-caused 4xx: a few malformed requests from one caller (bad image, oversized max_tokens, invalid schema) no longer open the circuit and take the shared provider away from every other caller
Changes
@@@ -23,6 +23,12 @@ const { createAnthropicProvider } = require('./anthropic');const { createHttpSeamProvider } = require('./http');const { createWhisperProvider } = require('./whisper');++/** 4xx that the request itself caused (bad image, schema, max_tokens...), as opposed to auth, rate or availability. */++function callerFault(err) {++ const st = err && err.status;++ return Number.isInteger(st) && st >= 400 && st < 500 && ![401, 403, 404, 408, 429].includes(st);++}++function createProviderPool({ db, registry, config, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, env = process.env, log = console }) {const cache = new Map(); // key -> { stamp, adapter }const stats = {}; // key -> { calls, failures, ... } (in memory; tests + /api/ready)@@@ -186,7 +192,9 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.nobreak;}}-- recordFailure(c.provider, lastErr);++ // A request the provider rejected as malformed is the caller's fault, not a sign the++ // provider is unhealthy: it must not open the circuit other callers depend on.++ if (!callerFault(lastErr)) recordFailure(c.provider, lastErr);tried.push({ provider: c.provider, error: String(lastErr && lastErr.message || lastErr).slice(0, 200) });log.warn(`[ai] ${operation} via ${c.provider}/${model || '-'} failed: ${lastErr && lastErr.message}`);}nnew file mode 100644@@@ -0,0 +1,48 @@++'use strict';++// The circuit breaker tracks PROVIDER health. A request the provider rejects as malformed (HTTP 400,++// 413, 422: a bad image, an oversized max_tokens, an invalid schema) is the caller's fault, so it++// must not count toward opening the circuit: otherwise any one caller can take a shared provider++// away from every other caller by sending a few bad requests in a row.++++const assert = require('assert');++const { boot, request, token, suite, seamServer, ALL } = require('./helpers');++++const t = suite('breaker');++const admin = token('live', ALL);++const attacker = token('news', ['ai.run.create', 'ai.run.read']);++let h;++let seam;++++t.test('boot: one seam provider that rejects "BAD" prompts with HTTP 400', async () => {++ seam = await seamServer((body) => {++ const text = JSON.stringify(body.messages || []);++ if (text.includes('BAD')) return { status: 400, body: { error: { message: 'invalid request: image could not be decoded' } } };++ return { text: 'fine', model: 'seam-model', usage: { input: 3, output: 1 } };++ });++ h = await boot({ env: { AI_STUB_FALLBACK: 'false', AI_BREAKER_FAILURES: '2', AI_BREAKER_COOLDOWN_MS: '600000' } });++ const p = await request(h.base, 'POST', '/api/v1/providers', { tok: admin, body: { key: 'seam', kind: 'http', auth_mode: 'none', base_url: seam.url, capabilities: ['chat', 'generate', 'json'], timeout_ms: 5000 } });++ assert.strictEqual(p.status, 201, p.text);++ const r = await request(h.base, 'POST', '/api/v1/routes/default.chat/versions', { tok: admin, body: { primary: { provider: 'seam' }, fallbacks: [], timeout_ms: 5000 } });++ assert.strictEqual(r.status, 201, r.text);++});++++t.test('bad requests from one caller fail that caller only; the circuit stays closed for everyone', async () => {++ for (let i = 0; i < 4; i++) {++ const bad = await request(h.base, 'POST', '/api/v1/generate', { tok: attacker, body: { prompt: `BAD ${i}`, options: { cache: false } } });++ assert.strictEqual(bad.body.run.status, 'failed', bad.text);++ }++ assert.strictEqual(h.pool.health('seam').state, 'closed', 'client errors must not open the circuit');++ const good = await request(h.base, 'POST', '/api/v1/generate', { tok: admin, body: { prompt: 'hello', options: { cache: false } } });++ assert.strictEqual(good.body.run.status, 'succeeded', good.text);++ assert.strictEqual(good.body.run.output.text, 'fine');++});++++t.test('server errors still open it', async () => {++ seam.handler = () => ({ status: 503, body: { error: { message: 'overloaded' } } });++ for (let i = 0; i < 2; i++) await request(h.base, 'POST', '/api/v1/generate', { tok: admin, body: { prompt: `x ${i}`, options: { cache: false } } });++ assert.strictEqual(h.pool.health('seam').state, 'open');++});++++t.test('shutdown', async () => { await h.stop(); await seam.close(); });++++t.run();
@@ -23,6 +23,12 @@ const { createAnthropicProvider } = require('./anthropic');
@@ -23,6 +23,12 @@ const { createAnthropicProvider } = require('./anthropic');
const { createHttpSeamProvider } = require('./http');
const { createHttpSeamProvider } = require('./http');
const { createWhisperProvider } = require('./whisper');
const { createWhisperProvider } = require('./whisper');
+/** 4xx that the request itself caused (bad image, schema, max_tokens...), as opposed to auth, rate or availability. */
+function callerFault(err) {
+ const st = err && err.status;
+ return Number.isInteger(st) && st >= 400 && st < 500 && ![401, 403, 404, 408, 429].includes(st);
+}
+
function createProviderPool({ db, registry, config, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, env = process.env, log = console }) {
function createProviderPool({ db, registry, config, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, env = process.env, log = console }) {
const cache = new Map(); // key -> { stamp, adapter }
const cache = new Map(); // key -> { stamp, adapter }
const stats = {}; // key -> { calls, failures, ... } (in memory; tests + /api/ready)
const stats = {}; // key -> { calls, failures, ... } (in memory; tests + /api/ready)
@@ -186,7 +192,9 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
@@ -186,7 +192,9 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
break;
break;
}
}
}
}
- recordFailure(c.provider, lastErr);
+ // A request the provider rejected as malformed is the caller's fault, not a sign the
+ // provider is unhealthy: it must not open the circuit other callers depend on.
+ if (!callerFault(lastErr)) recordFailure(c.provider, lastErr);
tried.push({ provider: c.provider, error: String(lastErr && lastErr.message || lastErr).slice(0, 200) });
tried.push({ provider: c.provider, error: String(lastErr && lastErr.message || lastErr).slice(0, 200) });
log.warn(`[ai] ${operation} via ${c.provider}/${model || '-'} failed: ${lastErr && lastErr.message}`);
log.warn(`[ai] ${operation} via ${c.provider}/${model || '-'} failed: ${lastErr && lastErr.message}`);
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,48 @@
@@ -0,0 +1,48 @@
+'use strict';
+// The circuit breaker tracks PROVIDER health. A request the provider rejects as malformed (HTTP 400,
+// 413, 422: a bad image, an oversized max_tokens, an invalid schema) is the caller's fault, so it
+// must not count toward opening the circuit: otherwise any one caller can take a shared provider
+// away from every other caller by sending a few bad requests in a row.
+
+const assert = require('assert');
+const { boot, request, token, suite, seamServer, ALL } = require('./helpers');
+
+const t = suite('breaker');
+const admin = token('live', ALL);
+const attacker = token('news', ['ai.run.create', 'ai.run.read']);
+let h;
+let seam;
+
+t.test('boot: one seam provider that rejects "BAD" prompts with HTTP 400', async () => {
+ seam = await seamServer((body) => {
+ const text = JSON.stringify(body.messages || []);
+ if (text.includes('BAD')) return { status: 400, body: { error: { message: 'invalid request: image could not be decoded' } } };
+ return { text: 'fine', model: 'seam-model', usage: { input: 3, output: 1 } };
+ });
+ h = await boot({ env: { AI_STUB_FALLBACK: 'false', AI_BREAKER_FAILURES: '2', AI_BREAKER_COOLDOWN_MS: '600000' } });
+ const p = await request(h.base, 'POST', '/api/v1/providers', { tok: admin, body: { key: 'seam', kind: 'http', auth_mode: 'none', base_url: seam.url, capabilities: ['chat', 'generate', 'json'], timeout_ms: 5000 } });
+ assert.strictEqual(p.status, 201, p.text);
+ const r = await request(h.base, 'POST', '/api/v1/routes/default.chat/versions', { tok: admin, body: { primary: { provider: 'seam' }, fallbacks: [], timeout_ms: 5000 } });
+ assert.strictEqual(r.status, 201, r.text);
+});
+
+t.test('bad requests from one caller fail that caller only; the circuit stays closed for everyone', async () => {
+ for (let i = 0; i < 4; i++) {
+ const bad = await request(h.base, 'POST', '/api/v1/generate', { tok: attacker, body: { prompt: `BAD ${i}`, options: { cache: false } } });
+ assert.strictEqual(bad.body.run.status, 'failed', bad.text);
+ }
+ assert.strictEqual(h.pool.health('seam').state, 'closed', 'client errors must not open the circuit');
+ const good = await request(h.base, 'POST', '/api/v1/generate', { tok: admin, body: { prompt: 'hello', options: { cache: false } } });
+ assert.strictEqual(good.body.run.status, 'succeeded', good.text);
+ assert.strictEqual(good.body.run.output.text, 'fine');
+});
+
+t.test('server errors still open it', async () => {
+ seam.handler = () => ({ status: 503, body: { error: { message: 'overloaded' } } });
+ for (let i = 0; i < 2; i++) await request(h.base, 'POST', '/api/v1/generate', { tok: admin, body: { prompt: `x ${i}`, options: { cache: false } } });
+ assert.strictEqual(h.pool.health('seam').state, 'open');
+});
+
+t.test('shutdown', async () => { await h.stop(); await seam.close(); });
+
+t.run();