diff --git a/modules/web_viewer/app.py b/modules/web_viewer/app.py index 59fd497..ddd4305 100644 --- a/modules/web_viewer/app.py +++ b/modules/web_viewer/app.py @@ -286,7 +286,7 @@ class BotDataViewer: """Initialize the packet_stream table in the web viewer database (same as [Bot] db_path by default).""" conn = None try: - with sqlite3.connect(self.db_path, timeout=30) as conn: + with sqlite3.connect(self.db_path, timeout=60) as conn: cursor = conn.cursor() # Create packet_stream table with schema matching the INSERT statements @@ -311,6 +311,12 @@ class BotDataViewer: ON packet_stream(type) ''') + # Enable WAL for better concurrent access (bot + web viewer use same DB) + try: + cursor.execute('PRAGMA journal_mode=WAL') + except sqlite3.OperationalError: + pass # Ignore if locked; WAL may already be set + conn.commit() self.logger.info(f"Initialized packet_stream table in {self.db_path}") @@ -322,7 +328,7 @@ class BotDataViewer: def _get_db_connection(self): """Get database connection - create new connection for each request to avoid threading issues""" try: - conn = sqlite3.connect(self.db_path, timeout=30) + conn = sqlite3.connect(self.db_path, timeout=60) conn.row_factory = sqlite3.Row return conn except Exception as e: @@ -1568,7 +1574,7 @@ class BotDataViewer: # Get commands from last 60 minutes cutoff_time = time.time() - (60 * 60) # 60 minutes ago - with sqlite3.connect(self.db_path, timeout=30) as conn: + with sqlite3.connect(self.db_path, timeout=60) as conn: cursor = conn.cursor() cursor.execute(''' @@ -2701,7 +2707,7 @@ class BotDataViewer: # Connect to database with timeout to prevent hanging # Use check_same_thread=False for thread safety, but be careful try: - conn = sqlite3.connect(self.db_path, timeout=30, check_same_thread=False) + conn = sqlite3.connect(self.db_path, timeout=60, check_same_thread=False) conn.row_factory = sqlite3.Row except sqlite3.OperationalError as conn_error: error_msg = str(conn_error) @@ -2754,8 +2760,8 @@ class BotDataViewer: finally: conn.close() - # Sleep before next poll - time.sleep(0.5) # Poll every 500ms + # Sleep before next poll (back off to reduce lock contention with bot writes) + time.sleep(2.0) # Poll every 2s except sqlite3.OperationalError as e: consecutive_errors += 1 @@ -2859,8 +2865,8 @@ class BotDataViewer: cutoff_time = time.time() - (days_to_keep * 24 * 60 * 60) - # Use shorter timeout and isolation_level for better concurrency - conn = sqlite3.connect(self.db_path, timeout=10, isolation_level='DEFERRED') + # Use DEFERRED isolation; longer timeout to wait out bot writes + conn = sqlite3.connect(self.db_path, timeout=60, isolation_level='DEFERRED') cursor = conn.cursor() # Use WAL mode for better concurrent access (if not already set) diff --git a/modules/web_viewer/integration.py b/modules/web_viewer/integration.py index bf01468..9acb6d1 100644 --- a/modules/web_viewer/integration.py +++ b/modules/web_viewer/integration.py @@ -126,7 +126,7 @@ class BotIntegration: db_path = self._get_web_viewer_db_path() # Connect to database and create table if it doesn't exist - conn = sqlite3.connect(str(db_path), timeout=30.0) + conn = sqlite3.connect(str(db_path), timeout=60.0) cursor = conn.cursor() # Create packet_stream table with schema matching the INSERT statements @@ -151,6 +151,12 @@ class BotIntegration: ON packet_stream(type) ''') + # Enable WAL for better concurrent access (bot + web viewer use same DB) + try: + cursor.execute('PRAGMA journal_mode=WAL') + except sqlite3.OperationalError: + pass # Ignore if locked; WAL may already be set + conn.commit() conn.close() @@ -161,12 +167,43 @@ class BotIntegration: # Don't raise - allow bot to continue even if table init fails # The error will be caught when trying to insert data + def _insert_packet_stream_row(self, data_json: str, row_type: str, log_prefix: str = "packet data"): + """Insert one row into packet_stream. Retries on database is locked. Logs and returns on failure.""" + import sqlite3 + import time + db_path = self._get_web_viewer_db_path() + max_retries = 3 + for attempt in range(max_retries): + conn = None + try: + conn = sqlite3.connect(str(db_path), timeout=60.0) + cursor = conn.cursor() + cursor.execute(''' + INSERT INTO packet_stream (timestamp, data, type) + VALUES (?, ?, ?) + ''', (time.time(), data_json, row_type)) + conn.commit() + return + except sqlite3.OperationalError as e: + if "locked" in str(e).lower() and attempt < max_retries - 1: + time.sleep(0.15 * (attempt + 1)) + continue + self.bot.logger.warning(f"Error storing {log_prefix} for web viewer: {e}") + return + except Exception as e: + self.bot.logger.warning(f"Error storing {log_prefix} for web viewer: {e}") + return + finally: + if conn is not None: + try: + conn.close() + except Exception: + pass + def capture_full_packet_data(self, packet_data): """Capture full packet data and store in database for web viewer""" try: - import sqlite3 import json - import time from datetime import datetime # Ensure packet_data is a dict (might be passed as dict already) @@ -191,22 +228,8 @@ class BotIntegration: # Convert non-serializable objects to strings serializable_data = self._make_json_serializable(packet_data) - # Store in database for web viewer to read (same path as viewer so packet log shows packets) - db_path = self._get_web_viewer_db_path() - conn = sqlite3.connect(str(db_path), timeout=30.0) - cursor = conn.cursor() - - # Insert packet data - cursor.execute(''' - INSERT INTO packet_stream (timestamp, data, type) - VALUES (?, ?, ?) - ''', (time.time(), json.dumps(serializable_data), 'packet')) - - conn.commit() - conn.close() - - # Note: Cleanup is handled by the web viewer subprocess to avoid - # database lock contention between bot and web viewer processes + # Store in database for web viewer to read (retries on database is locked) + self._insert_packet_stream_row(json.dumps(serializable_data), 'packet', "packet data") except Exception as e: self.bot.logger.warning(f"Error storing packet data for web viewer: {e}") @@ -214,7 +237,6 @@ class BotIntegration: def capture_command(self, message, command_name, response, success, command_id=None): """Capture command data and store in database for web viewer""" try: - import sqlite3 import json import time @@ -253,19 +275,8 @@ class BotIntegration: # Convert non-serializable objects to strings serializable_data = self._make_json_serializable(command_data) - # Store in database for web viewer to read - db_path = self._get_web_viewer_db_path() - conn = sqlite3.connect(str(db_path), timeout=30.0) - cursor = conn.cursor() - - # Insert command data - cursor.execute(''' - INSERT INTO packet_stream (timestamp, data, type) - VALUES (?, ?, ?) - ''', (time.time(), json.dumps(serializable_data), 'command')) - - conn.commit() - conn.close() + # Store in database for web viewer to read (retries on database is locked) + self._insert_packet_stream_row(json.dumps(serializable_data), 'command', "command data") except Exception as e: self.bot.logger.debug(f"Error storing command data: {e}") @@ -273,26 +284,13 @@ class BotIntegration: def capture_packet_routing(self, routing_data): """Capture packet routing data and store in database for web viewer""" try: - import sqlite3 import json - import time # Convert non-serializable objects to strings serializable_data = self._make_json_serializable(routing_data) - # Store in database for web viewer to read - db_path = self._get_web_viewer_db_path() - conn = sqlite3.connect(str(db_path), timeout=30.0) - cursor = conn.cursor() - - # Insert routing data - cursor.execute(''' - INSERT INTO packet_stream (timestamp, data, type) - VALUES (?, ?, ?) - ''', (time.time(), json.dumps(serializable_data), 'routing')) - - conn.commit() - conn.close() + # Store in database for web viewer to read (retries on database is locked) + self._insert_packet_stream_row(json.dumps(serializable_data), 'routing', "routing data") except Exception as e: self.bot.logger.debug(f"Error storing routing data: {e}") @@ -306,7 +304,7 @@ class BotIntegration: cutoff_time = time.time() - (days_to_keep * 24 * 60 * 60) db_path = self._get_web_viewer_db_path() - conn = sqlite3.connect(str(db_path), timeout=30.0) + conn = sqlite3.connect(str(db_path), timeout=60.0) cursor = conn.cursor() # Clean up old packet stream data