flatland_protocol/
frame.rs

1//! Length-prefixed postcard frames for TCP transport.
2//!
3//! Wire layout: `[magic:4][len:4 LE][postcard envelope]`
4//! Magic prevents interpreting payload bytes as a length after stream slip.
5
6use std::marker::Unpin;
7
8use crate::codec::Codec;
9use crate::{ClientMessage, Envelope, PostcardCodec, ServerMessage, PROTOCOL_VERSION};
10use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
11
12const MAX_FRAME_BYTES: usize = 1_048_576;
13/// Frame sentinel — must not appear as a misaligned length prefix.
14pub const FRAME_MAGIC: [u8; 4] = *b"FL3\x01";
15
16#[derive(Debug, thiserror::Error)]
17pub enum FrameError {
18    #[error("frame too large: {0} bytes")]
19    TooLarge(usize),
20    #[error("frame magic mismatch: expected FL3\\x01, got {got:?} (TCP frame read was interrupted — file a bug)")]
21    BadMagic { got: [u8; 4] },
22    #[error("io error: {0}")]
23    Io(#[from] std::io::Error),
24    #[error("codec error: {0}")]
25    Codec(#[from] crate::codec::CodecError),
26    #[error("protocol version mismatch: expected {PROTOCOL_VERSION}, got {0}")]
27    VersionMismatch(u16),
28}
29
30pub async fn write_client_message<W>(
31    writer: &mut W,
32    message: &ClientMessage,
33) -> Result<(), FrameError>
34where
35    W: AsyncWrite + Unpin + Send,
36{
37    write_payload(writer, &Envelope::new(message.clone())).await
38}
39
40pub async fn read_client_message<R>(reader: &mut R) -> Result<ClientMessage, FrameError>
41where
42    R: AsyncRead + Unpin + Send,
43{
44    let envelope: Envelope<ClientMessage> = read_payload(reader).await?;
45    Ok(envelope.payload)
46}
47
48pub async fn write_server_message<W>(
49    writer: &mut W,
50    message: &ServerMessage,
51) -> Result<(), FrameError>
52where
53    W: AsyncWrite + Unpin + Send,
54{
55    write_payload(writer, &Envelope::new(message.clone())).await
56}
57
58pub async fn read_server_message<R>(reader: &mut R) -> Result<ServerMessage, FrameError>
59where
60    R: AsyncRead + Unpin + Send,
61{
62    let envelope: Envelope<ServerMessage> = read_payload(reader).await?;
63    Ok(envelope.payload)
64}
65
66async fn write_payload<W, T>(writer: &mut W, envelope: &Envelope<T>) -> Result<(), FrameError>
67where
68    W: AsyncWrite + Unpin + Send,
69    T: serde::Serialize,
70{
71    let codec = PostcardCodec;
72    let bytes = codec.encode(envelope)?;
73    if bytes.len() > MAX_FRAME_BYTES {
74        return Err(FrameError::TooLarge(bytes.len()));
75    }
76    let len = u32::try_from(bytes.len()).map_err(|_| FrameError::TooLarge(bytes.len()))?;
77    writer.write_all(&FRAME_MAGIC).await?;
78    writer.write_u32_le(len).await?;
79    writer.write_all(&bytes).await?;
80    writer.flush().await?;
81    Ok(())
82}
83
84async fn read_payload<R, T>(reader: &mut R) -> Result<Envelope<T>, FrameError>
85where
86    R: AsyncRead + Unpin + Send,
87    T: serde::de::DeserializeOwned,
88{
89    let mut magic = [0_u8; 4];
90    reader.read_exact(&mut magic).await?;
91    if magic != FRAME_MAGIC {
92        return Err(FrameError::BadMagic { got: magic });
93    }
94    let len = reader.read_u32_le().await? as usize;
95    if len > MAX_FRAME_BYTES {
96        return Err(FrameError::TooLarge(len));
97    }
98    let mut buf = vec![0_u8; len];
99    reader.read_exact(&mut buf).await?;
100    let codec = PostcardCodec;
101    let envelope: Envelope<T> = codec.decode(&buf)?;
102    if envelope.protocol_version != PROTOCOL_VERSION {
103        return Err(FrameError::VersionMismatch(envelope.protocol_version));
104    }
105    Ok(envelope)
106}
107
108#[cfg(test)]
109mod tests {
110    use super::*;
111    use crate::types::EntityState;
112    use crate::{AuthCredential, ClientMessage, Hello, PROTOCOL_VERSION};
113    use uuid::Uuid;
114
115    #[tokio::test]
116    async fn roundtrip_client_message() {
117        let msg = ClientMessage::Hello(Hello {
118            client_name: "test".into(),
119            protocol_version: PROTOCOL_VERSION,
120            auth: Default::default(),
121            character_id: None,
122        });
123        let mut buf = Vec::new();
124        write_client_message(&mut buf, &msg).await.unwrap();
125        let parsed = read_client_message(&mut buf.as_slice()).await.unwrap();
126        assert_eq!(msg, parsed);
127    }
128
129    #[tokio::test]
130    async fn roundtrip_full_world_tick() {
131        use crate::types::{
132            PlayerSkills, PlayerVitals, PrimaryAttributes, ResourceNodeState, ResourceNodeView,
133            ServerMessage, TickDelta, Transform, Velocity2D, WorldClock, WorldCoord,
134        };
135
136        let delta = TickDelta {
137            tick: 42,
138            entities: vec![EntityState {
139                id: 1,
140                label: "Traveler".into(),
141                transform: Transform {
142                    position: WorldCoord::surface(128.0, 128.0),
143                    yaw: 0.0,
144                    velocity: Velocity2D { vx: 0.0, vy: 0.0 },
145                },
146                vitals: Some(PlayerVitals::default()),
147                attributes: Some(PrimaryAttributes::default()),
148                skills: Some(PlayerSkills::default()),
149                inside_building: None,
150                tile_id: None,
151                paperdoll_ref: None,
152                presentation_state: None,
153                sprite_mode: None,
154                progression_xp: None,
155                combat_cues: Vec::new(),
156            }],
157            resource_nodes: vec![ResourceNodeView {
158                id: "spawn-oak-1".into(),
159                label: "Oak tree".into(),
160                x: 126.0,
161                y: 134.0,
162                z: 0.0,
163                item_template: "oak_log".into(),
164                state: ResourceNodeState::Available,
165                blocking: true,
166                blocking_radius_m: 0.8,
167                tile_id: None,
168                yaw: 0.0,
169                pitch: 0.0,
170                roll: 0.0,
171                draw_scale: 1.0,
172                sprite_mode: None,
173                presentation_state: None,
174                growth_progress: None,
175                channel_start_tick: None,
176                channel_end_tick: None,
177                harvest_drop_templates: Vec::new(),
178            }],
179            buildings: Vec::new(),
180            doors: Vec::new(),
181            npcs: Vec::new(),
182            inventory: Vec::new(),
183            blueprints: Vec::new(),
184            world_clock: WorldClock::default(),
185            ground_drops: Vec::new(),
186            placed_containers: Vec::new(),
187            combat: None,
188            interior_map: None,
189            quest_log: Vec::new(),
190            hired_workers: Vec::new(),
191            interactables: Vec::new(),
192            ledger: None,
193            career: None,
194            combat_fx: Vec::new(),
195            property_plots: Vec::new(),
196            terrain_overlays: Vec::new(),
197        };
198        let msg = ServerMessage::Tick(delta);
199        let mut buf = Vec::new();
200        write_server_message(&mut buf, &msg).await.unwrap();
201        let parsed = read_server_message(&mut buf.as_slice()).await.unwrap();
202        assert_eq!(msg, parsed);
203    }
204
205    #[tokio::test]
206    async fn roundtrip_session_auth_hello() {
207        let msg = ClientMessage::Hello(Hello {
208            client_name: "traveler".into(),
209            protocol_version: PROTOCOL_VERSION,
210            auth: AuthCredential::Session {
211                token: "flat_sess_test".into(),
212            },
213            character_id: Some(Uuid::parse_str("019f399e-2b4e-7001-87a2-2e99c005e79d").unwrap()),
214        });
215        let mut buf = Vec::new();
216        write_client_message(&mut buf, &msg).await.unwrap();
217        let parsed = read_client_message(&mut buf.as_slice()).await.unwrap();
218        assert_eq!(msg, parsed);
219    }
220
221    #[tokio::test]
222    async fn rejects_garbage_without_length_desync() {
223        let mut buf = b"GET /health HTTP/1.1\r\n".as_slice();
224        let err = read_client_message(&mut buf).await.unwrap_err();
225        assert!(matches!(err, FrameError::BadMagic { .. }));
226    }
227
228    #[tokio::test]
229    async fn multiplexed_frames_stay_aligned() {
230        use crate::types::{TickDelta, Transform, WorldClock, WorldCoord};
231        use crate::{Intent, ServerMessage};
232
233        let (mut client_io, mut server_io) = tokio::io::duplex(64 * 1024);
234
235        for seq in 1..=200u32 {
236            write_client_message(
237                &mut client_io,
238                &ClientMessage::Intent(Intent::Stop { entity_id: 1, seq }),
239            )
240            .await
241            .unwrap();
242            write_server_message(
243                &mut server_io,
244                &ServerMessage::Tick(TickDelta {
245                    tick: seq as u64,
246                    entities: vec![EntityState {
247                        id: 1,
248                        label: "p".into(),
249                        transform: Transform {
250                            position: WorldCoord::surface(128.0, 128.0),
251                            yaw: 0.0,
252                            velocity: crate::types::Velocity2D { vx: 0.0, vy: 0.0 },
253                        },
254                        vitals: None,
255                        attributes: None,
256                        skills: None,
257                        inside_building: None,
258                        tile_id: None,
259                        paperdoll_ref: None,
260                        presentation_state: None,
261                        sprite_mode: None,
262                        progression_xp: None,
263                        combat_cues: Vec::new(),
264                    }],
265                    resource_nodes: vec![],
266                    buildings: vec![],
267                    doors: vec![],
268                    npcs: vec![],
269                    inventory: vec![],
270                    blueprints: vec![],
271                    world_clock: WorldClock::default(),
272                    ground_drops: vec![],
273                    placed_containers: vec![],
274                    combat: None,
275                    interior_map: None,
276                    quest_log: vec![],
277                    hired_workers: Vec::new(),
278                    interactables: vec![],
279                    ledger: None,
280                    career: None,
281                    combat_fx: Vec::new(),
282                    property_plots: Vec::new(),
283                    terrain_overlays: Vec::new(),
284                }),
285            )
286            .await
287            .unwrap();
288        }
289
290        for seq in 1..=200u32 {
291            let intent = read_client_message(&mut server_io).await.unwrap();
292            assert_eq!(
293                intent,
294                ClientMessage::Intent(Intent::Stop { entity_id: 1, seq })
295            );
296            let ServerMessage::Tick(delta) = read_server_message(&mut client_io).await.unwrap()
297            else {
298                panic!("expected tick");
299            };
300            assert_eq!(delta.tick, seq as u64);
301        }
302    }
303}
flatland_protocol/frame.rs

flatland_protocol/
frame.rs
