Skip to content

Commit 26340df

Browse files
committed
fix(capture): skip tombstoned pending embeds and serialize background passes
Join memory.version in findPendingChunks so a forget-then-add cannot sweep redacted placeholder text (tombstoned status and non-live generations). Coalesce overlapping background embed passes so concurrent adds share one in-flight drain instead of N retry storms against a busy Ollama.
1 parent 4bd8d2d commit 26340df

2 files changed

Lines changed: 247 additions & 9 deletions

File tree

‎src/services/capture.test.ts‎

Lines changed: 120 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -311,3 +311,123 @@ describe("background embed pass honors the configured embed timeout (CL-8615)",
311311
}
312312
});
313313
});
314+
315+
describe("findPendingChunks skips tombstoned and non-live versions", () => {
316+
it("joins memory.version and binds live generation so a forget-then-add cannot sweep [redacted] text", async () => {
317+
const { findPendingChunks } = await loadRealCapture();
318+
const queries: Array<{ sql: string; params: readonly unknown[] }> = [];
319+
const client = {
320+
query: (sql: string, params: readonly unknown[]) => {
321+
queries.push({ sql, params });
322+
return Promise.resolve([{ id: "chunk_live", text: "keep me" }]);
323+
},
324+
};
325+
const table = {
326+
tableName: `"memory"."embedding_0123456789abcdef"`,
327+
dims: 128,
328+
modelId: "test-model",
329+
};
330+
331+
const pending = await findPendingChunks(client, "tenant-1", table, 100);
332+
333+
expect(pending).toEqual([{ id: "chunk_live", text: "keep me" }]);
334+
expect(queries).toHaveLength(1);
335+
const { sql, params } = queries[0]!;
336+
expect(sql).toContain('INNER JOIN "memory"."version" v ON v.id = c.version_id');
337+
expect(sql).toContain("v.status <> 'tombstoned'");
338+
expect(sql).toContain("v.generation = $3");
339+
expect(params).toEqual(["tenant-1", 100, "live"]);
340+
});
341+
});
342+
343+
describe("background embed scheduler serializes overlapping adds", () => {
344+
it("overlapping enqueues for one tenant share one in-flight pass and never overlap run()", async () => {
345+
const { createBackgroundEmbedScheduler } = await loadRealCapture();
346+
const scheduler = createBackgroundEmbedScheduler();
347+
let current = 0;
348+
const batches: string[][] = [];
349+
let release!: () => void;
350+
const held = new Promise<void>((resolve) => {
351+
release = resolve;
352+
});
353+
let holding = true;
354+
355+
const run = async (chunks: EmbeddableChunk[]) => {
356+
current++;
357+
batches.push(chunks.map((chunk) => chunk.id));
358+
if (holding) {
359+
holding = false;
360+
await held;
361+
}
362+
current--;
363+
};
364+
365+
const p1 = scheduler.enqueue({
366+
tenantId: "tenant-1",
367+
chunks: [{ id: "a", text: "a" }],
368+
run,
369+
});
370+
const p2 = scheduler.enqueue({
371+
tenantId: "tenant-1",
372+
chunks: [{ id: "b", text: "b" }],
373+
run,
374+
});
375+
const p3 = scheduler.enqueue({
376+
tenantId: "tenant-1",
377+
chunks: [{ id: "c", text: "c" }],
378+
run,
379+
});
380+
381+
await Promise.resolve();
382+
expect(scheduler.passStarts()).toBe(1);
383+
expect(scheduler.maxConcurrent()).toBe(1);
384+
expect(current).toBe(1);
385+
release();
386+
await Promise.all([p1, p2, p3]);
387+
388+
expect(scheduler.maxConcurrent()).toBe(1);
389+
expect(scheduler.passStarts()).toBeLessThanOrEqual(2);
390+
expect(batches.flat().sort()).toEqual(["a", "b", "c"]);
391+
});
392+
393+
it("passes for two tenants still never run concurrently", async () => {
394+
const { createBackgroundEmbedScheduler } = await loadRealCapture();
395+
const scheduler = createBackgroundEmbedScheduler();
396+
let releaseFirst!: () => void;
397+
const held = new Promise<void>((resolve) => {
398+
releaseFirst = resolve;
399+
});
400+
let holding = true;
401+
const seenTenants: string[] = [];
402+
403+
const runFor = (tenantId: string) => async (_chunks: EmbeddableChunk[]) => {
404+
seenTenants.push(tenantId);
405+
if (holding) {
406+
holding = false;
407+
await held;
408+
}
409+
};
410+
411+
const pA = scheduler.enqueue({
412+
tenantId: "tenant-a",
413+
chunks: [{ id: "a", text: "a" }],
414+
run: runFor("tenant-a"),
415+
});
416+
const pB = scheduler.enqueue({
417+
tenantId: "tenant-b",
418+
chunks: [{ id: "b", text: "b" }],
419+
run: runFor("tenant-b"),
420+
});
421+
422+
await Promise.resolve();
423+
expect(scheduler.passStarts()).toBe(1);
424+
expect(scheduler.maxConcurrent()).toBe(1);
425+
expect(seenTenants).toEqual(["tenant-a"]);
426+
releaseFirst();
427+
await Promise.all([pA, pB]);
428+
429+
expect(scheduler.maxConcurrent()).toBe(1);
430+
expect(scheduler.passStarts()).toBe(2);
431+
expect(seenTenants).toEqual(["tenant-a", "tenant-b"]);
432+
});
433+
});

‎src/services/capture.ts‎

Lines changed: 127 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -642,6 +642,109 @@ const BACKGROUND_EMBED_PENDING_LIMIT = 100;
642642
// after a longer pause, rather than hammering it.
643643
const BACKGROUND_EMBED_RETRY_DELAYS_MS = [1_000, 5_000];
644644

645+
export type BackgroundEmbedPassFn = (
646+
chunks: EmbeddableChunk[],
647+
) => Promise<void>;
648+
649+
export type BackgroundEmbedEnqueue = {
650+
tenantId: string;
651+
chunks: readonly EmbeddableChunk[];
652+
run: BackgroundEmbedPassFn;
653+
};
654+
655+
export type BackgroundEmbedScheduler = {
656+
/**
657+
* Queue a pass. Overlapping calls for the same tenant coalesce into the
658+
* in-flight drain (at most one trailing rerun); different tenants share one
659+
* process-wide chain so N concurrent adds never fan out N retry storms.
660+
*/
661+
enqueue: (work: BackgroundEmbedEnqueue) => Promise<void>;
662+
/** Count of `run` invocations started (test seam). */
663+
passStarts: () => number;
664+
/** Peak overlapping `run` invocations (test seam). */
665+
maxConcurrent: () => number;
666+
};
667+
668+
/**
669+
* Serial, coalescing gate for detached embed passes. One `run` at a time
670+
* process-wide; same-tenant chunks that arrive while a drain is queued or
671+
* running join that drain instead of starting another 3-retry + sweep storm.
672+
*/
673+
export function createBackgroundEmbedScheduler(): BackgroundEmbedScheduler {
674+
type TenantState = {
675+
chunks: EmbeddableChunk[];
676+
run: BackgroundEmbedPassFn;
677+
waiters: Array<{
678+
resolve: () => void;
679+
reject: (err: unknown) => void;
680+
}>;
681+
queued: boolean;
682+
};
683+
684+
const tenants = new Map<string, TenantState>();
685+
let chain: Promise<void> = Promise.resolve();
686+
let passStarts = 0;
687+
let inRun = 0;
688+
let maxConcurrent = 0;
689+
690+
async function drainTenant(tenantId: string): Promise<void> {
691+
const state = tenants.get(tenantId);
692+
if (!state) return;
693+
let error: unknown;
694+
try {
695+
for (;;) {
696+
const batch = state.chunks.splice(0, state.chunks.length);
697+
if (batch.length === 0) break;
698+
passStarts++;
699+
inRun++;
700+
if (inRun > maxConcurrent) maxConcurrent = inRun;
701+
try {
702+
await state.run(batch);
703+
} finally {
704+
inRun--;
705+
}
706+
}
707+
} catch (err) {
708+
error = err;
709+
}
710+
const waiters = state.waiters.splice(0, state.waiters.length);
711+
tenants.delete(tenantId);
712+
if (error !== undefined) {
713+
for (const waiter of waiters) waiter.reject(error);
714+
return;
715+
}
716+
for (const waiter of waiters) waiter.resolve();
717+
}
718+
719+
return {
720+
enqueue(work) {
721+
let state = tenants.get(work.tenantId);
722+
if (!state) {
723+
state = { chunks: [], run: work.run, waiters: [], queued: false };
724+
tenants.set(work.tenantId, state);
725+
}
726+
state.chunks.push(...work.chunks);
727+
state.run = work.run;
728+
const done = new Promise<void>((resolve, reject) => {
729+
state.waiters.push({ resolve, reject });
730+
});
731+
if (!state.queued) {
732+
state.queued = true;
733+
const tenantId = work.tenantId;
734+
chain = chain.then(
735+
() => drainTenant(tenantId),
736+
() => drainTenant(tenantId),
737+
);
738+
}
739+
return done;
740+
},
741+
passStarts: () => passStarts,
742+
maxConcurrent: () => maxConcurrent,
743+
};
744+
}
745+
746+
const backgroundEmbedScheduler = createBackgroundEmbedScheduler();
747+
645748
function resolveBackgroundOpts(opts: CaptureBackgroundOpts = {}): {
646749
fetchImpl: typeof fetch;
647750
schedule: (task: () => Promise<void>) => void;
@@ -668,6 +771,11 @@ function resolveBackgroundOpts(opts: CaptureBackgroundOpts = {}): {
668771
* `limit` (oldest first) so a long outage sweeps incrementally, one
669772
* background pass at a time, instead of embedding an unbounded backlog in a
670773
* single call.
774+
*
775+
* Joins `memory.version` so a forget-then-add cannot sweep `[redacted]`
776+
* placeholder text: tombstoned versions and non-live generations are skipped.
777+
* Search already hides tombstones; this filter is wasted-work / placeholder-
778+
* vector prevention, not a hit-leak fix.
671779
*/
672780
export async function findPendingChunks(
673781
client: EmbedRegistrySqlClient,
@@ -682,11 +790,14 @@ export async function findPendingChunks(
682790
}
683791
const rows = await client.query(
684792
`SELECT c.id AS id, c.text AS text FROM "memory"."chunk" c
793+
INNER JOIN "memory"."version" v ON v.id = c.version_id
685794
LEFT JOIN ${activeTable.tableName} e ON e.chunk_id = c.id
686795
WHERE c.tenant_id = $1 AND e.chunk_id IS NULL
796+
AND v.status <> 'tombstoned'
797+
AND v.generation = $3
687798
ORDER BY c.created_at ASC
688799
LIMIT $2`,
689-
[tenantId, Math.max(1, Math.floor(limit))],
800+
[tenantId, Math.max(1, Math.floor(limit)), LIVE_GENERATION],
690801
);
691802
return rows.map((row) => ({
692803
id: row["id"] as string,
@@ -885,9 +996,11 @@ export async function captureDocument(
885996
// inference) must never stall the `add` tool call, so the embed pass runs
886997
// detached with retry plus a bounded pending-chunk sweep (see
887998
// runBackgroundEmbedPass), and failures stay in the logs rather than the
888-
// response. The synchronous paths below pay for no network: no chunks
889-
// means nothing to embed, and no embed endpoint (lexical-only) is decided
890-
// locally without touching the registry.
999+
// response. The success path therefore omits `degraded: ["embed_unavailable"]`
1000+
// even while chunks are still pending: a captured result means the row store
1001+
// committed, not that the embedder has caught up. The synchronous paths
1002+
// below pay for no network: no chunks means nothing to embed, and no embed
1003+
// endpoint (lexical-only) is decided locally without touching the registry.
8911004
const embedClientConfig = toEmbedClientConfig(deps.config.embed);
8921005
if (txResult.insertedChunks.length === 0 || !embedClientConfig) {
8931006
const { degraded } = await embedInsertedChunksWithConfig(
@@ -909,11 +1022,16 @@ export async function captureDocument(
9091022
const resolved = resolveBackgroundOpts(background);
9101023
const backgroundChunks = txResult.insertedChunks;
9111024
resolved.schedule(() =>
912-
runBackgroundEmbedPass(deps.sql, input.tenantId, backgroundChunks, embedClientConfig, {
913-
fetchImpl: resolved.fetchImpl,
914-
sleep: resolved.sleep,
915-
maxAttempts: resolved.maxAttempts,
916-
pendingLimit: resolved.pendingLimit,
1025+
backgroundEmbedScheduler.enqueue({
1026+
tenantId: input.tenantId,
1027+
chunks: backgroundChunks,
1028+
run: (chunks) =>
1029+
runBackgroundEmbedPass(deps.sql, input.tenantId, chunks, embedClientConfig, {
1030+
fetchImpl: resolved.fetchImpl,
1031+
sleep: resolved.sleep,
1032+
maxAttempts: resolved.maxAttempts,
1033+
pendingLimit: resolved.pendingLimit,
1034+
}),
9171035
}),
9181036
);
9191037

0 commit comments

Comments
 (0)