diff --git a/dascore/config.py b/dascore/config.py index e53012217..bef2473e2 100644 --- a/dascore/config.py +++ b/dascore/config.py @@ -83,9 +83,9 @@ class DascoreConfig(BaseModel): default_factory=lambda: _get_cache_root() / "data", 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.", + directory_index_map_dir: Path = Field( + default_factory=lambda: _get_cache_root() / "indexes" / "path_map", + description="Directory of per-data-directory external index-location entries.", ) # Progress display. @@ -129,7 +129,7 @@ class DascoreConfig(BaseModel): @field_validator( "downloader_cache_dir", - "directory_index_map_path", + "directory_index_map_dir", "remote_cache_dir", mode="before", ) diff --git a/dascore/io/index/indexer.py b/dascore/io/index/indexer.py index 219d5b3f0..05f2f46d1 100644 --- a/dascore/io/index/indexer.py +++ b/dascore/io/index/indexer.py @@ -32,52 +32,75 @@ from dascore.utils.paths import directory_writable, requires_local_directory -def _get_index_map(cache_path) -> dict: +def _path_digest(path) -> str: + """Stable per-path digest (hash() of a str/Path is per-process random). + + Uses the full sha256 so distinct paths never collide (a collision + would make two read-only directories share one index file). Prefer + os.fsencode so local filesystem paths (including non-UTF-8 filename + bytes) digest exactly; fall back to a plain string encoding for inputs + os.fsencode rejects, e.g. remote URL/UPath directories we may support + later. """ - Return a fresh dict of index locations read from disk. + try: + encoded = os.fsencode(path) + except (TypeError, ValueError, NotImplementedError): + encoded = str(path).encode("utf-8", "surrogatepass") + return hashlib.sha256(encoded).hexdigest() + + +def _map_entry_path(directory, map_dir) -> Path: + """Return the entry file recording one data directory's index location.""" + return Path(map_dir) / f"{_path_digest(directory)}.json" + - 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. +def _get_mapped_index_path(directory, map_dir) -> Path | None: """ - 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") + Return the index path recorded for a data directory, or None. + + Each directory's mapping is its own small JSON file. A corrupt or + unreadable entry simply reads as a miss; the next atomic write for + that directory self-heals it. Reads never delete the entry, which + would otherwise race a concurrent writer's repair. See #508. + """ + entry = _map_entry_path(directory, map_dir) + try: + with entry.open("r") as fi: + data = json.load(fi) + except (OSError, ValueError): + # Missing, unreadable, non-UTF-8, or non-JSON: treat as a miss. + return None + # Guard the (astronomically unlikely) digest collision and any + # malformed payload; a mismatch or bad shape is treated as a miss. + if not isinstance(data, dict) or data.get("directory") != str(directory): + return None + index_path = data.get("index_path") + if not isinstance(index_path, str) or not index_path: + return None + return Path(index_path) + + +def _set_mapped_index_path(directory, index_path, map_dir) -> None: + """ + Record a data directory's external index location. + + The entry is written to a sibling temp file and swapped into place, so + a concurrent reader never sees a half-written file. Different + directories use different files, so concurrent writers to distinct + directories cannot clobber one another. + """ + entry = _map_entry_path(directory, map_dir) + entry.parent.mkdir(exist_ok=True, parents=True) + payload = {"directory": str(directory), "index_path": str(index_path)} + fd, tmp = tempfile.mkstemp(dir=entry.parent, prefix=entry.name, suffix=".tmp") try: with os.fdopen(fd, "w") as fi: - json.dump(data, fi) - os.replace(tmp, path) + json.dump(payload, fi) + os.replace(tmp, entry) except BaseException: with suppress(OSError): os.unlink(tmp) raise - return data class DBDirectoryIndexer: @@ -94,8 +117,9 @@ class DBDirectoryIndexer: """ ext: str | None = None - # user-level file tracking index locations for unwritable data dirs - index_map_path: Path = config_attr("directory_index_map_path") + # cache dir holding per-data-directory index-location entries, used + # when a data directory itself is not writable + index_map_dir: Path = config_attr("directory_index_map_dir") def __init__( self, @@ -149,36 +173,31 @@ def _find_index_path(self, index_path=None) -> Path: default; when the data directory is read-only the index lives in the dascore cache and its location is recorded in the index map. """ - map_key = str(self.path) + map_dir = self.index_map_dir if index_path: index_path = Path(index_path).absolute() - update = {map_key: str(index_path)} - _update_index_map(update, cache_path=str(self.index_map_path)) + # A custom location is the only case worth recording: it is not + # otherwise rediscoverable. Read-only fallbacks are deterministic. + _set_mapped_index_path(self.path, index_path, map_dir) return index_path expected = self.path / self._index_name with suppress(PermissionError): if expected.exists(): return expected - path_map = _get_index_map(cache_path=str(self.index_map_path)) - if out := path_map.get(map_key): - mapped = Path(out) + mapped = _get_mapped_index_path(self.path, map_dir) + if mapped is not None and not self._is_legacy_or_foreign_index(mapped): # Index-map entries from older DASCore versions can point at # the retired PyTables (.h5) index; those are not usable and # a fresh SQLite index is built in their place. - if not self._is_legacy_or_foreign_index(mapped): - return mapped + return mapped if not directory_writable(self.path): # A stable digest, not hash(): str/Path hashing is randomized # per process (PYTHONHASHSEED), so hash() would name a new - # index file every session and orphan the previous one. - digest = hashlib.sha256(str(self.path).encode()).hexdigest()[:16] - name = f"_dascore_index_{digest}.sqlite3" - index_path = self.index_map_path.parent / name - _update_index_map( - {map_key: str(index_path.absolute())}, - cache_path=str(self.index_map_path), - ) - return index_path + # index file every session and orphan the previous one. The + # name is deterministic, so no map entry is needed to find it. + map_dir.mkdir(parents=True, exist_ok=True) + name = f"_dascore_index_{_path_digest(self.path)}.sqlite3" + return map_dir / name return expected def ensure_updated(self) -> bool: diff --git a/docs/changelog.qmd b/docs/changelog.qmd index 80113e64f..262631175 100644 --- a/docs/changelog.qmd +++ b/docs/changelog.qmd @@ -23,6 +23,7 @@ The [releases page](https://github.com/DASDAE/dascore/releases) tracks changes f - Chunk partitioning is unit-aware along the chunked dimension: patches whose coordinate units have different dimensionality (or unitful vs unitless) can never plan into one output — they were previously grouped by raw SI magnitude and failed only at load. Compatible unit spellings (e.g. metres and feet) now merge correctly, converted to the first member's units. - Spool selection now validates unknown names and quantity dimensionality, composes chained regular expressions with AND, supports explicit `_attrs` and `_coords` namespaces (as `name -> selector` mappings, or as a name/collection of names tagging bare keyword arguments), and applies relative offsets only to coordinate selectors. Values indexed without units stay candidates for quantity selectors, and an attribute observed with dimensionally incompatible units across files skips the incompatible values (with a warning) instead of failing the whole index update. - The legacy PyTables directory index has been replaced by SQLite. SQLite supports concurrent readers and one serialized writer on local filesystems; network filesystems with unreliable locking are not supported. +- The external index-location map (used only for read-only data directories and custom index paths) is now stored as one small file per data directory under a cache directory (`directory_index_map_dir`), replacing the single shared `cache_paths.json`. A corrupt entry only affects its own directory and self-heals, and concurrent writers to distinct directories can no longer clobber one another. The map is a disposable cache; any old `cache_paths.json` is ignored and left in place. - With the PyTables dependency removed, `dascore.utils.hdf5` no longer provides `PyTablesReader`, `PyTablesWriter`, or their `HDF5Reader`/`HDF5Writer` aliases. Use `H5Reader`/`H5Writer` (h5py-based) instead. - Added `CoordSegmented` and `concat_coords(...)` for representing exact piecewise-monotonic coordinates with queryable sampling changes and gaps. `patch.split_gaps()` splits segmented dimensions into contiguous patches, and `dc.write(..., split=True)` can perform that split for multi-patch formats. - Added `dc.scan_payloads(...)` for retrieving raw scan payloads with full coordinate managers and source provenance without loading data arrays. diff --git a/tests/conftest.py b/tests/conftest.py index 7d97dc9d4..9369bd370 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -103,9 +103,9 @@ 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" - with set_config(directory_index_map_path=tmp_map_path): + """For all tests cases, use a temporary index-map directory.""" + tmp_map_dir = tmp_path_factory.mktemp("cache_paths") / "path_map" + with set_config(directory_index_map_dir=tmp_map_dir): 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..0fc9eb2ea 100644 --- a/tests/test_io/test_index/test_index_edge_cases.py +++ b/tests/test_io/test_index/test_index_edge_cases.py @@ -1214,7 +1214,7 @@ def test_legacy_entry_ignored(self, tmp_path): import dascore as dc from dascore.io.index.indexer import ( - _update_index_map, + _set_mapped_index_path, ) # data directory with one file, plus a fake legacy index mapping. @@ -1224,9 +1224,9 @@ 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" - with dc.set_config(directory_index_map_path=map_path): - _update_index_map({str(data_dir): str(legacy)}, cache_path=str(map_path)) + map_dir = tmp_path / "path_map" + with dc.set_config(directory_index_map_dir=map_dir): + _set_mapped_index_path(data_dir, legacy, map_dir) spool = dc.spool(data_dir).update() assert len(spool) == 1 diff --git a/tests/test_io/test_indexer.py b/tests/test_io/test_indexer.py index 6bf09165d..79ad670ad 100644 --- a/tests/test_io/test_indexer.py +++ b/tests/test_io/test_indexer.py @@ -63,21 +63,11 @@ def unwritable_directory(self, tmp_path_factory): yield path os.chmod(path, 0o755) - @pytest.fixture() - 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'") - return cache_path - def test_directory_cant_write(self, unwritable_directory): """Ensure correct path is found when a read-only directory is used.""" dir_index = DBDirectoryIndexer(unwritable_directory) index_path = dir_index.index_path - index_map_path = dir_index.index_map_path - assert index_map_path.parent == index_path.parent + assert dir_index.index_map_dir == index_path.parent def test_read_only_index_name_is_stable(self, unwritable_directory, tmp_path): """The read-only fallback index name must not depend on hash(). @@ -86,12 +76,12 @@ def test_read_only_index_name_is_stable(self, unwritable_directory, tmp_path): name would differ every session and orphan the prior index. The name is a stable digest of the directory path. """ - import hashlib + from dascore.io.index.indexer import _path_digest - map_path = tmp_path / "cache_paths.json" - with set_config(directory_index_map_path=map_path): + map_dir = tmp_path / "path_map" + with set_config(directory_index_map_dir=map_dir): first = DBDirectoryIndexer(unwritable_directory).index_path - digest = hashlib.sha256(str(unwritable_directory).encode()).hexdigest()[:16] + digest = _path_digest(unwritable_directory) assert first.name == f"_dascore_index_{digest}.sqlite3" def test_specify_index_path(self, tmp_path_factory): @@ -118,13 +108,21 @@ def test_writable_dir_index_exists(self, tmp_path_factory): assert first.index_path == second.index_path assert first.index_path.exists() - def test_corrupt_cache(self, directory_indexer_bad_cache, tmp_path_factory): - """Ensure a corrupted cache doesn't crash indexing. See #508.""" - path = tmp_path_factory.mktemp("corrupt_cache_test") - 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() + def test_corrupt_cache(self, tmp_path): + """Ensure a corrupt map entry doesn't crash indexing. See #508.""" + from dascore.io.index.indexer import _map_entry_path + + data_dir = tmp_path / "data" + data_dir.mkdir() + map_dir = tmp_path / "path_map" + entry = _map_entry_path(data_dir, map_dir) + entry.parent.mkdir(parents=True) + entry.write_text("{'bad': 'json'") + with set_config(directory_index_map_dir=map_dir): + indexer = DBDirectoryIndexer(data_dir) + # The corrupt entry reads as a miss, so the writable data dir keeps + # its in-directory index rather than crashing. + assert indexer.index_path.parent == data_dir def test_remote_directory_not_supported(self): """Remote directory indexing should fail fast.""" @@ -139,53 +137,133 @@ def test_local_upath_normalized_to_path(self, tmp_path): assert isinstance(out.path, Path) assert out.path == Path(tmp_path).absolute() - 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" - with set_config(directory_index_map_path=index_map_path): + def test_index_map_dir_comes_from_config(self, tmp_path): + """Index map dir should be sourced from runtime configuration.""" + index_map_dir = tmp_path / "path_map" + with set_config(directory_index_map_dir=index_map_dir): out = DBDirectoryIndexer(tmp_path) - assert out.index_map_path == index_map_path + assert out.index_map_dir == index_map_dir class TestIndexMap: - """Tests for the index-location map helpers.""" + """Tests for the per-directory index-location entries.""" + + def test_digest_falls_back_for_non_fspath(self): + """A directory os.fsencode rejects still digests (future URL support).""" + import hashlib + + from dascore.io.index.indexer import _path_digest + + class _Remote: + """Stand-in for a remote UPath whose fspath is unavailable.""" + + def __fspath__(self): + raise NotImplementedError + + def __str__(self): + return "memory://data/dir" + + expected = hashlib.sha256(b"memory://data/dir").hexdigest() + assert _path_digest(_Remote()) == expected + + def test_roundtrip(self, tmp_path): + """A recorded index path is read back for the same directory.""" + from dascore.io.index.indexer import ( + _get_mapped_index_path, + _set_mapped_index_path, + ) + + map_dir = tmp_path / "path_map" + index_path = tmp_path / "idx.sqlite3" + _set_mapped_index_path(tmp_path / "data", index_path, map_dir) + assert _get_mapped_index_path(tmp_path / "data", map_dir) == index_path + + def test_missing_entry_is_none(self, tmp_path): + """An unmapped directory reads back as None (a cache miss).""" + from dascore.io.index.indexer import _get_mapped_index_path + + assert _get_mapped_index_path(tmp_path / "nope", tmp_path / "path_map") is None + + def test_distinct_dirs_dont_collide(self, tmp_path): + """Separate directories use separate entry files (no lost writes).""" + from dascore.io.index.indexer import ( + _get_mapped_index_path, + _set_mapped_index_path, + ) + + map_dir = tmp_path / "path_map" + _set_mapped_index_path(tmp_path / "a", "index-a", map_dir) + _set_mapped_index_path(tmp_path / "b", "index-b", map_dir) + assert _get_mapped_index_path(tmp_path / "a", map_dir) == Path("index-a") + assert _get_mapped_index_path(tmp_path / "b", map_dir) == Path("index-b") + + def test_corrupt_entry_is_miss(self, tmp_path): + """A corrupt entry reads as a miss (and is not deleted). See #508.""" + from dascore.io.index.indexer import ( + _get_mapped_index_path, + _map_entry_path, + ) + + map_dir = tmp_path / "path_map" + entry = _map_entry_path(tmp_path / "a", map_dir) + entry.parent.mkdir(parents=True) + entry.write_text("{not json") + assert _get_mapped_index_path(tmp_path / "a", map_dir) is None + # Reads never delete the entry; a later write self-heals it. + assert entry.exists() + + def test_bad_payload_shapes_are_miss(self, tmp_path): + """Non-string or empty index paths read as a miss, not an error.""" + from dascore.io.index.indexer import _get_mapped_index_path, _map_entry_path + + map_dir = tmp_path / "path_map" + entry = _map_entry_path(tmp_path / "a", map_dir) + entry.parent.mkdir(parents=True) + entry.write_text( + json.dumps({"directory": str(tmp_path / "a"), "index_path": []}) + ) + assert _get_mapped_index_path(tmp_path / "a", map_dir) is None + + def test_digest_collision_is_miss(self, tmp_path): + """An entry whose stored directory differs reads as a miss.""" + from dascore.io.index.indexer import _get_mapped_index_path, _map_entry_path + + map_dir = tmp_path / "path_map" + entry = _map_entry_path(tmp_path / "a", map_dir) + entry.parent.mkdir(parents=True) + # Same file, but recorded for a different directory. + entry.write_text(json.dumps({"directory": "other", "index_path": "x"})) + assert _get_mapped_index_path(tmp_path / "a", map_dir) is None 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.""" - from dascore.io.index.indexer import _get_index_map, _update_index_map - - cache_path = tmp_path / "cache_paths.json" - _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] - - def test_get_reads_fresh_each_call(self, tmp_path): - """Reads are not cached, so out-of-band changes are seen (no @cache).""" - from dascore.io.index.indexer import _get_index_map, _update_index_map - - cache_path = tmp_path / "cache_paths.json" - _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"})) - assert _get_index_map(str(cache_path)) == {"a": "1", "b": "2"} + """Writes swap a temp file into place, leaving no debris.""" + from dascore.io.index.indexer import ( + _get_mapped_index_path, + _map_entry_path, + _set_mapped_index_path, + ) + + map_dir = tmp_path / "path_map" + _set_mapped_index_path(tmp_path / "a", "1", map_dir) + _set_mapped_index_path(tmp_path / "a", "2", map_dir) + assert _get_mapped_index_path(tmp_path / "a", map_dir) == Path("2") + entry = _map_entry_path(tmp_path / "a", map_dir) + assert [p.name for p in map_dir.iterdir()] == [entry.name] def test_failed_swap_cleans_up_temp(self, tmp_path, monkeypatch): """A failure during the atomic swap unlinks the temp file and re-raises.""" from dascore.io.index import indexer as indexer_mod - cache_path = tmp_path / "cache_paths.json" + map_dir = tmp_path / "path_map" def boom(*args, **kwargs): raise RuntimeError("swap failed") monkeypatch.setattr(indexer_mod.os, "replace", boom) with pytest.raises(RuntimeError, match="swap failed"): - indexer_mod._update_index_map({"a": "1"}, cache_path=str(cache_path)) - # No temp debris and no half-written target left behind. - assert list(tmp_path.iterdir()) == [] + indexer_mod._set_mapped_index_path(tmp_path / "a", "1", map_dir) + # No temp debris and no half-written entry left behind. + assert list(map_dir.iterdir()) == [] class TestWalkResilience: