From b1cb6c51da14af95716dc52cbc55b3f8b403ea79 Mon Sep 17 00:00:00 2001 From: piyush-da-automation Date: Wed, 15 Jul 2026 18:05:47 +0530 Subject: [PATCH 1/2] fix(operation-queue): guard queue ownership by operation id --- apps/sim/stores/operation-queue/store.test.ts | 264 +++++++++++++++--- apps/sim/stores/operation-queue/store.ts | 95 +++++-- apps/sim/stores/operation-queue/types.ts | 2 +- 3 files changed, 290 insertions(+), 71 deletions(-) diff --git a/apps/sim/stores/operation-queue/store.test.ts b/apps/sim/stores/operation-queue/store.test.ts index 2380ed71036..7a837f4e773 100644 --- a/apps/sim/stores/operation-queue/store.test.ts +++ b/apps/sim/stores/operation-queue/store.test.ts @@ -1,16 +1,43 @@ /** * @vitest-environment node */ +import Module from 'node:module' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { registerEmitFunctions, useOperationQueueStore } from '@/stores/operation-queue/store' +function addWorkflowOperation(id: string) { + useOperationQueueStore.getState().addToQueue({ + id, + workflowId: 'workflow-a', + userId: 'user-1', + operation: { + operation: 'replace-state', + target: 'workflow', + payload: { state: { operationId: id } }, + }, + }) +} + +function addBlockOperation(id: string, blockId: string) { + useOperationQueueStore.getState().addToQueue({ + id, + workflowId: 'workflow-a', + userId: 'user-1', + operation: { + operation: 'block-update', + target: 'block', + payload: { id: blockId }, + }, + }) +} + describe('operation queue room gating', () => { beforeEach(() => { vi.clearAllMocks() useOperationQueueStore.setState({ operations: [], workflowOperationVersions: {}, - isProcessing: false, + processingOperationId: null, hasOperationError: false, }) registerEmitFunctions(vi.fn(), vi.fn(), vi.fn(), null) @@ -20,7 +47,7 @@ describe('operation queue room gating', () => { useOperationQueueStore.setState({ operations: [], workflowOperationVersions: {}, - isProcessing: false, + processingOperationId: null, hasOperationError: false, }) registerEmitFunctions(vi.fn(), vi.fn(), vi.fn(), null) @@ -92,7 +119,7 @@ describe('operation queue room gating', () => { expect(skippingEmit).toHaveBeenCalledTimes(1) const state = useOperationQueueStore.getState() - expect(state.isProcessing).toBe(false) + expect(state.processingOperationId).toBeNull() expect(state.hasOperationError).toBe(false) expect(state.operations).toEqual([ expect.objectContaining({ id: 'op-1', status: 'pending', retryCount: 0 }), @@ -133,63 +160,214 @@ describe('operation queue room gating', () => { useOperationQueueStore.getState().confirmOperation('op-1') }) - it('triggers offline mode for a non-retryable failure and recovers via clearError', () => { - registerEmitFunctions( - vi.fn(() => true), - vi.fn(), - vi.fn(), - 'workflow-a' - ) + it('advances exactly once after a successful acknowledgement', () => { + const workflowEmit = vi.fn(() => true) + registerEmitFunctions(workflowEmit, vi.fn(), vi.fn(), 'workflow-a') - useOperationQueueStore.getState().addToQueue({ - id: 'op-1', - workflowId: 'workflow-a', - userId: 'user-1', - operation: { - operation: 'replace-state', - target: 'workflow', - payload: { state: {} }, - }, + addWorkflowOperation('op-1') + addWorkflowOperation('op-2') + + expect(workflowEmit).toHaveBeenCalledTimes(1) + expect(useOperationQueueStore.getState().processingOperationId).toBe('op-1') + + useOperationQueueStore.getState().confirmOperation('op-1') + + expect(workflowEmit).toHaveBeenCalledTimes(2) + expect(useOperationQueueStore.getState().processingOperationId).toBe('op-2') + + useOperationQueueStore.getState().confirmOperation('op-2') + + expect(workflowEmit).toHaveBeenCalledTimes(2) + expect(useOperationQueueStore.getState().processingOperationId).toBeNull() + expect(useOperationQueueStore.getState().operations).toEqual([]) + }) + + it('does not release the active operation for an unknown acknowledgement', () => { + const workflowEmit = vi.fn(() => true) + registerEmitFunctions(workflowEmit, vi.fn(), vi.fn(), 'workflow-a') + + addWorkflowOperation('op-1') + addWorkflowOperation('op-2') + + useOperationQueueStore.getState().confirmOperation('unknown-operation') + + expect(workflowEmit).toHaveBeenCalledTimes(1) + expect(useOperationQueueStore.getState().processingOperationId).toBe('op-1') + expect(useOperationQueueStore.getState().operations.map((operation) => operation.id)).toEqual([ + 'op-1', + 'op-2', + ]) + + useOperationQueueStore.getState().confirmOperation('op-1') + useOperationQueueStore.getState().confirmOperation('op-2') + }) + + it('does not release the active operation for a stale acknowledgement', () => { + const workflowEmit = vi.fn(() => true) + registerEmitFunctions(workflowEmit, vi.fn(), vi.fn(), 'workflow-a') + + addWorkflowOperation('op-1') + addWorkflowOperation('op-2') + addWorkflowOperation('op-3') + + useOperationQueueStore.getState().confirmOperation('op-2') + + expect(workflowEmit).toHaveBeenCalledTimes(1) + expect(useOperationQueueStore.getState().processingOperationId).toBe('op-1') + expect(useOperationQueueStore.getState().operations.map((operation) => operation.id)).toEqual([ + 'op-1', + 'op-3', + ]) + + useOperationQueueStore.getState().confirmOperation('op-1') + + expect(workflowEmit).toHaveBeenCalledTimes(2) + expect(useOperationQueueStore.getState().processingOperationId).toBe('op-3') + useOperationQueueStore.getState().confirmOperation('op-3') + }) + + it('ignores a stale failure for a pending operation', () => { + const workflowEmit = vi.fn(() => true) + registerEmitFunctions(workflowEmit, vi.fn(), vi.fn(), 'workflow-a') + + addWorkflowOperation('op-1') + addWorkflowOperation('op-2') + + useOperationQueueStore.getState().failOperation('op-2') + + expect(workflowEmit).toHaveBeenCalledTimes(1) + expect(useOperationQueueStore.getState().processingOperationId).toBe('op-1') + expect(useOperationQueueStore.getState().operations).toEqual([ + expect.objectContaining({ id: 'op-1', status: 'processing', retryCount: 0 }), + expect.objectContaining({ id: 'op-2', status: 'pending', retryCount: 0 }), + ]) + + useOperationQueueStore.getState().confirmOperation('op-1') + useOperationQueueStore.getState().confirmOperation('op-2') + }) + + it('does not advance when cancelling an unrelated pending operation', () => { + const workflowEmit = vi.fn(() => true) + registerEmitFunctions(workflowEmit, vi.fn(), vi.fn(), 'workflow-a') + + addWorkflowOperation('op-1') + addBlockOperation('op-2', 'block-2') + addWorkflowOperation('op-3') + + useOperationQueueStore.getState().cancelOperationsForBlock('block-2') + + expect(workflowEmit).toHaveBeenCalledTimes(1) + expect(useOperationQueueStore.getState().processingOperationId).toBe('op-1') + expect(useOperationQueueStore.getState().operations.map((operation) => operation.id)).toEqual([ + 'op-1', + 'op-3', + ]) + + useOperationQueueStore.getState().confirmOperation('op-1') + + expect(workflowEmit).toHaveBeenCalledTimes(2) + expect(useOperationQueueStore.getState().processingOperationId).toBe('op-3') + useOperationQueueStore.getState().confirmOperation('op-3') + }) + + it('releases ownership and advances once when cancelling the active operation', () => { + const workflowEmit = vi.fn(() => true) + registerEmitFunctions(workflowEmit, vi.fn(), vi.fn(), 'workflow-a') + + addBlockOperation('op-1', 'block-1') + addWorkflowOperation('op-2') + + useOperationQueueStore.getState().cancelOperationsForBlock('block-1') + + expect(workflowEmit).toHaveBeenCalledTimes(2) + expect(useOperationQueueStore.getState().processingOperationId).toBe('op-2') + expect(useOperationQueueStore.getState().operations).toEqual([ + expect.objectContaining({ id: 'op-2', status: 'processing' }), + ]) + + useOperationQueueStore.getState().confirmOperation('op-2') + expect(workflowEmit).toHaveBeenCalledTimes(2) + expect(useOperationQueueStore.getState().processingOperationId).toBeNull() + }) + + it('drops a failed operation with a missing target and advances once', () => { + const originalRequire = Module.prototype.require + const requireSpy = vi.spyOn(Module.prototype, 'require').mockImplementation(function ( + moduleId: string + ) { + if (moduleId === '@/stores/workflows/workflow/store') { + return { useWorkflowStore: { getState: () => ({ blocks: {} }) } } + } + return originalRequire.call(this, moduleId) }) + try { + const workflowEmit = vi.fn(() => true) + registerEmitFunctions(workflowEmit, vi.fn(), vi.fn(), 'workflow-a') + + addBlockOperation('op-1', 'missing-block') + addWorkflowOperation('op-2') + + useOperationQueueStore.getState().failOperation('op-1', false) + + expect(workflowEmit).toHaveBeenCalledTimes(2) + expect(useOperationQueueStore.getState().hasOperationError).toBe(false) + expect(useOperationQueueStore.getState().processingOperationId).toBe('op-2') + expect(useOperationQueueStore.getState().operations).toEqual([ + expect.objectContaining({ id: 'op-2', status: 'processing' }), + ]) + + useOperationQueueStore.getState().confirmOperation('op-2') + } finally { + requireSpy.mockRestore() + } + }) + + it('triggers offline mode for a non-retryable failure and recovers via clearError', () => { + const workflowEmit = vi.fn(() => true) + registerEmitFunctions(workflowEmit, vi.fn(), vi.fn(), 'workflow-a') + + addWorkflowOperation('op-1') + addWorkflowOperation('op-2') + useOperationQueueStore.getState().failOperation('op-1', false) + expect(workflowEmit).toHaveBeenCalledTimes(1) expect(useOperationQueueStore.getState().hasOperationError).toBe(true) expect(useOperationQueueStore.getState().operations).toEqual([]) + expect(useOperationQueueStore.getState().processingOperationId).toBeNull() useOperationQueueStore.getState().clearError() expect(useOperationQueueStore.getState().hasOperationError).toBe(false) }) - it('triggers offline mode once retries exhaust for retryable failures', () => { - registerEmitFunctions( - vi.fn(() => true), - vi.fn(), - vi.fn(), - 'workflow-a' - ) + it('retries only after ownership is reacquired and enters offline mode after max retries', async () => { + vi.useFakeTimers() + try { + const workflowEmit = vi.fn(() => true) + registerEmitFunctions(workflowEmit, vi.fn(), vi.fn(), 'workflow-a') + addWorkflowOperation('op-1') - useOperationQueueStore.getState().addToQueue({ - id: 'op-1', - workflowId: 'workflow-a', - userId: 'user-1', - operation: { - operation: 'replace-state', - target: 'workflow', - payload: { state: {} }, - }, - }) + for (const delay of [2000, 4000, 8000]) { + useOperationQueueStore.getState().failOperation('op-1', true) - useOperationQueueStore.getState().failOperation('op-1', true) - useOperationQueueStore.getState().failOperation('op-1', true) - useOperationQueueStore.getState().failOperation('op-1', true) - expect(useOperationQueueStore.getState().hasOperationError).toBe(false) + expect(useOperationQueueStore.getState().processingOperationId).toBeNull() + expect(useOperationQueueStore.getState().hasOperationError).toBe(false) - useOperationQueueStore.getState().failOperation('op-1', true) + await vi.advanceTimersByTimeAsync(delay) + expect(useOperationQueueStore.getState().processingOperationId).toBe('op-1') + } - expect(useOperationQueueStore.getState().hasOperationError).toBe(true) - expect(useOperationQueueStore.getState().operations).toEqual([]) + useOperationQueueStore.getState().failOperation('op-1', true) + + expect(workflowEmit).toHaveBeenCalledTimes(4) + expect(useOperationQueueStore.getState().hasOperationError).toBe(true) + expect(useOperationQueueStore.getState().operations).toEqual([]) + expect(useOperationQueueStore.getState().processingOperationId).toBeNull() + } finally { + vi.useRealTimers() + } }) it('reports pending operations per workflow', () => { diff --git a/apps/sim/stores/operation-queue/store.ts b/apps/sim/stores/operation-queue/store.ts index 200abdf2b3f..2d3fb8d68f9 100644 --- a/apps/sim/stores/operation-queue/store.ts +++ b/apps/sim/stores/operation-queue/store.ts @@ -68,6 +68,25 @@ export function registerEmitFunctions( let currentRegisteredWorkflowId: string | null = null +function releaseQueueOwnership(operationId: string): boolean { + const state = useOperationQueueStore.getState() + if (state.processingOperationId !== operationId) { + return false + } + + useOperationQueueStore.setState({ processingOperationId: null }) + return true +} + +function advanceQueueAfterRelease(operationId: string): boolean { + if (!releaseQueueOwnership(operationId)) { + return false + } + + useOperationQueueStore.getState().processNextOperation() + return true +} + /** Targets whose payload id refers to a canvas block (subflow ids are loop/parallel blocks). */ const BLOCK_SCOPED_TARGETS = ['block', 'subblock', 'subflow'] @@ -107,18 +126,17 @@ function dropOperationForMissingTarget(operation: QueuedOperation): boolean { targetId, } ) - useOperationQueueStore.setState((s) => ({ - operations: s.operations.filter((op) => op.id !== operation.id), - isProcessing: false, + useOperationQueueStore.setState((state) => ({ + operations: state.operations.filter((op) => op.id !== operation.id), })) - useOperationQueueStore.getState().processNextOperation() + advanceQueueAfterRelease(operation.id) return true } export const useOperationQueueStore = create((set, get) => ({ operations: [], workflowOperationVersions: {}, - isProcessing: false, + processingOperationId: null, hasOperationError: false, addToQueue: (operation) => { @@ -222,6 +240,11 @@ export const useOperationQueueStore = create((set, get) => confirmOperation: (operationId) => { const state = get() const operation = state.operations.find((op) => op.id === operationId) + if (!operation) { + logger.debug('Ignoring confirmation for operation not in queue', { operationId }) + return + } + const newOperations = state.operations.filter((op) => op.id !== operationId) const retryTimeout = retryTimeouts.get(operationId) @@ -241,9 +264,8 @@ export const useOperationQueueStore = create((set, get) => remainingOps: newOperations.length, }) - set({ operations: newOperations, isProcessing: false }) - - get().processNextOperation() + set({ operations: newOperations }) + advanceQueueAfterRelease(operationId) }, failOperation: (operationId: string, retryable = true) => { @@ -254,6 +276,14 @@ export const useOperationQueueStore = create((set, get) => return } + if (state.processingOperationId !== operationId) { + logger.debug('Ignoring failure for operation that does not own the queue', { + operationId, + processingOperationId: state.processingOperationId, + }) + return + } + const operationTimeout = operationTimeouts.get(operationId) if (operationTimeout) { clearTimeout(operationTimeout) @@ -308,8 +338,8 @@ export const useOperationQueueStore = create((set, get) => ? { ...op, retryCount: newRetryCount, status: 'pending' as const } : op ), - isProcessing: false, })) + releaseQueueOwnership(operationId) const timeout = setTimeout(() => { retryTimeouts.delete(operationId) @@ -349,7 +379,7 @@ export const useOperationQueueStore = create((set, get) => processNextOperation: () => { const state = get() - if (state.isProcessing) { + if (state.processingOperationId) { return } @@ -368,7 +398,7 @@ export const useOperationQueueStore = create((set, get) => operations: state.operations.map((op) => op.id === nextOperation.id ? { ...op, status: 'processing' as const } : op ), - isProcessing: true, + processingOperationId: nextOperation.id, })) logger.debug('Processing operation sequentially', { @@ -421,8 +451,8 @@ export const useOperationQueueStore = create((set, get) => operations: state.operations.map((o) => o.id === nextOperation.id ? { ...o, status: 'pending' as const } : o ), - isProcessing: false, })) + releaseQueueOwnership(nextOperation.id) return } @@ -545,17 +575,20 @@ export const useOperationQueueStore = create((set, get) => return true }) - set({ - operations: newOperations, - isProcessing: false, - }) + set({ operations: newOperations }) logger.debug('Cancelled operations for block', { blockId, cancelledOperations: operationsToCancel.length, }) - get().processNextOperation() + const processingOperationId = state.processingOperationId + if ( + processingOperationId && + operationsToCancel.some((operation) => operation.id === processingOperationId) + ) { + advanceQueueAfterRelease(processingOperationId) + } }, cancelOperationsForVariable: (variableId: string) => { @@ -592,17 +625,20 @@ export const useOperationQueueStore = create((set, get) => ) ) - set({ - operations: newOperations, - isProcessing: false, - }) + set({ operations: newOperations }) logger.debug('Cancelled operations for variable', { variableId, cancelledOperations: operationsToCancel.length, }) - get().processNextOperation() + const processingOperationId = state.processingOperationId + if ( + processingOperationId && + operationsToCancel.some((operation) => operation.id === processingOperationId) + ) { + advanceQueueAfterRelease(processingOperationId) + } }, cancelOperationsForWorkflow: (workflowId: string) => { @@ -621,10 +657,15 @@ export const useOperationQueueStore = create((set, get) => operationTimeouts.delete(opId) } }) - set((s) => ({ - operations: s.operations.filter((op) => op.workflowId !== workflowId), - isProcessing: false, + const processingOperation = state.processingOperationId + ? state.operations.find((operation) => operation.id === state.processingOperationId) + : undefined + set((currentState) => ({ + operations: currentState.operations.filter((op) => op.workflowId !== workflowId), })) + if (processingOperation?.workflowId === workflowId) { + advanceQueueAfterRelease(processingOperation.id) + } }, triggerOfflineMode: () => { @@ -637,7 +678,7 @@ export const useOperationQueueStore = create((set, get) => set({ operations: [], - isProcessing: false, + processingOperationId: null, hasOperationError: true, }) }, @@ -662,7 +703,7 @@ export function useOperationQueue() { return useOperationQueueStore.getState().operations }, get isProcessing() { - return useOperationQueueStore.getState().isProcessing + return Boolean(useOperationQueueStore.getState().processingOperationId) }, hasOperationError, addToQueue: actions.addToQueue, diff --git a/apps/sim/stores/operation-queue/types.ts b/apps/sim/stores/operation-queue/types.ts index fb56a2a694b..ffec97ea0c8 100644 --- a/apps/sim/stores/operation-queue/types.ts +++ b/apps/sim/stores/operation-queue/types.ts @@ -44,7 +44,7 @@ export interface QueuedOperation { export interface OperationQueueState { operations: QueuedOperation[] workflowOperationVersions: Record - isProcessing: boolean + processingOperationId: string | null hasOperationError: boolean addToQueue: (operation: Omit) => void From 17b229502a72e0a8249063b258301c712e9fe9c3 Mon Sep 17 00:00:00 2001 From: piyush-da-automation Date: Wed, 15 Jul 2026 18:30:43 +0530 Subject: [PATCH 2/2] fix(operation-queue): advance after late retry acknowledgement --- apps/sim/stores/operation-queue/store.test.ts | 36 +++++++++++++++++++ apps/sim/stores/operation-queue/store.ts | 4 +++ 2 files changed, 40 insertions(+) diff --git a/apps/sim/stores/operation-queue/store.test.ts b/apps/sim/stores/operation-queue/store.test.ts index 7a837f4e773..6f5d46c596d 100644 --- a/apps/sim/stores/operation-queue/store.test.ts +++ b/apps/sim/stores/operation-queue/store.test.ts @@ -182,6 +182,42 @@ describe('operation queue room gating', () => { expect(useOperationQueueStore.getState().operations).toEqual([]) }) + it('advances when an acknowledgement arrives during retry backoff', async () => { + vi.useFakeTimers() + try { + const workflowEmit = vi.fn(() => true) + registerEmitFunctions(workflowEmit, vi.fn(), vi.fn(), 'workflow-a') + + addWorkflowOperation('op-1') + addWorkflowOperation('op-2') + + useOperationQueueStore.getState().failOperation('op-1', true) + + expect(workflowEmit).toHaveBeenCalledTimes(1) + expect(useOperationQueueStore.getState().processingOperationId).toBeNull() + expect(useOperationQueueStore.getState().operations).toEqual([ + expect.objectContaining({ id: 'op-1', status: 'pending', retryCount: 1 }), + expect.objectContaining({ id: 'op-2', status: 'pending', retryCount: 0 }), + ]) + + useOperationQueueStore.getState().confirmOperation('op-1') + + expect(workflowEmit).toHaveBeenCalledTimes(2) + expect(useOperationQueueStore.getState().processingOperationId).toBe('op-2') + expect(useOperationQueueStore.getState().operations).toEqual([ + expect.objectContaining({ id: 'op-2', status: 'processing' }), + ]) + + await vi.advanceTimersByTimeAsync(2000) + + expect(workflowEmit).toHaveBeenCalledTimes(2) + expect(useOperationQueueStore.getState().processingOperationId).toBe('op-2') + useOperationQueueStore.getState().confirmOperation('op-2') + } finally { + vi.useRealTimers() + } + }) + it('does not release the active operation for an unknown acknowledgement', () => { const workflowEmit = vi.fn(() => true) registerEmitFunctions(workflowEmit, vi.fn(), vi.fn(), 'workflow-a') diff --git a/apps/sim/stores/operation-queue/store.ts b/apps/sim/stores/operation-queue/store.ts index 2d3fb8d68f9..69a1baf9e42 100644 --- a/apps/sim/stores/operation-queue/store.ts +++ b/apps/sim/stores/operation-queue/store.ts @@ -265,6 +265,10 @@ export const useOperationQueueStore = create((set, get) => }) set({ operations: newOperations }) + if (state.processingOperationId === null) { + get().processNextOperation() + return + } advanceQueueAfterRelease(operationId) },