armdb 0.12.0

//! On-disk shape of one transaction-log record.
//!
//! ```text
//! record  := header || ops || marker
//! header  := magic b"ATXL" | format_version u16 | flags u16 | tx_id u64
//!            | op_count u32 | payload_len u32        // payload_len covers `ops` only
//! op      := collection_id u32 | name_len u16 | name | schema_fingerprint u64
//!            | key_len u32 | key | kind u8 | value_len u32 | value | version_tag
//! version_tag := tag u8 | 8 bytes (Bitcask: gsn u64; Fixed: slot_id u32 | target_meta u32)
//! marker  := magic b"ATXC" | crc32 u32 over header||ops
//! ```
//!
//! All integers are little-endian.

use crate::error::{DbError, DbResult};

pub(crate) const MAGIC_HEADER: [u8; 4] = *b"ATXL";
pub(crate) const MAGIC_MARKER: [u8; 4] = *b"ATXC";
pub(crate) const FORMAT_VERSION: u16 = 1;
pub(crate) const HEADER_LEN: usize = 24;
pub(crate) const MARKER_LEN: usize = 8;
pub(crate) const MAX_KEY_LEN: usize = 64 * 1024;
pub(crate) const MAX_RECORD_BYTES: usize = 8 * 1024 * 1024;
/// Bytes an AES-256-GCM seal adds over its plaintext.
pub(crate) const SEAL_TAG_LEN: usize = 16;
/// Smallest encoded operation: empty name/key/value plus a 9-byte version tag.
const MIN_OP_BYTES: usize = 4 + 2 + 8 + 4 + 1 + 4 + 9;

const TAG_BITCASK: u8 = 0;
const TAG_FIXED: u8 = 1;
const KIND_PUT: u8 = 0;
const KIND_DELETE: u8 = 1;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum OpKind {
    Put,
    Delete,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum VersionTag {
    /// Bitcask: the GSN reserved for this operation before the record was logged.
    Bitcask { gsn: u64 },
    /// FixedStore: the reserved slot and the packed meta the slot must end at.
    Fixed { slot_id: u32, target_meta: u32 },
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct TxOp {
    pub collection_id: u32,
    pub name: String,
    pub schema_fingerprint: u64,
    pub key: Vec<u8>,
    pub kind: OpKind,
    pub value: Vec<u8>,
    pub version: VersionTag,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct TxRecord {
    pub tx_id: u64,
    pub ops: Vec<TxOp>,
}

/// Outcome of decoding at one offset.
#[derive(Debug)]
pub(crate) enum Decoded {
    Record {
        record: TxRecord,
        len: usize,
    },
    /// Incomplete bytes or a failed CRC — legal only at the tail of the newest
    /// segment. The caller decides whether that position is legal.
    Torn,
}

/// Header flag: the payload is a sealed AES-256-GCM envelope.
pub(crate) const FLAG_ENCRYPTED: u16 = 1;

pub(crate) fn encode_record(rec: &TxRecord, out: &mut Vec<u8>) -> DbResult<()> {
    encode_record_with(rec, None, out)
}

pub(crate) fn encode_record_with(
    rec: &TxRecord,
    cipher: Option<&crate::txlog::LogCipher>,
    out: &mut Vec<u8>,
) -> DbResult<()> {
    // Ops first, into their own buffer: the header's `payload_len` must describe
    // what actually follows it (sealed or not), and it is also the AAD.
    let mut payload = Vec::new();
    for op in &rec.ops {
        encode_op(op, &mut payload)?;
    }
    let flags = if cipher.is_some() {
        FLAG_ENCRYPTED
    } else {
        0u16
    };

    let start = out.len();
    out.extend_from_slice(&MAGIC_HEADER);
    out.extend_from_slice(&FORMAT_VERSION.to_le_bytes());
    out.extend_from_slice(&flags.to_le_bytes());
    out.extend_from_slice(&rec.tx_id.to_le_bytes());
    let op_count = u32::try_from(rec.ops.len())
        .map_err(|_| DbError::Client("transaction has more than u32::MAX operations"))?;
    out.extend_from_slice(&op_count.to_le_bytes());

    // A sealed payload is the plaintext plus the 16-byte GCM tag. `TAG_LEN` is
    // named here rather than measured after sealing because `payload_len` is
    // inside the AAD: it must be final before the seal runs.
    let payload_len_value = payload.len() + if cipher.is_some() { SEAL_TAG_LEN } else { 0 };
    let payload_len = u32::try_from(payload_len_value)
        .map_err(|_| DbError::Client("transaction payload exceeds u32::MAX bytes"))?;
    out.extend_from_slice(&payload_len.to_le_bytes());

    match cipher {
        Some(c) => {
            let aad = out[start..start + HEADER_LEN].to_vec();
            let sealed = c.seal(rec.tx_id, &aad, &payload)?;
            debug_assert_eq!(sealed.len(), payload_len as usize);
            out.extend_from_slice(&sealed);
        }
        None => out.extend_from_slice(&payload),
    }

    let mut hasher = crc32fast::Hasher::new();
    hasher.update(&out[start..]);
    let crc = hasher.finalize();
    out.extend_from_slice(&MAGIC_MARKER);
    out.extend_from_slice(&crc.to_le_bytes());

    if out.len() - start > MAX_RECORD_BYTES {
        return Err(DbError::Client("transaction record exceeds max_bytes"));
    }
    Ok(())
}

fn encode_op(op: &TxOp, out: &mut Vec<u8>) -> DbResult<()> {
    if op.key.len() > MAX_KEY_LEN {
        return Err(DbError::Client("transaction key exceeds 64 KiB"));
    }
    let name_len = u16::try_from(op.name.len())
        .map_err(|_| DbError::Client("collection name exceeds u16::MAX bytes"))?;
    let value_len = u32::try_from(op.value.len())
        .map_err(|_| DbError::Client("transaction value exceeds u32::MAX bytes"))?;

    out.extend_from_slice(&op.collection_id.to_le_bytes());
    out.extend_from_slice(&name_len.to_le_bytes());
    out.extend_from_slice(op.name.as_bytes());
    out.extend_from_slice(&op.schema_fingerprint.to_le_bytes());
    out.extend_from_slice(&(op.key.len() as u32).to_le_bytes());
    out.extend_from_slice(&op.key);
    out.push(match op.kind {
        OpKind::Put => KIND_PUT,
        OpKind::Delete => KIND_DELETE,
    });
    out.extend_from_slice(&value_len.to_le_bytes());
    out.extend_from_slice(&op.value);
    match op.version {
        VersionTag::Bitcask { gsn } => {
            out.push(TAG_BITCASK);
            out.extend_from_slice(&gsn.to_le_bytes());
        }
        VersionTag::Fixed {
            slot_id,
            target_meta,
        } => {
            out.push(TAG_FIXED);
            out.extend_from_slice(&slot_id.to_le_bytes());
            out.extend_from_slice(&target_meta.to_le_bytes());
        }
    }
    Ok(())
}

pub(crate) fn decode_record(bytes: &[u8]) -> DbResult<Decoded> {
    decode_record_with(bytes, None)
}

pub(crate) fn decode_record_with(
    bytes: &[u8],
    cipher: Option<&crate::txlog::LogCipher>,
) -> DbResult<Decoded> {
    if bytes.len() < HEADER_LEN || bytes[0..4] != MAGIC_HEADER {
        return Ok(Decoded::Torn);
    }
    let payload_len = u32::from_le_bytes(le4(bytes, 20)) as usize;
    if payload_len > MAX_RECORD_BYTES {
        return Ok(Decoded::Torn);
    }
    let total = HEADER_LEN + payload_len + MARKER_LEN;
    if bytes.len() < total {
        return Ok(Decoded::Torn);
    }
    let marker_at = HEADER_LEN + payload_len;
    if bytes[marker_at..marker_at + 4] != MAGIC_MARKER {
        return Ok(Decoded::Torn);
    }
    let stored_crc = u32::from_le_bytes(le4(bytes, marker_at + 4));
    let mut hasher = crc32fast::Hasher::new();
    hasher.update(&bytes[..marker_at]);
    if hasher.finalize() != stored_crc {
        return Ok(Decoded::Torn);
    }

    // Only a CRC-valid record earns a fail-closed verdict: garbage in a torn
    // tail must not be mistaken for a record from a future format.
    let version = u16::from_le_bytes([bytes[4], bytes[5]]);
    if version != FORMAT_VERSION {
        return Err(DbError::FormatMismatch(format!(
            "txlog record format version {version}; this build writes {FORMAT_VERSION}"
        )));
    }
    let flags = u16::from_le_bytes([bytes[6], bytes[7]]);
    let known = FLAG_ENCRYPTED;
    if flags & !known != 0 {
        return Err(DbError::FormatMismatch(format!(
            "txlog record carries unknown flags {flags:#06x}"
        )));
    }

    // `tx_id` and the op-count sanity check are read exactly as part 1 wrote
    // them (`record.rs`), because both live in the clear: `tx_id` is the nonce
    // and the count guards the `with_capacity` below.
    let tx_id = u64::from_le_bytes(le8(bytes, 8));
    let op_count = u32::from_le_bytes(le4(bytes, 16)) as usize;
    if op_count
        .checked_mul(MIN_OP_BYTES)
        .is_none_or(|need| need > payload_len)
    {
        return Err(DbError::FormatMismatch(
            "txlog record op_count exceeds what the payload can hold".into(),
        ));
    }

    let payload: std::borrow::Cow<'_, [u8]> = if flags & FLAG_ENCRYPTED != 0 {
        // Without the `encryption` feature `cipher` is statically `None` (the
        // type is uninhabited), so this is the arm a feature-free build takes on
        // an encrypted database — a fail-closed refusal, never a silent skip.
        let Some(c) = cipher else {
            #[cfg(feature = "encryption")]
            {
                return Err(DbError::EncryptionError(format!(
                    "txlog record {tx_id} is encrypted but no log_encryption_key is configured \
                     (or this build lacks the `encryption` feature)"
                )));
            }
            #[cfg(not(feature = "encryption"))]
            {
                return Err(DbError::FormatMismatch(format!(
                    "txlog record {tx_id} is encrypted but this build lacks the \
                     `encryption` feature"
                )));
            }
        };
        std::borrow::Cow::Owned(c.open(
            tx_id,
            &bytes[..HEADER_LEN],
            &bytes[HEADER_LEN..marker_at],
        )?)
    } else {
        std::borrow::Cow::Borrowed(&bytes[HEADER_LEN..marker_at])
    };

    let mut cursor = 0usize;
    let mut ops = Vec::with_capacity(op_count);
    for _ in 0..op_count {
        let (op, used) = decode_op(&payload[cursor..])?;
        cursor += used;
        ops.push(op);
    }
    if cursor != payload.len() {
        return Err(DbError::FormatMismatch(
            "txlog record payload length disagrees with its operations".into(),
        ));
    }
    Ok(Decoded::Record {
        record: TxRecord { tx_id, ops },
        len: total,
    })
}

fn decode_op(b: &[u8]) -> DbResult<(TxOp, usize)> {
    let bad = || DbError::FormatMismatch("txlog operation is truncated".into());
    let mut at = 0usize;
    let need = |at: usize, n: usize, b: &[u8]| -> DbResult<()> {
        if at + n > b.len() {
            Err(DbError::FormatMismatch(
                "txlog operation is truncated".into(),
            ))
        } else {
            Ok(())
        }
    };

    need(at, 4, b)?;
    let collection_id = u32::from_le_bytes(le4(b, at));
    at += 4;
    need(at, 2, b)?;
    let name_len = u16::from_le_bytes([b[at], b[at + 1]]) as usize;
    at += 2;
    need(at, name_len, b)?;
    let name = String::from_utf8(b[at..at + name_len].to_vec())
        .map_err(|_| DbError::FormatMismatch("txlog collection name is not UTF-8".into()))?;
    at += name_len;
    need(at, 8, b)?;
    let schema_fingerprint = u64::from_le_bytes(le8(b, at));
    at += 8;
    need(at, 4, b)?;
    let key_len = u32::from_le_bytes(le4(b, at)) as usize;
    at += 4;
    if key_len > MAX_KEY_LEN {
        return Err(DbError::FormatMismatch("txlog key exceeds 64 KiB".into()));
    }
    need(at, key_len, b)?;
    let key = b[at..at + key_len].to_vec();
    at += key_len;
    need(at, 1, b)?;
    let kind = match b[at] {
        KIND_PUT => OpKind::Put,
        KIND_DELETE => OpKind::Delete,
        other => {
            return Err(DbError::FormatMismatch(format!(
                "txlog operation kind {other} is unknown"
            )));
        }
    };
    at += 1;
    need(at, 4, b)?;
    let value_len = u32::from_le_bytes(le4(b, at)) as usize;
    at += 4;
    need(at, value_len, b)?;
    let value = b[at..at + value_len].to_vec();
    at += value_len;
    need(at, 9, b).map_err(|_| bad())?;
    let version = match b[at] {
        TAG_BITCASK => VersionTag::Bitcask {
            gsn: u64::from_le_bytes(le8(b, at + 1)),
        },
        TAG_FIXED => VersionTag::Fixed {
            slot_id: u32::from_le_bytes(le4(b, at + 1)),
            target_meta: u32::from_le_bytes(le4(b, at + 5)),
        },
        other => {
            return Err(DbError::FormatMismatch(format!(
                "txlog version tag {other} is unknown"
            )));
        }
    };
    at += 9;

    Ok((
        TxOp {
            collection_id,
            name,
            schema_fingerprint,
            key,
            kind,
            value,
            version,
        },
        at,
    ))
}

#[inline]
fn le4(b: &[u8], at: usize) -> [u8; 4] {
    [b[at], b[at + 1], b[at + 2], b[at + 3]]
}

#[inline]
fn le8(b: &[u8], at: usize) -> [u8; 8] {
    [
        b[at],
        b[at + 1],
        b[at + 2],
        b[at + 3],
        b[at + 4],
        b[at + 5],
        b[at + 6],
        b[at + 7],
    ]
}

#[cfg(test)]
mod tests {
    use super::*;

    fn sample() -> TxRecord {
        TxRecord {
            tx_id: 7,
            ops: vec![
                TxOp {
                    collection_id: 3,
                    name: "users".to_string(),
                    schema_fingerprint: 0xDEAD_BEEF,
                    key: vec![1, 2, 3, 4],
                    kind: OpKind::Put,
                    value: vec![9; 33],
                    version: VersionTag::Bitcask { gsn: 4096 },
                },
                TxOp {
                    collection_id: 4,
                    name: "counters".to_string(),
                    schema_fingerprint: 1,
                    key: vec![0xFF; 8],
                    kind: OpKind::Delete,
                    value: Vec::new(),
                    version: VersionTag::Fixed {
                        slot_id: 12,
                        target_meta: 0x0000_0405,
                    },
                },
            ],
        }
    }

    #[test]
    fn txlog_record_round_trip() {
        let rec = sample();
        let mut buf = Vec::new();
        encode_record(&rec, &mut buf).unwrap();

        match decode_record(&buf).unwrap() {
            Decoded::Record { record, len } => {
                assert_eq!(record, rec);
                assert_eq!(len, buf.len(), "len must cover header+payload+marker");
            }
            Decoded::Torn => panic!("a freshly encoded record must not decode as torn"),
        }
    }

    #[test]
    fn txlog_record_decodes_from_a_longer_buffer() {
        // The scanner hands the decoder the whole remaining segment.
        let rec = sample();
        let mut buf = Vec::new();
        encode_record(&rec, &mut buf).unwrap();
        let one = buf.len();
        encode_record(&rec, &mut buf).unwrap();

        match decode_record(&buf).unwrap() {
            Decoded::Record { len, .. } => assert_eq!(len, one),
            Decoded::Torn => panic!("first record must decode"),
        }
    }

    #[test]
    fn txlog_record_truncated_bytes_are_torn_not_an_error() {
        let mut buf = Vec::new();
        encode_record(&sample(), &mut buf).unwrap();
        for cut in [0, 1, HEADER_LEN, buf.len() - MARKER_LEN, buf.len() - 1] {
            match decode_record(&buf[..cut]) {
                Ok(Decoded::Torn) => {}
                Ok(Decoded::Record { .. }) => panic!("cut {cut} must not decode"),
                Err(e) => panic!("cut {cut} must be torn, got {e}"),
            }
        }
    }

    #[test]
    fn txlog_record_crc_mismatch_is_torn() {
        let mut buf = Vec::new();
        encode_record(&sample(), &mut buf).unwrap();
        buf[HEADER_LEN + 2] ^= 0xFF; // flip a payload byte, leave the CRC alone
        assert!(matches!(decode_record(&buf), Ok(Decoded::Torn)));
    }

    #[test]
    fn txlog_record_unknown_version_fails_closed() {
        let mut buf = Vec::new();
        encode_record(&sample(), &mut buf).unwrap();
        buf[4..6].copy_from_slice(&2u16.to_le_bytes());
        rewrite_crc(&mut buf);
        let err = decode_record(&buf).expect_err("a CRC-valid future record must fail the open");
        assert!(format!("{err}").contains("format version 2"), "{err}");
    }

    #[test]
    fn txlog_record_unknown_flags_fail_closed() {
        let mut buf = Vec::new();
        encode_record(&sample(), &mut buf).unwrap();
        buf[6..8].copy_from_slice(&0x0002u16.to_le_bytes());
        rewrite_crc(&mut buf);
        let err = decode_record(&buf).expect_err("unknown flags must fail the open");
        assert!(format!("{err}").contains("unknown flags"), "{err}");
    }

    #[test]
    fn txlog_record_garbage_with_a_future_version_is_torn() {
        // A torn tail can contain anything, including a plausible version field.
        // Without a valid CRC it must be discarded, not treated as a future format.
        let mut buf = Vec::new();
        encode_record(&sample(), &mut buf).unwrap();
        buf[4..6].copy_from_slice(&99u16.to_le_bytes()); // CRC now stale
        assert!(matches!(decode_record(&buf), Ok(Decoded::Torn)));
    }

    #[test]
    fn txlog_record_rejects_an_oversized_key() {
        let mut rec = sample();
        rec.ops[0].key = vec![0u8; MAX_KEY_LEN + 1];
        let mut buf = Vec::new();
        assert!(encode_record(&rec, &mut buf).is_err());
    }

    #[test]
    fn txlog_record_huge_op_count_fails_closed() {
        // CRC-valid header with an impossible op_count must not allocate first.
        let rec = TxRecord {
            tx_id: 1,
            ops: Vec::new(),
        };
        let mut buf = Vec::new();
        encode_record(&rec, &mut buf).unwrap();
        buf[16..20].copy_from_slice(&u32::MAX.to_le_bytes());
        rewrite_crc(&mut buf);
        let err = decode_record(&buf).expect_err("impossible op_count must fail closed");
        assert!(format!("{err}").contains("op_count"), "{err}");
    }

    /// Recompute the marker CRC after mutating the header — used to build the
    /// "valid record from a future format" cases.
    fn rewrite_crc(buf: &mut [u8]) {
        let marker_at = buf.len() - MARKER_LEN;
        let mut hasher = crc32fast::Hasher::new();
        hasher.update(&buf[..marker_at]);
        let crc = hasher.finalize();
        buf[marker_at + 4..].copy_from_slice(&crc.to_le_bytes());
    }

    #[cfg(feature = "encryption")]
    mod encrypted {
        use super::super::*;
        use crate::txlog::crypt::RecordCipher;

        fn cipher() -> RecordCipher {
            RecordCipher::derive(&[0x77; 32], &[0x88; 16]).unwrap()
        }

        fn record() -> TxRecord {
            TxRecord {
                tx_id: 5,
                ops: vec![TxOp {
                    collection_id: 3,
                    name: "secrets".to_string(),
                    schema_fingerprint: 0xfeed,
                    key: b"k".to_vec(),
                    kind: OpKind::Put,
                    value: b"very secret value".to_vec(),
                    version: VersionTag::Bitcask { gsn: 12 },
                }],
            }
        }

        #[test]
        fn txlog_encrypted_record_round_trips() {
            let c = cipher();
            let mut bytes = Vec::new();
            encode_record_with(&record(), Some(&c), &mut bytes).unwrap();
            assert!(
                bytes.windows(17).all(|w| w != b"very secret value"),
                "the value must not appear in the clear"
            );
            let Decoded::Record { record: got, len } =
                decode_record_with(&bytes, Some(&c)).unwrap()
            else {
                panic!("expected a record");
            };
            assert_eq!(got, record());
            assert_eq!(len, bytes.len());
        }

        #[test]
        fn txlog_encrypted_record_without_a_key_fails_closed() {
            let c = cipher();
            let mut bytes = Vec::new();
            encode_record_with(&record(), Some(&c), &mut bytes).unwrap();
            let err = decode_record_with(&bytes, None).unwrap_err();
            assert!(
                matches!(
                    err,
                    DbError::EncryptionError(_) | DbError::FormatMismatch(_)
                ),
                "{err}"
            );
        }

        #[test]
        fn txlog_encrypted_record_rejects_a_wrong_key() {
            let mut bytes = Vec::new();
            encode_record_with(&record(), Some(&cipher()), &mut bytes).unwrap();
            let other = RecordCipher::derive(&[0x99; 32], &[0x88; 16]).unwrap();
            assert!(decode_record_with(&bytes, Some(&other)).is_err());
        }

        #[test]
        fn txlog_encrypted_record_rejects_a_tampered_payload() {
            let c = cipher();
            let mut bytes = Vec::new();
            encode_record_with(&record(), Some(&c), &mut bytes).unwrap();
            // Flip a payload byte and repair the CRC, so only the GCM tag can
            // catch it — which is exactly the property being claimed.
            bytes[HEADER_LEN] ^= 0xFF;
            let marker_at = bytes.len() - MARKER_LEN;
            let mut h = crc32fast::Hasher::new();
            h.update(&bytes[..marker_at]);
            let crc = h.finalize();
            bytes[marker_at + 4..].copy_from_slice(&crc.to_le_bytes());
            assert!(decode_record_with(&bytes, Some(&c)).is_err());
        }

        #[test]
        fn txlog_encrypted_torn_tail_is_still_torn() {
            let c = cipher();
            let mut bytes = Vec::new();
            encode_record_with(&record(), Some(&c), &mut bytes).unwrap();
            bytes.truncate(bytes.len() - 3);
            assert!(matches!(
                decode_record_with(&bytes, Some(&c)).unwrap(),
                Decoded::Torn
            ));
        }

        #[test]
        fn txlog_plaintext_record_still_reads_with_a_cipher_present() {
            // Turning encryption on must not orphan the records written before it.
            let mut bytes = Vec::new();
            encode_record(&record(), &mut bytes).unwrap();
            let Decoded::Record { record: got, .. } =
                decode_record_with(&bytes, Some(&cipher())).unwrap()
            else {
                panic!("expected a record");
            };
            assert_eq!(got, record());
        }
    }
}