otel-arrow-dfe-engine 0.60.0

// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0

//! Set of traits and structures used to implement processors.
//!
//! A processor is a node in the pipeline that transforms, filters, or otherwise processes messages
//! as they flow through the pipeline. Processors can perform operations such as:
//!
//! 1. Filtering messages based on certain criteria
//! 2. Transforming message content or format
//! 3. Aggregating multiple messages into a single message
//! 4. Splitting a single message into multiple messages
//! 5. Adding or removing attributes from messages
//!
//! # Lifecycle
//!
//! 1. The processor is instantiated and configured
//! 2. The processor receives and processes both data messages and control messages
//! 3. For each message, the processor can transform it, filter it, or split it into multiple messages
//! 4. The processor can maintain state between processing calls if needed
//! 5. The processor responds to control messages such as Config, TimerTick, or Shutdown
//! 6. The processor shuts down when it receives a `Shutdown` control message or encounters a fatal error
//!
//! # Thread Safety
//!
//! This implementation is designed for use in both single-threaded and multi-threaded environments.
//! The `Processor` trait requires the `Send` bound, enabling the use of thread-safe types.
//!
//! # Scalability
//!
//! To ensure scalability, the pipeline engine will start multiple instances of the same pipeline
//! in parallel on different cores, each with its own processor instance.

use crate::Interests;
#[cfg(any(test, feature = "test-utils"))]
use crate::control::WakeupRevision;
use crate::control::{AckMsg, NackMsg, RuntimeCtrlMsgSender, WakeupSlot};
use crate::effect_handler::{
    EffectHandlerCore, SourceTagging, TelemetryTimerCancelHandle, TimerCancelHandle,
};
use crate::error::{Error, TypedError};
use crate::flow_metrics::{
    DecisionFlowMetrics, EndFlowMetrics, FLOW_SIGNALS, FlowDroppedItemsMetrics,
    FlowDurationMetricSet, FlowInputItemsMetrics, FlowInputMessageMetrics, FlowInputSizeMetrics,
    FlowOutputItemsMetrics, FlowOutputMessageMetrics, FlowOutputSizeMetrics, InputFlowMetrics,
    SharedFlowMetricState, flow_signal_index, nanos_u64,
};
use crate::message::Message;
use crate::node::NodeId;
use crate::output_router::OutputRouter;
use crate::processor::ProcessorRuntimeRequirements;
use crate::runtime_services::{CodecEffectHandler, PipelineRuntimeServices};
use crate::shared::message::SharedSender;
use crate::{WakeupError, WakeupSetOutcome};
use async_trait::async_trait;
use otel_arrow_dfe_config::{PortName, SignalType};
use otel_arrow_dfe_pdata_codec::CodecService;
use otel_arrow_dfe_telemetry::common_attributes::SignalAttributes;
use otel_arrow_dfe_telemetry::error::Error as TelemetryError;
use otel_arrow_dfe_telemetry::metrics::{MeasurementMetricSet, MetricSet, MetricSetHandler};
use otel_arrow_dfe_telemetry::reporter::MetricsReporter;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

/// A trait for processors in the pipeline (Send definition).
#[async_trait]
pub trait Processor<PData> {
    /// Processes a message and optionally produces effects, such as generating new pdata messages.
    ///
    /// This method is called by the pipeline engine for each message that arrives at the processor.
    /// Unlike receivers, processors have known inputs (messages from previous stages), so the pipeline
    /// engine can control when to call this method and when the processor executes.
    ///
    /// This approach allows for greater flexibility and optimization, giving the pipeline engine
    /// the ability to decide whether to spawn one task per processor or one task for a group of processors.
    /// The method signature uses `&mut self` rather than `Box<Self>` because the engine only wants to
    /// temporarily allow mutation of the processor instance, not transfer ownership.
    ///
    /// The processor can:
    /// - Transform the message and return a new message
    /// - Filter the message by returning None
    /// - Split the message into multiple messages by returning a vector
    /// - Handle control messages (e.g., Config, TimerTick, Wakeup, Shutdown)
    ///
    /// Processor-local wakeups are scheduled through
    /// [`EffectHandler::set_wakeup`]. They are delivered back to the processor
    /// as `Message::Control(NodeControlMsg::Wakeup { .. })` through the normal
    /// inbox path and participate in the same control-vs-pdata fairness rules
    /// as other control traffic.
    ///
    /// # Parameters
    ///
    /// - `msg`: The message to process, which can be either a data message or a control message
    /// - `effect_handler`: A handler to perform side effects such as sending messages to the next node.
    ///    This can be either Send or !Send depending on the processor's Mode type.
    ///
    /// # Returns
    ///
    /// - `Ok(())`: The processor successfully processed the message
    /// - `Err(Error)`: The processor encountered an error and could not process the message
    ///
    /// # Errors
    ///
    /// Returns an [`Error`] if the processor encounters an unrecoverable error.
    async fn process(
        &mut self,
        msg: Message<PData>,
        effect_handler: &mut EffectHandler<PData>,
    ) -> Result<(), Error>;

    /// Returns whether the engine should deliver pdata messages to this processor right now.
    ///
    /// When this returns `false` the engine pauses pdata delivery and only forwards control
    /// messages (acks/nacks) until the processor signals it is ready again. Defaults to `true`.
    fn accept_pdata(&self) -> bool {
        true
    }

    /// Returns optional runtime services that this processor needs from the engine.
    ///
    /// This is the single source of truth for runtime wiring. For example,
    /// `local_wakeups: Some(...)` both enables processor-local wakeups and
    /// declares the live slot count the engine must provision.
    fn runtime_requirements(&self) -> ProcessorRuntimeRequirements {
        ProcessorRuntimeRequirements::none()
    }
}

/// A `Send` implementation of the EffectHandler.
#[derive(Clone)]
pub struct EffectHandler<PData> {
    pub(crate) core: EffectHandlerCore<PData>,
    /// Output-port router.
    pub router: OutputRouter<SharedSender<PData>>,
    /// Per-handler flow_metric state. See [`SharedFlowMetricState`] /
    /// [`EndFlowMetrics`] for field-level documentation.
    ///
    /// `Mutex` is used inside the marker/accumulator cells because shared
    /// processors run on worker threads -- contention is bounded to the
    /// per-processor sequential `process()` loop and the periodic
    /// telemetry drain.
    pub(crate) flow: SharedFlowMetricState,
}

/// Implementation for the `Send` effect handler.
impl<PData> EffectHandler<PData> {
    /// Creates a new shared (Send) `EffectHandler` with the given processor configuration and
    /// pipeline runtime services.
    #[must_use]
    pub fn new(
        node_id: NodeId,
        msg_senders: HashMap<PortName, SharedSender<PData>>,
        default_port: Option<PortName>,
        metrics_reporter: MetricsReporter,
        runtime_services: PipelineRuntimeServices,
    ) -> Self {
        let core = EffectHandlerCore::new(node_id.clone(), metrics_reporter, runtime_services);
        let router = OutputRouter::new(node_id, msg_senders, default_port);
        EffectHandler {
            core,
            router,
            flow: SharedFlowMetricState::default(),
        }
    }

    /// Returns the id of the processor associated with this handler.
    #[must_use]
    pub fn processor_id(&self) -> NodeId {
        self.core.node_id()
    }

    /// Sets outgoing messages source tagging mode.
    pub fn set_source_tagging(&mut self, value: SourceTagging) {
        self.core.set_source_tagging(value);
    }

    /// Returns outgoing messages source tagging mode. Enabled when
    /// the destination node has multiple input sources.
    #[must_use]
    pub const fn source_tagging(&self) -> SourceTagging {
        self.core.source_tagging()
    }

    /// Returns the list of connected output ports for this processor.
    #[must_use]
    pub fn connected_ports(&self) -> Vec<PortName> {
        self.router.connected_ports()
    }

    /// Returns the selected default output port name, if one exists.
    #[must_use]
    pub fn default_port(&self) -> Option<PortName> {
        self.router.default_port()
    }

    /// Returns the precomputed node interests.
    #[must_use]
    pub fn node_interests(&self) -> Interests {
        self.core.node_interests()
    }

    /// Sets flow_metric start/stop roles for this node.
    pub(crate) fn set_flow_roles(
        &mut self,
        is_start: bool,
        is_end: bool,
        input_message_metric: Option<MeasurementMetricSet<FlowInputMessageMetrics>>,
        input_items_metric: Option<MeasurementMetricSet<FlowInputItemsMetrics>>,
        input_size_metric: Option<MeasurementMetricSet<FlowInputSizeMetrics>>,
        duration_metric: Option<FlowDurationMetricSet>,
        output_items_metric: Option<MeasurementMetricSet<FlowOutputItemsMetrics>>,
        output_message_metric: Option<MeasurementMetricSet<FlowOutputMessageMetrics>>,
        output_size_metric: Option<MeasurementMetricSet<FlowOutputSizeMetrics>>,
        dropped_items_metric: Option<MeasurementMetricSet<FlowDroppedItemsMetrics>>,
        flow_metrics_active: bool,
        flow_needs_timing: bool,
    ) {
        self.flow.is_start = is_start;
        self.flow.is_end = is_end;
        self.flow.is_decision = dropped_items_metric.is_some();
        self.flow.interests.set(
            crate::flow_metrics::FlowMetricInterests::INPUT_MESSAGES,
            input_message_metric.is_some(),
        );
        self.flow.interests.set(
            crate::flow_metrics::FlowMetricInterests::INPUT_ITEMS,
            input_items_metric.is_some(),
        );
        self.flow.interests.set(
            crate::flow_metrics::FlowMetricInterests::INPUT_SIZE,
            input_size_metric.is_some(),
        );
        self.flow.interests.set(
            crate::flow_metrics::FlowMetricInterests::COMPUTE_DURATION,
            duration_metric.is_some(),
        );
        self.flow.interests.set(
            crate::flow_metrics::FlowMetricInterests::OUTPUT_ITEMS,
            output_items_metric.is_some(),
        );
        self.flow.interests.set(
            crate::flow_metrics::FlowMetricInterests::OUTPUT_MESSAGES,
            output_message_metric.is_some(),
        );
        self.flow.interests.set(
            crate::flow_metrics::FlowMetricInterests::OUTPUT_SIZE,
            output_size_metric.is_some(),
        );
        self.flow.interests.set(
            crate::flow_metrics::FlowMetricInterests::DROPPED_ITEMS,
            dropped_items_metric.is_some(),
        );
        self.flow.active = flow_metrics_active;
        self.flow.needs_timing = flow_needs_timing;
        self.flow.input = InputFlowMetrics {
            input_messages: input_message_metric
                .map(|metrics| (metrics, Arc::new(Mutex::new([0; 3])))),
            input_items: input_items_metric.map(|metrics| (metrics, Arc::new(Mutex::new([0; 3])))),
            input_size: input_size_metric.map(|metrics| (metrics, Arc::new(Mutex::new([0; 3])))),
        };
        self.flow.end = EndFlowMetrics {
            duration: duration_metric
                .map(FlowDurationMetricSet::into_measurement)
                .map(|measurement| Arc::new(Mutex::new(measurement))),
            output_messages: output_message_metric
                .map(|metrics| (metrics, Arc::new(Mutex::new([0; 3])))),
            output_items: output_items_metric
                .map(|metrics| (metrics, Arc::new(Mutex::new([0; 3])))),
            output_size: output_size_metric.map(|metrics| (metrics, Arc::new(Mutex::new([0; 3])))),
        };
        self.flow.decision = DecisionFlowMetrics {
            dropped_items: dropped_items_metric
                .map(|metrics| (metrics, Arc::new(Mutex::new([0; 3])))),
        };
    }

    /// Returns whether this node is a flow_metric start node.
    #[must_use]
    pub fn is_flow_start(&self) -> bool {
        self.flow.is_start
    }

    /// Returns whether this node is a flow_metric stop node.
    #[must_use]
    pub fn is_flow_end(&self) -> bool {
        self.flow.is_end
    }

    /// Returns whether this node is a decision node for some flow.
    #[must_use]
    pub fn is_flow_decision(&self) -> bool {
        self.flow.is_decision
    }

    /// Returns whether any flow_metric is configured in this pipeline.
    #[must_use]
    pub fn flow_metrics_active(&self) -> bool {
        self.flow.active
    }

    /// Begin per-message flow_metric timing for the upcoming `process()` call.
    ///
    /// Sets the send-marker to "now" so that the first
    /// [`take_elapsed_since_send_marker_ns`] call (typically from the send
    /// hook) measures elapsed time from the start of `process()`.
    /// No-op unless some flow in this pipeline tracks `compute.duration`
    /// (`needs_timing`); a count-only flow such as `dropped.items` pays no
    /// per-message `Instant::now()` cost.
    pub(crate) fn begin_process_timing(&self) {
        if self.flow.needs_timing {
            *self
                .flow
                .last_send_marker
                .lock()
                .expect("last_send_marker poisoned") = Some(Instant::now());
        }
    }

    /// Returns nanoseconds elapsed since the send-marker was last set or
    /// advanced, then advances the marker to "now". Returns 0 when no
    /// marker is active (e.g. flow_metrics disabled, or
    /// `begin_process_timing` was not called for this message).
    #[must_use]
    pub fn take_elapsed_since_send_marker_ns(&self) -> u64 {
        let mut guard = self
            .flow
            .last_send_marker
            .lock()
            .expect("last_send_marker poisoned");
        let Some(prev) = *guard else {
            return 0;
        };
        let now = Instant::now();
        *guard = Some(now);
        nanos_u64(now.duration_since(prev).as_nanos())
    }

    /// Record `total` nanoseconds as seconds into the shared flow metric histogram.
    pub fn record_flow_duration(&self, signal: SignalType, total: u64) {
        let Some(measurement) = self.flow.end.duration.as_ref() else {
            return;
        };
        let mut measurement = measurement
            .lock()
            .expect("flow duration accumulator poisoned");
        measurement.record(signal, total as f64 / 1_000_000_000.0);
    }

    /// Record input items into the shared flow accumulator.
    pub fn record_flow_input_items(&self, signal: SignalType, items: u64) {
        let Some((_, acc_mutex)) = self.flow.input.input_items.as_ref() else {
            return;
        };
        let mut acc = acc_mutex
            .lock()
            .expect("flow input_items accumulator poisoned");
        let index = flow_signal_index(signal);
        acc[index] = acc[index].saturating_add(items);
    }

    /// Record one message entering the flow.
    pub fn record_flow_input_message(&self, signal: SignalType) {
        let Some((_, acc_mutex)) = self.flow.input.input_messages.as_ref() else {
            return;
        };
        let mut acc = acc_mutex
            .lock()
            .expect("flow input_messages accumulator poisoned");
        let index = flow_signal_index(signal);
        acc[index] = acc[index].saturating_add(1);
    }

    /// Record logical payload bytes entering the flow.
    pub fn record_flow_input_size(&self, signal: SignalType, size: u64) {
        let Some((_, acc_mutex)) = self.flow.input.input_size.as_ref() else {
            return;
        };
        let mut acc = acc_mutex
            .lock()
            .expect("flow input_size accumulator poisoned");
        let index = flow_signal_index(signal);
        acc[index] = acc[index].saturating_add(size);
    }

    /// Record output items into the shared flow accumulator.
    pub fn record_flow_output_items(&self, signal: SignalType, items: u64) {
        let Some((_, acc_mutex)) = self.flow.end.output_items.as_ref() else {
            return;
        };
        let mut acc = acc_mutex
            .lock()
            .expect("flow output_items accumulator poisoned");
        let index = flow_signal_index(signal);
        acc[index] = acc[index].saturating_add(items);
    }

    /// Record one message leaving the flow.
    pub fn record_flow_output_message(&self, signal: SignalType) {
        let Some((_, acc_mutex)) = self.flow.end.output_messages.as_ref() else {
            return;
        };
        let mut acc = acc_mutex
            .lock()
            .expect("flow output_messages accumulator poisoned");
        let index = flow_signal_index(signal);
        acc[index] = acc[index].saturating_add(1);
    }

    /// Record logical payload bytes leaving the flow.
    pub fn record_flow_output_size(&self, signal: SignalType, size: u64) {
        let Some((_, acc_mutex)) = self.flow.end.output_size.as_ref() else {
            return;
        };
        let mut acc = acc_mutex
            .lock()
            .expect("flow output_size accumulator poisoned");
        let index = flow_signal_index(signal);
        acc[index] = acc[index].saturating_add(size);
    }

    /// Record the count of items this decision node chose to drop.
    ///
    /// No-op unless this node is wired as a decision node for a flow
    /// that enables `dropped.items`.
    pub fn record_flow_dropped_items(&self, signal: SignalType, items: u64) {
        let Some((_, acc_mutex)) = self.flow.decision.dropped_items.as_ref() else {
            return;
        };
        let mut acc = acc_mutex
            .lock()
            .expect("flow dropped_items accumulator poisoned");
        let index = flow_signal_index(signal);
        acc[index] = acc[index].saturating_add(items);
    }

    /// Drain accumulated flow_metric observations into the MetricSet and report.
    pub(crate) fn report_flow_metrics(&mut self) {
        if let Some((metrics, acc_mutex)) = self.flow.input.input_messages.as_mut() {
            let drained = {
                let mut guard = acc_mutex
                    .lock()
                    .expect("flow input_messages accumulator poisoned");
                std::mem::take(&mut *guard)
            };
            for signal in FLOW_SIGNALS {
                let count = drained[flow_signal_index(signal)];
                if count != 0 {
                    metrics
                        .with(SignalAttributes { signal })
                        .messages
                        .add(count);
                }
            }
            let _ = self.core.metrics_reporter.report_measurement(metrics);
        }
        if let Some((metrics, acc_mutex)) = self.flow.input.input_items.as_mut() {
            let drained = {
                let mut guard = acc_mutex
                    .lock()
                    .expect("flow input_items accumulator poisoned");
                std::mem::take(&mut *guard)
            };
            for signal in FLOW_SIGNALS {
                let count = drained[flow_signal_index(signal)];
                if count != 0 {
                    metrics.with(SignalAttributes { signal }).items.add(count);
                }
            }
            let _ = self.core.metrics_reporter.report_measurement(metrics);
        }
        if let Some((metrics, acc_mutex)) = self.flow.input.input_size.as_mut() {
            let drained = {
                let mut guard = acc_mutex
                    .lock()
                    .expect("flow input_size accumulator poisoned");
                std::mem::take(&mut *guard)
            };
            for signal in FLOW_SIGNALS {
                let size = drained[flow_signal_index(signal)];
                if size != 0 {
                    metrics.with(SignalAttributes { signal }).size.add(size);
                }
            }
            let _ = self.core.metrics_reporter.report_measurement(metrics);
        }
        if let Some(measurement) = self.flow.end.duration.as_mut() {
            measurement
                .lock()
                .expect("flow duration accumulator poisoned")
                .report(&mut self.core.metrics_reporter);
        }
        if let Some((metrics, acc_mutex)) = self.flow.end.output_messages.as_mut() {
            let drained = {
                let mut guard = acc_mutex
                    .lock()
                    .expect("flow output_messages accumulator poisoned");
                std::mem::take(&mut *guard)
            };
            for signal in FLOW_SIGNALS {
                let count = drained[flow_signal_index(signal)];
                if count != 0 {
                    metrics
                        .with(SignalAttributes { signal })
                        .messages
                        .add(count);
                }
            }
            let _ = self.core.metrics_reporter.report_measurement(metrics);
        }
        if let Some((metrics, acc_mutex)) = self.flow.end.output_items.as_mut() {
            let drained = {
                let mut guard = acc_mutex
                    .lock()
                    .expect("flow output_items accumulator poisoned");
                std::mem::take(&mut *guard)
            };
            for signal in FLOW_SIGNALS {
                let count = drained[flow_signal_index(signal)];
                if count != 0 {
                    metrics.with(SignalAttributes { signal }).items.add(count);
                }
            }
            let _ = self.core.metrics_reporter.report_measurement(metrics);
        }
        if let Some((metrics, acc_mutex)) = self.flow.end.output_size.as_mut() {
            let drained = {
                let mut guard = acc_mutex
                    .lock()
                    .expect("flow output_size accumulator poisoned");
                std::mem::take(&mut *guard)
            };
            for signal in FLOW_SIGNALS {
                let size = drained[flow_signal_index(signal)];
                if size != 0 {
                    metrics.with(SignalAttributes { signal }).size.add(size);
                }
            }
            let _ = self.core.metrics_reporter.report_measurement(metrics);
        }
        if let Some((metrics, acc_mutex)) = self.flow.decision.dropped_items.as_mut() {
            let drained = {
                let mut guard = acc_mutex
                    .lock()
                    .expect("flow dropped_items accumulator poisoned");
                std::mem::take(&mut *guard)
            };
            for signal in FLOW_SIGNALS {
                let count = drained[flow_signal_index(signal)];
                if count != 0 {
                    metrics.with(SignalAttributes { signal }).items.add(count);
                }
            }
            let _ = self.core.metrics_reporter.report_measurement(metrics);
        }
    }

    /// Drains and reliably hands off final flow metrics during processor shutdown.
    ///
    /// Every metric set shares the same absolute `deadline`, bounding the
    /// complete handoff rather than giving each set a fresh timeout.
    pub(crate) async fn report_flow_metrics_reliably(
        &mut self,
        deadline: Instant,
    ) -> Result<(), TelemetryError> {
        let reporter = self.core.metrics_reporter.clone();
        if let Some((metrics, acc_mutex)) = self.flow.input.input_messages.as_mut() {
            let drained = {
                let mut guard = acc_mutex
                    .lock()
                    .expect("flow input_messages accumulator poisoned");
                std::mem::take(&mut *guard)
            };
            for signal in FLOW_SIGNALS {
                let count = drained[flow_signal_index(signal)];
                if count != 0 {
                    metrics
                        .with(SignalAttributes { signal })
                        .messages
                        .add(count);
                }
            }
            let _ = reporter
                .report_measurement_reliably_until(metrics, deadline)
                .await?;
        }
        if let Some((metrics, acc_mutex)) = self.flow.input.input_items.as_mut() {
            let drained = {
                let mut guard = acc_mutex
                    .lock()
                    .expect("flow input_items accumulator poisoned");
                std::mem::take(&mut *guard)
            };
            for signal in FLOW_SIGNALS {
                let count = drained[flow_signal_index(signal)];
                if count != 0 {
                    metrics.with(SignalAttributes { signal }).items.add(count);
                }
            }
            let _ = reporter
                .report_measurement_reliably_until(metrics, deadline)
                .await?;
        }
        if let Some((metrics, acc_mutex)) = self.flow.input.input_size.as_mut() {
            let drained = {
                let mut guard = acc_mutex
                    .lock()
                    .expect("flow input_size accumulator poisoned");
                std::mem::take(&mut *guard)
            };
            for signal in FLOW_SIGNALS {
                let size = drained[flow_signal_index(signal)];
                if size != 0 {
                    metrics.with(SignalAttributes { signal }).size.add(size);
                }
            }
            let _ = reporter
                .report_measurement_reliably_until(metrics, deadline)
                .await?;
        }
        if let Some(measurement) = self.flow.end.duration.as_mut() {
            let snapshots = {
                let mut measurement = measurement
                    .lock()
                    .expect("flow duration accumulator poisoned");
                measurement.terminal_snapshots()
            };
            for snapshot in snapshots {
                let _ = reporter
                    .report_snapshot_reliably_until(snapshot, deadline)
                    .await?;
            }
        }
        if let Some((metrics, acc_mutex)) = self.flow.end.output_messages.as_mut() {
            let drained = {
                let mut guard = acc_mutex
                    .lock()
                    .expect("flow output_messages accumulator poisoned");
                std::mem::take(&mut *guard)
            };
            for signal in FLOW_SIGNALS {
                let count = drained[flow_signal_index(signal)];
                if count != 0 {
                    metrics
                        .with(SignalAttributes { signal })
                        .messages
                        .add(count);
                }
            }
            let _ = reporter
                .report_measurement_reliably_until(metrics, deadline)
                .await?;
        }
        if let Some((metrics, acc_mutex)) = self.flow.end.output_items.as_mut() {
            let drained = {
                let mut guard = acc_mutex
                    .lock()
                    .expect("flow output_items accumulator poisoned");
                std::mem::take(&mut *guard)
            };
            for signal in FLOW_SIGNALS {
                let count = drained[flow_signal_index(signal)];
                if count != 0 {
                    metrics.with(SignalAttributes { signal }).items.add(count);
                }
            }
            let _ = reporter
                .report_measurement_reliably_until(metrics, deadline)
                .await?;
        }
        if let Some((metrics, acc_mutex)) = self.flow.end.output_size.as_mut() {
            let drained = {
                let mut guard = acc_mutex
                    .lock()
                    .expect("flow output_size accumulator poisoned");
                std::mem::take(&mut *guard)
            };
            for signal in FLOW_SIGNALS {
                let size = drained[flow_signal_index(signal)];
                if size != 0 {
                    metrics.with(SignalAttributes { signal }).size.add(size);
                }
            }
            let _ = reporter
                .report_measurement_reliably_until(metrics, deadline)
                .await?;
        }
        if let Some((metrics, acc_mutex)) = self.flow.decision.dropped_items.as_mut() {
            let drained = {
                let mut guard = acc_mutex
                    .lock()
                    .expect("flow dropped_items accumulator poisoned");
                std::mem::take(&mut *guard)
            };
            for signal in FLOW_SIGNALS {
                let count = drained[flow_signal_index(signal)];
                if count != 0 {
                    metrics.with(SignalAttributes { signal }).items.add(count);
                }
            }
            let _ = reporter
                .report_measurement_reliably_until(metrics, deadline)
                .await?;
        }
        Ok(())
    }

    /// Sends a message to the next node(s) in the pipeline.
    ///
    /// # Errors
    ///
    /// Returns an [`Error::ProcessorError`] if the message could not be routed to a port.
    #[inline]
    pub async fn send_message(&self, mut data: PData) -> Result<(), TypedError<PData>>
    where
        PData: crate::processor::FlowMetricHook + Send,
    {
        data.before_processor_send(self);
        self.router.send_default(data).await
    }

    /// Attempts to send a message without awaiting.
    ///
    /// Unlike `send_message`, this method returns immediately if the downstream
    /// channel is full, allowing the caller to handle backpressure without awaiting.
    ///
    /// # Errors
    ///
    /// Returns a [`TypedError::ChannelSendError`] containing [`SendError::Full`] if the
    /// channel is full, or [`SendError::Closed`] if the channel is closed.
    /// Returns a [`TypedError::Error`] if no default port is configured.
    #[inline]
    pub fn try_send_message(&self, mut data: PData) -> Result<(), TypedError<PData>>
    where
        PData: crate::processor::FlowMetricHook + Send,
    {
        data.before_processor_send(self);
        self.router.try_send_default(data)
    }

    /// Sends a message to a specific named output port.
    #[inline]
    pub async fn send_message_to<P>(
        &self,
        port: P,
        mut data: PData,
    ) -> Result<(), TypedError<PData>>
    where
        P: Into<PortName>,
        PData: crate::processor::FlowMetricHook + Send,
    {
        data.before_processor_send(self);
        self.router.send_to(port, data).await
    }

    /// Attempts to send a message to a specific named output port without awaiting.
    ///
    /// Unlike `send_message_to`, this method returns immediately if the downstream
    /// channel is full, allowing the caller to handle backpressure without awaiting.
    ///
    /// # Errors
    ///
    /// Returns a [`TypedError::ChannelSendError`] containing [`SendError::Full`] if the
    /// channel is full, or [`SendError::Closed`] if the channel is closed.
    /// Returns a [`TypedError::Error`] if the port does not exist.
    #[inline]
    pub fn try_send_message_to<P>(&self, port: P, mut data: PData) -> Result<(), TypedError<PData>>
    where
        P: Into<PortName>,
        PData: crate::processor::FlowMetricHook + Send,
    {
        data.before_processor_send(self);
        self.router.try_send_to(port, data)
    }

    /// Print an info message to stdout.
    ///
    /// This method provides a standardized way for processors to output
    /// informational messages without blocking the async runtime.
    pub async fn info(&self, message: &str) {
        self.core.info(message).await;
    }

    /// Starts a cancellable periodic timer that emits TimerTick on the control channel.
    /// Returns a handle that can be used to cancel the timer.
    ///
    /// Current limitation: Only one timer can be started by a processor at a time.
    pub async fn start_periodic_timer(
        &self,
        duration: Duration,
    ) -> Result<TimerCancelHandle<PData>, Error> {
        self.core.start_periodic_timer(duration).await
    }

    /// Starts a cancellable periodic telemetry timer that emits CollectTelemetry.
    pub async fn start_periodic_telemetry(
        &self,
        duration: Duration,
    ) -> Result<TelemetryTimerCancelHandle<PData>, Error> {
        self.core.start_periodic_telemetry(duration).await
    }

    /// Requeue retained pdata onto this node later.
    pub fn requeue_later(&self, when: Instant, data: Box<PData>) -> Result<(), PData> {
        self.core.requeue_later(when, data)
    }

    /// Set or replace a processor-local wakeup.
    pub fn set_wakeup(
        &self,
        slot: WakeupSlot,
        when: Instant,
    ) -> Result<WakeupSetOutcome, WakeupError> {
        self.core.set_wakeup(slot, when)
    }

    /// Cancel a previously scheduled processor-local wakeup.
    #[must_use]
    pub fn cancel_wakeup(&self, slot: WakeupSlot) -> bool {
        self.core.cancel_wakeup(slot)
    }

    /// Pop the next wakeup from the local scheduler, regardless of whether
    /// it is due. Returns `None` when no wakeup is scheduled or when local
    /// wakeups are not enabled.
    ///
    /// This is intended for testing, where the inbox loop is not running and
    /// wakeups need to be manually delivered.
    #[cfg(any(test, feature = "test-utils"))]
    #[must_use]
    pub fn pop_wakeup(&self) -> Option<(WakeupSlot, Instant, WakeupRevision)> {
        self.core.pop_wakeup()
    }

    /// Reports metrics collected by the processor.
    #[allow(dead_code)] // Will be used in the future. ToDo report metrics from channel and messages.
    pub(crate) fn report_metrics<M: MetricSetHandler + 'static>(
        &mut self,
        metrics: &mut MetricSet<M>,
    ) -> Result<(), TelemetryError> {
        self.core.report_metrics(metrics)
    }

    /// Reports processor-local wakeup scheduler metrics, if enabled.
    pub fn report_local_scheduler_metrics(
        &self,
        metrics_reporter: &mut MetricsReporter,
    ) -> Result<(), TelemetryError> {
        self.core.report_local_scheduler_metrics(metrics_reporter)
    }

    /// Sets the runtime control message sender for this effect handler.
    ///
    /// Primarily used by tests and manual harnesses that construct an EffectHandler directly;
    /// the engine wiring sets this automatically in `prepare_runtime`.
    pub fn set_runtime_ctrl_msg_sender(
        &mut self,
        runtime_ctrl_msg_sender: RuntimeCtrlMsgSender<PData>,
    ) {
        self.core
            .set_runtime_ctrl_msg_sender(runtime_ctrl_msg_sender);
    }

    // More methods will be added in the future as needed.

    /// Sets the pipeline result message sender for this effect handler.
    ///
    /// Primarily used by tests and manual harnesses that construct an EffectHandler directly;
    /// the engine wiring sets this automatically in `prepare_runtime`.
    pub fn set_pipeline_completion_msg_sender(
        &mut self,
        pipeline_completion_msg_sender: crate::control::PipelineCompletionMsgSender<PData>,
    ) {
        self.core
            .set_pipeline_completion_msg_sender(pipeline_completion_msg_sender);
    }
}

impl<PData> CodecEffectHandler for EffectHandler<PData> {
    fn codec_service(&self) -> &CodecService {
        self.core.runtime_services.codecs()
    }
}

impl<PData> crate::processor::FlowMetricEffectHandler for EffectHandler<PData> {
    #[inline]
    fn is_flow_start(&self) -> bool {
        EffectHandler::is_flow_start(self)
    }
    #[inline]
    fn is_flow_end(&self) -> bool {
        EffectHandler::is_flow_end(self)
    }
    #[inline]
    fn flow_metric_interests(&self) -> crate::flow_metrics::FlowMetricInterests {
        self.flow.interests
    }
    #[inline]
    fn take_elapsed_since_send_marker_ns(&self) -> u64 {
        EffectHandler::take_elapsed_since_send_marker_ns(self)
    }
    #[inline]
    fn record_flow_duration(&self, signal: SignalType, total: u64) {
        EffectHandler::record_flow_duration(self, signal, total);
    }
    #[inline]
    fn record_flow_input_items(&self, signal: SignalType, items: u64) {
        EffectHandler::record_flow_input_items(self, signal, items);
    }
    #[inline]
    fn record_flow_input_message(&self, signal: SignalType) {
        EffectHandler::record_flow_input_message(self, signal);
    }
    #[inline]
    fn record_flow_input_size(&self, signal: SignalType, size: u64) {
        EffectHandler::record_flow_input_size(self, signal, size);
    }
    #[inline]
    fn record_flow_output_items(&self, signal: SignalType, items: u64) {
        EffectHandler::record_flow_output_items(self, signal, items);
    }
    #[inline]
    fn record_flow_output_message(&self, signal: SignalType) {
        EffectHandler::record_flow_output_message(self, signal);
    }
    #[inline]
    fn record_flow_output_size(&self, signal: SignalType, size: u64) {
        EffectHandler::record_flow_output_size(self, signal, size);
    }
}

#[async_trait(?Send)]
impl<PData: crate::Unwindable> crate::_private::AckNackRouting<PData> for EffectHandler<PData> {
    async fn route_ack(&self, ack: AckMsg<PData>) -> Result<(), Error> {
        self.core.route_ack(ack).await
    }

    async fn route_nack(&self, nack: NackMsg<PData>) -> Result<(), Error> {
        self.core.route_nack(nack).await
    }
}

#[cfg(test)]
mod tests {
    #![allow(missing_docs)]
    use super::*;
    use crate::flow_metrics::{
        FlowAttributeSet, FlowDurationNormalMetrics, FlowOutputItemsMetrics,
    };
    use crate::shared::message::SharedSender;
    use crate::testing::{test_node, test_pipeline_ctx};
    use otel_arrow_dfe_channel::error::SendError;
    use otel_arrow_dfe_telemetry::metrics::MetricValue;
    use otel_arrow_dfe_telemetry::reporter::MetricsReporter;
    use std::collections::HashMap;

    // Note: `impl FlowMetricHook for u64` lives in `crate::local::processor`
    // tests; trait impls are crate-wide so we share it across test modules.

    #[test]
    fn effect_handler_try_send_message_success() {
        let (tx, mut rx) = tokio::sync::mpsc::channel::<u64>(10);
        let mut senders = HashMap::new();
        let _ = senders.insert("out".into(), SharedSender::mpsc(tx));

        let (_metrics_rx, metrics_reporter) = MetricsReporter::create_new_and_receiver(1);
        let eh = EffectHandler::new(
            test_node("proc"),
            senders,
            Some("out".into()),
            metrics_reporter,
            crate::testing::test_pipeline_runtime_services(),
        );

        // Should succeed when channel has capacity
        assert!(eh.try_send_message(42).is_ok());
        assert_eq!(rx.try_recv().unwrap(), 42);
    }

    #[test]
    fn effect_handler_try_send_message_inbox_full() {
        // Create a channel with capacity 1
        let (tx, _rx) = tokio::sync::mpsc::channel::<u64>(1);
        let mut senders = HashMap::new();
        let _ = senders.insert("out".into(), SharedSender::mpsc(tx));

        let (_metrics_rx, metrics_reporter) = MetricsReporter::create_new_and_receiver(1);
        let eh = EffectHandler::new(
            test_node("proc"),
            senders,
            Some("out".into()),
            metrics_reporter,
            crate::testing::test_pipeline_runtime_services(),
        );

        // First send should succeed
        assert!(eh.try_send_message(1).is_ok());

        // Second send should fail with Full since channel capacity is 1
        let result = eh.try_send_message(2);
        assert!(matches!(
            result,
            Err(TypedError::ChannelSendError(SendError::Full(2)))
        ));
    }

    #[test]
    fn effect_handler_try_send_message_no_default_sender() {
        let (a_tx, _a_rx) = tokio::sync::mpsc::channel::<u64>(10);
        let (b_tx, _b_rx) = tokio::sync::mpsc::channel::<u64>(10);

        let mut senders = HashMap::new();
        let _ = senders.insert("a".into(), SharedSender::mpsc(a_tx));
        let _ = senders.insert("b".into(), SharedSender::mpsc(b_tx));

        let (_metrics_rx, metrics_reporter) = MetricsReporter::create_new_and_receiver(1);
        // No default port specified with multiple ports = ambiguous
        let eh = EffectHandler::new(
            test_node("proc"),
            senders,
            None,
            metrics_reporter,
            crate::testing::test_pipeline_runtime_services(),
        );

        // Should return configuration error when no default sender
        let result = eh.try_send_message(99);
        assert!(matches!(result, Err(TypedError::Error(_))));
    }

    #[test]
    fn effect_handler_try_send_message_to_success() {
        let (a_tx, mut a_rx) = tokio::sync::mpsc::channel::<u64>(10);
        let (b_tx, mut b_rx) = tokio::sync::mpsc::channel::<u64>(10);

        let mut senders = HashMap::new();
        let _ = senders.insert("a".into(), SharedSender::mpsc(a_tx));
        let _ = senders.insert("b".into(), SharedSender::mpsc(b_tx));

        let (_metrics_rx, metrics_reporter) = MetricsReporter::create_new_and_receiver(1);
        let eh = EffectHandler::new(
            test_node("proc"),
            senders,
            None,
            metrics_reporter,
            crate::testing::test_pipeline_runtime_services(),
        );

        // Should succeed when sending to a specific port
        assert!(eh.try_send_message_to("b", 42).is_ok());
        assert_eq!(b_rx.try_recv().unwrap(), 42);
        // Port 'a' should not have received anything
        assert!(a_rx.try_recv().is_err());
    }

    #[test]
    fn effect_handler_try_send_message_to_channel_full() {
        let (tx, _rx) = tokio::sync::mpsc::channel::<u64>(1);
        let mut senders = HashMap::new();
        let _ = senders.insert("out".into(), SharedSender::mpsc(tx));

        let (_metrics_rx, metrics_reporter) = MetricsReporter::create_new_and_receiver(1);
        let eh = EffectHandler::new(
            test_node("proc"),
            senders,
            None,
            metrics_reporter,
            crate::testing::test_pipeline_runtime_services(),
        );

        // First send should succeed
        assert!(eh.try_send_message_to("out", 1).is_ok());
        // Second send should fail with Full
        let result = eh.try_send_message_to("out", 2);
        assert!(matches!(
            result,
            Err(TypedError::ChannelSendError(SendError::Full(2)))
        ));
    }

    #[test]
    fn effect_handler_try_send_message_to_unknown_port() {
        let (tx, _rx) = tokio::sync::mpsc::channel::<u64>(10);
        let mut senders = HashMap::new();
        let _ = senders.insert("out".into(), SharedSender::mpsc(tx));

        let (_metrics_rx, metrics_reporter) = MetricsReporter::create_new_and_receiver(1);
        let eh = EffectHandler::new(
            test_node("proc"),
            senders,
            None,
            metrics_reporter,
            crate::testing::test_pipeline_runtime_services(),
        );

        // Should return error for unknown port
        let result = eh.try_send_message_to("unknown", 99);
        assert!(matches!(result, Err(TypedError::Error(_))));
    }

    #[test]
    fn flow_metric_marker_accumulates_after_begin_process_timing_shared() {
        let (_metrics_rx, metrics_reporter) = MetricsReporter::create_new_and_receiver(1);
        let mut eh = EffectHandler::<u64>::new(
            test_node("proc"),
            HashMap::new(),
            None,
            metrics_reporter,
            crate::testing::test_pipeline_runtime_services(),
        );
        eh.set_flow_roles(
            true, false, None, None, None, None, None, None, None, None, true, true,
        );
        assert!(eh.is_flow_start());
        assert!(eh.flow.active);
        assert!(eh.flow.needs_timing);

        eh.begin_process_timing();

        let mut value = 0u64;
        for i in 0..10_000 {
            value = value.wrapping_add(std::hint::black_box(i));
        }
        let _ = std::hint::black_box(value);

        let ns = eh.take_elapsed_since_send_marker_ns();
        assert!(
            ns > 0,
            "take_elapsed_since_send_marker_ns should be non-zero after begin_process_timing, got {ns}"
        );
    }

    /// Scenario: an active shared flow tracks no compute duration.
    /// Guarantees: processing leaves the send marker unarmed and reports zero duration.
    #[test]
    fn flow_metric_marker_not_armed_when_timing_disabled_shared() {
        let (_metrics_rx, metrics_reporter) = MetricsReporter::create_new_and_receiver(1);
        let mut eh = EffectHandler::<u64>::new(
            test_node("proc"),
            HashMap::new(),
            None,
            metrics_reporter,
            crate::testing::test_pipeline_runtime_services(),
        );
        // active = true, needs_timing = false.
        eh.set_flow_roles(
            true, false, None, None, None, None, None, None, None, None, true, false,
        );
        assert!(eh.flow.active);
        assert!(!eh.flow.needs_timing);

        eh.begin_process_timing();

        let mut value = 0u64;
        for i in 0..10_000 {
            value = value.wrapping_add(std::hint::black_box(i));
        }
        let _ = std::hint::black_box(value);

        assert_eq!(
            eh.take_elapsed_since_send_marker_ns(),
            0,
            "send marker must stay unarmed when timing is disabled"
        );
    }

    #[test]
    fn shared_handler_record_flow_duration_drains_to_metric_set() {
        let (ctx, _) = test_pipeline_ctx();
        let entity_key = ctx
            .metrics_registry()
            .register_entity(FlowAttributeSet::default());
        let registrar = ctx.metric_set_registrar_for_entity(entity_key);
        let start_metric_set = FlowInputItemsMetrics::register(&registrar);
        let duration_metric_set = FlowDurationNormalMetrics::register(&registrar);
        let outgoing_metric_set = FlowOutputItemsMetrics::register(&registrar);

        let (metrics_rx, metrics_reporter) = MetricsReporter::create_new_and_receiver(5);
        let mut eh = EffectHandler::<u64>::new(
            test_node("proc"),
            HashMap::new(),
            None,
            metrics_reporter,
            crate::testing::test_pipeline_runtime_services(),
        );
        eh.set_flow_roles(
            true,
            true,
            None,
            Some(start_metric_set),
            None,
            Some(duration_metric_set.into()),
            Some(outgoing_metric_set),
            None,
            None,
            None,
            true,
            true,
        );
        assert!(eh.is_flow_start());
        assert!(eh.is_flow_end());

        eh.record_flow_input_items(SignalType::Logs, 10);
        eh.record_flow_input_items(SignalType::Metrics, 20);
        for ns in [1000, 2000, 3000] {
            eh.record_flow_duration(SignalType::Logs, ns);
        }
        eh.record_flow_output_items(SignalType::Logs, 7);
        eh.record_flow_output_items(SignalType::Metrics, 8);

        let start_before_report = eh
            .flow
            .input
            .input_items
            .as_ref()
            .unwrap()
            .1
            .lock()
            .unwrap();
        assert_eq!(*start_before_report, [0, 20, 10]);

        let before_report = eh.flow.end.duration.as_ref().unwrap().lock().unwrap();
        let (count, sum, _, _) = before_report.pending_summary(SignalType::Logs);
        assert_eq!(count, 3);
        assert!((sum - 0.000_006).abs() < f64::EPSILON);
        drop(before_report);
        let output_before_report = eh.flow.end.output_items.as_ref().unwrap().1.lock().unwrap();
        assert_eq!(*output_before_report, [0, 8, 7]);

        drop(start_before_report);
        drop(output_before_report);
        eh.report_flow_metrics();

        let start_drained = eh
            .flow
            .input
            .input_items
            .as_ref()
            .unwrap()
            .1
            .lock()
            .unwrap();
        assert_eq!(
            *start_drained, [0; 3],
            "start accumulator should be drained"
        );

        let drained = eh.flow.end.duration.as_ref().unwrap().lock().unwrap();
        assert_eq!(
            drained.pending_summary(SignalType::Logs).0,
            0,
            "duration accumulator should be drained"
        );
        let output_drained = eh.flow.end.output_items.as_ref().unwrap().1.lock().unwrap();
        assert_eq!(
            *output_drained, [0; 3],
            "stop item accumulator should be drained"
        );

        let snapshot = metrics_rx
            .try_recv()
            .expect("start flow_metric metric should be reported");
        let [MetricValue::U64(consumed_snapshot)] = snapshot.get_metrics() else {
            panic!("expected one flow input-item metric");
        };
        assert_eq!(
            snapshot.measurement_attribute_value("signal"),
            Some("metrics")
        );
        assert_eq!(*consumed_snapshot, 20);

        let snapshot = metrics_rx
            .try_recv()
            .expect("logs flow input-item metric should be reported");
        let [MetricValue::U64(consumed_snapshot)] = snapshot.get_metrics() else {
            panic!("expected one flow input-item metric");
        };
        assert_eq!(snapshot.measurement_attribute_value("signal"), Some("logs"));
        assert_eq!(*consumed_snapshot, 10);

        let snapshot = metrics_rx
            .try_recv()
            .expect("flow duration metric should be reported");
        let [MetricValue::Distribution(duration_snapshot)] = snapshot.get_metrics() else {
            panic!("expected flow duration histogram");
        };
        let (count, sum, _, _) = duration_snapshot.summary();
        assert_eq!(count, 3);
        assert!((sum - 0.000_006).abs() < f64::EPSILON);
        assert_eq!(snapshot.measurement_attribute_value("signal"), Some("logs"));

        let snapshot = metrics_rx
            .try_recv()
            .expect("metrics flow output-item metric should be reported");
        let [MetricValue::U64(produced_snapshot)] = snapshot.get_metrics() else {
            panic!("expected flow output-item metric");
        };
        assert_eq!(
            snapshot.measurement_attribute_value("signal"),
            Some("metrics")
        );
        assert_eq!(*produced_snapshot, 8);

        let snapshot = metrics_rx
            .try_recv()
            .expect("logs flow output-item metric should be reported");
        let [MetricValue::U64(produced_snapshot)] = snapshot.get_metrics() else {
            panic!("expected flow output-item metric");
        };
        assert_eq!(snapshot.measurement_attribute_value("signal"), Some("logs"));
        assert_eq!(*produced_snapshot, 7);
    }

    #[test]
    fn shared_flow_message_and_size_accumulators_drain_to_metric_sets() {
        let (ctx, _) = test_pipeline_ctx();
        let entity_key = ctx
            .metrics_registry()
            .register_entity(FlowAttributeSet::default());
        let registrar = ctx.metric_set_registrar_for_entity(entity_key);
        let input_messages = FlowInputMessageMetrics::register(&registrar);
        let input_size = FlowInputSizeMetrics::register(&registrar);
        let output_messages = FlowOutputMessageMetrics::register(&registrar);
        let output_size = FlowOutputSizeMetrics::register(&registrar);
        let (_metrics_rx, metrics_reporter) = MetricsReporter::create_new_and_receiver(16);
        let mut handler = EffectHandler::<u64>::new(
            test_node("proc"),
            HashMap::new(),
            None,
            metrics_reporter,
            crate::testing::test_pipeline_runtime_services(),
        );
        handler.set_flow_roles(
            true,
            true,
            Some(input_messages),
            None,
            Some(input_size),
            None,
            None,
            Some(output_messages),
            Some(output_size),
            None,
            true,
            false,
        );

        handler.record_flow_input_message(SignalType::Logs);
        handler.record_flow_input_size(SignalType::Logs, 10);
        handler.record_flow_output_message(SignalType::Metrics);
        handler.record_flow_output_size(SignalType::Metrics, 20);

        assert_eq!(
            *handler
                .flow
                .input
                .input_messages
                .as_ref()
                .unwrap()
                .1
                .lock()
                .unwrap(),
            [0, 0, 1]
        );
        assert_eq!(
            *handler
                .flow
                .end
                .output_size
                .as_ref()
                .unwrap()
                .1
                .lock()
                .unwrap(),
            [0, 20, 0]
        );

        handler.report_flow_metrics();

        assert_eq!(
            *handler
                .flow
                .input
                .input_messages
                .as_ref()
                .unwrap()
                .1
                .lock()
                .unwrap(),
            [0; 3]
        );
        assert_eq!(
            *handler
                .flow
                .input
                .input_size
                .as_ref()
                .unwrap()
                .1
                .lock()
                .unwrap(),
            [0; 3]
        );
        assert_eq!(
            *handler
                .flow
                .end
                .output_messages
                .as_ref()
                .unwrap()
                .1
                .lock()
                .unwrap(),
            [0; 3]
        );
        assert_eq!(
            *handler
                .flow
                .end
                .output_size
                .as_ref()
                .unwrap()
                .1
                .lock()
                .unwrap(),
            [0; 3]
        );
    }
}