Mirror DB writes to a second database (#502) by tejas7777 · Pull Request #568 · tutorcruncher/morpheus

tejas7777 · 2026-10-08T09:33:16Z

🔍 Reviewed up to 9b8494378 — CHANGES_REQUESTED, 2026-10-13 (/full-review). Re-run after non-trivial changes.

Issue number: Part of #502 (not Close: backfill and switch still to do)

Technical description of changes

New MIRROR_DATABASE_URL (unset by default = off). When set, every committed ORM insert/update/delete queues Celery task mirror_rows, which copies those rows to the mirror with the same ids:
- upserts rows still on primary, deletes ids gone from primary
- copies missing parents (company, group, message) first
- skips messages.vector (mirror's trigger rebuilds it)
- queued once the session has returned its connection to the pool (after_transaction_end), not inside commit()
- 5 retries (~30s), then Sentry; a queue failure is logged, never fails the write
Copies run on their own mirror queue (task_routes), served by a new Procfile mirror process; the default worker never runs them.
Mirror engine: sslmode=require unless the URL sets one; hide_parameters=True, so errors name the SQL without row values.
Hourly beat reconcile_mirror_companies: re-copies any mirror company whose (id, code) differs from the primary (a purge or rename whose copy never landed).
Bulk statements skip ORM events, so subaccount rename and purge call mirror_ids themselves.
Not mirrored: delete_old_emails, materialized view refresh.
All in app/messages/mirror.py; delete it after the switch.

Description for the non-dev team to understand

No user-facing change. Once switched on, every email/SMS record is also saved to the new database.

Recreation Steps

Covered by tests/test_mirror.py (uses a real second Postgres database).

Manual testing required

None.

Switch-on steps (not run; need Heroku access):

heroku addons:create heroku-postgresql:standard-4 --as NEW_DATABASE -a <app>
Build schema. Must be create_db_and_tables: the purge relies on its CASCADE FKs.
heroku run -a <app> 'DATABASE_URL=$NEW_DATABASE_URL python -c "from app.core.database import create_db_and_tables; create_db_and_tables()"'
heroku ps:scale mirror=1 -a <app> (until it runs, copies wait in Redis).
heroku addons:attach <addon> --as MIRROR_DATABASE -a <app> (app restarts with mirroring on), then heroku addons:detach NEW_DATABASE -a <app>
Check plan connection limit: up to 4 mirror connections per mirror worker child, 2 children per dyno.

Follow-ups (separate PRs):

Backfill last year (ON CONFLICT (id) DO NOTHING), then re-copy events.
Before switch: compare row counts and max ids, refresh message_aggregation.
Switch, in this order (mirror_rows treats DATABASE_URL as the truth):
1. Maintenance on; worker and beat scaled to 0 so RDS stops taking writes.
2. Wait for the mirror queue to drain.
3. Run reconcile_mirror_companies and the content check; setval sequences.
4. One config change: DATABASE_URL to Heroku, unset MIRROR_DATABASE_URL.
5. Scale mirror to 0, worker/beat back up, maintenance off. Then delete mirror.py and turn on delete_old_emails.

Limits: one extra small Celery task per write (own mirror queue and process); legacy messages.spam_status/spam_reason not copied; an outage longer than the retries leaves gaps (pre-switch check catches them).

Checklist

Issue number added
Tests: tests/test_mirror.py 17 passed
Coverage: mirror.py 98.9% from tests/test_mirror.py (line 80, the unset early return in mirror_ids, is the miss)
make lint passes (ruff + ty)
Docstrings clear; type hints on new/changed signatures
API surface: none
Schema change: none on primary; mirror via create_db_and_tables
app/render changes: n/a

🤖 Generated with Claude Code

Part of #502. With MIRROR_DATABASE_URL set, every ORM insert, update and delete queues a Celery task once its transaction commits. The task copies the rows' current primary state to the mirror with the same ids, and brings missing parent rows across first. The two bulk statements (the subaccount rename and the company purge) queue the copy themselves. With the setting unset, nothing changes. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

codecov-commenter · 2026-10-08T09:35:10Z

Codecov Report

✅ All modified and coverable lines are covered by tests.

📢 Thoughts on this report? Let us know!

tejas7777 · 2026-10-08T14:09:39Z

I am not very onboard on the approach of using celery to write to the heroku db

tejas7777

CLAUDE SAYS:

Review of 9b84943.

The copy design is sound: each task syncs the rows' current primary state, so running it twice or out of order converges. It is off by default, and a failed enqueue never fails the primary write. Before MIRROR_DATABASE_URL is set, the mirror path needs time limits and to come off the send workers' queue.

Verdict: CHANGES_REQUESTED (no Blocker; 5 High, 2 Medium, 4 Nit inline).

Raised by more than one reviewer

mirror.py:52: no statement/lock/task time limit, and the task shares the send queue.
tasks.py:780: the mirror purge is one unbounded cascade delete. Rated High, not Blocker, because the primary purge already runs the same shape on master; the retries and the shared worker are what is new.
mirror.py:93: ignore_result=True is missing.
mirror.py:59: an unreachable guard.

Local verification

make lint: pass.
tests/test_mirror.py: 11 passed.

Functional run-through (local server and worker, primary and mirror both local test databases, synthetic data):

Email send with links, webhook event, click, subaccount delete, and SMS send on a re-created company code: after each, the mirror matched the primary in every column, ids included. The mirror's own trigger filled messages.vector.
Mirror unreachable: sends still returned 201 and were stored. Each copy retried 5 times and then failed.
MIRROR_DATABASE_URL unset: no copy was queued.
No dead ends.

Cross-cutting

The pre-switch check in the PR body (row counts and max ids) cannot see value drift on existing messages rows (status, update_ts, cost). See mirror.py:97 and mirror.py:131. It needs a per-id content comparison.
While the mirror is live, every one-off DDL in another PR (new column, ALTER TYPE ... ADD VALUE, index) must also run on the mirror before that code deploys. Otherwise every copy of that table fails. Worth a sentence in the mirror.py docstring.

tejas7777 · 2026-10-13T08:22:09Z

+            settings.mirror_database_url,
+            settings.db_worker_pool_size,
+            settings.db_worker_max_overflow,
+            connect_timeout=CONNECT_TIMEOUT,


[High] CLAUDE SAYS:

Only the connect has a time limit. The mirror connection has no statement_timeout, lock_timeout or keepalives, and mirror_rows has no time_limit. The task also runs on the default queue the send tasks use. A mirror that accepts the connection and then stalls (a lock wait behind the planned backfill, non-concurrent DDL, a half-open connection after a failover) pins each worker child that picks up a copy, and sends queue behind them.

Fix: options='-c timezone=UTC -c statement_timeout=30s -c lock_timeout=5s' (make_engine overwrites its own options, so repeat the timezone), keepalives (keepalives_idle=30, keepalives_interval=10, keepalives_count=3) and time_limit=120 on the task. Optionally give mirror_rows its own queue and worker.

tejas7777 · 2026-10-13T08:22:09Z

+
+def _queue_pending(session: DBSession) -> None:
+    for table_name, ids in session.info.pop(PENDING_KEY, {}).items():
+        mirror_ids(table_name, sorted(ids))


[High] CLAUDE SAYS:

Every commit that writes a mirrored row queues its own mirror_rows task: a Mandrill batch of K events goes from 1 task to 1+K (one commit per event), each click adds 1, and each stored message adds 2 (message, then links). They all land on the default queue send_email/send_sms use (no task_routes, one worker in the Procfile), so a webhook burst or a slow mirror sits ahead of customer sends.

Fix: SADD the ids per table in _queue_pending and drain them in batches from a beat task with one mirror_rows call per batch; route mirror_rows to its own queue and worker; skip the DELETE when every id was found.

tejas7777 · 2026-10-13T08:22:09Z

+event.listen(DBSession, 'after_rollback', _drop_pending)
+
+
+@celery_app.task(


[Medium] CLAUDE SAYS:

No ignore_result=True. With the Redis result backend and task_track_started=True (app/core/celery.py), every copy writes a STARTED and a SUCCESS celery-task-meta-* key that lives a day, one per committed write, and nothing reads them.

Fix: add ignore_result=True to the decorator.

tejas7777 · 2026-10-13T08:22:10Z

+    name='app.messages.mirror.mirror_rows',
+    autoretry_for=(SQLAlchemyError,),
+    retry_backoff=True,
+    max_retries=5,


[High] CLAUDE SAYS:

With retry_backoff=True and the default jitter, five retries span at most about 31s (about 15s on average). A mirror outage longer than that loses those copies for good. The pre-switch check (row counts and max ids) catches lost inserts and deletes. It cannot catch the one ORM update to an existing row, message.cost = m.price_amount (tasks.py:663): the row is already on the mirror with cost = NULL. After the switch, SMS spend and the cost_limit check would under-count. The backfill's ON CONFLICT (id) DO NOTHING won't repair it.

Fix: retry_backoff_max=600 with about 12 retries (roughly an hour), retrying only transient errors (OperationalError, IntegrityError, pool TimeoutError). Add a content comparison to the pre-switch step, e.g. count(cost)/sum(cost) per company and month for SMS, or an md5 of (id, status, cost, update_ts) per id range, and re-queue any mismatch.

tejas7777 · 2026-10-13T08:22:10Z

+    if not settings.mirror_database_url:
+        # Unset while copies were still queued: nothing to copy to.
+        return
+    with get_session() as db, get_mirror_engine().begin() as mirror_conn:


[High] CLAUDE SAYS:

The primary read on line 115 starts a transaction on the RDS connection, and nothing ends it until this with exits. The connection therefore stays idle in transaction through every mirror round trip: the parent checks, the recursive parent copies, the DELETE, the upsert and the mirror COMMIT. That happens once per committed write.

Fix: materialise the rows, then db.close() before the first mirror statement (the session is reusable after close). Do a fresh short primary read for any missing parents.

tejas7777 · 2026-10-13T08:22:10Z

        db.commit()
+    # The companies are gone from the primary, so the mirror deletes them, and its CASCADE foreign
+    # keys take their messages with them.
+    mirror_ids('companies', company_ids)


[High] CLAUDE SAYS:

Once the primary purge commits, mirror_rows('companies', ids) finds no rows and runs one DELETE FROM companies on the mirror. Its CASCADE FKs remove every group, message, event and link of the agency in one statement and one transaction. That is the same shape as the primary purge above, but here it runs inside a task that retries the whole cascade up to 5 times, with no mirror statement timeout, on a worker child that also serves sends.

Fix: a batched mirror purge: delete the company's messages in id batches (... WHERE id IN (SELECT id FROM messages WHERE company_id = ANY(:ids) LIMIT 5000)), commit per batch and re-enqueue until none are left, then message_groups, then companies. Combine it with a mirror statement_timeout.

tejas7777 · 2026-10-13T08:22:10Z

+
+@celery_app.task(
+    name='app.messages.mirror.mirror_rows',
+    autoretry_for=(SQLAlchemyError,),


[Nit] CLAUDE SAYS:

SQLAlchemyError also covers DataError/ProgrammingError (e.g. an enum value added to the primary but not the mirror), which fail the same way on every retry. Fix: autoretry_for=(OperationalError, IntegrityError).

tejas7777 · 2026-10-13T08:22:10Z

+@celery_app.task(
+    name='app.messages.mirror.mirror_rows',
+    autoretry_for=(SQLAlchemyError,),
+    retry_backoff=True,


[Nit] CLAUDE SAYS:

The PR body says "5 retries (~30s)". Celery's default jitter draws each countdown from 0..2**n, so 31s is the maximum and about 15s is typical. Fix: retry_jitter=False for a fixed window, or reword the PR body.

tejas7777 · 2026-10-13T08:22:10Z

+
+def mirror_ids(table_name: str, ids: list[int]) -> None:
+    """Queue a copy of these rows to the mirror database; a no-op unless MIRROR_DATABASE_URL is set."""
+    if not settings.mirror_database_url or not ids:


[Nit] CLAUDE SAYS:

or not ids cannot be true: _queue_pending only holds non-empty sets, and both bulk callers sit behind a non-empty company list. Fix: if not settings.mirror_database_url:.

tejas7777 · 2026-10-13T08:22:10Z



-def _make_engine(pool_size: int, max_overflow: int):
+def make_engine(url: str, pool_size: int, max_overflow: int, **connect_args):


[Nit] CLAUDE SAYS:

make_engine is now public and imported across modules, but has no return type and an untyped **connect_args. Fix: def make_engine(url: str, pool_size: int, max_overflow: int, **connect_args: Any) -> Engine:.

after_commit fires before SQLAlchemy hands the connection back to the pool, so the broker publish ran while the connection was still checked out. Mark the ids committed there and queue them from the root transaction's after_transaction_end instead. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

make_engine takes hide_parameters; the mirror engine sets it, so a failed copy's logged and retried error names the SQL without the bound row values. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Route app.messages.mirror.* to a `mirror` queue served by a new Procfile `mirror` process, so a slow or unreachable mirror never holds up send_email/send_sms on the default worker. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

The mirror engine passes sslmode=require unless MIRROR_DATABASE_URL sets its own sslmode; the test fixture sets sslmode=prefer because CI's Postgres service has no TLS. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

New beat task reconcile_mirror_companies compares (id, code) for every company on both databases and re-copies the ones the mirror holds out of step: a purge or rename whose copy never landed. The existing copy then deletes or renames them, and CASCADE takes a purged company's history with it. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

mirror_rows treats DATABASE_URL as the truth, so the switch stops writes to RDS and drains the mirror queue before DATABASE_URL moves and MIRROR_DATABASE_URL is unset. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

tejas7777 self-assigned this Oct 8, 2026

tejas7777 commented Oct 13, 2026

View reviewed changes

tejas7777 and others added 6 commits October 13, 2026 10:40

Mirror: keep row values out of mirror error messages

8214bc6

make_engine takes hide_parameters; the mirror engine sets it, so a failed copy's logged and retried error names the SQL without the bound row values. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Mirror: document the switch order

bcf344d

mirror_rows treats DATABASE_URL as the truth, so the switch stops writes to RDS and drains the mirror queue before DATABASE_URL moves and MIRROR_DATABASE_URL is unset. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

		event.listen(DBSession, 'after_rollback', _drop_pending)


		@celery_app.task(



		def _make_engine(pool_size: int, max_overflow: int):
		def make_engine(url: str, pool_size: int, max_overflow: int, **connect_args):

Conversation

tejas7777 commented Oct 8, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Technical description of changes

Description for the non-dev team to understand

Recreation Steps

Manual testing required

Checklist

Uh oh!

codecov-commenter commented Oct 8, 2026

Codecov Report

Uh oh!

tejas7777 commented Oct 8, 2026

Uh oh!

tejas7777 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

This branch has not been deployed

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

tejas7777 commented Oct 8, 2026 •

edited

Loading