fix: add resilience to manifest DB writes

Add retry with backoff for transient 'database is locked' errors in
upsert(), preventing a single lock failure from cascading into total
manifest write failure for the rest of the run.

Also add error handling to flush(), close(), and update_path() which
were previously unprotected against SQLite errors. Fix missing
_pending_count increment in update_path().

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: teh-hippo

src/icloudpd/manifest.py

@@ -13,6 +13,7 @@
 import logging
 import os
 import sqlite3
+import time
 from dataclasses import dataclass
 from datetime import datetime, timezone
 
@@ -133,6 +134,9 @@ class ManifestRow:
 class ManifestDB:
     """SQLite-backed asset manifest for tracking downloaded files."""
 
+    _MAX_WRITE_RETRIES = 3
+    _RETRY_BASE_DELAY = 0.1
+
     def __init__(self, download_dir: str) -> None:
         self._db_path = os.path.join(download_dir, ".icloudpd.db")
         self._conn: sqlite3.Connection | None = None
@@ -229,7 +233,13 @@ def close(self) -> None:
         """Close the manifest DB, committing any pending writes."""
         if self._conn:
             if self._dirty:
-                self._conn.commit()
+                try:
+                    self._conn.commit()
+                except sqlite3.OperationalError as e:
+                    if "locked" in str(e):
+                        logger.warning("Manifest commit on close failed: %s", e)
+                    else:
+                        raise
                 self._dirty = False
                 self._pending_count = 0
             self._conn.close()
@@ -238,9 +248,15 @@ def close(self) -> None:
     def flush(self) -> None:
         """Commit pending writes without closing."""
         if self._conn and self._dirty:
-            self._conn.commit()
-            self._dirty = False
-            self._pending_count = 0
+            try:
+                self._conn.commit()
+                self._dirty = False
+                self._pending_count = 0
+            except sqlite3.OperationalError as e:
+                if "locked" in str(e):
+                    logger.warning("Manifest flush failed: %s", e)
+                else:
+                    raise
 
     def __enter__(self) -> "ManifestDB":
         self.open()
@@ -308,69 +324,90 @@ def upsert(
         raw_fields: str | None = None,
     ) -> None:
         """Insert or update a manifest entry. Auto-flushes every 500 writes."""
+        now = datetime.now(tz=timezone.utc).isoformat()
+        params = (
+            asset_id, zone_id, local_path, version_size, version_checksum,
+            change_tag, now, now, item_type, filename,
+            asset_date, added_date, is_favorite, is_hidden, is_deleted,
+            original_width, original_height, duration, orientation,
+            title, description, keywords, gps_latitude, gps_longitude, gps_altitude,
+            gps_speed, gps_timestamp, timezone_offset, asset_subtype, hdr_type,
+            burst_flags, burst_flags_ext, burst_id, original_orientation, raw_fields,
+        )
+        sql = (
+            f"INSERT INTO manifest ({_ALL_COLUMNS}) "
+            "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) "
+            "ON CONFLICT(asset_id, zone_id, local_path) DO UPDATE SET "
+            "version_size=excluded.version_size, "
+            "version_checksum=excluded.version_checksum, "
+            "change_tag=excluded.change_tag, "
+            "last_updated_at=excluded.last_updated_at, "
+            "item_type=excluded.item_type, "
+            "filename=excluded.filename, "
+            "asset_date=excluded.asset_date, "
+            "added_date=excluded.added_date, "
+            "is_favorite=excluded.is_favorite, "
+            "is_hidden=excluded.is_hidden, "
+            "is_deleted=excluded.is_deleted, "
+            "original_width=excluded.original_width, "
+            "original_height=excluded.original_height, "
+            "duration=excluded.duration, "
+            "orientation=excluded.orientation, "
+            "title=excluded.title, "
+            "description=excluded.description, "
+            "keywords=excluded.keywords, "
+            "gps_latitude=excluded.gps_latitude, "
+            "gps_longitude=excluded.gps_longitude, "
+            "gps_altitude=excluded.gps_altitude, "
+            "gps_speed=excluded.gps_speed, "
+            "gps_timestamp=excluded.gps_timestamp, "
+            "timezone_offset=excluded.timezone_offset, "
+            "asset_subtype=excluded.asset_subtype, "
+            "hdr_type=excluded.hdr_type, "
+            "burst_flags=excluded.burst_flags, "
+            "burst_flags_ext=excluded.burst_flags_ext, "
+            "burst_id=excluded.burst_id, "
+            "original_orientation=excluded.original_orientation, "
+            "raw_fields=excluded.raw_fields"
+        )
+        for attempt in range(self._MAX_WRITE_RETRIES):
+            try:
+                self._db.execute(sql, params)
+                self._dirty = True
+                self._pending_count += 1
+                if self._pending_count >= self._flush_interval:
+                    self.flush()
+                return
+            except sqlite3.OperationalError as e:
+                if "locked" in str(e) and attempt < self._MAX_WRITE_RETRIES - 1:
+                    logger.debug(
+                        "Manifest write retry %d for %s: %s",
+                        attempt + 1, local_path, e,
+                    )
+                    time.sleep(self._RETRY_BASE_DELAY * (attempt + 1))
+                else:
+                    logger.warning("Manifest write failed for %s: %s", local_path, e)
+                    return
+            except sqlite3.Error as e:
+                logger.warning("Manifest write failed for %s: %s", local_path, e)
+                return
+
+    def update_path(self, asset_id: str, zone_id: str, old_path: str, new_path: str) -> None:
+        """Update local_path for an existing manifest entry."""
         try:
-            now = datetime.now(tz=timezone.utc).isoformat()
             self._db.execute(
-                f"INSERT INTO manifest ({_ALL_COLUMNS}) "
-                "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) "
-                "ON CONFLICT(asset_id, zone_id, local_path) DO UPDATE SET "
-                "version_size=excluded.version_size, "
-                "version_checksum=excluded.version_checksum, "
-                "change_tag=excluded.change_tag, "
-                "last_updated_at=excluded.last_updated_at, "
-                "item_type=excluded.item_type, "
-                "filename=excluded.filename, "
-                "asset_date=excluded.asset_date, "
-                "added_date=excluded.added_date, "
-                "is_favorite=excluded.is_favorite, "
-                "is_hidden=excluded.is_hidden, "
-                "is_deleted=excluded.is_deleted, "
-                "original_width=excluded.original_width, "
-                "original_height=excluded.original_height, "
-                "duration=excluded.duration, "
-                "orientation=excluded.orientation, "
-                "title=excluded.title, "
-                "description=excluded.description, "
-                "keywords=excluded.keywords, "
-                "gps_latitude=excluded.gps_latitude, "
-                "gps_longitude=excluded.gps_longitude, "
-                "gps_altitude=excluded.gps_altitude, "
-                "gps_speed=excluded.gps_speed, "
-                "gps_timestamp=excluded.gps_timestamp, "
-                "timezone_offset=excluded.timezone_offset, "
-                "asset_subtype=excluded.asset_subtype, "
-                "hdr_type=excluded.hdr_type, "
-                "burst_flags=excluded.burst_flags, "
-                "burst_flags_ext=excluded.burst_flags_ext, "
-                "burst_id=excluded.burst_id, "
-                "original_orientation=excluded.original_orientation, "
-                "raw_fields=excluded.raw_fields",
-                (
-                    asset_id, zone_id, local_path, version_size, version_checksum,
-                    change_tag, now, now, item_type, filename,
-                    asset_date, added_date, is_favorite, is_hidden, is_deleted,
-                    original_width, original_height, duration, orientation,
-                    title, description, keywords, gps_latitude, gps_longitude, gps_altitude,
-                    gps_speed, gps_timestamp, timezone_offset, asset_subtype, hdr_type,
-                    burst_flags, burst_flags_ext, burst_id, original_orientation, raw_fields,
-                ),
+                "UPDATE manifest SET local_path = ?, last_updated_at = ? "
+                "WHERE asset_id = ? AND zone_id = ? AND local_path = ?",
+                (new_path, datetime.now(tz=timezone.utc).isoformat(),
+                 asset_id, zone_id, old_path),
             )
             self._dirty = True
             self._pending_count += 1
-            if self._pending_count >= self._flush_interval:
-                self.flush()
         except sqlite3.Error as e:
-            logger.warning("Manifest write failed for %s: %s", local_path, e)
-
-    def update_path(self, asset_id: str, zone_id: str, old_path: str, new_path: str) -> None:
-        """Update local_path for an existing manifest entry."""
-        self._db.execute(
-            "UPDATE manifest SET local_path = ?, last_updated_at = ? "
-            "WHERE asset_id = ? AND zone_id = ? AND local_path = ?",
-            (new_path, datetime.now(tz=timezone.utc).isoformat(),
-             asset_id, zone_id, old_path),
-        )
-        self._dirty = True
+            logger.warning(
+                "Manifest path update failed for %s -> %s: %s",
+                old_path, new_path, e,
+            )
 
     def count_by_path(self, local_path: str) -> int:
         """Count how many manifest entries reference a given local path."""