enne2/pi-qmem
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf(fallback): connect timeout breve + circuit breaker persistente (fast-fail)

Browse Source 
Il gateway irraggiungibile costava ~30s x4 tentativi (fino a ~2 minuti) per ogni
chiamata: ora si distinguono i due casi e le chiamate successive sono immediate.

- due timeout separati: `connectTimeoutMs` (default 2500, connect+headers) e
  `timeoutMs` (default 30000, budget per il body). Nessuna risposta entro il
  primo = "gateway non raggiungibile"; body lento = "elaborazione lunga"
- circuit breaker persistente in ~/.local/share/pi-qmem/breaker.json
  (env QMEM_BREAKER_FILE): fallimento definitivo (connessione rifiutata/DNS/
  connect timeout) → nessun retry e apertura immediata per `breakerBaseMs`
  (default 120000 = 2 min) con escalation fino a `breakerMaxMs` (10 min);
  5xx/body lento sono ambigui → retry con Retry-After e apertura dopo
  `breakerTripAfter` (default 2). Un successo lo richiude; cambiando `url` lo
  stato riparte chiuso (endpoint-aware)
- con breaker aperto gatewayRequest ritorna in ~0 ms senza rete
  (`gateway_unreachable`, `breaker_open`, `retry_in_ms`): i tool passano subito
  al fallback locale e l'outbox accoda
- fix di due bug scoperti durante i test:
  * `res.json().catch(() => ({}))` trasformava un body non completato in
    "successo con dati vuoti" → l'agente vedeva "nessun risultato" invece del
    fallback locale. Ora è `timeout_body` (fallimento, ambiguo)
  * `submitOrQueue` passava un AbortSignal esterno, che con la nuova semantica
    sarebbe stato letto come annullamento utente (eccezione invece di coda)
- messaggi dei tool con lo stato del breaker e come forzare un tentativo;
  `details.breaker` per l'osservabilità
- comandi: `/qmem:local breaker [reset]` e `qmem-sqlite breaker [--reset]`;
  lo stato compare in `/qmem:local status` e nella CLI
- budget interni per enrich/pull/flush (niente AbortSignal esterni)

Misure: connessione rifiutata → 4-8 ms (prima: 4 x 30 s); front che risponde
503 dopo ~40 s → 3,5 s alla prima chiamata, poi 0 ms di rete a breaker aperto;
server che accetta e non risponde → 708 ms (connect timeout); body lento →
1,2 s senza aprire il breaker; persistenza verificata fra processi distinti.

Test: scripts/test-local.mjs 38/38 (nuova fase dedicata al breaker).
This commit is contained in:
Matteo Benedetto
2026-09-13 18:09:50 +02:00
parent d1985514e1
commit 9dd24298cc
11 changed files with 475 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/qmem-sqlite.mjs
+23 -2
View File
@@ -10,6 +10,7 @@
* node scripts/qmem-sqlite.mjs store --project P [--kind K] [--text "..."] [--queue-only]
* node scripts/qmem-sqlite.mjs queue [--status queued|synced|duplicate|failed]
* node scripts/qmem-sqlite.mjs flush [--limit N]
* node scripts/qmem-sqlite.mjs breaker [--reset]
* node scripts/qmem-sqlite.mjs enrich [--all] [--limit N] [--pace MS]
* node scripts/qmem-sqlite.mjs pull [--limit N]
*
@@ -27,7 +28,7 @@ process.emitWarning = (warning, ...rest) => {
};
const localDb = await import("../extensions/local-db.ts");
const { DEFAULT_DB_FILE, enrichFromGateway, flushQueue, importFromSessions, localDbPath, localDbReport, localSearch, pullFromGatewayExport, queueList, queueStats, queueStore, sessionRoots, submitOrQueue } = localDb;
const { loadConfig } = await import("../extensions/shared.ts");
const { breakerInfo, loadConfig, resetBreaker } = await import("../extensions/shared.ts");
process.emitWarning = originalEmitWarning;
const argv = process.argv.slice(2);
@@ -65,6 +66,10 @@ if (cmd === "status") {
}
if (r.pendingInIndex) console.log(` in indice : ${r.pendingInIndex} record marcati ⏳ (creati offline, non ancora sul gateway)`);
if (r.deleted) console.log(` tombstone : ${r.deleted} record cancellati sul gateway (soft delete)`);
{
const br = breakerInfo();
console.log(` breaker : ${br.open ? `APERTO (riprova tra ${Math.ceil(br.remainingMs / 1000)}s)` : "chiuso"} | fallimenti ${br.failures} | aperture ${br.trips}${br.lastError ? ` | ultimo errore: ${String(br.lastError).slice(0, 70)}` : ""}`);
}
console.log(` top progetti : ${r.topProjects.map((p) => `${p.project_id ?? "(null)"}=${p.n}`).join(", ") || "-"}`);
if (r.duplicates.length) {
console.log(` possibili duplicati (testo identico): ${r.duplicates.length} gruppi — es. ${r.duplicates[0].ids.map((i) => i.slice(0, 8)).join(", ")}`);
@@ -178,6 +183,22 @@ if (cmd === "status") {
if (res.stopped) console.log(` fermato: ${res.stopped}`);
if (res.errors.length) console.log(` errori: ${res.errors.join(" | ")}`);
}
} else if (cmd === "breaker") {
if (has("reset")) {
const br = resetBreaker();
out(json ? br : `Circuit breaker: chiuso (fallimenti ${br.failures})`);
} else {
const br = breakerInfo();
if (json) out(br);
else {
console.log(`Circuit breaker qmem (${br.file})`);
console.log(` stato: ${br.open ? `APERTO — riprova tra ${Math.ceil(br.remainingMs / 1000)}s` : "chiuso"}`);
console.log(` fallimenti consecutivi: ${br.failures} | aperture totali: ${br.trips}`);
if (br.lastError) console.log(` ultimo errore: ${br.lastError}`);
if (br.lastChange) console.log(` ultimo cambio: ${br.lastChange}`);
console.log(" reset: qmem-sqlite breaker --reset");
}
}
} else if (cmd === "pull") {
const cfg = loadConfig();
const res = await pullFromGatewayExport(cfg, { dbFile, limit: Number(opt("limit", 500)) || 500 });
@@ -187,7 +208,7 @@ if (cmd === "status") {
console.log(` supportato: ${res.supported} pagine: ${res.pages} record: ${res.fetched}${res.message ? ` — ${res.message}` : ""}`);
}
} else {
console.error(`Comando sconosciuto: ${cmd}\nComandi: status | import | find | store | queue | flush | enrich | pull`);
console.error(`Comando sconosciuto: ${cmd}\nComandi: status | import | find | store | queue | flush | breaker | enrich | pull`);
process.exit(2);
}
scripts/test-local.mjs
+101 -2
View File
@@ -66,7 +66,7 @@ const lines = [
fs.writeFileSync(path.join(SESS_DIR, "2026-09-01T10-00-00-000Z_test.jsonl"), lines.join("\n") + "\n");
// ---------------------------------------------------------------- config black-hole
fs.writeFileSync(CONFIG, JSON.stringify({ url: "http://127.0.0.1:9", apiKey: "test-key", timeoutMs: 1500, localDbPath: DB }, null, 2), { mode: 0o600 });
fs.writeFileSync(CONFIG, JSON.stringify({ url: "http://127.0.0.1:9", apiKey: "test-key", timeoutMs: 1500, connectTimeoutMs: 700, breakerBaseMs: 20000, breakerMaxMs: 60000, localDbPath: DB }, null, 2), { mode: 0o600 });
const env = { ...process.env, HOME, QMEM_SQLITE: DB, QMEM_SESSIONS_DIR: path.join(HOME, ".pi", "agent", "sessions") };
// isolamento anche per il processo di test: l'estensione caricata in-process
@@ -155,6 +155,7 @@ const ctx = {
const search = await tools.get("qmem_search").execute("t1", { query: "sqlite fts5" }, undefined, undefined, ctx);
check("qmem_search → fallback locale con gateway giù", search.details?.fallback === "local_sqlite" && search.details?.hits > 0, `hits=${search.details?.hits} status=${search.details?.gateway_status}`);
check("risultato locale etichettato come testuale", /INDICE LOCALE/i.test(search.content[0].text) && /non neurale/i.test(search.content[0].text) && /sqlite/i.test(search.content[0].text));
check("il messaggio indica il circuit breaker e come forzare un tentativo", /circuit breaker aperto/i.test(search.content[0].text) && /breaker reset/i.test(search.content[0].text), `breaker=${JSON.stringify(search.details?.breaker)}`);
const getRes = await tools.get("qmem_get").execute("t2", { memory_id: ID_E }, undefined, undefined, ctx);
if (process.env.DEBUG_GET) console.log("GET RESULT:", JSON.stringify(getRes, null, 1).slice(0, 900));
check("qmem_get → fallback locale per UUID", getRes.details?.fallback === "local_sqlite" && /nmap/.test(getRes.content[0].text));
@@ -205,7 +206,7 @@ const server = createServer((req, res) => {
});
await new Promise((r) => server.listen(0, "127.0.0.1", r));
const stub = `http://127.0.0.1:${server.address().port}`;
fs.writeFileSync(CONFIG, JSON.stringify({ url: stub, apiKey: "test-key", timeoutMs: 3000, localDbPath: DB }, null, 2), { mode: 0o600 });
fs.writeFileSync(CONFIG, JSON.stringify({ url: stub, apiKey: "test-key", timeoutMs: 3000, connectTimeoutMs: 700, breakerBaseMs: 20000, localDbPath: DB }, null, 2), { mode: 0o600 });
const enrichOut = await runCliAsync(["enrich", "--json"]);
let enrich = { stdout: enrichOut };
let enrichJson = null;
@@ -266,6 +267,104 @@ const dupItem = (qFinal.items ?? []).find((i) => i.local_id === dupStore.local_i
const badItem = (qFinal.items ?? []).find((i) => i.local_id === badStore.local_id);
check("duplicato marcato con remote_id del match", dupItem?.status === "duplicate" && dupItem?.remote_id === ID_A, `status=${dupItem?.status} remote=${String(dupItem?.remote_id).slice(0, 8)}`);
check("422 marcato failed con errore conservato", badItem?.status === "failed" && /422/.test(badItem?.last_error ?? ""), `err=${String(badItem?.last_error).slice(0, 40)}`);
// ---------------------------------------------------------------- 6) circuit breaker
if (process.env.SKIP_BREAKER !== "1") {
const origEmit = process.emitWarning;
process.emitWarning = (w, ...r) => {
const code = r[0]?.code ?? (typeof r[0] === "string" ? r[0] : undefined) ?? w?.code;
if (code === "MODULE_TYPELESS_PACKAGE_JSON") return;
return origEmit.call(process, w, ...r);
};
const shared = await import("../extensions/shared.ts");
process.emitWarning = origEmit;
const BH = { url: "http://127.0.0.1:9", apiKey: "k", connectTimeoutMs: 600, breakerBaseMs: 20000, breakerMaxMs: 60000 };
shared.resetBreaker();
check("breaker inizialmente chiuso", shared.breakerInfo().open === false);
// connessione rifiutata → fallimento immediato, nessun retry
let t = Date.now();
const r1 = await shared.gatewayRequest(BH, "GET", "/v1/status");
const ms1 = Date.now() - t;
check("connessione rifiutata: fallimento immediato senza retry", r1.status === 0 && r1.data?.error === "gateway_unreachable" && ms1 < 1500, `${ms1}ms`);
const br1 = shared.breakerInfo();
check("breaker APERTO dopo il fallimento di connessione", br1.open === true, `open=${br1.open} failures=${br1.failures} trips=${br1.trips}`);
// fast-fail: nessuna richiesta di rete
t = Date.now();
const r2 = await shared.gatewayRequest(BH, "GET", "/v1/status");
const ms2 = Date.now() - t;
check("fast-fail con breaker aperto (nessuna rete, < 50ms)", r2.status === 0 && r2.data?.breaker_open === true && ms2 < 50, `${ms2}ms`);
// persistenza fra processi (nuovo processo, stesso file di stato)
const code = `
const p = ${JSON.stringify(path.join(REPO, "extensions/shared.ts"))};
import(p).then(async (m) => {
const t = Date.now();
const r = await m.gatewayRequest({ url: "http://127.0.0.1:9", apiKey: "k", connectTimeoutMs: 600 }, "GET", "/v1/status");
console.log(JSON.stringify({ ms: Date.now() - t, status: r.status, breaker_open: !!r.data?.breaker_open, open: m.breakerInfo().open }));
});`;
const child = spawnSync("node", ["-e", code], { env, encoding: "utf8" });
let childOut = null;
try {
childOut = JSON.parse((child.stdout || "{}").trim().split("\n").pop());
} catch {
/* ignore */
}
check(
"breaker persistente fra processi (nuovo processo → fast-fail)",
childOut?.breaker_open === true && childOut?.ms < 200,
`ms=${childOut?.ms} open=${childOut?.open} stderr=${(child.stderr || "").slice(0, 60)}`,
);
// server che accetta ma non risponde mai → connect timeout breve (non 30s)
const hang = createServer(() => {});
await new Promise((r) => hang.listen(0, "127.0.0.1", r));
const hangUrl = `http://127.0.0.1:${hang.address().port}`;
shared.resetBreaker();
t = Date.now();
const r3 = await shared.gatewayRequest({ url: hangUrl, apiKey: "k", connectTimeoutMs: 700 }, "GET", "/v1/status");
const ms3 = Date.now() - t;
check("server che non risponde: connect timeout ~700ms (non 30s) e breaker aperto", r3.status === 0 && ms3 >= 600 && ms3 < 2600 && shared.breakerInfo().open === true, `${ms3}ms`);
hang.close();
// header subito ma body lento: elaborazione lunga → fallimento ambiguo, breaker NON aperto
const stall = createServer((_req, res) => {
res.writeHead(200, { "content-type": "application/json" });
res.write('{"status":');
// nessun end: il body resta appeso
});
await new Promise((r) => stall.listen(0, "127.0.0.1", r));
const stallUrl = `http://127.0.0.1:${stall.address().port}`;
shared.resetBreaker();
t = Date.now();
const r4 = await shared.gatewayRequest({ url: stallUrl, apiKey: "k", connectTimeoutMs: 700, timeoutMs: 1200 }, "GET", "/v1/status");
const ms4 = Date.now() - t;
const br4 = shared.breakerInfo();
check(
"body lento: timeout di elaborazione (~1.2s) senza aprire il breaker",
r4.status === 0 && r4.data?.error === "timeout_body" && ms4 < 2600 && br4.open === false,
`${ms4}ms failures=${br4.failures} open=${br4.open}`,
);
// store con gateway che accetta ma non risponde: deve accodare, non lanciare
fs.writeFileSync(CONFIG, JSON.stringify({ url: stallUrl, apiKey: "test-key", timeoutMs: 1200, connectTimeoutMs: 700, breakerBaseMs: 20000, localDbPath: DB }, null, 2), { mode: 0o600 });
const queuedOnStall = await tools.get("qmem_store").execute("t9", { text: "Record accodato con gateway che non risponde", project_id: "stall-proj" }, undefined, undefined, ctx);
check(
"gateway che accetta e non risponde: store ACCODATO (nessuna eccezione)",
queuedOnStall.details?.queued === true && /ACCODATO/i.test(queuedOnStall.content[0].text),
`queued=${queuedOnStall.details?.queued} status=${queuedOnStall.details?.gateway_status}`,
);
// cambio di endpoint → il breaker riparte chiuso
shared.tripBreaker("endpoint A", true, { ...BH, breakerBaseMs: 60000, url: "http://a" });
const r5 = await shared.gatewayRequest({ url: hangUrl, apiKey: "k", connectTimeoutMs: 400 }, "GET", "/v1/status");
check("cambio endpoint: il breaker non blocca il nuovo gateway", r5.data?.breaker_open !== true, `error=${r5.data?.error}`);
shared.resetBreaker();
check("reset manuale chiude il breaker", shared.breakerInfo().open === false);
stall.close();
}
// ---------------------------------------------------------------- report
const failed = results.filter((r) => !r.ok);
console.log(`\n${results.length - failed.length}/${results.length} controlli superati — HOME di test: ${HOME}`);