yo-kv 0.3.21

//! The RDB payload that `DUMP` hands out and `RESTORE` takes back.
//!
//! A payload is one value with no key and no deadline, wrapped in ten bytes that
//! say it is intact:
//!
//! ```text
//! +------+------------------+---------+-----------+
//! | type | the object       | version | crc64     |
//! | 1 B  | as many as it is | 2 B LE  | 8 B LE    |
//! +------+------------------+---------+-----------+
//!                                     ^ over everything to its left
//! ```
//!
//! This is not a file format even though it is spelled like one. There is no
//! header, no database selector and no end of file opcode, because the whole
//! point is that it fits in a bulk string. [`crate::snapshot`] is the file, and
//! it is these same bytes with a frame around them. The version is here so that a server
//! reading a payload can refuse one from a newer server rather than misread it,
//! and the checksum is here because `RESTORE` takes bytes from a client and a
//! client is allowed to be wrong.
//!
//! # Two version numbers and not one
//!
//! The version we stamp on a payload and the version we will read are different
//! numbers, because they are answers to opposite questions. What we write is a
//! promise about how old a server can be and still understand us, so it is as
//! low as it can be. What we read is a statement about how new a server can be
//! before a type byte might not mean what it used to, so it is as high as has
//! actually been checked. [`VERSION`] and [`READS_UP_TO`] say which is which.
//!
//! # It has to be Redis's bytes, not ours
//!
//! Nothing here is an internal format we get to choose. `MIGRATE` sends this to
//! another server and `RESTORE` accepts it from any client, so a payload we
//! produce has to load into a real Redis and a payload a real Redis produces has
//! to load here. That is the only reason CRC64 exists in `yo-common`, and it is
//! why the type bytes below are copied from `rdb.h` rather than numbered from
//! zero in the order this file happens to handle them.
//!
//! # Writing the simple shape and reading every shape
//!
//! The two directions are deliberately not symmetric. Reading accepts every
//! encoding a modern Redis emits, because we do not get to pick what arrives.
//! Writing picks the plainest legal type for each kind, a count followed by the
//! elements, because every one of those loads into Redis 8.2 and one shape per
//! kind is one shape to get right.
//!
//! # Copying the blob when there is one
//!
//! That is the shape for values that are stored as a structure. A value that is
//! already sitting in one packed blob does not go through it, because
//! [`crate::listpack`] and [`crate::intset`] are byte compatible with Redis's
//! own on purpose, so the payload for one of those is the blob with a length in
//! front of it. A small set, a small hash and a small sorted set are one memcpy
//! each instead of a walk that decodes every element and encodes it again, and
//! they are the overwhelming majority of what `DUMP` and `MIGRATE` are pointed
//! at.
//!
//! The rule for which type byte a value gets is the same word `OBJECT ENCODING`
//! answers with and not the body underneath it, so a set that calls itself a
//! hashtable is walked even in the corner where its members happen to still be
//! in one intset run. There is one rule and one place to read it.
//!
//! A hash that has been widened for field deadlines is not copied. That band
//! carries a third element per field and keeps it after the last deadline has
//! been taken off, so the blob it holds is not the blob `HASH_LISTPACK` means
//! and the walk is what makes it one.
//!
//! What that is worth, from `benches/rdb.rs` at a hundred elements, walked
//! against copied:
//!
//! ```text
//!   set of text      7.57 us    3.41 us    2.2x
//!   set of integers  1.32 us    0.57 us    2.3x
//!   hash            24.94 us    3.95 us    6.3x
//!   sorted set      21.04 us    3.82 us    5.5x
//! ```
//!
//! The same rows at a thousand elements, which is past every packed band and is
//! therefore the walk in both runs, moved by under one percent, so nothing here
//! was paid for by the values that do not benefit. What is left on the copied
//! rows is a checksum over the payload and one allocation to put it in, and both
//! of those are paid whichever way the payload was built, which is why the hash
//! and the sorted set gain more than the two sets do: their walk was the more
//! expensive one, not their copy the cheaper.
//!
//! The load side pays about five percent for this on a hash and a sorted set,
//! because a listpack entry has to be decoded where a count prefixed element is
//! read straight off a length. Copying the blob is not free on the way back in
//! and the trade is still worth making, since a payload is written once and this
//! is a five percent loss against a five hundred percent gain.
//!
//! The load side also stopped asking a listpack for element `i`. There is no
//! offset table in a listpack, so `get(i)` walks from the front and a loop that
//! asks for every element in turn costs the square of the count. A hundred field
//! hash loaded in 81 us and loads in 60. That was there before any of this and
//! the only thing that ever reached it was a payload from a real server, which
//! is the case that matters most.
//!
//! # Taking the blob back
//!
//! A payload for a sorted set on the packed band goes the other way too. The
//! blob that arrives is the layout that band uses, so it moves in whole rather
//! than being added a member at a time, and the difference is not small: adding
//! costs a scan to see whether the member is already there and a second scan to
//! find where it belongs, both over everything added so far, so it is the square
//! of the count twice over with a memmove on each one. A hundred member sorted
//! set restored in 534 us and restores in 4.6 us.
//!
//! The blob is checked before it is taken. This band answers a rank query by
//! position and by nothing else, so a payload that says it is a sorted set while
//! not being sorted would answer `ZRANGE` with the wrong members and never say
//! why, and a payload with the same member twice would report a length nothing
//! else agrees with. One pass rules out both, since strictly increasing means no
//! two members compare equal on the score and then equal on the bytes. A blob
//! that fails the check, or that is past this server's limits, is handed back
//! and walked, which is what the reader did with every payload before this.
//!
//! A sorted set past the band is sized from the count now, the way a set and a
//! hash already were. It used to start packed whatever the count said, fill to
//! the band limit at a scan a member, and throw the listpack away. A thousand
//! member sorted set restored in 1.23 ms and restores in 88 us.
//!
//! The hash gets the same treatment, and the only hard part was the bit the
//! sorted set got for free. A sorted set blob is ordered, so one pass proving it
//! is strictly increasing also proves no member is in it twice. A hash blob is
//! in insertion order, and a repeated field would give a hash whose `HLEN`
//! counts both rows and whose `HGET` and `HDEL` only ever reach the first, so
//! the length would disagree with `HGETALL` and a delete would leave the field
//! behind. `Hash::from_packed` rules that out by hashing each field into a
//! stack array and sorting it, which is one pass and a sort rather than the
//! square of the count, and a collision costs a fallback to the walk and not a
//! wrong answer. A hundred field hash restored in 62.8 us and restores in 5.8.
//!
//! Only the two element form. The band with a deadline after every value is not
//! handed over, for the same reason `Hash::packed_bytes` will not copy it on
//! the way out: that column has its own type byte and its own header, and a hash
//! that has been widened once keeps the third element per field forever after.
//!
//! # Compression
//!
//! Redis compresses strings over twenty bytes with LZF when `rdbcompression` is
//! on, which it is by default. Nothing here compresses on the way out, because
//! an uncompressed string is legal and every reader accepts it. Decompression on
//! the way in is not optional, because payloads arriving from a real Redis are
//! full of LZF strings.

use std::borrow::Cow;

use yo_common::crc::crc64;
use yo_common::num::{self, DIGITS_MAX};

use crate::hash::{self, Hash};
use crate::intset::Intset;
use crate::keys::{Body, Record};
use crate::list::{self, List};
use crate::listpack::{Entry, Listpack};
use crate::set::{self, Set};
use crate::stream::{Group, Id, Stream};
use crate::zset::{self, Zset};

/// The RDB version this server writes into the footer.
///
/// Redis refuses a payload whose version is above its own, so this being right
/// is the difference between a payload another server will look at and one it
/// throws away without reading. Lower is friendlier, and twelve is as low as
/// this can go: it is the version that introduced the hash with field deadlines,
/// which is a shape this server writes.
pub const VERSION: u16 = 12;

/// The highest version in a footer this server will still read.
///
/// A different number from [`VERSION`], and the two mean opposite things. What
/// we write is a promise about how old a server can be and still understand us.
/// What we read is a statement about how new a server can be before we stop
/// trusting that a type byte still means what it used to.
///
/// Fifteen because that is what a Redis 8.10.1 stamps on a payload, read off one
/// over a socket rather than out of a header file. Refusing it is not a small
/// bug: it means `RESTORE` turns down every payload a current server produces,
/// with a message about the checksum that sends the reader to entirely the wrong
/// place. That is what this constant existing separately is here to stop.
///
/// It goes up when a newer server has been checked and not before. The guard is
/// worth keeping rather than removing, because the day Redis reuses a type byte
/// for a different layout, refusing to read it is the only safe answer and a
/// wrong value is worse than no value.
pub const READS_UP_TO: u16 = 15;

/// The footer: two bytes of version and eight of checksum.
pub(crate) const FOOTER: usize = 10;

// The object type byte. These are `rdb.h`, and the gaps are types this server
// cannot hold, so they are not named.
const T_STRING: u8 = 0;
const T_LIST: u8 = 1;
const T_SET: u8 = 2;
const T_ZSET: u8 = 3;
const T_HASH: u8 = 4;
const T_ZSET_2: u8 = 5;
const T_SET_INTSET: u8 = 11;
const T_STREAM_LISTPACKS: u8 = 15;
const T_HASH_LISTPACK: u8 = 16;
const T_ZSET_LISTPACK: u8 = 17;
const T_LIST_QUICKLIST_2: u8 = 18;
const T_STREAM_LISTPACKS_2: u8 = 19;
const T_SET_LISTPACK: u8 = 20;
const T_STREAM_LISTPACKS_3: u8 = 21;
const T_HASH_METADATA: u8 = 24;
const T_HASH_LISTPACK_EX: u8 = 25;
const T_STREAM_LISTPACKS_4: u8 = 27;

// The length encoding, `00` and `01` in the top two bits for six and fourteen
// bit lengths, then two whole byte forms, and `11` for the special encodings.
const LEN_6BIT: u8 = 0;
const LEN_14BIT: u8 = 1;
const LEN_32BIT: u8 = 0x80;
const LEN_64BIT: u8 = 0x81;
const LEN_ENCODED: u8 = 3;

// What a `11` length means: three integer widths and a compressed blob.
const ENC_INT8: u64 = 0;
const ENC_INT16: u64 = 1;
const ENC_INT32: u64 = 2;
const ENC_LZF: u64 = 3;

/// A quicklist node holding a listpack rather than one long value.
const NODE_PACKED: u64 = 2;
/// A quicklist node that is one value too big for a listpack.
const NODE_PLAIN: u64 = 1;

/// Why a payload was not accepted.
///
/// Two variants because `RESTORE` has two complaints and a client can tell them
/// apart. A bad footer means the bytes were damaged or came from a newer server,
/// and everything else means they were intact and still did not make sense.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Bad {
    /// The version is from the future or the checksum does not match.
    Footer,
    /// The bytes are self consistent and are not a value this server can hold.
    Format,
}

/// Where a load is allowed to put what it builds.
///
/// The four representation thresholds, borrowed rather than copied, because a
/// restore has to land in the same band the same data would have landed in had
/// it been written a command at a time. A hash restored into a listpack on a
/// server configured for tables would answer the wrong thing to `OBJECT
/// ENCODING` and would be a different size for the rest of its life.
#[derive(Debug, Clone, Copy)]
pub struct Limits<'a> {
    /// `set-max-intset-entries` and the two listpack thresholds.
    pub set: &'a set::Limits,
    /// `hash-max-listpack-entries` and `hash-max-listpack-value`.
    pub hash: &'a hash::Limits,
    /// `list-max-listpack-size`, as bytes or as a count.
    pub list: &'a list::Limits,
    /// `zset-max-listpack-entries` and `zset-max-listpack-value`.
    pub zset: &'a zset::Limits,
}

// ---------------------------------------------------------------------------
// The wrapper: version and checksum.
// ---------------------------------------------------------------------------

/// Put the version and the checksum on the end of a serialised object.
fn seal(mut body: Vec<u8>) -> Vec<u8> {
    body.extend_from_slice(&VERSION.to_le_bytes());
    let crc = crc64(0, &body);
    body.extend_from_slice(&crc.to_le_bytes());
    body
}

/// Check the footer and hand back everything in front of it.
///
/// The version check comes before the checksum, which is the order Redis uses
/// and the order that gives the better answer: a payload from a newer server is
/// usually intact, and telling somebody their bytes are corrupt when they are
/// merely from next year sends them looking in the wrong place.
fn unseal(payload: &[u8]) -> Result<&[u8], Bad> {
    if payload.len() < FOOTER {
        return Err(Bad::Footer);
    }
    let split = payload.len() - FOOTER;
    let (body, foot) = payload.split_at(split);
    let version = u16::from_le_bytes([foot[0], foot[1]]);
    if version > READS_UP_TO {
        return Err(Bad::Footer);
    }
    let stored = u64::from_le_bytes(foot[2..].try_into().expect("ten byte footer, eight left"));
    if stored != crc64(0, &payload[..payload.len() - 8]) {
        return Err(Bad::Footer);
    }
    Ok(body)
}

// ---------------------------------------------------------------------------
// Writing.
// ---------------------------------------------------------------------------

/// Serialise a record's value, footer and all.
///
/// The deadline does not go in. `DUMP` deliberately drops it and `RESTORE` takes
/// a fresh one as an argument, because a payload that travels for a while would
/// otherwise arrive already expired or, worse, silently alive for longer than
/// anybody meant.
///
/// `None` for a value with no RDB shape at all, which today is only the sparse
/// array. No command on the wire can create one yet, so no client can reach this,
/// and it is a `None` rather than a panic so that the day the document commands
/// land the answer is a missing key and not a dead server.
pub(crate) fn dump(rec: &Record) -> Option<Vec<u8>> {
    let mut out = Vec::new();
    if !object(rec, None, &mut out) {
        return None;
    }
    Some(seal(out))
}

/// Serialise a record's value with no footer, and with a key if there is one.
///
/// The two callers want the same bytes in two frames. `DUMP` wants the value on
/// its own and passes `None`, and [`crate::snapshot`] wants it as one entry in a
/// file, where the key name sits between the type byte and the value. That is
/// the whole reason the key is threaded down here rather than written by the
/// caller: the type byte comes first and the key comes second, so there is no
/// point either of them could splice a name in without copying the value again.
///
/// `false` for a value with no RDB shape at all, and it is answered before
/// anything has been written, so a caller building a file does not have to undo
/// a half written entry.
pub(crate) fn object(rec: &Record, key: Option<&[u8]>, out: &mut Vec<u8>) -> bool {
    match rec.body() {
        // The same answer a sparse array gets, and for a stronger reason: a
        // foreign body is an engine that lives above this crate and there is no
        // byte shape for it here to write even in principle. `DUMP` on a graph
        // is refused by the dispatch before it reaches this, so nothing sees
        // the null bulk this would otherwise produce.
        //
        // The sparse array is ours and has no Redis number to write under, so
        // there is nothing for it to go out as either.
        Body::Foreign(_) | Body::Array(_) => return false,
        Body::String(bytes) => {
            put_head(out, T_STRING, key);
            put_str(out, bytes);
        }
        Body::List(list) => {
            put_head(out, T_LIST, key);
            put_len(out, list.len() as u64);
            for element in list.iter() {
                put_entry(out, element);
            }
        }
        Body::Set(set) => match (set.encoding(), set.packed_bytes()) {
            (set::Encoding::Intset, Some(blob)) => {
                put_head(out, T_SET_INTSET, key);
                put_str(out, blob);
            }
            (set::Encoding::Listpack, Some(blob)) => {
                put_head(out, T_SET_LISTPACK, key);
                put_str(out, blob);
            }
            _ => {
                put_head(out, T_SET, key);
                put_len(out, set.len() as u64);
                for member in set.iter() {
                    put_entry(out, member);
                }
            }
        },
        Body::Zset(zset) => match zset.packed_bytes() {
            Some(blob) => {
                put_head(out, T_ZSET_LISTPACK, key);
                put_str(out, blob);
            }
            None => {
                put_head(out, T_ZSET_2, key);
                put_len(out, zset.len() as u64);
                zset.walk(0, zset.len(), false, |member, score| {
                    put_entry(out, member);
                    out.extend_from_slice(&score.to_le_bytes());
                });
            }
        },
        Body::Hash(hash) => put_hash(out, hash, key),
        Body::Stream(stream) => put_stream(out, stream, key),
    }
    true
}

/// The type byte, and the key name behind it when this is going into a file.
fn put_head(out: &mut Vec<u8>, ty: u8, key: Option<&[u8]>) {
    out.push(ty);
    if let Some(key) = key {
        put_str(out, key);
    }
}

/// A hash, in the plain shape or the one that carries field deadlines.
///
/// Two types because the deadline costs a length prefixed number on every single
/// field, and the overwhelming majority of hashes have no deadline anywhere.
/// Redis makes the same split for the same reason, and the trick it uses is
/// worth copying: the earliest deadline in the hash goes in the header, and each
/// field stores the difference from it plus one, so a field with no deadline is
/// a zero and everything else is a small number rather than a full timestamp.
fn put_hash(out: &mut Vec<u8>, hash: &Hash, key: Option<&[u8]>) {
    let Some(soonest) = hash.soonest_deadline() else {
        if let Some(blob) = hash.packed_bytes() {
            put_head(out, T_HASH_LISTPACK, key);
            put_str(out, blob);
            return;
        }
        put_head(out, T_HASH, key);
        put_len(out, hash.len() as u64);
        for (field, value) in hash.iter() {
            put_entry(out, field);
            put_entry(out, value);
        }
        return;
    };
    put_head(out, T_HASH_METADATA, key);
    out.extend_from_slice(&soonest.to_le_bytes());
    put_len(out, hash.len() as u64);
    for i in 0..hash.len() {
        let (field, value) = hash.at(i).expect("index is under the length");
        // Saturating rather than subtracting, because `soonest_deadline` is
        // documented as a lower bound and a bound that is early by a millisecond
        // would underflow into a deadline a few hundred million years out.
        let ttl = match hash.deadline_at(i) {
            Some(at) => at.saturating_sub(soonest) + 1,
            None => 0,
        };
        put_len(out, ttl);
        put_entry(out, field);
        put_entry(out, value);
    }
}

/// A stream: the node blobs, then everything the nodes do not say.
///
/// This is where the node layout pays for itself. A node is already the rax key
/// and the listpack a real server writes, so the whole of the entry data goes
/// out as a length and a memcpy each and nothing is decoded on the way. What
/// follows is the counters, and then the consumer groups, which are the part
/// that is genuinely a structure rather than bytes.
///
/// `LISTPACKS_3` and not the newest type. A Redis 8.10 writes 27, which is this
/// with a per group field and an idempotency block on the end, and both are
/// empty for every stream this server can hold because it does not track
/// producer IDs. Writing the older type is the same data in a shape more servers
/// accept, which is the rule the rest of this file already follows.
fn put_stream(out: &mut Vec<u8>, s: &Stream, key: Option<&[u8]>) {
    put_head(out, T_STREAM_LISTPACKS_3, key);
    put_len(out, s.nodes() as u64);
    for (master, blob) in s.raw_nodes() {
        // The sixteen big endian bytes, which is the rax key Redis writes here
        // and is the only place in a payload where an ID is not a length.
        put_str(out, &master.to_bytes());
        put_str(out, blob);
    }
    put_len(out, s.len());
    put_id(out, s.last_id());
    // A stream with nothing left in it has no first entry, and Redis writes 0-0
    // for that rather than leaving the field out.
    put_id(out, s.first_id().unwrap_or(Id::MIN));
    put_id(out, s.max_deleted_id());
    put_len(out, s.added());

    put_len(out, s.groups().count() as u64);
    for (name, group) in s.groups() {
        put_str(out, name);
        put_id(out, group.last_id());
        // Minus one for a count that cannot be worked out, which is what Redis
        // writes for the same thing and reads back as unknown. Checked on
        // 8.10.1: a group over a stream something has been deleted from dumps
        // with eight bytes of ones here and reports a null `entries-read`.
        put_len(out, group.entries_read().unwrap_or(u64::MAX));

        // The whole ledger first and the consumers after it, so a reader meets
        // an entry before it meets whoever is holding it. That is not an
        // accident of the format: it is what lets an entry nobody holds be
        // written at all.
        put_len(out, group.pending_len() as u64);
        for (id, nack) in group.pending_all() {
            out.extend_from_slice(&id.to_bytes());
            put_millis(out, nack.time() as i64);
            put_len(out, nack.count());
        }

        put_len(out, group.consumers().count() as u64);
        for c in group.consumers() {
            put_str(out, c.name());
            put_millis(out, c.seen() as i64);
            // Minus one for a consumer that has never had anything, which is
            // what `XINFO CONSUMERS` on a real server reports for one made by
            // `XGROUP CREATECONSUMER` and never read from.
            put_millis(out, c.active().map_or(-1, |at| at as i64));
            put_len(out, c.len() as u64);
            for id in c.pending() {
                out.extend_from_slice(&id.to_bytes());
            }
        }
    }
}

/// An entry ID, as the two lengths a stream writes everywhere but a rax key.
fn put_id(out: &mut Vec<u8>, id: Id) {
    put_len(out, id.ms);
    put_len(out, id.seq);
}

/// A time, which a stream writes as eight signed little endian bytes.
///
/// Signed because minus one is a value the format uses, for a consumer that has
/// never read anything.
fn put_millis(out: &mut Vec<u8>, at: i64) {
    out.extend_from_slice(&at.to_le_bytes());
}

/// A length, in the smallest of the four forms that holds it.
pub(crate) fn put_len(out: &mut Vec<u8>, n: u64) {
    if n < 1 << 6 {
        out.push((LEN_6BIT << 6) | n as u8);
    } else if n < 1 << 14 {
        out.push((LEN_14BIT << 6) | (n >> 8) as u8);
        out.push(n as u8);
    } else if n <= u64::from(u32::MAX) {
        out.push(LEN_32BIT);
        out.extend_from_slice(&(n as u32).to_be_bytes());
    } else {
        out.push(LEN_64BIT);
        out.extend_from_slice(&n.to_be_bytes());
    }
}

/// A string, integer encoded when that is both possible and shorter.
pub(crate) fn put_str(out: &mut Vec<u8>, s: &[u8]) {
    // Redis only tries the integer encoding on strings short enough to be one,
    // which saves parsing every long value that starts with a digit.
    if s.len() <= 11
        && let Some(n) = num::parse_i64(s)
        && let mut buf = [0u8; DIGITS_MAX]
        && num::i64_digits(&mut buf, n) == s
        && put_int(out, n)
    {
        return;
    }
    put_len(out, s.len() as u64);
    out.extend_from_slice(s);
}

/// An element straight out of a collection.
///
/// A listpack already knows whether it is holding an integer, so an integer
/// element goes out in the integer encoding without ever being formatted into
/// digits and parsed back. That is the same saving the reply path makes and it
/// is why elements come back as an [`Entry`] rather than as bytes.
fn put_entry(out: &mut Vec<u8>, entry: Entry<'_>) {
    match entry {
        Entry::Int(n) => {
            if !put_int(out, n) {
                let mut buf = [0u8; DIGITS_MAX];
                let digits = num::i64_digits(&mut buf, n);
                put_len(out, digits.len() as u64);
                out.extend_from_slice(digits);
            }
        }
        Entry::Str(s) => put_str(out, s),
    }
}

/// An integer in one of the three widths, or `false` if it does not fit any.
///
/// There is no 64 bit form. A number past `i32` goes out as digits, which is
/// what Redis does, and it is not the oversight it looks like: the encoding is
/// there to make short strings shorter and a nineteen digit number in eight
/// bytes saves eleven bytes on a value that is already rare.
fn put_int(out: &mut Vec<u8>, n: i64) -> bool {
    if let Ok(v) = i8::try_from(n) {
        out.push((LEN_ENCODED << 6) | ENC_INT8 as u8);
        out.push(v as u8);
    } else if let Ok(v) = i16::try_from(n) {
        out.push((LEN_ENCODED << 6) | ENC_INT16 as u8);
        out.extend_from_slice(&v.to_le_bytes());
    } else if let Ok(v) = i32::try_from(n) {
        out.push((LEN_ENCODED << 6) | ENC_INT32 as u8);
        out.extend_from_slice(&v.to_le_bytes());
    } else {
        return false;
    }
    true
}

// ---------------------------------------------------------------------------
// Reading.
// ---------------------------------------------------------------------------

/// A position in a payload, and the only thing allowed to advance it.
///
/// Every read goes through here so that a truncated payload is one error at one
/// place rather than a bounds check per field that somebody eventually forgets.
struct Reader<'a> {
    buf: &'a [u8],
    at: usize,
}

impl<'a> Reader<'a> {
    const fn new(buf: &'a [u8]) -> Reader<'a> {
        Reader { buf, at: 0 }
    }

    fn byte(&mut self) -> Result<u8, Bad> {
        let b = *self.buf.get(self.at).ok_or(Bad::Format)?;
        self.at += 1;
        Ok(b)
    }

    fn take(&mut self, n: usize) -> Result<&'a [u8], Bad> {
        let end = self.at.checked_add(n).ok_or(Bad::Format)?;
        let s = self.buf.get(self.at..end).ok_or(Bad::Format)?;
        self.at = end;
        Ok(s)
    }

    /// How many elements follow, refusing a count the payload cannot hold.
    ///
    /// The count in a payload is four bytes wide and the payload is whatever
    /// length it happens to be, so nothing in the format stops one from claiming
    /// two billion members. Every reader that takes a count then hands it to a
    /// `with_hint`, which is the whole point of a hint, and a hint of two billion
    /// asks the allocator for thirty four gigabytes before a single element has
    /// been read. That is not a hypothetical: it is what a `RESTORE` of a
    /// truncated payload did, and on Linux the allocator refused and the process
    /// went down, which turns a bad payload from one client into an outage for
    /// everybody.
    ///
    /// The bound is the bytes that are left. An element takes at least one byte
    /// however it is encoded, so a count past what remains cannot be honest, and
    /// checking it here means every reader gets the check rather than the ones
    /// somebody remembered. It is deliberately loose: it is not trying to work
    /// out the real minimum for each type, only to keep an allocation in the same
    /// order of magnitude as the bytes that arrived.
    ///
    /// Zero is refused with it, for the reason [`non_empty`] gives.
    fn count(&mut self) -> Result<usize, Bad> {
        non_empty(self.bounded()?)
    }

    /// The same bound without the empty rule.
    ///
    /// A stream has four counts that are allowed to be zero and every one of
    /// them is an ordinary state rather than a payload nobody could have
    /// produced. It holds no nodes at all once everything in it has been
    /// deleted, and a real server dumps that and restores it as a live stream of
    /// length zero. It can have no consumer groups, a group can have nothing
    /// pending, and a consumer can be holding nothing.
    fn bounded(&mut self) -> Result<usize, Bad> {
        let n = self.len()?;
        if n > self.buf.len() - self.at {
            return Err(Bad::Format);
        }
        Ok(n)
    }

    /// A length, refusing the `11` forms that are not lengths at all.
    fn len(&mut self) -> Result<usize, Bad> {
        usize::try_from(self.num()?).map_err(|_| Bad::Format)
    }

    /// A number written in the length encoding, which is how a stream writes
    /// every counter it has and both halves of almost every ID.
    ///
    /// Separate from [`Reader::len`] because a stream's numbers are not lengths
    /// and are routinely past what a `usize` has to hold: a millisecond
    /// timestamp is one, and an unknown `entries-read` is written as the whole
    /// sixty four bits set.
    fn num(&mut self) -> Result<u64, Bad> {
        match self.len_or_encoding()? {
            (n, false) => Ok(n),
            (_, true) => Err(Bad::Format),
        }
    }

    /// An entry ID, as the two lengths a stream writes almost everywhere.
    fn id(&mut self) -> Result<Id, Bad> {
        Ok(Id::new(self.num()?, self.num()?))
    }

    /// An entry ID as the sixteen big endian bytes a pending list writes.
    fn raw_id(&mut self) -> Result<Id, Bad> {
        let b = self.take(16)?;
        Ok(Id::from_bytes(b.try_into().expect("sixteen bytes")))
    }

    /// A time, which a stream writes as eight signed little endian bytes.
    fn millis(&mut self) -> Result<i64, Bad> {
        let b = self.take(8)?;
        Ok(i64::from_le_bytes(b.try_into().expect("eight bytes")))
    }

    /// A length, and whether it was one of the special encodings instead.
    fn len_or_encoding(&mut self) -> Result<(u64, bool), Bad> {
        let first = self.byte()?;
        match first >> 6 {
            LEN_6BIT => Ok((u64::from(first & 0x3f), false)),
            LEN_14BIT => {
                let second = self.byte()?;
                Ok(((u64::from(first & 0x3f) << 8) | u64::from(second), false))
            }
            LEN_ENCODED => Ok((u64::from(first & 0x3f), true)),
            // The remaining two bit pattern is `10`, where the whole first byte
            // says which width follows rather than carrying any of the length.
            _ => match first {
                LEN_32BIT => {
                    let b = self.take(4)?;
                    Ok((
                        u64::from(u32::from_be_bytes(b.try_into().expect("four bytes"))),
                        false,
                    ))
                }
                LEN_64BIT => {
                    let b = self.take(8)?;
                    Ok((
                        u64::from_be_bytes(b.try_into().expect("eight bytes")),
                        false,
                    ))
                }
                _ => Err(Bad::Format),
            },
        }
    }

    /// A string, whichever of the five ways it was written.
    ///
    /// Borrowed when the bytes are already there and owned when they had to be
    /// built, which is the integer encodings and LZF. Most strings in a payload
    /// are plain, so most of them cost nothing here.
    fn str(&mut self) -> Result<Cow<'a, [u8]>, Bad> {
        let (n, encoded) = self.len_or_encoding()?;
        if !encoded {
            let n = usize::try_from(n).map_err(|_| Bad::Format)?;
            return Ok(Cow::Borrowed(self.take(n)?));
        }
        let value = match n {
            ENC_INT8 => i64::from(self.byte()? as i8),
            ENC_INT16 => {
                let b = self.take(2)?;
                i64::from(i16::from_le_bytes(b.try_into().expect("two bytes")))
            }
            ENC_INT32 => {
                let b = self.take(4)?;
                i64::from(i32::from_le_bytes(b.try_into().expect("four bytes")))
            }
            ENC_LZF => {
                let packed = self.len()?;
                let plain = self.len()?;
                let bytes = self.take(packed)?;
                return unpack(bytes, plain).map(Cow::Owned).ok_or(Bad::Format);
            }
            _ => return Err(Bad::Format),
        };
        let mut buf = [0u8; DIGITS_MAX];
        Ok(Cow::Owned(num::i64_digits(&mut buf, value).to_vec()))
    }

    /// A score in the binary form, which is `ZSET_2` and everything since.
    fn double(&mut self) -> Result<f64, Bad> {
        let b = self.take(8)?;
        Ok(f64::from_le_bytes(b.try_into().expect("eight bytes")))
    }

    /// A score in the old text form, which only `ZSET` uses.
    ///
    /// A length byte and then that many digits, with three of the lengths
    /// reserved to mean the three values that have no digits.
    fn double_text(&mut self) -> Result<f64, Bad> {
        match self.byte()? {
            255 => Ok(f64::NEG_INFINITY),
            254 => Ok(f64::INFINITY),
            253 => Ok(f64::NAN),
            n => {
                let digits = self.take(n as usize)?;
                num::parse_f64(digits).ok_or(Bad::Format)
            }
        }
    }

    /// Whether every byte has been read, which a well formed payload has.
    const fn done(&self) -> bool {
        self.at == self.buf.len()
    }
}

/// LZF, the one compression Redis puts in an RDB payload.
///
/// A control byte either introduces a run of literals or points backwards into
/// what has already been written. The back reference is allowed to overlap what
/// it is producing, which is how a long run of one byte compresses, so the copy
/// has to go one byte at a time rather than through a slice copy.
///
/// `plain` is the length the payload claims the result will be, and it is used
/// as the bound rather than trusted, so a payload claiming four bytes and
/// describing four gigabytes stops at four.
fn unpack(packed: &[u8], plain: usize) -> Option<Vec<u8>> {
    let mut out = Vec::with_capacity(plain.min(1 << 20));
    let mut i = 0;
    while i < packed.len() {
        let ctrl = usize::from(packed[i]);
        i += 1;
        if ctrl < 32 {
            let run = ctrl + 1;
            let end = i.checked_add(run)?;
            if end > packed.len() || out.len() + run > plain {
                return None;
            }
            out.extend_from_slice(&packed[i..end]);
            i = end;
        } else {
            let mut run = ctrl >> 5;
            if run == 7 {
                run += usize::from(*packed.get(i)?);
                i += 1;
            }
            let back = ((ctrl & 0x1f) << 8) + usize::from(*packed.get(i)?) + 1;
            i += 1;
            let run = run + 2;
            if back > out.len() || out.len() + run > plain {
                return None;
            }
            let from = out.len() - back;
            for at in from..from + run {
                out.push(out[at]);
            }
        }
    }
    (out.len() == plain).then_some(out)
}

/// Turn a payload back into a value.
///
/// `now` is here for one reason: a hash can carry deadlines and a field whose
/// deadline has already gone is not put back. Restoring it would leave a field
/// that the very next read would delete, and a count that is wrong until
/// somebody looks.
pub(crate) fn load(payload: &[u8], limits: Limits<'_>, now: u64) -> Result<Body, Bad> {
    let body = unseal(payload)?;
    let mut r = Reader::new(body);
    let kind = r.byte()?;
    let value = read_object(&mut r, kind, limits, now)?;
    // Trailing bytes mean the payload was not what it said it was, even though
    // everything read so far parsed. Redis is stricter than it looks here and so
    // is this, because a payload with something extra on the end is either a
    // different version's idea of the same type or somebody probing.
    if !r.done() {
        return Err(Bad::Format);
    }
    Ok(value)
}

/// Read one value, leaving the reader wherever that value ended.
///
/// Split out from [`load`] because a payload holds exactly one value and a file
/// holds a great many, so the file reader cannot use the check that there is
/// nothing left over. It is also what says how long a value is: nothing in the
/// format writes that down, and the only way to find the end of one is to read
/// it.
fn read_object(r: &mut Reader<'_>, kind: u8, limits: Limits<'_>, now: u64) -> Result<Body, Bad> {
    Ok(match kind {
        T_STRING => Body::String(r.str()?.into_owned()),
        T_LIST => read_list(r, limits.list)?,
        T_LIST_QUICKLIST_2 => read_quicklist(r, limits.list)?,
        T_SET => read_set(r, limits.set)?,
        T_SET_INTSET => read_intset(r, limits.set)?,
        T_SET_LISTPACK => read_set_listpack(r, limits.set)?,
        T_ZSET | T_ZSET_2 => read_zset(r, limits.zset, kind == T_ZSET_2)?,
        T_ZSET_LISTPACK => read_zset_listpack(r, limits.zset)?,
        T_HASH => read_hash(r, limits.hash)?,
        T_HASH_METADATA => read_hash_metadata(r, limits.hash, now)?,
        T_HASH_LISTPACK => read_hash_listpack(r, limits.hash, false, now)?,
        T_HASH_LISTPACK_EX => read_hash_listpack(r, limits.hash, true, now)?,
        T_STREAM_LISTPACKS | T_STREAM_LISTPACKS_2 | T_STREAM_LISTPACKS_3 | T_STREAM_LISTPACKS_4 => {
            read_stream(r, kind)?
        }
        _ => return Err(Bad::Format),
    })
}

/// How many bytes a value of type `ty` takes at the front of `bytes`.
///
/// Only the tests want this, and what they want it for is to walk a file this
/// crate wrote without trusting this crate's own idea of where each value ends.
/// It is here rather than in [`crate::snapshot`] because [`Reader`] is private
/// and should stay that way.
#[cfg(test)]
pub(crate) fn measure(ty: u8, bytes: &[u8]) -> Option<usize> {
    let (s, h, l, z) = (
        set::Limits::DEFAULT,
        hash::Limits::DEFAULT,
        list::Limits::default(),
        zset::Limits::DEFAULT,
    );
    let limits = Limits {
        set: &s,
        hash: &h,
        list: &l,
        zset: &z,
    };
    let mut r = Reader::new(bytes);
    read_object(&mut r, ty, limits, 0).ok()?;
    Some(r.at)
}

/// An empty collection is not a value, it is a deleted key.
///
/// Redis calls this `emptykey` and refuses the payload rather than creating a
/// key that every command would treat as missing. A zero length collection
/// cannot be produced by any command, so a payload holding one was either
/// hand written or corrupted in a way the checksum happened to survive.
fn non_empty(n: usize) -> Result<usize, Bad> {
    if n == 0 { Err(Bad::Format) } else { Ok(n) }
}

fn read_list(r: &mut Reader<'_>, limits: &list::Limits) -> Result<Body, Bad> {
    let n = r.count()?;
    let mut list = List::new();
    for _ in 0..n {
        list.push_back(&r.str()?, limits);
    }
    Ok(Body::List(list))
}

/// A quicklist, which is a count of nodes and then a blob each.
///
/// A packed node is a whole listpack and a plain node is a single value that was
/// too long to pack, and both of them are written as one string, so the only
/// difference is whether the string is parsed or pushed.
fn read_quicklist(r: &mut Reader<'_>, limits: &list::Limits) -> Result<Body, Bad> {
    let nodes = r.count()?;
    let mut list = List::new();
    for _ in 0..nodes {
        let container = r.len_or_encoding()?.0;
        let blob = r.str()?;
        match container {
            NODE_PLAIN => list.push_back(&blob, limits),
            NODE_PACKED => {
                let lp = Listpack::from_bytes(&blob).map_err(|_| Bad::Format)?;
                let mut buf = [0u8; DIGITS_MAX];
                for entry in lp.iter() {
                    list.push_back(text(entry, &mut buf), limits);
                }
            }
            _ => return Err(Bad::Format),
        }
    }
    if list.is_empty() {
        return Err(Bad::Format);
    }
    Ok(Body::List(list))
}

fn read_set(r: &mut Reader<'_>, limits: &set::Limits) -> Result<Body, Bad> {
    let n = r.count()?;
    let first = r.str()?;
    // The hint and the first member together are what decide the band, so the
    // first member is read before the set is built rather than after.
    let mut set = Set::with_hint(&first, n, limits);
    set.add(&first, limits);
    for _ in 1..n {
        set.add(&r.str()?, limits);
    }
    Ok(Body::Set(set))
}

fn read_intset(r: &mut Reader<'_>, limits: &set::Limits) -> Result<Body, Bad> {
    let blob = r.str()?;
    let ints = Intset::from_bytes(&blob).map_err(|_| Bad::Format)?;
    non_empty(ints.len())?;
    let mut buf = [0u8; DIGITS_MAX];
    let mut set = Set::with_hint(num::i64_digits(&mut buf, ints.at(0)), ints.len(), limits);
    for v in ints.iter() {
        set.add(num::i64_digits(&mut buf, v), limits);
    }
    Ok(Body::Set(set))
}

fn read_set_listpack(r: &mut Reader<'_>, limits: &set::Limits) -> Result<Body, Bad> {
    let blob = r.str()?;
    let lp = Listpack::from_bytes(&blob).map_err(|_| Bad::Format)?;
    non_empty(lp.len())?;
    let mut buf = [0u8; DIGITS_MAX];
    let first = text(lp.get(0).ok_or(Bad::Format)?, &mut buf).to_vec();
    let mut set = Set::with_hint(&first, lp.len(), limits);
    for entry in lp.iter() {
        set.add(text(entry, &mut buf), limits);
    }
    Ok(Body::Set(set))
}

fn read_zset(r: &mut Reader<'_>, limits: &zset::Limits, binary: bool) -> Result<Body, Bad> {
    let n = r.count()?;
    // Sized from the count, the way `read_set` and `read_hash` are. A sorted set
    // that is going to end up on the table should start there, rather than fill
    // the packed band to its limit at a scan a member and then throw it away.
    let mut zset = Zset::with_hint(n, limits);
    for _ in 0..n {
        let member = r.str()?;
        let score = if binary {
            r.double()?
        } else {
            r.double_text()?
        };
        zset.add(&member, score, limits);
    }
    Ok(Body::Zset(zset))
}

fn read_zset_listpack(r: &mut Reader<'_>, limits: &zset::Limits) -> Result<Body, Bad> {
    let blob = r.str()?;
    let lp = Listpack::from_bytes(&blob).map_err(|_| Bad::Format)?;
    if lp.is_empty() || lp.len() % 2 != 0 {
        return Err(Bad::Format);
    }
    // The payload is already the layout the packed band uses, so the fast answer
    // is to take it whole rather than to add a member at a time. `from_packed`
    // hands the blob back when it will not have it, and then the walk below
    // rebuilds it, which is what happens to a payload that is out of order or
    // past this server's limits.
    let lp = match Zset::from_packed(lp, limits) {
        Ok(zset) => return Ok(Body::Zset(zset)),
        Err(lp) => lp,
    };
    let mut zset = Zset::with_hint(lp.len() / 2, limits);
    let mut member = [0u8; DIGITS_MAX];
    let mut score = [0u8; DIGITS_MAX];
    // Walked and not indexed. A listpack has no offset table, so asking it for
    // element `i` costs a walk from the front and asking it for every element in
    // turn costs the square of the count.
    let mut walk = lp.iter();
    while let Some(entry) = walk.next() {
        let name = text(entry, &mut member).to_vec();
        let at = text(walk.next().ok_or(Bad::Format)?, &mut score);
        let at = num::parse_f64(at).ok_or(Bad::Format)?;
        zset.add(&name, at, limits);
    }
    Ok(Body::Zset(zset))
}

fn read_hash(r: &mut Reader<'_>, limits: &hash::Limits) -> Result<Body, Bad> {
    let n = r.count()?;
    let mut hash = Hash::with_hint(n, limits);
    for _ in 0..n {
        let field = r.str()?;
        let value = r.str()?;
        hash.set(&field, &value, limits);
    }
    Ok(Body::Hash(hash))
}

/// A hash with field deadlines, which is a header, a count and then triples.
///
/// The header is the earliest deadline in the hash and each field holds its own
/// distance from it, plus one so that a zero can mean no deadline at all.
fn read_hash_metadata(r: &mut Reader<'_>, limits: &hash::Limits, now: u64) -> Result<Body, Bad> {
    let soonest = u64::from_le_bytes(r.take(8)?.try_into().expect("eight bytes"));
    let n = r.count()?;
    let mut hash = Hash::with_hint(n, limits);
    for _ in 0..n {
        let ttl = r.len_or_encoding()?.0;
        let field = r.str()?;
        let value = r.str()?;
        put_field(
            &mut hash,
            &field,
            &value,
            deadline(soonest, ttl),
            limits,
            now,
        );
    }
    if hash.is_empty() {
        return Err(Bad::Format);
    }
    Ok(Body::Hash(hash))
}

/// A hash packed into one listpack, with or without the deadline column.
fn read_hash_listpack(
    r: &mut Reader<'_>,
    limits: &hash::Limits,
    with_ttl: bool,
    now: u64,
) -> Result<Body, Bad> {
    let soonest = if with_ttl {
        u64::from_le_bytes(r.take(8)?.try_into().expect("eight bytes"))
    } else {
        0
    };
    let blob = r.str()?;
    let lp = Listpack::from_bytes(&blob).map_err(|_| Bad::Format)?;
    let step = if with_ttl { 3 } else { 2 };
    if lp.is_empty() || lp.len() % step != 0 {
        return Err(Bad::Format);
    }
    // The payload is already the layout the packed band uses, so take it whole
    // rather than set a field at a time. Only the two element form: the deadline
    // column is a band this cannot hand over, for the reason `packed_bytes`
    // refuses to copy it on the way out.
    let lp = if with_ttl {
        lp
    } else {
        match Hash::from_packed(lp, limits) {
            Ok(hash) => return Ok(Body::Hash(hash)),
            Err(lp) => lp,
        }
    };
    let mut hash = Hash::with_hint(lp.len() / step, limits);
    let mut field_buf = [0u8; DIGITS_MAX];
    let mut value_buf = [0u8; DIGITS_MAX];
    // Walked and not indexed, for the reason `read_zset_listpack` gives: element
    // `i` of a listpack costs a walk from the front.
    let mut walk = lp.iter();
    while let Some(entry) = walk.next() {
        let field = text(entry, &mut field_buf).to_vec();
        let value = text(walk.next().ok_or(Bad::Format)?, &mut value_buf).to_vec();
        // The packed form holds the deadline as an absolute time, not as a
        // distance from the header, which is the one place the two hash layouts
        // disagree about the same number.
        let at = if with_ttl {
            match walk.next().ok_or(Bad::Format)? {
                Entry::Int(0) => None,
                Entry::Int(n) => Some(u64::try_from(n).map_err(|_| Bad::Format)?),
                Entry::Str(_) => return Err(Bad::Format),
            }
        } else {
            None
        };
        put_field(&mut hash, &field, &value, at, limits, now);
    }
    let _ = soonest;
    if hash.is_empty() {
        return Err(Bad::Format);
    }
    Ok(Body::Hash(hash))
}

/// A stream, in whichever of the four types it arrived as.
///
/// The nodes go straight in, since a payload's listpack is the blob this holds
/// anyway, and the rest of the payload is read into the counters and the groups
/// around them. Everything is checked on the way rather than trusted, because a
/// `RESTORE` takes these bytes from a client.
///
/// The four types are the same format with pieces added on the end of it, so
/// what varies is which fields are there and not where anything is:
///
/// ```text
/// 15  the nodes, the length and the last ID
/// 19  and the first ID, the deleted high water mark, the two read counters
/// 21  and a consumer's active time as well as its seen time
/// 27  and a per group field and a stream wide idempotency block
/// ```
///
/// The three fields 15 does not carry are worked out rather than left empty,
/// which is what Redis does with the same payload. There is nothing above the
/// oldest entry that has been deleted, since a type 15 payload has no way to say
/// there was, and everything that is there was added, since a stream that has
/// been trimmed cannot say so either. Both are the honest reading of a format
/// that predates the question.
fn read_stream(r: &mut Reader<'_>, kind: u8) -> Result<Body, Bad> {
    let edges = kind != T_STREAM_LISTPACKS;
    let active = kind == T_STREAM_LISTPACKS_3 || kind == T_STREAM_LISTPACKS_4;
    let mut s = Stream::new();

    let nodes = r.bounded()?;
    for _ in 0..nodes {
        let key = r.str()?;
        let key: [u8; 16] = key.as_ref().try_into().map_err(|_| Bad::Format)?;
        let blob = r.str()?;
        let lp = Listpack::from_bytes(&blob).map_err(|_| Bad::Format)?;
        // A node with nothing in it has no master entry, so every walk over it
        // would stop on the first read and the stream would be holding bytes
        // that answer nothing. Redis refuses the same node for the same reason.
        if lp.is_empty() || !s.push_raw_node(Id::from_bytes(key), lp) {
            return Err(Bad::Format);
        }
    }

    let length = r.num()?;
    let last = r.id()?;
    let (max_deleted, added) = if edges {
        // The first ID is read and dropped. It is the oldest entry that is still
        // there, which the nodes already say, and taking the payload's word for
        // it would let a wrong one disagree with what a range walk finds.
        let _first = r.id()?;
        (r.id()?, r.num()?)
    } else {
        (Id::MIN, length)
    };
    if !s.set_counters(length, last, max_deleted, added) {
        return Err(Bad::Format);
    }

    let groups = r.bounded()?;
    for _ in 0..groups {
        let name = r.str()?.into_owned();
        let last = r.id()?;
        // Minus one is how the format says the count is not known, and it is not
        // a count that happens to be enormous.
        let read = match edges {
            true => match r.num()? {
                u64::MAX => None,
                n => Some(n),
            },
            false => None,
        };
        let mut group = Group::new(last, read);

        let pending = r.bounded()?;
        for _ in 0..pending {
            let id = r.raw_id()?;
            let time = r.millis()?.max(0) as u64;
            let count = r.num()?;
            if !group.restore_nack(id, time, count) {
                return Err(Bad::Format);
            }
        }

        let consumers = r.bounded()?;
        for _ in 0..consumers {
            let name = r.str()?.into_owned();
            let seen = r.millis()?.max(0) as u64;
            // Before type 21 there was one time and it stood for both, which is
            // what Redis fills in when it reads an older payload.
            let at = if active { r.millis()? } else { seen as i64 };
            let Some(slot) = group.restore_consumer(&name, seen, (at >= 0).then_some(at as u64))
            else {
                return Err(Bad::Format);
            };
            let held = r.bounded()?;
            for _ in 0..held {
                let id = r.raw_id()?;
                if !group.restore_owner(id, slot) {
                    return Err(Bad::Format);
                }
            }
        }
        if kind == T_STREAM_LISTPACKS_4 {
            skip_group_idmp(r)?;
        }
        if !s.push_group(&name, group) {
            return Err(Bad::Format);
        }
    }
    if kind == T_STREAM_LISTPACKS_4 {
        skip_idmp(r)?;
    }
    Ok(Body::Stream(s))
}

/// The group's half of the idempotency block a Redis 8.10 writes.
///
/// One count, and it is zero in every payload a real server has been seen to
/// produce. Read and dropped, for the reason [`skip_idmp`] gives.
fn skip_group_idmp(r: &mut Reader<'_>) -> Result<(), Bad> {
    if r.num()? != 0 {
        return Err(Bad::Format);
    }
    Ok(())
}

/// The idempotency block a Redis 8.10 puts on the end of a stream.
///
/// Producer IDs are a thing this server does not track at all, so what is here
/// is read to find the end of the payload and then dropped. Dropping it is a
/// divergence and is registered as one, and it is a smaller one than it sounds:
/// a payload only ever carries the producer names and never any of the IDs
/// recorded against them, so a real Redis loading its own `DUMP` comes back with
/// `pids-tracked` and `iids-tracked` both at zero as well. Checked on 8.10.1 by
/// recording two IDs under one producer, dumping, restoring the payload under
/// another key and asking.
///
/// The two counts that must be zero are refused rather than skipped when they
/// are not. Nothing has been seen to write one, so what follows a nonzero count
/// is not something that can be read off a server, and guessing it would be
/// inventing a format instead of reading one. Refusing says so out loud, where a
/// skip would quietly turn the rest of the payload into nonsense.
fn skip_idmp(r: &mut Reader<'_>) -> Result<(), Bad> {
    let _duration = r.num()?;
    let _maxsize = r.num()?;
    let producers = r.bounded()?;
    for _ in 0..producers {
        let _name = r.str()?;
        if r.num()? != 0 {
            return Err(Bad::Format);
        }
    }
    let _added = r.num()?;
    let _duplicates = r.num()?;
    Ok(())
}

/// A field's absolute deadline from the header and its stored distance.
const fn deadline(soonest: u64, ttl: u64) -> Option<u64> {
    if ttl == 0 {
        None
    } else {
        Some(soonest + ttl - 1)
    }
}

/// Put one field in, unless its deadline has already gone.
fn put_field(
    hash: &mut Hash,
    field: &[u8],
    value: &[u8],
    at: Option<u64>,
    limits: &hash::Limits,
    now: u64,
) {
    if let Some(at) = at
        && at <= now
    {
        return;
    }
    hash.set(field, value, limits);
    if let Some(at) = at {
        hash.expire(field, at, crate::ttl::Cond::Always, now);
    }
}

/// A listpack entry as bytes, formatting an integer into the caller's buffer.
fn text<'a>(entry: Entry<'a>, buf: &'a mut [u8; DIGITS_MAX]) -> &'a [u8] {
    match entry {
        Entry::Int(n) => num::i64_digits(buf, n),
        Entry::Str(s) => s,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::stream::Retry;
    use crate::ttl::{Ask, Cond};

    fn limits() -> (set::Limits, hash::Limits, list::Limits, zset::Limits) {
        (
            set::Limits::DEFAULT,
            hash::Limits::DEFAULT,
            list::Limits::default(),
            zset::Limits::DEFAULT,
        )
    }

    fn round_trip(body: Body) -> Body {
        let (s, h, l, z) = limits();
        let all = Limits {
            set: &s,
            hash: &h,
            list: &l,
            zset: &z,
        };
        let rec = Record::new(body, None);
        let payload = dump(&rec).expect("this type has an RDB shape");
        load(&payload, all, 0).expect("what we wrote we can read")
    }

    fn set_of(members: &[&[u8]]) -> Set {
        let l = set::Limits::DEFAULT;
        let mut set = Set::with_hint(members[0], members.len(), &l);
        for m in members {
            set.add(m, &l);
        }
        set
    }

    #[test]
    fn a_payload_is_ten_bytes_longer_than_the_object() {
        let rec = Record::new(Body::String(b"hello".to_vec()), None);
        let payload = dump(&rec).expect("a string has an RDB shape");
        // One type byte, one length byte, five of text, and the footer.
        assert_eq!(payload.len(), 1 + 1 + 5 + FOOTER);
        assert_eq!(payload[0], T_STRING);
    }

    #[test]
    fn a_string_that_is_a_number_goes_out_as_one() {
        let rec = Record::new(Body::String(b"1234".to_vec()), None);
        let payload = dump(&rec).expect("a string has an RDB shape");
        // The type byte, the encoding byte, two bytes of integer, and the footer.
        assert_eq!(payload.len(), 1 + 1 + 2 + FOOTER);
        let Body::String(back) = round_trip(Body::String(b"1234".to_vec())) else {
            panic!("a string came back as something else");
        };
        assert_eq!(back, b"1234");
    }

    /// A number with a leading zero is not the same string as the number, so it
    /// has to stay text or the round trip changes the value.
    #[test]
    fn a_string_that_only_looks_like_a_number_stays_text() {
        for s in [&b"007"[..], b"+7", b"-0", b" 7", b"9223372036854775808"] {
            let Body::String(back) = round_trip(Body::String(s.to_vec())) else {
                panic!("a string came back as something else");
            };
            assert_eq!(back, s, "{} did not survive", String::from_utf8_lossy(s));
        }
    }

    #[test]
    fn every_integer_width_survives() {
        for n in [0i64, 1, -1, 127, -128, 128, -129, 32767, -32768, 32768] {
            let mut buf = [0u8; DIGITS_MAX];
            let s = num::i64_digits(&mut buf, n).to_vec();
            let Body::String(back) = round_trip(Body::String(s.clone())) else {
                panic!("a string came back as something else");
            };
            assert_eq!(back, s, "{n} did not survive");
        }
        // Past `i32` there is no encoding, so it goes as digits and still has to
        // come back the same.
        let big = b"2147483648".to_vec();
        let Body::String(back) = round_trip(Body::String(big.clone())) else {
            panic!("a string came back as something else");
        };
        assert_eq!(back, big);
    }

    #[test]
    fn a_set_comes_back_with_the_same_members() {
        let set = set_of(&[b"alpha", b"beta", b"gamma"]);
        let Body::Set(back) = round_trip(Body::Set(set)) else {
            panic!("a set came back as something else");
        };
        assert_eq!(back.len(), 3);
        for m in [&b"alpha"[..], b"beta", b"gamma"] {
            assert!(
                back.contains(m),
                "{} went missing",
                String::from_utf8_lossy(m)
            );
        }
    }

    /// An all integer set is held as an intset here and the round trip has to
    /// land it back in the same band, not in a listpack that happens to hold the
    /// same members.
    #[test]
    fn an_integer_set_comes_back_as_an_integer_set() {
        let set = set_of(&[b"1", b"2", b"3"]);
        let was = set.encoding();
        let Body::Set(back) = round_trip(Body::Set(set)) else {
            panic!("a set came back as something else");
        };
        assert_eq!(back.encoding(), was);
        assert_eq!(back.len(), 3);
        assert!(back.contains(b"2"));
    }

    #[test]
    fn a_list_keeps_its_order() {
        let l = list::Limits::default();
        let mut list = List::new();
        for v in [&b"one"[..], b"two", b"three"] {
            list.push_back(v, &l);
        }
        let Body::List(back) = round_trip(Body::List(list)) else {
            panic!("a list came back as something else");
        };
        let mut seen = Vec::new();
        for e in back.iter() {
            let mut buf = Vec::new();
            e.write_to(&mut buf);
            seen.push(buf);
        }
        assert_eq!(
            seen,
            vec![b"one".to_vec(), b"two".to_vec(), b"three".to_vec()]
        );
    }

    #[test]
    fn a_sorted_set_keeps_its_scores() {
        let l = zset::Limits::DEFAULT;
        let mut zset = Zset::new();
        zset.add(b"a", 1.5, &l);
        zset.add(b"b", -2.0, &l);
        zset.add(b"c", f64::INFINITY, &l);
        let Body::Zset(back) = round_trip(Body::Zset(zset)) else {
            panic!("a sorted set came back as something else");
        };
        assert_eq!(back.score(b"a"), Some(1.5));
        assert_eq!(back.score(b"b"), Some(-2.0));
        assert_eq!(back.score(b"c"), Some(f64::INFINITY));
    }

    #[test]
    fn a_hash_with_no_deadlines_uses_the_plain_type() {
        let l = hash::Limits::DEFAULT;
        let mut hash = Hash::new();
        // Past the packed band, so this is the table and there is no blob to
        // copy. The small case is the listpack one and it is tested below.
        for i in 0..1000 {
            hash.set(format!("f{i}").as_bytes(), b"1", &l);
        }
        assert_eq!(hash.encoding(), hash::Encoding::Hashtable);
        let rec = Record::new(Body::Hash(hash.clone()), None);
        assert_eq!(dump(&rec).expect("a hash has an RDB shape")[0], T_HASH);
        let Body::Hash(back) = round_trip(Body::Hash(hash)) else {
            panic!("a hash came back as something else");
        };
        assert_eq!(back.len(), 1000);
        assert_eq!(back.get(b"f7").map(|v| v.byte_len()), Some(1));
    }

    /// A value that is one packed blob goes out as the blob.
    ///
    /// The type byte is the thing being pinned here. Every one of these round
    /// trips already, through the walk, and the point of the check is that it is
    /// no longer going through the walk.
    #[test]
    fn a_packed_value_goes_out_as_its_blob() {
        let (sl, hl, _, zl) = limits();
        let mut hash = Hash::new();
        hash.set(b"one", b"1", &hl);
        hash.set(b"two", b"2", &hl);
        assert_eq!(hash.encoding(), hash::Encoding::Listpack);

        let mut set = Set::new();
        set.add(b"alpha", &sl);
        set.add(b"beta", &sl);
        assert_eq!(set.encoding(), set::Encoding::Listpack);

        let mut ints = Set::new();
        ints.add(b"1", &sl);
        ints.add(b"9", &sl);
        assert_eq!(ints.encoding(), set::Encoding::Intset);

        let mut zset = Zset::new();
        zset.add(b"a", 1.5, &zl);
        assert_eq!(zset.encoding(), zset::Encoding::Listpack);

        for (want, body) in [
            (T_HASH_LISTPACK, Body::Hash(hash)),
            (T_SET_LISTPACK, Body::Set(set)),
            (T_SET_INTSET, Body::Set(ints)),
            (T_ZSET_LISTPACK, Body::Zset(zset)),
        ] {
            let rec = Record::new(body.clone(), None);
            let payload = dump(&rec).expect("a packed value has an RDB shape");
            assert_eq!(payload[0], want, "wrong type byte for {body:?}");
            // And the walk is gone, not merely bypassed: what comes back has to
            // be the same value or the copy was of the wrong bytes.
            assert_eq!(
                format!("{:?}", round_trip(body.clone())),
                format!("{body:?}")
            );
        }
    }

    /// A hash that has been widened for deadlines is walked even once they have
    /// all gone, because the blob it holds still has the third element per field
    /// and `HASH_LISTPACK` has no room for it.
    #[test]
    fn a_widened_hash_is_not_copied() {
        let l = hash::Limits::DEFAULT;
        let mut hash = Hash::new();
        hash.set(b"one", b"1", &l);
        hash.expire(b"one", 5_000, Cond::Always, 0);
        hash.persist(b"one");
        // The bound leans early and only a reap that walks puts it right, so
        // this is what it takes to get a hash that is on the wider band and has
        // nothing left to say about deadlines.
        hash.reap(6_000);
        assert_eq!(hash.encoding(), hash::Encoding::ListpackEx);
        assert_eq!(hash.soonest_deadline(), None);
        let rec = Record::new(Body::Hash(hash.clone()), None);
        assert_eq!(dump(&rec).expect("a hash has an RDB shape")[0], T_HASH);
        let Body::Hash(back) = round_trip(Body::Hash(hash)) else {
            panic!("a hash came back as something else");
        };
        assert_eq!(back.len(), 1);
        assert_eq!(back.deadline(b"one"), Ask::NoDeadline);
    }

    #[test]
    fn a_hash_carries_its_field_deadlines_across() {
        let l = hash::Limits::DEFAULT;
        let mut hash = Hash::new();
        hash.set(b"keep", b"1", &l);
        hash.set(b"timed", b"2", &l);
        hash.expire(b"timed", 5_000, Cond::Always, 1_000);
        let rec = Record::new(Body::Hash(hash.clone()), None);
        assert_eq!(
            dump(&rec).expect("a hash has an RDB shape")[0],
            T_HASH_METADATA
        );
        let (s, h, li, z) = limits();
        let all = Limits {
            set: &s,
            hash: &h,
            list: &li,
            zset: &z,
        };
        let payload = dump(&rec).expect("a hash has an RDB shape");
        let Body::Hash(back) = load(&payload, all, 1_000).expect("it reads back") else {
            panic!("a hash came back as something else");
        };
        assert_eq!(back.len(), 2);
        assert_eq!(back.deadline(b"timed"), crate::ttl::Ask::At(5_000));
        assert_eq!(back.deadline(b"keep"), crate::ttl::Ask::NoDeadline);
    }

    /// A field whose deadline went while the payload was in flight is not put
    /// back, because the next read would delete it anyway and a count that is
    /// wrong until somebody looks is worse than a field that never arrived.
    #[test]
    fn a_field_that_expired_in_transit_does_not_come_back() {
        let l = hash::Limits::DEFAULT;
        let mut hash = Hash::new();
        hash.set(b"keep", b"1", &l);
        hash.set(b"gone", b"2", &l);
        hash.expire(b"gone", 5_000, Cond::Always, 1_000);
        let rec = Record::new(Body::Hash(hash), None);
        let payload = dump(&rec).expect("a hash has an RDB shape");
        let (s, h, li, z) = limits();
        let all = Limits {
            set: &s,
            hash: &h,
            list: &li,
            zset: &z,
        };
        let Body::Hash(back) = load(&payload, all, 9_000).expect("it reads back") else {
            panic!("a hash came back as something else");
        };
        assert_eq!(back.len(), 1);
        assert!(back.contains(b"keep"));
        assert!(!back.contains(b"gone"));
    }

    #[test]
    fn a_flipped_byte_is_caught() {
        let rec = Record::new(Body::String(b"hello there".to_vec()), None);
        let good = dump(&rec).expect("a string has an RDB shape");
        for i in 0..good.len() {
            let mut bad = good.clone();
            bad[i] ^= 1;
            let (s, h, l, z) = limits();
            let all = Limits {
                set: &s,
                hash: &h,
                list: &l,
                zset: &z,
            };
            assert!(
                load(&bad, all, 0).is_err(),
                "byte {i} could be changed without anything noticing"
            );
        }
    }

    #[test]
    fn a_payload_from_a_newer_server_is_refused() {
        let rec = Record::new(Body::String(b"hello".to_vec()), None);
        let mut payload = dump(&rec).expect("a string has an RDB shape");
        let n = payload.len();
        payload[n - 10] = 99;
        // The checksum has to be put right, or this would pass for the wrong
        // reason and the version check would never be reached.
        let crc = crc64(0, &payload[..n - 8]);
        payload[n - 8..].copy_from_slice(&crc.to_le_bytes());
        let (s, h, l, z) = limits();
        let all = Limits {
            set: &s,
            hash: &h,
            list: &l,
            zset: &z,
        };
        assert_eq!(load(&payload, all, 0).unwrap_err(), Bad::Footer);
    }

    /// Every version up to the one a current Redis stamps is read, and the one
    /// after it is not.
    ///
    /// This is the check that was missing when `RESTORE` was turning down every
    /// payload a real 8.10.1 produced. The old code compared against the version
    /// it writes, which is deliberately old so that old servers accept us, so
    /// making one number do both jobs meant refusing everything modern.
    #[test]
    fn a_payload_is_read_up_to_the_version_that_has_been_checked() {
        let rec = Record::new(Body::String(b"hello".to_vec()), None);
        let (s, h, l, z) = limits();
        let all = Limits {
            set: &s,
            hash: &h,
            list: &l,
            zset: &z,
        };
        for stamp in [VERSION, READS_UP_TO, READS_UP_TO + 1] {
            let mut payload = dump(&rec).expect("a string has an RDB shape");
            let n = payload.len();
            payload[n - 10..n - 8].copy_from_slice(&stamp.to_le_bytes());
            let crc = crc64(0, &payload[..n - 8]);
            payload[n - 8..].copy_from_slice(&crc.to_le_bytes());
            let got = load(&payload, all, 0);
            if stamp > READS_UP_TO {
                assert_eq!(got.unwrap_err(), Bad::Footer, "{stamp} should be refused");
            } else {
                assert!(got.is_ok(), "{stamp} should be read");
            }
        }
        const {
            assert!(
                VERSION <= READS_UP_TO,
                "a server that cannot read what it writes is no use to anybody"
            )
        };
    }

    #[test]
    fn a_payload_shorter_than_its_footer_is_refused() {
        for n in 0..FOOTER {
            assert_eq!(unseal(&vec![0u8; n]), Err(Bad::Footer));
        }
    }

    /// Nothing here should be able to panic on bytes a client made up, so the
    /// whole space of short payloads gets tried with a correct footer on it.
    #[test]
    fn arbitrary_bytes_are_an_error_and_not_a_panic() {
        let (s, h, l, z) = limits();
        let all = Limits {
            set: &s,
            hash: &h,
            list: &l,
            zset: &z,
        };
        for kind in 0u8..=26 {
            for len in 0usize..6 {
                for fill in [0u8, 1, 0x40, 0x80, 0x81, 0xc0, 0xc3, 0xff] {
                    let mut body = vec![kind];
                    body.extend(std::iter::repeat_n(fill, len));
                    let payload = seal(body);
                    let _ = load(&payload, all, 0);
                }
            }
        }
    }

    /// A count bigger than the payload is refused before anything is reserved.
    ///
    /// [`arbitrary_bytes_are_an_error_and_not_a_panic`] already sends these
    /// bytes and could not catch this, for two reasons worth writing down. An
    /// out of memory abort is not a panic, so a test that only says nothing
    /// panics will watch the process die and report nothing. And the dev machine
    /// overcommits, so the reservation succeeded there and only ever failed on
    /// Linux and Windows, which is to say in CI on the release tag and nowhere a
    /// person would see it.
    ///
    /// The bytes are the ones that did it. `0x80` opens a thirty two bit length
    /// and the four after it are the length, so the count comes out as
    /// `0x80808080`, and a row sixteen bytes wide makes that a request for
    /// thirty four gigabytes from a six byte payload.
    #[test]
    fn a_count_past_the_payload_is_refused_before_anything_is_reserved() {
        let (s, h, l, z) = limits();
        let all = Limits {
            set: &s,
            hash: &h,
            list: &l,
            zset: &z,
        };
        for kind in [T_SET, T_HASH, T_ZSET_2, T_ZSET, T_LIST_QUICKLIST_2] {
            let payload = seal(vec![kind, 0x80, 0x80, 0x80, 0x80, 0x80]);
            assert_eq!(
                load(&payload, all, 0).err(),
                Some(Bad::Format),
                "type {kind} took a count of 0x80808080 from six bytes"
            );
        }

        // The bound is the bytes that are left and not a fixed ceiling, so a
        // count that is small in absolute terms is still refused when the
        // payload cannot possibly hold it. Ten members and nothing after the
        // count to hold them.
        let mut body = vec![T_SET];
        put_len(&mut body, 10);
        assert_eq!(load(&seal(body), all, 0).err(), Some(Bad::Format));

        // And a count the payload can hold is read, so the bound is not simply
        // refusing everything.
        let mut body = vec![T_SET];
        put_len(&mut body, 2);
        put_str(&mut body, b"a");
        put_str(&mut body, b"b");
        assert!(load(&seal(body), all, 0).is_ok());
    }

    #[test]
    fn lzf_unpacks_a_literal_run() {
        // One control byte saying four literals, then the four.
        assert_eq!(
            unpack(&[3, b'a', b'b', b'c', b'd'], 4).as_deref(),
            Some(&b"abcd"[..])
        );
    }

    /// The case the byte at a time copy exists for: a back reference that reads
    /// bytes it is in the middle of writing.
    #[test]
    fn lzf_unpacks_an_overlapping_reference() {
        // One literal `a`, then a reference one byte back for five bytes. The
        // low five bits of the control byte and the byte after it are the
        // distance, and they are both zero because a distance is stored one
        // less than it is.
        let packed = [0u8, b'a', 3 << 5, 0];
        assert_eq!(unpack(&packed, 6).as_deref(), Some(&b"aaaaaa"[..]));
    }

    #[test]
    fn lzf_refuses_a_reference_to_nothing() {
        assert_eq!(unpack(&[(3 << 5), 0], 5), None);
        assert_eq!(unpack(&[3, b'a'], 4), None);
    }

    #[test]
    fn an_empty_collection_is_not_a_value() {
        let mut body = vec![T_SET];
        put_len(&mut body, 0);
        let payload = seal(body);
        let (s, h, l, z) = limits();
        let all = Limits {
            set: &s,
            hash: &h,
            list: &l,
            zset: &z,
        };
        assert_eq!(load(&payload, all, 0).unwrap_err(), Bad::Format);
    }

    #[test]
    fn trailing_bytes_are_refused() {
        let mut body = vec![T_STRING];
        put_str(&mut body, b"hello");
        body.push(0);
        let payload = seal(body);
        let (s, h, l, z) = limits();
        let all = Limits {
            set: &s,
            hash: &h,
            list: &l,
            zset: &z,
        };
        assert_eq!(load(&payload, all, 0).unwrap_err(), Bad::Format);
    }

    #[test]
    fn every_length_form_round_trips() {
        for n in [0u64, 63, 64, 16383, 16384, u64::from(u32::MAX), 1 << 40] {
            let mut out = Vec::new();
            put_len(&mut out, n);
            let mut r = Reader::new(&out);
            assert_eq!(r.len_or_encoding(), Ok((n, false)), "{n} did not survive");
            assert!(r.done(), "{n} left bytes behind");
        }
    }

    // -----------------------------------------------------------------------
    // Streams.
    // -----------------------------------------------------------------------

    /// The stream `sample()` builds, dumped by a Redis 8.10.1 over a socket.
    ///
    /// Type 27, which is what a current server writes: `LISTPACKS_3` with a
    /// count on the end of every group and an idempotency block on the end of
    /// the stream. Both are zero here and are zero on anything this server can
    /// hold, since it does not track producer IDs.
    ///
    /// The times in it are real times from the machine it was taken on, which is
    /// the point of hard coding the payload rather than building one: nothing in
    /// this file gets to pick what a delivery time looks like.
    const FROM_REDIS: &[u8] = &[
        0x1b, 0x01, 0x10, 0, 0, 0, 0, 0, 0, 0, 5, 0, 0, 0, 0, 0, 0, 0, 1, 0x40, 0x5f, 0x5f, 0, 0,
        0, 0x1f, 0, 3, 1, 1, 1, 2, 1, 0x86, b's', b'e', b'n', b's', b'o', b'r', 7, 0x87, b'r',
        b'e', b'a', b'd', b'i', b'n', b'g', 8, 0, 1, 2, 1, 0, 1, 0, 1, 0x81, b'a', 2, 1, 1, 5, 1,
        3, 1, 0, 1, 1, 1, 0x81, b'b', 2, 2, 1, 5, 1, 2, 1, 0xc2, 0xb7, 2, 0xdf, 0xff, 2, 0x81,
        b'c', 2, 3, 1, 5, 1, 0, 1, 0xc3, 0x7f, 2, 0xdf, 0xff, 2, 1, 1, 0x85, b'o', b't', b'h',
        b'e', b'r', 6, 0x81, b'x', 2, 6, 1, 0xff, 3, 0x43, 0x84, 0, 5, 1, 5, 2, 4, 2, 2, b'g',
        b'1', 0x42, 0xbc, 0, 0x81, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 2, 0, 0, 0, 0,
        0, 0, 0, 5, 0, 0, 0, 0, 0, 0, 0, 1, 0x50, 0xe1, 0x0a, 0x66, 0xa0, 1, 0, 0, 1, 0, 0, 0, 0,
        0, 0, 2, 0xbc, 0, 0, 0, 0, 0, 0, 0, 0, 0x50, 0xe1, 0x0a, 0x66, 0xa0, 1, 0, 0, 1, 2, 5,
        b'a', b'l', b'i', b'c', b'e', 0x50, 0xe1, 0x0a, 0x66, 0xa0, 1, 0, 0, 0x50, 0xe1, 0x0a,
        0x66, 0xa0, 1, 0, 0, 2, 0, 0, 0, 0, 0, 0, 0, 5, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0,
        2, 0xbc, 0, 0, 0, 0, 0, 0, 0, 0, 3, b'b', b'o', b'b', 0x70, 0xe1, 0x0a, 0x66, 0xa0, 1, 0,
        0, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0, 0, 2, b'g', b'2', 0x43, 0x84, 0,
        0x81, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0, 0, 0, 0x40, 0x64, 0x40, 0x64, 0,
        0, 0, 0x0f, 0, 0x5c, 0x1f, 0xdc, 0xf0, 0xb9, 0x7e, 0x8b, 0x75,
    ];

    /// When the two deliveries in [`FROM_REDIS`] happened, and when `bob` was
    /// made a few milliseconds later.
    const DELIVERED: u64 = 1_788_418_384_208;
    const MADE: u64 = 1_788_418_384_240;

    /// The stream [`FROM_REDIS`] is a dump of, built here instead.
    ///
    /// Four entries with one deleted, two groups, and one of them with two
    /// consumers where only one has anything. Built by the same commands in the
    /// same order, so that the two can be compared as values rather than as
    /// bytes.
    fn sample() -> Stream {
        let l = crate::stream::Limits::default();
        let mut s = Stream::new();
        s.append(Id::new(5, 1), &[(b"sensor", b"a"), (b"reading", b"1")], l)
            .expect("the first entry");
        s.append(Id::new(5, 2), &[(b"sensor", b"b"), (b"reading", b"2")], l)
            .expect("the second entry");
        s.append(Id::new(700, 0), &[(b"sensor", b"c"), (b"reading", b"3")], l)
            .expect("the third entry");
        s.append(Id::new(900, 0), &[(b"other", b"x")], l)
            .expect("the fourth entry");
        assert!(s.delete(Id::new(5, 2)), "the second entry was there");

        s.create_group(b"g1", Id::MIN, None);
        let g = s.group_mut(b"g1").expect("the group just made");
        let alice = g.consumer_or_create(b"alice", DELIVERED);
        g.deliver(alice, Id::new(5, 1), DELIVERED);
        g.deliver(alice, Id::new(700, 0), DELIVERED);
        // What the read path does once it knows the read handed something over,
        // and the reason alice has an active time in the payload while bob,
        // which was only ever declared, does not.
        g.touch(alice, DELIVERED, true);
        // A group over a stream something has been deleted from cannot work out
        // how far it has read, which is what a real server reports too.
        g.set_read(None);
        g.create_consumer(b"bob", MADE);

        s.create_group(b"g2", Id::new(900, 0), None);
        s.group_mut(b"g2")
            .expect("the group just made")
            .set_read(None);
        s
    }

    fn loaded(payload: &[u8]) -> Body {
        let (s, h, l, z) = limits();
        let all = Limits {
            set: &s,
            hash: &h,
            list: &l,
            zset: &z,
        };
        load(payload, all, 0).expect("a payload that should have loaded")
    }

    /// The one that matters, because it is the only test here that did not get
    /// to choose both sides of the comparison.
    #[test]
    fn a_stream_a_real_redis_dumped_loads_the_same_stream() {
        let Body::Stream(back) = loaded(FROM_REDIS) else {
            panic!("a stream came back as something else");
        };
        assert_eq!(back, sample());
    }

    /// The other direction, and the thing `MIGRATE` needs: what this writes has
    /// to be what that reads.
    ///
    /// Not byte for byte against `FROM_REDIS`, because that is type 27 and this
    /// writes type 21 on purpose. What is compared is the payload this writes
    /// against the same payload read back by the reader that already agrees with
    /// a real server.
    #[test]
    fn a_stream_this_wrote_reads_back_as_itself() {
        let s = sample();
        let rec = Record::new(Body::Stream(s.clone()), None);
        let payload = dump(&rec).expect("a stream has an RDB shape");
        assert_eq!(payload[0], T_STREAM_LISTPACKS_3);
        let Body::Stream(back) = loaded(&payload) else {
            panic!("a stream came back as something else");
        };
        assert_eq!(back, s);
    }

    /// A stream everything has been deleted from is still a stream, and a real
    /// server dumps one and restores it rather than treating it as a deleted
    /// key the way it does an empty set.
    #[test]
    fn an_empty_stream_survives_the_round_trip() {
        let l = crate::stream::Limits::default();
        let mut s = Stream::new();
        s.append(Id::new(1, 1), &[(b"a", b"1")], l)
            .expect("the only entry");
        assert!(s.delete(Id::new(1, 1)), "the only entry was there");

        let rec = Record::new(Body::Stream(s.clone()), None);
        let payload = dump(&rec).expect("a stream has an RDB shape");
        let Body::Stream(back) = loaded(&payload) else {
            panic!("a stream came back as something else");
        };
        assert_eq!(back, s);
        assert_eq!(back.len(), 0);
        assert_eq!(back.last_id(), Id::new(1, 1));
        assert_eq!(back.max_deleted_id(), Id::new(1, 1));
        assert_eq!(back.added(), 1);
    }

    /// The same payload a real server writes for one, checked against the bytes
    /// rather than against our own writer, since an empty stream is the one
    /// shape where a count being allowed to be zero could quietly be wrong.
    #[test]
    fn an_empty_stream_from_a_real_redis_loads() {
        let payload: &[u8] = &[
            0x1b, 0, 0, 1, 1, 0, 0, 1, 1, 1, 0, 0x40, 0x64, 0x40, 0x64, 0, 0, 0, 0x0f, 0, 0x30,
            0xd1, 0x23, 0xc3, 0x38, 0xd9, 0xd6, 0x40,
        ];
        let Body::Stream(back) = loaded(payload) else {
            panic!("a stream came back as something else");
        };
        assert_eq!(back.len(), 0);
        assert_eq!(back.nodes(), 0);
        assert_eq!(back.last_id(), Id::new(1, 1));
        assert_eq!(back.added(), 1);
    }

    /// An entry `XNACK` handed back sits in the ledger with nobody against it,
    /// and the format has room for that: the group's pending list is written
    /// before its consumers, so an entry no consumer claims is simply never
    /// claimed.
    #[test]
    fn a_pending_entry_nobody_holds_survives() {
        let l = crate::stream::Limits::default();
        let mut s = Stream::new();
        s.append(Id::new(1, 1), &[(b"a", b"1")], l)
            .expect("the only entry");
        s.create_group(b"g", Id::MIN, Some(0));
        let g = s.group_mut(b"g").expect("the group just made");
        let who = g.consumer_or_create(b"c", 100);
        g.deliver(who, Id::new(1, 1), 100);
        assert!(g.release(Id::new(1, 1), Retry::Keep), "it was delivered");
        assert_eq!(g.nacked_len(), 1);

        let rec = Record::new(Body::Stream(s.clone()), None);
        let payload = dump(&rec).expect("a stream has an RDB shape");
        let Body::Stream(back) = loaded(&payload) else {
            panic!("a stream came back as something else");
        };
        assert_eq!(back, s);
        let g = back.group(b"g").expect("the group came back");
        assert_eq!(g.nacked_len(), 1);
        assert!(
            g.nack(Id::new(1, 1))
                .expect("still pending")
                .owner()
                .is_none()
        );
    }

    /// Every way a stream payload can be wrong that the reader is meant to
    /// notice, each one made by breaking a payload that was fine.
    #[test]
    fn a_stream_payload_that_does_not_add_up_is_refused() {
        let rec = Record::new(Body::Stream(sample()), None);
        let good = dump(&rec).expect("a stream has an RDB shape");
        let body = &good[..good.len() - FOOTER];
        let (s, h, l, z) = limits();
        let all = Limits {
            set: &s,
            hash: &h,
            list: &l,
            zset: &z,
        };

        // The type byte, changed to one nothing writes.
        let mut bent = body.to_vec();
        bent[0] = 26;
        assert_eq!(load(&seal(bent), all, 0).unwrap_err(), Bad::Format);

        // A node key that is not sixteen bytes.
        let mut bent = body.to_vec();
        bent[2] = 15;
        assert_eq!(load(&seal(bent), all, 0).unwrap_err(), Bad::Format);

        // Cut short anywhere is short everywhere.
        for cut in [1, 3, 20, 60, 120, 180] {
            let bent = body[..cut.min(body.len())].to_vec();
            assert_eq!(
                load(&seal(bent), all, 0).unwrap_err(),
                Bad::Format,
                "a payload cut at {cut} was accepted"
            );
        }

        // Anything extra on the end, which is what a type 27 payload read as a
        // 21 would look like.
        let mut bent = body.to_vec();
        bent.push(0);
        assert_eq!(load(&seal(bent), all, 0).unwrap_err(), Bad::Format);
    }

    /// The idempotency block is dropped rather than read, and the payload it is
    /// on still has to end exactly where it says it does.
    #[test]
    fn the_idempotency_block_is_read_past_and_not_kept() {
        let Body::Stream(back) = loaded(FROM_REDIS) else {
            panic!("a stream came back as something else");
        };
        // Nothing here holds a producer ID, so there is nothing to check on the
        // value. What the test is for is that the payload was consumed to the
        // last byte, which `load` only accepts when it was.
        assert_eq!(back.len(), 3);

        // A producer with an ID recorded against it is a shape no `DUMP` has
        // been seen to write, and reading past it would be guessing.
        let body = &FROM_REDIS[..FROM_REDIS.len() - FOOTER];
        let at = body.len() - 3;
        assert_eq!(
            &body[at..],
            &[0, 0, 0],
            "the producer count and the two totals"
        );
        let mut bent = body.to_vec();
        bent[at] = 1;
        let (s, h, l, z) = limits();
        let all = Limits {
            set: &s,
            hash: &h,
            list: &l,
            zset: &z,
        };
        assert_eq!(load(&seal(bent), all, 0).unwrap_err(), Bad::Format);
    }
}