diff --git a/.agents/features/workers.md b/.agents/features/workers.md index 99ef9817a0a9..1ff9869d3df3 100644 --- a/.agents/features/workers.md +++ b/.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`. diff --git a/packages/server/api/src/app/core/websockets.service.ts b/packages/server/api/src/app/core/websockets.service.ts index 9ed56f6cda63..c3f3d7c2e503 100644 --- a/packages/server/api/src/app/core/websockets.service.ts +++ b/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 { - 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 { diff --git a/packages/server/api/src/app/flows/flow-run/engine-run-callback-service.ts b/packages/server/api/src/app/flows/flow-run/engine-run-callback-service.ts index ea12f9f5f158..5c0d2b98e655 100644 --- a/packages/server/api/src/app/flows/flow-run/engine-run-callback-service.ts +++ b/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 { 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 { +async function ensureLogsFileExists({ log, projectId, logsFileId, internalError }: EnsureLogsFileParams): Promise { 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 } diff --git a/packages/server/api/src/app/helper/system-validator.ts b/packages/server/api/src/app/helper/system-validator.ts index 7c809c45bde5..0b2567500171 100644 --- a/packages/server/api/src/app/helper/system-validator.ts +++ b/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, diff --git a/packages/server/api/src/app/helper/system/system-props.ts b/packages/server/api/src/app/helper/system/system-props.ts index 724d9d177abe..e89bff2884e9 100644 --- a/packages/server/api/src/app/helper/system/system-props.ts +++ b/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', diff --git a/packages/server/api/src/app/helper/system/system.ts b/packages/server/api/src/app/helper/system/system.ts index 130cca76d352..ae887e985871 100644 --- a/packages/server/api/src/app/helper/system/system.ts +++ b/packages/server/api/src/app/helper/system/system.ts @@ -26,6 +26,7 @@ const systemPropDefaultValues: Partial> = { [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', diff --git a/packages/server/api/src/app/pieces/piece-bundle.ts b/packages/server/api/src/app/pieces/piece-bundle.ts index 21f6838c3569..d5b0554e1e5e 100644 --- a/packages/server/api/src/app/pieces/piece-bundle.ts +++ b/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 { + 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() + 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 diff --git a/packages/server/api/src/app/server.ts b/packages/server/api/src/app/server.ts index 047aec580306..cc32ea555651 100644 --- a/packages/server/api/src/app/server.ts +++ b/packages/server/api/src/app/server.ts @@ -65,7 +65,6 @@ export const setupServer = async (): Promise => { .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()) { diff --git a/packages/server/api/src/app/workers/job-queue/job-assignment-tracker.ts b/packages/server/api/src/app/workers/job-queue/job-assignment-tracker.ts new file mode 100644 index 000000000000..f88b6b390d85 --- /dev/null +++ b/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() +const keysByConnectionId = new Map>() + +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 } diff --git a/packages/server/api/src/app/workers/job-queue/job-broker.ts b/packages/server/api/src/app/workers/job-queue/job-broker.ts index 3710f115609c..dbf2ae51e2f3 100644 --- a/packages/server/api/src/app/workers/job-queue/job-broker.ts +++ b/packages/server/api/src/app/workers/job-queue/job-broker.ts @@ -9,6 +9,7 @@ import { QueueName } from '../job' import { jobMigrations } from '../migrations/job-data-migrations' import { rateLimiterInterceptor } from './interceptors/rate-limiter-interceptor' import { zombiePollingInterceptor } from './interceptors/zombie-polling-interceptor' +import { jobAssignmentTracker } from './job-assignment-tracker' import { InterceptorVerdict, JobInterceptor } from './job-interceptor' import { isUserInteractionJobData } from './job-queue' import { createQueueDispatcher, QueueDispatcher } from './queue-dispatcher' @@ -211,10 +212,28 @@ export const jobBroker = (log: FastifyBaseLogger) => ({ log.info('[jobBroker] Job broker initialized') }, - async poll(queueName: string = QueueName.WORKER_JOBS): Promise { + async poll(queueName: string = QueueName.WORKER_JOBS, connectionId?: string): Promise { const worker = await ensureBullMQWorker(queueName, log) const dispatcher = ensureDispatcher(queueName, worker, log) - return dispatcher.poll() + const job = await dispatcher.poll() + if (!isNil(job) && !isNil(connectionId)) { + jobAssignmentTracker.record({ connectionId, jobId: job.jobId, token: job.token, queueName: job.queueName }) + } + return job + }, + + async releaseConnectionJobs(connectionId: string): Promise { + const jobs = jobAssignmentTracker.takeByConnection(connectionId) + if (jobs.length === 0) { + return + } + log.info({ connection: { id: connectionId }, jobCount: jobs.length }, '[jobBroker] Worker connection closed with in-flight jobs — returning them to the queue') + for (const { jobId, token, queueName } of jobs) { + const { error } = await tryCatch(() => returnJobToQueue(jobId, token, queueName, log)) + if (error) { + log.error({ connection: { id: connectionId }, job: { id: jobId }, error: String(error) }, '[jobBroker] Failed to return in-flight job on disconnect — leaving for stalled-scan recovery') + } + } }, async completeJob(input: ConsumeJobResponse & { jobId: string, token: string, queueName: string }): Promise { @@ -261,6 +280,8 @@ export const jobBroker = (log: FastifyBaseLogger) => ({ } } + jobAssignmentTracker.clear({ jobId: input.jobId, queueName: input.queueName }) + const failed = input.status === EngineResponseStatus.INTERNAL_ERROR || !isNil(error) for (const interceptor of interceptors) { const { error: interceptorError } = await tryCatch(() => interceptor.onJobFinished({ jobId: input.jobId, jobData, failed, log })) diff --git a/packages/server/api/src/app/workers/machine/machine-controller.ts b/packages/server/api/src/app/workers/machine/machine-controller.ts index 4541737a7704..c28186a5f32f 100644 --- a/packages/server/api/src/app/workers/machine/machine-controller.ts +++ b/packages/server/api/src/app/workers/machine/machine-controller.ts @@ -3,6 +3,7 @@ import { FastifyPluginAsyncZod } from 'fastify-type-provider-zod' import { z } from 'zod' import { securityAccess } from '../../core/security/authorization/fastify-security' import { websocketService } from '../../core/websockets.service' +import { jobBroker } from '../job-queue/job-broker' import { jobQueue } from '../job-queue/job-queue' import { createHandlers } from '../rpc/worker-rpc-service' import { machineService } from './machine-service' @@ -15,15 +16,19 @@ export const workerMachineController: FastifyPluginAsyncZod = async (app) => { const workerGroupId = typeof rawWorkerGroupId === 'string' ? rawWorkerGroupId : undefined const response = await machineService(app.log).onConnection(request, workerGroupId) callback?.(response) - createRpcServer(socket, createHandlers(app.log, workerGroupId)) + createRpcServer(socket, createHandlers(app.log, workerGroupId, socket.id)) } }) websocketService.addListener(PrincipalType.WORKER, WebsocketServerEvent.DISCONNECT, (socket) => { return async (_request: unknown, _principal) => { - await machineService(app.log).onDisconnect({ - workerId: socket.handshake.auth.workerId, - }) + // Return jobs dispatched to THIS connection that it never reported done — they sit + // orphaned in BullMQ `active` otherwise (graceful drain can't reach a job the worker never + // received), which inflated active past concurrency during deploys. Scoped to socket.id, + // not the stable workerId: a late disconnect for an old socket must not reclaim the jobs a + // reconnected socket (same workerId) has already polled. + await jobBroker(app.log).releaseConnectionJobs(socket.id) + await machineService(app.log).onDisconnect({ workerId: socket.handshake.auth.workerId }) } }) diff --git a/packages/server/api/src/app/workers/rpc/worker-rpc-service.ts b/packages/server/api/src/app/workers/rpc/worker-rpc-service.ts index 3c3ed6161cdf..bcbde2e1c810 100644 --- a/packages/server/api/src/app/workers/rpc/worker-rpc-service.ts +++ b/packages/server/api/src/app/workers/rpc/worker-rpc-service.ts @@ -43,7 +43,7 @@ function pageOnceForUnreadableAppVersion(log: FastifyBaseLogger, appVersion: str }) } -export function createHandlers(log: FastifyBaseLogger, workerGroupId?: string): WorkerToApiContract { +export function createHandlers(log: FastifyBaseLogger, workerGroupId?: string, connectionId?: string): WorkerToApiContract { return { async poll(input) { log.info({ worker: { id: input.workerId }, workerGroupId }, '[workerRpc#poll] Poll request received') @@ -64,7 +64,7 @@ export function createHandlers(log: FastifyBaseLogger, workerGroupId?: string): return null } const pollQueueName = getPollQueueName(workerGroupId) - const job = await jobBroker(log).poll(pollQueueName) + const job = await jobBroker(log).poll(pollQueueName, connectionId) if (job) { log.info({ worker: { id: input.workerId }, job: { id: job.jobId, type: job.jobData.jobType } }, '[workerRpc#poll] Returning job to worker') } diff --git a/packages/server/api/test/unit/app/workers/job-queue/active-invariant.test.ts b/packages/server/api/test/unit/app/workers/job-queue/active-invariant.test.ts new file mode 100644 index 000000000000..ca866af0ecbe --- /dev/null +++ b/packages/server/api/test/unit/app/workers/job-queue/active-invariant.test.ts @@ -0,0 +1,180 @@ +import { ConsumeJobRequest } from '@activepieces/shared' +import { Worker as BullMQWorker, Job, Queue } from 'bullmq' +import { FastifyBaseLogger } from 'fastify' +import IORedis from 'ioredis' +import { afterAll, beforeAll, describe, expect, it } from 'vitest' +import { jobAssignmentTracker } from '../../../../../src/app/workers/job-queue/job-assignment-tracker' +import { createQueueDispatcher } from '../../../../../src/app/workers/job-queue/queue-dispatcher' + +/** + * Pins the production invariant: the BullMQ `active` list must never exceed the total number of + * live worker slots (Σ worker concurrency). + * + * The app moves a job wait -> active when a worker polls (getNextJob) — i.e. it owns the job the + * moment it's dispatched, before the worker has even started it. The job leaves active only when + * the worker reports completion or the slow stalled-scan reaps it (~minutes). If a worker goes + * away mid-flight (a deploy stops every worker at once), its dispatched jobs are orphaned in + * active while the worker reconnects and is handed fresh ones — active accumulates far past + * concurrency (the prod incident). + * + * The fix: the app tracks which jobs each worker holds and, on disconnect, returns them to the + * queue (jobAssignmentTracker + releaseConnectionJobs). These tests drive the REAL tracker + REAL + * dispatcher + REAL BullMQ against the local test redis, contrasting abandon-on-stop vs reclaim. + */ + +const REDIS_HOST = process.env.AP_REDIS_HOST ?? 'localhost' +const REDIS_PORT = Number(process.env.AP_REDIS_PORT ?? '6379') +const PREFIX = 'ap-active-invariant-test' + +const log: FastifyBaseLogger = { + debug: () => {}, info: () => {}, warn: () => {}, error: () => {}, + fatal: () => {}, trace: () => {}, child: () => log, silent: () => {}, level: 'info', +} as unknown as FastifyBaseLogger + +const mkConn = (): IORedis => new IORedis({ host: REDIS_HOST, port: REDIS_PORT, maxRetriesPerRequest: null }) + +// Keep the stalled-checker effectively disabled for the short test window so the only thing that +// can move a job out of active is an explicit complete/return — isolating the invariant to the +// abandon-vs-reclaim behavior rather than stalled-checker timing. +const WORKER_OPTS = { + concurrency: 500, + autorun: false as const, + lockDuration: 120_000, + stalledInterval: 120_000, + maxStalledCount: 3, + prefix: PREFIX, +} + +type SimResult = { maxActive: number, totalConcurrency: number, completed: number, stops: number } + +async function runSimulation(params: { queueName: string, workers: number, jobs: number, reclaimOnStop: boolean, durationMs: number }): Promise { + const { queueName, workers, jobs, reclaimOnStop, durationMs } = params + jobAssignmentTracker.reset() + const conn = mkConn() + const redis = mkConn() + const queue = new Queue(queueName, { connection: conn, prefix: PREFIX }) + + const bullWorker = new BullMQWorker(queueName, undefined, { ...WORKER_OPTS, connection: mkConn() }) + await bullWorker.waitUntilReady() + + // Faithful mini-tryDequeue: getNextJob (wait -> active) and hand back the token. + const dequeue = async (worker: BullMQWorker): Promise => { + const token = `token-${Date.now()}-${Math.random().toString(36).slice(2)}` + const job = await worker.getNextJob(token) + if (!job) return null + return { jobId: job.id!, jobData: {} as ConsumeJobRequest['jobData'], timeoutInSeconds: 600, attempsStarted: 0, engineToken: 'x', token, queueName } + } + const returnJobToQueue = async (jobId: string, token: string): Promise => { + const job = await Job.fromId(bullWorker, jobId) + if (job) await job.moveToDelayed(Date.now() + 1, token) + } + const completeJob = async (jobId: string, token: string): Promise => { + const job = await Job.fromId(bullWorker, jobId) + if (job) await job.moveToCompleted({ ok: true }, token, false) + } + + const dispatcher = createQueueDispatcher({ + queueName, + worker: bullWorker, + dequeue, + onOrphanedJob: (jobId, token) => returnJobToQueue(jobId, token), + log, + }) + + await queue.addBulk(Array.from({ length: jobs }, (_, i) => ({ name: `j${i}`, data: { i }, opts: { attempts: 5 } }))) + + let maxActive = 0 + let stop = false + const activeKey = `${PREFIX}:${queueName}:active` + const sampler = (async () => { + while (!stop) { + const n = await redis.llen(activeKey) + if (n > maxActive) maxActive = n + await sleep(10) + } + })() + + let completed = 0 + let stops = 0 + const deadline = Date.now() + durationMs + + // Each simulated worker = one box, concurrency 1: it holds at most one job at a time. With + // prob 0.5 it "stops" mid-job (a deploy/restart) instead of looping to the next. + const workerLoop = async (workerId: string, seed: number): Promise => { + let rng = seed + const rand = (): number => { + rng = (rng * 1103515245 + 12345) & 0x7fffffff + return rng / 0x7fffffff + } + while (Date.now() < deadline) { + const job = await dispatcher.poll() + if (!job) { + await sleep(2) + continue + } + // Mirrors the server: jobBroker.poll records the assignment for this worker. + jobAssignmentTracker.record({ connectionId: workerId, jobId: job.jobId, token: job.token, queueName: job.queueName }) + const stopMidJob = rand() < 0.5 + if (stopMidJob) { + stops++ + if (reclaimOnStop) { + // The fix: the disconnect handler calls releaseWorkerJobs, which reads this + // worker's jobs from the real tracker and returns each to the queue. + for (const held of jobAssignmentTracker.takeByConnection(workerId)) { + await returnJobToQueue(held.jobId, held.token) + } + } + // else: stop without reclaiming — job orphaned in `active` (the prod incident). + } + else { + await completeJob(job.jobId, job.token) + jobAssignmentTracker.clear({ jobId: job.jobId, queueName: job.queueName }) + completed++ + } + } + } + + await Promise.all(Array.from({ length: workers }, (_, i) => workerLoop(`w${i + 1}`, i + 1))) + stop = true + await sampler + + dispatcher.close() + await bullWorker.close() + await queue.close() + conn.disconnect() + redis.disconnect() + return { maxActive, totalConcurrency: workers, completed, stops } +} + +function sleep(ms: number): Promise { + return new Promise(r => setTimeout(r, ms)) +} + +describe('BullMQ active-list invariant: active <= total worker concurrency', () => { + beforeAll(async () => { + const r = mkConn() + const keys = await r.keys(`${PREFIX}:*`) + if (keys.length) await r.del(...keys) + r.disconnect() + }) + afterAll(async () => { + const r = mkConn() + const keys = await r.keys(`${PREFIX}:*`) + if (keys.length) await r.del(...keys) + r.disconnect() + }) + + it('REPRODUCES the bug: abandoning the worker\'s jobs on stop makes active far exceed concurrency', async () => { + const res = await runSimulation({ queueName: 'repro-bug', workers: 4, jobs: 5000, reclaimOnStop: false, durationMs: 5000 }) + // eslint-disable-next-line no-console + console.log('[abandon-on-stop] maxActive=%d totalConcurrency=%d completed=%d stops=%d', res.maxActive, res.totalConcurrency, res.completed, res.stops) + expect(res.maxActive).toBeGreaterThan(res.totalConcurrency) + }, 60_000) + + it('FIX: returning the worker\'s jobs on disconnect keeps active <= concurrency', async () => { + const res = await runSimulation({ queueName: 'repro-fix', workers: 4, jobs: 5000, reclaimOnStop: true, durationMs: 5000 }) + // eslint-disable-next-line no-console + console.log('[reclaim-on-stop] maxActive=%d totalConcurrency=%d completed=%d stops=%d', res.maxActive, res.totalConcurrency, res.completed, res.stops) + expect(res.maxActive).toBeLessThanOrEqual(res.totalConcurrency) + }, 60_000) +}) diff --git a/packages/server/api/test/unit/app/workers/job-queue/job-assignment-tracker.test.ts b/packages/server/api/test/unit/app/workers/job-queue/job-assignment-tracker.test.ts new file mode 100644 index 000000000000..0a75c666851b --- /dev/null +++ b/packages/server/api/test/unit/app/workers/job-queue/job-assignment-tracker.test.ts @@ -0,0 +1,76 @@ +import { beforeEach, describe, expect, it } from 'vitest' +import { jobAssignmentTracker } from '../../../../../src/app/workers/job-queue/job-assignment-tracker' + +describe('jobAssignmentTracker', () => { + beforeEach(() => { + jobAssignmentTracker.reset() + }) + + it('returns a recorded job for its connection on takeByConnection', () => { + jobAssignmentTracker.record({ connectionId: 'c1', jobId: 'j1', token: 't1', queueName: 'q' }) + + expect(jobAssignmentTracker.takeByConnection('c1')).toEqual([{ jobId: 'j1', token: 't1', queueName: 'q' }]) + }) + + it('groups multiple in-flight jobs under the same connection', () => { + jobAssignmentTracker.record({ connectionId: 'c1', jobId: 'j1', token: 't1', queueName: 'q' }) + jobAssignmentTracker.record({ connectionId: 'c1', jobId: 'j2', token: 't2', queueName: 'q' }) + + const taken = jobAssignmentTracker.takeByConnection('c1') + + expect(taken).toHaveLength(2) + expect(taken.map(t => t.jobId).sort()).toEqual(['j1', 'j2']) + }) + + it('does not return another connection\'s jobs', () => { + jobAssignmentTracker.record({ connectionId: 'c1', jobId: 'j1', token: 't1', queueName: 'q' }) + jobAssignmentTracker.record({ connectionId: 'c2', jobId: 'j2', token: 't2', queueName: 'q' }) + + expect(jobAssignmentTracker.takeByConnection('c1')).toEqual([{ jobId: 'j1', token: 't1', queueName: 'q' }]) + expect(jobAssignmentTracker.takeByConnection('c2')).toEqual([{ jobId: 'j2', token: 't2', queueName: 'q' }]) + }) + + it('a stale socket disconnect does NOT reclaim the reconnected socket\'s jobs (same worker, new connection)', () => { + // old socket c1 polled j1; worker reconnects as c2 and polls j2 before c1's disconnect fires. + jobAssignmentTracker.record({ connectionId: 'c1', jobId: 'j1', token: 't1', queueName: 'q' }) + jobAssignmentTracker.record({ connectionId: 'c2', jobId: 'j2', token: 't2', queueName: 'q' }) + + // c1's late disconnect reclaims only j1 — j2 stays with the live connection c2. + expect(jobAssignmentTracker.takeByConnection('c1')).toEqual([{ jobId: 'j1', token: 't1', queueName: 'q' }]) + expect(jobAssignmentTracker.takeByConnection('c2')).toEqual([{ jobId: 'j2', token: 't2', queueName: 'q' }]) + }) + + it('keeps the same job id in different queues separate (no token clobbering)', () => { + jobAssignmentTracker.record({ connectionId: 'c1', jobId: 'shared', token: 'tA', queueName: 'qA' }) + jobAssignmentTracker.record({ connectionId: 'c2', jobId: 'shared', token: 'tB', queueName: 'qB' }) + + jobAssignmentTracker.clear({ jobId: 'shared', queueName: 'qA' }) + + expect(jobAssignmentTracker.takeByConnection('c1')).toEqual([]) + expect(jobAssignmentTracker.takeByConnection('c2')).toEqual([{ jobId: 'shared', token: 'tB', queueName: 'qB' }]) + }) + + it('cleared (completed) jobs are not returned on disconnect', () => { + jobAssignmentTracker.record({ connectionId: 'c1', jobId: 'j1', token: 't1', queueName: 'q' }) + jobAssignmentTracker.record({ connectionId: 'c1', jobId: 'j2', token: 't2', queueName: 'q' }) + + jobAssignmentTracker.clear({ jobId: 'j1', queueName: 'q' }) + + expect(jobAssignmentTracker.takeByConnection('c1')).toEqual([{ jobId: 'j2', token: 't2', queueName: 'q' }]) + }) + + it('takeByConnection is idempotent — a second call returns nothing', () => { + jobAssignmentTracker.record({ connectionId: 'c1', jobId: 'j1', token: 't1', queueName: 'q' }) + + expect(jobAssignmentTracker.takeByConnection('c1')).toHaveLength(1) + expect(jobAssignmentTracker.takeByConnection('c1')).toEqual([]) + }) + + it('returns empty for an unknown connection', () => { + expect(jobAssignmentTracker.takeByConnection('nope')).toEqual([]) + }) + + it('clear on an unknown job is a no-op', () => { + expect(() => jobAssignmentTracker.clear({ jobId: 'ghost', queueName: 'q' })).not.toThrow() + }) +}) diff --git a/packages/server/sandbox/src/lib/sandbox/sandbox.ts b/packages/server/sandbox/src/lib/sandbox/sandbox.ts index eca5dedebb81..714c48e818f6 100644 --- a/packages/server/sandbox/src/lib/sandbox/sandbox.ts +++ b/packages/server/sandbox/src/lib/sandbox/sandbox.ts @@ -2,7 +2,7 @@ import { ChildProcess } from 'child_process' import { randomBytes, timingSafeEqual } from 'crypto' import { createServer, Server as HttpServer } from 'http' import path from 'path' -import { ActivepiecesError, assertNotNullOrUndefined, ErrorCode, isNil } from '@activepieces/core-utils' +import { ActivepiecesError, assertNotNullOrUndefined, ErrorCode, isNil, tryCatch } from '@activepieces/core-utils' import { createNotifyServer, createRpcClient, EngineContract, EngineOperation, EngineOperationType, EngineResponse, EngineStderr, EngineStdout, WorkerNotifyContract } from '@activepieces/shared' import { Socket, Server as SocketIOServer } from 'socket.io' import treeKill from 'tree-kill' @@ -62,7 +62,6 @@ export function createSandbox( processMaker: SandboxProcessMaker, ): Sandbox { let childProcess: ChildProcess | null = null - let httpServer: HttpServer | null = null let io: SocketIOServer | null = null let connectedSocket: Socket | null = null let connectionResolve: (() => void) | null = null @@ -70,17 +69,9 @@ export function createSandbox( let busy = false let killedByShutdown = false - function createSocketServer(): number { - httpServer = createServer() - io = new SocketIOServer(httpServer, { - path: '/worker/ws', - maxHttpBufferSize: options.maxHttpBufferSizeBytes, - cors: { origin: '*' }, - }) - - io.use(authenticateHandshake({ getExpectedToken: () => wsRpcToken, log, sandboxId })) - - io.on('connection', (socket) => { + function wireConnectionHandler(ioServer: SocketIOServer): void { + ioServer.use(authenticateHandshake({ getExpectedToken: () => wsRpcToken, log, sandboxId })) + ioServer.on('connection', (socket) => { if (!isNil(connectedSocket) && connectedSocket.connected) { log.warn({ sandbox: { id: sandboxId }, socketId: socket.id }, '[WebSocket] Rejecting extra connection — sandbox already has an active socket') socket.disconnect(true) @@ -101,14 +92,51 @@ export function createSandbox( connectionResolve = null } }) + } - httpServer.listen(options.wsRpcPort ?? 0) + // In isolate mode the ws port is fixed per box (WS_RPC_BASE_PORT + boxId). A reused box whose + // previous server hasn't finished releasing that port — or a brief double-allocation under high + // concurrency — makes the bind emit EADDRINUSE. Without an 'error' listener that event is an + // UNHANDLED exception that crashes the ENTIRE worker process (and every in-flight sandbox on it), + // which is exactly what made busy workers crash-loop. Await the bind, and on a bind error retry a + // few times (a concurrent close frees the port within a beat); if it never frees, fail just this + // sandbox with a catchable error. A random port (listen(0)) can't collide, so it needs no retries. + async function createSocketServer(): Promise { + const requestedPort = options.wsRpcPort ?? 0 + const maxAttempts = requestedPort === 0 ? 1 : 30 + for (let attempt = 1; attempt <= maxAttempts; attempt++) { + const server = createServer() + const ioServer = new SocketIOServer(server, { + path: '/worker/ws', + maxHttpBufferSize: options.maxHttpBufferSizeBytes, + cors: { origin: '*' }, + }) + wireConnectionHandler(ioServer) + + const { error } = await tryCatch(() => listenOnce(server, requestedPort)) + if (isNil(error)) { + io = ioServer + const address = server.address() + if (typeof address === 'object' && address !== null) { + return address.port + } + throw new Error('Could not determine socket.io server port') + } - const address = httpServer.address() - if (typeof address === 'object' && address !== null) { - return address.port + await tryCatch(() => closeServer(ioServer)) + if (attempt === maxAttempts) { + throw new ActivepiecesError({ + code: ErrorCode.SANDBOX_INTERNAL_ERROR, + params: { + reason: `Failed to bind sandbox ws port ${requestedPort} after ${maxAttempts} attempts: ${String(error)}`, + standardOutput: '', + standardError: '', + }, + }) + } + await delay(100) } - throw new Error('Could not determine socket.io server port') + throw new Error('Could not start sandbox socket server') } function waitForConnection(): Promise { @@ -145,7 +173,7 @@ export function createSandbox( }, 'Starting sandbox') wsRpcToken = randomBytes(32).toString('hex') - const port = createSocketServer() + const port = await createSocketServer() const codeMount = buildCodeMount({ flowVersionId, reusable: options.reusable, basePath: options.basePath }) const customPieceMounts: SandboxMount[] = [] @@ -297,12 +325,37 @@ export function createSandbox( await io.close() } io = null - httpServer = null wsRpcToken = null }, } } +function listenOnce(server: HttpServer, port: number): Promise { + return new Promise((resolve, reject) => { + const onError = (err: Error): void => { + server.removeListener('listening', onListening) + reject(err) + } + const onListening = (): void => { + server.removeListener('error', onError) + resolve() + } + server.once('error', onError) + server.once('listening', onListening) + server.listen(port) + }) +} + +function closeServer(ioServer: SocketIOServer): Promise { + return new Promise((resolve) => { + ioServer.close(() => resolve()) + }) +} + +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)) +} + function handleProcessExit(log: SandboxLogger, params: ProcessExitParams): void { const { sandboxId, operationType, code, signal, killedByTimeout, killedByShutdown, stdOut, stdError, reject } = params log.info({ diff --git a/packages/server/sandbox/test/lib/sandbox/sandbox.test.ts b/packages/server/sandbox/test/lib/sandbox/sandbox.test.ts index 862a5c0b1dd9..f5ea88351e61 100644 --- a/packages/server/sandbox/test/lib/sandbox/sandbox.test.ts +++ b/packages/server/sandbox/test/lib/sandbox/sandbox.test.ts @@ -317,6 +317,35 @@ describe('createSandbox', () => { expect(createCall.env.AP_CUSTOM_PIECES_PATHS).toBeUndefined() }) + it('does NOT crash the process when a fixed ws port is already bound — fails just that sandbox', async () => { + // Isolate mode pins a fixed ws port per box. Two boxes contending the same port must not + // let the EADDRINUSE 'error' event become an uncaught exception that kills the whole worker. + const log = createMockLogger() + const fixedPort = 53777 + const pmA = createTestProcessMaker() + const pmB = createTestProcessMaker() + const sandboxA = createSandbox(log, 'sb-port-a', { ...defaultOptions, wsRpcPort: fixedPort }, pmA.maker) + const sandboxB = createSandbox(log, 'sb-port-b', { ...defaultOptions, wsRpcPort: fixedPort }, pmB.maker) + + await sandboxA.start(startOptions) + + let code: ErrorCode | undefined + try { + await sandboxB.start(startOptions) + } + catch (err) { + code = (err as ActivepiecesError).error.code + } + // A catchable error, NOT an uncaught crash (the test process is still running). + expect(code).toBe(ErrorCode.SANDBOX_INTERNAL_ERROR) + // A keeps working on its port. + expect(pmA.getClient().connected).toBe(true) + + await sandboxA.shutdown() + const clientB = pmB.getClient() + if (clientB?.connected) clientB.disconnect() + }, 20_000) + it('is idempotent when already connected', async () => { const log = createMockLogger() testPM = createTestProcessMaker() diff --git a/packages/server/worker/src/lib/worker.ts b/packages/server/worker/src/lib/worker.ts index be0fb167c82f..bb00b453c217 100644 --- a/packages/server/worker/src/lib/worker.ts +++ b/packages/server/worker/src/lib/worker.ts @@ -47,6 +47,13 @@ let healthServerInstance: ReturnType | null = null let runtime: Runtime | null = null +// Jobs executing across all poll loops. stop() waits for these to finish + report before tearing +// down, so a deploy doesn't orphan them in the app's BullMQ `active` list. Abrupt death (OOM/SIGKILL) +// still falls to the stalled-scan. +let inFlightJobs = 0 + +const DRAIN_TIMEOUT_MS = 25_000 + export const worker = { async start({ apiUrl, socketUrl, workerToken, withHealthServer = false }: WorkerStartParams): Promise { const workerGroupId = system.get(WorkerSystemProp.WORKER_GROUP_ID) @@ -71,11 +78,17 @@ export const worker = { connectionGeneration++ polling = false logger.warn({ reason }, 'Disconnected from API server') + // 'io client disconnect' is our own stop() — it already drained and tore down the runtime. + // For any other reason the socket dropped while a job may still be running locally; the app + // reclaims that job on disconnect, so kill the runtime now or the original keeps executing + // to completion and double-runs the requeued copy. (The reconnect path recreates it.) + if (reason !== 'io client disconnect') { + abortInFlightRuntime() + } // Socket.IO does NOT auto-reconnect when the server initiates the disconnect // (reason 'io server disconnect' — e.g. the API process restarts/hot-reloads). // Without a manual reconnect the worker stays dead and silently stops consuming - // jobs. 'io client disconnect' is our own shutdown, so leave that alone; every - // other reason already triggers Socket.IO's built-in reconnection. + // jobs. Every other non-client reason already triggers built-in reconnection. if (reason === 'io server disconnect') { socket?.connect() } @@ -93,6 +106,7 @@ export const worker = { async stop(): Promise { polling = false + await drainInFlightJobs() if (runtime) { await runtime.shutdown(logger) runtime = null @@ -122,25 +136,13 @@ async function startPollingWorkers(apiClient: WorkerToApiContract): Promise oldRuntime.shutdown(logger)).then(({ error }) => { - if (error) { - logger.error({ error }, 'Failed to shut down old runtime on reconnect') - } - }) - } + // Bring up a fresh runtime on every (re)connect — a prior connection's in-flight job is killed + // along with its box (usually already done by the disconnect handler), so it fails fast and is + // retried instead of lingering on a reused box. Reusing the box made the next generation's poll + // loop run a second operation on the same single-operation engine child (acquire() hands back a + // busy sandbox) and let the lingering job's lock lapse during the reconnect → BullMQ "stalled"; + // a deploy disconnects every worker at once, so reuse turned that into mass stalls. + abortInFlightRuntime() runtime = createSandboxRuntime({ concurrency, @@ -198,41 +200,74 @@ async function pollAndExecute(apiClient: WorkerToApiContract, runtime: Runtime, workerLog.debug({ job: { id: job.jobId, type: job.jobData.jobType } }, 'Job received from poll') - const lockExtensionInterval = setInterval(() => { - void tryCatch(() => apiClient.extendLock({ jobId: job.jobId, token: job.token, queueName: job.queueName })).then(({ error }) => { - if (error) { - workerLog.warn({ error, job: { id: job.jobId } }, 'Failed to extend lock') - } - }) - }, 30_000) - - const { data: result, error: execError } = await tryCatch(() => - executeJob(apiClient, job, runtime, workerIndex), - ) + // Counted for the duration of execute + completeJob so a graceful shutdown can wait for + // the report to land before tearing down the socket (otherwise the job is orphaned in active). + inFlightJobs++ + try { + const lockExtensionInterval = setInterval(() => { + void tryCatch(() => apiClient.extendLock({ jobId: job.jobId, token: job.token, queueName: job.queueName })).then(({ error }) => { + if (error) { + workerLog.warn({ error, job: { id: job.jobId } }, 'Failed to extend lock') + } + }) + }, 30_000) + + const { data: result, error: execError } = await tryCatch(() => + executeJob(apiClient, job, runtime, workerIndex), + ) + + const { error: completeError } = await tryCatch(() => + apiClient.completeJob({ + jobId: job.jobId, + token: job.token, + queueName: job.queueName, + status: execError + ? EngineResponseStatus.INTERNAL_ERROR + : result.status, + errorMessage: buildErrorMessage(execError ?? undefined, result ?? undefined), + logs: extractLogs(execError ?? undefined, result ?? undefined), + response: result?.kind === JobResultKind.SYNCHRONOUS ? result.response : undefined, + }), + ) + clearInterval(lockExtensionInterval) - const { error: completeError } = await tryCatch(() => - apiClient.completeJob({ - jobId: job.jobId, - token: job.token, - queueName: job.queueName, - status: execError - ? EngineResponseStatus.INTERNAL_ERROR - : result.status, - errorMessage: buildErrorMessage(execError ?? undefined, result ?? undefined), - logs: extractLogs(execError ?? undefined, result ?? undefined), - response: result?.kind === JobResultKind.SYNCHRONOUS ? result.response : undefined, - }), - ) - - clearInterval(lockExtensionInterval) - - if (completeError) { - workerLog.error({ error: completeError, job: { id: job.jobId } }, 'Failed to complete job') + if (completeError) { + workerLog.error({ error: completeError, job: { id: job.jobId } }, 'Failed to complete job') + } + } + finally { + inFlightJobs-- } } } +async function drainInFlightJobs(): Promise { + const deadline = Date.now() + DRAIN_TIMEOUT_MS + while (inFlightJobs > 0 && Date.now() < deadline) { + await sleep(100) + } + if (inFlightJobs > 0) { + logger.warn({ inFlightJobs }, 'Drain timeout reached with jobs still in flight; leaving them for the stalled-scan to reclaim') + } +} + +// Kill the current runtime (and the job running on it) without awaiting — awaiting the shutdown +// while a job is in-flight deadlocks (the await never resolves, polling never restarts). Nulling +// `runtime` lets the next (re)connect create a fresh box. +function abortInFlightRuntime(): void { + if (isNil(runtime)) { + return + } + const oldRuntime = runtime + runtime = null + void tryCatch(() => oldRuntime.shutdown(logger)).then(({ error }) => { + if (error) { + logger.error({ error }, 'Failed to shut down runtime') + } + }) +} + async function executeJob(apiClient: WorkerToApiContract, job: ConsumeJobRequest, runtime: Runtime, workerIndex: number): Promise { const rawData = job.jobData const jobData = JobData.parse(rawData)