UN-3445 [GATED-FEAT] PG-queue auxiliary-worker migration (OSS): loader fix + notification seam - #2217
UN-3445 [GATED-FEAT] PG-queue auxiliary-worker migration (OSS): loader fix + notification seam#2217muhammad-ali-e wants to merge 3 commits into
Conversation
#2197) * UN-3798 [FIX] Skip broken file-path task load for PG pluggable workers The top-level worker.py loaded a pluggable worker's tasks.py via spec_from_file_location("tasks", ...) — a bare module name with no parent package — so the plugin's relative imports (from .clients import ...) failed with "attempted relative import with no known parent package", crash-looping every cloud PG pluggable worker (agentic_callback/UN-3754, agentic_studio/ UN-3779, bulk_download/UN-3752) on startup under `python -m pg_queue_consumer`. The tasks are already registered by that point: WorkerBuilder.build_celery_app() (called just above) verifies a pluggable type by importing pluggable_worker.{type}.worker as a proper package (_verify_pluggable_worker_exists → import_module), which runs the plugin's `from . import tasks` and registers the tasks on this same app. So the file-path load is both redundant and broken. Fix: skip the file-path load for pluggable workers; non-pluggable (top-level) workers keep it unchanged. The Celery path is unaffected — it runs via `celery -A pluggable_worker.{type}.worker` (a dotted package import) and never touches this loader; the is_pluggable() file-path branch never ran successfully. Validated on a running dev stack via `python -m pg_queue_consumer`: agentic_callback and agentic_studio now start clean ("tasks already registered … skipping file-path task load" → "ready for Celery"); general (non-pluggable) loads unchanged. Prerequisite for UN-3752 / UN-3754 / UN-3779. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * UN-3798 [FIX] Address #2197 review: accurate loader comment, to_directory(), zero-task guard - Extract the task-load block into load_worker_tasks(worker_type) with guard-clause returns (depth 3 -> 1) and an accurate docstring: pluggable tasks register via WorkerBuilder's package import and Celery binds them on app finalize; the file-path load is skipped for them because it breaks any relative imports in the plugin's tasks.py. Softened the overstated "every worker crashes" and marked the cloud-plugin `from . import tasks` example as illustrative (contract, not internals). - Add WorkerType.to_directory() as the single source for the underscore->hyphen dir mapping; to_import_path() and the file-path loader both read it (no more slicing the import path). + tests. - Add a post-load zero-task check as a WARNING (not a hard raise): pluggable tasks bind on app finalize which can be after this point, so a raise would false-positive on a correctly-configured pluggable worker (the exact regression this PR fixes). - Move `import importlib.util` to the module-top imports. Deferred (per reviewer's hotfix note): the "spec_from_file_location never called for pluggable" regression test — worker.py runs infra init + builds the Celery app at import, so there is no clean seam to exercise the loader in isolation without a larger main()-guard refactor. The extraction creates that seam for a fast-follow. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
…nsport (flag-gated) (#2198) * UN-3753 [GATED-FEAT] Route webhook notifications through PG-queue transport (flag-gated) PG-queue analogue for the buffered-webhook dispatch, gated by the pg_queue_enabled Flipt flag. Flag OFF (prod default) keeps the existing Celery send_task byte-unchanged; flag ON enqueues send_webhook_notification onto the PG `notifications` queue, drained by the PG notification consumer. - Dispatch seam: notification_dispatch.py routes send_webhook_notification through resolve_transport — PG via enqueue_task when enabled for the org, else the prior celery_app.send_task (byte-identical, zero regression). _send_clubbed (the buffer-flush path) uses the seam; _org_identifier resolves the org pk -> string for the Flipt decision, keeping the pk in kwargs (the buffer/worker mark contract). + test_notification_dispatch.py (6 cases). - Sites 2 & 3 (WebhookSend / WebhookBatch internal endpoints) stay on Celery: they use countdown stagger, which the PG queue has no delayed-visibility for. Follow-up. Dev-tested on a live stack: flag-off took the Celery branch (0 PG rows); an enqueued send_webhook_notification was drained by the pg_queue_consumer (WORKER_TYPE=notification), delivered to a sink (200), and the row acked. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * UN-3753 [GATED-FEAT] Address #2198 review: error-classing, org-id naming, call-site tests - Robustness (P1): _send_clubbed's broad `except` mislabeled permanent dispatch errors as broker_failure and reverted to PENDING → retry-forever + mislabeled Sentry tracebacks until the attempt cap. Branch on class: ValueError/TypeError (enqueue_task validation / payload serialization) → dead-letter now with a distinct `result=dispatch_error` metric; keep revert-to-PENDING only for genuine transport/broker exceptions. - Type design (P2): renamed the seam's routing param organization_id → org_string_id so it can't be conflated with the org pk in `kwargs["organization_id"]` (the worker buffer-mark contract) — swapping them was a silent mis-route to Celery. - Observability (P2): _org_identifier now logs a warning (with org_pk) when the lookup returns None — a dangling FK is a data anomaly (CASCADE makes it otherwise unreachable), not an expected "org deleted" path; reworded the docstring and narrowed org_pk: int. Fixed the "before the webhook HTTP call" wording. - Comment accuracy (P2): dropped the aspirational "usable by callers that surface it in an API response" from the Returns block (the sole caller discards it). - Simplify (P3): dropped the redundant `or None` on the routing arg (resolve_transport already normalizes falsy); kept the load-bearing `or ""` on enqueue_task's org_id. - Tests (P1): new test_send_clubbed.py locks the two-org-identifier contract (string routes, pk in kwargs), the transient→PENDING vs permanent→DEAD_LETTER recovery split, and _org_identifier (string id / None+warn). 11 tests pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * UN-3753 [GATED-FEAT] Gate notification dead-letter classing to the PG path only Keep the flag-off (Celery) error flow byte-identical, per the "no change to the Celery path unless flag-gated" rule. The seam now raises a typed PermanentDispatchError ONLY on the PG branch (when enqueue_task rejects the message for a permanent reason — priority/exclusivity validation or a payload that won't JSON-serialize). _send_clubbed dead-letters on that exception; every other failure — including any Celery send_task error — falls to the transient PENDING branch exactly as before UN-3753. + seam test that a permanent enqueue ValueError/TypeError is wrapped; the call-site test now raises PermanentDispatchError (the real contract) instead of a raw ValueError. 12 tests pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
|
Important Review skippedDraft detected. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
✨ Finishing Touches🧪 Generate unit tests (beta)
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. Comment |
|
muhammad-ali-e
left a comment
muhammad-ali-e
left a comment
There was a problem hiding this comment.
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.
| # (@shared_task / connect_on_app_finalize), which can be after this point, so a | ||
| # raise here would false-positive on a correctly-configured pluggable worker. | ||
| _registered_tasks = [name for name in app.tasks if not name.startswith("celery.")] | ||
| if not _registered_tasks: |
There was a problem hiding this comment.
[Medium — silent failure] The empty-registry check only ever warns, but for a non-pluggable worker this state is terminal, not transient. load_worker_tasks loads a non-pluggable tasks.py synchronously above (or early-returns on a missing dir / tasks.py at L474 / L482), so an empty registry here has no late app.finalize() binding coming — the worker boots a healthy-looking Celery process with zero registered tasks, silently rejects every delivered message, and L509 still logs ✅ Successfully loaded. To k8s the pod is Ready. The WARNING-not-raise is justified only for pluggable workers (whose @shared_tasks bind on finalize).
Suggest splitting severity by is_pluggable(): warn for pluggable, raise RuntimeError(...) for non-pluggable — and likewise hard-fail the missing-dir / missing-tasks.py early-returns (L474 / L482) rather than return.
| org_id=org_string_id or "", | ||
| task_id=dispatch_id, | ||
| ) | ||
| except (ValueError, TypeError) as exc: |
There was a problem hiding this comment.
[Medium — error classification] except (ValueError, TypeError) catches only Python-layer permanent failures, but a permanent DB-layer rejection is neither type. Concretely: _json_safe uses json.dumps(default=str), which permits NaN/Infinity, so a webhook body carrying a NaN float round-trips un-sanitized and Postgres JSONB rejects it at insert with django.db.DataError. That is permanent (fails identically every flush tick) yet slips past this catch → the caller's generic except Exception reverts to PENDING and re-renders/re-dispatches until the attempt cap, logging a misleading result=broker_failure each tick — the exact loop PermanentDispatchError exists to prevent, just for DB-layer errors instead of Python-layer ones.
Fix options: json.dumps(..., allow_nan=False) inside _json_safe (so it raises ValueError at the intended seam), and/or widen the wrap to include django.db.DataError/IntegrityError while leaving OperationalError/InterfaceError (genuinely transient) to propagate. Self-heals at the cap, hence Medium not High.
| "organization_id": org_id, | ||
| }, | ||
| queue="notifications", | ||
| org_string_id=_org_identifier(org_id), |
There was a problem hiding this comment.
[Low — type safety] The PR's central invariant is that the two org identifiers must never be conflated: the org string id routes Flipt (org_string_id), while the org pk stays in kwargs['organization_id']. That's well expressed (keyword-only param, divergent naming, strong docstrings) but weakly enforced: _send_clubbed (L557), _penalize_render_failure (L653) and _dispatch_group (L673) all type org_id: Any, and resolve_transport does str(organization_id), so an accidental pk↔string swap silently mis-routes rather than failing loudly. Cheapest hardening: type those three params org_id: int to match _org_identifier(org_pk: int), letting mypy flag a mix-up along the whole caller chain.
| } | ||
| directory = directory_mapping.get(self.value, self.value) | ||
| return f"{directory}.tasks" | ||
| return directory_mapping.get(self.value, self.value) |
There was a problem hiding this comment.
[Low — enforcement] directory_mapping keys on self.value (the string "api_deployment"), so if an enum value were ever renamed the key would fall through .get(self.value, self.value) and silently return the wrong directory. The sibling to_health_port keys its port_mapping on the enum member (WorkerType.API_DEPLOYMENT). Keying this dict the same way ties the override to member identity and removes the stringly-typed rename hazard — one-line change, no behavior change.
| queue=queue, | ||
| args=args, | ||
| kwargs=kwargs, | ||
| # enqueue_task's org_id is str-typed — coerce None→"" (unlike the |
There was a problem hiding this comment.
[Low — comment accuracy] This comment calls the or "" "load-bearing (unlike the routing arg above)", but enqueue_task already re-coerces internally — producer.py:130 does org_id=org_id or "" — so passing None straight through would produce an identical row. The or "" is needed only for static-type correctness (enqueue_task's param is typed org_id: str), not at runtime; the claimed runtime asymmetry with the routing arg doesn't exist. Suggest: # enqueue_task is typed org_id: str; pass "" not None to satisfy the type (it re-coerces internally anyway).
| proper PACKAGE import that runs the plugin's own task-registration code (e.g. | ||
| ``from . import tasks``, which may use relative imports such as | ||
| ``from .clients import ...``). Celery binds those tasks to the worker's app | ||
| when it finalizes, so they are registered by this point. The generic file-path |
There was a problem hiding this comment.
[Low — comment accuracy] This docstring asserts pluggable tasks are "registered by this point", but the empty-registry warning ~40 lines below (L497) correctly says they "bind on app finalize … which can be after this point." Both can't be true — @shared_task / connect_on_app_finalize binding is deferred to app.finalize(). The skip is still correct, but the real justification is the relative-import breakage (loading tasks.py under a bare "tasks" spec), not that tasks are already bound. Suggest aligning the docstring with the sibling comment: the package import runs the plugin's registration code; binding completes at finalize.
| len(buffer_ids), | ||
| ) | ||
| except PermanentDispatchError: | ||
| # PG-ONLY permanent failure — enqueue_task validation (priority range / |
There was a problem hiding this comment.
[Low — comment accuracy] This call-site comment lists "priority range / reply_key+callback exclusivity" as the permanent failure modes, but dispatch_webhook_notification passes neither priority (defaults in-range) nor reply_key/callbacks, so those enqueue_task ValueErrors are unreachable from this path. The only reachable permanent failure here is payload JSON-serialization. Suggest narrowing this comment to the reachable mode — the general list is fine to keep on the PermanentDispatchError class docstring.
| .first() | ||
| ) | ||
| if org_string_id is None: | ||
| logger.warning( |
There was a problem hiding this comment.
[Low — observability] A None here is documented as a dangling FK — DB corruption "unreachable in normal operation" under on_delete=CASCADE. A genuine data-integrity anomaly ops must investigate is better surfaced as logger.error (Sentry-routed) than a WARNING no alert is keyed on. Failing closed to Celery is the right behavior; just consider bumping the severity so it's actually noticed.
| Non-pluggable (top-level) workers use absolute imports; their ``tasks.py`` is | ||
| loaded by file path with the worker directory on ``sys.path``. | ||
| """ | ||
| if worker_type.is_pluggable(): |
There was a problem hiding this comment.
[Low — test gap] This pluggable-skip early-return is the core fix the PR ships (it avoids the bare-"tasks" spec that breaks a plugin's relative imports), yet it has no direct test — only to_directory() is covered. If a future edit drops or inverts this guard, no test fails and every pluggable worker regresses to "attempted relative import with no known parent package" at startup (caught only by the soft warning). The blocker is that load_worker_tasks is invoked at module scope alongside heavy side effects, so import worker boots a real worker. Consider guarding the module-level orchestration behind a main() / __name__ guard so the function is importable and can be unit-tested (assert exec_module is never called for a pluggable type).
| # as a WARNING, not a hard failure: pluggable tasks bind on app finalize | ||
| # (@shared_task / connect_on_app_finalize), which can be after this point, so a | ||
| # raise here would false-positive on a correctly-configured pluggable worker. | ||
| _registered_tasks = [name for name in app.tasks if not name.startswith("celery.")] |
There was a problem hiding this comment.
[Nit] _registered_tasks is materialized solely to test emptiness. if not any(not n.startswith("celery.") for n in app.tasks): short-circuits and drops the throwaway list. Purely optional — the named list reads fine as-is.
1 participant
What
The OSS slice of the PG-queue auxiliary-worker migration wave (epic UN-3445). Two flag-gated changes, both inert with
pg_queue_enabledoff:workers/worker.py): skip the broken file-path task load for pluggable workers (their tasks are already registered bybuild_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.notification_v2/notification_dispatch.py): PG whenpg_queue_enabledfor the org, elsecelery_app.send_task(byte-identical).PermanentDispatchError(raised only on the PG branch) dead-letters permanent enqueue errors; the Celery error path is unchanged.Why
Can this PR break any existing features?
pg_queue_enabledFlipt flag; with it off (the production default)resolve_transportfail-closes to Celery and the seams call the samesend_taskas 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_dispatchregression suite green; worker enum/loader tests pass. Merged currentmainin (clean, no conflicts).Related Issues or PRs