liminal-server 0.25.4

//! Canonical transition-input event framing for the incarnation stream.
//!
//! The grammar carries INPUTS only. There is no header field in any event, in
//! any version: `server_incarnation` and `last_examined_connection_ordinal` are
//! consequences of re-executing these rows through `liminal-protocol`, never
//! bytes a writer may choose. That absence is the property the whole design
//! rests on, so it lives here where the encoder and decoder can be read
//! together.

use liminal_protocol::wire::ConnectionIncarnation;

use super::ConnectionFateClass;
use super::error::IncarnationStreamError;

const EVENT_MAGIC: [u8; 4] = *b"LPIE";
const FROZEN_EVENT_SCHEMA_VERSION: u8 = 1;
const EVENT_SCHEMA_VERSION: u8 = 2;
const STARTUP_TAG: u8 = 1;
const ALLOCATE_TAG: u8 = 2;
const OPEN_CONNECTION_FATE_TAG: u8 = 3;
const COMPLETE_CONNECTION_FATE_TAG: u8 = 4;
const EVENT_HEADER_LEN: usize = 10;
const ALLOCATE_BOUND_LEN: usize = 8;
const ALLOCATE_COUNT_LEN: usize = 8;
const ALLOCATE_FIXED_LEN: usize = ALLOCATE_BOUND_LEN + ALLOCATE_COUNT_LEN;
const CONNECTION_INCARNATION_LEN: usize = 16;
const CONNECTION_FATE_CLASS_LEN: usize = 1;
const CONVERSATION_BOUND_LEN: usize = 8;
const CONVERSATION_COUNT_LEN: usize = 8;
const CONVERSATION_ID_LEN: usize = 8;
const OPEN_CONNECTION_FATE_FIXED_LEN: usize = CONNECTION_INCARNATION_LEN
    + CONNECTION_FATE_CLASS_LEN
    + CONVERSATION_BOUND_LEN
    + CONVERSATION_COUNT_LEN;
const COMPLETE_CONNECTION_FATE_LEN: usize = 8;

impl ConnectionFateClass {
    const fn tag(self) -> u8 {
        match self {
            Self::CleanDisconnect => 1,
            Self::ServerShutdown => 2,
            Self::ConnectionLost => 3,
            Self::ProtocolError => 4,
        }
    }

    const fn from_tag(tag: u8) -> Result<Self, IncarnationStreamError> {
        match tag {
            1 => Ok(Self::CleanDisconnect),
            2 => Ok(Self::ServerShutdown),
            3 => Ok(Self::ConnectionLost),
            4 => Ok(Self::ProtocolError),
            unknown => Err(IncarnationStreamError::ConnectionFateClass(unknown)),
        }
    }
}

#[derive(Debug, PartialEq, Eq)]
pub(super) enum IncarnationEvent {
    Startup,
    Allocate {
        declared_reference_bound: usize,
        referenced_incarnations: Vec<ConnectionIncarnation>,
    },
    OpenConnectionFate {
        connection_incarnation: ConnectionIncarnation,
        class: ConnectionFateClass,
        declared_conversation_bound: usize,
        conversations: Vec<u64>,
    },
    CompleteConnectionFate {
        open_event_sequence: u64,
    },
}

pub(super) fn encode_event(event: &IncarnationEvent) -> Result<Vec<u8>, IncarnationStreamError> {
    let body_length = match event {
        IncarnationEvent::Startup => 0,
        IncarnationEvent::Allocate {
            referenced_incarnations,
            ..
        } => referenced_incarnations
            .len()
            .checked_mul(CONNECTION_INCARNATION_LEN)
            .and_then(|references| references.checked_add(ALLOCATE_FIXED_LEN))
            .ok_or(IncarnationStreamError::EventLengthOverflow)?,
        IncarnationEvent::OpenConnectionFate {
            declared_conversation_bound,
            conversations,
            ..
        } => {
            validate_conversations(*declared_conversation_bound, conversations)?;
            conversations
                .len()
                .checked_mul(CONVERSATION_ID_LEN)
                .and_then(|ids| ids.checked_add(OPEN_CONNECTION_FATE_FIXED_LEN))
                .ok_or(IncarnationStreamError::EventLengthOverflow)?
        }
        IncarnationEvent::CompleteConnectionFate { .. } => COMPLETE_CONNECTION_FATE_LEN,
    };
    let declared_body_length =
        u32::try_from(body_length).map_err(|_| IncarnationStreamError::EventLengthOverflow)?;
    let total_length = EVENT_HEADER_LEN
        .checked_add(body_length)
        .ok_or(IncarnationStreamError::EventLengthOverflow)?;
    let mut encoded = Vec::with_capacity(total_length);
    encoded.extend_from_slice(&EVENT_MAGIC);
    encoded.push(EVENT_SCHEMA_VERSION);
    encoded.push(match event {
        IncarnationEvent::Startup => STARTUP_TAG,
        IncarnationEvent::Allocate { .. } => ALLOCATE_TAG,
        IncarnationEvent::OpenConnectionFate { .. } => OPEN_CONNECTION_FATE_TAG,
        IncarnationEvent::CompleteConnectionFate { .. } => COMPLETE_CONNECTION_FATE_TAG,
    });
    encoded.extend_from_slice(&declared_body_length.to_be_bytes());
    match event {
        IncarnationEvent::Startup => {}
        IncarnationEvent::Allocate {
            declared_reference_bound,
            referenced_incarnations,
        } => {
            let declared_reference_bound = u64::try_from(*declared_reference_bound)
                .map_err(|_| IncarnationStreamError::EventLengthOverflow)?;
            let count = u64::try_from(referenced_incarnations.len())
                .map_err(|_| IncarnationStreamError::EventLengthOverflow)?;
            encoded.extend_from_slice(&declared_reference_bound.to_be_bytes());
            encoded.extend_from_slice(&count.to_be_bytes());
            for incarnation in referenced_incarnations {
                encoded.extend_from_slice(&incarnation.server_incarnation.to_be_bytes());
                encoded.extend_from_slice(&incarnation.connection_ordinal.to_be_bytes());
            }
        }
        IncarnationEvent::OpenConnectionFate {
            connection_incarnation,
            class,
            declared_conversation_bound,
            conversations,
        } => {
            let declared_conversation_bound = u64::try_from(*declared_conversation_bound)
                .map_err(|_| IncarnationStreamError::EventLengthOverflow)?;
            let count = u64::try_from(conversations.len())
                .map_err(|_| IncarnationStreamError::EventLengthOverflow)?;
            encoded.extend_from_slice(&connection_incarnation.server_incarnation.to_be_bytes());
            encoded.extend_from_slice(&connection_incarnation.connection_ordinal.to_be_bytes());
            encoded.push(class.tag());
            encoded.extend_from_slice(&declared_conversation_bound.to_be_bytes());
            encoded.extend_from_slice(&count.to_be_bytes());
            for conversation_id in conversations {
                encoded.extend_from_slice(&conversation_id.to_be_bytes());
            }
        }
        IncarnationEvent::CompleteConnectionFate {
            open_event_sequence,
        } => {
            encoded.extend_from_slice(&open_event_sequence.to_be_bytes());
        }
    }
    Ok(encoded)
}

pub(super) fn decode_event(bytes: &[u8]) -> Result<IncarnationEvent, IncarnationStreamError> {
    if bytes.len() < EVENT_HEADER_LEN {
        return Err(IncarnationStreamError::EventTruncated {
            required: EVENT_HEADER_LEN,
            actual: bytes.len(),
        });
    }
    if bytes.get(..4) != Some(EVENT_MAGIC.as_slice()) {
        return Err(IncarnationStreamError::EventMagic);
    }
    let version = take_u8(bytes, 4)?;
    if version != FROZEN_EVENT_SCHEMA_VERSION && version != EVENT_SCHEMA_VERSION {
        return Err(IncarnationStreamError::EventSchemaVersion(version));
    }
    let kind = take_u8(bytes, 5)?;
    let declared_body_length = take_u32(bytes, 6)?;
    let body = bytes
        .get(EVENT_HEADER_LEN..)
        .ok_or(IncarnationStreamError::EventTruncated {
            required: EVENT_HEADER_LEN,
            actual: bytes.len(),
        })?;
    if usize::try_from(declared_body_length)
        .map_err(|_| IncarnationStreamError::EventLengthOverflow)?
        != body.len()
    {
        return Err(IncarnationStreamError::EventBodyLength {
            declared: declared_body_length,
            actual: body.len(),
        });
    }
    match kind {
        STARTUP_TAG if body.is_empty() => Ok(IncarnationEvent::Startup),
        STARTUP_TAG => Err(IncarnationStreamError::StartupBodyLength { actual: body.len() }),
        ALLOCATE_TAG => decode_allocate_event(body),
        OPEN_CONNECTION_FATE_TAG | COMPLETE_CONNECTION_FATE_TAG
            if version == FROZEN_EVENT_SCHEMA_VERSION =>
        {
            Err(IncarnationStreamError::EventKindSchemaVersion { version, kind })
        }
        OPEN_CONNECTION_FATE_TAG => decode_open_connection_fate_event(body),
        COMPLETE_CONNECTION_FATE_TAG => decode_complete_connection_fate_event(body),
        unknown => Err(IncarnationStreamError::EventKind(unknown)),
    }
}

fn validate_conversations(
    declared_conversation_bound: usize,
    conversations: &[u64],
) -> Result<(), IncarnationStreamError> {
    if conversations.len() > declared_conversation_bound {
        return Err(IncarnationStreamError::ConversationCountExceedsBound {
            actual: conversations.len(),
            maximum: declared_conversation_bound,
        });
    }
    for (index, pair) in conversations.windows(2).enumerate() {
        if pair[0] >= pair[1] {
            return Err(IncarnationStreamError::ConversationsNotStrictlyIncreasing {
                index: index
                    .checked_add(1)
                    .ok_or(IncarnationStreamError::EventLengthOverflow)?,
            });
        }
    }
    Ok(())
}

fn decode_open_connection_fate_event(
    body: &[u8],
) -> Result<IncarnationEvent, IncarnationStreamError> {
    if body.len() < OPEN_CONNECTION_FATE_FIXED_LEN {
        return Err(IncarnationStreamError::EventTruncated {
            required: EVENT_HEADER_LEN + OPEN_CONNECTION_FATE_FIXED_LEN,
            actual: EVENT_HEADER_LEN + body.len(),
        });
    }
    let connection_incarnation = ConnectionIncarnation::new(take_u64(body, 0)?, take_u64(body, 8)?);
    let class = ConnectionFateClass::from_tag(take_u8(body, CONNECTION_INCARNATION_LEN)?)?;
    let bound_offset = CONNECTION_INCARNATION_LEN + CONNECTION_FATE_CLASS_LEN;
    let bound = take_u64(body, bound_offset)?;
    let declared_conversation_bound = usize::try_from(bound)
        .map_err(|_| IncarnationStreamError::ConversationBoundPlatformOverflow { bound })?;
    let count_offset = bound_offset + CONVERSATION_BOUND_LEN;
    let count = take_u64(body, count_offset)?;
    let count_usize = usize::try_from(count)
        .map_err(|_| IncarnationStreamError::ConversationCountPlatformOverflow { count })?;
    let expected = count_usize
        .checked_mul(CONVERSATION_ID_LEN)
        .and_then(|ids| ids.checked_add(OPEN_CONNECTION_FATE_FIXED_LEN))
        .ok_or(IncarnationStreamError::EventLengthOverflow)?;
    if expected != body.len() {
        return Err(IncarnationStreamError::OpenConnectionFateBodyLength {
            count,
            expected,
            actual: body.len(),
        });
    }
    let mut conversations = Vec::with_capacity(count_usize);
    let mut offset = OPEN_CONNECTION_FATE_FIXED_LEN;
    for _ in 0..count_usize {
        conversations.push(take_u64(body, offset)?);
        offset = offset
            .checked_add(CONVERSATION_ID_LEN)
            .ok_or(IncarnationStreamError::EventLengthOverflow)?;
    }
    validate_conversations(declared_conversation_bound, &conversations)?;
    Ok(IncarnationEvent::OpenConnectionFate {
        connection_incarnation,
        class,
        declared_conversation_bound,
        conversations,
    })
}

fn decode_complete_connection_fate_event(
    body: &[u8],
) -> Result<IncarnationEvent, IncarnationStreamError> {
    if body.len() != COMPLETE_CONNECTION_FATE_LEN {
        return Err(IncarnationStreamError::CompleteConnectionFateBodyLength {
            actual: body.len(),
        });
    }
    Ok(IncarnationEvent::CompleteConnectionFate {
        open_event_sequence: take_u64(body, 0)?,
    })
}

fn decode_allocate_event(body: &[u8]) -> Result<IncarnationEvent, IncarnationStreamError> {
    if body.len() < ALLOCATE_FIXED_LEN {
        return Err(IncarnationStreamError::EventTruncated {
            required: EVENT_HEADER_LEN + ALLOCATE_FIXED_LEN,
            actual: EVENT_HEADER_LEN + body.len(),
        });
    }
    let bound = take_u64(body, 0)?;
    let declared_reference_bound = usize::try_from(bound)
        .map_err(|_| IncarnationStreamError::ReferenceBoundPlatformOverflow { bound })?;
    let count = take_u64(body, ALLOCATE_BOUND_LEN)?;
    let count_usize = usize::try_from(count)
        .map_err(|_| IncarnationStreamError::ReferenceCountPlatformOverflow { count })?;
    if count_usize > declared_reference_bound {
        return Err(IncarnationStreamError::StoredReferenceCountExceedsBound {
            actual: count_usize,
            maximum: declared_reference_bound,
        });
    }
    let expected = count_usize
        .checked_mul(CONNECTION_INCARNATION_LEN)
        .and_then(|references| references.checked_add(ALLOCATE_FIXED_LEN))
        .ok_or(IncarnationStreamError::EventLengthOverflow)?;
    if expected != body.len() {
        return Err(IncarnationStreamError::AllocateBodyLength {
            count,
            expected,
            actual: body.len(),
        });
    }
    let mut referenced_incarnations = Vec::with_capacity(count_usize);
    let mut offset = ALLOCATE_FIXED_LEN;
    for _ in 0..count_usize {
        let server_incarnation = take_u64(body, offset)?;
        offset = offset
            .checked_add(8)
            .ok_or(IncarnationStreamError::EventLengthOverflow)?;
        let connection_ordinal = take_u64(body, offset)?;
        offset = offset
            .checked_add(8)
            .ok_or(IncarnationStreamError::EventLengthOverflow)?;
        referenced_incarnations.push(ConnectionIncarnation::new(
            server_incarnation,
            connection_ordinal,
        ));
    }
    Ok(IncarnationEvent::Allocate {
        declared_reference_bound,
        referenced_incarnations,
    })
}

fn take_u8(bytes: &[u8], offset: usize) -> Result<u8, IncarnationStreamError> {
    let required = offset
        .checked_add(1)
        .ok_or(IncarnationStreamError::EventLengthOverflow)?;
    bytes
        .get(offset)
        .copied()
        .ok_or(IncarnationStreamError::EventTruncated {
            required,
            actual: bytes.len(),
        })
}

fn take_u32(bytes: &[u8], offset: usize) -> Result<u32, IncarnationStreamError> {
    let end = offset
        .checked_add(4)
        .ok_or(IncarnationStreamError::EventLengthOverflow)?;
    let encoded: [u8; 4] = bytes
        .get(offset..end)
        .ok_or(IncarnationStreamError::EventTruncated {
            required: end,
            actual: bytes.len(),
        })?
        .try_into()
        .map_err(|_| IncarnationStreamError::EventLengthOverflow)?;
    Ok(u32::from_be_bytes(encoded))
}

fn take_u64(bytes: &[u8], offset: usize) -> Result<u64, IncarnationStreamError> {
    let end = offset
        .checked_add(8)
        .ok_or(IncarnationStreamError::EventLengthOverflow)?;
    let encoded: [u8; 8] = bytes
        .get(offset..end)
        .ok_or(IncarnationStreamError::EventTruncated {
            required: end,
            actual: bytes.len(),
        })?
        .try_into()
        .map_err(|_| IncarnationStreamError::EventLengthOverflow)?;
    Ok(u64::from_be_bytes(encoded))
}

#[cfg(test)]
pub(in crate::server) fn encode_startup_event_fixture() -> Result<Vec<u8>, IncarnationStreamError> {
    encode_event(&IncarnationEvent::Startup)
}

#[cfg(test)]
pub(in crate::server) fn encode_frozen_v1_startup_event_fixture()
-> Result<Vec<u8>, IncarnationStreamError> {
    let mut encoded = encode_startup_event_fixture()?;
    encoded[4] = FROZEN_EVENT_SCHEMA_VERSION;
    Ok(encoded)
}

#[cfg(test)]
pub(in crate::server) fn encode_allocate_event_fixture(
    declared_reference_bound: usize,
    referenced_incarnations: &[ConnectionIncarnation],
) -> Result<Vec<u8>, IncarnationStreamError> {
    encode_event(&IncarnationEvent::Allocate {
        declared_reference_bound,
        referenced_incarnations: referenced_incarnations.to_vec(),
    })
}

#[cfg(test)]
pub(in crate::server) fn encode_open_connection_fate_event_fixture(
    connection_incarnation: ConnectionIncarnation,
    class: ConnectionFateClass,
    declared_conversation_bound: usize,
    conversations: &[u64],
) -> Result<Vec<u8>, IncarnationStreamError> {
    encode_event(&IncarnationEvent::OpenConnectionFate {
        connection_incarnation,
        class,
        declared_conversation_bound,
        conversations: conversations.to_vec(),
    })
}

#[cfg(test)]
pub(in crate::server) fn encode_complete_connection_fate_event_fixture(
    open_event_sequence: u64,
) -> Result<Vec<u8>, IncarnationStreamError> {
    encode_event(&IncarnationEvent::CompleteConnectionFate {
        open_event_sequence,
    })
}