#!/usr/bin/env python3 """Conservative scratch recovery. Caller MUST hold the shared backup lock. Only private, recognized backup scratch is disposable. Recovery journals and manual restore extractions always win over attempts to reclaim disk space. """ import argparse import json import os import sys from pathlib import Path import re import shutil import subprocess import backup_set def private(path): return (path.is_dir() and path.resolve() == path and path.stat().st_uid == 0 and not path.stat().st_mode & 0o077) def mounted_paths(): # st_dev/ismount alone miss same-filesystem bind mounts. Linux mountinfo # lists file and directory mounts, escaping whitespace and backslashes. return {Path(re.sub(r'\\([0-7]{3})', lambda m: chr(int(m[1], 8)), line.split()[4])) for line in Path('/proc/self/mountinfo').read_text().splitlines()} def remove_tree(path): if not private(path): raise ValueError('Refusing non-private/aliased scratch: ' + str(path)) if os.path.ismount(path) or any(p == path or path in p.parents for p in mounted_paths()): raise ValueError('Refusing mounted scratch: ' + str(path)) device = path.stat().st_dev for parent, dirs, files in os.walk(path, followlinks=False): for name in dirs + files: child = Path(parent) / name if not child.is_symlink() and (child.stat().st_dev != device or os.path.ismount(child)): raise ValueError('Refusing mounted scratch: ' + str(child)) shutil.rmtree(path) # fd-based, symlink-safe implementation on Linux def clean(work, staging, archives, prefix, restore=None): for path in (work, staging, archives): if not private(path): raise ValueError('Cleanup requires private canonical directories: ' + str(path)) if list(work.glob('recovery-*.json')) or list(work.glob('recovery-*.json.tmp')): raise ValueError('Unresolved service/restore recovery journal; preserve all scratch and review services') pattern = re.compile(re.escape(prefix) + r'_\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}_[0-9a-f]{8}') for path in sorted(staging.iterdir()): run, dot, suffix = path.name.rpartition('.') if not (dot and re.fullmatch(r'[a-zA-Z0-9]{6}', suffix) and pattern.fullmatch(run) and private(path)): continue identity, version = path/'meta/run-id.txt', path/'meta/format.txt' if any(p.is_symlink() or not p.is_file() or p.stat().st_size > 256 for p in (identity, version)): continue if (path/'meta').is_symlink() or identity.read_text() != run+'\n' or not version.read_text().startswith('format=2\n'): continue partial = archives / ('.partial-' + run) if partial.exists(): # A completion marker might be the only surviving complete snapshot. if (partial/'COMPLETE').exists() or (partial/'COMPLETE').is_symlink(): continue remove_tree(partial) print('-- Removed abandoned partial archive: ' + partial.name, flush=True) remove_tree(path) print('-- Removed abandoned backup staging: ' + path.name, flush=True) if restore is not None and private(restore): for path in sorted(restore.iterdir()): run, _, suffix = path.name.rpartition('.') marker = path/'.verify-only' if (private(path) and pattern.fullmatch(run) and re.fullmatch(r'[a-zA-Z0-9]{6}', suffix) and marker.is_file() and not marker.is_symlink() and marker.stat().st_size < 256 and marker.read_text() == run+'\n'): remove_tree(path) print('-- Removed abandoned verification download: ' + path.name, flush=True) def uploaded_sets(archives, prefix): if not private(archives): raise ValueError('Cleanup requires private canonical archives: ' + str(archives)) valid = [] for path in archives.iterdir(): if not private(path) or not path.name.startswith(prefix+'_'): continue marker = path/'UPLOADED' if not marker.is_file() or marker.is_symlink(): continue try: manifest = backup_set.load_set(path, metadata_only=True) if manifest['run_id'] != path.name: continue except (ValueError, OSError, KeyError, TypeError): continue valid.append(path.name) return sorted(valid, reverse=True) def eligible(archives, prefix, keep, protected=False): if keep < 1: raise ValueError('At least one uploaded local set must be retained') # Pressure policy is independent of normal age/count retention. valid = uploaded_sets(archives, prefix) for name in (valid[:keep] if protected else valid[keep:][::-1]): print(name) def pending(archives, prefix): if not private(archives): raise ValueError('Pending uploads require private canonical archives') for path in sorted(archives.iterdir()): if not private(path) or not path.name.startswith(prefix+'_') or (path/'UPLOADED').exists() or (path/'UPLOADED').is_symlink(): continue try: data = backup_set.load_set(path, metadata_only=True) if data['run_id'] == path.name: print(path.name) except (ValueError, OSError, KeyError, TypeError): continue def remote_valid(folder, name): data = backup_set.load_set(folder, metadata_only=True) if data['run_id'] != name: raise ValueError('Remote directory/manifest identity mismatch') entries = json.loads((folder/'listing.json').read_text()) expected = {a['file']: a['size'] for a in data['archives']} expected.update({p: (folder/p).stat().st_size for p in ('manifest.json', 'COMPLETE')}) actual = {} for entry in entries: path = entry['Path'] if path in actual or entry['IsDir'] or type(entry['Size']) is not int: raise ValueError('Invalid remote file inventory') actual[path] = entry['Size'] if actual != expected: raise ValueError('Remote files missing, unexpected or wrong sizes') def quota(): try: data = json.load(sys.stdin) except (ValueError, TypeError): sys.exit(1) free = data.get('free') if isinstance(data, dict) else None if type(free) is not int or not 0 <= free <= 10**18: sys.exit(1) print(free) def inventory(archives, prefix, keep): valid = uploaded_sets(archives, prefix) protected = set(valid[:keep]) print('-- Remaining local archive inventory (allocated bytes; no deletion):', flush=True) for path in sorted(archives.iterdir()): if not private(path): print(f' {path.name}: protected unknown/aliased/non-private entry', flush=True) continue if path.name in protected: reason = 'protected local minimum' elif path.name in valid: reason = 'uploaded; retained (verification/source/device constraints)' else: reason = 'protected unconfirmed/incomplete/unknown set' result = subprocess.run(['du', '-sx', '-B1', '--', str(path)], text=True, capture_output=True, check=True) size = int(result.stdout.split()[0]) print(f' {path.name}: {size} bytes; {reason}', flush=True) if __name__ == '__main__': parser = argparse.ArgumentParser() parser.add_argument('mode', choices=('clean', 'eligible', 'protected', 'inventory', 'remove', 'pending', 'remote-valid', 'quota')) parser.add_argument('paths', nargs='*') args = parser.parse_args() if args.mode == 'quota': quota() elif args.mode == 'remote-valid': remote_valid(Path(args.paths[0]), args.paths[1]) elif args.mode == 'pending': pending(Path(args.paths[0]), args.paths[1]) elif args.mode == 'clean': clean(*(Path(p) for p in args.paths[:3]), args.paths[3], Path(args.paths[4]) if len(args.paths) > 4 else None) elif args.mode in ('eligible', 'protected'): eligible(Path(args.paths[0]), args.paths[1], int(args.paths[2]), args.mode == 'protected') elif args.mode == 'inventory': inventory(Path(args.paths[0]), args.paths[1], int(args.paths[2])) else: remove_tree(Path(args.paths[0]))