From dd661be2f19a4a27c4eb775171c648a82f2b8283 Mon Sep 17 00:00:00 2001 From: Nikhil Arora Date: Thu, 3 Sep 2026 21:22:34 +0530 Subject: [PATCH] Add AsyncWorker, async dispatch and serverless Implements Phase 6: a background AsyncWorker and non-blocking Logger dispatch. Logger gains async_dispatch, max_queue_size and backpressure (drop_oldest/drop_newest/block), plus flush()/flush_async()/close(). Child loggers share the parent's worker. Adds serverless helpers (with_lambda / with_cloud_function / with_azure_function) that flush loggers before handler return. Transport.flush() hook introduced. Config loading maps new async keys. Documentation (README/CHANGELOG) updated and comprehensive tests added for worker, async logging and serverless behavior. --- CHANGELOG.md | 34 +++ README.md | 82 ++++++- logquill/__init__.py | 6 + logquill/config.py | 19 +- logquill/logger.py | 101 +++++++-- logquill/plugins/alerting_plugin.py | 9 +- logquill/serverless.py | 83 +++++++ logquill/transports/transport.py | 10 + logquill/worker.py | 165 ++++++++++++++ tests/test_adapters/test_langchain_adapter.py | 2 +- tests/test_logger_async.py | 142 ++++++++++++ tests/test_serverless.py | 92 ++++++++ ...criteria.py => test_tracing_end_to_end.py} | 11 +- tests/test_worker.py | 205 ++++++++++++++++++ 14 files changed, 933 insertions(+), 28 deletions(-) create mode 100644 logquill/serverless.py create mode 100644 logquill/worker.py create mode 100644 tests/test_logger_async.py create mode 100644 tests/test_serverless.py rename tests/{test_phase5_exit_criteria.py => test_tracing_end_to_end.py} (91%) create mode 100644 tests/test_worker.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 45c530b..a8a51db 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,40 @@ All notable changes to this project are documented in this file. ## Unreleased +- Phase 6, async worker, shutdown & serverless safety, complete: + - `AsyncWorker` — a bounded, in-memory queue backed by a single daemon + thread, with a configurable `backpressure` policy for what happens once + that bound is hit under a sustained burst: `drop_oldest` (default, + evicts the oldest queued item), `drop_newest` (discards the item that + just overflowed the queue), or `block` (the submitting thread waits for + space instead of dropping anything). Either drop policy logs at most one + warning per minute while actively dropping, not one per drop. + - `Logger(async_dispatch=True, max_queue_size=10_000, backpressure= + "drop_oldest")` — moves each record's transport writes and `after_log` + plugin hooks onto that background thread, so `.info()`/`.error()`/... + return without waiting on a transport's I/O; `before_log` hooks still + run synchronously, since a later hook or transport needs to see their + result in order. `Logger.child()` shares its parent's worker rather than + starting a second background thread. + - `Logger.flush(timeout=None)` / `await Logger.flush_async(timeout=None)` + — drain any queued records and flush each transport's own internal + buffer (`Transport.flush()`, a new no-op-by-default hook; already + matched by `BatchingTransport`'s existing buffered-batch flush) without + closing anything, so the logger stays usable right after. `Logger.close + (timeout=5.0)` now drains the queue (up to `timeout`) before closing + every transport. + - `with_lambda(logger_or_loggers, timeout=5.0)` — wraps a handler so + `flush()`/`flush_async()` runs before the handler's result or exception + reaches the caller, covering both sync and `async def` handlers. + Flushes rather than closes, since a warm serverless container reuses + the same `Logger`/transports on its next invocation. `with_cloud_function` + and `with_azure_function` are the same decorator under a name that + reads naturally at each platform's own handler definition — the + flush-before-return behavior needed is identical across all three. + - `load_config`/`logger_from_file`/`logger_from_env` accept the new + `"async_dispatch"`/`"max_queue_size"`/`"backpressure"` config keys, + mapped straight onto the matching `Logger` constructor arguments. + ## 0.5.0 - 2026-09-01 - `LangGraphAdapter` (`pip install logquill[langgraph]`) — corrects an diff --git a/README.md b/README.md index af58a33..72d0fc1 100644 --- a/README.md +++ b/README.md @@ -27,9 +27,10 @@ for what's landed so far. - **Config from file/env** — `load_config(dict)`, `logger_from_file(path)` (JSON/YAML), `logger_from_env()` build a `Logger` from one config shape — see [Config](#config) - **Plugin pipeline** — `ContextPlugin`, `RedactPlugin` (by key), `PIIRedactPlugin` (by pattern), `SamplingPlugin` (with tail-based elevation), `TamperEvidentPlugin` (hash-chained logs), `TraceContextPlugin` (cross-service trace correlation), and `AlertingPlugin` (`SlackAlertPlugin`/`PagerDutyAlertPlugin`/`EmailAlertPlugin`, deduplicated) out of the box; a broken plugin can't crash logging; `.use()` also accepts a plain function, no subclassing required (see [Plugins](#plugins)) - **Agentic & harness tracing** — `.child()` loggers, `RunPlugin`, `.thought()/.action()/.observation()/.decision()`, `with agent_log.span(...)`, and framework adapters — `LangChainAdapter` (`pip install logquill[langchain]`), `LangGraphAdapter` (`pip install logquill[langgraph]`, adds checkpoint interrupt/resume events on top), `CrewAIAdapter` (`pip install logquill[crewai]`), `LlamaIndexAdapter` (`pip install logquill[llamaindex]`), and `AutoGenAdapter` (`pip install logquill[autogen]`) — see [Agentic & harness tracing](#agentic--harness-tracing) +- **Non-blocking async dispatch** — `Logger(async_dispatch=True)` moves transport writes onto a background thread with a bounded queue and a configurable backpressure policy (`drop_oldest`/`drop_newest`/`block`); `flush()`/`flush_async()` and a `with_lambda`/`with_cloud_function`/`with_azure_function` decorator make serverless shutdown safe — see [Async dispatch & serverless safety](#async-dispatch--serverless-safety) - **Zero required runtime dependencies** — stdlib only; `aiohttp` is opt-in, for async HTTP - **Typed throughout** — `mypy --strict` clean on the public API -- *(planned)* non-blocking async dispatch, `contextvars`-based context propagation — see `CHANGELOG.md` +- *(planned)* `contextvars`-based context propagation — see `CHANGELOG.md` ## Install @@ -700,6 +701,85 @@ event classes; that's a real divergence, not just a detail, so it needs its own adapter rather than reusing this one. `autogen-core` is never imported unless you import `logquill.adapters.autogen` yourself. +## Async dispatch & serverless safety + +By default, every log call dispatches to its transports synchronously — a +slow or down sink adds latency directly to the call that triggered it. Pass +`async_dispatch=True` to move dispatch (the transport writes, plus the +`after_log` plugin hooks that follow them) onto a background thread instead, +so `.info()`/`.error()`/... return as soon as `before_log` plugin hooks have +run, without waiting on any transport's I/O: + +```python +from logquill import ConsoleTransport, HTTPTransport, Logger + +logger = Logger( + "app", + transports=[ConsoleTransport(), HTTPTransport("https://logs.example.com/ingest")], + async_dispatch=True, + max_queue_size=10_000, # bounds memory if a transport stalls + backpressure="drop_oldest", # or "drop_newest" / "block" +) +``` + +`max_queue_size` bounds how many not-yet-dispatched records can pile up in +memory if a transport stalls (a down HTTP endpoint, a full disk). Once that +bound is hit, `backpressure` decides what happens next — `"drop_oldest"` +(default) evicts the oldest queued record to make room, `"drop_newest"` +discards the record that just triggered the overflow, and `"block"` makes +the calling thread wait for space instead of dropping anything. Either drop +policy logs at most one warning per minute while actively dropping, not one +per dropped record. `Logger.child()` shares its parent's queue/background +thread rather than starting a second one. + +Call `logger.close()` on ordinary process shutdown — it drains any records +still queued (up to an optional `timeout`, default 5 seconds) and then +closes every transport: + +```python +logger.close(timeout=5.0) +``` + +For code that keeps running afterward (a request handler, a serverless +invocation), use `logger.flush()` instead — it drains the queue and flushes +each transport's own internal buffer (see `BatchingTransport`) *without* +closing anything, so the logger is still usable right after: + +```python +logger.flush(timeout=2.0) # sync callers +await logger.flush_async(timeout=2.0) # async callers — awaits instead of blocking +``` + +### Serverless: flush before the container freezes + +A serverless execution environment (AWS Lambda, GCP Cloud Functions, Azure +Functions) can freeze or tear down immediately after your handler returns — +a record still sitting in the async queue at that instant may never reach +its transport. `with_lambda` wraps a handler so `flush()`/`flush_async()` +happens automatically, on both a normal return and an exception, before +control goes back to the platform: + +```python +from logquill import ConsoleTransport, Logger, with_lambda + +logger = Logger("app", transports=[ConsoleTransport()], async_dispatch=True) + + +@with_lambda(logger) +def handler(event, context): + logger.info("processing request", request_id=event["requestId"]) + return {"statusCode": 200} +``` + +It flushes, never closes — a warm container reuses the same `Logger`/ +transports on its next invocation, and `close()` would release resources +(an open file handle, a pooled connection) that invocation needs. Works +with `async def` handlers too, and accepts a list of loggers if a handler +logs through more than one. `with_cloud_function`/`with_azure_function` are +the same decorator under a name that reads naturally at each platform's own +handler definition — the flush-before-return behavior is identical across +all three. + ## Development ```bash diff --git a/logquill/__init__.py b/logquill/__init__.py index 4fd6076..c2744a2 100644 --- a/logquill/__init__.py +++ b/logquill/__init__.py @@ -16,6 +16,7 @@ from logquill.plugins.tamper_evident_plugin import TamperEvidentPlugin from logquill.plugins.trace_context_plugin import TraceContextPlugin from logquill.records import LogRecord +from logquill.serverless import with_azure_function, with_cloud_function, with_lambda from logquill.transports.batching_transport import BatchingTransport from logquill.transports.cloud.app_insights_transport import AppInsightsTransport from logquill.transports.cloud.cloud_logging_transport import CloudLoggingTransport @@ -40,12 +41,14 @@ from logquill.transports.sql.postgres_transport import PostgresTransport from logquill.transports.sql.sqlite_transport import SQLiteTransport from logquill.transports.transport import CollectingTransport, Transport +from logquill.worker import AsyncWorker __version__ = "0.5.0" __all__ = [ "AlertingPlugin", "AppInsightsTransport", + "AsyncWorker", "BaseQueueTransport", "BaseSQLTransport", "BatchingTransport", @@ -93,5 +96,8 @@ "logger_from_env", "logger_from_file", "parse_level", + "with_azure_function", + "with_cloud_function", + "with_lambda", "__version__", ] diff --git a/logquill/config.py b/logquill/config.py index f53cda4..945d80b 100644 --- a/logquill/config.py +++ b/logquill/config.py @@ -96,7 +96,10 @@ def load_config(data: dict[str, Any], *, name: str = "app") -> Logger: "plugins": [ {"type": "context", "options": {"service": "api"}}, {"type": "sampling", "options": {"rate": 0.1}} - ] + ], + "async_dispatch": true, + "max_queue_size": 10000, + "backpressure": "drop_oldest" } Each transport/plugin entry needs either `"type"` (a built-in shortcut — @@ -105,12 +108,24 @@ def load_config(data: dict[str, Any], *, name: str = "app") -> Logger: dictConfig` resolves a `class` key — the same trust boundary: only use this with config you trust, the same as any other deployment config). `"options"` becomes that class's constructor keyword arguments. + + `"async_dispatch"`/`"max_queue_size"`/`"backpressure"` are optional and + map directly onto `Logger`'s constructor arguments of the same name — + see there for what each does. """ logger_name = data.get("name", name) level = data.get("level", "INFO") transports = _build(data.get("transports"), _TRANSPORT_TYPES) plugins = _build(data.get("plugins"), _PLUGIN_TYPES) - return Logger(logger_name, level=level, transports=transports, plugins=plugins) + return Logger( + logger_name, + level=level, + transports=transports, + plugins=plugins, + async_dispatch=data.get("async_dispatch", False), + max_queue_size=data.get("max_queue_size", 10_000), + backpressure=data.get("backpressure", "drop_oldest"), + ) def logger_from_file(path: str | Path, *, name: str = "app") -> Logger: diff --git a/logquill/logger.py b/logquill/logger.py index 09d7c17..152e6ea 100644 --- a/logquill/logger.py +++ b/logquill/logger.py @@ -10,6 +10,7 @@ from logquill.records import LogRecord, create_record from logquill.span import SpanContext, current_span_id from logquill.transports.transport import Transport +from logquill.worker import AsyncWorker, BackpressurePolicy _logger = logging.getLogger("logquill") @@ -21,13 +22,33 @@ def __init__( level: int | str | Level = Level.INFO, transports: list[Transport] | None = None, plugins: list[Plugin | MiddlewareFunc] | None = None, + async_dispatch: bool = False, + max_queue_size: int = 10_000, + backpressure: BackpressurePolicy = "drop_oldest", ) -> None: + """`async_dispatch=True` moves per-record transport writes (and the + `after_log` plugin hooks that follow them) onto a background thread, + via an internal `AsyncWorker` — so `.info()`/`.error()`/... return + without waiting on a transport's I/O. `before_log` plugin hooks + still run synchronously on the caller's thread, since they can + filter/transform the record and a later hook or transport needs to + see the result in order. + + `max_queue_size`/`backpressure` are only meaningful with + `async_dispatch=True` — see `AsyncWorker` for what each + `backpressure` policy does under a sustained burst. + """ self.name = name self._level = parse_level(level) self.transports: list[Transport] = list(transports) if transports else [] self.plugins: list[Plugin] = [] for plugin in plugins or []: self.use(plugin) + self._worker: AsyncWorker | None = ( + AsyncWorker(max_queue_size=max_queue_size, backpressure=backpressure) + if async_dispatch + else None + ) @property def level(self) -> Level: @@ -62,10 +83,49 @@ def child(self, name: str, /, **fixed_meta: Any) -> Logger: child_logger = Logger(f"{self.name}.{name}", level=self._level, transports=self.transports) if fixed_meta: child_logger.use(ContextPlugin(**fixed_meta)) + # Share the parent's worker (if any) rather than spinning up a second + # background thread: both loggers write to the same transport + # instances, so their dispatch belongs on the same queue/thread. + child_logger._worker = self._worker return child_logger - def close(self) -> None: - """Close every attached transport. Call on shutdown to flush buffered writes.""" + def flush(self, timeout: float | None = None) -> bool: + """Wait for every record already submitted for async dispatch to + finish writing, then flush each transport's own internal buffer + (e.g. `BatchingTransport`) — without closing anything. + + Unlike `close()`, safe to call repeatedly mid-lifetime: this is + what `with_lambda`/`with_cloud_function`/`with_azure_function` call + before a serverless invocation returns, since a warm container + reuses this same `Logger`/its transports on the next invocation. + + Returns whether the async queue (if any) fully drained within + `timeout`; always `True` when `async_dispatch` wasn't enabled, since + dispatch already happened synchronously before this call. + """ + drained = self._worker.drain(timeout) if self._worker is not None else True + for transport in self.transports: + transport.flush() + return drained + + async def flush_async(self, timeout: float | None = None) -> bool: + """`asyncio`-friendly `flush()` — awaits the drain instead of + blocking the calling thread. See `flush()`. + """ + drained = await self._worker.drain_async(timeout) if self._worker is not None else True + for transport in self.transports: + transport.flush() + return drained + + def close(self, timeout: float | None = 5.0) -> None: + """Stop async dispatch (draining any queued records first, up to + `timeout` seconds) and close every attached transport. Call on + process shutdown — after this, the transports may release + resources a later log call would need, so don't call it on a + `Logger` you intend to keep using (see `flush()` for that case). + """ + if self._worker is not None: + self._worker.close(timeout) for transport in self.transports: transport.close() @@ -74,6 +134,26 @@ def _notify_error(self, plugin: Plugin, exc: Exception, record: LogRecord) -> No with contextlib.suppress(Exception): plugin.on_error(exc, record) + def _dispatch(self, record: LogRecord) -> None: + """Write `record` to every transport and run `after_log` hooks. + This is the half of `_log` that does I/O — with `async_dispatch=True` + it runs on the worker thread instead of the caller's, which is the + entire non-blocking-dispatch contract in one method boundary. + """ + for transport in self.transports: + try: + transport.write(transport.format(record), record) + except Exception: + # a transport that can't format or write this particular record + # (e.g. a circular reference in `meta`) must not crash the caller + _logger.exception("%s: failed to write a log record", type(transport).__name__) + + for plugin in self.plugins: + try: + plugin.after_log(record) + except Exception as exc: + self._notify_error(plugin, exc, record) + def _log(self, level: Level, message: str, meta: dict[str, Any]) -> LogRecord | None: if level < self._level: return None @@ -93,19 +173,10 @@ def _log(self, level: Level, message: str, meta: dict[str, Any]) -> LogRecord | return None record = result - for transport in self.transports: - try: - transport.write(transport.format(record), record) - except Exception: - # a transport that can't format or write this particular record - # (e.g. a circular reference in `meta`) must not crash the caller - _logger.exception("%s: failed to write a log record", type(transport).__name__) - - for plugin in self.plugins: - try: - plugin.after_log(record) - except Exception as exc: - self._notify_error(plugin, exc, record) + if self._worker is not None: + self._worker.submit(lambda: self._dispatch(record)) + else: + self._dispatch(record) return record diff --git a/logquill/plugins/alerting_plugin.py b/logquill/plugins/alerting_plugin.py index 86b1a9f..4a189a1 100644 --- a/logquill/plugins/alerting_plugin.py +++ b/logquill/plugins/alerting_plugin.py @@ -30,10 +30,11 @@ class AlertingPlugin(Plugin): default: level + logger + message) fires `send_alert` right away, on a short-lived background thread — so the log call that triggered it is never blocked on a webhook, SMTP handshake, or any other I/O, even if - the destination is slow or unreachable. This stands in for the shared - async dispatch queue a later phase will introduce; once that queue - exists, `AlertingPlugin` can route through it instead of spawning its - own thread per alert. + the destination is slow or unreachable. This plugin spawns its own + thread per alert rather than routing through `Logger`'s shared + `AsyncWorker` (see `logquill/worker.py`), since a plugin hook runs + before dispatch is decided and has no handle on that queue; unifying + the two is a possible future simplification, not a correctness gap. Any further record matching the same dedupe key within `dedupe_window_seconds` of the first is *not* sent again — it just diff --git a/logquill/serverless.py b/logquill/serverless.py new file mode 100644 index 0000000..8f5ad44 --- /dev/null +++ b/logquill/serverless.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +import functools +import inspect +from typing import Any, Callable, Sequence, TypeVar, Union, cast + +from logquill.logger import Logger + +F = TypeVar("F", bound=Callable[..., Any]) + +LoggerOrLoggers = Union[Logger, Sequence[Logger]] + + +def _as_loggers(loggers: LoggerOrLoggers) -> tuple[Logger, ...]: + return (loggers,) if isinstance(loggers, Logger) else tuple(loggers) + + +def with_lambda(loggers: LoggerOrLoggers, *, timeout: float | None = 5.0) -> Callable[[F], F]: + """Wrap a serverless function handler so any log records still queued + on a non-blocking `Logger` (`async_dispatch=True`) are flushed before + the handler's result (or exception) is returned to the platform. + + This exists because a serverless execution environment can freeze or + be torn down immediately after the handler returns — a record still + sitting in `AsyncWorker`'s queue at that instant may never actually + reach its transport. Wrapping the handler makes "flush before return" + automatic instead of something every handler has to remember to do + itself. + + Calls `logger.flush(timeout)` (or `flush_async` for an `async def` + handler), never `logger.close()` — a warm container reuses the same + `Logger`/transports on its next invocation, and `close()` would + release resources (e.g. an open file handle, a pooled HTTP connection) + that next invocation needs. Flushing happens whether the handler + returns normally or raises, so an unhandled error still ships its logs. + + Despite the name, this also covers GCP Cloud Functions and Azure + Functions handlers — see `with_cloud_function`/`with_azure_function`, + which are the same decorator under a name matching that platform. The + flush-before-return behavior needed is identical across all three; + only the handler's own argument signature (which this decorator never + inspects) differs per platform. + + `loggers` accepts a single `Logger` or a sequence of them, for a + handler that logs through more than one (e.g. an app logger and a + separate audit logger). + """ + targets = _as_loggers(loggers) + + def decorator(func: F) -> F: + if inspect.iscoroutinefunction(func): + + @functools.wraps(func) + async def async_wrapper(*args: Any, **kwargs: Any) -> Any: + try: + return await func(*args, **kwargs) + finally: + for logger in targets: + await logger.flush_async(timeout) + + return cast(F, async_wrapper) + + @functools.wraps(func) + def wrapper(*args: Any, **kwargs: Any) -> Any: + try: + return func(*args, **kwargs) + finally: + for logger in targets: + logger.flush(timeout) + + return cast(F, wrapper) + + return decorator + + +#: Same decorator as `with_lambda`, named for a GCP Cloud Functions handler. +#: See `with_lambda`'s docstring — the flush-before-return behavior is +#: identical across platforms; only the name differs, to read naturally at +#: each platform's own handler definition. +with_cloud_function = with_lambda + +#: Same decorator as `with_lambda`, named for an Azure Functions handler. +with_azure_function = with_lambda diff --git a/logquill/transports/transport.py b/logquill/transports/transport.py index 87abd7d..d47806a 100644 --- a/logquill/transports/transport.py +++ b/logquill/transports/transport.py @@ -20,6 +20,16 @@ def format(self, record: LogRecord) -> str: @abstractmethod def write(self, formatted: str, record: LogRecord) -> None: ... + def flush(self) -> None: # noqa: B027 — intentionally optional to override + """Push any internally buffered records out now, without releasing + the transport's resources — see `BatchingTransport`, whose + buffered-but-not-yet-sent batch this drains. No-op unless a + transport overrides it. Called by `Logger.flush()`/`flush_async()` + (e.g. from `with_lambda` before a serverless container may freeze), + which — unlike `Logger.close()` — must not close anything a warm + container will reuse on its next invocation. + """ + def close(self) -> None: # noqa: B027 — intentionally optional to override """Flush/release resources on shutdown. No-op unless a transport overrides it.""" diff --git a/logquill/worker.py b/logquill/worker.py new file mode 100644 index 0000000..4670edc --- /dev/null +++ b/logquill/worker.py @@ -0,0 +1,165 @@ +from __future__ import annotations + +import asyncio +import logging +import threading +import time +from collections import deque +from typing import Callable, Literal + +_logger = logging.getLogger("logquill") + +BackpressurePolicy = Literal["drop_oldest", "drop_newest", "block"] +_VALID_POLICIES = ("drop_oldest", "drop_newest", "block") + +WorkItem = Callable[[], None] + +#: Don't re-emit the "queue full, dropping records" warning on every single +#: drop — a stalled consumer under a sustained burst would otherwise flood +#: whatever's watching this logger's own diagnostic channel with thousands +#: of near-identical lines. Once per minute is enough to notice the +#: condition without becoming the next noisy-logging problem. +_DROP_WARNING_INTERVAL_SECONDS = 60.0 + + +class AsyncWorker: + """Background-thread dispatch queue backing `Logger(async_dispatch=True)`. + + A single daemon thread pulls submitted work items (closures) off a + bounded, in-memory queue and runs them one at a time, so `submit()` + returns to the caller without waiting on whatever the item actually + does (e.g. a transport's blocking I/O) — that's the whole point: a log + call's caller never blocks on the sink. + + `max_queue_size` bounds memory: a stalled or slow consumer (a down + HTTP sink, a full disk) can't grow this queue without limit. `backpressure` + decides what happens once that bound is hit: + + - `"drop_oldest"` (default) — evict the oldest queued item to make room + for the new one. Favors recent records over old ones and never blocks + the submitting thread. + - `"drop_newest"` — discard the item just submitted instead. Favors + records already queued over the newest one. + - `"block"` — the submitting thread waits for space. Favors completeness + over the non-blocking guarantee; only choose this if the caller can + tolerate an occasional stall. + + Either drop policy logs at most one warning per minute while actively + dropping, not one per dropped item. + """ + + def __init__( + self, + *, + max_queue_size: int = 10_000, + backpressure: BackpressurePolicy = "drop_oldest", + ) -> None: + if max_queue_size < 1: + raise ValueError(f"max_queue_size must be >= 1, got {max_queue_size}") + if backpressure not in _VALID_POLICIES: + raise ValueError(f"backpressure must be one of {_VALID_POLICIES}, got {backpressure!r}") + self.max_queue_size = max_queue_size + self.backpressure = backpressure + self._queue: deque[WorkItem] = deque() + self._pending = 0 + self._closed = False + self._cond = threading.Condition() + self._last_drop_warning = 0.0 + self._thread = threading.Thread(target=self._run, name="logquill-worker", daemon=True) + self._thread.start() + + @property + def qsize(self) -> int: + """Number of items currently queued (not yet started). Testing/introspection only.""" + with self._cond: + return len(self._queue) + + def submit(self, item: WorkItem) -> None: + """Enqueue `item` for the background thread to run. Never blocks + the caller unless `backpressure="block"` and the queue is full. + Silently dropped if the worker has already been `close()`d. + """ + with self._cond: + if self._closed: + return + if len(self._queue) >= self.max_queue_size: + if self.backpressure == "block": + while len(self._queue) >= self.max_queue_size and not self._closed: + self._cond.wait() + if self._closed: + return + elif self.backpressure == "drop_newest": + self._warn_dropping() + return + else: # drop_oldest + self._queue.popleft() + self._pending -= 1 + self._warn_dropping() + self._queue.append(item) + self._pending += 1 + self._cond.notify_all() + + def _warn_dropping(self) -> None: + now = time.monotonic() + if now - self._last_drop_warning >= _DROP_WARNING_INTERVAL_SECONDS: + self._last_drop_warning = now + _logger.warning( + "AsyncWorker: queue full at max_queue_size=%d, dropping records " + "under backpressure=%r — the consumer (a transport) isn't keeping " + "up, or max_queue_size is set too low for this burst rate", + self.max_queue_size, + self.backpressure, + ) + + def _run(self) -> None: + while True: + with self._cond: + while not self._queue and not self._closed: + self._cond.wait() + if not self._queue: + return + item = self._queue.popleft() + self._cond.notify_all() + try: + item() + except Exception: + _logger.exception("AsyncWorker: a queued work item raised") + finally: + with self._cond: + self._pending -= 1 + self._cond.notify_all() + + def drain(self, timeout: float | None = None) -> bool: + """Block the calling thread until every item submitted so far has + finished running, or `timeout` seconds elapse (`None` waits + indefinitely). Returns whether the queue actually fully drained. + """ + deadline = None if timeout is None else time.monotonic() + timeout + with self._cond: + while self._pending > 0: + if deadline is None: + self._cond.wait() + else: + remaining = deadline - time.monotonic() + if remaining <= 0: + return self._pending == 0 + self._cond.wait(remaining) + return True + + async def drain_async(self, timeout: float | None = None) -> bool: + """`asyncio`-friendly `drain()`: runs the blocking wait in an + executor thread so it doesn't block the event loop while waiting. + """ + loop = asyncio.get_running_loop() + return await loop.run_in_executor(None, self.drain, timeout) + + def close(self, timeout: float | None = 5.0) -> bool: + """Drain, then stop the background thread. Idempotent. Returns + whether the drain completed within `timeout` before shutdown. + """ + drained = self.drain(timeout) + with self._cond: + self._closed = True + self._cond.notify_all() + self._thread.join(timeout=1.0) + return drained diff --git a/tests/test_adapters/test_langchain_adapter.py b/tests/test_adapters/test_langchain_adapter.py index a2ea606..741cda8 100644 --- a/tests/test_adapters/test_langchain_adapter.py +++ b/tests/test_adapters/test_langchain_adapter.py @@ -72,7 +72,7 @@ def test_full_run_reconstructs_span_tree(monkeypatch: pytest.MonkeyPatch) -> Non llm_start, llm_end, tool_start, tool_end, finish, chain_close = sink.records # 5+ steps, at least one nested span — sorted by (parent_span_id, - # span_id) reconstructs the exact tree, per Phase 5's exit criterion. + # span_id) reconstructs the exact call tree. assert llm_start["meta"]["span_id"] == str(llm_run) assert llm_start["meta"]["parent_span_id"] == str(chain_run) assert llm_end["meta"]["span_id"] == str(llm_run) diff --git a/tests/test_logger_async.py b/tests/test_logger_async.py new file mode 100644 index 0000000..7f70060 --- /dev/null +++ b/tests/test_logger_async.py @@ -0,0 +1,142 @@ +from __future__ import annotations + +import threading +import time +from typing import Any + +from logquill.logger import Logger +from logquill.records import LogRecord +from logquill.transports.transport import CollectingTransport + +_POLL_TIMEOUT = 2.0 +_POLL_INTERVAL = 0.005 + + +def _wait_until(predicate: Any, timeout: float = _POLL_TIMEOUT) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(_POLL_INTERVAL) + return predicate() + + +class StallingTransport(CollectingTransport): + """A `CollectingTransport` whose `write()` blocks until released — stands + in for a slow/down sink so tests can assert the caller isn't the one + waiting on it. + """ + + def __init__(self) -> None: + super().__init__() + self.gate = threading.Event() + self.write_started = threading.Event() + + def write(self, formatted: str, record: LogRecord) -> None: + self.write_started.set() + self.gate.wait(timeout=5.0) + super().write(formatted, record) + + +def test_sync_dispatch_is_synchronous_by_default() -> None: + transport = StallingTransport() + transport.gate.set() # don't actually stall — just confirm default behavior + logger = Logger("app.test", transports=[transport]) + + logger.info("hello") + + assert len(transport.records) == 1 + + +def test_async_dispatch_returns_before_the_transport_write_completes() -> None: + transport = StallingTransport() + logger = Logger("app.test", transports=[transport], async_dispatch=True) + + start = time.monotonic() + record = logger.info("hello") + elapsed = time.monotonic() - start + + assert record is not None + assert elapsed < 1.0 # didn't block on the stalled transport + assert transport.records == [] + + transport.gate.set() + assert logger.flush(timeout=2.0) is True + assert len(transport.records) == 1 + + +def test_close_drains_queued_records_before_returning() -> None: + transport = StallingTransport() + transport.gate.set() + logger = Logger("app.test", transports=[transport], async_dispatch=True) + + for i in range(200): + logger.info("msg", i=i) + + logger.close(timeout=5.0) + assert len(transport.records) == 200 + + +def test_flush_does_not_close_the_transport() -> None: + transport = StallingTransport() + transport.gate.set() + logger = Logger("app.test", transports=[transport], async_dispatch=True) + + logger.info("hello") + assert logger.flush(timeout=2.0) is True + assert transport.closed is False + + logger.info("world") + assert logger.flush(timeout=2.0) is True + assert len(transport.records) == 2 + + +async def test_flush_async_awaits_the_drain() -> None: + transport = StallingTransport() + logger = Logger("app.test", transports=[transport], async_dispatch=True) + + logger.info("hello") + assert transport.records == [] + + transport.gate.set() + assert await logger.flush_async(timeout=2.0) is True + assert len(transport.records) == 1 + + +def test_child_logger_shares_the_parent_worker() -> None: + transport = StallingTransport() + transport.gate.set() + parent = Logger("app", transports=[transport], async_dispatch=True) + child = parent.child("child") + + parent.info("from parent") + child.info("from child") + + assert parent.flush(timeout=2.0) is True + assert len(transport.records) == 2 + + +def test_drop_oldest_backpressure_bounds_memory_under_a_stalled_transport() -> None: + transport = StallingTransport() + logger = Logger( + "app.test", + transports=[transport], + async_dispatch=True, + max_queue_size=50, + backpressure="drop_oldest", + ) + + logger.info("first") # popped immediately, stalls the worker on `gate` + assert _wait_until(transport.write_started.is_set) + + for i in range(5000): + logger.info("burst", i=i) + + assert logger._worker is not None + assert logger._worker.qsize <= 50 + + transport.gate.set() + logger.close(timeout=5.0) + # Bounded: nowhere near 5001 records made it through, but the caller was + # never blocked and the process never grew the queue past its limit. + assert len(transport.records) <= 52 diff --git a/tests/test_serverless.py b/tests/test_serverless.py new file mode 100644 index 0000000..8d01b33 --- /dev/null +++ b/tests/test_serverless.py @@ -0,0 +1,92 @@ +from __future__ import annotations + +import pytest + +from logquill.logger import Logger +from logquill.serverless import with_azure_function, with_cloud_function, with_lambda +from logquill.transports.transport import CollectingTransport + + +def test_with_lambda_flushes_a_sync_handler_before_returning() -> None: + transport = CollectingTransport() + logger = Logger("app.test", transports=[transport], async_dispatch=True) + + @with_lambda(logger) + def handler(event: dict, context: object) -> str: + logger.info("handling", event=event) + return "ok" + + result = handler({"key": "value"}, None) + + assert result == "ok" + # Flushed synchronously before `with_lambda` returned control — no + # sleeping/polling needed to observe the record. + assert len(transport.records) == 1 + + +def test_with_lambda_flushes_even_when_the_handler_raises() -> None: + transport = CollectingTransport() + logger = Logger("app.test", transports=[transport], async_dispatch=True) + + @with_lambda(logger) + def handler(event: dict, context: object) -> str: + logger.error("about to fail") + raise RuntimeError("boom") + + with pytest.raises(RuntimeError, match="boom"): + handler({}, None) + + assert len(transport.records) == 1 + + +async def test_with_lambda_supports_async_handlers() -> None: + transport = CollectingTransport() + logger = Logger("app.test", transports=[transport], async_dispatch=True) + + @with_lambda(logger) + async def handler(event: dict, context: object) -> str: + logger.info("handling async", event=event) + return "ok" + + result = await handler({}, None) + + assert result == "ok" + assert len(transport.records) == 1 + + +def test_with_lambda_does_not_close_the_transport() -> None: + transport = CollectingTransport() + logger = Logger("app.test", transports=[transport], async_dispatch=True) + + @with_lambda(logger) + def handler(event: dict, context: object) -> str: + logger.info("invocation 1") + return "ok" + + handler({}, None) + handler({}, None) + + assert transport.closed is False + assert len(transport.records) == 2 + + +def test_with_lambda_accepts_multiple_loggers() -> None: + app_transport = CollectingTransport() + audit_transport = CollectingTransport() + app_logger = Logger("app", transports=[app_transport], async_dispatch=True) + audit_logger = Logger("audit", transports=[audit_transport], async_dispatch=True) + + @with_lambda([app_logger, audit_logger]) + def handler(event: dict, context: object) -> None: + app_logger.info("app event") + audit_logger.info("audit event") + + handler({}, None) + + assert len(app_transport.records) == 1 + assert len(audit_transport.records) == 1 + + +def test_with_cloud_function_and_with_azure_function_are_the_same_behavior() -> None: + assert with_cloud_function is with_lambda + assert with_azure_function is with_lambda diff --git a/tests/test_phase5_exit_criteria.py b/tests/test_tracing_end_to_end.py similarity index 91% rename from tests/test_phase5_exit_criteria.py rename to tests/test_tracing_end_to_end.py index 7abcb70..7fb39df 100644 --- a/tests/test_phase5_exit_criteria.py +++ b/tests/test_tracing_end_to_end.py @@ -1,12 +1,13 @@ -"""End-to-end tests matching Phase 5's exit criteria verbatim (CLAUDE.md): +"""End-to-end tracing tests: 1. A full agent run (5+ steps, at least one nested span) reconstructs exact order and nesting when sorted by `meta.span_id`/`meta.parent_span_id`. 2. A synthetic multi-service call chain shares one `trace_id` end to end. -3. A `LangChainAdapter`-instrumented chain reconstructs its exact call tree - with zero manual instrumentation beyond passing the handler in once — - covered separately in `tests/test_adapters/test_langchain_adapter.py`, - since it needs the optional-dependency faking setup that lives there. + +A `LangChainAdapter`-instrumented chain reconstructing its exact call tree +with zero manual instrumentation is covered separately in +`tests/test_adapters/test_langchain_adapter.py`, since it needs the +optional-dependency faking setup that lives there. """ from logquill.logger import Logger diff --git a/tests/test_worker.py b/tests/test_worker.py new file mode 100644 index 0000000..bd0d01e --- /dev/null +++ b/tests/test_worker.py @@ -0,0 +1,205 @@ +from __future__ import annotations + +import threading +import time + +import pytest + +from logquill.worker import AsyncWorker + +_POLL_TIMEOUT = 2.0 +_POLL_INTERVAL = 0.005 + + +def _wait_until(predicate: object, timeout: float = _POLL_TIMEOUT) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): # type: ignore[operator] + return True + time.sleep(_POLL_INTERVAL) + return predicate() # type: ignore[operator] + + +def test_rejects_invalid_max_queue_size() -> None: + with pytest.raises(ValueError): + AsyncWorker(max_queue_size=0) + + +def test_rejects_invalid_backpressure_policy() -> None: + with pytest.raises(ValueError): + AsyncWorker(backpressure="explode") # type: ignore[arg-type] + + +def test_submitted_items_run_on_the_background_thread_and_drain_completes() -> None: + worker = AsyncWorker() + seen: list[int] = [] + for i in range(50): + worker.submit(lambda i=i: seen.append(i)) + + assert worker.drain(timeout=2.0) is True + assert seen == list(range(50)) + worker.close() + + +def test_drain_below_the_limit_loses_nothing() -> None: + # A burst *below* the configured queue limit must drain with zero + # dropped records — only exceeding the limit should ever lose one. + worker = AsyncWorker(max_queue_size=1000) + lock = threading.Lock() + seen: list[int] = [] + for i in range(500): + worker.submit(lambda i=i: (lock.acquire(), seen.append(i), lock.release())) + + assert worker.close(timeout=5.0) is True + assert sorted(seen) == list(range(500)) + + +def test_drop_oldest_bounds_the_queue_and_keeps_the_newest_items() -> None: + gate = threading.Event() + started = threading.Event() + worker = AsyncWorker(max_queue_size=10, backpressure="drop_oldest") + lock = threading.Lock() + seen: list[int] = [] + + def item(i: int) -> None: + if i == -1: + started.set() + gate.wait(timeout=5.0) + with lock: + seen.append(i) + + # Stall the very first item, and wait for the worker thread to actually + # start running it (i.e. pop it off the queue), before firing the burst + # below — otherwise it could itself be evicted as "oldest" before the + # worker ever gets to it, which would make this test flaky. + worker.submit(lambda: item(-1)) + assert _wait_until(started.is_set) + for i in range(1000): + worker.submit(lambda i=i: item(i)) + + assert worker.qsize <= 10 + gate.set() + assert worker.close(timeout=5.0) is True + + # The oldest queued items were evicted to make room, so only the most + # recently submitted ones (plus the stalled first item) survive. + assert -1 in seen + assert len(seen) <= 12 + assert max(seen) == 999 + + +def test_drop_newest_bounds_the_queue_and_keeps_the_oldest_items() -> None: + gate = threading.Event() + started = threading.Event() + worker = AsyncWorker(max_queue_size=10, backpressure="drop_newest") + lock = threading.Lock() + seen: list[int] = [] + + def item(i: int) -> None: + if i == -1: + started.set() + gate.wait(timeout=5.0) + with lock: + seen.append(i) + + worker.submit(lambda: item(-1)) + assert _wait_until(started.is_set) + for i in range(1000): + worker.submit(lambda i=i: item(i)) + + assert worker.qsize <= 10 + gate.set() + assert worker.close(timeout=5.0) is True + + assert -1 in seen + assert len(seen) <= 12 + # Later submissions were the ones discarded, so nothing near the tail + # of the burst should have made it through. + assert 999 not in seen + + +def test_block_backpressure_waits_for_space_and_drops_nothing() -> None: + gate = threading.Event() + worker = AsyncWorker(max_queue_size=5, backpressure="block") + lock = threading.Lock() + seen: list[int] = [] + + def item(i: int) -> None: + gate.wait(timeout=5.0) + with lock: + seen.append(i) + + def burst() -> None: + for i in range(50): + worker.submit(lambda i=i: item(i)) + + burst_thread = threading.Thread(target=burst) + burst_thread.start() + + # The submitting thread should be stalled, blocked on the full queue, + # not silently dropping or racing ahead. + assert not _wait_until(lambda: not burst_thread.is_alive(), timeout=0.3) + + gate.set() + burst_thread.join(timeout=5.0) + assert not burst_thread.is_alive() + assert worker.close(timeout=5.0) is True + assert sorted(seen) == list(range(50)) + + +def test_drain_respects_timeout_when_a_stalled_item_never_finishes() -> None: + gate = threading.Event() + worker = AsyncWorker() + worker.submit(lambda: gate.wait(timeout=5.0)) + + assert worker.drain(timeout=0.1) is False + + gate.set() + assert worker.drain(timeout=5.0) is True + worker.close() + + +def test_a_raising_work_item_does_not_stop_the_worker() -> None: + worker = AsyncWorker() + ran_after = threading.Event() + + def boom() -> None: + raise RuntimeError("boom") + + worker.submit(boom) + worker.submit(ran_after.set) + + assert _wait_until(ran_after.is_set) + assert worker.close(timeout=2.0) is True + + +def test_close_is_idempotent_and_drains_a_pending_backlog_first() -> None: + worker = AsyncWorker() + lock = threading.Lock() + seen: list[int] = [] + for i in range(20): + worker.submit(lambda i=i: (lock.acquire(), seen.append(i), lock.release())) + + assert worker.close(timeout=2.0) is True + assert worker.close(timeout=1.0) is True + assert sorted(seen) == list(range(20)) + + +def test_submit_after_close_is_a_silent_no_op() -> None: + worker = AsyncWorker() + worker.close() + ran = [] + worker.submit(lambda: ran.append(True)) + assert worker.qsize == 0 + assert ran == [] + + +async def test_drain_async_awaits_completion_without_blocking_the_event_loop() -> None: + worker = AsyncWorker() + seen: list[int] = [] + for i in range(20): + worker.submit(lambda i=i: seen.append(i)) + + assert await worker.drain_async(timeout=2.0) is True + assert sorted(seen) == list(range(20)) + worker.close()