diff --git a/CHANGELOG.md b/CHANGELOG.md index 90c2acf..f384ce3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -112,6 +112,11 @@ semantic versioning. - Data retention now runs shortly after startup and then daily. It no longer requires 24 hours of uninterrupted uptime before the first cleanup, and its timer remains independent from the nightly maintenance email. +- Retention deletes now commit in configurable chunks and yield between + batches, preventing a large first cleanup from monopolizing SQLite's writer + lock on SD-card installations. +- Linux service installers now use 1GB memory and 200% CPU ceilings, providing + Raspberry Pi graph and web-viewer workloads with practical headroom. ### Removed diff --git a/config.ini.example b/config.ini.example index 70418d0..17b985a 100644 --- a/config.ini.example +++ b/config.ini.example @@ -934,6 +934,13 @@ anonymize_users = false # The scheduler runs cleanup daily so retention is enforced even when the # standalone web viewer is not running. Shorter retention reduces DB size. # +# Old rows are deleted in committed chunks so live packet/contact/graph writers +# can run between batches. These defaults suit Raspberry Pi and SD-card installs. +# Smaller batches/longer pauses reduce I/O bursts but take longer to drain a +# large first-run backlog. +retention_delete_batch_size = 1000 +retention_delete_pause_seconds = 0.1 +# # Packet stream (web viewer real-time display and transmission_tracker) # 2-3 days is enough for most deployments; 7 days if you need longer history. packet_stream_retention_days = 3 @@ -1153,6 +1160,7 @@ graph_path_validation_obs_divisor = 50.0 # 14 - Good balance of coverage vs. memory (default for unconfigured installs) # 7 - Reduced memory footprint for Raspberry Pi Zero 2 W # Note: edges older than graph_edge_expiration_days are never loaded regardless of this value. +# Raspberry Pi 4 / SD-card profile: use 7 to make that effective bound explicit. graph_startup_load_days = 0 # Enable graph data capture from incoming packets (default: true) diff --git a/docs/path-command-config.md b/docs/path-command-config.md index 6094947..8744306 100644 --- a/docs/path-command-config.md +++ b/docs/path-command-config.md @@ -239,6 +239,63 @@ These settings control how graph edges are stored in the database. - Set to `false` on devices that don't use the path command - Default: `true` +## Raspberry Pi / Low-Power Profile + +For a Raspberry Pi 4 using an SD card, start with: + +```ini +[Path_Command] +graph_based_validation = true +graph_capture_enabled = true +min_edge_observations = 5 +graph_edge_expiration_days = 7 +graph_startup_load_days = 7 +graph_write_strategy = batched +graph_batch_interval_seconds = 60 +graph_batch_max_pending = 250 +graph_use_bidirectional = true +graph_use_hop_position = true +graph_multi_hop_enabled = true +graph_multi_hop_max_hops = 2 +graph_prefer_stored_keys = true + +[Data_Retention] +packet_stream_retention_days = 3 +observed_paths_retention_days = 30 +mesh_connections_retention_days = 7 +daily_stats_retention_days = 90 +purging_log_retention_days = 90 +retention_delete_batch_size = 1000 +retention_delete_pause_seconds = 0.1 +``` + +`batched` persistence groups dirty edges into one transaction, reducing WAL +and SD-card transaction churn. A 60-second interval can lose at most roughly +one minute of not-yet-flushed graph updates after abrupt power loss. Use +`hybrid` if immediate persistence of newly discovered edges matters more than +write volume. + +`observed_paths_retention_days` is the main bound on multi-byte graph +derivation work because lifetime edge identity and counts are derived from all +retained path evidence before the selected browser timeframe is applied. +Thirty days is a practical Pi baseline; use 14 days for very busy meshes or 90 +days when historical fidelity matters more than database size and query cost. + +Retention deletes are committed in chunks and pause briefly between batches. +This prevents a large first cleanup from holding SQLite's only writer lock for +minutes. Reduce `retention_delete_batch_size` or increase +`retention_delete_pause_seconds` if maintenance still causes visible I/O-wait +spikes; doing so lengthens the cleanup. + +For the web graph, use **Multi-byte Only**, a **72-hour edge window**, a +**7-day node window**, and a minimum observation threshold around **5**. These +settings reduce response and rendering work; retained-history length controls +the underlying multi-byte aggregation cost. + +The installed systemd service allows up to 1GB of memory and 200% CPU (two +cores). These are upper limits rather than reserved resources. A USB SSD is +still the most effective way to reduce SD-card wear on high-volume nodes. + ## Preset Configurations ### `balanced` (Default) diff --git a/docs/service-installation.md b/docs/service-installation.md index 64474f5..338d0ae 100644 --- a/docs/service-installation.md +++ b/docs/service-installation.md @@ -197,7 +197,7 @@ sudo systemctl restart meshcore-bot - No shell access for service user - Executable code and the virtual environment are root-owned - Only configuration, state, and log directories are service-writable -- Resource limits (512MB RAM, 50% CPU) +- Resource limits (1GB RAM, up to two CPU cores) ### Reliability - Automatic restart on failure @@ -238,10 +238,15 @@ environment from the current requirements. ### High Resource Usage The service has built-in limits: -- Memory: 512MB maximum -- CPU: 50% maximum +- Memory: 1GB maximum +- CPU: 200% maximum (up to two fully utilized CPU cores) - File descriptors: 65536 maximum +These are ceilings, not reservations. The 1GB/200% baseline gives the bot and +its web-viewer child process enough headroom for graph loading and bounded +SQLite maintenance on Raspberry Pi 4-class systems without allowing them to +consume the whole host. + ## Uninstallation To completely remove the service: diff --git a/meshcore-bot.service b/meshcore-bot.service index ba0a9a2..f9a59ef 100644 --- a/meshcore-bot.service +++ b/meshcore-bot.service @@ -39,8 +39,8 @@ ReadWritePaths=/var/log/meshcore-bot # Resource limits LimitNOFILE=65536 -MemoryMax=512M -CPUQuota=50% +MemoryMax=1G +CPUQuota=200% # Restart policy StartLimitInterval=60 diff --git a/modules/commands/stats_command.py b/modules/commands/stats_command.py index 9a1ea9e..05bd4ac 100644 --- a/modules/commands/stats_command.py +++ b/modules/commands/stats_command.py @@ -809,27 +809,26 @@ class StatsCommand(BaseCommand): cutoff_time = now - (days_to_keep * 24 * 60 * 60) future_cutoff = now + self.FUTURE_TIMESTAMP_GRACE_SECONDS - with self.bot.db_manager.connection() as conn: - cursor = conn.cursor() - - deleted = {} - for table in ('message_stats', 'command_stats', 'path_stats'): - cursor.execute( - f'DELETE FROM {table} WHERE timestamp < ? OR timestamp > ?', # noqa: S608 - fixed table names - (cutoff_time, future_cutoff), + deleted = {} + for table in ('message_stats', 'command_stats', 'path_stats'): + deleted[table] = ( + self.bot.db_manager.delete_timestamp_rows_in_chunks( + table, + 'timestamp', + cutoff_time, + future_cutoff=future_cutoff, + progress_label=table.replace('_', ' '), ) - deleted[table] = cursor.rowcount + ) - conn.commit() - - total_deleted = sum(deleted.values()) - if total_deleted > 0: - self.logger.info( - f"Cleaned up {total_deleted} old stats entries " - f"({deleted['message_stats']} messages, " - f"{deleted['command_stats']} commands, " - f"{deleted['path_stats']} paths)" - ) + total_deleted = sum(deleted.values()) + if total_deleted > 0: + self.logger.info( + f"Cleaned up {total_deleted} old stats entries " + f"({deleted['message_stats']} messages, " + f"{deleted['command_stats']} commands, " + f"{deleted['path_stats']} paths)" + ) except Exception as e: self.logger.error(f"Error cleaning up old stats: {e}") diff --git a/modules/db_manager.py b/modules/db_manager.py index a80c090..c287c4b 100644 --- a/modules/db_manager.py +++ b/modules/db_manager.py @@ -10,11 +10,15 @@ import re import sqlite3 from collections.abc import AsyncGenerator, Generator from contextlib import asynccontextmanager, contextmanager -from datetime import date, datetime +from datetime import date, datetime, timezone from pathlib import Path from typing import Any, Optional from .db_migrations import MigrationRunner +from .db_retention import ( + delete_timestamp_rows_in_chunks, + retention_delete_settings, +) from .security_utils import VALID_JOURNAL_MODES @@ -44,6 +48,8 @@ class DBManager: 'bot_metadata', 'packet_stream', 'message_stats', + 'command_stats', + 'path_stats', 'greeted_users', 'repeater_contacts', 'complete_contact_tracking', # Repeater manager @@ -250,22 +256,30 @@ class DBManager: expiration timestamp has passed. """ try: - with self.connection() as conn: - cursor = conn.cursor() + cutoff = ( + datetime.now(timezone.utc) + .replace(tzinfo=None) + .isoformat(sep=" ", timespec="seconds") + ) + geocoding_deleted = self.delete_timestamp_rows_in_chunks( + 'geocoding_cache', + 'expires_at', + cutoff, + progress_label='geocoding cache', + ) + generic_deleted = self.delete_timestamp_rows_in_chunks( + 'generic_cache', + 'expires_at', + cutoff, + progress_label='generic cache', + ) - # Clean up geocoding cache - cursor.execute("DELETE FROM geocoding_cache WHERE expires_at < datetime('now')") - geocoding_deleted = cursor.rowcount - - # Clean up generic cache - cursor.execute("DELETE FROM generic_cache WHERE expires_at < datetime('now')") - generic_deleted = cursor.rowcount - - conn.commit() - - total_deleted = geocoding_deleted + generic_deleted - if total_deleted > 0: - self.logger.info(f"Cleaned up {total_deleted} expired cache entries ({geocoding_deleted} geocoding, {generic_deleted} generic)") + total_deleted = geocoding_deleted + generic_deleted + if total_deleted > 0: + self.logger.info( + f"Cleaned up {total_deleted} expired cache entries " + f"({geocoding_deleted} geocoding, {generic_deleted} generic)" + ) except Exception as e: self.logger.error(f"Error cleaning up expired cache: {e}") @@ -273,13 +287,21 @@ class DBManager: def cleanup_geocoding_cache(self) -> None: """Remove expired geocoding cache entries""" try: - with self.connection() as conn: - cursor = conn.cursor() - cursor.execute("DELETE FROM geocoding_cache WHERE expires_at < datetime('now')") - deleted_count = cursor.rowcount - conn.commit() - if deleted_count > 0: - self.logger.info(f"Cleaned up {deleted_count} expired geocoding cache entries") + cutoff = ( + datetime.now(timezone.utc) + .replace(tzinfo=None) + .isoformat(sep=" ", timespec="seconds") + ) + deleted_count = self.delete_timestamp_rows_in_chunks( + 'geocoding_cache', + 'expires_at', + cutoff, + progress_label='geocoding cache', + ) + if deleted_count > 0: + self.logger.info( + f"Cleaned up {deleted_count} expired geocoding cache entries" + ) except Exception as e: self.logger.error(f"Error cleaning up geocoding cache: {e}") @@ -418,6 +440,35 @@ class DBManager: self.logger.error(f"Error executing update: {e}") return 0 + def delete_timestamp_rows_in_chunks( + self, + table: str, + timestamp_column: str, + cutoff: Any, + *, + future_cutoff: Any | None = None, + progress_label: str | None = None, + ) -> int: + """Delete retained history without monopolizing SQLite's writer lock.""" + if table not in self.ALLOWED_TABLES: + raise ValueError( + f"Table name '{table}' not in allowed tables whitelist" + ) + batch_size, pause_seconds = retention_delete_settings( + getattr(self.bot, "config", None) + ) + return delete_timestamp_rows_in_chunks( + self.connection, + table, + timestamp_column, + cutoff, + batch_size=batch_size, + pause_seconds=pause_seconds, + future_cutoff=future_cutoff, + logger=self.logger, + progress_label=progress_label, + ) + def execute_query_on_connection(self, conn: sqlite3.Connection, query: str, params: tuple = ()) -> list[dict]: """Execute a query on an existing connection. Caller owns the connection.""" cursor = conn.cursor() diff --git a/modules/db_retention.py b/modules/db_retention.py new file mode 100644 index 0000000..3fe803e --- /dev/null +++ b/modules/db_retention.py @@ -0,0 +1,113 @@ +"""SQLite retention helpers that keep writer lock hold times bounded.""" + +from __future__ import annotations + +import re +import sqlite3 +import time +from collections.abc import Callable +from contextlib import AbstractContextManager +from typing import Any + +DEFAULT_RETENTION_DELETE_BATCH_SIZE = 1000 +DEFAULT_RETENTION_DELETE_PAUSE_SECONDS = 0.1 +_MAX_RETENTION_DELETE_BATCH_SIZE = 10_000 +_MAX_RETENTION_DELETE_PAUSE_SECONDS = 5.0 +_VALID_IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") + + +def retention_delete_settings(config: Any) -> tuple[int, float]: + """Return bounded chunk size and inter-batch pause from configuration.""" + batch_size = DEFAULT_RETENTION_DELETE_BATCH_SIZE + pause_seconds = DEFAULT_RETENTION_DELETE_PAUSE_SECONDS + try: + batch_size = int( + config.getint( + "Data_Retention", + "retention_delete_batch_size", + fallback=batch_size, + ) + ) + except (AttributeError, TypeError, ValueError): + batch_size = DEFAULT_RETENTION_DELETE_BATCH_SIZE + try: + pause_seconds = float( + config.getfloat( + "Data_Retention", + "retention_delete_pause_seconds", + fallback=pause_seconds, + ) + ) + except (AttributeError, TypeError, ValueError): + pause_seconds = DEFAULT_RETENTION_DELETE_PAUSE_SECONDS + + return ( + max(1, min(batch_size, _MAX_RETENTION_DELETE_BATCH_SIZE)), + max(0.0, min(pause_seconds, _MAX_RETENTION_DELETE_PAUSE_SECONDS)), + ) + + +def delete_timestamp_rows_in_chunks( + connection_factory: Callable[[], AbstractContextManager[sqlite3.Connection]], + table: str, + timestamp_column: str, + cutoff: Any, + *, + batch_size: int = DEFAULT_RETENTION_DELETE_BATCH_SIZE, + pause_seconds: float = DEFAULT_RETENTION_DELETE_PAUSE_SECONDS, + future_cutoff: Any | None = None, + logger: Any | None = None, + progress_label: str | None = None, +) -> int: + """Delete timestamped rows in short committed transactions. + + A fresh connection is acquired for every batch. Each commit releases + SQLite's writer lock, and the optional pause gives live packet/contact/graph + writers an opportunity to proceed before retention acquires it again. + """ + for value, kind in ((table, "table"), (timestamp_column, "column")): + if not _VALID_IDENTIFIER.fullmatch(value): + raise ValueError(f"Invalid retention {kind} identifier: {value!r}") + if batch_size < 1: + raise ValueError("Retention delete batch_size must be at least 1") + if pause_seconds < 0: + raise ValueError("Retention delete pause_seconds cannot be negative") + + table_sql = f'"{table}"' + column_sql = f'"{timestamp_column}"' + where_sql = f"{column_sql} < ?" + comparison_params: tuple[Any, ...] = (cutoff,) + if future_cutoff is not None: + where_sql += f" OR {column_sql} > ?" + comparison_params += (future_cutoff,) + + delete_sql = ( + f"DELETE FROM {table_sql} WHERE rowid IN (" # noqa: S608 - identifiers validated above + f"SELECT rowid FROM {table_sql} WHERE {where_sql} LIMIT ?)" + ) + total_deleted = 0 + completed_batches = 0 + label = progress_label or table + + while True: + with connection_factory() as conn: + cursor = conn.cursor() + cursor.execute(delete_sql, (*comparison_params, batch_size)) + deleted = max(0, cursor.rowcount) + conn.commit() + + total_deleted += deleted + completed_batches += 1 + if deleted < batch_size: + break + + if logger is not None and completed_batches % 10 == 0: + logger.info( + "Retention cleanup progress for %s: %d rows deleted", + label, + total_deleted, + ) + if pause_seconds: + time.sleep(pause_seconds) + + return total_deleted diff --git a/modules/mesh_graph.py b/modules/mesh_graph.py index 7b49805..ce1a599 100644 --- a/modules/mesh_graph.py +++ b/modules/mesh_graph.py @@ -951,9 +951,12 @@ class MeshGraph: if days <= 0: return 0 try: - deleted = self.db_manager.execute_update( - "DELETE FROM mesh_connections WHERE last_seen < datetime('now', ?)", - (f'-{days} days',) + cutoff = (datetime.now() - timedelta(days=days)).isoformat() + deleted = self.db_manager.delete_timestamp_rows_in_chunks( + 'mesh_connections', + 'last_seen', + cutoff, + progress_label='mesh connections', ) if deleted > 0: self.logger.info(f"Cleaned up {deleted} old mesh_connections entries (older than {days} days)") diff --git a/modules/repeater_manager.py b/modules/repeater_manager.py index cf19da4..4481e7e 100644 --- a/modules/repeater_manager.py +++ b/modules/repeater_manager.py @@ -3248,9 +3248,11 @@ class RepeaterManager: cutoff_date = datetime.now() - timedelta(days=days_to_keep_logs) - deleted_count = self.db_manager.execute_update( - 'DELETE FROM purging_log WHERE timestamp < ?', - (cutoff_date.isoformat(),) + deleted_count = self.db_manager.delete_timestamp_rows_in_chunks( + 'purging_log', + 'timestamp', + cutoff_date.isoformat(), + progress_label='purging log', ) if deleted_count > 0: @@ -3271,17 +3273,21 @@ class RepeaterManager: # daily_stats and unique_advert_packets use date column cutoff_date = (datetime.now() - timedelta(days=daily_stats_days)).date().isoformat() - n = self.db_manager.execute_update( - 'DELETE FROM daily_stats WHERE date < ?', - (cutoff_date,) + n = self.db_manager.delete_timestamp_rows_in_chunks( + 'daily_stats', + 'date', + cutoff_date, + progress_label='daily stats', ) if n > 0: self.logger.info(f"Cleaned up {n} old daily_stats entries (older than {daily_stats_days} days)") total_deleted += n - n = self.db_manager.execute_update( - 'DELETE FROM unique_advert_packets WHERE date < ?', - (cutoff_date,) + n = self.db_manager.delete_timestamp_rows_in_chunks( + 'unique_advert_packets', + 'date', + cutoff_date, + progress_label='unique advert packets', ) if n > 0: self.logger.info(f"Cleaned up {n} old unique_advert_packets entries (older than {daily_stats_days} days)") @@ -3289,9 +3295,11 @@ class RepeaterManager: # observed_paths uses last_seen (timestamp) cutoff_ts = (datetime.now() - timedelta(days=observed_paths_days)).isoformat() - n = self.db_manager.execute_update( - 'DELETE FROM observed_paths WHERE last_seen < ?', - (cutoff_ts,) + n = self.db_manager.delete_timestamp_rows_in_chunks( + 'observed_paths', + 'last_seen', + cutoff_ts, + progress_label='observed paths', ) if n > 0: self.logger.info(f"Cleaned up {n} old observed_paths entries (older than {observed_paths_days} days)") diff --git a/modules/web_viewer/app.py b/modules/web_viewer/app.py index 1402728..6ae543b 100644 --- a/modules/web_viewer/app.py +++ b/modules/web_viewer/app.py @@ -49,6 +49,10 @@ from modules.database_restore import ( DatabaseRestoreError, stage_database_restore, ) +from modules.db_retention import ( + delete_timestamp_rows_in_chunks, + retention_delete_settings, +) from modules.ini_writer import IniValueError, update_ini_values from modules.security_utils import ( VALID_JOURNAL_MODES, @@ -1926,28 +1930,16 @@ class BotDataViewer: Returns: {"deleted": {: , ...}} — only tables that were purged """ _VALID_KEEP_DAYS = {"all", 1, 7, 14, 30, 60, 90} - # (table, sql, params) — tables created lazily by other modules may not exist + # (table, timestamp column) — some tables are created lazily. _purge_ops = [ - ('packet_stream', - 'DELETE FROM packet_stream WHERE timestamp < ?', - None), - ('message_stats', - 'DELETE FROM message_stats WHERE timestamp < ?', - None), - ('complete_contact_tracking', - 'DELETE FROM complete_contact_tracking WHERE last_heard < ?', - None), - ('purging_log', - 'DELETE FROM purging_log WHERE timestamp < ?', - None), - ('mesh_connections', - 'DELETE FROM mesh_connections WHERE last_seen < ?', - None), - ('daily_stats', - 'DELETE FROM daily_stats WHERE date < ?', - None), + ('packet_stream', 'timestamp'), + ('message_stats', 'timestamp'), + ('complete_contact_tracking', 'last_heard'), + ('purging_log', 'timestamp'), + ('mesh_connections', 'last_seen'), + ('daily_stats', 'date'), ] - _PURGEABLE = {t for t, _, _ in _purge_ops} + _PURGEABLE = {table for table, _ in _purge_ops} try: data = request.get_json(silent=True) or {} raw = data.get('keep_days', 'all') @@ -2005,24 +1997,30 @@ class BotDataViewer: } if tables_filter is None: - ops_to_run = [(t, sql, _params_for[t]) for t, sql, _ in _purge_ops] + ops_to_run = [ + (table, column, _params_for[table][0]) + for table, column in _purge_ops + ] else: want = set(tables_filter) ops_to_run = [ - (t, sql, _params_for[t]) - for t, sql, _ in _purge_ops - if t in want + (table, column, _params_for[table][0]) + for table, column in _purge_ops + if table in want ] - with self.db_manager.connection() as conn: - cur = conn.cursor() - for tbl, sql, params in ops_to_run: - try: - cur.execute(sql, params) - deleted[tbl] = cur.rowcount - except Exception: - deleted[tbl] = 0 - conn.commit() + for table, column, cutoff in ops_to_run: + try: + deleted[table] = ( + self.db_manager.delete_timestamp_rows_in_chunks( + table, + column, + cutoff, + progress_label=f'manual {table.replace("_", " ")} purge', + ) + ) + except Exception: + deleted[table] = 0 total = sum(deleted.values()) self.logger.info( @@ -4916,7 +4914,6 @@ class BotDataViewer: Uses [Data_Retention] packet_stream_retention_days when days_to_keep is not provided.""" try: import sqlite3 - import time if days_to_keep is None: days_to_keep = 3 @@ -4925,38 +4922,22 @@ class BotDataViewer: days_to_keep = self.config.getint('Data_Retention', 'packet_stream_retention_days') cutoff_time = time.time() - (days_to_keep * 24 * 60 * 60) - - # Use DEFERRED isolation; longer timeout to wait out bot writes - with closing(sqlite3.connect(self.db_path, timeout=60, isolation_level='DEFERRED')) as conn: - cursor = conn.cursor() - - # Use WAL mode for better concurrent access (if not already set) - try: - cursor.execute('PRAGMA journal_mode=WAL') - except sqlite3.OperationalError: - pass # Ignore if database is locked - WAL may already be set - - # Delete in smaller batches to avoid long locks - batch_size = 1000 - total_deleted = 0 - - while True: - cursor.execute( - 'DELETE FROM packet_stream WHERE id IN ' - '(SELECT id FROM packet_stream WHERE timestamp < ? LIMIT ?)', - (cutoff_time, batch_size) - ) - deleted_count = cursor.rowcount - conn.commit() - - if deleted_count == 0: - break - total_deleted += deleted_count - if deleted_count == batch_size: - time.sleep(0.1) - - if total_deleted > 0: - self.logger.info(f"Cleaned up {total_deleted} old packet stream entries (older than {days_to_keep} days)") + batch_size, pause_seconds = retention_delete_settings(self.config) + total_deleted = delete_timestamp_rows_in_chunks( + self._with_db_connection, + 'packet_stream', + 'timestamp', + cutoff_time, + batch_size=batch_size, + pause_seconds=pause_seconds, + logger=self.logger, + progress_label='packet stream', + ) + if total_deleted > 0: + self.logger.info( + f"Cleaned up {total_deleted} old packet stream entries " + f"(older than {days_to_keep} days)" + ) except sqlite3.OperationalError as e: self.logger.warning(f"Database busy during cleanup (will retry next cycle): {e}") diff --git a/modules/web_viewer/integration.py b/modules/web_viewer/integration.py index 98fcfb3..dd0ca2b 100644 --- a/modules/web_viewer/integration.py +++ b/modules/web_viewer/integration.py @@ -12,10 +12,14 @@ import subprocess import sys import threading import time -from contextlib import closing, suppress +from contextlib import closing, contextmanager, suppress from pathlib import Path from typing import Optional +from ..db_retention import ( + delete_timestamp_rows_in_chunks, + retention_delete_settings, +) from ..utils import resolve_path @@ -599,7 +603,6 @@ class BotIntegration: Uses [Data_Retention] packet_stream_retention_days when days_to_keep is not provided.""" try: import sqlite3 - import time if days_to_keep is None: days_to_keep = 3 @@ -610,14 +613,31 @@ class BotIntegration: cutoff_time = time.time() - (days_to_keep * 24 * 60 * 60) db_path = self._get_web_viewer_db_path() - with closing(sqlite3.connect(str(db_path), timeout=self.sqlite_connect_timeout_sec)) as conn: - cursor = conn.cursor() + batch_size, pause_seconds = retention_delete_settings(self.bot.config) - # Clean up old packet stream data - cursor.execute('DELETE FROM packet_stream WHERE timestamp < ?', (cutoff_time,)) - deleted_count = cursor.rowcount + @contextmanager + def cleanup_connection(): + with closing( + sqlite3.connect( + str(db_path), + timeout=self.sqlite_connect_timeout_sec, + ) + ) as conn: + conn.execute( + f"PRAGMA busy_timeout={int(self.sqlite_connect_timeout_sec * 1000)}" + ) + yield conn - conn.commit() + deleted_count = delete_timestamp_rows_in_chunks( + cleanup_connection, + 'packet_stream', + 'timestamp', + cutoff_time, + batch_size=batch_size, + pause_seconds=pause_seconds, + logger=self.bot.logger, + progress_label='packet stream', + ) if deleted_count > 0: self.bot.logger.info(f"Cleaned up {deleted_count} old packet stream entries (older than {days_to_keep} days)") diff --git a/scripts/build-deb.sh b/scripts/build-deb.sh index 1f03f3c..2b39ab8 100755 --- a/scripts/build-deb.sh +++ b/scripts/build-deb.sh @@ -138,7 +138,8 @@ ReadWritePaths=/etc/meshcore-bot ReadWritePaths=/var/log/meshcore-bot ReadWritePaths=/var/lib/meshcore-bot LimitNOFILE=65536 -MemoryMax=512M +MemoryMax=1G +CPUQuota=200% StartLimitInterval=60 StartLimitBurst=3 diff --git a/tests/test_db_retention.py b/tests/test_db_retention.py new file mode 100644 index 0000000..a01e31b --- /dev/null +++ b/tests/test_db_retention.py @@ -0,0 +1,137 @@ +"""Tests for bounded SQLite retention deletes.""" + +from configparser import ConfigParser +from contextlib import closing, contextmanager +from unittest.mock import Mock + +import pytest + +from modules.db_retention import ( + delete_timestamp_rows_in_chunks, + retention_delete_settings, +) + + +def test_chunked_delete_commits_and_yields_between_batches( + tmp_path, monkeypatch +): + import sqlite3 + + db_path = tmp_path / "retention.db" + with closing(sqlite3.connect(db_path)) as conn: + conn.executescript( + """ + CREATE TABLE events (id INTEGER PRIMARY KEY, seen INTEGER NOT NULL); + CREATE TABLE live_writes (id INTEGER PRIMARY KEY); + """ + ) + conn.executemany( + "INSERT INTO events(seen) VALUES (?)", + [(1,), (2,), (3,), (4,), (5,), (100,)], + ) + conn.commit() + + opened_connections = 0 + + @contextmanager + def connection(): + nonlocal opened_connections + opened_connections += 1 + with closing(sqlite3.connect(db_path, timeout=0.1)) as conn: + yield conn + + pauses = [] + + def live_writer_during_pause(seconds): + pauses.append(seconds) + with closing(sqlite3.connect(db_path, timeout=0.1)) as conn: + conn.execute( + "INSERT INTO live_writes DEFAULT VALUES" + ) + conn.commit() + + monkeypatch.setattr( + "modules.db_retention.time.sleep", + live_writer_during_pause, + ) + + deleted = delete_timestamp_rows_in_chunks( + connection, + "events", + "seen", + 10, + batch_size=2, + pause_seconds=0.01, + ) + + assert deleted == 5 + assert opened_connections == 3 + assert pauses == [0.01, 0.01] + with closing(sqlite3.connect(db_path)) as conn: + assert conn.execute("SELECT seen FROM events").fetchall() == [(100,)] + assert conn.execute("SELECT COUNT(*) FROM live_writes").fetchone()[0] == 2 + + +def test_chunked_delete_reports_progress_every_ten_full_batches(tmp_path): + import sqlite3 + + db_path = tmp_path / "retention.db" + with closing(sqlite3.connect(db_path)) as conn: + conn.execute( + "CREATE TABLE events (id INTEGER PRIMARY KEY, seen INTEGER NOT NULL)" + ) + conn.executemany( + "INSERT INTO events(seen) VALUES (?)", + [(1,) for _ in range(21)], + ) + conn.commit() + + @contextmanager + def connection(): + with closing(sqlite3.connect(db_path)) as conn: + yield conn + + logger = Mock() + deleted = delete_timestamp_rows_in_chunks( + connection, + "events", + "seen", + 10, + batch_size=2, + pause_seconds=0, + logger=logger, + progress_label="test events", + ) + + assert deleted == 21 + logger.info.assert_called_once_with( + "Retention cleanup progress for %s: %d rows deleted", + "test events", + 20, + ) + + +@pytest.mark.parametrize( + ("table", "column"), + [ + ("events; DROP TABLE events", "seen"), + ("events", "seen OR 1=1"), + ], +) +def test_chunked_delete_rejects_invalid_identifiers(table, column): + with pytest.raises(ValueError, match="Invalid retention"): + delete_timestamp_rows_in_chunks( + Mock(), + table, + column, + 10, + ) + + +def test_retention_settings_are_configurable_and_bounded(): + config = ConfigParser() + config.add_section("Data_Retention") + config.set("Data_Retention", "retention_delete_batch_size", "25000") + config.set("Data_Retention", "retention_delete_pause_seconds", "9") + + assert retention_delete_settings(config) == (10_000, 5.0) diff --git a/tests/test_repeater_manager.py b/tests/test_repeater_manager.py index daa6bbf..c1c3367 100644 --- a/tests/test_repeater_manager.py +++ b/tests/test_repeater_manager.py @@ -346,7 +346,11 @@ class TestCleanupRepeaterRetention: def test_does_not_raise_when_db_raises(self, rm): from unittest.mock import patch as _patch - with _patch.object(rm.db_manager, "execute_update", side_effect=Exception("db error")): + with _patch.object( + rm.db_manager, + "delete_timestamp_rows_in_chunks", + side_effect=Exception("db error"), + ): rm.cleanup_repeater_retention() # Should not raise rm.logger.error.assert_called() diff --git a/tests/test_service_packaging.py b/tests/test_service_packaging.py index 12014be..27a5605 100644 --- a/tests/test_service_packaging.py +++ b/tests/test_service_packaging.py @@ -25,8 +25,16 @@ def test_service_units_do_not_make_code_tree_writable(): assert "ReadWritePaths=/var/lib/meshcore-bot" in unit assert "ReadWritePaths=/etc/meshcore-bot" in unit assert "UMask=0077" in unit + assert "MemoryMax=1G" in unit + assert "CPUQuota=200%" in unit + assert "MemoryMax=512M" not in unit + assert "CPUQuota=50%" not in unit deb_builder = (REPO_ROOT / "scripts/build-deb.sh").read_text(encoding="utf-8") + assert "MemoryMax=1G" in deb_builder + assert "CPUQuota=200%" in deb_builder + assert "MemoryMax=512M" not in deb_builder + assert "CPUQuota=50%" not in deb_builder assert 'chown -R "${SERVICE_USER}:${SERVICE_USER}" "${INSTALL_ROOT}"' not in deb_builder assert "migrate_service_layout.py" in deb_builder assert 'chmod 0600 "${BUILD_DIR}${CONF_DIR}/config.ini"' in deb_builder diff --git a/tests/test_stats_command.py b/tests/test_stats_command.py index 65f7fc5..3103e5b 100644 --- a/tests/test_stats_command.py +++ b/tests/test_stats_command.py @@ -872,18 +872,15 @@ class TestCleanupOldStats: def test_cleanup_exception_handled(self): bot = _make_bot() - - @contextmanager - def _bad_conn(): - raise Exception("DB down") - yield - - bot.db_manager.connection = _bad_conn + bot.db_manager.delete_timestamp_rows_in_chunks = Mock( + side_effect=Exception("DB down") + ) cmd = StatsCommand.__new__(StatsCommand) cmd.bot = bot cmd.logger = bot.logger # Should not raise cmd.cleanup_old_stats(7) + cmd.logger.error.assert_called() # ---------------------------------------------------------------------------