jetstreamer_firehose/
node.rs

1use {
2    crate::{SharedError, block, dataframe, entry, epoch, rewards, subset, transaction, utils},
3    ahash::RandomState,
4    cid::Cid,
5    core::hash::Hasher,
6    crc::{CRC_64_GO_ISO, Crc},
7    fnv::FnvHasher,
8    std::{
9        collections::{HashMap, HashSet},
10        fmt,
11        io::{self, Read},
12        vec::Vec,
13    },
14};
15
16/// Pairing of a decoded [`Node`] with its [`Cid`].
17pub struct NodeWithCid {
18    cid: Cid,
19    node: Node,
20}
21
22impl NodeWithCid {
23    /// Creates a new `(CID, node)` pair.
24    pub const fn new(cid: Cid, node: Node) -> NodeWithCid {
25        NodeWithCid { cid, node }
26    }
27
28    /// Returns the CID associated with the node.
29    pub const fn get_cid(&self) -> &Cid {
30        &self.cid
31    }
32
33    /// Returns the decoded node.
34    pub const fn get_node(&self) -> &Node {
35        &self.node
36    }
37}
38
39/// Convenience collection that retains the CID for every stored node.
40#[derive(Default)]
41pub struct NodesWithCids(
42    #[doc = "Ordered collection of nodes paired with their content identifiers."]
43    pub  Vec<NodeWithCid>,
44    #[doc(hidden)] HashMap<Cid, usize, RandomState>,
45);
46
47impl NodesWithCids {
48    /// Creates an empty [`NodesWithCids`].
49    pub fn new() -> NodesWithCids {
50        NodesWithCids(vec![], HashMap::with_hasher(RandomState::new()))
51    }
52
53    /// Appends a node to the collection.
54    pub fn push(&mut self, node_with_cid: NodeWithCid) {
55        let index = self.0.len();
56        self.1.insert(*node_with_cid.get_cid(), index);
57        self.0.push(node_with_cid);
58    }
59
60    /// Returns the number of stored nodes.
61    pub const fn len(&self) -> usize {
62        self.0.len()
63    }
64
65    /// Returns `true` if no nodes are stored.
66    pub const fn is_empty(&self) -> bool {
67        self.len() == 0
68    }
69
70    /// Returns the node at `index`.
71    pub fn get(&self, index: usize) -> &NodeWithCid {
72        &self.0[index]
73    }
74
75    /// Looks up a node by CID.
76    pub fn get_by_cid(&self, cid: &Cid) -> Option<&NodeWithCid> {
77        self.1.get(cid).and_then(|index| self.0.get(*index))
78    }
79
80    /// Reassembles a potentially multi-part dataframe using the nodes in the collection.
81    pub fn reassemble_dataframes(
82        &self,
83        first_dataframe: &dataframe::DataFrame,
84    ) -> Result<Vec<u8>, SharedError> {
85        let mut data = Vec::with_capacity(first_dataframe.data.len());
86        data.extend_from_slice(first_dataframe.data.as_slice());
87
88        // A continuation may not revisit a frame the traversal already consumed; a cycle would reappend the same chunks forever.
89        let mut visited = HashSet::new();
90        let mut next_arr = first_dataframe.next.as_deref();
91        while let Some(next_cids) = next_arr {
92            let mut next_segment = None;
93            for next_cid in next_cids {
94                if !visited.insert(*next_cid) {
95                    return Err(Box::new(std::io::Error::other(std::format!(
96                        "Cyclic dataframe continuation: {:?}",
97                        next_cid
98                    ))) as SharedError);
99                }
100                let next_node = self.get_by_cid(next_cid).ok_or_else(|| {
101                    Box::new(std::io::Error::other(std::format!(
102                        "Missing CID: {:?}",
103                        next_cid
104                    ))) as SharedError
105                })?;
106
107                let next_dataframe = next_node.get_node().get_dataframe().ok_or_else(|| {
108                    Box::new(std::io::Error::other(std::format!(
109                        "Expected DataFrame, got {:?}",
110                        next_node.get_node()
111                    ))) as SharedError
112                })?;
113
114                data.extend_from_slice(next_dataframe.data.as_slice());
115                next_segment = next_dataframe.next.as_deref();
116            }
117            next_arr = next_segment;
118        }
119
120        if let Some(wanted_hash) = first_dataframe.hash {
121            verify_hash(&data, wanted_hash)?;
122        }
123        Ok(data)
124    }
125
126    /// Iterates over every node and invokes `f`.
127    pub fn each<F>(&self, mut f: F) -> Result<(), SharedError>
128    where
129        F: FnMut(&NodeWithCid) -> Result<(), SharedError>,
130    {
131        for node_with_cid in &self.0 {
132            f(node_with_cid)?;
133        }
134        Ok(())
135    }
136
137    /// Returns the CIDs for all stored nodes.
138    pub fn get_cids(&self) -> Vec<Cid> {
139        let mut cids = vec![];
140        for node_with_cid in &self.0 {
141            cids.push(*node_with_cid.get_cid());
142        }
143        cids
144    }
145
146    /// Returns a reference to the final [`block::Block`] in the collection.
147    pub fn get_block(&self) -> Result<&block::Block, SharedError> {
148        // the last node should be a block
149        let last_node = self.0.last();
150        if last_node.is_none() {
151            return Err(Box::new(std::io::Error::other("No nodes".to_owned())));
152        }
153        let last_node_un = last_node.unwrap();
154        if !last_node_un.get_node().is_block() {
155            return Err(Box::new(std::io::Error::other(std::format!(
156                "Expected Block, got {:?}",
157                last_node_un.get_node()
158            ))));
159        }
160        let block = last_node_un.get_node().get_block().unwrap();
161        Ok(block)
162    }
163}
164
165/// Validates the provided data against the expected CRC64 (or legacy FNV) hash.
166pub fn verify_hash(data: &[u8], hash: u64) -> Result<(), SharedError> {
167    let crc64 = checksum_crc64(data);
168    if crc64 != hash {
169        // Maybe it's the legacy checksum function?
170        let fnv = checksum_fnv(data);
171        if fnv != hash {
172            return Err(Box::new(std::io::Error::other(std::format!(
173                "data hash mismatch: wanted {:?}, got crc64={:?}, fnv={:?}",
174                hash,
175                crc64,
176                fnv
177            ))));
178        }
179    }
180    Ok(())
181}
182
183fn checksum_crc64(data: &[u8]) -> u64 {
184    let crc = Crc::<u64>::new(&CRC_64_GO_ISO);
185    let mut digest = crc.digest();
186    digest.update(data);
187    digest.finalize()
188}
189
190fn checksum_fnv(data: &[u8]) -> u64 {
191    let mut hasher = FnvHasher::default();
192    hasher.write(data);
193    hasher.finish()
194}
195
196/// Unified representation of all decoded firehose node types.
197#[derive(Clone, PartialEq, Eq, Hash, Debug)]
198pub enum Node {
199    /// Raw transaction node.
200    Transaction(transaction::Transaction),
201    /// Ledger entry node.
202    Entry(entry::Entry),
203    /// Block node containing ledger metadata and entries.
204    Block(block::Block),
205    /// Subset node linking to a contiguous block range.
206    Subset(subset::Subset),
207    /// Epoch node referencing subset descriptors.
208    Epoch(epoch::Epoch),
209    /// Rewards node containing per-account payouts.
210    Rewards(rewards::Rewards),
211    /// Data frame node wrapping arbitrary binary payloads.
212    DataFrame(dataframe::DataFrame),
213}
214
215impl Node {
216    /// Returns `true` if this node is a [`transaction::Transaction`].
217    pub const fn is_transaction(&self) -> bool {
218        matches!(self, Node::Transaction(_))
219    }
220
221    /// Returns `true` if this node is an [`entry::Entry`].
222    pub const fn is_entry(&self) -> bool {
223        matches!(self, Node::Entry(_))
224    }
225
226    /// Returns `true` if this node is a [`block::Block`].
227    pub const fn is_block(&self) -> bool {
228        matches!(self, Node::Block(_))
229    }
230
231    /// Returns `true` if this node is a [`subset::Subset`].
232    pub const fn is_subset(&self) -> bool {
233        matches!(self, Node::Subset(_))
234    }
235
236    /// Returns `true` if this node is an [`epoch::Epoch`].
237    pub const fn is_epoch(&self) -> bool {
238        matches!(self, Node::Epoch(_))
239    }
240
241    /// Returns `true` if this node is a [`rewards::Rewards`].
242    pub const fn is_rewards(&self) -> bool {
243        matches!(self, Node::Rewards(_))
244    }
245
246    /// Returns `true` if this node is a [`dataframe::DataFrame`].
247    pub const fn is_dataframe(&self) -> bool {
248        matches!(self, Node::DataFrame(_))
249    }
250
251    /// Returns the transaction if this node is [`Node::Transaction`].
252    pub const fn get_transaction(&self) -> Option<&transaction::Transaction> {
253        match self {
254            Node::Transaction(transaction) => Some(transaction),
255            _ => None,
256        }
257    }
258
259    /// Returns the entry if this node is [`Node::Entry`].
260    pub const fn get_entry(&self) -> Option<&entry::Entry> {
261        match self {
262            Node::Entry(entry) => Some(entry),
263            _ => None,
264        }
265    }
266
267    /// Returns the block if this node is [`Node::Block`].
268    pub const fn get_block(&self) -> Option<&block::Block> {
269        match self {
270            Node::Block(block) => Some(block),
271            _ => None,
272        }
273    }
274
275    /// Returns the subset if this node is [`Node::Subset`].
276    pub const fn get_subset(&self) -> Option<&subset::Subset> {
277        match self {
278            Node::Subset(subset) => Some(subset),
279            _ => None,
280        }
281    }
282
283    /// Returns the epoch if this node is [`Node::Epoch`].
284    pub const fn get_epoch(&self) -> Option<&epoch::Epoch> {
285        match self {
286            Node::Epoch(epoch) => Some(epoch),
287            _ => None,
288        }
289    }
290
291    /// Returns the rewards data if this node is [`Node::Rewards`].
292    pub const fn get_rewards(&self) -> Option<&rewards::Rewards> {
293        match self {
294            Node::Rewards(rewards) => Some(rewards),
295            _ => None,
296        }
297    }
298
299    /// Returns the dataframe if this node is [`Node::DataFrame`].
300    pub const fn get_dataframe(&self) -> Option<&dataframe::DataFrame> {
301        match self {
302            Node::DataFrame(dataframe) => Some(dataframe),
303            _ => None,
304        }
305    }
306}
307
308// parse_any_from_cbordata parses any CBOR data into either a Epoch, Subset, Block, Rewards, Entry, or Transaction
309/// Parses the raw CBOR payload into the appropriate [`Node`] variant.
310pub fn parse_any_from_cbordata(data: Vec<u8>) -> Result<Node, SharedError> {
311    let decoded_data: serde_cbor::Value = serde_cbor::from_slice(&data)?;
312    // Process the decoded data
313    // println!("Data: {:?}", decoded_data);
314    let cloned_data = decoded_data.clone();
315
316    // decoded_data is an serde_cbor.Array; print the kind, which is the first element of the array
317    if let serde_cbor::Value::Array(array) = decoded_data {
318        // println!("Kind: {:?}", array[0]);
319        if let Some(serde_cbor::Value::Integer(kind)) = array.first() {
320            // println!(
321            //     "Kind: {:?}",
322            //     Kind::from_u64(kind as u64).unwrap().to_string()
323            // );
324
325            // based on the kind, we can decode the rest of the data
326            let Some(kind) = Kind::from_u64(*kind as u64) else {
327                return Err(Box::new(std::io::Error::other(std::format!(
328                    "Invalid kind: {:?}",
329                    kind
330                ))));
331            };
332            match kind {
333                Kind::Transaction => {
334                    let transaction = transaction::Transaction::from_cbor(cloned_data)?;
335                    return Ok(Node::Transaction(transaction));
336                }
337                Kind::Entry => {
338                    let entry = entry::Entry::from_cbor(cloned_data)?;
339                    return Ok(Node::Entry(entry));
340                }
341                Kind::Block => {
342                    let block = block::Block::from_cbor(cloned_data)?;
343                    return Ok(Node::Block(block));
344                }
345                Kind::Subset => {
346                    let subset = subset::Subset::from_cbor(cloned_data)?;
347                    return Ok(Node::Subset(subset));
348                }
349                Kind::Epoch => {
350                    let epoch = epoch::Epoch::from_cbor(cloned_data)?;
351                    return Ok(Node::Epoch(epoch));
352                }
353                Kind::Rewards => {
354                    let rewards = rewards::Rewards::from_cbor(cloned_data)?;
355                    return Ok(Node::Rewards(rewards));
356                }
357                Kind::DataFrame => {
358                    let dataframe = dataframe::DataFrame::from_cbor(cloned_data)?;
359                    return Ok(Node::DataFrame(dataframe));
360                } // unknown => {
361                  //     return Err(Box::new(std::io::Error::new(
362                  //         std::io::ErrorKind::Other,
363                  //         std::format!("Unknown type: {:?}", unknown),
364                  //     )))
365                  // }
366            }
367        }
368    }
369
370    Err(Box::new(std::io::Error::other("Unknown type".to_owned())))
371}
372
373/// Numeric discriminant used in the CBOR encoding of [`Node`] variants.
374pub enum Kind {
375    /// Transaction node discriminant.
376    Transaction,
377    /// Entry node discriminant.
378    Entry,
379    /// Block node discriminant.
380    Block,
381    /// Subset node discriminant.
382    Subset,
383    /// Epoch node discriminant.
384    Epoch,
385    /// Rewards node discriminant.
386    Rewards,
387    /// Data frame node discriminant.
388    DataFrame,
389}
390
391impl fmt::Debug for Kind {
392    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
393        f.debug_struct("Kind")
394            .field("kind", &self.to_string())
395            .finish()
396    }
397}
398
399impl fmt::Display for Kind {
400    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
401        let kind = match self {
402            Kind::Transaction => "Transaction",
403            Kind::Entry => "Entry",
404            Kind::Block => "Block",
405            Kind::Subset => "Subset",
406            Kind::Epoch => "Epoch",
407            Kind::Rewards => "Rewards",
408            Kind::DataFrame => "DataFrame",
409        };
410        write!(f, "{}", kind)
411    }
412}
413
414impl Kind {
415    /// Converts a numeric discriminant into a [`Kind`].
416    pub const fn from_u64(kind: u64) -> Option<Kind> {
417        match kind {
418            0 => Some(Kind::Transaction),
419            1 => Some(Kind::Entry),
420            2 => Some(Kind::Block),
421            3 => Some(Kind::Subset),
422            4 => Some(Kind::Epoch),
423            5 => Some(Kind::Rewards),
424            6 => Some(Kind::DataFrame),
425            _ => None,
426        }
427    }
428
429    /// Returns the numeric discriminant for this [`Kind`].
430    pub const fn to_u64(&self) -> u64 {
431        match self {
432            Kind::Transaction => 0,
433            Kind::Entry => 1,
434            Kind::Block => 2,
435            Kind::Subset => 3,
436            Kind::Epoch => 4,
437            Kind::Rewards => 5,
438            Kind::DataFrame => 6,
439        }
440    }
441}
442
443/// Raw node extracted from an Old Faithful CAR segment.
444pub struct RawNode {
445    cid: Cid,
446    data: Vec<u8>,
447}
448
449// Debug trait for RawNode
450impl fmt::Debug for RawNode {
451    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
452        f.debug_struct("RawNode")
453            .field("cid", &self.cid)
454            .field("data", &self.data)
455            .finish()
456    }
457}
458
459impl RawNode {
460    /// Creates a [`RawNode`] from the provided CID and raw bytes read from Old Faithful.
461    pub const fn new(cid: Cid, data: Vec<u8>) -> RawNode {
462        RawNode { cid, data }
463    }
464
465    /// Parses the node into a typed [`Node`].
466    pub fn parse(&self) -> Result<Node, SharedError> {
467        let parsed = parse_any_from_cbordata(self.data.clone());
468        match parsed {
469            Ok(node) => Ok(node),
470            Err(_) => Err(Box::new(std::io::Error::other("Unknown type".to_owned()))),
471        }
472    }
473
474    /// Decodes a [`RawNode`] from an Old Faithful CAR cursor.
475    pub fn from_cursor(cursor: &mut io::Cursor<Vec<u8>>) -> Result<RawNode, SharedError> {
476        let cid_version = utils::read_uvarint(cursor)?;
477        // println!("CID version: {}", cid_version);
478
479        let multicodec = utils::read_uvarint(cursor)?;
480        // println!("Multicodec: {}", multicodec);
481
482        // Multihash hash function code.
483        let hash_function = utils::read_uvarint(cursor)?;
484        // println!("Hash function: {}", hash_function);
485
486        // Multihash digest length.
487        let digest_length = utils::read_uvarint(cursor)?;
488        // println!("Digest length: {}", digest_length);
489
490        if digest_length > 64 {
491            return Err(Box::new(std::io::Error::other(
492                "Digest length too long".to_owned(),
493            )));
494        }
495
496        // reac actual digest
497        let mut digest = vec![0u8; digest_length as usize];
498        cursor.read_exact(&mut digest)?;
499
500        // the rest is the data
501        let mut data = vec![];
502        cursor.read_to_end(&mut data)?;
503
504        // println!("Data: {:?}", data);
505
506        let ha = multihash::Multihash::wrap(hash_function, digest.as_slice())?;
507
508        match cid_version {
509            0 => {
510                let cid = Cid::new_v0(ha)?;
511                let raw_node = RawNode::new(cid, data);
512                Ok(raw_node)
513            }
514            1 => {
515                let cid = Cid::new_v1(multicodec, ha);
516                let raw_node = RawNode::new(cid, data);
517                Ok(raw_node)
518            }
519            _ => Err(Box::new(std::io::Error::other(
520                "Unknown CID version".to_owned(),
521            ))),
522        }
523    }
524}
525
526/// Old Faithful CAR reader that produces [`RawNode`] values from a synchronous source.
527pub struct NodeReader<R: Read> {
528    /// Underlying reader yielding Old Faithful CAR bytes.
529    reader: R,
530    /// Cached Old Faithful CAR header.
531    header: Vec<u8>,
532    /// Number of Old Faithful items that have been read.
533    item_index: u64,
534}
535
536impl<R: Read> NodeReader<R> {
537    /// Creates a new [`NodeReader`] around a blocking reader.
538    pub fn new(reader: R) -> Result<NodeReader<R>, SharedError> {
539        let node_reader = NodeReader {
540            reader,
541            header: vec![],
542            item_index: 0,
543        };
544        Ok(node_reader)
545    }
546
547    /// Returns the raw Old Faithful CAR header, caching it for subsequent calls.
548    pub fn read_raw_header(&mut self) -> Result<Vec<u8>, SharedError> {
549        if !self.header.is_empty() {
550            return Ok(self.header.clone());
551        };
552        let header_length = utils::read_uvarint(&mut self.reader)?;
553        if header_length > 1024 {
554            return Err(Box::new(std::io::Error::other(
555                "Header length too long".to_owned(),
556            )));
557        }
558        let mut header = vec![0u8; header_length as usize];
559        self.reader.read_exact(&mut header)?;
560
561        self.header.clone_from(&header);
562
563        let clone = header.clone();
564        Ok(clone.as_slice().to_owned())
565    }
566
567    #[allow(clippy::should_implement_trait)]
568    /// Reads the next [`RawNode`] without parsing it from Old Faithful data.
569    pub fn next(&mut self) -> Result<RawNode, SharedError> {
570        if self.header.is_empty() {
571            self.read_raw_header()?;
572        };
573
574        // println!("Item index: {}", item_index);
575        self.item_index += 1;
576
577        // Read and decode the uvarint prefix (length of CID + data)
578        let section_size = utils::read_uvarint(&mut self.reader)?;
579        // println!("Section size: {}", section_size);
580
581        if section_size > utils::MAX_ALLOWED_SECTION_SIZE as u64 {
582            return Err(Box::new(std::io::Error::other(
583                "Section size too long".to_owned(),
584            )));
585        }
586
587        // read whole item
588        let mut item = vec![0u8; section_size as usize];
589        self.reader.read_exact(&mut item)?;
590
591        // dump item bytes as numbers
592        // println!("Item bytes: {:?}", item);
593
594        // now create a cursor over the item
595        let mut cursor = io::Cursor::new(item);
596
597        RawNode::from_cursor(&mut cursor)
598    }
599
600    /// Reads and parses the next node, returning it with its [`Cid`].
601    pub fn next_parsed(&mut self) -> Result<NodeWithCid, SharedError> {
602        let raw_node = self.next()?;
603        let cid = raw_node.cid;
604        Ok(NodeWithCid::new(cid, raw_node.parse()?))
605    }
606
607    /// Iterates Old Faithful nodes until a block is encountered, returning the collected list.
608    pub fn read_until_block(&mut self) -> Result<NodesWithCids, SharedError> {
609        let mut nodes = NodesWithCids::new();
610        loop {
611            let node = self.next_parsed()?;
612            if node.get_node().is_block() {
613                nodes.push(node);
614                break;
615            }
616            nodes.push(node);
617        }
618        Ok(nodes)
619    }
620
621    /// Returns the number of CAR items read so far.
622    pub const fn get_item_index(&self) -> u64 {
623        self.item_index
624    }
625}
626
627#[cfg(test)]
628mod tests {
629    use {super::*, crate::dataframe::DataFrame, crate::utils::Buffer};
630
631    /// The standard CID bytes used by the decoder fixtures.
632    fn test_cid() -> Cid {
633        Cid::try_from(
634            vec![
635                1, 113, 18, 32, 56, 148, 167, 251, 237, 117, 200, 226, 181, 134, 79, 115, 131, 220,
636                232, 143, 20, 67, 224, 179, 48, 130, 197, 123, 226, 85, 85, 56, 38, 84, 106, 225,
637            ]
638            .as_slice(),
639        )
640        .unwrap()
641    }
642
643    /// A continuation that points back at a frame the traversal already
644    /// consumed keeps reappending the same chunk forever; the reassembly
645    /// must fail instead of looping.
646    #[test]
647    fn reassembly_fails_on_a_cyclic_continuation() {
648        let cid = test_cid();
649        let frame = DataFrame {
650            kind: Kind::DataFrame.to_u64(),
651            hash: None,
652            index: None,
653            total: None,
654            data: Buffer::from_vec(vec![1u8; 1024]),
655            next: Some(vec![cid]),
656        };
657        let mut nodes = NodesWithCids::new();
658        nodes.push(NodeWithCid::new(cid, Node::DataFrame(frame.clone())));
659
660        let err = nodes.reassemble_dataframes(&frame).unwrap_err();
661        assert!(
662            err.to_string().contains("Cyclic"),
663            "unexpected error: {err}"
664        );
665    }
666}
jetstreamer_firehose/node.rs

jetstreamer_firehose/
node.rs
