ab_client_archiving/
recreate.rs

1//! Segment re-creation
2
3use crate::task::encode_block;
4use ab_archiving::archiver::{Archiver, ArchiverInstantiationError, NewArchivedSegment};
5use ab_archiving::objects::BlockObject;
6use ab_client_api::{ChainInfo, ReadBlockError};
7use ab_core_primitives::block::BlockNumber;
8use ab_core_primitives::block::header::GenericBlockHeader;
9use ab_core_primitives::block::header::owned::GenericOwnedBlockHeader;
10use ab_core_primitives::block::owned::{GenericOwnedBlock, OwnedBeaconChainBlock};
11use ab_core_primitives::pieces::SegmentProof;
12use ab_core_primitives::segments::{SegmentHeader, SegmentPosition, SuperSegmentIndex};
13use ab_core_primitives::shard::ShardIndex;
14use ab_erasure_coding::ErasureCoding;
15use tokio::task::{JoinError, spawn_blocking};
16
17/// Re-create the genesis segment on demand.
18///
19/// This is a bit of a hack and is useful for deriving of the genesis beacon chain segment that is a
20/// special case since we don't have enough data in the blockchain history itself during genesis to
21/// do the archiving.
22pub fn recreate_genesis_segment(
23    owned_genesis_block: &OwnedBeaconChainBlock,
24    erasure_coding: ErasureCoding,
25) -> NewArchivedSegment {
26    let encoded_block = encode_block(owned_genesis_block);
27
28    let block_outcome = Archiver::new(ShardIndex::BEACON_CHAIN, erasure_coding)
29        .add_block(encoded_block, Vec::new())
30        .expect("Genesis block is much smaller than `u32::MAX` bytes; qed");
31    let mut archived_segment = block_outcome
32        .archived_segments
33        .into_iter()
34        .next()
35        .expect("Genesis block fills at least one segment; qed");
36
37    for piece in archived_segment.pieces.iter_mut() {
38        piece.header.super_segment_index = SuperSegmentIndex::ZERO.into();
39        // Since there is a single segment in super segment, the proof is empty
40    }
41
42    archived_segment.pieces = archived_segment.pieces.to_shared();
43
44    archived_segment
45}
46
47/// Error for [`recreate_segment()`]
48#[derive(Debug, thiserror::Error)]
49pub enum RecreateSegmentError {
50    /// Read block error
51    #[error("Read block error: {0}")]
52    ReadBlockError(#[from] ReadBlockError),
53    /// Archiver instantiation error
54    #[error("Archiver instantiation error: {0}")]
55    ArchiverInstantiationError(#[from] ArchiverInstantiationError),
56    /// Encoded block is too large to archive
57    #[error("Encoded block {block_number} is too large to archive: {size} bytes")]
58    BlockTooLarge {
59        /// Block number
60        block_number: BlockNumber,
61        /// Encoded block size in bytes
62        size: usize,
63    },
64    /// Blocking task join error
65    #[error("Blocking task join error: {0}")]
66    BlockingTaskJoinError(#[from] JoinError),
67}
68
69/// Super segment details for [`recreate_segment()`]
70#[derive(Debug, Copy, Clone)]
71#[repr(C)]
72pub struct RecreateSegmentSuperSegmentDetails {
73    /// Super segment index
74    pub super_segment_index: SuperSegmentIndex,
75    /// Segment position in a super segment
76    pub segment_position: SegmentPosition,
77    /// Segment proof
78    pub segment_proof: SegmentProof,
79}
80
81/// Re-create a segment on demand.
82///
83/// `last_segment_header` corresponds to the last segment before the segment being re-created and
84/// indicates where segment archiving should start.
85///
86/// `extract_block_objects` allows extracting objects stored in blocks to translate them into global
87/// objects.
88///
89/// Returns `Ok(None)` if one of the segment blocks is already pruned.
90pub async fn recreate_segment<Block, CI, EBO>(
91    last_segment_header: Option<SegmentHeader>,
92    chain_info: &CI,
93    erasure_coding: ErasureCoding,
94    super_segment_details: &RecreateSegmentSuperSegmentDetails,
95    mut extract_block_objects: EBO,
96) -> Result<Option<NewArchivedSegment>, RecreateSegmentError>
97where
98    Block: GenericOwnedBlock,
99    CI: ChainInfo<Block>,
100    EBO: FnMut(&Block) -> Vec<BlockObject>,
101{
102    let best_block_root = chain_info.best_root();
103
104    let (start_block_number, mut archiver) = if let Some(last_segment_header) = last_segment_header
105    {
106        let first_block_number = last_segment_header.last_archived_block.number.as_inner();
107
108        let archiver = {
109            let Some(first_block_root) = chain_info
110                .ancestor_header(first_block_number, &best_block_root)
111                .map(|header| *header.header().root())
112            else {
113                return Ok(None);
114            };
115            let block = chain_info.block(&first_block_root).await?;
116            let shard_index = block.header().header().prefix.shard_index;
117            let encoded_block = encode_block(&block);
118
119            Archiver::with_initial_state(
120                shard_index,
121                erasure_coding,
122                last_segment_header,
123                &encoded_block,
124                extract_block_objects(&block),
125            )?
126        };
127
128        (
129            first_block_number.saturating_add(BlockNumber::ONE),
130            archiver,
131        )
132    } else {
133        let best_header = chain_info.best_header();
134        let archiver = Archiver::new(best_header.header().prefix.shard_index, erasure_coding);
135
136        (BlockNumber::ZERO, archiver)
137    };
138
139    for block_number in start_block_number.. {
140        let (encoded_block, block_objects) = {
141            let Some(block_root) = chain_info
142                .ancestor_header(block_number, &best_block_root)
143                .map(|header| *header.header().root())
144            else {
145                return Ok(None);
146            };
147            let block = chain_info.block(&block_root).await?;
148
149            (encode_block(&block), extract_block_objects(&block))
150        };
151
152        let task_fut = spawn_blocking(move || {
153            let encoded_block_size = encoded_block.len();
154            let maybe_outcome = archiver.add_block(encoded_block, block_objects).ok_or(
155                RecreateSegmentError::BlockTooLarge {
156                    block_number,
157                    size: encoded_block_size,
158                },
159            );
160
161            (archiver, maybe_outcome)
162        });
163        let outcome;
164        (archiver, outcome) = task_fut.await?;
165        let outcome = outcome?;
166
167        // TODO: Return global objects once archiver API improves
168        if let Some(mut archived_segment) = outcome.archived_segments.into_iter().next() {
169            for piece in archived_segment.pieces.iter_mut() {
170                piece
171                    .header
172                    .super_segment_index
173                    .replace(super_segment_details.super_segment_index);
174                piece
175                    .header
176                    .segment_position
177                    .replace(super_segment_details.segment_position);
178                piece
179                    .header
180                    .segment_proof
181                    .copy_from_slice(super_segment_details.segment_proof.as_slice());
182            }
183
184            archived_segment.pieces = archived_segment.pieces.to_shared();
185
186            return Ok(Some(archived_segment));
187        }
188    }
189
190    Ok(None)
191}
ab_client_archiving/recreate.rs

ab_client_archiving/
recreate.rs
