feat(blend): support tx blending (#3321)

This commit is contained in:
Antonio
2026-08-19 05:14:57 +00:00
committed by GitHub
parent e595abf72c
commit 14c114a525
43 changed files with 1940 additions and 418 deletions
Generated
+2
View File
@@ -4135,8 +4135,10 @@ dependencies = [
"logos-blockchain-poq",
"logos-blockchain-sdp-service",
"logos-blockchain-services-utils",
"logos-blockchain-storage-service",
"logos-blockchain-time-service",
"logos-blockchain-tracing",
"logos-blockchain-tx-service",
"logos-blockchain-utils",
"overwatch",
"rand 0.8.6",
+1 -1
View File
@@ -9,4 +9,4 @@ mod message;
pub use encap::encapsulated::MessageIdentifier;
pub use error::Error;
pub use message::payload::{PaddedPayloadBody, PayloadType};
pub use message::payload::{MAX_PAYLOAD_BODY_SIZE, PaddedPayloadBody, PayloadType};
@@ -2,7 +2,7 @@ use core::hash::Hash;
use std::num::NonZeroU64;
use lb_blend_message::{
Error, PaddedPayloadBody, PayloadType, crypto::proofs::PoQVerificationInputsMinusSigningKey,
Error, PaddedPayloadBody, crypto::proofs::PoQVerificationInputsMinusSigningKey,
input::EncapsulationInput,
};
use lb_cryptarchia_engine::Epoch;
@@ -11,10 +11,28 @@ use crate::{
membership::Membership,
message_blend::{
crypto::EncapsulatedMessageWithVerifiedPublicHeader,
provers::{ProofsGeneratorSettings, WinningPolInfoStream, leader::LeaderProofsGenerator},
provers::{
BlendLayerProof, ProofsGeneratorSettings, WinningPolInfoStream,
leader_and_pow::LeaderAndPowProofsGenerator,
},
},
};
#[derive(Debug, Clone, Copy)]
enum PayloadType {
BlockProposal,
Transaction,
}
impl From<PayloadType> for lb_blend_message::PayloadType {
fn from(value: PayloadType) -> Self {
match value {
PayloadType::BlockProposal => Self::BlockProposal,
PayloadType::Transaction => Self::Transaction,
}
}
}
/// [`EpochCryptographicProcessor`] is responsible for only wrapping data
/// messages (no cover messages) for the message indistinguishability.
///
@@ -37,7 +55,7 @@ impl<NodeId, ProofsGenerator> EpochCryptographicProcessor<NodeId, ProofsGenerato
impl<NodeId, ProofsGenerator> EpochCryptographicProcessor<NodeId, ProofsGenerator>
where
ProofsGenerator: LeaderProofsGenerator,
ProofsGenerator: LeaderAndPowProofsGenerator,
{
#[must_use]
pub fn new(
@@ -66,18 +84,35 @@ where
impl<NodeId, ProofsGenerator> EpochCryptographicProcessor<NodeId, ProofsGenerator>
where
NodeId: Eq + Hash + 'static,
ProofsGenerator: LeaderProofsGenerator,
ProofsGenerator: LeaderAndPowProofsGenerator,
{
pub async fn encapsulate_block_proposal_payload(
&mut self,
payload: &[u8],
) -> Result<EncapsulatedMessageWithVerifiedPublicHeader, Error> {
self.encapsulate_payload(PayloadType::BlockProposal, payload)
.await
}
pub async fn encapsulate_transaction_payload(
&mut self,
payload: &[u8],
) -> Result<EncapsulatedMessageWithVerifiedPublicHeader, Error> {
self.encapsulate_payload(PayloadType::Transaction, payload)
.await
}
async fn encapsulate_payload(
&mut self,
payload_type: PayloadType,
payload: &[u8],
) -> Result<EncapsulatedMessageWithVerifiedPublicHeader, Error> {
// We validate the payload early on so we don't generate proofs unnecessarily.
let validated_payload = PaddedPayloadBody::try_from(payload)?;
let mut proofs = Vec::with_capacity(self.num_blend_layers.get() as usize);
for _ in 0..self.num_blend_layers.into() {
let Some(proof) = self.proofs_generator.get_next_proof().await else {
let Some(proof) = self.next_proof_for(payload_type).await else {
return Err(Error::ProofNotAvailable);
};
proofs.push(proof);
@@ -97,7 +132,7 @@ where
.enumerate()
.inspect(|(layer, (_, node_index))| {
tracing::trace!(
"Encapsulating layer {layer:?} of data message for node at index {node_index:?}."
"Encapsulating layer {layer:?} of data message type {payload_type:?} for node at index {node_index:?}."
);
})
// Map retrieved indices to the nodes' public keys.
@@ -126,10 +161,21 @@ where
Ok(EncapsulatedMessageWithVerifiedPublicHeader::try_new(
&inputs,
PayloadType::BlockProposal,
payload_type.into(),
validated_payload,
self.num_blend_layers.get() as usize,
)
.expect("Number of encapsulation inputs is in `1..=num_blend_layers`."))
}
/// The `PoQ` branch each payload type draws its layer proofs from.
///
/// An edge node has no core quota, so unlike a core node it has nothing to
/// spend on cover traffic — and it generates none.
async fn next_proof_for(&mut self, payload_type: PayloadType) -> Option<BlendLayerProof> {
match payload_type {
PayloadType::BlockProposal => self.proofs_generator.get_next_leader_proof().await,
PayloadType::Transaction => self.proofs_generator.get_next_pow_proof().await,
}
}
}
@@ -0,0 +1,76 @@
use async_trait::async_trait;
use lb_log_targets::blend;
use crate::message_blend::provers::{
BlendLayerProof, ProofsGeneratorSettings, WinningPolInfoStream,
leader::{LeaderProofsGenerator as _, RealLeaderProofsGenerator},
pow::{PowProofsGenerator as _, RealPowProofsGenerator},
};
#[cfg(test)]
mod tests;
const LOG_TARGET: &str = blend::scheduling::proofs::LEADER_AND_POW;
/// Proof generator for the two `PoQ` variants an edge node can reach.
///
/// An edge node holds no core quota, so it covers leadership — which needs
/// stake — and proof of work, which needs neither stake nor an SDP declaration
/// and is therefore what a node with no stake at all is left with. The variants
/// are indistinguishable to a verifier, so which one backs a given message is a
/// local decision.
#[async_trait]
pub trait LeaderAndPowProofsGenerator: Sized {
/// Instantiate a new generator for the duration of an epoch.
fn new(
settings: ProofsGeneratorSettings,
winning_pol_info_stream: WinningPolInfoStream,
) -> Self;
/// Request a new leadership proof from the prover. It returns `None` if all
/// the winning slots for the current epoch have been used up.
async fn get_next_leader_proof(&mut self) -> Option<BlendLayerProof>;
/// Request a new proof of work backed proof from the prover. It returns
/// `None` if the epoch's `PoW` public inputs admit no proof at all.
async fn get_next_pow_proof(&mut self) -> Option<BlendLayerProof>;
}
/// The generator an edge node runs for the duration of an epoch.
///
/// Unlike the core generator, this one needs no way to be told to stop: an edge
/// node replaces its whole message handler when an epoch rotates, and dropping
/// the generator with it is what abandons the mining stream it owns.
pub struct RealLeaderAndPowProofsGenerator {
leader_proofs_generator: RealLeaderProofsGenerator,
pow_proofs_generator: RealPowProofsGenerator,
}
#[async_trait]
impl LeaderAndPowProofsGenerator for RealLeaderAndPowProofsGenerator {
fn new(
settings: ProofsGeneratorSettings,
winning_pol_info_stream: WinningPolInfoStream,
) -> Self {
Self {
leader_proofs_generator: RealLeaderProofsGenerator::new(
settings,
winning_pol_info_stream,
),
pow_proofs_generator: RealPowProofsGenerator::new(settings),
}
}
async fn get_next_leader_proof(&mut self) -> Option<BlendLayerProof> {
self.leader_proofs_generator.get_next_proof().await
}
async fn get_next_pow_proof(&mut self) -> Option<BlendLayerProof> {
let proof = self.pow_proofs_generator.get_next_proof().await?;
tracing::trace!(
target: LOG_TARGET,
key_nullifier = ?proof.proof_of_quota.key_nullifier(),
signing_key = ?proof.ephemeral_signing_key.public_key(),
"generated PoW PoQ"
);
Some(proof)
}
}
@@ -0,0 +1,85 @@
use futures::stream;
use lb_blend_proofs::{quota::Quota, selection::inputs::VerifyInputs};
use lb_cryptarchia_engine::Epoch;
use test_log::test;
use crate::message_blend::provers::{
ProofsGeneratorSettings,
leader_and_pow::{LeaderAndPowProofsGenerator as _, RealLeaderAndPowProofsGenerator},
test_utils::{
poq_public_inputs_from_epoch_public_inputs_and_signing_key, valid_proof_of_leader_inputs,
valid_proof_of_work_inputs,
},
};
#[test(tokio::test)]
async fn pow_proof_generation() {
// The `PoW` fixture and the leadership fixture do not share public inputs,
// so the generator is built with the former and only its `PoW` branch is
// exercised here. Leadership generation is covered by the wrapped
// generator's own tests.
let public_inputs = valid_proof_of_work_inputs(Quota::ONE);
let mut generator = RealLeaderAndPowProofsGenerator::new(
ProofsGeneratorSettings {
local_node_index: None,
membership_size: 1,
public_inputs,
encapsulation_layers: 1.try_into().unwrap(),
epoch: Epoch::new(0),
},
Box::pin(stream::empty()),
);
let proof = generator.get_next_pow_proof().await.unwrap();
let verified_proof_of_quota = proof
.proof_of_quota
.into_inner()
.verify(&poq_public_inputs_from_epoch_public_inputs_and_signing_key(
(public_inputs, proof.ephemeral_signing_key.public_key()),
))
.unwrap();
proof
.proof_of_selection
.into_inner()
.verify(&VerifyInputs {
// Membership of 1 -> only a single index can be included
expected_node_index: 0,
key_nullifier: verified_proof_of_quota.key_nullifier(),
total_membership_size: 1,
})
.unwrap();
}
#[test(tokio::test)]
async fn leadership_proofs_are_delegated() {
let (public_inputs, leadership_private_inputs) = {
let (mut public_inputs, private_inputs) = valid_proof_of_leader_inputs(Quota::ONE);
// The wrapped `PoW` generator starts mining as soon as it is built, and
// the leadership fixture's difficulty is far too hard to ever solve. A
// zero quota switches the branch off, which is what this test wants
// anyway.
public_inputs.pow.pow_quota = Quota::ZERO;
(public_inputs, private_inputs)
};
let mut generator = RealLeaderAndPowProofsGenerator::new(
ProofsGeneratorSettings {
local_node_index: None,
membership_size: 1,
public_inputs,
encapsulation_layers: 1.try_into().unwrap(),
epoch: Epoch::new(0),
},
Box::pin(stream::repeat(leadership_private_inputs)),
);
let proof = generator.get_next_leader_proof().await.unwrap();
proof
.proof_of_quota
.into_inner()
.verify(&poq_public_inputs_from_epoch_public_inputs_and_signing_key(
(public_inputs, proof.ephemeral_signing_key.public_key()),
))
.unwrap();
}
@@ -12,6 +12,7 @@ pub mod core;
pub mod core_and_leader;
pub mod core_leader_and_pow;
pub mod leader;
pub mod leader_and_pow;
pub mod pow;
#[cfg(test)]
+2
View File
@@ -10,6 +10,8 @@ pub const TIME_INFO: &str = "/time/info";
pub const NETWORK_INFO: &str = "/network/info";
pub const BLEND_NETWORK_INFO: &str = "/blend/info";
pub const BLEND_JOIN_NETWORK: &str = "/blend/join";
pub const BLEND_DISPERSE_TRANSACTION: &str = "/blend/transactions/disperse";
pub const BLEND_PENDING_TRANSACTIONS: &str = "/blend/transactions/pending";
pub const MEMPOOL_ADD_TX: &str = "/mempool/add/tx";
pub const CHANNEL: &str = "/channel/:id";
pub const CHANNEL_DEPOSIT: &str = "/channel/deposit";
+13 -4
View File
@@ -44,10 +44,11 @@ use utoipa::OpenApi as _;
use utoipa_swagger_ui::SwaggerUi;
use super::handlers::{
add_tx, blend_info, block, block_events, blocks_range_stream, blocks_stream,
cryptarchia_headers, cryptarchia_info, cryptarchia_lib_stream, dial_peer, get_gas_prices,
get_sdp_declarations, get_sdp_snapshot, immutable_blocks, libp2p_info, mantle_metrics,
mantle_status, mempool_view, time_info, transaction, version, wallet,
add_tx, blend_info, blend_pending_transactions, blend_tx, block, block_events,
blocks_range_stream, blocks_stream, cryptarchia_headers, cryptarchia_info,
cryptarchia_lib_stream, dial_peer, get_gas_prices, get_sdp_declarations, get_sdp_snapshot,
immutable_blocks, libp2p_info, mantle_metrics, mantle_status, mempool_view, time_info,
transaction, version, wallet,
};
use crate::{
BlendService, TracingService, WalletService,
@@ -251,10 +252,18 @@ where
paths::BLEND_JOIN_NETWORK,
routing::post(blend_join_network::<BlendService, RuntimeServiceId>),
)
.route(
paths::BLEND_PENDING_TRANSACTIONS,
routing::get(blend_pending_transactions::<BlendService, RuntimeServiceId>),
)
.route(
paths::MEMPOOL_ADD_TX,
routing::post(add_tx::<MempoolStorageAdapter, RuntimeServiceId>),
)
.route(
paths::BLEND_DISPERSE_TRANSACTION,
routing::post(blend_tx::<BlendService, RuntimeServiceId>),
)
.route(
paths::MEMPOOL_VIEW,
routing::get(mempool_view::<MempoolStorageAdapter, RuntimeServiceId>),
+51
View File
@@ -686,6 +686,57 @@ where
))
}
#[utoipa::path(
get,
path = paths::BLEND_PENDING_TRANSACTIONS,
responses(
(status = 200, description = "Ids of the transactions waiting for a PoW solution before they can be blended", body = Vec<TxHash>),
(status = 500, description = "Internal server error", body = ErrorBody),
)
)]
pub async fn blend_pending_transactions<BlendService, RuntimeServiceId>(
State(handle): State<OverwatchHandle<RuntimeServiceId>>,
) -> Response
where
BlendService: ServiceData<
Message = ProxyServiceMessage<lb_blend_service::message::ServiceMessage<PeerId>>,
> + 'static,
RuntimeServiceId: Debug + Sync + Display + 'static + AsServiceId<BlendService>,
{
make_request_and_return_response!(blend::blend_pending_transactions::<
BlendService,
SignedMantleTx<Preverified>,
TxHash,
RuntimeServiceId,
>(&handle, Hashable::hash))
}
#[utoipa::path(
post,
path = paths::BLEND_DISPERSE_TRANSACTION,
responses(
(status = 200, description = "Id of the transaction accepted for blending, which was not added to this node's mempool", body = TxHash),
(status = 500, description = "Internal server error", body = ErrorBody),
)
)]
pub async fn blend_tx<BlendService, RuntimeServiceId>(
State(handle): State<OverwatchHandle<RuntimeServiceId>>,
Json(tx): Json<SignedMantleTx<Preverified>>,
) -> Response
where
BlendService: ServiceData<
Message = ProxyServiceMessage<lb_blend_service::message::ServiceMessage<PeerId>>,
> + 'static,
RuntimeServiceId: Debug + Sync + Display + 'static + AsServiceId<BlendService>,
{
make_request_and_return_response!(blend::blend_transaction::<
BlendService,
SignedMantleTx<Preverified>,
TxHash,
RuntimeServiceId,
>(&handle, tx, Hashable::hash))
}
#[utoipa::path(
post,
path = paths::MEMPOOL_ADD_TX,
+2
View File
@@ -15,6 +15,8 @@ use utoipa::OpenApi;
crate::api::handlers::libp2p_info,
crate::api::handlers::dial_peer,
crate::api::handlers::add_tx,
crate::api::handlers::blend_tx,
crate::api::handlers::blend_pending_transactions,
crate::api::handlers::mempool_view,
crate::api::handlers::channel,
crate::api::handlers::channel_deposit,
+1 -1
View File
@@ -1,7 +1,7 @@
use lb_blend_service::{
core::{
backends::libp2p::Libp2pBlendBackendSettings as Libp2pCoreBlendBackendSettings,
network::libp2p::Libp2pBroadcastSettings,
dispatcher::libp2p::Libp2pBroadcastSettings,
settings::{
CoverTrafficSettings, MessageDelayerSettings, SchedulerSettings,
StartingBlendConfig as BlendCoreSettings, ZkSettings,
@@ -4,8 +4,8 @@ use lb_blend::{
proofs::{quota::VerifiedProofOfQuota, selection::VerifiedProofOfSelection},
scheduling::message_blend::provers::{
BlendLayerProof, ProofsGeneratorSettings, WinningPolInfoStream,
core_leader_and_pow::RealCoreLeaderAndPowProofsGenerator,
leader::{LeaderProofsGenerator, RealLeaderProofsGenerator},
core_leader_and_pow::RealCoreLeaderAndPowProofsGenerator, leader::LeaderProofsGenerator,
leader_and_pow::RealLeaderAndPowProofsGenerator,
},
};
use lb_blend_service::{RealProofsVerifier, core::kms::PreloadKMSBackendCorePoQGenerator};
@@ -14,14 +14,25 @@ use lb_storage_service::{backends::rocksdb::RocksBackend, recovery::StorageRecov
use lb_time_service::backends::NtpTimeBackend;
use libp2p::PeerId;
use crate::generic_services::{CryptarchiaService, SdpService, blend::pol::PolInfoProvider};
use crate::generic_services::{
CryptarchiaService, MempoolNetworkAdapter, MempoolPool, SdpService, blend::pol::PolInfoProvider,
};
pub(crate) mod pol;
/// Blend's exit door on this node: block proposals go back onto the chain's
/// gossipsub topic, transactions go to the mempool.
pub type BlendPayloadDispatcher<RuntimeServiceId> =
lb_blend_service::core::dispatcher::libp2p::Libp2pPayloadDispatcher<
MempoolNetworkAdapter<RuntimeServiceId>,
MempoolPool<RuntimeServiceId>,
RuntimeServiceId,
>;
pub type BlendCoreRecoveryBackend<RuntimeServiceId> = StorageRecoveryBackend<
lb_blend_service::core::CoreServiceState<
lb_blend_service::core::backends::libp2p::Libp2pBlendBackendSettings,
<lb_blend_service::core::network::libp2p::Libp2pAdapter<RuntimeServiceId> as lb_blend_service::core::network::NetworkAdapter<RuntimeServiceId>>::Settings,
BlendBroadcastSettings<RuntimeServiceId>,
>,
lb_blend_service::core::settings::StartingBlendConfig<
lb_blend_service::core::backends::libp2p::Libp2pBlendBackendSettings,
@@ -34,7 +45,7 @@ pub type BlendCoreRecoveryBackend<RuntimeServiceId> = StorageRecoveryBackend<
pub type BlendCoreService<RuntimeServiceId> = lb_blend_service::core::BlendService<
lb_blend_service::core::backends::libp2p::Libp2pBlendBackend<RealProofsVerifier>,
PeerId,
lb_blend_service::core::network::libp2p::Libp2pAdapter<RuntimeServiceId>,
BlendPayloadDispatcher<RuntimeServiceId>,
SdpService<RuntimeServiceId>,
RealCoreLeaderAndPowProofsGenerator<PreloadKMSBackendCorePoQGenerator<RuntimeServiceId>>,
RealProofsVerifier,
@@ -69,7 +80,7 @@ impl LeaderProofsGenerator for MockLeaderProofsGenerator {
pub type BlendEdgeService<RuntimeServiceId> = lb_blend_service::edge::BlendService<
lb_blend_service::edge::backends::libp2p::Libp2pBlendBackend,
PeerId,
RealLeaderProofsGenerator,
RealLeaderAndPowProofsGenerator,
NtpTimeBackend,
CryptarchiaService<RuntimeServiceId>,
PolInfoProvider,
@@ -82,5 +93,6 @@ pub type BlendService<RuntimeServiceId> = lb_blend_service::BlendService<
RuntimeServiceId,
>;
pub type BlendBroadcastSettings<RuntimeServiceId> =
<lb_blend_service::core::network::libp2p::Libp2pAdapter<RuntimeServiceId> as lb_blend_service::core::network::NetworkAdapter<RuntimeServiceId>>::Settings;
pub type BlendBroadcastSettings<RuntimeServiceId> = <BlendPayloadDispatcher<RuntimeServiceId> as lb_blend_service::core::dispatcher::PayloadDispatcher<
RuntimeServiceId,
>>::Settings;
+20 -16
View File
@@ -20,26 +20,30 @@ use crate::generic_services::blend::BlendService;
pub mod blend;
pub mod sdp;
pub type TxMempoolService<RuntimeServiceId> = lb_tx_service::TxMempoolService<
pub type MempoolNetworkAdapter<RuntimeServiceId> =
lb_tx_service::network::adapters::libp2p::Libp2pAdapter<
SignedMantleTx<Preverified>,
<SignedMantleTx<Preverified> as Hashable>::Hash,
RuntimeServiceId,
>,
Mempool<
HeaderId,
SignedMantleTx<Preverified>,
TxHash,
RocksStorageAdapter<
SignedMantleTx<Preverified>,
<SignedMantleTx<Preverified> as Hashable>::Hash,
>,
RuntimeServiceId,
>,
RocksStorageAdapter<
SignedMantleTx<Preverified>,
<SignedMantleTx<Preverified> as Hashable>::Hash,
>,
>;
pub type MempoolRocksStorageAdapter = RocksStorageAdapter<
SignedMantleTx<Preverified>,
<SignedMantleTx<Preverified> as Hashable>::Hash,
>;
pub type MempoolPool<RuntimeServiceId> = Mempool<
HeaderId,
SignedMantleTx<Preverified>,
TxHash,
MempoolRocksStorageAdapter,
RuntimeServiceId,
>;
pub type TxMempoolService<RuntimeServiceId> = lb_tx_service::TxMempoolService<
MempoolNetworkAdapter<RuntimeServiceId>,
MempoolPool<RuntimeServiceId>,
MempoolRocksStorageAdapter,
RuntimeServiceId,
>;
+1 -4
View File
@@ -10,10 +10,7 @@ pub mod global_allocators;
use std::panic::set_hook;
use color_eyre::eyre::{Result, eyre};
pub use lb_blend_service::core::{
backends::libp2p::Libp2pBlendBackend as BlendBackend,
network::libp2p::Libp2pAdapter as BlendNetworkAdapter,
};
pub use lb_blend_service::core::backends::libp2p::Libp2pBlendBackend as BlendBackend;
use lb_core::mantle::transactions::states::Preverified;
pub use lb_core::{
codec,
+40 -3
View File
@@ -29,9 +29,10 @@ use lb_http_api_common::{
},
},
paths::{
BLEND_JOIN_NETWORK, BLOCK_EVENTS, BLOCKS, BLOCKS_DETAIL, BLOCKS_RANGE_STREAM,
BLOCKS_STREAM, CHANNEL, CRYPTARCHIA_INFO, CRYPTARCHIA_LIB_STREAM, LEADER_CLAIM_VOUCHERS,
MANTLE_GAS_PRICES, MEMPOOL_ADD_TX, NODE_VERSION, SDP_POST_DECLARATION, TIME_INFO,
BLEND_DISPERSE_TRANSACTION, BLEND_JOIN_NETWORK, BLEND_PENDING_TRANSACTIONS, BLOCK_EVENTS,
BLOCKS, BLOCKS_DETAIL, BLOCKS_RANGE_STREAM, BLOCKS_STREAM, CHANNEL, CRYPTARCHIA_INFO,
CRYPTARCHIA_LIB_STREAM, LEADER_CLAIM_VOUCHERS, MANTLE_GAS_PRICES, MEMPOOL_ADD_TX,
NODE_VERSION, SDP_POST_DECLARATION, TIME_INFO,
wallet::{BALANCE, FUND, TRANSACTIONS_TRANSFER_FUNDS},
},
queries::BlocksStreamQuery,
@@ -285,6 +286,42 @@ impl CommonHttpClient {
self.post(request_url, &transaction).await
}
/// Send a transaction through the Blend network, without adding it to the
/// node's own mempool.
///
/// The node it is submitted to never gossips the transaction itself — that
/// would say where it came from — so it reaches the network blended, and
/// comes back to this node's mempool like any other transaction once
/// whichever node exits it gossips it on.
///
/// Returns the transaction's id.
pub async fn blend_transaction<Tx, Id>(
&self,
base_url: Url,
transaction: Tx,
) -> Result<Id, Error>
where
Tx: Serialize + Send + Sync + 'static,
Id: for<'de> Deserialize<'de> + Send + Sync,
{
let request_url = base_url
.join(BLEND_DISPERSE_TRANSACTION.trim_start_matches('/'))
.map_err(Error::Url)?;
self.post(request_url, &transaction).await
}
/// The ids of the transactions the node is still waiting on a `PoW`
/// solution for before it can blend them.
pub async fn blend_pending_transactions<Id>(&self, base_url: Url) -> Result<Vec<Id>, Error>
where
Id: for<'de> Deserialize<'de> + Send + Sync,
{
let request_url = base_url
.join(BLEND_PENDING_TRANSACTIONS.trim_start_matches('/'))
.map_err(Error::Url)?;
self.get::<(), _>(request_url, None).await
}
/// Post a service declaration to the SDP endpoint.
pub async fn post_declaration(
&self,
+66 -1
View File
@@ -1,6 +1,7 @@
use std::fmt::{Debug, Display};
use lb_blend_service::message::{NetworkInfo, ProxyServiceMessage, ServiceMessage};
use lb_blend_service::message::{BlendPayload, NetworkInfo, ProxyServiceMessage, ServiceMessage};
use lb_core::codec::{DeserializeOp, SerializeOp};
use lb_network_service::backends::libp2p::PeerId;
use overwatch::services::{AsServiceId, ServiceData};
use tokio::sync::oneshot;
@@ -52,3 +53,67 @@ where
Ok(result)
}
/// Sends a transaction through the Blend network without adding it to this
/// node's own mempool.
///
/// Returns the transaction's id. Getting one back means the transaction was
/// accepted for blending, not that it was sent.
pub async fn blend_transaction<BlendService, Transaction, Id, RuntimeServiceId>(
handle: &overwatch::overwatch::handle::OverwatchHandle<RuntimeServiceId>,
transaction: Transaction,
id: impl Fn(&Transaction) -> Id,
) -> Result<Id, overwatch::DynError>
where
BlendService: ServiceData<Message = ProxyServiceMessage<ServiceMessage<PeerId>>>,
Transaction: SerializeOp,
RuntimeServiceId: AsServiceId<BlendService> + Debug + Sync + Display + 'static,
{
// Encoded the same way the mempool gossips transactions, so that whichever
// node exits this one decodes what it expects.
let payload = BlendPayload::transaction(transaction.to_bytes()?.to_vec())?;
let relay = handle.relay::<BlendService>().await?;
relay
.send(ServiceMessage::Blend(payload).into())
.await
.map_err(|(e, _)| e)?;
Ok(id(&transaction))
}
/// The ids of the transactions this node is still waiting on a `PoW` solution
/// for.
///
/// Only those: once a transaction has been encapsulated it is in the
/// scheduler's hands, waiting for a release round, and no longer reported here.
// TODO: Have Blend hand back transactions rather than bytes, so that this can
// report their ids without decoding them again.
pub async fn blend_pending_transactions<BlendService, Transaction, Id, RuntimeServiceId>(
handle: &overwatch::overwatch::handle::OverwatchHandle<RuntimeServiceId>,
id: impl Fn(&Transaction) -> Id,
) -> Result<Vec<Id>, overwatch::DynError>
where
BlendService: ServiceData<Message = ProxyServiceMessage<ServiceMessage<PeerId>>>,
Transaction: DeserializeOp,
RuntimeServiceId: AsServiceId<BlendService> + Debug + Sync + Display + 'static,
{
let relay = handle.relay::<BlendService>().await?;
let (sender, receiver) = oneshot::channel();
relay
.send(ServiceMessage::GetPendingTransactions { reply: sender }.into())
.await
.map_err(|(e, _)| e)?;
receiver
.await
.map_err(|e| Box::new(e) as overwatch::DynError)?
.iter()
.map(|encoded_transaction| {
// These are transactions this node encoded itself on the way in, so
// a decoding failure here is this node disagreeing with itself.
Ok(id(&Transaction::from_bytes(encoded_transaction)?))
})
.collect()
}
+2
View File
@@ -30,8 +30,10 @@ lb-network-service = { workspace = true }
lb-poq = { workspace = true }
lb-sdp-service = { workspace = true }
lb-services-utils = { workspace = true }
lb-storage-service = { workspace = true }
lb-time-service = { workspace = true }
lb-tracing = { workspace = true }
lb-tx-service = { workspace = true }
lb-utils = { workspace = true }
libp2p = { features = ["dns"], workspace = true }
libp2p-stream = { workspace = true }
@@ -0,0 +1,176 @@
use core::{
fmt::{Debug, Display},
marker::PhantomData,
};
use lb_core::{
codec::DeserializeOp as _,
header::HeaderId,
mantle::{traits::Hashable, transactions::hash::PrefixedKey},
};
use lb_log_targets::blend;
use lb_network_service::{
NetworkService,
backends::libp2p::{Command, Libp2p, PubSubCommand},
message::NetworkMsg,
};
use lb_storage_service::StorageService;
use lb_tx_service::{
MempoolMsg, TxMempoolService, backend::RecoverableMempool,
network::NetworkAdapter as MempoolNetworkAdapter, storage::MempoolStorageAdapter,
};
use overwatch::services::{AsServiceId, ServiceData, relay::OutboundRelay};
use serde::{Deserialize, Serialize};
use tokio::sync::oneshot;
use super::PayloadDispatcher;
use crate::message::BlendPayload;
const LOG_TARGET: &str = blend::service::CORE;
type MempoolRelay<Item, Key> = OutboundRelay<MempoolMsg<HeaderId, Item, Item, Key>>;
/// A payload dispatcher for a node whose network service uses the libp2p
/// backend.
pub struct Libp2pPayloadDispatcher<MempoolNetAdapter, Mempool, RuntimeServiceId>
where
Mempool: RecoverableMempool<BlockId = HeaderId>,
{
network_relay:
OutboundRelay<<NetworkService<Libp2p, RuntimeServiceId> as ServiceData>::Message>,
mempool_relay: MempoolRelay<Mempool::Item, Mempool::Key>,
settings: Libp2pBroadcastSettings,
_phantom: PhantomData<(MempoolNetAdapter, RuntimeServiceId)>,
}
/// Settings used to broadcast messages to the network service that uses libp2p
/// backend.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash)]
pub struct Libp2pBroadcastSettings {
pub topic: String,
}
impl<MempoolNetAdapter, Mempool, RuntimeServiceId>
Libp2pPayloadDispatcher<MempoolNetAdapter, Mempool, RuntimeServiceId>
where
Mempool: RecoverableMempool<BlockId = HeaderId>,
MempoolNetAdapter: Sync,
RuntimeServiceId: Sync,
{
/// Broadcast an unencrypted message to the network by publishing the
/// message under the configured gossipsub topic.
async fn broadcast_block_proposal(&self, proposal: Vec<u8>) {
if let Err((e, _)) = self
.network_relay
.send(NetworkMsg::Process(Command::PubSub(
PubSubCommand::Broadcast {
topic: self.settings.topic.clone(),
message: proposal.into_boxed_slice(),
},
)))
.await
{
tracing::error!(target: LOG_TARGET, "error broadcasting block proposal: {e}");
}
}
}
impl<MempoolNetAdapter, Mempool, RuntimeServiceId>
Libp2pPayloadDispatcher<MempoolNetAdapter, Mempool, RuntimeServiceId>
where
Mempool: RecoverableMempool<BlockId = HeaderId>,
Mempool::Item: Hashable<Hash = Mempool::Key> + serde::de::DeserializeOwned + Send + 'static,
Mempool::Key: PrefixedKey<Prefix: Send + Sync> + Send + 'static,
MempoolNetAdapter: Sync,
RuntimeServiceId: Sync,
{
/// Submit a decapsulated transaction to the local mempool after validating
/// its structure.
async fn submit_transaction(&self, transaction: Vec<u8>) {
let Ok(transaction) = Mempool::Item::from_bytes(&transaction).inspect_err(|e| {
tracing::error!(
target: LOG_TARGET,
"Discarding a decapsulated payload that does not decode as a transaction: {e}"
);
}) else {
return;
};
let (reply_channel, receiver) = oneshot::channel();
if let Err((e, _)) = self
.mempool_relay
.send(MempoolMsg::Add {
key: transaction.hash(),
payload: transaction,
reply_channel,
})
.await
{
tracing::error!(target: LOG_TARGET, "Error submitting a blended transaction to the mempool: {e}");
return;
}
let outcome = receiver
.await
.map_err(|e| format!("the mempool dropped the reply: {e}"))
.and_then(|added| added.map_err(|e| format!("the mempool refused it: {e}")));
if let Err(reason) = outcome {
tracing::debug!(target: LOG_TARGET, "Blended transaction was not added to the mempool: {reason}");
}
}
}
#[async_trait::async_trait]
impl<MempoolNetAdapter, Mempool, RuntimeServiceId> PayloadDispatcher<RuntimeServiceId>
for Libp2pPayloadDispatcher<MempoolNetAdapter, Mempool, RuntimeServiceId>
where
Mempool: RecoverableMempool<BlockId = HeaderId, RecoveryState: 'static> + Send + Sync + 'static,
Mempool::Item: Hashable<Hash = Mempool::Key> + serde::de::DeserializeOwned + Send + 'static,
Mempool::Key: PrefixedKey<Prefix: Send + Sync> + Send + 'static,
Mempool::Settings: Clone + Send + Sync + 'static,
Mempool::Storage: MempoolStorageAdapter<RuntimeServiceId> + Clone + Send + Sync + 'static,
MempoolNetAdapter: MempoolNetworkAdapter<RuntimeServiceId, Payload = Mempool::Item, Key = Mempool::Key>
+ Send
+ Sync
+ 'static,
MempoolNetAdapter::Settings: Clone + Send + Sync + 'static,
RuntimeServiceId: Clone
+ Debug
+ Display
+ Sync
+ Send
+ 'static
+ AsServiceId<
StorageService<
<Mempool::Storage as MempoolStorageAdapter<RuntimeServiceId>>::Backend,
RuntimeServiceId,
>,
>,
{
type Backend = Libp2p;
type MempoolService =
TxMempoolService<MempoolNetAdapter, Mempool, Mempool::Storage, RuntimeServiceId>;
type Settings = Libp2pBroadcastSettings;
fn new(
network_relay: OutboundRelay<
<NetworkService<Self::Backend, RuntimeServiceId> as ServiceData>::Message,
>,
mempool_relay: OutboundRelay<<Self::MempoolService as ServiceData>::Message>,
settings: Self::Settings,
) -> Self {
Self {
network_relay,
mempool_relay,
settings,
_phantom: PhantomData,
}
}
async fn dispatch(&self, payload: BlendPayload) {
match payload {
BlendPayload::BlockProposal(proposal) => self.broadcast_block_proposal(proposal).await,
BlendPayload::Transaction(transaction) => self.submit_transaction(transaction).await,
}
}
}
+37
View File
@@ -0,0 +1,37 @@
use std::fmt::Debug;
use lb_network_service::{NetworkService, backends::NetworkBackend};
use overwatch::services::{ServiceData, relay::OutboundRelay};
use serde::{Serialize, de::DeserializeOwned};
use crate::message::BlendPayload;
pub mod libp2p;
/// Hands a fully decapsulated payload over to the local service that owns it.
///
/// This is Blend's exit door: whatever comes through it has finished blending
/// and travels onwards in the clear. Where "onwards" is depends on what the
/// payload carries — a block proposal is republished under the chain's
/// gossipsub topic, a transaction goes to the mempool, which validates it and
/// gossips it on from there.
#[async_trait::async_trait]
pub trait PayloadDispatcher<RuntimeServiceId> {
/// The network backend used by the network service.
type Backend: NetworkBackend<RuntimeServiceId> + 'static;
/// The mempool service transactions are handed over to.
type MempoolService: ServiceData<Message: Send + 'static> + 'static;
/// Settings used to broadcast messages using the network service.
type Settings: Clone + Debug + Serialize + DeserializeOwned + Send + Sync + 'static;
fn new(
network_relay: OutboundRelay<
<NetworkService<Self::Backend, RuntimeServiceId> as ServiceData>::Message,
>,
mempool_relay: OutboundRelay<<Self::MempoolService as ServiceData>::Message>,
settings: Self::Settings,
) -> Self;
/// Deliver a decapsulated payload to the local service that owns it.
async fn dispatch(&self, payload: BlendPayload);
}
+336 -137
View File
@@ -1,4 +1,5 @@
use std::{
collections::VecDeque,
fmt::{Debug, Display},
hash::Hash,
marker::PhantomData,
@@ -7,6 +8,7 @@ use std::{
use async_trait::async_trait;
use backends::BlendBackend;
use dispatcher::PayloadDispatcher;
use fork_stream::StreamExt as _;
use futures::{
FutureExt as _, Stream, StreamExt as _,
@@ -58,7 +60,6 @@ use lb_services_utils::{
};
use lb_time_service::TimeService;
use lb_utils::blake_rng::BlakeRng;
use network::NetworkAdapter;
use overwatch::{
OpaqueServiceResourcesHandle,
overwatch::OverwatchHandle,
@@ -88,12 +89,12 @@ use crate::{
epoch_info::{PolEpochInfo, PolInfoProvider as PolInfoProviderTrait},
kms::PreloadKmsService,
membership::{self, ZkInfo, chain::BlendEpochState},
message::{NetworkMessage, ProcessedMessage, ServiceMessage},
message::{BlendPayload, ProcessedMessage, ServiceMessage},
};
pub mod backends;
pub mod dispatcher;
pub mod kms;
pub mod network;
pub mod settings;
pub(super) mod service_components;
@@ -117,7 +118,7 @@ const LOG_TARGET: &str = blend::service::CORE;
pub struct BlendService<
Backend,
NodeId,
Network,
Dispatcher,
SdpService,
ProofsGenerator,
ProofsVerifier,
@@ -128,15 +129,15 @@ pub struct BlendService<
RuntimeServiceId,
> where
Backend: BlendBackend<NodeId, BlakeRng, ProofsVerifier, RuntimeServiceId>,
Network: NetworkAdapter<RuntimeServiceId>,
Dispatcher: PayloadDispatcher<RuntimeServiceId>,
StateStorage: RecoveryBackendTrait<
RuntimeServiceId,
State = RecoveryServiceState<Backend::Settings, Network::Settings>,
State = RecoveryServiceState<Backend::Settings, Dispatcher::Settings>,
> + Send
+ Sync,
{
service_resources_handle: OpaqueServiceResourcesHandle<Self, RuntimeServiceId>,
last_saved_state: Option<ServiceState<Backend::Settings, Network::Settings>>,
last_saved_state: Option<ServiceState<Backend::Settings, Dispatcher::Settings>>,
_phantom: PhantomData<(
Backend,
SdpService,
@@ -151,7 +152,7 @@ pub struct BlendService<
impl<
Backend,
NodeId,
Network,
Dispatcher,
SdpService,
ProofsGenerator,
ProofsVerifier,
@@ -164,7 +165,7 @@ impl<
for BlendService<
Backend,
NodeId,
Network,
Dispatcher,
SdpService,
ProofsGenerator,
ProofsVerifier,
@@ -176,15 +177,15 @@ impl<
>
where
Backend: BlendBackend<NodeId, BlakeRng, ProofsVerifier, RuntimeServiceId>,
Network: NetworkAdapter<RuntimeServiceId>,
Dispatcher: PayloadDispatcher<RuntimeServiceId>,
StateStorage: RecoveryBackendTrait<
RuntimeServiceId,
State = RecoveryServiceState<Backend::Settings, Network::Settings>,
State = RecoveryServiceState<Backend::Settings, Dispatcher::Settings>,
> + Send
+ Sync,
{
type Settings = StartingBlendConfig<Backend::Settings, Network::Settings>;
type State = RecoveryServiceState<Backend::Settings, Network::Settings>;
type Settings = StartingBlendConfig<Backend::Settings, Dispatcher::Settings>;
type State = RecoveryServiceState<Backend::Settings, Dispatcher::Settings>;
type StateOperator = RecoveryOperator<StateStorage>;
type Message = ServiceMessage<NodeId>;
}
@@ -193,7 +194,7 @@ where
impl<
Backend,
NodeId,
Network,
Dispatcher,
SdpService,
ProofsGenerator,
ProofsVerifier,
@@ -206,7 +207,7 @@ impl<
for BlendService<
Backend,
NodeId,
Network,
Dispatcher,
SdpService,
ProofsGenerator,
ProofsVerifier,
@@ -219,7 +220,7 @@ impl<
where
Backend: BlendBackend<NodeId, BlakeRng, ProofsVerifier, RuntimeServiceId> + Send + Sync,
NodeId: membership::node_id::TryFrom + Clone + Debug + Send + Eq + Hash + Sync + 'static,
Network: NetworkAdapter<RuntimeServiceId> + Send + Sync,
Dispatcher: PayloadDispatcher<RuntimeServiceId> + Send + Sync,
ProofsGenerator:
CoreLeaderAndPowProofsGenerator<PreloadKMSBackendCorePoQGenerator<RuntimeServiceId>> + Send,
SdpService: ServiceData<Message = SdpMessage> + Send,
@@ -229,10 +230,11 @@ where
PolInfoProvider: PolInfoProviderTrait<RuntimeServiceId, Stream: Send + Unpin + 'static> + Send,
StateStorage: RecoveryBackendTrait<
RuntimeServiceId,
State = RecoveryServiceState<Backend::Settings, Network::Settings>,
State = RecoveryServiceState<Backend::Settings, Dispatcher::Settings>,
> + Send
+ Sync,
RuntimeServiceId: AsServiceId<NetworkService<Network::Backend, RuntimeServiceId>>
RuntimeServiceId: AsServiceId<NetworkService<Dispatcher::Backend, RuntimeServiceId>>
+ AsServiceId<Dispatcher::MempoolService>
+ AsServiceId<SdpService>
+ AsServiceId<TimeService<TimeBackend, RuntimeServiceId>>
+ AsServiceId<ChainService>
@@ -301,12 +303,16 @@ where
)
.await?;
let network_adapter = async {
let payload_dispatcher = async {
let network_relay = overwatch_handle
.relay::<NetworkService<_, _>>()
.await
.expect("Relay with network service should be available.");
Network::new(network_relay, blend_config.network.clone())
let mempool_relay = overwatch_handle
.relay::<Dispatcher::MempoolService>()
.await
.expect("Relay with mempool service should be available.");
Dispatcher::new(network_relay, mempool_relay, blend_config.network.clone())
}
.await;
@@ -374,13 +380,14 @@ where
current_public_info,
crypto_processor,
current_recovery_checkpoint,
pending_transactions,
message_scheduler,
mut backend,
mut rng,
) = initialize::<
NodeId,
Backend,
Network,
Dispatcher,
ProofsGenerator,
ProofsVerifier,
KmsServiceApi<PreloadKmsService<RuntimeServiceId>, RuntimeServiceId>,
@@ -424,10 +431,11 @@ where
&mut remaining_epoch_stream,
&running_blend_config,
&mut backend,
&network_adapter,
&payload_dispatcher,
&sdp_relay,
message_scheduler.into(),
&mut rng,
pending_transactions,
crypto_processor,
current_public_info,
current_recovery_checkpoint,
@@ -444,7 +452,7 @@ where
blend_messages.map(|(message, _)| message),
remaining_epoch_stream,
backend,
network_adapter,
payload_dispatcher,
sdp_relay,
old_epoch_message_scheduler,
rng,
@@ -466,7 +474,7 @@ where
async fn initialize<
NodeId,
Backend,
NetAdapter,
Dispatcher,
ProofsGenerator,
ProofsVerifier,
KmsAdapter,
@@ -477,9 +485,9 @@ async fn initialize<
overwatch_handle: OverwatchHandle<RuntimeServiceId>,
kms_adapter: KmsAdapter,
sdp_relay: &OutboundRelay<SdpMessage>,
mut last_saved_state: Option<ServiceState<Backend::Settings, NetAdapter::Settings>>,
mut last_saved_state: Option<ServiceState<Backend::Settings, Dispatcher::Settings>>,
state_updater: StateUpdater<
Option<RecoveryServiceState<Backend::Settings, NetAdapter::Settings>>,
Option<RecoveryServiceState<Backend::Settings, Dispatcher::Settings>>,
>,
) -> (
impl Stream<Item = EpochEvent<MaybeEmptyCoreEpochInfo<NodeId, KmsAdapter::CorePoQGenerator>>>
@@ -493,7 +501,8 @@ async fn initialize<
ProofsGenerator,
ProofsVerifier,
>,
ServiceState<Backend::Settings, NetAdapter::Settings>,
ServiceState<Backend::Settings, Dispatcher::Settings>,
VecDeque<Vec<u8>>,
SchedulerWrapper<BlakeRng, ProcessedMessage, EncapsulatedMessageWithVerifiedPublicHeader>,
Backend,
BlakeRng,
@@ -501,7 +510,7 @@ async fn initialize<
where
NodeId: Clone + Debug + Eq + Hash + Send + 'static,
Backend: BlendBackend<NodeId, BlakeRng, ProofsVerifier, RuntimeServiceId> + Sync,
NetAdapter: NetworkAdapter<RuntimeServiceId>,
Dispatcher: PayloadDispatcher<RuntimeServiceId>,
ProofsGenerator: CoreLeaderAndPowProofsGenerator<KmsAdapter::CorePoQGenerator>,
ProofsVerifier: ProofsVerifierTrait,
// To avoid bubbling up generics everywhere in the configs (current Overwatch limitation), we
@@ -619,24 +628,32 @@ where
// Initialize the current epoch state. If the epoch matches the stored one,
// retrieves the tracked consumed core quota. Else, fallback to `0`.
let current_recovery_checkpoint = if let Some(saved_state) = last_saved_state.take()
&& saved_state.last_seen_epoch() == current_epoch_public_info.epoch
{
tracing::trace!(
target: LOG_TARGET,
"Found recovery state for epoch {:?}: {saved_state:?}",
current_epoch_public_info.epoch
);
saved_state
} else {
tracing::trace!(
target: LOG_TARGET,
"No recovery state found for epoch {:?}. Initializing a new one.",
current_epoch_public_info.epoch
);
let current_recovery_checkpoint = match last_saved_state.take() {
Some(saved_state) if saved_state.last_seen_epoch() == current_epoch_public_info.epoch => {
tracing::trace!(
target: LOG_TARGET,
"Found recovery state for epoch {:?}: {saved_state:?}",
current_epoch_public_info.epoch
);
saved_state
}
maybe_stale_state => {
tracing::trace!(
target: LOG_TARGET,
"No recovery state found for epoch {:?}. Initializing a new one.",
current_epoch_public_info.epoch
);
ServiceState::with_epoch(
current_epoch_public_info.epoch,
// Everything else in a stale state belongs to the epoch it was
// saved under, but a transaction still waiting for a `PoW` solution
// has not been encapsulated and so belongs to none: it outlives the
// state that carried it, the same way it outlives an epoch rotation.
let pending_transactions =
maybe_stale_state.map_or_else(VecDeque::new, |state| state.into_components().4);
ServiceState::with_epoch(
current_epoch_public_info.epoch,
pending_transactions,
EpochBlendingTokenCollector::new(
&reward::EpochInfo::new(
current_epoch_public_info.epoch,
@@ -645,10 +662,12 @@ where
current_epoch_public_info.poq_core_public_inputs.quota,
blend_config.activity_threshold_sensitivity,
).expect("Reward epoch info must be created successfully. Panicking since the service cannot continue with this epoch")
),
None,
state_updater,
).expect("service state should be created successfully")
),
None,
state_updater,
)
.expect("service state should be created successfully")
}
};
// If there is the old epoch token collector loaded from `last_saved_state`,
@@ -699,11 +718,14 @@ where
// Rng for releasing messages.
let rng = BlakeRng::from_entropy();
let pending_transactions = current_recovery_checkpoint.pending_transactions().clone();
(
remaining_epoch_stream,
current_epoch_public_info,
crypto_processor,
current_recovery_checkpoint,
pending_transactions,
message_scheduler,
backend,
rng,
@@ -742,7 +764,7 @@ async fn run_event_loop<
NodeId,
Backend,
Rng,
NetAdapter,
Dispatcher,
ProofsGenerator,
ProofsVerifier,
CorePoQGenerator,
@@ -763,7 +785,7 @@ async fn run_event_loop<
),
blend_config: &RunningBlendConfig<Backend::Settings>,
backend: &mut Backend,
network_adapter: &NetAdapter,
payload_dispatcher: &Dispatcher,
sdp_relay: &OutboundRelay<SdpMessage>,
mut message_scheduler: EpochMessageScheduler<
Rng,
@@ -771,6 +793,7 @@ async fn run_event_loop<
EncapsulatedMessageWithVerifiedPublicHeader,
>,
rng: &mut Rng,
mut pending_transactions: VecDeque<Vec<u8>>,
mut crypto_processor: CoreCryptographicProcessor<
NodeId,
CorePoQGenerator,
@@ -778,7 +801,7 @@ async fn run_event_loop<
ProofsVerifier,
>,
mut current_epoch_info: CoreEpochPublicInfo<NodeId>,
mut recovery_checkpoint: ServiceState<Backend::Settings, NetAdapter::Settings>,
mut recovery_checkpoint: ServiceState<Backend::Settings, Dispatcher::Settings>,
) -> (
CoreCryptographicProcessor<NodeId, CorePoQGenerator, ProofsGenerator, ProofsVerifier>,
OldEpochMessageScheduler<Rng, ProcessedMessage, EncapsulatedMessageWithVerifiedPublicHeader>,
@@ -788,7 +811,7 @@ where
NodeId: Clone + Eq + Hash + Send + Sync + 'static,
Rng: rand::Rng + Clone + Send + Unpin,
Backend: BlendBackend<NodeId, BlakeRng, ProofsVerifier, RuntimeServiceId> + Sync + Send,
NetAdapter: NetworkAdapter<RuntimeServiceId> + Sync,
Dispatcher: PayloadDispatcher<RuntimeServiceId> + Sync,
ProofsGenerator: CoreLeaderAndPowProofsGenerator<CorePoQGenerator> + Send,
CorePoQGenerator: Send + Sync,
ProofsVerifier: ProofsVerifierTrait + Send + Sync,
@@ -816,28 +839,31 @@ where
tokio::select! {
Some(msg) = inbound_relay.next() => {
match msg {
ServiceMessage::Blend(message_payload) => {
// The Blend payload is exactly the message bytes: where a
// receiving node republishes them is its own configuration,
// so nothing about the destination travels with them.
let serialized_data_message = message_payload;
let message_copies = blend_config.data_replication_factor.checked_add(1).unwrap();
for _ in 0..message_copies {
recovery_checkpoint = handle_serialized_local_data_message(&serialized_data_message, &mut crypto_processor, &mut message_scheduler, recovery_checkpoint).await;
}
ServiceMessage::Blend(BlendPayload::Transaction(transaction)) => {
recovery_checkpoint = queue_transaction_for_encapsulation(transaction, &mut pending_transactions, recovery_checkpoint);
}
ServiceMessage::Blend(BlendPayload::BlockProposal(proposal)) => {
recovery_checkpoint = handle_local_block_proposal(&proposal, blend_config.data_replication_factor, &mut crypto_processor, &mut message_scheduler, recovery_checkpoint).await;
}
ServiceMessage::GetNetworkInfo { reply } => {
let info = backend.network_info().await;
drop(reply.send(info));
}
ServiceMessage::GetPendingTransactions { reply } => {
drop(reply.send(pending_transactions.iter().cloned().collect()));
}
}
}
// A queued transaction leaves as soon as a `PoW` solution backs it. The
// search is awaited here, so the rest of the loop keeps turning while it runs.
Some(encapsulation) = encapsulate_next_transaction(&pending_transactions, &mut crypto_processor) => {
recovery_checkpoint = handle_local_transaction(&encapsulation, &mut pending_transactions, &crypto_processor, &mut message_scheduler, recovery_checkpoint);
}
Some(incoming_message) = blend_messages.next() => {
recovery_checkpoint = handle_incoming_blend_message(incoming_message, &mut message_scheduler, old_epoch_message_scheduler.as_mut(), &crypto_processor, old_epoch_crypto_processor.as_ref(), recovery_checkpoint);
}
Some(round_info) = message_scheduler.next() => {
recovery_checkpoint = handle_release_round(round_info, &mut crypto_processor, rng, backend, network_adapter, recovery_checkpoint).await;
recovery_checkpoint = handle_release_round(round_info, &mut crypto_processor, rng, backend, payload_dispatcher, recovery_checkpoint).await;
}
Some((Some(round_info), previous_epoch)) = async {
match (&mut old_epoch_message_scheduler, old_epoch) {
@@ -847,7 +873,7 @@ where
_ => None
}
} => {
handle_release_round_for_old_epoch(round_info, rng, backend, network_adapter, previous_epoch).await;
handle_release_round_for_old_epoch(round_info, rng, backend, payload_dispatcher, previous_epoch).await;
}
Some(pol_secret_info) = secret_pol_info_stream.next() => {
if current_epoch_info.epoch == pol_secret_info.epoch {
@@ -895,6 +921,114 @@ where
}
}
/// Records a transaction as waiting for a `PoW` solution to back its layer
/// proofs.
fn queue_transaction_for_encapsulation<BackendSettings, NetworkSettings>(
transaction: Vec<u8>,
pending_transactions: &mut VecDeque<Vec<u8>>,
current_recovery_checkpoint: ServiceState<BackendSettings, NetworkSettings>,
) -> ServiceState<BackendSettings, NetworkSettings>
where
BackendSettings: Clone,
{
pending_transactions.push_back(transaction.clone());
let mut state_updater = current_recovery_checkpoint.start_updating();
state_updater.queue_unencapsulated_transaction(transaction);
state_updater.commit_changes()
}
/// Encapsulates the transaction that has been waiting longest, once a `PoW`
/// solution backs its layer proofs.
///
/// The transaction is only read here, never taken off the queue: `select!`
/// drops this future whenever another branch wins the race, and a future that
/// popped before awaiting would take the transaction down with it every time
/// that happened. It comes off the queue in [`handle_local_transaction`]
/// instead, which runs after the race is settled.
///
/// Returns `None` when there is nothing to hand back — either nothing is
/// queued, or the transaction at the head could not be encapsulated — which is
/// what leaves the `select!` branch free to wait on the others. The two are not
/// worth telling apart here: in both cases the right move is to do nothing this
/// time round.
///
/// A transaction that fails to encapsulate therefore stays queued and is tried
/// again.
async fn encapsulate_next_transaction<NodeId, ProofsGenerator, ProofsVerifier, CorePoQGenerator>(
pending_transactions: &VecDeque<Vec<u8>>,
cryptographic_processor: &mut CoreCryptographicProcessor<
NodeId,
CorePoQGenerator,
ProofsGenerator,
ProofsVerifier,
>,
) -> Option<EncapsulatedMessageWithVerifiedPublicHeader>
where
NodeId: Eq + Hash + 'static,
ProofsGenerator: CoreLeaderAndPowProofsGenerator<CorePoQGenerator>,
{
let transaction = pending_transactions.front()?;
cryptographic_processor
.encapsulate_transaction_payload(transaction)
.await
// Reported here rather than handed back: the encapsulation error is not
// `Send`, and a `select!` branch output has to be.
.inspect_err(|e| {
tracing::error!(target: LOG_TARGET, "Failed to encapsulate transaction: {e:?}");
})
.ok()
}
/// Processes a transaction whose wait for a `PoW` solution is over.
///
/// The counterpart to [`handle_local_block_proposal`], split in two because a
/// transaction cannot be dealt with where it arrives: leadership proofs are
/// ready on demand, whereas a transaction's have to be mined for.
fn handle_local_transaction<
NodeId,
Rng,
BackendSettings,
NetworkSettings,
ProofsGenerator,
ProofsVerifier,
CorePoQGenerator,
>(
encapsulation: &EncapsulatedMessageWithVerifiedPublicHeader,
pending_transactions: &mut VecDeque<Vec<u8>>,
cryptographic_processor: &CoreCryptographicProcessor<
NodeId,
CorePoQGenerator,
ProofsGenerator,
ProofsVerifier,
>,
scheduler: &mut EpochMessageScheduler<
Rng,
ProcessedMessage,
EncapsulatedMessageWithVerifiedPublicHeader,
>,
current_recovery_checkpoint: ServiceState<BackendSettings, NetworkSettings>,
) -> ServiceState<BackendSettings, NetworkSettings>
where
NodeId: Eq + Hash + Send + 'static,
Rng: RngCore + Clone + Send + Unpin,
BackendSettings: Clone + Send + Sync,
ProofsVerifier: ProofsVerifierTrait,
{
let recovery_checkpoint = schedule_local_encapsulated_message(
encapsulation,
cryptographic_processor,
scheduler,
current_recovery_checkpoint,
);
let transaction = pending_transactions
.pop_front()
.expect("Branch only yields while a transaction is queued.");
let mut state_updater = recovery_checkpoint.start_updating();
state_updater.dequeue_unencapsulated_transaction(&transaction);
state_updater.commit_changes()
}
/// Processes the old epoch during the epoch transition period
/// before retiring the core service.
#[expect(clippy::too_many_arguments, reason = "categorize args")]
@@ -902,7 +1036,7 @@ async fn retire<
NodeId,
Backend,
Rng,
NetAdapter,
Dispatcher,
ProofsGenerator,
ProofsVerifier,
CorePoQGenerator,
@@ -917,7 +1051,7 @@ async fn retire<
> + Send
+ Unpin,
mut backend: Backend,
network_adapter: NetAdapter,
payload_dispatcher: Dispatcher,
sdp_relay: OutboundRelay<SdpMessage>,
mut message_scheduler: OldEpochMessageScheduler<
Rng,
@@ -936,7 +1070,7 @@ async fn retire<
NodeId: Clone + Eq + Hash + Send + Sync + 'static,
Rng: rand::Rng + Clone + Send + Unpin,
Backend: BlendBackend<NodeId, BlakeRng, ProofsVerifier, RuntimeServiceId> + Send + Sync,
NetAdapter: NetworkAdapter<RuntimeServiceId> + Send + Sync,
Dispatcher: PayloadDispatcher<RuntimeServiceId> + Send + Sync,
ProofsGenerator: CoreLeaderAndPowProofsGenerator<CorePoQGenerator> + Send,
CorePoQGenerator: Send + Sync,
ProofsVerifier: ProofsVerifierTrait + Send + Sync,
@@ -948,7 +1082,7 @@ async fn retire<
handle_incoming_blend_message_from_old_epoch(incoming_message, &mut message_scheduler, &crypto_processor, &mut blending_token_collector);
}
Some(round_info) = message_scheduler.next() => {
handle_release_round_for_old_epoch(round_info, &mut rng, &backend, &network_adapter, crypto_processor.epoch()).await;
handle_release_round_for_old_epoch(round_info, &mut rng, &backend, &payload_dispatcher, crypto_processor.epoch()).await;
}
Some(EpochEvent::TransitionPeriodExpired) = remaining_epoch_stream.next() => {
handle_epoch_transition_expired(&mut backend, blending_token_collector, &sdp_relay).await;
@@ -1026,8 +1160,16 @@ where
// proof generator for the epoch transition period.
let mut current_cryptographic_processor = current_cryptographic_processor;
current_cryptographic_processor.stop_proof_generation();
let (_, _, _, _, current_epoch_blending_token_collector, _, state_updater) =
current_recovery_checkpoint.into_components();
let (
_,
_,
_,
_,
pending_transactions,
current_epoch_blending_token_collector,
_,
state_updater,
) = current_recovery_checkpoint.into_components();
let new_reward_epoch_info = reward::EpochInfo::new(
new_epoch_info.epoch,
@@ -1127,6 +1269,7 @@ where
old_scheduler: Box::new(old_scheduler),
new_recovery_checkpoint: ServiceState::with_epoch(
new_epoch_info.epoch,
pending_transactions,
new_epoch_blending_token_collector,
Some(old_epoch_blending_token_collector),
state_updater,
@@ -1145,7 +1288,7 @@ where
current_cryptographic_processor.stop_proof_generation();
current_cryptographic_processor
};
let (_, _, _, _, current_epoch_blending_token_collector, _, _) =
let (_, _, _, _, _, current_epoch_blending_token_collector, _, _) =
current_recovery_checkpoint.into_components();
let new_reward_epoch_info = reward::EpochInfo::new(
epoch,
@@ -1260,17 +1403,18 @@ enum HandleEpochEventOutput<
},
}
/// Processes an already-serialized local data message from another service.
/// Processes a block proposal handed over by another service.
///
/// The serialized payload is encapsulated with blend layers. Before scheduling,
/// the outermost layers addressed to this node are self-decapsulated so that
/// blending tokens are collected immediately and only the remaining layers (or
/// the fully unwrapped message) are scheduled for the next release round.
#[expect(
clippy::cognitive_complexity,
reason = "TODO: address this in a dedicated refactor"
)]
async fn handle_serialized_local_data_message<
/// Leadership proofs are ready the moment the epoch's secret `PoL` info is, so
/// unlike a transaction this can be encapsulated where it arrives without
/// holding up the event loop.
///
/// `data_replication_factor` extra copies go out alongside the first. Block
/// proposals are replicated and transactions are not: the spec's per-solution
/// `PoW` quota `Q_W` is `ß_max`, i.e. exactly one message's worth of
/// encapsulations, whereas the leadership quota `Q_L` budgets for the extra
/// copies on top.
async fn handle_local_block_proposal<
NodeId,
Rng,
BackendSettings,
@@ -1279,7 +1423,8 @@ async fn handle_serialized_local_data_message<
ProofsVerifier,
CorePoQGenerator,
>(
serialized_local_data_message: &[u8],
proposal: &[u8],
data_replication_factor: u64,
cryptographic_processor: &mut CoreCryptographicProcessor<
NodeId,
CorePoQGenerator,
@@ -1300,18 +1445,68 @@ where
ProofsGenerator: CoreLeaderAndPowProofsGenerator<CorePoQGenerator>,
ProofsVerifier: ProofsVerifierTrait,
{
// TODO: Change this to encapsulated differently depending on the type of
// payload.
let Ok(wrapped_message) = cryptographic_processor
.encapsulate_block_proposal_payload(serialized_local_data_message)
.await
.inspect_err(|e| {
tracing::error!(target: LOG_TARGET, "Failed to wrap message: {e:?}");
})
else {
return current_recovery_checkpoint;
};
let mut recovery_checkpoint = current_recovery_checkpoint;
for _ in 0..data_replication_factor.strict_add(1) {
let Ok(wrapped_message) = cryptographic_processor
.encapsulate_block_proposal_payload(proposal)
.await
.inspect_err(|e| {
tracing::error!(target: LOG_TARGET, "Failed to wrap message: {e:?}");
})
else {
return recovery_checkpoint;
};
recovery_checkpoint = schedule_local_encapsulated_message(
&wrapped_message,
cryptographic_processor,
scheduler,
recovery_checkpoint,
);
}
recovery_checkpoint
}
/// Schedules a locally-generated, already-encapsulated data message for
/// release.
///
/// Before scheduling, the outermost layers addressed to this node are
/// self-decapsulated so that blending tokens are collected immediately and only
/// the remaining layers (or the fully unwrapped message) are scheduled for the
/// next release round.
#[expect(
clippy::cognitive_complexity,
reason = "TODO: address this in a dedicated refactor"
)]
fn schedule_local_encapsulated_message<
NodeId,
Rng,
BackendSettings,
NetworkSettings,
ProofsGenerator,
ProofsVerifier,
CorePoQGenerator,
>(
wrapped_message: &EncapsulatedMessageWithVerifiedPublicHeader,
cryptographic_processor: &CoreCryptographicProcessor<
NodeId,
CorePoQGenerator,
ProofsGenerator,
ProofsVerifier,
>,
scheduler: &mut EpochMessageScheduler<
Rng,
ProcessedMessage,
EncapsulatedMessageWithVerifiedPublicHeader,
>,
current_recovery_checkpoint: ServiceState<BackendSettings, NetworkSettings>,
) -> ServiceState<BackendSettings, NetworkSettings>
where
NodeId: Eq + Hash + Send + 'static,
Rng: RngCore + Clone + Send + Unpin,
BackendSettings: Clone + Send + Sync,
ProofsVerifier: ProofsVerifierTrait,
{
let mut state_updater = current_recovery_checkpoint.start_updating();
// Before blending the data message, we try to peel off any outer layers that
@@ -1341,11 +1536,19 @@ where
let processed_message = match remaining_message_type {
// If all the layers are peeled off locally, then we are left with the initial data message.
DecapsulatedMessageType::Completed(fully_decapsulated_message) => {
assert!(
fully_decapsulated_message.payload_type().is_data_message(),
"Locally-generated and fully-decapsulated message should be a data message."
);
let data_message: NetworkMessage = fully_decapsulated_message.payload_body().to_vec();
let data_message = match fully_decapsulated_message.into_components() {
(PayloadType::BlockProposal, encoded_block_proposal) => {
BlendPayload::BlockProposal(encoded_block_proposal)
}
(PayloadType::Transaction, encoded_transaction) => {
BlendPayload::Transaction(encoded_transaction)
}
(PayloadType::Cover, _) => {
panic!(
"Locally-generated and fully-decapsulated message should be a data message."
);
}
};
tracing::trace!(target: LOG_TARGET, "Locally generated data message of {} bytes had all the {} layers addressed to this same node. Propagating only the fully decapsulated message.", data_message.len(), blending_tokens.len());
ProcessedMessage::from(data_message)
}
@@ -1664,31 +1867,27 @@ where
match decapsulated_message_type {
DecapsulatedMessageType::Completed(fully_decapsulated_message) => {
match fully_decapsulated_message.into_components() {
let data_message = match fully_decapsulated_message.into_components() {
(PayloadType::BlockProposal, encoded_block_proposal) => {
BlendPayload::BlockProposal(encoded_block_proposal)
}
(PayloadType::Transaction, encoded_transaction) => {
BlendPayload::Transaction(encoded_transaction)
}
(PayloadType::Cover, _) => {
tracing::trace!(target: LOG_TARGET, "Discarding received cover message.");
(None, blending_tokens.into_iter())
return (None, blending_tokens.into_iter());
}
(PayloadType::BlockProposal, data_message) => {
tracing::trace!(
target: LOG_TARGET,
"Processing a fully decapsulated data message of {} bytes.",
data_message.len()
);
let processed_message = ProcessedMessage::from(data_message);
scheduler.schedule_processed_message(processed_message.clone());
(Some(processed_message), blending_tokens.into_iter())
}
// TODO: Submit the tx to the mempool.
(PayloadType::Transaction, transaction) => {
tracing::warn!(
target: LOG_TARGET,
"Discarding a fully decapsulated transaction message of {} bytes: mempool submission is not wired yet.",
transaction.len()
);
(None, blending_tokens.into_iter())
}
}
};
tracing::trace!(
target: LOG_TARGET,
"Processing a fully decapsulated {:?} message of {} bytes.",
data_message.payload_type(),
data_message.len()
);
let processed_message = ProcessedMessage::from(data_message);
scheduler.schedule_processed_message(processed_message.clone());
(Some(processed_message), blending_tokens.into_iter())
}
DecapsulatedMessageType::Incompleted(remaining_encapsulated_message) => {
tracing::trace!(
@@ -1723,7 +1922,7 @@ async fn handle_release_round<
NodeId,
Rng,
Backend,
NetAdapter,
Dispatcher,
ProofsGenerator,
ProofsVerifier,
CorePoQGenerator,
@@ -1741,16 +1940,16 @@ async fn handle_release_round<
>,
rng: &mut Rng,
backend: &Backend,
network_adapter: &NetAdapter,
current_recovery_checkpoint: ServiceState<Backend::Settings, NetAdapter::Settings>,
) -> ServiceState<Backend::Settings, NetAdapter::Settings>
payload_dispatcher: &Dispatcher,
current_recovery_checkpoint: ServiceState<Backend::Settings, Dispatcher::Settings>,
) -> ServiceState<Backend::Settings, Dispatcher::Settings>
where
NodeId: Eq + Hash + 'static,
Rng: RngCore + Send,
Backend: BlendBackend<NodeId, BlakeRng, ProofsVerifier, RuntimeServiceId> + Sync,
ProofsGenerator: CoreLeaderAndPowProofsGenerator<CorePoQGenerator>,
ProofsVerifier: ProofsVerifierTrait,
NetAdapter: NetworkAdapter<RuntimeServiceId> + Sync,
Dispatcher: PayloadDispatcher<RuntimeServiceId> + Sync,
{
let (processed_messages, should_generate_cover_message) =
release_type.map_or_else(|| (vec![], false), RoundReleaseType::into_components);
@@ -1779,7 +1978,7 @@ where
let processed_messages_relay_futures = build_futures_to_release_processed_messages(
processed_messages,
backend,
network_adapter,
payload_dispatcher,
Some(&mut state_updater),
current_epoch,
);
@@ -1823,7 +2022,7 @@ async fn handle_release_round_for_old_epoch<
NodeId,
Rng,
Backend,
NetAdapter,
Dispatcher,
ProofsVerifier,
RuntimeServiceId,
>(
@@ -1833,13 +2032,13 @@ async fn handle_release_round_for_old_epoch<
}: RoundInfo<ProcessedMessage, EncapsulatedMessageWithVerifiedPublicHeader>,
rng: &mut Rng,
backend: &Backend,
network_adapter: &NetAdapter,
payload_dispatcher: &Dispatcher,
epoch: Epoch,
) where
NodeId: Eq + Hash + 'static,
Rng: RngCore + Send,
Backend: BlendBackend<NodeId, BlakeRng, ProofsVerifier, RuntimeServiceId> + Sync,
NetAdapter: NetworkAdapter<RuntimeServiceId> + Sync,
Dispatcher: PayloadDispatcher<RuntimeServiceId> + Sync,
{
// The old epoch never generates cover traffic, so the cover flag is always
// `false` here.
@@ -1864,7 +2063,7 @@ async fn handle_release_round_for_old_epoch<
.chain(build_futures_to_release_processed_messages(
processed_messages,
backend,
network_adapter,
payload_dispatcher,
None,
epoch,
))
@@ -1908,20 +2107,20 @@ fn build_futures_to_release_processed_messages<
'fut,
NodeId,
Backend,
NetAdapter,
Dispatcher,
ProofsVerifier,
RuntimeServiceId,
>(
processed_messages_to_release: Vec<ProcessedMessage>,
backend: &'fut Backend,
network_adapter: &'fut NetAdapter,
mut state_updater: Option<&mut ServiceStateUpdater<Backend::Settings, NetAdapter::Settings>>,
payload_dispatcher: &'fut Dispatcher,
mut state_updater: Option<&mut ServiceStateUpdater<Backend::Settings, Dispatcher::Settings>>,
epoch: Epoch,
) -> Vec<BoxFuture<'fut, ()>>
where
NodeId: Eq + Hash + 'static,
Backend: BlendBackend<NodeId, BlakeRng, ProofsVerifier, RuntimeServiceId> + Sync,
NetAdapter: NetworkAdapter<RuntimeServiceId> + Sync,
Dispatcher: PayloadDispatcher<RuntimeServiceId> + Sync,
{
processed_messages_to_release
.into_iter()
@@ -1942,8 +2141,8 @@ where
.map(
|processed_message_to_release| -> BoxFuture<'fut, ()> {
match processed_message_to_release {
ProcessedMessage::Network(message) => {
network_adapter.broadcast(message).boxed()
ProcessedMessage::Decapsulated(payload) => {
payload_dispatcher.dispatch(payload).boxed()
}
ProcessedMessage::Encapsulated(encapsulated_message) => {
backend.publish(*encapsulated_message, epoch).boxed()
-59
View File
@@ -1,59 +0,0 @@
use lb_network_service::{
NetworkService,
backends::libp2p::{Command, Libp2p, PubSubCommand},
message::NetworkMsg,
};
use overwatch::services::{ServiceData, relay::OutboundRelay};
use serde::{Deserialize, Serialize};
use super::NetworkAdapter;
/// A network adapter for the network service that uses libp2p backend.
#[derive(Clone)]
pub struct Libp2pAdapter<RuntimeServiceId> {
network_relay:
OutboundRelay<<NetworkService<Libp2p, RuntimeServiceId> as ServiceData>::Message>,
settings: Libp2pBroadcastSettings,
}
/// Settings used to broadcast messages to the network service that uses libp2p
/// backend.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash)]
pub struct Libp2pBroadcastSettings {
pub topic: String,
}
#[async_trait::async_trait]
impl<RuntimeServiceId> NetworkAdapter<RuntimeServiceId> for Libp2pAdapter<RuntimeServiceId> {
type Backend = Libp2p;
type Settings = Libp2pBroadcastSettings;
fn new(
network_relay: OutboundRelay<
<NetworkService<Self::Backend, RuntimeServiceId> as ServiceData>::Message,
>,
settings: Self::Settings,
) -> Self {
Self {
network_relay,
settings,
}
}
/// Broadcast an unencrypted message to the network by publishing the
/// message under the configured gossipsub topic.
async fn broadcast(&self, message: Vec<u8>) {
if let Err((e, _)) = self
.network_relay
.send(NetworkMsg::Process(Command::PubSub(
PubSubCommand::Broadcast {
topic: self.settings.topic.clone(),
message: message.into_boxed_slice(),
},
)))
.await
{
tracing::error!("error broadcasting message: {e}");
}
}
}
-27
View File
@@ -1,27 +0,0 @@
pub mod libp2p;
use std::fmt::Debug;
use lb_network_service::{NetworkService, backends::NetworkBackend};
use overwatch::services::{ServiceData, relay::OutboundRelay};
use serde::{Serialize, de::DeserializeOwned};
/// A trait for communicating with the network service, which is used to
/// broadcast fully unwrapped messages returned from the blend backend.
#[async_trait::async_trait]
pub trait NetworkAdapter<RuntimeServiceId> {
/// The network backend used by the network service.
type Backend: NetworkBackend<RuntimeServiceId> + 'static;
/// Settings used to broadcast messages using the network service.
type Settings: Clone + Debug + Serialize + DeserializeOwned + Send + Sync + 'static;
fn new(
network_relay: OutboundRelay<
<NetworkService<Self::Backend, RuntimeServiceId> as ServiceData>::Message,
>,
settings: Self::Settings,
) -> Self;
/// Broadcast a message to the network service using the configured
/// settings.
async fn broadcast(&self, message: Vec<u8>);
}
+34 -16
View File
@@ -2,15 +2,15 @@ use lb_utils::blake_rng::BlakeRng;
use tokio::sync::oneshot;
use crate::{
core::{BlendService, backends::BlendBackend, network::NetworkAdapter},
message::ServiceMessage,
core::{BlendService, backends::BlendBackend, dispatcher::PayloadDispatcher},
message::{BlendPayload, NetworkInfo, ServiceMessage},
};
/// Helper trait to help the Blend proxy service rely on the concrete types of
/// the core Blend service without having to specify all the generics the core
/// service expects.
pub trait ServiceComponents<RuntimeServiceId> {
type NetworkAdapter: NetworkAdapter<RuntimeServiceId>;
type PayloadDispatcher: PayloadDispatcher<RuntimeServiceId>;
type BackendSettings;
type NodeId;
type Rng;
@@ -45,35 +45,39 @@ impl<
>
where
Backend: BlendBackend<NodeId, BlakeRng, ProofsVerifier, RuntimeServiceId>,
Network: NetworkAdapter<RuntimeServiceId>,
Network: PayloadDispatcher<RuntimeServiceId>,
StateStorage: lb_services_utils::overwatch::recovery::RecoveryBackend<
RuntimeServiceId,
State = crate::core::state::RecoveryServiceState<Backend::Settings, Network::Settings>,
> + Send
+ Sync,
{
type NetworkAdapter = Network;
type PayloadDispatcher = Network;
type BackendSettings = Backend::Settings;
type NodeId = NodeId;
type Rng = BlakeRng;
type ProofsGenerator = ProofsGenerator;
}
pub type NetworkBackendOfService<Service, RuntimeServiceId> = <<Service as ServiceComponents<
RuntimeServiceId,
>>::NetworkAdapter as NetworkAdapter<RuntimeServiceId>>::Backend;
pub type NetworkBackendOfService<Service, RuntimeServiceId> =
<<Service as ServiceComponents<RuntimeServiceId>>::PayloadDispatcher as PayloadDispatcher<
RuntimeServiceId,
>>::Backend;
pub type BlendBackendSettingsOfService<Service, RuntimeServiceId> =
<Service as ServiceComponents<RuntimeServiceId>>::BackendSettings;
/// The settings the core service's network adapter needs in order to
/// republish a message — deployment configuration, never carried in a payload.
pub type NetworkAdapterSettingsOfService<Service, RuntimeServiceId> =
<<Service as ServiceComponents<RuntimeServiceId>>::NetworkAdapter as NetworkAdapter<
/// The mempool service the core service's dispatcher hands transactions to.
pub type MempoolOfService<Service, RuntimeServiceId> = <<Service as ServiceComponents<
RuntimeServiceId,
>>::PayloadDispatcher as PayloadDispatcher<RuntimeServiceId>>::MempoolService;
pub type PayloadDispatcherSettingsOfService<Service, RuntimeServiceId> =
<<Service as ServiceComponents<RuntimeServiceId>>::PayloadDispatcher as PayloadDispatcher<
RuntimeServiceId,
>>::Settings;
use crate::message::NetworkInfo;
pub trait MessageComponents<NodeId> {
type Payload;
@@ -87,16 +91,23 @@ pub trait MessageComponents<NodeId> {
) -> Result<oneshot::Sender<Option<NetworkInfo<NodeId>>>, Self>
where
Self: Sized;
/// Try to extract a pending-transactions request from the message.
/// Returns `Ok(sender)` if the message is a pending-transactions request,
/// or `Err(self)` if it is not.
fn try_into_pending_transactions_request(self) -> Result<oneshot::Sender<Vec<Vec<u8>>>, Self>
where
Self: Sized;
}
impl<NodeId> MessageComponents<NodeId> for ServiceMessage<NodeId> {
type Payload = Vec<u8>;
type Payload = BlendPayload;
fn into_payload(self) -> Self::Payload {
match self {
Self::Blend(message) => message,
Self::GetNetworkInfo { .. } => {
panic!("NetworkInfo messages should be handled before calling into_payload")
Self::GetNetworkInfo { .. } | Self::GetPendingTransactions { .. } => {
panic!("Request messages should be handled before calling into_payload")
}
}
}
@@ -106,7 +117,14 @@ impl<NodeId> MessageComponents<NodeId> for ServiceMessage<NodeId> {
) -> Result<oneshot::Sender<Option<NetworkInfo<NodeId>>>, Self> {
match self {
Self::GetNetworkInfo { reply } => Ok(reply),
other @ Self::Blend(_) => Err(other),
other @ (Self::Blend(_) | Self::GetPendingTransactions { .. }) => Err(other),
}
}
fn try_into_pending_transactions_request(self) -> Result<oneshot::Sender<Vec<Vec<u8>>>, Self> {
match self {
Self::GetPendingTransactions { reply } => Ok(reply),
other @ (Self::Blend(_) | Self::GetNetworkInfo { .. }) => Err(other),
}
}
}
+64 -3
View File
@@ -1,5 +1,5 @@
mod serde {
use std::collections::HashSet;
use std::collections::{HashSet, VecDeque};
use lb_blend::message::{
encap::validated::EncapsulatedMessageWithVerifiedPublicHeader,
@@ -23,6 +23,7 @@ mod serde {
spent_core_quota: Quota,
unsent_processed_messages: HashSet<ProcessedMessage>,
unsent_data_messages: HashSet<EncapsulatedMessageWithVerifiedPublicHeader>,
pending_transactions: VecDeque<Vec<u8>>,
current_epoch_token_collector: EpochBlendingTokenCollector,
old_epoch_token_collector: Option<OldEpochBlendingTokenCollector>,
}
@@ -45,6 +46,7 @@ mod serde {
self.spent_core_quota,
self.unsent_processed_messages,
self.unsent_data_messages,
self.pending_transactions,
self.current_epoch_token_collector,
self.old_epoch_token_collector,
state_updater,
@@ -61,6 +63,7 @@ mod serde {
spent_core_quota,
unsent_processed_messages,
unsent_data_messages,
pending_transactions,
current_epoch_token_collector,
old_epoch_token_collector,
_,
@@ -70,6 +73,7 @@ mod serde {
spent_core_quota,
unsent_processed_messages,
unsent_data_messages,
pending_transactions,
current_epoch_token_collector,
old_epoch_token_collector,
}
@@ -80,7 +84,7 @@ mod serde {
pub use self::service::ServiceState;
mod service {
use core::fmt::{self, Debug, Formatter};
use std::collections::HashSet;
use std::collections::{HashSet, VecDeque};
use lb_blend::message::{
encap::validated::EncapsulatedMessageWithVerifiedPublicHeader,
@@ -103,6 +107,9 @@ mod service {
spent_core_quota: Quota,
unsent_processed_messages: HashSet<ProcessedMessage>,
unsent_data_messages: HashSet<EncapsulatedMessageWithVerifiedPublicHeader>,
/// Transactions handed over for blending that are still waiting for a
/// `PoW` solution to back their layer proofs.
pending_transactions: VecDeque<Vec<u8>>,
current_epoch_token_collector: EpochBlendingTokenCollector,
old_epoch_token_collector: Option<OldEpochBlendingTokenCollector>,
state_updater: overwatch::services::state::StateUpdater<
@@ -117,6 +124,7 @@ mod service {
spent_core_quota: self.spent_core_quota,
unsent_processed_messages: self.unsent_processed_messages.clone(),
unsent_data_messages: self.unsent_data_messages.clone(),
pending_transactions: self.pending_transactions.clone(),
current_epoch_token_collector: self.current_epoch_token_collector.clone(),
old_epoch_token_collector: self.old_epoch_token_collector.clone(),
state_updater: self.state_updater.clone(),
@@ -131,6 +139,7 @@ mod service {
.field("spent_core_quota", &self.spent_core_quota)
.field("unsent_processed_messages", &self.unsent_processed_messages)
.field("unsent_data_messages", &self.unsent_data_messages)
.field("pending_transactions", &self.pending_transactions.len())
.field(
"current_epoch_token_collector",
&self.current_epoch_token_collector,
@@ -146,11 +155,16 @@ mod service {
{
// Creates a new instance with the provided fields, and saves it using
// `state_updater`.
#[expect(
clippy::too_many_arguments,
reason = "One argument per persisted field."
)]
pub(super) fn new(
last_seen_epoch: Epoch,
spent_core_quota: Quota,
unsent_processed_messages: HashSet<ProcessedMessage>,
unsent_data_messages: HashSet<EncapsulatedMessageWithVerifiedPublicHeader>,
pending_transactions: VecDeque<Vec<u8>>,
current_epoch_token_collector: EpochBlendingTokenCollector,
old_epoch_token_collector: Option<OldEpochBlendingTokenCollector>,
state_updater: overwatch::services::state::StateUpdater<
@@ -182,6 +196,7 @@ mod service {
spent_core_quota,
unsent_processed_messages,
unsent_data_messages,
pending_transactions,
current_epoch_token_collector,
old_epoch_token_collector,
state_updater,
@@ -196,9 +211,13 @@ mod service {
/// The new instance is saved immediately using `state_updater`.
///
/// This is typically used on epoch rotations or when no previous
/// state was recovered.
/// state was recovered. `pending_transactions` is carried in
/// rather than emptied: a transaction that has not been encapsulated
/// yet is tied to no epoch, so an epoch rotation is no reason to lose
/// it.
pub fn with_epoch(
epoch: Epoch,
pending_transactions: VecDeque<Vec<u8>>,
current_epoch_token_collector: EpochBlendingTokenCollector,
old_epoch_token_collector: Option<OldEpochBlendingTokenCollector>,
state_updater: overwatch::services::state::StateUpdater<
@@ -210,6 +229,7 @@ mod service {
Quota::ZERO,
HashSet::new(),
HashSet::new(),
pending_transactions,
current_epoch_token_collector,
old_epoch_token_collector,
state_updater,
@@ -290,6 +310,7 @@ mod service {
Quota,
HashSet<ProcessedMessage>,
HashSet<EncapsulatedMessageWithVerifiedPublicHeader>,
VecDeque<Vec<u8>>,
EpochBlendingTokenCollector,
Option<OldEpochBlendingTokenCollector>,
overwatch::services::state::StateUpdater<
@@ -301,6 +322,7 @@ mod service {
self.spent_core_quota,
self.unsent_processed_messages,
self.unsent_data_messages,
self.pending_transactions,
self.current_epoch_token_collector,
self.old_epoch_token_collector,
self.state_updater,
@@ -361,6 +383,26 @@ mod service {
) -> &HashSet<EncapsulatedMessageWithVerifiedPublicHeader> {
&self.unsent_data_messages
}
pub(super) fn queue_pending_transaction(&mut self, transaction: Vec<u8>) {
self.pending_transactions.push_back(transaction);
}
pub(super) fn dequeue_transaction(&mut self, expected_transaction: &[u8]) {
assert_eq!(
self.pending_transactions.pop_front().as_deref(),
Some(expected_transaction),
"Expected transaction to be dequeued does not match the oldest pending transaction."
);
}
/// The transactions still waiting for a `PoW` solution, oldest first.
///
/// This is what a restarting service reads to refill the queue it works
/// from.
pub const fn pending_transactions(&self) -> &VecDeque<Vec<u8>> {
&self.pending_transactions
}
}
}
@@ -481,6 +523,25 @@ mod state_updater {
self.changed = true;
self.inner.remove_sent_data_message(message)
}
/// Record a transaction as waiting for a `PoW` solution to back its
/// layer proofs.
pub fn queue_unencapsulated_transaction(&mut self, transaction: Vec<u8>) {
self.changed = true;
self.inner.queue_pending_transaction(transaction);
}
/// Take the longest-waiting transaction off the queue, whether it went
/// on to be encapsulated or could not be.
///
/// `expected_transaction` is what the caller believes is at the head,
/// so that a drift between this queue and the one the event
/// loop works from is caught here rather than silently dropping
/// the wrong transaction.
pub fn dequeue_unencapsulated_transaction(&mut self, expected_transaction: &[u8]) {
self.changed = true;
self.inner.dequeue_transaction(expected_transaction);
}
}
impl<BackendSettings, NetworkSettings> StateUpdater<BackendSettings, NetworkSettings>
+257 -27
View File
@@ -1,4 +1,5 @@
mod utils;
use core::time::Duration;
use std::collections::VecDeque;
use futures::{StreamExt as _, stream::repeat};
use lb_blend::{
@@ -26,20 +27,26 @@ use crate::{
state::ServiceState,
tests::utils::{
MockKmsAdapter, MockProofsVerifier, NodeId, TestBlendBackend, TestBlendBackendEvent,
TestNetworkAdapter, backend_epoch_info, dummy_overwatch_resources,
TestPayloadDispatcher, backend_epoch_info, dummy_overwatch_resources,
dummy_pol_private_inputs, new_crypto_processor, new_epoch_info, new_membership,
new_stream, recorded_set_epoch_private_calls, recorded_stop_proof_generation_calls,
reset_set_epoch_private_calls, reset_stop_proof_generation_calls, reward_epoch_info,
scheduler_epoch_info, scheduler_settings, sdp_relay, settings, timing_settings,
wait_for_blend_backend_event,
new_stream, outgoing_messages_recorder, recorded_set_epoch_private_calls,
recorded_stop_proof_generation_calls, reset_set_epoch_private_calls,
reset_stop_proof_generation_calls, reward_epoch_info, scheduler_epoch_info,
scheduler_settings, sdp_relay, settings, timing_settings, wait_for_blend_backend_event,
},
},
epoch::{CoreEpochInfo, CoreEpochPublicInfo},
epoch_info::PolEpochInfo,
membership::{MembershipInfo, ZkInfo, chain::BlendEpochState},
test_utils::{crypto::MockCoreAndLeaderProofsGenerator, epoch::OncePolStreamProvider},
message::{BlendPayload, ServiceMessage},
test_utils::{
crypto::{GatedPowProofsGenerator, MockCoreAndLeaderProofsGenerator, PowGate},
epoch::OncePolStreamProvider,
},
};
mod utils;
type RuntimeServiceId = ();
fn test_blend_epoch_state(
@@ -99,6 +106,7 @@ async fn test_handle_incoming_blend_message() {
);
let recovery_checkpoint = ServiceState::with_epoch(
epoch,
VecDeque::new(),
EpochBlendingTokenCollector::new(&reward_epoch_info(&public_info)),
None,
state_updater,
@@ -135,7 +143,7 @@ async fn test_handle_incoming_blend_message() {
);
let (mut new_scheduler, mut scheduler) =
scheduler.rotate_epoch(scheduler_epoch_info(&public_info), scheduler_settings);
let (_, _, _, _, current_token_collector, _, state_updater) =
let (_, _, _, _, _, current_token_collector, _, state_updater) =
recovery_checkpoint.into_components();
let (new_token_collector, old_token_collector) =
current_token_collector.rotate_epoch(&reward_epoch_info(&public_info));
@@ -145,6 +153,7 @@ async fn test_handle_incoming_blend_message() {
// scheduler.
let recovery_checkpoint = ServiceState::with_epoch(
epoch,
VecDeque::new(),
new_token_collector,
Some(old_token_collector),
state_updater,
@@ -342,6 +351,7 @@ async fn test_duplicate_decapsulated_replica_handled_gracefully() {
);
let recovery_checkpoint = ServiceState::with_epoch(
epoch,
VecDeque::new(),
EpochBlendingTokenCollector::new(&reward_epoch_info(&public_info)),
None,
state_updater,
@@ -434,6 +444,7 @@ async fn test_handle_incoming_blend_message_with_invalid_poq() {
);
let recovery_checkpoint = ServiceState::with_epoch(
epoch_1,
VecDeque::new(),
EpochBlendingTokenCollector::new(&reward_epoch_info(&public_info_1)),
None,
state_updater,
@@ -589,7 +600,14 @@ async fn test_handle_epoch_event() {
crypto_processor,
scheduler,
public_info,
ServiceState::with_epoch(epoch, token_collector, None, state_updater.clone()).unwrap(),
ServiceState::with_epoch(
epoch,
VecDeque::new(),
token_collector,
None,
state_updater.clone(),
)
.unwrap(),
&mut backend,
&sdp_relay,
&mut None,
@@ -762,7 +780,14 @@ async fn test_handle_epoch_event_membership_change_rewires_backend_and_generator
crypto_processor,
scheduler,
public_info,
ServiceState::with_epoch(epoch, token_collector, None, state_updater.clone()).unwrap(),
ServiceState::with_epoch(
epoch,
VecDeque::new(),
token_collector,
None,
state_updater.clone(),
)
.unwrap(),
&mut backend,
&sdp_relay,
&mut None,
@@ -855,7 +880,14 @@ async fn transition_to_new_epoch_with_secret(secret_epoch: Epoch) -> Vec<Epoch>
crypto_processor,
scheduler,
public_info,
ServiceState::with_epoch(epoch, token_collector, None, state_updater.clone()).unwrap(),
ServiceState::with_epoch(
epoch,
VecDeque::new(),
token_collector,
None,
state_updater.clone(),
)
.unwrap(),
&mut backend,
&sdp_relay,
&mut Some(secret_info),
@@ -949,7 +981,14 @@ async fn test_handle_epoch_event_empty_epoch_retires() {
crypto_processor,
scheduler,
public_info.clone(),
ServiceState::with_epoch(epoch, token_collector, None, state_updater.clone()).unwrap(),
ServiceState::with_epoch(
epoch,
VecDeque::new(),
token_collector,
None,
state_updater.clone(),
)
.unwrap(),
&mut backend,
&sdp_relay,
&mut None,
@@ -1022,7 +1061,14 @@ async fn test_handle_epoch_event_non_empty_without_local_core_path_retires() {
crypto_processor,
scheduler,
public_info.clone(),
ServiceState::with_epoch(epoch, token_collector, None, state_updater.clone()).unwrap(),
ServiceState::with_epoch(
epoch,
VecDeque::new(),
token_collector,
None,
state_updater.clone(),
)
.unwrap(),
&mut backend,
&sdp_relay,
&mut None,
@@ -1089,13 +1135,14 @@ async fn complete_old_epoch_after_main_loop_done() {
current_public_info,
crypto_processor,
current_recovery_checkpoint,
pending_transactions,
message_scheduler,
mut backend,
mut rng,
) = initialize::<
NodeId,
TestBlendBackend,
TestNetworkAdapter,
TestPayloadDispatcher,
MockCoreAndLeaderProofsGenerator,
MockProofsVerifier,
MockKmsAdapter,
@@ -1129,10 +1176,11 @@ async fn complete_old_epoch_after_main_loop_done() {
&mut remaining_epoch_stream,
&settings_cloned,
&mut backend,
&TestNetworkAdapter,
&TestPayloadDispatcher,
&sdp_relay,
message_scheduler.into(),
&mut rng,
pending_transactions,
crypto_processor,
current_public_info,
current_recovery_checkpoint,
@@ -1143,7 +1191,7 @@ async fn complete_old_epoch_after_main_loop_done() {
blend_message_stream.map(|(msg, _)| msg),
remaining_epoch_stream,
backend,
TestNetworkAdapter,
TestPayloadDispatcher,
sdp_relay,
old_epoch_message_scheduler,
rng,
@@ -1236,13 +1284,14 @@ async fn stop_on_empty_epoch() {
current_public_info,
crypto_processor,
current_recovery_checkpoint,
pending_transactions,
message_scheduler,
mut backend,
mut rng,
) = initialize::<
NodeId,
TestBlendBackend,
TestNetworkAdapter,
TestPayloadDispatcher,
MockCoreAndLeaderProofsGenerator,
MockProofsVerifier,
MockKmsAdapter,
@@ -1276,10 +1325,11 @@ async fn stop_on_empty_epoch() {
&mut remaining_epoch_stream,
&settings_cloned,
&mut backend,
&TestNetworkAdapter,
&TestPayloadDispatcher,
&sdp_relay,
message_scheduler.into(),
&mut rng,
pending_transactions,
crypto_processor,
current_public_info,
current_recovery_checkpoint,
@@ -1290,7 +1340,7 @@ async fn stop_on_empty_epoch() {
blend_message_stream.map(|(msg, _)| msg),
remaining_epoch_stream,
backend,
TestNetworkAdapter,
TestPayloadDispatcher,
sdp_relay,
old_epoch_message_scheduler,
rng,
@@ -1372,13 +1422,14 @@ async fn stop_on_non_empty_epoch_without_local_core_path() {
current_public_info,
crypto_processor,
current_recovery_checkpoint,
pending_transactions,
message_scheduler,
mut backend,
mut rng,
) = initialize::<
NodeId,
TestBlendBackend,
TestNetworkAdapter,
TestPayloadDispatcher,
MockCoreAndLeaderProofsGenerator,
MockProofsVerifier,
MockKmsAdapter,
@@ -1412,10 +1463,11 @@ async fn stop_on_non_empty_epoch_without_local_core_path() {
&mut remaining_epoch_stream,
&settings_cloned,
&mut backend,
&TestNetworkAdapter,
&TestPayloadDispatcher,
&sdp_relay,
message_scheduler.into(),
&mut rng,
pending_transactions,
crypto_processor,
current_public_info,
current_recovery_checkpoint,
@@ -1426,7 +1478,7 @@ async fn stop_on_non_empty_epoch_without_local_core_path() {
blend_message_stream.map(|(msg, _)| msg),
remaining_epoch_stream,
backend,
TestNetworkAdapter,
TestPayloadDispatcher,
sdp_relay,
old_epoch_message_scheduler,
rng,
@@ -1524,6 +1576,7 @@ async fn test_proof_generator_epoch_binding() {
);
let recovery_checkpoint = ServiceState::with_epoch(
epoch_0,
VecDeque::new(),
EpochBlendingTokenCollector::new(&reward_epoch_info(&public_info_0)),
None,
state_updater,
@@ -1554,6 +1607,7 @@ async fn test_proof_generator_epoch_binding() {
);
let recovery_checkpoint = ServiceState::with_epoch(
epoch_0,
VecDeque::new(),
EpochBlendingTokenCollector::new(&reward_epoch_info(&public_info_0)),
None,
state_updater,
@@ -1586,6 +1640,7 @@ async fn test_proof_generator_epoch_binding() {
);
let recovery_checkpoint = ServiceState::with_epoch(
epoch_1,
VecDeque::new(),
EpochBlendingTokenCollector::new(&reward_epoch_info(&public_info_1)),
None,
state_updater,
@@ -1647,11 +1702,17 @@ async fn test_initialize_recovers_matching_saved_state() {
// Build a pre-populated saved state with matching epoch and some spent quota.
let public_info = new_epoch_info(initial_epoch, membership.clone(), &settings);
let token_collector = EpochBlendingTokenCollector::new(&reward_epoch_info(&public_info));
let saved_state =
ServiceState::with_epoch(initial_epoch, token_collector, None, state_updater.clone())
.unwrap();
let saved_state = ServiceState::with_epoch(
initial_epoch,
VecDeque::new(),
token_collector,
None,
state_updater.clone(),
)
.unwrap();
let mut updater = saved_state.start_updating();
updater.consume_core_quota(Quota::new::<5>());
updater.queue_unencapsulated_transaction(b"transaction".to_vec());
let saved_state = updater.commit_changes();
let (
@@ -1659,13 +1720,14 @@ async fn test_initialize_recovers_matching_saved_state() {
_current_public_info,
_crypto_processor,
recovered_checkpoint,
_pending_transactions,
_message_scheduler,
_backend,
_rng,
) = initialize::<
NodeId,
TestBlendBackend,
TestNetworkAdapter,
TestPayloadDispatcher,
MockCoreAndLeaderProofsGenerator,
MockProofsVerifier,
MockKmsAdapter,
@@ -1687,6 +1749,13 @@ async fn test_initialize_recovers_matching_saved_state() {
"Matching epoch: spent_quota should be restored from saved state"
);
assert_eq!(recovered_checkpoint.last_seen_epoch(), initial_epoch);
// A transaction still waiting for a `PoW` solution has not been encapsulated
// and so belongs to no epoch: a restart must not lose it.
assert_eq!(
recovered_checkpoint.pending_transactions().front(),
Some(&b"transaction".to_vec()),
"Matching epoch: a queued transaction should be restored from saved state"
);
// Mismatched epoch: fresh state should be created
@@ -1715,6 +1784,7 @@ async fn test_initialize_recovers_matching_saved_state() {
EpochBlendingTokenCollector::new(&reward_epoch_info(&stale_public_info));
let stale_state = ServiceState::with_epoch(
99.into(),
VecDeque::new(),
stale_token_collector,
None,
state_updater2.clone(),
@@ -1722,6 +1792,7 @@ async fn test_initialize_recovers_matching_saved_state() {
.unwrap();
let mut updater = stale_state.start_updating();
updater.consume_core_quota(Quota::new::<42>());
updater.queue_unencapsulated_transaction(b"stale epoch transaction".to_vec());
let stale_state = updater.commit_changes();
let (
@@ -1729,13 +1800,14 @@ async fn test_initialize_recovers_matching_saved_state() {
_current_public_info2,
_crypto_processor2,
recovered_checkpoint2,
pending_transactions2,
_message_scheduler2,
_backend2,
_rng2,
) = initialize::<
NodeId,
TestBlendBackend,
TestNetworkAdapter,
TestPayloadDispatcher,
MockCoreAndLeaderProofsGenerator,
MockProofsVerifier,
MockKmsAdapter,
@@ -1761,4 +1833,162 @@ async fn test_initialize_recovers_matching_saved_state() {
initial_epoch,
"Mismatched epoch: should track the current epoch, not the stale one"
);
// The rest of a stale state belongs to the epoch it was saved under, but a
// transaction still waiting for a `PoW` solution has not been encapsulated
// and so belongs to none.
assert_eq!(
recovered_checkpoint2.pending_transactions().front(),
Some(&b"stale epoch transaction".to_vec()),
"Mismatched epoch: a queued transaction should outlive the state that carried it"
);
assert_eq!(
pending_transactions2.front(),
Some(&b"stale epoch transaction".to_vec()),
"Mismatched epoch: the queue handed to the event loop should carry it too"
);
}
/// A transaction waits for a `PoW` solution without holding up anything else.
///
/// The puzzle search behind a transaction's layer proofs takes long enough that
/// awaiting it where the transaction arrives would stop the service dead: no
/// incoming messages, no release rounds, no epoch events, no cover traffic. So
/// the transaction is queued and mined for on its own branch of the event loop,
/// and this test pins that down by getting a block proposal all the way out
/// while the transaction is still waiting for its solution.
#[test_log::test(tokio::test)]
async fn a_transaction_awaiting_a_pow_solution_does_not_stall_the_event_loop() {
let minimal_network_size = 2;
let (membership, local_private_key) = new_membership(minimal_network_size);
let mut settings = settings(
local_private_key.clone(),
u64::from(minimal_network_size).try_into().unwrap(),
(),
0,
);
// No cover traffic, so every message the service sends is one this test put
// in and the count below means what it says.
settings.scheduler.cover.message_frequency_per_round = 0.0.try_into().unwrap();
let (inbound_relay, inbound_message_sender) = new_stream();
let (mut blend_message_stream, _blend_message_sender) = new_stream();
let (membership_stream, membership_sender) = new_stream();
let membership_info = MembershipInfo {
membership,
zk: Some(ZkInfo {
root: ZkHash::ZERO,
core_and_path_selectors: Some([(ZkHash::ZERO, false); CORE_MERKLE_TREE_HEIGHT]),
}),
};
membership_sender
.send(test_blend_epoch_state(0, membership_info))
.await
.unwrap();
let (sdp_relay, _sdp_relay_receiver) = sdp_relay();
let (overwatch_handle, _overwatch_cmd_receiver, state_updater, _state_receiver) =
dummy_overwatch_resources();
// Both installed before the service exists, so nothing is missed.
let pow_gate = PowGate::setup();
let mut outgoing_messages = outgoing_messages_recorder();
let (
mut remaining_epoch_stream,
current_public_info,
crypto_processor,
current_recovery_checkpoint,
pending_transactions,
message_scheduler,
mut backend,
mut rng,
) = initialize::<
NodeId,
TestBlendBackend,
TestPayloadDispatcher,
GatedPowProofsGenerator,
MockProofsVerifier,
MockKmsAdapter,
RuntimeServiceId,
>(
settings.clone(),
membership_stream,
overwatch_handle.clone(),
MockKmsAdapter,
&sdp_relay,
None,
state_updater,
)
.await;
tokio::spawn(async move {
let secret_pol_info_stream =
post_initialize::<OncePolStreamProvider, RuntimeServiceId>(&overwatch_handle).await;
run_event_loop(
inbound_relay,
&mut blend_message_stream,
secret_pol_info_stream,
&mut remaining_epoch_stream,
&settings,
&mut backend,
&TestPayloadDispatcher,
&sdp_relay,
message_scheduler.into(),
&mut rng,
pending_transactions,
crypto_processor,
current_public_info,
current_recovery_checkpoint,
)
.await;
});
// The transaction goes in first, so if the loop blocked on mining, the
// proposal queued behind it could never get out.
inbound_message_sender
.send(ServiceMessage::Blend(BlendPayload::Transaction(
b"transaction".to_vec(),
)))
.await
.unwrap();
inbound_message_sender
.send(ServiceMessage::Blend(BlendPayload::BlockProposal(
b"proposal".to_vec(),
)))
.await
.unwrap();
// The proposal gets out while the transaction is still mining. Nothing else
// can be in flight: cover traffic is off and the gate is shut. A stalled
// loop shows up as a timeout here rather than as a hung test.
expect_outgoing_message(
&mut outgoing_messages,
"the block proposal should be sent while the transaction is still mining",
)
.await;
// And the transaction follows once its solution lands.
pow_gate.release();
expect_outgoing_message(
&mut outgoing_messages,
"the transaction should be sent once its PoW solution lands",
)
.await;
}
/// Waits for the service to send one message onwards, failing rather than
/// hanging if it never does.
async fn expect_outgoing_message(
outgoing_messages: &mut tokio::sync::mpsc::UnboundedReceiver<()>,
expectation: &str,
) {
// Generous next to the release round the message has to wait for, and only
// ever reached when the assertion has already failed.
const GRACE: Duration = Duration::from_secs(10);
tokio::time::timeout(GRACE, outgoing_messages.recv())
.await
.unwrap_or_else(|_| panic!("timed out: {expectation}"))
.unwrap_or_else(|| panic!("service stopped sending: {expectation}"));
}
+37 -5
View File
@@ -51,8 +51,8 @@ use tokio_stream::wrappers::{BroadcastStream, ReceiverStream};
use crate::{
core::{
backends::{BackendEpochInfo, BlendBackend},
dispatcher::PayloadDispatcher,
kms::KmsPoQAdapter,
network::NetworkAdapter,
processor::CoreCryptographicProcessor,
settings::{
CoverTrafficSettings, MessageDelayerSettings, RunningBlendConfig as BlendConfig,
@@ -62,9 +62,10 @@ use crate::{
tests::RuntimeServiceId,
},
epoch::CoreEpochPublicInfo,
message::NetworkInfo,
message::{BlendPayload, NetworkInfo},
settings::TimingSettings,
test_utils,
test_utils::mempool::TestMempoolService,
};
pub type NodeId = [u8; 32];
@@ -168,6 +169,7 @@ where
_msg: EncapsulatedMessageWithVerifiedPublicHeader,
_intended_epoch: Epoch,
) {
note_outgoing_message();
}
async fn rotate_epoch(&mut self, new_epoch_info: BackendEpochInfo<NodeId, ProofsVerifier>) {
@@ -240,23 +242,53 @@ pub async fn wait_for_blend_backend_event(
}
}
pub struct TestNetworkAdapter;
thread_local! {
/// Installed by [`record_outgoing_messages`] for the duration of a test.
static OUTGOING_MESSAGES: RefCell<Option<mpsc::UnboundedSender<()>>> =
const { RefCell::new(None) };
}
/// Starts recording every message the service sends onwards, whether it goes
/// to a Blend peer through the backend or to a local service through the
/// dispatcher.
pub fn outgoing_messages_recorder() -> mpsc::UnboundedReceiver<()> {
let (sender, receiver) = mpsc::unbounded_channel();
OUTGOING_MESSAGES.with_borrow_mut(|recorder| *recorder = Some(sender));
receiver
}
fn note_outgoing_message() {
OUTGOING_MESSAGES.with_borrow(|recorder| {
if let Some(sender) = recorder.as_ref() {
let _ = sender.send(());
}
});
}
pub struct TestPayloadDispatcher;
#[async_trait]
impl<RuntimeServiceId> NetworkAdapter<RuntimeServiceId> for TestNetworkAdapter {
impl<RuntimeServiceId> PayloadDispatcher<RuntimeServiceId> for TestPayloadDispatcher
where
RuntimeServiceId: Send + 'static,
{
type Backend = TestNetworkBackend;
type MempoolService = TestMempoolService<RuntimeServiceId>;
type Settings = ();
fn new(
_network_relay: OutboundRelay<
<NetworkService<Self::Backend, RuntimeServiceId> as ServiceData>::Message,
>,
_mempool_relay: OutboundRelay<<Self::MempoolService as ServiceData>::Message>,
_settings: Self::Settings,
) -> Self {
Self
}
async fn broadcast(&self, _message: Vec<u8>) {}
async fn dispatch(&self, _payload: BlendPayload) {
note_outgoing_message();
}
}
pub struct TestNetworkBackend {
+28 -8
View File
@@ -6,7 +6,7 @@ use lb_blend::{
membership::Membership,
message_blend::{
crypto::leader::send::EpochCryptographicProcessor,
provers::{WinningPolInfoStream, leader::LeaderProofsGenerator},
provers::{WinningPolInfoStream, leader_and_pow::LeaderAndPowProofsGenerator},
},
},
};
@@ -28,7 +28,7 @@ impl<Backend, NodeId, ProofsGenerator, RuntimeServiceId>
where
Backend: BlendBackend<NodeId, RuntimeServiceId>,
NodeId: Clone,
ProofsGenerator: LeaderProofsGenerator,
ProofsGenerator: LeaderAndPowProofsGenerator,
{
#[cfg(test)]
pub const fn epoch(&self) -> Epoch {
@@ -41,7 +41,7 @@ impl<Backend, NodeId, ProofsGenerator, RuntimeServiceId>
where
Backend: BlendBackend<NodeId, RuntimeServiceId>,
NodeId: Clone + Send + 'static,
ProofsGenerator: LeaderProofsGenerator,
ProofsGenerator: LeaderAndPowProofsGenerator,
{
/// Creates a [`MessageHandler`] with the given membership.
///
@@ -112,22 +112,42 @@ impl<Backend, NodeId, ProofsGenerator, RuntimeServiceId>
where
NodeId: Eq + Hash + Clone + Send + 'static,
Backend: BlendBackend<NodeId, RuntimeServiceId> + Sync,
ProofsGenerator: LeaderProofsGenerator,
ProofsGenerator: LeaderAndPowProofsGenerator,
{
/// Blend a new message received from another service.
pub async fn handle_message_to_blend(&mut self, message: Vec<u8>) {
/// Blend a block proposal, spending leadership quota on its layer proofs.
pub async fn handle_block_proposal_to_blend(&mut self, proposal: &[u8]) {
let Ok(message) = self
.cryptographic_processor
.encapsulate_block_proposal_payload(&message)
.encapsulate_block_proposal_payload(proposal)
.await
.inspect_err(|e| {
tracing::error!(target: LOG_TARGET, "Failed to encapsulate message: {e:?}");
tracing::error!(target: LOG_TARGET, "Failed to encapsulate block proposal: {e:?}");
})
else {
return;
};
self.backend.send(message).await;
}
/// Blend a transaction, whose layer proofs are backed by a proof of work.
///
/// Unlike a block proposal this cannot be answered on demand: the proofs
/// come from a puzzle search, so the caller has to be somewhere it can
/// afford to wait.
/// Returns `None` if the transaction could not be encapsulated, so the
/// caller can leave it queued and try again rather than losing it.
pub async fn handle_transaction_to_blend(&mut self, transaction: &[u8]) -> Option<()> {
let message = self
.cryptographic_processor
.encapsulate_transaction_payload(transaction)
.await
.inspect_err(|e| {
tracing::error!(target: LOG_TARGET, "Failed to encapsulate transaction: {e:?}");
})
.ok()?;
self.backend.send(message).await;
Some(())
}
}
#[derive(Debug, thiserror::Error)]
+72 -27
View File
@@ -6,6 +6,7 @@ pub mod settings;
mod tests;
use std::{
collections::VecDeque,
fmt::{Debug, Display},
hash::Hash,
marker::PhantomData,
@@ -19,7 +20,7 @@ use lb_blend::{
proofs::quota::inputs::prove::public::{CoreInputs, LeaderInputs, PowInputs},
scheduling::{
epoch::{EpochEvent, UninitializedEpochEventStream},
message_blend::provers::leader::LeaderProofsGenerator,
message_blend::provers::leader_and_pow::LeaderAndPowProofsGenerator,
},
};
use lb_chain_service::api::CryptarchiaServiceData;
@@ -51,7 +52,7 @@ use crate::{
epoch_info::{PolEpochInfo, PolInfoProvider as PolInfoProviderTrait},
kms::PreloadKmsService,
membership::{self, chain::BlendEpochState, node_id},
message::{NetworkInfo, ServiceMessage},
message::{BlendPayload, NetworkInfo, ServiceMessage},
};
const LOG_TARGET: &str = blend::service::EDGE;
@@ -111,7 +112,7 @@ impl<Backend, NodeId, ProofsGenerator, TimeBackend, ChainService, PolInfoProvide
where
Backend: BlendBackend<NodeId, RuntimeServiceId> + Send + Sync,
NodeId: Clone + Debug + Eq + Hash + Send + Sync + node_id::TryFrom + 'static,
ProofsGenerator: LeaderProofsGenerator + Send,
ProofsGenerator: LeaderAndPowProofsGenerator + Send,
TimeBackend: lb_time_service::backends::TimeBackend + Send,
ChainService: CryptarchiaServiceData<Tx: Send + Sync>,
PolInfoProvider: PolInfoProviderTrait<RuntimeServiceId, Stream: Send + Unpin + 'static> + Send,
@@ -191,23 +192,13 @@ where
)
.await;
let messages_to_blend_stream = Box::pin(inbound_relay.filter_map(async |msg| match msg {
ServiceMessage::Blend(message) => Some(message),
ServiceMessage::GetNetworkInfo { reply } => {
drop(reply.send(Some(NetworkInfo {
node_id: local_node_id.clone(),
core_info: None,
})));
None
}
}));
run::<Backend, _, ProofsGenerator, PolInfoProvider, _>(
UninitializedEpochEventStream::new(
public_epoch_stream,
settings.time.epoch_transition_period,
),
messages_to_blend_stream,
Box::pin(inbound_relay),
local_node_id,
RunningSettings::<Backend, _, _> {
backend: settings.backend,
cover: settings.cover,
@@ -263,7 +254,8 @@ async fn run<Backend, NodeId, ProofsGenerator, PolInfoProvider, RuntimeServiceId
public_epoch_stream: UninitializedEpochEventStream<
impl Stream<Item = BlendEpochState<NodeId>> + Unpin,
>,
mut incoming_message_stream: impl Stream<Item = Vec<u8>> + Send + Unpin,
mut inbound_relay: impl Stream<Item = ServiceMessage<NodeId>> + Send + Unpin,
local_node_id: NodeId,
settings: RunningSettings<Backend, NodeId, RuntimeServiceId>,
overwatch_handle: &OverwatchHandle<RuntimeServiceId>,
notify_ready: impl Fn(),
@@ -271,7 +263,7 @@ async fn run<Backend, NodeId, ProofsGenerator, PolInfoProvider, RuntimeServiceId
where
Backend: BlendBackend<NodeId, RuntimeServiceId> + Sync + Send,
NodeId: Clone + Debug + Eq + Hash + Send + Sync + 'static,
ProofsGenerator: LeaderProofsGenerator + Send,
ProofsGenerator: LeaderAndPowProofsGenerator + Send,
PolInfoProvider: PolInfoProviderTrait<RuntimeServiceId, Stream: Unpin>,
RuntimeServiceId: Clone + Send + Sync,
{
@@ -298,6 +290,8 @@ where
.expect("Should not fail to subscribe to secret PoL info stream.");
let mut current_secret_epoch_info: Option<PolEpochInfo> = None;
// Transactions waiting for a `PoW` solution to back their layer proofs.
let mut pending_transactions: VecDeque<Vec<u8>> = VecDeque::new();
let mut current_epoch_message_handler: Option<
MessageHandler<Backend, NodeId, ProofsGenerator, RuntimeServiceId>,
> = None;
@@ -332,17 +326,38 @@ where
Ok(()) => {}
}
}
Some(message) = incoming_message_stream.next() => {
// TODO: Investigate why secret PoL info at times arrives after the block proposal.
let Some(handler) = current_epoch_message_handler.as_mut() else {
tracing::warn!(target: LOG_TARGET, "Received a message to blend, but no active message handler is available to process it because the secret PoL info for the current epoch is not yet available. Ignoring the message.");
continue;
};
let message_copies = settings.data_replication_factor.checked_add(1).unwrap();
for _ in 0..message_copies {
handler.handle_message_to_blend(message.clone()).await;
Some(message) = inbound_relay.next() => {
match message {
ServiceMessage::Blend(BlendPayload::Transaction(transaction)) => {
pending_transactions.push_back(transaction);
}
ServiceMessage::Blend(BlendPayload::BlockProposal(proposal)) => {
// TODO: Investigate why secret PoL info at times arrives after the block proposal.
let Some(handler) = current_epoch_message_handler.as_mut() else {
tracing::warn!(target: LOG_TARGET, "Received a message to blend, but no active message handler is available to process it because the secret PoL info for the current epoch is not yet available. Ignoring the message.");
continue;
};
let message_copies = settings.data_replication_factor.checked_add(1).expect("Data replication factor should not overflow when incremented.");
for _ in 0..message_copies {
handler.handle_block_proposal_to_blend(&proposal).await;
}
}
ServiceMessage::GetNetworkInfo { reply } => {
drop(reply.send(Some(NetworkInfo {
node_id: local_node_id.clone(),
core_info: None,
})));
}
ServiceMessage::GetPendingTransactions { reply } => {
drop(reply.send(pending_transactions.iter().cloned().collect()));
}
}
}
// A queued transaction leaves as soon as a `PoW` solution backs it, awaited
// here as one branch among the others so the loop keeps turning meanwhile.
Some(()) = blend_next_transaction(&pending_transactions, &mut current_epoch_message_handler) => {
drop(pending_transactions.pop_front());
}
else => {
// All input streams have terminated (e.g. disorderly shutdown).
// Exit cleanly instead of letting `select!` panic.
@@ -353,6 +368,36 @@ where
}
}
/// Blends the transaction that has been waiting longest, once a `PoW` solution
/// backs its layer proofs.
///
/// The transaction is only read here, never taken off the queue: `select!`
/// drops this future whenever another branch wins the race, and one that popped
/// before awaiting would take the transaction down with it every time that
/// happened. It comes off the queue in the branch handler instead, which runs
/// once the race is settled.
///
/// Returns `None` when there is nothing to blend — no transaction queued, or no
/// handler for this epoch yet — which is what leaves the `select!` branch free
/// to wait on the others.
async fn blend_next_transaction<Backend, NodeId, ProofsGenerator, RuntimeServiceId>(
pending_transactions: &VecDeque<Vec<u8>>,
current_epoch_message_handler: &mut Option<
MessageHandler<Backend, NodeId, ProofsGenerator, RuntimeServiceId>,
>,
) -> Option<()>
where
Backend: BlendBackend<NodeId, RuntimeServiceId> + Sync,
NodeId: Clone + Debug + Eq + Hash + Send + Sync + 'static,
ProofsGenerator: LeaderAndPowProofsGenerator + Send,
{
let transaction = pending_transactions.front()?;
current_epoch_message_handler
.as_mut()?
.handle_transaction_to_blend(transaction)
.await
}
fn handle_new_epoch_event<Backend, NodeId, ProofsGenerator, RuntimeServiceId>(
current_public_epoch_info: &BlendEpochState<NodeId>,
maybe_current_secret_epoch_info: &mut Option<PolEpochInfo>,
@@ -365,7 +410,7 @@ fn handle_new_epoch_event<Backend, NodeId, ProofsGenerator, RuntimeServiceId>(
where
Backend: BlendBackend<NodeId, RuntimeServiceId>,
NodeId: Clone + Send + Eq + Hash + 'static,
ProofsGenerator: LeaderProofsGenerator,
ProofsGenerator: LeaderAndPowProofsGenerator,
{
// Whatever happens on a new epoch, we shut down the previous handler.
// It will be rebuilt below if the current public and secret info line up
+38 -2
View File
@@ -19,6 +19,7 @@ use crate::{
},
epoch_info::PolEpochInfo,
membership::chain::BlendEpochState,
message::BlendPayload,
test_utils::membership::membership,
};
@@ -39,7 +40,10 @@ async fn run_with_epoch_transition() {
.await;
// A message should be forwarded to the core node 0.
msg_sender.send(vec![0]).await.expect("channel opened");
msg_sender
.send(BlendPayload::BlockProposal(vec![0]).into())
.await
.expect("channel opened");
assert_eq!(
node_id_receiver.recv().await.expect("channel opened"),
core_node
@@ -54,7 +58,39 @@ async fn run_with_epoch_transition() {
sleep(Duration::from_millis(100)).await;
// A message should be forwarded to the core node 1.
msg_sender.send(vec![0]).await.expect("channel opened");
msg_sender
.send(BlendPayload::BlockProposal(vec![0]).into())
.await
.expect("channel opened");
assert_eq!(
node_id_receiver.recv().await.expect("channel opened"),
core_node
);
}
/// [`run`] blends a transaction, drawing its layer proofs from the `PoW` branch
/// rather than from leadership quota.
///
/// Unlike a block proposal, a transaction that arrives before the epoch's
/// secret `PoL` info does is not dropped: it waits in the queue until there is
/// a message handler to encapsulate it, which is the same queue that keeps the
/// puzzle search off the event loop.
#[test_log::test(tokio::test)]
async fn run_blends_a_transaction() {
let local_node = NodeId(99);
let core_node = NodeId(0);
let minimal_network_size = 1;
let (_, _epoch_sender, msg_sender, mut node_id_receiver) = spawn_run(
local_node,
minimal_network_size,
Some(membership(&[core_node], local_node)),
)
.await;
msg_sender
.send(BlendPayload::Transaction(vec![0]).into())
.await
.expect("channel opened");
assert_eq!(
node_id_receiver.recv().await.expect("channel opened"),
core_node
+10 -4
View File
@@ -10,7 +10,7 @@ use lb_blend::{
membership::Membership,
message_blend::provers::{
BlendLayerProof, ProofsGeneratorSettings, WinningPolInfoStream,
leader::LeaderProofsGenerator,
leader_and_pow::LeaderAndPowProofsGenerator,
},
},
};
@@ -26,6 +26,7 @@ use crate::{
backends::BlendBackend, handlers::Error, run, settings::RunningBlendConfig as BlendConfig,
tests::test_blend_epoch_state,
},
message::ServiceMessage,
settings::TimingSettings,
test_utils::{crypto::mock_blend_proof, epoch::OncePolStreamProvider, membership::key},
};
@@ -33,7 +34,7 @@ use crate::{
pub struct MockLeaderProofsGenerator;
#[async_trait]
impl LeaderProofsGenerator for MockLeaderProofsGenerator {
impl LeaderAndPowProofsGenerator for MockLeaderProofsGenerator {
fn new(
_settings: ProofsGeneratorSettings,
_winning_pol_info_stream: WinningPolInfoStream,
@@ -41,7 +42,11 @@ impl LeaderProofsGenerator for MockLeaderProofsGenerator {
Self
}
async fn get_next_proof(&mut self) -> Option<BlendLayerProof> {
async fn get_next_leader_proof(&mut self) -> Option<BlendLayerProof> {
Some(mock_blend_proof())
}
async fn get_next_pow_proof(&mut self) -> Option<BlendLayerProof> {
Some(mock_blend_proof())
}
}
@@ -53,7 +58,7 @@ pub async fn spawn_run(
) -> (
JoinHandle<Result<(), Error>>,
mpsc::Sender<Membership<NodeId>>,
mpsc::Sender<Vec<u8>>,
mpsc::Sender<ServiceMessage<NodeId>>,
mpsc::Receiver<NodeId>,
) {
let (epoch_sender, epoch_receiver) = mpsc::channel(1);
@@ -81,6 +86,7 @@ pub async fn spawn_run(
>(
UninitializedEpochEventStream::new(epoch_stream, Duration::ZERO),
ReceiverStream::new(msg_receiver),
local_node,
settings,
&overwatch_handle(),
|| {},
+23 -16
View File
@@ -12,13 +12,14 @@ use overwatch::{
use crate::{
core::{
network::NetworkAdapter as NetworkAdapterTrait,
dispatcher::PayloadDispatcher as PayloadDispatcherTrait,
service_components::{
MessageComponents, NetworkAdapterSettingsOfService, NetworkBackendOfService,
ServiceComponents as CoreServiceComponents,
MempoolOfService, MessageComponents, NetworkBackendOfService,
PayloadDispatcherSettingsOfService, ServiceComponents as CoreServiceComponents,
},
},
membership::MembershipInfo,
message::BlendPayload,
modes::{self, BroadcastMode, CoreMode, EdgeMode},
};
@@ -36,9 +37,9 @@ where
// Keep the previous core mode for the epoch transition period.
prev: CoreMode<CoreService, RuntimeServiceId>,
},
Broadcast(BroadcastMode<CoreService::NetworkAdapter, CoreService::NodeId, RuntimeServiceId>),
Broadcast(BroadcastMode<CoreService::PayloadDispatcher, CoreService::NodeId, RuntimeServiceId>),
BroadcastAfterCore {
mode: BroadcastMode<CoreService::NetworkAdapter, CoreService::NodeId, RuntimeServiceId>,
mode: BroadcastMode<CoreService::PayloadDispatcher, CoreService::NodeId, RuntimeServiceId>,
// Keep the previous core mode for the epoch transition period.
prev: CoreMode<CoreService, RuntimeServiceId>,
},
@@ -48,13 +49,13 @@ impl<CoreService, EdgeService, RuntimeServiceId>
Instance<CoreService, EdgeService, RuntimeServiceId>
where
CoreService: ServiceData<
Message: MessageComponents<CoreService::NodeId, Payload: Into<Vec<u8>>>
Message: MessageComponents<CoreService::NodeId, Payload: Into<BlendPayload>>
+ Send
+ Sync
+ 'static,
> + CoreServiceComponents<
RuntimeServiceId,
NetworkAdapter: NetworkAdapterTrait<RuntimeServiceId> + Send + Sync + 'static,
PayloadDispatcher: PayloadDispatcherTrait<RuntimeServiceId> + Send + Sync + 'static,
NodeId: Clone + Eq + Hash + Send + Sync,
> + 'static,
EdgeService: ServiceData<Message = CoreService::Message> + 'static,
@@ -65,7 +66,8 @@ where
NetworkBackendOfService<CoreService, RuntimeServiceId>,
RuntimeServiceId,
>,
> + Debug
> + AsServiceId<MempoolOfService<CoreService, RuntimeServiceId>>
+ Debug
+ Display
+ Clone
+ Send
@@ -77,7 +79,7 @@ where
mode: Mode,
local_node_id: CoreService::NodeId,
overwatch_handle: &OverwatchHandle<RuntimeServiceId>,
network_settings: NetworkAdapterSettingsOfService<CoreService, RuntimeServiceId>,
network_settings: PayloadDispatcherSettingsOfService<CoreService, RuntimeServiceId>,
) -> Result<Self, modes::Error> {
match mode {
Mode::Core => Ok(Self::Core(Self::new_core_mode(overwatch_handle).await?)),
@@ -103,9 +105,9 @@ where
async fn new_broadcast_mode(
overwatch_handle: &OverwatchHandle<RuntimeServiceId>,
local_node_id: CoreService::NodeId,
network_settings: NetworkAdapterSettingsOfService<CoreService, RuntimeServiceId>,
network_settings: PayloadDispatcherSettingsOfService<CoreService, RuntimeServiceId>,
) -> Result<
BroadcastMode<CoreService::NetworkAdapter, CoreService::NodeId, RuntimeServiceId>,
BroadcastMode<CoreService::PayloadDispatcher, CoreService::NodeId, RuntimeServiceId>,
modes::Error,
> {
BroadcastMode::new::<
@@ -140,7 +142,7 @@ where
overwatch_handle: &OverwatchHandle<RuntimeServiceId>,
minimal_network_size: usize,
local_node_id: CoreService::NodeId,
network_settings: NetworkAdapterSettingsOfService<CoreService, RuntimeServiceId>,
network_settings: PayloadDispatcherSettingsOfService<CoreService, RuntimeServiceId>,
) -> Result<Self, modes::Error> {
match event {
EpochEvent::NewEpoch(MembershipInfo { membership, .. }) => {
@@ -164,7 +166,7 @@ where
to_mode: Mode,
overwatch_handle: &OverwatchHandle<RuntimeServiceId>,
local_node_id: CoreService::NodeId,
network_settings: NetworkAdapterSettingsOfService<CoreService, RuntimeServiceId>,
network_settings: PayloadDispatcherSettingsOfService<CoreService, RuntimeServiceId>,
) -> Result<Self, modes::Error> {
match to_mode {
Mode::Core => self.transition_to_core(overwatch_handle).await,
@@ -227,7 +229,7 @@ where
self,
overwatch_handle: &OverwatchHandle<RuntimeServiceId>,
local_node_id: CoreService::NodeId,
network_settings: NetworkAdapterSettingsOfService<CoreService, RuntimeServiceId>,
network_settings: PayloadDispatcherSettingsOfService<CoreService, RuntimeServiceId>,
) -> Result<Self, modes::Error> {
match self {
Self::Core(mode) => Ok(Self::BroadcastAfterCore {
@@ -330,7 +332,10 @@ mod tests {
use tokio::time::sleep;
use super::*;
use crate::modes::broadcast_tests::{TestMessage, TestNetworkAdapter, TestNetworkBackend};
use crate::{
modes::broadcast_tests::{TestMessage, TestNetworkBackend, TestPayloadDispatcher},
test_utils::mempool::TestMempoolService,
};
const LOCAL_NODE_ID: u8 = 99;
@@ -698,6 +703,7 @@ mod tests {
core: CoreService,
edge: EdgeService,
network: NetworkService<TestNetworkBackend, RuntimeServiceId>,
mempool: TestMempoolService<RuntimeServiceId>,
}
async fn start_network_service(handle: &OverwatchHandle<RuntimeServiceId>) {
@@ -746,7 +752,7 @@ mod tests {
}
impl CoreServiceComponents<RuntimeServiceId> for CoreService {
type NetworkAdapter = TestNetworkAdapter;
type PayloadDispatcher = TestPayloadDispatcher;
type BackendSettings = ();
type NodeId = u8;
type Rng = ();
@@ -796,6 +802,7 @@ mod tests {
core: (),
edge: (),
network: NetworkConfig { backend: () },
mempool: (),
}
}
+10 -8
View File
@@ -34,10 +34,11 @@ use tracing::{debug, error, info};
use crate::{
core::{
network::NetworkAdapter as NetworkAdapterTrait,
dispatcher::PayloadDispatcher as PayloadDispatcherTrait,
service_components::{
BlendBackendSettingsOfService, MessageComponents, NetworkAdapterSettingsOfService,
NetworkBackendOfService, ServiceComponents as CoreServiceComponents,
BlendBackendSettingsOfService, MempoolOfService, MessageComponents,
NetworkBackendOfService, PayloadDispatcherSettingsOfService,
ServiceComponents as CoreServiceComponents,
},
},
edge::service_components::ServiceComponents as EdgeServiceComponents,
@@ -48,7 +49,7 @@ use crate::{
chain::BlendEpochState,
node_id::{self, TryFrom as _},
},
message::ProxyServiceMessage,
message::{BlendPayload, ProxyServiceMessage},
settings::Settings,
};
@@ -90,7 +91,7 @@ where
type Settings = Settings<
BlendBackendSettingsOfService<CoreService, RuntimeServiceId>,
<EdgeService as EdgeServiceComponents>::BackendSettings,
NetworkAdapterSettingsOfService<CoreService, RuntimeServiceId>,
PayloadDispatcherSettingsOfService<CoreService, RuntimeServiceId>,
>;
type State = NoState<Self::Settings>;
type StateOperator = NoOperator<Self::State>;
@@ -103,13 +104,13 @@ impl<CoreService, EdgeService, SdpService, RuntimeServiceId> ServiceCore<Runtime
for BlendService<CoreService, EdgeService, SdpService, RuntimeServiceId>
where
CoreService: ServiceData<
Message: MessageComponents<CoreService::NodeId, Payload: Into<Vec<u8>>>
Message: MessageComponents<CoreService::NodeId, Payload: Into<BlendPayload>>
+ Send
+ Sync
+ 'static,
> + CoreServiceComponents<
RuntimeServiceId,
NetworkAdapter: NetworkAdapterTrait<RuntimeServiceId> + Send + Sync + 'static,
PayloadDispatcher: PayloadDispatcherTrait<RuntimeServiceId> + Send + Sync + 'static,
NodeId: Clone + Debug + Hash + Eq + Send + Sync + node_id::TryFrom + 'static,
BackendSettings: Clone + Send + Sync,
> + Send
@@ -136,7 +137,8 @@ where
NetworkBackendOfService<CoreService, RuntimeServiceId>,
RuntimeServiceId,
>,
> + AsServiceId<SdpService>
> + AsServiceId<MempoolOfService<CoreService, RuntimeServiceId>>
+ AsServiceId<SdpService>
+ Debug
+ Display
+ Clone
+82 -9
View File
@@ -1,6 +1,9 @@
use core::fmt::{self, Debug, Formatter};
use lb_blend::message::encap::validated::EncapsulatedMessageWithVerifiedPublicHeader;
use lb_blend::message::{
MAX_PAYLOAD_BODY_SIZE, PayloadType,
encap::validated::EncapsulatedMessageWithVerifiedPublicHeader,
};
use lb_core::{
mantle::NoteId,
sdp::{DeclarationId, Locator},
@@ -42,13 +45,20 @@ impl<InnerMessage> From<InnerMessage> for ProxyServiceMessage<InnerMessage> {
/// A message that is handled by [`BlendService`].
pub enum ServiceMessage<NodeId> {
/// To send a message to the blend network and eventually broadcast it to
/// the [`NetworkService`].
Blend(NetworkMessage),
/// To send a payload through the blend network, for the exit node to
/// hand over to whichever local service owns that kind of payload.
Blend(BlendPayload),
/// Request the current blend network info (connected peers).
GetNetworkInfo {
reply: oneshot::Sender<Option<NetworkInfo<NodeId>>>,
},
/// Request the transactions still waiting for a `PoW` solution to back
/// their layer proofs, oldest first.
// TODO: Change this to be tx IDs once we have strong types at the API level and we don't blend
// `Vec<u8>`s but actual `SignedMantleTx`s.
GetPendingTransactions {
reply: oneshot::Sender<Vec<Vec<u8>>>,
},
}
impl<NodeId> Debug for ServiceMessage<NodeId> {
@@ -56,22 +66,85 @@ impl<NodeId> Debug for ServiceMessage<NodeId> {
match self {
Self::Blend(msg) => f.debug_tuple("Blend").field(msg).finish(),
Self::GetNetworkInfo { .. } => f.debug_struct("GetNetworkInfo").finish(),
Self::GetPendingTransactions { .. } => {
f.debug_struct("GetPendingTransactions").finish()
}
}
}
}
impl<NodeId> From<BlendPayload> for ServiceMessage<NodeId> {
fn from(value: BlendPayload) -> Self {
Self::Blend(value)
}
}
/// The plaintext body of a Blend data message, tagged with what it carries.
// TODO: Replace with strong types for each message type Blend supports.
pub type NetworkMessage = Vec<u8>;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash)]
pub enum BlendPayload {
BlockProposal(Vec<u8>),
Transaction(Vec<u8>),
}
impl BlendPayload {
/// Wraps a transaction for blending, refusing one that could never fit.
// TODO: This will go once we move away from `Vec<u8>` and into strong types
// for each message type Blend supports.
pub fn transaction(transaction: Vec<u8>) -> Result<Self, TransactionTooLarge> {
if transaction.len() > MAX_PAYLOAD_BODY_SIZE {
return Err(TransactionTooLarge {
size: transaction.len(),
maximum: MAX_PAYLOAD_BODY_SIZE,
});
}
Ok(Self::Transaction(transaction))
}
/// The wire discriminant this payload travels under.
#[must_use]
pub const fn payload_type(&self) -> PayloadType {
match self {
Self::BlockProposal(_) => PayloadType::BlockProposal,
Self::Transaction(_) => PayloadType::Transaction,
}
}
#[must_use]
pub fn body(&self) -> &[u8] {
match self {
Self::BlockProposal(body) | Self::Transaction(body) => body,
}
}
#[must_use]
pub fn len(&self) -> usize {
self.body().len()
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.body().is_empty()
}
}
/// A transaction too large to fit in a Blend payload.
#[derive(Debug, thiserror::Error, PartialEq, Eq)]
#[error("Transaction of {size} bytes exceeds the {maximum} a Blend payload can carry.")]
pub struct TransactionTooLarge {
pub size: usize,
pub maximum: usize,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash)]
pub enum ProcessedMessage {
Network(NetworkMessage),
Decapsulated(BlendPayload),
Encapsulated(Box<EncapsulatedMessageWithVerifiedPublicHeader>),
}
impl From<NetworkMessage> for ProcessedMessage {
fn from(value: NetworkMessage) -> Self {
Self::Network(value)
impl From<BlendPayload> for ProcessedMessage {
fn from(value: BlendPayload) -> Self {
Self::Decapsulated(value)
}
}
+57 -20
View File
@@ -12,8 +12,8 @@ use overwatch::{
};
use crate::{
core::{network::NetworkAdapter, service_components::MessageComponents},
message::NetworkInfo,
core::{dispatcher::PayloadDispatcher, service_components::MessageComponents},
message::{BlendPayload, NetworkInfo},
modes::Error,
};
@@ -25,7 +25,7 @@ pub struct BroadcastMode<Adapter, NodeId, RuntimeServiceId> {
impl<Adapter, NodeId, RuntimeServiceId> BroadcastMode<Adapter, NodeId, RuntimeServiceId>
where
Adapter: NetworkAdapter<RuntimeServiceId> + Send + Sync,
Adapter: PayloadDispatcher<RuntimeServiceId> + Send + Sync,
{
pub async fn new<NetworkService>(
overwatch_handle: &OverwatchHandle<RuntimeServiceId>,
@@ -35,7 +35,13 @@ where
where
NetworkService:
ServiceData<Message = BackendNetworkMsg<Adapter::Backend, RuntimeServiceId>>,
RuntimeServiceId: AsServiceId<NetworkService> + Debug + Display + Send + Sync + 'static,
RuntimeServiceId: AsServiceId<NetworkService>
+ AsServiceId<Adapter::MempoolService>
+ Debug
+ Display
+ Send
+ Sync
+ 'static,
{
wait_until_services_are_ready!(
&overwatch_handle,
@@ -44,7 +50,8 @@ where
)
.await?;
let relay = overwatch_handle.relay::<NetworkService>().await?;
let adapter = Adapter::new(relay, network_settings);
let mempool_relay = overwatch_handle.relay::<Adapter::MempoolService>().await?;
let adapter = Adapter::new(relay, mempool_relay, network_settings);
Ok(Self {
adapter,
node_id,
@@ -55,13 +62,13 @@ where
impl<Adapter, NodeId, RuntimeServiceId> BroadcastMode<Adapter, NodeId, RuntimeServiceId>
where
Adapter: NetworkAdapter<RuntimeServiceId> + Send + Sync + 'static,
Adapter: PayloadDispatcher<RuntimeServiceId> + Send + Sync + 'static,
NodeId: Clone + Send + Sync,
RuntimeServiceId: Send + Sync + 'static,
{
pub async fn handle_inbound_message<Message>(&self, message: Message) -> Result<(), Error>
where
Message: MessageComponents<NodeId, Payload: Into<Vec<u8>>> + Send + Sync + 'static,
Message: MessageComponents<NodeId, Payload: Into<BlendPayload>> + Send + Sync + 'static,
{
match message.try_into_network_info_request() {
Ok(reply) => {
@@ -71,10 +78,20 @@ where
})));
Ok(())
}
Err(message) => {
self.adapter.broadcast(message.into_payload().into()).await;
Ok(())
}
// Nothing waits for a `PoW` solution here: a node in broadcast mode
// does no blending, so a transaction goes straight to the mempool
// instead of queueing for one.
Err(message) => match message.try_into_pending_transactions_request() {
Ok(reply) => {
drop(reply.send(Vec::new()));
Ok(())
}
Err(message) => {
// Forward directly to the dispatcher.
self.adapter.dispatch(message.into_payload().into()).await;
Ok(())
}
},
}
}
}
@@ -97,6 +114,7 @@ pub mod tests {
use tracing::{debug, info};
use super::*;
use crate::test_utils::mempool::TestMempoolService;
#[test_log::test(test)]
fn broadcast_mode() {
@@ -113,7 +131,7 @@ pub mod tests {
.unwrap();
// Create the BroadcastMode
let mut mode = BroadcastMode::<TestNetworkAdapter, (), RuntimeServiceId>::new::<
let mut mode = BroadcastMode::<TestPayloadDispatcher, (), RuntimeServiceId>::new::<
TestNetworkService,
>(app.handle(), (), ())
.await
@@ -133,7 +151,7 @@ pub mod tests {
);
// Check if the mode can be created again.
let mut mode = BroadcastMode::<TestNetworkAdapter, (), RuntimeServiceId>::new::<
let mut mode = BroadcastMode::<TestPayloadDispatcher, (), RuntimeServiceId>::new::<
TestNetworkService,
>(app.handle(), (), ())
.await
@@ -155,6 +173,7 @@ pub mod tests {
#[overwatch::derive_services]
struct Services {
network: TestNetworkService,
mempool: TestMempoolService<RuntimeServiceId>,
}
pub struct TestNetworkService {
@@ -226,7 +245,7 @@ pub mod tests {
}
}
pub struct TestNetworkAdapter {
pub struct TestPayloadDispatcher {
relay: OutboundRelay<
<NetworkService<TestNetworkBackend, RuntimeServiceId> as ServiceData>::Message,
>,
@@ -235,14 +254,19 @@ pub mod tests {
}
#[async_trait::async_trait]
impl<RuntimeServiceId> NetworkAdapter<RuntimeServiceId> for TestNetworkAdapter {
impl<RuntimeServiceId> PayloadDispatcher<RuntimeServiceId> for TestPayloadDispatcher
where
RuntimeServiceId: Send + 'static,
{
type Backend = TestNetworkBackend;
type MempoolService = TestMempoolService<RuntimeServiceId>;
type Settings = ();
fn new(
relay: OutboundRelay<
<NetworkService<Self::Backend, RuntimeServiceId> as ServiceData>::Message,
>,
_mempool_relay: OutboundRelay<<Self::MempoolService as ServiceData>::Message>,
(): Self::Settings,
) -> Self {
let (broadcasted_messages_sender, broadcasted_messages_receiver) = mpsc::channel(100);
@@ -253,8 +277,9 @@ pub mod tests {
}
}
async fn broadcast(&self, message: Vec<u8>) {
debug!("Broadcasting message: {message:?}");
async fn dispatch(&self, payload: BlendPayload) {
debug!("Dispatching payload: {payload:?}");
let message = payload.body().to_vec();
self.relay
.send(NetworkMsg::Process(message.clone()))
.await
@@ -270,10 +295,19 @@ pub mod tests {
pub struct TestMessage(Vec<u8>);
impl<NodeId> MessageComponents<NodeId> for TestMessage {
type Payload = Vec<u8>;
type Payload = BlendPayload;
fn into_payload(self) -> Self::Payload {
self.0
BlendPayload::BlockProposal(self.0)
}
fn try_into_pending_transactions_request(
self,
) -> Result<oneshot::Sender<Vec<Vec<u8>>>, Self>
where
Self: Sized,
{
Err(self)
}
fn try_into_network_info_request(
@@ -287,6 +321,9 @@ pub mod tests {
}
fn settings() -> ServicesServiceSettings {
ServicesServiceSettings { network: () }
ServicesServiceSettings {
network: (),
mempool: (),
}
}
}
+76 -1
View File
@@ -1,4 +1,7 @@
use core::{cell::Cell, convert::Infallible};
use core::{
cell::{Cell, RefCell},
convert::Infallible,
};
use async_trait::async_trait;
use lb_blend::{
@@ -17,6 +20,7 @@ use lb_blend::{
};
use lb_chain_service::Epoch;
use lb_key_management_system_service::keys::{Ed25519PublicKey, UnsecuredEd25519Key};
use tokio::sync::watch;
pub struct MockCoreAndLeaderProofsGenerator;
@@ -141,3 +145,74 @@ pub fn mock_blend_proof() -> BlendLayerProof {
ephemeral_signing_key: UnsecuredEd25519Key::generate_with_blake_rng(),
}
}
/// A proofs generator whose `PoW` branch only yields once a test lets it.
///
/// Standing in for the puzzle search, which in production takes long enough
/// that awaiting it anywhere on the event loop's critical path would stall the
/// service. Core and leadership proofs stay immediate, as they are in
/// production, so a test can tell the two apart.
pub struct GatedPowProofsGenerator;
thread_local! {
static POW_GATE: RefCell<Option<watch::Receiver<bool>>> = const { RefCell::new(None) };
}
/// Holds the `PoW` branch shut until [`Self::release`] is called.
///
/// Level-triggered on purpose: the event loop re-creates the branch future on
/// every iteration, so an edge-triggered gate would lose the release whenever
/// it happened to fire between two of them.
pub struct PowGate(watch::Sender<bool>);
impl PowGate {
/// Sets up a closed gate for generators created on this thread.
#[must_use]
pub fn setup() -> Self {
let (sender, receiver) = watch::channel(false);
POW_GATE.with_borrow_mut(|gate| *gate = Some(receiver));
Self(sender)
}
/// Lets `PoW` proof requests through, now and from now on.
pub fn release(&self) {
self.0.send_replace(true);
}
}
#[async_trait]
impl<CorePoQGenerator> CoreLeaderAndPowProofsGenerator<CorePoQGenerator>
for GatedPowProofsGenerator
{
fn new(
_settings: ProofsGeneratorSettings,
_core_proof_of_quota_generator: CorePoQGenerator,
) -> Self {
Self
}
fn set_epoch_private(
&mut self,
_winning_pol_info_stream: WinningPolInfoStream,
_target_epoch: Epoch,
) {
}
fn drop_pow_proofs_stream(&mut self) {}
async fn get_next_core_proof(&mut self) -> Option<BlendLayerProof> {
Some(mock_blend_proof())
}
async fn get_next_leader_proof(&mut self) -> Option<BlendLayerProof> {
Some(mock_blend_proof())
}
async fn get_next_pow_proof(&mut self) -> Option<BlendLayerProof> {
let mut gate = POW_GATE.with_borrow(Clone::clone)?;
gate.wait_for(|open| *open)
.await
.expect("the gate should outlive the generator");
Some(mock_blend_proof())
}
}
+47
View File
@@ -0,0 +1,47 @@
use core::future::pending;
use overwatch::{
DynError, OpaqueServiceResourcesHandle,
services::{
ServiceCore, ServiceData,
state::{NoOperator, NoState},
},
};
/// A stand-in for the mempool service a [`PayloadDispatcher`] hands
/// transactions to.
///
/// A test dispatcher holds the relay but never sends on it, so the service only
/// has to exist: it registers under a runtime service ID and then parks.
///
/// [`PayloadDispatcher`]: crate::core::dispatcher::PayloadDispatcher
pub struct TestMempoolService<RuntimeServiceId> {
service_resources_handle: OpaqueServiceResourcesHandle<Self, RuntimeServiceId>,
}
impl<RuntimeServiceId> ServiceData for TestMempoolService<RuntimeServiceId> {
type Settings = ();
type State = NoState<Self::Settings>;
type StateOperator = NoOperator<Self::State>;
type Message = ();
}
#[async_trait::async_trait]
impl<RuntimeServiceId> ServiceCore<RuntimeServiceId> for TestMempoolService<RuntimeServiceId>
where
RuntimeServiceId: Send,
{
fn init(
service_resources_handle: OpaqueServiceResourcesHandle<Self, RuntimeServiceId>,
_initial_state: Self::State,
) -> Result<Self, DynError> {
Ok(Self {
service_resources_handle,
})
}
async fn run(self) -> Result<(), DynError> {
self.service_resources_handle.status_updater.notify_ready();
pending().await
}
}
+1
View File
@@ -1,6 +1,7 @@
pub mod crypto;
pub mod epoch;
pub mod membership;
pub mod mempool;
mod libp2p;
pub use self::libp2p::*;
+4 -2
View File
@@ -8,7 +8,7 @@
use std::marker::PhantomData;
use lb_blend_service::message::{ProxyServiceMessage, ServiceMessage};
use lb_blend_service::message::{BlendPayload, ProxyServiceMessage, ServiceMessage};
use lb_codec::BinaryEncode as _;
use lb_core::block::Proposal;
use overwatch::services::{ServiceData, relay::OutboundRelay};
@@ -49,7 +49,9 @@ where
pub async fn publish_proposal(&self, proposal: Proposal) {
if let Err((e, _)) = self
.relay
.send(ServiceMessage::Blend(proposal.encode_to_vec()).into())
.send(
ServiceMessage::Blend(BlendPayload::BlockProposal(proposal.encode_to_vec())).into(),
)
.await
{
error!(target: LOG_TARGET, "Failed to relay proposal to blend service: {e:?}");
@@ -51,3 +51,46 @@ Feature: Blend
Then I declare node "NODE_1" as blend core node via the API
And blend core SDP declaration for node "NODE_1" is included on node "NODE_1"
And I stop all nodes
@blend_ci
Scenario: Transactions submitted through Blend come back through the mempool
Given the genesis block has the following wallet resources:
| account_index | token_count | token_amount |
| 1 | 2 | 1000 |
| 2 | 2 | 1000 |
| 3 | 0 | 0 |
And I have deployment config override "cryptarchia.pow_config.blend.base_difficulty" as "1"
And I have a cluster with capacity of 4 nodes
And the first 2 nodes are declared as blend providers
And I start nodes with wallet resources:
| node_name | account_index | wallet_name | connected_to |
| NODE_1 | 1 | WALLET_CORE | |
| NODE_1 | 2 | WALLET_EDGE | |
| NODE_1 | 3 | WALLET_DEST | |
And I start peer node "NODE_2" connected to node "NODE_1"
And I start peer node "NODE_3" connected to node "NODE_1"
And I start peer node "NODE_4" connected to node "NODE_1"
When all nodes have at least 2 blocks and converged to within 1 blocks in 360 seconds
# NODE_1 is a core node
And I prepare transfer transaction "TX_VIA_CORE" of 100 LGO from wallet "WALLET_CORE" to wallet "WALLET_DEST"
And I submit prepared transaction "TX_VIA_CORE" through Blend on node "NODE_1"
Then transaction "TX_VIA_CORE" is not pending in mempool of all nodes in 10 seconds
And transaction "TX_VIA_CORE" is pending in mempool of nodes in 180 seconds:
| node_name |
| NODE_1 |
| NODE_2 |
| NODE_3 |
| NODE_4 |
# NODE_4 is not a core node
When I prepare transfer transaction "TX_VIA_EDGE" of 100 LGO from wallet "WALLET_EDGE" to wallet "WALLET_DEST"
And I submit prepared transaction "TX_VIA_EDGE" through Blend on node "NODE_4"
Then transaction "TX_VIA_EDGE" is not pending in mempool of all nodes in 10 seconds
And transaction "TX_VIA_EDGE" is pending in mempool of nodes in 180 seconds:
| node_name |
| NODE_1 |
| NODE_2 |
| NODE_3 |
| NODE_4 |
Then I stop all nodes
@@ -90,6 +90,35 @@ pub async fn submit_prepared_transaction_to_nodes(
Ok(())
}
/// Hands a prepared transaction to one node's Blend service rather than its
/// mempool.
///
/// Only that node is told, and it does not gossip the transaction itself, so
/// anything that later observes it anywhere came back through Blend.
pub async fn submit_prepared_transaction_through_blend(
world: &CucumberWorld,
step: &str,
transaction_alias: &str,
node_name: &str,
) -> Result<(), StepError> {
let signed_tx = world.resolve_prepared_transaction(transaction_alias)?;
let node = world.resolve_node_http_client(node_name).inspect_err(|e| {
warn!(target: TARGET, "Step `{step}` error: {e}");
})?;
let tx_hash = node.blend_transaction(&signed_tx).await.inspect_err(|e| {
warn!(target: TARGET, "Step `{step}` error: {e}");
})?;
info!(
target: TARGET,
"Submitted prepared transaction `{transaction_alias}` ({}) through Blend on `{node_name}`",
tx_hash_to_hex(&tx_hash)
);
Ok(())
}
pub async fn try_submit_invalid_transaction(
world: &mut CucumberWorld,
step: &str,
@@ -4,8 +4,9 @@ use crate::cucumber::{
error::{StepError, StepResult},
steps::manual_mempool::{
actions::{
prepare_transfer_transaction, submit_prepared_transaction_to_nodes,
try_submit_invalid_transaction, wait_for_mempool_recovery_flush,
prepare_transfer_transaction, submit_prepared_transaction_through_blend,
submit_prepared_transaction_to_nodes, try_submit_invalid_transaction,
wait_for_mempool_recovery_flush,
},
assertions::{
assert_transaction_not_pending_on_all_nodes, assert_transaction_pending_on_nodes,
@@ -52,6 +53,21 @@ async fn step_submit_prepared_transaction_to_nodes(
submit_prepared_transaction_to_nodes(world, &step.value, transaction_alias, node_names).await
}
#[when(expr = "I submit prepared transaction {string} through Blend on node {string}")]
#[expect(
clippy::needless_pass_by_ref_mut,
reason = "Cucumber step functions require `&mut World` as the first parameter"
)]
async fn step_submit_prepared_transaction_through_blend(
world: &mut CucumberWorld,
step: &Step,
transaction_alias: String,
node_name: String,
) -> StepResult {
submit_prepared_transaction_through_blend(world, &step.value, &transaction_alias, &node_name)
.await
}
#[when(expr = "I try to submit invalid transaction {string} to node {string}")]
async fn step_try_submit_invalid_transaction(
world: &mut CucumberWorld,
@@ -178,6 +178,30 @@ impl NodeHttpClient {
.await
}
pub async fn blend_transaction<State>(
&self,
tx: &SignedMantleTx<State>,
) -> Result<TxHash, Error>
where
State: VerificationState + Send + Sync + Clone + 'static,
{
self.with_timeout(
"Blend transaction request",
self.http_client
.blend_transaction(self.base_url.clone(), tx.clone()),
)
.await
}
pub async fn blend_pending_transactions(&self) -> Result<Vec<TxHash>, Error> {
self.with_timeout(
"Blend pending transactions request",
self.http_client
.blend_pending_transactions(self.base_url.clone()),
)
.await
}
pub async fn transfer_funds(
&self,
body: WalletTransferFundsRequestBody,
+1
View File
@@ -20,6 +20,7 @@ log_targets! {
proofs::CORE_AND_LEADER,
proofs::CORE_LEADER_AND_POW,
proofs::LEADER,
proofs::LEADER_AND_POW,
proofs::POW,
},
service::{