diff --git a/CHANGELOG.md b/CHANGELOG.md index 2bdad7c2..3b76807e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -278,6 +278,31 @@ only the control plane, not the segment frames. `main.rs`, through a direct `mimalloc` dependency rather than one inherited from a library. Same allocator, same version, same measured behaviour — the decision is just made in the place that owns the process (see the dependency entry under Changed). +- **Frame versions are a registry, and a frame says what it is** (1.0 plan X-FRAME; absorbs FRE-4, + TAG-2 and ROB-8's fault-class naming). `weft_physical_type::frame_registry` names the frame + layouts — `5` single-block, `6` paged, `7` envelope — on the rule stated there that a number is + never reused or renumbered once it has shipped. The v7 envelope is a header and a trailer around + a frame stored **verbatim**: `WEFTSEG | u16 7 | u16 body_version | extension table | frame | + crc32`, 16 bytes of overhead (a 12-byte header plus the trailer), so a v5 or v6 body keeps every + byte-for-byte expectation a released server already holds while a reader gains somewhere to put + material in front of a body it is not allowed to change. An extension is `{id, critical, + payload}`: a reader that does not know a **critical** one refuses the frame, because that body + was written on the assumption that the extension would be applied, and an optional one is + ignored. The set this build implements (`KNOWN_EXTENSIONS`) is deliberately empty — the envelope + ships with no extensions, and a number invented before there is a feature behind it can never be + taken back. `describe_frame`, `frame_layout_of`, `wrap_frame_envelope`/`unwrap_frame_envelope` + and the layout-aware readers `read_frame_segment`, `read_frame_paged_segment`, `read_frame_point`, + `read_frame_points`, `read_frame_range` and `read_frame_range_physical` read a frame without the + caller knowing how it is framed, and `frame_fault_of` gives the class an operator acts on (plan + C5): `Integrity`, the bytes are damaged, or `Undecodable { newer_format }`, this binary is behind + the file. `SegmentStore` picks its reader from the frame's own header rather than the + `format_version` it recorded in the index row, where any value that was not `6` used to send a + paged frame into the single-block decoder; the decode failures it returns carry + `weftdb::FrameFaultError` in the `anyhow` chain under the `decoding [paged ]segment : …` + sentence they have always produced. Over HTTP that is C5: a frame this build has no reader for is + `503 incompatible_store`, and any other frame that will not decode is `500 frame_undecodable`, + the code X-ERR gave a wire sentence but no producer. **No writer opts into the envelope yet** — + seals keep writing 5 and 6 — so this is the mechanism, not a format change. ### Changed diff --git a/ROADMAP.md b/ROADMAP.md index a7e0b21d..bc079094 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -1159,7 +1159,7 @@ any write, never purge an unproven frame), with its unfinished work on `durabili - [x] → **X-NAME** `aspect_name::encode` and `validate_new` at HTTP declare (absorbs FRE-7a, SEC-0) — *partly on main: `validate` and the declare check (#66), `encode` (#72). #91 is the rest: `weftdb::aspect_name::validate_new` (`weftdb/src/types/aspect_name.rs`) runs `validate`'s chain and then refuses an ASCII uppercase letter with a new `AspectNameReason::UppercaseAsciiLetter` (the enum is `#[non_exhaustive]`, so the variant is not a breaking change). Order is asserted: the capital rule runs **after** `reason()`, so `../Price` is refused for escaping the directory, not for its capital — the more dangerous reason is never hidden. The hole is real because the legacy frame file is `-.weftseg`, which writes the name literally, and NTFS, APFS (by default) and ext4 with `casefold` fold ASCII case: two aspects differing only in case share a file, whichever seal lands second overwrites the other's bytes, and reads serve one series' values for the other's name. `encode` (#72) removes that collision for the frames D-S10 will write; this closes the declaration of a new collision. Applied at **creation only** — `POST /api/v1/storage/aspects` — so `SegmentStore::declare`, every read, seal, maintenance and store-wide sweep keep plain `validate` and a store that already holds `Price` stays readable, writable and compactable (the test asserts both halves: the refusal, then a raw catalog declare of `Price` followed by a 201 write). Two things are deliberately **not** refused: the characters only Windows forbids (`< > : " | ? *`), because `job:rate5m` is blessed on Unix by this module's own docs and is a shape in use, and the cross-platform hazard belongs one layer down where `encode` escapes `:` for write-once frame names (D-S10, the slice the plan names for it); and Unicode NFC/NFD aliases, which need `unicode-normalization` this crate does not carry, while `encode` escapes every non-ASCII byte anyway. A refusal keeps the #90 shape: `400 invalid_aspect_name` with `details.aspect` and `details.reason`.* - [x] → **ROB-2** Storage-engine panic fail-stop: a turso-origin panic aborts — *#92: the write scope was already here — `weftdb::exec::control_plane_write`, and every `IndexTxn` operation runs inside it, which weftdb's own `every_op_runs_inside_the_control_plane_write_scope` asserts — so C12's "hook and knob" was the half that was missing, and it lives in the binary: `weft_server::panic_hook::install()` chains the hook it replaces, and when `in_control_plane_write()` says the panicking task held the scope it names the panic and calls `std::process::abort()`: SIGABRT for a supervisor, not the poison path's exit 70, and never an HTTP response (FRE-6a). Startup calls it immediately after `init_tracing`, because `set_hook` hands back the hook it replaced. It deliberately does not write the poison marker: that region reads before it writes (`txn.plans()`, `SegmentIndex::open`), and the marker survives the panic until an operator deletes a file. Nothing is caught, so ROB-3 stays the only catch layer and read-path panics still answer 500; the legacy `weftdb/src/types/aspect.rs` lane is ROB-5's. Proved by `weft-server/tests/control_plane_panic.rs`, which re-executes the test binary the way `poison_exit.rs` does.* - [x] **PAC-3** Library crate hygiene: no global allocator or wgpu features imposed on library users — *#93: two decisions were being made for every program that links these crates. `turso` 0.8.1 puts a `#[global_allocator]` in its own crate root behind its `mimalloc` feature, so the workspace's `features = ["mimalloc"]` on a **published library** (`weftdb/Cargo.toml`) chose the allocator of every embedder's process; and `splimes`' default features pull `wgpu`, so `wgpu` sat in the normal graph of `weftdb`, `weft-line-protocol`, `weft-reduce` and `weft-orchestration` — 52 packages of Vulkan, Metal and DX12 code for two crates that never call a GPU API. The allocator moved to the place that owns the process: `weft-server/src/main.rs` and `weft-tui/src/main.rs` each declare `static ALLOCATOR: mimalloc::MiMalloc` through a direct `mimalloc` dependency, so the same allocator on the same version still runs everything we benchmark and ship, and `weftdb` now carries no allocator feature at all — the plan's own lane row names this slice at `main.rs` (`docs/release/1.0-plan.md:126`). GPU is opt-in the same way: the workspace asks splimes for `default-features = false, features = ["serde"]`, `weft-server` and `weft-bench` add `gpu` themselves, and `weftdb`'s interpolation benches take it as a dev-dependency. Distinct packages in the normal graph: `weft-line-protocol` and `weft-reduce` 77 → 25, `weft-orchestration` 267 → 231, `weftdb` 266 → 227; `cargo tree -p weftdb -i wgpu -e normal` and `-i mimalloc` now return nothing. Two keeps are deliberate: splimes' `serde` stays on because `weftdb::BatchedMeasurement` holds a `splimes::Resolution` and derives `Serialize`/`Deserialize`, and turso's other default, `fts`, stays off exactly as it already was — an attempt to re-add it pulled `tantivy`, `zstd`, `datasketches` and friends, +435 lines of `Cargo.lock`, for full-text search nothing in this workspace uses. `tracing-subscriber` became a dev-dependency of `weftdb` (only test code installs a subscriber). One consequence for our own numbers: `cargo bench -p weftdb` used to inherit mimalloc and now runs on the system allocator, so those nine criterion benches are not comparable across this commit; the nightly regression track (EVI-5) uses `weft-bench`, which never linked `weftdb` and never had mimalloc, so no baseline moves there.* - - [ ] L-FMT: **X-FRAME** Frame registry, v7 envelope, `read_frame*`/`describe_frame`, `FrameFault` classes (absorbs FRE-4, TAG-2) + - [x] L-FMT: **X-FRAME** Frame registry, v7 envelope, `read_frame*`/`describe_frame`, `FrameFault` classes (absorbs FRE-4, TAG-2) — *#94: the three frame versions are named in one place, `weft_physical_type::frame_registry` — 5 single-block, 6 paged, 7 envelope, `layout_of`, `FrameExtension`, `FrameInfo`, `FrameFault { Integrity, Undecodable { newer_format } }` and `fault_class()`, with `KNOWN_EXTENSIONS` deliberately **empty** because an extension id is a permanent namespace. The envelope (`wrap_frame_envelope` in `weft-physical-type/src/weftseg.rs`) stores the older frame verbatim, magic, version and CRC included, for 16 bytes of overhead, so every byte-for-byte pin in this repo still holds and a legacy reader still answers `UnsupportedVersion { found: 7 }`; `frame_parts` verifies that CRC before it trusts a single header field, and `EnvelopeBodyMismatch` catches a header that disagrees with a body which is already under the checksum. Reading is now routed by **what the file says**: `weftdb/src/types/frame_read.rs` probes the header (`frame_layout_of`) instead of trusting `descriptor.format_version`, and its `FrameFaultError` reproduces the historical `decoding paged segment {path}: {sentence}` byte-for-byte, at all six call sites. Nothing opts into v7 yet — writers keep emitting 5 and 6, because `weftdb/src/types/segment_store.rs:7022` promises an unconfigured deployment bytes that are historical. `error::Code::FrameUndecodable` finally has a producer: a frame from a newer version is 503 `incompatible_store` and everything undecodable-but-current is 500, asserted in `weft-server/tests/error_contract.rs` by corrupting a real `.weftseg` (one byte, and a re-stamped-CRC version bump), which also shows the store surviving a damaged frame and the other aspect still reading. 306 `weft-physical-type` and 400 `weftdb` lib tests pass.* - [ ] Legacy lane: **FRE-14** Legacy data-directory marker, numbered migrations and measurement upgrade test - [ ] **PAC-2** Portable Linux binaries on a glibc 2.28 baseline - [ ] **EVI-5** Nightly benchmark-regression workflow and `weft-bench compare` diff --git a/weft-physical-type/src/frame_registry.rs b/weft-physical-type/src/frame_registry.rs new file mode 100644 index 00000000..dc4b561a --- /dev/null +++ b/weft-physical-type/src/frame_registry.rs @@ -0,0 +1,239 @@ +//! The version registry for `.weftseg` frame layouts, and the fault classes a frame +//! read can fail with. +//! +//! Every `.weftseg` file names its own layout: the seven magic bytes, then a `u16` +//! little-endian format version. That number is the whole of what a reader needs to +//! know how to decode what follows, so it is a shared, permanent name — three tracks +//! of this project allocated frame versions independently and would have shipped +//! different meanings for the same number (1.0 plan X-FRAME). This module is where +//! the names live: [`FRAME_SINGLE_BLOCK`] and [`FRAME_PAGED`] are the two layouts +//! WeftDB has always written, [`FRAME_ENVELOPE`] is the v7 wrapper that lets a frame +//! carry extension blocks in front of a body whose bytes are already frozen. +//! +//! **A number that has shipped is never reused, redefined, or retired**, so a file +//! written by an old server still means the same thing to a new one, and a file +//! written by a new server is refused rather than misread. The list is append-only: +//! the next layout takes 8. +//! +//! # Fault classes +//! +//! A frame read can fail in two ways that the operator should not treat the same way +//! (1.0 plan resolution C5). [`FrameFault::Integrity`] means the bytes disagree with +//! their own checksum, length, or magic: the file is damaged, and the quarantine +//! machinery is allowed to purge it. [`FrameFault::Undecodable`] means the file is +//! internally consistent but this build cannot make sense of it — a version it has +//! never heard of, an extension it cannot honour, a codec it was compiled without. +//! Those files are never damaged, and deleting one destroys data that a different +//! build could read: they go suspect, not corrupt, and `newer_format: true` is the +//! sub-case where a server should also say on `/ready` that the store on disk is a +//! format it cannot open. +//! +//! The quarantine machinery itself is D-S12's (verified by TAG-9 and ROB-22); what +//! lives here is the vocabulary every layer agrees on in the meantime. + +/// The single-block frame layout WeftDB has always written: one block per column, the +/// whole aspect's rows in one file. +pub const FRAME_SINGLE_BLOCK: u16 = 5; +/// The paged frame layout: fixed-size pages, each independently decodable, so a range +/// read touches only the pages it needs. +pub const FRAME_PAGED: u16 = 6; +/// The envelope: this version number's own layout, a header naming the body layout +/// plus an extension table, then a complete frame of that layout. +/// +/// The envelope exists because a body layout, once written by a released server, is +/// frozen: adding a field inside it would silently shift every byte after it. A frame +/// that needs something new puts it in front of the frozen body instead, where an +/// older reader can still refuse the file cleanly rather than decode garbage from a +/// moved offset. +pub const FRAME_ENVELOPE: u16 = 7; + +/// The highest frame version this build can read. Anything above it came from a newer +/// server, which is the difference between "this file is broken" and "this file is +/// ahead of me" — see [`FrameFault::Undecodable`]. +pub const FRAME_VERSION_MAX_KNOWN: u16 = FRAME_ENVELOPE; + +/// Which layout a frame body uses. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum FrameLayout { + /// [`FRAME_SINGLE_BLOCK`]. + SingleBlock, + /// [`FRAME_PAGED`]. + Paged, +} + +/// The layout a body version names, or `None` for a version this build has no reader +/// for. +/// +/// The version in a frame's own header is not necessarily a layout this build knows; +/// the caller decides whether that is an error worth reporting. +#[must_use] +pub const fn layout_of(version: u16) -> Option { + match version { + FRAME_SINGLE_BLOCK => Some(FrameLayout::SingleBlock), + FRAME_PAGED => Some(FrameLayout::Paged), + _ => None, + } +} + +/// An entry in a v7 envelope's extension table. +/// +/// `critical` is the whole of the compatibility contract: an extension the reader has +/// never heard of is skipped when it is informational, and refuses the frame when it +/// is critical, because "critical" is the writer's way of saying *the body below is +/// only correct if you applied me*. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct FrameExtension { + /// The extension's number. Extension numbers are a permanent namespace, on the + /// same rule as frame versions: never reuse one, and never renumber one that has + /// shipped. + pub id: u32, + /// Whether a reader that does not know this extension must refuse the frame. + pub critical: bool, + /// The extension's own bytes. The envelope does not interpret them; the extension + /// defines its own payload format, versioned by its own number. + pub payload: Vec, +} + +/// An extension number this build implements, and what it means. +/// +/// The description is not decoration: it is what the reader prints when a frame asks +/// for something the binary does not have, and an operator cannot decide between +/// "upgrade the server" and "this file is broken" without it. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct FrameExtensionSpec { + /// The extension's number. + pub id: u32, + /// What the extension does, in the sentence an operator reads. + pub name: &'static str, +} + +/// The extension numbers this build implements. +/// +/// Append-only, on exactly the rule stated for frame versions at the top of this +/// module: an extension that has shipped keeps its number, and a new one takes the +/// next free number. It is empty because the envelope ships with no extensions — the +/// point of X-FRAME is the *mechanism* that lets a later slice put a block in front of +/// a frozen body without moving a byte inside it, and a number invented before there +/// is a writer for it is a number wasted. +pub const KNOWN_EXTENSIONS: &[FrameExtensionSpec] = &[]; + +/// Whether this build knows what to do with extension `id`. +/// +/// A frame that names an extension absent from [`KNOWN_EXTENSIONS`] is a frame written +/// by a newer server, not a damaged one: the answer feeds +/// [`FrameFault::Undecodable`], and the file is kept. +#[must_use] +pub fn is_known_extension(id: u32) -> bool { + KNOWN_EXTENSIONS.iter().any(|spec| spec.id == id) +} + +/// What a frame is, without decoding it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct FrameInfo { + /// The version in the frame's own header: one of [`FRAME_SINGLE_BLOCK`], + /// [`FRAME_PAGED`], [`FRAME_ENVELOPE`], or a number this build has no reader for. + pub version: u16, + /// The version of the body bytes themselves — the same as [`Self::version`] for a + /// plain v5/v6 frame, and the wrapped layout's number for an envelope. + pub body_version: u16, + /// The body's layout, or `None` if this build has no reader for it. + pub layout: Option, + /// The envelope's extension table; empty for a plain frame. + pub extensions: Vec, + /// Set when the frame carries a **critical** extension this build does not + /// implement: the body is not safe to serve, because it was written on the + /// assumption that a reader would apply what is missing. + pub unknown_critical: bool, +} + +impl FrameInfo { + /// Whether the frame is safe to decode with this build: the body layout is known + /// and no critical extension is being ignored. + #[must_use] + pub const fn decodable(&self) -> bool { + self.layout.is_some() && !self.unknown_critical + } +} + +/// Why a frame could not be read, in the two classes the plan's C5 resolution defines. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum FrameFault { + /// The bytes disagree with their own magic, checksum or length: the file is + /// damaged. Quarantine and, under the store's own policy, TTL purge are allowed. + Integrity, + /// The file is internally consistent but this build cannot read it. Its contents + /// are not damaged, so it is never purged automatically. + Undecodable { + /// The cause is a frame version this build predates, so the store holds data + /// written by a newer server. `/ready` reports `incompatible_store` for this + /// case and only this case. + newer_format: bool, + }, +} + +impl FrameFault { + /// The fault class name, for logs and for the operator-facing quarantine reason: + /// `integrity`, `newer_format`, or `undecodable`. + #[must_use] + pub const fn fault_class(&self) -> &'static str { + match self { + Self::Integrity => "integrity", + Self::Undecodable { newer_format: true } => "newer_format", + Self::Undecodable { newer_format: false } => "undecodable", + } + } + + /// Whether the fault means the file itself is damaged. + #[must_use] + pub const fn is_integrity(&self) -> bool { + matches!(self, Self::Integrity) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_registry_numbers_are_the_ones_the_formats_use() { + // Pinned against the constants the writers use, so a renumber here and a + // renumber there cannot drift apart. + assert_eq!(FRAME_SINGLE_BLOCK, crate::segment::SEGMENT_FORMAT_VERSION); + assert_eq!(FRAME_PAGED, crate::page::PAGED_SEGMENT_FORMAT_VERSION); + assert_eq!(FRAME_ENVELOPE, 7); + assert_eq!(FRAME_VERSION_MAX_KNOWN, FRAME_ENVELOPE); + } + + #[test] + fn layout_of_names_only_what_this_build_can_read() { + assert_eq!(layout_of(FRAME_SINGLE_BLOCK), Some(FrameLayout::SingleBlock)); + assert_eq!(layout_of(FRAME_PAGED), Some(FrameLayout::Paged)); + for unknown in [0, 1, 2, 3, 4, FRAME_ENVELOPE, 8, u16::MAX] { + assert_eq!(layout_of(unknown), None, "{unknown} has no reader in this build"); + } + } + + #[test] + fn no_extension_number_is_claimed_yet() { + // The envelope mechanism is what X-FRAME ships; the extensions come later, one + // per slice that needs a block in front of a frozen body. This assertion is the + // place where a new extension becomes visible instead of incidental. + assert!(KNOWN_EXTENSIONS.is_empty()); + for id in 0..=16 { + assert!(!is_known_extension(id), "extension {id} has no implementation in this build"); + } + // Numbers are a permanent namespace: a table may only ever gain distinct ids. + for (i, spec) in KNOWN_EXTENSIONS.iter().enumerate() { + assert!(!KNOWN_EXTENSIONS[i + 1..].iter().any(|later| later.id == spec.id), "extension number {} claimed twice", spec.id); + } + } + + #[test] + fn fault_classes_are_the_three_c5_names() { + assert_eq!(FrameFault::Integrity.fault_class(), "integrity"); + assert!(FrameFault::Integrity.is_integrity()); + assert_eq!(FrameFault::Undecodable { newer_format: true }.fault_class(), "newer_format"); + assert_eq!(FrameFault::Undecodable { newer_format: false }.fault_class(), "undecodable"); + assert!(!FrameFault::Undecodable { newer_format: false }.is_integrity()); + } +} diff --git a/weft-physical-type/src/lib.rs b/weft-physical-type/src/lib.rs index 08a7f0ac..7dd62854 100644 --- a/weft-physical-type/src/lib.rs +++ b/weft-physical-type/src/lib.rs @@ -106,6 +106,9 @@ pub mod catalog; pub mod column; #[cfg(feature = "experimental-codecs")] pub mod floatcodec; +/// The `.weftseg` frame version registry, and the two classes of frame-fault a read +/// can fail with (1.0 plan X-FRAME, resolution C5). +pub mod frame_registry; pub mod nulls; pub mod page; pub mod schema; @@ -122,6 +125,7 @@ pub use catalog::{SegmentDescriptor, SegmentIndex}; pub use column::{encode_column, recommend_encoding, CascadeInner, ColumnEncodeError, ColumnEncoding, DeltaCascadePlan}; #[cfg(feature = "experimental-codecs")] pub use floatcodec::{best_f64_bytes, best_f64_codec, chimp128_f64_bytes, chimp128_f64_decode, chimp128_f64_encode, chimp_f64_bytes, chimp_f64_decode, chimp_f64_encode, elf_f64_bytes, elf_f64_decode, elf_f64_encode, xor_f64_bytes, xor_f64_decode, xor_f64_encode}; +pub use frame_registry::{is_known_extension, layout_of, FrameExtension, FrameExtensionSpec, FrameFault, FrameInfo, FrameLayout, FRAME_ENVELOPE, FRAME_PAGED, FRAME_SINGLE_BLOCK, FRAME_VERSION_MAX_KNOWN, KNOWN_EXTENSIONS}; pub use nulls::{NullMask, NullMaskError}; pub use page::{Page, PagedSegment, DEFAULT_ROWS_PER_PAGE, PAGED_SEGMENT_FORMAT_VERSION}; pub use schema::{AspectSchema, SealError}; @@ -132,7 +136,7 @@ pub use tags::{SeriesKey, SeriesSelector, TagError, TagSet, MAX_SERIES_KEY_BYTES pub use timestamp::{bitpack_bytes, bitpack_decode, bitpack_encode, bitpack_width, decode_delta, decode_delta_of_delta, encode_delta, encode_delta_of_delta, first_order_violation, rle_decode, rle_encode, rle_varint_bytes, uvarint_len, zigzag_varint_bytes, zigzag_varint_len, DeltaColumn, DeltaOfDeltaColumn, TimeUnit}; #[cfg(feature = "experimental-codecs")] pub use timestamp::{fire_estimated_bytes, fire_reconstruct, fire_residuals}; -pub use weftseg::{crc32, frame_value_codec, read_paged_segment, read_segment, write_paged_segment, write_segment, write_segment_with, ByteReader, ByteWriter, FrameOptions, WeftSegError}; +pub use weftseg::{crc32, describe_frame, frame_fault_of, frame_layout_of, frame_value_codec, frame_version, read_frame_paged_segment, read_frame_point, read_frame_points, read_frame_range, read_frame_range_physical, read_frame_segment, read_paged_segment, read_segment, unwrap_frame_envelope, wrap_frame_envelope, write_paged_segment, write_segment, write_segment_with, ByteReader, ByteWriter, FrameOptions, WeftSegError}; /// A schema-declared physical encoding for an aspect's numeric values. /// diff --git a/weft-physical-type/src/weftseg.rs b/weft-physical-type/src/weftseg.rs index ea61e05f..c782579a 100644 --- a/weft-physical-type/src/weftseg.rs +++ b/weft-physical-type/src/weftseg.rs @@ -31,7 +31,13 @@ //! by construction, because both count the same varint widths. /// Why a `.weftseg` byte stream could not be read. +/// +/// The variants are open: a new frame layout or a new decode restriction adds one +/// without breaking a caller's `match`. Pair a failure with +/// [`frame_fault_of`] to get the class an operator +/// acts on rather than the byte that tripped. #[derive(Debug, Clone, PartialEq, Eq)] +#[non_exhaustive] pub enum WeftSegError { /// The reader ran out of bytes before a value was fully decoded. UnexpectedEof { @@ -89,6 +95,25 @@ pub enum WeftSegError { /// The cargo feature of `weft-physical-type` that compiles the codec in. feature: &'static str, }, + /// A v7 envelope carries a **critical** extension this build does not implement, so + /// the body it wraps cannot be trusted: "critical" is the writer's way of saying the + /// body below is only correct once a reader has applied this. The file is not + /// damaged — a build that implements extension `id` reads it — so it is quarantined + /// as suspect and never purged (1.0 plan resolution C5). + UnknownCriticalExtension { + /// The extension number the frame requires. + id: u32, + }, + /// A v7 envelope declares a body layout that the frame inside it does not name. Both + /// fields are covered by the same checksum, so no writer produces this: the file was + /// edited, or a writer has a bug. It is counted as damage rather than as an unknown + /// format, because there is no build that could read it. + EnvelopeBodyMismatch { + /// The layout the envelope's `body_version` field declares. + declared: u16, + /// The version the wrapped frame names in its own header. + found: u16, + }, } impl std::fmt::Display for WeftSegError { @@ -105,6 +130,8 @@ impl std::fmt::Display for WeftSegError { Self::TrailingBytes { remaining } => write!(f, "{remaining} trailing bytes after segment frame"), Self::InvalidNullMask(source) => write!(f, "invalid quality column: {source}"), Self::CodecNotEnabled { codec, feature } => write!(f, "segment uses the {codec} codec, which this build does not include; rebuild with the `{feature}` cargo feature (weft-physical-type; forwarded by weftdb and weft-server) to read it"), + Self::UnknownCriticalExtension { id } => write!(f, "frame carries critical extension {id}, which this build does not implement; the frame is left intact because a build that implements it can read it"), + Self::EnvelopeBodyMismatch { declared, found } => write!(f, "v7 envelope declares a {declared} body but the frame inside it names version {found}; the envelope's own checksum is correct, so this file was written or edited inconsistently"), } } } @@ -2205,7 +2232,7 @@ fn read_dod_codec(r: &mut ByteReader, count: usize) -> Result<(Vec, Option< // --------------------------------------------------------------------------- use crate::{ - nulls::NullMask, page::{Page, PagedSegment, PAGED_SEGMENT_FORMAT_VERSION}, segment::SegmentStats, Segment, SEGMENT_FORMAT_VERSION + frame_registry::{layout_of, FrameExtension, FrameFault, FrameInfo, FrameLayout, FRAME_ENVELOPE, FRAME_VERSION_MAX_KNOWN}, nulls::NullMask, page::{Page, PagedSegment, PAGED_SEGMENT_FORMAT_VERSION}, segment::SegmentStats, Segment, SEGMENT_FORMAT_VERSION }; /// The magic prefix every `.weftseg` frame starts with. @@ -2319,6 +2346,352 @@ fn read_segment_stats(r: &mut ByteReader) -> Result Ok(SegmentStats { row_count, null_count, time_sorted, min_ts, max_ts, min_value, max_value }) } +// --------------------------------------------------------------------------- +// The v7 envelope: a header in front of a body whose bytes are already frozen. +// --------------------------------------------------------------------------- + +/// The layout version a frame names in its own header, without touching anything +/// after it. The cheapest thing a server can do with a file it is about to open. +/// +/// # Errors +/// +/// [`WeftSegError::UnexpectedEof`] for a file too short to carry a header, +/// [`WeftSegError::BadMagic`] for bytes that are not a `.weftseg` frame at all, and +/// never a version error — this reports the version rather than judging it. +pub fn frame_version(bytes: &[u8]) -> Result { + if bytes.len() < MAGIC.len() + 2 { + return Err(WeftSegError::UnexpectedEof { needed: MAGIC.len() + 2, remaining: bytes.len() }); + } + if !is_known_magic(&bytes[..MAGIC.len()]) { + return Err(WeftSegError::BadMagic); + } + Ok(u16::from_le_bytes([bytes[MAGIC.len()], bytes[MAGIC.len() + 1]])) +} + +/// Which reader a frame needs, read from its header alone. +/// +/// This is a routing decision, not a verdict: nothing past the version fields is +/// trusted or verified here, and the reader this chooses checks every byte that +/// follows. That is what makes it safe to route on the number in the file rather +/// than the number copied into an index row when the frame was written — a frame +/// whose header was damaged lands in whichever reader the damaged byte suggests and +/// fails there on its own checksum, which is the honest answer either way. +/// +/// An envelope's answer is the layout of the frame it wraps, so a caller can pick a +/// layout-specific entry point without knowing whether the envelope is in use. +#[must_use] +pub fn frame_layout_of(bytes: &[u8]) -> Result { + let version = frame_version(bytes)?; + if let Some(layout) = layout_of(version) { + return Ok(layout); + } + if version != FRAME_ENVELOPE { + return Err(WeftSegError::UnsupportedVersion { found: version }); + } + // The envelope's own version is followed by the version of the frame inside it. + if bytes.len() < MAGIC.len() + 4 { + return Err(WeftSegError::UnexpectedEof { needed: MAGIC.len() + 4, remaining: bytes.len() }); + } + let body_version = ByteReader::new(&bytes[MAGIC.len() + 2..]).read_u16_le()?; + layout_of(body_version).ok_or(WeftSegError::UnsupportedVersion { found: body_version }) +} + +/// A frame taken apart: header, extension table, and the body bytes a reader of +/// `body_version` can be handed as they stand. +struct FrameParts<'a> { + /// The version in the frame's own header. + version: u16, + /// The version of the body layout. + body_version: u16, + /// The extension table, empty for a plain frame. + extensions: Vec, + /// A critical extension this build does not implement was named as one. + unknown_critical: bool, + /// The complete frame of `body_version`: magic, version, body and that frame's own + /// CRC trailer. For a plain v5/v6 frame this is the whole input, which is what + /// every existing reader in this file expects to be handed. + body: &'a [u8], +} + +/// Split a frame into its header, extension table and body, checking the frame's own +/// trailing CRC on the way. +/// +/// A v5 or v6 frame comes back with `body == bytes`: the envelope pass changes nothing +/// about how the two historical layouts are read. A v7 frame is checked end to end — +/// the outer CRC covers the header, the table and the wrapped frame including that +/// frame's own CRC — and its wrapped frame is returned as the body, verbatim. +fn frame_parts(bytes: &[u8]) -> Result, WeftSegError> { + let version = frame_version(bytes)?; + if bytes.len() < MAGIC.len() + 6 { + // Header plus a CRC-32 trailer is the smallest frame that could ever be valid. + return Err(WeftSegError::UnexpectedEof { needed: MAGIC.len() + 6, remaining: bytes.len() }); + } + let (framed, trailer) = bytes.split_at(bytes.len() - 4); + let stored = u32::from_le_bytes([trailer[0], trailer[1], trailer[2], trailer[3]]); + let computed = crc32(framed); + if stored != computed { + return Err(WeftSegError::ChecksumMismatch { stored, computed }); + } + if version != FRAME_ENVELOPE { + return Ok(FrameParts { version, body_version: version, extensions: Vec::new(), unknown_critical: false, body: bytes }); + } + // Skip the magic and the envelope's own version; what follows is body_version, + // then the extension table, then the wrapped frame. + let rest = &framed[MAGIC.len() + 2..]; + let mut r = ByteReader::new(rest); + let body_version = r.read_u16_le()?; + let count = r.read_uvarint()?; + // The count is not trusted as a size hint: entries are parsed one at a time and + // each needs bytes that must actually be present, so a count invented by damage + // stops at the first entry that is not there instead of reserving anything. + let mut extensions = Vec::new(); + let mut unknown_critical = false; + for _ in 0..count { + let id = u32::try_from(r.read_uvarint()?).map_err(|_| WeftSegError::VarintTooLong)?; + let critical = match r.read_u8()? { + 0 => false, + 1 => true, + value => return Err(WeftSegError::InvalidTag { kind: "extension critical flag", value }), + }; + let payload = r.read_bytes()?.to_vec(); + if critical && !crate::frame_registry::is_known_extension(id) { + unknown_critical = true; + } + extensions.push(FrameExtension { id, critical, payload }); + } + let body = &rest[r.pos..]; + // The wrapped frame carries its own header and CRC, and it has to name the layout the + // envelope says it is. Both fields sit under the same checksum, so a writer cannot + // produce a pair that disagrees: this is a file that was edited, or a writer with a + // bug. The check lives here rather than in each caller so that every reader sees the + // same answer, and the error names what was found rather than what was claimed — a + // version error here would be a lie about a version this build has never heard of. + let found = frame_version(body)?; + if found != body_version { + return Err(WeftSegError::EnvelopeBodyMismatch { declared: body_version, found }); + } + Ok(FrameParts { version, body_version, extensions, unknown_critical, body }) +} + +/// Wrap a complete frame in a v7 envelope carrying `extensions`. +/// +/// The frame inside is stored **verbatim** — magic, version and its own CRC included — +/// and the envelope adds its header plus a CRC over everything. That costs 16 bytes — +/// a 12-byte header (magic, envelope version, body version, entry count) plus a 4-byte +/// trailer — +/// and buys the envelope's whole point: not one byte inside a v5 or v6 frame moves, so +/// the byte-for-byte expectations that released servers and on-disk stores already +/// hold keep holding, while a reader that understands envelopes can put new material +/// in front of a body it is not allowed to change. +/// +/// # Errors +/// +/// [`WeftSegError::BadMagic`] or [`WeftSegError::UnsupportedVersion`] if `inner` is not +/// a frame of a layout this build can read — an envelope must wrap something real. +pub fn wrap_frame_envelope(inner: &[u8], extensions: &[FrameExtension]) -> Result, WeftSegError> { + let body_version = frame_version(inner)?; + if layout_of(body_version).is_none() { + return Err(WeftSegError::UnsupportedVersion { found: body_version }); + } + let extra = extensions.iter().map(|e| e.payload.len() + 6).sum::(); + let mut w = ByteWriter::with_capacity(inner.len() + extra + 32); + w.put_raw(MAGIC); + w.put_u16_le(FRAME_ENVELOPE); + w.put_u16_le(body_version); + w.put_uvarint(extensions.len() as u64); + for e in extensions { + w.put_uvarint(u64::from(e.id)); + w.put_u8(u8::from(e.critical)); + w.put_bytes(&e.payload); + } + w.put_raw(inner); + let checksum = crc32(w.as_slice()); + w.put_u32_le(checksum); + Ok(w.into_vec()) +} + +/// The complete frame inside a v7 envelope, with the envelope's CRC verified and its +/// extension table checked. +/// +/// A plain v5/v6 frame is returned unchanged, so a caller that reads frames through +/// this function does not have to know which layout it was given. +/// +/// # Errors +/// +/// [`WeftSegError::UnknownCriticalExtension`] if the frame carries a critical extension +/// this build does not implement — the body was written on the assumption that a reader +/// would apply it, so decoding the body without it would serve data the writer never +/// meant. Also the CRC, magic, length and varint errors of the envelope check itself. +pub fn unwrap_frame_envelope(bytes: &[u8]) -> Result<&[u8], WeftSegError> { + let parts = frame_parts(bytes)?; + if parts.unknown_critical { + // Named separately from the generic refusal so the message can say which number + // the operator's server has to grow. + if let Some(e) = parts.extensions.iter().find(|e| e.critical && !crate::frame_registry::is_known_extension(e.id)) { + return Err(WeftSegError::UnknownCriticalExtension { id: e.id }); + } + } + // `frame_parts` has already checked that a wrapped frame names the layout its + // envelope declares, and the legacy readers below re-check its CRC over exactly + // these bytes. + Ok(parts.body) +} + +/// Describe a frame without decoding it: which layout it is, which extensions it +/// carries, and whether this build can serve it. +/// +/// This is what a server answers for a frame it cannot read, and what an operator +/// script reads to decide whether a file is damaged or simply ahead of the software. +/// The answer is already classified: a `FrameFault` says which of the two, and +/// `FrameInfo::decodable()` says whether the body may be served at all. +/// +/// # Errors +/// +/// The [`FrameFault`] class of whatever went wrong — see [`frame_fault_of`]. +pub fn describe_frame(bytes: &[u8]) -> Result { + match frame_parts(bytes) { + Ok(parts) => Ok(FrameInfo { version: parts.version, body_version: parts.body_version, layout: layout_of(parts.body_version), extensions: parts.extensions, unknown_critical: parts.unknown_critical }), + Err(error) => Err(frame_fault_of(&error)), + } +} + +/// The operator-facing class of a frame read failure, per the plan's C5 resolution. +/// +/// **Integrity** is damage: the bytes disagree with their own magic, checksum or +/// length, and the quarantine machinery may purge the file. Everything else is +/// **undecodable**: the file is self-consistent, and the only reason it failed is that +/// this build has no reader for it — a version it predates, a codec feature it was +/// compiled without, an extension it does not implement, or a tag it does not +/// recognise. Those files hold data that some other build can read, so they are never +/// purged automatically. +/// +/// `newer_format` is set when the cause is a frame version above +/// [`FRAME_VERSION_MAX_KNOWN`], which is the case where the store as a whole should +/// report `incompatible_store` rather than blame one file. +#[must_use] +pub const fn frame_fault_of(error: &WeftSegError) -> FrameFault { + match error { + WeftSegError::UnsupportedVersion { found } => FrameFault::Undecodable { newer_format: *found > FRAME_VERSION_MAX_KNOWN }, + WeftSegError::CodecNotEnabled { .. } | WeftSegError::UnknownCriticalExtension { .. } => FrameFault::Undecodable { newer_format: false }, + // BadMagic, ChecksumMismatch, UnexpectedEof, VarintTooLong, TrailingBytes, + // InvalidUtf8, InvalidDecimal, InvalidNullMask, EnvelopeBodyMismatch: the file + // does not agree with itself. InvalidTag is counted with them because a tag this + // build does not know arrives with a new frame or body version, not inside a + // version it already understands. + _ => FrameFault::Integrity, + } +} + +/// The body layout of a frame this build can read, envelope or not, together with the +/// body bytes that reader should be handed. +/// +/// # Errors +/// +/// [`WeftSegError::UnsupportedVersion`] for a body version with no reader (including a +/// frame from a newer server), and [`WeftSegError::UnknownCriticalExtension`] for a +/// critical extension this build does not implement. +fn read_frame_body(bytes: &[u8]) -> Result<(&[u8], FrameLayout), WeftSegError> { + let parts = frame_parts(bytes)?; + if parts.unknown_critical { + if let Some(e) = parts.extensions.iter().find(|e| e.critical && !crate::frame_registry::is_known_extension(e.id)) { + return Err(WeftSegError::UnknownCriticalExtension { id: e.id }); + } + } + let body = if parts.version == FRAME_ENVELOPE { + // `frame_parts` already proved the wrapped frame names the layout declared in the + // envelope header, so the layout below is the layout of these bytes. + parts.body + } else { + bytes + }; + match layout_of(parts.body_version) { + Some(layout) => Ok((body, layout)), + None => Err(WeftSegError::UnsupportedVersion { found: parts.body_version }), + } +} + +/// Read a single-block frame, whether or not it is wrapped in a v7 envelope. +/// +/// The envelope-aware form of [`read_segment`], for a caller that does not know how the +/// file was framed. +/// +/// # Errors +/// +/// The [`WeftSegError`]s of [`unwrap_frame_envelope`] and [`read_segment`]. Use +/// [`frame_fault_of`] to turn one into the class the operator acts on. +pub fn read_frame_segment(bytes: &[u8]) -> Result { + read_segment(unwrap_frame_envelope(bytes)?) +} + +/// Read a paged frame, whether or not it is wrapped in a v7 envelope. +/// +/// The envelope-aware form of [`read_paged_segment`]. +/// +/// # Errors +/// +/// The [`WeftSegError`]s of [`unwrap_frame_envelope`] and [`read_paged_segment`]. +pub fn read_frame_paged_segment(bytes: &[u8]) -> Result { + read_paged_segment(unwrap_frame_envelope(bytes)?) +} + +/// Read one point from a frame of either layout, wrapped or not. +/// +/// The envelope-aware form of [`read_segment_point`] and [`read_paged_segment_point`], +/// dispatching on the layout the file itself names rather than on what a caller +/// believes it wrote. +/// +/// # Errors +/// +/// [`WeftSegError::UnsupportedVersion`] for a body layout this build has no point +/// reader for, otherwise the errors of the two underlying readers. +pub fn read_frame_point(bytes: &[u8], t: i64) -> Result, WeftSegError> { + let (body, layout) = read_frame_body(bytes)?; + match layout { + FrameLayout::SingleBlock => read_segment_point(body, t), + FrameLayout::Paged => read_paged_segment_point(body, t), + } +} + +/// Read several points from a frame of either layout, wrapped or not. +/// +/// # Errors +/// +/// As [`read_frame_point`]. +pub fn read_frame_points(bytes: &[u8], ts: &[i64]) -> Result>, WeftSegError> { + let (body, layout) = read_frame_body(bytes)?; + match layout { + FrameLayout::SingleBlock => read_segment_points(body, ts), + FrameLayout::Paged => read_paged_segment_points(body, ts), + } +} + +/// Read a timestamp range from a frame of either layout, wrapped or not. +/// +/// # Errors +/// +/// As [`read_frame_point`]. +pub fn read_frame_range(bytes: &[u8], start: i64, end: i64) -> Result<(Vec, Vec>), WeftSegError> { + let (body, layout) = read_frame_body(bytes)?; + match layout { + FrameLayout::SingleBlock => read_segment_range(body, start, end), + FrameLayout::Paged => read_paged_segment_range(body, start, end), + } +} + +/// Read a timestamp range as physical values from a frame of either layout, wrapped or +/// not. +/// +/// # Errors +/// +/// As [`read_frame_point`]. +pub fn read_frame_range_physical(bytes: &[u8], start: i64, end: i64) -> Result<(Vec, Vec>), WeftSegError> { + let (body, layout) = read_frame_body(bytes)?; + match layout { + FrameLayout::SingleBlock => read_segment_range_physical(body, start, end), + FrameLayout::Paged => read_paged_segment_range_physical(body, start, end), + } +} + /// Encode a [`Segment`] into a complete, self-describing, checksummed `.weftseg` /// byte frame. /// @@ -2696,6 +3069,7 @@ pub fn read_paged_segment(bytes: &[u8]) -> Result { #[cfg(test)] mod tests { use super::*; + use crate::frame_registry::{FrameExtension, FrameFault, FrameLayout, FRAME_ENVELOPE, FRAME_PAGED, FRAME_SINGLE_BLOCK, FRAME_VERSION_MAX_KNOWN}; #[test] fn crc32_slicing_by_16_matches_the_byte_at_a_time_crc() { @@ -4174,6 +4548,233 @@ mod tests { assert_eq!(read_segment(&bytes), Err(WeftSegError::UnsupportedVersion { found: SEGMENT_FORMAT_VERSION + 1 })); } + // --------------------------------------------------------------------- + // The v7 envelope, the frame registry, and the two fault classes C5 names + // --------------------------------------------------------------------- + + /// Re-stamp a frame's trailing CRC-32 over everything before it, so a body edit can + /// be made with the trailer telling the truth about the damaged bytes. + fn restamp(frame: &[u8]) -> Vec { + let mut bytes = frame.to_vec(); + let body_len = bytes.len() - 4; + let new_crc = crc32(&bytes[..body_len]).to_le_bytes(); + bytes[body_len..].copy_from_slice(&new_crc); + bytes + } + + /// Rewrite the u16 version at `at` and re-stamp the trailer: a frame that is + /// internally consistent apart from the number that was changed, which is how a + /// version this build has never seen arrives from a future release. + fn reversion(frame: &[u8], at: usize, version: u16) -> Vec { + let mut bytes = frame.to_vec(); + bytes[at..at + 2].copy_from_slice(&version.to_le_bytes()); + restamp(&bytes) + } + + /// A single-block frame with enough rows that point and range reads do real work. + fn single_block_frame() -> Vec { + write_segment(&single_block_segment()) + } + + fn single_block_segment() -> Segment { + let ts: Vec = (0..400).map(|i| 1_600_000_000 + i * 60).collect(); + let vs: Vec = (0..400).map(|i| BigDecimal::from(i * 3)).collect(); + Segment::build(&ts, &vs, TimeUnit::Seconds, &BigDecimal::from(0)).expect("builds") + } + + /// A paged frame spanning several pages. + fn paged_frame() -> Vec { + write_paged_segment(&paged_segment()) + } + + fn paged_segment() -> PagedSegment { + let ts: Vec = (0..400).map(|i| 1_600_000_000 + i * 60).collect(); + let vs: Vec = (0..400).map(|i| BigDecimal::from(i * 3)).collect(); + let seg = PagedSegment::build(&ts, &vs, TimeUnit::Seconds, &BigDecimal::from(0), 64).expect("builds"); + assert!(seg.page_count() > 1, "the point of these tests is more than one page"); + seg + } + + #[test] + fn an_envelope_stores_the_frame_it_wraps_without_moving_a_byte() { + let inner = single_block_frame(); + let wrapped = wrap_frame_envelope(&inner, &[]).expect("wraps"); + // magic(7) + envelope version(2) + body version(2) + entry count(1) = 12 bytes of + // header, then the inner frame entire, then the envelope's own CRC. + let header = MAGIC.len() + 2 + 2 + 1; + assert_eq!(&wrapped[header..header + inner.len()], &inner[..], "the wrapped frame keeps its own magic, version and CRC untouched"); + assert_eq!(wrapped.len(), inner.len() + 16, "an envelope with an empty extension table costs 16 bytes"); + assert_eq!(frame_version(&wrapped).expect("version"), FRAME_ENVELOPE); + assert_eq!(unwrap_frame_envelope(&wrapped).expect("unwraps"), &inner[..]); + } + + #[test] + fn the_frame_readers_answer_the_same_for_a_wrapped_frame() { + for plain in [single_block_frame(), paged_frame()] { + let wrapped = wrap_frame_envelope(&plain, &[]).expect("wraps"); + // A plain frame has to go through the same entry points too, or the family would + // be a second format a caller has to detect rather than a superset of the first. + assert_eq!(read_frame_point(&wrapped, 1_600_000_060).expect("wrapped point"), read_frame_point(&plain, 1_600_000_060).expect("plain point")); + assert_eq!(read_frame_points(&wrapped, &[1_600_000_060, 1_600_001_200]).expect("wrapped points"), read_frame_points(&plain, &[1_600_000_060, 1_600_001_200]).expect("plain points")); + assert_eq!(read_frame_range(&wrapped, 1_600_000_000, 1_600_003_000).expect("wrapped range"), read_frame_range(&plain, 1_600_000_000, 1_600_003_000).expect("plain range")); + assert_eq!(read_frame_range_physical(&wrapped, 1_600_000_000, 1_600_003_000).expect("wrapped physical"), read_frame_range_physical(&plain, 1_600_000_000, 1_600_003_000).expect("plain physical")); + } + // The layout-specific entry points keep working too, wrapped or not. + let single = single_block_frame(); + let single_wrapped = wrap_frame_envelope(&single, &[]).expect("wraps"); + assert_eq!(read_frame_segment(&single_wrapped).expect("segment"), read_segment(&single).expect("segment")); + let paged = paged_frame(); + let paged_wrapped = wrap_frame_envelope(&paged, &[]).expect("wraps"); + assert_eq!(read_frame_paged_segment(&paged_wrapped).expect("paged"), read_paged_segment(&paged).expect("paged")); + } + + #[test] + fn a_critical_extension_this_build_lacks_refuses_the_frame() { + let inner = single_block_frame(); + let wrapped = wrap_frame_envelope(&inner, &[FrameExtension { id: 7, critical: true, payload: vec![1, 2, 3] }]).expect("wraps"); + // Decoding the body without applying the extension would serve data the writer + // never meant, so every reader in the family refuses instead. + assert_eq!(read_frame_point(&wrapped, 1_600_000_060), Err(WeftSegError::UnknownCriticalExtension { id: 7 })); + assert_eq!(read_frame_range(&wrapped, 1_600_000_000, 1_600_003_000), Err(WeftSegError::UnknownCriticalExtension { id: 7 })); + assert_eq!(unwrap_frame_envelope(&wrapped), Err(WeftSegError::UnknownCriticalExtension { id: 7 })); + // Describing it is still allowed: the fault class is what an operator acts on, and + // the file is undecodable-but-intact, not damaged. + let info = describe_frame(&wrapped).expect("describes"); + assert_eq!(info.version, FRAME_ENVELOPE); + assert_eq!(info.body_version, FRAME_SINGLE_BLOCK); + assert!(info.unknown_critical); + assert!(!info.decodable()); + assert_eq!(info.extensions.len(), 1); + assert_eq!(info.extensions[0].id, 7); + assert!(info.extensions[0].critical); + let error = read_frame_point(&wrapped, 1_600_000_060).expect_err("refuses"); + assert_eq!(frame_fault_of(&error), FrameFault::Undecodable { newer_format: false }); + // A build that has never heard of the envelope refuses it too, rather than reading + // the extension table as if it were a segment header. + assert_eq!(read_segment(&wrapped), Err(WeftSegError::UnsupportedVersion { found: FRAME_ENVELOPE })); + } + + #[test] + fn an_informational_extension_is_carried_and_ignored() { + let inner = single_block_frame(); + let wrapped = wrap_frame_envelope(&inner, &[FrameExtension { id: 3, critical: false, payload: b"written-by/version".to_vec() }]).expect("wraps"); + assert_eq!(read_frame_range(&wrapped, 1_600_000_000, 1_600_003_000).expect("range"), read_frame_range(&inner, 1_600_000_000, 1_600_003_000).expect("range")); + let info = describe_frame(&wrapped).expect("describes"); + assert!(!info.unknown_critical); + assert!(info.decodable()); + assert_eq!(info.extensions[0].payload, b"written-by/version"); + } + + #[test] + fn a_frame_from_a_newer_build_is_the_newer_format_fault() { + let renamed = reversion(&single_block_frame(), MAGIC.len(), FRAME_VERSION_MAX_KNOWN + 1); + // Describing is not decoding: the shape is legible, there is just no reader. + let info = describe_frame(&renamed).expect("describes"); + assert_eq!(info.version, FRAME_VERSION_MAX_KNOWN + 1); + assert_eq!(info.layout, None, "no reader in this build is named for it"); + assert!(!info.decodable()); + let error = read_frame_point(&renamed, 1_600_000_060).expect_err("refuses"); + assert_eq!(error, WeftSegError::UnsupportedVersion { found: FRAME_VERSION_MAX_KNOWN + 1 }); + // The whole store is suspect in this case, not one file: a build that wrote it is + // ahead of this one, which is what /ready reports as incompatible_store. + assert_eq!(frame_fault_of(&error), FrameFault::Undecodable { newer_format: true }); + // Writing one is refused too: an envelope has to wrap something this build can read. + assert_eq!(wrap_frame_envelope(&renamed, &[]), Err(WeftSegError::UnsupportedVersion { found: FRAME_VERSION_MAX_KNOWN + 1 })); + } + + #[test] + fn an_older_version_is_not_reported_as_a_newer_build() { + // The distinction matters because the two faults get different treatment: a frame + // this old was written by something that could not have known better, so blaming + // the store's version would send an operator hunting for a server that does not + // exist. + let renamed = reversion(&single_block_frame(), MAGIC.len(), FRAME_SINGLE_BLOCK - 1); + let error = read_frame_point(&renamed, 1_600_000_060).expect_err("refuses"); + assert_eq!(error, WeftSegError::UnsupportedVersion { found: FRAME_SINGLE_BLOCK - 1 }); + assert_eq!(frame_fault_of(&error), FrameFault::Undecodable { newer_format: false }); + } + + #[test] + fn an_envelope_that_disagrees_with_its_body_is_damage() { + let wrapped = wrap_frame_envelope(&single_block_frame(), &[]).expect("wraps"); + // The envelope's body_version sits after the magic and its own version. + let contradicted = reversion(&wrapped, MAGIC.len() + 2, FRAME_PAGED); + assert_eq!(read_frame_segment(&contradicted), Err(WeftSegError::EnvelopeBodyMismatch { declared: FRAME_PAGED, found: FRAME_SINGLE_BLOCK })); + assert_eq!(read_frame_point(&contradicted, 1_600_000_060), Err(WeftSegError::EnvelopeBodyMismatch { declared: FRAME_PAGED, found: FRAME_SINGLE_BLOCK })); + // Both numbers sit under the same checksum, so nothing wrote this file honestly: + // it is damage, not a format this build merely predates. + assert_eq!(describe_frame(&contradicted), Err(FrameFault::Integrity)); + } + + #[test] + fn an_inner_frame_keeps_its_own_checksum_under_the_envelope() { + let inner = single_block_frame(); + let wrapped = wrap_frame_envelope(&inner, &[]).expect("wraps"); + let header = MAGIC.len() + 2 + 2 + 1; + // Damage in the body, with the envelope's trailer honestly recomputed over it: the + // outer check passes and the inner frame's own CRC still catches the byte. + let mut damaged = wrapped.clone(); + // A byte in the middle of the wrapped frame's column data: not the magic, not the + // version, nothing the envelope inspects for itself. + damaged[header + inner.len() / 2] ^= 0xFF; + let stamped = restamp(&damaged); + assert_eq!(describe_frame(&stamped).expect("describes").version, FRAME_ENVELOPE, "the envelope itself reads as intact"); + let error = read_frame_segment(&stamped).expect_err("the body does not"); + assert!(matches!(error, WeftSegError::ChecksumMismatch { .. }), "the wrapped frame's own trailer is what catches it: {error:?}"); + assert!(frame_fault_of(&error).is_integrity()); + // The same byte without re-stamping is caught one layer earlier, and still classifies + // the same way, so the two layers cannot tell an operator two different stories. + let error = read_frame_segment(&damaged).expect_err("refuses"); + assert!(matches!(error, WeftSegError::ChecksumMismatch { .. }), "{error:?}"); + assert!(frame_fault_of(&error).is_integrity()); + } + + #[test] + fn describe_frame_names_the_layout_of_every_frame() { + let single = single_block_frame(); + let paged = paged_frame(); + let info = describe_frame(&single).expect("v5"); + assert_eq!((info.version, info.body_version, info.layout), (FRAME_SINGLE_BLOCK, FRAME_SINGLE_BLOCK, Some(FrameLayout::SingleBlock))); + assert!(info.extensions.is_empty() && !info.unknown_critical && info.decodable()); + let info = describe_frame(&paged).expect("v6"); + assert_eq!((info.version, info.body_version, info.layout), (FRAME_PAGED, FRAME_PAGED, Some(FrameLayout::Paged))); + let info = describe_frame(&wrap_frame_envelope(&paged, &[]).expect("wraps")).expect("v7 of v6"); + assert_eq!((info.version, info.body_version, info.layout), (FRAME_ENVELOPE, FRAME_PAGED, Some(FrameLayout::Paged))); + // Damage is described as damage, including a frame too short to hold a trailer. + assert_eq!(describe_frame(&[]), Err(FrameFault::Integrity)); + assert_eq!(describe_frame(&[0; 3]), Err(FrameFault::Integrity)); + } + + #[test] + fn an_envelope_wraps_a_checkpointed_frame() { + // The opt-in layouts are exactly the frames whose bytes the envelope exists to + // protect, so a wrapped checkpointed frame has to read back identically. + let ts: Vec = (0..400).map(|i| 1_600_000_000 + i * 60).collect(); + let vs: Vec = (0..400).map(|i| BigDecimal::from(i * 3)).collect(); + let _ = (&ts, &vs); + let seg = single_block_segment(); + let plain = write_segment_checkpointed(&seg, 64); + let wrapped = wrap_frame_envelope(&plain, &[]).expect("wraps"); + assert_eq!(read_frame_segment(&wrapped).expect("segment"), read_segment(&plain).expect("segment")); + let paged = paged_segment(); + let plain = write_paged_segment_checkpointed(&paged, 64); + let wrapped = wrap_frame_envelope(&plain, &[]).expect("wraps"); + assert_eq!(read_frame_paged_segment(&wrapped).expect("paged"), read_paged_segment(&plain).expect("paged")); + assert_eq!(read_frame_range(&wrapped, 1_600_000_000, 1_600_003_000).expect("range"), read_paged_segment_range(&plain, 1_600_000_000, 1_600_003_000).expect("legacy range")); + } + + #[test] + fn the_envelope_rejects_a_frame_it_was_handed_that_is_not_a_frame() { + // Not a .weftseg at all, correctly checksummed: refused by magic, and classed as + // damage because there is no build that could read it. + let mut junk = b"NOTSEG!".to_vec(); + junk.extend_from_slice(&crc32(&junk).to_le_bytes()); + assert_eq!(unwrap_frame_envelope(&junk), Err(WeftSegError::BadMagic)); + assert_eq!(describe_frame(&junk), Err(FrameFault::Integrity)); + // A truncated envelope: the header does not even fit. + assert!(matches!(unwrap_frame_envelope(&[b'W', b'E', b'F', b'T', b'S', b'E', b'G', 7, 0, 5, 0]), Err(WeftSegError::UnexpectedEof { .. }))); + } + #[test] fn realized_frame_decodes_to_the_original_data() { // End-to-end: raw data -> segment -> bytes -> segment -> raw data. diff --git a/weft-server/src/error.rs b/weft-server/src/error.rs index 4d55cf20..e5ac641c 100644 --- a/weft-server/src/error.rs +++ b/weft-server/src/error.rs @@ -571,6 +571,15 @@ pub(crate) fn classify_typed(err: &anyhow::Error) -> Option { if let Some(e) = err.downcast_ref::() { return Some(aspect_name_refusal(e)); } + // A frame on disk that this build could not decode. The fault class decides the status, + // not the decoder's words: a frame written by a newer build is a store this binary must + // refuse (C5 sends `incompatible_store`, and the operator's next step is the binary, not + // the file), while damage or a frame needing an extension or codec this build lacks is + // `frame_undecodable`. Neither is ever derived from the message, which is kept + // byte-for-byte what these sites have always logged. + if let Some(e) = err.downcast_ref::() { + return Some(from_frame_fault(e)); + } // A store on disk this binary cannot open: exit `65` for a CLI, `503 incompatible_store` // here, and the same code for every reason (newer layout, unreadable `STORE_FORMAT`). if let Some(e) = err.downcast_ref::() { @@ -588,6 +597,23 @@ pub(crate) fn classify_typed(err: &anyhow::Error) -> Option { None } +/// A frame whose bytes this build could not turn into rows. +/// +/// `503 incompatible_store` only for `newer_format`, because that is the one case where the +/// file is fine and the running binary is the thing that is wrong: the same store answers once +/// the server is updated, so it is reported as a store/version condition and `/ready` says so +/// too. Damage stays `500 frame_undecodable` — the request was not the problem, but the store as +/// a whole is not unusable either, and C5's quarantine of the file happens in the storage layer. +/// A frame needing a critical extension or an off-by-default codec also stays `500`: the build is +/// behind on a feature, not on the format, and the message names what to enable. +pub fn from_frame_fault(e: &weftdb::FrameFaultError) -> ApiError { + use weft_physical_type::FrameFault; + + let fault = e.fault(); + let code = if matches!(fault, FrameFault::Undecodable { newer_format: true }) { Code::IncompatibleStore } else { Code::FrameUndecodable }; + ApiError::server(code, e.to_string()).with_details(serde_json::json!({ "fault": fault.fault_class() })) +} + /// A batch that will not seal under the aspect's declared schema. /// /// This is where the old code's `400 bad_request` for every seal failure splits apart, as the diff --git a/weft-server/tests/frame_fault.rs b/weft-server/tests/frame_fault.rs new file mode 100644 index 00000000..977fe26a --- /dev/null +++ b/weft-server/tests/frame_fault.rs @@ -0,0 +1,115 @@ +//! What a client gets when a frame on disk cannot be decoded. +//! +//! X-ERR gave `frame_undecodable` a sentence and X-FRAME gave it a producer, and this file is +//! where the two meet. It cannot be a unit test: `weftdb::FrameFaultError` has no public +//! constructor and `WeftSegError` is `#[non_exhaustive]`, precisely so that a status cannot be +//! invented here — the only way to reach these branches is a real file the store cannot read. +//! +//! The two cases are the two halves of plan `C5`. Damage is the store's problem and not the +//! caller's, so it is `500 frame_undecodable` with the `integrity` class: the request was fine, +//! the file is not, and quarantining it is a storage-layer decision that belongs to the slice +//! which verifies frames on read. A frame written by a newer build is the *binary's* problem, so +//! it is `503 incompatible_store` with the `newer_format` class — the same store answers as soon +//! as the server is updated, which is why the code is the one `/ready` already uses to say "this +//! store needs a newer server". Neither answer is derived from the decoder's wording: the class +//! comes from the fault object, and the sentence stays the fixed one X-ERR chose. + +use std::{ + fs, path::{Path, PathBuf}, sync::Arc +}; + +use axum::{ + body::Body, http::{Request, StatusCode} +}; +use bigdecimal::BigDecimal; +use tower::ServiceExt; +use weft_physical_type::{crc32, AspectSchema, PhysicalType, TimeUnit, FRAME_VERSION_MAX_KNOWN}; +use weft_server::{app_with_state, AppState}; +use weftdb::SegmentStore; + +fn schema() -> AspectSchema { + AspectSchema::new(PhysicalType::F64, BigDecimal::from(0), TimeUnit::Seconds) +} + +/// A store holding one sealed frame of three points, closed at the end of the call. +/// +/// The store is dropped on purpose: a frame read while it is still open can be served from a +/// cache, and the whole point is to make the decoder look at damaged bytes. +async fn seal_one_frame(dir: &Path) { + let store = SegmentStore::open(dir).await.expect("opens the store"); + store.declare("price", &schema()).await.expect("declares the aspect"); + store.seal("price", &schema(), &[10, 20, 30], &[BigDecimal::from(1), BigDecimal::from(2), BigDecimal::from(3)]).await.expect("seals the frame"); + drop(store); +} + +/// The single `.weftseg` the store wrote. +fn only_frame(dir: &Path) -> PathBuf { + let segments = dir.join("segments"); + let mut names = fs::read_dir(&segments).expect("the segment directory exists").map(|entry| entry.expect("a directory entry").path()).filter(|path| path.extension().is_some_and(|ext| ext == "weftseg")).collect::>(); + names.sort(); + assert_eq!(names.len(), 1, "one seal writes one frame, found {names:?}"); + names.remove(0) +} + +/// Re-write `path` with `bytes`, re-stamping nothing: callers that change the body recompute the +/// trailer themselves so that each case says which checksum it expects to pass. +fn write_frame(path: &Path, bytes: &[u8]) { + fs::write(path, bytes).expect("writes the frame back"); +} + +/// `GET` the range that covers the sealed points, and return the status plus the JSON body. +async fn read_range(dir: &Path) -> (StatusCode, serde_json::Value) { + let store = SegmentStore::open(dir).await.expect("re-opens the store"); + let router = app_with_state(AppState::new().with_store(Arc::new(store))); + let response = router.oneshot(Request::builder().method("GET").uri("/api/v1/storage/price/points?start=0&end=100").body(Body::empty()).expect("builds the request")).await.expect("answers"); + let status = response.status(); + let bytes = axum::body::to_bytes(response.into_body(), usize::MAX).await.expect("reads the body"); + (status, serde_json::from_slice(&bytes).expect("a JSON body")) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_frame_that_fails_its_checksum_answers_500_frame_undecodable() { + let dir = tempfile::TempDir::new().expect("tempdir"); + seal_one_frame(dir.path()).await; + let frame = only_frame(dir.path()); + + // Flip a byte in the middle of the body and leave the trailer alone: the checksum does not + // match, which is the plain `integrity` case. + let mut bytes = fs::read(&frame).expect("reads the frame"); + let middle = bytes.len() / 2; + bytes[middle] ^= 0xff; + write_frame(&frame, &bytes); + + let (status, body) = read_range(dir.path()).await; + assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR, "damaged data on disk is a server fault, not a bad request: {body}"); + assert_eq!(body["code"], "frame_undecodable", "{body}"); + assert_eq!(body["details"]["fault"], "integrity", "the fault class is what the caller branches on: {body}"); + assert_eq!(body["error"], "the server could not decode data it has on disk", "the sentence is X-ERR's fixed one, not the decoder's: {body}"); + // The decoder's own words are kept for the log, and the envelope's `error` field is not the + // place they leak: this asserts the sanitisation rule holds on this path too. + assert!(!body["error"].as_str().unwrap_or_default().contains("weftseg"), "the path does not belong in the body: {body}"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_frame_from_a_newer_build_answers_503_incompatible_store() { + let dir = tempfile::TempDir::new().expect("tempdir"); + seal_one_frame(dir.path()).await; + let frame = only_frame(dir.path()); + + // Claim a version this build has never heard of, and re-stamp the trailer so the checksum + // passes: the only thing wrong with the file is that it is newer than the server. The frame + // header is `MAGIC | u16 version`, and the trailer is `crc32` over everything before it, so + // both have to move together or the test would only be proving the case above. + let mut bytes = fs::read(&frame).expect("reads the frame"); + let newer = FRAME_VERSION_MAX_KNOWN + 1; + bytes[7..9].copy_from_slice(&newer.to_le_bytes()); + let end = bytes.len() - 4; + let stamped = crc32(&bytes[..end]).to_le_bytes(); + bytes[end..].copy_from_slice(&stamped); + write_frame(&frame, &bytes); + + let (status, body) = read_range(dir.path()).await; + assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE, "a store this binary cannot read is a store/version condition: {body}"); + assert_eq!(body["code"], "incompatible_store", "{body}"); + assert_eq!(body["details"]["fault"], "newer_format", "the class says the binary is what is behind: {body}"); +} diff --git a/weftdb/src/types/frame_read.rs b/weftdb/src/types/frame_read.rs new file mode 100644 index 00000000..29d4bfaf --- /dev/null +++ b/weftdb/src/types/frame_read.rs @@ -0,0 +1,266 @@ +//! Reading a frame by what the **frame** says it is, and keeping its fault class +//! attached to the failure (1.0 plan X-FRAME). +//! +//! Every frame-decoding call site in [`crate::SegmentStore`] used to choose its reader +//! from `descriptor.format_version`, a u16 copied into the index row when the frame was +//! sealed. The frame registry (5 = single-block, 6 = paged, 7 = envelope) makes the file +//! the authority on its own layout, and the index copy is at best a cache of what the +//! writer decided years ago. It is also the least protected number in the path: +//! `types/segment_index.rs` reads it out of a damaged row with `unwrap_or(0)`, which +//! quietly routes a paged frame to the single-block reader and reports a corrupt file as +//! a version problem. +//! +//! So the layout is probed from the frame's own header — magic plus the version fields, +//! nothing else — and the reader that gets chosen verifies every byte after that. A +//! header that has been damaged still fails, and fails on the frame's own checksum, +//! which is the honest answer whichever way the routing guess went. +//! +//! What travels out of here is deliberately two things, not one: +//! +//! - the **operator-visible sentence**, byte-for-byte what these sites have always +//! produced (`decoding paged segment : `). Corrupt-frame triage +//! is done by reading logs, and a slice that changes how a fault is *classified* has no +//! business also changing what an operator greps for. When the header cannot even say +//! which layout it is, the sentence uses the unmarked word `segment`, which is what the +//! not-paged case has always said. +//! - a **typed fault** underneath it ([`FrameFaultError`]), carrying the fault class the +//! plan's C5 policy is written against, so `weft-server` can answer with the status the +//! condition deserves rather than deriving one from the sentence — which is the rule +//! X-ERR established: read the type under the `anyhow` context, not the context's words. + +use std::fmt; + +use anyhow::Result; +use bigdecimal::BigDecimal; +use weft_physical_type::{frame_fault_of, frame_layout_of, read_frame_paged_segment, read_frame_point, read_frame_points, read_frame_range, read_frame_range_physical, read_frame_segment, FrameFault, FrameLayout, PagedSegment, PhysicalValue, Segment, WeftSegError}; + +/// A frame that this build could not decode, tagged with why in the vocabulary the +/// storage policy is written against. +/// +/// The class, not the decoder's phrasing, is what the storage layer's behaviour and the +/// HTTP status follow: +/// +/// - [`FrameFault::Integrity`] — damage: a checksum that does not match, a length that +/// does not add up, an envelope that contradicts the frame inside it. The file is +/// corrupt and stays corrupt in every build this one could be updated to. +/// - [`FrameFault::Undecodable`] — this build is the problem, not the file. Either it is +/// older than the frame (`newer_format`, which is a reason to refuse the store rather +/// than to quarantine a file), or the frame needs something this build does not +/// implement, such as a critical extension or a codec behind an off-by-default feature. +/// +/// It sits under an `anyhow::Context` whose sentence is the historical one, so nothing +/// that reads the message changes while the type becomes available to anyone downcasting. +#[derive(Debug)] +pub struct FrameFaultError { + /// The C5 fault class this failure belongs to. + fault: FrameFault, + /// The decoder's own complaint, kept so the log line and `source()` stay informative. + source: WeftSegError, +} + +impl FrameFaultError { + /// The class a caller acts on. + #[must_use] + pub fn fault(&self) -> FrameFault { + self.fault + } + + /// The decoder's own complaint. + #[must_use] + pub fn source(&self) -> &WeftSegError { + &self.source + } +} + +impl fmt::Display for FrameFaultError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + // The class leads, because this is the error a caller reaches for when the + // surrounding context has already been read past. + write!(formatter, "{} frame could not be decoded: {}", self.fault.fault_class(), self.source) + } +} + +impl std::error::Error for FrameFaultError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + Some(&self.source) + } +} + +/// A whole frame, decoded, in whichever layout the frame itself declared. +#[derive(Debug)] +pub(crate) enum FullFrame { + /// Layout 5. + SingleBlock(Segment), + /// Layout 6. + Paged(PagedSegment), +} + +impl FullFrame { + /// Rows per page, for the index columns that only mean something to a paged frame. + pub(crate) fn rows_per_page(&self) -> Option { + match self { + Self::SingleBlock(_) => None, + Self::Paged(segment) => Some(segment.rows_per_page), + } + } + + /// Every row in the frame, nulls kept. + pub(crate) fn decode_nullable(self) -> (Vec, Vec>) { + match self { + Self::SingleBlock(segment) => segment.decode_nullable(), + Self::Paged(segment) => segment.decode_nullable(), + } + } +} + +/// Turns a decoder failure into the chain the store has always returned, with the typed +/// fault added underneath it. +fn decode_error(error: WeftSegError, path: &str, paged: bool) -> anyhow::Error { + // Kept before `error` moves: the historical sentence is the point of this function. + let shown = error.to_string(); + let fault = frame_fault_of(&error); + let word = if paged { "paged segment" } else { "segment" }; + anyhow::Error::new(FrameFaultError { fault, source: error }).context(format!("decoding {word} {path}: {shown}")) +} + +/// Whether the frame calls itself paged, for the sentence in its error path. +fn frame_is_paged(bytes: &[u8]) -> bool { + matches!(frame_layout_of(bytes), Ok(FrameLayout::Paged)) +} + +/// Decode a frame and return the physical (encoded) values in a time range. +pub(crate) fn read_range_physical(path: &str, bytes: &[u8], start: i64, end: i64) -> Result<(Vec, Vec>)> { + let paged = frame_is_paged(bytes); + read_frame_range_physical(bytes, start, end).map_err(|error| decode_error(error, path, paged)) +} + +/// Decode a frame and return the values in a time range. +pub(crate) fn read_range(path: &str, bytes: &[u8], start: i64, end: i64) -> Result<(Vec, Vec>), anyhow::Error> { + let paged = frame_is_paged(bytes); + read_frame_range(bytes, start, end).map_err(|error| decode_error(error, path, paged)) +} + +/// Decode a frame and look up one timestamp. +pub(crate) fn read_point(path: &str, bytes: &[u8], t: i64) -> Result, anyhow::Error> { + let paged = frame_is_paged(bytes); + read_frame_point(bytes, t).map_err(|error| decode_error(error, path, paged)) +} + +/// Decode a frame and look up several timestamps. +pub(crate) fn read_points(path: &str, bytes: &[u8], timestamps: &[i64]) -> Result>, anyhow::Error> { + let paged = frame_is_paged(bytes); + read_frame_points(bytes, timestamps).map_err(|error| decode_error(error, path, paged)) +} + +/// Decode a whole frame, keeping its layout so the caller can report rows per page. +pub(crate) fn read_full(path: &str, bytes: &[u8]) -> Result { + match frame_layout_of(bytes) { + Ok(FrameLayout::Paged) => read_frame_paged_segment(bytes).map(FullFrame::Paged).map_err(|error| decode_error(error, path, true)), + Ok(FrameLayout::SingleBlock) => read_frame_segment(bytes).map(FullFrame::SingleBlock).map_err(|error| decode_error(error, path, false)), + // The header could not name a layout this build knows. There is no paged/non-paged + // answer to make, and the unmarked word is the historical default. + Err(error) => Err(decode_error(error, path, false)), + } +} + +#[cfg(test)] +mod tests { + use bigdecimal::BigDecimal; + use weft_physical_type::{crc32, wrap_frame_envelope, write_paged_segment, FrameExtension, FrameFault, PagedSegment, TimeUnit, FRAME_VERSION_MAX_KNOWN}; + + use super::*; + + /// A ten-row paged frame in the layout most stores actually use. + fn paged_frame() -> PagedSegment { + let timestamps: Vec = (0..10).collect(); + let values: Vec = (0..10).map(BigDecimal::from).collect(); + PagedSegment::build(×tamps, &values, TimeUnit::Nanos, &BigDecimal::from(0), 4).expect("ten rows in pages of four build") + } + + /// The bytes of that frame, unwrapped. + fn paged_bytes() -> Vec { + write_paged_segment(&paged_frame()) + } + + /// Rewrite a frame's own version field and re-stamp its trailer, so a reader reaches + /// the version check instead of tripping the checksum on the way to it. + fn stamp_version(bytes: &mut [u8], version: u16) { + // `MAGIC` is 7 bytes and the version is the u16 straight after it; the trailer is + // the last 4 bytes and covers everything before it. + bytes[7..9].copy_from_slice(&version.to_le_bytes()); + let body = bytes.len() - 4; + let checksum = crc32(&bytes[..body]); + bytes[body..].copy_from_slice(&checksum.to_le_bytes()); + } + + #[test] + fn a_frame_is_read_from_what_the_file_says() { + let body = paged_bytes(); + let wrapped = wrap_frame_envelope(&body, &[]).expect("an envelope around a real frame"); + let frame = read_full("price-1.weftseg", &wrapped).expect("the envelope's own header says which reader to use"); + assert_eq!(frame.rows_per_page(), Some(4), "the wrapped frame is still the paged one"); + let (timestamps, values) = frame.decode_nullable(); + assert_eq!(timestamps.len(), 10); + assert!(values.iter().all(Option::is_some), "no row was lost to the envelope"); + } + + #[test] + fn damage_keeps_the_operator_sentence_and_adds_a_fault_class() { + let mut bytes = paged_bytes(); + // One byte in the middle: away from the magic, away from the trailer, so the + // checksum is what complains. + let middle = bytes.len() / 2; + bytes[middle] ^= 0xff; + let error = read_range("/var/lib/weft/segments/price-1.weftseg", &bytes, i64::MIN, i64::MAX).expect_err("a damaged frame does not decode"); + let fault = error.downcast_ref::().expect("the fault class travels under the context"); + assert!(matches!(fault.fault(), FrameFault::Integrity), "damage is damage: {fault:?}"); + // The sentence an operator greps for is the one these sites have always printed. + assert!(error.to_string().starts_with("decoding paged segment /var/lib/weft/segments/price-1.weftseg: "), "{error}"); + } + + #[test] + fn a_frame_written_by_a_newer_build_is_the_binary_being_behind() { + let mut bytes = paged_bytes(); + stamp_version(&mut bytes, FRAME_VERSION_MAX_KNOWN + 1); + let error = read_range("price-1.weftseg", &bytes, i64::MIN, i64::MAX).expect_err("this build has no reader for that version"); + let fault = error.downcast_ref::().expect("the fault class travels under the context"); + assert!(matches!(fault.fault(), FrameFault::Undecodable { newer_format: true }), "a newer frame is not a corrupt file: {fault:?}"); + } + + #[test] + fn an_envelope_wrapping_a_body_this_build_cannot_read_is_still_newer_format() { + // The case the envelope exists for: a later server wraps a body this build has no + // reader for. `read_full` must not then double-check the CRC of a frame it has + // already refused, which would report the same file as damaged. + let mut envelope = wrap_frame_envelope(&paged_bytes(), &[]).expect("an envelope around a real frame"); + let outer = envelope.len() - 4; + let inner = envelope[1..outer].windows(7).position(|w| w == b"WEFTSEG").expect("an envelope wraps a frame") + 1; + let bumped = FRAME_VERSION_MAX_KNOWN + 1; + envelope[7..9].copy_from_slice(&bumped.to_le_bytes()); + envelope[inner + 7..inner + 9].copy_from_slice(&bumped.to_le_bytes()); + let inner_crc = crc32(&envelope[inner..outer]); + envelope[outer - 4..outer].copy_from_slice(&inner_crc.to_le_bytes()); + let outer_crc = crc32(&envelope[..outer]); + envelope[outer..].copy_from_slice(&outer_crc.to_le_bytes()); + let error = read_full("price-9.weftseg", &envelope).expect_err("the body layout has no reader here"); + let fault = error.downcast_ref::().expect("the fault class travels under the context"); + assert!(matches!(fault.fault(), FrameFault::Undecodable { newer_format: true }), "{fault:?}"); + } + + #[test] + fn a_critical_extension_this_build_lacks_is_a_refusal_not_damage() { + // The frame is intact and this build is simply not allowed to serve it: decoding + // the body without the extension would answer with data the writer never meant. + let wrapped = wrap_frame_envelope(&paged_bytes(), &[FrameExtension { id: 9001, critical: true, payload: vec![7] }]).expect("the envelope does not require a number this build knows"); + let error = read_full("price-1.weftseg", &wrapped).expect_err("a critical extension this build cannot apply stops the read"); + let fault = error.downcast_ref::().expect("the fault class travels under the context"); + assert!(matches!(fault.fault(), FrameFault::Undecodable { newer_format: false }), "{fault:?}"); + } + + #[test] + fn a_non_critical_extension_is_ignored_and_the_frame_reads() { + let wrapped = wrap_frame_envelope(&paged_bytes(), &[FrameExtension { id: 9002, critical: false, payload: vec![7, 7] }]).expect("an extension this build does not know may still be written"); + let frame = read_full("price-1.weftseg", &wrapped).expect("an optional extension does not stop a read"); + assert_eq!(frame.decode_nullable().0.len(), 10); + } +} diff --git a/weftdb/src/types/mod.rs b/weftdb/src/types/mod.rs index 8dd44fa2..05d46115 100644 --- a/weftdb/src/types/mod.rs +++ b/weftdb/src/types/mod.rs @@ -16,6 +16,7 @@ pub use dataset::{Dataset, DatasetId}; pub use dictionary::{Dictionary, DictionaryConstraints, DictionaryId, DictionaryMetadata, Steps, Variability, VariablilityType}; pub use dictionary_name::InvalidDictionaryName; pub use event::{Event, EventID, EventName, Events, Manifestation, ManifestationId}; +pub use frame_read::FrameFaultError; pub use input_measurement::InputMeasurement; pub use measurement::{Measurement, MeasurementId}; pub use measurement_vector::MeasurementVector; @@ -55,6 +56,7 @@ pub mod error; pub mod event; pub mod exec; pub(crate) mod frame_name; +pub(crate) mod frame_read; pub(crate) mod index_txn; pub mod input_measurement; pub mod interpolate_range; diff --git a/weftdb/src/types/segment_store.rs b/weftdb/src/types/segment_store.rs index 8bb99383..bb0016e6 100644 --- a/weftdb/src/types/segment_store.rs +++ b/weftdb/src/types/segment_store.rs @@ -84,14 +84,14 @@ use anyhow::{bail, Context, Result}; use bigdecimal::BigDecimal; use chrono::{DateTime, Utc}; use splimes::{Point, Resolution}; -use weft_physical_type::{merge_newer_wins, split_index, AspectSchema, FrameOptions, PagedSegment, PhysicalValue, Segment, SegmentDescriptor, SplitDecision, SplitPolicy, TimeUnit, PAGED_SEGMENT_FORMAT_VERSION}; +use weft_physical_type::{merge_newer_wins, split_index, AspectSchema, FrameOptions, PagedSegment, PhysicalValue, Segment, SegmentDescriptor, SplitDecision, SplitPolicy, TimeUnit}; use weft_reduce::{Aggregation, Bucket, PartialReduction}; use crate::{ aspect_name, types::{ aspect_locks::{AspectLocks, AspectState, CommitGuard, MaintGuard}, durable::{ create_dir_all_durable, fault::{self, FaultPoint}, poison_global, write_new_durable, LockHolder, RealFs, RootLock, RootLockError, StoreFs, SyncPolicy, WritePoints - }, frame_name, index_txn::{IndexOp, IndexRow, IndexTxn, IndexTxnError, TxnApplied, TxnErrorKind, TxnPoints}, migrations, reaper::{LiveFrames, ReadPin, Reaper}, store_format::{self, StoreFormat, StoreScope} + }, frame_name, frame_read, index_txn::{IndexOp, IndexRow, IndexTxn, IndexTxnError, TxnApplied, TxnErrorKind, TxnPoints}, migrations, reaper::{LiveFrames, ReadPin, Reaper}, store_format::{self, StoreFormat, StoreScope} }, AspectCatalog, AspectMetadata, AspectMetadataStore, CatalogStore, PartialSidecar, PartialSidecarPolicy, SegmentIndexStore, SnapshotReport, StoreError, CONTROL_PLANE_FILES, SIDECAR_AGGREGATIONS, SUPPORTED_LAYOUT }; @@ -289,7 +289,7 @@ fn scaled_rows(ts: &[i64], vs: &[Option], start: i64, end: i64, u fn segment_partial(descriptor: &SegmentDescriptor, bytes: &[u8], start: i64, end: i64, unit: TimeUnit, resolution: Resolution, aggregations: &[Aggregation]) -> Result> { // Read the window in its physical encoding, so a ScaledI64 column can be reduced on its // stored mantissas without building a BigDecimal per row. - let (ts, vs) = if descriptor.format_version == PAGED_SEGMENT_FORMAT_VERSION { weft_physical_type::weftseg::read_paged_segment_range_physical(bytes, start, end).map_err(|e| anyhow::anyhow!("decoding paged segment {}: {e}", descriptor.path))? } else { weft_physical_type::weftseg::read_segment_range_physical(bytes, start, end).map_err(|e| anyhow::anyhow!("decoding segment {}: {e}", descriptor.path))? }; + let (ts, vs) = frame_read::read_range_physical(descriptor.path.as_str(), bytes, start, end)?; // Integer-native path: every present in-window value is ScaledI64 at one scale and every // requested reduction is a streaming one. `reduce_partial_scaled` then builds the same // mergeable partial `reduce_partial` would, from integer accumulators. @@ -1921,16 +1921,10 @@ impl SegmentStore { bail!("segment {} does not end in the CRC its row was committed with ({expected:#010x})", descriptor.path); } } - if descriptor.format_version == PAGED_SEGMENT_FORMAT_VERSION { - let segment = PagedSegment::read_from(&bytes).map_err(|e| anyhow::anyhow!("decoding paged segment {}: {e}", descriptor.path))?; - let rows_per_page = segment.rows_per_page; - let (timestamps, values) = segment.decode_nullable(); - Ok(DecodedFrame { rows_per_page: Some(rows_per_page), timestamps, values }) - } else { - let segment = Segment::read_from(&bytes).map_err(|e| anyhow::anyhow!("decoding segment {}: {e}", descriptor.path))?; - let (timestamps, values) = segment.decode_nullable(); - Ok(DecodedFrame { rows_per_page: None, timestamps, values }) - } + let frame = frame_read::read_full(descriptor.path.as_str(), &bytes)?; + let rows_per_page = frame.rows_per_page(); + let (timestamps, values) = frame.decode_nullable(); + Ok(DecodedFrame { rows_per_page, timestamps, values }) } /// Read the frames of `members` (ascending by id), checking each is the frame its row @@ -3054,8 +3048,9 @@ impl SegmentStore { /// /// # Errors /// - /// Propagates a libSQL prune failure, a filesystem read error, or a - /// [`weft_physical_type::weftseg::WeftSegError`] for a corrupt/unreadable `.weftseg`. + /// Propagates a libSQL prune failure, a filesystem read error, or a frame-decode + /// failure carrying a [`crate::FrameFaultError`] — the C5 fault class (`integrity`, + /// `newer_format` or `undecodable`) for a corrupt or unreadable `.weftseg`. pub async fn read_time_range(&self, aspect: &str, start: i64, end: i64) -> Result<(Vec, Vec>)> { aspect_name::validate(aspect)?; let _pin = self.pin_reads(); @@ -3070,7 +3065,7 @@ impl SegmentStore { // index range + per-present-row value read) — the paged variant additionally skips whole // pages disjoint from the window — instead of the whole segment; any other shape falls // back to a full decode. Already filtered to `[start, end]`. - let (ts, vs) = if descriptor.format_version == PAGED_SEGMENT_FORMAT_VERSION { weft_physical_type::weftseg::read_paged_segment_range(&bytes, start, end).map_err(|e| anyhow::anyhow!("decoding paged segment {}: {e}", descriptor.path))? } else { weft_physical_type::weftseg::read_segment_range(&bytes, start, end).map_err(|e| anyhow::anyhow!("decoding segment {}: {e}", descriptor.path))? }; + let (ts, vs) = frame_read::read_range(descriptor.path.as_str(), &bytes, start, end)?; for (t, v) in ts.into_iter().zip(vs) { if start <= t && t <= end { timestamps.push(t); @@ -3193,8 +3188,9 @@ impl SegmentStore { /// /// # Errors /// - /// Propagates a libSQL prune failure, a filesystem read error, or a - /// [`weft_physical_type::weftseg::WeftSegError`] for a corrupt/unreadable `.weftseg`. + /// Propagates a libSQL prune failure, a filesystem read error, or a frame-decode + /// failure carrying a [`crate::FrameFaultError`] — the C5 fault class (`integrity`, + /// `newer_format` or `undecodable`) for a corrupt or unreadable `.weftseg`. pub async fn read_point(&self, aspect: &str, t: i64) -> Result> { aspect_name::validate(aspect)?; let _pin = self.pin_reads(); @@ -3205,7 +3201,7 @@ impl SegmentStore { // Streaming single-value read: prunes/skips the pages and value-column blocks a point // lookup does not touch, unpacking only the one block covering `t` on a per-block codec // (roadmap Phase 4/6). Equal to `…read_from(&bytes)?.value_at(t)` for every frame. - let hit = if descriptor.format_version == PAGED_SEGMENT_FORMAT_VERSION { weft_physical_type::weftseg::read_paged_segment_point(&bytes, t).map_err(|e| anyhow::anyhow!("decoding paged segment {}: {e}", descriptor.path))? } else { weft_physical_type::weftseg::read_segment_point(&bytes, t).map_err(|e| anyhow::anyhow!("decoding segment {}: {e}", descriptor.path))? }; + let hit = frame_read::read_point(descriptor.path.as_str(), &bytes, t)?; if hit.is_some() { found = hit; } @@ -3228,8 +3224,9 @@ impl SegmentStore { /// /// # Errors /// - /// Propagates a libSQL prune failure, a filesystem read error, or a - /// [`weft_physical_type::weftseg::WeftSegError`] for a corrupt/unreadable `.weftseg`. + /// Propagates a libSQL prune failure, a filesystem read error, or a frame-decode + /// failure carrying a [`crate::FrameFaultError`] — the C5 fault class (`integrity`, + /// `newer_format` or `undecodable`) for a corrupt or unreadable `.weftseg`. pub async fn read_points(&self, aspect: &str, ts: &[i64]) -> Result>> { aspect_name::validate(aspect)?; let _pin = self.pin_reads(); @@ -3242,7 +3239,7 @@ impl SegmentStore { let mut found = vec![None; ts.len()]; for descriptor in &descriptors { let bytes = self.read_frame(descriptor).await?; - let hits = if descriptor.format_version == PAGED_SEGMENT_FORMAT_VERSION { weft_physical_type::weftseg::read_paged_segment_points(&bytes, ts).map_err(|e| anyhow::anyhow!("decoding paged segment {}: {e}", descriptor.path))? } else { weft_physical_type::weftseg::read_segment_points(&bytes, ts).map_err(|e| anyhow::anyhow!("decoding segment {}: {e}", descriptor.path))? }; + let hits = frame_read::read_points(descriptor.path.as_str(), &bytes, ts)?; // Later (higher seal-id) segments override earlier ones per instant — last-writer-wins. for (slot, hit) in found.iter_mut().zip(hits) { if hit.is_some() { @@ -4239,8 +4236,9 @@ impl SegmentStore { /// /// # Errors /// - /// Propagates a libSQL read failure, a filesystem read error, or a - /// [`weft_physical_type::weftseg::WeftSegError`] for a corrupt/unreadable `.weftseg`. + /// Propagates a libSQL read failure, a filesystem read error, or a frame-decode + /// failure carrying a [`crate::FrameFaultError`] — the C5 fault class (`integrity`, + /// `newer_format` or `undecodable`) for a corrupt or unreadable `.weftseg`. pub async fn read_value_range(&self, aspect: &str, lo: &BigDecimal, hi: &BigDecimal) -> Result<(Vec, Vec)> { aspect_name::validate(aspect)?; let _pin = self.pin_reads(); @@ -4265,13 +4263,7 @@ impl SegmentStore { /// rows aligned `(timestamps, values)` with `None` at every null row. async fn decode_all(&self, descriptor: &SegmentDescriptor) -> Result<(Vec, Vec>)> { let bytes = self.read_frame(descriptor).await?; - if descriptor.format_version == PAGED_SEGMENT_FORMAT_VERSION { - let segment = PagedSegment::read_from(&bytes).map_err(|e| anyhow::anyhow!("decoding paged segment {}: {e}", descriptor.path))?; - Ok(segment.decode_nullable()) - } else { - let segment = Segment::read_from(&bytes).map_err(|e| anyhow::anyhow!("decoding segment {}: {e}", descriptor.path))?; - Ok(segment.decode_nullable()) - } + Ok(frame_read::read_full(descriptor.path.as_str(), &bytes)?.decode_nullable()) } /// Number of sealed segments recorded for `aspect`. @@ -4692,7 +4684,7 @@ mod tests { use serial_test::serial; use tempfile::TempDir; - use weft_physical_type::{timestamp::TimeUnit, PhysicalType}; + use weft_physical_type::{timestamp::TimeUnit, PhysicalType, PAGED_SEGMENT_FORMAT_VERSION}; use super::*; use crate::{types::index_txn::TxnPoints, STORE_FORMAT_FILE};