fix(host): await cancelled cold-summary batches
This commit is contained in:
@@ -46,7 +46,7 @@ const DEFAULT_MAX_MESSAGES = 50
|
|||||||
/** Provider work budget: at most 100 calls and 2,000 inspected hits. */
|
/** Provider work budget: at most 100 calls and 2,000 inspected hits. */
|
||||||
const SESSION_SEARCH_PROVIDER_CALL_LIMIT = 100
|
const SESSION_SEARCH_PROVIDER_CALL_LIMIT = 100
|
||||||
|
|
||||||
/** Bound cold-log stat fan-out so an aborted search stops launching new work. */
|
/** Bound cold-log stat fan-out and settle each started batch before cancellation returns. */
|
||||||
const COLD_SUMMARY_BATCH_SIZE = 16
|
const COLD_SUMMARY_BATCH_SIZE = 16
|
||||||
|
|
||||||
/** Surface message event types (the pagination counting unit). */
|
/** Surface message event types (the pagination counting unit). */
|
||||||
@@ -615,10 +615,23 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
|||||||
for (let offset = 0; offset < cold.length; offset += COLD_SUMMARY_BATCH_SIZE) {
|
for (let offset = 0; offset < cold.length; offset += COLD_SUMMARY_BATCH_SIZE) {
|
||||||
signal?.throwIfAborted()
|
signal?.throwIfAborted()
|
||||||
const batch = cold.slice(offset, offset + COLD_SUMMARY_BATCH_SIZE)
|
const batch = cold.slice(offset, offset + COLD_SUMMARY_BATCH_SIZE)
|
||||||
items.push(...await Promise.all(
|
const settled = await Promise.allSettled(
|
||||||
batch.map(meta => summarizeCold(persistence, meta, signal)),
|
batch.map(meta => summarizeCold(persistence, meta, signal)),
|
||||||
))
|
)
|
||||||
|
const summaries: SessionSummary[] = []
|
||||||
|
let rejected = false
|
||||||
|
let failure: unknown
|
||||||
|
for (const result of settled) {
|
||||||
|
if (result.status === 'fulfilled') {
|
||||||
|
summaries.push(result.value)
|
||||||
|
} else if (!rejected) {
|
||||||
|
rejected = true
|
||||||
|
failure = result.reason
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (rejected) throw failure
|
||||||
signal?.throwIfAborted()
|
signal?.throwIfAborted()
|
||||||
|
items.push(...summaries)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
items.sort((a, b) => b.updatedAt - a.updatedAt)
|
items.sort((a, b) => b.updatedAt - a.updatedAt)
|
||||||
|
|||||||
@@ -6,6 +6,7 @@
|
|||||||
|
|
||||||
import { describe, expect, it, vi } from 'vitest'
|
import { describe, expect, it, vi } from 'vitest'
|
||||||
import { Context } from 'cordis'
|
import { Context } from 'cordis'
|
||||||
|
import { stat } from 'node:fs/promises'
|
||||||
import AgentRegistry from '@deepseek-ai/dsh-agent'
|
import AgentRegistry from '@deepseek-ai/dsh-agent'
|
||||||
import SessionStore from '@deepseek-ai/dsh-session'
|
import SessionStore from '@deepseek-ai/dsh-session'
|
||||||
import type { SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
|
import type { SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
|
||||||
@@ -19,6 +20,11 @@ import type { RpcRequest } from '@deepseek-ai/dsh-host-apiproxy/api'
|
|||||||
import { RpcId } from '@deepseek-ai/dsh-host-apiproxy/api'
|
import { RpcId } from '@deepseek-ai/dsh-host-apiproxy/api'
|
||||||
import { createApiProxy } from '@deepseek-ai/dsh-host-apiproxy'
|
import { createApiProxy } from '@deepseek-ai/dsh-host-apiproxy'
|
||||||
|
|
||||||
|
vi.mock('node:fs/promises', async (importOriginal) => {
|
||||||
|
const actual = await importOriginal<typeof import('node:fs/promises')>()
|
||||||
|
return { ...actual, stat: vi.fn(actual.stat) }
|
||||||
|
})
|
||||||
|
|
||||||
const sid = (value: string): SessionId => value as SessionId
|
const sid = (value: string): SessionId => value as SessionId
|
||||||
const defaults = { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' }
|
const defaults = { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' }
|
||||||
|
|
||||||
@@ -777,6 +783,48 @@ describe('session.search', () => {
|
|||||||
expect(searchSessions).not.toHaveBeenCalled()
|
expect(searchSessions).not.toHaveBeenCalled()
|
||||||
})
|
})
|
||||||
|
|
||||||
|
it('awaits every started cold-summary stat before returning cancellation', async () => {
|
||||||
|
const ctx = await baseContext()
|
||||||
|
const controller = new AbortController()
|
||||||
|
const cold = Array.from({ length: 16 }, (_, index) => header(`cold-${index}`, `/cold-${index}`))
|
||||||
|
const statGates = cold.map(() => Promise.withResolvers<{ mtimeMs: number }>())
|
||||||
|
const statMock = vi.mocked(stat)
|
||||||
|
statMock.mockClear()
|
||||||
|
for (const gate of statGates) {
|
||||||
|
statMock.mockImplementationOnce((() => gate.promise) as never)
|
||||||
|
}
|
||||||
|
ctx.provide('sessionPersistence', {
|
||||||
|
list: () => Promise.resolve(cold),
|
||||||
|
locate: (meta: SessionHeader) => ({ kind: 'jsonl', path: `/logs/${meta.id}.jsonl` }),
|
||||||
|
} as never)
|
||||||
|
const searchSessions = vi.fn()
|
||||||
|
ctx.provide('sessionQuery', { searchSessions } as never)
|
||||||
|
|
||||||
|
let settled = false
|
||||||
|
const responsePromise = createApiProxy(ctx, defaults).sessions.search(
|
||||||
|
request('cancel-during-cold-stats'),
|
||||||
|
controller.signal,
|
||||||
|
).finally(() => {
|
||||||
|
settled = true
|
||||||
|
})
|
||||||
|
await vi.waitFor(() => {
|
||||||
|
expect(statMock).toHaveBeenCalledTimes(16)
|
||||||
|
})
|
||||||
|
|
||||||
|
controller.abort()
|
||||||
|
statGates[0]!.resolve({ mtimeMs: 101 })
|
||||||
|
await new Promise<void>(resolve => setImmediate(resolve))
|
||||||
|
expect(settled).toBe(false)
|
||||||
|
|
||||||
|
for (const gate of statGates.slice(1)) gate.resolve({ mtimeMs: 102 })
|
||||||
|
const response = await responsePromise
|
||||||
|
expect(response.result).toMatchObject({
|
||||||
|
ok: false,
|
||||||
|
error: { code: 'cancelled' },
|
||||||
|
})
|
||||||
|
expect(searchSessions).not.toHaveBeenCalled()
|
||||||
|
})
|
||||||
|
|
||||||
it('maps missing composition, query cancellation, and provider failure', async () => {
|
it('maps missing composition, query cancellation, and provider failure', async () => {
|
||||||
const missingCtx = await baseContext()
|
const missingCtx = await baseContext()
|
||||||
missingCtx.sessions.create(sid('visible'), { meta: header('visible') })
|
missingCtx.sessions.create(sid('visible'), { meta: header('visible') })
|
||||||
|
|||||||
Reference in New Issue
Block a user