aetheris_encoder_serde/
serde_encoder.rs

1//! Implementation of the `SerdeEncoder` using `rmp-serde`.
2
3use std::io::Cursor;
4
5use serde::{Deserialize, Serialize};
6
7use aetheris_protocol::error::EncodeError;
8use aetheris_protocol::events::{ComponentUpdate, NetworkEvent, ReplicationEvent, WireEvent};
9use aetheris_protocol::traits::Encoder;
10use aetheris_protocol::types::{ClientId, ComponentKind, NetworkId};
11
12/// Internal header for serialized replication events.
13///
14/// Ensures a stable binary format across different `rmp-serde` configurations.
15#[derive(Debug, Serialize, Deserialize)]
16struct PacketHeader {
17    network_id: NetworkId,
18    component_kind: ComponentKind,
19    tick: u64,
20}
21
22/// A `serde`-based encoder that uses `rmp-serde` (`MessagePack`) for binary serialization.
23///
24/// This implementation targets Phase 1 MVP requirements for rapid iteration.
25/// It uses a fixed-size header followed by the raw component payload.
26#[derive(Debug, Default)]
27pub struct SerdeEncoder;
28
29impl SerdeEncoder {
30    /// Creates a new `SerdeEncoder`.
31    #[must_use]
32    pub fn new() -> Self {
33        Self
34    }
35
36    /// Encodes a `NetworkEvent` into raw bytes for transmission.
37    ///
38    /// # Errors
39    /// Returns `EncodeError` if the event fails to serialize or is a local-only variant.
40    pub fn encode_event(
41        &self,
42        event: &aetheris_protocol::events::NetworkEvent,
43    ) -> Result<Vec<u8>, EncodeError> {
44        let wire_event = match event {
45            NetworkEvent::Ping { tick, .. } if event.is_wire() => WireEvent::Ping { tick: *tick },
46            NetworkEvent::Pong { tick } => WireEvent::Pong { tick: *tick },
47            NetworkEvent::Auth { session_token } => WireEvent::Auth {
48                session_token: session_token.clone(),
49            },
50            NetworkEvent::Fragment { fragment, .. } => WireEvent::Fragment(fragment.clone()),
51            NetworkEvent::StressTest { count, rotate, .. } => WireEvent::StressTest {
52                count: *count,
53                rotate: *rotate,
54            },
55            NetworkEvent::Spawn {
56                entity_type,
57                x,
58                y,
59                rot,
60                ..
61            } => WireEvent::Spawn {
62                entity_type: *entity_type,
63                x: *x,
64                y: *y,
65                rot: *rot,
66            },
67            NetworkEvent::ClearWorld { .. } => WireEvent::ClearWorld,
68            NetworkEvent::ClientConnected(_)
69            | NetworkEvent::ClientDisconnected(_)
70            | NetworkEvent::UnreliableMessage { .. }
71            | NetworkEvent::ReliableMessage { .. }
72            | NetworkEvent::Ping { .. }
73            | NetworkEvent::SessionClosed(_)
74            | NetworkEvent::StreamReset(_)
75            | NetworkEvent::Disconnected(_) => {
76                return Err(EncodeError::Io(std::io::Error::other(format!(
77                    "Cannot encode local-only variant as wire event: {event:?}"
78                ))));
79            }
80        };
81        rmp_serde::to_vec(&wire_event)
82            .map_err(|e| EncodeError::Io(std::io::Error::other(e.to_string())))
83    }
84
85    /// Decodes raw bytes into a `NetworkEvent`.
86    ///
87    /// # Errors
88    /// Returns `EncodeError` if the bytes are not a valid `WireEvent`.
89    pub fn decode_event(
90        &self,
91        data: &[u8],
92    ) -> Result<aetheris_protocol::events::NetworkEvent, EncodeError> {
93        let wire_event: WireEvent = rmp_serde::from_slice(data).map_err(|e| {
94            EncodeError::MalformedPayload {
95                offset: 0, // In Phase 1 we don't track exact rmp-serde offset easily
96                message: e.to_string(),
97            }
98        })?;
99
100        Ok(match wire_event {
101            WireEvent::Ping { tick } => NetworkEvent::Ping {
102                client_id: ClientId(0), // Populated by transport/server
103                tick,
104            },
105            WireEvent::Pong { tick } => NetworkEvent::Pong { tick },
106            WireEvent::Auth { session_token } => NetworkEvent::Auth { session_token },
107            WireEvent::Fragment(fragment) => NetworkEvent::Fragment {
108                client_id: ClientId(0),
109                fragment,
110            },
111            WireEvent::StressTest { count, rotate } => NetworkEvent::StressTest {
112                client_id: ClientId(0), // Populated by server
113                count,
114                rotate,
115            },
116            WireEvent::Spawn {
117                entity_type,
118                x,
119                y,
120                rot,
121            } => NetworkEvent::Spawn {
122                client_id: ClientId(0),
123                entity_type,
124                x,
125                y,
126                rot,
127            },
128            WireEvent::ClearWorld => NetworkEvent::ClearWorld {
129                client_id: ClientId(0),
130            },
131        })
132    }
133}
134
135impl Encoder for SerdeEncoder {
136    fn encode_event(&self, event: &NetworkEvent) -> Result<Vec<u8>, EncodeError> {
137        self.encode_event(event)
138    }
139
140    fn decode_event(&self, data: &[u8]) -> Result<NetworkEvent, EncodeError> {
141        self.decode_event(data)
142    }
143
144    fn encode(&self, event: &ReplicationEvent, buffer: &mut [u8]) -> Result<usize, EncodeError> {
145        #[cfg(not(target_arch = "wasm32"))]
146        let start = std::time::Instant::now();
147        let header = PacketHeader {
148            network_id: event.network_id,
149            component_kind: event.component_kind,
150            tick: event.tick,
151        };
152
153        let mut cursor = Cursor::new(buffer);
154        let mut serializer = rmp_serde::Serializer::new(&mut cursor);
155
156        header.serialize(&mut serializer).map_err(|_e| {
157            metrics::counter!("aetheris_encoder_errors_total", "type" => "header_serialize_fail")
158                .increment(1);
159            // If it fails to serialize, it's likely a buffer overflow.
160            EncodeError::BufferOverflow {
161                needed: 32, // PacketHeader is small (~20 bytes)
162                available: cursor.get_ref().len(),
163            }
164        })?;
165
166        let header_len = usize::try_from(cursor.position()).unwrap_or(usize::MAX);
167        let payload_len = event.payload.len();
168        let total_needed = header_len + payload_len;
169
170        if total_needed > cursor.get_ref().len() {
171            metrics::counter!("aetheris_encoder_errors_total", "type" => "buffer_overflow")
172                .increment(1);
173            return Err(EncodeError::BufferOverflow {
174                needed: total_needed,
175                available: cursor.get_ref().len(),
176            });
177        }
178
179        // Copy payload manually after the header
180        let slice = cursor.into_inner();
181        slice[header_len..total_needed].copy_from_slice(&event.payload);
182
183        #[allow(clippy::cast_precision_loss)]
184        metrics::histogram!(
185            "aetheris_encoder_payload_size_bytes",
186            "operation" => "encode"
187        )
188        .record(total_needed as f64);
189
190        #[cfg(not(target_arch = "wasm32"))]
191        metrics::histogram!(
192            "aetheris_encoder_encode_duration_seconds",
193            "kind" => event.component_kind.0.to_string()
194        )
195        .record(start.elapsed().as_secs_f64());
196
197        Ok(total_needed)
198    }
199
200    fn decode(&self, buffer: &[u8]) -> Result<ComponentUpdate, EncodeError> {
201        #[cfg(not(target_arch = "wasm32"))]
202        let start = std::time::Instant::now();
203        let mut cursor = Cursor::new(buffer);
204        let mut deserializer = rmp_serde::Deserializer::new(&mut cursor);
205
206        let header = PacketHeader::deserialize(&mut deserializer).map_err(|e| {
207            metrics::counter!("aetheris_encoder_errors_total", "type" => "malformed_payload")
208                .increment(1);
209            EncodeError::MalformedPayload {
210                offset: usize::try_from(cursor.position()).unwrap_or(usize::MAX),
211                message: e.to_string(),
212            }
213        })?;
214
215        let header_len = usize::try_from(cursor.position()).unwrap_or(usize::MAX);
216        let payload = buffer
217            .get(header_len..)
218            .ok_or(EncodeError::MalformedPayload {
219                offset: header_len,
220                message: "Payload slice out of bounds".to_string(),
221            })?
222            .to_vec();
223
224        #[allow(clippy::cast_precision_loss)]
225        metrics::histogram!(
226            "aetheris_encoder_payload_size_bytes",
227            "operation" => "decode"
228        )
229        .record(buffer.len() as f64);
230
231        #[cfg(not(target_arch = "wasm32"))]
232        metrics::histogram!(
233            "aetheris_encoder_decode_duration_seconds",
234            "kind" => header.component_kind.0.to_string()
235        )
236        .record(start.elapsed().as_secs_f64());
237
238        Ok(ComponentUpdate {
239            network_id: header.network_id,
240            component_kind: header.component_kind,
241            payload,
242            tick: header.tick,
243        })
244    }
245
246    fn max_encoded_size(&self) -> usize {
247        aetheris_protocol::MAX_SAFE_PAYLOAD_SIZE
248    }
249}
250
251#[cfg(test)]
252mod tests {
253    use super::*;
254    use proptest::prelude::*;
255
256    #[test]
257    fn test_roundtrip() {
258        let encoder = SerdeEncoder::new();
259        let event = ReplicationEvent {
260            network_id: NetworkId(42),
261            component_kind: ComponentKind(1),
262            payload: vec![1, 2, 3, 4],
263            tick: 100,
264        };
265
266        let mut buffer = [0u8; 1200];
267        let bytes_written = encoder.encode(&event, &mut buffer).unwrap();
268        assert!(bytes_written > 0);
269
270        let update = encoder.decode(&buffer[..bytes_written]).unwrap();
271        assert_eq!(update.network_id, event.network_id);
272        assert_eq!(update.component_kind, event.component_kind);
273        assert_eq!(update.tick, event.tick);
274        assert_eq!(update.payload, event.payload);
275    }
276    #[test]
277    fn test_fragment_roundtrip() {
278        let encoder = SerdeEncoder::new();
279        let fragment = aetheris_protocol::events::FragmentedEvent {
280            message_id: 123,
281            fragment_index: 1,
282            total_fragments: 5,
283            payload: vec![1, 2, 3],
284        };
285
286        let event = NetworkEvent::Fragment {
287            client_id: aetheris_protocol::types::ClientId(0),
288            fragment: fragment.clone(),
289        };
290
291        let output = encoder.encode_event(&event).unwrap();
292        let decoded = encoder.decode_event(&output).unwrap();
293
294        if let NetworkEvent::Fragment {
295            client_id: _,
296            fragment: decoded_fragment,
297        } = decoded
298        {
299            assert_eq!(decoded_fragment.message_id, fragment.message_id);
300            assert_eq!(decoded_fragment.fragment_index, fragment.fragment_index);
301            assert_eq!(decoded_fragment.total_fragments, fragment.total_fragments);
302            assert_eq!(decoded_fragment.payload, fragment.payload);
303        } else {
304            panic!("Decoded event is not a Fragment: {decoded:?}");
305        }
306    }
307
308    #[test]
309    fn test_buffer_overflow() {
310        let encoder = SerdeEncoder::new();
311        let event = ReplicationEvent {
312            network_id: NetworkId(42),
313            component_kind: ComponentKind(1),
314            payload: vec![1, 2, 3, 4],
315            tick: 100,
316        };
317
318        let mut small_buffer = [0u8; 1];
319        let result = encoder.encode(&event, &mut small_buffer);
320        assert!(matches!(result, Err(EncodeError::BufferOverflow { .. })));
321    }
322
323    #[test]
324    fn test_malformed_payload() {
325        let encoder = SerdeEncoder::new();
326        let garbage = [0xff, 0xff, 0xff, 0xff];
327        let result = encoder.decode(&garbage);
328        if let Err(EncodeError::MalformedPayload { message, .. }) = result {
329            assert!(!message.is_empty());
330        } else {
331            panic!("Expected MalformedPayload error, got {result:?}");
332        }
333    }
334
335    proptest! {
336        #[test]
337        fn test_fuzz_decode(ref bytes in any::<Vec<u8>>()) {
338            let encoder = SerdeEncoder::new();
339            // Should never panic
340            let _ = encoder.decode(bytes);
341        }
342
343        #[test]
344        fn test_fuzz_roundtrip(
345            nid in any::<u64>(),
346            kind in any::<u16>(),
347            tick in any::<u64>(),
348            ref payload in any::<Vec<u8>>()
349        ) {
350            let encoder = SerdeEncoder::new();
351            let event = ReplicationEvent {
352                network_id: NetworkId(nid),
353                component_kind: ComponentKind(kind),
354                payload: payload.clone(),
355                tick,
356            };
357
358            let mut buffer = vec![0u8; 2048 + payload.len()];
359            if let Ok(written) = encoder.encode(&event, &mut buffer) {
360                let update = encoder.decode(&buffer[..written])
361                    .expect("Round-trip decode failed during fuzzed test");
362                assert_eq!(update.network_id, event.network_id);
363                assert_eq!(update.component_kind, event.component_kind);
364                assert_eq!(update.tick, event.tick);
365                assert_eq!(update.payload, event.payload);
366            }
367        }
368    }
369
370    #[test]
371    fn test_disconnected_not_serializable() {
372        let encoder = SerdeEncoder::new();
373        let event = NetworkEvent::Disconnected(ClientId(42));
374
375        // Attempting to encode a local-only event should return an error
376        let result = encoder.encode_event(&event);
377        assert!(result.is_err());
378        if let Err(EncodeError::Io(e)) = result {
379            assert!(e.to_string().contains("Cannot encode local-only variant"));
380        } else {
381            panic!("Expected EncodeError::Io with local-only message, got {result:?}");
382        }
383    }
384
385    #[test]
386    fn test_wire_event_exclusivity() {
387        let encoder = SerdeEncoder::new();
388        // ClientConnected is local-only
389        let event = NetworkEvent::ClientConnected(ClientId(1));
390        assert!(encoder.encode_event(&event).is_err());
391
392        // Ping is wire-capable (but we should have a way to distinguish wire pings if needed)
393        // For now, our implementation handles Ping in the match arm correctly by excluding it
394        // from certain paths or variants.
395        // Let's verify that a standard Auth event (which is wire) works.
396        let auth = NetworkEvent::Auth {
397            session_token: "token".to_string(),
398        };
399        assert!(encoder.encode_event(&auth).is_ok());
400    }
401}
aetheris_encoder_serde/serde_encoder.rs

aetheris_encoder_serde/
serde_encoder.rs
