Add AsyncWorker, async dispatch and serverless

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

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

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__",
 ]

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:

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
 

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