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

ghostlink_core/
protocol.rs
