zentralopensource / zentral / 29831792481

89%

push

github

Recycle the DB connection for all preprocessors on a recoverable DB error

The preprocess worker runs a long-lived consumer loop with no request cycle,
so Django never recycles its pooled connection (CONN_HEALTH_CHECKS and
CONN_MAX_AGE are driven by the request_started / request_finished signals).
When a pooler/proxy drops an idle connection, the next commit raises
InterfaceError / OperationalError.

Only the inventory preprocessor caught that, and it reused the same dead
connection, so every following inventory_machine_snapshot message failed with
"connection already closed" until the process restarted — the worker
re-enqueued relentlessly and dead-lettered a large share of check-ins. The
other preprocessors (mdm, jamf, puppet, wsone) had no handling and just let the
worker crash.

Add iter_preprocessed_events(), a single entry point every queue backend uses
to run a preprocessor: it resolves the preprocessor for the routing key, yields
the events it produces, and on a recoverable DB error drops the dead connection
with close_old_connections() and raises RetryLater so the retry and the
following messages reconnect. The three backends (SQS, Pub/Sub, kombu) share it
instead of each duplicating the routing-key lookup and retry handling. The
inventory preprocessor no longer handles the DB error itself; it only re-raises
it so its deterministic-drop can't swallow it.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

41 of 41 new or added lines in 5 files covered. (100.0%)

47546 of 53427 relevant lines covered (88.99%)

0.89 hits per line