iggy_binary_protocol 0.11.0

// 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 bytemuck::{CheckedBitPattern, NoUninit};

/// Replicated operation discriminant. Identifies the state machine operation
/// carried in a consensus message body.
#[derive(Default, Debug, Clone, Copy, PartialEq, Eq, NoUninit, CheckedBitPattern)]
#[repr(u8)]
pub enum Operation {
    /// The value 0 is reserved to prevent a spurious zero from being
    /// interpreted as a valid operation.
    #[default]
    Reserved = 0,

    /// Register a client session with the cluster. Goes through the same
    /// consensus pipeline (prepare/replicate/commit) as normal operations
    /// but skips state machine dispatch at commit time, the metadata
    /// plane calls `commit_register` directly, which mints the session's
    /// fence epoch from the committed Register log position.
    Register = 1,

    /// Non-replicated client request carried in VSR framing. The concrete
    /// legacy command code is stored outside the operation discriminant by the
    /// transport/server bridge and must not enter the replicated log.
    NonReplicated = 2,

    /// Unregister a client session with the cluster. Carries the legacy
    /// `LogoutUser` command through the consensus pipeline and removes the
    /// client-table entry at commit time.
    Logout = 3,
    // Internal metadata operations (journal / replica-only)
    CreateTopicWithAssignments = 64,
    CreatePartitionsWithAssignments = 65,
    /// Server-originated: drop a disconnected client from every consumer group
    /// it joined and rebalance. Applied as a deterministic side-effect of the
    /// `Logout` commit (not its own client command), so it has no wire code.
    RemoveConsumerGroupMember = 66,

    /// Reconciler-originated: complete a pending cooperative revocation, moving
    /// a drained partition to its target member. Submitted through metadata
    /// consensus by the partition reconciler; no client wire code.
    CompleteConsumerGroupRevocation = 67,

    /// Server-originated: set a partition's delete watermark (the offset below
    /// which sealed segments are removed). The dispatch handler resolves a
    /// client `DeleteSegments` count to a concrete offset on the owning shard
    /// and submits this through metadata consensus, so every replica applies
    /// the same offset deterministically. No client wire code.
    TruncatePartition = 68,

    // Metadata operations (shard 0)
    CreateStream = 128,
    UpdateStream = 129,
    DeleteStream = 130,
    PurgeStream = 131,
    CreateTopic = 132,
    UpdateTopic = 133,
    DeleteTopic = 134,
    PurgeTopic = 135,
    CreatePartitions = 136,
    DeletePartitions = 137,
    // Client op handled specially: the dispatch layer resolves the requested
    // segment count to an offset on the owning shard and replicates a
    // `TruncatePartition` through metadata, so `DeleteSegments` itself is
    // neither a metadata nor a partition consensus op (see `is_partition`).
    DeleteSegments = 138,
    CreateConsumerGroup = 139,
    DeleteConsumerGroup = 140,
    CreateUser = 141,
    UpdateUser = 142,
    DeleteUser = 143,
    ChangePassword = 144,
    UpdatePermissions = 145,
    CreatePersonalAccessToken = 146,
    DeletePersonalAccessToken = 147,
    JoinConsumerGroup = 148,
    LeaveConsumerGroup = 149,

    // Partition operations (routed by namespace)
    SendMessages = 160,
    StoreConsumerOffset = 161,
    DeleteConsumerOffset = 162,
}

impl Operation {
    /// Whether `code` is a discriminant this build defines.
    ///
    /// The typed decode needs this to tell an operation a newer release added
    /// from a corrupted header byte: bytemuck's checked cast rejects both with
    /// one undifferentiated error, and only the former is fixable by upgrading
    /// this node.
    #[must_use]
    pub fn is_known_code(code: u8) -> bool {
        bytemuck::checked::try_cast::<u8, Self>(code).is_ok()
    }

    pub const INTERNAL_START: u8 = Self::CreateTopicWithAssignments as u8;
    pub const METADATA_START: u8 = Self::CreateStream as u8;
    pub const PARTITION_START: u8 = Self::SendMessages as u8;

    /// Internal-only operations reserved for replica / journal use.
    #[must_use]
    #[inline]
    pub const fn is_internal(&self) -> bool {
        (*self as u8) >= Self::INTERNAL_START && (*self as u8) < Self::METADATA_START
    }

    /// Metadata / control-plane operations handled by shard 0.
    #[must_use]
    #[inline]
    pub const fn is_metadata(&self) -> bool {
        if self.is_internal() {
            return true;
        }

        matches!(
            self,
            Self::CreateStream
                | Self::UpdateStream
                | Self::DeleteStream
                | Self::PurgeStream
                | Self::CreateTopic
                | Self::UpdateTopic
                | Self::DeleteTopic
                | Self::PurgeTopic
                | Self::CreatePartitions
                | Self::DeletePartitions
                | Self::CreateConsumerGroup
                | Self::DeleteConsumerGroup
                | Self::CreateUser
                | Self::UpdateUser
                | Self::DeleteUser
                | Self::ChangePassword
                | Self::UpdatePermissions
                | Self::CreatePersonalAccessToken
                | Self::DeletePersonalAccessToken
                | Self::JoinConsumerGroup
                | Self::LeaveConsumerGroup
        )
    }

    /// VSR protocol-level operations that are not mapped directly to a
    /// legacy command code.
    #[must_use]
    #[inline]
    pub const fn is_vsr_reserved(&self) -> bool {
        matches!(
            self,
            Self::Reserved | Self::Register | Self::Logout | Self::NonReplicated
        )
    }

    /// Whether a committed/rejected reply for this operation carries the
    /// result section (`[count: u32]` then `count` x `{index, result}`) ahead
    /// of the payload. This is the single source of truth the encode sites
    /// (`ApplyReply::write_reply_body`, the partition plane's
    /// `committed_reply_body`, `build_result_rejection_reply` callers) and the
    /// SDK decode carve-out (`split_metadata_result`) must agree on; a drifted
    /// site decodes a rejection as `Ok` or misreads a payload as the count.
    /// Metadata ops are all result-framed; on the partition plane only the
    /// consumer-offset ops are (they reject with typed errors at admission).
    /// `Register` is result-framed only when non-empty -- that nuance stays in
    /// the SDK, the one place that sees Register replies.
    #[must_use]
    pub const fn is_result_framed(&self) -> bool {
        self.is_metadata() || matches!(self, Self::StoreConsumerOffset | Self::DeleteConsumerOffset)
    }

    /// Data-plane operations routed to the shard owning the partition.
    #[must_use]
    #[inline]
    pub const fn is_partition(&self) -> bool {
        (*self as u8) >= Self::PARTITION_START
    }

    /// Operations that replicate through the METADATA consensus group and live
    /// in its WAL.
    ///
    /// Wider than [`Self::is_metadata`]: the session ops replicate on the
    /// metadata plane without being metadata mutations. The single source of
    /// truth for "does the metadata plane own this op", shared by the plane's
    /// own applicability predicate and the repair router's legacy-stamp
    /// acceptance -- the two drifting is how a metadata op ends up offered to
    /// the partition arm.
    #[must_use]
    #[inline]
    pub const fn is_metadata_plane(&self) -> bool {
        self.is_metadata() || matches!(self, Self::Register | Self::Logout)
    }

    /// Whether a consensus plane claims this operation. Anything else the plane
    /// chain drops in its `()` terminator, so entry paths must refuse it first.
    #[must_use]
    #[inline]
    pub const fn is_plane_routable(&self) -> bool {
        self.is_metadata_plane() || self.is_partition()
    }

    /// Operations clients are allowed to send directly.
    #[must_use]
    #[inline]
    pub const fn is_client_allowed(&self) -> bool {
        !matches!(self, Self::Reserved) && !self.is_internal()
    }

    /// Bidirectional mapping: `Operation` -> client command code.
    ///
    /// Delegates to the dispatch table as the single source of truth.
    #[must_use]
    pub const fn to_command_code(&self) -> Option<u32> {
        match self {
            Self::Reserved
            | Self::Register
            | Self::Logout
            | Self::NonReplicated
            | Self::CreateTopicWithAssignments
            | Self::CreatePartitionsWithAssignments
            | Self::RemoveConsumerGroupMember
            | Self::CompleteConsumerGroupRevocation
            | Self::TruncatePartition => None,
            Self::CreateStream
            | Self::UpdateStream
            | Self::DeleteStream
            | Self::PurgeStream
            | Self::CreateTopic
            | Self::UpdateTopic
            | Self::DeleteTopic
            | Self::PurgeTopic
            | Self::CreatePartitions
            | Self::DeletePartitions
            | Self::DeleteSegments
            | Self::CreateConsumerGroup
            | Self::DeleteConsumerGroup
            | Self::CreateUser
            | Self::UpdateUser
            | Self::DeleteUser
            | Self::ChangePassword
            | Self::UpdatePermissions
            | Self::CreatePersonalAccessToken
            | Self::DeletePersonalAccessToken
            | Self::JoinConsumerGroup
            | Self::LeaveConsumerGroup
            | Self::SendMessages
            | Self::StoreConsumerOffset
            | Self::DeleteConsumerOffset => match crate::dispatch::lookup_by_operation(*self) {
                Some(meta) => Some(meta.code),
                None => None,
            },
        }
    }

    /// Bidirectional mapping: client command code -> `Operation`.
    ///
    /// Delegates to the dispatch table as the single source of truth.
    #[must_use]
    pub const fn from_command_code(code: u32) -> Option<Self> {
        match crate::dispatch::lookup_command(code) {
            Some(meta) => meta.operation,
            None => None,
        }
    }
}

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

    #[test]
    fn command_code_roundtrip() {
        let ops = [
            Operation::CreateStream,
            Operation::UpdateStream,
            Operation::DeleteStream,
            Operation::PurgeStream,
            Operation::CreateTopic,
            Operation::UpdateTopic,
            Operation::DeleteTopic,
            Operation::PurgeTopic,
            Operation::CreatePartitions,
            Operation::DeletePartitions,
            Operation::DeleteSegments,
            Operation::CreateConsumerGroup,
            Operation::DeleteConsumerGroup,
            Operation::CreateUser,
            Operation::UpdateUser,
            Operation::DeleteUser,
            Operation::ChangePassword,
            Operation::UpdatePermissions,
            Operation::CreatePersonalAccessToken,
            Operation::DeletePersonalAccessToken,
            Operation::JoinConsumerGroup,
            Operation::LeaveConsumerGroup,
            Operation::SendMessages,
            Operation::StoreConsumerOffset,
            Operation::DeleteConsumerOffset,
        ];
        for op in ops {
            let code = op
                .to_command_code()
                .unwrap_or_else(|| panic!("Operation {op:?} has no command code mapping"));
            let back = Operation::from_command_code(code)
                .unwrap_or_else(|| panic!("command code {code} has no Operation mapping"));
            assert_eq!(op, back, "roundtrip failed for {op:?} (code={code})");
        }
    }

    #[test]
    fn vsr_reserved_have_no_code() {
        assert_eq!(Operation::Reserved.to_command_code(), None);
        assert_eq!(Operation::Register.to_command_code(), None);
        assert_eq!(Operation::Logout.to_command_code(), None);
        assert_eq!(Operation::NonReplicated.to_command_code(), None);
    }

    #[test]
    fn vsr_reserved_classification() {
        assert!(Operation::Reserved.is_vsr_reserved());
        assert!(Operation::Register.is_vsr_reserved());
        assert!(Operation::Logout.is_vsr_reserved());
        assert!(Operation::NonReplicated.is_vsr_reserved());
        assert!(!Operation::CreateStream.is_vsr_reserved());
        assert!(!Operation::SendMessages.is_vsr_reserved());
        assert!(!Operation::Register.is_metadata());
        assert!(!Operation::Register.is_partition());
        assert!(!Operation::Logout.is_metadata());
        assert!(!Operation::Logout.is_partition());
        assert_eq!(
            Operation::CreateTopicWithAssignments.to_command_code(),
            None
        );
        assert_eq!(
            Operation::CreatePartitionsWithAssignments.to_command_code(),
            None
        );
    }

    #[test]
    fn read_only_commands_have_no_operation() {
        use crate::codes::{GET_STATS_CODE, GET_STREAM_CODE, PING_CODE, POLL_MESSAGES_CODE};
        assert!(Operation::from_command_code(PING_CODE).is_none());
        assert!(Operation::from_command_code(GET_STATS_CODE).is_none());
        assert!(Operation::from_command_code(GET_STREAM_CODE).is_none());
        assert!(Operation::from_command_code(POLL_MESSAGES_CODE).is_none());
    }

    #[test]
    fn metadata_vs_partition() {
        assert!(Operation::CreateTopicWithAssignments.is_internal());
        assert!(Operation::CreateTopicWithAssignments.is_metadata());
        assert!(!Operation::CreateTopicWithAssignments.is_client_allowed());
        assert!(Operation::CreateStream.is_metadata());
        assert!(!Operation::CreateStream.is_partition());
        assert!(Operation::CreateStream.is_client_allowed());
        assert!(Operation::SendMessages.is_partition());
        assert!(!Operation::SendMessages.is_metadata());
        assert!(!Operation::DeleteSegments.is_partition());
        assert!(!Operation::DeleteSegments.is_metadata());
        assert!(Operation::TruncatePartition.is_internal());
        assert!(Operation::TruncatePartition.is_metadata());
        assert!(!Operation::TruncatePartition.is_client_allowed());
        assert!(Operation::StoreConsumerOffset.is_partition());
        assert!(Operation::DeleteConsumerOffset.is_partition());
    }

    /// Every operation belongs to exactly one plane, or to the short list
    /// answered before the chain. Walks `is_known_code` so a new variant fails
    /// here rather than on the first frame carrying it.
    #[test]
    fn unroutable_operations_are_listed() {
        // Answered earlier: `Reserved` by `validate_request_fields`,
        // `NonReplicated` by the reads router, `DeleteSegments` by resolution to
        // `TruncatePartition` (`server::dispatch::classify`).
        const UNROUTABLE: [Operation; 3] = [
            Operation::Reserved,
            Operation::NonReplicated,
            Operation::DeleteSegments,
        ];

        for code in 0..=u8::MAX {
            if !Operation::is_known_code(code) {
                continue;
            }
            let operation: Operation = bytemuck::checked::cast(code);
            assert!(
                !(operation.is_metadata_plane() && operation.is_partition()),
                "{operation:?} is claimed by both planes; the chain takes the first"
            );
            assert_eq!(
                operation.is_plane_routable(),
                !UNROUTABLE.contains(&operation),
                "{operation:?}: is_plane_routable={} disagrees with the unroutable list",
                operation.is_plane_routable(),
            );
        }
    }
}