"""Exact raw NoNE page-object pack sets. This module is intentionally independent from ``none_paging`` so the storage primitive can be validated before it is admitted into generation authority. It never interprets revision-6 deltas and never compresses or reconstructs a page object. Every packed entry is the byte-for-byte content of one complete revision 2, 3, or 7 safetensors object. The pack set is striped into disjoint shards. A single sequential device on the target host is not fast enough for the all-knowledge cold-ingestion gate; parallel shard reads provide one unique-byte measurement without counting replica copies more than once. """ from __future__ import annotations import hashlib import mmap import os import stat import time from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, replace from pathlib import Path from typing import Final, Sequence import torch from safetensors import safe_open from safetensors.torch import save_file DIRECT_PAGE_PACK_INDEX_REVISION: Final[int] = 1 DIRECT_PAGE_PACK_ALIGNMENT_BYTES: Final[int] = 4096 DIRECT_PAGE_PACK_IO_WAVE_BYTES: Final[int] = 64 * 1024 * 1024 DIRECT_PAGE_PACK_MAX_COALESCED_READ_BYTES: Final[int] = ( DIRECT_PAGE_PACK_IO_WAVE_BYTES ) DIRECT_PAGE_PACK_MIN_BYTES_PER_SECOND: Final[int] = 1_000_000_000 DIRECT_PAGE_PACK_SUFFIX: Final[str] = ".none-direct-pack" DIRECT_PAGE_PACK_INDEX_SUFFIX: Final[str] = ".none-direct-index.safetensors" DIRECT_PAGE_PACK_FORMAT_REVISIONS: Final[frozenset[int]] = frozenset({2, 3, 7}) _PAGE_WEIGHT_TENSOR_NAMES: Final[tuple[str, ...]] = ( "ffn_mode_t", "gate_t", "up_t", "down_t", "glyph_down_t", "glyph_up_t", "translation_gate_t", "outcome_memory_t", "repair_memory_t", "transfer_memory_t", ) _REVISION_2_KEYS: Final[frozenset[str]] = frozenset( { "format_revision_t", "page_ids_t", "optimizer_mean_t", "optimizer_square_t", "step_t", *_PAGE_WEIGHT_TENSOR_NAMES, } ) _REVISION_3_KEYS: Final[frozenset[str]] = frozenset( { "format_revision_t", "page_ids_t", "optimizer_width_t", *_PAGE_WEIGHT_TENSOR_NAMES, } ) _REVISION_7_KEYS: Final[frozenset[str]] = frozenset( { "format_revision_t", "page_ids_t", "optimizer_width_t", "step_t", *_PAGE_WEIGHT_TENSOR_NAMES, } ) _FORBIDDEN_OBJECT_KEY_FRAGMENTS: Final[tuple[str, ...]] = ( "_delta_", "base_object_", "base_generation_", "base_manifest_", "compression", "compressed", "semantic_pack", ) _INDEX_TENSOR_NAMES: Final[frozenset[str]] = frozenset( { "index_revision_t", "alignment_bytes_t", "page_count_t", "shard_count_t", "logical_object_bytes_t", "physical_pack_bytes_t", "page_ids_t", "object_sha256s_t", "object_bytes_t", "shard_indices_t", "object_offsets_t", "object_spans_t", "format_revisions_t", "shard_sha256s_t", "shard_bytes_t", "shard_logical_bytes_t", "shard_page_counts_t", "page_ids_sha256_t", "page_map_sha256_t", "pack_set_sha256_t", } ) @dataclass(frozen=True) class DirectPagePackSourcePacket: """External-I/O source rows for a deterministic raw pack build.""" page_ids_t: torch.Tensor object_sha256s_t: torch.Tensor object_bytes_t: torch.Tensor object_paths: tuple[Path, ...] @dataclass(frozen=True) class DirectPagePackIndexPacket: """Tensor-native logical and physical index for one striped pack set.""" page_ids_t: torch.Tensor object_sha256s_t: torch.Tensor object_bytes_t: torch.Tensor shard_indices_t: torch.Tensor object_offsets_t: torch.Tensor object_spans_t: torch.Tensor format_revisions_t: torch.Tensor shard_sha256s_t: torch.Tensor shard_bytes_t: torch.Tensor shard_logical_bytes_t: torch.Tensor shard_page_counts_t: torch.Tensor page_ids_sha256_t: torch.Tensor page_map_sha256_t: torch.Tensor pack_set_sha256_t: torch.Tensor @dataclass(frozen=True) class DirectPagePackSetAuthorityPacket: """Filesystem locator plus tensor identities for one immutable pack set.""" shard_roots: tuple[Path, ...] shard_relative_paths: tuple[str, ...] index_root: Path index_relative_path: str shard_sha256s_t: torch.Tensor index_sha256_t: torch.Tensor shard_bytes_t: torch.Tensor index_bytes_t: torch.Tensor logical_object_bytes_t: torch.Tensor page_count_t: torch.Tensor alignment_bytes_t: torch.Tensor page_ids_sha256_t: torch.Tensor page_map_sha256_t: torch.Tensor pack_set_sha256_t: torch.Tensor @dataclass(frozen=True) class DirectPagePackBuildPacket: """Durable pack-set build result.""" authority: DirectPagePackSetAuthorityPacket index: DirectPagePackIndexPacket @dataclass(frozen=True) class DirectPagePackSelectedPacket: """Exact selected raw objects in caller order.""" page_ids_t: torch.Tensor object_sha256s_t: torch.Tensor object_bytes_t: torch.Tensor payload_offsets_t: torch.Tensor object_payload_t: torch.Tensor shard_physical_read_bytes_t: torch.Tensor direct_io_t: torch.Tensor @dataclass(frozen=True) class DirectPagePackColdVerificationPacket: """Full unique-shard cold-ingestion evidence.""" pack_set_sha256_t: torch.Tensor shard_sha256s_t: torch.Tensor shard_physical_read_bytes_t: torch.Tensor shard_logical_bytes_per_second_t: torch.Tensor logical_object_bytes_t: torch.Tensor physical_read_bytes_t: torch.Tensor elapsed_nanoseconds_t: torch.Tensor aggregate_unique_logical_bytes_per_second_t: torch.Tensor direct_io_t: torch.Tensor zero_padding_verified_t: torch.Tensor @dataclass(frozen=True) class DirectPagePackReplicaReceiptPacket: """One non-inflating durable replica-set result.""" authority: DirectPagePackSetAuthorityPacket newly_written_shards_t: torch.Tensor newly_written_logical_bytes_t: torch.Tensor shard_replica_bytes_per_second_t: torch.Tensor total_newly_written_logical_bytes_t: torch.Tensor elapsed_nanoseconds_t: torch.Tensor aggregate_unique_logical_bytes_per_second_t: torch.Tensor direct_io_t: torch.Tensor cold_verification: DirectPagePackColdVerificationPacket @dataclass(frozen=True) class DirectPagePackShardReplicaReceiptPacket: """One exact proof-only shard copy with zero duplicate speed credit.""" shard_index_t: torch.Tensor shard_root: Path shard_relative_path: str shard_sha256_t: torch.Tensor shard_bytes_t: torch.Tensor logical_object_bytes_t: torch.Tensor newly_written_t: torch.Tensor newly_written_logical_bytes_t: torch.Tensor elapsed_nanoseconds_t: torch.Tensor direct_io_t: torch.Tensor @dataclass(frozen=True) class _BuiltShard: shard_index: int sha256: str physical_bytes: int logical_bytes: int page_count: int relative_path: str def _stable_cpu_tensor(value_t: torch.Tensor, *, dtype: torch.dtype) -> torch.Tensor: return value_t.detach().cpu().to(dtype=dtype).contiguous() def _digest_tensor(raw_digest: bytes) -> torch.Tensor: if len(raw_digest) != 32: raise RuntimeError("direct page pack digest geometry differs") return torch.frombuffer(bytearray(raw_digest), dtype=torch.uint8).clone() def _tensor_digest(value_t: torch.Tensor) -> bytes: stable_t = _stable_cpu_tensor(value_t, dtype=torch.uint8).reshape(-1) if stable_t.shape != (32,): raise RuntimeError("direct page pack digest tensor differs") return stable_t.numpy().tobytes(order="C") def _tensor_payload_digest(value_t: torch.Tensor) -> torch.Tensor: stable_t = value_t.detach().cpu().contiguous() digest = hashlib.sha256() digest.update(str(stable_t.dtype).encode("ascii") + b"\x00") digest.update( torch.tensor(stable_t.shape, dtype=torch.long).numpy().tobytes(order="C") ) digest.update( stable_t.reshape(-1).view(torch.uint8).numpy().tobytes(order="C") ) return _digest_tensor(digest.digest()) def _page_map_digest( page_ids_t: torch.Tensor, object_sha256s_t: torch.Tensor, object_bytes_t: torch.Tensor, ) -> torch.Tensor: digest = hashlib.sha256() for name, value_t in ( ("page_ids_t", page_ids_t), ("object_sha256s_t", object_sha256s_t), ("object_bytes_t", object_bytes_t), ): stable_t = value_t.detach().cpu().contiguous() digest.update(name.encode("ascii") + b"\x00") digest.update(str(stable_t.dtype).encode("ascii") + b"\x00") digest.update( torch.tensor(stable_t.shape, dtype=torch.long) .numpy() .tobytes(order="C") ) digest.update( stable_t.reshape(-1).view(torch.uint8).numpy().tobytes(order="C") ) return _digest_tensor(digest.digest()) def _pack_set_digest(index: DirectPagePackIndexPacket) -> torch.Tensor: digest = hashlib.sha256() for name, value_t in ( ("page_ids_t", index.page_ids_t), ("object_sha256s_t", index.object_sha256s_t), ("object_bytes_t", index.object_bytes_t), ("shard_indices_t", index.shard_indices_t), ("object_offsets_t", index.object_offsets_t), ("object_spans_t", index.object_spans_t), ("format_revisions_t", index.format_revisions_t), ("shard_sha256s_t", index.shard_sha256s_t), ("shard_bytes_t", index.shard_bytes_t), ("shard_logical_bytes_t", index.shard_logical_bytes_t), ("shard_page_counts_t", index.shard_page_counts_t), ): stable_t = value_t.detach().cpu().contiguous() digest.update(name.encode("ascii") + b"\x00") digest.update(str(stable_t.dtype).encode("ascii") + b"\x00") digest.update( torch.tensor(stable_t.shape, dtype=torch.long) .numpy() .tobytes(order="C") ) digest.update( stable_t.reshape(-1).view(torch.uint8).numpy().tobytes(order="C") ) return _digest_tensor(digest.digest()) def _align_up(value: int, alignment: int = DIRECT_PAGE_PACK_ALIGNMENT_BYTES) -> int: if value < 0 or alignment < 1 or alignment & (alignment - 1): raise RuntimeError("direct page pack alignment differs") return (value + alignment - 1) // alignment * alignment def _file_identity(path: Path) -> tuple[int, int, int, int, int, int]: identity = path.stat(follow_symlinks=False) return ( identity.st_dev, identity.st_ino, identity.st_size, identity.st_mtime_ns, identity.st_ctime_ns, identity.st_nlink, ) def _file_sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: while True: chunk = handle.read(8 * 1024 * 1024) if not chunk: break digest.update(chunk) return digest.hexdigest() def _fsync_directory(path: Path) -> None: descriptor = os.open( path, os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_DIRECTORY", 0), ) try: os.fsync(descriptor) finally: os.close(descriptor) def _reject_symlink_components( path: Path, *, allow_missing_tail: bool, ) -> Path: """Reject unresolved symlinks in every existing path component.""" absolute = Path(os.path.abspath(os.fspath(path.expanduser()))) current = Path(absolute.anchor) for part in absolute.parts[1:]: current = current / part try: identity = current.lstat() except FileNotFoundError: if allow_missing_tail: break raise RuntimeError( "direct page pack path component is absent" ) from None if stat.S_ISLNK(identity.st_mode): raise RuntimeError( "direct page pack path contains a symlink component" ) return absolute def _safe_root(path: Path, *, create: bool) -> Path: expanded = _reject_symlink_components( path, allow_missing_tail=create, ) if create: expanded.mkdir(parents=True, exist_ok=True) _reject_symlink_components(expanded, allow_missing_tail=False) resolved = expanded.resolve(strict=True) identity = resolved.lstat() if not stat.S_ISDIR(identity.st_mode): raise RuntimeError("direct page pack storage root is not a directory") return resolved def _safe_relative_path( root: Path, relative_path: str, *, must_exist: bool, create_parent: bool = False, ) -> Path: relative = Path(relative_path) if ( not relative_path or relative.is_absolute() or ".." in relative.parts or relative.parts in {(), (".",)} ): raise RuntimeError("direct page pack path escapes its storage root") resolved_root = _safe_root(root, create=create_parent) candidate = resolved_root / relative if create_parent: candidate.parent.mkdir(parents=True, exist_ok=True) _reject_symlink_components( candidate, allow_missing_tail=not must_exist, ) if must_exist: resolved = candidate.resolve(strict=True) if not resolved.is_relative_to(resolved_root): raise RuntimeError("direct page pack path escapes its storage root") identity = resolved.lstat() if not stat.S_ISREG(identity.st_mode): raise RuntimeError("direct page pack path is not a regular file") return resolved if not candidate.parent.resolve(strict=True).is_relative_to(resolved_root): raise RuntimeError("direct page pack path escapes its storage root") return candidate def _source_object_identity(path: Path, *, expected_bytes: int) -> tuple[int, ...]: expanded = _reject_symlink_components( path, allow_missing_tail=False, ) resolved = expanded.resolve(strict=True) identity = resolved.lstat() if ( not stat.S_ISREG(identity.st_mode) or identity.st_size != expected_bytes or identity.st_nlink != 1 ): raise RuntimeError("direct page pack source object identity differs") return _file_identity(resolved) def _validate_source_object_schema(path: Path, *, expected_page_id: int) -> int: with safe_open( # type: ignore[no-untyped-call] str(path), framework="pt", device="cpu", ) as handle: metadata = handle.metadata() if metadata not in (None, {}): raise RuntimeError( "direct page pack source metadata must be empty" ) keys = frozenset(str(name) for name in handle.keys()) if any( fragment in name for name in keys for fragment in _FORBIDDEN_OBJECT_KEY_FRAGMENTS ): raise RuntimeError( "direct page pack source contains indirect or compressed state" ) if "format_revision_t" not in keys or "page_ids_t" not in keys: raise RuntimeError("direct page pack source scalar authority is absent") revision_t = handle.get_tensor("format_revision_t").reshape(-1) page_ids_t = handle.get_tensor("page_ids_t").reshape(-1) if ( revision_t.dtype != torch.long or revision_t.shape != (1,) or page_ids_t.dtype != torch.long or page_ids_t.shape != (1,) ): raise RuntimeError("direct page pack source scalar authority differs") revision = int(revision_t[0]) if revision not in DIRECT_PAGE_PACK_FORMAT_REVISIONS: raise RuntimeError("direct page pack source revision is not 2, 3, or 7") if int(page_ids_t[0]) != expected_page_id: raise RuntimeError("direct page pack source page identity differs") expected_keys = { 2: _REVISION_2_KEYS, 3: _REVISION_3_KEYS, 7: _REVISION_7_KEYS, }[revision] if keys != expected_keys: raise RuntimeError("direct page pack source tensor key set differs") weight_tensors = tuple( handle.get_tensor(name) for name in _PAGE_WEIGHT_TENSOR_NAMES ) storage_dtype = weight_tensors[0].dtype if any( tensor.dtype != storage_dtype or tensor.ndim < 2 or tensor.shape[0] != 1 for tensor in weight_tensors ): raise RuntimeError("direct page pack source weight geometry differs") if revision in {3, 7}: width_t = handle.get_tensor("optimizer_width_t").reshape(-1) if ( width_t.dtype != torch.long or width_t.shape != (1,) or int(width_t[0]) < 1 ): raise RuntimeError( "direct page pack source optimizer width differs" ) if revision in {2, 7}: step_t = handle.get_tensor("step_t").reshape(-1) if step_t.dtype != torch.long or step_t.shape != (1,): raise RuntimeError("direct page pack source optimizer step differs") if revision == 2: mean_t = handle.get_tensor("optimizer_mean_t") square_t = handle.get_tensor("optimizer_square_t") if ( mean_t.shape != square_t.shape or mean_t.ndim != 2 or mean_t.shape[0] != 1 or mean_t.dtype not in {torch.float32, torch.bfloat16} or square_t.dtype not in {torch.float32, torch.bfloat16} ): raise RuntimeError( "direct page pack source optimizer tensors differ" ) return revision def _validate_source_packet( source: DirectPagePackSourcePacket, ) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor, tuple[Path, ...]]: page_ids_t = _stable_cpu_tensor(source.page_ids_t, dtype=torch.long).reshape(-1) object_sha256s_t = _stable_cpu_tensor( source.object_sha256s_t, dtype=torch.uint8, ) object_bytes_t = _stable_cpu_tensor( source.object_bytes_t, dtype=torch.long, ).reshape(-1) count = int(page_ids_t.numel()) if ( source.page_ids_t.dtype != torch.long or source.object_sha256s_t.dtype != torch.uint8 or source.object_bytes_t.dtype != torch.long or count < 1 or object_sha256s_t.shape != (count, 32) or object_bytes_t.shape != (count,) or len(source.object_paths) != count or bool(page_ids_t.lt(0).any()) or bool(object_bytes_t.le(0).any()) or torch.unique(page_ids_t).numel() != count or torch.unique(object_sha256s_t, dim=0).shape[0] != count ): raise RuntimeError("direct page pack source packet differs") order_t = torch.argsort(page_ids_t, stable=True) page_ids_t = page_ids_t[order_t].contiguous() object_sha256s_t = object_sha256s_t[order_t].contiguous() object_bytes_t = object_bytes_t[order_t].contiguous() order = tuple(int(value) for value in order_t) object_paths = tuple(source.object_paths[index] for index in order) return page_ids_t, object_sha256s_t, object_bytes_t, object_paths def _balanced_contiguous_shards( object_spans_t: torch.Tensor, *, shard_count: int, ) -> torch.Tensor: count = int(object_spans_t.numel()) if shard_count < 1 or shard_count > count: raise ValueError("direct page pack shard count differs") total_bytes = int(object_spans_t.sum()) target_bytes = (total_bytes + shard_count - 1) // shard_count assignments = torch.empty(count, dtype=torch.long) shard_index = 0 current_bytes = 0 for row_index in range(count): span = int(object_spans_t[row_index]) remaining_rows = count - row_index remaining_shards = shard_count - shard_index if ( shard_index + 1 < shard_count and current_bytes > 0 and current_bytes + span > target_bytes and remaining_rows >= remaining_shards ): shard_index += 1 current_bytes = 0 assignments[row_index] = shard_index current_bytes += span if int(assignments[-1]) != shard_count - 1: # Very skewed object sizes can leave empty trailing shards. Move one # terminal row into each missing shard without changing global order. assignments = torch.arange(count, dtype=torch.long).mul(shard_count).div( count, rounding_mode="floor", ) return assignments.contiguous() def _write_source_into_shard( *, descriptor: int, source_path: Path, expected_page_id: int, expected_sha256: bytes, expected_bytes: int, offset: int, span: int, ) -> int: identity_before = _source_object_identity( source_path, expected_bytes=expected_bytes, ) resolved_source = source_path.expanduser().resolve(strict=True) expected_hex = expected_sha256.hex() if resolved_source.name != f"{expected_hex}.safetensors": raise RuntimeError("direct page pack source content-addressed name differs") revision = _validate_source_object_schema( resolved_source, expected_page_id=expected_page_id, ) source_descriptor = os.open( resolved_source, os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0), ) digest = hashlib.sha256() copied = 0 try: while copied < expected_bytes: chunk = os.read( source_descriptor, min(8 * 1024 * 1024, expected_bytes - copied), ) if not chunk: raise RuntimeError("direct page pack source read was incomplete") digest.update(chunk) written = 0 while written < len(chunk): count = os.pwrite( descriptor, chunk[written:], offset + copied + written, ) if count < 1: raise RuntimeError( "direct page pack destination write was incomplete" ) written += count copied += len(chunk) finally: os.close(source_descriptor) if copied != expected_bytes or digest.digest() != expected_sha256: raise RuntimeError("direct page pack source object hash differs") padding_bytes = span - expected_bytes if padding_bytes < 0 or padding_bytes >= DIRECT_PAGE_PACK_ALIGNMENT_BYTES: raise RuntimeError("direct page pack source span differs") if padding_bytes: padding = b"\x00" * padding_bytes written = os.pwrite(descriptor, padding, offset + expected_bytes) if written != padding_bytes: raise RuntimeError("direct page pack zero padding write was incomplete") if _file_identity(resolved_source) != identity_before: raise RuntimeError("direct page pack source changed during build") return revision def _build_one_shard( *, shard_index: int, shard_root: Path, row_indices: tuple[int, ...], page_ids_t: torch.Tensor, object_sha256s_t: torch.Tensor, object_bytes_t: torch.Tensor, object_paths: tuple[Path, ...], object_offsets_t: torch.Tensor, object_spans_t: torch.Tensor, format_revisions_t: torch.Tensor, ) -> _BuiltShard: root = _safe_root(shard_root, create=True) temporary_root = _safe_relative_path( root, "direct-page-packs/sha256/.build", must_exist=False, create_parent=True, ) temporary_root.mkdir(parents=True, exist_ok=True) if temporary_root.is_symlink(): raise RuntimeError("direct page pack temporary root is a symlink") temporary_path = temporary_root / ( f".shard-{shard_index}.{os.getpid()}.{time.monotonic_ns()}.tmp" ) descriptor = -1 logical_bytes = 0 try: descriptor = os.open( temporary_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0), 0o600, ) for row_index in row_indices: object_sha256 = object_sha256s_t[row_index].numpy().tobytes() revision = _write_source_into_shard( descriptor=descriptor, source_path=object_paths[row_index], expected_page_id=int(page_ids_t[row_index]), expected_sha256=object_sha256, expected_bytes=int(object_bytes_t[row_index]), offset=int(object_offsets_t[row_index]), span=int(object_spans_t[row_index]), ) format_revisions_t[row_index] = revision logical_bytes += int(object_bytes_t[row_index]) physical_bytes = sum(int(object_spans_t[index]) for index in row_indices) os.ftruncate(descriptor, physical_bytes) os.fsync(descriptor) os.close(descriptor) descriptor = -1 shard_sha256 = _file_sha256(temporary_path) relative_path = ( f"direct-page-packs/sha256/{shard_sha256}{DIRECT_PAGE_PACK_SUFFIX}" ) final_path = _safe_relative_path( root, relative_path, must_exist=False, create_parent=True, ) if final_path.exists(): if ( final_path.is_symlink() or final_path.stat().st_size != temporary_path.stat().st_size or _file_sha256(final_path) != shard_sha256 ): raise RuntimeError("existing direct page pack shard differs") temporary_path.unlink() else: os.rename(temporary_path, final_path) _fsync_directory(final_path.parent) return _BuiltShard( shard_index=shard_index, sha256=shard_sha256, physical_bytes=final_path.stat().st_size, logical_bytes=logical_bytes, page_count=len(row_indices), relative_path=relative_path, ) finally: if descriptor >= 0: os.close(descriptor) temporary_path.unlink(missing_ok=True) def _index_serialization_tensors( index: DirectPagePackIndexPacket, ) -> dict[str, torch.Tensor]: return { "index_revision_t": torch.tensor( [DIRECT_PAGE_PACK_INDEX_REVISION], dtype=torch.long, ), "alignment_bytes_t": torch.tensor( [DIRECT_PAGE_PACK_ALIGNMENT_BYTES], dtype=torch.long, ), "page_count_t": torch.tensor( [index.page_ids_t.numel()], dtype=torch.long, ), "shard_count_t": torch.tensor( [index.shard_sha256s_t.shape[0]], dtype=torch.long, ), "logical_object_bytes_t": index.object_bytes_t.sum().reshape(1), "physical_pack_bytes_t": index.shard_bytes_t.sum().reshape(1), "page_ids_t": index.page_ids_t, "object_sha256s_t": index.object_sha256s_t, "object_bytes_t": index.object_bytes_t, "shard_indices_t": index.shard_indices_t, "object_offsets_t": index.object_offsets_t, "object_spans_t": index.object_spans_t, "format_revisions_t": index.format_revisions_t, "shard_sha256s_t": index.shard_sha256s_t, "shard_bytes_t": index.shard_bytes_t, "shard_logical_bytes_t": index.shard_logical_bytes_t, "shard_page_counts_t": index.shard_page_counts_t, "page_ids_sha256_t": index.page_ids_sha256_t, "page_map_sha256_t": index.page_map_sha256_t, "pack_set_sha256_t": index.pack_set_sha256_t, } def validate_direct_page_pack_index_boundary( index: DirectPagePackIndexPacket, ) -> DirectPagePackIndexPacket: """Fail closed on any noncanonical or indirect pack-set index.""" page_ids_t = _stable_cpu_tensor(index.page_ids_t, dtype=torch.long).reshape(-1) object_sha256s_t = _stable_cpu_tensor( index.object_sha256s_t, dtype=torch.uint8, ) object_bytes_t = _stable_cpu_tensor( index.object_bytes_t, dtype=torch.long, ).reshape(-1) shard_indices_t = _stable_cpu_tensor( index.shard_indices_t, dtype=torch.long, ).reshape(-1) object_offsets_t = _stable_cpu_tensor( index.object_offsets_t, dtype=torch.long, ).reshape(-1) object_spans_t = _stable_cpu_tensor( index.object_spans_t, dtype=torch.long, ).reshape(-1) format_revisions_t = _stable_cpu_tensor( index.format_revisions_t, dtype=torch.long, ).reshape(-1) shard_sha256s_t = _stable_cpu_tensor( index.shard_sha256s_t, dtype=torch.uint8, ) shard_bytes_t = _stable_cpu_tensor( index.shard_bytes_t, dtype=torch.long, ).reshape(-1) shard_logical_bytes_t = _stable_cpu_tensor( index.shard_logical_bytes_t, dtype=torch.long, ).reshape(-1) shard_page_counts_t = _stable_cpu_tensor( index.shard_page_counts_t, dtype=torch.long, ).reshape(-1) count = int(page_ids_t.numel()) shard_count = int(shard_bytes_t.numel()) if ( index.page_ids_t.dtype != torch.long or index.object_sha256s_t.dtype != torch.uint8 or index.object_bytes_t.dtype != torch.long or index.shard_indices_t.dtype != torch.long or index.object_offsets_t.dtype != torch.long or index.object_spans_t.dtype != torch.long or index.format_revisions_t.dtype != torch.long or index.shard_sha256s_t.dtype != torch.uint8 or index.shard_bytes_t.dtype != torch.long or index.shard_logical_bytes_t.dtype != torch.long or index.shard_page_counts_t.dtype != torch.long or count < 1 or shard_count < 1 or object_sha256s_t.shape != (count, 32) or object_bytes_t.shape != (count,) or shard_indices_t.shape != (count,) or object_offsets_t.shape != (count,) or object_spans_t.shape != (count,) or format_revisions_t.shape != (count,) or shard_sha256s_t.shape != (shard_count, 32) or shard_logical_bytes_t.shape != (shard_count,) or shard_page_counts_t.shape != (shard_count,) or not torch.equal(page_ids_t, torch.sort(page_ids_t).values) or torch.unique(page_ids_t).numel() != count or torch.unique(object_sha256s_t, dim=0).shape[0] != count or bool(object_bytes_t.le(0).any()) or bool(shard_indices_t.lt(0).any()) or bool(shard_indices_t.ge(shard_count).any()) or bool(object_offsets_t.lt(0).any()) or bool(object_spans_t.le(0).any()) or bool(object_spans_t.remainder(DIRECT_PAGE_PACK_ALIGNMENT_BYTES).any()) or bool(object_offsets_t.remainder(DIRECT_PAGE_PACK_ALIGNMENT_BYTES).any()) or bool(object_spans_t.lt(object_bytes_t).any()) or bool( object_spans_t .ne( object_bytes_t.add(DIRECT_PAGE_PACK_ALIGNMENT_BYTES - 1) .div(DIRECT_PAGE_PACK_ALIGNMENT_BYTES, rounding_mode="floor") .mul(DIRECT_PAGE_PACK_ALIGNMENT_BYTES) ) .any() ) or bool( ~torch.isin( format_revisions_t, torch.tensor( sorted(DIRECT_PAGE_PACK_FORMAT_REVISIONS), dtype=torch.long, ), ).all() ) or bool(shard_bytes_t.le(0).any()) or bool(shard_logical_bytes_t.le(0).any()) or bool(shard_page_counts_t.le(0).any()) or int(shard_page_counts_t.sum()) != count ): raise RuntimeError("direct page pack index tensor authority differs") derived_shard_bytes = torch.zeros(shard_count, dtype=torch.long) derived_logical_bytes = torch.zeros(shard_count, dtype=torch.long) derived_page_counts = torch.zeros(shard_count, dtype=torch.long) for shard_index in range(shard_count): rows_t = shard_indices_t.eq(shard_index).nonzero( as_tuple=False ).reshape(-1) if rows_t.numel() < 1: raise RuntimeError("direct page pack index contains an empty shard") offsets_t = object_offsets_t[rows_t] spans_t = object_spans_t[rows_t] expected_offsets_t = torch.cat( ( torch.zeros(1, dtype=torch.long), torch.cumsum(spans_t[:-1], dim=0), ) ) if not torch.equal(offsets_t, expected_offsets_t): raise RuntimeError( "direct page pack index overlaps or contains a physical hole" ) derived_shard_bytes[shard_index] = spans_t.sum() derived_logical_bytes[shard_index] = object_bytes_t[rows_t].sum() derived_page_counts[shard_index] = rows_t.numel() if ( not torch.equal(derived_shard_bytes, shard_bytes_t) or not torch.equal(derived_logical_bytes, shard_logical_bytes_t) or not torch.equal(derived_page_counts, shard_page_counts_t) ): raise RuntimeError("direct page pack shard aggregates differ") canonical = DirectPagePackIndexPacket( page_ids_t=page_ids_t, object_sha256s_t=object_sha256s_t, object_bytes_t=object_bytes_t, shard_indices_t=shard_indices_t, object_offsets_t=object_offsets_t, object_spans_t=object_spans_t, format_revisions_t=format_revisions_t, shard_sha256s_t=shard_sha256s_t, shard_bytes_t=shard_bytes_t, shard_logical_bytes_t=shard_logical_bytes_t, shard_page_counts_t=shard_page_counts_t, page_ids_sha256_t=_stable_cpu_tensor( index.page_ids_sha256_t, dtype=torch.uint8, ).reshape(-1), page_map_sha256_t=_stable_cpu_tensor( index.page_map_sha256_t, dtype=torch.uint8, ).reshape(-1), pack_set_sha256_t=_stable_cpu_tensor( index.pack_set_sha256_t, dtype=torch.uint8, ).reshape(-1), ) expected_page_ids_sha256_t = _tensor_payload_digest(page_ids_t) expected_page_map_sha256_t = _page_map_digest( page_ids_t, object_sha256s_t, object_bytes_t, ) expected_pack_set_sha256_t = _pack_set_digest( replace( canonical, page_ids_sha256_t=expected_page_ids_sha256_t, page_map_sha256_t=expected_page_map_sha256_t, pack_set_sha256_t=torch.zeros(32, dtype=torch.uint8), ) ) if ( canonical.page_ids_sha256_t.shape != (32,) or canonical.page_map_sha256_t.shape != (32,) or canonical.pack_set_sha256_t.shape != (32,) or not torch.equal( canonical.page_ids_sha256_t, expected_page_ids_sha256_t, ) or not torch.equal( canonical.page_map_sha256_t, expected_page_map_sha256_t, ) or not torch.equal( canonical.pack_set_sha256_t, expected_pack_set_sha256_t, ) ): raise RuntimeError("direct page pack index digest authority differs") return canonical def build_direct_page_pack_set_boundary( source: DirectPagePackSourcePacket, *, shard_roots: Sequence[Path], index_root: Path, ) -> DirectPagePackBuildPacket: """Build deterministic disjoint aligned raw shards without whole-pack memory.""" ( page_ids_t, object_sha256s_t, object_bytes_t, object_paths, ) = _validate_source_packet(source) roots = tuple(_safe_root(Path(root), create=True) for root in shard_roots) if not roots or len(roots) > page_ids_t.numel(): raise ValueError("direct page pack shard roots differ") object_spans_t = ( object_bytes_t.add(DIRECT_PAGE_PACK_ALIGNMENT_BYTES - 1) .div(DIRECT_PAGE_PACK_ALIGNMENT_BYTES, rounding_mode="floor") .mul(DIRECT_PAGE_PACK_ALIGNMENT_BYTES) .contiguous() ) shard_indices_t = _balanced_contiguous_shards( object_spans_t, shard_count=len(roots), ) object_offsets_t = torch.empty_like(object_spans_t) for shard_index in range(len(roots)): rows_t = shard_indices_t.eq(shard_index).nonzero( as_tuple=False ).reshape(-1) spans_t = object_spans_t[rows_t] object_offsets_t[rows_t] = torch.cat( ( torch.zeros(1, dtype=torch.long), torch.cumsum(spans_t[:-1], dim=0), ) ) format_revisions_t = torch.zeros_like(page_ids_t) futures = [] with ThreadPoolExecutor( max_workers=len(roots), thread_name_prefix="nnf-direct-pack-build", ) as executor: for shard_index, root in enumerate(roots): row_indices = tuple( int(value) for value in shard_indices_t.eq(shard_index) .nonzero(as_tuple=False) .reshape(-1) ) futures.append( executor.submit( _build_one_shard, shard_index=shard_index, shard_root=root, row_indices=row_indices, page_ids_t=page_ids_t, object_sha256s_t=object_sha256s_t, object_bytes_t=object_bytes_t, object_paths=object_paths, object_offsets_t=object_offsets_t, object_spans_t=object_spans_t, format_revisions_t=format_revisions_t, ) ) built_shards = tuple(future.result() for future in futures) built_shards = tuple( sorted(built_shards, key=lambda shard: shard.shard_index) ) shard_sha256s_t = torch.stack( tuple( _digest_tensor(bytes.fromhex(shard.sha256)) for shard in built_shards ) ) shard_bytes_t = torch.tensor( [shard.physical_bytes for shard in built_shards], dtype=torch.long, ) shard_logical_bytes_t = torch.tensor( [shard.logical_bytes for shard in built_shards], dtype=torch.long, ) shard_page_counts_t = torch.tensor( [shard.page_count for shard in built_shards], dtype=torch.long, ) page_ids_sha256_t = _tensor_payload_digest(page_ids_t) page_map_sha256_t = _page_map_digest( page_ids_t, object_sha256s_t, object_bytes_t, ) provisional = DirectPagePackIndexPacket( page_ids_t=page_ids_t, object_sha256s_t=object_sha256s_t, object_bytes_t=object_bytes_t, shard_indices_t=shard_indices_t, object_offsets_t=object_offsets_t, object_spans_t=object_spans_t, format_revisions_t=format_revisions_t, shard_sha256s_t=shard_sha256s_t, shard_bytes_t=shard_bytes_t, shard_logical_bytes_t=shard_logical_bytes_t, shard_page_counts_t=shard_page_counts_t, page_ids_sha256_t=page_ids_sha256_t, page_map_sha256_t=page_map_sha256_t, pack_set_sha256_t=torch.zeros(32, dtype=torch.uint8), ) index = replace( provisional, pack_set_sha256_t=_pack_set_digest(provisional), ) index = validate_direct_page_pack_index_boundary(index) resolved_index_root = _safe_root(index_root, create=True) temporary_index = _safe_relative_path( resolved_index_root, ( "direct-page-packs/index/" f".index.{os.getpid()}.{time.monotonic_ns()}.tmp" ), must_exist=False, create_parent=True, ) save_file(_index_serialization_tensors(index), str(temporary_index)) with temporary_index.open("rb") as handle: os.fsync(handle.fileno()) index_sha256 = _file_sha256(temporary_index) index_relative_path = ( f"direct-page-packs/index/{index_sha256}" f"{DIRECT_PAGE_PACK_INDEX_SUFFIX}" ) final_index = _safe_relative_path( resolved_index_root, index_relative_path, must_exist=False, create_parent=True, ) try: if final_index.exists(): if ( final_index.is_symlink() or final_index.stat().st_size != temporary_index.stat().st_size or _file_sha256(final_index) != index_sha256 ): raise RuntimeError("existing direct page pack index differs") temporary_index.unlink() else: os.rename(temporary_index, final_index) _fsync_directory(final_index.parent) finally: temporary_index.unlink(missing_ok=True) authority = DirectPagePackSetAuthorityPacket( shard_roots=roots, shard_relative_paths=tuple( shard.relative_path for shard in built_shards ), index_root=resolved_index_root, index_relative_path=index_relative_path, shard_sha256s_t=index.shard_sha256s_t.clone(), index_sha256_t=_digest_tensor(bytes.fromhex(index_sha256)), shard_bytes_t=index.shard_bytes_t.clone(), index_bytes_t=torch.tensor( final_index.stat().st_size, dtype=torch.long, ), logical_object_bytes_t=index.object_bytes_t.sum().reshape(()), page_count_t=torch.tensor(index.page_ids_t.numel(), dtype=torch.long), alignment_bytes_t=torch.tensor( DIRECT_PAGE_PACK_ALIGNMENT_BYTES, dtype=torch.long, ), page_ids_sha256_t=index.page_ids_sha256_t.clone(), page_map_sha256_t=index.page_map_sha256_t.clone(), pack_set_sha256_t=index.pack_set_sha256_t.clone(), ) validate_direct_page_pack_set_authority_boundary(authority, index=index) return DirectPagePackBuildPacket(authority=authority, index=index) def _load_index_file(path: Path) -> DirectPagePackIndexPacket: with safe_open( # type: ignore[no-untyped-call] str(path), framework="pt", device="cpu", ) as handle: keys = frozenset(str(name) for name in handle.keys()) if keys != _INDEX_TENSOR_NAMES: raise RuntimeError("direct page pack index key set differs") scalars = { name: handle.get_tensor(name).reshape(-1) for name in ( "index_revision_t", "alignment_bytes_t", "page_count_t", "shard_count_t", "logical_object_bytes_t", "physical_pack_bytes_t", ) } if any( value.dtype != torch.long or value.shape != (1,) for value in scalars.values() ): raise RuntimeError("direct page pack index scalar authority differs") if ( int(scalars["index_revision_t"][0]) != DIRECT_PAGE_PACK_INDEX_REVISION or int(scalars["alignment_bytes_t"][0]) != DIRECT_PAGE_PACK_ALIGNMENT_BYTES ): raise RuntimeError("direct page pack index revision differs") index = DirectPagePackIndexPacket( page_ids_t=handle.get_tensor("page_ids_t"), object_sha256s_t=handle.get_tensor("object_sha256s_t"), object_bytes_t=handle.get_tensor("object_bytes_t"), shard_indices_t=handle.get_tensor("shard_indices_t"), object_offsets_t=handle.get_tensor("object_offsets_t"), object_spans_t=handle.get_tensor("object_spans_t"), format_revisions_t=handle.get_tensor("format_revisions_t"), shard_sha256s_t=handle.get_tensor("shard_sha256s_t"), shard_bytes_t=handle.get_tensor("shard_bytes_t"), shard_logical_bytes_t=handle.get_tensor( "shard_logical_bytes_t" ), shard_page_counts_t=handle.get_tensor("shard_page_counts_t"), page_ids_sha256_t=handle.get_tensor("page_ids_sha256_t"), page_map_sha256_t=handle.get_tensor("page_map_sha256_t"), pack_set_sha256_t=handle.get_tensor("pack_set_sha256_t"), ) validated = validate_direct_page_pack_index_boundary(index) if ( int(scalars["page_count_t"][0]) != validated.page_ids_t.numel() or int(scalars["shard_count_t"][0]) != validated.shard_sha256s_t.shape[0] or int(scalars["logical_object_bytes_t"][0]) != int(validated.object_bytes_t.sum()) or int(scalars["physical_pack_bytes_t"][0]) != int(validated.shard_bytes_t.sum()) ): raise RuntimeError("direct page pack index aggregate scalar differs") return validated def reopen_direct_page_pack_set_boundary( *, storage_root: Path, index_relative_path: str, ) -> DirectPagePackBuildPacket: """Reopen one exact canonical pack set without rebuilding or copying it. The caller supplies the immutable storage root and the exact content-addressed index path. The index is the sole logical authority: shard paths are derived from its shard digests, never discovered by walking source objects or storage directories. This boundary reads and validates the small index file, but it deliberately does not open or hash any shard payload. Full payload hashing remains the responsibility of the explicit cold-verification boundary. """ resolved_root = _safe_root(storage_root, create=False) index_path = _safe_relative_path( resolved_root, index_relative_path, must_exist=True, ) index_identity = _file_identity(index_path) if index_identity[5] != 1: raise RuntimeError("direct page pack index file authority differs") index_sha256 = _file_sha256(index_path) expected_index_relative_path = ( "direct-page-packs/index/" f"{index_sha256}{DIRECT_PAGE_PACK_INDEX_SUFFIX}" ) if index_relative_path != expected_index_relative_path: raise RuntimeError( "direct page pack index content-addressed path differs" ) if _file_identity(index_path) != index_identity: raise RuntimeError("direct page pack index changed during validation") index = _load_index_file(index_path) if _file_identity(index_path) != index_identity: raise RuntimeError("direct page pack index changed during validation") shard_relative_paths = tuple( ( "direct-page-packs/sha256/" f"{_tensor_digest(shard_sha256_t).hex()}" f"{DIRECT_PAGE_PACK_SUFFIX}" ) for shard_sha256_t in index.shard_sha256s_t ) shard_roots = (resolved_root,) * len(shard_relative_paths) authority = DirectPagePackSetAuthorityPacket( shard_roots=shard_roots, shard_relative_paths=shard_relative_paths, index_root=resolved_root, index_relative_path=index_relative_path, shard_sha256s_t=index.shard_sha256s_t.clone(), index_sha256_t=_digest_tensor(bytes.fromhex(index_sha256)), shard_bytes_t=index.shard_bytes_t.clone(), index_bytes_t=torch.tensor(index_identity[2], dtype=torch.long), logical_object_bytes_t=index.object_bytes_t.sum().reshape(()), page_count_t=torch.tensor( index.page_ids_t.numel(), dtype=torch.long, ), alignment_bytes_t=torch.tensor( DIRECT_PAGE_PACK_ALIGNMENT_BYTES, dtype=torch.long, ), page_ids_sha256_t=index.page_ids_sha256_t.clone(), page_map_sha256_t=index.page_map_sha256_t.clone(), pack_set_sha256_t=index.pack_set_sha256_t.clone(), ) validate_direct_page_pack_set_authority_boundary( authority, index=index, ) return DirectPagePackBuildPacket(authority=authority, index=index) def load_existing_direct_page_pack_set_for_source_boundary( source: DirectPagePackSourcePacket, *, shard_roots: Sequence[Path], index_root: Path, ) -> DirectPagePackBuildPacket | None: """Reopen one exact content-addressed pack without rewriting its payload. This is a metadata and immutable-file-identity resume boundary. It requires the existing index to name the exact source page/object map and deterministic physical layout, then revalidates every source object's direct schema. Callers that use this to publish generation authority must still perform the full cold shard hash/read proof; this boundary never substitutes cached metadata for that proof. """ ( page_ids_t, object_sha256s_t, object_bytes_t, object_paths, ) = _validate_source_packet(source) roots = tuple(_safe_root(Path(root), create=True) for root in shard_roots) if not roots or len(roots) > page_ids_t.numel(): raise ValueError("direct page pack shard roots differ") resolved_index_root = _safe_root(index_root, create=True) index_directory = ( resolved_index_root / "direct-page-packs" / "index" ) if not index_directory.exists(): return None safe_index_directory = _safe_root(index_directory, create=False) object_spans_t = ( object_bytes_t.add(DIRECT_PAGE_PACK_ALIGNMENT_BYTES - 1) .div(DIRECT_PAGE_PACK_ALIGNMENT_BYTES, rounding_mode="floor") .mul(DIRECT_PAGE_PACK_ALIGNMENT_BYTES) .contiguous() ) shard_indices_t = _balanced_contiguous_shards( object_spans_t, shard_count=len(roots), ) object_offsets_t = torch.empty_like(object_spans_t) expected_shard_bytes: list[int] = [] expected_shard_logical_bytes: list[int] = [] expected_shard_page_counts: list[int] = [] for shard_index in range(len(roots)): rows_t = shard_indices_t.eq(shard_index).nonzero( as_tuple=False ).reshape(-1) spans_t = object_spans_t[rows_t] object_offsets_t[rows_t] = torch.cat( ( torch.zeros(1, dtype=torch.long), torch.cumsum(spans_t[:-1], dim=0), ) ) expected_shard_bytes.append(int(spans_t.sum())) expected_shard_logical_bytes.append( int(object_bytes_t[rows_t].sum()) ) expected_shard_page_counts.append(int(rows_t.numel())) candidates: list[ tuple[Path, DirectPagePackIndexPacket] ] = [] for index_path in sorted( safe_index_directory.glob(f"*{DIRECT_PAGE_PACK_INDEX_SUFFIX}") ): identity = index_path.lstat() if ( not stat.S_ISREG(identity.st_mode) or identity.st_nlink != 1 ): continue index_sha256 = _file_sha256(index_path) if ( index_path.name != f"{index_sha256}{DIRECT_PAGE_PACK_INDEX_SUFFIX}" ): continue try: existing = _load_index_file(index_path) except RuntimeError: continue if ( not torch.equal(existing.page_ids_t, page_ids_t) or not torch.equal( existing.object_sha256s_t, object_sha256s_t, ) or not torch.equal(existing.object_bytes_t, object_bytes_t) or not torch.equal( existing.shard_indices_t, shard_indices_t, ) or not torch.equal( existing.object_offsets_t, object_offsets_t, ) or not torch.equal(existing.object_spans_t, object_spans_t) or existing.shard_bytes_t.tolist() != expected_shard_bytes or existing.shard_logical_bytes_t.tolist() != expected_shard_logical_bytes or existing.shard_page_counts_t.tolist() != expected_shard_page_counts ): continue candidates.append((index_path, existing)) if not candidates: return None source_revisions_t = torch.tensor( [ _validated_existing_pack_source_revision_boundary( path=path, expected_page_id=int(page_id), expected_sha256=object_sha256.numpy().tobytes(), expected_bytes=int(object_bytes), ) for page_id, object_sha256, object_bytes, path in zip( page_ids_t, object_sha256s_t, object_bytes_t, object_paths, strict=True, ) ], dtype=torch.long, ) candidates = [ (path, index) for path, index in candidates if torch.equal(index.format_revisions_t, source_revisions_t) ] if not candidates: return None if len(candidates) != 1: raise RuntimeError( "existing direct page pack source authority is ambiguous" ) index_path, index = candidates[0] shard_relative_paths = tuple( ( "direct-page-packs/sha256/" f"{_tensor_digest(index.shard_sha256s_t[shard_index]).hex()}" f"{DIRECT_PAGE_PACK_SUFFIX}" ) for shard_index in range(len(roots)) ) authority = DirectPagePackSetAuthorityPacket( shard_roots=roots, shard_relative_paths=shard_relative_paths, index_root=resolved_index_root, index_relative_path=str(index_path.relative_to(resolved_index_root)), shard_sha256s_t=index.shard_sha256s_t.clone(), index_sha256_t=_digest_tensor( bytes.fromhex(_file_sha256(index_path)) ), shard_bytes_t=index.shard_bytes_t.clone(), index_bytes_t=torch.tensor( index_path.stat().st_size, dtype=torch.long, ), logical_object_bytes_t=index.object_bytes_t.sum().reshape(()), page_count_t=torch.tensor( index.page_ids_t.numel(), dtype=torch.long, ), alignment_bytes_t=torch.tensor( DIRECT_PAGE_PACK_ALIGNMENT_BYTES, dtype=torch.long, ), page_ids_sha256_t=index.page_ids_sha256_t.clone(), page_map_sha256_t=index.page_map_sha256_t.clone(), pack_set_sha256_t=index.pack_set_sha256_t.clone(), ) validate_direct_page_pack_set_authority_boundary( authority, index=index, ) return DirectPagePackBuildPacket(authority=authority, index=index) def _validated_existing_pack_source_revision_boundary( *, path: Path, expected_page_id: int, expected_sha256: bytes, expected_bytes: int, ) -> int: """Validate one source identity without rereading its tensor payload.""" identity_before = _source_object_identity( path, expected_bytes=expected_bytes, ) resolved = path.expanduser().resolve(strict=True) if resolved.name != f"{expected_sha256.hex()}.safetensors": raise RuntimeError( "direct page pack source content-addressed name differs" ) revision = _validate_source_object_schema( resolved, expected_page_id=expected_page_id, ) if _file_identity(resolved) != identity_before: raise RuntimeError( "direct page pack source changed during resume validation" ) return revision def validate_direct_page_pack_set_authority_boundary( authority: DirectPagePackSetAuthorityPacket, *, index: DirectPagePackIndexPacket | None = None, ) -> DirectPagePackIndexPacket: """Validate path containment, immutable identities, and tensor authority.""" shard_count = len(authority.shard_roots) if ( shard_count < 1 or len(authority.shard_relative_paths) != shard_count or authority.shard_sha256s_t.dtype != torch.uint8 or authority.shard_sha256s_t.shape != (shard_count, 32) or authority.shard_bytes_t.dtype != torch.long or authority.shard_bytes_t.shape != (shard_count,) or authority.index_sha256_t.dtype != torch.uint8 or authority.index_sha256_t.reshape(-1).shape != (32,) or authority.index_bytes_t.dtype != torch.long or authority.index_bytes_t.reshape(-1).shape != (1,) or int(authority.index_bytes_t) < 1 or authority.logical_object_bytes_t.dtype != torch.long or authority.logical_object_bytes_t.reshape(-1).shape != (1,) or int(authority.logical_object_bytes_t) < 1 or authority.page_count_t.dtype != torch.long or authority.page_count_t.reshape(-1).shape != (1,) or int(authority.page_count_t) < 1 or authority.alignment_bytes_t.dtype != torch.long or authority.alignment_bytes_t.reshape(-1).shape != (1,) or int(authority.alignment_bytes_t) != DIRECT_PAGE_PACK_ALIGNMENT_BYTES ): raise RuntimeError("direct page pack set authority tensor geometry differs") index_path = _safe_relative_path( authority.index_root, authority.index_relative_path, must_exist=True, ) index_identity = _file_identity(index_path) if ( index_identity[2] != int(authority.index_bytes_t) or bytes.fromhex(_file_sha256(index_path)) != _tensor_digest(authority.index_sha256_t) ): raise RuntimeError("direct page pack index file authority differs") loaded_index = _load_index_file(index_path) if index is not None: supplied_index = validate_direct_page_pack_index_boundary(index) for field_name in DirectPagePackIndexPacket.__dataclass_fields__: loaded_value = getattr(loaded_index, field_name) supplied_value = getattr(supplied_index, field_name) if not torch.equal(loaded_value, supplied_value): raise RuntimeError( "direct page pack supplied index differs from its file" ) if _file_identity(index_path) != index_identity: raise RuntimeError("direct page pack index changed during validation") if ( not torch.equal( loaded_index.shard_sha256s_t, authority.shard_sha256s_t.detach().cpu(), ) or not torch.equal( loaded_index.shard_bytes_t, authority.shard_bytes_t.detach().cpu(), ) or int(loaded_index.object_bytes_t.sum()) != int(authority.logical_object_bytes_t) or loaded_index.page_ids_t.numel() != int(authority.page_count_t) or not torch.equal( loaded_index.page_ids_sha256_t, authority.page_ids_sha256_t.detach().cpu(), ) or not torch.equal( loaded_index.page_map_sha256_t, authority.page_map_sha256_t.detach().cpu(), ) or not torch.equal( loaded_index.pack_set_sha256_t, authority.pack_set_sha256_t.detach().cpu(), ) ): raise RuntimeError("direct page pack set authority differs from its index") for shard_index, (root, relative_path) in enumerate( zip( authority.shard_roots, authority.shard_relative_paths, strict=True, ) ): shard_path = _safe_relative_path( root, relative_path, must_exist=True, ) identity = shard_path.lstat() if ( identity.st_size != int(loaded_index.shard_bytes_t[shard_index]) or identity.st_size % DIRECT_PAGE_PACK_ALIGNMENT_BYTES or identity.st_nlink != 1 ): raise RuntimeError("direct page pack shard file authority differs") return loaded_index def load_direct_page_pack_index_boundary( authority: DirectPagePackSetAuthorityPacket, ) -> DirectPagePackIndexPacket: return validate_direct_page_pack_set_authority_boundary(authority) def _open_pack_descriptor(path: Path, *, writable: bool, direct_io: bool) -> int: flags = ( (os.O_WRONLY if writable else os.O_RDONLY) | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0) ) if direct_io: direct_flag = getattr(os, "O_DIRECT", 0) if not direct_flag: raise RuntimeError("direct page pack O_DIRECT is unavailable") flags |= direct_flag if writable: flags |= os.O_CREAT | os.O_EXCL try: return os.open(path, flags, 0o600) except OSError as error: raise RuntimeError("direct page pack descriptor open failed") from error def _pread_exact_into( descriptor: int, view: memoryview, *, offset: int, direct_io: bool, ) -> None: completed = 0 while completed < len(view): tail = view[completed:] try: count = os.preadv(descriptor, (tail,), offset + completed) finally: tail.release() if count < 1: raise RuntimeError("direct page pack read was incomplete") completed += count if ( direct_io and completed < len(view) and completed % DIRECT_PAGE_PACK_ALIGNMENT_BYTES ): raise RuntimeError("direct page pack direct read lost alignment") def _pwrite_exact_from( descriptor: int, view: memoryview, *, offset: int, ) -> None: completed = 0 while completed < len(view): tail = view[completed:] try: count = os.pwritev(descriptor, (tail,), offset + completed) finally: tail.release() if count < 1: raise RuntimeError("direct page pack write was incomplete") completed += count if ( completed < len(view) and completed % DIRECT_PAGE_PACK_ALIGNMENT_BYTES ): raise RuntimeError("direct page pack direct write lost alignment") def _shard_path( authority: DirectPagePackSetAuthorityPacket, shard_index: int, ) -> Path: return _safe_relative_path( authority.shard_roots[shard_index], authority.shard_relative_paths[shard_index], must_exist=True, ) def _read_selected_shard( *, authority: DirectPagePackSetAuthorityPacket, index: DirectPagePackIndexPacket, shard_index: int, requested_rows: tuple[tuple[int, int], ...], direct_io: bool, ) -> tuple[tuple[tuple[int, bytes], ...], int]: shard_path = _shard_path(authority, shard_index) identity_before = _file_identity(shard_path) descriptor = _open_pack_descriptor( shard_path, writable=False, direct_io=direct_io, ) ordered_rows = tuple( sorted( requested_rows, key=lambda pair: int(index.object_offsets_t[pair[1]]), ) ) groups: list[list[tuple[int, int]]] = [] for request in ordered_rows: row_index = request[1] row_span = int(index.object_spans_t[row_index]) if ( groups and int(index.object_offsets_t[row_index]) == int(index.object_offsets_t[groups[-1][-1][1]]) + int(index.object_spans_t[groups[-1][-1][1]]) and ( int(index.object_offsets_t[row_index]) + row_span - int(index.object_offsets_t[groups[-1][0][1]]) <= DIRECT_PAGE_PACK_MAX_COALESCED_READ_BYTES ) ): groups[-1].append(request) else: groups.append([request]) payloads: list[tuple[int, bytes]] = [] physical_read_bytes = 0 try: for group in groups: first_row = group[0][1] last_row = group[-1][1] group_offset = int(index.object_offsets_t[first_row]) group_end = ( int(index.object_offsets_t[last_row]) + int(index.object_spans_t[last_row]) ) group_bytes = group_end - group_offset bounce = mmap.mmap(-1, group_bytes, access=mmap.ACCESS_WRITE) view = memoryview(bounce) try: _pread_exact_into( descriptor, view, offset=group_offset, direct_io=direct_io, ) physical_read_bytes += group_bytes for request_index, row_index in group: local_offset = ( int(index.object_offsets_t[row_index]) - group_offset ) logical_bytes = int(index.object_bytes_t[row_index]) span = int(index.object_spans_t[row_index]) payload_view = view[ local_offset : local_offset + logical_bytes ] padding_view = view[ local_offset + logical_bytes : local_offset + span ] try: expected_sha256 = ( index.object_sha256s_t[row_index] .numpy() .tobytes(order="C") ) if hashlib.sha256(payload_view).digest() != expected_sha256: raise RuntimeError( "direct page pack selected object hash differs" ) if padding_view and padding_view.tobytes().strip(b"\x00"): raise RuntimeError( "direct page pack selected padding is nonzero" ) payloads.append( (request_index, payload_view.tobytes()) ) finally: payload_view.release() padding_view.release() finally: view.release() bounce.close() finally: os.close(descriptor) if _file_identity(shard_path) != identity_before: raise RuntimeError("direct page pack shard changed during selected read") return tuple(payloads), physical_read_bytes def read_direct_page_pack_selected_boundary( authority: DirectPagePackSetAuthorityPacket, selected_page_ids_t: torch.Tensor, *, direct_io: bool = True, ) -> DirectPagePackSelectedPacket: """Read selected complete objects, coalescing within and parallelizing shards.""" index = load_direct_page_pack_index_boundary(authority) selected_t = _stable_cpu_tensor( selected_page_ids_t, dtype=torch.long, ).reshape(-1) if ( selected_page_ids_t.dtype != torch.long or selected_t.numel() < 1 or torch.unique(selected_t).numel() != selected_t.numel() ): raise RuntimeError("direct page pack selection differs") positions_t = torch.searchsorted(index.page_ids_t, selected_t) bounded_positions_t = positions_t.clamp(max=index.page_ids_t.numel() - 1) if bool(positions_t.ge(index.page_ids_t.numel()).any()) or not torch.equal( index.page_ids_t[bounded_positions_t], selected_t, ): raise FileNotFoundError("selected direct page is absent from the pack set") requested_by_shard: dict[int, list[tuple[int, int]]] = {} for request_index, row_index_t in enumerate(positions_t): row_index = int(row_index_t) shard_index = int(index.shard_indices_t[row_index]) requested_by_shard.setdefault(shard_index, []).append( (request_index, row_index) ) shard_physical_read_bytes_t = torch.zeros( index.shard_sha256s_t.shape[0], dtype=torch.long, ) payload_by_request: dict[int, bytes] = {} with ThreadPoolExecutor( max_workers=len(requested_by_shard), thread_name_prefix="nnf-direct-pack-read", ) as executor: futures = { shard_index: executor.submit( _read_selected_shard, authority=authority, index=index, shard_index=shard_index, requested_rows=tuple(requested_rows), direct_io=direct_io, ) for shard_index, requested_rows in requested_by_shard.items() } for shard_index, future in futures.items(): payloads, physical_read_bytes = future.result() shard_physical_read_bytes_t[shard_index] = physical_read_bytes payload_by_request.update(payloads) selected_bytes_t = index.object_bytes_t[positions_t].clone() payload_offsets_t = torch.cat( ( torch.zeros(1, dtype=torch.long), torch.cumsum(selected_bytes_t, dim=0), ) ) object_payload_t = torch.empty( int(payload_offsets_t[-1]), dtype=torch.uint8, ) for request_index in range(selected_t.numel()): payload = payload_by_request.get(request_index) if payload is None: raise RuntimeError("direct page pack selected payload is absent") target_t = object_payload_t[ int(payload_offsets_t[request_index]) : int( payload_offsets_t[request_index + 1] ) ] target_t.copy_( torch.frombuffer(bytearray(payload), dtype=torch.uint8) ) return DirectPagePackSelectedPacket( page_ids_t=selected_t, object_sha256s_t=index.object_sha256s_t[positions_t].clone(), object_bytes_t=selected_bytes_t, payload_offsets_t=payload_offsets_t, object_payload_t=object_payload_t, shard_physical_read_bytes_t=shard_physical_read_bytes_t, direct_io_t=torch.tensor(direct_io, dtype=torch.bool), ) def _verify_one_shard_cold( *, authority: DirectPagePackSetAuthorityPacket, index: DirectPagePackIndexPacket, shard_index: int, direct_io: bool, ) -> tuple[int, int]: shard_path = _shard_path(authority, shard_index) identity_before = _file_identity(shard_path) shard_bytes = int(index.shard_bytes_t[shard_index]) descriptor = _open_pack_descriptor( shard_path, writable=False, direct_io=direct_io, ) bounce_bytes = min(DIRECT_PAGE_PACK_IO_WAVE_BYTES, shard_bytes) bounce = mmap.mmap(-1, bounce_bytes, access=mmap.ACCESS_WRITE) view = memoryview(bounce) shard_digest = hashlib.sha256() row_indices = tuple( int(row_index) for row_index in index.shard_indices_t.eq(shard_index) .nonzero(as_tuple=False) .reshape(-1) ) row_position = 0 object_digest = hashlib.sha256() object_verified = False offset = 0 started_ns = time.perf_counter_ns() try: while offset < shard_bytes: wave_bytes = min(bounce_bytes, shard_bytes - offset) wave = view[:wave_bytes] try: _pread_exact_into( descriptor, wave, offset=offset, direct_io=direct_io, ) shard_digest.update(wave) wave_end = offset + wave_bytes cursor = offset while cursor < wave_end: if row_position >= len(row_indices): raise RuntimeError( "direct page pack cold object layout differs" ) row_index = row_indices[row_position] object_start = int(index.object_offsets_t[row_index]) object_end = object_start + int( index.object_bytes_t[row_index] ) span_end = object_start + int( index.object_spans_t[row_index] ) if cursor < object_start or cursor >= span_end: raise RuntimeError( "direct page pack cold object layout differs" ) if cursor < object_end: payload_end = min(wave_end, object_end) payload_view = wave[ cursor - offset : payload_end - offset ] try: object_digest.update(payload_view) finally: payload_view.release() cursor = payload_end if cursor < object_end: continue if not object_verified: if object_digest.digest() != _tensor_digest( index.object_sha256s_t[row_index] ): raise RuntimeError( "direct page pack cold object hash differs" ) object_verified = True if cursor < span_end: padding_end = min(wave_end, span_end) padding_view = wave[ cursor - offset : padding_end - offset ] try: if padding_view.tobytes().strip(b"\x00"): raise RuntimeError( "direct page pack cold padding is nonzero" ) finally: padding_view.release() cursor = padding_end if cursor < span_end: continue row_position += 1 object_digest = hashlib.sha256() object_verified = False finally: wave.release() offset += wave_bytes finally: view.release() bounce.close() os.close(descriptor) elapsed_ns = time.perf_counter_ns() - started_ns if row_position != len(row_indices): raise RuntimeError("direct page pack cold object layout differs") if shard_digest.digest() != _tensor_digest( index.shard_sha256s_t[shard_index] ): raise RuntimeError("direct page pack cold shard hash differs") if _file_identity(shard_path) != identity_before: raise RuntimeError("direct page pack shard changed during cold read") return shard_bytes, elapsed_ns def verify_direct_page_pack_set_cold_boundary( authority: DirectPagePackSetAuthorityPacket, *, minimum_unique_logical_bytes_per_second: int = 0, direct_io: bool = True, ) -> DirectPagePackColdVerificationPacket: """Cold-read every unique shard once and report parallel ingestion speed.""" if minimum_unique_logical_bytes_per_second < 0: raise ValueError("direct page pack cold minimum rate differs") index = load_direct_page_pack_index_boundary(authority) shard_count = index.shard_sha256s_t.shape[0] wall_started_ns = time.perf_counter_ns() with ThreadPoolExecutor( max_workers=shard_count, thread_name_prefix="nnf-direct-pack-cold", ) as executor: futures = tuple( executor.submit( _verify_one_shard_cold, authority=authority, index=index, shard_index=shard_index, direct_io=direct_io, ) for shard_index in range(shard_count) ) rows = tuple(future.result() for future in futures) wall_elapsed_ns = time.perf_counter_ns() - wall_started_ns shard_physical_read_bytes_t = torch.tensor( [row[0] for row in rows], dtype=torch.long, ) shard_logical_rates_t = torch.tensor( [ int(index.shard_logical_bytes_t[shard_index]) * 1_000_000_000 // max(1, rows[shard_index][1]) for shard_index in range(shard_count) ], dtype=torch.long, ) logical_object_bytes = int(index.shard_logical_bytes_t.sum()) aggregate_rate = ( logical_object_bytes * 1_000_000_000 // max(1, wall_elapsed_ns) ) if aggregate_rate < minimum_unique_logical_bytes_per_second: raise RuntimeError( "direct page pack unique cold-ingestion rate is below authority" ) return DirectPagePackColdVerificationPacket( pack_set_sha256_t=index.pack_set_sha256_t.clone(), shard_sha256s_t=index.shard_sha256s_t.clone(), shard_physical_read_bytes_t=shard_physical_read_bytes_t, shard_logical_bytes_per_second_t=shard_logical_rates_t, logical_object_bytes_t=torch.tensor( logical_object_bytes, dtype=torch.long, ), physical_read_bytes_t=shard_physical_read_bytes_t.sum().reshape(()), elapsed_nanoseconds_t=torch.tensor(wall_elapsed_ns, dtype=torch.long), aggregate_unique_logical_bytes_per_second_t=torch.tensor( aggregate_rate, dtype=torch.long, ), direct_io_t=torch.tensor(direct_io, dtype=torch.bool), zero_padding_verified_t=torch.tensor(True, dtype=torch.bool), ) def _copy_file_buffered_atomic( source: Path, destination: Path, *, expected_sha256: bytes, ) -> bool: if destination.exists(): if ( destination.is_symlink() or bytes.fromhex(_file_sha256(destination)) != expected_sha256 ): raise RuntimeError("existing direct page pack index replica differs") return False temporary = destination.parent / ( f".{destination.name}.{os.getpid()}.{time.monotonic_ns()}.tmp" ) descriptor = os.open( temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0), 0o600, ) digest = hashlib.sha256() try: with source.open("rb") as source_handle: while True: chunk = source_handle.read(1024 * 1024) if not chunk: break digest.update(chunk) written = 0 while written < len(chunk): count = os.write(descriptor, chunk[written:]) if count < 1: raise RuntimeError( "direct page pack index replica write was incomplete" ) written += count os.fsync(descriptor) finally: os.close(descriptor) try: if digest.digest() != expected_sha256: raise RuntimeError("direct page pack index source hash differs") if destination.exists(): raise RuntimeError( "direct page pack index replica appeared during publication" ) os.rename(temporary, destination) _fsync_directory(destination.parent) finally: temporary.unlink(missing_ok=True) return True def _copy_one_shard_direct( *, source_path: Path, destination_path: Path, expected_sha256: bytes, physical_bytes: int, logical_bytes: int, ) -> tuple[bool, int, int]: if destination_path.exists(): identity_before = _file_identity(destination_path) if ( destination_path.is_symlink() or identity_before[2] != physical_bytes ): raise RuntimeError( "existing direct page pack shard replica differs" ) descriptor = _open_pack_descriptor( destination_path, writable=False, direct_io=True, ) bounce = mmap.mmap( -1, min(DIRECT_PAGE_PACK_IO_WAVE_BYTES, physical_bytes), access=mmap.ACCESS_WRITE, ) view = memoryview(bounce) digest = hashlib.sha256() offset = 0 try: while offset < physical_bytes: wave_bytes = min(len(view), physical_bytes - offset) wave = view[:wave_bytes] try: _pread_exact_into( descriptor, wave, offset=offset, direct_io=True, ) digest.update(wave) finally: wave.release() offset += wave_bytes finally: view.release() bounce.close() os.close(descriptor) if ( digest.digest() != expected_sha256 or _file_identity(destination_path) != identity_before ): raise RuntimeError( "existing direct page pack shard replica differs" ) return False, 0, 0 temporary = destination_path.parent / ( f".{destination_path.name}.{os.getpid()}.{time.monotonic_ns()}.tmp" ) source_descriptor = -1 destination_descriptor = -1 try: source_descriptor = _open_pack_descriptor( source_path, writable=False, direct_io=True, ) destination_descriptor = _open_pack_descriptor( temporary, writable=True, direct_io=True, ) if hasattr(os, "posix_fallocate"): os.posix_fallocate(destination_descriptor, 0, physical_bytes) else: os.ftruncate(destination_descriptor, physical_bytes) bounce = mmap.mmap( -1, min(DIRECT_PAGE_PACK_IO_WAVE_BYTES, physical_bytes), access=mmap.ACCESS_WRITE, ) view = memoryview(bounce) digest = hashlib.sha256() offset = 0 started_ns = time.perf_counter_ns() try: while offset < physical_bytes: wave_bytes = min(len(view), physical_bytes - offset) wave = view[:wave_bytes] try: _pread_exact_into( source_descriptor, wave, offset=offset, direct_io=True, ) digest.update(wave) _pwrite_exact_from( destination_descriptor, wave, offset=offset, ) finally: wave.release() offset += wave_bytes os.fsync(destination_descriptor) finally: view.release() bounce.close() elapsed_ns = time.perf_counter_ns() - started_ns os.close(destination_descriptor) destination_descriptor = -1 os.close(source_descriptor) source_descriptor = -1 if digest.digest() != expected_sha256: raise RuntimeError("direct page pack replica source hash differs") if destination_path.exists(): raise RuntimeError( "direct page pack shard replica appeared during publication" ) os.rename(temporary, destination_path) _fsync_directory(destination_path.parent) return True, logical_bytes, elapsed_ns finally: if destination_descriptor >= 0: os.close(destination_descriptor) if source_descriptor >= 0: os.close(source_descriptor) temporary.unlink(missing_ok=True) def replicate_direct_page_pack_set_durable_boundary( authority: DirectPagePackSetAuthorityPacket, *, destination_shard_roots: Sequence[Path], destination_index_root: Path, minimum_unique_logical_bytes_per_second: int = 0, ) -> DirectPagePackReplicaReceiptPacket: """Durably copy each unique shard once through O_DIRECT. This API creates one destination pack set. It records durable-copy speed but does not apply the 1 GB/s cold-ingestion gate by default. It never multiplies the numerator by replica count: each disjoint logical shard contributes at most its own logical object bytes. """ if minimum_unique_logical_bytes_per_second < 0: raise ValueError("direct page pack replica minimum rate differs") index = load_direct_page_pack_index_boundary(authority) destination_roots = tuple( _safe_root(Path(root), create=True) for root in destination_shard_roots ) if len(destination_roots) != len(authority.shard_roots): raise ValueError("direct page pack destination shard count differs") destination_paths = tuple( _safe_relative_path( destination_roots[shard_index], authority.shard_relative_paths[shard_index], must_exist=False, create_parent=True, ) for shard_index in range(len(destination_roots)) ) created_paths: list[Path] = [] index_created_path: Path | None = None wall_started_ns = time.perf_counter_ns() try: with ThreadPoolExecutor( max_workers=len(destination_roots), thread_name_prefix="nnf-direct-pack-replica", ) as executor: futures = tuple( executor.submit( _copy_one_shard_direct, source_path=_shard_path(authority, shard_index), destination_path=destination_paths[shard_index], expected_sha256=_tensor_digest( index.shard_sha256s_t[shard_index] ), physical_bytes=int(index.shard_bytes_t[shard_index]), logical_bytes=int( index.shard_logical_bytes_t[shard_index] ), ) for shard_index in range(len(destination_roots)) ) shard_rows = tuple(future.result() for future in futures) for shard_index, row in enumerate(shard_rows): if row[0]: created_paths.append(destination_paths[shard_index]) resolved_destination_index_root = _safe_root( destination_index_root, create=True, ) destination_index_path = _safe_relative_path( resolved_destination_index_root, authority.index_relative_path, must_exist=False, create_parent=True, ) source_index_path = _safe_relative_path( authority.index_root, authority.index_relative_path, must_exist=True, ) if _copy_file_buffered_atomic( source_index_path, destination_index_path, expected_sha256=_tensor_digest(authority.index_sha256_t), ): index_created_path = destination_index_path wall_elapsed_ns = time.perf_counter_ns() - wall_started_ns newly_written_logical_bytes_t = torch.tensor( [row[1] for row in shard_rows], dtype=torch.long, ) newly_written_shards_t = newly_written_logical_bytes_t.gt(0) total_newly_written = int(newly_written_logical_bytes_t.sum()) aggregate_rate = ( total_newly_written * 1_000_000_000 // max(1, wall_elapsed_ns) ) if ( total_newly_written < 1 and minimum_unique_logical_bytes_per_second > 0 ) or aggregate_rate < minimum_unique_logical_bytes_per_second: raise RuntimeError( "direct page pack durable replica rate is below authority" ) shard_rates_t = torch.tensor( [ row[1] * 1_000_000_000 // max(1, row[2]) if row[1] > 0 else 0 for row in shard_rows ], dtype=torch.long, ) destination_authority = replace( authority, shard_roots=destination_roots, index_root=resolved_destination_index_root, ) validate_direct_page_pack_set_authority_boundary( destination_authority ) cold_verification = verify_direct_page_pack_set_cold_boundary( destination_authority, minimum_unique_logical_bytes_per_second=0, direct_io=True, ) return DirectPagePackReplicaReceiptPacket( authority=destination_authority, newly_written_shards_t=newly_written_shards_t, newly_written_logical_bytes_t=newly_written_logical_bytes_t, shard_replica_bytes_per_second_t=shard_rates_t, total_newly_written_logical_bytes_t=torch.tensor( total_newly_written, dtype=torch.long, ), elapsed_nanoseconds_t=torch.tensor( wall_elapsed_ns, dtype=torch.long, ), aggregate_unique_logical_bytes_per_second_t=torch.tensor( aggregate_rate, dtype=torch.long, ), direct_io_t=torch.tensor(True, dtype=torch.bool), cold_verification=cold_verification, ) except Exception: touched_roots: set[Path] = set() for path in reversed(created_paths): if path.exists() and not path.is_symlink(): path.unlink() touched_roots.add(path.parent) if ( index_created_path is not None and index_created_path.exists() and not index_created_path.is_symlink() ): index_created_path.unlink() touched_roots.add(index_created_path.parent) for root in sorted(touched_roots, key=str): _fsync_directory(root) raise def replicate_direct_page_pack_shard_durable_boundary( authority: DirectPagePackSetAuthorityPacket, *, shard_index: int, destination_root: Path, ) -> DirectPagePackShardReplicaReceiptPacket: """Durably copy or verify one proof-only shard through O_DIRECT.""" index = load_direct_page_pack_index_boundary(authority) if ( isinstance(shard_index, bool) or shard_index < 0 or shard_index >= index.shard_sha256s_t.shape[0] ): raise ValueError("direct page pack proof shard index differs") root = _safe_root(destination_root, create=True) destination_path = _safe_relative_path( root, authority.shard_relative_paths[shard_index], must_exist=False, create_parent=True, ) written, logical_bytes, elapsed_ns = _copy_one_shard_direct( source_path=_shard_path(authority, shard_index), destination_path=destination_path, expected_sha256=_tensor_digest( index.shard_sha256s_t[shard_index] ), physical_bytes=int(index.shard_bytes_t[shard_index]), logical_bytes=int(index.shard_logical_bytes_t[shard_index]), ) return DirectPagePackShardReplicaReceiptPacket( shard_index_t=torch.tensor(shard_index, dtype=torch.long), shard_root=root, shard_relative_path=authority.shard_relative_paths[shard_index], shard_sha256_t=index.shard_sha256s_t[shard_index].clone(), shard_bytes_t=index.shard_bytes_t[shard_index].clone(), logical_object_bytes_t=index.shard_logical_bytes_t[ shard_index ].clone(), newly_written_t=torch.tensor(written, dtype=torch.bool), newly_written_logical_bytes_t=torch.tensor( logical_bytes, dtype=torch.long, ), elapsed_nanoseconds_t=torch.tensor( elapsed_ns, dtype=torch.long, ), direct_io_t=torch.tensor(True, dtype=torch.bool), )