Skip to content

R2 Concurrency + correctness audit — 2026-05-31

Scope: every .py under services/wp-pulse-server/api/{routers,services,utils}/ touched in the last 2 days (same set as R1). Read-only review; no code modified.

Summary

  • Critical: 0
  • High: 5
  • Medium: 9
  • Low: 6
  • Info: 4

The most load-bearing pieces (FastAPI BackgroundTasks pattern, fleet job module-level lock, lighthouse semaphore, visual-diff strong-ref pool, version_diff temp cleanup, monthly_report idempotency, ssl_check / metrics_rollup ON-CONFLICT) are all wired correctly. Findings cluster around:

  1. A repeated Popen.communicate(timeout=…) recipe with no proc.kill() on TimeoutExpired — leaks docker-compose-exec children.
  2. A few subprocess.run / popen / asyncio operations missing a timeout.
  3. A handful of TOCTOU windows on "one active job per X" check-then-insert flows.
  4. Two except Exception: pass swallows that hide real failures and one that could let a docker stack stay up while a destructive rsync runs.

Findings

[HIGH] subprocess.Popen.communicate(timeout=) never kills the child on timeout (4 sites)

  • Files:
  • api/utils/versions.py:285-312 (mysqldump → gzip path)
  • api/utils/versions.py:637-672 (load_version mysql restore)
  • api/utils/version_preview.py:530-549 (time-travel preview mysql restore)
  • api/routers/pulls.py:1134-1162 (_restore_db_dump)
  • Class: subprocess lifecycle / resource leak
  • Detail: All four follow the same pattern: proc = subprocess.Popen(...) → stream stdin → proc.communicate(timeout=600/900) → except (OSError, subprocess.TimeoutExpired) as exc: return False. Python's docs explicitly state that on TimeoutExpired from communicate() the child is not reaped — you must call proc.kill() then proc.communicate() again. As written, a wedged docker compose exec (e.g. the db container is alive but mysql is hung) leaves an orphan child + open pipes for the duration of the FastAPI worker.
  • Recommendation: wrap each in try/except subprocess.TimeoutExpired: proc.kill(); proc.communicate() (or use proc.wait(timeout=...) inside a try/finally that always kills on failure). Module-level helper would deduplicate the 4 copies.

[HIGH] state_machine._promote_rehearsal_to_main continues after docker compose down fails

  • File: api/services/rehearsal/state_machine.py:357-384
  • Class: error masking + destructive op
  • Detail: subprocess.run(["docker", "compose", "down"], …, timeout=60) is wrapped in try: …; except Exception: pass. If the down command times out (60s ceiling is tight on a loaded box) or returns non-zero, the next step is rsync -a --delete from the rehearsal worktree into the main worktree — overwriting wp-content while the live php container may still be holding file handles. Worst-case: half-written PHP files become live, and the operator can't roll back without restoring from a snapshot. The bare except Exception: pass also masks the symptom from logs.
  • Recommendation: at minimum log the failure and bail if returncode != 0; ideally verify docker compose ps reports the stack down before running rsync, or wrap rsync in a snapshot-and-revert guard.

[HIGH] TOCTOU on "max-active previews" + port allocation

  • File: api/routers/version_previews.py:148-194 (callers) + api/utils/version_preview.py:107-132 (helpers)
  • Class: race / cap bypass
  • Detail: count_active(db) followed by allocate_port(db) then db.add(row) + commit(). Two concurrent POSTs to /preview can both see count == MAX-1 and both reserve the same lowest-free port; the second insert succeeds because there's no unique constraint on (port, status='running'). End state: two previews sharing port 91xx; docker bind fails for the loser and we end up with a half-baked row + a wasted port slot.
  • Recommendation: either (a) add a partial unique index WHERE status IN ('provisioning','running') on port (catches the dupe), or (b) wrap the cap-check + insert in a SERIALIZABLE transaction / advisory lock on a preview_pool key. Same pattern applies to _enforce_concurrency_caps in rehearsals.py:146-176.

[HIGH] subprocess.run calls without timeout= (5 hot paths)

  • Files:
  • api/routers/scratch.py:66, 77, 78 — git status / reset --hard / clean -fd on every scratch-mode request. A wedged .git lock on disk hangs the request indefinitely.
  • api/routers/pulls.py:1025, 1052 — docker compose ps + mysqladmin ping inside the DB-health wait loop. Outer loop has its own deadline but each individual run can block forever; deadline is checked between iterations only.
  • api/routers/pulls.py:1300 — tar -xzf {tarball} -C {wp_content} (files restore). On a corrupt tarball glob seek loop, can hang.
  • api/routers/pulls.py:1372 — openssl req -x509 …. Normally fast but -newkey rsa:2048 can stall on a low-entropy box.
  • api/utils/worktree.py:78 — subprocess.run(cmd, cwd=cwd, ...) is the generic helper used by _write_gitignore_if_changed etc. No timeout.
  • api/routers/local_stacks.py:78 — docker compose -p {slug} {args} helper. Used by every local-stack action; no timeout.
  • api/routers/pushes.py:186, 193 — _git and _git_bytes helpers. Used for git status -z --porcelain on every push start.
  • Class: hung subprocess / event-loop blocker
  • Recommendation: add a sane default timeout= (30-60s for git/docker-ps, longer for tar). Standardise on a project helper _run(cmd, *, timeout) so the timeout is mandatory in the type system.

[HIGH] register_preview_job / register_apply_job not held under _FLEET_JOB_LOCK

  • File: api/services/fleet_search_replace.py:200, 737, 1165 + api/routers/fleet_search_replace.py:189-201, 270-297
  • Class: cap bypass / race
  • Detail: The router does if svc.is_fleet_job_running(): 409 then svc.register_preview_job(...) then background.add_task(svc.run_preview_job, ...). The is_fleet_job_running() scan and the register_*_job insert are not under _FLEET_JOB_LOCK. Two concurrent POSTs (e.g. operator double-clicks "Apply") both see no in-flight job, both register, both background tasks fire — run_preview_job then serialises on the lock, so the second one waits for the first to finish (correct end state), but the user sees two job_ids and the 409 promise is broken.
  • Recommendation: take _FLEET_JOB_LOCK for the check + register sequence, OR move the is_fleet_job_running() check into register_*_job itself (which is a sync function — would need to switch to async or use a threading.Lock). Easiest fix: async with _FLEET_JOB_LOCK: if running: raise; job = register(...) inside the router.

[MEDIUM] monthly_reports_%Y-%m dedup is in-memory only

  • File: api/utils/health_cron.py:551-568
  • Class: idempotency on restart
  • Detail: _LAST_FIRE_TS[month_key] = ts survives only within one process. If the server is restarted on day-1 between 06:00 UTC and the next tick after the cron has run, the month_key is missing on boot, now.day == 1 and now.hour >= 6 is still true, so the cron re-fires. generate_one_site_report itself short-circuits via the (site_id, period_start, status='done') check, so the DB state is safe; but the cron will re-launch Playwright for every enabled site, which costs ~10-30s of chromium boot per site and burns memory. Same exposure for latest_versions_sync, ssl_check, metrics_rollup (all keyed on _LAST_RUN[name] in-memory). The work itself is idempotent at DB level but the wasted re-runs are real.
  • Recommendation: persist last-fire timestamps to a cron_state table (or use the existing version_reference KV) so a restart preserves the cadence.

[MEDIUM] preview_cleanup / rehearsal_cleanup / scheduled_applies_tick set _LAST_FIRE_TS BEFORE awaiting the job

  • File: api/utils/health_cron.py:572, 584, 600
  • Class: idempotency / dropped runs
  • Detail: _LAST_FIRE_TS["preview_cleanup"] = ts is assigned before await preview_cleanup_tick(). If the tick raises before the body runs (e.g. import error after a hot reload) the timestamp is already moved forward and the next tick won't retry for 5/10/1 min. The wrapper at _run_job (used by _due-style jobs) puts the assignment in finally:, so failure-after-attempt is handled — but the wall-clock-cadence jobs don't get the same treatment.
  • Recommendation: hoist the assignment into the finally: clause of each try block, or factor out a shared helper similar to _run_job for the _LAST_FIRE_TS family.

[MEDIUM] apply_all_to_prod commit-after-add_task on the quarantine branch

  • File: api/routers/pending.py:610-631
  • Class: transaction boundary
  • Detail: _push_worktree_quarantine_to_prod(...) is called (which performs its own commits internally including enqueuing the push row + background task), THEN site.prod_quarantined_plugins = list(...) is set and await db.commit() is issued. If that final commit fails (e.g. row was modified concurrently, optimistic-lock retry), the push has already been queued + background worker spawned. The push will go through but prod_quarantined_plugins won't reflect it until the agent's post-DONE hook lands — temporarily inconsistent UI state. Not a correctness bug per se but worth a comment / try/except with a compensating recovery.
  • Recommendation: at minimum log the failure prominently; ideally update prod_quarantined_plugins BEFORE the helper enqueues the push so the commit is atomic with the push row.

[MEDIUM] scheduled_applies_tick selects pending rows without FOR UPDATE

  • File: api/services/scheduled_apply_runner.py:247-275
  • Class: multi-process race
  • Detail: The tick reads ScheduledApply.status == 'pending' AND scheduled_for <= now, then mutates each row in-loop without a row-level lock. The codebase appears single-process (uvicorn workers=1 implied by _FLEET_JOB_LOCK being module-level asyncio.Lock), so this is currently safe. But the moment someone bumps uvicorn --workers 2 for capacity, both workers will fire each scheduled apply — and apply_all_to_prod is not idempotent (it creates a new pre-apply snapshot + push every call).
  • Recommendation: either document the "single-worker-only" invariant prominently, or use with_for_update(skip_locked=True) when selecting the due rows. Same advice applies to health_cron_tick's tick body which assumes single-instance.

[MEDIUM] health_cron.py:90 swallows urllib failures as logger.warning

  • File: api/utils/health_cron.py:89-91
  • Class: error masking
  • Detail: except Exception as exc: logger.warning(...); return None. Treating URLError, socket.timeout, JSONDecodeError, and unexpected errors (e.g. KeyboardInterrupt is caught because it's not derived from Exception — OK there) all the same. A schema change on api.wordpress.org returning malformed JSON would silently leave version_reference.wp_latest stale forever. Same shape on cdn_helper.py:91-100, dns_provisioner.py:80-95, file_integrity.py:76-90, patchstack_sync.py:72-85, plugin_sandbox.py:105-115.
  • Recommendation: narrow to urllib.error.URLError, socket.timeout, json.JSONDecodeError so a real bug (NameError, TypeError from a refactor) surfaces in the logs as an exception instead of a warning.

[MEDIUM] _LAST_RUN / _LAST_FIRE_TS are mutated without a lock

  • File: api/utils/health_cron.py:444-448, 463-470, 512-617
  • Class: race (currently theoretical)
  • Detail: Only safe because health_cron_tick is a single coroutine that awaits sequentially. If someone parallelises the tick body (e.g. wrapping each job in asyncio.create_task for speed), the dicts become unsynchronised.
  • Recommendation: add a comment pinning the invariant, or use an asyncio.Lock.

[MEDIUM] _VISUAL_DIFF_TASKS strong-ref pool is correct, but _BG_TASKS in reports.py is referenced without being defined nearby

  • File: api/routers/reports.py:170-174
  • Class: completeness check
  • Detail: The pattern follows the _VISUAL_DIFF_TASKS idiom from pulls.py:450-458 correctly. Worth grepping for any new asyncio.create_task callers that don't use the pattern — services/fleet_search_replace.py:926 is the one I found: 
    asyncio.create_task(_run_push_worker(push_uuid, [...], slug))
    No strong-ref retention; on a low-load loop the task could be GC'd before completion. This is the same bug the visual-diff pool was designed to fix.
  • Recommendation: route through _spawn_visual_diff_task (or a renamed shared helper) in fleet_search_replace.py:926.

[MEDIUM] cf_purge outer AsyncClient has no explicit timeout

  • File: api/utils/cf_purge.py:103
  • Class: HTTP client timeout
  • Detail: async with httpx.AsyncClient(headers=headers) as client: — relies on httpx's 5s default. Each call inside (_purge_one at line 50 uses timeout=15.0, _zone_id_for at line 60 uses 10.0) does specify per-request, so the omission is moot in practice. Flag because the default is a fragile thing to depend on (a future httpx version could change it).
  • Recommendation: set an explicit timeout=httpx.Timeout(15.0, connect=5.0) on the client.

[MEDIUM] kick_off_run returns a Task whose caller discards it

  • File: api/services/lighthouse.py:89-95 + api/routers/performance.py:397
  • Class: garbage-collected task
  • Detail: kick_off_run(run.id) returns asyncio.create_task(...) and the router discards the return value. The semaphore inside _drive_run will keep the task alive while it's awaiting, but during the startup window (after create_task returns, before the coroutine first awaits the semaphore), only the loop's weak-ref points at it. Python 3.11+ documents this as a real bug class.
  • Recommendation: store the task in a module-level set inside lighthouse.py (same pattern as _VISUAL_DIFF_TASKS), or have the router await the task briefly.

[LOW] subprocess.run(["scp"|"ssh"|"openssl"], ..., timeout=N) — N is sometimes tight

  • Files: api/utils/preview.py:388 (30s), 547 (20s), 604 (120s); api/routers/pulls.py:1372 (no timeout)
  • Class: marginal timeout selection
  • Detail: SSH connect timeout of 5s + caddy reload window of 120s is reasonable. The 20s scp timeout is fine for small snippets but borderline if the LXC's disk is busy. Flagged for awareness, not correction.

[LOW] _VERSION_DIFF_JOBS worker pool — semaphore acquired AFTER size guard

  • File: api/utils/version_diff.py:102-105, 116-168
  • Class: minor — concurrency cap honoured but request-side latency is unbounded
  • Detail: async with DIFF_SEMAPHORE: wraps the whole _run_diff_sync call. Two requests + a long-running third causes requests 1 and 2 to hold the slots for the full duration; request 3's HTTP poll just sees status="running" indefinitely (no timeout on the queue side).
  • Recommendation: add a wait_for(...) ceiling so a queued diff job that's been waiting >30 min is marked failed: queue_timeout.

[LOW] wpcli_audit rollback nest catches all exceptions

  • Files: api/routers/wpcli.py:144-145; api/routers/db_tools.py:334-335, 666-667; api/routers/wp_options.py:334-335; api/routers/users.py:670-671; api/routers/status_page.py:226-227
  • Class: defensive exception swallow
  • Detail: All are except Exception: pass on await db.rollback(). Acceptable in a "last-ditch audit" context but the bare swallow makes a connection-pool exhaustion hard to diagnose.
  • Recommendation: log at debug level so the swallow is at least visible.

[LOW] _LAST_FIRE_TS["preview_cleanup"] & co. grow unbounded for month_key

  • File: api/utils/health_cron.py:551
  • Class: long-running process memory creep
  • Detail: _LAST_FIRE_TS[f"monthly_reports_{YYYY-MM}"] adds a new key every month. After 10 years, 120 keys. Trivial size but the dict has no pruning logic.
  • Recommendation: prune keys older than 13 months at the top of each tick.

[LOW] _run_diff_job doesn't mark the row "failed" if factory() raises

  • File: api/routers/version_diff.py:119-161
  • Class: incomplete state
  • Detail: If the second async with factory() as session block (the finalise block) raises, the row stays in running forever. There's no surrounding except for that path.
  • Recommendation: wrap the finalise in try/except, log, and let the cron's stuck-job sweeper (analogous to rehearsal_cleanup) clean up. Currently no stuck-job sweeper exists for version_diff_jobs.

[LOW] _FLEET_JOB_LOCK is held for the entire duration of a fleet job

  • File: api/services/fleet_search_replace.py:654-734, 997-…
  • Class: liveness — works as designed
  • Detail: A 30-min apply blocks every other fleet job (preview included). The router has the in-progress check + 409, so the user never blocks; the lock just enforces serialisation inside the worker. Worth a comment explicitly stating "we hold this lock for the entire job, not just registration".

[INFO] Pattern BackgroundTasks.add_task(coro_fn) correctly passes the function, not the coroutine

  • Files: api/routers/imports.py:178-187, api/routers/fresh_install.py:149-156
  • Class: regression prevention
  • Detail: Both files have explicit comments documenting the prior bug ("BackgroundTasks.add_task accepts a coroutine FUNCTION, not a coroutine object"). The audit confirmed no remaining call sites pass coro() instead of coro to add_task.

[INFO] version_diff worker correctly rmtree's on success AND failure

  • File: api/utils/version_diff.py:115-168
  • Class: cleanup
  • Detail: try: … finally: shutil.rmtree(workdir, ignore_errors=True) — the temp dir is removed even on DiffTooLargeError / tar extract failure. No leak.

[INFO] monthly_report_cron is DB-level idempotent

  • File: api/services/report_generator.py:302-318
  • Class: cron idempotency
  • Detail: Checks (site_id, period_start, status='done') AND verifies the artifact path on disk before short-circuiting. Safe to double-fire from cron.

[INFO] latest_versions_sync / ssl_check / metrics_rollup use ON CONFLICT DO UPDATE/DO NOTHING

  • File: api/utils/health_cron.py:189-202, plus the equivalent helpers in the rollup section
  • Class: cron idempotency
  • Detail: All DB writes use PG upsert. Double-firing within the same window writes the same data; no duplicate-key explosions, no row count drift.

Methodology

  1. Listed .py files under services/wp-pulse-server/api/{routers,services,utils}/ with mtime < 2 days (149 files).
  2. Searched for: subprocess.{run,Popen}, httpx.{AsyncClient,Client}, requests.*, urllib.request.urlopen, asyncio.{create_task,Semaphore,Lock,gather}, BackgroundTasks.add_task, except Exception: pass, except BaseException, bare except, with_for_update.
  3. For each subprocess call, verified a timeout= kwarg.
  4. For each Popen, traced the communicate / wait / kill lifecycle.
  5. For each add_task / create_task, traced what's being scheduled.
  6. For each "cap + insert" flow, looked for a surrounding lock or DB constraint.
  7. For each cron job in health_cron.py, traced the dedup mechanism.