diff --git a/package.json b/package.json index 56461f78d..645c7c2bd 100644 --- a/package.json +++ b/package.json @@ -74,9 +74,9 @@ "@tanstack/react-start": "1.168.26", "@tanstack/react-start-client": "1.168.14", "@tanstack/react-table": "^8.21.3", - "@tanstack/workflow-core": "0.0.3", - "@tanstack/workflow-runtime": "0.0.2", - "@tanstack/workflow-store-drizzle-postgres": "0.0.4", + "@tanstack/workflow-core": "0.0.4", + "@tanstack/workflow-runtime": "0.0.3", + "@tanstack/workflow-store-drizzle-postgres": "0.0.5", "@types/d3": "^7.4.3", "@uploadthing/react": "^7.3.3", "@visx/hierarchy": "^3.12.0", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index d935d005a..b70fc56b3 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -139,14 +139,14 @@ importers: specifier: ^8.21.3 version: 8.21.3(react-dom@19.2.3(react@19.2.3))(react@19.2.3) '@tanstack/workflow-core': - specifier: 0.0.3 - version: 0.0.3 + specifier: 0.0.4 + version: 0.0.4(@opentelemetry/api@1.9.1) '@tanstack/workflow-runtime': - specifier: 0.0.2 - version: 0.0.2 + specifier: 0.0.3 + version: 0.0.3(@opentelemetry/api@1.9.1) '@tanstack/workflow-store-drizzle-postgres': - specifier: 0.0.4 - version: 0.0.4(@opentelemetry/api@1.9.1)(@types/pg@8.20.0)(postgres@3.4.8) + specifier: 0.0.5 + version: 0.0.5(@opentelemetry/api@1.9.1)(@types/pg@8.20.0)(postgres@3.4.8) '@types/d3': specifier: ^7.4.3 version: 7.4.3 @@ -3600,16 +3600,20 @@ packages: resolution: {integrity: sha512-uhOeFyxLcU41HzvrxsGpiWdcMbScY1EDgbZ5K7DVRMYInbLYWAC0EA/kx9wXAoSM8q82bUG2hRl8+EAjE6XAbA==} engines: {node: '>=20.19'} - '@tanstack/workflow-core@0.0.3': - resolution: {integrity: sha512-0ev5ISQR8knUBAco46bc8h2/bMGS3ZPlEC/I/EgeD7PXE0DnmNJLvrkwZ5KfdzcbdM8pqJQLR5coBHtNurOcEg==} + '@tanstack/workflow-core@0.0.4': + resolution: {integrity: sha512-Ytl8uF1v+xKDCez1P7Qcfoe6OZ2HskQzfey8Kci5h83UoEOt8azi/KmfrkrE28ev46QPSqkOn4tTLvILgCi67w==} engines: {node: '>=18'} + peerDependencies: + '@opentelemetry/api': ^1.9.0 - '@tanstack/workflow-runtime@0.0.2': - resolution: {integrity: sha512-v+cv8rIOaE0a50Md6WSPge11wLn9NVxcGbVanKQe2BlZuNtDVIayZFUgn3RXHQp68MQmrdu2bK6aT9BSMZ4z2g==} + '@tanstack/workflow-runtime@0.0.3': + resolution: {integrity: sha512-U2IRw+sednNI7kqG301o6RBrgAB7zob9uDjcsVr5RPQPjIWHk2a9tAJkVYQKCCWUbE6L37GW+Pl8nU2HqIISRQ==} engines: {node: '>=18'} + peerDependencies: + '@opentelemetry/api': ^1.9.0 - '@tanstack/workflow-store-drizzle-postgres@0.0.4': - resolution: {integrity: sha512-IguipqLh04rvO3FnWxuPp/BiTJPnJiB7jlGH/M/qlfhUkFA0x4bPJ+MjYxa1sQaJLJbpYJI/Z2AVzRwfhDAFRw==} + '@tanstack/workflow-store-drizzle-postgres@0.0.5': + resolution: {integrity: sha512-f7t5gI56Cy4hucnuY0hA1W5tKCjH1gKtQ+xX3uRX8/GDK5AeMIaGw4pL/YxiRwHFZYPxYppkTFmbb3i8z17g6w==} engines: {node: '>=18'} '@tweenjs/tween.js@23.1.3': @@ -9972,18 +9976,20 @@ snapshots: '@tanstack/virtual-file-routes@1.162.0': {} - '@tanstack/workflow-core@0.0.3': + '@tanstack/workflow-core@0.0.4(@opentelemetry/api@1.9.1)': dependencies: + '@opentelemetry/api': 1.9.1 '@standard-schema/spec': 1.1.0 - '@tanstack/workflow-runtime@0.0.2': + '@tanstack/workflow-runtime@0.0.3(@opentelemetry/api@1.9.1)': dependencies: - '@tanstack/workflow-core': 0.0.3 + '@opentelemetry/api': 1.9.1 + '@tanstack/workflow-core': 0.0.4(@opentelemetry/api@1.9.1) - '@tanstack/workflow-store-drizzle-postgres@0.0.4(@opentelemetry/api@1.9.1)(@types/pg@8.20.0)(postgres@3.4.8)': + '@tanstack/workflow-store-drizzle-postgres@0.0.5(@opentelemetry/api@1.9.1)(@types/pg@8.20.0)(postgres@3.4.8)': dependencies: - '@tanstack/workflow-core': 0.0.3 - '@tanstack/workflow-runtime': 0.0.2 + '@tanstack/workflow-core': 0.0.4(@opentelemetry/api@1.9.1) + '@tanstack/workflow-runtime': 0.0.3(@opentelemetry/api@1.9.1) drizzle-orm: 0.45.2(@opentelemetry/api@1.9.1)(@types/pg@8.20.0)(postgres@3.4.8) transitivePeerDependencies: - '@aws-sdk/client-rds-data' diff --git a/src/routes/admin/intent.tsx b/src/routes/admin/intent.tsx index 38851524d..5533cc4e7 100644 --- a/src/routes/admin/intent.tsx +++ b/src/routes/admin/intent.tsx @@ -180,7 +180,7 @@ function IntentAdminPage() { processMutation.isPending || (stats?.pendingVersions ?? 0) + (stats?.failedVersions ?? 0) === 0 } - title="Download tarballs and extract skills until the workflow nears its time budget" + title="Download tarballs and extract skills for pending versions" > 0 - ? [ - `${processMutation.data.deferred} version(s) deferred by the time budget`, - ] - : []), - ...processMutation.data.results.map( - (r) => - `${r.packageName}@${r.version}: ${r.status === 'synced' ? `${r.skillCount} skills` : `FAILED — ${r.error}`}`, - ), - ]} - errors={processMutation.data.results - .filter((r) => r.status === 'failed') - .map((r) => `${r.packageName}@${r.version}: ${r.error}`)} + title={ + processMutation.data.kind === 'completed' + ? `Processed ${processMutation.data.summary.processed} version(s)` + : 'Queue processing started' + } + items={ + processMutation.data.kind === 'completed' + ? processMutation.data.summary.results.map( + (result) => + `${result.packageName}@${result.version}: ${result.status === 'synced' ? `${result.skillCount} skills` : `FAILED — ${result.error}`}`, + ) + : ['Remaining work will continue during scheduled processing.'] + } + errors={ + processMutation.data.kind === 'completed' + ? processMutation.data.summary.results + .filter((result) => result.status === 'failed') + .map( + (result) => + `${result.packageName}@${result.version}: ${result.error}`, + ) + : [] + } onDismiss={() => processMutation.reset()} /> )} diff --git a/src/server/scheduled.server.ts b/src/server/scheduled.server.ts index cb2dc5252..90f07aa99 100644 --- a/src/server/scheduled.server.ts +++ b/src/server/scheduled.server.ts @@ -4,13 +4,14 @@ import { refreshHomepageNpmStatsSummary } from '~/utils/homepage-npm-stats.serve import { refreshGitHubOrgStats } from '~/utils/stats.functions' import { reconcileWorkflowRuntimeStore, + WORKFLOW_RUNTIME_MAX_DURATION_MS, + WORKFLOW_RUNTIME_MIN_REMAINING_MS, workflowRuntime, } from '~/utils/workflow-runtime.server' const CONTENT_CACHE_PRUNE_CRON = '0 9 * * *' const STATS_REFRESH_CRON = '0 */6 * * *' const WORKFLOW_SWEEP_CRON = '* * * * *' -const WORKFLOW_SWEEP_MAX_DURATION_MS = 25_000 export async function runScheduledTasks(cron: string, scheduledTime: number) { switch (cron) { @@ -43,7 +44,8 @@ async function runWorkflowSweep(cron: string, scheduledTime: number) { const sweep = await workflowRuntime.sweep({ now: scheduledTime, leaseOwner: `cloudflare:${cron}:${scheduledTime}`, - maxDurationMs: WORKFLOW_SWEEP_MAX_DURATION_MS, + maxDurationMs: WORKFLOW_RUNTIME_MAX_DURATION_MS, + minYieldRemainingMs: WORKFLOW_RUNTIME_MIN_REMAINING_MS, maxScheduledRuns: 10, maxTimers: 10, includeEvents: false, diff --git a/src/utils/intent-admin.server.ts b/src/utils/intent-admin.server.ts index 4d4782cde..20a92268d 100644 --- a/src/utils/intent-admin.server.ts +++ b/src/utils/intent-admin.server.ts @@ -36,6 +36,8 @@ import { import { getWorkflowRuntimeHealth, reconcileWorkflowRuntimeStore, + WORKFLOW_RUNTIME_MAX_DURATION_MS, + WORKFLOW_RUNTIME_MIN_REMAINING_MS, workflowExecutionStore, workflowRuntime, } from '~/utils/workflow-runtime.server' @@ -219,10 +221,25 @@ export async function triggerIntentProcess() { input: { source: 'admin', }, + maxDurationMs: WORKFLOW_RUNTIME_MAX_DURATION_MS, + minYieldRemainingMs: WORKFLOW_RUNTIME_MIN_REMAINING_MS, includeEvents: false, }) - return intentProcessResultSchema.parse(getCompletedWorkflowOutput(result)) + if (result.kind === 'paused') { + return { + kind: 'continuing' as const, + runId: result.runId, + } + } + + return { + kind: 'completed' as const, + runId: result.runId, + summary: intentProcessResultSchema.parse( + getCompletedWorkflowOutput(result), + ), + } } // --------------------------------------------------------------------------- diff --git a/src/utils/intent-workflows.server.ts b/src/utils/intent-workflows.server.ts index 514adbdc2..3cb2ee436 100644 --- a/src/utils/intent-workflows.server.ts +++ b/src/utils/intent-workflows.server.ts @@ -16,18 +16,9 @@ const intentDiscoverInputSchema = z.object({ source: z.enum(['schedule', 'admin']).default('schedule'), }) -const PROCESS_WORKFLOW_DEFAULT_MAX_DURATION_MS = 4 * 60 * 1000 -const PROCESS_WORKFLOW_MAX_DURATION_MS = 4 * 60 * 1000 -const PROCESS_WORKFLOW_MIN_REMAINING_MS = 30_000 const PROCESS_WORKFLOW_SELECT_LIMIT = 50 const intentProcessInputSchema = z.object({ - maxDurationMs: z - .number() - .int() - .positive() - .max(PROCESS_WORKFLOW_MAX_DURATION_MS) - .default(PROCESS_WORKFLOW_DEFAULT_MAX_DURATION_MS), source: z.enum(['schedule', 'admin']).default('schedule'), }) @@ -71,14 +62,11 @@ export function createIntentProcessWorkflow( id: INTENT_PROCESS_WORKFLOW_ID, input: intentProcessInputSchema, }).handler(async (ctx) => { - const deadline = Date.now() + ctx.input.maxDurationMs const attemptedIds = new Set() const results: Array = [] - let deferred = 0 let selectIteration = 0 - let shouldContinue = true - while (shouldContinue && hasProcessBudget(deadline)) { + while (true) { const versions = await ctx.step( `select-pending-versions:${selectIteration}`, () => @@ -97,14 +85,7 @@ export function createIntentProcessWorkflow( ) if (unattemptedVersions.length === 0) break - for (let index = 0; index < unattemptedVersions.length; index++) { - if (!hasProcessBudget(deadline)) { - deferred += unattemptedVersions.length - index - shouldContinue = false - break - } - - const version = unattemptedVersions[index]! + for (const version of unattemptedVersions) { attemptedIds.add(version.id) try { @@ -128,7 +109,7 @@ export function createIntentProcessWorkflow( if (versions.length < PROCESS_WORKFLOW_SELECT_LIMIT) break } - return summarizeIntentProcessResults(results, { deferred }) + return summarizeIntentProcessResults(results) }) } @@ -155,7 +136,6 @@ export const intentWorkflowRegistrations = { schedule: every.minutes(15), overlapPolicy: 'skip', input: { - maxDurationMs: PROCESS_WORKFLOW_DEFAULT_MAX_DURATION_MS, source: 'schedule', }, }, @@ -166,7 +146,3 @@ export const intentWorkflowRegistrations = { function getErrorMessage(error: unknown): string { return error instanceof Error ? error.message : String(error) } - -function hasProcessBudget(deadline: number): boolean { - return Date.now() + PROCESS_WORKFLOW_MIN_REMAINING_MS < deadline -} diff --git a/src/utils/workflow-runtime.server.ts b/src/utils/workflow-runtime.server.ts index f6380023b..a5482e3ba 100644 --- a/src/utils/workflow-runtime.server.ts +++ b/src/utils/workflow-runtime.server.ts @@ -10,6 +10,9 @@ import { intentWorkflowRegistrations } from '~/utils/intent-workflows.server' export const workflowExecutionStore = createDrizzlePostgresWorkflowStore({ db }) +export const WORKFLOW_RUNTIME_MAX_DURATION_MS = 25_000 +export const WORKFLOW_RUNTIME_MIN_REMAINING_MS = 5_000 + export const workflowRuntime = defineWorkflowRuntime({ store: workflowExecutionStore, workflows: { diff --git a/tests/intent-workflow.test.ts b/tests/intent-workflow.test.ts index 25d70ac47..22089ad6f 100644 --- a/tests/intent-workflow.test.ts +++ b/tests/intent-workflow.test.ts @@ -1,5 +1,5 @@ import assert from 'node:assert/strict' -import { test } from 'node:test' +import { mock, test } from 'node:test' import { defineWorkflowRuntime, every, @@ -114,54 +114,84 @@ test('failed package version step does not prevent other versions from processin assert.deepEqual(result.run.output, expected) }) -test('process workflow continues from queue state across scheduled invocations', async () => { - const queue = [ - { id: 1, packageName: '@example/one', version: '1.0.0' }, - { id: 2, packageName: '@example/two', version: '1.0.0' }, - ] - const processed: Array = [] - const store = inMemoryWorkflowExecutionStore() - const operations: IntentSyncOperations = { - ...noopOperations, - selectPendingIntentVersions: async () => { - const next = queue.shift() - return next ? [next] : [] - }, - processIntentVersion: async (versionId: number) => { - const packageName = versionId === 1 ? '@example/one' : '@example/two' - const version = '1.0.0' - processed.push(`${packageName}@${version}`) - return { - packageName, - version, - status: 'synced', - skillCount: 1, - } - }, - } - const first = createTestIntentRuntime({ - store, - operations, - }) - const second = createTestIntentRuntime({ - store, - operations, - }) - const firstBucket = Date.UTC(2026, 4, 26, 12, 0, 0) - const secondBucket = Date.UTC(2026, 4, 26, 12, 15, 0) +test('process workflow yields near its runtime deadline and resumes the same run', async () => { + let wallNow = 1_000 + mock.method(Date, 'now', () => wallNow) - await materializeWorkflowSchedules(first.runtime, { now: firstBucket }) - await first.runtime.sweep({ now: firstBucket, includeEvents: false }) - await materializeWorkflowSchedules(second.runtime, { now: secondBucket }) - await second.runtime.sweep({ now: secondBucket, includeEvents: false }) + try { + const queue = [ + { id: 1, packageName: '@example/one', version: '1.0.0' }, + { id: 2, packageName: '@example/two', version: '1.0.0' }, + ] + const processed: Array = [] + const store = inMemoryWorkflowExecutionStore() + const { processWorkflow, runtime } = createTestIntentRuntime({ + store, + operations: { + ...noopOperations, + selectPendingIntentVersions: async ({ limit }) => + queue.splice(0, limit), + processIntentVersion: async (versionId: number) => { + const packageName = versionId === 1 ? '@example/one' : '@example/two' + const version = '1.0.0' + processed.push(`${packageName}@${version}`) + wallNow += 600 + return { + packageName, + version, + status: 'synced', + skillCount: 1, + } + }, + }, + }) + const runId = 'intent-process:deadline-resume' - const runs = await store.listRuns({ - workflowId: first.processWorkflow.id, - limit: 10, - }) + const first = await runtime.startRun({ + workflowId: processWorkflow.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.deepEqual(processed, ['@example/one@1.0.0', '@example/two@1.0.0']) - assert.equal(runs.length, 2) + assert.equal(first.kind, 'paused') + assert.equal(paused?.status, 'paused') + assert.equal(paused?.lease, undefined) + assert.equal(resumed.timers[0]?.kind, 'completed') + assert.equal(resumed.timers[0]?.runId, runId) + assert.deepEqual(processed, ['@example/one@1.0.0', '@example/two@1.0.0']) + assert.deepEqual(resumed.timers[0]?.run?.output, { + processed: 2, + failed: 0, + deferred: 0, + results: [ + { + packageName: '@example/one', + version: '1.0.0', + status: 'synced', + skillCount: 1, + }, + { + packageName: '@example/two', + version: '1.0.0', + status: 'synced', + skillCount: 1, + }, + ], + }) + } finally { + mock.restoreAll() + } }) const noopOperations: IntentSyncOperations = {