diff --git a/libervia/backend/core/main.py b/libervia/backend/core/main.py index 42762712..17ef078b 100644 --- a/libervia/backend/core/main.py +++ b/libervia/backend/core/main.py @@ -540,7 +540,7 @@ class LiberviaBackend(service.Service): continue else: defers_list.append(utils.as_deferred(unload)) - return defers_list + return defer.DeferredList(defers_list) async def _sync_components_profiles(self) -> None: """Synchronize components profiles data with settings. @@ -799,7 +799,7 @@ class LiberviaBackend(service.Service): shutil.rmtree(self.local_shared_path, ignore_errors=True) log.info("Salut aussi à Rantanplan.") - return self.plugins_unload() + return defer.DeferredList([self.plugins_unload(), G.storage.unload()]) def run(self): log.debug(_("running app")) diff --git a/libervia/backend/memory/sqla.py b/libervia/backend/memory/sqla.py index 2b265cb8..e277d7bc 100644 --- a/libervia/backend/memory/sqla.py +++ b/libervia/backend/memory/sqla.py @@ -24,7 +24,7 @@ import json from pathlib import Path import sys import time -from typing import Any, Dict, List, Optional, Tuple, Union +from typing import Any, Dict, List, Literal, Optional, Tuple, Union from alembic import config as al_config, script as al_script from alembic.runtime import migration as al_migration @@ -128,6 +128,7 @@ class Storage: def __init__(self): self.initialized = defer.Deferred() + self.engine: AsyncEngine # we keep cache for the profiles (key: profile name, value: profile id) # profile id to component entry point self.components: Dict[int, str] = {} @@ -142,6 +143,17 @@ class Storage: def get_profile_by_name(self, profile: str) -> str: return self.profiles[profile] + @aio + async def unload(self) -> None: + """Clean up database resources. + + Flushes WAL and disposes engine. + This method is called on shutdown. + """ + if self.engine.dialect.name == "sqlite": + await self.wal_checkpoint.aio_func() + await self.engine.dispose() + async def migrate_apply(self, *args: str, log_output: bool = False) -> None: """Do a migration command @@ -197,11 +209,19 @@ class Storage: return set(context.get_current_heads()) == set(directory.get_heads()) def _sqlite_set_journal_mode_wal(self, conn: Connection) -> None: - """Check if journal mode is WAL, and set it if necesssary""" + """Check if journal mode is WAL, set it if necessary and checkpoint + + If WAL was already active from a previous run, checkpoints at startup + so the main DB has all data, avoiding slow reads on first login. + """ result = conn.execute(text("PRAGMA journal_mode")) if result.scalar() != "wal": log.info("WAL mode not activated, activating it") conn.execute(text("PRAGMA journal_mode=WAL")) + else: + # WAL was already active, checkpoint at startup + result = conn.execute(text("PRAGMA wal_checkpoint(TRUNCATE)")) + log.debug(f"WAL checkpoint at startup: {result.fetchone()}") async def check_and_update_db(self, engine: AsyncEngine, db_config: dict) -> None: """Check that database is up-to-date, and update if necessary""" @@ -220,7 +240,7 @@ class Storage: log.info(_("Connecting database")) db_config = sqla_config.get_db_config() - engine = create_async_engine( + engine = self.engine = create_async_engine( db_config["url"], future=True, json_serializer=lambda obj: json.dumps(obj, ensure_ascii=False), @@ -248,6 +268,26 @@ class Storage: self.initialized.callback(None) + @aio + async def wal_checkpoint( + self, mode: Literal["PASSIVE", "FULL", "RESTART", "TRUNCATE", "NOOP"] = "TRUNCATE" + ) -> tuple[int, int, int]: + """Checkpoint WAL file to flush pending changes to main DB + + See https://sqlite.org/pragma.html#pragma_wal_checkpoint for details. + + @param mode: checkpoint mode + @return: (num_checkpointed, num_dirty_pages, num_pages) + """ + async with self.session() as session: + result = await session.execute(text(f"PRAGMA wal_checkpoint({mode})")) + fetched = result.fetchone() + log.debug( + f"WAL checkpoint ({mode}): " + f"checkpointed={fetched[0]}, dirty_pages={fetched[1]}, pages={fetched[2]}" + ) + return fetched + ## Generic @aio