yo_resp/dispatch/

mod.rs

//! From a decoded command to a written reply.
//!
//! This is the layer Y23 exists to keep thin. The wire and the embedded API
//! both have to reach the same code, or there are two implementations of `INCR`
//! and one of them is wrong. So `yo-kv` holds one method per command taking
//! ordinary Rust values, and everything here is about the part that is only
//! true on a socket: which keyword goes where, which combinations a real server
//! refuses, and which of the two protocols the answer is spelled in.
//!
//! # What runs a command
//!
//! [`Server`] holds the databases. [`Session`] holds what one connection has
//! chosen: which database, which name it gave itself, what its id is. The
//! protocol version lives in the [`Out`] because that is what needs it, and
//! `HELLO` changes it there.
//!
//! ```
//! use yo_resp::{Argv, Limits, Out, Proto};
//! use yo_resp::dispatch::{Args, Flow, Server, Session, execute};
//!
//! let mut server = Server::new();
//! let mut session = Session::new(1);
//! let mut out = Out::new(Proto::Resp2);
//!
//! let wire = b"*3\r\n$3\r\nSET\r\n$1\r\nk\r\n$1\r\nv\r\n";
//! let mut argv = Argv::new();
//! argv.decode(wire, &Limits::default())?;
//! let flow = execute(&mut server, &mut session, Args::new(&argv, wire), &mut out);
//!
//! assert_eq!(flow, Flow::Continue);
//! assert_eq!(out.as_slice(), b"+OK\r\n");
//! # Ok::<(), yo_resp::ProtocolError>(())
//! ```
//!
//! # Errors are values until the last moment
//!
//! A command body returns a [`Result`], and this module turns the error into
//! the line that goes on the wire. That is what keeps the same body usable from
//! the embedded API, where an error is a value with a [`Code`] on it and not a
//! sentence to be parsed.
//!
//! The reply buffer is rolled back to where it was before a failing command
//! wrote anything, so a body that checks its arguments halfway through cannot
//! leave half a reply in front of the error.
//!
//! # Nothing here allocates
//!
//! Arguments are slices of the connection's read buffer, keywords are compared
//! in place, numbers are written straight into the reply, and the pairs of
//! `MSET` reach the store as an iterator rather than a `Vec`. The two places
//! that do allocate, an error message and the text of `INFO`, say so and wrap
//! it, because a shard thread that allocates aborts.

mod args;
mod arrays;
mod blocking;
mod cpu;
mod hashes;
mod keyspace;
mod lists;
mod scan;
mod scripting;
mod server;
mod sets;
mod strings;
pub mod table;
mod zsets;

pub use args::Args;
pub use blocking::{Parked, Waiters};
pub use table::{COMMANDS, Spec, arity_ok, lookup};

use crate::reply::Out;
use yo_common::{Code, Error};
use yo_kv::{Clock, Keyspace};

/// How many databases a server has.
///
/// Redis's default is sixteen and its `databases` setting can change it. Ours
/// is sixteen and cannot, which is why `CONFIG GET databases` can answer with a
/// constant. Nothing in the design needs the number to be fixed; nothing yet
/// needs it not to be.
pub const DATABASES: usize = 16;

/// Every database's bit in [`Server::dirty`], which is what a fresh server
/// starts on so that the first maintenance turn asks all of them.
///
/// A `u64` holds sixteen bits with room to spare, and the assertion below is
/// what turns raising [`DATABASES`] past sixty four into a build failure rather
/// than a shift that silently drops the databases past the end.
const ALL_DATABASES: u64 = if DATABASES == 64 {
    u64::MAX
} else {
    (1u64 << DATABASES) - 1
};
const _: () = assert!(DATABASES <= 64);

/// How many keys one command throws away before it leaves the rest to the next.
///
/// A bound and not a loop to the end, because this runs in front of a client
/// that is waiting for its reply, and a server a long way over its limit would
/// otherwise hold that client for as long as it took to walk all the way back
/// under. Sixty four is a batch's worth of commands, so a server that went over
/// by what one batch allocated comes back under in one command, and a server
/// whose limit was just cut in half works through it over the next few thousand
/// rather than in one long stall. Redis bounds the same loop by a time slice
/// instead of a count and hands the rest to a timer; there is no timer here, so
/// the rest goes to the next command that runs.
const EVICT_BUDGET: usize = 64;

/// What a server says to a command that would allocate when it has no room.
///
/// Redis's `shared.oomerr`, word for word including the full stop, because
/// clients match on the `OOM` prefix and people match on the sentence.
const OOM: &[u8] = b"command not allowed when used memory > 'maxmemory'.";

/// What the connection should do after a command.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Flow {
    /// Read the next command.
    Continue,
    /// Write what is buffered and then close, which is what `QUIT` asks for.
    Close,
    /// Nothing was written and nothing is owed yet.
    ///
    /// The client is on the waiter list and its reply comes when a key it named
    /// has something in it or when its deadline passes, whichever happens first.
    /// Until then the connection stops reading commands, because a client that
    /// is waiting for an answer is not a client that has sent another question.
    Block,
}

/// The numbers `INFO` reports that this layer cannot see for itself.
///
/// The reactor owns the sockets, so the reactor is what knows how many clients
/// there are. It writes these directly and nothing here does anything with them
/// except report them.
#[derive(Debug, Clone, Copy, Default)]
pub struct Stats {
    /// Connections open right now.
    pub clients: u64,
    /// Connections accepted since the server started.
    pub connections: u64,
    /// Commands run since the server started, which this layer counts itself.
    pub commands: u64,
}

/// Everything a server holds.
///
/// One of these per shard thread, not one per process: the databases inside are
/// not `Sync` and are reached by sending their thread a command. What makes
/// this a server rather than a shard is that it is the whole of what a
/// connection can address.
pub struct Server {
    dbs: Vec<Keyspace>,
    clock: Clock,
    started_ms: u64,
    /// Where the next maintenance turn starts looking, so that a database
    /// under constant write load cannot hold the other fifteen's space.
    next_db: usize,
    /// One bit per database, set when a command ran against it.
    ///
    /// The maintenance turn after every batch used to ask all sixteen
    /// databases whether they had anything to collect, and asking costs a load
    /// and a store in each one. Fifteen of those are cold lines on a server
    /// where every client is on database zero, which is every server, and the
    /// answer is no every time. This is the cheap half of the question: a
    /// database nobody has touched since it last said no cannot have started
    /// saying yes.
    dirty: u64,
    /// What the connections are holding, kept by the engine.
    conn_bytes: usize,
    /// The `maxmemory` limit in bytes, zero when there is not one.
    ///
    /// Zero is the default and it is the whole reason the check in front of
    /// every write is one comparison against a field that is already warm.
    maxmemory: u64,
    /// What [`Server::memory_bytes`] said at the last maintenance turn.
    ///
    /// The reading is a walk over every collection in every database and cannot
    /// go on a command path, so the command path reads this instead and is at
    /// most one batch behind. What that costs is overshoot: a server can end a
    /// batch holding one batch's worth of allocation more than its limit before
    /// anything notices. A batch is 64 commands, so that is bounded by what 64
    /// commands can allocate and not by how long the server runs.
    ///
    /// Only kept up to date when there is a limit to judge it against. A server
    /// with no `maxmemory` never reads it and never pays for it.
    used: usize,
    /// Which database the next eviction draws from.
    ///
    /// Its own cursor and not [`Server::next_db`], because eviction and
    /// compaction move at different rates and sharing one would make the
    /// database that gets compacted depend on how many keys were evicted.
    evict_db: usize,
    /// Which database the next active expiry sweep starts at.
    ///
    /// A third cursor for the same reason there is a second one. A sweep runs on
    /// every turn of the loop and compaction runs when there is dead space, so
    /// sharing a cursor would make which database gets swept depend on which one
    /// was last collected.
    expire_db: usize,
    /// The millisecond the last active expiry sweep ran on, so the next one on
    /// the same millisecond does not bother.
    expire_ms: u64,
    /// Clients parked on a blocking command.
    waiters: Waiters,
    /// The numbers the reactor keeps for `INFO`.
    pub stats: Stats,
}

impl Server {
    /// A server with [`DATABASES`] empty databases on the system clock.
    #[must_use]
    pub fn new() -> Server {
        let clock = Clock::system();
        Server {
            dbs: (0..DATABASES)
                .map(|_| Keyspace::with_clock(clock))
                .collect(),
            clock,
            started_ms: clock.now_ms(),
            next_db: 0,
            dirty: ALL_DATABASES,
            conn_bytes: 0,
            maxmemory: 0,
            used: 0,
            evict_db: 0,
            expire_db: 0,
            expire_ms: 0,
            waiters: Waiters::default(),
            stats: Stats::default(),
        }
    }

    /// A server on a clock the caller moves by hand, for tests.
    #[must_use]
    pub fn with_clock(clock: Clock) -> Server {
        Server {
            dbs: (0..DATABASES)
                .map(|_| Keyspace::with_clock(clock))
                .collect(),
            clock,
            started_ms: clock.now_ms(),
            next_db: 0,
            dirty: ALL_DATABASES,
            conn_bytes: 0,
            maxmemory: 0,
            used: 0,
            evict_db: 0,
            expire_db: 0,
            expire_ms: 0,
            waiters: Waiters::default(),
            stats: Stats::default(),
        }
    }

    /// One database, by index.
    ///
    /// # Panics
    ///
    /// If `i` is not a database. `SELECT` is the only way a client changes the
    /// index and it checks, so an index that is out of range here is a bug in
    /// the caller and not something a client can ask for.
    pub fn db(&mut self, i: usize) -> &mut Keyspace {
        // The borrow is mutable, so assume it is used. Anything that only reads
        // has [`Server::db_ref`] and does not come through here.
        self.dirty |= 1u64 << i;
        &mut self.dbs[i]
    }

    /// One database, by index, without taking it mutably.
    ///
    /// What the prefetch stage needs. It runs for all 64 commands in a batch
    /// before any of them executes, so it cannot hold the mutable borrow `run`
    /// is about to want, and it does not need one: warming a cache line reads
    /// nothing and changes nothing.
    ///
    /// # Panics
    ///
    /// As [`Server::db`].
    #[must_use]
    pub fn db_ref(&self, i: usize) -> &Keyspace {
        &self.dbs[i]
    }

    /// Take a new clock reading and give it to every database.
    ///
    /// Once per turn of the event loop, which is the only place time moves. A
    /// command asking what the time is gets the answer the whole batch got, so
    /// two keys written by the same batch expire together (`04` section 3).
    pub fn refresh_clock(&mut self) {
        self.clock.refresh();
        let now = self.clock.now_ms();
        for db in &mut self.dbs {
            db.clock_mut().set(now);
        }
    }

    /// Move every clock here to `ms` by hand, for tests about expiry.
    ///
    /// A test cannot wait a hundred seconds and a test that waits a hundred
    /// milliseconds is a test that fails on a loaded machine, so time moves on
    /// request. The system clock underneath will overwrite this on the next
    /// [`Server::refresh_clock`], which is why this is only useful in a test
    /// that drives commands directly rather than through the event loop.
    pub fn set_clock_ms(&mut self, ms: u64) {
        self.clock.set(ms);
        for db in &mut self.dbs {
            db.clock_mut().set(ms);
        }
    }

    /// Seconds since this server was built.
    #[must_use]
    pub fn uptime_secs(&self) -> u64 {
        self.clock.now_ms().saturating_sub(self.started_ms) / 1000
    }

    /// Bytes held by every database's index and arena, plus the read and reply
    /// buffers of every connection.
    ///
    /// The buffers are in here because they are real and because Redis counts
    /// its own, so leaving them out would make the one number people compare
    /// flattering rather than true. They are not a database, so nothing in the
    /// keyspace can change them and the engine has to say when they move.
    #[must_use]
    pub fn memory_bytes(&self) -> usize {
        self.dbs.iter().map(Keyspace::memory_bytes).sum::<usize>() + self.conn_bytes
    }

    /// What the keyspace itself is holding, live records only.
    ///
    /// `used_memory` minus this is what the store costs to run: the index, the
    /// space dead records are sitting in until compaction gets to them, and the
    /// connections' buffers.
    #[must_use]
    pub fn dataset_bytes(&self) -> usize {
        self.dbs
            .iter()
            .map(|db| db.map().arena().live_bytes() as usize)
            .sum()
    }

    /// Bytes the arenas are holding, live and dead together.
    #[must_use]
    pub fn arena_bytes(&self) -> usize {
        self.dbs
            .iter()
            .map(|db| db.map().arena().reserved_bytes() as usize)
            .sum()
    }

    /// Bytes the indexes are holding.
    #[must_use]
    pub fn index_bytes(&self) -> usize {
        self.dbs
            .iter()
            .map(|db| db.map().index().memory_bytes())
            .sum()
    }

    /// Arena segments whose pages are real, across every database.
    #[must_use]
    pub fn segment_count(&self) -> usize {
        self.dbs
            .iter()
            .map(|db| db.map().arena().resident_segments())
            .sum()
    }

    /// What the connections' read and reply buffers are holding.
    #[must_use]
    pub const fn conn_bytes(&self) -> usize {
        self.conn_bytes
    }

    /// Note that the connections are holding `delta` bytes more than they were,
    /// or fewer when it is negative.
    ///
    /// A delta and not a total because the alternative is a walk over every
    /// connection, and the walk would have to happen on a turn of the loop
    /// rather than when `INFO` asks, which puts the cost of a report on the
    /// command path of a server nobody is asking.
    pub fn note_conn_bytes(&mut self, delta: isize) {
        self.conn_bytes = self.conn_bytes.saturating_add_signed(delta);
    }

    /// Keys reclaimed by running into them after their deadline.
    #[must_use]
    pub fn expired_keys(&self) -> u64 {
        self.dbs.iter().map(Keyspace::expired_keys).sum()
    }

    /// Keys thrown away to make room, which is the other number entirely.
    #[must_use]
    pub fn evicted_keys(&self) -> u64 {
        self.dbs.iter().map(Keyspace::evicted_keys).sum()
    }

    /// The `maxmemory` limit in bytes, zero when there is not one.
    #[must_use]
    pub const fn maxmemory(&self) -> u64 {
        self.maxmemory
    }

    /// Set the limit, and take a reading straight away.
    ///
    /// The reading is here rather than left to the next maintenance turn because
    /// a client that sets the limit and sends a write in the same batch expects
    /// the write to be judged against the limit it just set, and because the
    /// cached number is meaningless until the first time there is a limit to
    /// compare it with.
    ///
    /// Turning the limit on also turns on the running total every slab keeps of
    /// what its collections hold, and turning it off turns that back off, so a
    /// server with no limit is not paying to count something nobody reads. The
    /// first reading after switching it on is the walk that the total starts
    /// from, and it is the only walk.
    pub fn set_maxmemory(&mut self, bytes: u64) {
        self.maxmemory = bytes;
        for db in &mut self.dbs {
            db.track_memory(bytes != 0);
        }
        self.used = self.settled_memory();
    }

    /// Take a fresh memory reading, which the maintenance turn does once a batch.
    ///
    /// Nothing at all when there is no limit, which is the default and is every
    /// server that has not asked for one.
    pub fn refresh_memory(&mut self) {
        if self.maxmemory != 0 {
            self.used = self.settled_memory();
        }
    }

    /// [`Server::memory_bytes`], asked the cheap way.
    ///
    /// The same number. The difference is that this asks each database only
    /// about the collections that could have moved since the last time, which is
    /// what a batch touched rather than what the server holds, so it can be
    /// asked once a batch and again on every command that is over the limit.
    fn settled_memory(&mut self) -> usize {
        self.dbs
            .iter_mut()
            .map(Keyspace::settled_memory_bytes)
            .sum::<usize>()
            + self.conn_bytes
    }

    /// Make room under the `maxmemory` limit, throwing keys away if that is what
    /// it takes. Answers whether there is anything left it could throw away.
    ///
    /// Redis runs the same thing from `processCommand` before every command and
    /// so does this: a client that writes has to be judged at the moment it
    /// writes, not a batch later, or the limit is a suggestion.
    ///
    /// Three things happen in the loop and all three are needed. Eviction picks
    /// a key and drops it. Compaction gives the pages back, because dropping a
    /// key marks its record dead and returns nothing on its own, so a loop that
    /// only evicted would throw the whole keyspace away and watch the number
    /// stay where it was. The reading is taken again each time round, because
    /// the two of them together are the only thing that moves it.
    ///
    /// # Why running out of budget is not a no
    ///
    /// `false` means there was nothing left to evict, which is `noeviction`, or
    /// a `volatile` policy on a database where nothing has a deadline, or a
    /// keyspace that is already empty. It does not mean the server is still over
    /// its limit, and that difference is Redis's: `performEvictions` answers
    /// `EVICT_FAIL` only when it has run out of things to delete, and
    /// `processCommand` refuses the client on that and on nothing else. Running
    /// out of time part way through a job it is doing well comes back as
    /// `EVICT_RUNNING` and the command goes through, because a server that is
    /// evicting steadily and refusing every write while it does it is worse for
    /// the client than a little overshoot.
    ///
    /// # What the limit is worth
    ///
    /// Space comes back a segment at a time and a segment is two megabytes, so
    /// this holds a server to its limit give or take a segment. A `maxmemory` of
    /// a few hundred megabytes gets what it asked for. A `maxmemory` of four
    /// megabytes is asking for a precision this store does not have.
    pub fn make_room(&mut self) -> bool {
        if self.maxmemory == 0 || self.used as u64 <= self.maxmemory {
            return true;
        }
        // The cached reading is a batch old and the batch may have compacted
        // since, so take a fresh one before throwing anything away. It is the
        // settled reading and not the walk, so what this costs is the handful of
        // collections the last batch touched and not the whole database.
        self.used = self.settled_memory();
        let mut budget = EVICT_BUDGET;
        while self.used as u64 > self.maxmemory {
            if !self.evict_step() {
                return false;
            }
            self.compact_hard_step();
            self.used = self.settled_memory();
            budget -= 1;
            if budget == 0 {
                break;
            }
        }
        true
    }

    /// Throw one key away, from whichever database has one to give.
    ///
    /// Round robin from a cursor rather than always starting at database zero,
    /// so a server using more than one of them does not empty the first before
    /// touching the second. Almost every server is on database zero only, where
    /// this is one call that answers and fifteen that say the map is empty.
    fn evict_step(&mut self) -> bool {
        for turn in 0..self.dbs.len() {
            let i = (self.evict_db + turn) % self.dbs.len();
            if self.dbs[i].evict_one() {
                self.evict_db = (i + 1) % self.dbs.len();
                self.dirty |= 1u64 << i;
                return true;
            }
        }
        false
    }

    /// The sweep the shard loop calls, at most once a millisecond.
    ///
    /// The gate is the whole difference between this and [`Server::expire_step`].
    /// A maintenance slice runs on every turn of the loop and a turn is a
    /// hundred nanoseconds, so an ungated sweep would draw a fresh sample ten
    /// thousand times per millisecond and spend a real share of the shard on
    /// looking for keys that cannot have died since the last look. Nothing in a
    /// database changes fast enough to be worth asking about more often than the
    /// clock can tell the difference, and the clock here is milliseconds.
    ///
    /// A millisecond is also far finer than Redis, whose slow cycle runs at ten
    /// hertz, so this is not the thing that decides how promptly memory comes
    /// back. What it decides is that an idle server sweeps a thousand times a
    /// second rather than a million.
    pub fn expire_slice(&mut self, budget: usize) -> usize {
        let now = self.clock.now_ms();
        if now == self.expire_ms {
            return 0;
        }
        self.expire_ms = now;
        self.expire_step(budget)
    }

    /// Sweep dead keys out of the databases, spending at most `budget` looks.
    ///
    /// Answers what it spent, so the caller can charge its maintenance slice for
    /// it. See [`yo_kv::expiry`] for why the budget is in keys looked at.
    ///
    /// Round robin from its own cursor, and every database gets offered whatever
    /// is left of the budget rather than a sixteenth of it each, so a server on
    /// database zero only, which is nearly every server, spends the whole slice
    /// where the keys are. The fifteen empty ones cost a comparison apiece
    /// because a database with no key carrying a deadline says so without
    /// drawing anything.
    ///
    /// The cursor moves to the database after whichever one did the work, so two
    /// busy databases take turns instead of the lower numbered one starving the
    /// other.
    pub fn expire_step(&mut self, budget: usize) -> usize {
        let mut spent = 0;
        for turn in 0..self.dbs.len() {
            if spent >= budget {
                break;
            }
            let i = (self.expire_db + turn) % self.dbs.len();
            let c = self.dbs[i].expire_cycle(budget - spent);
            spent += c.examined;
            if c.expired > 0 {
                self.expire_db = (i + 1) % self.dbs.len();
                self.dirty |= 1u64 << i;
            }
        }
        spent
    }

    /// One slice of compaction for a server that is over its limit.
    ///
    /// Takes the databases in the same order [`Server::compact_step`] does and
    /// stops at the first one that had something to move, and it asks with the
    /// ratios off. See [`Keyspace::compact_hard`] for what that changes.
    fn compact_hard_step(&mut self) -> Option<usize> {
        for turn in 0..self.dbs.len() {
            let i = (self.next_db + turn) % self.dbs.len();
            if let Some(moved) = self.dbs[i].compact_hard() {
                self.next_db = (i + 1) % self.dbs.len();
                return Some(moved);
            }
        }
        None
    }

    /// Give one database's dead space back, if any database has enough of it to
    /// be worth the move. `None` when no database had a candidate.
    ///
    /// Once per batch, next to the clock. Overwriting a key writes a new record
    /// and counts the old one dead, so without this a server holds everything
    /// it has ever written: 400000 sets over 100000 keys measured at 742 bytes
    /// a key against Redis at 144 for the same load, and the whole difference
    /// was dead records nothing ever came back for.
    ///
    /// At most one segment moves per call and the search starts one database
    /// further along each time, so the cost of asking is a comparison per
    /// database and the cost of acting is bounded by a segment.
    pub fn compact_step(&mut self) -> Option<usize> {
        for turn in 0..self.dbs.len() {
            let i = (self.next_db + turn) % self.dbs.len();
            // Nothing has run against this database since it last said it had
            // nothing to collect, so it still has nothing to collect and the
            // line it lives on stays where it is.
            if self.dirty & (1 << i) == 0 {
                continue;
            }
            if let Some(moved) = self.dbs[i].compact_step() {
                self.next_db = (i + 1) % self.dbs.len();
                return Some(moved);
            }
            self.dirty &= !(1u64 << i);
        }
        None
    }
}

impl Default for Server {
    fn default() -> Server {
        Server::new()
    }
}

/// What one connection has chosen.
pub struct Session {
    db: usize,
    id: u64,
    name: Vec<u8>,
}

impl Session {
    /// A new connection, on database zero with no name.
    #[must_use]
    pub fn new(id: u64) -> Session {
        Session {
            db: 0,
            id,
            name: Vec::new(),
        }
    }

    /// The connection id, which `HELLO` reports and `CLIENT` will.
    #[must_use]
    pub const fn id(&self) -> u64 {
        self.id
    }

    /// Which database this connection is working in.
    #[must_use]
    pub const fn db(&self) -> usize {
        self.db
    }

    /// The name the client gave itself, empty if it gave none.
    #[must_use]
    pub fn name(&self) -> &[u8] {
        &self.name
    }

    /// Put everything back the way it was when the connection was opened.
    ///
    /// The protocol is not here because it is not here: it lives in the reply
    /// buffer, and `RESET` sets it back there.
    pub fn reset(&mut self) {
        self.db = 0;
        self.name.clear();
    }

    /// Record the name from `HELLO ... SETNAME`.
    fn set_name(&mut self, name: &[u8]) {
        yo_alloc::allow(|| {
            self.name.clear();
            self.name.extend_from_slice(name);
        });
    }
}

/// Run one command and write its reply.
///
/// The name is looked up and the arity is checked here, once, so that no body
/// has to. Everything after that is the command's own.
pub fn execute(server: &mut Server, session: &mut Session, args: Args<'_>, out: &mut Out) -> Flow {
    // The decoder never produces a command with no name. If one ever arrives,
    // it is not something to answer.
    if args.is_empty() {
        return Flow::Continue;
    }
    server.stats.commands += 1;

    let Some(spec) = lookup(args.name()) else {
        write_error(out, &args::unknown_command(args));
        return Flow::Continue;
    };
    if !arity_ok(spec, args.len()) {
        write_error(out, &args::wrong_arity(spec.name));
        return Flow::Continue;
    }

    // The limit first, so a server with no `maxmemory`, which is the default and
    // is nearly all of them, pays one comparison against a field that is already
    // warm. Every command and not only the writes, because that is where Redis
    // puts it: making room is the server's job whatever the client asked for,
    // and the flag only decides who gets told no when there is no room to make.
    //
    // The flag is Redis's own `denyoom` and the list of commands carrying it is
    // Redis's list, so a command that only frees is let through with nothing
    // left, which is what lets a client dig itself out with `DEL`.
    if server.maxmemory != 0 && !server.make_room() && spec.flags.contains(&"denyoom") {
        out.error_line(b"OOM ", OOM);
        return Flow::Continue;
    }

    // Which databases the maintenance turn after this batch has to ask. Marked
    // for every command and not only for the writes, because a read can make
    // garbage too: a `GET` on a key whose expiry has passed reaps it, and the
    // record it dropped is exactly the kind of thing the collector is for.
    // `COPY`, `SWAPDB` and `FLUSHALL` reach a database nobody selected, so the
    // two groups that hold them mark all of them rather than the session's.
    server.dirty |= match spec.group {
        "string" | "set" | "hash" | "list" | "zset" | "array" => 1u64 << session.db,
        _ => ALL_DATABASES,
    };

    let mark = out.len();
    // Before the group, because the five that block are list commands and would
    // otherwise land in `lists`, which is handed one database and nothing that
    // could park a client. The flag is the right thing to branch on rather than
    // a list of names: it is what `COMMAND INFO` reports about exactly these
    // commands, and the sorted set and stream ones that arrive later carry it
    // too.
    let done = if spec.flags.contains(&"blocking") {
        blocking::execute(server, session, spec, args, out)
    } else {
        match spec.group {
            "string" => {
                let db = session.db;
                strings::execute(&mut server.dbs[db], spec, args, out).map(|()| Flow::Continue)
            }
            "set" => {
                let db = session.db;
                sets::execute(&mut server.dbs[db], spec, args, out).map(|()| Flow::Continue)
            }
            "hash" => {
                let db = session.db;
                hashes::execute(&mut server.dbs[db], spec, args, out).map(|()| Flow::Continue)
            }
            "list" => {
                let db = session.db;
                lists::execute(&mut server.dbs[db], spec, args, out).map(|()| Flow::Continue)
            }
            "zset" => {
                let db = session.db;
                zsets::execute(&mut server.dbs[db], spec, args, out).map(|()| Flow::Continue)
            }
            "array" => {
                let db = session.db;
                arrays::execute(&mut server.dbs[db], spec, args, out).map(|()| Flow::Continue)
            }
            // Every database and not the one the session is on, because `COPY` takes
            // a `DB n` and writes into a database nobody selected.
            "keyspace" => keyspace::execute(&mut server.dbs, session.db, spec, args, out)
                .map(|()| Flow::Continue),
            "scripting" => scripting::execute(spec, args, out).map(|()| Flow::Continue),
            _ => server::execute(server, session, spec, args, out),
        }
    };
    match done {
        Ok(flow) => flow,
        Err(e) => {
            out.truncate(mark);
            write_error(out, &e);
            Flow::Continue
        }
    }
}

/// The error line for an error value.
///
/// The prefix is what a client branches on, and there are only two of them in
/// this milestone: `WRONGTYPE` for a command sent at the wrong kind of value,
/// and `ERR` for everything else. The three errors that need a different one,
/// `NOPROTO`, `WRONGPASS` and `OOM`, are written where they are decided rather
/// than routed through here. `OOM` is not a [`Code`] of its own because
/// [`Code::Full`] already covers the string that is too long for
/// `proto-max-bulk-len`, and that one goes out as `ERR` on a real server.
fn write_error(out: &mut Out, e: &Error) {
    let prefix: &[u8] = match e.code() {
        Code::WrongType => b"WRONGTYPE ",
        _ => b"ERR ",
    };
    out.error_line(prefix, e.message().as_bytes());
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::proto::{Limits, Proto};
    use crate::request::Argv;

    /// Build the wire bytes for a command.
    ///
    /// Tests go through the codec rather than around it, so an argument in a
    /// test is the same borrowed slice a connection produces.
    pub(crate) fn encode(parts: &[&[u8]]) -> Vec<u8> {
        let mut wire = format!("*{}\r\n", parts.len()).into_bytes();
        for p in parts {
            wire.extend_from_slice(format!("${}\r\n", p.len()).as_bytes());
            wire.extend_from_slice(p);
            wire.extend_from_slice(b"\r\n");
        }
        wire
    }

    /// A server, a connection and a buffer, driven the way the reactor will.
    struct Fixture {
        server: Server,
        session: Session,
        argv: Argv,
        out: Out,
    }

    impl Fixture {
        fn new() -> Fixture {
            Fixture {
                server: Server::new(),
                session: Session::new(7),
                argv: Argv::new(),
                out: Out::new(Proto::Resp2),
            }
        }

        /// Run one command and answer with the bytes it wrote.
        fn run(&mut self, parts: &[&[u8]]) -> String {
            self.flow(parts).1
        }

        /// Move every clock in the server on by `ms`.
        fn advance(&mut self, ms: u64) {
            for db in 0..DATABASES {
                self.server.db(db).clock_mut().advance(ms);
            }
        }

        /// The same, with what the connection should do next.
        fn flow(&mut self, parts: &[&[u8]]) -> (Flow, String) {
            let wire = encode(parts);
            self.argv.decode(&wire, &Limits::default()).unwrap();
            self.out.clear();
            let flow = execute(
                &mut self.server,
                &mut self.session,
                Args::new(&self.argv, &wire),
                &mut self.out,
            );
            (
                flow,
                String::from_utf8_lossy(self.out.as_slice()).into_owned(),
            )
        }
    }

    /// What a client does all day: write the same keys again and again. Every
    /// one of those writes leaves the previous record behind, so a server that
    /// never compacts holds every version of every key it has ever been sent.
    #[test]
    fn rewriting_the_same_keys_does_not_grow_the_server() {
        let mut f = Fixture::new();
        let val = vec![b'v'; 1024];
        let keys: Vec<Vec<u8>> = (0..64).map(|i| format!("key:{i}").into_bytes()).collect();

        for k in &keys {
            f.run(&[b"SET", k, &val]);
        }
        f.server.compact_step();
        let after_first = f.server.memory_bytes();

        // 64 KiB a pass, five hundred passes, and the same 64 keys at the end
        // of it. Thirty two megabytes written to hold sixty four kilobytes,
        // which is the shape of a real workload and is enough churn to fill
        // sixteen segments if nothing ever comes back.
        for _ in 0..500 {
            for k in &keys {
                f.run(&[b"SET", k, &val]);
            }
            f.server.compact_step();
        }

        assert!(
            f.server.memory_bytes() <= after_first * 2,
            "held {} after five hundred passes against {after_first} after one",
            f.server.memory_bytes()
        );
        assert_eq!(f.run(&[b"DBSIZE"]), format!(":{}\r\n", keys.len()));
        assert_eq!(f.run(&[b"STRLEN", b"key:7"]), ":1024\r\n");
    }

    /// The same churn on a database nobody starts on, either side of a quiet
    /// spell long enough for the maintenance turn to stop asking about it.
    ///
    /// The turn after each batch skips a database that has already said it has
    /// nothing to collect and has not been touched since, which is what keeps a
    /// server whose clients are all on database zero from loading and storing
    /// in the other fifteen every batch to be told no. Two things could go
    /// wrong with that. A database might never be marked at all, so this uses
    /// database nine, which nothing marks by accident. And a database whose
    /// mark was cleared might never get it back, so this drains the collector
    /// until it says there is nothing left, checks the mark really is gone, and
    /// then writes another thirty two megabytes through the same sixty four
    /// keys. If either went wrong the server would hold all of it.
    #[test]
    fn a_database_nobody_started_on_is_still_collected() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"SELECT", b"9"]), "+OK\r\n");
        let val = vec![b'v'; 1024];
        let keys: Vec<Vec<u8>> = (0..64).map(|i| format!("key:{i}").into_bytes()).collect();

        for k in &keys {
            f.run(&[b"SET", k, &val]);
        }
        while f.server.compact_step().is_some() {}
        assert_eq!(
            f.server.dirty & (1 << 9),
            0,
            "database nine was drained and should not be asked again until it is written to"
        );
        let after_first = f.server.memory_bytes();

        for _ in 0..500 {
            for k in &keys {
                f.run(&[b"SET", k, &val]);
            }
            f.server.compact_step();
        }

        assert!(
            f.server.memory_bytes() <= after_first * 2,
            "held {} after five hundred passes against {after_first} after one",
            f.server.memory_bytes()
        );
        assert_eq!(f.run(&[b"DBSIZE"]), format!(":{}\r\n", keys.len()));
        assert_eq!(f.run(&[b"STRLEN", b"key:7"]), ":1024\r\n");
        // And nothing landed anywhere else on the way.
        f.run(&[b"SELECT", b"0"]);
        assert_eq!(f.run(&[b"DBSIZE"]), ":0\r\n");
    }

    #[test]
    fn a_command_goes_from_bytes_to_bytes() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"SET", b"k", b"v"]), "+OK\r\n");
        assert_eq!(f.run(&[b"GET", b"k"]), "$1\r\nv\r\n");
        assert_eq!(f.run(&[b"GET", b"nosuch"]), "$-1\r\n");
        assert_eq!(f.run(&[b"STRLEN", b"k"]), ":1\r\n");
        // The name is matched whatever case it came in, and so are the options.
        assert_eq!(f.run(&[b"set", b"k", b"v2", b"xx"]), "+OK\r\n");
        assert_eq!(f.run(&[b"GET", b"k"]), "$2\r\nv2\r\n");
    }

    #[test]
    fn deleting_counts_keys_removed_and_existing_counts_arguments_matched() {
        let mut f = Fixture::new();
        f.run(&[b"MSET", b"a", b"1", b"b", b"2", b"c", b"3"]);
        // A key named twice exists twice and can only be deleted once, and both
        // of those are Redis's answers rather than tidier ones.
        assert_eq!(f.run(&[b"EXISTS", b"a", b"a", b"nosuch"]), ":2\r\n");
        assert_eq!(f.run(&[b"DEL", b"a", b"a", b"nosuch"]), ":1\r\n");
        assert_eq!(f.run(&[b"EXISTS", b"a"]), ":0\r\n");
        // UNLINK is the same body and reports the same way.
        assert_eq!(f.run(&[b"UNLINK", b"b", b"c"]), ":2\r\n");
        assert_eq!(f.run(&[b"DBSIZE"]), ":0\r\n");
    }

    #[test]
    fn type_is_a_simple_string_and_says_none_for_a_key_that_is_not_there() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"k", b"v"]);
        // A simple string on both protocols, which is unusual: most replies
        // that carry a word are bulk strings.
        assert_eq!(f.run(&[b"TYPE", b"k"]), "+string\r\n");
        assert_eq!(f.run(&[b"TYPE", b"nosuch"]), "+none\r\n");
    }

    #[test]
    fn touch_counts_the_way_exists_counts() {
        let mut f = Fixture::new();
        f.run(&[b"MSET", b"a", b"1", b"b", b"2"]);
        assert_eq!(f.run(&[b"TOUCH", b"a", b"b"]), ":2\r\n");
        assert_eq!(
            f.run(&[b"TOUCH", b"a", b"a"]),
            ":2\r\n",
            "twice counts twice"
        );
        assert_eq!(f.run(&[b"TOUCH", b"a", b"nosuch"]), ":1\r\n");
        assert_eq!(f.run(&[b"TOUCH", b"nosuch"]), ":0\r\n");
    }

    #[test]
    fn a_rename_moves_the_deadline_with_the_value_and_drops_the_one_it_lands_on() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"a", b"v1", b"EX", b"100"]);
        f.run(&[b"SET", b"b", b"v2", b"EX", b"500"]);

        assert_eq!(f.run(&[b"RENAME", b"a", b"b"]), "+OK\r\n");
        assert_eq!(f.run(&[b"GET", b"b"]), "$2\r\nv1\r\n");
        assert_eq!(
            f.run(&[b"TTL", b"b"]),
            ":100\r\n",
            "the source's and not b's"
        );
        assert_eq!(f.run(&[b"EXISTS", b"a"]), ":0\r\n");
    }

    #[test]
    fn a_rename_with_no_source_is_an_error_and_not_a_zero() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"RENAME", b"a", b"b"]), "-ERR no such key\r\n");
        // The source is checked before the destination, so this is the error
        // and not the zero RENAMENX would otherwise answer for a taken name.
        assert_eq!(f.run(&[b"RENAMENX", b"a", b"a"]), "-ERR no such key\r\n");
    }

    #[test]
    fn renamenx_refuses_a_taken_name_including_the_one_it_already_has() {
        let mut f = Fixture::new();
        f.run(&[b"MSET", b"a", b"v1", b"b", b"v2"]);

        assert_eq!(f.run(&[b"RENAMENX", b"a", b"b"]), ":0\r\n");
        assert_eq!(f.run(&[b"GET", b"b"]), "$2\r\nv2\r\n");
        // Renaming onto itself is 0 here and OK for plain RENAME, which is the
        // one call the two disagree about and neither does any work for.
        assert_eq!(f.run(&[b"RENAMENX", b"a", b"a"]), ":0\r\n");
        assert_eq!(f.run(&[b"RENAME", b"a", b"a"]), "+OK\r\n");
        assert_eq!(f.run(&[b"RENAMENX", b"a", b"c"]), ":1\r\n");
        assert_eq!(f.run(&[b"GET", b"c"]), "$2\r\nv1\r\n");
    }

    #[test]
    fn renaming_a_set_does_not_touch_a_member() {
        let mut f = Fixture::new();
        for i in 0..300 {
            f.run(&[b"SADD", b"s", format!("m{i}").as_bytes()]);
        }
        let before = f.server.memory_bytes();

        assert_eq!(f.run(&[b"RENAME", b"s", b"t"]), "+OK\r\n");
        assert_eq!(f.run(&[b"SCARD", b"t"]), ":300\r\n");
        assert_eq!(f.run(&[b"TYPE", b"t"]), "+set\r\n");
        assert!(
            f.server.memory_bytes().abs_diff(before) < 256,
            "the members were copied: {} against {before}",
            f.server.memory_bytes()
        );
    }

    #[test]
    fn a_copy_is_a_second_value_and_not_a_second_name() {
        let mut f = Fixture::new();
        f.run(&[b"SADD", b"s", b"m1", b"m2"]);

        assert_eq!(f.run(&[b"COPY", b"s", b"t"]), ":1\r\n");
        f.run(&[b"SADD", b"t", b"m3"]);
        assert_eq!(f.run(&[b"SCARD", b"s"]), ":2\r\n", "the original is intact");
        assert_eq!(f.run(&[b"SCARD", b"t"]), ":3\r\n");
    }

    /// Every type a key can hold, copied, because two of them used to panic.
    ///
    /// `COPY` reads the value out of the source through one match on the type
    /// tag, and that match had a catch all at the bottom from back when a set
    /// and a hash were the only bodies. The list and the sorted set landed after
    /// it and nobody came back, so `COPY mylist other` took the shard down. It
    /// is an ordinary command against a type the server supports everywhere
    /// else, so this walks all five rather than the two that were broken: the
    /// point is that the next type cannot land the same way.
    #[test]
    fn every_type_can_be_copied() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"str", b"v1"]);
        f.run(&[b"SADD", b"set", b"m1"]);
        f.run(&[b"HSET", b"hash", b"f", b"v"]);
        f.run(&[b"RPUSH", b"list", b"a", b"b"]);
        f.run(&[b"ZADD", b"zset", b"1", b"m1"]);

        for name in [
            &b"str"[..],
            &b"set"[..],
            &b"hash"[..],
            &b"list"[..],
            &b"zset"[..],
        ] {
            let dst = [name, b":copy"].concat();
            assert_eq!(
                f.run(&[b"COPY", name, &dst]),
                ":1\r\n",
                "copying {}",
                String::from_utf8_lossy(name)
            );
            assert_eq!(f.run(&[b"TYPE", name]), f.run(&[b"TYPE", &dst]));
        }

        assert_eq!(f.run(&[b"LRANGE", b"list:copy", b"0", b"-1"]), {
            let mut want = String::from("*2\r\n");
            want.push_str("$1\r\na\r\n$1\r\nb\r\n");
            want
        });
        assert_eq!(f.run(&[b"ZSCORE", b"zset:copy", b"m1"]), "$1\r\n1\r\n");

        // And the copy is its own value, not a second name for the source.
        f.run(&[b"RPUSH", b"list:copy", b"c"]);
        assert_eq!(f.run(&[b"LLEN", b"list"]), ":2\r\n");
        assert_eq!(f.run(&[b"LLEN", b"list:copy"]), ":3\r\n");
    }

    #[test]
    fn a_copy_refuses_a_taken_destination_until_it_is_told_it_can_have_it() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"a", b"v1", b"EX", b"100"]);
        f.run(&[b"SET", b"b", b"v2"]);

        assert_eq!(f.run(&[b"COPY", b"a", b"b"]), ":0\r\n");
        assert_eq!(f.run(&[b"GET", b"b"]), "$2\r\nv2\r\n");
        assert_eq!(f.run(&[b"COPY", b"a", b"b", b"REPLACE"]), ":1\r\n");
        assert_eq!(f.run(&[b"GET", b"b"]), "$2\r\nv1\r\n");
        assert_eq!(f.run(&[b"TTL", b"b"]), ":100\r\n", "the deadline came too");
        assert_eq!(f.run(&[b"COPY", b"nosuch", b"z"]), ":0\r\n");
    }

    #[test]
    fn a_copy_into_another_database_is_a_copy_and_onto_itself_there_is_too() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"a", b"v1"]);

        // Same key, different database, so this is not the same object and is
        // an ordinary copy. Same key in the same database is the error below.
        assert_eq!(f.run(&[b"COPY", b"a", b"a", b"DB", b"1"]), ":1\r\n");
        f.run(&[b"SELECT", b"1"]);
        assert_eq!(f.run(&[b"GET", b"a"]), "$2\r\nv1\r\n");
        assert_eq!(
            f.run(&[b"COPY", b"a", b"a", b"DB", b"0"]),
            ":0\r\n",
            "taken"
        );
        assert_eq!(
            f.run(&[b"COPY", b"a", b"a", b"DB", b"0", b"REPLACE"]),
            ":1\r\n"
        );
    }

    #[test]
    fn copy_checks_its_options_before_it_looks_for_anything() {
        let mut f = Fixture::new();
        // No key exists at all, and every one of these is still the option
        // complaint rather than a zero, which is the order a real server uses.
        assert_eq!(
            f.run(&[b"COPY", b"a", b"b", b"DB", b"99"]),
            "-ERR DB index is out of range\r\n"
        );
        assert_eq!(
            f.run(&[b"COPY", b"a", b"b", b"DB", b"-1"]),
            "-ERR DB index is out of range\r\n"
        );
        assert_eq!(
            f.run(&[b"COPY", b"a", b"b", b"DB", b"x"]),
            "-ERR value is not an integer or out of range\r\n"
        );
        assert_eq!(
            f.run(&[b"COPY", b"a", b"b", b"nonsense"]),
            "-ERR syntax error\r\n"
        );
        assert_eq!(
            f.run(&[b"COPY", b"a", b"a"]),
            "-ERR source and destination objects are the same\r\n"
        );
        // Repeated, reordered and lowercased, and the last DB wins.
        assert_eq!(
            f.run(&[b"COPY", b"a", b"b", b"dB", b"1", b"rEpLaCe", b"db", b"2"]),
            ":0\r\n"
        );
    }

    #[test]
    fn time_is_two_bulk_strings_and_moves() {
        let mut f = Fixture::new();
        let first = f.run(&[b"TIME"]);
        assert!(first.starts_with("*2\r\n$"), "got {first}");
        let parts: Vec<&str> = first.split("\r\n").collect();
        let secs: i64 = parts[2].parse().expect("seconds as decimal text");
        let micros: i64 = parts[4].parse().expect("microseconds as decimal text");
        assert!(secs > 1_700_000_000, "a real wall clock, got {secs}");
        assert!((0..1_000_000).contains(&micros), "got {micros}");
        // The coarse clock the keyspace uses is a cached millisecond that a
        // background tick refreshes, so a TIME built on it would answer the
        // same microsecond twice in a row here.
        assert_ne!(first, f.run(&[b"TIME"]));
    }

    #[test]
    fn a_keyspace_scan_walks_every_key_once() {
        let mut f = Fixture::new();
        for i in 0..500 {
            f.run(&[b"SET", format!("k{i}").as_bytes(), b"v"]);
        }

        let mut seen: Vec<String> = Vec::new();
        let mut cursor = "0".to_owned();
        let mut calls = 0;
        loop {
            let (next, keys) = scan_reply(&f.run(&[b"SCAN", cursor.as_bytes(), b"COUNT", b"32"]));
            seen.extend(keys);
            cursor = next;
            calls += 1;
            assert!(calls < 10_000, "the cursor is not advancing");
            if cursor == "0" {
                break;
            }
        }

        seen.sort();
        seen.dedup();
        assert_eq!(seen.len(), 500, "every key once and only once");
        // And more than one call to get them, or the COUNT is being ignored and
        // the loop above proved nothing about resuming.
        assert!(calls > 1, "500 keys came back in one batch");
    }

    #[test]
    fn a_scan_narrows_by_pattern_and_by_type() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"str", b"v"]);
        f.run(&[b"SADD", b"members", b"a"]);
        f.run(&[b"HSET", b"fields", b"f", b"v"]);

        let all = |f: &mut Fixture, args: &[&[u8]]| {
            let mut out: Vec<String> = Vec::new();
            let mut cursor = "0".to_owned();
            loop {
                let mut line: Vec<&[u8]> = vec![b"SCAN", cursor.as_bytes()];
                line.extend_from_slice(args);
                let (next, keys) = scan_reply(&f.run(&line));
                out.extend(keys);
                cursor = next;
                if cursor == "0" {
                    break;
                }
            }
            out.sort();
            out
        };

        assert_eq!(all(&mut f, &[]), ["fields", "members", "str"]);
        assert_eq!(all(&mut f, &[b"MATCH", b"*e*"]), ["fields", "members"]);
        assert_eq!(all(&mut f, &[b"TYPE", b"set"]), ["members"]);
        // Case insensitive, the same as Redis's own comparison.
        assert_eq!(all(&mut f, &[b"TYPE", b"HASH"]), ["fields"]);
        // A type nothing can hold is not an error, it just matches nothing.
        assert!(all(&mut f, &[b"TYPE", b"list"]).is_empty());
        assert!(all(&mut f, &[b"TYPE", b"banana"]).is_empty());
        // Both filters at once, and they are an and rather than an or.
        assert!(all(&mut f, &[b"MATCH", b"str*", b"TYPE", b"set"]).is_empty());
    }

    #[test]
    fn a_scan_says_what_is_wrong_with_it() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"SCAN", b"nope"]), "-ERR invalid cursor\r\n");
        assert_eq!(f.run(&[b"SCAN", b"-1"]), "-ERR invalid cursor\r\n");
        assert_eq!(f.run(&[b"SCAN", b"0", b"MATCH"]), "-ERR syntax error\r\n");
        assert_eq!(
            f.run(&[b"SCAN", b"0", b"COUNT", b"0"]),
            "-ERR syntax error\r\n"
        );
        assert_eq!(
            f.run(&[b"SCAN", b"0", b"COUNT", b"x"]),
            "-ERR value is not an integer or out of range\r\n"
        );
        assert_eq!(
            f.run(&[b"SCAN", b"0", b"WAT", b"1"]),
            "-ERR syntax error\r\n"
        );
        // A cursor the client made up is a cursor. It resumes somewhere
        // arbitrary and answers whatever is there, which is what Redis does and
        // is the only behaviour that does not need the server to remember every
        // cursor it has handed out.
        assert!(f.run(&[b"SCAN", b"18446744073709551615"]).starts_with("*2"));
    }

    #[test]
    fn keys_and_randomkey_look_at_the_whole_database() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"KEYS", b"*"]), "*0\r\n");
        assert_eq!(f.run(&[b"RANDOMKEY"]), "$-1\r\n");

        for name in ["one", "two", "three"] {
            f.run(&[b"SET", name.as_bytes(), b"v"]);
        }
        assert_eq!(sorted(&f.run(&[b"KEYS", b"*"])), ["one", "three", "two"]);
        assert_eq!(sorted(&f.run(&[b"KEYS", b"t*"])), ["three", "two"]);
        assert_eq!(f.run(&[b"KEYS", b"nothing"]), "*0\r\n");

        for _ in 0..50 {
            let got = f.run(&[b"RANDOMKEY"]);
            assert!(
                ["$3\r\none\r\n", "$3\r\ntwo\r\n", "$5\r\nthree\r\n"].contains(&got.as_str()),
                "got {got}"
            );
        }
    }

    #[test]
    fn a_walk_does_not_answer_keys_that_have_expired() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"alive", b"v"]);
        f.run(&[b"SET", b"dead", b"v", b"PX", b"1"]);
        f.server.db(0).clock_mut().advance(2);
        assert_eq!(
            f.run(&[b"DBSIZE"]),
            ":2\r\n",
            "nothing has collected it yet"
        );

        assert_eq!(f.run(&[b"KEYS", b"*"]), "*1\r\n$5\r\nalive\r\n");
        let (_, keys) = scan_reply(&f.run(&[b"SCAN", b"0", b"COUNT", b"1000"]));
        assert_eq!(keys, ["alive"]);
        for _ in 0..20 {
            assert_eq!(f.run(&[b"RANDOMKEY"]), "$5\r\nalive\r\n");
        }
        // The walk collected it on the way past, which is what makes DBSIZE
        // here answer what Redis answers once its own cycle has been round.
        assert_eq!(f.run(&[b"DBSIZE"]), ":1\r\n");
    }

    #[test]
    fn a_key_deadline_goes_on_and_comes_back_in_all_four_units() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"k", b"v"]);
        assert_eq!(f.run(&[b"TTL", b"k"]), ":-1\r\n", "there and no deadline");
        assert_eq!(f.run(&[b"TTL", b"nosuch"]), ":-2\r\n", "not there at all");

        assert_eq!(f.run(&[b"EXPIRE", b"k", b"100"]), ":1\r\n");
        assert_eq!(f.run(&[b"TTL", b"k"]), ":100\r\n");
        let ms = int(&f.run(&[b"PTTL", b"k"]));
        assert!((99_000..=100_000).contains(&ms), "got {ms}");

        // The absolute pair, derived from the same one number the store kept.
        let at = int(&f.run(&[b"EXPIRETIME", b"k"]));
        let at_ms = int(&f.run(&[b"PEXPIRETIME", b"k"]));
        assert_eq!(at, (at_ms + 500) / 1000);
        assert!(at_ms > 1_700_000_000_000, "an absolute moment, got {at_ms}");

        assert_eq!(f.run(&[b"PERSIST", b"k"]), ":1\r\n");
        assert_eq!(f.run(&[b"TTL", b"k"]), ":-1\r\n");
        assert_eq!(
            f.run(&[b"PERSIST", b"k"]),
            ":0\r\n",
            "nothing to take off the second time"
        );
        assert_eq!(f.run(&[b"PERSIST", b"nosuch"]), ":0\r\n");
        assert_eq!(
            f.run(&[b"GET", b"k"]),
            "$1\r\nv\r\n",
            "and the value went through all of that untouched"
        );
    }

    #[test]
    fn every_type_can_be_given_a_deadline_and_it_is_the_same_deadline() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"str", b"v"]);
        f.run(&[b"SADD", b"set", b"a", b"b"]);
        f.run(&[b"HSET", b"hash", b"f", b"v"]);

        for key in [b"str".as_slice(), b"set", b"hash"] {
            assert_eq!(f.run(&[b"EXPIRE", key, b"100"]), ":1\r\n");
            assert_eq!(f.run(&[b"TTL", key]), ":100\r\n");
        }
        // The body is not touched by any of that, which is the whole reason the
        // deadline lives in the record and the body lives somewhere else.
        assert_eq!(f.run(&[b"SCARD", b"set"]), ":2\r\n");
        assert_eq!(f.run(&[b"HGET", b"hash", b"f"]), "$1\r\nv\r\n");
        assert_eq!(f.run(&[b"GET", b"str"]), "$1\r\nv\r\n");
    }

    #[test]
    fn a_deadline_that_has_already_gone_deletes_the_key_now() {
        let mut f = Fixture::new();
        for key in [b"a".as_slice(), b"b", b"c", b"d"] {
            f.run(&[b"SET", key, b"v"]);
        }
        // Four ways of naming a moment that has passed, and all four are a
        // delete answering 1 rather than an error. Zero is a moment, minus one
        // is a moment, and the hash field commands refuse the negative one.
        assert_eq!(f.run(&[b"EXPIRE", b"a", b"0"]), ":1\r\n");
        assert_eq!(f.run(&[b"EXPIRE", b"b", b"-1"]), ":1\r\n");
        assert_eq!(f.run(&[b"EXPIREAT", b"c", b"1"]), ":1\r\n");
        assert_eq!(f.run(&[b"PEXPIREAT", b"d", b"1"]), ":1\r\n");
        assert_eq!(f.run(&[b"DBSIZE"]), ":0\r\n");
        assert_eq!(
            f.run(&[b"EXPIRE", b"a", b"100"]),
            ":0\r\n",
            "and the key really went, so there is nothing to put a deadline on"
        );
    }

    #[test]
    fn the_four_conditions_decide_whether_the_deadline_moves() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"k", b"v"]);

        assert_eq!(f.run(&[b"EXPIRE", b"k", b"100", b"XX"]), ":0\r\n");
        assert_eq!(f.run(&[b"TTL", b"k"]), ":-1\r\n", "and XX left it alone");
        assert_eq!(f.run(&[b"EXPIRE", b"k", b"100", b"GT"]), ":0\r\n");
        assert_eq!(
            f.run(&[b"EXPIRE", b"k", b"100", b"LT"]),
            ":1\r\n",
            "no deadline reads as infinitely far away, so LT passes where GT fails"
        );

        assert_eq!(f.run(&[b"EXPIRE", b"k", b"50", b"NX"]), ":0\r\n");
        assert_eq!(f.run(&[b"EXPIRE", b"k", b"50", b"GT"]), ":0\r\n");
        assert_eq!(f.run(&[b"TTL", b"k"]), ":100\r\n");
        assert_eq!(f.run(&[b"EXPIRE", b"k", b"50", b"LT"]), ":1\r\n");
        assert_eq!(f.run(&[b"EXPIRE", b"k", b"200", b"GT"]), ":1\r\n");
        assert_eq!(f.run(&[b"TTL", b"k"]), ":200\r\n");

        // The condition is answered before the past check, so this is a 0 and
        // the key survives. The other order would delete it.
        assert_eq!(f.run(&[b"EXPIRE", b"k", b"0", b"NX"]), ":0\r\n");
        assert_eq!(f.run(&[b"EXISTS", b"k"]), ":1\r\n");
        assert_eq!(f.run(&[b"EXPIRE", b"k", b"0", b"XX"]), ":1\r\n");
        assert_eq!(f.run(&[b"EXISTS", b"k"]), ":0\r\n", "and XX let it through");
    }

    #[test]
    fn the_conditions_are_a_set_and_not_a_keyword() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"k", b"v"]);

        assert_eq!(f.run(&[b"EXPIRE", b"k", b"100", b"nx"]), ":1\r\n");
        assert_eq!(
            f.run(&[b"EXPIRE", b"k", b"100", b"nx", b"nx"]),
            ":0\r\n",
            "the same keyword twice means it once, and NX now has a deadline to fail on"
        );

        // XX with LT is the one pair that is not either of them on its own: LT
        // alone would accept a key with no deadline and this does not.
        assert_eq!(f.run(&[b"EXPIRE", b"k", b"200", b"xx", b"gt"]), ":1\r\n");
        assert_eq!(f.run(&[b"TTL", b"k"]), ":200\r\n");
        assert_eq!(f.run(&[b"EXPIRE", b"k", b"100", b"gt", b"xx"]), ":0\r\n");
        assert_eq!(f.run(&[b"EXPIRE", b"k", b"100", b"XX", b"LT"]), ":1\r\n");
        assert_eq!(f.run(&[b"TTL", b"k"]), ":100\r\n");
        f.run(&[b"PERSIST", b"k"]);
        assert_eq!(
            f.run(&[b"EXPIRE", b"k", b"100", b"XX", b"LT"]),
            ":0\r\n",
            "where LT on its own would have taken it"
        );
        assert_eq!(f.run(&[b"EXPIRE", b"k", b"100", b"LT"]), ":1\r\n");
    }

    #[test]
    fn a_key_is_gone_once_its_moment_passes() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"k", b"v"]);
        f.run(&[b"EXPIRE", b"k", b"100"]);

        let at = int(&f.run(&[b"PEXPIRETIME", b"k"]));
        f.server.set_clock_ms(at as u64 + 1);
        assert_eq!(f.run(&[b"GET", b"k"]), "$-1\r\n");
        assert_eq!(f.run(&[b"TTL", b"k"]), ":-2\r\n");
        assert_eq!(f.run(&[b"EXISTS", b"k"]), ":0\r\n");
        assert_eq!(f.run(&[b"DBSIZE"]), ":0\r\n");
    }

    #[test]
    fn the_expiry_commands_refuse_what_a_real_server_refuses() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"k", b"v"]);
        for (bad, want) in [
            (
                &[b"EXPIRE".as_slice(), b"k", b"soon"][..],
                "-ERR value is not an integer or out of range\r\n",
            ),
            (
                &[b"EXPIRE", b"k", b"100", b"MAYBE"],
                "-ERR Unsupported option MAYBE\r\n",
            ),
            (
                &[b"EXPIRE", b"k", b"100", b"NX", b"XX"],
                "-ERR NX and XX, GT or LT options at the same time are not compatible\r\n",
            ),
            (
                &[b"EXPIRE", b"k", b"100", b"NX", b"GT"],
                "-ERR NX and XX, GT or LT options at the same time are not compatible\r\n",
            ),
            (
                &[b"EXPIRE", b"k", b"100", b"GT", b"LT", b"GT"],
                "-ERR GT and LT options at the same time are not compatible\r\n",
            ),
            // Seconds that overflow when multiplied into milliseconds. Every
            // message names the command it came from.
            (
                &[b"EXPIRE", b"k", b"9223372036854775807"],
                "-ERR invalid expire time in 'expire' command\r\n",
            ),
            (
                &[b"EXPIREAT", b"k", b"9223372036854775807"],
                "-ERR invalid expire time in 'expireat' command\r\n",
            ),
            (
                &[b"PEXPIRE", b"k", b"9223372036854775807"],
                "-ERR invalid expire time in 'pexpire' command\r\n",
            ),
        ] {
            assert_eq!(f.run(bad), want, "for {bad:?}");
        }
        assert_eq!(
            f.run(&[b"TTL", b"k"]),
            ":-1\r\n",
            "and none of those put a deadline on anything"
        );

        // The one of the four that has no arithmetic to overflow. Redis takes
        // it and holds the number as given, and a record here holds forty six
        // bits, so it lands in the year 4199 instead. D-17.
        assert_eq!(
            f.run(&[b"PEXPIREAT", b"k", b"9223372036854775807"]),
            ":1\r\n"
        );
        assert_eq!(f.run(&[b"PEXPIRETIME", b"k"]), ":70368744177663\r\n");
    }

    #[test]
    fn flushing_empties_this_database_or_every_one_of_them() {
        let mut f = Fixture::new();
        f.run(&[b"SELECT", b"0"]);
        f.run(&[b"MSET", b"a", b"1", b"b", b"2"]);
        f.run(&[b"SELECT", b"1"]);
        f.run(&[b"SET", b"c", b"3"]);
        assert_eq!(f.run(&[b"DBSIZE"]), ":1\r\n");
        // ASYNC and SYNC are both taken and neither changes anything, since the
        // keyspace is empty before the OK goes out either way.
        assert_eq!(f.run(&[b"FLUSHDB", b"async"]), "+OK\r\n");
        assert_eq!(f.run(&[b"DBSIZE"]), ":0\r\n");
        // Only database one was emptied.
        f.run(&[b"SELECT", b"0"]);
        assert_eq!(f.run(&[b"DBSIZE"]), ":2\r\n");
        assert_eq!(f.run(&[b"FLUSHALL", b"SYNC"]), "+OK\r\n");
        assert_eq!(f.run(&[b"DBSIZE"]), ":0\r\n");
        f.run(&[b"SELECT", b"1"]);
        assert_eq!(f.run(&[b"DBSIZE"]), ":0\r\n");
        // Anything else after the name is a syntax error, and so is a third
        // argument even when the second one is a word we take.
        assert_eq!(f.run(&[b"FLUSHALL", b"nope"]), "-ERR syntax error\r\n");
        assert_eq!(
            f.run(&[b"FLUSHDB", b"sync", b"sync"]),
            "-ERR syntax error\r\n"
        );
    }

    #[test]
    fn the_script_cache_and_the_library_set_answer_for_being_empty() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"SCRIPT", b"FLUSH"]), "+OK\r\n");
        assert_eq!(f.run(&[b"SCRIPT", b"FLUSH", b"async"]), "+OK\r\n");
        assert_eq!(f.run(&[b"FUNCTION", b"FLUSH", b"SYNC"]), "+OK\r\n");
        // Nothing is cached, so nothing is there, one answer per hash asked
        // about.
        assert_eq!(
            f.run(&[b"SCRIPT", b"EXISTS", b"aaaa", b"bbbb"]),
            "*2\r\n:0\r\n:0\r\n"
        );
        assert_eq!(f.run(&[b"FUNCTION", b"LIST"]), "*0\r\n");
        assert_eq!(
            f.run(&[b"FUNCTION", b"LIST", b"LIBRARYNAME", b"x", b"WITHCODE"]),
            "*0\r\n"
        );
        assert_eq!(
            f.run(&[b"FUNCTION", b"DELETE", b"nosuch"]),
            "-ERR Library not found\r\n"
        );

        // Redis's two messages here are its own, one per container, and one of
        // them reads like a typo.
        assert_eq!(
            f.run(&[b"SCRIPT", b"FLUSH", b"nope"]),
            "-ERR SCRIPT FLUSH only support SYNC|ASYNC option\r\n"
        );
        assert_eq!(
            f.run(&[b"FUNCTION", b"FLUSH", b"nope"]),
            "-ERR FUNCTION FLUSH only supports SYNC|ASYNC option\r\n"
        );
        // A second argument after the mode is the generic one instead, because
        // the count is checked before the word is looked at.
        assert_eq!(
            f.run(&[b"FUNCTION", b"FLUSH", b"sync", b"sync"]),
            "-ERR unknown subcommand or wrong number of arguments for 'flush'. Try FUNCTION HELP.\r\n"
        );
        assert_eq!(
            f.run(&[b"FUNCTION", b"LIST", b"bogus"]),
            "-ERR Unknown argument bogus\r\n"
        );
        assert_eq!(
            f.run(&[b"SCRIPT", b"EXISTS"]),
            "-ERR wrong number of arguments for 'script|exists' command\r\n"
        );

        // The ones that need an interpreter are not here, and say so rather
        // than answering OK to a load that loaded nothing.
        assert_eq!(
            f.run(&[b"SCRIPT", b"LOAD", b"return 1"]),
            "-ERR unknown subcommand 'LOAD'. Try SCRIPT HELP.\r\n"
        );
        assert_eq!(
            f.run(&[b"FUNCTION", b"STATS"]),
            "-ERR unknown subcommand 'STATS'. Try FUNCTION HELP.\r\n"
        );
    }

    #[test]
    fn a_counter_is_an_integer_and_not_a_string_of_digits() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"INCR", b"c"]), ":1\r\n");
        assert_eq!(f.run(&[b"INCRBY", b"c", b"41"]), ":42\r\n");
        assert_eq!(f.run(&[b"DECRBY", b"c", b"2"]), ":40\r\n");
        // Read back as a string it is still an integer, written out as digits
        // only because somebody asked for them.
        assert_eq!(f.run(&[b"GET", b"c"]), "$2\r\n40\r\n");
        assert_eq!(f.run(&[b"INCRBYFLOAT", b"c", b"0.5"]), "$4\r\n40.5\r\n");
        // A counter that is not a number is the error the store raises and this
        // layer only spells, which is the whole point of the split.
        f.run(&[b"SET", b"k", b"hello"]);
        assert_eq!(
            f.run(&[b"INCR", b"k"]),
            "-ERR value is not an integer or out of range\r\n"
        );
        assert_eq!(
            f.run(&[b"INCRBYFLOAT", b"c", b"inf"]),
            "-ERR increment would produce NaN or Infinity\r\n"
        );
    }

    /// Every one of these was read off a running 8.8. They are the answers a
    /// client library's own test suite checks, and the shapes are not
    /// guessable: `DIGEST` is hexadecimal in a bulk string, `MSETEX` is an
    /// integer, `INCREX` is a pair.
    #[test]
    fn the_newer_commands_reply_in_the_shapes_a_real_server_sends() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"SET", b"k", b"hello"]), "+OK\r\n");
        // The same digest a real 8.8 answers for the same five bytes, which is
        // what makes `IFDEQ` usable against a mixed deployment.
        assert_eq!(f.run(&[b"DIGEST", b"k"]), "$16\r\n9555e8555c62dcfd\r\n");
        assert_eq!(f.run(&[b"DIGEST", b"nosuch"]), "$-1\r\n");
        assert_eq!(f.run(&[b"MSETEX", b"1", b"a", b"1"]), ":1\r\n");
        assert_eq!(f.run(&[b"MSETEX", b"1", b"a", b"2", b"NX"]), ":0\r\n");
        assert_eq!(f.run(&[b"GET", b"a"]), "$1\r\n1\r\n");
        assert_eq!(f.run(&[b"INCREX", b"n"]), "*2\r\n:1\r\n:1\r\n");
        assert_eq!(
            f.run(&[b"INCREX", b"n", b"BYINT", b"5", b"UBOUND", b"3"]),
            "*2\r\n:1\r\n:0\r\n",
            "a refused increment reports the value it left alone and applied nothing"
        );
        assert_eq!(
            f.run(&[
                b"INCREX",
                b"n",
                b"BYINT",
                b"5",
                b"UBOUND",
                b"3",
                b"SATURATE"
            ]),
            "*2\r\n:3\r\n:2\r\n"
        );
        assert_eq!(f.run(&[b"DELEX", b"a", b"IFEQ", b"2"]), ":0\r\n");
        assert_eq!(f.run(&[b"DELEX", b"a", b"IFEQ", b"1"]), ":1\r\n");
    }

    #[test]
    fn the_same_answers_come_out_in_resp3_spelling() {
        let mut f = Fixture::new();
        assert!(f.run(&[b"HELLO", b"3"]).starts_with("%7\r\n"));
        assert_eq!(f.run(&[b"GET", b"nosuch"]), "_\r\n");
        // A float counter is a double on RESP3 and the digits in a bulk string
        // on RESP2, and `INCRBYFLOAT` is a bulk string on both.
        assert_eq!(
            f.run(&[b"INCREX", b"c", b"BYFLOAT", b"1.5"]),
            "*2\r\n,1.5\r\n,1.5\r\n"
        );
        assert_eq!(f.run(&[b"INCRBYFLOAT", b"f", b"2.5"]), "$3\r\n2.5\r\n");
        // `RESET` puts the protocol back, which is the part that is easy to
        // miss and leaves a pooled connection speaking the wrong one.
        assert_eq!(f.run(&[b"RESET"]), "+RESET\r\n");
        assert_eq!(f.run(&[b"GET", b"nosuch"]), "$-1\r\n");
    }

    #[test]
    fn a_command_nobody_has_heard_of_is_an_error_and_not_a_closed_socket() {
        let mut f = Fixture::new();
        let (flow, reply) = f.flow(&[b"NOPE", b"a", b"b"]);
        assert_eq!(flow, Flow::Continue);
        assert_eq!(
            reply,
            "-ERR unknown command 'NOPE', with args beginning with: 'a' 'b' \r\n"
        );
        // A name with a line ending in it cannot write its own frame into the
        // stream, which is the reason the error writer maps them to spaces.
        let reply = f.run(&[b"NO\r\n+PONG\r\nPE"]);
        assert_eq!(reply.matches("\r\n").count(), 1);
    }

    #[test]
    fn arity_is_checked_before_the_command_is() {
        let mut f = Fixture::new();
        assert_eq!(
            f.run(&[b"GET"]),
            "-ERR wrong number of arguments for 'get' command\r\n"
        );
        assert_eq!(
            f.run(&[b"MSET", b"k"]),
            "-ERR wrong number of arguments for 'mset' command\r\n"
        );
        // The table says `PING` takes one or more and a real server then
        // refuses three, which is the sort of thing that only shows up against
        // the real thing.
        assert_eq!(
            f.run(&[b"PING", b"a", b"b"]),
            "-ERR wrong number of arguments for 'ping' command\r\n"
        );
        assert_eq!(f.run(&[b"PING"]), "+PONG\r\n");
        assert_eq!(f.run(&[b"PING", b"hi"]), "$2\r\nhi\r\n");
        // `DELEX` takes two or four and nothing between.
        assert_eq!(
            f.run(&[b"DELEX", b"k", b"IFEQ"]),
            "-ERR wrong number of arguments for 'delex' command\r\n"
        );
    }

    /// The option rules, all of them measured against 8.8 rather than read off
    /// the documentation. The surprising one is that `SET` accepts the same
    /// keyword twice and `INCREX` does not.
    #[test]
    fn the_option_combinations_are_the_ones_a_real_server_accepts() {
        let mut f = Fixture::new();
        let syntax = "-ERR syntax error\r\n";
        assert_eq!(f.run(&[b"SET", b"k", b"v", b"NX", b"XX"]), syntax);
        assert_eq!(f.run(&[b"SET", b"k", b"v", b"NX", b"IFEQ", b"a"]), syntax);
        assert_eq!(
            f.run(&[b"SET", b"k", b"v", b"KEEPTTL", b"EX", b"5"]),
            syntax
        );
        assert_eq!(
            f.run(&[b"SET", b"k", b"v", b"EX", b"5", b"PX", b"5"]),
            syntax
        );
        assert_eq!(f.run(&[b"SET", b"k", b"v", b"PERSIST"]), syntax);
        // Twice is fine, and the last one wins.
        assert_eq!(
            f.run(&[b"SET", b"k", b"v", b"EX", b"5", b"EX", b"100"]),
            "+OK\r\n"
        );
        assert_eq!(f.run(&[b"SET", b"k", b"v", b"XX", b"XX"]), "+OK\r\n");
        assert_eq!(f.run(&[b"SET", b"k", b"v", b"GET", b"GET"]), "$1\r\nv\r\n");
        // `INCREX` refuses what `SET` allows.
        assert_eq!(
            f.run(&[b"INCREX", b"n", b"BYINT", b"1", b"BYINT", b"2"]),
            syntax
        );
        assert_eq!(
            f.run(&[b"INCREX", b"n", b"ENX"]),
            "-ERR ENX flag requires an expiration\r\n"
        );
        assert_eq!(
            f.run(&[b"INCREX", b"n", b"UBOUND", b"abc"]),
            "-ERR UBOUND is not an integer or out of range\r\n"
        );
        assert_eq!(
            f.run(&[b"INCREX", b"n", b"LBOUND", b"10", b"UBOUND", b"5"]),
            "-ERR LBOUND can't be greater than UBOUND\r\n"
        );
        assert_eq!(
            f.run(&[b"LCS", b"a", b"b", b"LEN", b"IDX"]),
            "-ERR If you want both the length and indexes, please just use IDX.\r\n"
        );
    }

    /// Where the expiration rules bite. The one worth the test is `GETEX` on a
    /// key that is not there, which answers null without ever looking at the
    /// expiration it was given.
    #[test]
    fn the_expiry_rules_are_redis_own() {
        let mut f = Fixture::new();
        let bad = "-ERR invalid expire time in 'set' command\r\n";
        assert_eq!(f.run(&[b"SET", b"k", b"v", b"EX", b"0"]), bad);
        assert_eq!(f.run(&[b"SET", b"k", b"v", b"EX", b"-1"]), bad);
        assert_eq!(f.run(&[b"SET", b"k", b"v", b"EXAT", b"0"]), bad);
        assert_eq!(
            f.run(&[b"SET", b"k", b"v", b"EX", b"9999999999999999"]),
            bad
        );
        assert_eq!(
            f.run(&[b"SET", b"k", b"v", b"PX", b"99999999999999999999"]),
            "-ERR value is not an integer or out of range\r\n"
        );
        assert_eq!(
            f.run(&[b"SETEX", b"k", b"0", b"v"]),
            "-ERR invalid expire time in 'setex' command\r\n"
        );
        assert_eq!(f.run(&[b"GETEX", b"nosuch", b"EX", b"0"]), "$-1\r\n");
        assert_eq!(f.run(&[b"GETEX", b"nosuch", b"EX", b"abc"]), "$-1\r\n");
        assert_eq!(
            f.run(&[b"GETEX", b"nosuch", b"KEEPTTL"]),
            "-ERR syntax error\r\n",
            "the option list is still checked before the key is looked up"
        );
        // A deadline in the past is accepted and the key goes with it.
        assert_eq!(f.run(&[b"SET", b"k", b"v"]), "+OK\r\n");
        assert_eq!(f.run(&[b"SET", b"k", b"v", b"EXAT", b"1"]), "+OK\r\n");
        assert_eq!(f.run(&[b"GET", b"k"]), "$-1\r\n");
    }

    #[test]
    fn mset_takes_its_pairs_from_the_read_buffer() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"MSET", b"a", b"1", b"b", b"2"]), "+OK\r\n");
        assert_eq!(
            f.run(&[b"MGET", b"a", b"b", b"nosuch"]),
            "*3\r\n$1\r\n1\r\n$1\r\n2\r\n$-1\r\n"
        );
        assert_eq!(f.run(&[b"MSETNX", b"b", b"9", b"c", b"3"]), ":0\r\n");
        assert_eq!(f.run(&[b"MSETNX", b"c", b"3", b"d", b"4"]), ":1\r\n");
        assert_eq!(
            f.run(&[b"MSETEX", b"2", b"e", b"5"]),
            "-ERR wrong number of key-value pairs\r\n"
        );
        assert_eq!(
            f.run(&[b"MSETEX", b"0", b"e", b"5"]),
            "-ERR invalid numkeys value\r\n"
        );
        assert_eq!(
            f.run(&[b"MSETEX", b"abc", b"e", b"5"]),
            "-ERR invalid numkeys value\r\n"
        );
    }

    #[test]
    fn lcs_answers_the_length_the_string_and_the_runs() {
        let mut f = Fixture::new();
        f.run(&[b"MSET", b"a", b"ohmytext", b"b", b"mynewtext"]);
        assert_eq!(f.run(&[b"LCS", b"a", b"b"]), "$6\r\nmytext\r\n");
        assert_eq!(f.run(&[b"LCS", b"a", b"b", b"LEN"]), ":6\r\n");
        assert_eq!(
            f.run(&[b"LCS", b"a", b"b", b"IDX", b"MINMATCHLEN", b"4"]),
            "*4\r\n$7\r\nmatches\r\n*1\r\n*2\r\n*2\r\n:4\r\n:7\r\n*2\r\n:5\r\n:8\r\n$3\r\nlen\r\n:6\r\n"
        );
        // Without `IDX` the two options that only mean something with it are
        // accepted and ignored, which is what a real server does.
        assert_eq!(
            f.run(&[b"LCS", b"a", b"b", b"MINMATCHLEN", b"4", b"WITHMATCHLEN"]),
            "$6\r\nmytext\r\n"
        );
    }

    #[test]
    fn select_moves_the_connection_and_the_databases_stay_apart() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"k", b"zero"]);
        assert_eq!(f.run(&[b"SELECT", b"4"]), "+OK\r\n");
        assert_eq!(f.run(&[b"GET", b"k"]), "$-1\r\n");
        f.run(&[b"SET", b"k", b"four"]);
        assert_eq!(f.run(&[b"SELECT", b"0"]), "+OK\r\n");
        assert_eq!(f.run(&[b"GET", b"k"]), "$4\r\nzero\r\n");
        assert_eq!(
            f.run(&[b"SELECT", b"99"]),
            "-ERR DB index is out of range\r\n"
        );
        assert_eq!(
            f.run(&[b"SELECT", b"-1"]),
            "-ERR DB index is out of range\r\n"
        );
        assert_eq!(
            f.run(&[b"SELECT", b"abc"]),
            "-ERR value is not an integer or out of range\r\n"
        );
        // `RESET` brings it back to zero.
        f.run(&[b"SELECT", b"4"]);
        f.run(&[b"RESET"]);
        assert_eq!(f.run(&[b"GET", b"k"]), "$4\r\nzero\r\n");
    }

    #[test]
    fn hello_agrees_on_a_protocol_and_refuses_the_ones_that_do_not_exist() {
        let mut f = Fixture::new();
        let reply = f.run(&[b"HELLO"]);
        assert!(reply.starts_with("*14\r\n"), "{reply}");
        assert!(reply.contains("$5\r\nredis\r\n"), "{reply}");
        assert!(reply.contains("$5\r\n8.8.0\r\n"), "{reply}");
        assert!(
            reply.contains(":7\r\n"),
            "the connection id is in there: {reply}"
        );
        assert_eq!(
            f.run(&[b"HELLO", b"4"]),
            "-NOPROTO unsupported protocol version\r\n"
        );
        assert_eq!(
            f.run(&[b"HELLO", b"abc"]),
            "-ERR Protocol version is not an integer or out of range\r\n"
        );
        assert_eq!(
            f.run(&[b"HELLO", b"3", b"SETNAME"]),
            "-ERR Syntax error in HELLO option 'SETNAME'\r\n"
        );
        assert!(
            f.run(&[b"HELLO", b"3", b"SETNAME", b"bob"])
                .starts_with("%7\r\n")
        );
        assert_eq!(f.session.name(), b"bob");
        f.run(&[b"RESET"]);
        assert_eq!(f.session.name(), b"");
    }

    #[test]
    fn command_describes_this_server_in_the_shape_a_driver_reads() {
        let mut f = Fixture::new();
        let count = format!(":{}\r\n", COMMANDS.len());
        assert_eq!(f.run(&[b"COMMAND", b"COUNT"]), count);
        let info = f.run(&[b"COMMAND", b"INFO", b"get"]);
        assert_eq!(
            info,
            "*1\r\n*10\r\n$3\r\nget\r\n:2\r\n*2\r\n+readonly\r\n+fast\r\n:1\r\n:1\r\n:1\r\n\
             *3\r\n+@read\r\n+@string\r\n+@fast\r\n*0\r\n*0\r\n*0\r\n"
        );
        // A null in the list, and the plain one: `$-1` and not `*-1`.
        assert_eq!(f.run(&[b"COMMAND", b"INFO", b"nosuch"]), "*1\r\n$-1\r\n");
        assert_eq!(
            f.run(&[b"COMMAND", b"LIST", b"FILTERBY", b"PATTERN", b"getr*"]),
            "*1\r\n$8\r\ngetrange\r\n"
        );
        assert_eq!(
            f.run(&[b"COMMAND", b"NOPE"]),
            "-ERR unknown subcommand 'NOPE'. Try COMMAND HELP.\r\n"
        );
    }

    /// A cluster aware client asks this question and then routes on the
    /// answer, so `MSETEX`, whose keys are not where the table says, is the one
    /// that matters.
    #[test]
    fn command_getkeys_finds_the_keys_including_the_hidden_ones() {
        let mut f = Fixture::new();
        assert_eq!(
            f.run(&[b"COMMAND", b"GETKEYS", b"get", b"k"]),
            "*1\r\n$1\r\nk\r\n"
        );
        assert_eq!(
            f.run(&[b"COMMAND", b"GETKEYS", b"mset", b"a", b"1", b"b", b"2"]),
            "*2\r\n$1\r\na\r\n$1\r\nb\r\n"
        );
        assert_eq!(
            f.run(&[
                b"COMMAND", b"GETKEYS", b"msetex", b"2", b"a", b"1", b"b", b"2"
            ]),
            "*2\r\n$1\r\na\r\n$1\r\nb\r\n"
        );
        assert_eq!(
            f.run(&[b"COMMAND", b"GETKEYS", b"ping"]),
            "-ERR The command has no key arguments\r\n"
        );
        assert_eq!(
            f.run(&[b"COMMAND", b"GETKEYS", b"set"]),
            "-ERR Invalid number of arguments specified for command\r\n"
        );
    }

    #[test]
    fn config_answers_what_it_can_and_refuses_what_it_cannot() {
        let mut f = Fixture::new();
        assert_eq!(
            f.run(&[b"CONFIG", b"GET", b"maxmemory"]),
            "*2\r\n$9\r\nmaxmemory\r\n$1\r\n0\r\n"
        );
        // A pattern matches more than one, and a setting two patterns both ask
        // for is still sent once.
        let both = f.run(&[b"CONFIG", b"GET", b"maxmemory*", b"maxmemory"]);
        assert!(both.starts_with("*6\r\n"), "{both}");
        assert_eq!(f.run(&[b"CONFIG", b"GET", b"nosuch"]), "*0\r\n");
        assert_eq!(f.run(&[b"CONFIG", b"SET", b"appendonly", b"no"]), "+OK\r\n");
        assert_eq!(
            f.run(&[b"CONFIG", b"SET", b"appendonly", b"yes"]),
            "-ERR CONFIG SET failed (possibly related to argument 'appendonly') - can't set immutable config\r\n"
        );
        assert_eq!(
            f.run(&[b"CONFIG", b"SET", b"nosuch", b"1"]),
            "-ERR Unknown option or number of arguments for CONFIG SET - 'nosuch'\r\n"
        );
        assert_eq!(
            f.run(&[b"CONFIG", b"GET"]),
            "-ERR wrong number of arguments for 'config|get' command\r\n"
        );
        // Too few arguments and an odd number of them are different
        // complaints, which is the sort of thing only the real server tells
        // you.
        assert_eq!(
            f.run(&[b"CONFIG", b"SET", b"appendonly"]),
            "-ERR wrong number of arguments for 'config|set' command\r\n"
        );
        assert_eq!(
            f.run(&[b"CONFIG", b"SET", b"appendonly", b"no", b"maxmemory"]),
            "-ERR syntax error\r\n"
        );
        assert_eq!(f.run(&[b"CONFIG", b"RESETSTAT"]), "+OK\r\n");
        assert_eq!(
            f.run(&[b"CONFIG", b"REWRITE"]),
            "-ERR The server is running without a config file\r\n"
        );
    }

    #[test]
    fn the_eviction_policy_reads_back_what_was_written_to_it() {
        let mut f = Fixture::new();
        assert_eq!(
            f.run(&[b"CONFIG", b"GET", b"maxmemory-policy"]),
            "*2\r\n$16\r\nmaxmemory-policy\r\n$10\r\nnoeviction\r\n"
        );
        assert_eq!(
            f.run(&[b"CONFIG", b"SET", b"maxmemory-policy", b"AllKeys-LFU"]),
            "+OK\r\n",
            "the name is matched without regard to case, like every other one"
        );
        assert_eq!(
            f.run(&[b"CONFIG", b"GET", b"maxmemory-policy"]),
            "*2\r\n$16\r\nmaxmemory-policy\r\n$11\r\nallkeys-lfu\r\n"
        );
        // And INFO agrees with CONFIG, which it did not when it was a literal.
        assert!(
            f.run(&[b"INFO", b"memory"])
                .contains("maxmemory_policy:allkeys-lfu"),
            "INFO and CONFIG disagree about the policy"
        );
        // The refusal names every legal value in the order the real server's
        // enum table lists them, because a client comparing the message compares
        // the whole string.
        assert_eq!(
            f.run(&[b"CONFIG", b"SET", b"maxmemory-policy", b"garbage"]),
            "-ERR CONFIG SET failed (possibly related to argument 'maxmemory-policy') - argument(s) must be one of the following: volatile-lru, volatile-lfu, volatile-random, volatile-ttl, volatile-lrm, allkeys-lru, allkeys-lfu, allkeys-random, allkeys-lrm, noeviction\r\n"
        );
        // A bad pair leaves the good one in the same command alone, and the
        // policy is checked by the same pass that checks the numbers.
        assert_eq!(
            f.run(&[b"CONFIG", b"GET", b"maxmemory-policy"]),
            "*2\r\n$16\r\nmaxmemory-policy\r\n$11\r\nallkeys-lfu\r\n"
        );
        f.run(&[
            b"CONFIG",
            b"SET",
            b"hash-max-listpack-entries",
            b"7",
            b"maxmemory-policy",
            b"nonsense",
        ]);
        assert_eq!(
            f.run(&[b"CONFIG", b"GET", b"hash-max-listpack-entries"]),
            "*2\r\n$25\r\nhash-max-listpack-entries\r\n$3\r\n512\r\n"
        );
    }

    #[test]
    fn the_three_eviction_numbers_read_back_too() {
        let mut f = Fixture::new();
        for (name, default, set) in [
            ("maxmemory-samples", "5", "12"),
            ("lfu-log-factor", "10", "3"),
            ("lfu-decay-time", "1", "60"),
        ] {
            let get = || {
                format!(
                    "*2\r\n${}\r\n{name}\r\n${}\r\n{default}\r\n",
                    name.len(),
                    default.len()
                )
            };
            assert_eq!(f.run(&[b"CONFIG", b"GET", name.as_bytes()]), get());
            assert_eq!(
                f.run(&[b"CONFIG", b"SET", name.as_bytes(), set.as_bytes()]),
                "+OK\r\n"
            );
            assert_eq!(
                f.run(&[b"CONFIG", b"GET", name.as_bytes()]),
                format!(
                    "*2\r\n${}\r\n{name}\r\n${}\r\n{set}\r\n",
                    name.len(),
                    set.len()
                )
            );
            // A number that is not a number is refused with the same sentence
            // every other number gets, which names the setting the client typed.
            assert_eq!(
                f.run(&[b"CONFIG", b"SET", name.as_bytes(), b"soon"]),
                format!(
                    "-ERR CONFIG SET failed (possibly related to argument '{name}') - argument couldn't be parsed into an integer\r\n"
                )
            );
        }
    }

    #[test]
    fn the_memory_limit_reads_back_in_bytes_whatever_the_unit_was() {
        let mut f = Fixture::new();
        assert_eq!(
            f.run(&[b"CONFIG", b"GET", b"maxmemory"]),
            "*2\r\n$9\r\nmaxmemory\r\n$1\r\n0\r\n",
            "no limit is the default"
        );
        // The pairing is Redis's and it is a trap: the bare letter is a power of
        // ten and the one with the b is a power of two.
        for (typed, bytes) in [
            (&b"1024"[..], "1024"),
            (b"1k", "1000"),
            (b"1kb", "1024"),
            (b"1M", "1000000"),
            (b"1Mb", "1048576"),
            (b"1gb", "1073741824"),
            (b"100mb", "104857600"),
        ] {
            assert_eq!(f.run(&[b"CONFIG", b"SET", b"maxmemory", typed]), "+OK\r\n");
            assert_eq!(
                f.run(&[b"CONFIG", b"GET", b"maxmemory"]),
                format!("*2\r\n$9\r\nmaxmemory\r\n${}\r\n{bytes}\r\n", bytes.len()),
                "set {}",
                String::from_utf8_lossy(typed)
            );
        }
        assert!(
            f.run(&[b"INFO", b"memory"]).contains("maxmemory:104857600"),
            "the report agrees with the setting"
        );

        // A unit nobody has heard of, and a negative number, which is not a very
        // large one however it is spelled.
        for bad in [&b"1tb"[..], b"-1", b"", b"lots"] {
            assert_eq!(
                f.run(&[b"CONFIG", b"SET", b"maxmemory", bad]),
                "-ERR CONFIG SET failed (possibly related to argument 'maxmemory') - argument must be a memory value\r\n",
                "refused {}",
                String::from_utf8_lossy(bad)
            );
        }
        assert!(
            f.run(&[b"INFO", b"memory"]).contains("maxmemory:104857600"),
            "and the refusal left the old one alone"
        );
    }

    #[test]
    fn a_write_is_refused_when_there_is_no_room_and_nothing_to_evict() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"here", b"already"]);
        // A byte, which is under what an empty server holds, so nothing this
        // command could do would get it under. The default policy is
        // `noeviction`, so nothing is what it does.
        f.run(&[b"CONFIG", b"SET", b"maxmemory", b"1"]);
        assert_eq!(
            f.run(&[b"SET", b"k", b"v"]),
            "-OOM command not allowed when used memory > 'maxmemory'.\r\n"
        );
        assert_eq!(
            f.run(&[b"LPUSH", b"l", b"v"]),
            "-OOM command not allowed when used memory > 'maxmemory'.\r\n"
        );
        // Reading is allowed, and so is the one thing that would help.
        assert_eq!(f.run(&[b"GET", b"here"]), "$7\r\nalready\r\n");
        assert_eq!(f.run(&[b"DEL", b"here"]), ":1\r\n");
        assert!(f.run(&[b"INFO", b"stats"]).contains("evicted_keys:0"));

        // Taking the limit away lets the write through again.
        f.run(&[b"CONFIG", b"SET", b"maxmemory", b"0"]);
        assert_eq!(f.run(&[b"SET", b"k", b"v"]), "+OK\r\n");
    }

    #[test]
    fn an_allkeys_policy_makes_room_instead_of_refusing() {
        let mut f = Fixture::new();
        let val = vec![b'v'; 256];
        for i in 0..24000u32 {
            let k = format!("key:{i:08}");
            f.run(&[b"SET", k.as_bytes(), &val]);
        }
        let full = f.server.memory_bytes();
        assert!(
            full > 3 * 1024 * 1024,
            "the arena is several segments: {full}"
        );

        // Two megabytes under what it is holding, which is one segment's worth,
        // so getting there means giving a whole segment back and not just
        // dropping a few records.
        let limit = full - 2 * 1024 * 1024;
        f.run(&[b"CONFIG", b"SET", b"maxmemory-policy", b"allkeys-lru"]);
        f.run(&[
            b"CONFIG",
            b"SET",
            b"maxmemory",
            limit.to_string().as_bytes(),
        ]);

        // Writes keep working the whole way down. The budget means one command
        // does not do it all, so this runs until the server has settled and
        // checks that nothing was refused on the way.
        for i in 0..2000u32 {
            let k = format!("new:{i:08}");
            assert_eq!(
                f.run(&[b"SET", k.as_bytes(), &val]),
                "+OK\r\n",
                "write {i} was refused"
            );
            f.server.refresh_memory();
            if f.server.memory_bytes() <= limit {
                break;
            }
        }
        assert!(
            f.server.memory_bytes() <= limit,
            "it never got under: {} against {limit}",
            f.server.memory_bytes()
        );
        let info = f.run(&[b"INFO", b"stats"]);
        assert!(!info.contains("evicted_keys:0"), "{info}");
        assert!(
            f.run(&[b"DBSIZE"]) != ":0\r\n",
            "and it did not empty the database to get there"
        );
    }

    #[test]
    fn the_running_total_and_the_walk_agree_on_a_mixed_keyspace() {
        // The limit is judged against a number kept as the collections move,
        // rather than found by asking all of them, and the two have to be the
        // same number or the limit is enforced against a fiction. This does the
        // things that move it, which is growing a collection, shrinking one,
        // changing its representation, deleting it and reusing its slot, across
        // all five types, and checks the two against each other as it goes.
        let mut f = Fixture::new();
        f.run(&[b"CONFIG", b"SET", b"maxmemory", b"1gb"]);
        let big = vec![b'v'; 200];

        for i in 0..400u32 {
            let n = i.to_string();
            let n = n.as_bytes();
            f.run(&[b"SADD", b"s", n]);
            f.run(&[b"SADD", b"s2", &big]);
            f.run(&[b"HSET", b"h", n, &big]);
            f.run(&[b"RPUSH", b"l", &big]);
            f.run(&[b"ZADD", b"z", n, n]);
            f.run(&[b"ARSET", b"a", n, &big]);
            if i % 7 == 0 {
                f.run(&[b"SREM", b"s", n]);
                f.run(&[b"HDEL", b"h", n]);
                f.run(&[b"LPOP", b"l"]);
                f.run(&[b"ZREM", b"z", n]);
                f.run(&[b"ARDEL", b"a", n]);
            }
            if i % 53 == 0 {
                // Every type deleted and made again, so a slot goes on the free
                // list and comes back holding something else.
                f.run(&[b"DEL", b"s2"]);
            }
            assert_eq!(
                f.server.settled_memory(),
                f.server.memory_bytes(),
                "after round {i}"
            );
        }

        // The run has to have built something, or the two numbers agreeing is
        // two zeroes agreeing.
        assert_eq!(f.run(&[b"DBSIZE"]), ":6\r\n");
        assert!(
            f.server.memory_bytes() > 512 * 1024,
            "{}",
            f.server.memory_bytes()
        );

        // And it survives the collections going away entirely.
        f.run(&[b"FLUSHALL"]);
        assert_eq!(f.server.settled_memory(), f.server.memory_bytes());
    }

    #[test]
    fn taking_the_limit_away_stops_the_counting_and_putting_it_back_starts_again() {
        // A server with no limit does not keep the running total, so setting a
        // limit on a database that is already full has to start it from a walk.
        // If it did not, the first reading would be zero and the server would
        // think it had all the room in the world.
        let mut f = Fixture::new();
        for i in 0..200u32 {
            let n = i.to_string();
            f.run(&[b"SADD", b"s", n.as_bytes()]);
            f.run(&[b"HSET", b"h", n.as_bytes(), b"value"]);
        }
        f.run(&[b"CONFIG", b"SET", b"maxmemory", b"1gb"]);
        assert_eq!(f.server.settled_memory(), f.server.memory_bytes());

        f.run(&[b"CONFIG", b"SET", b"maxmemory", b"0"]);
        for i in 200..400u32 {
            let n = i.to_string();
            f.run(&[b"SADD", b"s", n.as_bytes()]);
        }
        f.run(&[b"CONFIG", b"SET", b"maxmemory", b"1gb"]);
        assert_eq!(
            f.server.settled_memory(),
            f.server.memory_bytes(),
            "the writes it was not watching are in the number it started from"
        );
    }

    #[test]
    fn evicted_keys_and_expired_keys_are_different_numbers() {
        let mut f = Fixture::new();
        // Nothing has been evicted and nothing can be under the default policy,
        // so this stays at zero while the other one moves.
        f.run(&[b"SET", b"gone", b"v", b"PX", b"1"]);
        f.server.db(0).clock_mut().advance(20);
        f.run(&[b"GET", b"gone"]);
        let info = f.run(&[b"INFO", b"stats"]);
        assert!(info.contains("expired_keys:1"), "{info}");
        assert!(info.contains("evicted_keys:0"), "{info}");
    }

    #[test]
    fn the_object_subcommands_follow_the_policy() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"s", b"v"]);
        // Under the default the clock is kept and the counter is not, and under
        // an LFU policy it is the other way round. Each subcommand refuses on
        // the side where its reading of the three bytes means nothing.
        assert_eq!(f.run(&[b"OBJECT", b"IDLETIME", b"s"]), ":0\r\n");
        assert!(
            f.run(&[b"OBJECT", b"FREQ", b"s"])
                .starts_with("-ERR An LFU maxmemory policy is not selected"),
        );

        f.run(&[b"CONFIG", b"SET", b"maxmemory-policy", b"allkeys-lfu"]);
        assert!(
            f.run(&[b"OBJECT", b"IDLETIME", b"s"])
                .starts_with("-ERR An LFU maxmemory policy is selected"),
        );
        // The key was written under a clock policy, so what comes back is that
        // clock read as a counter. It is a number and not an error, which is the
        // point: switching at runtime does not invalidate anything, it only makes
        // the old field mean something else until the key is used again.
        assert!(
            f.run(&[b"OBJECT", b"FREQ", b"s"]).starts_with(':'),
            "FREQ should answer under an LFU policy"
        );
    }

    #[test]
    fn object_says_which_rung_of_the_ladder_a_key_is_on() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"s", b"hello"]);
        f.run(&[b"SET", b"n", b"123"]);
        f.run(&[b"SADD", b"si", b"1", b"2", b"3"]);
        f.run(&[b"SADD", b"ss", b"a", b"b"]);
        f.run(&[b"HSET", b"h", b"f", b"v"]);
        for (key, want) in [
            (b"s".as_slice(), "embstr"),
            (b"n", "int"),
            (b"si", "intset"),
            (b"ss", "listpack"),
            (b"h", "listpack"),
        ] {
            let reply = f.run(&[b"OBJECT", b"ENCODING", key]);
            assert_eq!(reply, format!("${}\r\n{want}\r\n", want.len()));
        }

        // A field deadline widens the blob rather than promoting it, and this
        // is the only place a client can see that happen.
        f.run(&[b"HEXPIRE", b"h", b"100", b"FIELDS", b"1", b"f"]);
        assert_eq!(
            f.run(&[b"OBJECT", b"ENCODING", b"h"]),
            "$10\r\nlistpackex\r\n"
        );

        assert_eq!(f.run(&[b"OBJECT", b"REFCOUNT", b"s"]), ":1\r\n");
        assert_eq!(f.run(&[b"OBJECT", b"IDLETIME", b"s"]), ":0\r\n");
        assert!(f.run(&[b"OBJECT", b"HELP"]).starts_with("*14\r\n+OBJECT "));
    }

    #[test]
    fn object_answers_nil_for_a_key_that_is_not_there() {
        let mut f = Fixture::new();
        for sub in [b"ENCODING".as_slice(), b"REFCOUNT", b"IDLETIME", b"FREQ"] {
            assert_eq!(
                f.run(&[b"OBJECT", sub, b"nokey"]),
                "$-1\r\n",
                "a nil and not an error, which is what 8.10.1 does"
            );
        }
        // And the key is looked up before FREQ has its complaint, so the
        // complaint only reaches a key that exists.
        f.run(&[b"SET", b"s", b"v"]);
        assert!(
            f.run(&[b"OBJECT", b"FREQ", b"s"])
                .starts_with("-ERR An LFU maxmemory policy is not"),
        );
        assert_eq!(
            f.run(&[b"OBJECT", b"NOPE", b"s"]),
            "-ERR unknown subcommand 'NOPE'. Try OBJECT HELP.\r\n"
        );
        assert_eq!(
            f.run(&[b"OBJECT", b"ENCODING"]),
            "-ERR wrong number of arguments for 'object|encoding' command\r\n"
        );
        assert_eq!(
            f.run(&[b"OBJECT", b"ENCODING", b"s", b"extra"]),
            "-ERR wrong number of arguments for 'object|encoding' command\r\n"
        );
        assert_eq!(
            f.run(&[b"OBJECT"]),
            "-ERR wrong number of arguments for 'object' command\r\n"
        );
    }

    #[test]
    fn config_moves_the_ladder_and_object_encoding_agrees() {
        let mut f = Fixture::new();
        assert_eq!(
            f.run(&[b"CONFIG", b"GET", b"hash-max-listpack-entries"]),
            "*2\r\n$25\r\nhash-max-listpack-entries\r\n$3\r\n512\r\n",
            "512 and not the 128 everyone remembers, which is what 8.10.1 says"
        );
        // The old spelling is the same number under a different name, and a
        // glob that catches both sends both.
        assert_eq!(
            f.run(&[b"CONFIG", b"GET", b"hash-max-ziplist-entries"]),
            "*2\r\n$24\r\nhash-max-ziplist-entries\r\n$3\r\n512\r\n"
        );
        assert!(
            f.run(&[b"CONFIG", b"GET", b"hash-max-*"])
                .starts_with("*8\r\n")
        );
        assert!(
            f.run(&[b"CONFIG", b"GET", b"set-max-*"])
                .starts_with("*6\r\n")
        );

        f.run(&[b"HSET", b"h", b"a", b"1", b"b", b"2", b"c", b"3"]);
        assert_eq!(f.run(&[b"OBJECT", b"ENCODING", b"h"]), "$8\r\nlistpack\r\n");

        assert_eq!(
            f.run(&[b"CONFIG", b"SET", b"hash-max-ziplist-entries", b"2"]),
            "+OK\r\n",
            "written under the old name and read back under the new one"
        );
        assert_eq!(
            f.run(&[b"CONFIG", b"GET", b"hash-max-listpack-entries"]),
            "*2\r\n$25\r\nhash-max-listpack-entries\r\n$1\r\n2\r\n"
        );
        assert_eq!(
            f.run(&[b"OBJECT", b"ENCODING", b"h"]),
            "$8\r\nlistpack\r\n",
            "the hash that already exists is left exactly where it was"
        );
        f.run(&[b"HSET", b"h2", b"a", b"1", b"b", b"2", b"c", b"3"]);
        assert_eq!(
            f.run(&[b"OBJECT", b"ENCODING", b"h2"]),
            "$9\r\nhashtable\r\n",
            "and the next one built goes straight to a table"
        );

        // The set has three of these and all three move.
        f.run(&[b"CONFIG", b"SET", b"set-max-intset-entries", b"2"]);
        f.run(&[b"SADD", b"s", b"1", b"2", b"3"]);
        assert_eq!(f.run(&[b"OBJECT", b"ENCODING", b"s"]), "$8\r\nlistpack\r\n");
        f.run(&[b"CONFIG", b"SET", b"set-max-listpack-value", b"2"]);
        f.run(&[b"SADD", b"s2", b"abcdefgh"]);
        assert_eq!(
            f.run(&[b"OBJECT", b"ENCODING", b"s2"]),
            "$9\r\nhashtable\r\n"
        );
    }

    #[test]
    fn config_set_takes_all_of_the_ladder_or_none_of_it() {
        let mut f = Fixture::new();
        assert_eq!(
            f.run(&[
                b"CONFIG",
                b"SET",
                b"hash-max-listpack-entries",
                b"7",
                b"set-max-listpack-entries",
                b"abc"
            ]),
            "-ERR CONFIG SET failed (possibly related to argument 'set-max-listpack-entries') - argument couldn't be parsed into an integer\r\n"
        );
        assert_eq!(
            f.run(&[b"CONFIG", b"GET", b"hash-max-listpack-entries"]),
            "*2\r\n$25\r\nhash-max-listpack-entries\r\n$3\r\n512\r\n",
            "the pair in front of the bad one did not go in"
        );
        // The name in the complaint is the one that was typed, so the old
        // spelling comes back as the old spelling.
        assert_eq!(
            f.run(&[b"CONFIG", b"SET", b"hash-max-ziplist-entries", b"abc"]),
            "-ERR CONFIG SET failed (possibly related to argument 'hash-max-ziplist-entries') - argument couldn't be parsed into an integer\r\n"
        );
        assert_eq!(
            f.run(&[b"CONFIG", b"SET", b"set-max-intset-entries", b"-1"]),
            "-ERR CONFIG SET failed (possibly related to argument 'set-max-intset-entries') - argument must be between 0 and 9223372036854775807 inclusive\r\n"
        );
        // A number past what an i64 holds is the parse complaint and not the
        // range one, which is upstream reading it before it checks it.
        assert_eq!(
            f.run(&[
                b"CONFIG",
                b"SET",
                b"set-max-intset-entries",
                b"99999999999999999999"
            ]),
            "-ERR CONFIG SET failed (possibly related to argument 'set-max-intset-entries') - argument couldn't be parsed into an integer\r\n"
        );
        assert_eq!(
            f.run(&[
                b"CONFIG",
                b"SET",
                b"set-max-intset-entries",
                b"9223372036854775807"
            ]),
            "+OK\r\n"
        );
    }

    #[test]
    fn a_setting_moved_on_one_database_moved_on_all_of_them() {
        let mut f = Fixture::new();
        f.run(&[b"CONFIG", b"SET", b"hash-max-listpack-entries", b"1"]);
        f.run(&[b"SELECT", b"3"]);
        f.run(&[b"HSET", b"h", b"a", b"1", b"b", b"2"]);
        assert_eq!(
            f.run(&[b"OBJECT", b"ENCODING", b"h"]),
            "$9\r\nhashtable\r\n",
            "these are one server wide number in Redis, whatever a Keyspace carries"
        );
    }

    #[test]
    fn info_reports_the_numbers_it_can_stand_behind() {
        let mut f = Fixture::new();
        f.run(&[b"MSET", b"a", b"1", b"b", b"2"]);
        let all = f.run(&[b"INFO"]);
        assert!(all.contains("redis_version:8.8.0"), "{all}");
        assert!(
            all.contains(concat!("yo_version:", env!("CARGO_PKG_VERSION"))),
            "{all}"
        );
        assert!(all.contains("db0:keys=2,expires=0,avg_ttl=0"), "{all}");
        assert!(all.contains("role:master"), "{all}");
        // One section is one section.
        let clients = f.run(&[b"INFO", b"clients"]);
        assert!(clients.contains("connected_clients:0"), "{clients}");
        assert!(!clients.contains("redis_version"), "{clients}");
        assert_eq!(f.run(&[b"INFO", b"nosuch"]), "$0\r\n\r\n");
    }

    /// A cache that writes with a deadline and never reads back used to hold
    /// every key it had ever written, because lazy expiry needs somebody to walk
    /// past a key before it can reclaim it and nobody ever did.
    #[test]
    fn the_active_sweep_reclaims_keys_no_client_comes_back_for() {
        let mut f = Fixture::new();
        for i in 0..3_000u32 {
            f.run(&[b"SET", format!("d{i}").as_bytes(), b"v", b"PX", b"50"]);
        }
        for i in 0..1_000u32 {
            f.run(&[b"SET", format!("k{i}").as_bytes(), b"v"]);
        }
        assert_eq!(f.run(&[b"DBSIZE"]), ":4000\r\n");
        f.advance(100);
        assert_eq!(
            f.run(&[b"DBSIZE"]),
            ":4000\r\n",
            "DBSIZE counts records and nothing has read past the dead ones yet"
        );

        // What the shard loop does, one slice at a time.
        let mut spent = 0;
        for _ in 0..2_000 {
            spent += f.server.expire_step(4096);
            if f.run(&[b"DBSIZE"]) == ":1000\r\n" {
                break;
            }
        }
        assert_eq!(f.run(&[b"DBSIZE"]), ":1000\r\n", "spent {spent} looks");
        assert!(f.run(&[b"INFO", b"stats"]).contains("expired_keys:3000"));
        for i in 0..1_000u32 {
            assert_eq!(
                f.run(&[b"GET", format!("k{i}").as_bytes()]),
                "$1\r\nv\r\n",
                "it took a key that had no deadline"
            );
        }
    }

    #[test]
    fn a_sweep_of_a_server_with_no_deadlines_anywhere_costs_nothing() {
        let mut f = Fixture::new();
        for i in 0..2_000u32 {
            f.run(&[b"SET", format!("k{i}").as_bytes(), b"v"]);
        }
        assert_eq!(f.server.expire_step(4096), 0);
        // And one database having them does not make the other fifteen pay.
        f.run(&[b"SELECT", b"3"]);
        f.run(&[b"SET", b"x", b"v", b"PX", b"50"]);
        f.advance(100);
        for _ in 0..64 {
            f.server.expire_step(4096);
        }
        assert_eq!(f.run(&[b"DBSIZE"]), ":0\r\n");
        f.run(&[b"SELECT", b"0"]);
        assert_eq!(f.run(&[b"DBSIZE"]), ":2000\r\n");
        assert_eq!(f.server.expire_step(4096), 0, "and it is quiet again");
    }

    /// The gate, which is what stops a maintenance slice that runs every hundred
    /// nanoseconds from drawing a sample every hundred nanoseconds.
    #[test]
    fn the_sweep_the_loop_calls_runs_at_most_once_a_millisecond() {
        let mut f = Fixture::new();
        for i in 0..500u32 {
            f.run(&[b"SET", format!("d{i}").as_bytes(), b"v", b"PX", b"50"]);
        }
        f.advance(100);
        let at = f.server.db(0).clock().now_ms();
        f.server.set_clock_ms(at);
        // A small budget, so that one slice cannot finish the job and a second
        // one having nothing to do would mean the gate and not an empty
        // database.
        assert!(f.server.expire_slice(8) > 0, "the first one works");
        for _ in 0..1_000 {
            assert_eq!(
                f.server.expire_slice(8),
                0,
                "the millisecond has not moved and neither should this"
            );
        }
        assert!(
            f.server.db(0).expires() > 400,
            "there is plenty left to take"
        );
        f.server.set_clock_ms(at + 1);
        assert!(f.server.expire_slice(8) > 0, "and then it goes again");
    }

    /// `expires=` used to be a hardcoded zero, which meant a dashboard watching
    /// how much of a cache is volatile was reading a constant.
    #[test]
    fn info_keyspace_counts_the_keys_that_have_a_deadline() {
        let mut f = Fixture::new();
        f.run(&[b"MSET", b"a", b"1", b"b", b"2", b"c", b"3"]);
        assert!(
            f.run(&[b"INFO", b"keyspace"])
                .contains("db0:keys=3,expires=0"),
            "none of them has one yet"
        );
        f.run(&[b"EXPIRE", b"a", b"1000"]);
        f.run(&[b"EXPIRE", b"b", b"1000"]);
        let two = f.run(&[b"INFO", b"keyspace"]);
        assert!(two.contains("db0:keys=3,expires=2"), "{two}");
        f.run(&[b"PERSIST", b"a"]);
        f.run(&[b"DEL", b"b"]);
        let none = f.run(&[b"INFO", b"keyspace"]);
        assert!(none.contains("db0:keys=2,expires=0"), "{none}");

        // Each database answers for itself, the way Redis reports it.
        f.run(&[b"SELECT", b"1"]);
        f.run(&[b"SET", b"x", b"1", b"EX", b"1000"]);
        let both = f.run(&[b"INFO", b"keyspace"]);
        assert!(both.contains("db0:keys=2,expires=0"), "{both}");
        assert!(both.contains("db1:keys=1,expires=1"), "{both}");
    }

    #[cfg(unix)]
    #[test]
    fn info_cpu_reports_processor_time_that_was_really_measured() {
        let mut f = Fixture::new();
        let cpu = f.run(&[b"INFO", b"cpu"]);
        assert!(cpu.contains("# CPU"), "{cpu}");
        // Redis's unit/info-command asks for this one by name in three tests.
        assert!(cpu.contains("used_cpu_user:"), "{cpu}");
        assert!(cpu.contains("used_cpu_sys:"), "{cpu}");
        assert!(cpu.contains("used_cpu_user_children:0.000000"), "{cpu}");
        assert!(!cpu.contains("redis_version"), "{cpu}");

        // It is a measurement and not a constant, so it goes up when work
        // happens. A tight loop rather than a sleep, because sleeping is the
        // one thing that does not move this number.
        let before = used_cpu_user(&cpu);
        let mut n = 0u64;
        let mut rounds = 0;
        while used_cpu_user(&f.run(&[b"INFO", b"cpu"])) <= before {
            for i in 0..1_000_000u64 {
                n = n.wrapping_add(i.wrapping_mul(i));
            }
            rounds += 1;
            // A bound rather than a spin, so a platform where this number does
            // not move fails here instead of hanging. Even a clock with whole
            // millisecond granularity gets there in the first round or two.
            assert!(rounds < 1_000, "cpu time never moved, n is {n}");
        }
    }

    /// Pull `used_cpu_user` back out of an `INFO cpu` reply.
    #[cfg(unix)]
    fn used_cpu_user(info: &str) -> f64 {
        info.lines()
            .find_map(|l| l.strip_prefix("used_cpu_user:"))
            .expect("no used_cpu_user in the reply")
            .trim()
            .parse()
            .expect("used_cpu_user is not a number")
    }

    /// The safety net under the rule that a body checks its arguments before
    /// it writes anything. `MGET` writes its array header first and then reads
    /// each key, so if a later argument could fail the header would already be
    /// out. Nothing in the string group does that today and this is what would
    /// catch the first one that did.
    #[test]
    fn a_command_that_fails_leaves_nothing_half_written() {
        let mut f = Fixture::new();
        let reply = f.run(&[b"SETRANGE", b"k", b"-1", b"x"]);
        assert_eq!(reply, "-ERR offset is out of range\r\n");
        assert!(!reply.contains(':'), "no integer went out in front of it");
    }

    #[test]
    fn quit_answers_first_and_closes_after() {
        let mut f = Fixture::new();
        let (flow, reply) = f.flow(&[b"QUIT"]);
        assert_eq!(reply, "+OK\r\n");
        assert_eq!(flow, Flow::Close);
    }

    #[test]
    fn the_command_counter_counts_every_command_including_the_bad_ones() {
        let mut f = Fixture::new();
        f.run(&[b"PING"]);
        f.run(&[b"NOPE"]);
        f.run(&[b"GET"]);
        assert_eq!(f.server.stats.commands, 3);
    }

    #[test]
    fn a_set_goes_from_bytes_to_bytes() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"SADD", b"s", b"a", b"b", b"c"]), ":3\r\n");
        assert_eq!(f.run(&[b"SADD", b"s", b"b", b"d"]), ":1\r\n");
        assert_eq!(f.run(&[b"SCARD", b"s"]), ":4\r\n");
        assert_eq!(f.run(&[b"SISMEMBER", b"s", b"a"]), ":1\r\n");
        assert_eq!(f.run(&[b"SISMEMBER", b"s", b"z"]), ":0\r\n");
        assert_eq!(f.run(&[b"TYPE", b"s"]), "+set\r\n");
        assert_eq!(
            f.run(&[b"SMISMEMBER", b"s", b"a", b"z", b"d"]),
            "*3\r\n:1\r\n:0\r\n:1\r\n"
        );
        assert_eq!(f.run(&[b"SREM", b"s", b"a", b"z"]), ":1\r\n");
        assert_eq!(f.run(&[b"SCARD", b"s"]), ":3\r\n");
    }

    #[test]
    fn a_set_command_at_a_key_that_is_not_there_answers_empty() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"SCARD", b"nope"]), ":0\r\n");
        assert_eq!(f.run(&[b"SISMEMBER", b"nope", b"a"]), ":0\r\n");
        assert_eq!(f.run(&[b"SREM", b"nope", b"a"]), ":0\r\n");
        assert_eq!(f.run(&[b"SMEMBERS", b"nope"]), "*0\r\n");
        assert_eq!(
            f.run(&[b"SMISMEMBER", b"nope", b"a", b"b"]),
            "*2\r\n:0\r\n:0\r\n"
        );
        assert_eq!(f.run(&[b"DBSIZE"]), ":0\r\n", "and made nothing");
    }

    #[test]
    fn smembers_answers_a_set_on_resp3_and_an_array_on_resp2() {
        // Not cosmetic. A RESP3 client that gets a `~` hands the caller a set
        // and one that gets a `*` hands it a list, without either of them being
        // told which command was sent.
        let mut f = Fixture::new();
        f.run(&[b"SADD", b"s", b"one"]);
        assert_eq!(f.run(&[b"SMEMBERS", b"s"]), "*1\r\n$3\r\none\r\n");

        f.run(&[b"HELLO", b"3"]);
        assert_eq!(f.run(&[b"SMEMBERS", b"s"]), "~1\r\n$3\r\none\r\n");
    }

    #[test]
    fn an_integer_member_comes_back_as_the_digits_it_never_stored() {
        // An intset holds the number, so these digits exist for the first time
        // in the reply buffer.
        let mut f = Fixture::new();
        f.run(&[b"SADD", b"s", b"42"]);
        assert_eq!(f.run(&[b"SMEMBERS", b"s"]), "*1\r\n$2\r\n42\r\n");
        assert_eq!(f.run(&[b"SISMEMBER", b"s", b"42"]), ":1\r\n");
        assert_eq!(
            f.run(&[b"SISMEMBER", b"s", b"042"]),
            ":0\r\n",
            "the member is the bytes and not the number they parse to"
        );
    }

    #[test]
    fn the_wrong_command_at_the_wrong_type_says_so_both_ways() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"str", b"v"]);
        f.run(&[b"SADD", b"set", b"a"]);

        let wrong = "-WRONGTYPE Operation against a key holding the wrong kind of value\r\n";
        assert_eq!(f.run(&[b"SADD", b"str", b"a"]), wrong);
        assert_eq!(f.run(&[b"SCARD", b"str"]), wrong);
        assert_eq!(f.run(&[b"SMEMBERS", b"str"]), wrong);
        assert_eq!(f.run(&[b"SMISMEMBER", b"str", b"a"]), wrong);
        assert_eq!(f.run(&[b"GET", b"set"]), wrong);
        assert_eq!(f.run(&[b"APPEND", b"set", b"x"]), wrong);
        assert_eq!(f.run(&[b"INCR", b"set"]), wrong);
        assert_eq!(f.run(&[b"STRLEN", b"set"]), wrong);

        // MGET is the one that does not, because Redis gives nil for the odd
        // key out rather than failing the good keys next to it.
        assert_eq!(
            f.run(&[b"MGET", b"str", b"set", b"nope"]),
            "*3\r\n$1\r\nv\r\n$-1\r\n$-1\r\n"
        );
        // And plain SET overwrites any type, which takes the body with it.
        assert_eq!(f.run(&[b"SET", b"set", b"now a string"]), "+OK\r\n");
        assert_eq!(f.run(&[b"TYPE", b"set"]), "+string\r\n");
    }

    #[test]
    fn a_wrongtype_leaves_nothing_half_written() {
        // SMISMEMBER writes an array header and then one reply per member, so
        // it is the first command in the server that could get a header out in
        // front of an error if it checked its key in the wrong order.
        let mut f = Fixture::new();
        f.run(&[b"SET", b"k", b"v"]);
        let reply = f.run(&[b"SMISMEMBER", b"k", b"a", b"b"]);
        assert!(reply.starts_with("-WRONGTYPE"), "got {reply}");
        assert!(!reply.contains('*'), "an array header went out in front");
    }

    #[test]
    fn emptying_a_set_takes_the_key_with_it() {
        let mut f = Fixture::new();
        f.run(&[b"SADD", b"s", b"a", b"b"]);
        assert_eq!(f.run(&[b"DBSIZE"]), ":1\r\n");
        assert_eq!(f.run(&[b"SREM", b"s", b"a", b"b"]), ":2\r\n");
        assert_eq!(f.run(&[b"EXISTS", b"s"]), ":0\r\n");
        assert_eq!(f.run(&[b"TYPE", b"s"]), "+none\r\n");
        assert_eq!(f.run(&[b"DBSIZE"]), ":0\r\n");
    }

    /// Pull the cursor and the members out of one `SSCAN` reply.
    ///
    /// Crude on purpose. A test that walked a set through a real client would
    /// be testing the client, and what these tests are about is the shape of
    /// the bytes and the fact that a walk sees every member once.
    fn split_scan(reply: &str) -> (String, Vec<String>) {
        let mut lines = reply.split("\r\n");
        assert_eq!(lines.next(), Some("*2"), "got {reply}");
        lines.next().expect("the cursor header");
        let cursor = lines.next().expect("the cursor").to_owned();
        let header = lines.next().expect("the member header");
        let n: usize = header[1..].parse().expect("a member count");
        let mut members = Vec::with_capacity(n);
        for _ in 0..n {
            lines.next().expect("a member header");
            members.push(lines.next().expect("a member").to_owned());
        }
        (cursor, members)
    }

    #[test]
    fn popping_takes_a_member_off_the_set_and_hands_it_back() {
        let mut f = Fixture::new();
        f.run(&[b"SADD", b"s", b"a", b"b", b"c", b"d"]);

        let one = f.run(&[b"SPOP", b"s"]);
        assert!(
            ["$1\r\na\r\n", "$1\r\nb\r\n", "$1\r\nc\r\n", "$1\r\nd\r\n"].contains(&one.as_str()),
            "got {one}"
        );
        assert_eq!(f.run(&[b"SCARD", b"s"]), ":3\r\n");

        // A count takes that many, and the last one takes the key with it.
        let (_, rest) = ("", f.run(&[b"SPOP", b"s", b"3"]));
        assert!(rest.starts_with("*3\r\n"), "got {rest}");
        assert_eq!(f.run(&[b"EXISTS", b"s"]), ":0\r\n");
        // And a pop at a key that is not there is a nil, not an empty bulk.
        assert_eq!(f.run(&[b"SPOP", b"s"]), "$-1\r\n");
        assert_eq!(f.run(&[b"SPOP", b"s", b"2"]), "*0\r\n");
    }

    #[test]
    fn the_two_draws_disagree_about_the_reply_type_and_they_are_right_to() {
        // The one place in the server where the reply type carries something
        // the command name does not. SPOP's members are distinct so a RESP3
        // client can build a set out of them. SRANDMEMBER with a negative count
        // can hand back the same member three times, and a set would lose two.
        let mut f = Fixture::new();
        f.run(&[b"HELLO", b"3"]);
        f.run(&[b"SADD", b"s", b"a", b"b", b"c"]);

        assert!(f.run(&[b"SPOP", b"s", b"2"]).starts_with("~2\r\n"));
        // And a positive count is an array too, since Redis makes it one.
        assert!(f.run(&[b"SRANDMEMBER", b"s", b"1"]).starts_with("*1\r\n"));

        // A negative count against a set of one is where the difference bites:
        // the same member three times, which is a three element reply and would
        // have been a one element reply if it had gone out as a set.
        f.run(&[b"SADD", b"one", b"z"]);
        assert_eq!(
            f.run(&[b"SRANDMEMBER", b"one", b"-3"]),
            "*3\r\n$1\r\nz\r\n$1\r\nz\r\n$1\r\nz\r\n"
        );
    }

    #[test]
    fn drawing_a_member_removes_nothing_and_says_nil_at_a_missing_key() {
        let mut f = Fixture::new();
        f.run(&[b"SADD", b"s", b"only"]);
        assert_eq!(f.run(&[b"SRANDMEMBER", b"s"]), "$4\r\nonly\r\n");
        assert_eq!(f.run(&[b"SRANDMEMBER", b"s"]), "$4\r\nonly\r\n");
        assert_eq!(f.run(&[b"SCARD", b"s"]), ":1\r\n");

        assert_eq!(f.run(&[b"SRANDMEMBER", b"nope"]), "$-1\r\n");
        // The count form answers an empty array rather than a nil, which is the
        // pair of answers Redis gives and is not the pair it looks like.
        assert_eq!(f.run(&[b"SRANDMEMBER", b"nope", b"3"]), "*0\r\n");
        assert_eq!(f.run(&[b"SRANDMEMBER", b"nope", b"-3"]), "*0\r\n");
        // Asking for more than is there answers all of it once and not padding.
        assert_eq!(f.run(&[b"SRANDMEMBER", b"s", b"9"]), "*1\r\n$4\r\nonly\r\n");
    }

    #[test]
    fn a_pop_count_that_is_not_a_positive_number_says_so() {
        let mut f = Fixture::new();
        f.run(&[b"SADD", b"s", b"a"]);
        let bad = "-ERR value is out of range, must be positive\r\n";
        assert_eq!(f.run(&[b"SPOP", b"s", b"-1"]), bad);
        assert_eq!(f.run(&[b"SPOP", b"s", b"abc"]), bad);
        assert_eq!(f.run(&[b"SCARD", b"s"]), ":1\r\n", "and took nothing");
        // Zero is allowed and is a real answer rather than an error.
        assert_eq!(f.run(&[b"SPOP", b"s", b"0"]), "*0\r\n");
        assert_eq!(f.run(&[b"EXISTS", b"s"]), ":1\r\n");
    }

    #[test]
    fn a_scan_walks_a_set_of_any_size_exactly_once() {
        let mut f = Fixture::new();
        let members: Vec<Vec<u8>> = (0..300).map(|i| format!("m{i}").into_bytes()).collect();
        let args: Vec<&[u8]> = [&b"SADD"[..], &b"s"[..]]
            .into_iter()
            .chain(members.iter().map(Vec::as_slice))
            .collect();
        f.run(&args);

        let mut seen = Vec::new();
        let mut cursor = "0".to_owned();
        loop {
            let reply = f.run(&[b"SSCAN", b"s", cursor.as_bytes()]);
            let (next, got) = split_scan(&reply);
            seen.extend(got);
            cursor = next;
            if cursor == "0" {
                break;
            }
        }
        seen.sort();
        seen.dedup();
        assert_eq!(seen.len(), 300, "a walk saw a member twice or missed one");

        // A set small enough to be a listpack answers in one call whatever
        // cursor it was handed, which is what Redis does for that encoding.
        f.run(&[b"SADD", b"small", b"a", b"b", b"c"]);
        let (cursor, got) = split_scan(&f.run(&[b"SSCAN", b"small", b"0", b"COUNT", b"1"]));
        assert_eq!(cursor, "0");
        assert_eq!(got.len(), 3);
        // And a key that is not there is a finished scan of nothing.
        assert_eq!(f.run(&[b"SSCAN", b"nope", b"0"]), "*2\r\n$1\r\n0\r\n*0\r\n");
    }

    #[test]
    fn a_scan_takes_match_and_count_and_refuses_anything_else() {
        let mut f = Fixture::new();
        f.run(&[b"SADD", b"s", b"aa", b"ab", b"ba", b"12", b"13"]);

        let (_, got) = split_scan(&f.run(&[b"SSCAN", b"s", b"0", b"MATCH", b"a*"]));
        let mut got = got;
        got.sort();
        assert_eq!(got, ["aa", "ab"]);

        // An integer member has no digits stored anywhere, so MATCH is the one
        // place a scan pays to write some.
        let (_, got) = split_scan(&f.run(&[b"SSCAN", b"s", b"0", b"MATCH", b"1?"]));
        let mut got = got;
        got.sort();
        assert_eq!(got, ["12", "13"]);

        assert_eq!(f.run(&[b"SSCAN", b"s", b"abc"]), "-ERR invalid cursor\r\n");
        assert_eq!(f.run(&[b"SSCAN", b"s", b"-1"]), "-ERR invalid cursor\r\n");
        assert_eq!(
            f.run(&[b"SSCAN", b"s", b"0", b"NOPE", b"1"]),
            "-ERR syntax error\r\n"
        );
        // A count under one is a syntax error and not a range error, which is
        // the odder of Redis's two answers and the reason it is copied exactly.
        assert_eq!(
            f.run(&[b"SSCAN", b"s", b"0", b"COUNT", b"0"]),
            "-ERR syntax error\r\n"
        );
    }

    #[test]
    fn moving_a_member_takes_it_off_one_set_and_puts_it_on_another() {
        let mut f = Fixture::new();
        f.run(&[b"SADD", b"src", b"a", b"b"]);
        f.run(&[b"SADD", b"dst", b"c"]);

        assert_eq!(f.run(&[b"SMOVE", b"src", b"dst", b"a"]), ":1\r\n");
        assert_eq!(f.run(&[b"SISMEMBER", b"src", b"a"]), ":0\r\n");
        assert_eq!(f.run(&[b"SISMEMBER", b"dst", b"a"]), ":1\r\n");
        // A member that is not in the source is a zero and moves nothing.
        assert_eq!(f.run(&[b"SMOVE", b"src", b"dst", b"zz"]), ":0\r\n");
        assert_eq!(f.run(&[b"SCARD", b"dst"]), ":2\r\n");

        // A destination that does not exist gets made, and a source that runs
        // out goes away.
        assert_eq!(f.run(&[b"SMOVE", b"src", b"fresh", b"b"]), ":1\r\n");
        assert_eq!(f.run(&[b"EXISTS", b"src"]), ":0\r\n");
        assert_eq!(f.run(&[b"SMEMBERS", b"fresh"]), "*1\r\n$1\r\nb\r\n");
    }

    #[test]
    fn moving_checks_the_types_in_the_order_redis_checks_them() {
        // Not the order it looks like it should be. A source that is not there
        // answers zero without ever looking at the destination, so this is a
        // zero and not a WRONGTYPE even though the destination is a string.
        let mut f = Fixture::new();
        f.run(&[b"SET", b"str", b"v"]);
        f.run(&[b"SADD", b"set", b"a"]);

        let wrong = "-WRONGTYPE Operation against a key holding the wrong kind of value\r\n";
        assert_eq!(f.run(&[b"SMOVE", b"nope", b"str", b"a"]), ":0\r\n");
        assert_eq!(f.run(&[b"SMOVE", b"str", b"set", b"a"]), wrong);
        assert_eq!(f.run(&[b"SMOVE", b"set", b"str", b"a"]), wrong);
        assert_eq!(f.run(&[b"SPOP", b"str"]), wrong);
        assert_eq!(f.run(&[b"SRANDMEMBER", b"str"]), wrong);
        assert_eq!(f.run(&[b"SSCAN", b"str", b"0"]), wrong);
        assert_eq!(
            f.run(&[b"SISMEMBER", b"set", b"a"]),
            ":1\r\n",
            "and none of that moved anything"
        );
    }

    #[test]
    fn a_scan_leaves_nothing_half_written_when_its_arguments_are_wrong() {
        // SSCAN writes an outer array header before it walks, so it is the
        // command most likely to get bytes out in front of an error.
        let mut f = Fixture::new();
        f.run(&[b"SADD", b"s", b"a"]);
        for bad in [
            &[b"SSCAN".as_slice(), b"s", b"abc"][..],
            &[b"SSCAN".as_slice(), b"s", b"0", b"COUNT", b"nope"][..],
            &[b"SSCAN".as_slice(), b"s", b"0", b"MATCH"][..],
        ] {
            let reply = f.run(bad);
            assert!(reply.starts_with("-ERR"), "got {reply}");
            assert!(!reply.contains('*'), "an array header went out in front");
        }
    }

    #[test]
    fn a_hash_writes_reads_and_deletes_its_fields() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"HSET", b"h", b"a", b"1", b"b", b"2"]), ":2\r\n");
        assert_eq!(f.run(&[b"HSET", b"h", b"a", b"9"]), ":0\r\n", "a was there");
        assert_eq!(f.run(&[b"HGET", b"h", b"a"]), "$1\r\n9\r\n");
        assert_eq!(f.run(&[b"HGET", b"h", b"nope"]), "$-1\r\n");
        assert_eq!(f.run(&[b"HGET", b"nokey", b"a"]), "$-1\r\n");
        assert_eq!(f.run(&[b"HLEN", b"h"]), ":2\r\n");
        assert_eq!(f.run(&[b"HEXISTS", b"h", b"a"]), ":1\r\n");
        assert_eq!(f.run(&[b"HEXISTS", b"h", b"nope"]), ":0\r\n");
        assert_eq!(f.run(&[b"HSTRLEN", b"h", b"a"]), ":1\r\n");
        assert_eq!(f.run(&[b"HSTRLEN", b"h", b"nope"]), ":0\r\n");

        // The value the client sent is `9`, so HGET h b must not find the `2`
        // that is a value. A search with a step of one would have.
        assert_eq!(f.run(&[b"HGET", b"h", b"2"]), "$-1\r\n");

        assert_eq!(f.run(&[b"HDEL", b"h", b"a", b"nope"]), ":1\r\n");
        assert_eq!(f.run(&[b"HDEL", b"h", b"b"]), ":1\r\n");
        assert_eq!(
            f.run(&[b"EXISTS", b"h"]),
            ":0\r\n",
            "and losing the last field lost the key"
        );
    }

    #[test]
    fn hgetall_answers_a_map_on_resp3_and_the_same_pairs_flat_on_resp2() {
        let mut f = Fixture::new();
        f.run(&[b"HSET", b"h", b"a", b"1"]);
        assert_eq!(f.run(&[b"HGETALL", b"h"]), "*2\r\n$1\r\na\r\n$1\r\n1\r\n");
        assert_eq!(f.run(&[b"HGETALL", b"nokey"]), "*0\r\n");
        assert_eq!(f.run(&[b"HKEYS", b"h"]), "*1\r\n$1\r\na\r\n");
        assert_eq!(f.run(&[b"HVALS", b"h"]), "*1\r\n$1\r\n1\r\n");
        assert_eq!(f.run(&[b"HKEYS", b"nokey"]), "*0\r\n");

        f.run(&[b"HELLO", b"3"]);
        assert_eq!(f.run(&[b"HGETALL", b"h"]), "%1\r\n$1\r\na\r\n$1\r\n1\r\n");
        assert_eq!(
            f.run(&[b"HGETALL", b"nokey"]),
            "%0\r\n",
            "a missing key is the empty hash and never a nil"
        );
        assert_eq!(
            f.run(&[b"HKEYS", b"h"]),
            "*1\r\n$1\r\na\r\n",
            "and the two that answer one side stay arrays"
        );
    }

    #[test]
    fn hmget_answers_once_per_field_and_hmset_answers_ok() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"HMSET", b"h", b"a", b"1", b"c", b"3"]), "+OK\r\n");
        assert_eq!(
            f.run(&[b"HMGET", b"h", b"a", b"b", b"c"]),
            "*3\r\n$1\r\n1\r\n$-1\r\n$1\r\n3\r\n",
            "the reply is positional, so b is a nil and not a gap"
        );
        assert_eq!(
            f.run(&[b"HMGET", b"nokey", b"a", b"b"]),
            "*2\r\n$-1\r\n$-1\r\n",
            "and a missing key is all nils rather than an empty array"
        );

        assert_eq!(f.run(&[b"HSETNX", b"h", b"a", b"9"]), ":0\r\n");
        assert_eq!(f.run(&[b"HSETNX", b"h", b"z", b"9"]), ":1\r\n");
        assert_eq!(f.run(&[b"HGET", b"h", b"a"]), "$1\r\n1\r\n");
    }

    #[test]
    fn a_hash_counts_up_and_says_so_when_it_cannot() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"HINCRBY", b"h", b"n", b"5"]), ":5\r\n");
        assert_eq!(f.run(&[b"HINCRBY", b"h", b"n", b"-7"]), ":-2\r\n");
        assert_eq!(f.run(&[b"HGET", b"h", b"n"]), "$2\r\n-2\r\n");
        assert_eq!(
            f.run(&[b"HINCRBYFLOAT", b"h", b"f", b"10.5"]),
            "$4\r\n10.5\r\n",
            "a bulk string and not a double, on both protocols"
        );

        f.run(&[b"HSET", b"h", b"s", b"words"]);
        let bad = f.run(&[b"HINCRBY", b"h", b"s", b"1"]);
        assert!(
            bad.starts_with("-ERR hash value is not an integer"),
            "{bad}"
        );
        let bad = f.run(&[b"HINCRBY", b"h", b"n", b"nope"]);
        assert!(
            bad.starts_with("-ERR value is not an integer"),
            "a bad argument is not yet a hash value, {bad}"
        );
        assert_eq!(
            f.run(&[b"HGET", b"h", b"s"]),
            "$5\r\nwords\r\n",
            "and neither of them wrote anything"
        );
    }

    #[test]
    fn a_hash_scan_walks_every_pair_once_and_novalues_drops_half_of_it() {
        let mut f = Fixture::new();
        for i in 0..500 {
            let field = format!("field-{i}");
            let value = format!("value-{i}");
            f.run(&[b"HSET", b"h", field.as_bytes(), value.as_bytes()]);
        }

        let mut seen: Vec<String> = Vec::new();
        let mut cursor = "0".to_owned();
        loop {
            let reply = f.run(&[b"HSCAN", b"h", cursor.as_bytes(), b"COUNT", b"32"]);
            let (next, items) = scan_reply(&reply);
            assert_eq!(items.len() % 2, 0, "a pair went out half written");
            for pair in items.chunks(2) {
                assert_eq!(
                    pair[0].strip_prefix("field-"),
                    pair[1].strip_prefix("value-"),
                    "a field came back with someone else's value"
                );
                seen.push(pair[0].clone());
            }
            cursor = next;
            if cursor == "0" {
                break;
            }
        }
        seen.sort();
        seen.dedup();
        assert_eq!(seen.len(), 500, "every field once and only once");

        let (_, items) = scan_reply(&f.run(&[b"HSCAN", b"h", b"0", b"NOVALUES", b"COUNT", b"32"]));
        assert!(
            items.iter().all(|s| s.starts_with("field-")),
            "NOVALUES still sent the values"
        );

        let (_, one) = scan_reply(&f.run(&[
            b"HSCAN",
            b"h",
            b"0",
            b"MATCH",
            b"field-499",
            b"COUNT",
            b"1000",
        ]));
        assert_eq!(one, ["field-499", "value-499"], "MATCH is on the field");
    }

    #[test]
    fn hrandfield_draws_what_it_was_asked_for_and_nests_values_on_resp3() {
        let mut f = Fixture::new();
        f.run(&[b"HSET", b"h", b"a", b"1"]);
        assert_eq!(f.run(&[b"HRANDFIELD", b"h"]), "$1\r\na\r\n");
        assert_eq!(f.run(&[b"HRANDFIELD", b"nokey"]), "$-1\r\n");
        assert_eq!(f.run(&[b"HRANDFIELD", b"nokey", b"3"]), "*0\r\n");
        assert_eq!(
            f.run(&[b"HRANDFIELD", b"h", b"3"]),
            "*1\r\n$1\r\na\r\n",
            "a positive count is capped at the size of the hash"
        );
        assert_eq!(
            f.run(&[b"HRANDFIELD", b"h", b"-3"]),
            "*3\r\n$1\r\na\r\n$1\r\na\r\n$1\r\na\r\n",
            "and a negative one repeats itself"
        );
        assert_eq!(
            f.run(&[b"HRANDFIELD", b"h", b"1", b"WITHVALUES"]),
            "*2\r\n$1\r\na\r\n$1\r\n1\r\n",
            "flat on RESP2"
        );

        f.run(&[b"HELLO", b"3"]);
        assert_eq!(
            f.run(&[b"HRANDFIELD", b"h", b"1", b"WITHVALUES"]),
            "*1\r\n*2\r\n$1\r\na\r\n$1\r\n1\r\n",
            "and nested on RESP3, but still an array and never a map"
        );
    }

    #[test]
    fn every_hash_command_says_wrongtype_and_writes_nothing() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"str", b"v"]);
        let wrong = "-WRONGTYPE Operation against a key holding the wrong kind of value\r\n";

        for cmd in [
            &[b"HSET".as_slice(), b"str", b"f", b"v"][..],
            &[b"HMSET".as_slice(), b"str", b"f", b"v"][..],
            &[b"HSETNX".as_slice(), b"str", b"f", b"v"][..],
            &[b"HGET".as_slice(), b"str", b"f"][..],
            &[b"HMGET".as_slice(), b"str", b"f"][..],
            &[b"HDEL".as_slice(), b"str", b"f"][..],
            &[b"HLEN".as_slice(), b"str"][..],
            &[b"HEXISTS".as_slice(), b"str", b"f"][..],
            &[b"HSTRLEN".as_slice(), b"str", b"f"][..],
            &[b"HGETALL".as_slice(), b"str"][..],
            &[b"HKEYS".as_slice(), b"str"][..],
            &[b"HVALS".as_slice(), b"str"][..],
            &[b"HINCRBY".as_slice(), b"str", b"f", b"1"][..],
            &[b"HINCRBYFLOAT".as_slice(), b"str", b"f", b"1"][..],
            &[b"HRANDFIELD".as_slice(), b"str"][..],
            &[b"HRANDFIELD".as_slice(), b"str", b"2"][..],
            &[b"HSCAN".as_slice(), b"str", b"0"][..],
        ] {
            let reply = f.run(cmd);
            assert_eq!(reply, wrong, "{:?}", cmd[0]);
        }
        assert_eq!(
            f.run(&[b"GET", b"str"]),
            "$1\r\nv\r\n",
            "and none of them touched the value"
        );
    }

    #[test]
    fn a_hash_scan_leaves_nothing_half_written_when_its_arguments_are_wrong() {
        let mut f = Fixture::new();
        f.run(&[b"HSET", b"h", b"f", b"v"]);
        for bad in [
            &[b"HSCAN".as_slice(), b"h", b"abc"][..],
            &[b"HSCAN".as_slice(), b"h", b"0", b"COUNT", b"nope"][..],
            &[b"HSCAN".as_slice(), b"h", b"0", b"COUNT", b"0"][..],
            &[b"HSCAN".as_slice(), b"h", b"0", b"MATCH"][..],
        ] {
            let reply = f.run(bad);
            assert!(reply.starts_with("-ERR"), "got {reply}");
            assert!(!reply.contains('*'), "an array header went out in front");
        }
    }

    #[test]
    fn a_field_deadline_goes_on_and_comes_back_in_all_four_units() {
        let mut f = Fixture::new();
        f.run(&[b"HSET", b"h", b"a", b"1", b"b", b"2"]);
        assert_eq!(
            f.run(&[b"HEXPIRE", b"h", b"100", b"FIELDS", b"1", b"a"]),
            "*1\r\n:1\r\n"
        );
        assert_eq!(
            f.run(&[b"HTTL", b"h", b"FIELDS", b"3", b"a", b"b", b"nope"]),
            "*3\r\n:100\r\n:-1\r\n:-2\r\n",
            "one answer per field, and the two sentinels are TTL's own"
        );

        // The same deadline in the other three units, all of them derived from
        // the one number the store kept.
        let ms = int_reply(&f.run(&[b"HPTTL", b"h", b"FIELDS", b"1", b"a"]));
        assert!((99_000..=100_000).contains(&ms), "got {ms}");
        let at = int_reply(&f.run(&[b"HEXPIRETIME", b"h", b"FIELDS", b"1", b"a"]));
        let at_ms = int_reply(&f.run(&[b"HPEXPIRETIME", b"h", b"FIELDS", b"1", b"a"]));
        assert_eq!(at, at_ms.div_euclid(1000) + i64::from(at_ms % 1000 != 0));
        assert!(at_ms > 1_700_000_000_000, "an absolute moment, got {at_ms}");

        assert_eq!(
            f.run(&[b"HPERSIST", b"h", b"FIELDS", b"3", b"a", b"b", b"nope"]),
            "*3\r\n:1\r\n:-1\r\n:-2\r\n",
            "one for the deadline taken off, and it does not say what it was"
        );
        assert_eq!(
            f.run(&[b"HTTL", b"h", b"FIELDS", b"1", b"a"]),
            "*1\r\n:-1\r\n"
        );
        assert_eq!(
            f.run(&[b"HGET", b"h", b"a"]),
            "$1\r\n1\r\n",
            "and the field is still there with the value it had"
        );
    }

    #[test]
    fn a_deadline_that_has_already_gone_deletes_the_field_now() {
        let mut f = Fixture::new();
        f.run(&[b"HSET", b"h", b"a", b"1", b"b", b"2"]);
        assert_eq!(
            f.run(&[b"HEXPIREAT", b"h", b"1", b"FIELDS", b"1", b"a"]),
            "*1\r\n:2\r\n",
            "two, and not one, because nothing was stored"
        );
        assert_eq!(f.run(&[b"HGET", b"h", b"a"]), "$-1\r\n");
        assert_eq!(f.run(&[b"HLEN", b"h"]), ":1\r\n");

        assert_eq!(
            f.run(&[b"HPEXPIREAT", b"h", b"1", b"FIELDS", b"1", b"b"]),
            "*1\r\n:2\r\n"
        );
        assert_eq!(
            f.run(&[b"EXISTS", b"h"]),
            ":0\r\n",
            "and the last field going took the key with it"
        );

        // Zero is a delete and not an error, where minus one is an error. That
        // is Redis's split and it is easy to get backwards.
        f.run(&[b"HSET", b"h", b"a", b"1"]);
        assert_eq!(
            f.run(&[b"HEXPIRE", b"h", b"0", b"FIELDS", b"1", b"a"]),
            "*1\r\n:2\r\n"
        );
    }

    #[test]
    fn a_field_is_gone_once_its_moment_passes() {
        let mut f = Fixture::new();
        f.run(&[b"HSET", b"h", b"a", b"1", b"b", b"2"]);
        assert_eq!(
            f.run(&[b"HPEXPIRE", b"h", b"20", b"FIELDS", b"1", b"a"]),
            "*1\r\n:1\r\n"
        );
        assert_eq!(f.run(&[b"HGET", b"h", b"a"]), "$1\r\n1\r\n", "not yet");

        // Time moves once per turn of the event loop and nowhere else, so a
        // test moves it by hand rather than by sleeping. There is nothing to
        // sleep for: the deadline is a number and so is the clock.
        f.server.db(0).clock_mut().advance(60);
        assert_eq!(f.run(&[b"HLEN", b"h"]), ":1\r\n");
        assert_eq!(f.run(&[b"HGET", b"h", b"a"]), "$-1\r\n");
        assert_eq!(
            f.run(&[b"HGETALL", b"h"]),
            "*2\r\n$1\r\nb\r\n$1\r\n2\r\n",
            "and the walks do not hand back a field that has expired"
        );
    }

    #[test]
    fn a_missing_key_answers_the_no_field_sentinel_for_every_field() {
        let mut f = Fixture::new();
        for cmd in [
            &[
                b"HEXPIRE".as_slice(),
                b"nokey",
                b"100",
                b"FIELDS",
                b"2",
                b"a",
                b"b",
            ][..],
            &[b"HTTL".as_slice(), b"nokey", b"FIELDS", b"2", b"a", b"b"][..],
            &[b"HPTTL".as_slice(), b"nokey", b"FIELDS", b"2", b"a", b"b"][..],
            &[
                b"HEXPIRETIME".as_slice(),
                b"nokey",
                b"FIELDS",
                b"2",
                b"a",
                b"b",
            ][..],
            &[
                b"HPERSIST".as_slice(),
                b"nokey",
                b"FIELDS",
                b"2",
                b"a",
                b"b",
            ][..],
        ] {
            assert_eq!(f.run(cmd), "*2\r\n:-2\r\n:-2\r\n", "{:?}", cmd[0]);
        }
    }

    #[test]
    fn writing_a_field_clears_the_deadline_that_was_on_it() {
        let mut f = Fixture::new();
        f.run(&[b"HSET", b"h", b"a", b"1"]);
        f.run(&[b"HEXPIRE", b"h", b"100", b"FIELDS", b"1", b"a"]);
        f.run(&[b"HSET", b"h", b"a", b"2"]);
        assert_eq!(
            f.run(&[b"HTTL", b"h", b"FIELDS", b"1", b"a"]),
            "*1\r\n:-1\r\n",
            "Redis has done this since 7.4, and it is why HGETEX exists"
        );
    }

    #[test]
    fn the_four_conditions_reach_the_store_the_way_they_were_written() {
        let mut f = Fixture::new();
        f.run(&[b"HSET", b"h", b"a", b"1"]);
        assert_eq!(
            f.run(&[b"HEXPIRE", b"h", b"100", b"XX", b"FIELDS", b"1", b"a"]),
            "*1\r\n:0\r\n",
            "XX on a field with no deadline changes nothing"
        );
        assert_eq!(
            f.run(&[b"HEXPIRE", b"h", b"100", b"NX", b"FIELDS", b"1", b"a"]),
            "*1\r\n:1\r\n"
        );
        assert_eq!(
            f.run(&[b"HEXPIRE", b"h", b"200", b"NX", b"FIELDS", b"1", b"a"]),
            "*1\r\n:0\r\n",
            "and NX will not move one that is already there"
        );
        assert_eq!(
            f.run(&[b"HEXPIRE", b"h", b"50", b"GT", b"FIELDS", b"1", b"a"]),
            "*1\r\n:0\r\n"
        );
        assert_eq!(
            f.run(&[b"HEXPIRE", b"h", b"500", b"GT", b"FIELDS", b"1", b"a"]),
            "*1\r\n:1\r\n"
        );
        assert_eq!(
            f.run(&[b"HEXPIRE", b"h", b"50", b"LT", b"FIELDS", b"1", b"a"]),
            "*1\r\n:1\r\n"
        );
        assert_eq!(
            f.run(&[b"HTTL", b"h", b"FIELDS", b"1", b"a"]),
            "*1\r\n:50\r\n"
        );
    }

    #[test]
    fn the_field_ttl_family_leaves_nothing_half_written_on_a_bad_argument() {
        let mut f = Fixture::new();
        f.run(&[b"HSET", b"h", b"a", b"1"]);
        for (bad, want) in [
            (
                &[b"HEXPIRE".as_slice(), b"h", b"-1", b"FIELDS", b"1", b"a"][..],
                "-ERR invalid expire time, must be >= 0",
            ),
            (
                &[
                    b"HEXPIRE".as_slice(),
                    b"h",
                    b"9999999999999999",
                    b"FIELDS",
                    b"1",
                    b"a",
                ][..],
                "-ERR invalid expire time in 'hexpire' command",
            ),
            (
                &[b"HEXPIRE".as_slice(), b"h", b"100", b"FIELD", b"1", b"a"][..],
                "-ERR wrong number of arguments for 'hexpire' command",
            ),
            (
                &[b"HEXPIRE".as_slice(), b"h", b"100", b"FIELDS", b"0", b"a"][..],
                "-ERR Parameter `numFields` should be greater than 0",
            ),
            (
                &[b"HEXPIRE".as_slice(), b"h", b"100", b"FIELDS", b"2", b"a"][..],
                "-ERR wrong number of arguments",
            ),
            (
                &[b"HTTL".as_slice(), b"h", b"FIELDS", b"3", b"a", b"b"][..],
                "-ERR wrong number of arguments",
            ),
        ] {
            let reply = f.run(bad);
            assert!(reply.starts_with(want), "wanted {want}, got {reply}");
            assert!(!reply.contains('*'), "an array header went out in front");
        }
        assert_eq!(
            f.run(&[b"HTTL", b"h", b"FIELDS", b"1", b"a"]),
            "*1\r\n:-1\r\n",
            "and not one of them put a deadline on anything"
        );
    }

    #[test]
    fn every_field_ttl_command_says_wrongtype_and_writes_nothing() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"str", b"v"]);
        let wrong = "-WRONGTYPE Operation against a key holding the wrong kind of value\r\n";

        for cmd in [
            &[b"HEXPIRE".as_slice(), b"str", b"100", b"FIELDS", b"1", b"f"][..],
            &[
                b"HPEXPIRE".as_slice(),
                b"str",
                b"100",
                b"FIELDS",
                b"1",
                b"f",
            ][..],
            &[
                b"HEXPIREAT".as_slice(),
                b"str",
                b"9999999999",
                b"FIELDS",
                b"1",
                b"f",
            ][..],
            &[
                b"HPEXPIREAT".as_slice(),
                b"str",
                b"9999999999999",
                b"FIELDS",
                b"1",
                b"f",
            ][..],
            &[b"HTTL".as_slice(), b"str", b"FIELDS", b"1", b"f"][..],
            &[b"HPTTL".as_slice(), b"str", b"FIELDS", b"1", b"f"][..],
            &[b"HEXPIRETIME".as_slice(), b"str", b"FIELDS", b"1", b"f"][..],
            &[b"HPEXPIRETIME".as_slice(), b"str", b"FIELDS", b"1", b"f"][..],
            &[b"HPERSIST".as_slice(), b"str", b"FIELDS", b"1", b"f"][..],
        ] {
            assert_eq!(f.run(cmd), wrong, "{:?}", cmd[0]);
        }
        assert_eq!(
            f.run(&[b"GET", b"str"]),
            "$1\r\nv\r\n",
            "and none of them touched the value"
        );
    }

    #[test]
    fn hgetdel_hands_the_value_out_and_then_takes_the_field() {
        let mut f = Fixture::new();
        f.run(&[b"HSET", b"h", b"a", b"1", b"b", b"2"]);
        assert_eq!(
            f.run(&[b"HGETDEL", b"h", b"FIELDS", b"2", b"a", b"nope"]),
            "*2\r\n$1\r\n1\r\n$-1\r\n",
            "positional, so the field that was not there is a nil in its place"
        );
        assert_eq!(f.run(&[b"HLEN", b"h"]), ":1\r\n");
        assert_eq!(
            f.run(&[b"HGETDEL", b"nokey", b"FIELDS", b"1", b"a"]),
            "*1\r\n$-1\r\n"
        );
        assert_eq!(
            f.run(&[b"HGETDEL", b"h", b"FIELDS", b"1", b"b"]),
            "*1\r\n$1\r\n2\r\n"
        );
        assert_eq!(
            f.run(&[b"EXISTS", b"h"]),
            ":0\r\n",
            "and the last field took the key"
        );
    }

    #[test]
    fn hgetex_reads_and_moves_the_deadline_in_one_command() {
        let mut f = Fixture::new();
        f.run(&[b"HSET", b"h", b"a", b"1"]);
        assert_eq!(
            f.run(&[b"HGETEX", b"h", b"FIELDS", b"1", b"a"]),
            "*1\r\n$1\r\n1\r\n"
        );
        assert_eq!(
            f.run(&[b"HTTL", b"h", b"FIELDS", b"1", b"a"]),
            "*1\r\n:-1\r\n",
            "no option means leave it alone, which is the one place this is not GETEX"
        );

        f.run(&[b"HGETEX", b"h", b"EX", b"100", b"FIELDS", b"1", b"a"]);
        assert_eq!(
            f.run(&[b"HTTL", b"h", b"FIELDS", b"1", b"a"]),
            "*1\r\n:100\r\n"
        );
        f.run(&[b"HGETEX", b"h", b"FIELDS", b"1", b"a"]);
        assert_eq!(
            f.run(&[b"HTTL", b"h", b"FIELDS", b"1", b"a"]),
            "*1\r\n:100\r\n",
            "and a plain read really does leave it alone"
        );
        assert_eq!(
            f.run(&[b"HGETEX", b"h", b"PERSIST", b"FIELDS", b"1", b"a"]),
            "*1\r\n$1\r\n1\r\n"
        );
        assert_eq!(
            f.run(&[b"HTTL", b"h", b"FIELDS", b"1", b"a"]),
            "*1\r\n:-1\r\n"
        );

        assert_eq!(
            f.run(&[b"HGETEX", b"h", b"EXAT", b"1", b"FIELDS", b"1", b"a"]),
            "*1\r\n$1\r\n1\r\n",
            "the value goes out before the deadline that has already gone is applied"
        );
        assert_eq!(f.run(&[b"EXISTS", b"h"]), ":0\r\n");
        assert_eq!(
            f.run(&[b"HGETEX", b"nokey", b"EX", b"100", b"FIELDS", b"1", b"a"]),
            "*1\r\n$-1\r\n"
        );
    }

    #[test]
    fn hsetex_writes_all_of_it_or_none_of_it() {
        let mut f = Fixture::new();
        assert_eq!(
            f.run(&[b"HSETEX", b"h", b"FIELDS", b"1", b"a", b"1"]),
            ":1\r\n"
        );
        assert_eq!(f.run(&[b"HGET", b"h", b"a"]), "$1\r\n1\r\n");
        assert_eq!(
            f.run(&[
                b"HSETEX", b"h", b"FNX", b"FIELDS", b"2", b"a", b"9", b"new", b"9"
            ]),
            ":0\r\n",
            "FNX wants every field named to be missing"
        );
        assert_eq!(f.run(&[b"HGET", b"h", b"a"]), "$1\r\n1\r\n");
        assert_eq!(
            f.run(&[b"HEXISTS", b"h", b"new"]),
            ":0\r\n",
            "and none of the list was written"
        );
        assert_eq!(
            f.run(&[
                b"HSETEX", b"h", b"FXX", b"FIELDS", b"2", b"a", b"9", b"nope", b"9"
            ]),
            ":0\r\n",
            "and FXX wants every one of them to be there"
        );
        assert_eq!(f.run(&[b"HGET", b"h", b"a"]), "$1\r\n1\r\n");
        assert_eq!(
            f.run(&[b"HSETEX", b"h", b"FXX", b"FIELDS", b"1", b"a", b"9"]),
            ":1\r\n"
        );
        assert_eq!(f.run(&[b"HGET", b"h", b"a"]), "$1\r\n9\r\n");

        assert_eq!(
            f.run(&[b"HSETEX", b"gone", b"FXX", b"FIELDS", b"1", b"a", b"1"]),
            ":0\r\n"
        );
        assert_eq!(
            f.run(&[b"EXISTS", b"gone"]),
            ":0\r\n",
            "a key with no fields cannot meet FXX and is not created trying"
        );
    }

    #[test]
    fn hsetex_clears_the_deadline_unless_it_is_told_to_keep_it() {
        let mut f = Fixture::new();
        f.run(&[b"HSETEX", b"h", b"EX", b"100", b"FIELDS", b"1", b"a", b"1"]);
        assert_eq!(
            f.run(&[b"HTTL", b"h", b"FIELDS", b"1", b"a"]),
            "*1\r\n:100\r\n"
        );

        f.run(&[b"HSETEX", b"h", b"KEEPTTL", b"FIELDS", b"1", b"a", b"2"]);
        assert_eq!(f.run(&[b"HGET", b"h", b"a"]), "$1\r\n2\r\n");
        assert_eq!(
            f.run(&[b"HTTL", b"h", b"FIELDS", b"1", b"a"]),
            "*1\r\n:100\r\n",
            "KEEPTTL put back what the write cleared"
        );

        f.run(&[b"HSETEX", b"h", b"FIELDS", b"1", b"a", b"3"]);
        assert_eq!(
            f.run(&[b"HTTL", b"h", b"FIELDS", b"1", b"a"]),
            "*1\r\n:-1\r\n",
            "and without it a write clears the deadline the way HSET does"
        );

        // Any order, because Redis reads these in a loop and not in a fixed
        // sequence.
        assert_eq!(
            f.run(&[
                b"HSETEX", b"h", b"PX", b"100000", b"FXX", b"FIELDS", b"1", b"a", b"4"
            ]),
            ":1\r\n"
        );
        assert_eq!(
            f.run(&[b"HTTL", b"h", b"FIELDS", b"1", b"a"]),
            "*1\r\n:100\r\n"
        );

        assert_eq!(
            f.run(&[b"HSETEX", b"h", b"EXAT", b"1", b"FIELDS", b"1", b"a", b"5"]),
            ":1\r\n",
            "written, and not the separate code the HEXPIRE family has for this"
        );
        assert_eq!(
            f.run(&[b"EXISTS", b"h"]),
            ":0\r\n",
            "and storing it and then removing it emptied the hash"
        );
    }

    #[test]
    fn the_last_three_hash_commands_word_their_mistakes_their_own_way() {
        let mut f = Fixture::new();
        f.run(&[b"HSET", b"h", b"a", b"1"]);
        for (bad, want) in [
            // HGETDEL has three sentences of its own for these three mistakes.
            (
                &[b"HGETDEL".as_slice(), b"h", b"FIELDS", b"0", b"a"][..],
                "-ERR Number of fields must be a positive integer",
            ),
            (
                &[b"HGETDEL".as_slice(), b"h", b"FIELDS", b"2", b"a"][..],
                "-ERR The `numfields` parameter must match the number of arguments",
            ),
            (
                &[b"HGETDEL".as_slice(), b"h", b"FIELD", b"1", b"a"][..],
                "-ERR Mandatory argument FIELDS is missing or not at the right position",
            ),
            // And HGETEX and HSETEX have three different ones between them.
            (
                &[b"HGETEX".as_slice(), b"h", b"FIELDS", b"0", b"a"][..],
                "-ERR invalid number of fields",
            ),
            (
                &[b"HGETEX".as_slice(), b"h", b"FIELDS", b"2", b"a"][..],
                "-ERR wrong number of arguments",
            ),
            (
                &[b"HGETEX".as_slice(), b"h", b"FIELD", b"1", b"a"][..],
                "-ERR unknown argument: FIELD",
            ),
            (
                &[
                    b"HGETEX".as_slice(),
                    b"h",
                    b"KEEPTTL",
                    b"FIELDS",
                    b"1",
                    b"a",
                ][..],
                "-ERR unknown argument: KEEPTTL",
            ),
            (
                &[
                    b"HGETEX".as_slice(),
                    b"h",
                    b"EX",
                    b"100",
                    b"PERSIST",
                    b"FIELDS",
                    b"1",
                    b"a",
                ][..],
                "-ERR Only one of EX, PX, EXAT, PXAT or PERSIST arguments can be specified",
            ),
            (
                &[
                    b"HSETEX".as_slice(),
                    b"h",
                    b"EX",
                    b"1",
                    b"KEEPTTL",
                    b"FIELDS",
                    b"1",
                    b"a",
                    b"1",
                ][..],
                "-ERR Only one of EX, PX, EXAT, PXAT or KEEPTTL arguments can be specified",
            ),
            (
                &[
                    b"HSETEX".as_slice(),
                    b"h",
                    b"FNX",
                    b"FXX",
                    b"FIELDS",
                    b"1",
                    b"a",
                    b"1",
                ][..],
                "-ERR Only one of FXX or FNX arguments can be specified",
            ),
            (
                &[
                    b"HSETEX".as_slice(),
                    b"h",
                    b"FIELDS",
                    b"2",
                    b"a",
                    b"1",
                    b"b",
                ][..],
                "-ERR wrong number of arguments",
            ),
            (
                &[
                    b"HGETEX".as_slice(),
                    b"h",
                    b"EX",
                    b"-1",
                    b"FIELDS",
                    b"1",
                    b"a",
                ][..],
                "-ERR invalid expire time, must be >= 0",
            ),
            (
                &[
                    b"HGETEX".as_slice(),
                    b"h",
                    b"PXAT",
                    b"99999999999999",
                    b"FIELDS",
                    b"1",
                    b"a",
                ][..],
                "-ERR invalid expire time in 'hgetex' command",
            ),
            (
                &[
                    b"HSETEX".as_slice(),
                    b"h",
                    b"EX",
                    b"abc",
                    b"FIELDS",
                    b"1",
                    b"a",
                    b"1",
                ][..],
                "-ERR value is not an integer or out of range",
            ),
        ] {
            let reply = f.run(bad);
            assert!(reply.starts_with(want), "wanted {want}, got {reply}");
            assert!(!reply.contains('*'), "an array header went out in front");
        }
        assert_eq!(
            f.run(&[b"HGET", b"h", b"a"]),
            "$1\r\n1\r\n",
            "and not one of them wrote anything"
        );
        assert_eq!(
            f.run(&[b"HTTL", b"h", b"FIELDS", b"1", b"a"]),
            "*1\r\n:-1\r\n"
        );
    }

    #[test]
    fn the_last_three_hash_commands_say_wrongtype_and_write_nothing() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"str", b"v"]);
        let wrong = "-WRONGTYPE Operation against a key holding the wrong kind of value\r\n";
        for cmd in [
            &[b"HGETDEL".as_slice(), b"str", b"FIELDS", b"1", b"f"][..],
            &[b"HGETEX".as_slice(), b"str", b"FIELDS", b"1", b"f"][..],
            &[
                b"HGETEX".as_slice(),
                b"str",
                b"EX",
                b"100",
                b"FIELDS",
                b"1",
                b"f",
            ][..],
            &[b"HSETEX".as_slice(), b"str", b"FIELDS", b"1", b"f", b"v"][..],
        ] {
            assert_eq!(f.run(cmd), wrong, "{:?}", cmd[0]);
        }
        assert_eq!(f.run(&[b"GET", b"str"]), "$1\r\nv\r\n");
    }

    /// The one integer of a single element array reply.
    /// The number out of a plain integer reply.
    ///
    /// [`int_reply`] is the same thing wrapped in a one element array, which is
    /// the shape every hash field command answers in.
    fn int(reply: &str) -> i64 {
        let body = reply
            .strip_prefix(':')
            .and_then(|s| s.strip_suffix("\r\n"))
            .unwrap_or_else(|| panic!("wanted an integer, got {reply}"));
        body.parse().expect("an integer")
    }

    fn int_reply(reply: &str) -> i64 {
        let body = reply
            .strip_prefix("*1\r\n:")
            .and_then(|s| s.strip_suffix("\r\n"))
            .unwrap_or_else(|| panic!("wanted one integer, got {reply}"));
        body.parse().expect("an integer")
    }

    /// The cursor and the flat items of a scan reply.
    fn scan_reply(reply: &str) -> (String, Vec<String>) {
        let mut lines = reply.split("\r\n");
        assert_eq!(lines.next(), Some("*2"), "got {reply}");
        lines.next().expect("the cursor header");
        let cursor = lines.next().expect("a cursor").to_owned();
        let header = lines.next().expect("an item count");
        let n: usize = header[1..].parse().expect("a count");
        let mut items = Vec::with_capacity(n);
        for _ in 0..n {
            lines.next().expect("an item header");
            items.push(lines.next().expect("an item").to_owned());
        }
        (cursor, items)
    }

    /// The members of a set reply, sorted, since none of these promise an
    /// order and a test that asserted one would be asserting an accident.
    fn sorted(reply: &str) -> Vec<String> {
        let mut lines = reply.split("\r\n");
        let header = lines.next().expect("a header");
        assert!(
            header.starts_with('*') || header.starts_with('~'),
            "got {reply}"
        );
        let n: usize = header[1..].parse().expect("a member count");
        let mut got = Vec::with_capacity(n);
        for _ in 0..n {
            lines.next().expect("a member header");
            got.push(lines.next().expect("a member").to_owned());
        }
        got.sort();
        got
    }

    #[test]
    fn the_algebra_answers_what_the_sets_share_and_do_not() {
        let mut f = Fixture::new();
        f.run(&[b"SADD", b"a", b"1", b"2", b"3"]);
        f.run(&[b"SADD", b"b", b"2", b"3", b"4"]);
        f.run(&[b"SADD", b"c", b"3", b"4", b"5"]);

        assert_eq!(sorted(&f.run(&[b"SINTER", b"a", b"b", b"c"])), ["3"]);
        assert_eq!(
            sorted(&f.run(&[b"SUNION", b"a", b"b", b"c"])),
            ["1", "2", "3", "4", "5"]
        );
        assert_eq!(sorted(&f.run(&[b"SDIFF", b"a", b"b"])), ["1"]);
        assert_eq!(sorted(&f.run(&[b"SINTER", b"a"])), ["1", "2", "3"]);

        // A key that is not there is an empty set, which empties an
        // intersection and does nothing at all to a union.
        assert_eq!(f.run(&[b"SINTER", b"a", b"nope"]), "*0\r\n");
        assert_eq!(sorted(&f.run(&[b"SUNION", b"a", b"nope"])), ["1", "2", "3"]);
        assert_eq!(f.run(&[b"SDIFF", b"nope", b"a"]), "*0\r\n");
        assert_eq!(f.run(&[b"DBSIZE"]), ":3\r\n", "and none of it made a key");
    }

    #[test]
    fn the_algebra_answers_a_set_on_resp3_and_an_array_on_resp2() {
        let mut f = Fixture::new();
        f.run(&[b"SADD", b"a", b"x"]);
        assert_eq!(f.run(&[b"SINTER", b"a"]), "*1\r\n$1\r\nx\r\n");
        assert_eq!(f.run(&[b"SUNION", b"a"]), "*1\r\n$1\r\nx\r\n");
        assert_eq!(f.run(&[b"SDIFF", b"a"]), "*1\r\n$1\r\nx\r\n");

        f.run(&[b"HELLO", b"3"]);
        assert_eq!(f.run(&[b"SINTER", b"a"]), "~1\r\n$1\r\nx\r\n");
        assert_eq!(f.run(&[b"SUNION", b"a"]), "~1\r\n$1\r\nx\r\n");
        assert_eq!(f.run(&[b"SDIFF", b"a"]), "~1\r\n$1\r\nx\r\n");
        assert_eq!(f.run(&[b"SINTER", b"nope"]), "~0\r\n");
    }

    #[test]
    fn a_store_form_writes_a_key_and_answers_how_big_it_is() {
        let mut f = Fixture::new();
        f.run(&[b"SADD", b"a", b"1", b"2", b"3"]);
        f.run(&[b"SADD", b"b", b"2", b"3", b"4"]);

        assert_eq!(f.run(&[b"SINTERSTORE", b"d", b"a", b"b"]), ":2\r\n");
        assert_eq!(sorted(&f.run(&[b"SMEMBERS", b"d"])), ["2", "3"]);
        assert_eq!(f.run(&[b"SUNIONSTORE", b"d", b"a", b"b"]), ":4\r\n");
        assert_eq!(sorted(&f.run(&[b"SMEMBERS", b"d"])), ["1", "2", "3", "4"]);
        assert_eq!(f.run(&[b"SDIFFSTORE", b"d", b"a", b"b"]), ":1\r\n");
        assert_eq!(f.run(&[b"SMEMBERS", b"d"]), "*1\r\n$1\r\n1\r\n");

        // An empty answer deletes the destination rather than leaving an empty
        // set behind, and the destination may be one of the sources.
        assert_eq!(f.run(&[b"SDIFFSTORE", b"d", b"a", b"a"]), ":0\r\n");
        assert_eq!(f.run(&[b"EXISTS", b"d"]), ":0\r\n");
        assert_eq!(f.run(&[b"SINTERSTORE", b"a", b"a", b"b"]), ":2\r\n");
        assert_eq!(sorted(&f.run(&[b"SMEMBERS", b"a"])), ["2", "3"]);

        // And a destination holding something else is overwritten, the same way
        // SET overwrites, rather than refused.
        f.run(&[b"SET", b"str", b"v"]);
        assert_eq!(f.run(&[b"SUNIONSTORE", b"str", b"b"]), ":3\r\n");
        assert_eq!(f.run(&[b"TYPE", b"str"]), "+set\r\n");
    }

    #[test]
    fn sintercard_counts_without_building_and_stops_at_a_limit() {
        let mut f = Fixture::new();
        f.run(&[b"SADD", b"a", b"1", b"2", b"3", b"4"]);
        f.run(&[b"SADD", b"b", b"2", b"3", b"4", b"5"]);

        assert_eq!(f.run(&[b"SINTERCARD", b"2", b"a", b"b"]), ":3\r\n");
        assert_eq!(
            f.run(&[b"SINTERCARD", b"2", b"a", b"b", b"LIMIT", b"2"]),
            ":2\r\n"
        );
        assert_eq!(
            f.run(&[b"SINTERCARD", b"2", b"a", b"b", b"LIMIT", b"0"]),
            ":3\r\n",
            "a limit of zero is no limit"
        );
        assert_eq!(f.run(&[b"SINTERCARD", b"1", b"a"]), ":4\r\n");
        assert_eq!(f.run(&[b"SINTERCARD", b"2", b"a", b"nope"]), ":0\r\n");

        // The counted keys are what make its three error messages its own.
        assert_eq!(
            f.run(&[b"SINTERCARD", b"0", b"a"]),
            "-ERR numkeys should be greater than 0\r\n"
        );
        assert_eq!(
            f.run(&[b"SINTERCARD", b"abc", b"a"]),
            "-ERR numkeys should be greater than 0\r\n"
        );
        assert_eq!(
            f.run(&[b"SINTERCARD", b"3", b"a", b"b"]),
            "-ERR Number of keys can't be greater than number of args\r\n"
        );
        assert_eq!(
            f.run(&[b"SINTERCARD", b"2", b"a", b"b", b"LIMIT", b"-1"]),
            "-ERR LIMIT can't be negative\r\n"
        );
        assert_eq!(
            f.run(&[b"SINTERCARD", b"2", b"a", b"b", b"NOPE", b"1"]),
            "-ERR syntax error\r\n"
        );
        // A key really can be called LIMIT, which is why the count exists.
        f.run(&[b"SADD", b"LIMIT", b"2"]);
        assert_eq!(f.run(&[b"SINTERCARD", b"2", b"a", b"LIMIT"]), ":1\r\n");
    }

    #[test]
    fn the_algebra_answers_wrongtype_before_it_writes_anything() {
        let mut f = Fixture::new();
        f.run(&[b"SADD", b"a", b"1"]);
        f.run(&[b"SADD", b"d", b"old"]);
        f.run(&[b"SET", b"str", b"v"]);

        let wrong = "-WRONGTYPE Operation against a key holding the wrong kind of value\r\n";
        for bad in [
            &[b"SINTER".as_slice(), b"a", b"str"][..],
            &[b"SUNION".as_slice(), b"str"][..],
            &[b"SDIFF".as_slice(), b"a", b"str"][..],
            &[b"SINTERCARD".as_slice(), b"2", b"a", b"str"][..],
            &[b"SINTERSTORE".as_slice(), b"d", b"a", b"str"][..],
            &[b"SUNIONSTORE".as_slice(), b"d", b"str"][..],
            &[b"SDIFFSTORE".as_slice(), b"d", b"a", b"str"][..],
        ] {
            let reply = f.run(bad);
            assert_eq!(reply, wrong, "for {:?}", bad[0]);
        }
        assert_eq!(
            f.run(&[b"SMEMBERS", b"d"]),
            "*1\r\n$3\r\nold\r\n",
            "and the destination was left alone every time"
        );
    }

    /// The leak a set can spring that nothing on the wire would ever show: the
    /// key goes, the body does not, and `DBSIZE` looks right the whole time.
    #[test]
    fn churning_sets_does_not_grow_the_server() {
        let mut f = Fixture::new();
        let members: Vec<Vec<u8>> = (0..200).map(|i| format!("m{i}").into_bytes()).collect();
        let args: Vec<&[u8]> = std::iter::once(&b"SADD"[..])
            .chain(std::iter::once(&b"s"[..]))
            .chain(members.iter().map(Vec::as_slice))
            .collect();

        f.run(&args);
        f.run(&[b"DEL", b"s"]);
        f.server.compact_step();
        let after_first = f.server.memory_bytes();

        for _ in 0..200 {
            f.run(&args);
            f.run(&[b"DEL", b"s"]);
            f.server.compact_step();
        }
        assert_eq!(f.run(&[b"DBSIZE"]), ":0\r\n");
        assert!(
            f.server.memory_bytes() <= after_first * 2,
            "held {} after two hundred passes against {after_first} after one",
            f.server.memory_bytes()
        );
    }

    /// A RESP2 array of bulk strings, which is what most of the list replies
    /// are and what writing them out by hand in every assertion looks like.
    fn bulks(parts: &[&str]) -> String {
        let mut s = format!("*{}\r\n", parts.len());
        for p in parts {
            s.push_str(&format!("${}\r\n{p}\r\n", p.len()));
        }
        s
    }

    #[test]
    fn a_list_is_pushed_from_both_ends_and_the_left_one_reverses() {
        let mut f = Fixture::new();
        // Each element in turn goes at the head, so the last one sent is at the
        // front when it is over. That reads like a bug in the client and it is
        // what every Redis has always done.
        assert_eq!(f.run(&[b"LPUSH", b"k", b"a", b"b", b"c"]), ":3\r\n");
        assert_eq!(
            f.run(&[b"LRANGE", b"k", b"0", b"-1"]),
            bulks(&["c", "b", "a"])
        );
        assert_eq!(f.run(&[b"RPUSH", b"k", b"d"]), ":4\r\n");
        assert_eq!(f.run(&[b"LLEN", b"k"]), ":4\r\n");
        assert_eq!(f.run(&[b"LPOP", b"k"]), "$1\r\nc\r\n");
        assert_eq!(f.run(&[b"RPOP", b"k"]), "$1\r\nd\r\n");
        assert_eq!(f.run(&[b"LRANGE", b"k", b"0", b"-1"]), bulks(&["b", "a"]));
        assert_eq!(f.run(&[b"TYPE", b"k"]), "+list\r\n");
    }

    #[test]
    fn the_x_pushes_refuse_to_bring_a_list_back_to_life() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"LPUSHX", b"k", b"a"]), ":0\r\n");
        assert_eq!(f.run(&[b"RPUSHX", b"k", b"a"]), ":0\r\n");
        assert_eq!(f.run(&[b"EXISTS", b"k"]), ":0\r\n");
        f.run(&[b"RPUSH", b"k", b"a"]);
        assert_eq!(f.run(&[b"LPUSHX", b"k", b"z"]), ":2\r\n");
        assert_eq!(f.run(&[b"RPUSHX", b"k", b"y"]), ":3\r\n");
        assert_eq!(
            f.run(&[b"LRANGE", b"k", b"0", b"-1"]),
            bulks(&["z", "a", "y"])
        );
    }

    /// The four ways a pop can come back with nothing, which are three
    /// different replies and a RESP2 client can tell all of them apart.
    #[test]
    fn an_empty_pop_is_a_different_nothing_with_a_count_and_without() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"LPOP", b"nope"]), "$-1\r\n");
        assert_eq!(f.run(&[b"LPOP", b"nope", b"2"]), "*-1\r\n");
        assert_eq!(f.run(&[b"RPOP", b"nope"]), "$-1\r\n");
        assert_eq!(f.run(&[b"RPOP", b"nope", b"2"]), "*-1\r\n");
        f.run(&[b"RPUSH", b"k", b"a", b"b", b"c"]);
        // A count of zero against a list that is there is an empty array and
        // not a null array, which is the fourth answer.
        assert_eq!(f.run(&[b"LPOP", b"k", b"0"]), "*0\r\n");
        assert_eq!(f.run(&[b"LPOP", b"k", b"1"]), bulks(&["a"]));
        // More than there is takes what there is and the key goes with it.
        assert_eq!(f.run(&[b"RPOP", b"k", b"9"]), bulks(&["c", "b"]));
        assert_eq!(f.run(&[b"EXISTS", b"k"]), ":0\r\n");
    }

    #[test]
    fn a_pop_count_has_its_own_sentence_and_a_third_argument_is_an_arity_error() {
        let mut f = Fixture::new();
        f.run(&[b"RPUSH", b"k", b"a"]);
        let range = "-ERR value is out of range, must be positive\r\n";
        assert_eq!(f.run(&[b"LPOP", b"k", b"-1"]), range);
        assert_eq!(f.run(&[b"LPOP", b"k", b"abc"]), range);
        assert_eq!(f.run(&[b"RPOP", b"k", b"-1"]), range);
        // Redis calls this an arity error and not a syntax error, which is a
        // distinction it does not always make.
        assert_eq!(
            f.run(&[b"LPOP", b"k", b"1", b"2"]),
            "-ERR wrong number of arguments for 'lpop' command\r\n"
        );
        assert_eq!(f.run(&[b"LLEN", b"k"]), ":1\r\n");
    }

    #[test]
    fn a_range_takes_negative_ends_and_clamps_the_ones_that_run_off() {
        let mut f = Fixture::new();
        f.run(&[b"RPUSH", b"k", b"a", b"b", b"c"]);
        assert_eq!(
            f.run(&[b"LRANGE", b"k", b"0", b"-1"]),
            bulks(&["a", "b", "c"])
        );
        assert_eq!(f.run(&[b"LRANGE", b"k", b"-2", b"-1"]), bulks(&["b", "c"]));
        assert_eq!(f.run(&[b"LRANGE", b"k", b"1", b"1"]), bulks(&["b"]));
        assert_eq!(f.run(&[b"LRANGE", b"k", b"5", b"10"]), "*0\r\n");
        assert_eq!(f.run(&[b"LRANGE", b"k", b"2", b"1"]), "*0\r\n");
        assert_eq!(
            f.run(&[b"LRANGE", b"k", b"-100", b"100"]),
            bulks(&["a", "b", "c"])
        );
        // A key that is not there is an empty range and not a nil, which is the
        // one place a list disagrees with a set.
        assert_eq!(f.run(&[b"LRANGE", b"nope", b"0", b"-1"]), "*0\r\n");
        assert_eq!(
            f.run(&[b"LRANGE", b"k", b"a", b"b"]),
            "-ERR value is not an integer or out of range\r\n"
        );
    }

    #[test]
    fn an_index_reads_and_writes_from_whichever_end_is_nearer() {
        let mut f = Fixture::new();
        f.run(&[b"RPUSH", b"k", b"a", b"b", b"c"]);
        assert_eq!(f.run(&[b"LINDEX", b"k", b"0"]), "$1\r\na\r\n");
        assert_eq!(f.run(&[b"LINDEX", b"k", b"-1"]), "$1\r\nc\r\n");
        assert_eq!(f.run(&[b"LINDEX", b"k", b"99"]), "$-1\r\n");
        assert_eq!(f.run(&[b"LINDEX", b"nope", b"0"]), "$-1\r\n");
        assert_eq!(f.run(&[b"LSET", b"k", b"-1", b"z"]), "+OK\r\n");
        assert_eq!(
            f.run(&[b"LRANGE", b"k", b"0", b"-1"]),
            bulks(&["a", "b", "z"])
        );
        // Both ways of missing are errors here rather than a nil, because a
        // list is never empty and there is nothing else the reply could be.
        assert_eq!(
            f.run(&[b"LSET", b"k", b"99", b"z"]),
            "-ERR index out of range\r\n"
        );
        assert_eq!(
            f.run(&[b"LSET", b"nope", b"0", b"z"]),
            "-ERR no such key\r\n"
        );
    }

    #[test]
    fn linsert_says_three_things_with_one_signed_number() {
        let mut f = Fixture::new();
        // Zero for a key that is not there, which is not the same as minus one
        // for a pivot that is not in a list that is.
        assert_eq!(
            f.run(&[b"LINSERT", b"nope", b"BEFORE", b"a", b"x"]),
            ":0\r\n"
        );
        f.run(&[b"RPUSH", b"k", b"a", b"b"]);
        assert_eq!(f.run(&[b"LINSERT", b"k", b"before", b"a", b"X"]), ":3\r\n");
        assert_eq!(f.run(&[b"LINSERT", b"k", b"AFTER", b"b", b"Y"]), ":4\r\n");
        assert_eq!(
            f.run(&[b"LRANGE", b"k", b"0", b"-1"]),
            bulks(&["X", "a", "b", "Y"])
        );
        assert_eq!(
            f.run(&[b"LINSERT", b"k", b"BEFORE", b"zz", b"x"]),
            ":-1\r\n"
        );
        assert_eq!(
            f.run(&[b"LINSERT", b"k", b"SIDEWAYS", b"a", b"x"]),
            "-ERR syntax error\r\n"
        );
    }

    #[test]
    fn lrem_counts_in_three_directions_and_takes_the_key_when_it_empties() {
        let mut f = Fixture::new();
        f.run(&[b"RPUSH", b"k", b"a", b"b", b"a", b"c", b"a"]);
        assert_eq!(f.run(&[b"LREM", b"k", b"2", b"a"]), ":2\r\n");
        assert_eq!(
            f.run(&[b"LRANGE", b"k", b"0", b"-1"]),
            bulks(&["b", "c", "a"])
        );
        assert_eq!(f.run(&[b"LREM", b"k", b"-1", b"a"]), ":1\r\n");
        assert_eq!(f.run(&[b"LRANGE", b"k", b"0", b"-1"]), bulks(&["b", "c"]));
        assert_eq!(f.run(&[b"LREM", b"k", b"0", b"b"]), ":1\r\n");
        assert_eq!(f.run(&[b"LREM", b"k", b"0", b"c"]), ":1\r\n");
        assert_eq!(f.run(&[b"EXISTS", b"k"]), ":0\r\n");
        assert_eq!(f.run(&[b"LREM", b"nope", b"0", b"a"]), ":0\r\n");
    }

    #[test]
    fn ltrim_keeps_a_window_and_an_empty_one_deletes_the_key() {
        let mut f = Fixture::new();
        f.run(&[b"RPUSH", b"k", b"a", b"b", b"c", b"d"]);
        assert_eq!(f.run(&[b"LTRIM", b"k", b"1", b"-2"]), "+OK\r\n");
        assert_eq!(f.run(&[b"LRANGE", b"k", b"0", b"-1"]), bulks(&["b", "c"]));
        // `LTRIM k 1 0` is the documented way to empty a list, so it has to
        // leave `EXISTS` answering zero rather than leaving an empty one.
        assert_eq!(f.run(&[b"LTRIM", b"k", b"1", b"0"]), "+OK\r\n");
        assert_eq!(f.run(&[b"EXISTS", b"k"]), ":0\r\n");
        assert_eq!(f.run(&[b"LTRIM", b"nope", b"0", b"-1"]), "+OK\r\n");
    }

    #[test]
    fn lpos_walks_from_either_end_and_stops_where_it_is_told() {
        let mut f = Fixture::new();
        f.run(&[b"RPUSH", b"p", b"a", b"b", b"c", b"a", b"b", b"c", b"a"]);
        assert_eq!(f.run(&[b"LPOS", b"p", b"a"]), ":0\r\n");
        assert_eq!(f.run(&[b"LPOS", b"p", b"a", b"RANK", b"-1"]), ":6\r\n");
        assert_eq!(f.run(&[b"LPOS", b"p", b"a", b"RANK", b"2"]), ":3\r\n");
        assert_eq!(
            f.run(&[b"LPOS", b"p", b"a", b"COUNT", b"2"]),
            "*2\r\n:0\r\n:3\r\n"
        );
        assert_eq!(
            f.run(&[b"LPOS", b"p", b"a", b"RANK", b"-1", b"COUNT", b"0"]),
            "*3\r\n:6\r\n:3\r\n:0\r\n"
        );
        // MAXLEN counts elements looked at and not matches found, so three
        // stops after `a b c` and finds the one match in it.
        assert_eq!(
            f.run(&[b"LPOS", b"p", b"a", b"COUNT", b"0", b"MAXLEN", b"3"]),
            "*1\r\n:0\r\n"
        );
        // Nothing found is three different replies depending on how it was
        // asked and whether the key is there at all.
        assert_eq!(f.run(&[b"LPOS", b"p", b"zz"]), "$-1\r\n");
        assert_eq!(f.run(&[b"LPOS", b"p", b"zz", b"COUNT", b"0"]), "*0\r\n");
        assert_eq!(f.run(&[b"LPOS", b"nope", b"a"]), "$-1\r\n");
        assert_eq!(f.run(&[b"LPOS", b"nope", b"a", b"COUNT", b"2"]), "*0\r\n");
    }

    #[test]
    fn lpos_words_its_three_mistakes_the_way_redis_does() {
        let mut f = Fixture::new();
        f.run(&[b"RPUSH", b"p", b"a"]);
        // The whole sentence and not a prefix, because the older wording of it
        // is still all over the internet and clients match on the text.
        assert_eq!(
            f.run(&[b"LPOS", b"p", b"a", b"RANK", b"0"]),
            "-ERR RANK can't be zero: use 1 to start from the first match, 2 from the second ... or use negative to start from the end of the list\r\n"
        );
        assert_eq!(
            f.run(&[b"LPOS", b"p", b"a", b"COUNT", b"-1"]),
            "-ERR COUNT can't be negative\r\n"
        );
        assert_eq!(
            f.run(&[b"LPOS", b"p", b"a", b"MAXLEN", b"-1"]),
            "-ERR MAXLEN can't be negative\r\n"
        );
        assert_eq!(
            f.run(&[b"LPOS", b"p", b"a", b"RANK"]),
            "-ERR syntax error\r\n"
        );
        assert_eq!(
            f.run(&[b"LPOS", b"p", b"a", b"FOO", b"1"]),
            "-ERR syntax error\r\n"
        );
    }

    #[test]
    fn a_move_takes_from_one_end_and_gives_to_another_even_on_one_key() {
        let mut f = Fixture::new();
        f.run(&[b"RPUSH", b"k", b"a", b"b", b"c"]);
        assert_eq!(f.run(&[b"RPOPLPUSH", b"k", b"d"]), "$1\r\nc\r\n");
        assert_eq!(f.run(&[b"LRANGE", b"k", b"0", b"-1"]), bulks(&["a", "b"]));
        assert_eq!(f.run(&[b"LRANGE", b"d", b"0", b"-1"]), bulks(&["c"]));
        assert_eq!(
            f.run(&[b"LMOVE", b"k", b"d", b"LEFT", b"RIGHT"]),
            "$1\r\na\r\n"
        );
        assert_eq!(f.run(&[b"LRANGE", b"d", b"0", b"-1"]), bulks(&["c", "a"]));
        // The same key twice is the documented way to rotate a list and falls
        // out of taking the element before deciding where to put it.
        f.run(&[b"DEL", b"r"]);
        f.run(&[b"RPUSH", b"r", b"1", b"2", b"3"]);
        assert_eq!(f.run(&[b"RPOPLPUSH", b"r", b"r"]), "$1\r\n3\r\n");
        assert_eq!(
            f.run(&[b"LRANGE", b"r", b"0", b"-1"]),
            bulks(&["3", "1", "2"])
        );
        assert_eq!(
            f.run(&[b"LMOVE", b"nope", b"d", b"LEFT", b"LEFT"]),
            "$-1\r\n"
        );
        assert_eq!(
            f.run(&[b"LMOVE", b"r", b"d", b"LEFT", b"SIDEWAYS"]),
            "-ERR syntax error\r\n"
        );
    }

    #[test]
    fn a_move_checks_the_destination_before_it_takes_anything() {
        let mut f = Fixture::new();
        f.run(&[b"RPUSH", b"k", b"a", b"b"]);
        f.run(&[b"SET", b"str", b"v"]);
        assert_eq!(
            f.run(&[b"LMOVE", b"k", b"str", b"LEFT", b"LEFT"]),
            "-WRONGTYPE Operation against a key holding the wrong kind of value\r\n"
        );
        // The element is still where it was, rather than having gone nowhere.
        assert_eq!(f.run(&[b"LRANGE", b"k", b"0", b"-1"]), bulks(&["a", "b"]));
    }

    #[test]
    fn lmpop_answers_from_the_first_key_that_has_anything() {
        let mut f = Fixture::new();
        f.run(&[b"RPUSH", b"b", b"1", b"2", b"3"]);
        // The name of the key that answered comes back with the elements,
        // because the client cannot work out which one it was.
        assert_eq!(
            f.run(&[b"LMPOP", b"2", b"a", b"b", b"LEFT", b"COUNT", b"2"]),
            "*2\r\n$1\r\nb\r\n*2\r\n$1\r\n1\r\n$1\r\n2\r\n"
        );
        assert_eq!(
            f.run(&[b"LMPOP", b"2", b"a", b"b", b"RIGHT"]),
            "*2\r\n$1\r\nb\r\n*1\r\n$1\r\n3\r\n"
        );
        assert_eq!(f.run(&[b"EXISTS", b"b"]), ":0\r\n");
        // A null array and not a null, even though what it stands in for is an
        // array holding a key name and then another array.
        assert_eq!(f.run(&[b"LMPOP", b"2", b"a", b"b", b"LEFT"]), "*-1\r\n");
    }

    #[test]
    fn lmpop_has_its_own_words_for_a_count_and_for_a_key_count() {
        let mut f = Fixture::new();
        f.run(&[b"RPUSH", b"k", b"a"]);
        assert_eq!(
            f.run(&[b"LMPOP", b"0", b"k", b"LEFT"]),
            "-ERR numkeys should be greater than 0\r\n"
        );
        assert_eq!(
            f.run(&[b"LMPOP", b"-1", b"k", b"LEFT"]),
            "-ERR numkeys should be greater than 0\r\n"
        );
        assert_eq!(
            f.run(&[b"LMPOP", b"1", b"k", b"LEFT", b"COUNT", b"0"]),
            "-ERR count should be greater than 0\r\n"
        );
        // A key count that eats the direction is a syntax error and not a
        // sentence about key counts, because the direction is simply not there.
        assert_eq!(
            f.run(&[b"LMPOP", b"3", b"k", b"LEFT"]),
            "-ERR syntax error\r\n"
        );
        assert_eq!(
            f.run(&[b"LMPOP", b"1", b"k", b"LEFT", b"COUNT", b"1", b"x"]),
            "-ERR syntax error\r\n"
        );
        assert_eq!(
            f.run(&[b"LMPOP", b"1", b"k", b"LEFT", b"FOO", b"1"]),
            "-ERR syntax error\r\n"
        );
        assert_eq!(
            f.run(&[b"LMPOP", b"1", b"k", b"SIDEWAYS"]),
            "-ERR syntax error\r\n"
        );
        assert_eq!(f.run(&[b"LLEN", b"k"]), ":1\r\n");
    }

    #[test]
    fn every_list_command_says_wrongtype_and_writes_nothing() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"str", b"v"]);
        let wrong = "-WRONGTYPE Operation against a key holding the wrong kind of value\r\n";
        for cmd in [
            &[b"LPUSH".as_slice(), b"str", b"a"][..],
            &[b"RPUSH", b"str", b"a"],
            &[b"LPUSHX", b"str", b"a"],
            &[b"RPUSHX", b"str", b"a"],
            &[b"LPOP", b"str"],
            &[b"LPOP", b"str", b"2"],
            &[b"RPOP", b"str"],
            &[b"LLEN", b"str"],
            &[b"LRANGE", b"str", b"0", b"-1"],
            &[b"LINDEX", b"str", b"0"],
            &[b"LSET", b"str", b"0", b"a"],
            &[b"LINSERT", b"str", b"BEFORE", b"a", b"b"],
            &[b"LREM", b"str", b"0", b"a"],
            &[b"LTRIM", b"str", b"0", b"-1"],
            &[b"LPOS", b"str", b"a"],
            &[b"LPOS", b"str", b"a", b"COUNT", b"0"],
            &[b"RPOPLPUSH", b"str", b"d"],
            &[b"LMOVE", b"str", b"d", b"LEFT", b"LEFT"],
            &[b"LMPOP", b"1", b"str", b"LEFT"],
        ] {
            assert_eq!(f.run(cmd), wrong, "{:?}", String::from_utf8_lossy(cmd[0]));
        }
        assert_eq!(f.run(&[b"GET", b"str"]), "$1\r\nv\r\n");
        assert_eq!(f.run(&[b"EXISTS", b"d"]), ":0\r\n");
    }

    /// A timeout is not an integer and it is not an ordinary float either: the
    /// three sentences it can answer with are its own, and which one a given
    /// argument gets is not what reading the code would suggest.
    #[test]
    fn a_timeout_has_three_ways_of_being_wrong() {
        let mut f = Fixture::new();
        let not_float = "-ERR timeout is not a float or out of range\r\n";
        let range = "-ERR timeout is out of range\r\n";
        for (bad, want) in [
            (&[b"BLPOP".as_slice(), b"k", b"abc"][..], not_float),
            (&[b"BLPOP", b"k", b"nan"], not_float),
            (&[b"BLPOP", b"k", b""], not_float),
            // Whitespace on either side, which `strtold` would take and Redis
            // does not.
            (&[b"BLPOP", b"k", b" 1"], not_float),
            (&[b"BLPOP", b"k", b"1 "], not_float),
            (&[b"BLPOP", b"k", b"-1"], "-ERR timeout is negative\r\n"),
            (&[b"BLPOP", b"k", b"-0.1"], "-ERR timeout is negative\r\n"),
            // These three parse, so they are not the not-a-float error, and all
            // three are further off than an i64 of milliseconds reaches.
            (&[b"BLPOP", b"k", b"1e400"], range),
            (&[b"BLPOP", b"k", b"inf"], range),
            (&[b"BLPOP", b"k", b"9999999999999999"], range),
            (&[b"BRPOP", b"k", b"abc"], not_float),
            (
                &[b"BLMOVE", b"a", b"b", b"LEFT", b"RIGHT", b"abc"],
                not_float,
            ),
            (
                &[b"BRPOPLPUSH", b"a", b"b", b"-1"],
                "-ERR timeout is negative\r\n",
            ),
            (&[b"BLMPOP", b"abc", b"1", b"k", b"LEFT"], not_float),
        ] {
            assert_eq!(f.run(bad), want, "for {bad:?}");
        }
    }

    /// A timeout of exactly zero means no timeout, and there are two ways of
    /// writing exactly zero.
    #[test]
    fn a_zero_timeout_waits_and_the_smallest_positive_one_does_not() {
        let mut f = Fixture::new();
        for timeout in [b"0".as_slice(), b"0.0", b"-0.0"] {
            let (flow, out) = f.flow(&[b"BLPOP", b"k", timeout]);
            assert_eq!(flow, Flow::Block, "for {timeout:?}");
            assert!(out.is_empty(), "for {timeout:?}");
        }
        // Positive, so it is a real deadline, and the deadline is this
        // millisecond. Nothing is written here either: the reply comes from the
        // sweep, which is the engine's and not this layer's.
        let (flow, out) = f.flow(&[b"BLPOP", b"k", b"0.0000001"]);
        assert_eq!(flow, Flow::Block);
        assert!(out.is_empty());
    }

    #[test]
    fn a_blocking_command_that_can_be_answered_answers_like_the_one_it_wraps() {
        let mut f = Fixture::new();
        f.run(&[b"RPUSH", b"L", b"a", b"b", b"c", b"d", b"e"]);

        // The one difference from LPOP: the reply names the key that answered,
        // which is what makes BLPOP over several keys usable.
        assert_eq!(
            f.flow(&[b"BLPOP", b"nope", b"L", b"0"]),
            (Flow::Continue, "*2\r\n$1\r\nL\r\n$1\r\na\r\n".to_owned())
        );
        assert_eq!(
            f.run(&[b"BRPOP", b"L", b"0"]),
            "*2\r\n$1\r\nL\r\n$1\r\ne\r\n"
        );
        assert_eq!(
            f.run(&[
                b"BLMPOP", b"0", b"2", b"nope", b"L", b"LEFT", b"COUNT", b"2"
            ]),
            "*2\r\n$1\r\nL\r\n*2\r\n$1\r\nb\r\n$1\r\nc\r\n"
        );
        assert_eq!(
            f.run(&[b"BLMOVE", b"L", b"D", b"LEFT", b"RIGHT", b"0"]),
            "$1\r\nd\r\n"
        );
        assert_eq!(
            f.run(&[b"EXISTS", b"L"]),
            ":0\r\n",
            "and the key went with it"
        );
        assert_eq!(f.run(&[b"LRANGE", b"D", b"0", b"-1"]), "*1\r\n$1\r\nd\r\n");
        // Onto itself, which is how a list is rotated and is a real thing to ask
        // a blocking move for.
        f.run(&[b"RPUSH", b"D", b"x"]);
        assert_eq!(f.run(&[b"BRPOPLPUSH", b"D", b"D", b"0"]), "$1\r\nx\r\n");
        assert_eq!(
            f.run(&[b"LRANGE", b"D", b"0", b"-1"]),
            "*2\r\n$1\r\nx\r\n$1\r\nd\r\n"
        );
    }

    #[test]
    fn blmpop_reads_its_count_and_its_key_count_the_way_lmpop_does() {
        let mut f = Fixture::new();
        f.run(&[b"RPUSH", b"k", b"a"]);
        for (bad, want) in [
            (
                &[b"BLMPOP".as_slice(), b"0", b"0", b"k", b"LEFT"][..],
                "-ERR numkeys should be greater than 0\r\n",
            ),
            (
                &[b"BLMPOP", b"0", b"-1", b"k", b"LEFT"],
                "-ERR numkeys should be greater than 0\r\n",
            ),
            // Two keys named and one given, so the word that should have been
            // the direction is a key and there is no direction left.
            (
                &[b"BLMPOP", b"0", b"2", b"k", b"LEFT"],
                "-ERR syntax error\r\n",
            ),
            (
                &[b"BLMPOP", b"0", b"1", b"k", b"SIDEWAYS"],
                "-ERR syntax error\r\n",
            ),
            (
                &[b"BLMPOP", b"0", b"1", b"k", b"LEFT", b"COUNT"],
                "-ERR syntax error\r\n",
            ),
            (
                &[b"BLMPOP", b"0", b"1", b"k", b"LEFT", b"COUNT", b"2", b"x"],
                "-ERR syntax error\r\n",
            ),
            // A count that is not a number at all gets the same sentence a zero
            // or a negative one gets, rather than the usual one about integers.
            (
                &[b"BLMPOP", b"0", b"1", b"k", b"LEFT", b"COUNT", b"0"],
                "-ERR count should be greater than 0\r\n",
            ),
            (
                &[b"BLMPOP", b"0", b"1", b"k", b"LEFT", b"COUNT", b"abc"],
                "-ERR count should be greater than 0\r\n",
            ),
        ] {
            assert_eq!(f.run(bad), want, "for {bad:?}");
        }
        assert_eq!(f.run(&[b"LLEN", b"k"]), ":1\r\n", "and none of them popped");
    }

    #[test]
    fn a_blocking_move_reads_its_directions_before_its_timeout() {
        let mut f = Fixture::new();
        // Both are wrong. Redis checks the directions first, so this is the
        // syntax error and not a complaint about the timeout.
        assert_eq!(
            f.run(&[b"BLMOVE", b"a", b"b", b"UP", b"DOWN", b"abc"]),
            "-ERR syntax error\r\n"
        );
        assert_eq!(
            f.run(&[b"BLMOVE", b"a", b"b", b"LEFT", b"DOWN", b"0.05"]),
            "-ERR syntax error\r\n"
        );
    }

    /// The four ways a blocking command sees a key of another type, and the one
    /// way it does not.
    #[test]
    fn a_blocking_command_errors_on_a_wrong_type_rather_than_waiting_on_it() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"S", b"v"]);
        f.run(&[b"RPUSH", b"D", b"x"]);
        let wrong = "-WRONGTYPE Operation against a key holding the wrong kind of value\r\n";

        assert_eq!(f.run(&[b"BLPOP", b"S", b"0"]), wrong);
        // Every key is checked even when an earlier one would have blocked, so
        // an empty key in front of a string does not hide it.
        assert_eq!(f.run(&[b"BLPOP", b"E", b"S", b"0"]), wrong);
        assert_eq!(f.run(&[b"BRPOP", b"S", b"0"]), wrong);
        assert_eq!(f.run(&[b"BLMPOP", b"0", b"1", b"S", b"LEFT"]), wrong);
        assert_eq!(f.run(&[b"BRPOPLPUSH", b"S", b"D", b"0"]), wrong);
        // The destination, which is only reached because the source has
        // something in it.
        assert_eq!(f.run(&[b"BRPOPLPUSH", b"D", b"S", b"0"]), wrong);
        assert_eq!(f.run(&[b"LRANGE", b"D", b"0", b"-1"]), "*1\r\n$1\r\nx\r\n");

        // And the one that does not: an empty source means the destination is
        // never looked at, so this waits rather than erroring, and on a real
        // server it times out.
        assert_eq!(
            f.flow(&[b"BLMOVE", b"E", b"S", b"LEFT", b"RIGHT", b"0.1"])
                .0,
            Flow::Block
        );
    }

    /// The same churn the set and the string get, because a list that leaks a
    /// chunk per push looks exactly like one that does not until it has run for
    /// an afternoon.
    #[test]
    fn churning_lists_does_not_grow_the_server() {
        let mut f = Fixture::new();
        let vals: Vec<Vec<u8>> = (0..200).map(|i| format!("v{i}").into_bytes()).collect();
        let args: Vec<&[u8]> = [&b"RPUSH"[..], &b"k"[..]]
            .into_iter()
            .chain(vals.iter().map(Vec::as_slice))
            .collect();

        f.run(&args);
        f.run(&[b"DEL", b"k"]);
        f.server.compact_step();
        let after_first = f.server.memory_bytes();

        for _ in 0..200 {
            f.run(&args);
            f.run(&[b"LTRIM", b"k", b"1", b"0"]);
            f.server.compact_step();
        }
        assert_eq!(f.run(&[b"DBSIZE"]), ":0\r\n");
        assert!(
            f.server.memory_bytes() <= after_first * 2,
            "held {} after two hundred passes against {after_first} after one",
            f.server.memory_bytes()
        );
    }

    // ------------------------------------------------------------ sorted set

    #[test]
    fn a_sorted_set_takes_scores_and_gives_them_back() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"ZADD", b"z", b"1", b"a", b"2", b"b"]), ":2\r\n");
        assert_eq!(f.run(&[b"ZADD", b"z", b"1", b"a", b"3", b"c"]), ":1\r\n");
        assert_eq!(f.run(&[b"ZCARD", b"z"]), ":3\r\n");
        assert_eq!(f.run(&[b"ZSCORE", b"z", b"b"]), "$1\r\n2\r\n");
        assert_eq!(f.run(&[b"ZSCORE", b"z", b"nope"]), "$-1\r\n");
        assert_eq!(f.run(&[b"ZSCORE", b"nokey", b"b"]), "$-1\r\n");
        assert_eq!(
            f.run(&[b"ZMSCORE", b"z", b"a", b"nope", b"c"]),
            "*3\r\n$1\r\n1\r\n$-1\r\n$1\r\n3\r\n"
        );
        assert_eq!(f.run(&[b"ZREM", b"z", b"a", b"nope"]), ":1\r\n");
        assert_eq!(f.run(&[b"ZCARD", b"z"]), ":2\r\n");
        // The key goes when the last member does.
        assert_eq!(f.run(&[b"ZREM", b"z", b"b", b"c"]), ":2\r\n");
        assert_eq!(f.run(&[b"EXISTS", b"z"]), ":0\r\n");
    }

    #[test]
    fn a_score_is_a_double_on_resp3_and_digits_on_resp2() {
        let mut f = Fixture::new();
        f.run(&[b"ZADD", b"z", b"1.5", b"a", b"inf", b"b", b"-inf", b"c"]);
        assert_eq!(f.run(&[b"ZSCORE", b"z", b"a"]), "$3\r\n1.5\r\n");
        assert_eq!(f.run(&[b"ZSCORE", b"z", b"b"]), "$3\r\ninf\r\n");
        assert_eq!(f.run(&[b"ZSCORE", b"z", b"c"]), "$4\r\n-inf\r\n");

        f.out = Out::new(Proto::Resp3);
        assert_eq!(f.run(&[b"ZSCORE", b"z", b"a"]), ",1.5\r\n");
        assert_eq!(f.run(&[b"ZSCORE", b"z", b"b"]), ",inf\r\n");
        assert_eq!(f.run(&[b"ZSCORE", b"z", b"c"]), ",-inf\r\n");
        assert_eq!(f.run(&[b"ZSCORE", b"z", b"nope"]), "_\r\n");
    }

    #[test]
    fn the_zadd_options_gate_what_gets_written() {
        let mut f = Fixture::new();
        f.run(&[b"ZADD", b"z", b"5", b"a"]);
        // NX leaves a member that is there alone, XX will not create one.
        assert_eq!(f.run(&[b"ZADD", b"z", b"NX", b"9", b"a"]), ":0\r\n");
        assert_eq!(f.run(&[b"ZSCORE", b"z", b"a"]), "$1\r\n5\r\n");
        assert_eq!(f.run(&[b"ZADD", b"z", b"XX", b"9", b"new"]), ":0\r\n");
        assert_eq!(f.run(&[b"EXISTS", b"z"]), ":1\r\n");
        // GT and LT only move a score one way.
        assert_eq!(f.run(&[b"ZADD", b"z", b"GT", b"CH", b"3", b"a"]), ":0\r\n");
        assert_eq!(f.run(&[b"ZADD", b"z", b"GT", b"CH", b"7", b"a"]), ":1\r\n");
        assert_eq!(f.run(&[b"ZADD", b"z", b"LT", b"CH", b"9", b"a"]), ":0\r\n");
        // CH counts a moved score and plain ZADD does not.
        assert_eq!(f.run(&[b"ZADD", b"z", b"1", b"a", b"1", b"b"]), ":1\r\n");
        assert_eq!(
            f.run(&[b"ZADD", b"z", b"CH", b"2", b"a", b"2", b"c"]),
            ":2\r\n"
        );
    }

    #[test]
    fn zadd_incr_answers_a_score_or_nothing_at_all() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"ZADD", b"z", b"INCR", b"5", b"m"]), "$1\r\n5\r\n");
        assert_eq!(f.run(&[b"ZADD", b"z", b"INCR", b"2", b"m"]), "$1\r\n7\r\n");
        // A gate that refuses is the string nil, because the reply it stands in
        // for is a score.
        assert_eq!(
            f.run(&[b"ZADD", b"z", b"NX", b"INCR", b"2", b"m"]),
            "$-1\r\n"
        );
        assert_eq!(
            f.run(&[b"ZADD", b"z", b"XX", b"INCR", b"2", b"gone"]),
            "$-1\r\n"
        );
        assert_eq!(
            f.run(&[b"ZADD", b"z", b"GT", b"INCR", b"-1", b"m"]),
            "$-1\r\n"
        );
        assert_eq!(
            f.run(&[b"ZADD", b"z", b"GT", b"INCR", b"1", b"m"]),
            "$1\r\n8\r\n"
        );
        assert_eq!(f.run(&[b"ZINCRBY", b"z", b"2", b"m"]), "$2\r\n10\r\n");
        assert_eq!(f.run(&[b"ZINCRBY", b"z", b"1", b"fresh"]), "$1\r\n1\r\n");
    }

    #[test]
    fn the_two_infinities_will_not_be_added_together() {
        let mut f = Fixture::new();
        f.run(&[b"ZADD", b"z", b"inf", b"m"]);
        let nan = "-ERR resulting score is not a number (NaN)\r\n";
        assert_eq!(f.run(&[b"ZINCRBY", b"z", b"-inf", b"m"]), nan);
        assert_eq!(f.run(&[b"ZADD", b"z", b"INCR", b"-inf", b"m"]), nan);
        assert_eq!(f.run(&[b"ZSCORE", b"z", b"m"]), "$3\r\ninf\r\n");
        // And a key made for an increment that then fails does not stay behind.
        assert_eq!(f.run(&[b"ZINCRBY", b"gone", b"1", b"m"]), "$1\r\n1\r\n");
    }

    #[test]
    fn zadd_says_its_mistakes_the_way_redis_says_them() {
        let mut f = Fixture::new();
        // The pairs are counted before the options are looked at, so this is a
        // syntax error about having none and not a complaint about NX and XX.
        assert_eq!(
            f.run(&[b"ZADD", b"z", b"NX", b"XX"]),
            "-ERR syntax error\r\n"
        );
        assert_eq!(
            f.run(&[b"ZADD", b"z", b"NX", b"XX", b"1", b"a"]),
            "-ERR XX and NX options at the same time are not compatible\r\n"
        );
        let gtlt = "-ERR GT, LT, and/or NX options at the same time are not compatible\r\n";
        assert_eq!(f.run(&[b"ZADD", b"z", b"NX", b"GT", b"1", b"a"]), gtlt);
        assert_eq!(f.run(&[b"ZADD", b"z", b"GT", b"LT", b"1", b"a"]), gtlt);
        assert_eq!(
            f.run(&[b"ZADD", b"z", b"INCR", b"1", b"a", b"2", b"b"]),
            "-ERR INCR option supports a single increment-element pair\r\n"
        );
        // An odd number of arguments after the options.
        assert_eq!(
            f.run(&[b"ZADD", b"z", b"1", b"a", b"2"]),
            "-ERR syntax error\r\n"
        );
        // Every score is read before the first is stored.
        assert_eq!(
            f.run(&[b"ZADD", b"z", b"1", b"a", b"nonsense", b"b"]),
            "-ERR value is not a valid float\r\n"
        );
        assert_eq!(f.run(&[b"EXISTS", b"z"]), ":0\r\n");
    }

    #[test]
    fn a_rank_says_where_a_member_sits_from_either_end() {
        let mut f = Fixture::new();
        f.run(&[b"ZADD", b"z", b"1", b"a", b"2", b"b", b"3", b"c"]);
        assert_eq!(f.run(&[b"ZRANK", b"z", b"a"]), ":0\r\n");
        assert_eq!(f.run(&[b"ZRANK", b"z", b"c"]), ":2\r\n");
        assert_eq!(f.run(&[b"ZREVRANK", b"z", b"c"]), ":0\r\n");
        assert_eq!(f.run(&[b"ZREVRANK", b"z", b"a"]), ":2\r\n");
        // WITHSCORE changes both shapes: the answer and the nothing.
        assert_eq!(
            f.run(&[b"ZRANK", b"z", b"b", b"WITHSCORE"]),
            "*2\r\n:1\r\n$1\r\n2\r\n"
        );
        assert_eq!(f.run(&[b"ZRANK", b"z", b"nope"]), "$-1\r\n");
        assert_eq!(f.run(&[b"ZRANK", b"z", b"nope", b"WITHSCORE"]), "*-1\r\n");
        assert_eq!(f.run(&[b"ZRANK", b"nokey", b"a", b"WITHSCORE"]), "*-1\r\n");
        // A bad option is a syntax error and one argument too many is an arity
        // error, which is Redis's split.
        assert_eq!(
            f.run(&[b"ZRANK", b"z", b"b", b"bogus"]),
            "-ERR syntax error\r\n"
        );
        assert_eq!(
            f.run(&[b"ZREVRANK", b"z", b"b", b"WITHSCORE", b"more"]),
            "-ERR wrong number of arguments for 'zrevrank' command\r\n"
        );
    }

    #[test]
    fn the_two_counts_read_their_two_kinds_of_bound() {
        let mut f = Fixture::new();
        f.run(&[b"ZADD", b"z", b"1", b"a", b"2", b"b", b"3", b"c"]);
        assert_eq!(f.run(&[b"ZCOUNT", b"z", b"-inf", b"+inf"]), ":3\r\n");
        assert_eq!(f.run(&[b"ZCOUNT", b"z", b"2", b"3"]), ":2\r\n");
        assert_eq!(f.run(&[b"ZCOUNT", b"z", b"(1", b"3"]), ":2\r\n");
        assert_eq!(f.run(&[b"ZCOUNT", b"z", b"(1", b"(3"]), ":1\r\n");
        assert_eq!(f.run(&[b"ZCOUNT", b"nokey", b"-inf", b"+inf"]), ":0\r\n");
        assert_eq!(
            f.run(&[b"ZCOUNT", b"z", b"bogus", b"3"]),
            "-ERR min or max is not a float\r\n"
        );

        f.run(&[b"ZADD", b"l", b"0", b"a", b"0", b"b", b"0", b"c"]);
        assert_eq!(f.run(&[b"ZLEXCOUNT", b"l", b"-", b"+"]), ":3\r\n");
        assert_eq!(f.run(&[b"ZLEXCOUNT", b"l", b"[a", b"(c"]), ":2\r\n");
        assert_eq!(f.run(&[b"ZLEXCOUNT", b"l", b"(a", b"+"]), ":2\r\n");
        // A bare member is not a bound, because a member can start with any
        // byte and there would be no way to say the bracket if it were optional.
        assert_eq!(
            f.run(&[b"ZLEXCOUNT", b"l", b"a", b"c"]),
            "-ERR min or max not valid string range item\r\n"
        );
    }

    /// The three ways `ZRANGE` can be asked for a window, forwards and back.
    ///
    /// Every byte in here was read off a real 8.10.1 rather than worked out,
    /// because the interesting part of this command is not what it selects, it
    /// is which of the two ends the client is expected to name first.
    #[test]
    fn one_range_command_selects_by_rank_or_score_or_name() {
        let mut f = Fixture::new();
        f.run(&[b"ZADD", b"z", b"1", b"a", b"2", b"b", b"3", b"c"]);
        assert_eq!(
            f.run(&[b"ZRANGE", b"z", b"0", b"-1"]),
            "*3\r\n$1\r\na\r\n$1\r\nb\r\n$1\r\nc\r\n"
        );
        assert_eq!(
            f.run(&[b"ZRANGE", b"z", b"-2", b"-1"]),
            "*2\r\n$1\r\nb\r\n$1\r\nc\r\n"
        );
        assert_eq!(f.run(&[b"ZRANGE", b"z", b"5", b"9"]), "*0\r\n");
        assert_eq!(f.run(&[b"ZRANGE", b"nokey", b"0", b"-1"]), "*0\r\n");
        // REV over ranks reverses the walk and leaves the two arguments alone,
        // because a rank counts from the end the walk starts at.
        assert_eq!(
            f.run(&[b"ZRANGE", b"z", b"0", b"-1", b"REV"]),
            "*3\r\n$1\r\nc\r\n$1\r\nb\r\n$1\r\na\r\n"
        );
        assert_eq!(
            f.run(&[b"ZRANGE", b"z", b"(1", b"+inf", b"BYSCORE"]),
            "*2\r\n$1\r\nb\r\n$1\r\nc\r\n"
        );
        // And REV over scores does swap them, since a bound does not count from
        // anywhere. This is the one line of the parse that tells the two apart.
        assert_eq!(
            f.run(&[b"ZRANGE", b"z", b"+inf", b"(1", b"BYSCORE", b"REV"]),
            "*2\r\n$1\r\nc\r\n$1\r\nb\r\n"
        );
        assert_eq!(
            f.run(&[b"ZRANGE", b"z", b"-", b"+", b"BYLEX"]),
            "*3\r\n$1\r\na\r\n$1\r\nb\r\n$1\r\nc\r\n"
        );
        assert_eq!(
            f.run(&[b"ZRANGE", b"z", b"+", b"-", b"BYLEX", b"REV"]),
            "*3\r\n$1\r\nc\r\n$1\r\nb\r\n$1\r\na\r\n"
        );
    }

    /// The older spellings, which are the same six windows with the mode in the
    /// name and the high end named first on the three that go backwards.
    #[test]
    fn the_older_range_spellings_name_their_high_end_first() {
        let mut f = Fixture::new();
        f.run(&[b"ZADD", b"z", b"1", b"a", b"2", b"b", b"3", b"c"]);
        assert_eq!(
            f.run(&[b"ZREVRANGE", b"z", b"0", b"-1"]),
            "*3\r\n$1\r\nc\r\n$1\r\nb\r\n$1\r\na\r\n"
        );
        assert_eq!(
            f.run(&[b"ZREVRANGE", b"z", b"0", b"0", b"WITHSCORES"]),
            "*2\r\n$1\r\nc\r\n$1\r\n3\r\n"
        );
        assert_eq!(
            f.run(&[b"ZRANGEBYSCORE", b"z", b"(1", b"3"]),
            "*2\r\n$1\r\nb\r\n$1\r\nc\r\n"
        );
        assert_eq!(
            f.run(&[b"ZREVRANGEBYSCORE", b"z", b"3", b"(1"]),
            "*2\r\n$1\r\nc\r\n$1\r\nb\r\n"
        );
        // The two arguments the wrong way round is an empty answer and not an
        // error, which is what the swap being in the parse rather than in the
        // window buys.
        assert_eq!(f.run(&[b"ZREVRANGEBYSCORE", b"z", b"(1", b"3"]), "*0\r\n");
        assert_eq!(
            f.run(&[b"ZRANGEBYLEX", b"z", b"[a", b"(c"]),
            "*2\r\n$1\r\na\r\n$1\r\nb\r\n"
        );
        assert_eq!(
            f.run(&[b"ZREVRANGEBYLEX", b"z", b"(c", b"[a"]),
            "*2\r\n$1\r\nb\r\n$1\r\na\r\n"
        );
        // BYSCORE, BYLEX and REV mean nothing to these, so they are not another
        // way of spelling the mode, they are a syntax error.
        for cmd in [
            &[b"ZREVRANGE".as_slice(), b"z", b"0", b"-1", b"BYSCORE"][..],
            &[b"ZRANGEBYSCORE", b"z", b"1", b"3", b"REV"],
            &[b"ZRANGEBYLEX", b"z", b"[a", b"[c", b"BYLEX"],
        ] {
            assert_eq!(f.run(cmd), "-ERR syntax error\r\n", "{:?}", cmd[0]);
        }
    }

    /// `LIMIT` and `WITHSCORES`, which every one of these commands reads and
    /// only some of them accept.
    #[test]
    fn limit_and_withscores_are_read_by_all_of_them_and_refused_afterwards() {
        let mut f = Fixture::new();
        f.run(&[b"ZADD", b"z", b"1", b"a", b"2", b"b", b"3", b"c"]);
        assert_eq!(
            f.run(&[
                b"ZRANGE", b"z", b"-inf", b"+inf", b"BYSCORE", b"LIMIT", b"1", b"1"
            ]),
            "*1\r\n$1\r\nb\r\n"
        );
        // A negative offset skips past everything, a negative count is no bound.
        assert_eq!(
            f.run(&[
                b"ZRANGE", b"z", b"-inf", b"+inf", b"BYSCORE", b"LIMIT", b"-1", b"2"
            ]),
            "*0\r\n"
        );
        assert_eq!(
            f.run(&[
                b"ZRANGE", b"z", b"-inf", b"+inf", b"BYSCORE", b"LIMIT", b"0", b"-1"
            ]),
            "*3\r\n$1\r\na\r\n$1\r\nb\r\n$1\r\nc\r\n"
        );
        // The two options in either order, which falls out of the parse loop.
        let both = "*4\r\n$1\r\na\r\n$1\r\n1\r\n$1\r\nb\r\n$1\r\n2\r\n";
        assert_eq!(
            f.run(&[
                b"ZRANGEBYSCORE",
                b"z",
                b"1",
                b"3",
                b"WITHSCORES",
                b"LIMIT",
                b"0",
                b"2"
            ]),
            both
        );
        assert_eq!(
            f.run(&[
                b"ZRANGEBYSCORE",
                b"z",
                b"1",
                b"3",
                b"LIMIT",
                b"0",
                b"2",
                b"WITHSCORES"
            ]),
            both
        );
        // LIMIT on a range by rank is refused after the whole option list has
        // been read, so this complains about LIMIT and not about WITHSCORES.
        let needs_by = "-ERR syntax error, LIMIT is only supported in combination with either BYSCORE or BYLEX\r\n";
        assert_eq!(
            f.run(&[
                b"ZREVRANGE",
                b"z",
                b"0",
                b"-1",
                b"WITHSCORES",
                b"LIMIT",
                b"0",
                b"1"
            ]),
            needs_by
        );
        assert_eq!(
            f.run(&[b"ZRANGE", b"z", b"0", b"-1", b"LIMIT", b"0", b"1"]),
            needs_by
        );
        let not_bylex = "-ERR syntax error, WITHSCORES not supported in combination with BYLEX\r\n";
        assert_eq!(
            f.run(&[b"ZRANGE", b"z", b"-", b"+", b"BYLEX", b"WITHSCORES"]),
            not_bylex
        );
        assert_eq!(
            f.run(&[b"ZRANGEBYLEX", b"z", b"[a", b"[c", b"WITHSCORES"]),
            not_bylex
        );
        // Two modes at once, an option nobody knows, a LIMIT missing its count,
        // and the three number errors, which are three different sentences.
        for cmd in [
            &[
                b"ZRANGE".as_slice(),
                b"z",
                b"0",
                b"-1",
                b"BYSCORE",
                b"BYLEX",
            ][..],
            &[b"ZRANGE", b"z", b"0", b"-1", b"junk"],
            &[b"ZRANGEBYSCORE", b"z", b"1", b"3", b"LIMIT", b"0"],
        ] {
            assert_eq!(f.run(cmd), "-ERR syntax error\r\n", "{cmd:?}");
        }
        assert_eq!(
            f.run(&[b"ZRANGEBYSCORE", b"z", b"bad", b"3"]),
            "-ERR min or max is not a float\r\n"
        );
        assert_eq!(
            f.run(&[b"ZRANGEBYLEX", b"z", b"a", b"[c"]),
            "-ERR min or max not valid string range item\r\n"
        );
        assert_eq!(
            f.run(&[b"ZRANGEBYSCORE", b"z", b"1", b"3", b"LIMIT", b"a", b"2"]),
            "-ERR value is not an integer or out of range\r\n"
        );
    }

    /// `WITHSCORES` is the one place in this group where the two protocols
    /// disagree about the shape of the reply and not just the type of a value.
    #[test]
    fn withscores_nests_on_resp3_and_flattens_on_resp2() {
        let mut f = Fixture::new();
        f.run(&[b"ZADD", b"z", b"1", b"a", b"2", b"b", b"3", b"c"]);
        assert_eq!(
            f.run(&[b"ZRANGE", b"z", b"0", b"-1", b"WITHSCORES"]),
            "*6\r\n$1\r\na\r\n$1\r\n1\r\n$1\r\nb\r\n$1\r\n2\r\n$1\r\nc\r\n$1\r\n3\r\n"
        );
        f.out = Out::new(Proto::Resp3);
        assert_eq!(
            f.run(&[b"ZRANGE", b"z", b"0", b"-1", b"WITHSCORES"]),
            "*3\r\n*2\r\n$1\r\na\r\n,1\r\n*2\r\n$1\r\nb\r\n,2\r\n*2\r\n$1\r\nc\r\n,3\r\n"
        );
        assert_eq!(
            f.run(&[b"ZRANGE", b"z", b"0", b"-1"]),
            "*3\r\n$1\r\na\r\n$1\r\nb\r\n$1\r\nc\r\n"
        );
    }

    /// The store form, which is the same parse with the destination in front.
    #[test]
    fn a_range_store_writes_the_window_into_another_key() {
        let mut f = Fixture::new();
        f.run(&[b"ZADD", b"z", b"1", b"a", b"2", b"b", b"3", b"c"]);
        assert_eq!(f.run(&[b"ZRANGESTORE", b"d", b"z", b"0", b"-1"]), ":3\r\n");
        // A window that selects nothing deletes the destination rather than
        // leaving an empty sorted set, because an empty one does not exist.
        assert_eq!(f.run(&[b"ZRANGESTORE", b"d", b"z", b"5", b"9"]), ":0\r\n");
        assert_eq!(f.run(&[b"EXISTS", b"d"]), ":0\r\n");
        assert_eq!(
            f.run(&[b"ZRANGESTORE", b"d", b"z", b"(1", b"+inf", b"BYSCORE"]),
            ":2\r\n"
        );
        assert_eq!(
            f.run(&[b"ZRANGE", b"d", b"0", b"-1", b"WITHSCORES"]),
            "*4\r\n$1\r\nb\r\n$1\r\n2\r\n$1\r\nc\r\n$1\r\n3\r\n"
        );
        // The destination is allowed to be the source, because the result is
        // built whole before anything is written over.
        assert_eq!(f.run(&[b"ZRANGESTORE", b"z", b"z", b"1", b"2"]), ":2\r\n");
        assert_eq!(
            f.run(&[b"ZRANGE", b"z", b"0", b"-1", b"WITHSCORES"]),
            "*4\r\n$1\r\nb\r\n$1\r\n2\r\n$1\r\nc\r\n$1\r\n3\r\n"
        );
        // It takes every option ZRANGE takes except WITHSCORES, which is a
        // plain syntax error here and not the sentence about BYLEX.
        assert_eq!(
            f.run(&[b"ZRANGESTORE", b"d", b"z", b"0", b"-1", b"WITHSCORES"]),
            "-ERR syntax error\r\n"
        );
    }

    /// The three removals, which are the read side's window with the walk
    /// turned into a removal and no options at all.
    #[test]
    fn the_three_removals_share_their_window_with_the_reads() {
        let mut f = Fixture::new();
        f.run(&[b"ZADD", b"z", b"1", b"a", b"2", b"b", b"3", b"c"]);
        assert_eq!(f.run(&[b"ZREMRANGEBYRANK", b"z", b"0", b"0"]), ":1\r\n");
        assert_eq!(
            f.run(&[b"ZRANGE", b"z", b"0", b"-1"]),
            "*2\r\n$1\r\nb\r\n$1\r\nc\r\n"
        );
        assert_eq!(
            f.run(&[b"ZREMRANGEBYSCORE", b"z", b"(2", b"+inf"]),
            ":1\r\n"
        );
        assert_eq!(f.run(&[b"ZRANGE", b"z", b"0", b"-1"]), "*1\r\n$1\r\nb\r\n");
        // The last member going takes the key with it.
        assert_eq!(f.run(&[b"ZREMRANGEBYLEX", b"z", b"-", b"+"]), ":1\r\n");
        assert_eq!(f.run(&[b"EXISTS", b"z"]), ":0\r\n");
        assert_eq!(
            f.run(&[b"ZREMRANGEBYRANK", b"nokey", b"0", b"-1"]),
            ":0\r\n"
        );
        assert_eq!(
            f.run(&[b"ZREMRANGEBYRANK", b"z", b"0", b"x"]),
            "-ERR value is not an integer or out of range\r\n"
        );
    }

    /// The algebra, which is one gather and three names for it.
    #[test]
    fn the_three_algebra_commands_combine_scores_and_order_the_answer_once() {
        let mut f = Fixture::new();
        f.run(&[b"ZADD", b"z", b"1", b"a", b"2", b"b", b"3", b"c"]);
        f.run(&[b"ZADD", b"y", b"10", b"b", b"20", b"d"]);
        assert_eq!(
            f.run(&[b"ZUNION", b"2", b"z", b"y"]),
            "*4\r\n$1\r\na\r\n$1\r\nc\r\n$1\r\nb\r\n$1\r\nd\r\n"
        );
        // The scores are added where a member is in both, and the answer comes
        // out in the order those combined scores put it in.
        assert_eq!(
            f.run(&[b"ZUNION", b"2", b"z", b"y", b"WITHSCORES"]),
            "*8\r\n$1\r\na\r\n$1\r\n1\r\n$1\r\nc\r\n$1\r\n3\r\n$1\r\nb\r\n$2\r\n12\r\n$1\r\nd\r\n$2\r\n20\r\n"
        );
        assert_eq!(
            f.run(&[
                b"ZUNION",
                b"2",
                b"z",
                b"y",
                b"WEIGHTS",
                b"2",
                b"3",
                b"WITHSCORES"
            ]),
            "*8\r\n$1\r\na\r\n$1\r\n2\r\n$1\r\nc\r\n$1\r\n6\r\n$1\r\nb\r\n$2\r\n34\r\n$1\r\nd\r\n$2\r\n60\r\n"
        );
        assert_eq!(
            f.run(&[
                b"ZUNION",
                b"2",
                b"z",
                b"y",
                b"AGGREGATE",
                b"MIN",
                b"WITHSCORES"
            ]),
            "*8\r\n$1\r\na\r\n$1\r\n1\r\n$1\r\nb\r\n$1\r\n2\r\n$1\r\nc\r\n$1\r\n3\r\n$1\r\nd\r\n$2\r\n20\r\n"
        );
        assert_eq!(
            f.run(&[
                b"ZUNION",
                b"2",
                b"z",
                b"y",
                b"AGGREGATE",
                b"MAX",
                b"WITHSCORES"
            ]),
            "*8\r\n$1\r\na\r\n$1\r\n1\r\n$1\r\nc\r\n$1\r\n3\r\n$1\r\nb\r\n$2\r\n10\r\n$1\r\nd\r\n$2\r\n20\r\n"
        );
        assert_eq!(
            f.run(&[b"ZINTER", b"2", b"z", b"y", b"WITHSCORES"]),
            "*2\r\n$1\r\nb\r\n$2\r\n12\r\n"
        );
        assert_eq!(
            f.run(&[b"ZDIFF", b"2", b"z", b"y", b"WITHSCORES"]),
            "*4\r\n$1\r\na\r\n$1\r\n1\r\n$1\r\nc\r\n$1\r\n3\r\n"
        );
        assert_eq!(f.run(&[b"ZUNION", b"1", b"nokey"]), "*0\r\n");
        // A plain set is an input, and it behaves as a sorted set in which
        // every member scores one.
        f.run(&[b"SADD", b"p", b"a", b"d"]);
        assert_eq!(
            f.run(&[b"ZUNION", b"2", b"z", b"p", b"WITHSCORES"]),
            "*8\r\n$1\r\nd\r\n$1\r\n1\r\n$1\r\na\r\n$1\r\n2\r\n$1\r\nb\r\n$1\r\n2\r\n$1\r\nc\r\n$1\r\n3\r\n"
        );
        // A difference never combines two scores, so it has nothing for either
        // of the two options to do and refuses both.
        for cmd in [
            &[
                b"ZDIFF".as_slice(),
                b"2",
                b"z",
                b"y",
                b"WEIGHTS",
                b"1",
                b"1",
            ][..],
            &[b"ZDIFF", b"2", b"z", b"y", b"AGGREGATE", b"MIN"],
        ] {
            assert_eq!(f.run(cmd), "-ERR syntax error\r\n", "{cmd:?}");
        }
    }

    /// The count of keys, which is what lets a key be named `WEIGHTS`.
    #[test]
    fn the_algebra_counts_its_keys_and_says_so_when_the_count_is_wrong() {
        let mut f = Fixture::new();
        f.run(&[b"ZADD", b"z", b"1", b"a"]);
        f.run(&[b"ZADD", b"y", b"2", b"b"]);
        // Redis names the command in this one, so each spelling says its own.
        assert_eq!(
            f.run(&[b"ZUNION", b"0", b"z"]),
            "-ERR at least 1 input key is needed for 'zunion' command\r\n"
        );
        assert_eq!(
            f.run(&[b"ZUNION", b"-1", b"z"]),
            "-ERR at least 1 input key is needed for 'zunion' command\r\n"
        );
        assert_eq!(
            f.run(&[b"ZINTERCARD", b"0", b"z"]),
            "-ERR at least 1 input key is needed for 'zintercard' command\r\n"
        );
        // A count bigger than the line is a plain syntax error, which reads
        // oddly and is what Redis says.
        assert_eq!(
            f.run(&[b"ZUNION", b"3", b"z", b"y"]),
            "-ERR syntax error\r\n"
        );
        assert_eq!(
            f.run(&[b"ZUNION", b"x", b"z"]),
            "-ERR value is not an integer or out of range\r\n"
        );
        // A WEIGHTS list that is not one per key is a syntax error, and a
        // weight that is not a number gets a sentence of its own.
        assert_eq!(
            f.run(&[b"ZUNION", b"2", b"z", b"y", b"WEIGHTS", b"1"]),
            "-ERR syntax error\r\n"
        );
        assert_eq!(
            f.run(&[b"ZUNION", b"2", b"z", b"y", b"WEIGHTS", b"a", b"b"]),
            "-ERR weight value is not a float\r\n"
        );
        assert_eq!(
            f.run(&[b"ZUNION", b"2", b"z", b"y", b"AGGREGATE", b"NOPE"]),
            "-ERR syntax error\r\n"
        );
    }

    /// The three store forms, which answer a count and take no WITHSCORES.
    #[test]
    fn the_algebra_stores_answer_a_count_and_delete_an_empty_destination() {
        let mut f = Fixture::new();
        f.run(&[b"ZADD", b"z", b"1", b"a", b"2", b"b", b"3", b"c"]);
        f.run(&[b"ZADD", b"y", b"10", b"b", b"20", b"d"]);
        assert_eq!(f.run(&[b"ZUNIONSTORE", b"d", b"2", b"z", b"y"]), ":4\r\n");
        assert_eq!(
            f.run(&[b"ZRANGE", b"d", b"0", b"-1", b"WITHSCORES"]),
            "*8\r\n$1\r\na\r\n$1\r\n1\r\n$1\r\nc\r\n$1\r\n3\r\n$1\r\nb\r\n$2\r\n12\r\n$1\r\nd\r\n$2\r\n20\r\n"
        );
        assert_eq!(f.run(&[b"ZINTERSTORE", b"d", b"2", b"z", b"y"]), ":1\r\n");
        assert_eq!(f.run(&[b"ZDIFFSTORE", b"d", b"2", b"z", b"y"]), ":2\r\n");
        // An empty result deletes the destination rather than leaving an empty
        // sorted set, because an empty one does not exist.
        assert_eq!(
            f.run(&[b"ZINTERSTORE", b"d", b"2", b"z", b"nokey"]),
            ":0\r\n"
        );
        assert_eq!(f.run(&[b"EXISTS", b"d"]), ":0\r\n");
        // The destination is allowed to name its own source.
        assert_eq!(f.run(&[b"ZUNIONSTORE", b"z", b"2", b"z", b"y"]), ":4\r\n");
        assert_eq!(f.run(&[b"ZCARD", b"z"]), ":4\r\n");
        for cmd in [
            &[
                b"ZUNIONSTORE".as_slice(),
                b"d",
                b"2",
                b"z",
                b"y",
                b"WITHSCORES",
            ][..],
            &[
                b"ZDIFFSTORE",
                b"d",
                b"2",
                b"z",
                b"y",
                b"WEIGHTS",
                b"1",
                b"1",
            ],
        ] {
            assert_eq!(f.run(cmd), "-ERR syntax error\r\n", "{cmd:?}");
        }
    }

    /// `ZINTERCARD`, which counts without building anything.
    #[test]
    fn intercard_counts_and_stops_at_its_limit() {
        let mut f = Fixture::new();
        f.run(&[b"ZADD", b"z", b"1", b"a", b"2", b"b", b"3", b"c"]);
        f.run(&[b"ZADD", b"y", b"10", b"b", b"20", b"c", b"30", b"d"]);
        assert_eq!(f.run(&[b"ZINTERCARD", b"2", b"z", b"y"]), ":2\r\n");
        // A limit of zero is no limit, which is Redis's reading of it.
        assert_eq!(
            f.run(&[b"ZINTERCARD", b"2", b"z", b"y", b"LIMIT", b"0"]),
            ":2\r\n"
        );
        assert_eq!(
            f.run(&[b"ZINTERCARD", b"2", b"z", b"y", b"LIMIT", b"1"]),
            ":1\r\n"
        );
        // A negative limit and a limit that is not a number at all get the same
        // sentence, which looks like a mistake in Redis and is copied as one.
        let bad = "-ERR LIMIT can't be negative\r\n";
        assert_eq!(
            f.run(&[b"ZINTERCARD", b"2", b"z", b"y", b"LIMIT", b"-1"]),
            bad
        );
        assert_eq!(
            f.run(&[b"ZINTERCARD", b"2", b"z", b"y", b"LIMIT", b"x"]),
            bad
        );
        for cmd in [
            &[b"ZINTERCARD".as_slice(), b"3", b"z", b"y"][..],
            &[b"ZINTERCARD", b"2", b"z", b"y", b"LIMIT"],
            &[b"ZINTERCARD", b"2", b"z", b"y", b"junk", b"1"],
        ] {
            assert_eq!(f.run(cmd), "-ERR syntax error\r\n", "{cmd:?}");
        }
    }

    /// `ZRANDMEMBER`, which answers two different shapes out of one name.
    #[test]
    fn a_draw_answers_one_member_or_an_array_of_them() {
        let mut f = Fixture::new();
        f.run(&[b"ZADD", b"z", b"1", b"a", b"2", b"b", b"3", b"c"]);
        // No count is one member or a nil, a count is an array that may be
        // empty, and those are two reply types the client has to tell apart.
        assert_eq!(f.run(&[b"ZRANDMEMBER", b"nokey"]), "$-1\r\n");
        assert_eq!(f.run(&[b"ZRANDMEMBER", b"nokey", b"3"]), "*0\r\n");
        assert_eq!(f.run(&[b"ZRANDMEMBER", b"z", b"0"]), "*0\r\n");
        assert!(f.run(&[b"ZRANDMEMBER", b"z"]).starts_with("$1\r\n"));
        // A positive count draws without replacement, so a count over the size
        // answers the whole set and never a member twice.
        let all = f.run(&[b"ZRANDMEMBER", b"z", b"10"]);
        assert!(all.starts_with("*3\r\n"), "{all}");
        for m in ["a", "b", "c"] {
            assert!(all.contains(m), "{all}");
        }
        // A negative one draws with replacement and answers exactly as many as
        // it was asked for, whatever the size of the set.
        assert!(
            f.run(&[b"ZRANDMEMBER", b"z", b"-5"]).starts_with("*5\r\n"),
            "five draws with replacement"
        );
        assert!(
            f.run(&[b"ZRANDMEMBER", b"z", b"2", b"WITHSCORES"])
                .starts_with("*4\r\n"),
            "two pairs, flat on RESP2"
        );
        f.out = Out::new(Proto::Resp3);
        let got = f.run(&[b"ZRANDMEMBER", b"z", b"2", b"WITHSCORES"]);
        assert!(got.starts_with("*2\r\n*2\r\n"), "{got}");
        assert_eq!(f.run(&[b"ZRANDMEMBER", b"nokey"]), "_\r\n");
        f.out = Out::new(Proto::Resp2);
        assert_eq!(
            f.run(&[b"ZRANDMEMBER", b"z", b"2", b"junk"]),
            "-ERR syntax error\r\n"
        );
        assert_eq!(
            f.run(&[b"ZRANDMEMBER", b"z", b"x"]),
            "-ERR value is not an integer or out of range\r\n"
        );
    }

    /// `ZSCAN`, and the one sorted set reply where a score is not a double.
    #[test]
    fn a_sorted_set_scan_answers_pairs_of_strings_on_both_protocols() {
        let mut f = Fixture::new();
        f.run(&[b"ZADD", b"z", b"1", b"a", b"2", b"b", b"3", b"c"]);
        let all = "*2\r\n$1\r\n0\r\n*6\r\n$1\r\na\r\n$1\r\n1\r\n$1\r\nb\r\n$1\r\n2\r\n$1\r\nc\r\n$1\r\n3\r\n";
        assert_eq!(f.run(&[b"ZSCAN", b"z", b"0"]), all);
        assert_eq!(f.run(&[b"ZSCAN", b"z", b"0", b"COUNT", b"10"]), all);
        assert_eq!(
            f.run(&[b"ZSCAN", b"z", b"0", b"MATCH", b"a*"]),
            "*2\r\n$1\r\n0\r\n*2\r\n$1\r\na\r\n$1\r\n1\r\n"
        );
        assert_eq!(
            f.run(&[b"ZSCAN", b"nokey", b"0"]),
            "*2\r\n$1\r\n0\r\n*0\r\n"
        );
        // A score stays a bulk string on RESP3, which is the one place the two
        // protocols agree about a score and everywhere else they do not.
        f.out = Out::new(Proto::Resp3);
        assert_eq!(f.run(&[b"ZSCAN", b"z", b"0"]), all);
        f.out = Out::new(Proto::Resp2);
        assert_eq!(
            f.run(&[b"ZSCAN", b"z", b"0", b"NOVALUES"]),
            "-ERR NOVALUES option can only be used in HSCAN\r\n"
        );
        assert_eq!(f.run(&[b"ZSCAN", b"z", b"-1"]), "-ERR invalid cursor\r\n");
        assert_eq!(
            f.run(&[b"ZSCAN", b"z", b"0", b"COUNT", b"0"]),
            "-ERR syntax error\r\n"
        );
    }

    /// The count is what decides the shape, and its value is not.
    #[test]
    fn a_sorted_set_pop_changes_shape_when_it_is_given_a_count() {
        let mut f = Fixture::new();
        f.run(&[b"ZADD", b"z", b"1", b"a", b"2", b"b", b"3", b"c"]);
        // No count, so one flat pair, and the score is a bulk string on RESP2.
        assert_eq!(f.run(&[b"ZPOPMIN", b"z"]), "*2\r\n$1\r\na\r\n$1\r\n1\r\n");
        assert_eq!(f.run(&[b"ZPOPMAX", b"z"]), "*2\r\n$1\r\nc\r\n$1\r\n3\r\n");
        f.run(&[b"ZADD", b"z", b"1", b"a", b"3", b"c"]);
        // A count, so pairs, and on RESP2 they are flattened into one run.
        assert_eq!(
            f.run(&[b"ZPOPMIN", b"z", b"2"]),
            "*4\r\n$1\r\na\r\n$1\r\n1\r\n$1\r\nb\r\n$1\r\n2\r\n"
        );
        // An empty array rather than a null, which is where a sorted set pop and
        // a list pop part company, and the same answer a count of zero gives.
        assert_eq!(f.run(&[b"ZPOPMIN", b"nokey"]), "*0\r\n");
        assert_eq!(f.run(&[b"ZPOPMIN", b"nokey", b"2"]), "*0\r\n");
        assert_eq!(f.run(&[b"ZPOPMIN", b"z", b"0"]), "*0\r\n");
        // The last member takes the key with it.
        assert_eq!(
            f.run(&[b"ZPOPMIN", b"z", b"9"]),
            "*2\r\n$1\r\nc\r\n$1\r\n3\r\n"
        );
        assert_eq!(f.run(&[b"EXISTS", b"z"]), ":0\r\n");

        f.run(&[b"ZADD", b"z", b"1", b"a", b"2", b"b"]);
        f.out = Out::new(Proto::Resp3);
        assert_eq!(f.run(&[b"ZPOPMIN", b"z"]), "*2\r\n$1\r\na\r\n,1\r\n");
        assert_eq!(
            f.run(&[b"ZPOPMIN", b"z", b"1"]),
            "*1\r\n*2\r\n$1\r\nb\r\n,2\r\n"
        );
        f.out = Out::new(Proto::Resp2);
        // Both of these are the range error rather than the usual sentence about
        // integers, which is the odd answer and so the one worth copying.
        let bad = "-ERR value is out of range, must be positive\r\n";
        assert_eq!(f.run(&[b"ZPOPMIN", b"z", b"x"]), bad);
        assert_eq!(f.run(&[b"ZPOPMIN", b"z", b"-1"]), bad);
        assert_eq!(
            f.run(&[b"ZPOPMIN", b"z", b"1", b"2"]),
            "-ERR syntax error\r\n"
        );
    }

    /// `ZMPOP`, which is `LMPOP` with scores and the same parse.
    #[test]
    fn a_multi_key_pop_names_the_key_that_answered_and_nests_its_pairs() {
        let mut f = Fixture::new();
        f.run(&[b"ZADD", b"z", b"1", b"a", b"2", b"b", b"3", b"c"]);
        assert_eq!(
            f.run(&[b"ZMPOP", b"2", b"nokey", b"z", b"MIN"]),
            "*2\r\n$1\r\nz\r\n*1\r\n*2\r\n$1\r\na\r\n$1\r\n1\r\n"
        );
        // Nested on RESP2 as well, because the key name is already in front of
        // the pairs and there is nothing left to flatten into.
        assert_eq!(
            f.run(&[b"ZMPOP", b"1", b"z", b"MAX", b"COUNT", b"2"]),
            "*2\r\n$1\r\nz\r\n*2\r\n*2\r\n$1\r\nc\r\n$1\r\n3\r\n*2\r\n$1\r\nb\r\n$1\r\n2\r\n"
        );
        // A null array and not a null, the same as LMPOP.
        assert_eq!(f.run(&[b"ZMPOP", b"1", b"nokey", b"MIN"]), "*-1\r\n");
        f.out = Out::new(Proto::Resp3);
        assert_eq!(f.run(&[b"ZMPOP", b"1", b"nokey", b"MIN"]), "_\r\n");
        f.out = Out::new(Proto::Resp2);
        let numkeys = "-ERR numkeys should be greater than 0\r\n";
        for bad in [
            &[b"ZMPOP".as_slice(), b"0", b"z", b"MIN"][..],
            &[b"ZMPOP", b"-1", b"z", b"MIN"],
            &[b"ZMPOP", b"x", b"z", b"MIN"],
        ] {
            assert_eq!(f.run(bad), numkeys, "{:?}", bad[1]);
        }
        let count = "-ERR count should be greater than 0\r\n";
        for bad in [
            &[b"ZMPOP".as_slice(), b"1", b"z", b"MIN", b"COUNT", b"0"][..],
            &[b"ZMPOP", b"1", b"z", b"MIN", b"COUNT", b"-1"],
            &[b"ZMPOP", b"1", b"z", b"MIN", b"COUNT", b"x"],
        ] {
            assert_eq!(f.run(bad), count, "{:?}", bad[5]);
        }
        let syntax = "-ERR syntax error\r\n";
        for bad in [
            // Two keys named and one given, so the word that should have been
            // the direction is a key and there is no direction left.
            &[b"ZMPOP".as_slice(), b"2", b"z", b"MIN"][..],
            &[b"ZMPOP", b"1", b"z", b"SIDEWAYS"],
            &[b"ZMPOP", b"1", b"z", b"MIN", b"junk"],
            &[b"ZMPOP", b"1", b"z", b"MIN", b"COUNT", b"1", b"junk"],
        ] {
            assert_eq!(f.run(bad), syntax, "{bad:?}");
        }
    }

    /// The three that wait, when there is something there and they do not have
    /// to. `BZPOPMIN` is the one reply in the group that is three flat elements.
    #[test]
    fn the_sorted_set_pops_that_wait_answer_like_the_ones_they_wrap() {
        let mut f = Fixture::new();
        f.run(&[b"ZADD", b"z", b"1", b"a", b"2", b"b", b"3", b"c"]);
        assert_eq!(
            f.flow(&[b"BZPOPMIN", b"nokey", b"z", b"0"]),
            (
                Flow::Continue,
                "*3\r\n$1\r\nz\r\n$1\r\na\r\n$1\r\n1\r\n".to_owned()
            )
        );
        assert_eq!(
            f.run(&[b"BZPOPMAX", b"z", b"0"]),
            "*3\r\n$1\r\nz\r\n$1\r\nc\r\n$1\r\n3\r\n"
        );
        f.run(&[b"ZADD", b"z", b"1", b"a", b"3", b"c"]);
        assert_eq!(
            f.run(&[
                b"BZMPOP", b"0", b"2", b"nokey", b"z", b"MIN", b"COUNT", b"2"
            ]),
            "*2\r\n$1\r\nz\r\n*2\r\n*2\r\n$1\r\na\r\n$1\r\n1\r\n*2\r\n$1\r\nb\r\n$1\r\n2\r\n"
        );
        f.out = Out::new(Proto::Resp3);
        assert_eq!(
            f.run(&[b"BZPOPMIN", b"z", b"0"]),
            "*3\r\n$1\r\nz\r\n$1\r\nc\r\n,3\r\n"
        );
        f.out = Out::new(Proto::Resp2);
        // Nothing to take, so the client is parked and nothing was written.
        assert_eq!(
            f.flow(&[b"BZPOPMIN", b"z", b"0"]),
            (Flow::Block, String::new())
        );
        assert_eq!(
            f.flow(&[b"BZMPOP", b"0", b"1", b"z", b"MIN"]),
            (Flow::Block, String::new())
        );
        // The timeout is read before the key count, so this complains about the
        // timeout and not about the count.
        assert_eq!(
            f.run(&[b"BZMPOP", b"abc", b"0", b"z", b"MIN"]),
            "-ERR timeout is not a float or out of range\r\n"
        );
        assert_eq!(
            f.run(&[b"BZMPOP", b"0", b"0", b"z", b"MIN"]),
            "-ERR numkeys should be greater than 0\r\n"
        );
        assert_eq!(
            f.run(&[b"BZPOPMIN", b"z", b"-1"]),
            "-ERR timeout is negative\r\n"
        );
    }

    /// A parked sorted set client is served by whatever puts a member under one
    /// of its keys, and is not served by something of another type landing
    /// there.
    #[test]
    fn a_parked_sorted_set_client_waits_for_a_member_and_not_for_a_key() {
        let mut f = Fixture::new();
        assert_eq!(f.flow(&[b"BZPOPMIN", b"z", b"0"]).0, Flow::Block);
        assert_eq!(f.server.waiters().len(), 1);
        // A string under the key is not what it asked for, so it stays parked
        // rather than being handed a WRONGTYPE on a command that was accepted.
        f.run(&[b"SET", b"z", b"v"]);
        let mut out = Out::new(Proto::Resp2);
        assert!(!f.server.serve_waiter(0, 0, &mut out));
        assert!(out.as_slice().is_empty());
        f.run(&[b"DEL", b"z"]);
        f.run(&[b"ZADD", b"z", b"5", b"m"]);
        assert!(f.server.serve_waiter(0, 0, &mut out));
        assert_eq!(
            core::str::from_utf8(out.as_slice()).expect("ascii"),
            "*3\r\n$1\r\nz\r\n$1\r\nm\r\n$1\r\n5\r\n"
        );
        // And the member is gone, which is what makes a queue of workers on a
        // sorted set work at all.
        assert_eq!(f.run(&[b"EXISTS", b"z"]), ":0\r\n");
    }

    #[test]
    fn every_sorted_set_command_says_wrongtype_and_writes_nothing() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"s", b"v"]);
        let wrong = "-WRONGTYPE Operation against a key holding the wrong kind of value\r\n";
        for cmd in [
            &[b"ZADD".as_slice(), b"s", b"1", b"a"][..],
            &[b"ZINCRBY", b"s", b"1", b"a"],
            &[b"ZCARD", b"s"],
            &[b"ZSCORE", b"s", b"a"],
            &[b"ZMSCORE", b"s", b"a"],
            &[b"ZREM", b"s", b"a"],
            &[b"ZRANK", b"s", b"a"],
            &[b"ZREVRANK", b"s", b"a"],
            &[b"ZCOUNT", b"s", b"1", b"2"],
            &[b"ZLEXCOUNT", b"s", b"-", b"+"],
            &[b"ZRANGE", b"s", b"0", b"-1"],
            &[b"ZREVRANGE", b"s", b"0", b"-1"],
            &[b"ZRANGEBYSCORE", b"s", b"1", b"2"],
            &[b"ZREVRANGEBYSCORE", b"s", b"2", b"1"],
            &[b"ZRANGEBYLEX", b"s", b"-", b"+"],
            &[b"ZREVRANGEBYLEX", b"s", b"+", b"-"],
            &[b"ZRANGESTORE", b"d", b"s", b"0", b"-1"],
            &[b"ZREMRANGEBYRANK", b"s", b"0", b"-1"],
            &[b"ZREMRANGEBYSCORE", b"s", b"1", b"2"],
            &[b"ZREMRANGEBYLEX", b"s", b"-", b"+"],
            &[b"ZUNION", b"1", b"s"],
            &[b"ZINTER", b"1", b"s"],
            &[b"ZDIFF", b"1", b"s"],
            &[b"ZUNIONSTORE", b"d", b"1", b"s"],
            &[b"ZINTERSTORE", b"d", b"1", b"s"],
            &[b"ZDIFFSTORE", b"d", b"1", b"s"],
            &[b"ZINTERCARD", b"1", b"s"],
            &[b"ZRANDMEMBER", b"s"],
            &[b"ZSCAN", b"s", b"0"],
            &[b"ZPOPMIN", b"s"],
            &[b"ZPOPMAX", b"s", b"2"],
            &[b"ZMPOP", b"1", b"s", b"MIN"],
            &[b"BZPOPMIN", b"s", b"0"],
            &[b"BZPOPMAX", b"s", b"0"],
            &[b"BZMPOP", b"0", b"1", b"s", b"MIN"],
        ] {
            assert_eq!(f.run(cmd), wrong, "{:?}", cmd[0]);
        }
        assert_eq!(f.run(&[b"GET", b"s"]), "$1\r\nv\r\n");
    }

    /// The same churn the set, the string and the list get, because a sorted
    /// set that leaks a tree node per add looks exactly like one that does not
    /// until it has run for an afternoon.
    #[test]
    fn churning_sorted_sets_does_not_grow_the_server() {
        let mut f = Fixture::new();
        let members: Vec<Vec<u8>> = (0..200).map(|i| format!("m{i}").into_bytes()).collect();
        let scores: Vec<Vec<u8>> = (0..200).map(|i| format!("{i}").into_bytes()).collect();
        let mut args: Vec<&[u8]> = vec![b"ZADD", b"z"];
        for i in 0..200 {
            args.push(&scores[i]);
            args.push(&members[i]);
        }

        f.run(&args);
        f.run(&[b"DEL", b"z"]);
        f.server.compact_step();
        let after_first = f.server.memory_bytes();

        for _ in 0..200 {
            f.run(&args);
            f.run(&[b"DEL", b"z"]);
            f.server.compact_step();
        }
        assert_eq!(f.run(&[b"DBSIZE"]), ":0\r\n");
        assert!(
            f.server.memory_bytes() <= after_first * 2,
            "held {} after two hundred passes against {after_first} after one",
            f.server.memory_bytes()
        );
    }

    // ----------------------------------------------------------------- array

    #[test]
    fn an_array_writes_at_any_index_and_reads_back_what_it_sent() {
        let mut f = Fixture::new();
        // Three consecutive positions from a high index, and the reply is how
        // many of them were empty before rather than how many were written.
        assert_eq!(
            f.run(&[b"ARSET", b"a", b"1000", b"x", b"y", b"z"]),
            ":3\r\n"
        );
        assert_eq!(f.run(&[b"ARSET", b"a", b"1000", b"X", b"Y"]), ":0\r\n");
        assert_eq!(f.run(&[b"ARGET", b"a", b"1000"]), "$1\r\nX\r\n");
        assert_eq!(f.run(&[b"ARGET", b"a", b"1002"]), "$1\r\nz\r\n");
        // A hole and a key that is not there are the same answer.
        assert_eq!(f.run(&[b"ARGET", b"a", b"999"]), "$-1\r\n");
        assert_eq!(f.run(&[b"ARGET", b"nope", b"0"]), "$-1\r\n");
        assert_eq!(
            f.run(&[b"ARMGET", b"a", b"1002", b"999", b"1000"]),
            "*3\r\n$1\r\nz\r\n$-1\r\n$1\r\nX\r\n"
        );
        // Scattered pairs in one command, last write wins within it.
        assert_eq!(f.run(&[b"ARMSET", b"a", b"5", b"p", b"5", b"q"]), ":1\r\n");
        assert_eq!(f.run(&[b"ARGET", b"a", b"5"]), "$1\r\nq\r\n");
    }

    /// The two numbers an array reports are not the same number, and one of
    /// them does not fit a signed integer.
    #[test]
    fn the_length_is_the_high_water_mark_and_the_count_is_the_population() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"ARLEN", b"nope"]), ":0\r\n");
        assert_eq!(f.run(&[b"ARCOUNT", b"nope"]), ":0\r\n");
        f.run(&[b"ARMSET", b"a", b"0", b"x", b"9", b"y"]);
        assert_eq!(f.run(&[b"ARLEN", b"a"]), ":10\r\n");
        assert_eq!(f.run(&[b"ARCOUNT", b"a"]), ":2\r\n");
        // Deleting in the middle leaves the high water mark where it was.
        assert_eq!(f.run(&[b"ARDEL", b"a", b"0"]), ":1\r\n");
        assert_eq!(f.run(&[b"ARLEN", b"a"]), ":10\r\n");
        assert_eq!(f.run(&[b"ARCOUNT", b"a"]), ":1\r\n");

        // The top of the space is addressable, and its length is a number with
        // bit sixty three set, so the reply has to be unsigned or it comes back
        // negative.
        f.run(&[b"ARSET", b"top", b"18446744073709551614", b"z"]);
        assert_eq!(f.run(&[b"ARLEN", b"top"]), ":18446744073709551615\r\n");
        assert_eq!(f.run(&[b"ARCOUNT", b"top"]), ":1\r\n");
        // And one past it does not exist, so a write that would reach it fails
        // before any of it lands.
        assert_eq!(
            f.run(&[b"ARSET", b"over", b"18446744073709551614", b"a", b"b"]),
            "-ERR array index overflow\r\n"
        );
        assert_eq!(f.run(&[b"EXISTS", b"over"]), ":0\r\n");
    }

    /// One reply per position and not one per element, which is the whole
    /// reason the range is capped.
    #[test]
    fn a_range_read_answers_for_the_holes_too_and_is_capped_at_a_million() {
        let mut f = Fixture::new();
        f.run(&[b"ARSET", b"a", b"1", b"x"]);
        assert_eq!(
            f.run(&[b"ARGETRANGE", b"a", b"0", b"3"]),
            "*4\r\n$-1\r\n$1\r\nx\r\n$-1\r\n$-1\r\n"
        );
        // The two ends may come in either order, and the answer is reversed
        // rather than empty.
        assert_eq!(
            f.run(&[b"ARGETRANGE", b"a", b"3", b"0"]),
            "*4\r\n$-1\r\n$-1\r\n$1\r\nx\r\n$-1\r\n"
        );
        // A key that is not there reads like an array of nothing but holes.
        assert_eq!(
            f.run(&[b"ARGETRANGE", b"nope", b"0", b"1"]),
            "*2\r\n$-1\r\n$-1\r\n"
        );
        // A range wider than a million positions is refused and not trimmed,
        // because against a missing key it is a request for as many nulls as
        // the range is wide.
        assert_eq!(
            f.run(&[b"ARGETRANGE", b"nope", b"0", b"18446744073709551614"]),
            "-ERR range exceeds maximum of 1000000 items\r\n"
        );
    }

    /// Every index in the argument list is read before the key is touched, so
    /// a bad one at the end leaves nothing half written.
    #[test]
    fn a_bad_index_late_in_the_line_writes_none_of_the_earlier_ones() {
        let mut f = Fixture::new();
        assert_eq!(
            f.run(&[b"ARMSET", b"a", b"0", b"x", b"-1", b"y"]),
            "-ERR invalid array index\r\n"
        );
        assert_eq!(f.run(&[b"EXISTS", b"a"]), ":0\r\n");
        f.run(&[b"ARSET", b"a", b"0", b"x", b"y", b"z"]);
        assert_eq!(
            f.run(&[b"ARDEL", b"a", b"0", b"01"]),
            "-ERR invalid array index\r\n"
        );
        assert_eq!(f.run(&[b"ARCOUNT", b"a"]), ":3\r\n");
        // An index is unsigned here, so the numbers a list would take are not
        // the last element, they are errors.
        assert_eq!(
            f.run(&[b"ARGET", b"a", b"-1"]),
            "-ERR invalid array index\r\n"
        );
        // And a pair list with an odd tail is an arity error rather than a
        // syntax one.
        assert_eq!(
            f.run(&[b"ARMSET", b"a", b"0", b"x", b"1"]),
            "-ERR wrong number of arguments for 'armset' command\r\n"
        );
        assert_eq!(
            f.run(&[b"ARDELRANGE", b"a", b"0", b"1", b"2"]),
            "-ERR wrong number of arguments for 'ardelrange' command\r\n"
        );
    }

    #[test]
    fn a_range_delete_costs_the_elements_and_takes_the_key_when_it_empties() {
        let mut f = Fixture::new();
        f.run(&[b"ARSET", b"a", b"0", b"0", b"1", b"2", b"3", b"4"]);
        assert_eq!(f.run(&[b"ARDELRANGE", b"a", b"3", b"1"]), ":3\r\n");
        assert_eq!(f.run(&[b"ARCOUNT", b"a"]), ":2\r\n");
        // Two ranges in one command, and the second one covers the whole space
        // without walking it.
        assert_eq!(
            f.run(&[
                b"ARDELRANGE",
                b"a",
                b"100",
                b"200",
                b"0",
                b"18446744073709551614"
            ]),
            ":2\r\n"
        );
        assert_eq!(f.run(&[b"EXISTS", b"a"]), ":0\r\n");
        assert_eq!(f.run(&[b"ARDELRANGE", b"nope", b"0", b"1"]), ":0\r\n");
        assert_eq!(f.run(&[b"ARDEL", b"nope", b"0"]), ":0\r\n");
    }

    /// A value goes out as the bytes it came in as, whichever of the three ways
    /// the array found to store it.
    #[test]
    fn a_value_comes_back_byte_for_byte_however_it_was_packed() {
        let mut f = Fixture::new();
        let long = vec![b'v'; 200];
        f.run(&[
            b"ARMSET", b"a", b"0", b"42", b"1", b"007", b"2", b"3.5", b"3", b"3.14", b"4",
            b"short", b"5", &long, b"6", b"-0",
        ]);
        // 42 is an integer, 007 is not one because it does not print back the
        // same, 3.5 survives a double and 3.14 does not, and the last two are a
        // word packed string and a blob.
        assert_eq!(
            f.run(&[b"ARGETRANGE", b"a", b"0", b"6"]),
            format!(
                "*7\r\n$2\r\n42\r\n$3\r\n007\r\n$3\r\n3.5\r\n$4\r\n3.14\r\n$5\r\nshort\r\n$200\r\n{}\r\n$2\r\n-0\r\n",
                String::from_utf8_lossy(&long)
            )
        );
    }

    #[test]
    fn an_array_is_a_type_and_an_encoding_a_client_can_see() {
        let mut f = Fixture::new();
        f.run(&[b"ARSET", b"a", b"0", b"x"]);
        assert_eq!(f.run(&[b"TYPE", b"a"]), "+array\r\n");
        assert_eq!(
            f.run(&[b"OBJECT", b"ENCODING", b"a"]),
            "$12\r\nsliced-array\r\n"
        );
        // And it is a body like any other, so the key commands work on it.
        assert_eq!(f.run(&[b"EXPIRE", b"a", b"100"]), ":1\r\n");
        assert_eq!(f.run(&[b"PERSIST", b"a"]), ":1\r\n");
        assert_eq!(f.run(&[b"COPY", b"a", b"b"]), ":1\r\n");
        assert_eq!(f.run(&[b"ARGET", b"b", b"0"]), "$1\r\nx\r\n");
        assert_eq!(f.run(&[b"RENAME", b"a", b"c"]), "+OK\r\n");
        assert_eq!(f.run(&[b"ARCOUNT", b"c"]), ":1\r\n");
    }

    #[test]
    fn every_array_command_refuses_a_key_holding_something_else() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"s", b"v"]);
        let wrong = "-WRONGTYPE Operation against a key holding the wrong kind of value\r\n";
        for cmd in [
            &[b"ARSET".as_ref(), b"s", b"0", b"x"][..],
            &[b"ARMSET".as_ref(), b"s", b"0", b"x"][..],
            &[b"ARGET".as_ref(), b"s", b"0"][..],
            &[b"ARMGET".as_ref(), b"s", b"0"][..],
            &[b"ARGETRANGE".as_ref(), b"s", b"0", b"1"][..],
            &[b"ARLEN".as_ref(), b"s"][..],
            &[b"ARCOUNT".as_ref(), b"s"][..],
            &[b"ARDEL".as_ref(), b"s", b"0"][..],
            &[b"ARDELRANGE".as_ref(), b"s", b"0", b"1"][..],
            &[b"ARINSERT".as_ref(), b"s", b"x"][..],
            &[b"ARRING".as_ref(), b"s", b"4", b"x"][..],
            &[b"ARNEXT".as_ref(), b"s"][..],
            &[b"ARSEEK".as_ref(), b"s", b"1"][..],
            &[b"ARLASTITEMS".as_ref(), b"s", b"1"][..],
            &[b"ARSCAN".as_ref(), b"s", b"0", b"1"][..],
            &[b"ARGREP".as_ref(), b"s", b"0", b"1", b"EXACT", b"v"][..],
            &[b"AROP".as_ref(), b"s", b"0", b"1", b"SUM"][..],
            &[b"ARINFO".as_ref(), b"s"][..],
        ] {
            assert_eq!(f.run(cmd), wrong, "{}", String::from_utf8_lossy(cmd[0]));
        }
    }

    /// Two of the array commands look the key up before they read the index and
    /// the rest read the index first, so the same broken argument gets two
    /// different errors depending on which command it went to.
    #[test]
    fn a_bad_index_reports_the_type_only_where_redis_reports_it() {
        let mut f = Fixture::new();
        f.run(&[b"SET", b"s", b"v"]);
        let wrong = "-WRONGTYPE Operation against a key holding the wrong kind of value\r\n";
        let bad = "-ERR invalid array index\r\n";
        assert_eq!(f.run(&[b"ARGET", b"s", b"-1"]), wrong);
        assert_eq!(f.run(&[b"ARMGET", b"s", b"0", b"-1"]), wrong);
        assert_eq!(f.run(&[b"ARSET", b"s", b"-1", b"x"]), bad);
        assert_eq!(f.run(&[b"ARDEL", b"s", b"-1"]), bad);
        assert_eq!(f.run(&[b"ARSCAN", b"s", b"-1", b"0"]), bad);
        assert_eq!(f.run(&[b"ARGREP", b"s", b"-1", b"0", b"EXACT", b"v"]), bad);
        // And on a key that is an array the index is just an index.
        f.run(&[b"ARSET", b"a", b"0", b"x"]);
        assert_eq!(f.run(&[b"ARGET", b"a", b"-1"]), bad);
        assert_eq!(f.run(&[b"ARGET", b"nope", b"-1"]), bad);
    }

    #[test]
    fn an_append_follows_a_cursor_the_client_can_move() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"ARNEXT", b"nope"]), ":0\r\n");
        assert_eq!(f.run(&[b"ARINSERT", b"a", b"x", b"y"]), ":1\r\n");
        assert_eq!(f.run(&[b"ARNEXT", b"a"]), ":2\r\n");
        assert_eq!(f.run(&[b"ARINSERT", b"a", b"z"]), ":2\r\n");
        assert_eq!(f.run(&[b"ARGET", b"a", b"2"]), "$1\r\nz\r\n");

        // A seek says where the next one goes, and a missing key has no cursor
        // to move and is not created by the asking.
        assert_eq!(f.run(&[b"ARSEEK", b"nope", b"5"]), ":0\r\n");
        assert_eq!(f.run(&[b"EXISTS", b"nope"]), ":0\r\n");
        assert_eq!(f.run(&[b"ARSEEK", b"a", b"100"]), ":1\r\n");
        assert_eq!(f.run(&[b"ARNEXT", b"a"]), ":100\r\n");
        assert_eq!(f.run(&[b"ARINSERT", b"a", b"far"]), ":100\r\n");
        assert_eq!(f.run(&[b"ARSEEK", b"a", b"0"]), ":1\r\n");
        assert_eq!(f.run(&[b"ARNEXT", b"a"]), ":0\r\n");

        // The top of the space is the one index only ARSEEK will take, and it
        // leaves the cursor with nowhere to go.
        assert_eq!(f.run(&[b"ARSEEK", b"a", b"18446744073709551615"]), ":1\r\n");
        assert_eq!(f.run(&[b"ARNEXT", b"a"]), "$-1\r\n");
        assert_eq!(
            f.run(&[b"ARINSERT", b"a", b"x"]),
            "-ERR insert index overflow\r\n"
        );
        assert_eq!(
            f.run(&[b"ARSET", b"a", b"18446744073709551615", b"x"]),
            "-ERR invalid array index\r\n"
        );
    }

    #[test]
    fn a_ring_keeps_the_newest_and_renumbers_them_when_it_is_resized() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"ARRING", b"r", b"3", b"a", b"b", b"c"]), ":2\r\n");
        assert_eq!(f.run(&[b"ARRING", b"r", b"3", b"d", b"e"]), ":1\r\n");
        assert_eq!(f.run(&[b"ARLEN", b"r"]), ":3\r\n");
        assert_eq!(
            f.run(&[b"ARGETRANGE", b"r", b"0", b"2"]),
            "*3\r\n$1\r\nd\r\n$1\r\ne\r\n$1\r\nc\r\n"
        );
        // Growing it after it has wrapped puts the survivors back in the order
        // they arrived, which is the whole point of paying for the rebuild.
        assert_eq!(f.run(&[b"ARRING", b"r", b"5", b"f"]), ":3\r\n");
        assert_eq!(
            f.run(&[b"ARGETRANGE", b"r", b"0", b"3"]),
            "*4\r\n$1\r\nc\r\n$1\r\nd\r\n$1\r\ne\r\n$1\r\nf\r\n"
        );
        // The size is read before the key, so a bad one is a bad size wherever
        // it is sent.
        assert_eq!(
            f.run(&[b"ARRING", b"r", b"0", b"x"]),
            "-ERR size must be positive\r\n"
        );
        assert_eq!(
            f.run(&[b"ARRING", b"r", b"big", b"x"]),
            "-ERR invalid size\r\n"
        );
    }

    #[test]
    fn the_last_items_walk_back_from_the_cursor_and_report_the_holes() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"ARLASTITEMS", b"nope", b"5"]), "*0\r\n");
        f.run(&[b"ARRING", b"r", b"4", b"a", b"b", b"c", b"d", b"e"]);
        assert_eq!(
            f.run(&[b"ARLASTITEMS", b"r", b"3"]),
            "*3\r\n$1\r\nc\r\n$1\r\nd\r\n$1\r\ne\r\n"
        );
        assert_eq!(
            f.run(&[b"ARLASTITEMS", b"r", b"3", b"rev"]),
            "*3\r\n$1\r\ne\r\n$1\r\nd\r\n$1\r\nc\r\n"
        );
        assert_eq!(
            f.run(&[b"ARLASTITEMS", b"r", b"99"]),
            "*4\r\n$1\r\nb\r\n$1\r\nc\r\n$1\r\nd\r\n$1\r\ne\r\n",
            "more than there is gets what there is"
        );
        // Nothing asked for is an empty reply, and Redis answers that before it
        // has read the option or looked at the key.
        assert_eq!(f.run(&[b"ARLASTITEMS", b"r", b"0", b"junk"]), "*0\r\n");
        assert_eq!(
            f.run(&[b"ARLASTITEMS", b"r", b"1", b"junk"]),
            "-ERR syntax error\r\n"
        );
        assert_eq!(
            f.run(&[b"ARLASTITEMS", b"r", b"nine"]),
            "-ERR invalid COUNT\r\n"
        );

        // With no cursor the tail of the array is the anchor, and a hole inside
        // the window is reported as one.
        f.run(&[b"ARMSET", b"h", b"0", b"x", b"2", b"z"]);
        assert_eq!(
            f.run(&[b"ARLASTITEMS", b"h", b"5"]),
            "*2\r\n$-1\r\n$1\r\nz\r\n"
        );
    }

    #[test]
    fn a_scan_answers_pairs_for_what_is_there_and_skips_what_is_not() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"ARSCAN", b"nope", b"0", b"10"]), "*0\r\n");
        f.run(&[b"ARMSET", b"a", b"0", b"x", b"7", b"y", b"1000000", b"z"]);
        // The whole index space, which ARGETRANGE refuses and this one answers
        // in three visits because holes cost nothing.
        assert_eq!(
            f.run(&[b"ARSCAN", b"a", b"0", b"18446744073709551614"]),
            "*3\r\n*2\r\n:0\r\n$1\r\nx\r\n*2\r\n:7\r\n$1\r\ny\r\n*2\r\n:1000000\r\n$1\r\nz\r\n"
        );
        assert_eq!(
            f.run(&[
                b"ARSCAN",
                b"a",
                b"18446744073709551614",
                b"0",
                b"LIMIT",
                b"1"
            ]),
            "*1\r\n*2\r\n:1000000\r\n$1\r\nz\r\n"
        );
        assert_eq!(f.run(&[b"ARSCAN", b"a", b"1", b"6"]), "*0\r\n");
        assert_eq!(
            f.run(&[b"ARSCAN", b"a", b"0", b"10", b"LIMIT", b"0"]),
            "-ERR LIMIT must be positive\r\n"
        );
        assert_eq!(
            f.run(&[b"ARSCAN", b"a", b"0", b"10", b"NOPE", b"1"]),
            "-ERR syntax error\r\n"
        );
        assert_eq!(
            f.run(&[b"ARSCAN", b"a", b"0", b"10", b"LIMIT"]),
            "-ERR wrong number of arguments for 'arscan' command\r\n"
        );
    }

    #[test]
    fn a_grep_answers_the_indexes_whose_elements_match() {
        let mut f = Fixture::new();
        assert_eq!(
            f.run(&[b"ARGREP", b"nope", b"0", b"10", b"EXACT", b"x"]),
            "*0\r\n"
        );
        f.run(&[b"ARSET", b"a", b"0", b"alpha", b"beta", b"gamma", b"ALPHA"]);

        // The two bounds take the ends of the array as well as an index, and a
        // reversed range is walked backwards the way ARSCAN walks one.
        assert_eq!(
            f.run(&[b"ARGREP", b"a", b"-", b"+", b"GLOB", b"*a"]),
            "*3\r\n:0\r\n:1\r\n:2\r\n"
        );
        assert_eq!(
            f.run(&[b"ARGREP", b"a", b"+", b"-", b"GLOB", b"*a"]),
            "*3\r\n:2\r\n:1\r\n:0\r\n"
        );
        assert_eq!(
            f.run(&[b"ARGREP", b"a", b"1", b"2", b"GLOB", b"*a"]),
            "*2\r\n:1\r\n:2\r\n"
        );

        // One test each. NOCASE reaches all four of them and it may be written
        // after the pattern it applies to.
        assert_eq!(
            f.run(&[b"ARGREP", b"a", b"-", b"+", b"EXACT", b"alpha"]),
            "*1\r\n:0\r\n"
        );
        assert_eq!(
            f.run(&[b"ARGREP", b"a", b"-", b"+", b"EXACT", b"alpha", b"NOCASE"]),
            "*2\r\n:0\r\n:3\r\n"
        );
        assert_eq!(
            f.run(&[b"ARGREP", b"a", b"-", b"+", b"MATCH", b"mm"]),
            "*1\r\n:2\r\n"
        );
        assert_eq!(
            f.run(&[b"ARGREP", b"a", b"-", b"+", b"RE", b"^[bg]"]),
            "*2\r\n:1\r\n:2\r\n"
        );

        // OR is the default and AND has to be asked for, and either way the
        // last of a repeated option wins.
        let both: &[&[u8]] = &[
            b"ARGREP", b"a", b"-", b"+", b"EXACT", b"beta", b"MATCH", b"al",
        ];
        assert_eq!(f.run(both), "*2\r\n:0\r\n:1\r\n");
        assert_eq!(
            f.run(&[
                b"ARGREP", b"a", b"-", b"+", b"EXACT", b"beta", b"MATCH", b"al", b"AND"
            ]),
            "*0\r\n"
        );
        assert_eq!(
            f.run(&[
                b"ARGREP", b"a", b"-", b"+", b"EXACT", b"beta", b"MATCH", b"al", b"AND", b"OR"
            ]),
            "*2\r\n:0\r\n:1\r\n"
        );

        // WITHVALUES turns each hit into a pair, and LIMIT counts the hits and
        // not the positions it had to look at.
        assert_eq!(
            f.run(&[
                b"ARGREP",
                b"a",
                b"-",
                b"+",
                b"MATCH",
                b"a",
                b"WITHVALUES",
                b"LIMIT",
                b"2"
            ]),
            "*2\r\n*2\r\n:0\r\n$5\r\nalpha\r\n*2\r\n:1\r\n$4\r\nbeta\r\n"
        );
        assert_eq!(
            f.run(&[
                b"ARGREP", b"a", b"-", b"+", b"EXACT", b"ALPHA", b"LIMIT", b"1"
            ]),
            "*1\r\n:3\r\n"
        );
    }

    /// Everything ARGREP refuses, in the order it refuses it.
    #[test]
    fn a_grep_reports_a_broken_command_the_way_redis_does() {
        let mut f = Fixture::new();
        f.run(&[b"ARSET", b"a", b"0", b"alpha"]);
        let syntax = "-ERR syntax error\r\n";

        // The bounds are read before the plan, so a bad index beats a bad
        // predicate whichever way round the two are written.
        assert_eq!(
            f.run(&[b"ARGREP", b"a", b"-1", b"0", b"NOPE", b"x"]),
            "-ERR invalid array index\r\n"
        );
        assert_eq!(f.run(&[b"ARGREP", b"a", b"0", b"1", b"NOPE", b"x"]), syntax);
        // A keyword with nothing after it, and a command that asks for nothing.
        assert_eq!(
            f.run(&[b"ARGREP", b"a", b"0", b"1", b"NOCASE", b"EXACT"]),
            syntax
        );
        assert_eq!(
            f.run(&[b"ARGREP", b"a", b"0", b"1", b"EXACT", b"x", b"LIMIT"]),
            syntax
        );
        assert_eq!(
            f.run(&[b"ARGREP", b"a", b"0", b"1", b"NOCASE", b"WITHVALUES"]),
            syntax,
            "a command with no predicate in it at all"
        );
        assert_eq!(
            f.run(&[b"ARGREP", b"a", b"0", b"1", b"EXACT", b"x", b"LIMIT", b"0"]),
            "-ERR LIMIT must be positive\r\n"
        );
        assert_eq!(
            f.run(&[
                b"ARGREP", b"a", b"0", b"1", b"EXACT", b"x", b"LIMIT", b"nine"
            ]),
            "-ERR value is not an integer or out of range\r\n"
        );
        assert_eq!(
            f.run(&[b"ARGREP", b"a", b"0", b"1", b"RE", b""]),
            "-ERR regular expression is empty\r\n"
        );
        assert_eq!(
            f.run(&[b"ARGREP", b"a", b"0", b"1", b"RE", b"(a"]),
            "-ERR invalid regular expression: Missing ')'\r\n"
        );
        assert_eq!(
            f.run(&[b"ARGREP", b"a", b"0", b"1", b"RE", br"(a)\1"]),
            "-ERR regular expression backreferences are not supported\r\n"
        );
        // The arity is minus six, so a predicate keyword with no pattern after
        // it is short by one and never reaches the parser.
        let arity = "-ERR wrong number of arguments for 'argrep' command\r\n";
        assert_eq!(f.run(&[b"ARGREP", b"a", b"0", b"1", b"EXACT"]), arity);
        assert_eq!(f.run(&[b"ARGREP", b"a", b"0", b"1"]), arity);
    }

    #[test]
    fn an_op_reduces_a_range_to_one_number() {
        let mut f = Fixture::new();
        f.run(&[b"ARSET", b"a", b"0", b"1", b"2.5", b"word", b"-4"]);
        assert_eq!(
            f.run(&[b"AROP", b"a", b"0", b"10", b"SUM"]),
            "$4\r\n-0.5\r\n"
        );
        assert_eq!(f.run(&[b"AROP", b"a", b"0", b"10", b"min"]), "$2\r\n-4\r\n");
        assert_eq!(
            f.run(&[b"AROP", b"a", b"0", b"10", b"MAX"]),
            "$3\r\n2.5\r\n"
        );
        assert_eq!(f.run(&[b"AROP", b"a", b"0", b"10", b"USED"]), ":4\r\n");
        assert_eq!(
            f.run(&[b"AROP", b"a", b"0", b"10", b"MATCH", b"word"]),
            ":1\r\n"
        );
        // An aggregate is written with seventeen significant digits, which is
        // Redis's own choice and not what a score comes back as.
        f.run(&[b"ARSET", b"t", b"0", b"0.1", b"0.2"]);
        assert_eq!(
            f.run(&[b"AROP", b"t", b"0", b"10", b"SUM"]),
            "$19\r\n0.30000000000000004\r\n"
        );
        assert_eq!(f.run(&[b"ZADD", b"z", b"0.3", b"m"]), ":1\r\n");
        assert_eq!(f.run(&[b"ZSCORE", b"z", b"m"]), "$3\r\n0.3\r\n");

        // Nothing to work with is a null, and a missing key is a null for the
        // aggregates and a zero for the two that count.
        f.run(&[b"ARSET", b"w", b"0", b"word"]);
        assert_eq!(f.run(&[b"AROP", b"w", b"0", b"10", b"SUM"]), "$-1\r\n");
        assert_eq!(f.run(&[b"AROP", b"nope", b"0", b"10", b"SUM"]), "$-1\r\n");
        assert_eq!(f.run(&[b"AROP", b"nope", b"0", b"10", b"USED"]), ":0\r\n");

        assert_eq!(
            f.run(&[b"AROP", b"a", b"0", b"10", b"NOPE"]),
            "-ERR unknown operation\r\n"
        );
        assert_eq!(
            f.run(&[b"AROP", b"a", b"0", b"10", b"MATCH"]),
            "-ERR MATCH requires a value argument\r\n"
        );
        assert_eq!(
            f.run(&[b"AROP", b"a", b"0", b"10", b"SUM", b"extra"]),
            "-ERR wrong number of arguments for 'arop' command\r\n"
        );
    }

    #[test]
    fn the_info_is_a_map_and_a_missing_key_is_an_error() {
        let mut f = Fixture::new();
        assert_eq!(f.run(&[b"ARINFO", b"nope"]), "-ERR no such key\r\n");
        f.run(&[b"ARINSERT", b"a", b"x", b"y"]);
        let short = f.run(&[b"ARINFO", b"a"]);
        assert!(
            short.starts_with("*14\r\n"),
            "seven pairs on RESP2: {short}"
        );
        assert!(short.contains("$5\r\ncount\r\n:2\r\n"), "{short}");
        assert!(
            short.contains("$17\r\nnext-insert-index\r\n:2\r\n"),
            "{short}"
        );
        assert!(short.contains("$10\r\nslice-size\r\n:4096\r\n"), "{short}");
        let full = f.run(&[b"ARINFO", b"a", b"full"]);
        assert!(full.starts_with("*24\r\n"), "twelve pairs: {full}");
        // Two values one apart are held sparsely, so the dense count is zero and
        // the two dense averages have nothing to average.
        assert!(full.contains("$12\r\ndense-slices\r\n:0\r\n"), "{full}");
        assert!(full.contains("$13\r\nsparse-slices\r\n:1\r\n"), "{full}");
        assert!(
            full.contains("$14\r\navg-dense-size\r\n$1\r\n0\r\n"),
            "{full}"
        );
        assert_eq!(f.run(&[b"ARINFO", b"a", b"nope"]), "-ERR syntax error\r\n");

        // On RESP3 the same reply is a map and the averages are doubles.
        let mut g = Fixture::new();
        g.run(&[b"HELLO", b"3"]);
        g.run(&[b"ARINSERT", b"a", b"x"]);
        let map = g.run(&[b"ARINFO", b"a", b"FULL"]);
        assert!(map.starts_with("%12\r\n"), "{map}");
        assert!(map.contains("$5\r\ncount\r\n:1\r\n"), "{map}");
        assert!(map.contains("$14\r\navg-dense-size\r\n,0\r\n"), "{map}");
    }

    #[test]
    fn a_double_on_the_wire_is_written_the_way_redis_writes_one() {
        let mut f = Fixture::new();
        // Whole numbers up to two to the sixty second come back as integers,
        // and past that the digit generator takes over and uses an exponent.
        for (score, want) in [
            ("3", "3"),
            ("3.5", "3.5"),
            ("0.3", "0.3"),
            ("1e30", "1e+30"),
            ("1e19", "1e+19"),
            ("1e-7", "1e-7"),
            ("0.000001", "0.000001"),
            ("4611686018427387904", "4611686018427387904"),
            ("-0", "-0"),
        ] {
            f.run(&[b"ZADD", b"z", score.as_bytes(), b"m"]);
            assert_eq!(
                f.run(&[b"ZSCORE", b"z", b"m"]),
                format!("${}\r\n{want}\r\n", want.len()),
                "score {score}"
            );
        }

        // The same bytes on RESP3, where the reply is a double rather than a
        // bulk string.
        let mut g = Fixture::new();
        g.run(&[b"HELLO", b"3"]);
        g.run(&[b"ZADD", b"z", b"1e30", b"m"]);
        assert_eq!(g.run(&[b"ZSCORE", b"z", b"m"]), ",1e+30\r\n");
        // The two float increments are not this printer. They go through
        // ld2string in its human mode, which is a fixed point conversion with
        // the trailing zeros taken off, so they never write an exponent, and
        // they reply with a bulk string on both protocols.
        assert_eq!(
            g.run(&[b"INCRBYFLOAT", b"s", b"1e30"]),
            "$31\r\n1000000000000000000000000000000\r\n"
        );
        assert_eq!(g.run(&[b"INCRBYFLOAT", b"t", b"0.1"]), "$3\r\n0.1\r\n");
        assert_eq!(
            g.run(&[b"HINCRBYFLOAT", b"h", b"f", b"1e19"]),
            "$20\r\n10000000000000000000\r\n"
        );
    }
}