Route LokiHandler emits without a running loop through sync fallback

[woodwardmw]

Summary

• LokiHandler.emit() no longer drops logs with a generic "RuntimeError" when called from threads that have no running event loop (e.g. work dispatched via asyncio.to_thread, or any thread-pool task inside an async app).
• It now lazily spawns an internal SyncLokiHandler the first time it sees a loopless thread, and hands off the prebuilt payload via a new SyncLokiHandler.enqueue_payload() helper. Async-only callers pay nothing — the fallback is built on demand.
• Set enable_thread_fallback=False on the constructor to restore the strict async-only behaviour.
• close() and aclose() now tear the fallback worker down alongside the aiohttp session.

Motivation

Caller-side symptom was the unhelpful repeated line:

Failed to send log to Loki (async): RuntimeError

with no detail (the handler intentionally suppresses the exception's str()). The actual cause is asyncio.get_running_loop() raising RuntimeError("no running event loop") inside emit, because the calling thread isn't the one running the loop. This is normal in async apps that do CPU/IO work via asyncio.to_thread or a thread pool, and the dropped records are typically the most useful ones (deep helper functions).

Test plan

• Existing async-loop path still creates a task on the running loop
• Loopless emit now routes through the fallback (verified with mock)
• enable_thread_fallback=False still logs the send-failure
• Fallback is lazy — _fallback is None until first loopless emit
• Fallback is built once and reused
• close() tears the fallback worker down
• Full suite green: pytest -q → 55 passed

[Copilot]

Pull request overview

This PR updates LokiHandler.emit() to avoid dropping log records when called from a thread without a running asyncio event loop by lazily routing those records through a background-threaded SyncLokiHandler fallback. It also adds configurability to disable this behavior and expands test coverage around the new fallback lifecycle.

Changes:

• Add enable_thread_fallback option and a lazy _get_thread_fallback() path to route loopless emit() calls to SyncLokiHandler.
• Add SyncLokiHandler.enqueue_payload() to accept pre-built Loki payloads from the async handler.
• Extend handler tests to cover fallback routing, laziness/reuse, teardown, queue-full behavior, and post-close behavior.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 5 comments.

File	Description
observability_library/handler.py	Implements lazy sync fallback for loopless threads; adds teardown of fallback in close()/aclose()/__del__.
observability_library/sync_handler.py	Adds enqueue_payload() helper to enqueue pre-built payloads.
tests/test_handler.py	Adds tests validating fallback routing, lifecycle (lazy/reuse/teardown), and failure cases.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[claude]

Review summary — 5 findings across integrity, consistency, and test coverage.

[Integrity, medium] Lock race in _get_thread_fallback (handler.py:71): the inner double-check guard if self._fallback is None: does not re-verify _enable_thread_fallback under the lock. A concurrent close() can set _fallback = None and _enable_thread_fallback = False, then a thread that already passed the outer (unlocked) check re-enters the lock and rebuilds the fallback after close has torn it down — leaving an unjoined daemon thread. Fix: if self._fallback is None and self._enable_thread_fallback:.

[Integrity, medium] test_async_send_logs_send_failure_on_exception (test_handler.py:240-253) was inadvertently broken by the new loop-mismatch check. The test does not set fake_session._loop, so getattr(MagicMock(), '_loop', None) returns an auto-generated child MagicMock ≠ current_loop, causing the handler to discard fake_session and create a real aiohttp.ClientSession. The test still passes (via a connection-error path) but no longer exercises fake_session.post.side_effect. Fix: add fake_session._loop = asyncio.get_running_loop() as done in test_async_send_posts_payload_with_auth_header.

[Integrity, low] Stale session not closed on loop change (handler.py:132-139): when session_loop is not current_loop the old ClientSession is silently replaced with no cleanup. A best-effort asyncio.run(old.close()) or at minimum a comment explaining why cleanup is skipped would prevent resource leaks in the Modal-worker case.

[Consistency, low] getattr(self._session, '_loop', None) (handler.py:122-124) reads a private, undocumented aiohttp attribute. The safe fallback (missing → session recreated) means breakage is silent rather than loud. A comment tying this to a specific aiohttp version keeps the fragility visible at upgrade time.

[Integrity, very low] SyncLokiHandler.enqueue_payload (sync_handler.py:56-67) accepts items after close() — they pile up in the queue forever once the drain thread exits. Upstream protection in LokiHandler.close() covers the normal path, but the lock race above can bypass it. Adding if self._stop.is_set(): return False makes the method self-contained.

Overall the PR is well-structured, the test coverage is thorough, and the lazy-fallback and double-checked locking patterns are sound — the race described above is the only correctness gap.

[claude]

Finding 1/4 — [Integrity, medium] Lock race in _get_thread_fallback (handler.py:71)

The double-checked locking pattern is almost correct but the inner guard — inside the lock — only checks self._fallback is None; it does not re-verify _enable_thread_fallback:

The race:

1. Thread A reads _enable_thread_fallback = True at the outer (unlocked) check on line 66.
2. Thread B (close()) acquires the lock, sets _enable_thread_fallback = False, clears _fallback = None, releases the lock.
3. Thread A acquires the lock, sees _fallback is None, and rebuilds the fallback after close() has already torn it down — leaving an unjoined daemon thread.

Fix: change the inner guard from if self._fallback is None: to if self._fallback is None and self._enable_thread_fallback:.

[claude]

Finding 2/4 — [Integrity, medium] test_async_send_logs_send_failure_on_exception inadvertently broken (test_handler.py:240-253)

This pre-existing test does not set fake_session._loop. With the new loop-mismatch check added by this PR:

session_loop = (
    getattr(self._session, '_loop', None) if self._session else None
)
if ... or session_loop is not current_loop:
    self._session = aiohttp.ClientSession(...)   # real session created

getattr(MagicMock(), '_loop', None) returns an auto-generated child MagicMock (not current_loop), so the handler silently discards fake_session and creates a real aiohttp.ClientSession. The test still passes — the real session gets a connection error and log_send_failure is called — but it no longer exercises fake_session.post.side_effect = RuntimeError('boom').

Compare: test_async_send_posts_payload_with_auth_header was correctly updated in this PR with:

fake_session._loop = _asyncio.get_running_loop()

The same one-liner is needed in test_async_send_logs_send_failure_on_exception.

[claude]

Finding 3/4 — [Integrity, low] Stale session not closed on loop change (handler.py:132-139)

When session_loop is not current_loop, the old aiohttp.ClientSession is overwritten without being closed — the connector and its underlying TCP connections are never explicitly released:

if (
    self._session is None
    or self._session.closed
    or session_loop is not current_loop   # old session silently dropped
):
    self._session = aiohttp.ClientSession(...)

In the Modal-worker scenario the old loop may already be torn down, making a proper async close impossible. Either document this explicitly in the comment (e.g. "old loop is torn down; best-effort close not attempted") or attempt asyncio.run(old_session.close()) on a fresh loop — matching the pattern already used in close() for the session teardown.

[claude]

Finding 4/4 — [Consistency, low] Private aiohttp.ClientSession._loop attribute (handler.py:122-124) + minor: enqueue_payload after close() (sync_handler.py:56-67)

4a — getattr(self._session, '_loop', None) reads a private, undocumented aiohttp attribute. The safe fallback (attribute missing → None → session recreated) means breakage will be silent rather than loud. A comment like # aiohttp 3.x stores the bound loop as a private _loop attribute tied to a known version range makes this visible at upgrade time.

4b (very low) — SyncLokiHandler.enqueue_payload does not guard against a stopped worker. After close() sets _stop, the drain loop exits and any payload pushed here sits in the queue permanently. Upstream protection in LokiHandler.close() (which clears _fallback) covers the normal path, but the lock race in finding 1 can bypass that guard. Adding if self._stop.is_set(): return False at the top makes the method self-contained.

[alejandroaquintero]

@claude

[claude]

Code Review — PR #13: Route LokiHandler emits without a running loop through sync fallback

Overall this is a well-motivated, well-documented change. The lazy-init pattern, TOCTOU-guarded construction, and teardown sequencing are solid. Test coverage is comprehensive. A few issues worth addressing before merge, ordered by severity.

---

🔴 Medium — Race: _get_thread_fallback can spawn a never-closed worker after close()

observability_library/handler.py lines 66–86

The outer pre-lock checks read _enable_thread_fallback and _fallback without holding the lock:

if not self._enable_thread_fallback:   # (A) read outside lock
    return None
if self._fallback is not None:         # (B) read outside lock
    return self._fallback
with self._fallback_lock:
    if self._fallback is None:
        self._fallback = SyncLokiHandler(...)

close() holds _fallback_lock when it sets _enable_thread_fallback = False. But a thread that passed check (A) before close() acquired the lock will enter the with block after close() releases it, see _fallback is None, and create a fresh SyncLokiHandler that nobody ever closes — leaking a daemon thread and its queue.

Fix: re-check the gate inside the lock:

with self._fallback_lock:
    if not self._enable_thread_fallback:   # re-check under lock
        return None
    if self._fallback is None:
        self._fallback = SyncLokiHandler(...)

The unguarded fast path (if self._fallback is not None: return self._fallback) is safe to keep for the hot path once the initial construction is done; the lock only needs to protect construction and the close→gate interaction.

---

🟡 Medium — Old aiohttp session leaks when the loop changes

observability_library/handler.py lines 132–139

When session_loop is not current_loop, self._session is overwritten without closing the old one:

if (
    self._session is None
    or self._session.closed
    or session_loop is not current_loop
):
    self._session = aiohttp.ClientSession(...)

For session.closed, an unclosed session is the existing pre-PR behaviour (minor, well-known). But the loop-mismatch branch is new. The old session's connector and its underlying TCP connections are abandoned. In Modal-style multi-loop environments this can happen on every function invocation.

The old loop may already be closed, so you can't await old_session.close() there. A low-cost option is to call the sync connector close on a thread via asyncio.get_event_loop().run_in_executor on the old loop if it's still running, or at minimum suppress the ResourceWarning by calling old_session.connector.close() synchronously (connectors can be closed without a loop). Worth at least a note in the comment about why the teardown is intentionally skipped.

---

🟡 Low — __del__ reaches into SyncLokiHandler._stop (private attribute coupling)

observability_library/handler.py lines 257–259

fallback._stop.set()

__del__ on LokiHandler directly accesses a private implementation detail of SyncLokiHandler. If SyncLokiHandler is ever refactored (e.g. the event renamed, or replaced with a threading.Barrier), this breaks silently at GC time.

Prefer a public method on SyncLokiHandler:

def signal_stop(self) -> None:
    """Signal the worker to exit without joining it."""
    self._stop.set()

Then __del__ calls fallback.signal_stop() — stable interface, intent is clear.

---

🟡 Low — Duplicated fallback teardown block in close() and aclose()

observability_library/handler.py lines 179–204 and 222–237

Both methods contain an identical sequence: acquire _fallback_lock, flip _enable_thread_fallback, grab and clear _fallback, release the lock. The only difference is how the captured fallback is then closed (executor vs. sync). Extracting the shared preamble to a _take_fallback() -> Optional[SyncLokiHandler] helper would remove the duplication and make future changes to the teardown protocol (e.g. adding _session cleanup here) apply in one place:

def _take_fallback(self) -> Optional[SyncLokiHandler]:
    lock = getattr(self, "_fallback_lock", None)
    if lock is None:
        return None
    with lock:
        self._enable_thread_fallback = False
        fallback, self._fallback = self._fallback, None
    return fallback

---

🔵 Nit — asyncio imported mid-function in tests

tests/test_handler.py lines 217 and 294

import asyncio as _asyncio
fake_session._loop = _asyncio.get_running_loop()

import asyncio as _asyncio
await _asyncio.sleep(0)

asyncio is not imported at the module level in this file; both usages reach for it inside the function body with an aliased name. Moving import asyncio to the top of the module alongside the other stdlib imports is the project convention and avoids the visual noise of _asyncio.

---

🔵 Nit — test_emit_after_close_does_not_re_enqueue_to_dead_fallback missing exception-type assertion

tests/test_handler.py lines 178–185

The test verifies log_send_failure is called but doesn't assert which error type is passed. After close(), _enable_thread_fallback is False, so _get_thread_fallback() returns None and the NoRunningEventLoopError path is hit. Pinning that:

assert isinstance(failure.call_args.args[1], NoRunningEventLoopError)

would make the test self-documenting and guard against the fallback silently enqueuing into a dead worker (which would return True from enqueue_payload without logging anything).

---

✅ What's well done

• The double-checked locking pattern for construction is correct under CPython's GIL; the _enable_thread_fallback = False + log_send_failure on construction failure prevents unbounded retry noise.
• LOKI_DEBUG env var + _sanitise_exc_message regex is a clean opt-in for verbose diagnostics. The regex correctly handles ://user:password@host while leaving ://host:port/path untouched, and the "best-effort" framing is appropriate.
• __del__ correctly avoids calling fallback.close() (which would join the worker thread and stall interpreter shutdown) — the comment explaining why is exactly the right level of detail.
• The create_task → fallback path for the closed-loop race is a real edge case (worker/interpreter shutdown) and the test test_emit_falls_back_when_loop_create_task_raises_runtime_error covers it precisely.
• Test coverage across all new code paths is thorough.

[alejandroaquintero]

@claude-reviewer

[claude]

Fragile private API access — will silently break with aiohttp ≥ 3.9

ClientSession._loop is an aiohttp implementation detail that was removed in 3.9 when the library dropped explicit-loop support. With the project's constraint aiohttp>=3.8.0, any installation on 3.9+ will return None here, making session_loop is not current_loop (None is not <running loop>) permanently True. The result is that _async_send recreates the session on every call — connectors are never reused, and the abandoned sessions/connectors accumulate until GC collects them.

The tests don't catch this because they mock the session and set _loop manually:

fake_session._loop = _asyncio.get_running_loop()

so they pass regardless of what real aiohttp exposes.

Suggested fix — track the loop explicitly on the handler, not via aiohttp internals:

# in __init__
self._session_loop: Optional[asyncio.AbstractEventLoop] = None

# in _async_send, replace the session_loop / getattr lines with:
session_loop = self._session_loop
current_loop = asyncio.get_running_loop()
if (
    self._session is None
    or self._session.closed
    or session_loop is not current_loop
):
    self._session = aiohttp.ClientSession(
        timeout=aiohttp.ClientTimeout(total=self.timeout)
    )
    self._session_loop = current_loop

This avoids private-attribute coupling and works across all aiohttp versions.

[alejandroaquintero]

@claude-reviewer

[claude]

Bug: inner lock check doesn't re-validate _enable_thread_fallback — zombie worker thread on close race

The double-checked locking pattern here is incomplete. Consider this interleaving:

1. Thread A passes the outer if not self._enable_thread_fallback check (sees True)
2. Thread B calls close(), acquires the lock, sets _enable_thread_fallback = False, sets _fallback = None, releases the lock
3. Thread A acquires the lock, finds _fallback is None → True, and builds a new SyncLokiHandler

The code comment ("Flip the gate before joining so a concurrent emit() in _get_thread_fallback() can't spin up a new worker after we've torn the current one down") describes the intention correctly, but the guard isn't checked inside the lock. This new handler's worker thread leaks — _enable_thread_fallback is now False so nobody will ever call close() on it.

Suggested change

	if self._fallback is None:
	with self._fallback_lock:
	if self._fallback is None and self._enable_thread_fallback:

[claude]

Session leak when loop mismatch is detected

When session_loop is not current_loop, the old self._session is silently overwritten without being closed. The underlying aiohttp.TCPConnector (and any open sockets) are leaked and Python will emit ResourceWarning: Unclosed client session noise. Since we're already in an async context here, the old session can be closed before replacement:

old_session = self._session
if (
    old_session is None
    or old_session.closed
    or session_loop is not current_loop
):
    self._session = aiohttp.ClientSession(
        timeout=aiohttp.ClientTimeout(total=self.timeout)
    )
    if old_session is not None and not old_session.closed and session_loop is not current_loop:
        # old_session belongs to a different loop; schedule its close on
        # the new loop so we don't block.
        try:
            await old_session.close()
        except Exception:
            pass

The loop-mismatch case is admittedly awkward (the old session may already be broken), but at minimum the try/except prevents a close attempt from hiding real errors.

[claude]

Fragile dependency on aiohttp private attribute

aiohttp.ClientSession._loop is undocumented and has already been removed in some aiohttp builds. If it disappears, getattr(..., "_loop", None) returns None, making session_loop is not current_loop always True, which forces a new session on every _async_send call — effectively a connector-per-log-record regression.

If you need to handle the loop-mismatch case, consider storing the loop yourself when the session is created:

# in _async_send, after creating the session:
self._session = aiohttp.ClientSession(...)
self._session_loop = asyncio.get_running_loop()

Then check self._session_loop is not current_loop instead of accessing _loop on the aiohttp object. Add self._session_loop: Optional[asyncio.AbstractEventLoop] = None to __init__.

[claude]

Double-logging on first emit() when fallback construction fails

When SyncLokiHandler(...) raises here, this line logs the construction error. Control then returns None to emit(), which falls through to line 117:

log_send_failure("async", NoRunningEventLoopError())

So the same emit() call produces two error lines: one for the construction failure and one for NoRunningEventLoopError. An operator tailing the log will see a confusing pair with no explanation that they're from the same record.

Consider either:

• Not calling log_send_failure here (let the NoRunningEventLoopError at the call site be the sole signal; the construction failure is usually a config issue that will recur), or
• Setting _fallback to a sentinel and adjusting the caller to emit a single combined message.

[claude]

enqueue_payload silently succeeds on a stopped handler

After SyncLokiHandler.close() has joined the worker thread, this method still returns True — put_nowait succeeds because the queue isn't flushed, but no thread is left to drain it. The record is silently discarded.

LokiHandler.close() guards against this by zeroing _fallback before calling fallback.close(), so the normal sequential close-then-emit path is safe. However, a thread that already holds the _fallback reference (via the outer fast-path in _get_thread_fallback()) can still call enqueue_payload after the worker has stopped and get a false True.

Adding a _stop.is_set() guard makes the contract explicit and avoids misleading callers:

Suggested change

	self._queue.put_nowait(payload)
	try:
	if self._stop.is_set():
	return False
	self._queue.put_nowait(payload)
	return True

[claude]

Review summary

The motivation and general approach are sound — lazy fallback construction, the NoRunningEventLoopError / ThreadFallbackQueueFullError distinction, and the LOKI_DEBUG opt-in for richer diagnostics are all good additions. The test coverage is thorough for the sequential paths.

Issues found

1. Zombie worker thread on close() race (handler.py line 71 — blocking)
The inner if self._fallback is None: guard inside the lock does not re-check _enable_thread_fallback. A thread that passed the outer fast-path check just before close() acquired the lock will build a new SyncLokiHandler worker thread after close() has already torn the old one down — and nothing will ever close the new one. One-line fix: if self._fallback is None and self._enable_thread_fallback:.

2. Stale aiohttp session leak on loop mismatch (handler.py lines 132–139 — blocking)
When session_loop is not current_loop, the old session is overwritten without being closed. This leaks the TCPConnector and produces ResourceWarning: Unclosed client session noise in the Modal-worker scenario the fix is specifically targeting.

3. Private aiohttp._loop attribute (handler.py line 123 — should fix)
aiohttp.ClientSession._loop has been removed in some aiohttp versions. When it's absent getattr returns None, making the mismatch check always True — a silent per-record session churn regression. Store the loop yourself (self._session_loop = asyncio.get_running_loop()) when creating the session instead.

4. Double-logging when fallback construction fails (handler.py line 85 — minor)
The construction error is logged here, then emit() also logs a NoRunningEventLoopError for the same record, producing a confusing pair on first failure.

5. enqueue_payload returns True on a stopped handler (sync_handler.py line 64 — minor)
After close() the worker is gone but the queue still accepts items, so enqueue_payload returns True for records that will never be drained. A _stop.is_set() guard before put_nowait makes the contract accurate.

6. README not updated (documentation — required per contributing guidelines)
enable_thread_fallback is a new public constructor parameter and LOKI_DEBUG is a new operational control — both need to appear in the README per the contributing guidelines ("Any change to the public surface … must be reflected in README.md").

What's good

• Lazy fallback construction is the right call; async-only paths pay nothing.
• ThreadFallbackQueueFullError vs NoRunningEventLoopError class names giving operators a distinguishable signal via log_send_failure is clean.
• _sanitise_exc_message + LOKI_DEBUG opt-in is a well-reasoned improvement over the current silent class-name-only default.
• The close() / aclose() / __del__ teardown logic is careful about blocking the event loop.
• Test coverage for the new paths is comprehensive.

[claude]

Code Review

1. 🔴 Race: _get_thread_fallback() can create a worker after close() — handler.py:66–86

The gate checks (_enable_thread_fallback, _fallback is not None) run outside the lock. A thread that passes both checks and then blocks waiting for _fallback_lock can race with close():

1. Thread A passes the two pre-lock checks (flag is True, fallback is None).
2. close() acquires the lock, sets _enable_thread_fallback = False and _fallback = None, and releases.
3. Thread A acquires the lock, sees _fallback is None, and creates a brand-new SyncLokiHandler worker that will never be torn down.

Fix — re-check the flag inside the lock before constructing:

with self._fallback_lock:
    if self._fallback is None:
        if not self._enable_thread_fallback:   # re-check inside the lock
            return None
        try:
            self._fallback = SyncLokiHandler(...)

---

2. 🔴 Stale aiohttp session leaked on loop mismatch — handler.py:132–139

When session_loop is not current_loop, self._session is overwritten without the old (still-open) session being closed. Unlike the .closed branch (where a closed session has already released its resources), a loop-mismatched session holds a live connector and open sockets.

# Current code — stale session is silently abandoned:
if (
    self._session is None
    or self._session.closed
    or session_loop is not current_loop   # ← open session replaced, never closed
):
    self._session = aiohttp.ClientSession(...)

Fix — explicitly close the outgoing session before replacing it:

if (
    self._session is None
    or self._session.closed
    or session_loop is not current_loop
):
    old = self._session
    self._session = aiohttp.ClientSession(
        timeout=aiohttp.ClientTimeout(total=self.timeout)
    )
    if old and not old.closed:
        await old.close()

---

3. 🔴 test_async_send_logs_send_failure_on_exception silently tests the wrong path — tests/test_handler.py:240–253

This PR adds a loop-mismatch check in _async_send that reads getattr(self._session, "_loop", None). MagicMock auto-creates attributes on access, so getattr(fake_session, "_loop", None) returns a child MagicMock — not None — which is always is not current_loop. The handler therefore discards the fake session and creates a real aiohttp.ClientSession that tries to POST to http://loki/push. The test still passes because that connection is refused and log_send_failure is called, but it's exercising the connection-error path, not the RuntimeError("boom") path it claims to cover.

The PR already fixed test_async_send_posts_payload_with_auth_header for this exact reason (adding fake_session._loop = _asyncio.get_running_loop()). The same one-liner is needed here:

fake_session = MagicMock()
fake_session.closed = False
fake_session._loop = asyncio.get_running_loop()   # ← add this
fake_session.post = MagicMock(side_effect=RuntimeError("boom"))
handler._session = fake_session

---

4. 🟡 enqueue_payload narrow exception catch violates handler contract — sync_handler.py:63–67

enqueue_payload only catches queue.Full. Any other exception from put_nowait (e.g. a RuntimeError if the queue object is in an unexpected state) propagates uncaught through LokiHandler.emit(), which breaks the logging.Handler contract (handlers must not let exceptions escape emit). The risk is low in practice, but a broad catch is cheap:

except Exception:
    return False

---

Summary

#	Severity	File	Issue
1	🔴 Critical	handler.py:70	Post-close worker created; _enable_thread_fallback not re-checked inside lock
2	🔴 Critical	handler.py:132–139	Open aiohttp session leaked when loop changes
3	🔴 Critical	tests/test_handler.py:244	Loop-mismatch guard silently breaks existing test; wrong path covered
4	🟡 Moderate	sync_handler.py:63–67	Non-Full exceptions escape enqueue_payload into emit()

The overall design (lazy fallback, double-checked locking intent, LOKI_DEBUG credential scrubbing, teardown via close/aclose/__del__) is sound and the new test coverage is thorough. Issues 1–3 need to be addressed before merge.