From 37d6a41b2eaf131de5a71b6dc074c324e9503953 Mon Sep 17 00:00:00 2001 From: Tanner Linsley Date: Mon, 7 Sep 2026 08:00:26 -0600 Subject: [PATCH] Fix intent discovery workflow timeouts --- src/routes/admin/intent.tsx | 32 ++++-- src/utils/intent-admin.server.ts | 17 ++- src/utils/intent-sync.server.ts | 149 ++++++++++----------------- src/utils/intent-workflows.server.ts | 100 ++++++++++++++++-- src/utils/intent.server.ts | 19 ++-- tests/intent-workflow.test.ts | 133 ++++++++++++++++++++++-- 6 files changed, 323 insertions(+), 127 deletions(-) diff --git a/src/routes/admin/intent.tsx b/src/routes/admin/intent.tsx index bc310e087..8d0f85fc4 100644 --- a/src/routes/admin/intent.tsx +++ b/src/routes/admin/intent.tsx @@ -234,16 +234,28 @@ function IntentAdminPage() { {/* Mutation result banners */} {discoverMutation.data && ( 0 - ? [`${discoverMutation.data.errors.length} errors`] - : []), - ]} - errors={discoverMutation.data.errors} + title={ + discoverMutation.data.kind === 'completed' + ? 'Discovery complete' + : 'Discovery started' + } + items={ + discoverMutation.data.kind === 'completed' + ? [ + `${discoverMutation.data.summary.packagesDiscovered} found on NPM`, + `${discoverMutation.data.summary.packagesVerified} verified`, + `${discoverMutation.data.summary.versionsEnqueued} versions enqueued`, + ...(discoverMutation.data.summary.errors.length > 0 + ? [`${discoverMutation.data.summary.errors.length} errors`] + : []), + ] + : ['Discovery will continue during scheduled processing.'] + } + errors={ + discoverMutation.data.kind === 'completed' + ? discoverMutation.data.summary.errors + : [] + } onDismiss={() => discoverMutation.reset()} /> )} diff --git a/src/utils/intent-admin.server.ts b/src/utils/intent-admin.server.ts index 4c733d3cf..9c0d82f31 100644 --- a/src/utils/intent-admin.server.ts +++ b/src/utils/intent-admin.server.ts @@ -202,10 +202,25 @@ export async function triggerIntentDiscover() { workflowId: INTENT_DISCOVER_WORKFLOW_ID, runId: createAdminRunId(INTENT_DISCOVER_WORKFLOW_ID), input: { source: 'admin' }, + maxDurationMs: WORKFLOW_RUNTIME_MAX_DURATION_MS, + minYieldRemainingMs: WORKFLOW_RUNTIME_MIN_REMAINING_MS, includeEvents: false, }) - return intentDiscoveryResultSchema.parse(getCompletedWorkflowOutput(result)) + if (result.kind === 'paused') { + return { + kind: 'continuing' as const, + runId: result.runId, + } + } + + return { + kind: 'completed' as const, + runId: result.runId, + summary: intentDiscoveryResultSchema.parse( + getCompletedWorkflowOutput(result), + ), + } } // --------------------------------------------------------------------------- diff --git a/src/utils/intent-sync.server.ts b/src/utils/intent-sync.server.ts index e4319dea0..f15b5e9fa 100644 --- a/src/utils/intent-sync.server.ts +++ b/src/utils/intent-sync.server.ts @@ -4,9 +4,10 @@ import { extractSkillsFromTarball, fetchPackument, isIntentCompatible, - searchIntentPackages, + searchIntentPackagesPage, selectVersionsToSync, } from '~/utils/intent.server' +import { fetchWithTimeout } from '~/utils/outbound-fetch.server' import { enqueuePackageVersion, getKnownVersions, @@ -32,6 +33,11 @@ const githubContentResponseSchema = z.object({ content: z.string().optional(), }) +const intentGitHubCandidateSchema = z.object({ + repo: z.string(), + path: z.string(), +}) + const packageJsonSchema = z.object({ name: z.string().optional(), private: z.boolean().optional(), @@ -66,6 +72,7 @@ export const intentProcessResultSchema = z.object({ }) export type IntentDiscoveryResult = z.infer +export type IntentGitHubCandidate = z.infer export type IntentVersionProcessResult = z.infer< typeof intentVersionProcessResultSchema > @@ -78,7 +85,15 @@ export interface IntentVersionToProcess { } export interface IntentSyncOperations { - discoverIntentPackages: () => Promise + searchIntentNpmPackagesPage: (options: { + from: number + size: number + }) => Promise<{ packageNames: Array; total: number }> + discoverIntentNpmPackage: (packageName: string) => Promise + searchIntentGitHubCandidates: () => Promise> + discoverIntentGitHubPackage: ( + candidate: IntentGitHubCandidate, + ) => Promise selectPendingIntentVersions: (options: { limit: number excludeIds?: Array @@ -89,61 +104,22 @@ export interface IntentSyncOperations { } export const defaultIntentSyncOperations: IntentSyncOperations = { - discoverIntentPackages, + searchIntentNpmPackagesPage, + discoverIntentNpmPackage, + searchIntentGitHubCandidates, + discoverIntentGitHubPackage, selectPendingIntentVersions, processIntentVersion, } -export async function discoverIntentPackages(): Promise { - const errors: Array = [] - let packagesDiscovered = 0 - let githubCandidates = 0 - let packagesVerified = 0 - let versionsEnqueued = 0 - - try { - const searchResults = await searchIntentPackages() - const packageNames = dedupe( - searchResults.objects.map((item) => item.package.name), - ) - packagesDiscovered = packageNames.length - - for (const packageName of packageNames) { - try { - const enqueued = await discoverNpmPackage(packageName) - if (enqueued !== null) { - packagesVerified++ - versionsEnqueued += enqueued - } - } catch (error) { - errors.push(`npm/${packageName}: ${getErrorMessage(error)}`) - } - } - } catch (error) { - errors.push(`npm-search: ${getErrorMessage(error)}`) - } - - const githubToken = - getCurrentHostRuntimeEnv()?.GITHUB_AUTH_TOKEN ?? - process.env.GITHUB_AUTH_TOKEN - if (githubToken) { - try { - const githubResult = await discoverGitHubPackages(githubToken) - githubCandidates = githubResult.githubCandidates - packagesVerified += githubResult.packagesVerified - versionsEnqueued += githubResult.versionsEnqueued - errors.push(...githubResult.errors) - } catch (error) { - errors.push(`github-search: ${getErrorMessage(error)}`) - } - } - +export async function searchIntentNpmPackagesPage(options: { + from: number + size: number +}) { + const page = await searchIntentPackagesPage(options) return { - packagesDiscovered, - githubCandidates, - packagesVerified, - versionsEnqueued, - errors, + packageNames: page.objects.map((item) => item.package.name), + total: page.total, } } @@ -174,7 +150,9 @@ export function summarizeIntentProcessResults( } } -async function discoverNpmPackage(packageName: string): Promise { +export async function discoverIntentNpmPackage( + packageName: string, +): Promise { await upsertIntentPackage({ name: packageName, verified: false }) const packument = await fetchPackument(packageName) @@ -188,19 +166,17 @@ async function discoverNpmPackage(packageName: string): Promise { return enqueueVersionsFromPackument(packageName, packument) } -async function discoverGitHubPackages(githubToken: string): Promise<{ - githubCandidates: number - packagesVerified: number - versionsEnqueued: number - errors: Array -}> { +export async function searchIntentGitHubCandidates() { + const githubToken = getGitHubToken() + if (!githubToken) return [] + const ghHeaders = { Authorization: `Bearer ${githubToken}`, Accept: 'application/vnd.github.v3+json', } - const searchRes = await fetch( + const searchRes = await fetchWithTimeout( 'https://api.github.com/search/code?q=%22%40tanstack%2Fintent%22+filename%3Apackage.json&per_page=100', - { headers: ghHeaders }, + { headers: ghHeaders, timeoutMs: 10_000 }, ) if (!searchRes.ok) throw new Error(`GitHub search ${searchRes.status}`) @@ -212,39 +188,22 @@ async function discoverGitHubPackages(githubToken: string): Promise<{ })), (item) => `${item.repo}|${item.path}`, ) - let packagesVerified = 0 - let versionsEnqueued = 0 - const errors: Array = [] - - for (const candidate of candidates) { - try { - const enqueued = await discoverGitHubPackage(candidate, ghHeaders) - if (enqueued !== null) { - packagesVerified++ - versionsEnqueued += enqueued - } - } catch (error) { - errors.push( - `github/${candidate.repo}/${candidate.path}: ${getErrorMessage(error)}`, - ) - } - } - - return { - githubCandidates: candidates.length, - packagesVerified, - versionsEnqueued, - errors, - } + return candidates } -async function discoverGitHubPackage( - candidate: { repo: string; path: string }, - headers: HeadersInit, +export async function discoverIntentGitHubPackage( + candidate: IntentGitHubCandidate, ): Promise { - const contentRes = await fetch( + const githubToken = getGitHubToken() + if (!githubToken) return null + + const headers = { + Authorization: `Bearer ${githubToken}`, + Accept: 'application/vnd.github.v3+json', + } + const contentRes = await fetchWithTimeout( `https://api.github.com/repos/${candidate.repo}/contents/${candidate.path}`, - { headers }, + { headers, timeoutMs: 10_000 }, ) if (!contentRes.ok) return null @@ -256,8 +215,9 @@ async function discoverGitHubPackage( ) if (!packageJson.name || packageJson.private) return null - const npmRes = await fetch( + const npmRes = await fetchWithTimeout( `https://registry.npmjs.org/${encodeURIComponent(packageJson.name)}/latest`, + { timeoutMs: 10_000 }, ) if (!npmRes.ok) return null @@ -343,8 +303,11 @@ export async function processIntentVersion( } } -function dedupe(values: Array): Array { - return dedupeBy(values, (value) => value) +function getGitHubToken() { + return ( + getCurrentHostRuntimeEnv()?.GITHUB_AUTH_TOKEN ?? + process.env.GITHUB_AUTH_TOKEN + ) } function dedupeBy(values: Array, getKey: (value: T) => string): Array { diff --git a/src/utils/intent-workflows.server.ts b/src/utils/intent-workflows.server.ts index 3cb2ee436..49a477a15 100644 --- a/src/utils/intent-workflows.server.ts +++ b/src/utils/intent-workflows.server.ts @@ -24,9 +24,12 @@ const intentProcessInputSchema = z.object({ const discoverStepOptions = { retry: { maxAttempts: 2, backoff: 'exponential', baseMs: 1_000 }, - timeout: 10 * 60 * 1000, + timeout: 20_000, } satisfies StepOptions +const NPM_SEARCH_PAGE_SIZE = 250 +const MAX_NPM_SEARCH_RESULTS = 1_000 + const selectPendingVersionsStepOptions = { timeout: 30_000, } satisfies StepOptions @@ -40,19 +43,98 @@ export const INTENT_PROCESS_WORKFLOW_ID = 'intent-process-workflow' export const INTENT_DISCOVER_SCHEDULE_ID = 'intent-discover-every-6h' export const INTENT_PROCESS_SCHEDULE_ID = 'intent-process-every-15m' -function createIntentDiscoverWorkflow( +export function createIntentDiscoverWorkflow( operations: IntentSyncOperations = defaultIntentSyncOperations, ) { return createWorkflow({ id: INTENT_DISCOVER_WORKFLOW_ID, input: intentDiscoverInputSchema, - }).handler((ctx) => - ctx.step( - 'discover-intent-packages', - () => operations.discoverIntentPackages(), - discoverStepOptions, - ), - ) + }).handler(async (ctx) => { + const errors: Array = [] + const discoveredPackageNames: Array = [] + let packagesVerified = 0 + let versionsEnqueued = 0 + let searchOffset = 0 + + while (discoveredPackageNames.length < MAX_NPM_SEARCH_RESULTS) { + try { + const page = await ctx.step( + `search-npm-packages:${searchOffset}`, + () => + operations.searchIntentNpmPackagesPage({ + from: searchOffset, + size: NPM_SEARCH_PAGE_SIZE, + }), + discoverStepOptions, + ) + discoveredPackageNames.push(...page.packageNames) + searchOffset += NPM_SEARCH_PAGE_SIZE + + if (searchOffset >= page.total || page.packageNames.length === 0) break + } catch (error) { + errors.push(`npm-search: ${getErrorMessage(error)}`) + break + } + } + + const packageNames = [...new Set(discoveredPackageNames)].slice( + 0, + MAX_NPM_SEARCH_RESULTS, + ) + for (const packageName of packageNames) { + try { + const enqueued = await ctx.step( + `discover-npm-package:${packageName}`, + () => operations.discoverIntentNpmPackage(packageName), + discoverStepOptions, + ) + if (enqueued !== null) { + packagesVerified++ + versionsEnqueued += enqueued + } + } catch (error) { + errors.push(`npm/${packageName}: ${getErrorMessage(error)}`) + } + } + + let githubCandidates = 0 + try { + const candidates = await ctx.step( + 'search-github-packages', + () => operations.searchIntentGitHubCandidates(), + discoverStepOptions, + ) + githubCandidates = candidates.length + + for (const candidate of candidates) { + try { + const enqueued = await ctx.step( + `discover-github-package:${candidate.repo}:${candidate.path}`, + () => operations.discoverIntentGitHubPackage(candidate), + discoverStepOptions, + ) + if (enqueued !== null) { + packagesVerified++ + versionsEnqueued += enqueued + } + } catch (error) { + errors.push( + `github/${candidate.repo}/${candidate.path}: ${getErrorMessage(error)}`, + ) + } + } + } catch (error) { + errors.push(`github-search: ${getErrorMessage(error)}`) + } + + return { + packagesDiscovered: packageNames.length, + githubCandidates, + packagesVerified, + versionsEnqueued, + errors, + } + }) } export function createIntentProcessWorkflow( diff --git a/src/utils/intent.server.ts b/src/utils/intent.server.ts index 37e4cccfb..bbe2e8ebf 100644 --- a/src/utils/intent.server.ts +++ b/src/utils/intent.server.ts @@ -197,20 +197,27 @@ function isSafeSkillPath(skillPath: string) { ) } +export async function searchIntentPackagesPage(options: { + from: number + size: number +}): Promise { + return fetchNpmSearch( + `${NPM_REGISTRY}/-/v1/search?text=keywords:tanstack-intent&size=${options.size}&from=${options.from}`, + ) +} + export async function searchIntentPackages(): Promise { const results: NpmSearchResult = { objects: [], total: 0, time: '' } let from = 0 const size = 250 while (true) { - const url = `${NPM_REGISTRY}/-/v1/search?text=keywords:tanstack-intent&size=${size}&from=${from}` - const page = await fetchNpmSearch(url) - + const page = await searchIntentPackagesPage({ from, size }) results.objects.push(...page.objects) results.total = page.total results.time = page.time - from += size + if ( from >= page.total || page.objects.length === 0 || @@ -218,11 +225,9 @@ export async function searchIntentPackages(): Promise { ) { results.objects = results.objects.slice(0, MAX_INTENT_SEARCH_RESULTS) results.total = Math.min(results.total, MAX_INTENT_SEARCH_RESULTS) - break + return results } } - - return results } export async function searchIntentPackagesByText( diff --git a/tests/intent-workflow.test.ts b/tests/intent-workflow.test.ts index 22089ad6f..bedc7a59c 100644 --- a/tests/intent-workflow.test.ts +++ b/tests/intent-workflow.test.ts @@ -7,7 +7,10 @@ import { materializeWorkflowSchedules, } from '@tanstack/workflow-runtime' import type { WorkflowExecutionStore } from '@tanstack/workflow-runtime' -import { createIntentProcessWorkflow } from '../src/utils/intent-workflows.server' +import { + createIntentDiscoverWorkflow, + createIntentProcessWorkflow, +} from '../src/utils/intent-workflows.server' import type { IntentProcessResult, IntentSyncOperations, @@ -194,20 +197,136 @@ test('process workflow yields near its runtime deadline and resumes the same run } }) -const noopOperations: IntentSyncOperations = { - discoverIntentPackages: async () => ({ - packagesDiscovered: 0, +test('discovery yields near its runtime deadline and resumes completed package steps', async () => { + let wallNow = 1_000 + mock.method(Date, 'now', () => wallNow) + + try { + const discovered: Array = [] + const store = inMemoryWorkflowExecutionStore() + const { discoverWorkflow, runtime } = createTestDiscoverRuntime({ + store, + operations: { + ...noopOperations, + searchIntentNpmPackagesPage: async () => ({ + packageNames: ['@example/one', '@example/two'], + total: 2, + }), + discoverIntentNpmPackage: async (packageName) => { + discovered.push(packageName) + wallNow += 600 + return 1 + }, + }, + }) + const runId = 'intent-discover:deadline-resume' + + const first = await runtime.startRun({ + workflowId: discoverWorkflow.id, + runId, + input: { source: 'schedule' }, + now: 100, + deadline: 2_000, + minYieldRemainingMs: 500, + includeEvents: false, + }) + const paused = await store.loadRun(runId) + const resumed = await runtime.sweep({ + now: 101, + deadline: 10_000, + minYieldRemainingMs: 500, + includeEvents: false, + }) + + assert.equal(first.kind, 'paused') + assert.equal(paused?.status, 'paused') + assert.equal(resumed.timers[0]?.kind, 'completed') + assert.deepEqual(discovered, ['@example/one', '@example/two']) + assert.deepEqual(resumed.timers[0]?.run?.output, { + packagesDiscovered: 2, + githubCandidates: 0, + packagesVerified: 2, + versionsEnqueued: 2, + errors: [], + }) + } finally { + mock.restoreAll() + } +}) + +test('failed discovery package does not prevent later packages from running', async () => { + let failedCalls = 0 + let goodCalls = 0 + const { discoverWorkflow, runtime } = createTestDiscoverRuntime({ + operations: { + ...noopOperations, + searchIntentNpmPackagesPage: async () => ({ + packageNames: ['@example/bad', '@example/good'], + total: 2, + }), + discoverIntentNpmPackage: async (packageName) => { + if (packageName === '@example/bad') { + failedCalls++ + throw new Error('bad package') + } + goodCalls++ + return 2 + }, + }, + }) + + const result = await runtime.startRun({ + workflowId: discoverWorkflow.id, + runId: 'intent-discover:partial-failure', + input: { source: 'admin' }, + includeEvents: false, + }) + + assert.equal(result.kind, 'completed') + assert.equal(failedCalls, 2) + assert.equal(goodCalls, 1) + assert.deepEqual(result.run?.output, { + packagesDiscovered: 2, githubCandidates: 0, - packagesVerified: 0, - versionsEnqueued: 0, - errors: [], + packagesVerified: 1, + versionsEnqueued: 2, + errors: ['npm/@example/bad: bad package'], + }) +}) + +const noopOperations: IntentSyncOperations = { + searchIntentNpmPackagesPage: async () => ({ + packageNames: [], + total: 0, }), + discoverIntentNpmPackage: async () => null, + searchIntentGitHubCandidates: async () => [], + discoverIntentGitHubPackage: async () => null, selectPendingIntentVersions: async () => [], processIntentVersion: async () => { throw new Error('No test version configured') }, } +function createTestDiscoverRuntime(options: { + operations: IntentSyncOperations + store?: WorkflowExecutionStore +}) { + const discoverWorkflow = createIntentDiscoverWorkflow(options.operations) + + return { + discoverWorkflow, + runtime: defineWorkflowRuntime({ + store: options.store ?? inMemoryWorkflowExecutionStore(), + workflows: { + [discoverWorkflow.id]: { + load: async () => discoverWorkflow, + }, + }, + }), + } +} + function createTestIntentRuntime(options: { operations: IntentSyncOperations store?: WorkflowExecutionStore