improvement(webhooks): cache-first deployed-state reads keyed by deploymentVersionId

loadWorkflowDeploymentVersionState consults the existing 5-min LRU before
the SELECT (the id is immutable, and entries now carry their workflowId so
a mismatched pair still falls through to the query). blockExistsInDeployment
routes through that loader when the webhook row's admitted version id is
known, instead of re-reading the entire state jsonb for one boolean — which
also warms the in-process cache the inline execution path reads moments
later. The raw active-version read remains the null-id fallback, and any
failure still answers false.

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

Author: icecrasher321

apps/sim/lib/webhooks/processor.ts

@@ -870,7 +870,10 @@ export async function dispatchResolvedWebhookTarget(
   }
 
   if (webhookRecord.blockId) {
-    const blockExists = await blockExistsInDeployment(foundWorkflow.id, webhookRecord.blockId)
+    const blockExists = await blockExistsInDeployment(foundWorkflow.id, webhookRecord.blockId, {
+      deploymentVersionId: webhookRecord.deploymentVersionId,
+      workspaceId: foundWorkflow.workspaceId,
+    })
     if (!blockExists) {
       const verificationResponse = handlePreDeploymentVerification(webhookRecord, options.requestId)
       return {
@@ -956,7 +959,10 @@ export async function processPolledWebhookEvent(
   let reservationTransferred = false
   try {
     if (foundWebhook.blockId) {
-      const blockExists = await blockExistsInDeployment(foundWorkflow.id, foundWebhook.blockId)
+      const blockExists = await blockExistsInDeployment(foundWorkflow.id, foundWebhook.blockId, {
+        deploymentVersionId: foundWebhook.deploymentVersionId,
+        workspaceId: foundWorkflow.workspaceId,
+      })
       if (!blockExists) {
         logger.info(
           `[${requestId}] Trigger block ${foundWebhook.blockId} not found in deployment for workflow ${foundWorkflow.id}`

apps/sim/lib/workflows/persistence/utils.test.ts

@@ -1389,6 +1389,78 @@ describe('Database Helpers', () => {
       expect(dbChainMockFns.where).toHaveBeenCalledTimes(1)
     })
 
+    it('serves a warm admitted version from the cache without the version SELECT', async () => {
+      queueTableRows(schemaMock.workflowDeploymentVersion, [
+        { id: 'dv-warm', state: buildDeployedState() },
+      ])
+
+      await dbHelpers.loadWorkflowDeploymentVersionState('wf-warm', 'dv-warm', 'workspace-1')
+      const second = await dbHelpers.loadWorkflowDeploymentVersionState(
+        'wf-warm',
+        'dv-warm',
+        'workspace-1'
+      )
+
+      expect(second.deploymentVersionId).toBe('dv-warm')
+      expect(dbChainMockFns.where).toHaveBeenCalledTimes(1)
+      expect(mockSanitizeAgentToolsInBlocks).toHaveBeenCalledTimes(1)
+    })
+
+    it('does not serve a cached version to a different workflow id', async () => {
+      queueTableRows(schemaMock.workflowDeploymentVersion, [
+        { id: 'dv-mine', state: buildDeployedState() },
+      ])
+      await dbHelpers.loadWorkflowDeploymentVersionState('wf-mine', 'dv-mine', 'workspace-1')
+
+      await expect(
+        dbHelpers.loadWorkflowDeploymentVersionState('wf-other', 'dv-mine', 'workspace-1')
+      ).rejects.toThrow('Deployment dv-mine was not found for workflow wf-other')
+    })
+
+    it('blockExistsInDeployment answers from the admitted version and warms the cache', async () => {
+      queueTableRows(schemaMock.workflowDeploymentVersion, [
+        { id: 'dv-block', state: buildDeployedState() },
+      ])
+
+      await expect(
+        dbHelpers.blockExistsInDeployment('wf-block', 'block-1', {
+          deploymentVersionId: 'dv-block',
+          workspaceId: 'workspace-1',
+        })
+      ).resolves.toBe(true)
+
+      const deployed = await dbHelpers.loadWorkflowDeploymentVersionState(
+        'wf-block',
+        'dv-block',
+        'workspace-1'
+      )
+      expect(deployed.blocks['block-1']).toBeDefined()
+      expect(dbChainMockFns.where).toHaveBeenCalledTimes(1)
+
+      await expect(
+        dbHelpers.blockExistsInDeployment('wf-block', 'missing-block', {
+          deploymentVersionId: 'dv-block',
+          workspaceId: 'workspace-1',
+        })
+      ).resolves.toBe(false)
+    })
+
+    it('blockExistsInDeployment answers false when the admitted version is missing', async () => {
+      await expect(
+        dbHelpers.blockExistsInDeployment('wf-x', 'block-1', {
+          deploymentVersionId: 'dv-missing',
+          workspaceId: 'workspace-1',
+        })
+      ).resolves.toBe(false)
+    })
+
+    it('blockExistsInDeployment falls back to the raw active-version read without a version id', async () => {
+      queueTableRows(schemaMock.workflowDeploymentVersion, [{ state: buildDeployedState() }])
+
+      await expect(dbHelpers.blockExistsInDeployment('wf-raw', 'block-1')).resolves.toBe(true)
+      expect(mockSanitizeAgentToolsInBlocks).not.toHaveBeenCalled()
+    })
+
     it('invalidateDeployedStateCache(id) forces a rebuild on the next call', async () => {
       queueActiveVersion('dv-inv', buildDeployedState())
       queueActiveVersion('dv-inv', buildDeployedState())

apps/sim/lib/workflows/persistence/utils.ts

@@ -94,11 +94,29 @@ export interface DeployedWorkflowData extends NormalizedWorkflowData {
   variables?: Record<string, unknown>
 }
 
+/**
+ * Answers whether a trigger block exists in the workflow's deployment. When the
+ * caller already knows the admitted `deploymentVersionId` (webhook rows carry
+ * it), the check goes through the LRU-backed version loader — warming the cache
+ * the execution path reads moments later — instead of re-reading the full state
+ * jsonb for one boolean. Without an id it falls back to the raw active-version
+ * read. Any failure answers `false`, matching the historical contract.
+ */
 export async function blockExistsInDeployment(
   workflowId: string,
-  blockId: string
+  blockId: string,
+  options?: { deploymentVersionId?: string | null; workspaceId?: string | null }
 ): Promise<boolean> {
   try {
+    if (options?.deploymentVersionId) {
+      const deployed = await loadWorkflowDeploymentVersionState(
+        workflowId,
+        options.deploymentVersionId,
+        options.workspaceId ?? undefined
+      )
+      return Boolean(deployed.blocks[blockId])
+    }
+
     const [result] = await db
       .select({ state: workflowDeploymentVersion.state })
       .from(workflowDeploymentVersion)
@@ -131,10 +149,12 @@ const DEPLOYED_STATE_CACHE_TTL_MS = 5 * 60 * 1000
  * absolute on purpose — it bounds the one non-immutable part, the live credential
  * remap in `applyBlockMigrations` — so credential changes still propagate.
  */
-const deployedStateCache = new LRUCache<string, DeployedWorkflowData>({
-  max: DEPLOYED_STATE_CACHE_MAX_ENTRIES,
-  ttl: DEPLOYED_STATE_CACHE_TTL_MS,
-})
+const deployedStateCache = new LRUCache<string, { workflowId: string; data: DeployedWorkflowData }>(
+  {
+    max: DEPLOYED_STATE_CACHE_MAX_ENTRIES,
+    ttl: DEPLOYED_STATE_CACHE_TTL_MS,
+  }
+)
 
 /** Evicts one deployed-state entry, or clears the cache when no id is given. */
 export function invalidateDeployedStateCache(deploymentVersionId?: string): void {
@@ -191,8 +211,8 @@ export async function materializeDeploymentState(
   executor?: DbOrTx
 ): Promise<DeployedWorkflowData> {
   const cached = deployedStateCache.get(version.id)
-  if (cached) {
-    return structuredClone(cached)
+  if (cached?.workflowId === workflowId) {
+    return structuredClone(cached.data)
   }
 
   const state = version.state as WorkflowState & { variables?: Record<string, unknown> }
@@ -241,7 +261,7 @@ export async function materializeDeploymentState(
     deploymentVersionId: version.id,
   }
 
-  deployedStateCache.set(version.id, deployedState)
+  deployedStateCache.set(version.id, { workflowId, data: deployedState })
   return structuredClone(deployedState)
 }
 
@@ -283,12 +303,21 @@ export async function loadDeployedWorkflowState(
 
 /**
  * Loads an immutable deployment snapshot by ID for work admitted before a later cutover.
+ *
+ * Cache-first: the id is immutable, so a warm LRU entry (guarded by matching
+ * `workflowId`) is byte-identical to what the SELECT + materialization below
+ * would produce, minus the full-state jsonb round trip.
  */
 export async function loadWorkflowDeploymentVersionState(
   workflowId: string,
   deploymentVersionId: string,
   providedWorkspaceId?: string
 ): Promise<DeployedWorkflowData> {
+  const cached = deployedStateCache.get(deploymentVersionId)
+  if (cached?.workflowId === workflowId) {
+    return structuredClone(cached.data)
+  }
+
   const [version] = await db
     .select({
       id: workflowDeploymentVersion.id,