From 8fd68731bab2696abfba3a2d38fd5c47c27811d3 Mon Sep 17 00:00:00 2001 From: Hypatia May Date: Sat, 11 Jul 2026 12:02:51 +0800 Subject: [PATCH] fix(session-query): harden provider operation lifecycle --- docs/cordis-catalog/services.md | 2 +- .../cordis/tool-cordis/src/api-catalog.ts | 2 +- .../session-query/session-query/README.md | 2 +- .../session-query/session-query/src/index.ts | 4 +- .../session-query/src/provider.ts | 52 ++++++------ .../session-query/tests/session-query.spec.ts | 83 ++++++++++++++++++- 6 files changed, 114 insertions(+), 31 deletions(-) diff --git a/docs/cordis-catalog/services.md b/docs/cordis-catalog/services.md index 04d07f5795..c8d083593e 100644 --- a/docs/cordis-catalog/services.md +++ b/docs/cordis-catalog/services.md @@ -179,7 +179,7 @@ async listEvents(sessionId: SessionId): Promise async readEvent(request: SessionEventReadRequest): Promise async traceSession(sessionId: SessionId): Promise async traceEvent(sessionId: SessionId, seq: number): Promise -registerSearchProvider(provider: SessionSearchProvider): () => void +registerSearchProvider(provider: SessionSearchProvider): () => Promise registerEventTextExtractor( type: K, extractor: SessionEventTextExtractor, ): () => void registerContentTextExtractor( type: K, extractor: SessionContentTextExtractor, ): () => void searchSessions( request: SessionSearchRequest, exec?: SessionQueryExecContext, ): Promise> diff --git a/packages/cordis/tool-cordis/src/api-catalog.ts b/packages/cordis/tool-cordis/src/api-catalog.ts index 12872d689f..2179326c49 100644 --- a/packages/cordis/tool-cordis/src/api-catalog.ts +++ b/packages/cordis/tool-cordis/src/api-catalog.ts @@ -144,7 +144,7 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [ 'async readEvent(request: SessionEventReadRequest): Promise', 'async traceSession(sessionId: SessionId): Promise', 'async traceEvent(sessionId: SessionId, seq: number): Promise', - 'registerSearchProvider(provider: SessionSearchProvider): () => void', + 'registerSearchProvider(provider: SessionSearchProvider): () => Promise', 'registerEventTextExtractor( type: K, extractor: SessionEventTextExtractor, ): () => void', 'registerContentTextExtractor( type: K, extractor: SessionContentTextExtractor, ): () => void', 'searchSessions( request: SessionSearchRequest, exec?: SessionQueryExecContext, ): Promise>', diff --git a/packages/session-query/session-query/README.md b/packages/session-query/session-query/README.md index 5dc5aecc34..47977e6f7a 100644 --- a/packages/session-query/session-query/README.md +++ b/packages/session-query/session-query/README.md @@ -22,7 +22,7 @@ Session filters cover id, exact cwd, inclusive creation time, parent id/root, an ## Full-text providers -`registerSearchProvider(provider)` is effect-scoped and ids are unique. Without `searchProvider`, exactly one locally available provider must be registered; explicit selection fails loudly when the named provider is missing or unavailable. Search pages default to 20 hits and reject limits above 100; a provider returning more hits than the normalized request limit fails with a typed provider error rather than silently dropping cursor-addressable results. Provider scores never cross the public API: event hits carry a plain snippet, while each session hit carries exactly one best matching event. +`registerSearchProvider(provider)` is effect-scoped and ids are unique. Its async disposer removes the provider from selection immediately, lets already accepted transactions finish, and settles after they drain. Without `searchProvider`, exactly one locally available provider must be registered; explicit selection fails loudly when the named provider is missing or unavailable. Search pages default to 20 hits and reject limits above 100; a provider returning more hits than the normalized request limit fails with a typed provider error rather than silently dropping cursor-addressable results. Provider scores never cross the public API: event hits carry a plain snippet, while each session hit carries exactly one best matching event. The service feeds providers two independent layers: a durable persisted base (`persistedInventory`, `replacePersisted`, `removePersisted`, `setPersistedActive`) and an ephemeral live override (`replaceLive`, `removeLive`). A search waits for the relevant source state observed before its call: the whole corpus for session search, only the target for a live event search. Failed derived updates do not fail session writes; affected searches receive `SESSION_QUERY_INDEX_FAILED`, and a later search retries the dirty state. `AbortSignal` lets a caller stop waiting and is also passed to provider search. diff --git a/packages/session-query/session-query/src/index.ts b/packages/session-query/session-query/src/index.ts index edab597dd5..c07c0b8d05 100644 --- a/packages/session-query/session-query/src/index.ts +++ b/packages/session-query/session-query/src/index.ts @@ -151,9 +151,9 @@ export class SessionQueryService extends Service { /** * Register one full-text provider with effect-scoped disposal. * @param provider - provider and synchronization implementation. - * @returns disposer that unregisters the provider. + * @returns async disposer that immediately unregisters selection and awaits accepted provider work. */ - registerSearchProvider(provider: SessionSearchProvider): () => void { + registerSearchProvider(provider: SessionSearchProvider): () => Promise { return this._providers.register(this.ctx, provider) } diff --git a/packages/session-query/session-query/src/provider.ts b/packages/session-query/session-query/src/provider.ts index c3ec4ad0f4..e4235c6d8a 100644 --- a/packages/session-query/session-query/src/provider.ts +++ b/packages/session-query/session-query/src/provider.ts @@ -23,7 +23,6 @@ import { filterEventResults, filterSessionResults } from './filters.ts' interface ProviderState { provider: SessionSearchProvider - active: boolean chain: Promise liveIds: Set } @@ -49,26 +48,25 @@ export class SessionProviderCoordinator { * Register one effect-scoped provider. * @param ctx - contributing caller context. * @param provider - provider implementation. - * @returns disposer for the registration. + * @returns async disposer that deselects immediately and drains accepted work. */ - register(ctx: Context, provider: SessionSearchProvider): () => void { + register(ctx: Context, provider: SessionSearchProvider): () => Promise { if (this._providers.has(provider.id)) { throw new SessionQueryError(`a session-query provider with id "${provider.id}" is already registered`, 'SESSION_QUERY_DUPLICATE_PROVIDER') } const state: ProviderState = { provider, - active: true, chain: Promise.resolve(), liveIds: new Set(), } const dispose = ctx.effect(function* (this: SessionProviderCoordinator) { this._providers.set(provider.id, state) - yield () => { - state.active = false + yield async () => { this._providers.delete(provider.id) + await state.chain } }.bind(this), 'sessionQuery.registerSearchProvider()') - return () => void dispose() + return async () => { await dispose() } } /** @@ -83,7 +81,7 @@ export class SessionProviderCoordinator { ): Promise> { const state = this._resolveProvider() const normalized = this._normalizeSessionSearch(request) - const work = this._runFullSearch(state, async () => { + const work = this._runFullSearch(state, undefined, async () => { if (exec?.signal?.aborted) throw aborted() const result = await state.provider.searchSessions(normalized, exec) return this._validateSearchPage(state, result, normalized.limit) @@ -113,22 +111,25 @@ export class SessionProviderCoordinator { if (live !== undefined) { work = this._runLiveSearch(state, live, query) } else { - const persistence = await this._corpus().persistenceView() - if (persistence === undefined || !persistence.headers.some(header => header.id === request.sessionId)) { - throw new SessionQueryError(`session "${request.sessionId}" not found`, 'SESSION_QUERY_SESSION_NOT_FOUND') - } - work = this._runFullSearch(state, query) + work = this._runFullSearch(state, request.sessionId, query) } return waitFor(work, exec?.signal) } - private _runFullSearch(state: ProviderState, query: () => Promise): Promise { + private _runFullSearch( + state: ProviderState, + requiredSessionId: SessionId | undefined, + query: () => Promise, + ): Promise { const liveSessions = this._corpus().listLive() return this._serialize(state, async () => { await this._synchronize(state, async () => { - /* v8 ignore next -- a provider can be disposed while queued behind an in-flight update */ - if (!state.active) return const persistence = await this._corpus().persistenceView() + const missingRequired = requiredSessionId !== undefined + && (persistence === undefined || !persistence.headers.some(header => header.id === requiredSessionId)) + if (missingRequired) { + throw new SessionQueryError(`session "${requiredSessionId}" not found`, 'SESSION_QUERY_SESSION_NOT_FOUND') + } if (persistence === undefined) { await state.provider.setPersistedActive(false) } else { @@ -172,8 +173,6 @@ export class SessionProviderCoordinator { } return this._serialize(state, async () => { await this._synchronize(state, async () => { - /* v8 ignore next -- a provider can be disposed while queued behind an in-flight update */ - if (!state.active) return await state.provider.replaceLive(snapshot) state.liveIds.add(session.id) }) @@ -272,32 +271,35 @@ export class SessionProviderCoordinator { } function waitFor(work: Promise, signal: AbortSignal | undefined): Promise { - if (signal === undefined) return work + const observed = work.catch((error: unknown) => { throw operationError(error) }) + if (signal === undefined) return observed if (signal.aborted) { // Cancellation supersedes the caller's result, but shared work must still // have a rejection observer when it has already failed synchronously. - void work.catch((_supersededError: unknown) => undefined) + void observed.catch((_supersededError: unknown) => undefined) return Promise.reject(aborted()) } return new Promise((resolve, reject) => { const onAbort = () => { reject(aborted()) } signal.addEventListener('abort', onAbort, { once: true }) - work.then( + observed.then( (value) => { signal.removeEventListener('abort', onAbort) resolve(value) }, (error: unknown) => { signal.removeEventListener('abort', onAbort) - /* v8 ignore next -- Promise contracts reject with Error; retain a typed boundary for third-party providers */ - reject(error instanceof Error - ? error - : new SessionQueryError('session-query operation failed with a non-Error rejection', 'SESSION_QUERY_PROVIDER_ERROR', { cause: error })) + reject(operationError(error)) }, ) }) } +function operationError(error: unknown): Error { + if (error instanceof Error) return error + return new SessionQueryError('session-query operation failed with a non-Error rejection', 'SESSION_QUERY_PROVIDER_ERROR', { cause: error }) +} + function aborted(): SessionQueryError { return new SessionQueryError('session-query operation aborted', 'SESSION_QUERY_ABORTED') } diff --git a/packages/session-query/session-query/tests/session-query.spec.ts b/packages/session-query/session-query/tests/session-query.spec.ts index 642bfb263f..36a8676640 100644 --- a/packages/session-query/session-query/tests/session-query.spec.ts +++ b/packages/session-query/session-query/tests/session-query.spec.ts @@ -400,10 +400,37 @@ describe('provider selection and synchronization', () => { await expect(ctx.sessionQuery.searchEvents({ sessionId: session.id, query: 'x', limit: 1 }, { signal: new AbortController().signal })) .resolves.toMatchObject({ providerId: provider.id }) - dispose() + await dispose() await expect(ctx.sessionQuery.searchSessions({ query: 'x' })).rejects.toThrow(expectCode('SESSION_QUERY_PROVIDER_UNAVAILABLE')) }) + it('deselects immediately and drains accepted work before disposal settles', async () => { + const ctx = await liveContext() + const provider = new FakeProvider() + const queryStarted = deferred() + const releaseQuery = deferred() + provider.searchSessions = async () => { + queryStarted.resolve() + await releaseQuery.promise + return { providerId: provider.id, items: [] } + } + const dispose = ctx.sessionQuery.registerSearchProvider(provider) + + const accepted = ctx.sessionQuery.searchSessions({ query: 'accepted' }) + await queryStarted.promise + let disposed = false + const disposal = dispose().then(() => { disposed = true }) + await expect(ctx.sessionQuery.searchSessions({ query: 'future' })) + .rejects.toThrow(expectCode('SESSION_QUERY_PROVIDER_UNAVAILABLE')) + await Promise.resolve() + expect(disposed).toBe(false) + + releaseQuery.resolve() + await expect(accepted).resolves.toMatchObject({ providerId: provider.id }) + await disposal + expect(disposed).toBe(true) + }) + it('serializes concurrent synchronization and supports cancellation while provider search is pending', async () => { const ctx = await liveContext() const session = ctx.sessions.create(SessionId('coalesce')) @@ -559,6 +586,60 @@ describe('provider selection and synchronization', () => { expect(provider.persisted.get(persisted.id)?.documents[0]?.text).toBe('persisted target') }) + it('cancels a persisted-only event search while persistence listing is blocked', async () => { + const persisted = header('blocked-persisted-target', 1) + TestPersistence.reset([{ meta: persisted, events: eventLog('persisted target') }]) + const listStarted = deferred() + const releaseList = deferred() + TestPersistence.onList = listStarted.resolve + TestPersistence.listBarrier = releaseList.promise + const ctx = await liveContext() + const persistenceFiber = await ctx.plugin(TestPersistence) + await listStarted.promise + const provider = new FakeProvider() + const disposeProvider = ctx.sessionQuery.registerSearchProvider(provider) + const controller = new AbortController() + + const pending = ctx.sessionQuery.searchEvents( + { sessionId: persisted.id, query: 'target' }, + { signal: controller.signal }, + ) + controller.abort() + await expect(pending).rejects.toThrow(expectCode('SESSION_QUERY_ABORTED')) + expect(provider.eventRequests).toEqual([]) + + releaseList.resolve() + await disposeProvider() + await persistenceFiber.dispose() + TestPersistence.listBarrier = undefined + TestPersistence.onList = undefined + }) + + it('normalizes non-Error query rejections and preserves Error identity', async () => { + const ctx = await liveContext() + const provider = new FakeProvider() + ctx.sessionQuery.registerSearchProvider(provider) + const signals = [undefined, new AbortController().signal] + + for (const [index, signal] of signals.entries()) { + const exec = signal === undefined ? undefined : { signal } + const identity = new Error(`query failure ${index}`) + provider.searchSessions = () => Promise.reject(identity) + const preserved = await ctx.sessionQuery.searchSessions({ query: 'x' }, exec) + .then(() => undefined, (error: unknown) => error) + expect(preserved).toBe(identity) + + const rejection = { index } + // Deliberately violate the Promise convention to test the provider boundary. + // eslint-disable-next-line @typescript-eslint/prefer-promise-reject-errors + provider.searchSessions = () => Promise.reject(rejection) + const normalized = await ctx.sessionQuery.searchSessions({ query: 'x' }, exec) + .then(() => undefined, (error: unknown) => error) + expect(normalized).toBeInstanceOf(SessionQueryError) + expect(normalized).toMatchObject({ code: 'SESSION_QUERY_PROVIDER_ERROR', cause: rejection }) + } + }) + it('fails loudly for duplicate, configured, unavailable, and ambiguous providers', async () => { const ctx = await liveContext() const first = new FakeProvider('first')