UN-4046 [MISC] Roll the PG queue out of the pg_queue_enabled feature flag

[muhammad-ali-e]

What

Removes the pg_queue_enabled feature flag and makes the PostgreSQL queue the
unconditional default — in code, in environment defaults, and in the local Docker stack.

• workflow_manager/workflow_v2/transport.py deleted. resolve_transport() was
  entirely flag logic; its three callers now use the PG constant directly. The
  transport wire field is kept, hardcoded to pg_queue — workers read it at eight
  sites, and removing it is separate work.
• Remaining flag reads gone: prompt_studio_core_v2 task_status, scheduler/ownership,
  both pg_queue/executor_rpc factories, and the shared resolve_pg_transport /
  RoutingExecutionDispatcher in unstract/workflow-execution.
• Two defaults flipped to PG. Neither moves with the flag, and each independently
  leaves a producer with no consumer:
  • select_backend() read WORKER_PG_QUEUE_ENABLED_TASKS, an allow-list set
    nowhere (commented out in sample.env), so every dispatch without an explicit
    backend= went to Celery. The live case is api-deployment/tasks.py dispatching
    process_batch_callback_api. The allow-list is deleted along with its plumbing.
  • PG_SCHEDULER_ENABLED default false → true. It gates whether the PG scheduler
    owns a pipeline's cron; left off with Beat disabled, nothing fires schedules at all.
• WORKER_BARRIER_BACKEND is deliberately not touched — _barrier_for_transport()
  returns a PgBarrier whenever the transport is PG, regardless of that variable.
• docker/docker-compose.yaml is PG by default. The eleven Celery worker services are
  removed, profiles: [pg-queue] is dropped from the PG consumers, and six queues that
  had no PG consumer are closed (file_processing_priority, celery_callback,
  celery_executor_table, _smart_table, _simple_prompt_studio, _lookup_test).
  worker-log-history-scheduler-v2 (a bash loop, not Celery) and
  worker-log-stream-consumer stay.
• Dead Celery-coupled code removed where the flag removal made it trivially reachable:
  four callerless webhook views and their serializers, submit_file_batch_for_processing
  end to end, and three unused constants.
• Retry jitter in the workers' HTTP client is now unconditional. It was gated so the
  Celery flow kept an exact exponential cadence; it is transport-agnostic hardening and
  the jittered cadence is what integration has run under load.

Why

QA passed 100% PG on integration, so the flag has served its purpose and staging is next.

It has also become a liability rather than a safety net. On 2026-08-28 a routine node
rebalancing eviction restarted Flipt on integration; resolve_transport failed closed to
Celery and published to RabbitMQ, where a 100%-PG topology has no consumers. Two
executions stranded permanently — the undispatched sweep excludes them (task_id is set)
and stuck-execution recovery excludes them (the file never reached a terminal state). With
no flag there is no Flipt call in the dispatch path and this class of strand disappears.

The two default flips matter for the same reason. A producer without a consumer does not
merely lose the work: the messages accumulate in RabbitMQ until it hits its memory
high-watermark and starts blocking publishers, which degrades everything still using the
broker.

How

Deletion-led. Almost every hunk removes a branch rather than adding code — the current
flag-on behaviour becomes the only behaviour, with no business-logic change. Net
−2 000-odd lines.

Tests were reworked per assertion rather than wholesale: anything pinning a payload, queue
or error contract is kept; anything that only tested the branch choice is removed, with
the reason recorded where it was. Two examples:

• TestDispatchAttachesFairness patched current_app and read send_task.call_args.
  That branch is now unreachable, so the cases assert the same property against the PG
  path — fairness must reach the transport without being folded into the business kwargs,
  which on PG means the payload slot and the org_id / priority columns the dequeue
  orders by. One case keeps the Celery branch covered through an explicit backend=
  override, its only remaining reachability.
• TestBackoffJitter's two flag-off cases are gone; they asserted the exact exponential
  cadence the Celery flow kept, and the band assertion already pins the result from both
  sides.

A new zero-producer canary (TestNoCeleryProducer) asserts that no dispatch can
resolve to Celery, so a future call site cannot silently reintroduce a broker publish.

Can this PR break any existing features?

The intended behaviour change is that execution dispatch no longer has a Celery path.
Everything else is preserved, and the two risks are deployment-shaped rather than
code-shaped:

• Deploy order is load-bearing and one-directional. The PG worker fleet must be up in
  the same release as this code. Deploying this against a topology whose PG consumers are
  off would enqueue to queues nothing drains — no fallback, no reaper recovery, no error.
  The chart's partial-fleet guard refuses the half-configured version at render time.
• There is no runtime rollback any more. Reverting means redeploying the previous
  image and chart, not flipping a flag.

Not affected: queue payloads and task names are unchanged; the transport wire field
still travels (hardcoded to pg_queue), so workers need no coordinated change; the Celery
library and the @app.task registry stay permanently, because the PG consumer is a
Celery app and resolves work through self._app.tasks.get(name).

step_execution (single-step workflow execution) now raises unconditionally. It has had
no UI for some time — both live front-end call sites pass isStepExecution=false — so it
is reachable only by a direct API call, and any organisation already on PG hits this
today. Deleting the endpoint outright is tracked separately as UN-4051.

Database Migrations

None.

Env Config

• WORKER_PG_QUEUE_ENABLED_TASKS — removed. Delete it from any environment that sets
  it (it was commented out in sample.env and set nowhere).
• PG_SCHEDULER_ENABLED — default changes false → true.
• No new variables.

Notes on Testing

Run locally against a live Postgres:

Backend suite	1005 passed, 29 skipped, 0 failed
Workers suite	1501 passed, 1 skipped, 0 failed
pre-commit, changed files	clean
ruff (0.3.4, changed files)	24 → 23 errors vs main

The backend suite was run in a clean worktree. A working checkout with git-ignored
enterprise plugins present produces unrelated import failures that look like regressions.

Two checks beyond the suites, because no CI job covers them:

• docker/docker-compose.yaml — parses with no profile into twelve worker services,
  all PG except the two log services; zero Celery workers, zero beat.
• Queue coverage — every queue with a PG consumer in the local stack has one in the
  deployment chart. Zero gaps in that direction.

Related Issues or PRs

UN-4046, under epic UN-3445. Follow-up filed as UN-4051 (delete the deprecated
single-step execution endpoint).

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
• I have checked my code and corrected any misspellings

[muhammad-ali-e]

Ran the local Docker stack while preparing this — worth recording, since no CI job builds or runs docker/docker-compose.yaml and that file is where eleven services were deleted.

It found a real defect, now fixed in 3e962ae. Removing worker-log-consumer-v2 left the stack with a Redis log consumer and zero publishers pointed at it: LOG_TRANSPORT was set on worker-log-stream-consumer — the consumer — and nowhere else. Every publisher kept the celery default and would have written execution logs to celery_log_task_queue on RabbitMQ, which this stack no longer drains. Silent in both directions: LogPublisher swallows its errors, so the symptom is execution logs that stay empty while the queue grows toward the broker's memory high-watermark. The requirement was already written in the compose comment above the consumer; the change just did not honour it.

It is now set on all fourteen publishers — backend, runner and the twelve workers — in environment: rather than essentials.env. That file is git-ignored and generated, and copy_or_merge_envs only merges new keys on --only-env/--update, so every existing checkout would have silently kept the old default on a plain ./run-platform.sh.

What the run verified

• Stack comes up with no profile: twelve worker containers, all PG except worker-log-history-scheduler-v2 (a bash loop) and worker-log-stream-consumer. Zero Celery workers, zero beat.
• Every PG consumer starts and polls its queues. The six previously-uncovered queues are now drained — file_processing_priority on worker-pg-fileproc, celery_callback on worker-pg-callback, and celery_executor_table / _smart_table / _simple_prompt_studio / _lookup_test on worker-pg-executor.
• 0 ERROR-level lines across all workers. (One startup line in worker-log-stream-consumer is a RabbitMQ race that retried to ready.)
• All 21 named RabbitMQ queues at 0 messages except two carrying pre-existing backlog from earlier sessions; both stayed flat across the run.

What it could not verify. Docker Hub is unreachable from this machine (TLS handshake timeout), so images containing these changes could not be built — the containers ran a prebuilt worker-unified image. This exercises the compose wiring (service set, queue assignments, env), not the Python changes; those are covered by the suites. A full ETL end-to-end still needs to happen on integration.

[muhammad-ali-e]

Standardized review — BLOCK

Mode: INITIAL · reviewed at 3e962aec8 · 16/16 lenses · ten specialist agents across both repos, findings deduplicated and re-rated against the shared rubric, every Critical and most Highs independently verified.

Summary (both repos): Critical 2 · High 12 · Medium 21 · Low 8. This PR carries the Critical.

Findings are in the body rather than as inline comments by necessity: most of the severe ones sit on lines this diff does not touch (the headers= call sites, DEFAULT_WORKFLOW_TRANSPORT, run-worker.sh, the ownership.py docstring), and GitHub rejects inline anchors outside a diff hunk. Every finding carries file:line.

---

CRITICAL — every document extraction raises TypeError

workers/file_processing/structure_tool_task.py:501, 544, 761 → unstract/workflow-execution/src/unstract/workflow_execution/executor_rpc.py:135

get_executor_dispatcher() now returns PgExecutionDispatcher, whose dispatch(self, context, timeout=None) has no headers parameter and no **kwargs. The deleted RoutingExecutionDispatcher.dispatch(context, timeout, headers) absorbed it. Three live call sites still pass headers=; :544 is the main structure_pipeline path.

Reproduced directly:

TypeError: PgExecutionDispatcher.dispatch() got an unexpected keyword argument 'headers'

Caught by the wrapper at :231 and surfaced as ExecutionResult.failure, so every document fails extraction with a message naming a Python signature rather than the cause. dispatch_async lost headers too.

It ships green because every dispatcher double is a bare MagicMock() with no spec= (workers/tests/test_structure_tool_pipeline.py:177 and ~11 siblings), so the suite cannot see a dispatcher signature change at all.

Fix: drop the three headers= arguments (PG carries fairness in the enqueue payload), and add spec=PgExecutionDispatcher to the doubles so this class of drift fails a test instead of production.

---

HIGH

H-a · The safe default inverted without moving — unstract/core/src/unstract/core/data_models.py:219 (not in this diff)
DEFAULT_WORKFLOW_TRANSPORT = WorkflowTransport.CELERY.value, and normalize_transport still fails closed to it, at eight consumer signature sites. Producers now always send pg_queue, but a payload missing the field selects CeleryChordBarrier via _barrier_for_transport and skips the PG orchestration-claim and terminal-redelivery guards — publishing the fan-out to RabbitMQ, which this PR empties of consumers. No barrier row is created, so the reaper is blind and the execution sits EXECUTING forever. Reachable on a rolling deploy: a not-yet-upgraded backend returns transport: "celery", select_backend routes to PG anyway, and the worker fans out over the dead broker. One-line fix: flip the constant to PG_QUEUE in this PR.

H-b · The debug stack drains nothing — docker/compose.debug.yaml:71-186
15 celery -A worker invocations, 0 pg-queue-consumer. Service keys were renamed to worker-pg-*; the bodies were not. As an override this replaces command: ["pg-queue-consumer"], so the documented debug flow turns every PG consumer back into a Celery worker bound to RabbitMQ. Executions hang with empty logs — the exact symptom this migration exists to remove.

H-c · PG_SCHEDULER_ADOPT_PERIODICS is set nowhere — repo-wide grep returns zero
converge_pg_scheduler adopts dashboard_metrics.* only with --periodics, which backend/entrypoint.sh:66 derives from that variable. This PR deletes celery-beat and worker-metrics while those periodics stay Beat-owned, so nothing fires them. The cloud chart refuses this combination at render time; the OSS compose has no equivalent guard.

H-d · run-worker.sh has no pg-notification role — workers/run-worker.sh:61-93
No notification role exists at all, and three queue lists drifted from compose (celery_callback, file_processing_priority, and four celery_executor_*). ./run-worker.sh pg silently drops every buffered webhook — _enqueue_pg succeeds, nothing drains.

H-e · Operator-facing docs invert the most consequential behaviour change — backend/entrypoint.sh:36-45, backend/scheduler/ownership.py:14-25
Both still describe two gates with PG_SCHEDULER_ENABLED defaulting off, and state that a restart "moves nothing". It is now one gate defaulting on, so the first --migrate start hands every mirrored schedule to PG and disables Beat's rows. entrypoint.sh even records that a previous version of itself made this same mistake.

---

MEDIUM / LOW — grouped (29 findings)

Lens 16 dominates. Docstrings still promising a Celery branch: both pg_queue/executor_rpc.py module headers, dispatch() (workers/queue_backend/dispatch.py:93-104, incl. a reference to the deleted _log_allow_list_once), workers/queue_backend/__init__.py:27-34 ("observable but inert"), workers/queue_backend/pg_queue/__init__.py:19-20, workers/queue_backend/pg_queue/README.md:6-7,56-59 (tables the deleted flag and allow-list as live knobs), docker/sample.env:147-151, docs/local-dev-setup-executor-migration.md:132-145 (commands for deleted services). Half-edits: docker/docker-compose.yaml:430 claims workers get LOG_TRANSPORT "via essentials.env" — there are 14 explicit per-service entries and 0 in that file, and the correct rationale is written 370 lines earlier; :458-459 still calls worker-pg-executor "Dark until the pg_queue_enabled gate is flipped" when it is now the only executor. Orphans: backend/pg_queue/flags.py (zero importers, docstring cites the deleted transport.py), and the PG_QUEUE_FLAG_KEY comment now heading EXECUTE_TASK at executor_rpc.py:47-49. An operator-facing RuntimeError at workers/queue_backend/pg_barrier.py:475 still says "before enabling PG transport (pg_queue_enabled)".

Correctness / quality. backend/backend/internal_api_constants.py:68 — webhook_send() -> str lost its body and returns None. docker/sample.compose.override.yaml:299 builds worker-pg-metrics from backend.Dockerfile, which rejects pg-queue-consumer and crash-loops. Dead assignment at workflow_helper.py:1023. QueueBackend.CELERY is unreachable by construction yet dispatch() still executes send_task for it — an unreachable branch that silently drops messages is worse than one that raises. api-deployment/tasks.py:755 now routes to PG without PG_TRANSPORT_CALLBACK_KWARG, so the callback's duplicate guard is skipped (enclosing comment argues the branch is unreachable — worth confirming).

Testing. Dispatcher doubles lack spec= — this is why the Critical shipped green. test_flipt_is_never_consulted_while_the_gate_is_off (backend/scheduler/tests/test_pg_schedule_ownership.py:296) now has zero assertions. Two test_step_execution_transport cases were deleted, one of which existed specifically to pin the guard placement this PR moved. test_uuid_job_id_is_stringified deleted though the coercion is live. TestNoCeleryProducer overclaims: it enumerates six names against an unconditional return QueueBackend.PG and inspects no call site, so a reintroduced Celery publish would ship green in all three realistic forms.

---

Lens checklist (16/16)

#	Lens	Result
1	Spec & intent	See Critical, H-a — "no business-logic change" does not hold
2	Architectural fit	See Medium — QueueBackend.CELERY still executable into a dead broker
3	Correctness	See Critical, H-a, webhook_send
4	Security	Clean — assessed directly: no authn/authz, tenant-scoping or secrets surface
5	Data integrity	See H-c. No migrations; sweep predicate verified correct for pre- and post-upgrade rows
6	Concurrency	See H-a. Diff adds no concurrency surface — barrier selection untouched
7	API & contract	See Critical (dispatcher signature), removed internal routes
8	Reliability	See H-a, H-b
9	Performance & cost	Clean — assessed directly, no hot-path change
10	Observability	LOG_TRANSPORT defaults to celery in pubsub_helper.py:42 and appears in no sample.env
11	Operational safety	See H-b, H-c, H-d — heaviest lens on this PR
12	LLM/agent	N/A — verified: zero added lines touching prompts, model config, tool schemas or evals
13	Testing	See testing group
14	Dependencies & build	Clean — verified: no dependency, lockfile or Dockerfile changes
15	Code quality	See Medium/Low
16	Doc & comment accuracy	Largest category by volume

Open questions

1. In-flight RabbitMQ work at cutover. Production runs Celery today. This release deletes the Celery Deployments, so claimed tasks drain but unclaimed queued messages have no consumer and strand permanently. Since the new code publishes nothing to RabbitMQ, keeping the Celery workers enabled for one release drains the backlog at no cost — the same posture already taken for workerLogConsumerV2.
2. Should the deprecated single-step endpoint be deleted now rather than left as an unconditional error? (UN-4051 exists.)

Assumptions

Reviewed the committed head 3e962aec8. helm-unittest 0.5.1 / helm 3.15.3, the pair CI pins.

[muhammad-ali-e]

Worked through the review findings — a501ae754.

Critical: fixed and mutation-verified

PgExecutionDispatcher.dispatch() takes no headers; three live call sites passed it, including the main structure_pipeline path. Reproduced the TypeError, then removed the argument and the now-dead _fairness_headers(). Nothing is lost — the router never forwarded headers to the PG path (its own docstring said so), and PG carries org routing in the enqueue payload.

The reason it shipped green was 12 bare MagicMock() dispatcher doubles. MagicMock(spec=...) does not fix that — I tried it first and the mutation stayed green, because spec= restricts attribute names, not signatures. Switched to create_autospec; re-adding headers= now turns 11 tests red.

Highs: fixed

• compose.debug.yaml had 15 celery -A worker invocations and zero PG consumers — renamed keys, unchanged bodies. Rewritten to wrap python -m pg_queue_consumer, with the queue set deliberately not restated (the module reads it from the base file; restating it is what let this drift). Merged debug stack now has zero Celery workers.
• PG_SCHEDULER_ADOPT_PERIODICS was set nowhere; added to the backend service.
• run-worker.sh was missing pg-notification and pg-ide-callback, plus three drifted queue lists. Synced against docker compose config — 23 queues, exact parity both ways.
• entrypoint.sh / ownership.py told an SRE a restart moves nothing. Corrected.

Plus deletions (flags.py, the gutted webhook_send() returning None from a -> str, TestFairnessHeaders), the sample.compose.override.yaml build fix, and ~15 docstrings/comments that described deleted machinery as live.

One High deliberately NOT fixed, and why

DEFAULT_WORKFLOW_TRANSPORT stays CELERY. Flipping it is the right fix and it is one line — but I ran it, and it breaks 19 tests across 6 files, every one relying on the implicit Celery default, most characterising PgBarrier's Celery-link branch. Rewriting those to state their transport explicitly is a behaviour change that deserves its own review; bundling it into a doc sweep is how regressions ship, and it is the pattern this review criticised elsewhere. The hazard, the mechanism and the exact blast radius are now documented at the constant.

Verified

Backend 1005 passed, workers 1498 passed, pre-commit clean. One honest note: an earlier run reported 6 failed / 108 errors — that was three suites racing a single Postgres (test_unstract_db does not exist), not a regression. Re-run serially it is green.

[greptile-apps]

Greptile Summary

The PR removes the PostgreSQL queue rollout flag and makes PostgreSQL the default execution and scheduling transport.

• Replaces flag-based transport selection with direct PostgreSQL queue dispatch.
• Enables PostgreSQL schedule ownership by default.
• Reconfigures the local Compose stack around PostgreSQL consumers and removes obsolete Celery workers and endpoints.
• Retains transport fields required by worker payload contracts while removing dead rollout plumbing.

Confidence Score: 5/5

The PR appears safe to merge because no blocking failure eligible for this follow-up review remains.

No blocking failure remains.

Important Files Changed

Filename	Overview
backend/workflow_manager/workflow_v2/workflow_helper.py	Makes PostgreSQL queue dispatch unconditional while preserving dispatch bookkeeping and the worker transport payload.
workers/queue_backend/routing.py	Removes rollout allow-list selection and makes PostgreSQL the default backend while retaining explicit backend overrides.
backend/scheduler/ownership.py	Removes per-schedule rollout decisions and makes PostgreSQL ownership depend only on the scheduler enablement setting.
docker/docker-compose.yaml	Promotes PostgreSQL queue consumers into the default stack and removes the replaced Celery worker services.
backend/notification_v2/notification_dispatch.py	Simplifies buffered webhook delivery to an unconditional PostgreSQL enqueue while retaining terminal-failure semantics.
backend/prompt_studio/prompt_studio_core_v2/views.py	Uses PostgreSQL task results unconditionally for Prompt Studio task-status polling.
unstract/workflow-execution/src/unstract/workflow_execution/executor_rpc.py	Removes routing-dispatcher fallback behavior so executor request-reply operations use PostgreSQL directly.

Flowchart

%%{init: {'theme': 'neutral'}}%%
flowchart LR
    A[Backend producers] --> B[PostgreSQL queue]
    B --> C[Role-specific PG consumers]
    C --> D[Registered worker tasks]
    D --> E[Workflow and notification state]
    F[Backend startup] --> G[Schedule ownership convergence]
    G --> H[PostgreSQL scheduler]
    H --> B

Loading

_{Reviews (2): Last reviewed commit: "UN-4046 [FIX] e2e: retarget the test com..." | Re-trigger Greptile}

[sonarqubecloud]

Quality Gate passed

Issues
2 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	25.1
✅	e2e-coowners	e2e	1	0	0	0	1.4
✅	e2e-etl	e2e	1	0	0	0	10.7
✅	e2e-login	e2e	2	0	0	0	1.5
✅	e2e-prompt-studio	e2e	1	0	0	0	4.5
✅	e2e-smoke	e2e	2	0	0	0	1.3
✅	e2e-workflow	e2e	1	0	0	0	20.6
✅	integration-backend	integration	310	0	0	26	47.2
✅	integration-connectors	integration	1	0	0	7	8.0
✅	integration-workers	integration	157	0	0	1	50.5
✅	unit-backend	unit	1124	0	0	1	41.4
✅	unit-connectors	unit	63	0	0	0	10.0
✅	unit-core	unit	33	0	0	0	1.4
✅	unit-platform-service	unit	15	0	0	0	2.6
✅	unit-rig	unit	117	0	0	0	5.3
✅	unit-runner	unit	5	0	0	0	2.9
✅	unit-sdk1	unit	563	0	0	0	28.8
✅	unit-workers	unit	1365	0	0	1	123.1
	TOTAL		3764	0	0	36	386.4

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