hotshot_new_protocol/
fetch.rs

1use std::collections::{HashMap, HashSet};
2
3use hotshot_types::{
4    data::{VidCommitment2, ViewNumber},
5    epoch_membership::EpochMembershipCoordinator,
6    traits::{
7        BlockPayload,
8        block_contents::{BlockHeader, EncodeBytes},
9        node_implementation::NodeType,
10        signature_key::SignatureKey,
11    },
12    vote::HasViewNumber,
13};
14use tokio::task::JoinSet;
15use tracing::{debug, error, warn};
16
17use crate::{
18    consensus::Consensus,
19    message::{
20        Message, MessageType, Validated,
21        fetch::Request,
22        payload::{
23            PayloadFetchMessage, PayloadFetchResponse, PayloadRequestBody, PayloadResponseBody,
24        },
25    },
26    vid::{ObtainedPayload, expected_vid_param, matches_commitment},
27};
28
29/// The requesting half of payload recovery.
30pub struct Fetcher<T: NodeType> {
31    public_key: T::SignatureKey,
32
33    /// Outstanding fetches.
34    requested: HashMap<(ViewNumber, VidCommitment2), Fetch<T::SignatureKey>>,
35
36    /// Verifications of received payloads, surfaced by [`Fetcher::next`].
37    tasks: JoinSet<Option<ObtainedPayload<T>>>,
38}
39
40#[derive(Clone, Copy, Debug, Eq, PartialEq)]
41pub enum Retry {
42    NewRound,
43    SameRound,
44}
45
46/// The peers one fetch has involved.
47///
48/// `asked` decides who not to draw again, and a new round forgets it.
49/// `pending` is who still owes us an answer, and outlives the round.
50struct Fetch<K> {
51    asked: HashSet<K>,
52    pending: HashSet<K>,
53}
54
55impl<K> Default for Fetch<K> {
56    fn default() -> Self {
57        Self {
58            asked: HashSet::new(),
59            pending: HashSet::new(),
60        }
61    }
62}
63
64impl<T: NodeType> Fetcher<T> {
65    pub fn new(public_key: T::SignatureKey) -> Self {
66        Self {
67            public_key,
68            requested: HashMap::new(),
69            tasks: JoinSet::new(),
70        }
71    }
72
73    pub async fn next(&mut self) -> Option<ObtainedPayload<T>> {
74        loop {
75            match self.tasks.join_next().await? {
76                Ok(Some(out)) => return Some(out),
77                Ok(None) => {},
78                Err(err) => {
79                    if err.is_panic() {
80                        error!(%err, "fetched payload verification panicked");
81                    }
82                },
83            }
84        }
85    }
86
87    pub fn request(
88        &mut self,
89        view: ViewNumber,
90        commitment: VidCommitment2,
91        retry: Retry,
92        consensus: &Consensus<T>,
93        membership: &EpochMembershipCoordinator<T>,
94    ) -> Option<(T::SignatureKey, Message<T, Validated>)> {
95        let Some(target) = self.select_peer(view, commitment, retry, consensus, membership) else {
96            warn!(%view, "no peer to fetch the payload from");
97            return None;
98        };
99
100        let fetch = self.requested.entry((view, commitment)).or_default();
101
102        fetch.asked.insert(target.clone());
103        fetch.pending.insert(target.clone());
104
105        let message = Message {
106            sender: self.public_key.clone(),
107            message_type: {
108                let request = Request::new(view, PayloadRequestBody);
109                MessageType::PayloadFetch(PayloadFetchMessage::Req(request))
110            },
111        };
112
113        Some((target, message))
114    }
115
116    pub fn response(
117        &mut self,
118        response: PayloadFetchResponse,
119        sender: &T::SignatureKey,
120        consensus: &Consensus<T>,
121        membership: &EpochMembershipCoordinator<T>,
122    ) -> Vec<VidCommitment2> {
123        let view = response.view_number();
124        match response.into_body() {
125            PayloadResponseBody::NotAvailable | PayloadResponseBody::TooLarge => {
126                let refused = self.refused(view, sender);
127                if refused.is_empty() {
128                    warn!(%view, %sender, "payload refusal from unexpected peer");
129                }
130                refused
131            },
132            PayloadResponseBody::Payload {
133                commitment: payload_commitment,
134                data,
135            } => {
136                // A request names only the view, and a peer that kept another
137                // proposal's payload for it answers with that one. That is no
138                // answer to what was asked, so it counts as a refusal.
139                if !self
140                    .requested
141                    .get(&(view, payload_commitment))
142                    .is_some_and(|fetch| fetch.pending.contains(sender))
143                {
144                    let refused = self.refused(view, sender);
145                    if refused.is_empty() {
146                        warn!(%view, %sender, "payload response from unexpected peer");
147                    } else {
148                        debug!(%view, %sender, "payload response for another payload than asked");
149                    }
150                    return refused;
151                }
152
153                let Some(proposal) = consensus.proposal_with_payload(view, payload_commitment)
154                else {
155                    warn!(%view, "payload response does not match a proposal's commitment");
156                    return Vec::new();
157                };
158
159                if consensus.is_reconstructed(view, payload_commitment) {
160                    debug!(%view, "payload already obtained; dropping the response");
161                    if let Some(fetch) = self.requested.get_mut(&(view, payload_commitment)) {
162                        fetch.pending.remove(sender);
163                    }
164                    return Vec::new();
165                }
166
167                if let Some(fetch) = self.requested.get_mut(&(view, payload_commitment)) {
168                    fetch.pending.remove(sender);
169                }
170
171                let Some(param) = expected_vid_param(membership, proposal.epoch) else {
172                    warn!(%view, "no VID param for fetched payload; dropping");
173                    return Vec::new();
174                };
175
176                let epoch = proposal.epoch;
177                let metadata = proposal.block_header.metadata().clone();
178
179                self.tasks.spawn_blocking(move || {
180                    if !matches_commitment(
181                        view,
182                        &param,
183                        &metadata.encode(),
184                        &data,
185                        &payload_commitment,
186                    ) {
187                        return None;
188                    }
189                    let payload = T::BlockPayload::from_bytes(&data, &metadata);
190                    let tx_commitments = payload.transaction_commitments(&metadata);
191                    Some(ObtainedPayload {
192                        view,
193                        epoch,
194                        payload_commitment,
195                        payload,
196                        metadata,
197                        tx_commitments,
198                    })
199                });
200
201                Vec::new()
202            },
203        }
204    }
205
206    pub fn gc(&mut self, view: ViewNumber) {
207        self.requested.retain(|(v, _), _| *v > view);
208    }
209
210    /// `sender` answered a request at `view` without the payload asked for.
211    ///
212    /// Every payload pending from it at `view` counts as refused, not only the
213    /// one it answered: a peer serves one block per view (`Server::blocks`), so
214    /// one that refuses or sends another payload has none of the others either.
215    /// Returns those payloads, which are the ones worth asking another peer for.
216    fn refused(&mut self, view: ViewNumber, sender: &T::SignatureKey) -> Vec<VidCommitment2> {
217        self.requested
218            .iter_mut()
219            .filter(|((asked_view, _), _)| *asked_view == view)
220            .filter_map(|((_, commitment), fetch)| {
221                fetch.pending.remove(sender).then_some(*commitment)
222            })
223            .collect()
224    }
225
226    fn select_peer(
227        &mut self,
228        view: ViewNumber,
229        payload_commitment: VidCommitment2,
230        retry: Retry,
231        consensus: &Consensus<T>,
232        membership: &EpochMembershipCoordinator<T>,
233    ) -> Option<T::SignatureKey> {
234        use rand::seq::IteratorRandom;
235
236        let epoch = consensus
237            .proposal_with_payload(view, payload_commitment)?
238            .epoch;
239        let membership = membership.stake_table_for_epoch(Some(epoch)).ok()?;
240
241        let mut rng = rand::thread_rng();
242        let mut fetch = self.requested.get_mut(&(view, payload_commitment));
243        let me = &self.public_key;
244
245        let draw = |fetch: Option<&Fetch<T::SignatureKey>>, rng: &mut _| {
246            membership
247                .stake_table()
248                .map(|peer| T::SignatureKey::public_key(&peer.stake_table_entry))
249                .filter(|peer| peer != me && fetch.is_none_or(|fetch| !fetch.asked.contains(peer)))
250                .choose(rng)
251        };
252
253        if let Some(peer) = draw(fetch.as_deref(), &mut rng) {
254            return Some(peer);
255        }
256
257        if retry == Retry::SameRound {
258            return None;
259        }
260
261        let fetch = fetch.as_mut()?;
262        fetch.asked.clear();
263
264        draw(None, &mut rng)
265    }
266}
267
268#[cfg(test)]
269mod tests {
270    use std::collections::HashSet;
271
272    use hotshot_example_types::{
273        block_types::{TestBlockHeader, TestBlockPayload, TestMetadata, TestTransaction},
274        node_types::{TEST_VERSIONS, TestTypes},
275    };
276    use hotshot_types::{
277        data::{EpochNumber, Leaf2, VidCommitment, VidCommitment2, ViewNumber, ns_table},
278        signature_key::BLSPubKey,
279        traits::{block_contents::EncodeBytes, signature_key::SignatureKey},
280        vid::avidm_gf2::AvidmGf2Scheme,
281    };
282
283    use super::{Fetch, Fetcher, Retry};
284    use crate::{
285        message::{
286            fetch::Response,
287            payload::{PayloadFetchResponse, PayloadResponseBody},
288        },
289        tests::common::utils::{ConsensusHarness, TestData},
290        vid::expected_vid_param,
291    };
292
293    fn key(index: u64) -> BLSPubKey {
294        BLSPubKey::generated_from_seed_indexed([0; 32], index).0
295    }
296
297    fn view() -> ViewNumber {
298        ViewNumber::new(1)
299    }
300
301    /// A payload, and a proposal at view 1 that commits to it.
302    ///
303    /// The commitment is computed the way a disperser computes it, from the
304    /// committee's erasure parameters, so `response` derives the same one from
305    /// the proposal and the stake table.
306    async fn payload_and_consensus() -> (Vec<u8>, VidCommitment2, ConsensusHarness) {
307        let mut harness = ConsensusHarness::new(0).await;
308        let test_data = TestData::new(1).await;
309
310        let metadata = TestMetadata {
311            num_transactions: 1,
312        };
313        let payload = TestBlockPayload {
314            transactions: vec![TestTransaction::new(vec![1, 2, 3])],
315        };
316        let bytes = payload.encode().to_vec();
317
318        let param = expected_vid_param(&harness.membership_coordinator, EpochNumber::genesis())
319            .expect("committee resolves");
320        let ns_table = ns_table::parse_ns_table(bytes.len(), &metadata.encode());
321        let (commitment, _) = AvidmGf2Scheme::commit(&param, &bytes, ns_table).expect("commit");
322
323        let template = &test_data.views[0];
324        let parent_leaf: Leaf2<TestTypes> = template.proposal.data.clone().into();
325        let mut proposal = template.proposal.data.clone();
326        proposal.block_header = TestBlockHeader::new(
327            &parent_leaf,
328            VidCommitment::V2(commitment),
329            template
330                .proposal
331                .data
332                .block_header
333                .builder_commitment
334                .clone(),
335            metadata,
336            TEST_VERSIONS.test.base,
337        );
338
339        harness.force_set_proposal(view(), proposal);
340        (bytes, commitment, harness)
341    }
342
343    fn fetcher_expecting(peer: BLSPubKey, commitment: VidCommitment2) -> Fetcher<TestTypes> {
344        let mut fetcher = Fetcher::new(key(0));
345        fetcher.requested.insert(
346            (view(), commitment),
347            Fetch {
348                asked: HashSet::from([peer]),
349                pending: HashSet::from([peer]),
350            },
351        );
352        fetcher
353    }
354
355    fn payload_response(commitment: VidCommitment2, data: Vec<u8>) -> PayloadFetchResponse {
356        Response::new(view(), PayloadResponseBody::Payload { commitment, data })
357    }
358
359    fn accept(
360        fetcher: &mut Fetcher<TestTypes>,
361        response: PayloadFetchResponse,
362        sender: BLSPubKey,
363        harness: &ConsensusHarness,
364    ) -> bool {
365        !fetcher
366            .response(
367                response,
368                &sender,
369                &harness.consensus,
370                &harness.membership_coordinator,
371            )
372            .is_empty()
373    }
374
375    /// The payload a peer we asked sends, which re-commits to what our own
376    /// proposal names, comes back out of the fetcher.
377    #[tokio::test]
378    async fn a_payload_that_recommits_is_yielded() {
379        let (bytes, commitment, harness) = payload_and_consensus().await;
380        let peer = key(1);
381        let mut fetcher = fetcher_expecting(peer, commitment);
382
383        assert!(!accept(
384            &mut fetcher,
385            payload_response(commitment, bytes),
386            peer,
387            &harness
388        ));
389
390        let obtained = fetcher.next().await.expect("payload is verified");
391        assert_eq!(obtained.view, view());
392        assert_eq!(obtained.payload_commitment, commitment);
393    }
394
395    /// Bytes that do not re-commit are dropped, however well-formed the rest
396    /// of the response is. Nothing but the commitment vouches for them.
397    #[tokio::test]
398    async fn a_payload_that_does_not_recommit_is_dropped() {
399        let (mut bytes, commitment, harness) = payload_and_consensus().await;
400        bytes.push(0);
401        let peer = key(1);
402        let mut fetcher = fetcher_expecting(peer, commitment);
403
404        accept(
405            &mut fetcher,
406            payload_response(commitment, bytes),
407            peer,
408            &harness,
409        );
410
411        assert!(fetcher.next().await.is_none());
412    }
413
414    /// A peer we did not ask does not get to spend our verification, and does
415    /// not get to send us to another peer either.
416    #[tokio::test]
417    async fn an_unsolicited_payload_is_not_verified() {
418        let (bytes, commitment, harness) = payload_and_consensus().await;
419        let mut fetcher = fetcher_expecting(key(1), commitment);
420
421        assert!(!accept(
422            &mut fetcher,
423            payload_response(commitment, bytes),
424            key(2),
425            &harness
426        ));
427
428        assert!(fetcher.next().await.is_none());
429        assert!(
430            fetcher
431                .requested
432                .get(&(view(), commitment))
433                .is_some_and(|fetch| fetch.pending.contains(&key(1))),
434            "the peer we asked still owes us an answer"
435        );
436    }
437
438    /// A response naming another commitment than the one asked for is not the
439    /// payload we wanted, so it counts as a refusal.
440    ///
441    /// A request names only the view, and a peer that kept another proposal's
442    /// payload for it answers with that one. Waiting on it would stall the fetch,
443    /// so another peer is drawn, once: the answer uses up the peer's slot.
444    #[tokio::test]
445    async fn a_response_for_another_commitment_is_a_refusal() {
446        let (bytes, commitment, harness) = payload_and_consensus().await;
447        let peer = key(1);
448        let mut fetcher = fetcher_expecting(peer, commitment);
449
450        assert!(accept(
451            &mut fetcher,
452            payload_response(VidCommitment2::default(), bytes.clone()),
453            peer,
454            &harness
455        ));
456        assert!(
457            !fetcher
458                .requested
459                .get(&(view(), commitment))
460                .is_some_and(|fetch| fetch.pending.contains(&peer)),
461            "the peer no longer owes us an answer"
462        );
463        assert!(!accept(
464            &mut fetcher,
465            payload_response(VidCommitment2::default(), bytes),
466            peer,
467            &harness
468        ));
469
470        assert!(fetcher.next().await.is_none());
471    }
472
473    /// Answering consumes the peer's slot, so one request buys one
474    /// verification and a peer cannot keep sending payloads to verify.
475    #[tokio::test]
476    async fn answering_consumes_the_peers_slot() {
477        let (bytes, commitment, harness) = payload_and_consensus().await;
478        let peer = key(1);
479        let mut fetcher = fetcher_expecting(peer, commitment);
480
481        accept(
482            &mut fetcher,
483            payload_response(commitment, bytes.clone()),
484            peer,
485            &harness,
486        );
487        accept(
488            &mut fetcher,
489            payload_response(commitment, bytes),
490            peer,
491            &harness,
492        );
493
494        assert!(fetcher.next().await.is_some());
495        assert!(
496            fetcher.next().await.is_none(),
497            "the second response was not verified"
498        );
499    }
500
501    /// The draw excludes this node and everyone this fetch already asked, so
502    /// a peer that stays silent is not drawn again while the round lasts.
503    #[tokio::test]
504    async fn the_draw_excludes_this_node_and_the_peers_already_asked() {
505        let (_, commitment, harness) = payload_and_consensus().await;
506        let mut fetcher = Fetcher::new(key(0));
507        let mut drawn = HashSet::new();
508
509        // Ten committee members, one of them us.
510        for _ in 0..9 {
511            let peer = fetcher
512                .select_peer(
513                    view(),
514                    commitment,
515                    Retry::SameRound,
516                    &harness.consensus,
517                    &harness.membership_coordinator,
518                )
519                .expect("a peer this round has not asked");
520            assert!(peer != key(0), "never ourselves");
521            assert!(drawn.insert(peer), "never the same peer twice");
522            fetcher
523                .requested
524                .entry((view(), commitment))
525                .or_default()
526                .asked
527                .insert(peer);
528        }
529        assert_eq!(drawn.len(), 9);
530    }
531
532    /// Once every peer has been asked the round is over, and only the
533    /// periodic retry may start another one. A refusal must not, or peers
534    /// refusing at message speed would keep the draw going round the
535    /// committee.
536    #[tokio::test]
537    async fn only_a_new_round_starts_the_draw_over() {
538        let (_, commitment, harness) = payload_and_consensus().await;
539        let mut fetcher = Fetcher::new(key(0));
540        let everyone: HashSet<BLSPubKey> = (1..10).map(key).collect();
541        fetcher.requested.insert(
542            (view(), commitment),
543            Fetch {
544                asked: everyone.clone(),
545                pending: HashSet::new(),
546            },
547        );
548
549        assert!(
550            fetcher
551                .select_peer(
552                    view(),
553                    commitment,
554                    Retry::SameRound,
555                    &harness.consensus,
556                    &harness.membership_coordinator,
557                )
558                .is_none(),
559            "the round is exhausted"
560        );
561
562        let peer = fetcher
563            .select_peer(
564                view(),
565                commitment,
566                Retry::NewRound,
567                &harness.consensus,
568                &harness.membership_coordinator,
569            )
570            .expect("a new round draws again");
571        assert!(everyone.contains(&peer));
572        assert!(
573            fetcher.requested[&(view(), commitment)].asked.is_empty(),
574            "the new round forgot who was asked"
575        );
576    }
577
578    /// Only a peer that owes us an answer sends us to another one, and only
579    /// once: refusing spends the slot, so a peer cannot repeat it to keep the
580    /// draw going round the committee.
581    #[tokio::test]
582    async fn only_a_peer_we_asked_redraws_and_only_once() {
583        let (_, commitment, harness) = payload_and_consensus().await;
584        let peer = key(1);
585        let mut fetcher = fetcher_expecting(peer, commitment);
586        let refusal = || Response::new(view(), PayloadResponseBody::NotAvailable);
587
588        assert!(!accept(&mut fetcher, refusal(), key(2), &harness));
589        assert!(accept(&mut fetcher, refusal(), peer, &harness));
590        assert!(
591            !accept(&mut fetcher, refusal(), peer, &harness),
592            "the slot was spent by the first refusal"
593        );
594        assert!(
595            fetcher
596                .requested
597                .get(&(view(), commitment))
598                .is_some_and(|fetch| fetch.asked.contains(&peer)),
599            "a spent peer is still not drawn again"
600        );
601    }
602}
hotshot_new_protocol/fetch.rs

hotshot_new_protocol/
fetch.rs
