Observability

engrava exposes a snapshot metrics API via await store.metrics(). The returned EngravaMetrics dataclass aggregates thought/edge counts, storage footprint, and a rolling-window search-latency histogram.

Quick Example

from engrava import SqliteEngravaCore
import aiosqlite

async def main() -> None:
    conn = await aiosqlite.connect("engrava.db")
    conn.row_factory = aiosqlite.Row
    store = SqliteEngravaCore(conn)
    await store.ensure_schema()
    try:
        metrics = await store.metrics()
        print(metrics.thoughts.total)
        print(metrics.edges.by_type)
        print(metrics.search_latency.p95_ms)
    finally:
        await conn.close()

Metrics Fields

store.metrics() returns a stable EngravaMetrics dataclass with:

• thoughts — counts by type and lifecycle status
• edges — counts by edge type
• storage — on-disk footprint for the main SQLite database and WAL
• search_latency — rolling-window p50/p95/p99 search latency
• measured — True only when this snapshot came from actually querying the store; False for the zero-filled placeholder returned when metrics.enabled is False. Check this before trusting a zero — an unmeasured snapshot and an honestly empty store both read thoughts.total == 0.

Configuration

metrics:
  enabled: true
  window_size: 1000

When enabled: false, store.metrics() returns a zero-filled snapshot and does not issue SQL queries. That snapshot has measured=False. Nothing else in the snapshot states whether a measurement happened — storage.db_bytes reads 0 for a disabled store and for a measured in-memory one, and search_latency.sample_count reads 0 for a disabled store and for a measured store that has simply served no searches yet. measured is the one field that states it directly.

CLI

engrava info renders the same snapshot the Python API returns, with one deliberate difference: its own schema_version is exposed as metrics_schema_version, alongside a database_schema_version field the Python snapshot does not carry (the database’s PRAGMA user_version, read separately).

engrava --db mydata.db info
engrava --db mydata.db --format json info

Notes

• The latency histogram tracks completed public search calls.
• Nested calls inside search_hybrid() are suppressed, so one hybrid search contributes one latency sample.
• This snapshot API tracks only aggregate counts and search latency — not individual events.

Production monitoring

store.metrics() is a pull snapshot — there is no built-in exporter. To monitor a deployment, scrape the snapshot on an interval and feed the fields into your metrics system (Prometheus, OpenTelemetry, StatsD, …).

Exporting the snapshot

The snapshot is a plain dataclass, so mapping it to any client is straightforward — but a plain Gauge is the wrong tool: it exposes a sample (0.0 by default) from the moment it is registered, whether or not it has ever been .set(), so a “register once, update on each scrape” scheme has a window where a scrape reads a value nobody measured. Use a custom collector instead — it queries the store fresh on every scrape and yields nothing when measured is False, so there is no registration window and no stale value between ticks.

A store belongs to one event loop (see Concurrency → “One store belongs to one event loop”): its internal locks bind to whichever loop first has to wait on one, and a later waiter from a different loop then raises RuntimeError. collect() calling a fresh asyncio.run(store.metrics()) on every scrape — a new loop each time — risks exactly that against a long-lived store built on a different (or now-closed) loop, once anything contends for it. The pattern below is written for a WSGI-style exposition (start_http_server()), where collect() runs synchronously on the server’s own thread with no persistent loop of its own: it runs the store, and every scrape, on one dedicated background thread’s loop instead of a fresh one per call, using run_coroutine_threadsafe to hand the coroutine to it and block for the result. It is the wrong shape for an ASGI application (e.g. one mounting prometheus_client.make_asgi_app()): there, collect() runs on the app’s own running loop, which is also where the store should already live — do the async store read directly in your own request handler (no thread hand-off needed, since you’re already on the store’s loop) and register a collector over the already-computed metric families instead.

import asyncio
import threading

from prometheus_client import REGISTRY
from prometheus_client.core import GaugeMetricFamily

from engrava import SqliteEngravaCore

_METRICS = (
    ("engrava_thoughts_total", "Total thoughts"),
    ("engrava_db_bytes", "Main database size in bytes"),
    ("engrava_wal_bytes", "WAL size in bytes"),
    ("engrava_search_p95_ms", "Search p95 latency (ms)"),
    ("engrava_search_p99_ms", "Search p99 latency (ms)"),
)


def _run_store_loop(ready: threading.Event, out: dict) -> None:
    """Construct the store here, then run this thread's loop forever.

    Everything the store ever does — every scrape included — runs on this one
    loop, since the store's own internal locks are asyncio.Lock instances
    that bind to whichever loop first has to wait on them. Construction can
    itself fail (a bad `engrava.yaml`, a corrupt journal) -- reported via
    `out["error"]` rather than left to hang the waiting main thread forever.
    """
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        out["store"] = loop.run_until_complete(SqliteEngravaCore.from_config("engrava.yaml"))
        out["loop"] = loop
    except BaseException as exc:  # noqa: BLE001 - reported on the main thread, not swallowed
        out["error"] = exc
        return
    finally:
        ready.set()
    loop.run_forever()


class EngravaCollector:
    """Runs every scrape on the store's own loop; yields nothing when unmeasured."""

    def __init__(self, store, loop: asyncio.AbstractEventLoop):
        self._store = store
        self._loop = loop

    def describe(self):
        # Sample-free descriptors so the registry learns names/help text at
        # registration time, without needing to run collect() (and therefore
        # the store query) from the registering thread.
        return [GaugeMetricFamily(name, help_text) for name, help_text in _METRICS]

    def collect(self):
        # Hand the coroutine to the store's own loop and block this (scrape)
        # thread for the result -- never asyncio.run(), which would spin up
        # a different loop on every call.
        future = asyncio.run_coroutine_threadsafe(self._store.metrics(), self._loop)
        m = future.result(timeout=10)
        if not m.measured:
            return
        values = (m.thoughts.total, m.storage.db_bytes, m.storage.wal_bytes,
                  m.search_latency.p95_ms, m.search_latency.p99_ms)
        for (name, help_text), value in zip(_METRICS, values, strict=True):
            yield GaugeMetricFamily(name, help_text, value=value)


_ready = threading.Event()
_out: dict = {}
threading.Thread(target=_run_store_loop, args=(_ready, _out), daemon=True).start()
if not _ready.wait(timeout=30):
    raise RuntimeError("timed out waiting for the store's background loop to start")
if "error" in _out:
    raise RuntimeError("failed to construct the store") from _out["error"]
REGISTRY.register(EngravaCollector(_out["store"], _out["loop"]))

This only reports meaningful search_latency when the application’s own search_hybrid() / recall() calls are also routed through this same _out["store"] instance — for example, by driving them through run_coroutine_threadsafe onto this same thread’s loop too. Each SqliteEngravaCore keeps its own in-memory, per-instance latency histogram, never shared or persisted across instances; a store built solely for scraping, with no application search traffic ever passing through it, always reports an empty search_latency histogram (sample_count stays 0) — silently, with no error to signal it.

The full field set on EngravaMetrics: thoughts (total, by_type, by_status), edges (total, by_type), storage (db_bytes, wal_bytes, vec_index_bytes, total_bytes), and search_latency (sample_count, p50_ms, p95_ms, p99_ms, min_ms, max_ms, mean_ms). The snapshot also carries schema_version, snapshot_timestamp, and measured for the snapshot itself — always check measured before trusting the rest.

Scrape cadence

Treat metrics() like any pull endpoint: a 30–60 s scrape interval is typically plenty. Counts and storage change slowly, and the latency histogram is a rolling window (metrics.window_size, default 1000 samples) that already smooths short spikes. Avoid sub-second scrapes — each call runs a few aggregate SQL queries.

What to alert on

The expired-backlog and audit-integrity signals are not in the metrics snapshot — compute them from the calls shown above on your own cadence.

engrava gc --expired needs the vector extra to delete from a database that carries a sqlite-vec index. Such a pass stops before deleting anything and exits 1 unless the environment running the CLI — frequently not the application’s — can load the extension. A pass that deletes nothing is never refused: a dry run, a sweep with nothing expired, an archiving TTL strategy. Reinstall as pip install 'engrava[vec]' and retry; see Upgrade to 0.6.

The audit-integrity check is useful once journaling has been enabled (journal.enabled: true — see Configuration → journal). Use the store-level verifier; it checks the on-disk journal even if journaling is currently disabled:

async def journal_ok(store) -> bool:
    result = await store.verify_journal()
    return result.valid

Health check

For a readiness probe you want a call that actually touches the database. Note that metrics() is not reliable for this when metrics are disabled: with metrics.enabled: false, store.metrics() returns a zero-filled snapshot without issuing any SQL, so it would report healthy even if the database were unreadable. Use a lightweight real read instead — count_thoughts() always queries the database (independent of the metrics setting):

async def healthcheck(store) -> bool:
    try:
        await store.count_thoughts()  # issues SQL — confirms DB + schema are readable
    except Exception:
        return False
    return True

(If you know metrics are enabled in your deployment, await store.metrics() works too and additionally returns the live counts.)

Logging

The library logs through the standard logging module under the engrava.* namespace (each module uses logging.getLogger(__name__), e.g. engrava.extensions.dreaming, engrava.config). As a general rule it logs at WARNING (degraded conditions, e.g. sqlite-vec unavailable → numpy fallback), INFO (dreaming progress), and DEBUG (detailed internals), and raises failures as typed exceptions for the caller to handle rather than logging them. CRITICAL is never used.

The one departure: when the derived-records extension is enabled (derive.enabled=True, off by default) and a derived child’s own failed write cannot be unwound — quarantining the connection — after its source has already committed, that failure logs at ERROR (under the default derive.on_error: "log") and then still propagates: a quarantined connection can no longer be trusted for the remaining children, so even under the default policy this is not swallowed. An ordinary per-child failure that unwinds cleanly stays at the usual WARNING level and does not raise under "log". This is the library’s only ERROR call site. Configure it like any library logger:

import logging

logging.getLogger("engrava").setLevel(logging.WARNING)  # quiet, production default
# logging.getLogger("engrava").setLevel(logging.INFO)   # see dreaming activity

Out of scope

The snapshot is deliberately small. It does not include:

• write / mutation counters or error counters — track those at your application layer (Engrava raises typed exceptions you can count there);
• dreaming metrics — run_consolidation() returns a ConsolidationResult (promoted / edges / reflections counts) per run; consume that directly;
• journal size or per-event audit metrics — the audit history lives in the journal itself, which you query and verify directly, not via the metrics snapshot.