feat: add journaled transfer materialization

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
+The daemon currently executes heartbeat, inventory, state-snapshot, and
-route-provisioning commands. Other mutation commands are durably rejected as
+route-provisioning commands. Transfer storage primitives are implemented and
-unsupported until their service and file-operation executors are added; they
+tested but are not yet wired to assignment/step commands; those and eviction
-are never falsely acknowledged as accepted.
+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:
 

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,

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)

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()

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()