diff --git a/layers/artifact-delivery/server/api/v1/resolutions/[id].get.ts b/layers/artifact-delivery/server/api/v1/resolutions/[id].get.ts index 279a3620..5d6fa32c 100644 --- a/layers/artifact-delivery/server/api/v1/resolutions/[id].get.ts +++ b/layers/artifact-delivery/server/api/v1/resolutions/[id].get.ts @@ -10,6 +10,7 @@ import { } from '../../../utils/github-app' import { canReadPrivateResolution } from '../../../utils/private-access' import { privateArtifactAccessEnabled } from '../../../utils/private-feature' +import { waitForResolutionChange } from '../../../utils/resolution-wait' import { setSkillPageUrlHeader } from '../../../utils/skill-page' import { getResolution } from '../../../utils/state' @@ -50,8 +51,15 @@ export default withArtifactProblems(defineApiHandler({ throw createError({ statusCode: 404, message: 'Resolution not found' }) } } - await setSkillPageUrlHeader(event, row) - return row + // A building Resolution answers when its state changes, so the CLI sees + // each stage, and the finished build, within one check of it landing. + const current = await waitForResolutionChange({ + load: () => getResolution(platform.db, row.id), + sleep: milliseconds => new Promise(resolve => setTimeout(resolve, milliseconds)), + now: Date.now, + }, row) + await setSkillPageUrlHeader(event, current) + return current }, presenter: presentArtifactResolution, })) diff --git a/layers/artifact-delivery/server/api/v1/resolutions/index.post.ts b/layers/artifact-delivery/server/api/v1/resolutions/index.post.ts index 631a6983..98da91b5 100644 --- a/layers/artifact-delivery/server/api/v1/resolutions/index.post.ts +++ b/layers/artifact-delivery/server/api/v1/resolutions/index.post.ts @@ -1,3 +1,4 @@ +import type { H3Event } from 'h3' import { getHeader, setHeader, setResponseStatus } from 'h3' import { defineApiHandler } from '#shared/server/handler' import { presentArtifactResolution } from '../../../presenters/resolution' @@ -7,11 +8,13 @@ import { problemCodeSchema, } from '../../../schemas/contracts' import { withArtifactProblems } from '../../../utils/artifact-problem' +import { failResolution } from '../../../utils/build' import { findPrivateRepositoryAccess } from '../../../utils/private-access' import { privateArtifactAccessEnabled } from '../../../utils/private-feature' import { enqueueArtifactBuild } from '../../../utils/queue' -import { fetchAdmittedSkillIdentity, requestResolution } from '../../../utils/request-resolution' +import { enqueueAfterResponse, fetchAdmittedSkillIdentity, requestResolution } from '../../../utils/request-resolution' import { setSkillPageUrlHeader } from '../../../utils/skill-page' +import { getResolution } from '../../../utils/state' export default withArtifactProblems(defineApiHandler({ schema: createResolutionRequestSchema, @@ -36,7 +39,21 @@ export default withArtifactProblems(defineApiHandler({ const result = await requestResolution({ db: platform.db, lookupAdmitted: fetchAdmittedSkillIdentity(event.context), - enqueue: resolutionId => enqueueArtifactBuild(platform.env, resolutionId), + enqueue: enqueueAfterResponse({ + schedule: work => runAfterResponse(event, work), + enqueue: resolutionId => enqueueArtifactBuild(platform.env, resolutionId), + failUnqueued: async (resolutionId, error) => { + console.error(JSON.stringify({ + operation: 'artifact-build', + outcome: 'enqueue-failed', + resolutionId, + error: error instanceof Error ? error.message : String(error), + })) + const row = await getResolution(platform.db, resolutionId) + if (row) + await failResolution({ db: platform.db, now: () => Math.floor(Date.now() / 1000) }, row, 'SERVICE_UNAVAILABLE', true) + }, + }), now: () => Math.floor(Date.now() / 1000), }, { source: body.source, @@ -88,3 +105,20 @@ export default withArtifactProblems(defineApiHandler({ }, presenter: presentArtifactResolution, })) + +/** + * Keep the queue send alive past the response on Workers. + * + * H3Event has no `waitUntil` on h3 1.15, so the send goes through the + * Cloudflare context Nitro mounts. Off Workers (local dev, tests) it does not + * block the response. The registry layer keeps the same helper, and ADR-0001 + * keeps utilities out of other layers, so this route holds its own copy. + */ +function runAfterResponse(event: H3Event, promise: Promise): void { + const ctx = (event.context as { cloudflare?: { context?: { waitUntil?: (p: Promise) => void } } }).cloudflare?.context + if (ctx?.waitUntil) { + ctx.waitUntil(promise) + return + } + void promise +} diff --git a/layers/artifact-delivery/server/utils/request-resolution.ts b/layers/artifact-delivery/server/utils/request-resolution.ts index d9556bb6..83fdaa93 100644 --- a/layers/artifact-delivery/server/utils/request-resolution.ts +++ b/layers/artifact-delivery/server/utils/request-resolution.ts @@ -60,3 +60,31 @@ export async function requestResolution( await dependencies.enqueue(result.row.id) return result } + +export interface AfterResponseEnqueueDependencies { + /** Keep work alive after the response, such as the Worker's `waitUntil`. */ + schedule: (work: Promise) => void + enqueue: (resolutionId: string) => Promise + /** Settle a Resolution whose build never reached the queue. */ + failUnqueued: (resolutionId: string, error: unknown) => Promise +} + +/** + * An `enqueue` that sends the build message after the response. + * + * The queue send took 446 to 500 ms of a 590 to 650 ms request in three + * production traces from Sydney and Bangkok on 2026-10-07. The CLI polls + * for the result anyway, so the send need not hold the first answer. A send + * that fails settles the Resolution as a retryable failure, so the polling + * CLI requests a new Resolution instead of waiting out its deadline. + */ +export function enqueueAfterResponse( + dependencies: AfterResponseEnqueueDependencies, +): (resolutionId: string) => Promise { + return async (resolutionId) => { + dependencies.schedule( + dependencies.enqueue(resolutionId) + .catch(error => dependencies.failUnqueued(resolutionId, error)), + ) + } +} diff --git a/layers/artifact-delivery/server/utils/resolution-wait.ts b/layers/artifact-delivery/server/utils/resolution-wait.ts new file mode 100644 index 00000000..9e2314bf --- /dev/null +++ b/layers/artifact-delivery/server/utils/resolution-wait.ts @@ -0,0 +1,51 @@ +import type { ResolutionRow } from './state' +import { ACTIVE_BUILD_STATES } from './state' + +/** The longest one Resolution read waits for its build to move. */ +export const RESOLUTION_WAIT_MS = 8_000 + +/** How often a waiting read loads the Resolution again. */ +export const RESOLUTION_CHECK_MS = 250 + +export interface ResolutionWaitDependencies { + /** Load the Resolution as D1 holds it now. */ + load: () => Promise + sleep: (milliseconds: number) => Promise + /** Milliseconds since the epoch. */ + now: () => number +} + +/** + * The Resolution once its build leaves the state it had, or as it stands + * when the wait runs out. + * + * A read used to answer at once, and the CLI came back a second later. A + * build that finished just after a read cost the run up to that second, and a + * Resolution passes through seven states. A held read answers within one + * check of each change, so the CLI sees `ready` about 250 ms after it lands. + * A settled Resolution answers at once. + */ +export async function waitForResolutionChange( + dependencies: ResolutionWaitDependencies, + row: ResolutionRow, + budgetMs: number = RESOLUTION_WAIT_MS, +): Promise { + if (!isBuilding(row)) + return row + const deadline = dependencies.now() + budgetMs + let current = row + while (dependencies.now() + RESOLUTION_CHECK_MS <= deadline) { + await dependencies.sleep(RESOLUTION_CHECK_MS) + const next = await dependencies.load() + if (!next) + return current + current = next + if (current.state !== row.state) + return current + } + return current +} + +function isBuilding(row: ResolutionRow): boolean { + return (ACTIVE_BUILD_STATES as readonly string[]).includes(row.state) +} diff --git a/layers/artifact-delivery/server/utils/state.ts b/layers/artifact-delivery/server/utils/state.ts index bea13ef6..3d1988df 100644 --- a/layers/artifact-delivery/server/utils/state.ts +++ b/layers/artifact-delivery/server/utils/state.ts @@ -468,7 +468,9 @@ export function presentResolution(row: ResolutionRow): ResolutionResponse { state: 'pending', resolutionId: row.id, stage: row.state as ActiveBuildState, - pollAfterMs: 1000, + // A read of a building Resolution waits for its next state, so the + // CLI may come back at once. See `waitForResolutionChange`. + pollAfterMs: 250, } } if (row.state === 'blocked') { diff --git a/test/unit/artifact-resolution-latency.test.ts b/test/unit/artifact-resolution-latency.test.ts new file mode 100644 index 00000000..d336ef76 --- /dev/null +++ b/test/unit/artifact-resolution-latency.test.ts @@ -0,0 +1,254 @@ +import type { SourceRequest } from '../../layers/artifact-delivery/server/schemas/contracts' +import type { ResolutionRow } from '../../layers/artifact-delivery/server/utils/state' +import { createApp, createError, eventHandler, readBody, toWebHandler } from 'h3' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { failResolution } from '../../layers/artifact-delivery/server/utils/build' +import { enqueueAfterResponse } from '../../layers/artifact-delivery/server/utils/request-resolution' +import { + RESOLUTION_CHECK_MS, + RESOLUTION_WAIT_MS, + waitForResolutionChange, +} from '../../layers/artifact-delivery/server/utils/resolution-wait' +import { + createResolution, + getResolution, + presentResolution, + resolutionRequestIdentity, +} from '../../layers/artifact-delivery/server/utils/state' +import { createSqliteD1 } from './helpers/d1-sqlite' + +const NOW = 1_787_227_200 +const request: SourceRequest = { + provider: 'github', + owner: 'skilld-dev', + repository: 'skills', + selector: { type: 'named-skill', name: 'demo' }, +} + +const MIGRATIONS = [ + 'migrations/0017_users.sql', + 'migrations/0110_artifact_delivery.sql', + 'migrations/0111_github_app_delivery.sql', + 'migrations/0122_artifact_resolution_retry_after.sql', +] + +async function requestedRow(): Promise { + const sqlite = createSqliteD1(MIGRATIONS) + const identity = await resolutionRequestIdentity(request, 'resolution-wait-test-key') + const created = await createResolution(sqlite.db, request, identity, NOW) + sqlite.close() + if (created._tag === 'idempotency-conflict') + throw new Error('Test Resolution conflicted') + return created.row +} + +/** A clock that only moves when the wait sleeps, and a row that changes on a schedule. */ +function scripted(states: Array<{ atMs: number, row: ResolutionRow }>) { + let clock = 0 + let loads = 0 + return { + loads: () => loads, + elapsed: () => clock, + dependencies: { + now: () => clock, + sleep: async (milliseconds: number) => { + clock += milliseconds + }, + load: async () => { + loads += 1 + return states.filter(state => state.atMs <= clock).at(-1)?.row ?? null + }, + }, + } +} + +describe('waiting for a Resolution to move', () => { + it('answers within one check after the build changes state', async () => { + const requested = await requestedRow() + const resolving = { ...requested, state: 'resolving' as const, state_version: 1 } + const run = scripted([{ atMs: 0, row: requested }, { atMs: 1_100, row: resolving }]) + + const answer = await waitForResolutionChange(run.dependencies, requested) + + expect(answer.state).toBe('resolving') + expect(run.elapsed()).toBe(1_250) + expect(run.elapsed() - 1_100).toBeLessThanOrEqual(RESOLUTION_CHECK_MS) + }) + + it('answers a settled Resolution at once', async () => { + const ready = { ...(await requestedRow()), state: 'ready' as const } + const run = scripted([{ atMs: 0, row: ready }]) + + const answer = await waitForResolutionChange(run.dependencies, ready) + + expect(answer).toBe(ready) + expect(run.loads()).toBe(0) + }) + + it('answers the unchanged Resolution when the wait runs out', async () => { + const requested = await requestedRow() + const run = scripted([{ atMs: 0, row: requested }]) + + const answer = await waitForResolutionChange(run.dependencies, requested) + + expect(answer.state).toBe('requested') + expect(run.elapsed()).toBe(RESOLUTION_WAIT_MS) + }) + + it('keeps waiting through a write that leaves the state as it was', async () => { + const signing = { ...(await requestedRow()), state: 'signing' as const, state_version: 5 } + const staged = { ...signing, state_version: 6 } + const publishing = { ...signing, state: 'publishing' as const, state_version: 7 } + const run = scripted([ + { atMs: 0, row: signing }, + { atMs: 300, row: staged }, + { atMs: 900, row: publishing }, + ]) + + const answer = await waitForResolutionChange(run.dependencies, signing) + + expect(answer.state).toBe('publishing') + expect(run.elapsed()).toBe(1_000) + }) +}) + +describe('queueing a build after the response', () => { + it('answers before the queue send finishes', async () => { + const scheduled: Array> = [] + let release = () => {} + const sent = new Promise((resolve) => { + release = resolve + }) + const enqueue = enqueueAfterResponse({ + schedule: work => scheduled.push(work), + enqueue: async () => await sent, + failUnqueued: async () => expect.fail('the send did not fail'), + }) + + await enqueue('6d3c8f1e-8d5f-4a43-9f0e-2a5e1d4b7c11') + + expect(scheduled).toHaveLength(1) + release() + await Promise.all(scheduled) + }) + + it('fails the Resolution as retryable when the send fails', async () => { + const sqlite = createSqliteD1(MIGRATIONS) + const identity = await resolutionRequestIdentity(request, 'resolution-enqueue-test-key') + const created = await createResolution(sqlite.db, request, identity, NOW) + if (created._tag === 'idempotency-conflict') + throw new Error('Test Resolution conflicted') + const scheduled: Array> = [] + const enqueue = enqueueAfterResponse({ + schedule: work => scheduled.push(work), + enqueue: async () => { + throw new Error('Queue send failed: 503') + }, + failUnqueued: async (resolutionId) => { + const row = await getResolution(sqlite.db, resolutionId) + if (row) + await failResolution({ db: sqlite.db, now: () => NOW }, row, 'SERVICE_UNAVAILABLE', true) + }, + }) + + await enqueue(created.row.id) + await Promise.all(scheduled) + + const settled = await getResolution(sqlite.db, created.row.id) + expect(presentResolution(settled!)).toMatchObject({ + state: 'failed', + code: 'SERVICE_UNAVAILABLE', + retryable: true, + }) + sqlite.close() + }) +}) + +describe('pOST /api/v1/resolutions queues the build from the real event', () => { + beforeEach(() => { + vi.resetModules() + // `defineApiHandler` reads Nuxt auto-imports, so the route needs them before it loads. + vi.stubGlobal('defineEventHandler', (handler: unknown) => handler) + vi.stubGlobal('createError', createError) + vi.stubGlobal('readBody', readBody) + }) + + afterEach(() => { + vi.unstubAllGlobals() + }) + + it('answers 202 with the send scheduled off the h3 event', async () => { + const postResolutions = (await import( + '../../layers/artifact-delivery/server/api/v1/resolutions/index.post', + )).default + const sqlite = createSqliteD1(MIGRATIONS) + const sent: Array<{ version: number, resolutionId: string }> = [] + const app = createApp() + app.use(eventHandler((event) => { + event.context.platform = { + db: sqlite.db, + env: { + ARTIFACT_BUILD_QUEUE: { + send: async (message: { version: number, resolutionId: string }) => { + sent.push(message) + }, + }, + }, + } as never + })) + app.use(postResolutions) + const handle = toWebHandler(app) + + const response = await handle(new Request('http://localhost/api/v1/resolutions', { + method: 'POST', + headers: { 'content-type': 'application/json', 'idempotency-key': 'post-route-queue-test-key' }, + // A path selector asks no registry lookup, so the route's HTTP seam stays out of this test. + body: JSON.stringify({ source: { provider: 'github', owner: 'skilld-dev', repository: 'skills', selector: { type: 'path', path: 'skills/demo' } } }), + })) + + expect(response.status).toBe(202) + expect(response.headers.get('content-type')).not.toBe('application/problem+json') + const body = await response.json() + expect(body).toMatchObject({ state: 'pending', stage: 'requested', pollAfterMs: 250 }) + expect(sent.map(message => message.resolutionId)).toEqual([body.resolutionId]) + sqlite.close() + }) + + it('hands the send to the Worker waitUntil when the runtime mounts one', async () => { + const postResolutions = (await import( + '../../layers/artifact-delivery/server/api/v1/resolutions/index.post', + )).default + const sqlite = createSqliteD1(MIGRATIONS) + const awaited: Array> = [] + const app = createApp() + app.use(eventHandler((event) => { + ;(event.context as { cloudflare?: object }).cloudflare = { + context: { waitUntil: (promise: Promise) => awaited.push(promise) }, + } + event.context.platform = { + db: sqlite.db, + env: { + ARTIFACT_BUILD_QUEUE: { + send: async () => { + throw new Error('the send must only be awaited, not resolved here') + }, + }, + }, + } as never + })) + app.use(postResolutions) + const handle = toWebHandler(app) + + const response = await handle(new Request('http://localhost/api/v1/resolutions', { + method: 'POST', + headers: { 'content-type': 'application/json', 'idempotency-key': 'post-route-waituntil-key1' }, + body: JSON.stringify({ source: { provider: 'github', owner: 'skilld-dev', repository: 'skills', selector: { type: 'path', path: 'skills/demo' } } }), + })) + + expect(response.status).toBe(202) + expect(awaited).toHaveLength(1) + // Let the scheduled send and its failure settlement finish before the database closes. + await Promise.allSettled(awaited) + sqlite.close() + }) +})