From 867396b809bd1b16dc91c7ed7d84dd923472f58e Mon Sep 17 00:00:00 2001 From: Mo AbuAboud Date: Fri, 21 Aug 2026 20:15:27 +0200 Subject: [PATCH] fix(flows): stop large flows from pegging app CPU and tripping healthchecks (#14984) --- .agents/skills/profile-app-cpu/SKILL.md | 127 ++++++++++++++++++ .../engineering/ci-pr-review-hygiene.md | 1 + brain/knowledge/flows-execution/flows.md | 1 + .../flow-version/flow-version.service.ts | 19 +-- ...remove-connections-and-sample-data.test.ts | 109 +++++++++++++++ 5 files changed, 249 insertions(+), 8 deletions(-) create mode 100644 .agents/skills/profile-app-cpu/SKILL.md create mode 100644 packages/server/api/test/unit/app/flows/flow-version/remove-connections-and-sample-data.test.ts diff --git a/.agents/skills/profile-app-cpu/SKILL.md b/.agents/skills/profile-app-cpu/SKILL.md new file mode 100644 index 000000000000..1344789eaecd --- /dev/null +++ b/.agents/skills/profile-app-cpu/SKILL.md @@ -0,0 +1,127 @@ +--- +name: profile-app-cpu +description: "Find what is burning CPU in a live app container without restarting it: separate the JS thread from GC and libuv, open the V8 inspector in place with SIGUSR1, take a CPU profile over CDP, and aggregate it back to the function that owns the time. Use when containers sit at their CPU cap, healthchecks time out with nothing crashed, or a self-hoster reports the app 'unhealthy' but the process is alive." +--- + +# Profile a CPU-Bound App Container + +The sibling of `profile-worker-memory`. That one answers *what is holding memory*; this one answers *what is burning CPU*, and it does it on a running production container without a restart — a restart destroys the very state you are trying to measure. + +**A profile stays on the box until you have looked at it.** A `.cpuprofile` embeds function names, file paths and script snippets. Move the aggregated hot-frame list, not the raw file, off a customer's host. + +## 1. Rule out the database before you profile + +An app waiting on Postgres is *idle*, not busy. If containers are actually pegged, the DB is almost never the cause — but check first. Take the shape of the connection pool, including how many sessions are blocked on the *client*: + +```sql +select count(*) total, + count(*) filter (where state='active') active, + count(*) filter (where state='idle in transaction') idle_in_txn, + count(*) filter (where wait_event_type='Lock') waiting_on_lock, + count(*) filter (where wait_event='ClientRead') waiting_on_client +from pg_stat_activity; +``` + +Then look at the individual sessions, because the wait event per session is what actually decides this: + +```sql +select now() - query_start as dur, state, wait_event_type, wait_event, left(query, 120) as q +from pg_stat_activity +where state <> 'idle' and query_start is not null +order by 1 desc +limit 15; +``` + +The tell that the **app** is the bottleneck: sessions sitting in `wait_event=ClientRead` for seconds at a time, with `waiting_on_lock` at zero. `ClientRead` means Postgres has done its work and is waiting for the client to read the results — the event loop is blocked, and the DB is a victim rather than a cause. A genuinely struggling database looks the opposite: non-zero lock waits, or long `dur` on sessions whose `wait_event_type` is `IO` or `LWLock`. + +## 2. Find which thread is hot + +`docker stats` gives a percentage where 100% = one core, so a container capped at `cpus: 1` reads ~100% when saturated. Break it down per thread: + +```bash +PID=$(docker inspect -f '{{.State.Pid}}' ) +top -H -b -n 1 -p $PID +``` + +Read the `TIME+` column, not just `%CPU`: + +- **`MainThread` hot** → JS on the event loop. Continue to step 3. +- **`V8Worker` threads hot** → GC. Look at heap pressure, not application code. +- **`libuv-worker` hot** → fs/crypto/zlib in the threadpool. + +Compare `TIME+` against container age. 105 CPU-minutes over 27 hours is ~6% average — that is **bursty**, not a steady spin, and it means you must sample while it is actually hot. + +## 3. Open the inspector in place + +Node starts the inspector on `SIGUSR1` without restarting. It binds to `127.0.0.1:9229` **inside the container's network namespace**, so it is not reachable from outside the host. + +```bash +docker exec sh -c 'kill -USR1 1' +docker exec node -e 'require("http").get({host:"127.0.0.1",port:9229,path:"/json/version"},r=>r.pipe(process.stdout))' +``` + +Two caveats worth knowing before you do this on production: + +- The inspector cannot be closed again; it lives until the process exits. Prefer a container you can recycle afterwards, and say so in the incident notes. +- The `Profiler` domain only *samples*. It does not pause the process the way a breakpoint would. + +## 4. Take the profile + +Node 22+ ships a global `WebSocket`, so the CDP client needs no dependency. Run it *inside* the container so it shares the network namespace: + +```js +const http = require('http'), fs = require('fs') +const getWs = () => new Promise((res, rej) => + http.get({ host: '127.0.0.1', port: 9229, path: '/json/list' }, r => { + let d = ''; r.on('data', c => d += c) + r.on('end', () => { try { res(JSON.parse(d)[0].webSocketDebuggerUrl) } catch (e) { rej(new Error(d)) } }) + }).on('error', rej)) + +;(async () => { + const ws = new WebSocket(await getWs()) + let id = 0; const pending = new Map() + await new Promise((r, j) => { ws.onopen = r; ws.onerror = j }) + ws.onmessage = ev => { const m = JSON.parse(ev.data); if (pending.has(m.id)) { pending.get(m.id)(m.result); pending.delete(m.id) } } + const send = (method, params = {}) => new Promise(r => { const i = ++id; pending.set(i, r); ws.send(JSON.stringify({ id: i, method, params })) }) + await send('Profiler.enable') + await send('Profiler.setSamplingInterval', { interval: 400 }) + await send('Profiler.start') + await new Promise(r => setTimeout(r, 30000)) + const { profile } = await send('Profiler.stop') + fs.writeFileSync('/tmp/cpu.cpuprofile', JSON.stringify(profile)) + console.log('samples=' + profile.samples.length) + process.exit(0) +})() +``` + +`docker cp` it in, run it with `node`, `docker cp` the result out. 30s at a 400µs interval is ~50k samples — plenty, and light enough not to distort the result. + +## 5. Aggregate by function, not by stack node + +This is the step people skip, and it is why profiles get misread. V8 emits a **separate node per call stack**, so one hot function shows up dozens of times, each with a small percentage, and none of them look significant. Sum self-time by `(functionName, url, line)` first: + +```js +const p = JSON.parse(require('fs').readFileSync(process.argv[2], 'utf8')) +const byId = new Map(p.nodes.map(n => [n.id, n])) +const self = new Map() +for (const s of p.samples) self.set(s, (self.get(s) || 0) + 1) +const agg = new Map() +for (const [id, c] of self) { + const f = byId.get(id).callFrame + const k = `${f.functionName || '(anon)'} @ ${f.url}:${f.lineNumber + 1}` + agg.set(k, (agg.get(k) || 0) + c) +} +const total = p.samples.length +;[...agg.entries()].sort((a, b) => b[1] - a[1]).slice(0, 15) + .forEach(([k, c]) => console.log((100 * c / total).toFixed(2).padStart(6) + '% ' + k)) +``` + +Read the result against `(idle)`: at 49% idle, a frame at 42% of wall-clock owns ~84% of the CPU actually being spent. Report both numbers — "42% of wall-clock, 84% of non-idle" is the sentence that makes the finding land. + +To spot **runaway recursion**, walk each sampled stack and count repeats of one function. A long chain of the same frame with a slowly decaying percentage (22% → 19% over 30 frames) is recursion whose per-level cost is proportional to what remains below it — the signature of an accidental O(N²). + +## Gotchas + +- **Healthcheck timeouts are a symptom of a blocked event loop, not a crash.** A container shows `unhealthy` with `FailingStreak` climbing while the app still serves traffic; the `curl` healthcheck simply cannot be answered within its timeout. Zombie `curl` processes accumulating inside the container (`ps -eo stat | grep ^Z`) are those killed healthchecks, not a leak. Which container looks unhealthy rotates with traffic, so do not over-index on the one the alert names. +- **`docker stats` CPU is per-core, not per-host.** 100% means one full core. Against `cpus: 1` that is the cap, even though the host shows plenty of idle. +- Bursty load means an instantaneous `docker stats` disagrees with `TIME+` and with load average. Pick your profiling target from a fresh `docker stats` reading taken seconds before you attach. diff --git a/brain/knowledge/engineering/ci-pr-review-hygiene.md b/brain/knowledge/engineering/ci-pr-review-hygiene.md index 724d63efdf58..4e92cbaff600 100644 --- a/brain/knowledge/engineering/ci-pr-review-hygiene.md +++ b/brain/knowledge/engineering/ci-pr-review-hygiene.md @@ -29,5 +29,6 @@ Enforcement is the **`Codeowners review` repository ruleset** (active on the def - **A decision authored on a long-lived branch will collide on its number.** `brain/decisions/` numbers are assigned once and never reused, but the next free number is only knowable against `main` — two branches in flight both grab it. #14593 carried a `000024` that `main` had since filled, and `000025` too, so it landed as `000026`. Renumber against `main` at merge time and update every referring link; nothing in CI catches a duplicate number or a dead decision link. - **Preview environments resurrect on PR close because `setup-environment.yml` also triggers on `closed`.** Both workflows fire on the same close event; Remove Environment tears the env down correctly (compose down, nginx, repo), then Setup Environment sees the `preview` label (labels survive merge) and re-provisions the whole thing minutes later — verified on #14832: remove finished 11:20, setup rebuilt it by 11:30. This is why merged PRs kept live zombie environments on the preview box. Both workflows are thin SSH wrappers; the real setup/remove logic lives in `/root/environments` on the preview server (`secrets.PREVIEW_HOST`), not in this repo. Fixed by dropping `closed` from setup's trigger list. - **The preview-server remove tool can't clean containers once the repo dir is gone.** Its `stop()` skips `docker compose down` when `repos//docker-compose.yml` doesn't exist, so an env whose repo folder was deleted first leaves containers running forever — re-running `remove` is a no-op for them. Clean those manually via compose labels: `docker ps -aq --filter "label=com.docker.compose.project="` (same filter works for `docker volume ls`). When auditing envs against PR state: read the real branch from the clone's HEAD (`git -C repos/ symbolic-ref --short HEAD`) since subdomains flatten `/` to `-`; a clone sitting on `main` means the branch was deleted after merge; and an env with **no PR at all** is a manual `workflow_dispatch` preview — don't auto-delete those (bulk cleanup 2026-08-20 removed 27 closed-PR envs, reclaimed 32.5GB). +- **A unit test added under `packages/server/api/test/unit/` never runs in CI.** `ci.yml` runs exactly two test commands: `turbo run test` filtered to engine/shared/sandbox/ai-providers/pieces-framework/web, and `turbo run test-ce test-ee test-cloud check-migrations --filter=api`. The api package *has* a `test-unit` script (`vitest run test/unit`), but no workflow invokes it and the root `test-unit` filter list does not include api — so the 10+ files already sitting in `test/unit/**` are dead weight, and a new one passes review while protecting nothing. `packages/core/execution` is in the same position. Until the wiring changes, put api coverage that must actually gate merges in `test/integration/ce|ee|cloud`, and if you do add a unit test, say in the PR that you ran it locally and paste the result. - **`tools/scripts/` is outside the lint and test wiring.** ESLint ignores it, and `npm run test-unit` only covers engine/shared/web. A script there with real policy logic must run its own tests from its own workflow — `pr-size.yml` runs `bun test tools/scripts/pr-size-check.test.ts` as a step before the check itself. - **A branch that predates the `brain/` → `brain/knowledge/` move cannot edit a brain page in place — GitHub will call the PR conflicting even when `git merge` is clean locally.** Git follows the rename and merges the modification into the new path; GitHub's mergeability check does not, so it reports `modify/delete` on the old path and the PR goes `dirty`. Local `git merge-tree --write-tree` exits 0 and hides the problem; reproduce what GitHub sees with `git merge -X no-renames origin/main`. Fix: merge `origin/main` into the branch first, which lands the edit at the new path, then push. diff --git a/brain/knowledge/flows-execution/flows.md b/brain/knowledge/flows-execution/flows.md index 9b39ac85d103..3a984bc120bf 100644 --- a/brain/knowledge/flows-execution/flows.md +++ b/brain/knowledge/flows-execution/flows.md @@ -35,6 +35,7 @@ Flows are the core automation primitive: a versioned directed graph of trigger + - **`transaction()` (`core/db/transaction.ts`) is a bare `dataSource.transaction()`** — it acquires a *new* connection, not a savepoint. Nesting it deadlocks, so check every caller before wrapping a service method that others may already call inside a transaction. - Step settings split a piece's props into an always-visible **essential** set and a collapsed **Advanced** section: a prop is Advanced only when it sets `advanced: true` (everything else — incl. `MARKDOWN`, tab/section group members, and checkbox reveal targets — stays essential). `propertyGroups` render as tabs, sectioned cards, or the "Add filter" builder. - **Flows stuck in `DELETING` keep eating the active-flow limit.** Deletion is a durable BullMQ system job (`delete-flow-`), not synchronous: `delete()` sets `operationStatus=DELETING` and enqueues, and the row plus `status=ENABLED` only go away when the job finishes. That job runs `sampleDataService.deleteForFlow`, whose `DELETE FROM file … metadata->>'flowId'=?` had no index — on the large prod `file` table it seq-scans, blows `statement_timeout`, exhausts its 2 attempts and lands **permanently** in the failed set. The flow is then hidden from the UI list (which filters `!=DELETING`) but still counted by the active-flows quota (`getUsage` counts `status=ENABLED`), so Publish silently shows the "Purchase Extra Active Flows" dialog instead of publishing — this is what breaks the `webhook-should-return-response` e2e monitor. Stuck flows are functionally dead (`preDelete` disables the trigger before the failing delete), so forcing their rows away is safe. Fixes on `fix/flow-delete-sample-data-timeout`: a partial expression index `idx_file_sample_data_flow_id` on `file (type, (metadata->>'flowId'))`, plus `operationStatus != DELETING` in the active-flow counts so the quota stops depending on delete-job success. +- **`transferFlow` already deep-clones the whole flow — a callback that clones `step` again is quadratic.** `flowStructureUtil.transferFlow` opens with `JSON.parse(JSON.stringify(flowVersion))`, so the callback is handed a private copy and can mutate in place. Cloning per step instead is O(N²), because a `step` carries `nextAction` (the entire rest of the chain) plus loop/router children: cloning step *i* copies the remaining `N-i` steps. Measured on prod app containers (CDP CPU profile, 2026-08-21): the callback at `flow-version.service.ts` was **42% of wall-clock / ~84% of non-idle CPU**, at `transferStep` recursion depth 255 ≈ 32k step serializations per call, plus the GC churn behind ~2.5 GB RSS. It ran on **every** `getFlowVersionOrThrow` — including the default `removeConnectionsName=false, removeSampleData=false`, where the callback does nothing but the cloning still happens. Symptom was containers pegged at their `cpus: 1` cap and the 5s healthcheck `curl` timing out, which reads as "app unhealthy" with nothing crashed (the leftover zombie `curl`s are those killed healthchecks). Same pattern at `ee/…/project-state/diff/flow-diff.service.ts` (colder path, untouched here). When you write a `transferFlow` callback, mutate and return `step` — don't re-clone it. ### Editions CE has full authoring/publishing/folders/forms. EE/Cloud add owner transfer, piece filtering, template sharing, and active-flow quota enforcement on publish/enable. diff --git a/packages/server/api/src/app/flows/flow-version/flow-version.service.ts b/packages/server/api/src/app/flows/flow-version/flow-version.service.ts index c439ce1b5a4e..fd9c6bc1bed6 100644 --- a/packages/server/api/src/app/flows/flow-version/flow-version.service.ts +++ b/packages/server/api/src/app/flows/flow-version/flow-version.service.ts @@ -288,16 +288,19 @@ export const flowVersionService = (log: FastifyBaseLogger) => ({ removeSampleData: boolean, ): FlowVersion { return flowStructureUtil.transferFlow(flowVersion, (step) => { - const clonedStep = JSON.parse(JSON.stringify(step)) - if (removeConnectionNames) { - clonedStep.settings.input = removeConnectionsFromInput(clonedStep.settings.input) + const settings = { ...step.settings } + if (removeConnectionNames && !isNil(settings.input)) { + settings.input = removeConnectionsFromInput(settings.input) } - if (removeSampleData && !isNil(clonedStep?.settings?.sampleData)) { - clonedStep.settings.sampleData.sampleDataFileId = undefined - clonedStep.settings.sampleData.sampleDataInputFileId = undefined - clonedStep.settings.sampleData.lastTestDate = undefined + if (removeSampleData && !isNil(settings.sampleData)) { + settings.sampleData = { + ...settings.sampleData, + sampleDataFileId: undefined, + sampleDataInputFileId: undefined, + lastTestDate: undefined, + } } - return clonedStep + return { ...step, settings } }) }, }) diff --git a/packages/server/api/test/unit/app/flows/flow-version/remove-connections-and-sample-data.test.ts b/packages/server/api/test/unit/app/flows/flow-version/remove-connections-and-sample-data.test.ts new file mode 100644 index 000000000000..1da382e55162 --- /dev/null +++ b/packages/server/api/test/unit/app/flows/flow-version/remove-connections-and-sample-data.test.ts @@ -0,0 +1,109 @@ +import { CodeAction, FlowAction, FlowActionType, flowStructureUtil, FlowTrigger, FlowTriggerType, FlowVersion, FlowVersionState } from '@activepieces/shared' +import { FastifyBaseLogger } from 'fastify' +import { describe, expect, it, vi } from 'vitest' +import { flowVersionService } from '../../../../../src/app/flows/flow-version/flow-version.service' + +const mockLog = { + info: vi.fn(), + error: vi.fn(), + warn: vi.fn(), + debug: vi.fn(), +} as unknown as FastifyBaseLogger + +function buildCodeAction(name: string, nextAction?: FlowAction): CodeAction { + return { + name, + type: FlowActionType.CODE, + valid: true, + displayName: name, + lastUpdatedDate: '2026-05-02T00:00:00.000Z', + settings: { + sourceCode: { code: '', packageJson: '{}' }, + input: { + url: 'https://example.com', + token: '{{connections.my_connection}}', + }, + sampleData: { + sampleDataFileId: 'file-1', + sampleDataInputFileId: 'file-2', + lastTestDate: '2026-05-02T00:00:00.000Z', + }, + errorHandlingOptions: { + continueOnFailure: { value: false }, + retryOnFailure: { value: false }, + }, + }, + nextAction, + } +} + +function buildChainedFlowVersion(stepCount: number): FlowVersion { + let nextAction: FlowAction | undefined = undefined + for (let i = stepCount; i >= 1; i--) { + nextAction = buildCodeAction(`step_${i}`, nextAction) + } + const trigger: FlowTrigger = { + name: 'trigger', + type: FlowTriggerType.EMPTY, + valid: false, + displayName: 'Select Trigger', + lastUpdatedDate: '2026-05-02T00:00:00.000Z', + settings: {}, + nextAction, + } + return { + id: 'flow-version-id', + created: '2026-05-02T00:00:00.000Z', + updated: '2026-05-02T00:00:00.000Z', + flowId: 'flow-id', + displayName: 'quadratic clone regression', + trigger, + valid: false, + state: FlowVersionState.DRAFT, + schemaVersion: '1', + connectionIds: [], + agentIds: [], + } +} + +function removeAll(flowVersion: FlowVersion): FlowVersion { + return flowVersionService(mockLog).removeConnectionsAndSampleDataFromFlowVersion(flowVersion, true, true) +} + +describe('removeConnectionsAndSampleDataFromFlowVersion', () => { + it('strips connection references and sample data from every step in the chain', () => { + const result = removeAll(buildChainedFlowVersion(25)) + + const codeSteps = flowStructureUtil.getAllSteps(result.trigger).filter((step) => step.type === FlowActionType.CODE) + expect(codeSteps).toHaveLength(25) + for (const step of codeSteps) { + expect(step.settings.input.token).toBeUndefined() + expect(step.settings.input.url).toBe('https://example.com') + expect(step.settings.sampleData?.sampleDataFileId).toBeUndefined() + expect(step.settings.sampleData?.sampleDataInputFileId).toBeUndefined() + expect(step.settings.sampleData?.lastTestDate).toBeUndefined() + } + }) + + it('leaves the input flow version untouched', () => { + const flowVersion = buildChainedFlowVersion(5) + const before = JSON.stringify(flowVersion) + + removeAll(flowVersion) + + expect(JSON.stringify(flowVersion)).toBe(before) + }) + + it('clones the flow a constant number of times regardless of step count', () => { + const measure = (stepCount: number): number => { + const flowVersion = buildChainedFlowVersion(stepCount) + const spy = vi.spyOn(JSON, 'stringify') + removeAll(flowVersion) + const calls = spy.mock.calls.length + spy.mockRestore() + return calls + } + + expect(measure(200)).toBe(measure(20)) + }) +})