saluki_components/encoders/datadog/metrics/

mod.rs

use std::{collections::VecDeque, ops::Range, time::Duration};

use agent_data_plane_config::shared::{Endpoints, MetricsEncoding};
use async_trait::async_trait;
use ddsketch::DDSketch;
use facet::Facet;
use http::{HeaderValue, Method, Request};
use protobuf::{rt::WireType, CodedOutputStream};
use saluki_common::{
    buf::{ChunkedBytesBuffer, FrozenChunkedBytesBuffer},
    iter::ReusableDeduplicator,
    task::HandleExt as _,
};
use saluki_config::GenericConfiguration;
use saluki_context::tags::{SharedTagSet, Tag};
use saluki_core::accounting::{MemoryBounds, MemoryBoundsBuilder};
use saluki_core::{
    components::{encoders::*, ComponentContext},
    data_model::{
        event::{
            metric::{Metric, MetricOrigin, MetricValues},
            EventType,
        },
        payload::{HttpPayload, Payload, PayloadMetadata, PayloadType},
    },
    observability::ComponentMetricsExt as _,
    topology::{EventsBuffer, PayloadsBuffer},
};
use saluki_error::{generic_error, ErrorContext as _, GenericError};
use saluki_io::compression::{CompressionScheme, Compressor};
use saluki_metrics::MetricsBuilder;
use serde::Deserialize;
use tokio::{io::AsyncWriteExt as _, select, sync::mpsc, time::sleep};
use tracing::{debug, error, warn};
use url::Url;
use uuid::Uuid;

#[cfg(test)]
use self::shadow::shadow_sample_matches;
use self::{
    shadow::{SeriesShadowConfig, SeriesShadowState},
    v3::{
        V3EncodedMetrics, V3EncodedRequest, V3EncoderStats, V3MetricType, V3PayloadLimits, V3PayloadRequest,
        V3PayloadSplitReason, V3SerializerTelemetry, V3Writer,
    },
};
use crate::{
    common::datadog::{
        clamp_payload_limits, default_serializer_compressor_kind,
        endpoints::{
            calculate_resolved_endpoint, default_site, deserialize_dd_url, series_v3_config_can_enable_v3,
            AdditionalEndpoints, EndpointV3Settings, ResolvedEndpoint, V3EndpointConfig, DEFAULT_SITE,
        },
        io::RB_BUFFER_CHUNK_SIZE,
        protocol::{MetricsPayloadInfo, UseV3ApiSeriesConfig, V3ApiConfig},
        request_builder::{RequestBuilder, RequestBuilderError},
        resolve_zstd_compressor_level,
        telemetry::ComponentTelemetry,
        DEFAULT_SERIALIZER_COMPRESSED_SIZE_LIMIT, DEFAULT_SERIALIZER_UNCOMPRESSED_SIZE_LIMIT, METRICS_SERIES_V3_PATH,
        METRICS_SKETCHES_V3_PATH,
    },
    encoders::datadog::metrics::v2::MetricsEndpointEncoder,
};

mod endpoint;
use self::endpoint::{EndpointConfiguration, MetricsEndpoint};

mod shadow;
mod v1;
mod v2;
mod v3;

const V3_SERIES_ENDPOINT_URI: &str = METRICS_SERIES_V3_PATH;
const V3_SKETCHES_ENDPOINT_URI: &str = METRICS_SKETCHES_V3_PATH;

const fn default_max_metrics_per_payload() -> usize {
    10_000
}

const fn default_max_payload_size() -> usize {
    DEFAULT_SERIALIZER_COMPRESSED_SIZE_LIMIT
}

const fn default_max_uncompressed_payload_size() -> usize {
    DEFAULT_SERIALIZER_UNCOMPRESSED_SIZE_LIMIT
}

const fn default_max_series_payload_size() -> usize {
    v2::SERIES_V2_COMPRESSED_SIZE_LIMIT
}

const fn default_max_series_uncompressed_payload_size() -> usize {
    v2::SERIES_V2_UNCOMPRESSED_SIZE_LIMIT
}

const fn default_max_series_points_per_payload() -> usize {
    10_000
}

const fn default_flush_timeout_secs() -> u64 {
    2
}

const fn default_use_v2_api_series() -> bool {
    true
}

const fn default_log_payloads() -> bool {
    false
}

fn series_shadow_config_for_endpoint(
    series_endpoint: MetricsEndpoint, sample_rate: f64, metrics_v3_disabled_by_compressor: bool,
) -> SeriesShadowConfig {
    SeriesShadowConfig::new(
        if !metrics_v3_disabled_by_compressor && series_endpoint == MetricsEndpoint::SeriesV2 {
            sample_rate
        } else {
            0.0
        },
    )
}

/// Encoding mode for a metrics endpoint.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum MetricsEncoderMode {
    /// Send V2 payloads only.
    V2Only,
    /// V3 is enabled for at least one endpoint; generate tagged V2 and V3 payloads so each endpoint
    /// receives the protocol version configured for it.
    V3Enabled,
    /// Send both V2 and V3 payloads simultaneously with a shared batch ID for backend validation.
    Validation,
}

impl MetricsEncoderMode {
    fn from_config(use_v3: bool, validate: bool) -> Self {
        match (use_v3, validate) {
            (false, _) => Self::V2Only,
            (true, false) => Self::V3Enabled,
            (true, true) => Self::Validation,
        }
    }

    fn needs_v3(self) -> bool {
        matches!(self, Self::V3Enabled | Self::Validation)
    }

    fn needs_batch_id(self) -> bool {
        matches!(self, Self::Validation)
    }

    fn needs_tagging(self) -> bool {
        matches!(self, Self::V3Enabled | Self::Validation)
    }
}

fn metrics_encoder_mode_for_config(
    use_v3: bool, validate: bool, metrics_v3_disabled_by_compressor: bool,
) -> MetricsEncoderMode {
    let use_v3 = use_v3 && !metrics_v3_disabled_by_compressor;
    MetricsEncoderMode::from_config(use_v3, use_v3 && validate)
}

fn selected_metrics_primary_v3_override(
    opw_enabled: bool, opw_url: &str, opw_use_v3_series: bool, vector_enabled: bool, vector_url: &str,
    vector_use_v3_series: bool,
) -> Option<bool> {
    selected_metrics_primary_endpoint(
        opw_enabled,
        opw_url,
        opw_use_v3_series,
        vector_enabled,
        vector_url,
        vector_use_v3_series,
    )
    .map(|(_, use_v3_series)| use_v3_series)
}

fn selected_metrics_primary_endpoint<'a>(
    opw_enabled: bool, opw_url: &'a str, opw_use_v3_series: bool, vector_enabled: bool, vector_url: &'a str,
    vector_use_v3_series: bool,
) -> Option<(&'a str, bool)> {
    if opw_enabled {
        let opw_url = opw_url.trim();
        metrics_primary_url_can_resolve(opw_url).then_some((opw_url, opw_use_v3_series))
    } else if vector_enabled {
        let vector_url = vector_url.trim();
        metrics_primary_url_can_resolve(vector_url).then_some((vector_url, vector_use_v3_series))
    } else {
        None
    }
}

fn metrics_primary_url_can_resolve(url: &str) -> bool {
    if url.is_empty() {
        return false;
    }
    if url.starts_with("http://") || url.starts_with("https://") {
        Url::parse(url).is_ok_and(|url| url.host_str().is_some())
    } else {
        Url::parse(&format!("https://{url}")).is_ok_and(|url| url.host_str().is_some())
    }
}

fn series_v3_can_be_enabled_for_config(
    use_v2_api_series: bool, serializer_use_v3_series: bool, metrics_primary_v3_override: Option<bool>,
    has_additional_endpoints: bool, series_config: &UseV3ApiSeriesConfig,
) -> bool {
    use_v2_api_series
        && (serializer_use_v3_series
            || metrics_primary_v3_override == Some(true)
            || ((metrics_primary_v3_override != Some(false) || has_additional_endpoints)
                && series_v3_config_can_enable_v3(series_config)))
}

/// Datadog Metrics encoder.
///
/// Generates Datadog metrics payloads for the Datadog platform.
#[derive(Clone, Deserialize, Facet)]
#[cfg_attr(test, derive(Debug, PartialEq, serde::Serialize))]
pub struct DatadogMetricsConfiguration {
    /// Maximum number of input metrics to encode into a single request payload.
    ///
    /// This applies both to the series and sketches endpoints.
    ///
    /// Defaults to 10,000.
    #[serde(
        rename = "serializer_max_metrics_per_payload",
        default = "default_max_metrics_per_payload"
    )]
    max_metrics_per_payload: usize,

    /// Maximum compressed size, in bytes, of generic payloads.
    ///
    /// This applies to V1 JSON series payloads and sketch payloads, matching the Datadog Agent's generic payload
    /// builder. V2 series payloads use `serializer_max_series_payload_size` instead. The effective value is clamped to
    /// the Agent's default intake-safe limit of 2,621,440 bytes, so larger configured values do not allow payloads that
    /// intake may reject. If set to `0`, every non-empty compressed payload exceeds the limit and is dropped during
    /// flush.
    ///
    /// Defaults to 2,621,440 bytes.
    #[serde(rename = "serializer_max_payload_size", default = "default_max_payload_size")]
    max_payload_size: usize,

    /// Maximum uncompressed size, in bytes, of generic payloads.
    ///
    /// This applies to V1 JSON series payloads and sketch payloads, matching the Datadog Agent's generic payload
    /// builder. V2 series payloads use `serializer_max_series_uncompressed_payload_size` instead. The effective value
    /// is clamped to the Agent's default intake-safe limit of 4,194,304 bytes, so larger configured values do not allow
    /// payloads that intake may reject. Values smaller than the minimum endpoint framing size prevent the request
    /// builder from starting.
    ///
    /// Defaults to 4,194,304 bytes.
    #[serde(
        rename = "serializer_max_uncompressed_payload_size",
        default = "default_max_uncompressed_payload_size"
    )]
    max_uncompressed_payload_size: usize,

    /// Maximum compressed size, in bytes, of a V2 series payload.
    ///
    /// This applies only when `use_v2_api.series` is `true`. V1 series and sketches use `serializer_max_payload_size`
    /// instead. The effective value is clamped to the V2 series API limit of 512,000 bytes, so larger configured values
    /// do not allow payloads that intake would reject. High-throughput workloads may increase this up to that API limit
    /// to reduce request count, at the cost of larger individual requests. If set to `0`, every non-empty compressed
    /// payload exceeds the limit and is dropped during flush.
    ///
    /// Defaults to 512,000 bytes.
    #[serde(
        rename = "serializer_max_series_payload_size",
        default = "default_max_series_payload_size"
    )]
    max_series_payload_size: usize,

    /// Maximum uncompressed size, in bytes, of a V2 series payload.
    ///
    /// This applies only when `use_v2_api.series` is `true`. V1 series and sketches use
    /// `serializer_max_uncompressed_payload_size` instead. The effective value is clamped to the V2 series API limit of
    /// 5,242,880 bytes, so larger configured values do not allow payloads that intake would reject. This limit protects
    /// the encoder before compression, so compressed payload size may still force a separate flush. Values smaller than
    /// the minimum endpoint framing size prevent the request builder from starting.
    ///
    /// Defaults to 5,242,880 bytes.
    #[serde(
        rename = "serializer_max_series_uncompressed_payload_size",
        default = "default_max_series_uncompressed_payload_size"
    )]
    max_series_uncompressed_payload_size: usize,

    /// Maximum number of data points, across all series, to encode into a single series request payload.
    ///
    /// This applies only to series metrics (counters, gauges, rates, sets) and not to sketch metrics (histograms,
    /// distributions). A single metric series may contribute multiple data points when it carries more than one
    /// timestamp/value pair. When encoding an input would cause the running data point total to exceed this limit, the
    /// current payload is flushed first and the input is placed in the next payload.
    ///
    /// Defaults to 10,000.
    #[serde(
        rename = "serializer_max_series_points_per_payload",
        default = "default_max_series_points_per_payload"
    )]
    max_series_points_per_payload: usize,

    /// Flush timeout for pending requests, in seconds.
    ///
    /// When the destination has written metrics to the in-flight request payload, but it has not yet reached the
    /// payload size limits that would force the payload to be flushed, the destination will wait for a period of time
    /// before flushing the in-flight request payload. This allows for the possibility of other events to be processed
    /// and written into the request payload, thereby maximizing the payload size and reducing the number of requests
    /// generated and sent overall.
    ///
    /// Defaults to 2 seconds.
    #[serde(default = "default_flush_timeout_secs")]
    flush_timeout_secs: u64,

    /// Compression kind to use for the request payloads.
    ///
    /// Defaults to `zstd`.
    #[serde(
        rename = "serializer_compressor_kind",
        default = "default_serializer_compressor_kind"
    )]
    compressor_kind: String,

    /// ADP-specific zstd compression level, taking precedence over `serializer_zstd_compressor_level`.
    ///
    /// See [`resolve_zstd_compressor_level`] for how the effective level is determined.
    #[serde(rename = "data_plane_serializer_zstd_compressor_level", default)]
    data_plane_zstd_compressor_level: Option<i32>,

    /// The Core Agent's zstd compression level. Used only when set to a non-default value (not 1),
    /// since the Agent forwards its default over the config stream.
    ///
    /// See [`resolve_zstd_compressor_level`] for how the effective level is determined.
    #[serde(rename = "serializer_zstd_compressor_level", default)]
    serializer_zstd_compressor_level: Option<i32>,

    /// Whether to use the V2 API for series metrics.
    ///
    /// When `true` (the default), series metrics are sent to the V2 protobuf endpoint (`/api/v2/series`). When
    /// `false`, series metrics are sent to the legacy V1 JSON endpoint (`/api/v1/series`). Sketch metrics always use
    /// the V2 endpoint (`/api/beta/sketches`) regardless of this setting.
    ///
    /// Defaults to `true`.
    #[serde(default = "default_use_v2_api_series")]
    use_v2_api_series: bool,

    /// Whether to log metric payload contents before encoding.
    ///
    /// This logs decoded metric objects, not the encoded JSON/protobuf HTTP body.
    ///
    /// Defaults to `false`.
    #[serde(default = "default_log_payloads")]
    log_payloads: bool,

    /// Additional tags to apply to all forwarded metrics.
    #[serde(default, skip)]
    #[facet(opaque)]
    additional_tags: Option<SharedTagSet>,

    /// V3 API configuration for per-endpoint V3 support.
    ///
    /// Configures which endpoints receive V3 payloads and whether validation mode is enabled.
    #[serde(rename = "serializer_experimental_use_v3_api", default)]
    v3_api: V3ApiConfig,

    /// Agent-compatible V3 API configuration for series metrics.
    #[serde(flatten)]
    use_v3_api_series: UseV3ApiSeriesConfig,

    /// Enables routing all metrics to Observability Pipelines Worker.
    #[serde(default, rename = "observability_pipelines_worker_metrics_enabled")]
    observability_pipelines_worker_metrics_enabled: bool,

    /// Endpoint of the Observability Pipelines Worker instance to route metrics to.
    #[serde(default, rename = "observability_pipelines_worker_metrics_url")]
    observability_pipelines_worker_metrics_url: String,

    /// Enables V3 series metrics when routing to Observability Pipelines Worker.
    ///
    /// Defaults to `false`.
    #[serde(default, rename = "observability_pipelines_worker_metrics_use_v3_api_series")]
    observability_pipelines_worker_metrics_use_v3_api_series: bool,

    /// Enables routing all metrics to Vector.
    #[serde(default, rename = "vector_metrics_enabled")]
    vector_metrics_enabled: bool,

    /// Endpoint of the Vector instance to route metrics to.
    #[serde(default, rename = "vector_metrics_url")]
    vector_metrics_url: String,

    /// Enables V3 series metrics when routing to Vector.
    ///
    /// Deprecated in favor of `observability_pipelines_worker.metrics.use_v3_api.series`.
    ///
    /// Defaults to `false`.
    #[serde(default, rename = "vector_metrics_use_v3_api_series")]
    vector_metrics_use_v3_api_series: bool,

    /// The Datadog site used to resolve the primary metrics endpoint.
    ///
    /// Defaults to `datadoghq.com`.
    #[serde(default = "default_site")]
    site: String,

    /// The optional explicit primary metrics endpoint.
    ///
    /// Defaults to unset, in which case `site` determines the endpoint.
    #[serde(default, alias = "url", deserialize_with = "deserialize_dd_url")]
    dd_url: Option<String>,

    /// Additional endpoints that metrics may be dual-shipped to.
    #[serde(default)]
    additional_endpoints: AdditionalEndpoints,
}

impl DatadogMetricsConfiguration {
    /// Creates a new `DatadogMetricsConfiguration` from the given configuration.
    pub fn from_configuration(config: &GenericConfiguration) -> Result<Self, GenericError> {
        Ok(config.as_typed()?)
    }

    /// Creates a metrics encoder using authoritative typed metrics-routing configuration.
    pub fn from_configuration_with_metrics_routing(
        config: &GenericConfiguration, metrics: &MetricsEncoding, endpoints: &Endpoints,
    ) -> Result<Self, GenericError> {
        let mut metrics_config = Self::from_configuration(config)?;

        metrics_config.compressor_kind = endpoints.compression.compressor_kind.clone();
        metrics_config.use_v2_api_series = metrics.use_v2_series_api;
        metrics_config.v3_api = (&metrics.v3_api).into();
        metrics_config.use_v3_api_series = (&metrics.v3_series_mode).into();
        metrics_config.observability_pipelines_worker_metrics_enabled = endpoints.opw_intake.enabled;
        metrics_config.observability_pipelines_worker_metrics_url = endpoints.opw_intake.url.clone();
        metrics_config.observability_pipelines_worker_metrics_use_v3_api_series = endpoints.opw_intake.use_v3_series;
        metrics_config.vector_metrics_enabled = endpoints.vector_intake.enabled;
        metrics_config.vector_metrics_url = endpoints.vector_intake.url.clone();
        metrics_config.vector_metrics_use_v3_api_series = endpoints.vector_intake.use_v3_series;

        Ok(metrics_config)
    }

    /// Sets additional tags to be applied uniformly to all metrics forwarded by this destination.
    pub fn with_additional_tags(mut self, additional_tags: SharedTagSet) -> Self {
        self.additional_tags = Some(additional_tags);
        self
    }

    /// Restricts endpoint-aware protocol selection to a single overridden metrics endpoint.
    ///
    /// This mirrors a forwarder branch that replaces the normal primary endpoint and removes additional and
    /// OPW/Vector endpoints, such as Multi-Region Failover.
    pub fn with_metrics_endpoint_override(mut self, dd_url: String) -> Self {
        self.dd_url = Some(dd_url);
        self.additional_endpoints = AdditionalEndpoints::default();
        self.observability_pipelines_worker_metrics_enabled = false;
        self.vector_metrics_enabled = false;
        self
    }

    /// Forces series metrics to use V2 without producing V3 shadow payloads.
    ///
    /// This is used for local destinations that only accept the V2 series protocol, such as the Cluster Agent.
    pub fn with_v2_series_only(mut self) -> Self {
        self.use_v3_api_series.enabled = "false".to_string();
        self.use_v3_api_series.endpoints.clear();
        self.v3_api.series.endpoints.clear();
        self.v3_api.series.shadow_sample_rate = 0.0;
        self.observability_pipelines_worker_metrics_use_v3_api_series = false;
        self.vector_metrics_use_v3_api_series = false;
        self
    }

    fn v3_payload_limits(&self) -> V3PayloadLimits {
        V3PayloadLimits::new(
            self.max_series_payload_size,
            self.max_series_uncompressed_payload_size,
            self.max_metrics_per_payload,
            self.max_series_points_per_payload,
        )
    }

    fn endpoint_v3_settings(
        &self, endpoint: &ResolvedEndpoint, metrics_primary_v3_override: Option<bool>,
        serializer_v3_configured_endpoint: Option<&str>,
    ) -> EndpointV3Settings {
        EndpointV3Settings::from_v3_config(V3EndpointConfig {
            configured_endpoint: endpoint.configured_endpoint(),
            resolved_endpoint: endpoint.endpoint(),
            serializer_v3_configured_endpoint,
            series_config: &self.use_v3_api_series,
            metrics_primary_v3_override,
            serializer_v3_series_endpoints: &self.v3_api.series.endpoints,
            serializer_v3_sketches_endpoints: &self.v3_api.sketches.endpoints,
            series_validate: self.v3_api.series.validate,
            sketches_validate: self.v3_api.sketches.validate,
            series_shadow_sites: &self.v3_api.series.shadow_sites,
        })
    }

    fn configured_primary_endpoint(&self) -> String {
        match self.dd_url.as_deref() {
            Some(url) => url.to_string(),
            None => {
                let base_domain = if self.site.is_empty() { DEFAULT_SITE } else { &self.site };
                format!("https://app.{base_domain}")
            }
        }
    }

    fn any_series_endpoint_matches(
        &self, mut predicate: impl FnMut(&EndpointV3Settings) -> bool,
    ) -> Result<bool, GenericError> {
        let configured_primary_endpoint = self.configured_primary_endpoint();
        if let Some((metrics_primary_url, metrics_primary_v3_override)) = selected_metrics_primary_endpoint(
            self.observability_pipelines_worker_metrics_enabled,
            &self.observability_pipelines_worker_metrics_url,
            self.observability_pipelines_worker_metrics_use_v3_api_series,
            self.vector_metrics_enabled,
            &self.vector_metrics_url,
            self.vector_metrics_use_v3_api_series,
        ) {
            let metrics_primary = calculate_resolved_endpoint(Some(metrics_primary_url), &self.site, "")
                .error_context("Failed parsing/resolving the metrics primary destination endpoint.")?;
            let settings = self.endpoint_v3_settings(
                &metrics_primary,
                Some(metrics_primary_v3_override),
                Some(&configured_primary_endpoint),
            );
            if predicate(&settings) {
                return Ok(true);
            }
        } else {
            let primary = calculate_resolved_endpoint(self.dd_url.as_deref(), &self.site, "")
                .error_context("Failed parsing/resolving the primary destination endpoint.")?;
            let settings = self.endpoint_v3_settings(&primary, None, None);
            if predicate(&settings) {
                return Ok(true);
            }
        }

        for endpoint in self
            .additional_endpoints
            .resolved_endpoints(None)
            .error_context("Failed parsing/resolving the additional destination endpoints.")?
        {
            let settings = self.endpoint_v3_settings(&endpoint, None, None);
            if predicate(&settings) {
                return Ok(true);
            }
        }

        Ok(false)
    }

    fn requires_v2_series(&self, metrics_v3_disabled_by_compressor: bool) -> Result<bool, GenericError> {
        if !self.use_v2_api_series || metrics_v3_disabled_by_compressor {
            return Ok(true);
        }

        self.any_series_endpoint_matches(|settings| !settings.use_v3_series || settings.series_validation_mode)
    }

    fn requires_v3_series(&self, metrics_v3_disabled_by_compressor: bool) -> Result<bool, GenericError> {
        if metrics_v3_disabled_by_compressor {
            return Ok(false);
        }

        let metrics_primary_v3_override = selected_metrics_primary_v3_override(
            self.observability_pipelines_worker_metrics_enabled,
            &self.observability_pipelines_worker_metrics_url,
            self.observability_pipelines_worker_metrics_use_v3_api_series,
            self.vector_metrics_enabled,
            &self.vector_metrics_url,
            self.vector_metrics_use_v3_api_series,
        );
        if !series_v3_can_be_enabled_for_config(
            self.use_v2_api_series,
            self.v3_api.use_v3_series(),
            metrics_primary_v3_override,
            !self.additional_endpoints.is_empty(),
            &self.use_v3_api_series,
        ) {
            return Ok(false);
        }

        self.any_series_endpoint_matches(|settings| settings.use_v3_series)
    }
}

#[async_trait]
impl EncoderBuilder for DatadogMetricsConfiguration {
    fn input_event_type(&self) -> EventType {
        EventType::Metric
    }

    fn output_payload_type(&self) -> PayloadType {
        PayloadType::Http
    }

    async fn build(&self, context: ComponentContext) -> Result<Box<dyn Encoder + Send>, GenericError> {
        let metrics_builder = MetricsBuilder::from_component_context(&context);
        let telemetry = ComponentTelemetry::from_builder(&metrics_builder);
        let v3_serializer_telemetry = V3SerializerTelemetry::from_builder(&metrics_builder);

        let zstd_compressor_level = resolve_zstd_compressor_level(
            self.data_plane_zstd_compressor_level,
            self.serializer_zstd_compressor_level,
        );
        let v2_compression_scheme = CompressionScheme::new(&self.compressor_kind, zstd_compressor_level);
        let v3_compression_scheme = if self.v3_api.compression_level > 0 {
            CompressionScheme::new(&self.compressor_kind, self.v3_api.compression_level)
        } else {
            v2_compression_scheme
        };
        let series_endpoint_uri = if self.v3_api.series.use_beta {
            self.v3_api.series.beta_route.clone()
        } else {
            V3_SERIES_ENDPOINT_URI.to_string()
        };
        let shadow_series_endpoint_uri = self.v3_api.series.beta_route.clone();
        let payload_limits = self.v3_payload_limits();

        let v2_endpoint_config = EndpointConfiguration::new(
            v2_compression_scheme,
            self.max_metrics_per_payload,
            self.max_series_points_per_payload,
            self.additional_tags.clone(),
        );
        let endpoint_config = EndpointConfiguration::new(
            v3_compression_scheme,
            self.max_metrics_per_payload,
            // Actually enforced by V3PayloadLimits, required for the
            // constructor shared between V1/V2/V3.
            usize::MAX,
            self.additional_tags.clone(),
        );

        // Derive the encoding mode for each metric type from the configuration.
        let metrics_v3_disabled_by_compressor = matches!(v3_compression_scheme, CompressionScheme::Zlib(_));
        let use_v3_series = self.requires_v3_series(metrics_v3_disabled_by_compressor)?;
        let series_mode = metrics_encoder_mode_for_config(
            use_v3_series,
            self.v3_api.series.validate,
            metrics_v3_disabled_by_compressor,
        );
        let sketches_mode = metrics_encoder_mode_for_config(
            self.v3_api.use_v3_sketches(),
            self.v3_api.sketches.validate,
            metrics_v3_disabled_by_compressor,
        );
        let series_endpoint = if self.use_v2_api_series {
            MetricsEndpoint::SeriesV2
        } else {
            MetricsEndpoint::SeriesV1
        };
        let series_shadow_config = series_shadow_config_for_endpoint(
            series_endpoint,
            self.v3_api.series.shadow_sample_rate,
            metrics_v3_disabled_by_compressor,
        );
        let v3_runtime_config = V3RuntimeConfig {
            endpoint_config,
            payload_limits,
            series_endpoint_uri,
            shadow_series_endpoint_uri,
            series_shadow_config,
            serializer_telemetry: v3_serializer_telemetry,
        };
        let generic_payload_limits = clamp_payload_limits(
            self.max_uncompressed_payload_size,
            self.max_payload_size,
            DEFAULT_SERIALIZER_UNCOMPRESSED_SIZE_LIMIT,
            DEFAULT_SERIALIZER_COMPRESSED_SIZE_LIMIT,
        );
        let (series_uncompressed_limit, series_compressed_limit) = if series_endpoint == MetricsEndpoint::SeriesV2 {
            clamp_payload_limits(
                self.max_series_uncompressed_payload_size,
                self.max_series_payload_size,
                v2::SERIES_V2_UNCOMPRESSED_SIZE_LIMIT,
                v2::SERIES_V2_COMPRESSED_SIZE_LIMIT,
            )
        } else {
            generic_payload_limits
        };
        let v2_series_builder = if self.requires_v2_series(metrics_v3_disabled_by_compressor)? {
            let mut builder = v2::create_v2_request_builder(series_endpoint, &v2_endpoint_config)
                .await
                .error_context("Failed to create V2 series request builder.")?;
            builder.with_len_limits(series_uncompressed_limit, series_compressed_limit)?;
            Some(builder)
        } else {
            debug!("All metrics series endpoints use authoritative V3; disabling V2 series encoding.");
            None
        };

        let (sketches_uncompressed_limit, sketches_compressed_limit) = generic_payload_limits;
        let mut v2_sketch_builder = v2::create_v2_request_builder(MetricsEndpoint::Sketches, &v2_endpoint_config)
            .await
            .error_context("Failed to create V2 sketches request builder.")?;
        v2_sketch_builder.with_len_limits(sketches_uncompressed_limit, sketches_compressed_limit)?;
        let v2_sketch_builder = Some(v2_sketch_builder);

        let flush_timeout = match self.flush_timeout_secs {
            // We always give ourselves a minimum flush timeout of 10ms to allow for some very minimal amount of
            // batching, while still practically flushing things almost immediately.
            0 => Duration::from_millis(10),
            secs => Duration::from_secs(secs),
        };

        if series_mode.needs_v3() || sketches_mode.needs_v3() {
            debug!(
                ?series_mode,
                ?sketches_mode,
                v3_series_endpoints = ?self.v3_api.series.endpoints,
                v3_sketches_endpoints = ?self.v3_api.sketches.endpoints,
                "V3 encoding support is enabled."
            );
        }

        Ok(Box::new(DatadogMetrics {
            v2_series_builder,
            v2_sketch_builder,
            series_mode,
            sketches_mode,
            v3_runtime_config,
            telemetry,
            flush_timeout,
            log_payloads: self.log_payloads,
        }))
    }
}

impl MemoryBounds for DatadogMetricsConfiguration {
    fn specify_bounds(&self, builder: &mut MemoryBoundsBuilder) {
        // TODO: How do we properly represent the requests we can generate that may be sitting around in-flight?
        //
        // Theoretically, we'll end up being limited by the size of the downstream forwarder's interconnect, and however
        // many payloads it will buffer internally... so realistically the firm limit boils down to the forwarder itself
        // but we'll have a hard time in the forwarder knowing the maximum size of any given payload being sent in, which
        // then makes it hard to calculate a proper firm bound even though we know the rest of the values required to
        // calculate the firm bound.
        builder
            .minimum()
            .with_single_value::<DatadogMetrics>("component struct")
            .with_array::<EventsBuffer>("request builder events channel", 8)
            .with_array::<PayloadsBuffer>("request builder payloads channel", 8);

        builder
            .firm()
            // Capture the size of the "split re-encode" buffers in the request builders, which is where we keep owned
            // versions of metrics that we encode in case we need to actually re-encode them during a split operation.
            .with_array::<Metric>("series metrics split re-encode buffer", self.max_metrics_per_payload)
            .with_array::<Metric>("sketch metrics split re-encode buffer", self.max_metrics_per_payload);
    }
}

pub struct DatadogMetrics {
    v2_series_builder: Option<RequestBuilder<v2::MetricsEndpointEncoder>>,
    v2_sketch_builder: Option<RequestBuilder<v2::MetricsEndpointEncoder>>,
    series_mode: MetricsEncoderMode,
    sketches_mode: MetricsEncoderMode,
    v3_runtime_config: V3RuntimeConfig,
    telemetry: ComponentTelemetry,
    flush_timeout: Duration,
    log_payloads: bool,
}

struct V3RuntimeConfig {
    endpoint_config: EndpointConfiguration,
    payload_limits: V3PayloadLimits,
    series_endpoint_uri: String,
    shadow_series_endpoint_uri: String,
    series_shadow_config: SeriesShadowConfig,
    serializer_telemetry: V3SerializerTelemetry,
}

#[async_trait]
impl Encoder for DatadogMetrics {
    async fn run(mut self: Box<Self>, mut context: EncoderContext) -> Result<(), GenericError> {
        let Self {
            v2_series_builder,
            v2_sketch_builder,
            series_mode,
            sketches_mode,
            v3_runtime_config,
            telemetry,
            flush_timeout,
            log_payloads,
        } = *self;

        let mut health = context.take_health_handle();

        // Spawn our request builder task.
        let (events_tx, events_rx) = mpsc::channel(8);
        let (payloads_tx, mut payloads_rx) = mpsc::channel(8);
        let request_builder_fut = run_request_builder(
            v2_series_builder,
            v2_sketch_builder,
            series_mode,
            sketches_mode,
            v3_runtime_config,
            telemetry,
            events_rx,
            payloads_tx,
            flush_timeout,
            log_payloads,
        );
        let request_builder_handle = context
            .topology_context()
            .global_thread_pool()
            .spawn_traced_named("dd-metrics-request-builder", request_builder_fut);

        health.mark_ready();
        debug!("Datadog Metrics encoder started.");

        loop {
            select! {
                biased;

                _ = health.live() => continue,
                maybe_payload = payloads_rx.recv() => match maybe_payload {
                    Some(payload) => {
                        if let Err(e) = context.dispatcher().dispatch(payload).await {
                            error!("Failed to dispatch payload: {}", e);
                        }
                    }
                    None => break,
                },
                maybe_event_buffer = context.events().next() => match maybe_event_buffer {
                    Some(event_buffer) => {
                        // Both channels between this task and the request builder task are bounded, and each task is
                        // the producer on one and the consumer on the other. Blocking outright on `events_tx` would
                        // stop us draining `payloads_rx`, which can deadlock: the builder task blocks sending a
                        // payload once `payloads_tx` is full, so it never returns to receive from `events_rx`, so
                        // capacity never frees up here. Keep draining payloads while we wait for capacity so that at
                        // least one of the two channels is always making progress.
                        let permit = loop {
                            select! {
                                biased;

                                permit = events_tx.reserve() => break permit
                                    .error_context("Failed to reserve capacity for event buffer.")?,
                                maybe_payload = payloads_rx.recv() => match maybe_payload {
                                    Some(payload) => if let Err(e) = context.dispatcher().dispatch(payload).await {
                                        error!("Failed to dispatch payload: {}", e);
                                    },

                                    // Our payloads channel is gone, which means our request builder task went away unexpectedly.
                                    None => return Err(generic_error!("Request builder task stopped before accepting event buffer.")),
                                }
                            }
                        };

                        permit.send(event_buffer);
                    }
                    None => break,
                },
            }
        }

        // Drop the events sender, which signals the request builder task to stop.
        drop(events_tx);

        // Continue draining the payloads receiver until it is closed.
        while let Some(payload) = payloads_rx.recv().await {
            if let Err(e) = context.dispatcher().dispatch(payload).await {
                error!("Failed to dispatch payload: {}", e);
            }
        }

        // Request build task should now be stopped.
        match request_builder_handle.await {
            Ok(Ok(())) => debug!("Request builder task stopped."),
            Ok(Err(e)) => error!(error = %e, "Request builder task failed."),
            Err(e) => error!(error = %e, "Request builder task panicked."),
        }

        debug!("Datadog Metrics encoder stopped.");

        Ok(())
    }
}

/// Logs the decoded contents of a metric prior to encoding.
///
/// This logs the metric object itself, not the encoded JSON/protobuf HTTP body.
fn log_metric_payload(metric: &Metric) {
    match metric.values() {
        MetricValues::Counter(..) | MetricValues::Rate(..) | MetricValues::Gauge(..) | MetricValues::Set(..) => {
            debug!(?metric, "Flushing series metric.")
        }
        MetricValues::Histogram(..) | MetricValues::Distribution(..) => {
            debug!(?metric, "Flushing sketch metric.")
        }
    }
}

#[allow(clippy::too_many_arguments)]
async fn run_request_builder(
    mut v2_series_builder: Option<RequestBuilder<v2::MetricsEndpointEncoder>>,
    mut v2_sketch_builder: Option<RequestBuilder<v2::MetricsEndpointEncoder>>, series_mode: MetricsEncoderMode,
    sketches_mode: MetricsEncoderMode, v3_runtime_config: V3RuntimeConfig, telemetry: ComponentTelemetry,
    mut events_rx: mpsc::Receiver<EventsBuffer>, mut payloads_tx: mpsc::Sender<PayloadsBuffer>,
    flush_timeout: Duration, log_payloads: bool,
) -> Result<(), GenericError> {
    let mut pending_flush = false;
    let pending_flush_timeout = sleep(flush_timeout);
    tokio::pin!(pending_flush_timeout);

    let mut v3_series_metrics =
        (series_mode.needs_v3() || v3_runtime_config.series_shadow_config.is_enabled()).then(Vec::<Metric>::new);
    let mut v3_sketch_metrics = sketches_mode.needs_v3().then(Vec::<Metric>::new);
    let mut v3_series_points = 0usize;
    let mut v3_sketch_points = 0usize;
    let mut v3_series_ratio = V3CompressionRatio::default();
    let mut v3_sketch_ratio = V3CompressionRatio::default();

    let mut series_batch_id = None;
    let mut sketches_batch_id = None;
    let mut series_shadow_state = SeriesShadowState::default();
    let series_shadow_config = v3_runtime_config.series_shadow_config;

    let tag_series = series_mode.needs_tagging();
    let tag_sketches = sketches_mode.needs_tagging();
    let v3_flush_context = V3FlushContext {
        endpoint_config: &v3_runtime_config.endpoint_config,
        payload_limits: v3_runtime_config.payload_limits,
        series_endpoint_uri: &v3_runtime_config.series_endpoint_uri,
        serializer_telemetry: &v3_runtime_config.serializer_telemetry,
        telemetry: &telemetry,
    };
    let v3_shadow_flush_context = V3FlushContext {
        endpoint_config: &v3_runtime_config.endpoint_config,
        payload_limits: v3_runtime_config.payload_limits,
        series_endpoint_uri: &v3_runtime_config.shadow_series_endpoint_uri,
        serializer_telemetry: &v3_runtime_config.serializer_telemetry,
        telemetry: &telemetry,
    };

    loop {
        select! {
            Some(event_buffer) = events_rx.recv() => {
                for event in event_buffer {
                    let metric = match event.try_into_metric() {
                        Some(metric) => metric,
                        None => continue,
                    };

                    if log_payloads {
                        log_metric_payload(&metric);
                    }

                    // A series metric whose points are all non-finite would encode to a series with no points, which
                    // intake rejects as an empty value set. Drop it whole rather than emit an empty series.
                    if !v1::has_emittable_point(&metric) {
                        debug!(metric = %metric.context().name(), "Dropping series metric with no finite points.");
                        telemetry.events_dropped_encoder().increment(1);
                        continue;
                    }

                    // Figure out which endpoint the metric belongs to, and grab the relevant V2 builder/V3 storage.
                    let endpoint = MetricsEndpoint::from_metric(&metric);
                    let (endpoint_mode, maybe_v2_builder, maybe_v3_metrics, v3_points, batch_id) = match endpoint {
                        MetricsEndpoint::SeriesV1 | MetricsEndpoint::SeriesV2 => (
                            series_mode,
                            &mut v2_series_builder,
                            &mut v3_series_metrics,
                            &mut v3_series_points,
                            &mut series_batch_id,
                        ),
                        MetricsEndpoint::Sketches => (
                            sketches_mode,
                            &mut v2_sketch_builder,
                            &mut v3_sketch_metrics,
                            &mut v3_sketch_points,
                            &mut sketches_batch_id,
                        ),
                    };
                    let metric_point_count = metric.values().len();
                    let is_series = matches!(endpoint, MetricsEndpoint::SeriesV1 | MetricsEndpoint::SeriesV2);
                    let series_shadow_active = is_series
                        && matches!(endpoint_mode, MetricsEncoderMode::V2Only)
                        && series_shadow_state.ensure_decision(series_shadow_config);
                    let needs_batch_id = endpoint_mode.needs_batch_id() || series_shadow_active;
                    if needs_batch_id && batch_id.is_none() {
                        *batch_id = Some(Uuid::now_v7());
                    }
                    let active_batch_id = needs_batch_id.then_some(batch_id.as_ref()).flatten();
                    let should_buffer_v3 = endpoint_mode.needs_v3() || series_shadow_active;

                    // Store a copy of the metric in `maybe_v3_metrics` if it's present.
                    //
                    // We have to do this before encoding because `RequestBuilder::encode` consumes the metric. This also means we'll
                    // need to _remove_ the metric if encoding fails.
                    if should_buffer_v3 {
                        if let Some(metrics) = maybe_v3_metrics {
                            metrics.push(metric.clone());
                            *v3_points += metric_point_count;
                        }
                    }

                    // Attempt encoding the metric for V2 if configured.
                    //
                    // If the metric couldn't be encoded (too big, some other issue), the call returns `false` which is
                    // our signal to remove the metric from `maybe_v3_metrics` (if we added it), since we know now that
                    // the metric wasn't encoded for V2 and we want our V2/V3 payload batches to be consistent in
                    // validation mode.
                    let v2_payload_info = match endpoint {
                        MetricsEndpoint::SeriesV1 | MetricsEndpoint::SeriesV2 => {
                            if series_shadow_active {
                                Some(MetricsPayloadInfo::v2_shadow_series())
                            } else {
                                tag_series.then(MetricsPayloadInfo::v2_series)
                            }
                        }
                        MetricsEndpoint::Sketches => tag_sketches.then(MetricsPayloadInfo::v2_sketches),
                    };
                    // If V2 flushes while this batch is not shadowed, the current metric may start the next V2 batch.
                    // Keep a clone so we can add it to the next V3 shadow batch if that next batch samples in.
                    let metric_for_next_shadow_batch = (is_series
                        && matches!(endpoint_mode, MetricsEncoderMode::V2Only)
                        && series_shadow_config.is_enabled()
                        && !series_shadow_active)
                        .then(|| metric.clone());
                    let mut v2_encoded = false;
                    let v2_flushed = if let Some(builder) = maybe_v2_builder {
                        let result =
                            encode_v2_metrics(builder, metric, &telemetry, &mut payloads_tx, active_batch_id, v2_payload_info).await?;
                        v2_encoded = result.encoded();
                        if should_buffer_v3
                            && !result.encoded()
                            && !matches!(endpoint_mode, MetricsEncoderMode::V3Enabled)
                        {
                            if let Some(metrics) = maybe_v3_metrics {
                                let _ = metrics.pop();
                                *v3_points = v3_points.saturating_sub(metric_point_count);
                            }
                        }

                        result.flushed()
                    } else {
                        false
                    };

                    // Validation and shadow payloads must keep V2/V3 batch boundaries aligned. Authoritative V3 can
                    // batch independently, so it flushes on V3-specific limits instead of following V2 flushes.
                    let v3_payload_info = match endpoint {
                        MetricsEndpoint::SeriesV1 | MetricsEndpoint::SeriesV2 => {
                            if series_shadow_active {
                                Some(MetricsPayloadInfo::v3_shadow_series())
                            } else {
                                tag_series.then(MetricsPayloadInfo::v3_series)
                            }
                        }
                        MetricsEndpoint::Sketches => tag_sketches.then(MetricsPayloadInfo::v3_sketches),
                    };
                    let mut split_metric = None;
                    let v3_flushed = if let Some(v3_metrics) = maybe_v3_metrics {
                        let should_flush_v3 = match endpoint_mode {
                            MetricsEncoderMode::V2Only => series_shadow_active && v2_flushed,
                            MetricsEncoderMode::V3Enabled => {
                                if v3_flush_context.payload_limits.point_count_fits(metric_point_count)
                                    && v3_flush_context.payload_limits.point_count_exceeds_limit(*v3_points)
                                    && v3_metrics.len() > 1
                                {
                                    v3_flush_context
                                        .serializer_telemetry
                                        .record_split_reason(V3PayloadSplitReason::MaxPoints);
                                    if let Some(metric) = v3_metrics.pop() {
                                        *v3_points = v3_points.saturating_sub(metric.values().len());
                                        split_metric = Some(metric);
                                    }
                                    true
                                } else {
                                    v3_flush_context.payload_limits.should_flush_point_count_limit(*v3_points)
                                        || v3_flush_context
                                            .payload_limits
                                            .should_flush_metric_count_limit(v3_metrics)
                                }
                            }
                            MetricsEncoderMode::Validation => v2_flushed,
                        };
                        if should_flush_v3 {
                            if !matches!(endpoint_mode, MetricsEncoderMode::V3Enabled) {
                                // V2 flushes the previous batch without the current metric. Pop it
                                // from V3 before flushing so both batches cover the same set of metrics.
                                if v2_flushed {
                                    if let Some(metric) = v3_metrics.pop() {
                                        *v3_points = v3_points.saturating_sub(metric.values().len());
                                        split_metric = Some(metric);
                                    }
                                }
                            }
                            let flush_context = if series_shadow_active {
                                v3_shadow_flush_context
                            } else {
                                v3_flush_context
                            };
                            encode_and_flush_v3_metrics(
                                endpoint,
                                flush_context,
                                v3_metrics,
                                &mut v3_series_ratio,
                                &mut v3_sketch_ratio,
                                &mut payloads_tx,
                                active_batch_id,
                                v3_payload_info,
                            )
                            .await?;
                            *v3_points = 0;
                            true
                        } else {
                            false
                        }
                    } else {
                        false
                    };

                    if matches!(endpoint_mode, MetricsEncoderMode::V3Enabled) {
                        // Authoritative V3 may flush the previous point-limit batch while keeping the current metric
                        // buffered for the next V3 payload.
                        if let Some(m) = split_metric.take() {
                            let point_count = m.values().len();
                            if let Some(metrics) = maybe_v3_metrics {
                                metrics.push(m);
                                *v3_points += point_count;
                            }
                        }
                    } else if endpoint_mode.needs_batch_id() && (v2_flushed || v3_flushed) {
                        // A validation flush completes the current V2/V3 pair. If V2 carried the current metric into
                        // the next batch, keep the V3 copy and assign that next pair a fresh validation ID.
                        *batch_id = if let Some(m) = split_metric.take() {
                            let point_count = m.values().len();
                            if let Some(metrics) = maybe_v3_metrics {
                                metrics.push(m);
                                *v3_points += point_count;
                            }
                            Some(Uuid::now_v7())
                        } else {
                            None
                        };
                    } else if is_series && series_shadow_active && (v2_flushed || v3_flushed) {
                        // A shadow flush completes the current sampled pair. The next V2 batch gets a new shadow sample
                        // decision, so only carry the split metric into V3 if that next batch samples in.
                        series_shadow_state.reset();
                        *batch_id = if let Some(m) = split_metric.take() {
                            if series_shadow_state.ensure_decision(series_shadow_config) {
                                let point_count = m.values().len();
                                if let Some(metrics) = maybe_v3_metrics {
                                    metrics.push(m);
                                    *v3_points += point_count;
                                }
                                Some(Uuid::now_v7())
                            } else {
                                None
                            }
                        } else {
                            None
                        };
                    } else if is_series
                        && matches!(endpoint_mode, MetricsEncoderMode::V2Only)
                        && series_shadow_config.is_enabled()
                        && v2_flushed
                    {
                        // This V2 batch was not shadowed, but the flushed metric may have started the next V2 batch.
                        // Re-sample for that next batch and seed the V3 buffer only if the new decision samples in.
                        series_shadow_state.reset();
                        *batch_id = None;
                        if v2_encoded {
                            if let Some(m) = metric_for_next_shadow_batch {
                                if series_shadow_state.ensure_decision(series_shadow_config) {
                                    let point_count = m.values().len();
                                    if let Some(metrics) = maybe_v3_metrics {
                                        metrics.push(m);
                                        *v3_points += point_count;
                                    }
                                    *batch_id = Some(Uuid::now_v7());
                                }
                            }
                        }
                    }
                }

                debug!("Processed event buffer.");

                // If we're not already pending a flush, we'll start the countdown.
                if !pending_flush {
                    pending_flush_timeout.as_mut().reset(tokio::time::Instant::now() + flush_timeout);
                    pending_flush = true;
                }
            },
            _ = &mut pending_flush_timeout, if pending_flush => {
                debug!("Flushing pending request(s).");

                pending_flush = false;

                // Flush any pending series metrics.
                // Timeout flushes complete the current batch, so reuse the existing shadow decision instead of sampling
                // a new one for metrics that are already buffered.
                let series_shadow_active = matches!(series_mode, MetricsEncoderMode::V2Only)
                    && series_shadow_state.active.unwrap_or(false);
                let v2_series_payload_info = if series_shadow_active {
                    Some(MetricsPayloadInfo::v2_shadow_series())
                } else {
                    tag_series.then(MetricsPayloadInfo::v2_series)
                };
                let series_active_batch_id = (series_mode.needs_batch_id() || series_shadow_active)
                    .then_some(series_batch_id.as_ref())
                    .flatten();
                let mut v2_series_flush_succeeded = true;
                if let Some(builder) = &mut v2_series_builder {
                    if let Err(e) = flush_v2_metrics(builder, &mut payloads_tx, series_active_batch_id, v2_series_payload_info).await {
                        error!(error = %e, "Failed to flush V2 series metrics: {}", e);
                        v2_series_flush_succeeded = false;
                    }
                }

                let v3_series_payload_info = if series_shadow_active {
                    Some(MetricsPayloadInfo::v3_shadow_series())
                } else {
                    tag_series.then(MetricsPayloadInfo::v3_series)
                };
                if let Some(metrics) = &mut v3_series_metrics {
                    if v2_series_flush_succeeded || matches!(series_mode, MetricsEncoderMode::V3Enabled) {
                        // Shadow series use the V3 beta route; normal V3 series use the configured authoritative route.
                        let flush_context = if series_shadow_active {
                            v3_shadow_flush_context
                        } else {
                            v3_flush_context
                        };
                        if let Err(e) = encode_and_flush_v3_series_metrics(
                            flush_context,
                            metrics,
                            &mut v3_series_ratio,
                            &mut payloads_tx,
                            series_active_batch_id,
                            v3_series_payload_info,
                        )
                        .await
                        {
                            error!(error = %e, "Failed to flush V3 series metrics: {}", e);
                        }
                        v3_series_points = 0;
                    } else {
                        // Validation/shadow V3 must not outlive a failed V2 baseline flush.
                        warn!("Failed to flush V2 series metrics, skipping V3 series flush.");
                        metrics.clear();
                        v3_series_points = 0;
                    }
                }
                if series_mode.needs_batch_id() {
                    series_batch_id = None;
                }
                if matches!(series_mode, MetricsEncoderMode::V2Only) && series_shadow_config.is_enabled() {
                    series_shadow_state.reset();
                    series_batch_id = None;
                }

                // Flush any pending sketch metrics.
                let v2_sketches_payload_info = tag_sketches.then(MetricsPayloadInfo::v2_sketches);
                let sketches_active_batch_id = sketches_mode.needs_batch_id().then_some(sketches_batch_id.as_ref()).flatten();
                let mut v2_sketches_flush_succeeded = true;
                if let Some(builder) = &mut v2_sketch_builder {
                    if let Err(e) = flush_v2_metrics(builder, &mut payloads_tx, sketches_active_batch_id, v2_sketches_payload_info).await {
                        error!(error = %e, "Failed to flush V2 sketch metrics: {}", e);
                        v2_sketches_flush_succeeded = false;
                    }
                }

                let v3_sketches_payload_info = tag_sketches.then(MetricsPayloadInfo::v3_sketches);
                if let Some(metrics) = &mut v3_sketch_metrics {
                    if v2_sketches_flush_succeeded || matches!(sketches_mode, MetricsEncoderMode::V3Enabled) {
                        if let Err(e) = encode_and_flush_v3_sketch_metrics(
                            v3_flush_context,
                            metrics,
                            &mut v3_sketch_ratio,
                            &mut payloads_tx,
                            sketches_active_batch_id,
                            v3_sketches_payload_info,
                        )
                        .await
                        {
                            error!(error = %e, "Failed to flush V3 sketch metrics: {}", e);
                        }
                        v3_sketch_points = 0;
                    } else {
                        warn!("Failed to flush V2 sketch metrics, skipping V3 sketch flush.");
                        metrics.clear();
                        v3_sketch_points = 0;
                    }
                }
                if sketches_mode.needs_batch_id() {
                    sketches_batch_id = None;
                }

                debug!("All flushed requests sent to I/O task. Waiting for next event buffer...");
            },

            // Event buffers channel has been closed, and we have no pending flushing, so we're all done.
            else => break,
        }
    }

    Ok(())
}

struct EncodeResult {
    encoded: bool,
    flushed: bool,
}

impl EncodeResult {
    pub const fn new(encoded: bool, flushed: bool) -> Self {
        Self { encoded, flushed }
    }

    pub const fn encoded(&self) -> bool {
        self.encoded
    }

    pub const fn flushed(&self) -> bool {
        self.flushed
    }
}

async fn encode_v2_metrics(
    request_builder: &mut RequestBuilder<v2::MetricsEndpointEncoder>, metric: Metric, telemetry: &ComponentTelemetry,
    payloads_tx: &mut mpsc::Sender<Payload>, batch_id: Option<&Uuid>, payload_info: Option<MetricsPayloadInfo>,
) -> Result<EncodeResult, GenericError> {
    // Encode the metric. If we get it back, that means the current request is full, and we need to
    // flush it before we can try to encode the metric again... so we'll hold on to it in that case
    // before flushing and trying to encode it again.
    let metric_to_retry = match request_builder.encode(metric).await {
        Ok(None) => return Ok(EncodeResult::new(true, false)),
        Ok(Some(metric)) => metric,
        Err(RequestBuilderError::InvalidInput { input }) => {
            debug!(metric_name = %input.context().name(), "Dropping metric with no emittable values.");
            telemetry.events_dropped_encoder().increment(1);
            return Ok(EncodeResult::new(false, false));
        }
        Err(e) => {
            error!(error = %e, "Failed to encode metric.");
            telemetry.events_dropped_encoder().increment(1);
            return Ok(EncodeResult::new(false, false));
        }
    };

    flush_v2_metrics(request_builder, payloads_tx, batch_id, payload_info).await?;

    // Now try to encode the metric again. If it fails again, we'll just log it because it shouldn't
    // be possible to fail at this point, otherwise we would have already caught that the first
    // time.
    match request_builder.encode(metric_to_retry).await {
        Ok(None) => Ok(EncodeResult::new(true, true)),
        Ok(Some(_)) => unreachable!(
            "failure to encode due to size should never occur after flush for metrics which aren't unencodable"
        ),
        Err(e) => {
            error!(error = %e, "Failed to encode metric.");
            telemetry.events_dropped_encoder().increment(1);
            Ok(EncodeResult::new(false, true))
        }
    }
}

async fn flush_v2_metrics(
    request_builder: &mut RequestBuilder<MetricsEndpointEncoder>, payloads_tx: &mut mpsc::Sender<Payload>,
    batch_id: Option<&Uuid>, payload_info: Option<MetricsPayloadInfo>,
) -> Result<usize, GenericError> {
    let mut requests_flushed = 0;

    let maybe_requests = request_builder.flush().await;
    let batch_len = maybe_requests.len();
    for (batch_seq, maybe_request) in maybe_requests.into_iter().enumerate() {
        match maybe_request {
            Ok((events, data_points, request)) => {
                requests_flushed += 1;

                flush_payload(
                    request,
                    events,
                    data_points,
                    payloads_tx,
                    batch_id,
                    batch_seq,
                    batch_len,
                    payload_info,
                )
                .await?;
            }

            // TODO: Increment a counter here that metrics were dropped due to a flush failure.
            Err(e) => {
                if !e.is_recoverable() {
                    return Err(GenericError::from(e).context("Failed to flush request."));
                }
            }
        }
    }

    Ok(requests_flushed)
}

#[derive(Clone, Copy)]
struct V3FlushContext<'a> {
    endpoint_config: &'a EndpointConfiguration,
    payload_limits: V3PayloadLimits,
    series_endpoint_uri: &'a str,
    serializer_telemetry: &'a V3SerializerTelemetry,
    telemetry: &'a ComponentTelemetry,
}

#[allow(clippy::too_many_arguments)]
async fn encode_and_flush_v3_metrics(
    endpoint: MetricsEndpoint, context: V3FlushContext<'_>, metrics: &mut Vec<Metric>,
    series_ratio: &mut V3CompressionRatio, sketches_ratio: &mut V3CompressionRatio,
    payloads_tx: &mut mpsc::Sender<Payload>, batch_id: Option<&Uuid>, payload_info: Option<MetricsPayloadInfo>,
) -> Result<(), GenericError> {
    match endpoint {
        MetricsEndpoint::SeriesV1 | MetricsEndpoint::SeriesV2 => {
            encode_and_flush_v3_series_metrics(context, metrics, series_ratio, payloads_tx, batch_id, payload_info)
                .await
        }
        MetricsEndpoint::Sketches => {
            encode_and_flush_v3_sketch_metrics(context, metrics, sketches_ratio, payloads_tx, batch_id, payload_info)
                .await
        }
    }
}

async fn encode_and_flush_v3_series_metrics(
    context: V3FlushContext<'_>, metrics: &mut Vec<Metric>, ratio: &mut V3CompressionRatio,
    payloads_tx: &mut mpsc::Sender<Payload>, batch_id: Option<&Uuid>, payload_info: Option<MetricsPayloadInfo>,
) -> Result<(), GenericError> {
    if metrics.is_empty() {
        return Ok(());
    }
    let metrics_to_flush = std::mem::take(metrics);

    encode_and_flush_v3_payload_requests(
        context.series_endpoint_uri,
        &metrics_to_flush,
        context,
        "series",
        ratio,
        payloads_tx,
        batch_id,
        payload_info,
    )
    .await
}

async fn encode_and_flush_v3_sketch_metrics(
    context: V3FlushContext<'_>, metrics: &mut Vec<Metric>, ratio: &mut V3CompressionRatio,
    payloads_tx: &mut mpsc::Sender<Payload>, batch_id: Option<&Uuid>, payload_info: Option<MetricsPayloadInfo>,
) -> Result<(), GenericError> {
    if metrics.is_empty() {
        return Ok(());
    }
    let metrics_to_flush = std::mem::take(metrics);

    encode_and_flush_v3_payload_requests(
        V3_SKETCHES_ENDPOINT_URI,
        &metrics_to_flush,
        context,
        "sketches",
        ratio,
        payloads_tx,
        batch_id,
        payload_info,
    )
    .await
}

/// Encodes `metrics` into one or more V3 payload requests, forwarding each one to the I/O task.
///
/// Requests are sent as they are built rather than accumulated, so a batch that splits into many payloads does not
/// hold all of them in memory at once. The exception is validation/shadow batches, which carry an
/// `X-Metrics-Request-Len` header that is only known once every payload in the batch has been built; those are
/// collected first and sent afterwards.
#[allow(clippy::too_many_arguments)]
/// Smallest compression ratio used when translating the compressed size limit into an uncompressed byte target.
///
/// Guards against dividing by a pathologically small ratio and producing an enormous target.
const V3_MIN_COMPRESSION_RATIO: f64 = 0.02;

/// Fraction of the derived byte target that a batch is actually cut at.
///
/// Leaves headroom so that ordinary variation between batches does not push the finalized payload over the real limit
/// and force the expensive re-encode path.
const V3_BATCH_TARGET_MARGIN: f64 = 0.9;

/// Weight given to the newest observation when updating the compression ratio.
const V3_RATIO_SMOOTHING: f64 = 0.25;

/// Tracks the compression ratio observed on emitted V3 payloads, so batches can be sized against the compressed limit.
///
/// V3 payloads are columnar and only compressed once the whole batch is finalized, so unlike the V2 request builder
/// there is no live compressor to consult with a [`CompressionEstimator`] while batching. Instead we remember how well
/// recent payloads compressed and use that to turn the compressed size limit (which is the limit that actually binds
/// for metrics intake) into an uncompressed byte target for the next batch.
#[derive(Default)]
struct V3CompressionRatio {
    ratio: Option<f64>,
}

impl V3CompressionRatio {
    /// Returns the uncompressed byte target that a batch should be cut at.
    fn uncompressed_target(&self, limits: V3PayloadLimits) -> usize {
        // Until a payload has been observed, assume the data will not compress at all. That errs towards batches that
        // are too small, which costs an extra payload, rather than too large, which costs a full re-encode.
        let ratio = self.ratio.unwrap_or(1.0).clamp(V3_MIN_COMPRESSION_RATIO, 1.0);
        let target_from_compressed_limit = (limits.max_compressed_size as f64 / ratio) as usize;
        let target = target_from_compressed_limit.min(limits.max_uncompressed_size);

        // Always leave room for at least one metric, however small the limits are.
        (((target as f64) * V3_BATCH_TARGET_MARGIN) as usize).max(1)
    }

    /// Records the uncompressed and compressed sizes of an emitted payload.
    fn record(&mut self, uncompressed_len: usize, compressed_len: usize) {
        if uncompressed_len == 0 {
            return;
        }

        let observed = compressed_len as f64 / uncompressed_len as f64;
        self.ratio = Some(match self.ratio {
            Some(current) => (current * (1.0 - V3_RATIO_SMOOTHING)) + (observed * V3_RATIO_SMOOTHING),
            None => observed,
        });
    }
}

/// A contiguous run of metrics encoded into a single V3 payload.
struct V3Batch {
    encoded: V3EncodedMetrics,
    event_count: usize,
    data_point_count: usize,
}

/// Encodes metrics starting at `*idx` into a single V3 payload, stopping once a payload limit or
/// `target_uncompressed_len` is reached, and advances `*idx` past everything consumed.
///
/// Sizing the batch while encoding it is what keeps this to a single encode pass. The alternative (encoding a large
/// batch and halving it whenever the result is too big) re-encodes and re-compresses the same metrics once per level of
/// splitting, which is dominated by dictionary rebuilding on high-cardinality data
fn build_v3_batch(
    metrics: &[Metric], idx: &mut usize, context: V3FlushContext<'_>, payload_kind: &'static str,
    target_uncompressed_len: usize,
) -> Option<V3Batch> {
    let mut writer = V3Writer::new();
    let mut tags_deduplicator = ReusableDeduplicator::new();
    let additional_tags = context.endpoint_config.additional_tags();
    let mut data_point_count = 0usize;
    let mut cut_for_size = false;

    while *idx < metrics.len() {
        let metric = &metrics[*idx];

        if !metric_has_emittable_values(metric) {
            debug!(metric_name = %metric.context().name(), "Dropping metric with no emittable values.");
            context.telemetry.events_dropped_encoder().increment(1);
            *idx += 1;
            continue;
        }

        let metric_points = metric.values().len();
        if !context.payload_limits.point_count_fits(metric_points) {
            // This metric exceeds the point limit on its own, so it cannot fit in any payload.
            context.serializer_telemetry.record_item_too_big();
            context
                .serializer_telemetry
                .record_split_reason(V3PayloadSplitReason::ItemTooBig);
            warn!(
                payload_kind,
                data_points = metric_points,
                point_limit = context.payload_limits.max_points_per_payload,
                "Dropping oversized V3 metric that exceeds the point-count limit."
            );
            context.telemetry.events_dropped_encoder().increment(1);
            *idx += 1;
            continue;
        }

        // Limits that have to be honoured before the metric goes in. The first metric of a batch is always admitted,
        // since it has already been checked to fit by itself.
        if writer.metric_count() > 0 {
            if !context
                .payload_limits
                .point_count_fits(data_point_count + metric_points)
            {
                context
                    .serializer_telemetry
                    .record_split_reason(V3PayloadSplitReason::MaxPoints);
                break;
            }

            if context.payload_limits.metric_count_reached(writer.metric_count()) {
                break;
            }
        }

        write_metric_to_v3(&mut writer, metric, additional_tags, &mut tags_deduplicator);
        data_point_count += metric_points;
        *idx += 1;

        // The size contribution of a metric is only knowable once it is written, since dictionary deduplication makes
        // it depend on what came before. Cutting after the fact can overshoot by one metric, which the target's
        // headroom absorbs; the finalized payload is checked against the real limits regardless.
        if writer.estimated_uncompressed_len() >= target_uncompressed_len {
            cut_for_size = true;
            break;
        }
    }

    // Only count this as a payload boundary if metrics actually remain: a batch that ends on the last metric was not
    // split by anything.
    if cut_for_size && *idx < metrics.len() {
        context
            .serializer_telemetry
            .record_split_reason(V3PayloadSplitReason::PayloadFull);
    }

    let event_count = writer.metric_count();
    if event_count == 0 {
        return None;
    }

    match writer.finalize() {
        Ok(encoded) => Some(V3Batch {
            encoded,
            event_count,
            data_point_count,
        }),
        Err(e) => {
            error!(error = %e, payload_kind, events = event_count, "Failed to encode V3 metrics payload request.");
            context.telemetry.events_dropped_encoder().increment(event_count as u64);
            None
        }
    }
}

/// Turns an encoded batch into a payload request, returning `None` if it does not fit the payload limits.
async fn create_v3_batch_request(
    endpoint_uri: &str, batch: V3Batch, context: V3FlushContext<'_>, payload_kind: &'static str,
    ratio: &mut V3CompressionRatio,
) -> Option<V3PayloadRequest> {
    let V3Batch {
        encoded,
        event_count,
        data_point_count,
    } = batch;

    let mut encoded_request =
        match create_v3_request(endpoint_uri, encoded, context.endpoint_config.compression_scheme()).await {
            Ok(request) => request,
            Err(e) => {
                error!(error = %e, payload_kind, events = event_count, "Failed to create V3 metrics request.");
                context.telemetry.events_dropped_encoder().increment(event_count as u64);
                return None;
            }
        };

    // Feed the observation back even when the request is too big: an oversized batch is exactly the case where the
    // target needs correcting.
    ratio.record(encoded_request.uncompressed_len, encoded_request.compressed_len);

    if !context.payload_limits.request_fits(&encoded_request) {
        return None;
    }

    // Per-column compressed sizes are measured only for requests we actually emit. Measuring them eagerly in
    // `create_v3_request` would compress every column of every oversized attempt that gets discarded, which is pure
    // waste.
    if let Err(e) =
        measure_v3_column_compressed_sizes(&mut encoded_request.stats, context.endpoint_config.compression_scheme())
            .await
    {
        error!(error = %e, payload_kind, "Failed to measure V3 column compressed sizes.");
    } else {
        record_v3_serializer_stats(context.serializer_telemetry, &encoded_request.stats);
    }

    Some(V3PayloadRequest {
        request: encoded_request.request,
        event_count,
        data_point_count,
    })
}

/// Re-encodes an oversized range as progressively smaller ranges until each one fits.
///
/// This is the fallback for a batch that was sized against the byte target but still exceeded the real payload limits.
/// It re-encodes and re-compresses at every level of splitting, so it is deliberately only reached when the target
/// guessed wrong.
async fn split_and_encode_oversized_v3_range(
    endpoint_uri: &str, metrics: &[Metric], range: Range<usize>, context: V3FlushContext<'_>,
    payload_kind: &'static str, ratio: &mut V3CompressionRatio,
) -> Vec<V3PayloadRequest> {
    // `build_v3_batch` advances past metrics it drops. Preserve that decision when the accepted batch has to be
    // re-encoded: retrying the raw source range would otherwise resurrect those metrics.
    let accepted_metrics = metrics[range]
        .iter()
        .filter(|metric| {
            metric_has_emittable_values(metric) && context.payload_limits.point_count_fits(metric.values().len())
        })
        .collect::<Vec<_>>();

    let mut requests = Vec::new();
    let mut pending_ranges = VecDeque::new();
    pending_ranges.push_back(0..accepted_metrics.len());

    while let Some(range) = pending_ranges.pop_front() {
        if range.is_empty() {
            continue;
        }

        let metrics_in_range = &accepted_metrics[range.clone()];
        let event_count = metrics_in_range.len();
        let data_point_count = metrics_in_range.iter().map(|metric| metric.values().len()).sum();

        let encoded = match encode_v3_metrics_batch(
            metrics_in_range.iter().copied(),
            context.endpoint_config.additional_tags(),
        ) {
            Ok(encoded) => encoded,
            Err(e) => {
                error!(error = %e, payload_kind, events = event_count, "Failed to encode V3 metrics payload request.");
                context.telemetry.events_dropped_encoder().increment(event_count as u64);
                continue;
            }
        };
        let batch = V3Batch {
            encoded,
            event_count,
            data_point_count,
        };

        if let Some(request) = create_v3_batch_request(endpoint_uri, batch, context, payload_kind, ratio).await {
            requests.push(request);
            continue;
        }

        if range.len() == 1 {
            // The encoded request is too large and this range cannot be split any further.
            context.serializer_telemetry.record_item_too_big();
            context
                .serializer_telemetry
                .record_split_reason(V3PayloadSplitReason::ItemTooBig);
            warn!(
                payload_kind,
                compressed_limit = context.payload_limits.max_compressed_size,
                uncompressed_limit = context.payload_limits.max_uncompressed_size,
                "Dropping oversized V3 metric that cannot be split further."
            );
            context.telemetry.events_dropped_encoder().increment(1);
            continue;
        }

        // Retry this oversized range as two smaller ranges, preserving the original metric order.
        context
            .serializer_telemetry
            .record_split_reason(V3PayloadSplitReason::PayloadFull);
        let pivot = range.start + range.len() / 2;
        pending_ranges.push_front(pivot..range.end);
        pending_ranges.push_front(range.start..pivot);
    }

    requests
}

#[allow(clippy::too_many_arguments)]
async fn encode_and_flush_v3_payload_requests(
    endpoint_uri: &str, metrics: &[Metric], context: V3FlushContext<'_>, payload_kind: &'static str,
    ratio: &mut V3CompressionRatio, payloads_tx: &mut mpsc::Sender<Payload>, batch_id: Option<&Uuid>,
    payload_info: Option<MetricsPayloadInfo>,
) -> Result<(), GenericError> {
    // Batches that carry validation headers need the total payload count up front, so they cannot be streamed.
    let stream_payloads = batch_id.is_none();
    let mut pending_requests = Vec::new();
    let target_uncompressed_len = ratio.uncompressed_target(context.payload_limits);
    let mut idx = 0;

    while idx < metrics.len() {
        let batch_start = idx;
        let Some(batch) = build_v3_batch(metrics, &mut idx, context, payload_kind, target_uncompressed_len) else {
            // `build_v3_batch` always consumes at least one metric when one is available, but guard against spinning
            // on a batch that somehow produced nothing.
            if idx == batch_start {
                error!(
                    payload_kind,
                    "V3 batching made no progress; dropping remaining metrics."
                );
                context
                    .telemetry
                    .events_dropped_encoder()
                    .increment((metrics.len() - idx) as u64);
                break;
            }
            continue;
        };

        let requests = match create_v3_batch_request(endpoint_uri, batch, context, payload_kind, ratio).await {
            Some(request) => vec![request],
            // The batch we sized against the byte target still came out over the real limits, so fall back to
            // re-encoding it as progressively smaller ranges. The target self-corrects from the ratio we record on
            // every emitted payload, so this should be rare.
            None => {
                split_and_encode_oversized_v3_range(
                    endpoint_uri,
                    metrics,
                    batch_start..idx,
                    context,
                    payload_kind,
                    ratio,
                )
                .await
            }
        };

        for payload_request in requests {
            if stream_payloads {
                flush_payload(
                    payload_request.request,
                    payload_request.event_count,
                    payload_request.data_point_count,
                    payloads_tx,
                    None,
                    0,
                    0,
                    payload_info,
                )
                .await?;
                debug!(
                    payload_kind,
                    events = payload_request.event_count,
                    data_points = payload_request.data_point_count,
                    "Sent V3 payload."
                );
            } else {
                pending_requests.push(payload_request);
            }
        }
    }

    let batch_len = pending_requests.len();
    for (batch_seq, payload_request) in pending_requests.into_iter().enumerate() {
        flush_payload(
            payload_request.request,
            payload_request.event_count,
            payload_request.data_point_count,
            payloads_tx,
            batch_id,
            batch_seq,
            batch_len,
            payload_info,
        )
        .await?;
        debug!(
            payload_kind,
            events = payload_request.event_count,
            data_points = payload_request.data_point_count,
            "Sent V3 payload."
        );
    }

    Ok(())
}

fn record_v3_serializer_stats(telemetry: &V3SerializerTelemetry, stats: &V3EncoderStats) {
    telemetry.record_values_count(stats.value_encoding_stats);

    for column in &stats.columns {
        let uncompressed_size = column.bytes.len() as u64;
        let compressed_size = column.compressed_len as u64;
        telemetry.record_column_size(column.field_number, uncompressed_size, compressed_size);
    }
}

/// Measures the compressed size of each V3 column, for telemetry purposes.
///
/// This is deliberately separate from building the request: it is only worth paying for columns belonging to a request
/// that will actually be sent, since oversized requests are discarded and re-encoded as smaller ranges.
async fn measure_v3_column_compressed_sizes(
    stats: &mut V3EncoderStats, compression_scheme: CompressionScheme,
) -> Result<(), GenericError> {
    for column in &mut stats.columns {
        column.compressed_len = compressed_v3_len(&column.bytes, compression_scheme)
            .await
            .error_context("Failed to measure V3 column compressed size.")?;
    }

    Ok(())
}

async fn compressed_v3_len(bytes: &[u8], compression_scheme: CompressionScheme) -> Result<usize, GenericError> {
    if matches!(compression_scheme, CompressionScheme::Noop) {
        return Ok(bytes.len());
    }

    let buffer = ChunkedBytesBuffer::new(RB_BUFFER_CHUNK_SIZE);
    let mut compressor = Compressor::from_scheme(compression_scheme, buffer);
    compressor
        .write_all(bytes)
        .await
        .error_context("Failed to compress V3 bytes.")?;
    compressor
        .flush()
        .await
        .error_context("Failed to flush V3 compressor.")?;
    compressor
        .shutdown()
        .await
        .error_context("Failed to shutdown V3 compressor.")?;

    Ok(compressor.into_inner().freeze().len())
}

/// Converts a `Uuid` to a `HeaderValue`.
fn uuid_to_header_value(uuid: &Uuid) -> HeaderValue {
    let s = uuid.as_hyphenated().to_string();
    HeaderValue::try_from(s).expect("hyphenated UUID should be a valid header value")
}

/// Converts a `usize` to a `HeaderValue`.
fn usize_to_header_value(value: usize) -> HeaderValue {
    let s = value.to_string();
    HeaderValue::try_from(s).expect("usize should be a valid header value")
}

async fn flush_payload(
    mut request: Request<FrozenChunkedBytesBuffer>, event_count: usize, data_point_count: usize,
    payloads_tx: &mut mpsc::Sender<Payload>, batch_id: Option<&Uuid>, batch_seq: usize, batch_len: usize,
    payload_info: Option<MetricsPayloadInfo>,
) -> Result<(), GenericError> {
    // Attach the validation batch UUID and sequence headers if present.
    if let Some(batch_id) = batch_id {
        let headers = request.headers_mut();
        headers.insert("X-Metrics-Request-ID", uuid_to_header_value(batch_id));
        headers.insert("X-Metrics-Request-Seq", usize_to_header_value(batch_seq));
        headers.insert("X-Metrics-Request-Len", usize_to_header_value(batch_len));
    }

    let mut payload_meta = PayloadMetadata::from_event_and_data_point_count(event_count, data_point_count);
    if let Some(info) = payload_info {
        payload_meta = payload_meta.with(info);
    }
    let http_payload = HttpPayload::new(payload_meta, request);
    let payload = Payload::Http(http_payload);

    payloads_tx
        .send(payload)
        .await
        .error_context("Failed to send payload.")?;

    Ok(())
}

// Encodes a batch of metrics to V3 columnar format.
fn encode_v3_metrics_batch<'a>(
    metrics: impl IntoIterator<Item = &'a Metric>, additional_tags: &SharedTagSet,
) -> Result<V3EncodedMetrics, GenericError> {
    let mut writer = V3Writer::new();
    let mut tags_deduplicator = ReusableDeduplicator::new();

    for metric in metrics {
        write_metric_to_v3(&mut writer, metric, additional_tags, &mut tags_deduplicator);
    }

    writer
        .finalize()
        .map_err(|e| generic_error!("Failed to serialize V3 payload: {}", e))
}

/// Writes a single metric to the V3 writer.
pub(super) fn sketch_has_emittable_values(sketch: &DDSketch) -> bool {
    !sketch.is_empty() || !sketch.bins().is_empty()
}

pub(super) fn metric_has_emittable_values(metric: &Metric) -> bool {
    match metric.values() {
        MetricValues::Counter(points) | MetricValues::Rate(points, _) | MetricValues::Gauge(points) => {
            points.into_iter().any(|(_, value)| emittable_scalar_point(value))
        }
        MetricValues::Set(points) => points.into_iter().any(|(_, value)| emittable_scalar_point(value)),
        MetricValues::Distribution(sketches) => sketches
            .into_iter()
            .any(|(_, sketch)| sketch_has_emittable_values(sketch)),
        MetricValues::Histogram(points) => points.into_iter().any(|(_, histogram)| !histogram.samples().is_empty()),
    }
}

fn write_metric_to_v3(
    writer: &mut V3Writer, metric: &Metric, additional_tags: &SharedTagSet,
    tags_deduplicator: &mut ReusableDeduplicator<Tag>,
) {
    if !metric_has_emittable_values(metric) {
        debug!(metric_name = %metric.context().name(), "Dropping metric with no emittable values.");
        return;
    }

    let metric_type = match metric.values() {
        MetricValues::Counter(..) => V3MetricType::Count,
        MetricValues::Rate(..) => V3MetricType::Rate,
        MetricValues::Gauge(..) | MetricValues::Set(..) => V3MetricType::Gauge,
        MetricValues::Histogram(..) | MetricValues::Distribution(..) => V3MetricType::Sketch,
    };
    let is_sketch = metric_type == V3MetricType::Sketch;

    let mut builder = writer.write(metric_type, metric.context().name());

    // Tags - chain instrumented + additional + origin tags
    let chained_tags = metric
        .context()
        .tags()
        .into_iter()
        .chain(additional_tags)
        .chain(metric.context().origin_tags());
    let all_tags = tags_deduplicator
        .deduplicated(chained_tags)
        .filter(|t| is_sketch || !is_v3_series_resource_tag(t) && !is_v3_series_device_tag(t))
        .map(|t| t.as_str());
    builder.set_tags(all_tags);

    // Resources - extract host and, for series, promoted resource tags.
    let mut resources = Vec::new();
    if let Some(host) = metric.context().host().filter(|host| !host.is_empty()) {
        resources.push(("host", host));
    }
    if !is_sketch {
        let mut device_resource = None;
        let chained_tags = metric
            .context()
            .origin_tags()
            .into_iter()
            .chain(metric.context().tags())
            .chain(additional_tags);
        for tag in tags_deduplicator.deduplicated(chained_tags) {
            if is_v3_series_device_tag(tag) {
                device_resource = tag.value().filter(|device| !device.is_empty());
            } else if is_v3_series_resource_tag(tag) {
                if let Some((rtype, rname)) = tag.value().and_then(|value| value.split_once(':')) {
                    if !rtype.is_empty() && !rname.is_empty() {
                        resources.push((rtype, rname));
                    }
                }
            }
        }
        if let Some(device) = device_resource {
            let device_idx = usize::from(metric.context().host().is_some_and(|host| !host.is_empty()));
            resources.insert(device_idx, ("device", device));
        }
    }
    builder.set_resources(&resources);

    // Origin metadata
    if let Some(origin) = metric.metadata().origin() {
        match origin {
            MetricOrigin::SourceType(source_type) => {
                builder.set_source_type(source_type.as_ref());
            }
            MetricOrigin::OriginMetadata {
                product,
                subproduct,
                product_detail,
            } => {
                builder.set_origin(*product, *subproduct, *product_detail, false);
            }
        }
    }

    if metric_type != V3MetricType::Sketch {
        if let Some(unit) = metric.metadata().unit() {
            builder.set_unit(unit);
        }
    }

    // Points based on metric type
    match metric.values() {
        MetricValues::Counter(points) | MetricValues::Gauge(points) => {
            for (ts, val) in points {
                if !emittable_scalar_point(val) {
                    continue;
                }
                let timestamp = ts.map(|t| t.get() as i64).unwrap_or(0);
                builder.add_point(timestamp, val);
            }
        }
        MetricValues::Rate(points, interval) => {
            builder.set_interval(interval.as_secs());
            for (ts, val) in points {
                // Scale by interval as done in V2
                let scaled = val / interval.as_secs_f64();
                if !emittable_scalar_point(scaled) {
                    continue;
                }
                let timestamp = ts.map(|t| t.get() as i64).unwrap_or(0);
                builder.add_point(timestamp, scaled);
            }
        }
        MetricValues::Set(points) => {
            // Set values are already converted to count in the iterator
            for (ts, count) in points {
                if !emittable_scalar_point(count) {
                    continue;
                }
                let timestamp = ts.map(|t| t.get() as i64).unwrap_or(0);
                builder.add_point(timestamp, count);
            }
        }
        MetricValues::Distribution(sketches) => {
            for (ts, sketch) in sketches {
                let timestamp = ts.map(|t| t.get() as i64).unwrap_or(0);
                if sketch_has_emittable_values(sketch) {
                    let bin_keys: Vec<i32> = sketch.bins().iter().map(|b| b.key()).collect();
                    let bin_counts: Vec<u32> = sketch.bins().iter().map(|b| b.count()).collect();
                    builder.add_sketch(
                        timestamp,
                        sketch.count() as i64,
                        sketch.stored_sum(),
                        sketch.stored_min(),
                        sketch.stored_max(),
                        &bin_keys,
                        &bin_counts,
                    );
                }
            }
        }
        MetricValues::Histogram(histograms) => {
            for (ts, histogram) in histograms {
                let timestamp = ts.map(|t| t.get() as i64).unwrap_or(0);
                // Convert histogram to DDSketch
                let mut sketch = DDSketch::default();
                for sample in histogram.samples() {
                    sketch.insert_n(sample.value.into_inner(), sample.weight.0 as u64);
                }
                if !sketch.is_empty() {
                    let bin_keys: Vec<i32> = sketch.bins().iter().map(|b| b.key()).collect();
                    let bin_counts: Vec<u32> = sketch.bins().iter().map(|b| b.count()).collect();
                    builder.add_sketch(
                        timestamp,
                        sketch.count() as i64,
                        sketch.sum().unwrap_or(0.0),
                        sketch.min().unwrap_or(0.0),
                        sketch.max().unwrap_or(0.0),
                        &bin_keys,
                        &bin_counts,
                    );
                }
            }
        }
    }

    builder.close();
}

#[inline]
fn emittable_scalar_point(point: f64) -> bool {
    point.is_finite()
}

fn is_v3_series_device_tag(tag: &Tag) -> bool {
    tag.name() == "device" && tag.value().is_some()
}

fn is_v3_series_resource_tag(tag: &Tag) -> bool {
    tag.name() == "dd.internal.resource" && tag.value().is_some()
}

/// Creates a V3 HTTP request from encoded payload data.
async fn create_v3_request(
    endpoint_uri: &str, encoded: V3EncodedMetrics, compression_scheme: CompressionScheme,
) -> Result<V3EncodedRequest, GenericError> {
    // Keep the wire payload as one continuous compressed stream. Per-column compressed sizes are measured
    // independently for telemetry.
    let mut header_buf = [0; 16];
    let header_len = {
        let mut header_writer = CodedOutputStream::bytes(&mut header_buf);
        header_writer.write_tag(3, WireType::LengthDelimited)?;
        header_writer.write_uint64_no_tag(encoded.payload.len() as u64)?;
        header_writer.flush()?;
        header_writer.total_bytes_written() as usize
    };

    let uncompressed_len = header_len + encoded.payload.len();

    let buffer = ChunkedBytesBuffer::new(RB_BUFFER_CHUNK_SIZE);
    let mut compressor = Compressor::from_scheme(compression_scheme, buffer);
    compressor
        .write_all(&header_buf[..header_len])
        .await
        .error_context("Failed to compress V3 payload.")?;
    compressor
        .write_all(&encoded.payload)
        .await
        .error_context("Failed to compress V3 payload.")?;
    compressor
        .flush()
        .await
        .error_context("Failed to flush V3 compressor.")?;
    compressor
        .shutdown()
        .await
        .error_context("Failed to shutdown V3 compressor.")?;

    let compressed_buf = compressor.into_inner().freeze();
    let compressed_len = compressed_buf.len();

    let mut builder = Request::builder()
        .method(Method::POST)
        .uri(endpoint_uri)
        .header(http::header::CONTENT_TYPE, "application/x-protobuf");

    if let Some(encoding) = content_encoding_for_scheme(compression_scheme) {
        builder = builder.header(http::header::CONTENT_ENCODING, encoding);
    }

    let request = builder
        .body(compressed_buf)
        .map_err(|e| generic_error!("Failed to build V3 request: {}", e))?;

    Ok(V3EncodedRequest {
        request,
        compressed_len,
        uncompressed_len,
        stats: encoded.stats,
    })
}

fn content_encoding_for_scheme(compression_scheme: CompressionScheme) -> Option<HeaderValue> {
    match compression_scheme {
        CompressionScheme::Noop => None,
        CompressionScheme::Gzip(_) => Some(HeaderValue::from_static("gzip")),
        CompressionScheme::Zlib(_) => Some(HeaderValue::from_static("deflate")),
        CompressionScheme::Zstd(_) => Some(HeaderValue::from_static("zstd")),
    }
}

#[cfg(test)]
mod tests {
    use std::{collections::HashMap, io::Cursor};

    use bytes::Bytes;
    use datadog_protos::metrics::v3::MetricData as V3MetricData;
    use protobuf::Message as _;
    use saluki_config::ConfigurationLoader;
    use saluki_context::{
        tags::{Tag, TagSet},
        Context,
    };
    use saluki_core::data_model::{
        event::{metric::MetricMetadata, Event},
        payload::Payload,
    };
    use saluki_metrics::test::TestRecorder;
    use stringtheory::MetaString;
    use tokio::time::timeout;

    use super::*;

    #[test]
    fn deser_agent_v3_api_nested_settings() {
        let raw = r#"
serializer_experimental_use_v3_api:
  compression_level: 7
  series:
    endpoints:
      - https://app.datadoghq.com
    validate: true
    use_beta: true
    beta_route: /api/intake/metrics/custom/series
    shadow_sample_rate: 0.25
    shadow_sites:
      - datadoghq.eu
  sketches:
    endpoints:
      - https://app.datadoghq.eu
"#;

        let config =
            serde_yaml::from_str::<DatadogMetricsConfiguration>(raw).expect("configuration should deserialize");

        assert_eq!(7, config.v3_api.compression_level);
        assert_eq!(
            Some("https://app.datadoghq.com"),
            config.v3_api.series.endpoints.first().map(String::as_str)
        );
        assert!(config.v3_api.series.validate);
        assert!(config.v3_api.series.use_beta);
        assert_eq!("/api/intake/metrics/custom/series", config.v3_api.series.beta_route);
        assert_eq!(0.25, config.v3_api.series.shadow_sample_rate);
        assert_eq!(vec!["datadoghq.eu"], config.v3_api.series.shadow_sites);
        assert_eq!(
            Some("https://app.datadoghq.eu"),
            config.v3_api.sketches.endpoints.first().map(String::as_str)
        );
    }

    #[tokio::test]
    async fn typed_metrics_routing_is_authoritative() {
        let (raw, _) = ConfigurationLoader::for_tests(
            Some(serde_json::json!({
                "serializer_compressor_kind": "zlib",
                "use_v2_api_series": true,
                "use_v3_api_series_enabled": "true",
                "observability_pipelines_worker_metrics_enabled": false,
            })),
            None,
            false,
        )
        .await;

        let mut endpoints = Endpoints::default();
        endpoints.compression.compressor_kind = "zstd".to_string();
        endpoints.opw_intake.enabled = true;
        endpoints.opw_intake.url = "https://opw.example.com".to_string();
        endpoints.opw_intake.use_v3_series = true;
        let mut metrics = MetricsEncoding {
            use_v2_series_api: false,
            ..Default::default()
        };
        metrics.v3_api.compression_level = 7;
        metrics.v3_api.series.validate = true;
        metrics.v3_series_mode.mode = "false".to_string();
        metrics.v3_series_mode.endpoint_modes =
            HashMap::from([("https://app.datadoghq.com".to_string(), "true".to_string())]);

        let config = DatadogMetricsConfiguration::from_configuration_with_metrics_routing(&raw, &metrics, &endpoints)
            .expect("configuration should deserialize");

        assert_eq!(config.compressor_kind, "zstd");
        assert!(!config.use_v2_api_series);
        assert_eq!(config.v3_api.compression_level, 7);
        assert!(config.v3_api.series.validate);
        assert_eq!(config.use_v3_api_series.enabled, "false");
        assert_eq!(
            config.use_v3_api_series.endpoints.get("https://app.datadoghq.com"),
            Some(&"true".to_string())
        );
        assert!(config.observability_pipelines_worker_metrics_enabled);
        assert_eq!(
            config.observability_pipelines_worker_metrics_url,
            "https://opw.example.com"
        );
        assert!(config.observability_pipelines_worker_metrics_use_v3_api_series);
    }

    #[test]
    fn agent_v3_api_shadow_defaults_match_agent() {
        let config = serde_yaml::from_str::<DatadogMetricsConfiguration>("").expect("configuration should deserialize");

        assert_eq!(0.0, config.v3_api.series.shadow_sample_rate);
        assert_eq!(vec!["datadoghq.com"], config.v3_api.series.shadow_sites);
    }

    #[test]
    fn shadow_sample_matches_agent_threshold_behavior() {
        assert!(!shadow_sample_matches(0.0, 0.0));
        assert!(shadow_sample_matches(0.5, 0.4));
        assert!(!shadow_sample_matches(0.5, 0.5));
        assert!(!shadow_sample_matches(0.5, 0.6));
    }

    #[test]
    fn shadow_sampling_is_disabled_for_v1_series_baseline_or_v3_incompatible_compressor() {
        assert!(series_shadow_config_for_endpoint(MetricsEndpoint::SeriesV2, 1.0, false).is_enabled());
        assert!(!series_shadow_config_for_endpoint(MetricsEndpoint::SeriesV1, 1.0, false).is_enabled());
        assert!(!series_shadow_config_for_endpoint(MetricsEndpoint::SeriesV2, 1.0, true).is_enabled());
    }

    #[test]
    fn metrics_v3_disabled_by_compressor_uses_v2_only() {
        assert_eq!(
            MetricsEncoderMode::V2Only,
            metrics_encoder_mode_for_config(true, false, true)
        );
        assert_eq!(
            MetricsEncoderMode::V2Only,
            metrics_encoder_mode_for_config(true, true, true)
        );
        assert_eq!(
            MetricsEncoderMode::V3Enabled,
            metrics_encoder_mode_for_config(true, false, false)
        );
        assert_eq!(
            MetricsEncoderMode::Validation,
            metrics_encoder_mode_for_config(true, true, false)
        );
    }

    fn v3_series_config(raw: &str) -> DatadogMetricsConfiguration {
        serde_yaml::from_str(raw).expect("configuration should deserialize")
    }

    #[test]
    fn mixed_v2_and_v3_endpoints_require_both_series_encoders() {
        let config = v3_series_config(
            r#"
use_v3_api_series_enabled: "datadog_only"
additional_endpoints:
  https://custom.example.com:
    - additional-api-key
"#,
        );

        assert!(config.requires_v2_series(false).expect("endpoints should resolve"));
        assert!(config.requires_v3_series(false).expect("endpoints should resolve"));
    }

    #[test]
    fn all_v2_endpoints_do_not_require_v3_series() {
        let config = v3_series_config(
            r#"
dd_url: http://127.0.0.1:9091
use_v3_api_series_enabled: "datadog_only"
"#,
        );

        assert!(config.requires_v2_series(false).expect("endpoints should resolve"));
        assert!(!config.requires_v3_series(false).expect("endpoints should resolve"));
    }

    #[test]
    fn validation_requires_v2_series() {
        let config = v3_series_config(
            r#"
use_v3_api_series_enabled: "true"
serializer_experimental_use_v3_api:
  series:
    validate: true
"#,
        );

        assert!(config.requires_v2_series(false).expect("endpoints should resolve"));
        assert!(config.requires_v3_series(false).expect("endpoints should resolve"));
    }

    #[test]
    fn v1_series_configuration_takes_precedence_over_v3() {
        let config = v3_series_config(
            r#"
use_v2_api_series: false
use_v3_api_series_enabled: "true"
"#,
        );

        let series_v3_can_be_enabled = series_v3_can_be_enabled_for_config(
            config.use_v2_api_series,
            false,
            None,
            false,
            &config.use_v3_api_series,
        );

        assert!(!series_v3_can_be_enabled);

        assert_eq!(
            MetricsEncoderMode::V2Only,
            metrics_encoder_mode_for_config(series_v3_can_be_enabled, false, false)
        );

        assert!(
            config.requires_v2_series(false).expect("endpoint should resolve"),
            "V1 configuration must retain the legacy series builder even when V3 is enabled"
        );
    }

    #[test]
    fn all_v3_serializer_endpoints_require_only_v3_series() {
        let config = v3_series_config(
            r#"
dd_url: https://agent.datad0g.com.
use_v3_api_series_enabled: "false"
additional_endpoints:
  https://agent.datadoghq.com.:
    - additional-api-key
serializer_experimental_use_v3_api:
  series:
    endpoints:
      - https://agent.datad0g.com.
      - https://agent.datadoghq.com.
"#,
        );

        assert!(!config.requires_v2_series(false).expect("endpoints should resolve"));
        assert!(config.requires_v3_series(false).expect("endpoints should resolve"));
    }

    #[test]
    fn endpoint_override_uses_the_overridden_endpoint_protocol() {
        let config = v3_series_config(
            r#"
dd_url: https://primary.example.com
use_v3_api_series_enabled: "false"
serializer_experimental_use_v3_api:
  series:
    endpoints:
      - https://primary.example.com
      - https://v3-mrf.example.com
"#,
        );

        let v2_mrf_config = config
            .clone()
            .with_metrics_endpoint_override("https://v2-mrf.example.com".to_string());
        let v3_mrf_config = config.with_metrics_endpoint_override("https://v3-mrf.example.com".to_string());

        assert!(v2_mrf_config
            .requires_v2_series(false)
            .expect("V2 MRF endpoint should resolve"));
        assert!(!v2_mrf_config
            .requires_v3_series(false)
            .expect("V2 MRF endpoint should resolve"));
        assert!(!v3_mrf_config
            .requires_v2_series(false)
            .expect("V3 MRF endpoint should resolve"));
        assert!(v3_mrf_config
            .requires_v3_series(false)
            .expect("V3 MRF endpoint should resolve"));
    }

    #[test]
    fn v2_series_only_override_keeps_v2_and_disables_shadowing() {
        let config = v3_series_config(
            r#"
use_v3_api_series_enabled: "true"
serializer_experimental_use_v3_api:
  series:
    shadow_sample_rate: 1.0
"#,
        )
        .with_v2_series_only();

        assert_eq!("false", config.use_v3_api_series.enabled);
        assert!(config.use_v3_api_series.endpoints.is_empty());
        assert!(config.v3_api.series.endpoints.is_empty());
        assert_eq!(0.0, config.v3_api.series.shadow_sample_rate);
        assert!(config.requires_v2_series(false).expect("endpoint should resolve"));
        assert!(!config.requires_v3_series(false).expect("endpoint should resolve"));
    }

    #[test]
    fn agent_default_v3_does_not_enable_opw_only_encoder_mode() {
        let series_config = UseV3ApiSeriesConfig::default();
        let invalid_metrics_primary_override =
            selected_metrics_primary_v3_override(true, "http://[::1", false, true, "http://vector.example.com", false);

        assert!(!series_v3_can_be_enabled_for_config(
            true,
            false,
            Some(false),
            false,
            &series_config
        ));
        assert!(series_v3_can_be_enabled_for_config(
            true,
            false,
            Some(true),
            false,
            &series_config
        ));
        assert!(series_v3_can_be_enabled_for_config(
            true,
            false,
            Some(false),
            true,
            &series_config
        ));
        assert!(series_v3_can_be_enabled_for_config(
            true,
            true,
            Some(false),
            false,
            &series_config
        ));
        assert_eq!(None, invalid_metrics_primary_override);
        assert!(series_v3_can_be_enabled_for_config(
            true,
            false,
            invalid_metrics_primary_override,
            false,
            &series_config
        ));
    }

    #[test]
    fn v3_encodes_zero_count_distribution_with_bins() {
        let mut sketch = DDSketch::default();
        sketch.insert_n(42.0, 10);
        sketch.set_count(0);
        sketch.set_min(40.0);
        sketch.set_max(45.0);
        sketch.set_sum(123.0);
        sketch.set_avg(f64::NAN);

        let metric = Metric::from_parts(
            Context::from_static_parts("zero.count.distribution", &[]),
            MetricValues::distribution((123_u64, sketch)),
            MetricMetadata::default(),
        );

        let encoded = encode_v3_metrics_batch(&[metric], &SharedTagSet::default())
            .expect("zero-count distribution should encode to V3");
        let metric_data = V3MetricData::parse_from_bytes(&encoded.payload).expect("V3 metric data should decode");

        assert_eq!(metric_data.numPoints, vec![1]);
    }

    #[test]
    fn v3_skips_empty_distribution() {
        let metric = Metric::from_parts(
            Context::from_static_parts("empty.distribution", &[]),
            MetricValues::distribution((123_u64, DDSketch::default())),
            MetricMetadata::default(),
        );

        let encoded =
            encode_v3_metrics_batch(&[metric], &SharedTagSet::default()).expect("empty distribution should encode");
        let metric_data = V3MetricData::parse_from_bytes(&encoded.payload).expect("V3 metric data should decode");

        assert!(metric_data.numPoints.is_empty());
    }

    #[tokio::test]
    async fn create_v3_request_uses_configured_endpoint_uri() {
        let encoded = V3Writer::new().finalize().expect("empty V3 payload should encode");
        let request = create_v3_request("/api/intake/metrics/custom/series", encoded, CompressionScheme::noop())
            .await
            .expect("request should be created");

        assert_eq!("/api/intake/metrics/custom/series", request.request.uri());
    }

    #[tokio::test]
    async fn create_v3_request_uses_single_stream_body_and_column_telemetry() {
        let metrics = vec![Metric::counter("v3.single.stream", 42.0)];
        let encoded = encode_v3_metrics_batch(&metrics, &SharedTagSet::default()).expect("metrics should encode to V3");
        let expected_payload = encoded.payload.clone();

        let mut request = create_v3_request(V3_SERIES_ENDPOINT_URI, encoded, CompressionScheme::noop())
            .await
            .expect("request should be created");
        measure_v3_column_compressed_sizes(&mut request.stats, CompressionScheme::noop())
            .await
            .expect("column compressed sizes should be measured");

        for column in &request.stats.columns {
            assert_eq!(column.compressed_len, column.bytes.len());
        }

        let mut expected_body = Vec::new();
        {
            let mut os = CodedOutputStream::vec(&mut expected_body);
            os.write_tag(3, WireType::LengthDelimited).unwrap();
            os.write_uint64_no_tag(expected_payload.len() as u64).unwrap();
            os.flush().unwrap();
        }
        expected_body.extend_from_slice(&expected_payload);

        assert_eq!(request.request.into_body().into_bytes(), Bytes::from(expected_body));
    }

    #[tokio::test]
    async fn create_v3_request_zstd_body_decodes_to_metric_payload() {
        let metrics = vec![Metric::counter("v3.single.stream.zstd", 42.0)];
        let encoded = encode_v3_metrics_batch(&metrics, &SharedTagSet::default()).expect("metrics should encode to V3");
        let expected_payload = encoded.payload.clone();

        let mut request = create_v3_request(V3_SERIES_ENDPOINT_URI, encoded, CompressionScheme::zstd_default())
            .await
            .expect("request should be created");
        measure_v3_column_compressed_sizes(&mut request.stats, CompressionScheme::zstd_default())
            .await
            .expect("column compressed sizes should be measured");

        for column in &request.stats.columns {
            let expected_compressed_len = compressed_v3_len(&column.bytes, CompressionScheme::zstd_default())
                .await
                .expect("column compressed size should be measured");
            assert_eq!(column.compressed_len, expected_compressed_len);
        }

        let mut expected_body = Vec::new();
        {
            let mut os = CodedOutputStream::vec(&mut expected_body);
            os.write_tag(3, WireType::LengthDelimited).unwrap();
            os.write_uint64_no_tag(expected_payload.len() as u64).unwrap();
            os.flush().unwrap();
        }
        expected_body.extend_from_slice(&expected_payload);

        let decoded_body = zstd::stream::decode_all(Cursor::new(request.request.into_body().into_bytes()))
            .expect("compressed V3 body should decode");
        assert_eq!(decoded_body, expected_body);
    }

    #[test]
    fn v3_drops_non_finite_scalar_points() {
        const NUM_POINTS_FIELD_NUMBER: u32 = 15;
        const VALUE_SINT64_FIELD_NUMBER: u32 = 17;

        let metrics = vec![
            Metric::from_parts(
                Context::from_static_parts("v3.finite.counter", &[]),
                MetricValues::counter([(1, 1.0_f64), (2, f64::NAN), (3, f64::INFINITY), (4, 2.0)]),
                MetricMetadata::default(),
            ),
            Metric::from_parts(
                Context::from_static_parts("v3.finite.gauge", &[]),
                MetricValues::gauge([(1, 3.0_f64), (2, f64::NAN), (3, 4.0)]),
                MetricMetadata::default(),
            ),
            Metric::from_parts(
                Context::from_static_parts("v3.finite.rate", &[]),
                MetricValues::rate(
                    [(1, 30.0_f64), (2, f64::INFINITY), (3, f64::NAN)],
                    Duration::from_secs(10),
                ),
                MetricMetadata::default(),
            ),
            Metric::set("v3.finite.set", "alpha"),
        ];

        let encoded = encode_v3_metrics_batch(&metrics, &SharedTagSet::default()).expect("metrics should encode to V3");
        let column = |field_number| {
            encoded
                .stats
                .columns
                .iter()
                .find(|column| column.field_number == field_number)
                .map(|column| column.bytes.as_slice())
        };

        assert_eq!(column(NUM_POINTS_FIELD_NUMBER), Some(&[2, 2, 1, 1][..]));
        // Values are 1, 2, 3, 4, 3 (rate scaled by 10s), and 1 (set cardinality), encoded as sint64.
        assert_eq!(column(VALUE_SINT64_FIELD_NUMBER), Some(&[2, 4, 6, 8, 6, 2][..]));
        assert_eq!(encoded.stats.value_encoding_stats.sint64, 6);
        assert_eq!(encoded.stats.value_encoding_stats.float64, 0);
    }

    #[tokio::test]
    async fn v3_serializer_stats_record_agent_style_value_and_column_counters() {
        const VALUE_SINT64_FIELD_NUMBER: u32 = 17;

        let recorder = TestRecorder::default();
        let _local = metrics::set_default_local_recorder(&recorder);
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());
        let metrics = vec![
            Metric::gauge("v3.telemetry.zero", [(123, 0.0), (124, 0.0)]),
            Metric::counter("v3.telemetry.sint64", [(123, 100.0), (124, 200.0)]),
            Metric::gauge("v3.telemetry.float32", [(123, 1.5), (124, 2.25)]),
            Metric::gauge("v3.telemetry.float64", [(123, (1i64 << 30) as f64), (124, 1.5)]),
        ];
        let encoded = encode_v3_metrics_batch(&metrics, &SharedTagSet::default()).expect("metrics should encode to V3");
        let mut request = create_v3_request(V3_SERIES_ENDPOINT_URI, encoded, CompressionScheme::noop())
            .await
            .expect("request should be created");
        measure_v3_column_compressed_sizes(&mut request.stats, CompressionScheme::noop())
            .await
            .expect("column compressed sizes should be measured");

        let value_sint64_column = request
            .stats
            .columns
            .iter()
            .find(|column| column.field_number == VALUE_SINT64_FIELD_NUMBER)
            .expect("sint64 value column should be present");
        let value_sint64_len = value_sint64_column.bytes.len() as u64;
        assert_eq!(value_sint64_column.compressed_len as u64, value_sint64_len);

        record_v3_serializer_stats(&serializer_telemetry, &request.stats);

        assert_eq!(
            recorder.counter(("serializer.v3_values_count", &[("type", "zero")])),
            Some(2)
        );
        assert_eq!(
            recorder.counter(("serializer.v3_values_count", &[("type", "sint64")])),
            Some(2)
        );
        assert_eq!(
            recorder.counter(("serializer.v3_values_count", &[("type", "float32")])),
            Some(2)
        );
        assert_eq!(
            recorder.counter(("serializer.v3_values_count", &[("type", "float64")])),
            Some(2)
        );
        assert_eq!(
            recorder.counter((
                "serializer.v3_column_size",
                &[("column", "ValueSint64"), ("compressed", "uncompressed")]
            )),
            Some(value_sint64_len)
        );
        assert_eq!(
            recorder.counter((
                "serializer.v3_column_size",
                &[("column", "ValueSint64"), ("compressed", "compressed")]
            )),
            Some(value_sint64_len)
        );
    }

    async fn create_v3_test_request(metrics: &[Metric]) -> V3EncodedRequest {
        let encoded = encode_v3_metrics_batch(metrics, &SharedTagSet::default()).expect("metrics should encode to V3");
        create_v3_request(V3_SERIES_ENDPOINT_URI, encoded, CompressionScheme::noop())
            .await
            .expect("request should be created")
    }

    fn test_v3_flush_context<'a>(
        ep_config: &'a EndpointConfiguration, payload_limits: V3PayloadLimits,
        serializer_telemetry: &'a V3SerializerTelemetry, telemetry: &'a ComponentTelemetry,
    ) -> V3FlushContext<'a> {
        V3FlushContext {
            endpoint_config: ep_config,
            payload_limits,
            series_endpoint_uri: V3_SERIES_ENDPOINT_URI,
            serializer_telemetry,
            telemetry,
        }
    }

    /// Collected form of a payload emitted by `encode_and_flush_v3_payload_requests`.
    struct CollectedV3Payload {
        event_count: usize,
        data_point_count: usize,
        request: Request<FrozenChunkedBytesBuffer>,
    }

    /// Drives `encode_and_flush_v3_payload_requests` and collects the payloads it emits.
    ///
    /// The channel is sized generously because nothing drains it concurrently here; production drains it from the
    /// encoder's main task.
    async fn collect_v3_payload_requests(metrics: &[Metric], context: V3FlushContext<'_>) -> Vec<CollectedV3Payload> {
        let (mut payloads_tx, mut payloads_rx) = tokio::sync::mpsc::channel(64);
        let mut ratio = V3CompressionRatio::default();
        encode_and_flush_v3_payload_requests(
            V3_SERIES_ENDPOINT_URI,
            metrics,
            context,
            "series",
            &mut ratio,
            &mut payloads_tx,
            None,
            None,
        )
        .await
        .expect("payload requests should encode and flush");
        drop(payloads_tx);

        let mut collected = Vec::new();
        while let Some(payload) = payloads_rx.recv().await {
            let Payload::Http(http_payload) = payload else {
                panic!("expected HTTP payload");
            };
            let (metadata, request) = http_payload.into_parts();
            collected.push(CollectedV3Payload {
                event_count: metadata.event_count(),
                data_point_count: metadata.data_point_count(),
                request,
            });
        }

        collected
    }

    #[tokio::test]
    async fn v3_payload_requests_split_by_compressed_size_limit() {
        let metrics = vec![
            Metric::counter("v3.compressed.split.one", 1.0),
            Metric::counter("v3.compressed.split.two", 2.0),
        ];
        let single_request = create_v3_test_request(&metrics[..1]).await;
        let combined_request = create_v3_test_request(&metrics).await;
        assert!(combined_request.compressed_len > single_request.compressed_len);

        let limits = V3PayloadLimits::new(single_request.compressed_len, usize::MAX, 10_000, 10_000);
        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());

        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);
        let requests = collect_v3_payload_requests(&metrics, context).await;

        assert_eq!(2, requests.len());
        assert_eq!(
            vec![1, 1],
            requests.iter().map(|request| request.event_count).collect::<Vec<_>>()
        );
        assert!(requests
            .iter()
            .all(|request| request.request.body().len() <= limits.max_compressed_size));
    }

    #[tokio::test]
    async fn v3_byte_fallback_does_not_resurrect_metric_over_point_limit() {
        let point_oversized_metric = Metric::counter("v3.fallback.over.point.limit", [(123, 1.0), (124, 2.0)]);

        let byte_oversized_metric = Metric::counter(
            concat!(
                "v3.fallback.over.byte.limit.",
                "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
                "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
                "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
                "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd",
            ),
            3.0,
        );

        let point_oversized_request = create_v3_test_request(std::slice::from_ref(&point_oversized_metric)).await;
        let byte_oversized_request = create_v3_test_request(std::slice::from_ref(&byte_oversized_metric)).await;

        assert!(byte_oversized_request.compressed_len > point_oversized_request.compressed_len);

        let limits = V3PayloadLimits::new(point_oversized_request.compressed_len, usize::MAX, 10_000, 1);

        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());
        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);

        let requests = collect_v3_payload_requests(&[point_oversized_metric, byte_oversized_metric], context).await;

        let emitted_point_counts = requests
            .iter()
            .map(|request| request.data_point_count)
            .collect::<Vec<_>>();

        assert!(
            emitted_point_counts
                .iter()
                .all(|point_count| *point_count <= limits.max_points_per_payload),
            "fallback emitted request point counts {emitted_point_counts:?} above limit {}",
            limits.max_points_per_payload
        );
    }

    #[tokio::test]
    async fn v3_serializer_stats_record_payload_full_split_reason() {
        let metrics = vec![
            Metric::counter("v3.telemetry.payload_full.one", 1.0),
            Metric::counter("v3.telemetry.payload_full.two", 2.0),
        ];
        let single_request = create_v3_test_request(&metrics[..1]).await;
        let limits = V3PayloadLimits::new(single_request.compressed_len, usize::MAX, 10_000, 10_000);
        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
        let recorder = TestRecorder::default();
        let _local = metrics::set_default_local_recorder(&recorder);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());

        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);
        let requests = collect_v3_payload_requests(&metrics, context).await;

        assert_eq!(2, requests.len());
        assert_eq!(
            recorder.counter(("serializer.v3_payload_split_reason", &[("reason", "payload_full")])),
            Some(1)
        );
    }

    #[tokio::test]
    async fn v3_serializer_stats_record_item_too_big_split_reason() {
        let metrics = vec![Metric::counter("v3.telemetry.item_too_big", 1.0)];
        let request = create_v3_test_request(&metrics).await;
        let limits = V3PayloadLimits::new(request.compressed_len - 1, usize::MAX, 10_000, 10_000);
        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
        let recorder = TestRecorder::default();
        let _local = metrics::set_default_local_recorder(&recorder);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());

        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);
        let requests = collect_v3_payload_requests(&metrics, context).await;

        assert!(requests.is_empty());
        assert_eq!(recorder.counter("serializer.v3_item_too_big"), Some(1));
        assert_eq!(
            recorder.counter(("serializer.v3_payload_split_reason", &[("reason", "item_too_big")])),
            Some(1)
        );
    }

    #[tokio::test]
    async fn v3_payload_requests_split_by_uncompressed_size_limit() {
        let metrics = vec![
            Metric::counter("v3.uncompressed.split.one", 1.0),
            Metric::counter("v3.uncompressed.split.two", 2.0),
        ];
        let single_request = create_v3_test_request(&metrics[..1]).await;
        let combined_request = create_v3_test_request(&metrics).await;
        assert!(combined_request.uncompressed_len > single_request.uncompressed_len);

        let limits = V3PayloadLimits::new(usize::MAX, single_request.uncompressed_len, 10_000, 10_000);
        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());

        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);
        let requests = collect_v3_payload_requests(&metrics, context).await;

        assert_eq!(2, requests.len());
        assert_eq!(
            vec![1, 1],
            requests.iter().map(|request| request.event_count).collect::<Vec<_>>()
        );
    }

    /// Drives `build_v3_batch` to exhaustion, returning the metric range and event count of each batch it cut.
    fn collect_v3_batches(
        metrics: &[Metric], context: V3FlushContext<'_>, target_uncompressed_len: usize,
    ) -> Vec<(Range<usize>, usize)> {
        let mut batches = Vec::new();
        let mut idx = 0;
        while idx < metrics.len() {
            let start = idx;
            match build_v3_batch(metrics, &mut idx, context, "series", target_uncompressed_len) {
                Some(batch) => batches.push((start..idx, batch.event_count)),
                None if idx == start => break,
                None => {}
            }
        }

        batches
    }

    #[test]
    fn v3_batches_are_cut_at_the_point_limit() {
        let metrics = vec![
            Metric::counter("v3.points.split.one", [(123, 1.0), (124, 2.0)]),
            Metric::counter("v3.points.split.two", [(123, 3.0), (124, 4.0)]),
            Metric::counter("v3.points.split.three", 5.0),
        ];
        let limits = V3PayloadLimits::new(usize::MAX, usize::MAX, 10_000, 3);
        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());
        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);

        let batches = collect_v3_batches(&metrics, context, usize::MAX);

        assert_eq!(vec![(0..1, 1), (1..3, 2)], batches);
    }

    #[test]
    fn v3_batches_are_cut_at_the_metric_limit() {
        let metrics = vec![
            Metric::counter("v3.metric.limit.a", 1.0),
            Metric::counter("v3.metric.limit.b", 2.0),
            Metric::counter("v3.metric.limit.c", 3.0),
            Metric::counter("v3.metric.limit.d", 4.0),
            Metric::counter("v3.metric.limit.e", 5.0),
        ];
        let limits = V3PayloadLimits::new(usize::MAX, usize::MAX, 2, 10_000);
        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 2, usize::MAX, None);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());
        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);

        let batches = collect_v3_batches(&metrics, context, usize::MAX);

        assert_eq!(vec![(0..2, 2), (2..4, 2), (4..5, 1)], batches);
    }

    #[test]
    fn v3_batches_are_cut_at_the_uncompressed_byte_target() {
        let metrics = vec![
            Metric::counter("v3.byte.target.a", 1.0),
            Metric::counter("v3.byte.target.b", 2.0),
            Metric::counter("v3.byte.target.c", 3.0),
            Metric::counter("v3.byte.target.d", 4.0),
        ];
        let limits = V3PayloadLimits::new(usize::MAX, usize::MAX, 10_000, 10_000);
        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());
        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);

        // A target of one byte forces a cut after every metric, since the estimate is non-zero once anything is
        // written. This is the mechanism that keeps high-cardinality batches from being encoded and then re-split.
        let batches = collect_v3_batches(&metrics, context, 1);

        assert_eq!(4, batches.len());
        assert!(batches.iter().all(|(_, event_count)| *event_count == 1));
    }

    #[test]
    fn v3_serializer_stats_record_max_points_split_reason() {
        let metrics = vec![
            Metric::counter("v3.telemetry.max_points.one", [(123, 1.0), (124, 2.0)]),
            Metric::counter("v3.telemetry.max_points.two", [(123, 3.0), (124, 4.0)]),
        ];
        let limits = V3PayloadLimits::new(usize::MAX, usize::MAX, 10_000, 2);
        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
        let recorder = TestRecorder::default();
        let _local = metrics::set_default_local_recorder(&recorder);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());
        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);

        let batches = collect_v3_batches(&metrics, context, usize::MAX);

        assert_eq!(vec![(0..1, 1), (1..2, 1)], batches);
        assert_eq!(
            recorder.counter(("serializer.v3_payload_split_reason", &[("reason", "max_points")])),
            Some(1)
        );
    }

    #[test]
    fn v3_batches_drop_oversized_metric_and_keep_batching() {
        let metrics = vec![
            Metric::counter("v3.points.oversized.before", [(123, 1.0), (124, 2.0)]),
            Metric::counter(
                "v3.points.oversized.too_big",
                [(123, 3.0), (124, 4.0), (125, 5.0), (126, 6.0)],
            ),
            Metric::counter("v3.points.oversized.after", 7.0),
        ];
        let limits = V3PayloadLimits::new(usize::MAX, usize::MAX, 10_000, 3);
        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
        let recorder = TestRecorder::default();
        let _local = metrics::set_default_local_recorder(&recorder);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());
        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);

        let batches = collect_v3_batches(&metrics, context, usize::MAX);

        // Dropping the oversized metric no longer forces a payload boundary: the metrics either side of it share a
        // batch, since together they still fit the point limit.
        assert_eq!(vec![(0..3, 2)], batches);
        assert_eq!(recorder.counter("serializer.v3_item_too_big"), Some(1));
    }

    #[test]
    fn v3_batches_skip_zero_point_metrics_without_splitting() {
        let metrics = vec![
            Metric::counter("v3.points.zero.before", 1.0),
            Metric::counter("v3.points.zero.empty", &[] as &[f64]),
            Metric::counter("v3.points.zero.after", 2.0),
        ];
        let limits = V3PayloadLimits::new(usize::MAX, usize::MAX, 10_000, 10_000);
        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());
        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);

        let batches = collect_v3_batches(&metrics, context, usize::MAX);

        assert_eq!(vec![(0..3, 2)], batches);
    }

    #[test]
    fn v3_compression_ratio_targets_the_compressed_limit() {
        let limits = V3PayloadLimits::new(500_000, 5_000_000, 10_000, 10_000);
        let mut ratio = V3CompressionRatio::default();

        // With nothing observed yet, assume no compression: the target tracks the compressed limit directly so the
        // first batch cannot wildly overshoot.
        let initial_target = ratio.uncompressed_target(limits);
        assert_eq!((500_000.0 * V3_BATCH_TARGET_MARGIN) as usize, initial_target);

        // After observing 4:1 compression, the target grows towards the uncompressed budget those bytes imply.
        ratio.record(4_000_000, 1_000_000);
        let observed_target = ratio.uncompressed_target(limits);
        assert!(
            observed_target > initial_target,
            "target should grow once compression is observed: {observed_target} vs {initial_target}"
        );

        // It is still capped by the uncompressed limit, however well the data compresses.
        ratio.record(4_000_000, 40_000);
        ratio.record(4_000_000, 40_000);
        ratio.record(4_000_000, 40_000);
        ratio.record(4_000_000, 40_000);
        ratio.record(4_000_000, 40_000);
        assert!(ratio.uncompressed_target(limits) <= limits.max_uncompressed_size);
    }

    #[tokio::test]
    async fn v3_split_flush_uses_payload_request_batch_headers() {
        let mut metrics = vec![
            Metric::counter("v3.headers.split.one", 1.0),
            Metric::counter("v3.headers.split.two", 2.0),
        ];
        let single_request = create_v3_test_request(&metrics[..1]).await;
        let combined_request = create_v3_test_request(&metrics).await;
        assert!(combined_request.compressed_len > single_request.compressed_len);

        let limits = V3PayloadLimits::new(single_request.compressed_len, usize::MAX, 10_000, 10_000);
        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());
        let batch_id = Uuid::now_v7();
        let (mut payloads_tx, mut payloads_rx) = tokio::sync::mpsc::channel(8);

        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);
        let mut ratio = V3CompressionRatio::default();
        encode_and_flush_v3_series_metrics(
            context,
            &mut metrics,
            &mut ratio,
            &mut payloads_tx,
            Some(&batch_id),
            Some(MetricsPayloadInfo::v3_series()),
        )
        .await
        .expect("V3 metrics should flush");

        for expected_seq in 0..2 {
            let payload = payloads_rx.recv().await.expect("payload should be emitted");
            let Payload::Http(http_payload) = payload else {
                panic!("expected HTTP payload");
            };
            let (_, request) = http_payload.into_parts();
            assert_eq!(
                batch_id.as_hyphenated().to_string(),
                request
                    .headers()
                    .get("X-Metrics-Request-ID")
                    .expect("batch ID header should be present")
                    .to_str()
                    .expect("batch ID header should be valid")
            );
            assert_eq!(
                expected_seq.to_string(),
                request
                    .headers()
                    .get("X-Metrics-Request-Seq")
                    .expect("batch sequence header should be present")
                    .to_str()
                    .expect("batch sequence header should be valid")
            );
            assert_eq!(
                "2",
                request
                    .headers()
                    .get("X-Metrics-Request-Len")
                    .expect("batch length header should be present")
                    .to_str()
                    .expect("batch length header should be valid")
            );
        }

        assert!(metrics.is_empty());
    }

    #[tokio::test]
    async fn v3_sketch_flush_uses_split_payload_requests() {
        let mut metrics = vec![
            Metric::distribution("v3.sketch.split.one", [1.0, 2.0, 3.0]),
            Metric::distribution("v3.sketch.split.two", [4.0, 5.0, 6.0]),
        ];
        let single_request = create_v3_test_request(&metrics[..1]).await;
        let combined_request = create_v3_test_request(&metrics).await;
        assert!(combined_request.compressed_len > single_request.compressed_len);

        let limits = V3PayloadLimits::new(single_request.compressed_len, usize::MAX, 10_000, 10_000);
        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());
        let batch_id = Uuid::now_v7();
        let (mut payloads_tx, mut payloads_rx) = tokio::sync::mpsc::channel(8);

        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);
        let mut ratio = V3CompressionRatio::default();
        encode_and_flush_v3_sketch_metrics(
            context,
            &mut metrics,
            &mut ratio,
            &mut payloads_tx,
            Some(&batch_id),
            Some(MetricsPayloadInfo::v3_sketches()),
        )
        .await
        .expect("V3 sketches should flush");

        for expected_seq in 0..2 {
            let payload = payloads_rx.recv().await.expect("payload should be emitted");
            let Payload::Http(http_payload) = payload else {
                panic!("expected HTTP payload");
            };
            let (_, request) = http_payload.into_parts();
            assert_eq!(V3_SKETCHES_ENDPOINT_URI, request.uri());
            assert_eq!(
                expected_seq.to_string(),
                request
                    .headers()
                    .get("X-Metrics-Request-Seq")
                    .expect("batch sequence header should be present")
                    .to_str()
                    .expect("batch sequence header should be valid")
            );
            assert_eq!(
                "2",
                request
                    .headers()
                    .get("X-Metrics-Request-Len")
                    .expect("batch length header should be present")
                    .to_str()
                    .expect("batch length header should be valid")
            );
        }

        assert!(metrics.is_empty());
    }

    #[test]
    fn v3_series_metric_unit_refs_are_encoded_sparsely() {
        let context = Context::from_static_parts("my.timer.avg", &[]);
        let metadata = MetricMetadata::default().with_unit(MetaString::from_static("millisecond"));
        let gauge = Metric::from_parts(context, MetricValues::gauge([1.0_f64]), metadata);
        let context = Context::from_static_parts("my.counter", &[]);
        let no_unit = Metric::from_parts(context, MetricValues::gauge([2.0_f64]), MetricMetadata::default());
        let context = Context::from_static_parts("my.timer.max", &[]);
        let metadata = MetricMetadata::default().with_unit(MetaString::from_static("millisecond"));
        let same_unit = Metric::from_parts(context, MetricValues::gauge([3.0_f64]), metadata);

        let payload = encode_v3_metrics_batch(&[gauge, no_unit, same_unit], &SharedTagSet::default())
            .expect("V3 metric should encode successfully")
            .payload;

        let expected_unit_dict = [
            0xca, 0x01, // field 25, length-delimited.
            0x0c, // field payload length: varint string length + string bytes.
            0x0b, b'm', b'i', b'l', b'l', b'i', b's', b'e', b'c', b'o', b'n', b'd',
        ];
        assert!(
            payload
                .windows(expected_unit_dict.len())
                .any(|window| window == expected_unit_dict),
            "V3 payload should contain DictUnitStr field for 'millisecond', got bytes: {:?}",
            payload
        );

        let expected_unit_ref = [
            0xd2, 0x01, // field 26, length-delimited.
            0x02, // packed field payload length.
            0x02, 0x00, // sparse unit refs for metrics 1 and 3 only: refs [1, 1] -> deltas [1, 0].
        ];
        assert!(
            payload
                .windows(expected_unit_ref.len())
                .any(|window| window == expected_unit_ref),
            "V3 payload should contain UnitRef field for 'millisecond', got bytes: {:?}",
            payload
        );
    }

    #[test]
    fn v3_sketch_metric_unit_not_encoded() {
        let context = Context::from_static_parts("my.histogram", &[]);
        let metadata = MetricMetadata::default().with_unit(MetaString::from_static("millisecond"));
        let histogram = Metric::from_parts(context, MetricValues::histogram([1.0_f64]), metadata);

        let payload = encode_v3_metrics_batch(&[histogram], &SharedTagSet::default())
            .expect("V3 sketch metric should encode successfully")
            .payload;

        assert!(
            !payload
                .windows(b"millisecond".len())
                .any(|window| window == b"millisecond"),
            "V3 sketch payload should not contain unit bytes, matching the Agent V3 sketch builder: {:?}",
            payload
        );
    }

    #[test]
    fn v3_series_promotes_device_and_internal_resource_tags_to_resources() {
        let context = Context::from_static_parts(
            "series.resources",
            &[
                "env:prod",
                "device:switch1",
                "dd.internal.resource:pod:pod-a",
                "dd.internal.resource:malformed",
            ],
        );
        let context = context.with_host(Some(MetaString::from_static("host-a")));
        let metadata = MetricMetadata::default();
        let metric = Metric::from_parts(context, MetricValues::gauge([1.0_f64]), metadata);

        let payload = encode_v3_metrics_batch(&[metric], &SharedTagSet::default())
            .expect("V3 series should encode successfully")
            .payload;

        assert_contains_bytes(&payload, b"env:prod");
        assert!(!contains_bytes(&payload, b"device:switch1"));
        assert!(!contains_bytes(&payload, b"dd.internal.resource:pod:pod-a"));
        assert!(!contains_bytes(&payload, b"dd.internal.resource:malformed"));

        let expected_resource_dict = [
            0x22, // field 4, length-delimited.
            0x25, // field payload length.
            0x04, b'h', b'o', b's', b't', 0x06, b'h', b'o', b's', b't', b'-', b'a', 0x06, b'd', b'e', b'v', b'i', b'c',
            b'e', 0x07, b's', b'w', b'i', b't', b'c', b'h', b'1', 0x03, b'p', b'o', b'd', 0x05, b'p', b'o', b'd', b'-',
            b'a',
        ];
        assert_contains_bytes(&payload, &expected_resource_dict);
    }

    #[test]
    fn v3_series_promotes_additional_and_origin_resource_tags_without_empty_host() {
        let context = Context::from_static_parts("series.additional_origin_resources", &["env:prod"])
            .with_origin_tags(tag_set(["dd.internal.resource:pod:pod-origin"]));
        let additional_tags = SharedTagSet::from(tag_set([
            "team:core",
            "device:switch1",
            "dd.internal.resource:container:container-a",
        ]));
        let context = context.with_host(Some(MetaString::empty()));
        let metadata = MetricMetadata::default();
        let metric = Metric::from_parts(context, MetricValues::gauge([1.0_f64]), metadata);

        let payload = encode_v3_metrics_batch(&[metric], &additional_tags)
            .expect("V3 series should encode successfully")
            .payload;

        assert_contains_bytes(&payload, b"env:prod");
        assert_contains_bytes(&payload, b"team:core");
        assert!(!contains_bytes(&payload, b"device:switch1"));
        assert!(!contains_bytes(&payload, b"dd.internal.resource:container:container-a"));
        assert!(!contains_bytes(&payload, b"dd.internal.resource:pod:pod-origin"));

        let expected_resource_dict = [
            0x22, // field 4, length-delimited.
            0x34, // field payload length.
            0x06, b'd', b'e', b'v', b'i', b'c', b'e', 0x07, b's', b'w', b'i', b't', b'c', b'h', b'1', 0x03, b'p', b'o',
            b'd', 0x0a, b'p', b'o', b'd', b'-', b'o', b'r', b'i', b'g', b'i', b'n', 0x09, b'c', b'o', b'n', b't', b'a',
            b'i', b'n', b'e', b'r', 0x0b, b'c', b'o', b'n', b't', b'a', b'i', b'n', b'e', b'r', b'-', b'a',
        ];
        assert_contains_bytes(&payload, &expected_resource_dict);
        assert!(!contains_bytes(&payload, b"host"));
    }

    #[test]
    fn v3_sketch_keeps_device_and_internal_resource_tags_as_tags() {
        let context = Context::from_static_parts(
            "sketch.resources",
            &["env:prod", "device:switch1", "dd.internal.resource:pod:pod-a"],
        );
        let context = context.with_host(Some(MetaString::from_static("host-a")));
        let metadata = MetricMetadata::default();
        let metric = Metric::from_parts(context, MetricValues::histogram([1.0_f64]), metadata);

        let payload = encode_v3_metrics_batch(&[metric], &SharedTagSet::default())
            .expect("V3 sketch should encode successfully")
            .payload;

        assert_contains_bytes(&payload, b"env:prod");
        assert_contains_bytes(&payload, b"device:switch1");
        assert_contains_bytes(&payload, b"dd.internal.resource:pod:pod-a");

        let expected_resource_dict = [
            0x22, // field 4, length-delimited.
            0x0c, // field payload length.
            0x04, b'h', b'o', b's', b't', 0x06, b'h', b'o', b's', b't', b'-', b'a',
        ];
        assert_contains_bytes(&payload, &expected_resource_dict);
    }

    /// Distinct inputs for a [`run_request_builder`] integration test.
    struct RequestBuilderScenario {
        v2_series_max_metrics: Option<usize>,
        series_mode: MetricsEncoderMode,
        sketches_mode: MetricsEncoderMode,
        payload_limits: V3PayloadLimits,
        shadow_series_endpoint_uri: String,
        series_shadow_config: SeriesShadowConfig,
        flush_timeout: Duration,
    }

    impl RequestBuilderScenario {
        /// Creates a scenario with no V2 series builder, unbounded V3 size/point limits, shadow sampling disabled,
        /// and a 10 ms flush timeout.
        fn new(series_mode: MetricsEncoderMode, sketches_mode: MetricsEncoderMode) -> Self {
            Self {
                v2_series_max_metrics: None,
                series_mode,
                sketches_mode,
                payload_limits: V3PayloadLimits::new(usize::MAX, usize::MAX, 10_000, 10_000),
                shadow_series_endpoint_uri: "/api/intake/metrics/v3beta/series".to_string(),
                series_shadow_config: SeriesShadowConfig::new(0.0),
                flush_timeout: Duration::from_millis(10),
            }
        }

        /// Attaches a V2 series request builder that flushes after `max_metrics_per_payload` metrics.
        fn with_v2_series_builder(mut self, max_metrics_per_payload: usize) -> Self {
            self.v2_series_max_metrics = Some(max_metrics_per_payload);
            self
        }

        /// Sets the V3 per-payload point limit that drives point-count split flushes.
        fn with_max_points_per_payload(mut self, max_points_per_payload: usize) -> Self {
            self.payload_limits = V3PayloadLimits::new(usize::MAX, usize::MAX, 10_000, max_points_per_payload);
            self
        }

        /// Enables V3 series shadow sampling at `sample_rate`, routed to `shadow_series_endpoint_uri`.
        fn with_series_shadow(mut self, sample_rate: f64, shadow_series_endpoint_uri: &str) -> Self {
            self.series_shadow_config = SeriesShadowConfig::new(sample_rate);
            self.shadow_series_endpoint_uri = shadow_series_endpoint_uri.to_string();
            self
        }

        /// Overrides the flush timeout used to bound pending flushes.
        fn with_flush_timeout(mut self, flush_timeout: Duration) -> Self {
            self.flush_timeout = flush_timeout;
            self
        }

        /// Spawns [`run_request_builder`] for this scenario, returning a harness for pushing metrics and draining
        /// flushed payloads.
        async fn spawn(self) -> RequestBuilderHarness {
            let v2_series_builder = match self.v2_series_max_metrics {
                Some(max_metrics_per_payload) => {
                    let v2_endpoint_config = EndpointConfiguration::new(
                        CompressionScheme::noop(),
                        max_metrics_per_payload,
                        usize::MAX,
                        None,
                    );
                    Some(
                        v2::create_v2_request_builder(MetricsEndpoint::SeriesV2, &v2_endpoint_config)
                            .await
                            .expect("V2 request builder should be created"),
                    )
                }
                None => None,
            };
            let v3_endpoint_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
            let metrics_builder = MetricsBuilder::default();
            let telemetry = ComponentTelemetry::from_builder(&metrics_builder);
            let serializer_telemetry = V3SerializerTelemetry::from_builder(&metrics_builder);
            let (events_tx, events_rx) = tokio::sync::mpsc::channel(1);
            let (payloads_tx, payloads_rx) = tokio::sync::mpsc::channel(8);

            let handle = tokio::spawn(run_request_builder(
                v2_series_builder,
                None,
                self.series_mode,
                self.sketches_mode,
                V3RuntimeConfig {
                    endpoint_config: v3_endpoint_config,
                    payload_limits: self.payload_limits,
                    series_endpoint_uri: V3_SERIES_ENDPOINT_URI.to_string(),
                    shadow_series_endpoint_uri: self.shadow_series_endpoint_uri,
                    series_shadow_config: self.series_shadow_config,
                    serializer_telemetry,
                },
                telemetry,
                events_rx,
                payloads_tx,
                self.flush_timeout,
                false,
            ));

            RequestBuilderHarness {
                events_tx,
                payloads_rx,
                handle,
            }
        }
    }

    /// A spawned [`run_request_builder`] task plus its input/output channels.
    struct RequestBuilderHarness {
        events_tx: tokio::sync::mpsc::Sender<EventsBuffer>,
        payloads_rx: tokio::sync::mpsc::Receiver<PayloadsBuffer>,
        handle: tokio::task::JoinHandle<Result<(), GenericError>>,
    }

    impl RequestBuilderHarness {
        /// Sends a single event buffer carrying `metrics` to the request builder.
        async fn push_metrics(&self, metrics: impl IntoIterator<Item = Metric>) {
            let mut events = EventsBuffer::default();
            for metric in metrics {
                assert!(
                    events.try_push(Event::Metric(metric)).is_none(),
                    "event buffer should accept metric"
                );
            }
            self.events_tx
                .send(events)
                .await
                .expect("events should be sent to request builder");
        }

        /// Receives the next flushed payload, unwrapping it into its HTTP metadata and request.
        async fn next_http_request(&mut self) -> (PayloadMetadata, Request<FrozenChunkedBytesBuffer>) {
            let payload = timeout(Duration::from_secs(1), self.payloads_rx.recv())
                .await
                .expect("payload should arrive before timeout")
                .expect("payload channel should remain open");
            match payload {
                Payload::Http(http_payload) => http_payload.into_parts(),
                _ => panic!("expected HTTP payload"),
            }
        }

        /// Asserts that no payload is flushed within `window`.
        async fn assert_no_payload_within(&mut self, window: Duration) {
            assert!(
                timeout(window, self.payloads_rx.recv()).await.is_err(),
                "no payload should be flushed within {window:?}"
            );
        }

        /// Drops the events sender and waits for the request builder task to stop cleanly.
        async fn shutdown(self) {
            drop(self.events_tx);
            self.handle
                .await
                .expect("request builder task should complete")
                .expect("request builder should stop cleanly");
        }
    }

    #[tokio::test]
    async fn validation_split_flush_assigns_batch_id_to_carried_metric() {
        // Validation mode keeps each V2 flush aligned with a V3 flush under the same batch ID. With the V2 builder
        // flushing after one metric, two input metrics produce two V2/V3 pairs, and the metric carried into the
        // second batch gets a fresh validation ID.
        let mut harness = RequestBuilderScenario::new(MetricsEncoderMode::Validation, MetricsEncoderMode::V2Only)
            .with_v2_series_builder(1)
            .spawn()
            .await;

        harness
            .push_metrics([
                Metric::counter("validation.split.one", 1.0),
                Metric::counter("validation.split.two", 2.0),
            ])
            .await;

        let mut flushed_requests = Vec::new();
        for _ in 0..4 {
            let (_, request) = harness.next_http_request().await;
            let batch_id = request
                .headers()
                .get("X-Metrics-Request-ID")
                .expect("validation batch ID header should be present")
                .to_str()
                .expect("validation batch ID should be valid header text")
                .to_string();
            flushed_requests.push((request.uri().to_string(), batch_id));
        }

        assert_eq!("/api/v2/series", flushed_requests[0].0);
        assert_eq!(V3_SERIES_ENDPOINT_URI, flushed_requests[1].0);
        assert_eq!("/api/v2/series", flushed_requests[2].0);
        assert_eq!(V3_SERIES_ENDPOINT_URI, flushed_requests[3].0);

        assert_eq!(flushed_requests[0].1, flushed_requests[1].1);
        assert_eq!(flushed_requests[2].1, flushed_requests[3].1);
        assert_ne!(flushed_requests[0].1, flushed_requests[2].1);

        harness.shutdown().await;
    }

    #[tokio::test]
    async fn authoritative_v3_flushes_previous_point_limit_batch() {
        // Authoritative V3 batches independently of V2. With a 3-point limit, the first two-point metric fits but
        // the second would exceed it, so the first flushes as a point-limit split and the second flushes on the
        // pending-flush timeout.
        let recorder = TestRecorder::default();
        let _local = metrics::set_default_local_recorder(&recorder);

        let mut harness = RequestBuilderScenario::new(MetricsEncoderMode::V3Enabled, MetricsEncoderMode::V2Only)
            .with_max_points_per_payload(3)
            .with_flush_timeout(Duration::from_millis(250))
            .spawn()
            .await;

        harness
            .push_metrics([
                Metric::counter("authoritative.v3.points.one", [(123, 1.0), (124, 2.0)]),
                Metric::counter("authoritative.v3.points.two", [(123, 3.0), (124, 4.0)]),
            ])
            .await;

        // Point-count split flushes the first metric before the second exceeds the limit.
        let (_, request) = harness.next_http_request().await;
        assert_eq!(V3_SERIES_ENDPOINT_URI, request.uri());
        harness.assert_no_payload_within(Duration::from_millis(50)).await;

        // The carried-over metric flushes when the pending-flush timeout fires.
        let (_, request) = harness.next_http_request().await;
        assert_eq!(V3_SERIES_ENDPOINT_URI, request.uri());
        assert_eq!(
            recorder.counter(("serializer.v3_payload_split_reason", &[("reason", "max_points")])),
            Some(1)
        );

        harness.shutdown().await;
    }

    #[tokio::test]
    async fn authoritative_v3_sketches_flush_previous_point_limit_batch() {
        // The same authoritative point-limit split applies to sketches, which always target the V3 sketches
        // endpoint: the first distribution flushes as a point-limit split, the second on the pending-flush timeout.
        let recorder = TestRecorder::default();
        let _local = metrics::set_default_local_recorder(&recorder);

        let mut harness = RequestBuilderScenario::new(MetricsEncoderMode::V2Only, MetricsEncoderMode::V3Enabled)
            .with_max_points_per_payload(3)
            .with_flush_timeout(Duration::from_millis(250))
            .spawn()
            .await;

        harness
            .push_metrics([
                Metric::distribution("authoritative.v3.sketch.points.one", [(123, 1.0), (124, 2.0)]),
                Metric::distribution("authoritative.v3.sketch.points.two", [(123, 3.0), (124, 4.0)]),
            ])
            .await;

        for stage in ["point-limit split", "timeout"] {
            let (_, request) = harness.next_http_request().await;
            assert_eq!(
                V3_SKETCHES_ENDPOINT_URI,
                request.uri(),
                "{stage} sketches payload should target the V3 sketches endpoint"
            );
        }
        assert_eq!(
            recorder.counter(("serializer.v3_payload_split_reason", &[("reason", "max_points")])),
            Some(1)
        );

        harness.shutdown().await;
    }

    #[tokio::test]
    async fn authoritative_v3_does_not_flush_on_v2_boundary() {
        // The V2 builder flushes at its one-metric boundary, but authoritative V3 batches independently and must
        // not flush just because V2 did. The remaining V2 and V3 batches only flush together on the timeout.
        let mut harness = RequestBuilderScenario::new(MetricsEncoderMode::V3Enabled, MetricsEncoderMode::V2Only)
            .with_v2_series_builder(1)
            .with_flush_timeout(Duration::from_millis(250))
            .spawn()
            .await;

        harness
            .push_metrics([
                Metric::counter("authoritative.v3.decouple.one", 1.0),
                Metric::counter("authoritative.v3.decouple.two", 2.0),
            ])
            .await;

        let (_, request) = harness.next_http_request().await;
        assert_eq!("/api/v2/series", request.uri());
        harness.assert_no_payload_within(Duration::from_millis(50)).await;

        let mut timeout_flush_uris = Vec::new();
        for _ in 0..2 {
            let (_, request) = harness.next_http_request().await;
            timeout_flush_uris.push(request.uri().to_string());
        }
        assert_eq!(2, timeout_flush_uris.len());
        assert!(timeout_flush_uris.iter().any(|uri| uri == "/api/v2/series"));
        assert!(timeout_flush_uris.iter().any(|uri| uri == V3_SERIES_ENDPOINT_URI));

        harness.shutdown().await;
    }

    #[tokio::test]
    async fn shadow_sampled_series_flush_sends_v2_and_v3_beta_with_same_batch_id() {
        // With shadow sampling forced on (rate 1.0), a V2-only series flush also emits a sampled V3 beta shadow
        // payload to the configured shadow route, tagged as shadow and sharing the V2 batch ID.
        let mut harness = RequestBuilderScenario::new(MetricsEncoderMode::V2Only, MetricsEncoderMode::V2Only)
            .with_v2_series_builder(10_000)
            .with_series_shadow(1.0, "/api/intake/metrics/v3beta/custom")
            .spawn()
            .await;

        harness.push_metrics([Metric::counter("shadow.sampled", 1.0)]).await;

        let mut flushed_requests = Vec::new();
        for _ in 0..2 {
            let (metadata, request) = harness.next_http_request().await;
            let payload_info = *metadata
                .get::<MetricsPayloadInfo>()
                .expect("metrics payload info should be present");
            let batch_id = request
                .headers()
                .get("X-Metrics-Request-ID")
                .expect("shadow batch ID header should be present")
                .to_str()
                .expect("shadow batch ID should be valid header text")
                .to_string();
            flushed_requests.push((request.uri().to_string(), payload_info, batch_id));
        }

        assert_eq!("/api/v2/series", flushed_requests[0].0);
        assert_eq!(MetricsPayloadInfo::v2_shadow_series(), flushed_requests[0].1);
        assert_eq!("/api/intake/metrics/v3beta/custom", flushed_requests[1].0);
        assert_eq!(MetricsPayloadInfo::v3_shadow_series(), flushed_requests[1].1);
        assert_eq!(flushed_requests[0].2, flushed_requests[1].2);

        harness.shutdown().await;
    }

    #[tokio::test]
    async fn shadow_sample_rate_zero_sends_only_v2_without_validation_headers() {
        // With shadow sampling disabled (rate 0.0), a V2-only series flush emits only the V2 payload, with no
        // shadow/validation batch ID header and no second payload.
        let mut harness = RequestBuilderScenario::new(MetricsEncoderMode::V2Only, MetricsEncoderMode::V2Only)
            .with_v2_series_builder(10_000)
            .spawn()
            .await;

        harness.push_metrics([Metric::counter("shadow.disabled", 1.0)]).await;

        let (_, request) = harness.next_http_request().await;
        assert_eq!("/api/v2/series", request.uri());
        assert!(!request.headers().contains_key("X-Metrics-Request-ID"));
        harness.assert_no_payload_within(Duration::from_millis(50)).await;

        harness.shutdown().await;
    }

    fn contains_bytes(haystack: &[u8], needle: &[u8]) -> bool {
        haystack.windows(needle.len()).any(|window| window == needle)
    }

    fn assert_contains_bytes(haystack: &[u8], needle: &[u8]) {
        assert!(
            contains_bytes(haystack, needle),
            "expected payload to contain bytes {:?}, got {:?}",
            needle,
            haystack
        );
    }

    fn tag_set<const N: usize>(tags: [&'static str; N]) -> TagSet {
        tags.into_iter().map(Tag::from_static).collect()
    }

    // Regression test to ensure the V2 series request builder enforces `max_series_points_per_payload`.
    //
    // The test encodes more total points than the configured limit and asserts the builder splits them across
    // multiple payloads without any payload exceeding the limit.
    #[tokio::test]
    async fn v2_series_builder_enforces_max_series_points_per_payload() {
        let v2_endpoint_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, 10_000, None);
        let mut builder = v2::create_v2_request_builder(MetricsEndpoint::SeriesV2, &v2_endpoint_config)
            .await
            .expect("V2 request builder should be created");
        builder
            .with_len_limits(usize::MAX, usize::MAX)
            .expect("byte limits should be accepted");

        let mut total_points = 0;
        for i in 0..6_000u32 {
            let metric = Metric::gauge(
                Context::from_parts(MetaString::from(format!("g{i}")), TagSet::default()),
                [(1u64, 1.0), (2u64, 2.0)],
            );
            let mut pending = Some(metric);
            while let Some(metric) = pending.take() {
                if let Some(returned) = builder.encode(metric).await.expect("encode should not error") {
                    for request in builder.flush().await {
                        let (_, points, _) = request.expect("request should build");
                        assert!(
                            points <= 10_000,
                            "payload carried {} points, over the 10000 limit",
                            points
                        );
                        total_points += points;
                    }
                    pending = Some(returned);
                }
            }
        }
        for request in builder.flush().await {
            let (_, points, _) = request.expect("request should build");
            assert!(
                points <= 10_000,
                "final payload carried {} points, over the 10000 limit",
                points
            );
            total_points += points;
        }

        assert_eq!(
            total_points, 12_000,
            "all points should be emitted across the split payloads"
        );
    }
}

#[cfg(test)]
mod config_smoke {
    use datadog_agent_config_testing::config_registry::structs;
    use datadog_agent_config_testing::run_config_smoke_tests;
    use serde_json::json;

    use super::DatadogMetricsConfiguration;
    use crate::config::{DatadogRemapper, KEY_ALIASES};

    #[tokio::test]
    async fn smoke_test() {
        run_config_smoke_tests(
            structs::DATADOG_METRICS_CONFIGURATION,
            &[
                "serializer_experimental_use_v3_api.sketches.beta_route",
                "serializer_experimental_use_v3_api.sketches.shadow_sample_rate",
                "serializer_experimental_use_v3_api.sketches.shadow_sites",
                "serializer_experimental_use_v3_api.sketches.use_beta",
            ],
            json!({}),
            |cfg| {
                cfg.as_typed::<DatadogMetricsConfiguration>()
                    .expect("DatadogMetricsConfiguration should deserialize")
            },
            KEY_ALIASES,
            DatadogRemapper::from_env_vars,
        )
        .await
    }
}

#[cfg(test)]
mod use_v2_api_series_default {
    use saluki_config::ConfigurationLoader;
    use serde_json::json;

    use super::{v2, DatadogMetricsConfiguration};
    use crate::{common::datadog::clamp_payload_limits, config::KEY_ALIASES};

    /// `use_v2_api_series` defaults to `true`, preserving V2 protobuf behavior when the flag is absent.
    #[tokio::test]
    async fn defaults_to_true_when_absent() {
        let cfg = ConfigurationLoader::default()
            .with_key_aliases(KEY_ALIASES)
            .add_providers([figment::providers::Serialized::defaults(json!({}))])
            .into_generic()
            .await
            .expect("config should load");
        let parsed: DatadogMetricsConfiguration = cfg.as_typed().expect("should deserialize");
        assert!(parsed.use_v2_api_series);
    }

    #[tokio::test]
    async fn deserializes_payload_limit_keys() {
        let cfg = ConfigurationLoader::default()
            .with_key_aliases(KEY_ALIASES)
            .add_providers([figment::providers::Serialized::defaults(json!({
                "serializer_max_payload_size": 4321,
                "serializer_max_uncompressed_payload_size": 8765,
                "serializer_max_series_payload_size": 1234,
                "serializer_max_series_uncompressed_payload_size": 5678,
            }))])
            .into_generic()
            .await
            .expect("config should load");
        let parsed: DatadogMetricsConfiguration = cfg.as_typed().expect("should deserialize");

        assert_eq!(parsed.max_payload_size, 4321);
        assert_eq!(parsed.max_uncompressed_payload_size, 8765);
        assert_eq!(parsed.max_series_payload_size, 1234);
        assert_eq!(parsed.max_series_uncompressed_payload_size, 5678);
    }

    #[tokio::test]
    async fn deserializes_max_series_points_per_payload() {
        // Default should be 10,000.
        let cfg = ConfigurationLoader::default()
            .with_key_aliases(KEY_ALIASES)
            .add_providers([figment::providers::Serialized::defaults(json!({}))])
            .into_generic()
            .await
            .expect("config should load");
        let parsed: DatadogMetricsConfiguration = cfg.as_typed().expect("should deserialize");
        assert_eq!(parsed.max_series_points_per_payload, 10_000);
        assert_eq!(parsed.v3_payload_limits().max_points_per_payload, 10_000);

        // Explicit value should round-trip.
        let cfg = ConfigurationLoader::default()
            .with_key_aliases(KEY_ALIASES)
            .add_providers([figment::providers::Serialized::defaults(json!({
                "serializer_max_series_points_per_payload": 500,
            }))])
            .into_generic()
            .await
            .expect("config should load");
        let parsed: DatadogMetricsConfiguration = cfg.as_typed().expect("should deserialize");
        assert_eq!(parsed.max_series_points_per_payload, 500);
        assert_eq!(parsed.v3_payload_limits().max_points_per_payload, 500);
    }

    #[test]
    fn clamps_series_payload_limit_keys_to_api_limits() {
        let (uncompressed_limit, compressed_limit) = clamp_payload_limits(
            v2::SERIES_V2_UNCOMPRESSED_SIZE_LIMIT + 1,
            v2::SERIES_V2_COMPRESSED_SIZE_LIMIT + 1,
            v2::SERIES_V2_UNCOMPRESSED_SIZE_LIMIT,
            v2::SERIES_V2_COMPRESSED_SIZE_LIMIT,
        );
        assert_eq!(uncompressed_limit, v2::SERIES_V2_UNCOMPRESSED_SIZE_LIMIT);
        assert_eq!(compressed_limit, v2::SERIES_V2_COMPRESSED_SIZE_LIMIT);

        let (uncompressed_limit, compressed_limit) = clamp_payload_limits(
            5678,
            1234,
            v2::SERIES_V2_UNCOMPRESSED_SIZE_LIMIT,
            v2::SERIES_V2_COMPRESSED_SIZE_LIMIT,
        );
        assert_eq!(uncompressed_limit, 5678);
        assert_eq!(compressed_limit, 1234);
    }
}