Skip to main content

ghostlink_core/
protocol.rs

1//! Binary Protocol with CRC32 Checksums for Ghost-Link Discovery
2//!
3//! This module implements a fixed-width binary protocol using:
4//! - Fixed-size fields for zero-copy parsing
5//! - CRC32 checksums for frame integrity (hardware-accelerated via crc32fast)
6//! - Sequence numbers and versioning for ordering
7//! - Zero-copy encoding with thread-local buffers
8
9use crc32fast::Hasher;
10
11/// Ghost-Link EtherType (0x88B5)
12pub const GHOSTLINK_ETHERTYPE: u16 = 0x88B5;
13
14/// Protocol version
15pub const PROTOCOL_VERSION: u8 = 1;
16
17/// Maximum payload size for discovery frames
18pub const MAX_PAYLOAD_SIZE: usize = 256;
19
20/// Maximum frame size (header + max payload)
21pub const MAX_FRAME_SIZE: usize = 8 + MAX_PAYLOAD_SIZE;
22
23/// Frame header structure (fixed-width, 8 bytes)
24#[derive(Clone, Copy, Debug)]
25pub struct FrameHeader {
26    /// EtherType identifying the protocol
27    pub ether_type: u16,
28    /// Frame kind (see FrameKind enum)
29    pub kind: u8,
30    /// Protocol version
31    pub version: u8,
32    /// CRC32 checksum of payload
33    pub crc: u32,
34}
35
36impl FrameHeader {
37    const HEADER_SIZE: usize = 8;
38
39    /// Create a new frame header with computed CRC from pre-encoded payload
40    #[inline]
41    pub fn new(ether_type: u16, kind: u8, payload: &[u8]) -> Self {
42        let crc = crc32(payload);
43        Self {
44            ether_type,
45            kind,
46            version: PROTOCOL_VERSION,
47            crc,
48        }
49    }
50
51    /// Encode header directly into a pre-allocated buffer (zero-copy)
52    #[inline]
53    pub fn encode_into(&self, buf: &mut [u8; Self::HEADER_SIZE]) {
54        let et_bytes = self.ether_type.to_le_bytes();
55        buf[0] = et_bytes[0];
56        buf[1] = et_bytes[1];
57        buf[2] = self.kind;
58        buf[3] = self.version;
59        let crc_bytes = self.crc.to_le_bytes();
60        buf[4] = crc_bytes[0];
61        buf[5] = crc_bytes[1];
62        buf[6] = crc_bytes[2];
63        buf[7] = crc_bytes[3];
64    }
65
66    /// Decode header from bytes (zero-copy, no allocation)
67    #[inline]
68    pub fn decode(buf: &[u8]) -> Option<Self> {
69        if buf.len() < Self::HEADER_SIZE {
70            return None;
71        }
72        let ether_type = u16::from_le_bytes([buf[0], buf[1]]);
73        let kind = buf[2];
74        let version = buf[3];
75        let crc = u32::from_le_bytes([buf[4], buf[5], buf[6], buf[7]]);
76        Some(Self {
77            ether_type,
78            kind,
79            version,
80            crc,
81        })
82    }
83}
84
85/// Compute CRC32 checksum using hardware-accelerated crc32fast
86#[inline]
87pub fn crc32(data: &[u8]) -> u32 {
88    let mut hasher = Hasher::new();
89    hasher.update(data);
90    hasher.finalize()
91}
92
93/// Frame kind enumeration
94#[derive(Clone, Copy, Debug, PartialEq, Eq)]
95pub enum FrameKind {
96    /// Discovery frame - announces node presence
97    Discovery = 1,
98    /// Join frame - requests to join cluster
99    Join = 2,
100    /// Attestation frame - resource verification
101    Attestation = 3,
102    /// Health check frame - periodic liveness probe
103    HealthCheck = 4,
104    /// Resource advertisement frame - capability update
105    ResourceAdvert = 5,
106}
107
108impl FrameKind {
109    const fn as_u8(&self) -> u8 {
110        match self {
111            Self::Discovery => 1,
112            Self::Join => 2,
113            Self::Attestation => 3,
114            Self::HealthCheck => 4,
115            Self::ResourceAdvert => 5,
116        }
117    }
118}
119
120impl TryFrom<u8> for FrameKind {
121    type Error = &'static str;
122
123    fn try_from(value: u8) -> Result<Self, Self::Error> {
124        match value {
125            1 => Ok(Self::Discovery),
126            2 => Ok(Self::Join),
127            3 => Ok(Self::Attestation),
128            4 => Ok(Self::HealthCheck),
129            5 => Ok(Self::ResourceAdvert),
130            _ => Err("unknown frame kind"),
131        }
132    }
133}
134
135/// Node resources structure for discovery frames
136#[derive(Clone, Debug, Default)]
137pub struct NodeResources {
138    /// Unique node identifier
139    pub id: String,
140    /// GPU VRAM in GB (f32 for precision)
141    pub vram_gb: f32,
142    /// System memory in GB
143    pub system_memory_gb: f32,
144    /// CUDA compute capability string (e.g., "8.9")
145    pub compute_capability: String,
146    /// GPU name/model
147    pub gpu_name: Option<String>,
148    /// Port this node's `ggml-rpc-server` is listening on, when it has opted
149    /// in to contributing compute to the cluster (see
150    /// `ghost_link::rpc_cluster`). `None` means this node either isn't
151    /// running one or the peer that sent this frame predates this field —
152    /// both are indistinguishable and both correctly mean "don't route
153    /// distributed inference through this node."
154    pub rpc_port: Option<u16>,
155}
156
157impl NodeResources {
158    /// Create new node resources
159    pub fn new(
160        id: impl Into<String>,
161        vram_gb: f32,
162        system_memory_gb: f32,
163        compute_capability: impl Into<String>,
164        gpu_name: Option<String>,
165    ) -> Self {
166        Self {
167            id: id.into(),
168            vram_gb,
169            system_memory_gb,
170            compute_capability: compute_capability.into(),
171            gpu_name,
172            rpc_port: None,
173        }
174    }
175
176    /// Builder-style setter so adding this field didn't require touching
177    /// every existing `NodeResources::new(...)` call site across the
178    /// workspace's tests and CLI commands.
179    pub fn with_rpc_port(mut self, port: u16) -> Self {
180        self.rpc_port = Some(port);
181        self
182    }
183
184    /// Serialize node resources into an existing binary buffer.
185    ///
186    /// Returns the length of the written payload on success, or an Error on validation failure.
187    #[inline]
188    pub fn encode_payload_into(
189        &self,
190        buffer: &mut Vec<u8>,
191        max_size: usize,
192    ) -> Result<usize, &'static str> {
193        let id_bytes = self.id.as_bytes();
194        let cc_bytes = self.compute_capability.as_bytes();
195        let gpu_bytes = self.gpu_name.as_ref().map(|name| name.as_bytes());
196
197        if id_bytes.len() > u8::MAX as usize || cc_bytes.len() > u8::MAX as usize {
198            return Err("field length exceeds u8::MAX");
199        }
200        if let Some(gpu_bytes) = gpu_bytes {
201            if gpu_bytes.len() > u8::MAX as usize {
202                return Err("GPU name length exceeds u8::MAX");
203            }
204        }
205
206        // Trailing 2 bytes: rpc_port as little-endian u16, 0 meaning "none"
207        // (real port 0 is not a valid bind target, so it's an unambiguous
208        // sentinel). Appended *after* the gpu_name section so a decoder built
209        // before this field existed simply never reads these bytes — it
210        // never checked for or rejected trailing data — and this encoder's
211        // output stays parseable by that older decoder. A decoder built
212        // after this field existed, reading an older (shorter) payload from
213        // a peer that predates it, just doesn't find the 2 extra bytes and
214        // defaults to `None` — see `decode_payload`.
215        let payload_len =
216            11 + id_bytes.len() + cc_bytes.len() + gpu_bytes.map_or(0, |bytes| 1 + bytes.len()) + 2;
217        if payload_len > max_size {
218            return Err("payload length exceeds max_size");
219        }
220
221        buffer.push(id_bytes.len() as u8);
222        buffer.extend_from_slice(id_bytes);
223        buffer.extend_from_slice(&self.vram_gb.to_le_bytes());
224        buffer.extend_from_slice(&self.system_memory_gb.to_le_bytes());
225        buffer.push(cc_bytes.len() as u8);
226        buffer.extend_from_slice(cc_bytes);
227
228        if let Some(gpu_bytes) = gpu_bytes {
229            buffer.push(1);
230            buffer.push(gpu_bytes.len() as u8);
231            buffer.extend_from_slice(gpu_bytes);
232        } else {
233            buffer.push(0);
234        }
235
236        buffer.extend_from_slice(&self.rpc_port.unwrap_or(0).to_le_bytes());
237
238        Ok(payload_len)
239    }
240
241    /// Serialize node resources to fixed-width binary payload
242    ///
243    /// Format: [id_len(1) + id + vram_f32_le + mem_f32_le + cc_len(1) + cc]
244    #[inline]
245    pub fn encode_payload(&self, max_size: usize) -> Vec<u8> {
246        let id_bytes_len = self.id.len();
247        let cc_bytes_len = self.compute_capability.len();
248        let gpu_bytes_len = self.gpu_name.as_ref().map_or(0, |name| name.len());
249        let est_len = 11
250            + id_bytes_len
251            + cc_bytes_len
252            + if gpu_bytes_len > 0 {
253                2 + gpu_bytes_len
254            } else {
255                0
256            }
257            + 2;
258
259        let mut payload = Vec::with_capacity(est_len);
260        if self.encode_payload_into(&mut payload, max_size).is_ok() {
261            payload
262        } else {
263            Vec::new()
264        }
265    }
266
267    /// Deserialize node resources from binary payload
268    pub fn decode_payload(payload: &[u8]) -> Result<Self, String> {
269        let len = payload.len();
270        if len < 11 {
271            return Err("payload too short".into());
272        }
273
274        let mut cursor = 0usize;
275
276        let id_len = payload[cursor] as usize;
277        cursor += 1;
278        let id_end = cursor + id_len;
279        if id_end > len {
280            return Err("payload truncated".into());
281        }
282        // Optimize: Avoid allocating and copying to a temporary Vec first.
283        // Direct zero-copy slice validation using std::str::from_utf8, then allocating only on success.
284        let id = std::str::from_utf8(&payload[cursor..id_end])
285            .map_err(|_| "invalid UTF-8 in ID".to_string())?
286            .to_string();
287        cursor = id_end;
288
289        let vram_end = cursor + 8;
290        if vram_end > len {
291            return Err("payload truncated".into());
292        }
293        let vram_gb = f32::from_le_bytes([
294            payload[cursor],
295            payload[cursor + 1],
296            payload[cursor + 2],
297            payload[cursor + 3],
298        ]);
299        let system_memory_gb = f32::from_le_bytes([
300            payload[cursor + 4],
301            payload[cursor + 5],
302            payload[cursor + 6],
303            payload[cursor + 7],
304        ]);
305        cursor = vram_end;
306
307        if cursor >= len {
308            return Err("payload truncated".into());
309        }
310        let cc_len = payload[cursor] as usize;
311        cursor += 1;
312        let cc_end = cursor + cc_len;
313        if cc_end > len {
314            return Err("payload truncated".into());
315        }
316        // Optimize: Zero-copy validation followed by direct to_string allocation on success.
317        let compute_capability = std::str::from_utf8(&payload[cursor..cc_end])
318            .map_err(|_| "invalid UTF-8 in CC".to_string())?
319            .to_string();
320        cursor = cc_end;
321
322        if cursor >= len {
323            return Err("payload truncated".into());
324        }
325        let has_gpu = payload[cursor] == 1;
326        cursor += 1;
327
328        let gpu_name = if has_gpu {
329            if cursor >= len {
330                return Err("payload truncated".into());
331            }
332            let gpu_len = payload[cursor] as usize;
333            cursor += 1;
334            let gpu_end = cursor + gpu_len;
335            if gpu_end > len {
336                return Err("payload truncated".into());
337            }
338            // Optimize: Zero-copy validation followed by direct to_string allocation on success.
339            let name = std::str::from_utf8(&payload[cursor..gpu_end])
340                .map_err(|_| "invalid UTF-8 in GPU name".to_string())?
341                .to_string();
342            // Was previously left unset here — harmless before this field
343            // existed (cursor was never read again), but the rpc_port read
344            // below now depends on this pointing past the GPU name bytes
345            // rather than into the middle of them.
346            cursor = gpu_end;
347            Some(name)
348        } else {
349            None
350        };
351
352        // Trailing rpc_port field, added after this format shipped — a
353        // payload from an older peer simply won't have these 2 bytes, which
354        // correctly decodes as "no RPC contribution advertised" rather than
355        // an error.
356        let rpc_port = if cursor + 2 <= len {
357            let port = u16::from_le_bytes([payload[cursor], payload[cursor + 1]]);
358            if port == 0 {
359                None
360            } else {
361                Some(port)
362            }
363        } else {
364            None
365        };
366
367        Ok(Self {
368            id,
369            vram_gb,
370            system_memory_gb,
371            compute_capability,
372            gpu_name,
373            rpc_port,
374        })
375    }
376}
377
378/// Discovery frame with binary encoding and CRC32
379#[derive(Clone, Debug)]
380pub struct DiscoveryFrame {
381    pub kind: FrameKind,
382    pub node: NodeResources,
383}
384
385impl DiscoveryFrame {
386    /// Encode discovery frame to bytes using a stack-allocated fixed-size buffer.
387    /// Avoids all Vec capacity checks and `extend_from_slice` overhead.
388    #[inline]
389    pub fn encode(&self) -> Vec<u8> {
390        let mut buf = [0u8; MAX_FRAME_SIZE];
391        let et = GHOSTLINK_ETHERTYPE.to_le_bytes();
392        buf[0] = et[0];
393        buf[1] = et[1];
394        buf[2] = self.kind.as_u8();
395        buf[3] = PROTOCOL_VERSION;
396
397        let id_bytes = self.node.id.as_bytes();
398        let cc_bytes = self.node.compute_capability.as_bytes();
399
400        if id_bytes.len() > u8::MAX as usize || cc_bytes.len() > u8::MAX as usize {
401            buf[4..8].copy_from_slice(&crc32(&[]).to_le_bytes());
402            return buf[..8].to_vec();
403        }
404        if let Some(ref gpu_name) = self.node.gpu_name {
405            if gpu_name.len() > u8::MAX as usize {
406                buf[4..8].copy_from_slice(&crc32(&[]).to_le_bytes());
407                return buf[..8].to_vec();
408            }
409        }
410
411        // Guard against the combined fields overflowing the fixed-size stack
412        // buffer: each field is individually bounded by u8::MAX above, but
413        // their sum can still exceed MAX_PAYLOAD_SIZE and panic on the
414        // slice-index writes below if left unchecked.
415        let gpu_len = self.node.gpu_name.as_ref().map(|name| name.len());
416        // +2 for the trailing rpc_port field this hand-rolled duplicate of
417        // `NodeResources::encode_payload_into` previously omitted entirely —
418        // UDP discovery replies silently dropped rpc_port while the mDNS
419        // path (which does call encode_payload_into's TXT-record analog)
420        // carried it correctly, so a UDP-discovered peer could never be
421        // selected for distributed inference. See `decode_payload`, which
422        // already tolerated this field being absent/present either way.
423        let payload_len =
424            11 + id_bytes.len() + cc_bytes.len() + gpu_len.map_or(0, |len| 1 + len) + 2;
425        if payload_len > MAX_PAYLOAD_SIZE {
426            buf[4..8].copy_from_slice(&crc32(&[]).to_le_bytes());
427            return buf[..8].to_vec();
428        }
429
430        let mut pos = 8usize;
431
432        buf[pos] = id_bytes.len() as u8;
433        pos += 1;
434        buf[pos..pos + id_bytes.len()].copy_from_slice(id_bytes);
435        pos += id_bytes.len();
436
437        buf[pos..pos + 4].copy_from_slice(&self.node.vram_gb.to_le_bytes());
438        pos += 4;
439
440        buf[pos..pos + 4].copy_from_slice(&self.node.system_memory_gb.to_le_bytes());
441        pos += 4;
442
443        buf[pos] = cc_bytes.len() as u8;
444        pos += 1;
445        buf[pos..pos + cc_bytes.len()].copy_from_slice(cc_bytes);
446        pos += cc_bytes.len();
447
448        if let Some(ref gpu_name) = self.node.gpu_name {
449            let gpu_bytes = gpu_name.as_bytes();
450            buf[pos] = 1;
451            pos += 1;
452            buf[pos] = gpu_bytes.len() as u8;
453            pos += 1;
454            buf[pos..pos + gpu_bytes.len()].copy_from_slice(gpu_bytes);
455            pos += gpu_bytes.len();
456        } else {
457            buf[pos] = 0;
458            pos += 1;
459        }
460
461        let rpc_port_bytes = self.node.rpc_port.unwrap_or(0).to_le_bytes();
462        buf[pos..pos + 2].copy_from_slice(&rpc_port_bytes);
463        pos += 2;
464
465        let crc = crc32(&buf[8..pos]);
466        buf[4..8].copy_from_slice(&crc.to_le_bytes());
467
468        buf[..pos].to_vec()
469    }
470
471    /// Encode discovery frame into a pre-allocated buffer (zero-copy hot path)
472    /// Returns the number of bytes written.
473    #[inline]
474    pub fn encode_into(&self, buf: &mut Vec<u8>) -> usize {
475        buf.clear();
476        buf.extend_from_slice(&[0u8; 8]);
477
478        if self.node.encode_payload_into(buf, MAX_PAYLOAD_SIZE).is_ok() {
479            let payload = &buf[8..];
480            let crc = crc32(payload);
481
482            let header = FrameHeader {
483                ether_type: GHOSTLINK_ETHERTYPE,
484                kind: self.kind.as_u8(),
485                version: PROTOCOL_VERSION,
486                crc,
487            };
488            header.encode_into(unsafe { &mut *(buf.as_mut_ptr() as *mut [u8; 8]) });
489
490            buf.len()
491        } else {
492            buf.clear();
493            buf.extend_from_slice(&[0u8; 8]);
494            let header = FrameHeader {
495                ether_type: GHOSTLINK_ETHERTYPE,
496                kind: self.kind.as_u8(),
497                version: PROTOCOL_VERSION,
498                crc: crc32(&[]),
499            };
500            header.encode_into(unsafe { &mut *(buf.as_mut_ptr() as *mut [u8; 8]) });
501            buf.len()
502        }
503    }
504
505    /// Decode discovery frame from bytes (header + payload) - optimized path
506    #[inline]
507    pub fn decode(bytes: &[u8]) -> Result<Self, String> {
508        if bytes.len() < FrameHeader::HEADER_SIZE {
509            return Err("frame too short".into());
510        }
511
512        // Fast header decode using zero-copy method
513        let header = FrameHeader::decode(bytes).ok_or("invalid header")?;
514
515        if header.ether_type != GHOSTLINK_ETHERTYPE {
516            return Err(format!("unexpected EtherType 0x{:04x}", header.ether_type));
517        }
518
519        if header.version != PROTOCOL_VERSION {
520            return Err(format!("unsupported protocol version {}", header.version));
521        }
522
523        // Fast CRC check
524        let payload = &bytes[FrameHeader::HEADER_SIZE..];
525        let computed_crc = crc32(payload);
526
527        if computed_crc != header.crc {
528            return Err(format!(
529                "CRC mismatch: expected 0x{:08x}, got 0x{:08x}",
530                header.crc, computed_crc
531            ));
532        }
533
534        // Decode node resources from payload
535        let node = NodeResources::decode_payload(payload)?;
536
537        Ok(Self {
538            kind: FrameKind::try_from(header.kind)?,
539            node,
540        })
541    }
542}
543
544#[cfg(test)]
545mod tests {
546    use super::*;
547
548    #[test]
549    fn discovery_frames_round_trip() {
550        let frame = DiscoveryFrame {
551            kind: FrameKind::Join,
552            node: NodeResources::new("node-b", 12.0, 32.0, "8.6", None),
553        };
554
555        let encoded = frame.encode();
556        let decoded = DiscoveryFrame::decode(&encoded).unwrap();
557
558        assert_eq!(decoded.kind, frame.kind);
559        assert_eq!(decoded.node.id, frame.node.id);
560        assert_eq!(decoded.node.vram_gb, frame.node.vram_gb);
561    }
562
563    #[test]
564    fn discovery_frame_encode_round_trips_rpc_port() {
565        // Regression test: `DiscoveryFrame::encode()` is a separate,
566        // hand-duplicated serializer from `NodeResources::encode_payload_into`
567        // (kept for its zero-copy-buffer calling convention) — it silently
568        // omitted the rpc_port field entirely when that field was added,
569        // so a UDP-discovered peer's contribute-compute port never survived
570        // the wire even though the equivalent mDNS path (which does reuse
571        // the shared encoder) carried it correctly. `encode_into` already
572        // covered this path; `encode()` (what `discovery.rs`'s UDP broadcast
573        // path actually calls) did not.
574        let frame = DiscoveryFrame {
575            kind: FrameKind::Join,
576            node: NodeResources::new("node-c", 12.0, 32.0, "8.6", None).with_rpc_port(50052),
577        };
578
579        let encoded = frame.encode();
580        let decoded = DiscoveryFrame::decode(&encoded).unwrap();
581
582        assert_eq!(decoded.node.rpc_port, Some(50052));
583    }
584
585    #[test]
586    fn discovery_frame_encode_without_rpc_port_decodes_to_none() {
587        let frame = DiscoveryFrame {
588            kind: FrameKind::Join,
589            node: NodeResources::new("node-d", 8.0, 16.0, "7.5", None),
590        };
591
592        let encoded = frame.encode();
593        let decoded = DiscoveryFrame::decode(&encoded).unwrap();
594
595        assert_eq!(decoded.node.rpc_port, None);
596    }
597
598    #[test]
599    fn crc_verification_fails_on_modified_payload() {
600        let frame = DiscoveryFrame {
601            kind: FrameKind::Discovery,
602            node: NodeResources::new("node-a", 24.0, 64.0, "8.9", Some("RTX4090".to_string())),
603        };
604
605        let encoded = frame.encode();
606        let mut modified = encoded.clone();
607        modified[10] ^= 0xFF; // Modify payload
608
609        let result = DiscoveryFrame::decode(&modified);
610        assert!(result.is_err());
611        assert!(result.unwrap_err().contains("CRC mismatch"));
612    }
613
614    #[test]
615    fn rejects_wrong_ether_type() {
616        let mut fake_frame = vec![0u8; 10];
617        fake_frame[0] = 0xB5u8; // Low byte of GHOSTLINK_ETHERTYPE (0x88B5 LE)
618        fake_frame[1] = 0xFF; // Wrong high byte
619
620        let result = DiscoveryFrame::decode(&fake_frame);
621        assert!(result.is_err());
622    }
623
624    #[test]
625    fn encode_payload_into_accepts_exact_max_size_with_gpu_name() {
626        // Regression test: encode_payload_into previously overcounted the
627        // gpu-name field length by 1 byte, causing it to reject payloads
628        // that actually fit exactly within max_size.
629        let node = NodeResources::new("a", 1.0, 1.0, "b", Some("c".to_string()));
630        let mut buffer = Vec::new();
631        // Actual encoded length: id_len(1)+id(1)+vram(4)+mem(4)+cc_len(1)+cc(1)
632        // +has_gpu(1)+gpu_len(1)+gpu(1)+rpc_port(2) = 17 bytes.
633        let written = node
634            .encode_payload_into(&mut buffer, 17)
635            .expect("payload of exactly 17 bytes must fit within max_size 17");
636        assert_eq!(written, 17);
637        assert_eq!(buffer.len(), 17);
638    }
639
640    #[test]
641    fn node_resources_round_trip_preserves_rpc_port() {
642        let node = NodeResources::new("node-e", 12.0, 32.0, "8.6", None).with_rpc_port(50052);
643        let encoded = node.encode_payload(64);
644        let decoded = NodeResources::decode_payload(&encoded).expect("decode");
645        assert_eq!(decoded.rpc_port, Some(50052));
646    }
647
648    #[test]
649    fn node_resources_round_trip_without_rpc_port_is_none() {
650        let node = NodeResources::new("node-f", 8.0, 16.0, "7.5", None);
651        let encoded = node.encode_payload(64);
652        let decoded = NodeResources::decode_payload(&encoded).expect("decode");
653        assert_eq!(decoded.rpc_port, None);
654    }
655
656    #[test]
657    fn node_resources_round_trip_with_gpu_name_and_rpc_port_together() {
658        // Regression test: decode_payload previously never advanced `cursor`
659        // past the GPU name bytes (harmless while nothing after it was ever
660        // read), which would have made a trailing field like rpc_port read
661        // from the middle of the GPU name instead of after it.
662        let node = NodeResources::new("node-g", 24.0, 64.0, "8.9", Some("RTX 4090".to_string()))
663            .with_rpc_port(50053);
664        let encoded = node.encode_payload(128);
665        let decoded = NodeResources::decode_payload(&encoded).expect("decode");
666        assert_eq!(decoded.gpu_name, Some("RTX 4090".to_string()));
667        assert_eq!(decoded.rpc_port, Some(50053));
668    }
669
670    #[test]
671    fn decode_payload_from_before_rpc_port_existed_defaults_to_none() {
672        // Simulates a frame from an older peer build: same format, just
673        // without the trailing 2 bytes.
674        let node = NodeResources::new("node-h", 8.0, 16.0, "7.5", None);
675        let mut encoded = node.encode_payload(64);
676        assert!(encoded.len() >= 2);
677        encoded.truncate(encoded.len() - 2);
678        let decoded = NodeResources::decode_payload(&encoded).expect("decode");
679        assert_eq!(decoded.rpc_port, None);
680    }
681
682    #[test]
683    fn encode_does_not_panic_when_combined_field_lengths_exceed_max_payload() {
684        // Regression test: DiscoveryFrame::encode only validated each field
685        // individually against u8::MAX, so combined field lengths could
686        // exceed MAX_PAYLOAD_SIZE and panic on the fixed-size stack buffer
687        // writes. It should now fall back to a safe header-only frame.
688        let big_id = "x".repeat(200);
689        let big_cc = "y".repeat(200);
690        let frame = DiscoveryFrame {
691            kind: FrameKind::Discovery,
692            node: NodeResources::new(big_id, 1.0, 1.0, big_cc, None),
693        };
694
695        let encoded = frame.encode();
696        assert_eq!(encoded.len(), 8, "should fall back to header-only frame");
697
698        let decoded_err = DiscoveryFrame::decode(&encoded).unwrap_err();
699        assert!(decoded_err.contains("payload too short") || decoded_err.contains("truncated"));
700    }
701
702    #[test]
703    fn handles_gpu_name_field() {
704        let frame = DiscoveryFrame {
705            kind: FrameKind::Discovery,
706            node: NodeResources::new(
707                "gpu-node-1",
708                24.0,
709                64.0,
710                "9.0",
711                Some("NVIDIA GeForce RTX 4090".to_string()),
712            ),
713        };
714
715        let encoded = frame.encode();
716        let decoded = DiscoveryFrame::decode(&encoded).unwrap();
717
718        assert_eq!(
719            decoded.node.gpu_name,
720            Some("NVIDIA GeForce RTX 4090".to_string())
721        );
722    }
723
724    // ====================================================================
725    // PROPERTY-BASED TESTS (proptest)
726    // ====================================================================
727
728    #[cfg(test)]
729    mod proptest_protocol {
730        use super::super::*;
731        use proptest::prelude::*;
732
733        proptest! {
734            /// Property: Any valid discovery frame should round-trip encode/decode
735            #[test]
736            fn prop_discovery_frame_round_trip(
737                node_id in "[a-z0-9]{1,12}",
738                vram in 1.0f32..512.0,
739                ram in 8.0f32..1024.0,
740            ) {
741                let frame = DiscoveryFrame {
742                    kind: FrameKind::Discovery,
743                    node: NodeResources::new(format!("node-{node_id}"), vram, ram, "8.9", None),
744                };
745
746                let encoded = frame.encode();
747                let decoded = DiscoveryFrame::decode(&encoded).expect("round-trip failed");
748
749                assert_eq!(decoded.node.id, frame.node.id);
750                assert!((decoded.node.vram_gb - frame.node.vram_gb).abs() < 0.01);
751                assert!((decoded.node.system_memory_gb - frame.node.system_memory_gb).abs() < 0.01);
752            }
753
754            /// Property: CRC should detect corruption with high probability
755            #[test]
756            fn prop_crc_detects_any_payload_corruption(
757                seed in 0u8..255,
758                corruption_pos in 0usize..256,
759            ) {
760                let frame = DiscoveryFrame {
761                    kind: FrameKind::Join,
762                    node: NodeResources::new("node-test", 24.0, 64.0, "8.9", None),
763                };
764
765                let mut encoded = frame.encode();
766                if corruption_pos < encoded.len() && corruption_pos < 200 {
767                    // Corrupt payload (not header/CRC area)
768                    if corruption_pos > 10 && corruption_pos < encoded.len().saturating_sub(4) {
769                        encoded[corruption_pos] = encoded[corruption_pos].wrapping_add(seed.wrapping_add(1));
770
771                        // Should fail CRC check
772                        let result = DiscoveryFrame::decode(&encoded);
773                        assert!(result.is_err(), "Corruption at byte {corruption_pos} should be detected");
774                    }
775                }
776            }
777
778            /// Property: Frame encoding is deterministic
779            #[test]
780            fn prop_frame_encoding_is_deterministic(
781                node_id in "[a-z0-9]{1,8}",
782                vram in 1.0f32..100.0,
783            ) {
784                let frame1 = DiscoveryFrame {
785                    kind: FrameKind::Discovery,
786                    node: NodeResources::new(format!("node-{node_id}"), vram, 64.0, "8.9", None),
787                };
788
789                let frame2 = DiscoveryFrame {
790                    kind: FrameKind::Discovery,
791                    node: NodeResources::new(format!("node-{node_id}"), vram, 64.0, "8.9", None),
792                };
793
794                let enc1 = frame1.encode();
795                let enc2 = frame2.encode();
796
797                // Same input should produce same output
798                assert_eq!(enc1, enc2, "Encoding should be deterministic");
799            }
800        }
801    }
802}