From b81c633299ff835e5fb78f977dc5aaf7602685c6 Mon Sep 17 00:00:00 2001 From: Alex Rubinsteyn Date: Fri, 2 Oct 2026 10:41:08 -0400 Subject: [PATCH 1/3] Add transactional derived-artifact materialization --- .github/workflows/tests.yml | 1 + CHANGELOG.md | 7 + README.md | 9 +- datacache/__init__.py | 4 + datacache/bundles.py | 12 +- datacache/materialization.py | 513 +++++++++++++++ datacache/provenance.py | 11 +- datacache/resume.py | 3 +- docs/api.md | 47 ++ docs/data.md | 5 + docs/design/materialization.md | 36 ++ docs/materialization.md | 173 +++++ examples/reference_dna_materialization.py | 55 ++ tests/test_materialization.py | 747 ++++++++++++++++++++++ 14 files changed, 1611 insertions(+), 12 deletions(-) create mode 100644 datacache/materialization.py create mode 100644 docs/design/materialization.md create mode 100644 docs/materialization.md create mode 100644 examples/reference_dna_materialization.py create mode 100644 tests/test_materialization.py diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 9f3ad89..f77ce6d 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -51,6 +51,7 @@ jobs: run: | python -m examples.basic_usage python -m examples.versioned_datasets + python -m examples.reference_dna_materialization - name: Publish coverage to Coveralls # Coverage upload is best-effort: parallel matrix jobs race to post to # the same Coveralls build, and the loser gets a "build already closed" diff --git a/CHANGELOG.md b/CHANGELOG.md index 2c78229..b56e7cd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,13 @@ ## 1.17.0 +- Add `materialize` and offline `inspect_materialization` for caller-built + multi-file artifacts with complete raw-input/transform/output dependency + receipts. Preserve completed inputs and HTTP resume state across failures, + publish immutable output generations atomically, and discard owned inputs + only after successful publication unless retention is requested. Local + sources stay caller-owned; scientific validation stays in the builder (#90). + - Add a `store_path(name, version)` callback to `VersionedDatasetRegistry` for exact consumer-chosen managed-store locations. It is mutually exclusive with root configuration, does not run at construction, and preserves immutable diff --git a/README.md b/README.md index 58adc24..2c0a44c 100644 --- a/README.md +++ b/README.md @@ -19,8 +19,9 @@ python -m pip install datacache python -m pip install "datacache[html]" ``` -tqdm is installed by default; progress displays are opt-in with -`show_progress=True`. Normal use does not import tqdm or configure your +tqdm is installed by default; existing download APIs enable displays with +`show_progress=True`, while `materialize` enables them by default. Quiet use +does not import tqdm or configure your application's logging. The old `datacache[progress]` install syntax remains supported as a compatibility alias. @@ -165,6 +166,7 @@ lists every public signature, default, return value, exception, and example. | Task | API | Result | | --- | --- | --- | | Install or reuse a versioned dataset | `VersionedDatasetRegistry`, `install_bundle(...)` | Mapping of asset names to snapshot paths | +| Build derived files from versioned dependencies | `materialize(...)` | Mapping of output names to snapshot paths | | Install versioned archive trees | `VersionedArchiveRegistry`, `install_archive(...)` | Immutable extracted-generation `Path` | | Reuse an established fixed-path versioned file cache | `VersionedFileRegistry` | One Path and a legacy-compatible root receipt | | Inspect a complete dataset generation | `inspect_bundle(...)` | `BundleInspection` | @@ -189,6 +191,9 @@ FASTA files with `fetch_file`, then parse them in the consuming library. ## Guides +- [Derived-artifact materialization](docs/materialization.md): caller-owned + scientific builders, atomic output generations, dependency receipts, resumable + inputs and post-publication input cleanup. - [Versioned datasets and bundles](docs/bundles.md): pinned versions, atomic installation, offline recovery, path lifetime, and downstream adoption. - [Archive directory installation](docs/archives.md): safe tar extraction, split archives, diff --git a/datacache/__init__.py b/datacache/__init__.py index 679192a..a9d4360 100644 --- a/datacache/__init__.py +++ b/datacache/__init__.py @@ -36,6 +36,7 @@ from .archives import ( ArchiveInspection, VersionedArchiveRegistry, inspect_archive, install_archive) from .file_registry import VersionedFileRegistry +from .materialization import MaterializationInspection, inspect_materialization, materialize from .version import __version__ __all__ = [ @@ -43,6 +44,9 @@ 'fetch_file', 'discard_partial', 'BundleInspection', + 'MaterializationInspection', + 'inspect_materialization', + 'materialize', 'ArchiveInspection', 'VersionedArchiveRegistry', 'VersionedDatasetRegistry', diff --git a/datacache/bundles.py b/datacache/bundles.py index fed6a90..23ecc2e 100644 --- a/datacache/bundles.py +++ b/datacache/bundles.py @@ -88,10 +88,10 @@ def _directory(path): raise FileValidationError(path, 'expected a directory, not a link') -def _store(path): +def _store(path, *, marker=STORE): _directory(path) - if read_json(path / STORE) != {'format': FORMAT}: - raise FileValidationError(path, 'unrecognized bundle store') + if read_json(path / marker) != {'format': FORMAT}: + raise FileValidationError(path, 'unrecognized generation store') _directory(path / 'generations') @@ -214,19 +214,19 @@ def _paths(inspection): return {name: value.path for name, value in inspection.files.items()} -def _initialize(path): +def _initialize(path, *, marker=STORE): existing_mode = None if path_present(path): _directory(path) if any(path.iterdir()): - _store(path) # Never take over an arbitrary nonempty directory. + _store(path, marker=marker) # Never adopt a foreign directory. return existing_mode = stat.S_IMODE(path.lstat().st_mode) # Publish a complete store skeleton at once: concurrent first-time readers # see absence or a recognized store, never a half-written ownership marker. staging = Path(tempfile.mkdtemp(prefix='.datacache-store-', dir=path.parent)) try: - write_json(staging / STORE, {'format': FORMAT}, mode=_file_mode(path.parent)) + write_json(staging / marker, {'format': FORMAT}, mode=_file_mode(path.parent)) (staging / 'generations').mkdir() os.chmod(staging, existing_mode if existing_mode is not None else stat.S_IMODE((staging / 'generations').stat().st_mode)) diff --git a/datacache/materialization.py b/datacache/materialization.py new file mode 100644 index 0000000..50d0379 --- /dev/null +++ b/datacache/materialization.py @@ -0,0 +1,513 @@ +"""Transactional caller-built artifacts with durable raw-input dependencies.""" + +from dataclasses import dataclass, field +from datetime import datetime, timezone +import hashlib +import json +import os +from pathlib import Path +import shutil +import stat +from urllib.parse import urlsplit +from urllib.request import url2pathname +from uuid import uuid4 + +from ._filesystem import file_lock, open_regular, path_present, read_json, write_json +from .bundles import _directory, _file_mode, _generation, _initialize, _relative_name, _store +from .inspection import FileInspection +from .integrity import FileValidationError, _validate_expectations +from .progress import Progress +from . import provenance + +STORE = '.datacache-materialization.json' +MANIFEST = '.datacache-manifest.json' +CURRENT = 'current.json' +INPUTS = '.datacache-inputs.json' +FORMAT = 1 +CHUNK_SIZE = 2 ** 20 + + +def _json_copy(value): + # Reject non-JSON identities and NaNs, and detach from caller mutation. + return json.loads(json.dumps(value, sort_keys=True, allow_nan=False)) + + +def _fingerprint(value): + return hashlib.sha256(json.dumps(value, sort_keys=True, allow_nan=False).encode()).hexdigest() + + +def _inventory(mapping): + if not isinstance(mapping, dict) or not mapping: + raise ValueError('sources and outputs must be nonempty mappings') + for name in mapping: + _relative_name(name) + folded = {name.casefold() for name in mapping} + parents = {'/'.join(name.split('/')[:i]).casefold() + for name in mapping for i in range(1, len(name.split('/')))} + if len(folded) != len(mapping) or folded & parents: + raise ValueError('paths collide as files/directories or ignoring case') + return mapping + + +def _expectations(spec): + if not isinstance(spec, dict) or set(spec) - {'sha256', 'size'}: + raise ValueError('output expectations must contain only sha256 and size') + digest, size = spec.get('sha256'), spec.get('size') + _validate_expectations(digest, size) + return dict(sha256=digest.lower() if digest else None, size=size) + + +def _transform(value): + if (not isinstance(value, dict) or set(value) - {'version', 'options'} or + not isinstance(value.get('version'), str) or not value['version']): + raise ValueError('transform requires a nonempty version and optional JSON options') + return dict(version=value['version'], options=_json_copy(value.get('options', {}))) + + +def _definition(sources, transform, outputs): + normalized, acquisition = {}, {} + for name, spec in _inventory(sources).items(): + if not isinstance(spec, dict) or set(spec) - {'url', 'path', 'sha256', 'size'}: + raise ValueError('source metadata supports url or path, sha256 and size') + if ('url' in spec) == ('path' in spec): + raise ValueError('each source requires exactly one of url or path') + expected = _expectations({key: spec[key] for key in ('sha256', 'size') if key in spec}) + if 'path' in spec: + source = Path(spec['path']).absolute() + origin = source.as_uri() + acquisition[name] = ('path', str(source)) + else: + origin = spec['url'] + if not isinstance(origin, str) or not origin: + raise ValueError('source URL must be a nonempty string') + parts = urlsplit(origin) + if parts.scheme not in ('http', 'https', 'ftp', 'file'): + raise ValueError('source URL must use HTTP(S), FTP or file') + if parts.scheme == 'file': + if parts.netloc not in ('', 'localhost') or parts.query or parts.fragment: + raise ValueError('local file URLs cannot have remote hosts, queries or fragments') + acquisition[name] = ('path', str(Path(url2pathname(parts.path)).absolute())) + else: + acquisition[name] = ('url', origin) + normalized[name] = dict(expected, identity=_fingerprint(origin), origin=provenance.redact_url(origin)) + sidecars = {Path(provenance.sidecar_path(name)).as_posix().casefold() for name in normalized} + occupied = {name.casefold() for name in normalized} + parents = {'/'.join(name.split('/')[:i]).casefold() for name in normalized + for i in range(1, len(name.split('/')))} + if sidecars & (occupied | parents): + raise ValueError('source paths collide with automatic provenance sidecars') + output_specs = {name: _expectations(spec) for name, spec in _inventory(outputs).items()} + return dict(sources=normalized, transform=_transform(transform), outputs=output_specs), acquisition + + +def _recorded_definition(value): + if not isinstance(value, dict) or set(value) != {'sources', 'transform', 'outputs'}: + raise ValueError('invalid materialization dependency definition') + for spec in _inventory(value['sources']).values(): + if not isinstance(spec, dict) or set(spec) != {'origin', 'identity', 'sha256', 'size'}: + raise ValueError('invalid source dependency definition') + _expectations({key: spec[key] for key in ('sha256', 'size')}) + _validate_expectations(spec['identity'], None) + if spec['identity'] is None or not isinstance(spec['origin'], str): + raise ValueError('source identity and origin are required') + for spec in _inventory(value['outputs']).values(): + _expectations(spec) + if _transform(value['transform']) != value['transform']: + raise ValueError('invalid transform definition') + return value + + +def _check_observed(path, observed, expected): + if not isinstance(observed, dict) or set(observed) != {'sha256', 'size'}: + raise ValueError('invalid observed file metadata') + _validate_expectations(observed['sha256'], observed['size']) + if observed['sha256'] is None or observed['size'] is None: + raise ValueError('observed hashes and sizes are required') + for key in ('sha256', 'size'): + if expected[key] is not None and observed[key] != expected[key]: + raise FileValidationError(path, '%s disagrees with dependency expectations' % key) + + +def _observe(path, expected=None, *, verify_files=True, show_progress=False): + with os.fdopen(open_regular(path), 'rb') as handle: + info = os.fstat(handle.fileno()) + if verify_files: + digest, size = hashlib.sha256(), 0 + with Progress(show_progress, 'Verifying', info.st_size) as progress: + for chunk in iter(lambda: handle.read(CHUNK_SIZE), b''): + digest.update(chunk) + size += len(chunk) + progress(size, info.st_size) + if size != info.st_size or os.fstat(handle.fileno()).st_mtime_ns != info.st_mtime_ns: + raise FileValidationError(path, 'file changed during verification') + observed = dict(sha256=digest.hexdigest(), size=size) + if expected is not None: + _check_observed(path, observed, expected) + else: + observed = dict(sha256=None, size=info.st_size) + if expected is not None and expected['size'] is not None and info.st_size != expected['size']: + raise FileValidationError(path, 'file size disagrees with receipt') + return observed, info + + +def _source_records(value, definition): + if not isinstance(value, dict) or set(value) != set(definition): + raise ValueError('source receipt inventory disagrees with definition') + for name, record in value.items(): + if not isinstance(record, dict): + raise ValueError('invalid source receipt') + observed = {key: record.get(key) for key in ('sha256', 'size')} + _check_observed(name, observed, definition[name]) + trusted = definition[name]['sha256'] is not None + if (record.get('identity') != definition[name]['identity'] or + record.get('origin') != definition[name]['origin'] or + record.get('verified') is not trusted or + not isinstance(record.get('fetched_at'), str)): + raise ValueError('source receipt identity or trust disagrees with definition') + transport = record.get('transport') + if transport is not None: + from .resume import _strong_etag + if (not isinstance(transport, dict) or set(transport) != {'header', 'value'} or + transport['header'] not in ('ETag', 'Last-Modified') or + not isinstance(transport['value'], str) or + '\r' in transport['value'] or '\n' in transport['value'] or + (transport['header'] == 'ETag' and not _strong_etag(transport['value']))): + raise ValueError('invalid validated transport metadata') + return value + + +def _tree(root, recorded, trusted=None, *, verify_files=True, show_progress=False, allow_manifest=True): + """Check the exact inventory without following directories or special files.""" + files, found, stack = {}, set(), [(Path(root), '')] + parents = {'/'.join(name.split('/')[:i]) for name in recorded + for i in range(1, len(name.split('/')))} + while stack: + directory, prefix = stack.pop() + _directory(directory) + with os.scandir(directory) as entries: + for entry in entries: + name = prefix + entry.name + if allow_manifest and not prefix and name == MANIFEST: + continue + _relative_name(name) + if entry.is_dir(follow_symlinks=False): + if name not in parents: + raise FileValidationError(entry.path, 'undeclared output directory') + stack.append((Path(entry.path), name + '/')) + continue + if name not in recorded: + raise FileValidationError(entry.path, 'undeclared output file') + observed, info = _observe(entry.path, recorded[name], verify_files=verify_files, + show_progress=show_progress) + found.add(name) + files[name] = FileInspection(entry.path, 'available', + verified=bool(verify_files and trusted and trusted[name]['sha256']), + size=info.st_size, mtime=info.st_mtime) + if found != set(recorded): + raise FileValidationError(root, 'missing declared outputs: %s' % sorted(set(recorded) - found)) + return files + + +@dataclass(frozen=True) +class MaterializationInspection: + """One read-only snapshot, including source provenance and transform identity. + + verified refers only to outputs matched against caller-trusted hashes now. + Source records distinguish acquisition-time trusted hashes from observation. + """ + path: str + status: str + verified: bool = False + generation: object = None + files: dict = field(default_factory=dict) + sources: dict = field(default_factory=dict) + transform: object = None + error: object = None + + +def _inspect_tree(store, directory, expected=None, *, verify_files=True, show_progress=False): + receipt = read_json(directory / MANIFEST, limit=None) + if not isinstance(receipt, dict) or receipt.get('format') != FORMAT: + raise ValueError('unrecognized materialization receipt') + definition = _recorded_definition(receipt.get('definition')) + if expected is not None and definition != expected: + raise FileValidationError(directory, 'dependency identity changed; use force=True to refresh') + sources = _source_records(receipt.get('sources'), definition['sources']) + outputs = receipt.get('outputs') + if not isinstance(outputs, dict) or set(outputs) != set(definition['outputs']): + raise ValueError('output receipt inventory disagrees with definition') + for name, observed in outputs.items(): + _check_observed(name, observed, definition['outputs'][name]) + trusted = expected['outputs'] if expected is not None else None + files = _tree(directory, outputs, trusted, verify_files=verify_files, show_progress=show_progress) + verified = verify_files and trusted is not None and all(spec['sha256'] for spec in trusted.values()) + return MaterializationInspection(str(store), 'available', bool(verified), str(directory), + files, sources, definition['transform']) + + +def inspect_materialization(destination, sources=None, *, transform=None, outputs=None, verify_files=True): + """Inspect offline without writes, locks, source access or automatic recovery. + + Supply all of sources/transform/outputs to check dependency identity; omit + all three for receipt-only consistency checks. Metadata-only inspection + checks regular files and sizes, never claims verification of payload hashes. + """ + if not isinstance(verify_files, bool): + raise ValueError('verify_files must be a boolean') + supplied = sum(value is not None for value in (sources, transform, outputs)) + if supplied not in (0, 3): + raise ValueError('provide sources, transform and outputs together, or omit all three') + expected = _definition(sources, transform, outputs)[0] if supplied else None + return _inspect_materialization(destination, expected, verify_files=verify_files) + + +def _inspect_materialization(destination, expected, *, verify_files=True): + path = Path(destination) + try: + if not path_present(path): + return MaterializationInspection(str(path), 'missing') + try: + _store(path, marker=STORE) + except FileNotFoundError: + _directory(path) + if not any(path.iterdir()): + return MaterializationInspection(str(path), 'missing') + _store(path, marker=STORE) + try: + pointer = read_json(path / CURRENT) + except FileNotFoundError: + status = 'recovery-required' if any((path / 'generations').iterdir()) else 'missing' + return MaterializationInspection(str(path), status) + return _inspect_tree(path, _generation(path, pointer['generation']), expected, verify_files=verify_files) + except PermissionError as error: + return MaterializationInspection(str(path), 'inaccessible', error=error) + except (OSError, ValueError, KeyError, TypeError, RecursionError) as error: + return MaterializationInspection(str(path), 'invalid', error=error) + + +def _private(path): + if not path_present(path): + path.mkdir(mode=0o700) + info = path.lstat() + if (not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid() or + stat.S_IMODE(info.st_mode) & 0o077): + raise FileValidationError(path, 'working directory must be private and owner-owned') + + +def _input_directory(store, definition): + return store / ('.inputs-%s-%d' % (_fingerprint(definition['sources']), os.getuid())) + + +def _cleanup_inputs(store, definition, retain_sources): + directory = _input_directory(store, definition) + if not retain_sources and path_present(directory): + _private(directory) + shutil.rmtree(directory) + + +def _acquire_inputs(directory, definition, acquisition, options, *, force=False): + from .download import fetch_file + _private(directory) + receipt_path = directory / INPUTS + records = read_json(receipt_path, limit=None) if path_present(receipt_path) else {} + if not isinstance(records, dict) or set(records) - set(definition): + raise FileValidationError(receipt_path, 'invalid private input receipt') + paths = {} + for name, spec in definition.items(): + target = directory / name + for parent in reversed(target.relative_to(directory).parents): + current = directory / parent + if not path_present(current): + current.mkdir(mode=0o700) + _directory(current) + if name in records: + _source_records({name: records[name]}, {name: spec}) + try: + _observe(target, {key: records[name][key] for key in ('sha256', 'size')}) + except (FileNotFoundError, FileValidationError): + if not force: + raise FileValidationError(target, 'invalid private input; use force=True to repair') + if path_present(target): + fd = open_regular(target) # Never repair a planted link/special file. + os.close(fd) + target.unlink() + provenance.remove(target) + del records[name] + write_json(receipt_path, records) + else: + paths[name] = str(target) + continue + if path_present(target): + # Only installer-created, private completed bytes can exist here. + # Validate their type before fetch_file's ordinary cache-hit path. + fd = open_regular(target) + os.close(fd) + # A hash-pinned acquisition is independently reusable. A bare + # size-only/unverified file has no durable completion receipt yet. + if spec['sha256'] is None: + target.unlink() + else: + try: + _observe(target, spec) + except FileValidationError: + if not force: + raise FileValidationError(target, 'invalid private input; use force=True to repair') + target.unlink() + provenance.remove(target) + kind, source = acquisition[name] + if kind == 'url': + fetch_file(source, destination=target, raw=True, record_provenance=True, + expected_sha256=spec['sha256'], expected_size=spec['size'], **options) + elif not path_present(target): + temporary = directory / ('.copy-' + uuid4().hex) + try: + with os.fdopen(open_regular(source), 'rb') as handle, temporary.open('xb') as output: + with Progress(options['show_progress'], 'Copying', os.fstat(handle.fileno()).st_size) as progress: + size = 0 + for chunk in iter(lambda: handle.read(CHUNK_SIZE), b''): + output.write(chunk) + size += len(chunk) + progress(size, progress.total) + output.flush() + os.fsync(output.fileno()) + _observe(temporary, spec, show_progress=options['show_progress']) + os.replace(temporary, target) + finally: + temporary.unlink(missing_ok=True) + observed, info = _observe(target, spec, show_progress=options['show_progress']) + origin_record = provenance.read(target, info) or {} + record = dict(observed, origin=spec['origin'], identity=spec['identity'], + verified=spec['sha256'] is not None, + fetched_at=origin_record.get('fetched_at', datetime.now(timezone.utc).isoformat())) + if origin_record.get('transport') is not None: + record['transport'] = origin_record['transport'] + records[name] = record + write_json(receipt_path, records) + paths[name] = str(target) + return paths, records + + +def materialize(destination, sources, *, transform, outputs, builder, force=False, + retain_sources=False, verify_files=True, download_options=None): + """Build all declared outputs and their receipt as one immutable generation. + + builder(source_paths, output_paths) must create/close exactly the declared + outputs and leave inputs unchanged. Source definitions contain url or path + plus optional raw sha256/size; outputs map names to optional sha256/size. + transform is {version: nonempty string, options: opaque JSON value}. + Changed dependencies and corrupt generations require explicit force=True. + Force rebuilds outputs while reusing any complete matching private inputs. + + Inputs survive failures, including without HTTP resume. resume=True enables + existing raw HTTP resume rules for remote sources; local inputs are copied. + Successful publication removes owned inputs unless retain_sources=True. + Caller-owned inputs and old output generations are never deleted. Cache + hits are offline and read-only; verify_files=False skips payload hashing + only on reuse. New publication and explicit recovery always check bytes. + Download/copy/verification progress is on by default; builder progress is + caller-owned. Installation requires POSIX locks and atomic local renames. + """ + if not all(isinstance(value, bool) for value in (force, retain_sources, verify_files)): + raise ValueError('force, retain_sources and verify_files must be booleans') + if not callable(builder): + raise ValueError('builder must be callable') + definition, acquisition = _definition(sources, transform, outputs) + options = dict(download_options or {}) + allowed = {'timeout', 'chunk_size', 'progress_callback', 'show_progress', + 'max_retries', 'retry_backoff', 'retry_max_delay', 'resume'} + if set(options) - allowed: + raise ValueError('unsupported materialization download options') + options.setdefault('show_progress', True) + if not isinstance(options['show_progress'], bool) or not isinstance(options.get('resume', False), bool): + raise ValueError('show_progress and resume must be booleans') + from .retries import validate_retry_options + from .download import DEFAULT_CHUNK_SIZE, DEFAULT_MAX_RETRIES, DEFAULT_RETRY_BACKOFF, DEFAULT_RETRY_MAX_DELAY + validate_retry_options(options.get('max_retries', DEFAULT_MAX_RETRIES), + options.get('retry_backoff', DEFAULT_RETRY_BACKOFF), + options.get('retry_max_delay', DEFAULT_RETRY_MAX_DELAY)) + chunk_size = options.get('chunk_size', DEFAULT_CHUNK_SIZE) + if not isinstance(chunk_size, int) or isinstance(chunk_size, bool) or chunk_size <= 0: + raise ValueError('chunk_size must be a positive integer') + if options.get('progress_callback') is not None and not callable(options['progress_callback']): + raise ValueError('progress_callback must be callable') + if options.get('resume'): + from .resume import validate_resume + for name, (kind, source) in acquisition.items(): + if kind == 'url': + validate_resume(source, definition['sources'][name]['sha256'], definition['sources'][name]['size']) + path = Path(destination).absolute() + + def inspect_current(): + return _inspect_materialization(path, definition, verify_files=verify_files) + + def check_hit(inspection): + if inspection.status == 'inaccessible': + raise inspection.error + if not force and inspection.status == 'invalid': + raise FileValidationError(path, 'invalid materialization; use force=True to refresh or repair') from inspection.error + return not force and inspection.status == 'available' + + inspection = inspect_current() + if check_hit(inspection): + return {name: value.path for name, value in inspection.files.items()} + if os.name != 'posix': + raise NotImplementedError('Materialization requires a POSIX local filesystem') + path.parent.mkdir(parents=True, exist_ok=True) + key = hashlib.sha256(os.fsencode(path.name)).hexdigest()[:32] + with file_lock(path.parent / ('.datacache-materialization-lock-' + key)): + _initialize(path, marker=STORE) + inspection = inspect_current() + if check_hit(inspection): + return {name: value.path for name, value in inspection.files.items()} + if inspection.status in ('missing', 'recovery-required'): + for entry in sorted((path / 'generations').iterdir(), reverse=True): + try: + candidate = _inspect_tree(path, _generation(path, entry.name), definition) + except (OSError, ValueError, KeyError, TypeError, RecursionError): + continue + write_json(path / CURRENT, {'generation': entry.name}, mode=_file_mode(path)) + _cleanup_inputs(path, definition, retain_sources) + return {name: value.path for name, value in candidate.files.items()} + inputs = _input_directory(path, definition) + source_paths, records = _acquire_inputs(inputs, definition['sources'], acquisition, options, force=force) + working = path / ('.staging-%s-%d' % (_fingerprint(definition), os.getuid())) + _private(working) + # Only this user's installer-owned failed outputs are discarded. + for entry in working.iterdir(): + if entry.is_dir() and not entry.is_symlink(): + shutil.rmtree(entry) + else: + entry.unlink() + staged = working / 'files' + staged.mkdir() + output_paths = {name: str(staged / name) for name in definition['outputs']} + for target in output_paths.values(): + Path(target).parent.mkdir(parents=True, exist_ok=True) + try: + builder(dict(source_paths), dict(output_paths)) + # Builder callbacks are trusted code, not a sandbox; enforce their + # read-only-input contract before creating a dependency receipt. + for name, source in source_paths.items(): + _observe(source, {key: records[name][key] for key in ('sha256', 'size')}) + observed = {} + # Reject extras/links/specials before hashing any declared output. + _tree(staged, definition['outputs'], verify_files=False, allow_manifest=False) + for name, target in output_paths.items(): + observed[name] = _observe(target, definition['outputs'][name], + show_progress=options['show_progress'])[0] + os.chmod(target, _file_mode(staged)) + receipt = dict(format=FORMAT, created_at=datetime.now(timezone.utc).isoformat(), + definition=definition, sources=records, outputs=observed) + write_json(staged / MANIFEST, receipt, mode=_file_mode(staged)) + _inspect_tree(path, staged, definition) + generation = uuid4().hex + final = path / 'generations' / generation + os.replace(staged, final) + # Remap only after validating the receipt in its private staging. + candidate = _inspect_tree(path, final, definition, verify_files=False) + write_json(path / CURRENT, {'generation': generation}, mode=_file_mode(path)) + _cleanup_inputs(path, definition, retain_sources) + return {name: value.path for name, value in candidate.files.items()} + finally: + _private(working) + shutil.rmtree(working) diff --git a/datacache/provenance.py b/datacache/provenance.py index 34fa4d1..69858a2 100644 --- a/datacache/provenance.py +++ b/datacache/provenance.py @@ -60,22 +60,27 @@ def redact_url(url): return urlunsplit((parts.scheme, host, parts.path, "", "")) -def describe(download_url, info, sha256=None): +def describe(download_url, info, sha256=None, *, transport=None): """Provenance, as JSON text, for the bytes that stat info describes. info must come from the exact bytes being published (the staged file), not from whatever is at the destination afterwards, which a concurrent writer may have replaced. sha256 is recorded only when those bytes were verified against a trusted expectation: a digest of what a server sent proves nothing. + transport optionally records the validator accepted by resumable transport; + it is evidence of representation consistency, not a trusted content hash. """ - return json.dumps({ + record = { "format": FORMAT, "url": redact_url(download_url), "fetched_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), "size": info.st_size, "mtime_ns": info.st_mtime_ns, "sha256": sha256.lower() if sha256 else None, - }, sort_keys=True) + } + if transport is not None: + record['transport'] = transport + return json.dumps(record, sort_keys=True) def remove(path): diff --git a/datacache/resume.py b/datacache/resume.py index 13239f4..b8e130e 100644 --- a/datacache/resume.py +++ b/datacache/resume.py @@ -244,7 +244,8 @@ def report(): staged = handle.name with partial.open('rb') as source: download.copyfileobj(source, handle) - record = (provenance.describe(download_url, os.stat(staged), expected_sha256) + record = (provenance.describe(download_url, os.stat(staged), expected_sha256, + transport=metadata.get('validator')) if record_provenance else None) download._publish_file(staged, destination, record) finally: diff --git a/docs/api.md b/docs/api.md index dc0796a..f98be0a 100644 --- a/docs/api.md +++ b/docs/api.md @@ -17,6 +17,7 @@ explain the longer workflows and compatibility guarantees. | Downloading | [fetch_file](#fetch_file), [fetch_csv_dataframe](#fetch_csv_dataframe), [fetch_and_transform](#fetch_and_transform) | | Complete archive trees | [install_archive](#install_archive), [inspect_archive](#inspect_archive), [ArchiveInspection](#archiveinspection), [VersionedArchiveRegistry](#versionedarchiveregistry) | | Versioned file bundles | [install_bundle](#install_bundle), [inspect_bundle](#inspect_bundle), [BundleInspection](#bundleinspection), [VersionedDatasetRegistry](#versioneddatasetregistry), [VersionedFileRegistry](#versionedfileregistry) | +| Derived artifacts | [materialize](#materialize), [inspect_materialization](#inspect_materialization), [MaterializationInspection](#materializationinspection) | | Paths and presence | [expected_path](#expected_path), [file_exists](#file_exists), [build_local_filename](#build_local_filename), [get_data_dir](#get_data_dir), [get_cache_root](#get_cache_root), [resolve_path](#resolve_path), [build_path](#build_path), [ensure_dir](#ensure_dir), [clear_cache](#clear_cache) | | Integrity and permissions | [validate_file](#validate_file), [inspect_file](#inspect_file), [inspect_files](#inspect_files), [make_file_readable](#make_file_readable) | | Results and exceptions | [FileInspection](#fileinspection), [CacheInspection](#cacheinspection), [FileValidationError](#filevalidationerror) | @@ -1266,6 +1267,52 @@ The [bundle guide](bundles.md) includes an example and downstream migration note | `is_cached(name, version=None, *, verify_files=False)` | Whether inspection reports `available`; payload verification is opt-in. | | `status(*, verify_files=False)` | One metadata-only row per pinned default: name, version, description, available_versions and inspection. Pass `verify_files=True` for full checks. | +## materialize + +```text +materialize(destination, sources, *, transform, outputs, builder, force=False, + retain_sources=False, verify_files=True, download_options=None) +``` + +Install caller-built outputs and their dependency receipt as one immutable +generation. Sources map relative names to `{url: ...}` or `{path: ...}`, plus +optional raw-byte `sha256`/`size`; outputs map names to optional output-byte +`sha256`/`size`. Transform is `{version: nonempty_string, options: JSON_value}`. +The builder receives private `(source_paths, output_paths)` mappings, must leave +inputs unchanged and create/close exactly the declared regular output files. +Returns name-to-string snapshot paths; builder return values are ignored. + +Dependency mismatch/corruption requires explicit `force=True`. Completed inputs +survive failures; successful publication removes owned inputs unless retention +is requested. Caller-owned local inputs and old generations are never deleted. +Cache hits are offline/read-only. Fast reuse is opt-in; new publication/recovery +always hash files. Download/copy/verification progress defaults on, configurable +through download options. Installation initially requires POSIX local storage. +See the [materialization guide](materialization.md) for receipts, resume, callback +contracts, retention, recovery and peak/retained disk use. + +## inspect_materialization + +```text +inspect_materialization(destination, sources=None, *, transform=None, outputs=None, + verify_files=True) +``` + +Return `MaterializationInspection` offline without writes, locks, original +input access or repair. Supply all of sources/transform/outputs for requested +dependency checks, or omit all three for receipt-only consistency checks. +`verify_files=False` validates inventory, types and sizes without payload reads +and never claims trusted verification. + +## MaterializationInspection + +A read-only snapshot with `path`, `status`, `verified`, `generation`, `files`, +`sources`, `transform` and `error`. Status is available, missing, invalid, +inaccessible or recovery-required. `verified` means all output files matched +caller-trusted hashes now. Source records separately report raw observed hashes, +sizes, redacted origins, identity, acquisition time, acquisition-time trusted +verification and available validated transport metadata. + ## VersionedFileRegistry `VersionedFileRegistry(datasets, *, cache_dir, error_cls=RuntimeError)` diff --git a/docs/data.md b/docs/data.md index 26d6c91..552280e 100644 --- a/docs/data.md +++ b/docs/data.md @@ -150,6 +150,11 @@ rebuild a matching database; also increment `version` when its data changes. ## Custom single-file transformations +For versioned dependencies, multiple outputs, resumable compressed inputs or +automatic post-publication source cleanup, use the +[derived-artifact materialization API](materialization.md). The existing helper +below retains its compatible single-file behavior. + ```python from pathlib import Path from datacache import fetch_and_transform diff --git a/docs/design/materialization.md b/docs/design/materialization.md new file mode 100644 index 0000000..9686f7d --- /dev/null +++ b/docs/design/materialization.md @@ -0,0 +1,36 @@ +# Derived-artifact transaction (#90) + +Add `materialize` and `inspect_materialization`, leaving `fetch_and_transform` +unchanged. A caller chooses the exact store, raw source definitions, a JSON +transform identity (`version` plus opaque `options`), required output inventory, +and `builder(source_paths, output_paths)`. Biological parsing stays in the +builder. DataCache validates regular files, hashes bytes in bounded memory, +records dependency receipts and publishes one immutable generation atomically. + +Use existing filesystem, transport, resume, progress and generation-store +primitives. Materialization stores have a distinct ownership marker: neither +bundles nor arbitrary legacy directories can be overwritten even with force. +Source and transform identity mismatches raise until explicit `force=True`. +Input expectations describe raw (including compressed) bytes; output +expectations describe builder-produced bytes. Observed digests are not trust. + +Writers serialize per artifact. Inputs live in owner-only, source-keyed working +directories, survive download/build/publication interruptions, and can be +reused after a transform-only change. Builders receive only private staged +paths and must close all handles, leave inputs unchanged, and create exactly +the declared regular outputs. Outputs and the receipt publish together through +one generation rename and one current-pointer replacement. Failed refreshes +never modify the previous pointer or old generation. Missing pointers can be +recovered explicitly from fully verified matching local generations. + +Default successful publication removes only the transaction's private owned +inputs; retention is opt-in. Cache hits stay read-only and never clean up. +Local inputs are copied and never modified/deleted. Old output generations are +retained. Installation initially has the same POSIX-local-filesystem constraint +as resumable bundle installation; cross-platform bundle support is separate. + +Test tiny gzip and two-output builds, actual HTTP range resumption, killed +builders, completed-input retry without network, identity mismatch, source and +output corruption, unsafe paths/files, failure at both publication boundaries, +read-only offline reuse, per-artifact serialization and unrelated writer +independence. Document disk use including transport's temporary resume copy. diff --git a/docs/materialization.md b/docs/materialization.md new file mode 100644 index 0000000..38c593c --- /dev/null +++ b/docs/materialization.md @@ -0,0 +1,173 @@ +# Derived-artifact materialization + +Use `materialize` when an application must download raw inputs, build one or +more derived files, validate them, and publish them together. For example, +PyEnsembl can keep its FASTA parsing and biological validation in a builder +while DataCache owns reference-DNA download, retry, receipts and publication. +Existing `fetch_and_transform` keeps its compatible single-file behavior. + +```python +from datacache import materialize, inspect_materialization + +sources = { + "dna.fa.gz": { + "url": compressed_dna_url, + "sha256": compressed_dna_sha256, + "size": compressed_dna_size, + }, +} +outputs = {"dna.fa": {}, "index.json": {}} +transform = {"version": "dna-parser-1", "options": {"assembly": "GRCh38"}} + +def build(source_paths, output_paths): + # These are private staged paths, not the caller's source or public output. + # Stream gzip into output_paths["dna.fa"], validate sequence/contig contents, + # and create output_paths["index.json"]. Scientific policy stays here. + parse_and_validate_dna(source_paths["dna.fa.gz"], output_paths) + +paths = materialize( + artifact_store, sources, transform=transform, outputs=outputs, builder=build, + download_options={"resume": True, "timeout": 60}, +) +dna_path = paths["dna.fa"] +index_path = paths["index.json"] +state = inspect_materialization( + artifact_store, sources, transform=transform, outputs=outputs, +) +``` + +The [offline reference-DNA example](../examples/reference_dna_materialization.py) +is runnable with `python -m examples.reference_dna_materialization` and uses +tiny gzip data without any biological-library dependency. + +## Definitions and builder contract + +Choose an exact managed store path, independently of the Python package +version. Sources are a nonempty mapping from relative names to either `url` or +`path`, plus optional `sha256` and `size`. Local paths and `file://` URLs are +copied into private working storage; their original bytes and permissions are +never modified, and caller-owned inputs are never deleted. Links, hard-linked +files, FIFOs and other nonregular local inputs are rejected without blocking. + +Source expectations describe the **raw input bytes**, including compression. +Outputs are a nonempty mapping from relative names to optional `sha256` and +`size`, describing **builder-produced bytes**. Empty `{}` expectations are +supported; observation and a local consistency receipt do not authenticate +these bytes. An expected size of zero supports explicitly empty inputs/outputs. +Names cannot escape the store, collide by case or as files/directories, or +occupy reserved metadata/provenance paths. + +`transform` contains a nonempty opaque `version` string and optional JSON +`options`. Bump this identity when parsing, validation, index format or other +scientific decisions change. DataCache does not infer code changes or decide +biological compatibility. Do not put secrets in options: they are recorded. + +`builder(source_paths, output_paths)` receives name-to-string mappings. Output +paths are initially absent, with required parent directories already present. +The builder must create exactly those outputs, close all handles, and leave +inputs unchanged before returning. Only regular, single-link files and the +declared output parent directories are accepted. Return values are ignored; +`materialize` returns name-to-absolute-path mappings belonging to the published generation. +The callback is trusted application code, not a sandbox. Builder progress, +streamed decompression and biological validation are caller-owned. + +## Identity, receipts and trust + +Each generation records the complete source/transform/output dependency +definition and observed sizes/SHA-256 hashes. Source receipts also record +redacted origins, acquisition time, acquisition-time trusted-hash verification, +and validated transport metadata when available. Resumable HTTP acquisitions +record the accepted strong ETag or Last-Modified validator. These validators +establish transport consistency, not scientific correctness or authenticity. +Ordinary transfers without validated transport metadata do not invent it. + +Full source references are represented by SHA-256 identity fingerprints. +Display origins omit URL credentials, query text and fragments, but retain the +path; avoid secret-bearing paths. Changing the full URL/path (including signed +query text), integrity expectations, transform identity or output inventory +makes an existing generation `invalid` for that request. Refresh is explicit: +`materialize(..., force=True)` builds a replacement. This API does not claim +different mirror URLs are the same dependency; caller-declared source identity +policy is separate. No remote `latest` lookup occurs. + +`inspect_materialization` reports `available`, `missing`, `invalid`, +`inaccessible` or `recovery-required`, with generation paths, output inspections, +source receipts, transform identity and an error cause. Supply all three of +`sources`, `transform`, and `outputs` to check the requested identity, or omit +all three for receipt-only consistency checks. Inspection never accesses the +original inputs, performs network requests, writes, locks or repairs. + +Full inspection and cache-hit checks hash outputs by default. `verified=True` +means all outputs matched caller-supplied trusted hashes **now**; source receipt +verification is explicitly historical. Without trusted output hashes, even +fully consistent results have `verified=False`. `verify_files=False` checks +ownership, dependency identity, exact inventory, types, readability and sizes, +but cannot detect same-size corruption and always reports unverified outputs. +This fast flag applies only to reuse. New output publication and explicit +recovery always hash the bytes. + +## Interruption, publication and retention + +Writers serialize per artifact; unrelated artifacts have independent locks. +Completed inputs persist in owner-only source-keyed directories even without +HTTP resume. `download_options={"resume": True}` also preserves partial raw +HTTP(S) downloads, using existing hash/size or size/strong-ETag resume rules. +Remote resumable inputs require `size`; local inputs are simply copied. +The supported transport options are timeout, chunk size, progress callback, +progress display, retry count/backoff/delay and resume. Download, copy and hash +progress is enabled by default; set `show_progress=False` in download options +for noninteractive use. Valid cache hits remain quiet. + +Download, gzip, validation and builder failures leave the previously selected +output generation untouched. Failed builder outputs are discarded; completed +inputs survive for retry without another download. A killed builder can leave +private partial outputs, which the next transaction with the same definition +discards. Changing only the transform can reuse retained matching inputs. +`force=True` rebuilds outputs, reuses sound inputs, and can repair corrupted +regular private inputs; it never adopts unsafe paths or foreign directories. + +All declared outputs and `.datacache-manifest.json` are staged and checked as +one tree. DataCache renames that tree into `generations/` and atomically +replaces `current.json`. It does not present independent fixed-file renames as +a multi-file transaction. Readers see the complete old or complete new tree; +returned snapshot paths continue to exist across later refreshes. Do not add +indexes or edit files in a published generation; declare them as outputs or +place independently mutable files outside the managed store. + +If the first pointer publication is interrupted, explicit `materialize` can +fully check and select a matching complete local generation without fetching +inputs or calling the builder. Inspection alone does not recover anything. + +After successful pointer publication/recovery, owned inputs for that source +definition are removed by default. `retain_sources=True` keeps them privately +for future builds. Cache hits never perform cleanup: a crash between pointer +publication and cleanup may leave inputs for an explicit later refresh. Inputs +for other source definitions/users and old output generations are never pruned +implicitly. Abandoned input/staging variants require deliberate operator +cleanup; general inventory/pruning is separate work. Retention is installation +policy, not output identity, so toggling it on a cache hit does not mutate disk. + +The distinct `.datacache-materialization.json` ownership marker prevents +adoption of populated legacy directories or bundle stores, even with force. +Empty precreated stores are allowed and keep their directory modes. As with +current resumable bundles, installation requires a POSIX local filesystem with +`flock` and atomic sibling renames. Read-only existing hits require no lock. +This is atomic visibility, not a guarantee against filesystem/power-loss faults. + +## Disk use + +Let `C` be total completed raw inputs, `D` the new output generation, and `O` +all old retained output generations. During a build, storage is approximately +`O + C + D`, plus scratch files created by the builder. Generation publication +is a rename and does not duplicate `D`. After successful default cleanup, +retained storage is `O + D`; opt-in source retention uses `O + C + D`. + +The existing resumable transport copies a completed private partial into file +publication staging before removing the partial. That phase can need an extra +compressed-input-sized copy: peak is approximately +`O + max(C + largest_input, C + D + builder_scratch)`, excluding any previously +abandoned input/staging variants. For one 1 GB compressed / 3 GB decompressed +human-DNA input, a first build needs roughly 4 GB plus builder scratch, then +retains roughly 3 GB (or 4 GB with source retention). A refresh retaining an old +3 GB output needs roughly 7 GB peak and retains 6 GB afterward. Caller-owned +local originals occupy additional storage outside this managed store. diff --git a/examples/reference_dna_materialization.py b/examples/reference_dna_materialization.py new file mode 100644 index 0000000..03ff9d4 --- /dev/null +++ b/examples/reference_dna_materialization.py @@ -0,0 +1,55 @@ +"""Tiny offline reference-DNA transaction; scientific work stays in the builder.""" + +import gzip +from hashlib import sha256 +import json +from pathlib import Path +import shutil +from tempfile import TemporaryDirectory + +from datacache import inspect_materialization, materialize + + +def build_reference(source_paths, output_paths): + with gzip.open(source_paths['dna.fa.gz'], 'rb') as source, open(output_paths['dna.fa'], 'wb') as output: + shutil.copyfileobj(source, output, length=2 ** 20) + # A toy caller-side biological check, not a DataCache FASTA parser. + with open(output_paths['dna.fa'], 'rb') as fasta: + if fasta.readline() != b'>chr1\n': + raise ValueError('unexpected contig header') + length = sum(len(line.strip()) for line in fasta) + Path(output_paths['index.json']).write_text(json.dumps({'length': length})) + + +def main(): + with TemporaryDirectory() as temporary: + root = Path(temporary) + dna = b'>chr1\nACGTACGT\n' + compressed = gzip.compress(dna, mtime=0) + original = root / 'upstream.fa.gz' + original.write_bytes(compressed) + sources = {'dna.fa.gz': {'path': original, 'sha256': sha256(compressed).hexdigest(), + 'size': len(compressed)}} + outputs = {'dna.fa': {'sha256': sha256(dna).hexdigest(), 'size': len(dna)}, 'index.json': {}} + transform = {'version': 'toy-fasta-1', 'options': {'contig': 'chr1'}} + store = root / 'reference' / 'release-110' + paths = materialize(store, sources, transform=transform, outputs=outputs, builder=build_reference, + download_options={'show_progress': False}) + assert Path(paths['dna.fa']).read_bytes() == dna + assert json.loads(Path(paths['index.json']).read_text()) == {'length': 8} + assert not list(store.glob('.inputs-*')) + original.unlink() + # Neither the upstream source nor a builder is needed on an offline hit. + def forbidden(*args): + raise AssertionError('offline reuse attempted to rebuild') + assert materialize(store, sources, transform=transform, outputs=outputs, + builder=forbidden, verify_files=False) == paths + state = inspect_materialization(store, sources, transform=transform, outputs=outputs) + assert state.status == 'available' + assert not state.verified # Index has observed metadata, not a trusted hash. + assert state.sources['dna.fa.gz']['verified'] + print('Reference DNA: atomic pair, compressed provenance, input cleanup and offline reuse pass') + + +if __name__ == '__main__': + main() diff --git a/tests/test_materialization.py b/tests/test_materialization.py new file mode 100644 index 0000000..b638459 --- /dev/null +++ b/tests/test_materialization.py @@ -0,0 +1,747 @@ +"""Caller-built generations, private retained inputs, and raw gzip dependencies.""" + +from concurrent.futures import ThreadPoolExecutor +import gzip +from hashlib import sha256 +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import os +from pathlib import Path +import socket +import subprocess +import sys +import threading +import time + +import pytest + +from datacache import FileValidationError, inspect_bundle, inspect_materialization, materialize +from datacache import download, materialization as module +from datacache.resume import _state_directory + +pytestmark = pytest.mark.skipif(os.name != 'posix', reason='POSIX materialization locks') +DATA = b'>chr1\nACGTACGT\n' +COMPRESSED = gzip.compress(DATA, mtime=0) +TRANSFORM = {'version': 'fasta-1', 'options': {'validate': True}} + + +def metadata(data): + return {'sha256': sha256(data).hexdigest(), 'size': len(data)} + + +@pytest.fixture +def sources(tmp_path): + source = tmp_path / 'caller-owned.fa.gz' + source.write_bytes(COMPRESSED) + return {'dna.fa.gz': dict(path=source, **metadata(COMPRESSED))} + + +@pytest.fixture +def outputs(): + return {'dna.fa': metadata(DATA), 'index/info.json': metadata(b'{"length":8}\n')} + + +def build(sources, outputs): + assert all(Path(path).parent.stat().st_mode & 0o077 == 0 for path in sources.values()) + with gzip.open(sources['dna.fa.gz'], 'rb') as source, open(outputs['dna.fa'], 'wb') as target: + while chunk := source.read(4): + target.write(chunk) + Path(outputs['index/info.json']).write_bytes(b'{"length":8}\n') + + +def install(store, sources, outputs, **options): + defaults = dict(transform=TRANSFORM, outputs=outputs, builder=build, + download_options={'show_progress': False}) + defaults.update(options) + return materialize(store, sources, **defaults) + + +def inspected(store, sources, outputs, **options): + return inspect_materialization(store, sources, transform=TRANSFORM, outputs=outputs, **options) + + +def input_dir(store): + return next(store.glob('.inputs-*')) + + +def test_transaction_receipt_cleanup_and_immutable_paths(tmp_path, sources, outputs): + store = tmp_path / 'artifacts' / 'dna-110' + paths = install(store, sources, outputs) + assert Path(paths['dna.fa']).read_bytes() == DATA + assert Path(paths['index/info.json']).read_bytes() == b'{"length":8}\n' + state = inspected(store, sources, outputs) + assert state.status == 'available' and state.verified + assert all(file.verified for file in state.files.values()) + assert state.transform == TRANSFORM + record = state.sources['dna.fa.gz'] + assert record['verified'] is True + assert {key: record[key] for key in ('sha256', 'size')} == metadata(COMPRESSED) + receipt = json.loads((Path(state.generation) / module.MANIFEST).read_text()) + assert receipt['definition']['transform'] == TRANSFORM + assert set(receipt['outputs']) == set(outputs) + assert Path(sources['dna.fa.gz']['path']).read_bytes() == COMPRESSED + assert not list(store.glob('.inputs-*')) + assert not list(store.glob('.staging-*')) + assert inspect_bundle(store).status == 'invalid' # Distinct store ownership. + assert not inspect_materialization(store).verified + new_paths = install(store, sources, outputs, force=True) + assert paths != new_paths + assert all(Path(path).exists() for path in paths.values()) + + +def forbidden(*args, **kwargs): + raise AssertionError('offline cache hit tried to write, build, or acquire') + + +def test_readonly_offline_hit_never_touches_sources(tmp_path, sources, outputs, monkeypatch): + store = tmp_path / 'dna' + paths = install(store, sources, outputs) + Path(sources['dna.fa.gz']['path']).unlink() + changed_modes = [] + for path in [*store.rglob('*'), store]: + changed_modes.append((path, path.stat().st_mode & 0o777)) + os.chmod(path, 0o555 if path.is_dir() else 0o444) + try: + monkeypatch.setattr(download, 'fetch_file', forbidden) + monkeypatch.setattr(module, 'file_lock', forbidden) + monkeypatch.setattr(module, 'write_json', forbidden) + monkeypatch.setattr(Path, 'mkdir', forbidden) + assert install(store, sources, outputs, builder=forbidden) == paths + assert inspected(store, sources, outputs).verified + fast = inspected(store, sources, outputs, verify_files=False) + assert fast.status == 'available' and not fast.verified + assert not any(file.verified for file in fast.files.values()) + finally: + for path, mode in reversed(changed_modes): + os.chmod(path, mode) + + +@pytest.mark.parametrize('failure', ['builder', 'keyboard', 'last-output', 'validation', 'rename', 'pointer']) +def test_failed_refresh_preserves_old_and_keeps_complete_input(tmp_path, sources, outputs, monkeypatch, failure): + store = tmp_path / 'dna' + old_paths = install(store, sources, outputs) + old_pointer = (store / module.CURRENT).read_bytes() + + def failing_build(inputs, targets): + build(inputs, targets) + assert (store / module.CURRENT).read_bytes() == old_pointer + if failure == 'builder': + raise RuntimeError('biological validation failed') + if failure == 'keyboard': + raise KeyboardInterrupt + if failure == 'last-output': + Path(targets['index/info.json']).unlink() + if failure == 'validation': + Path(targets['index/info.json']).write_bytes(b'wrong') + + original_replace, original_write = module.os.replace, module.write_json + if failure == 'rename': + def fail_replace(source, target): + if Path(target).parent.name == 'generations': + raise OSError('generation rename failed') + return original_replace(source, target) + monkeypatch.setattr(module.os, 'replace', fail_replace) + if failure == 'pointer': + def fail_write(path, value, **options): + if Path(path).name == module.CURRENT: + raise OSError('pointer failed') + return original_write(path, value, **options) + monkeypatch.setattr(module, 'write_json', fail_write) + with pytest.raises((RuntimeError, KeyboardInterrupt, FileValidationError, OSError)): + install(store, sources, outputs, builder=failing_build, force=True) + assert (store / module.CURRENT).read_bytes() == old_pointer + assert inspected(store, sources, outputs).verified + assert all(Path(path).exists() for path in old_paths.values()) + assert (input_dir(store) / 'dna.fa.gz').read_bytes() == COMPRESSED + assert not list(store.glob('.staging-*')) + Path(sources['dna.fa.gz']['path']).unlink() + monkeypatch.setattr(module.os, 'replace', original_replace) + monkeypatch.setattr(module, 'write_json', original_write) + assert install(store, sources, outputs, force=True) != old_paths + assert not list(store.glob('.inputs-*')) + + +def test_missing_pointer_recovers_locally_and_only_then_discards_inputs(tmp_path, sources, outputs, monkeypatch): + store = tmp_path / 'dna' + original = module.write_json + def interrupted(path, value, **options): + if Path(path).name == module.CURRENT: + raise KeyboardInterrupt + return original(path, value, **options) + monkeypatch.setattr(module, 'write_json', interrupted) + with pytest.raises(KeyboardInterrupt): + install(store, sources, outputs) + assert inspected(store, sources, outputs).status == 'recovery-required' + assert (input_dir(store) / 'dna.fa.gz').exists() + Path(sources['dna.fa.gz']['path']).unlink() + monkeypatch.setattr(module, 'write_json', original) + monkeypatch.setattr(download, 'fetch_file', forbidden) + paths = install(store, sources, outputs, builder=forbidden) + assert Path(paths['dna.fa']).read_bytes() == DATA + assert inspected(store, sources, outputs).verified + assert not list(store.glob('.inputs-*')) + + +def test_recovery_always_hashes_outputs_even_with_fast_hit_flag(tmp_path, sources, outputs): + store = tmp_path / 'dna' + paths = install(store, sources, outputs, retain_sources=True) + (store / module.CURRENT).unlink() + Path(paths['dna.fa']).write_bytes(b'x' * len(DATA)) + repaired = install(store, sources, outputs, verify_files=False) + assert repaired != paths + assert Path(repaired['dna.fa']).read_bytes() == DATA + + +@pytest.mark.parametrize('change', ['source', 'transform-version', 'transform-options', 'output-inventory']) +def test_dependency_mismatch_is_explicit_and_does_not_touch_old(tmp_path, sources, outputs, change): + store = tmp_path / 'dna' + paths = install(store, sources, outputs, retain_sources=True) + new_sources = {name: dict(spec) for name, spec in sources.items()} + options = {} + new_outputs = outputs + if change == 'source': + replacement = tmp_path / 'another.fa.gz' + replacement.write_bytes(COMPRESSED) + new_sources['dna.fa.gz']['path'] = replacement + elif change == 'transform-version': + options['transform'] = {'version': 'fasta-2', 'options': TRANSFORM['options']} + elif change == 'transform-options': + options['transform'] = {'version': TRANSFORM['version'], 'options': {'validate': False}} + else: + new_outputs = {'dna.fa': outputs['dna.fa']} + options['builder'] = lambda inputs, targets: Path(targets['dna.fa']).write_bytes(DATA) + with pytest.raises(FileValidationError, match='force=True'): + install(store, new_sources, new_outputs, **options) + assert inspected(store, sources, outputs).verified + assert install(store, new_sources, new_outputs, force=True, **options) != paths + + +def test_transform_change_can_reuse_retained_inputs_after_upstream_removed(tmp_path, sources, outputs): + store = tmp_path / 'dna' + install(store, sources, outputs, retain_sources=True) + owned = input_dir(store) + Path(sources['dna.fa.gz']['path']).unlink() + paths = install(store, sources, outputs, force=True, transform={'version': 'fasta-2'}) + assert Path(paths['dna.fa']).read_bytes() == DATA + assert not owned.exists() + + +def test_untrusted_observation_is_not_trusted_verification(tmp_path, sources, outputs): + store = tmp_path / 'dna' + untrusted_sources = {'dna.fa.gz': {'path': sources['dna.fa.gz']['path']}} + untrusted_outputs = {name: {} for name in outputs} + paths = install(store, untrusted_sources, untrusted_outputs) + state = inspected(store, untrusted_sources, untrusted_outputs) + assert state.status == 'available' and not state.verified + assert state.sources['dna.fa.gz']['verified'] is False + assert state.sources['dna.fa.gz']['sha256'] == sha256(COMPRESSED).hexdigest() + Path(paths['dna.fa']).write_bytes(b'x' * len(DATA)) + assert inspected(store, untrusted_sources, untrusted_outputs, verify_files=False).status == 'available' + assert inspected(store, untrusted_sources, untrusted_outputs).status == 'invalid' + with pytest.raises(FileValidationError): + install(store, untrusted_sources, untrusted_outputs) + assert install(store, untrusted_sources, untrusted_outputs, force=True) != paths + + +@pytest.mark.parametrize('kind', ['symlink', 'hardlink', 'fifo', 'extra', 'extra-directory', 'input-mutation']) +def test_unsafe_or_undeclared_builder_results_are_never_published(tmp_path, sources, outputs, kind): + store = tmp_path / 'dna' + def unsafe(inputs, targets): + build(inputs, targets) + target = Path(targets['dna.fa']) + if kind in ('symlink', 'hardlink', 'fifo'): + target.unlink() + if kind == 'symlink': + target.symlink_to(sources['dna.fa.gz']['path']) + elif kind == 'hardlink': + os.link(inputs['dna.fa.gz'], target) + else: + os.mkfifo(target) + elif kind == 'extra': + (target.parent / 'undeclared').write_text('extra') + elif kind == 'extra-directory': + (target.parent / 'unused').mkdir() + else: + Path(inputs['dna.fa.gz']).write_bytes(b'changed') + with pytest.raises(FileValidationError): + install(store, sources, outputs, builder=unsafe) + assert inspected(store, sources, outputs).status == 'missing' + assert list((store / 'generations').iterdir()) == [] + assert Path(sources['dna.fa.gz']['path']).read_bytes() == COMPRESSED + + +@pytest.mark.parametrize('kind', ['symlink', 'hardlink', 'fifo', 'directory']) +def test_nonregular_local_sources_rejected_without_blocking(tmp_path, sources, outputs, kind): + target = tmp_path / 'unsafe' + if kind == 'symlink': + target.symlink_to(sources['dna.fa.gz']['path']) + elif kind == 'hardlink': + os.link(sources['dna.fa.gz']['path'], target) + elif kind == 'fifo': + os.mkfifo(target) + else: + target.mkdir() + sources['dna.fa.gz']['path'] = target + with pytest.raises(FileValidationError): + install(tmp_path / 'dna', sources, outputs) + + +@pytest.mark.parametrize('force', [False, True]) +def test_foreign_directory_never_adopted(tmp_path, sources, outputs, force): + store = tmp_path / 'foreign' + store.mkdir() + (store / 'precious').write_text('keep') + with pytest.raises((FileValidationError, OSError)): + install(store, sources, outputs, force=force) + assert (store / 'precious').read_text() == 'keep' + + +def test_complete_gzip_but_failed_decompression_reuses_input(tmp_path, sources, outputs): + invalid_gzip = b'not-gzip' + source = sources['dna.fa.gz']['path'] + source.write_bytes(invalid_gzip) + sources['dna.fa.gz'].update(metadata(invalid_gzip)) + store = tmp_path / 'dna' + with pytest.raises(gzip.BadGzipFile): + install(store, sources, outputs) + source.unlink() + owned = input_dir(store) / 'dna.fa.gz' + assert owned.read_bytes() == invalid_gzip + calls = [] + def failure(inputs, targets): + calls.append(Path(inputs['dna.fa.gz']).read_bytes()) + raise RuntimeError('retry reached builder with complete input') + with pytest.raises(RuntimeError, match='retry reached'): + install(store, sources, outputs, builder=failure) + assert calls == [invalid_gzip] and owned.exists() + + +def test_same_artifact_serializes_but_unrelated_builders_can_run(tmp_path, sources, outputs): + store = tmp_path / 'dna' + entered, release = threading.Event(), threading.Event() + calls = [] + def paused(inputs, targets): + calls.append(1) + entered.set() + assert release.wait(8) + build(inputs, targets) + with ThreadPoolExecutor(max_workers=3) as pool: + first = pool.submit(install, store, sources, outputs, builder=paused) + assert entered.wait(8) + second = pool.submit(install, store, sources, outputs, builder=paused) + unrelated = pool.submit(install, tmp_path / 'other', sources, outputs) + try: + assert unrelated.result(timeout=8) + assert not second.done() + finally: + release.set() + assert first.result(timeout=8) == second.result(timeout=8) + assert calls == [1] + + +def test_reader_sees_whole_old_pair_until_pointer_publication(tmp_path, sources, outputs, monkeypatch): + store = tmp_path / 'dna' + old = install(store, sources, outputs) + entered, release = threading.Event(), threading.Event() + original = module.write_json + def paused(path, value, **options): + if Path(path).name == module.CURRENT: + entered.set() + assert release.wait(8) + return original(path, value, **options) + monkeypatch.setattr(module, 'write_json', paused) + with ThreadPoolExecutor() as pool: + future = pool.submit(install, store, sources, outputs, force=True) + assert entered.wait(8) + try: + state = inspected(store, sources, outputs) + assert state.verified + assert {name: file.path for name, file in state.files.items()} == old + assert (input_dir(store) / 'dna.fa.gz').exists() + finally: + release.set() + new = future.result(timeout=8) + assert new != old + assert Path(new['index/info.json']).parents[1] == Path(new['dna.fa']).parent + + +def test_large_dependency_receipts_are_readable(tmp_path, sources, outputs): + store = tmp_path / 'dna' + transform = {'version': '1', 'options': 'x' * (1024 * 1024 + 1)} + paths = install(store, sources, outputs, transform=transform) + receipt = Path(paths['dna.fa']).parent / module.MANIFEST + assert receipt.stat().st_size > 1024 * 1024 + assert inspect_materialization(store, sources, transform=transform, outputs=outputs).verified + + +@pytest.fixture +def server(): + requests = [] + actions = [] + class Handler(BaseHTTPRequestHandler): + def do_GET(self): + requests.append(dict(self.headers)) + action = actions.pop(0) if actions else {} + offset = int(self.headers.get('Range', 'bytes=0-')[6:-1]) + body = COMPRESSED[offset:] + self.send_response(206 if offset else 200) + self.send_header('Content-Length', str(len(body))) + self.send_header('ETag', '"dna-110"') + if offset: + self.send_header('Content-Range', 'bytes %d-%d/%d' % (offset, len(COMPRESSED) - 1, len(COMPRESSED))) + self.end_headers() + try: + self.wfile.write(body[:action.get('drop', len(body))]) + self.wfile.flush() + if 'drop' in action: + self.connection.shutdown(socket.SHUT_RDWR) + except OSError: + pass + def log_message(self, *args): + pass + srv = ThreadingHTTPServer(('127.0.0.1', 0), Handler) + thread = threading.Thread(target=srv.serve_forever, daemon=True) + thread.start() + yield 'http://127.0.0.1:%d/dna.gz?token=secret' % srv.server_port, requests, actions + srv.shutdown() + srv.server_close() + thread.join() + + +@pytest.mark.parametrize('trusted', [True, False]) +def test_resume_gzip_and_record_validated_transport(server, tmp_path, outputs, trusted): + url, requests, actions = server + actions.append({'drop': 12}) + sources = {'dna.fa.gz': dict(url=url, size=len(COMPRESSED))} + if trusted: + sources['dna.fa.gz']['sha256'] = sha256(COMPRESSED).hexdigest() + store = tmp_path / 'dna' + paths = install(store, sources, outputs, download_options={'resume': True, 'chunk_size': 4, + 'retry_backoff': 0, 'show_progress': False}) + assert Path(paths['dna.fa']).read_bytes() == DATA + assert requests[1]['Range'] == 'bytes=12-' + assert requests[1]['If-Range'] == '"dna-110"' + state = inspected(store, sources, outputs) + record = state.sources['dna.fa.gz'] + assert record['verified'] is trusted + assert record['transport'] == {'header': 'ETag', 'value': '"dna-110"'} + assert record['origin'] == url.split('?')[0] + assert 'secret' not in (Path(state.generation) / module.MANIFEST).read_text() + assert not list(store.glob('.inputs-*')) + + +def test_interrupted_download_and_builder_retry_without_network(server, tmp_path, outputs, monkeypatch): + url, requests, actions = server + sources = {'dna.fa.gz': dict(url=url, **metadata(COMPRESSED))} + store = tmp_path / 'dna' + def interrupt(done, total): + if done >= 8: + raise KeyboardInterrupt + options = {'resume': True, 'chunk_size': 4, 'progress_callback': interrupt, 'show_progress': False} + with pytest.raises(KeyboardInterrupt): + install(store, sources, outputs, download_options=options) + owned = input_dir(store) / 'dna.fa.gz' + assert (_state_directory(owned) / 'partial').read_bytes() == COMPRESSED[:8] + assert inspected(store, sources, outputs).status == 'missing' + del options['progress_callback'] + def interrupted_build(inputs, targets): + with gzip.open(inputs['dna.fa.gz'], 'rb') as source, open(targets['dna.fa'], 'wb') as target: + target.write(source.read(4)) + raise KeyboardInterrupt + with pytest.raises(KeyboardInterrupt): + install(store, sources, outputs, builder=interrupted_build, download_options=options) + assert requests[1]['Range'] == 'bytes=8-' + assert owned.read_bytes() == COMPRESSED + assert not (_state_directory(owned) / 'partial').exists() + monkeypatch.setattr(download, 'fetch_file', forbidden) + assert install(store, sources, outputs, download_options=options) + assert len(requests) == 2 + assert not owned.exists() + + +def test_killed_builder_retries_with_completed_local_input(tmp_path, sources, outputs): + store = tmp_path / 'dna' + ready = tmp_path / 'builder-ready' + code = ''' +import json, sys, time +from pathlib import Path +from datacache import materialize +store, source, ready, outputs = sys.argv[1:] +def builder(inputs, targets): + Path(targets['dna.fa']).write_bytes(b'partial') + Path(ready).write_text('ready') + time.sleep(30) +materialize(store, {'dna.fa.gz': {'path': source}}, transform={'version':'fasta-1','options':{'validate':True}}, + outputs=json.loads(outputs), builder=builder, download_options={'show_progress':False}) +''' + process = subprocess.Popen([sys.executable, '-c', code, str(store), str(sources['dna.fa.gz']['path']), + str(ready), json.dumps(outputs)], stdout=subprocess.PIPE, stderr=subprocess.PIPE) + try: + deadline = time.monotonic() + 8 + while not ready.exists() and process.poll() is None and time.monotonic() < deadline: + time.sleep(0.02) + assert ready.exists(), process.communicate(timeout=2) + process.kill() + process.communicate(timeout=5) + assert list(store.glob('.staging-*')) + Path(sources['dna.fa.gz']['path']).unlink() + untrusted = {'dna.fa.gz': {'path': sources['dna.fa.gz']['path']}} + paths = install(store, untrusted, outputs) + assert Path(paths['dna.fa']).read_bytes() == DATA + assert not list(store.glob('.staging-*')) + assert not list(store.glob('.inputs-*')) + finally: + if process.poll() is None: + process.kill() + process.communicate(timeout=5) + + +@pytest.mark.parametrize('options', [ + {'force': 1}, {'retain_sources': 'yes'}, {'verify_files': None}, {'builder': None}, + {'transform': {'version': ''}}, {'transform': {'version': '1', 'options': float('nan')}}, + {'outputs': {'../escape': {}}}, {'outputs': {'A': {}, 'a/b': {}}}, + {'download_options': {'raw': False}}, {'download_options': {'show_progress': 1}}, + {'download_options': {'resume': 'yes'}}, + {'download_options': {'chunk_size': 0}}, {'download_options': {'progress_callback': 'yes'}}, + {'download_options': {'max_retries': -1}}, +]) +def test_invalid_options_fail_before_creating_store(tmp_path, sources, outputs, options): + store = tmp_path / 'missing' / 'dna' + with pytest.raises((ValueError, TypeError)): + install(store, sources, outputs, **options) + assert not store.parent.exists() + + +def test_fast_reuse_does_not_hash_payloads(tmp_path, sources, outputs, monkeypatch): + store = tmp_path / 'dna' + paths = install(store, sources, outputs) + original = module._observe + def no_hash(*args, **options): + assert options.get('verify_files') is False + return original(*args, **options) + monkeypatch.setattr(module, '_observe', no_hash) + assert install(store, sources, outputs, verify_files=False, builder=forbidden) == paths + + +def test_input_hash_failure_and_bad_final_output_never_publish(tmp_path, sources, outputs): + store = tmp_path / 'dna' + sources['dna.fa.gz']['sha256'] = '0' * 64 + with pytest.raises(FileValidationError): + install(store, sources, outputs, verify_files=False) + assert not list((store / 'generations').iterdir()) + sources['dna.fa.gz'].update(metadata(COMPRESSED)) + outputs['dna.fa']['sha256'] = '0' * 64 + with pytest.raises(FileValidationError): + install(store, sources, outputs, verify_files=False) + assert not list((store / 'generations').iterdir()) + + +def test_source_file_url_is_private_copy_and_not_deleted(tmp_path, sources, outputs): + source = sources['dna.fa.gz']['path'] + sources['dna.fa.gz'] = dict(url=source.as_uri(), **metadata(COMPRESSED)) + assert install(tmp_path / 'dna', sources, outputs) + assert source.read_bytes() == COMPRESSED + + +def test_private_input_corruption_requires_explicit_repair(tmp_path, sources, outputs): + store = tmp_path / 'dna' + def interrupted(inputs, targets): + raise KeyboardInterrupt + with pytest.raises(KeyboardInterrupt): + install(store, sources, outputs, builder=interrupted) + owned = input_dir(store) / 'dna.fa.gz' + owned.write_bytes(b'x' * len(COMPRESSED)) + with pytest.raises(FileValidationError, match='force=True'): + install(store, sources, outputs) + assert install(store, sources, outputs, force=True) + assert not owned.exists() + + +def test_bare_complete_input_without_receipt_can_be_explicitly_repaired(tmp_path, sources, outputs): + store = tmp_path / 'dna' + def interrupted(inputs, targets): + raise KeyboardInterrupt + with pytest.raises(KeyboardInterrupt): + install(store, sources, outputs, builder=interrupted) + directory = input_dir(store) + (directory / module.INPUTS).unlink() + (directory / 'dna.fa.gz').write_bytes(b'x' * len(COMPRESSED)) + with pytest.raises(FileValidationError, match='force=True'): + install(store, sources, outputs) + assert install(store, sources, outputs, force=True) + + +def test_untrusted_bare_input_requires_new_acquisition(tmp_path, sources, outputs): + store = tmp_path / 'dna' + sources = {'dna.fa.gz': {'path': sources['dna.fa.gz']['path']}} + def interrupted(inputs, targets): + raise KeyboardInterrupt + with pytest.raises(KeyboardInterrupt): + install(store, sources, outputs, builder=interrupted) + directory = input_dir(store) + (directory / module.INPUTS).unlink() + (directory / 'dna.fa.gz').write_bytes(b'unverified unknown') + assert install(store, sources, outputs) + + +def test_inspection_requires_complete_definition_or_none(tmp_path, sources, outputs): + with pytest.raises(ValueError): + inspect_materialization(tmp_path / 'dna', sources, outputs=outputs) + with pytest.raises(ValueError): + inspect_materialization(tmp_path / 'dna', verify_files='yes') + + +def test_precreated_empty_store_preserves_mode(tmp_path, sources, outputs): + store = tmp_path / 'dna' + store.mkdir(mode=0o750) + assert inspected(store, sources, outputs).status == 'missing' + install(store, sources, outputs) + assert store.stat().st_mode & 0o777 == 0o750 + + +def test_retention_toggle_on_hit_does_not_cleanup(tmp_path, sources, outputs): + store = tmp_path / 'dna' + paths = install(store, sources, outputs, retain_sources=True) + owned = input_dir(store) / 'dna.fa.gz' + assert install(store, sources, outputs) == paths + assert owned.exists() + assert install(store, sources, outputs, force=True) != paths + assert not owned.exists() + + +@pytest.mark.parametrize('kind', ['symlink', 'fifo', 'shared-directory']) +def test_planted_private_state_is_not_followed_or_adopted(tmp_path, sources, outputs, kind): + store = tmp_path / 'dna' + def interrupted(inputs, targets): + raise KeyboardInterrupt + with pytest.raises(KeyboardInterrupt): + install(store, sources, outputs, builder=interrupted) + owned_directory = input_dir(store) + owned = owned_directory / 'dna.fa.gz' + if kind == 'shared-directory': + os.chmod(owned_directory, 0o755) + else: + owned.unlink() + if kind == 'symlink': + owned.symlink_to(sources['dna.fa.gz']['path']) + else: + os.mkfifo(owned) + with pytest.raises(FileValidationError): + install(store, sources, outputs, force=True) + assert Path(sources['dna.fa.gz']['path']).read_bytes() == COMPRESSED + + +@pytest.mark.parametrize('malformation', [ + 'missing-output', 'extra-output', 'source-size', 'source-identity', 'source-trust', + 'source-inventory', 'transform', 'output-digest', 'transport', 'definition', +]) +def test_malformed_manifest_is_invalid_even_for_fast_inspection(tmp_path, sources, outputs, malformation): + store = tmp_path / 'dna' + paths = install(store, sources, outputs) + manifest = Path(paths['dna.fa']).parent / module.MANIFEST + receipt = json.loads(manifest.read_text()) + if malformation == 'missing-output': + del receipt['outputs']['dna.fa'] + elif malformation == 'extra-output': + receipt['outputs']['extra'] = metadata(b'') + elif malformation == 'source-size': + receipt['sources']['dna.fa.gz']['size'] += 1 + elif malformation == 'source-identity': + receipt['sources']['dna.fa.gz']['identity'] = '0' * 64 + elif malformation == 'source-trust': + receipt['sources']['dna.fa.gz']['verified'] = False + elif malformation == 'source-inventory': + receipt['sources'] = {} + elif malformation == 'transform': + receipt['definition']['transform'] = {'version': ''} + elif malformation == 'output-digest': + receipt['outputs']['dna.fa']['sha256'] = None + elif malformation == 'transport': + receipt['sources']['dna.fa.gz']['transport'] = {'header': 'ETag', 'value': 'W/"weak"'} + else: + receipt['definition'] = None + manifest.write_text(json.dumps(receipt)) + assert inspected(store, sources, outputs, verify_files=False).status == 'invalid' + assert inspect_materialization(store, verify_files=False).status == 'invalid' + + +def test_inspection_permission_error_is_inaccessible(tmp_path, sources, outputs, monkeypatch): + store = tmp_path / 'dna' + install(store, sources, outputs) + def permission_denied(*args, **kwargs): + raise PermissionError('cannot read receipt') + monkeypatch.setattr(module, 'read_json', permission_denied) + assert inspected(store, sources, outputs).status == 'inaccessible' + with pytest.raises(PermissionError): + install(store, sources, outputs) + + +@pytest.mark.parametrize('bad_sources', [ + {}, {'a': {}}, {'a': {'url': 'https://example.test', 'path': 'local'}}, + {'a': {'url': ''}}, {'a': {'url': 'unsupported://example.test'}}, + {'a': {'url': 'file://remote/file'}}, {'a': {'url': 'file:///file?query'}}, + {'a': {'url': 'https://example.test', 'decompress': True}}, + {'a': {'path': 'local'}, '.a.datacache.json': {'path': 'local'}}, + {'a': {'path': 'local'}, '.a.datacache.json/extra': {'path': 'local'}}, +]) +def test_invalid_source_definitions_fail_without_writes(tmp_path, outputs, bad_sources): + store = tmp_path / 'missing' / 'dna' + with pytest.raises(ValueError): + install(store, bad_sources, outputs) + assert not store.parent.exists() + + +def test_multiple_inputs_keep_completed_first_source_on_later_failure(tmp_path, sources, outputs, monkeypatch): + store = tmp_path / 'dna' + original = Path(sources['dna.fa.gz']['path']) + sources['auxiliary'] = {'path': tmp_path / 'missing'} + with pytest.raises(FileNotFoundError): + install(store, sources, outputs) + completed = input_dir(store) / 'dna.fa.gz' + assert completed.read_bytes() == COMPRESSED + original.unlink() + Path(sources['auxiliary']['path']).write_bytes(b'auxiliary') + paths = install(store, sources, outputs) + assert Path(paths['dna.fa']).read_bytes() == DATA + assert set(inspected(store, sources, outputs).sources) == {'dna.fa.gz', 'auxiliary'} + + +def test_explicitly_empty_input_and_output(tmp_path): + source = tmp_path / 'empty' + source.touch() + sources = {'empty': dict(path=source, **metadata(b''))} + outputs = {'empty-result': metadata(b'')} + def empty(inputs, targets): + Path(targets['empty-result']).touch() + store = tmp_path / 'artifact' + materialize(store, sources, outputs=outputs, transform={'version': '1'}, builder=empty, + download_options={'show_progress': False}) + assert inspect_materialization(store, sources, transform={'version': '1'}, outputs=outputs).verified + assert source.exists() + + +def test_progress_defaults_on_and_existing_hits_stay_quiet(tmp_path, sources, outputs, monkeypatch): + enabled = [] + original = module.Progress + def progress(enabled_flag, *args, **options): + enabled.append(enabled_flag) + return original(False, *args, **options) + monkeypatch.setattr(module, 'Progress', progress) + store = tmp_path / 'dna' + paths = install(store, sources, outputs, download_options={}) + assert any(enabled) + enabled.clear() + assert install(store, sources, outputs, download_options={}) == paths + assert not any(enabled) + + +def test_builder_paths_stay_absolute_after_working_directory_change(tmp_path, sources, outputs, monkeypatch): + monkeypatch.chdir(tmp_path) + other = tmp_path / 'other' + other.mkdir() + def changing_directory(inputs, targets): + assert all(Path(path).is_absolute() for path in [*inputs.values(), *targets.values()]) + monkeypatch.chdir(other) + build(inputs, targets) + paths = install(Path('relative') / 'dna', sources, outputs, builder=changing_directory) + assert Path(paths['dna.fa']).read_bytes() == DATA From 3e81c40957d37a752ceafb4273c3e341cd19658d Mon Sep 17 00:00:00 2001 From: Alex Rubinsteyn Date: Fri, 2 Oct 2026 10:50:42 -0400 Subject: [PATCH 2/3] Honor progress callbacks for local materialization inputs --- datacache/materialization.py | 4 +++- docs/materialization.md | 3 ++- tests/test_materialization.py | 30 ++++++++++++++++++++++++++---- 3 files changed, 31 insertions(+), 6 deletions(-) diff --git a/datacache/materialization.py b/datacache/materialization.py index 50d0379..dc66deb 100644 --- a/datacache/materialization.py +++ b/datacache/materialization.py @@ -364,10 +364,12 @@ def _acquire_inputs(directory, definition, acquisition, options, *, force=False) with os.fdopen(open_regular(source), 'rb') as handle, temporary.open('xb') as output: with Progress(options['show_progress'], 'Copying', os.fstat(handle.fileno()).st_size) as progress: size = 0 - for chunk in iter(lambda: handle.read(CHUNK_SIZE), b''): + for chunk in iter(lambda: handle.read(options.get('chunk_size', CHUNK_SIZE)), b''): output.write(chunk) size += len(chunk) progress(size, progress.total) + if options.get('progress_callback') is not None: + options['progress_callback'](size, progress.total) output.flush() os.fsync(output.fileno()) _observe(temporary, spec, show_progress=options['show_progress']) diff --git a/docs/materialization.md b/docs/materialization.md index 38c593c..1c26a27 100644 --- a/docs/materialization.md +++ b/docs/materialization.md @@ -55,7 +55,8 @@ Outputs are a nonempty mapping from relative names to optional `sha256` and supported; observation and a local consistency receipt do not authenticate these bytes. An expected size of zero supports explicitly empty inputs/outputs. Names cannot escape the store, collide by case or as files/directories, or -occupy reserved metadata/provenance paths. +occupy reserved metadata names. Sources also cannot collide with their +automatic provenance-sidecar paths. `transform` contains a nonempty opaque `version` string and optional JSON `options`. Bump this identity when parsing, validation, index format or other diff --git a/tests/test_materialization.py b/tests/test_materialization.py index b638459..8820c80 100644 --- a/tests/test_materialization.py +++ b/tests/test_materialization.py @@ -463,14 +463,15 @@ def test_killed_builder_retries_with_completed_local_input(tmp_path, sources, ou store = tmp_path / 'dna' ready = tmp_path / 'builder-ready' code = ''' -import json, sys, time +import gzip, json, sys, time from pathlib import Path from datacache import materialize store, source, ready, outputs = sys.argv[1:] def builder(inputs, targets): - Path(targets['dna.fa']).write_bytes(b'partial') - Path(ready).write_text('ready') - time.sleep(30) + with gzip.open(inputs['dna.fa.gz'], 'rb') as compressed: + Path(targets['dna.fa']).write_bytes(compressed.read(4)) + Path(ready).write_text('ready') + time.sleep(30) materialize(store, {'dna.fa.gz': {'path': source}}, transform={'version':'fasta-1','options':{'validate':True}}, outputs=json.loads(outputs), builder=builder, download_options={'show_progress':False}) ''' @@ -745,3 +746,24 @@ def changing_directory(inputs, targets): build(inputs, targets) paths = install(Path('relative') / 'dna', sources, outputs, builder=changing_directory) assert Path(paths['dna.fa']).read_bytes() == DATA + + +def test_local_copy_honors_chunk_size_and_progress_callback(tmp_path, sources, outputs): + events = [] + install(tmp_path / 'dna', sources, outputs, + download_options={'chunk_size': 4, 'progress_callback': lambda *event: events.append(event), + 'show_progress': False}) + assert events == [(count, len(COMPRESSED)) for count in range(4, len(COMPRESSED) + 1, 4)] + + +def test_interrupted_local_copy_never_modifies_original(tmp_path, sources, outputs): + store = tmp_path / 'dna' + def interrupt(done, total): + raise KeyboardInterrupt + with pytest.raises(KeyboardInterrupt): + install(store, sources, outputs, download_options={'chunk_size': 4, 'progress_callback': interrupt, + 'show_progress': False}) + assert Path(sources['dna.fa.gz']['path']).read_bytes() == COMPRESSED + assert not (input_dir(store) / 'dna.fa.gz').exists() + assert not list(input_dir(store).glob('.copy-*')) + assert install(store, sources, outputs) From 97865f2eaf973e0cd2026fa9d7b9177ff84fcb40 Mon Sep 17 00:00:00 2001 From: Alex Rubinsteyn Date: Wed, 7 Oct 2026 16:06:29 -0400 Subject: [PATCH 3/3] Address review of materialize; recover the newest generation everywhere - Recovery after a lost pointer picks the newest complete generation by the creation time its receipt records (directory mtime as a fallback), via one helper shared by bundles, archives and materialization. Generation names are random UUIDs, so the previous reverse-name order could roll a store back to older content. - force=True always rebuilds: it never recovers an older generation and no longer hashes the outputs it is about to replace. - A forced refresh copies a local source again while the original exists, so an in-place correction is picked up; once the original is gone, a retained private copy still serves. - force=True repairs a corrupt private input receipt or record instead of raising before the repair branch. - Staging cleanup never replaces a builder's error or fails a published build; a cleanup failure is logged. - Local inputs are hashed while they are copied (one read instead of three); the file mode is computed once; outputs aren't hashed a second time before publication, since the receipt was just built from their digests. - One _valid_validator rule in resume.py checks transport validators for both resumable downloads and materialization receipts. - Not here: a shared generation-store implementation and download-option validator (#103). - Merged the 1.19.0 stack (#97, #98) and master (1.18.1); README, interface table and site navigation carry the materialization guide; fixed its link to the example outside docs/. Claude-Session: https://claude.ai/code/session_016ZPLkDy3Gf7KWzJGEgVtZV --- CHANGELOG.md | 7 ++- datacache/archives.py | 5 +- datacache/bundles.py | 23 ++++++- datacache/materialization.py | 88 +++++++++++++++++++-------- datacache/resume.py | 17 ++++-- docs/materialization.md | 19 ++++-- tests/test_bundles.py | 23 +++++++ tests/test_materialization.py | 109 ++++++++++++++++++++++++++++++++++ 8 files changed, 249 insertions(+), 42 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c465575..2e775ac 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,12 @@ publish immutable output generations atomically, and discard owned inputs only after successful publication unless retention is requested. Local sources stay caller-owned; scientific validation stays in the builder (#90). - + `force=True` always rebuilds, copies local sources again while the original + exists, and repairs corrupt private input receipts. +- Recover the newest complete generation when a bundle, archive or + materialization pointer is lost. Generation names are random, so recovery + previously chose an arbitrary matching generation and could roll back to + older content; it now orders by the creation time each receipt records. - Add a `store_path(name, version)` callback to `VersionedDatasetRegistry` for exact consumer-chosen managed-store locations. It is mutually exclusive with root configuration and does not run at construction; the first lookup diff --git a/datacache/archives.py b/datacache/archives.py index fa383ca..506bff5 100644 --- a/datacache/archives.py +++ b/datacache/archives.py @@ -26,7 +26,7 @@ from filelock import FileLock from ._filesystem import open_regular, path_present, read_json, write_json -from .bundles import _relative_name +from .bundles import _newest_generations_first, _relative_name from .inspection import FileInspection from .integrity import FileValidationError, _validate_expectations from .provenance import redact_url @@ -419,7 +419,8 @@ def _initialize(path): def _recover(path, definition): - for entry in sorted((path / "generations").iterdir(), reverse=True): + for name in _newest_generations_first(path, MANIFEST, "fetched_at"): + entry = path / "generations" / name try: candidate = _inspect_generation(path, entry.name, definition) except (OSError, ValueError, KeyError, TypeError, RecursionError): diff --git a/datacache/bundles.py b/datacache/bundles.py index 301bbdc..7183ff4 100644 --- a/datacache/bundles.py +++ b/datacache/bundles.py @@ -252,9 +252,28 @@ def _file_mode(path): return _normal_creation_mode(path) +def _newest_generations_first(store, manifest, time_key): + """Generation names, newest first by the UTC time their receipt records + under time_key, else by directory modification time. Names are random + UUIDs, so their own order says nothing about age.""" + def created(entry): + try: + recorded = datetime.fromisoformat(read_json(entry / manifest, limit=None)[time_key]) + if recorded.tzinfo is None: + recorded = recorded.replace(tzinfo=timezone.utc) + return recorded.timestamp() + except (OSError, ValueError, KeyError, TypeError, RecursionError, OverflowError): + try: + return entry.lstat().st_mtime + except OSError: + return float('-inf') + entries = list((store / 'generations').iterdir()) + return [entry.name for entry in sorted(entries, key=lambda e: (created(e), e.name), reverse=True)] + + def _recover(path, assets): - generations = path / 'generations' - for entry in sorted(generations.iterdir(), reverse=True): + for name in _newest_generations_first(path, MANIFEST, 'fetched_at'): + entry = path / 'generations' / name try: candidate = _inspect_generation(path, entry.name, assets) except (OSError, ValueError, KeyError, TypeError, RecursionError): diff --git a/datacache/materialization.py b/datacache/materialization.py index dc66deb..77db025 100644 --- a/datacache/materialization.py +++ b/datacache/materialization.py @@ -4,6 +4,7 @@ from datetime import datetime, timezone import hashlib import json +import logging import os from pathlib import Path import shutil @@ -13,7 +14,9 @@ from uuid import uuid4 from ._filesystem import file_lock, open_regular, path_present, read_json, write_json -from .bundles import _directory, _file_mode, _generation, _initialize, _relative_name, _store +from .bundles import ( + _directory, _file_mode, _generation, _initialize, _newest_generations_first, _relative_name, _store, +) from .inspection import FileInspection from .integrity import FileValidationError, _validate_expectations from .progress import Progress @@ -26,6 +29,8 @@ FORMAT = 1 CHUNK_SIZE = 2 ** 20 +logger = logging.getLogger(__name__) + def _json_copy(value): # Reject non-JSON identities and NaNs, and detach from caller mutation. @@ -166,12 +171,8 @@ def _source_records(value, definition): raise ValueError('source receipt identity or trust disagrees with definition') transport = record.get('transport') if transport is not None: - from .resume import _strong_etag - if (not isinstance(transport, dict) or set(transport) != {'header', 'value'} or - transport['header'] not in ('ETag', 'Last-Modified') or - not isinstance(transport['value'], str) or - '\r' in transport['value'] or '\n' in transport['value'] or - (transport['header'] == 'ETag' and not _strong_etag(transport['value']))): + from .resume import _valid_validator + if not _valid_validator(transport): raise ValueError('invalid validated transport metadata') return value @@ -309,9 +310,17 @@ def _acquire_inputs(directory, definition, acquisition, options, *, force=False) from .download import fetch_file _private(directory) receipt_path = directory / INPUTS - records = read_json(receipt_path, limit=None) if path_present(receipt_path) else {} - if not isinstance(records, dict) or set(records) - set(definition): - raise FileValidationError(receipt_path, 'invalid private input receipt') + try: + records = read_json(receipt_path, limit=None) if path_present(receipt_path) else {} + if not isinstance(records, dict) or set(records) - set(definition): + raise FileValidationError(receipt_path, 'invalid private input receipt') + except (ValueError, RecursionError) as error: + if not force: + raise FileValidationError( + receipt_path, 'invalid private input receipt; use force=True to repair') from error + # Unrecorded inputs are checked or acquired again below. + records = {} + write_json(receipt_path, records) paths = {} for name, spec in definition.items(): target = directory / name @@ -320,13 +329,25 @@ def _acquire_inputs(directory, definition, acquisition, options, *, force=False) if not path_present(current): current.mkdir(mode=0o700) _directory(current) + kind, source = acquisition[name] + if force and kind == 'path' and name in records and path_present(Path(source)): + # The caller may have corrected a local source in place without + # changing its path, so a forced refresh copies it again while the + # original exists; once it's gone, the retained copy still serves. + del records[name] + write_json(receipt_path, records) + if path_present(target): + os.close(open_regular(target)) # Never remove a planted link/special file. + target.unlink() + provenance.remove(target) if name in records: - _source_records({name: records[name]}, {name: spec}) try: + _source_records({name: records[name]}, {name: spec}) _observe(target, {key: records[name][key] for key in ('sha256', 'size')}) - except (FileNotFoundError, FileValidationError): + except (FileNotFoundError, ValueError) as error: if not force: - raise FileValidationError(target, 'invalid private input; use force=True to repair') + raise FileValidationError( + target, 'invalid private input; use force=True to repair') from error if path_present(target): fd = open_regular(target) # Never repair a planted link/special file. os.close(fd) @@ -354,7 +375,7 @@ def _acquire_inputs(directory, definition, acquisition, options, *, force=False) raise FileValidationError(target, 'invalid private input; use force=True to repair') target.unlink() provenance.remove(target) - kind, source = acquisition[name] + copied = None if kind == 'url': fetch_file(source, destination=target, raw=True, record_provenance=True, expected_sha256=spec['sha256'], expected_size=spec['size'], **options) @@ -362,21 +383,27 @@ def _acquire_inputs(directory, definition, acquisition, options, *, force=False) temporary = directory / ('.copy-' + uuid4().hex) try: with os.fdopen(open_regular(source), 'rb') as handle, temporary.open('xb') as output: + digest = hashlib.sha256() with Progress(options['show_progress'], 'Copying', os.fstat(handle.fileno()).st_size) as progress: size = 0 for chunk in iter(lambda: handle.read(options.get('chunk_size', CHUNK_SIZE)), b''): output.write(chunk) + digest.update(chunk) # Hash while copying: one read. size += len(chunk) progress(size, progress.total) if options.get('progress_callback') is not None: options['progress_callback'](size, progress.total) output.flush() os.fsync(output.fileno()) - _observe(temporary, spec, show_progress=options['show_progress']) + copied = dict(sha256=digest.hexdigest(), size=size) + _check_observed(temporary, copied, spec) os.replace(temporary, target) finally: temporary.unlink(missing_ok=True) - observed, info = _observe(target, spec, show_progress=options['show_progress']) + if copied is not None: + observed, info = copied, os.stat(target) + else: + observed, info = _observe(target, spec, show_progress=options['show_progress']) origin_record = provenance.read(target, info) or {} record = dict(observed, origin=spec['origin'], identity=spec['identity'], verified=spec['sha256'] is not None, @@ -440,7 +467,9 @@ def materialize(destination, sources, *, transform, outputs, builder, force=Fals path = Path(destination).absolute() def inspect_current(): - return _inspect_materialization(path, definition, verify_files=verify_files) + # A forced refresh rebuilds whatever is there; it only needs to know + # whether the store is readable, not to hash the old outputs. + return _inspect_materialization(path, definition, verify_files=verify_files and not force) def check_hit(inspection): if inspection.status == 'inaccessible': @@ -461,13 +490,13 @@ def check_hit(inspection): inspection = inspect_current() if check_hit(inspection): return {name: value.path for name, value in inspection.files.items()} - if inspection.status in ('missing', 'recovery-required'): - for entry in sorted((path / 'generations').iterdir(), reverse=True): + if not force and inspection.status in ('missing', 'recovery-required'): + for name in _newest_generations_first(path, MANIFEST, 'created_at'): try: - candidate = _inspect_tree(path, _generation(path, entry.name), definition) + candidate = _inspect_tree(path, _generation(path, name), definition) except (OSError, ValueError, KeyError, TypeError, RecursionError): continue - write_json(path / CURRENT, {'generation': entry.name}, mode=_file_mode(path)) + write_json(path / CURRENT, {'generation': name}, mode=_file_mode(path)) _cleanup_inputs(path, definition, retain_sources) return {name: value.path for name, value in candidate.files.items()} inputs = _input_directory(path, definition) @@ -494,14 +523,17 @@ def check_hit(inspection): observed = {} # Reject extras/links/specials before hashing any declared output. _tree(staged, definition['outputs'], verify_files=False, allow_manifest=False) + mode = _file_mode(staged) for name, target in output_paths.items(): observed[name] = _observe(target, definition['outputs'][name], show_progress=options['show_progress'])[0] - os.chmod(target, _file_mode(staged)) + os.chmod(target, mode) receipt = dict(format=FORMAT, created_at=datetime.now(timezone.utc).isoformat(), definition=definition, sources=records, outputs=observed) - write_json(staged / MANIFEST, receipt, mode=_file_mode(staged)) - _inspect_tree(path, staged, definition) + write_json(staged / MANIFEST, receipt, mode=mode) + # The outputs were just hashed into the receipt; check its structure, + # inventory and sizes without hashing them again. + _inspect_tree(path, staged, definition, verify_files=False) generation = uuid4().hex final = path / 'generations' / generation os.replace(staged, final) @@ -511,5 +543,9 @@ def check_hit(inspection): _cleanup_inputs(path, definition, retain_sources) return {name: value.path for name, value in candidate.files.items()} finally: - _private(working) - shutil.rmtree(working) + # Cleanup never replaces the builder's error or fails a publication. + try: + _private(working) # Only ever remove our own private directory. + shutil.rmtree(working) + except (OSError, ValueError) as error: + logger.warning("Could not remove the staging directory %s: %s", working, error) diff --git a/datacache/resume.py b/datacache/resume.py index b8e130e..2636f76 100644 --- a/datacache/resume.py +++ b/datacache/resume.py @@ -61,6 +61,17 @@ def discard_partial(destination): (directory / name).unlink(missing_ok=True) +def _valid_validator(validator): + """Whether a recorded transport validator is a usable If-Range value: + {header: ETag (strong) or Last-Modified, value: one header line}.""" + return ( + isinstance(validator, dict) and set(validator) == {'header', 'value'} + and validator['header'] in ('ETag', 'Last-Modified') + and isinstance(validator['value'], str) + and '\r' not in validator['value'] and '\n' not in validator['value'] + and (validator['header'] != 'ETag' or _strong_etag(validator['value']))) + + def _strong_etag(value): # RFC 9110 entity-tag syntax, without the weak W/ prefix. Embedded quotes, # whitespace, and control characters cannot authorize an If-Range request. @@ -107,11 +118,7 @@ def download_resumable(download_url, destination, *, expected_sha256, expected_s # Unlink, never truncate an unknown path planted in the state dir. partial.unlink(missing_ok=True) validator = metadata.get('validator') - valid_validator = ( - isinstance(validator, dict) and validator.get('header') in ('ETag', 'Last-Modified') - and isinstance(validator.get('value'), str) - and '\r' not in validator['value'] and '\n' not in validator['value'] - and (validator['header'] != 'ETag' or _strong_etag(validator['value']))) + valid_validator = _valid_validator(validator) strong_validator = (valid_validator and validator['header'] == 'ETag') if ((validator is not None and not valid_validator) or (expected_sha256 is None and not strong_validator)): diff --git a/docs/materialization.md b/docs/materialization.md index 1c26a27..9fddf10 100644 --- a/docs/materialization.md +++ b/docs/materialization.md @@ -36,7 +36,7 @@ state = inspect_materialization( ) ``` -The [offline reference-DNA example](../examples/reference_dna_materialization.py) +The [offline reference-DNA example](https://github.com/openvax/datacache/blob/master/examples/reference_dna_materialization.py) is runnable with `python -m examples.reference_dna_materialization` and uses tiny gzip data without any biological-library dependency. @@ -124,8 +124,12 @@ output generation untouched. Failed builder outputs are discarded; completed inputs survive for retry without another download. A killed builder can leave private partial outputs, which the next transaction with the same definition discards. Changing only the transform can reuse retained matching inputs. -`force=True` rebuilds outputs, reuses sound inputs, and can repair corrupted -regular private inputs; it never adopts unsafe paths or foreign directories. +`force=True` always rebuilds outputs: it never selects an older generation, +even when `current.json` is missing, and it doesn't hash the outputs it replaces. +It reuses remote inputs that still match, copies a local source again while +the original exists (so an in-place correction is picked up; once the original +is gone, a retained copy still serves), and repairs corrupted private inputs or +input receipts. It never adopts unsafe paths or foreign directories. All declared outputs and `.datacache-manifest.json` are staged and checked as one tree. DataCache renames that tree into `generations/` and atomically @@ -135,9 +139,12 @@ returned snapshot paths continue to exist across later refreshes. Do not add indexes or edit files in a published generation; declare them as outputs or place independently mutable files outside the managed store. -If the first pointer publication is interrupted, explicit `materialize` can -fully check and select a matching complete local generation without fetching -inputs or calling the builder. Inspection alone does not recover anything. +If the first pointer publication is interrupted, explicit `materialize` +without `force` can fully check and select a matching complete local +generation without fetching inputs or calling the builder. It tries the newest +generation first, by the creation time its receipt records. Inspection alone +does not recover anything. A failure to remove the private staging directory is +logged; it never replaces a builder's error or fails a published build. After successful pointer publication/recovery, owned inputs for that source definition are removed by default. `retain_sources=True` keeps them privately diff --git a/tests/test_bundles.py b/tests/test_bundles.py index 449ca24..46e60f6 100644 --- a/tests/test_bundles.py +++ b/tests/test_bundles.py @@ -6,6 +6,7 @@ import os from pathlib import Path import threading +import time import pytest @@ -519,3 +520,25 @@ def pause_after_missing_marker(path, *args, **kwargs): resume_inspection.set() assert future.result() == winner assert len(list((dest / 'generations').iterdir())) == 1 + + +def test_recovery_picks_the_newest_generation_not_the_highest_name(tmp_path, assets): + dest = tmp_path / 'bundle' + older = Path(next(iter(install_bundle(dest, assets).values()))).parent + newer = Path(next(iter(install_bundle(dest, assets, force=True).values()))).parent + # Generation names are random: make the older one's name sort highest. + older.rename(dest / 'generations' / ('f' * 32)) + newer.rename(dest / 'generations' / ('0' * 32)) + (dest / bundles.CURRENT).unlink() + recovered = install_bundle(dest, assets) + generations = {Path(path).relative_to(dest / 'generations').parts[0] for path in recovered.values()} + assert generations == {'0' * 32} + + +def test_generation_age_falls_back_to_modification_time(tmp_path): + store = tmp_path / 'store' + for name, age in [('a' * 32, 100), ('b' * 32, 10)]: + directory = store / 'generations' / name + directory.mkdir(parents=True) + os.utime(directory, (time.time() - age, time.time() - age)) + assert bundles._newest_generations_first(store, bundles.MANIFEST, 'fetched_at') == ['b' * 32, 'a' * 32] diff --git a/tests/test_materialization.py b/tests/test_materialization.py index 8820c80..20bed5b 100644 --- a/tests/test_materialization.py +++ b/tests/test_materialization.py @@ -767,3 +767,112 @@ def interrupt(done, total): assert not (input_dir(store) / 'dna.fa.gz').exists() assert not list(input_dir(store).glob('.copy-*')) assert install(store, sources, outputs) + + +# Review fixes + + +def unpinned(tmp_path, data): + source = tmp_path / 'local.fa.gz' + source.write_bytes(gzip.compress(data, mtime=0)) + return source, {'dna.fa.gz': dict(path=source)} + + +def unpinned_outputs(): + return {'dna.fa': {}, 'index/info.json': {}} + + +def test_recovery_picks_the_newest_generation_not_the_highest_name(tmp_path): + source, sources = unpinned(tmp_path, DATA) + store = tmp_path / 'dna' + older = Path(install(store, sources, unpinned_outputs())['dna.fa']).parent + source.write_bytes(gzip.compress(b'>chr1\nTTTTTTTT\n', mtime=0)) + newer = Path(install(store, sources, unpinned_outputs(), force=True)['dna.fa']).parent + # Names are random: make the older generation's name sort highest. + older.rename(store / 'generations' / ('f' * 32)) + newer.rename(store / 'generations' / ('0' * 32)) + (store / module.CURRENT).unlink() + recovered = install(store, sources, unpinned_outputs()) + assert Path(recovered['dna.fa']).parent.name == '0' * 32 + assert Path(recovered['dna.fa']).read_bytes() == b'>chr1\nTTTTTTTT\n' + + +def test_forced_refresh_copies_an_edited_local_source_again(tmp_path): + source, sources = unpinned(tmp_path, DATA) + store = tmp_path / 'dna' + install(store, sources, unpinned_outputs(), retain_sources=True) + source.write_bytes(gzip.compress(b'>chr1\nTTTTTTTT\n', mtime=0)) + paths = install(store, sources, unpinned_outputs(), force=True) + assert Path(paths['dna.fa']).read_bytes() == b'>chr1\nTTTTTTTT\n' + state = inspect_materialization(store) + assert state.sources['dna.fa.gz']['sha256'] == sha256(source.read_bytes()).hexdigest() + + +def test_force_repairs_a_corrupt_private_input_receipt(tmp_path, sources, outputs): + import shutil + store = tmp_path / 'dna' + install(store, sources, outputs, retain_sources=True) + # No generations left, so the next call has to acquire inputs again. + (store / module.CURRENT).unlink() + for generation in (store / 'generations').iterdir(): + shutil.rmtree(generation) + (input_dir(store) / module.INPUTS).write_text('{') + with pytest.raises(FileValidationError, match='use force=True to repair'): + install(store, sources, outputs) + paths = install(store, sources, outputs, force=True) + assert Path(paths['dna.fa']).read_bytes() == DATA + + +def test_force_rebuilds_instead_of_recovering_a_lost_pointer(tmp_path, sources, outputs): + store = tmp_path / 'dna' + old = install(store, sources, outputs) + (store / module.CURRENT).unlink() + built = [] + + def counting_build(inputs, targets): + built.append(True) + build(inputs, targets) + + new = install(store, sources, outputs, builder=counting_build, force=True) + assert built == [True] and new != old + (store / module.CURRENT).unlink() + assert install(store, sources, outputs, builder=counting_build) == new # Plain call recovers. + assert built == [True] + + +def test_forced_refresh_does_not_hash_the_old_outputs(tmp_path, sources, outputs, monkeypatch): + store = tmp_path / 'dna' + install(store, sources, outputs) + modes = [] + original = module._inspect_materialization + + def recording(path, expected, *, verify_files=True): + modes.append(verify_files) + return original(path, expected, verify_files=verify_files) + + monkeypatch.setattr(module, '_inspect_materialization', recording) + install(store, sources, outputs, force=True) + assert modes and not any(modes) + + +@pytest.mark.parametrize('builder_fails', [True, False]) +def test_cleanup_failure_never_masks_the_outcome(tmp_path, sources, outputs, monkeypatch, caplog, builder_fails): + store = tmp_path / 'dna' + original = module.shutil.rmtree + + def failing_rmtree(path, *args, **kwargs): + if Path(path).name.startswith('.staging-'): + raise OSError('busy') + return original(path, *args, **kwargs) + + monkeypatch.setattr(module.shutil, 'rmtree', failing_rmtree) + + def failing_build(inputs, targets): + raise RuntimeError('builder failed') + + if builder_fails: + with pytest.raises(RuntimeError, match='builder failed'): + install(store, sources, outputs, builder=failing_build) + else: + assert Path(install(store, sources, outputs)['dna.fa']).read_bytes() == DATA + assert 'Could not remove the staging directory' in caplog.text