improvement(webhooks): stop redoing ingest work in the same-process worker

preprocessExecution gains trustWorkflowRecord (skip the archived-state
re-read for a row fetched in the same request) and skipAccountChecks
(skip the ban + subscription re-reads; guarded to checkRateLimit: false).
Webhook ingest passes trust for the row findAllWebhooksForPath returned;
the inline runner closures hand the ingest-loaded workflow + webhook rows
to executeWebhookJob as memory-only warm context, so the worker drops the
third workflow fetch, the webhook re-select, and the duplicate account
checks. Trigger.dev and recovery jobs pass no warm context and are
byte-identical to before.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: icecrasher321

apps/sim/background/webhook-execution.test.ts

@@ -351,6 +351,154 @@ describe('executeWebhookJob fault vs error handling', () => {
     )
   })
 
+  it('reuses ingest-loaded rows and skips duplicate account checks with warm context', async () => {
+    mockExecuteWorkflowCore.mockResolvedValue({
+      success: true,
+      status: 'completed',
+      output: {},
+      logs: [],
+      executionState: {
+        blockStates: {},
+        executedBlocks: [],
+        blockLogs: [],
+        decisions: {},
+        completedLoops: [],
+        activeExecutionPath: [],
+      },
+    })
+    const warmContext = {
+      workflowRecord: { id: 'workflow-1', workspaceId: 'workspace-1', userId: 'user-1' },
+      webhookRecord: { id: 'webhook-1', providerConfig: { warm: true } },
+    } as unknown as NonNullable<Parameters<typeof executeWebhookJob>[2]>
+
+    await executeWebhookJob(payload, undefined, warmContext)
+
+    expect(executionPreprocessingMockFns.mockPreprocessExecution).toHaveBeenCalledWith(
+      expect.objectContaining({
+        workflowRecord: expect.objectContaining({ id: 'workflow-1' }),
+        trustWorkflowRecord: true,
+        skipAccountChecks: true,
+      })
+    )
+    expect(dbChainMockFns.limit).not.toHaveBeenCalled()
+    expect(mockResolveWebhookRecordProviderConfig).toHaveBeenCalledWith(
+      expect.objectContaining({ id: 'webhook-1', providerConfig: { warm: true } }),
+      'user-1',
+      'workspace-1',
+      expect.any(Object)
+    )
+  })
+
+  it('loads rows and keeps account checks without warm context', async () => {
+    mockExecuteWorkflowCore.mockResolvedValue({
+      success: true,
+      status: 'completed',
+      output: {},
+      logs: [],
+      executionState: {
+        blockStates: {},
+        executedBlocks: [],
+        blockLogs: [],
+        decisions: {},
+        completedLoops: [],
+        activeExecutionPath: [],
+      },
+    })
+
+    await executeWebhookJob(payload)
+
+    expect(executionPreprocessingMockFns.mockPreprocessExecution).toHaveBeenCalledWith(
+      expect.objectContaining({
+        workflowRecord: undefined,
+        trustWorkflowRecord: false,
+        skipAccountChecks: false,
+      })
+    )
+    expect(dbChainMockFns.limit).toHaveBeenCalled()
+  })
+
+  it('ignores warm rows whose ids do not match the payload', async () => {
+    mockExecuteWorkflowCore.mockResolvedValue({
+      success: true,
+      status: 'completed',
+      output: {},
+      logs: [],
+      executionState: {
+        blockStates: {},
+        executedBlocks: [],
+        blockLogs: [],
+        decisions: {},
+        completedLoops: [],
+        activeExecutionPath: [],
+      },
+    })
+    const warmContext = {
+      workflowRecord: { id: 'other-workflow' },
+      webhookRecord: { id: 'other-webhook' },
+    } as unknown as NonNullable<Parameters<typeof executeWebhookJob>[2]>
+
+    await executeWebhookJob(payload, undefined, warmContext)
+
+    expect(executionPreprocessingMockFns.mockPreprocessExecution).toHaveBeenCalledWith(
+      expect.objectContaining({
+        workflowRecord: undefined,
+        trustWorkflowRecord: false,
+        skipAccountChecks: false,
+      })
+    )
+    expect(dbChainMockFns.limit).toHaveBeenCalled()
+  })
+
+  it('logs phase timings and the executor-start metric for latency-tracked payloads', async () => {
+    mockExecuteWorkflowCore.mockResolvedValue({
+      success: true,
+      status: 'completed',
+      output: {},
+      logs: [],
+      executionState: {
+        blockStates: {},
+        executedBlocks: [],
+        blockLogs: [],
+        decisions: {},
+        completedLoops: [],
+        activeExecutionPath: [],
+      },
+    })
+
+    await executeWebhookJob({
+      ...payload,
+      webhookReceivedAt: Date.now() - 100,
+      triggerTimestampMs: Date.now() - 500,
+    })
+
+    expect(webhookExecutionLogger.info).toHaveBeenCalledWith(
+      '[request-1] Webhook dispatch latency',
+      expect.objectContaining({
+        dispatchLatencyMs: expect.any(Number),
+        triggerAgeMs: expect.any(Number),
+        preprocessMs: expect.any(Number),
+        loadsMs: expect.any(Number),
+        providerConfigMs: expect.any(Number),
+        formatInputMs: expect.any(Number),
+      })
+    )
+
+    const coreOptions = mockExecuteWorkflowCore.mock.calls[0]?.[0] as {
+      callbacks: { onBlockStart?: () => Promise<void> }
+    }
+    expect(coreOptions.callbacks.onBlockStart).toBeTypeOf('function')
+    await coreOptions.callbacks.onBlockStart?.()
+    await coreOptions.callbacks.onBlockStart?.()
+    const executorStartCalls = webhookExecutionLogger.info.mock.calls.filter(
+      ([message]: [string]) => String(message).includes('Webhook executor started')
+    )
+    expect(executorStartCalls).toHaveLength(1)
+    expect(executorStartCalls[0]?.[1]).toMatchObject({
+      executorStartLatencyMs: expect.any(Number),
+      executorStartTriggerAgeMs: expect.any(Number),
+    })
+  })
+
   it('does not pass provider-config provenance absent from the trigger input', async () => {
     mockGetEffectiveEnvironmentSnapshot.mockResolvedValue({
       personalEncrypted: { WEBHOOK_SECRET: 'personal-ciphertext' },

apps/sim/background/webhook-execution.ts

@@ -1,5 +1,5 @@
 import { db } from '@sim/db'
-import { account, webhook } from '@sim/db/schema'
+import { account, webhook, type workflow } from '@sim/db/schema'
 import { createLogger, runWithRequestContext } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
@@ -291,9 +291,21 @@ export type WebhookExecutionPayload = {
   executionTimeoutMs?: number
 }
 
+/**
+ * Memory-only rows the same-process inline runner hands over so the worker does
+ * not re-read what ingest just loaded. Never serialized into the persisted job
+ * payload — Trigger.dev and recovery paths run without it and load everything
+ * themselves.
+ */
+export interface WebhookWarmContext {
+  workflowRecord?: typeof workflow.$inferSelect
+  webhookRecord?: typeof webhook.$inferSelect
+}
+
 export async function executeWebhookJob(
   payload: WebhookExecutionPayload,
-  externalAbortSignal?: AbortSignal
+  externalAbortSignal?: AbortSignal,
+  warmContext?: WebhookWarmContext
 ) {
   const correlation = buildWebhookCorrelation(payload)
   const executionId = correlation.executionId
@@ -358,7 +370,8 @@ export async function executeWebhookJob(
           payload,
           correlation,
           timeoutController,
-          admissionCompleted
+          admissionCompleted,
+          warmContext
         )
       }
 
@@ -482,7 +495,8 @@ async function executeWebhookJobInternal(
   payload: WebhookExecutionPayload,
   correlation: AsyncExecutionCorrelation,
   timeoutController: ReturnType<typeof createTimeoutAbortController>,
-  admissionCompleted: boolean
+  admissionCompleted: boolean,
+  warmContext?: WebhookWarmContext
 ) {
   const { executionId, requestId } = correlation
   const loggingSession = new LoggingSession(
@@ -493,6 +507,8 @@ async function executeWebhookJobInternal(
   )
   loggingSession.setExecutionDeadlineAt(getExecutionDeadlineAt(timeoutController.signal))
 
+  const warmWorkflowRecord =
+    warmContext?.workflowRecord?.id === payload.workflowId ? warmContext.workflowRecord : undefined
   const preprocessStartedAt = Date.now()
   const preprocessResult = await preprocessExecution({
     workflowId: payload.workflowId,
@@ -509,6 +525,9 @@ async function executeWebhookJobInternal(
     billingAttribution: payload.billingAttribution,
     executionType: 'async',
     executionDeadlineAt: getExecutionDeadlineAt(timeoutController.signal)?.getTime(),
+    workflowRecord: warmWorkflowRecord,
+    trustWorkflowRecord: Boolean(warmWorkflowRecord),
+    skipAccountChecks: admissionCompleted && Boolean(warmWorkflowRecord),
   })
   const preprocessEndedAt = Date.now()
 
@@ -559,9 +578,13 @@ async function executeWebhookJobInternal(
           workspaceId
         )
       : loadDeployedWorkflowState(payload.workflowId, workspaceId)
+    const warmWebhookRecord =
+      warmContext?.webhookRecord?.id === payload.webhookId ? warmContext.webhookRecord : undefined
     const [workflowData, webhookRows, resolvedCredentialUserId] = await Promise.all([
       workflowStatePromise,
-      db.select().from(webhook).where(eq(webhook.id, payload.webhookId)).limit(1),
+      warmWebhookRecord
+        ? Promise.resolve([warmWebhookRecord])
+        : db.select().from(webhook).where(eq(webhook.id, payload.webhookId)).limit(1),
       payload.credentialId
         ? resolveCredentialAccountUserId(payload.credentialId)
         : Promise.resolve(undefined),

apps/sim/lib/execution/preprocessing.test.ts

@@ -145,6 +145,124 @@ describe('preprocessExecution deployment checks', () => {
   })
 })
 
+describe('preprocessExecution prefetched record trust and account-check skips', () => {
+  type PrefetchedRecord = NonNullable<Parameters<typeof preprocessExecution>[0]['workflowRecord']>
+  const prefetchedRecord = {
+    id: 'workflow-1',
+    userId: 'creator-1',
+    workspaceId: 'workspace-1',
+    isDeployed: true,
+    archivedAt: null,
+  } as unknown as PrefetchedRecord
+
+  beforeEach(() => {
+    vi.clearAllMocks()
+  })
+
+  it('skips the active-record re-read when the prefetched record is trusted', async () => {
+    const result = await preprocessExecution({
+      workflowId: 'workflow-1',
+      userId: 'user-1',
+      triggerType: 'webhook',
+      executionId: 'execution-1',
+      requestId: 'request-1',
+      checkRateLimit: false,
+      workflowRecord: prefetchedRecord,
+      trustWorkflowRecord: true,
+    })
+
+    expect(result.success).toBe(true)
+    expect(workflowAuthzMockFns.mockGetActiveWorkflowRecord).not.toHaveBeenCalled()
+  })
+
+  it('re-reads the active record for a prefetched record without trust', async () => {
+    const result = await preprocessExecution({
+      workflowId: 'workflow-1',
+      userId: 'user-1',
+      triggerType: 'webhook',
+      executionId: 'execution-1',
+      requestId: 'request-1',
+      checkRateLimit: false,
+      workflowRecord: prefetchedRecord,
+    })
+
+    expect(result.success).toBe(true)
+    expect(workflowAuthzMockFns.mockGetActiveWorkflowRecord).toHaveBeenCalledWith('workflow-1')
+  })
+
+  it('still rejects a trusted prefetched record that is archived', async () => {
+    const result = await preprocessExecution({
+      workflowId: 'workflow-1',
+      userId: 'user-1',
+      triggerType: 'webhook',
+      executionId: 'execution-1',
+      requestId: 'request-1',
+      checkRateLimit: false,
+      workflowRecord: {
+        ...prefetchedRecord,
+        archivedAt: new Date('2026-08-01T00:00:00.000Z'),
+      } as unknown as PrefetchedRecord,
+      trustWorkflowRecord: true,
+    })
+
+    expect(result).toEqual({
+      success: false,
+      error: { message: 'Workflow not found', statusCode: 404 },
+    })
+    expect(workflowAuthzMockFns.mockGetActiveWorkflowRecord).not.toHaveBeenCalled()
+  })
+
+  it('skips the ban and subscription reads when skipAccountChecks is set', async () => {
+    const result = await preprocessExecution({
+      workflowId: 'workflow-1',
+      userId: 'user-1',
+      triggerType: 'webhook',
+      executionId: 'execution-1',
+      requestId: 'request-1',
+      checkRateLimit: false,
+      skipAccountChecks: true,
+      workflowRecord: prefetchedRecord,
+      trustWorkflowRecord: true,
+    })
+
+    expect(result.success).toBe(true)
+    if (result.success) {
+      expect(result.actorSubscription).toBeNull()
+    }
+    expect(mockGetActivelyBannedUserIds).not.toHaveBeenCalled()
+    expect(getHighestPrioritySubscription).not.toHaveBeenCalled()
+  })
+
+  it('keeps the ban and subscription reads by default', async () => {
+    const result = await preprocessExecution({
+      workflowId: 'workflow-1',
+      userId: 'user-1',
+      triggerType: 'webhook',
+      executionId: 'execution-1',
+      requestId: 'request-1',
+      checkRateLimit: false,
+    })
+
+    expect(result.success).toBe(true)
+    expect(mockGetActivelyBannedUserIds).toHaveBeenCalled()
+    expect(getHighestPrioritySubscription).toHaveBeenCalled()
+  })
+
+  it('rejects skipAccountChecks combined with rate limiting', async () => {
+    await expect(
+      preprocessExecution({
+        workflowId: 'workflow-1',
+        userId: 'user-1',
+        triggerType: 'webhook',
+        executionId: 'execution-1',
+        requestId: 'request-1',
+        checkRateLimit: true,
+        skipAccountChecks: true,
+      })
+    ).rejects.toThrow('skipAccountChecks requires checkRateLimit: false')
+  })
+})
+
 describe('preprocessExecution correlation logging', () => {
   it('preserves trigger correlation when logging preprocessing failures', async () => {
     mockResolveSystemBillingAttribution.mockRejectedValueOnce(