#![forbid(unsafe_code)]
use sectorsync_core::prelude::{
BarrierId, BarrierState, ClientId, CommandEnvelope, CommandId, CommandPriority, ComponentId,
ComponentStore, EntityHandle, EntityId, EventId, EventKind, EventPriority, OwnerEpoch,
ReplicationPlan, Station, StationEvent, StationId, Tick,
};
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum FrameKind {
Replication = 0,
CommandAck = 1,
Barrier = 2,
Command = 3,
StationEvent = 4,
CommandDispatch = 5,
}
impl FrameKind {
pub const fn from_byte(byte: u8) -> Option<Self> {
match byte {
0 => Some(Self::Replication),
1 => Some(Self::CommandAck),
2 => Some(Self::Barrier),
3 => Some(Self::Command),
4 => Some(Self::StationEvent),
5 => Some(Self::CommandDispatch),
_ => None,
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ReplicationFrame {
pub client_id: ClientId,
pub server_tick: Tick,
pub entity_count: u32,
pub estimated_payload_bytes: u32,
pub entities: Vec<EntityDelta>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct EntityDelta {
pub entity_id: EntityId,
pub owner_epoch: OwnerEpoch,
pub components: Vec<ComponentDelta>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ComponentDelta {
pub component_id: ComponentId,
pub version: u64,
pub flags: u8,
pub bytes: Vec<u8>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ReplicationFrameRef<'a> {
pub client_id: ClientId,
pub server_tick: Tick,
pub entity_count: u32,
pub estimated_payload_bytes: u32,
encoded_entities: &'a [u8],
encoded_entity_count: usize,
}
impl<'a> ReplicationFrameRef<'a> {
pub const fn encoded_entity_count(self) -> usize {
self.encoded_entity_count
}
pub fn entities(self) -> EntityDeltaRefIter<'a> {
EntityDeltaRefIter {
cursor: Cursor::new(self.encoded_entities),
remaining: self.encoded_entity_count,
}
}
pub fn to_owned(self) -> ReplicationFrame {
ReplicationFrame {
client_id: self.client_id,
server_tick: self.server_tick,
entity_count: self.entity_count,
estimated_payload_bytes: self.estimated_payload_bytes,
entities: self.entities().map(EntityDeltaRef::to_owned).collect(),
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct EntityDeltaRef<'a> {
pub entity_id: EntityId,
pub owner_epoch: OwnerEpoch,
encoded_components: &'a [u8],
encoded_component_count: usize,
}
impl<'a> EntityDeltaRef<'a> {
pub const fn encoded_component_count(self) -> usize {
self.encoded_component_count
}
pub fn components(self) -> ComponentDeltaRefIter<'a> {
ComponentDeltaRefIter {
cursor: Cursor::new(self.encoded_components),
remaining: self.encoded_component_count,
}
}
pub fn to_owned(self) -> EntityDelta {
EntityDelta {
entity_id: self.entity_id,
owner_epoch: self.owner_epoch,
components: self.components().map(ComponentDeltaRef::to_owned).collect(),
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ComponentDeltaRef<'a> {
pub component_id: ComponentId,
pub version: u64,
pub flags: u8,
pub bytes: &'a [u8],
}
impl ComponentDeltaRef<'_> {
pub fn to_owned(self) -> ComponentDelta {
ComponentDelta {
component_id: self.component_id,
version: self.version,
flags: self.flags,
bytes: self.bytes.to_vec(),
}
}
}
#[derive(Clone, Debug)]
pub struct EntityDeltaRefIter<'a> {
cursor: Cursor<'a>,
remaining: usize,
}
impl<'a> Iterator for EntityDeltaRefIter<'a> {
type Item = EntityDeltaRef<'a>;
fn next(&mut self) -> Option<Self::Item> {
if self.remaining == 0 {
return None;
}
let entity = decode_entity_ref(&mut self.cursor)
.expect("ReplicationFrameRef validates all entity bytes before iteration");
self.remaining -= 1;
Some(entity)
}
fn size_hint(&self) -> (usize, Option<usize>) {
(self.remaining, Some(self.remaining))
}
}
impl ExactSizeIterator for EntityDeltaRefIter<'_> {}
#[derive(Clone, Debug)]
pub struct ComponentDeltaRefIter<'a> {
cursor: Cursor<'a>,
remaining: usize,
}
impl<'a> Iterator for ComponentDeltaRefIter<'a> {
type Item = ComponentDeltaRef<'a>;
fn next(&mut self) -> Option<Self::Item> {
if self.remaining == 0 {
return None;
}
let component = decode_component_ref(&mut self.cursor)
.expect("ReplicationFrameRef validates all component bytes before iteration");
self.remaining -= 1;
Some(component)
}
fn size_hint(&self) -> (usize, Option<usize>) {
(self.remaining, Some(self.remaining))
}
}
impl ExactSizeIterator for ComponentDeltaRefIter<'_> {}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ReplicationFrameLimits {
pub max_entity_deltas: usize,
pub max_components_per_entity: usize,
pub max_component_bytes: usize,
}
impl Default for ReplicationFrameLimits {
fn default() -> Self {
Self {
max_entity_deltas: 256,
max_components_per_entity: 16,
max_component_bytes: 1024,
}
}
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct ComponentSelection {
pub component_ids: Vec<ComponentId>,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct ReplicationFrameBuildStats {
pub planned_entities: usize,
pub encoded_entities: usize,
pub encoded_components: usize,
pub skipped_entities_by_limit: usize,
pub skipped_entities_by_frame_bytes: usize,
pub skipped_components_by_limit: usize,
pub skipped_components_by_size: usize,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ReplicationFrameBuild {
pub frame: ReplicationFrame,
pub stats: ReplicationFrameBuildStats,
}
#[derive(Clone, Copy, Debug, Default)]
pub struct ReplicationFrameBuilder {
pub limits: ReplicationFrameLimits,
}
impl ReplicationFrameBuilder {
const BINARY_HEADER_BYTES: usize = 1 + 8 + 8 + 4 + 4 + 4;
const BINARY_ENTITY_METADATA_BYTES: usize = 8 + 8 + 2;
const BINARY_COMPONENT_METADATA_BYTES: usize = 2 + 8 + 1 + 4;
pub const fn new(limits: ReplicationFrameLimits) -> Self {
Self { limits }
}
pub fn binary_capacity_hint(
&self,
plan: &ReplicationPlan,
selection: &ComponentSelection,
) -> usize {
let entities = plan.entities.len().min(self.limits.max_entity_deltas);
let components = selection
.component_ids
.len()
.min(self.limits.max_components_per_entity);
let fixed_bytes =
Self::BINARY_HEADER_BYTES.saturating_add(entities.saturating_mul(
Self::BINARY_ENTITY_METADATA_BYTES.saturating_add(
components.saturating_mul(Self::BINARY_COMPONENT_METADATA_BYTES),
),
));
let maximum_payload_bytes = entities
.saturating_mul(components)
.saturating_mul(self.limits.max_component_bytes);
fixed_bytes.saturating_add(plan.stats.estimated_bytes.min(maximum_payload_bytes))
}
pub fn sampled_binary_capacity_hint(
&self,
station: &Station,
plan: &ReplicationPlan,
components: &ComponentStore,
selection: &ComponentSelection,
) -> usize {
const MAX_SAMPLES: usize = 4;
let candidates = plan.entities.len();
let samples = candidates.min(MAX_SAMPLES);
if samples == 0 {
return 0;
}
let denominator = samples.saturating_mul(2);
let mut sampled_bytes = 0_usize;
for sample in 0..samples {
let numerator = sample.saturating_mul(2).saturating_add(1);
let index = numerator.saturating_mul(candidates) / denominator;
let entity_bytes = self.binary_entity_bytes(
station,
plan.entities[index.min(candidates - 1)],
components,
selection,
);
if entity_bytes == 0 {
return 0;
}
sampled_bytes = sampled_bytes.saturating_add(entity_bytes);
}
let estimated_entity_bytes = sampled_bytes.saturating_mul(candidates).div_ceil(samples);
let maximum_entities = candidates.min(self.limits.max_entity_deltas);
let maximum_components = selection
.component_ids
.len()
.min(self.limits.max_components_per_entity);
let maximum_entity_bytes = Self::BINARY_ENTITY_METADATA_BYTES.saturating_add(
maximum_components.saturating_mul(
Self::BINARY_COMPONENT_METADATA_BYTES
.saturating_add(self.limits.max_component_bytes),
),
);
Self::BINARY_HEADER_BYTES.saturating_add(
estimated_entity_bytes.min(maximum_entities.saturating_mul(maximum_entity_bytes)),
)
}
fn binary_entity_bytes(
&self,
station: &Station,
handle: EntityHandle,
components: &ComponentStore,
selection: &ComponentSelection,
) -> usize {
if station.get(handle).is_none() {
return 0;
}
let mut bytes = Self::BINARY_ENTITY_METADATA_BYTES;
let mut encoded_components = 0_usize;
for component_id in &selection.component_ids {
if encoded_components >= self.limits.max_components_per_entity {
break;
}
let Some(blob) = components.get_blob(*component_id, handle) else {
continue;
};
if !blob.dirty || blob.bytes.len() > self.limits.max_component_bytes {
continue;
}
bytes = bytes
.saturating_add(Self::BINARY_COMPONENT_METADATA_BYTES)
.saturating_add(blob.bytes.len());
encoded_components += 1;
}
if encoded_components == 0 { 0 } else { bytes }
}
pub fn build(
&self,
client_id: ClientId,
server_tick: Tick,
station: &Station,
plan: &ReplicationPlan,
components: &ComponentStore,
selection: &ComponentSelection,
) -> ReplicationFrameBuild {
let mut stats = ReplicationFrameBuildStats {
planned_entities: plan.entities.len(),
..ReplicationFrameBuildStats::default()
};
let mut entity_deltas =
Vec::with_capacity(plan.entities.len().min(self.limits.max_entity_deltas));
let mut estimated_payload_bytes = 0_usize;
for handle in &plan.entities {
if entity_deltas.len() >= self.limits.max_entity_deltas {
stats.skipped_entities_by_limit += 1;
continue;
}
let Some(entity) = station.get(*handle) else {
continue;
};
let mut component_deltas = Vec::new();
for component_id in &selection.component_ids {
if component_deltas.len() >= self.limits.max_components_per_entity {
stats.skipped_components_by_limit += 1;
continue;
}
let Some(blob) = components.get_blob(*component_id, *handle) else {
continue;
};
if !blob.dirty {
continue;
}
if blob.bytes.len() > self.limits.max_component_bytes {
stats.skipped_components_by_size += 1;
continue;
}
estimated_payload_bytes = estimated_payload_bytes
.saturating_add(2 + 8 + 1 + 4)
.saturating_add(blob.bytes.len());
component_deltas.push(ComponentDelta {
component_id: *component_id,
version: blob.version,
flags: 0,
bytes: blob.bytes.clone(),
});
}
if component_deltas.is_empty() {
continue;
}
stats.encoded_components += component_deltas.len();
estimated_payload_bytes = estimated_payload_bytes.saturating_add(8 + 8 + 2);
entity_deltas.push(EntityDelta {
entity_id: entity.id,
owner_epoch: entity.role.owner_epoch(),
components: component_deltas,
});
}
stats.encoded_entities = entity_deltas.len();
ReplicationFrameBuild {
frame: ReplicationFrame {
client_id,
server_tick,
entity_count: u32::try_from(plan.entities.len()).unwrap_or(u32::MAX),
estimated_payload_bytes: u32::try_from(estimated_payload_bytes).unwrap_or(u32::MAX),
entities: entity_deltas,
},
stats,
}
}
#[allow(clippy::too_many_arguments)]
pub fn encode_binary_into(
&self,
client_id: ClientId,
server_tick: Tick,
station: &Station,
plan: &ReplicationPlan,
components: &ComponentStore,
selection: &ComponentSelection,
out: &mut Vec<u8>,
) -> Result<ReplicationFrameBuildStats, BinaryEncodeError> {
self.encode_binary_bounded_into(
client_id,
server_tick,
station,
plan,
components,
selection,
usize::MAX,
out,
)
}
#[allow(clippy::too_many_arguments)]
pub fn encode_binary_bounded_into(
&self,
client_id: ClientId,
server_tick: Tick,
station: &Station,
plan: &ReplicationPlan,
components: &ComponentStore,
selection: &ComponentSelection,
max_frame_bytes: usize,
out: &mut Vec<u8>,
) -> Result<ReplicationFrameBuildStats, BinaryEncodeError> {
if max_frame_bytes < Self::BINARY_HEADER_BYTES {
return Err(BinaryEncodeError::FrameBudgetTooSmall {
budget: max_frame_bytes,
required: Self::BINARY_HEADER_BYTES,
});
}
let frame_start = out.len();
let mut stats = ReplicationFrameBuildStats {
planned_entities: plan.entities.len(),
..ReplicationFrameBuildStats::default()
};
let mut estimated_payload_bytes = 0_usize;
out.push(FrameKind::Replication as u8);
out.extend_from_slice(&client_id.get().to_le_bytes());
out.extend_from_slice(&server_tick.get().to_le_bytes());
out.extend_from_slice(
&u32::try_from(plan.entities.len())
.unwrap_or(u32::MAX)
.to_le_bytes(),
);
let estimated_payload_offset = out.len();
out.extend_from_slice(&0_u32.to_le_bytes());
let entity_count_offset = out.len();
out.extend_from_slice(&0_u32.to_le_bytes());
for handle in &plan.entities {
if stats.encoded_entities >= self.limits.max_entity_deltas {
stats.skipped_entities_by_limit += 1;
continue;
}
let Some(entity) = station.get(*handle) else {
continue;
};
let entity_start = out.len();
let estimated_before_entity = estimated_payload_bytes;
out.extend_from_slice(&entity.id.get().to_le_bytes());
out.extend_from_slice(&entity.role.owner_epoch().get().to_le_bytes());
let component_count_offset = out.len();
out.extend_from_slice(&0_u16.to_le_bytes());
let mut component_count = 0_usize;
for component_id in &selection.component_ids {
if component_count >= self.limits.max_components_per_entity {
stats.skipped_components_by_limit += 1;
continue;
}
let Some(blob) = components.get_blob(*component_id, *handle) else {
continue;
};
if !blob.dirty {
continue;
}
if blob.bytes.len() > self.limits.max_component_bytes {
stats.skipped_components_by_size += 1;
continue;
}
out.extend_from_slice(&component_id.get().to_le_bytes());
out.extend_from_slice(&blob.version.to_le_bytes());
out.push(0);
write_bytes("replication.component.bytes", &blob.bytes, out)?;
estimated_payload_bytes = estimated_payload_bytes
.saturating_add(2 + 8 + 1 + 4)
.saturating_add(blob.bytes.len());
component_count += 1;
}
if component_count == 0 {
out.truncate(entity_start);
continue;
}
if out.len().saturating_sub(frame_start) > max_frame_bytes {
out.truncate(entity_start);
estimated_payload_bytes = estimated_before_entity;
stats.skipped_entities_by_frame_bytes =
stats.skipped_entities_by_frame_bytes.saturating_add(1);
break;
}
let component_count =
u16::try_from(component_count).map_err(|_| BinaryEncodeError::TooManyItems {
field: "replication.entity.components",
actual: component_count,
})?;
out[component_count_offset..component_count_offset + 2]
.copy_from_slice(&component_count.to_le_bytes());
stats.encoded_components += usize::from(component_count);
stats.encoded_entities += 1;
estimated_payload_bytes = estimated_payload_bytes.saturating_add(8 + 8 + 2);
}
let encoded_entities =
u32::try_from(stats.encoded_entities).map_err(|_| BinaryEncodeError::TooManyItems {
field: "replication.entities",
actual: stats.encoded_entities,
})?;
out[estimated_payload_offset..estimated_payload_offset + 4].copy_from_slice(
&u32::try_from(estimated_payload_bytes)
.unwrap_or(u32::MAX)
.to_le_bytes(),
);
out[entity_count_offset..entity_count_offset + 4]
.copy_from_slice(&encoded_entities.to_le_bytes());
Ok(stats)
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CommandAckFrame {
pub client_id: ClientId,
pub command_id: CommandId,
pub server_tick: Tick,
pub accepted: bool,
pub reason_code: u16,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CommandFrame {
pub client_id: ClientId,
pub command_id: CommandId,
pub entity_id: EntityId,
pub sequence: u64,
pub kind: u32,
pub priority: CommandPriority,
pub payload: Vec<u8>,
}
impl CommandFrame {
pub fn into_envelope(self, received_at: Tick) -> CommandEnvelope {
CommandEnvelope {
id: self.command_id,
client_id: self.client_id,
entity_id: self.entity_id,
sequence: self.sequence,
received_at,
kind: self.kind,
priority: self.priority,
payload: self.payload,
}
}
pub fn from_envelope(envelope: &CommandEnvelope) -> Self {
Self {
client_id: envelope.client_id,
command_id: envelope.id,
entity_id: envelope.entity_id,
sequence: envelope.sequence,
kind: envelope.kind,
priority: envelope.priority,
payload: envelope.payload.clone(),
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CommandDispatchFrame {
pub station_id: StationId,
pub client_id: ClientId,
pub command_id: CommandId,
pub entity_id: EntityId,
pub sequence: u64,
pub received_at: Tick,
pub kind: u32,
pub priority: CommandPriority,
pub payload: Vec<u8>,
}
impl CommandDispatchFrame {
pub fn from_envelope(station_id: StationId, envelope: &CommandEnvelope) -> Self {
Self {
station_id,
client_id: envelope.client_id,
command_id: envelope.id,
entity_id: envelope.entity_id,
sequence: envelope.sequence,
received_at: envelope.received_at,
kind: envelope.kind,
priority: envelope.priority,
payload: envelope.payload.clone(),
}
}
pub fn into_envelope(self) -> CommandEnvelope {
CommandEnvelope {
id: self.command_id,
client_id: self.client_id,
entity_id: self.entity_id,
sequence: self.sequence,
received_at: self.received_at,
kind: self.kind,
priority: self.priority,
payload: self.payload,
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StationEventFrame {
pub event_id: EventId,
pub source_station: StationId,
pub target_station: StationId,
pub source_tick: Tick,
pub target_tick: Tick,
pub priority: EventPriority,
pub kind: EventKind,
}
impl StationEventFrame {
pub fn from_event(event: &StationEvent) -> Self {
Self {
event_id: event.id,
source_station: event.source,
target_station: event.target,
source_tick: event.source_tick,
target_tick: event.target_tick,
priority: event.priority,
kind: event.kind.clone(),
}
}
pub fn into_event(self) -> StationEvent {
StationEvent {
id: self.event_id,
source: self.source_station,
target: self.target_station,
source_tick: self.source_tick,
target_tick: self.target_tick,
priority: self.priority,
kind: self.kind,
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct BarrierFrame {
pub client_id: ClientId,
pub barrier_id: BarrierId,
pub server_tick: Tick,
pub state: BarrierState,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum RuntimeFrame {
Replication(ReplicationFrame),
Command(CommandFrame),
CommandDispatch(CommandDispatchFrame),
CommandAck(CommandAckFrame),
Barrier(BarrierFrame),
StationEvent(StationEventFrame),
}
pub trait FrameEncoder {
type Error;
fn encode_replication(
&mut self,
frame: &ReplicationFrame,
out: &mut Vec<u8>,
) -> Result<(), Self::Error>;
fn encode_command_ack(
&mut self,
frame: &CommandAckFrame,
out: &mut Vec<u8>,
) -> Result<(), Self::Error>;
fn encode_command(
&mut self,
frame: &CommandFrame,
out: &mut Vec<u8>,
) -> Result<(), Self::Error>;
fn encode_command_dispatch(
&mut self,
frame: &CommandDispatchFrame,
out: &mut Vec<u8>,
) -> Result<(), Self::Error>;
fn encode_station_event(
&mut self,
frame: &StationEventFrame,
out: &mut Vec<u8>,
) -> Result<(), Self::Error>;
fn encode_barrier(
&mut self,
frame: &BarrierFrame,
out: &mut Vec<u8>,
) -> Result<(), Self::Error>;
}
pub trait FrameDecoder {
type Error;
fn decode(&mut self, input: &[u8]) -> Result<RuntimeFrame, Self::Error>;
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum BinaryDecodeError {
Empty,
UnknownFrameKind(u8),
Truncated {
needed: usize,
available: usize,
},
InvalidBarrierState(u8),
InvalidCommandPriority(u8),
InvalidEventPriority(u8),
InvalidEventKind(u8),
TrailingBytes(usize),
}
impl core::fmt::Display for BinaryDecodeError {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
match self {
Self::Empty => f.write_str("empty frame"),
Self::UnknownFrameKind(kind) => write!(f, "unknown frame kind {kind}"),
Self::Truncated { needed, available } => {
write!(f, "truncated frame: needed {needed}, available {available}")
}
Self::InvalidBarrierState(state) => write!(f, "invalid barrier state {state}"),
Self::InvalidCommandPriority(priority) => {
write!(f, "invalid command priority {priority}")
}
Self::InvalidEventPriority(priority) => write!(f, "invalid event priority {priority}"),
Self::InvalidEventKind(kind) => write!(f, "invalid event kind {kind}"),
Self::TrailingBytes(bytes) => write!(f, "frame has {bytes} trailing bytes"),
}
}
}
impl std::error::Error for BinaryDecodeError {}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum ReplicationFrameRefDecodeError {
Binary(BinaryDecodeError),
UnexpectedFrameKind(FrameKind),
}
impl core::fmt::Display for ReplicationFrameRefDecodeError {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
match self {
Self::Binary(error) => error.fmt(f),
Self::UnexpectedFrameKind(actual) => {
write!(f, "expected Replication frame, found {actual:?}")
}
}
}
}
impl std::error::Error for ReplicationFrameRefDecodeError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::Binary(error) => Some(error),
Self::UnexpectedFrameKind(_) => None,
}
}
}
impl From<BinaryDecodeError> for ReplicationFrameRefDecodeError {
fn from(value: BinaryDecodeError) -> Self {
Self::Binary(value)
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum BinaryEncodeError {
FrameBudgetTooSmall {
budget: usize,
required: usize,
},
TooManyItems {
field: &'static str,
actual: usize,
},
PayloadTooLarge {
field: &'static str,
actual: usize,
},
}
impl core::fmt::Display for BinaryEncodeError {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
match self {
Self::FrameBudgetTooSmall { budget, required } => write!(
f,
"replication frame budget {budget} bytes is smaller than {required}-byte header"
),
Self::TooManyItems { field, actual } => {
write!(f, "{field} has too many items: {actual}")
}
Self::PayloadTooLarge { field, actual } => {
write!(f, "{field} payload is too large: {actual} bytes")
}
}
}
}
impl std::error::Error for BinaryEncodeError {}
#[derive(Clone, Copy, Debug, Default)]
pub struct BinaryFrameDecoder;
impl BinaryFrameDecoder {
pub fn decode_replication<'a>(
&mut self,
input: &'a [u8],
) -> Result<ReplicationFrameRef<'a>, ReplicationFrameRefDecodeError> {
let mut cursor = Cursor::new(input);
let kind_byte = cursor.read_u8()?;
let kind = FrameKind::from_byte(kind_byte)
.ok_or(BinaryDecodeError::UnknownFrameKind(kind_byte))?;
if kind != FrameKind::Replication {
return Err(ReplicationFrameRefDecodeError::UnexpectedFrameKind(kind));
}
let frame = decode_replication_ref(&mut cursor)?;
cursor.finish()?;
Ok(frame)
}
}
impl FrameDecoder for BinaryFrameDecoder {
type Error = BinaryDecodeError;
fn decode(&mut self, input: &[u8]) -> Result<RuntimeFrame, Self::Error> {
let mut cursor = Cursor::new(input);
let kind_byte = cursor.read_u8()?;
let kind = FrameKind::from_byte(kind_byte)
.ok_or(BinaryDecodeError::UnknownFrameKind(kind_byte))?;
let frame = match kind {
FrameKind::Replication => {
RuntimeFrame::Replication(decode_replication_owned(&mut cursor)?)
}
FrameKind::CommandAck => RuntimeFrame::CommandAck(CommandAckFrame {
client_id: ClientId::new(cursor.read_u64()?),
command_id: CommandId::new(cursor.read_u64()?),
server_tick: Tick::new(cursor.read_u64()?),
accepted: cursor.read_u8()? != 0,
reason_code: cursor.read_u16()?,
}),
FrameKind::Command => RuntimeFrame::Command(CommandFrame {
client_id: ClientId::new(cursor.read_u64()?),
command_id: CommandId::new(cursor.read_u64()?),
entity_id: EntityId::new(cursor.read_u64()?),
sequence: cursor.read_u64()?,
kind: cursor.read_u32()?,
priority: decode_command_priority(cursor.read_u8()?)?,
payload: {
let byte_len = cursor.read_u32()? as usize;
cursor.read_bytes(byte_len)?
},
}),
FrameKind::CommandDispatch => RuntimeFrame::CommandDispatch(CommandDispatchFrame {
station_id: StationId::new(cursor.read_u32()?),
client_id: ClientId::new(cursor.read_u64()?),
command_id: CommandId::new(cursor.read_u64()?),
entity_id: EntityId::new(cursor.read_u64()?),
sequence: cursor.read_u64()?,
received_at: Tick::new(cursor.read_u64()?),
kind: cursor.read_u32()?,
priority: decode_command_priority(cursor.read_u8()?)?,
payload: {
let byte_len = cursor.read_u32()? as usize;
cursor.read_bytes(byte_len)?
},
}),
FrameKind::Barrier => RuntimeFrame::Barrier(BarrierFrame {
client_id: ClientId::new(cursor.read_u64()?),
barrier_id: BarrierId::new(cursor.read_u64()?),
server_tick: Tick::new(cursor.read_u64()?),
state: decode_barrier_state(cursor.read_u8()?)?,
}),
FrameKind::StationEvent => RuntimeFrame::StationEvent(StationEventFrame {
event_id: EventId::new(cursor.read_u64()?),
source_station: StationId::new(cursor.read_u32()?),
target_station: StationId::new(cursor.read_u32()?),
source_tick: Tick::new(cursor.read_u64()?),
target_tick: Tick::new(cursor.read_u64()?),
priority: decode_event_priority(cursor.read_u8()?)?,
kind: decode_event_kind(&mut cursor)?,
}),
};
cursor.finish()?;
Ok(frame)
}
}
#[derive(Clone, Copy, Debug, Default)]
pub struct BinaryFrameEncoder;
impl BinaryFrameEncoder {
pub fn encode_command_dispatch_envelope(
&mut self,
station_id: StationId,
envelope: &CommandEnvelope,
out: &mut Vec<u8>,
) -> Result<(), BinaryEncodeError> {
out.push(FrameKind::CommandDispatch as u8);
out.extend_from_slice(&station_id.get().to_le_bytes());
out.extend_from_slice(&envelope.client_id.get().to_le_bytes());
out.extend_from_slice(&envelope.id.get().to_le_bytes());
out.extend_from_slice(&envelope.entity_id.get().to_le_bytes());
out.extend_from_slice(&envelope.sequence.to_le_bytes());
out.extend_from_slice(&envelope.received_at.get().to_le_bytes());
out.extend_from_slice(&envelope.kind.to_le_bytes());
out.push(encode_command_priority(envelope.priority));
write_bytes("command_dispatch.payload", &envelope.payload, out)
}
}
impl FrameEncoder for BinaryFrameEncoder {
type Error = BinaryEncodeError;
fn encode_replication(
&mut self,
frame: &ReplicationFrame,
out: &mut Vec<u8>,
) -> Result<(), Self::Error> {
out.push(FrameKind::Replication as u8);
out.extend_from_slice(&frame.client_id.get().to_le_bytes());
out.extend_from_slice(&frame.server_tick.get().to_le_bytes());
out.extend_from_slice(&frame.entity_count.to_le_bytes());
out.extend_from_slice(&frame.estimated_payload_bytes.to_le_bytes());
write_len_u32("replication.entities", frame.entities.len(), out)?;
for entity in &frame.entities {
out.extend_from_slice(&entity.entity_id.get().to_le_bytes());
out.extend_from_slice(&entity.owner_epoch.get().to_le_bytes());
write_len_u16(
"replication.entity.components",
entity.components.len(),
out,
)?;
for component in &entity.components {
out.extend_from_slice(&component.component_id.get().to_le_bytes());
out.extend_from_slice(&component.version.to_le_bytes());
out.push(component.flags);
write_bytes("replication.component.bytes", &component.bytes, out)?;
}
}
Ok(())
}
fn encode_command_ack(
&mut self,
frame: &CommandAckFrame,
out: &mut Vec<u8>,
) -> Result<(), Self::Error> {
out.push(FrameKind::CommandAck as u8);
out.extend_from_slice(&frame.client_id.get().to_le_bytes());
out.extend_from_slice(&frame.command_id.get().to_le_bytes());
out.extend_from_slice(&frame.server_tick.get().to_le_bytes());
out.push(u8::from(frame.accepted));
out.extend_from_slice(&frame.reason_code.to_le_bytes());
Ok(())
}
fn encode_command(
&mut self,
frame: &CommandFrame,
out: &mut Vec<u8>,
) -> Result<(), Self::Error> {
out.push(FrameKind::Command as u8);
out.extend_from_slice(&frame.client_id.get().to_le_bytes());
out.extend_from_slice(&frame.command_id.get().to_le_bytes());
out.extend_from_slice(&frame.entity_id.get().to_le_bytes());
out.extend_from_slice(&frame.sequence.to_le_bytes());
out.extend_from_slice(&frame.kind.to_le_bytes());
out.push(encode_command_priority(frame.priority));
write_bytes("command.payload", &frame.payload, out)?;
Ok(())
}
fn encode_command_dispatch(
&mut self,
frame: &CommandDispatchFrame,
out: &mut Vec<u8>,
) -> Result<(), Self::Error> {
out.push(FrameKind::CommandDispatch as u8);
out.extend_from_slice(&frame.station_id.get().to_le_bytes());
out.extend_from_slice(&frame.client_id.get().to_le_bytes());
out.extend_from_slice(&frame.command_id.get().to_le_bytes());
out.extend_from_slice(&frame.entity_id.get().to_le_bytes());
out.extend_from_slice(&frame.sequence.to_le_bytes());
out.extend_from_slice(&frame.received_at.get().to_le_bytes());
out.extend_from_slice(&frame.kind.to_le_bytes());
out.push(encode_command_priority(frame.priority));
write_bytes("command_dispatch.payload", &frame.payload, out)
}
fn encode_station_event(
&mut self,
frame: &StationEventFrame,
out: &mut Vec<u8>,
) -> Result<(), Self::Error> {
out.push(FrameKind::StationEvent as u8);
out.extend_from_slice(&frame.event_id.get().to_le_bytes());
out.extend_from_slice(&frame.source_station.get().to_le_bytes());
out.extend_from_slice(&frame.target_station.get().to_le_bytes());
out.extend_from_slice(&frame.source_tick.get().to_le_bytes());
out.extend_from_slice(&frame.target_tick.get().to_le_bytes());
out.push(encode_event_priority(frame.priority));
encode_event_kind(&frame.kind, out);
Ok(())
}
fn encode_barrier(
&mut self,
frame: &BarrierFrame,
out: &mut Vec<u8>,
) -> Result<(), Self::Error> {
out.push(FrameKind::Barrier as u8);
out.extend_from_slice(&frame.client_id.get().to_le_bytes());
out.extend_from_slice(&frame.barrier_id.get().to_le_bytes());
out.extend_from_slice(&frame.server_tick.get().to_le_bytes());
out.push(encode_barrier_state(frame.state));
Ok(())
}
}
fn write_len_u16(
field: &'static str,
len: usize,
out: &mut Vec<u8>,
) -> Result<(), BinaryEncodeError> {
let len =
u16::try_from(len).map_err(|_| BinaryEncodeError::TooManyItems { field, actual: len })?;
out.extend_from_slice(&len.to_le_bytes());
Ok(())
}
fn write_len_u32(
field: &'static str,
len: usize,
out: &mut Vec<u8>,
) -> Result<(), BinaryEncodeError> {
let len =
u32::try_from(len).map_err(|_| BinaryEncodeError::TooManyItems { field, actual: len })?;
out.extend_from_slice(&len.to_le_bytes());
Ok(())
}
fn write_bytes(
field: &'static str,
bytes: &[u8],
out: &mut Vec<u8>,
) -> Result<(), BinaryEncodeError> {
let len = u32::try_from(bytes.len()).map_err(|_| BinaryEncodeError::PayloadTooLarge {
field,
actual: bytes.len(),
})?;
out.extend_from_slice(&len.to_le_bytes());
out.extend_from_slice(bytes);
Ok(())
}
fn encode_barrier_state(state: BarrierState) -> u8 {
match state {
BarrierState::Running => 0,
BarrierState::Requested => 1,
BarrierState::WaitingTickBoundary => 2,
BarrierState::Frozen => 3,
BarrierState::Resuming => 4,
}
}
fn decode_barrier_state(state: u8) -> Result<BarrierState, BinaryDecodeError> {
match state {
0 => Ok(BarrierState::Running),
1 => Ok(BarrierState::Requested),
2 => Ok(BarrierState::WaitingTickBoundary),
3 => Ok(BarrierState::Frozen),
4 => Ok(BarrierState::Resuming),
_ => Err(BinaryDecodeError::InvalidBarrierState(state)),
}
}
fn encode_command_priority(priority: CommandPriority) -> u8 {
match priority {
CommandPriority::Normal => 0,
CommandPriority::High => 1,
CommandPriority::Low => 2,
}
}
fn decode_command_priority(priority: u8) -> Result<CommandPriority, BinaryDecodeError> {
match priority {
0 => Ok(CommandPriority::Normal),
1 => Ok(CommandPriority::High),
2 => Ok(CommandPriority::Low),
_ => Err(BinaryDecodeError::InvalidCommandPriority(priority)),
}
}
fn encode_event_priority(priority: EventPriority) -> u8 {
match priority {
EventPriority::Critical => 0,
EventPriority::Important => 1,
EventPriority::BestEffort => 2,
}
}
fn decode_event_priority(priority: u8) -> Result<EventPriority, BinaryDecodeError> {
match priority {
0 => Ok(EventPriority::Critical),
1 => Ok(EventPriority::Important),
2 => Ok(EventPriority::BestEffort),
_ => Err(BinaryDecodeError::InvalidEventPriority(priority)),
}
}
fn encode_event_kind(kind: &EventKind, out: &mut Vec<u8>) {
match kind {
EventKind::Custom(kind) => {
out.push(0);
out.extend_from_slice(&kind.to_le_bytes());
}
EventKind::HandoffPrepare { entity_id } => {
out.push(1);
out.extend_from_slice(&entity_id.get().to_le_bytes());
}
EventKind::HandoffCommit {
entity_id,
owner_epoch,
} => {
out.push(2);
out.extend_from_slice(&entity_id.get().to_le_bytes());
out.extend_from_slice(&owner_epoch.get().to_le_bytes());
}
}
}
fn decode_event_kind(cursor: &mut Cursor<'_>) -> Result<EventKind, BinaryDecodeError> {
match cursor.read_u8()? {
0 => Ok(EventKind::Custom(cursor.read_u32()?)),
1 => Ok(EventKind::HandoffPrepare {
entity_id: EntityId::new(cursor.read_u64()?),
}),
2 => Ok(EventKind::HandoffCommit {
entity_id: EntityId::new(cursor.read_u64()?),
owner_epoch: OwnerEpoch::new(cursor.read_u64()?),
}),
kind => Err(BinaryDecodeError::InvalidEventKind(kind)),
}
}
fn decode_replication_ref<'a>(
cursor: &mut Cursor<'a>,
) -> Result<ReplicationFrameRef<'a>, BinaryDecodeError> {
let client_id = ClientId::new(cursor.read_u64()?);
let server_tick = Tick::new(cursor.read_u64()?);
let entity_count = cursor.read_u32()?;
let estimated_payload_bytes = cursor.read_u32()?;
let encoded_entity_count = cursor.read_u32()? as usize;
let entities_start = cursor.offset;
for _ in 0..encoded_entity_count {
decode_entity_ref(cursor)?;
}
let encoded_entities = &cursor.input[entities_start..cursor.offset];
Ok(ReplicationFrameRef {
client_id,
server_tick,
entity_count,
estimated_payload_bytes,
encoded_entities,
encoded_entity_count,
})
}
fn decode_replication_owned(
cursor: &mut Cursor<'_>,
) -> Result<ReplicationFrame, BinaryDecodeError> {
let client_id = ClientId::new(cursor.read_u64()?);
let server_tick = Tick::new(cursor.read_u64()?);
let entity_count = cursor.read_u32()?;
let estimated_payload_bytes = cursor.read_u32()?;
let encoded_entity_count = cursor.read_u32()? as usize;
let mut entities = Vec::with_capacity(encoded_entity_count);
for _ in 0..encoded_entity_count {
let entity_id = EntityId::new(cursor.read_u64()?);
let owner_epoch = OwnerEpoch::new(cursor.read_u64()?);
let encoded_component_count = cursor.read_u16()? as usize;
let mut components = Vec::with_capacity(encoded_component_count);
for _ in 0..encoded_component_count {
components.push(decode_component_ref(cursor)?.to_owned());
}
entities.push(EntityDelta {
entity_id,
owner_epoch,
components,
});
}
Ok(ReplicationFrame {
client_id,
server_tick,
entity_count,
estimated_payload_bytes,
entities,
})
}
fn decode_entity_ref<'a>(cursor: &mut Cursor<'a>) -> Result<EntityDeltaRef<'a>, BinaryDecodeError> {
let entity_id = EntityId::new(cursor.read_u64()?);
let owner_epoch = OwnerEpoch::new(cursor.read_u64()?);
let encoded_component_count = cursor.read_u16()? as usize;
let components_start = cursor.offset;
for _ in 0..encoded_component_count {
decode_component_ref(cursor)?;
}
let encoded_components = &cursor.input[components_start..cursor.offset];
Ok(EntityDeltaRef {
entity_id,
owner_epoch,
encoded_components,
encoded_component_count,
})
}
fn decode_component_ref<'a>(
cursor: &mut Cursor<'a>,
) -> Result<ComponentDeltaRef<'a>, BinaryDecodeError> {
let component_id = ComponentId::new(cursor.read_u16()?);
let version = cursor.read_u64()?;
let flags = cursor.read_u8()?;
let byte_len = cursor.read_u32()? as usize;
let bytes = cursor.read_slice(byte_len)?;
Ok(ComponentDeltaRef {
component_id,
version,
flags,
bytes,
})
}
#[derive(Clone, Debug)]
struct Cursor<'a> {
input: &'a [u8],
offset: usize,
}
impl<'a> Cursor<'a> {
fn new(input: &'a [u8]) -> Self {
Self { input, offset: 0 }
}
fn read_u8(&mut self) -> Result<u8, BinaryDecodeError> {
self.require(1)?;
let value = self.input[self.offset];
self.offset += 1;
Ok(value)
}
fn read_u16(&mut self) -> Result<u16, BinaryDecodeError> {
let bytes = self.read_array::<2>()?;
Ok(u16::from_le_bytes(bytes))
}
fn read_u32(&mut self) -> Result<u32, BinaryDecodeError> {
let bytes = self.read_array::<4>()?;
Ok(u32::from_le_bytes(bytes))
}
fn read_u64(&mut self) -> Result<u64, BinaryDecodeError> {
let bytes = self.read_array::<8>()?;
Ok(u64::from_le_bytes(bytes))
}
fn read_array<const N: usize>(&mut self) -> Result<[u8; N], BinaryDecodeError> {
self.require(N)?;
let mut out = [0_u8; N];
out.copy_from_slice(&self.input[self.offset..self.offset + N]);
self.offset += N;
Ok(out)
}
fn read_bytes(&mut self, len: usize) -> Result<Vec<u8>, BinaryDecodeError> {
Ok(self.read_slice(len)?.to_vec())
}
fn read_slice(&mut self, len: usize) -> Result<&'a [u8], BinaryDecodeError> {
self.require(len)?;
let bytes = &self.input[self.offset..self.offset + len];
self.offset += len;
Ok(bytes)
}
fn require(&self, count: usize) -> Result<(), BinaryDecodeError> {
let needed = self.offset.saturating_add(count);
if needed > self.input.len() {
Err(BinaryDecodeError::Truncated {
needed,
available: self.input.len(),
})
} else {
Ok(())
}
}
fn finish(&self) -> Result<(), BinaryDecodeError> {
if self.offset == self.input.len() {
Ok(())
} else {
Err(BinaryDecodeError::TrailingBytes(
self.input.len().saturating_sub(self.offset),
))
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn binary_codec_roundtrips_replication_frame() {
let frame = ReplicationFrame {
client_id: ClientId::new(9),
server_tick: Tick::new(42),
entity_count: 17,
estimated_payload_bytes: 544,
entities: vec![EntityDelta {
entity_id: EntityId::new(100),
owner_epoch: OwnerEpoch::new(2),
components: vec![
ComponentDelta {
component_id: ComponentId::new(1),
version: 9,
flags: 0,
bytes: vec![1, 2, 3, 4],
},
ComponentDelta {
component_id: ComponentId::new(2),
version: 10,
flags: 1,
bytes: vec![5, 6],
},
],
}],
};
let mut encoder = BinaryFrameEncoder;
let mut bytes = Vec::new();
encoder
.encode_replication(&frame, &mut bytes)
.expect("encoder is infallible");
let borrowed = BinaryFrameDecoder
.decode_replication(&bytes)
.expect("borrowed decode should work");
assert_eq!(borrowed.client_id, frame.client_id);
assert_eq!(borrowed.server_tick, frame.server_tick);
assert_eq!(borrowed.encoded_entity_count(), 1);
let mut entities = borrowed.entities();
assert_eq!(entities.len(), 1);
let entity = entities.next().expect("entity should be borrowed");
assert_eq!(entity.entity_id, frame.entities[0].entity_id);
assert_eq!(entity.encoded_component_count(), 2);
let mut components = entity.components();
assert_eq!(components.len(), 2);
let first = components.next().expect("component should be borrowed");
assert_eq!(first.bytes, frame.entities[0].components[0].bytes);
let input_start = bytes.as_ptr() as usize;
let input_end = input_start.saturating_add(bytes.len());
let payload_start = first.bytes.as_ptr() as usize;
assert!(payload_start >= input_start);
assert!(payload_start.saturating_add(first.bytes.len()) <= input_end);
assert_eq!(borrowed.to_owned(), frame);
let decoded = BinaryFrameDecoder
.decode(&bytes)
.expect("decode should work");
assert_eq!(decoded, RuntimeFrame::Replication(frame));
}
#[test]
fn borrowed_replication_decode_rejects_wrong_kind_truncation_and_trailing_bytes() {
assert_eq!(
BinaryFrameDecoder.decode_replication(&[FrameKind::CommandAck as u8]),
Err(ReplicationFrameRefDecodeError::UnexpectedFrameKind(
FrameKind::CommandAck
))
);
let frame = ReplicationFrame {
client_id: ClientId::new(9),
server_tick: Tick::new(42),
entity_count: 1,
estimated_payload_bytes: 1,
entities: vec![EntityDelta {
entity_id: EntityId::new(100),
owner_epoch: OwnerEpoch::new(2),
components: vec![ComponentDelta {
component_id: ComponentId::new(1),
version: 9,
flags: 0,
bytes: vec![1],
}],
}],
};
let mut bytes = Vec::new();
BinaryFrameEncoder
.encode_replication(&frame, &mut bytes)
.expect("frame should encode");
let truncated = &bytes[..bytes.len() - 1];
assert!(matches!(
BinaryFrameDecoder.decode_replication(truncated),
Err(ReplicationFrameRefDecodeError::Binary(
BinaryDecodeError::Truncated { .. }
))
));
bytes.push(0xff);
assert_eq!(
BinaryFrameDecoder.decode_replication(&bytes),
Err(ReplicationFrameRefDecodeError::Binary(
BinaryDecodeError::TrailingBytes(1)
))
);
}
#[test]
fn binary_codec_roundtrips_command_ack_frame() {
let frame = CommandAckFrame {
client_id: ClientId::new(1),
command_id: CommandId::new(2),
server_tick: Tick::new(3),
accepted: false,
reason_code: 7,
};
let mut encoder = BinaryFrameEncoder;
let mut bytes = Vec::new();
encoder
.encode_command_ack(&frame, &mut bytes)
.expect("encoder is infallible");
let decoded = BinaryFrameDecoder
.decode(&bytes)
.expect("decode should work");
assert_eq!(decoded, RuntimeFrame::CommandAck(frame));
}
#[test]
fn binary_codec_roundtrips_command_frame() {
let frame = CommandFrame {
client_id: ClientId::new(1),
command_id: CommandId::new(2),
entity_id: EntityId::new(3),
sequence: 4,
kind: 5,
priority: CommandPriority::High,
payload: vec![9, 8, 7],
};
let mut encoder = BinaryFrameEncoder;
let mut bytes = Vec::new();
encoder
.encode_command(&frame, &mut bytes)
.expect("encoder is infallible");
let decoded = BinaryFrameDecoder
.decode(&bytes)
.expect("decode should work");
assert_eq!(decoded, RuntimeFrame::Command(frame));
}
#[test]
fn binary_codec_roundtrips_command_dispatch_frame() {
let frame = CommandDispatchFrame {
station_id: StationId::new(10),
client_id: ClientId::new(1),
command_id: CommandId::new(2),
entity_id: EntityId::new(3),
sequence: 4,
received_at: Tick::new(99),
kind: 5,
priority: CommandPriority::High,
payload: vec![9, 8, 7],
};
let mut encoder = BinaryFrameEncoder;
let mut bytes = Vec::new();
encoder
.encode_command_dispatch(&frame, &mut bytes)
.expect("encoder is infallible");
let envelope = CommandEnvelope {
id: frame.command_id,
client_id: frame.client_id,
entity_id: frame.entity_id,
sequence: frame.sequence,
received_at: frame.received_at,
kind: frame.kind,
priority: frame.priority,
payload: frame.payload.clone(),
};
let mut direct = Vec::new();
encoder
.encode_command_dispatch_envelope(frame.station_id, &envelope, &mut direct)
.expect("direct encoder is infallible");
assert_eq!(direct, bytes);
let decoded = BinaryFrameDecoder
.decode(&bytes)
.expect("decode should work");
assert_eq!(decoded, RuntimeFrame::CommandDispatch(frame));
}
#[test]
fn command_frame_converts_to_runtime_envelope() {
let frame = CommandFrame {
client_id: ClientId::new(1),
command_id: CommandId::new(2),
entity_id: EntityId::new(3),
sequence: 4,
kind: 5,
priority: CommandPriority::Low,
payload: vec![1, 2, 3],
};
let envelope = frame.clone().into_envelope(Tick::new(99));
assert_eq!(envelope.id, frame.command_id);
assert_eq!(envelope.received_at, Tick::new(99));
assert_eq!(CommandFrame::from_envelope(&envelope), frame);
}
#[test]
fn command_dispatch_frame_preserves_stamped_envelope_tick() {
let envelope = CommandEnvelope {
id: CommandId::new(2),
client_id: ClientId::new(1),
entity_id: EntityId::new(3),
sequence: 4,
received_at: Tick::new(99),
kind: 5,
priority: CommandPriority::Low,
payload: vec![1, 2, 3],
};
let frame = CommandDispatchFrame::from_envelope(StationId::new(10), &envelope);
assert_eq!(frame.station_id, StationId::new(10));
assert_eq!(frame.received_at, Tick::new(99));
assert_eq!(frame.into_envelope(), envelope);
}
#[test]
fn binary_codec_roundtrips_barrier_frame() {
let frame = BarrierFrame {
client_id: ClientId::new(1),
barrier_id: BarrierId::new(99),
server_tick: Tick::new(11),
state: BarrierState::Frozen,
};
let mut encoder = BinaryFrameEncoder;
let mut bytes = Vec::new();
encoder
.encode_barrier(&frame, &mut bytes)
.expect("encoder is infallible");
let decoded = BinaryFrameDecoder
.decode(&bytes)
.expect("decode should work");
assert_eq!(decoded, RuntimeFrame::Barrier(frame));
}
#[test]
fn binary_codec_roundtrips_station_event_frame() {
let frames = [
StationEventFrame {
event_id: EventId::new(1),
source_station: StationId::new(10),
target_station: StationId::new(11),
source_tick: Tick::new(2),
target_tick: Tick::new(3),
priority: EventPriority::Critical,
kind: EventKind::Custom(7),
},
StationEventFrame {
event_id: EventId::new(2),
source_station: StationId::new(10),
target_station: StationId::new(11),
source_tick: Tick::new(2),
target_tick: Tick::new(3),
priority: EventPriority::Important,
kind: EventKind::HandoffPrepare {
entity_id: EntityId::new(99),
},
},
StationEventFrame {
event_id: EventId::new(3),
source_station: StationId::new(10),
target_station: StationId::new(11),
source_tick: Tick::new(2),
target_tick: Tick::new(3),
priority: EventPriority::BestEffort,
kind: EventKind::HandoffCommit {
entity_id: EntityId::new(99),
owner_epoch: OwnerEpoch::new(5),
},
},
];
for frame in frames {
let mut encoder = BinaryFrameEncoder;
let mut bytes = Vec::new();
encoder
.encode_station_event(&frame, &mut bytes)
.expect("encoder is infallible");
let decoded = BinaryFrameDecoder
.decode(&bytes)
.expect("decode should work");
assert_eq!(decoded, RuntimeFrame::StationEvent(frame));
}
}
#[test]
fn station_event_frame_converts_to_runtime_event() {
let event = StationEvent {
id: EventId::new(1),
source: StationId::new(10),
target: StationId::new(11),
source_tick: Tick::new(2),
target_tick: Tick::new(3),
priority: EventPriority::Critical,
kind: EventKind::HandoffPrepare {
entity_id: EntityId::new(99),
},
};
let frame = StationEventFrame::from_event(&event);
assert_eq!(frame.event_id, event.id);
assert_eq!(frame.clone().into_event(), event);
}
#[test]
fn frame_builder_materializes_dirty_component_deltas() {
use sectorsync_core::prelude::{
Bounds, ComponentDescriptor, ComponentMigrationMode, ComponentSyncMode, InstanceId,
NodeId, PolicyId, Position3, ReplicationPlan, StationConfig, Vec3, Vec3LeCodec,
};
let mut station = Station::new(StationConfig {
station_id: sectorsync_core::prelude::StationId::new(1),
node_id: NodeId::new(1),
instance_id: InstanceId::new(1),
tick_rate_hz: 20,
});
let handle = station
.spawn_owned(
EntityId::new(10),
Position3::new(0.0, 0.0, 0.0),
Bounds::Point,
PolicyId::new(0),
)
.expect("spawn should work");
let descriptor = ComponentDescriptor::sparse_blob(
ComponentId::new(1),
"velocity",
ComponentSyncMode::Delta,
ComponentMigrationMode::Copy,
12,
);
let mut components = ComponentStore::default();
components
.set_typed(
&descriptor,
handle,
3,
&Vec3LeCodec,
&Vec3::new(1.0, 2.0, 3.0),
)
.expect("typed component should encode");
let builder = ReplicationFrameBuilder::new(ReplicationFrameLimits {
max_entity_deltas: 8,
max_components_per_entity: 4,
max_component_bytes: 64,
});
let plan = ReplicationPlan {
entities: vec![handle],
stats: sectorsync_core::prelude::ReplicationStats::default(),
};
let selection = ComponentSelection {
component_ids: vec![ComponentId::new(1)],
};
let build = builder.build(
ClientId::new(5),
Tick::new(9),
&station,
&plan,
&components,
&selection,
);
assert_eq!(build.stats.encoded_entities, 1);
assert_eq!(build.stats.encoded_components, 1);
assert_eq!(build.frame.entities[0].entity_id, EntityId::new(10));
assert_eq!(build.frame.entities[0].components[0].version, 3);
let mut materialized_bytes = Vec::new();
BinaryFrameEncoder
.encode_replication(&build.frame, &mut materialized_bytes)
.expect("materialized frame should encode");
let mut direct_bytes = Vec::new();
let direct_stats = builder
.encode_binary_into(
ClientId::new(5),
Tick::new(9),
&station,
&plan,
&components,
&selection,
&mut direct_bytes,
)
.expect("plan should encode directly");
assert_eq!(direct_stats, build.stats);
assert_eq!(direct_bytes, materialized_bytes);
assert_eq!(
builder.sampled_binary_capacity_hint(&station, &plan, &components, &selection),
direct_bytes.len()
);
assert_bounded_binary_encoding(
builder,
&station,
&plan,
&components,
&selection,
direct_bytes.len(),
);
}
fn assert_bounded_binary_encoding(
builder: ReplicationFrameBuilder,
station: &Station,
plan: &ReplicationPlan,
components: &ComponentStore,
selection: &ComponentSelection,
unbounded_len: usize,
) {
let mut bounded_bytes = Vec::new();
let bounded = builder
.encode_binary_bounded_into(
ClientId::new(5),
Tick::new(9),
station,
plan,
components,
selection,
unbounded_len - 1,
&mut bounded_bytes,
)
.expect("bounded frame should roll back the oversized entity");
assert_eq!(bounded.encoded_entities, 0);
assert_eq!(bounded.encoded_components, 0);
assert_eq!(bounded.skipped_entities_by_frame_bytes, 1);
assert!(bounded_bytes.len() < unbounded_len);
let RuntimeFrame::Replication(frame) = BinaryFrameDecoder
.decode(&bounded_bytes)
.expect("bounded empty frame should remain valid")
else {
panic!("expected replication frame");
};
assert!(frame.entities.is_empty());
let mut prefix = vec![9, 8, 7];
assert_eq!(
builder.encode_binary_bounded_into(
ClientId::new(5),
Tick::new(9),
station,
plan,
components,
selection,
ReplicationFrameBuilder::BINARY_HEADER_BYTES - 1,
&mut prefix,
),
Err(BinaryEncodeError::FrameBudgetTooSmall {
budget: ReplicationFrameBuilder::BINARY_HEADER_BYTES - 1,
required: ReplicationFrameBuilder::BINARY_HEADER_BYTES,
})
);
assert_eq!(prefix, [9, 8, 7]);
}
#[test]
fn binary_capacity_hint_is_bounded_by_active_builder_limits() {
let builder = ReplicationFrameBuilder::new(ReplicationFrameLimits {
max_entity_deltas: 2,
max_components_per_entity: 2,
max_component_bytes: 32,
});
let plan = ReplicationPlan {
entities: vec![
EntityHandle::new(1, 0),
EntityHandle::new(2, 0),
EntityHandle::new(3, 0),
],
stats: sectorsync_core::prelude::ReplicationStats {
estimated_bytes: usize::MAX,
..sectorsync_core::prelude::ReplicationStats::default()
},
};
let selection = ComponentSelection {
component_ids: vec![
ComponentId::new(1),
ComponentId::new(2),
ComponentId::new(3),
],
};
let fixed_bytes = 29 + 2 * (18 + 2 * 15);
let maximum_payload_bytes = 2 * 2 * 32;
assert_eq!(
builder.binary_capacity_hint(&plan, &selection),
fixed_bytes + maximum_payload_bytes
);
}
#[test]
fn sampled_capacity_hint_falls_back_when_any_sample_has_no_dirty_data() {
use sectorsync_core::prelude::{
Bounds, ComponentDescriptor, ComponentMigrationMode, ComponentSyncMode, InstanceId,
NodeId, PolicyId, Position3, ReplicationStats, StationConfig, U32LeCodec,
};
let mut station = Station::new(StationConfig {
station_id: StationId::new(1),
node_id: NodeId::new(1),
instance_id: InstanceId::new(1),
tick_rate_hz: 20,
});
let descriptor = ComponentDescriptor::sparse_blob(
ComponentId::new(1),
"health",
ComponentSyncMode::Delta,
ComponentMigrationMode::Copy,
4,
);
let mut components = ComponentStore::default();
let mut handles = Vec::new();
for entity in 0_u64..4 {
let handle = station
.spawn_owned(
EntityId::new(entity),
Position3::new(0.0, 0.0, 0.0),
Bounds::Point,
PolicyId::new(1),
)
.expect("entity ids are unique");
if entity < 3 {
components
.set_typed(&descriptor, handle, 1, &U32LeCodec, &100)
.expect("component should fit");
}
handles.push(handle);
}
let plan = ReplicationPlan {
entities: handles,
stats: ReplicationStats {
selected: 4,
estimated_bytes: 128,
..ReplicationStats::default()
},
};
let selection = ComponentSelection {
component_ids: vec![ComponentId::new(1)],
};
assert_eq!(
ReplicationFrameBuilder::default().sampled_binary_capacity_hint(
&station,
&plan,
&components,
&selection,
),
0
);
}
}