hotshot_new_protocol/
block.rs

1use std::{
2    collections::{BTreeMap, BTreeSet, HashMap, HashSet},
3    num::{NonZeroU64, NonZeroUsize},
4    panic::resume_unwind,
5    sync::Arc,
6    time::Duration,
7};
8
9use committable::{Commitment, Committable};
10use hotshot_types::{
11    PeerConfig,
12    consensus::PayloadWithMetadata,
13    data::{EpochNumber, Leaf2, VidCommitment, VidCommitment2, ViewNumber},
14    epoch_membership::EpochMembershipCoordinator,
15    message::UpgradeLock,
16    traits::{
17        BlockPayload, EncodeBytes, ValidatedState as _,
18        block_contents::{BuilderFee, Transaction},
19        metrics::Histogram,
20        node_implementation::NodeType,
21        signature_key::BuilderSignatureKey,
22    },
23    utils::BuilderCommitment,
24};
25use tokio::{
26    task::{AbortHandle, JoinSet, spawn_blocking},
27    time::sleep,
28};
29use tracing::{debug_span, error, warn};
30use versions::{NEW_PROTOCOL_VERSION, NO_BUILDER_COMMITMENT_VERSION, Version};
31
32use crate::{
33    consensus::ConsensusInput,
34    coordinator::metrics::{Measurement, finish_measurement, ignore_measurement},
35    helpers::proposal_commitment,
36    leader_trace::{LeaderEvent, LeaderTracerHandle},
37    message::{DedupManifest, Proposal, TransactionMessage},
38    network::message_limit,
39    state::HeaderRequest,
40    trace_leader_event,
41    vid::{BuiltDispersal, VidDisperseError},
42};
43
44#[derive(Debug, thiserror::Error)]
45pub enum BlockError {
46    #[error("payload construction failed: {0}")]
47    PayloadConstruction(String),
48
49    #[error("stake table unavailable")]
50    StakeTableUnavailable,
51
52    #[error("builder signature failed")]
53    BuilderSignature,
54
55    #[error("vid dispersal failed: {0}")]
56    Vid(#[from] VidDisperseError),
57
58    #[error("version {0} predates the new protocol")]
59    UnsupportedVersion(Version),
60
61    #[error("block builder task cancelled")]
62    Cancelled,
63}
64
65/// Why [`BlockBuilder::on_submit_transaction`] refused a transaction.
66#[derive(Debug, thiserror::Error)]
67pub enum SubmitError {
68    /// Bigger than a block or a forwarded message.
69    #[error("transaction of {size} bytes exceeds the {limit} byte limit")]
70    TooLarge { size: u64, limit: u64 },
71    /// The retry buffer is full, retrying later can succeed.
72    #[error("transaction retry buffer is full")]
73    RetryBufferFull,
74}
75
76#[derive(Clone, Eq, PartialEq, Debug)]
77pub struct BlockAndHeaderRequest<T: NodeType> {
78    pub view: ViewNumber,
79    pub epoch: EpochNumber,
80    pub parent_proposal: Proposal<T>,
81}
82
83pub struct BlockBuilderOutput<T: NodeType> {
84    pub view: ViewNumber,
85    pub epoch: EpochNumber,
86    pub payload: PayloadWithMetadata<T>,
87    pub parent_proposal: Proposal<T>,
88    pub builder_commitment: BuilderCommitment,
89    pub builder_fee: BuilderFee<T>,
90    pub payload_commitment: VidCommitment,
91    pub manifest: DedupManifest<T>,
92    /// The shares behind `payload_commitment`, until the coordinator takes them for the
93    /// disperser.
94    pub dispersal: Option<BuiltDispersal<T>>,
95}
96
97/// Commitments the leader computes for a block it built.
98pub struct BlockCommitments<T: NodeType> {
99    pub block_size: u64,
100    pub payload_commitment: VidCommitment,
101    pub builder_commitment: BuilderCommitment,
102    pub hashes: Vec<Commitment<T::Transaction>>,
103    pub dispersal: BuiltDispersal<T>,
104}
105
106/// The leader's commitment work for a built block, on the proposal's critical path. The
107/// `block_build` bench in espresso-types calls this; its spans time each step.
108///
109/// `pooled_hashes` are the pool's hashes of the transactions the payload was built from, and
110/// `stake_table` the nodes of the block's epoch. The payload commitment is folded from the
111/// block's dispersal, so the leader encodes the payload once for both. The new protocol runs
112/// from `NEW_PROTOCOL_VERSION`, whose VID scheme is always AvidmGf2, so an earlier `version` is
113/// refused.
114pub fn block_commitments<T: NodeType>(
115    payload: &PayloadWithMetadata<T>,
116    pooled_hashes: &[Commitment<T::Transaction>],
117    stake_table: &[PeerConfig<T>],
118    version: Version,
119) -> Result<BlockCommitments<T>, BlockError> {
120    if version < NEW_PROTOCOL_VERSION {
121        return Err(BlockError::UnsupportedVersion(version));
122    }
123    let _span = debug_span!("block_commitments").entered();
124    let (payload_bytes, metadata_bytes) =
125        debug_span!("encode").in_scope(|| (payload.payload.encode(), payload.metadata.encode()));
126    let block_size = payload_bytes.len() as u64;
127    // Independent work, run in parallel rather than paid for in turn on the leader's proposal
128    // path.
129    let (hashes, (dispersal, builder_commitment)) = rayon::join(
130        || transaction_hashes(payload, pooled_hashes),
131        || {
132            rayon::join(
133                || {
134                    debug_span!("vid_encode").in_scope(|| {
135                        BuiltDispersal::new(stake_table, payload_bytes, &metadata_bytes)
136                    })
137                },
138                // From 0.7 the header carries no builder commitment, so the leader skips the
139                // SHA-256 over the whole payload that produced it.
140                || {
141                    if version >= NO_BUILDER_COMMITMENT_VERSION {
142                        BuilderCommitment::empty()
143                    } else {
144                        debug_span!("builder_commitment")
145                            .in_scope(|| payload.payload.builder_commitment(&payload.metadata))
146                    }
147                },
148            )
149        },
150    );
151    let dispersal = dispersal?;
152    Ok(BlockCommitments {
153        block_size,
154        payload_commitment: VidCommitment::V2(dispersal.payload_commitment()?),
155        builder_commitment,
156        hashes,
157        dispersal,
158    })
159}
160
161/// The payload's transaction hashes, in any order: every consumer uses them as a set.
162///
163/// Building a payload can drop transactions but never changes, repeats or adds one, so a payload
164/// that kept as many transactions as the pool gave it holds exactly the pooled ones, and the
165/// pool's hashes save hashing them again. Debug builds check this.
166pub(crate) fn transaction_hashes<T: NodeType>(
167    payload: &PayloadWithMetadata<T>,
168    pooled_hashes: &[Commitment<T::Transaction>],
169) -> Vec<Commitment<T::Transaction>> {
170    if payload.payload.num_transactions(&payload.metadata) == pooled_hashes.len() {
171        debug_assert_eq!(
172            pooled_hashes.iter().copied().collect::<HashSet<_>>(),
173            payload
174                .payload
175                .transaction_commitments(&payload.metadata)
176                .into_iter()
177                .collect::<HashSet<_>>(),
178            "a payload that kept every pooled transaction holds exactly those"
179        );
180        pooled_hashes.to_vec()
181    } else {
182        debug_span!("transaction_hashes")
183            .in_scope(|| payload.payload.transaction_commitments(&payload.metadata))
184    }
185}
186
187/// Room in a forwarded message for everything but the transactions.
188const FORWARD_ENVELOPE_BYTES: u64 = 4096;
189
190/// Views between a send and the first view it targets. The next view's leader takes its block
191/// as soon as it pairs this view's proposal, before most of this view's submissions reach it,
192/// so a copy sent to it would mostly wait in its pool for a later turn.
193const SEND_LEAD: u64 = 2;
194
195pub struct BlockBuilderConfig {
196    pub max_retry_bytes: u64,
197    /// `max_block_size` per protocol version; a missing version inherits the previous one.
198    pub block_sizes: BTreeMap<Version, u64>,
199    pub ttl: u64,
200    /// Views a leader remembers included transactions. Keep it at least `ttl`, or a node still
201    /// resending a transaction gets it included again once leaders have forgotten it.
202    pub dedup_window_size: u64,
203    pub empty_block_delay: Duration,
204    /// How many upcoming leaders each transaction is sent to. A leader holds up to
205    /// `fanout + 1` blocks of transactions for its later views.
206    pub fanout: NonZeroU64,
207}
208
209impl Default for BlockBuilderConfig {
210    fn default() -> Self {
211        let ttl = 10;
212        Self {
213            max_retry_bytes: 100 * 1024 * 1024,
214            block_sizes: BTreeMap::from([(versions::version(0, 0), 2 * 1024 * 1024)]),
215            ttl,
216            dedup_window_size: ttl,
217            empty_block_delay: Duration::from_millis(500),
218            fanout: NonZeroU64::new(2).expect("2 is non-zero"),
219        }
220    }
221}
222
223/// Encoded bytes of transactions that fit in one message of `max_message_size`.
224pub fn forward_budget(max_message_size: NonZeroUsize) -> u64 {
225    (max_message_size.get() as u64).saturating_sub(FORWARD_ENVELOPE_BYTES)
226}
227
228struct RetryEntry<T: NodeType> {
229    tx: T::Transaction,
230    valid_until: ViewNumber,
231    size: u64,
232    /// Bytes on the wire, which can exceed `size`.
233    encoded_size: u64,
234    /// The last view whose leader was sent this transaction.
235    sent_until: ViewNumber,
236}
237
238struct PoolEntry<T: NodeType> {
239    tx: T::Transaction,
240    /// The view the transaction was sent for, or the view it arrived in if that is later.
241    view: ViewNumber,
242}
243
244/// Memory charged per retry entry on top of its block size, so tiny
245/// transactions cannot outgrow `max_retry_bytes`. Doubled for table slack.
246const fn retry_entry_overhead<T: NodeType>() -> u64 {
247    let map = size_of::<(Commitment<T::Transaction>, RetryEntry<T>)>();
248    let order = size_of::<(ViewNumber, Commitment<T::Transaction>)>();
249    2 * (map + order) as u64
250}
251
252/// Transactions taken from the pool for a view, with the hashes the pool keyed them by.
253#[derive(Clone)]
254struct PooledBlock<T: NodeType> {
255    txs: Arc<Vec<T::Transaction>>,
256    hashes: Arc<Vec<Commitment<T::Transaction>>>,
257}
258
259pub struct BlockBuilder<T: NodeType> {
260    instance: Arc<T::InstanceState>,
261    membership: EpochMembershipCoordinator<T>,
262    retry_pending: HashMap<Commitment<T::Transaction>, RetryEntry<T>>,
263    retry_order: BTreeSet<(ViewNumber, Commitment<T::Transaction>)>,
264    retry_total_bytes: u64,
265    leader_buffer: HashMap<Commitment<T::Transaction>, PoolEntry<T>>,
266    leader_order: BTreeSet<(ViewNumber, Commitment<T::Transaction>)>,
267    leader_total_bytes: u64,
268    dedups: BTreeMap<ViewNumber, HashSet<Commitment<T::Transaction>>>,
269    // Transactions of the blocks this node proposed or reconstructed that have not
270    // decided yet. They leave `retry_pending` only once their block decides.
271    undecided: BTreeMap<(ViewNumber, VidCommitment2), Vec<Commitment<T::Transaction>>>,
272    // Blocks that decided within the retry TTL, for a reconstruction that completes
273    // after the decide.
274    decided: BTreeSet<(ViewNumber, VidCommitment2)>,
275    config: BlockBuilderConfig,
276    upgrade_lock: UpgradeLock<T>,
277    current_view: ViewNumber,
278    // Keyed by (view, parent_proposal commitment) so that two requests for
279    // the same view but different parents (e.g. one from
280    // `handle_proposal_with_vid_share` and one from
281    // `handle_timeout_certificate`) don't dedup against each other.
282    calculations: BTreeMap<(ViewNumber, Commitment<Leaf2<T>>), AbortHandle>,
283    /// The transactions taken for the first block of each view, with their hashes.
284    ///
285    /// Every later block for the view is built from exactly these. A leader
286    /// disperses a block as soon as it is built, and peers keep one share per
287    /// leader and view, so a second block for the view is votable only if its
288    /// payload is the same. It is the same whenever the new parent does not
289    /// change how the payload is built.
290    view_transactions: BTreeMap<ViewNumber, PooledBlock<T>>,
291    tasks: JoinSet<Result<BlockBuilderOutput<T>, BlockError>>,
292    commitments_duration_metric: Option<Arc<dyn Histogram>>,
293    /// Records the four block-builder events (request, payload start, commitments
294    /// start/end), when set.
295    tracer: Option<LeaderTracerHandle>,
296}
297
298impl<T: NodeType> BlockBuilder<T> {
299    pub fn new(
300        instance: Arc<T::InstanceState>,
301        membership: EpochMembershipCoordinator<T>,
302        config: BlockBuilderConfig,
303        upgrade_lock: UpgradeLock<T>,
304    ) -> Self {
305        Self {
306            instance,
307            membership,
308            config,
309            upgrade_lock,
310            retry_pending: HashMap::new(),
311            retry_order: BTreeSet::new(),
312            retry_total_bytes: 0,
313            leader_buffer: HashMap::new(),
314            leader_order: BTreeSet::new(),
315            leader_total_bytes: 0,
316            dedups: BTreeMap::new(),
317            undecided: BTreeMap::new(),
318            decided: BTreeSet::new(),
319            current_view: ViewNumber::genesis(),
320            calculations: BTreeMap::new(),
321            view_transactions: BTreeMap::new(),
322            tasks: JoinSet::new(),
323            commitments_duration_metric: None,
324            tracer: None,
325        }
326    }
327
328    /// Registers the leader tracer; builder-style like `with_metrics`.
329    pub fn with_tracer(mut self, tracer: Option<LeaderTracerHandle>) -> Self {
330        self.tracer = tracer;
331        self
332    }
333
334    pub fn with_metrics(mut self, commitments_duration: Option<Arc<dyn Histogram>>) -> Self {
335        self.commitments_duration_metric = commitments_duration;
336        self
337    }
338
339    pub fn request_block(&mut self, request: BlockAndHeaderRequest<T>) {
340        let view = request.view;
341        let parent_commitment = proposal_commitment(&request.parent_proposal);
342        if self.calculations.contains_key(&(view, parent_commitment)) {
343            return;
344        }
345        let Ok(version) = self.upgrade_lock.version(view) else {
346            warn!(%view, "unsupported version");
347            return;
348        };
349        let epoch = request.epoch;
350        let PooledBlock {
351            txs,
352            hashes: pooled_hashes,
353        } = self.transactions_for(view);
354        let instance = self.instance.clone();
355        let membership = self.membership.clone();
356
357        let empty_block_delay = self.config.empty_block_delay;
358        let duration_metric = self.commitments_duration_metric.clone();
359        let tracer = self.tracer.clone();
360
361        trace_leader_event!(tracer, view, LeaderEvent::BlockRequestStarted);
362        let handle = self.tasks.spawn(async move {
363            // Without this an idle network produces empty blocks as fast as consensus can run
364            // them, flooding the coordinator's event queue.
365            if txs.is_empty() {
366                sleep(empty_block_delay).await;
367            }
368            trace_leader_event!(tracer, view, LeaderEvent::BlockPayloadStart);
369
370            let validated_state =
371                T::ValidatedState::from_header(&request.parent_proposal.block_header);
372            let (payload, metadata) =
373                T::BlockPayload::from_transactions(&txs, &validated_state, &instance)
374                    .await
375                    .map_err(|e| BlockError::PayloadConstruction(e.to_string()))?;
376            let payload: PayloadWithMetadata<T> = PayloadWithMetadata { payload, metadata };
377
378            let stake_table: Vec<_> = membership
379                .stake_table_for_epoch(Some(epoch))
380                .map_err(|_| BlockError::StakeTableUnavailable)?
381                .stake_table()
382                .cloned()
383                .collect();
384            let commitments = spawn_blocking(move || {
385                let measurement = duration_metric.map(Measurement::start);
386                trace_leader_event!(tracer, view, LeaderEvent::BlockCommitmentsStart);
387                let commitments =
388                    block_commitments(&payload, &pooled_hashes, &stake_table, version);
389                trace_leader_event!(tracer, view, LeaderEvent::BlockCommitmentsEnd);
390                match &commitments {
391                    Ok(_) => finish_measurement(measurement),
392                    Err(_) => ignore_measurement(measurement),
393                }
394                (payload, commitments)
395            });
396            let (payload, commitments) = match commitments.await {
397                Ok(out) => out,
398                Err(e) if e.is_panic() => resume_unwind(e.into_panic()),
399                Err(_) => return Err(BlockError::Cancelled),
400            };
401            let BlockCommitments {
402                block_size,
403                payload_commitment,
404                builder_commitment,
405                hashes,
406                dispersal,
407            } = commitments?;
408            let manifest = DedupManifest {
409                view,
410                epoch,
411                hashes,
412            };
413            let (builder_key, builder_private_key) =
414                T::BuilderSignatureKey::generated_from_seed_indexed([0u8; 32], 0);
415            let offered_fee = block_size;
416            let builder_fee = BuilderFee {
417                fee_amount: offered_fee,
418                fee_account: builder_key,
419                fee_signature: T::BuilderSignatureKey::sign_fee(
420                    &builder_private_key,
421                    offered_fee,
422                    &payload.metadata,
423                )
424                .map_err(|_| BlockError::BuilderSignature)?,
425            };
426            Ok(BlockBuilderOutput {
427                view,
428                epoch,
429                payload,
430                parent_proposal: request.parent_proposal,
431                builder_commitment,
432                builder_fee,
433                payload_commitment,
434                manifest,
435                dispersal: Some(dispersal),
436            })
437        });
438        self.calculations.insert((view, parent_commitment), handle);
439    }
440
441    fn transactions_for(&mut self, view: ViewNumber) -> PooledBlock<T> {
442        if let Some(pooled) = self.view_transactions.get(&view) {
443            return pooled.clone();
444        }
445        let (txs, hashes) = self.take_block(view);
446        let pooled = PooledBlock {
447            txs: Arc::new(txs),
448            hashes: Arc::new(hashes),
449        };
450        self.view_transactions.insert(view, pooled.clone());
451        pooled
452    }
453
454    /// Removes up to one block of pooled transactions, those sent for the earliest views first,
455    /// and returns them with their hashes.
456    fn take_block(
457        &mut self,
458        view: ViewNumber,
459    ) -> (Vec<T::Transaction>, Vec<Commitment<T::Transaction>>) {
460        let max_bytes = self.block_size(view);
461        let mut bytes = 0;
462        let mut taken = Vec::new();
463        for (_, hash) in &self.leader_order {
464            let size = self.leader_buffer[hash].tx.minimum_block_size();
465            if bytes + size > max_bytes {
466                continue;
467            }
468            bytes += size;
469            taken.push(*hash);
470        }
471        let txs = taken
472            .iter()
473            .map(|hash| {
474                self.remove_pooled(hash)
475                    .expect("hashes come from the pool's own order")
476            })
477            .collect();
478        (txs, taken)
479    }
480
481    fn remove_pooled(&mut self, hash: &Commitment<T::Transaction>) -> Option<T::Transaction> {
482        let entry = self.leader_buffer.remove(hash)?;
483        self.leader_order.remove(&(entry.view, *hash));
484        self.leader_total_bytes -= entry.tx.minimum_block_size();
485        Some(entry.tx)
486    }
487
488    pub async fn next(&mut self) -> Option<Result<BlockBuilderOutput<T>, BlockError>> {
489        loop {
490            match self.tasks.join_next().await {
491                Some(Ok(result)) => return Some(result),
492                Some(Err(err)) => {
493                    if err.is_panic() {
494                        error!(%err, "block builder task panicked");
495                    }
496                },
497                None => return None,
498            }
499        }
500    }
501
502    pub fn gc(&mut self, view_number: ViewNumber) {
503        self.calculations.retain(|(view, _), handle| {
504            if *view < view_number {
505                handle.abort();
506                false
507            } else {
508                true
509            }
510        });
511        self.view_transactions = self.view_transactions.split_off(&view_number);
512    }
513
514    pub fn fanout(&self) -> NonZeroU64 {
515        self.config.fanout
516    }
517
518    /// Starts the builder at `view` instead of genesis. Transactions can be submitted before
519    /// the first `on_view_changed`, and would otherwise target the leaders of the first views
520    /// and expire at the first view change.
521    pub fn start_at(&mut self, view: ViewNumber) {
522        self.current_view = view;
523    }
524
525    pub fn outstanding_transactions(&self) -> (usize, usize) {
526        (self.retry_pending.len(), self.retry_total_bytes as usize)
527    }
528
529    /// Returns the message for the upcoming leaders, addressed to the view `SEND_LEAD` ahead,
530    /// which the coordinator sends on to the leaders of the `fanout` views from there.
531    /// Resubmitting a queued transaction succeeds without queueing or sending it twice.
532    pub fn on_submit_transaction(
533        &mut self,
534        tx: T::Transaction,
535    ) -> Result<Option<TransactionMessage<T>>, SubmitError> {
536        let hash = tx.commit();
537
538        if self.retry_pending.contains_key(&hash) {
539            return Ok(None);
540        }
541
542        let size = tx.minimum_block_size();
543        let encoded_size = bincode::serialized_size(&tx).expect("transactions serialize");
544        // Forwarding uses the first target view's block size, which an upgrade can raise.
545        let max_bytes = self
546            .block_size(self.current_view)
547            .max(self.block_size(self.current_view + SEND_LEAD));
548        let budget = forward_budget(message_limit(max_bytes));
549        if size > max_bytes {
550            return Err(SubmitError::TooLarge {
551                size,
552                limit: max_bytes,
553            });
554        }
555        if encoded_size > budget {
556            return Err(SubmitError::TooLarge {
557                size: encoded_size,
558                limit: budget,
559            });
560        }
561        let charge = size + retry_entry_overhead::<T>();
562        if self.retry_total_bytes + charge > self.config.max_retry_bytes {
563            warn!("retry buffer full, rejecting {hash}");
564            return Err(SubmitError::RetryBufferFull);
565        }
566
567        let valid_until = self.current_view + self.config.ttl;
568        let first_target = self.current_view + SEND_LEAD;
569        let sent_until = first_target + (self.config.fanout.get() - 1);
570        let message = TransactionMessage {
571            view: first_target,
572            transactions: Vec::from([tx.clone()]),
573        };
574
575        self.retry_total_bytes += charge;
576        self.retry_order.insert((valid_until, hash));
577        self.retry_pending.insert(
578            hash,
579            RetryEntry {
580                tx,
581                valid_until,
582                size,
583                encoded_size,
584                sent_until,
585            },
586        );
587        Ok(Some(message))
588    }
589
590    pub fn on_transactions(&mut self, msg: TransactionMessage<T>) {
591        // A sender behind this node may name a view that has passed. Pooling it as of now keeps
592        // it from expiring before it can be built.
593        let view = msg.view.max(self.current_view);
594        let block_size = self.block_size(view);
595        let max_bytes = block_size.saturating_mul(self.config.fanout.get() + 1);
596        for tx in msg.transactions {
597            let hash = tx.commit();
598
599            if self.dedups.values().any(|hs| hs.contains(&hash)) {
600                continue;
601            }
602
603            if self.leader_buffer.contains_key(&hash) {
604                continue;
605            }
606
607            let size = tx.minimum_block_size();
608            // It could never be built, and would hold pool space until it expires.
609            if size > block_size {
610                continue;
611            }
612            if self.leader_total_bytes + size > max_bytes {
613                continue;
614            }
615
616            self.leader_total_bytes += size;
617            self.leader_order.insert((view, hash));
618            self.leader_buffer.insert(hash, PoolEntry { tx, view });
619        }
620    }
621
622    pub fn on_dedup_manifest(&mut self, manifest: DedupManifest<T>) {
623        let DedupManifest { view, hashes, .. } = manifest;
624        self.mark_included(view, hashes);
625    }
626
627    /// Returns the pending transactions to send again, within one block and one message,
628    /// addressed like `on_submit_transaction`. A transaction is sent again only once every
629    /// leader it went to has had its turn without including it.
630    pub fn on_view_changed(&mut self, view: ViewNumber) -> Option<TransactionMessage<T>> {
631        self.current_view = view;
632        while let Some(&(valid_until, hash)) = self.retry_order.first() {
633            if valid_until >= view {
634                break;
635            }
636            self.remove_pending(&hash);
637        }
638        self.expire_pooled(view);
639        self.expire_block_records(view);
640
641        let batch = self.resend_batch(view);
642        if batch.is_empty() {
643            return None;
644        }
645        Some(TransactionMessage {
646            view: view + SEND_LEAD,
647            transactions: batch,
648        })
649    }
650
651    fn resend_batch(&mut self, view: ViewNumber) -> Vec<T::Transaction> {
652        let first_target = view + SEND_LEAD;
653        let max_bytes = self.block_size(first_target);
654        let max_encoded = forward_budget(message_limit(max_bytes));
655        let mut batch = Vec::new();
656        let mut unfit = Vec::new();
657        let (mut bytes, mut encoded) = (0u64, 0u64);
658        for (_, hash) in &self.retry_order {
659            let entry = &self.retry_pending[hash];
660            if entry.size > max_bytes || entry.encoded_size > max_encoded {
661                unfit.push(*hash);
662                continue;
663            }
664            // A node sees a block's transactions as included only once it has reconstructed
665            // the block, about a view after its leader built it. Resending in that view would
666            // send again most of what the last leader just included.
667            if entry.sent_until + 1 >= view {
668                continue;
669            }
670            if bytes + entry.size > max_bytes || encoded + entry.encoded_size > max_encoded {
671                continue;
672            }
673            bytes += entry.size;
674            encoded += entry.encoded_size;
675            batch.push(*hash);
676        }
677        for hash in &unfit {
678            warn!(%hash, "pending transaction no longer fits a block, dropping");
679            self.remove_pending(hash);
680        }
681        let sent_until = first_target + (self.config.fanout.get() - 1);
682        batch
683            .into_iter()
684            .map(|hash| {
685                let entry = self
686                    .retry_pending
687                    .get_mut(&hash)
688                    .expect("batched hashes come from the retry buffer");
689                entry.sent_until = sent_until;
690                entry.tx.clone()
691            })
692            .collect()
693    }
694
695    /// Drops pooled transactions sent for views more than `ttl` behind `view`: their senders
696    /// have stopped retrying them.
697    fn expire_pooled(&mut self, view: ViewNumber) {
698        while let Some(&(sent_for, hash)) = self.leader_order.first() {
699            if sent_for + self.config.ttl >= view {
700                break;
701            }
702            self.remove_pooled(&hash);
703        }
704    }
705
706    /// Drops the records of blocks more than `ttl` views behind `view`: everything
707    /// submitted before such a block was built has expired by now, so its record can no
708    /// longer clear anything from `retry_pending`.
709    fn expire_block_records(&mut self, view: ViewNumber) {
710        let ttl = self.config.ttl;
711        self.undecided
712            .retain(|(block_view, _), _| *block_view + ttl >= view);
713        self.decided
714            .retain(|(block_view, _)| *block_view + ttl >= view);
715    }
716
717    fn remove_pending(&mut self, hash: &Commitment<T::Transaction>) {
718        if let Some(entry) = self.retry_pending.remove(hash) {
719            self.retry_order.remove(&(entry.valid_until, *hash));
720            self.retry_total_bytes -= entry.size + retry_entry_overhead::<T>();
721        }
722    }
723
724    /// The block size of the protocol version running at `view`.
725    fn block_size(&self, view: ViewNumber) -> u64 {
726        let version = self.upgrade_lock.version_infallible(view);
727        *self
728            .config
729            .block_sizes
730            .range(..=version)
731            .next_back()
732            .expect("block sizes start at or below the running version")
733            .1
734    }
735
736    /// Call for every block this node proposes or reconstructs, so it drops copies of the
737    /// block's transactions that reach it later. It keeps forwarding its own submissions
738    /// until [`Self::on_blocks_decided`] confirms the block: a proposed, or even locked,
739    /// block can still be abandoned, and its transactions would otherwise be lost.
740    pub fn on_block_reconstructed(
741        &mut self,
742        view: ViewNumber,
743        payload_commitment: VidCommitment2,
744        tx_commitments: Vec<Commitment<T::Transaction>>,
745    ) {
746        if self.decided.contains(&(view, payload_commitment)) {
747            // The decide came first, as it can when this node's shares arrive slowly.
748            for hash in &tx_commitments {
749                self.remove_pending(hash);
750            }
751        } else {
752            self.undecided
753                .insert((view, payload_commitment), tx_commitments.clone());
754        }
755        self.mark_included(view, tx_commitments);
756    }
757
758    /// Call with every decided block. Its transactions leave `retry_pending`. The records
759    /// of older blocks stay: a view that decides after a newer one is a gap fill, not an
760    /// abandoned branch, and records expire with the retry TTL anyway.
761    pub fn on_blocks_decided(
762        &mut self,
763        blocks: impl IntoIterator<Item = (ViewNumber, VidCommitment2)>,
764    ) {
765        for key in blocks {
766            for hash in self.undecided.remove(&key).unwrap_or_default() {
767                self.remove_pending(&hash);
768            }
769            self.decided.insert(key);
770        }
771    }
772
773    fn mark_included(&mut self, view: ViewNumber, hashes: Vec<Commitment<T::Transaction>>) {
774        for hash in &hashes {
775            self.remove_pooled(hash);
776        }
777
778        let lower_bound: ViewNumber = self
779            .current_view
780            .saturating_sub(self.config.dedup_window_size)
781            .into();
782
783        if view >= lower_bound {
784            self.dedups.entry(view).or_default().extend(hashes);
785        }
786
787        self.dedups = self.dedups.split_off(&lower_bound);
788    }
789
790    #[cfg(test)]
791    pub(crate) fn drain(&mut self, view: ViewNumber) -> Vec<T::Transaction> {
792        self.take_block(view).0
793    }
794}
795
796impl<T: NodeType> From<&BlockBuilderOutput<T>> for HeaderRequest<T> {
797    fn from(output: &BlockBuilderOutput<T>) -> Self {
798        HeaderRequest {
799            view: output.view,
800            epoch: output.epoch,
801            parent_proposal: output.parent_proposal.clone(),
802            payload_commitment: output.payload_commitment,
803            builder_commitment: output.builder_commitment.clone(),
804            metadata: output.payload.metadata.clone(),
805            builder_fee: output.builder_fee.clone(),
806        }
807    }
808}
809
810impl<T: NodeType> From<BlockBuilderOutput<T>> for ConsensusInput<T> {
811    fn from(output: BlockBuilderOutput<T>) -> Self {
812        ConsensusInput::BlockBuilt {
813            view: output.view,
814            epoch: output.epoch,
815            payload: output.payload.payload,
816            metadata: output.payload.metadata,
817            payload_commitment: output.payload_commitment,
818        }
819    }
820}
hotshot_new_protocol/block.rs

hotshot_new_protocol/
block.rs
