consolidate

Author: icecrasher321

apps/sim/app/api/workflows/[id]/execute/cancel/route.ts

@@ -1,7 +1,7 @@
 import { type NextRequest, NextResponse } from 'next/server'
 import { z } from 'zod'
 import { checkHybridAuth } from '@/lib/auth/hybrid'
-import { requestCancellation } from '@/lib/execution/active-executors'
+import { requestCancellation } from '@/lib/execution/cancellation'
 
 const CancelExecutionSchema = z.object({
   executionId: z.string().uuid(),

apps/sim/app/api/workflows/[id]/execute/route.ts

@@ -7,7 +7,7 @@ import { isTriggerDevEnabled } from '@/lib/core/config/feature-flags'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { SSE_HEADERS } from '@/lib/core/utils/sse'
 import { getBaseUrl } from '@/lib/core/utils/urls'
-import { clearCancellation } from '@/lib/execution/active-executors'
+import { clearCancellation } from '@/lib/execution/cancellation'
 import { processInputFileFields } from '@/lib/execution/files'
 import { preprocessExecution } from '@/lib/execution/preprocessing'
 import { createLogger } from '@/lib/logs/console/logger'

apps/sim/executor/execution/engine.ts

@@ -1,4 +1,4 @@
-import { isCancellationRequested } from '@/lib/execution/active-executors'
+import { isCancellationRequested } from '@/lib/execution/cancellation'
 import { createLogger } from '@/lib/logs/console/logger'
 import { BlockType } from '@/executor/constants'
 import type { DAG } from '@/executor/dag/builder'

apps/sim/lib/execution/active-executors.ts

@@ -0,0 +1,50 @@
+import { getRedisClient } from '@/lib/core/config/redis'
+
+const KEY_PREFIX = 'execution:cancel:'
+const TTL_SECONDS = 300
+const TTL_MS = TTL_SECONDS * 1000
+
+const memoryStore = new Map<string, number>()
+
+export async function requestCancellation(executionId: string): Promise<boolean> {
+  const redis = getRedisClient()
+  if (redis) {
+    try {
+      await redis.set(`${KEY_PREFIX}${executionId}`, '1', 'EX', TTL_SECONDS)
+      return true
+    } catch {
+      return false
+    }
+  }
+  memoryStore.set(executionId, Date.now() + TTL_MS)
+  return true
+}
+
+export async function isCancellationRequested(executionId: string): Promise<boolean> {
+  const redis = getRedisClient()
+  if (redis) {
+    try {
+      return (await redis.exists(`${KEY_PREFIX}${executionId}`)) === 1
+    } catch {
+      return false
+    }
+  }
+  const expiry = memoryStore.get(executionId)
+  if (!expiry) return false
+  if (Date.now() > expiry) {
+    memoryStore.delete(executionId)
+    return false
+  }
+  return true
+}
+
+export async function clearCancellation(executionId: string): Promise<void> {
+  const redis = getRedisClient()
+  if (redis) {
+    try {
+      await redis.del(`${KEY_PREFIX}${executionId}`)
+    } catch {}
+    return
+  }
+  memoryStore.delete(executionId)
+}