From 95a378a8f1e14fd4756190936e6b98c55e8d2b57 Mon Sep 17 00:00:00 2001 From: BiFangKNT <1320414964@qq.com> Date: Sat, 24 Oct 2026 02:03:41 +0800 Subject: [PATCH] fix(persistence): reclaim oversized SQLite WAL files - Require successful WAL truncation before starting application producers - Preserve completed migrations and retry checkpoint failures on restart - Set a 64 MiB WAL retention limit on every SQLite connection - Preserve default automatic checkpoint behavior and reader snapshots - Cover startup failures, connection recreation, and WAL reclamation --- ARCHITECTURE.md | 2 + .../pkg/persistence/databases/sqlite.py | 14 ++ src/langbot/pkg/persistence/mgr.py | 6 + src/langbot/pkg/persistence/sqlite_startup.py | 27 +++ .../persistence/test_sqlite_startup.py | 203 ++++++++++++++++++ 5 files changed, 252 insertions(+) create mode 100644 src/langbot/pkg/persistence/sqlite_startup.py create mode 100644 tests/integration/persistence/test_sqlite_startup.py diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 9cb242493..913fc982b 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -273,6 +273,8 @@ Maintenance rule: when adding, removing, or changing an HTTP endpoint that shoul Persistence is centered on `pkg/persistence/mgr.py`. - SQLite is the default database; PostgreSQL is supported. +- Every SQLite connection sets a 64 MiB `journal_size_limit`, including connections recreated after migration or pool disposal. SQLite's default automatic checkpoint policy and synchronous setting remain unchanged. The limit reduces retained WAL space when the WAL resets; it does not cap active transactions or WAL pinned by readers. Online maintenance does not periodically force checkpoints or truncate WAL files. Diagnose persistent excess growth by checking checkpoint progress and actual transaction lifetimes, then shorten the responsible transaction rather than forcing readers to finish. +- After schema initialization and migration, every SQLite startup completes `wal_checkpoint(TRUNCATE)` before runtime setup and producers start. Lock waits allow up to 60 seconds; checkpoint I/O may take longer. An incomplete checkpoint fails startup without undoing successful migrations, and the next startup retries even at schema head. The connection busy timeout is restored afterward. PostgreSQL is unaffected. - SQLite files use incremental auto-vacuum. Migration `0042_sqlite_incremental_vacuum` converts existing files before producers start, with a verified backup and restoration on conversion failure when a full rebuild is required. Fresh files enable the format before creating tables. Monitoring cleanup reclaims free pages in independent adaptive batches (1–512 pages, starting at 8, targeting 25 ms), pausing 250 ms between batches and starting no new batches after a two-second round budget. Busy writers take priority. No full `VACUUM` or forced WAL checkpoint runs during online cleanup. - Models live under `pkg/entity/persistence/`. - `pkg/persistence/media.py::MediaReferenceText` validates bound conversation payloads on ORM/Core writes while preserving the existing TEXT schema and string API. It rejects recognized structured inline images: retained images go through `MediaCache.externalize_chain_dump()` before serialization, while metadata-only snapshots and transcripts use `redact_inline_images()`. Reads and comparisons preserve historical values for incremental cleanup. Runtime/adapter messages, arbitrary diagnostic text, and opaque strings are outside this contract. diff --git a/src/langbot/pkg/persistence/databases/sqlite.py b/src/langbot/pkg/persistence/databases/sqlite.py index f314b2522..f0d699e3a 100644 --- a/src/langbot/pkg/persistence/databases/sqlite.py +++ b/src/langbot/pkg/persistence/databases/sqlite.py @@ -6,6 +6,19 @@ import sqlalchemy.ext.asyncio as sqlalchemy_asyncio from .. import database +WAL_RETAIN_BYTES = 64 * 1024 * 1024 + + +def _configure_connection(dbapi_connection, _connection_record): + # This limit applies when a WAL resets, not while a transaction or reader + # still needs its frames. Keep SQLite's automatic checkpoint policy. + cursor = dbapi_connection.cursor() + try: + cursor.execute(f'PRAGMA journal_size_limit={WAL_RETAIN_BYTES}') + finally: + cursor.close() + + @database.manager_class('sqlite') class SQLiteDatabaseManager(database.BaseDatabaseManager): """SQLite database manager""" @@ -14,6 +27,7 @@ class SQLiteDatabaseManager(database.BaseDatabaseManager): db_file_path = self.ap.instance_config.data.get('database', {}).get('sqlite', {}).get('path', 'data/langbot.db') engine_url = f'sqlite+aiosqlite:///{db_file_path}' self.engine = sqlalchemy_asyncio.create_async_engine(engine_url) + sqlalchemy.event.listen(self.engine.sync_engine, 'connect', _configure_connection) # Reader transactions must not block independent plugin callback commits. # Configure the persistent journal before serving requests, not per write. async with self.engine.connect() as connection: diff --git a/src/langbot/pkg/persistence/mgr.py b/src/langbot/pkg/persistence/mgr.py index 6327cd3d4..445af890a 100644 --- a/src/langbot/pkg/persistence/mgr.py +++ b/src/langbot/pkg/persistence/mgr.py @@ -198,6 +198,12 @@ class PersistenceManager: return await self._initialize_managed_schema() + from .sqlite_startup import checkpoint_after_migration + + if engine.dialect.name == 'sqlite': + self.ap.logger.info('Completing SQLite startup WAL checkpoint before starting producers...') + await checkpoint_after_migration(engine) + self.ap.logger.info('SQLite startup WAL checkpoint completed.') await self._enable_sqlite_foreign_keys_after_migration() from .resource_urls import install diff --git a/src/langbot/pkg/persistence/sqlite_startup.py b/src/langbot/pkg/persistence/sqlite_startup.py new file mode 100644 index 000000000..e97051229 --- /dev/null +++ b/src/langbot/pkg/persistence/sqlite_startup.py @@ -0,0 +1,27 @@ +"""Finish SQLite WAL maintenance before application producers start.""" + +STARTUP_BUSY_TIMEOUT_MS = 60_000 + + +async def checkpoint_after_migration(engine): + """Require a complete WAL checkpoint, including on a retry at schema head.""" + if engine.dialect.name != 'sqlite': + return + async with engine.connect() as connection: + connection = await connection.execution_options(isolation_level='AUTOCOMMIT') + if (await connection.exec_driver_sql('PRAGMA journal_mode')).scalar_one().lower() != 'wal': + return + timeout = (await connection.exec_driver_sql('PRAGMA busy_timeout')).scalar_one() + try: + await connection.exec_driver_sql(f'PRAGMA busy_timeout={STARTUP_BUSY_TIMEOUT_MS}') + busy, frames, checkpointed = ( + await connection.exec_driver_sql('PRAGMA main.wal_checkpoint(TRUNCATE)') + ).one() + if (busy, frames, checkpointed) != (0, 0, 0): + raise RuntimeError( + 'SQLite startup WAL checkpoint did not complete: ' + f'busy={busy}, frames={frames}, checkpointed={checkpointed}. ' + 'Stop other database users and restart LangBot to retry.' + ) + finally: + await connection.exec_driver_sql(f'PRAGMA busy_timeout={timeout}') diff --git a/tests/integration/persistence/test_sqlite_startup.py b/tests/integration/persistence/test_sqlite_startup.py new file mode 100644 index 000000000..4bfd0e8eb --- /dev/null +++ b/tests/integration/persistence/test_sqlite_startup.py @@ -0,0 +1,203 @@ +"""Verify startup checkpoint completion, contention, and retry semantics.""" + +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest +from sqlalchemy.ext.asyncio import create_async_engine + +from langbot.pkg.persistence import alembic_runner, resource_urls, sqlite_startup +from tests.integration.persistence.test_sqlite_migration_backup import _manager + +pytestmark = [pytest.mark.integration, pytest.mark.asyncio] + + +async def _prepare(engine): + async with engine.connect() as connection: + await connection.exec_driver_sql('PRAGMA journal_mode=WAL') + await connection.exec_driver_sql('PRAGMA wal_autocheckpoint=0') + await connection.exec_driver_sql('CREATE TABLE retained (value TEXT)') + await connection.exec_driver_sql("INSERT INTO retained VALUES ('keep')") + await connection.commit() + await alembic_runner.run_alembic_stamp(engine, alembic_runner.get_alembic_head()) + + +async def test_checkpoint_truncates_wal_and_preserves_committed_data(tmp_path): + path = tmp_path / 'startup.db' + engine = create_async_engine(f'sqlite+aiosqlite:///{path}') + try: + await _prepare(engine) + wal = path.with_name(path.name + '-wal') + assert wal.stat().st_size > 0 + async with engine.connect() as connection: + timeout = (await connection.exec_driver_sql('PRAGMA busy_timeout')).scalar_one() + await sqlite_startup.checkpoint_after_migration(engine) + assert wal.stat().st_size == 0 + async with engine.connect() as connection: + assert (await connection.exec_driver_sql('SELECT value FROM retained')).scalar_one() == 'keep' + assert (await connection.exec_driver_sql('PRAGMA busy_timeout')).scalar_one() == timeout + finally: + await engine.dispose() + + +@pytest.mark.parametrize('blocker', ['reader', 'writer']) +async def test_busy_checkpoint_preserves_revision_and_retries(tmp_path, monkeypatch, blocker): + path = tmp_path / 'busy.db' + engine = create_async_engine(f'sqlite+aiosqlite:///{path}') + monkeypatch.setattr(sqlite_startup, 'STARTUP_BUSY_TIMEOUT_MS', 20) + try: + await _prepare(engine) + async with engine.connect() as occupied: + await occupied.exec_driver_sql('BEGIN' if blocker == 'reader' else 'BEGIN IMMEDIATE') + await occupied.exec_driver_sql('SELECT * FROM retained') + if blocker == 'reader': + async with engine.begin() as writer: + await writer.exec_driver_sql("INSERT INTO retained VALUES ('new')") + with pytest.raises(RuntimeError, match='startup WAL checkpoint did not complete'): + await sqlite_startup.checkpoint_after_migration(engine) + async with engine.connect() as connection: + assert (await connection.exec_driver_sql('PRAGMA busy_timeout')).scalar_one() == 5000 + assert await alembic_runner.get_alembic_current(engine) == alembic_runner.get_alembic_head() + await occupied.rollback() + await sqlite_startup.checkpoint_after_migration(engine) + assert path.with_name(path.name + '-wal').stat().st_size == 0 + finally: + await engine.dispose() + + +async def test_initialize_stops_before_runtime_setup_and_retries_at_head(tmp_path, monkeypatch): + from langbot.pkg.persistence import database + + engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "gate.db"}') + manager = _manager(engine) + manager.ap.instance_config = SimpleNamespace(data={'database': {'use': 'sqlite'}}) + db = SimpleNamespace(engine=engine, initialize=AsyncMock(), get_engine=lambda: engine) + + class TestDatabase: + name = 'sqlite' + + def __new__(cls, *args, **kwargs): + return db + + monkeypatch.setattr(database, 'preregistered_managers', [TestDatabase]) + manager._initialize_managed_schema = AsyncMock() + manager._enable_sqlite_foreign_keys_after_migration = AsyncMock() + manager.write_space_model_providers = AsyncMock() + install = Mock() + monkeypatch.setattr(resource_urls, 'install', install) + checkpoint = AsyncMock(side_effect=RuntimeError('checkpoint failure')) + monkeypatch.setattr(sqlite_startup, 'checkpoint_after_migration', checkpoint) + try: + with pytest.raises(RuntimeError, match='checkpoint failure'): + await manager.initialize() + install.assert_not_called() + manager.write_space_model_providers.assert_not_awaited() + manager._initialize_managed_schema.assert_awaited_once() + manager._enable_sqlite_foreign_keys_after_migration.assert_not_awaited() + checkpoint.side_effect = None + await manager.initialize() + assert checkpoint.await_count == 2 + install.assert_called_once() + manager._enable_sqlite_foreign_keys_after_migration.assert_awaited_once() + manager.write_space_model_providers.assert_awaited_once() + finally: + await engine.dispose() + + +async def test_checkpoint_waits_for_writer_to_finish(tmp_path, monkeypatch): + engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "wait.db"}') + monkeypatch.setattr(sqlite_startup, 'STARTUP_BUSY_TIMEOUT_MS', 2000) + try: + await _prepare(engine) + async with engine.connect() as writer: + await writer.exec_driver_sql('BEGIN IMMEDIATE') + task = asyncio.create_task(sqlite_startup.checkpoint_after_migration(engine)) + try: + await asyncio.sleep(0.05) + assert not task.done() + await writer.rollback() + await task + finally: + await writer.rollback() + await task + finally: + await engine.dispose() + + +async def test_non_sqlite_checkpoint_does_not_connect(): + engine = SimpleNamespace(dialect=SimpleNamespace(name='postgresql'), connect=Mock()) + await sqlite_startup.checkpoint_after_migration(engine) + engine.connect.assert_not_called() + + +def _database_manager(tmp_path): + from langbot.pkg.persistence.databases.sqlite import SQLiteDatabaseManager + + ap = SimpleNamespace( + instance_config=SimpleNamespace(data={'database': {'sqlite': {'path': str(tmp_path / 'wal.db')}}}) + ) + return SQLiteDatabaseManager(ap) + + +async def test_wal_policy_applies_to_all_connections_after_pool_disposal(tmp_path): + manager = _database_manager(tmp_path) + await manager.initialize() + try: + for _ in range(2): + async with manager.engine.connect() as first, manager.engine.connect() as second: + for connection in (first, second): + assert (await connection.exec_driver_sql('PRAGMA journal_size_limit')).scalar_one() == 64 * 1024**2 + assert (await connection.exec_driver_sql('PRAGMA wal_autocheckpoint')).scalar_one() == 1000 + assert (await connection.exec_driver_sql('PRAGMA synchronous')).scalar_one() == 2 + await manager.engine.dispose() + finally: + await manager.engine.dispose() + + +async def test_large_wal_shrinks_on_reuse_without_forced_checkpoint(tmp_path): + manager = _database_manager(tmp_path) + await manager.initialize() + wal = tmp_path / 'wal.db-wal' + try: + async with manager.engine.connect() as connection: + await connection.exec_driver_sql('CREATE TABLE payload (data BLOB)') + await connection.exec_driver_sql('INSERT INTO payload VALUES (zeroblob(73400320))') + await connection.commit() + assert wal.stat().st_size > 64 * 1024**2 + await connection.exec_driver_sql('INSERT INTO payload VALUES (NULL)') + await connection.commit() + assert wal.stat().st_size == 64 * 1024**2 + assert (await connection.exec_driver_sql('SELECT count(*) FROM payload')).scalar_one() == 2 + finally: + await manager.engine.dispose() + + +async def test_retention_limit_preserves_wal_needed_by_reader(tmp_path): + manager = _database_manager(tmp_path) + await manager.initialize() + wal = tmp_path / 'wal.db-wal' + try: + async with manager.engine.connect() as writer, manager.engine.connect() as reader: + await writer.exec_driver_sql('CREATE TABLE payload (data BLOB)') + await writer.exec_driver_sql('INSERT INTO payload VALUES (zeroblob(8388608))') + await writer.commit() + await writer.exec_driver_sql('PRAGMA wal_checkpoint(TRUNCATE)') + await writer.commit() + await reader.exec_driver_sql('BEGIN') + await reader.exec_driver_sql('SELECT length(data) FROM payload') + for _ in range(9): + await writer.exec_driver_sql('UPDATE payload SET data=randomblob(8388608)') + await writer.commit() + assert wal.stat().st_size > 64 * 1024**2 + assert (await reader.exec_driver_sql('SELECT hex(substr(data, 1, 1)) FROM payload')).scalar_one() == '00' + _, frames, checkpointed = (await writer.exec_driver_sql('PRAGMA wal_checkpoint(PASSIVE)')).one() + assert frames > checkpointed + await reader.rollback() + _, frames, checkpointed = (await writer.exec_driver_sql('PRAGMA wal_checkpoint(PASSIVE)')).one() + assert frames == checkpointed + await writer.exec_driver_sql('INSERT INTO payload VALUES (NULL)') + await writer.commit() + assert wal.stat().st_size == 64 * 1024**2 + finally: + await manager.engine.dispose()