store

store

SQLite-backed telemetry storage.

Classes

TelemetryStore

TelemetryStore(db_path: str | Path, batch_size: int = 50, flush_interval_seconds: float = 5.0)

Append-only SQLite store for inference telemetry records.

Writes are batched in memory and flushed to SQLite when a batch reaches batch_size, when flush_interval_seconds elapses (a background flusher thread guarantees this even with no further writes), on any read through this store, and on close(). Readers that open their OWN connection to the database file (e.g. TelemetryAggregator) therefore see new rows within flush_interval_seconds at the latest; call flush() first for immediate visibility. Pass flush_interval_seconds=0 to disable time-based flushing (batch-size and read/close flushes still apply).

Source code in src/openjarvis/telemetry/store.py

def __init__(
    self,
    db_path: str | Path,
    batch_size: int = 50,
    flush_interval_seconds: float = 5.0,
) -> None:
    if batch_size < 1:
        raise ValueError("batch_size must be >= 1")
    if flush_interval_seconds < 0:
        raise ValueError("flush_interval_seconds must be >= 0")

    self._db_path = str(db_path)
    if self._db_path != ":memory:":
        from openjarvis.security.file_utils import secure_create

        secure_create(Path(self._db_path))
    self._conn = sqlite3.connect(self._db_path, check_same_thread=False)
    self._lock = threading.Lock()
    self._conn.execute("PRAGMA journal_mode=WAL")
    self._conn.execute("PRAGMA synchronous=NORMAL")
    self._conn.execute("PRAGMA busy_timeout=5000")
    self._conn.execute(_CREATE_TABLE)
    self._conn.execute(_CREATE_MINING_STATS_TABLE)
    self._conn.commit()
    self._migrate_schema()

    self._batch_size = batch_size
    self._flush_interval_seconds = flush_interval_seconds
    self._last_flush_time = time.monotonic()
    self._telemetry_batch: list[tuple[Any, ...]] = []
    self._mining_batch: list[tuple[Any, ...]] = []
    self._closed = False

    # Background flusher: without it, a partial batch written just before
    # traffic stops would stay invisible to other connections until the
    # NEXT write arrived (the stale check in ``_maybe_flush_unlocked``
    # only runs inside record calls). Daemon so it never blocks exit.
    self._stop_flusher = threading.Event()
    self._flusher: threading.Thread | None = None
    if flush_interval_seconds > 0:
        self._flusher = threading.Thread(
            target=self._flush_loop,
            name="telemetry-store-flusher",
            daemon=True,
        )
        self._flusher.start()

Functions

record

record(rec: TelemetryRecord) -> None

Persist a single telemetry record.

Source code in src/openjarvis/telemetry/store.py

def record(self, rec: TelemetryRecord) -> None:
    """Persist a single telemetry record."""
    row = (
        rec.timestamp,
        rec.model_id,
        rec.engine,
        rec.agent,
        rec.prompt_tokens,
        rec.prompt_tokens_evaluated,
        rec.completion_tokens,
        rec.total_tokens,
        rec.latency_seconds,
        rec.ttft,
        rec.cost_usd,
        rec.energy_joules,
        rec.power_watts,
        rec.gpu_utilization_pct,
        rec.gpu_memory_used_gb,
        rec.gpu_temperature_c,
        rec.throughput_tok_per_sec,
        rec.prefill_latency_seconds,
        rec.decode_latency_seconds,
        rec.energy_method,
        rec.energy_vendor,
        rec.batch_id,
        1 if rec.is_warmup else 0,
        rec.cpu_energy_joules,
        rec.gpu_energy_joules,
        rec.dram_energy_joules,
        rec.tokens_per_joule,
        rec.energy_per_output_token_joules,
        rec.throughput_per_watt,
        rec.prefill_energy_joules,
        rec.decode_energy_joules,
        rec.mean_itl_ms,
        rec.median_itl_ms,
        rec.p90_itl_ms,
        rec.p95_itl_ms,
        rec.p99_itl_ms,
        rec.std_itl_ms,
        1 if rec.is_streaming else 0,
        rec.token_counting_version,
        rec.mining_session_id,
        json.dumps(rec.metadata),
    )
    with self._lock:
        self._telemetry_batch.append(row)
        self._maybe_flush_unlocked()

record_mining_stats

record_mining_stats(stats: Any) -> None

Persist one mining stats snapshot.

stats is duck-typed to keep telemetry usable without importing the optional mining package at module import time.

Source code in src/openjarvis/telemetry/store.py

def record_mining_stats(self, stats: Any) -> None:
    """Persist one mining stats snapshot.

    ``stats`` is duck-typed to keep telemetry usable without importing the
    optional mining package at module import time.
    """
    row = (
        time.time(),
        stats.provider_id,
        stats.shares_submitted,
        stats.shares_accepted,
        stats.blocks_found,
        stats.hashrate,
        stats.uptime_seconds,
        stats.last_share_at,
        stats.last_error,
        stats.payout_target,
        stats.fees_owed,
    )
    with self._lock:
        self._mining_batch.append(row)
        self._maybe_flush_unlocked()

flush

flush() -> None

Write all pending records to the database.

Source code in src/openjarvis/telemetry/store.py

def flush(self) -> None:
    """Write all pending records to the database."""
    with self._lock:
        self._flush_unlocked()

list_recent

list_recent(limit: int = 50) -> list[dict[str, Any]]

Return recent telemetry rows as dictionaries.

Source code in src/openjarvis/telemetry/store.py

def list_recent(self, limit: int = 50) -> list[dict[str, Any]]:
    """Return recent telemetry rows as dictionaries."""
    with self._lock:
        self._flush_unlocked()
        return self._select_dicts_unlocked(
            "SELECT * FROM telemetry ORDER BY timestamp DESC LIMIT ?",
            (limit,),
        )

list_recent_mining_stats

list_recent_mining_stats(limit: int = 50) -> list[dict[str, Any]]

Return recent mining stats snapshots as dictionaries.

Source code in src/openjarvis/telemetry/store.py

def list_recent_mining_stats(self, limit: int = 50) -> list[dict[str, Any]]:
    """Return recent mining stats snapshots as dictionaries."""
    with self._lock:
        self._flush_unlocked()
        return self._select_dicts_unlocked(
            "SELECT * FROM mining_stats ORDER BY recorded_at DESC LIMIT ?",
            (limit,),
        )

subscribe_to_bus

subscribe_to_bus(bus: EventBus) -> None

Subscribe to TELEMETRY_RECORD events on bus.

Source code in src/openjarvis/telemetry/store.py

def subscribe_to_bus(self, bus: EventBus) -> None:
    """Subscribe to ``TELEMETRY_RECORD`` events on *bus*."""
    bus.subscribe(EventType.TELEMETRY_RECORD, self._on_event)

close

close() -> None

Flush pending records and close the underlying SQLite connection.

Source code in src/openjarvis/telemetry/store.py

def close(self) -> None:
    """Flush pending records and close the underlying SQLite connection."""
    # Set the stop event BEFORE taking the lock: a flusher iteration
    # already waiting on the lock re-checks the event after acquiring it
    # and exits instead of touching the closed connection.
    self._stop_flusher.set()
    with self._lock:
        if self._closed:
            return
        self._flush_unlocked()
        self._conn.close()
        self._closed = True
    if self._flusher is not None:
        self._flusher.join(timeout=1.0)
        self._flusher = None