Uns 611 clubbed notification dispatch

[kirtimanmishrazipstack]

What

Why

How

Can this PR break any existing features. If yes, please list possible items. If no, please explain why. (PS: Admins do not merge the PR without this section filled)

Database Migrations

Env Config

Relevant Docs

Related Issues or PRs

Dependencies Versions

Notes on Testing

Screenshots

Checklist

I have read and understood the Contribution Guidelines.

[coderabbitai]

Important

Review skipped

Auto reviews are disabled on base/target branches other than the default branch.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 85f0737f-49b7-494f-b12d-71a49b0d4a9a

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch UNS-611-clubbed-notification-dispatch

Tip

💬 Introducing Slack Agent: The best way for teams to turn conversations into code.

Slack Agent is built on CodeRabbit's deep understanding of your code, so your team can collaborate across the entire SDLC without losing context.

• Generate code and open pull requests
• Plan features and break down work
• Investigate incidents and troubleshoot customer tickets together
• Automate recurring tasks and respond to alerts with triggers
• Summarize progress and report instantly

Built for teams:

• Shared memory across your entire org—no repeating context
• Per-thread sandboxes to safely plan and execute work
• Governance built-in—scoped access, auditability, and budget controls

One agent for your entire SDLC. Right inside Slack.

👉 Get started

---

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 and usage tips.}

[sonarqubecloud]

Quality Gate failed

Failed conditions
2 Security Hotspots

See analysis details on SonarQube Cloud

[greptile-apps]

Greptile Summary

This PR introduces "clubbed" (batched) notification dispatch for pipelines and API deployments. When a Notification is set to BATCHED delivery mode, execution events are buffered in a new NotificationBuffer table and flushed as a single aggregated webhook message per (org, webhook_url, auth_sig) group on a configurable cadence, rather than firing one request per execution.

• Adds the NotificationBuffer model, a DeliveryMode enum, and a partial DB index on PENDING rows; a new migration covers both the new table and the delivery_mode column on Notification.
• Two new internal endpoints (buffer/enqueue/, buffer/process/) handle worker-side buffering and backend-side flush; a new process_notification_buffer.py script and a scheduler.sh update drive the flush loop alongside the existing log-history consumer.
• Frontend gains a delivery-mode selector on the notification form (with auto-IMMEDIATE when "notify on failures" is checked) and a batch-interval admin UI in Platform Settings.

Confidence Score: 3/5

The clubbed dispatch path has a window where a transient DB error after broker publish can leave rows PENDING while notifications have already been sent, leading to duplicates on the next flush tick.

The _dispatch_group function calls celery_app.send_task() inside transaction.atomic() before rows are marked DISPATCHED. A transient DB failure in the .update() call rolls back the transaction while the broker has already received the task — rows stay PENDING and are re-dispatched on the next tick, sending duplicate notifications. The settings serializer also accepts zero or negative intervals with no validation, collapsing the batch window for every future enqueue.

backend/notification_v2/internal_api_views.py (the _dispatch_group flush transaction) and backend/notification_v2/serializers.py (the settings serializer validation) warrant a second look before merging.

Important Files Changed

Filename	Overview
backend/notification_v2/internal_api_views.py	Core flush logic: adds enqueue_notification_buffer and process_notification_buffer endpoints; _dispatch_group calls celery_app.send_task() inside transaction.atomic() — a duplicate-dispatch race exists if the subsequent .update() fails before the transaction commits.
backend/notification_v2/serializers.py	Adds NotificationSettingsSerializer and delivery_mode field; club_interval_seconds is missing a min_value=1 guard, allowing a zero or negative interval to be persisted and silently collapse batch windows.
backend/notification_v2/helper.py	Adds enqueue, dispatch_with_delivery_mode, split_by_delivery_mode, and auth-sig helpers; implementation is sound, BATCHED routing correctly branches before reaching the immediate-dispatch path.
workers/shared/patterns/notification/helper.py	Adds _route_notification / _enqueue_to_buffer branching; BATCHED failures are silently dropped by design — no metric/counter emitted beyond an error log, which can obscure backend outages.
backend/notification_v2/models.py	Adds delivery_mode to Notification and new NotificationBuffer model with correct partial index on (org, url, auth_sig, flush_after) scoped to PENDING; schema looks correct.
backend/notification_v2/tasks.py	New mark_buffer_dead_letter Celery errback task; signature correctly matches Celery's link_error (request, exc, traceback) positional convention with buffer_row_ids pre-bound as a kwarg.
backend/notification_v2/clubbed_renderer.py	New renderer for batched messages; build_envelope caps at MAX_BATCH_SIZE=500 matching _PROCESS_BUFFER_CAP; Slack display truncation at 25 events with overflow footer looks correct.
backend/notification_v2/migrations/0003_add_notification_buffer.py	Migration adds delivery_mode column and NotificationBuffer table with partial index; defaults and choices align with model and enum definitions.
workers/log_consumer/process_notification_buffer.py	Thin HTTP wrapper that triggers the backend's buffer flush endpoint every scheduler tick; non-fatal failure handling is correct — logs and returns False without crashing the scheduler loop.
workers/log_consumer/scheduler.sh	Adds a second task slot for the notification buffer flush alongside log history; run_task helper catches individual failures without aborting the loop; -e removed from set flags intentionally.

Prompt To Fix All With AI

Fix the following 4 code review issues. Work through them one at a time, proposing concise fixes.

---

### Issue 1 of 4
backend/notification_v2/internal_api_views.py:429-472
**Celery task dispatched inside `transaction.atomic()` — duplicate notifications on commit failure**

`celery_app.send_task()` is called while the `transaction.atomic()` block is still open. If `send_task` succeeds but the subsequent `.update(status=DISPATCHED)` raises (e.g. a transient DB error), the entire transaction rolls back: rows remain `PENDING` while the broker has already received — and may have already executed — the webhook task. On the very next flush tick these rows are re-selected as `PENDING` and dispatched again, sending duplicate notifications.

The standard fix is to emit the Celery task only after the transaction commits, via `transaction.on_commit()`. One safe restructuring: perform the `.update()` first inside the transaction, then schedule the task via `on_commit` so that the broker is only notified once the DB state is durable.

### Issue 2 of 4
backend/notification_v2/serializers.py:11-13
No `min_value` guard on `club_interval_seconds`. A caller can `PATCH` a value of `0` or a negative integer, which passes serializer validation and is stored in the configuration table. `get_org_club_interval_seconds` returns it as-is, so `flush_after = now + timedelta(seconds=0_or_negative)` — the new row is immediately past its window and will be dispatched on the very next tick, effectively bypassing batching. The rendered message would also display `interval_minutes: 0`.

```suggestion
    # Backend ConfigSpec accepts any int; a floor of 1 prevents a zero or
    # negative value from collapsing the batch window to "immediate".
    club_interval_seconds = serializers.IntegerField(min_value=1)
```

### Issue 3 of 4
workers/shared/patterns/notification/helper.py:56-85
**Silent BATCHED notification drop on backend unavailability**

When `_enqueue_to_buffer` returns `False`, `_route_notification` returns immediately with no fallback and no metric. The execution event is lost — it won't appear in any future clubbed message nor as an immediate notification. The comment acknowledges this is intentional ("we DON'T fall back"), but there is no counter, no structured metric log, and no dead-letter record created on the worker side. During a prolonged backend outage, BATCHED notifications can silently disappear with only a single `logger.error` line as evidence.

### Issue 4 of 4
backend/notification_v2/internal_api_views.py:499-506
**`_PROCESS_BUFFER_CAP` referenced before it is defined in the module**

`_PROCESS_BUFFER_CAP` is used in `_dispatch_group` (line ~406) but defined at the module level only after that function's definition. Python resolves it at call time rather than definition time, so this doesn't raise a `NameError` at import, but any reader or static analyzer that encounters the reference first will be confused. Moving the constant above `_dispatch_group` (or at the top of the file with the other module-level constants) is the conventional placement.

_{Reviews (1): Last reviewed commit: "batch notification" | Re-trigger Greptile}

[greptile-apps]

Celery task dispatched inside transaction.atomic() — duplicate notifications on commit failure

celery_app.send_task() is called while the transaction.atomic() block is still open. If send_task succeeds but the subsequent .update(status=DISPATCHED) raises (e.g. a transient DB error), the entire transaction rolls back: rows remain PENDING while the broker has already received — and may have already executed — the webhook task. On the very next flush tick these rows are re-selected as PENDING and dispatched again, sending duplicate notifications.

The standard fix is to emit the Celery task only after the transaction commits, via transaction.on_commit(). One safe restructuring: perform the .update() first inside the transaction, then schedule the task via on_commit so that the broker is only notified once the DB state is durable.

Prompt To Fix With AI

This is a comment left during a code review.
Path: backend/notification_v2/internal_api_views.py
Line: 429-472

Comment:
**Celery task dispatched inside `transaction.atomic()` — duplicate notifications on commit failure**

`celery_app.send_task()` is called while the `transaction.atomic()` block is still open. If `send_task` succeeds but the subsequent `.update(status=DISPATCHED)` raises (e.g. a transient DB error), the entire transaction rolls back: rows remain `PENDING` while the broker has already received — and may have already executed — the webhook task. On the very next flush tick these rows are re-selected as `PENDING` and dispatched again, sending duplicate notifications.

The standard fix is to emit the Celery task only after the transaction commits, via `transaction.on_commit()`. One safe restructuring: perform the `.update()` first inside the transaction, then schedule the task via `on_commit` so that the broker is only notified once the DB state is durable.

How can I resolve this? If you propose a fix, please make it concise.

[greptile-apps]

No min_value guard on club_interval_seconds. A caller can PATCH a value of 0 or a negative integer, which passes serializer validation and is stored in the configuration table. get_org_club_interval_seconds returns it as-is, so flush_after = now + timedelta(seconds=0_or_negative) — the new row is immediately past its window and will be dispatched on the very next tick, effectively bypassing batching. The rendered message would also display interval_minutes: 0.

Suggested change

	# No min/max here: mfbt is silent on bounds. Backend ConfigSpec accepts
	# any int; constraining is a follow-up if/when product gives a number.
	club_interval_seconds = serializers.IntegerField()
	# Backend ConfigSpec accepts any int; a floor of 1 prevents a zero or
	# negative value from collapsing the batch window to "immediate".
	club_interval_seconds = serializers.IntegerField(min_value=1)

Prompt To Fix With AI

This is a comment left during a code review.
Path: backend/notification_v2/serializers.py
Line: 11-13

Comment:
No `min_value` guard on `club_interval_seconds`. A caller can `PATCH` a value of `0` or a negative integer, which passes serializer validation and is stored in the configuration table. `get_org_club_interval_seconds` returns it as-is, so `flush_after = now + timedelta(seconds=0_or_negative)` — the new row is immediately past its window and will be dispatched on the very next tick, effectively bypassing batching. The rendered message would also display `interval_minutes: 0`.

```suggestion
    # Backend ConfigSpec accepts any int; a floor of 1 prevents a zero or
    # negative value from collapsing the batch window to "immediate".
    club_interval_seconds = serializers.IntegerField(min_value=1)
```

How can I resolve this? If you propose a fix, please make it concise.

[greptile-apps]

Silent BATCHED notification drop on backend unavailability

When _enqueue_to_buffer returns False, _route_notification returns immediately with no fallback and no metric. The execution event is lost — it won't appear in any future clubbed message nor as an immediate notification. The comment acknowledges this is intentional ("we DON'T fall back"), but there is no counter, no structured metric log, and no dead-letter record created on the worker side. During a prolonged backend outage, BATCHED notifications can silently disappear with only a single logger.error line as evidence.

Prompt To Fix With AI

This is a comment left during a code review.
Path: workers/shared/patterns/notification/helper.py
Line: 56-85

Comment:
**Silent BATCHED notification drop on backend unavailability**

When `_enqueue_to_buffer` returns `False`, `_route_notification` returns immediately with no fallback and no metric. The execution event is lost — it won't appear in any future clubbed message nor as an immediate notification. The comment acknowledges this is intentional ("we DON'T fall back"), but there is no counter, no structured metric log, and no dead-letter record created on the worker side. During a prolonged backend outage, BATCHED notifications can silently disappear with only a single `logger.error` line as evidence.

How can I resolve this? If you propose a fix, please make it concise.

[greptile-apps]

_PROCESS_BUFFER_CAP referenced before it is defined in the module

_PROCESS_BUFFER_CAP is used in _dispatch_group (line ~406) but defined at the module level only after that function's definition. Python resolves it at call time rather than definition time, so this doesn't raise a NameError at import, but any reader or static analyzer that encounters the reference first will be confused. Moving the constant above _dispatch_group (or at the top of the file with the other module-level constants) is the conventional placement.

Prompt To Fix With AI

This is a comment left during a code review.
Path: backend/notification_v2/internal_api_views.py
Line: 499-506

Comment:
**`_PROCESS_BUFFER_CAP` referenced before it is defined in the module**

`_PROCESS_BUFFER_CAP` is used in `_dispatch_group` (line ~406) but defined at the module level only after that function's definition. Python resolves it at call time rather than definition time, so this doesn't raise a `NameError` at import, but any reader or static analyzer that encounters the reference first will be confused. Moving the constant above `_dispatch_group` (or at the top of the file with the other module-level constants) is the conventional placement.

How can I resolve this? If you propose a fix, please make it concise.