From 5b39d77e70d3edee6b9006892e85d3c215f9cf11 Mon Sep 17 00:00:00 2001 From: MyPrototypeWhat Date: Fri, 3 Jul 2026 14:19:22 +0800 Subject: [PATCH 1/9] =?UTF-8?q?feat(core):=20resilience=20primitives=20?= =?UTF-8?q?=E2=80=94=20withTimeout=20+=20retryingFetch?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/src/__tests__/resilience.test.ts | 94 +++++++++++++++++++ packages/core/src/index.ts | 2 + packages/core/src/resilience.ts | 80 ++++++++++++++++ 3 files changed, 176 insertions(+) create mode 100644 packages/core/src/__tests__/resilience.test.ts create mode 100644 packages/core/src/resilience.ts diff --git a/packages/core/src/__tests__/resilience.test.ts b/packages/core/src/__tests__/resilience.test.ts new file mode 100644 index 0000000..d03f827 --- /dev/null +++ b/packages/core/src/__tests__/resilience.test.ts @@ -0,0 +1,94 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { retryingFetch, withTimeout } from '../resilience' + +const okResponse = () => new Response('ok', { status: 200 }) +const status = (s: number) => new Response('x', { status: s }) + +describe('withTimeout', () => { + beforeEach(() => vi.useFakeTimers()) + afterEach(() => vi.useRealTimers()) + + it('aborts the signal and rejects `expired` after timeoutMs', async () => { + const t = withTimeout(undefined, 1000) + expect(t.signal.aborted).toBe(false) + const raced = Promise.race([new Promise(() => {}), t.expired]).catch(e => e) + await vi.advanceTimersByTimeAsync(1000) + expect(t.signal.aborted).toBe(true) + expect(String(await raced)).toContain('timeout after 1000ms') + t.cancel() + }) + + it('propagates a parent abort immediately', async () => { + const parent = new AbortController() + const t = withTimeout(parent.signal, 60_000) + parent.abort() + expect(t.signal.aborted).toBe(true) + t.cancel() + }) + + it('cancel() clears the timer so nothing fires later', async () => { + const t = withTimeout(undefined, 1000) + t.cancel() + await vi.advanceTimersByTimeAsync(5000) + expect(t.signal.aborted).toBe(false) + }) +}) + +describe('retryingFetch', () => { + beforeEach(() => vi.useFakeTimers()) + afterEach(() => vi.useRealTimers()) + + it('retries a 500 then returns the success', async () => { + const impl = vi.fn().mockResolvedValueOnce(status(500)).mockResolvedValueOnce(okResponse()) + const f = retryingFetch(impl as unknown as typeof fetch, { retries: 1 }) + const p = f('https://x/') + await vi.runAllTimersAsync() + expect((await p).status).toBe(200) + expect(impl).toHaveBeenCalledTimes(2) + }) + + it('retries 429 and a rejected network error', async () => { + const impl = vi.fn() + .mockResolvedValueOnce(status(429)) + .mockRejectedValueOnce(new TypeError('fetch failed')) + .mockResolvedValueOnce(okResponse()) + const f = retryingFetch(impl as unknown as typeof fetch, { retries: 2 }) + const p = f('https://x/') + await vi.runAllTimersAsync() + expect((await p).status).toBe(200) + expect(impl).toHaveBeenCalledTimes(3) + }) + + it('returns the last 5xx response (does not throw) once retries are exhausted', async () => { + const impl = vi.fn().mockResolvedValue(status(503)) + const f = retryingFetch(impl as unknown as typeof fetch, { retries: 1 }) + const p = f('https://x/') + await vi.runAllTimersAsync() + expect((await p).status).toBe(503) + expect(impl).toHaveBeenCalledTimes(2) + }) + + it('does not retry non-retryable statuses (400/404)', async () => { + const impl = vi.fn().mockResolvedValue(status(404)) + const f = retryingFetch(impl as unknown as typeof fetch, { retries: 3 }) + expect((await f('https://x/')).status).toBe(404) + expect(impl).toHaveBeenCalledTimes(1) + }) + + it('does not retry after an abort; abort during backoff cancels the wait', async () => { + const ac = new AbortController() + const abortErr = Object.assign(new Error('aborted'), { name: 'AbortError' }) + const impl = vi.fn().mockRejectedValue(abortErr) + const f = retryingFetch(impl as unknown as typeof fetch, { retries: 3 }) + await expect(f('https://x/', { signal: ac.signal })).rejects.toMatchObject({ name: 'AbortError' }) + expect(impl).toHaveBeenCalledTimes(1) + + const impl2 = vi.fn().mockResolvedValue(status(500)) + const f2 = retryingFetch(impl2 as unknown as typeof fetch, { retries: 3 }) + const p2 = f2('https://x/', { signal: ac.signal }).catch(e => e) + ac.abort() + await vi.runAllTimersAsync() + await p2 + expect(impl2).toHaveBeenCalledTimes(1) // aborted before/during first backoff — no second attempt + }) +}) diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 59aa547..6b59849 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -64,3 +64,5 @@ export type { } from './client' export { lexicalReranker, tokenize } from './rerank' export type { Reranker, RerankInput, LexicalRerankOptions } from './rerank' +export { withTimeout, retryingFetch } from './resilience' +export type { TimeoutHandle, RetryOptions } from './resilience' diff --git a/packages/core/src/resilience.ts b/packages/core/src/resilience.ts new file mode 100644 index 0000000..b78deb6 --- /dev/null +++ b/packages/core/src/resilience.ts @@ -0,0 +1,80 @@ +// Resilience primitives for the search orchestrator (H8/H9). Pure: the fetch +// implementation is always injected; core never references a global fetch. + +export interface TimeoutHandle { + /** Aborts when the parent aborts OR the timer fires. Pass as ProviderContext.signal. */ + signal: AbortSignal + /** Rejects with `timeout after Nms` when the timer fires. Race the provider against it. */ + expired: Promise + /** Clear the timer + detach the parent listener. ALWAYS call once settled. */ + cancel(): void +} + +/** Compose a parent signal with a deadline — manual composition, no AbortSignal.any + * (H9: keeps core runtime-agnostic). */ +export function withTimeout(parent: AbortSignal | undefined, timeoutMs: number): TimeoutHandle { + const ctrl = new AbortController() + const onParentAbort = () => ctrl.abort(parent?.reason) + if (parent?.aborted) ctrl.abort(parent.reason) + else parent?.addEventListener('abort', onParentAbort, { once: true }) + + let rejectExpired: (e: Error) => void + const expired = new Promise((_, reject) => { rejectExpired = reject }) + expired.catch(() => {}) // the race may settle first; never let this become an unhandled rejection + const timer = setTimeout(() => { + const err = new Error(`timeout after ${timeoutMs}ms`) + ctrl.abort(err) + rejectExpired(err) + }, timeoutMs) + + return { + signal: ctrl.signal, + expired, + cancel() { + clearTimeout(timer) + parent?.removeEventListener('abort', onParentAbort) + }, + } +} + +export interface RetryOptions { + /** Extra attempts after the first (H8 default 1). */ + retries: number + /** Base backoff delay; grows 2^attempt with full jitter. Default 250. */ + baseDelayMs?: number +} + +function abortAware(ms: number, signal: AbortSignal | null | undefined): Promise { + return new Promise((resolve, reject) => { + if (signal?.aborted) return reject(signal.reason ?? new Error('aborted')) + const onAbort = () => { clearTimeout(t); reject(signal?.reason ?? new Error('aborted')) } + const t = setTimeout(() => { signal?.removeEventListener('abort', onAbort); resolve() }, ms) + signal?.addEventListener('abort', onAbort, { once: true }) + }) +} + +const isRetryableStatus = (s: number) => s === 429 || s >= 500 + +/** Wrap an injected fetch with bounded retries on 429/5xx/network-error (H8). + * Aborts are never retried; an exhausted 429/5xx returns the response so the + * provider's own `!res.ok` error path still owns the failure message. */ +export function retryingFetch(fetchImpl: typeof fetch, opts: RetryOptions): typeof fetch { + const base = opts.baseDelayMs ?? 250 + const wrapped = async ( + input: Parameters[0], + init?: Parameters[1], + ): Promise => { + for (let attempt = 0; ; attempt++) { + try { + const res = await fetchImpl(input, init) + if (!isRetryableStatus(res.status) || attempt >= opts.retries) return res + } catch (err) { + const name = (err as { name?: string } | null)?.name + if (name === 'AbortError' || init?.signal?.aborted || attempt >= opts.retries) throw err + } + // exponential backoff with full jitter; an abort during the wait cancels the retry + await abortAware(base * 2 ** attempt * (0.5 + Math.random() * 0.5), init?.signal) + } + } + return wrapped as typeof fetch +} From d4081e0833fc6a4fd5f344d37da4eb32573b3af5 Mon Sep 17 00:00:00 2001 From: MyPrototypeWhat Date: Fri, 3 Jul 2026 14:30:35 +0800 Subject: [PATCH 2/9] fix(core): drain discarded retry responses; self-clean withTimeout on parent abort --- packages/core/src/__tests__/resilience.test.ts | 12 ++++++++++++ packages/core/src/resilience.ts | 14 +++++++++++--- 2 files changed, 23 insertions(+), 3 deletions(-) diff --git a/packages/core/src/__tests__/resilience.test.ts b/packages/core/src/__tests__/resilience.test.ts index d03f827..49fd79b 100644 --- a/packages/core/src/__tests__/resilience.test.ts +++ b/packages/core/src/__tests__/resilience.test.ts @@ -47,6 +47,18 @@ describe('retryingFetch', () => { expect(impl).toHaveBeenCalledTimes(2) }) + it('drains (cancels) the body of a discarded retryable response', async () => { + const res500 = status(500) + expect(res500.body).not.toBeNull() // guard: the spy below must target a real stream + const cancelSpy = vi.spyOn(res500.body!, 'cancel') + const impl = vi.fn().mockResolvedValueOnce(res500).mockResolvedValueOnce(okResponse()) + const f = retryingFetch(impl as unknown as typeof fetch, { retries: 1 }) + const p = f('https://x/') + await vi.runAllTimersAsync() + expect((await p).status).toBe(200) + expect(cancelSpy).toHaveBeenCalledTimes(1) + }) + it('retries 429 and a rejected network error', async () => { const impl = vi.fn() .mockResolvedValueOnce(status(429)) diff --git a/packages/core/src/resilience.ts b/packages/core/src/resilience.ts index b78deb6..772cade 100644 --- a/packages/core/src/resilience.ts +++ b/packages/core/src/resilience.ts @@ -14,14 +14,18 @@ export interface TimeoutHandle { * (H9: keeps core runtime-agnostic). */ export function withTimeout(parent: AbortSignal | undefined, timeoutMs: number): TimeoutHandle { const ctrl = new AbortController() - const onParentAbort = () => ctrl.abort(parent?.reason) + let timer: ReturnType | undefined + const onParentAbort = () => { + clearTimeout(timer) // self-clean: a parent abort makes the deadline moot + ctrl.abort(parent?.reason) + } if (parent?.aborted) ctrl.abort(parent.reason) else parent?.addEventListener('abort', onParentAbort, { once: true }) let rejectExpired: (e: Error) => void const expired = new Promise((_, reject) => { rejectExpired = reject }) expired.catch(() => {}) // the race may settle first; never let this become an unhandled rejection - const timer = setTimeout(() => { + timer = setTimeout(() => { const err = new Error(`timeout after ${timeoutMs}ms`) ctrl.abort(err) rejectExpired(err) @@ -38,7 +42,7 @@ export function withTimeout(parent: AbortSignal | undefined, timeoutMs: number): } export interface RetryOptions { - /** Extra attempts after the first (H8 default 1). */ + /** Extra attempts after the first. The orchestrator passes its own default (H8: 1). */ retries: number /** Base backoff delay; grows 2^attempt with full jitter. Default 250. */ baseDelayMs?: number @@ -65,13 +69,17 @@ export function retryingFetch(fetchImpl: typeof fetch, opts: RetryOptions): type init?: Parameters[1], ): Promise => { for (let attempt = 0; ; attempt++) { + let discarded: Response | undefined try { const res = await fetchImpl(input, init) if (!isRetryableStatus(res.status) || attempt >= opts.retries) return res + discarded = res } catch (err) { const name = (err as { name?: string } | null)?.name if (name === 'AbortError' || init?.signal?.aborted || attempt >= opts.retries) throw err } + // drain the discarded body so undici can reuse the socket during retries + void discarded?.body?.cancel().catch(() => {}) // exponential backoff with full jitter; an abort during the wait cancels the retry await abortAware(base * 2 ** attempt * (0.5 + Math.random() * 0.5), init?.signal) } From d8e99ca755cd7d9c7517ef175d860683240d0069 Mon Sep 17 00:00:00 2001 From: MyPrototypeWhat Date: Fri, 3 Jul 2026 14:37:33 +0800 Subject: [PATCH 3/9] feat(core): per-provider timeout, retrying fetch, latencyMs in search meta --- packages/core/src/__tests__/client.test.ts | 79 ++++++++++++++- packages/core/src/client.ts | 108 +++++++++++++-------- packages/core/src/index.ts | 1 + 3 files changed, 144 insertions(+), 44 deletions(-) diff --git a/packages/core/src/__tests__/client.test.ts b/packages/core/src/__tests__/client.test.ts index f043d30..a21a519 100644 --- a/packages/core/src/__tests__/client.test.ts +++ b/packages/core/src/__tests__/client.test.ts @@ -88,16 +88,21 @@ describe('createRefkit', () => { }) it('defaults fetch to globalThis.fetch when options.fetch is omitted', async () => { + // resilience defaults ON (H8), so ctx.fetch is a retrying wrapper rather than + // globalThis.fetch itself — assert on the underlying implementation it delegates to. + const globalFetchSpy = vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response('ok', { status: 200 })) let capturedFetch: typeof fetch | undefined const capturingProvider = defineProvider({ id: 'cap', modalities: ['image'], queryFeatures: ['keyword'], - search: async (_q, ctx) => { capturedFetch = ctx.fetch; return [] }, + search: async (_q, ctx) => { capturedFetch = ctx.fetch; await ctx.fetch('https://cap/x'); return [] }, }) const rk = createRefkit({ providers: [capturingProvider] }) await rk.search({ query: 'x', modalities: ['image'] }) - expect(capturedFetch).toBe(globalThis.fetch) + expect(capturedFetch).not.toBe(globalThis.fetch) // wrapped by retryingFetch + expect(globalFetchSpy.mock.calls[0]?.[0]).toBe('https://cap/x') + globalFetchSpy.mockRestore() }) it('throws a clear Error (not AggregateError) when no provider supports the requested modality', async () => { @@ -215,8 +220,8 @@ describe('createRefkit', () => { expect(out.references.map(r => r.canonicalUrl)).toEqual(['https://ok/1']) expect(out.meta.providers).toEqual([ - { providerId: 'ok', status: 'fulfilled', returned: 2, accepted: 2, rejected: 0 }, - { providerId: 'bad', status: 'failed', error: 'boom' }, + { providerId: 'ok', status: 'fulfilled', returned: 2, accepted: 2, rejected: 0, latencyMs: expect.any(Number) }, + { providerId: 'bad', status: 'failed', error: 'boom', latencyMs: expect.any(Number) }, { providerId: 'text', status: 'skipped', reason: 'unsupported-modality' }, ]) expect(out.meta.gate).toEqual({ intent: 'commercial-product', before: 2, after: 1, dropped: 1 }) @@ -264,4 +269,70 @@ describe('createRefkit', () => { ignoredByProvider: { controlled: ['safety'], plain: ['orientation', 'color', 'safety'] }, }) }) + + it('times out a hanging provider, returns partial results, and reports the timeout', async () => { + vi.useFakeTimers() + try { + const hanging = defineProvider({ + id: 'hang', modalities: ['image'], queryFeatures: ['keyword'], + search: () => new Promise(() => {}), // never settles, ignores ctx.signal + }) + const rk = createRefkit({ providers: [provider('a', [ref('a-1', 'https://a/1')]), hanging] }) + const p = rk.searchWithMeta({ query: 'x', modalities: ['image'] }) + await vi.advanceTimersByTimeAsync(10_000) + const out = await p + expect(out.references).toHaveLength(1) + const hangStatus = out.meta.providers.find(s => s.providerId === 'hang') + expect(hangStatus?.status).toBe('failed') + expect(hangStatus?.error).toContain('timeout after 10000ms') + } finally { + vi.useRealTimers() + } + }) + + it('resilience: false disables the timeout entirely', async () => { + vi.useFakeTimers() + try { + let done = false + const slow = defineProvider({ + id: 'slow', modalities: ['image'], queryFeatures: ['keyword'], + search: () => new Promise(resolve => setTimeout(() => { done = true; resolve([ref('slow-1', 'https://s/1')]) }, 60_000)), + }) + const rk = createRefkit({ providers: [slow], resilience: false }) + const p = rk.search({ query: 'x', modalities: ['image'] }) + await vi.advanceTimersByTimeAsync(60_000) + expect(await p).toHaveLength(1) + expect(done).toBe(true) + } finally { + vi.useRealTimers() + } + }) + + it('gives providers a retrying ctx.fetch: a 500-then-200 upstream succeeds transparently', async () => { + const upstream = vi.fn() + .mockResolvedValueOnce(new Response('x', { status: 500 })) + .mockResolvedValueOnce(new Response('ok', { status: 200 })) + const usesFetch = defineProvider({ + id: 'net', modalities: ['image'], queryFeatures: ['keyword'], + search: async (_q, ctx) => { + const res = await ctx.fetch('https://net/api', { signal: ctx.signal }) + if (!res.ok) throw new Error(`net failed: ${res.status}`) + return [ref('net-1', 'https://net/1')] + }, + }) + const rk = createRefkit({ providers: [usesFetch], fetch: upstream as unknown as typeof fetch, resilience: { retries: 1, timeoutMs: 10_000 } }) + const out = await rk.search({ query: 'x', modalities: ['image'] }) + expect(out).toHaveLength(1) + expect(upstream).toHaveBeenCalledTimes(2) + }) + + it('reports latencyMs on fulfilled and failed providers, not on skipped', async () => { + const textOnly = defineProvider({ id: 'text', modalities: ['text'], queryFeatures: [], search: async () => [] }) + const rk = createRefkit({ providers: [provider('a', [ref('a-1', 'https://a/1')]), failing('bad'), textOnly] }) + const out = await rk.searchWithMeta({ query: 'x', modalities: ['image'] }) + const byId = Object.fromEntries(out.meta.providers.map(s => [s.providerId, s])) + expect(byId.a.latencyMs).toEqual(expect.any(Number)) + expect(byId.bad.latencyMs).toEqual(expect.any(Number)) + expect(byId.text.latencyMs).toBeUndefined() + }) }) diff --git a/packages/core/src/client.ts b/packages/core/src/client.ts index 7f98f3f..859cce8 100644 --- a/packages/core/src/client.ts +++ b/packages/core/src/client.ts @@ -17,6 +17,14 @@ import type { } from './provider' import { mergeReferences, type MergeOptions } from './merge' import { mergeSearchControls, normalizeQuery, requestedControlKeys, supportedControlKeys, unsupportedControlKeys } from './query' +import { retryingFetch, withTimeout } from './resilience' + +export interface ResilienceOptions { + /** Soft deadline per provider search. Default 10_000. */ + timeoutMs?: number + /** Extra fetch attempts on 429/5xx/network-error. Default 1. */ + retries?: number +} export interface RefkitOptions { providers: ReferenceProvider[] @@ -24,6 +32,8 @@ export interface RefkitOptions { cache?: KeyValueCache signal?: AbortSignal merge?: MergeOptions + /** Per-provider timeout + retry (H8). Defaults ON; pass `false` to disable both. */ + resilience?: ResilienceOptions | false } export interface ProviderError { @@ -39,6 +49,7 @@ export interface ProviderSearchStatus { rejected?: number reason?: 'unsupported-modality' error?: string + latencyMs?: number } export interface SearchGateMeta { @@ -105,6 +116,8 @@ export interface RefkitClient { const DEFAULT_LIMIT = 30 const DEFAULT_POOL_FACTOR = 4 const MAX_POOL_LIMIT = 100 // never ask a single source for more than this, even at high limits +const DEFAULT_TIMEOUT_MS = 10_000 +const DEFAULT_RETRIES = 1 function errorSummary(error: unknown): string { if (error instanceof Error) return error.message @@ -122,12 +135,6 @@ export function createRefkit(options: RefkitOptions): RefkitClient { if (typeof doFetch !== 'function') { throw new Error('createRefkit: no fetch available — pass options.fetch') } - const ctx: ProviderContext = { - fetch: doFetch, - cache: options.cache, - signal: input.signal ?? options.signal, - } - const chosen = options.providers.filter(p => p.modalities.some(m => input.modalities.includes(m))) if (chosen.length === 0) { throw new Error(`refkit.search: no registered provider supports modalities [${input.modalities.join(', ')}]`) @@ -148,55 +155,76 @@ export function createRefkit(options: RefkitOptions): RefkitClient { for (const p of options.providers) { if (!chosen.includes(p)) statusByProvider.set(p.id, { providerId: p.id, status: 'skipped', reason: 'unsupported-modality' }) } - const settled = await Promise.allSettled( - chosen.map(p => - p.search( - normalizeQuery({ - query: input.query, - modalities: input.modalities, - filters: input.filters, - controls: input.controls, - providerOptions: input.providerOptions, - limit: fetchLimit, - }, p), - ctx, - ), - ), - ) + const resilience = options.resilience === false ? undefined : { + timeoutMs: options.resilience?.timeoutMs ?? DEFAULT_TIMEOUT_MS, + retries: options.resilience?.retries ?? DEFAULT_RETRIES, + } - const perSource: Reference[][] = [] - let anyOk = false - settled.forEach((res, i) => { - const provider = chosen[i] - if (res.status === 'fulfilled') { - anyOk = true + type ProviderRun = + | { ok: true; valid: Reference[]; returned: number; latencyMs: number } + | { ok: false; error: unknown; latencyMs: number } + + const runProvider = async (p: ReferenceProvider): Promise => { + const started = Date.now() + const timeout = resilience ? withTimeout(input.signal ?? options.signal, resilience.timeoutMs) : undefined + const ctx: ProviderContext = { + fetch: resilience && resilience.retries > 0 ? retryingFetch(doFetch, { retries: resilience.retries }) : doFetch, + cache: options.cache, + signal: timeout?.signal ?? input.signal ?? options.signal, + } + const query = normalizeQuery({ + query: input.query, + modalities: input.modalities, + filters: input.filters, + controls: input.controls, + providerOptions: input.providerOptions, + limit: fetchLimit, + }, p) + try { + const searching = p.search(query, ctx) + searching.catch(() => {}) // a raced-past provider must not become an unhandled rejection + const raw = await (timeout ? Promise.race([searching, timeout.expired]) : searching) const valid: Reference[] = [] - for (const raw of res.value) { + for (const item of raw) { try { - valid.push(parseReference(raw)) + valid.push(parseReference(item)) } catch (error) { - input.onProviderError?.({ providerId: provider.id, error }) + input.onProviderError?.({ providerId: p.id, error }) } } + return { ok: true, valid, returned: raw.length, latencyMs: Date.now() - started } + } catch (error) { + input.onProviderError?.({ providerId: p.id, error }) + return { ok: false, error, latencyMs: Date.now() - started } + } finally { + timeout?.cancel() + } + } + + const runs = await Promise.all(chosen.map(runProvider)) + + const perSource: Reference[][] = [] + let anyOk = false + runs.forEach((run, i) => { + const provider = chosen[i] + if (run.ok) { + anyOk = true statusByProvider.set(provider.id, { providerId: provider.id, status: 'fulfilled', - returned: res.value.length, - accepted: valid.length, - rejected: res.value.length - valid.length, + returned: run.returned, + accepted: run.valid.length, + rejected: run.returned - run.valid.length, + latencyMs: run.latencyMs, }) - perSource.push(valid) + perSource.push(run.valid) } else { - input.onProviderError?.({ providerId: provider.id, error: res.reason }) - statusByProvider.set(provider.id, { providerId: provider.id, status: 'failed', error: errorSummary(res.reason) }) + statusByProvider.set(provider.id, { providerId: provider.id, status: 'failed', error: errorSummary(run.error), latencyMs: run.latencyMs }) } }) if (!anyOk) { - const reasons = settled - .filter((s): s is PromiseRejectedResult => s.status === 'rejected') - .map(s => s.reason) - throw new AggregateError(reasons, 'refkit.search: all providers failed') + throw new AggregateError(runs.filter(r => !r.ok).map(r => (r as { error: unknown }).error), 'refkit.search: all providers failed') } let refs = mergeReferences(perSource, options.merge) diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 6b59849..fd3a43e 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -54,6 +54,7 @@ export { createRefkit } from './client' export type { RefkitClient, RefkitOptions, + ResilienceOptions, SearchInput, SearchResult, SearchMeta, From 5011f95e245e5e523bb079c8a0d964195f09c9aa Mon Sep 17 00:00:00 2001 From: MyPrototypeWhat Date: Fri, 3 Jul 2026 14:50:04 +0800 Subject: [PATCH 4/9] feat(core): per-provider result cache on the KeyValueCache port --- packages/core/src/__tests__/client.test.ts | 70 ++++++++++++++++++++++ packages/core/src/client.ts | 26 +++++++- 2 files changed, 95 insertions(+), 1 deletion(-) diff --git a/packages/core/src/__tests__/client.test.ts b/packages/core/src/__tests__/client.test.ts index a21a519..40bb672 100644 --- a/packages/core/src/__tests__/client.test.ts +++ b/packages/core/src/__tests__/client.test.ts @@ -335,4 +335,74 @@ describe('createRefkit', () => { expect(byId.bad.latencyMs).toEqual(expect.any(Number)) expect(byId.text.latencyMs).toBeUndefined() }) + + const mapCache = () => { + const m = new Map() + return { + store: m, + ttls: [] as (number | undefined)[], + async get(k: string) { return m.get(k) }, + async set(k: string, v: string, ttlMs?: number) { m.set(k, v); this.ttls.push(ttlMs) }, + } + } + + it('serves a repeat query from the cache without re-hitting the provider', async () => { + const cache = mapCache() + let calls = 0 + const counted = defineProvider({ + id: 'c', modalities: ['image'], queryFeatures: ['keyword'], + search: async () => { calls++; return [ref('c-1', 'https://c/1')] }, + }) + const rk = createRefkit({ providers: [counted], cache }) + await rk.search({ query: 'x', modalities: ['image'] }) + const out = await rk.searchWithMeta({ query: 'x', modalities: ['image'] }) + expect(calls).toBe(1) + expect(out.references).toHaveLength(1) + expect(out.meta.providers[0]).toMatchObject({ status: 'fulfilled', cached: true }) + expect(cache.ttls).toEqual([300_000]) // default cacheTtlMs, one set for the first (live) search + }) + + it('different queries use different cache keys', async () => { + const cache = mapCache() + let calls = 0 + const counted = defineProvider({ + id: 'c', modalities: ['image'], queryFeatures: ['keyword'], + search: async () => { calls++; return [ref('c-1', 'https://c/1')] }, + }) + const rk = createRefkit({ providers: [counted], cache }) + await rk.search({ query: 'x', modalities: ['image'] }) + await rk.search({ query: 'y', modalities: ['image'] }) + expect(calls).toBe(2) + }) + + it('a corrupt or invalid cache entry falls back to a live fetch', async () => { + const cache = mapCache() + let calls = 0 + const counted = defineProvider({ + id: 'c', modalities: ['image'], queryFeatures: ['keyword'], + search: async () => { calls++; return [ref('c-1', 'https://c/1')] }, + }) + const rk = createRefkit({ providers: [counted], cache }) + await rk.search({ query: 'x', modalities: ['image'] }) + for (const k of cache.store.keys()) cache.store.set(k, '{not json') + await rk.search({ query: 'x', modalities: ['image'] }) + expect(calls).toBe(2) + }) + + it('cache errors are non-fatal: a throwing cache degrades to live search', async () => { + const broken = { + async get(): Promise { throw new Error('cache down') }, + async set(): Promise { throw new Error('cache down') }, + } + const rk = createRefkit({ providers: [provider('a', [ref('a-1', 'https://a/1')])], cache: broken }) + const out = await rk.search({ query: 'x', modalities: ['image'] }) + expect(out).toHaveLength(1) + }) + + it('honors a custom cacheTtlMs', async () => { + const cache = mapCache() + const rk = createRefkit({ providers: [provider('a', [ref('a-1', 'https://a/1')])], cache, cacheTtlMs: 1234 }) + await rk.search({ query: 'x', modalities: ['image'] }) + expect(cache.ttls).toEqual([1234]) + }) }) diff --git a/packages/core/src/client.ts b/packages/core/src/client.ts index 859cce8..ed83a14 100644 --- a/packages/core/src/client.ts +++ b/packages/core/src/client.ts @@ -18,6 +18,7 @@ import type { import { mergeReferences, type MergeOptions } from './merge' import { mergeSearchControls, normalizeQuery, requestedControlKeys, supportedControlKeys, unsupportedControlKeys } from './query' import { retryingFetch, withTimeout } from './resilience' +import { fnv1a } from './hash' export interface ResilienceOptions { /** Soft deadline per provider search. Default 10_000. */ @@ -34,6 +35,8 @@ export interface RefkitOptions { merge?: MergeOptions /** Per-provider timeout + retry (H8). Defaults ON; pass `false` to disable both. */ resilience?: ResilienceOptions | false + /** TTL for per-provider cached results; used only when `cache` is set. Default 300_000. */ + cacheTtlMs?: number } export interface ProviderError { @@ -50,6 +53,7 @@ export interface ProviderSearchStatus { reason?: 'unsupported-modality' error?: string latencyMs?: number + cached?: boolean } export interface SearchGateMeta { @@ -118,6 +122,7 @@ const DEFAULT_POOL_FACTOR = 4 const MAX_POOL_LIMIT = 100 // never ask a single source for more than this, even at high limits const DEFAULT_TIMEOUT_MS = 10_000 const DEFAULT_RETRIES = 1 +const DEFAULT_CACHE_TTL_MS = 300_000 function errorSummary(error: unknown): string { if (error instanceof Error) return error.message @@ -161,7 +166,7 @@ export function createRefkit(options: RefkitOptions): RefkitClient { } type ProviderRun = - | { ok: true; valid: Reference[]; returned: number; latencyMs: number } + | { ok: true; valid: Reference[]; returned: number; latencyMs: number; cached?: boolean } | { ok: false; error: unknown; latencyMs: number } const runProvider = async (p: ReferenceProvider): Promise => { @@ -180,6 +185,20 @@ export function createRefkit(options: RefkitOptions): RefkitClient { providerOptions: input.providerOptions, limit: fetchLimit, }, p) + const cacheKey = options.cache + ? `refkit:v1:${p.id}:${fnv1a(JSON.stringify(query))}` + : undefined + if (options.cache && cacheKey) { + // best-effort: a broken/corrupt/stale cache degrades to a live search + const hit = await options.cache.get(cacheKey).catch(() => undefined) + if (hit !== undefined) { + try { + // cached refs keep their original verifiedAt — staleness is bounded by the TTL + const parsed = (JSON.parse(hit) as unknown[]).map(item => parseReference(item)) + return { ok: true, valid: parsed, returned: parsed.length, latencyMs: Date.now() - started, cached: true } + } catch { /* fall through to live */ } + } + } try { const searching = p.search(query, ctx) searching.catch(() => {}) // a raced-past provider must not become an unhandled rejection @@ -192,6 +211,10 @@ export function createRefkit(options: RefkitOptions): RefkitClient { input.onProviderError?.({ providerId: p.id, error }) } } + if (options.cache && cacheKey) { + // fire-and-forget: cache write failure must never fail the search + void options.cache.set(cacheKey, JSON.stringify(valid), options.cacheTtlMs ?? DEFAULT_CACHE_TTL_MS).catch(() => {}) + } return { ok: true, valid, returned: raw.length, latencyMs: Date.now() - started } } catch (error) { input.onProviderError?.({ providerId: p.id, error }) @@ -216,6 +239,7 @@ export function createRefkit(options: RefkitOptions): RefkitClient { accepted: run.valid.length, rejected: run.returned - run.valid.length, latencyMs: run.latencyMs, + ...(run.cached ? { cached: true } : {}), }) perSource.push(run.valid) } else { From 9b781053cb645f1459c0abf4451511b86c1ab22c Mon Sep 17 00:00:00 2001 From: MyPrototypeWhat Date: Fri, 3 Jul 2026 14:56:39 +0800 Subject: [PATCH 5/9] fix(core): stable cache keys, deadline-bounded cache reads, gate-freshness test --- packages/core/src/__tests__/client.test.ts | 23 +++++++++++++++++++ packages/core/src/client.ts | 26 +++++++++++++++++++--- packages/core/src/provider.ts | 2 ++ 3 files changed, 48 insertions(+), 3 deletions(-) diff --git a/packages/core/src/__tests__/client.test.ts b/packages/core/src/__tests__/client.test.ts index 40bb672..c13672f 100644 --- a/packages/core/src/__tests__/client.test.ts +++ b/packages/core/src/__tests__/client.test.ts @@ -405,4 +405,27 @@ describe('createRefkit', () => { await rk.search({ query: 'x', modalities: ['image'] }) expect(cache.ttls).toEqual([1234]) }) + + it('a cache hit still flows through the license gate (hits are pre-merge; the gate stays live)', async () => { + const cache = mapCache() + const rk = createRefkit({ providers: [provider('a', [ref('a-1', 'https://a/1', 'proprietary')])], cache }) + await rk.search({ query: 'x', modalities: ['image'] }) + const out = await rk.searchWithMeta({ query: 'x', modalities: ['image'], gateFor: 'commercial-product' }) + expect(out.meta.providers[0]).toMatchObject({ status: 'fulfilled', cached: true }) + expect(out.references).toHaveLength(0) + expect(out.meta.gate).toMatchObject({ intent: 'commercial-product', before: 1, after: 0, dropped: 1 }) + }) + + it('providerOptions key order does not change the cache key', async () => { + const cache = mapCache() + let calls = 0 + const counted = defineProvider({ + id: 'c', modalities: ['image'], queryFeatures: ['keyword'], + search: async () => { calls++; return [ref('c-1', 'https://c/1')] }, + }) + const rk = createRefkit({ providers: [counted], cache }) + await rk.search({ query: 'x', modalities: ['image'], providerOptions: { c: { b: 2, a: 1 } } }) + await rk.search({ query: 'x', modalities: ['image'], providerOptions: { c: { a: 1, b: 2 } } }) + expect(calls).toBe(1) // second search is a cache hit despite the different key order + }) }) diff --git a/packages/core/src/client.ts b/packages/core/src/client.ts index ed83a14..41654fb 100644 --- a/packages/core/src/client.ts +++ b/packages/core/src/client.ts @@ -130,6 +130,17 @@ function errorSummary(error: unknown): string { return 'unknown error' } +// Deterministic JSON for cache keys: object keys sorted recursively, so a +// caller's providerOptions key order can't split otherwise-identical keys. +function stableStringify(value: unknown): string { + if (Array.isArray(value)) return `[${value.map(stableStringify).join(',')}]` + if (value && typeof value === 'object') { + const rec = value as Record + return `{${Object.keys(rec).sort().map(k => `${JSON.stringify(k)}:${stableStringify(rec[k])}`).join(',')}}` + } + return JSON.stringify(value) ?? 'null' +} + export function createRefkit(options: RefkitOptions): RefkitClient { if (!options.providers || options.providers.length === 0) { throw new Error('createRefkit: at least one provider is required') @@ -186,14 +197,23 @@ export function createRefkit(options: RefkitOptions): RefkitClient { limit: fetchLimit, }, p) const cacheKey = options.cache - ? `refkit:v1:${p.id}:${fnv1a(JSON.stringify(query))}` + ? `refkit:v1:${p.id}:${fnv1a(stableStringify(query))}` : undefined if (options.cache && cacheKey) { // best-effort: a broken/corrupt/stale cache degrades to a live search - const hit = await options.cache.get(cacheKey).catch(() => undefined) + const pending = options.cache.get(cacheKey) + // A slow cache read must not outlive the provider deadline: at expiry it + // degrades to a miss, and the live search below then fails fast on the + // same (already-expired) deadline. + const hit = await (timeout + ? Promise.race([pending, timeout.expired.catch(() => undefined)]) + : pending + ).catch(() => undefined) + pending.catch(() => {}) // raced-past rejection must not go unhandled if (hit !== undefined) { try { - // cached refs keep their original verifiedAt — staleness is bounded by the TTL + // cached refs keep their original verifiedAt — staleness is bounded + // by the TTL when the cache honors ttlMs const parsed = (JSON.parse(hit) as unknown[]).map(item => parseReference(item)) return { ok: true, valid: parsed, returned: parsed.length, latencyMs: Date.now() - started, cached: true } } catch { /* fall through to live */ } diff --git a/packages/core/src/provider.ts b/packages/core/src/provider.ts index c68bdd3..e115236 100644 --- a/packages/core/src/provider.ts +++ b/packages/core/src/provider.ts @@ -90,6 +90,8 @@ export interface NormalizedQuery { limit?: number } +/** Implementations SHOULD honor ttlMs — refkit's cached-result freshness is + * bounded by the TTL only when the cache enforces it. */ export interface KeyValueCache { get(key: string): Promise set(key: string, value: string, ttlMs?: number): Promise From 52e5485f766a28fa98d4e84d57bffd7a5d8f2b59 Mon Sep 17 00:00:00 2001 From: MyPrototypeWhat Date: Fri, 3 Jul 2026 15:04:41 +0800 Subject: [PATCH 6/9] =?UTF-8?q?test(core):=20review=20carry-overs=20?= =?UTF-8?q?=E2=80=94=20jitter=20term,=20abort-clause=20locks,=20deadline-b?= =?UTF-8?q?ounded=20cache=20read;=20surface=20latencyMs=20in=20mcp=20meta?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/core/src/__tests__/client.test.ts | 46 +++++++++++++++++++ .../core/src/__tests__/resilience.test.ts | 30 ++++++++++-- packages/core/src/provider.ts | 3 ++ packages/core/src/resilience.ts | 4 +- packages/mcp/src/__tests__/mcp.test.ts | 6 +-- packages/mcp/src/index.ts | 2 + 6 files changed, 81 insertions(+), 10 deletions(-) diff --git a/packages/core/src/__tests__/client.test.ts b/packages/core/src/__tests__/client.test.ts index c13672f..e41efaa 100644 --- a/packages/core/src/__tests__/client.test.ts +++ b/packages/core/src/__tests__/client.test.ts @@ -290,6 +290,31 @@ describe('createRefkit', () => { } }) + it('a well-behaved provider that rejects on ctx.signal abort observes the timeout and is reported failed', async () => { + vi.useFakeTimers() + try { + let observedAbort = false + const wellBehaved = defineProvider({ + id: 'wb', modalities: ['image'], queryFeatures: ['keyword'], + search: (_q, ctx) => new Promise((_resolve, reject) => { + ctx.signal?.addEventListener('abort', () => { + observedAbort = true + reject(ctx.signal?.reason ?? new Error('aborted')) + }) + }), + }) + const rk = createRefkit({ providers: [provider('a', [ref('a-1', 'https://a/1')]), wellBehaved] }) + const p = rk.searchWithMeta({ query: 'x', modalities: ['image'] }) + await vi.advanceTimersByTimeAsync(10_000) + const out = await p + expect(observedAbort).toBe(true) + const wbStatus = out.meta.providers.find(s => s.providerId === 'wb') + expect(wbStatus?.status).toBe('failed') + } finally { + vi.useRealTimers() + } + }) + it('resilience: false disables the timeout entirely', async () => { vi.useFakeTimers() try { @@ -416,6 +441,27 @@ describe('createRefkit', () => { expect(out.meta.gate).toMatchObject({ intent: 'commercial-product', before: 1, after: 0, dropped: 1 }) }) + it('a never-resolving cache.get does not hang the search — deadline-bounded cache read falls back to live results', async () => { + vi.useFakeTimers() + try { + const hangingCache = { + get: () => new Promise(() => {}), // never resolves + set: async () => {}, + } + const rk = createRefkit({ + providers: [provider('a', [ref('a-1', 'https://a/1')])], + cache: hangingCache, + resilience: { timeoutMs: 100 }, + }) + const p = rk.search({ query: 'x', modalities: ['image'] }) + await vi.advanceTimersByTimeAsync(100) + const out = await p + expect(out).toHaveLength(1) // live results, not a hang + } finally { + vi.useRealTimers() + } + }) + it('providerOptions key order does not change the cache key', async () => { const cache = mapCache() let calls = 0 diff --git a/packages/core/src/__tests__/resilience.test.ts b/packages/core/src/__tests__/resilience.test.ts index 49fd79b..ea5d4dd 100644 --- a/packages/core/src/__tests__/resilience.test.ts +++ b/packages/core/src/__tests__/resilience.test.ts @@ -47,16 +47,20 @@ describe('retryingFetch', () => { expect(impl).toHaveBeenCalledTimes(2) }) - it('drains (cancels) the body of a discarded retryable response', async () => { + it('drains (cancels) the body of a discarded retryable response, never the returned one', async () => { const res500 = status(500) - expect(res500.body).not.toBeNull() // guard: the spy below must target a real stream - const cancelSpy = vi.spyOn(res500.body!, 'cancel') - const impl = vi.fn().mockResolvedValueOnce(res500).mockResolvedValueOnce(okResponse()) + const res200 = okResponse() + expect(res500.body).not.toBeNull() // guard: the spies below must target real streams + expect(res200.body).not.toBeNull() + const cancelSpy500 = vi.spyOn(res500.body!, 'cancel') + const cancelSpy200 = vi.spyOn(res200.body!, 'cancel') + const impl = vi.fn().mockResolvedValueOnce(res500).mockResolvedValueOnce(res200) const f = retryingFetch(impl as unknown as typeof fetch, { retries: 1 }) const p = f('https://x/') await vi.runAllTimersAsync() expect((await p).status).toBe(200) - expect(cancelSpy).toHaveBeenCalledTimes(1) + expect(cancelSpy500).toHaveBeenCalledTimes(1) + expect(cancelSpy200).not.toHaveBeenCalled() // never drained on the return path }) it('retries 429 and a rejected network error', async () => { @@ -103,4 +107,20 @@ describe('retryingFetch', () => { await p2 expect(impl2).toHaveBeenCalledTimes(1) // aborted before/during first backoff — no second attempt }) + + it('does not retry a deadline abort even when the rejection is a plain Error (not name AbortError)', async () => { + // undici rejects fetch with the abort *reason* when the signal fires — and + // withTimeout's reason is a plain `Error('timeout after Nms')` whose name is + // 'Error', not 'AbortError'. retryingFetch must still recognize this as an + // abort via `init?.signal?.aborted` and not burn a retry on it. + const ac = new AbortController() + const timeoutErr = new Error('timeout after 100ms') + const impl = vi.fn().mockImplementation(() => { + ac.abort(timeoutErr) + return Promise.reject(timeoutErr) + }) + const f = retryingFetch(impl as unknown as typeof fetch, { retries: 3 }) + await expect(f('https://x/', { signal: ac.signal })).rejects.toBe(timeoutErr) + expect(impl).toHaveBeenCalledTimes(1) + }) }) diff --git a/packages/core/src/provider.ts b/packages/core/src/provider.ts index e115236..123e05d 100644 --- a/packages/core/src/provider.ts +++ b/packages/core/src/provider.ts @@ -102,6 +102,9 @@ export interface KeyValueCache { export interface ProviderContext { fetch: typeof fetch cache?: KeyValueCache + /** Forward into fetch init.signal — it carries the orchestrator's per-provider + * deadline; forwarding enables timeout cancellation and prevents burning a + * retry on an aborted request. */ signal?: AbortSignal } diff --git a/packages/core/src/resilience.ts b/packages/core/src/resilience.ts index 772cade..3f9a12c 100644 --- a/packages/core/src/resilience.ts +++ b/packages/core/src/resilience.ts @@ -44,7 +44,7 @@ export function withTimeout(parent: AbortSignal | undefined, timeoutMs: number): export interface RetryOptions { /** Extra attempts after the first. The orchestrator passes its own default (H8: 1). */ retries: number - /** Base backoff delay; grows 2^attempt with full jitter. Default 250. */ + /** Base backoff delay; grows 2^attempt with equal jitter. Default 250. */ baseDelayMs?: number } @@ -80,7 +80,7 @@ export function retryingFetch(fetchImpl: typeof fetch, opts: RetryOptions): type } // drain the discarded body so undici can reuse the socket during retries void discarded?.body?.cancel().catch(() => {}) - // exponential backoff with full jitter; an abort during the wait cancels the retry + // exponential backoff with equal jitter (half fixed + half random); an abort during the wait cancels the retry await abortAware(base * 2 ** attempt * (0.5 + Math.random() * 0.5), init?.signal) } } diff --git a/packages/mcp/src/__tests__/mcp.test.ts b/packages/mcp/src/__tests__/mcp.test.ts index 2008bb7..3875c00 100644 --- a/packages/mcp/src/__tests__/mcp.test.ts +++ b/packages/mcp/src/__tests__/mcp.test.ts @@ -212,12 +212,12 @@ describe('@refkit/mcp', () => { }) const structured = res.structuredContent as { references: Array<{ useExplanation?: string }> - meta?: { providers: Array<{ providerId: string; status: string; error?: string }>; warnings: string[] } + meta?: { providers: Array<{ providerId: string; status: string; error?: string; latencyMs?: number }>; warnings: string[] } } expect(structured.references[0].useExplanation).toContain('allowed-with-attribution') expect(structured.meta?.providers).toEqual([ - { providerId: 'good', status: 'fulfilled', returned: 1, accepted: 1, rejected: 0 }, - { providerId: 'bad', status: 'failed', error: 'offline' }, + { providerId: 'good', status: 'fulfilled', returned: 1, accepted: 1, rejected: 0, latencyMs: expect.any(Number) }, + { providerId: 'bad', status: 'failed', error: 'offline', latencyMs: expect.any(Number) }, ]) expect(structured.meta?.warnings).toContain('1 provider(s) failed; returning partial results.') await client.close() diff --git a/packages/mcp/src/index.ts b/packages/mcp/src/index.ts index e780092..170b322 100644 --- a/packages/mcp/src/index.ts +++ b/packages/mcp/src/index.ts @@ -139,6 +139,8 @@ const searchMetaSchema: z.ZodType = z.object({ rejected: z.number().optional(), reason: z.enum(['unsupported-modality']).optional(), error: z.string().optional(), + latencyMs: z.number().optional(), + cached: z.boolean().optional(), })), gate: z.object({ intent: z.enum(INTENTS), From c6b6061e2dc613aec937338e7fe5c47c44987552 Mon Sep 17 00:00:00 2001 From: MyPrototypeWhat Date: Fri, 3 Jul 2026 15:04:44 +0800 Subject: [PATCH 7/9] docs: resilience & caching section + changeset for orchestrator hardening --- .changeset/orchestrator-hardening.md | 12 ++++++++++++ README.md | 17 ++++++++++++++++- 2 files changed, 28 insertions(+), 1 deletion(-) create mode 100644 .changeset/orchestrator-hardening.md diff --git a/.changeset/orchestrator-hardening.md b/.changeset/orchestrator-hardening.md new file mode 100644 index 0000000..ed84da8 --- /dev/null +++ b/.changeset/orchestrator-hardening.md @@ -0,0 +1,12 @@ +--- +"@refkit/core": minor +--- + +Harden the search orchestrator: per-provider soft timeout (default 10s) and +bounded retry on 429/5xx/network errors (default 1, exponential backoff) — on by +default, tunable or disabled via `createRefkit({ resilience })`. Provider +statuses in `searchWithMeta` now carry `latencyMs`, and supplying a `cache` +(`KeyValueCache`) now memoizes per-provider results (key +`refkit:v1::`, TTL `cacheTtlMs`, default 5 min) with hits +flagged `cached: true`. Merge, rerank, and the license gate always run fresh. +New exports: `withTimeout`, `retryingFetch`, `ResilienceOptions`. diff --git a/README.md b/README.md index 909e73e..a9f4037 100644 --- a/README.md +++ b/README.md @@ -34,7 +34,7 @@ const refkit = createRefkit({ met(), // keyless unsplash({ accessKey: process.env.UNSPLASH_KEY! }), // BYOK ], - // fetch defaults to globalThis.fetch — inject your own to add caching/retries + // fetch defaults to globalThis.fetch — timeouts/retries/caching are built in (see below) }) // Fan out, merge (Reciprocal Rank Fusion) + dedup; every result carries rights. @@ -149,6 +149,21 @@ const refkit = createRefkit({ }) ``` +## Resilience & caching + +Fan-out is hardened by default: each provider gets a **soft 10s timeout** and **one retry** (429/5xx/network errors, exponential backoff). A slow or hanging source is reported in `meta.providers` as `failed` with `timeout after Nms` — the search still returns everyone else. Tune or disable per client: + +```ts +createRefkit({ providers, resilience: { timeoutMs: 4000, retries: 2 } }) +createRefkit({ providers, resilience: false }) // raw fan-out, no timeout/retry +``` + +Pass a `cache` to memoize **per-provider** results (keyed by provider + normalized query, TTL `cacheTtlMs`, default 5 min). Merging, reranking, and the license gate always run fresh; cache hits are flagged `cached: true` in `meta.providers`, and every provider status carries `latencyMs`: + +```ts +createRefkit({ providers, cache: myKvCache, cacheTtlMs: 60_000 }) +``` + ## Providers | Package | Source | Modality | Auth | License | From eba188a0dd6bbb60d817e8bc8273cbaff3a9eac8 Mon Sep 17 00:00:00 2001 From: MyPrototypeWhat Date: Fri, 3 Jul 2026 15:09:49 +0800 Subject: [PATCH 8/9] chore: bump @refkit/mcp for surfaced meta fields; complete changeset exports list; commit wave plan --- .changeset/orchestrator-hardening.md | 11 +- .../2026-07-03-orchestrator-hardening.md | 616 ++++++++++++++++++ 2 files changed, 624 insertions(+), 3 deletions(-) create mode 100644 docs/superpowers/plans/2026-07-03-orchestrator-hardening.md diff --git a/.changeset/orchestrator-hardening.md b/.changeset/orchestrator-hardening.md index ed84da8..6630f05 100644 --- a/.changeset/orchestrator-hardening.md +++ b/.changeset/orchestrator-hardening.md @@ -1,5 +1,6 @@ --- "@refkit/core": minor +"@refkit/mcp": minor --- Harden the search orchestrator: per-provider soft timeout (default 10s) and @@ -7,6 +8,10 @@ bounded retry on 429/5xx/network errors (default 1, exponential backoff) — on default, tunable or disabled via `createRefkit({ resilience })`. Provider statuses in `searchWithMeta` now carry `latencyMs`, and supplying a `cache` (`KeyValueCache`) now memoizes per-provider results (key -`refkit:v1::`, TTL `cacheTtlMs`, default 5 min) with hits -flagged `cached: true`. Merge, rerank, and the license gate always run fresh. -New exports: `withTimeout`, `retryingFetch`, `ResilienceOptions`. +`refkit:v1::`, TTL via the new `cacheTtlMs` option, default +5 min) with hits flagged `cached: true`. Merge, rerank, and the license gate +always run fresh. New exports: `withTimeout`, `retryingFetch`, and the +`ResilienceOptions`, `TimeoutHandle`, `RetryOptions` types. + +The MCP `search_references` structured output (`explain: true`) now surfaces +`latencyMs` per provider and `cached` on cache hits. diff --git a/docs/superpowers/plans/2026-07-03-orchestrator-hardening.md b/docs/superpowers/plans/2026-07-03-orchestrator-hardening.md new file mode 100644 index 0000000..cae8bce --- /dev/null +++ b/docs/superpowers/plans/2026-07-03-orchestrator-hardening.md @@ -0,0 +1,616 @@ +# Wave 2 — Orchestrator Hardening Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Per-provider soft timeout + retry-with-backoff, per-provider `latencyMs` metadata, and an opt-in per-provider result cache on the existing `KeyValueCache` port — all in `@refkit/core`, honoring decisions H8–H11 in `2026-07-03-hardening-index.md`. + +**Architecture:** New focused module `packages/core/src/resilience.ts` (signal composition + retrying fetch — pure, zero-network). `client.ts`'s fan-out refactors into a per-provider `runProvider` (own ctx, timeout race, latency measurement, cache), replacing `Promise.allSettled` with never-throwing discriminated results. `evaluateUse`/merge/rerank untouched. + +**Tech Stack:** TypeScript ESM, vitest (fake timers), zod. No new dependencies. Branch: `m13t/wave2-orchestrator-hardening`. + +**Locked decisions (do not re-litigate):** H8 defaults ON (`timeoutMs: 10_000`, `retries: 1`, retry only 429/5xx/network-error; `resilience: false` disables); H9 no `AbortSignal.any` (manual composition; `setTimeout` in core is fine — the no-network test only bans `fetch(` calls and `http(s)://` literals); H10 cache per-provider pre-merge, key `refkit:v1::`, TTL `cacheTtlMs` default 300_000, hits marked `cached: true`; H11 `latencyMs` around each provider run. + +--- + +### Task W2.1: `resilience.ts` — `withTimeout` + `retryingFetch` + +**Files:** +- Create: `packages/core/src/resilience.ts` +- Modify: `packages/core/src/index.ts` (exports) +- Test: `packages/core/src/__tests__/resilience.test.ts` + +- [ ] **Step 1: Write the failing tests** — create `resilience.test.ts`: + +```ts +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { retryingFetch, withTimeout } from '../resilience' + +const okResponse = () => new Response('ok', { status: 200 }) +const status = (s: number) => new Response('x', { status: s }) + +describe('withTimeout', () => { + beforeEach(() => vi.useFakeTimers()) + afterEach(() => vi.useRealTimers()) + + it('aborts the signal and rejects `expired` after timeoutMs', async () => { + const t = withTimeout(undefined, 1000) + expect(t.signal.aborted).toBe(false) + const raced = Promise.race([new Promise(() => {}), t.expired]).catch(e => e) + await vi.advanceTimersByTimeAsync(1000) + expect(t.signal.aborted).toBe(true) + expect(String(await raced)).toContain('timeout after 1000ms') + t.cancel() + }) + + it('propagates a parent abort immediately', async () => { + const parent = new AbortController() + const t = withTimeout(parent.signal, 60_000) + parent.abort() + expect(t.signal.aborted).toBe(true) + t.cancel() + }) + + it('cancel() clears the timer so nothing fires later', async () => { + const t = withTimeout(undefined, 1000) + t.cancel() + await vi.advanceTimersByTimeAsync(5000) + expect(t.signal.aborted).toBe(false) + }) +}) + +describe('retryingFetch', () => { + beforeEach(() => vi.useFakeTimers()) + afterEach(() => vi.useRealTimers()) + + it('retries a 500 then returns the success', async () => { + const impl = vi.fn().mockResolvedValueOnce(status(500)).mockResolvedValueOnce(okResponse()) + const f = retryingFetch(impl as unknown as typeof fetch, { retries: 1 }) + const p = f('https://x/') + await vi.runAllTimersAsync() + expect((await p).status).toBe(200) + expect(impl).toHaveBeenCalledTimes(2) + }) + + it('retries 429 and a rejected network error', async () => { + const impl = vi.fn() + .mockResolvedValueOnce(status(429)) + .mockRejectedValueOnce(new TypeError('fetch failed')) + .mockResolvedValueOnce(okResponse()) + const f = retryingFetch(impl as unknown as typeof fetch, { retries: 2 }) + const p = f('https://x/') + await vi.runAllTimersAsync() + expect((await p).status).toBe(200) + expect(impl).toHaveBeenCalledTimes(3) + }) + + it('returns the last 5xx response (does not throw) once retries are exhausted', async () => { + const impl = vi.fn().mockResolvedValue(status(503)) + const f = retryingFetch(impl as unknown as typeof fetch, { retries: 1 }) + const p = f('https://x/') + await vi.runAllTimersAsync() + expect((await p).status).toBe(503) + expect(impl).toHaveBeenCalledTimes(2) + }) + + it('does not retry non-retryable statuses (400/404)', async () => { + const impl = vi.fn().mockResolvedValue(status(404)) + const f = retryingFetch(impl as unknown as typeof fetch, { retries: 3 }) + expect((await f('https://x/')).status).toBe(404) + expect(impl).toHaveBeenCalledTimes(1) + }) + + it('does not retry after an abort; abort during backoff cancels the wait', async () => { + const ac = new AbortController() + const abortErr = Object.assign(new Error('aborted'), { name: 'AbortError' }) + const impl = vi.fn().mockRejectedValue(abortErr) + const f = retryingFetch(impl as unknown as typeof fetch, { retries: 3 }) + await expect(f('https://x/', { signal: ac.signal })).rejects.toMatchObject({ name: 'AbortError' }) + expect(impl).toHaveBeenCalledTimes(1) + + const impl2 = vi.fn().mockResolvedValue(status(500)) + const f2 = retryingFetch(impl2 as unknown as typeof fetch, { retries: 3 }) + const p2 = f2('https://x/', { signal: ac.signal }).catch(e => e) + ac.abort() + await vi.runAllTimersAsync() + await p2 + expect(impl2).toHaveBeenCalledTimes(1) // aborted before/during first backoff — no second attempt + }) +}) +``` + +- [ ] **Step 2: Run to verify failure** — `pnpm exec vitest run packages/core/src/__tests__/resilience.test.ts` → FAIL (module missing). + +- [ ] **Step 3: Implement** `packages/core/src/resilience.ts` (zero-network: no `fetch(` call, no endpoint literals — the injected impl is always named `fetchImpl`): + +```ts +// Resilience primitives for the search orchestrator (H8/H9). Pure: the fetch +// implementation is always injected; core never references a global fetch. + +export interface TimeoutHandle { + /** Aborts when the parent aborts OR the timer fires. Pass as ProviderContext.signal. */ + signal: AbortSignal + /** Rejects with `timeout after Nms` when the timer fires. Race the provider against it. */ + expired: Promise + /** Clear the timer + detach the parent listener. ALWAYS call once settled. */ + cancel(): void +} + +/** Compose a parent signal with a deadline — manual composition, no AbortSignal.any + * (H9: keeps core runtime-agnostic). */ +export function withTimeout(parent: AbortSignal | undefined, timeoutMs: number): TimeoutHandle { + const ctrl = new AbortController() + const onParentAbort = () => ctrl.abort(parent?.reason) + if (parent?.aborted) ctrl.abort(parent.reason) + else parent?.addEventListener('abort', onParentAbort, { once: true }) + + let rejectExpired: (e: Error) => void + const expired = new Promise((_, reject) => { rejectExpired = reject }) + expired.catch(() => {}) // the race may settle first; never let this become an unhandled rejection + const timer = setTimeout(() => { + const err = new Error(`timeout after ${timeoutMs}ms`) + ctrl.abort(err) + rejectExpired(err) + }, timeoutMs) + + return { + signal: ctrl.signal, + expired, + cancel() { + clearTimeout(timer) + parent?.removeEventListener('abort', onParentAbort) + }, + } +} + +export interface RetryOptions { + /** Extra attempts after the first (H8 default 1). */ + retries: number + /** Base backoff delay; grows 2^attempt with full jitter. Default 250. */ + baseDelayMs?: number +} + +function abortAware(ms: number, signal: AbortSignal | null | undefined): Promise { + return new Promise((resolve, reject) => { + if (signal?.aborted) return reject(signal.reason ?? new Error('aborted')) + const onAbort = () => { clearTimeout(t); reject(signal?.reason ?? new Error('aborted')) } + const t = setTimeout(() => { signal?.removeEventListener('abort', onAbort); resolve() }, ms) + signal?.addEventListener('abort', onAbort, { once: true }) + }) +} + +const isRetryableStatus = (s: number) => s === 429 || s >= 500 + +/** Wrap an injected fetch with bounded retries on 429/5xx/network-error (H8). + * Aborts are never retried; an exhausted 429/5xx returns the response so the + * provider's own `!res.ok` error path still owns the failure message. */ +export function retryingFetch(fetchImpl: typeof fetch, opts: RetryOptions): typeof fetch { + const base = opts.baseDelayMs ?? 250 + const wrapped = async ( + input: Parameters[0], + init?: Parameters[1], + ): Promise => { + for (let attempt = 0; ; attempt++) { + try { + const res = await fetchImpl(input, init) + if (!isRetryableStatus(res.status) || attempt >= opts.retries) return res + } catch (err) { + const name = (err as { name?: string } | null)?.name + if (name === 'AbortError' || init?.signal?.aborted || attempt >= opts.retries) throw err + } + // exponential backoff with full jitter; an abort during the wait cancels the retry + await abortAware(base * 2 ** attempt * (0.5 + Math.random() * 0.5), init?.signal) + } + } + return wrapped as typeof fetch +} +``` + +Add to `packages/core/src/index.ts` (new lines, keep existing exports): + +```ts +export { withTimeout, retryingFetch } from './resilience' +export type { TimeoutHandle, RetryOptions } from './resilience' +``` + +- [ ] **Step 4: Run** — `pnpm exec vitest run packages/core && pnpm --filter @refkit/core typecheck` → PASS, including `no-network.test.ts` (it scans every core source file — `resilience.ts` must not trip `\bfetch\s*\(` or `https?://`). + +- [ ] **Step 5: Commit** + +```bash +git add packages/core/src/resilience.ts packages/core/src/index.ts packages/core/src/__tests__/resilience.test.ts +git commit -m "feat(core): resilience primitives — withTimeout + retryingFetch" +``` + +--- + +### Task W2.2: client.ts — per-provider timeout, retries, latencyMs + +**Files:** +- Modify: `packages/core/src/client.ts` +- Test: `packages/core/src/__tests__/client.test.ts` + +- [ ] **Step 1: Write the failing tests.** In `client.test.ts` add (fake timers are needed ONLY in the timeout tests — scope them per-test exactly as shown, so the other async tests keep real timers): + +```ts + it('times out a hanging provider, returns partial results, and reports the timeout', async () => { + vi.useFakeTimers() + try { + const hanging = defineProvider({ + id: 'hang', modalities: ['image'], queryFeatures: ['keyword'], + search: () => new Promise(() => {}), // never settles, ignores ctx.signal + }) + const rk = createRefkit({ providers: [provider('a', [ref('a-1', 'https://a/1')]), hanging] }) + const p = rk.searchWithMeta({ query: 'x', modalities: ['image'] }) + await vi.advanceTimersByTimeAsync(10_000) + const out = await p + expect(out.references).toHaveLength(1) + const hangStatus = out.meta.providers.find(s => s.providerId === 'hang') + expect(hangStatus?.status).toBe('failed') + expect(hangStatus?.error).toContain('timeout after 10000ms') + } finally { + vi.useRealTimers() + } + }) + + it('resilience: false disables the timeout entirely', async () => { + vi.useFakeTimers() + try { + let done = false + const slow = defineProvider({ + id: 'slow', modalities: ['image'], queryFeatures: ['keyword'], + search: () => new Promise(resolve => setTimeout(() => { done = true; resolve([ref('slow-1', 'https://s/1')]) }, 60_000)), + }) + const rk = createRefkit({ providers: [slow], resilience: false }) + const p = rk.search({ query: 'x', modalities: ['image'] }) + await vi.advanceTimersByTimeAsync(60_000) + expect(await p).toHaveLength(1) + expect(done).toBe(true) + } finally { + vi.useRealTimers() + } + }) + + it('gives providers a retrying ctx.fetch: a 500-then-200 upstream succeeds transparently', async () => { + const upstream = vi.fn() + .mockResolvedValueOnce(new Response('x', { status: 500 })) + .mockResolvedValueOnce(new Response('ok', { status: 200 })) + const usesFetch = defineProvider({ + id: 'net', modalities: ['image'], queryFeatures: ['keyword'], + search: async (_q, ctx) => { + const res = await ctx.fetch('https://net/api', { signal: ctx.signal }) + if (!res.ok) throw new Error(`net failed: ${res.status}`) + return [ref('net-1', 'https://net/1')] + }, + }) + const rk = createRefkit({ providers: [usesFetch], fetch: upstream as unknown as typeof fetch, resilience: { retries: 1, timeoutMs: 10_000 } }) + const out = await rk.search({ query: 'x', modalities: ['image'] }) + expect(out).toHaveLength(1) + expect(upstream).toHaveBeenCalledTimes(2) + }) + + it('reports latencyMs on fulfilled and failed providers, not on skipped', async () => { + const textOnly = defineProvider({ id: 'text', modalities: ['text'], queryFeatures: [], search: async () => [] }) + const rk = createRefkit({ providers: [provider('a', [ref('a-1', 'https://a/1')]), failing('bad'), textOnly] }) + const out = await rk.searchWithMeta({ query: 'x', modalities: ['image'] }) + const byId = Object.fromEntries(out.meta.providers.map(s => [s.providerId, s])) + expect(byId.a.latencyMs).toEqual(expect.any(Number)) + expect(byId.bad.latencyMs).toEqual(expect.any(Number)) + expect(byId.text.latencyMs).toBeUndefined() + }) +``` + +ALSO update the existing strict assertion in `'searchWithMeta returns provider status, warnings, and gate summary'` (~line 217): statuses now carry `latencyMs`, so replace the `toEqual([...])` with: + +```ts + expect(out.meta.providers).toEqual([ + { providerId: 'ok', status: 'fulfilled', returned: 2, accepted: 2, rejected: 0, latencyMs: expect.any(Number) }, + { providerId: 'bad', status: 'failed', error: 'boom', latencyMs: expect.any(Number) }, + { providerId: 'text', status: 'skipped', reason: 'unsupported-modality' }, + ]) +``` + +- [ ] **Step 2: Run to verify failure** — `pnpm exec vitest run packages/core/src/__tests__/client.test.ts` → new tests FAIL (typecheck: `resilience` option unknown). + +- [ ] **Step 3: Implement in `client.ts`:** + +Types (additions): + +```ts +export interface ResilienceOptions { + /** Soft deadline per provider search. Default 10_000. */ + timeoutMs?: number + /** Extra fetch attempts on 429/5xx/network-error. Default 1. */ + retries?: number +} + +export interface RefkitOptions { + providers: ReferenceProvider[] + fetch?: typeof fetch + cache?: KeyValueCache + signal?: AbortSignal + merge?: MergeOptions + /** Per-provider timeout + retry (H8). Defaults ON; pass `false` to disable both. */ + resilience?: ResilienceOptions | false +} +``` + +`ProviderSearchStatus` gains `latencyMs?: number` (fulfilled/failed only). + +Imports: `import { retryingFetch, withTimeout } from './resilience'`. Constants: `const DEFAULT_TIMEOUT_MS = 10_000`, `const DEFAULT_RETRIES = 1`. + +Replace the shared-ctx + `Promise.allSettled` block (currently `const ctx: ProviderContext = {...}` through the whole `settled.forEach`) with a per-provider runner. The discriminated result keeps every downstream shape intact: + +```ts + const resilience = options.resilience === false ? undefined : { + timeoutMs: options.resilience?.timeoutMs ?? DEFAULT_TIMEOUT_MS, + retries: options.resilience?.retries ?? DEFAULT_RETRIES, + } + + type ProviderRun = + | { ok: true; valid: Reference[]; returned: number; latencyMs: number } + | { ok: false; error: unknown; latencyMs: number } + + const runProvider = async (p: ReferenceProvider): Promise => { + const started = Date.now() + const timeout = resilience ? withTimeout(input.signal ?? options.signal, resilience.timeoutMs) : undefined + const ctx: ProviderContext = { + fetch: resilience && resilience.retries > 0 ? retryingFetch(doFetch, { retries: resilience.retries }) : doFetch, + cache: options.cache, + signal: timeout?.signal ?? input.signal ?? options.signal, + } + const query = normalizeQuery({ + query: input.query, + modalities: input.modalities, + filters: input.filters, + controls: input.controls, + providerOptions: input.providerOptions, + limit: fetchLimit, + }, p) + try { + const searching = p.search(query, ctx) + searching.catch(() => {}) // a raced-past provider must not become an unhandled rejection + const raw = await (timeout ? Promise.race([searching, timeout.expired]) : searching) + const valid: Reference[] = [] + for (const item of raw) { + try { + valid.push(parseReference(item)) + } catch (error) { + input.onProviderError?.({ providerId: p.id, error }) + } + } + return { ok: true, valid, returned: raw.length, latencyMs: Date.now() - started } + } catch (error) { + input.onProviderError?.({ providerId: p.id, error }) + return { ok: false, error, latencyMs: Date.now() - started } + } finally { + timeout?.cancel() + } + } + + const runs = await Promise.all(chosen.map(runProvider)) + + const perSource: Reference[][] = [] + let anyOk = false + runs.forEach((run, i) => { + const provider = chosen[i] + if (run.ok) { + anyOk = true + statusByProvider.set(provider.id, { + providerId: provider.id, + status: 'fulfilled', + returned: run.returned, + accepted: run.valid.length, + rejected: run.returned - run.valid.length, + latencyMs: run.latencyMs, + }) + perSource.push(run.valid) + } else { + statusByProvider.set(provider.id, { providerId: provider.id, status: 'failed', error: errorSummary(run.error), latencyMs: run.latencyMs }) + } + }) + + if (!anyOk) { + throw new AggregateError(runs.filter(r => !r.ok).map(r => (r as { error: unknown }).error), 'refkit.search: all providers failed') + } +``` + +Notes for the implementer: `errorSummary` already renders `Error.message`, so the timeout error surfaces as `timeout after 10000ms`. The old `settled.forEach` block and the shared `ctx` are deleted — `onProviderError` for provider failure now fires inside `runProvider` (was previously fired in the forEach; keep exactly one call per failure). Export `ResilienceOptions` from `client.ts` and add it to the type re-exports in `packages/core/src/index.ts`. + +- [ ] **Step 4: Run** — `pnpm exec vitest run packages/core && pnpm --filter @refkit/core typecheck` → PASS (all pre-existing client tests must stay green — the AggregateError, onProviderError, malformed-item, and rerank-signal tests exercise the refactored path). + +- [ ] **Step 5: Commit** + +```bash +git add packages/core/src/client.ts packages/core/src/index.ts packages/core/src/__tests__/client.test.ts +git commit -m "feat(core): per-provider timeout, retrying fetch, latencyMs in search meta" +``` + +--- + +### Task W2.3: per-provider result cache on the KeyValueCache port + +**Files:** +- Modify: `packages/core/src/client.ts` +- Test: `packages/core/src/__tests__/client.test.ts` + +- [ ] **Step 1: Write the failing tests** — add to `client.test.ts`: + +```ts + const mapCache = () => { + const m = new Map() + return { + store: m, + ttls: [] as (number | undefined)[], + async get(k: string) { return m.get(k) }, + async set(k: string, v: string, ttlMs?: number) { m.set(k, v); this.ttls.push(ttlMs) }, + } + } + + it('serves a repeat query from the cache without re-hitting the provider', async () => { + const cache = mapCache() + let calls = 0 + const counted = defineProvider({ + id: 'c', modalities: ['image'], queryFeatures: ['keyword'], + search: async () => { calls++; return [ref('c-1', 'https://c/1')] }, + }) + const rk = createRefkit({ providers: [counted], cache }) + await rk.search({ query: 'x', modalities: ['image'] }) + const out = await rk.searchWithMeta({ query: 'x', modalities: ['image'] }) + expect(calls).toBe(1) + expect(out.references).toHaveLength(1) + expect(out.meta.providers[0]).toMatchObject({ status: 'fulfilled', cached: true }) + expect(cache.ttls).toEqual([300_000]) // default cacheTtlMs, one set for the first (live) search + }) + + it('different queries use different cache keys', async () => { + const cache = mapCache() + let calls = 0 + const counted = defineProvider({ + id: 'c', modalities: ['image'], queryFeatures: ['keyword'], + search: async () => { calls++; return [ref('c-1', 'https://c/1')] }, + }) + const rk = createRefkit({ providers: [counted], cache }) + await rk.search({ query: 'x', modalities: ['image'] }) + await rk.search({ query: 'y', modalities: ['image'] }) + expect(calls).toBe(2) + }) + + it('a corrupt or invalid cache entry falls back to a live fetch', async () => { + const cache = mapCache() + let calls = 0 + const counted = defineProvider({ + id: 'c', modalities: ['image'], queryFeatures: ['keyword'], + search: async () => { calls++; return [ref('c-1', 'https://c/1')] }, + }) + const rk = createRefkit({ providers: [counted], cache }) + await rk.search({ query: 'x', modalities: ['image'] }) + for (const k of cache.store.keys()) cache.store.set(k, '{not json') + await rk.search({ query: 'x', modalities: ['image'] }) + expect(calls).toBe(2) + }) + + it('cache errors are non-fatal: a throwing cache degrades to live search', async () => { + const broken = { + async get(): Promise { throw new Error('cache down') }, + async set(): Promise { throw new Error('cache down') }, + } + const rk = createRefkit({ providers: [provider('a', [ref('a-1', 'https://a/1')])], cache: broken }) + const out = await rk.search({ query: 'x', modalities: ['image'] }) + expect(out).toHaveLength(1) + }) + + it('honors a custom cacheTtlMs', async () => { + const cache = mapCache() + const rk = createRefkit({ providers: [provider('a', [ref('a-1', 'https://a/1')])], cache, cacheTtlMs: 1234 }) + await rk.search({ query: 'x', modalities: ['image'] }) + expect(cache.ttls).toEqual([1234]) + }) +``` + +- [ ] **Step 2: Run to verify failure** — cache is currently passed into ctx but never used by the client → repeat-query test FAILS (calls === 2), `cached`/`cacheTtlMs` don't typecheck. + +- [ ] **Step 3: Implement in `client.ts`:** + +Types: `RefkitOptions` gains `/** TTL for per-provider cached results; used only when `cache` is set. Default 300_000. */ cacheTtlMs?: number`. `ProviderSearchStatus` gains `cached?: boolean`. `ProviderRun`'s ok-arm gains `cached?: boolean`. Import `fnv1a` from `./hash` and `referenceSchema` (or reuse `parseReference` per item) from `./reference`. Constant `const DEFAULT_CACHE_TTL_MS = 300_000`. + +Inside `runProvider`, wrap the live path (H10 — per-provider, PRE-merge, stores only validated refs): + +```ts + const cacheKey = options.cache + ? `refkit:v1:${p.id}:${fnv1a(JSON.stringify(query))}` + : undefined + if (options.cache && cacheKey) { + // best-effort: a broken/corrupt/stale cache degrades to a live search + const hit = await options.cache.get(cacheKey).catch(() => undefined) + if (hit !== undefined) { + try { + const parsed = (JSON.parse(hit) as unknown[]).map(item => parseReference(item)) + return { ok: true, valid: parsed, returned: parsed.length, latencyMs: Date.now() - started, cached: true } + } catch { /* fall through to live */ } + } + } +``` + +…and after the live `valid` array is built (only on the ok path, before `return`): + +```ts + if (options.cache && cacheKey) { + // fire-and-forget: cache write failure must never fail the search + void options.cache.set(cacheKey, JSON.stringify(valid), options.cacheTtlMs ?? DEFAULT_CACHE_TTL_MS).catch(() => {}) + } +``` + +Status mapping: spread `...(run.cached ? { cached: true } : {})` into the fulfilled status. Note in a comment that cached refs keep their original `verifiedAt` (staleness bounded by the TTL). + +- [ ] **Step 4: Run** — `pnpm exec vitest run packages/core && pnpm --filter @refkit/core typecheck` → PASS. + +- [ ] **Step 5: Commit** + +```bash +git add packages/core/src/client.ts packages/core/src/__tests__/client.test.ts +git commit -m "feat(core): per-provider result cache on the KeyValueCache port" +``` + +--- + +### Task W2.4: docs, changeset, full verify + +**Files:** +- Modify: `README.md` +- Create: `.changeset/orchestrator-hardening.md` + +- [ ] **Step 1: README.** Two edits: + 1. In the Quickstart code block, the comment `// fetch defaults to globalThis.fetch — inject your own to add caching/retries` is now stale → change to `// fetch defaults to globalThis.fetch — timeouts/retries/caching are built in (see below)`. + 2. After the "Ranking & rerank" section add: + +```md +## Resilience & caching + +Fan-out is hardened by default: each provider gets a **soft 10s timeout** and **one retry** (429/5xx/network errors, exponential backoff). A slow or hanging source is reported in `meta.providers` as `failed` with `timeout after Nms` — the search still returns everyone else. Tune or disable per client: + +```ts +createRefkit({ providers, resilience: { timeoutMs: 4000, retries: 2 } }) +createRefkit({ providers, resilience: false }) // raw fan-out, no timeout/retry +``` + +Pass a `cache` to memoize **per-provider** results (keyed by provider + normalized query, TTL `cacheTtlMs`, default 5 min). Merging, reranking, and the license gate always run fresh; cache hits are flagged `cached: true` in `meta.providers`, and every provider status carries `latencyMs`: + +```ts +createRefkit({ providers, cache: myKvCache, cacheTtlMs: 60_000 }) +``` +``` + +- [ ] **Step 2: Changeset** — create `.changeset/orchestrator-hardening.md`: + +```md +--- +"@refkit/core": minor +--- + +Harden the search orchestrator: per-provider soft timeout (default 10s) and +bounded retry on 429/5xx/network errors (default 1, exponential backoff) — on by +default, tunable or disabled via `createRefkit({ resilience })`. Provider +statuses in `searchWithMeta` now carry `latencyMs`, and supplying a `cache` +(`KeyValueCache`) now memoizes per-provider results (key +`refkit:v1::`, TTL `cacheTtlMs`, default 5 min) with hits +flagged `cached: true`. Merge, rerank, and the license gate always run fresh. +New exports: `withTimeout`, `retryingFetch`, `ResilienceOptions`. +``` + +- [ ] **Step 3: Full verify** — `pnpm -r --parallel typecheck && pnpm test:run` → ALL green (Wave 1 baseline 259 + the new resilience/client tests). Confirm `no-network` still green. + +- [ ] **Step 4: Commit** + +```bash +git add README.md .changeset/orchestrator-hardening.md +git commit -m "docs: resilience & caching section + changeset for orchestrator hardening" +``` + +--- + +## Self-review notes + +- H8 (Task W2.2), H9 (Task W2.1 — no AbortSignal.any; no-network stays green), H10 (Task W2.3), H11 (Task W2.2). +- The pre-existing strict `toEqual` on `meta.providers` (~client.test.ts:217) is explicitly updated in Task W2.2 — do not skip it. +- MCP's `searchMetaSchema` (packages/mcp/src/index.ts) validates provider statuses — check in Task W2.4's full verify: its zod object for providers has no `latencyMs`/`cached` keys; zod objects strip unknown keys by default, but the schema is typed `z.ZodType` — if typecheck fails there, add `latencyMs: z.number().optional(), cached: z.boolean().optional()` to the mcp provider-status schema and include the file in Task W2.4's commit with a note. +- Line anchors may drift — locate by quoted code. From 1d4c39aabe8f16010669cbf03f59147c948c6f7a Mon Sep 17 00:00:00 2001 From: MyPrototypeWhat Date: Fri, 3 Jul 2026 16:24:00 +0800 Subject: [PATCH 9/9] =?UTF-8?q?fix:=20review=20findings=20=E2=80=94=20cach?= =?UTF-8?q?e-hit=20timeout=20leak,=20per-item=20cached=20validation,=20sta?= =?UTF-8?q?ble=20keys=20w/=20undefined,=20parent-abort=20fast-fail,=20Requ?= =?UTF-8?q?est=20signal,=20race=20dedup?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/core/src/__tests__/client.test.ts | 94 +++++++++++++++++++ .../core/src/__tests__/resilience.test.ts | 33 +++++++ packages/core/src/client.ts | 72 +++++++++----- packages/core/src/resilience.ts | 22 +++-- 4 files changed, 189 insertions(+), 32 deletions(-) diff --git a/packages/core/src/__tests__/client.test.ts b/packages/core/src/__tests__/client.test.ts index e41efaa..c5e642b 100644 --- a/packages/core/src/__tests__/client.test.ts +++ b/packages/core/src/__tests__/client.test.ts @@ -105,6 +105,25 @@ describe('createRefkit', () => { globalFetchSpy.mockRestore() }) + it('resolves globalThis.fetch at search time, not at createRefkit time (late-binding)', async () => { + // createRefkit is called BEFORE globalThis.fetch is replaced — a client that + // resolved options.fetch ?? globalThis.fetch once at creation time would be + // stuck delegating to the pre-replacement implementation forever. + let capturedFetch: typeof fetch | undefined + const capturingProvider = defineProvider({ + id: 'cap', + modalities: ['image'], + queryFeatures: ['keyword'], + search: async (_q, ctx) => { capturedFetch = ctx.fetch; await ctx.fetch('https://cap/late'); return [] }, + }) + const rk = createRefkit({ providers: [capturingProvider] }) + const globalFetchSpy = vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response('ok', { status: 200 })) + await rk.search({ query: 'x', modalities: ['image'] }) + expect(capturedFetch).not.toBe(globalThis.fetch) // still wrapped by retryingFetch + expect(globalFetchSpy.mock.calls[0]?.[0]).toBe('https://cap/late') // delegated to the NEW globalThis.fetch + globalFetchSpy.mockRestore() + }) + it('throws a clear Error (not AggregateError) when no provider supports the requested modality', async () => { const imageOnly = provider('img', [ref('img-1', 'https://img/1')]) const rk = createRefkit({ providers: [imageOnly] }) @@ -333,6 +352,27 @@ describe('createRefkit', () => { } }) + it('a user abort of input.signal fast-fails a provider that ignores ctx.signal, instead of waiting for the deadline', async () => { + vi.useFakeTimers() + try { + const ac = new AbortController() + const ignoresSignal = defineProvider({ + id: 'ignorer', modalities: ['image'], queryFeatures: ['keyword'], + search: () => new Promise(() => {}), // never settles, never looks at ctx.signal + }) + const rk = createRefkit({ providers: [ignoresSignal] }) + const p = rk.search({ query: 'x', modalities: ['image'], signal: ac.signal }).catch(e => e) + ac.abort(new Error('user cancelled')) + // advance only a small amount — far less than the 10s default deadline — + // the search must already have settled from the parent abort, not the timer + await vi.advanceTimersByTimeAsync(50) + const result = await p + expect(result).toBeInstanceOf(AggregateError) // all providers failed → AggregateError + } finally { + vi.useRealTimers() + } + }) + it('gives providers a retrying ctx.fetch: a 500-then-200 upstream succeeds transparently', async () => { const upstream = vi.fn() .mockResolvedValueOnce(new Response('x', { status: 500 })) @@ -474,4 +514,58 @@ describe('createRefkit', () => { await rk.search({ query: 'x', modalities: ['image'], providerOptions: { c: { a: 1, b: 2 } } }) expect(calls).toBe(1) // second search is a cache hit despite the different key order }) + + it('a cache hit cancels its timeout handle — no leaked timers/listeners across repeated hits', async () => { + vi.useFakeTimers() + try { + const cache = mapCache() + const counted = defineProvider({ + id: 'c', modalities: ['image'], queryFeatures: ['keyword'], + search: async () => [ref('c-1', 'https://c/1')], + }) + const rk = createRefkit({ providers: [counted], cache }) + await rk.search({ query: 'x', modalities: ['image'] }) // live search, populates cache + expect(vi.getTimerCount()).toBe(0) + for (let i = 0; i < 5; i++) { + await rk.search({ query: 'x', modalities: ['image'] }) // cache hit + expect(vi.getTimerCount()).toBe(0) // timeout handle must be cancelled on every exit path + } + } finally { + vi.useRealTimers() + } + }) + + it('a cache entry with one malformed item validates per-item (matches the live path): good kept, bad reported+rejected', async () => { + const cache = mapCache() + const onProviderError = vi.fn() + const counted = defineProvider({ + id: 'c', modalities: ['image'], queryFeatures: ['keyword'], + search: async () => [ref('c-1', 'https://c/1')], + }) + const rk = createRefkit({ providers: [counted], cache }) + await rk.search({ query: 'x', modalities: ['image'] }) // live search, populates cache + // seed the cache entry with 1 valid + 1 malformed item + for (const k of cache.store.keys()) { + const valid = JSON.parse(cache.store.get(k)!) + cache.store.set(k, JSON.stringify([...valid, { id: '', modality: 'image' }])) + } + const out = await rk.searchWithMeta({ query: 'x', modalities: ['image'], onProviderError }) + expect(out.references.map(r => r.id)).toEqual(['c-1']) + expect(out.meta.providers[0]).toMatchObject({ cached: true, returned: 2, accepted: 1, rejected: 1 }) + expect(onProviderError).toHaveBeenCalledTimes(1) + expect(onProviderError).toHaveBeenCalledWith(expect.objectContaining({ providerId: 'c' })) + }) + + it('stableStringify cache keys treat undefined-valued keys the same as absent keys', async () => { + const cache = mapCache() + let calls = 0 + const counted = defineProvider({ + id: 'c', modalities: ['image'], queryFeatures: ['keyword'], + search: async () => { calls++; return [ref('c-1', 'https://c/1')] }, + }) + const rk = createRefkit({ providers: [counted], cache }) + await rk.search({ query: 'x', modalities: ['image'], providerOptions: { c: { a: 1, b: undefined } } }) + await rk.search({ query: 'x', modalities: ['image'], providerOptions: { c: { a: 1 } } }) + expect(calls).toBe(1) // second search hits the same cache entry as the first + }) }) diff --git a/packages/core/src/__tests__/resilience.test.ts b/packages/core/src/__tests__/resilience.test.ts index ea5d4dd..0c644bd 100644 --- a/packages/core/src/__tests__/resilience.test.ts +++ b/packages/core/src/__tests__/resilience.test.ts @@ -26,6 +26,28 @@ describe('withTimeout', () => { t.cancel() }) + it('a parent abort also rejects `expired` (fast-fails a race against a provider ignoring ctx.signal)', async () => { + const parent = new AbortController() + const t = withTimeout(parent.signal, 60_000) + const raced = Promise.race([new Promise(() => {}), t.expired]).catch(e => e) + const reason = new Error('user cancelled') + parent.abort(reason) + expect(t.signal.aborted).toBe(true) + expect(await raced).toBe(reason) + t.cancel() + }) + + it('a parent abort without an Error reason still rejects `expired` with an Error', async () => { + const parent = new AbortController() + const t = withTimeout(parent.signal, 60_000) + const raced = Promise.race([new Promise(() => {}), t.expired]).catch(e => e) + parent.abort('some string reason') + const err = await raced + expect(err).toBeInstanceOf(Error) + expect(String(err)).toContain('some string reason') + t.cancel() + }) + it('cancel() clears the timer so nothing fires later', async () => { const t = withTimeout(undefined, 1000) t.cancel() @@ -123,4 +145,15 @@ describe('retryingFetch', () => { await expect(f('https://x/', { signal: ac.signal })).rejects.toBe(timeoutErr) expect(impl).toHaveBeenCalledTimes(1) }) + + it('does not retry when the abort signal is carried by a Request object, not init.signal', async () => { + const ac = new AbortController() + ac.abort(new Error('user cancelled')) + const req = new Request('https://x/', { signal: ac.signal }) + const plainErr = new Error('network down') // impl rejects with a plain Error, no name: 'AbortError' + const impl = vi.fn().mockRejectedValue(plainErr) + const f = retryingFetch(impl as unknown as typeof fetch, { retries: 3 }) + await expect(f(req)).rejects.toBe(plainErr) + expect(impl).toHaveBeenCalledTimes(1) // aborted Request signal → no retry + }) }) diff --git a/packages/core/src/client.ts b/packages/core/src/client.ts index 41654fb..8dfa756 100644 --- a/packages/core/src/client.ts +++ b/packages/core/src/client.ts @@ -132,11 +132,13 @@ function errorSummary(error: unknown): string { // Deterministic JSON for cache keys: object keys sorted recursively, so a // caller's providerOptions key order can't split otherwise-identical keys. +// Keys whose value is `undefined` are skipped, matching JSON.stringify's own +// object semantics — `{ a: 1, b: undefined }` and `{ a: 1 }` must key alike. function stableStringify(value: unknown): string { if (Array.isArray(value)) return `[${value.map(stableStringify).join(',')}]` if (value && typeof value === 'object') { const rec = value as Record - return `{${Object.keys(rec).sort().map(k => `${JSON.stringify(k)}:${stableStringify(rec[k])}`).join(',')}}` + return `{${Object.keys(rec).filter(k => rec[k] !== undefined).sort().map(k => `${JSON.stringify(k)}:${stableStringify(rec[k])}`).join(',')}}` } return JSON.stringify(value) ?? 'null' } @@ -175,6 +177,10 @@ export function createRefkit(options: RefkitOptions): RefkitClient { timeoutMs: options.resilience?.timeoutMs ?? DEFAULT_TIMEOUT_MS, retries: options.resilience?.retries ?? DEFAULT_RETRIES, } + // Built once per search (doFetch/retries are fixed for the whole call) and + // shared across every provider in the fan-out below, instead of allocating + // a fresh wrapper per provider. + const sharedFetch = resilience && resilience.retries > 0 ? retryingFetch(doFetch, { retries: resilience.retries }) : doFetch type ProviderRun = | { ok: true; valid: Reference[]; returned: number; latencyMs: number; cached?: boolean } @@ -184,7 +190,7 @@ export function createRefkit(options: RefkitOptions): RefkitClient { const started = Date.now() const timeout = resilience ? withTimeout(input.signal ?? options.signal, resilience.timeoutMs) : undefined const ctx: ProviderContext = { - fetch: resilience && resilience.retries > 0 ? retryingFetch(doFetch, { retries: resilience.retries }) : doFetch, + fetch: sharedFetch, cache: options.cache, signal: timeout?.signal ?? input.signal ?? options.signal, } @@ -199,30 +205,15 @@ export function createRefkit(options: RefkitOptions): RefkitClient { const cacheKey = options.cache ? `refkit:v1:${p.id}:${fnv1a(stableStringify(query))}` : undefined - if (options.cache && cacheKey) { - // best-effort: a broken/corrupt/stale cache degrades to a live search - const pending = options.cache.get(cacheKey) - // A slow cache read must not outlive the provider deadline: at expiry it - // degrades to a miss, and the live search below then fails fast on the - // same (already-expired) deadline. - const hit = await (timeout - ? Promise.race([pending, timeout.expired.catch(() => undefined)]) - : pending - ).catch(() => undefined) - pending.catch(() => {}) // raced-past rejection must not go unhandled - if (hit !== undefined) { - try { - // cached refs keep their original verifiedAt — staleness is bounded - // by the TTL when the cache honors ttlMs - const parsed = (JSON.parse(hit) as unknown[]).map(item => parseReference(item)) - return { ok: true, valid: parsed, returned: parsed.length, latencyMs: Date.now() - started, cached: true } - } catch { /* fall through to live */ } - } + // Race a promise against the deadline without leaking an unhandled rejection + // for whichever side loses the race. + const raceDeadline = (p: Promise): Promise => { + p.catch(() => {}) + return timeout ? Promise.race([p, timeout.expired]) : p } - try { - const searching = p.search(query, ctx) - searching.catch(() => {}) // a raced-past provider must not become an unhandled rejection - const raw = await (timeout ? Promise.race([searching, timeout.expired]) : searching) + // Parse raw provider items one at a time — a single bad item must not + // discard the rest (shared by the cache-hit and live-search paths). + const parseItems = (raw: unknown[]): Reference[] => { const valid: Reference[] = [] for (const item of raw) { try { @@ -231,6 +222,37 @@ export function createRefkit(options: RefkitOptions): RefkitClient { input.onProviderError?.({ providerId: p.id, error }) } } + return valid + } + try { + if (options.cache && cacheKey) { + // best-effort: a broken/corrupt/stale cache degrades to a live search + const pending = options.cache.get(cacheKey) + // A slow cache read must not outlive the provider deadline: at expiry it + // degrades to a miss, and the live search below then fails fast on the + // same (already-expired) deadline. + const hit = await (timeout + ? Promise.race([pending, timeout.expired.catch(() => undefined)]) + : pending + ).catch(() => undefined) + pending.catch(() => {}) // raced-past rejection must not go unhandled + if (hit !== undefined) { + try { + // cached refs keep their original verifiedAt — staleness is bounded + // by the TTL when the cache honors ttlMs. Only a whole-payload + // failure (not JSON, not an array) falls through to live; a single + // bad item within an otherwise-valid array is reported and dropped, + // same as the live path. + const raw = JSON.parse(hit) as unknown[] + if (!Array.isArray(raw)) throw new Error('cached payload is not an array') + const valid = parseItems(raw) + return { ok: true, valid, returned: raw.length, latencyMs: Date.now() - started, cached: true } + } catch { /* fall through to live */ } + } + } + const searching = p.search(query, ctx) + const raw = await raceDeadline(searching) + const valid = parseItems(raw) if (options.cache && cacheKey) { // fire-and-forget: cache write failure must never fail the search void options.cache.set(cacheKey, JSON.stringify(valid), options.cacheTtlMs ?? DEFAULT_CACHE_TTL_MS).catch(() => {}) diff --git a/packages/core/src/resilience.ts b/packages/core/src/resilience.ts index 3f9a12c..3e8e712 100644 --- a/packages/core/src/resilience.ts +++ b/packages/core/src/resilience.ts @@ -4,7 +4,9 @@ export interface TimeoutHandle { /** Aborts when the parent aborts OR the timer fires. Pass as ProviderContext.signal. */ signal: AbortSignal - /** Rejects with `timeout after Nms` when the timer fires. Race the provider against it. */ + /** Rejects when the deadline fires (`timeout after Nms`) OR the parent aborts + * (with the parent's reason) — whichever comes first. Race the provider against + * it so a user abort fast-fails even a provider that ignores ctx.signal. */ expired: Promise /** Clear the timer + detach the parent listener. ALWAYS call once settled. */ cancel(): void @@ -15,16 +17,19 @@ export interface TimeoutHandle { export function withTimeout(parent: AbortSignal | undefined, timeoutMs: number): TimeoutHandle { const ctrl = new AbortController() let timer: ReturnType | undefined + + let rejectExpired: (e: Error) => void + const expired = new Promise((_, reject) => { rejectExpired = reject }) + expired.catch(() => {}) // the race may settle first; never let this become an unhandled rejection + const onParentAbort = () => { clearTimeout(timer) // self-clean: a parent abort makes the deadline moot ctrl.abort(parent?.reason) + rejectExpired(parent?.reason instanceof Error ? parent.reason : new Error(String(parent?.reason ?? 'aborted'))) } - if (parent?.aborted) ctrl.abort(parent.reason) + if (parent?.aborted) onParentAbort() else parent?.addEventListener('abort', onParentAbort, { once: true }) - let rejectExpired: (e: Error) => void - const expired = new Promise((_, reject) => { rejectExpired = reject }) - expired.catch(() => {}) // the race may settle first; never let this become an unhandled rejection timer = setTimeout(() => { const err = new Error(`timeout after ${timeoutMs}ms`) ctrl.abort(err) @@ -68,6 +73,9 @@ export function retryingFetch(fetchImpl: typeof fetch, opts: RetryOptions): type input: Parameters[0], init?: Parameters[1], ): Promise => { + // A signal can be carried on `init.signal` OR on a `Request` object passed + // as `input` — `init.signal` takes precedence (matches fetch's own rule). + const signal = init?.signal ?? (typeof Request !== 'undefined' && input instanceof Request ? input.signal : undefined) for (let attempt = 0; ; attempt++) { let discarded: Response | undefined try { @@ -76,12 +84,12 @@ export function retryingFetch(fetchImpl: typeof fetch, opts: RetryOptions): type discarded = res } catch (err) { const name = (err as { name?: string } | null)?.name - if (name === 'AbortError' || init?.signal?.aborted || attempt >= opts.retries) throw err + if (name === 'AbortError' || signal?.aborted || attempt >= opts.retries) throw err } // drain the discarded body so undici can reuse the socket during retries void discarded?.body?.cancel().catch(() => {}) // exponential backoff with equal jitter (half fixed + half random); an abort during the wait cancels the retry - await abortAware(base * 2 ** attempt * (0.5 + Math.random() * 0.5), init?.signal) + await abortAware(base * 2 ** attempt * (0.5 + Math.random() * 0.5), signal) } } return wrapped as typeof fetch