Skip to content

apollo_network_benchmark: add ReveresedSqmrSender struct and implementation #11554

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
sirandreww-starkware wants to merge 1 commit into
base: 01-08-apollo_network_benchmark_add_reveresedsqmr_variant_to_networkprotocol_enum
Choose a base branch
from
Open

apollo_network_benchmark: add ReveresedSqmrSender struct and implementation #11554

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 49 additions & 1 deletion crates/apollo_network_benchmark/src/bin/broadcast_network_stress_test_node/protocol.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ use apollo_network_benchmark::node_args::NetworkProtocol;
use futures::StreamExt;
use libp2p::gossipsub::{Sha256Topic, Topic};
use libp2p::PeerId;
use tracing::error;
use tracing::{error, info, trace};

// ================================
// Types and Constants
Expand Down Expand Up @@ -77,6 +77,48 @@ pub fn register_protocol_channels(
pub enum MessageSender {
Gossipsub(BroadcastTopicClient<TopicType>),
Sqmr(SqmrClientSender<TopicType, TopicType>),
ReveresedSqmr(ReveresedSqmrSender),
}

/// Wrapper for ReveresedSqmr that maintains the last active query
pub struct ReveresedSqmrSender {
server: SqmrServerReceiver<TopicType, TopicType>,
active_query: Option<apollo_network::network_manager::ServerQueryManager<TopicType, TopicType>>,
}

impl ReveresedSqmrSender {
pub fn new(server: SqmrServerReceiver<TopicType, TopicType>) -> Self {
Self { server, active_query: None }
}

async fn collect_new_queries(&mut self) {
// Non-blocking check for new queries, keeping only the last one
while let Ok(query) =
tokio::time::timeout(tokio::time::Duration::from_millis(1), self.server.next()).await
{
if let Some(query) = query {
info!("ReveresedSqmr: Received new query, replacing previous query");
self.active_query = Some(query);
} else {
break;
}
}
}
Copy link

@cursor cursor bot Feb 16, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Only one query stored despite broadcast-to-all intent

Medium Severity

collect_new_queries replaces active_query each iteration, so only the most recent query is retained and all prior queries are silently dropped. The comment on line 143 says "broadcast the message to all active queries," and the method is named broadcast_to_queries (plural), but the struct only holds a single Option — meaning in a multi-receiver benchmark scenario, only the last peer to connect actually receives data. If broadcasting to all connected peers is intended, active_query needs to be a collection.

Additional Locations (1)

Fix in Cursor Fix in Web

Copy link

@cursor cursor bot Feb 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

1ms timeout per send inflates benchmark latency metrics

Medium Severity

collect_new_queries uses a tokio::time::timeout of 1ms in its while loop. When no queries are pending (the common steady-state case), the final iteration always blocks for 1ms before the timeout fails and the loop exits. Since collect_new_queries is called on every send_message invocation, this adds at least 1ms of artificial latency to every send. In a stress test benchmark, this caps throughput at ~1000 msg/s and inflates the BROADCAST_MESSAGE_SEND_DELAY_SECONDS metric recorded by the caller, producing misleading benchmark results.

Fix in Cursor Fix in Web


async fn broadcast_to_queries(&mut self, message: TopicType) {
if let Some(query) = &mut self.active_query {
match query.send_response(message).await {
Ok(()) => {
trace!("ReveresedSqmr: Sent response to active query");
}
Err(e) => {
// Query failed, remove it
error!("ReveresedSqmr: Active query failed, removing it, error: {:?}", e);
self.active_query = None;
}
}
}
}
Copy link

@cursor cursor bot Feb 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Silent message drop when no active query exists

Medium Severity

broadcast_to_queries silently discards the message when active_query is None — no warning or error is logged. The caller in handlers.rs unconditionally increments BROADCAST_MESSAGE_COUNT and records send delay after calling send_message, so dropped messages are counted as successfully sent. This produces inaccurate benchmark metrics. The other protocol variants either panic or log errors on failure, making this inconsistency especially easy to miss during debugging.

Fix in Cursor Fix in Web

}

impl MessageSender {
Expand All @@ -95,6 +137,12 @@ impl MessageSender {
error!("Failed to send SQMR query: {:?}", e);
}
},
MessageSender::ReveresedSqmr(sender) => {
// Collect any new queries first
sender.collect_new_queries().await;
// Then broadcast the message to all active queries
sender.broadcast_to_queries(message).await;
}
}
}
}
Expand Down
Loading