feat(job): refactor job logging so it isn't a bottleneck

[mihow]

Observation

During a moderately-sized async ML job (~742 images), the Celery worker running process_nats_pipeline_result begins producing a recurring pair of errors once multiple NATS results are in flight:

ERROR Failed to save logs for job #N: SoftTimeLimitExceeded()
ERROR Failed to save logs for job #N: sending query failed: another command is already in progress
ERROR Error processing pipeline result for job N: sending query failed: another command is already in progress. NATS will re-deliver the task message.

The same 5-minute window also shows Django containers logging terminating connection due to idle-in-transaction timeout. Errors are spread across all ForkPoolWorker-* processes (not clustered on one) and are only ever attached to the single actively-processing job row. A live pg_stat_activity snapshot during the error window showed no lock contention and plenty of connection headroom (~90 of 500) — i.e. the DB is not saturated.

This appears (tentatively) to be a logical within-connection collision, not a capacity problem.

Current implementation

ami/jobs/models.py:335–361 — JobLogHandler.emit:

def emit(self, record: logging.LogRecord):
    logger.log(record.levelno, self.format(record))
    try:
        self.job.refresh_from_db(fields=["logs"])          # SELECT
        # ...in-memory insert at position 0, dedupe check, truncate to 1000...
        self.job.save(update_fields=["logs"], update_progress=False)  # UPDATE
    except Exception as e:
        logger.error(f"Failed to save logs for job #{self.job.pk}: {e}")

The refresh_from_db exists for a real reason (see PR #1162): each concurrent worker holds its own stale in-memory JobLogs pydantic object, and without the refresh the last save() wins and earlier entries are silently dropped. So the current design is trading concurrency-safety of log entries for concurrency-risk at the connection level.

Likely failure path (hypothesis)

_update_job_progress (same file, ~L179–209) wraps a transaction.atomic() + Job.objects.select_for_update().get(pk=job_id) and then calls job.logger.info(...) inside that block. JobLogHandler.emit then issues:

1. SELECT (refresh_from_db) on the same connection that already holds a SELECT … FOR UPDATE cursor on the same row
2. An UPDATE on the same row for logs only

Under ~180 result tasks per 5 minutes × concurrency 8, several workers end up waiting on the FOR UPDATE lock. They stall past Postgres's idle_in_transaction_session_timeout (Postgres terminates the connection server-side), or they stall past Celery's 300s soft_time_limit (soft timeout interrupts the task mid-transaction). The resulting exception is caught by the bare except Exception in emit, which then calls logger.error(...) — which re-enters emit — which issues another query on the same now-broken connection. That produces the "another command is already in progress" cascade.

What still needs to be verified: whether CONN_MAX_AGE > 0 + CONN_HEALTH_CHECKS = False on the celeryworker is causing Django to hand dead server-killed connections to subsequent tasks; and the exact value of idle_in_transaction_session_timeout on the Postgres instances used for async jobs. Both would affect how often this fires in practice.

Consequence

• Log entries are dropped during bursts (the inner save is the exception site)
• Connection churn / soft-timeouts make individual process_nats_pipeline_result tasks much slower than they otherwise would be, which compounds queue backlog
• The bare-except cascade can convert a single DB blip into dozens of error lines per task

Proposals

Ordered by scope.

A. Separate JobLog table (recommended)

class JobLog(models.Model):
    job = models.ForeignKey(Job, on_delete=models.CASCADE, related_name="log_entries")
    level = models.CharField(max_length=10)
    message = models.TextField()
    created_at = models.DateTimeField(auto_now_add=True, db_index=True)

    class Meta:
        indexes = [models.Index(fields=["job", "-created_at"])]

emit becomes a single INSERT:

def emit(self, record):
    logger.log(record.levelno, self.format(record))
    try:
        JobLog.objects.create(
            job_id=self.job.pk,
            level=record.levelname,
            message=self.format(record),
        )
    except Exception as e:
        logger.error(f"Failed to save log for job #{self.job.pk}: {e}")

Why this is the right shape:

• Pure INSERT — no read-before-write, no nested cursor, no row-level FOR UPDATE contention
• Natural concurrency: multiple workers inserting into the same child table is a solved pattern (already the case for Detection, Classification, etc.)
• Removes the arbitrary 1000-entry truncation; paginable / time-range queryable
• Removes the if msg not in stdout dedupe scan (which is O(n) per write, ~n=1000)
• Log levels become first-class (filter by ERROR for the "stderr" view; filter by INFO+ for the full view)

Costs:

• One migration + a compatibility shim
• Frontend read path changes, if it consumes Job.logs.stdout / .stderr directly

Proposed rollout to avoid a hard cutover:

1. Add JobLog model + migration
2. Dual-write: keep the existing JSON path; also JobLog.objects.create(...) in emit
3. Switch Job.logs.stdout / .stderr to @property that reads the last N rows from JobLog (keeps existing consumers working with no API contract change)
4. Verify frontend, then stop writing to the JSON field
5. Migrate existing data + drop the JSON column in a separate release

B. Atomic jsonb_insert (no schema change)

Replace the refresh+mutate+save with a single SQL statement:

UPDATE jobs_job
SET logs = jsonb_set(
    COALESCE(logs, '{}'::jsonb),
    '{stdout}',
    jsonb_insert(COALESCE(logs->'stdout', '[]'::jsonb), '{0}', to_jsonb(%s::text), false)
)
WHERE id = %s;

This eliminates the read-before-write and therefore the nested cursor bug. But:

• Truncation to 1000 entries becomes awkward (subquery on jsonb_array_length, two statements or a CTE)
• Dedup (if msg not in stdout) disappears
• Critically, the UPDATE still targets the same jobs_job row that _update_job_progress holds a SELECT FOR UPDATE on — so we trade a connection-state bug for a straight row-lock wait. Doesn't fix the contention, only the specific cursor error.

C. Redis-buffered + periodic flush

emit does LPUSH to a per-job Redis list; a celerybeat task flushes into the DB every N seconds.

• Fully decouples log writes from the request/task path
• Batched writes = much higher throughput
• Adds eventual-consistency (logs lag real-time by up to N seconds)
• More moving parts; loss-risk if the key is evicted before flush

Could also be layered on top of (A) as an optimisation later.

Recommendation

Go with A, rolled out dual-write as above. The nested-cursor bug is the visible symptom, but the deeper issue is that log writes share the row that the job-progress path locks. Moving them to their own table removes the coupling entirely and is consistent with how every other high-volume write in the model lives today.

B is tempting if we want something in this week — but because the remaining row-lock contention on jobs_job is a real source of stall, it only partially addresses the symptom set we saw on the ~742-image job.

C is a good follow-up once the write path is a plain insert; at that point it becomes an incremental throughput optimisation rather than a bug fix.

Adjacent quick wins worth considering in the same PR

• Set CONN_HEALTH_CHECKS = True (Django ≥4.1) on the worker so a server-killed connection is detected before the next query reuses it
• Move job.logger.* calls in _update_job_progress outside the transaction.atomic() block regardless of which proposal lands — logging shouldn't be inside a held row lock
• Tighten the bare except Exception in emit to catch a narrower exception class so a broken connection can propagate instead of being swallowed, then re-enter emit again on the same broken connection

Related

• [Draft] Don't overwrite logs & status in concurrent background tasks #1026 (draft, open) — "Don't overwrite logs & status in concurrent background tasks" — pulled from Update status of disconnected and stale jobs #981, directly addresses the same concurrency-on-the-same-row class of bug. This ticket's proposal (A) is one concrete way to land [Draft] Don't overwrite logs & status in concurrent background tasks #1026's goal; worth folding them together rather than two parallel efforts.
• Update status of disconnected and stale jobs #981 (closed) — "Update status of disconnected and stale jobs" — origin of [Draft] Don't overwrite logs & status in concurrent background tasks #1026's patch set; introduced the periodic unfinished-job status check and the initial concurrent-update safety work.
• Support ML async job cancellation, fail jobs on redis errors #1162 — introduced the current try/except wrap and the refresh-before-save pattern in JobLogHandler.emit
• Review and simplify job logs #1236 — separate cleanup ticket for log noise (signal-to-noise in the existing logs), complementary to this but scoped to volume, not write-path
• feat(jobs): make NATS queue activity visible in async ML job logs #1222, feat(jobs): show who requests which tasks in ML processing job logs #1238, feat(jobs): show worker availability in async job logs #1240 — recent additions that write to job.logger more frequently and amplify the exposure of this bug
• Follow-ups from PR #1231 review: remaining Redis/retry gaps in async job state #1232 — open follow-ups on Redis/retry gaps (another axis on which emit can fail)

[mihow]

Empirical validation + new implementation angles (2026-04-20)

Hit this exact bottleneck on a staging deployment with 5 concurrent async_api jobs (2×10 + 2×100 + 1×500 images, same pipeline). Confirmed the diagnosis in the ticket body and added py-spy dumps + Postgres lock-graph snapshots. Key findings beyond what's already written up:

Evidence collected

py-spy dumps of the ml-results celery worker with CELERY_WORKER_CONCURRENCY=8 — 7 of 8 ForkPoolWorker threads stuck in exactly:

wait (psycopg/connection.py:957)
execute (psycopg/cursor.py:719)
... cachalot/monkey_patch.py ...
_update_job_progress (ami/jobs/tasks.py:493)
process_nats_pipeline_result (ami/jobs/tasks.py:224)

The 8th was stuck in JobLogHandler.emit → save → ... on the same row. Row throughput under this contention: 0.3 tasks/s (measured across a 20s window, 8 workers). With one worker alone, ~0.7 tasks/s.

pg_stat_activity under load — consistent pattern: one gunicorn backend idle in transaction for 130-140s, last query UPDATE jobs_job SET logs = ..., blocking a chain of 6-10 other UPDATEs/SELECTs. Cross-referenced the blocking graph:

blocked_pid  blocker_pid  blocker_state         blocker_xact_age
3525453      3525445      idle in transaction   136s
3525454      3525453      active                135s
3525488      3525454      active                135s
3525411      3525454      active                135s
...

The view-side root blocker, when we eliminated the ML-worker-side contention (stubbed _update_job_progress) and reduced ml concurrency to 1, was a gunicorn request thread holding a tx open across process_nats_pipeline_result.delay() — broker RPC under ATOMIC_REQUESTS=True with the row lock already taken from the prior job.logger.info(...) UPDATE.

Fixes tested live (in order)

1. Drop refresh_from_db in JobLogHandler.emit — removes the nested SELECT-cursor issue from the ticket body. Helpful but not sufficient.
2. CELERY_WORKER_CONCURRENCY: "1" on the ml-results worker — eliminates worker-vs-worker contention on _update_job_progress's select_for_update. Surfaces the gunicorn side.
3. Stub emit + _update_job_progress to no-op — validated that jobs_job row-lock contention is the dominant bottleneck. Detections still land; UI progress freezes; queue drains much faster (though cachalot-on-save_results is the next bottleneck, ~2.5s/task).
4. transaction.on_commit(lambda: ...apply_async(...)) wrapping .delay() in /result, plus cachalot_disabled() around process_nats_pipeline_result — fixes the view-awaits-broker variant. Doesn't fix the workers-stack-on-select_for_update variant if concurrency >1.

Bottom line: fix #1 (recommended in the ticket) addresses part of the problem. It does not address:

• _update_job_progress itself holds select_for_update on the same row (ami/jobs/tasks.py:493), independently of logging. Under concurrent ADC result delivery for one job, ml workers serialize on that lock — identical throughput collapse, just without the log-write trigger.
• ATOMIC_REQUESTS=True + per-iteration job.logger.info inside /result's for-loop — 4-16 UPDATEs in a single view tx, each taking the row lock briefly but the tx doesn't commit until the view returns.

New implementation options beyond A/B/C

D. Replace select_for_update in _update_job_progress with atomic jsonb_set.

The progress object is an intentionally-generic JSON blob per Job (so that non-ML job types can surface arbitrary status counts in the UI). Dedicated integer columns would break that polymorphism. But Postgres jsonb_set + a conditional WHERE can atomically increment a nested integer in-place without a read-modify-write cycle:

UPDATE jobs_job
SET progress = jsonb_set(
    progress,
    '{stages,1,params,0,value}',           -- path to the `processed` count
    to_jsonb(
        COALESCE((progress#>>'{stages,1,params,0,value}')::int, 0) + %(delta)s
    )
)
WHERE id = %(pk)s
  AND ((progress#>>'{stages,1,progress}')::float IS NULL
       OR (progress#>>'{stages,1,progress}')::float < %(new_progress)s);

• Single statement → no select_for_update, no row lock held across Python code.
• Conditional WHERE handles the "don't regress progress" guarantee the current code uses max(existing, new) for.
• Works over the existing generic JSON shape — no schema migration, no loss of job-type flexibility.
• Django ORM form via Func/RawSQL + F() — uglier but wrappable in a helper update_progress_field(job_pk, path, delta).

Downsides:

• The JSON path is fragile if the progress schema changes (but it's already implicitly fragile — current code uses job.progress.get_stage(stage) which depends on the same structure).
• Postgres-only (fine, antenna is already Postgres-specific).
• Harder to read than select_for_update + Python math.

E. Decorate /result and /tasks with @transaction.non_atomic_requests.

One-line change per view (plus @non_atomic_requests at the ViewSet level won't work — needs to be per-method via @method_decorator or via the DB config). Removes the long-lived view tx entirely; each ORM call is its own autocommit tx, released immediately. The per-iteration job.logger.info UPDATEs still hit the row lock but only while their UPDATE is running (ms scale), not across the whole request lifetime.

Pairs well with D (fixes the worker side) — together they decouple the view and worker contention paths from each other.

Downside: the view must not rely on request-level atomicity. Current /result view body does 1 job.logger.info(...) per loop iteration + queued_tasks.append (in-memory) + transaction.on_commit(...) — no multi-write atomicity depended on. Likewise /tasks. Safe to apply.

F. Drop the per-iteration job.logger.info("Queued pipeline result: ...") in /result.

It mirrors the module-logger line one line above it — redundant. Keeping the module-level logger.info preserves ops-level visibility (container logs) without adding N UPDATEs per POST. Cheapest win, zero risk.

On (C) Redis-buffered logs

One caveat on the ticket's proposal (C): with CACHALOT_CACHE_RANDOM = True and cachalot's write-invalidation wrappers, a LPUSH-only emit is not quite free on the connection — no DB write, but the flush task still does a bulk-insert + invalidation. Layering C on top of A (as the ticket suggests) avoids this by making both the per-emit write and the flush path go to JobLog, not the Job.logs JSON.

Reproduction

Not VM/staging-specific. The pathology is pure Postgres (row lock + ATOMIC_REQUESTS + cachalot) + Python concurrency. To reproduce in the full-stack local dev compose:

• WEB_CONCURRENCY=4 or higher on django (default is 1 in dev — hides it)
• CELERY_WORKER_CONCURRENCY=8 on the ml-results worker
• Kick off 3+ concurrent async_api jobs via manage.py test_ml_job_e2e
• Either run an ADC worker locally against the same API, or use the existing chaos_monkey / a simple shell loop posting fake /result payloads to simulate load

The staging VM setup just reaches the load threshold faster because two real H100 workers pump results through simultaneously; the Postgres end of the path is the same code and will show the same graph.

Recommendations

For the first PR:

• D + E + F together — smallest diff that fully breaks the contention on both sides. Keep generic JSON progress. No migrations. No frontend changes.
• Keep the ticket's recommended A (dedicated JobLog table) as a follow-up — it's still the structurally right move and decouples the log-write path from any job-progress work forever.
• Keep the already-demonstrated @transaction.on_commit(lambda: ...apply_async(...)) pattern for any view-side .delay() (whether or not ATOMIC_REQUESTS is removed from that specific view — it's defensive and fires only on successful commit).
• Keep cachalot_disabled() around process_nats_pipeline_result (and probably save_results). Invalidation cost on per-detection writes is ~50% of task time.

Adjacent fixes from the ticket body (CONN_HEALTH_CHECKS, move job.logger.* outside transaction.atomic, narrow the bare except Exception in emit) still apply and are cheap.