#342 feat(storage): WebhookAuditSink close() lifecycle — stop signal, drain, thread-leak fix (#337)

[Anonymous]

Originally created by: Maqbool61

Slice

• Feature / Slice: F8.5 follow-up — WebhookAuditSink lifecycle (#317)
• Closes [#337]

What this PR does

Adds a close() shutdown path to WebhookAuditSink, caches thread-start
failure as an inert sink, adds a pytest fixture that cleans up worker threads
after each test, and replaces two time.sleep races with deterministic
Event-based sync.

Changes

src/doberman/storage/sinks.py

• _worker loop now checks not self._stop_event.is_set() (1 s queue timeout
  lets it notice the signal promptly). A drain loop after the main loop flushes
  records already in the queue before exiting.
• close(drain_timeout_s=5.0): marks sink inert immediately (concurrent
  emit() after close() → silent no-op), sets the stop event, joins the
  worker with a bounded timeout. Idempotent; never raises.
• Thread.start() wrapped in try/except: failure leaves _active=False so
  subsequent decisions don't retry the thread spawn — no retry storm under
  resource pressure.

tests/unit/test_webhook_audit_sink.py

• active_sink fixture: yields a live sink with a urlopen stub and calls
  close() in teardown — no worker threads leak across tests.
• All 15 tests that construct live sinks now call sink.close() explicitly or
  use the fixture.
• test_drop_oldest_when_queue_full and test_drop_counter_increments_on_each_overflow:
  replaced time.sleep(0.05) with a worker_blocked Event — deterministic
  under any CI load.
• 6 new tests covering the lifecycle contract (see below).

New tests (47 total, up from 45)

Test	Proves
test_close_stops_worker_thread	Worker thread is not alive after close() returns
test_close_is_idempotent	Second close() is a no-op, never raises
test_emit_after_close_is_noop	emit() after close() discards silently, no enqueue
test_close_drains_queued_records	All queued records are POSTed before worker exits
test_close_on_inert_sink_is_noop	close() on a no-config sink never raises
test_thread_start_failure_leaves_sink_inert	Start failure → _active=False, no retry

Full test suite

Run locally across all 152 unit test files: 2279 passed, 5 skipped, 0 failures.
The 5 skips are pre-existing and unrelated.

Decision-path contract unchanged

emit() remains non-blocking. close() is never called from the decision
path. Sink failures remain swallowed and isolated.

Security checklist

• [x] Fails closed on error / uncertainty
• [x] No secret, full file, or unredacted prompt logged or committed
• [x] Any guardrail/learning change is raise-only (no silent loosening)
• [x] Every BLOCK/AUTH carries reason codes + a human explanation

[Anonymous]

Originally posted by: Maqbool61

Fixed. The active-state transition and enqueue are now atomic under _state_lock.

What changed (sinks.py)

Added _state_lock = threading.Lock() to __init__. emit() acquires it for the entire check-and-enqueue as one atomic unit. close() acquires it before flipping _active = False, so it must wait for any in-progress emit() to finish putting before it stops the worker. Records enqueued while close() was waiting on the lock are delivered by the drain loop before the worker exits.

Regression test added

test_emit_close_race_no_stranded_record deterministically forces the exact interleaving you flagged via a put_nowait monkeypatch that blocks emit() inside the lock until close() has been called. Asserts captured == 1 — the record must be delivered, never stranded.

Smoke test results (20 forced trials of the exact interleaving)

Test | Result -- | -- Fixed code: 20 trials of the race interleaving — stranded records | 0 / 20 Old code (no lock): same 20 trials — stranded records | 20 / 20 close() drains 5 queued records before stopping | ✅ emit() after close() → silent no-op | ✅ Worker not alive after close() | ✅ close() idempotent | ✅ Thread-start failure → inert sink | ✅

Full unit suite: 2279 passed, 5 skipped, 0 failures (152 files).

[Anonymous]

Originally posted by: Maqbool61

@fu351 I think it is ready to merge. Can you confirm it?

[Anonymous]

Ticket changed by: fu351

• status: open --> closed

[Anonymous]

Originally posted by: fu351

The fix is exactly right: taking _state_lock across both the active-check and the enqueue in emit(), and the same lock around the _active = False flip in close(), closes the window where a record could slip in after the worker exits. close() has to wait for any in-flight emit() to release the lock, and any emit() that starts afterward sees _active=False under that same lock. The audit path stays fail-safe too — a full queue or an error is swallowed, never raised into the decision path. Merging.

One honest note on the regression test. test_emit_close_race_no_stranded_record pauses the enqueue only briefly, so the worker is usually still parked in queue.get() when the deferred put_nowait finally fires — which means it likely passes on the pre-fix code too, not only on the fixed code. The production fix is correct regardless (I checked the lock directly); it's the test's proof that's weaker than it looks. If you feel like a quick follow-up, forcing the worker to fully exit before the deferred put — shrinking the poll interval, or patching get — would make the test fail red on the old code and truly pin the race.

For a next issue, [#245] (OpenTelemetry AuditSink) is the natural one: same enqueue-only, drain-on-close sink pattern you just got right, one seam over. Want me to assign it to you?

Thanks, Maqbool61.

[Anonymous]

Originally posted by: Maqbool61

The fix is exactly right: taking _state_lock across both the active-check and the enqueue in emit(), and the same lock around the _active = False flip in close(), closes the window where a record could slip in after the worker exits. close() has to wait for any in-flight emit() to release the lock, and any emit() that starts afterward sees _active=False under that same lock. The audit path stays fail-safe too — a full queue or an error is swallowed, never raised into the decision path. Merging.

One honest note on the regression test. test_emit_close_race_no_stranded_record pauses the enqueue only briefly, so the worker is usually still parked in queue.get() when the deferred put_nowait finally fires — which means it likely passes on the pre-fix code too, not only on the fixed code. The production fix is correct regardless (I checked the lock directly); it's the test's proof that's weaker than it looks. If you feel like a quick follow-up, forcing the worker to fully exit before the deferred put — shrinking the poll interval, or patching get — would make the test fail red on the old code and truly pin the race.

For a next issue, [#245] (OpenTelemetry AuditSink) is the natural one: same enqueue-only, drain-on-close sink pattern you just got right, one seam over. Want me to assign it to you?

Thanks, Maqbool61.

I will do the follow up and yes assign me to the [#245].