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