[pull] main from activepieces:main

.agents/features/workers.md

@@ -4,10 +4,11 @@
 Workers are separate Node processes that poll the app for jobs and execute flows/triggers. **The worker *is* the sandbox**: each worker forks the engine **in-process** (`@activepieces/sandbox`, `createSandboxRuntime`) — there is no separate sandbox pool or Cloud Run runtime indirection. The destination model is one box per worker (`concurrency 1`), scaling out horizontally with replicas (each capped small, e.g. 0.5 vCPU / 1 GB) so an OOM kills one job, not a shared pool; a transitional compatibility mode still honours `AP_WORKER_CONCURRENCY` (N independent boxes). They connect to the app over a Socket.IO channel: on connect a worker fetches its runtime settings (`WorkerSettingsResponse`) and the app registers an RPC server (`WorkerToApiContract`) for that socket. Jobs are pulled by the worker via `poll()` rather than pushed. A worker advertises liveness and config through `MachineInformation` (heartbeat), whose `workerProps` carry its identity including `version`. In the default Docker image both `activepieces-app` and `activepieces-worker` run under PM2 from `WORKDIR /usr/src/app`; `AP_CONTAINER_TYPE` (`APP` / `WORKER` / `WORKER_AND_APP`) selects which start.
 
 ## Key Files
-- `packages/server/api/src/app/workers/machine/machine-controller.ts` — Socket.IO listeners (`FETCH_WORKER_SETTINGS`, `DISCONNECT`); registers the RPC server per connection
+- `packages/server/api/src/app/workers/machine/machine-controller.ts` — Socket.IO listeners (`FETCH_WORKER_SETTINGS`, `DISCONNECT`); registers the RPC server per connection. The `DISCONNECT` handler returns that connection's in-flight jobs to the queue (`jobBroker.releaseConnectionJobs`) before `machineService.onDisconnect`
+- `packages/server/api/src/app/workers/job-queue/job-assignment-tracker.ts` — in-memory `(socket.id → in-flight jobs)` map (keyed by queue+jobId); recorded on `poll`, cleared on `completeJob`, drained on disconnect to bound BullMQ `active` to live worker concurrency
 - `packages/server/api/src/app/workers/machine/machine-service.ts` — `onConnection` / `onDisconnect`, `buildSettingsResponse` (emits `APP_VERSION`), worker listing
 - `packages/server/api/src/app/workers/rpc/worker-rpc-service.ts` — `createHandlers()`: `poll` (with version gate), `completeJob`, `extendLock`, progress/log RPCs, `getFlowBundle`, `prepareFlowBundleUpload`, `uploadFlowBundle`
-- `packages/server/worker/src/lib/worker.ts` — worker lifecycle (`worker.start/stop`), `pollAndExecute` loop (with version gate), `getWorkerProps`; builds the `Runtime` via `createSandboxRuntime` and a per-job `Resolver` via `createResolver`, and drives the per-job lifecycle through them
+- `packages/server/worker/src/lib/worker.ts` — worker lifecycle (`worker.start/stop`), `pollAndExecute` loop (with version gate, tracks `inFlightJobs`), `getWorkerProps`; builds the `Runtime` via `createSandboxRuntime` and a per-job `Resolver` via `createResolver`. `stop()` drains in-flight jobs (`drainInFlightJobs`) before teardown; an unexpected disconnect aborts the runtime (`abortInFlightRuntime`)
 - `packages/server/worker/src/lib/runtime/sandbox-config.ts` — bridges worker settings → `SandboxSettings` + cache base path (merges the env-only `AP_REUSE_SANDBOX` override into the fetched `WorkerSettings`)
 - `packages/server/sandbox/` — standalone `@activepieces/sandbox` library holding the in-process sandbox (`createSandboxRuntime`, `sandbox-manager`, isolate/fork process makers), cache/piece installation, the worker-side `createResolver`, and the `Runtime` / `Resolver` / `ProvisionInput` type contracts (`src/lib/types.ts`)
 - `packages/server/worker/src/lib/config/configs.ts` — worker env vars incl. `AP_REUSE_SANDBOX` (`WorkerSystemProp.REUSE_SANDBOX`, reuse the engine process between jobs) and `AP_WORKER_CONCURRENCY`
@@ -28,9 +29,10 @@ Workers are separate Node processes that poll the app for jobs and execute flows
 ## Connection & Poll Flow
 1. Worker connects → emits `FETCH_WORKER_SETTINGS`; app's `machineService.onConnection` returns `WorkerSettingsResponse` (incl. `APP_VERSION`) and registers `createHandlers` for the socket.
 2. Worker caches settings, **shuts down any old `Runtime` (fire-and-forget) and creates a fresh one**, then spawns `concurrency` `pollAndExecute` loops. Recreating per (re)connect means an in-flight job from a prior connection is killed with its box (it fails fast and BullMQ retries it) rather than lingering on a reused box — reusing a busy box let the new generation collide on the single-operation engine child and let the lingering job's lock lapse → BullMQ `Job stalled`.
-3. Each loop calls `apiClient.poll(machineInfo)`; the app's `poll` handler returns the next job for the worker's queue, or `null`.
-4. On job: worker executes in a sandbox, periodically `extendLock`, then `completeJob`.
-5. On disconnect, `connectionGeneration++` stops the loops; Socket.IO auto-reconnects (incl. a manual reconnect on `io server disconnect`, e.g. an app restart during a deploy) and the cycle repeats — recreating the runtime as in step 2.
+3. Each loop calls `apiClient.poll(machineInfo)`; the app's `poll` handler returns the next job for the worker's queue, or `null`. Handing out a job moves it to BullMQ `active`, so the handler **records the job under the polling connection's `socket.id`** (`jobAssignmentTracker`, keyed by queue+jobId) — only the app knows which connection holds it. (Scoped to `socket.id`, not the stable `workerId`, so a late disconnect of an old socket can't reclaim jobs a reconnected socket already polled.)
+4. On job: worker executes in a sandbox, periodically `extendLock`, then `completeJob`; `completeJob` **clears the assignment**.
+5. On disconnect, `connectionGeneration++` stops the loops; the app **returns that connection's still-held jobs to the queue** (`jobBroker.releaseConnectionJobs(socket.id)`, called from the `DISCONNECT` handler — scoped to the socket so a late disconnect can't reclaim a reconnected socket's jobs) so they don't sit orphaned in `active` past worker concurrency (the deploy "Job stalled" storm) waiting on the minutes-long stalled-scan; the worker **aborts its in-flight runtime** (`abortInFlightRuntime`) so it stops executing a job the app is reclaiming. Socket.IO auto-reconnects (incl. a manual reconnect on `io server disconnect`, e.g. an app restart during a deploy) and the cycle repeats — recreating the runtime as in step 2.
+6. On graceful shutdown (`stop()`, e.g. SIGTERM), the worker **drains in-flight jobs first** (`drainInFlightJobs`, ≤25s) so the executing job completes and reports before the runtime/socket are torn down — avoiding a kill+retry on a clean deploy. The stalled-scan remains the backstop for abrupt death (OOM/SIGKILL) where no drain is possible.
 
 > **Payload resolution is engine-side, not worker-side.** Jobs carry a `JobPayload` (`inline` value or `ref` `fileId`). The worker forwards it unchanged into the engine operation; the engine hydrates a `ref` via the file-download path (direct bytes or an S3 signed-link redirect). There is no worker→API payload-fetch RPC — the contract exposes no `getPayloadFile`.
 

packages/server/api/src/app/core/websockets.service.ts

@@ -59,17 +59,10 @@ export const websocketService = {
             }
         }
         for (const [event, handler] of Object.entries(listener[castedType])) {
-            socket.on(event, async (data, callback) => rejectedPromiseHandler(handler(socket)(data, principal, projectId, callback), log))
-        }
-        for (const handler of Object.values(listener[castedType][WebsocketServerEvent.CONNECT] ?? {})) {
-            handler(socket)
-        }
-    },
-    async onDisconnect(socket: Socket): Promise<void> {
-        const principal = await websocketService.verifyPrincipal(socket)
-        const castedType = principal.type as keyof typeof listener
-        for (const handler of Object.values(listener[castedType][WebsocketServerEvent.DISCONNECT] ?? {})) {
-            handler(socket)
+            // Socket.IO emits the reserved lowercase 'disconnect' per-socket; map our DISCONNECT enum
+            // onto it or the handler never fires (it never did — worker cleanup relied on the 60s sweep).
+            const socketEvent = event === WebsocketServerEvent.DISCONNECT ? 'disconnect' : event
+            socket.on(socketEvent, async (data, callback) => rejectedPromiseHandler(handler(socket)(data, principal, projectId, callback), log))
         }
     },
     async verifyPrincipal(socket: Socket): Promise<Principal> {

packages/server/api/src/app/flows/flow-run/engine-run-callback-service.ts

@@ -28,12 +28,14 @@ export const engineRunCallbackService = (log: FastifyBaseLogger) => ({
 
     async uploadRunLog({ projectId, request }: UploadRunLogParams): Promise<void> {
         const internalErrorEnabled = request.internalError?.source === RunInternalErrorSource.ENGINE || system.getEdition() !== ApEdition.CLOUD
-        if (internalErrorEnabled && !isNil(request.internalError) && !isNil(request.logsFileId)) {
-            await persistInternalErrorToLogs({
+        const internalError = internalErrorEnabled ? request.internalError : undefined
+        const isTerminal = isFlowRunStateTerminal({ status: request.status, ignoreInternalError: false })
+        if (isTerminal && !isNil(request.logsFileId)) {
+            await ensureLogsFileExists({
                 log,
                 projectId,
                 logsFileId: request.logsFileId,
-                internalError: request.internalError,
+                internalError,
             })
         }
         const logData: RunsMetadataUpsertData = {
@@ -66,19 +68,26 @@ export const engineRunCallbackService = (log: FastifyBaseLogger) => ({
     },
 })
 
-async function persistInternalErrorToLogs({ log, projectId, logsFileId, internalError }: PersistInternalErrorParams): Promise<void> {
+async function ensureLogsFileExists({ log, projectId, logsFileId, internalError }: EnsureLogsFileParams): Promise<void> {
     const { error } = await tryCatch(async () => {
-        const existing = await fileService(log).getDataOrUndefined({
+        const fileExists = await fileService(log).exists({
             projectId,
             fileId: logsFileId,
             type: FileType.FLOW_RUN_LOG,
         })
+        if (fileExists && isNil(internalError)) {
+            return
+        }
+
+        const existing = fileExists
+            ? await fileService(log).getDataOrUndefined({ projectId, fileId: logsFileId, type: FileType.FLOW_RUN_LOG })
+            : undefined
         const outputFile: ExecutioOutputFile = !isNil(existing)
             ? JSON.parse(existing.data.toString('utf-8'))
             : { executionState: { steps: {}, tags: [] } }
 
         const data = await fileCompressor.compress({
-            data: await logSerializer.serialize({ ...outputFile, internalError }),
+            data: await logSerializer.serialize(isNil(internalError) ? outputFile : { ...outputFile, internalError }),
             compression: FileCompression.ZSTD,
         })
 
@@ -95,7 +104,7 @@ async function persistInternalErrorToLogs({ log, projectId, logsFileId, internal
     })
 
     if (error) {
-        log.error({ error, logsFileId, project: { id: projectId } }, '[uploadRunLog] Failed to persist internal error to logs file')
+        log.error({ error, logsFileId, project: { id: projectId } }, '[uploadRunLog] Failed to ensure logs file exists')
     }
 }
 
@@ -118,9 +127,9 @@ type UploadRunLogParams = {
     request: UploadRunLogsRequest
 }
 
-type PersistInternalErrorParams = {
+type EnsureLogsFileParams = {
     log: FastifyBaseLogger
     projectId: string
     logsFileId: string
-    internalError: RunInternalError
+    internalError?: RunInternalError
 }

packages/server/api/src/app/helper/system-validator.ts

@@ -127,6 +127,7 @@ const systemPropValidators: {
     [AppSystemProp.REDIS_SENTINEL_ROLE]: stringValidator,
     [AppSystemProp.REDIS_SENTINEL_HOSTS]: stringValidator,
     [AppSystemProp.REDIS_SENTINEL_NAME]: stringValidator,
+    [AppSystemProp.USE_CDN_FOR_BUNDLES]: booleanValidator,
     [AppSystemProp.S3_ACCESS_KEY_ID]: stringValidator,
     [AppSystemProp.S3_BUCKET]: stringValidator,
     [AppSystemProp.S3_ENDPOINT]: stringValidator,

packages/server/api/src/app/helper/system/system-props.ts

@@ -119,6 +119,7 @@ export enum AppSystemProp {
     TRIGGER_DEFAULT_POLL_INTERVAL = 'TRIGGER_DEFAULT_POLL_INTERVAL',
     TRIGGER_HOOKS_TIMEOUT_SECONDS = 'TRIGGER_HOOKS_TIMEOUT_SECONDS',
     TRIGGER_TIMEOUT_SECONDS = 'TRIGGER_TIMEOUT_SECONDS',
+    USE_CDN_FOR_BUNDLES = 'USE_CDN_FOR_BUNDLES',
     WEBHOOK_TIMEOUT_SECONDS = 'WEBHOOK_TIMEOUT_SECONDS',
     OPENROUTER_PROVISION_KEY = 'OPENROUTER_PROVISION_KEY',
     EVENT_DESTINATION_TIMEOUT_SECONDS = 'EVENT_DESTINATION_TIMEOUT_SECONDS',

packages/server/api/src/app/helper/system/system.ts

@@ -26,6 +26,7 @@ const systemPropDefaultValues: Partial<Record<SystemProp, string>> = {
     [AppSystemProp.EXECUTION_DATA_RETENTION_DAYS]: '30',
     [AppSystemProp.PAUSED_FLOW_TIMEOUT_DAYS]: '30',
     [AppSystemProp.PIECES_SYNC_MODE]: PieceSyncMode.OFFICIAL_AUTO,
+    [AppSystemProp.USE_CDN_FOR_BUNDLES]: 'false',
     [AppSystemProp.ENVIRONMENT]: 'prod',
     [AppSystemProp.EXECUTION_MODE]: ExecutionMode.UNSANDBOXED,
     [AppSystemProp.WEBHOOK_TIMEOUT_SECONDS]: '30',

packages/server/api/src/app/pieces/piece-bundle.ts

@@ -1,6 +1,6 @@
 import { isNil, tryCatch } from '@activepieces/core-utils'
 import { apDayjs, safeHttp } from '@activepieces/server-utils'
-import { FileType, PackageType } from '@activepieces/shared'
+import { FileType, PackageType, PieceType } from '@activepieces/shared'
 import { FastifyBaseLogger } from 'fastify'
 import { fileRepo } from '../file/file.service'
 import { s3Helper } from '../file/s3-helper'
@@ -44,6 +44,13 @@ export const pieceBundle = (log: FastifyBaseLogger) => ({
             }
             void tryCatch(() => enqueueBundleJob({ name, version, log }))
         }
+        // CDN only mirrors official pieces — dev/custom/private registry pieces may 404 there, so fall back to npm.
+        if (metadata.pieceType === PieceType.OFFICIAL && system.getBoolean(AppSystemProp.USE_CDN_FOR_BUNDLES)) {
+            const cdnUrl = cdnTarballUrl({ name, version })
+            if (await cdnBundleExists({ url: cdnUrl, log })) {
+                return { type: 'redirect', url: cdnUrl }
+            }
+        }
         return { type: 'redirect', url: npmTarballUrl({ name, version }) }
     },
     registerJobHandler(): void {
@@ -71,6 +78,29 @@ async function enqueueBundleJob({ name, version, log }: EnqueueBundleJobParams):
     })
 }
 
+function cdnTarballUrl({ name, version }: PieceRef): string {
+    return `${CDN_PIECES_URL}${name.replace('/', '-')}-${version}.tgz`
+}
+
+// Piece tarballs are immutable per (name, version), so a positive result is cached forever.
+async function cdnBundleExists({ url, log }: CdnBundleExistsParams): Promise<boolean> {
+    if (cdnVerifiedUrls.has(url)) {
+        return true
+    }
+    const { data: response, error } = await tryCatch(() =>
+        safeHttp.axios.head(url, { validateStatus: (status) => status < 500 }),
+    )
+    if (error !== null) {
+        log.warn({ error, url }, '[pieceBundle] CDN bundle HEAD check failed, falling back to npm')
+        return false
+    }
+    const exists = response.status >= 200 && response.status < 300
+    if (exists) {
+        cdnVerifiedUrls.add(url)
+    }
+    return exists
+}
+
 function npmTarballUrl({ name, version }: PieceRef): string {
     const unscopedName = name.startsWith('@') ? name.split('/')[1] : name
     return `${NPM_REGISTRY_URL}/${name}/-/${unscopedName}-${version}.tgz`
@@ -80,7 +110,10 @@ function pieceBundleS3Key({ name, version }: PieceRef): string {
     return `${S3_PIECES_PREFIX}${name.replace('/', '-')}-${version}.tgz`
 }
 
+const cdnVerifiedUrls = new Set<string>()
+
 const NPM_REGISTRY_URL = 'https://registry.npmjs.org'
+const CDN_PIECES_URL = 'https://cdn.activepieces.com/pieces/retro/'
 const S3_PIECES_PREFIX = 'pieces/'
 
 type PieceRef = {
@@ -92,6 +125,11 @@ type EnqueueBundleJobParams = PieceRef & {
     log: FastifyBaseLogger
 }
 
+type CdnBundleExistsParams = {
+    url: string
+    log: FastifyBaseLogger
+}
+
 type ResolveParams = {
     name?: string
     version?: string

packages/server/api/src/app/server.ts

@@ -65,7 +65,6 @@ export const setupServer = async (): Promise<FastifyInstance> => {
                 .catch(() => next(new Error('Authentication error')))
         })
         app.io.on('connection', (socket: Socket) => rejectedPromiseHandler(websocketService.init(socket, app!.log), app!.log))
-        app.io.on('disconnect', (socket: Socket) => rejectedPromiseHandler(websocketService.onDisconnect(socket), app!.log))
     }
 
     if (system.isApp()) {

packages/server/api/src/app/workers/job-queue/job-assignment-tracker.ts

@@ -0,0 +1,79 @@
+import { isNil } from '@activepieces/core-utils'
+
+// Tracks which in-flight job each worker CONNECTION holds. The app moves a job to BullMQ `active`
+// when a worker polls — BEFORE the worker has received/started it — so a job orphaned by a
+// disconnect can be sitting in `active` that the worker itself never saw and could never drain.
+// Only the app knows it dispatched that job, so on disconnect the broker reads that connection's
+// jobs back from here and returns them to the queue, instead of leaving them in `active` for the
+// minutes-long stalled-scan (which is what inflated active past concurrency during deploys).
+// A connection's poll / completeJob / disconnect all land on the same app instance, so this
+// in-memory map is authoritative for that instance; the stalled-scan backstops an app-instance crash.
+//
+// Keyed by the per-connection socket id, NOT the stable workerId: a worker keeps the same workerId
+// across Socket.IO reconnects, so a late `disconnect` for an old socket — fired after the worker has
+// already reconnected and polled fresh jobs — would otherwise reclaim those fresh jobs out from under
+// the live connection. Scoping to the socket id means an old socket's disconnect only reclaims its
+// own jobs. Assignments themselves are keyed by (queueName, jobId) so the same job id on the shared
+// and a worker-group queue can't clobber each other's token.
+
+const assignmentByKey = new Map<string, JobAssignment>()
+const keysByConnectionId = new Map<string, Set<string>>()
+
+export const jobAssignmentTracker = {
+    record(params: { connectionId: string, jobId: string, token: string, queueName: string }): void {
+        const { connectionId, jobId, token, queueName } = params
+        const key = keyOf(queueName, jobId)
+        assignmentByKey.set(key, { connectionId, jobId, token, queueName })
+        const existing = keysByConnectionId.get(connectionId)
+        if (existing) {
+            existing.add(key)
+        }
+        else {
+            keysByConnectionId.set(connectionId, new Set([key]))
+        }
+    },
+    clear(params: { jobId: string, queueName: string }): void {
+        const key = keyOf(params.queueName, params.jobId)
+        const assignment = assignmentByKey.get(key)
+        if (isNil(assignment)) {
+            return
+        }
+        assignmentByKey.delete(key)
+        const connectionKeys = keysByConnectionId.get(assignment.connectionId)
+        if (isNil(connectionKeys)) {
+            return
+        }
+        connectionKeys.delete(key)
+        if (connectionKeys.size === 0) {
+            keysByConnectionId.delete(assignment.connectionId)
+        }
+    },
+    takeByConnection(connectionId: string): TakenAssignment[] {
+        const keys = keysByConnectionId.get(connectionId)
+        if (isNil(keys)) {
+            return []
+        }
+        keysByConnectionId.delete(connectionId)
+        const taken: TakenAssignment[] = []
+        for (const key of keys) {
+            const assignment = assignmentByKey.get(key)
+            if (isNil(assignment)) {
+                continue
+            }
+            assignmentByKey.delete(key)
+            taken.push({ jobId: assignment.jobId, token: assignment.token, queueName: assignment.queueName })
+        }
+        return taken
+    },
+    reset(): void {
+        assignmentByKey.clear()
+        keysByConnectionId.clear()
+    },
+}
+
+function keyOf(queueName: string, jobId: string): string {
+    return `${queueName} ${jobId}`
+}
+
+type JobAssignment = { connectionId: string, jobId: string, token: string, queueName: string }
+type TakenAssignment = { jobId: string, token: string, queueName: string }