zksync_consensus_roles/validator/messages/v2/
replica_timeout.rs

1use std::collections::{BTreeMap, HashMap};
2
3use anyhow::Context as _;
4use zksync_protobuf::{read_optional, read_required, ProtoFmt};
5
6use super::{
7    BlockHeader, CommitQC, CommitQCVerifyError, ReplicaCommit, ReplicaCommitVerifyError, Signers,
8    View,
9};
10use crate::{
11    proto::validator as proto,
12    validator::{self, EpochNumber, GenesisHash, Signed},
13};
14
15/// A timeout message from a replica.
16/// WARNING: any change to this struct may invalidate preexisting signatures. See `TimeoutQC` docs.
17#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
18pub struct ReplicaTimeout {
19    /// View of this message.
20    pub view: View,
21    /// The highest block that the replica has committed to.
22    pub high_vote: Option<ReplicaCommit>,
23    /// The highest CommitQC that the replica has seen.
24    pub high_qc: Option<CommitQC>,
25}
26
27impl ReplicaTimeout {
28    /// Verifies the message.
29    pub fn verify(
30        &self,
31        genesis_hash: GenesisHash,
32        epoch: EpochNumber,
33        validators_schedule: &validator::Schedule,
34    ) -> Result<(), ReplicaTimeoutVerifyError> {
35        self.view
36            .verify(genesis_hash, epoch)
37            .map_err(ReplicaTimeoutVerifyError::BadView)?;
38
39        if let Some(v) = &self.high_vote {
40            v.verify(genesis_hash, epoch)
41                .map_err(ReplicaTimeoutVerifyError::InvalidHighVote)?;
42        }
43
44        if let Some(qc) = &self.high_qc {
45            qc.verify(genesis_hash, epoch, validators_schedule)
46                .map_err(ReplicaTimeoutVerifyError::InvalidHighQC)?;
47        }
48
49        Ok(())
50    }
51}
52
53impl ProtoFmt for ReplicaTimeout {
54    type Proto = proto::ReplicaTimeoutV2;
55
56    fn read(r: &Self::Proto) -> anyhow::Result<Self> {
57        Ok(Self {
58            view: read_required(&r.view).context("view")?,
59            high_vote: read_optional(&r.high_vote).context("high_vote")?,
60            high_qc: read_optional(&r.high_qc).context("high_qc")?,
61        })
62    }
63
64    fn build(&self) -> Self::Proto {
65        Self::Proto {
66            view: Some(self.view.build()),
67            high_vote: self.high_vote.as_ref().map(ProtoFmt::build),
68            high_qc: self.high_qc.as_ref().map(ProtoFmt::build),
69        }
70    }
71}
72
73/// Error returned by `ReplicaTimeout::verify()`.
74#[derive(thiserror::Error, Debug)]
75pub enum ReplicaTimeoutVerifyError {
76    /// View.
77    #[error("view: {0:#}")]
78    BadView(anyhow::Error),
79    /// Invalid High Vote.
80    #[error("invalid high_vote: {0:#}")]
81    InvalidHighVote(ReplicaCommitVerifyError),
82    /// Invalid High QC.
83    #[error("invalid high_qc: {0:#}")]
84    InvalidHighQC(CommitQCVerifyError),
85}
86
87/// A quorum certificate of ReplicaTimeout messages. Since not all ReplicaTimeout messages are
88/// identical (they have different high blocks and high QCs), we need to keep the ReplicaTimeout
89/// messages in a map. We can still aggregate the signatures though.
90///
91/// WARNING: `TimeoutQC` message contains a map indexed by `ReplicaTimeout` messages.
92/// Therefore, `Ord` implementation of `ReplicaTimeout` affects the unique encoding of the `TimeoutQC` message.
93/// This `Ord` implementation is the derived lexicographic ordering of the fields of
94/// `ReplicaTimeout` (and transitively on types of the fields AS WELL).
95/// As a result ANY change to type of ANY transitive field of ReplicaTimeout struct may invalidate
96/// preexisting signatures of `TimeoutQC` messages (or messages containing `TimeoutQC`).
97///
98/// The proper fix would be to add support for unordered collections on the protobuf level
99/// (for example, by adding a custom option "unordered" for repeated fields, which would make the encoder
100/// sort the encoded elements before encoding the whole message). However the current protobuf
101/// encoding of TimeoutQC keeps the keys and values in separate repeated fields. Until this is
102/// fixed, ANY change to the above mentioned types may be backward incompatible.
103#[derive(Clone, Debug, PartialEq, Eq)]
104pub struct TimeoutQC {
105    /// View of this QC.
106    pub view: View,
107    /// Map from replica Timeout messages to the validators that signed them.
108    pub map: BTreeMap<ReplicaTimeout, Signers>,
109    /// Aggregate signature of the ReplicaTimeout messages.
110    pub signature: validator::AggregateSignature,
111}
112
113impl TimeoutQC {
114    /// Create a new empty TimeoutQC for a given view.
115    pub fn new(view: View) -> Self {
116        Self {
117            view,
118            map: BTreeMap::new(),
119            signature: validator::AggregateSignature::default(),
120        }
121    }
122
123    /// Get the highest block voted and check if there's a subquorum of votes for it. To have a subquorum
124    /// in this situation, we require n-3*f votes, where f is the maximum number of faulty replicas.
125    /// Note that it is possible to have 2 subquorums: vote A and vote B, each with >n-3*f weight, in a single
126    /// TimeoutQC. In such a situation we say that there is no high vote.
127    pub fn high_vote(&self, validators_schedule: &validator::Schedule) -> Option<BlockHeader> {
128        let mut count: HashMap<_, u64> = HashMap::new();
129        for (msg, signers) in &self.map {
130            if let Some(v) = &msg.high_vote {
131                *count.entry(v.proposal).or_default() += signers.weight(validators_schedule);
132            }
133        }
134
135        let min = validators_schedule.subquorum_threshold();
136        let mut high_votes: Vec<_> = count.into_iter().filter(|x| x.1 >= min).collect();
137
138        if high_votes.len() == 1 {
139            high_votes.pop().map(|x| x.0)
140        } else {
141            None
142        }
143    }
144
145    /// Get the highest CommitQC.
146    pub fn high_qc(&self) -> Option<&CommitQC> {
147        self.map
148            .keys()
149            .filter_map(|m| m.high_qc.as_ref())
150            .max_by_key(|qc| qc.view().number)
151    }
152
153    /// Add a validator's signed message. This also verifies the message and the signature before adding.
154    pub fn add(
155        &mut self,
156        msg: &Signed<ReplicaTimeout>,
157        genesis_hash: GenesisHash,
158        epoch: EpochNumber,
159        validators_schedule: &validator::Schedule,
160    ) -> Result<(), TimeoutQCAddError> {
161        // Check if the signer is in the committee.
162        let Some(i) = validators_schedule.index(&msg.key) else {
163            return Err(TimeoutQCAddError::SignerNotInCommittee {
164                signer: Box::new(msg.key.clone()),
165            });
166        };
167
168        // Check if we already have a message from the same signer.
169        if self.map.values().any(|s| s.0[i]) {
170            return Err(TimeoutQCAddError::DuplicateSigner {
171                signer: Box::new(msg.key.clone()),
172            });
173        };
174
175        // Verify the signature.
176        msg.verify().map_err(TimeoutQCAddError::BadSignature)?;
177
178        // Check that the view is consistent with the TimeoutQC.
179        if msg.msg.view != self.view {
180            return Err(TimeoutQCAddError::InconsistentViews);
181        };
182
183        // Check that the message itself is valid.
184        msg.msg
185            .verify(genesis_hash, epoch, validators_schedule)
186            .map_err(TimeoutQCAddError::InvalidMessage)?;
187
188        // Add the message plus signer to the map, and the signature to the aggregate signature.
189        let e = self
190            .map
191            .entry(msg.msg.clone())
192            .or_insert_with(|| Signers::new(validators_schedule.len()));
193        e.0.set(i, true);
194        self.signature.add(&msg.sig);
195
196        Ok(())
197    }
198
199    /// Verifies the integrity of the TimeoutQC.
200    pub fn verify(
201        &self,
202        genesis_hash: GenesisHash,
203        epoch: EpochNumber,
204        validators_schedule: &validator::Schedule,
205    ) -> Result<(), TimeoutQCVerifyError> {
206        self.view
207            .verify(genesis_hash, epoch)
208            .map_err(TimeoutQCVerifyError::BadView)?;
209
210        let mut sum = Signers::new(validators_schedule.len());
211
212        // Check the ReplicaTimeout messages.
213        for (i, (msg, signers)) in self.map.iter().enumerate() {
214            if msg.view != self.view {
215                return Err(TimeoutQCVerifyError::InconsistentView(i));
216            }
217            if signers.len() != sum.len() {
218                return Err(TimeoutQCVerifyError::WrongSignersLength(i));
219            }
220            if signers.is_empty() {
221                return Err(TimeoutQCVerifyError::NoSignersAssigned(i));
222            }
223            if !(&sum & signers).is_empty() {
224                return Err(TimeoutQCVerifyError::OverlappingSignatureSet(i));
225            }
226            msg.verify(genesis_hash, epoch, validators_schedule)
227                .map_err(|err| TimeoutQCVerifyError::InvalidMessage(i, err))?;
228
229            sum |= signers;
230        }
231
232        // Check if the signers' weight is enough.
233        let weight = sum.weight(validators_schedule);
234        let threshold = validators_schedule.quorum_threshold();
235        if weight < threshold {
236            return Err(TimeoutQCVerifyError::NotEnoughWeight {
237                got: weight,
238                want: threshold,
239            });
240        }
241
242        // Now we can verify the signature.
243        let messages_and_keys = self.map.clone().into_iter().flat_map(|(msg, signers)| {
244            validators_schedule
245                .keys()
246                .enumerate()
247                .filter(|(i, _)| signers.0[*i])
248                .map(|(_, pk)| (msg.clone(), pk))
249                .collect::<Vec<_>>()
250        });
251
252        // TODO: This reaggregating is suboptimal.
253        self.signature
254            .verify_messages(messages_and_keys)
255            .map_err(TimeoutQCVerifyError::BadSignature)
256    }
257
258    /// Calculates the weight of current TimeoutQC signing validators
259    pub fn weight(&self, validators_schedule: &validator::Schedule) -> u64 {
260        self.map
261            .values()
262            .map(|signers| signers.weight(validators_schedule))
263            .sum()
264    }
265}
266
267impl Ord for TimeoutQC {
268    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
269        self.view.number.cmp(&other.view.number)
270    }
271}
272
273impl PartialOrd for TimeoutQC {
274    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
275        Some(self.cmp(other))
276    }
277}
278
279impl ProtoFmt for TimeoutQC {
280    type Proto = proto::TimeoutQcv2;
281
282    fn read(r: &Self::Proto) -> anyhow::Result<Self> {
283        let mut map = BTreeMap::new();
284
285        for (msg, signers) in r.msgs.iter().zip(r.signers.iter()) {
286            map.insert(
287                ReplicaTimeout::read(msg).context("msg")?,
288                Signers::read(signers).context("signers")?,
289            );
290        }
291
292        Ok(Self {
293            view: read_required(&r.view).context("view")?,
294            map,
295            signature: read_required(&r.sig).context("sig")?,
296        })
297    }
298
299    fn build(&self) -> Self::Proto {
300        let (msgs, signers) = self
301            .map
302            .iter()
303            .map(|(msg, signers)| (msg.build(), signers.build()))
304            .unzip();
305
306        Self::Proto {
307            view: Some(self.view.build()),
308            msgs,
309            signers,
310            sig: Some(self.signature.build()),
311        }
312    }
313}
314
315/// Error returned by `TimeoutQC::add()`.
316#[derive(thiserror::Error, Debug)]
317pub enum TimeoutQCAddError {
318    /// Signer not present in the committee.
319    #[error("Signer not in committee: {signer:?}")]
320    SignerNotInCommittee {
321        /// Signer of the message.
322        signer: Box<validator::PublicKey>,
323    },
324    /// Message from the same signer already present in QC.
325    #[error("Message from the same signer already in QC: {signer:?}")]
326    DuplicateSigner {
327        /// Signer of the message.
328        signer: Box<validator::PublicKey>,
329    },
330    /// Bad signature.
331    #[error("Bad signature: {0:#}")]
332    BadSignature(#[source] anyhow::Error),
333    /// Inconsistent views.
334    #[error("Trying to add a message from a different view")]
335    InconsistentViews,
336    /// Invalid message.
337    #[error("Invalid message: {0:#}")]
338    InvalidMessage(ReplicaTimeoutVerifyError),
339}
340
341/// Error returned by `TimeoutQC::verify()`.
342#[derive(thiserror::Error, Debug)]
343pub enum TimeoutQCVerifyError {
344    /// Bad view.
345    #[error("Bad view: {0:#}")]
346    BadView(anyhow::Error),
347    /// Inconsistent views.
348    #[error("Message with inconsistent view: number [{0}]")]
349    InconsistentView(usize),
350    /// Invalid message.
351    #[error("Invalid message: number [{0}], {1:#}")]
352    InvalidMessage(usize, ReplicaTimeoutVerifyError),
353    /// Wrong signers length.
354    #[error("Message with wrong signers length: number [{0}]")]
355    WrongSignersLength(usize),
356    /// No signers assigned.
357    #[error("Message with no signers assigned: number [{0}]")]
358    NoSignersAssigned(usize),
359    /// Overlapping signature sets.
360    #[error("Message with overlapping signature set: number [{0}]")]
361    OverlappingSignatureSet(usize),
362    /// Weight not reached.
363    #[error("Signers have not reached threshold weight: got {got}, want {want}")]
364    NotEnoughWeight {
365        /// Got weight.
366        got: u64,
367        /// Want weight.
368        want: u64,
369    },
370    /// Bad signature.
371    #[error("Bad signature: {0:#}")]
372    BadSignature(#[source] anyhow::Error),
373}
zksync_consensus_roles/validator/messages/v2/replica_timeout.rs

zksync_consensus_roles/validator/messages/v2/
replica_timeout.rs
