tycho-collator 0.3.9

A collator node.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93

//! Module for managing external messages reading state.
//!
//! # State hierarchy for externals
//!
//! ```text
//! ExternalsReaderState
//!
//! ├── ranges: BTreeMap<BlockSeqno, ExternalsRangeReaderState>
//! │   └── by_partitions: BTreeMap<QueuePartitionIdx, ExternalsPartitionRangeReaderState>
//! └── by_partitions: BTreeMap<QueuePartitionIdx, ExternalsPartitionReaderState>
//! ```
//!
//! - `ExternalsReaderState` - top-level state, tracks all external message ranges and per-partition progress
//! - `ExternalsRangeReaderState` - state for a block range (anchor interval), tracks partition states within range
//! - `ExternalsPartitionReaderState` - global partition state, stores `processed_to` position across all ranges
//! - `ExternalsPartitionRangeReaderState` - partition state within a range, contains message buffer and offsets

use crate::collator::messages_reader::state::ext::partition_range_reader::ExternalsPartitionRangeReaderState;
use crate::mempool::MempoolAnchorId;
use crate::types::processed_upto::ExternalsRangeInfo;

pub mod partition_range_reader;
pub mod partition_reader;
pub mod range_reader;
pub mod reader;

#[derive(Debug, Clone)]
pub struct ExternalsReaderRange {
    pub from: ExternalKey,
    pub to: ExternalKey,
    pub current_position: ExternalKey,
    /// Chain time of the block during whose collation the range was read
    pub chain_time: u64,
}

impl ExternalsReaderRange {
    pub fn from_range_info(range_info: &ExternalsRangeInfo, processed_to: ExternalKey) -> Self {
        let from = range_info.from.into();
        let to = range_info.to.into();
        let current_position = if processed_to < from {
            from
        } else if processed_to < to {
            processed_to
        } else {
            to
        };
        Self {
            from,
            to,
            current_position,
            chain_time: range_info.chain_time,
        }
    }
}

impl From<(&ExternalsReaderRange, &ExternalsPartitionRangeReaderState)> for ExternalsRangeInfo {
    fn from((range, state): (&ExternalsReaderRange, &ExternalsPartitionRangeReaderState)) -> Self {
        Self {
            from: range.from.into(),
            to: range.to.into(),
            chain_time: range.chain_time,
            skip_offset: *state.skip_offset,
            processed_offset: *state.processed_offset,
        }
    }
}

#[derive(Default, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub struct ExternalKey {
    pub anchor_id: MempoolAnchorId,
    pub msgs_offset: u64,
}

impl std::fmt::Debug for ExternalKey {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "({}, {})", self.anchor_id, self.msgs_offset)
    }
}

impl From<(MempoolAnchorId, u64)> for ExternalKey {
    fn from(value: (MempoolAnchorId, u64)) -> Self {
        Self {
            anchor_id: value.0,
            msgs_offset: value.1,
        }
    }
}

impl From<ExternalKey> for (MempoolAnchorId, u64) {
    fn from(value: ExternalKey) -> Self {
        (value.anchor_id, value.msgs_offset)
    }
}