Fix sqlite3 "database is locked" in pipelined shards cache

conda_libmamba_solver/shards_cache.py

@@ -49,9 +49,15 @@ def connect(dburi="cache.db"):
 
     dburi: uri-style sqlite database filename; accepts certain ?= parameters.
     """
-    conn = sqlite3.connect(dburi, uri=True)
+    conn = sqlite3.connect(dburi, uri=True, timeout=30.0)
     conn.row_factory = sqlite3.Row
     with conn as c:
+        try:
+            mode = c.execute("PRAGMA journal_mode = WAL").fetchone()[0]
+        except sqlite3.DatabaseError:
+            mode = None
+        if mode and mode.lower().startswith("wal"):
+            c.execute("PRAGMA synchronous = NORMAL")
         c.execute("PRAGMA foreign_keys = ON")
     return conn
 

news/924-fix-shards-db-locked

@@ -0,0 +1,23 @@
+### Enhancements
+
+* <news item>
+
+### Bug fixes
+
+* Open the sharded repodata cache (`repodata_shards.db`) in WAL mode with a
+  longer SQLite busy timeout, so the pipelined cache reader thread no longer
+  races with the network writer thread and raises
+  `sqlite3.OperationalError: database is locked`. Fall back to the default
+  journal mode on filesystems where WAL is not supported. (#924)
+
+### Deprecations
+
+* <news item>
+
+### Docs
+
+* <news item>
+
+### Other
+
+* <news item>

tests/test_shards.py

@@ -13,6 +13,7 @@
 import os
 import sqlite3
 import tempfile
+import threading
 import time
 from contextlib import contextmanager
 from pathlib import Path
@@ -633,6 +634,58 @@ def test_shards_cache_recovery(tmp_path: Path):
     assert cache.retrieve("notfound") is None
 
 
+def test_shards_cache_uses_wal(tmp_path: Path):
+    """WAL journal mode is enabled on a fresh cache."""
+    with shards_cache.ShardCache(tmp_path) as cache:
+        mode = cache.conn.execute("PRAGMA journal_mode").fetchone()[0]
+    assert mode == "wal"
+
+
+def test_shards_cache_concurrent_read_write(tmp_path: Path):
+    """Concurrent readers and writers must not raise OperationalError (#924)."""
+    compressor = zstandard.ZstdCompressor(level=1)
+    errors: list[Exception] = []
+    stop = threading.Event()
+
+    def writer(base):
+        try:
+            with shards_cache.ShardCache(base, create=False) as cache_copy:
+                for i in range(200):
+                    if stop.is_set():
+                        break
+                    shard = shards_cache.AnnotatedRawShard(
+                        f"https://shard{i}",
+                        f"pkg{i}",
+                        compressor.compress(msgpack.dumps({f"pkg{i}": "data"})),
+                    )
+                    cache_copy.insert(shard)
+        except Exception as exc:
+            errors.append(exc)
+
+    def reader(base):
+        try:
+            with shards_cache.ShardCache(base, create=False) as cache_copy:
+                for i in range(200):
+                    if stop.is_set():
+                        break
+                    urls = [f"https://shard{j}" for j in range(i + 1)]
+                    cache_copy.retrieve_multiple(urls)
+        except Exception as exc:
+            errors.append(exc)
+
+    with shards_cache.ShardCache(tmp_path) as cache:
+        w = threading.Thread(target=writer, args=(cache.base,))
+        r = threading.Thread(target=reader, args=(cache.base,))
+        w.start()
+        r.start()
+        w.join(timeout=10)
+        r.join(timeout=10)
+        stop.set()
+
+    # No sqlite3.OperationalError from either thread
+    assert errors == []
+
+
 NUM_FAKE_SHARDS = 64