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

ghostlink_core/
protocol.rs
