diff --git a/README.md b/README.md index b4a04da..17b0c64 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,13 @@ bidirectional nonce/acknowledgement files. Route attempt nonces, ownership, and ordered updates survive restart; accepted attempts resume on reconnect without concurrent duplicate execution. Conflicting existing folders and foreign data are reported without being overwritten. +The transfer storage foundation now publishes isolated per-job namespaces with +canonical protobuf-JSON manifests and digest-bound ready markers. Selected +regular files and metainfo artifacts are materialized with hardlink, reflink, +then sparse-aware copy fallback; every intent/result is journaled in SQLite. +Receiver materialization never overwrites a path, can reuse a same-size regular +file for later qBittorrent verification, and replays completed operations +idempotently after a database reopen. ```bash archive-client --config /etc/archive-control/client.toml --check-config @@ -44,10 +51,12 @@ archive-client-backup --database /var/lib/archive-control/client.db \ Secrets must be regular files without group/world permissions. The daemon never stores them in SQLite or sends the shared token after registration. -This foundation currently executes heartbeat, inventory, state-snapshot, and -route-provisioning commands. Other mutation commands are durably rejected as -unsupported until their service and file-operation executors are added; they -are never falsely acknowledged as accepted. +The daemon currently executes heartbeat, inventory, state-snapshot, and +route-provisioning commands. Transfer storage primitives are implemented and +tested but are not yet wired to assignment/step commands; those and eviction +commands remain durably rejected as unsupported until their orchestration and +qBittorrent executors are added. They are never falsely acknowledged as +accepted. Run tests and build using containers: diff --git a/src/archive_clients/state.py b/src/archive_clients/state.py index 898d3cd..358cf6e 100644 --- a/src/archive_clients/state.py +++ b/src/archive_clients/state.py @@ -22,6 +22,10 @@ class JobConflict(RuntimeError): pass +class FileOperationConflict(RuntimeError): + pass + + @dataclass(frozen=True) class CommandAcceptance: duplicate: bool @@ -183,6 +187,87 @@ class ClientStore: ), ) + def begin_file_operation( + self, + operation_id: str, + job_id: str, + intent_json: str, + ) -> dict[str, object]: + intent = _canonical(json.loads(intent_json)) + with self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + existing = connection.execute( + "SELECT * FROM file_journal WHERE operation_id = ?", + (operation_id,), + ).fetchone() + if existing: + if ( + existing["job_id"] != job_id + or existing["intent_json"] != intent + ): + raise FileOperationConflict( + "file operation ID was reused with different intent" + ) + return dict(existing) + connection.execute( + """ + INSERT INTO file_journal ( + operation_id, job_id, intent_json, state + ) VALUES (?, ?, ?, 'intent') + """, + (operation_id, job_id, intent), + ) + row = connection.execute( + "SELECT * FROM file_journal WHERE operation_id = ?", + (operation_id,), + ).fetchone() + return dict(row) + + def complete_file_operation( + self, + operation_id: str, + result_json: str, + ) -> dict[str, object]: + result = _canonical(json.loads(result_json)) + with self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + existing = connection.execute( + "SELECT * FROM file_journal WHERE operation_id = ?", + (operation_id,), + ).fetchone() + if existing is None: + raise FileOperationConflict("file operation intent does not exist") + if existing["state"] == "completed": + if existing["result_json"] != result: + raise FileOperationConflict( + "completed file operation has a different result" + ) + return dict(existing) + connection.execute( + """ + UPDATE file_journal + SET result_json = ?, state = 'completed' + WHERE operation_id = ? + """, + (result, operation_id), + ) + row = connection.execute( + "SELECT * FROM file_journal WHERE operation_id = ?", + (operation_id,), + ).fetchone() + return dict(row) + + def file_operation_rows(self, job_id: str) -> list[dict[str, object]]: + with self._connect() as connection: + rows = connection.execute( + """ + SELECT operation_id, job_id, intent_json, result_json, state + FROM file_journal WHERE job_id = ? ORDER BY operation_id + """, + (job_id,), + ).fetchall() + return [dict(row) for row in rows] + def begin_route_attempt( self, command_id: str, diff --git a/src/archive_clients/transfer.py b/src/archive_clients/transfer.py new file mode 100644 index 0000000..29d49d0 --- /dev/null +++ b/src/archive_clients/transfer.py @@ -0,0 +1,713 @@ +"""Safe, journaled transfer staging and target materialization.""" + +from __future__ import annotations + +import errno +import fcntl +import hashlib +import json +import os +import stat +import uuid +from dataclasses import dataclass +from pathlib import Path, PurePosixPath +from typing import Mapping + +from google.protobuf import json_format + +from archive_clients.state import ClientStore +from archive_control.v1 import transfer_pb2 + + +_FICLONE = 0x40049409 +_COPY_CHUNK_BYTES = 1024 * 1024 +_FALLBACK_ERRNOS = { + errno.EXDEV, + errno.EPERM, + errno.EACCES, + errno.EMLINK, + errno.ENOSYS, + errno.EINVAL, + errno.ENOTTY, + errno.EOPNOTSUPP, +} + + +class TransferError(RuntimeError): + pass + + +class UnsafeTransferPath(TransferError): + pass + + +class TransferCollision(TransferError): + pass + + +class TransferIntegrityError(TransferError): + pass + + +@dataclass(frozen=True) +class MaterializedFile: + method: int + logical_bytes: int + allocated_bytes: int + sparse: bool + destination_preexisted: bool + destination_device: int + destination_inode: int + + +@dataclass(frozen=True) +class PublishedTransfer: + job_directory: Path + manifest_path: Path + ready_path: Path + manifest_sha256_hex: str + manifest: transfer_pb2.TransferManifest + + +class FileMaterializer: + """Materialize one regular file without overwriting an existing path.""" + + def __init__( + self, + store: ClientStore, + *, + allow_hardlink: bool = True, + allow_reflink: bool = True, + copy_chunk_bytes: int = _COPY_CHUNK_BYTES, + ): + if copy_chunk_bytes < 1: + raise ValueError("copy_chunk_bytes must be positive") + self.store = store + self.allow_hardlink = allow_hardlink + self.allow_reflink = allow_reflink + self.copy_chunk_bytes = copy_chunk_bytes + + def materialize( + self, + *, + job_id: str, + operation_id: str, + source_root: Path, + source_relative_path: str, + destination_root: Path, + destination_relative_path: str, + expected_bytes: int, + allow_preexisting_reuse: bool, + sparse_supported: bool, + ) -> MaterializedFile: + source_relative = _relative_path(source_relative_path) + destination_relative = _relative_path(destination_relative_path) + source = _existing_regular_file(source_root, source_relative) + source_stat = source.stat(follow_symlinks=False) + if source_stat.st_size != expected_bytes: + raise TransferIntegrityError( + f"source size is {source_stat.st_size}, expected {expected_bytes}" + ) + source_sparse = _is_sparse(source_stat) + if source_sparse and not sparse_supported: + raise TransferError( + "source is sparse but the destination filesystem lacks sparse support" + ) + destination = _prepare_destination( + destination_root, destination_relative + ) + intent = { + "destination": destination_relative.as_posix(), + "destination_root_device": destination_root.stat().st_dev, + "destination_root_inode": destination_root.stat().st_ino, + "expected_bytes": expected_bytes, + "source": source_relative.as_posix(), + "source_device": source_stat.st_dev, + "source_inode": source_stat.st_ino, + "source_mtime_ns": source_stat.st_mtime_ns, + } + row = self.store.begin_file_operation( + operation_id, job_id, _canonical_json(intent).decode() + ) + if row["state"] == "completed": + result = json.loads(str(row["result_json"])) + replayed = _result_from_json(result) + _verify_completed_destination(destination, replayed) + return replayed + + if destination.exists() or destination.is_symlink(): + if not allow_preexisting_reuse: + raise TransferCollision("destination already exists") + existing = destination.lstat() + if not stat.S_ISREG(existing.st_mode): + raise TransferCollision( + "pre-existing destination is not a regular file" + ) + if existing.st_size != expected_bytes: + raise TransferCollision( + "pre-existing destination size does not match" + ) + result = MaterializedFile( + transfer_pb2.MATERIALIZATION_METHOD_PREEXISTING_REUSED, + existing.st_size, + _allocated_bytes(existing), + _is_sparse(existing), + True, + existing.st_dev, + existing.st_ino, + ) + self._complete(operation_id, result) + return result + + method: int | None = None + if self.allow_hardlink: + try: + os.link(source, destination, follow_symlinks=False) + method = transfer_pb2.MATERIALIZATION_METHOD_HARD_LINK + except OSError as exc: + if exc.errno == errno.EEXIST: + raise TransferCollision("destination appeared concurrently") from exc + if exc.errno not in _FALLBACK_ERRNOS: + raise + if method is None and self.allow_reflink: + try: + _reflink(source, destination, operation_id) + method = transfer_pb2.MATERIALIZATION_METHOD_REFLINK + except OSError as exc: + if exc.errno == errno.EEXIST: + raise TransferCollision("destination appeared concurrently") from exc + if exc.errno not in _FALLBACK_ERRNOS: + raise + if method is None: + _sparse_copy( + source, + destination, + operation_id, + self.copy_chunk_bytes, + ) + method = transfer_pb2.MATERIALIZATION_METHOD_COPY + + created = destination.stat(follow_symlinks=False) + if created.st_size != expected_bytes: + raise TransferIntegrityError( + "materialized destination has an unexpected size" + ) + result = MaterializedFile( + method, + created.st_size, + _allocated_bytes(created), + _is_sparse(created), + False, + created.st_dev, + created.st_ino, + ) + self._complete(operation_id, result) + return result + + def _complete( + self, operation_id: str, result: MaterializedFile + ) -> None: + self.store.complete_file_operation( + operation_id, + _canonical_json( + { + "allocated_bytes": result.allocated_bytes, + "destination_preexisted": result.destination_preexisted, + "destination_device": result.destination_device, + "destination_inode": result.destination_inode, + "logical_bytes": result.logical_bytes, + "method": result.method, + "sparse": result.sparse, + } + ).decode(), + ) + + +def stage_transfer( + manifest: transfer_pb2.TransferManifest, + *, + source_root: Path, + sync_root: Path, + store: ClientStore, + artifact_sources: Mapping[str, Path] | None = None, + allow_hardlink: bool = True, + allow_reflink: bool = True, + sparse_supported: bool = True, +) -> PublishedTransfer: + """Publish a complete isolated transfer namespace and ready marker.""" + + _validate_manifest(manifest) + staged = transfer_pb2.TransferManifest() + staged.CopyFrom(manifest) + job_directory = _job_directory(sync_root, staged.job_id) + materializer = FileMaterializer( + store, + allow_hardlink=allow_hardlink, + allow_reflink=allow_reflink, + ) + for entry in staged.files: + result = materializer.materialize( + job_id=staged.job_id, + operation_id=_operation_id( + staged.job_id, "source-stage", entry.file_index, + entry.payload_relative_path, + ), + source_root=source_root, + source_relative_path=entry.target_canonical_path, + destination_root=job_directory, + destination_relative_path=entry.payload_relative_path, + expected_bytes=entry.logical_bytes, + allow_preexisting_reuse=False, + sparse_supported=sparse_supported, + ) + entry.allocated_bytes = result.allocated_bytes + entry.sparse = result.sparse + entry.source_staging_method = result.method + + artifact_sources = artifact_sources or {} + for artifact in staged.artifacts: + source = artifact_sources.get(artifact.payload_relative_path) + if source is None: + raise TransferIntegrityError( + f"artifact source is missing: {artifact.payload_relative_path}" + ) + result = materializer.materialize( + job_id=staged.job_id, + operation_id=_operation_id( + staged.job_id, "artifact-stage", artifact.kind, + artifact.payload_relative_path, + ), + source_root=source.parent, + source_relative_path=source.name, + destination_root=job_directory, + destination_relative_path=artifact.payload_relative_path, + expected_bytes=artifact.logical_bytes, + allow_preexisting_reuse=False, + sparse_supported=sparse_supported, + ) + artifact.allocated_bytes = result.allocated_bytes + artifact.sparse = result.sparse + digest = _sha256_file( + _existing_regular_file( + job_directory, _relative_path(artifact.payload_relative_path) + ) + ) + if artifact.sha256_hex and artifact.sha256_hex != digest: + raise TransferIntegrityError("artifact SHA-256 does not match") + artifact.sha256_hex = digest + + manifest_bytes = canonical_message_json(staged) + manifest_path = job_directory / "manifest.json" + _atomic_create_or_verify(manifest_path, manifest_bytes) + digest = hashlib.sha256(manifest_bytes).hexdigest() + ready = transfer_pb2.ReadyMarker( + job_id=staged.job_id, + manifest_sha256_hex=digest, + ) + ready.ready_at.GetCurrentTime() + ready_path = job_directory / "ready.json" + if ready_path.exists(): + existing = load_published_transfer(job_directory) + if existing.manifest_sha256_hex != digest: + raise TransferCollision("existing ready marker describes another manifest") + return existing + _atomic_create_or_verify(ready_path, canonical_message_json(ready)) + return PublishedTransfer( + job_directory, + manifest_path, + ready_path, + digest, + staged, + ) + + +def load_published_transfer(job_directory: Path) -> PublishedTransfer: + manifest_path = job_directory / "manifest.json" + ready_path = job_directory / "ready.json" + manifest_bytes = _read_regular_file(manifest_path) + ready_bytes = _read_regular_file(ready_path) + ready = transfer_pb2.ReadyMarker() + manifest = transfer_pb2.TransferManifest() + try: + json_format.Parse(ready_bytes.decode(), ready) + json_format.Parse(manifest_bytes.decode(), manifest) + except (UnicodeDecodeError, json_format.ParseError) as exc: + raise TransferIntegrityError("published transfer JSON is invalid") from exc + digest = hashlib.sha256(manifest_bytes).hexdigest() + if ready.job_id != manifest.job_id: + raise TransferIntegrityError("ready marker job ID does not match manifest") + if ready.manifest_sha256_hex != digest: + raise TransferIntegrityError("ready marker manifest digest does not match") + _validate_manifest(manifest) + return PublishedTransfer( + job_directory, + manifest_path, + ready_path, + digest, + manifest, + ) + + +def materialize_transfer( + published: PublishedTransfer, + *, + target_root: Path, + store: ClientStore, + allow_hardlink: bool = True, + allow_reflink: bool = True, + sparse_supported: bool = True, +) -> transfer_pb2.TransferManifest: + """Materialize a verified published payload into a target content root.""" + + verified = load_published_transfer(published.job_directory) + result_manifest = transfer_pb2.TransferManifest() + result_manifest.CopyFrom(verified.manifest) + materializer = FileMaterializer( + store, + allow_hardlink=allow_hardlink, + allow_reflink=allow_reflink, + ) + for entry in result_manifest.files: + result = materializer.materialize( + job_id=result_manifest.job_id, + operation_id=_operation_id( + result_manifest.job_id, "target-materialize", + entry.file_index, entry.target_canonical_path, + ), + source_root=verified.job_directory, + source_relative_path=entry.payload_relative_path, + destination_root=target_root, + destination_relative_path=entry.target_canonical_path, + expected_bytes=entry.logical_bytes, + allow_preexisting_reuse=True, + sparse_supported=sparse_supported, + ) + entry.target_method = result.method + entry.target_preexisted = result.destination_preexisted + return result_manifest + + +def canonical_message_json(message: object) -> bytes: + value = json_format.MessageToDict( + message, + preserving_proto_field_name=False, + always_print_fields_with_no_presence=True, + ) + return _canonical_json(value) + + +def _validate_manifest(manifest: transfer_pb2.TransferManifest) -> None: + if manifest.manifest_version != 1: + raise TransferIntegrityError("unsupported transfer manifest version") + try: + uuid.UUID(manifest.job_id) + except (ValueError, AttributeError) as exc: + raise TransferIntegrityError("manifest job ID is not a UUID") from exc + if not manifest.source_client_id or not manifest.target_client_id: + raise TransferIntegrityError("manifest source and target are required") + if not manifest.route_id: + raise TransferIntegrityError("manifest route is required") + if not manifest.HasField("created_at"): + raise TransferIntegrityError("manifest creation timestamp is required") + indices: set[int] = set() + payload_paths: set[str] = set() + for entry in manifest.files: + if entry.file_index in indices: + raise TransferIntegrityError("manifest contains duplicate file indices") + indices.add(entry.file_index) + payload = _relative_path(entry.payload_relative_path).as_posix() + _relative_path(entry.target_canonical_path) + if payload in payload_paths: + raise TransferIntegrityError("manifest contains duplicate payload paths") + payload_paths.add(payload) + for artifact in manifest.artifacts: + payload = _relative_path(artifact.payload_relative_path).as_posix() + if payload in payload_paths: + raise TransferIntegrityError("manifest contains duplicate payload paths") + payload_paths.add(payload) + + +def _relative_path(value: str) -> PurePosixPath: + if ( + not value + or "\x00" in value + or "\\" in value + or value.startswith("/") + or any(part in {"", ".", ".."} for part in value.split("/")) + ): + raise UnsafeTransferPath("transfer path is not a safe relative POSIX path") + path = PurePosixPath(value) + if path.is_absolute(): + raise UnsafeTransferPath("transfer path must be relative") + return path + + +def _job_directory(sync_root: Path, job_id: str) -> Path: + try: + uuid.UUID(job_id) + except ValueError as exc: + raise TransferIntegrityError("job ID is not a UUID") from exc + return _prepare_directory( + sync_root, PurePosixPath(".archive-control", "jobs", job_id) + ) + + +def _prepare_destination(root: Path, relative: PurePosixPath) -> Path: + root_stat = root.stat(follow_symlinks=False) + if not stat.S_ISDIR(root_stat.st_mode): + raise UnsafeTransferPath("configured root is not a directory") + parent = _prepare_directory(root, PurePosixPath(*relative.parts[:-1])) + return parent / relative.name + + +def _prepare_directory(root: Path, relative: PurePosixPath) -> Path: + current = root + root_stat = current.stat(follow_symlinks=False) + if not stat.S_ISDIR(root_stat.st_mode): + raise UnsafeTransferPath("configured root is not a directory") + for component in relative.parts: + current = current / component + try: + metadata = current.lstat() + except FileNotFoundError: + current.mkdir(mode=0o700) + metadata = current.lstat() + if not stat.S_ISDIR(metadata.st_mode): + raise UnsafeTransferPath("transfer parent is not a real directory") + return current + + +def _existing_regular_file(root: Path, relative: PurePosixPath) -> Path: + current = root + if not stat.S_ISDIR(current.stat(follow_symlinks=False).st_mode): + raise UnsafeTransferPath("configured root is not a directory") + for component in relative.parts[:-1]: + current = current / component + metadata = current.lstat() + if not stat.S_ISDIR(metadata.st_mode): + raise UnsafeTransferPath("source parent is not a real directory") + result = current / relative.name + metadata = result.lstat() + if not stat.S_ISREG(metadata.st_mode): + raise UnsafeTransferPath("source is not a regular file") + return result + + +def _read_regular_file(path: Path) -> bytes: + metadata = path.lstat() + if not stat.S_ISREG(metadata.st_mode): + raise TransferIntegrityError("transfer metadata is not a regular file") + return path.read_bytes() + + +def _reflink(source: Path, destination: Path, operation_id: str) -> None: + temporary = _temporary_path(destination, operation_id) + source_fd = os.open(source, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) + destination_fd = -1 + try: + destination_fd = os.open( + temporary, + os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), + stat.S_IMODE(source.stat(follow_symlinks=False).st_mode), + ) + fcntl.ioctl(destination_fd, _FICLONE, source_fd) + os.fsync(destination_fd) + _publish_temporary(temporary, destination) + finally: + os.close(source_fd) + if destination_fd >= 0: + os.close(destination_fd) + temporary.unlink(missing_ok=True) + + +def _sparse_copy( + source: Path, + destination: Path, + operation_id: str, + chunk_bytes: int, +) -> None: + temporary = _temporary_path(destination, operation_id) + source_fd = os.open(source, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) + destination_fd = -1 + try: + source_stat = os.fstat(source_fd) + destination_fd = os.open( + temporary, + os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), + stat.S_IMODE(source_stat.st_mode), + ) + _copy_extents(source_fd, destination_fd, source_stat.st_size, chunk_bytes) + os.ftruncate(destination_fd, source_stat.st_size) + os.fsync(destination_fd) + _publish_temporary(temporary, destination) + finally: + os.close(source_fd) + if destination_fd >= 0: + os.close(destination_fd) + temporary.unlink(missing_ok=True) + + +def _copy_extents( + source_fd: int, + destination_fd: int, + size: int, + chunk_bytes: int, +) -> None: + if size == 0: + return + try: + position = 0 + while position < size: + try: + data_offset = os.lseek(source_fd, position, os.SEEK_DATA) + except OSError as exc: + if exc.errno == errno.ENXIO: + break + raise + hole_offset = os.lseek(source_fd, data_offset, os.SEEK_HOLE) + os.lseek(source_fd, data_offset, os.SEEK_SET) + os.lseek(destination_fd, data_offset, os.SEEK_SET) + remaining = min(hole_offset, size) - data_offset + while remaining: + data = os.read(source_fd, min(chunk_bytes, remaining)) + if not data: + raise TransferIntegrityError("source ended during sparse copy") + _write_all(destination_fd, data) + remaining -= len(data) + position = hole_offset + except OSError as exc: + if exc.errno not in {errno.EINVAL, errno.ENOTSUP, errno.ENOSYS}: + raise + os.lseek(source_fd, 0, os.SEEK_SET) + os.lseek(destination_fd, 0, os.SEEK_SET) + remaining = size + while remaining: + data = os.read(source_fd, min(chunk_bytes, remaining)) + if not data: + raise TransferIntegrityError("source ended during copy") + _write_all(destination_fd, data) + remaining -= len(data) + + +def _write_all(file_descriptor: int, data: bytes) -> None: + view = memoryview(data) + while view: + written = os.write(file_descriptor, view) + view = view[written:] + + +def _publish_temporary(temporary: Path, destination: Path) -> None: + try: + os.link(temporary, destination, follow_symlinks=False) + except FileExistsError as exc: + raise TransferCollision("destination appeared concurrently") from exc + _fsync_directory(destination.parent) + + +def _temporary_path(destination: Path, operation_id: str) -> Path: + digest = hashlib.sha256(operation_id.encode()).hexdigest()[:16] + return destination.with_name(f".{destination.name}.archive-control-{digest}.tmp") + + +def _atomic_create_or_verify(path: Path, content: bytes) -> None: + if path.exists() or path.is_symlink(): + if _read_regular_file(path) != content: + raise TransferCollision(f"{path.name} already exists with other content") + return + temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp") + file_descriptor = os.open( + temporary, + os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), + 0o600, + ) + try: + _write_all(file_descriptor, content) + os.fsync(file_descriptor) + finally: + os.close(file_descriptor) + try: + _publish_temporary(temporary, path) + finally: + temporary.unlink(missing_ok=True) + + +def _verify_completed_destination( + destination: Path, expected: MaterializedFile +) -> None: + try: + metadata = destination.lstat() + except FileNotFoundError as exc: + raise TransferIntegrityError( + "completed journal entry has no destination" + ) from exc + if ( + not stat.S_ISREG(metadata.st_mode) + or metadata.st_size != expected.logical_bytes + or metadata.st_dev != expected.destination_device + or metadata.st_ino != expected.destination_inode + ): + raise TransferIntegrityError( + "completed journal destination no longer matches" + ) + + +def _result_from_json(value: object) -> MaterializedFile: + if not isinstance(value, dict): + raise TransferIntegrityError("file journal result is invalid") + try: + return MaterializedFile( + int(value["method"]), + int(value["logical_bytes"]), + int(value["allocated_bytes"]), + bool(value["sparse"]), + bool(value["destination_preexisted"]), + int(value["destination_device"]), + int(value["destination_inode"]), + ) + except (KeyError, TypeError, ValueError) as exc: + raise TransferIntegrityError("file journal result is invalid") from exc + + +def _operation_id( + job_id: str, phase: str, index: int, path: str +) -> str: + digest = hashlib.sha256( + f"{job_id}\0{phase}\0{index}\0{path}".encode() + ).hexdigest() + return f"{phase}:{digest}" + + +def _allocated_bytes(metadata: os.stat_result) -> int: + return int(getattr(metadata, "st_blocks", 0)) * 512 + + +def _sha256_file(path: Path) -> str: + digest = hashlib.sha256() + descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) + try: + while data := os.read(descriptor, _COPY_CHUNK_BYTES): + digest.update(data) + finally: + os.close(descriptor) + return digest.hexdigest() + + +def _is_sparse(metadata: os.stat_result) -> bool: + return metadata.st_size > 0 and _allocated_bytes(metadata) < metadata.st_size + + +def _canonical_json(value: object) -> bytes: + return json.dumps( + value, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode() + + +def _fsync_directory(path: Path) -> None: + descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) diff --git a/tests/test_state.py b/tests/test_state.py index 7bce711..b90812f 100644 --- a/tests/test_state.py +++ b/tests/test_state.py @@ -4,7 +4,12 @@ import os from pathlib import Path from uuid import uuid4 -from archive_clients.state import ClientStore, CommandConflict, JobConflict +from archive_clients.state import ( + ClientStore, + CommandConflict, + FileOperationConflict, + JobConflict, +) class ClientStoreTests(unittest.TestCase): @@ -86,6 +91,35 @@ class ClientStoreTests(unittest.TestCase): command_id, 2, "ready", '{"sequence":2,"changed":true}' ) + def test_file_operation_journal_is_idempotent_and_content_addressed(self): + with tempfile.TemporaryDirectory() as directory: + store = ClientStore(Path(directory) / "state.db") + store.initialize() + first = store.begin_file_operation( + "operation-1", "job-1", '{"destination":"a","source":"b"}' + ) + repeated = store.begin_file_operation( + "operation-1", "job-1", '{"source":"b","destination":"a"}' + ) + self.assertEqual(first["state"], "intent") + self.assertEqual(repeated["state"], "intent") + completed = store.complete_file_operation( + "operation-1", '{"method":1}' + ) + duplicate = store.complete_file_operation( + "operation-1", '{"method":1}' + ) + self.assertEqual(completed["state"], "completed") + self.assertEqual(duplicate["result_json"], '{"method":1}') + with self.assertRaises(FileOperationConflict): + store.begin_file_operation( + "operation-1", "job-1", '{"source":"other"}' + ) + with self.assertRaises(FileOperationConflict): + store.complete_file_operation( + "operation-1", '{"method":2}' + ) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_transfer.py b/tests/test_transfer.py new file mode 100644 index 0000000..afcf855 --- /dev/null +++ b/tests/test_transfer.py @@ -0,0 +1,247 @@ +import hashlib +import os +import tempfile +import unittest +from pathlib import Path +from unittest import mock +from uuid import uuid4 + +from archive_clients.state import ClientStore +from archive_clients.transfer import ( + FileMaterializer, + canonical_message_json, + load_published_transfer, + materialize_transfer, + stage_transfer, +) +from archive_control.v1 import transfer_pb2 + + +class TransferHappyPathTests(unittest.TestCase): + def setUp(self): + self.temporary = tempfile.TemporaryDirectory() + self.root = Path(self.temporary.name) + self.source = self.root / "source" + self.sync = self.root / "sync" + self.target = self.root / "target" + self.source.mkdir() + self.sync.mkdir() + self.target.mkdir() + (self.source / "album").mkdir() + (self.source / "album" / "one.bin").write_bytes(b"one" * 4096) + (self.source / "album" / "two.bin").write_bytes(b"two" * 2048) + self.metainfo = self.root / "source.torrent" + self.metainfo.write_bytes(b"d4:infod4:name5:albume") + self.store = ClientStore(self.root / "state.db") + self.store.initialize() + + def tearDown(self): + self.temporary.cleanup() + + def test_stage_publish_and_receiver_materialization_hardlink(self): + manifest = self._manifest() + published = stage_transfer( + manifest, + source_root=self.source, + sync_root=self.sync, + store=self.store, + artifact_sources={"metainfo/source.torrent": self.metainfo}, + ) + + staged_one = published.job_directory / "payload/album/one.bin" + self.assertEqual( + staged_one.stat().st_ino, + (self.source / "album/one.bin").stat().st_ino, + ) + self.assertTrue(published.ready_path.is_file()) + self.assertEqual( + hashlib.sha256(published.manifest_path.read_bytes()).hexdigest(), + published.manifest_sha256_hex, + ) + loaded = load_published_transfer(published.job_directory) + self.assertEqual(loaded.manifest.job_id, manifest.job_id) + self.assertEqual( + loaded.manifest.files[0].source_staging_method, + transfer_pb2.MATERIALIZATION_METHOD_HARD_LINK, + ) + self.assertEqual( + loaded.manifest.artifacts[0].sha256_hex, + hashlib.sha256(self.metainfo.read_bytes()).hexdigest(), + ) + + target_manifest = materialize_transfer( + loaded, + target_root=self.target, + store=self.store, + ) + self.assertEqual( + (self.target / "album/one.bin").read_bytes(), + (self.source / "album/one.bin").read_bytes(), + ) + self.assertEqual( + target_manifest.files[0].target_method, + transfer_pb2.MATERIALIZATION_METHOD_HARD_LINK, + ) + self.assertFalse(target_manifest.files[0].target_preexisted) + + def test_stage_and_materialize_are_idempotent_after_store_reopen(self): + manifest = self._manifest() + first = stage_transfer( + manifest, + source_root=self.source, + sync_root=self.sync, + store=self.store, + artifact_sources={"metainfo/source.torrent": self.metainfo}, + ) + first_target = materialize_transfer( + first, target_root=self.target, store=self.store + ) + reopened = ClientStore(self.root / "state.db") + second = stage_transfer( + manifest, + source_root=self.source, + sync_root=self.sync, + store=reopened, + artifact_sources={"metainfo/source.torrent": self.metainfo}, + ) + second_target = materialize_transfer( + second, target_root=self.target, store=reopened + ) + + self.assertEqual( + first.manifest_sha256_hex, second.manifest_sha256_hex + ) + self.assertEqual( + first_target.files[0].target_method, + second_target.files[0].target_method, + ) + self.assertEqual( + len(reopened.file_operation_rows(manifest.job_id)), 5 + ) + + def test_same_size_preexisting_target_is_reused(self): + manifest = self._manifest() + published = stage_transfer( + manifest, + source_root=self.source, + sync_root=self.sync, + store=self.store, + artifact_sources={"metainfo/source.torrent": self.metainfo}, + ) + (self.target / "album").mkdir() + preexisting = self.target / "album/one.bin" + preexisting.write_bytes((self.source / "album/one.bin").read_bytes()) + inode = preexisting.stat().st_ino + + result = materialize_transfer( + published, target_root=self.target, store=self.store + ) + + self.assertEqual(preexisting.stat().st_ino, inode) + self.assertEqual( + result.files[0].target_method, + transfer_pb2.MATERIALIZATION_METHOD_PREEXISTING_REUSED, + ) + self.assertTrue(result.files[0].target_preexisted) + + def test_reflink_fallback_records_reflink_method(self): + destination_root = self.root / "reflink-target" + destination_root.mkdir() + + def fake_reflink(source, destination, operation_id): + del operation_id + destination.write_bytes(source.read_bytes()) + + with mock.patch( + "archive_clients.transfer._reflink", side_effect=fake_reflink + ): + result = FileMaterializer( + self.store, allow_hardlink=False, allow_reflink=True + ).materialize( + job_id=str(uuid4()), + operation_id="reflink-operation", + source_root=self.source, + source_relative_path="album/one.bin", + destination_root=destination_root, + destination_relative_path="album/one.bin", + expected_bytes=(self.source / "album/one.bin").stat().st_size, + allow_preexisting_reuse=False, + sparse_supported=True, + ) + + self.assertEqual( + result.method, transfer_pb2.MATERIALIZATION_METHOD_REFLINK + ) + + def test_copy_fallback_preserves_sparse_file(self): + sparse = self.source / "album/sparse.bin" + with sparse.open("wb") as output: + output.seek(4 * 1024 * 1024) + output.write(b"tail") + destination_root = self.root / "copy-target" + destination_root.mkdir() + result = FileMaterializer( + self.store, allow_hardlink=False, allow_reflink=False + ).materialize( + job_id=str(uuid4()), + operation_id="copy-operation", + source_root=self.source, + source_relative_path="album/sparse.bin", + destination_root=destination_root, + destination_relative_path="album/sparse.bin", + expected_bytes=sparse.stat().st_size, + allow_preexisting_reuse=False, + sparse_supported=True, + ) + + copied = destination_root / "album/sparse.bin" + self.assertEqual(copied.read_bytes()[-4:], b"tail") + self.assertEqual(copied.stat().st_size, sparse.stat().st_size) + self.assertEqual( + result.method, transfer_pb2.MATERIALIZATION_METHOD_COPY + ) + self.assertTrue(result.sparse) + self.assertLess( + os.stat(copied).st_blocks * 512, os.stat(copied).st_size + ) + + def test_canonical_manifest_json_is_stable(self): + manifest = self._manifest() + first = canonical_message_json(manifest) + second = canonical_message_json(self._manifest(job_id=manifest.job_id)) + self.assertEqual(first, second) + self.assertEqual(first, canonical_message_json(manifest)) + + def _manifest(self, job_id=None): + manifest = transfer_pb2.TransferManifest( + manifest_version=1, + job_id=job_id or str(uuid4()), + source_client_id="cache-1", + target_client_id="archive-1", + route_id="route-1", + ) + manifest.resource_id.info_hash_v1_hex = "a" * 40 + manifest.created_at.seconds = 1_700_000_000 + for index, relative in enumerate( + ("album/one.bin", "album/two.bin") + ): + size = (self.source / relative).stat().st_size + entry = manifest.files.add( + file_index=index, + payload_relative_path=f"payload/{relative}", + target_canonical_path=relative, + logical_bytes=size, + ) + del entry + artifact = manifest.artifacts.add( + kind=transfer_pb2.ARTIFACT_KIND_TORRENT_FILE, + payload_relative_path="metainfo/source.torrent", + logical_bytes=self.metainfo.stat().st_size, + format="application/x-bittorrent", + ) + del artifact + return manifest + + +if __name__ == "__main__": + unittest.main()