sectorsync_wire/
lib.rs

1//! Wire frame traits and default frame shapes for `SectorSync`.
2
3#![forbid(unsafe_code)]
4
5use sectorsync_core::prelude::{
6    BarrierId, BarrierState, ClientId, CommandEnvelope, CommandId, CommandPriority, ComponentId,
7    ComponentStore, EntityId, EventId, EventKind, EventPriority, OwnerEpoch, ReplicationPlan,
8    Station, StationEvent, StationId, Tick,
9};
10
11/// Runtime frame kind.
12#[derive(Clone, Copy, Debug, PartialEq, Eq)]
13pub enum FrameKind {
14    /// Replication update frame.
15    Replication = 0,
16    /// Command acknowledgement frame.
17    CommandAck = 1,
18    /// Runtime barrier notification.
19    Barrier = 2,
20    /// Client command ingress frame.
21    Command = 3,
22    /// Cross-station event frame.
23    StationEvent = 4,
24    /// Gateway-to-station command dispatch frame.
25    CommandDispatch = 5,
26}
27
28impl FrameKind {
29    /// Converts a byte into a frame kind.
30    pub const fn from_byte(byte: u8) -> Option<Self> {
31        match byte {
32            0 => Some(Self::Replication),
33            1 => Some(Self::CommandAck),
34            2 => Some(Self::Barrier),
35            3 => Some(Self::Command),
36            4 => Some(Self::StationEvent),
37            5 => Some(Self::CommandDispatch),
38            _ => None,
39        }
40    }
41}
42
43/// Replication frame metadata produced per client.
44#[derive(Clone, Debug, PartialEq, Eq)]
45pub struct ReplicationFrame {
46    /// Target client.
47    pub client_id: ClientId,
48    /// Server tick represented by this frame.
49    pub server_tick: Tick,
50    /// Number of entity updates in this frame.
51    pub entity_count: u32,
52    /// Estimated payload bytes before transport overhead.
53    pub estimated_payload_bytes: u32,
54    /// Concrete entity/component deltas included in this frame.
55    pub entities: Vec<EntityDelta>,
56}
57
58/// Entity delta included in a replication frame.
59#[derive(Clone, Debug, PartialEq, Eq)]
60pub struct EntityDelta {
61    /// Entity being updated.
62    pub entity_id: EntityId,
63    /// Owner epoch observed by the sender.
64    pub owner_epoch: OwnerEpoch,
65    /// Component deltas for this entity.
66    pub components: Vec<ComponentDelta>,
67}
68
69/// Component delta included in an entity delta.
70#[derive(Clone, Debug, PartialEq, Eq)]
71pub struct ComponentDelta {
72    /// Component id.
73    pub component_id: ComponentId,
74    /// Component version.
75    pub version: u64,
76    /// Runtime-defined flags.
77    pub flags: u8,
78    /// Encoded component bytes.
79    pub bytes: Vec<u8>,
80}
81
82/// Limits used by `ReplicationFrameBuilder`.
83#[derive(Clone, Copy, Debug, PartialEq, Eq)]
84pub struct ReplicationFrameLimits {
85    /// Maximum entity deltas to materialize in one frame.
86    pub max_entity_deltas: usize,
87    /// Maximum component deltas to include per entity.
88    pub max_components_per_entity: usize,
89    /// Maximum component payload bytes to include per component.
90    pub max_component_bytes: usize,
91}
92
93impl Default for ReplicationFrameLimits {
94    fn default() -> Self {
95        Self {
96            max_entity_deltas: 256,
97            max_components_per_entity: 16,
98            max_component_bytes: 1024,
99        }
100    }
101}
102
103/// Component selection for frame building.
104#[derive(Clone, Debug, Default, PartialEq, Eq)]
105pub struct ComponentSelection {
106    /// Component ids to include when present and dirty.
107    pub component_ids: Vec<ComponentId>,
108}
109
110/// Frame builder statistics.
111#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
112pub struct ReplicationFrameBuildStats {
113    /// Entity handles selected by the replication plan.
114    pub planned_entities: usize,
115    /// Entity deltas materialized into the frame.
116    pub encoded_entities: usize,
117    /// Component deltas materialized into the frame.
118    pub encoded_components: usize,
119    /// Entity deltas skipped by builder limits.
120    pub skipped_entities_by_limit: usize,
121    /// Component deltas skipped by builder limits.
122    pub skipped_components_by_limit: usize,
123    /// Component payloads skipped because they exceed byte limits.
124    pub skipped_components_by_size: usize,
125}
126
127/// Result of building a replication frame.
128#[derive(Clone, Debug, PartialEq, Eq)]
129pub struct ReplicationFrameBuild {
130    /// Built frame.
131    pub frame: ReplicationFrame,
132    /// Build statistics.
133    pub stats: ReplicationFrameBuildStats,
134}
135
136/// Builds concrete replication frames from a core replication plan.
137#[derive(Clone, Copy, Debug, Default)]
138pub struct ReplicationFrameBuilder {
139    /// Builder limits.
140    pub limits: ReplicationFrameLimits,
141}
142
143impl ReplicationFrameBuilder {
144    /// Creates a frame builder with explicit limits.
145    pub const fn new(limits: ReplicationFrameLimits) -> Self {
146        Self { limits }
147    }
148
149    /// Builds a frame from a station plan and component store.
150    pub fn build(
151        &self,
152        client_id: ClientId,
153        server_tick: Tick,
154        station: &Station,
155        plan: &ReplicationPlan,
156        components: &ComponentStore,
157        selection: &ComponentSelection,
158    ) -> ReplicationFrameBuild {
159        let mut stats = ReplicationFrameBuildStats {
160            planned_entities: plan.entities.len(),
161            ..ReplicationFrameBuildStats::default()
162        };
163        let mut entity_deltas =
164            Vec::with_capacity(plan.entities.len().min(self.limits.max_entity_deltas));
165        let mut estimated_payload_bytes = 0_usize;
166
167        for handle in &plan.entities {
168            if entity_deltas.len() >= self.limits.max_entity_deltas {
169                stats.skipped_entities_by_limit += 1;
170                continue;
171            }
172            let Some(entity) = station.get(*handle) else {
173                continue;
174            };
175
176            let mut component_deltas = Vec::new();
177            for component_id in &selection.component_ids {
178                if component_deltas.len() >= self.limits.max_components_per_entity {
179                    stats.skipped_components_by_limit += 1;
180                    continue;
181                }
182                let Some(blob) = components.get_blob(*component_id, *handle) else {
183                    continue;
184                };
185                if !blob.dirty {
186                    continue;
187                }
188                if blob.bytes.len() > self.limits.max_component_bytes {
189                    stats.skipped_components_by_size += 1;
190                    continue;
191                }
192                estimated_payload_bytes = estimated_payload_bytes
193                    .saturating_add(2 + 8 + 1 + 4)
194                    .saturating_add(blob.bytes.len());
195                component_deltas.push(ComponentDelta {
196                    component_id: *component_id,
197                    version: blob.version,
198                    flags: 0,
199                    bytes: blob.bytes.clone(),
200                });
201            }
202
203            if component_deltas.is_empty() {
204                continue;
205            }
206
207            stats.encoded_components += component_deltas.len();
208            estimated_payload_bytes = estimated_payload_bytes.saturating_add(8 + 8 + 2);
209            entity_deltas.push(EntityDelta {
210                entity_id: entity.id,
211                owner_epoch: entity.role.owner_epoch(),
212                components: component_deltas,
213            });
214        }
215
216        stats.encoded_entities = entity_deltas.len();
217        ReplicationFrameBuild {
218            frame: ReplicationFrame {
219                client_id,
220                server_tick,
221                entity_count: u32::try_from(plan.entities.len()).unwrap_or(u32::MAX),
222                estimated_payload_bytes: u32::try_from(estimated_payload_bytes).unwrap_or(u32::MAX),
223                entities: entity_deltas,
224            },
225            stats,
226        }
227    }
228}
229
230/// Command acknowledgement frame.
231#[derive(Clone, Debug, PartialEq, Eq)]
232pub struct CommandAckFrame {
233    /// Target client.
234    pub client_id: ClientId,
235    /// Acknowledged command.
236    pub command_id: CommandId,
237    /// Server tick at acknowledgement.
238    pub server_tick: Tick,
239    /// Whether the command was accepted by the runtime pipeline.
240    pub accepted: bool,
241    /// Game/runtime reject reason code.
242    pub reason_code: u16,
243}
244
245/// Client command ingress frame.
246///
247/// The server stamps `received_at` when converting this into a
248/// `CommandEnvelope`; game validation and anti-cheat checks remain outside the
249/// wire codec.
250#[derive(Clone, Debug, PartialEq, Eq)]
251pub struct CommandFrame {
252    /// Client that submitted the command.
253    pub client_id: ClientId,
254    /// Command id used for replay and audit.
255    pub command_id: CommandId,
256    /// Entity the command intends to control.
257    pub entity_id: EntityId,
258    /// Client-side sequence number.
259    pub sequence: u64,
260    /// Game-defined command kind.
261    pub kind: u32,
262    /// Command priority.
263    pub priority: CommandPriority,
264    /// Opaque payload owned by the embedding game.
265    pub payload: Vec<u8>,
266}
267
268impl CommandFrame {
269    /// Converts an ingress frame into a runtime command envelope.
270    pub fn into_envelope(self, received_at: Tick) -> CommandEnvelope {
271        CommandEnvelope {
272            id: self.command_id,
273            client_id: self.client_id,
274            entity_id: self.entity_id,
275            sequence: self.sequence,
276            received_at,
277            kind: self.kind,
278            priority: self.priority,
279            payload: self.payload,
280        }
281    }
282
283    /// Converts a command envelope into a wire frame, dropping server-only tick
284    /// metadata.
285    pub fn from_envelope(envelope: &CommandEnvelope) -> Self {
286        Self {
287            client_id: envelope.client_id,
288            command_id: envelope.id,
289            entity_id: envelope.entity_id,
290            sequence: envelope.sequence,
291            kind: envelope.kind,
292            priority: envelope.priority,
293            payload: envelope.payload.clone(),
294        }
295    }
296}
297
298/// Internal gateway-to-station command dispatch frame.
299///
300/// Unlike `CommandFrame`, this preserves the server `received_at` tick stamped
301/// by the gateway pipeline before the command is forwarded to a station node.
302#[derive(Clone, Debug, PartialEq, Eq)]
303pub struct CommandDispatchFrame {
304    /// Target station selected by gateway/deployment routing.
305    pub station_id: StationId,
306    /// Client that submitted the command.
307    pub client_id: ClientId,
308    /// Command id used for replay and audit.
309    pub command_id: CommandId,
310    /// Entity the command intends to control.
311    pub entity_id: EntityId,
312    /// Client-side sequence number.
313    pub sequence: u64,
314    /// Server tick observed when the command entered `SectorSync`.
315    pub received_at: Tick,
316    /// Game-defined command kind.
317    pub kind: u32,
318    /// Command priority.
319    pub priority: CommandPriority,
320    /// Opaque payload owned by the embedding game.
321    pub payload: Vec<u8>,
322}
323
324impl CommandDispatchFrame {
325    /// Converts a stamped command envelope into an internal dispatch frame.
326    pub fn from_envelope(station_id: StationId, envelope: &CommandEnvelope) -> Self {
327        Self {
328            station_id,
329            client_id: envelope.client_id,
330            command_id: envelope.id,
331            entity_id: envelope.entity_id,
332            sequence: envelope.sequence,
333            received_at: envelope.received_at,
334            kind: envelope.kind,
335            priority: envelope.priority,
336            payload: envelope.payload.clone(),
337        }
338    }
339
340    /// Converts an internal dispatch frame back into a command envelope.
341    pub fn into_envelope(self) -> CommandEnvelope {
342        CommandEnvelope {
343            id: self.command_id,
344            client_id: self.client_id,
345            entity_id: self.entity_id,
346            sequence: self.sequence,
347            received_at: self.received_at,
348            kind: self.kind,
349            priority: self.priority,
350            payload: self.payload,
351        }
352    }
353}
354
355/// Cross-station event frame.
356#[derive(Clone, Debug, PartialEq, Eq)]
357pub struct StationEventFrame {
358    /// Idempotency key.
359    pub event_id: EventId,
360    /// Source station.
361    pub source_station: StationId,
362    /// Target station.
363    pub target_station: StationId,
364    /// Tick observed at source.
365    pub source_tick: Tick,
366    /// Tick at which target should apply the event.
367    pub target_tick: Tick,
368    /// Priority class.
369    pub priority: EventPriority,
370    /// Event payload kind.
371    pub kind: EventKind,
372}
373
374impl StationEventFrame {
375    /// Converts a runtime station event into a wire frame.
376    pub fn from_event(event: &StationEvent) -> Self {
377        Self {
378            event_id: event.id,
379            source_station: event.source,
380            target_station: event.target,
381            source_tick: event.source_tick,
382            target_tick: event.target_tick,
383            priority: event.priority,
384            kind: event.kind.clone(),
385        }
386    }
387
388    /// Converts a wire frame into a runtime station event.
389    pub fn into_event(self) -> StationEvent {
390        StationEvent {
391            id: self.event_id,
392            source: self.source_station,
393            target: self.target_station,
394            source_tick: self.source_tick,
395            target_tick: self.target_tick,
396            priority: self.priority,
397            kind: self.kind,
398        }
399    }
400}
401
402/// Runtime barrier notification frame.
403#[derive(Clone, Debug, PartialEq, Eq)]
404pub struct BarrierFrame {
405    /// Target client.
406    pub client_id: ClientId,
407    /// Barrier id.
408    pub barrier_id: BarrierId,
409    /// Server tick associated with this barrier state.
410    pub server_tick: Tick,
411    /// Current barrier state.
412    pub state: BarrierState,
413}
414
415/// Decoded runtime frame.
416#[derive(Clone, Debug, PartialEq, Eq)]
417pub enum RuntimeFrame {
418    /// Replication update.
419    Replication(ReplicationFrame),
420    /// Client command ingress.
421    Command(CommandFrame),
422    /// Gateway-to-station command dispatch.
423    CommandDispatch(CommandDispatchFrame),
424    /// Command acknowledgement.
425    CommandAck(CommandAckFrame),
426    /// Barrier notification.
427    Barrier(BarrierFrame),
428    /// Cross-station event.
429    StationEvent(StationEventFrame),
430}
431
432/// Encodes frames into bytes.
433pub trait FrameEncoder {
434    /// Encoder error type.
435    type Error;
436
437    /// Encodes a replication frame into `out`.
438    fn encode_replication(
439        &mut self,
440        frame: &ReplicationFrame,
441        out: &mut Vec<u8>,
442    ) -> Result<(), Self::Error>;
443
444    /// Encodes a command acknowledgement frame into `out`.
445    fn encode_command_ack(
446        &mut self,
447        frame: &CommandAckFrame,
448        out: &mut Vec<u8>,
449    ) -> Result<(), Self::Error>;
450
451    /// Encodes a client command frame into `out`.
452    fn encode_command(
453        &mut self,
454        frame: &CommandFrame,
455        out: &mut Vec<u8>,
456    ) -> Result<(), Self::Error>;
457
458    /// Encodes an internal command dispatch frame into `out`.
459    fn encode_command_dispatch(
460        &mut self,
461        frame: &CommandDispatchFrame,
462        out: &mut Vec<u8>,
463    ) -> Result<(), Self::Error>;
464
465    /// Encodes a cross-station event frame into `out`.
466    fn encode_station_event(
467        &mut self,
468        frame: &StationEventFrame,
469        out: &mut Vec<u8>,
470    ) -> Result<(), Self::Error>;
471
472    /// Encodes a barrier frame into `out`.
473    fn encode_barrier(
474        &mut self,
475        frame: &BarrierFrame,
476        out: &mut Vec<u8>,
477    ) -> Result<(), Self::Error>;
478}
479
480/// Decodes frames from bytes.
481pub trait FrameDecoder {
482    /// Decoder error type.
483    type Error;
484
485    /// Decodes one runtime frame.
486    fn decode(&mut self, input: &[u8]) -> Result<RuntimeFrame, Self::Error>;
487}
488
489/// Binary decode error.
490#[derive(Clone, Debug, PartialEq, Eq)]
491pub enum BinaryDecodeError {
492    /// Input is empty.
493    Empty,
494    /// Unknown frame kind byte.
495    UnknownFrameKind(u8),
496    /// Frame ended before all fields were available.
497    Truncated {
498        /// Required bytes.
499        needed: usize,
500        /// Available bytes.
501        available: usize,
502    },
503    /// Barrier state byte is invalid.
504    InvalidBarrierState(u8),
505    /// Command priority byte is invalid.
506    InvalidCommandPriority(u8),
507    /// Event priority byte is invalid.
508    InvalidEventPriority(u8),
509    /// Event kind byte is invalid.
510    InvalidEventKind(u8),
511    /// Trailing bytes were present after a complete frame.
512    TrailingBytes(usize),
513}
514
515impl core::fmt::Display for BinaryDecodeError {
516    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
517        match self {
518            Self::Empty => f.write_str("empty frame"),
519            Self::UnknownFrameKind(kind) => write!(f, "unknown frame kind {kind}"),
520            Self::Truncated { needed, available } => {
521                write!(f, "truncated frame: needed {needed}, available {available}")
522            }
523            Self::InvalidBarrierState(state) => write!(f, "invalid barrier state {state}"),
524            Self::InvalidCommandPriority(priority) => {
525                write!(f, "invalid command priority {priority}")
526            }
527            Self::InvalidEventPriority(priority) => write!(f, "invalid event priority {priority}"),
528            Self::InvalidEventKind(kind) => write!(f, "invalid event kind {kind}"),
529            Self::TrailingBytes(bytes) => write!(f, "frame has {bytes} trailing bytes"),
530        }
531    }
532}
533
534impl std::error::Error for BinaryDecodeError {}
535
536/// Binary encode error.
537#[derive(Clone, Debug, PartialEq, Eq)]
538pub enum BinaryEncodeError {
539    /// A list length exceeded `u32::MAX`.
540    TooManyItems {
541        /// Field being encoded.
542        field: &'static str,
543        /// Actual item count.
544        actual: usize,
545    },
546    /// A byte payload exceeded `u32::MAX`.
547    PayloadTooLarge {
548        /// Field being encoded.
549        field: &'static str,
550        /// Actual byte count.
551        actual: usize,
552    },
553}
554
555impl core::fmt::Display for BinaryEncodeError {
556    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
557        match self {
558            Self::TooManyItems { field, actual } => {
559                write!(f, "{field} has too many items: {actual}")
560            }
561            Self::PayloadTooLarge { field, actual } => {
562                write!(f, "{field} payload is too large: {actual} bytes")
563            }
564        }
565    }
566}
567
568impl std::error::Error for BinaryEncodeError {}
569
570/// Simple little-endian binary frame decoder.
571#[derive(Clone, Copy, Debug, Default)]
572pub struct BinaryFrameDecoder;
573
574impl FrameDecoder for BinaryFrameDecoder {
575    type Error = BinaryDecodeError;
576
577    fn decode(&mut self, input: &[u8]) -> Result<RuntimeFrame, Self::Error> {
578        let mut cursor = Cursor::new(input);
579        let kind_byte = cursor.read_u8()?;
580        let kind = FrameKind::from_byte(kind_byte)
581            .ok_or(BinaryDecodeError::UnknownFrameKind(kind_byte))?;
582        let frame = match kind {
583            FrameKind::Replication => RuntimeFrame::Replication(ReplicationFrame {
584                client_id: ClientId::new(cursor.read_u64()?),
585                server_tick: Tick::new(cursor.read_u64()?),
586                entity_count: cursor.read_u32()?,
587                estimated_payload_bytes: cursor.read_u32()?,
588                entities: {
589                    let entity_delta_count = cursor.read_u32()? as usize;
590                    let mut entities = Vec::with_capacity(entity_delta_count);
591                    for _ in 0..entity_delta_count {
592                        let entity_id = EntityId::new(cursor.read_u64()?);
593                        let owner_epoch = OwnerEpoch::new(cursor.read_u64()?);
594                        let component_count = cursor.read_u16()? as usize;
595                        let mut components = Vec::with_capacity(component_count);
596                        for _ in 0..component_count {
597                            let component_id = ComponentId::new(cursor.read_u16()?);
598                            let version = cursor.read_u64()?;
599                            let flags = cursor.read_u8()?;
600                            let byte_len = cursor.read_u32()? as usize;
601                            let bytes = cursor.read_bytes(byte_len)?;
602                            components.push(ComponentDelta {
603                                component_id,
604                                version,
605                                flags,
606                                bytes,
607                            });
608                        }
609                        entities.push(EntityDelta {
610                            entity_id,
611                            owner_epoch,
612                            components,
613                        });
614                    }
615                    entities
616                },
617            }),
618            FrameKind::CommandAck => RuntimeFrame::CommandAck(CommandAckFrame {
619                client_id: ClientId::new(cursor.read_u64()?),
620                command_id: CommandId::new(cursor.read_u64()?),
621                server_tick: Tick::new(cursor.read_u64()?),
622                accepted: cursor.read_u8()? != 0,
623                reason_code: cursor.read_u16()?,
624            }),
625            FrameKind::Command => RuntimeFrame::Command(CommandFrame {
626                client_id: ClientId::new(cursor.read_u64()?),
627                command_id: CommandId::new(cursor.read_u64()?),
628                entity_id: EntityId::new(cursor.read_u64()?),
629                sequence: cursor.read_u64()?,
630                kind: cursor.read_u32()?,
631                priority: decode_command_priority(cursor.read_u8()?)?,
632                payload: {
633                    let byte_len = cursor.read_u32()? as usize;
634                    cursor.read_bytes(byte_len)?
635                },
636            }),
637            FrameKind::CommandDispatch => RuntimeFrame::CommandDispatch(CommandDispatchFrame {
638                station_id: StationId::new(cursor.read_u32()?),
639                client_id: ClientId::new(cursor.read_u64()?),
640                command_id: CommandId::new(cursor.read_u64()?),
641                entity_id: EntityId::new(cursor.read_u64()?),
642                sequence: cursor.read_u64()?,
643                received_at: Tick::new(cursor.read_u64()?),
644                kind: cursor.read_u32()?,
645                priority: decode_command_priority(cursor.read_u8()?)?,
646                payload: {
647                    let byte_len = cursor.read_u32()? as usize;
648                    cursor.read_bytes(byte_len)?
649                },
650            }),
651            FrameKind::Barrier => RuntimeFrame::Barrier(BarrierFrame {
652                client_id: ClientId::new(cursor.read_u64()?),
653                barrier_id: BarrierId::new(cursor.read_u64()?),
654                server_tick: Tick::new(cursor.read_u64()?),
655                state: decode_barrier_state(cursor.read_u8()?)?,
656            }),
657            FrameKind::StationEvent => RuntimeFrame::StationEvent(StationEventFrame {
658                event_id: EventId::new(cursor.read_u64()?),
659                source_station: StationId::new(cursor.read_u32()?),
660                target_station: StationId::new(cursor.read_u32()?),
661                source_tick: Tick::new(cursor.read_u64()?),
662                target_tick: Tick::new(cursor.read_u64()?),
663                priority: decode_event_priority(cursor.read_u8()?)?,
664                kind: decode_event_kind(&mut cursor)?,
665            }),
666        };
667        cursor.finish()?;
668        Ok(frame)
669    }
670}
671
672/// Simple little-endian binary frame encoder.
673#[derive(Clone, Copy, Debug, Default)]
674pub struct BinaryFrameEncoder;
675
676impl FrameEncoder for BinaryFrameEncoder {
677    type Error = BinaryEncodeError;
678
679    fn encode_replication(
680        &mut self,
681        frame: &ReplicationFrame,
682        out: &mut Vec<u8>,
683    ) -> Result<(), Self::Error> {
684        out.push(FrameKind::Replication as u8);
685        out.extend_from_slice(&frame.client_id.get().to_le_bytes());
686        out.extend_from_slice(&frame.server_tick.get().to_le_bytes());
687        out.extend_from_slice(&frame.entity_count.to_le_bytes());
688        out.extend_from_slice(&frame.estimated_payload_bytes.to_le_bytes());
689        write_len_u32("replication.entities", frame.entities.len(), out)?;
690        for entity in &frame.entities {
691            out.extend_from_slice(&entity.entity_id.get().to_le_bytes());
692            out.extend_from_slice(&entity.owner_epoch.get().to_le_bytes());
693            write_len_u16(
694                "replication.entity.components",
695                entity.components.len(),
696                out,
697            )?;
698            for component in &entity.components {
699                out.extend_from_slice(&component.component_id.get().to_le_bytes());
700                out.extend_from_slice(&component.version.to_le_bytes());
701                out.push(component.flags);
702                write_bytes("replication.component.bytes", &component.bytes, out)?;
703            }
704        }
705        Ok(())
706    }
707
708    fn encode_command_ack(
709        &mut self,
710        frame: &CommandAckFrame,
711        out: &mut Vec<u8>,
712    ) -> Result<(), Self::Error> {
713        out.push(FrameKind::CommandAck as u8);
714        out.extend_from_slice(&frame.client_id.get().to_le_bytes());
715        out.extend_from_slice(&frame.command_id.get().to_le_bytes());
716        out.extend_from_slice(&frame.server_tick.get().to_le_bytes());
717        out.push(u8::from(frame.accepted));
718        out.extend_from_slice(&frame.reason_code.to_le_bytes());
719        Ok(())
720    }
721
722    fn encode_command(
723        &mut self,
724        frame: &CommandFrame,
725        out: &mut Vec<u8>,
726    ) -> Result<(), Self::Error> {
727        out.push(FrameKind::Command as u8);
728        out.extend_from_slice(&frame.client_id.get().to_le_bytes());
729        out.extend_from_slice(&frame.command_id.get().to_le_bytes());
730        out.extend_from_slice(&frame.entity_id.get().to_le_bytes());
731        out.extend_from_slice(&frame.sequence.to_le_bytes());
732        out.extend_from_slice(&frame.kind.to_le_bytes());
733        out.push(encode_command_priority(frame.priority));
734        write_bytes("command.payload", &frame.payload, out)?;
735        Ok(())
736    }
737
738    fn encode_command_dispatch(
739        &mut self,
740        frame: &CommandDispatchFrame,
741        out: &mut Vec<u8>,
742    ) -> Result<(), Self::Error> {
743        out.push(FrameKind::CommandDispatch as u8);
744        out.extend_from_slice(&frame.station_id.get().to_le_bytes());
745        out.extend_from_slice(&frame.client_id.get().to_le_bytes());
746        out.extend_from_slice(&frame.command_id.get().to_le_bytes());
747        out.extend_from_slice(&frame.entity_id.get().to_le_bytes());
748        out.extend_from_slice(&frame.sequence.to_le_bytes());
749        out.extend_from_slice(&frame.received_at.get().to_le_bytes());
750        out.extend_from_slice(&frame.kind.to_le_bytes());
751        out.push(encode_command_priority(frame.priority));
752        write_bytes("command_dispatch.payload", &frame.payload, out)?;
753        Ok(())
754    }
755
756    fn encode_station_event(
757        &mut self,
758        frame: &StationEventFrame,
759        out: &mut Vec<u8>,
760    ) -> Result<(), Self::Error> {
761        out.push(FrameKind::StationEvent as u8);
762        out.extend_from_slice(&frame.event_id.get().to_le_bytes());
763        out.extend_from_slice(&frame.source_station.get().to_le_bytes());
764        out.extend_from_slice(&frame.target_station.get().to_le_bytes());
765        out.extend_from_slice(&frame.source_tick.get().to_le_bytes());
766        out.extend_from_slice(&frame.target_tick.get().to_le_bytes());
767        out.push(encode_event_priority(frame.priority));
768        encode_event_kind(&frame.kind, out);
769        Ok(())
770    }
771
772    fn encode_barrier(
773        &mut self,
774        frame: &BarrierFrame,
775        out: &mut Vec<u8>,
776    ) -> Result<(), Self::Error> {
777        out.push(FrameKind::Barrier as u8);
778        out.extend_from_slice(&frame.client_id.get().to_le_bytes());
779        out.extend_from_slice(&frame.barrier_id.get().to_le_bytes());
780        out.extend_from_slice(&frame.server_tick.get().to_le_bytes());
781        out.push(encode_barrier_state(frame.state));
782        Ok(())
783    }
784}
785
786fn write_len_u16(
787    field: &'static str,
788    len: usize,
789    out: &mut Vec<u8>,
790) -> Result<(), BinaryEncodeError> {
791    let len =
792        u16::try_from(len).map_err(|_| BinaryEncodeError::TooManyItems { field, actual: len })?;
793    out.extend_from_slice(&len.to_le_bytes());
794    Ok(())
795}
796
797fn write_len_u32(
798    field: &'static str,
799    len: usize,
800    out: &mut Vec<u8>,
801) -> Result<(), BinaryEncodeError> {
802    let len =
803        u32::try_from(len).map_err(|_| BinaryEncodeError::TooManyItems { field, actual: len })?;
804    out.extend_from_slice(&len.to_le_bytes());
805    Ok(())
806}
807
808fn write_bytes(
809    field: &'static str,
810    bytes: &[u8],
811    out: &mut Vec<u8>,
812) -> Result<(), BinaryEncodeError> {
813    let len = u32::try_from(bytes.len()).map_err(|_| BinaryEncodeError::PayloadTooLarge {
814        field,
815        actual: bytes.len(),
816    })?;
817    out.extend_from_slice(&len.to_le_bytes());
818    out.extend_from_slice(bytes);
819    Ok(())
820}
821
822fn encode_barrier_state(state: BarrierState) -> u8 {
823    match state {
824        BarrierState::Running => 0,
825        BarrierState::Requested => 1,
826        BarrierState::WaitingTickBoundary => 2,
827        BarrierState::Frozen => 3,
828        BarrierState::Resuming => 4,
829    }
830}
831
832fn decode_barrier_state(state: u8) -> Result<BarrierState, BinaryDecodeError> {
833    match state {
834        0 => Ok(BarrierState::Running),
835        1 => Ok(BarrierState::Requested),
836        2 => Ok(BarrierState::WaitingTickBoundary),
837        3 => Ok(BarrierState::Frozen),
838        4 => Ok(BarrierState::Resuming),
839        _ => Err(BinaryDecodeError::InvalidBarrierState(state)),
840    }
841}
842
843fn encode_command_priority(priority: CommandPriority) -> u8 {
844    match priority {
845        CommandPriority::Normal => 0,
846        CommandPriority::High => 1,
847        CommandPriority::Low => 2,
848    }
849}
850
851fn decode_command_priority(priority: u8) -> Result<CommandPriority, BinaryDecodeError> {
852    match priority {
853        0 => Ok(CommandPriority::Normal),
854        1 => Ok(CommandPriority::High),
855        2 => Ok(CommandPriority::Low),
856        _ => Err(BinaryDecodeError::InvalidCommandPriority(priority)),
857    }
858}
859
860fn encode_event_priority(priority: EventPriority) -> u8 {
861    match priority {
862        EventPriority::Critical => 0,
863        EventPriority::Important => 1,
864        EventPriority::BestEffort => 2,
865    }
866}
867
868fn decode_event_priority(priority: u8) -> Result<EventPriority, BinaryDecodeError> {
869    match priority {
870        0 => Ok(EventPriority::Critical),
871        1 => Ok(EventPriority::Important),
872        2 => Ok(EventPriority::BestEffort),
873        _ => Err(BinaryDecodeError::InvalidEventPriority(priority)),
874    }
875}
876
877fn encode_event_kind(kind: &EventKind, out: &mut Vec<u8>) {
878    match kind {
879        EventKind::Custom(kind) => {
880            out.push(0);
881            out.extend_from_slice(&kind.to_le_bytes());
882        }
883        EventKind::HandoffPrepare { entity_id } => {
884            out.push(1);
885            out.extend_from_slice(&entity_id.get().to_le_bytes());
886        }
887        EventKind::HandoffCommit {
888            entity_id,
889            owner_epoch,
890        } => {
891            out.push(2);
892            out.extend_from_slice(&entity_id.get().to_le_bytes());
893            out.extend_from_slice(&owner_epoch.get().to_le_bytes());
894        }
895    }
896}
897
898fn decode_event_kind(cursor: &mut Cursor<'_>) -> Result<EventKind, BinaryDecodeError> {
899    match cursor.read_u8()? {
900        0 => Ok(EventKind::Custom(cursor.read_u32()?)),
901        1 => Ok(EventKind::HandoffPrepare {
902            entity_id: EntityId::new(cursor.read_u64()?),
903        }),
904        2 => Ok(EventKind::HandoffCommit {
905            entity_id: EntityId::new(cursor.read_u64()?),
906            owner_epoch: OwnerEpoch::new(cursor.read_u64()?),
907        }),
908        kind => Err(BinaryDecodeError::InvalidEventKind(kind)),
909    }
910}
911
912struct Cursor<'a> {
913    input: &'a [u8],
914    offset: usize,
915}
916
917impl<'a> Cursor<'a> {
918    fn new(input: &'a [u8]) -> Self {
919        Self { input, offset: 0 }
920    }
921
922    fn read_u8(&mut self) -> Result<u8, BinaryDecodeError> {
923        self.require(1)?;
924        let value = self.input[self.offset];
925        self.offset += 1;
926        Ok(value)
927    }
928
929    fn read_u16(&mut self) -> Result<u16, BinaryDecodeError> {
930        let bytes = self.read_array::<2>()?;
931        Ok(u16::from_le_bytes(bytes))
932    }
933
934    fn read_u32(&mut self) -> Result<u32, BinaryDecodeError> {
935        let bytes = self.read_array::<4>()?;
936        Ok(u32::from_le_bytes(bytes))
937    }
938
939    fn read_u64(&mut self) -> Result<u64, BinaryDecodeError> {
940        let bytes = self.read_array::<8>()?;
941        Ok(u64::from_le_bytes(bytes))
942    }
943
944    fn read_array<const N: usize>(&mut self) -> Result<[u8; N], BinaryDecodeError> {
945        self.require(N)?;
946        let mut out = [0_u8; N];
947        out.copy_from_slice(&self.input[self.offset..self.offset + N]);
948        self.offset += N;
949        Ok(out)
950    }
951
952    fn read_bytes(&mut self, len: usize) -> Result<Vec<u8>, BinaryDecodeError> {
953        self.require(len)?;
954        let bytes = self.input[self.offset..self.offset + len].to_vec();
955        self.offset += len;
956        Ok(bytes)
957    }
958
959    fn require(&self, count: usize) -> Result<(), BinaryDecodeError> {
960        let needed = self.offset.saturating_add(count);
961        if needed > self.input.len() {
962            Err(BinaryDecodeError::Truncated {
963                needed,
964                available: self.input.len(),
965            })
966        } else {
967            Ok(())
968        }
969    }
970
971    fn finish(&self) -> Result<(), BinaryDecodeError> {
972        if self.offset == self.input.len() {
973            Ok(())
974        } else {
975            Err(BinaryDecodeError::TrailingBytes(
976                self.input.len().saturating_sub(self.offset),
977            ))
978        }
979    }
980}
981
982#[cfg(test)]
983mod tests {
984    use super::*;
985
986    #[test]
987    fn binary_codec_roundtrips_replication_frame() {
988        let frame = ReplicationFrame {
989            client_id: ClientId::new(9),
990            server_tick: Tick::new(42),
991            entity_count: 17,
992            estimated_payload_bytes: 544,
993            entities: vec![EntityDelta {
994                entity_id: EntityId::new(100),
995                owner_epoch: OwnerEpoch::new(2),
996                components: vec![ComponentDelta {
997                    component_id: ComponentId::new(1),
998                    version: 9,
999                    flags: 0,
1000                    bytes: vec![1, 2, 3, 4],
1001                }],
1002            }],
1003        };
1004        let mut encoder = BinaryFrameEncoder;
1005        let mut bytes = Vec::new();
1006        encoder
1007            .encode_replication(&frame, &mut bytes)
1008            .expect("encoder is infallible");
1009
1010        let decoded = BinaryFrameDecoder
1011            .decode(&bytes)
1012            .expect("decode should work");
1013        assert_eq!(decoded, RuntimeFrame::Replication(frame));
1014    }
1015
1016    #[test]
1017    fn binary_codec_roundtrips_command_ack_frame() {
1018        let frame = CommandAckFrame {
1019            client_id: ClientId::new(1),
1020            command_id: CommandId::new(2),
1021            server_tick: Tick::new(3),
1022            accepted: false,
1023            reason_code: 7,
1024        };
1025        let mut encoder = BinaryFrameEncoder;
1026        let mut bytes = Vec::new();
1027        encoder
1028            .encode_command_ack(&frame, &mut bytes)
1029            .expect("encoder is infallible");
1030
1031        let decoded = BinaryFrameDecoder
1032            .decode(&bytes)
1033            .expect("decode should work");
1034        assert_eq!(decoded, RuntimeFrame::CommandAck(frame));
1035    }
1036
1037    #[test]
1038    fn binary_codec_roundtrips_command_frame() {
1039        let frame = CommandFrame {
1040            client_id: ClientId::new(1),
1041            command_id: CommandId::new(2),
1042            entity_id: EntityId::new(3),
1043            sequence: 4,
1044            kind: 5,
1045            priority: CommandPriority::High,
1046            payload: vec![9, 8, 7],
1047        };
1048        let mut encoder = BinaryFrameEncoder;
1049        let mut bytes = Vec::new();
1050        encoder
1051            .encode_command(&frame, &mut bytes)
1052            .expect("encoder is infallible");
1053
1054        let decoded = BinaryFrameDecoder
1055            .decode(&bytes)
1056            .expect("decode should work");
1057        assert_eq!(decoded, RuntimeFrame::Command(frame));
1058    }
1059
1060    #[test]
1061    fn binary_codec_roundtrips_command_dispatch_frame() {
1062        let frame = CommandDispatchFrame {
1063            station_id: StationId::new(10),
1064            client_id: ClientId::new(1),
1065            command_id: CommandId::new(2),
1066            entity_id: EntityId::new(3),
1067            sequence: 4,
1068            received_at: Tick::new(99),
1069            kind: 5,
1070            priority: CommandPriority::High,
1071            payload: vec![9, 8, 7],
1072        };
1073        let mut encoder = BinaryFrameEncoder;
1074        let mut bytes = Vec::new();
1075        encoder
1076            .encode_command_dispatch(&frame, &mut bytes)
1077            .expect("encoder is infallible");
1078
1079        let decoded = BinaryFrameDecoder
1080            .decode(&bytes)
1081            .expect("decode should work");
1082        assert_eq!(decoded, RuntimeFrame::CommandDispatch(frame));
1083    }
1084
1085    #[test]
1086    fn command_frame_converts_to_runtime_envelope() {
1087        let frame = CommandFrame {
1088            client_id: ClientId::new(1),
1089            command_id: CommandId::new(2),
1090            entity_id: EntityId::new(3),
1091            sequence: 4,
1092            kind: 5,
1093            priority: CommandPriority::Low,
1094            payload: vec![1, 2, 3],
1095        };
1096
1097        let envelope = frame.clone().into_envelope(Tick::new(99));
1098        assert_eq!(envelope.id, frame.command_id);
1099        assert_eq!(envelope.received_at, Tick::new(99));
1100        assert_eq!(CommandFrame::from_envelope(&envelope), frame);
1101    }
1102
1103    #[test]
1104    fn command_dispatch_frame_preserves_stamped_envelope_tick() {
1105        let envelope = CommandEnvelope {
1106            id: CommandId::new(2),
1107            client_id: ClientId::new(1),
1108            entity_id: EntityId::new(3),
1109            sequence: 4,
1110            received_at: Tick::new(99),
1111            kind: 5,
1112            priority: CommandPriority::Low,
1113            payload: vec![1, 2, 3],
1114        };
1115
1116        let frame = CommandDispatchFrame::from_envelope(StationId::new(10), &envelope);
1117        assert_eq!(frame.station_id, StationId::new(10));
1118        assert_eq!(frame.received_at, Tick::new(99));
1119        assert_eq!(frame.into_envelope(), envelope);
1120    }
1121
1122    #[test]
1123    fn binary_codec_roundtrips_barrier_frame() {
1124        let frame = BarrierFrame {
1125            client_id: ClientId::new(1),
1126            barrier_id: BarrierId::new(99),
1127            server_tick: Tick::new(11),
1128            state: BarrierState::Frozen,
1129        };
1130        let mut encoder = BinaryFrameEncoder;
1131        let mut bytes = Vec::new();
1132        encoder
1133            .encode_barrier(&frame, &mut bytes)
1134            .expect("encoder is infallible");
1135
1136        let decoded = BinaryFrameDecoder
1137            .decode(&bytes)
1138            .expect("decode should work");
1139        assert_eq!(decoded, RuntimeFrame::Barrier(frame));
1140    }
1141
1142    #[test]
1143    fn binary_codec_roundtrips_station_event_frame() {
1144        let frames = [
1145            StationEventFrame {
1146                event_id: EventId::new(1),
1147                source_station: StationId::new(10),
1148                target_station: StationId::new(11),
1149                source_tick: Tick::new(2),
1150                target_tick: Tick::new(3),
1151                priority: EventPriority::Critical,
1152                kind: EventKind::Custom(7),
1153            },
1154            StationEventFrame {
1155                event_id: EventId::new(2),
1156                source_station: StationId::new(10),
1157                target_station: StationId::new(11),
1158                source_tick: Tick::new(2),
1159                target_tick: Tick::new(3),
1160                priority: EventPriority::Important,
1161                kind: EventKind::HandoffPrepare {
1162                    entity_id: EntityId::new(99),
1163                },
1164            },
1165            StationEventFrame {
1166                event_id: EventId::new(3),
1167                source_station: StationId::new(10),
1168                target_station: StationId::new(11),
1169                source_tick: Tick::new(2),
1170                target_tick: Tick::new(3),
1171                priority: EventPriority::BestEffort,
1172                kind: EventKind::HandoffCommit {
1173                    entity_id: EntityId::new(99),
1174                    owner_epoch: OwnerEpoch::new(5),
1175                },
1176            },
1177        ];
1178
1179        for frame in frames {
1180            let mut encoder = BinaryFrameEncoder;
1181            let mut bytes = Vec::new();
1182            encoder
1183                .encode_station_event(&frame, &mut bytes)
1184                .expect("encoder is infallible");
1185
1186            let decoded = BinaryFrameDecoder
1187                .decode(&bytes)
1188                .expect("decode should work");
1189            assert_eq!(decoded, RuntimeFrame::StationEvent(frame));
1190        }
1191    }
1192
1193    #[test]
1194    fn station_event_frame_converts_to_runtime_event() {
1195        let event = StationEvent {
1196            id: EventId::new(1),
1197            source: StationId::new(10),
1198            target: StationId::new(11),
1199            source_tick: Tick::new(2),
1200            target_tick: Tick::new(3),
1201            priority: EventPriority::Critical,
1202            kind: EventKind::HandoffPrepare {
1203                entity_id: EntityId::new(99),
1204            },
1205        };
1206
1207        let frame = StationEventFrame::from_event(&event);
1208        assert_eq!(frame.event_id, event.id);
1209        assert_eq!(frame.clone().into_event(), event);
1210    }
1211
1212    #[test]
1213    fn frame_builder_materializes_dirty_component_deltas() {
1214        use sectorsync_core::prelude::{
1215            Bounds, ComponentDescriptor, ComponentMigrationMode, ComponentSyncMode, InstanceId,
1216            NodeId, PolicyId, Position3, ReplicationPlan, StationConfig, Vec3, Vec3LeCodec,
1217        };
1218
1219        let mut station = Station::new(StationConfig {
1220            station_id: sectorsync_core::prelude::StationId::new(1),
1221            node_id: NodeId::new(1),
1222            instance_id: InstanceId::new(1),
1223            tick_rate_hz: 20,
1224        });
1225        let handle = station
1226            .spawn_owned(
1227                EntityId::new(10),
1228                Position3::new(0.0, 0.0, 0.0),
1229                Bounds::Point,
1230                PolicyId::new(0),
1231            )
1232            .expect("spawn should work");
1233
1234        let descriptor = ComponentDescriptor::sparse_blob(
1235            ComponentId::new(1),
1236            "velocity",
1237            ComponentSyncMode::Delta,
1238            ComponentMigrationMode::Copy,
1239            12,
1240        );
1241        let mut components = ComponentStore::default();
1242        components
1243            .set_typed(
1244                &descriptor,
1245                handle,
1246                3,
1247                &Vec3LeCodec,
1248                &Vec3::new(1.0, 2.0, 3.0),
1249            )
1250            .expect("typed component should encode");
1251
1252        let build = ReplicationFrameBuilder::new(ReplicationFrameLimits {
1253            max_entity_deltas: 8,
1254            max_components_per_entity: 4,
1255            max_component_bytes: 64,
1256        })
1257        .build(
1258            ClientId::new(5),
1259            Tick::new(9),
1260            &station,
1261            &ReplicationPlan {
1262                entities: vec![handle],
1263                stats: sectorsync_core::prelude::ReplicationStats::default(),
1264            },
1265            &components,
1266            &ComponentSelection {
1267                component_ids: vec![ComponentId::new(1)],
1268            },
1269        );
1270
1271        assert_eq!(build.stats.encoded_entities, 1);
1272        assert_eq!(build.stats.encoded_components, 1);
1273        assert_eq!(build.frame.entities[0].entity_id, EntityId::new(10));
1274        assert_eq!(build.frame.entities[0].components[0].version, 3);
1275    }
1276}
sectorsync_wire/lib.rs

sectorsync_wire/
lib.rs
