From 52a0b432fb96cb7c392f534a09426f3dddc252d3 Mon Sep 17 00:00:00 2001 From: vlad-ko Date: Thu, 9 Jul 2026 11:17:11 -0400 Subject: [PATCH 1/2] fix: Batch vector upserts + background timeout budget (Fixes BATTLE-MAGE-5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 2s VECTOR_OP_TIMEOUT_MS was sized for interactive paths (KB recall, search arms inside a turn) but also governed the background embed pipelines, where one upsert carries a whole file's chunks (up to ~130) for server-side embedding — legitimately seconds of work. Large files timed out on every 5-minute tick, permanently stalling the code index on exactly the files most worth indexing, while the un-abortable request often succeeded server-side anyway (wasted embedding spend on each retry). The docs pipeline had the same bug latent: it upserts an entire corpus in one call, fire-and-forget. - vectorUpsert now batches at UPSERT_BATCH_SIZE (20) with a per-batch timeout, stopping on the first failed batch; deterministic ids make the caller's retry of already-written batches idempotent - Per-call timeout override: background pipelines (code index, docs embed) pass VECTOR_BACKGROUND_TIMEOUT_MS (30s), absorbed by the ticks' existing wall-clock budgets; interactive paths keep 2s Co-Authored-By: Claude Fable 5 --- src/lib/code-index.test.ts | 3 ++ src/lib/code-index.ts | 5 +++ src/lib/repo-index.test.ts | 10 +++-- src/lib/repo-index.ts | 5 ++- src/lib/vector.test.ts | 81 ++++++++++++++++++++++++++++++++++++++ src/lib/vector.ts | 40 ++++++++++++++++--- 6 files changed, 135 insertions(+), 9 deletions(-) diff --git a/src/lib/code-index.test.ts b/src/lib/code-index.test.ts index 5d4c69f..452172e 100644 --- a/src/lib/code-index.test.ts +++ b/src/lib/code-index.test.ts @@ -51,6 +51,7 @@ vi.mock("./logger", () => ({ })); vi.mock("./vector", () => ({ + VECTOR_BACKGROUND_TIMEOUT_MS: 30_000, isVectorConfigured: (...args: unknown[]) => h.isVectorConfiguredMock(...args), srcNamespace: () => "acme_backend:src", docsNamespace: (sha: string) => `acme_backend:docs:${sha}`, @@ -221,6 +222,8 @@ describe("runCodeIndexTick", () => { expect(vectorUpsertSpy).toHaveBeenCalledWith( "acme_backend:src", [expect.objectContaining({ id: "src/a.ts#0", metadata: expect.objectContaining({ path: "src/a.ts", startLine: 1 }) })], + // Background pipeline uses the generous embed budget (BATTLE-MAGE-5). + { timeoutMs: 30_000 }, ); expect(kvData.get("srcindex:manifest")).toEqual({ "src/a.ts": { sha: "blob-a", chunks: 1 } }); expect(kvData.get("srcindex:sha")).toBe("sha-2"); diff --git a/src/lib/code-index.ts b/src/lib/code-index.ts index 1564c95..8ab8481 100644 --- a/src/lib/code-index.ts +++ b/src/lib/code-index.ts @@ -37,6 +37,7 @@ import { log } from "./logger"; import { isVectorConfigured, srcNamespace, + VECTOR_BACKGROUND_TIMEOUT_MS, vectorUpsert, vectorDelete, } from "./vector"; @@ -324,6 +325,10 @@ export async function runCodeIndexTick( const ok = await vectorUpsert( namespace, chunks.map((c) => ({ id: c.id, text: c.text, metadata: c.metadata })), + // Background pipeline: server-side embedding of a file's chunks + // legitimately takes seconds (BATTLE-MAGE-5); the tick's + // wall-clock budget absorbs the slack. + { timeoutMs: VECTOR_BACKGROUND_TIMEOUT_MS }, ); if (!ok) { degraded = true; diff --git a/src/lib/repo-index.test.ts b/src/lib/repo-index.test.ts index eb19790..a0ae82a 100644 --- a/src/lib/repo-index.test.ts +++ b/src/lib/repo-index.test.ts @@ -39,6 +39,7 @@ vi.mock("./kv", () => ({ })); vi.mock("./vector", () => ({ + VECTOR_BACKGROUND_TIMEOUT_MS: 30_000, isVectorConfigured: () => true, docsNamespace: (sha: string) => `acme_backend:docs:${sha}`, vectorUpsert: (...a: unknown[]) => vectorUpsertSpy(...a), @@ -610,9 +611,12 @@ describe("getOrRebuildIndex — doc embedding hook (#127)", () => { expect(logSpy).toHaveBeenCalledWith("docs_embedded", expect.anything()), ); - expect(vectorUpsertSpy).toHaveBeenCalledWith("acme_backend:docs:new-sha", [ - expect.objectContaining({ id: "docs/setup.md#0" }), - ]); + expect(vectorUpsertSpy).toHaveBeenCalledWith( + "acme_backend:docs:new-sha", + [expect.objectContaining({ id: "docs/setup.md#0" })], + // Background pipeline uses the generous embed budget (BATTLE-MAGE-5). + { timeoutMs: 30_000 }, + ); expect(kvData.get("index:vector_docs_ns")).toBe("acme_backend:docs:new-sha"); expect(vectorDeleteNamespaceSpy).toHaveBeenCalledWith( "acme_backend:docs:old-sha", diff --git a/src/lib/repo-index.ts b/src/lib/repo-index.ts index 4c77463..7b9d708 100644 --- a/src/lib/repo-index.ts +++ b/src/lib/repo-index.ts @@ -5,6 +5,7 @@ import { docsNamespace, vectorUpsert, vectorDeleteNamespace, + VECTOR_BACKGROUND_TIMEOUT_MS, } from "./vector"; import { type BattleMageConfig, @@ -521,7 +522,9 @@ async function embedDocChunks(sha: string, docPaths: string[]): Promise { } const namespace = docsNamespace(sha); - const ok = await vectorUpsert(namespace, items); + // Background pipeline: one docs corpus can be many chunks; use the + // generous embed budget, not the interactive 2s cap (BATTLE-MAGE-5). + const ok = await vectorUpsert(namespace, items, { timeoutMs: VECTOR_BACKGROUND_TIMEOUT_MS }); if (!ok) { log("docs_embed_failed", { sha, docCount, chunkCount: chunks.length }); return; diff --git a/src/lib/vector.test.ts b/src/lib/vector.test.ts index f08bd94..5853bee 100644 --- a/src/lib/vector.test.ts +++ b/src/lib/vector.test.ts @@ -22,6 +22,8 @@ import { docsNamespace, srcNamespace, VECTOR_OP_TIMEOUT_MS, + VECTOR_BACKGROUND_TIMEOUT_MS, + UPSERT_BATCH_SIZE, __setVectorStoreFactoryForTests, type VectorStore, } from "./vector"; @@ -277,6 +279,85 @@ describe("degradation: store errors never throw", () => { }); }); +describe("upsert batching + timeout override (BATTLE-MAGE-5)", () => { + beforeEach(() => stubVectorEnv()); + + const item = (i: number) => ({ id: `c${i}`, text: `chunk ${i}` }); + + it("pins the background timeout and batch size constants", () => { + expect(VECTOR_BACKGROUND_TIMEOUT_MS).toBe(30_000); + expect(UPSERT_BATCH_SIZE).toBe(20); + expect(VECTOR_BACKGROUND_TIMEOUT_MS).toBeGreaterThan(VECTOR_OP_TIMEOUT_MS); + }); + + it("splits a large item list into sequential store calls of at most UPSERT_BATCH_SIZE", async () => { + const store = makeFakeStore(); + __setVectorStoreFactoryForTests(() => store); + const items = Array.from({ length: UPSERT_BATCH_SIZE * 2 + 5 }, (_, i) => item(i)); + const ok = await vectorUpsert("ns", items); + expect(ok).toBe(true); + expect(store.upsert).toHaveBeenCalledTimes(3); + expect(vi.mocked(store.upsert).mock.calls[0][1]).toHaveLength(UPSERT_BATCH_SIZE); + expect(vi.mocked(store.upsert).mock.calls[1][1]).toHaveLength(UPSERT_BATCH_SIZE); + expect(vi.mocked(store.upsert).mock.calls[2][1]).toHaveLength(5); + // One success log carrying the TOTAL count. + expect(logSpy).toHaveBeenCalledWith( + "vector_op", + expect.objectContaining({ op: "upsert", count: UPSERT_BATCH_SIZE * 2 + 5 }), + ); + }); + + it("stops on the first failed batch and returns false (later batches never sent)", async () => { + const store = makeFakeStore({ + upsert: vi + .fn() + .mockResolvedValueOnce(undefined) + .mockRejectedValueOnce(new Error("boom")), + }); + __setVectorStoreFactoryForTests(() => store); + const items = Array.from({ length: UPSERT_BATCH_SIZE * 3 }, (_, i) => item(i)); + const ok = await vectorUpsert("ns", items); + expect(ok).toBe(false); + expect(store.upsert).toHaveBeenCalledTimes(2); + }); + + it("a list at or under the batch size still makes exactly one store call", async () => { + const store = makeFakeStore(); + __setVectorStoreFactoryForTests(() => store); + await vectorUpsert("ns", Array.from({ length: UPSERT_BATCH_SIZE }, (_, i) => item(i))); + expect(store.upsert).toHaveBeenCalledTimes(1); + }); + + it("honors a per-call timeout override: survives past the default, fails past the override", async () => { + vi.useFakeTimers(); + __setVectorStoreFactoryForTests(() => + makeFakeStore({ upsert: () => new Promise(() => {}) }), // hangs forever + ); + const pending = vectorUpsert("ns", [item(1)], { timeoutMs: VECTOR_BACKGROUND_TIMEOUT_MS }); + await vi.advanceTimersByTimeAsync(VECTOR_OP_TIMEOUT_MS + 1); + // Default budget elapsed — the overridden call must still be alive. + let settled = false; + pending.then(() => { settled = true; }); + await Promise.resolve(); + expect(settled).toBe(false); + await vi.advanceTimersByTimeAsync(VECTOR_BACKGROUND_TIMEOUT_MS); + await expect(pending).resolves.toBe(false); + const err = logSpy.mock.calls.find((c) => c[0] === "vector_error"); + expect(err?.[1]).toMatchObject({ op: "upsert", errorClass: "VectorTimeoutError" }); + expect(String(err?.[1].errorMessage)).toContain("30000"); + }); + + it("default timeout still applies when no override is passed", async () => { + vi.useFakeTimers(); + __setVectorStoreFactoryForTests(() => + makeFakeStore({ upsert: () => new Promise(() => {}) }), + ); + const pending = vectorUpsert("ns", [item(1)]); + await vi.advanceTimersByTimeAsync(VECTOR_OP_TIMEOUT_MS + 1); + await expect(pending).resolves.toBe(false); + }); +}); + describe("degradation: timeout", () => { beforeEach(() => stubVectorEnv()); diff --git a/src/lib/vector.ts b/src/lib/vector.ts index 609f3da..6eb971d 100644 --- a/src/lib/vector.ts +++ b/src/lib/vector.ts @@ -172,17 +172,38 @@ function getStore(): VectorStore { /** Hard latency cap per vector op — recall must never stall a turn. */ export const VECTOR_OP_TIMEOUT_MS = 2000; +/** + * Timeout for background embed pipelines (code index, doc embedding), + * where a single upsert batch is server-side-embedded and legitimately + * needs seconds — the 2s interactive cap starved them and stalled the + * index on large files (BATTLE-MAGE-5). Cron budgets absorb the slack. + */ +export const VECTOR_BACKGROUND_TIMEOUT_MS = 30_000; + +/** + * Max items per underlying store call. One upsert request carrying a + * whole large file (or a whole docs corpus) embeds slowly and fails as + * a unit; smaller batches keep per-request latency inside the timeout + * and make retries finer-grained. Ids are deterministic, so re-sending + * a batch after a mid-list failure is harmless. + */ +export const UPSERT_BATCH_SIZE = 20; + export class VectorTimeoutError extends Error { - constructor(op: string) { - super(`vector ${op} exceeded ${VECTOR_OP_TIMEOUT_MS}ms`); + constructor(op: string, timeoutMs: number) { + super(`vector ${op} exceeded ${timeoutMs}ms`); this.name = "VectorTimeoutError"; } } -function withTimeout(op: string, promise: Promise): Promise { +function withTimeout( + op: string, + promise: Promise, + timeoutMs: number = VECTOR_OP_TIMEOUT_MS, +): Promise { let timer: ReturnType | undefined; const timeout = new Promise((_, reject) => { - timer = setTimeout(() => reject(new VectorTimeoutError(op)), VECTOR_OP_TIMEOUT_MS); + timer = setTimeout(() => reject(new VectorTimeoutError(op, timeoutMs)), timeoutMs); }); return Promise.race([promise, timeout]).finally(() => clearTimeout(timer)) as Promise; } @@ -221,20 +242,29 @@ function logError(op: VectorOp, namespace: string, startedAt: number, err: unkno export async function vectorUpsert( namespace: string, items: VectorUpsertItem[], + opts?: { timeoutMs?: number }, ): Promise { if (items.length === 0) return true; if (!isVectorConfigured()) { logUnavailable("upsert"); return false; } + const timeoutMs = opts?.timeoutMs ?? VECTOR_OP_TIMEOUT_MS; const startedAt = Date.now(); try { - await withTimeout("upsert", getStore().upsert(namespace, items)); + // Sequential batches; stop on the first failure. Completed batches + // stay written — deterministic ids make the caller's retry re-send + // them idempotently. + for (let i = 0; i < items.length; i += UPSERT_BATCH_SIZE) { + const batch = items.slice(i, i + UPSERT_BATCH_SIZE); + await withTimeout("upsert", getStore().upsert(namespace, batch), timeoutMs); + } log("vector_op", { op: "upsert", namespace, durationMs: Date.now() - startedAt, count: items.length, + batches: Math.ceil(items.length / UPSERT_BATCH_SIZE), }); return true; } catch (err) { From e3bc18b73d6ffa421b8ab4835afc8b953cea36ef Mon Sep 17 00:00:00 2001 From: vlad-ko Date: Thu, 9 Jul 2026 11:24:06 -0400 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20Address=20PR=20#144=20review=20findi?= =?UTF-8?q?ngs=20=E2=80=94=20end-to-end=20upsert=20deadline=20+=20constant?= =?UTF-8?q?=20flow=20in=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - timeoutMs is now a per-CALL deadline: each batch races the remaining budget, so a multi-batch upsert can never exceed the caller's cap (previously each batch got a fresh budget — batches × 30s could overrun the tick's wall-clock). Deadline test pins 3×12s batches failing at a 30s cap. - Test assertions reference VECTOR_BACKGROUND_TIMEOUT_MS instead of a raw 30_000 literal; the vector mocks now spread importOriginal so real constants flow through and can't drift from production. Co-Authored-By: Claude Fable 5 --- src/lib/code-index.test.ts | 10 ++++++---- src/lib/repo-index.test.ts | 10 ++++++---- src/lib/vector.test.ts | 21 +++++++++++++++++++++ src/lib/vector.ts | 9 +++++++-- 4 files changed, 40 insertions(+), 10 deletions(-) diff --git a/src/lib/code-index.test.ts b/src/lib/code-index.test.ts index 452172e..04fe51f 100644 --- a/src/lib/code-index.test.ts +++ b/src/lib/code-index.test.ts @@ -6,6 +6,7 @@ // tree snapshot SHA. import { describe, it, expect, vi, beforeEach } from "vitest"; +import { VECTOR_BACKGROUND_TIMEOUT_MS } from "./vector"; import type { BattleMageConfig } from "./config"; const h = vi.hoisted(() => { @@ -50,9 +51,10 @@ vi.mock("./logger", () => ({ log: (...args: unknown[]) => h.logSpy(...args), })); -vi.mock("./vector", () => ({ - VECTOR_BACKGROUND_TIMEOUT_MS: 30_000, - isVectorConfigured: (...args: unknown[]) => h.isVectorConfiguredMock(...args), +vi.mock("./vector", async (importOriginal) => ({ + // Real constants flow through; only the side-effectful fns are mocked. + ...(await importOriginal()), + isVectorConfigured: (...args: unknown[]) => h.isVectorConfiguredMock(...args), srcNamespace: () => "acme_backend:src", docsNamespace: (sha: string) => `acme_backend:docs:${sha}`, kbNamespace: () => "acme_backend:kb", @@ -223,7 +225,7 @@ describe("runCodeIndexTick", () => { "acme_backend:src", [expect.objectContaining({ id: "src/a.ts#0", metadata: expect.objectContaining({ path: "src/a.ts", startLine: 1 }) })], // Background pipeline uses the generous embed budget (BATTLE-MAGE-5). - { timeoutMs: 30_000 }, + { timeoutMs: VECTOR_BACKGROUND_TIMEOUT_MS }, ); expect(kvData.get("srcindex:manifest")).toEqual({ "src/a.ts": { sha: "blob-a", chunks: 1 } }); expect(kvData.get("srcindex:sha")).toBe("sha-2"); diff --git a/src/lib/repo-index.test.ts b/src/lib/repo-index.test.ts index a0ae82a..966faf3 100644 --- a/src/lib/repo-index.test.ts +++ b/src/lib/repo-index.test.ts @@ -1,4 +1,5 @@ import { describe, it, expect, vi, beforeEach } from "vitest"; +import { VECTOR_BACKGROUND_TIMEOUT_MS } from "./vector"; // ── #127 mocks: kv / github / vector / logger for the rebuild-hook tests ── // The pure-function tests below never touch these; the mocks only feed @@ -38,9 +39,10 @@ vi.mock("./kv", () => ({ }, })); -vi.mock("./vector", () => ({ - VECTOR_BACKGROUND_TIMEOUT_MS: 30_000, - isVectorConfigured: () => true, +vi.mock("./vector", async (importOriginal) => ({ + // Real constants flow through; only the side-effectful fns are mocked. + ...(await importOriginal()), + isVectorConfigured: () => true, docsNamespace: (sha: string) => `acme_backend:docs:${sha}`, vectorUpsert: (...a: unknown[]) => vectorUpsertSpy(...a), vectorDeleteNamespace: (...a: unknown[]) => vectorDeleteNamespaceSpy(...a), @@ -615,7 +617,7 @@ describe("getOrRebuildIndex — doc embedding hook (#127)", () => { "acme_backend:docs:new-sha", [expect.objectContaining({ id: "docs/setup.md#0" })], // Background pipeline uses the generous embed budget (BATTLE-MAGE-5). - { timeoutMs: 30_000 }, + { timeoutMs: VECTOR_BACKGROUND_TIMEOUT_MS }, ); expect(kvData.get("index:vector_docs_ns")).toBe("acme_backend:docs:new-sha"); expect(vectorDeleteNamespaceSpy).toHaveBeenCalledWith( diff --git a/src/lib/vector.test.ts b/src/lib/vector.test.ts index 5853bee..26b76d0 100644 --- a/src/lib/vector.test.ts +++ b/src/lib/vector.test.ts @@ -347,6 +347,27 @@ describe("upsert batching + timeout override (BATTLE-MAGE-5)", () => { expect(String(err?.[1].errorMessage)).toContain("30000"); }); + it("timeoutMs is an end-to-end deadline for the whole call, not per batch", async () => { + vi.useFakeTimers(); + // Each batch takes 12s. With a 30s per-CALL budget, batch 1 (12s) and + // batch 2 (24s) fit; batch 3 would finish at 36s — the call must fail + // at the 30s deadline instead of granting each batch a fresh 30s. + __setVectorStoreFactoryForTests(() => + makeFakeStore({ + upsert: () => new Promise((resolve) => setTimeout(() => resolve(undefined), 12_000)), + }), + ); + const items = Array.from({ length: UPSERT_BATCH_SIZE * 3 }, (_, i) => ({ + id: `c${i}`, + text: `chunk ${i}`, + })); + const pending = vectorUpsert("ns", items, { timeoutMs: 30_000 }); + await vi.advanceTimersByTimeAsync(40_000); + await expect(pending).resolves.toBe(false); + const err = logSpy.mock.calls.find((c) => c[0] === "vector_error"); + expect(err?.[1]).toMatchObject({ op: "upsert", errorClass: "VectorTimeoutError" }); + }); + it("default timeout still applies when no override is passed", async () => { vi.useFakeTimers(); __setVectorStoreFactoryForTests(() => diff --git a/src/lib/vector.ts b/src/lib/vector.ts index 6eb971d..856b026 100644 --- a/src/lib/vector.ts +++ b/src/lib/vector.ts @@ -254,10 +254,15 @@ export async function vectorUpsert( try { // Sequential batches; stop on the first failure. Completed batches // stay written — deterministic ids make the caller's retry re-send - // them idempotently. + // them idempotently. timeoutMs is an END-TO-END deadline for the + // whole call: each batch races the remaining budget, so a + // multi-batch upsert can never exceed the caller's per-call cap. + const deadline = startedAt + timeoutMs; for (let i = 0; i < items.length; i += UPSERT_BATCH_SIZE) { + const remaining = deadline - Date.now(); + if (remaining <= 0) throw new VectorTimeoutError("upsert", timeoutMs); const batch = items.slice(i, i + UPSERT_BATCH_SIZE); - await withTimeout("upsert", getStore().upsert(namespace, batch), timeoutMs); + await withTimeout("upsert", getStore().upsert(namespace, batch), remaining); } log("vector_op", { op: "upsert",