feat(logs): less slow syslog.db rotation

Dockerfile

@@ -798,6 +798,7 @@ ENV DENO_DIR="/config/cache/deno" \
     PYTHONPATH="/app" \
     PYTHONPYCACHEPREFIX="/config/cache/pycache" \
     S6_CMD_WAIT_FOR_SERVICES_MAXTIME="0" \
+    SQLITE_TMPDIR="/config/cache" \
     XDG_CACHE_HOME="/config/cache" \
     XDG_CONFIG_HOME="/config/tubesync" \
     XDG_STATE_HOME="/config/state"

config/root/etc/s6-overlay/s6-rc.d/hat-syslog-server/finish

@@ -0,0 +1,60 @@
+#!/command/with-contenv bash
+
+set -e
+cd /app
+
+# Setup cleanup framework
+_cleanup_at_exit() {
+    for target in "${CLEANUP_TARGETS[@]}"; do
+        # If file is missing, exits 0 early. If it exists, calls rm.
+        ! [[ -e "${target}" ]] || s6-setuidgid app rm -f "${target}"
+    done
+}
+declare -a CLEANUP_TARGETS=()
+trap _cleanup_at_exit EXIT
+
+DB_HIGH_SIZE='1000000'
+DB_LOW_SIZE='100000'
+
+DEST_DIR='/downloads/cache/'
+DIR="${XDG_STATE_HOME:-/config/state}/hat"
+DATABASE="${DIR}/syslog.db"
+
+if ! [[ -s "${DATABASE}" ]]; then
+    exit 0
+fi
+
+ROW_RAW="$(s6-setuidgid app sqlite3 "${DATABASE}" 'SELECT count(rowid) FROM log;')"
+
+case "${ROW_RAW}" in
+    ('('*')') ROW_COUNT="$(s6-setuidgid app python3 -c 'print( tuple('"${ROW_RAW}"')[0] )')" ;;
+    (*) ROW_COUNT="${ROW_RAW}" ;;
+esac
+
+if ! [[ "${ROW_COUNT:-0}" -gt "${DB_HIGH_SIZE}" ]]; then
+    exit 0
+fi
+
+# Allocate an anchor file on the same filesystem to isolate modification times
+s6-setuidgid app mkdir -p "${DEST_DIR}"
+ANCHOR="$(s6-setuidgid app mktemp "${DEST_DIR}/anchor.XXXXXX")"
+CLEANUP_TARGETS+=("${ANCHOR}")
+
+# Clear the sub-second/same-second timestamp collision window
+sleep 1
+
+SQLITE_TMPDIR="${DEST_DIR}" s6-setuidgid app python3 /app/manage.py archive_hat_syslog --stopped "${DATABASE}" "${DEST_DIR}" "${DB_LOW_SIZE}" &&
+    s6-setuidgid app find "${DEST_DIR}" -name "${DATABASE##*/}" -type f -newer "${ANCHOR}" -exec mv -v -T '{}' "${DATABASE}.0.new" ';' -quit &&
+    [[ -s "${DATABASE}.0.new" ]] &&
+    s6-setuidgid app mv -v -T "${DATABASE}" "${DATABASE}.0" &&
+    s6-setuidgid app mv -v -T "${DATABASE}.0.new" "${DATABASE}" &&
+    s6-setuidgid app rm -f "${ANCHOR}" || exit 1
+
+i=1
+# Loop as long as a file OR symlink exists
+while [[ -e "${DATABASE}.${i}" || -L "${DATABASE}.${i}" ]]; do
+    ((++i))
+done
+s6-setuidgid app mv -v -T "${DATABASE}.0" "${DATABASE}.${i}"
+
+exit 0

config/root/etc/s6-overlay/s6-rc.d/hat-syslog-server/run

@@ -3,6 +3,9 @@
 set -e
 cd /app
 
+DB_HIGH_SIZE='500000'
+DB_LOW_SIZE='100000'
+
 DIR="${XDG_STATE_HOME:-/config/state}/hat"
 DATABASE="${DIR}/syslog.db"
 
@@ -12,4 +15,5 @@ chmod 0700 "${DIR}"
 
 exec s6-setuidgid app \
     hat-syslog-server --log-level 'INFO' \
+        --db-high-size "${DB_HIGH_SIZE}" --db-low-size "${DB_LOW_SIZE}" \
         --db-enable-archive --db-path "${DATABASE}"

config/root/etc/s6-overlay/s6-rc.d/hat-syslog-server/timeout-finish

@@ -0,0 +1 @@
+120000

tubesync/common/management/commands/archive_hat_syslog.py

@@ -0,0 +1,309 @@
+import sqlite3
+import tempfile
+import time
+from pathlib import Path
+from django.core.management.base import BaseCommand, CommandError
+
+
+class Command(BaseCommand):
+    help = 'Copies a number of rows from a live or stopped source SQLite syslog database into a new file using safety constraints.'
+    avg_row_size_bytes: int = 768
+    batch_size: int = 1000
+    batch_sleep: float = 0.05
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            '--stopped',
+            action='store_true',
+            default=False,
+            help='Source service is stopped. Allows raw attached-db file transfers instead of safe live streaming.',
+        )
+
+        parser.add_argument(
+            '--vacuum',
+            action='store_true',
+            default=False,
+            help='Optimize the database after all the rows were added (Only applies to --stopped)',
+        )
+
+        parser.add_argument('source_db_path', type=str, help='Path to the source SQLite database file')
+
+        parser.add_argument('destination_path', type=str, help='Arbitrary destination directory path for the new directory')
+
+        parser.add_argument('limit', type=int, help='The maximum number of rows to copy from the source table')
+
+    def handle(self, *args, **options):
+        row_limit: int = options['limit']
+        if not (200 <= row_limit <= 10_000_000):
+            raise CommandError(f'Invalid row limit: {row_limit:,}. Must be between 200 and 10,000,000.')
+
+        source_path: Path = Path(options['source_db_path'])
+        try:
+            source_path = source_path.resolve(strict=True)
+        except (FileNotFoundError, OSError) as e:
+            raise CommandError(f'Source database file not found or inaccessible at: {source_path}. Error: {e}')
+
+        destination_path: Path = Path(options['destination_path']).resolve(strict=False)
+        if not destination_path.exists():
+            raise CommandError(f'Destination path is not a valid directory or does not exist: {destination_path}')
+
+        blocked_bytes: tuple[int, ...] = (39, 34, 0, 42, 63, 58, 60, 62, 124, 96, 59)
+
+        # Create the unique temporary directory INSIDE the specified arbitrary destination path
+        temp_dir: str = tempfile.mkdtemp(prefix='tmp_', dir=destination_path)
+        target_path: Path = Path(temp_dir) / source_path.name
+
+        resolved_target: str = str(target_path.resolve(strict=False))
+        if any(b in blocked_bytes for b in resolved_target.encode('utf-8')):
+            self._cleanup_temp_dir(temp_dir)
+            raise CommandError('Operation aborted! Resolved target path contains invalid filesystem or quote bytes.')
+
+        self.stdout.write(self.style.SUCCESS(f'Created temp directory: {temp_dir}'))
+        self.stdout.write(f'Target database file: {target_path}')
+
+        if options['stopped']:
+            self.stdout.write('Service stopped flag detected. Executing attached legacy migration pathway...')
+            self._handle_stopped(source_path, target_path, temp_dir, row_limit, options['vacuum'])
+        else:
+            self.stdout.write('Service active. Executing un-locked in-memory streaming loop pipeline...')
+            self._handle_live(source_path, target_path, temp_dir, row_limit)
+
+    def _cleanup_temp_file(self, path: Path | str) -> None:
+        path = Path(path)
+        if path.exists():
+            try:
+                path.unlink()
+            except OSError:
+                pass
+
+    def _cleanup_temp_dir(self, directory: Path | str) -> None:
+        directory = Path(directory)
+        try:
+            for p in directory.iterdir():
+                if p.is_dir():
+                    self._cleanup_temp_dir(p)
+                elif p.is_symlink():
+                    # symbolic links may not "exist"
+                    p.unlink()
+                else:
+                    self._cleanup_temp_file(p)
+            directory.rmdir()
+        except OSError:
+            pass
+
+    def _handle_stopped(self, source_path: Path, target_path: Path, temp_dir: str, row_limit: int, run_vacuum: bool) -> None:
+        """Legacy configuration pathway optimized for an explicitly stopped service context."""
+        conn = None
+        try:
+            # Establish connection to the new target database file
+            conn = sqlite3.connect(target_path, isolation_level=None)
+            cursor = conn.cursor()
+
+            # Set auto_vacuum on the completely blank disk database file
+            cursor.execute('PRAGMA auto_vacuum = FULL;')
+
+            # Attach the old source database strictly as read-only using URI mode
+            cursor.execute(f"ATTACH DATABASE 'file:{source_path}?mode=ro' AS old;")
+
+            # Dynamically discover the active logging table name from the attached schema
+            cursor.execute(
+                "SELECT name, sql FROM old.sqlite_schema WHERE type='table' AND name NOT LIKE 'sqlite_%'"
+            )
+            table_info = cursor.fetchone()
+            if table_info is None:
+                conn.close()
+                self._cleanup_temp_file(target_path)
+                raise CommandError('Source database does not contain any user tables.')
+
+            table_name: str
+            create_schema_sql: str
+            table_name, create_schema_sql = table_info
+
+            # Discover and save all associated indexes for deferred execution AFTER insertion
+            cursor.execute(f"SELECT sql FROM old.sqlite_schema WHERE type='index' AND tbl_name='{table_name}' AND sql IS NOT NULL;")
+            index_sqls: list[str] = [row.__getitem__(0) for row in cursor.fetchall()]
+
+            # Gather the rest of the column names cleanly, filtering out rowid metadata structures
+            cursor.execute(f"PRAGMA old.table_info('{table_name}');")
+            cols = [row.__getitem__(1) for row in cursor.fetchall() if row.__getitem__(1) != 'rowid']
+            col_string = ', '.join(cols)
+
+            # Lock down the exact starting rowid boundary for our target subset
+            cursor.execute(f'''
+                SELECT MIN(rowid), MAX(rowid) FROM (
+                    SELECT rowid FROM old.{table_name}
+                    ORDER BY rowid DESC
+                    LIMIT {row_limit}
+                )
+            ''')
+            boundary_res = cursor.fetchone()
+            if boundary_res is None or boundary_res.__getitem__(0) is None:
+                conn.close()
+                self._cleanup_temp_file(target_path)
+                raise CommandError(f"Source table '{table_name}' is empty.")
+
+            start_rowid: int = boundary_res.__getitem__(0)
+            max_rowid: int = boundary_res.__getitem__(1)
+
+            # Create a temporary staging table by attaching a specialized in-memory database
+            cursor.execute("ATTACH DATABASE ':memory:' AS mem;")
+
+            staging_table: str = 'tmp_staging_log_table'
+            modified_sql = create_schema_sql.replace(table_name, f'mem.{staging_table}', 1).replace('(', '(rowid INTEGER PRIMARY KEY, ', 1)
+            cursor.execute(modified_sql)
+
+            # Re-create the actual clean table layout matching the original structure exactly inside the main disk file
+            cursor.execute(create_schema_sql)
+
+            # Step forward chronologically from the lowest bounded rowid up to the maximum rowid
+            self.stdout.write(f"Streaming {row_limit:,} log records chronologically from '{table_name}' in batches of {self.batch_size}...")
+
+            current_chunk_start: int = start_rowid
+            while current_chunk_start <= max_rowid:
+                current_chunk_end: int = min(current_chunk_start + self.batch_size - 1, max_rowid)
+
+                # Batch Step 1: Read a chronological chunk from the source directly into attached RAM database
+                cursor.execute(f'''
+                    INSERT INTO mem.{staging_table} (rowid, {col_string})
+                    SELECT rowid, {col_string} FROM old.{table_name}
+                    WHERE rowid BETWEEN ? AND ?
+                    ORDER BY rowid ASC
+                ''', (current_chunk_start, current_chunk_end))
+
+                # Batch Step 2: Flush from the attached RAM database directly into the clean main disk table
+                cursor.execute(f'''
+                    INSERT INTO main.{table_name} (rowid, {col_string})
+                    SELECT rowid, {col_string} FROM mem.{staging_table}
+                    ORDER BY rowid ASC
+                ''')
+
+                # Batch Step 3: Clear the memory staging table completely for the next iteration pass
+                cursor.execute(f'DELETE FROM mem.{staging_table}')
+
+                current_chunk_start += self.batch_size
+
+            # Success: Close files handles and drop the source database link as early as possible
+            cursor.execute('DETACH DATABASE old;')
+
+            # Securely drop and detach the memory container from the operational workspace
+            cursor.execute(f'DROP TABLE mem.{staging_table}')
+            cursor.execute('DETACH DATABASE mem;')
+
+            # Performance Win: Rebuild all indexes in a single sequential pass now that clean data is loaded
+            if index_sqls:
+                self.stdout.write('Rebuilding index structures...')
+                for index_sql in index_sqls:
+                    cursor.execute(index_sql)
+
+            conn.commit()
+            if run_vacuum:
+                self.stdout.write('Optimizing database structure...')
+                cursor.execute('VACUUM;')
+            conn.close()
+            conn = None
+
+            self.stdout.write(self.style.SUCCESS(f'Successfully archived database to: {target_path}'))
+
+        except sqlite3.Error as e:
+            raise CommandError(f'SQLite error occurred during migration: {e}')
+        finally:
+            if conn:
+                conn.close()
+                self._cleanup_temp_file(target_path)
+            if not target_path.exists():
+                self._cleanup_temp_dir(temp_dir)
+
+    def _handle_live(self, source_path: Path, target_path: Path, temp_dir: str, rows_limit: int) -> None:
+        """Safe execution path optimized for a live, un-locked running hat-syslog-server environment."""
+        src_conn = sqlite3.connect(source_path, isolation_level=None)
+        src_cursor = src_conn.cursor()
+
+        calculated_kib: int = (max(rows_limit, self.batch_size) * self.avg_row_size_bytes) // 1024
+        src_cursor.execute(f'PRAGMA cache_size = -{calculated_kib}')
+
+        src_cursor.execute("SELECT name, sql FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'")
+        table_info = src_cursor.fetchone()
+        if table_info is None:
+            src_conn.close()
+            self._cleanup_temp_dir(temp_dir)
+            raise CommandError('Source database does not contain any user tables.')
+
+        table_name: str
+        create_schema_sql: str
+        table_name, create_schema_sql = table_info
+
+        src_cursor.execute(f"SELECT sql FROM sqlite_master WHERE type='index' AND tbl_name='{table_name}' AND sql IS NOT NULL")
+        index_sqls: list[str] = [row.__getitem__(0) for row in src_cursor.fetchall()]
+
+        src_cursor.execute(f'SELECT MAX(rowid) FROM {table_name}')
+        res = src_cursor.fetchone()
+
+        if res is None or (max_rowid := res.__getitem__(0)) is None:
+            src_conn.close()
+            self._cleanup_temp_dir(temp_dir)
+            raise CommandError(f"Source table '{table_name}' is empty.")
+
+        min_needed_rowid: int = 1 + max_rowid - rows_limit
+
+        mem_conn = sqlite3.connect(':memory:', isolation_level=None)
+        mem_cursor = mem_conn.cursor()
+
+        staging_table: str = 'tmp_staging_log_table'
+        modified_schema_sql: str = create_schema_sql.replace(table_name, staging_table, 1).replace('(', '(rowid INTEGER PRIMARY KEY,', 1)
+        mem_cursor.execute(modified_schema_sql)
+
+        current_start_rowid: int = min_needed_rowid
+        total_copied: int = 0
+
+        try:
+            while current_start_rowid <= max_rowid:
+                current_end_rowid: int = min(current_start_rowid + self.batch_size - 1, max_rowid)
+
+                src_cursor.execute(f'SELECT rowid, * FROM {table_name} WHERE rowid BETWEEN ? AND ?', (current_start_rowid, current_end_rowid))
+                rows = src_cursor.fetchall()
+
+                if rows:
+                    mem_cursor.execute(f'SELECT COUNT(*) FROM {staging_table}')
+                    pre_count: int = mem_cursor.fetchone().__getitem__(0)
+
+                    placeholders: str = ','.join(['?'] * len(rows.__getitem__(0)))
+                    mem_cursor.executemany(f'INSERT INTO {staging_table} VALUES ({placeholders})', rows)
+
+                    mem_cursor.execute(f'SELECT COUNT(*) FROM {staging_table}')
+                    post_count: int = mem_cursor.fetchone().__getitem__(0)
+
+                    inserted_in_batch: int = post_count - pre_count
+                    if not (inserted_in_batch == len(rows)):
+                        src_conn.close()
+                        mem_conn.close()
+                        self._cleanup_temp_dir(temp_dir)
+                        raise CommandError(f'Data insertion mismatch! Expected {len(rows)} inserts, but only {inserted_in_batch} committed.')
+
+                    total_copied += inserted_in_batch
+
+                current_start_rowid += self.batch_size
+                time.sleep(self.batch_sleep)
+
+        finally:
+            src_conn.close()
+
+        if 0 < total_copied:
+            mem_cursor.execute(create_schema_sql)
+
+            mem_cursor.execute(f"SELECT name FROM pragma_table_info('{table_name}')")
+            col_list: str = ','.join([row.__getitem__(0) for row in mem_cursor.fetchall()])
+
+            mem_cursor.execute(f'INSERT INTO {table_name} (rowid, {col_list}) SELECT rowid, {col_list} FROM {staging_table}')
+            mem_cursor.execute(f'DROP TABLE {staging_table}')
+
+            for index_sql in index_sqls:
+                mem_cursor.execute(index_sql)
+
+        if 0 < total_copied:
+            target_path.parent.mkdir(parents=True, exist_ok=True)
+            mem_cursor.execute(f"VACUUM main INTO '{str(target_path.resolve(strict=False))}'")
+            self.stdout.write(self.style.SUCCESS(f'Successfully streamed database archive to: {target_path}'))
+        else:
+            self.stdout.write('No rows matched within the live boundaries; target archive empty.')
+
+        mem_conn.close()