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>(writer: &mut W, message: &ClientMessage) -> Result<(), FrameError>
31where
32    W: AsyncWrite + Unpin + Send,
33{
34    write_payload(writer, &Envelope::new(message.clone())).await
35}
36
37pub async fn read_client_message<R>(reader: &mut R) -> Result<ClientMessage, FrameError>
38where
39    R: AsyncRead + Unpin + Send,
40{
41    let envelope: Envelope<ClientMessage> = read_payload(reader).await?;
42    Ok(envelope.payload)
43}
44
45pub async fn write_server_message<W>(writer: &mut W, message: &ServerMessage) -> Result<(), FrameError>
46where
47    W: AsyncWrite + Unpin + Send,
48{
49    write_payload(writer, &Envelope::new(message.clone())).await
50}
51
52pub async fn read_server_message<R>(reader: &mut R) -> Result<ServerMessage, FrameError>
53where
54    R: AsyncRead + Unpin + Send,
55{
56    let envelope: Envelope<ServerMessage> = read_payload(reader).await?;
57    Ok(envelope.payload)
58}
59
60async fn write_payload<W, T>(writer: &mut W, envelope: &Envelope<T>) -> Result<(), FrameError>
61where
62    W: AsyncWrite + Unpin + Send,
63    T: serde::Serialize,
64{
65    let codec = PostcardCodec;
66    let bytes = codec.encode(envelope)?;
67    if bytes.len() > MAX_FRAME_BYTES {
68        return Err(FrameError::TooLarge(bytes.len()));
69    }
70    let len = u32::try_from(bytes.len()).map_err(|_| FrameError::TooLarge(bytes.len()))?;
71    writer.write_all(&FRAME_MAGIC).await?;
72    writer.write_u32_le(len).await?;
73    writer.write_all(&bytes).await?;
74    writer.flush().await?;
75    Ok(())
76}
77
78async fn read_payload<R, T>(reader: &mut R) -> Result<Envelope<T>, FrameError>
79where
80    R: AsyncRead + Unpin + Send,
81    T: serde::de::DeserializeOwned,
82{
83    let mut magic = [0_u8; 4];
84    reader.read_exact(&mut magic).await?;
85    if magic != FRAME_MAGIC {
86        return Err(FrameError::BadMagic { got: magic });
87    }
88    let len = reader.read_u32_le().await? as usize;
89    if len > MAX_FRAME_BYTES {
90        return Err(FrameError::TooLarge(len));
91    }
92    let mut buf = vec![0_u8; len];
93    reader.read_exact(&mut buf).await?;
94    let codec = PostcardCodec;
95    let envelope: Envelope<T> = codec.decode(&buf)?;
96    if envelope.protocol_version != PROTOCOL_VERSION {
97        return Err(FrameError::VersionMismatch(envelope.protocol_version));
98    }
99    Ok(envelope)
100}
101
102#[cfg(test)]
103mod tests {
104    use super::*;
105    use crate::types::EntityState;
106    use crate::{AuthCredential, ClientMessage, Hello, PROTOCOL_VERSION};
107    use uuid::Uuid;
108
109    #[tokio::test]
110    async fn roundtrip_client_message() {
111        let msg = ClientMessage::Hello(Hello {
112            client_name: "test".into(),
113            protocol_version: PROTOCOL_VERSION,
114            auth: Default::default(),
115            character_id: None,
116        });
117        let mut buf = Vec::new();
118        write_client_message(&mut buf, &msg).await.unwrap();
119        let parsed = read_client_message(&mut buf.as_slice()).await.unwrap();
120        assert_eq!(msg, parsed);
121    }
122
123    #[tokio::test]
124    async fn roundtrip_full_world_tick() {
125        use crate::types::{
126            PlayerSkills, PlayerVitals, PrimaryAttributes, ResourceNodeState, ResourceNodeView,
127            ServerMessage, TickDelta, Transform, Velocity2D, WorldClock, WorldCoord,
128        };
129
130        let delta = TickDelta {
131            tick: 42,
132            entities: vec![EntityState {
133                id: 1,
134                label: "Traveler".into(),
135                transform: Transform {
136                    position: WorldCoord::surface(128.0, 128.0),
137                    yaw: 0.0,
138                    velocity: Velocity2D { vx: 0.0, vy: 0.0 },
139                },
140                vitals: Some(PlayerVitals::default()),
141                attributes: Some(PrimaryAttributes::default()),
142                skills: Some(PlayerSkills::default()),
143                inside_building: None,
144            }],
145            resource_nodes: vec![ResourceNodeView {
146                id: "spawn-oak-1".into(),
147                label: "Oak tree".into(),
148                x: 126.0,
149                y: 134.0,
150                z: 0.0,
151                item_template: "oak_log".into(),
152                state: ResourceNodeState::Available,
153                blocking: true,
154                blocking_radius_m: 0.8,
155            }],
156            buildings: Vec::new(),
157            doors: Vec::new(),
158            npcs: Vec::new(),
159            inventory: Vec::new(),
160            blueprints: Vec::new(),
161            world_clock: WorldClock::default(),
162            ground_drops: Vec::new(),
163            placed_containers: Vec::new(),
164            combat: None,
165        };
166        let msg = ServerMessage::Tick(delta);
167        let mut buf = Vec::new();
168        write_server_message(&mut buf, &msg).await.unwrap();
169        let parsed = read_server_message(&mut buf.as_slice()).await.unwrap();
170        assert_eq!(msg, parsed);
171    }
172
173    #[tokio::test]
174    async fn roundtrip_session_auth_hello() {
175        let msg = ClientMessage::Hello(Hello {
176            client_name: "traveler".into(),
177            protocol_version: PROTOCOL_VERSION,
178            auth: AuthCredential::Session {
179                token: "flat_sess_test".into(),
180            },
181            character_id: Some(Uuid::parse_str("019f399e-2b4e-7001-87a2-2e99c005e79d").unwrap()),
182        });
183        let mut buf = Vec::new();
184        write_client_message(&mut buf, &msg).await.unwrap();
185        let parsed = read_client_message(&mut buf.as_slice()).await.unwrap();
186        assert_eq!(msg, parsed);
187    }
188
189    #[tokio::test]
190    async fn rejects_garbage_without_length_desync() {
191        let mut buf = b"GET /health HTTP/1.1\r\n".as_slice();
192        let err = read_client_message(&mut buf).await.unwrap_err();
193        assert!(matches!(err, FrameError::BadMagic { .. }));
194    }
195
196    #[tokio::test]
197    async fn multiplexed_frames_stay_aligned() {
198        use crate::types::{TickDelta, Transform, WorldClock, WorldCoord};
199        use crate::{Intent, ServerMessage};
200
201        let (mut client_io, mut server_io) = tokio::io::duplex(64 * 1024);
202
203        for seq in 1..=200u32 {
204            write_client_message(
205                &mut client_io,
206                &ClientMessage::Intent(Intent::Stop {
207                    entity_id: 1,
208                    seq,
209                }),
210            )
211            .await
212            .unwrap();
213            write_server_message(
214                &mut server_io,
215                &ServerMessage::Tick(TickDelta {
216                    tick: seq as u64,
217                    entities: vec![EntityState {
218                        id: 1,
219                        label: "p".into(),
220                        transform: Transform {
221                            position: WorldCoord::surface(128.0, 128.0),
222                            yaw: 0.0,
223                            velocity: crate::types::Velocity2D { vx: 0.0, vy: 0.0 },
224                        },
225                        vitals: None,
226                        attributes: None,
227                        skills: None,
228                        inside_building: None,
229                    }],
230                    resource_nodes: vec![],
231                    buildings: vec![],
232                    doors: vec![],
233                    npcs: vec![],
234                    inventory: vec![],
235                    blueprints: vec![],
236                    world_clock: WorldClock::default(),
237                    ground_drops: vec![],
238                    placed_containers: vec![],
239                    combat: None,
240                }),
241            )
242            .await
243            .unwrap();
244        }
245
246        for seq in 1..=200u32 {
247            let intent = read_client_message(&mut server_io).await.unwrap();
248            assert_eq!(
249                intent,
250                ClientMessage::Intent(Intent::Stop {
251                    entity_id: 1,
252                    seq,
253                })
254            );
255            let ServerMessage::Tick(delta) = read_server_message(&mut client_io).await.unwrap() else {
256                panic!("expected tick");
257            };
258            assert_eq!(delta.tick, seq as u64);
259        }
260    }
261}
flatland_protocol/frame.rs

flatland_protocol/
frame.rs
