Skip to content

Commit 099fad8

Browse files
committed
fix(workflows): pin the deployed-state hint to the active version
`loadDeployedWorkflowState` asks Postgres to withhold the version blob for the version id it hinted for the workflow, but the hint was written on every materialize miss, by-id reads included, and never on a cache hit. An execution or webhook loading an older version by id repointed the hint away from the active one, and a rollback to a version still in the cache left it on the version rolled back from, so each later active read pulled the full blob again until the hint expired. The hint is now written only by the active read, on every successful return, hit or miss, and `materializeDeploymentState` no longer touches it. Autopsy: the cache tests only ever read the active version, one version per workflow or one cutover, so the hint always matched; no test interleaved a by-id read or returned to a cached version, and the integration test checks answers, which stay correct when the hint is stale (it costs bytes, not correctness). The two new cases queue the withheld row a correct hint produces and fail on the old code with NoActiveDeploymentError.
1 parent 339b321 commit 099fad8

2 files changed

Lines changed: 59 additions & 14 deletions

File tree

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

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -817,6 +817,44 @@ describe('Database Helpers', () => {
817817
expect(dbChainMockFns.where).toHaveBeenCalledTimes(1)
818818
})
819819

820+
/**
821+
* The hint naming the version Postgres may withhold used to follow the last
822+
* materialize miss, by-id reads included, so an execution of an older version
823+
* repointed it and the next active reads pulled the full blob again until the
824+
* entry expired. `state: null` stands in for the row Postgres returns when it
825+
* withholds the blob for the hinted version.
826+
*/
827+
it('keeps withholding the active state after a by-id read of another version', async () => {
828+
queueActiveVersion('dv-active', buildDeployedState())
829+
await dbHelpers.loadDeployedWorkflowState('wf-hint', 'workspace-1')
830+
831+
queueTableRows(schemaMock.workflowDeploymentVersion, [
832+
{ id: 'dv-older', state: buildDeployedState() },
833+
])
834+
await dbHelpers.loadWorkflowDeploymentVersionState('wf-hint', 'dv-older', 'workspace-1')
835+
836+
queueActiveVersion('dv-active', null)
837+
const deployed = await dbHelpers.loadDeployedWorkflowState('wf-hint', 'workspace-1')
838+
839+
expect(deployed.deploymentVersionId).toBe('dv-active')
840+
expect(mockSanitizeAgentToolsInBlocks).toHaveBeenCalledTimes(2)
841+
})
842+
843+
it('withholds the active state again after a rollback to a version still cached', async () => {
844+
queueActiveVersion('dv-a', buildDeployedState())
845+
await dbHelpers.loadDeployedWorkflowState('wf-rollback', 'workspace-1')
846+
queueActiveVersion('dv-b', buildDeployedState())
847+
await dbHelpers.loadDeployedWorkflowState('wf-rollback', 'workspace-1')
848+
queueActiveVersion('dv-a', buildDeployedState())
849+
await dbHelpers.loadDeployedWorkflowState('wf-rollback', 'workspace-1')
850+
851+
queueActiveVersion('dv-a', null)
852+
const deployed = await dbHelpers.loadDeployedWorkflowState('wf-rollback', 'workspace-1')
853+
854+
expect(deployed.deploymentVersionId).toBe('dv-a')
855+
expect(mockSanitizeAgentToolsInBlocks).toHaveBeenCalledTimes(2)
856+
})
857+
820858
it('invalidateDeployedStateCache(id) forces a rebuild on the next call', async () => {
821859
queueActiveVersion('dv-inv', buildDeployedState())
822860
queueActiveVersion('dv-inv', buildDeployedState())

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

Lines changed: 21 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -146,9 +146,12 @@ const deployedStateCache = new LRUCache<
146146
})
147147

148148
/**
149-
* The version id each workflow last cached, so the active-version read can ask
150-
* Postgres to withhold a `state` this process already holds. Only a hint: a
151-
* stale or evicted entry costs a re-read, never a wrong answer.
149+
* The active version id each workflow last resolved to, so the active-version
150+
* read can ask Postgres to withhold a `state` this process already holds. Only
151+
* {@link loadDeployedWorkflowState} writes it: a by-id read of an older version
152+
* (an execution or webhook admitted before a cutover) must not repoint it away
153+
* from the active one. Only a hint: a stale or evicted entry costs a re-read,
154+
* never a wrong answer.
152155
*/
153156
const cachedVersionByWorkflow = new LRUCache<string, string>({
154157
max: DEPLOYED_STATE_CACHE_MAX_ENTRIES,
@@ -272,7 +275,6 @@ export async function materializeDeploymentState(
272275
if (options.cache === false) return deployedState
273276
/** The cached entry is shared, so each caller gets its own copy. */
274277
deployedStateCache.set(version.id, { workflowId, state: deployedState })
275-
cachedVersionByWorkflow.set(workflowId, version.id)
276278
return structuredClone(deployedState)
277279
}
278280

@@ -311,20 +313,25 @@ export async function loadDeployedWorkflowState(
311313
throw new NoActiveDeploymentError(workflowId)
312314
}
313315

316+
let deployedState: DeployedWorkflowData
314317
if (active.id === cachedVersionId && active.state == null) {
315318
// Served from the cache, or re-read by id if the entry was evicted since the hint.
316-
return loadWorkflowDeploymentVersionState(workflowId, active.id, providedWorkspaceId)
317-
}
318-
319-
if (!active.state) {
319+
deployedState = await loadWorkflowDeploymentVersionState(
320+
workflowId,
321+
active.id,
322+
providedWorkspaceId
323+
)
324+
} else if (!active.state) {
320325
throw new NoActiveDeploymentError(workflowId)
326+
} else {
327+
deployedState = await materializeDeploymentState(
328+
workflowId,
329+
active,
330+
await resolveWorkspaceId(workflowId, providedWorkspaceId)
331+
)
321332
}
322-
323-
return materializeDeploymentState(
324-
workflowId,
325-
active,
326-
await resolveWorkspaceId(workflowId, providedWorkspaceId)
327-
)
333+
cachedVersionByWorkflow.set(workflowId, active.id)
334+
return deployedState
328335
} catch (error) {
329336
// An undeployed workflow is an outcome each caller handles, not a load failure.
330337
if (!(error instanceof NoActiveDeploymentError)) {

0 commit comments

Comments
 (0)