iggy_binary_protocol 0.11.1-edge.2

// 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::codes::{LOGIN_REGISTER_CODE, LOGIN_REGISTER_WITH_PAT_CODE, LOGOUT_USER_CODE};
use crate::consensus::{Command, HEADER_SIZE, Operation, RequestHeader};
use crate::error::WireError;
use ring::rand::{SecureRandom, SystemRandom};
use std::ops::Range;

/// Where `RequestHeader::reserved` carries the legacy command code of a
/// `NonReplicated` request, as little-endian `u32`.
pub const NON_REPLICATED_CODE_RANGE: Range<usize> = 0..4;

/// Maps a command code onto the consensus operation a client sends it as.
///
/// `COMMAND_TABLE` is a protocol registry, not a per-server capability list,
/// so a client 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 stamps) and the
/// server answers with a proper error if it does not know it.
#[must_use]
pub fn operation_for_code(code: u32) -> Operation {
    match code {
        LOGIN_REGISTER_CODE | LOGIN_REGISTER_WITH_PAT_CODE => Operation::Register,
        LOGOUT_USER_CODE => Operation::Logout,
        _ => Operation::from_command_code(code).unwrap_or(Operation::NonReplicated),
    }
}

impl RequestHeader {
    /// Builds the header a client sends for `code` with `payload`, given the
    /// identity the session assigned: `client`, `request` and `session`.
    ///
    /// Every rule that depends only on the code and the payload lives here so
    /// that each client, and the fixtures other SDKs verify against, agree on
    /// the bytes: metadata and session operations stamp `request_checksum`
    /// with a random per-request identifier, partition and non-replicated
    /// operations leave it zero (send batches carry their own checksum and non-replicated
    /// requests bypass dedup), and a non-replicated request carries its
    /// command code in `reserved`.
    ///
    /// The partition context fields stay zero. A client partition operation
    /// or poll stamps them afterwards with
    /// [`PartitionContext::stamp`](crate::PartitionContext::stamp). The server
    /// refuses such a request whose incarnation is not the partition's current
    /// one, zero included, with `HistoryUnavailable`.
    ///
    /// # Errors
    ///
    /// Returns `WireError::PayloadTooLarge` when the frame would not fit the
    /// `u32` size field, or `WireError::RandomSourceUnavailable` if the secure
    /// random source fails for a metadata or session request.
    pub fn for_request(
        code: u32,
        client: u128,
        request: u64,
        session: u64,
        payload: &[u8],
    ) -> Result<Self, WireError> {
        let operation = operation_for_code(code);
        let total_size = HEADER_SIZE.saturating_add(payload.len());
        let size = u32::try_from(total_size).map_err(|_| WireError::PayloadTooLarge {
            size: payload.len(),
            max: u32::MAX as usize - HEADER_SIZE,
        })?;
        let request_checksum = if operation.is_partition() || operation == Operation::NonReplicated
        {
            0
        } else {
            let random = SystemRandom::new();
            loop {
                let mut bytes = [0u8; 16];
                random
                    .fill(&mut bytes)
                    .map_err(|_| WireError::RandomSourceUnavailable)?;
                let id = u128::from_le_bytes(bytes);
                if id != 0 {
                    break id;
                }
            }
        };
        let mut reserved = [0; 4];
        if operation == Operation::NonReplicated {
            reserved[NON_REPLICATED_CODE_RANGE].copy_from_slice(&code.to_le_bytes());
        }
        Ok(Self {
            command: Command::Request,
            operation,
            size,
            client,
            request,
            session,
            request_checksum,
            // Replicated prepares get a server timestamp; no consumer needs a
            // client clock read here.
            timestamp: 0,
            reserved,
            ..Default::default()
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::codes::{CREATE_STREAM_CODE, PING_CODE, SEND_MESSAGES_CODE};
    use twox_hash::XxHash3_64;

    #[test]
    fn login_codes_register_and_logout_logs_out() {
        assert_eq!(operation_for_code(LOGIN_REGISTER_CODE), Operation::Register);
        assert_eq!(
            operation_for_code(LOGIN_REGISTER_WITH_PAT_CODE),
            Operation::Register
        );
        assert_eq!(operation_for_code(LOGOUT_USER_CODE), Operation::Logout);
        assert_eq!(operation_for_code(u32::MAX), Operation::NonReplicated);
    }

    #[test]
    fn given_identical_metadata_bodies_when_building_requests_should_stamp_distinct_random_ids() {
        let payload = [1, 2, 3];
        let header = RequestHeader::for_request(CREATE_STREAM_CODE, 7, 8, 9, &payload).unwrap();
        let another = RequestHeader::for_request(CREATE_STREAM_CODE, 7, 9, 9, &payload).unwrap();
        assert_eq!(header.command, Command::Request);
        assert_eq!(header.size as usize, HEADER_SIZE + payload.len());
        assert_eq!((header.client, header.request, header.session), (7, 8, 9));
        assert_ne!(header.request_checksum, 0);
        assert_ne!(header.request_checksum, another.request_checksum);
        assert_ne!(
            header.request_checksum,
            u128::from(XxHash3_64::oneshot(&payload))
        );
        assert_eq!(header.reserved, [0; 4]);
    }

    #[test]
    fn partition_and_non_replicated_requests_leave_the_stamp_zero() {
        let send = RequestHeader::for_request(SEND_MESSAGES_CODE, 1, 2, 3, &[9; 16]).unwrap();
        assert_eq!(send.request_checksum, 0);
        assert!(send.operation.is_partition());

        let ping = RequestHeader::for_request(PING_CODE, 1, 2, 3, &[]).unwrap();
        assert_eq!(ping.operation, Operation::NonReplicated);
        assert_eq!(ping.request_checksum, 0);
        assert_eq!(
            ping.reserved[NON_REPLICATED_CODE_RANGE],
            PING_CODE.to_le_bytes()
        );
    }
}