#[inline]
fn record_fsp_owner_match(owner_matches_current_worker: bool) {
crate::perf_profile::record_event(if owner_matches_current_worker {
crate::perf_profile::Event::DecryptFspOwnerSame
} else {
crate::perf_profile::Event::DecryptFspOwnerMismatch
});
}
#[inline]
fn record_fsp_path_local(lane: DecryptWorkerLane) {
crate::perf_profile::record_event(crate::perf_profile::Event::DecryptFspPathLocal);
crate::perf_profile::record_event(match lane {
DecryptWorkerLane::Priority => crate::perf_profile::Event::DecryptFspPathLocalPriority,
DecryptWorkerLane::Bulk => crate::perf_profile::Event::DecryptFspPathLocalBulk,
});
}
#[inline]
fn record_fsp_path_handoff(lane: DecryptWorkerLane) {
crate::perf_profile::record_event(crate::perf_profile::Event::DecryptFspPathHandoff);
crate::perf_profile::record_event(match lane {
DecryptWorkerLane::Priority => crate::perf_profile::Event::DecryptFspPathHandoffPriority,
DecryptWorkerLane::Bulk => crate::perf_profile::Event::DecryptFspPathHandoffBulk,
});
}
#[inline]
fn record_fsp_path_worker_open_bulk() {
crate::perf_profile::record_event(crate::perf_profile::Event::DecryptFspPathWorkerOpen);
crate::perf_profile::record_event(crate::perf_profile::Event::DecryptFspPathWorkerOpenBulk);
}
fn drop_fsp_owner_handoff_job(job: FspDecryptJob) {
record_fsp_owner_handoff_drop(job.lane(), 1);
}
fn drop_fsp_owner_handoff_jobs(jobs: Vec<FspDecryptJob>) {
let mut priority = 0usize;
let mut bulk = 0usize;
for job in jobs {
match job.lane() {
DecryptWorkerLane::Priority => priority += 1,
DecryptWorkerLane::Bulk => bulk += 1,
}
}
if priority > 0 {
record_fsp_owner_handoff_drop(DecryptWorkerLane::Priority, priority);
}
if bulk > 0 {
record_fsp_owner_handoff_drop(DecryptWorkerLane::Bulk, bulk);
}
}
fn record_fsp_owner_handoff_drop(lane: DecryptWorkerLane, count: usize) {
if count == 0 {
return;
}
crate::perf_profile::record_event_count(
crate::perf_profile::Event::DecryptFspOwnerHandoffDropped,
count as u64,
);
let lane_event = match lane {
DecryptWorkerLane::Priority => crate::perf_profile::Event::DecryptWorkerPriorityDropped,
DecryptWorkerLane::Bulk => crate::perf_profile::Event::DecryptWorkerBulkDropped,
};
crate::perf_profile::record_event_count(lane_event, count as u64);
}
#[inline]
fn record_fsp_open_worker_completion_backlog_fallback() {
crate::perf_profile::record_event(
crate::perf_profile::Event::DecryptFspOpenWorkerCompletionBacklogFallback,
);
}
#[inline]
fn record_fsp_open_pool_bulk_drop(count: usize) {
if count == 0 {
return;
}
crate::perf_profile::record_event_count(
crate::perf_profile::Event::DecryptFspOpenPoolQueueFullFallback,
count as u64,
);
crate::perf_profile::record_event_count(
crate::perf_profile::Event::DecryptWorkerBulkDropped,
count as u64,
);
}
#[inline]
fn record_fsp_aead_completion_drop(event: crate::perf_profile::Event, count: usize) {
crate::perf_profile::record_event_count(event, count as u64);
}
#[inline]
fn record_fsp_aead_completion_order_error(error: &OrderedCompletionError) {
let event = match error {
OrderedCompletionError::Stale => crate::perf_profile::Event::FspAeadCompletionStaleTicket,
OrderedCompletionError::Duplicate => {
crate::perf_profile::Event::FspAeadCompletionDuplicateTicket
}
OrderedCompletionError::WindowExceeded => {
crate::perf_profile::Event::FspAeadCompletionWindowExceeded
}
};
crate::perf_profile::record_event(event);
}
fn record_fsp_aead_completion_wait(
source: FspAeadCompletionSource,
completed_at: Option<crate::perf_profile::TraceStamp>,
) {
if source.is_worker_open() {
crate::perf_profile::record_since_count(
crate::perf_profile::Stage::FspAeadWorkerOpenCompletionWait,
completed_at,
1,
);
}
}
enum FspOpenWorkerPrepareError {
Ineligible(FspDecryptJob),
OwnerCompletionBacklog(FspDecryptJob),
}
struct DecryptWorkerShard {
pool: DecryptWorkerPool,
sessions: HashMap<DecryptSessionKey, OwnedSessionState>,
fsp_sessions: HashMap<NodeAddr, OwnedFspSessionState>,
}
impl DecryptWorkerShard {
fn new(pool: DecryptWorkerPool) -> Self {
Self {
pool,
sessions: HashMap::new(),
fsp_sessions: HashMap::new(),
}
}
fn handle_msg(&mut self, idx: usize, msg: WorkerMsg) {
match msg {
WorkerMsg::Job(job) => {
self.handle_job_msg(idx, job);
}
WorkerMsg::FspJob(job) => {
self.handle_fsp_job_msg(idx, job);
}
WorkerMsg::RegisterSession { session_key, state } => {
self.register_session(idx, session_key, state);
}
WorkerMsg::RegisterFspSession { source_addr, state } => {
self.register_fsp_session(idx, source_addr, state);
}
WorkerMsg::UnregisterSession { session_key } => {
self.unregister_session(idx, session_key);
}
WorkerMsg::UnregisterFspSession { source_addr } => {
self.unregister_fsp_session(idx, source_addr);
}
}
}
fn handle_job_msg(&mut self, idx: usize, job: DecryptJob) {
match self.handle_job_action(idx, job) {
Ok(actions) => self.handle_job_actions_immediate(idx, actions),
Err(err) => {
debug!(worker = idx, error = %err, "decrypt worker job failed");
}
}
}
fn handle_bulk_job_msg(
&mut self,
idx: usize,
job: DecryptJob,
plaintext_batch: &mut DecryptPlaintextFallbackBatch,
) {
match self.handle_job_action(idx, job) {
Ok(actions) => {
self.push_job_actions_output(idx, actions, plaintext_batch, None, None);
}
Err(err) => {
debug!(worker = idx, error = %err, "decrypt worker job failed");
}
}
}
fn handle_fsp_job_msg(&mut self, idx: usize, job: FspDecryptJob) {
job.record_queue_wait();
let _t_service =
crate::perf_profile::Timer::start(crate::perf_profile::Stage::DecryptFspWorkerService);
let mut plaintext_batch = DecryptPlaintextFallbackBatch::new();
self.push_fsp_job_outputs(idx, job, &mut plaintext_batch);
plaintext_batch.flush();
trace!(worker = idx, "processed FSP decrypt worker job");
}
fn handle_bulk_fsp_job_msg(
&mut self,
idx: usize,
job: FspDecryptJob,
plaintext_batch: &mut DecryptPlaintextFallbackBatch,
) {
job.record_queue_wait();
let _t_service =
crate::perf_profile::Timer::start(crate::perf_profile::Stage::DecryptFspWorkerService);
self.push_fsp_job_outputs(idx, job, plaintext_batch);
trace!(worker = idx, "processed bulk FSP decrypt worker job");
}
fn register_session(
&mut self,
idx: usize,
session_key: DecryptSessionKey,
state: OwnedSessionState,
) {
trace!(
worker = idx,
?session_key,
"DecryptWorker: register session"
);
self.sessions.insert(session_key, state);
}
fn unregister_session(&mut self, idx: usize, session_key: DecryptSessionKey) {
trace!(
worker = idx,
?session_key,
"DecryptWorker: unregister session"
);
self.sessions.remove(&session_key);
}
fn register_fsp_session(
&mut self,
idx: usize,
source_addr: NodeAddr,
mut state: OwnedFspSessionState,
) {
trace!(
worker = idx,
%source_addr,
"DecryptWorker: register FSP session"
);
if let Some(previous) = self.fsp_sessions.remove(&source_addr) {
state.preserve_receive_order_from(previous);
}
let shared = self
.pool
.fsp_bulk_open_worker_enabled()
.then(|| state.shared_crypto_session(idx))
.flatten()
.map(Arc::new);
if let Some(shared) = &shared {
state.attach_shared_crypto_session(Arc::clone(shared));
}
self.pool.publish_fsp_aead_session(source_addr, shared);
self.fsp_sessions.insert(source_addr, state);
}
fn unregister_fsp_session(&mut self, idx: usize, source_addr: NodeAddr) {
trace!(
worker = idx,
%source_addr,
"DecryptWorker: unregister FSP session"
);
self.fsp_sessions.remove(&source_addr);
self.pool.publish_fsp_aead_session(source_addr, None);
}
#[cfg(test)]
fn handle_job(
&mut self,
job: DecryptJob,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
if let Some(output) = self.handle_job_output(0, job)? {
let _ = output.send();
}
Ok(())
}
fn handle_job_actions_immediate(&mut self, idx: usize, actions: DecryptWorkerJobActions) {
actions.for_each(|action| self.handle_job_action_immediate(idx, action));
}
fn handle_job_action_immediate(&mut self, idx: usize, action: DecryptWorkerJobAction) {
match action {
DecryptWorkerJobAction::Output(output) => {
let _ = output.send();
}
DecryptWorkerJobAction::FspJob(job) => {
for output in self.dispatch_or_handle_fsp_job(idx, job) {
let _ = output.send();
}
}
}
}
fn push_job_actions_output(
&mut self,
idx: usize,
actions: DecryptWorkerJobActions,
plaintext_batch: &mut DecryptPlaintextFallbackBatch,
fsp_batcher: Option<&mut FspDecryptJobBatcher>,
fsp_open_batcher: Option<&mut FspAeadOpenJobBatcher>,
) {
let mut fsp_batcher = fsp_batcher;
let mut fsp_open_batcher = fsp_open_batcher;
actions.for_each(|action| {
self.push_job_action_output(
idx,
action,
plaintext_batch,
fsp_batcher.as_deref_mut(),
fsp_open_batcher.as_deref_mut(),
);
});
}
fn push_job_action_output(
&mut self,
idx: usize,
action: DecryptWorkerJobAction,
plaintext_batch: &mut DecryptPlaintextFallbackBatch,
fsp_batcher: Option<&mut FspDecryptJobBatcher>,
fsp_open_batcher: Option<&mut FspAeadOpenJobBatcher>,
) {
match action {
DecryptWorkerJobAction::Output(output) => plaintext_batch.push_output(output),
DecryptWorkerJobAction::FspJob(job) => {
let owner_idx = self.pool.worker_idx_for_fsp(&job.source_addr);
record_fsp_owner_match(owner_idx == idx);
let job = if let Some(fsp_open_batcher) = fsp_open_batcher {
match self.try_prepare_fsp_bulk_open_worker_job(idx, job) {
Ok((open_idx, owner_idx, open_job)) => {
record_fsp_path_worker_open_bulk();
let returned =
fsp_open_batcher.push(&self.pool, open_idx, owner_idx, open_job);
if !returned.is_empty() {
self.drop_returned_fsp_aead_open_jobs(
idx,
returned,
plaintext_batch,
);
}
return;
}
Err(FspOpenWorkerPrepareError::Ineligible(job)) => job,
Err(FspOpenWorkerPrepareError::OwnerCompletionBacklog(job)) => {
debug_assert_ne!(owner_idx, idx);
record_fsp_open_worker_completion_backlog_fallback();
drop_fsp_owner_handoff_job(job);
return;
}
}
} else {
match self.try_start_fsp_bulk_open_worker(idx, job, plaintext_batch) {
Ok(()) => return,
Err(job) => job,
}
};
if owner_idx == idx {
record_fsp_path_local(job.lane());
self.push_fsp_job_outputs(idx, job, plaintext_batch);
return;
}
record_fsp_path_handoff(job.lane());
if let Some(fsp_batcher) = fsp_batcher {
fsp_batcher.push(&self.pool, job);
return;
}
match self.pool.dispatch_fsp_job_or_return(job) {
Ok(()) => {}
Err(job) => {
crate::perf_profile::record_event(
crate::perf_profile::Event::DecryptFspPathFallback,
);
drop_fsp_owner_handoff_job(job);
}
}
}
}
}
fn push_job_actions_output_with_fsp_batchers(
&mut self,
idx: usize,
actions: DecryptWorkerJobActions,
plaintext_batch: &mut DecryptPlaintextFallbackBatch,
) {
let mut fsp_batcher = FspDecryptJobBatcher::new();
let mut fsp_open_batcher = FspAeadOpenJobBatcher::new();
self.push_job_actions_output(
idx,
actions,
plaintext_batch,
Some(&mut fsp_batcher),
Some(&mut fsp_open_batcher),
);
fsp_batcher.flush(&self.pool);
let returned = fsp_open_batcher.flush(&self.pool);
if !returned.is_empty() {
self.drop_returned_fsp_aead_open_jobs(idx, returned, plaintext_batch);
}
}
fn local_established_fsp_meta(
packet_data: &[u8],
local_node_addr: NodeAddr,
link_msg_start: usize,
link_msg_end: usize,
) -> Option<FspDecryptJobMeta> {
local_established_fsp_datagram_meta(
packet_data,
local_node_addr,
link_msg_start,
link_msg_end,
)
}
#[allow(clippy::result_large_err)]
fn direct_session_delivery_from_message(
source_addr: NodeAddr,
local_node_addr: NodeAddr,
message: AuthenticatedSessionMessage,
) -> Result<DecryptDirectSessionDelivery, AuthenticatedSessionMessage> {
match SessionMessageType::from_byte(message.msg_type()) {
Some(SessionMessageType::EndpointData) => Ok(
DecryptDirectSessionDelivery::EndpointData(message.into_endpoint_data_delivery()),
),
Some(SessionMessageType::DataPacket) => {
let body = message.body();
if body.len() < FSP_PORT_HEADER_SIZE {
return Err(message);
}
let dst_port = u16::from_le_bytes([body[2], body[3]]);
if dst_port != FSP_PORT_IPV6_SHIM {
return Err(message);
}
let src_ipv6 = FipsAddress::from_node_addr(&source_addr).to_ipv6().octets();
let dst_ipv6 = FipsAddress::from_node_addr(&local_node_addr)
.to_ipv6()
.octets();
let Some(packet) = crate::upper::ipv6_shim::decompress_ipv6(
&body[FSP_PORT_HEADER_SIZE..],
src_ipv6,
dst_ipv6,
) else {
return Err(message);
};
Ok(DecryptDirectSessionDelivery::Ipv6Packet(packet))
}
_ => Err(message),
}
}
#[allow(clippy::too_many_arguments)]
fn direct_session_event(
sink: &DecryptDirectSessionDeliverySink,
fmp: DecryptFmpBookkeeping,
source_addr: NodeAddr,
previous_hop_peer: PeerIdentity,
ce_flag: bool,
body_len: usize,
delivery: DecryptDirectSessionDelivery,
receive_sync: FspReceiveSync,
lane: DecryptWorkerLane,
) -> (DecryptWorkerEvent, Option<PendingDirectSessionDelivery>) {
let source_peer = match &delivery {
DecryptDirectSessionDelivery::EndpointData(delivery) => delivery.source_peer,
DecryptDirectSessionDelivery::Ipv6Packet(_) => fmp.source_peer,
};
let direct_hop = previous_hop_peer.node_addr() == &source_addr;
let delivered_ipv6 = matches!(delivery, DecryptDirectSessionDelivery::Ipv6Packet(_));
if direct_hop && sink.can_deliver(&delivery) {
return (
DecryptWorkerEvent::DirectSessionCommit(DecryptDirectSessionCommit {
fmp,
source_addr,
previous_hop_peer,
ce_flag,
receive_sync,
body_len,
delivered_ipv6,
lane,
trace_enqueued_at: None,
}),
Some(PendingDirectSessionDelivery {
sink: sink.clone(),
source_addr,
source_peer,
ce_flag,
delivery,
}),
);
}
(
DecryptWorkerEvent::DirectSessionData(DecryptDirectSessionData {
fmp,
source_addr,
previous_hop_peer,
ce_flag,
receive_sync,
body_len,
delivery,
lane,
trace_enqueued_at: None,
}),
None,
)
}
#[allow(clippy::result_large_err)]
fn try_prepare_fsp_bulk_open_worker_job(
&mut self,
idx: usize,
job: FspDecryptJob,
) -> Result<(usize, usize, FspAeadOpenJob), FspOpenWorkerPrepareError> {
if !matches!(job.lane(), DecryptWorkerLane::Bulk) {
return Err(FspOpenWorkerPrepareError::Ineligible(job));
}
let source_addr = job.source_addr;
let Some(shared) = self.pool.fsp_aead_session(&source_addr) else {
return Err(FspOpenWorkerPrepareError::Ineligible(job));
};
let owner_idx = shared.owner_idx;
let eligible_open_idx = if owner_idx == idx {
if !self.pool.fsp_local_bulk_open_worker_enabled() {
return Err(FspOpenWorkerPrepareError::Ineligible(job));
}
self.pool.worker_idx_for_fsp_open_avoiding(&source_addr, idx)
} else {
if !self.pool.fsp_remote_bulk_open_worker_enabled() {
return Err(FspOpenWorkerPrepareError::Ineligible(job));
}
self.pool
.worker_idx_for_fsp_open_avoiding_pair(&source_addr, idx, owner_idx)
};
if eligible_open_idx.is_none() {
return Err(FspOpenWorkerPrepareError::Ineligible(job));
}
if owner_idx != idx
&& !self
.pool
.fsp_open_worker_owner_completion_backlog_ready(owner_idx)
{
return Err(FspOpenWorkerPrepareError::OwnerCompletionBacklog(job));
}
let payload_end = job.fsp_payload_offset.saturating_add(job.fsp_payload_len);
let Some(payload) = job.fallback.packet_data.get(job.fsp_payload_offset..payload_end)
else {
return Err(FspOpenWorkerPrepareError::Ineligible(job));
};
let Some(header) = FspEncryptedHeader::parse(payload) else {
return Err(FspOpenWorkerPrepareError::Ineligible(job));
};
let received_k_bit = header.flags & FSP_FLAG_K != 0;
if received_k_bit != shared.current_k_bit {
return Err(FspOpenWorkerPrepareError::Ineligible(job));
}
let Some(ticket) = shared.try_issue_ticket() else {
crate::perf_profile::record_event_count(
crate::perf_profile::Event::DecryptFspOpenWorkerWindowFallback,
1,
);
return Err(FspOpenWorkerPrepareError::Ineligible(job));
};
let open_idx = if owner_idx == idx {
self.pool.worker_idx_for_fsp_open_avoiding(&source_addr, idx)
} else {
self.pool
.worker_idx_for_fsp_open_avoiding_pair(&source_addr, idx, owner_idx)
}
.unwrap_or_else(|| eligible_open_idx.expect("checked eligible FSP open worker"));
let open_job = FspAeadOpenJob {
source_addr,
receive_order_id: shared.receive_order_id,
ticket,
cipher: Arc::clone(&shared.cipher),
job,
header,
completion_source: FspAeadCompletionSource::WorkerOpen,
completion_tx: None,
open_queued_at: None,
};
Ok((open_idx, owner_idx, open_job))
}
#[allow(clippy::result_large_err)]
fn try_start_fsp_bulk_open_worker(
&mut self,
idx: usize,
job: FspDecryptJob,
plaintext_batch: &mut DecryptPlaintextFallbackBatch,
) -> Result<(), FspDecryptJob> {
let (open_idx, owner_idx, open_job) = match self.try_prepare_fsp_bulk_open_worker_job(
idx, job,
) {
Ok(prepared) => prepared,
Err(FspOpenWorkerPrepareError::Ineligible(job)) => return Err(job),
Err(FspOpenWorkerPrepareError::OwnerCompletionBacklog(job)) => {
let owner_idx = self.pool.worker_idx_for_fsp(&job.source_addr);
debug_assert_ne!(owner_idx, idx);
record_fsp_open_worker_completion_backlog_fallback();
drop_fsp_owner_handoff_job(job);
return Ok(());
}
};
record_fsp_path_worker_open_bulk();
match self
.pool
.dispatch_fsp_aead_open_worker_job(open_idx, owner_idx, open_job)
{
Ok(()) => Ok(()),
Err(open_job) => {
self.drop_returned_fsp_aead_open_job(idx, open_job, plaintext_batch);
Ok(())
}
}
}
fn drop_returned_fsp_aead_open_job(
&mut self,
idx: usize,
mut open_job: FspAeadOpenJob,
plaintext_batch: &mut DecryptPlaintextFallbackBatch,
) {
record_fsp_open_pool_bulk_drop(1);
let completion_tx = open_job.completion_tx.take();
open_job.mark_returned_completion();
let completion = open_job.into_dropped_completion();
if completion_tx
.as_ref()
.is_some_and(|tx| self.pool.fsp_aead_completion_sender_is(idx, tx))
|| completion_tx.is_none()
{
self.handle_fsp_aead_completion_msg(idx, completion, plaintext_batch);
return;
}
if let Some(tx) = completion_tx {
let _ = tx.send(FspAeadCompletionBatch::one(completion));
}
}
fn drop_returned_fsp_aead_open_jobs(
&mut self,
idx: usize,
jobs: Vec<FspAeadOpenJob>,
plaintext_batch: &mut DecryptPlaintextFallbackBatch,
) {
record_fsp_open_pool_bulk_drop(jobs.len());
let mut current_tx: Option<Sender<FspAeadCompletionBatch>> = None;
let mut current_local = false;
let mut current_source_addr = None;
let mut current_receive_order_id = None;
let mut current_batch: Option<FspAeadCompletionBatch> = None;
let completion_batch_max = fsp_aead_completion_batch_max();
for mut job in jobs {
let completion_tx = job.completion_tx.take();
let local_completion = completion_tx
.as_ref()
.is_some_and(|tx| self.pool.fsp_aead_completion_sender_is(idx, tx))
|| completion_tx.is_none();
let source_addr = job.source_addr;
let receive_order_id = job.receive_order_id;
job.mark_returned_completion();
let same_batch = current_batch
.as_ref()
.is_some_and(|batch| batch.len() < completion_batch_max)
&& current_local == local_completion
&& current_source_addr == Some(source_addr)
&& current_receive_order_id == Some(receive_order_id)
&& (local_completion
|| current_tx
.as_ref()
.zip(completion_tx.as_ref())
.is_some_and(|(current, next)| current.same_channel(next)));
if !same_batch {
self.flush_dropped_fsp_aead_open_completion_batch(
idx,
current_local,
current_tx.take(),
current_batch.take(),
plaintext_batch,
);
current_local = local_completion;
current_tx = completion_tx.filter(|_| !local_completion);
current_source_addr = Some(source_addr);
current_receive_order_id = Some(receive_order_id);
current_batch = Some(FspAeadCompletionBatch::one(job.into_dropped_completion()));
continue;
}
let Some(batch) = current_batch.as_mut() else {
unreachable!("same_batch requires an active completion batch")
};
batch.push(job.into_dropped_completion());
}
self.flush_dropped_fsp_aead_open_completion_batch(
idx,
current_local,
current_tx,
current_batch,
plaintext_batch,
);
}
fn flush_dropped_fsp_aead_open_completion_batch(
&mut self,
idx: usize,
local_completion: bool,
completion_tx: Option<Sender<FspAeadCompletionBatch>>,
batch: Option<FspAeadCompletionBatch>,
plaintext_batch: &mut DecryptPlaintextFallbackBatch,
) {
let Some(batch) = batch else { return };
if local_completion {
self.handle_fsp_aead_completion_batch_msg(idx, batch, plaintext_batch);
return;
}
if let Some(tx) = completion_tx {
let _ = tx.send(batch);
}
}
fn handle_fsp_aead_completion_msg(
&mut self,
idx: usize,
completion: FspAeadCompletion,
plaintext_batch: &mut DecryptPlaintextFallbackBatch,
) {
let _t_service = crate::perf_profile::Timer::start(
crate::perf_profile::Stage::FspAeadCompletionService,
);
record_fsp_aead_completion_wait(completion.source, completion.completed_at);
let FspAeadCompletion {
source_addr,
receive_order_id,
ticket,
source: _,
result,
completed_at: _,
} = completion;
let Some(state) = self.fsp_sessions.get_mut(&source_addr) else {
record_fsp_aead_completion_drop(
crate::perf_profile::Event::FspAeadCompletionStaleSession,
1,
);
return;
};
if state.fsp_receive_order_id() != receive_order_id {
record_fsp_aead_completion_drop(
crate::perf_profile::Event::FspAeadCompletionStaleOrder,
1,
);
return;
}
let drain = match state.complete_ordered_fsp_open(ticket, result) {
Ok(drain) => drain,
Err(error) => {
record_fsp_aead_completion_order_error(&error);
debug!(
worker = idx,
?error,
%source_addr,
"dropping invalid ordered FSP AEAD completion"
);
return;
}
};
let next_ready = state.fsp_receive_order_next_ready();
debug_assert_eq!(
drain.ready,
drain.accepted
+ drain.aead_failures
+ drain.epoch_mismatches
+ drain.replay_drops
+ drain.dropped
);
crate::perf_profile::record_fsp_aead_completion_drain(
drain.ready,
drain.accepted,
drain.aead_failures,
drain.epoch_mismatches,
drain.replay_drops,
);
drain.aead_failure_sources.record();
drain.replay_drop_sources.record();
if let Some(shared) = self.pool.fsp_aead_session(&source_addr)
&& shared.receive_order_id == receive_order_id
{
shared.mark_next_ready(next_ready);
}
self.push_fsp_ready_completion_outputs(drain.outputs, plaintext_batch);
}
fn handle_fsp_aead_completion_batch_msg(
&mut self,
idx: usize,
completions: FspAeadCompletionBatch,
plaintext_batch: &mut DecryptPlaintextFallbackBatch,
) {
let Some((source_addr, receive_order_id)) = completions.common_source_order() else {
completions.for_each(|completion| {
self.handle_fsp_aead_completion_msg(idx, completion, plaintext_batch);
});
return;
};
let completions = completions.into_vec();
if completions.len() <= 1 {
for completion in completions {
self.handle_fsp_aead_completion_msg(idx, completion, plaintext_batch);
}
return;
}
self.handle_fsp_aead_completion_same_source_batch(
idx,
source_addr,
receive_order_id,
completions,
plaintext_batch,
);
}
fn handle_fsp_aead_completion_same_source_batch(
&mut self,
idx: usize,
source_addr: NodeAddr,
receive_order_id: u64,
completions: Vec<FspAeadCompletion>,
plaintext_batch: &mut DecryptPlaintextFallbackBatch,
) {
let _t_service = crate::perf_profile::Timer::start(
crate::perf_profile::Stage::FspAeadCompletionService,
);
for completion in &completions {
record_fsp_aead_completion_wait(completion.source, completion.completed_at);
}
let mut ready = 0usize;
let mut accepted = 0usize;
let mut aead_failures = 0usize;
let mut epoch_mismatches = 0usize;
let mut replay_drops = 0usize;
let mut dropped = 0usize;
let mut aead_failure_sources = FspAeadFailureSources::default();
let mut replay_drop_sources = FspReplayDropSources::default();
let mut outputs = Vec::new();
let next_ready;
{
let Some(state) = self.fsp_sessions.get_mut(&source_addr) else {
record_fsp_aead_completion_drop(
crate::perf_profile::Event::FspAeadCompletionStaleSession,
completions.len(),
);
return;
};
if state.fsp_receive_order_id() != receive_order_id {
record_fsp_aead_completion_drop(
crate::perf_profile::Event::FspAeadCompletionStaleOrder,
completions.len(),
);
return;
}
for completion in completions {
let FspAeadCompletion {
source_addr: _,
receive_order_id: _,
ticket,
source: _,
result,
completed_at: _,
} = completion;
let drain = match state.complete_ordered_fsp_open(ticket, result) {
Ok(drain) => drain,
Err(error) => {
record_fsp_aead_completion_order_error(&error);
debug!(
worker = idx,
?error,
%source_addr,
"dropping invalid ordered FSP AEAD completion"
);
continue;
}
};
debug_assert_eq!(
drain.ready,
drain.accepted
+ drain.aead_failures
+ drain.epoch_mismatches
+ drain.replay_drops
+ drain.dropped
);
ready += drain.ready;
accepted += drain.accepted;
aead_failures += drain.aead_failures;
epoch_mismatches += drain.epoch_mismatches;
replay_drops += drain.replay_drops;
dropped += drain.dropped;
aead_failure_sources.add_sources(drain.aead_failure_sources);
replay_drop_sources.add_sources(drain.replay_drop_sources);
outputs.extend(drain.outputs);
}
next_ready = state.fsp_receive_order_next_ready();
}
debug_assert_eq!(
ready,
accepted + aead_failures + epoch_mismatches + replay_drops + dropped
);
crate::perf_profile::record_fsp_aead_completion_drain(
ready,
accepted,
aead_failures,
epoch_mismatches,
replay_drops,
);
aead_failure_sources.record();
replay_drop_sources.record();
if let Some(shared) = self.pool.fsp_aead_session(&source_addr)
&& shared.receive_order_id == receive_order_id
{
shared.mark_next_ready(next_ready);
}
self.push_fsp_ready_completion_outputs(outputs, plaintext_batch);
}
fn push_fsp_ready_completion_outputs(
&self,
outputs: Vec<FspReadyCompletion>,
plaintext_batch: &mut DecryptPlaintextFallbackBatch,
) {
for output in self.outputs_for_fsp_ready_completions(outputs) {
plaintext_batch.push_output(output);
}
}
fn outputs_for_fsp_ready_completions(
&self,
outputs: Vec<FspReadyCompletion>,
) -> Vec<DecryptWorkerOutput> {
let mut ready = Vec::with_capacity(outputs.len());
for completion in outputs {
match completion {
FspReadyCompletion::Opened {
opened,
slot,
source_peer,
} => {
if let Some(output) =
self.output_for_opened_fsp_job(source_peer, opened, slot)
{
ready.push(output);
}
}
FspReadyCompletion::AeadFailed { job, header } => {
ready.push(self.output_for_fsp_aead_failure(job, &header));
}
}
}
ready
}
fn dispatch_or_handle_fsp_job(
&mut self,
idx: usize,
job: FspDecryptJob,
) -> Vec<DecryptWorkerOutput> {
let owner_idx = self.pool.worker_idx_for_fsp(&job.source_addr);
record_fsp_owner_match(owner_idx == idx);
if owner_idx == idx {
record_fsp_path_local(job.lane());
return self.handle_fsp_job_outputs(idx, job);
}
record_fsp_path_handoff(job.lane());
match self.pool.dispatch_fsp_job_or_return(job) {
Ok(()) => Vec::new(),
Err(job) => {
crate::perf_profile::record_event(
crate::perf_profile::Event::DecryptFspPathFallback,
);
drop_fsp_owner_handoff_job(job);
Vec::new()
}
}
}
fn output_for_fsp_aead_failure(
&self,
job: FspDecryptJob,
_header: &FspEncryptedHeader,
) -> DecryptWorkerOutput {
let FspDecryptJob {
fallback_tx,
fallback,
lane: _,
local_node_addr: _,
source_addr: _,
previous_hop_peer: _,
path_mtu: _,
ce_flag: _,
inner_timestamp_ms: _,
fsp_payload_offset: _,
fsp_payload_len: _,
trace_enqueued_at: _,
} = job;
crate::perf_profile::record_event(crate::perf_profile::Event::DecryptFspPathFallback);
DecryptWorkerOutput {
fallback_tx,
event: DecryptWorkerEvent::Plaintext(fallback),
direct_delivery: None,
}
}
fn output_for_malformed_fsp_drop(
&self,
fallback_tx: DecryptWorkerFallbackSender,
fallback: DecryptFallback,
lane: DecryptWorkerLane,
inner_timestamp_ms: u32,
previous_hop_peer: PeerIdentity,
) -> DecryptWorkerOutput {
crate::perf_profile::record_event(crate::perf_profile::Event::DecryptFspMalformedDropped);
DecryptWorkerOutput {
fallback_tx,
event: DecryptWorkerEvent::AuthenticatedFmpReceive(DecryptAuthenticatedFmpReceive {
fmp: DecryptFmpBookkeeping {
source_peer: fallback.source_peer,
transport_id: fallback.transport_id,
remote_addr: fallback.remote_addr,
packet_timestamp_ms: fallback.timestamp_ms,
packet_len: fallback.packet_len,
fmp_counter: fallback.fmp_counter,
inner_timestamp_ms,
fmp_flags: fallback.fmp_flags,
},
previous_hop_peer: Some(previous_hop_peer),
lane,
trace_enqueued_at: None,
}),
direct_delivery: None,
}
}
fn output_for_opened_fsp_job(
&self,
source_peer: PeerIdentity,
opened: FspOpenedJob,
slot: EpochSlot,
) -> Option<DecryptWorkerOutput> {
let FspOpenedJob {
job,
header,
plaintext_len,
} = opened;
let FspDecryptJob {
fallback_tx,
fallback,
lane,
local_node_addr,
source_addr,
previous_hop_peer,
path_mtu,
ce_flag,
inner_timestamp_ms,
fsp_payload_offset,
fsp_payload_len: _,
trace_enqueued_at: _,
} = job;
let ciphertext_offset = fsp_payload_offset + FSP_HEADER_SIZE;
let plaintext = fallback
.packet_data
.get(ciphertext_offset..ciphertext_offset + plaintext_len)?;
let (timestamp, msg_type, inner_flags_byte, _body) = fsp_strip_inner_header(plaintext)?;
let received_k_bit = header.flags & FSP_FLAG_K != 0;
let spin_bit = inner_flags_byte & 0x01 != 0;
let fmp = DecryptFmpBookkeeping {
source_peer: fallback.source_peer,
transport_id: fallback.transport_id,
remote_addr: fallback.remote_addr.clone(),
packet_timestamp_ms: fallback.timestamp_ms,
packet_len: fallback.packet_len,
fmp_counter: fallback.fmp_counter,
inner_timestamp_ms,
fmp_flags: fallback.fmp_flags,
};
let sync = FspReceiveSync {
counter: header.counter,
slot,
received_k_bit,
timestamp,
plaintext_len,
ce_flag,
path_mtu,
spin_bit,
};
let message = AuthenticatedSessionMessage::from_buffer(
source_peer,
fallback.packet_data,
ciphertext_offset,
plaintext_len,
msg_type,
inner_flags_byte,
timestamp,
);
let body_len = message.body_len();
match Self::direct_session_delivery_from_message(source_addr, local_node_addr, message) {
Ok(delivery) => {
let (event, direct_delivery) = Self::direct_session_event(
&self.pool.direct_delivery_sink,
fmp,
source_addr,
previous_hop_peer,
ce_flag,
body_len,
delivery,
sync,
lane,
);
Some(DecryptWorkerOutput {
fallback_tx,
event,
direct_delivery,
})
}
Err(message) => Some(DecryptWorkerOutput {
fallback_tx,
event: DecryptWorkerEvent::AuthenticatedSession(DecryptAuthenticatedSession {
fmp,
source_addr,
previous_hop_peer,
ce_flag,
message,
receive_sync: sync,
lane,
trace_enqueued_at: None,
}),
direct_delivery: None,
}),
}
}
fn push_fsp_job_outputs(
&mut self,
idx: usize,
job: FspDecryptJob,
plaintext_batch: &mut DecryptPlaintextFallbackBatch,
) {
for output in self.handle_fsp_job_outputs(idx, job) {
plaintext_batch.push_output(output);
}
}
fn handle_fsp_job_outputs(
&mut self,
idx: usize,
job: FspDecryptJob,
) -> Vec<DecryptWorkerOutput> {
let FspDecryptJob {
fallback_tx,
mut fallback,
lane,
local_node_addr,
source_addr,
previous_hop_peer,
path_mtu,
ce_flag,
inner_timestamp_ms,
fsp_payload_offset,
fsp_payload_len,
trace_enqueued_at: _,
} = job;
let Some(state) = self.fsp_sessions.get_mut(&source_addr) else {
return vec![DecryptWorkerOutput {
fallback_tx,
event: DecryptWorkerEvent::Plaintext(fallback),
direct_delivery: None,
}];
};
let payload_end = fsp_payload_offset.saturating_add(fsp_payload_len);
let header = {
let Some(payload) = fallback.packet_data.get(fsp_payload_offset..payload_end) else {
return vec![self.output_for_malformed_fsp_drop(
fallback_tx,
fallback,
lane,
inner_timestamp_ms,
previous_hop_peer,
)];
};
let Some(header) = FspEncryptedHeader::parse(payload) else {
return vec![self.output_for_malformed_fsp_drop(
fallback_tx,
fallback,
lane,
inner_timestamp_ms,
previous_hop_peer,
)];
};
header
};
let fmp = DecryptFmpBookkeeping {
source_peer: fallback.source_peer,
transport_id: fallback.transport_id,
remote_addr: fallback.remote_addr.clone(),
packet_timestamp_ms: fallback.timestamp_ms,
packet_len: fallback.packet_len,
fmp_counter: fallback.fmp_counter,
inner_timestamp_ms,
fmp_flags: fallback.fmp_flags,
};
if state.has_single_current_epoch() {
let Some(ticket) = state.issue_fsp_receive_ticket() else {
match lane {
DecryptWorkerLane::Priority => {
record_decrypt_worker_priority_drop(idx, "fsp-receive-window");
}
DecryptWorkerLane::Bulk => {
record_decrypt_worker_bulk_drop_count(idx, 1);
}
}
return Vec::new();
};
let ciphertext_offset = fsp_payload_offset + FSP_HEADER_SIZE;
let Some(ciphertext) = fallback.packet_data.get_mut(ciphertext_offset..payload_end)
else {
return vec![self.output_for_malformed_fsp_drop(
fallback_tx,
fallback,
lane,
inner_timestamp_ms,
previous_hop_peer,
)];
};
let open_result = state.current_epoch_matches(&header).then(|| {
let _t_fsp =
crate::perf_profile::Timer::start(crate::perf_profile::Stage::FspDecrypt);
state.open_current_established_frame_in_place_deferred_replay(&header, ciphertext)
});
let job = FspDecryptJob {
fallback_tx,
fallback,
lane,
local_node_addr,
source_addr,
previous_hop_peer,
path_mtu,
ce_flag,
inner_timestamp_ms,
fsp_payload_offset,
fsp_payload_len,
trace_enqueued_at: None,
};
let completion = match open_result {
Some(Ok(plaintext_len)) => FspOrderedCompletion::Opened {
opened: FspOpenedJob {
job,
header,
plaintext_len,
},
source: FspAeadCompletionSource::Local,
},
Some(Err(FspOpenError::Aead)) => {
crate::perf_profile::record_fsp_aead_completion_local_open_aead_failure();
FspOrderedCompletion::AeadFailed {
job,
header,
source: FspAeadCompletionSource::Local,
}
}
Some(Err(FspOpenError::Replay)) => {
FspOrderedCompletion::AeadFailed {
job,
header,
source: FspAeadCompletionSource::Local,
}
}
None => FspOrderedCompletion::EpochMismatch {
job,
header,
source: FspAeadCompletionSource::Local,
},
};
let drain = match state.complete_ordered_fsp_open(ticket, completion) {
Ok(drain) => drain,
Err(error) => {
record_fsp_aead_completion_order_error(&error);
debug!(
worker = idx,
?error,
%source_addr,
"dropping invalid local ordered FSP completion"
);
return Vec::new();
}
};
let next_ready = state.fsp_receive_order_next_ready();
debug_assert_eq!(
drain.ready,
drain.accepted
+ drain.aead_failures
+ drain.epoch_mismatches
+ drain.replay_drops
+ drain.dropped
);
crate::perf_profile::record_fsp_aead_completion_drain(
drain.ready,
drain.accepted,
drain.aead_failures,
drain.epoch_mismatches,
drain.replay_drops,
);
drain.aead_failure_sources.record();
drain.replay_drop_sources.record();
if let Some(shared) = self.pool.fsp_aead_session(&source_addr)
&& shared.receive_order_id == state.fsp_receive_order_id()
{
shared.mark_next_ready(next_ready);
}
return self.outputs_for_fsp_ready_completions(drain.outputs);
}
let Some(payload) = fallback.packet_data.get(fsp_payload_offset..payload_end) else {
return vec![self.output_for_malformed_fsp_drop(
fallback_tx,
fallback,
lane,
inner_timestamp_ms,
previous_hop_peer,
)];
};
let ciphertext = &payload[FSP_HEADER_SIZE..];
let received_k_bit = header.flags & FSP_FLAG_K != 0;
let open_result = {
let _t_fsp =
crate::perf_profile::Timer::start(crate::perf_profile::Stage::FspDecrypt);
state.open_established_frame(&header, ciphertext)
};
let FspOpenSuccess { plaintext, slot } = match open_result {
Ok(success) => success,
Err(FspOpenError::Replay) => {
crate::perf_profile::record_event(
crate::perf_profile::Event::DecryptFspWorkerReplayDropped,
);
return Vec::new();
}
Err(FspOpenError::Aead) => {
let job = FspDecryptJob {
fallback_tx,
fallback,
lane,
local_node_addr,
source_addr,
previous_hop_peer,
path_mtu,
ce_flag,
inner_timestamp_ms,
fsp_payload_offset,
fsp_payload_len,
trace_enqueued_at: None,
};
return vec![self.output_for_fsp_aead_failure(job, &header)];
}
};
let Some((timestamp, msg_type, inner_flags_byte, _body)) =
fsp_strip_inner_header(&plaintext)
else {
return Vec::new();
};
let spin_bit = inner_flags_byte & 0x01 != 0;
let plaintext_len = plaintext.len();
let sync = FspReceiveSync {
counter: header.counter,
slot,
received_k_bit,
timestamp,
plaintext_len,
ce_flag,
path_mtu,
spin_bit,
};
let message = AuthenticatedSessionMessage::new(
state.source_peer,
plaintext,
msg_type,
inner_flags_byte,
timestamp,
);
let body_len = message.body_len();
let event =
match Self::direct_session_delivery_from_message(source_addr, local_node_addr, message)
{
Ok(delivery) => {
let (event, direct_delivery) = Self::direct_session_event(
&self.pool.direct_delivery_sink,
fmp,
source_addr,
previous_hop_peer,
ce_flag,
body_len,
delivery,
sync,
lane,
);
return vec![DecryptWorkerOutput {
fallback_tx,
event,
direct_delivery,
}];
}
Err(message) => {
DecryptWorkerEvent::AuthenticatedSession(DecryptAuthenticatedSession {
fmp,
source_addr,
previous_hop_peer,
ce_flag,
message,
receive_sync: sync,
lane,
trace_enqueued_at: None,
})
}
};
vec![DecryptWorkerOutput {
fallback_tx,
event,
direct_delivery: None,
}]
}
fn handle_job_action(
&mut self,
idx: usize,
job: DecryptJob,
) -> Result<DecryptWorkerJobActions, Box<dyn std::error::Error + Send + Sync>> {
job.record_queue_wait();
let DecryptJob {
mut packet_data,
lane,
session_key,
_transport_id: transport_id,
_remote_addr: remote_addr,
local_node_addr,
timestamp_ms,
fmp_counter,
fmp_flags,
fmp_header,
fmp_ciphertext_offset,
fallback_tx,
trace_enqueued_at: _,
} = job;
let packet_len = packet_data.len();
if !self.pool.fmp_aead_helpers_enabled() {
let (source_peer, fmp_plaintext_len) = {
let state = match self.sessions.get_mut(&session_key) {
Some(s) => s,
None => {
let _ = fallback_tx; let _ = packet_data;
return Ok(DecryptWorkerJobActions::None);
}
};
let source_peer = state.source_peer;
let precheck = match state.precheck_fmp_replay(fmp_counter) {
Ok(precheck) => precheck,
Err(FmpOpenError::Replay) => return Ok(DecryptWorkerJobActions::None),
#[cfg(test)]
Err(FmpOpenError::Aead { .. }) => {
unreachable!("FMP replay precheck cannot run AEAD")
}
};
let outcome = match OwnedSessionState::open_fmp_aead_in_place(
&state.fmp_cipher,
&mut packet_data,
fmp_ciphertext_offset,
fmp_counter,
fmp_flags,
&fmp_header,
) {
Ok(outcome) => outcome,
Err(()) => {
return Ok(DecryptWorkerJobActions::one(DecryptWorkerJobAction::Output(
DecryptWorkerOutput {
fallback_tx,
event: DecryptWorkerEvent::DecryptFailure(
DecryptFailureReport {
source_peer,
fmp_counter,
fmp_replay_highest: precheck.replay_highest,
trace_enqueued_at: None,
},
),
direct_delivery: None,
},
)));
}
};
if OwnedSessionState::accept_prechecked_fmp_replay_on(
&mut state.fmp_replay,
precheck,
)
.is_err()
{
return Ok(DecryptWorkerJobActions::None);
}
(source_peer, outcome.plaintext_len)
};
let opened = OpenedFmpJob {
packet_data,
lane,
source_peer,
transport_id,
remote_addr,
local_node_addr,
timestamp_ms,
packet_len,
fmp_counter,
fmp_flags,
fmp_plaintext_offset: fmp_ciphertext_offset,
fmp_plaintext_len,
fallback_tx,
};
return Ok(Self::handle_opened_fmp_job(opened)
.map(DecryptWorkerJobActions::one)
.unwrap_or(DecryptWorkerJobActions::None));
}
let (source_peer, receive_order_id, replay_precheck, ticket, cipher) = {
let state = match self.sessions.get_mut(&session_key) {
Some(s) => s,
None => {
let _ = fallback_tx; let _ = packet_data;
return Ok(DecryptWorkerJobActions::None);
}
};
let source_peer = state.source_peer;
let replay_precheck = match state.precheck_fmp_replay(fmp_counter) {
Ok(precheck) => precheck,
Err(FmpOpenError::Replay) => return Ok(DecryptWorkerJobActions::None),
#[cfg(test)]
Err(FmpOpenError::Aead { .. }) => {
unreachable!("FMP replay precheck cannot run AEAD")
}
};
let Some(ticket) = state.issue_fmp_receive_ticket() else {
crate::perf_profile::record_event(crate::perf_profile::Event::DecryptWorkerQueueFull);
return Ok(DecryptWorkerJobActions::None);
};
(
source_peer,
state.fmp_receive_order_id(),
replay_precheck,
ticket,
Arc::clone(&state.fmp_cipher),
)
};
let helper_job = FmpAeadHelperJob {
session_key,
receive_order_id,
ticket,
replay: FmpReplayDecision::Prechecked(replay_precheck),
cipher,
fmp_header,
opened: OpenedFmpJob {
packet_data,
lane,
source_peer,
transport_id,
remote_addr,
local_node_addr,
timestamp_ms,
packet_len,
fmp_counter,
fmp_flags,
fmp_plaintext_offset: fmp_ciphertext_offset,
fmp_plaintext_len: 0,
fallback_tx,
},
completion_tx: None,
helper_queued_at: None,
};
match self.pool.dispatch_fmp_aead_helper_job(idx, helper_job) {
Ok(()) => Ok(DecryptWorkerJobActions::None),
Err(helper_job) => {
let completion = helper_job.into_completion();
Ok(self.handle_fmp_aead_completion_action(completion))
}
}
}
fn handle_fmp_aead_completion_msg(
&mut self,
idx: usize,
completion: FmpAeadCompletion,
plaintext_batch: &mut DecryptPlaintextFallbackBatch,
) {
let _t_service =
crate::perf_profile::Timer::start(crate::perf_profile::Stage::FmpAeadHelperCompletionService);
let actions = self.handle_fmp_aead_completion_action(completion);
self.push_job_actions_output_with_fsp_batchers(idx, actions, plaintext_batch);
}
fn handle_fmp_aead_completion_batch_msg(
&mut self,
idx: usize,
completions: FmpAeadCompletionBatch,
plaintext_batch: &mut DecryptPlaintextFallbackBatch,
) {
let Some((session_key, receive_order_id)) = completions.common_session_order() else {
completions.for_each(|completion| {
self.handle_fmp_aead_completion_msg(idx, completion, plaintext_batch);
});
return;
};
let completions = completions.into_vec();
if completions.len() <= 1 {
for completion in completions {
self.handle_fmp_aead_completion_msg(idx, completion, plaintext_batch);
}
return;
}
self.handle_fmp_aead_completion_same_session_batch(
idx,
session_key,
receive_order_id,
completions,
plaintext_batch,
);
}
fn handle_fmp_aead_completion_same_session_batch(
&mut self,
idx: usize,
session_key: DecryptSessionKey,
receive_order_id: u64,
completions: Vec<FmpAeadCompletion>,
plaintext_batch: &mut DecryptPlaintextFallbackBatch,
) {
let _t_service = crate::perf_profile::Timer::start(
crate::perf_profile::Stage::FmpAeadHelperCompletionService,
);
for completion in &completions {
let (priority_count, bulk_count) = match completion.result.lane() {
DecryptWorkerLane::Priority => (1, 0),
DecryptWorkerLane::Bulk => (0, 1),
};
crate::perf_profile::record_since_split_count(
crate::perf_profile::Stage::FmpAeadHelperCompletionWait,
crate::perf_profile::Stage::FmpAeadHelperPriorityCompletionWait,
crate::perf_profile::Stage::FmpAeadHelperBulkCompletionWait,
completion.completed_at,
1,
priority_count,
bulk_count,
);
}
let mut ready = 0usize;
let mut accepted = 0usize;
let mut aead_failures = 0usize;
let mut replay_drops = 0usize;
let mut actions = DecryptWorkerJobActions::None;
{
let Some(state) = self.sessions.get_mut(&session_key) else {
return;
};
if state.fmp_receive_order_id() != receive_order_id {
return;
}
for completion in completions {
let FmpAeadCompletion {
session_key: _,
receive_order_id: _,
ticket,
completed_at: _,
result,
} = completion;
let ordered = match result {
FmpAeadCompletionResult::Opened { replay, opened } => {
FmpOrderedCompletion::Opened {
replay,
value: opened,
}
}
FmpAeadCompletionResult::AeadFailed(failure) => {
FmpOrderedCompletion::AeadFailed(failure)
}
};
let drain_result =
state.complete_ordered_fmp_open_with_value(ticket, ordered, |ready| {
match ready {
FmpReadyCompletion::Opened(opened_job) => {
if let Some(action) = Self::handle_opened_fmp_job(opened_job) {
actions.push(action);
}
}
FmpReadyCompletion::AeadFailed(failure) => {
actions.push(Self::fmp_aead_failure_action(failure));
}
}
});
match drain_result {
Ok(drain) => {
ready += drain.ready;
accepted += drain.accepted;
aead_failures += drain.aead_failures;
replay_drops += drain.replay_drops;
}
Err(FmpOpenError::Replay) => {}
#[cfg(test)]
Err(FmpOpenError::Aead { .. }) => {
unreachable!("ordered FMP completion cannot run AEAD")
}
}
}
}
debug_assert_eq!(ready, accepted + aead_failures + replay_drops);
crate::perf_profile::record_fmp_aead_completion_drain(
ready,
accepted,
aead_failures,
replay_drops,
);
self.push_job_actions_output_with_fsp_batchers(idx, actions, plaintext_batch);
}
fn handle_fmp_aead_completion_action(
&mut self,
completion: FmpAeadCompletion,
) -> DecryptWorkerJobActions {
let (priority_count, bulk_count) = match completion.result.lane() {
DecryptWorkerLane::Priority => (1, 0),
DecryptWorkerLane::Bulk => (0, 1),
};
crate::perf_profile::record_since_split_count(
crate::perf_profile::Stage::FmpAeadHelperCompletionWait,
crate::perf_profile::Stage::FmpAeadHelperPriorityCompletionWait,
crate::perf_profile::Stage::FmpAeadHelperBulkCompletionWait,
completion.completed_at,
1,
priority_count,
bulk_count,
);
let FmpAeadCompletion {
session_key,
receive_order_id,
ticket,
completed_at: _,
result,
} = completion;
let Some(state) = self.sessions.get_mut(&session_key) else {
return DecryptWorkerJobActions::None;
};
if state.fmp_receive_order_id() != receive_order_id {
return DecryptWorkerJobActions::None;
}
match result {
FmpAeadCompletionResult::Opened { replay, opened } => {
let mut actions = DecryptWorkerJobActions::None;
let drain_result = state.complete_ordered_fmp_open_with_value(
ticket,
FmpOrderedCompletion::Opened {
replay,
value: opened,
},
|ready| match ready {
FmpReadyCompletion::Opened(opened_job) => {
if let Some(action) = Self::handle_opened_fmp_job(opened_job) {
actions.push(action);
}
}
FmpReadyCompletion::AeadFailed(failure) => {
actions.push(Self::fmp_aead_failure_action(failure));
}
},
);
match drain_result {
Ok(drain) => {
crate::perf_profile::record_fmp_aead_completion_drain(
drain.ready,
drain.accepted,
drain.aead_failures,
drain.replay_drops,
);
}
Err(FmpOpenError::Replay) => return actions,
#[cfg(test)]
Err(FmpOpenError::Aead { .. }) => {
unreachable!("ordered FMP completion cannot run AEAD")
}
}
actions
}
FmpAeadCompletionResult::AeadFailed(failure) => {
let mut actions = DecryptWorkerJobActions::None;
let drain_result = state.complete_ordered_fmp_open_with_value(
ticket,
FmpOrderedCompletion::AeadFailed(failure),
|ready| match ready {
FmpReadyCompletion::Opened(opened_job) => {
if let Some(action) = Self::handle_opened_fmp_job(opened_job) {
actions.push(action);
}
}
FmpReadyCompletion::AeadFailed(failure) => {
actions.push(Self::fmp_aead_failure_action(failure));
}
},
);
if let Ok(drain) = drain_result {
crate::perf_profile::record_fmp_aead_completion_drain(
drain.ready,
drain.accepted,
drain.aead_failures,
drain.replay_drops,
);
}
actions
}
}
}
fn fmp_aead_failure_action(failure: FmpAeadFailure) -> DecryptWorkerJobAction {
DecryptWorkerJobAction::Output(DecryptWorkerOutput {
fallback_tx: failure.fallback_tx,
event: DecryptWorkerEvent::DecryptFailure(DecryptFailureReport {
source_peer: failure.source_peer,
fmp_counter: failure.fmp_counter,
fmp_replay_highest: failure.fmp_replay_highest.unwrap_or(0),
trace_enqueued_at: None,
}),
direct_delivery: None,
})
}
fn handle_opened_fmp_job(job: OpenedFmpJob) -> Option<DecryptWorkerJobAction> {
let OpenedFmpJob {
packet_data,
lane: _,
source_peer,
transport_id,
remote_addr,
local_node_addr,
timestamp_ms,
packet_len,
fmp_counter,
fmp_flags,
fmp_plaintext_offset,
fmp_plaintext_len,
fallback_tx,
} = job;
let fmp_plaintext_start = fmp_plaintext_offset;
let fmp_plaintext_end = fmp_plaintext_offset + fmp_plaintext_len;
const INNER_TIMESTAMP_LEN: usize = 4;
if fmp_plaintext_len < INNER_TIMESTAMP_LEN {
return None;
}
let inner_timestamp_ms = u32::from_le_bytes([
packet_data[fmp_plaintext_start],
packet_data[fmp_plaintext_start + 1],
packet_data[fmp_plaintext_start + 2],
packet_data[fmp_plaintext_start + 3],
]);
if fmp_plaintext_len == INNER_TIMESTAMP_LEN {
let fmp = DecryptFmpBookkeeping {
source_peer,
transport_id,
remote_addr,
packet_timestamp_ms: timestamp_ms,
packet_len,
fmp_counter,
inner_timestamp_ms,
fmp_flags,
};
return Some(DecryptWorkerJobAction::Output(DecryptWorkerOutput {
fallback_tx,
event: DecryptWorkerEvent::AuthenticatedFmpReceive(
DecryptAuthenticatedFmpReceive {
fmp,
previous_hop_peer: None,
lane: DecryptWorkerLane::Priority,
trace_enqueued_at: None,
},
),
direct_delivery: None,
}));
}
let link_msg_start = fmp_plaintext_start + INNER_TIMESTAMP_LEN;
let link_msg_end = fmp_plaintext_end;
let fsp_meta = Self::local_established_fsp_meta(
&packet_data,
local_node_addr,
link_msg_start,
link_msg_end,
);
let fallback = DecryptFallback::new(
source_peer,
transport_id,
remote_addr,
timestamp_ms,
packet_len,
fmp_counter,
fmp_flags,
packet_data,
fmp_plaintext_start,
fmp_plaintext_len,
);
if let Some(meta) = fsp_meta {
let fsp_job = FspDecryptJob {
fallback_tx: fallback_tx.clone(),
fallback,
lane: DecryptWorkerLane::Bulk,
local_node_addr,
source_addr: meta.source_addr,
previous_hop_peer: source_peer,
path_mtu: meta.path_mtu,
ce_flag: fmp_flags & crate::node::wire::FLAG_CE != 0,
inner_timestamp_ms,
fsp_payload_offset: meta.fsp_payload_offset,
fsp_payload_len: meta.fsp_payload_len,
trace_enqueued_at: None,
};
return Some(DecryptWorkerJobAction::FspJob(fsp_job));
}
let event = DecryptWorkerEvent::Plaintext(fallback);
Some(DecryptWorkerJobAction::Output(DecryptWorkerOutput {
fallback_tx,
event,
direct_delivery: None,
}))
}
#[cfg(test)]
fn handle_job_output(
&mut self,
idx: usize,
job: DecryptJob,
) -> Result<Option<DecryptWorkerOutput>, Box<dyn std::error::Error + Send + Sync>> {
let actions = self.handle_job_action(idx, job)?;
Ok(self.handle_job_actions_output(idx, actions))
}
fn fmp_receive_order_window_available(&self, session_key: DecryptSessionKey) -> bool {
self.sessions
.get(&session_key)
.is_none_or(OwnedSessionState::can_issue_fmp_receive_ticket)
}
#[cfg(test)]
fn handle_job_actions_output(
&mut self,
idx: usize,
actions: DecryptWorkerJobActions,
) -> Option<DecryptWorkerOutput> {
let mut first_output = None;
actions.for_each(|action| {
let outputs = match action {
DecryptWorkerJobAction::Output(output) => vec![output],
DecryptWorkerJobAction::FspJob(job) => self.dispatch_or_handle_fsp_job(idx, job),
};
for output in outputs {
if first_output.is_none() {
first_output = Some(output);
} else {
let _ = output.send();
}
}
});
first_output
}
#[cfg(test)]
fn contains_session(&self, session_key: DecryptSessionKey) -> bool {
self.sessions.contains_key(&session_key)
}
#[cfg(test)]
fn fmp_replay_highest(&self, session_key: DecryptSessionKey) -> Option<u64> {
self.sessions
.get(&session_key)
.map(|state| state.fmp_replay.highest())
}
}