diff --git a/dascore/config.py b/dascore/config.py index e53012217..92ad5b9b4 100644 --- a/dascore/config.py +++ b/dascore/config.py @@ -84,8 +84,8 @@ class DascoreConfig(BaseModel): description="Persistent directory used to cache downloaded example data.", ) directory_index_map_path: Path = Field( - default_factory=lambda: _get_cache_root() / "indexes" / "cache_paths.json", - description="Path to the cache that records external index-file locations.", + default_factory=lambda: _get_cache_root() / "indexes" / "cache_paths.sqlite3", + description="Path to the SQLite cache of external index-file locations.", ) # Progress display. diff --git a/dascore/io/index/indexer.py b/dascore/io/index/indexer.py index 219d5b3f0..0a101bce7 100644 --- a/dascore/io/index/indexer.py +++ b/dascore/io/index/indexer.py @@ -11,11 +11,11 @@ from __future__ import annotations import hashlib -import json import os -import tempfile -from contextlib import suppress +import sqlite3 +from contextlib import contextmanager, suppress from pathlib import Path +from threading import Lock import pandas as pd from typing_extensions import Self @@ -31,53 +31,274 @@ from dascore.utils.misc import _iter_filesystem from dascore.utils.paths import directory_writable, requires_local_directory +_INDEX_MAP_SCHEMA = """ +CREATE TABLE IF NOT EXISTS index_map ( + directory TEXT PRIMARY KEY, + index_path TEXT NOT NULL +) +""" +_INDEX_MAP_UPSERT = """ +INSERT INTO index_map (directory, index_path) +VALUES (?, ?) +ON CONFLICT(directory) DO UPDATE SET index_path = excluded.index_path +""" +_INDEX_MAP_CORRUPTION_CODES = { + getattr(sqlite3, "SQLITE_CORRUPT", 11), + getattr(sqlite3, "SQLITE_NOTADB", 26), +} +_INDEX_MAP_CORRUPTION_MESSAGES = ("database disk image is malformed", "not a database") +_INDEX_MAP_RECOVERY_LOCK = Lock() -def _get_index_map(cache_path) -> dict: - """ - Return a fresh dict of index locations read from disk. - Read (not cached): another process may have updated the map, and a - per-process cache would dump this process's stale copy on the next - write, erasing entries others added. - """ - path = Path(cache_path) - out = {} - successful_read = True - if path.exists(): - try: - with path.open("r") as fi: - out = json.load(fi) - # On rare occasions, the file can become corrupt. See #508. - except (OSError, json.JSONDecodeError): - successful_read = False - if not isinstance(out, dict) or not successful_read: - out = {} - with suppress(FileNotFoundError, PermissionError): - path.unlink(missing_ok=True) - return out - - -def _update_index_map(updates, cache_path) -> dict: - """Update the index map to track a new index, writing atomically.""" - data = _get_index_map(cache_path=cache_path) - data.update(updates) - path = Path(cache_path) - path.parent.mkdir(exist_ok=True, parents=True) - # Write to a sibling temp file and os.replace() it into place. A direct - # write is not atomic: a concurrent reader hitting a half-written file - # raises JSONDecodeError, which _get_index_map treats as corruption and - # deletes the whole map (see #508). replace() is atomic on the same - # filesystem, so readers only ever see a complete file. - fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=path.name, suffix=".tmp") +def _acquire_index_map_recovery_lock_before_fork(): + """Wait for the recovery guard before allowing a process fork.""" + _INDEX_MAP_RECOVERY_LOCK.acquire() + + +def _release_index_map_recovery_lock_after_fork(): + """Release the parent recovery lock acquired before a process fork.""" + _INDEX_MAP_RECOVERY_LOCK.release() + + +def _reset_index_map_recovery_lock(): + """Replace thread-lock state inherited by a forked child.""" + global _INDEX_MAP_RECOVERY_LOCK + _INDEX_MAP_RECOVERY_LOCK = Lock() + + +if hasattr(os, "register_at_fork"): + os.register_at_fork( + before=_acquire_index_map_recovery_lock_before_fork, + after_in_parent=_release_index_map_recovery_lock_after_fork, + after_in_child=_reset_index_map_recovery_lock, + ) + + +def _is_corrupt_index_map_error(exc: sqlite3.DatabaseError) -> bool: + """Return whether an SQLite error means the disposable map is corrupt.""" + error_code = getattr(exc, "sqlite_errorcode", None) + if error_code is not None: + return error_code & 0xFF in _INDEX_MAP_CORRUPTION_CODES + message = str(exc).lower() + return any(part in message for part in _INDEX_MAP_CORRUPTION_MESSAGES) + + +def _open_index_map(database_path: Path) -> sqlite3.Connection: + """Open and initialize one SQLite index-map connection.""" + connection = sqlite3.connect(database_path, timeout=30, isolation_level=None) try: - with os.fdopen(fd, "w") as fi: - json.dump(data, fi) - os.replace(tmp, path) + connection.execute("PRAGMA busy_timeout = 30000") + connection.execute(_INDEX_MAP_SCHEMA) + return connection except BaseException: - with suppress(OSError): - os.unlink(tmp) + connection.close() raise - return data + + +def _open_index_map_read_only(database_path: Path) -> sqlite3.Connection: + """Open an existing SQLite index map without requiring write access.""" + database_uri = f"{database_path.absolute().as_uri()}?mode=ro" + connection = sqlite3.connect( + database_uri, + timeout=30, + isolation_level=None, + uri=True, + ) + connection.execute("PRAGMA busy_timeout = 30000") + return connection + + +def _acquire_recovery_file_lock(lock_file) -> bool: + """Acquire a content-independent process lock on an open file.""" + if os.name == "nt": # pragma: no cover + import errno + import msvcrt + import time + + lock_file.seek(0, os.SEEK_END) + if not lock_file.tell(): + lock_file.write(b"\0") + lock_file.flush() + lock_file.seek(0) + transient_errors = {errno.EACCES, errno.EAGAIN, errno.EDEADLK} + while True: + try: + msvcrt.locking(lock_file.fileno(), msvcrt.LK_NBLCK, 1) + except OSError as exc: + if exc.errno not in transient_errors: + raise + time.sleep(0.05) + else: + break + return True + else: + try: + import fcntl + except ImportError: # Emscripten can omit this Unix-only module. + return False + + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) + return True + + +def _release_recovery_file_lock(lock_file): + """Release a process lock acquired by `_acquire_recovery_file_lock`.""" + if os.name == "nt": # pragma: no cover + import msvcrt + + lock_file.seek(0) + msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1) + else: + import fcntl + + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + + +@contextmanager +def _index_map_recovery_guard(database_path: Path): + """Keep map operations out of the way of destructive recovery.""" + lock_path = database_path.with_name(f"{database_path.name}.recovery.lock") + with _INDEX_MAP_RECOVERY_LOCK, lock_path.open("a+b") as lock_file: + lock_acquired = _acquire_recovery_file_lock(lock_file) + try: + yield + finally: + if lock_acquired: + _release_recovery_file_lock(lock_file) + + +@contextmanager +def _index_map_read_only_guard(database_path: Path): + """Coordinate a read-only map access when its lock cannot be written.""" + lock_path = database_path.with_name(f"{database_path.name}.recovery.lock") + with _INDEX_MAP_RECOVERY_LOCK: + # Windows byte-range locks require a writable file handle. SQLite still + # provides a consistent read transaction when the cache is read-only. + if os.name == "nt": # pragma: no cover + yield + return + try: + lock_file = lock_path.open("rb") + except OSError: + yield + return + with lock_file: + try: + lock_acquired = _acquire_recovery_file_lock(lock_file) + except OSError: + yield + return + try: + yield + finally: + if lock_acquired: + _release_recovery_file_lock(lock_file) + + +def _index_map_is_healthy(database_path: Path) -> bool: + """Validate the complete SQLite database, including unused pages.""" + connection = None + try: + connection = _open_index_map(database_path) + result = connection.execute("PRAGMA integrity_check").fetchall() + return result == [("ok",)] + except sqlite3.DatabaseError as exc: + if not _is_corrupt_index_map_error(exc): + raise + return False + finally: + if connection is not None: + connection.close() + + +def _remove_index_map(database_path: Path) -> None: + """Remove a corrupt map and SQLite sidecars before rebuilding.""" + for suffix in ("", "-journal", "-shm", "-wal"): + Path(f"{database_path}{suffix}").unlink(missing_ok=True) + + +def _recover_index_map(database_path: Path) -> None: + """Rebuild a corrupt map while the caller holds the recovery guard.""" + # A process that held the recovery lock before us may have fixed it. + if _index_map_is_healthy(database_path): + return + _remove_index_map(database_path) + _open_index_map(database_path).close() + + +def _run_index_map_operation(cache_path, operation): + """Run one complete map operation, rebuilding and retrying on corruption.""" + database_path = Path(cache_path) + database_path.parent.mkdir(exist_ok=True, parents=True) + # Recovery removes and recreates the database, so participating access + # is deliberately exclusive, including reads. This ensures no process retains + # an open handle to the unlinked database or its name-based SQLite sidecars. + # SQLite transactions also protect against connections outside this guard. + with _index_map_recovery_guard(database_path): + for attempt in range(2): + connection = None + try: + connection = _open_index_map(database_path) + return operation(connection) + except sqlite3.DatabaseError as exc: + if attempt or not _is_corrupt_index_map_error(exc): + raise + finally: + if connection is not None: + connection.close() + _recover_index_map(database_path) + raise AssertionError("unreachable") + + +def _get_index_map(cache_path) -> dict[str, str]: + """ + Return a fresh dict of index locations read from the SQLite database. + + Read (not cached): another process may have updated the map. + """ + + def read(connection): + rows = connection.execute( + "SELECT directory, index_path FROM index_map" + ).fetchall() + return dict(rows) + + try: + return _run_index_map_operation(cache_path, read) + except OSError: + # A shared cache can expose a readable map and recovery sidecar without + # granting write access. Join its existing lock when possible, then use + # SQLite's read-only URI mode rather than discarding a valid mapping. + database_path = Path(cache_path) + with _index_map_read_only_guard(database_path): + connection = None + try: + connection = _open_index_map_read_only(database_path) + return read(connection) + except sqlite3.DatabaseError as exc: + if not _is_corrupt_index_map_error(exc): + raise + return {} + finally: + if connection is not None: + connection.close() + + +def _update_index_map(updates, cache_path) -> dict[str, str]: + """Transactionally upsert index locations without losing other writers.""" + rows = [(str(key), str(value)) for key, value in updates.items()] + + def update(connection): + with connection: + # Reserve the single writer slot before reading or updating rows. + connection.execute("BEGIN IMMEDIATE") + connection.executemany(_INDEX_MAP_UPSERT, rows) + data = connection.execute( + "SELECT directory, index_path FROM index_map" + ).fetchall() + return dict(data) + + return _run_index_map_operation(cache_path, update) class DBDirectoryIndexer: @@ -159,7 +380,11 @@ def _find_index_path(self, index_path=None) -> Path: with suppress(PermissionError): if expected.exists(): return expected - path_map = _get_index_map(cache_path=str(self.index_map_path)) + path_map = {} + # A writable data directory can fall back to its local index when the + # optional global map is unavailable, such as on a read-only cache. + with suppress(OSError, sqlite3.OperationalError): + path_map = _get_index_map(cache_path=str(self.index_map_path)) if out := path_map.get(map_key): mapped = Path(out) # Index-map entries from older DASCore versions can point at diff --git a/tests/conftest.py b/tests/conftest.py index 7d97dc9d4..9d9e67b67 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -104,7 +104,7 @@ def allow_legacy_dasdae_coord_unpickle(): @pytest.fixture(scope="session", autouse=True) def swap_index_map_path(tmp_path_factory): """For all tests cases, use a temporary index file.""" - tmp_map_path = tmp_path_factory.mktemp("cache_paths") / "cache_paths.json" + tmp_map_path = tmp_path_factory.mktemp("cache_paths") / "cache_paths.sqlite3" with set_config(directory_index_map_path=tmp_map_path): yield diff --git a/tests/test_io/test_index/test_index_edge_cases.py b/tests/test_io/test_index/test_index_edge_cases.py index 993f15ce1..b8c225eb8 100644 --- a/tests/test_io/test_index/test_index_edge_cases.py +++ b/tests/test_io/test_index/test_index_edge_cases.py @@ -1224,7 +1224,7 @@ def test_legacy_entry_ignored(self, tmp_path): legacy = tmp_path / "legacy_index.h5" with h5py.File(legacy, "w") as fh: fh.create_dataset("x", data=[1, 2, 3]) - map_path = tmp_path / "cache_paths.json" + map_path = tmp_path / "cache_paths.sqlite3" with dc.set_config(directory_index_map_path=map_path): _update_index_map({str(data_dir): str(legacy)}, cache_path=str(map_path)) spool = dc.spool(data_dir).update() diff --git a/tests/test_io/test_indexer.py b/tests/test_io/test_indexer.py index 26304f702..573980be1 100644 --- a/tests/test_io/test_indexer.py +++ b/tests/test_io/test_indexer.py @@ -2,11 +2,15 @@ from __future__ import annotations -import json +import multiprocessing import os import platform import shutil -from contextlib import suppress +import sqlite3 +import sys +import threading +from concurrent.futures import ThreadPoolExecutor +from contextlib import closing, suppress from pathlib import Path import pandas as pd @@ -16,9 +20,27 @@ from dascore.config import set_config from dascore.exceptions import InvalidSpoolError from dascore.io.index.indexer import DBDirectoryIndexer +from dascore.utils.misc import suppress_warnings from dascore.utils.patch import get_patch_names +def _update_corrupt_index_map_process(cache_path, key, ready, start): + """Update a corrupt index map in a child process.""" + from dascore.io.index.indexer import _update_index_map + + ready.set() + if not start.wait(timeout=20): + raise RuntimeError("timed out waiting to update index map") + _update_index_map({key: f"index-{key}"}, cache_path=cache_path) + + +def _read_index_map_process(cache_path): + """Read the index map in a child process.""" + from dascore.io.index.indexer import _get_index_map + + _get_index_map(cache_path) + + @pytest.fixture(scope="class") def basic_indexer(two_patch_directory): """Return an indexer on the basic spool directory.""" @@ -67,9 +89,8 @@ def unwritable_directory(self, tmp_path_factory): def directory_indexer_bad_cache(self, tmp_path_factory): """Create a bad index_map file.""" path = tmp_path_factory.mktemp("corrupt_cache_test") - cache_path = path / "corrupt_cache.json" - with cache_path.open("wt") as fi: - fi.write("{'bad': 'json'") + cache_path = path / "corrupt_cache.sqlite3" + cache_path.write_bytes(b"not a sqlite database") return cache_path def test_directory_cant_write(self, unwritable_directory): @@ -88,7 +109,7 @@ def test_read_only_index_name_is_stable(self, unwritable_directory, tmp_path): """ import hashlib - map_path = tmp_path / "cache_paths.json" + map_path = tmp_path / "cache_paths.sqlite3" with set_config(directory_index_map_path=map_path): first = DBDirectoryIndexer(unwritable_directory).index_path digest = hashlib.sha256(str(unwritable_directory).encode()).hexdigest()[:16] @@ -110,6 +131,81 @@ def test_writeable_dir_index_not_there(self, tmp_path_factory): dir_indexer = DBDirectoryIndexer(path) assert dir_indexer.index_path.parent == path + def test_writable_dir_survives_unwritable_map( + self, unwritable_directory, tmp_path_factory + ): + """Use a local index when the global map directory is read-only.""" + data_path = tmp_path_factory.mktemp("writable_data") + map_path = unwritable_directory / "cache_paths.sqlite3" + with set_config(directory_index_map_path=map_path): + out = DBDirectoryIndexer(data_path) + assert out.index_path.parent == data_path + + def test_writable_dir_survives_unavailable_sqlite_map( + self, tmp_path_factory, monkeypatch + ): + """Use a local index when SQLite reports an unavailable map.""" + import dascore.io.index.indexer as indexer + + def unavailable_map(*args, **kwargs): + raise sqlite3.OperationalError("database is locked") + + monkeypatch.setattr(indexer, "_get_index_map", unavailable_map) + data_path = tmp_path_factory.mktemp("writable_data") + out = DBDirectoryIndexer(data_path) + assert out.index_path.parent == data_path + + @pytest.mark.skipif(os.name == "nt", reason="requires POSIX permissions") + def test_read_only_map_reuses_existing_mapping(self, tmp_path): + """A readable map remains useful when its cache cannot be written.""" + data_path = tmp_path / "data" + cache_dir = tmp_path / "cache" + mapped_dir = tmp_path / "mapped" + data_path.mkdir() + cache_dir.mkdir() + mapped_dir.mkdir() + map_path = cache_dir / "cache_paths.sqlite3" + mapped_path = mapped_dir / "index.sqlite3" + with set_config(directory_index_map_path=map_path): + first = DBDirectoryIndexer(data_path, index_path=mapped_path) + first.close() + + lock_path = cache_dir / f"{map_path.name}.recovery.lock" + os.chmod(data_path, 0o555) + os.chmod(cache_dir, 0o555) + os.chmod(map_path, 0o444) + os.chmod(lock_path, 0o444) + try: + with set_config(directory_index_map_path=map_path): + second = DBDirectoryIndexer(data_path) + assert second.index_path == mapped_path + second.close() + finally: + os.chmod(map_path, 0o644) + os.chmod(lock_path, 0o644) + os.chmod(cache_dir, 0o755) + os.chmod(data_path, 0o755) + + @pytest.mark.skipif(os.name == "nt", reason="requires POSIX permissions") + def test_writable_dir_survives_corrupt_read_only_map(self, tmp_path): + """An unrecoverable optional map falls back to the local index.""" + data_path = tmp_path / "data" + cache_dir = tmp_path / "cache" + data_path.mkdir() + cache_dir.mkdir() + map_path = cache_dir / "cache_paths.sqlite3" + map_path.write_bytes(b"not a sqlite database") + os.chmod(cache_dir, 0o555) + os.chmod(map_path, 0o444) + try: + with set_config(directory_index_map_path=map_path): + out = DBDirectoryIndexer(data_path) + assert out.index_path.parent == data_path + out.close() + finally: + os.chmod(map_path, 0o644) + os.chmod(cache_dir, 0o755) + def test_writable_dir_index_exists(self, tmp_path_factory): """A test case where the index does exist.""" path = tmp_path_factory.mktemp("normal_indexer_test") @@ -124,7 +220,7 @@ def test_corrupt_cache(self, directory_indexer_bad_cache, tmp_path_factory): assert directory_indexer_bad_cache.exists() with set_config(directory_index_map_path=directory_indexer_bad_cache): DBDirectoryIndexer(path) - assert not directory_indexer_bad_cache.exists() + assert directory_indexer_bad_cache.read_bytes().startswith(b"SQLite format 3") def test_remote_directory_not_supported(self): """Remote directory indexing should fail fast.""" @@ -141,7 +237,7 @@ def test_local_upath_normalized_to_path(self, tmp_path): def test_index_map_path_comes_from_config(self, tmp_path): """Index map paths should be sourced from runtime configuration.""" - index_map_path = tmp_path / "cache_paths.json" + index_map_path = tmp_path / "cache_paths.sqlite3" with set_config(directory_index_map_path=index_map_path): out = DBDirectoryIndexer(tmp_path) assert out.index_map_path == index_map_path @@ -150,28 +246,342 @@ def test_index_map_path_comes_from_config(self, tmp_path): class TestIndexMap: """Tests for the index-location map helpers.""" - def test_update_is_atomic_and_leaves_no_temp(self, tmp_path): - """Updates write via a temp file and swap it in, leaving no debris.""" + def test_default_map_path_uses_sqlite(self): + """The default map path names the SQLite database directly.""" + from dascore.config import DascoreConfig + + assert DascoreConfig().directory_index_map_path.name == "cache_paths.sqlite3" + + def test_updates_use_sqlite_database(self, tmp_path): + """Updates persist transactionally without temporary database files.""" from dascore.io.index.indexer import _get_index_map, _update_index_map - cache_path = tmp_path / "cache_paths.json" + cache_path = tmp_path / "cache_paths.sqlite3" _update_index_map({"a": "1"}, cache_path=str(cache_path)) _update_index_map({"b": "2"}, cache_path=str(cache_path)) assert _get_index_map(str(cache_path)) == {"a": "1", "b": "2"} - # the swap target is the only file left in the directory. - assert [p.name for p in tmp_path.iterdir()] == [cache_path.name] + assert cache_path.read_bytes().startswith(b"SQLite format 3") + assert {path.name for path in tmp_path.iterdir()} == { + cache_path.name, + f"{cache_path.name}.recovery.lock", + } + + def test_corrupt_sqlite_is_rebuilt(self, tmp_path): + """A corrupt disposable SQLite map is replaced transparently.""" + from dascore.io.index.indexer import _get_index_map, _update_index_map + + cache_path = tmp_path / "cache_paths.sqlite3" + cache_path.write_bytes(b"not a sqlite database") + assert _get_index_map(cache_path) == {} + assert cache_path.read_bytes().startswith(b"SQLite format 3") + _update_index_map({"a": "1"}, cache_path) + assert _get_index_map(cache_path) == {"a": "1"} + + def test_corrupt_recovery_lock_does_not_block_rebuild(self, tmp_path): + """Recovery locking must not depend on the lock file's contents.""" + from dascore.io.index.indexer import _get_index_map + + cache_path = tmp_path / "cache_paths.sqlite3" + lock_path = tmp_path / "cache_paths.sqlite3.recovery.lock" + cache_path.write_bytes(b"not a sqlite database") + lock_path.write_bytes(b"arbitrary non-database contents") + + assert _get_index_map(cache_path) == {} + assert cache_path.read_bytes().startswith(b"SQLite format 3") + + @pytest.mark.parametrize( + "message", + ["database disk image is malformed", "file is not a database"], + ) + def test_corruption_detection_without_error_code(self, message): + """Python 3.10 SQLite messages still identify disposable corruption.""" + from dascore.io.index.indexer import _is_corrupt_index_map_error + + error = sqlite3.DatabaseError(message) + assert not hasattr(error, "sqlite_errorcode") + assert _is_corrupt_index_map_error(error) + + def test_unrelated_database_error_propagates(self, tmp_path, monkeypatch): + """Operational failures are not mistaken for disposable corruption.""" + import dascore.io.index.indexer as indexer + + error = sqlite3.OperationalError("database is locked") + + def fail_open(*args): + raise error + + monkeypatch.setattr(indexer, "_open_index_map", fail_open) + with pytest.raises(sqlite3.OperationalError, match="database is locked"): + indexer._get_index_map(tmp_path / "cache_paths.sqlite3") + + def test_corruption_during_select_is_rebuilt(self, tmp_path, monkeypatch): + """Corruption raised by a query triggers a rebuild and retry.""" + import dascore.io.index.indexer as indexer + + cache_path = tmp_path / "cache_paths.sqlite3" + indexer._update_index_map({"a": "1"}, cache_path) + original_open = indexer._open_index_map + open_calls = 0 + + class CorruptOnSelect: + """Make the first map connection fail when queried.""" + + def __init__(self, connection): + self.connection = connection + + def execute(self, statement, *args, **kwargs): + if statement.lstrip().startswith("SELECT"): + self.connection.close() + cache_path.write_bytes(b"not a sqlite database") + error = sqlite3.DatabaseError("database disk image is malformed") + error.sqlite_errorcode = getattr(sqlite3, "SQLITE_CORRUPT", 11) + raise error + return self.connection.execute(statement, *args, **kwargs) + + def close(self): + self.connection.close() + + def open_map(*args): + nonlocal open_calls + open_calls += 1 + connection = original_open(*args) + if open_calls == 1: + return CorruptOnSelect(connection) + return connection + + monkeypatch.setattr(indexer, "_open_index_map", open_map) + assert indexer._get_index_map(cache_path) == {} + assert cache_path.read_bytes().startswith(b"SQLite format 3") + + def test_integrity_check_rebuilds_hidden_corruption(self, tmp_path, monkeypatch): + """Recovery checks pages that a successful row query may not visit.""" + import dascore.io.index.indexer as indexer + + cache_path = tmp_path / "cache_paths.sqlite3" + indexer._update_index_map({"a": "1"}, cache_path) + original_open = indexer._open_index_map + open_calls = 0 + + class IntegrityResult: + """Return a deterministic non-OK SQLite integrity result.""" + + @staticmethod + def fetchall(): + return [("invalid freelist page",)] + + class HiddenCorruption: + """Expose corruption only to SQLite's full integrity check.""" + + def __init__(self, connection): + self.connection = connection + + def execute(self, statement, *args, **kwargs): + if statement == "PRAGMA integrity_check": + return IntegrityResult() + return self.connection.execute(statement, *args, **kwargs) + + def close(self): + self.connection.close() + + def open_map(*args): + nonlocal open_calls + open_calls += 1 + connection = original_open(*args) + if open_calls == 1: + return HiddenCorruption(connection) + return connection + + monkeypatch.setattr(indexer, "_open_index_map", open_map) + indexer._recover_index_map(cache_path) + assert indexer._get_index_map(cache_path) == {} + + @pytest.mark.skipif(os.name == "nt", reason="fcntl is POSIX-only") + def test_missing_fcntl_uses_process_lock(self, tmp_path, monkeypatch): + """Platforms without fcntl retain single-process map support.""" + import dascore.io.index.indexer as indexer + + cache_path = tmp_path / "cache_paths.sqlite3" + monkeypatch.setitem(sys.modules, "fcntl", None) + + indexer._update_index_map({"a": "1"}, cache_path) + assert indexer._get_index_map(cache_path) == {"a": "1"} + + @pytest.mark.parametrize("_repeat", range(3)) + def test_concurrent_process_recovery_preserves_updates(self, tmp_path, _repeat): + """Separate processes repeatedly coordinate corruption recovery.""" + from dascore.io.index.indexer import _get_index_map + + cache_path = tmp_path / "cache_paths.sqlite3" + cache_path.write_bytes(b"not a sqlite database") + context = multiprocessing.get_context("spawn") + start = context.Event() + ready = [context.Event(), context.Event()] + keys = ("a", "b") + processes = [ + context.Process( + target=_update_corrupt_index_map_process, + args=(str(cache_path), key, event, start), + ) + for key, event in zip(keys, ready, strict=True) + ] + try: + for process in processes: + process.start() + assert all(event.wait(timeout=20) for event in ready) + start.set() + for process in processes: + process.join(timeout=30) + assert [process.exitcode for process in processes] == [0, 0] + finally: + start.set() + for process in processes: + if process.is_alive(): + process.terminate() + process.join(timeout=5) + + assert _get_index_map(cache_path) == {"a": "index-a", "b": "index-b"} + + @pytest.mark.skipif( + "fork" not in multiprocessing.get_all_start_methods(), + reason="requires POSIX fork", + ) + def test_recovery_guard_blocks_fork(self, tmp_path, monkeypatch): + """Fork waits until no recovery file-lock descriptor can be inherited.""" + import dascore.io.index.indexer as indexer + + cache_path = tmp_path / "cache_paths.sqlite3" + indexer._update_index_map({"a": "1"}, cache_path) + + class ObservedLock: + """Expose when the at-fork callback waits on the guard lock.""" + + def __init__(self): + self._lock = threading.Lock() + self._count_lock = threading.Lock() + self.acquire_count = 0 + self.fork_waiting = threading.Event() + + def acquire(self): + with self._count_lock: + self.acquire_count += 1 + if self.acquire_count == 2: + self.fork_waiting.set() + return self._lock.acquire() + + def release(self): + self._lock.release() + + def __enter__(self): + self.acquire() + return self + + def __exit__(self, *_): + self.release() + + observed_lock = ObservedLock() + monkeypatch.setattr(indexer, "_INDEX_MAP_RECOVERY_LOCK", observed_lock) + guard_held = threading.Event() + release_guard = threading.Event() + fork_errors = [] + + def hold_guard(): + with indexer._index_map_recovery_guard(cache_path): + guard_held.set() + release_guard.wait(timeout=20) + + context = multiprocessing.get_context("fork") + process = context.Process( + target=_read_index_map_process, + args=(str(cache_path),), + ) + + def start_child(): + try: + with suppress_warnings(DeprecationWarning): + process.start() + process.join(timeout=10) + except BaseException as exc: + fork_errors.append(exc) + + guard_thread = threading.Thread(target=hold_guard) + fork_thread = threading.Thread(target=start_child) + guard_thread.start() + try: + assert guard_held.wait(timeout=5) + fork_thread.start() + assert observed_lock.fork_waiting.wait(timeout=5) + assert process.pid is None + release_guard.set() + fork_thread.join(timeout=15) + assert not fork_thread.is_alive() + assert not fork_errors + assert process.exitcode == 0 + finally: + release_guard.set() + guard_thread.join(timeout=5) + if fork_thread.is_alive(): + fork_thread.join(timeout=15) + if process.pid is not None and process.is_alive(): + process.terminate() + process.join(timeout=5) def test_get_reads_fresh_each_call(self, tmp_path): - """Reads are not cached, so out-of-band changes are seen (no @cache).""" + """Reads are not cached, so out-of-band SQLite changes are visible.""" from dascore.io.index.indexer import _get_index_map, _update_index_map - cache_path = tmp_path / "cache_paths.json" + cache_path = tmp_path / "cache_paths.sqlite3" _update_index_map({"a": "1"}, cache_path=str(cache_path)) assert _get_index_map(str(cache_path)) == {"a": "1"} - # simulate another process rewriting the map underneath us. - cache_path.write_text(json.dumps({"a": "1", "b": "2"})) + with closing(sqlite3.connect(cache_path)) as connection, connection: + connection.execute( + "INSERT INTO index_map (directory, index_path) VALUES (?, ?)", + ("b", "2"), + ) assert _get_index_map(str(cache_path)) == {"a": "1", "b": "2"} + def test_concurrent_updates_preserve_every_entry(self, tmp_path): + """Concurrent callers preserve entries when they start together.""" + from dascore.io.index.indexer import _get_index_map, _update_index_map + + cache_path = tmp_path / "cache_paths.sqlite3" + keys = [f"source-{number}" for number in range(8)] + barrier = threading.Barrier(len(keys)) + + def update(key): + barrier.wait(timeout=5) + _update_index_map({key: f"index-{key}"}, cache_path=cache_path) + + with ThreadPoolExecutor(max_workers=len(keys)) as pool: + futures = [pool.submit(update, key) for key in keys] + for future in futures: + future.result(timeout=10) + + expected = {key: f"index-{key}" for key in keys} + assert _get_index_map(cache_path) == expected + + def test_legacy_json_is_ignored(self, tmp_path): + """A neighboring JSON map is neither read nor changed.""" + from dascore.io.index.indexer import _get_index_map + + legacy_path = tmp_path / "cache_paths.json" + database_path = tmp_path / "cache_paths.sqlite3" + legacy_contents = '{"a": "1", "b": "2"}' + legacy_path.write_text(legacy_contents) + + assert _get_index_map(database_path) == {} + assert database_path.read_bytes().startswith(b"SQLite format 3") + assert legacy_path.read_text() == legacy_contents + + def test_configured_suffix_does_not_select_storage_format(self, tmp_path): + """The configured map path is SQLite regardless of its suffix.""" + from dascore.io.index.indexer import _get_index_map, _update_index_map + + cache_path = tmp_path / "custom-index-map.json" + _update_index_map({"external-data": "external-index.sqlite3"}, cache_path) + + assert cache_path.read_bytes().startswith(b"SQLite format 3") + assert _get_index_map(cache_path) == {"external-data": "external-index.sqlite3"} + class TestBasics: """Basic tests for indexer."""