✨(workers) upgrade from Celery to Dramatiq - #772
Conversation
This is a rebase/rework of #560
📝 WalkthroughWalkthroughThe application migrates background processing from Celery to Dramatiq with Redis Streams. It adds shared task registration, queue routing, scheduling, progress and ownership tracking, a Django admin dashboard, dedicated workers, legacy-table cleanup, and updated tests and documentation. ChangesDramatiq task platform
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant API
participant TaskUtils
participant RedisStreams
participant Worker
participant ResultBackend
API->>TaskUtils: dispatch registered task
TaskUtils->>RedisStreams: enqueue message with queue and priority
RedisStreams->>Worker: deliver task message
Worker->>ResultBackend: store result and progress
API->>ResultBackend: retrieve task status
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 18
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
src/backend/core/tests/test_worker.py (1)
331-338: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winThis test now always costs 10 seconds.
Success is defined by
TimeoutExpiredfiring, socommunicate(timeout=10)never returns early on the happy path. Raising the value from 3 to 10 makes every green run 10 seconds slower. Poll for liveness instead, and only read the output when the process has already exited.♻️ Proposed refactor
try: - # Wait briefly for startup - if it crashes immediately, we'll know - # Use communicate with timeout to capture output - try: - stdout, _ = process.communicate(timeout=10) - except subprocess.TimeoutExpired: - # Worker is still running after 10 seconds - this is expected - stdout = "" + # Poll for an early crash instead of waiting out a fixed timeout: + # a worker still running after the grace period is the success case. + deadline = time.monotonic() + 10 + while time.monotonic() < deadline and process.poll() is None: + time.sleep(0.2) + stdout = "" if process.poll() is None else process.communicate()[0]Add
import timealongside the existingimport subprocess.🤖 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 `@src/backend/core/tests/test_worker.py` around lines 331 - 338, Update the worker startup test around process.communicate to poll for process liveness with a short interval instead of waiting for a 10-second communicate timeout. Import and use time for the polling loop, treating a still-running process as success and only calling communicate to capture stdout after the process exits.src/backend/core/api/viewsets/calendar.py (1)
152-167: 🗄️ Data Integrity & Integration | 🔴 Critical | 🏗️ Heavy liftDecouple
track_owner()failures from.delay()failures in both calendar task views. BothCalendarRsvpView.postandCalendarAddEventView.postwraptask.track_owner(request.user.id)in the sametryblock as the preceding.delay()call.track_owner()only records ownership metadata after the task is already enqueued; if it fails, theexceptblock still reports a 503 "could not schedule" error even though the CalDAV side effect (RSVP reply / event creation) already happened. A client that retries on that false 503 dispatches the non-idempotent write a second time.
src/backend/core/api/viewsets/calendar.py#L152-L167: split into twotry/exceptblocks — keep the 503 response tied only tocalendar_rsvp_task.delay(...)failing, and log (without failing the request) if the subsequenttask.track_owner(request.user.id)raises.src/backend/core/api/viewsets/calendar.py#L225-L238: apply the identical split tocalendar_add_event_task.delay(...)and itstask.track_owner(request.user.id)call.🤖 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 `@src/backend/core/api/viewsets/calendar.py` around lines 152 - 167, In src/backend/core/api/viewsets/calendar.py lines 152-167, split task.track_owner(request.user.id) out of the try block surrounding calendar_rsvp_task.delay(...); return the 503 response only when dispatch fails, and log ownership-tracking failures without failing the request. Apply the identical separation in src/backend/core/api/viewsets/calendar.py lines 225-238 for calendar_add_event_task.delay(...) and task.track_owner(request.user.id).
🤖 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.
Inline comments:
In `@compose.yaml`:
- Around line 49-51: Update the Redis Streams broker service in compose.yaml to
start Redis with maxmemory-policy noeviction and AOF persistence enabled, and
mount a named volume at /data. Declare the redis-rediss-data named volume in the
top-level volumes section while preserving the existing redis:8.2 image and port
configuration.
In `@docs/architecture.md`:
- Line 47: Correct the wording for the “Background tasks” entries in
docs/architecture.md: remove the duplicated “tasks” in the OpenSearch sentence
and use a plural verb at the corresponding email-flow entry on line 55.
In `@docs/push-notifications.md`:
- Line 381: Add a blank line immediately after the “Task model — one background
task per notification” heading so the following paragraph is separated according
to Markdown formatting requirements.
In `@src/backend/core/mda/inbound_tasks.py`:
- Around line 61-66: Update the inbound task lock handling around
_INBOUND_TASK_LOCK_TTL and the TaskTimeLimitExceeded path so a live task cannot
lose its per-message lock when timeout delivery is delayed. Refresh the lock
periodically during task execution, or re-assert/extend it in the timeout
handler immediately before _retry_or_abandon persists the retry or abandonment;
preserve TTL cleanup for dead workers and the existing retry behavior.
In `@src/backend/core/services/push/tasks.py`:
- Around line 173-180: Add durable idempotency to the push task’s send flow by
recording and enforcing a unique delivery key composed of channel_id and
message_id before invoking send_push_notification(). Reuse the existing delivery
record when the same task is replayed, and pass a provider-supported idempotency
key where available so accepted requests cannot produce duplicate notifications
after a worker crash.
In `@src/backend/core/task_utils.py`:
- Line 210: Make result storage opt-in for the high-volume inbound delivery and
search reindex task declarations by setting store_results=False, while leaving
it enabled for tasks used by the task-status endpoint. Do not change the global
task result behavior or TTL unless needed for these specific task registrations.
In `@src/backend/core/tests/api/test_import_file_upload.py`:
- Around line 73-81: Convert the test helper `_stub_result` into a
`contextlib.contextmanager` that starts the `Message` patch, configures
`message_cls.return_value.get_result`, yields control, and reliably stops the
patch in `finally`. Update all five callers to use `with self._stub_result(...)`
and remove their duplicated try/finally patch cleanup.
In `@src/backend/core/tests/test_task_dashboard.py`:
- Around line 112-125: Add a test alongside
test_a_staff_get_is_proxied_through_with_status_and_headers that posts to the
dashboard endpoint with the same-origin HTTP_ORIGIN, uses an active staff user,
and verifies the WSGI app receives POST and returns a successful response. This
should exercise the combined method and origin guard in _make_view.
- Around line 21-27: Update test_not_mounted_without_the_streams_broker to
explicitly patch dramatiq.get_broker with a non-Streams broker, matching the
setup used by test_mounted_under_the_broker_supports_it, before asserting the
dashboard URL patterns are empty.
In `@src/backend/core/tests/test_task_utils.py`:
- Around line 235-243: Update the test around _report_progress_task to create
the user with factories.UserFactory() instead of
get_user_model().objects.create, preserving the existing user reference for
task.track_owner(user.id). Move the get_user_model and APIClient imports out of
the test and into module-level imports, removing them locally if no longer used.
- Around line 262-279: Update the test containing _explodes and _also_runs to
accept pytest’s monkeypatch fixture, replace the manual _shutdown_callbacks
save/assignment/try-finally restoration with monkeypatch.setattr, and remove the
protected-access disables associated with that manual setup.
In `@src/backend/core/tests/test_worker.py`:
- Around line 45-59: In src/backend/core/tests/test_worker.py lines 45-59, add a
shared actor-registry guard that imports core.tasks, asserts the broker actor
list is non-empty, and ensure the priority test compares at least one defined
actor so the queue loop cannot become a no-op. Apply the same guard before
building offenders at lines 111-139 and assert declared is non-empty before the
subset check at lines 158-168; reuse the helper across all three tests.
- Around line 16-24: Update the queue-definition assertions in the worker tests
to explicitly assert that worker.ALL_QUEUES and the imported
core.task_utils.ALL_QUEUES are equal, while retaining the existing literal-list
and declared-queue checks. Anchor the new assertion near the existing ALL_QUEUES
import and comparison so divergence between the CLI and task registration
definitions is detected.
- Around line 187-197: Update test_parse_args_defaults to remove or unset
WORKER_THREADS before calling self._parse(), ensuring the test exercises the
parser’s default of one thread regardless of the surrounding environment.
- Around line 254-261: The test_queue_order_preserved_after_exclusion assertion
should derive its expected queues from ALL_QUEUES rather than hardcoding the
full list. Filter out “outbound” and “imports” while retaining ALL_QUEUES order,
then compare _resolve’s result against that derived sequence; leave full
queue-content validation to test_all_queues_defined_in_priority_order.
In `@src/backend/messages/__init__.py`:
- Around line 10-11: Remove the DJANGO_CONFIGURATION defaulting from the package
initialization, leaving only the DJANGO_SETTINGS_MODULE setup so missing
configuration continues to fail loudly. If the Dramatiq CLI re-exec path
requires a configuration, set it explicitly in that path rather than restoring a
package-level default.
In `@src/backend/messages/settings.py`:
- Around line 1186-1202: Add dramatiq.results.Results to the middleware list
returned by the settings middleware configuration in
src/backend/messages/settings.py:1186-1202, and update EagerBroker.enqueue in
src/backend/core/task_utils.py:255-266 to invoke after_process_message so stored
results are available in eager execution.
In `@src/backend/pyproject.toml`:
- Around line 49-52: Update the dramatiq-redis-streams dependency entry in the
project dependency list to use a direct URL reference pinned to the intended git
revision, preventing fallback to PyPI; retain the existing [tool.uv.sources]
entry for local development and apply the same change to the corresponding
dependency entry noted in the review.
---
Outside diff comments:
In `@src/backend/core/api/viewsets/calendar.py`:
- Around line 152-167: In src/backend/core/api/viewsets/calendar.py lines
152-167, split task.track_owner(request.user.id) out of the try block
surrounding calendar_rsvp_task.delay(...); return the 503 response only when
dispatch fails, and log ownership-tracking failures without failing the request.
Apply the identical separation in src/backend/core/api/viewsets/calendar.py
lines 225-238 for calendar_add_event_task.delay(...) and
task.track_owner(request.user.id).
In `@src/backend/core/tests/test_worker.py`:
- Around line 331-338: Update the worker startup test around process.communicate
to poll for process liveness with a short interval instead of waiting for a
10-second communicate timeout. Import and use time for the polling loop,
treating a still-running process as success and only calling communicate to
capture stdout after the process exits.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: dbc3d69b-8196-43f9-91dc-6eb4b0d1ad5c
⛔ Files ignored due to path filters (1)
src/backend/uv.lockis excluded by!**/*.lock
📒 Files selected for processing (78)
CHANGELOG.mdMakefileProcfileREADME.mdcompose.yamldeploy/env/backend.e2edocs/architecture.mddocs/env.mddocs/imports.mddocs/mobile.mddocs/push-notifications.mddocs/search-indexation.mddocs/selfcheck.mddocs/tiered-storage.mddocs/worker.mdsrc/backend/core/api/serializers.pysrc/backend/core/api/viewsets/calendar.pysrc/backend/core/api/viewsets/send.pysrc/backend/core/api/viewsets/submit.pysrc/backend/core/api/viewsets/task.pysrc/backend/core/management/commands/re_store_blobs.pysrc/backend/core/management/commands/run_task.pysrc/backend/core/mda/dispatch_webhooks.pysrc/backend/core/mda/inbound_pipeline.pysrc/backend/core/mda/inbound_tasks.pysrc/backend/core/mda/outbound.pysrc/backend/core/mda/outbound_tasks.pysrc/backend/core/migrations/0035_drop_celery_tables.pysrc/backend/core/services/blob_gc.pysrc/backend/core/services/calendar/tasks.pysrc/backend/core/services/dns/tasks.pysrc/backend/core/services/exporter/tasks.pysrc/backend/core/services/importer/imap.pysrc/backend/core/services/importer/mbox.pysrc/backend/core/services/importer/tasks.pysrc/backend/core/services/importer/utils.pysrc/backend/core/services/push/__init__.pysrc/backend/core/services/push/common.pysrc/backend/core/services/push/tasks.pysrc/backend/core/services/search/coalescer.pysrc/backend/core/services/search/exceptions.pysrc/backend/core/services/search/index.pysrc/backend/core/services/search/tasks.pysrc/backend/core/services/tiered_storage_tasks.pysrc/backend/core/signals.pysrc/backend/core/task_dashboard.pysrc/backend/core/task_utils.pysrc/backend/core/tasks.pysrc/backend/core/templates/admin/index.htmlsrc/backend/core/tests/api/test_calendar.pysrc/backend/core/tests/api/test_draft_attachments.pysrc/backend/core/tests/api/test_import_file_upload.pysrc/backend/core/tests/exporter/test_export_task.pysrc/backend/core/tests/mda/test_arc.pysrc/backend/core/tests/mda/test_dispatch_webhooks.pysrc/backend/core/tests/mda/test_inbound_auth.pysrc/backend/core/tests/mda/test_inbound_spoofed_sender.pysrc/backend/core/tests/mda/test_outbound_e2e.pysrc/backend/core/tests/mda/test_retry.pysrc/backend/core/tests/mda/test_spam_processing.pysrc/backend/core/tests/search/test_e2e.pysrc/backend/core/tests/search/test_e2e_modifiers.pysrc/backend/core/tests/search/test_search.pysrc/backend/core/tests/search/test_search_reindex_command.pysrc/backend/core/tests/tasks/test_task_send_message.pysrc/backend/core/tests/tasks/test_tiered_storage_tasks.pysrc/backend/core/tests/test_push.pysrc/backend/core/tests/test_signals.pysrc/backend/core/tests/test_task_dashboard.pysrc/backend/core/tests/test_task_utils.pysrc/backend/core/tests/test_worker.pysrc/backend/core/utils.pysrc/backend/messages/__init__.pysrc/backend/messages/celery_app.pysrc/backend/messages/settings.pysrc/backend/messages/urls.pysrc/backend/pyproject.tomlsrc/backend/worker.py
💤 Files with no reviewable changes (1)
- src/backend/messages/celery_app.py
| # The Streams broker needs Redis >= 7 (consumer groups on streams | ||
| # with per-entry deletion). Keep this in step with production. | ||
| image: redis:8.2 |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n -C 6 'redis:|appendonly|maxmemory-policy|redis-data|/data' compose.yamlRepository: suitenumerique/messages
Length of output: 4278
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== compose.yaml redis service =="
sed -n '34,56p' compose.yaml
echo "== top-level volumes in compose.yaml =="
sed -n '590,640p' compose.yaml || true
echo "== occurrences of required persistence docs/settings across tracked files =="
rg -n -C 3 'AOF|appendonly|maxmemory-policy|redis-streams|Streams broker|at-least-once|at-least-once delivery' -S --glob '!*.lock' --glob '!node_modules/**' --glob '!dist/**' .
echo "== deterministic docker compose config probe =="
python3 - <<'PY'
import yaml
from pathlib import Path
p=Path('compose.yaml')
data=yaml.safe_load(p.read_text())
svc=data['services']['redis']
print({
'redis_command': svc.get('command'),
'redis_volumes': svc.get('volumes'),
'top_level_volumes': data.get('volumes'),
})
for cfg_name in ['redis'].append(['redis-persistence'] if False else []):
if cfg_name in data['services']:
print(cfg_name)
for k in svc:
print(k, svc[k])
PYRepository: suitenumerique/messages
Length of output: 14891
Configure Redis persistence for the Streams broker.
docs/workers.md, CHANGELOG.md, and TASK_BROKER_URL require Redis maxmemory-policy noeviction plus AOF persistence. compose.yaml only declares redis:8.2 and the port, so container restarts can lose stream tasks. Add the Redis server flags and a named volume for /data.
Proposed configuration
redis:
image: redis:8.2
+ command:
+ - redis-server
+ - --appendonly
+ - "yes"
+ - --maxmemory-policy
+ - noeviction
+ volumes:
+ - redis-data:/dataDeclare redis-rediss-data in the top-level volumes section.
🤖 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 `@compose.yaml` around lines 49 - 51, Update the Redis Streams broker service
in compose.yaml to start Redis with maxmemory-policy noeviction and AOF
persistence enabled, and mount a named volume at /data. Declare the
redis-rediss-data named volume in the top-level volumes section while preserving
the existing redis:8.2 image and port configuration.
| 2. **MTA-In** validates recipients against Django backend | ||
| 3. **MDA** parses and stores messages in PostgreSQL | ||
| 4. **Celery** tasks index content in OpenSearch | ||
| 4. **Background tasks** tasks index content in OpenSearch |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Fix the task grammar in the email flow.
Line 47 repeats tasks. Line 55 uses a singular verb with the plural subject Background tasks.
Proposed wording
-4. **Background tasks** tasks index content in OpenSearch
+4. **Background tasks** index content in OpenSearch
-4. **Background tasks** processes sending via **MTA-Out**
+4. **Background tasks** process sending via **MTA-Out**Also applies to: 55-55
🤖 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 `@docs/architecture.md` at line 47, Correct the wording for the “Background
tasks” entries in docs/architecture.md: remove the duplicated “tasks” in the
OpenSearch sentence and use a plural verb at the corresponding email-flow entry
on line 55.
| ## 12. Delivery & operations (as implemented) | ||
|
|
||
| ### Task model — one Celery task per notification | ||
| ### Task model — one background task per notification |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Add a blank line after this heading.
The paragraph starts immediately after the heading. This violates MD022.
Proposed fix
### Task model — one background task per notification
+
`enqueue_push_notifications` (on commit) schedules `send_push_for_message`, the🧰 Tools
🪛 markdownlint-cli2 (0.23.1)
[warning] 381-381: Headings should be surrounded by blank lines
Expected: 1; Actual: 0; Below
(MD022, blanks-around-headings)
🤖 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 `@docs/push-notifications.md` at line 381, Add a blank line immediately after
the “Task model — one background task per notification” heading so the following
paragraph is separated according to Markdown formatting requirements.
Source: Linters/SAST tools
| # The per-message lock must outlive the time limit. On a clean (or timed-out) | ||
| # exit the ``finally`` releases it immediately; on a worker OOM/kill the lock is | ||
| # freed only by this TTL. Setting it past the time limit means a *live* task can | ||
| # never have its lock stolen, while a *dead* task's lock frees ~a minute later | ||
| # so the 5-min sweep can retry. | ||
| _INBOUND_TASK_LOCK_TTL = _INBOUND_TASK_TIME_LIMIT + 60 |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
The lock TTL no longer guarantees that a live task keeps its lock.
Under Celery the hard limit terminated the worker process, so the 60-second margin between the limit and the lock TTL was a real bound. Dramatiq's TimeLimit middleware instead raises TaskTimeLimitExceeded asynchronously in the worker thread. That interrupt is only delivered at a bytecode boundary, so a thread blocked inside a C call (a socket read, a long DB statement) receives it late. Nothing terminates the task at 600 seconds.
If delivery slips past 660 seconds, the lock expires while the task is still running, and the 5-minute sweep can dispatch a second concurrent task for the same InboundMessage row. The comment states the opposite as a safety property.
Refresh the lock while the task runs, or re-assert it in the TaskTimeLimitExceeded handler before _retry_or_abandon writes the row.
🤖 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 `@src/backend/core/mda/inbound_tasks.py` around lines 61 - 66, Update the
inbound task lock handling around _INBOUND_TASK_LOCK_TTL and the
TaskTimeLimitExceeded path so a live task cannot lose its per-message lock when
timeout delivery is delayed. Refresh the lock periodically during task
execution, or re-assert/extend it in the timeout handler immediately before
_retry_or_abandon persists the retry or abandonment; preserve TTL cleanup for
dead workers and the existing retry behavior.
| single-item batch to that platform's sender. On a *transient* failure | ||
| (429 / 5xx / network) it raises :class:`PushTransientError` so Celery retries | ||
| just this notification with exponential backoff; retrying is idempotent | ||
| (429 / 5xx / network) it raises :class:`PushTransientError` so just this | ||
| notification is retried with exponential backoff; retrying is idempotent | ||
| on-device because the collapse key / Topic coalesces it onto the same | ||
| notification. Dead-token devices are deleted inside the sender; permanent | ||
| rejections (bad payload, auth) end the task. ``acks_late`` means a worker | ||
| crash re-runs this one push (again collapse-deduped), not the whole fan-out. | ||
| rejections (bad payload, auth) end the task. Delivery is at-least-once, so | ||
| a worker crash re-runs this one push (again collapse-deduped), not the | ||
| whole fan-out. |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n -C 8 'collapse_key|apns-collapse-id|Topic|idempot|PushDelivery|message_id' \
src/backend/core/services/pushRepository: suitenumerique/messages
Length of output: 35023
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== tasks.py relevant sections =="
sed -n '1,240p' src/backend/core/services/push/tasks.py | cat -n
echo
echo "== senders around request/response handling =="
sed -n '1,220p' src/backend/core/services/push/apns.py | cat -n
echo
sed -n '80,220p' src/backend/core/services/push/fcm.py | cat -n
echo
sed -n '160,280p' src/backend/core/services/push/webpush.py | cat -n
echo
echo "== Django/ORM models mentioning push channels or delivery =="
rg -n -C 5 'class .*Channel|PushPlatformChoices|PushDelivery|Push.*Delivery|delivery|idempot|collapse|topic|apns-collapse|collapse_key' src/backend/core -g '*.py'
echo
echo "== migrations around push/channel models =="
rg -n -C 4 'class .*Channel|PushPlatformChoices|PushDelivery|Push.*Delivery' src/backend -g '*.py'Repository: suitenumerique/messages
Length of output: 50380
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== targeted task/senders slices =="
sed -n '168,218p' src/backend/core/services/push/tasks.py | cat -n
sed -n '87,200p' src/backend/core/services/push/apns.py | cat -n
sed -n '87,200p' src/backend/core/services/push/fcm.py | cat -n
sed -n '165,270p' src/backend/core/services/push/web.py || true
sed -n '165,275p' src/backend/core/services/push/webpush.py | cat -n
echo
echo "== focused model/schema references =="
rg -n -C 3 '\.objects\.get\(id=channel_id|PushPlatformChoices|class.*Channel|push_platform|push_delivery|delivery|idempot|collapse_key|apns-collapse-id|topic' \
src/backend/core/services/push \
src/backend/core/models.py \
src/backend/core/migrations \
--glob '!**/__pycache__/**'Repository: suitenumerique/messages
Length of output: 50379
🌐 Web query:
APNs HTTP/2 provider keys collapse-id semantics pending notification replacement
💡 Result:
In the Apple Push Notification service (APNs) HTTP/2 API, the apns-collapse-id header is the mechanism used to coalesce, or replace, multiple pending notifications into a single notification on the user's device [1][2][3]. When you send a notification request to APNs with a specific apns-collapse-id value, the system treats notifications with the same identifier as a group [1][4]. If a device is offline or the notifications have not yet been delivered, APNs coalesces these requests, typically resulting in only the most recent notification being presented to the user [2][3][5]. Key semantics and implementation details include: * Header Usage: The apns-collapse-id must be sent as an HTTP/2 request header [6][7]. It is not part of the JSON notification payload [6]. * Size Constraint: The value of this header must not exceed 64 bytes [1][2][4]. * Coalescing Behavior: When multiple notifications with the same collapse identifier are sent, they are merged into a single notification for the user [1][2]. This is particularly useful for reducing noise, such as when a service sends multiple updates for the same event or headline [2][3]. * Limitations: This mechanism primarily affects notifications that have not yet been displayed to the user [2][5]. If a user has already received and is viewing or has interacted with a notification, a new request with the same collapse ID might result in a new notification, depending on the current state of the device and the application [8][2][6]. * Implementation Note: Because this relies on the HTTP/2 API, your provider must be configured to use the APNs HTTP/2 protocol [5][7]. If your push notification service or library does not explicitly support setting custom HTTP/2 headers, you will be unable to utilize this feature [5][7]. For developers, ensuring that this header is correctly encoded according to HPACK (HTTP/2 header compression) requirements is also important when sending high volumes of notifications to maintain efficiency in the dynamic table [1][9].
Citations:
- 1: https://developer.apple.com/documentation/usernotifications/sending-notification-requests-to-apns
- 2: https://stackoverflow.com/questions/43463175/updating-a-delivered-notifications-ios-10
- 3: https://stackoverflow.com/questions/12413477/apple-push-notification-collapse-key-equivalent
- 4: https://developer.apple.com/library/archive/documentation/NetworkingInternet/Conceptual/RemoteNotificationsPG/CommunicatingwithAPNs.html
- 5: https://medium.com/the-guardian-mobile-innovation-lab/how-to-replace-the-content-of-an-ios-notification-2d8d93766446
- 6: https://stackoverflow.com/questions/44560851/apns-collapse-id-not-merging-multiple-notifications-ios-10
- 7: https://developer.apple.com/forums/thread/50606
- 8: https://developer.apple.com/forums/thread/737181
- 9: https://developer.apple.com/documentation/usernotifications/establishing-a-connection-to-apns
🌐 Web query:
Firebase Cloud Messaging CollapseKey replacement message still waiting delivery semantics
💡 Result:
In Firebase Cloud Messaging (FCM), a collapsible message is one that may be replaced by a newer message if the original has not yet been delivered to the device [1]. This mechanism is primarily used to ensure that only the most recent information is sent when multiple updates are queued for an offline device [2][3][4]. Delivery Semantics and Behavior: 1. Replacement Process: If a device is offline (or in a restricted state like Doze mode) and you send a new message with the same collapse_key as a message already waiting in FCM storage, the old message is discarded and replaced by the new one [5][6]. 2. Storage Limits: FCM can store a maximum of four different collapsible messages per device at any given time, each distinguished by a unique collapse_key [1][7]. 3. Default Behavior: Except for notification messages, all messages are non-collapsible by default [1]. If no collapse_key is set, FCM stores messages for future delivery without automatically discarding or collapsing them [5][6]. 4. Non-collapsible Messages: If you need to ensure all messages are delivered (e.g., chat messages or critical alerts), you should not use a collapse_key [1]. Non-collapsible messages remain in the queue until they are delivered or reach their time-to-live (TTL) expiration [5]. Important Constraints: - Notification messages are always collapsible by default and ignore the collapse_key parameter [1]. - If you exceed the limit of four different collapse keys for a single device, FCM will discard existing keys to make room for new ones, though the specific selection of which keys are kept is not guaranteed [1]. - Collapsing is an expected, intentional feature. When a message is collapsed, it is considered successfully replaced rather than failed [2]. For scenarios where you need to manage notifications already displayed in the system tray (rather than messages waiting in the FCM queue), you should use notification tags on the client side via the Android NotificationManager, which allows you to update or replace existing notifications displayed to the user [8].
Citations:
- 1: https://firebase.google.com/docs/cloud-messaging/customize-messages/collapsible-message-types
- 2: https://firebase.blog/posts/2024/07/understand-fcm-delivery-rates/
- 3: https://firebase.google.com/docs/cloud-messaging/understand-delivery
- 4: https://firebase.blog/posts/2019/02/life-of-a-message/
- 5: https://firebase.google.com/docs/cloud-messaging/customize-messages/setting-message-lifespan
- 6: https://stackoverflow.com/questions/50245336/will-push-notification-be-collapsed-if-collapse-key-not-passed-in-payload
- 7: https://firebase.google.com/docs/reference/fcm/rest/v1/projects.messages
- 8: FCM - To replace/cancel notification posted when app in background firebase/quickstart-android#744
🌐 Web query:
Web Push Protocol Topic header replace subsequent messages same topic documentation
💡 Result:
In the Web Push Protocol, as defined by RFC 8030, the Topic header is used to correlate push messages sent to the same subscription, allowing newer messages to replace outstanding (pending) ones if they share the same topic [1][2]. Key characteristics of this mechanism include: Replacement Behavior: When a push service receives a new message with a Topic header, it checks for any outstanding, undelivered messages to the same subscription that have an identical topic [1][3]. If one is found, the existing message resource is deleted and replaced with the new content [1][2]. This prevents the accumulation of redundant or outdated notifications while the user agent is offline [1][4]. Replacements Properties: The replacement request updates the stored Time-to-Live (TTL), urgency, and any receipt subscription associated with the previous message in that topic [1][2]. Constraints: 1. The Topic header value must be a string following the token grammar defined in RFC 7230 [1][2]. 2. The value is restricted to a maximum of 32 characters, using only the URL and filename-safe Base 64 alphabet [1][2]. 3. If a request includes a Topic header that does not meet these constraints, the push service must return a 400 (Bad Request) status code [1][3]. 4. The Topic header value is for the push service's internal use and must not be forwarded to the user agent [2][4]. If a message is sent with a topic that does not match any currently outstanding messages for that subscription, it is stored or delivered normally [1][4]. Delivery receipts for deleted (replaced) messages should be suppressed by the push service [1][2].
Citations:
- 1: https://datatracker.ietf.org/doc/html/rfc8030
- 2: https://www.rfc-editor.org/rfc/rfc8030.txt
- 3: https://datatracker.ietf.org/doc/rfc8030/
- 4: https://datatracker.ietf.org/doc/html/draft-ietf-webpush-protocol-11
Add durable idempotency for the at-least-once push task.
Collapse keys and Topic only coalesce pending platform queue messages, and send_push_notification() has no durable (channel_id, message_id) delivery record. A replay after an accepted request plus worker crash can still deliver a duplicate notification; add a unique delivery key or provider-supported idempotency key.
🤖 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 `@src/backend/core/services/push/tasks.py` around lines 173 - 180, Add durable
idempotency to the push task’s send flow by recording and enforcing a unique
delivery key composed of channel_id and message_id before invoking
send_push_notification(). Reuse the existing delivery record when the same task
is replayed, and pass a provider-supported idempotency key where available so
accepted requests cannot produce duplicate notifications after a worker crash.
| def test_parse_args_defaults(self): | ||
| """Test default argument values.""" | ||
| args = self._parse() | ||
|
|
||
| import worker | ||
| assert args.queues is None | ||
| assert args.exclude is None | ||
| assert args.disable_scheduler is False | ||
| assert args.loglevel == "INFO" | ||
| # One task at a time per process, so --concurrency alone says how many | ||
| # tasks run in parallel. | ||
| assert args.threads == 1 |
There was a problem hiding this comment.
🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick win
The default-threads assertion depends on the environment.
parse_args reads --threads from WORKER_THREADS with a fallback of "1" (src/backend/worker.py:158-166). If the test environment exports WORKER_THREADS, line 197 fails. The deployment layer in this stack sets worker environment variables, so this is reachable in CI and in a developer shell. Clear the variable inside the test.
🛡️ Proposed fix
- def test_parse_args_defaults(self):
+ def test_parse_args_defaults(self, monkeypatch):
"""Test default argument values."""
+ monkeypatch.delenv("WORKER_THREADS", raising=False)
args = self._parse()📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def test_parse_args_defaults(self): | |
| """Test default argument values.""" | |
| args = self._parse() | |
| import worker | |
| assert args.queues is None | |
| assert args.exclude is None | |
| assert args.disable_scheduler is False | |
| assert args.loglevel == "INFO" | |
| # One task at a time per process, so --concurrency alone says how many | |
| # tasks run in parallel. | |
| assert args.threads == 1 | |
| def test_parse_args_defaults(self, monkeypatch): | |
| """Test default argument values.""" | |
| monkeypatch.delenv("WORKER_THREADS", raising=False) | |
| args = self._parse() | |
| assert args.queues is None | |
| assert args.exclude is None | |
| assert args.disable_scheduler is False | |
| assert args.loglevel == "INFO" | |
| # One task at a time per process, so --concurrency alone says how many | |
| # tasks run in parallel. | |
| assert args.threads == 1 |
🤖 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 `@src/backend/core/tests/test_worker.py` around lines 187 - 197, Update
test_parse_args_defaults to remove or unset WORKER_THREADS before calling
self._parse(), ensuring the test exercises the parser’s default of one thread
regardless of the surrounding environment.
| def test_queue_order_preserved_after_exclusion(self): | ||
| """Test that queue priority order is preserved after exclusion.""" | ||
| import worker | ||
|
|
||
| queues = worker.ALL_QUEUES.copy() | ||
| exclude = ["outbound", "imports"] | ||
| result = [q for q in queues if q not in exclude] | ||
|
|
||
| expected = ["management", "inbound", "default", "reindex"] | ||
| assert result == expected | ||
|
|
||
|
|
||
| class TestBeatScheduleQueues: | ||
| """Test that beat schedule tasks use correct queues.""" | ||
| assert self._resolve(exclude="outbound,imports") == [ | ||
| "inbound", | ||
| "default", | ||
| "blobs", | ||
| "reindex", | ||
| ] |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Derive the expected list instead of hardcoding it.
This test verifies that resolve_queues preserves priority order. The hardcoded list also encodes the full contents of ALL_QUEUES, so adding any queue breaks this test even though order preservation still works. test_all_queues_defined_in_priority_order already owns that assertion.
♻️ Proposed refactor
def test_queue_order_preserved_after_exclusion(self):
"""Test that queue priority order is preserved after exclusion."""
- assert self._resolve(exclude="outbound,imports") == [
- "inbound",
- "default",
- "blobs",
- "reindex",
- ]
+ import worker
+
+ excluded = {"outbound", "imports"}
+ assert self._resolve(exclude="outbound,imports") == [
+ q for q in worker.ALL_QUEUES if q not in excluded
+ ]📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def test_queue_order_preserved_after_exclusion(self): | |
| """Test that queue priority order is preserved after exclusion.""" | |
| import worker | |
| queues = worker.ALL_QUEUES.copy() | |
| exclude = ["outbound", "imports"] | |
| result = [q for q in queues if q not in exclude] | |
| expected = ["management", "inbound", "default", "reindex"] | |
| assert result == expected | |
| class TestBeatScheduleQueues: | |
| """Test that beat schedule tasks use correct queues.""" | |
| assert self._resolve(exclude="outbound,imports") == [ | |
| "inbound", | |
| "default", | |
| "blobs", | |
| "reindex", | |
| ] | |
| def test_queue_order_preserved_after_exclusion(self): | |
| """Test that queue priority order is preserved after exclusion.""" | |
| import worker | |
| excluded = {"outbound", "imports"} | |
| assert self._resolve(exclude="outbound,imports") == [ | |
| q for q in worker.ALL_QUEUES if q not in excluded | |
| ] |
🤖 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 `@src/backend/core/tests/test_worker.py` around lines 254 - 261, The
test_queue_order_preserved_after_exclusion assertion should derive its expected
queues from ALL_QUEUES rather than hardcoding the full list. Filter out
“outbound” and “imports” while retaining ALL_QUEUES order, then compare
_resolve’s result against that derived sequence; leave full queue-content
validation to test_all_queues_defined_in_priority_order.
| os.environ.setdefault("DJANGO_SETTINGS_MODULE", "messages.settings") | ||
| os.environ.setdefault("DJANGO_CONFIGURATION", "Development") |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Do not default DJANGO_CONFIGURATION to Development.
Importing the messages package now silently selects the Development configuration when DJANGO_CONFIGURATION is unset. Before this change a process without that variable failed loudly. A deployment that loses the variable (a new dyno type, a cron container, an ad-hoc management command) now boots with development settings: LocMemCache instead of Redis, development-oriented defaults, and no boot-time signal that the wrong configuration is active.
Set only DJANGO_SETTINGS_MODULE here and let django-configurations fail when DJANGO_CONFIGURATION is missing. If a default is required for the Dramatiq CLI re-exec path, make that path pass the variable explicitly instead.
🔒️ Proposed fix
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "messages.settings")
-os.environ.setdefault("DJANGO_CONFIGURATION", "Development")📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| os.environ.setdefault("DJANGO_SETTINGS_MODULE", "messages.settings") | |
| os.environ.setdefault("DJANGO_CONFIGURATION", "Development") | |
| os.environ.setdefault("DJANGO_SETTINGS_MODULE", "messages.settings") |
🤖 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 `@src/backend/messages/__init__.py` around lines 10 - 11, Remove the
DJANGO_CONFIGURATION defaulting from the package initialization, leaving only
the DJANGO_SETTINGS_MODULE setup so missing configuration continues to fail
loudly. If the Dramatiq CLI re-exec path requires a configuration, set it
explicitly in that path rather than restoring a package-level default.
| middleware = [] | ||
| if self.TASK_PROMETHEUS_ENABLED: | ||
| middleware.append("dramatiq.middleware.prometheus.Prometheus") | ||
| middleware += [ | ||
| "dramatiq.middleware.AgeLimit", | ||
| "dramatiq.middleware.TimeLimit", | ||
| "dramatiq.middleware.Callbacks", | ||
| "dramatiq.middleware.Retries", | ||
| "dramatiq.middleware.CurrentMessage", | ||
| # Recycles connections closed by Postgres between messages; without | ||
| # it a worker idling past ``CONN_MAX_AGE`` fails its next task. | ||
| "django_dramatiq.middleware.DbConnectionsMiddleware", | ||
| "core.task_utils.WorkerShutdownMiddleware", | ||
| ] | ||
| if self.TASK_HISTORY_ENABLED: | ||
| middleware.append("django_dramatiq.middleware.AdminMiddleware") | ||
| return middleware |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check whether django_dramatiq registers the Results middleware from DRAMATIQ_RESULT_BACKEND.
fd -t f -p --full-path '*django_dramatiq*' -e py | xargs -r rg -n -C6 'RESULT_BACKEND|Results\('
# Confirm the project does not add it anywhere itself.
rg -nP -C3 'results\.Results|dramatiq\.results' src/backendRepository: suitenumerique/messages
Length of output: 3539
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== relevant package settings =="
rg -n -C3 'dramatiq_middleware|DRAMATIQ_RESULT_BACKEND|TASK_HISTORY|results\.Results|dramatiq\.results|store_results|EagerBroker|Results\(' src/backend/messages/settings.py src/backend/core/task_utils.py src/backend
echo
echo "== candidate files =="
git ls-files | rg '(^|/)django[_-]?dramatiq|dramatiq|settings\.py|task_utils\.py$' | sed -n '1,200p'Repository: suitenumerique/messages
Length of output: 18591
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== task_utils middleware lookup =="
sed -n '230,270p;459,505p' src/backend/core/task_utils.py.py 2>/dev/null || sed -n '230,270p;459,505p' src/backend/core/task_utils.py
echo
echo "== task status result lookup =="
rg -n -C4 'store_results|ResultGettingMiddleware|Results|ResultBackend|GetResult|ResultMissing|ResultGetting' src/backend/core/src src/backend/core | sed -n '1,240p'
echo
echo "== package manifests for django_dramatiq/dramatiq =="
rg -n -C3 'django[_-]?dramatiq|dramatiq' pyproject.toml setup.py setup.cfg requirements* src/backend pyproject.toml 2>/dev/null | sed -n '1,240p'Repository: suitenumerique/messages
Length of output: 22640
🌐 Web query:
django-dramatiq DRAMATIQ_RESULT_BACKEND register middleware Results middleware
💡 Result:
To use a results backend with django-dramatiq, you must configure the Results middleware within your Dramatiq settings. While standard Dramatiq requires manually adding the Results middleware to the broker, the django-dramatiq integration allows you to specify a DRAMATIQ_RESULT_BACKEND configuration in your Django settings.py [1][2]. Configuration Steps: 1. Define the backend in settings.py: You can use the DRAMATIQ_RESULT_BACKEND dictionary to define which backend to use (such as Redis) and provide middleware-specific options [1][2]. Example configuration: DRAMATIQ_RESULT_BACKEND = { "BACKEND": "dramatiq.results.backends.redis.RedisBackend", "BACKEND_OPTIONS": { "url": "redis://localhost:6379", }, "MIDDLEWARE_OPTIONS": { "result_ttl": 1000 * 60 * 10 # Results expire after 10 minutes (in milliseconds) } } 2. Enable the middleware: The Results middleware is not enabled by default because it is not needed for every application [3]. To enable it, you must add the Results middleware to the list of middlewares in your DRAMATIQ_BROKER configuration [1][2]. Note that the Results middleware requires a backend instance to be passed to it [4][5]. When using the django-dramatiq setup, ensure your broker configuration correctly initializes the middleware as expected by the Dramatiq framework [4]. 3. Actor Usage: Once configured, you must explicitly enable result storage for specific actors by setting store_results=True in the @dramatiq.actor decorator [6][5]: @dramatiq.actor(store_results=True) def my_task(x, y): return x + y Important Considerations: - Result Expiration: Results will expire after the time specified in result_ttl (default is 10 minutes) to prevent the backend from running out of space [4][6]. - Retries: If an actor has retries enabled, the result of the message may be delayed until all retries have completed [6]. - Alternative Backends: Besides standard Redis or Memcached backends, third-party packages like django-dramatiq-result-backend exist if you prefer to store results using the Django ORM [7].
Citations:
- 1: https://github.com/bogdanp/django_dramatiq
- 2: https://github.com/Bogdanp/django_dramatiq/blob/master/README.md
- 3: https://dramatiq.io/advanced.html
- 4: https://dramatiq.io/cookbook.html
- 5: https://github.com/Bogdanp/dramatiq/blob/master/docs/source/cookbook.rst
- 6: https://github.com/Bogdanp/dramatiq/blob/master/dramatiq/results/middleware.py
- 7: https://pypi.org/project/django-dramatiq-result-backend/
Add dramatiq.results.Results to the broker middleware. Setting DRAMATIQ_RESULT_BACKEND alone does not enable result storage. Without the middleware, actors with store_results=True do not save task return values, and EagerBroker.enqueue() does not call after_process_message, so tests cannot poll task results even though the option defaults to True.
📍 Affects 2 files
src/backend/messages/settings.py#L1186-L1202(this comment)src/backend/core/task_utils.py#L255-L266
🤖 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 `@src/backend/messages/settings.py` around lines 1186 - 1202, Add
dramatiq.results.Results to the middleware list returned by the settings
middleware configuration in src/backend/messages/settings.py:1186-1202, and
update EagerBroker.enqueue in src/backend/core/task_utils.py:255-266 to invoke
after_process_message so stored results are available in eager execution.
| # Redis Streams broker shared across La Suite (see [tool.uv.sources] below): | ||
| # event-driven XREADGROUP consumption, deadline-based recovery of tasks from | ||
| # dead workers, and the queue dashboard mounted in the Django admin. | ||
| "dramatiq-redis-streams", |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Pin dramatiq-redis-streams to the git source in the dependency entry.
The dependency list contains a bare dramatiq-redis-streams requirement. Only [tool.uv.sources] redirects it to the git revision. Any resolver that ignores [tool.uv.sources] — pip install ., a Docker build stage that does not use uv, or a downstream consumer — resolves the name from PyPI instead. That substitutes an unrelated or attacker-controlled distribution for the broker in the task path.
Express the requirement so it cannot resolve from PyPI, for example as a direct URL reference pinned to the same revision, and keep the uv source entry for local development.
🔒️ Proposed direction
- "dramatiq-redis-streams",
+ "dramatiq-redis-streams @ git+https://github.com/sylvinus/dramatiq-redis-streams.git@819912de7259361688fcfa1b91b0c465ecd2b6e4",Also applies to: 107-109
🤖 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 `@src/backend/pyproject.toml` around lines 49 - 52, Update the
dramatiq-redis-streams dependency entry in the project dependency list to use a
direct URL reference pinned to the intended git revision, preventing fallback to
PyPI; retain the existing [tool.uv.sources] entry for local development and
apply the same change to the corresponding dependency entry noted in the review.
1 participant
This is a rebase/rework of #560
Summary by CodeRabbit