From 14c114a525ca86458aadb7b13607aa5fc6e4ee83 Mon Sep 17 00:00:00 2001 From: Antonio Date: Wed, 19 Aug 2026 05:14:57 +0000 Subject: [PATCH] feat(blend): support tx blending (#3321) --- Cargo.lock | 2 + blend/message/src/lib.rs | 2 +- .../src/message_blend/crypto/leader/send.rs | 60 ++- .../provers/leader_and_pow/mod.rs | 76 +++ .../provers/leader_and_pow/tests.rs | 85 ++++ .../src/message_blend/provers/mod.rs | 1 + nodes/api-common/src/paths.rs | 2 + nodes/node/binary/src/api/backend.rs | 17 +- nodes/node/binary/src/api/handlers.rs | 51 ++ nodes/node/binary/src/api/openapi.rs | 2 + nodes/node/binary/src/config/blend/mod.rs | 2 +- .../binary/src/generic_services/blend/mod.rs | 28 +- nodes/node/binary/src/generic_services/mod.rs | 36 +- nodes/node/binary/src/lib.rs | 5 +- nodes/node/http-client/src/lib.rs | 43 +- services/api/src/http/blend.rs | 67 ++- services/blend/Cargo.toml | 2 + services/blend/src/core/dispatcher/libp2p.rs | 176 +++++++ services/blend/src/core/dispatcher/mod.rs | 37 ++ services/blend/src/core/mod.rs | 473 +++++++++++++----- services/blend/src/core/network/libp2p.rs | 59 --- services/blend/src/core/network/mod.rs | 27 - services/blend/src/core/service_components.rs | 50 +- services/blend/src/core/state.rs | 67 ++- services/blend/src/core/tests/mod.rs | 284 ++++++++++- services/blend/src/core/tests/utils.rs | 42 +- services/blend/src/edge/handlers.rs | 36 +- services/blend/src/edge/mod.rs | 99 +++- services/blend/src/edge/tests/mod.rs | 40 +- services/blend/src/edge/tests/utils.rs | 14 +- services/blend/src/instance.rs | 39 +- services/blend/src/lib.rs | 18 +- services/blend/src/message.rs | 91 +++- services/blend/src/modes/broadcast.rs | 77 ++- services/blend/src/test_utils/crypto.rs | 77 ++- services/blend/src/test_utils/mempool.rs | 47 ++ services/blend/src/test_utils/mod.rs | 1 + services/chain/chain-leader/src/blend.rs | 6 +- tests/cucumber_tests/features/blend.feature | 43 ++ .../cucumber/steps/manual_mempool/actions.rs | 29 ++ .../cucumber/steps/manual_mempool/steps.rs | 20 +- .../testing_framework/src/node/http_client.rs | 24 + tracing/targets/src/blend.rs | 1 + 43 files changed, 1940 insertions(+), 418 deletions(-) create mode 100644 blend/scheduling/src/message_blend/provers/leader_and_pow/mod.rs create mode 100644 blend/scheduling/src/message_blend/provers/leader_and_pow/tests.rs create mode 100644 services/blend/src/core/dispatcher/libp2p.rs create mode 100644 services/blend/src/core/dispatcher/mod.rs delete mode 100644 services/blend/src/core/network/libp2p.rs delete mode 100644 services/blend/src/core/network/mod.rs create mode 100644 services/blend/src/test_utils/mempool.rs diff --git a/Cargo.lock b/Cargo.lock index 1d9ae1563..1bd8fc16c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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", diff --git a/blend/message/src/lib.rs b/blend/message/src/lib.rs index 985f4c59a..c7baba925 100644 --- a/blend/message/src/lib.rs +++ b/blend/message/src/lib.rs @@ -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}; diff --git a/blend/scheduling/src/message_blend/crypto/leader/send.rs b/blend/scheduling/src/message_blend/crypto/leader/send.rs index 940a80e52..7acafd0f8 100644 --- a/blend/scheduling/src/message_blend/crypto/leader/send.rs +++ b/blend/scheduling/src/message_blend/crypto/leader/send.rs @@ -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 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 EpochCryptographicProcessor EpochCryptographicProcessor where - ProofsGenerator: LeaderProofsGenerator, + ProofsGenerator: LeaderAndPowProofsGenerator, { #[must_use] pub fn new( @@ -66,18 +84,35 @@ where impl EpochCryptographicProcessor where NodeId: Eq + Hash + 'static, - ProofsGenerator: LeaderProofsGenerator, + ProofsGenerator: LeaderAndPowProofsGenerator, { pub async fn encapsulate_block_proposal_payload( &mut self, payload: &[u8], + ) -> Result { + self.encapsulate_payload(PayloadType::BlockProposal, payload) + .await + } + + pub async fn encapsulate_transaction_payload( + &mut self, + payload: &[u8], + ) -> Result { + self.encapsulate_payload(PayloadType::Transaction, payload) + .await + } + + async fn encapsulate_payload( + &mut self, + payload_type: PayloadType, + payload: &[u8], ) -> Result { // 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 { + match payload_type { + PayloadType::BlockProposal => self.proofs_generator.get_next_leader_proof().await, + PayloadType::Transaction => self.proofs_generator.get_next_pow_proof().await, + } + } } diff --git a/blend/scheduling/src/message_blend/provers/leader_and_pow/mod.rs b/blend/scheduling/src/message_blend/provers/leader_and_pow/mod.rs new file mode 100644 index 000000000..df560060c --- /dev/null +++ b/blend/scheduling/src/message_blend/provers/leader_and_pow/mod.rs @@ -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; + /// 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; +} + +/// 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 { + self.leader_proofs_generator.get_next_proof().await + } + + async fn get_next_pow_proof(&mut self) -> Option { + 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) + } +} diff --git a/blend/scheduling/src/message_blend/provers/leader_and_pow/tests.rs b/blend/scheduling/src/message_blend/provers/leader_and_pow/tests.rs new file mode 100644 index 000000000..a31d63ec8 --- /dev/null +++ b/blend/scheduling/src/message_blend/provers/leader_and_pow/tests.rs @@ -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(); +} diff --git a/blend/scheduling/src/message_blend/provers/mod.rs b/blend/scheduling/src/message_blend/provers/mod.rs index 8295c0dc5..3b3808734 100644 --- a/blend/scheduling/src/message_blend/provers/mod.rs +++ b/blend/scheduling/src/message_blend/provers/mod.rs @@ -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)] diff --git a/nodes/api-common/src/paths.rs b/nodes/api-common/src/paths.rs index 5e7613450..d8f22e94e 100644 --- a/nodes/api-common/src/paths.rs +++ b/nodes/api-common/src/paths.rs @@ -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"; diff --git a/nodes/node/binary/src/api/backend.rs b/nodes/node/binary/src/api/backend.rs index 56be9c1ab..7f98dba18 100644 --- a/nodes/node/binary/src/api/backend.rs +++ b/nodes/node/binary/src/api/backend.rs @@ -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::), ) + .route( + paths::BLEND_PENDING_TRANSACTIONS, + routing::get(blend_pending_transactions::), + ) .route( paths::MEMPOOL_ADD_TX, routing::post(add_tx::), ) + .route( + paths::BLEND_DISPERSE_TRANSACTION, + routing::post(blend_tx::), + ) .route( paths::MEMPOOL_VIEW, routing::get(mempool_view::), diff --git a/nodes/node/binary/src/api/handlers.rs b/nodes/node/binary/src/api/handlers.rs index b8e06cdd4..43cf8ae7f 100644 --- a/nodes/node/binary/src/api/handlers.rs +++ b/nodes/node/binary/src/api/handlers.rs @@ -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), + (status = 500, description = "Internal server error", body = ErrorBody), + ) +)] +pub async fn blend_pending_transactions( + State(handle): State>, +) -> Response +where + BlendService: ServiceData< + Message = ProxyServiceMessage>, + > + 'static, + RuntimeServiceId: Debug + Sync + Display + 'static + AsServiceId, +{ + make_request_and_return_response!(blend::blend_pending_transactions::< + BlendService, + SignedMantleTx, + 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( + State(handle): State>, + Json(tx): Json>, +) -> Response +where + BlendService: ServiceData< + Message = ProxyServiceMessage>, + > + 'static, + RuntimeServiceId: Debug + Sync + Display + 'static + AsServiceId, +{ + make_request_and_return_response!(blend::blend_transaction::< + BlendService, + SignedMantleTx, + TxHash, + RuntimeServiceId, + >(&handle, tx, Hashable::hash)) +} + #[utoipa::path( post, path = paths::MEMPOOL_ADD_TX, diff --git a/nodes/node/binary/src/api/openapi.rs b/nodes/node/binary/src/api/openapi.rs index 08b86931c..74ef15e25 100644 --- a/nodes/node/binary/src/api/openapi.rs +++ b/nodes/node/binary/src/api/openapi.rs @@ -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, diff --git a/nodes/node/binary/src/config/blend/mod.rs b/nodes/node/binary/src/config/blend/mod.rs index 35e847be1..00b5b992b 100644 --- a/nodes/node/binary/src/config/blend/mod.rs +++ b/nodes/node/binary/src/config/blend/mod.rs @@ -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, diff --git a/nodes/node/binary/src/generic_services/blend/mod.rs b/nodes/node/binary/src/generic_services/blend/mod.rs index 2364ce6d6..f554174d3 100644 --- a/nodes/node/binary/src/generic_services/blend/mod.rs +++ b/nodes/node/binary/src/generic_services/blend/mod.rs @@ -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 = + lb_blend_service::core::dispatcher::libp2p::Libp2pPayloadDispatcher< + MempoolNetworkAdapter, + MempoolPool, + RuntimeServiceId, + >; + pub type BlendCoreRecoveryBackend = StorageRecoveryBackend< lb_blend_service::core::CoreServiceState< lb_blend_service::core::backends::libp2p::Libp2pBlendBackendSettings, - as lb_blend_service::core::network::NetworkAdapter>::Settings, + BlendBroadcastSettings, >, lb_blend_service::core::settings::StartingBlendConfig< lb_blend_service::core::backends::libp2p::Libp2pBlendBackendSettings, @@ -34,7 +45,7 @@ pub type BlendCoreRecoveryBackend = StorageRecoveryBackend< pub type BlendCoreService = lb_blend_service::core::BlendService< lb_blend_service::core::backends::libp2p::Libp2pBlendBackend, PeerId, - lb_blend_service::core::network::libp2p::Libp2pAdapter, + BlendPayloadDispatcher, SdpService, RealCoreLeaderAndPowProofsGenerator>, RealProofsVerifier, @@ -69,7 +80,7 @@ impl LeaderProofsGenerator for MockLeaderProofsGenerator { pub type BlendEdgeService = lb_blend_service::edge::BlendService< lb_blend_service::edge::backends::libp2p::Libp2pBlendBackend, PeerId, - RealLeaderProofsGenerator, + RealLeaderAndPowProofsGenerator, NtpTimeBackend, CryptarchiaService, PolInfoProvider, @@ -82,5 +93,6 @@ pub type BlendService = lb_blend_service::BlendService< RuntimeServiceId, >; -pub type BlendBroadcastSettings = - as lb_blend_service::core::network::NetworkAdapter>::Settings; +pub type BlendBroadcastSettings = as lb_blend_service::core::dispatcher::PayloadDispatcher< + RuntimeServiceId, +>>::Settings; diff --git a/nodes/node/binary/src/generic_services/mod.rs b/nodes/node/binary/src/generic_services/mod.rs index dd919da94..2695c6cd0 100644 --- a/nodes/node/binary/src/generic_services/mod.rs +++ b/nodes/node/binary/src/generic_services/mod.rs @@ -20,26 +20,30 @@ use crate::generic_services::blend::BlendService; pub mod blend; pub mod sdp; -pub type TxMempoolService = lb_tx_service::TxMempoolService< +pub type MempoolNetworkAdapter = lb_tx_service::network::adapters::libp2p::Libp2pAdapter< SignedMantleTx, as Hashable>::Hash, RuntimeServiceId, - >, - Mempool< - HeaderId, - SignedMantleTx, - TxHash, - RocksStorageAdapter< - SignedMantleTx, - as Hashable>::Hash, - >, - RuntimeServiceId, - >, - RocksStorageAdapter< - SignedMantleTx, - as Hashable>::Hash, - >, + >; + +pub type MempoolRocksStorageAdapter = RocksStorageAdapter< + SignedMantleTx, + as Hashable>::Hash, +>; + +pub type MempoolPool = Mempool< + HeaderId, + SignedMantleTx, + TxHash, + MempoolRocksStorageAdapter, + RuntimeServiceId, +>; + +pub type TxMempoolService = lb_tx_service::TxMempoolService< + MempoolNetworkAdapter, + MempoolPool, + MempoolRocksStorageAdapter, RuntimeServiceId, >; diff --git a/nodes/node/binary/src/lib.rs b/nodes/node/binary/src/lib.rs index b0a5a8fb1..a08fced41 100644 --- a/nodes/node/binary/src/lib.rs +++ b/nodes/node/binary/src/lib.rs @@ -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, diff --git a/nodes/node/http-client/src/lib.rs b/nodes/node/http-client/src/lib.rs index 7fafc2fb3..58db7e7f8 100644 --- a/nodes/node/http-client/src/lib.rs +++ b/nodes/node/http-client/src/lib.rs @@ -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( + &self, + base_url: Url, + transaction: Tx, + ) -> Result + 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(&self, base_url: Url) -> Result, 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, diff --git a/services/api/src/http/blend.rs b/services/api/src/http/blend.rs index 86db6d250..ae19f635d 100644 --- a/services/api/src/http/blend.rs +++ b/services/api/src/http/blend.rs @@ -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( + handle: &overwatch::overwatch::handle::OverwatchHandle, + transaction: Transaction, + id: impl Fn(&Transaction) -> Id, +) -> Result +where + BlendService: ServiceData>>, + Transaction: SerializeOp, + RuntimeServiceId: AsServiceId + 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::().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( + handle: &overwatch::overwatch::handle::OverwatchHandle, + id: impl Fn(&Transaction) -> Id, +) -> Result, overwatch::DynError> +where + BlendService: ServiceData>>, + Transaction: DeserializeOp, + RuntimeServiceId: AsServiceId + Debug + Sync + Display + 'static, +{ + let relay = handle.relay::().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() +} diff --git a/services/blend/Cargo.toml b/services/blend/Cargo.toml index 49e463807..e8eae9d59 100644 --- a/services/blend/Cargo.toml +++ b/services/blend/Cargo.toml @@ -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 } diff --git a/services/blend/src/core/dispatcher/libp2p.rs b/services/blend/src/core/dispatcher/libp2p.rs new file mode 100644 index 000000000..afddcebd7 --- /dev/null +++ b/services/blend/src/core/dispatcher/libp2p.rs @@ -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 = OutboundRelay>; + +/// A payload dispatcher for a node whose network service uses the libp2p +/// backend. +pub struct Libp2pPayloadDispatcher +where + Mempool: RecoverableMempool, +{ + network_relay: + OutboundRelay< as ServiceData>::Message>, + mempool_relay: MempoolRelay, + 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 + Libp2pPayloadDispatcher +where + Mempool: RecoverableMempool, + 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) { + 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 + Libp2pPayloadDispatcher +where + Mempool: RecoverableMempool, + Mempool::Item: Hashable + serde::de::DeserializeOwned + Send + 'static, + Mempool::Key: PrefixedKey + 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) { + 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 PayloadDispatcher + for Libp2pPayloadDispatcher +where + Mempool: RecoverableMempool + Send + Sync + 'static, + Mempool::Item: Hashable + serde::de::DeserializeOwned + Send + 'static, + Mempool::Key: PrefixedKey + Send + 'static, + Mempool::Settings: Clone + Send + Sync + 'static, + Mempool::Storage: MempoolStorageAdapter + Clone + Send + Sync + 'static, + MempoolNetAdapter: MempoolNetworkAdapter + + Send + + Sync + + 'static, + MempoolNetAdapter::Settings: Clone + Send + Sync + 'static, + RuntimeServiceId: Clone + + Debug + + Display + + Sync + + Send + + 'static + + AsServiceId< + StorageService< + >::Backend, + RuntimeServiceId, + >, + >, +{ + type Backend = Libp2p; + type MempoolService = + TxMempoolService; + type Settings = Libp2pBroadcastSettings; + + fn new( + network_relay: OutboundRelay< + as ServiceData>::Message, + >, + mempool_relay: OutboundRelay<::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, + } + } +} diff --git a/services/blend/src/core/dispatcher/mod.rs b/services/blend/src/core/dispatcher/mod.rs new file mode 100644 index 000000000..5e96eace1 --- /dev/null +++ b/services/blend/src/core/dispatcher/mod.rs @@ -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 { + /// The network backend used by the network service. + type Backend: NetworkBackend + 'static; + /// The mempool service transactions are handed over to. + type MempoolService: ServiceData + 'static; + /// Settings used to broadcast messages using the network service. + type Settings: Clone + Debug + Serialize + DeserializeOwned + Send + Sync + 'static; + + fn new( + network_relay: OutboundRelay< + as ServiceData>::Message, + >, + mempool_relay: OutboundRelay<::Message>, + settings: Self::Settings, + ) -> Self; + + /// Deliver a decapsulated payload to the local service that owns it. + async fn dispatch(&self, payload: BlendPayload); +} diff --git a/services/blend/src/core/mod.rs b/services/blend/src/core/mod.rs index a4adbae09..bfac5a56d 100644 --- a/services/blend/src/core/mod.rs +++ b/services/blend/src/core/mod.rs @@ -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, - Network: NetworkAdapter, + Dispatcher: PayloadDispatcher, StateStorage: RecoveryBackendTrait< RuntimeServiceId, - State = RecoveryServiceState, + State = RecoveryServiceState, > + Send + Sync, { service_resources_handle: OpaqueServiceResourcesHandle, - last_saved_state: Option>, + last_saved_state: Option>, _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, - Network: NetworkAdapter, + Dispatcher: PayloadDispatcher, StateStorage: RecoveryBackendTrait< RuntimeServiceId, - State = RecoveryServiceState, + State = RecoveryServiceState, > + Send + Sync, { - type Settings = StartingBlendConfig; - type State = RecoveryServiceState; + type Settings = StartingBlendConfig; + type State = RecoveryServiceState; type StateOperator = RecoveryOperator; type Message = ServiceMessage; } @@ -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 + Send + Sync, NodeId: membership::node_id::TryFrom + Clone + Debug + Send + Eq + Hash + Sync + 'static, - Network: NetworkAdapter + Send + Sync, + Dispatcher: PayloadDispatcher + Send + Sync, ProofsGenerator: CoreLeaderAndPowProofsGenerator> + Send, SdpService: ServiceData + Send, @@ -229,10 +230,11 @@ where PolInfoProvider: PolInfoProviderTrait + Send, StateStorage: RecoveryBackendTrait< RuntimeServiceId, - State = RecoveryServiceState, + State = RecoveryServiceState, > + Send + Sync, - RuntimeServiceId: AsServiceId> + RuntimeServiceId: AsServiceId> + + AsServiceId + AsServiceId + AsServiceId> + AsServiceId @@ -301,12 +303,16 @@ where ) .await?; - let network_adapter = async { + let payload_dispatcher = async { let network_relay = overwatch_handle .relay::>() .await .expect("Relay with network service should be available."); - Network::new(network_relay, blend_config.network.clone()) + let mempool_relay = overwatch_handle + .relay::() + .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, 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, kms_adapter: KmsAdapter, sdp_relay: &OutboundRelay, - mut last_saved_state: Option>, + mut last_saved_state: Option>, state_updater: StateUpdater< - Option>, + Option>, >, ) -> ( impl Stream>> @@ -493,7 +501,8 @@ async fn initialize< ProofsGenerator, ProofsVerifier, >, - ServiceState, + ServiceState, + VecDeque>, SchedulerWrapper, Backend, BlakeRng, @@ -501,7 +510,7 @@ async fn initialize< where NodeId: Clone + Debug + Eq + Hash + Send + 'static, Backend: BlendBackend + Sync, - NetAdapter: NetworkAdapter, + Dispatcher: PayloadDispatcher, ProofsGenerator: CoreLeaderAndPowProofsGenerator, 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: &mut Backend, - network_adapter: &NetAdapter, + payload_dispatcher: &Dispatcher, sdp_relay: &OutboundRelay, mut message_scheduler: EpochMessageScheduler< Rng, @@ -771,6 +793,7 @@ async fn run_event_loop< EncapsulatedMessageWithVerifiedPublicHeader, >, rng: &mut Rng, + mut pending_transactions: VecDeque>, mut crypto_processor: CoreCryptographicProcessor< NodeId, CorePoQGenerator, @@ -778,7 +801,7 @@ async fn run_event_loop< ProofsVerifier, >, mut current_epoch_info: CoreEpochPublicInfo, - mut recovery_checkpoint: ServiceState, + mut recovery_checkpoint: ServiceState, ) -> ( CoreCryptographicProcessor, OldEpochMessageScheduler, @@ -788,7 +811,7 @@ where NodeId: Clone + Eq + Hash + Send + Sync + 'static, Rng: rand::Rng + Clone + Send + Unpin, Backend: BlendBackend + Sync + Send, - NetAdapter: NetworkAdapter + Sync, + Dispatcher: PayloadDispatcher + Sync, ProofsGenerator: CoreLeaderAndPowProofsGenerator + 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( + transaction: Vec, + pending_transactions: &mut VecDeque>, + current_recovery_checkpoint: ServiceState, +) -> ServiceState +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( + pending_transactions: &VecDeque>, + cryptographic_processor: &mut CoreCryptographicProcessor< + NodeId, + CorePoQGenerator, + ProofsGenerator, + ProofsVerifier, + >, +) -> Option +where + NodeId: Eq + Hash + 'static, + ProofsGenerator: CoreLeaderAndPowProofsGenerator, +{ + 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>, + cryptographic_processor: &CoreCryptographicProcessor< + NodeId, + CorePoQGenerator, + ProofsGenerator, + ProofsVerifier, + >, + scheduler: &mut EpochMessageScheduler< + Rng, + ProcessedMessage, + EncapsulatedMessageWithVerifiedPublicHeader, + >, + current_recovery_checkpoint: ServiceState, +) -> ServiceState +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, 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 + Send + Sync, - NetAdapter: NetworkAdapter + Send + Sync, + Dispatcher: PayloadDispatcher + Send + Sync, ProofsGenerator: CoreLeaderAndPowProofsGenerator + 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, 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, +) -> ServiceState +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, -) -> ServiceState + payload_dispatcher: &Dispatcher, + current_recovery_checkpoint: ServiceState, +) -> ServiceState where NodeId: Eq + Hash + 'static, Rng: RngCore + Send, Backend: BlendBackend + Sync, ProofsGenerator: CoreLeaderAndPowProofsGenerator, ProofsVerifier: ProofsVerifierTrait, - NetAdapter: NetworkAdapter + Sync, + Dispatcher: PayloadDispatcher + 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, rng: &mut Rng, backend: &Backend, - network_adapter: &NetAdapter, + payload_dispatcher: &Dispatcher, epoch: Epoch, ) where NodeId: Eq + Hash + 'static, Rng: RngCore + Send, Backend: BlendBackend + Sync, - NetAdapter: NetworkAdapter + Sync, + Dispatcher: PayloadDispatcher + 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, backend: &'fut Backend, - network_adapter: &'fut NetAdapter, - mut state_updater: Option<&mut ServiceStateUpdater>, + payload_dispatcher: &'fut Dispatcher, + mut state_updater: Option<&mut ServiceStateUpdater>, epoch: Epoch, ) -> Vec> where NodeId: Eq + Hash + 'static, Backend: BlendBackend + Sync, - NetAdapter: NetworkAdapter + Sync, + Dispatcher: PayloadDispatcher + 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() diff --git a/services/blend/src/core/network/libp2p.rs b/services/blend/src/core/network/libp2p.rs deleted file mode 100644 index bda5769ef..000000000 --- a/services/blend/src/core/network/libp2p.rs +++ /dev/null @@ -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 { - network_relay: - OutboundRelay< 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 NetworkAdapter for Libp2pAdapter { - type Backend = Libp2p; - type Settings = Libp2pBroadcastSettings; - - fn new( - network_relay: OutboundRelay< - 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) { - 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}"); - } - } -} diff --git a/services/blend/src/core/network/mod.rs b/services/blend/src/core/network/mod.rs deleted file mode 100644 index 56354832f..000000000 --- a/services/blend/src/core/network/mod.rs +++ /dev/null @@ -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 { - /// The network backend used by the network service. - type Backend: NetworkBackend + 'static; - /// Settings used to broadcast messages using the network service. - type Settings: Clone + Debug + Serialize + DeserializeOwned + Send + Sync + 'static; - - fn new( - network_relay: OutboundRelay< - as ServiceData>::Message, - >, - settings: Self::Settings, - ) -> Self; - /// Broadcast a message to the network service using the configured - /// settings. - async fn broadcast(&self, message: Vec); -} diff --git a/services/blend/src/core/service_components.rs b/services/blend/src/core/service_components.rs index 05e41e3ca..fc081549d 100644 --- a/services/blend/src/core/service_components.rs +++ b/services/blend/src/core/service_components.rs @@ -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 { - type NetworkAdapter: NetworkAdapter; + type PayloadDispatcher: PayloadDispatcher; type BackendSettings; type NodeId; type Rng; @@ -45,35 +45,39 @@ impl< > where Backend: BlendBackend, - Network: NetworkAdapter, + Network: PayloadDispatcher, StateStorage: lb_services_utils::overwatch::recovery::RecoveryBackend< RuntimeServiceId, State = crate::core::state::RecoveryServiceState, > + Send + Sync, { - type NetworkAdapter = Network; + type PayloadDispatcher = Network; type BackendSettings = Backend::Settings; type NodeId = NodeId; type Rng = BlakeRng; type ProofsGenerator = ProofsGenerator; } -pub type NetworkBackendOfService = <>::NetworkAdapter as NetworkAdapter>::Backend; +pub type NetworkBackendOfService = + <>::PayloadDispatcher as PayloadDispatcher< + RuntimeServiceId, + >>::Backend; pub type BlendBackendSettingsOfService = >::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 = - <>::NetworkAdapter as NetworkAdapter< +/// The mempool service the core service's dispatcher hands transactions to. +pub type MempoolOfService = <>::PayloadDispatcher as PayloadDispatcher>::MempoolService; + +pub type PayloadDispatcherSettingsOfService = + <>::PayloadDispatcher as PayloadDispatcher< RuntimeServiceId, >>::Settings; -use crate::message::NetworkInfo; - pub trait MessageComponents { type Payload; @@ -87,16 +91,23 @@ pub trait MessageComponents { ) -> Result>>, 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>>, Self> + where + Self: Sized; } impl MessageComponents for ServiceMessage { - type Payload = Vec; + 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 MessageComponents for ServiceMessage { ) -> Result>>, 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>>, Self> { + match self { + Self::GetPendingTransactions { reply } => Ok(reply), + other @ (Self::Blend(_) | Self::GetNetworkInfo { .. }) => Err(other), } } } diff --git a/services/blend/src/core/state.rs b/services/blend/src/core/state.rs index 5d8b8019d..926ca9e3b 100644 --- a/services/blend/src/core/state.rs +++ b/services/blend/src/core/state.rs @@ -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, unsent_data_messages: HashSet, + pending_transactions: VecDeque>, current_epoch_token_collector: EpochBlendingTokenCollector, old_epoch_token_collector: Option, } @@ -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, unsent_data_messages: HashSet, + /// Transactions handed over for blending that are still waiting for a + /// `PoW` solution to back their layer proofs. + pending_transactions: VecDeque>, current_epoch_token_collector: EpochBlendingTokenCollector, old_epoch_token_collector: Option, 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, unsent_data_messages: HashSet, + pending_transactions: VecDeque>, current_epoch_token_collector: EpochBlendingTokenCollector, old_epoch_token_collector: Option, 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>, current_epoch_token_collector: EpochBlendingTokenCollector, old_epoch_token_collector: Option, 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, HashSet, + VecDeque>, EpochBlendingTokenCollector, Option, 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 { &self.unsent_data_messages } + + pub(super) fn queue_pending_transaction(&mut self, transaction: Vec) { + 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> { + &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) { + 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 StateUpdater diff --git a/services/blend/src/core/tests/mod.rs b/services/blend/src/core/tests/mod.rs index cf1cee52b..da4e61f4d 100644 --- a/services/blend/src/core/tests/mod.rs +++ b/services/blend/src/core/tests/mod.rs @@ -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 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::(&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}")); } diff --git a/services/blend/src/core/tests/utils.rs b/services/blend/src/core/tests/utils.rs index 783ea2da6..d577bcceb 100644 --- a/services/blend/src/core/tests/utils.rs +++ b/services/blend/src/core/tests/utils.rs @@ -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) { @@ -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>> = + 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 NetworkAdapter for TestNetworkAdapter { +impl PayloadDispatcher for TestPayloadDispatcher +where + RuntimeServiceId: Send + 'static, +{ type Backend = TestNetworkBackend; + type MempoolService = TestMempoolService; type Settings = (); fn new( _network_relay: OutboundRelay< as ServiceData>::Message, >, + _mempool_relay: OutboundRelay<::Message>, _settings: Self::Settings, ) -> Self { Self } - async fn broadcast(&self, _message: Vec) {} + async fn dispatch(&self, _payload: BlendPayload) { + note_outgoing_message(); + } } pub struct TestNetworkBackend { diff --git a/services/blend/src/edge/handlers.rs b/services/blend/src/edge/handlers.rs index 70014af95..b838fa400 100644 --- a/services/blend/src/edge/handlers.rs +++ b/services/blend/src/edge/handlers.rs @@ -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 where Backend: BlendBackend, NodeId: Clone, - ProofsGenerator: LeaderProofsGenerator, + ProofsGenerator: LeaderAndPowProofsGenerator, { #[cfg(test)] pub const fn epoch(&self) -> Epoch { @@ -41,7 +41,7 @@ impl where Backend: BlendBackend, NodeId: Clone + Send + 'static, - ProofsGenerator: LeaderProofsGenerator, + ProofsGenerator: LeaderAndPowProofsGenerator, { /// Creates a [`MessageHandler`] with the given membership. /// @@ -112,22 +112,42 @@ impl where NodeId: Eq + Hash + Clone + Send + 'static, Backend: BlendBackend + Sync, - ProofsGenerator: LeaderProofsGenerator, + ProofsGenerator: LeaderAndPowProofsGenerator, { - /// Blend a new message received from another service. - pub async fn handle_message_to_blend(&mut self, message: Vec) { + /// 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)] diff --git a/services/blend/src/edge/mod.rs b/services/blend/src/edge/mod.rs index b33f1ca37..f1b5649bd 100644 --- a/services/blend/src/edge/mod.rs +++ b/services/blend/src/edge/mod.rs @@ -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 + 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, PolInfoProvider: PolInfoProviderTrait + 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::( UninitializedEpochEventStream::new( public_epoch_stream, settings.time.epoch_transition_period, ), - messages_to_blend_stream, + Box::pin(inbound_relay), + local_node_id, RunningSettings:: { backend: settings.backend, cover: settings.cover, @@ -263,7 +254,8 @@ async fn run> + Unpin, >, - mut incoming_message_stream: impl Stream> + Send + Unpin, + mut inbound_relay: impl Stream> + Send + Unpin, + local_node_id: NodeId, settings: RunningSettings, overwatch_handle: &OverwatchHandle, notify_ready: impl Fn(), @@ -271,7 +263,7 @@ async fn run + Sync + Send, NodeId: Clone + Debug + Eq + Hash + Send + Sync + 'static, - ProofsGenerator: LeaderProofsGenerator + Send, + ProofsGenerator: LeaderAndPowProofsGenerator + Send, PolInfoProvider: PolInfoProviderTrait, 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 = None; + // Transactions waiting for a `PoW` solution to back their layer proofs. + let mut pending_transactions: VecDeque> = VecDeque::new(); let mut current_epoch_message_handler: Option< MessageHandler, > = 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( + pending_transactions: &VecDeque>, + current_epoch_message_handler: &mut Option< + MessageHandler, + >, +) -> Option<()> +where + Backend: BlendBackend + 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( current_public_epoch_info: &BlendEpochState, maybe_current_secret_epoch_info: &mut Option, @@ -365,7 +410,7 @@ fn handle_new_epoch_event( where Backend: BlendBackend, 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 diff --git a/services/blend/src/edge/tests/mod.rs b/services/blend/src/edge/tests/mod.rs index 7d4abf50f..e6995788f 100644 --- a/services/blend/src/edge/tests/mod.rs +++ b/services/blend/src/edge/tests/mod.rs @@ -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 diff --git a/services/blend/src/edge/tests/utils.rs b/services/blend/src/edge/tests/utils.rs index e5cca5962..e74df62ee 100644 --- a/services/blend/src/edge/tests/utils.rs +++ b/services/blend/src/edge/tests/utils.rs @@ -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 { + async fn get_next_leader_proof(&mut self) -> Option { + Some(mock_blend_proof()) + } + + async fn get_next_pow_proof(&mut self) -> Option { Some(mock_blend_proof()) } } @@ -53,7 +58,7 @@ pub async fn spawn_run( ) -> ( JoinHandle>, mpsc::Sender>, - mpsc::Sender>, + mpsc::Sender>, mpsc::Receiver, ) { 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(), || {}, diff --git a/services/blend/src/instance.rs b/services/blend/src/instance.rs index b1130ee8a..7fba6e870 100644 --- a/services/blend/src/instance.rs +++ b/services/blend/src/instance.rs @@ -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, }, - Broadcast(BroadcastMode), + Broadcast(BroadcastMode), BroadcastAfterCore { - mode: BroadcastMode, + mode: BroadcastMode, // Keep the previous core mode for the epoch transition period. prev: CoreMode, }, @@ -48,13 +49,13 @@ impl Instance where CoreService: ServiceData< - Message: MessageComponents>> + Message: MessageComponents> + Send + Sync + 'static, > + CoreServiceComponents< RuntimeServiceId, - NetworkAdapter: NetworkAdapterTrait + Send + Sync + 'static, + PayloadDispatcher: PayloadDispatcherTrait + Send + Sync + 'static, NodeId: Clone + Eq + Hash + Send + Sync, > + 'static, EdgeService: ServiceData + 'static, @@ -65,7 +66,8 @@ where NetworkBackendOfService, RuntimeServiceId, >, - > + Debug + > + AsServiceId> + + Debug + Display + Clone + Send @@ -77,7 +79,7 @@ where mode: Mode, local_node_id: CoreService::NodeId, overwatch_handle: &OverwatchHandle, - network_settings: NetworkAdapterSettingsOfService, + network_settings: PayloadDispatcherSettingsOfService, ) -> Result { 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, local_node_id: CoreService::NodeId, - network_settings: NetworkAdapterSettingsOfService, + network_settings: PayloadDispatcherSettingsOfService, ) -> Result< - BroadcastMode, + BroadcastMode, modes::Error, > { BroadcastMode::new::< @@ -140,7 +142,7 @@ where overwatch_handle: &OverwatchHandle, minimal_network_size: usize, local_node_id: CoreService::NodeId, - network_settings: NetworkAdapterSettingsOfService, + network_settings: PayloadDispatcherSettingsOfService, ) -> Result { match event { EpochEvent::NewEpoch(MembershipInfo { membership, .. }) => { @@ -164,7 +166,7 @@ where to_mode: Mode, overwatch_handle: &OverwatchHandle, local_node_id: CoreService::NodeId, - network_settings: NetworkAdapterSettingsOfService, + network_settings: PayloadDispatcherSettingsOfService, ) -> Result { match to_mode { Mode::Core => self.transition_to_core(overwatch_handle).await, @@ -227,7 +229,7 @@ where self, overwatch_handle: &OverwatchHandle, local_node_id: CoreService::NodeId, - network_settings: NetworkAdapterSettingsOfService, + network_settings: PayloadDispatcherSettingsOfService, ) -> Result { 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, + mempool: TestMempoolService, } async fn start_network_service(handle: &OverwatchHandle) { @@ -746,7 +752,7 @@ mod tests { } impl CoreServiceComponents 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: (), } } diff --git a/services/blend/src/lib.rs b/services/blend/src/lib.rs index adcd5f316..9f0dc7655 100644 --- a/services/blend/src/lib.rs +++ b/services/blend/src/lib.rs @@ -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, ::BackendSettings, - NetworkAdapterSettingsOfService, + PayloadDispatcherSettingsOfService, >; type State = NoState; type StateOperator = NoOperator; @@ -103,13 +104,13 @@ impl ServiceCore where CoreService: ServiceData< - Message: MessageComponents>> + Message: MessageComponents> + Send + Sync + 'static, > + CoreServiceComponents< RuntimeServiceId, - NetworkAdapter: NetworkAdapterTrait + Send + Sync + 'static, + PayloadDispatcher: PayloadDispatcherTrait + 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, RuntimeServiceId, >, - > + AsServiceId + > + AsServiceId> + + AsServiceId + Debug + Display + Clone diff --git a/services/blend/src/message.rs b/services/blend/src/message.rs index f89d8eaa5..bb9fcc26b 100644 --- a/services/blend/src/message.rs +++ b/services/blend/src/message.rs @@ -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 From for ProxyServiceMessage { /// A message that is handled by [`BlendService`]. pub enum ServiceMessage { - /// 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>>, }, + /// 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`s but actual `SignedMantleTx`s. + GetPendingTransactions { + reply: oneshot::Sender>>, + }, } impl Debug for ServiceMessage { @@ -56,22 +66,85 @@ impl Debug for ServiceMessage { 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 From for ServiceMessage { + 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; +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash)] +pub enum BlendPayload { + BlockProposal(Vec), + Transaction(Vec), +} + +impl BlendPayload { + /// Wraps a transaction for blending, refusing one that could never fit. + // TODO: This will go once we move away from `Vec` and into strong types + // for each message type Blend supports. + pub fn transaction(transaction: Vec) -> Result { + 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), } -impl From for ProcessedMessage { - fn from(value: NetworkMessage) -> Self { - Self::Network(value) +impl From for ProcessedMessage { + fn from(value: BlendPayload) -> Self { + Self::Decapsulated(value) } } diff --git a/services/blend/src/modes/broadcast.rs b/services/blend/src/modes/broadcast.rs index 11ca50019..853fb6341 100644 --- a/services/blend/src/modes/broadcast.rs +++ b/services/blend/src/modes/broadcast.rs @@ -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 { impl BroadcastMode where - Adapter: NetworkAdapter + Send + Sync, + Adapter: PayloadDispatcher + Send + Sync, { pub async fn new( overwatch_handle: &OverwatchHandle, @@ -35,7 +35,13 @@ where where NetworkService: ServiceData>, - RuntimeServiceId: AsServiceId + Debug + Display + Send + Sync + 'static, + RuntimeServiceId: AsServiceId + + AsServiceId + + Debug + + Display + + Send + + Sync + + 'static, { wait_until_services_are_ready!( &overwatch_handle, @@ -44,7 +50,8 @@ where ) .await?; let relay = overwatch_handle.relay::().await?; - let adapter = Adapter::new(relay, network_settings); + let mempool_relay = overwatch_handle.relay::().await?; + let adapter = Adapter::new(relay, mempool_relay, network_settings); Ok(Self { adapter, node_id, @@ -55,13 +62,13 @@ where impl BroadcastMode where - Adapter: NetworkAdapter + Send + Sync + 'static, + Adapter: PayloadDispatcher + Send + Sync + 'static, NodeId: Clone + Send + Sync, RuntimeServiceId: Send + Sync + 'static, { pub async fn handle_inbound_message(&self, message: Message) -> Result<(), Error> where - Message: MessageComponents>> + Send + Sync + 'static, + Message: MessageComponents> + 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::::new::< + let mut mode = BroadcastMode::::new::< TestNetworkService, >(app.handle(), (), ()) .await @@ -133,7 +151,7 @@ pub mod tests { ); // Check if the mode can be created again. - let mut mode = BroadcastMode::::new::< + let mut mode = BroadcastMode::::new::< TestNetworkService, >(app.handle(), (), ()) .await @@ -155,6 +173,7 @@ pub mod tests { #[overwatch::derive_services] struct Services { network: TestNetworkService, + mempool: TestMempoolService, } pub struct TestNetworkService { @@ -226,7 +245,7 @@ pub mod tests { } } - pub struct TestNetworkAdapter { + pub struct TestPayloadDispatcher { relay: OutboundRelay< as ServiceData>::Message, >, @@ -235,14 +254,19 @@ pub mod tests { } #[async_trait::async_trait] - impl NetworkAdapter for TestNetworkAdapter { + impl PayloadDispatcher for TestPayloadDispatcher + where + RuntimeServiceId: Send + 'static, + { type Backend = TestNetworkBackend; + type MempoolService = TestMempoolService; type Settings = (); fn new( relay: OutboundRelay< as ServiceData>::Message, >, + _mempool_relay: OutboundRelay<::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) { - 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); impl MessageComponents for TestMessage { - type Payload = Vec; + type Payload = BlendPayload; fn into_payload(self) -> Self::Payload { - self.0 + BlendPayload::BlockProposal(self.0) + } + + fn try_into_pending_transactions_request( + self, + ) -> Result>>, 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: (), + } } } diff --git a/services/blend/src/test_utils/crypto.rs b/services/blend/src/test_utils/crypto.rs index 0c898adfc..e2f8cd887 100644 --- a/services/blend/src/test_utils/crypto.rs +++ b/services/blend/src/test_utils/crypto.rs @@ -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>> = 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); + +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 CoreLeaderAndPowProofsGenerator + 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 { + Some(mock_blend_proof()) + } + + async fn get_next_leader_proof(&mut self) -> Option { + Some(mock_blend_proof()) + } + + async fn get_next_pow_proof(&mut self) -> Option { + 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()) + } +} diff --git a/services/blend/src/test_utils/mempool.rs b/services/blend/src/test_utils/mempool.rs new file mode 100644 index 000000000..dc4ed6632 --- /dev/null +++ b/services/blend/src/test_utils/mempool.rs @@ -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 { + service_resources_handle: OpaqueServiceResourcesHandle, +} + +impl ServiceData for TestMempoolService { + type Settings = (); + type State = NoState; + type StateOperator = NoOperator; + type Message = (); +} + +#[async_trait::async_trait] +impl ServiceCore for TestMempoolService +where + RuntimeServiceId: Send, +{ + fn init( + service_resources_handle: OpaqueServiceResourcesHandle, + _initial_state: Self::State, + ) -> Result { + Ok(Self { + service_resources_handle, + }) + } + + async fn run(self) -> Result<(), DynError> { + self.service_resources_handle.status_updater.notify_ready(); + pending().await + } +} diff --git a/services/blend/src/test_utils/mod.rs b/services/blend/src/test_utils/mod.rs index c81a62638..228894f18 100644 --- a/services/blend/src/test_utils/mod.rs +++ b/services/blend/src/test_utils/mod.rs @@ -1,6 +1,7 @@ pub mod crypto; pub mod epoch; pub mod membership; +pub mod mempool; mod libp2p; pub use self::libp2p::*; diff --git a/services/chain/chain-leader/src/blend.rs b/services/chain/chain-leader/src/blend.rs index 46a82b46e..285a7c304 100644 --- a/services/chain/chain-leader/src/blend.rs +++ b/services/chain/chain-leader/src/blend.rs @@ -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:?}"); diff --git a/tests/cucumber_tests/features/blend.feature b/tests/cucumber_tests/features/blend.feature index 47d93b328..1296c429d 100644 --- a/tests/cucumber_tests/features/blend.feature +++ b/tests/cucumber_tests/features/blend.feature @@ -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 diff --git a/tests/src/cucumber/steps/manual_mempool/actions.rs b/tests/src/cucumber/steps/manual_mempool/actions.rs index 73ed14153..aa992cd4c 100644 --- a/tests/src/cucumber/steps/manual_mempool/actions.rs +++ b/tests/src/cucumber/steps/manual_mempool/actions.rs @@ -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, diff --git a/tests/src/cucumber/steps/manual_mempool/steps.rs b/tests/src/cucumber/steps/manual_mempool/steps.rs index 45168c99e..39a303d6e 100644 --- a/tests/src/cucumber/steps/manual_mempool/steps.rs +++ b/tests/src/cucumber/steps/manual_mempool/steps.rs @@ -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, diff --git a/tests/testing_framework/src/node/http_client.rs b/tests/testing_framework/src/node/http_client.rs index 615aab55d..c11cbc4c2 100644 --- a/tests/testing_framework/src/node/http_client.rs +++ b/tests/testing_framework/src/node/http_client.rs @@ -178,6 +178,30 @@ impl NodeHttpClient { .await } + pub async fn blend_transaction( + &self, + tx: &SignedMantleTx, + ) -> Result + 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, 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, diff --git a/tracing/targets/src/blend.rs b/tracing/targets/src/blend.rs index def34e07a..b6402ccd3 100644 --- a/tracing/targets/src/blend.rs +++ b/tracing/targets/src/blend.rs @@ -20,6 +20,7 @@ log_targets! { proofs::CORE_AND_LEADER, proofs::CORE_LEADER_AND_POW, proofs::LEADER, + proofs::LEADER_AND_POW, proofs::POW, }, service::{