UN-3445 [GATED-FEAT] PG-queue auxiliary-worker migration (OSS): loader fix + notification seam

[muhammad-ali-e]

What

The OSS slice of the PG-queue auxiliary-worker migration wave (epic UN-3445). Two flag-gated changes, both inert with pg_queue_enabled off:

• UN-3798 — fix the PG pluggable-worker task loader (workers/worker.py): skip the broken file-path task load for pluggable workers (their tasks are already registered by build_celery_app's package import). Non-pluggable workers keep the file-path load. Fixes a startup crash-loop for cloud PG pluggable workers; the Celery path is untouched.
• UN-3753 — route the buffered webhook-notification dispatch through the PG-queue transport seam (notification_v2/notification_dispatch.py): PG when pg_queue_enabled for the org, else celery_app.send_task (byte-identical). PermanentDispatchError (raised only on the PG branch) dead-letters permanent enqueue errors; the Celery error path is unchanged.

Why

• Part of migrating the remaining dispatch/auxiliary workers off Celery onto the PG queue so Celery can eventually be decommissioned (sub-tasks of UN-3445).

Can this PR break any existing features?

• No. Both changes are gated by the pg_queue_enabled Flipt flag; with it off (the production default) resolve_transport fail-closes to Celery and the seams call the same send_task as before — byte-identical. The loader fix only skips a load path for pluggable workers (not deployed with the flag off). Dev-tested on a live stack: flag-off produced 0 PG rows (Celery taken); PG round-trips delivered end-to-end.

Notes on Testing

• notification_v2/tests/ (12 tests) pass; pipeline_dispatch regression suite green; worker enum/loader tests pass. Merged current main in (clean, no conflicts).

Related Issues or PRs

• UN-3445 (epic), UN-3798, UN-3753. Cloud counterpart PR covers UN-3752/3754/3779 + the chart. Flag-gated wave; merge-safe with the flag off.

[coderabbitai]

Note

Reviews paused

It looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

• @coderabbitai resume to resume automatic reviews.
• @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

• ▶️ Resume reviews
• 🔍 Trigger review

Summary by CodeRabbit

• New Features

  • Webhook notifications now use organization-aware delivery routing.
  • Added reliable delivery through multiple supported transport paths.

• Bug Fixes

  • Invalid numeric values are rejected before storage.
  • Permanent delivery failures are dead-lettered, while transient failures can be retried.
  • Improved worker startup validation and delivery failure logging.

• Refactor

  • Standardized worker task loading and directory naming.

• Tests

  • Expanded coverage for routing, retries, validation, and worker configuration.

Walkthrough

Webhook notifications now select PG or Celery using an organization string identifier, classify dispatch failures, and reject non-standard JSON numbers. Worker startup centralizes directory mapping, strictly loads non-pluggable tasks, and validates task registration.

Changes

Webhook dispatch and queue validation

Layer / File(s)	Summary
Transport dispatch routing backend/notification_v2/notification_dispatch.py, backend/notification_v2/tests/test_notification_dispatch.py	dispatch_webhook_notification resolves PG or Celery, preserves task arguments, shares a dispatch ID, and wraps permanent PG enqueue errors.
Organization-aware buffer dispatch backend/notification_v2/internal_api_views.py, backend/notification_v2/tests/test_send_clubbed.py	_send_clubbed resolves the organization string identifier for routing, retains the organization PK for workers, and separates dead-letter handling from transient buffer recovery.
PG queue serialization validation backend/pg_queue/producer.py, backend/pg_queue/tests/test_producer.py	PG queue serialization rejects NaN and Infinity before database insertion and logs construction failures with enqueue context.

Worker startup and diagnostics

Layer / File(s)	Summary
Worker directory mapping workers/shared/enums/worker_enums_base.py, workers/tests/test_worker_enums_directory.py	WorkerType.to_directory() centralizes directory naming, including the API_DEPLOYMENT hyphenated directory, and to_import_path() derives from it.
Worker task loading and validation workers/worker.py	Worker startup skips file loading for pluggable workers, fails on missing non-pluggable task files, dynamically loads tasks.py, and validates task registration.
Consumer diagnostics and configuration workers/queue_backend/pg_queue/consumer.py, workers/tests/test_pg_queue_consumer.py	Poison-drop logs include queue context and normalized attempt counts. Tests cover max_attempts overrides, defaults, and invalid values.

Estimated code review effort: 4 (Complex) | ~45 minutes

Sequence Diagram(s)

sequenceDiagram
  participant _send_clubbed
  participant dispatch_webhook_notification
  participant resolve_transport
  participant PGQueue
  participant Celery
  _send_clubbed->>dispatch_webhook_notification: dispatch webhook with org string id
  dispatch_webhook_notification->>resolve_transport: resolve transport
  resolve_transport-->>dispatch_webhook_notification: PG or Celery
  dispatch_webhook_notification->>PGQueue: enqueue with dispatch id
  dispatch_webhook_notification->>Celery: send task with original arguments

Loading

sequenceDiagram
  participant Worker
  participant WorkerType
  participant tasks_py
  participant CeleryRegistry
  Worker->>WorkerType: resolve worker directory
  Worker->>tasks_py: load non-pluggable tasks.py
  tasks_py->>CeleryRegistry: register tasks
  Worker->>CeleryRegistry: validate task registration

Loading

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 25.00% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly identifies the PG-queue auxiliary-worker migration and its two main changes: the loader fix and notification dispatch seam.
Description check	✅ Passed	The description covers the changes, rationale, impact, testing, and related issues; the required breakage section is filled, although several template sections are omitted.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

✨ Finishing Touches 💡 1

📝 Generate docstrings 💡

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat/UN-3445-pg-queue-aux-workers

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[muhammad-ali-e]

PR Review Toolkit — automated multi-agent review

Ran six specialist agents (Code Reviewer, Silent-Failure Hunter, Type-Design Analyzer, Test Analyzer, Comment Analyzer, Code Simplifier) plus an independent verification pass against enqueue_task / resolve_transport / is_pg_transport.

Verdict: no blocking issues. The transport-routing seam is correct, the flag-off (Celery) path is byte-identical to the prior send_task, the permanent-vs-transient error taxonomy is sound and SENDING-guarded, and the two-org-identifier contract is well-documented and pinned by tests. Everything below is an improvement, not a defect — the two [Medium] items are the only ones I'd act on before merge.

Findings are inline. Priority: 2 Medium, 6 Low, 1 Nit.

[muhammad-ali-e]

Review feedback addressed (commit 02dcdcb)

Worked through the automated review. All changes are either new-in-this-PR code or gated to the PG branch, so the flag-off (Celery) path stays byte-identical.

Addressed (9):

• [Medium] loader empty-registry — split by `is_pluggable()`: non-pluggable now `raise RuntimeError` on empty registry / missing dir / missing `tasks.py` (terminal — no late finalize binding); pluggable stays a WARN. Celery hot path unaffected (general/executor ship a `tasks.py`).
• [Medium] NaN → re-dispatch loop — `_json_safe` now uses `json.dumps(..., allow_nan=False)` so NaN/Infinity raise `ValueError` at the enqueue seam and funnel into `PermanentDispatchError`, instead of surfacing later as a permanent `django.db.DataError` at the JSONB insert. Except left narrow (transient DB errors still propagate). New unit test added. PG-branch only.
• [Low] org-id typing — `_send_clubbed` / `_penalize_render_failure` / `_dispatch_group` now type `org_id: int` to match `_org_identifier(org_pk: int)` (annotation-only).
• [Low] observability — dangling-FK org resolution bumped WARNING → `logger.error` (Sentry-routed); fail-closed-to-Celery unchanged.
• [Low/Nit] comments + short-circuit — corrected the loader docstring (import runs registration; binding completes at `app.finalize`; skip exists because a bare `"tasks"` spec breaks relative imports), the `or ""` note, and the `PermanentDispatchError` call-site comment; replaced the materialized registry list with a short-circuiting `any(...)`; re-keyed `directory_mapping` on the enum member.

Deferred (follow-up ticket): `worker.py` loader test-gap — needs a `main()`/`name` guard refactor to make the module import-safe for unit testing.

Tests: 28 backend + 3 worker green; pre-commit clean.

[greptile-apps]

Greptile Summary

The PR extends the feature-gated PostgreSQL queue migration while preserving the existing Celery fallback.

• Routes buffered webhook notifications through the organization-scoped transport resolver and adds explicit transient and permanent enqueue-failure handling.
• Normalizes PG queue payloads, preserves stable task IDs during consumption, and redacts sensitive notification data from poison-message logs.
• Updates worker task loading so pluggable workers rely on package registration while non-pluggable workers validate and load their task files directly.
• Adds focused tests for notification routing, terminal behavior, queue consumption, payload serialization, and worker-directory mapping.

Confidence Score: 5/5

The PR appears safe to merge.

No blocking failure remains.

Important Files Changed

Filename	Overview
backend/notification_v2/internal_api_views.py	Routes clubbed notification dispatch through the transport seam and adds bounded transient recovery plus permanent enqueue dead-lettering.
backend/notification_v2/notification_dispatch.py	Introduces organization-gated Celery/PG routing while adapting notification retry semantics for eager PG consumption.
backend/pg_queue/producer.py	Rejects non-finite JSON values before insertion and logs message-construction failures with dispatch context.
workers/queue_backend/pg_queue/consumer.py	Preserves producer task IDs during eager execution and redacts credentials and customer payloads from poison-drop logs.
workers/worker.py	Separates pluggable package registration from validated file-based loading for non-pluggable workers.
workers/shared/enums/worker_enums_base.py	Centralizes worker enum-to-directory mapping and rejects invalid directory resolution for pluggable worker types.

Sequence Diagram

sequenceDiagram
    participant Flush as Notification buffer flush
    participant Resolver as Transport resolver
    participant Celery as Celery broker
    participant PG as PG queue
    participant Worker as Notification worker
    participant Backend as Buffer status API

    Flush->>Resolver: Resolve using org string ID
    alt PG queue enabled
        Resolver-->>Flush: PG transport
        Flush->>PG: Enqueue notification task
        PG->>Worker: "apply(task_id, max_retries=0)"
    else Celery fallback
        Resolver-->>Flush: Celery transport
        Flush->>Celery: send_task with original kwargs
        Celery->>Worker: Execute notification task
    end
    Worker->>Backend: Mark dispatched or dead-letter

Loading

_{Reviews (9): Last reviewed commit: "Revert "UN-3893 [FIX] Make ConcurrencyMo..." | Re-trigger Greptile}

[coderabbitai]

🧹 Nitpick comments (1)

backend/notification_v2/tests/test_send_clubbed.py (1)

75-91: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Permanent-error test doesn't verify the SENDING guard on the dead-letter update.

Unlike test_transient_failure_reverts_sending_rows_to_pending, this test only checks .update() kwargs, not .filter() kwargs — so the documented status=BufferStatus.SENDING.value clobber-guard on the dead-letter path is unpinned.

♻️ Proposed addition

     def test_permanent_pg_error_dead_letters(self):
         ...
             _send()
+        fkw = buf.objects.filter.call_args.kwargs
+        assert fkw["status"] == BufferStatus.SENDING.value
         # Permanent error → terminal DEAD_LETTER, no PENDING revert / refund.
         ukw = buf.objects.filter.return_value.update.call_args.kwargs
         assert ukw == {"status": BufferStatus.DEAD_LETTER.value}

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@backend/notification_v2/tests/test_send_clubbed.py` around lines 75 - 91, The
test_permanent_pg_error_dead_letters test must also verify the SENDING status
guard used when dead-lettering. Inspect buf.objects.filter.call_args.kwargs and
assert it includes status=BufferStatus.SENDING.value, while preserving the
existing update kwargs assertion.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Nitpick comments:
In `@backend/notification_v2/tests/test_send_clubbed.py`:
- Around line 75-91: The test_permanent_pg_error_dead_letters test must also
verify the SENDING status guard used when dead-lettering. Inspect
buf.objects.filter.call_args.kwargs and assert it includes
status=BufferStatus.SENDING.value, while preserving the existing update kwargs
assertion.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 17fd0acc-5a58-48f3-98c2-17484311087e

📥 Commits

Reviewing files that changed from the base of the PR and between a642b44 and 3e7961c.

📒 Files selected for processing (10)

• backend/notification_v2/internal_api_views.py
• backend/notification_v2/notification_dispatch.py
• backend/notification_v2/tests/__init__.py
• backend/notification_v2/tests/test_notification_dispatch.py
• backend/notification_v2/tests/test_send_clubbed.py
• backend/pg_queue/producer.py
• backend/pg_queue/tests/test_producer.py
• workers/shared/enums/worker_enums_base.py
• workers/tests/test_worker_enums_directory.py
• workers/worker.py

[muhammad-ali-e]

Follow-up automated review pass (PR Review Toolkit). One net-new finding below; everything else surfaced was observability-level, a comment nit, a test-coverage gap, or on a file outside this diff — reported to the author out-of-band rather than posted here to keep the bar high.

[muhammad-ali-e]

Addressed the CodeRabbit nitpick (commit 617b2fb): test_permanent_pg_error_dead_letters now also asserts the dead-letter .update() is guarded to rows still status=SENDING (via filter.call_args.kwargs), mirroring the transient-revert test's clobber-guard assertion. 5 tests green.

[muhammad-ali-e]

Addressed the follow-up review finding (commit 1eb15c6): moved the _json_safe message construction inside enqueue_task's try/except, so a serialization ValueError (now reachable via allow_nan=False) gets the same task=/queue=/org= breadcrumb as a DB-insert failure — no more context-free permanent drop on the orchestrator path. The comment's "serialization" coverage claim is now accurate. Test asserts the breadcrumb fires and the DB insert is never reached; 16 producer tests green.

[coderabbitai]

🧹 Nitpick comments (1)

workers/tests/test_pg_queue_consumer.py (1)

275-291: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Assert the queue context added by the log change.

The test checks poison-dropped, read_ct, and org_id, but it does not check queue. Add an assertion for the expected queue value so a regression cannot remove this diagnostic context while the test remains green.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@workers/tests/test_pg_queue_consumer.py` around lines 275 - 291, The test
method test_poison_log_enriched_with_org_and_read_ct must also verify the queue
context in the poison-drop log. Add an assertion that caplog.text contains the
expected queue value for the consumer initialized with ["q"], while preserving
the existing poison-dropped, read_ct, and org_id assertions.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Nitpick comments:
In `@workers/tests/test_pg_queue_consumer.py`:
- Around line 275-291: The test method
test_poison_log_enriched_with_org_and_read_ct must also verify the queue context
in the poison-drop log. Add an assertion that caplog.text contains the expected
queue value for the consumer initialized with ["q"], while preserving the
existing poison-dropped, read_ct, and org_id assertions.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 3e6d8bfd-e63f-4ae0-8299-e93f86f4790d

📥 Commits

Reviewing files that changed from the base of the PR and between 1eb15c6 and 533cf78.

📒 Files selected for processing (2)

• workers/queue_backend/pg_queue/consumer.py
• workers/tests/test_pg_queue_consumer.py

[vishnuszipstack]

PR Review Toolkit — consolidated findings

Ran the Code Reviewer, Silent Failure Hunter, Type Design Analyzer, PR Test Analyzer, Comment Analyzer and Code Simplifier agents over the 12 changed files, then verified each surviving finding against the source. Findings already raised on this PR are omitted.

Two blockers before the flag can be turned on for any org:

1. The raise_on_final_failure=False override is unreachable for any subscriber with max_retries >= 1 — verified by running the consumer's exact task.apply(throw=True) shape. The buffers are never dead-lettered and the subscriber is re-POSTed on every vt-expiry redelivery.
2. Nothing in this repo drains the PG notifications queue — none of the seven pg-queue-consumer compose services polls it, and run-worker.sh has no notification PG role. Flag-on means every buffered webhook for that org is enqueued and silently lost.

Also flagged: an unconditional attempt-refund that makes the dispatch cap unreachable, subscriber credentials reaching Sentry and PG-at-rest via the poison-drop path, an empty-registry guard whose stated rationale is contradicted by Celery's auto-finalize, and several test/comment accuracy gaps.

Nice work on the seam tests themselves — they assert forwarded payloads and negatives rather than mock tautologies, and test_permanent_pg_error_dead_letters proving the absence of a refund via exact-dict equality is the right instinct.

[sonarqubecloud]

Quality Gate passed

Issues
1 New issue
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	16.7
✅	e2e-coowners	e2e	1	0	0	0	1.4
✅	e2e-etl	e2e	1	0	0	0	8.2
✅	e2e-login	e2e	2	0	0	0	1.2
✅	e2e-prompt-studio	e2e	1	0	0	0	4.3
✅	e2e-smoke	e2e	2	0	0	0	1.0
✅	e2e-workflow	e2e	1	0	0	0	17.9
✅	integration-backend	integration	205	0	0	26	42.9
✅	integration-connectors	integration	1	0	0	7	8.4
✅	integration-workers	integration	140	0	0	1	52.0
✅	unit-backend	unit	299	0	0	1	36.4
✅	unit-connectors	unit	63	0	0	0	9.7
✅	unit-core	unit	33	0	0	0	1.3
✅	unit-platform-service	unit	15	0	0	0	2.6
✅	unit-rig	unit	109	0	0	0	5.3
✅	unit-sdk1	unit	480	0	0	0	23.1
✅	unit-workers	unit	1335	0	0	1	94.5
	TOTAL		2691	0	0	36	327.0

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
• 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