fn record_decrypt_worker_bulk_drop_count(worker: usize, count: usize) {
    crate::perf_profile::record_event_count(
        crate::perf_profile::Event::DecryptWorkerQueueFull,
        count as u64,
    );
    crate::perf_profile::record_event_count(
        crate::perf_profile::Event::DecryptWorkerBulkDropped,
        count as u64,
    );
    static FULL_COUNT: AtomicU64 = AtomicU64::new(0);
    let n = FULL_COUNT.fetch_add(count as u64, Ordering::Relaxed);
    if n < 8 || n.is_multiple_of(10000) {
        warn!(
            worker,
            drops = n + count as u64,
            dropped = count,
            "DecryptWorker bulk channel full; dropping inbound packets"
        );
    }
}

fn try_reserve_bulk_packets_with_previous(
    counter: &AtomicUsize,
    capacity: usize,
    count: usize,
) -> Option<usize> {
    if count == 0 {
        return Some(counter.load(Ordering::Relaxed));
    }

    counter
        .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
            current.checked_add(count).filter(|next| *next <= capacity)
        })
        .ok()
}

fn try_reserve_bulk_packets(counter: &AtomicUsize, capacity: usize, count: usize) -> bool {
    try_reserve_bulk_packets_with_previous(counter, capacity, count).is_some()
}

fn try_reserve_bulk_packets_partial(
    counter: &AtomicUsize,
    capacity: usize,
    count: usize,
) -> usize {
    if count == 0 {
        return 0;
    }

    let mut reserved = 0;
    let result = counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
        let available = capacity.saturating_sub(current);
        if available == 0 {
            reserved = 0;
            return None;
        }
        reserved = count.min(available);
        current
            .checked_add(reserved)
            .filter(|next| *next <= capacity)
    });

    if result.is_ok() { reserved } else { 0 }
}

fn release_bulk_packets(counter: &AtomicUsize, count: usize) {
    if count == 0 {
        return;
    }

    let previous = counter.fetch_sub(count, Ordering::Relaxed);
    debug_assert!(
        previous >= count,
        "decrypt worker bulk job accounting underflow: previous={previous}, release={count}"
    );
}

fn record_decrypt_worker_priority_drop(worker: usize, kind: &'static str) {
    crate::perf_profile::record_event(crate::perf_profile::Event::DecryptWorkerQueueFull);
    crate::perf_profile::record_event(crate::perf_profile::Event::DecryptWorkerPriorityDropped);
    static FULL_COUNT: AtomicU64 = AtomicU64::new(0);
    let n = FULL_COUNT.fetch_add(1, Ordering::Relaxed);
    if n < 8 || n.is_multiple_of(10000) {
        warn!(
            worker,
            kind,
            drops = n + 1,
            "DecryptWorker priority channel full; dropping inbound item"
        );
    }
}

fn record_decrypt_worker_return_drop_count(
    event: crate::perf_profile::Event,
    lane: DecryptWorkerLane,
    count: usize,
) {
    crate::perf_profile::record_event_count(event, count as u64);
    static FULL_COUNT: AtomicU64 = AtomicU64::new(0);
    let n = FULL_COUNT.fetch_add(count as u64, Ordering::Relaxed);
    if n < 8 || n.is_multiple_of(10000) {
        warn!(
            ?lane,
            drops = n + count as u64,
            dropped = count,
            "DecryptWorker return channel full; dropping worker event"
        );
    }
}

fn run_worker(
    idx: usize,
    pool: DecryptWorkerPool,
    control_rx: Receiver<WorkerMsg>,
    priority_rx: Receiver<WorkerMsg>,
    fmp_aead_completion_rx: Receiver<FmpAeadCompletionBatch>,
    fsp_aead_completion_rx: Receiver<FspAeadCompletionBatch>,
    bulk_rx: Receiver<DecryptWorkerBulkItem>,
    bulk_queued_packets: Arc<AtomicUsize>,
) {
    trace!(worker = idx, "FMP+FSP decrypt worker thread starting");

    let mut shard = DecryptWorkerShard::new(pool);
    let mut plaintext_batch = DecryptPlaintextFallbackBatch::new();

    loop {
        drain_worker_queues(
            idx,
            &mut shard,
            &control_rx,
            &priority_rx,
            &fmp_aead_completion_rx,
            &fsp_aead_completion_rx,
            &bulk_rx,
            &bulk_queued_packets,
            &mut plaintext_batch,
        );
        match recv_worker_item_biased(
            &control_rx,
            &priority_rx,
            &fmp_aead_completion_rx,
            &fsp_aead_completion_rx,
            &bulk_rx,
        ) {
            DecryptWorkerQueueItem::Control(msg) => {
                crate::perf_profile::record_decrypt_worker_select_control();
                let mut batch_stats = DecryptWorkerBatchStats::default();
                batch_stats.add_msg(&msg);
                shard.handle_msg(idx, msg);
                batch_stats.record(idx);
            }
            DecryptWorkerQueueItem::Priority(msg) => {
                crate::perf_profile::record_decrypt_worker_select_priority();
                let mut batch_stats = DecryptWorkerBatchStats::default();
                batch_stats.add_msg(&msg);
                shard.handle_msg(idx, msg);
                batch_stats.record(idx);
            }
            DecryptWorkerQueueItem::FmpAeadCompletion(completion) => {
                crate::perf_profile::record_decrypt_worker_select_fmp_completion();
                shard.handle_fmp_aead_completion_batch_msg(idx, completion, &mut plaintext_batch);
                plaintext_batch.flush();
            }
            DecryptWorkerQueueItem::FspAeadCompletion(completions) => {
                crate::perf_profile::record_decrypt_worker_select_fsp_completion(
                    completions.len(),
                );
                shard.handle_fsp_aead_completion_batch_msg(idx, completions, &mut plaintext_batch);
                plaintext_batch.flush();
            }
            DecryptWorkerQueueItem::Bulk(item) => {
                crate::perf_profile::record_decrypt_worker_select_bulk(item.packet_count());
                release_bulk_packets(&bulk_queued_packets, item.packet_count());
                let mut batch_stats = DecryptWorkerBatchStats::default();
                batch_stats.add_bulk_item(&item);
                handle_bulk_item(
                    idx,
                    &mut shard,
                    &control_rx,
                    &priority_rx,
                    &fmp_aead_completion_rx,
                    &fsp_aead_completion_rx,
                    item,
                    &mut plaintext_batch,
                    &mut batch_stats,
                );
                plaintext_batch.flush();
                batch_stats.record(idx);
            }
            DecryptWorkerQueueItem::Closed => {
                drain_worker_queues(
                    idx,
                    &mut shard,
                    &control_rx,
                    &priority_rx,
                    &fmp_aead_completion_rx,
                    &fsp_aead_completion_rx,
                    &bulk_rx,
                    &bulk_queued_packets,
                    &mut plaintext_batch,
                );
                break;
            }
        }
    }
    trace!(worker = idx, "FMP+FSP decrypt worker thread exiting");
}

#[allow(clippy::large_enum_variant)]
enum DecryptWorkerQueueItem {
    Control(WorkerMsg),
    Priority(WorkerMsg),
    FmpAeadCompletion(FmpAeadCompletionBatch),
    FspAeadCompletion(FspAeadCompletionBatch),
    Bulk(DecryptWorkerBulkItem),
    Closed,
}

#[allow(clippy::large_enum_variant)]
enum DecryptWorkerAeadCompletion {
    Fmp(FmpAeadCompletionBatch),
    Fsp(FspAeadCompletionBatch),
}

fn try_recv_aead_completion_fair(
    fmp_aead_completion_rx: &Receiver<FmpAeadCompletionBatch>,
    fsp_aead_completion_rx: &Receiver<FspAeadCompletionBatch>,
) -> Option<DecryptWorkerAeadCompletion> {
    let fmp_backlog_packets = fmp_aead_completion_rx
        .len()
        .saturating_mul(fmp_aead_completion_batch_max());
    let fsp_backlog_packets = fsp_aead_completion_rx
        .len()
        .saturating_mul(fsp_aead_completion_batch_max());
    let prefer_fsp = fsp_backlog_packets > fmp_backlog_packets;
    if prefer_fsp {
        if let Ok(completion) = fsp_aead_completion_rx.try_recv() {
            return Some(DecryptWorkerAeadCompletion::Fsp(completion));
        }
        if let Ok(completion) = fmp_aead_completion_rx.try_recv() {
            return Some(DecryptWorkerAeadCompletion::Fmp(completion));
        }
    } else {
        if let Ok(completion) = fmp_aead_completion_rx.try_recv() {
            return Some(DecryptWorkerAeadCompletion::Fmp(completion));
        }
        if let Ok(completion) = fsp_aead_completion_rx.try_recv() {
            return Some(DecryptWorkerAeadCompletion::Fsp(completion));
        }
    }
    None
}

fn handle_aead_completion(
    idx: usize,
    shard: &mut DecryptWorkerShard,
    completion: DecryptWorkerAeadCompletion,
    plaintext_batch: &mut DecryptPlaintextFallbackBatch,
) -> usize {
    match completion {
        DecryptWorkerAeadCompletion::Fmp(completion) => {
            let count = completion.len();
            shard.handle_fmp_aead_completion_batch_msg(idx, completion, plaintext_batch);
            count
        }
        DecryptWorkerAeadCompletion::Fsp(completions) => {
            let count = completions.len();
            shard.handle_fsp_aead_completion_batch_msg(idx, completions, plaintext_batch);
            count
        }
    }
}

fn drain_aead_completions_for_bulk_item(
    idx: usize,
    shard: &mut DecryptWorkerShard,
    fmp_aead_completion_rx: &Receiver<FmpAeadCompletionBatch>,
    fsp_aead_completion_rx: &Receiver<FspAeadCompletionBatch>,
    plaintext_batch: &mut DecryptPlaintextFallbackBatch,
    remaining_budget: &mut usize,
) {
    let started_with_budget = *remaining_budget;
    let mut drained_packets = 0usize;
    let mut drained_messages = 0usize;
    while *remaining_budget > 0 {
        let Some(completion) =
            try_recv_aead_completion_fair(fmp_aead_completion_rx, fsp_aead_completion_rx)
        else {
            break;
        };
        let handled = handle_aead_completion(idx, shard, completion, plaintext_batch);
        drained_packets = drained_packets.saturating_add(handled.max(1));
        drained_messages = drained_messages.saturating_add(1);
        *remaining_budget = remaining_budget.saturating_sub(handled.max(1));
    }
    crate::perf_profile::record_decrypt_worker_bulk_interleave_aead_completion(
        drained_messages,
        drained_packets,
    );
    if started_with_budget > 0
        && *remaining_budget == 0
        && (!fmp_aead_completion_rx.is_empty() || !fsp_aead_completion_rx.is_empty())
    {
        crate::perf_profile::record_decrypt_worker_bulk_interleave_budget_exhausted();
    }
}

fn recv_worker_item_biased(
    control_rx: &Receiver<WorkerMsg>,
    priority_rx: &Receiver<WorkerMsg>,
    fmp_aead_completion_rx: &Receiver<FmpAeadCompletionBatch>,
    fsp_aead_completion_rx: &Receiver<FspAeadCompletionBatch>,
    bulk_rx: &Receiver<DecryptWorkerBulkItem>,
) -> DecryptWorkerQueueItem {
    if let Ok(msg) = control_rx.try_recv() {
        return DecryptWorkerQueueItem::Control(msg);
    }
    if let Ok(msg) = priority_rx.try_recv() {
        return DecryptWorkerQueueItem::Priority(msg);
    }
    if let Some(completion) =
        try_recv_aead_completion_fair(fmp_aead_completion_rx, fsp_aead_completion_rx)
    {
        return match completion {
            DecryptWorkerAeadCompletion::Fmp(completion) => {
                DecryptWorkerQueueItem::FmpAeadCompletion(completion)
            }
            DecryptWorkerAeadCompletion::Fsp(completion) => {
                DecryptWorkerQueueItem::FspAeadCompletion(completion)
            }
        };
    }
    crossbeam_channel::select_biased! {
        recv(control_rx) -> msg => match msg {
            Ok(msg) => DecryptWorkerQueueItem::Control(msg),
            Err(_) => DecryptWorkerQueueItem::Closed,
        },
        recv(priority_rx) -> msg => match msg {
            Ok(msg) => DecryptWorkerQueueItem::Priority(msg),
            Err(_) => DecryptWorkerQueueItem::Closed,
        },
        recv(fmp_aead_completion_rx) -> completion => match completion {
            Ok(completion) => DecryptWorkerQueueItem::FmpAeadCompletion(completion),
            Err(_) => DecryptWorkerQueueItem::Closed,
        },
        recv(fsp_aead_completion_rx) -> completion => match completion {
            Ok(completion) => DecryptWorkerQueueItem::FspAeadCompletion(completion),
            Err(_) => DecryptWorkerQueueItem::Closed,
        },
        recv(bulk_rx) -> item => match item {
            Ok(item) => DecryptWorkerQueueItem::Bulk(item),
            Err(_) => DecryptWorkerQueueItem::Closed,
        },
    }
}

fn drain_worker_queues(
    idx: usize,
    shard: &mut DecryptWorkerShard,
    control_rx: &Receiver<WorkerMsg>,
    priority_rx: &Receiver<WorkerMsg>,
    fmp_aead_completion_rx: &Receiver<FmpAeadCompletionBatch>,
    fsp_aead_completion_rx: &Receiver<FspAeadCompletionBatch>,
    bulk_rx: &Receiver<DecryptWorkerBulkItem>,
    bulk_queued_packets: &AtomicUsize,
    plaintext_batch: &mut DecryptPlaintextFallbackBatch,
) {
    let mut batch_stats = DecryptWorkerBatchStats::default();
    while let Ok(msg) = control_rx.try_recv() {
        crate::perf_profile::record_decrypt_worker_drain_control();
        batch_stats.add_msg(&msg);
        shard.handle_msg(idx, msg);
    }
    while let Ok(msg) = priority_rx.try_recv() {
        crate::perf_profile::record_decrypt_worker_drain_priority();
        batch_stats.add_msg(&msg);
        shard.handle_msg(idx, msg);
    }
    let mut drained_completion_packets = 0usize;
    let mut drained_bulk_jobs = 0;
    while drained_bulk_jobs < DECRYPT_WORKER_BULK_BURST_BUDGET {
        if let Ok(msg) = control_rx.try_recv() {
            plaintext_batch.flush();
            crate::perf_profile::record_decrypt_worker_drain_control();
            batch_stats.add_msg(&msg);
            shard.handle_msg(idx, msg);
            continue;
        }
        if let Ok(msg) = priority_rx.try_recv() {
            plaintext_batch.flush();
            crate::perf_profile::record_decrypt_worker_drain_priority();
            batch_stats.add_msg(&msg);
            shard.handle_msg(idx, msg);
            continue;
        }
        if drained_completion_packets < DECRYPT_WORKER_AEAD_COMPLETION_DRAIN_BUDGET {
            if let Some(completion) =
                try_recv_aead_completion_fair(fmp_aead_completion_rx, fsp_aead_completion_rx)
            {
                let handled = handle_aead_completion(idx, shard, completion, plaintext_batch);
                drained_completion_packets =
                    drained_completion_packets.saturating_add(handled.max(1));
                crate::perf_profile::record_decrypt_worker_drain_aead_completion(1, handled);
                continue;
            }
        }
        match bulk_rx.try_recv() {
            Ok(item) => {
                crate::perf_profile::record_decrypt_worker_drain_bulk(item.packet_count());
                release_bulk_packets(bulk_queued_packets, item.packet_count());
                batch_stats.add_bulk_item(&item);
                drained_bulk_jobs += handle_bulk_item(
                    idx,
                    shard,
                    control_rx,
                    priority_rx,
                    fmp_aead_completion_rx,
                    fsp_aead_completion_rx,
                    item,
                    plaintext_batch,
                    &mut batch_stats,
                );
            }
            Err(_) => break,
        }
    }
    plaintext_batch.flush();
    batch_stats.record(idx);
}

fn wait_for_fmp_receive_order_window(
    idx: usize,
    shard: &mut DecryptWorkerShard,
    control_rx: &Receiver<WorkerMsg>,
    priority_rx: &Receiver<WorkerMsg>,
    fmp_aead_completion_rx: &Receiver<FmpAeadCompletionBatch>,
    session_key: DecryptSessionKey,
    plaintext_batch: &mut DecryptPlaintextFallbackBatch,
    batch_stats: &mut DecryptWorkerBatchStats,
) -> bool {
    if !shard.pool.fmp_aead_helpers_enabled() {
        return true;
    }

    let mut wait_started_at = None;
    while !shard.fmp_receive_order_window_available(session_key) {
        if wait_started_at.is_none() {
            wait_started_at = crate::perf_profile::stamp();
        }
        plaintext_batch.flush();
        crossbeam_channel::select_biased! {
            recv(control_rx) -> msg => match msg {
                Ok(msg) => {
                    crate::perf_profile::record_decrypt_worker_drain_control();
                    batch_stats.add_msg(&msg);
                    shard.handle_msg(idx, msg);
                }
                Err(_) => {
                    record_fmp_receive_order_window_wait(wait_started_at);
                    return false;
                }
            },
            recv(priority_rx) -> msg => match msg {
                Ok(msg) => {
                    crate::perf_profile::record_decrypt_worker_drain_priority();
                    batch_stats.add_msg(&msg);
                    shard.handle_msg(idx, msg);
                }
                Err(_) => {
                    record_fmp_receive_order_window_wait(wait_started_at);
                    return false;
                }
            },
            recv(fmp_aead_completion_rx) -> completion => match completion {
                Ok(completion) => {
                    shard.handle_fmp_aead_completion_batch_msg(idx, completion, plaintext_batch);
                }
                Err(_) => {
                    record_fmp_receive_order_window_wait(wait_started_at);
                    return false;
                }
            },
        }
    }
    record_fmp_receive_order_window_wait(wait_started_at);
    true
}

#[inline]
fn record_fmp_receive_order_window_wait(wait_started_at: Option<crate::perf_profile::TraceStamp>) {
    if wait_started_at.is_some() {
        crate::perf_profile::record_since_count(
            crate::perf_profile::Stage::FmpReceiveOrderWindowWait,
            wait_started_at,
            1,
        );
    }
}

#[inline]
fn record_fsp_worker_bulk_input_head_wait(job: &FspDecryptJob) {
    crate::perf_profile::record_since_count(
        crate::perf_profile::Stage::DecryptFspWorkerBulkInputHeadWait,
        job.trace_enqueued_at,
        1,
    );
}

#[inline]
fn record_fsp_worker_bulk_input_head_wait_batch(jobs: &[FspDecryptJob]) {
    for job in jobs {
        record_fsp_worker_bulk_input_head_wait(job);
    }
}

#[inline]
fn record_fsp_worker_bulk_input_tail_wait(
    item_started_at: Option<crate::perf_profile::TraceStamp>,
) {
    crate::perf_profile::record_since_count(
        crate::perf_profile::Stage::DecryptFspWorkerBulkInputTailWait,
        item_started_at,
        1,
    );
}

#[inline]
fn record_decrypt_worker_bulk_input_head_wait(
    queued_at: Option<crate::perf_profile::TraceStamp>,
    count: usize,
) {
    crate::perf_profile::record_decrypt_worker_bulk_input_wait(queued_at, count as u64);
}

#[inline]
fn record_decrypt_worker_bulk_input_tail_wait(
    item_started_at: Option<crate::perf_profile::TraceStamp>,
) {
    crate::perf_profile::record_since_count(
        crate::perf_profile::Stage::DecryptWorkerBulkInputTailWait,
        item_started_at,
        1,
    );
}

#[inline]
fn record_decrypt_worker_bulk_item_service(
    item_started_at: Option<crate::perf_profile::TraceStamp>,
    count: usize,
) {
    crate::perf_profile::record_since_count(
        crate::perf_profile::Stage::DecryptWorkerBulkItemService,
        item_started_at,
        count as u64,
    );
}

fn send_fsp_aead_open_completion_batch(
    idx: usize,
    completion_tx: Sender<FspAeadCompletionBatch>,
    batch: FspAeadCompletionBatch,
) {
    if completion_tx.send(batch).is_err() {
        debug!(
            worker = idx,
            "FSP AEAD opener completion owner gone; dropping completion"
        );
    }
}

fn complete_fsp_aead_open_jobs(idx: usize, jobs: Vec<FspAeadOpenJob>) {
    let completion_batch_max = fsp_aead_completion_batch_max();
    let mut current_tx: Option<Sender<FspAeadCompletionBatch>> = None;
    let mut current_source_addr = None;
    let mut current_receive_order_id = None;
    let mut current_batch: Option<FspAeadCompletionBatch> = None;

    for mut job in jobs {
        let Some(completion_tx) = job.completion_tx.take() else {
            continue;
        };
        let source_addr = job.source_addr;
        let receive_order_id = job.receive_order_id;
        let same_batch = current_tx
            .as_ref()
            .is_some_and(|tx| tx.same_channel(&completion_tx))
            && current_source_addr == Some(source_addr)
            && current_receive_order_id == Some(receive_order_id)
            && current_batch
                .as_ref()
                .is_some_and(|batch| batch.len() < completion_batch_max);

        if !same_batch {
            if let (Some(tx), Some(batch)) = (current_tx.take(), current_batch.take()) {
                send_fsp_aead_open_completion_batch(idx, tx, batch);
            }
            current_tx = Some(completion_tx);
            current_source_addr = Some(source_addr);
            current_receive_order_id = Some(receive_order_id);
            current_batch = Some(FspAeadCompletionBatch::one(job.into_completion()));
            continue;
        }

        let Some(batch) = current_batch.as_mut() else {
            unreachable!("same_batch requires an active completion batch")
        };
        batch.push(job.into_completion());
    }

    if let (Some(tx), Some(batch)) = (current_tx, current_batch) {
        send_fsp_aead_open_completion_batch(idx, tx, batch);
    }
}

fn flush_fsp_open_batcher(
    idx: usize,
    shard: &mut DecryptWorkerShard,
    plaintext_batch: &mut DecryptPlaintextFallbackBatch,
    fsp_open_batcher: &mut FspAeadOpenJobBatcher,
) {
    let returned = fsp_open_batcher.flush(&shard.pool);
    if !returned.is_empty() {
        shard.drop_returned_fsp_aead_open_jobs(idx, returned, plaintext_batch);
    }
}

fn handle_bulk_item(
    idx: usize,
    shard: &mut DecryptWorkerShard,
    control_rx: &Receiver<WorkerMsg>,
    priority_rx: &Receiver<WorkerMsg>,
    fmp_aead_completion_rx: &Receiver<FmpAeadCompletionBatch>,
    fsp_aead_completion_rx: &Receiver<FspAeadCompletionBatch>,
    item: DecryptWorkerBulkItem,
    plaintext_batch: &mut DecryptPlaintextFallbackBatch,
    batch_stats: &mut DecryptWorkerBatchStats,
) -> usize {
    match item {
        DecryptWorkerBulkItem::Job(job) => {
            let item_service_started_at = crate::perf_profile::stamp();
            let item_started_at = crate::perf_profile::stamp();
            record_decrypt_worker_bulk_input_head_wait(job.trace_enqueued_at, 1);
            record_decrypt_worker_bulk_input_tail_wait(item_started_at);
            shard.handle_bulk_job_msg(idx, job, plaintext_batch);
            record_decrypt_worker_bulk_item_service(item_service_started_at, 1);
            1
        }
        DecryptWorkerBulkItem::FspJob(job) => {
            let item_started_at = crate::perf_profile::stamp();
            record_fsp_worker_bulk_input_head_wait(&job);
            record_fsp_worker_bulk_input_tail_wait(item_started_at);
            shard.handle_bulk_fsp_job_msg(idx, job, plaintext_batch);
            1
        }
        DecryptWorkerBulkItem::FspAeadOpen(job) => {
            let item_service_started_at = crate::perf_profile::stamp();
            complete_fsp_aead_open_jobs(idx, vec![job]);
            record_decrypt_worker_bulk_item_service(item_service_started_at, 1);
            1
        }
        DecryptWorkerBulkItem::FspAeadOpenBatch(jobs) => {
            let item_service_started_at = crate::perf_profile::stamp();
            let count = jobs.len();
            complete_fsp_aead_open_jobs(idx, jobs);
            record_decrypt_worker_bulk_item_service(item_service_started_at, count);
            count
        }
        DecryptWorkerBulkItem::Batch(jobs) => {
            let item_service_started_at = crate::perf_profile::stamp();
            let count = jobs.len();
            let item_started_at = crate::perf_profile::stamp();
            if let Some(job) = jobs.first() {
                record_decrypt_worker_bulk_input_head_wait(job.trace_enqueued_at, count);
            }
            let mut fsp_batcher = FspDecryptJobBatcher::new();
            let mut fsp_open_batcher = FspAeadOpenJobBatcher::new();
            for job in jobs {
                while let Ok(msg) = control_rx.try_recv() {
                    fsp_batcher.flush(&shard.pool);
                    flush_fsp_open_batcher(
                        idx,
                        shard,
                        plaintext_batch,
                        &mut fsp_open_batcher,
                    );
                    plaintext_batch.flush();
                    crate::perf_profile::record_decrypt_worker_drain_control();
                    batch_stats.add_msg(&msg);
                    shard.handle_msg(idx, msg);
                }
                while let Ok(msg) = priority_rx.try_recv() {
                    fsp_batcher.flush(&shard.pool);
                    flush_fsp_open_batcher(
                        idx,
                        shard,
                        plaintext_batch,
                        &mut fsp_open_batcher,
                    );
                    plaintext_batch.flush();
                    crate::perf_profile::record_decrypt_worker_drain_priority();
                    batch_stats.add_msg(&msg);
                    shard.handle_msg(idx, msg);
                }
                let mut completion_interleave_budget =
                    DECRYPT_WORKER_AEAD_COMPLETION_INTERLEAVE_BUDGET;
                drain_aead_completions_for_bulk_item(
                    idx,
                    shard,
                    fmp_aead_completion_rx,
                    fsp_aead_completion_rx,
                    plaintext_batch,
                    &mut completion_interleave_budget,
                );
                if !wait_for_fmp_receive_order_window(
                    idx,
                    shard,
                    control_rx,
                    priority_rx,
                    fmp_aead_completion_rx,
                    job.session_key,
                    plaintext_batch,
                    batch_stats,
                ) {
                    break;
                }
                record_decrypt_worker_bulk_input_tail_wait(item_started_at);
                match shard.handle_job_action(idx, job) {
                    Ok(actions) => {
                        shard.push_job_actions_output(
                            idx,
                            actions,
                            plaintext_batch,
                            Some(&mut fsp_batcher),
                            Some(&mut fsp_open_batcher),
                        );
                    }
                    Err(err) => {
                        debug!(worker = idx, error = %err, "decrypt worker job failed");
                    }
                }
            }
            fsp_batcher.flush(&shard.pool);
            flush_fsp_open_batcher(idx, shard, plaintext_batch, &mut fsp_open_batcher);
            record_decrypt_worker_bulk_item_service(item_service_started_at, count);
            count
        }
        DecryptWorkerBulkItem::FspBatch(jobs) => {
            let item_service_started_at = crate::perf_profile::stamp();
            let item_started_at = crate::perf_profile::stamp();
            record_fsp_worker_bulk_input_head_wait_batch(&jobs);
            let count = jobs.len();
            for job in jobs {
                while let Ok(msg) = control_rx.try_recv() {
                    plaintext_batch.flush();
                    crate::perf_profile::record_decrypt_worker_drain_control();
                    batch_stats.add_msg(&msg);
                    shard.handle_msg(idx, msg);
                }
                while let Ok(msg) = priority_rx.try_recv() {
                    plaintext_batch.flush();
                    crate::perf_profile::record_decrypt_worker_drain_priority();
                    batch_stats.add_msg(&msg);
                    shard.handle_msg(idx, msg);
                }
                let mut completion_interleave_budget =
                    DECRYPT_WORKER_AEAD_COMPLETION_INTERLEAVE_BUDGET;
                drain_aead_completions_for_bulk_item(
                    idx,
                    shard,
                    fmp_aead_completion_rx,
                    fsp_aead_completion_rx,
                    plaintext_batch,
                    &mut completion_interleave_budget,
                );
                record_fsp_worker_bulk_input_tail_wait(item_started_at);
                shard.handle_bulk_fsp_job_msg(idx, job, plaintext_batch);
            }
            record_decrypt_worker_bulk_item_service(item_service_started_at, count);
            count
        }
    }
}

struct DecryptWorkerOutput {
    fallback_tx: DecryptWorkerFallbackSender,
    event: DecryptWorkerEvent,
    direct_delivery: Option<PendingDirectSessionDelivery>,
}

#[allow(clippy::large_enum_variant)]
enum DecryptWorkerJobAction {
    Output(DecryptWorkerOutput),
    FspJob(FspDecryptJob),
}

#[allow(clippy::large_enum_variant)]
enum DecryptWorkerJobActions {
    None,
    One(DecryptWorkerJobAction),
    Many(Vec<DecryptWorkerJobAction>),
}

impl DecryptWorkerJobActions {
    fn one(action: DecryptWorkerJobAction) -> Self {
        Self::One(action)
    }

    fn push(&mut self, action: DecryptWorkerJobAction) {
        match std::mem::replace(self, Self::None) {
            Self::None => {
                *self = Self::One(action);
            }
            Self::One(existing) => {
                *self = Self::Many(vec![existing, action]);
            }
            Self::Many(mut actions) => {
                actions.push(action);
                *self = Self::Many(actions);
            }
        }
    }

    fn for_each(self, mut on_action: impl FnMut(DecryptWorkerJobAction)) {
        match self {
            Self::None => {}
            Self::One(action) => on_action(action),
            Self::Many(actions) => {
                for action in actions {
                    on_action(action);
                }
            }
        }
    }
}

impl DecryptWorkerOutput {
    fn send(mut self) -> bool {
        let direct_delivery = self.direct_delivery.take();
        if !self.fallback_tx.send(self.event) {
            return false;
        }
        if let Some(delivery) = direct_delivery {
            delivery.deliver();
        }
        true
    }

    fn is_batchable_bulk_plaintext(&self) -> bool {
        matches!(
            &self.event,
            DecryptWorkerEvent::Plaintext(fallback)
                if matches!(fallback.lane(), DecryptWorkerLane::Bulk)
        )
    }

    fn is_batchable_authenticated_session(&self) -> bool {
        matches!(
            (&self.event, &self.direct_delivery),
            (DecryptWorkerEvent::AuthenticatedSession(session), None)
                if matches!(session.lane, DecryptWorkerLane::Bulk)
        )
    }

    fn is_batchable_direct_endpoint(&self) -> bool {
        matches!(
            (&self.event, &self.direct_delivery),
            (
                DecryptWorkerEvent::DirectSessionCommit(commit),
                Some(delivery),
            ) if matches!(commit.lane, DecryptWorkerLane::Bulk) && delivery.is_endpoint_data()
        )
    }

    fn is_batchable_direct_ipv6(&self) -> bool {
        matches!(
            (&self.event, &self.direct_delivery),
            (
                DecryptWorkerEvent::DirectSessionCommit(commit),
                Some(delivery),
            ) if matches!(commit.lane, DecryptWorkerLane::Bulk) && delivery.is_ipv6_packet()
        )
    }

    fn is_batchable_direct_data(&self) -> bool {
        matches!(
            (&self.event, &self.direct_delivery),
            (DecryptWorkerEvent::DirectSessionData(direct), None)
                if matches!(direct.lane, DecryptWorkerLane::Bulk)
        )
    }
}
