cfxcore/sync/message/
state_sync_candidate_request.rs

1use crate::{
2    message::{
3        GetMaybeRequestId, Message, MessageProtocolVersionBound, MsgId,
4        RequestId, SetRequestId,
5    },
6    sync::{
7        message::{
8            msgid, Context, DynamicCapability, Handleable, KeyContainer,
9            StateSyncCandidateResponse,
10        },
11        request_manager::{AsAny, Request},
12        state::storage::SnapshotSyncCandidate,
13        Error, ProtocolConfiguration, SYNC_PROTO_V1, SYNC_PROTO_V3,
14    },
15};
16use malloc_size_of_derive::MallocSizeOf as DeriveMallocSizeOf;
17use network::service::ProtocolVersion;
18use rlp::Encodable;
19use rlp_derive::{RlpDecodable, RlpEncodable};
20use std::{any::Any, time::Duration};
21
22#[derive(Clone, RlpEncodable, RlpDecodable, Debug, DeriveMallocSizeOf)]
23pub struct StateSyncCandidateRequest {
24    pub request_id: RequestId,
25    pub candidates: Vec<SnapshotSyncCandidate>,
26}
27
28build_msg_with_request_id_impl! {
29    StateSyncCandidateRequest, msgid::STATE_SYNC_CANDIDATE_REQUEST,
30    "StateSyncCandidateRequest", SYNC_PROTO_V1, SYNC_PROTO_V3
31}
32
33impl Handleable for StateSyncCandidateRequest {
34    fn handle(self, ctx: &Context) -> Result<(), Error> {
35        let mut supported_candidates =
36            Vec::with_capacity(self.candidates.len());
37        let storage_manager = ctx
38            .manager
39            .graph
40            .data_man
41            .storage_manager
42            .get_storage_manager();
43        for candidate in self.candidates {
44            match candidate {
45                SnapshotSyncCandidate::FullSync {
46                    height,
47                    snapshot_epoch_id,
48                } => {
49                    match storage_manager
50                        .get_snapshot_info_at_epoch(&snapshot_epoch_id)
51                    {
52                        Some(snapshot_info) => {
53                            if snapshot_info.height == height {
54                                supported_candidates.push(
55                                    SnapshotSyncCandidate::FullSync {
56                                        height,
57                                        snapshot_epoch_id,
58                                    },
59                                );
60                            } else {
61                                warn!(
62                                    "Invalid SnapshotSyncCandidate, height unmatch: get {:?}, \
63                                    local_height of the snapshot is {}",
64                                    candidate, snapshot_info.height);
65                            }
66                        }
67                        None => {
68                            debug!(
69                                "Requested snapshot not exist: {:?}",
70                                candidate
71                            );
72                        }
73                    }
74                }
75            }
76        }
77        ctx.send_response(&StateSyncCandidateResponse {
78            request_id: self.request_id,
79            supported_candidates,
80        })?;
81
82        Ok(())
83    }
84}
85
86impl AsAny for StateSyncCandidateRequest {
87    fn as_any(&self) -> &dyn Any { self }
88
89    fn as_any_mut(&mut self) -> &mut dyn Any { self }
90}
91
92impl Request for StateSyncCandidateRequest {
93    fn timeout(&self, conf: &ProtocolConfiguration) -> Duration {
94        conf.snapshot_candidate_request_timeout
95    }
96
97    fn on_removed(&self, _inflight_keys: &KeyContainer) {}
98
99    fn with_inflight(&mut self, _inflight_keys: &KeyContainer) {}
100
101    fn is_empty(&self) -> bool { false }
102
103    fn resend(&self) -> Option<Box<dyn Request>> { None }
104
105    fn required_capability(&self) -> Option<DynamicCapability> { None }
106}
cfxcore/sync/message/state_sync_candidate_request.rs

cfxcore/sync/message/
state_sync_candidate_request.rs
