praxis-proxy-filter 0.7.2

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2024 Praxis Contributors

//! Filter pipeline engine: the runtime representation of a listener's
//! filter processing.
//!
//! ## Module Layout
//!
//! | Module | Responsibility |
//! |---|---|
//! | [`build`] | `FilterPipeline` construction from config entries |
//! | [`build_branch`] | Recursive branch chain resolution |
//! | [`http`] | Request, response, and body execution loops |
//! | [`tcp`] | Connect/disconnect execution |
//! | [`evaluate`] | Branch condition checking and dispatch |
//! | [`branch`] | Runtime branch types ([`ResolvedBranch`], [`BranchOutcome`]) |
//! | [`filter`] | [`PipelineFilter`], the per-filter wrapper |
//! | [`body`] | Body chunk processing utilities |
//! | [`checks`] | Ordering validation (router before LB, etc.) |
//! | [`clusters`] | Cluster reference collection |
//! | [`extension`] | [`PipelineExtension`] trait for injecting per-request resources |
//!
//! At runtime, chains do not exist. All listener chains are
//! concatenated into a flat `Vec<PipelineFilter>`. Branch chains are
//! stored as nested `Vec<PipelineFilter>` inside each filter's
//! [`branches`] field.
//!
//! [`ResolvedBranch`]: branch::ResolvedBranch
//! [`BranchOutcome`]: branch::BranchOutcome
//! [`PipelineFilter`]: filter::PipelineFilter
//! [`branches`]: filter::PipelineFilter::branches

pub(crate) mod body;
pub(crate) mod branch;
mod build;
pub(crate) mod build_branch;
#[cfg(feature = "upstream-binding")]
pub(crate) mod catalog;
mod checks;
mod clusters;
pub(crate) mod evaluate;
mod extension;
pub(crate) mod filter;
mod http;
mod http_utils;
/// Public pipeline introspection snapshots for admin surfaces.
pub(crate) mod introspection;
/// Sub-request execution for iterative request routing.
pub(crate) mod subrequest;
mod tcp;
#[cfg(test)]
mod test_filters;

#[cfg(test)]
#[expect(clippy::allow_attributes, reason = "blanket test suppressions")]
#[allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::indexing_slicing,
    clippy::panic,
    clippy::field_reassign_with_default,
    clippy::type_complexity,
    clippy::too_many_lines,
    clippy::redundant_closure_for_method_calls,
    clippy::significant_drop_tightening,
    clippy::doc_markdown,
    reason = "tests"
)]
mod tests;

use std::sync::Arc;

pub use extension::PipelineExtension;
use praxis_core::{
    config::{ABSOLUTE_MAX_BODY_BYTES, FailureMode, InsecureOptions},
    health::HealthRegistry,
    id::IdGenerator,
    kv::KvStoreRegistry,
    time::TimeSource,
};
use tracing::{error, warn};

use self::filter::PipelineFilter;
use crate::{
    FilterError,
    any_filter::AnyFilter,
    body::{BodyCapabilities, BodyMode},
    builtins::http::payload_processing::compression_config::CompressionConfig,
    extensions::RequestExtensions,
};

// -----------------------------------------------------------------------------
// FilterPipeline
// -----------------------------------------------------------------------------

/// An ordered list of filters executed on every request.
///
/// ```
/// use praxis_filter::{FilterPipeline, FilterRegistry};
///
/// let registry = FilterRegistry::with_builtins();
/// let pipeline = FilterPipeline::build(&mut [], &registry).unwrap();
/// assert!(pipeline.is_empty());
/// ```
#[expect(
    clippy::struct_excessive_bools,
    reason = "independent server-injected runtime toggles, not a state machine"
)]
pub struct FilterPipeline {
    /// Pre-computed body processing capabilities for this pipeline.
    body_capabilities: BodyCapabilities,

    /// Compression configuration, if a compression filter is present.
    compression: Option<CompressionConfig>,

    /// Ordered list of filters with their conditions and branches.
    pub(crate) filters: Vec<PipelineFilter>,

    /// Whether per-filter duration metrics are recorded.
    record_filter_duration_metrics: bool,

    /// Compiled path templates for the `route` metric label.
    route_templates: Arc<praxis_core::config::RouteTemplates>,

    /// Shared health registry for endpoint health lookups.
    health_registry: Option<HealthRegistry>,

    /// Shared ID generator for request correlation IDs.
    id_generator: Arc<IdGenerator>,

    /// Named key-value stores for runtime mappings.
    kv_stores: Option<KvStoreRegistry>,

    /// Per-cluster session stores for sticky session affinity, preserved across reloads.
    session_stores: Option<Arc<crate::SessionStoreRegistry>>,

    /// Shared sub-request client for iterative sub-requests.
    subrequest_client: Option<praxis_core::subrequest::SubRequestClient>,

    /// Whether any filter, including branch filters, may select a streaming
    /// sub-request response.
    may_select_streaming_subrequest_response: bool,

    /// Top-level `trace_context` filters whose conditions gate early propagation.
    trace_context_filter_indices: Vec<usize>,

    /// External pipeline extensions injected after construction.
    pipeline_extensions: Vec<Box<dyn PipelineExtension>>,

    /// Wall-clock time source for filters that need timestamps.
    time_source: Arc<dyn TimeSource>,

    /// Global request body ceiling, enforced by counting in Stream mode.
    request_body_ceiling: Option<usize>,

    /// Global response body ceiling, enforced by counting in Stream mode.
    response_body_ceiling: Option<usize>,

    /// Indices into `filters` of filters declaring request-body access.
    request_body_filter_indices: Vec<usize>,

    /// Indices into `filters` of filters declaring response-body access.
    response_body_filter_indices: Vec<usize>,

    /// Indices into `filters` of filters declaring selected-upstream
    /// request-body access.
    selected_upstream_request_body_filter_indices: Vec<usize>,

    /// Indices into `filters` of top-level filters declaring bound-upstream
    /// request-body access, drained once per request by the barrier in
    /// [`execute_http_request`].
    ///
    /// [`execute_http_request`]: FilterPipeline::execute_http_request
    #[cfg(feature = "bound-upstream-request-body")]
    bound_upstream_request_body_filter_indices: Vec<usize>,

    /// Whether upstream hostnames may resolve to private or reserved IPs.
    ///
    /// Mirrors `insecure_options.allow_private_upstreams`; consumed by the
    /// upstream peer builders when they resolve a hostname.
    allow_private_upstreams: bool,
    /// Indices into `filters` of filters declaring response-trailer access.
    response_trailer_filter_indices: Vec<usize>,
}

#[expect(
    clippy::multiple_inherent_impl,
    reason = "pipeline concerns are split across modules"
)]
impl FilterPipeline {
    /// Apply global body size ceilings.
    ///
    /// When no filter requires body access (mode is [`Stream`]),
    /// uses [`SizeLimit`] to enforce the ceiling without
    /// buffering. When a filter already requested
    /// [`StreamBuffer`], the ceiling tightens the existing limit.
    ///
    /// # Errors
    ///
    /// Returns [`FilterError`] if a [`StreamBuffer`] has no byte limit
    /// and `allow_unbounded` is `false`.
    ///
    /// [`Stream`]: BodyMode::Stream
    /// [`SizeLimit`]: BodyMode::SizeLimit
    /// [`StreamBuffer`]: BodyMode::StreamBuffer
    pub fn apply_body_limits(
        &mut self,
        max_request: Option<usize>,
        max_response: Option<usize>,
        allow_unbounded: bool,
    ) -> Result<(), FilterError> {
        self.apply_nested_body_limits(max_request, max_response, allow_unbounded)?;

        if let Some(ceiling) = max_request {
            self.body_capabilities.request_body_mode = clamp_body_mode(
                self.body_capabilities.request_body_mode,
                ceiling,
                self.body_capabilities.needs_request_body,
            );
            self.body_capabilities.needs_request_body = true;
        }

        if let Some(ceiling) = max_response {
            self.body_capabilities.response_body_mode = clamp_body_mode(
                self.body_capabilities.response_body_mode,
                ceiling,
                self.body_capabilities.needs_response_body,
            );
            self.body_capabilities.needs_response_body = true;
        }

        check_unbounded_stream_buffer(
            "request",
            &mut self.body_capabilities.request_body_mode,
            allow_unbounded,
        )?;
        check_unbounded_stream_buffer(
            "response",
            &mut self.body_capabilities.response_body_mode,
            allow_unbounded,
        )?;

        self.request_body_ceiling = max_request;
        self.response_body_ceiling = max_response;

        Ok(())
    }

    /// Apply the listener ceilings recursively to nested pipelines.
    fn apply_nested_body_limits(
        &mut self,
        max_request: Option<usize>,
        max_response: Option<usize>,
        allow_unbounded: bool,
    ) -> Result<(), FilterError> {
        let mut nested_error = None;
        self.visit_nested_pipelines(&mut |pipeline| {
            if nested_error.is_none()
                && let Err(error) = pipeline.apply_body_limits(max_request, max_response, allow_unbounded)
            {
                nested_error = Some(error);
            }
        });
        if let Some(error) = nested_error {
            return Err(error);
        }
        Ok(())
    }

    /// Global request body ceiling; `None` means unbounded was allowed.
    #[must_use]
    pub fn request_body_ceiling(&self) -> Option<usize> {
        self.request_body_ceiling
    }

    /// Effective request-body byte limit for the selected-upstream phase.
    ///
    /// The adapted output is checked against the merged `StreamBuffer` limit
    /// clamped to the listener ceiling — the same bound that governed the
    /// canonical pre-read — never an individual filter's limit. `min` keeps this
    /// correct whether or not `request_body_mode` is already clamped at build.
    ///
    /// ```
    /// use praxis_core::config::ABSOLUTE_MAX_BODY_BYTES;
    /// use praxis_filter::{FilterPipeline, FilterRegistry};
    ///
    /// let registry = FilterRegistry::with_builtins();
    /// let pipeline = FilterPipeline::build(&mut [], &registry).unwrap();
    /// assert_eq!(
    ///     pipeline.selected_upstream_request_body_limit(),
    ///     ABSOLUTE_MAX_BODY_BYTES,
    ///     "an empty pipeline uses the defensive absolute limit",
    /// );
    /// ```
    #[must_use]
    pub fn selected_upstream_request_body_limit(&self) -> usize {
        let mode_limit = match self.body_capabilities().request_body_mode {
            BodyMode::StreamBuffer { max_bytes } => max_bytes.unwrap_or(ABSOLUTE_MAX_BODY_BYTES),
            // Defensive default: selected-upstream participation always merges the
            // request-body mode to StreamBuffer (see `selected_upstream_body_mode`
            // in crates/filter/src/pipeline/body.rs), so this arm is not reachable
            // for the phase; it keeps the limit well-defined if that ever changes.
            BodyMode::SizeLimit { max_bytes } => max_bytes,
            _ => ABSOLUTE_MAX_BODY_BYTES,
        };
        self.request_body_ceiling()
            .map_or(mode_limit, |ceiling| mode_limit.min(ceiling))
    }

    /// Global response body ceiling; `None` means unbounded was allowed.
    #[must_use]
    pub fn response_body_ceiling(&self) -> Option<usize> {
        self.response_body_ceiling
    }

    /// Pre-computed body processing capabilities for this pipeline.
    pub fn body_capabilities(&self) -> &BodyCapabilities {
        &self.body_capabilities
    }

    /// Reset request-body completion flags before replaying an upstream attempt.
    pub fn clear_request_body_done(&self, body_done_indices: &mut [bool]) {
        for &idx in &self.request_body_filter_indices {
            if let Some(done) = body_done_indices.get_mut(idx) {
                *done = false;
            }
        }
    }

    /// Whether any filter in the pipeline needs body access.
    pub fn needs_body_filters(&self) -> bool {
        self.body_capabilities.needs_request_body || self.body_capabilities.needs_response_body
    }

    /// Number of filters in the pipeline.
    pub fn len(&self) -> usize {
        self.filters.len()
    }

    /// Whether the pipeline has no filters.
    pub fn is_empty(&self) -> bool {
        self.filters.is_empty()
    }

    /// Whether any top-level filter has the given type name.
    ///
    /// ```
    /// use praxis_filter::{FilterPipeline, FilterRegistry};
    ///
    /// let registry = FilterRegistry::with_builtins();
    /// let pipeline = FilterPipeline::build(&mut [], &registry).unwrap();
    /// assert!(!pipeline.contains_filter("access_log"));
    /// ```
    pub fn contains_filter(&self, type_name: &str) -> bool {
        self.filters.iter().any(|pf| pf.filter.name() == type_name)
    }

    /// Whether a top-level `trace_context` filter matches this request.
    pub(crate) fn enables_trace_propagation(&self, request: &crate::Request) -> bool {
        self.trace_context_filter_indices.iter().any(|&idx| {
            self.filters
                .get(idx)
                .is_some_and(|pf| crate::condition::should_execute(&pf.conditions, request))
        })
    }

    /// Whether a top-level `trace_context` filter matches the supplied header view.
    pub(crate) fn enables_trace_propagation_from<S: crate::condition::HeaderSource>(
        &self,
        request: &crate::Request,
        headers: &S,
    ) -> Result<bool, S::Error> {
        for &idx in &self.trace_context_filter_indices {
            let Some(pf) = self.filters.get(idx) else {
                continue;
            };
            if crate::condition::should_execute_from(
                &pf.conditions,
                request,
                headers,
                crate::condition::BoundUpstreamView::default(),
                crate::condition::SelectedUpstream::none(),
            )? {
                return Ok(true);
            }
        }
        Ok(false)
    }

    /// Names of filters in this pipeline whose protocol level is not
    /// supported by `listener_protocol`.
    ///
    /// A TCP listener silently skips HTTP-level filters at runtime, so an
    /// HTTP security filter placed on a TCP listener would never run. This
    /// surfaces that mismatch at build time.
    pub fn filters_unsupported_by(&self, listener_protocol: praxis_core::config::ProtocolKind) -> Vec<&'static str> {
        self.filters
            .iter()
            .filter(|pf| !listener_protocol.supports(pf.filter.protocol_level()))
            .map(|pf| pf.filter.name())
            .collect()
    }

    /// Names of filters in this pipeline that operate below the HTTP layer.
    ///
    /// Unlike [`filters_unsupported_by`], which honors a listener's protocol
    /// stack (an HTTP listener legitimately runs TCP-level filters at the
    /// connection layer), this reports filters that a *filtered sub-request*
    /// cannot run: that executor drives only the HTTP request phase, so any
    /// filter whose protocol level is not [`ProtocolKind::Http`] is never
    /// invoked. Binding one into an outbound chain is a silent runtime no-op;
    /// this surfaces it at build time.
    ///
    /// [`filters_unsupported_by`]: Self::filters_unsupported_by
    /// [`ProtocolKind::Http`]: praxis_core::config::ProtocolKind::Http
    pub fn non_http_filters(&self) -> Vec<&'static str> {
        let mut names = Vec::new();
        for_each_pipeline_filter(&self.filters, &mut |pf| {
            if pf.filter.protocol_level() != praxis_core::config::ProtocolKind::Http {
                names.push(pf.filter.name());
            }
        });
        names
    }

    /// Names of terminal filters in this pipeline, recursively including those
    /// nested inside branch sub-chains.
    ///
    /// A terminal filter short-circuits the request phase with a response and no
    /// upstream. The HTTP filtered sub-request executor forwards to a resolved
    /// upstream and cannot surface such a response, so a terminal filter bound
    /// into an outbound chain (at the top level or buried in a branch) must be
    /// rejected at build time rather than silently activate and drop its response
    /// at runtime.
    ///
    /// Detection is capability-based
    /// ([`HttpFilter::produces_terminal_response`]), not a match against the
    /// hard-coded builtin names in [`TERMINAL_FILTERS`], so a custom filter that
    /// can return a terminal action is caught even though its name is unknown to
    /// the builtin list.
    ///
    /// [`HttpFilter::produces_terminal_response`]: crate::HttpFilter::produces_terminal_response
    /// [`TERMINAL_FILTERS`]: praxis_core::config::TERMINAL_FILTERS
    pub fn terminal_filters(&self) -> Vec<&'static str> {
        let mut names = Vec::new();
        for_each_pipeline_filter(&self.filters, &mut |pf| {
            if let AnyFilter::Http(filter) = &pf.filter
                && filter.produces_terminal_response()
            {
                names.push(filter.name());
            }
        });
        names
    }

    /// Whether any filter of `type_name` has request conditions matching the
    /// request carried by `ctx` (an unconditional entry always matches).
    ///
    /// Used by protocol-level fallbacks that emit on a filter's behalf, so
    /// an operator's `when`/`unless` scoping is honored outside the normal
    /// request phase. Evaluates against both the request's bound-upstream view
    /// and the load balancer's published selection carried by `ctx`, so
    /// `bound_upstream` and `selected_upstream` predicates are honored on the
    /// fallback path exactly as during the normal request phase. The selection
    /// restored by `logging_cleanup` is on the context, so a
    /// `selected_upstream`-scoped filter is matched against it rather than
    /// silently dropped; passing the request alone would evaluate every such
    /// predicate against an empty view and mis-gate the fallback record. Absent
    /// metadata still fails closed.
    pub fn filter_request_conditions_match(&self, type_name: &str, ctx: &crate::HttpFilterContext<'_>) -> bool {
        let bound = ctx.bound_upstream_view();
        let selected = http_utils::ctx_selected_upstream(ctx);
        self.filters
            .iter()
            .filter(|pf| pf.filter.name() == type_name)
            .any(|pf| crate::condition::should_execute_bound_selected(&pf.conditions, ctx.request, bound, selected))
    }

    /// Ask each filter to emit its end-of-request record, returning
    /// whether any did.
    ///
    /// Used by the protocol layer's logging phase for requests that never
    /// reached a filter's own completion hooks. Filters that log nothing
    /// are no-ops, so this is cheap for pipelines without an access log.
    pub fn emit_deferred_records(&self, ctx: &crate::HttpFilterContext<'_>, status: u16) -> bool {
        // Short-circuits on the first filter that claims the record: one
        // access record per request, matching the completion hooks, which
        // mark the request logged after the first emit.
        self.filters
            .iter()
            .any(|pf| pf.filter.emit_deferred_record(ctx, status))
    }

    /// Whether any filter in this pipeline rewrites response trailers.
    ///
    /// Lets the protocol layer skip the trailer hook entirely for the
    /// pipelines — nearly all of them — that do not need it.
    pub fn needs_response_trailers(&self) -> bool {
        self.body_capabilities.needs_response_trailers
    }

    /// Compression configuration, if a compression filter is present.
    pub fn compression_config(&self) -> Option<&CompressionConfig> {
        self.compression.as_ref()
    }

    /// Set the shared [`HealthRegistry`] for this pipeline.
    pub fn set_health_registry(&mut self, registry: HealthRegistry) {
        self.visit_nested_pipelines(&mut |pipeline| pipeline.set_health_registry(Arc::clone(&registry)));
        self.health_registry = Some(registry);
    }

    /// Enable or disable recording of per-filter duration metrics.
    pub fn set_record_filter_duration_metrics(&mut self, enabled: bool) {
        self.visit_nested_pipelines(&mut |pipeline| pipeline.set_record_filter_duration_metrics(enabled));
        self.record_filter_duration_metrics = enabled;
    }

    /// Whether per-filter duration metrics are recorded.
    pub fn records_filter_duration_metrics(&self) -> bool {
        self.record_filter_duration_metrics
    }

    /// Set the compiled path templates used for the `route` metric label.
    pub fn set_route_templates(&mut self, templates: Arc<praxis_core::config::RouteTemplates>) {
        self.visit_nested_pipelines(&mut |pipeline| pipeline.set_route_templates(Arc::clone(&templates)));
        self.route_templates = templates;
    }

    /// The compiled path templates for the `route` metric label.
    pub fn route_templates(&self) -> &praxis_core::config::RouteTemplates {
        &self.route_templates
    }

    /// The shared health registry, if set.
    pub fn health_registry(&self) -> Option<&HealthRegistry> {
        self.health_registry.as_ref()
    }

    /// The shared request ID generator.
    pub fn id_generator(&self) -> &IdGenerator {
        &self.id_generator
    }

    /// Override the [`IdGenerator`] for this pipeline.
    pub fn set_id_generator(&mut self, generator: Arc<IdGenerator>) {
        self.visit_nested_pipelines(&mut |pipeline| pipeline.set_id_generator(Arc::clone(&generator)));
        self.id_generator = generator;
    }

    /// The shared KV store registry, if set.
    pub fn kv_stores(&self) -> Option<&KvStoreRegistry> {
        self.kv_stores.as_ref()
    }

    /// Set the shared [`KvStoreRegistry`] for this pipeline.
    pub fn set_kv_stores(&mut self, stores: KvStoreRegistry) {
        self.visit_nested_pipelines(&mut |pipeline| pipeline.set_kv_stores(stores.clone()));
        self.kv_stores = Some(stores);
    }

    /// The shared session store registry, if set.
    pub fn session_stores(&self) -> Option<&Arc<crate::SessionStoreRegistry>> {
        self.session_stores.as_ref()
    }

    /// Set the shared [`crate::SessionStoreRegistry`] for this pipeline.
    pub fn set_session_stores(&mut self, stores: Arc<crate::SessionStoreRegistry>) {
        self.visit_nested_pipelines(&mut |pipeline| pipeline.set_session_stores(Arc::clone(&stores)));
        self.session_stores = Some(stores);
    }

    /// The shared sub-request client, if set.
    pub fn subrequest_client(&self) -> Option<&praxis_core::subrequest::SubRequestClient> {
        self.subrequest_client.as_ref()
    }

    /// Whether any filter in this pipeline or its branches may select a
    /// streaming sub-request response.
    pub fn may_select_streaming_subrequest_response(&self) -> bool {
        self.may_select_streaming_subrequest_response
    }

    /// Set the shared [`SubRequestClient`] for this pipeline.
    ///
    /// [`SubRequestClient`]: praxis_core::subrequest::SubRequestClient
    pub fn set_subrequest_client(&mut self, client: praxis_core::subrequest::SubRequestClient) {
        self.visit_nested_pipelines(&mut |pipeline| pipeline.set_subrequest_client(client.clone()));
        self.subrequest_client = Some(client);
    }

    /// Register an external pipeline extension.
    ///
    /// Extensions are called once per request via
    /// [`prepare_extensions`] to inject pipeline-scoped resources
    /// into the per-request [`RequestExtensions`] container.
    ///
    /// [`prepare_extensions`]: FilterPipeline::prepare_extensions
    /// [`RequestExtensions`]: crate::RequestExtensions
    pub fn add_pipeline_extension(&mut self, ext: Box<dyn PipelineExtension>) {
        self.pipeline_extensions.push(ext);
    }

    /// Inject pipeline-level resources into per-request extensions.
    ///
    /// Called by the protocol adapter when building each request's
    /// filter context. Delegates to each registered
    /// [`PipelineExtension`].
    pub fn prepare_extensions(&self, extensions: &mut RequestExtensions) {
        for ext in &self.pipeline_extensions {
            ext.prepare(extensions);
        }
    }

    /// The wall-clock time source.
    pub fn time_source(&self) -> &dyn TimeSource {
        &*self.time_source
    }

    /// Override the [`TimeSource`] for this pipeline.
    pub fn set_time_source(&mut self, source: Arc<dyn TimeSource>) {
        self.visit_nested_pipelines(&mut |pipeline| pipeline.set_time_source(Arc::clone(&source)));
        self.time_source = source;
    }

    /// Filesystem paths the filters in this pipeline read configuration from,
    /// beyond the main Praxis config.
    ///
    /// Used by the config watcher to decide which files should trigger a reload.
    /// Without this, a filter that loads an external document never picks up edits
    /// to it, because the reload gate only ever sees the main config's bytes.
    ///
    /// Walks branch sub-chains recursively, so a document referenced solely from
    /// a filter inside a branch is observed too.
    pub fn referenced_files(&self) -> Vec<std::path::PathBuf> {
        let mut files = Vec::new();
        for_each_pipeline_filter(&self.filters, &mut |pf| {
            if let AnyFilter::Http(f) = &pf.filter {
                files.extend(f.referenced_files());
            }
        });
        files
    }

    /// Whether upstream hostnames are allowed to resolve to private or
    /// reserved IP addresses.
    ///
    /// Mirrors `insecure_options.allow_private_upstreams`. Defaults to
    /// `false`, so a pipeline that was never configured fails closed.
    pub fn allow_private_upstreams(&self) -> bool {
        self.allow_private_upstreams
    }

    /// Set the runtime private-upstream override, including for nested pipelines.
    ///
    /// Called once per (re)configuration from
    /// `insecure_options.allow_private_upstreams`, so a hot reload that
    /// flips the flag takes effect with the swapped-in pipeline.
    pub fn set_allow_private_upstreams(&mut self, allow: bool) {
        self.visit_nested_pipelines(&mut |pipeline| pipeline.set_allow_private_upstreams(allow));
        self.allow_private_upstreams = allow;
    }

    /// Apply [`InsecureOptions`] to all filters in the pipeline.
    ///
    /// Delegates to each filter's [`apply_insecure_options`] method, recursing
    /// into branch sub-chains so a filter buried in a branch honors the override
    /// too. Filters that support insecure overrides (e.g. CSRF log-only mode)
    /// handle the relevant flags; others ignore the call.
    ///
    /// [`apply_insecure_options`]: crate::HttpFilter::apply_insecure_options
    /// [`InsecureOptions`]: praxis_core::config::InsecureOptions
    pub fn apply_insecure_options(&self, options: &InsecureOptions) {
        for_each_pipeline_filter(&self.filters, &mut |pf| {
            if let AnyFilter::Http(f) = &pf.filter {
                f.apply_insecure_options(options);
            }
        });
    }

    /// Apply a mutation to every pipeline directly embedded by a filter,
    /// including filters nested inside branch sub-chains.
    fn visit_nested_pipelines(&mut self, visitor: &mut dyn FnMut(&mut FilterPipeline)) {
        visit_branch_nested_pipelines(&mut self.filters, visitor);
    }

    /// Whether any filter in this pipeline (including branch sub-chains and
    /// nested framework pipelines) selects its cluster from the frozen logical
    /// binding.
    ///
    /// A framework filter that owns nested pipelines (the IRR) folds its
    /// reachable steps' consumption up through [`consumes_bound_upstream`], so
    /// this walk sees an IRR-step bound consumer even though
    /// [`for_each_pipeline_filter`] does not descend into step pipelines
    /// directly.
    ///
    /// [`consumes_bound_upstream`]: crate::HttpFilter::consumes_bound_upstream
    #[cfg(feature = "iterative-request-router")]
    pub(crate) fn consumes_bound_upstream(&self) -> bool {
        let mut found = false;
        for_each_pipeline_filter(&self.filters, &mut |pf| {
            if let AnyFilter::Http(f) = &pf.filter {
                found = found || f.consumes_bound_upstream();
            }
        });
        found
    }

    /// Whether this pipeline contains any logical-binding observer or consumer.
    #[cfg(feature = "iterative-request-router")]
    pub(crate) fn uses_bound_upstream(&self) -> bool {
        checks::uses_bound_upstream(&self.filters)
    }

    /// Clusters this pipeline's bound-source consumers declare, at any branch
    /// depth.
    #[cfg(feature = "iterative-request-router")]
    pub(crate) fn bound_upstream_candidates(&self) -> std::collections::HashSet<String> {
        checks::bound_consumer_clusters(&self.filters)
    }

    /// The `when: bound_upstream` matchers on this pipeline's filters, at any
    /// branch depth, in pipeline order.
    #[cfg(feature = "iterative-request-router")]
    pub(crate) fn bound_when_matchers(&self) -> Vec<praxis_core::config::ApplicationMatch> {
        checks::bound_when_matchers(&self.filters)
    }

    /// Whether every path through this pipeline, entered already bound to
    /// `cluster`, reaches a load balancer serving it or an answer.
    #[cfg(feature = "iterative-request-router")]
    pub(crate) fn serves_bound_cluster(
        &self,
        cluster: &str,
        metadata: Option<&catalog::ClusterApplicationMetadata>,
    ) -> bool {
        checks::serves_bound_cluster(&self.filters, cluster, metadata)
    }

    /// Cluster application-metadata declarations from every filter in this
    /// pipeline, descending into branch sub-chains.
    ///
    /// A framework filter that owns nested pipelines folds its steps'
    /// declarations up through [`declared_cluster_metadata`], so an IRR step's
    /// cluster metadata reaches the parent catalog and its conflict check.
    ///
    /// [`declared_cluster_metadata`]: crate::HttpFilter::declared_cluster_metadata
    #[cfg(feature = "iterative-request-router")]
    pub(crate) fn cluster_metadata_declarations(&self) -> Vec<catalog::ClusterMetadataDeclaration> {
        collect_cluster_declarations(&self.filters)
    }
}

/// Visit every filter in `filters` and, recursively, every filter nested inside
/// their branch sub-chains.
///
/// The shared read-only walk behind the pipeline-wide scans (non-HTTP filter
/// detection, terminal-filter detection, referenced-file discovery, insecure
/// option application) so a filter buried in a branch is treated exactly like a
/// top-level one rather than silently skipped.
fn for_each_pipeline_filter(filters: &[PipelineFilter], visit: &mut dyn FnMut(&PipelineFilter)) {
    for pf in filters {
        visit(pf);
        for branch in &pf.branches {
            for_each_pipeline_filter(&branch.filters, visit);
        }
    }
}

/// Collect every filter's cluster application-metadata declarations,
/// descending into branch sub-chains.
///
/// Shared by pipeline construction (which folds these into the runtime
/// catalog) and validation (which detects conflicting declarations). Walks
/// the same reachable filter set as the other pipeline-wide scans so a
/// cluster declared only inside a branch still contributes.
#[cfg(feature = "upstream-binding")]
pub(super) fn collect_cluster_declarations(filters: &[PipelineFilter]) -> Vec<catalog::ClusterMetadataDeclaration> {
    let mut declarations = Vec::new();
    for_each_pipeline_filter(filters, &mut |pf| {
        if let AnyFilter::Http(f) = &pf.filter {
            declarations.extend(f.declared_cluster_metadata());
        }
    });
    declarations
}

/// Invoke each filter's [`visit_nested_pipelines`], descending recursively into
/// branch sub-chains so pipelines embedded by branch-contained filters (for
/// example an outbound chain bound by a callout placed inside a branch) receive
/// the same mutation as top-level ones.
///
/// [`visit_nested_pipelines`]: crate::HttpFilter::visit_nested_pipelines
fn visit_branch_nested_pipelines(filters: &mut [PipelineFilter], visitor: &mut dyn FnMut(&mut FilterPipeline)) {
    for pf in filters {
        if let AnyFilter::Http(filter) = &mut pf.filter {
            filter.visit_nested_pipelines(visitor);
        }
        for branch in &mut pf.branches {
            visit_branch_nested_pipelines(&mut branch.filters, visitor);
        }
    }
}

// -----------------------------------------------------------------------------
// Body Limit Utilities
// -----------------------------------------------------------------------------

/// Tighten a body mode's size limit to the given ceiling.
/// When `filter_declared` is true a filter explicitly chose Stream
/// mode; preserve it so streaming filters keep working. When false
/// the mode is the default (no filter needs body); convert to
/// `SizeLimit` so the body limit is enforced by buffering.
fn clamp_body_mode(mode: BodyMode, ceiling: usize, filter_declared: bool) -> BodyMode {
    match mode {
        BodyMode::StreamBuffer { max_bytes } => BodyMode::StreamBuffer {
            max_bytes: Some(max_bytes.map_or(ceiling, |m| m.min(ceiling))),
        },
        BodyMode::SizeLimit { max_bytes } => BodyMode::SizeLimit {
            max_bytes: max_bytes.min(ceiling),
        },
        BodyMode::Stream if filter_declared => BodyMode::Stream,
        BodyMode::Stream => BodyMode::SizeLimit { max_bytes: ceiling },
    }
}

/// Reject or clamp unbounded [`StreamBuffer`] body modes.
///
/// When `allow_unbounded` is `true`, the mode is clamped to
/// [`ABSOLUTE_MAX_BODY_BYTES`] and a warning is emitted.
///
/// # Errors
///
/// Returns [`FilterError`] when the body mode is unbounded
/// and `allow_unbounded` is `false`.
///
/// [`StreamBuffer`]: BodyMode::StreamBuffer
/// [`ABSOLUTE_MAX_BODY_BYTES`]: praxis_core::config::ABSOLUTE_MAX_BODY_BYTES
fn check_unbounded_stream_buffer(
    direction: &str,
    mode: &mut BodyMode,
    allow_unbounded: bool,
) -> Result<(), FilterError> {
    if let BodyMode::StreamBuffer { max_bytes: max @ None } = mode {
        if allow_unbounded {
            warn!(
                direction = direction,
                ceiling = ABSOLUTE_MAX_BODY_BYTES,
                "StreamBuffer body mode has no per-filter size limit; \
                 clamped to absolute ceiling ({} MiB)",
                ABSOLUTE_MAX_BODY_BYTES / 1_048_576
            );
            *max = Some(ABSOLUTE_MAX_BODY_BYTES);
        } else {
            return Err(format!(
                "StreamBuffer {direction} body mode has no size limit; \
                 set max_{direction}_body_bytes or set \
                 insecure_options.allow_unbounded_body: true to allow"
            )
            .into());
        }
    }
    Ok(())
}

// -----------------------------------------------------------------------------
// Failure Mode
// -----------------------------------------------------------------------------

/// Check failure mode and either swallow or propagate a filter error.
///
/// When `failure_mode` is [`FailureMode::Open`], the error is logged as a
/// warning and `Ok(())` is returned so the caller can continue.
pub(crate) fn check_failure_mode(
    filter_name: &str,
    error: FilterError,
    phase: &str,
    failure_mode: FailureMode,
) -> Result<(), FilterError> {
    match failure_mode {
        FailureMode::Open => {
            warn!(
                filter = filter_name,
                error = %error,
                phase,
                failure_mode = "open",
                "filter error, continuing"
            );
            Ok(())
        },
        FailureMode::Closed => {
            error!(
                filter = filter_name,
                error = %error,
                phase,
                failure_mode = "closed",
                "filter error, aborting request"
            );
            Err(error)
        },
    }
}