linera_execution/

committee.rs

// Copyright (c) Facebook, Inc. and its affiliates.
// Copyright (c) Zefchain Labs, Inc.
// SPDX-License-Identifier: Apache-2.0

use std::{borrow::Cow, collections::BTreeMap, sync::Arc};

use allocative::Allocative;
use linera_base::{
    crypto::{AccountPublicKey, ValidatorPublicKey},
    data_types::Epoch,
};
use serde::{Deserialize, Serialize};

use crate::policy::ResourceControlPolicy;

/// Public state of a validator.
#[derive(Eq, PartialEq, Hash, Clone, Debug, Serialize, Deserialize, Allocative)]
pub struct ValidatorState {
    /// The network address (in a string format understood by the networking layer).
    pub network_address: String,
    /// The voting power.
    pub votes: u64,
    /// The public key of the account associated with the validator.
    pub account_public_key: AccountPublicKey,
}

/// A set of validators (identified by their public keys) and their voting rights.
#[derive(Eq, PartialEq, Hash, Clone, Debug, Default, Allocative)]
#[cfg_attr(with_graphql, derive(async_graphql::InputObject))]
pub struct Committee {
    /// The validators in the committee.
    pub validators: BTreeMap<ValidatorPublicKey, ValidatorState>,
    /// The sum of all voting rights.
    total_votes: u64,
    /// The threshold to form a quorum.
    quorum_threshold: u64,
    /// The threshold to prove the validity of a statement. I.e. the assumption is that strictly
    /// less than `validity_threshold` are faulty.
    validity_threshold: u64,
    /// The policy agreed on for this epoch.
    policy: ResourceControlPolicy,
}

impl Serialize for Committee {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::ser::Serializer,
    {
        if serializer.is_human_readable() {
            CommitteeFull::from(self).serialize(serializer)
        } else {
            CommitteeMinimal::from(self).serialize(serializer)
        }
    }
}

impl<'de> Deserialize<'de> for Committee {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::de::Deserializer<'de>,
    {
        if deserializer.is_human_readable() {
            let committee_full = CommitteeFull::deserialize(deserializer)?;
            Committee::try_from(committee_full).map_err(serde::de::Error::custom)
        } else {
            let committee_minimal = CommitteeMinimal::deserialize(deserializer)?;
            Ok(Committee::from(committee_minimal))
        }
    }
}

#[derive(Serialize, Deserialize)]
#[serde(rename = "Committee")]
struct CommitteeFull<'a> {
    validators: Cow<'a, BTreeMap<ValidatorPublicKey, ValidatorState>>,
    total_votes: u64,
    quorum_threshold: u64,
    validity_threshold: u64,
    policy: Cow<'a, ResourceControlPolicy>,
}

#[derive(Serialize, Deserialize)]
#[serde(rename = "Committee")]
struct CommitteeMinimal<'a> {
    validators: Cow<'a, BTreeMap<ValidatorPublicKey, ValidatorState>>,
    policy: Cow<'a, ResourceControlPolicy>,
}

impl TryFrom<CommitteeFull<'static>> for Committee {
    type Error = String;

    fn try_from(committee_full: CommitteeFull) -> Result<Committee, Self::Error> {
        let CommitteeFull {
            validators,
            total_votes,
            quorum_threshold,
            validity_threshold,
            policy,
        } = committee_full;
        let committee = Committee::new(validators.into_owned(), policy.into_owned());
        if total_votes != committee.total_votes {
            Err(format!(
                "invalid committee: total_votes is {}; should be {}",
                total_votes, committee.total_votes,
            ))
        } else if quorum_threshold != committee.quorum_threshold {
            Err(format!(
                "invalid committee: quorum_threshold is {}; should be {}",
                quorum_threshold, committee.quorum_threshold,
            ))
        } else if validity_threshold != committee.validity_threshold {
            Err(format!(
                "invalid committee: validity_threshold is {}; should be {}",
                validity_threshold, committee.validity_threshold,
            ))
        } else {
            Ok(committee)
        }
    }
}

impl<'a> From<&'a Committee> for CommitteeFull<'a> {
    fn from(committee: &'a Committee) -> CommitteeFull<'a> {
        let Committee {
            validators,
            total_votes,
            quorum_threshold,
            validity_threshold,
            policy,
        } = committee;
        CommitteeFull {
            validators: Cow::Borrowed(validators),
            total_votes: *total_votes,
            quorum_threshold: *quorum_threshold,
            validity_threshold: *validity_threshold,
            policy: Cow::Borrowed(policy),
        }
    }
}

impl From<CommitteeMinimal<'static>> for Committee {
    fn from(committee_min: CommitteeMinimal) -> Committee {
        let CommitteeMinimal { validators, policy } = committee_min;
        Committee::new(validators.into_owned(), policy.into_owned())
    }
}

impl<'a> From<&'a Committee> for CommitteeMinimal<'a> {
    fn from(committee: &'a Committee) -> CommitteeMinimal<'a> {
        let Committee {
            validators,
            total_votes: _,
            quorum_threshold: _,
            validity_threshold: _,
            policy,
        } = committee;
        CommitteeMinimal {
            validators: Cow::Borrowed(validators),
            policy: Cow::Borrowed(policy),
        }
    }
}

impl Committee {
    pub fn new(
        validators: BTreeMap<ValidatorPublicKey, ValidatorState>,
        policy: ResourceControlPolicy,
    ) -> Self {
        let total_votes = validators.values().fold(0, |sum, state| sum + state.votes);
        // Let N = 3f + 1 + k such that 0 <= k <= 2. (Notably ⌊k / 3⌋ = 0 and ⌊(2 - k) / 3⌋ = 0.)
        // The following thresholds verify:
        // * ⌊2 N / 3⌋ + 1 = ⌊(6f + 2 + 2k) / 3⌋ + 1 = 2f + 1 + k + ⌊(2 - k) / 3⌋ = N - f
        // * ⌊(N + 2) / 3⌋= ⌊(3f + 3 + k) / 3⌋ = f + 1 + ⌊k / 3⌋ = f + 1
        let quorum_threshold = 2 * total_votes / 3 + 1;
        let validity_threshold = total_votes.div_ceil(3);

        Committee {
            validators,
            total_votes,
            quorum_threshold,
            validity_threshold,
            policy,
        }
    }

    #[cfg(with_testing)]
    pub fn make_simple(keys: Vec<(ValidatorPublicKey, AccountPublicKey)>) -> Self {
        let map = keys
            .into_iter()
            .map(|(validator_key, account_key)| {
                (
                    validator_key,
                    ValidatorState {
                        network_address: "Tcp:localhost:8080".to_string(),
                        votes: 100,
                        account_public_key: account_key,
                    },
                )
            })
            .collect();
        Committee::new(map, ResourceControlPolicy::default())
    }

    pub fn weight(&self, author: &ValidatorPublicKey) -> u64 {
        match self.validators.get(author) {
            Some(state) => state.votes,
            None => 0,
        }
    }

    pub fn account_keys_and_weights(&self) -> impl Iterator<Item = (AccountPublicKey, u64)> + '_ {
        self.validators
            .values()
            .map(|validator| (validator.account_public_key, validator.votes))
    }

    pub fn quorum_threshold(&self) -> u64 {
        self.quorum_threshold
    }

    pub fn validity_threshold(&self) -> u64 {
        self.validity_threshold
    }

    pub fn validators(&self) -> &BTreeMap<ValidatorPublicKey, ValidatorState> {
        &self.validators
    }

    pub fn validator_addresses(&self) -> impl Iterator<Item = (ValidatorPublicKey, &str)> {
        self.validators
            .iter()
            .map(|(name, validator)| (*name, &*validator.network_address))
    }

    pub fn total_votes(&self) -> u64 {
        self.total_votes
    }

    pub fn policy(&self) -> &ResourceControlPolicy {
        &self.policy
    }

    /// Returns a mutable reference to this committee's [`ResourceControlPolicy`].
    pub fn policy_mut(&mut self) -> &mut ResourceControlPolicy {
        &mut self.policy
    }
}

/// Process-global, append-only cache of committees by epoch.
///
/// Committees are network-global state (created by the admin chain, agreed on
/// by every validator), so caching them once per process avoids holding a
/// separate copy in every chain's execution state. The map is populated
/// lazily by `get_or_load`-style lookups in [`crate::ExecutionRuntimeContext`]
/// and in the storage layer.
#[derive(Clone, Debug, Default)]
pub struct SharedCommittees {
    map: Arc<papaya::HashMap<Epoch, Arc<Committee>>>,
}

impl SharedCommittees {
    pub fn new() -> Self {
        Self::default()
    }

    /// Returns the cached committee for `epoch`, if any.
    pub fn get(&self, epoch: Epoch) -> Option<Arc<Committee>> {
        self.map.pin().get(&epoch).cloned()
    }

    /// Inserts `committee` for `epoch`. If an entry was already present, the
    /// existing value wins and is returned (avoiding spurious clones when two
    /// callers race to populate the same epoch).
    pub fn insert(&self, epoch: Epoch, committee: Arc<Committee>) -> Arc<Committee> {
        let pinned = self.map.pin();
        match pinned.try_insert(epoch, committee) {
            Ok(inserted) => inserted.clone(),
            Err(e) => e.current.clone(),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn shared_committees_insert_and_get() {
        let shared = SharedCommittees::new();
        assert!(shared.get(Epoch(0)).is_none());
        let committee = Arc::new(Committee::default());
        let inserted = shared.insert(Epoch(0), committee.clone());
        assert!(Arc::ptr_eq(&inserted, &committee));
        let fetched = shared.get(Epoch(0)).unwrap();
        assert!(Arc::ptr_eq(&fetched, &committee));
    }

    #[test]
    fn shared_committees_insert_is_first_writer_wins() {
        let shared = SharedCommittees::new();
        let first = Arc::new(Committee::default());
        let second = Arc::new(Committee::default());
        let winner = shared.insert(Epoch(5), first.clone());
        assert!(Arc::ptr_eq(&winner, &first));
        let loser = shared.insert(Epoch(5), second.clone());
        assert!(Arc::ptr_eq(&loser, &first));
        assert!(!Arc::ptr_eq(&loser, &second));
    }

    #[test]
    fn shared_committees_clones_share_storage() {
        let a = SharedCommittees::new();
        let b = a.clone();
        let committee = Arc::new(Committee::default());
        a.insert(Epoch(1), committee.clone());
        let fetched = b.get(Epoch(1)).unwrap();
        assert!(Arc::ptr_eq(&fetched, &committee));
    }
}