obj_core/codec/

mod.rs

//! Document codec (L4) — per-document header + postcard payload.
//!
//! See `docs/format.md` § Document records for the authoritative
//! byte-by-byte specification. This module is the reference
//! implementation.
//!
//! The codec sits between the typed `Document` API (this module) and
//! the B+tree (L3): every stored value in a collection's primary
//! B-tree is the byte string produced by [`encode`], and every
//! `Db::get` decodes through [`decode`]. The catalog (L5) supplies
//! the `collection_id` that the per-document header pins; the
//! catalog itself stores `CollectionDescriptor`s through this same
//! codec (M5 issue #38).
//!
//! # Power-of-ten posture
//!
//! - **Rule 5.** Encode and decode are runtime boundaries: every
//!   header field is validated explicitly before any `postcard` call
//!   touches caller-controlled bytes. CRC32C, collection-id, and
//!   version-range checks are the three layers of defense in
//!   [`decode`].
//! - **Rule 7.** No `unwrap` / `expect` on any error-bearing path;
//!   postcard errors propagate via the `?` operator into
//!   [`Error::Codec`].
//! - **Rule 9.** Hot-path dispatch is static — all codec calls are
//!   monomorphised over `T: Document`. No `dyn` anywhere in this
//!   module.

#![forbid(unsafe_code)]

pub mod dynamic;
pub mod header;
pub mod migrate;
pub mod schema;

pub use crate::codec::dynamic::Dynamic;
pub use crate::codec::header::{DocumentHeader, DOC_HEADER_SIZE, MAX_INLINE_DOC};
pub use crate::codec::migrate::Migrate;
pub use crate::codec::schema::{DynamicSchema, EnumVariantSchema, Schema, MAX_SCHEMA_DEPTH};

use crate::error::{Error, Result};
use crate::pager::checksum::crc32c;

use serde::de::DeserializeOwned;
use serde::Serialize;

/// The trait every user document type implements.
///
/// `Document` types are `serde::Serialize + DeserializeOwned` so
/// they round-trip through postcard. Each implementation provides
/// two associated constants:
///
/// - [`COLLECTION`](Document::COLLECTION) — the collection name
///   under which records of this type are stored. The catalog
///   resolves it to a numeric `collection_id` at registration time;
///   the codec takes that id as an argument to [`encode`]/[`decode`].
/// - [`VERSION`](Document::VERSION) — the type's schema version.
///   Stored in every record's header; the decoder routes a stored-
///   version mismatch through [`Document::migrate`].
///
/// `Document` is `'static` so type-erased catalog rows can carry the
/// collection name as `&'static str`. M5 ships hand-impls; M9
/// introduces a `#[derive(obj::Document)]` proc macro.
pub trait Document: Serialize + DeserializeOwned + 'static {
    /// The collection name this document type stores into.
    ///
    /// Must be a stable, application-chosen identifier. The
    /// catalog resolves it to a `collection_id` on first
    /// registration; subsequent opens reuse the existing id.
    const COLLECTION: &'static str;

    /// The schema version of this `Document` implementation.
    ///
    /// Bump on any breaking change (added/removed/renamed fields,
    /// changed semantics). The decoder enforces:
    ///
    /// - `header.type_version < VERSION` → dispatch through
    ///   [`Document::migrate`].
    /// - `header.type_version == VERSION` → decode directly.
    /// - `header.type_version > VERSION` →
    ///   [`Error::SchemaVersionFromFuture`].
    const VERSION: u32;

    /// Transform an older stored record into `Self`.
    ///
    /// The codec calls [`migrate`](Document::migrate) when a stored
    /// record's `type_version` is strictly less than [`VERSION`](Document::VERSION).
    /// `dynamic` is a structured [`Dynamic`] view of the older
    /// record — the codec walks the on-disk payload through the
    /// schema registered for `from_version` (see
    /// [`historical_schemas`](Document::historical_schemas)) and
    /// hands the resulting map-shaped `Dynamic` to this method.
    /// Concrete overrides read the fields they care about with
    /// [`Dynamic::get`](crate::codec::Dynamic::get) /
    /// [`Dynamic::get_str`](crate::codec::Dynamic::get_str) /
    /// [`Dynamic::deserialize`](crate::codec::Dynamic::deserialize)
    /// and construct the target `Self`.
    ///
    /// `from_version` is the on-disk `type_version` — always `<
    /// Self::VERSION` when this is invoked by the codec.
    ///
    /// # Default body
    ///
    /// Returns [`Error::SchemaMigrationNotImplemented`]. Real types
    /// override this method to handle older versions.
    ///
    /// # Errors
    ///
    /// User overrides MAY return any [`Error`] variant. The default
    /// returns [`Error::SchemaMigrationNotImplemented`].
    fn migrate(_dynamic: crate::codec::Dynamic, from_version: u32) -> Result<Self> {
        Err(Error::SchemaMigrationNotImplemented {
            collection: Self::COLLECTION,
            from_version,
            to_version: Self::VERSION,
        })
    }

    /// Schemas for stored records of OLDER `type_version`s than
    /// [`VERSION`](Document::VERSION).
    ///
    /// Returns a list of `(version, schema)` pairs sorted strictly
    /// ascending by version.  The codec consults this list whenever
    /// it observes `header.type_version < Self::VERSION`:
    ///
    /// 1. Look up the matching `version`.
    /// 2. Walk the on-disk payload bytes through
    ///    [`Dynamic::from_postcard_bytes`](crate::codec::Dynamic::from_postcard_bytes)
    ///    using the registered `schema`.
    /// 3. Hand the resulting structured `Dynamic` to
    ///    [`migrate`](Document::migrate) along with the stored
    ///    version.
    ///
    /// A miss (no entry for the stored version) surfaces as
    /// [`Error::SchemaNotRegistered`].
    /// The default body returns an empty list — a `Document` with
    /// no `historical_schemas()` cannot migrate any older payload.
    ///
    /// # Ordering
    ///
    /// The returned slice MUST be sorted strictly ascending by
    /// version. The codec debug-asserts on read; out-of-order
    /// entries are a programming bug.
    ///
    /// # Power-of-ten posture
    ///
    /// - **Rule 9.** Static dispatch — concrete `Document`
    ///   implementations return their own `Vec<(u32, DynamicSchema)>`;
    ///   the codec never reaches for a `dyn Trait`.
    /// - **Rule 7.** Empty default is intentional — a
    ///   `Document` that has never been versioned cannot have
    ///   historical schemas.
    #[must_use]
    fn historical_schemas() -> Vec<(u32, crate::codec::schema::DynamicSchema)> {
        Vec::new()
    }

    /// Declared secondary indexes for this `Document` type.
    ///
    /// Default body returns the empty vector — no indexes. Override
    /// to declare per-collection indexes; the catalog reconciler
    /// (M7 #57) compares this list against the catalog's stored
    /// descriptors on the FIRST `WriteTxn::collection::<T>()` call
    /// per process per collection and:
    ///
    /// - declares specs absent from the catalog,
    /// - flips active descriptors absent from this list to
    ///   `DroppedPending`,
    /// - leaves unchanged matches alone (idempotent).
    ///
    /// The reconciler runs inside the user's WAL transaction so a
    /// rolled-back txn leaves the catalog clean.
    ///
    /// `&self` is intentionally **not** taken — indexes are a
    /// type-level property of the `Document`, not a per-instance
    /// one.
    #[must_use]
    fn indexes() -> Vec<crate::index::IndexSpec> {
        Vec::new()
    }
}

/// Encode `doc` into the on-disk record format.
///
/// Layout: `DocumentHeader` (16 bytes) followed by
/// `postcard::to_allocvec(doc)`. Returns the assembled bytes in a
/// fresh `Vec<u8>`; allocation is unavoidable here because the
/// payload length is not known until postcard runs.
///
/// # Errors
///
/// - [`Error::Codec`] if postcard encoding fails.
/// - [`Error::DocumentTooLarge`] if the resulting record exceeds
///   [`MAX_INLINE_DOC`] — overflow chains for oversize records are
///   deferred to a later format-minor (see `docs/format.md`
///   § Document records).
pub fn encode<T: Document>(doc: &T, collection_id: u32) -> Result<Vec<u8>> {
    let payload = postcard::to_allocvec(doc)?;
    let payload_len = u32::try_from(payload.len()).map_err(|_| Error::DocumentTooLarge {
        len: payload.len(),
        max: MAX_INLINE_DOC,
    })?;
    let payload_crc32c = crc32c(&payload);
    let header = DocumentHeader {
        collection_id,
        type_version: T::VERSION,
        payload_len,
        payload_crc32c,
    };
    let total = DOC_HEADER_SIZE
        .checked_add(payload.len())
        .ok_or(Error::DocumentTooLarge {
            len: usize::MAX,
            max: MAX_INLINE_DOC,
        })?;
    if total > MAX_INLINE_DOC {
        return Err(Error::DocumentTooLarge {
            len: total,
            max: MAX_INLINE_DOC,
        });
    }
    let mut out = Vec::with_capacity(total);
    header.write_to(&mut out);
    out.extend_from_slice(&payload);
    debug_assert_eq!(out.len(), total, "encode: assembled size mismatch");
    Ok(out)
}

/// Decode an on-disk record into a `T: Document` instance.
///
/// Validates the per-document header at the runtime boundary
/// (power-of-ten Rule 5):
///
/// 1. `bytes.len() >= DOC_HEADER_SIZE`.
/// 2. `header.collection_id == expected_collection_id`.
/// 3. `bytes.len() == DOC_HEADER_SIZE + header.payload_len`.
/// 4. CRC32C of the payload matches `header.payload_crc32c`.
/// 5. `header.type_version <= T::VERSION` (no future versions).
/// 6. If `header.type_version < T::VERSION`, the codec consults
///    `T::historical_schemas()` for the matching version, walks
///    the payload through that schema via
///    [`Dynamic::from_postcard_bytes`](crate::codec::Dynamic::from_postcard_bytes),
///    and dispatches into [`Migrate::migrate`]
///    with the structured `Dynamic` (M10 #83). If no schema is
///    registered for `header.type_version`, the codec returns
///    [`Error::SchemaNotRegistered`] without invoking `migrate` —
///    silent fallback hides schema-evolution bugs.
///    Otherwise (versions match) the payload is decoded directly
///    via `postcard::from_bytes::<T>(payload)`.
///
/// # Errors
///
/// - [`Error::Corruption`] with `page_id = 0` on a malformed header
///   or CRC mismatch (the codec does not know the page id; callers
///   that need a specific id should wrap or re-emit).
/// - [`Error::CollectionIdMismatch`] on a collection-id mismatch.
/// - [`Error::SchemaVersionFromFuture`] on a stored record newer
///   than `T::VERSION`.
/// - [`Error::SchemaNotRegistered`] when the stored
///   `type_version` is older than `T::VERSION` and
///   `T::historical_schemas()` has no entry for it.
/// - [`Error::SchemaMigrationNotImplemented`] when the registered
///   `Migrate::migrate` body returns the default error.
/// - [`Error::SchemaTypeMismatch`] / [`Error::SchemaDepthExceeded`]
///   on a schema / payload disagreement.
/// - [`Error::Codec`] on postcard decode failures.
pub fn decode<T: Document>(bytes: &[u8], expected_collection_id: u32) -> Result<T> {
    let header = DocumentHeader::read_from(bytes)?;
    if header.collection_id != expected_collection_id {
        return Err(Error::CollectionIdMismatch {
            expected: expected_collection_id,
            found: header.collection_id,
        });
    }
    let payload_len =
        usize::try_from(header.payload_len).map_err(|_| Error::Corruption { page_id: 0 })?;
    let total = DOC_HEADER_SIZE
        .checked_add(payload_len)
        .ok_or(Error::Corruption { page_id: 0 })?;
    if bytes.len() != total {
        return Err(Error::Corruption { page_id: 0 });
    }
    let payload = &bytes[DOC_HEADER_SIZE..total];
    if crc32c(payload) != header.payload_crc32c {
        return Err(Error::Corruption { page_id: 0 });
    }
    if header.type_version > T::VERSION {
        return Err(Error::SchemaVersionFromFuture {
            collection: T::COLLECTION,
            from: header.type_version,
            to: T::VERSION,
        });
    }
    if header.type_version < T::VERSION {
        // Migration dispatch (#36 → #83). The codec walks the
        // payload bytes through the schema registered for
        // `header.type_version` and hands the resulting structured
        // `Dynamic` to `T::migrate`. Concrete types override
        // `Document::migrate` to perform the actual conversion;
        // the default body errors with
        // `Error::SchemaMigrationNotImplemented`.
        //
        // Power-of-ten Rule 5: every entry in
        // `historical_schemas()` is sorted ascending — assert here
        // so a malformed registry surfaces during development.
        let history = <T as Document>::historical_schemas();
        debug_assert!(
            history.windows(2).all(|w| w[0].0 < w[1].0),
            "historical_schemas() must be strictly ascending by version",
        );
        let schema = history
            .iter()
            .find(|(v, _)| *v == header.type_version)
            .map(|(_, s)| s)
            .ok_or(Error::SchemaNotRegistered {
                collection: T::COLLECTION,
                version: header.type_version,
            })?;
        let dynamic = Dynamic::from_postcard_bytes(payload, schema)?;
        return <T as Migrate>::migrate(dynamic, header.type_version);
    }
    postcard::from_bytes::<T>(payload).map_err(Error::from)
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde::{Deserialize, Serialize};

    #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
    struct TinyDoc {
        a: u32,
        b: String,
    }

    impl Document for TinyDoc {
        const COLLECTION: &'static str = "tiny";
        const VERSION: u32 = 1;
    }

    #[test]
    fn round_trip_small_document() {
        let d = TinyDoc {
            a: 42,
            b: "hello".to_owned(),
        };
        let bytes = encode(&d, 7).expect("encode");
        let back: TinyDoc = decode(&bytes, 7).expect("decode");
        assert_eq!(back, d);
    }

    #[test]
    fn collection_id_mismatch_errors() {
        let d = TinyDoc {
            a: 1,
            b: "x".to_owned(),
        };
        let bytes = encode(&d, 7).expect("encode");
        let err = decode::<TinyDoc>(&bytes, 9).expect_err("mismatched id");
        assert!(matches!(
            err,
            Error::CollectionIdMismatch {
                expected: 9,
                found: 7
            }
        ));
    }

    #[test]
    fn crc_mismatch_errors() {
        let d = TinyDoc {
            a: 1,
            b: "x".to_owned(),
        };
        let mut bytes = encode(&d, 7).expect("encode");
        bytes[DOC_HEADER_SIZE] ^= 0xFF;
        let err = decode::<TinyDoc>(&bytes, 7).expect_err("crc mismatch");
        assert!(matches!(err, Error::Corruption { page_id: 0 }));
    }

    #[test]
    fn truncated_payload_errors() {
        let d = TinyDoc {
            a: 1,
            b: "x".to_owned(),
        };
        let bytes = encode(&d, 7).expect("encode");
        let truncated = &bytes[..bytes.len() - 1];
        let err = decode::<TinyDoc>(truncated, 7).expect_err("truncated");
        assert!(matches!(err, Error::Corruption { page_id: 0 }));
    }

    #[test]
    fn header_too_short_errors() {
        let bytes = [0u8; DOC_HEADER_SIZE - 1];
        let err = decode::<TinyDoc>(&bytes, 0).expect_err("short header");
        assert!(matches!(err, Error::Corruption { page_id: 0 }));
    }

    #[test]
    fn oversize_document_errors() {
        #[derive(Serialize, Deserialize)]
        struct Big {
            blob: Vec<u8>,
        }
        impl Document for Big {
            const COLLECTION: &'static str = "big";
            const VERSION: u32 = 1;
        }
        let huge: Vec<u8> = vec![0xAB; MAX_INLINE_DOC + 64];
        let big = Big { blob: huge };
        let err = encode(&big, 1).expect_err("oversize");
        match err {
            Error::DocumentTooLarge { len, max } => {
                assert!(len > max, "len {len} should exceed max {max}");
                assert_eq!(max, MAX_INLINE_DOC);
            }
            other => panic!("expected DocumentTooLarge, got {other:?}"),
        }
    }

    #[test]
    fn future_version_errors() {
        let d = TinyDoc {
            a: 1,
            b: "x".to_owned(),
        };
        let mut bytes = encode(&d, 7).expect("encode");
        bytes[4..8].copy_from_slice(&99u32.to_le_bytes());
        let err = decode::<TinyDoc>(&bytes, 7).expect_err("future");
        assert!(matches!(
            err,
            Error::SchemaVersionFromFuture {
                collection: "tiny",
                from: 99,
                to: 1
            }
        ));
    }

    #[test]
    fn missing_schema_errors_schema_not_registered() {
        // Hand-construct a v0 record (TinyDoc::VERSION = 1) so the
        // decoder dispatches into the migration path. TinyDoc has
        // no `historical_schemas()` override, so the codec surfaces
        // SchemaNotRegistered before reaching `migrate`.
        let payload = postcard::to_allocvec(&TinyDoc {
            a: 1,
            b: "x".to_owned(),
        })
        .expect("postcard");
        let payload_crc = crc32c(&payload);
        let header = DocumentHeader {
            collection_id: 7,
            type_version: 0,
            payload_len: u32::try_from(payload.len()).expect("fits u32"),
            payload_crc32c: payload_crc,
        };
        let mut bytes = Vec::with_capacity(DOC_HEADER_SIZE + payload.len());
        header.write_to(&mut bytes);
        bytes.extend_from_slice(&payload);
        let err = decode::<TinyDoc>(&bytes, 7).expect_err("v0 stored, v1 reader");
        assert!(matches!(
            err,
            Error::SchemaNotRegistered {
                collection: "tiny",
                version: 0,
            }
        ));
    }

    // Migration override test — a v2 doc that adds a new field,
    // with a hand-impl Migrate (via the inherent
    // Document::migrate override).
    //
    // M5 #36 had this test using `Dynamic::Bytes(payload)` as the
    // migrate input.  M10 #83 replaces that with the schema-driven
    // structured Dynamic: the codec consults
    // `historical_schemas()` to find the v1 schema and passes a
    // structured `Dynamic::Map` to the override.
    #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
    struct EvolvingV1 {
        a: u32,
    }

    impl Document for EvolvingV1 {
        const COLLECTION: &'static str = "evolving";
        const VERSION: u32 = 1;
    }

    #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
    struct EvolvingV2 {
        a: u32,
        b: String,
    }

    impl Document for EvolvingV2 {
        const COLLECTION: &'static str = "evolving";
        const VERSION: u32 = 2;

        fn historical_schemas() -> Vec<(u32, crate::codec::schema::DynamicSchema)> {
            vec![(
                1,
                crate::codec::schema::DynamicSchema::map([(
                    "a",
                    crate::codec::schema::DynamicSchema::U64,
                )]),
            )]
        }

        fn migrate(dynamic: crate::codec::Dynamic, from_version: u32) -> Result<Self> {
            if from_version != 1 {
                return Err(Error::SchemaMigrationNotImplemented {
                    collection: Self::COLLECTION,
                    from_version,
                    to_version: Self::VERSION,
                });
            }
            let a = match dynamic.get("a") {
                Some(crate::codec::Dynamic::U64(n)) => {
                    u32::try_from(*n).map_err(|_| Error::SchemaMigrationNotImplemented {
                        collection: Self::COLLECTION,
                        from_version,
                        to_version: Self::VERSION,
                    })?
                }
                _ => {
                    return Err(Error::SchemaMigrationNotImplemented {
                        collection: Self::COLLECTION,
                        from_version,
                        to_version: Self::VERSION,
                    });
                }
            };
            Ok(EvolvingV2 {
                a,
                b: "<default>".to_owned(),
            })
        }
    }

    #[test]
    fn migrate_override_lifts_v1_to_v2() {
        // 1. Encode a v1 record. We can't use `encode<EvolvingV1>`
        //    directly because EvolvingV2 occupies the "evolving"
        //    collection at compile time; instead we hand-build the
        //    record at type_version = 1 by re-using the codec's
        //    header layout.
        let v1 = EvolvingV1 { a: 99 };
        let payload = postcard::to_allocvec(&v1).expect("postcard");
        let header = DocumentHeader {
            collection_id: 13,
            type_version: 1,
            payload_len: u32::try_from(payload.len()).expect("fits u32"),
            payload_crc32c: crc32c(&payload),
        };
        let mut record = Vec::with_capacity(DOC_HEADER_SIZE + payload.len());
        header.write_to(&mut record);
        record.extend_from_slice(&payload);

        // 2. Decode as v2 — the codec spots type_version < VERSION
        //    and dispatches through EvolvingV2::migrate after
        //    walking the v1 schema.
        let decoded: EvolvingV2 = decode(&record, 13).expect("migrate succeeds");
        assert_eq!(
            decoded,
            EvolvingV2 {
                a: 99,
                b: "<default>".to_owned(),
            }
        );
    }

    #[test]
    fn current_version_does_not_route_through_migrate() {
        // Sanity check: a v2 record (matching EvolvingV2::VERSION) is
        // decoded directly by postcard, NOT through `migrate`.
        let v2 = EvolvingV2 {
            a: 7,
            b: "in-band".to_owned(),
        };
        let bytes = encode(&v2, 13).expect("encode");
        let back: EvolvingV2 = decode(&bytes, 13).expect("decode");
        assert_eq!(back, v2);
    }

    // M10 #83: unregistered historical version → SchemaNotRegistered.
    #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
    struct UnregisteredV2 {
        a: u32,
    }

    impl Document for UnregisteredV2 {
        // Distinct collection name so the catalog dispatch does not
        // collide with EvolvingV2.
        const COLLECTION: &'static str = "unregistered";
        const VERSION: u32 = 2;

        // No historical_schemas() override — the default empty Vec
        // applies.  Decoding a v1 record must surface
        // SchemaNotRegistered.
        fn migrate(_dynamic: crate::codec::Dynamic, _from_version: u32) -> Result<Self> {
            // Never reached — the codec errors before dispatch.
            unimplemented!("not reached when no schema is registered")
        }
    }

    #[test]
    fn missing_history_entry_errors_schema_not_registered() {
        let payload = postcard::to_allocvec(&UnregisteredV2 { a: 1 }).expect("postcard");
        let header = DocumentHeader {
            collection_id: 17,
            type_version: 1,
            payload_len: u32::try_from(payload.len()).expect("fits u32"),
            payload_crc32c: crc32c(&payload),
        };
        let mut record = Vec::with_capacity(DOC_HEADER_SIZE + payload.len());
        header.write_to(&mut record);
        record.extend_from_slice(&payload);
        let err = decode::<UnregisteredV2>(&record, 17).expect_err("unregistered");
        assert!(matches!(
            err,
            Error::SchemaNotRegistered {
                collection: "unregistered",
                version: 1,
            }
        ));
    }
}