improvement(processing): reduce redundant DB queries in execution preprocessing#3320
Merged
waleedlatif1 merged 19 commits intostagingfrom Feb 24, 2026
Merged
improvement(processing): reduce redundant DB queries in execution preprocessing#3320waleedlatif1 merged 19 commits intostagingfrom
waleedlatif1 merged 19 commits intostagingfrom
Conversation
…processing Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
|
The latest updates on your projects. Learn more about Vercel for GitHub. |
Collaborator
Author
|
@cursor review |
Collaborator
Author
|
@greptile |
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
Contributor
Greptile SummaryThis PR reduces redundant database queries in the execution preprocessing pipeline by threading pre-fetched data (workflow records, subscription info, workspace IDs) through call chains instead of re-querying. It also removes verbose
Confidence Score: 3/5
Important Files Changed
Sequence DiagramsequenceDiagram
participant Client
participant ExecuteRoute as execute/route.ts
participant AuthZ as authorizeWorkflow
participant Preprocess as preprocessExecution
participant SubDB as Subscription DB
participant UsageCheck as checkServerSideUsageLimits
participant Core as executeWorkflowCore
participant Logging as LoggingSession
Client->>ExecuteRoute: POST /workflows/{id}/execute
ExecuteRoute->>AuthZ: authorizeWorkflowByWorkspacePermission()
AuthZ-->>ExecuteRoute: {workflow record}
ExecuteRoute->>Preprocess: preprocessExecution({workflowRecord})
Note over Preprocess: Skip Step 1 DB query (reuse record)
Preprocess->>SubDB: getHighestPrioritySubscription(actorUserId)
SubDB-->>Preprocess: subscription (fetched once)
Preprocess->>UsageCheck: checkServerSideUsageLimits(userId, subscription)
Note over UsageCheck: Reuses subscription (no re-fetch)
UsageCheck-->>Preprocess: {isExceeded, ...}
Preprocess-->>ExecuteRoute: {workflowRecord, subscription, ...}
ExecuteRoute->>Core: executeWorkflowCore(snapshot, ...)
Core->>Core: Execute workflow blocks
Core-->>ExecuteRoute: result (returned immediately)
Note over Core,Logging: Fire-and-forget (void async)
Core->>Logging: safeComplete / safeCompleteWithError
Logging->>Logging: DB update (may race with process exit)
Last reviewed commit: 2860663 |
Collaborator
Author
|
@cursor review |
…ow record Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Collaborator
Author
|
@cursor review |
Collaborator
Author
|
@greptile |
Replace `as any` cast in non-SSE error path with proper `buildTraceSpans()` transformation, matching the SSE error path. Remove redundant `as any` cast in preprocessing.ts where the types already align. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Collaborator
Author
|
@greptile |
Collaborator
Author
|
@cursor review |
…logging - logger.ts: cast JSONB cost column to `WorkflowExecutionLog['cost']` instead of `any` in both `completeWorkflowExecution` and `getWorkflowExecution` - logger.ts: replace `(orgUsageBefore as any)?.toString?.()` with `String()` since COALESCE guarantees a non-null SQL aggregate value - logging-session.ts: cast JSONB cost to `AccumulatedCost` (the local interface) instead of `any` in `loadExistingCost` Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
…e in usage.ts Replace inline `Awaited<ReturnType<typeof getHighestPrioritySubscription>>` with the already-exported `HighestPrioritySubscription` type alias. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Collaborator
Author
|
@cursor review |
… types - preprocessing.ts: use exported `HighestPrioritySubscription` type instead of redeclaring via `Awaited<ReturnType<...>>` - deploy/route.ts, status/route.ts: cast `hasWorkflowChanged` args to `WorkflowState` instead of `any` (JSONB + object literal narrowing) - state/route.ts: type block sanitization and save with `BlockState` and `WorkflowState` instead of `any` - search-suggestions.ts: remove 8 unnecessary `as any` casts on `'date'` literal that already satisfies the `Suggestion['category']` union Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
…tion When executeWorkflowCore throws, its catch block fire-and-forgets safeCompleteWithError, then re-throws. The caller's catch block also fire-and-forgets safeCompleteWithError on the same LoggingSession. Both check this.completed (still false) before either's async DB write resolves, so both proceed to completeWorkflowExecution which uses additive SQL for billing — doubling the charged cost on every failed execution. Fix: add a synchronous `completing` flag set immediately before the async work begins. This blocks concurrent callers at the guard check. On failure, the flag is reset so the safe* fallback path (completeWithCostOnlyLog) can still attempt recovery. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
…vent completion races Move waitForCompletion() into markAsFailed() so every call site is automatically safe against in-flight fire-and-forget completions. Remove the now-redundant external waitForCompletion() calls in route.ts. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Collaborator
Author
|
@cursor review |
Collaborator
Author
|
@greptile |
…empty catch - completeWithCostOnlyLog now resets this.completing = false when the fallback itself fails, preventing a permanently stuck session - Use _disconnectError in MCP test-connection to signal intentional ignore Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Revert unrelated debug log removal — this file isn't part of the processing improvements and the log aids connection leak detection. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Collaborator
Author
|
@cursor review |
- preprocessing.ts: use undefined (not null) for failed subscription fetch so getUserUsageLimit does a fresh lookup instead of silently falling back to free-tier limits - deployed/route.ts: log warning on loadDeployedWorkflowState failure instead of silently swallowing the error - schedule-execution.ts: remove dead successLog parameter and all call-site arguments left over from logger.debug cleanup - mcp/middleware.ts: drop unused error binding in empty catch - audit/log.ts, wand.ts: promote logger.debug to logger.warn in catch blocks where these are the only failure signal Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
getHighestPrioritySubscription never throws (it catches internally and returns null), so the catch block in preprocessExecution is dead code. The null vs undefined distinction doesn't matter and the coercions added unnecessary complexity. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
…itySubscription getHighestPrioritySubscription catches internally and returns null on error, so the wrapping try/catch was unreachable dead code. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
No longer called after createSnapshotWithDeduplication was refactored to use a single upsert instead of select-then-insert. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Collaborator
Author
|
@cursor review |
Collaborator
Author
|
@greptile |
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 2 potential issues.
Bugbot Autofix is OFF. To automatically fix reported issues with Cloud Agents, enable Autofix in the Cursor dashboard.
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
Comment on lines
+363
to
+404
| // Fire-and-forget: post-execution logging, billing, and cleanup | ||
| void (async () => { | ||
| try { | ||
| const { traceSpans, totalDuration } = buildTraceSpans(result) | ||
|
|
||
| if (result.success && result.status !== 'paused') { | ||
| try { | ||
| await updateWorkflowRunCounts(workflowId) | ||
| } catch (runCountError) { | ||
| logger.error(`[${requestId}] Failed to update run counts`, { error: runCountError }) | ||
| } | ||
| } | ||
|
|
||
| await loggingSession.safeComplete({ | ||
| endedAt: new Date().toISOString(), | ||
| totalDurationMs: totalDuration || 0, | ||
| finalOutput: result.output || {}, | ||
| traceSpans: traceSpans || [], | ||
| workflowInput: processedInput, | ||
| executionState: result.executionState, | ||
| }) | ||
| if (result.status === 'cancelled') { | ||
| await loggingSession.safeCompleteWithCancellation({ | ||
| endedAt: new Date().toISOString(), | ||
| totalDurationMs: totalDuration || 0, | ||
| traceSpans: traceSpans || [], | ||
| }) | ||
| } else if (result.status === 'paused') { | ||
| await loggingSession.safeCompleteWithPause({ | ||
| endedAt: new Date().toISOString(), | ||
| totalDurationMs: totalDuration || 0, | ||
| traceSpans: traceSpans || [], | ||
| workflowInput: processedInput, | ||
| }) | ||
| } else { | ||
| await loggingSession.safeComplete({ | ||
| endedAt: new Date().toISOString(), | ||
| totalDurationMs: totalDuration || 0, | ||
| finalOutput: result.output || {}, | ||
| traceSpans: traceSpans || [], | ||
| workflowInput: processedInput, | ||
| executionState: result.executionState, | ||
| }) | ||
| } | ||
|
|
||
| await clearExecutionCancellation(executionId) | ||
| await clearExecutionCancellation(executionId) | ||
| } catch (postExecError) { | ||
| logger.error(`[${requestId}] Post-execution logging failed`, { error: postExecError }) | ||
| } | ||
| })() |
Contributor
There was a problem hiding this comment.
Fire-and-forget may silently lose execution logs and billing
Wrapping all post-execution work (logging, billing, run-count updates, cancellation cleanup) in void (async () => {...})() means the function returns result before any of this completes. If the Node.js process shuts down or the serverless function's lifetime ends before the IIFE settles (common in edge/serverless environments, and also during Trigger.dev task completion), all of this work is silently dropped — no logs, no billing updates, no clearExecutionCancellation.
This is particularly risky because:
workflow-execution.ts(Trigger.dev background job) callsexecuteWorkflowCore, and after it returns, the Trigger.dev task may complete and the process may exit, racing against the fire-and-forget IIFE.- The same applies in the catch block (line ~417–440) where error logging is also fire-and-forget.
Consider either awaiting the post-execution work, or at minimum returning the completion promise so callers who can wait (like background jobs) have the option to await it.
Contributor
Additional Comments (2)
Every other completion method (
The catch block here does not reset |
royceP2
pushed a commit
to arenadeveloper02/p2-sim
that referenced
this pull request
Mar 3, 2026
…processing (simstudioai#3320) * improvement(processing): reduce redundant DB queries in execution preprocessing * improvement(processing): add defensive ID check for prefetched workflow record * improvement(processing): fix type safety in execution error logging Replace `as any` cast in non-SSE error path with proper `buildTraceSpans()` transformation, matching the SSE error path. Remove redundant `as any` cast in preprocessing.ts where the types already align. * improvement(processing): replace `as any` casts with proper types in logging - logger.ts: cast JSONB cost column to `WorkflowExecutionLog['cost']` instead of `any` in both `completeWorkflowExecution` and `getWorkflowExecution` - logger.ts: replace `(orgUsageBefore as any)?.toString?.()` with `String()` since COALESCE guarantees a non-null SQL aggregate value - logging-session.ts: cast JSONB cost to `AccumulatedCost` (the local interface) instead of `any` in `loadExistingCost` * improvement(processing): use exported HighestPrioritySubscription type in usage.ts Replace inline `Awaited<ReturnType<typeof getHighestPrioritySubscription>>` with the already-exported `HighestPrioritySubscription` type alias. * improvement(processing): replace remaining `as any` casts with proper types - preprocessing.ts: use exported `HighestPrioritySubscription` type instead of redeclaring via `Awaited<ReturnType<...>>` - deploy/route.ts, status/route.ts: cast `hasWorkflowChanged` args to `WorkflowState` instead of `any` (JSONB + object literal narrowing) - state/route.ts: type block sanitization and save with `BlockState` and `WorkflowState` instead of `any` - search-suggestions.ts: remove 8 unnecessary `as any` casts on `'date'` literal that already satisfies the `Suggestion['category']` union * fix(processing): prevent double-billing race in LoggingSession completion When executeWorkflowCore throws, its catch block fire-and-forgets safeCompleteWithError, then re-throws. The caller's catch block also fire-and-forgets safeCompleteWithError on the same LoggingSession. Both check this.completed (still false) before either's async DB write resolves, so both proceed to completeWorkflowExecution which uses additive SQL for billing — doubling the charged cost on every failed execution. Fix: add a synchronous `completing` flag set immediately before the async work begins. This blocks concurrent callers at the guard check. On failure, the flag is reset so the safe* fallback path (completeWithCostOnlyLog) can still attempt recovery. * fix(processing): unblock error responses and isolate run-count failures Remove unnecessary `await waitForCompletion()` from non-SSE and SSE error paths where no `markAsFailed()` follows — these were blocking error responses on log persistence for no reason. Wrap `updateWorkflowRunCounts` in its own try/catch so a run-count DB failure cannot prevent session completion, billing, and trace span persistence. * improvement(processing): remove dead setupExecutor method The method body was just a debug log with an `any` parameter — logging now works entirely through trace spans with no executor integration. * remove logger.debug * fix(processing): guard completionPromise as write-once (singleton promise) Prevent concurrent safeComplete* calls from overwriting completionPromise with a no-op. The guard now lives at the assignment site — if a completion is already in-flight, return its promise instead of starting a new one. This ensures waitForCompletion() always awaits the real work. * improvement(processing): remove empty else/catch blocks left by debug log cleanup * fix(processing): enforce waitForCompletion inside markAsFailed to prevent completion races Move waitForCompletion() into markAsFailed() so every call site is automatically safe against in-flight fire-and-forget completions. Remove the now-redundant external waitForCompletion() calls in route.ts. * fix(processing): reset completing flag on fallback failure, clean up empty catch - completeWithCostOnlyLog now resets this.completing = false when the fallback itself fails, preventing a permanently stuck session - Use _disconnectError in MCP test-connection to signal intentional ignore * fix(processing): restore disconnect error logging in MCP test-connection Revert unrelated debug log removal — this file isn't part of the processing improvements and the log aids connection leak detection. * fix(processing): address audit findings across branch - preprocessing.ts: use undefined (not null) for failed subscription fetch so getUserUsageLimit does a fresh lookup instead of silently falling back to free-tier limits - deployed/route.ts: log warning on loadDeployedWorkflowState failure instead of silently swallowing the error - schedule-execution.ts: remove dead successLog parameter and all call-site arguments left over from logger.debug cleanup - mcp/middleware.ts: drop unused error binding in empty catch - audit/log.ts, wand.ts: promote logger.debug to logger.warn in catch blocks where these are the only failure signal * revert: undo unnecessary subscription null→undefined change getHighestPrioritySubscription never throws (it catches internally and returns null), so the catch block in preprocessExecution is dead code. The null vs undefined distinction doesn't matter and the coercions added unnecessary complexity. * improvement(processing): remove dead try/catch around getHighestPrioritySubscription getHighestPrioritySubscription catches internally and returns null on error, so the wrapping try/catch was unreachable dead code. * improvement(processing): remove dead getSnapshotByHash method No longer called after createSnapshotWithDeduplication was refactored to use a single upsert instead of select-then-insert. ---------
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
getHighestPrioritySubscriptionfetches per executionpreprocessResult.workflowRecordin background execution instead of re-fetching withgetWorkflowByIdworkspaceIdtoloadDeployedWorkflowStateto skip internal workflow table re-queryTest plan
bunx tsc --noEmitpasses cleanbun run lintpasses clean