Database Service

LOG_RECORD_COLUMNS = ('seq', 'timestamp', 'level', 'logger_name', 'func_name', 'lineno', 'message', 'exc_info', 'app_key', 'instance_name', 'instance_index', 'execution_id', 'author') module-attribute

log_records table columns (excluding the autoincrement id). Public so scripts/seed_db.py can import the single source of truth instead of hand-keeping a duplicate tuple.

DatabaseService

Bases: DatabaseWriteQueueMixin, DatabaseRetentionMixin, DatabaseSizeFailsafeMixin, Service

Manages the SQLite database for operational telemetry.

Handles PRAGMA user_version migrations, heartbeat updates, the write queue, retention cleanup of old execution records, and the database size failsafe.

Source code in src/hassette/core/database_service.py

class DatabaseService(DatabaseWriteQueueMixin, DatabaseRetentionMixin, DatabaseSizeFailsafeMixin, Service):
    """Manages the SQLite database for operational telemetry.

    Handles PRAGMA user_version migrations, heartbeat updates, the write queue, retention
    cleanup of old execution records, and the database size failsafe.
    """

    required: ClassVar[bool] = True

    shutdown_hook_timeout_applies: ClassVar[bool] = False
    """A foundation service: the root's foundation stage bounds its shutdown, and its drain has a fixed cap."""

    shutdown_snapshot_task: "asyncio.Task[bool] | None"
    """The shutdown snapshot's copy, once ``write_shutdown_snapshot()`` started it; the hard exit waits for it."""

    _db: aiosqlite.Connection | None
    """The aiosqlite write connection, set during on_initialize."""

    _read_db: aiosqlite.Connection | None
    """Dedicated read-only connection for TelemetryQueryService. Opened on a separate
    WAL snapshot so reads never block the write worker."""

    CONN_ATTRS = ("_read_db", "_db")
    """Connection attributes, in the order both teardown paths close them.

    Read first so the write connection closes last and performs the WAL checkpoint. Named once so
    a rename cannot leave one of the two reflection-based loops behind.
    """

    _db_path: Path
    """Resolved path to the SQLite database file."""

    _consecutive_heartbeat_failures: int
    """Counter for consecutive heartbeat failures; triggers RuntimeError after threshold."""

    _progress_at_last_heartbeat: int | None
    """``_write_progress_count`` when the previous heartbeat attempt finished, or None before the first
    attempt of this lifecycle. A failed attempt only counts a strike when the count hasn't moved since."""

    _db_write_queue: asyncio.Queue[_WriteQueueItem] | None
    """Bounded queue of pending write coroutines; each paired with an optional Future for result delivery."""

    _db_worker_task: asyncio.Task[None] | None
    """Background task that drains _db_write_queue sequentially."""

    _write_queue_detached: bool
    """Whether a teardown path has taken ``_db_write_queue`` away — see ``detach_write_queue()``."""

    _executing_future: asyncio.Future[Any] | None
    """Future of the ``submit()`` item the write worker is currently running, if any. Lets
    ``submit()`` tell a write still waiting in the queue (safe to withdraw on timeout) from one
    already executing (whose outcome must be awaited, since it may commit)."""

    _write_progress_count: int
    """Monotonic count of completed write work: write-queue jobs that returned without raising, plus
    each retention or size-failsafe batch that committed. Never reset, so a reader compares two
    readings to tell a busy write worker from a stuck one."""

    _consecutive_size_triggers: int
    """Counter for consecutive hourly size failsafe triggers; logged as a warning."""

    _consecutive_exhaustion_triggers: int
    """Counter for consecutive size failsafe runs that drained every retention tier and left
    the database still over the configured size limit; logged as a warning."""

    _consecutive_parent_guard_skips: int
    """Counter for consecutive retention cycles that skipped parent-guard deletes because a
    retention target failed or was incomplete; escalated to an error past a threshold."""

    def __init__(self, hassette: "Hassette", *, parent: "Resource | None" = None) -> None:
        super().__init__(hassette, parent=parent)
        self._db = None
        self._read_db = None
        self._db_path = Path()
        self._consecutive_heartbeat_failures = 0
        self._progress_at_last_heartbeat = None
        self._consecutive_size_triggers = 0
        self._consecutive_exhaustion_triggers = 0
        self._consecutive_parent_guard_skips = 0
        self._db_write_queue = None
        self._db_worker_task = None
        self._write_queue_detached = False
        self._executing_future = None
        self._waiting_submits = 0
        self._write_progress_count = 0
        self.shutdown_snapshot_task = None

    @property
    def config_log_level(self) -> LogLevel:
        """Log level for this service, read from ``logging.database_service`` in the config."""
        return self.hassette.config.logging.database_service

    @property
    def is_db_ready(self) -> bool:
        """Whether the write database connection is open and usable."""
        return self._db is not None

    @property
    def is_accepting_writes(self) -> bool:
        """Whether the write queue is live and accepting submissions.

        False before ``on_initialize()`` creates the queue and after shutdown drains it.
        """
        return self._db_write_queue is not None

    @property
    def db(self) -> aiosqlite.Connection:
        """Return the active write database connection.

        Raises:
            RuntimeError: If the database connection is not initialized.
        """
        if self._db is None:
            raise RuntimeError("Database connection is not initialized")
        return self._db

    @property
    def read_db(self) -> aiosqlite.Connection:
        """Return the dedicated read-only database connection.

        Uses a separate WAL snapshot so reads never block the write worker.

        Raises:
            RuntimeError: If the read connection is not initialized.
        """
        if self._read_db is None:
            raise RuntimeError("Read database connection is not initialized")
        return self._read_db

    async def on_initialize(self) -> None:
        """Set up the database: check schema version, run migrations and open connection."""
        self._consecutive_heartbeat_failures = 0
        self._progress_at_last_heartbeat = None
        self._consecutive_size_triggers = 0
        self._consecutive_exhaustion_triggers = 0
        self._consecutive_parent_guard_skips = 0
        # Cleared up front, not at queue-creation time below: a re-initialization that
        # fails before it gets that far must still report the pre-init cause rather than a
        # stale post-teardown one left over from the previous lifecycle.
        self._write_queue_detached = False
        self._db_path = self.resolve_db_path()
        self._db_path.parent.mkdir(parents=True, exist_ok=True)

        await self.handle_schema_version(self._db_path)

        self.logger.debug("Running database migrations for %s", self._db_path)
        timeout = self.hassette.config.database.migration_timeout_seconds
        try:
            await asyncio.wait_for(asyncio.to_thread(self.run_migrations), timeout=timeout)
        except TimeoutError as exc:
            raise TimeoutError(
                f"Database migrations did not finish within {timeout}s (database.migration_timeout_seconds)"
            ) from exc

        self._db = await connect_daemon(self._db_path, isolation_level=None)
        self._db.row_factory = aiosqlite.Row

        # Open a dedicated read connection on a separate WAL snapshot (F1).
        # This ensures read queries never block the write worker.
        self._read_db = await connect_daemon(self._db_path, isolation_level=None)
        self._read_db.row_factory = aiosqlite.Row
        await self._read_db.execute("PRAGMA query_only = ON")
        await self._read_db.execute(f"PRAGMA busy_timeout = {SQLITE_BUSY_TIMEOUT_MS}")

        await self.set_pragmas()
        await self.catch_up_cleanup_at_startup()

        self._db_write_queue = asyncio.Queue(maxsize=self.hassette.config.database.write_queue_max)
        # Bypass the loop's global task factory so the worker is not tracked by any TaskBucket. A bare
        # asyncio.create_task() falls through to the root Hassette bucket via make_task_factory(); the root
        # bucket's cancel_all() runs before wave-based child shutdown begins, killing the worker before
        # on_shutdown() can drain the queue — producing two 10s stalls per test (one in each downstream wave
        # that tries to submit() a write to the dead worker). The worker's lifecycle is managed by
        # on_shutdown() (drain-and-close) and _force_terminal() (hard cancel when a deadline owner walks away).
        self._db_worker_task = create_lifecycle_task(
            run_write_queue_worker(self), name=f"db_write_worker:{self.unique_name}"
        )
        self._db_worker_task.add_done_callback(lambda task: log_worker_exit(self, task))

    def _force_terminal(self) -> None:
        """Also cancel the untracked database write worker, drain its queue, and close connections.

        ``_db_worker_task`` bypasses TaskBucket entirely (see ``on_initialize()``), so
        ``Service._force_terminal()``'s ``TaskBucket.cancel_all_sync()`` never reaches it.
        Without cancelling it here, a force-terminal call from a deadline owner walking away (which skips
        ``on_shutdown()``, the only other place this task is cancelled) leaves the worker
        and its queue/connections running after the process has declared shutdown complete.

        Detaches ``_db_write_queue`` via ``detach_write_queue()`` (mirroring ``on_shutdown()``'s
        drain-and-close pattern, minus the graceful ``queue.join()`` this synchronous path can't
        await) before closing remaining items via ``close_remaining_queue_items()``. Without this,
        two things go wrong: any coroutine still queued when the worker is cancelled is never
        closed (GC eventually raises "coroutine was never awaited"), and ``submit()``/``enqueue()``
        only reject once ``_db_write_queue`` is ``None`` -- leaving it set would let a caller
        enqueue into a queue with a cancelled worker, stalling every ``submit()`` caller for
        its full queue timeout and silently dropping every ``enqueue()``.

        Closing ``_db``/``_read_db`` here (via the same synchronous ``stop_connection_sync()``
        used by ``App._force_terminal()`` for its cache) matters for the same reason: this path
        intentionally skips ``on_shutdown()``/``cleanup()`` (see ``Resource._force_terminal()``'s
        docstring) because production assumes force-terminal is nearly always followed by process
        exit. In a long-lived process -- notably the test suite -- a connection left open here
        would otherwise sit until the garbage collector reclaims it, firing an unraisable-exception
        warning attributed to whichever test happens to be running at that moment.
        """
        if self._db_worker_task is not None and not self._db_worker_task.done():
            self._db_worker_task.cancel()
        self.close_remaining_queue_items(detach_write_queue(self))
        for attr in self.CONN_ATTRS:
            stop_connection_sync(getattr(self, attr))
            setattr(self, attr, None)
        super()._force_terminal()

    async def serve(self) -> None:
        """Run the heartbeat, retention, and size failsafe loop until shutdown."""
        mark_ready(self, reason="Database service started")

        # Each task remembers its last attempt, and every pass derives the deadlines from it and the current
        # config. Every wake goes back to the top, so a config reload in either direction and a shutdown
        # request are all handled by that one derivation. Sleeping until the earliest deadline lets a
        # maintenance interval shorter than the heartbeat interval run on its own cadence.
        start = time.monotonic()
        last_heartbeat = start
        # Seeded as a successful attempt at startup, so each cleanup's first run waits a full interval.
        retention_attempt = CleanupAttempt(attempted_at=start, enqueued=True)
        size_failsafe_attempt = CleanupAttempt(attempted_at=start, enqueued=True)

        while True:
            if self.shutdown_event.is_set():
                mark_not_ready(self, reason="Shutting down")
                return

            config = self.hassette.config.database
            next_heartbeat, next_retention, next_size_failsafe = maintenance_deadlines(
                config, last_heartbeat, retention_attempt, size_failsafe_attempt
            )
            earliest = min(next_heartbeat, next_retention, next_size_failsafe)
            now = time.monotonic()
            if now < earliest:
                with contextlib.suppress(TimeoutError):
                    await asyncio.wait_for(self.shutdown_event.wait(), timeout=earliest - now)
                continue

            if time.monotonic() >= next_heartbeat:
                await self.update_heartbeat()
                last_heartbeat = time.monotonic()

                if self._consecutive_heartbeat_failures >= config.max_consecutive_heartbeat_failures:
                    raise RuntimeError(f"Heartbeat failed {self._consecutive_heartbeat_failures} consecutive times")

            if time.monotonic() >= next_retention:
                enqueued = await self.run_retention_cleanup()
                retention_attempt = CleanupAttempt(attempted_at=time.monotonic(), enqueued=enqueued)

            if time.monotonic() >= next_size_failsafe:
                enqueued = await self.run_size_failsafe()
                size_failsafe_attempt = CleanupAttempt(attempted_at=time.monotonic(), enqueued=enqueued)

    async def on_shutdown(self) -> None:
        """Drain the write queue, cancel the worker, and close the database connections.

        The root writes the shutdown snapshot after the foundation stage (``write_shutdown_snapshot()``).
        """
        queue: asyncio.Queue[_WriteQueueItem] | None = None
        try:
            if self._db_worker_task is not None:
                queue = detach_write_queue(self)
                if queue is not None:
                    await self.drain_write_queue(
                        queue, clamp_to_shutdown_deadline(self, SHUTDOWN_DRAIN_TIMEOUT_SECONDS)
                    )
                await cancel_and_drain(self._db_worker_task)
                self._db_worker_task = None
        except STORAGE_ERRORS:
            self.logger.exception("Error draining write queue during shutdown")
        finally:
            if self._db_worker_task is not None:
                await cancel_and_drain(self._db_worker_task)
                self._db_worker_task = None
            self.close_remaining_queue_items(queue)
            await self.close_connections()

    async def write_shutdown_snapshot(self) -> None:
        """Write ``database.shutdown_snapshot_path``, last in shutdown, after the foundation stage.

        The root calls it even when the drain, the close or the foundation stage itself failed, so the snapshot
        then holds every write that committed. It opens its own connection to the database file, so the closed
        service connections don't affect it. A failure is logged, never raised: shutdown carries on, and the
        previous snapshot stays in place.

        Nothing bounds the copy. It runs in a worker thread inside a shielded task kept on
        ``shutdown_snapshot_task``, so if this await is cancelled the copy still finishes and logs its outcome,
        and the hard exit waits for it. Only a kill (the supervisor's stop grace period) cuts it short, which
        leaves the previous snapshot in place.
        """
        snapshot_path = self.hassette.config.database.shutdown_snapshot_path
        if snapshot_path is None:
            return
        # A lifecycle task, not a bare shield(coro): the TaskBucket factory is sealed this late in shutdown.
        copy = create_lifecycle_task(
            asyncio.to_thread(write_snapshot, self._db_path, snapshot_path),
            name=f"shutdown_snapshot:{self.unique_name}",
        )
        copy.add_done_callback(partial(log_shutdown_snapshot_outcome, self, snapshot_path))
        self.shutdown_snapshot_task = copy
        # failures are logged by the callback, which also runs if this await is cancelled
        with contextlib.suppress(sqlite3.Error, OSError):
            await asyncio.shield(copy)

    async def close_connections(self) -> None:
        """Close both database connections. Idempotent -- safe to call multiple times.

        aiosqlite's worker threads are set to daemon in on_initialize() as a safety net, but this
        method still does a best-effort close to avoid resource warnings and ensure clean WAL
        checkpoints.

        Propagates the first close failure rather than swallowing it: both call sites feed teardown
        accounting that has to tell a clean close from one that left a connection in an unknown
        state -- an escaping exception makes the shutdown outcome ``FAILED``, from ``cleanup()`` and
        ``on_shutdown()`` alike. See
        ``close_connection_pair()`` for the close and cancellation mechanics.
        """
        await close_connection_pair(self, self.CONN_ATTRS, self.logger)

    async def cleanup(self) -> None:
        """Close database connections if on_shutdown was interrupted or never ran."""
        await self.close_connections()
        await super().cleanup()

    def resolve_db_path(self) -> Path:
        """Resolve the database path from config or use default."""
        if self.hassette.config.database.path is not None:
            return self.hassette.config.database.path.resolve()
        return self.hassette.config.data_dir / "hassette.db"

    def get_expected_head_version(self) -> int:
        """Return the highest migration version number from migrations_sql/ (synchronous).

        Scans the migrations_sql/ directory for *.sql files with numeric stems
        and returns the largest version number found.
        """
        sql_files = _collect_migrations(None)
        if not sql_files:
            raise RuntimeError("No migration files found in migrations_sql/")
        return max(sql_files)

    def get_current_db_version(self, db_path: Path) -> int:
        """Return PRAGMA user_version from the on-disk DB (synchronous). Returns 0 for fresh databases."""
        return _read_user_version(db_path)

    async def handle_schema_version(self, db_path: Path) -> None:
        """Check schema version and handle mismatches.

        If the DB file does not exist yet, does nothing (migrations will create it).
        If the DB version matches the expected head, does nothing.
        If the DB version is behind head (0 < current < head), does nothing —
        run_migrations() will apply pending migrations incrementally, preserving data.
        If the DB version is 0 on an existing file (pre-PRAGMA-era or fresh DB with
        no migrations applied), deletes the file so migrations recreate it cleanly.
        If the DB version is *ahead* of head (newer DB on older binary), logs an ERROR
        and raises SchemaVersionError — auto-delete is refused in this case.

        Args:
            db_path: Path to the SQLite database file.

        Raises:
            SchemaVersionError: When the DB version is ahead of the expected head version.
            RuntimeError: When the DB file cannot be deleted due to permissions.
        """
        if not await asyncio.to_thread(db_path.exists):
            return

        expected_head = await asyncio.to_thread(self.get_expected_head_version)
        current_version = await asyncio.to_thread(self.get_current_db_version, db_path)

        if current_version == expected_head:
            return

        if current_version > expected_head:
            self.logger.error(
                "Database schema version %d is ahead of the code's expected head %d. "
                "This usually means a newer binary created this database. "
                "Refusing to auto-delete — upgrade the binary or remove the database manually.",
                current_version,
                expected_head,
            )
            raise SchemaVersionError(
                f"Database schema version {current_version} is ahead of expected head "
                f"{expected_head}. Cannot start safely."
            )

        if current_version > 0:
            self.logger.info(
                "Database schema version %d is behind head %d — pending migrations will be applied.",
                current_version,
                expected_head,
            )
            return

        if current_version == 0:
            self.logger.warning(
                "Database has no schema version (expected %d) — recreating database (no production data to preserve).",
                expected_head,
            )
            try:
                await asyncio.to_thread(delete_db_files, db_path)
            except PermissionError as exc:
                raise RuntimeError(
                    f"Cannot delete stale database file {db_path}: {exc}. Please remove it manually and restart."
                ) from exc

    def run_migrations(self) -> None:
        """Run PRAGMA user_version migrations to the latest version (synchronous, called via to_thread).

        auto_vacuum = INCREMENTAL is set by the runner before any tables are created.
        """
        run_migrations(self._db_path)

    async def catch_up_cleanup_at_startup(self) -> None:
        """Run the size failsafe, then retention cleanup, once during on_initialize().

        Runs directly on the write connection, before the write worker starts.
        """
        # Both passes commit per batch, so a re-run after a restart converges without saved
        # progress. They share one budget so a large backlog can't exhaust the readiness timeout
        # wait_for_ready() enforces around on_initialize(). The rollback after an interrupt is
        # unbudgeted, but it waits on at most one batch-bounded statement.
        budget = self.hassette.config.lifecycle.startup_timeout_seconds * _STARTUP_CLEANUP_BUDGET_FRACTION
        try:
            async with asyncio.timeout(budget):
                await self._check_size_failsafe()
                await self._do_run_retention_cleanup()
        except TimeoutError:
            # Cancelling the await doesn't stop a statement already running on aiosqlite's worker
            # thread; interrupt it so the rollback isn't queued behind it.
            try:
                await self.db.interrupt()
            except STORAGE_ERRORS:
                self.logger.exception("Interrupting the startup cleanup catch-up statement failed")
            await safe_rollback(self.db, self, "startup cleanup catch-up")
            self.logger.warning(
                "Startup cleanup catch-up exceeded its %.1fs budget; the hourly cleanup loop will pick up the rest",
                budget,
            )
        except STORAGE_ERRORS:
            # A failure can leave a retention batch between its BEGIN and commit; roll back so the
            # write worker doesn't inherit an open transaction. Earlier batches stay committed.
            await safe_rollback(self.db, self, "startup cleanup catch-up")
            self.logger.exception("Startup cleanup catch-up failed; continuing without cleanup")

    async def set_pragmas(self) -> None:
        """Configure SQLite PRAGMAs for performance and safety."""
        db = self.db
        await db.execute("PRAGMA journal_mode = WAL")
        await db.execute("PRAGMA wal_autocheckpoint = 1000")
        # NORMAL is an intentional performance tradeoff: in WAL mode, the last committed
        # writes before an OS crash (not app crash) may be lost if not yet checkpointed.
        # This is acceptable for operational telemetry — the orphan-session mechanism
        # compensates for session rows but not for individual telemetry records.
        await db.execute("PRAGMA synchronous = NORMAL")
        await db.execute(f"PRAGMA busy_timeout = {SQLITE_BUSY_TIMEOUT_MS}")
        await db.execute("PRAGMA foreign_keys = ON")
        # Intentionally a no-op — auto_vacuum is set by the migration runner before table creation.
        # This line documents intent only.
        await db.execute("PRAGMA auto_vacuum = INCREMENTAL")

    async def update_heartbeat(self) -> None:
        """Await a heartbeat update for the current session.

        Early-return guards run inline; the DB write is awaited via submit()
        so that _consecutive_heartbeat_failures is updated before returning.

        The submit() call is bounded by _HEARTBEAT_WRITE_TIMEOUT_SECONDS. submit()'s own queue
        timeout only covers the wait before the write starts; a worker wedged *inside* a write
        never raises, so without this outer bound serve() would park on submit() forever and
        never reach its failure-count escalation. Either timeout surfaces as the same
        TimeoutError: the queue was backed up past its limit, or the worker is wedged.
        A timeout, or a WriteQueueUnavailableError from submit(), is a failed attempt identically to a
        raised sqlite3.Error/OSError/ValueError. A failed attempt counts a strike only when
        _write_progress_count has not moved since the previous attempt finished (or, on the first
        attempt, since this one started): a worker still committing writes, such as a long batched
        retention pass holding the queue, is busy rather than stuck and must not be failed.

        This method is the only place _consecutive_heartbeat_failures moves, so one attempt
        costs at most one strike. Counting the timeout here and the raise inside the queued
        coroutine would let a single slow-then-failing write consume two of the three,
        failing the service a full heartbeat interval early.
        """
        if self._db is None:
            return
        if self._db_write_queue is None:
            return
        try:
            _ = self.hassette.session_id
        except RuntimeError:
            return
        progress_baseline = self._progress_at_last_heartbeat
        if progress_baseline is None:
            progress_baseline = self._write_progress_count
        try:
            async with asyncio.timeout(_HEARTBEAT_WRITE_TIMEOUT_SECONDS):
                await self.submit(self._do_update_heartbeat())
        except TimeoutError as exc:
            self.record_heartbeat_failure(
                "Heartbeat write timed out — write queue backed up or worker wedged", exc, progress_baseline
            )
        except (sqlite3.Error, OSError, ValueError, WriteQueueUnavailableError) as exc:
            self.record_heartbeat_failure("Failed to update heartbeat", exc, progress_baseline)
        else:
            if self._consecutive_heartbeat_failures > 0:
                self.logger.info("Heartbeat recovered after %d failure(s)", self._consecutive_heartbeat_failures)
                self._consecutive_heartbeat_failures = 0
        finally:
            self._progress_at_last_heartbeat = self._write_progress_count

    def record_heartbeat_failure(self, message: str, exc: BaseException, progress_baseline: int) -> None:
        """Count a strike for a failed heartbeat attempt unless writes progressed past ``progress_baseline``.

        ``exc`` is the attempt's failure; its traceback is attached to the logged entry either way.
        """
        max_failures = self.hassette.config.database.max_consecutive_heartbeat_failures
        progress = self._write_progress_count - progress_baseline
        if progress > 0:
            self.logger.warning(
                "%s, but %d write(s) completed since the previous attempt; not counted (failures %d/%d)",
                message,
                progress,
                self._consecutive_heartbeat_failures,
                max_failures,
                exc_info=exc,
            )
            return
        self._consecutive_heartbeat_failures += 1
        self.logger.error(
            "%s with no write progress since the previous attempt (failure %d/%d)",
            message,
            self._consecutive_heartbeat_failures,
            max_failures,
            exc_info=exc,
        )

    async def _do_update_heartbeat(self) -> None:
        """Execute the heartbeat DB write; called by the write-queue worker.

        Failures propagate to update_heartbeat() through submit()'s future rather than
        being counted here — see that method for why the count lives in one place.
        """
        session_id = self.hassette.session_id
        now = time.time()
        await self.db.execute(
            "UPDATE sessions SET last_heartbeat_at = ? WHERE id = ?",
            (now, session_id),
        )
        await self.db.commit()
        self.logger.debug("Heartbeat updated for session %d", session_id)

    async def _insert_log_records(self, records: list[dict]) -> None:
        """Batch-insert log records into the log_records table.

        Must only be called via enqueue() — never await directly.
        """
        if not records:
            return
        db = self.db
        try:
            await db.execute(SQL_BEGIN)
            await db.executemany(_LOG_INSERT_SQL, records)
            await db.commit()
        except Exception:  # broad on purpose: rolls back, then re-raises whatever failed
            await safe_rollback(db, self, "log record insert")
            raise

shutdown_hook_timeout_applies: bool = False class-attribute

A foundation service: the root's foundation stage bounds its shutdown, and its drain has a fixed cap.

CONN_ATTRS = ('_read_db', '_db') class-attribute instance-attribute

Connection attributes, in the order both teardown paths close them.

Read first so the write connection closes last and performs the WAL checkpoint. Named once so a rename cannot leave one of the two reflection-based loops behind.

shutdown_snapshot_task: asyncio.Task[bool] | None = None instance-attribute

The shutdown snapshot's copy, once write_shutdown_snapshot() started it; the hard exit waits for it.

config_log_level: LogLevel property

Log level for this service, read from logging.database_service in the config.

is_db_ready: bool property

Whether the write database connection is open and usable.

is_accepting_writes: bool property

Whether the write queue is live and accepting submissions.

False before on_initialize() creates the queue and after shutdown drains it.

db: aiosqlite.Connection property

Return the active write database connection.

Raises:

Type	Description
RuntimeError	If the database connection is not initialized.

read_db: aiosqlite.Connection property

Return the dedicated read-only database connection.

Uses a separate WAL snapshot so reads never block the write worker.

Raises:

Type	Description
RuntimeError	If the read connection is not initialized.

on_initialize() -> None async

Set up the database: check schema version, run migrations and open connection.

Source code in src/hassette/core/database_service.py

async def on_initialize(self) -> None:
    """Set up the database: check schema version, run migrations and open connection."""
    self._consecutive_heartbeat_failures = 0
    self._progress_at_last_heartbeat = None
    self._consecutive_size_triggers = 0
    self._consecutive_exhaustion_triggers = 0
    self._consecutive_parent_guard_skips = 0
    # Cleared up front, not at queue-creation time below: a re-initialization that
    # fails before it gets that far must still report the pre-init cause rather than a
    # stale post-teardown one left over from the previous lifecycle.
    self._write_queue_detached = False
    self._db_path = self.resolve_db_path()
    self._db_path.parent.mkdir(parents=True, exist_ok=True)

    await self.handle_schema_version(self._db_path)

    self.logger.debug("Running database migrations for %s", self._db_path)
    timeout = self.hassette.config.database.migration_timeout_seconds
    try:
        await asyncio.wait_for(asyncio.to_thread(self.run_migrations), timeout=timeout)
    except TimeoutError as exc:
        raise TimeoutError(
            f"Database migrations did not finish within {timeout}s (database.migration_timeout_seconds)"
        ) from exc

    self._db = await connect_daemon(self._db_path, isolation_level=None)
    self._db.row_factory = aiosqlite.Row

    # Open a dedicated read connection on a separate WAL snapshot (F1).
    # This ensures read queries never block the write worker.
    self._read_db = await connect_daemon(self._db_path, isolation_level=None)
    self._read_db.row_factory = aiosqlite.Row
    await self._read_db.execute("PRAGMA query_only = ON")
    await self._read_db.execute(f"PRAGMA busy_timeout = {SQLITE_BUSY_TIMEOUT_MS}")

    await self.set_pragmas()
    await self.catch_up_cleanup_at_startup()

    self._db_write_queue = asyncio.Queue(maxsize=self.hassette.config.database.write_queue_max)
    # Bypass the loop's global task factory so the worker is not tracked by any TaskBucket. A bare
    # asyncio.create_task() falls through to the root Hassette bucket via make_task_factory(); the root
    # bucket's cancel_all() runs before wave-based child shutdown begins, killing the worker before
    # on_shutdown() can drain the queue — producing two 10s stalls per test (one in each downstream wave
    # that tries to submit() a write to the dead worker). The worker's lifecycle is managed by
    # on_shutdown() (drain-and-close) and _force_terminal() (hard cancel when a deadline owner walks away).
    self._db_worker_task = create_lifecycle_task(
        run_write_queue_worker(self), name=f"db_write_worker:{self.unique_name}"
    )
    self._db_worker_task.add_done_callback(lambda task: log_worker_exit(self, task))

serve() -> None async

Run the heartbeat, retention, and size failsafe loop until shutdown.

Source code in src/hassette/core/database_service.py

async def serve(self) -> None:
    """Run the heartbeat, retention, and size failsafe loop until shutdown."""
    mark_ready(self, reason="Database service started")

    # Each task remembers its last attempt, and every pass derives the deadlines from it and the current
    # config. Every wake goes back to the top, so a config reload in either direction and a shutdown
    # request are all handled by that one derivation. Sleeping until the earliest deadline lets a
    # maintenance interval shorter than the heartbeat interval run on its own cadence.
    start = time.monotonic()
    last_heartbeat = start
    # Seeded as a successful attempt at startup, so each cleanup's first run waits a full interval.
    retention_attempt = CleanupAttempt(attempted_at=start, enqueued=True)
    size_failsafe_attempt = CleanupAttempt(attempted_at=start, enqueued=True)

    while True:
        if self.shutdown_event.is_set():
            mark_not_ready(self, reason="Shutting down")
            return

        config = self.hassette.config.database
        next_heartbeat, next_retention, next_size_failsafe = maintenance_deadlines(
            config, last_heartbeat, retention_attempt, size_failsafe_attempt
        )
        earliest = min(next_heartbeat, next_retention, next_size_failsafe)
        now = time.monotonic()
        if now < earliest:
            with contextlib.suppress(TimeoutError):
                await asyncio.wait_for(self.shutdown_event.wait(), timeout=earliest - now)
            continue

        if time.monotonic() >= next_heartbeat:
            await self.update_heartbeat()
            last_heartbeat = time.monotonic()

            if self._consecutive_heartbeat_failures >= config.max_consecutive_heartbeat_failures:
                raise RuntimeError(f"Heartbeat failed {self._consecutive_heartbeat_failures} consecutive times")

        if time.monotonic() >= next_retention:
            enqueued = await self.run_retention_cleanup()
            retention_attempt = CleanupAttempt(attempted_at=time.monotonic(), enqueued=enqueued)

        if time.monotonic() >= next_size_failsafe:
            enqueued = await self.run_size_failsafe()
            size_failsafe_attempt = CleanupAttempt(attempted_at=time.monotonic(), enqueued=enqueued)

on_shutdown() -> None async

Drain the write queue, cancel the worker, and close the database connections.

The root writes the shutdown snapshot after the foundation stage (write_shutdown_snapshot()).

Source code in src/hassette/core/database_service.py

async def on_shutdown(self) -> None:
    """Drain the write queue, cancel the worker, and close the database connections.

    The root writes the shutdown snapshot after the foundation stage (``write_shutdown_snapshot()``).
    """
    queue: asyncio.Queue[_WriteQueueItem] | None = None
    try:
        if self._db_worker_task is not None:
            queue = detach_write_queue(self)
            if queue is not None:
                await self.drain_write_queue(
                    queue, clamp_to_shutdown_deadline(self, SHUTDOWN_DRAIN_TIMEOUT_SECONDS)
                )
            await cancel_and_drain(self._db_worker_task)
            self._db_worker_task = None
    except STORAGE_ERRORS:
        self.logger.exception("Error draining write queue during shutdown")
    finally:
        if self._db_worker_task is not None:
            await cancel_and_drain(self._db_worker_task)
            self._db_worker_task = None
        self.close_remaining_queue_items(queue)
        await self.close_connections()

write_shutdown_snapshot() -> None async

Write database.shutdown_snapshot_path, last in shutdown, after the foundation stage.

The root calls it even when the drain, the close or the foundation stage itself failed, so the snapshot then holds every write that committed. It opens its own connection to the database file, so the closed service connections don't affect it. A failure is logged, never raised: shutdown carries on, and the previous snapshot stays in place.

Nothing bounds the copy. It runs in a worker thread inside a shielded task kept on shutdown_snapshot_task, so if this await is cancelled the copy still finishes and logs its outcome, and the hard exit waits for it. Only a kill (the supervisor's stop grace period) cuts it short, which leaves the previous snapshot in place.

Source code in src/hassette/core/database_service.py

async def write_shutdown_snapshot(self) -> None:
    """Write ``database.shutdown_snapshot_path``, last in shutdown, after the foundation stage.

    The root calls it even when the drain, the close or the foundation stage itself failed, so the snapshot
    then holds every write that committed. It opens its own connection to the database file, so the closed
    service connections don't affect it. A failure is logged, never raised: shutdown carries on, and the
    previous snapshot stays in place.

    Nothing bounds the copy. It runs in a worker thread inside a shielded task kept on
    ``shutdown_snapshot_task``, so if this await is cancelled the copy still finishes and logs its outcome,
    and the hard exit waits for it. Only a kill (the supervisor's stop grace period) cuts it short, which
    leaves the previous snapshot in place.
    """
    snapshot_path = self.hassette.config.database.shutdown_snapshot_path
    if snapshot_path is None:
        return
    # A lifecycle task, not a bare shield(coro): the TaskBucket factory is sealed this late in shutdown.
    copy = create_lifecycle_task(
        asyncio.to_thread(write_snapshot, self._db_path, snapshot_path),
        name=f"shutdown_snapshot:{self.unique_name}",
    )
    copy.add_done_callback(partial(log_shutdown_snapshot_outcome, self, snapshot_path))
    self.shutdown_snapshot_task = copy
    # failures are logged by the callback, which also runs if this await is cancelled
    with contextlib.suppress(sqlite3.Error, OSError):
        await asyncio.shield(copy)

close_connections() -> None async

Close both database connections. Idempotent -- safe to call multiple times.

aiosqlite's worker threads are set to daemon in on_initialize() as a safety net, but this method still does a best-effort close to avoid resource warnings and ensure clean WAL checkpoints.

Propagates the first close failure rather than swallowing it: both call sites feed teardown accounting that has to tell a clean close from one that left a connection in an unknown state -- an escaping exception makes the shutdown outcome FAILED, from cleanup() and on_shutdown() alike. See close_connection_pair() for the close and cancellation mechanics.

Source code in src/hassette/core/database_service.py

async def close_connections(self) -> None:
    """Close both database connections. Idempotent -- safe to call multiple times.

    aiosqlite's worker threads are set to daemon in on_initialize() as a safety net, but this
    method still does a best-effort close to avoid resource warnings and ensure clean WAL
    checkpoints.

    Propagates the first close failure rather than swallowing it: both call sites feed teardown
    accounting that has to tell a clean close from one that left a connection in an unknown
    state -- an escaping exception makes the shutdown outcome ``FAILED``, from ``cleanup()`` and
    ``on_shutdown()`` alike. See
    ``close_connection_pair()`` for the close and cancellation mechanics.
    """
    await close_connection_pair(self, self.CONN_ATTRS, self.logger)

cleanup() -> None async

Close database connections if on_shutdown was interrupted or never ran.

Source code in src/hassette/core/database_service.py

async def cleanup(self) -> None:
    """Close database connections if on_shutdown was interrupted or never ran."""
    await self.close_connections()
    await super().cleanup()

resolve_db_path() -> Path

Resolve the database path from config or use default.

Source code in src/hassette/core/database_service.py

def resolve_db_path(self) -> Path:
    """Resolve the database path from config or use default."""
    if self.hassette.config.database.path is not None:
        return self.hassette.config.database.path.resolve()
    return self.hassette.config.data_dir / "hassette.db"

get_expected_head_version() -> int

Return the highest migration version number from migrations_sql/ (synchronous).

Scans the migrations_sql/ directory for *.sql files with numeric stems and returns the largest version number found.

Source code in src/hassette/core/database_service.py

def get_expected_head_version(self) -> int:
    """Return the highest migration version number from migrations_sql/ (synchronous).

    Scans the migrations_sql/ directory for *.sql files with numeric stems
    and returns the largest version number found.
    """
    sql_files = _collect_migrations(None)
    if not sql_files:
        raise RuntimeError("No migration files found in migrations_sql/")
    return max(sql_files)

get_current_db_version(db_path: Path) -> int

Return PRAGMA user_version from the on-disk DB (synchronous). Returns 0 for fresh databases.

Source code in src/hassette/core/database_service.py

def get_current_db_version(self, db_path: Path) -> int:
    """Return PRAGMA user_version from the on-disk DB (synchronous). Returns 0 for fresh databases."""
    return _read_user_version(db_path)

handle_schema_version(db_path: Path) -> None async

Check schema version and handle mismatches.

If the DB file does not exist yet, does nothing (migrations will create it). If the DB version matches the expected head, does nothing. If the DB version is behind head (0 < current < head), does nothing — run_migrations() will apply pending migrations incrementally, preserving data. If the DB version is 0 on an existing file (pre-PRAGMA-era or fresh DB with no migrations applied), deletes the file so migrations recreate it cleanly. If the DB version is ahead of head (newer DB on older binary), logs an ERROR and raises SchemaVersionError — auto-delete is refused in this case.

Parameters:

Name	Type	Description	Default
db_path	Path	Path to the SQLite database file.	required

Raises:

Type	Description
SchemaVersionError	When the DB version is ahead of the expected head version.
RuntimeError	When the DB file cannot be deleted due to permissions.

Source code in src/hassette/core/database_service.py

async def handle_schema_version(self, db_path: Path) -> None:
    """Check schema version and handle mismatches.

    If the DB file does not exist yet, does nothing (migrations will create it).
    If the DB version matches the expected head, does nothing.
    If the DB version is behind head (0 < current < head), does nothing —
    run_migrations() will apply pending migrations incrementally, preserving data.
    If the DB version is 0 on an existing file (pre-PRAGMA-era or fresh DB with
    no migrations applied), deletes the file so migrations recreate it cleanly.
    If the DB version is *ahead* of head (newer DB on older binary), logs an ERROR
    and raises SchemaVersionError — auto-delete is refused in this case.

    Args:
        db_path: Path to the SQLite database file.

    Raises:
        SchemaVersionError: When the DB version is ahead of the expected head version.
        RuntimeError: When the DB file cannot be deleted due to permissions.
    """
    if not await asyncio.to_thread(db_path.exists):
        return

    expected_head = await asyncio.to_thread(self.get_expected_head_version)
    current_version = await asyncio.to_thread(self.get_current_db_version, db_path)

    if current_version == expected_head:
        return

    if current_version > expected_head:
        self.logger.error(
            "Database schema version %d is ahead of the code's expected head %d. "
            "This usually means a newer binary created this database. "
            "Refusing to auto-delete — upgrade the binary or remove the database manually.",
            current_version,
            expected_head,
        )
        raise SchemaVersionError(
            f"Database schema version {current_version} is ahead of expected head "
            f"{expected_head}. Cannot start safely."
        )

    if current_version > 0:
        self.logger.info(
            "Database schema version %d is behind head %d — pending migrations will be applied.",
            current_version,
            expected_head,
        )
        return

    if current_version == 0:
        self.logger.warning(
            "Database has no schema version (expected %d) — recreating database (no production data to preserve).",
            expected_head,
        )
        try:
            await asyncio.to_thread(delete_db_files, db_path)
        except PermissionError as exc:
            raise RuntimeError(
                f"Cannot delete stale database file {db_path}: {exc}. Please remove it manually and restart."
            ) from exc

run_migrations() -> None

Run PRAGMA user_version migrations to the latest version (synchronous, called via to_thread).

auto_vacuum = INCREMENTAL is set by the runner before any tables are created.

Source code in src/hassette/core/database_service.py

def run_migrations(self) -> None:
    """Run PRAGMA user_version migrations to the latest version (synchronous, called via to_thread).

    auto_vacuum = INCREMENTAL is set by the runner before any tables are created.
    """
    run_migrations(self._db_path)

catch_up_cleanup_at_startup() -> None async

Run the size failsafe, then retention cleanup, once during on_initialize().

Runs directly on the write connection, before the write worker starts.

Source code in src/hassette/core/database_service.py

async def catch_up_cleanup_at_startup(self) -> None:
    """Run the size failsafe, then retention cleanup, once during on_initialize().

    Runs directly on the write connection, before the write worker starts.
    """
    # Both passes commit per batch, so a re-run after a restart converges without saved
    # progress. They share one budget so a large backlog can't exhaust the readiness timeout
    # wait_for_ready() enforces around on_initialize(). The rollback after an interrupt is
    # unbudgeted, but it waits on at most one batch-bounded statement.
    budget = self.hassette.config.lifecycle.startup_timeout_seconds * _STARTUP_CLEANUP_BUDGET_FRACTION
    try:
        async with asyncio.timeout(budget):
            await self._check_size_failsafe()
            await self._do_run_retention_cleanup()
    except TimeoutError:
        # Cancelling the await doesn't stop a statement already running on aiosqlite's worker
        # thread; interrupt it so the rollback isn't queued behind it.
        try:
            await self.db.interrupt()
        except STORAGE_ERRORS:
            self.logger.exception("Interrupting the startup cleanup catch-up statement failed")
        await safe_rollback(self.db, self, "startup cleanup catch-up")
        self.logger.warning(
            "Startup cleanup catch-up exceeded its %.1fs budget; the hourly cleanup loop will pick up the rest",
            budget,
        )
    except STORAGE_ERRORS:
        # A failure can leave a retention batch between its BEGIN and commit; roll back so the
        # write worker doesn't inherit an open transaction. Earlier batches stay committed.
        await safe_rollback(self.db, self, "startup cleanup catch-up")
        self.logger.exception("Startup cleanup catch-up failed; continuing without cleanup")

set_pragmas() -> None async

Configure SQLite PRAGMAs for performance and safety.

Source code in src/hassette/core/database_service.py

async def set_pragmas(self) -> None:
    """Configure SQLite PRAGMAs for performance and safety."""
    db = self.db
    await db.execute("PRAGMA journal_mode = WAL")
    await db.execute("PRAGMA wal_autocheckpoint = 1000")
    # NORMAL is an intentional performance tradeoff: in WAL mode, the last committed
    # writes before an OS crash (not app crash) may be lost if not yet checkpointed.
    # This is acceptable for operational telemetry — the orphan-session mechanism
    # compensates for session rows but not for individual telemetry records.
    await db.execute("PRAGMA synchronous = NORMAL")
    await db.execute(f"PRAGMA busy_timeout = {SQLITE_BUSY_TIMEOUT_MS}")
    await db.execute("PRAGMA foreign_keys = ON")
    # Intentionally a no-op — auto_vacuum is set by the migration runner before table creation.
    # This line documents intent only.
    await db.execute("PRAGMA auto_vacuum = INCREMENTAL")

update_heartbeat() -> None async

Await a heartbeat update for the current session.

Early-return guards run inline; the DB write is awaited via submit() so that _consecutive_heartbeat_failures is updated before returning.

The submit() call is bounded by _HEARTBEAT_WRITE_TIMEOUT_SECONDS. submit()'s own queue timeout only covers the wait before the write starts; a worker wedged inside a write never raises, so without this outer bound serve() would park on submit() forever and never reach its failure-count escalation. Either timeout surfaces as the same TimeoutError: the queue was backed up past its limit, or the worker is wedged. A timeout, or a WriteQueueUnavailableError from submit(), is a failed attempt identically to a raised sqlite3.Error/OSError/ValueError. A failed attempt counts a strike only when _write_progress_count has not moved since the previous attempt finished (or, on the first attempt, since this one started): a worker still committing writes, such as a long batched retention pass holding the queue, is busy rather than stuck and must not be failed.

This method is the only place _consecutive_heartbeat_failures moves, so one attempt costs at most one strike. Counting the timeout here and the raise inside the queued coroutine would let a single slow-then-failing write consume two of the three, failing the service a full heartbeat interval early.

Source code in src/hassette/core/database_service.py

async def update_heartbeat(self) -> None:
    """Await a heartbeat update for the current session.

    Early-return guards run inline; the DB write is awaited via submit()
    so that _consecutive_heartbeat_failures is updated before returning.

    The submit() call is bounded by _HEARTBEAT_WRITE_TIMEOUT_SECONDS. submit()'s own queue
    timeout only covers the wait before the write starts; a worker wedged *inside* a write
    never raises, so without this outer bound serve() would park on submit() forever and
    never reach its failure-count escalation. Either timeout surfaces as the same
    TimeoutError: the queue was backed up past its limit, or the worker is wedged.
    A timeout, or a WriteQueueUnavailableError from submit(), is a failed attempt identically to a
    raised sqlite3.Error/OSError/ValueError. A failed attempt counts a strike only when
    _write_progress_count has not moved since the previous attempt finished (or, on the first
    attempt, since this one started): a worker still committing writes, such as a long batched
    retention pass holding the queue, is busy rather than stuck and must not be failed.

    This method is the only place _consecutive_heartbeat_failures moves, so one attempt
    costs at most one strike. Counting the timeout here and the raise inside the queued
    coroutine would let a single slow-then-failing write consume two of the three,
    failing the service a full heartbeat interval early.
    """
    if self._db is None:
        return
    if self._db_write_queue is None:
        return
    try:
        _ = self.hassette.session_id
    except RuntimeError:
        return
    progress_baseline = self._progress_at_last_heartbeat
    if progress_baseline is None:
        progress_baseline = self._write_progress_count
    try:
        async with asyncio.timeout(_HEARTBEAT_WRITE_TIMEOUT_SECONDS):
            await self.submit(self._do_update_heartbeat())
    except TimeoutError as exc:
        self.record_heartbeat_failure(
            "Heartbeat write timed out — write queue backed up or worker wedged", exc, progress_baseline
        )
    except (sqlite3.Error, OSError, ValueError, WriteQueueUnavailableError) as exc:
        self.record_heartbeat_failure("Failed to update heartbeat", exc, progress_baseline)
    else:
        if self._consecutive_heartbeat_failures > 0:
            self.logger.info("Heartbeat recovered after %d failure(s)", self._consecutive_heartbeat_failures)
            self._consecutive_heartbeat_failures = 0
    finally:
        self._progress_at_last_heartbeat = self._write_progress_count

record_heartbeat_failure(message: str, exc: BaseException, progress_baseline: int) -> None

Count a strike for a failed heartbeat attempt unless writes progressed past progress_baseline.

exc is the attempt's failure; its traceback is attached to the logged entry either way.

Source code in src/hassette/core/database_service.py

def record_heartbeat_failure(self, message: str, exc: BaseException, progress_baseline: int) -> None:
    """Count a strike for a failed heartbeat attempt unless writes progressed past ``progress_baseline``.

    ``exc`` is the attempt's failure; its traceback is attached to the logged entry either way.
    """
    max_failures = self.hassette.config.database.max_consecutive_heartbeat_failures
    progress = self._write_progress_count - progress_baseline
    if progress > 0:
        self.logger.warning(
            "%s, but %d write(s) completed since the previous attempt; not counted (failures %d/%d)",
            message,
            progress,
            self._consecutive_heartbeat_failures,
            max_failures,
            exc_info=exc,
        )
        return
    self._consecutive_heartbeat_failures += 1
    self.logger.error(
        "%s with no write progress since the previous attempt (failure %d/%d)",
        message,
        self._consecutive_heartbeat_failures,
        max_failures,
        exc_info=exc,
    )

CleanupAttempt dataclass

When a maintenance task last tried to enqueue its cleanup, and whether the enqueue went through.

Source code in src/hassette/core/database_service.py

@dataclass(frozen=True)
class CleanupAttempt:
    """When a maintenance task last tried to enqueue its cleanup, and whether the enqueue went through."""

    attempted_at: float
    enqueued: bool

    def next_due(self, interval: float, *, retry_interval: float) -> float:
        """Return the monotonic time of the next attempt.

        A successful enqueue waits the full ``interval``. A dropped one retries after
        ``min(interval, retry_interval)``: skipping a whole interval would starve the cleanup
        under the sustained write-queue backlog it exists to relieve.
        """
        delay = interval if self.enqueued else min(interval, retry_interval)
        return self.attempted_at + delay

next_due(interval: float, *, retry_interval: float) -> float

Return the monotonic time of the next attempt.

A successful enqueue waits the full interval. A dropped one retries after min(interval, retry_interval): skipping a whole interval would starve the cleanup under the sustained write-queue backlog it exists to relieve.

Source code in src/hassette/core/database_service.py

def next_due(self, interval: float, *, retry_interval: float) -> float:
    """Return the monotonic time of the next attempt.

    A successful enqueue waits the full ``interval``. A dropped one retries after
    ``min(interval, retry_interval)``: skipping a whole interval would starve the cleanup
    under the sustained write-queue backlog it exists to relieve.
    """
    delay = interval if self.enqueued else min(interval, retry_interval)
    return self.attempted_at + delay

maintenance_deadlines(config: DatabaseConfig, last_heartbeat: float, retention_attempt: CleanupAttempt, size_failsafe_attempt: CleanupAttempt) -> tuple[float, float, float]

Return the monotonic (heartbeat, retention, size failsafe) deadlines under config.

A dropped cleanup enqueue retries on the heartbeat cadence (see CleanupAttempt.next_due).

Source code in src/hassette/core/database_service.py

def maintenance_deadlines(
    config: DatabaseConfig,
    last_heartbeat: float,
    retention_attempt: CleanupAttempt,
    size_failsafe_attempt: CleanupAttempt,
) -> tuple[float, float, float]:
    """Return the monotonic (heartbeat, retention, size failsafe) deadlines under ``config``.

    A dropped cleanup enqueue retries on the heartbeat cadence (see ``CleanupAttempt.next_due``).
    """
    retry = config.heartbeat_interval_seconds
    return (
        last_heartbeat + config.heartbeat_interval_seconds,
        retention_attempt.next_due(config.retention_interval_seconds, retry_interval=retry),
        size_failsafe_attempt.next_due(config.size_failsafe_interval_seconds, retry_interval=retry),
    )

log_shutdown_snapshot_outcome(service: DatabaseService, snapshot_path: Path, copy: asyncio.Task[bool]) -> None

Log how DatabaseService.write_shutdown_snapshot()'s copy ended.

Source code in src/hassette/core/database_service.py

def log_shutdown_snapshot_outcome(service: DatabaseService, snapshot_path: Path, copy: "asyncio.Task[bool]") -> None:
    """Log how ``DatabaseService.write_shutdown_snapshot()``'s copy ended."""
    if copy.cancelled():
        service.logger.warning("The shutdown snapshot to %s was cancelled", snapshot_path)
        return
    exc = copy.exception()
    if exc is not None:
        service.logger.error("Couldn't write the shutdown snapshot to %s", snapshot_path, exc_info=exc)
    elif copy.result():
        service.logger.info("Wrote the shutdown snapshot to %s", snapshot_path)
    else:
        service.logger.debug("No database at %s; skipped the shutdown snapshot", service.resolve_db_path())