Queued log transport should detect (and ideally survive) fork/snapshot-restore without consumer intervention

Problem

QueuedLogDestination starts its listener thread at construction and deliberately carries no fork hooks or pid checks (documented in destinations/queued.py). That is the right simplicity trade-off inside the package, but it moves a correctness obligation onto every consumer whose process forks or restores from a memory snapshot: they must know to call restart_observability() at the right lifecycle moment, after credentials are re-materialized.

The failure mode when a consumer misses this is the worst kind: the service serves traffic normally while every Google-bound record silently drops (reason="full" once the queue fills, dead listener thread never drains). Nothing crashes, nothing alerts on the request path.

We just paid this cost in practice: PolicyEngine/policyengine-household-api#1597 adopts 1.4.0 on Modal, whose workers restore from memory snapshots (@modal.enter(snap=True) builds the app — and with it the queued transport — at snapshot creation). The fix was a restart_observability() call in the post-restore hook, plus a unit test pinning the ordering. The next consumer with a forking or snapshotting runtime (gunicorn --preload with fork, celery prefork, another snapshot platform) has to rediscover all of this from scratch.

Ask

Make dead-listener state detectable — and ideally self-healing — inside the package, so consumers are safe by default rather than safe by convention.

Two tiers, smallest first:

1. Detect and report loudly (minimum). QueuedLogDestination.emit() already counts drops with a throttled report. Add a cheap staleness check — e.g. compare os.getpid() against the pid captured at construction — and when it differs, report through the internal-error channel with an actionable message naming restart_observability() and a distinct drop reason (reason="stale-process" rather than "full"). This turns a silent outage into a diagnosable one for the cost of one integer comparison per emit.

2. Self-heal (stretch, only if it stays simple). On detecting a stale pid, rebuild the destination the way restart_log_destinations() does. Constraints that killed the previous attempt at this problem space (#23) still apply and are worth restating: fail-open above all, rebuild fresh clients rather than resurrect old ones, no locking or state machine on the hot path, and no rebuild-inside-emit races — if the rebuild can't be made trivially safe (e.g. single os.register_at_fork hook rather than emit-path mutation), tier 1 alone is still a big win.

Related prior art: #22 (accept/emit split), #24 (half-open recovery for disabled destinations, closed as deferred), and the consumer-side integration in PolicyEngine/policyengine-household-api#1597.

🤖 Generated with Claude Code

Start by reading destinations/queued.py, then trace restart_log_destinations() and restart_observability(), including the consumer-side post-restore ordering described in PolicyEngine/policyengine-household-api#1597. Review the existing drop-reporting path and relevant unit coverage. Done means stale-process state is reported with restart guidance and a distinct drop reason; self-healing is optional only if it remains fail-open and race-free.