UN-3445 [GATED-FEAT] Complete the PostgreSQL queue transport: scheduler, metrics, log streaming and strand recovery

[muhammad-ali-e]

What

Completes the PostgreSQL queue transport so the platform can run with every Celery
worker scaled to zero. All of it sits behind the pg_queue_enabled feature flag.

• PG scheduler — fires the non-pipeline Beat periodics, with ownership converging
  in both directions so rollback is a values change rather than a manual DB edit.
• PG metrics execution path — worker-pg-metrics plus its internal API, the
  consumer that finally lets celery-beat and worker-metrics go to zero.
• Log streaming without Celery — a Redis-list transport, removing the last worker
  that forced a Celery dependency for logs.
• Delayed visibility — the countdown/eta primitive the queue lacked.
• Strand recovery — the sweep that finalizes executions abandoned by a transport
  cutover, plus the undispatched sweep for executions that died before dispatch.

Why

Celery cannot be decommissioned while any part of the platform still depends on it.
Each item above was a remaining dependency: the scheduler, the metrics consumer, log
transport, delayed dispatch, and recovery of work stranded mid-cutover.

The recovery work in particular came out of rehearsing the cutover on a live
environment rather than from review. A 30-file ETL was deliberately stranded mid-run
by removing its workers; it stranded exactly as designed (every file COMPLETED, the
execution stuck EXECUTING, no duplicate work and no data loss) and then was not
recovered, because the sweep's selection could never reach it. That is fixed here.

How

Dispatch goes through a single seam that resolves the transport per execution and
fails closed to Celery. Transport is pinned at dispatch, so an in-flight execution
finishes on the transport it started on and only new work moves.

Schedule ownership is complementary rather than duplicated: a schedule is owned by
PG or by Beat, never both, and the convergence command moves ownership either way on
every deploy — which is what makes rollback real.

Can this PR break any existing features. If yes, please list possible items. If no, please explain why.

The flag-off path is unchanged: this adds PG paths and removes nothing. With
pg_queue_enabled off, resolve_transport() returns celery and every existing
code path runs as before.

Two things worth a reviewer's attention:

• Schedule ownership touches rows shared with django-celery-beat. Ownership is
  complementary and the convergence command is idempotent in both directions, but this
  is the highest-risk surface in the change and the one to look at hardest.
• Recovery finalizes executions. It acts only when every file execution is already
  terminal and the files have been quiet past the window, so it cannot terminalise
  live work. It re-computes the same status the callback would have written.

Database Migrations

Four, all additive:

• pg_queue/0002_pgqueuemessage_available_at — delayed visibility
• pg_queue/0003_pgperiodictask — periodic schedules
• dashboard_metrics/0004_pg_periodic_tasks — metrics periodics
• workflow_manager/workflow_v2/0026_workflowexecution_undispatched_idx — index, built
  CONCURRENTLY

No column is dropped or altered; a deployment that never enables the flag carries the
schema inert.

Env Config

• PG_SCHEDULER_ENABLED — hands the periodics to PG; the complement of Beat's own
  enabled flag
• LOG_TRANSPORT — redis or celery; must be set on every log publisher, not
  just the consumer
• WORKER_PG_STUCK_EXECUTION_RECOVERY_ENABLED (default on) and
  WORKER_PG_STUCK_EXECUTION_RECOVERY_SECONDS (default 600)

Notes on Testing

Rehearsed end-to-end on a live environment rather than only in unit tests:

• 30-file ETL on Celery as a clean baseline — completed normally in ~14 minutes
• the same ETL cut to PG mid-run with the Celery fleet removed — stranded exactly as
  predicted, with 30 destination rows for 30 files, no duplicates and no data loss;
  only the execution status was left dangling
• schedule hand-over and release both exercised, including confirming that releasing
  back to Beat does not replay every missed interval

Unit tests cover the strand-recovery selection (starvation regression, drain property,
both halves of the staleness predicate) and the schedule-ownership convergence.

One gap stated plainly: the tests added in the final commit have not been executed.
The backend suite cannot start in the author's checkout — every test in that file errors
at setup on a pre-existing schema-provisioning issue, including tests the commit does not
touch. Lint, formatting and compilation are clean. CI is what will verify them.

Related Issues or PRs

UN-3445 (epic), UN-3796, UN-3755, UN-3843

Checklist

• I have added an appropriate PR title and description
• I have read and understood the Contribution Guidelines
• My code follows the style guidelines of this project
• I have performed a self-review of my code
• I have commented on my code, particularly in hard-to-understand areas
• I have added tests that prove my fix is effective or that my feature works
• New and existing unit tests pass locally with my changes — see Notes on Testing
• I have checked my code and corrected any misspellings

[muhammad-ali-e]

PR Review — Standardized (INITIAL)

Verdict: BLOCK

Summary — Critical: 1 · High: 13 (this repo) · Medium: ~20 · Low: ~6 · Lenses run: 16/16

Reviewed under the team standardized 16-lens rubric, executed via the PR Review Toolkit specialist agents (5 scoped passes over this repo) plus an orchestrator pass on lenses 5/6/11/12/14 and the cross-repo seam with Zipstack/unstract-cloud#1744.

Coverage caveat: ~6.1k added lines across 60 files. Past what one pass covers exhaustively; depth is deliberately uneven — the sweep, scheduler ownership and reaper surfaces got the most attention.

---

The blocking finding

C1 — the undispatched sweep can delete the staged input of a running execution. Inline on undispatched_sweep.py. It is Critical rather than High on the rubric's irreversible state change / data loss anchor: the ERROR status is recoverable, the deleted input is not.

The evidence spans a file this diff does not touch, so it could not be attached inline:

workflow_helper.py:667 sets dispatched = True, then :679-692 records the dispatch handle best-effort with the exception swallowed — under the comment "continuing — the orchestrator is already running". Two further paths return without writing a handle after the message is already on the transport: :545-551 (empty handle) and :556-564 (unparseable PG handle). All three leave task_id and queue_message_id NULL on an execution that is running, which is exactly the sweep's "never dispatched" predicate. workflow_helper.py:665-666 states the invariant the sweep violates.

---

Unanchored findings

These have no line in this diff to attach to.

[High] [Lens 13, 11] — Both PRs are drafts, so no test tier is running. The PR body says of the never-executed tests: "CI is what will verify them." It is not. ci-test.yaml:64 gates on github.event.pull_request.draft == false, and gh pr checks currently reports test, e2e and report as skipping. The same is true on the cloud side, including the shadow guard that would check the on-prem render claim. Nothing verifies the new tests until these are marked ready for review.

[High] [Lens 13] — TestReconcileAtomicityRealDB can never execute (backend/scheduler/tests/test_pg_schedule_ownership.py:364-387, lines this PR does not modify). No @pytest.mark.django_db and not a TestCase, so pytest-django's blocker raises RuntimeError on the first ORM call — which is then caught by except Exception as exc: pytest.skip(...) at :387. It reports SKIPPED in every lane. ownership.py:1-12 calls the property it guards "load-bearing" (that pg_owned and PeriodicTask.enabled commit or roll back together); it has zero executing coverage. Pre-existing, but this PR makes that property considerably more load-bearing.

[High] [Lens 13, 11] — runner/tests/ is registered in no rig group. tests/groups.yaml defines no unit-runner group and tox.ini has no runner alias, so runner/tests/test_sidecar_log_transport.py — the sole guard that LOG_TRANSPORT and LOG_STREAM_QUEUE_NAME reach the sidecar's hand-picked env allowlist — is never collected. Combined with the disclosure that the new tests were not run locally, this file has been verified by nothing.

[Medium] [Lens 13, 5] — available_at missing from the worker schema-drift contract. workers/tests/test_pg_schema_drift.py:38 (pg_queue_message set, unmodified here). This PR adds worker raw SQL depending on that column (reaper.py:397-402) and adds pg_periodic_task to the manifest in the same commit, but not available_at. Rename or drop the column and the guard stays green while delayed-visibility delivery breaks at runtime.

---

Other Mediums worth naming

• The sweep does 500 x 3 remote round-trips inside a 30s client timeout that is then retried 3x on timeout (undispatched_sweep.py:59, :190-238) — a per-row WorkflowExecution query, a Redis call and an object-store call for every claimed row, including ETL rows that never had API storage.
• The under-lock re-check is strictly weaker than the selection predicate it re-confirms — last_file_at is never re-evaluated under the row lock (internal_views.py:744-745 vs :785).
• cron_from_periodic_task silently drops CrontabSchedule.timezone (mirror_pg_periodic_tasks.py:107-109); a non-UTC periodic is adopted at the wrong hour.
• Beat-side PeriodicTask.objects.filter(...).update(...) discards the match count on both write sites (mirror_pg_periodic_tasks.py:403, ownership.py:264), so a release can orphan a schedule and still report success — on the rollback path.
• converge_pg_scheduler's docstring claim that "the pipeline path no-ops when ownership already matches" is false: 2N row writes plus N PeriodicTasks.update_changed() Beat reloads per backend start.
• LOG_STREAM_QUEUE_NAME is set only on the consumer in the sibling chart while publishers rely on a matching code default (pubsub_helper.py:30) — the one place the "derive it so the bad pairing is unrepresentable" discipline was not applied.

Lens checklist

#	Lens	Result
1	Spec & intent	See findings
2	Architectural fit	See findings
3	Correctness & edge cases	See C1, H1, H2
4	Security	Clean — new internal endpoints match the repo baseline; InternalAPIAuthMiddleware gates every /internal/ path on the Bearer key before the view runs
5	Data integrity & migrations	See C1. The migrations themselves are careful — 0026 is exemplary (atomic = False, CONCURRENTLY, explicit INVALID-index guard)
6	Concurrency	Clean on the primitives — claim atomicity (FOR UPDATE SKIP LOCKED), the _CLAIM_SQL re-carried predicate, and lock ordering all verified sound. Residual issues are in findings
7	API & contract compatibility	See findings (untyped log envelope, queue-name derivation gap)
8	Reliability & resilience	See H1, H2, H4, H9
9	Performance & cost	See findings
10	Observability	See H1
11	Operational safety	See findings
12	LLM/agent	N/A — no model, prompt, tool or agent path touched
13	Testing	See findings
14	Dependencies & build	N/A — no pyproject.toml, lockfile or Dockerfile touched
15	Code quality	See Lows
16	Doc & comment accuracy	See findings

Open questions

1. Are the three no-handle paths in _record_dispatch_handle genuinely reachable in production? If they provably are not, C1 drops to High.
2. Merge/release ordering of this PR's backend image relative to the cloud chart — the sibling PR's post-upgrade hook runs converge_pg_scheduler on every cloud environment including production, and that command ships here.
3. Was widening stuck-recovery to the Celery transport and cutting the window to 600s intended as one change, or did they land together incidentally?

[muhammad-ali-e]

[Critical] [Lens 3, 5, 8] — The sweep deletes the staged input of a dispatched, running execution

Failure mode. The sweep infers "never dispatched" from the absence of two nullable columns (task_id IS NULL AND queue_message_id IS NULL, :112-130). Three paths in workflow_helper.py reach exactly that state after the message is already on the transport:

• :679-692 — handle recording is best-effort, exception swallowed, immediately after dispatched = True at :667
• :545-551 — empty handle, early return
• :556-564 — unparseable PG handle, early return

After the 15-minute grace (DEFAULT_MIN_AGE_SECONDS = 900) such a row is claimed, and _release_abandoned_resources deletes the API storage directory at :229-231 while the worker is still going to read it. The user is simultaneously told "You can safely run it again" (:70-74) — inviting a duplicate run against deleted input.

The predicate is also not transport-filtered, so this reaches Celery executions, not just PG ones.

Why Critical and not High. The rubric's Critical anchor covers data loss and irreversible state change. Marking the row ERROR is recoverable; deleting the staged input is not. workflow_helper.py:665-666 states the invariant this violates in so many words: "Past this point the orchestrator is on its transport: a failure in the bookkeeping below must NOT flip the (now-running) row to ERROR."

Suggested fix. Make dispatch a positive fact rather than an inferred absence: stamp a dispatched_at in the same call that records the handle — including on the three non-recording branches, which know the dispatch succeeded even when the handle is unusable — and key both the sweep predicate and the 0026 partial index on it. Until the predicate is sound, drop delete_api_storage_dir from the sweep; the status write is the part that matters and it is reversible.

Confidence: High that the state is reachable and that the predicate matches it. Medium on frequency — the defensive parses make the handle paths uncommon; what makes it Critical is that the consequence is irreversible.

[muhammad-ali-e]

[High] [Lens 10, 13, 3] — This metric can never increment; the alert for the incident this feature exists to detect is dead on arrival

sweep_undispatched_executions() returns self.post(...) -> BaseAPIClient.post -> _make_request -> _parse_response, which returns the parsed JSON body as a plain dict ({"swept": N}). A dict has no .data, so getattr(response, "data", None) is None -> {} -> swept is always 0.

pg_reaper_undispatched_swept_total never increments and the "terminalised N undispatched execution(s)" line never fires. The sweep works; every operator-facing signal for it reports nothing. The metric's own help text says a sustained non-zero rate means requests are dying between create_workflow_execution and dispatch — that is precisely the 967-orphan condition this feature was built for, and it is now invisible.

The sibling path already documents this exact trap and works around it, which is what makes this a regression rather than an unknown — execution_client.py:352-358:

# The endpoint returns a FLAT body ... convert_dict_response() reads response["data"],
# which is absent -> it would silently zero every counter the reaper logs from.
return APIResponse.success_response(data=response if isinstance(response, dict) else None)

Found independently by four of the five review agents.

Suggested fix. Either wrap in internal_client.py:1448-1461 the way execution_client does, or read the flat dict here: swept = (response or {}).get("swept", 0) behind an isinstance guard. Then fix the test fixture — see the comment on test_reaper_undispatched_sweep.py.

Confidence: High.

[muhammad-ali-e]

[High] [Lens 13] — This fixture encodes a contract the real client does not honour, and is why the dead metric shipped

SimpleNamespace(data={"swept": 3}) is a shape nothing in the call chain produces. The real InternalAPIClient.sweep_undispatched_executions() returns a bare dict. So inc.assert_called_once_with(3) passes against a fiction while production always increments by 0. CI cannot catch the bug because the test asserts the bug away.

Separately, and worse: nothing in this file pins that the sweep is ever invoked. Every test calls r._sweep_undispatched_executions() directly on a PgReaper.__new__ stand-in; no test drives tick() or _maybe_sweep(). Delete the call site at reaper.py:1355 and the entire feature goes dead with a green suite — PENDING rows accumulate forever with no error anywhere. The file's own docstring claims the opposite: "These pin the wiring, which is the part the backend tests cannot see." grep -rn "_maybe_sweep" workers/tests/ returns nothing.

Suggested fix. Return {"swept": 3} from the mock so the test exercises the real contract, and add one test that drives _maybe_sweep() with _sweep_undispatched_executions patched — asserting it is called once and that an immediate second call is cadence-gated out.

Confidence: High.

[muhammad-ali-e]

[High] [Lens 3, 8, 11] — Stuck-recovery was widened to the Celery transport and had its window cut 9000s -> 600s in the same change, default-on

Verified against the base ref. The predicate changed from:

queue_message_id__isnull=False,  # PG-only; Celery uses task_id

to Q(queue_message_id__isnull=False) | Q(task_id__isnull=False) — every Celery execution is now in scope — while the recovery window default moved from self._stuck_timeout_seconds (the barrier timeout, ~9000s) to a hard-coded _DEFAULT_STUCK_RECOVERY_SECONDS = 600 (reaper.py:203), with recovery enabled by default.

Failure mode. Any execution whose files all went terminal >=10 minutes ago and whose chord callback has not yet written a terminal status gets finalized by the reaper. A backlogged callback queue produces exactly that state for healthy work — and a backlogged callback queue is the condition under which strands occur in the first place. The user sees the execution flip while the callback is still pending; the API-deployment rate-limit slot is released early; and the real callback's write is then refused by the protected-status guard, so the failure notification never fires.

The mitigation the code offers does not cover this. The comment at :730-742 argues that requiring last_file_at < cutoff closes the race. It closes only the immediately-after-completion window: an execution whose files finished 20 minutes ago with a still-queued callback satisfies both predicates.

Suggested fix. Separate the two changes. Keep the Celery widening if it is wanted, but leave the window at the barrier stuck-timeout for task_id-only rows, or gate the Celery half behind its own switch defaulting off so this GATED-FEAT PR does not alter live Celery behaviour. At minimum emit a distinct metric for Celery-transport recoveries so premature finalizations are countable.

Confidence: High that the predicate admits backlogged-callback executions; Medium on frequency, which depends on the real callback-latency distribution.

[muhammad-ali-e]

[High] [Lens 16] — This comment tells the next maintainer the threshold is not load-bearing; this PR is what made it load-bearing

The comment argues the recovery window "only needs to outlast a callback that is about to fire — seconds" and that "a legitimately running execution is never a candidate no matter how short this is."

That was arguably true when the only guard was all-files-terminal. This PR adds a second, threshold-dependent guard (last_file_at__lt=cutoff) precisely because all-files-terminal is not sufficient — and says so 500 lines away in internal_views.py:733-741: "the all-files-terminal filter above stops that being catastrophic, but it leaves a live race... Requiring the LAST FILE to also be older than the cutoff closes it."

Someone who trusts this comment and drops WORKER_PG_STUCK_EXECUTION_RECOVERY_SECONDS to 30s reopens the callback race. Per lens 16 that rates High: a comment that would actively mislead a maintainer into a bug.

Suggested fix. Invert the claim — the threshold must exceed the worst-case callback latency after the last file goes terminal, because the selection predicate now includes last_file_at < cutoff. Keep the "not the barrier timeout" rationale; delete "no matter how short this is" and "— seconds".

Confidence: High.

[muhammad-ali-e]

[High] [Lens 8, 11, 16] — adopt=[] means adopt everything, on every backend start, with no check that a consumer exists

In _set_ownership, if names: is falsy for [], so the name filter is skipped and all rows are adopted. _mirror runs first and mirrors every PeriodicTask not in the three-entry _EXCLUDED_TASK_PATHS, with no verification that task_name is registered in the PG worker's task registry or that any PG consumer polls that queue.

Because entrypoint.sh runs this on every backend start, any periodic added to Beat afterwards — by a plugin, a cloud-repo migration, an operator via the admin — is silently mirrored and adopted on the next pod restart: pg_owned=True and the Beat row disabled in one transaction. If nothing implements that task, the periodic stops running entirely and exits 0.

The docstring at :25 claims "Adoption is not unilateral: it happens only because someone set the env var." That holds for rows present when the var was set; it is false for every row created afterwards. This is not hypothetical — mirror_pg_periodic_tasks.py:49-53 records that a real Beat table was found carrying a legacy execute_pipeline_task_v2 row "only excluded here by luck of being disabled."

Suggested fix. Have the unattended path adopt a declared allowlist (e.g. the names seeded by dashboard_metrics/migrations/0004) rather than []-means-everything, and have plan_mirror refuse to adopt a task_name absent from the worker registry, reporting it as a skip.

Confidence: High on the mechanism.

[muhammad-ali-e]

[High] [Lens 8, 11, 2] — A Flipt outage at backend start silently releases every PG-owned schedule

resolve_schedule_owner returns False when FLIPT_SERVICE_AVAILABLE != "true" or when the Flipt call raises. reconcile_ownership_for then writes pg_owned=False, next_run_at=NULL and re-enables the Beat row — for every schedule in the sweep.

Fail-closed-to-Beat was the right call when this ran as an explicit operator action. It is a different risk now that entrypoint.sh:63 runs the full sweep (no mirror_only) unattended on every container start. If Beat has already been scaled down — which is the stated goal of this epic — nothing fires any scheduled pipeline. It does not self-heal when Flipt recovers; only the next backend restart re-adopts. The command still exits 0, so there is no alert, only per-pipeline logger.warning.

reconcile_pg_schedules.py:11-13 still describes the old posture: "kept a command here so the ramp stays an explicit, auditable ops action."

Suggested fix. Distinguish "flag says no" from "could not evaluate", and treat the latter as leave ownership unchanged rather than release. Alternatively abort _reconcile_all with a non-zero exit when FLIPT_SERVICE_AVAILABLE != "true", rather than sweeping thousands of rows to Beat.

Confidence: High on the mechanism; Medium on blast radius, which is only an outage once Beat is genuinely retired.

[muhammad-ali-e]

[High] [Lens 16, 11] — This comment block describes a command that is not run, and asserts the opposite of what the container now does

These lines document a reconcile_pg_schedules --mirror-only invocation that does not exist in this file, and state:

--mirror-only: purely additive, writes only pg_periodic_schedule and never a Beat PeriodicTask, so it is safe at any flag state. Ownership hand-over stays an explicit operator action.

What actually runs at :63 is converge_pg_scheduler, which with PG_SCHEDULER_ENABLED=true flips pg_owned and disables Beat PeriodicTask rows for every mirrored pipeline. An SRE auditing "can a rolling restart move schedules off Beat?" gets the wrong answer — and the answer is yes, for the single riskiest behaviour this PR introduces.

The staleness is easy to miss because two contradictory "Best-effort by design" paragraphs survive side by side (:44-45 and :55-56), describing two different commands.

Flagged independently by three review agents. The same contradiction exists at reconcile_pg_schedules.py:102-107.

Suggested fix. Delete :31-45; the block at :46-56 already describes what runs. Update reconcile_pg_schedules.py:102-107 to say the entrypoint runs the non-mirror_only mode when the gate is on.

Confidence: High.

[muhammad-ali-e]

[High] [Lens 16, 15] — The docstring promises reorder-safety that positional unpacking does not provide

The type is justified as binding "the field names to the SELECT's column order at exactly one site", and its sibling _DueSchedule states outright that this means "a future reorder of the SELECT can't silently misassign fields."

_DuePeriodicTask(*row) at :303 is pure positional unpacking — no name binding, no runtime validation. A reorder is exactly what it cannot catch. Six of the eight fields are same-typed TEXT (name, task_name, queue, org_id, cron_string), so swapping task_name and queue constructs cleanly and then fires the queue name as a task, onto a queue named after the task — a periodic that silently stops running, with a MAX_ATTEMPTS=1 message dropped on an unwatched queue.

A maintainer who reads this docstring will reorder the SELECT believing the type protects them. That is the bug the comment claims to prevent.

Suggested fix. Make the claim true instead of weakening it — derive the column list from the type:

f"SELECT {', '.join(_DuePeriodicTask._fields)} FROM ..."

One line, and it makes reordering structurally impossible for both this and _DueSchedule. Or use psycopg2.extras.NamedTupleCursor.

Confidence: High.

[muhammad-ali-e]

[High] [Lens 13] — _release_stale is the rollback half and runs on every backend start, with no test

entrypoint.sh runs converge_pg_scheduler on every backend start with migrate; gate-off routes here via --mirror-only --release-stale. Three separately-fatal branches are unexercised:

1. the if pg_scheduler_enabled(): return 0, 0 early-out — invert or drop it and every deploy releases the whole live PG-owned fleet back to Beat mid-rollout, giving two firers per schedule
2. the filter(pg_owned=True) scoping — widen it and every clean install does N writes plus a PeriodicTasks.update_changed() per row on each boot, contradicting the "Beat's tables stay untouched" guarantee the design rests on
3. active=row.enabled — pass True and every operator-paused pipeline is resurrected on deploy

test_converge_pg_scheduler.py patches call_command and asserts only kwargs["release_stale"] is True, so it never reaches this body. grep -n "release_stale" backend/pg_queue/tests/ matches nothing else.

Suggested fix. Add DB-free tests mirroring the existing TestChunking mock style: gate-on returns (0, 0) and calls nothing; gate-off releases exactly the pg_owned=True rows; active comes from row.enabled; --dry-run writes nothing.

Confidence: High.

[greptile-apps]

Greptile Summary

The PR completes the PostgreSQL queue rollout path behind feature flags, including periodic scheduling, dashboard-metrics execution, Redis-backed log streaming, delayed visibility, and execution recovery.

• Adds complementary PostgreSQL/Celery Beat schedule ownership and deployment-time convergence.
• Adds PostgreSQL metrics workers and internal backend endpoints.
• Adds positive dispatch tracking and recovery sweeps for stranded or undispatched executions.
• Adds Redis-list log transport and the corresponding worker/runtime configuration.

Confidence Score: 5/5

The PR appears safe to merge.

No blocking failure remains.

Important Files Changed

Filename	Overview
backend/workflow_manager/workflow_v2/workflow_helper.py	Stamps successful dispatch before best-effort transport-handle bookkeeping while preserving the post-dispatch boundary.
backend/workflow_manager/workflow_v2/undispatched_sweep.py	Uses the positive dispatch marker together with legacy handle checks to avoid claiming live executions during rolling deployment.
backend/workflow_manager/workflow_v2/migrations/0027_workflowexecution_dispatched_at.py	Adds the nullable dispatch marker needed to distinguish queued work from executions abandoned before dispatch.
backend/workflow_manager/workflow_v2/migrations/0028_undispatched_idx_dispatched_at.py	Replaces the recovery partial index concurrently while preserving index coverage during deployment and rollback.
backend/pg_queue/management/commands/converge_pg_scheduler.py	Converges schedule ownership in either direction according to the PostgreSQL scheduler rollout configuration.
backend/pg_queue/management/commands/mirror_pg_periodic_tasks.py	Mirrors eligible Beat schedules and transfers ownership atomically between Beat and PostgreSQL.
workers/queue_backend/pg_queue/pg_scheduler.py	Implements PostgreSQL-owned periodic dispatch with transactional schedule advancement.
workers/queue_backend/pg_queue/reaper.py	Extends leader-driven recovery for queue claims, stranded executions, and undispatched execution cleanup.

Flowchart

%%{init: {'theme': 'neutral'}}%%
flowchart LR
  Trigger[Workflow or periodic trigger] --> Route{Selected transport}
  Route -->|Celery| Celery[Celery queue and Beat]
  Route -->|PostgreSQL| PG[PostgreSQL queue and scheduler]
  PG --> Worker[PG worker]
  Celery --> Worker
  Worker --> Execute[Execute workflow or metric task]
  Execute --> Finalize[Persist terminal execution state]
  Recovery[Recovery sweeps] --> Finalize
  Logs[Execution logs] --> Redis[Redis list]
  Redis --> LogConsumer[Log consumer]

Loading

_{Reviews (5): Last reviewed commit: "UN-3796 [FIX] Match 0027's field to the ..." | Re-trigger Greptile}

[greptile-apps]

Live executions lose concurrency slots

When dispatch succeeds but handle recording returns an empty or non-numeric handle or raises after enqueue, the execution retains both NULL handles; after 15 minutes this sweep marks the active execution ERROR and releases its API concurrency slot, causing an incorrect failure state and allowing the organization to exceed its configured concurrency limit while the worker is still running.

Knowledge Base Used:

• Workflow execution system
• Queue-backed worker orchestration

Prompt To Fix With AI

This is a comment left during a code review.
Path: backend/workflow_manager/workflow_v2/undispatched_sweep.py
Line: 120-121

Comment:
**Live executions lose concurrency slots**

When dispatch succeeds but handle recording returns an empty or non-numeric handle or raises after enqueue, the execution retains both NULL handles; after 15 minutes this sweep marks the active execution `ERROR` and releases its API concurrency slot, causing an incorrect failure state and allowing the organization to exceed its configured concurrency limit while the worker is still running.

**Knowledge Base Used:**
- [Workflow execution system](https://app.greptile.com/zipstack/-/custom-context/knowledge-base/zipstack/unstract/-/docs/workflow-execution.md)
- [Queue-backed worker orchestration](https://app.greptile.com/zipstack/-/custom-context/knowledge-base/zipstack/unstract/-/docs/worker-queue-orchestration.md)

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

[muhammad-ali-e]

Valid finding — not disputing it. Partially addressed; the root cause is still open and tracked. Breaking it down, since "fixed" would overstate it:

Fixed — the irreversible half. _release_abandoned_resources no longer calls delete_api_storage_dir (bb772503b). That was the part that could not self-correct: the sweep deleted a live execution's staged input while the worker was still going to read it, and told the user "You can safely run it again". Tests now pin the absence of the call so it cannot be reinstated without revisiting the predicate.

Mitigated — the window. DEFAULT_MIN_AGE_SECONDS raised 900 → 3600 (1b7a26f65), so this comment's "after 15 minutes" is now an hour. That was chosen to sit clear of any realistic dequeue latency, because the failure needs an unrecorded handle and a queue wait longer than the grace, at the same time.

Not fixed — the predicate itself. It is still an inference from the absence of two nullable columns, and the three workflow_helper paths you describe still reach that state after the message is on its transport. The unsoundness is now documented at the predicate rather than left implicit, but documenting is not fixing.

One correction worth recording, because it cuts against us. The concurrency-slot consequence you name was not actually reachable when this review ran: release_slot(cls, organization_id: str, ...) was being passed an Organization instance, so its Redis key formatted to ...:org:Organization object (12) while acquire_slot had built the key from str(organization.organization_id). The ZREM removed a non-member, returned 0, and raised nothing — a mis-claimed live execution kept its slot by accident. That is fixed in 6c658c060, which makes your finding's specific consequence real for the first time.

We still think that is the right trade: the previous behaviour leaked a slot for the full 6h limiter TTL on every genuinely-abandoned execution, which throttles the whole org — a 502 storm could hold a tenant down for hours. One org briefly exceeding its limit in a rare mis-claim is the smaller harm. But it is a genuine cost of the fix and worth having in the thread.

The actual close is to make dispatch a positive fact rather than an inferred absence: stamp a dispatched_at in the same call that records the handle — including on the three branches that currently return without writing one — and key both this predicate and the 0026 partial index on it. That needs a migration and an index rebuild, so it is tracked separately rather than folded into this PR.

For context on the score: this review ran against 9e502f7da, which predates every mitigation above (7f3cc38ab, 6c658c060, 1b7a26f65, 02ecc60a2). Leaving the thread open deliberately — it should stay open until dispatched_at lands.

[greptile-apps]

Agreed. The finding remains open: the irreversible cleanup is removed and the grace period reduces the mis-claim window, but nullable-handle absence is still not a sound proof that dispatch never occurred. The correction on concurrency-slot behavior is also noted—the slot-exceedance consequence became reachable only after release_slot was fixed. Keeping this thread open until dispatched_at makes dispatch a positive fact is appropriate.

[muhammad-ali-e]

Root fix landed in 1c4f1c785 (+ a713fb6e4). This is no longer mitigated — the inference is gone.

What changed. dispatched_at is stamped by the dispatcher the instant dispatch returns, at workflow_helper.py:667, before any handle bookkeeping runs. That single write sits upstream of all three paths you identified — the swallowed exception, the empty handle, the unparseable PG handle — so the sweep now tests a recorded fact instead of inferring one from absence. A live execution always carries the stamp and can no longer be claimed.

Rollout is single-phase, deliberately. The predicate keeps both handle checks alongside the new one:

WHERE status = 'PENDING' AND created_at < now() - grace
  AND dispatched_at IS NULL
  AND task_id IS NULL AND queue_message_id IS NULL

During a rolling upgrade an old pod dispatches without stamping; that row matches dispatched_at IS NULL but is excluded by task_id IS NOT NULL. Existing rows need no backfill for the same reason. Dropping the handle checks would have required two releases, or swept live work mid-deploy.

A gap this opened, and closed. Requiring dispatched_at IS NULL means a row that was dispatched but whose handle write failed stops matching this sweep — and it did not match recover_stuck_pg_executions either, which tested the handles alone. It would have matched neither, trading "wrongly terminalised" for "never terminalised". Recovery is now a three-way test (dispatched_at OR either handle), so the two sweeps are disjoint and jointly exhaustive: undispatched takes rows where all three are absent, recovery takes rows where any one is present. Cleaner than before this change.

Verified against a live Postgres, both directions — not asserted:

• makemigrations --check --dry-run → no changes detected (this is what caught a713fb6e4: the migration and model had divergent db_comment text, which Django read as a pending alteration)
• 0026 → 0027 → 0028 apply cleanly; dispatched_at is nullable timestamptz
• old index dropped, we_undispatched_dispatch_idx created, 0 INVALID indexes
• EXPLAIN on the sweep's exact predicate → Index Scan using we_undispatched_dispatch_idx, so the partial index is matched rather than bypassed
• migrate back to 0026 unapplies both, restores the old index and drops the column
• 139 tests pass across workflow_v2 and execution, including three new regressions: a dispatched row with no handles is never swept (the exact shape your three paths produce), an unstamped row with a handle is not swept (the rolling-deploy guarantee), and a genuinely undispatched row still is swept (the fix must not disable the sweep)

Residual, stated plainly. If the stamp itself fails, that one execution falls back to the old ambiguity — three paths narrowed to one, not zero. The log line at that site says so rather than staying silent. There is also one extra small UPDATE per dispatch, and 0028 builds its index CONCURRENTLY on a multi-million-row table (the migration docstring carries the out-of-band build command for production).

Follow-up this unlocks: the grace period was raised 900s → 3600s in 1b7a26f65 purely to narrow this window. With the inference gone it is no longer load-bearing and can return to 15 minutes, terminalising genuine orphans 4× sooner. Left as-is for now — separate, reversible tuning.

Leaving the thread open for you to close.

[sonarqubecloud]

Quality Gate passed

Issues
14 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

[github-actions]

Unstract test results

Per-group results

Status	Group	Tier	Passed	Failed	Errors	Skipped	Duration (s)
✅	e2e-api-deployment	e2e	3	0	0	0	21.0
✅	e2e-coowners	e2e	1	0	0	0	1.6
✅	e2e-etl	e2e	1	0	0	0	8.4
✅	e2e-login	e2e	2	0	0	0	1.2
✅	e2e-prompt-studio	e2e	1	0	0	0	4.5
✅	e2e-smoke	e2e	2	0	0	0	1.5
✅	e2e-workflow	e2e	1	0	0	0	16.4
✅	integration-backend	integration	310	0	0	26	42.2
✅	integration-connectors	integration	1	0	0	7	7.6
✅	integration-workers	integration	157	0	0	1	49.3
✅	unit-backend	unit	1140	0	0	1	42.5
✅	unit-connectors	unit	63	0	0	0	10.2
✅	unit-core	unit	33	0	0	0	1.4
✅	unit-platform-service	unit	15	0	0	0	2.8
✅	unit-rig	unit	117	0	0	0	5.5
✅	unit-runner	unit	5	0	0	0	3.1
✅	unit-sdk1	unit	563	0	0	0	30.2
✅	unit-workers	unit	1397	0	0	1	128.5
	TOTAL		3812	0	0	36	377.8

Critical paths

⚠️ Critical paths not yet covered

• workflow-execution-fan-out — Multi-file workflow execution fans out to file-processing workers and rejoins. (declared coverage: no groups declared)

✅ Covered critical paths

• auth-login — covered by e2e-login
• adapter-register-llm — covered by integration-backend
• workflow-author — covered by integration-backend
• co-owner-manage — covered by integration-backend, e2e-coowners
• workflow-create-execute — covered by e2e-workflow
• api-deployment-provision — covered by integration-backend
• api-deployment-auth — covered by integration-backend
• api-deployment-run — covered by e2e-api-deployment
• mcp-server-auth — covered by integration-backend
• mcp-platform-auth — covered by integration-backend
• prompt-studio-author — covered by integration-backend
• prompt-studio-fetch-response — covered by e2e-prompt-studio
• connector-register-test — covered by integration-backend
• pipeline-etl-execute — covered by e2e-etl
• usage-aggregate-read — covered by integration-backend
• usage-token-tracking — covered by e2e-api-deployment
• callback-result-delivery — covered by e2e-api-deployment