const DATAPLANE_DEFERRED_RAW_INGRESS_MAX_RETRIES: u8 = 8;

#[derive(Debug)]
pub(crate) struct DataplaneTurnDriver {
    mover: Dataplane,
    prepared_work: Vec<PreparedCryptoWork>,
    completion_batches: Vec<CryptoCompletionBatch>,
    raw_ingress_drops: Vec<DataplaneRawIngressDrop>,
    output_drops: Vec<DataplaneOutputDrop>,
    outputs: Vec<PacketOutput>,
    output_rewrite_buffer: Vec<PacketOutput>,
    raw_socket_packets: Vec<SocketPacket>,
    retired_outbound_packets: Vec<OutboundPacket>,
    transport_output: DataplaneTransportSendGroups,
    drops: Vec<PacketDrop>,
    fmp_ingress_receipts: Vec<DataplaneFmpIngressReceipt>,
    fmp_link_ingress: Vec<DataplaneFmpLinkIngress>,
    fsp_coord_warmups: Vec<DataplaneFspCoordWarmup>,
    fsp_local_session_ingress: Vec<DataplaneFspLocalSessionIngress>,
    fsp_authenticated_ingress: DataplaneFspAuthenticatedIngress,
}

struct DataplaneLiveAdmissionResult {
    summary: DataplaneRuntimeSummary,
    outbound_buffers: DataplaneRouteTableOutboundBuffers,
    endpoint_drained: usize,
    tun_drained: usize,
}

impl DataplaneLiveAdmissionResult {
    fn has_activity(&self) -> bool {
        self.summary.has_activity()
            || self.endpoint_drained > 0
            || self.tun_drained > 0
            || self.outbound_buffers.has_activity()
    }
}

struct DataplaneLiveAdmissionRequest<'a, RI> {
    summary: DataplaneRuntimeSummary,
    fast_ingress: Option<DataplaneFastIngressBatch>,
    raw_ingress: &'a mut RI,
    routes: &'a mut DataplaneLiveRouteTable,
    raw_ingress_limit: usize,
    endpoint_data_rx: &'a mut EndpointDataBatchRx,
    endpoint_limit: usize,
    tun_outbound_rx: &'a mut TunOutboundRx,
    tun_limit: usize,
    outbound_firsts: DataplaneLiveOutboundFirsts,
    deferred_raw_ingress: &'a mut std::collections::VecDeque<(DataplaneRawIngress, u8)>,
}

struct DataplaneLiveOutputRequest<'a> {
    summary: DataplaneRuntimeSummary,
    routes: &'a mut DataplaneLiveRouteTable,
    endpoint_tx: &'a EndpointEventSender,
    transports: &'a HashMap<TransportId, TransportHandle>,
    deferred_raw_ingress: &'a mut std::collections::VecDeque<(DataplaneRawIngress, u8)>,
    crypto_limit: usize,
    collect_transport_sent_receipts: bool,
    crypto_worker: &'a mut DataplaneAeadWorkerPool,
    transport_send_batch_packets: usize,
}

struct DataplaneLiveFinishRequest<'a> {
    admission: DataplaneLiveAdmissionResult,
    routes: &'a mut DataplaneLiveRouteTable,
    endpoint_tx: &'a EndpointEventSender,
    transports: &'a HashMap<TransportId, TransportHandle>,
    crypto_limit: usize,
    collect_transport_sent_receipts: bool,
    crypto_worker: &'a mut DataplaneAeadWorkerPool,
    transport_send_batch_packets: usize,
    deferred_raw_ingress: &'a mut std::collections::VecDeque<(DataplaneRawIngress, u8)>,
    deferred_endpoint_data_batches: &'a mut Vec<NodeEndpointDataBatch>,
    deferred_tun_packets: &'a mut Vec<Vec<u8>>,
}

struct DataplaneLivePumpRequest<'a, RI> {
    summary: DataplaneRuntimeSummary,
    crypto_worker: &'a mut DataplaneAeadWorkerPool,
    fast_ingress: Option<DataplaneFastIngressBatch>,
    raw_ingress: &'a mut RI,
    routes: &'a mut DataplaneLiveRouteTable,
    raw_ingress_limit: usize,
    endpoint_data_rx: &'a mut EndpointDataBatchRx,
    endpoint_limit: usize,
    tun_outbound_rx: &'a mut TunOutboundRx,
    tun_limit: usize,
    outbound_firsts: DataplaneLiveOutboundFirsts,
    deferred_endpoint_data_batches: &'a mut Vec<NodeEndpointDataBatch>,
    deferred_tun_packets: &'a mut Vec<Vec<u8>>,
    deferred_raw_ingress: &'a mut std::collections::VecDeque<(DataplaneRawIngress, u8)>,
    endpoint_tx: &'a EndpointEventSender,
    transports: &'a HashMap<TransportId, TransportHandle>,
    crypto_limit: usize,
    transport_send_batch_packets: usize,
}

impl DataplaneTurnDriver {
    pub(crate) fn new(config: AdmissionConfig) -> Self {
        Self {
            mover: Dataplane::new(config),
            prepared_work: Vec::new(),
            completion_batches: Vec::new(),
            raw_ingress_drops: Vec::new(),
            output_drops: Vec::new(),
            outputs: Vec::new(),
            output_rewrite_buffer: Vec::new(),
            raw_socket_packets: Vec::new(),
            retired_outbound_packets: Vec::new(),
            transport_output: DataplaneTransportSendGroups::new(),
            drops: Vec::new(),
            fmp_ingress_receipts: Vec::new(),
            fmp_link_ingress: Vec::new(),
            fsp_coord_warmups: Vec::new(),
            fsp_local_session_ingress: Vec::new(),
            fsp_authenticated_ingress: DataplaneFspAuthenticatedIngress::default(),
        }
    }

    pub(crate) fn register_owner(&mut self, owner: OwnerId, config: OwnerConfig) {
        self.mover.register_owner(owner, config);
    }

    pub(crate) fn unregister_owner(&mut self, owner: OwnerId) {
        self.mover.unregister_owner(owner);
    }

    pub(crate) fn has_runnable_work(&self) -> bool {
        self.mover.has_runnable_work()
    }

    pub(crate) fn has_owner(&self, owner: OwnerId) -> bool {
        self.mover.has_owner(owner)
    }

    pub(crate) fn fsp_owner_destinations(&self) -> Vec<NodeAddr> {
        self.mover.fsp_owner_destinations()
    }

    pub(crate) fn owner_active_path(&self, owner: OwnerId) -> Option<TransportPath> {
        self.mover.owner_active_path(owner)
    }

    pub(crate) fn owner_fsp_next_hop(&self, owner: OwnerId) -> Option<NodeAddr> {
        self.mover.owner_fsp_next_hop(owner)
    }

    pub(crate) fn owner_fsp_activity(
        &self,
        owner: OwnerId,
    ) -> Option<DataplaneFspOwnerActivity> {
        self.mover.owner_fsp_activity(owner)
    }

    pub(crate) fn owner_has_fsp_pending_receive_epoch(
        &self,
        owner: OwnerId,
        received_k_bit: bool,
    ) -> bool {
        self.mover
            .owner_has_fsp_pending_receive_epoch(owner, received_k_bit)
    }

    pub(crate) fn owner_has_fmp_pending_receive_epoch(
        &self,
        owner: OwnerId,
        received_k_bit: bool,
    ) -> bool {
        self.mover
            .owner_has_fmp_pending_receive_epoch(owner, received_k_bit)
    }

    pub(crate) fn owner_fsp_mmp_snapshot(
        &self,
        owner: OwnerId,
    ) -> Option<DataplaneFspMmpSnapshot> {
        self.mover.owner_fsp_mmp_snapshot(owner)
    }

    pub(crate) fn owner_fsp_send_context(
        &self,
        owner: OwnerId,
    ) -> Option<DataplaneFspSendContext> {
        self.mover.owner_fsp_send_context(owner)
    }

    pub(crate) fn owner_fmp_send_context(
        &self,
        owner: OwnerId,
    ) -> Option<DataplaneFmpSendContext> {
        self.mover.owner_fmp_send_context(owner)
    }

    pub(crate) fn owner_fmp_link_metrics(
        &self,
        owner: OwnerId,
        now: std::time::Instant,
    ) -> Option<DataplaneFmpLinkMetrics> {
        self.mover.owner_fmp_link_metrics(owner, now)
    }

    pub(crate) fn owner_fmp_link_cost(&self, owner: OwnerId) -> Option<f64> {
        self.mover.owner_fmp_link_cost(owner)
    }

    pub(crate) fn owner_fmp_has_srtt(&self, owner: OwnerId) -> bool {
        self.mover.owner_fmp_has_srtt(owner)
    }

    pub(crate) fn collect_fmp_mmp_reports(
        &mut self,
        now: std::time::Instant,
    ) -> DataplaneFmpMmpReportBatch {
        self.mover.collect_fmp_mmp_reports(now)
    }

    pub(crate) fn collect_fsp_mmp_reports(
        &mut self,
        now: std::time::Instant,
    ) -> DataplaneFspMmpReportBatch {
        self.mover.collect_fsp_mmp_reports(now)
    }

    pub(crate) fn record_fsp_mmp_send_result(
        &mut self,
        owner: OwnerId,
        success: bool,
    ) -> Option<DataplaneFspMmpReportingResumed> {
        self.mover.record_fsp_mmp_send_result(owner, success)
    }

    pub(crate) fn seed_fsp_path_mtu(
        &mut self,
        owner: OwnerId,
        path_mtu: u16,
    ) -> Result<(), DataplaneFspMmpSkip> {
        self.mover.seed_fsp_path_mtu(owner, path_mtu)
    }

    pub(crate) fn process_fsp_mmp_receiver_report(
        &mut self,
        owner: OwnerId,
        rr: &crate::mmp::report::ReceiverReport,
        last_outbound_next_hop: Option<NodeAddr>,
        now_ms: u64,
        now: std::time::Instant,
        min_loss_sample: u64,
    ) -> Result<DataplaneFspReceiverReportResult, DataplaneFspMmpSkip> {
        self.mover.process_fsp_mmp_receiver_report(
            owner,
            rr,
            last_outbound_next_hop,
            now_ms,
            now,
            min_loss_sample,
        )
    }

    pub(crate) fn apply_fsp_path_mtu_signal(
        &mut self,
        owner: OwnerId,
        path_mtu: u16,
        now: std::time::Instant,
    ) -> Result<DataplaneFspPathMtuApplyResult, DataplaneFspMmpSkip> {
        self.mover.apply_fsp_path_mtu_signal(owner, path_mtu, now)
    }

    pub(crate) fn min_fsp_rx_age_for_next_hop(
        &self,
        next_hop: &NodeAddr,
        now_ms: u64,
    ) -> Option<u64> {
        self.mover.min_fsp_rx_age_for_next_hop(next_hop, now_ms)
    }

    pub(crate) fn min_fsp_data_rx_age_for_next_hop(
        &self,
        next_hop: &NodeAddr,
        now_ms: u64,
    ) -> Option<u64> {
        self.mover
            .min_fsp_data_rx_age_for_next_hop(next_hop, now_ms)
    }

    pub(crate) fn any_fsp_recent_outbound_without_inbound_for_next_hop(
        &self,
        next_hop: &NodeAddr,
        now_ms: u64,
        timeout_ms: u64,
    ) -> bool {
        self.mover
            .any_fsp_recent_outbound_without_inbound_for_next_hop(next_hop, now_ms, timeout_ms)
    }

    pub(crate) fn owner_mut(&mut self, owner: OwnerId) -> Option<&mut OwnerState> {
        self.mover.owner_mut(owner)
    }

    pub(crate) fn record_authenticated_fsp_session(
        &mut self,
        session: DataplaneAuthenticatedFspSession,
    ) -> Option<bool> {
        self.mover.record_authenticated_fsp_session(session)
    }

    fn record_fsp_session_ingress_activity(
        &mut self,
        ingress: &DataplaneFspSessionIngress,
    ) -> bool {
        let body_len = ingress
            .receive_sync
            .plaintext_len
            .saturating_sub(FSP_INNER_HEADER_SIZE);
        self.record_authenticated_fsp_session(DataplaneAuthenticatedFspSession::new(
            ingress.source_addr,
            ingress.previous_hop_addr,
            ingress.msg_type,
            body_len,
            ingress.receive_sync,
            ingress.activity_tick,
            std::time::Instant::now(),
        ))
        .unwrap_or(false)
    }

    pub(crate) fn record_fsp_decrypt_failure(&mut self, owner: OwnerId) -> Option<u32> {
        self.mover.record_fsp_decrypt_failure(owner)
    }

    async fn finish_aead_live_node_output_turn(
        &mut self,
        request: DataplaneLiveOutputRequest<'_>,
    ) -> DataplaneLiveNodeTurn {
        let DataplaneLiveOutputRequest {
            summary,
            routes,
            endpoint_tx,
            transports,
            deferred_raw_ingress,
            crypto_limit,
            collect_transport_sent_receipts,
            crypto_worker,
            transport_send_batch_packets,
        } = request;
        let compact_endpoint_data = endpoint_tx.direct_sink().is_some();
        let summary = self.collect_live_session_outputs(
            summary,
            routes,
            crypto_limit,
            crypto_worker,
            compact_endpoint_data,
            deferred_raw_ingress,
        );
        let mut transport_output = std::mem::take(&mut self.transport_output);
        transport_output.clear();
        let mut report = {
            let mut sink = DataplaneLiveOutputSink::new(&mut transport_output);
            let turn = self.send_collected_outputs(summary, &mut sink);
            DataplaneLiveNodeTurn::from_runtime_turn(&turn)
        };
        self.deliver_direct_endpoint_packet_batches(endpoint_tx.direct_sink());
        report.fmp_ingress_receipts = std::mem::take(&mut self.fmp_ingress_receipts);
        report.fmp_link_ingress = std::mem::take(&mut self.fmp_link_ingress);
        report.fsp_coord_warmups = std::mem::take(&mut self.fsp_coord_warmups);
        report.fsp_local_session_ingress = std::mem::take(&mut self.fsp_local_session_ingress);
        report.transport_planned = transport_output.planned_packets();
        let dropped_before = report.output_drops.len();
        let mut transport_sent_receipts = if collect_transport_sent_receipts {
            Some(&mut report.transport_sent_receipts)
        } else {
            None
        };
        report.transport_sent = {
            let _transport_send_timer = crate::perf_profile::Timer::start(
                crate::perf_profile::Stage::DataplaneTransportSend,
            );
            let groups = transport_output.take_groups_preserving_capacity();
            send_dataplane_transport_groups(
                transports,
                groups,
                &mut report.output_drops,
                transport_send_batch_packets,
                transport_sent_receipts.take(),
            )
            .await
        };
        report.transport_dropped = report.output_drops.len().saturating_sub(dropped_before);
        debug_assert_eq!(
            report.transport_planned,
            report.transport_sent + report.transport_dropped
        );
        report.summary.outputs_sent = report
            .summary
            .outputs_sent
            .saturating_sub(report.transport_dropped);
        report.summary.outputs_dropped = report
            .summary
            .outputs_dropped
            .saturating_add(report.transport_dropped);
        report.fsp_authenticated_ingress = std::mem::take(&mut self.fsp_authenticated_ingress);
        self.transport_output = transport_output;
        report
    }

    fn start_aead_completion_turn(
        &mut self,
        completions: &mut DataplaneAeadWorkerPool,
        completion_limit: usize,
        compact_endpoint_data: bool,
    ) -> DataplaneRuntimeSummary {
        self.reset_turn_buffers();
        self.drain_aead_completion_turn_into_summary(
            DataplaneRuntimeSummary::default(),
            completions,
            completion_limit,
            compact_endpoint_data,
        )
    }

    fn drain_aead_completion_turn_into_summary(
        &mut self,
        mut summary: DataplaneRuntimeSummary,
        completions: &mut DataplaneAeadWorkerPool,
        completion_limit: usize,
        compact_endpoint_data: bool,
    ) -> DataplaneRuntimeSummary {
        let _completion_timer = crate::perf_profile::Timer::start(
            crate::perf_profile::Stage::DataplaneCompletionDrain,
        );
        let completion_limit = self.completion_drain_limit(completion_limit);
        let queued = completions.drain_completion_batches_into_sink(
            completion_limit,
            &mut self.mover,
        );
        summary.completions = summary.completions.saturating_add(queued);
        self.retire_queued_completed_aead_outputs(completion_limit, compact_endpoint_data);
        self.admit_retired_outbound_packets(summary)
    }

    fn completion_drain_limit(&self, limit: usize) -> usize {
        if limit < DATAPLANE_AEAD_WORKER_FAIRNESS_PACKETS || self.mover.has_priority_pending() {
            return limit;
        }
        limit.saturating_mul(DATAPLANE_AEAD_WORKER_FAIRNESS_PACKETS)
    }

    async fn pump_aead_live_node_route_table_turn_after_completion_with_firsts<RI>(
        &mut self,
        request: DataplaneLivePumpRequest<'_, RI>,
    ) -> DataplaneLiveNodeTurn
    where
        RI: DataplaneRawIngressSource,
    {
        let DataplaneLivePumpRequest {
            summary,
            crypto_worker,
            fast_ingress,
            raw_ingress,
            routes,
            raw_ingress_limit,
            endpoint_data_rx,
            endpoint_limit,
            tun_outbound_rx,
            tun_limit,
            outbound_firsts,
            deferred_endpoint_data_batches,
            deferred_tun_packets,
            deferred_raw_ingress,
            endpoint_tx,
            transports,
            crypto_limit,
            transport_send_batch_packets,
        } = request;
        let collect_transport_sent_receipts = outbound_firsts.collect_transport_sent_receipts;
        let mut completion_report = None;
        let mut admission_summary = summary;
        let mut completion_deferred_raw_ingress = std::collections::VecDeque::new();
        // Ready completions are already owner-ordered work. Let their local
        // continuations consume this turn's bounded crypto budget before
        // admitting fresh raw/outbound packets.
        let mut remaining_crypto_limit = crypto_limit;
        if admission_summary.has_activity() {
            let report = self
                .finish_aead_live_node_output_turn(DataplaneLiveOutputRequest {
                    summary: admission_summary,
                    routes,
                    endpoint_tx,
                    transports,
                    deferred_raw_ingress: &mut completion_deferred_raw_ingress,
                    crypto_limit: remaining_crypto_limit,
                    collect_transport_sent_receipts,
                    crypto_worker: &mut *crypto_worker,
                    transport_send_batch_packets,
                })
                .await;
            remaining_crypto_limit =
                remaining_crypto_limit.saturating_sub(report.summary().dispatched());
            self.reset_turn_buffers();
            completion_report = Some(report);
            admission_summary = DataplaneRuntimeSummary::default();
        }
        let admission = self.admit_live_node_route_table_turn_with_firsts(
            DataplaneLiveAdmissionRequest {
                summary: admission_summary,
                fast_ingress,
                raw_ingress,
                routes,
                raw_ingress_limit,
                endpoint_data_rx,
                endpoint_limit,
                tun_outbound_rx,
                tun_limit,
                outbound_firsts,
                deferred_raw_ingress,
            },
        );
        deferred_raw_ingress.append(&mut completion_deferred_raw_ingress);
        if let Some(mut completion_report) = completion_report {
            if !admission.has_activity() {
                return completion_report;
            }
            let report = self
                .finish_live_node_turn_after_admission(
                    DataplaneLiveFinishRequest {
                        admission,
                        routes,
                        endpoint_tx,
                        transports,
                        crypto_limit: remaining_crypto_limit,
                        collect_transport_sent_receipts,
                        crypto_worker,
                        transport_send_batch_packets,
                        deferred_raw_ingress,
                        deferred_endpoint_data_batches,
                        deferred_tun_packets,
                    },
                )
                .await;
            completion_report.absorb(report);
            completion_report
        } else {
            self.finish_live_node_turn_after_admission(
                DataplaneLiveFinishRequest {
                    admission,
                    routes,
                    endpoint_tx,
                    transports,
                    crypto_limit: remaining_crypto_limit,
                    collect_transport_sent_receipts,
                    crypto_worker,
                    transport_send_batch_packets,
                    deferred_raw_ingress,
                    deferred_endpoint_data_batches,
                    deferred_tun_packets,
                },
            )
            .await
        }
    }

    fn admit_live_node_route_table_turn_with_firsts<RI>(
        &mut self,
        request: DataplaneLiveAdmissionRequest<'_, RI>,
    ) -> DataplaneLiveAdmissionResult
    where
        RI: DataplaneRawIngressSource,
    {
        let DataplaneLiveAdmissionRequest {
            mut summary,
            fast_ingress,
            raw_ingress,
            routes,
            raw_ingress_limit,
            endpoint_data_rx,
            endpoint_limit,
            tun_outbound_rx,
            tun_limit,
            outbound_firsts,
            deferred_raw_ingress,
        } = request;
        let admit_timer =
            crate::perf_profile::Timer::start(crate::perf_profile::Stage::DataplaneLiveAdmit);
        let trace_enabled = crate::perf_profile::enabled();
        let mut outbound_firsts = outbound_firsts;
        if let Some(packet) = outbound_firsts.initial_outbound.take() {
            self.admit_outbound_packet(packet, &mut summary);
        }

        let routed_outbound_limit = endpoint_limit.saturating_add(tun_limit);
        let reserved_outbound_limit =
            reserved_live_outbound_progress_limit(endpoint_limit, tun_limit, routed_outbound_limit);
        let mut outbound_buffers = DataplaneRouteTableOutboundBuffers::default();
        let mut endpoint_drained = 0usize;
        let mut tun_drained = 0usize;
        let mut outbound_drained = 0usize;
        let mut outbound_admitted = DataplaneOutboundAdmissionCounts::default();

        if reserved_outbound_limit > 0 {
            let mut outbound_source = DataplaneRouteTableOutboundSource::new(
                endpoint_data_rx,
                endpoint_limit,
                tun_outbound_rx,
                tun_limit,
                routes,
                &mut outbound_buffers,
            )
            .with_firsts(outbound_firsts);
            let (drained_total, endpoint, tun) = {
                let mut admission = DataplaneOutboundAdmission {
                    driver: self,
                    summary: &mut summary,
                    trace_enabled,
                    counts: &mut outbound_admitted,
                };
                outbound_source.drain_outbound_batched(reserved_outbound_limit, &mut admission)
            };
            outbound_drained = drained_total;
            endpoint_drained = endpoint_drained.saturating_add(endpoint);
            tun_drained = tun_drained.saturating_add(tun);
            outbound_firsts = outbound_source.take_firsts();
        }

        let mut raw_socket_packets = std::mem::take(&mut self.raw_socket_packets);
        raw_socket_packets.clear();
        let raw_admitted_before = if trace_enabled {
            summary.inbound_admitted
        } else {
            0
        };
        if let Some(fast_ingress) = fast_ingress {
            self.admit_fast_ingress_runs(fast_ingress, &mut summary);
        }
        {
            let raw_ingress_drops = &mut self.raw_ingress_drops;
            let deferred_available = deferred_raw_ingress.len();
            let fresh_drained = raw_ingress.drain_raw_ingress(raw_ingress_limit, |packet| {
                if let Some(socket_packet) =
                    Self::raw_ingress_socket_packet(
                        packet,
                        routes,
                        &mut summary,
                        raw_ingress_drops,
                        deferred_raw_ingress,
                        0,
                    )
                {
                    raw_socket_packets.push(socket_packet);
                }
            });
            let deferred_limit = raw_ingress_limit
                .saturating_sub(fresh_drained)
                .min(deferred_available);
            for _ in 0..deferred_limit {
                let Some((packet, retry_count)) = deferred_raw_ingress.pop_front() else {
                    break;
                };
                if let Some(socket_packet) = Self::raw_ingress_socket_packet(
                    packet,
                    routes,
                    &mut summary,
                    raw_ingress_drops,
                    deferred_raw_ingress,
                    retry_count,
                ) {
                    raw_socket_packets.push(socket_packet);
                }
            }
        }
        self.admit_socket_packets(&mut raw_socket_packets, &mut summary);
        self.raw_socket_packets = raw_socket_packets;
        if trace_enabled {
            crate::perf_profile::record_event_count(
                crate::perf_profile::Event::DataplaneLiveRawAdmitted,
                summary
                    .inbound_admitted
                    .saturating_sub(raw_admitted_before) as u64,
            );
        }

        let remaining_outbound_limit =
            routed_outbound_limit.saturating_sub(outbound_drained.min(routed_outbound_limit));
        if remaining_outbound_limit > 0 {
            let mut outbound_source = DataplaneRouteTableOutboundSource::new(
                endpoint_data_rx,
                endpoint_limit,
                tun_outbound_rx,
                tun_limit,
                routes,
                &mut outbound_buffers,
            )
            .with_firsts(outbound_firsts);
            let (_, endpoint, tun) = {
                let mut admission = DataplaneOutboundAdmission {
                    driver: self,
                    summary: &mut summary,
                    trace_enabled,
                    counts: &mut outbound_admitted,
                };
                outbound_source.drain_outbound_batched(remaining_outbound_limit, &mut admission)
            };
            endpoint_drained = endpoint_drained.saturating_add(endpoint);
            tun_drained = tun_drained.saturating_add(tun);
        }
        if trace_enabled {
            crate::perf_profile::record_event_count(
                crate::perf_profile::Event::DataplaneLiveEndpointAdmitted,
                outbound_admitted.endpoint as u64,
            );
            crate::perf_profile::record_event_count(
                crate::perf_profile::Event::DataplaneLiveTunAdmitted,
                outbound_admitted.tun as u64,
            );
        }
        drop(admit_timer);

        DataplaneLiveAdmissionResult {
            summary,
            outbound_buffers,
            endpoint_drained,
            tun_drained,
        }
    }

    async fn finish_live_node_turn_after_admission(
        &mut self,
        request: DataplaneLiveFinishRequest<'_>,
    ) -> DataplaneLiveNodeTurn {
        let DataplaneLiveFinishRequest {
            admission,
            routes,
            endpoint_tx,
            transports,
            crypto_limit,
            collect_transport_sent_receipts,
            crypto_worker,
            transport_send_batch_packets,
            deferred_raw_ingress,
            deferred_endpoint_data_batches,
            deferred_tun_packets,
        } = request;
        let DataplaneLiveAdmissionResult {
            summary,
            mut outbound_buffers,
            endpoint_drained,
            tun_drained,
        } = admission;
        let mut report = self
            .finish_aead_live_node_output_turn(DataplaneLiveOutputRequest {
                summary,
                routes,
                endpoint_tx,
                transports,
                deferred_raw_ingress,
                crypto_limit,
                collect_transport_sent_receipts,
                crypto_worker,
                transport_send_batch_packets,
            })
            .await;
        let endpoint_deferred_count = outbound_buffers.deferred_endpoint_data_batches.len();
        deferred_endpoint_data_batches.append(&mut outbound_buffers.deferred_endpoint_data_batches);
        let tun_deferred_count = outbound_buffers.tun_deferred_packets.len();
        deferred_tun_packets.append(&mut outbound_buffers.tun_deferred_packets);
        report.endpoint_data_drops = outbound_buffers.endpoint_drops;
        report.endpoint_source_drained = endpoint_drained;
        report.deferred_endpoint_data_batches_count = endpoint_deferred_count;
        report.tun_outbound_drops = outbound_buffers.tun_drops;
        report.tun_deferred_packets = tun_deferred_count;
        report.tun_source_drained = tun_drained;
        report
    }

    fn reset_turn_buffers(&mut self) {
        self.outputs.clear();
        self.output_rewrite_buffer.clear();
        self.raw_socket_packets.clear();
        self.retired_outbound_packets.clear();
        self.transport_output.clear();
        self.drops.clear();
        self.raw_ingress_drops.clear();
        self.output_drops.clear();
        self.fmp_ingress_receipts.clear();
        self.fmp_link_ingress.clear();
        self.fsp_coord_warmups.clear();
        self.fsp_local_session_ingress.clear();
        self.fsp_authenticated_ingress.clear();
    }

    fn raw_ingress_socket_packet<R>(
        packet: DataplaneRawIngress,
        router: &mut R,
        summary: &mut DataplaneRuntimeSummary,
        raw_ingress_drops: &mut Vec<DataplaneRawIngressDrop>,
        deferred_raw_ingress: &mut std::collections::VecDeque<(DataplaneRawIngress, u8)>,
        retry_count: u8,
    ) -> Option<SocketPacket>
    where
        R: DataplaneIngressRouter,
    {
        let header = match packet.protocol {
            PacketProtocol::Fmp => match FmpWireHeader::parse(packet.payload.as_slice()) {
                Ok(header) => DataplaneIngressHeader::Fmp(header),
                Err(error) => {
                    summary.raw_ingress_dropped += 1;
                    raw_ingress_drops.push(DataplaneRawIngressDrop::from_packet(
                        packet,
                        DataplaneRawIngressDropReason::Wire(error),
                    ));
                    return None;
                }
            },
            PacketProtocol::Fsp => match FspWireHeader::parse(packet.payload.as_slice()) {
                Ok(header) => DataplaneIngressHeader::Fsp(header),
                Err(error) => {
                    summary.raw_ingress_dropped += 1;
                    raw_ingress_drops.push(DataplaneRawIngressDrop::from_packet(
                        packet,
                        DataplaneRawIngressDropReason::Wire(error),
                    ));
                    return None;
                }
            },
        };

        let counter = header.counter();
        let Some(route) = router.route(&packet, header) else {
            if packet.protocol == PacketProtocol::Fsp
                && packet.fsp_source.is_some()
                && retry_count < DATAPLANE_DEFERRED_RAW_INGRESS_MAX_RETRIES
            {
                deferred_raw_ingress.push_back((packet, retry_count.saturating_add(1)));
                return None;
            }
            summary.raw_ingress_dropped += 1;
            raw_ingress_drops.push(DataplaneRawIngressDrop::from_packet(
                packet,
                DataplaneRawIngressDropReason::Unrouted,
            ));
            return None;
        };

        let wire_flags = header.flags();
        let DataplaneRawIngress {
            path: source_path,
            previous_hop,
            ce_flag,
            path_mtu,
            activity_tick,
            payload,
            ..
        } = packet;
        let mut socket_packet = SocketPacket::new(
            route.owner,
            route.generation,
            counter,
            route.class,
            route.output,
            payload,
        )
        .with_source_path(source_path);
        socket_packet = socket_packet.with_path_mtu(path_mtu);
        socket_packet = socket_packet.with_receive_epoch(route.receive_epoch);
        if let Some(tick) = activity_tick {
            socket_packet = socket_packet.with_activity_tick(tick);
        }
        if let Some(previous_hop) = previous_hop {
            socket_packet = socket_packet.with_previous_hop(previous_hop);
        }
        socket_packet = socket_packet.with_ce_flag(ce_flag);
        socket_packet = socket_packet.with_wire_flags(wire_flags);
        Some(socket_packet)
    }

    fn admit_socket_packet(
        &mut self,
        packet: SocketPacket,
        summary: &mut DataplaneRuntimeSummary,
    ) {
        match self.mover.submit_socket_packet(packet) {
            Ok(_) => summary.inbound_admitted += 1,
            Err(_) => summary.inbound_dropped += 1,
        }
    }

    fn admit_socket_packets(
        &mut self,
        packets: &mut Vec<SocketPacket>,
        summary: &mut DataplaneRuntimeSummary,
    ) {
        match packets.len() {
            0 => {}
            1 => {
                let packet = packets.pop().expect("checked one packet");
                self.admit_socket_packet(packet, summary);
            }
            _ => {
                let capacity = packets.capacity();
                let batch = std::mem::replace(packets, Vec::with_capacity(capacity));
                let (admitted, dropped) = self.mover.submit_socket_packet_batch(batch);
                summary.inbound_admitted = summary.inbound_admitted.saturating_add(admitted);
                summary.inbound_dropped = summary.inbound_dropped.saturating_add(dropped);
            }
        }
    }

    fn admit_fast_ingress_runs(
        &mut self,
        fast_ingress: DataplaneFastIngressBatch,
        summary: &mut DataplaneRuntimeSummary,
    ) {
        for run in fast_ingress.into_runs() {
            let run_len = run.len();
            crate::perf_profile::record_dataplane_fast_ingress_owner_run(run_len);
            let (owner, lane, packets) = run.into_parts();
            let (admitted, dropped) =
                self.mover.submit_socket_packet_run(Some(owner), Some(lane), packets);
            summary.inbound_admitted = summary.inbound_admitted.saturating_add(admitted);
            summary.inbound_dropped = summary.inbound_dropped.saturating_add(dropped);
        }
    }

    fn admit_outbound_packet(
        &mut self,
        packet: OutboundPacket,
        summary: &mut DataplaneRuntimeSummary,
    ) {
        match self.mover.submit_outbound_packet(packet) {
            Ok(_) => summary.outbound_admitted += 1,
            Err(_) => summary.outbound_dropped += 1,
        }
    }

    fn admit_outbound_packet_batch(
        &mut self,
        packets: Vec<OutboundPacket>,
        summary: &mut DataplaneRuntimeSummary,
    ) {
        let packet_count = packets.len();
        crate::perf_profile::record_event(crate::perf_profile::Event::DataplaneOutboundBatchAdmit);
        crate::perf_profile::record_event_count(
            crate::perf_profile::Event::DataplaneOutboundBatchPackets,
            packet_count as u64,
        );
        let (admitted, dropped) = self.mover.submit_outbound_packet_batch(packets);
        summary.outbound_admitted = summary.outbound_admitted.saturating_add(admitted);
        summary.outbound_dropped = summary.outbound_dropped.saturating_add(dropped);
    }

    fn admit_outbound_packets(
        &mut self,
        packets: &mut Vec<OutboundPacket>,
        summary: &mut DataplaneRuntimeSummary,
    ) {
        match packets.len() {
            0 => {}
            1 => {
                let packet = packets.pop().expect("checked one packet");
                self.admit_outbound_packet(packet, summary);
            }
            _ => {
                let capacity = packets.capacity();
                let batch = std::mem::replace(packets, Vec::with_capacity(capacity));
                self.admit_outbound_packet_batch(batch, summary);
            }
        }
    }

    fn send_collected_outputs<S>(
        &mut self,
        mut summary: DataplaneRuntimeSummary,
        sink: &mut S,
    ) -> DataplaneRuntimeTurn<'_>
    where
        S: DataplaneOutputSink,
    {
        let dropped_before = self.output_drops.len();
        let sent = if self.outputs.is_empty() {
            0
        } else {
            crate::perf_profile::record_dataplane_live_output_batch(self.outputs.len());
            let _output_sink_timer = crate::perf_profile::Timer::start(
                crate::perf_profile::Stage::DataplaneOutputSink,
            );
            sink.send_batch(self.outputs.drain(..), &mut self.output_drops)
        };
        summary.outputs_sent += sent;
        summary.outputs_dropped += self.output_drops.len().saturating_sub(dropped_before);

        DataplaneRuntimeTurn {
            summary,
            raw_ingress_drops: &self.raw_ingress_drops,
            output_drops: &self.output_drops,
            outputs: &self.outputs,
            drops: &self.drops,
        }
    }

    fn process_live_internal_outputs<R>(
        &mut self,
        router: &mut R,
        summary: &mut DataplaneRuntimeSummary,
        deferred_raw_ingress: &mut std::collections::VecDeque<(DataplaneRawIngress, u8)>,
    ) -> usize
    where
        R: DataplaneIngressRouter,
    {
        let mut outputs = self.take_outputs_for_rewrite();
        let mut raw_socket_packets = std::mem::take(&mut self.raw_socket_packets);
        raw_socket_packets.clear();
        let dropped_before = self.output_drops.len();
        let admitted_before = summary.inbound_admitted;
        for output in outputs.drain(..) {
            match output.target {
                OutputTarget::SessionIngress { local_addr } => {
                    let receipt = DataplaneFmpIngressReceipt::from_output(&output);
                    match dataplane_session_ingress_from_output(output, local_addr) {
                        DataplaneSessionIngressHandoff::Raw { raw, coord_warmup } => {
                            if let Some(receipt) = receipt {
                                self.fmp_ingress_receipts.push(receipt);
                            }
                            if !coord_warmup.is_empty() {
                                self.fsp_coord_warmups.push(coord_warmup);
                            }
                            if let Some(socket_packet) = Self::raw_ingress_socket_packet(
                                raw,
                                router,
                                summary,
                                &mut self.raw_ingress_drops,
                                deferred_raw_ingress,
                                1,
                            ) {
                                raw_socket_packets.push(socket_packet);
                            }
                        }
                        DataplaneSessionIngressHandoff::Local(ingress) => {
                            if let Some(receipt) = receipt {
                                self.fmp_ingress_receipts.push(receipt);
                            }
                            self.fsp_local_session_ingress.push(ingress);
                        }
                        DataplaneSessionIngressHandoff::Rejected {
                            output,
                            error: DataplaneSessionHandoffError::NoRoute,
                        } => {
                            if let Some(receipt) = receipt {
                                self.fmp_link_ingress
                                    .push(DataplaneFmpLinkIngress::from_output(output, receipt));
                            } else {
                                self.output_drops.push(DataplaneOutputDrop::from_output(
                                    &output,
                                    DataplaneOutputError::NoRoute,
                                ));
                            }
                        }
                        DataplaneSessionIngressHandoff::Rejected { output, error } => {
                            self.output_drops.push(DataplaneOutputDrop::from_output(
                                &output,
                                dataplane_output_error_from_session_handoff(error),
                            ))
                        }
                    }
                }
                OutputTarget::SessionPayload { .. } => {
                    let mut output = output;
                    if let Some(ingress) =
                        DataplaneFspSessionIngress::take_from_output(&mut output)
                    {
                        self.record_fsp_session_ingress_activity(&ingress);
                        self.fsp_authenticated_ingress.push_session(ingress);
                    } else {
                        self.output_drops.push(DataplaneOutputDrop::from_output(
                            &output,
                            DataplaneOutputError::InvalidPacket,
                        ));
                    }
                }
                _ => self.outputs.push(output),
            }
        }
        self.admit_socket_packets(&mut raw_socket_packets, summary);
        self.raw_socket_packets = raw_socket_packets;
        self.output_rewrite_buffer = outputs;
        summary.outputs = self.outputs.len();
        summary.outputs_dropped = summary
            .outputs_dropped
            .saturating_add(self.output_drops.len().saturating_sub(dropped_before));
        summary.inbound_admitted.saturating_sub(admitted_before)
    }

    fn retire_queued_completed_aead_outputs(
        &mut self,
        limit: usize,
        compact_endpoint_data: bool,
    ) {
        let retired_completions = self
            .mover
            .retire_queued_completions_into(
                limit,
                &mut DataplaneRetiredOutputSink::new(
                    &mut self.outputs,
                    &mut self.retired_outbound_packets,
                    &mut self.fsp_authenticated_ingress,
                ),
                compact_endpoint_data,
            );
        crate::perf_profile::record_dataplane_live_completions_retired(retired_completions);
        let mut mover_drops = self.mover.drain_drops();
        self.drops.append(&mut mover_drops);
    }

    fn collect_live_session_outputs<R>(
        &mut self,
        mut summary: DataplaneRuntimeSummary,
        router: &mut R,
        crypto_limit: usize,
        crypto_worker: &mut DataplaneAeadWorkerPool,
        compact_endpoint_data: bool,
        deferred_raw_ingress: &mut std::collections::VecDeque<(DataplaneRawIngress, u8)>,
    ) -> DataplaneRuntimeSummary
    where
        R: DataplaneIngressRouter,
    {
        let mut remaining = crypto_limit;
        self.process_live_internal_outputs(router, &mut summary, deferred_raw_ingress);
        loop {
            let dispatched_before = summary.dispatched;
            summary = self.collect_aead_outputs(
                summary,
                remaining,
                crypto_worker,
                compact_endpoint_data,
            );
            let dispatched = summary.dispatched.saturating_sub(dispatched_before);
            remaining = remaining.saturating_sub(dispatched);
            if remaining == 0 {
                break;
            }

            if self.process_live_internal_outputs(router, &mut summary, deferred_raw_ingress) == 0 {
                break;
            }
        }
        self.process_live_internal_outputs(router, &mut summary, deferred_raw_ingress);
        summary
    }

    fn take_outputs_for_rewrite(&mut self) -> Vec<PacketOutput> {
        let mut outputs = std::mem::take(&mut self.output_rewrite_buffer);
        std::mem::swap(&mut self.outputs, &mut outputs);
        outputs
    }

    fn collect_aead_outputs(
        &mut self,
        mut summary: DataplaneRuntimeSummary,
        limit: usize,
        crypto_worker: &mut DataplaneAeadWorkerPool,
        compact_endpoint_data: bool,
    ) -> DataplaneRuntimeSummary {
        let mut remaining = limit;
        while remaining > 0 {
            let dispatched = {
                let _dispatch_timer = crate::perf_profile::Timer::start(
                    crate::perf_profile::Stage::DataplaneAeadDispatch,
                );
                self.mover.run_aead_available_into(
                    remaining,
                    DataplaneAeadRunBuffers::new(
                        &mut self.prepared_work,
                        &mut self.completion_batches,
                        &mut self.outputs,
                        &mut self.retired_outbound_packets,
                        &mut self.fsp_authenticated_ingress,
                        &mut self.drops,
                    ),
                    crypto_worker,
                    compact_endpoint_data,
                )
            };
            summary.dispatched = summary.dispatched.saturating_add(dispatched);
            remaining = remaining.saturating_sub(dispatched);

            let outbound_admitted_before = summary.outbound_admitted;
            summary = self.admit_retired_outbound_packets(summary);

            if dispatched == 0 && summary.outbound_admitted == outbound_admitted_before {
                break;
            }
        }

        summary.outputs = self.outputs.len();
        summary.drops = self.drops.len();
        summary
    }

    fn admit_retired_outbound_packets(
        &mut self,
        mut summary: DataplaneRuntimeSummary,
    ) -> DataplaneRuntimeSummary {
        let mut outbound_packets = std::mem::take(&mut self.retired_outbound_packets);
        for packet in &mut outbound_packets {
            self.refresh_wrapped_fsp_outbound_context(packet);
        }
        self.admit_outbound_packets(&mut outbound_packets, &mut summary);
        self.retired_outbound_packets = outbound_packets;
        summary.outputs = self.outputs.len();
        summary.drops = self.drops.len();
        summary
    }

    fn refresh_wrapped_fsp_outbound_context(&self, packet: &mut OutboundPacket) {
        if !packet.has_fsp_send_receipt() || packet.owner().protocol() != PacketProtocol::Fmp {
            return;
        }
        let Some(context) = self.owner_fmp_send_context(packet.owner()) else {
            return;
        };
        packet.refresh_fmp_send_context(
            context.generation(),
            context.receiver_idx(),
            context.flags(),
        );
    }

    fn deliver_direct_endpoint_packet_batches(&mut self, direct_sink: Option<&EndpointDirectSink>) {
        let Some(direct_sink) = direct_sink else {
            return;
        };

        let mut dropped = 0usize;
        let mut sink_failed = false;
        for bulk in self.fsp_authenticated_ingress.endpoint_data_batches_mut() {
            if !bulk.has_direct_packet_runs() {
                continue;
            }
            let count = bulk.len();
            if sink_failed {
                let _ = bulk.take_direct_packet_batch();
                dropped = dropped.saturating_add(count);
                continue;
            }
            let packet_batch = bulk.take_direct_packet_batch();
            if direct_sink.deliver_direct_packet_batch(packet_batch).is_err() {
                dropped = dropped.saturating_add(count);
                sink_failed = true;
            }
        }

        if dropped > 0 {
            crate::perf_profile::record_event_count(
                crate::perf_profile::Event::EndpointEventBulkDropped,
                dropped as u64,
            );
        }
    }
}
