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 98da91b5..c0554ddd 100644 --- a/layers/artifact-delivery/server/api/v1/resolutions/index.post.ts +++ b/layers/artifact-delivery/server/api/v1/resolutions/index.post.ts @@ -8,11 +8,11 @@ import { problemCodeSchema, } from '../../../schemas/contracts' import { withArtifactProblems } from '../../../utils/artifact-problem' -import { failResolution } from '../../../utils/build' +import { failResolution, processArtifactBuild } from '../../../utils/build' import { findPrivateRepositoryAccess } from '../../../utils/private-access' import { privateArtifactAccessEnabled } from '../../../utils/private-feature' -import { enqueueArtifactBuild } from '../../../utils/queue' -import { enqueueAfterResponse, fetchAdmittedSkillIdentity, requestResolution } from '../../../utils/request-resolution' +import { createArtifactBuildDependencies, enqueueArtifactBuild } from '../../../utils/queue' +import { buildAfterResponse, fetchAdmittedSkillIdentity, requestResolution } from '../../../utils/request-resolution' import { setSkillPageUrlHeader } from '../../../utils/skill-page' import { getResolution } from '../../../utils/state' @@ -39,9 +39,16 @@ export default withArtifactProblems(defineApiHandler({ const result = await requestResolution({ db: platform.db, lookupAdmitted: fetchAdmittedSkillIdentity(event.context), - enqueue: enqueueAfterResponse({ + enqueue: buildAfterResponse({ schedule: work => runAfterResponse(event, work), - enqueue: resolutionId => enqueueArtifactBuild(platform.env, resolutionId), + build: resolutionId => processArtifactBuild(createArtifactBuildDependencies(platform.env), resolutionId), + enqueue: (resolutionId, delaySeconds) => enqueueArtifactBuild(platform.env, resolutionId, delaySeconds), + reportBuildError: (resolutionId, error) => console.error(JSON.stringify({ + operation: 'artifact-build', + outcome: 'failed-in-request', + resolutionId, + error: error instanceof Error ? error.message : String(error), + })), failUnqueued: async (resolutionId, error) => { console.error(JSON.stringify({ operation: 'artifact-build', @@ -107,9 +114,9 @@ export default withArtifactProblems(defineApiHandler({ })) /** - * Keep the queue send alive past the response on Workers. + * Keep the scheduled build alive past the response on Workers. * - * H3Event has no `waitUntil` on h3 1.15, so the send goes through the + * H3Event has no `waitUntil` on h3 1.15, so the work 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. diff --git a/layers/artifact-delivery/server/utils/request-resolution.ts b/layers/artifact-delivery/server/utils/request-resolution.ts index 83fdaa93..0aa9a731 100644 --- a/layers/artifact-delivery/server/utils/request-resolution.ts +++ b/layers/artifact-delivery/server/utils/request-resolution.ts @@ -1,6 +1,7 @@ import type { H3EventContext } from 'h3' import type { SourceRequest } from '../schemas/contracts' import type { FetchAdmittedSkillIdentity } from './admitted-identity' +import type { ArtifactBuildOutcome } from './build' import type { CreateResolutionResult } from './state' import { admittedSourceRequest } from './admitted-identity' import { createResolution, resolutionRequestIdentity } from './state' @@ -61,30 +62,63 @@ export async function requestResolution( return result } -export interface AfterResponseEnqueueDependencies { +/** + * When the queue takes over a build that ran in its request. A Worker keeps a + * request alive 30 s after its response, so a build the runtime stopped + * resumes on the queue a few seconds later. + */ +export const IN_REQUEST_BUILD_FALLBACK_SECONDS = 35 + +export interface AfterResponseBuildDependencies { /** 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. */ + /** Run the build here, as the queue consumer would. */ + build: (resolutionId: string) => Promise + enqueue: (resolutionId: string, delaySeconds?: number) => Promise + /** Settle a Resolution that neither this request nor the queue can build. */ failUnqueued: (resolutionId: string, error: unknown) => Promise + reportBuildError: (resolutionId: string, error: unknown) => void } /** - * An `enqueue` that sends the build message after the response. + * An `enqueue` that builds the Resolution in its own request, after the + * response, with the queue as the fallback. * - * 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. + * The queue delivered a build 1.2 s after the request at the median and 3.4 s + * at worst, in 20 production runs on 2026-10-07. Its consumer ran in US + * colos, where each D1 call to the Sydney primary took 158 to 244 ms. A + * request runs where the client reached Cloudflare, and from Sydney the same + * calls took 8 to 36 ms. Queue placement cannot move a consumer. + * + * A delayed message goes out first, so a build the runtime stops resumes on + * the queue. A build that throws goes to the queue at once, for its retry + * ladder. A build that follows an earlier build of the same Skill goes back + * on the queue after the wait it names, as the consumer does. */ -export function enqueueAfterResponse( - dependencies: AfterResponseEnqueueDependencies, +export function buildAfterResponse( + dependencies: AfterResponseBuildDependencies, ): (resolutionId: string) => Promise { return async (resolutionId) => { - dependencies.schedule( - dependencies.enqueue(resolutionId) - .catch(error => dependencies.failUnqueued(resolutionId, error)), - ) + dependencies.schedule(buildHere(dependencies, resolutionId)) } } + +async function buildHere(dependencies: AfterResponseBuildDependencies, resolutionId: string): Promise { + const fallback = dependencies.enqueue(resolutionId, IN_REQUEST_BUILD_FALLBACK_SECONDS) + .then(() => ({ _tag: 'queued' as const }), (error: unknown) => ({ _tag: 'unqueued' as const, error })) + const outcome = await dependencies.build(resolutionId) + .then(value => ({ _tag: 'built' as const, value }), (error: unknown) => ({ _tag: 'threw' as const, error })) + const queued = await fallback + if (outcome._tag === 'built' && outcome.value._tag !== 'deferred') + return + if (outcome._tag === 'threw') + dependencies.reportBuildError(resolutionId, outcome.error) + const delaySeconds = outcome._tag === 'built' && outcome.value._tag === 'deferred' + ? outcome.value.delaySeconds + : undefined + const sent = await dependencies.enqueue(resolutionId, delaySeconds) + .then(() => null, (error: unknown) => error) + // With the delayed message queued, the consumer still takes the build over. + if (sent !== null && queued._tag === 'unqueued') + await dependencies.failUnqueued(resolutionId, sent) +} diff --git a/test/unit/artifact-resolution-latency.test.ts b/test/unit/artifact-resolution-latency.test.ts index d336ef76..41b26aca 100644 --- a/test/unit/artifact-resolution-latency.test.ts +++ b/test/unit/artifact-resolution-latency.test.ts @@ -1,9 +1,11 @@ import type { SourceRequest } from '../../layers/artifact-delivery/server/schemas/contracts' +import type { ArtifactBuildOutcome } from '../../layers/artifact-delivery/server/utils/build' 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 { bytesToBase64Url } from '../../layers/artifact-delivery/server/utils/encoding' +import { buildAfterResponse, IN_REQUEST_BUILD_FALLBACK_SECONDS } from '../../layers/artifact-delivery/server/utils/request-resolution' import { RESOLUTION_CHECK_MS, RESOLUTION_WAIT_MS, @@ -112,35 +114,89 @@ describe('waiting for a Resolution to move', () => { }) }) -describe('queueing a build after the response', () => { - it('answers before the queue send finishes', async () => { +describe('building in the request after the response', () => { + const ID = '6d3c8f1e-8d5f-4a43-9f0e-2a5e1d4b7c11' + + function harness(options: { + build?: (resolutionId: string) => Promise + enqueue?: (resolutionId: string, delaySeconds?: number) => Promise + } = {}) { const scheduled: Array> = [] + const enqueue = vi.fn(options.enqueue ?? (async () => {})) + const failUnqueued = vi.fn(async () => {}) + const reportBuildError = vi.fn() + const start = buildAfterResponse({ + schedule: work => scheduled.push(work), + build: options.build ?? (async resolutionId => ({ _tag: 'ready', resolutionId })), + enqueue, + failUnqueued, + reportBuildError, + }) + return { start, enqueue, failUnqueued, reportBuildError, settle: () => Promise.all(scheduled), scheduled } + } + + it('answers before the build runs', async () => { let release = () => {} - const sent = new Promise((resolve) => { + const built = 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'), - }) + const run = harness({ build: async (resolutionId) => { + await built + return { _tag: 'ready', resolutionId } + } }) - await enqueue('6d3c8f1e-8d5f-4a43-9f0e-2a5e1d4b7c11') + await run.start(ID) - expect(scheduled).toHaveLength(1) + expect(run.scheduled).toHaveLength(1) release() - await Promise.all(scheduled) + await run.settle() + }) + + it('builds in the request and leaves only the delayed fallback on the queue', async () => { + const build = vi.fn(async (resolutionId: string) => ({ _tag: 'ready' as const, resolutionId })) + const run = harness({ build }) + + await run.start(ID) + await run.settle() + + expect(build).toHaveBeenCalledWith(ID) + expect(run.enqueue.mock.calls).toEqual([[ID, IN_REQUEST_BUILD_FALLBACK_SECONDS]]) + }) + + it('sends a build that throws to the queue at once', async () => { + const run = harness({ build: async () => { + throw new Error('GitHub read failed at tree: The operation was aborted due to timeout') + } }) + + await run.start(ID) + await run.settle() + + expect(run.enqueue.mock.calls).toEqual([[ID, IN_REQUEST_BUILD_FALLBACK_SECONDS], [ID, undefined]]) + expect(run.reportBuildError).toHaveBeenCalledOnce() + expect(run.failUnqueued).not.toHaveBeenCalled() + }) + + it('puts a build that follows an earlier one back on the queue after the wait it names', async () => { + const run = harness({ build: async resolutionId => ({ _tag: 'deferred', resolutionId, delaySeconds: 2 }) }) + + await run.start(ID) + await run.settle() + + expect(run.enqueue.mock.calls).toEqual([[ID, IN_REQUEST_BUILD_FALLBACK_SECONDS], [ID, 2]]) }) - it('fails the Resolution as retryable when the send fails', async () => { + it('fails the Resolution as retryable when neither the request nor the queue can build it', async () => { const sqlite = createSqliteD1(MIGRATIONS) - const identity = await resolutionRequestIdentity(request, 'resolution-enqueue-test-key') + const identity = await resolutionRequestIdentity(request, 'resolution-build-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({ + const start = buildAfterResponse({ schedule: work => scheduled.push(work), + build: async () => { + throw new Error('D1_ERROR: Network connection lost.') + }, enqueue: async () => { throw new Error('Queue send failed: 503') }, @@ -149,9 +205,10 @@ describe('queueing a build after the response', () => { if (row) await failResolution({ db: sqlite.db, now: () => NOW }, row, 'SERVICE_UNAVAILABLE', true) }, + reportBuildError: () => {}, }) - await enqueue(created.row.id) + await start(created.row.id) await Promise.all(scheduled) const settled = await getResolution(sqlite.db, created.row.id) @@ -162,37 +219,60 @@ describe('queueing a build after the response', () => { }) sqlite.close() }) + + it('keeps a build that threw on the delayed fallback when the second send fails', async () => { + let sends = 0 + const run = harness({ + build: async () => { + throw new Error('The operation was aborted due to timeout') + }, + enqueue: async () => { + sends += 1 + if (sends > 1) + throw new Error('Queue send failed: 503') + }, + }) + + await run.start(ID) + await run.settle() + + expect(run.failUnqueued).not.toHaveBeenCalled() + }) }) -describe('pOST /api/v1/resolutions queues the build from the real event', () => { +describe('pOST /api/v1/resolutions builds 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) + // The route reports a failed build through `console.error`; the tests assert the queue flow. + vi.spyOn(console, 'error').mockImplementation(() => {}) }) afterEach(() => { + vi.restoreAllMocks() vi.unstubAllGlobals() }) - it('answers 202 with the send scheduled off the h3 event', async () => { + it('answers 202 and schedules the build 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 sent: Array<{ version: number, resolutionId: string, delaySeconds?: number }> = [] 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) + send: async (message: { version: number, resolutionId: string }, options?: { delaySeconds?: number }) => { + sent.push({ ...message, delaySeconds: options?.delaySeconds }) }, }, + ARTIFACT_TRUSTED_ROOT_JSON: trustedRootConfig(), }, } as never })) @@ -201,7 +281,7 @@ describe('pOST /api/v1/resolutions queues the build from the real event', () => 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' }, + headers: { 'content-type': 'application/json', 'idempotency-key': 'post-route-build-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' } } }), })) @@ -210,11 +290,19 @@ describe('pOST /api/v1/resolutions queues the build from the real event', () => 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]) + + // The build runs after the response. This env has no database, so the + // build throws: the delayed fallback goes out first, and the failure + // re-queues the build at once for its retry ladder. + await vi.waitFor(() => expect(sent).toHaveLength(2)) + expect(sent[0]?.resolutionId).toBe(body.resolutionId) + expect(sent[0]?.delaySeconds).toBe(IN_REQUEST_BUILD_FALLBACK_SECONDS) + expect(sent[1]?.resolutionId).toBe(body.resolutionId) + expect(sent[1]?.delaySeconds).toBeUndefined() sqlite.close() }) - it('hands the send to the Worker waitUntil when the runtime mounts one', async () => { + it('hands the build to the Worker waitUntil and settles it when every send fails', async () => { const postResolutions = (await import( '../../layers/artifact-delivery/server/api/v1/resolutions/index.post', )).default @@ -233,6 +321,7 @@ describe('pOST /api/v1/resolutions queues the build from the real event', () => throw new Error('the send must only be awaited, not resolved here') }, }, + ARTIFACT_TRUSTED_ROOT_JSON: trustedRootConfig(), }, } as never })) @@ -247,8 +336,48 @@ describe('pOST /api/v1/resolutions queues the build from the real event', () => expect(response.status).toBe(202) expect(awaited).toHaveLength(1) - // Let the scheduled send and its failure settlement finish before the database closes. + // Let the scheduled build and its failure settlement finish before the database closes. await Promise.allSettled(awaited) + const body = await response.json() + const settled = await getResolution(sqlite.db, body.resolutionId) + expect(presentResolution(settled!)).toMatchObject({ + state: 'failed', + code: 'SERVICE_UNAVAILABLE', + retryable: true, + }) sqlite.close() }) }) + +/** + * A trusted root config that parses. The scheduled build only reaches its + * first D1 read, where the missing database throws, so nothing verifies + * signatures. + */ +function trustedRootConfig(): string { + const statement = { + version: 1, + rootKeyId: 'skilld-root-2026', + keyId: 'skilld-production-2026-08', + algorithm: 'Ed25519', + publicKey: 'B'.repeat(43), + notBefore: '2026-08-20T00:00:00.000Z', + notAfter: '2026-11-20T00:00:00.000Z', + status: 'active', + } as const + return JSON.stringify({ + version: 1, + rootKeyId: statement.rootKeyId, + rootPublicKey: 'A'.repeat(43), + keys: [{ + keyId: statement.keyId, + algorithm: statement.algorithm, + publicKey: statement.publicKey, + notBefore: statement.notBefore, + notAfter: statement.notAfter, + status: statement.status, + statement: bytesToBase64Url(new TextEncoder().encode(JSON.stringify(statement))), + rootSignature: 'C'.repeat(86), + }], + }) +}