impl Node {
async fn queue_dataplane_unrouted_endpoint_batch(
&mut self,
remote: PeerIdentity,
payloads: Vec<EndpointDataPayload>,
enqueued_at_ms: u64,
) {
let dest_addr = *remote.node_addr();
let dest_pubkey = remote.pubkey_full();
self.register_identity(dest_addr, dest_pubkey);
let _ = self
.queue_dataplane_unrouted_endpoint_payloads(
dest_addr,
dest_pubkey,
payloads,
enqueued_at_ms,
)
.await;
}
async fn queue_dataplane_unrouted_endpoint_payloads(
&mut self,
dest_addr: NodeAddr,
dest_pubkey: secp256k1::PublicKey,
payloads: Vec<EndpointDataPayload>,
enqueued_at_ms: u64,
) -> Result<(), NodeError> {
if payloads.is_empty() {
return Ok(());
}
match self.dataplane_outbound_session_state(&dest_addr) {
OutboundSessionState::Established => {
let route_available = self.find_next_hop(&dest_addr).is_some();
if route_available && self.dataplane_has_fsp_owner(&dest_addr) {
if let Err(error) = self
.send_dataplane_cached_endpoint_payloads(&dest_addr, payloads)
.await
{
tracing::debug!(
dest = %self.peer_display_name(&dest_addr),
error = %error,
"Failed to send established endpoint data through dataplane"
);
}
return Ok(());
}
self.queue_pending_endpoint_data_batch_with_enqueued_at_ms(
dest_addr,
payloads,
enqueued_at_ms,
);
if !route_available {
self.maybe_initiate_path_recovery_lookup(&dest_addr).await;
}
Ok(())
}
OutboundSessionState::Pending => {
self.queue_pending_endpoint_data_batch_with_enqueued_at_ms(
dest_addr,
payloads,
enqueued_at_ms,
);
let should_discover = self.config.node.routing.mode
== crate::config::RoutingMode::ReplyLearned
|| self.find_next_hop(&dest_addr).is_none();
if should_discover {
self.maybe_initiate_lookup(&dest_addr).await;
}
Ok(())
}
OutboundSessionState::Missing => {
if self.find_next_hop(&dest_addr).is_none() {
self.queue_pending_endpoint_data_batch_with_enqueued_at_ms(
dest_addr,
payloads,
enqueued_at_ms,
);
self.maybe_initiate_lookup(&dest_addr).await;
return Ok(());
}
match self.initiate_session(dest_addr, dest_pubkey).await {
Ok(()) => {}
Err(NodeError::SendFailed { node_addr, reason })
if node_addr == dest_addr && reason == "no route to destination" =>
{
self.queue_pending_endpoint_data_batch_with_enqueued_at_ms(
dest_addr,
payloads,
enqueued_at_ms,
);
self.maybe_initiate_lookup(&dest_addr).await;
return Ok(());
}
Err(error) => return Err(error),
}
self.queue_pending_endpoint_data_batch_with_enqueued_at_ms(
dest_addr,
payloads,
enqueued_at_ms,
);
Ok(())
}
}
}
pub(in crate::node) fn requeue_deferred_endpoint_data_batch(
&mut self,
batch: NodeEndpointDataBatch,
) {
let (remote, payloads, _, enqueued_at_ms) = batch.into_parts();
let dest_addr = *remote.node_addr();
self.register_identity(dest_addr, remote.pubkey_full());
self.queue_pending_endpoint_data_batch_with_enqueued_at_ms(
dest_addr,
payloads,
enqueued_at_ms,
);
}
}