PLAT-3188: Recover Celery workers from dropped DB connections

Jira: PLAT-3188. This blocks the October bulk resend under PLAT-3134.

Problem

web.py caches one DB connection per thread and only replaces it at the end of a web request. Celery workers have no requests, so a connection the server has closed stays in use forever.

On 2026-11-08 00:24 Postgres closed its connections. Until the workers were restarted on 2026-11-11, every task on the API pod's workers failed with InterfaceError('connection already closed'). For process-events-batch, the retries, salvage_batch and the quarantine save all failed on the same connection, so the batches were dropped after a 202 (Failed to save to fallback storage was logged ~4,400 times).

Changes

• Reconnect before each task (api.reset_stale_connection, task_prerun hook in tasks/celery_app.py): drop the thread's connection if it's closed or fails a SELECT 1, so web.py opens a new one. It's only pinged when idle, so an open transaction is never touched.
• Connection errors retry instead of losing the batch (tasks/events_ingestion.py): backoff of 1, 2, 4, 8 min, then every 10 min (~4.5h). The backoff counts data-error retries too, so a connection lost during salvage starts at 8 min. Countdowns are kept under RabbitMQ's 30 min consumer_timeout; if a backlog pushes a retry past it, RabbitMQ redelivers it, which is harmless as the inserts are idempotent. They never enter the salvage path. salvage_batch now raises connection errors instead of quarantining them.
  • A connection error is an InterfaceError, or an OperationalError with no SQLSTATE, class 08, or 57P01–57P03. Other OperationalErrors, such as a statement timeout (57014) or out of memory (53200), still go through the normal retry → salvage path.
  • If retries run out, one CRITICAL line with the marker EVENTS_BATCH_LOST gives the task id, uploader, measures and timestamp range, so the batch can be resent. The same marker is logged when salvage can't quarantine events for another reason (e.g. read-only after failover): salvage_batch raises QuarantineError with the lost events. Timestamps are ordered by instant, so mixed UTC offsets (a resend across the DST change) give the right range.
• JSON logging in workers and beat (setup_logging hook) with a JsonFormatter that escapes quotes and newlines and keeps tracebacks in the entry. This replaces the _log_safe workaround. The level comes from API_DEBUG, so -l info was removed from the supervisord commands. kombu and amqp are capped at WARNING, and celery stays at INFO in debug mode.
• Workers pod logs reach Cloud Logging: supervisord-workers.conf set stdout_logfile etc. on [group:] sections, which supervisor ignores. The Celery output of beat and the cache, popularity and report workers was going to files in /tmp. Those settings are now per program.
• Bulk worker node name metrics-bulk@%h, so celery inspect/control replies no longer collide with save_to_database_tasks.
• Supervisor autorestart per program in supervisord.conf: it was set on [group:api_and_tasks], which supervisor ignores, so the Celery workers weren't restarted after a clean exit.
• Supervisor socket (/tmp/supervisor.sock, 0700) in both configs, so supervisorctl works in the pods.
• Removed a leftover debug print().

Testing

• Full unit suite passes locally against Postgres and Redis (333 tests).
• tests/test_db_reconnect.py kills the backend with pg_terminate_backend (as the outage did) and checks that the connection is replaced, both directly and through a task's task_prerun. It also checks that a real statement timeout isn't classed as a connection error.
• Task tests cover the backoff, the separate retry limit, the EVENTS_BATCH_LOST line, and salvage raising connection errors.
• Both supervisord configs parse with supervisor 4.2.5, and every program logs to /dev/stdout//dev/stderr.
• A worker started locally against an unreachable broker logs valid single-line JSON (including Celery's multi-line connection errors).

Before deploy

• Update the FALLBACK_EVENTS_PENDING alert filter: it matches textPayload:"FALLBACK_EVENTS_PENDING", and after this MR the line arrives as jsonPayload.message. A bare "FALLBACK_EVENTS_PENDING" search term matches both.
• Add a log-based alert on EVENTS_BATCH_LOST ("metrics-api: event batch lost", CRITICAL, same notification channel).

After deploy

• Check worker logs from both pods show up as jsonPayload, and that FALLBACK_EVENTS_PENDING still matches.
• Check supervisorctl status works in both pods.

🤖 Generated with Claude Code