use crate::{
dkg::{
ParticipantsProvider, Registrar, ReshareBlock, SecretStore,
fence::Fence,
network::{
AddressableManager, Addresses, Directory, Manager as DkgManager, MissingAddress,
},
orchestrator, probe as dkg_probe,
reshare::{self, Input as ReshareInput},
state_sync::{Config as StateSyncConfig, Plan as StateSyncPlan, StateSync},
tests::{
max_supported_mode,
mocks::{FilteredReceiver, MemorySecretStore},
},
types::*,
},
simulate::{
engine::{EngineDefinition, InitContext},
processed::ProcessedHeight,
reporter::MonitorReporter,
},
stateful::{
Application, Config as StatefulConfig, Input, Proposed, Stateful as StatefulActor,
SyncPlan,
db::{
DatabaseSet, Merkleized as _, Shared, SyncEngineConfig, Unmerkleized as _,
p2p as qmdb_resolver,
},
},
};
use commonware_broadcast::buffered;
use commonware_codec::{
Encode, EncodeSize, Error as CodecError, RangeCfg, Read, ReadExt as _, Write,
};
use commonware_consensus::{
Block as ConsensusBlock, CertifiableBlock, Heightable, Reporters,
marshal::{
self,
ancestry::Ancestry,
core::{Actor as MarshalActor, CommitmentFallback, Mailbox as MarshalMailbox},
resolver::p2p as marshal_resolver,
standard::{Deferred, Standard},
},
simplex::{
self,
config::{ForwardPolicy, SkipPolicy},
elector::RoundRobin,
types::{Context, Finalization},
},
types::{Epoch, Epocher as _, FixedEpocher, Height, Round, View, ViewDelta},
};
use commonware_cryptography::{
Digest as _, Digestible, Hasher, Sha256, Signer as _,
bls12381::{
dkg::feldman_desmedt::{Reveal, deal},
primitives::{group::Share, sharing::Mode, variant::MinPk},
},
certificate::{ConstantProvider, Provider as CertificateProvider, Scoped},
ed25519,
sha256::{self, Digest as Sha256Digest},
};
use commonware_formatting::hex;
use commonware_math::algebra::Random;
use commonware_p2p::{Address, Provider, TrackedPeers, simulated};
use commonware_parallel::Sequential;
use commonware_runtime::{
Buf, BufMut, BufferPooler, Clock, Handle, Metrics, Quota, Spawner, Storage, Supervisor as _,
buffer::paged::CacheRef, deterministic::Context as DeterministicContext,
};
use commonware_storage::{
archive::prunable,
journal::contiguous::fixed::Config as FixedLogConfig,
mmr::{self, Location, full::Config as MmrJournalConfig},
qmdb::{
any::{FixedConfig, unordered::fixed},
sync::Target,
},
translator::TwoCap,
};
use commonware_utils::{
N3f1, NZDuration, NZU16, NZU32, NZU64, NZUsize, TestRng, non_empty_range,
ordered::{Map, Set},
range::NonEmptyRange,
sequence::Unit,
sync::Mutex,
test_rng,
};
use rand::Rng;
use std::{
collections::{BTreeMap, HashMap, HashSet, btree_map::Entry},
marker::PhantomData,
net::{IpAddr, Ipv4Addr, SocketAddr},
num::{NonZeroU16, NonZeroU32, NonZeroU64, NonZeroUsize},
sync::Arc,
time::Duration,
};
type Qmdb<E> =
fixed::Db<mmr::Family, E, sha256::Digest, sha256::Digest, Sha256, TwoCap, Sequential>;
type Database<E> = Shared<Qmdb<E>>;
type Scheme = simplex::scheme::bls12381_threshold::vrf::Scheme<ed25519::PublicKey, MinPk>;
type MarshalVariant = Standard<Block>;
type Marshal = MarshalMailbox<Scheme, MarshalVariant>;
type PublicKey = ed25519::PublicKey;
type DiscoveryManager = simulated::Manager<PublicKey, DeterministicContext>;
type LookupManager = AddressableManager<simulated::SocketManager<PublicKey, DeterministicContext>>;
pub(super) const EPOCH_LENGTH: NonZeroU64 = NZU64!(32);
const NAMESPACE: &[u8] = b"_COMMONWARE_GLUE_DKG_RESHARE_E2E";
const PAGE_SIZE: NonZeroU16 = NZU16!(1024);
const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(10);
const IO_BUFFER_SIZE: NonZeroUsize = NZUsize!(2048);
const TEST_QUOTA: Quota = Quota::per_second(NZU32!(1_000_000));
const MAX_PARTICIPANTS: NonZeroU32 = NZU32!(16);
const VOTE_CHANNEL: u64 = 0;
const CERTIFICATE_CHANNEL: u64 = 1;
const RESOLVER_CHANNEL: u64 = 2;
const BACKFILL_CHANNEL: u64 = 3;
const BROADCAST_CHANNEL: u64 = 4;
const QMDB_CHANNEL: u64 = 5;
const DKG_CHANNEL: u64 = 6;
const DKG_PROBE_CHANNEL: u64 = 7;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum Network {
Discovery,
Lookup,
}
impl Network {
fn directory(
self,
peers: &Set<PublicKey>,
addresses: &Map<PublicKey, Address>,
) -> TestDirectory {
match self {
Self::Discovery => TestDirectory(None),
Self::Lookup => TestDirectory(Some(
peers
.iter()
.map(|peer| {
let address = addresses
.get_value(peer)
.expect("participant must have an address")
.clone();
(peer.clone(), address)
})
.collect(),
)),
}
}
fn manager(self, oracle: &simulated::Oracle<PublicKey, DeterministicContext>) -> TestManager {
match self {
Self::Discovery => TestManager::Discovery(oracle.manager()),
Self::Lookup => TestManager::Lookup(AddressableManager::new(oracle.socket_manager())),
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) struct TestDirectory(Option<Addresses<PublicKey>>);
impl Write for TestDirectory {
fn write(&self, buf: &mut impl BufMut) {
self.0.write(buf);
}
}
impl EncodeSize for TestDirectory {
fn encode_size(&self) -> usize {
self.0.encode_size()
}
}
impl Read for TestDirectory {
type Cfg = RangeCfg<usize>;
fn read_cfg(buf: &mut impl Buf, cfg: &Self::Cfg) -> Result<Self, CodecError> {
Option::<Addresses<PublicKey>>::read_cfg(buf, cfg).map(Self)
}
}
impl Directory<PublicKey> for TestDirectory {
fn codec_config(peers: &Set<PublicKey>) -> Self::Cfg {
RangeCfg::exact(peers.len())
}
fn matches(&self, peers: &Set<PublicKey>) -> bool {
self.0
.as_ref()
.is_none_or(|addresses| addresses.matches(peers))
}
}
#[derive(Clone, Debug)]
enum TestManager {
Discovery(DiscoveryManager),
Lookup(LookupManager),
}
impl Provider for TestManager {
type PublicKey = PublicKey;
async fn peer_set(&mut self, id: u64) -> Option<TrackedPeers<Self::PublicKey>> {
match self {
Self::Discovery(manager) => manager.peer_set(id).await,
Self::Lookup(manager) => manager.peer_set(id).await,
}
}
async fn subscribe(&mut self) -> commonware_p2p::PeerSetSubscription<Self::PublicKey> {
match self {
Self::Discovery(manager) => manager.subscribe().await,
Self::Lookup(manager) => manager.subscribe().await,
}
}
}
impl DkgManager for TestManager {
type Directory = TestDirectory;
type Error = MissingAddress<PublicKey>;
fn track(
&mut self,
epoch: Epoch,
peers: TrackedPeers<Self::PublicKey>,
directory: &Self::Directory,
) -> Result<(), Self::Error> {
match (self, &directory.0) {
(Self::Discovery(manager), None) => {
DkgManager::track(manager, epoch, peers, &Unit).map_err(|error| match error {})
}
(Self::Lookup(manager), Some(addresses)) => {
DkgManager::track(manager, epoch, peers, addresses)
}
_ => panic!("network and directory must use the same transport"),
}
}
}
#[derive(Clone, PartialEq, Eq)]
pub(super) struct Block {
context: Context<sha256::Digest, ed25519::PublicKey>,
parent: sha256::Digest,
height: Height,
state_root: sha256::Digest,
range: NonEmptyRange<Location>,
payload: Option<Payload<MinPk, ed25519::PrivateKey, TestDirectory>>,
}
impl Write for Block {
fn write(&self, buf: &mut impl BufMut) {
self.context.write(buf);
self.parent.write(buf);
self.height.write(buf);
self.state_root.write(buf);
self.range.write(buf);
self.payload.write(buf);
}
}
impl EncodeSize for Block {
fn encode_size(&self) -> usize {
self.context.encode_size()
+ self.parent.encode_size()
+ self.height.encode_size()
+ self.state_root.encode_size()
+ self.range.encode_size()
+ self.payload.encode_size()
}
}
impl Read for Block {
type Cfg = ();
fn read_cfg(buf: &mut impl Buf, _: &Self::Cfg) -> Result<Self, CodecError> {
Ok(Self {
context: Context::read(buf)?,
parent: sha256::Digest::read(buf)?,
height: Height::read(buf)?,
state_root: sha256::Digest::read(buf)?,
range: NonEmptyRange::read(buf)?,
payload: Option::<Payload<MinPk, ed25519::PrivateKey, TestDirectory>>::read_cfg(
buf,
&(MAX_PARTICIPANTS, max_supported_mode()),
)?,
})
}
}
impl Digestible for Block {
type Digest = sha256::Digest;
fn digest(&self) -> sha256::Digest {
Sha256::hash(&[&self.encode()])
}
}
impl Heightable for Block {
fn height(&self) -> Height {
self.height
}
}
impl ConsensusBlock for Block {
fn parent(&self) -> sha256::Digest {
self.parent
}
}
impl CertifiableBlock for Block {
type Context = Context<sha256::Digest, ed25519::PublicKey>;
fn context(&self) -> Self::Context {
self.context.clone()
}
}
impl ReshareBlock for Block {
type Variant = MinPk;
type Signer = ed25519::PrivateKey;
type Directory = TestDirectory;
fn payload(&self) -> Option<Payload<Self::Variant, Self::Signer, Self::Directory>> {
self.payload.clone()
}
}
impl Block {
fn genesis(
leader: ed25519::PublicKey,
info: EpochInfo<MinPk, ed25519::PublicKey, TestDirectory>,
) -> Self {
Self {
context: Context {
round: Round::new(Epoch::zero(), View::zero()),
leader,
parent: (View::zero(), sha256::Digest::EMPTY),
},
parent: sha256::Digest::EMPTY,
height: Height::zero(),
state_root: empty_db_root(),
range: non_empty_range!(Location::new(0), Location::new(1)),
payload: Some(Payload::EpochInfo(info)),
}
}
}
#[derive(Clone)]
struct App {
genesis: Block,
processed: Arc<Mutex<BTreeMap<ed25519::PublicKey, u64>>>,
public_key: ed25519::PublicKey,
hold: Arc<Mutex<Option<(ed25519::PublicKey, u64)>>>,
}
impl App {
async fn execute<E: Rng + Spawner + Metrics + Clock + Storage + BufferPooler>(
height: Height,
mut batches: <Database<E> as DatabaseSet<E>>::Unmerkleized,
) -> <Database<E> as DatabaseSet<E>>::Merkleized {
let key = Sha256::hash(&[b"height"]);
batches = batches.write(key, Some(u64_to_digest(height.get())));
batches.merkleize().await.unwrap()
}
}
impl<E: Rng + Spawner + Metrics + Clock + Storage + BufferPooler> Application<E> for App {
type SigningScheme = Scheme;
type Context = Context<sha256::Digest, ed25519::PublicKey>;
type Block = Block;
type Databases = Database<E>;
type Captured = ();
type Provider = ();
type Input = ReshareInput<(), MinPk, ed25519::PrivateKey, TestDirectory>;
async fn genesis(&mut self) -> Self::Block {
self.genesis.clone()
}
async fn propose(
&mut self,
context: (E, Self::Context),
ancestry: impl Ancestry<Self::Block>,
batches: <Self::Databases as DatabaseSet<E>>::Unmerkleized,
input: Input<Self::Input, Self::Provider>,
) -> Option<Proposed<Self, E>> {
let parent = ancestry.peek()?.clone();
let height = Height::new(parent.height().get() + 1);
let payload = input.upstream.payload;
let merkleized = Self::execute(height, batches).await;
let bounds = merkleized.bounds();
let block = Block {
context: context.1,
parent: parent.digest(),
height,
state_root: merkleized.root(),
range: non_empty_range!(bounds.inactivity_floor, bounds.tip.size),
payload,
};
Some(Proposed { block, merkleized })
}
async fn verify(
&mut self,
_context: (E, Self::Context),
ancestry: impl Ancestry<Self::Block>,
batches: <Self::Databases as DatabaseSet<E>>::Unmerkleized,
) -> Option<<Self::Databases as DatabaseSet<E>>::Merkleized> {
let tip = ancestry.peek()?.clone();
let merkleized = Self::execute(tip.height(), batches).await;
Some(merkleized)
}
async fn apply(
&mut self,
_context: (E, Self::Context),
block: &Self::Block,
batches: <Self::Databases as DatabaseSet<E>>::Unmerkleized,
) -> Option<<Self::Databases as DatabaseSet<E>>::Merkleized> {
Some(Self::execute(block.height(), batches).await)
}
async fn capture(
&mut self,
_context: (E, Self::Context),
_block: &Self::Block,
_batches: &<Self::Databases as DatabaseSet<E>>::Merkleized,
_readers: <Self::Databases as DatabaseSet<E>>::Readers,
) {
}
async fn finalized(
&mut self,
context: (E, Self::Context),
block: &Self::Block,
_captured: Self::Captured,
_readers: <Self::Databases as DatabaseSet<E>>::Readers,
) {
self.processed
.lock()
.insert(self.public_key.clone(), block.height().get());
while self.hold.lock().as_ref().is_some_and(|(held, height)| {
*held == self.public_key && block.height().get() >= *height
}) {
context.0.sleep(Duration::from_millis(25)).await;
}
}
fn sync_targets(block: &Self::Block) -> <Self::Databases as DatabaseSet<E>>::SyncTargets {
Target::new(block.state_root, block.range.clone())
}
}
#[derive(Clone)]
struct DynamicProvider {
schemes: Arc<Mutex<HashMap<Epoch, Arc<Scheme>>>>,
}
impl DynamicProvider {
pub(super) fn new() -> Self {
Self {
schemes: Arc::new(Mutex::new(HashMap::new())),
}
}
fn register(&self, epoch: Epoch, scheme: Scheme) {
self.schemes.lock().insert(epoch, Arc::new(scheme));
}
}
impl CertificateProvider for DynamicProvider {
type Scope = Epoch;
type Scheme = Scheme;
fn scoped(&self, scope: Self::Scope) -> Option<Scoped<Self::Scheme>> {
self.schemes.lock().get(&scope).cloned().map(Scoped::scheme)
}
fn scheme(&self, scope: Self::Scope) -> Option<Arc<Self::Scheme>> {
self.schemes.lock().get(&scope).cloned()
}
}
#[derive(Clone)]
struct TestRegistrar {
provider: DynamicProvider,
events: Arc<Mutex<BTreeMap<ed25519::PublicKey, Vec<Registration>>>>,
public_key: ed25519::PublicKey,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum RegistrationRole {
Signer,
Verifier,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) struct Registration {
pub(super) epoch: Epoch,
pub(super) role: RegistrationRole,
}
impl Registrar for TestRegistrar {
type Variant = MinPk;
type PublicKey = ed25519::PublicKey;
async fn register(&self, epoch: Epoch, info: SchemeInfo<Self::Variant, Self::PublicKey>) {
let (scheme, role) = match info {
SchemeInfo::Verifier {
participants,
sharing,
} => (
Scheme::verifier(NAMESPACE, participants, sharing),
RegistrationRole::Verifier,
),
SchemeInfo::Signer {
participants,
sharing,
share,
} => (
Scheme::signer(NAMESPACE, participants, sharing, share)
.expect("share must match participant set"),
RegistrationRole::Signer,
),
};
self.provider.register(epoch, scheme);
self.events
.lock()
.entry(self.public_key.clone())
.or_default()
.push(Registration { epoch, role });
}
}
#[derive(Clone)]
struct ScheduleProvider {
pub(super) schedule: Arc<CommitteeSchedule>,
network: Network,
addresses: Arc<Map<PublicKey, Address>>,
}
impl ParticipantsProvider for ScheduleProvider {
type PublicKey = PublicKey;
type Directory = TestDirectory;
async fn participants(&mut self, epoch: Epoch) -> Set<Self::PublicKey> {
self.schedule.players(epoch)
}
async fn directory(&mut self, _: Epoch, peers: Set<Self::PublicKey>) -> Self::Directory {
self.network.directory(&peers, &self.addresses)
}
}
#[derive(Clone)]
pub(super) struct CommitteeSchedule {
participants: Vec<ed25519::PublicKey>,
committee_sizes: Vec<usize>,
}
impl CommitteeSchedule {
pub(super) fn players(&self, epoch: Epoch) -> Set<ed25519::PublicKey> {
let offset = epoch.get() as usize % self.participants.len();
let committee_size =
self.committee_sizes[epoch.get() as usize % self.committee_sizes.len()];
let players = (0..committee_size)
.map(|i| self.participants[(offset + i) % self.participants.len()].clone());
Set::from_iter_dedup(players)
}
}
#[derive(Clone)]
pub(super) struct ReshareEngine {
network: Network,
signers: Vec<ed25519::PrivateKey>,
pub(super) participants: Vec<ed25519::PublicKey>,
addresses: Arc<Map<PublicKey, Address>>,
pub(super) schedule: Arc<CommitteeSchedule>,
initial: Arc<InitialState>,
sharing_mode: Mode,
stores: Arc<Mutex<BTreeMap<ed25519::PublicKey, MemorySecretStore>>>,
pub(super) registrations: Arc<Mutex<BTreeMap<ed25519::PublicKey, Vec<Registration>>>>,
pub(super) state_syncs: Arc<Mutex<BTreeMap<ed25519::PublicKey, u64>>>,
pub(super) state_sync_starts: Arc<Mutex<BTreeMap<ed25519::PublicKey, u64>>>,
state_sync_floor: Option<Height>,
processed_hold: Arc<Mutex<Option<(ed25519::PublicKey, u64)>>>,
epoch_cross_during_sync: bool,
processed: Arc<Mutex<BTreeMap<ed25519::PublicKey, u64>>>,
marshals: Arc<Mutex<BTreeMap<ed25519::PublicKey, Marshal>>>,
failures: Arc<HashSet<u64>>,
}
pub(super) struct ValidatorEngine {
context: DeterministicContext,
handles: [Handle<()>; 6],
}
#[derive(Clone)]
struct InitialState {
info: EpochInfo<MinPk, ed25519::PublicKey, TestDirectory>,
shares: Map<ed25519::PublicKey, Share>,
}
impl InitialState {
async fn register_epoch_zero(
&self,
provider: &DynamicProvider,
store: &MemorySecretStore,
) -> RegistrationRole {
let mut store = store.clone();
let participants = self.info.output.players();
let sharing = self.info.output.public();
store.get_share(Epoch::zero()).await.map_or_else(
|| {
provider.register(
Epoch::zero(),
Scheme::verifier(NAMESPACE, participants.clone(), sharing.clone()),
);
RegistrationRole::Verifier
},
|share| {
provider.register(
Epoch::zero(),
Scheme::signer(NAMESPACE, participants.clone(), sharing.clone(), share)
.expect("initial signer share"),
);
RegistrationRole::Signer
},
)
}
}
impl ReshareEngine {
pub(super) fn new(network: Network) -> Self {
Self::with_committee(network, 5, 4)
}
pub(super) fn with_committee(network: Network, total: u32, committee_size: usize) -> Self {
Self::with_committees(network, total, vec![committee_size])
}
pub(super) fn with_committees(
network: Network,
total: u32,
committee_sizes: Vec<usize>,
) -> Self {
assert!(!committee_sizes.is_empty());
for committee_size in &committee_sizes {
assert!(*committee_size > 0);
assert!(*committee_size <= total as usize);
}
let mut rng = test_rng();
let signers = (0..total)
.map(|_| ed25519::PrivateKey::random(&mut rng))
.collect::<Vec<_>>();
let participants = signers.iter().map(|s| s.public_key()).collect::<Vec<_>>();
let addresses = Arc::new(Map::from_iter_dedup(participants.iter().enumerate().map(
|(index, participant)| {
let socket =
SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 10_000 + index as u16);
(participant.clone(), Address::Symmetric(socket))
},
)));
let schedule = Arc::new(CommitteeSchedule {
participants: participants.clone(),
committee_sizes,
});
let players = schedule.players(Epoch::zero());
let next_players = schedule.players(Epoch::new(1));
let (output, shares) =
deal::<MinPk, _, N3f1>(TestRng::new(10), Mode::NonZeroCounter, players.clone())
.expect("trusted initial deal");
let initial_peers = Set::from_iter_dedup(
output
.players()
.iter()
.chain(players.iter())
.chain(next_players.iter())
.cloned(),
);
let info = EpochInfo {
outcome: EpochOutcome::Success,
epoch: Epoch::zero(),
output,
players,
next_players,
directory: network.directory(&initial_peers, &addresses),
};
Self {
network,
signers,
participants,
addresses,
schedule,
initial: Arc::new(InitialState { info, shares }),
sharing_mode: Mode::NonZeroCounter,
stores: Arc::new(Mutex::new(BTreeMap::new())),
registrations: Arc::new(Mutex::new(BTreeMap::new())),
state_syncs: Arc::new(Mutex::new(BTreeMap::new())),
state_sync_starts: Arc::new(Mutex::new(BTreeMap::new())),
state_sync_floor: None,
processed_hold: Arc::new(Mutex::new(None)),
epoch_cross_during_sync: false,
processed: Arc::new(Mutex::new(BTreeMap::new())),
marshals: Arc::new(Mutex::new(BTreeMap::new())),
failures: Arc::new(HashSet::new()),
}
}
pub(super) fn with_failures(mut self, failures: impl IntoIterator<Item = u64>) -> Self {
self.failures = Arc::new(failures.into_iter().collect());
self
}
pub(super) const fn with_sharing_mode(mut self, sharing_mode: Mode) -> Self {
self.sharing_mode = sharing_mode;
self
}
pub(super) const fn with_state_sync_floor(mut self, height: Height) -> Self {
self.state_sync_floor = Some(height);
self
}
pub(super) fn with_processed_hold(
self,
participant: ed25519::PublicKey,
height: Height,
) -> Self {
*self.processed_hold.lock() = Some((participant, height.get()));
self
}
pub(super) const fn with_epoch_cross_during_sync(mut self) -> Self {
self.epoch_cross_during_sync = true;
self
}
pub(super) const fn state_sync_floor(&self) -> Option<Height> {
self.state_sync_floor
}
fn store(&self, public_key: &ed25519::PublicKey) -> MemorySecretStore {
let mut stores = self.stores.lock();
match stores.entry(public_key.clone()) {
Entry::Occupied(entry) => entry.get().clone(),
Entry::Vacant(entry) => {
let store = MemorySecretStore::default();
if let Some(share) = self.initial.shares.get_value(public_key).cloned() {
store.seed_share(Epoch::zero(), share);
}
entry.insert(store.clone());
store
}
}
}
}
impl EngineDefinition for ReshareEngine {
type PublicKey = ed25519::PublicKey;
type Engine = ValidatorEngine;
type State = ValidatorState;
fn participants(&self) -> Vec<Self::PublicKey> {
self.participants.clone()
}
fn channels(&self) -> Vec<(u64, Quota)> {
vec![
(VOTE_CHANNEL, TEST_QUOTA),
(CERTIFICATE_CHANNEL, TEST_QUOTA),
(RESOLVER_CHANNEL, TEST_QUOTA),
(BACKFILL_CHANNEL, TEST_QUOTA),
(BROADCAST_CHANNEL, TEST_QUOTA),
(QMDB_CHANNEL, TEST_QUOTA),
(DKG_CHANNEL, TEST_QUOTA),
(DKG_PROBE_CHANNEL, TEST_QUOTA),
]
}
async fn init(&self, ctx: InitContext<'_, Self::PublicKey>) -> (Self::Engine, Self::State) {
let InitContext {
context,
index,
delayed,
public_key,
oracle,
channels,
participants: _,
monitor,
} = ctx;
let signer = self.signers[index].clone();
let partition_prefix = format!("reshare-e2e-{index}");
let page_cache = CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE);
{
let mut hold = self.processed_hold.lock();
if hold.as_ref().is_some_and(|(held, _)| held == public_key)
&& self.stores.lock().contains_key(public_key)
{
hold.take();
}
}
let mut channels = channels.into_iter();
let vote_network = channels.next().unwrap();
let certificate_network = channels.next().unwrap();
let resolver_network = channels.next().unwrap();
let backfill_network = channels.next().unwrap();
let broadcast_network = channels.next().unwrap();
let qmdb_network = channels.next().unwrap();
let dkg_network = channels.next().unwrap();
let probe_boundary_network = channels.next().unwrap();
let provider = DynamicProvider::new();
let store = self.store(public_key);
self.initial.register_epoch_zero(&provider, &store).await;
let dkg_manager = self.network.manager(oracle);
let resolver = marshal_resolver::init(
context.child("marshal_resolver"),
marshal_resolver::Config {
public_key: public_key.clone(),
peer_provider: oracle.manager(),
blocker: oracle.control(public_key.clone()),
mailbox_size: NZUsize!(100),
timeout: Duration::from_secs(2),
fetch_retry_timeout: Duration::from_millis(100),
priority_requests: false,
priority_responses: false,
},
backfill_network,
);
let broadcast_config = buffered::Config {
public_key: public_key.clone(),
mailbox_size: NZUsize!(100),
deque_size: 10,
priority: false,
codec_config: (),
peer_provider: oracle.manager(),
};
let (broadcast_engine, buffer) =
buffered::Engine::new(context.child("broadcast"), broadcast_config);
broadcast_engine.start(broadcast_network);
let finalizations_by_height = prunable::Archive::init(
context.child("finalizations_by_height"),
archive_config(&partition_prefix, "finalizations", page_cache.clone(), ()),
)
.await
.expect("finalizations archive");
let finalized_blocks = prunable::Archive::init(
context.child("finalized_blocks"),
archive_config(&partition_prefix, "blocks", page_cache.clone(), ()),
)
.await
.expect("blocks archive");
let genesis = Block::genesis(self.participants[0].clone(), self.initial.info.clone());
let (probe_actor, probe_mailbox) = dkg_probe::Actor::new(dkg_probe::Config {
context: context.child("dkg_probe"),
manager: dkg_manager.clone(),
bootstrap: dkg_probe::Bootstrap {
epoch: Epoch::zero(),
participants: self.initial.info.participants(),
directory: self.initial.info.directory.clone(),
},
verifier: Scheme::certificate_verifier(
NAMESPACE,
*self.initial.info.output.public().public(),
),
genesis: self.initial.info.clone(),
strategy: Sequential,
blocker: oracle.control(public_key.clone()),
blocks_per_epoch: EPOCH_LENGTH,
retry_timeout: NZDuration!(Duration::from_millis(500)),
mailbox_size: NZUsize!(100),
block_codec_config: (),
});
let probe_handle = probe_actor.start(probe_boundary_network);
let stateful_startup_context = context.child("stateful_startup");
let mut plan = SyncPlan::init(&stateful_startup_context, partition_prefix.clone()).await;
let should_state_sync = plan.should_state_sync(delayed);
if should_state_sync {
*self
.state_sync_starts
.lock()
.entry(public_key.clone())
.or_default() += 1;
}
let probe_artifact = if should_state_sync {
let artifact = probe_mailbox.subscribe().await.expect("probe stopped");
provider.register(
artifact.info.epoch,
Scheme::verifier(
NAMESPACE,
artifact.info.output.players().clone(),
artifact.info.output.public().clone(),
),
);
Some(artifact)
} else {
None
};
if self.epoch_cross_during_sync && should_state_sync {
let artifact = probe_artifact
.as_ref()
.expect("epoch cross hold requires state-sync bootstrap");
let committee = artifact.info.output.players().clone();
assert!(
committee.position(public_key).is_none(),
"delayed node must not be a sampled-committee member for the epoch-cross hold"
);
let target = FixedEpocher::new(EPOCH_LENGTH)
.first(artifact.info.epoch.next())
.expect("next epoch must be supported")
.get();
loop {
let crossed = {
let processed = self.processed.lock();
committee
.iter()
.all(|member| processed.get(member).is_some_and(|height| *height > target))
};
if crossed {
break;
}
context.sleep(Duration::from_millis(50)).await;
}
}
if should_state_sync {
let finalization = match self.state_sync_floor {
Some(height) => {
let source = self
.marshals
.lock()
.values()
.next()
.cloned()
.expect("state-sync floor source must be available");
source
.get_finalization(height)
.await
.expect("configured state-sync floor must be finalized")
}
None => probe_artifact
.as_ref()
.expect("state-sync startup must have a probe artifact")
.floor
.clone(),
};
plan = plan.with_floor(finalization);
}
let (marshal_actor, marshal, floor) = MarshalActor::init(
context.child("marshal"),
finalizations_by_height,
finalized_blocks,
marshal::Config {
provider: ConstantProvider::<_, Epoch>::new(Scheme::certificate_verifier(
NAMESPACE,
*self.initial.info.output.public().public(),
)),
epocher: FixedEpocher::new(EPOCH_LENGTH),
start: plan.marshal_start(genesis.clone()),
partition_prefix: partition_prefix.clone(),
mailbox_size: NZUsize!(100),
view_retention: ViewDelta::new(10),
prunable_items_per_section: NZU64!(10),
page_cache: page_cache.clone(),
replay_buffer: IO_BUFFER_SIZE,
key_write_buffer: IO_BUFFER_SIZE,
value_write_buffer: IO_BUFFER_SIZE,
block_codec_config: (),
max_repair: NZUsize!(10),
max_pending_acks: NZUsize!(1),
strategy: Sequential,
},
)
.await;
self.marshals
.lock()
.insert(public_key.clone(), marshal.clone());
let db_config = FixedConfig {
merkle_config: MmrJournalConfig {
journal_partition: format!("{partition_prefix}-qmdb-mmr-journal"),
metadata_partition: format!("{partition_prefix}-qmdb-mmr-metadata"),
items_per_blob: NZU64!(11),
write_buffer: IO_BUFFER_SIZE,
replay_buffer: IO_BUFFER_SIZE,
strategy: Sequential,
page_cache: page_cache.clone(),
},
journal_config: FixedLogConfig {
partition: format!("{partition_prefix}-qmdb-log-journal"),
items_per_blob: NZU64!(7),
page_cache: page_cache.clone(),
write_buffer: IO_BUFFER_SIZE,
replay_buffer: IO_BUFFER_SIZE,
},
translator: TwoCap,
init_cache_size: Some(NZUsize!(1024)),
init_buffer: NZUsize!(1 << 21),
init_concurrency: (),
};
let (qmdb_resolver_actor, qmdb_sync_resolver) = qmdb_resolver::Actor::new(
context.child("qmdb_resolver"),
qmdb_resolver::Config {
peer_provider: oracle.manager(),
blocker: oracle.control(public_key.clone()),
database: None,
mailbox_size: NZUsize!(100),
me: Some(public_key.clone()),
timeout: Duration::from_secs(2),
fetch_retry_timeout: Duration::from_millis(100),
max_serve_ops: NZU64!(16),
priority_requests: false,
priority_responses: false,
},
);
let qmdb_handle = qmdb_resolver_actor.start(qmdb_network);
let fence_epoch = probe_artifact
.as_ref()
.map_or_else(Epoch::zero, |artifact| artifact.info.epoch);
let state_sync = probe_artifact.map(|artifact| {
let floor = plan
.floor()
.cloned()
.expect("state-sync startup must have a probe floor");
StateSync {
info: artifact.info,
floor,
}
});
let registrar = TestRegistrar {
provider: provider.clone(),
events: self.registrations.clone(),
public_key: public_key.clone(),
};
let (fence, gate) = Fence::new(fence_epoch);
let sync_floor: Option<Finalization<Scheme, sha256::Digest>> = state_sync
.as_ref()
.map(|state_sync| state_sync.floor.clone());
let state_sync = StateSyncPlan::init(
context.child("dkg_state_sync_plan"),
StateSyncConfig {
partition_prefix: partition_prefix.clone(),
max_participants: MAX_PARTICIPANTS,
max_supported_mode: max_supported_mode(),
},
state_sync,
)
.await;
let (reshare_actor, reshare_mailbox) = reshare::Actor::new(
context.child("reshare"),
reshare::Config {
signer: signer.clone(),
manager: dkg_manager.clone(),
blocker: oracle.control(public_key.clone()),
participants_provider: ScheduleProvider {
schedule: self.schedule.clone(),
network: self.network,
addresses: self.addresses.clone(),
},
secret_store: store,
strategy: Sequential,
registrar,
marshal: marshal.clone(),
state_sync: state_sync.clone(),
fence,
namespace: NAMESPACE,
sharing_mode: self.sharing_mode,
reveal: Reveal::V1,
mailbox_size: NZUsize!(100),
partition_prefix: format!("{partition_prefix}-reshare"),
max_participants: MAX_PARTICIPANTS,
blocks_per_epoch: EPOCH_LENGTH,
batch_verifier: PhantomData::<ed25519::Batch>,
},
);
let dkg_network = (
dkg_network.0,
FilteredReceiver::epochs(dkg_network.1, self.failures.clone()),
);
let reshare_handle = reshare_actor.start(dkg_network);
let (stateful_actor, stateful_mailbox) = StatefulActor::init(
context.child("stateful"),
StatefulConfig {
application: App {
genesis: genesis.clone(),
processed: self.processed.clone(),
public_key: public_key.clone(),
hold: self.processed_hold.clone(),
},
db_config,
provider: (),
marshal: (marshal.clone(), floor),
mailbox_size: NZUsize!(100),
plan,
resolvers: qmdb_sync_resolver,
sync_config: SyncEngineConfig {
fetch_batch_size: NZU64!(16),
apply_batch_size: NZU64!(64),
max_outstanding_requests: 8,
update_channel_size: NZUsize!(256),
max_retained_roots: 8,
},
prune_config: None,
},
);
let deferred = Deferred::new(
context.child("deferred"),
reshare::Application::new(
stateful_mailbox.clone(),
reshare_mailbox.clone(),
EPOCH_LENGTH,
),
marshal.clone(),
FixedEpocher::new(EPOCH_LENGTH),
);
let (orchestrator_actor, orchestrator_mailbox) = orchestrator::Actor::new(
context.child("orchestrator"),
orchestrator::Config {
oracle: oracle.control(public_key.clone()),
manager: dkg_manager,
provider: provider.clone(),
marshal: marshal.clone(),
application: deferred,
strategy: Sequential,
simplex: orchestrator::SimplexConfig {
elector: RoundRobin::<Sha256>::default(),
mailbox_size: NZUsize!(3),
replay_buffer: IO_BUFFER_SIZE,
write_buffer: IO_BUFFER_SIZE,
page_cache_page_size: PAGE_SIZE,
page_cache_pages: PAGE_CACHE_SIZE,
leader_timeout: Duration::from_secs(1),
certification_timeout: Duration::from_secs(2),
timeout_retry: Duration::from_millis(500),
fetch_timeout: Duration::from_secs(2),
view_retention: ViewDelta::new(10),
skip: SkipPolicy::Enabled {
timeout: Duration::from_secs(5),
budget: simplex::SkipBudget::Participants,
},
forward: ForwardPolicy::Disabled,
track_historical_votes: false,
},
gate,
state_sync,
blocks_per_epoch: EPOCH_LENGTH,
muxer_size: 128,
mailbox_size: NZUsize!(100),
partition_prefix: format!("{partition_prefix}-orchestrator"),
},
);
let orchestrator_handle =
orchestrator_actor.start(vote_network, certificate_network, resolver_network);
let reporters = Reporters::from((
stateful_mailbox,
Reporters::from((orchestrator_mailbox, reshare_mailbox)),
));
let marshal_handle = marshal_actor.start(
MonitorReporter::new(public_key.clone(), monitor, reporters),
buffer,
resolver,
);
probe_mailbox.attach(marshal.clone());
if let Some(finalization) = sync_floor {
let marshal = marshal.clone();
let state_syncs = self.state_syncs.clone();
let public_key = public_key.clone();
context
.child("sync_floor_recorder")
.spawn(move |_| async move {
let Ok(block) = marshal
.subscribe_by_commitment(
finalization.proposal.payload,
CommitmentFallback::Wait,
)
.await
else {
return;
};
state_syncs.lock().insert(public_key, block.height().get());
});
}
let stateful_handle = stateful_actor.start();
(
ValidatorEngine {
context,
handles: [
probe_handle,
qmdb_handle,
reshare_handle,
orchestrator_handle,
marshal_handle,
stateful_handle,
],
},
ValidatorState {
marshal,
processed: self.processed.clone(),
registrations: self.registrations.clone(),
state_syncs: self.state_syncs.clone(),
public_key: public_key.clone(),
},
)
}
fn start(engine: Self::Engine) -> Handle<()> {
let ValidatorEngine { context, handles } = engine;
context.spawn(move |_| async move {
Handle::select(handles)
.await
.expect("validator actor failed");
})
}
}
#[derive(Clone)]
pub(super) struct ValidatorState {
pub(super) marshal: Marshal,
processed: Arc<Mutex<BTreeMap<ed25519::PublicKey, u64>>>,
registrations: Arc<Mutex<BTreeMap<ed25519::PublicKey, Vec<Registration>>>>,
pub(super) state_syncs: Arc<Mutex<BTreeMap<ed25519::PublicKey, u64>>>,
public_key: ed25519::PublicKey,
}
impl PartialEq for ValidatorState {
fn eq(&self, other: &Self) -> bool {
self.public_key == other.public_key
}
}
impl ProcessedHeight for ValidatorState {
async fn processed_height(&self) -> u64 {
self.processed
.lock()
.get(&self.public_key)
.copied()
.unwrap_or_default()
}
}
impl ValidatorState {
pub(super) fn public_key(&self) -> &ed25519::PublicKey {
&self.public_key
}
pub(super) fn registrations(&self) -> Vec<Registration> {
self.registrations
.lock()
.get(&self.public_key)
.cloned()
.unwrap_or_default()
}
pub(super) fn state_sync_height(&self) -> Option<u64> {
self.state_syncs.lock().get(&self.public_key).copied()
}
}
#[test]
fn restart_after_epoch_zero_pruning_does_not_reseed_share() {
let engine = ReshareEngine::new(Network::Discovery);
let public_key = engine
.initial
.shares
.get(0)
.cloned()
.expect("epoch-zero player");
let store = engine.store(&public_key);
assert!(store.has_share(Epoch::zero()));
futures::executor::block_on(async {
let mut pruned = store.clone();
pruned.prune(Epoch::new(1)).await;
assert!(pruned.get_share(Epoch::zero()).await.is_none());
let restarted = engine.store(&public_key);
let mut restarted_view = restarted.clone();
assert!(restarted_view.get_share(Epoch::zero()).await.is_none());
let provider = DynamicProvider::new();
let role = engine
.initial
.register_epoch_zero(&provider, &restarted)
.await;
assert_eq!(role, RegistrationRole::Verifier);
});
}
fn archive_config<C>(
prefix: &str,
name: &str,
page_cache: CacheRef,
codec_config: C,
) -> prunable::Config<TwoCap, C> {
prunable::Config {
translator: TwoCap,
metadata_partition: format!("{prefix}-{name}-metadata"),
key_partition: format!("{prefix}-{name}-key"),
key_page_cache: page_cache,
value_partition: format!("{prefix}-{name}-value"),
compression: None,
codec_config,
items_per_section: NZU64!(10),
key_write_buffer: IO_BUFFER_SIZE,
value_write_buffer: IO_BUFFER_SIZE,
replay_buffer: IO_BUFFER_SIZE,
}
}
fn empty_db_root() -> sha256::Digest {
Sha256Digest::from(hex!(
"ea6e0567a525372add5e4ef4d0600c18ed47fa5dd041a0ab0d25b60ea8c35978"
))
}
fn u64_to_digest(v: u64) -> sha256::Digest {
let mut bytes = [0u8; 32];
bytes[..8].copy_from_slice(&v.to_be_bytes());
sha256::Digest::from(bytes)
}
pub(super) fn final_height(epoch: u64) -> Height {
FixedEpocher::new(EPOCH_LENGTH)
.last(Epoch::new(epoch))
.expect("test epoch should be supported")
}
pub(super) fn height_round(height: Height) -> Round {
let info = FixedEpocher::new(EPOCH_LENGTH)
.containing(height)
.expect("test height should be supported");
Round::new(info.epoch(), View::new(info.relative().get()))
}