Use Workflow runtime deadlines for intent processing (#1054)

Use workflow runtime deadlines for intent processing

Author: tannerlinsley

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",

pnpm-lock.yaml

@@ -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"
           >
             <Play
               className={
@@ -249,21 +249,29 @@ function IntentAdminPage() {
       )}
       {processMutation.data && (
         <ResultBanner
-          title={`Processed ${processMutation.data.processed} version(s)`}
-          items={[
-            ...(processMutation.data.deferred > 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()}
         />
       )}

src/routes/admin/intent.tsx

@@ -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,

src/server/scheduled.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),
+    ),
+  }
 }
 
 // ---------------------------------------------------------------------------

src/utils/intent-admin.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<number>()
     const results: Array<IntentVersionProcessResult> = []
-    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
-}

src/utils/intent-workflows.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: {