1use bytes::Bytes;
57use bytes::BytesMut;
58use serde::Deserialize;
59use serde::Serialize;
60use serde::de::DeserializeOwned;
61use serde_multipart::Part;
62
63use crate::attrs::AttrValue;
64use crate::attrs::Attrs;
65use crate::attrs::Key;
66
67const HEADER_SIZE: usize = 2;
69
70const ENTRY_HEADER_SIZE: usize = 12;
72
73#[derive(Clone)]
79pub struct Flattrs {
80 buffer: BytesMut,
84}
85
86impl Default for Flattrs {
87 fn default() -> Self {
88 Self::new()
89 }
90}
91
92#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
94#[non_exhaustive]
95pub enum FlattrsValidationError {
96 #[error("flattrs buffer is missing its entry-count header")]
98 MissingHeader,
99
100 #[error("flattrs declares {declared} entries but contains {parsed} complete entries")]
102 EntryCountMismatch {
103 declared: usize,
105 parsed: usize,
107 },
108
109 #[error("flattrs buffer has {trailing} trailing bytes")]
111 TrailingBytes {
112 trailing: usize,
114 },
115}
116
117impl Flattrs {
118 pub fn new() -> Self {
120 let mut buffer = BytesMut::with_capacity(HEADER_SIZE);
121 buffer.extend_from_slice(&0u16.to_le_bytes());
122 Self { buffer }
123 }
124
125 pub fn from_part(part: Part) -> Self {
127 Self {
128 buffer: BytesMut::from(part.into_bytes().as_ref()),
129 }
130 }
131
132 pub fn to_part(&self) -> Part {
136 Part::from(Bytes::copy_from_slice(&self.buffer))
137 }
138
139 pub fn encoded_len(&self) -> usize {
141 self.buffer.len()
142 }
143
144 pub fn validate_wire(&self) -> Result<(), FlattrsValidationError> {
150 if self.buffer.len() < HEADER_SIZE {
151 return Err(FlattrsValidationError::MissingHeader);
152 }
153
154 let declared = self.len();
155 let mut parsed = 0;
156 let mut parsed_bytes = HEADER_SIZE;
157 for (_, value) in self.iter() {
158 parsed += 1;
159 parsed_bytes += ENTRY_HEADER_SIZE + value.len();
160 }
161 if parsed != declared {
162 return Err(FlattrsValidationError::EntryCountMismatch { declared, parsed });
163 }
164 if parsed_bytes != self.buffer.len() {
165 return Err(FlattrsValidationError::TrailingBytes {
166 trailing: self.buffer.len().saturating_sub(parsed_bytes),
167 });
168 }
169 Ok(())
170 }
171
172 pub fn set<T: Serialize>(&mut self, key: Key<T>, value: T) {
178 let key_hash = key.key_hash();
179 let serialized = bincode::serde::encode_to_vec(&value, bincode::config::legacy())
180 .expect("serialization failed");
181 self.set_serialized(key_hash, &serialized);
182 }
183
184 pub fn set_serialized(&mut self, key_hash: u64, serialized: &[u8]) {
199 if let Some((offset, old_len)) = self.find_entry_location(key_hash) {
201 if serialized.len() == old_len {
202 let value_start = offset + ENTRY_HEADER_SIZE;
204 self.buffer[value_start..value_start + old_len].copy_from_slice(serialized);
205 return;
206 }
207
208 let entry_size = ENTRY_HEADER_SIZE + old_len;
210 let end = offset + entry_size;
211
212 if end < self.buffer.len() {
213 self.buffer.copy_within(end.., offset);
214 }
215 self.buffer.truncate(self.buffer.len() - entry_size);
216
217 let count = self.len();
219 self.buffer[0..2].copy_from_slice(&((count - 1) as u16).to_le_bytes());
220 }
221
222 self.append_entry(key_hash, serialized);
223 }
224
225 pub fn get<T: AttrValue + DeserializeOwned>(&self, key: Key<T>) -> Option<T> {
230 let key_hash = key.key_hash();
231 let value_bytes = self.find_value(key_hash)?;
232 bincode::serde::decode_from_slice(value_bytes, bincode::config::legacy())
233 .map(|(v, _)| v)
234 .ok()
235 }
236
237 #[inline]
239 pub fn contains_key<T>(&self, key: Key<T>) -> bool {
240 self.find_value(key.key_hash()).is_some()
241 }
242
243 #[inline]
245 pub fn is_empty(&self) -> bool {
246 self.len() == 0
247 }
248
249 #[inline]
251 pub fn len(&self) -> usize {
252 if self.buffer.len() < HEADER_SIZE {
253 return 0;
254 }
255 u16::from_le_bytes([self.buffer[0], self.buffer[1]]) as usize
256 }
257
258 pub fn from_attrs(attrs: &Attrs) -> Self {
260 let mut flattrs = Self::new();
261 for (name, value) in attrs.iter() {
262 let key_hash = crate::attrs::fnv1a_hash(name.as_bytes());
263 let serialized = value.serialize_bincode();
264 flattrs.append_entry(key_hash, &serialized);
265 }
266 flattrs
267 }
268
269 pub fn iter(&self) -> FlattrsIter<'_> {
278 FlattrsIter {
279 buffer: &self.buffer,
280 remaining: self.len(),
281 offset: HEADER_SIZE,
282 }
283 }
284
285 fn find_value(&self, key_hash: u64) -> Option<&[u8]> {
287 if self.buffer.len() < HEADER_SIZE {
288 return None;
289 }
290
291 let num_entries = u16::from_le_bytes([self.buffer[0], self.buffer[1]]) as usize;
292 let mut offset = HEADER_SIZE;
293
294 for _ in 0..num_entries {
295 if offset + ENTRY_HEADER_SIZE > self.buffer.len() {
296 return None;
297 }
298
299 let entry_key_hash =
300 u64::from_le_bytes(self.buffer[offset..offset + 8].try_into().unwrap_or([0; 8]));
301 let entry_len = u32::from_le_bytes(
302 self.buffer[offset + 8..offset + 12]
303 .try_into()
304 .unwrap_or([0; 4]),
305 ) as usize;
306
307 let value_start = offset + ENTRY_HEADER_SIZE;
308 let value_end = value_start + entry_len;
309
310 if value_end > self.buffer.len() {
311 return None;
312 }
313
314 if entry_key_hash == key_hash {
315 return Some(&self.buffer[value_start..value_end]);
316 }
317
318 offset = value_end;
319 }
320
321 None
322 }
323
324 fn find_entry_location(&self, key_hash: u64) -> Option<(usize, usize)> {
326 if self.buffer.len() < HEADER_SIZE {
327 return None;
328 }
329
330 let num_entries = u16::from_le_bytes([self.buffer[0], self.buffer[1]]) as usize;
331 let mut offset = HEADER_SIZE;
332
333 for _ in 0..num_entries {
334 if offset + ENTRY_HEADER_SIZE > self.buffer.len() {
335 return None;
336 }
337
338 let entry_key_hash =
339 u64::from_le_bytes(self.buffer[offset..offset + 8].try_into().unwrap_or([0; 8]));
340 let entry_len = u32::from_le_bytes(
341 self.buffer[offset + 8..offset + 12]
342 .try_into()
343 .unwrap_or([0; 4]),
344 ) as usize;
345
346 if entry_key_hash == key_hash {
347 return Some((offset, entry_len));
348 }
349
350 offset += ENTRY_HEADER_SIZE + entry_len;
351 }
352
353 None
354 }
355
356 fn append_entry(&mut self, key_hash: u64, value: &[u8]) {
358 let len = self.len();
359 self.buffer[0..2].copy_from_slice(&((len + 1) as u16).to_le_bytes());
360
361 self.buffer.extend_from_slice(&key_hash.to_le_bytes());
363 self.buffer
364 .extend_from_slice(&(value.len() as u32).to_le_bytes());
365 self.buffer.extend_from_slice(value);
366 }
367}
368
369pub struct FlattrsIter<'a> {
373 buffer: &'a [u8],
374 remaining: usize,
375 offset: usize,
376}
377
378impl<'a> Iterator for FlattrsIter<'a> {
379 type Item = (u64, &'a [u8]);
380
381 fn next(&mut self) -> Option<Self::Item> {
382 if self.remaining == 0 {
383 return None;
384 }
385 if self.offset + ENTRY_HEADER_SIZE > self.buffer.len() {
386 return None;
387 }
388 let key_hash = u64::from_le_bytes(
389 self.buffer[self.offset..self.offset + 8]
390 .try_into()
391 .unwrap_or([0; 8]),
392 );
393 let entry_len = u32::from_le_bytes(
394 self.buffer[self.offset + 8..self.offset + 12]
395 .try_into()
396 .unwrap_or([0; 4]),
397 ) as usize;
398 let value_start = self.offset + ENTRY_HEADER_SIZE;
399 let value_end = value_start + entry_len;
400 if value_end > self.buffer.len() {
401 return None;
402 }
403 let value = &self.buffer[value_start..value_end];
404 self.offset = value_end;
405 self.remaining -= 1;
406 Some((key_hash, value))
407 }
408}
409
410impl From<Attrs> for Flattrs {
411 fn from(attrs: Attrs) -> Self {
412 Self::from_attrs(&attrs)
413 }
414}
415
416impl From<&Attrs> for Flattrs {
417 fn from(attrs: &Attrs) -> Self {
418 Self::from_attrs(attrs)
419 }
420}
421
422impl std::fmt::Debug for Flattrs {
423 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
424 f.debug_struct("Flattrs").field("len", &self.len()).finish()
425 }
426}
427
428impl PartialEq for Flattrs {
431 fn eq(&self, other: &Self) -> bool {
432 self.len() == other.len()
433 && self
434 .iter()
435 .all(|(key_hash, value)| other.find_value(key_hash) == Some(value))
436 }
437}
438
439impl Eq for Flattrs {}
440
441impl std::fmt::Display for Flattrs {
442 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
443 use crate::attrs::lookup_key_info;
444
445 let mut offset = HEADER_SIZE;
446 let mut first = true;
447
448 for _ in 0..self.len() {
449 let key_hash = u64::from_le_bytes(self.buffer[offset..offset + 8].try_into().unwrap());
450 let entry_len =
451 u32::from_le_bytes(self.buffer[offset + 8..offset + 12].try_into().unwrap())
452 as usize;
453 let value_bytes = &self.buffer[offset + ENTRY_HEADER_SIZE..][..entry_len];
454
455 if !first {
456 write!(f, ",")?;
457 }
458 first = false;
459
460 let info =
461 lookup_key_info(key_hash).expect("key should be registered via declare_attrs!");
462
463 let value = (info.deserialize_bincode)(value_bytes).expect("value should deserialize");
464 write!(f, "{}={}", info.name, (info.display)(value.as_ref()))?;
465
466 offset += ENTRY_HEADER_SIZE + entry_len;
467 }
468
469 Ok(())
470 }
471}
472
473impl Serialize for Flattrs {
474 fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
475 where
476 S: serde::Serializer,
477 {
478 self.to_part().serialize(serializer)
479 }
480}
481
482impl<'de> Deserialize<'de> for Flattrs {
483 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
484 where
485 D: serde::Deserializer<'de>,
486 {
487 let part: Part = Deserialize::deserialize(deserializer)?;
488 Ok(Self::from_part(part))
489 }
490}
491
492#[cfg(test)]
493mod tests {
494 use super::*;
495 use crate::attrs::declare_attrs;
496
497 declare_attrs! {
498 attr TEST_U64: u64;
499 attr TEST_STRING: String;
500 attr TEST_BOOL: bool;
501 }
502
503 #[test]
504 fn test_basic_set_get() {
505 let mut attrs = Flattrs::new();
506
507 attrs.set(TEST_U64, 42u64);
508 attrs.set(TEST_STRING, "hello".to_string());
509 attrs.set(TEST_BOOL, true);
510
511 assert_eq!(attrs.get(TEST_U64), Some(42u64));
512 assert_eq!(attrs.get(TEST_STRING), Some("hello".to_string()));
513 assert_eq!(attrs.get(TEST_BOOL), Some(true));
514 }
515
516 #[test]
517 fn test_missing_key() {
518 let attrs = Flattrs::new();
519 assert_eq!(attrs.get::<u64>(TEST_U64), None);
520 }
521
522 #[test]
523 fn equality_ignores_entry_order() {
524 let mut first = Flattrs::new();
525 first.set(TEST_U64, 42u64);
526 first.set(TEST_BOOL, true);
527 let mut second = Flattrs::new();
528 second.set(TEST_BOOL, true);
529 second.set(TEST_U64, 42u64);
530 assert_eq!(first, second, "the same entries in a different order");
531
532 second.set(TEST_U64, 43u64);
533 assert_ne!(first, second, "a differing value");
534 let mut subset = Flattrs::new();
535 subset.set(TEST_U64, 42u64);
536 assert_ne!(first, subset, "a missing entry");
537 }
538
539 #[test]
540 fn default_is_valid_and_mutable() {
541 let mut attrs = Flattrs::default();
542 assert_eq!(attrs.validate_wire(), Ok(()));
543 attrs.set(TEST_U64, 42u64);
544 assert_eq!(attrs.get(TEST_U64), Some(42u64));
545 }
546
547 #[test]
548 fn validate_wire_rejects_malformed_buffers() {
549 let mut valid = Flattrs::new();
550 valid.set(TEST_U64, 42u64);
551 assert_eq!(valid.validate_wire(), Ok(()));
552 assert_eq!(valid.encoded_len(), valid.to_part().len());
553
554 assert_eq!(
555 Flattrs::from_part(Part::from(Vec::<u8>::new())).validate_wire(),
556 Err(FlattrsValidationError::MissingHeader),
557 );
558 assert_eq!(
559 Flattrs::from_part(Part::from(vec![1, 0])).validate_wire(),
560 Err(FlattrsValidationError::EntryCountMismatch {
561 declared: 1,
562 parsed: 0,
563 }),
564 );
565 assert_eq!(
566 Flattrs::from_part(Part::from(vec![0, 0, 1])).validate_wire(),
567 Err(FlattrsValidationError::TrailingBytes { trailing: 1 }),
568 );
569
570 let mut truncated_value = 1u16.to_le_bytes().to_vec();
571 truncated_value.extend_from_slice(&TEST_U64.key_hash().to_le_bytes());
572 truncated_value.extend_from_slice(&1u32.to_le_bytes());
573 assert_eq!(
574 Flattrs::from_part(Part::from(truncated_value)).validate_wire(),
575 Err(FlattrsValidationError::EntryCountMismatch {
576 declared: 1,
577 parsed: 0,
578 }),
579 );
580
581 let mut truncated_second_entry = 2u16.to_le_bytes().to_vec();
582 truncated_second_entry.extend_from_slice(&TEST_U64.key_hash().to_le_bytes());
583 truncated_second_entry.extend_from_slice(&0u32.to_le_bytes());
584 truncated_second_entry.extend_from_slice(&[0; 4]);
585 assert_eq!(
586 Flattrs::from_part(Part::from(truncated_second_entry)).validate_wire(),
587 Err(FlattrsValidationError::EntryCountMismatch {
588 declared: 2,
589 parsed: 1,
590 }),
591 );
592 }
593
594 #[test]
595 fn test_set_replaces_existing() {
596 let mut attrs = Flattrs::new();
597 attrs.set(TEST_U64, 42u64);
598 attrs.set(TEST_U64, 100u64);
599 assert_eq!(attrs.get(TEST_U64), Some(100u64));
600 assert_eq!(attrs.len(), 1);
601 }
602
603 #[test]
604 fn test_set_replaces_different_size() {
605 let mut attrs = Flattrs::new();
606 attrs.set(TEST_STRING, "short".to_string());
607 attrs.set(TEST_STRING, "a much longer string".to_string());
608 assert_eq!(
609 attrs.get(TEST_STRING),
610 Some("a much longer string".to_string())
611 );
612 assert_eq!(attrs.len(), 1);
613 }
614
615 #[test]
618 fn test_set_serialized_overwrites_existing_same_size() {
619 let mut attrs = Flattrs::new();
620 let key_hash = TEST_U64.key_hash();
621 let first = bincode::serde::encode_to_vec(42u64, bincode::config::legacy()).unwrap();
622 attrs.set_serialized(key_hash, &first);
623 assert_eq!(attrs.get(TEST_U64), Some(42u64));
624 assert_eq!(attrs.len(), 1);
625
626 let second = bincode::serde::encode_to_vec(100u64, bincode::config::legacy()).unwrap();
627 assert_eq!(first.len(), second.len(), "same-size precondition");
628 attrs.set_serialized(key_hash, &second);
629
630 assert_eq!(attrs.get(TEST_U64), Some(100u64));
631 assert_eq!(attrs.len(), 1);
632 }
633
634 #[test]
637 fn test_set_serialized_overwrites_existing_different_size() {
638 let mut attrs = Flattrs::new();
639 let key_hash = TEST_STRING.key_hash();
640 let short =
641 bincode::serde::encode_to_vec("short".to_string(), bincode::config::legacy()).unwrap();
642 attrs.set_serialized(key_hash, &short);
643 assert_eq!(attrs.get(TEST_STRING), Some("short".to_string()));
644 assert_eq!(attrs.len(), 1);
645
646 let long = bincode::serde::encode_to_vec(
647 "a much longer string".to_string(),
648 bincode::config::legacy(),
649 )
650 .unwrap();
651 assert_ne!(short.len(), long.len(), "different-size precondition");
652 attrs.set_serialized(key_hash, &long);
653
654 assert_eq!(
655 attrs.get(TEST_STRING),
656 Some("a much longer string".to_string())
657 );
658 assert_eq!(attrs.len(), 1);
659 }
660
661 #[test]
664 fn test_set_serialized_interops_with_typed_set() {
665 let mut attrs = Flattrs::new();
666 attrs.set(TEST_U64, 1u64);
667 attrs.set_serialized(
668 TEST_U64.key_hash(),
669 &bincode::serde::encode_to_vec(2u64, bincode::config::legacy()).unwrap(),
670 );
671 assert_eq!(attrs.get(TEST_U64), Some(2u64));
672 assert_eq!(attrs.len(), 1);
673
674 let mut attrs = Flattrs::new();
675 attrs.set_serialized(
676 TEST_U64.key_hash(),
677 &bincode::serde::encode_to_vec(1u64, bincode::config::legacy()).unwrap(),
678 );
679 attrs.set(TEST_U64, 2u64);
680 assert_eq!(attrs.get(TEST_U64), Some(2u64));
681 assert_eq!(attrs.len(), 1);
682 }
683
684 #[test]
686 fn test_set_serialized_new_key_appends() {
687 let mut attrs = Flattrs::new();
688 attrs.set_serialized(
689 TEST_U64.key_hash(),
690 &bincode::serde::encode_to_vec(7u64, bincode::config::legacy()).unwrap(),
691 );
692 attrs.set_serialized(
693 TEST_STRING.key_hash(),
694 &bincode::serde::encode_to_vec("x".to_string(), bincode::config::legacy()).unwrap(),
695 );
696 assert_eq!(attrs.get(TEST_U64), Some(7u64));
697 assert_eq!(attrs.get(TEST_STRING), Some("x".to_string()));
698 assert_eq!(attrs.len(), 2);
699 }
700
701 #[test]
702 fn test_contains_key() {
703 let mut attrs = Flattrs::new();
704
705 assert!(!attrs.contains_key(TEST_U64));
706 attrs.set(TEST_U64, 42u64);
707 assert!(attrs.contains_key(TEST_U64));
708 }
709
710 #[test]
711 fn test_serde_roundtrip() {
712 let mut attrs = Flattrs::new();
713 attrs.set(TEST_U64, 42u64);
714 attrs.set(TEST_STRING, "hello".to_string());
715
716 let serialized =
717 bincode::serde::encode_to_vec(&attrs, bincode::config::legacy()).expect("serialize");
718 let deserialized: Flattrs =
719 bincode::serde::decode_from_slice(&serialized, bincode::config::legacy())
720 .map(|(v, _)| v)
721 .expect("deserialize");
722
723 assert_eq!(deserialized.get(TEST_U64), Some(42u64));
724 assert_eq!(deserialized.get(TEST_STRING), Some("hello".to_string()));
725 assert_eq!(deserialized.len(), 2);
726 }
727
728 #[test]
729 fn test_wire_roundtrip() {
730 let mut attrs = Flattrs::new();
731 attrs.set(TEST_U64, 42u64);
732 attrs.set(TEST_STRING, "hello".to_string());
733
734 let wire = attrs.to_part();
735 let received = Flattrs::from_part(wire);
736
737 assert_eq!(received.get(TEST_U64), Some(42u64));
738 assert_eq!(received.get(TEST_STRING), Some("hello".to_string()));
739 assert_eq!(received.len(), 2);
740 }
741
742 #[test]
743 fn test_multiple_keys() {
744 let mut attrs = Flattrs::new();
745 attrs.set(TEST_U64, 1u64);
746 attrs.set(TEST_STRING, "two".to_string());
747 attrs.set(TEST_BOOL, true);
748
749 assert_eq!(attrs.get(TEST_U64), Some(1u64));
750 assert_eq!(attrs.get(TEST_STRING), Some("two".to_string()));
751 assert_eq!(attrs.get(TEST_BOOL), Some(true));
752 assert_eq!(attrs.len(), 3);
753 }
754
755 #[test]
756 fn test_is_empty() {
757 let attrs = Flattrs::new();
758 assert!(attrs.is_empty());
759
760 let mut attrs2 = Flattrs::new();
761 attrs2.set(TEST_U64, 42u64);
762 assert!(!attrs2.is_empty());
763 }
764
765 #[test]
766 fn test_display() {
767 use crate::attrs::Attrs;
768
769 let empty_flattrs = Flattrs::new();
771 let empty_attrs = Attrs::new();
772 assert_eq!(format!("{}", empty_flattrs), format!("{}", empty_attrs));
773 assert_eq!(format!("{}", empty_flattrs), "");
774
775 let mut single_flattrs = Flattrs::new();
777 single_flattrs.set(TEST_U64, 42u64);
778 let mut single_attrs = Attrs::new();
779 single_attrs.set(TEST_U64, 42u64);
780 assert_eq!(format!("{}", single_flattrs), format!("{}", single_attrs));
781 assert_eq!(
782 format!("{}", single_flattrs),
783 "hyperactor_config::flattrs::tests::test_u64=42"
784 );
785
786 let mut multi_flattrs = Flattrs::new();
789 multi_flattrs.set(TEST_U64, 1u64);
790 multi_flattrs.set(TEST_STRING, "hello".to_string());
791 assert_eq!(
792 format!("{}", multi_flattrs),
793 "hyperactor_config::flattrs::tests::test_u64=1,hyperactor_config::flattrs::tests::test_string=hello"
794 );
795 }
796}