consensus_core/
test_dag_builder.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use std::{
5    collections::{BTreeMap, HashSet},
6    ops::{Bound::Included, RangeInclusive},
7    sync::Arc,
8};
9
10use consensus_config::AuthorityIndex;
11use consensus_types::block::{BlockDigest, BlockRef, BlockTimestampMs, Round, TransactionIndex};
12use parking_lot::RwLock;
13use rand::{Rng, SeedableRng, rngs::StdRng, seq::SliceRandom, thread_rng};
14
15use crate::{
16    CommitRef, CommittedSubDag, Transaction,
17    block::{BlockAPI, BlockTransactionVotes, Slot, TestBlock, VerifiedBlock, genesis_blocks},
18    commit::{CertifiedCommit, CommitDigest, TrustedCommit},
19    context::Context,
20    dag_state::DagState,
21    leader_schedule::{LeaderSchedule, LeaderSwapTable},
22    linearizer::{BlockStoreAPI, Linearizer},
23};
24
25/// DagBuilder API
26///
27/// Usage:
28///
29/// DAG Building
30/// ```ignore
31/// let context = Arc::new(Context::new_for_test(4).0);
32/// let mut dag_builder = DagBuilder::new(context);
33/// dag_builder.layer(1).build(); // Round 1 is fully connected with parents by default.
34/// dag_builder.layers(2..=10).build(); // Rounds 2 ~ 10 are fully connected with parents by default.
35/// dag_builder.layer(11).min_parent_links().build(); // Round 11 is minimally and randomly connected with parents, without weak links.
36/// dag_builder.layer(12).no_leader_block(0).build(); // Round 12 misses leader block. Other blocks are fully connected with parents.
37/// dag_builder.layer(13).no_leader_link(12, 0).build(); // Round 13 misses votes for leader block. Other blocks are fully connected with parents.
38/// dag_builder.layer(14).authorities(vec![3,5]).skip_block().build(); // Round 14 authorities 3 and 5 will not propose any block.
39/// dag_builder.layer(15).authorities(vec![3,5]).skip_ancestor_links(vec![1,2]).build(); // Round 15 authorities 3 and 5 will not link to ancestors 1 and 2
40/// dag_builder.layer(16).authorities(vec![3,5]).equivocate(3).build(); // Round 16 authorities 3 and 5 will produce 3 equivocating blocks.
41/// ```
42///
43/// Persisting to DagState by Layer
44/// ```ignore
45/// let dag_state = Arc::new(RwLock::new(DagState::new(
46///    dag_builder.context.clone(),
47///    Arc::new(MemStore::new()),
48/// )));
49/// let context = Arc::new(Context::new_for_test(4).0);
50/// let mut dag_builder = DagBuilder::new(context);
51/// dag_builder.layer(1).build().persist_layers(dag_state.clone()); // persist the layer
52/// ```
53///
54/// Persisting entire DAG to DagState
55/// ```ignore
56/// let dag_state = Arc::new(RwLock::new(DagState::new(
57///    dag_builder.context.clone(),
58///    Arc::new(MemStore::new()),
59/// )));
60/// let context = Arc::new(Context::new_for_test(4).0);
61/// let mut dag_builder = DagBuilder::new(context);
62/// dag_builder.layer(1).build();
63/// dag_builder.layers(2..=10).build();
64/// dag_builder.persist_all_blocks(dag_state.clone()); // persist entire DAG
65/// ```
66///
67/// Printing DAG
68/// ```ignore
69/// let context = Arc::new(Context::new_for_test(4).0);
70/// let mut dag_builder = DagBuilder::new(context);
71/// dag_builder.layer(1).build();
72/// dag_builder.print(); // pretty print the entire DAG
73/// ```
74pub struct DagBuilder {
75    pub(crate) context: Arc<Context>,
76    pub(crate) leader_schedule: LeaderSchedule,
77    // The genesis blocks
78    pub(crate) genesis: BTreeMap<BlockRef, VerifiedBlock>,
79    // The current set of ancestors that any new layer will attempt to connect to.
80    pub(crate) last_ancestors: Vec<BlockRef>,
81    // All blocks created by dag builder. Will be used to pretty print or to be
82    // retrieved for testing/persiting to dag state.
83    pub(crate) blocks: BTreeMap<BlockRef, VerifiedBlock>,
84    // All the committed sub dags created by the dag builder.
85    pub(crate) committed_sub_dags: Vec<(CommittedSubDag, TrustedCommit)>,
86    pub(crate) last_committed_rounds: Vec<Round>,
87
88    number_of_leaders: u32,
89}
90
91#[allow(unused)]
92impl DagBuilder {
93    pub fn new(context: Arc<Context>) -> Self {
94        let leader_schedule = LeaderSchedule::new(context.clone(), LeaderSwapTable::default());
95        let genesis_blocks = genesis_blocks(context.as_ref());
96        let genesis: BTreeMap<BlockRef, VerifiedBlock> = genesis_blocks
97            .into_iter()
98            .map(|block| (block.reference(), block))
99            .collect();
100        let last_ancestors = genesis.keys().cloned().collect();
101        Self {
102            last_committed_rounds: vec![0; context.committee.size()],
103            context,
104            leader_schedule,
105            number_of_leaders: 1,
106            genesis,
107            last_ancestors,
108            blocks: BTreeMap::new(),
109            committed_sub_dags: vec![],
110        }
111    }
112
113    pub(crate) fn blocks(&self, rounds: RangeInclusive<Round>) -> Vec<VerifiedBlock> {
114        assert!(
115            !self.blocks.is_empty(),
116            "No blocks have been created, please make sure that you have called build method"
117        );
118        self.blocks
119            .iter()
120            .filter_map(|(block_ref, block)| rounds.contains(&block_ref.round).then_some(block))
121            .cloned()
122            .collect::<Vec<VerifiedBlock>>()
123    }
124
125    pub fn all_blocks(&self) -> Vec<VerifiedBlock> {
126        assert!(
127            !self.blocks.is_empty(),
128            "No blocks have been created, please make sure that you have called build method"
129        );
130        self.blocks.values().cloned().collect()
131    }
132
133    pub(crate) fn get_sub_dag_and_commits(
134        &mut self,
135        leader_rounds: RangeInclusive<Round>,
136    ) -> Vec<(CommittedSubDag, TrustedCommit)> {
137        let (last_leader_round, mut last_commit_ref, mut last_timestamp_ms) =
138            if let Some((sub_dag, _)) = self.committed_sub_dags.last() {
139                (
140                    sub_dag.leader.round,
141                    sub_dag.commit_ref,
142                    sub_dag.timestamp_ms,
143                )
144            } else {
145                (0, CommitRef::new(0, CommitDigest::MIN), 0)
146            };
147
148        struct BlockStorage {
149            gc_round: Round,
150            blocks: BTreeMap<BlockRef, (VerifiedBlock, bool)>, // the tuple represends the block and whether it is committed
151            genesis: BTreeMap<BlockRef, VerifiedBlock>,
152        }
153        impl BlockStoreAPI for BlockStorage {
154            fn get_blocks(&self, refs: &[BlockRef]) -> Vec<Option<VerifiedBlock>> {
155                refs.iter()
156                    .map(|block_ref| {
157                        if block_ref.round == 0 {
158                            return self.genesis.get(block_ref).cloned();
159                        }
160                        self.blocks
161                            .get(block_ref)
162                            .map(|(block, _committed)| block.clone())
163                    })
164                    .collect()
165            }
166
167            fn gc_round(&self) -> Round {
168                self.gc_round
169            }
170
171            fn set_committed(&mut self, block_ref: &BlockRef) -> bool {
172                let Some((_block, committed)) = self.blocks.get_mut(block_ref) else {
173                    panic!("Block {:?} should be found in store", block_ref);
174                };
175                if !*committed {
176                    *committed = true;
177                    return true;
178                }
179                false
180            }
181
182            fn is_committed(&self, block_ref: &BlockRef) -> bool {
183                self.blocks
184                    .get(block_ref)
185                    .map(|(_, committed)| *committed)
186                    .expect("Block should be found in store")
187            }
188        }
189
190        let mut storage = BlockStorage {
191            blocks: self
192                .blocks
193                .clone()
194                .into_iter()
195                .map(|(k, v)| (k, (v, false)))
196                .collect(),
197            genesis: self.genesis.clone(),
198            gc_round: 0,
199        };
200
201        // Create any remaining committed sub dags
202        for leader_block in self
203            .leader_blocks(last_leader_round + 1..=*leader_rounds.end())
204            .into_iter()
205            .flatten()
206        {
207            // set the gc round to the round of the leader block
208            storage.gc_round = leader_block
209                .round()
210                .saturating_sub(1)
211                .saturating_sub(self.context.protocol_config.gc_depth());
212
213            let leader_block_ref = leader_block.reference();
214
215            let to_commit = Linearizer::linearize_sub_dag(leader_block.clone(), &mut storage);
216
217            last_timestamp_ms = Linearizer::calculate_commit_timestamp(
218                &self.context.clone(),
219                &mut storage,
220                &leader_block,
221                last_timestamp_ms,
222            );
223
224            // Update the last committed rounds
225            for block in &to_commit {
226                self.last_committed_rounds[block.author()] =
227                    self.last_committed_rounds[block.author()].max(block.round());
228            }
229
230            let commit = TrustedCommit::new_for_test(
231                last_commit_ref.index + 1,
232                last_commit_ref.digest,
233                last_timestamp_ms,
234                leader_block_ref,
235                to_commit
236                    .iter()
237                    .map(|block| block.reference())
238                    .collect::<Vec<_>>(),
239            );
240
241            last_commit_ref = commit.reference();
242
243            let sub_dag = CommittedSubDag::new(
244                leader_block_ref,
245                to_commit,
246                last_timestamp_ms,
247                commit.reference(),
248            );
249
250            self.committed_sub_dags.push((sub_dag, commit));
251        }
252
253        self.committed_sub_dags
254            .clone()
255            .into_iter()
256            .filter(|(sub_dag, _)| leader_rounds.contains(&sub_dag.leader.round))
257            .collect()
258    }
259
260    pub(crate) fn get_sub_dag_and_certified_commits(
261        &mut self,
262        leader_rounds: RangeInclusive<Round>,
263    ) -> Vec<(CommittedSubDag, CertifiedCommit)> {
264        let commits = self.get_sub_dag_and_commits(leader_rounds);
265        commits
266            .into_iter()
267            .map(|(sub_dag, commit)| {
268                let certified_commit =
269                    CertifiedCommit::new_certified(commit, sub_dag.blocks.clone());
270                (sub_dag, certified_commit)
271            })
272            .collect()
273    }
274
275    pub fn leader_blocks(&self, rounds: RangeInclusive<Round>) -> Vec<Option<VerifiedBlock>> {
276        assert!(
277            !self.blocks.is_empty(),
278            "No blocks have been created, please make sure that you have called build method"
279        );
280        rounds
281            .into_iter()
282            .map(|round| self.leader_block(round))
283            .collect()
284    }
285
286    pub(crate) fn leader_block(&self, round: Round) -> Option<VerifiedBlock> {
287        assert!(
288            !self.blocks.is_empty(),
289            "No blocks have been created, please make sure that you have called build method"
290        );
291        self.blocks
292            .iter()
293            .find(|(block_ref, _block)| {
294                block_ref.round == round
295                    && block_ref.author == self.leader_schedule.elect_leader(round, 0)
296            })
297            .map(|(_block_ref, block)| block.clone())
298    }
299
300    pub(crate) fn layer(&mut self, round: Round) -> LayerBuilder<'_> {
301        LayerBuilder::new(self, round)
302    }
303
304    pub fn layers(&mut self, rounds: RangeInclusive<Round>) -> LayerBuilder<'_> {
305        let mut builder = LayerBuilder::new(self, *rounds.start());
306        builder.end_round = Some(*rounds.end());
307        builder
308    }
309
310    pub(crate) fn persist_all_blocks(&self, dag_state: Arc<RwLock<DagState>>) {
311        dag_state
312            .write()
313            .accept_blocks(self.blocks.values().cloned().collect());
314    }
315
316    pub(crate) fn print(&self) {
317        let mut dag_str = "DAG {\n".to_string();
318
319        let mut round = 0;
320        for block in self.blocks.values() {
321            if block.round() > round {
322                round = block.round();
323                dag_str.push_str(&format!("Round {round} : \n"));
324            }
325            dag_str.push_str(&format!("    Block {block:#?}\n"));
326        }
327        dag_str.push_str("}\n");
328
329        tracing::info!("{dag_str}");
330    }
331
332    // TODO: merge into layer builder?
333    // This method allows the user to specify specific links to ancestors. The
334    // layer is written to dag state and the blocks are cached in [`DagBuilder`]
335    // state.
336    pub(crate) fn layer_with_connections(
337        &mut self,
338        connections: Vec<(AuthorityIndex, Vec<BlockRef>)>,
339        round: Round,
340    ) {
341        let mut references = Vec::new();
342        for (authority, ancestors) in connections {
343            let author = authority.value() as u32;
344            let base_ts = round as BlockTimestampMs * 1000;
345            let block = VerifiedBlock::new_for_test(
346                TestBlock::new(round, author)
347                    .set_ancestors(ancestors)
348                    .set_timestamp_ms(base_ts + author as u64)
349                    .build(),
350            );
351            references.push(block.reference());
352            self.blocks.insert(block.reference(), block.clone());
353        }
354        self.last_ancestors = references;
355    }
356
357    /// Gets all uncommitted blocks in a slot.
358    pub(crate) fn get_uncommitted_blocks_at_slot(&self, slot: Slot) -> Vec<VerifiedBlock> {
359        let mut blocks = vec![];
360        for (_block_ref, block) in self.blocks.range((
361            Included(BlockRef::new(slot.round, slot.authority, BlockDigest::MIN)),
362            Included(BlockRef::new(slot.round, slot.authority, BlockDigest::MAX)),
363        )) {
364            blocks.push(block.clone())
365        }
366        blocks
367    }
368
369    pub(crate) fn genesis_block_refs(&self) -> Vec<BlockRef> {
370        self.genesis.keys().cloned().collect()
371    }
372}
373
374/// Refer to doc comments for [`DagBuilder`] for usage information.
375pub struct LayerBuilder<'a> {
376    dag_builder: &'a mut DagBuilder,
377
378    start_round: Round,
379    end_round: Option<Round>,
380
381    // Configuration options applied to specified authorities
382    // TODO: convert configuration options into an enum
383    specified_authorities: Option<Vec<AuthorityIndex>>,
384    // Number of transactions to include per block.
385    num_transactions: u32,
386    // Whether to include rejected transactions
387    rejected_transactions_pct: u8,
388    rejected_transactions_seed: u64,
389    // Number of equivocating blocks per specified authority
390    equivocations: usize,
391    // Skip block proposal for specified authorities
392    skip_block: bool,
393    // Skip specified ancestor links for specified authorities
394    skip_ancestor_links: Option<Vec<AuthorityIndex>>,
395    // Skip leader link for specified authorities
396    no_leader_link: bool,
397    // Use to override last ancestors in dag builder
398    override_last_ancestors: bool,
399
400    // Skip leader block proposal
401    no_leader_block: bool,
402    // Used for leader based configurations
403    specified_leader_link_offsets: Option<Vec<u32>>,
404    specified_leader_block_offsets: Option<Vec<u32>>,
405    leader_round: Option<Round>,
406
407    // All ancestors will be linked to the current layer
408    fully_linked_ancestors: bool,
409    // Only 2f+1 random ancestors will be linked to the current layer using a
410    // seed, if provided
411    min_ancestor_links: bool,
412    min_ancestor_links_random_seed: Option<u64>,
413    // Add random weak links to the current layer using a seed, if provided
414    random_weak_links: bool,
415    random_weak_links_random_seed: Option<u64>,
416
417    // Ancestors to link to the current layer
418    ancestors: Vec<BlockRef>,
419    // override last ancestors in dag_builder
420    specified_ancestors: Vec<BlockRef>,
421
422    // The block timestamps for the layer for each specified authority. This will work as base timestamp and the round will be added to make sure that timestamps do offset.
423    timestamps: Vec<BlockTimestampMs>,
424
425    // Accumulated blocks to write to dag state
426    blocks: Vec<VerifiedBlock>,
427}
428
429#[allow(unused)]
430impl<'a> LayerBuilder<'a> {
431    fn new(dag_builder: &'a mut DagBuilder, start_round: Round) -> Self {
432        assert!(start_round > 0, "genesis round is created by default");
433        let ancestors = dag_builder.last_ancestors.clone();
434        Self {
435            dag_builder,
436            start_round,
437            end_round: None,
438            specified_authorities: None,
439            num_transactions: 0,
440            rejected_transactions_pct: 0,
441            rejected_transactions_seed: 0,
442            equivocations: 0,
443            skip_block: false,
444            skip_ancestor_links: None,
445            override_last_ancestors: false,
446            no_leader_link: false,
447            no_leader_block: false,
448            specified_leader_link_offsets: None,
449            specified_leader_block_offsets: None,
450            leader_round: None,
451            fully_linked_ancestors: true,
452            min_ancestor_links: false,
453            min_ancestor_links_random_seed: None,
454            random_weak_links: false,
455            random_weak_links_random_seed: None,
456            ancestors,
457            specified_ancestors: vec![],
458            timestamps: vec![],
459            blocks: vec![],
460        }
461    }
462
463    // Configuration methods
464
465    // If you try to add blocks to the same layer you will end up with the last
466    // ancestors replaced with blocks from the current layer. Use this method
467    // to override the last ancestors with the references you manually have kept.
468    pub fn override_last_ancestors(mut self, ancestors: Vec<BlockRef>) -> Self {
469        self.specified_ancestors = ancestors;
470        self.override_last_ancestors = true;
471        self.build()
472    }
473
474    // Only link 2f+1 random ancestors to the current layer round using a seed,
475    // if provided. Also provide a flag to guarantee the leader is included.
476    // note: configuration is terminal and layer will be built after this call.
477    pub fn min_ancestor_links(mut self, include_leader: bool, seed: Option<u64>) -> Self {
478        self.min_ancestor_links = true;
479        self.min_ancestor_links_random_seed = seed;
480        if include_leader {
481            self.leader_round = Some(self.ancestors.iter().max_by_key(|b| b.round).unwrap().round);
482        }
483        self.fully_linked_ancestors = false;
484        self.build()
485    }
486
487    // No links will be created between the specified ancestors and the specified
488    // authorities at the layer round.
489    // note: configuration is terminal and layer will be built after this call.
490    pub fn skip_ancestor_links(mut self, ancestors_to_skip: Vec<AuthorityIndex>) -> Self {
491        // authorities must be specified for this to apply
492        assert!(self.specified_authorities.is_some());
493        self.skip_ancestor_links = Some(ancestors_to_skip);
494        self.fully_linked_ancestors = false;
495        self.build()
496    }
497
498    // Add random weak links to the current layer round using a seed, if provided
499    pub fn random_weak_links(mut self, seed: Option<u64>) -> Self {
500        self.random_weak_links = true;
501        self.random_weak_links_random_seed = seed;
502        self
503    }
504
505    // Should be called when building a leader round. Will ensure leader block is missing.
506    // A list of specified leader offsets can be provided to skip those leaders.
507    // If none are provided all potential leaders for the round will be skipped.
508    pub fn no_leader_block(mut self, specified_leader_offsets: Vec<u32>) -> Self {
509        self.no_leader_block = true;
510        self.specified_leader_block_offsets = Some(specified_leader_offsets);
511        self
512    }
513
514    // Should be called when building a voting round. Will ensure vote is missing.
515    // A list of specified leader offsets can be provided to skip those leader links.
516    // If none are provided all potential leaders for the round will be skipped.
517    // note: configuration is terminal and layer will be built after this call.
518    pub fn no_leader_link(
519        mut self,
520        leader_round: Round,
521        specified_leader_offsets: Vec<u32>,
522    ) -> Self {
523        self.no_leader_link = true;
524        self.specified_leader_link_offsets = Some(specified_leader_offsets);
525        self.leader_round = Some(leader_round);
526        self.fully_linked_ancestors = false;
527        self.build()
528    }
529
530    pub fn authorities(mut self, authorities: Vec<AuthorityIndex>) -> Self {
531        assert!(
532            self.specified_authorities.is_none(),
533            "Specified authorities already set"
534        );
535        self.specified_authorities = Some(authorities);
536        self
537    }
538
539    // Number of transactions to include per block.
540    pub fn num_transactions(mut self, num_transactions: u32) -> Self {
541        self.num_transactions = num_transactions;
542        self
543    }
544
545    pub fn rejected_transactions_pct(mut self, pct: u8, seed: Option<u64>) -> Self {
546        self.rejected_transactions_pct = pct;
547        self.rejected_transactions_seed = if let Some(seed) = seed {
548            seed
549        } else {
550            thread_rng().r#gen()
551        };
552        self
553    }
554
555    // Multiple blocks will be created for the specified authorities at the layer round.
556    pub fn equivocate(mut self, equivocations: usize) -> Self {
557        // authorities must be specified for this to apply
558        assert!(self.specified_authorities.is_some());
559        self.equivocations = equivocations;
560        self
561    }
562
563    // No blocks will be created for the specified authorities at the layer round.
564    pub fn skip_block(mut self) -> Self {
565        // authorities must be specified for this to apply
566        assert!(self.specified_authorities.is_some());
567        self.skip_block = true;
568        self
569    }
570
571    pub fn with_timestamps(mut self, timestamps: Vec<BlockTimestampMs>) -> Self {
572        // authorities must be specified for this to apply
573        assert!(self.specified_authorities.is_some());
574        assert_eq!(
575            self.specified_authorities.as_ref().unwrap().len(),
576            timestamps.len(),
577            "Timestamps should be provided for each specified authority"
578        );
579        self.timestamps = timestamps;
580        self
581    }
582
583    // Apply the configurations & build the dag layer(s).
584    pub fn build(mut self) -> Self {
585        for round in self.start_round..=self.end_round.unwrap_or(self.start_round) {
586            tracing::debug!("BUILDING LAYER ROUND {round}...");
587
588            let authorities = if self.specified_authorities.is_some() {
589                self.specified_authorities.clone().unwrap()
590            } else {
591                self.dag_builder
592                    .context
593                    .committee
594                    .authorities()
595                    .map(|x| x.0)
596                    .collect()
597            };
598
599            // TODO: investigate if these configurations can be called in combination
600            // for the same layer
601            let mut connections = if self.override_last_ancestors {
602                self.configure_specifed_ancestors()
603            } else if self.fully_linked_ancestors {
604                self.configure_fully_linked_ancestors(round)
605            } else if self.min_ancestor_links {
606                self.configure_min_parent_links(round)
607            } else if self.no_leader_link {
608                self.configure_no_leader_links(authorities.clone(), round)
609            } else if self.skip_ancestor_links.is_some() {
610                self.configure_skipped_ancestor_links(
611                    authorities,
612                    self.skip_ancestor_links.clone().unwrap(),
613                )
614            } else {
615                vec![]
616            };
617
618            if self.random_weak_links {
619                connections.append(&mut self.configure_random_weak_links());
620            }
621
622            self.create_blocks(round, connections);
623        }
624
625        self.dag_builder.last_ancestors = self.ancestors.clone();
626        self
627    }
628
629    pub fn persist_layers(&self, dag_state: Arc<RwLock<DagState>>) {
630        assert!(
631            !self.blocks.is_empty(),
632            "Called to persist layers although no blocks have been created. Make sure you have called build before."
633        );
634        dag_state.write().accept_blocks(self.blocks.clone());
635    }
636
637    // Layer round is minimally and randomly connected with ancestors.
638    pub fn configure_min_parent_links(
639        &mut self,
640        round: Round,
641    ) -> Vec<(AuthorityIndex, Vec<BlockRef>)> {
642        let quorum_threshold = self.dag_builder.context.committee.quorum_threshold() as usize;
643        let mut authorities: Vec<AuthorityIndex> = self
644            .dag_builder
645            .context
646            .committee
647            .authorities()
648            .map(|authority| authority.0)
649            .collect();
650
651        let mut rng = match self.min_ancestor_links_random_seed {
652            Some(s) => StdRng::seed_from_u64(s),
653            None => StdRng::from_entropy(),
654        };
655
656        let mut authorities_to_shuffle = authorities.clone();
657
658        let mut leaders = vec![];
659        if let Some(leader_round) = self.leader_round {
660            let leader_offsets = (0..self.dag_builder.number_of_leaders).collect::<Vec<_>>();
661
662            for leader_offset in leader_offsets {
663                leaders.push(
664                    self.dag_builder
665                        .leader_schedule
666                        .elect_leader(leader_round, leader_offset),
667                );
668            }
669        }
670
671        authorities
672            .iter()
673            .map(|authority| {
674                authorities_to_shuffle.shuffle(&mut rng);
675
676                // TODO: handle quroum threshold properly with stake
677                let min_ancestors: HashSet<AuthorityIndex> = authorities_to_shuffle
678                    .iter()
679                    .take(quorum_threshold)
680                    .cloned()
681                    .collect();
682
683                (
684                    *authority,
685                    self.ancestors
686                        .iter()
687                        .filter(|a| {
688                            leaders.contains(&a.author)
689                                || min_ancestors.contains(&a.author)
690                                || a.round != round
691                        })
692                        .cloned()
693                        .collect::<Vec<BlockRef>>(),
694                )
695            })
696            .collect()
697    }
698
699    // TODO: configure layer round randomly connected with weak links.
700    fn configure_random_weak_links(&mut self) -> Vec<(AuthorityIndex, Vec<BlockRef>)> {
701        unimplemented!("configure_random_weak_links");
702    }
703
704    // Layer round misses link to leader, but other blocks are fully connected with ancestors.
705    fn configure_no_leader_links(
706        &mut self,
707        authorities: Vec<AuthorityIndex>,
708        round: Round,
709    ) -> Vec<(AuthorityIndex, Vec<BlockRef>)> {
710        let mut missing_leaders = Vec::new();
711        let mut specified_leader_offsets = self
712            .specified_leader_link_offsets
713            .clone()
714            .expect("specified_leader_offsets should be set");
715        let leader_round = self.leader_round.expect("leader round should be set");
716
717        // When no specified leader offsets are available, all leaders are
718        // expected to be missing.
719        if specified_leader_offsets.is_empty() {
720            specified_leader_offsets.extend(0..self.dag_builder.number_of_leaders);
721        }
722
723        for leader_offset in specified_leader_offsets {
724            missing_leaders.push(
725                self.dag_builder
726                    .leader_schedule
727                    .elect_leader(leader_round, leader_offset),
728            );
729        }
730
731        self.configure_skipped_ancestor_links(authorities, missing_leaders)
732    }
733
734    fn configure_specifed_ancestors(&mut self) -> Vec<(AuthorityIndex, Vec<BlockRef>)> {
735        self.dag_builder
736            .context
737            .committee
738            .authorities()
739            .map(|authority| (authority.0, self.specified_ancestors.clone()))
740            .collect::<Vec<_>>()
741    }
742
743    fn configure_fully_linked_ancestors(
744        &mut self,
745        round: Round,
746    ) -> Vec<(AuthorityIndex, Vec<BlockRef>)> {
747        self.dag_builder
748            .context
749            .committee
750            .authorities()
751            .map(|authority| {
752                (
753                    authority.0,
754                    // don't connect to ancestors of this round
755                    self.ancestors
756                        .clone()
757                        .into_iter()
758                        .filter(|a| a.round != round)
759                        .collect::<Vec<_>>(),
760                )
761            })
762            .collect::<Vec<_>>()
763    }
764
765    fn configure_skipped_ancestor_links(
766        &mut self,
767        authorities: Vec<AuthorityIndex>,
768        ancestors_to_skip: Vec<AuthorityIndex>,
769    ) -> Vec<(AuthorityIndex, Vec<BlockRef>)> {
770        let filtered_ancestors = self
771            .ancestors
772            .clone()
773            .into_iter()
774            .filter(|ancestor| !ancestors_to_skip.contains(&ancestor.author))
775            .collect::<Vec<_>>();
776        authorities
777            .into_iter()
778            .map(|authority| (authority, filtered_ancestors.clone()))
779            .collect::<Vec<_>>()
780    }
781
782    // Creates the blocks for the new layer based on configured connections, also
783    // sets the ancestors for future layers to be linked to
784    fn create_blocks(&mut self, round: Round, connections: Vec<(AuthorityIndex, Vec<BlockRef>)>) {
785        let mut references = Vec::new();
786        let mut reject_rng =
787            StdRng::seed_from_u64(self.rejected_transactions_seed ^ (round as u64));
788        for (authority, ancestors) in connections {
789            if self.should_skip_block(round, authority) {
790                continue;
791            };
792            let transactions = (0..self.num_transactions)
793                .map(|_| Transaction::new(vec![1_u8; 16]))
794                .collect::<Vec<_>>();
795            let num_blocks = self.num_blocks_to_create(authority);
796            for num_block in 0..num_blocks {
797                let mut votes = vec![];
798                if self.rejected_transactions_pct > 0 {
799                    for ancestor in &ancestors {
800                        let mut rejects = vec![];
801                        for i in 0..self.num_transactions {
802                            if reject_rng.gen_range(1..=100) <= self.rejected_transactions_pct {
803                                rejects.push(i as TransactionIndex);
804                            }
805                        }
806                        if !rejects.is_empty() {
807                            votes.push(BlockTransactionVotes {
808                                block_ref: *ancestor,
809                                rejects,
810                            });
811                        }
812                    }
813                }
814                let timestamp = self.block_timestamp(authority, round, num_block);
815                let block = VerifiedBlock::new_for_test(
816                    TestBlock::new(round, authority.value() as u32)
817                        .set_transactions(transactions.clone())
818                        .set_transaction_votes(votes)
819                        .set_ancestors(ancestors.clone())
820                        .set_timestamp_ms(timestamp)
821                        .build(),
822                );
823                references.push(block.reference());
824                self.dag_builder
825                    .blocks
826                    .insert(block.reference(), block.clone());
827                self.blocks.push(block);
828            }
829        }
830        self.ancestors = references;
831    }
832
833    fn num_blocks_to_create(&self, authority: AuthorityIndex) -> u32 {
834        if self.specified_authorities.is_some()
835            && self
836                .specified_authorities
837                .clone()
838                .unwrap()
839                .contains(&authority)
840        {
841            // Always create 1 block and then the equivocating blocks on top of that.
842            1 + self.equivocations as u32
843        } else {
844            1
845        }
846    }
847
848    fn block_timestamp(
849        &self,
850        authority: AuthorityIndex,
851        round: Round,
852        num_block: u32,
853    ) -> BlockTimestampMs {
854        if self.specified_authorities.is_some() && !self.timestamps.is_empty() {
855            let specified_authorities = self.specified_authorities.as_ref().unwrap();
856
857            if let Some(position) = specified_authorities.iter().position(|&x| x == authority) {
858                return self.timestamps[position] + (round + num_block) as u64;
859            }
860        }
861        let author = authority.value() as u32;
862        let base_ts = round as BlockTimestampMs * 1000;
863        base_ts + (author + round + num_block) as u64
864    }
865
866    fn should_skip_block(&self, round: Round, authority: AuthorityIndex) -> bool {
867        // Safe to unwrap as specified authorites has to be set before skip
868        // is specified.
869        if self.skip_block
870            && self
871                .specified_authorities
872                .clone()
873                .unwrap()
874                .contains(&authority)
875        {
876            return true;
877        }
878        if self.no_leader_block {
879            let mut specified_leader_offsets = self
880                .specified_leader_block_offsets
881                .clone()
882                .expect("specified_leader_block_offsets should be set");
883
884            // When no specified leader offsets are available, all leaders are
885            // expected to be skipped.
886            if specified_leader_offsets.is_empty() {
887                specified_leader_offsets.extend(0..self.dag_builder.number_of_leaders);
888            }
889
890            for leader_offset in specified_leader_offsets {
891                let leader = self
892                    .dag_builder
893                    .leader_schedule
894                    .elect_leader(round, leader_offset);
895
896                if leader == authority {
897                    return true;
898                }
899            }
900        }
901        false
902    }
903}
904
905// TODO: add unit tests
consensus_core/test_dag_builder.rs

consensus_core/
test_dag_builder.rs
