iggy 0.11.0-edge.4

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

use crate::session::ConsensusSession;
use bytes::{BufMut, Bytes, BytesMut};
use iggy_binary_protocol::codes::{
    LOGIN_REGISTER_CODE, LOGIN_REGISTER_WITH_PAT_CODE, LOGOUT_USER_CODE,
};
use iggy_binary_protocol::consensus::{
    Command, EvictionHeader, EvictionReason, GenericHeader, HEADER_SIZE, Operation, ReplyHeader,
    RequestHeader, read_size_field, result_code, result_section_len,
};
use iggy_common::{IggyError, calculate_checksum, eviction_reason_to_error};

const NON_REPLICATED_CODE_RANGE: std::ops::Range<usize> = 0..4;

// TODO(vsr): transparent retry-after-disconnect can weaken at-most-once semantics
// for replicated writes. The transports currently disconnect, reconnect, and encode
// the retry from the current ConsensusSession. If disconnect created a fresh VSR
// client/session, the retried request gets a new (client_id, request_id) tuple, so
// server-side deduplication cannot match a mutation that may already have committed
// before the transport failure.
//
// The fix is to keep the ConsensusSession's client_id and request counter across
// reconnects and retry replicated writes under the same (client_id, request_id)
// instead of re-registering fresh. Resume happens through the LOGIN path: the
// reconnecting client re-authenticates presenting its previous client_id, the
// server verifies the authenticated user owns that entry, and the rebind commits
// a Register that adopts the entry with its watermark and reply ring intact. Note
// the epoch changes -- the rebind moves the fence to the new register's op -- so
// the session field must be taken from the new login reply, not carried over.
//
// There is deliberately no credential-free rebind: presenting (client, session)
// on an unauthenticated transport is refused, since that pair is a dedup key and
// never a bearer token.
pub(crate) fn encode_contiguous_request(
    session: &mut ConsensusSession,
    code: u32,
    payload: &Bytes,
) -> Result<Bytes, IggyError> {
    let (header, total_size) = encode_request_header(session, code, payload)?;
    let mut request = BytesMut::with_capacity(total_size);
    request.put_slice(bytemuck::bytes_of(&header));
    request.put_slice(payload);
    Ok(request.freeze())
}

pub(crate) fn encode_request_header(
    session: &mut ConsensusSession,
    code: u32,
    payload: &Bytes,
) -> Result<(RequestHeader, usize), IggyError> {
    let (operation, request_id, session_id) = match code {
        LOGIN_REGISTER_CODE | LOGIN_REGISTER_WITH_PAT_CODE => {
            // A re-login reuses this `ConsensusSession`; `begin_register`
            // re-arms it (fresh session) so the Register encodes cleanly instead
            // of tripping the one-shot register guard. Safe under the transport
            // stream lock: it is lockstep (one request in flight), so no
            // in-flight request observes the reset.
            (Operation::Register, session.begin_register(), 0)
        }
        _ => {
            let operation = operation_for_code(code);
            // NonReplicated ops (ping, reads) bypass server-side dedup --
            // `ClientTable` only tracks request_ids for replicated ops, and
            // the table accepts any id above the watermark with no
            // contiguity requirement (client_table.rs: "There is no
            // `RequestGap`"), so consuming the counter would not break the
            // next metadata op. Read the current id without advancing
            // because the server ignores it for NonReplicated and burning
            // ids for requests the table never sees buys nothing.
            //
            // They are also sessionless on the server (routed by transport
            // id; protected codes are auth-gated server-side), so send with
            // session 0 when unregistered -- preserving the legacy "ping
            // works without auth" contract. Replicated ops still require a
            // bound session for dedup and fail fast here.
            if operation == Operation::NonReplicated {
                (
                    operation,
                    session.current_request_id(),
                    session.session().unwrap_or(0),
                )
            } else if operation.is_partition() {
                // Partition ops replicate in their own per-partition group,
                // which is at-least-once with no `ClientTable` dedup -- the
                // metadata table never records their request ids, so there
                // is nothing for a consumed id to deduplicate against. Every
                // partition request on a session therefore carries the id
                // the next metadata op will claim, and a partition-plane
                // replay is at-least-once.
                let session_id = session.session().ok_or(IggyError::Unauthenticated)?;
                (operation, session.current_request_id(), session_id)
            } else {
                let session_id = session.session().ok_or(IggyError::Unauthenticated)?;
                (operation, session.next_request_id(), session_id)
            }
        }
    };
    // Stamped only for ops the server's `ClientTable` dedups. Partition ops are
    // at-least-once with no reply cache to poison, and theirs are the large payloads,
    // already covered client-side by `batch_checksum` over the same bytes.
    // NonReplicated ops bypass dedup too.
    let request_checksum = if operation.is_partition() || operation == Operation::NonReplicated {
        0
    } else {
        u128::from(calculate_checksum(payload))
    };
    let total_size = HEADER_SIZE
        .checked_add(payload.len())
        .ok_or(IggyError::InvalidConfiguration)?;
    let size = u32::try_from(total_size).map_err(|_| IggyError::InvalidConfiguration)?;
    let mut reserved = [0; 60];
    if operation == Operation::NonReplicated {
        reserved[NON_REPLICATED_CODE_RANGE].copy_from_slice(&code.to_le_bytes());
    }
    let header = RequestHeader {
        command: Command::Request,
        operation,
        size,
        client: session.client_id(),
        request: request_id,
        session: session_id,
        // Lets the client table tell a genuine retry from a `request` number reused
        // for different arguments. Zero means unstamped, which is what an SDK
        // predating this sends. A server that rewrites the body (PAT, password)
        // carries it through untouched, so it keeps describing what the client sent.
        request_checksum,
        // Zeroed: the field is "informational" -- the server copies it into
        // `ReplyHeader.timestamp` for RTT but nothing else reads it. Paying
        // a `clock_gettime` syscall per encoded request (formerly held the
        // `consensus_session` lock too) for an unused field is waste.
        // Reintroduce a real stamp here when an RTT consumer actually wires
        // it up.
        timestamp: 0,
        reserved,
        ..Default::default()
    };

    Ok((header, total_size))
}

/// `COMMAND_TABLE` is a protocol registry, not a per-server capability list, so
/// an SDK build cannot know which codes a given server implements. The server is
/// the authority: an unmapped code is forwarded as non-replicated (the code
/// rides `RequestHeader.reserved`, which that path already stamps) and the
/// server answers with a proper error if it does not know it.
fn operation_for_code(code: u32) -> Operation {
    if code == LOGOUT_USER_CODE {
        return Operation::Logout;
    }

    Operation::from_command_code(code).unwrap_or(Operation::NonReplicated)
}

pub(crate) fn response_size(header: &[u8]) -> Result<usize, IggyError> {
    let size = read_size_field(header).ok_or(IggyError::InvalidCommand)? as usize;
    if size < HEADER_SIZE {
        return Err(IggyError::InvalidCommand);
    }
    Ok(size)
}

pub(crate) fn decode_response(response: Bytes) -> Result<Bytes, IggyError> {
    if response.len() < HEADER_SIZE {
        return Err(IggyError::EmptyResponse);
    }

    let header_bytes: &[u8; HEADER_SIZE] = response[..HEADER_SIZE]
        .try_into()
        .map_err(|_| IggyError::InvalidCommand)?;
    match peek_command(header_bytes) {
        Command::Eviction => Err(decode_eviction(header_bytes)),
        Command::Reply => {
            let total_size = response_size(header_bytes)?;
            if response.len() < total_size {
                return Err(IggyError::InvalidCommand);
            }
            if let Some(error) = read_reply_status(header_bytes) {
                return Err(error);
            }
            let operation = read_operation(header_bytes)?;
            split_metadata_result(operation, response.slice(HEADER_SIZE..total_size))
        }
        _ => Err(IggyError::InvalidCommand),
    }
}

/// Decode a reply when the header and body have been read into separate
/// buffers. Saves the 64B header `put_slice` that `decode_response` would
/// otherwise perform when callers concatenate header + body before decoding.
///
/// Also surfaces session-terminal `Command::Eviction` frames as typed
/// errors: callers waiting on a Reply for an unbound session would otherwise
/// hit a read-timeout because the SDK previously only accepted
/// `Command::Reply`. Returns the body slice on a normal Reply, or maps the
/// eviction reason to an `IggyError` so the request fails fast.
pub(crate) fn decode_response_split(
    header_bytes: &[u8; HEADER_SIZE],
    body: Bytes,
) -> Result<Bytes, IggyError> {
    match peek_command(header_bytes) {
        Command::Eviction => Err(decode_eviction(header_bytes)),
        Command::Reply => {
            let expected_body = response_size(header_bytes)? - HEADER_SIZE;
            if body.len() < expected_body {
                return Err(IggyError::InvalidCommand);
            }
            if let Some(error) = read_reply_status(header_bytes) {
                return Err(error);
            }
            let operation = read_operation(header_bytes)?;
            split_metadata_result(operation, body.slice(..expected_body))
        }
        _ => Err(IggyError::InvalidCommand),
    }
}

/// Peek `ReplyHeader.status`, the pre-commit deny channel shared by every deny
/// frame: dispatch-time denies (authorization and a malformed user-password
/// body) via `build_deny_reply`, and partition-primary admission rejects via
/// `consensus::build_deny_reply_from_request`. Read before any body decode:
/// a nonzero status maps to its [`IggyError`] and fails the request, an empty
/// or partial body notwithstanding. `None` (status 0) lets the reply flow to
/// the body/result-section decode. Read by wire offset for the same
/// misalignment reason as [`read_operation`]; a committed metadata reply stamps
/// 0 here and carries its business rejection in the result section (see
/// `split_metadata_result`).
fn read_reply_status(header_bytes: &[u8; HEADER_SIZE]) -> Option<IggyError> {
    const STATUS_OFFSET: usize = std::mem::offset_of!(ReplyHeader, status);
    let mut bytes = [0u8; 4];
    bytes.copy_from_slice(&header_bytes[STATUS_OFFSET..STATUS_OFFSET + 4]);
    let status = u32::from_le_bytes(bytes);
    (status != 0).then(|| IggyError::from_code(status))
}

/// Read the [`Operation`] discriminant from a reply header by wire offset.
/// Mirrors [`peek_command`]: a full `ReplyHeader` struct cast is avoided
/// because the response buffers are not 16-aligned, so the leading `u128`
/// fields would make `try_from_bytes::<ReplyHeader>` fail on an unlucky
/// placement. One validated byte is enough to drive `split_metadata_result`.
fn read_operation(header_bytes: &[u8; HEADER_SIZE]) -> Result<Operation, IggyError> {
    const OPERATION_OFFSET: usize = std::mem::offset_of!(ReplyHeader, operation);
    bytemuck::checked::try_from_bytes::<Operation>(
        &header_bytes[OPERATION_OFFSET..=OPERATION_OFFSET],
    )
    .copied()
    .map_err(|_| IggyError::InvalidCommand)
}

/// Interpret the committed result section that leads a metadata reply body.
///
/// Metadata ops ([`Operation::is_metadata`]) commit a result
/// section ahead of their typed payload (encode mirror:
/// `metadata::stm::result::ApplyReply::write_reply_body`): success carries
/// `count == 0` then the payload; a committed business rejection carries one
/// `{index, result}` entry and no payload. Strip the section on success and map
/// a nonzero committed code to its [`IggyError`] -- the result discriminants
/// share the `IggyError` code space, so this is the same [`IggyError::from_code`]
/// mapping the legacy transport applies to a status word.
///
/// Reads, the partition data plane, and Register/Logout carry no result section
/// and pass through untouched. A metadata body that is not a well-formed result
/// section is corruption, never a silent success, so it maps to `InvalidCommand`
/// rather than risk a rejection decoding as `Ok`.
fn split_metadata_result(operation: Operation, body: Bytes) -> Result<Bytes, IggyError> {
    // Register (login/register) replies are result-framed too, so a transient
    // login decodes to `TransientNotCommitted` and the SDK replays it. The one
    // exception is a terminal failure, which ships an empty body (no result
    // section) and is passed through to fail the typed `LoginRegisterResponse`
    // decode. `Operation::is_result_framed` is the shared source of truth with
    // the server-side encode sites; the Register empty-body-is-terminal nuance
    // is the one SDK-side addition. Other reads, data-plane ops, and Logout
    // carry no result section and pass through untouched.
    let result_framed =
        operation.is_result_framed() || (operation == Operation::Register && !body.is_empty());
    if !result_framed {
        return Ok(body);
    }
    match result_code(&body) {
        Some(0) => {
            let payload_start = result_section_len(&body).ok_or(IggyError::InvalidCommand)?;
            Ok(body.slice(payload_start..))
        }
        Some(code) => Err(IggyError::from_code(code)),
        None => Err(IggyError::InvalidCommand),
    }
}

/// `Command` lives at a fixed offset shared by every consensus header
/// (Reply, Eviction, Prepare, ...), so a byte read is enough to discriminate
/// the frame.
fn peek_command(header_bytes: &[u8; HEADER_SIZE]) -> Command {
    const COMMAND_OFFSET: usize = std::mem::offset_of!(GenericHeader, command);
    match header_bytes[COMMAND_OFFSET] {
        x if x == Command::Reply as u8 => Command::Reply,
        x if x == Command::Eviction as u8 => Command::Eviction,
        _ => Command::Reserved,
    }
}

/// Map a session-terminal Eviction frame to a typed error. Fields are read
/// by wire offset instead of an `EvictionHeader` struct cast: the response
/// buffers are not 16-aligned (the header holds `u128`s, so the cast would
/// fail on an unlucky buffer placement), and reading raw lets the SDK apply
/// the same window sanity check as `EvictionHeader::validate` rather than
/// trusting the remote frame.
fn decode_eviction(header_bytes: &[u8; HEADER_SIZE]) -> IggyError {
    const REASON_OFFSET: usize = std::mem::offset_of!(EvictionHeader, reason);
    const VERSION_OFFSET: usize = std::mem::offset_of!(EvictionHeader, server_protocol_version);
    const VERSION_MIN_OFFSET: usize =
        std::mem::offset_of!(EvictionHeader, server_protocol_version_min);

    let Ok(&reason) = bytemuck::checked::try_from_bytes::<EvictionReason>(
        &header_bytes[REASON_OFFSET..=REASON_OFFSET],
    ) else {
        return IggyError::Unauthenticated;
    };
    eviction_reason_to_error(
        reason,
        read_window_field(header_bytes, VERSION_OFFSET),
        read_window_field(header_bytes, VERSION_MIN_OFFSET),
    )
}

fn read_window_field(header_bytes: &[u8; HEADER_SIZE], offset: usize) -> u32 {
    let mut value = [0u8; 4];
    value.copy_from_slice(&header_bytes[offset..offset + 4]);
    u32::from_le_bytes(value)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::session::ConsensusSession;
    use iggy_binary_protocol::codes::{CREATE_STREAM_CODE, GET_STREAM_CODE, PING_CODE};
    use iggy_binary_protocol::requests::streams::CreateStreamRequest;
    use iggy_binary_protocol::requests::users::LoginRegisterRequest;
    use iggy_binary_protocol::version::IGGY_PROTOCOL_VERSION;
    use iggy_binary_protocol::{ClientVersionInfo, WireEncode, WireName, WireOptions};
    use secrecy::SecretString;

    fn decode_request_header(bytes: &Bytes) -> RequestHeader {
        *bytemuck::checked::try_from_bytes::<RequestHeader>(&bytes[..HEADER_SIZE]).unwrap()
    }

    #[test]
    fn second_register_on_bound_session_re_arms_instead_of_panicking() {
        let request = LoginRegisterRequest {
            version_info: ClientVersionInfo {
                protocol_version: IGGY_PROTOCOL_VERSION,
                sdk_name: WireName::new("rust-sdk").unwrap(),
                sdk_version: WireName::new("1.0.0").unwrap(),
            },
            username: WireName::new("admin").unwrap(),
            password: SecretString::from("secret"),
            client_context: None,
        };

        let mut session = ConsensusSession::with_client_id(7);
        encode_contiguous_request(&mut session, LOGIN_REGISTER_CODE, &request.to_bytes()).unwrap();
        session.bind(42);

        // A second login on the same bound session must encode a fresh Register
        // (request 0, session 0), not panic in the one-shot register guard.
        let bytes =
            encode_contiguous_request(&mut session, LOGIN_REGISTER_CODE, &request.to_bytes())
                .unwrap();
        let header = decode_request_header(&bytes);
        assert_eq!(header.operation, Operation::Register);
        assert_eq!(header.request, 0);
        assert_eq!(header.session, 0);
        assert!(!session.is_bound());
    }

    #[test]
    fn eviction_incompatible_protocol_decodes_to_typed_error() {
        use iggy_binary_protocol::version::IGGY_PROTOCOL_VERSION_MIN;

        // Header lands at a guaranteed-misaligned address (aligned start +1):
        // eviction decode reads fields by offset and must not care.
        #[repr(C, align(16))]
        struct Misaligner([u8; HEADER_SIZE + 1]);

        let header = EvictionHeader::incompatible_protocol(
            0,
            0,
            0,
            0xCAFE,
            IGGY_PROTOCOL_VERSION,
            IGGY_PROTOCOL_VERSION_MIN,
        );
        let mut raw = Misaligner([0; HEADER_SIZE + 1]);
        raw.0[1..].copy_from_slice(bytemuck::bytes_of(&header));
        let shifted: &[u8; HEADER_SIZE] = raw.0[1..].try_into().unwrap();

        let result = decode_response_split(shifted, Bytes::new());
        assert!(matches!(
            result,
            Err(IggyError::IncompatibleProtocolVersion(client, min, max))
                if client == IGGY_PROTOCOL_VERSION
                    && min == IGGY_PROTOCOL_VERSION_MIN
                    && max == IGGY_PROTOCOL_VERSION
        ));
    }

    #[test]
    fn eviction_with_invalid_window_degrades_to_unauthenticated() {
        for (server_max, server_min) in [(1, 0), (1, 2)] {
            let mut header = EvictionHeader::incompatible_protocol(0, 0, 0, 0xCAFE, 1, 1);
            header.server_protocol_version = server_max;
            header.server_protocol_version_min = server_min;
            let mut buf = [0u8; HEADER_SIZE];
            buf.copy_from_slice(bytemuck::bytes_of(&header));

            let result = decode_response_split(&buf, Bytes::new());
            assert!(
                matches!(result, Err(IggyError::Unauthenticated)),
                "window [{server_min}, {server_max}] must not surface as typed error"
            );
        }
    }

    #[test]
    fn reply_with_nonzero_status_surfaces_as_typed_error() {
        // A dispatch-time authorization denial rides `ReplyHeader.status`; the
        // decode funnel surfaces it as the typed error before any body decode,
        // even though the deny body is empty.
        let header = ReplyHeader {
            command: Command::Reply,
            size: HEADER_SIZE as u32,
            status: IggyError::Unauthorized.as_code(),
            ..Default::default()
        };
        let mut buf = [0u8; HEADER_SIZE];
        buf.copy_from_slice(bytemuck::bytes_of(&header));
        let result = decode_response_split(&buf, Bytes::new());
        assert!(matches!(result, Err(IggyError::Unauthorized)));
    }

    #[test]
    fn reply_with_zero_status_passes_body_through() {
        // status 0 is the ok channel: a non-metadata reply returns its body.
        let header = ReplyHeader {
            command: Command::Reply,
            operation: Operation::NonReplicated,
            size: (HEADER_SIZE + 3) as u32,
            ..Default::default()
        };
        let mut buf = [0u8; HEADER_SIZE];
        buf.copy_from_slice(bytemuck::bytes_of(&header));
        let out = decode_response_split(&buf, Bytes::from_static(b"abc")).unwrap();
        assert_eq!(&out[..], b"abc");
    }

    #[test]
    fn replicated_request_increments_request_counter() {
        let mut session = ConsensusSession::with_client_id(42);
        let _ = session.register_request_id();
        session.bind(99);
        let payload = CreateStreamRequest {
            name: WireName::new("stream").unwrap(),
            options: WireOptions::empty(),
        }
        .to_bytes();

        let first = encode_contiguous_request(&mut session, CREATE_STREAM_CODE, &payload).unwrap();
        let second = encode_contiguous_request(&mut session, CREATE_STREAM_CODE, &payload).unwrap();

        assert_eq!(decode_request_header(&first).request, 1);
        assert_eq!(decode_request_header(&second).request, 2);
        assert_eq!(decode_request_header(&second).session, 99);
    }

    #[test]
    fn request_checksum_is_stamped_only_for_deduped_operations() {
        // The stamp exists to stop a reused `request` number returning the wrong
        // cached reply, so it is worth its hashing pass only where `ClientTable`
        // dedups. Partition payloads are the large ones and carry `batch_checksum`
        // over the same bytes already; hashing them again is pure cost.
        let mut session = ConsensusSession::with_client_id(42);
        session.bind(99);
        let payload = Bytes::from_static(b"payload");

        let deduped =
            encode_contiguous_request(&mut session, CREATE_STREAM_CODE, &payload).unwrap();
        assert_eq!(
            decode_request_header(&deduped).request_checksum,
            u128::from(calculate_checksum(&payload)),
        );

        let ping = encode_contiguous_request(&mut session, PING_CODE, &Bytes::new()).unwrap();
        assert_eq!(decode_request_header(&ping).request_checksum, 0);
    }

    #[test]
    fn ping_uses_non_replicated_operation() {
        let mut session = ConsensusSession::with_client_id(42);
        session.bind(99);
        let bytes = encode_contiguous_request(&mut session, PING_CODE, &Bytes::new()).unwrap();
        let header = decode_request_header(&bytes);

        assert_eq!(header.operation, Operation::NonReplicated);
        assert_eq!(
            u32::from_le_bytes(
                header.reserved[NON_REPLICATED_CODE_RANGE]
                    .try_into()
                    .unwrap()
            ),
            PING_CODE
        );
        assert_eq!(header.session, 99);
    }

    #[test]
    fn logout_uses_replicated_logout_operation() {
        let mut session = ConsensusSession::with_client_id(42);
        session.bind(99);
        let bytes =
            encode_contiguous_request(&mut session, LOGOUT_USER_CODE, &Bytes::new()).unwrap();
        let header = decode_request_header(&bytes);

        assert_eq!(header.operation, Operation::Logout);
        assert_eq!(header.request, 1);
        assert_eq!(header.session, 99);
    }

    #[test]
    fn read_only_request_uses_non_replicated_operation() {
        let mut session = ConsensusSession::with_client_id(42);
        session.bind(99);
        let bytes =
            encode_contiguous_request(&mut session, GET_STREAM_CODE, &Bytes::new()).unwrap();
        let header = decode_request_header(&bytes);

        assert_eq!(header.operation, Operation::NonReplicated);
        assert_eq!(
            u32::from_le_bytes(
                header.reserved[NON_REPLICATED_CODE_RANGE]
                    .try_into()
                    .unwrap()
            ),
            GET_STREAM_CODE
        );
        assert_eq!(header.session, 99);
    }

    #[test]
    fn unknown_code_encodes_as_non_replicated_and_carries_the_code() {
        // An extended server may implement codes this SDK build has never heard
        // of. The registry is not a capability list, so the request must reach
        // the server rather than fail at encode time.
        const UNKNOWN_CODE: u32 = 60_000;
        assert!(
            iggy_binary_protocol::dispatch::lookup_command(UNKNOWN_CODE).is_none(),
            "test needs a code absent from COMMAND_TABLE"
        );

        let mut session = ConsensusSession::with_client_id(42);
        session.bind(99);
        let bytes = encode_contiguous_request(&mut session, UNKNOWN_CODE, &Bytes::new()).unwrap();
        let header = decode_request_header(&bytes);

        assert_eq!(header.operation, Operation::NonReplicated);
        assert_eq!(
            u32::from_le_bytes(
                header.reserved[NON_REPLICATED_CODE_RANGE]
                    .try_into()
                    .unwrap()
            ),
            UNKNOWN_CODE
        );
    }

    #[test]
    fn no_replicated_command_ever_resolves_to_non_replicated() {
        // The safety asymmetry that makes forwarding unknown codes acceptable:
        // an unknown code is the server's business, but a *known* replicated
        // command sent as non-replicated would apply on one node only and
        // silently diverge the replicas. Swept over the whole registry so a
        // future entry cannot regress it.
        for meta in iggy_binary_protocol::dispatch::COMMAND_TABLE {
            if !meta.is_replicated() {
                continue;
            }
            assert_ne!(
                operation_for_code(meta.code),
                Operation::NonReplicated,
                "replicated command {} ({}) must never encode as NonReplicated",
                meta.name,
                meta.code
            );
        }
    }

    #[test]
    fn metadata_success_reply_strips_result_section_and_returns_payload() {
        let mut body = BytesMut::new();
        body.put_u32_le(0); // count = 0 (success)
        body.put_slice(b"payload");
        let payload = split_metadata_result(Operation::CreateStream, body.freeze()).unwrap();
        assert_eq!(&payload[..], b"payload");
    }

    #[test]
    fn metadata_rejection_reply_maps_committed_code_to_iggy_error() {
        let mut body = BytesMut::new();
        body.put_u32_le(1); // count = 1 (one rejection entry)
        body.put_u32_le(0); // index
        body.put_u32_le(IggyError::StreamIdNotFound(Default::default()).as_code());
        let err = split_metadata_result(Operation::DeleteStream, body.freeze()).unwrap_err();
        assert_eq!(
            err.as_code(),
            IggyError::StreamIdNotFound(Default::default()).as_code()
        );
    }

    #[test]
    fn metadata_reply_with_truncated_section_is_invalid_command_never_ok() {
        // count claims an entry the body cannot hold: corruption, never a
        // rejection silently flipping to success.
        let mut body = BytesMut::new();
        body.put_u32_le(1);
        let err = split_metadata_result(Operation::CreateStream, body.freeze()).unwrap_err();
        assert!(matches!(err, IggyError::InvalidCommand));
    }

    #[test]
    fn non_metadata_reply_passes_through_without_a_result_section() {
        // Reads / partition-plane / Register-Logout replies carry no section.
        let body = Bytes::from_static(b"raw-non-metadata-body");
        let out = split_metadata_result(Operation::NonReplicated, body.clone()).unwrap();
        assert_eq!(out, body);
    }

    /// `[count = 1][index = 0][result = code]` -- the single-entry rejection
    /// shape (`ApplyReply::write_reply_body` / `build_transient_reply`).
    fn rejection_body(code: u32) -> Bytes {
        let mut body = Vec::with_capacity(12);
        body.extend_from_slice(&1u32.to_le_bytes());
        body.extend_from_slice(&0u32.to_le_bytes());
        body.extend_from_slice(&code.to_le_bytes());
        Bytes::from(body)
    }

    /// `[count = 0]` followed by the payload -- the result-framed success shape.
    fn success_body(payload: &[u8]) -> Bytes {
        let mut body = Vec::with_capacity(4 + payload.len());
        body.extend_from_slice(&0u32.to_le_bytes());
        body.extend_from_slice(payload);
        Bytes::from(body)
    }

    #[test]
    fn delete_consumer_offset_rejection_decodes_to_terminal_error() {
        let code = IggyError::ConsumerOffsetNotFound(0).as_code();
        let result = split_metadata_result(Operation::DeleteConsumerOffset, rejection_body(code));
        assert_eq!(
            result.unwrap_err().as_code(),
            code,
            "delete rejection must surface as the typed error, not decode as Ok"
        );
    }

    #[test]
    fn store_consumer_offset_rejection_decodes_to_terminal_error() {
        let code = IggyError::InvalidOffset(42).as_code();
        let result = split_metadata_result(Operation::StoreConsumerOffset, rejection_body(code));
        assert_eq!(result.unwrap_err().as_code(), code);
    }

    #[test]
    fn consumer_offset_success_strips_the_empty_result_section() {
        let out = split_metadata_result(Operation::StoreConsumerOffset, success_body(b"")).unwrap();
        assert!(out.is_empty());
        let out =
            split_metadata_result(Operation::DeleteConsumerOffset, success_body(b"")).unwrap();
        assert!(out.is_empty());
    }

    #[test]
    fn metadata_transient_code_decodes_to_transient_not_committed() {
        let code = IggyError::TransientNotCommitted.as_code();
        let result = split_metadata_result(Operation::CreateStream, rejection_body(code));
        assert!(matches!(
            result.unwrap_err(),
            IggyError::TransientNotCommitted
        ));
    }

    #[test]
    fn metadata_success_returns_payload_after_result_section() {
        let out = split_metadata_result(Operation::CreateStream, success_body(b"payload")).unwrap();
        assert_eq!(out.as_ref(), b"payload");
    }

    #[test]
    fn send_messages_body_is_never_interpreted_as_a_result_section() {
        // A data-plane reply whose first bytes happen to look like a rejection
        // must pass through untouched: `SendMessages` is not result-framed.
        let body = rejection_body(IggyError::InvalidOffset(1).as_code());
        let out = split_metadata_result(Operation::SendMessages, body.clone()).unwrap();
        assert_eq!(out, body);
    }
}