From a47c95fb771377cb2525838f3d8291abf4a06a63 Mon Sep 17 00:00:00 2001 From: Nelson Spence Date: Fri, 24 Jul 2026 14:52:37 -0500 Subject: [PATCH 1/5] feat(manifest): add plan-verified streaming decode Signed-off-by: Nelson Spence --- CHANGELOG.md | 38 +- Cargo.lock | 3 + THREAT_MODEL.md | 10 +- docs/INDEX_PROVENANCE.md | 17 +- ordvec-manifest/Cargo.toml | 11 +- ordvec-manifest/README.md | 44 +- ordvec-manifest/src/lib.rs | 835 ++++++++++++++++++++++- ordvec-manifest/tests/deterministic.rs | 152 +++-- ordvec-manifest/tests/verified_decode.rs | 414 +++++++++++ src/quant.rs | 11 + src/rank_io.rs | 133 +++- src/sign_bitmap.rs | 11 + tests/index/loader_validation.rs | 193 +++++- tests/release_publish_invariants.py | 17 + 14 files changed, 1749 insertions(+), 140 deletions(-) create mode 100644 ordvec-manifest/tests/verified_decode.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 2bc72fda..12d17a93 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,27 @@ _No unreleased changes._ ### Added +- **Forward-only, exactly-sized decoders.** `RankQuant::read_from_sized` and + `SignBitmap::read_from_sized` accept a `Read` plus the exact encoded bytes + remaining at its current position. Existing seek-capable readers delegate to + them; truncation and trailing data remain fail-closed without a seek or read + beyond the declared record. +- **Plan-verified artifact decoding.** `ordvec-manifest` now exposes + `VerifiedLoadPlan::decode_primary_with` and + `VerifiedAuxiliaryArtifactPlan::decode_verified_with`. They open one + no-follow regular-file descriptor, cap and hash the decoder's reads, drain + unread bytes through a 64 KiB buffer, compare initial/final descriptor size, + and return typed access, type, stale, decoder, or incomplete-consumption + errors with documented precedence. Optional-absent auxiliaries are typed. +- **Composable current-schema loading.** Public + `read_manifest_bytes_bounded` and `parse_current_manifest_bytes` let policy + layers dispatch over one bounded read while retaining strict v2 parsing. + Manifest reads now reject a final symlink/reparse point or nonregular file. +- **Stable JSON-number feature profile.** The production manifest dependency + now enables `serde_json/float_roundtrip` rather than + `serde_json/arbitrary_precision`. Current-schema parsing, plus a public + validator for compatibility schemas, rejects lossy number tokens from the + original bounded bytes before nested `Value` fields can alias. - **`ordvec-manifest`: typed verification classification.** Every verification issue code is now a named `pub const` in a `codes` module (zero bare literals at emit sites), with a `#[non_exhaustive]` @@ -146,12 +167,13 @@ _No unreleased changes._ ### Fixed -- **`ordvec-manifest`: canonical content addresses are independent of - `serde_json` feature unification.** Nested extension and attestation maps are - recursively key-sorted and JSON numbers are normalized before serialization, - so downstream `preserve_order` / `arbitrary_precision` features cannot change - the same logical manifest's bytes. Manifest creation now rejects non-UTF-8 - artifact paths instead of embedding the lossy replacement character. +- **`ordvec-manifest`: canonical content addresses normalize nested JSON.** + Nested extension and attestation maps are recursively key-sorted and supplied + JSON numeric values are normalized before serialization. Release consumers + enforce one reviewed serde_json feature profile; current parsing rejects + lossy original number tokens before they become `Value`s. Manifest creation + now rejects non-UTF-8 artifact paths instead of embedding the lossy + replacement character. - **`ordvec-manifest`: manifest writes are atomic and fail before replacement.** `write_manifest_file` writes and syncs a same-directory temporary file before atomically replacing the destination, and rejects non-finite distortion @@ -164,6 +186,10 @@ _No unreleased changes._ ### Security +- Plan-verified decoding closes the verify/path-reopen gap for the bytes it + consumes on controlled local storage. It does not authenticate a producer, + confine hostile ancestor replacement, detect a mutation perfectly restored + between observations, or make a later file-backed mapping immutable. - The release CI gate now waits, with a 30-minute fail-closed deadline, for the latest exact-HEAD push runs instead of treating normal Actions visibility/in-progress delay as an immediate failure. It rejects terminal diff --git a/Cargo.lock b/Cargo.lock index b36e6f10..c14404b6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -863,13 +863,16 @@ dependencies = [ "chrono", "clap", "hex", + "libc", "ordvec", "rusqlite", + "rustix", "serde", "serde_json", "sha2", "tempfile", "uuid", + "windows-sys", ] [[package]] diff --git a/THREAT_MODEL.md b/THREAT_MODEL.md index aad3d559..c765fd29 100644 --- a/THREAT_MODEL.md +++ b/THREAT_MODEL.md @@ -414,8 +414,14 @@ CAN still cause I/O and CPU proportional to the byte size it declares and actually supplies on disk. The flat `ResourceLimits` byte caps are opt-in ceilings (unbounded by default) for deployments that must bound worst-case verification time on attacker-supplied bundles. A `VerifiedLoadPlan` remains -a verification snapshot, not a byte pin: bytes can change between -verification and use by a local actor with write access (see scope). +a verification snapshot, not a byte pin. Its plan-verified decode methods open +the final component without following it, require a regular file (and, on +Windows, a disk handle), and +return decoded data only after delivered bytes and initial/final descriptor +size match the plan. This closes the path-reopen race within the documented +trusted-parent-directory boundary. It does not confine hostile ancestor +replacement, authenticate the producer, detect a transient mutation perfectly +restored between observations, or make a long-lived file-backed mmap immutable. **THREAT-QUERY-002 (P3): Panic on contract violation in Rust server contexts.** Rust APIs fail fast on invalid contract input (non-finite floats, dimension / diff --git a/docs/INDEX_PROVENANCE.md b/docs/INDEX_PROVENANCE.md index 4e92113b..a7f57cf3 100644 --- a/docs/INDEX_PROVENANCE.md +++ b/docs/INDEX_PROVENANCE.md @@ -65,19 +65,20 @@ required artifacts and `--optional-aux NAME=PATH` for optional artifacts. Rust callers can use `verify_for_load(manifest_path, VerifyOptions)` to get a `VerifiedLoadPlan` containing the canonical artifact path, probed metadata, row-identity summary, auxiliary artifact states, and the full verification -report, then call `require_auxiliary(name)` for sidecars that must be present -before loading. Callers that already hold a `ManifestDocument` can use +report, then call `decode_primary_with` and an auxiliary plan's +`decode_verified_with` to decode through one no-follow, bounded, hashing file +descriptor. Callers that already hold a `ManifestDocument` can use `verify_document_for_load(&document, VerifyOptions)` without re-reading the -manifest file. The plan helpers do not call an ordvec loader, pin file -descriptors, or make mutable shared storage immutable; callers still own the -final policy decision and should load from the returned paths only while the -verified files remain under their control. +manifest file. The caller supplies the concrete decoder. The plan methods +establish that the bytes delivered during that call match the verified plan, +but do not make mutable shared storage or a later file-backed mmap immutable; +callers still own the final policy decision. `ordvec-manifest/README.md` shows the intended verify-then-immediate-load pattern, a concrete `manifest.json + index.ovrq + ids.bin` sidecar-backed bundle, and the stable report fields/codes for sidecar audit logs. If another process can mutate the manifest, index, row map, or sidecar between -verification and load, re-run `verify_for_load` at the load boundary or load -from immutable storage or a caller-owned loading path that pins bytes. +verification and use, decode through the plan at the load boundary or load +from immutable storage. The manifest verifier checks: diff --git a/ordvec-manifest/Cargo.toml b/ordvec-manifest/Cargo.toml index 40357d3e..73b8f222 100644 --- a/ordvec-manifest/Cargo.toml +++ b/ordvec-manifest/Cargo.toml @@ -32,11 +32,20 @@ hex = "0.4.3" ordvec = { version = "0.6.0", path = ".." } rusqlite = { version = "0.39.0", optional = true } serde = { version = "1.0", features = ["derive"] } -serde_json = { version = "1.0", features = ["arbitrary_precision"] } +serde_json = { version = "1.0", features = ["float_roundtrip"] } sha2 = "0.11.0" tempfile = "3.27.0" uuid = { version = "1.23.2", features = ["v4"] } +[target.'cfg(unix)'.dependencies] +libc = "0.2.183" + +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.61.2", features = ["Win32_Foundation", "Win32_Storage_FileSystem"] } + +[target.'cfg(target_os = "wasi")'.dependencies] +rustix = { version = "1.1.4", features = ["fs"] } + [dev-dependencies] serde_json = { version = "1.0", features = ["preserve_order"] } diff --git a/ordvec-manifest/README.md b/ordvec-manifest/README.md index 332f90a3..080e4b94 100644 --- a/ordvec-manifest/README.md +++ b/ordvec-manifest/README.md @@ -94,20 +94,28 @@ The helper verifies the manifest with the supplied options, fails closed by returning `VerifiedLoadPlanError::VerificationFailed(report)` when report errors exist, and otherwise returns the canonical primary artifact path, probed metadata, row-identity summary, declared auxiliary artifact states, and the full -report. Callers that already hold a `ManifestDocument` can use +report. `decode_primary_with` and an auxiliary plan's `decode_verified_with` +then consume one no-follow regular-file descriptor through a forward-only, +size-capped hashing reader. Callers that already hold a `ManifestDocument` can use `verify_document_for_load(&document, VerifyOptions)` without re-reading the -manifest file. These helpers do not call `Rank::load`, `RankQuant::load`, -`Bitmap::load`, or `SignBitmap::load`, and they do not pin file descriptors or -lock mutable storage. Callers should load from the returned paths immediately on -storage they control, or re-verify if the backing files can change between -verification and load. +manifest file. The verification helpers do not choose a core index decoder. +The plan-verified decode methods call the decoder supplied by the application, +then return its value only after the descriptor's size and delivered-byte +digest still match the plan. They trust caller-selected ancestor directories +and do not lock mutable storage or authenticate the manifest producer. Controlled-storage load pattern: ```rust let plan = ordvec_manifest::verify_for_load(&manifest_path, options)?; -let _app_ids = plan.require_auxiliary("app.ids")?; -let index = ordvec::RankQuant::load(plan.artifact_path())?; +let index = plan.decode_primary_with( + ordvec_manifest::ManifestIndexKind::RankQuant, + |reader, encoded_len| ordvec::RankQuant::read_from_sized(reader, encoded_len), +)?; +let app_ids = plan + .auxiliary_by_name("app.ids") + .ok_or("app.ids is not declared")? + .decode_verified_with(parse_caller_owned_ids_streaming)?; ``` Concrete sidecar-backed bundle pattern: @@ -167,10 +175,14 @@ let plan = ordvec_manifest::verify_for_load( let metadata = plan.metadata(); assert_eq!(metadata.vector_count, expected_rows); -let index = ordvec::RankQuant::load(plan.artifact_path())?; -let ids_path = plan.require_auxiliary("app.ids")?; -let ids_bytes = std::fs::read(ids_path)?; -let doc_ids = parse_caller_owned_ids(&ids_bytes)?; +let index = plan.decode_primary_with( + ordvec_manifest::ManifestIndexKind::RankQuant, + |reader, encoded_len| ordvec::RankQuant::read_from_sized(reader, encoded_len), +)?; +let doc_ids = plan + .auxiliary_by_name("app.ids") + .ok_or("app.ids is not declared")? + .decode_verified_with(parse_caller_owned_ids_streaming)?; ``` `ordvec-manifest` owns the path, size, SHA-256, and index-metadata checks. @@ -189,8 +201,12 @@ queue_for_later(plan); If a manifest or artifact lives on shared, writable, or otherwise mutable storage, re-run `verify_for_load` immediately before loading, load from -immutable storage, or use a caller-owned loading path that pins bytes. -`VerifiedLoadPlan` is not a byte pin. +immutable storage, or use the plan-verified decode methods. Those methods +detect mutations reflected in delivered bytes or descriptor sizes observed +during the call. They cannot detect a transient mutation perfectly restored +between those observations, and they do not make a long-lived file-backed mmap +immutable after open. `VerifiedLoadPlan` itself remains a snapshot, not a byte +pin. Verification uses bounded parser/report defaults on both CLI and library paths. Stable limit codes are part of the contract: diff --git a/ordvec-manifest/src/lib.rs b/ordvec-manifest/src/lib.rs index 6bdbb6ab..a1198951 100644 --- a/ordvec-manifest/src/lib.rs +++ b/ordvec-manifest/src/lib.rs @@ -15,7 +15,8 @@ //! for command-line use. //! //! ```no_run -//! use ordvec_manifest::{verify_for_load, VerifyOptions}; +//! use ordvec::RankQuant; +//! use ordvec_manifest::{verify_for_load, ManifestIndexKind, VerifyOptions}; //! //! # fn main() -> Result<(), Box> { //! let plan = verify_for_load("quickstart.manifest.json", VerifyOptions::default())?; @@ -24,8 +25,11 @@ //! plan.metadata().vector_count, //! plan.artifact_path().display() //! ); -//! // Load the index from plan.artifact_path() immediately, while the verified -//! // bytes remain under the caller's control. +//! let index = plan.decode_primary_with( +//! ManifestIndexKind::RankQuant, +//! |reader, encoded_len| RankQuant::read_from_sized(reader, encoded_len), +//! )?; +//! assert_eq!(index.len(), plan.metadata().vector_count); //! # Ok(()) //! # } //! ``` @@ -42,7 +46,9 @@ use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use std::collections::{BTreeMap, HashSet}; use std::fmt; -use std::fs::{self, File}; +#[cfg(any(unix, windows))] +use std::fs::OpenOptions; +use std::fs::{self, File, Metadata}; use std::io::{self, BufRead, BufReader, Read, Write}; use std::path::{Component, Path, PathBuf}; use uuid::Uuid; @@ -135,27 +141,8 @@ pub fn load_manifest_file_with_options( options: &VerifyOptions, ) -> Result { let path = path.as_ref(); - let manifest_bytes = read_bounded_file( - path, - options.limits.max_manifest_bytes, - codes::MANIFEST_FILE_TOO_LARGE, - "manifest file", - )?; - let manifest: IndexManifest = serde_json::from_slice(&manifest_bytes) - .map_err(|err| manifest_parse_error(&manifest_bytes, err))?; - // A genuine v1 manifest fails the parse above (its required `manifest_id` - // / `created_at` fields trip `deny_unknown_fields`), but a document that is - // v2-shaped yet labels itself an unsupported `schema_version` would - // otherwise load and only be caught at verify. Enforce the version at load - // so loading any non-current schema fails here, as documented. - if manifest.schema_version != SCHEMA_VERSION { - return Err(ManifestError::invalid(format!( - "manifest declares schema_version {:?} but this build supports \ - {SCHEMA_VERSION:?}; the manifest was written by an older or newer \ - manifest schema", - manifest.schema_version - ))); - } + let manifest_bytes = read_manifest_bytes_bounded(path, options.limits.max_manifest_bytes)?; + let manifest = parse_current_manifest_bytes(&manifest_bytes)?; let base_dir = path .parent() .filter(|p| !p.as_os_str().is_empty()) @@ -168,6 +155,130 @@ pub fn load_manifest_file_with_options( }) } +/// Read a manifest through a bounded, forward-only, final-component-safe file handle. +/// +/// The caller-selected parent directory remains trusted. The final component is +/// opened without following a Unix/WASI symlink or Windows reparse point and +/// must be a regular file; Windows additionally requires a disk handle. At +/// most `max_bytes + 1` bytes are read, permitting an exact limit check without +/// an unbounded allocation. +pub fn read_manifest_bytes_bounded( + path: impl AsRef, + max_bytes: u64, +) -> Result, ManifestError> { + read_bounded_file( + path.as_ref(), + max_bytes, + codes::MANIFEST_FILE_TOO_LARGE, + "manifest file", + ) +} + +/// Strictly parse bytes as the current manifest schema. +/// +/// Older or newer schema generations are rejected with schema-version context; +/// this function performs no filesystem access and no compatibility fallback. +pub fn parse_current_manifest_bytes(bytes: &[u8]) -> Result { + validate_manifest_json_number_tokens(bytes)?; + let manifest: IndexManifest = + serde_json::from_slice(bytes).map_err(|err| manifest_parse_error(bytes, err))?; + if manifest.schema_version != SCHEMA_VERSION { + return Err(ManifestError::invalid(format!( + "manifest declares schema_version {:?} but this build supports \ + {SCHEMA_VERSION:?}; the manifest was written by an older or newer \ + manifest schema", + manifest.schema_version + ))); + } + Ok(manifest) +} + +/// Reject JSON number tokens that `serde_json` could only retain by changing +/// their exact decimal value. +/// +/// OrdinalDB-compatible dependency graphs deliberately use +/// `serde_json/float_roundtrip` and reject representation-changing features +/// such as `arbitrary_precision`. Under that profile, an oversized or overly +/// precise token can otherwise be rounded before it reaches a nested +/// [`serde_json::Value`]. Call this on the original bytes before deserializing +/// a compatible legacy schema so schema dispatch cannot introduce aliases. +pub fn validate_manifest_json_number_tokens(bytes: &[u8]) -> Result<(), ManifestError> { + let mut offset = 0usize; + while offset < bytes.len() { + match bytes[offset] { + b'"' => { + offset += 1; + while offset < bytes.len() { + match bytes[offset] { + b'\\' => { + // Escapes consume the following byte. Full escape + // syntax remains the JSON parser's responsibility; + // this scan only keeps number-looking string bytes + // out of the numeric-token path. + offset = offset.saturating_add(2); + } + b'"' => { + offset += 1; + break; + } + _ => offset += 1, + } + } + } + b'-' | b'0'..=b'9' => { + let start = offset; + offset += 1; + while offset < bytes.len() + && !matches!(bytes[offset], b',' | b']' | b'}') + && !bytes[offset].is_ascii_whitespace() + { + offset += 1; + } + let token = std::str::from_utf8(&bytes[start..offset]).map_err(|_| { + ManifestError::invalid("manifest contains a non-UTF-8 JSON number token") + })?; + validate_lossless_json_number_token(token)?; + } + _ => offset += 1, + } + } + Ok(()) +} + +fn validate_lossless_json_number_token(token: &str) -> Result<(), ManifestError> { + let identity = decimal_identity(token).ok_or_else(|| { + ManifestError::invalid(format!( + "manifest contains JSON number {token:?} that cannot be represented canonically" + )) + })?; + + // Integer syntax within serde_json's native signed/unsigned domain is + // retained exactly without going through f64. Decimal/exponent syntax is + // always parsed as f64, so it must pass the round-trip check below even + // when its mathematical value happens to be integral. + let integer_syntax = !token.bytes().any(|byte| matches!(byte, b'.' | b'e' | b'E')); + if integer_syntax && canonical_integer_from_identity(&identity).is_some() { + return Ok(()); + } + + let value = token.parse::().map_err(|_| { + ManifestError::invalid(format!( + "manifest contains JSON number {token:?} that cannot be represented canonically" + )) + })?; + let normalized = serde_json::Number::from_f64(value).ok_or_else(|| { + ManifestError::invalid(format!( + "manifest contains JSON number {token:?} that cannot be represented canonically" + )) + })?; + if decimal_identity(&normalized.to_string()).as_ref() != Some(&identity) { + return Err(ManifestError::invalid(format!( + "manifest contains JSON number {token:?} outside the exact canonical i64/u64/f64 domain" + ))); + } + Ok(()) +} + /// Wraps a manifest parse failure with schema-version context. Old or new /// schema generations fail the strict `deny_unknown_fields` parse before the /// `schema_version` field is ever validated, so a targeted probe of that one @@ -192,13 +303,230 @@ fn manifest_parse_error(manifest_bytes: &[u8], err: serde_json::Error) -> Manife ManifestError::Json(err) } +/// Filesystem operation at which a plan-verified artifact became inaccessible. +#[derive(Copy, Clone, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub enum VerifiedArtifactAccessStage { + Open, + InitialMetadata, + Read, + FinalMetadata, +} + +impl fmt::Display for VerifiedArtifactAccessStage { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Open => f.write_str("open"), + Self::InitialMetadata => f.write_str("initial metadata"), + Self::Read => f.write_str("read"), + Self::FinalMetadata => f.write_str("final metadata"), + } + } +} + +/// Why an opened final path component is not an acceptable artifact file. +#[derive(Copy, Clone, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub enum VerifiedArtifactTypeRejection { + FinalSymlinkOrReparsePoint, + NonRegularFile, + NonDiskFile, + UnsupportedPlatform, +} + +impl fmt::Display for VerifiedArtifactTypeRejection { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::FinalSymlinkOrReparsePoint => { + f.write_str("the final component is a symlink or reparse point") + } + Self::NonRegularFile => f.write_str("the opened component is not a regular file"), + Self::NonDiskFile => f.write_str("the opened handle is not a disk file"), + Self::UnsupportedPlatform => { + f.write_str("the target platform cannot provide a no-follow final-component open") + } + } + } +} + +enum OpenRegularFileError { + Access { + stage: VerifiedArtifactAccessStage, + source: io::Error, + }, + Type(VerifiedArtifactTypeRejection), +} + +#[cfg(unix)] +fn is_final_symlink_open_error(source: &io::Error) -> bool { + let Some(code) = source.raw_os_error() else { + return false; + }; + if code == libc::ELOOP { + return true; + } + #[cfg(any(target_os = "freebsd", target_os = "dragonfly"))] + if code == libc::EMLINK { + return true; + } + #[cfg(target_os = "netbsd")] + if code == libc::EFTYPE { + return true; + } + false +} + +#[cfg(unix)] +fn open_regular_file_no_follow(path: &Path) -> Result<(File, Metadata), OpenRegularFileError> { + use std::os::unix::fs::OpenOptionsExt; + + let file = OpenOptions::new() + .read(true) + .custom_flags(libc::O_NOFOLLOW | libc::O_NONBLOCK) + .open(path) + .map_err(|source| { + if is_final_symlink_open_error(&source) { + OpenRegularFileError::Type( + VerifiedArtifactTypeRejection::FinalSymlinkOrReparsePoint, + ) + } else { + OpenRegularFileError::Access { + stage: VerifiedArtifactAccessStage::Open, + source, + } + } + })?; + let metadata = file + .metadata() + .map_err(|source| OpenRegularFileError::Access { + stage: VerifiedArtifactAccessStage::InitialMetadata, + source, + })?; + if !metadata.file_type().is_file() { + return Err(OpenRegularFileError::Type( + VerifiedArtifactTypeRejection::NonRegularFile, + )); + } + Ok((file, metadata)) +} + +#[cfg(windows)] +fn open_regular_file_no_follow(path: &Path) -> Result<(File, Metadata), OpenRegularFileError> { + use std::os::windows::fs::{MetadataExt, OpenOptionsExt}; + use std::os::windows::io::AsRawHandle; + use windows_sys::Win32::Foundation::{GetLastError, SetLastError, ERROR_SUCCESS}; + use windows_sys::Win32::Storage::FileSystem::{ + GetFileType, FILE_ATTRIBUTE_REPARSE_POINT, FILE_FLAG_BACKUP_SEMANTICS, + FILE_FLAG_OPEN_REPARSE_POINT, FILE_TYPE_DISK, + }; + + let file = OpenOptions::new() + .read(true) + .custom_flags(FILE_FLAG_OPEN_REPARSE_POINT | FILE_FLAG_BACKUP_SEMANTICS) + .open(path) + .map_err(|source| OpenRegularFileError::Access { + stage: VerifiedArtifactAccessStage::Open, + source, + })?; + let metadata = file + .metadata() + .map_err(|source| OpenRegularFileError::Access { + stage: VerifiedArtifactAccessStage::InitialMetadata, + source, + })?; + if metadata.file_attributes() & FILE_ATTRIBUTE_REPARSE_POINT != 0 { + return Err(OpenRegularFileError::Type( + VerifiedArtifactTypeRejection::FinalSymlinkOrReparsePoint, + )); + } + // `GetFileType` uses the last-error slot to disambiguate a legitimate + // FILE_TYPE_UNKNOWN result, but it need not clear a stale value on + // success. Clear it immediately before the query. + // SAFETY: these calls only access the current thread's last-error slot; + // `file` owns a live HANDLE for the duration of `GetFileType`. + unsafe { SetLastError(ERROR_SUCCESS) }; + let file_type = unsafe { GetFileType(file.as_raw_handle().cast()) }; + if file_type == windows_sys::Win32::Storage::FileSystem::FILE_TYPE_UNKNOWN { + // `FILE_TYPE_UNKNOWN` can be a legitimate handle classification or a + // failed query. Windows requires consulting the thread-local last + // error immediately to distinguish the two cases. + // SAFETY: `GetLastError` reads the current thread's last-error value. + let code = unsafe { GetLastError() }; + if code != ERROR_SUCCESS { + return Err(OpenRegularFileError::Access { + stage: VerifiedArtifactAccessStage::InitialMetadata, + source: io::Error::from_raw_os_error(code as i32), + }); + } + } + if file_type != FILE_TYPE_DISK { + return Err(OpenRegularFileError::Type( + VerifiedArtifactTypeRejection::NonDiskFile, + )); + } + if !metadata.file_type().is_file() { + return Err(OpenRegularFileError::Type( + VerifiedArtifactTypeRejection::NonRegularFile, + )); + } + Ok((file, metadata)) +} + +#[cfg(target_os = "wasi")] +fn open_regular_file_no_follow(path: &Path) -> Result<(File, Metadata), OpenRegularFileError> { + use rustix::fs::{openat, Mode, OFlags, CWD}; + + let descriptor = openat( + CWD, + path, + OFlags::RDONLY | OFlags::NOFOLLOW | OFlags::NONBLOCK, + Mode::empty(), + ) + .map_err(|error| { + if error == rustix::io::Errno::LOOP { + OpenRegularFileError::Type(VerifiedArtifactTypeRejection::FinalSymlinkOrReparsePoint) + } else { + OpenRegularFileError::Access { + stage: VerifiedArtifactAccessStage::Open, + source: io::Error::from_raw_os_error(error.raw_os_error()), + } + } + })?; + let file = File::from(descriptor); + let metadata = file + .metadata() + .map_err(|source| OpenRegularFileError::Access { + stage: VerifiedArtifactAccessStage::InitialMetadata, + source, + })?; + if !metadata.file_type().is_file() { + return Err(OpenRegularFileError::Type( + VerifiedArtifactTypeRejection::NonRegularFile, + )); + } + Ok((file, metadata)) +} + +#[cfg(not(any(unix, windows, target_os = "wasi")))] +fn open_regular_file_no_follow(_path: &Path) -> Result<(File, Metadata), OpenRegularFileError> { + Err(OpenRegularFileError::Type( + VerifiedArtifactTypeRejection::UnsupportedPlatform, + )) +} + fn read_bounded_file( path: &Path, max_bytes: u64, code: &'static str, context: &'static str, ) -> Result, ManifestError> { - let mut file = File::open(path)?; + let (mut file, _) = open_regular_file_no_follow(path).map_err(|err| match err { + OpenRegularFileError::Access { source, .. } => ManifestError::Io(source), + OpenRegularFileError::Type(rejection) => ManifestError::invalid(format!( + "{} is not an acceptable regular manifest file: {rejection}", + path.display() + )), + })?; let max_len = usize::try_from(max_bytes).map_err(|_| { ManifestError::limit_exceeded( code, @@ -3137,6 +3465,132 @@ fn require_manifest_coverage(kind: CoreIndexKind) -> Result<(), UnsupportedCoreM } } +/// Stable identity of one artifact covered by a verified load plan. +#[derive(Clone, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub struct VerifiedArtifactIdentity { + pub name: String, + pub kind: VerifiedArtifactKind, + pub expected_size_bytes: u64, + pub expected_sha256: String, +} + +/// Role and, for a primary artifact, persisted index kind. +#[derive(Clone, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub enum VerifiedArtifactKind { + Primary(ManifestIndexKind), + Auxiliary, +} + +/// An observed departure from the bytes recorded in a verified load plan. +#[derive(Clone, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub enum VerifiedArtifactChange { + InitialSize { expected: u64, observed: u64 }, + FinalSize { expected: u64, observed: u64 }, + Digest { expected: String, observed: String }, +} + +/// Failure while consuming an artifact recorded in a verified load plan. +/// +/// Stale-content errors take precedence over decoder and incomplete-consumption +/// errors after all declared bytes have been drained. Filesystem access and type +/// failures take precedence over content classification. +#[derive(Debug)] +#[non_exhaustive] +pub enum VerifiedArtifactUseError { + KindMismatch { + identity: VerifiedArtifactIdentity, + expected: ManifestIndexKind, + observed: ManifestIndexKind, + }, + OptionalAbsent { + name: String, + }, + Access { + identity: VerifiedArtifactIdentity, + stage: VerifiedArtifactAccessStage, + source: io::Error, + }, + TypeRejected { + identity: VerifiedArtifactIdentity, + rejection: VerifiedArtifactTypeRejection, + }, + Stale { + identity: VerifiedArtifactIdentity, + changes: Vec, + }, + Decoder { + identity: VerifiedArtifactIdentity, + source: E, + }, + IncompleteConsumption { + identity: VerifiedArtifactIdentity, + consumed: u64, + expected: u64, + }, +} + +impl fmt::Display for VerifiedArtifactUseError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::KindMismatch { + identity, + expected, + observed, + } => write!( + f, + "artifact {:?} has kind {observed:?}, expected {expected:?}", + identity.name + ), + Self::OptionalAbsent { name } => { + write!(f, "optional auxiliary artifact {name:?} is absent") + } + Self::Access { + identity, + stage, + source, + } => write!( + f, + "cannot use artifact {:?} during {stage}: {source}", + identity.name + ), + Self::TypeRejected { + identity, + rejection, + } => write!(f, "artifact {:?} was rejected: {rejection}", identity.name), + Self::Stale { identity, changes } => write!( + f, + "artifact {:?} is stale relative to the verified plan ({changes:?})", + identity.name + ), + Self::Decoder { identity, source } => { + write!(f, "artifact {:?} failed to decode: {source}", identity.name) + } + Self::IncompleteConsumption { + identity, + consumed, + expected, + } => write!( + f, + "artifact {:?} decoder consumed {consumed} of {expected} encoded bytes", + identity.name + ), + } + } +} + +impl std::error::Error for VerifiedArtifactUseError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Access { source, .. } => Some(source), + Self::Decoder { source, .. } => Some(source), + _ => None, + } + } +} + /// Verified paths and metadata for a caller-managed load. /// /// A `VerifiedLoadPlan` means the manifest, primary artifact, row-identity @@ -3150,6 +3604,7 @@ fn require_manifest_coverage(kind: CoreIndexKind) -> Result<(), UnsupportedCoreM pub struct VerifiedLoadPlan { manifest_path: Option, artifact_path: PathBuf, + primary_identity: VerifiedArtifactIdentity, metadata: MetadataReport, row_identity: VerifiedRowIdentityPlan, auxiliary_artifacts: Vec, @@ -3180,6 +3635,22 @@ impl VerifiedLoadPlan { message: "verified report is missing probed artifact metadata".to_string(), } })?; + let primary_identity = VerifiedArtifactIdentity { + name: "primary".to_string(), + kind: VerifiedArtifactKind::Primary(metadata.kind), + expected_size_bytes: report.artifact.size_bytes.ok_or_else(|| { + VerifiedLoadPlanError::IncompletePlan { + report: Box::new(report.clone()), + message: "verified report is missing primary artifact size".to_string(), + } + })?, + expected_sha256: report.artifact.sha256.clone().ok_or_else(|| { + VerifiedLoadPlanError::IncompletePlan { + report: Box::new(report.clone()), + message: "verified report is missing primary artifact digest".to_string(), + } + })?, + }; let row_identity = VerifiedRowIdentityPlan::from_report(paths.row_identity_path.as_ref(), &report)?; let auxiliary_artifacts = report @@ -3201,6 +3672,7 @@ impl VerifiedLoadPlan { Ok(Self { manifest_path: document.source_path.clone(), artifact_path, + primary_identity, metadata, row_identity, auxiliary_artifacts, @@ -3221,6 +3693,31 @@ impl VerifiedLoadPlan { &self.artifact_path } + pub fn primary_identity(&self) -> &VerifiedArtifactIdentity { + &self.primary_identity + } + + /// Decode the primary artifact from one no-follow descriptor while checking + /// its plan-recorded kind, size, and digest. + /// + /// The `u64` passed to `decoder` is the exact encoded byte count remaining + /// at the reader's current position. The reader is forward-only and never + /// accesses bytes beyond that boundary. + pub fn decode_primary_with( + &self, + expected_kind: ManifestIndexKind, + decoder: impl FnOnce(&mut dyn Read, u64) -> Result, + ) -> Result> { + if self.metadata.kind != expected_kind { + return Err(VerifiedArtifactUseError::KindMismatch { + identity: self.primary_identity.clone(), + expected: expected_kind, + observed: self.metadata.kind, + }); + } + decode_plan_verified(&self.artifact_path, self.primary_identity.clone(), decoder) + } + pub fn metadata(&self) -> &MetadataReport { &self.metadata } @@ -3415,6 +3912,33 @@ impl VerifiedAuxiliaryArtifactPlan { }); } }; + let (sha256, size_bytes) = + match entry.state { + AuxiliaryArtifactState::Verified => ( + Some(entry.sha256.clone().ok_or_else(|| { + VerifiedLoadPlanError::IncompletePlan { + report: Box::new(report.clone()), + message: format!( + "verified auxiliary artifact {:?} is missing its digest", + entry.name + ), + } + })?), + Some(entry.size_bytes.ok_or_else(|| { + VerifiedLoadPlanError::IncompletePlan { + report: Box::new(report.clone()), + message: format!( + "verified auxiliary artifact {:?} is missing its size", + entry.name + ), + } + })?), + ), + AuxiliaryArtifactState::OptionalAbsent => (None, None), + AuxiliaryArtifactState::MissingRequired | AuxiliaryArtifactState::Failed => { + unreachable!("non-loadable auxiliary states returned above") + } + }; Ok(Self { name: entry.name.clone(), @@ -3422,8 +3946,8 @@ impl VerifiedAuxiliaryArtifactPlan { required: entry.required, state: entry.state, reason_code: entry.reason_code.clone(), - sha256: entry.sha256.clone(), - size_bytes: entry.size_bytes, + sha256, + size_bytes, }) } @@ -3454,6 +3978,40 @@ impl VerifiedAuxiliaryArtifactPlan { pub fn size_bytes(&self) -> Option { self.size_bytes } + + pub fn identity(&self) -> Option { + Some(VerifiedArtifactIdentity { + name: self.name.clone(), + kind: VerifiedArtifactKind::Auxiliary, + expected_size_bytes: self.size_bytes?, + expected_sha256: self.sha256.clone()?, + }) + } + + /// Decode this auxiliary artifact from one no-follow descriptor while + /// checking its plan-recorded size and digest. + /// + /// Optional-absent artifacts return [`VerifiedArtifactUseError::OptionalAbsent`] + /// before any path is opened. The `u64` passed to `decoder` is the exact + /// encoded byte count remaining at the reader's current position. + pub fn decode_verified_with( + &self, + decoder: impl FnOnce(&mut dyn Read, u64) -> Result, + ) -> Result> { + if self.state == AuxiliaryArtifactState::OptionalAbsent { + return Err(VerifiedArtifactUseError::OptionalAbsent { + name: self.name.clone(), + }); + } + let identity = self + .identity() + .expect("verified auxiliary plans always carry a size and digest"); + let path = self + .path + .as_deref() + .expect("verified auxiliary plans always carry a captured path"); + decode_plan_verified(path, identity, decoder) + } } #[derive(Debug)] @@ -3497,6 +4055,162 @@ impl From for VerifiedLoadPlanError { } } +const VERIFIED_DECODE_DRAIN_BYTES: usize = 64 * 1024; + +struct DigestingBoundedReader<'a, R: Read + ?Sized> { + reader: &'a mut R, + remaining: u64, + consumed: u64, + hasher: Sha256, + read_error: Option, +} + +impl<'a, R: Read + ?Sized> DigestingBoundedReader<'a, R> { + fn new(reader: &'a mut R, encoded_len: u64) -> Self { + Self { + reader, + remaining: encoded_len, + consumed: 0, + hasher: Sha256::new(), + read_error: None, + } + } + + fn finish(self) -> (String, Option) { + (hex::encode(self.hasher.finalize()), self.read_error) + } +} + +impl Read for DigestingBoundedReader<'_, R> { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + if self.remaining == 0 || buf.is_empty() { + return Ok(0); + } + let limit = usize::try_from(self.remaining.min(buf.len() as u64)) + .expect("bounded read length never exceeds the caller buffer"); + loop { + match self.reader.read(&mut buf[..limit]) { + Ok(read) => { + self.hasher.update(&buf[..read]); + self.remaining -= read as u64; + self.consumed += read as u64; + return Ok(read); + } + Err(source) if source.kind() == io::ErrorKind::Interrupted => continue, + Err(source) => { + if self.read_error.is_none() { + self.read_error = Some(clone_io_error(&source)); + } + return Err(source); + } + } + } + } +} + +fn clone_io_error(source: &io::Error) -> io::Error { + match source.raw_os_error() { + Some(code) => io::Error::from_raw_os_error(code), + None => io::Error::new(source.kind(), source.to_string()), + } +} + +fn decode_plan_verified( + path: &Path, + identity: VerifiedArtifactIdentity, + decoder: impl FnOnce(&mut dyn Read, u64) -> Result, +) -> Result> { + let (mut file, initial_metadata) = match open_regular_file_no_follow(path) { + Ok(opened) => opened, + Err(OpenRegularFileError::Access { stage, source }) => { + return Err(VerifiedArtifactUseError::Access { + identity, + stage, + source, + }); + } + Err(OpenRegularFileError::Type(rejection)) => { + return Err(VerifiedArtifactUseError::TypeRejected { + identity, + rejection, + }); + } + }; + + let expected_size = identity.expected_size_bytes; + let initial_size = initial_metadata.len(); + if initial_size != expected_size { + return Err(VerifiedArtifactUseError::Stale { + identity, + changes: vec![VerifiedArtifactChange::InitialSize { + expected: expected_size, + observed: initial_size, + }], + }); + } + + let mut reader = DigestingBoundedReader::new(&mut file, expected_size); + let decoded = decoder(&mut reader, expected_size); + let decoder_consumed = reader.consumed; + + let mut scratch = [0u8; VERIFIED_DECODE_DRAIN_BYTES]; + while reader.remaining != 0 { + match reader.read(&mut scratch) { + Ok(0) => break, + Ok(_) => {} + Err(source) if source.kind() == io::ErrorKind::Interrupted => continue, + Err(_) => break, + } + } + let (observed_digest, read_error) = reader.finish(); + + if let Some(source) = read_error { + return Err(VerifiedArtifactUseError::Access { + identity, + stage: VerifiedArtifactAccessStage::Read, + source, + }); + } + + let final_size = file + .metadata() + .map_err(|source| VerifiedArtifactUseError::Access { + identity: identity.clone(), + stage: VerifiedArtifactAccessStage::FinalMetadata, + source, + })? + .len(); + + let mut changes = Vec::new(); + if final_size != expected_size { + changes.push(VerifiedArtifactChange::FinalSize { + expected: expected_size, + observed: final_size, + }); + } + if observed_digest != identity.expected_sha256 { + changes.push(VerifiedArtifactChange::Digest { + expected: identity.expected_sha256.clone(), + observed: observed_digest, + }); + } + if !changes.is_empty() { + return Err(VerifiedArtifactUseError::Stale { identity, changes }); + } + + match decoded { + Err(source) => Err(VerifiedArtifactUseError::Decoder { identity, source }), + Ok(_) if decoder_consumed != expected_size => { + Err(VerifiedArtifactUseError::IncompleteConsumption { + identity, + consumed: decoder_consumed, + expected: expected_size, + }) + } + Ok(value) => Ok(value), + } +} + fn report_issue_summary(errors: &[ReportIssue]) -> String { if errors.is_empty() { return String::new(); @@ -4455,10 +5169,12 @@ fn create_auxiliary_artifacts( /// schema-version event, not a cosmetic change. /// /// Nested [`serde_json::Value`] maps carried in `extensions` / `attestations` -/// are recursively key-sorted and their numbers are normalized before -/// serialization. The stored bytes therefore do not depend on downstream -/// feature unification of `serde_json/preserve_order` or -/// `serde_json/arbitrary_precision`. +/// are recursively key-sorted and their supplied numeric values are normalized +/// before serialization. File-loading policy must validate original number +/// tokens through [`parse_current_manifest_bytes`] (or +/// [`validate_manifest_json_number_tokens`] for a compatibility schema) before +/// deserialization. A writer cannot recover a source lexeme that some earlier, +/// caller-owned parser already rounded into a `Value`. /// /// The destination is replaced atomically from a temporary file in the same /// directory after the bytes and portable permission bits are synced. New @@ -5274,6 +5990,61 @@ fn hex_digest_eq(a: &str, b: &str) -> bool { mod tests { use super::*; + #[cfg(unix)] + #[test] + fn unix_nofollow_symlink_errnos_have_typed_classification() { + assert!(is_final_symlink_open_error(&io::Error::from_raw_os_error( + libc::ELOOP, + ))); + assert!(!is_final_symlink_open_error(&io::Error::from_raw_os_error( + libc::EACCES + ),)); + + #[cfg(any(target_os = "freebsd", target_os = "dragonfly"))] + assert!(is_final_symlink_open_error(&io::Error::from_raw_os_error( + libc::EMLINK, + ))); + #[cfg(target_os = "netbsd")] + assert!(is_final_symlink_open_error(&io::Error::from_raw_os_error( + libc::EFTYPE, + ))); + } + + #[test] + fn digesting_bounded_reader_retries_interrupted_reads() { + struct InterruptOnce { + inner: R, + interrupted: bool, + } + + impl Read for InterruptOnce { + fn read(&mut self, buffer: &mut [u8]) -> io::Result { + if !self.interrupted { + self.interrupted = true; + return Err(io::Error::from(io::ErrorKind::Interrupted)); + } + self.inner.read(buffer) + } + } + + let bytes = b"plan-verified bytes survive an interrupted read"; + let mut source = InterruptOnce { + inner: io::Cursor::new(bytes), + interrupted: false, + }; + let mut reader = DigestingBoundedReader::new(&mut source, bytes.len() as u64); + let mut delivered = Vec::new(); + reader.read_to_end(&mut delivered).unwrap(); + assert_eq!(reader.remaining, 0); + assert_eq!(reader.consumed, bytes.len() as u64); + let (digest, read_error) = reader.finish(); + + assert!(source.interrupted); + assert_eq!(delivered, bytes); + assert!(read_error.is_none()); + assert_eq!(digest, hex::encode(Sha256::digest(bytes))); + } + #[test] fn canonical_manifest_path_form_is_policy_aware() { assert!(is_canonical_manifest_path("index.ovrq", false)); diff --git a/ordvec-manifest/tests/deterministic.rs b/ordvec-manifest/tests/deterministic.rs index 950e18c1..06c92973 100644 --- a/ordvec-manifest/tests/deterministic.rs +++ b/ordvec-manifest/tests/deterministic.rs @@ -1,11 +1,11 @@ use ordvec::RankQuant; use ordvec_manifest::{ create_manifest_for_index, create_manifest_for_index_with_options, load_manifest_file, - sha256_file, verify_manifest_with_base, write_manifest_file, CreateAuxiliaryArtifact, - CreateManifestOptions, CreateRowIdentity, DistortionBounds, DistortionEvidence, - DistortionEvidenceKind, DistortionScope, EncoderDistortionProfileRef, EncoderSpec, - IndexManifest, ManifestError, MetricSpec, VerifyOptions, ENCODER_DISTORTION_SCHEMA_VERSION, - SCHEMA_VERSION, + parse_current_manifest_bytes, sha256_file, verify_manifest_with_base, write_manifest_file, + CreateAuxiliaryArtifact, CreateManifestOptions, CreateRowIdentity, DistortionBounds, + DistortionEvidence, DistortionEvidenceKind, DistortionScope, EncoderDistortionProfileRef, + EncoderSpec, IndexManifest, ManifestError, MetricSpec, VerifyOptions, + ENCODER_DISTORTION_SCHEMA_VERSION, SCHEMA_VERSION, }; use serde_json::{json, Map, Value}; use std::fs; @@ -58,6 +58,16 @@ fn manifest_bytes_with_extension_number( dir: &Path, number: &str, ) -> Result, ManifestError> { + let (manifest_path, raw) = raw_manifest_bytes_with_extension_number(dir, number)?; + let manifest = parse_current_manifest_bytes(&raw)?; + write_manifest_file(&manifest, &manifest_path)?; + fs::read(manifest_path).map_err(ManifestError::Io) +} + +fn raw_manifest_bytes_with_extension_number( + dir: &Path, + number: &str, +) -> Result<(PathBuf, Vec), ManifestError> { let index = write_index(dir); let manifest_path = dir.join("manifest.json"); let mut manifest = create_manifest_for_index( @@ -66,12 +76,22 @@ fn manifest_bytes_with_extension_number( "test-embedding", &manifest_path, )?; + const MARKER: &str = "__ordvec_raw_number_marker__"; manifest.extensions.insert( "canonical_number".to_string(), - serde_json::from_str(number).map_err(ManifestError::Json)?, + Value::String(MARKER.to_string()), ); - write_manifest_file(&manifest, &manifest_path)?; - fs::read(manifest_path).map_err(ManifestError::Io) + let encoded = serde_json::to_vec(&manifest).map_err(ManifestError::Json)?; + let quoted_marker = format!("\"{MARKER}\""); + let marker_offset = encoded + .windows(quoted_marker.len()) + .position(|window| window == quoted_marker.as_bytes()) + .expect("unique raw-number marker is present in the serialized fixture"); + let mut raw = Vec::with_capacity(encoded.len() - quoted_marker.len() + number.len()); + raw.extend_from_slice(&encoded[..marker_offset]); + raw.extend_from_slice(number.as_bytes()); + raw.extend_from_slice(&encoded[marker_offset + quoted_marker.len()..]); + Ok((manifest_path, raw)) } fn distortion_profile_with_zero( @@ -268,33 +288,6 @@ fn equivalent_json_number_spellings_have_one_canonical_manifest_representation() &["9007199254740992", "9007199254740992.0"], "9007199254740992", ), - ( - "two to 53 plus one", - &[ - "9007199254740993", - "9007199254740993.0", - "9007199254740993e0", - ], - "9007199254740993", - ), - ( - "i64 min", - &[ - "-9223372036854775808", - "-9223372036854775808.0", - "-9.223372036854775808e18", - ], - "-9223372036854775808", - ), - ( - "u64 max", - &[ - "18446744073709551615", - "18446744073709551615.0", - "1.8446744073709551615e19", - ], - "18446744073709551615", - ), ( "out of integer range canonical f64", &["100000000000000000000", "100000000000000000000.0", "1e20"], @@ -384,9 +377,9 @@ fn typed_signed_zero_fields_have_one_canonical_manifest_representation() { #[test] fn adjacent_large_json_integers_keep_distinct_manifest_addresses() { for spellings in [ - ["9007199254740991.0", "9007199254740992.0"], - ["9007199254740992.0", "9007199254740993.0"], - ["18446744073709551614.0", "18446744073709551615.0"], + ["9007199254740991", "9007199254740992"], + ["9007199254740992", "9007199254740993"], + ["18446744073709551614", "18446744073709551615"], ] { let temp_a = tempfile::tempdir().unwrap(); let temp_b = tempfile::tempdir().unwrap(); @@ -398,8 +391,6 @@ fn adjacent_large_json_integers_keep_distinct_manifest_addresses() { #[test] fn arbitrary_precision_numbers_that_would_alias_are_rejected() { - let temp = tempfile::tempdir().unwrap(); - let index = write_index(temp.path()); for (case, number, expected_error) in [ ( "two_to_64_integer", @@ -429,23 +420,78 @@ fn arbitrary_precision_numbers_that_would_alias_are_rejected() { ("f64_overflow", "1e400", "cannot be represented canonically"), ("f64_underflow", "1e-400", "outside the exact canonical"), ] { - let manifest_path = temp.path().join(format!("{case}.json")); - let mut manifest = create_manifest_for_index( - &index, - CreateRowIdentity::RowIdIdentity, - "test-embedding", - &manifest_path, - ) - .unwrap(); - manifest - .extensions - .insert("precise".to_string(), serde_json::from_str(number).unwrap()); - let err = write_manifest_file(&manifest, &manifest_path).unwrap_err(); + let temp = tempfile::tempdir().unwrap(); + let manifest_path = temp.path().join("manifest.json"); + let err = manifest_bytes_with_extension_number(temp.path(), number).unwrap_err(); assert!(err.to_string().contains(expected_error), "{number}: {err}"); - assert!(!manifest_path.exists()); + assert!(!manifest_path.exists(), "{case}"); } } +#[test] +fn lossy_number_tokens_fail_before_current_file_or_attestation_parsing() { + let temp = tempfile::tempdir().unwrap(); + let (manifest_path, raw) = + raw_manifest_bytes_with_extension_number(temp.path(), "9007199254740993.0").unwrap(); + let parse_error = parse_current_manifest_bytes(&raw).unwrap_err(); + assert!(parse_error + .to_string() + .contains("outside the exact canonical")); + + fs::write(&manifest_path, &raw).unwrap(); + let load_error = load_manifest_file(&manifest_path).unwrap_err(); + assert!(load_error + .to_string() + .contains("outside the exact canonical")); + + let index = write_index(temp.path()); + let mut manifest = create_manifest_for_index( + &index, + CreateRowIdentity::RowIdIdentity, + "test-embedding", + &manifest_path, + ) + .unwrap(); + const MARKER: &str = "__ordvec_attestation_number_marker__"; + manifest + .attestations + .push(json!({"predicate": {"measurement": MARKER}})); + let encoded = serde_json::to_vec(&manifest).unwrap(); + let quoted_marker = format!("\"{MARKER}\""); + let marker_offset = encoded + .windows(quoted_marker.len()) + .position(|window| window == quoted_marker.as_bytes()) + .unwrap(); + let mut attestation_raw = Vec::new(); + attestation_raw.extend_from_slice(&encoded[..marker_offset]); + attestation_raw.extend_from_slice(b"0.100000000000000005"); + attestation_raw.extend_from_slice(&encoded[marker_offset + quoted_marker.len()..]); + assert!(parse_current_manifest_bytes(&attestation_raw).is_err()); +} + +#[test] +fn lossless_number_scan_ignores_numeric_text_and_escapes_inside_strings() { + let temp = tempfile::tempdir().unwrap(); + let index = write_index(temp.path()); + let manifest_path = temp.path().join("manifest.json"); + let mut manifest = create_manifest_for_index( + &index, + CreateRowIdentity::RowIdIdentity, + "test-embedding", + &manifest_path, + ) + .unwrap(); + manifest.extensions.insert( + "number_text".to_string(), + Value::String( + "9007199254740993.0 and \\\"18446744073709551617\\\" after \\\\ escape".to_string(), + ), + ); + let bytes = serde_json::to_vec(&manifest).unwrap(); + let parsed = parse_current_manifest_bytes(&bytes).unwrap(); + assert_eq!(parsed.extensions, manifest.extensions); +} + #[test] fn manifest_write_atomically_replaces_the_destination_inode() { let temp = tempfile::tempdir().unwrap(); diff --git a/ordvec-manifest/tests/verified_decode.rs b/ordvec-manifest/tests/verified_decode.rs new file mode 100644 index 00000000..80db87b7 --- /dev/null +++ b/ordvec-manifest/tests/verified_decode.rs @@ -0,0 +1,414 @@ +use ordvec::RankQuant; +#[cfg(any(unix, windows))] +use ordvec_manifest::VerifiedArtifactTypeRejection; +use ordvec_manifest::{ + create_manifest_for_index_with_options, parse_current_manifest_bytes, + read_manifest_bytes_bounded, verify_for_load, write_manifest_file, AuxiliaryArtifactState, + CreateAuxiliaryArtifact, CreateManifestOptions, CreateRowIdentity, ManifestIndexKind, + VerifiedArtifactChange, VerifiedArtifactUseError, VerifiedLoadPlan, VerifyOptions, +}; +use std::fs::{self, OpenOptions}; +use std::io::{self, Write}; +use std::path::PathBuf; +use tempfile::TempDir; + +struct Fixture { + _dir: TempDir, + manifest_path: PathBuf, + index_path: PathBuf, + auxiliary_path: PathBuf, + plan: VerifiedLoadPlan, +} + +fn fixture(optional_auxiliary_absent: bool) -> Fixture { + let dir = tempfile::tempdir().unwrap(); + let index_path = dir.path().join("index.ovrq"); + let auxiliary_path = dir.path().join("ids.bin"); + let manifest_path = dir.path().join("index.manifest.json"); + + let mut index = RankQuant::new(8, 2); + index.add(&[8.0, 3.0, 6.0, 1.0, 7.0, 2.0, 5.0, 4.0]); + index.write(&index_path).unwrap(); + fs::write(&auxiliary_path, b"final auxiliary representation").unwrap(); + + let manifest = create_manifest_for_index_with_options( + &index_path, + CreateRowIdentity::RowIdIdentity, + "fixture-model", + &manifest_path, + CreateManifestOptions { + auxiliary_artifacts: vec![CreateAuxiliaryArtifact { + name: "ids".to_string(), + path: auxiliary_path.clone(), + required: false, + }], + ..CreateManifestOptions::default() + }, + ) + .unwrap(); + write_manifest_file(&manifest, &manifest_path).unwrap(); + if optional_auxiliary_absent { + fs::remove_file(&auxiliary_path).unwrap(); + } + let plan = verify_for_load(&manifest_path, VerifyOptions::default()).unwrap(); + + Fixture { + _dir: dir, + manifest_path, + index_path, + auxiliary_path, + plan, + } +} + +#[test] +fn public_manifest_byte_reader_and_current_parser_share_the_strict_loader_path() { + let fixture = fixture(false); + let bytes = read_manifest_bytes_bounded(&fixture.manifest_path, 1024 * 1024).unwrap(); + let parsed = parse_current_manifest_bytes(&bytes).unwrap(); + assert_eq!(parsed.schema_version, ordvec_manifest::SCHEMA_VERSION); + + let err = + read_manifest_bytes_bounded(&fixture.manifest_path, bytes.len() as u64 - 1).unwrap_err(); + assert_eq!(err.code(), Some("manifest_file_too_large")); +} + +#[test] +fn primary_and_auxiliary_decode_from_plan_verified_forward_only_readers() { + let fixture = fixture(false); + let decoded = fixture + .plan + .decode_primary_with(ManifestIndexKind::RankQuant, |reader, encoded_len| { + RankQuant::read_from_sized(reader, encoded_len) + }) + .unwrap(); + assert_eq!(decoded.len(), 1); + assert_eq!( + fixture.plan.primary_identity().expected_size_bytes, + fs::metadata(&fixture.index_path).unwrap().len() + ); + + let auxiliary = fixture.plan.require_auxiliary("ids").unwrap(); + assert_eq!(auxiliary, fixture.auxiliary_path); + let decoded = fixture + .plan + .auxiliary_by_name("ids") + .unwrap() + .decode_verified_with(|reader, _| { + let mut bytes = Vec::new(); + reader.read_to_end(&mut bytes)?; + Ok::<_, io::Error>(bytes) + }) + .unwrap(); + assert_eq!(decoded, b"final auxiliary representation"); +} + +#[test] +fn primary_kind_mismatch_is_typed_and_does_not_open_the_file() { + let fixture = fixture(false); + fs::remove_file(&fixture.index_path).unwrap(); + let err = fixture + .plan + .decode_primary_with(ManifestIndexKind::SignBitmap, |_, _| Ok::<_, io::Error>(())) + .unwrap_err(); + assert!(matches!( + err, + VerifiedArtifactUseError::KindMismatch { + expected: ManifestIndexKind::SignBitmap, + observed: ManifestIndexKind::RankQuant, + .. + } + )); +} + +#[test] +fn intact_early_parser_failure_wins_after_the_remainder_is_drained() { + let fixture = fixture(false); + let err = fixture + .plan + .decode_primary_with(ManifestIndexKind::RankQuant, |reader, _| { + let mut first = [0u8; 1]; + reader.read_exact(&mut first)?; + Err::<(), _>(io::Error::new( + io::ErrorKind::InvalidData, + "deliberate early parser failure", + )) + }) + .unwrap_err(); + match err { + VerifiedArtifactUseError::Decoder { source, .. } => { + assert_eq!(source.kind(), io::ErrorKind::InvalidData); + assert!(source + .to_string() + .contains("deliberate early parser failure")); + } + other => panic!("expected decoder error, got {other:?}"), + } +} + +#[test] +fn stale_digest_wins_over_an_early_parser_failure() { + let fixture = fixture(false); + let mut bytes = fs::read(&fixture.index_path).unwrap(); + *bytes.last_mut().unwrap() ^= 0x01; + fs::write(&fixture.index_path, bytes).unwrap(); + + let err = fixture + .plan + .decode_primary_with(ManifestIndexKind::RankQuant, |reader, _| { + let mut first = [0u8; 1]; + reader.read_exact(&mut first)?; + Err::<(), _>(io::Error::new( + io::ErrorKind::InvalidData, + "deliberate early parser failure", + )) + }) + .unwrap_err(); + assert!(matches!( + err, + VerifiedArtifactUseError::Stale { ref changes, .. } + if changes.iter().any(|change| matches!(change, VerifiedArtifactChange::Digest { .. })) + )); +} + +#[test] +fn initial_size_change_is_stale_before_decoder_invocation() { + let fixture = fixture(false); + OpenOptions::new() + .append(true) + .open(&fixture.index_path) + .unwrap() + .write_all(b"growth") + .unwrap(); + let decoder_called = std::cell::Cell::new(false); + let err = fixture + .plan + .decode_primary_with(ManifestIndexKind::RankQuant, |_, _| { + decoder_called.set(true); + Ok::<_, io::Error>(()) + }) + .unwrap_err(); + assert!(!decoder_called.get()); + assert!(matches!( + err, + VerifiedArtifactUseError::Stale { ref changes, .. } + if matches!(changes.as_slice(), [VerifiedArtifactChange::InitialSize { .. }]) + )); +} + +#[test] +fn initial_truncation_is_stale_before_decoder_invocation() { + let fixture = fixture(false); + let mut bytes = fs::read(&fixture.index_path).unwrap(); + bytes.pop().unwrap(); + fs::write(&fixture.index_path, bytes).unwrap(); + let decoder_called = std::cell::Cell::new(false); + let err = fixture + .plan + .decode_primary_with(ManifestIndexKind::RankQuant, |_, _| { + decoder_called.set(true); + Ok::<_, io::Error>(()) + }) + .unwrap_err(); + assert!(!decoder_called.get()); + assert!(matches!( + err, + VerifiedArtifactUseError::Stale { ref changes, .. } + if matches!(changes.as_slice(), [VerifiedArtifactChange::InitialSize { expected, observed }] if observed + 1 == *expected) + )); +} + +#[test] +fn missing_artifact_is_a_typed_open_access_failure() { + let fixture = fixture(false); + fs::remove_file(&fixture.index_path).unwrap(); + let err = fixture + .plan + .decode_primary_with(ManifestIndexKind::RankQuant, |_, _| Ok::<_, io::Error>(())) + .unwrap_err(); + assert!(matches!( + err, + VerifiedArtifactUseError::Access { + stage: ordvec_manifest::VerifiedArtifactAccessStage::Open, + .. + } + )); +} + +#[test] +fn final_descriptor_growth_is_stale_without_reading_past_the_plan_boundary() { + let fixture = fixture(false); + let path = fixture.index_path.clone(); + let err = fixture + .plan + .decode_primary_with(ManifestIndexKind::RankQuant, move |reader, _| { + let mut first = [0u8; 1]; + reader.read_exact(&mut first)?; + OpenOptions::new() + .append(true) + .open(path)? + .write_all(b"growth")?; + Err::<(), _>(io::Error::new( + io::ErrorKind::InvalidData, + "deliberate early parser failure", + )) + }) + .unwrap_err(); + assert!(matches!( + err, + VerifiedArtifactUseError::Stale { ref changes, .. } + if changes.iter().any(|change| matches!(change, VerifiedArtifactChange::FinalSize { .. })) + )); +} + +#[test] +fn successful_decoder_must_consume_the_entire_declared_artifact() { + let fixture = fixture(false); + let err = fixture + .plan + .decode_primary_with(ManifestIndexKind::RankQuant, |reader, _| { + let mut first = [0u8; 1]; + reader.read_exact(&mut first)?; + Ok::<_, io::Error>(first[0]) + }) + .unwrap_err(); + assert!(matches!( + err, + VerifiedArtifactUseError::IncompleteConsumption { + consumed: 1, + expected, + .. + } if expected > 1 + )); +} + +#[test] +fn optional_absent_auxiliary_is_a_typed_state() { + let fixture = fixture(true); + let auxiliary = fixture.plan.auxiliary_by_name("ids").unwrap(); + assert_eq!(auxiliary.state(), AuxiliaryArtifactState::OptionalAbsent); + let err = auxiliary + .decode_verified_with(|_, _| Ok::<_, io::Error>(())) + .unwrap_err(); + assert!(matches!( + err, + VerifiedArtifactUseError::OptionalAbsent { ref name } if name == "ids" + )); +} + +#[cfg(unix)] +#[test] +fn final_symlink_is_rejected_by_manifest_and_artifact_openers() { + use std::os::unix::fs::symlink; + + let fixture = fixture(false); + let saved_index = fixture.index_path.with_extension("saved"); + fs::rename(&fixture.index_path, &saved_index).unwrap(); + symlink(&saved_index, &fixture.index_path).unwrap(); + let err = fixture + .plan + .decode_primary_with(ManifestIndexKind::RankQuant, |_, _| Ok::<_, io::Error>(())) + .unwrap_err(); + assert!(matches!( + err, + VerifiedArtifactUseError::TypeRejected { + rejection: VerifiedArtifactTypeRejection::FinalSymlinkOrReparsePoint, + .. + } + )); + + let saved_manifest = fixture.manifest_path.with_extension("saved"); + fs::rename(&fixture.manifest_path, &saved_manifest).unwrap(); + symlink(&saved_manifest, &fixture.manifest_path).unwrap(); + let err = read_manifest_bytes_bounded(&fixture.manifest_path, 1024 * 1024).unwrap_err(); + assert!(err.to_string().contains("symlink or reparse point")); +} + +#[cfg(windows)] +#[test] +fn final_junction_reparse_point_is_rejected() { + let fixture = fixture(false); + fs::remove_file(&fixture.index_path).unwrap(); + let junction_target = fixture.index_path.with_extension("junction-target"); + fs::create_dir(&junction_target).unwrap(); + let status = std::process::Command::new("cmd.exe") + .args(["/D", "/C", "mklink", "/J"]) + .arg(&fixture.index_path) + .arg(&junction_target) + .status() + .expect("cmd.exe must be available on supported Windows test hosts"); + assert!(status.success(), "failed to create test junction: {status}"); + + let err = fixture + .plan + .decode_primary_with(ManifestIndexKind::RankQuant, |_, _| Ok::<_, io::Error>(())) + .unwrap_err(); + assert!(matches!( + err, + VerifiedArtifactUseError::TypeRejected { + rejection: VerifiedArtifactTypeRejection::FinalSymlinkOrReparsePoint, + .. + } + )); +} + +#[cfg(unix)] +#[test] +fn fifo_decode_child() { + if std::env::var_os("ORDVEC_MANIFEST_FIFO_CHILD").is_none() { + return; + } + use std::ffi::CString; + use std::os::unix::ffi::OsStrExt; + + let fixture = fixture(false); + fs::remove_file(&fixture.index_path).unwrap(); + let path = CString::new(fixture.index_path.as_os_str().as_bytes()).unwrap(); + // SAFETY: `path` is a NUL-terminated pathname and the mode is valid. + assert_eq!(unsafe { libc::mkfifo(path.as_ptr(), 0o600) }, 0); + let err = fixture + .plan + .decode_primary_with(ManifestIndexKind::RankQuant, |_, _| Ok::<_, io::Error>(())) + .unwrap_err(); + assert!(matches!( + err, + VerifiedArtifactUseError::TypeRejected { + rejection: VerifiedArtifactTypeRejection::NonRegularFile, + .. + } + )); +} + +#[cfg(unix)] +#[test] +fn fifo_rejection_subprocess_cannot_hang() { + use std::process::{Command, Stdio}; + use std::thread; + use std::time::{Duration, Instant}; + + let mut child = Command::new(std::env::current_exe().unwrap()) + .args(["--exact", "fifo_decode_child", "--nocapture"]) + .env("ORDVEC_MANIFEST_FIFO_CHILD", "1") + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + let deadline = Instant::now() + Duration::from_secs(5); + loop { + if let Some(status) = child.try_wait().unwrap() { + assert!(status.success(), "FIFO child failed with {status}"); + break; + } + if Instant::now() >= deadline { + child.kill().ok(); + let output = child.wait_with_output().unwrap(); + panic!( + "FIFO child hung; stdout={} stderr={}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + } + thread::sleep(Duration::from_millis(20)); + } +} diff --git a/src/quant.rs b/src/quant.rs index 321f748b..b0425aea 100644 --- a/src/quant.rs +++ b/src/quant.rs @@ -943,6 +943,17 @@ impl RankQuant { Self::from_persisted_parts(bits, dim, n_vectors, packed) } + /// Load one exactly-sized `.ovrq`/legacy `.tvrq` record from a forward-only reader. + /// + /// `encoded_len` is the exact number of encoded bytes remaining at the + /// reader's current position. The decoder never seeks and rejects both + /// truncation and bytes beyond the declared record. + pub fn read_from_sized(reader: R, encoded_len: u64) -> std::io::Result { + let (bits, dim, n_vectors, packed) = + crate::rank_io::load_rankquant_from_sized(reader, encoded_len)?; + Self::from_persisted_parts(bits, dim, n_vectors, packed) + } + /// Load a `.ovrq`/legacy `.tvrq` index from an in-memory byte slice. pub fn load_from_bytes(bytes: &[u8]) -> std::io::Result { Self::read_from(std::io::Cursor::new(bytes)) diff --git a/src/rank_io.rs b/src/rank_io.rs index d0b29572..ce17fabd 100644 --- a/src/rank_io.rs +++ b/src/rank_io.rs @@ -157,13 +157,14 @@ fn try_alloc_zeroed(n: usize) -> io::Result> { /// Read `n` little-endian `W`-byte elements directly into a fallibly /// pre-reserved Vec, so an oversized/under-memory load returns an -/// io::Error instead of aborting (and avoids the 2x byte-buffer + typed-Vec peak). +/// io::Error instead of aborting (and avoids the 2x artifact-buffer + typed-Vec peak). /// /// Building the typed `Vec` via `bytes.chunks_exact(W).map(..).collect()` /// uses an infallible allocation — an OOM there is a `SIGABRT`, not a /// recoverable error — and holds both the byte buffer and the typed `Vec` -/// live at once (2x peak). Reserving fallibly and reading element-by-element -/// removes both problems. The size guards ([`check_payload_bytes`], +/// live at once (2x peak). Reserving fallibly and decoding through one fixed +/// 64 KiB stack chunk removes both problems without degrading an unbuffered +/// forward-only reader into one kernel read per element. The size guards ([`check_payload_bytes`], /// [`check_payload_matches_file`]) run *before* this call; `read_le_vec` /// reserves the same element count those guards validated. fn read_le_vec( @@ -171,13 +172,28 @@ fn read_le_vec( n: usize, parse: impl Fn([u8; W]) -> T, ) -> io::Result> { + const READ_CHUNK_BYTES: usize = 64 * 1024; + if W == 0 || W > READ_CHUNK_BYTES { + return Err(invalid("unsupported persisted element width")); + } let mut v: Vec = Vec::new(); v.try_reserve_exact(n) .map_err(|_| invalid("payload allocation too large"))?; - let mut buf = [0u8; W]; - for _ in 0..n { - r.read_exact(&mut buf)?; - v.push(parse(buf)); + let elements_per_chunk = READ_CHUNK_BYTES / W; + let mut scratch = [0u8; READ_CHUNK_BYTES]; + let mut remaining = n; + while remaining != 0 { + let count = remaining.min(elements_per_chunk); + let byte_count = count + .checked_mul(W) + .ok_or_else(|| invalid("persisted read chunk size overflow"))?; + r.read_exact(&mut scratch[..byte_count])?; + for bytes in scratch[..byte_count].chunks_exact(W) { + let mut element = [0u8; W]; + element.copy_from_slice(bytes); + v.push(parse(element)); + } + remaining -= count; } Ok(v) } @@ -229,6 +245,43 @@ fn stream_len_from_current(reader: &mut R) -> io::Result { Ok(end) } +fn stream_remaining_len(reader: &mut R) -> io::Result { + let start = reader.stream_position()?; + let end = reader.seek(SeekFrom::End(0))?; + reader.seek(SeekFrom::Start(start))?; + end.checked_sub(start) + .ok_or_else(|| invalid("stream end precedes current position")) +} + +fn check_payload_matches_sized_input( + label: &str, + encoded_len: u64, + header_bytes: u64, + payload_bytes: usize, +) -> io::Result<()> { + let payload_bytes = u64::try_from(payload_bytes) + .map_err(|_| invalid(format!("{label} payload size does not fit u64")))?; + let expected = header_bytes + .checked_add(payload_bytes) + .ok_or_else(|| invalid(format!("{label} encoded size overflows u64")))?; + if expected > encoded_len { + return Err(invalid(format!( + "{label} payload truncated: header declares {payload_bytes} B but input has {} B remaining after the header", + encoded_len.saturating_sub(header_bytes) + ))); + } + if expected < encoded_len { + return Err(invalid(format!( + "{label} payload has trailing bytes: header declares {payload_bytes} B but input has {} B remaining after the header", + encoded_len.saturating_sub(header_bytes) + ))); + } + Ok(()) +} + +const RANK_QUANT_HEADER_BYTES: u64 = 4 + 1 + 1 + 4 + 4; +const SIGN_BITMAP_HEADER_BYTES: u64 = 4 + 1 + 4 + 4; + fn check_dim(dim: usize) -> io::Result<()> { if !(2..=MAX_DIM).contains(&dim) { return Err(invalid(format!("dim {dim} out of range [2, {MAX_DIM}]"))); @@ -699,22 +752,31 @@ pub(crate) fn load_rankquant(path: impl AsRef) -> io::Result<(u8, usize, u // the trailing-byte check. Both are wrong on a metadata race (NFS/procfs). let file_len = file.metadata()?.len(); let mut f = BufReader::new(file); - load_rankquant_from_stream(&mut f, file_len) + load_rankquant_from_sized(&mut f, file_len) } pub(crate) fn load_rankquant_from( reader: R, ) -> io::Result<(u8, usize, usize, Vec)> { let mut reader = reader; - let file_len = stream_len_from_current(&mut reader)?; - load_rankquant_from_stream(&mut reader, file_len) + let encoded_len = stream_remaining_len(&mut reader)?; + load_rankquant_from_sized(&mut reader, encoded_len) } -fn load_rankquant_from_stream( - mut f: &mut R, - file_len: u64, +pub(crate) fn load_rankquant_from_sized( + f: R, + encoded_len: u64, ) -> io::Result<(u8, usize, usize, Vec)> { - let magic = read_magic(f, "OVRQ")?; + if encoded_len < RANK_QUANT_HEADER_BYTES { + return Err(invalid(format!( + "OVRQ header truncated: input declares {encoded_len} B, header requires {RANK_QUANT_HEADER_BYTES} B" + ))); + } + // Cap the reader from the first byte, not only for the payload. A caller + // may be decoding one record from a concatenated stream, so even a forged + // undersized length must never consume bytes belonging to the next record. + let mut f = f.take(encoded_len); + let magic = read_magic(&mut f, "OVRQ")?; if &magic != OVRQ_MAGIC && &magic != TVRQ_MAGIC { return Err(invalid("not an OVRQ/TVRQ (RankQuant) file: wrong magic")); } @@ -751,7 +813,7 @@ fn load_rankquant_from_stream( check_n_vectors(n_vectors)?; let payload_bytes = rankquant_payload_bytes(dim, n_vectors, bits)?; check_payload_bytes(payload_bytes)?; - check_payload_matches_file(&mut f, "OVRQ", file_len, payload_bytes)?; + check_payload_matches_sized_input("OVRQ", encoded_len, RANK_QUANT_HEADER_BYTES, payload_bytes)?; let mut packed = try_alloc_zeroed(payload_bytes)?; f.read_exact(&mut packed)?; // Constant-composition invariant: every document must place exactly @@ -1028,22 +1090,30 @@ pub(crate) fn load_sign_bitmap(path: impl AsRef) -> io::Result<(usize, usi // the trailing-byte check. Both are wrong on a metadata race (NFS/procfs). let file_len = file.metadata()?.len(); let mut f = BufReader::new(file); - load_sign_bitmap_from_stream(&mut f, file_len) + load_sign_bitmap_from_sized(&mut f, file_len) } pub(crate) fn load_sign_bitmap_from( reader: R, ) -> io::Result<(usize, usize, Vec)> { let mut reader = reader; - let file_len = stream_len_from_current(&mut reader)?; - load_sign_bitmap_from_stream(&mut reader, file_len) + let encoded_len = stream_remaining_len(&mut reader)?; + load_sign_bitmap_from_sized(&mut reader, encoded_len) } -fn load_sign_bitmap_from_stream( - mut f: &mut R, - file_len: u64, +pub(crate) fn load_sign_bitmap_from_sized( + f: R, + encoded_len: u64, ) -> io::Result<(usize, usize, Vec)> { - let magic = read_magic(f, "OVSB")?; + if encoded_len < SIGN_BITMAP_HEADER_BYTES { + return Err(invalid(format!( + "OVSB header truncated: input declares {encoded_len} B, header requires {SIGN_BITMAP_HEADER_BYTES} B" + ))); + } + // See the RankQuant counterpart: the declared boundary covers the fixed + // header as well as the payload. + let mut f = f.take(encoded_len); + let magic = read_magic(&mut f, "OVSB")?; if &magic != OVSB_MAGIC && &magic != TVSB_MAGIC { return Err(invalid("not an OVSB/TVSB (SignBitmap) file: wrong magic")); } @@ -1054,7 +1124,12 @@ fn load_sign_bitmap_from_stream( check_n_vectors(n_vectors)?; let payload_bytes = bitmap_payload_bytes(dim, n_vectors, "OVSB")?; check_payload_bytes(payload_bytes)?; - check_payload_matches_file(&mut f, "OVSB", file_len, payload_bytes)?; + check_payload_matches_sized_input( + "OVSB", + encoded_len, + SIGN_BITMAP_HEADER_BYTES, + payload_bytes, + )?; // `payload_bytes == n_vectors * qpv * 8`, so the u64 element count is // `payload_bytes / 8`. Read directly into a fallibly reserved Vec // rather than allocating a byte buffer and `.collect()`-ing it. @@ -1873,6 +1948,18 @@ mod tests { } } + #[test] + fn fallible_payload_reservations_report_capacity_overflow_without_oom() { + let byte_err = super::try_alloc_zeroed(usize::MAX).unwrap_err(); + assert_eq!(byte_err.kind(), std::io::ErrorKind::InvalidData); + assert!(byte_err.to_string().contains("allocation too large")); + + let typed_err = + super::read_le_vec(&mut std::io::empty(), usize::MAX, u64::from_le_bytes).unwrap_err(); + assert_eq!(typed_err.kind(), std::io::ErrorKind::InvalidData); + assert!(typed_err.to_string().contains("allocation too large")); + } + // OVFS (FastScan) write path: valid round-trip, and fail-loud (io::Error, not // a panic) on invalid `dim`/`n_vectors`/payload — the now-public persistence // API must never abort the caller or silently truncate the header. diff --git a/src/sign_bitmap.rs b/src/sign_bitmap.rs index 649443d0..02dd5c60 100644 --- a/src/sign_bitmap.rs +++ b/src/sign_bitmap.rs @@ -622,6 +622,17 @@ impl SignBitmap { Self::from_persisted_parts(dim, n_vectors, bitmaps) } + /// Load one exactly-sized `.ovsb`/legacy `.tvsb` record from a forward-only reader. + /// + /// `encoded_len` is the exact number of encoded bytes remaining at the + /// reader's current position. The decoder never seeks and rejects both + /// truncation and bytes beyond the declared record. + pub fn read_from_sized(reader: R, encoded_len: u64) -> std::io::Result { + let (dim, n_vectors, bitmaps) = + crate::rank_io::load_sign_bitmap_from_sized(reader, encoded_len)?; + Self::from_persisted_parts(dim, n_vectors, bitmaps) + } + /// Load a `.ovsb`/legacy `.tvsb` index from an in-memory byte slice. pub fn load_from_bytes(bytes: &[u8]) -> std::io::Result { Self::read_from(std::io::Cursor::new(bytes)) diff --git a/tests/index/loader_validation.rs b/tests/index/loader_validation.rs index 2c45f9a6..695d0f8f 100644 --- a/tests/index/loader_validation.rs +++ b/tests/index/loader_validation.rs @@ -9,7 +9,9 @@ //! Each case pairs a positive control (a freshly-written valid index still //! round-trips) with a corrupted-but-well-shaped negative case. -use std::io::{Cursor, Write}; +use std::cell::Cell; +use std::io::{self, Cursor, Read, Write}; +use std::rc::Rc; use ordvec::{Bitmap, Rank, RankQuant, SignBitmap}; @@ -58,6 +60,61 @@ fn append_trailer(mut bytes: Vec) -> Cursor> { Cursor::new(bytes) } +struct FragmentedInterruptingSpy { + bytes: Vec, + record_len: usize, + position: usize, + interrupt_next: bool, + overread_attempts: Rc>, +} + +impl FragmentedInterruptingSpy { + fn new(bytes: Vec, record_len: usize, overread_attempts: Rc>) -> Self { + Self { + bytes, + record_len, + position: 0, + interrupt_next: true, + overread_attempts, + } + } +} + +impl Read for FragmentedInterruptingSpy { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + if self.interrupt_next { + self.interrupt_next = false; + return Err(io::Error::new(io::ErrorKind::Interrupted, "spy interrupt")); + } + self.interrupt_next = true; + if self.position == self.bytes.len() || buf.is_empty() { + return Ok(0); + } + let read = buf.len().min(3).min(self.bytes.len() - self.position); + if self.position.saturating_add(read) > self.record_len { + self.overread_attempts + .set(self.overread_attempts.get().saturating_add(1)); + } + buf[..read].copy_from_slice(&self.bytes[self.position..self.position + read]); + self.position += read; + Ok(read) + } +} + +struct CountingReader { + inner: Cursor>, + calls: Rc>, +} + +impl Read for CountingReader { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + if !buf.is_empty() { + self.calls.set(self.calls.get().saturating_add(1)); + } + self.inner.read(buf) + } +} + #[test] fn public_stream_persistence_roundtrips_core_formats() { let corpus = make_corpus(90_001); @@ -167,6 +224,140 @@ fn public_stream_persistence_roundtrips_core_formats() { } } +#[test] +fn sized_forward_only_readers_handle_fragmentation_interrupts_and_never_overread() { + let corpus = make_corpus(90_051); + + let mut rank_quant = RankQuant::new(D, 2); + rank_quant.add(&corpus); + let mut rank_quant_bytes = Vec::new(); + rank_quant.write_to(&mut rank_quant_bytes).unwrap(); + let rank_quant_len = rank_quant_bytes.len(); + rank_quant_bytes.extend_from_slice(b"unrelated-next-record"); + let rank_quant_overreads = Rc::new(Cell::new(0)); + let decoded = RankQuant::read_from_sized( + FragmentedInterruptingSpy::new( + rank_quant_bytes, + rank_quant_len, + Rc::clone(&rank_quant_overreads), + ), + rank_quant_len as u64, + ) + .unwrap(); + assert_eq!(decoded.len(), rank_quant.len()); + assert_eq!(rank_quant_overreads.get(), 0); + + let mut sign = SignBitmap::new(D); + sign.add(&corpus); + let mut sign_bytes = Vec::new(); + sign.write_to(&mut sign_bytes).unwrap(); + let sign_len = sign_bytes.len(); + sign_bytes.extend_from_slice(b"unrelated-next-record"); + let sign_overreads = Rc::new(Cell::new(0)); + let decoded = SignBitmap::read_from_sized( + FragmentedInterruptingSpy::new(sign_bytes, sign_len, Rc::clone(&sign_overreads)), + sign_len as u64, + ) + .unwrap(); + assert_eq!(decoded.len(), sign.len()); + assert_eq!(sign_overreads.get(), 0); +} + +#[test] +fn sized_forward_only_readers_reject_inexact_lengths_before_payload_allocation() { + let corpus = make_corpus(90_052); + + let mut rank_quant = RankQuant::new(D, 2); + rank_quant.add(&corpus); + let mut bytes = Vec::new(); + rank_quant.write_to(&mut bytes).unwrap(); + assert_load_err_contains( + RankQuant::read_from_sized(Cursor::new(&bytes), bytes.len() as u64 - 1), + "payload truncated", + ); + assert_load_err_contains( + RankQuant::read_from_sized(Cursor::new(&bytes), u64::MAX), + "payload has trailing bytes", + ); + + let mut sign = SignBitmap::new(D); + sign.add(&corpus); + let mut bytes = Vec::new(); + sign.write_to(&mut bytes).unwrap(); + assert_load_err_contains( + SignBitmap::read_from_sized(Cursor::new(&bytes), bytes.len() as u64 - 1), + "payload truncated", + ); + assert_load_err_contains( + SignBitmap::read_from_sized(Cursor::new(&bytes), u64::MAX), + "payload has trailing bytes", + ); +} + +#[test] +fn sized_forward_only_readers_never_cross_an_undersized_header_boundary() { + let corpus = make_corpus(90_053); + + let mut rank_quant = RankQuant::new(D, 2); + rank_quant.add(&corpus); + let mut rank_quant_bytes = Vec::new(); + rank_quant.write_to(&mut rank_quant_bytes).unwrap(); + let rank_quant_declared = 13; + let rank_quant_overreads = Rc::new(Cell::new(0)); + assert_load_err_contains( + RankQuant::read_from_sized( + FragmentedInterruptingSpy::new( + rank_quant_bytes, + rank_quant_declared, + Rc::clone(&rank_quant_overreads), + ), + rank_quant_declared as u64, + ), + "header truncated", + ); + assert_eq!(rank_quant_overreads.get(), 0); + + let mut sign = SignBitmap::new(D); + sign.add(&corpus); + let mut sign_bytes = Vec::new(); + sign.write_to(&mut sign_bytes).unwrap(); + let sign_declared = 12; + let sign_overreads = Rc::new(Cell::new(0)); + assert_load_err_contains( + SignBitmap::read_from_sized( + FragmentedInterruptingSpy::new(sign_bytes, sign_declared, Rc::clone(&sign_overreads)), + sign_declared as u64, + ), + "header truncated", + ); + assert_eq!(sign_overreads.get(), 0); +} + +#[test] +fn sized_sign_reader_batches_typed_payload_reads() { + let corpus = make_corpus(90_054); + let mut sign = SignBitmap::new(D); + for _ in 0..32 { + sign.add(&corpus); + } + let mut bytes = Vec::new(); + sign.write_to(&mut bytes).unwrap(); + assert!(bytes.len() > 64 * 1024); + + let calls = Rc::new(Cell::new(0)); + let reader = CountingReader { + inner: Cursor::new(bytes.clone()), + calls: Rc::clone(&calls), + }; + let decoded = SignBitmap::read_from_sized(reader, bytes.len() as u64).unwrap(); + assert_eq!(decoded.len(), sign.len()); + assert!( + calls.get() < 32, + "fixed-chunk decoding regressed to {} underlying reads", + calls.get() + ); +} + #[test] fn public_readers_do_not_buffer_past_reported_trailing_bytes() { let corpus = make_corpus(90_101); diff --git a/tests/release_publish_invariants.py b/tests/release_publish_invariants.py index e2f266f4..db11832d 100644 --- a/tests/release_publish_invariants.py +++ b/tests/release_publish_invariants.py @@ -729,6 +729,23 @@ def check_registry_metadata_parity() -> None: def check_manifest_cli_defaults() -> None: manifest = load_toml("ordvec-manifest/Cargo.toml") + dependencies = mapping( + manifest.get("dependencies"), "ordvec-manifest/Cargo.toml: dependencies" + ) + serde_json = mapping( + dependencies.get("serde_json"), + "ordvec-manifest/Cargo.toml: dependencies.serde_json", + ) + serde_features = string_sequence( + serde_json.get("features"), + "ordvec-manifest/Cargo.toml: dependencies.serde_json.features", + ) + if serde_features != ["float_roundtrip"]: + fail( + "ordvec-manifest/Cargo.toml: production serde_json features must be " + "exactly ['float_roundtrip']; arbitrary_precision, preserve_order, " + "raw_value, and unbounded_depth are forbidden" + ) features = mapping(manifest.get("features"), "ordvec-manifest/Cargo.toml: features") default_features = string_sequence( features.get("default"), "ordvec-manifest/Cargo.toml: features.default" From 9a23bc46aa6d7111a31cfc848bf02f549c7b4874 Mon Sep 17 00:00:00 2001 From: Nelson Spence Date: Fri, 24 Jul 2026 15:22:03 -0500 Subject: [PATCH 2/5] test(manifest): compare canonical plan paths Signed-off-by: Nelson Spence --- ordvec-manifest/tests/verified_decode.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/ordvec-manifest/tests/verified_decode.rs b/ordvec-manifest/tests/verified_decode.rs index 80db87b7..1840c407 100644 --- a/ordvec-manifest/tests/verified_decode.rs +++ b/ordvec-manifest/tests/verified_decode.rs @@ -89,7 +89,10 @@ fn primary_and_auxiliary_decode_from_plan_verified_forward_only_readers() { ); let auxiliary = fixture.plan.require_auxiliary("ids").unwrap(); - assert_eq!(auxiliary, fixture.auxiliary_path); + assert_eq!( + auxiliary, + fs::canonicalize(&fixture.auxiliary_path).unwrap() + ); let decoded = fixture .plan .auxiliary_by_name("ids") From 6607c98b233a9193018cbd3db5188acf8aecd686 Mon Sep 17 00:00:00 2001 From: Nelson Spence Date: Fri, 24 Jul 2026 15:24:01 -0500 Subject: [PATCH 3/5] fix(io): allocate decode scratch fallibly Signed-off-by: Nelson Spence --- src/rank_io.rs | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/src/rank_io.rs b/src/rank_io.rs index ce17fabd..e01190be 100644 --- a/src/rank_io.rs +++ b/src/rank_io.rs @@ -180,7 +180,14 @@ fn read_le_vec( v.try_reserve_exact(n) .map_err(|_| invalid("payload allocation too large"))?; let elements_per_chunk = READ_CHUNK_BYTES / W; - let mut scratch = [0u8; READ_CHUNK_BYTES]; + // Keep the batching buffer off the caller's stack. Loaders are valid on + // deliberately small-stack worker threads, and allocation failure must + // remain a recoverable I/O error rather than a stack-overflow abort. + let mut scratch = Vec::new(); + scratch + .try_reserve_exact(READ_CHUNK_BYTES) + .map_err(|_| invalid("read scratch allocation too large"))?; + scratch.resize(READ_CHUNK_BYTES, 0); let mut remaining = n; while remaining != 0 { let count = remaining.min(elements_per_chunk); From 32035d1b6ed0c7df71a5dcc06fbc7992fb26ab5b Mon Sep 17 00:00:00 2001 From: Nelson Spence Date: Fri, 24 Jul 2026 15:32:24 -0500 Subject: [PATCH 4/5] fix(manifest): heap-allocate verification scratch Signed-off-by: Nelson Spence --- ordvec-manifest/src/lib.rs | 33 ++++++++++++++++++++++++++------- 1 file changed, 26 insertions(+), 7 deletions(-) diff --git a/ordvec-manifest/src/lib.rs b/ordvec-manifest/src/lib.rs index a1198951..4ffd4fcb 100644 --- a/ordvec-manifest/src/lib.rs +++ b/ordvec-manifest/src/lib.rs @@ -4153,13 +4153,32 @@ fn decode_plan_verified( let decoded = decoder(&mut reader, expected_size); let decoder_consumed = reader.consumed; - let mut scratch = [0u8; VERIFIED_DECODE_DRAIN_BYTES]; - while reader.remaining != 0 { - match reader.read(&mut scratch) { - Ok(0) => break, - Ok(_) => {} - Err(source) if source.kind() == io::ErrorKind::Interrupted => continue, - Err(_) => break, + if reader.remaining != 0 { + // Keep the fixed verification scratch off small caller stacks, and + // preserve the public recoverable-error contract if it cannot be + // allocated. Fully consuming decoders do not need this allocation. + let mut scratch = Vec::new(); + if scratch + .try_reserve_exact(VERIFIED_DECODE_DRAIN_BYTES) + .is_err() + { + return Err(VerifiedArtifactUseError::Access { + identity, + stage: VerifiedArtifactAccessStage::Read, + source: io::Error::new( + io::ErrorKind::OutOfMemory, + "verification read scratch allocation failed", + ), + }); + } + scratch.resize(VERIFIED_DECODE_DRAIN_BYTES, 0); + while reader.remaining != 0 { + match reader.read(&mut scratch) { + Ok(0) => break, + Ok(_) => {} + Err(source) if source.kind() == io::ErrorKind::Interrupted => continue, + Err(_) => break, + } } } let (observed_digest, read_error) = reader.finish(); From 26c3ec983e730dfd9afd8c05e4be42808afdc847 Mon Sep 17 00:00:00 2001 From: Nelson Spence Date: Sat, 25 Jul 2026 10:28:46 -0500 Subject: [PATCH 5/5] fix(io): preserve truncated header errors Signed-off-by: Nelson Spence --- src/rank_io.rs | 10 ---------- tests/index/loader_validation.rs | 17 ++++++++++++----- 2 files changed, 12 insertions(+), 15 deletions(-) diff --git a/src/rank_io.rs b/src/rank_io.rs index e01190be..3b19f4dc 100644 --- a/src/rank_io.rs +++ b/src/rank_io.rs @@ -774,11 +774,6 @@ pub(crate) fn load_rankquant_from_sized( f: R, encoded_len: u64, ) -> io::Result<(u8, usize, usize, Vec)> { - if encoded_len < RANK_QUANT_HEADER_BYTES { - return Err(invalid(format!( - "OVRQ header truncated: input declares {encoded_len} B, header requires {RANK_QUANT_HEADER_BYTES} B" - ))); - } // Cap the reader from the first byte, not only for the payload. A caller // may be decoding one record from a concatenated stream, so even a forged // undersized length must never consume bytes belonging to the next record. @@ -1112,11 +1107,6 @@ pub(crate) fn load_sign_bitmap_from_sized( f: R, encoded_len: u64, ) -> io::Result<(usize, usize, Vec)> { - if encoded_len < SIGN_BITMAP_HEADER_BYTES { - return Err(invalid(format!( - "OVSB header truncated: input declares {encoded_len} B, header requires {SIGN_BITMAP_HEADER_BYTES} B" - ))); - } // See the RankQuant counterpart: the declared boundary covers the fixed // header as well as the payload. let mut f = f.take(encoded_len); diff --git a/tests/index/loader_validation.rs b/tests/index/loader_validation.rs index 695d0f8f..695f15ba 100644 --- a/tests/index/loader_validation.rs +++ b/tests/index/loader_validation.rs @@ -31,7 +31,7 @@ fn tmp(name: &str) -> std::path::PathBuf { )) } -fn assert_load_err_contains(result: std::io::Result, expected: &str) { +fn load_err_contains(result: std::io::Result, expected: &str) -> std::io::Error { let Err(err) = result else { panic!("expected error containing {expected:?}, got Ok(_)"); }; @@ -40,6 +40,11 @@ fn assert_load_err_contains(result: std::io::Result, expected: &str) { text.contains(expected), "expected error containing {expected:?}, got {text:?}" ); + err +} + +fn assert_load_err_contains(result: std::io::Result, expected: &str) { + let _ = load_err_contains(result, expected); } fn set_u32_field(bytes: &mut [u8], offset: usize, value: u32) { @@ -304,7 +309,7 @@ fn sized_forward_only_readers_never_cross_an_undersized_header_boundary() { rank_quant.write_to(&mut rank_quant_bytes).unwrap(); let rank_quant_declared = 13; let rank_quant_overreads = Rc::new(Cell::new(0)); - assert_load_err_contains( + let rank_quant_err = load_err_contains( RankQuant::read_from_sized( FragmentedInterruptingSpy::new( rank_quant_bytes, @@ -313,8 +318,9 @@ fn sized_forward_only_readers_never_cross_an_undersized_header_boundary() { ), rank_quant_declared as u64, ), - "header truncated", + "OVRQ header truncated while reading n_vectors", ); + assert_eq!(rank_quant_err.kind(), std::io::ErrorKind::UnexpectedEof); assert_eq!(rank_quant_overreads.get(), 0); let mut sign = SignBitmap::new(D); @@ -323,13 +329,14 @@ fn sized_forward_only_readers_never_cross_an_undersized_header_boundary() { sign.write_to(&mut sign_bytes).unwrap(); let sign_declared = 12; let sign_overreads = Rc::new(Cell::new(0)); - assert_load_err_contains( + let sign_err = load_err_contains( SignBitmap::read_from_sized( FragmentedInterruptingSpy::new(sign_bytes, sign_declared, Rc::clone(&sign_overreads)), sign_declared as u64, ), - "header truncated", + "OVSB header truncated while reading n_vectors", ); + assert_eq!(sign_err.kind(), std::io::ErrorKind::UnexpectedEof); assert_eq!(sign_overreads.get(), 0); }