Buckets:
| """Apply a reviewed output-layout plan without rewriting historical artifacts.""" | |
| from __future__ import annotations | |
| import argparse | |
| from contextlib import ExitStack | |
| import fcntl | |
| import hashlib | |
| import json | |
| import os | |
| from pathlib import Path | |
| import stat | |
| import time | |
| def inside(root: Path, relative: str) -> Path: | |
| part = Path(relative) | |
| if part.is_absolute() or '..' in part.parts or not part.parts: | |
| raise ValueError(f'unsafe relative path: {relative}') | |
| path = root / part | |
| if path.resolve() != path: | |
| raise ValueError(f'symlinked migration path: {path}') | |
| return path | |
| def relocate_path(path: Path, root: Path, records: list[dict]) -> Path: | |
| for row in sorted(records, key=lambda r: len(r['source']), reverse=True): | |
| source = root / row['source'] | |
| if path == source or source in path.parents: | |
| return root / row['destination'] / path.relative_to(source) | |
| return path | |
| def inventory(run: Path): | |
| for directory, dirs, files in os.walk(run, followlinks=False): | |
| for path in [Path(directory), *(Path(directory) / name for name in dirs + files)]: | |
| info = path.lstat() | |
| if stat.S_ISDIR(info.st_mode) and path != Path(directory): | |
| continue | |
| row = dict(path=str(path.relative_to(run)), inode=info.st_ino, | |
| device=info.st_dev, mode=info.st_mode, size=info.st_size, | |
| mtime_ns=info.st_mtime_ns, links=info.st_nlink) | |
| if stat.S_ISLNK(info.st_mode): | |
| row.update(target=os.readlink(path), existed=path.exists()) | |
| yield row | |
| def fingerprint(rows): | |
| digest = hashlib.sha256() | |
| count = total = 0 | |
| for row in sorted(rows, key=lambda r: r['path']): | |
| if stat.S_ISDIR(row['mode']): | |
| fields = (row['path'], row['inode'], row['device'], row['mode']) | |
| else: | |
| fields = tuple((key, row[key]) for key in sorted(row) if key != 'existed') | |
| count += 1 | |
| total += row['size'] | |
| digest.update(repr(fields).encode()) | |
| return dict(sha256=digest.hexdigest(), files=count, logical_bytes=total) | |
| def apply(plan_path: Path, expected_hash: str) -> dict: | |
| raw = plan_path.read_bytes() | |
| if hashlib.sha256(raw).hexdigest() != expected_hash: | |
| raise ValueError('reviewed plan hash changed') | |
| plan = json.loads(raw) | |
| root = Path(plan['output_root']) | |
| if root.name != 'output' or root.parent.name != 'experiments' or root.resolve() != root: | |
| raise ValueError('expected a real experiments/output directory') | |
| records = plan['records'] | |
| audit = plan_path.parent | |
| for row in records: | |
| if not row.get('destination') or not row.get('owner'): | |
| raise ValueError(f'unclassified run: {row["source"]}') | |
| source, target = inside(root, row['source']), inside(root, row['destination']) | |
| if not source.is_dir() or target.exists() or target == source: | |
| raise ValueError(f'missing source or occupied destination: {source} -> {target}') | |
| if len(target.relative_to(root).parts) != 6: | |
| raise ValueError(f'noncanonical destination: {target}') | |
| sources = [root / row['source'] for row in records] | |
| for source in sources: | |
| if any(other != source and other in source.parents for other in sources): | |
| raise ValueError(f'nested migration roots: {source}') | |
| before = {} | |
| links = [] | |
| with ExitStack() as locks: | |
| with (audit / 'inventory-before.jsonl').open('x') as out: | |
| for row in records: | |
| source = root / row['source'] | |
| entries = list(inventory(source)) | |
| for entry in entries: | |
| if Path(entry['path']).name == '.active-writer.lock': | |
| handle = locks.enter_context((source / entry['path']).open('rb')) | |
| fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB) | |
| if 'target' in entry: | |
| links.append((source / entry['path'], entry)) | |
| before[row['source']] = fingerprint(entries) | |
| for entry in entries: | |
| out.write(json.dumps(dict(run=row['source'], **entry)) + '\n') | |
| print('INVENTORIED', len(records), 'runs', sum(x['files'] for x in before.values()), 'files', flush=True) | |
| with (audit / 'renamed.jsonl').open('x') as journal: | |
| for row in records: | |
| source, target = root / row['source'], root / row['destination'] | |
| target.parent.mkdir(parents=True, exist_ok=True) | |
| if target.exists(): | |
| raise FileExistsError(target) | |
| os.rename(source, target) | |
| check = fingerprint(list(inventory(target))) | |
| if check != before[row['source']]: | |
| raise RuntimeError(f'artifact changed while relocating: {source}') | |
| journal.write(json.dumps(dict(**row, verified=check, at=time.time())) + '\n') | |
| journal.flush() | |
| os.fsync(journal.fileno()) | |
| repaired = [] | |
| for original, entry in links: | |
| new_link = relocate_path(original, root, records) | |
| old_target = Path(entry['target']) | |
| resolved = old_target if old_target.is_absolute() else Path(os.path.normpath(original.parent / old_target)) | |
| resolved = relocate_path(resolved, root, records) | |
| target = str(resolved) if old_target.is_absolute() else os.path.relpath(resolved, new_link.parent) | |
| if target != entry['target']: | |
| if '_submission' in original.parts: | |
| raise RuntimeError(f'frozen snapshot symlink needs explicit review: {original}') | |
| temporary = new_link.with_name(new_link.name + '.relocating') | |
| if os.path.lexists(temporary): | |
| raise FileExistsError(temporary) | |
| os.symlink(target, temporary) | |
| os.replace(temporary, new_link) | |
| repaired.append(dict(path=str(new_link), before=entry['target'], after=target)) | |
| if entry['existed'] and not new_link.exists(): | |
| raise RuntimeError(f'newly broken symlink: {new_link}') | |
| (audit / 'symlink-repairs.json').write_text(json.dumps(repaired, indent=2) + '\n') | |
| empty_parents = {parent for source in sources for parent in source.parents if parent != root and root in parent.parents} | |
| removed_empty = [] | |
| for directory in sorted(empty_parents, key=lambda p: len(p.parts), reverse=True): | |
| try: | |
| directory.rmdir() | |
| removed_empty.append(str(directory.relative_to(root))) | |
| except OSError: | |
| pass | |
| result = dict(runs=len(records), files=sum(x['files'] for x in before.values()), | |
| logical_bytes=sum(x['logical_bytes'] for x in before.values()), | |
| symlinks_repaired=len(repaired), removed_empty_directories=removed_empty, | |
| owners=dict(__import__('collections').Counter(r['owner'] for r in records)), | |
| verification='Every non-directory entry retained inode, mode, size and mtime across rename; directories retained inode and mode; symlink repairs journaled separately.', | |
| completed_at=time.time()) | |
| (audit / 'result.json').write_text(json.dumps(result, indent=2) + '\n') | |
| return result | |
| def main(): | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument('--plan', type=Path, required=True) | |
| parser.add_argument('--apply-sha256', required=True) | |
| args = parser.parse_args() | |
| print(json.dumps(apply(args.plan, args.apply_sha256)), flush=True) | |
| if __name__ == '__main__': | |
| main() | |
Xet Storage Details
- Size:
- 7.69 kB
- Xet hash:
- 5fec06a5e8b4a9e786cba3de8b1a8ef23d6c5342ea2dd1d5364b03cbc2be2577
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.