From 41a1f2c8eb4274cfc2edd7a11d577d5c8dc0350a Mon Sep 17 00:00:00 2001 From: qer Date: Tue, 8 Sep 2026 01:56:14 +0800 Subject: [PATCH 1/6] fix(kap-server): verify indexed search results against session sources --- .changeset/search-deleted-sessions.md | 5 + packages/kap-server/src/search/docs.ts | 5 + packages/kap-server/src/search/indexCore.ts | 98 ++++++++-- .../kap-server/src/search/searchService.ts | 2 +- .../test/search/searchRoute.test.ts | 1 + .../test/search/searchService.test.ts | 170 +++++++++++++++++- 6 files changed, 269 insertions(+), 12 deletions(-) create mode 100644 .changeset/search-deleted-sessions.md diff --git a/.changeset/search-deleted-sessions.md b/.changeset/search-deleted-sessions.md new file mode 100644 index 00000000000..117bcd0e163 --- /dev/null +++ b/.changeset/search-deleted-sessions.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Stop returning deleted sessions from global search before the search index catches up. diff --git a/packages/kap-server/src/search/docs.ts b/packages/kap-server/src/search/docs.ts index 5899ac42eff..4637338506e 100644 --- a/packages/kap-server/src/search/docs.ts +++ b/packages/kap-server/src/search/docs.ts @@ -2,6 +2,7 @@ export const MAX_DOC_TEXT_CHARS = 20_000; export interface MessageDoc { readonly kind: 'message'; + readonly sessionIdentity?: string; readonly sessionId: string; readonly workspaceId: string; readonly sessionTitle: string; @@ -15,6 +16,7 @@ export interface MessageDoc { export interface TitleDoc { readonly kind: 'title'; + readonly sessionIdentity?: string; readonly sessionId: string; readonly workspaceId: string; readonly sessionTitle: string; @@ -56,6 +58,9 @@ export interface FileMetaDoc { export interface SessionMetaDoc { readonly kind: 'sessionMeta'; + readonly title?: string; + readonly dir?: string; + readonly identity?: string; } export interface StatsDoc { diff --git a/packages/kap-server/src/search/indexCore.ts b/packages/kap-server/src/search/indexCore.ts index 70750caeebf..09a63f36bb6 100644 --- a/packages/kap-server/src/search/indexCore.ts +++ b/packages/kap-server/src/search/indexCore.ts @@ -65,6 +65,34 @@ function errorMessage(error: unknown): string { return error instanceof Error ? error.message : String(error); } +async function sessionDirectoryIdentity(dir: string): Promise { + try { + const info = await stat(dir, { bigint: true }); + return info.isDirectory() ? `${info.dev}:${info.ino}:${info.birthtimeNs}` : undefined; + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + if (code === 'ENOENT' || code === 'ENOTDIR') return undefined; + throw error; + } +} + +async function sessionDirectoryTitle(dir: string, log: SearchCoreLog): Promise { + for (const scope of ['', 'session-meta']) { + try { + const meta: unknown = JSON.parse(await readFile(join(dir, scope, 'state.json'), 'utf8')); + if (typeof meta === 'object' && meta !== null && 'title' in meta && typeof meta.title === 'string') { + return meta.title; + } + return ''; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') { + log.warn('search index: cannot read session title', { dir, error: errorMessage(error) }); + } + } + } + return ''; +} + const INITIAL_TURN_STATE: TurnCounterState = { next: 0, hasTurn: false, openers: [] }; function initialTurnState(): TurnCounterState { @@ -511,6 +539,22 @@ export class SearchIndexCore { } private async syncSession(db: MiniDb, summary: SyncSessionInput): Promise { + const identity = await sessionDirectoryIdentity(summary.dir); + if (identity === undefined) { + await this.deleteSessionDocs(db, summary.id); + return; + } + const title = await sessionDirectoryTitle(summary.dir, this.log); + const metaKey = SESSION_META_PREFIX + summary.id; + const previous = db.get(metaKey); + if (previous?.kind !== 'sessionMeta' || previous.identity !== identity || previous.dir !== summary.dir) { + await this.deleteSessionDocs(db, summary.id); + if (previous !== undefined) this.syncReplaced = true; + } + const meta: SessionMetaDoc = { kind: 'sessionMeta', dir: summary.dir, identity, title }; + if (previous?.kind !== 'sessionMeta' || previous.identity !== identity || previous.dir !== summary.dir || previous.title !== title) { + await db.set(metaKey, meta); + } const wireFiles = await collectWireFiles(summary.dir); const seenPaths = new Set(wireFiles.map((file) => file.path)); @@ -523,16 +567,16 @@ export class SearchIndexCore { } for (const file of wireFiles) { - await this.syncWireFile(db, summary, file); + await this.syncWireFile(db, { ...summary, title, sessionIdentity: identity }, file); } - const title = summary.title ?? ''; const titleKey = `${summary.id}/$title`; const existing = db.get(titleKey); if (title.length > 0) { if (existing?.kind !== 'title' || existing.text !== title) { const doc: TitleDoc = { kind: 'title', + sessionIdentity: identity, sessionId: summary.id, workspaceId: summary.workspaceId, sessionTitle: title, @@ -547,10 +591,6 @@ export class SearchIndexCore { } else if (existing !== undefined) { await db.del(titleKey); } - if (db.get(SESSION_META_PREFIX + summary.id) === undefined) { - const sessionMeta: SessionMetaDoc = { kind: 'sessionMeta' }; - await db.set(SESSION_META_PREFIX + summary.id, sessionMeta); - } } private async deleteFileDocs(db: MiniDb, meta: FileMetaDoc): Promise { @@ -562,7 +602,7 @@ export class SearchIndexCore { private async syncWireFile( db: MiniDb, - summary: SyncSessionInput, + summary: SyncSessionInput & { readonly sessionIdentity: string }, file: WireFileRef, ): Promise { let st: { size: number; mtimeMs: number; ino: number }; @@ -693,7 +733,7 @@ export class SearchIndexCore { private collectWireLine( ops: BatchInputOp[], - summary: SyncSessionInput, + summary: SyncSessionInput & { readonly sessionIdentity: string }, file: WireFileRef, line: string, lineOffset: number, @@ -717,6 +757,7 @@ export class SearchIndexCore { const stepOrdinal = e.stepUuid !== undefined ? stepState.byUuid[e.stepUuid] : undefined; const doc: MessageDoc = { kind: 'message', + sessionIdentity: summary.sessionIdentity, sessionId: summary.id, workspaceId: summary.workspaceId, sessionTitle: summary.title ?? '', @@ -865,14 +906,51 @@ export class SearchIndexCore { const boundary = page.kind === 'keyset' ? page.boundary : undefined; const matched = matchDocs(q, candidates, boundary, budget); incomplete ??= matched.incomplete; - const { pageRows, hasMore } = paginateRows(q, page, matched.rows); + const index = this.readIndexView(serveDb, freshnessStale); + const sources = new Map(); + for (const row of matched.rows) { + const id = row.value.sessionId; + if (sources.has(id)) continue; + if (Date.now() > budget.deadlineAt) { + incomplete ??= 'deadline'; + break; + } + const meta = serveDb.get(SESSION_META_PREFIX + id); + sources.set(id, meta?.kind === 'sessionMeta' ? meta : undefined); + } + const identities = new Map(); + const visible: MatchedRow[] = []; + for (const row of matched.rows) { + if (Date.now() > budget.deadlineAt) { + incomplete ??= 'deadline'; + break; + } + const meta = sources.get(row.value.sessionId); + if (meta?.dir === undefined || row.value.sessionIdentity === undefined || meta.identity !== row.value.sessionIdentity) { + freshnessStale = true; + continue; + } + if (!identities.has(meta.dir)) { + try { + identities.set(meta.dir, await sessionDirectoryIdentity(meta.dir)); + } catch (error) { + throw new GlobalSearchError('index_unavailable', `cannot verify search source: ${errorMessage(error)}`); + } + } + if (identities.get(meta.dir) === row.value.sessionIdentity) { + visible.push({ ...row, value: { ...row.value, sessionTitle: meta.title ?? '' } }); + } else { + freshnessStale = true; + } + } + const { pageRows, hasMore } = paginateRows(q, page, visible); return { kind: 'page', rows: pageRows, hasMore, incomplete, generation, - index: this.readIndexView(serveDb, freshnessStale), + index: { ...index, freshnessStale: freshnessStale || this.db !== serveDb }, }; } diff --git a/packages/kap-server/src/search/searchService.ts b/packages/kap-server/src/search/searchService.ts index fd14272d6f0..71bb58c67e1 100644 --- a/packages/kap-server/src/search/searchService.ts +++ b/packages/kap-server/src/search/searchService.ts @@ -620,7 +620,7 @@ export class GlobalSearchService implements IGlobalSearchService { return { sessionId: doc.sessionId, workspaceId: doc.workspaceId, - sessionTitle: this.summaries.get(doc.sessionId)?.title ?? doc.sessionTitle, + sessionTitle: doc.sessionTitle, agentId: doc.agentId, role: doc.role, snippet: diff --git a/packages/kap-server/test/search/searchRoute.test.ts b/packages/kap-server/test/search/searchRoute.test.ts index 7b7292c2709..92df53fefd5 100644 --- a/packages/kap-server/test/search/searchRoute.test.ts +++ b/packages/kap-server/test/search/searchRoute.test.ts @@ -96,6 +96,7 @@ describe('server-v2 /api/v1/search', () => { ].join('\n') + '\n', 'utf8', ); + await writeFile(join(home, 'sessions', WS, 's1', 'state.json'), JSON.stringify({ title: '苹果询价' })); const summaries: SessionSummary[] = [ { id: 's1', diff --git a/packages/kap-server/test/search/searchService.test.ts b/packages/kap-server/test/search/searchService.test.ts index 497faea16e5..c56f9fd429f 100644 --- a/packages/kap-server/test/search/searchService.test.ts +++ b/packages/kap-server/test/search/searchService.test.ts @@ -1,6 +1,7 @@ import { createHash } from 'node:crypto'; -import { appendFile, mkdir, mkdtemp, readFile, rm, stat, writeFile } from 'node:fs/promises'; +import fs, { appendFile, mkdir, mkdtemp, readFile, rm, stat, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; +import { syncBuiltinESMExports } from 'node:module'; import { join } from 'node:path'; import { monitorEventLoopDelay, performance } from 'node:perf_hooks'; import { Worker } from 'node:worker_threads'; @@ -118,6 +119,12 @@ async function writeWire( return file; } +async function writeTitle(home: string, sessionId: string, title: string): Promise { + const dir = join(home, 'sessions', WS, sessionId); + await mkdir(dir, { recursive: true }); + await writeFile(join(dir, 'state.json'), JSON.stringify({ title })); +} + const noopLog = { error: () => {}, warn: () => {}, @@ -276,6 +283,7 @@ describe('GlobalSearchService', () => { it('indexes user and assistant text and finds Chinese and English terms', async () => { const s1 = summary('s1', '搜索重构讨论', T1); + await writeTitle(home!, s1.id, s1.title!); await writeWire(home!, 's1', 'main', [ userLine('帮我看看苹果怎么挑', T1), assistantLine('Here is the apple picking guide.', T2), @@ -303,8 +311,164 @@ describe('GlobalSearchService', () => { expect(injected.items).toEqual([]); }); + it.each([makeService, makeInlineService])('filters deleted sources before pagination without waiting for a writer sync (%#)', async (make) => { + const removed = summary('removed', 'needle title', T3); + await writeTitle(home!, removed.id, removed.title!); + const retained = summary('retained', 'retained', T1); + await writeWire(home!, removed.id, 'main', [userLine('needle deleted', T3)]); + await writeWire(home!, retained.id, 'main', [userLine('needle first', T1), userLine('needle second', T2)]); + const writer = track(make(home!, staticIndex([removed, retained]))); + await writer.reindex(); + const reader = track(make(home!, staticIndex([removed, retained]))); + await settleSync(reader); + expect((await reader.status()).lifecycle.state).toBe('ready'); + expect((await reader.search({ query: 'needle' })).indexState.state).toBe('readonly'); + await rm(join(home!, 'sessions', WS, removed.id), { recursive: true }); + for (const mode of ['terms', 'literal'] as const) { + const first = await reader.search({ query: 'needle', mode, sort: 'time_desc', pageSize: 1 }); + expect(first.items).toHaveLength(1); + expect(first.items[0]?.sessionId).toBe(retained.id); + expect(first.hasMore).toBe(true); + const second = await reader.search({ query: 'needle', mode, sort: 'time_desc', pageSize: 1, pageToken: first.pageToken }); + expect(second.items).toHaveLength(1); + expect(second.items[0]?.sessionId).toBe(retained.id); + expect(second.items[0]?.time).not.toBe(first.items[0]?.time); + expect(second.hasMore).toBe(false); + } + }); + + it.each([makeService, makeInlineService])('does not serve an old incarnation after the same directory is recreated (%#)', async (make) => { + const s1 = summary('s1', 'original title', T1); + await writeTitle(home!, s1.id, s1.title!); + await writeWire(home!, s1.id, 'main', [userLine('original secret', T1)]); + const writer = track(make(home!, staticIndex([s1]))); + await writer.reindex(); + const reader = track(make(home!, staticIndex([s1]))); + await settleSync(reader); + expect((await reader.search({ query: 'original' })).items.length).toBeGreaterThan(0); + await rm(join(home!, 'sessions', WS, s1.id), { recursive: true }); + await writeWire(home!, s1.id, 'main', [userLine('replacement message', T2)]); + await writeTitle(home!, s1.id, 'replacement title'); + expect((await reader.search({ query: 'original' })).items).toEqual([]); + await settleSync(writer); + await refreshNow(reader); + expect((await reader.search({ query: 'replacement', role: 'user' })).items).toHaveLength(1); + expect((await reader.search({ query: 'original' })).items).toEqual([]); + expect((await reader.search({ query: 'replacement', role: 'user' })).items[0]?.sessionTitle).toBe('replacement title'); + }); + + it('keeps a query valid when readonly refresh replaces its handle during source validation', async () => { + const s1 = summary('s1', 'one', T1); + await writeWire(home!, s1.id, 'main', [userLine('needle body', T1)]); + const writer = track(makeInlineService(home!, staticIndex([s1]))); + await writer.reindex(); + const reader = track(makeInlineService(home!, staticIndex([s1]))); + await settleSync(reader); + let enter!: () => void; + let release!: () => void; + const entered = new Promise((resolve) => { enter = resolve; }); + const gate = new Promise((resolve) => { release = resolve; }); + const original = fs.stat.bind(fs); + const intercept = vi.spyOn(fs, 'stat').mockImplementation((async (path, options) => { + const result = await original(path, options); + if (path === join(home!, 'sessions', WS, s1.id) && options?.bigint === true) { + enter(); + await gate; + } + return result; + }) as typeof fs.stat); + syncBuiltinESMExports(); + try { + const searching = reader.search({ query: 'needle' }); + await entered; + await coreOf(writer).db!.compact(); + await refreshNow(reader); + release(); + expect((await searching).items).toHaveLength(1); + } finally { + release(); + intercept.mockRestore(); + syncBuiltinESMExports(); + } + }); + + it('fails a query when its session source cannot be verified', async () => { + const s1 = summary('s1', 'one', T1); + await writeWire(home!, s1.id, 'main', [userLine('needle body', T1)]); + const writer = track(makeInlineService(home!, staticIndex([s1]))); + await writer.reindex(); + const reader = track(makeInlineService(home!, staticIndex([s1]))); + await settleSync(reader); + const original = fs.stat.bind(fs); + const intercept = vi.spyOn(fs, 'stat').mockImplementation((async (path, options) => { + if (path === join(home!, 'sessions', WS, s1.id) && options?.bigint === true) { + throw Object.assign(new Error('source unavailable'), { code: 'EACCES' }); + } + return original(path, options); + }) as typeof fs.stat); + syncBuiltinESMExports(); + try { + await expect(reader.search({ query: 'needle' })).rejects.toMatchObject({ reason: 'index_unavailable' }); + } finally { + intercept.mockRestore(); + syncBuiltinESMExports(); + } + }); + + it('updates the displayed title without rewriting unchanged message documents', async () => { + const s1 = summary('s1', 'stale summary', T1); + await writeWire(home!, s1.id, 'main', [userLine('needle body', T1)]); + await writeTitle(home!, s1.id, 'original title'); + const service = track(makeInlineService(home!, staticIndex([s1]))); + await service.reindex(); + expect((await service.search({ query: 'needle' })).items[0]?.sessionTitle).toBe('original title'); + await writeTitle(home!, s1.id, 'renamed title'); + await settleSync(service); + expect((await service.search({ query: 'needle' })).items[0]?.sessionTitle).toBe('renamed title'); + expect((await service.search({ query: 'original', role: 'title' })).items).toEqual([]); + }); + + it.each([false, true])('keeps indexing healthy messages when primary title metadata is damaged (legacy=%s)', async (legacy) => { + const s1 = summary('s1', 'cached title', T1); + const wire = await writeWire(home!, s1.id, 'main', [userLine('needle initial', T1)]); + await writeTitle(home!, s1.id, 'original title'); + const service = track(makeInlineService(home!, staticIndex([s1]))); + await service.reindex(); + await writeFile(join(home!, 'sessions', WS, s1.id, 'state.json'), '{broken'); + if (legacy) { + const dir = join(home!, 'sessions', WS, s1.id, 'session-meta'); + await mkdir(dir); + await writeFile(join(dir, 'state.json'), JSON.stringify({ title: 'legacy title' })); + } + await appendFile(wire, `${userLine('needle appended', T2)}\n`); + await settleSync(service); + const page = await service.search({ query: 'appended' }); + expect(page.items).toHaveLength(1); + expect(page.items[0]?.sessionTitle).toBe(legacy ? 'legacy title' : ''); + }); + + it('rebuilds legacy indexed documents before trusting their session identity', async () => { + const s1 = summary('s1', 'one', T1); + await writeWire(home!, s1.id, 'main', [userLine('needle legacy', T1)]); + const writer = track(makeInlineService(home!, staticIndex([s1]))); + await writer.reindex(); + const db = coreOf(writer).db!; + for (const row of db.query({ key: { prefix: 's1/' } })) { + const { sessionIdentity: _identity, ...legacy } = row.value; + await db.set(row.key, legacy); + } + await db.set('\0meta\\session\\s1', { kind: 'sessionMeta' }); + const reader = track(makeInlineService(home!, staticIndex([s1]))); + await settleSync(reader); + expect((await reader.search({ query: 'needle' })).items).toEqual([]); + await settleSync(writer); + await refreshNow(reader); + expect((await reader.search({ query: 'needle' })).items).toHaveLength(1); + }); + it('hits session titles as title docs', async () => { const s1 = summary('s1', '季度总结报告', T1); + await writeTitle(home!, s1.id, s1.title!); await writeWire(home!, 's1', 'main', [userLine('随便说点什么', T1)]); const service = track(makeService(home!, staticIndex([s1]))); await service.reindex(); @@ -426,6 +590,7 @@ describe('GlobalSearchService', () => { it('reports indexState building before the first full sync and ready after', async () => { const s1 = summary('s1', 'state', T1); + await writeTitle(home!, s1.id, s1.title!); await writeWire(home!, 's1', 'main', [userLine('苹果 state', T1)]); let release!: () => void; @@ -557,6 +722,7 @@ describe('GlobalSearchService', () => { it('runs a second instance read-only and catches up from the WAL', async () => { const s1 = summary('s1', 'shared', T1); + await writeTitle(home!, s1.id, s1.title!); const file = await writeWire(home!, 's1', 'main', [userLine('苹果 base', T1)]); const index = staticIndex([s1]); @@ -806,6 +972,7 @@ describe('GlobalSearchService', () => { it('assigns transcript step ids to assistant hits; user and title hits carry none', async () => { const s1 = summary('s1', '苹果 steps', T1); + await writeTitle(home!, s1.id, s1.title!); await writeWire(home!, 's1', 'main', [ userLine('苹果 question', T1), stepBeginLine('u1', 1, T1 + 100), @@ -2032,6 +2199,7 @@ describe('GlobalSearchService', () => { it('returns identical literal results on both routes for equivalent data', async () => { const s1 = summary('s1', '无关标题', T1); + await writeTitle(home!, s1.id, s1.title!); await writeWire(home!, 's1', 'main', [ userLine('帮我看看苹果怎么挑', T1), stepBeginLine('u1', 1, T1 + 100), From 3ea3c627e52474830cb931c91f348d6a4e080ffa Mon Sep 17 00:00:00 2001 From: qer Date: Tue, 8 Sep 2026 02:07:00 +0800 Subject: [PATCH 2/6] fix(kap-server): reject unverifiable search source identities --- packages/kap-server/src/search/indexCore.ts | 3 +- .../test/search/searchService.test.ts | 45 +++++++++++++++++++ 2 files changed, 47 insertions(+), 1 deletion(-) diff --git a/packages/kap-server/src/search/indexCore.ts b/packages/kap-server/src/search/indexCore.ts index 09a63f36bb6..93827ca055b 100644 --- a/packages/kap-server/src/search/indexCore.ts +++ b/packages/kap-server/src/search/indexCore.ts @@ -68,7 +68,8 @@ function errorMessage(error: unknown): string { async function sessionDirectoryIdentity(dir: string): Promise { try { const info = await stat(dir, { bigint: true }); - return info.isDirectory() ? `${info.dev}:${info.ino}:${info.birthtimeNs}` : undefined; + if (!info.isDirectory() || info.ino <= 0n || info.birthtimeNs <= 0n) return undefined; + return `${info.dev}:${info.ino}:${info.birthtimeNs}`; } catch (error) { const code = (error as NodeJS.ErrnoException).code; if (code === 'ENOENT' || code === 'ENOTDIR') return undefined; diff --git a/packages/kap-server/test/search/searchService.test.ts b/packages/kap-server/test/search/searchService.test.ts index c56f9fd429f..220ee384448 100644 --- a/packages/kap-server/test/search/searchService.test.ts +++ b/packages/kap-server/test/search/searchService.test.ts @@ -357,6 +357,51 @@ describe('GlobalSearchService', () => { expect((await reader.search({ query: 'replacement', role: 'user' })).items[0]?.sessionTitle).toBe('replacement title'); }); + it.each(['inode', 'birthtime', 'both'])('rejects matching stored identities with unavailable %s and recovers after reindexing', async (missing) => { + const s1 = summary('s1', 'one', T1); + await writeWire(home!, s1.id, 'main', [userLine('original secret', T1)]); + const dir = join(home!, 'sessions', WS, s1.id); + const writer = track(makeInlineService(home!, staticIndex([s1]))); + await writer.reindex(); + const info = await stat(dir, { bigint: true }); + const ino = missing === 'birthtime' ? info.ino : 0n; + const birthtimeNs = missing === 'inode' ? info.birthtimeNs : 0n; + const identity = `${info.dev}:${ino}:${birthtimeNs}`; + const db = coreOf(writer).db!; + for (const row of db.query({ key: { prefix: 's1/' } })) { + await db.set(row.key, { ...row.value, sessionIdentity: identity }); + } + await db.set('\0meta\\session\\s1', { kind: 'sessionMeta', dir, identity }); + const original = fs.stat.bind(fs); + const intercept = vi.spyOn(fs, 'stat').mockImplementation((async (path, options) => { + const result = await original(path, options); + if (path === dir && options?.bigint === true) { + return Object.assign(result, { ino, birthtimeNs }); + } + return result; + }) as typeof fs.stat); + syncBuiltinESMExports(); + const reader = track(makeInlineService(home!, staticIndex([s1]))); + try { + await settleSync(reader); + expect((await reader.search({ query: 'original' })).items).toEqual([]); + await rm(dir, { recursive: true }); + await writeWire(home!, s1.id, 'main', [userLine('replacement message', T2)]); + expect((await reader.search({ query: 'original' })).items).toEqual([]); + await settleSync(writer); + await refreshNow(reader); + expect((await reader.search({ query: 'replacement' })).items).toEqual([]); + expect([...db.query({ key: { prefix: 's1/' } })]).toEqual([]); + } finally { + intercept.mockRestore(); + syncBuiltinESMExports(); + } + await settleSync(writer); + await refreshNow(reader); + expect((await reader.search({ query: 'replacement' })).items).toHaveLength(1); + expect((await reader.search({ query: 'original' })).items).toEqual([]); + }); + it('keeps a query valid when readonly refresh replaces its handle during source validation', async () => { const s1 = summary('s1', 'one', T1); await writeWire(home!, s1.id, 'main', [userLine('needle body', T1)]); From a30e2c89730359c73e39e87a343f8073c1d29e00 Mon Sep 17 00:00:00 2001 From: qer Date: Tue, 8 Sep 2026 02:16:25 +0800 Subject: [PATCH 3/6] fix(kap-server): bound search source checks by the query deadline --- packages/kap-server/src/search/indexCore.ts | 54 ++++++++++++------- .../test/search/searchService.test.ts | 54 +++++++++++++++++++ 2 files changed, 88 insertions(+), 20 deletions(-) diff --git a/packages/kap-server/src/search/indexCore.ts b/packages/kap-server/src/search/indexCore.ts index 93827ca055b..894a94b882f 100644 --- a/packages/kap-server/src/search/indexCore.ts +++ b/packages/kap-server/src/search/indexCore.ts @@ -921,28 +921,42 @@ export class SearchIndexCore { } const identities = new Map(); const visible: MatchedRow[] = []; - for (const row of matched.rows) { - if (Date.now() > budget.deadlineAt) { - incomplete ??= 'deadline'; - break; - } - const meta = sources.get(row.value.sessionId); - if (meta?.dir === undefined || row.value.sessionIdentity === undefined || meta.identity !== row.value.sessionIdentity) { - freshnessStale = true; - continue; - } - if (!identities.has(meta.dir)) { - try { - identities.set(meta.dir, await sessionDirectoryIdentity(meta.dir)); - } catch (error) { - throw new GlobalSearchError('index_unavailable', `cannot verify search source: ${errorMessage(error)}`); + let deadlineTimer: ReturnType | undefined; + const deadline = new Promise((resolve) => { + deadlineTimer = setTimeout(() => resolve(null), Math.max(0, budget.deadlineAt - Date.now())); + deadlineTimer.unref?.(); + }); + try { + for (const row of matched.rows) { + if (Date.now() > budget.deadlineAt) { + incomplete ??= 'deadline'; + break; + } + const meta = sources.get(row.value.sessionId); + if (meta?.dir === undefined || row.value.sessionIdentity === undefined || meta.identity !== row.value.sessionIdentity) { + freshnessStale = true; + continue; + } + if (!identities.has(meta.dir)) { + try { + const identity = await Promise.race([sessionDirectoryIdentity(meta.dir), deadline]); + if (identity === null || Date.now() > budget.deadlineAt) { + incomplete ??= 'deadline'; + break; + } + identities.set(meta.dir, identity); + } catch (error) { + throw new GlobalSearchError('index_unavailable', `cannot verify search source: ${errorMessage(error)}`); + } + } + if (identities.get(meta.dir) === row.value.sessionIdentity) { + visible.push({ ...row, value: { ...row.value, sessionTitle: meta.title ?? '' } }); + } else { + freshnessStale = true; } } - if (identities.get(meta.dir) === row.value.sessionIdentity) { - visible.push({ ...row, value: { ...row.value, sessionTitle: meta.title ?? '' } }); - } else { - freshnessStale = true; - } + } finally { + if (deadlineTimer !== undefined) clearTimeout(deadlineTimer); } const { pageRows, hasMore } = paginateRows(q, page, visible); return { diff --git a/packages/kap-server/test/search/searchService.test.ts b/packages/kap-server/test/search/searchService.test.ts index 220ee384448..6d393c1159e 100644 --- a/packages/kap-server/test/search/searchService.test.ts +++ b/packages/kap-server/test/search/searchService.test.ts @@ -437,6 +437,60 @@ describe('GlobalSearchService', () => { } }); + it.each(['resolve', 'reject'])('returns verified partial results when a source stat outlives the deadline and later %ss', async (outcome) => { + const slow = summary('slow', 'slow', T1); + const healthy = summary('healthy', 'healthy', T2); + await writeWire(home!, slow.id, 'main', [userLine('needle slow extra text', T1)]); + await writeWire(home!, healthy.id, 'main', [userLine('needle', T2)]); + const writer = track(makeInlineService(home!, staticIndex([slow, healthy]))); + await writer.reindex(); + const reader = track(makeInlineService(home!, staticIndex([slow, healthy]))); + await settleSync(reader); + expect((await reader.search({ query: 'needle' })).items).toHaveLength(2); + let enter!: () => void; + let release!: () => void; + let finish!: () => void; + const entered = new Promise((resolve) => { enter = resolve; }); + const gate = new Promise((resolve) => { release = resolve; }); + const finished = new Promise((resolve) => { finish = resolve; }); + const original = fs.stat.bind(fs); + const intercept = vi.spyOn(fs, 'stat').mockImplementation((async (path, options) => { + const result = await original(path, options); + if (path === join(home!, 'sessions', WS, slow.id) && options?.bigint === true) { + enter(); + try { + await gate; + if (outcome === 'reject') throw Object.assign(new Error('disconnected source'), { code: 'EIO' }); + } finally { + finish(); + } + } + return result; + }) as typeof fs.stat); + syncBuiltinESMExports(); + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'Date'] }); + reader.queryDeadlineMs = 100; + let completed: Awaited> | undefined; + const searching = reader.search({ query: 'needle', sort: 'time_desc' }).then((page) => { completed = page; }); + try { + await entered; + await vi.advanceTimersByTimeAsync(100); + expect(completed).toMatchObject({ incomplete: 'deadline', items: [{ sessionId: healthy.id }] }); + expect(completed?.items).toHaveLength(1); + release(); + await finished; + await searching; + expect(completed?.items).toHaveLength(1); + } finally { + release(); + await searching.catch(() => {}); + intercept.mockRestore(); + syncBuiltinESMExports(); + vi.useRealTimers(); + } + expect((await reader.search({ query: 'needle' })).items).toHaveLength(2); + }); + it('fails a query when its session source cannot be verified', async () => { const s1 = summary('s1', 'one', T1); await writeWire(home!, s1.id, 'main', [userLine('needle body', T1)]); From 95c7a63d915baf94b1c0e1015ccb51ce9f2c7712 Mon Sep 17 00:00:00 2001 From: qer Date: Tue, 8 Sep 2026 03:20:43 +0800 Subject: [PATCH 4/6] fix(search): report sessions skipped during indexing --- packages/kap-server/src/search/docs.ts | 1 + packages/kap-server/src/search/indexCore.ts | 15 ++++++++------- .../kap-server/test/search/searchService.test.ts | 6 ++++++ 3 files changed, 15 insertions(+), 7 deletions(-) diff --git a/packages/kap-server/src/search/docs.ts b/packages/kap-server/src/search/docs.ts index 4637338506e..77978159ab5 100644 --- a/packages/kap-server/src/search/docs.ts +++ b/packages/kap-server/src/search/docs.ts @@ -65,6 +65,7 @@ export interface SessionMetaDoc { export interface StatsDoc { readonly kind: 'stats'; + readonly degraded?: string; readonly sessions: number; readonly documents: number; readonly lastIndexedAt: number; diff --git a/packages/kap-server/src/search/indexCore.ts b/packages/kap-server/src/search/indexCore.ts index 894a94b882f..57ad9f2fdc8 100644 --- a/packages/kap-server/src/search/indexCore.ts +++ b/packages/kap-server/src/search/indexCore.ts @@ -488,8 +488,7 @@ export class SearchIndexCore { for (const summary of sessions) { if (this.disposed) return { noop: true, sessions: 0, documents: 0 }; try { - await this.syncSession(db, summary); - indexed++; + if (await this.syncSession(db, summary)) indexed++; } catch (error) { this.log.warn('global search: failed to index session', { sessionId: summary.id, @@ -502,6 +501,7 @@ export class SearchIndexCore { const metaCount = db.query({ key: { prefix: '\0meta\\' }, project: [] }).length; const stats: StatsDoc = { kind: 'stats', + degraded: indexed < sessions.length ? `Skipped ${sessions.length - indexed} session(s) during indexing` : undefined, sessions: indexed, documents: db.size - metaCount, lastIndexedAt: Date.now(), @@ -539,11 +539,11 @@ export class SearchIndexCore { await db.del(SESSION_META_PREFIX + sessionId); } - private async syncSession(db: MiniDb, summary: SyncSessionInput): Promise { + private async syncSession(db: MiniDb, summary: SyncSessionInput): Promise { const identity = await sessionDirectoryIdentity(summary.dir); if (identity === undefined) { await this.deleteSessionDocs(db, summary.id); - return; + return false; } const title = await sessionDirectoryTitle(summary.dir, this.log); const metaKey = SESSION_META_PREFIX + summary.id; @@ -592,6 +592,7 @@ export class SearchIndexCore { } else if (existing !== undefined) { await db.del(titleKey); } + return true; } private async deleteFileDocs(db: MiniDb, meta: FileMetaDoc): Promise { @@ -1017,7 +1018,7 @@ export class SearchIndexCore { generation: this.generation, readOnly: this.db?.readOnly === true, lockToken: this.lockToken, - degraded: this.lastRefreshError?.message, + degraded: this.lastRefreshError?.message ?? (stats?.kind === 'stats' ? stats.degraded : undefined), lifecycle: this.lifecycleState(), }; } @@ -1032,7 +1033,7 @@ export class SearchIndexCore { documents: stats?.kind === 'stats' ? stats.documents : 0, readOnly: handle?.readOnly === true, freshnessStale: true, - degraded: this.lastRefreshError?.message, + degraded: this.lastRefreshError?.message ?? (stats?.kind === 'stats' ? stats.degraded : undefined), lockToken: this.lockToken, }; } @@ -1048,7 +1049,7 @@ export class SearchIndexCore { documents, readOnly: db.readOnly, freshnessStale, - degraded: this.lastRefreshError?.message, + degraded: this.lastRefreshError?.message ?? (stats?.kind === 'stats' ? stats.degraded : undefined), lockToken: this.lockToken, }; } diff --git a/packages/kap-server/test/search/searchService.test.ts b/packages/kap-server/test/search/searchService.test.ts index 6d393c1159e..fbcfd3aa172 100644 --- a/packages/kap-server/test/search/searchService.test.ts +++ b/packages/kap-server/test/search/searchService.test.ts @@ -392,6 +392,10 @@ describe('GlobalSearchService', () => { await refreshNow(reader); expect((await reader.search({ query: 'replacement' })).items).toEqual([]); expect([...db.query({ key: { prefix: 's1/' } })]).toEqual([]); + expect(await writer.status()).toMatchObject({ sessions: 0, degraded: 'Skipped 1 session(s) during indexing' }); + expect((await reader.search({ query: 'replacement' })).indexState).toMatchObject({ + indexedSessions: 0, degraded: 'Skipped 1 session(s) during indexing', + }); } finally { intercept.mockRestore(); syncBuiltinESMExports(); @@ -399,6 +403,8 @@ describe('GlobalSearchService', () => { await settleSync(writer); await refreshNow(reader); expect((await reader.search({ query: 'replacement' })).items).toHaveLength(1); + expect(await writer.status()).toMatchObject({ sessions: 1, degraded: undefined }); + expect((await reader.search({ query: 'replacement' })).indexState.degraded).toBeUndefined(); expect((await reader.search({ query: 'original' })).items).toEqual([]); }); From fedf9b8fab342029cf5ff454e38e2492cd420747 Mon Sep 17 00:00:00 2001 From: qer Date: Tue, 8 Sep 2026 03:36:00 +0800 Subject: [PATCH 5/6] fix(search): bound outstanding source identity checks --- packages/kap-server/src/search/indexCore.ts | 28 ++++++++++++- .../test/search/searchService.test.ts | 41 ++++++++++++++++++- 2 files changed, 67 insertions(+), 2 deletions(-) diff --git a/packages/kap-server/src/search/indexCore.ts b/packages/kap-server/src/search/indexCore.ts index 57ad9f2fdc8..76f6e472cdc 100644 --- a/packages/kap-server/src/search/indexCore.ts +++ b/packages/kap-server/src/search/indexCore.ts @@ -77,6 +77,32 @@ async function sessionDirectoryIdentity(dir: string): Promise void>(); + +async function queryDirectoryIdentity(dir: string, deadlineAt: number, deadline: Promise): Promise { + while (checkingQuerySource) { + let wake!: () => void; + const available = new Promise((resolve) => { wake = () => resolve(true); }); + querySourceWaiters.add(wake); + try { + if (await Promise.race([available, deadline]) === null) return null; + } finally { + querySourceWaiters.delete(wake); + } + } + if (Date.now() >= deadlineAt) return null; + checkingQuerySource = true; + const identity = sessionDirectoryIdentity(dir); + const release = () => { + checkingQuerySource = false; + for (const wake of querySourceWaiters) wake(); + querySourceWaiters.clear(); + }; + void identity.then(release, release); + return Promise.race([identity, deadline]); +} + async function sessionDirectoryTitle(dir: string, log: SearchCoreLog): Promise { for (const scope of ['', 'session-meta']) { try { @@ -940,7 +966,7 @@ export class SearchIndexCore { } if (!identities.has(meta.dir)) { try { - const identity = await Promise.race([sessionDirectoryIdentity(meta.dir), deadline]); + const identity = await queryDirectoryIdentity(meta.dir, budget.deadlineAt, deadline); if (identity === null || Date.now() > budget.deadlineAt) { incomplete ??= 'deadline'; break; diff --git a/packages/kap-server/test/search/searchService.test.ts b/packages/kap-server/test/search/searchService.test.ts index fbcfd3aa172..42d54f73811 100644 --- a/packages/kap-server/test/search/searchService.test.ts +++ b/packages/kap-server/test/search/searchService.test.ts @@ -497,6 +497,46 @@ describe('GlobalSearchService', () => { expect((await reader.search({ query: 'needle' })).items).toHaveLength(2); }); + it('bounds outstanding source checks across timed-out searches and verifies fresh state after recovery', async () => { + const s1 = summary('s1', 'one', T1); + await writeWire(home!, s1.id, 'main', [userLine('needle body', T1)]); + const writer = track(makeInlineService(home!, staticIndex([s1]))); + await writer.reindex(); + const reader = track(makeInlineService(home!, staticIndex([s1]))); + await settleSync(reader); + writer.queryDeadlineMs = 30; + reader.queryDeadlineMs = 30; + let release!: () => void; + const gate = new Promise((resolve) => { release = resolve; }); + let calls = 0; + const dir = join(home!, 'sessions', WS, s1.id); + const original = fs.stat.bind(fs); + const intercept = vi.spyOn(fs, 'stat').mockImplementation((async (path, options) => { + const result = await original(path, options); + if (path === dir && options?.bigint === true) { + calls++; + await gate; + } + return result; + }) as typeof fs.stat); + syncBuiltinESMExports(); + try { + expect(await reader.search({ query: 'needle' })).toMatchObject({ incomplete: 'deadline', items: [] }); + const pages = await Promise.all(Array.from({ length: 8 }, (_, i) => + (i % 2 === 0 ? reader : writer).search({ query: 'needle' }))); + for (const page of pages) expect(page).toMatchObject({ incomplete: 'deadline', items: [] }); + expect(calls).toBe(1); + await rm(dir, { recursive: true }); + release(); + expect((await reader.search({ query: 'needle' })).items).toEqual([]); + } finally { + release(); + intercept.mockRestore(); + syncBuiltinESMExports(); + } + expect((await writer.search({ query: 'needle' })).items).toEqual([]); + }); + it('fails a query when its session source cannot be verified', async () => { const s1 = summary('s1', 'one', T1); await writeWire(home!, s1.id, 'main', [userLine('needle body', T1)]); @@ -3272,4 +3312,3 @@ describe('search lifecycle diagnostics (stage 5)', () => { expect(status.degraded).toContain('worker'); }); }); - From c3b0da012720e016469c4bff04f4c361119b0dc3 Mon Sep 17 00:00:00 2001 From: qer Date: Tue, 8 Sep 2026 03:36:20 +0800 Subject: [PATCH 6/6] refactor(search): group source check coordination state --- packages/kap-server/src/search/indexCore.ts | 19 +++++++++---------- 1 file changed, 9 insertions(+), 10 deletions(-) diff --git a/packages/kap-server/src/search/indexCore.ts b/packages/kap-server/src/search/indexCore.ts index 76f6e472cdc..ec3e5f0d1e1 100644 --- a/packages/kap-server/src/search/indexCore.ts +++ b/packages/kap-server/src/search/indexCore.ts @@ -77,27 +77,26 @@ async function sessionDirectoryIdentity(dir: string): Promise void>(); +const querySourceChecks = { running: false, waiters: new Set<() => void>() }; async function queryDirectoryIdentity(dir: string, deadlineAt: number, deadline: Promise): Promise { - while (checkingQuerySource) { + while (querySourceChecks.running) { let wake!: () => void; - const available = new Promise((resolve) => { wake = () => resolve(true); }); - querySourceWaiters.add(wake); + const available = new Promise((resolve) => { wake = () => { resolve(true); }; }); + querySourceChecks.waiters.add(wake); try { if (await Promise.race([available, deadline]) === null) return null; } finally { - querySourceWaiters.delete(wake); + querySourceChecks.waiters.delete(wake); } } if (Date.now() >= deadlineAt) return null; - checkingQuerySource = true; + querySourceChecks.running = true; const identity = sessionDirectoryIdentity(dir); const release = () => { - checkingQuerySource = false; - for (const wake of querySourceWaiters) wake(); - querySourceWaiters.clear(); + querySourceChecks.running = false; + for (const wake of querySourceChecks.waiters) wake(); + querySourceChecks.waiters.clear(); }; void identity.then(release, release); return Promise.race([identity, deadline]);