1use crc32fast::Hasher;
10
11pub const GHOSTLINK_ETHERTYPE: u16 = 0x88B5;
13
14pub const PROTOCOL_VERSION: u8 = 1;
16
17pub const MAX_PAYLOAD_SIZE: usize = 256;
19
20pub const MAX_FRAME_SIZE: usize = 8 + MAX_PAYLOAD_SIZE;
22
23#[derive(Clone, Copy, Debug)]
25pub struct FrameHeader {
26 pub ether_type: u16,
28 pub kind: u8,
30 pub version: u8,
32 pub crc: u32,
34}
35
36impl FrameHeader {
37 const HEADER_SIZE: usize = 8;
38
39 #[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 #[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 #[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..2].try_into().ok()?);
75 let kind = buf[2];
76 let version = buf[3];
77 let crc = u32::from_le_bytes(buf[4..8].try_into().ok()?);
78 Some(Self {
79 ether_type,
80 kind,
81 version,
82 crc,
83 })
84 }
85}
86
87#[inline]
89pub fn crc32(data: &[u8]) -> u32 {
90 let mut hasher = Hasher::new();
91 hasher.update(data);
92 hasher.finalize()
93}
94
95#[derive(Clone, Copy, Debug, PartialEq, Eq)]
97pub enum FrameKind {
98 Discovery = 1,
100 Join = 2,
102 Attestation = 3,
104 HealthCheck = 4,
106 ResourceAdvert = 5,
108}
109
110impl FrameKind {
111 const fn as_u8(&self) -> u8 {
112 match self {
113 Self::Discovery => 1,
114 Self::Join => 2,
115 Self::Attestation => 3,
116 Self::HealthCheck => 4,
117 Self::ResourceAdvert => 5,
118 }
119 }
120}
121
122impl TryFrom<u8> for FrameKind {
123 type Error = &'static str;
124
125 fn try_from(value: u8) -> Result<Self, Self::Error> {
126 match value {
127 1 => Ok(Self::Discovery),
128 2 => Ok(Self::Join),
129 3 => Ok(Self::Attestation),
130 4 => Ok(Self::HealthCheck),
131 5 => Ok(Self::ResourceAdvert),
132 _ => Err("unknown frame kind"),
133 }
134 }
135}
136
137#[derive(Clone, Debug, Default)]
139pub struct NodeResources {
140 pub id: String,
142 pub vram_gb: f32,
144 pub system_memory_gb: f32,
146 pub compute_capability: String,
148 pub gpu_name: Option<String>,
150 pub rpc_port: Option<u16>,
157}
158
159impl NodeResources {
160 pub fn new(
162 id: impl Into<String>,
163 vram_gb: f32,
164 system_memory_gb: f32,
165 compute_capability: impl Into<String>,
166 gpu_name: Option<String>,
167 ) -> Self {
168 Self {
169 id: id.into(),
170 vram_gb,
171 system_memory_gb,
172 compute_capability: compute_capability.into(),
173 gpu_name,
174 rpc_port: None,
175 }
176 }
177
178 pub fn with_rpc_port(mut self, port: u16) -> Self {
182 self.rpc_port = Some(port);
183 self
184 }
185
186 #[inline]
190 pub fn encode_payload_into(
191 &self,
192 buffer: &mut Vec<u8>,
193 max_size: usize,
194 ) -> Result<usize, &'static str> {
195 let id_bytes = self.id.as_bytes();
196 let cc_bytes = self.compute_capability.as_bytes();
197 let gpu_bytes = self.gpu_name.as_ref().map(|name| name.as_bytes());
198
199 if id_bytes.len() > u8::MAX as usize || cc_bytes.len() > u8::MAX as usize {
200 return Err("field length exceeds u8::MAX");
201 }
202 if let Some(gpu_bytes) = gpu_bytes {
203 if gpu_bytes.len() > u8::MAX as usize {
204 return Err("GPU name length exceeds u8::MAX");
205 }
206 }
207
208 let payload_len =
218 11 + id_bytes.len() + cc_bytes.len() + gpu_bytes.map_or(0, |bytes| 1 + bytes.len()) + 2;
219 if payload_len > max_size {
220 return Err("payload length exceeds max_size");
221 }
222
223 buffer.push(id_bytes.len() as u8);
224 buffer.extend_from_slice(id_bytes);
225 buffer.extend_from_slice(&self.vram_gb.to_le_bytes());
226 buffer.extend_from_slice(&self.system_memory_gb.to_le_bytes());
227 buffer.push(cc_bytes.len() as u8);
228 buffer.extend_from_slice(cc_bytes);
229
230 if let Some(gpu_bytes) = gpu_bytes {
231 buffer.push(1);
232 buffer.push(gpu_bytes.len() as u8);
233 buffer.extend_from_slice(gpu_bytes);
234 } else {
235 buffer.push(0);
236 }
237
238 buffer.extend_from_slice(&self.rpc_port.unwrap_or(0).to_le_bytes());
239
240 Ok(payload_len)
241 }
242
243 #[inline]
247 pub fn encode_payload(&self, max_size: usize) -> Vec<u8> {
248 let id_bytes_len = self.id.len();
249 let cc_bytes_len = self.compute_capability.len();
250 let gpu_bytes_len = self.gpu_name.as_ref().map_or(0, |name| name.len());
251 let est_len = 11
252 + id_bytes_len
253 + cc_bytes_len
254 + if gpu_bytes_len > 0 {
255 2 + gpu_bytes_len
256 } else {
257 0
258 }
259 + 2;
260
261 let mut payload = Vec::with_capacity(est_len);
262 if self.encode_payload_into(&mut payload, max_size).is_ok() {
263 payload
264 } else {
265 Vec::new()
266 }
267 }
268
269 pub fn decode_payload(payload: &[u8]) -> Result<Self, String> {
271 let len = payload.len();
272 if len < 11 {
273 return Err("payload too short".into());
274 }
275
276 let mut cursor = 0usize;
277
278 let id_len = payload[cursor] as usize;
279 cursor += 1;
280 let id_end = cursor + id_len;
281 if id_end > len {
282 return Err("payload truncated".into());
283 }
284 let id = std::str::from_utf8(&payload[cursor..id_end])
287 .map_err(|_| "invalid UTF-8 in ID".to_string())?
288 .to_string();
289 cursor = id_end;
290
291 let vram_end = cursor + 8;
292 if vram_end > len {
293 return Err("payload truncated".into());
294 }
295 let vram_gb = f32::from_le_bytes(
298 payload[cursor..cursor + 4]
299 .try_into()
300 .map_err(|_| "vram parsing failed".to_string())?,
301 );
302 let system_memory_gb = f32::from_le_bytes(
303 payload[cursor + 4..cursor + 8]
304 .try_into()
305 .map_err(|_| "system memory parsing failed".to_string())?,
306 );
307 cursor = vram_end;
308
309 if cursor >= len {
310 return Err("payload truncated".into());
311 }
312 let cc_len = payload[cursor] as usize;
313 cursor += 1;
314 let cc_end = cursor + cc_len;
315 if cc_end > len {
316 return Err("payload truncated".into());
317 }
318 let compute_capability = std::str::from_utf8(&payload[cursor..cc_end])
320 .map_err(|_| "invalid UTF-8 in CC".to_string())?
321 .to_string();
322 cursor = cc_end;
323
324 if cursor >= len {
325 return Err("payload truncated".into());
326 }
327 let has_gpu = payload[cursor] == 1;
328 cursor += 1;
329
330 let gpu_name = if has_gpu {
331 if cursor >= len {
332 return Err("payload truncated".into());
333 }
334 let gpu_len = payload[cursor] as usize;
335 cursor += 1;
336 let gpu_end = cursor + gpu_len;
337 if gpu_end > len {
338 return Err("payload truncated".into());
339 }
340 let name = std::str::from_utf8(&payload[cursor..gpu_end])
342 .map_err(|_| "invalid UTF-8 in GPU name".to_string())?
343 .to_string();
344 cursor = gpu_end;
349 Some(name)
350 } else {
351 None
352 };
353
354 let rpc_port = if cursor + 2 <= len {
360 let port = u16::from_le_bytes(
361 payload[cursor..cursor + 2]
362 .try_into()
363 .map_err(|_| "rpc_port parsing failed".to_string())?,
364 );
365 if port == 0 {
366 None
367 } else {
368 Some(port)
369 }
370 } else {
371 None
372 };
373
374 Ok(Self {
375 id,
376 vram_gb,
377 system_memory_gb,
378 compute_capability,
379 gpu_name,
380 rpc_port,
381 })
382 }
383}
384
385#[derive(Clone, Debug)]
387pub struct DiscoveryFrame {
388 pub kind: FrameKind,
389 pub node: NodeResources,
390}
391
392impl DiscoveryFrame {
393 #[inline]
396 pub fn encode(&self) -> Vec<u8> {
397 let mut buf = [0u8; MAX_FRAME_SIZE];
398 let et = GHOSTLINK_ETHERTYPE.to_le_bytes();
399 buf[0] = et[0];
400 buf[1] = et[1];
401 buf[2] = self.kind.as_u8();
402 buf[3] = PROTOCOL_VERSION;
403
404 let id_bytes = self.node.id.as_bytes();
405 let cc_bytes = self.node.compute_capability.as_bytes();
406
407 if id_bytes.len() > u8::MAX as usize || cc_bytes.len() > u8::MAX as usize {
408 buf[4..8].copy_from_slice(&crc32(&[]).to_le_bytes());
409 return buf[..8].to_vec();
410 }
411 if let Some(ref gpu_name) = self.node.gpu_name {
412 if gpu_name.len() > u8::MAX as usize {
413 buf[4..8].copy_from_slice(&crc32(&[]).to_le_bytes());
414 return buf[..8].to_vec();
415 }
416 }
417
418 let gpu_len = self.node.gpu_name.as_ref().map(|name| name.len());
423 let payload_len =
431 11 + id_bytes.len() + cc_bytes.len() + gpu_len.map_or(0, |len| 1 + len) + 2;
432 if payload_len > MAX_PAYLOAD_SIZE {
433 buf[4..8].copy_from_slice(&crc32(&[]).to_le_bytes());
434 return buf[..8].to_vec();
435 }
436
437 let mut pos = 8usize;
438
439 buf[pos] = id_bytes.len() as u8;
440 pos += 1;
441 buf[pos..pos + id_bytes.len()].copy_from_slice(id_bytes);
442 pos += id_bytes.len();
443
444 buf[pos..pos + 4].copy_from_slice(&self.node.vram_gb.to_le_bytes());
445 pos += 4;
446
447 buf[pos..pos + 4].copy_from_slice(&self.node.system_memory_gb.to_le_bytes());
448 pos += 4;
449
450 buf[pos] = cc_bytes.len() as u8;
451 pos += 1;
452 buf[pos..pos + cc_bytes.len()].copy_from_slice(cc_bytes);
453 pos += cc_bytes.len();
454
455 if let Some(ref gpu_name) = self.node.gpu_name {
456 let gpu_bytes = gpu_name.as_bytes();
457 buf[pos] = 1;
458 pos += 1;
459 buf[pos] = gpu_bytes.len() as u8;
460 pos += 1;
461 buf[pos..pos + gpu_bytes.len()].copy_from_slice(gpu_bytes);
462 pos += gpu_bytes.len();
463 } else {
464 buf[pos] = 0;
465 pos += 1;
466 }
467
468 let rpc_port_bytes = self.node.rpc_port.unwrap_or(0).to_le_bytes();
469 buf[pos..pos + 2].copy_from_slice(&rpc_port_bytes);
470 pos += 2;
471
472 let crc = crc32(&buf[8..pos]);
473 buf[4..8].copy_from_slice(&crc.to_le_bytes());
474
475 buf[..pos].to_vec()
476 }
477
478 #[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 header.encode_into(unsafe { &mut *(buf.as_mut_ptr() as *mut [u8; 8]) });
496
497 buf.len()
498 } else {
499 buf.clear();
500 buf.extend_from_slice(&[0u8; 8]);
501 let header = FrameHeader {
502 ether_type: GHOSTLINK_ETHERTYPE,
503 kind: self.kind.as_u8(),
504 version: PROTOCOL_VERSION,
505 crc: crc32(&[]),
506 };
507 header.encode_into(unsafe { &mut *(buf.as_mut_ptr() as *mut [u8; 8]) });
508 buf.len()
509 }
510 }
511
512 #[inline]
514 pub fn decode(bytes: &[u8]) -> Result<Self, String> {
515 if bytes.len() < FrameHeader::HEADER_SIZE {
516 return Err("frame too short".into());
517 }
518
519 let header = FrameHeader::decode(bytes).ok_or("invalid header")?;
521
522 if header.ether_type != GHOSTLINK_ETHERTYPE {
523 return Err(format!("unexpected EtherType 0x{:04x}", header.ether_type));
524 }
525
526 if header.version != PROTOCOL_VERSION {
527 return Err(format!("unsupported protocol version {}", header.version));
528 }
529
530 let payload = &bytes[FrameHeader::HEADER_SIZE..];
532 let computed_crc = crc32(payload);
533
534 if computed_crc != header.crc {
535 return Err(format!(
536 "CRC mismatch: expected 0x{:08x}, got 0x{:08x}",
537 header.crc, computed_crc
538 ));
539 }
540
541 let node = NodeResources::decode_payload(payload)?;
543
544 Ok(Self {
545 kind: FrameKind::try_from(header.kind)?,
546 node,
547 })
548 }
549}
550
551#[cfg(test)]
552mod tests {
553 use super::*;
554
555 #[test]
556 fn discovery_frames_round_trip() {
557 let frame = DiscoveryFrame {
558 kind: FrameKind::Join,
559 node: NodeResources::new("node-b", 12.0, 32.0, "8.6", None),
560 };
561
562 let encoded = frame.encode();
563 let decoded = DiscoveryFrame::decode(&encoded).unwrap();
564
565 assert_eq!(decoded.kind, frame.kind);
566 assert_eq!(decoded.node.id, frame.node.id);
567 assert_eq!(decoded.node.vram_gb, frame.node.vram_gb);
568 }
569
570 #[test]
571 fn discovery_frame_encode_round_trips_rpc_port() {
572 let frame = DiscoveryFrame {
582 kind: FrameKind::Join,
583 node: NodeResources::new("node-c", 12.0, 32.0, "8.6", None).with_rpc_port(50052),
584 };
585
586 let encoded = frame.encode();
587 let decoded = DiscoveryFrame::decode(&encoded).unwrap();
588
589 assert_eq!(decoded.node.rpc_port, Some(50052));
590 }
591
592 #[test]
593 fn discovery_frame_encode_without_rpc_port_decodes_to_none() {
594 let frame = DiscoveryFrame {
595 kind: FrameKind::Join,
596 node: NodeResources::new("node-d", 8.0, 16.0, "7.5", None),
597 };
598
599 let encoded = frame.encode();
600 let decoded = DiscoveryFrame::decode(&encoded).unwrap();
601
602 assert_eq!(decoded.node.rpc_port, None);
603 }
604
605 #[test]
606 fn crc_verification_fails_on_modified_payload() {
607 let frame = DiscoveryFrame {
608 kind: FrameKind::Discovery,
609 node: NodeResources::new("node-a", 24.0, 64.0, "8.9", Some("RTX4090".to_string())),
610 };
611
612 let encoded = frame.encode();
613 let mut modified = encoded.clone();
614 modified[10] ^= 0xFF; let result = DiscoveryFrame::decode(&modified);
617 assert!(result.is_err());
618 assert!(result.unwrap_err().contains("CRC mismatch"));
619 }
620
621 #[test]
622 fn rejects_wrong_ether_type() {
623 let mut fake_frame = vec![0u8; 10];
624 fake_frame[0] = 0xB5u8; fake_frame[1] = 0xFF; let result = DiscoveryFrame::decode(&fake_frame);
628 assert!(result.is_err());
629 }
630
631 #[test]
632 fn encode_payload_into_accepts_exact_max_size_with_gpu_name() {
633 let node = NodeResources::new("a", 1.0, 1.0, "b", Some("c".to_string()));
637 let mut buffer = Vec::new();
638 let written = node
641 .encode_payload_into(&mut buffer, 17)
642 .expect("payload of exactly 17 bytes must fit within max_size 17");
643 assert_eq!(written, 17);
644 assert_eq!(buffer.len(), 17);
645 }
646
647 #[test]
648 fn node_resources_round_trip_preserves_rpc_port() {
649 let node = NodeResources::new("node-e", 12.0, 32.0, "8.6", None).with_rpc_port(50052);
650 let encoded = node.encode_payload(64);
651 let decoded = NodeResources::decode_payload(&encoded).expect("decode");
652 assert_eq!(decoded.rpc_port, Some(50052));
653 }
654
655 #[test]
656 fn node_resources_round_trip_without_rpc_port_is_none() {
657 let node = NodeResources::new("node-f", 8.0, 16.0, "7.5", None);
658 let encoded = node.encode_payload(64);
659 let decoded = NodeResources::decode_payload(&encoded).expect("decode");
660 assert_eq!(decoded.rpc_port, None);
661 }
662
663 #[test]
664 fn node_resources_round_trip_with_gpu_name_and_rpc_port_together() {
665 let node = NodeResources::new("node-g", 24.0, 64.0, "8.9", Some("RTX 4090".to_string()))
670 .with_rpc_port(50053);
671 let encoded = node.encode_payload(128);
672 let decoded = NodeResources::decode_payload(&encoded).expect("decode");
673 assert_eq!(decoded.gpu_name, Some("RTX 4090".to_string()));
674 assert_eq!(decoded.rpc_port, Some(50053));
675 }
676
677 #[test]
678 fn decode_payload_from_before_rpc_port_existed_defaults_to_none() {
679 let node = NodeResources::new("node-h", 8.0, 16.0, "7.5", None);
682 let mut encoded = node.encode_payload(64);
683 assert!(encoded.len() >= 2);
684 encoded.truncate(encoded.len() - 2);
685 let decoded = NodeResources::decode_payload(&encoded).expect("decode");
686 assert_eq!(decoded.rpc_port, None);
687 }
688
689 #[test]
690 fn encode_does_not_panic_when_combined_field_lengths_exceed_max_payload() {
691 let big_id = "x".repeat(200);
696 let big_cc = "y".repeat(200);
697 let frame = DiscoveryFrame {
698 kind: FrameKind::Discovery,
699 node: NodeResources::new(big_id, 1.0, 1.0, big_cc, None),
700 };
701
702 let encoded = frame.encode();
703 assert_eq!(encoded.len(), 8, "should fall back to header-only frame");
704
705 let decoded_err = DiscoveryFrame::decode(&encoded).unwrap_err();
706 assert!(decoded_err.contains("payload too short") || decoded_err.contains("truncated"));
707 }
708
709 #[test]
710 fn handles_gpu_name_field() {
711 let frame = DiscoveryFrame {
712 kind: FrameKind::Discovery,
713 node: NodeResources::new(
714 "gpu-node-1",
715 24.0,
716 64.0,
717 "9.0",
718 Some("NVIDIA GeForce RTX 4090".to_string()),
719 ),
720 };
721
722 let encoded = frame.encode();
723 let decoded = DiscoveryFrame::decode(&encoded).unwrap();
724
725 assert_eq!(
726 decoded.node.gpu_name,
727 Some("NVIDIA GeForce RTX 4090".to_string())
728 );
729 }
730
731 #[cfg(test)]
736 mod proptest_protocol {
737 use super::super::*;
738 use proptest::prelude::*;
739
740 proptest! {
741 #[test]
743 fn prop_discovery_frame_round_trip(
744 node_id in "[a-z0-9]{1,12}",
745 vram in 1.0f32..512.0,
746 ram in 8.0f32..1024.0,
747 ) {
748 let frame = DiscoveryFrame {
749 kind: FrameKind::Discovery,
750 node: NodeResources::new(format!("node-{node_id}"), vram, ram, "8.9", None),
751 };
752
753 let encoded = frame.encode();
754 let decoded = DiscoveryFrame::decode(&encoded).expect("round-trip failed");
755
756 assert_eq!(decoded.node.id, frame.node.id);
757 assert!((decoded.node.vram_gb - frame.node.vram_gb).abs() < 0.01);
758 assert!((decoded.node.system_memory_gb - frame.node.system_memory_gb).abs() < 0.01);
759 }
760
761 #[test]
763 fn prop_crc_detects_any_payload_corruption(
764 seed in 0u8..255,
765 corruption_pos in 0usize..256,
766 ) {
767 let frame = DiscoveryFrame {
768 kind: FrameKind::Join,
769 node: NodeResources::new("node-test", 24.0, 64.0, "8.9", None),
770 };
771
772 let mut encoded = frame.encode();
773 if corruption_pos < encoded.len() && corruption_pos < 200 {
774 if corruption_pos > 10 && corruption_pos < encoded.len().saturating_sub(4) {
776 encoded[corruption_pos] = encoded[corruption_pos].wrapping_add(seed.wrapping_add(1));
777
778 let result = DiscoveryFrame::decode(&encoded);
780 assert!(result.is_err(), "Corruption at byte {corruption_pos} should be detected");
781 }
782 }
783 }
784
785 #[test]
787 fn prop_frame_encoding_is_deterministic(
788 node_id in "[a-z0-9]{1,8}",
789 vram in 1.0f32..100.0,
790 ) {
791 let frame1 = DiscoveryFrame {
792 kind: FrameKind::Discovery,
793 node: NodeResources::new(format!("node-{node_id}"), vram, 64.0, "8.9", None),
794 };
795
796 let frame2 = DiscoveryFrame {
797 kind: FrameKind::Discovery,
798 node: NodeResources::new(format!("node-{node_id}"), vram, 64.0, "8.9", None),
799 };
800
801 let enc1 = frame1.encode();
802 let enc2 = frame2.encode();
803
804 assert_eq!(enc1, enc2, "Encoding should be deterministic");
806 }
807 }
808 }
809}