diff --git a/CHANGELOG.md b/CHANGELOG.md index ffdbe2fb..73d72895 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). ## [Unreleased] +### Added +- Show stage-level tqdm progress for long-running standalone offline verify and restore operations. +- Add `--verbose` diagnostics for object roles, BASE/DELTA details, byte counts, state digests, and timings. + +### Changed +- Report object verification progress as an object count instead of summed compressed-plus-logical bytes, avoiding misleading apparent repository sizes. + ## [0.1.4] - 2026-10-06 ### Changed diff --git a/README.md b/README.md index 899c5ffc..217ec459 100644 --- a/README.md +++ b/README.md @@ -189,6 +189,10 @@ endkeep-offline-env/bin/python endkeep-offline.py --repo /path/to/backups/repo v endkeep-offline-env/bin/python endkeep-offline.py --repo /path/to/backups/repo restore /path/to/new-world ``` +Long-running `verify` and `restore` operations show stage-level tqdm progress by default. Add +`--verbose` to either command to include object roles, BASE/DELTA details, logical/compressed +byte counts, state digests, and per-step timings. + To restore a specific recovery point: ```bash diff --git a/requirements-offline.txt b/requirements-offline.txt index f3434e7d..0b38f794 100644 --- a/requirements-offline.txt +++ b/requirements-offline.txt @@ -1,2 +1,3 @@ amulet-leveldb==3.0.7a0 zstandard>=0.23,<1 +tqdm>=4.70,<5 diff --git a/src/endstone_endkeep/logical/amulet_reader.py b/src/endstone_endkeep/logical/amulet_reader.py index 71df970c..77ef29cc 100644 --- a/src/endstone_endkeep/logical/amulet_reader.py +++ b/src/endstone_endkeep/logical/amulet_reader.py @@ -1,6 +1,6 @@ from __future__ import annotations -from collections.abc import Generator, Iterable, Iterator +from collections.abc import Callable, Generator, Iterable, Iterator from contextlib import contextmanager from pathlib import Path @@ -33,6 +33,7 @@ def write_fresh_leveldb( state: Iterable[tuple[bytes, bytes]], *, batch_records: int = 4096, + progress: Callable[[int], None] | None = None, ) -> tuple[int, int]: """Build a fresh Bedrock LevelDB from canonical visible state.""" @@ -56,10 +57,16 @@ def write_fresh_leveldb( value_bytes += len(value) previous = key if len(batch) >= batch_records: + completed = len(batch) db.put_batch(batch) batch.clear() + if progress is not None: + progress(completed) if batch: + completed = len(batch) db.put_batch(batch) + if progress is not None: + progress(completed) finally: db.close() return records, value_bytes diff --git a/src/endstone_endkeep/logical/merge.py b/src/endstone_endkeep/logical/merge.py index b35dd52d..b977138a 100644 --- a/src/endstone_endkeep/logical/merge.py +++ b/src/endstone_endkeep/logical/merge.py @@ -1,7 +1,7 @@ from __future__ import annotations import hashlib -from collections.abc import Iterable, Iterator +from collections.abc import Callable, Iterable, Iterator from dataclasses import dataclass from .format import DeltaOperation, encode_uvarint @@ -61,10 +61,19 @@ class StateStats: sha256: str -def hash_state(state: Iterable[tuple[bytes, bytes]]) -> StateStats: +def hash_state( + state: Iterable[tuple[bytes, bytes]], + *, + progress: Callable[[int], None] | None = None, + progress_interval: int = 16384, +) -> StateStats: + if progress_interval <= 0: + raise ValueError("progress_interval must be positive") + hasher = hashlib.sha256() records = 0 value_bytes = 0 + pending_progress = 0 previous: bytes | None = None for key, value in state: if previous is not None and key <= previous: @@ -76,4 +85,14 @@ def hash_state(state: Iterable[tuple[bytes, bytes]]) -> StateStats: previous = key records += 1 value_bytes += len(value) + + if progress is not None: + pending_progress += 1 + if pending_progress >= progress_interval: + progress(pending_progress) + pending_progress = 0 + + if progress is not None and pending_progress: + progress(pending_progress) + return StateStats(records, value_bytes, hasher.hexdigest()) diff --git a/src/endstone_endkeep/offline/cli.py b/src/endstone_endkeep/offline/cli.py index ae2ffac3..b5826443 100644 --- a/src/endstone_endkeep/offline/cli.py +++ b/src/endstone_endkeep/offline/cli.py @@ -4,6 +4,8 @@ import json import os import shutil +import sys +import time from collections.abc import Sequence from pathlib import Path @@ -12,7 +14,7 @@ from ..logical.sidecar import extract_sidecar from ..repository.lock import RepositoryLock from ..repository.manifest import ManifestStore, RepositoryManifest, SnapshotNode -from ..repository.objects import ObjectStore +from ..repository.objects import ObjectMetadata, ObjectStore from ..repository.reader import RepositoryReader from ..repository.verify import RepositoryVerifier @@ -39,25 +41,123 @@ def _resolve_node(manifest: RepositoryManifest, snapshot: str | None) -> Snapsho raise KeyError(f"snapshot not found: {snapshot}") +def _new_progress_bar( + *, + total: int, + desc: str, + unit: str, + unit_scale: bool = False, +): + try: + from tqdm import tqdm + except ImportError as exc: + raise RuntimeError("tqdm is required for offline progress display; install requirements-offline.txt") from exc + + return tqdm( + total=total, + desc=desc, + unit=unit, + unit_scale=unit_scale, + dynamic_ncols=True, + mininterval=0.2, + file=sys.stderr, + leave=True, + ) + + +def _progress_write(message: str) -> None: + try: + from tqdm import tqdm + except ImportError as exc: + raise RuntimeError("tqdm is required for offline progress display; install requirements-offline.txt") from exc + tqdm.write(message, file=sys.stderr) + + +def _stage(index: int, total: int, message: str) -> None: + print(f"[{index}/{total}] {message}", file=sys.stderr) + + +def _stage_done() -> None: + print("done", file=sys.stderr) + + +def _verbose(message: str, *, enabled: bool) -> None: + if enabled: + _progress_write(f" {message}") + + +def _unique_referenced_objects(manifest: RepositoryManifest) -> list[ObjectMetadata]: + result: list[ObjectMetadata] = [] + checked: set[str] = set() + for node in manifest.chain: + for metadata in (node.object, node.sidecar): + if metadata.logical_sha256 in checked: + continue + checked.add(metadata.logical_sha256) + result.append(metadata) + return result + + +def _restore_dependencies( + manifest: RepositoryManifest, + target: SnapshotNode, +) -> list[tuple[SnapshotNode, str, ObjectMetadata]]: + target_index = next(index for index, node in enumerate(manifest.chain) if node.snapshot == target.snapshot) + result: list[tuple[SnapshotNode, str, ObjectMetadata]] = [] + checked: set[str] = set() + + for node in manifest.chain[: target_index + 1]: + metadata = node.object + if metadata.logical_sha256 in checked: + continue + checked.add(metadata.logical_sha256) + result.append((node, node.type.upper(), metadata)) + + if target.sidecar.logical_sha256 not in checked: + result.append((target, "SIDECAR", target.sidecar)) + + return result + + def _verify_restore_dependencies( manifest: RepositoryManifest, objects: ObjectStore, target: SnapshotNode, + *, + verbose: bool, + show_progress: bool, ) -> None: """Deep-verify only the immutable objects required to restore one snapshot.""" manifest.validate_chain() - target_index = next(index for index, node in enumerate(manifest.chain) if node.snapshot == target.snapshot) - checked: set[str] = set() + required = _restore_dependencies(manifest, target) + bar = None + if show_progress: + _stage(1, 5, "Verifying required objects") + bar = _new_progress_bar( + total=len(required), + desc="Verifying objects", + unit="obj", + ) - for node in manifest.chain[: target_index + 1]: - metadata = node.object - if metadata.logical_sha256 not in checked: + try: + for node, role, metadata in required: + started = time.perf_counter() + _verbose( + f"{role} snapshot={node.snapshot} object={metadata.logical_sha256} " + f"logical_bytes={metadata.logical_bytes} compressed_bytes={metadata.compressed_bytes}", + enabled=verbose, + ) objects.verify(metadata) - checked.add(metadata.logical_sha256) - - if target.sidecar.logical_sha256 not in checked: - objects.verify(target.sidecar) + if bar is not None: + bar.update(1) + _verbose( + f"verified object={metadata.logical_sha256[:16]} elapsed={time.perf_counter() - started:.3f}s", + enabled=verbose, + ) + finally: + if bar is not None: + bar.close() def command_list(repo: Path, *, as_json: bool) -> int: @@ -89,18 +189,133 @@ def command_list(repo: Path, *, as_json: bool) -> int: return 0 -def command_verify(repo: Path) -> int: +def command_verify( + repo: Path, + *, + verbose: bool = False, + show_progress: bool = False, +) -> int: manifests, objects, _reader = _runtime(repo) with RepositoryLock(repo): - report = RepositoryVerifier(manifests, objects).verify(deep=True) + manifest = _load_manifest(manifests) + + if show_progress: + print("EndKeep repository verification", file=sys.stderr) + print(f"Repository: {repo}", file=sys.stderr) + print(f"Generation: {manifest.generation}", file=sys.stderr) + print(f"Snapshots: {len(manifest.chain)}", file=sys.stderr) + print(file=sys.stderr) + _stage(1, 2, "Verifying objects") + + unique_objects = _unique_referenced_objects(manifest) + object_bar = ( + _new_progress_bar( + total=len(unique_objects), + desc="Verifying objects", + unit="obj", + ) + if show_progress + else None + ) + object_started: dict[str, float] = {} + object_completed: set[str] = set() + state_bar = None + state_snapshot: str | None = None + state_started = 0.0 + state_stage_printed = False + + def on_object( + node: SnapshotNode, + role: str, + metadata: ObjectMetadata, + current: int, + total: int, + ) -> None: + key = metadata.logical_sha256 + if current == 0: + object_started[key] = time.perf_counter() + _verbose( + f"{role} snapshot={node.snapshot} object={key} " + f"logical_bytes={metadata.logical_bytes} compressed_bytes={metadata.compressed_bytes}", + enabled=verbose, + ) + if current >= total and key not in object_completed: + object_completed.add(key) + if object_bar is not None: + object_bar.update(1) + _verbose( + f"verified object={key[:16]} elapsed={time.perf_counter() - object_started.pop(key):.3f}s", + enabled=verbose, + ) + + def on_state(node: SnapshotNode, current: int, total: int) -> None: + nonlocal state_bar, state_snapshot, state_started, state_stage_printed + if state_snapshot != node.snapshot: + if state_bar is not None: + state_bar.close() + if show_progress and not state_stage_printed: + if object_bar is not None: + object_bar.close() + print(file=sys.stderr) + _stage(2, 2, "Verifying logical states") + state_stage_printed = True + state_snapshot = node.snapshot + state_started = time.perf_counter() + _verbose( + f"{node.type.upper()} snapshot={node.snapshot} records={node.records} " + f"value_bytes={node.value_bytes} state_sha256={node.state_sha256}", + enabled=verbose, + ) + state_bar = ( + _new_progress_bar( + total=total, + desc=node.snapshot, + unit="rec", + unit_scale=True, + ) + if show_progress + else None + ) + + if state_bar is not None and current > state_bar.n: + state_bar.update(current - int(state_bar.n)) + if current >= total: + if state_bar is not None: + state_bar.close() + state_bar = None + _verbose( + f"verified state={node.snapshot} elapsed={time.perf_counter() - state_started:.3f}s", + enabled=verbose, + ) + + try: + report = RepositoryVerifier(manifests, objects).verify( + deep=True, + object_progress=on_object if show_progress or verbose else None, + state_progress=on_state if show_progress or verbose else None, + ) + finally: + if object_bar is not None: + object_bar.close() + if state_bar is not None: + state_bar.close() + + print("PASS") print( - f"PASS generation={report.generation} snapshots={report.snapshots} " + f"generation={report.generation} snapshots={report.snapshots} " f"objects={report.referenced_objects} orphans={report.orphan_objects}" ) return 0 -def command_restore(repo: Path, destination: Path, snapshot: str | None) -> int: +def command_restore( + repo: Path, + destination: Path, + snapshot: str | None, + *, + verbose: bool = False, + show_progress: bool = False, +) -> int: if destination.exists(): raise FileExistsError(f"restore destination already exists: {destination}") @@ -109,28 +324,101 @@ def command_restore(repo: Path, destination: Path, snapshot: str | None) -> int: manifest = _load_manifest(manifests) node = _resolve_node(manifest, snapshot) - # A restore only depends on the BASE/DELTA prefix through the selected - # snapshot plus that snapshot's SIDECAR. Later recovery points must not - # prevent restoring an older intact point. The reconstructed DB is still - # reopened and checked against the target state digest below. - _verify_restore_dependencies(manifest, objects, node) + if show_progress: + print("EndKeep restore", file=sys.stderr) + print(f"Snapshot: {node.snapshot}", file=sys.stderr) + print(f"Captured: {node.captured_at}", file=sys.stderr) + print(f"World: {node.world_name}", file=sys.stderr) + print(f"Destination: {destination}", file=sys.stderr) + print(f"Records: {node.records}", file=sys.stderr) + print(file=sys.stderr) + + _verify_restore_dependencies( + manifest, + objects, + node, + verbose=verbose, + show_progress=show_progress, + ) destination.parent.mkdir(parents=True, exist_ok=True) destination.mkdir() try: + started = time.perf_counter() + if show_progress: + print(file=sys.stderr) + _stage(2, 5, "Restoring SIDECAR") with objects.open_logical(node.sidecar) as stream: - extract_sidecar(stream, destination, strip_prefix=node.world_name) + sidecar = extract_sidecar(stream, destination, strip_prefix=node.world_name) + _verbose( + f"SIDECAR files={sidecar.files} bytes={sidecar.bytes} elapsed={time.perf_counter() - started:.3f}s", + enabled=verbose, + ) + if show_progress: + _stage_done() state = reader.iter_state(manifest, snapshot=node.snapshot) - records, value_bytes = write_fresh_leveldb(destination / "db", state) + if show_progress: + print(file=sys.stderr) + _stage(3, 5, "Rebuilding LevelDB") + started = time.perf_counter() + rebuild_bar = ( + _new_progress_bar( + total=node.records, + desc="Rebuilding LevelDB", + unit="rec", + unit_scale=True, + ) + if show_progress + else None + ) + try: + records, value_bytes = write_fresh_leveldb( + destination / "db", + state, + progress=rebuild_bar.update if rebuild_bar is not None else None, + ) + finally: + if rebuild_bar is not None: + rebuild_bar.close() + _verbose( + f"rebuilt records={records} value_bytes={value_bytes} elapsed={time.perf_counter() - started:.3f}s", + enabled=verbose, + ) if records != node.records or value_bytes != node.value_bytes: raise RuntimeError( f"restored DB stats mismatch: records={records}/{node.records} " f"value_bytes={value_bytes}/{node.value_bytes}" ) - with iter_visible_state(destination / "db") as reopened: - reopened_stats = hash_state(reopened) + if show_progress: + print(file=sys.stderr) + _stage(4, 5, "Verifying restored LevelDB") + started = time.perf_counter() + verify_bar = ( + _new_progress_bar( + total=node.records, + desc="Verifying LevelDB", + unit="rec", + unit_scale=True, + ) + if show_progress + else None + ) + try: + with iter_visible_state(destination / "db") as reopened: + reopened_stats = hash_state( + reopened, + progress=verify_bar.update if verify_bar is not None else None, + ) + finally: + if verify_bar is not None: + verify_bar.close() + _verbose( + f"verified records={reopened_stats.records} value_bytes={reopened_stats.value_bytes} " + f"state_sha256={reopened_stats.sha256} elapsed={time.perf_counter() - started:.3f}s", + enabled=verbose, + ) if ( reopened_stats.sha256 != node.state_sha256 or reopened_stats.records != node.records @@ -140,14 +428,25 @@ def command_restore(repo: Path, destination: Path, snapshot: str | None) -> int: f"restored world state digest mismatch: {reopened_stats.sha256} != {node.state_sha256}" ) + if show_progress: + print(file=sys.stderr) + _stage(5, 5, "Flushing restored world") + started = time.perf_counter() _fsync_tree(destination) + _verbose( + f"fsync destination={destination} elapsed={time.perf_counter() - started:.3f}s", + enabled=verbose, + ) + if show_progress: + _stage_done() except Exception: shutil.rmtree(destination, ignore_errors=True) raise + print("RESTORED") print( - f"RESTORED snapshot={node.snapshot} world={node.world_name} " - f"records={node.records} state_sha256={node.state_sha256} destination={destination}" + f"snapshot={node.snapshot} world={node.world_name} records={node.records} " + f"state_sha256={node.state_sha256} destination={destination}" ) return 0 @@ -202,7 +501,13 @@ def build_parser() -> argparse.ArgumentParser: list_parser = subparsers.add_parser("list", help="List committed recovery points") list_parser.add_argument("--json", action="store_true", help="Emit machine-readable JSON") - subparsers.add_parser("verify", help="Deep-verify objects and logical state digests") + verify_parser = subparsers.add_parser("verify", help="Deep-verify objects and logical state digests") + verify_parser.add_argument( + "-v", + "--verbose", + action="store_true", + help="Show object, BASE/DELTA, byte-count, and timing details", + ) restore_parser = subparsers.add_parser("restore", help="Restore a snapshot into a fresh world directory") restore_parser.add_argument( @@ -215,6 +520,12 @@ def build_parser() -> argparse.ArgumentParser: default="latest", help="Snapshot ID to restore (default: latest)", ) + restore_parser.add_argument( + "-v", + "--verbose", + action="store_true", + help="Show object, BASE/DELTA, byte-count, and timing details", + ) return parser @@ -226,9 +537,19 @@ def main(argv: Sequence[str] | None = None) -> int: if args.command == "list": return command_list(repo, as_json=args.json) if args.command == "verify": - return command_verify(repo) + return command_verify( + repo, + verbose=args.verbose, + show_progress=True, + ) if args.command == "restore": - return command_restore(repo, args.destination.resolve(), args.snapshot) + return command_restore( + repo, + args.destination.resolve(), + args.snapshot, + verbose=args.verbose, + show_progress=True, + ) parser.error(f"unknown command: {args.command}") return 2 diff --git a/src/endstone_endkeep/repository/verify.py b/src/endstone_endkeep/repository/verify.py index 0ac756fa..0c1a3d44 100644 --- a/src/endstone_endkeep/repository/verify.py +++ b/src/endstone_endkeep/repository/verify.py @@ -1,14 +1,18 @@ from __future__ import annotations +from collections.abc import Callable from dataclasses import dataclass from endstone_endkeep.logical.merge import hash_state from .gc import referenced_object_hashes -from .manifest import ManifestStore, RepositoryManifest +from .manifest import ManifestStore, RepositoryManifest, SnapshotNode from .objects import ObjectMetadata, ObjectStore from .reader import RepositoryReader +type ObjectProgressCallback = Callable[[SnapshotNode, str, ObjectMetadata, int, int], None] +type StateProgressCallback = Callable[[SnapshotNode, int, int], None] + class VerificationError(RuntimeError): pass @@ -29,7 +33,13 @@ def __init__(self, manifests: ManifestStore, objects: ObjectStore) -> None: self.objects = objects self.reader = RepositoryReader(objects) - def verify(self, *, deep: bool = False) -> VerifyReport: + def verify( + self, + *, + deep: bool = False, + object_progress: ObjectProgressCallback | None = None, + state_progress: StateProgressCallback | None = None, + ) -> VerifyReport: manifest = self.manifests.load_current() if manifest is None: return VerifyReport(None, 0, 0, self._count_object_files(), deep) @@ -40,7 +50,11 @@ def verify(self, *, deep: bool = False) -> VerifyReport: orphans = existing - referenced if deep: - self._verify_deep(manifest) + self._verify_deep( + manifest, + object_progress=object_progress, + state_progress=state_progress, + ) return VerifyReport( generation=manifest.generation, @@ -68,16 +82,48 @@ def _verify_metadata(self, metadata: ObjectMetadata) -> None: f"{path.stat().st_size} != {metadata.compressed_bytes}" ) - def _verify_deep(self, manifest: RepositoryManifest) -> None: + def _verify_deep( + self, + manifest: RepositoryManifest, + *, + object_progress: ObjectProgressCallback | None, + state_progress: StateProgressCallback | None, + ) -> None: checked: set[str] = set() for node in manifest.chain: - for metadata in (node.object, node.sidecar): - if metadata.logical_sha256 not in checked: - self.objects.verify(metadata) - checked.add(metadata.logical_sha256) + for role, metadata in ((node.type.upper(), node.object), ("SIDECAR", node.sidecar)): + if metadata.logical_sha256 in checked: + continue + + if object_progress is not None: + object_progress(node, role, metadata, 0, 1) + self.objects.verify(metadata) + if object_progress is not None: + object_progress(node, role, metadata, 1, 1) + checked.add(metadata.logical_sha256) for node in manifest.chain: - stats = hash_state(self.reader.iter_state(manifest, snapshot=node.snapshot)) + completed_records = 0 + if state_progress is not None: + state_progress(node, 0, node.records) + + def advance_state( + amount: int, + current_node: SnapshotNode = node, + ) -> None: + nonlocal completed_records + completed_records += amount + if state_progress is not None: + state_progress( + current_node, + completed_records, + current_node.records, + ) + + stats = hash_state( + self.reader.iter_state(manifest, snapshot=node.snapshot), + progress=advance_state if state_progress is not None else None, + ) if ( stats.sha256 != node.state_sha256 or stats.records != node.records diff --git a/tests/test_offline_progress.py b/tests/test_offline_progress.py new file mode 100644 index 00000000..2c45d88c --- /dev/null +++ b/tests/test_offline_progress.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +from pathlib import Path + +from endstone_endkeep.logical.amulet_reader import write_fresh_leveldb +from endstone_endkeep.repository.manifest import ManifestStore +from endstone_endkeep.repository.objects import ObjectStore +from endstone_endkeep.repository.verify import RepositoryVerifier +from tests.standalone_fixture import build_fixture + + +def test_deep_verify_reports_object_and_state_progress(tmp_path: Path) -> None: + storage = tmp_path / "backups" + build_fixture(storage) + repo = storage / "repo" + + manifest = ManifestStore(repo).load_current() + assert manifest is not None + + objects = ObjectStore(repo, compression_level=6, compression_threads=1) + object_progress: dict[str, tuple[int, int]] = {} + state_progress: dict[str, tuple[int, int]] = {} + + def on_object(node, role, metadata, current: int, total: int) -> None: + del node, role + object_progress[metadata.logical_sha256] = (current, total) + + def on_state(node, current: int, total: int) -> None: + state_progress[node.snapshot] = (current, total) + + report = RepositoryVerifier(ManifestStore(repo), objects).verify( + deep=True, + object_progress=on_object, + state_progress=on_state, + ) + + assert report.snapshots == len(manifest.chain) + assert object_progress + assert all(current == total == 1 for current, total in object_progress.values()) + assert state_progress == {node.snapshot: (node.records, node.records) for node in manifest.chain} + + +def test_fresh_leveldb_reports_committed_record_progress(tmp_path: Path) -> None: + updates: list[int] = [] + records, value_bytes = write_fresh_leveldb( + tmp_path / "db", + [(b"a", b"1"), (b"b", b"22"), (b"c", b"333")], + batch_records=2, + progress=updates.append, + ) + + assert records == 3 + assert value_bytes == 6 + assert updates == [2, 1]