praxis-proxy-filter 0.7.2

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

//! Branch evaluation and execution during the HTTP request phase.
//!
//! Called by [`execute_http_request`] after each filter runs.
//! [`evaluate_branches`] checks every branch on the current filter:
//!
//! 1. Does the [`on_result`] condition match `ctx.filter_results`?
//! 2. Is the re-entrance limit (if any) still within bounds?
//! 3. Execute the branch's filter list.
//! 4. Map the rejoin target to a [`BranchOutcome`] that controls the parent while-loop index.
//!
//! **First-match-wins**: once a branch fires and produces a
//! non-[`Continue`] outcome, evaluation stops.
//!
//! **Results clearing**: `ctx.filter_results` is cleared after
//! branch evaluation at each filter. For [`ReEnter`], results
//! are also cleared to prevent stale data from re-triggering
//! conditional branches without the filter re-executing.
//!
//! [`execute_http_request`]: super::FilterPipeline::execute_http_request
//! [`on_result`]: praxis_core::config::BranchChainConfig::on_result
//! [`BranchOutcome`]: super::branch::BranchOutcome
//! [`Continue`]: super::branch::BranchOutcome::Continue
//! [`ReEnter`]: super::branch::RejoinTarget::ReEnter

use std::{pin::Pin, sync::Arc};

use praxis_core::config::FailureMode;
use tracing::{debug, trace, warn};

use super::{
    branch::{BranchOutcome, RejoinTarget, ResolvedBranch, record_executed_branch_filter},
    check_failure_mode,
    filter::PipelineFilter,
};
use crate::{
    FilterError, actions::FilterAction, any_filter::AnyFilter, condition::should_execute_bound_selected,
    context::HttpFilterContext, trace_context::ensure_trace_context,
};

// -----------------------------------------------------------------------------
// Branch Evaluation
// -----------------------------------------------------------------------------

/// Evaluate all branches on a filter, executing matching ones.
///
/// The branch-free case takes a synchronous fast path; nested branch
/// recursion goes through [`evaluate_branches_boxed`].
pub(crate) async fn evaluate_branches(
    branches: &[ResolvedBranch],
    ctx: &mut HttpFilterContext<'_>,
) -> Result<BranchOutcome, FilterError> {
    if branches.is_empty() {
        retain_filter_results(ctx);
        ctx.filter_results.clear();
        return Ok(BranchOutcome::Continue);
    }
    evaluate_branches_inner(branches, ctx).await
}

/// Boxed entry point breaking the async recursion cycle for nested
/// branches (branches within branches).
fn evaluate_branches_boxed<'a>(
    branches: &'a [ResolvedBranch],
    ctx: &'a mut HttpFilterContext<'_>,
) -> Pin<Box<dyn Future<Output = Result<BranchOutcome, FilterError>> + Send + 'a>> {
    Box::pin(evaluate_branches_inner(branches, ctx))
}

/// Inner implementation of branch evaluation.
#[expect(clippy::too_many_lines, reason = "branch execution and optional result retention")]
async fn evaluate_branches_inner(
    branches: &[ResolvedBranch],
    ctx: &mut HttpFilterContext<'_>,
) -> Result<BranchOutcome, FilterError> {
    // Branch conditions read the host filter's results. Snapshot them
    // lazily: only executing a branch chain can clear
    // `ctx.filter_results` mid-loop and wipe later siblings' condition
    // inputs, so the clone (a map plus per-entry string copies) is paid
    // just before the first branch fires — the common
    // no-branch-fires case reads the live results for free.
    let mut host_results: Option<std::collections::HashMap<&'static str, crate::FilterResultSet>> = None;
    for branch in branches {
        let fires = should_branch_fire(branch, host_results.as_ref().unwrap_or(&ctx.filter_results));
        if !fires {
            trace!(branch = %branch.name, "branch condition not met");
            continue;
        }

        if !check_reentrance_limit(branch, ctx) {
            continue;
        }

        debug!(
            branch = %branch.name,
            "executing branch chain"
        );

        if host_results.is_none() {
            host_results = Some(ctx.filter_results.clone());
        }
        let action = execute_branch_filters(&branch.filters, ctx).await?;

        match action {
            FilterAction::Reject(r) => return Ok(BranchOutcome::Reject(r)),
            FilterAction::TerminalResponse(t) => return Ok(BranchOutcome::TerminalResponse(t)),
            FilterAction::StreamingTerminalResponse(t) => {
                return Ok(BranchOutcome::StreamingTerminalResponse(t));
            },
            _ => {},
        }

        match &branch.rejoin {
            RejoinTarget::Next => {},
            RejoinTarget::Terminal => return Ok(BranchOutcome::Terminal),
            RejoinTarget::SkipTo(target) => {
                retain_filter_results(ctx);
                ctx.filter_results.clear();
                return Ok(BranchOutcome::SkipTo(*target));
            },
            RejoinTarget::ReEnter(target) => {
                retain_filter_results(ctx);
                ctx.filter_results.clear();
                return Ok(BranchOutcome::ReEnter(*target));
            },
        }
    }

    retain_filter_results(ctx);
    ctx.filter_results.clear();

    Ok(BranchOutcome::Continue)
}

/// Snapshot ephemeral results when a caller opted into cross-phase retention.
fn retain_filter_results(ctx: &mut HttpFilterContext<'_>) {
    if ctx.filter_results.is_empty() {
        return;
    }
    if let Some(retained) = ctx.extensions.get_mut::<crate::results::RetainedFilterResults>() {
        // Every caller clears `filter_results` right after this, so the
        // sets can be moved rather than cloned; the follow-up clear
        // becomes a no-op.
        retained.0.extend(ctx.filter_results.drain());
    }
}

/// Check whether a branch's condition is met against the host filter's
/// results (snapshotted before any sibling branch chain executes).
fn should_branch_fire(
    branch: &ResolvedBranch,
    host_results: &std::collections::HashMap<&'static str, crate::FilterResultSet>,
) -> bool {
    match &branch.condition {
        None => true,
        Some(cond) => crate::matches_filter_result(
            host_results,
            cond.filter_name.as_ref(),
            cond.key.as_ref(),
            cond.value.as_ref(),
        ),
    }
}

/// Check re-entrance limits, returning false if exceeded.
fn check_reentrance_limit(branch: &ResolvedBranch, ctx: &mut HttpFilterContext<'_>) -> bool {
    if let RejoinTarget::ReEnter(_) = branch.rejoin {
        let count = ctx.branch_iterations.entry(Arc::clone(&branch.name)).or_insert(0);
        *count += 1;
        if let Some(max) = branch.max_iterations
            && *count > max
        {
            debug!(
                branch = %branch.name,
                iterations = *count,
                "max iterations exceeded, falling through"
            );
            return false;
        }
    }
    true
}

/// Execute a branch's filter list.
///
/// Each filter whose `on_request` ran (including one that rejected or
/// answered) is marked in the context's [`executed_branch_filters`] so the
/// response phase runs its `on_response`; a filter skipped by conditions, or
/// whose error propagated under `failure_mode: closed`, is not.
///
/// [`executed_branch_filters`]: HttpFilterContext::executed_branch_filters
async fn execute_branch_filters(
    filters: &[PipelineFilter],
    ctx: &mut HttpFilterContext<'_>,
) -> Result<FilterAction, FilterError> {
    ensure_branch_trace_context(filters, ctx);
    for pf in filters {
        let AnyFilter::Http(http_filter) = &pf.filter else {
            continue;
        };
        if !pf.conditions.is_empty()
            && !should_execute_bound_selected(
                &pf.conditions,
                ctx.request,
                ctx.bound_upstream_view(),
                super::http_utils::ctx_selected_upstream(ctx),
            )
        {
            continue;
        }
        ctx.current_filter_id = Some(pf.filter_id);
        let result = http_filter.on_request(ctx).await;
        ctx.current_filter_id = None;
        let stop = branch_request_outcome(result, http_filter.name(), pf.failure_mode)?;
        record_executed_branch_filter(&mut ctx.executed_branch_filters, pf.filter_id);
        if let Some(action) = stop {
            return Ok(action);
        }
        if let Some(action) = dispatch_nested_outcome(&pf.branches, ctx).await? {
            return Ok(action);
        }
    }
    Ok(FilterAction::Continue)
}

/// Initialize correlation only after a branch is selected and its trace filter matches.
fn ensure_branch_trace_context(filters: &[PipelineFilter], ctx: &mut HttpFilterContext<'_>) {
    let bound = ctx.bound_upstream_view();
    let selected = super::http_utils::ctx_selected_upstream(ctx);
    if filters.iter().any(|pf| {
        pf.filter.name() == "trace_context"
            && should_execute_bound_selected(&pf.conditions, ctx.request, bound, selected)
    }) {
        ensure_trace_context(ctx);
    }
}

/// Map a branch filter's request-phase result to a parent-loop action.
///
/// `None` continues the branch loop; `Some` stops it (a terminal response or
/// rejection). An error is routed through [`check_failure_mode`], which either
/// swallows it (fail-open) or propagates it (fail-closed).
fn branch_request_outcome(
    result: Result<FilterAction, FilterError>,
    filter_name: &str,
    failure_mode: FailureMode,
) -> Result<Option<FilterAction>, FilterError> {
    match result {
        Ok(FilterAction::Continue | FilterAction::Release | FilterAction::BodyDone) => Ok(None),
        Ok(FilterAction::Reject(r)) => Ok(Some(FilterAction::Reject(r))),
        Ok(FilterAction::TerminalResponse(t)) => Ok(Some(FilterAction::TerminalResponse(t))),
        Ok(FilterAction::StreamingTerminalResponse(t)) => Ok(Some(FilterAction::StreamingTerminalResponse(t))),
        Err(e) => {
            check_failure_mode(filter_name, e, "branch request", failure_mode)?;
            Ok(None)
        },
    }
}

/// Evaluate nested branches and convert their outcome for the parent.
///
/// Returns `Some(action)` when the parent should stop iteration
/// (terminal or reject), `None` to continue.
///
/// `SkipTo` and `ReEnter` from nested branches are **deliberately
/// discarded** because they reference indices in the parent
/// pipeline, which the nested branch has no authority to control.
/// Only `Terminal` and `Reject` propagate upward.
///
/// A nested `terminal`/`client` rejoin whose sub-chain produced no
/// response fails closed with a 500, matching the top-level pipeline:
/// "stop; respond to client" must never degrade into continuing the
/// pipeline with the remaining filters skipped.
async fn dispatch_nested_outcome(
    branches: &[ResolvedBranch],
    ctx: &mut HttpFilterContext<'_>,
) -> Result<Option<FilterAction>, FilterError> {
    let outcome = if branches.is_empty() {
        retain_filter_results(ctx);
        ctx.filter_results.clear();
        BranchOutcome::Continue
    } else {
        evaluate_branches_boxed(branches, ctx).await?
    };
    Ok(map_nested_outcome(outcome))
}

/// Convert a nested branch outcome into the parent's stopping action.
fn map_nested_outcome(outcome: BranchOutcome) -> Option<FilterAction> {
    match outcome {
        BranchOutcome::Continue => None,
        BranchOutcome::SkipTo(target) => {
            warn!(
                target,
                "discarding SkipTo from nested branch; nested control flow does not propagate"
            );
            None
        },
        BranchOutcome::ReEnter(target) => {
            warn!(
                target,
                "discarding ReEnter from nested branch; nested control flow does not propagate"
            );
            None
        },
        BranchOutcome::Terminal => {
            warn!("nested terminal branch produced no response; failing closed with 500");
            Some(FilterAction::Reject(crate::actions::Rejection::status(500)))
        },
        BranchOutcome::Reject(r) => Some(FilterAction::Reject(r)),
        BranchOutcome::TerminalResponse(t) => Some(FilterAction::TerminalResponse(t)),
        BranchOutcome::StreamingTerminalResponse(t) => Some(FilterAction::StreamingTerminalResponse(t)),
    }
}

// -----------------------------------------------------------------------------
// Tests
// -----------------------------------------------------------------------------

#[cfg(test)]
#[expect(clippy::allow_attributes, reason = "blanket test suppressions")]
#[allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::indexing_slicing,
    clippy::panic,
    clippy::too_many_lines,
    clippy::doc_markdown,
    reason = "tests"
)]
mod tests {
    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};

    use async_trait::async_trait;
    use bytes::Bytes;
    use http::Method;

    use super::*;
    use crate::{
        Rejection, StreamingTerminalResponse,
        filter::HttpFilter,
        pipeline::branch::{ResolvedBranchCondition, branch_filter_executed},
        results::FilterResultSet,
    };

    #[tokio::test]
    async fn opted_in_results_survive_branch_evaluation() {
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        ctx.extensions.insert(crate::results::RetainedFilterResults::default());
        ctx.filter_results
            .entry("probe")
            .or_default()
            .set("result", "yes")
            .unwrap();

        let outcome = evaluate_branches(&[], &mut ctx).await.unwrap();

        assert!(matches!(outcome, BranchOutcome::Continue));
        assert!(ctx.filter_results.is_empty());
        assert!(
            ctx.extensions
                .get::<crate::results::RetainedFilterResults>()
                .unwrap()
                .0
                .get("probe")
                .is_some_and(|results| results.matches("result", "yes"))
        );
    }

    #[tokio::test]
    async fn unconditional_branch_fires() {
        let counter = Arc::new(AtomicUsize::new(0));
        let branches = vec![make_branch(
            "uncond",
            None,
            RejoinTarget::Next,
            None,
            vec![counting_pf(Arc::clone(&counter))],
        )];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let outcome = evaluate_branches(&branches, &mut ctx).await.unwrap();
        assert!(
            matches!(outcome, BranchOutcome::Continue),
            "unconditional branch with Next rejoin should continue"
        );
        assert_eq!(
            counter.load(Ordering::SeqCst),
            1,
            "unconditional branch filter should have executed"
        );
    }

    #[tokio::test]
    async fn conditional_branch_fires_on_match() {
        let counter = Arc::new(AtomicUsize::new(0));
        let branches = vec![make_branch(
            "cond_match",
            Some(("cache", "status", "hit")),
            RejoinTarget::Next,
            None,
            vec![counting_pf(Arc::clone(&counter))],
        )];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let mut rs = FilterResultSet::new();
        rs.set("status", "hit").unwrap();
        ctx.filter_results.insert("cache", rs);
        let outcome = evaluate_branches(&branches, &mut ctx).await.unwrap();
        assert!(
            matches!(outcome, BranchOutcome::Continue),
            "matching conditional branch should continue"
        );
        assert_eq!(
            counter.load(Ordering::SeqCst),
            1,
            "matching branch filter should have executed"
        );
    }

    #[tokio::test]
    async fn conditional_branch_skips_on_mismatch() {
        let counter = Arc::new(AtomicUsize::new(0));
        let branches = vec![make_branch(
            "cond_miss",
            Some(("cache", "status", "hit")),
            RejoinTarget::Next,
            None,
            vec![counting_pf(Arc::clone(&counter))],
        )];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let mut rs = FilterResultSet::new();
        rs.set("status", "miss").unwrap();
        ctx.filter_results.insert("cache", rs);
        let outcome = evaluate_branches(&branches, &mut ctx).await.unwrap();
        assert!(
            matches!(outcome, BranchOutcome::Continue),
            "mismatched branch should continue"
        );
        assert_eq!(
            counter.load(Ordering::SeqCst),
            0,
            "mismatched branch filter should not have executed"
        );
    }

    #[tokio::test]
    async fn selected_branch_enables_trace_context_before_earlier_callouts() {
        let saw_context = Arc::new(AtomicBool::new(false));
        let branches = vec![make_branch(
            "selected_trace",
            None,
            RejoinTarget::Next,
            None,
            vec![
                trace_probe_pf(Arc::clone(&saw_context)),
                named_pf("trace_context", vec![]),
            ],
        )];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);

        drop(evaluate_branches(&branches, &mut ctx).await.unwrap());

        assert!(saw_context.load(Ordering::SeqCst));
    }

    #[tokio::test]
    async fn unselected_branch_does_not_enable_trace_context() {
        let branches = vec![make_branch(
            "skipped_trace",
            Some(("cache", "status", "hit")),
            RejoinTarget::Next,
            None,
            vec![named_pf("trace_context", vec![])],
        )];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let mut results = FilterResultSet::new();
        results.set("status", "miss").unwrap();
        ctx.filter_results.insert("cache", results);

        drop(evaluate_branches(&branches, &mut ctx).await.unwrap());

        assert!(ctx.extensions.get::<crate::trace_context::TraceContext>().is_none());
    }

    #[tokio::test]
    async fn selected_branch_honors_trace_context_filter_conditions() {
        let conditions = vec![praxis_core::config::Condition::When(
            praxis_core::config::ConditionMatch {
                grpc: None,
                path: None,
                path_prefix: Some("/api".to_owned()),
                methods: None,
                headers: None,
                bound_upstream: None,
                selected_upstream: None,
            },
        )];
        let branches = vec![make_branch(
            "conditional_trace",
            None,
            RejoinTarget::Next,
            None,
            vec![named_pf("trace_context", conditions)],
        )];
        let req = crate::test_utils::make_request(Method::GET, "/healthz");
        let mut ctx = crate::test_utils::make_filter_context(&req);

        drop(evaluate_branches(&branches, &mut ctx).await.unwrap());

        assert!(ctx.extensions.get::<crate::trace_context::TraceContext>().is_none());
    }

    #[tokio::test]
    async fn terminal_rejoin_stops_parent() {
        let branches = vec![make_branch("term", None, RejoinTarget::Terminal, None, vec![])];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let outcome = evaluate_branches(&branches, &mut ctx).await.unwrap();
        assert!(
            matches!(outcome, BranchOutcome::Terminal),
            "terminal branch should stop parent"
        );
    }

    #[tokio::test]
    async fn skip_to_advances_to_target() {
        let branches = vec![make_branch("skip", None, RejoinTarget::SkipTo(5), None, vec![])];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let outcome = evaluate_branches(&branches, &mut ctx).await.unwrap();
        assert!(
            matches!(outcome, BranchOutcome::SkipTo(5)),
            "SkipTo branch should advance to target index 5"
        );
    }

    #[tokio::test]
    async fn skip_to_clears_filter_results() {
        let branches = vec![make_branch("skip", None, RejoinTarget::SkipTo(5), None, vec![])];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let mut rs = FilterResultSet::new();
        rs.set("status", "hit").unwrap();
        ctx.filter_results.insert("cache", rs);

        let outcome = evaluate_branches(&branches, &mut ctx).await.unwrap();

        assert!(matches!(outcome, BranchOutcome::SkipTo(5)), "branch should skip");
        assert!(
            ctx.filter_results.is_empty(),
            "SkipTo must clear ephemeral results like every other outcome"
        );
    }

    #[tokio::test]
    async fn reenter_loops_back() {
        let counter = Arc::new(AtomicUsize::new(0));
        let branches = vec![make_branch(
            "reenter",
            None,
            RejoinTarget::ReEnter(1),
            Some(3),
            vec![counting_pf(Arc::clone(&counter))],
        )];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let outcome = evaluate_branches(&branches, &mut ctx).await.unwrap();
        assert!(
            matches!(outcome, BranchOutcome::ReEnter(1)),
            "ReEnter branch should loop back to target index"
        );
        assert_eq!(
            counter.load(Ordering::SeqCst),
            1,
            "branch filter should execute once per evaluation"
        );
    }

    #[tokio::test]
    async fn reenter_max_iterations_exceeded_falls_through() {
        let counter = Arc::new(AtomicUsize::new(0));
        let branches = vec![make_branch(
            "limited",
            None,
            RejoinTarget::ReEnter(0),
            Some(2),
            vec![counting_pf(Arc::clone(&counter))],
        )];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);

        evaluate_branches(&branches, &mut ctx).await.unwrap();
        evaluate_branches(&branches, &mut ctx).await.unwrap();

        let outcome = evaluate_branches(&branches, &mut ctx).await.unwrap();
        assert!(
            matches!(outcome, BranchOutcome::Continue),
            "exceeded max_iterations should fall through to Continue"
        );
        assert_eq!(
            counter.load(Ordering::SeqCst),
            2,
            "branch should execute max_iterations times then stop"
        );
    }

    #[tokio::test]
    async fn branch_filter_reject_propagates() {
        let branches = vec![make_branch(
            "reject",
            None,
            RejoinTarget::Next,
            None,
            vec![reject_pf(403)],
        )];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let outcome = evaluate_branches(&branches, &mut ctx).await.unwrap();
        assert!(
            matches!(outcome, BranchOutcome::Reject(r) if r.status == 403),
            "branch filter rejection should propagate"
        );
    }

    #[tokio::test]
    async fn branch_filter_error_respects_failure_mode_open() {
        let branches = vec![make_branch(
            "open_error",
            None,
            RejoinTarget::Next,
            None,
            vec![error_pf(FailureMode::Open)],
        )];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let outcome = evaluate_branches(&branches, &mut ctx).await.unwrap();
        assert!(
            matches!(outcome, BranchOutcome::Continue),
            "failure_mode=open should swallow branch filter errors"
        );
        assert_eq!(ctx.current_filter_id, None, "filter id should be cleared after error");
    }

    #[tokio::test]
    async fn branch_filter_error_respects_failure_mode_closed() {
        let branches = vec![make_branch(
            "closed_error",
            None,
            RejoinTarget::Next,
            None,
            vec![error_pf(FailureMode::Closed)],
        )];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let Err(err) = evaluate_branches(&branches, &mut ctx).await else {
            panic!("failure_mode=closed should propagate branch filter errors");
        };
        assert!(
            err.to_string().contains("branch error"),
            "unexpected branch filter error: {err}"
        );
        assert_eq!(ctx.current_filter_id, None, "filter id should be cleared after error");
    }

    #[tokio::test]
    async fn results_cleared_after_evaluation() {
        let branches: Vec<ResolvedBranch> = vec![];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let mut rs = FilterResultSet::new();
        rs.set("status", "hit").unwrap();
        ctx.filter_results.insert("cache", rs);
        assert!(
            !ctx.filter_results.is_empty(),
            "results should be present before evaluation"
        );
        evaluate_branches(&branches, &mut ctx).await.unwrap();
        assert!(
            ctx.filter_results.is_empty(),
            "results should be cleared after evaluation"
        );
    }

    #[tokio::test]
    async fn multiple_branches_first_match_wins() {
        let counter_a = Arc::new(AtomicUsize::new(0));
        let counter_b = Arc::new(AtomicUsize::new(0));
        let branches = vec![
            make_branch(
                "first",
                None,
                RejoinTarget::Terminal,
                None,
                vec![counting_pf(Arc::clone(&counter_a))],
            ),
            make_branch(
                "second",
                None,
                RejoinTarget::Terminal,
                None,
                vec![counting_pf(Arc::clone(&counter_b))],
            ),
        ];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let outcome = evaluate_branches(&branches, &mut ctx).await.unwrap();
        assert!(
            matches!(outcome, BranchOutcome::Terminal),
            "first matching branch should win"
        );
        assert_eq!(counter_a.load(Ordering::SeqCst), 1, "first branch should execute");
        assert_eq!(counter_b.load(Ordering::SeqCst), 0, "second branch should not execute");
    }

    #[tokio::test]
    async fn sibling_branch_condition_survives_earlier_branch_execution() {
        let counter_a = Arc::new(AtomicUsize::new(0));
        let counter_b = Arc::new(AtomicUsize::new(0));
        let branches = vec![
            make_branch(
                "first",
                Some(("cache", "status", "hit")),
                RejoinTarget::Next,
                None,
                vec![counting_pf(Arc::clone(&counter_a))],
            ),
            make_branch(
                "second",
                Some(("cache", "status", "hit")),
                RejoinTarget::Next,
                None,
                vec![counting_pf(Arc::clone(&counter_b))],
            ),
        ];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let mut rs = FilterResultSet::new();
        rs.set("status", "hit").unwrap();
        ctx.filter_results.insert("cache", rs);
        let outcome = evaluate_branches(&branches, &mut ctx).await.unwrap();
        assert!(
            matches!(outcome, BranchOutcome::Continue),
            "next-rejoining branches should continue"
        );
        assert_eq!(counter_a.load(Ordering::SeqCst), 1, "first branch should execute");
        assert_eq!(
            counter_b.load(Ordering::SeqCst),
            1,
            "second branch's condition must not be wiped by the first branch's execution"
        );
    }

    #[tokio::test]
    async fn empty_branches_is_noop() {
        let branches: Vec<ResolvedBranch> = vec![];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let outcome = evaluate_branches(&branches, &mut ctx).await.unwrap();
        assert!(
            matches!(outcome, BranchOutcome::Continue),
            "empty branches should continue"
        );
    }

    #[tokio::test]
    async fn nested_branch_terminal_propagates() {
        let inner_branch = make_branch("inner", None, RejoinTarget::Terminal, None, vec![]);
        let outer_filter = PipelineFilter {
            filter_id: 100,
            is_security: false,
            branches: vec![inner_branch],
            conditions: vec![],
            failure_mode: FailureMode::default(),
            filter: AnyFilter::Http(Box::new(NoopFilter)),
            name: None,
            response_conditions: vec![],
        };
        let outer_branches = vec![make_branch("outer", None, RejoinTarget::Next, None, vec![outer_filter])];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let outcome = evaluate_branches(&outer_branches, &mut ctx).await.unwrap();
        assert!(
            matches!(&outcome, BranchOutcome::Reject(r) if r.status == 500),
            "nested terminal with no response must fail closed, not continue the pipeline"
        );
    }

    #[tokio::test]
    async fn conditional_branch_skips_when_filter_absent_from_results() {
        let counter = Arc::new(AtomicUsize::new(0));
        let branches = vec![make_branch(
            "absent_filter",
            Some(("cache", "status", "hit")),
            RejoinTarget::Next,
            None,
            vec![counting_pf(Arc::clone(&counter))],
        )];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let outcome = evaluate_branches(&branches, &mut ctx).await.unwrap();
        assert!(
            matches!(outcome, BranchOutcome::Continue),
            "branch should not fire when referenced filter has no results"
        );
        assert_eq!(
            counter.load(Ordering::SeqCst),
            0,
            "branch filter should not execute when referenced filter is absent from results"
        );
    }

    #[tokio::test]
    async fn reenter_does_not_carry_stale_results_into_conditional_branch() {
        let counter = Arc::new(AtomicUsize::new(0));
        let branches = vec![make_branch(
            "cond_reenter",
            Some(("tracker", "status", "stale")),
            RejoinTarget::ReEnter(0),
            Some(3),
            vec![counting_pf(Arc::clone(&counter))],
        )];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);

        let mut rs = FilterResultSet::new();
        rs.set("status", "stale").unwrap();
        ctx.filter_results.insert("tracker", rs);

        let outcome = evaluate_branches(&branches, &mut ctx).await.unwrap();
        assert!(
            matches!(outcome, BranchOutcome::ReEnter(0)),
            "first call should fire and produce ReEnter"
        );
        assert_eq!(
            counter.load(Ordering::SeqCst),
            1,
            "branch should execute once on first evaluation"
        );

        let outcome = evaluate_branches(&branches, &mut ctx).await.unwrap();
        assert!(
            matches!(outcome, BranchOutcome::Continue),
            "second call should not fire because stale results were cleared"
        );
        assert_eq!(
            counter.load(Ordering::SeqCst),
            1,
            "branch should not execute again when stale result is gone"
        );
    }

    #[tokio::test]
    async fn reenter_max_iterations_at_ceiling_fires_exactly_100_times() {
        let counter = Arc::new(AtomicUsize::new(0));
        let branches = vec![make_branch(
            "ceiling",
            None,
            RejoinTarget::ReEnter(0),
            Some(100),
            vec![counting_pf(Arc::clone(&counter))],
        )];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);

        for i in 1..=100 {
            let outcome = evaluate_branches(&branches, &mut ctx).await.unwrap();
            assert!(
                matches!(outcome, BranchOutcome::ReEnter(0)),
                "iteration {i} should re-enter"
            );
        }
        assert_eq!(
            counter.load(Ordering::SeqCst),
            100,
            "branch should fire exactly 100 times"
        );

        let outcome = evaluate_branches(&branches, &mut ctx).await.unwrap();
        assert!(
            matches!(outcome, BranchOutcome::Continue),
            "iteration 101 should fall through"
        );
        assert_eq!(
            counter.load(Ordering::SeqCst),
            100,
            "branch should not fire after max_iterations exceeded"
        );
    }

    // -------------------------------------------------------------------------
    // Filter State in Branches
    // -------------------------------------------------------------------------

    #[tokio::test]
    async fn branch_filter_can_insert_and_read_own_state() {
        let log: ObsLog = Arc::new(std::sync::Mutex::new(Vec::new()));
        let branch_pf_id = NEXT_TEST_ID.fetch_add(1, Ordering::SeqCst);
        let branch_pf = PipelineFilter {
            filter_id: branch_pf_id,
            is_security: false,
            branches: vec![],
            conditions: vec![],
            failure_mode: FailureMode::default(),
            filter: AnyFilter::Http(Box::new(BranchStatefulFilter {
                id: 42,
                log: Arc::clone(&log),
            })),
            name: None,
            response_conditions: vec![],
        };
        let branches = vec![make_branch(
            "state_branch",
            None,
            RejoinTarget::Next,
            None,
            vec![branch_pf],
        )];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        evaluate_branches(&branches, &mut ctx).await.unwrap();
        let recorded = log.lock().unwrap().clone();
        assert_eq!(recorded, vec![(42, "insert")], "branch filter should insert state");
        assert!(
            ctx.filter_state.contains_key(&branch_pf_id),
            "state should exist at the branch filter's unique id"
        );
    }

    #[tokio::test]
    async fn branch_filter_state_does_not_collide_with_parent() {
        let log: ObsLog = Arc::new(std::sync::Mutex::new(Vec::new()));
        let parent_id = NEXT_TEST_ID.fetch_add(1, Ordering::SeqCst);
        let branch_id = NEXT_TEST_ID.fetch_add(1, Ordering::SeqCst);
        let branch_pf = PipelineFilter {
            filter_id: branch_id,
            is_security: false,
            branches: vec![],
            conditions: vec![],
            failure_mode: FailureMode::default(),
            filter: AnyFilter::Http(Box::new(BranchStatefulFilter {
                id: 99,
                log: Arc::clone(&log),
            })),
            name: None,
            response_conditions: vec![],
        };
        let branches = vec![make_branch(
            "state_branch",
            None,
            RejoinTarget::Next,
            None,
            vec![branch_pf],
        )];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        ctx.filter_state.insert(parent_id, Box::new(77_u64));
        evaluate_branches(&branches, &mut ctx).await.unwrap();
        let parent_state = ctx.filter_state.get(&parent_id).unwrap().downcast_ref::<u64>().unwrap();
        let branch_state = ctx.filter_state.get(&branch_id).unwrap().downcast_ref::<u64>().unwrap();
        assert_eq!(*parent_state, 77, "parent state should be unchanged");
        assert_eq!(*branch_state, 99, "branch filter should have its own state");
    }

    #[tokio::test]
    async fn two_branch_filters_of_same_type_get_independent_state() {
        let log: ObsLog = Arc::new(std::sync::Mutex::new(Vec::new()));
        let pf_a = stateful_pf(100, &log);
        let pf_b = stateful_pf(200, &log);
        let id_a = pf_a.filter_id;
        let id_b = pf_b.filter_id;
        let branches = vec![make_branch("dual", None, RejoinTarget::Next, None, vec![pf_a, pf_b])];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        evaluate_branches(&branches, &mut ctx).await.unwrap();
        let recorded = log.lock().unwrap().clone();
        assert_eq!(
            recorded,
            vec![(100, "insert"), (200, "insert")],
            "both branch filters should insert"
        );
        let state_a = ctx.filter_state.get(&id_a).unwrap().downcast_ref::<u64>().unwrap();
        let state_b = ctx.filter_state.get(&id_b).unwrap().downcast_ref::<u64>().unwrap();
        assert_eq!(*state_a, 100, "first branch filter state");
        assert_eq!(*state_b, 200, "second branch filter state");
    }

    #[tokio::test]
    async fn identity_is_none_after_branch_evaluation() {
        let branches = vec![make_branch(
            "check",
            None,
            RejoinTarget::Next,
            None,
            vec![counting_pf(Arc::new(AtomicUsize::new(0)))],
        )];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        evaluate_branches(&branches, &mut ctx).await.unwrap();
        assert!(
            ctx.current_filter_id.is_none(),
            "current_filter_id should be None after branch evaluation"
        );
    }

    #[tokio::test]
    async fn identity_is_none_after_branch_rejection() {
        let branches = vec![make_branch(
            "reject_branch",
            None,
            RejoinTarget::Next,
            None,
            vec![reject_pf(403)],
        )];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let outcome = evaluate_branches(&branches, &mut ctx).await.unwrap();
        assert!(matches!(outcome, BranchOutcome::Reject(_)), "should reject");
        assert!(
            ctx.current_filter_id.is_none(),
            "current_filter_id should be None after branch rejection"
        );
    }

    #[tokio::test]
    async fn reenter_without_max_iterations_always_fires() {
        let counter = Arc::new(AtomicUsize::new(0));
        let branches = vec![make_branch(
            "unlimited",
            None,
            RejoinTarget::ReEnter(0),
            None,
            vec![counting_pf(Arc::clone(&counter))],
        )];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);

        for i in 1..=10 {
            let outcome = evaluate_branches(&branches, &mut ctx).await.unwrap();
            assert!(
                matches!(outcome, BranchOutcome::ReEnter(0)),
                "iteration {i} should re-enter when max_iterations is None"
            );
        }
        assert_eq!(
            counter.load(Ordering::SeqCst),
            10,
            "branch should fire every time when max_iterations is None"
        );
    }

    #[tokio::test]
    async fn nested_branch_skip_to_discarded() {
        let inner_branch = make_branch("inner_skip", None, RejoinTarget::SkipTo(42), None, vec![]);
        let outer_filter = PipelineFilter {
            filter_id: NEXT_TEST_ID.fetch_add(1, Ordering::SeqCst),
            is_security: false,
            branches: vec![inner_branch],
            conditions: vec![],
            failure_mode: FailureMode::default(),
            filter: AnyFilter::Http(Box::new(NoopFilter)),
            name: None,
            response_conditions: vec![],
        };
        let outer_branches = vec![make_branch("outer", None, RejoinTarget::Next, None, vec![outer_filter])];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let outcome = evaluate_branches(&outer_branches, &mut ctx).await.unwrap();
        assert!(
            matches!(outcome, BranchOutcome::Continue),
            "nested SkipTo should be discarded and outer should continue"
        );
    }

    #[tokio::test]
    async fn branch_records_filters_whose_on_request_ran() {
        let gated = named_pf(
            "gated",
            vec![praxis_core::config::Condition::When(
                praxis_core::config::ConditionMatch {
                    grpc: None,
                    path: None,
                    path_prefix: Some("/api".to_owned()),
                    methods: None,
                    headers: None,
                    bound_upstream: None,
                    selected_upstream: None,
                },
            )],
        );
        let ran = named_pf("ran", vec![]);
        let rejecting = reject_pf(403);
        let unreached = named_pf("unreached", vec![]);
        let ids = [gated.filter_id, ran.filter_id, rejecting.filter_id, unreached.filter_id];
        let branches = vec![make_branch(
            "record",
            None,
            RejoinTarget::Next,
            None,
            vec![gated, ran, rejecting, unreached],
        )];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);

        drop(evaluate_branches(&branches, &mut ctx).await.unwrap());

        assert_eq!(
            ids.map(|id| branch_filter_executed(&ctx.executed_branch_filters, id)),
            [false, true, true, false],
            "only filters whose on_request ran (including the rejecting one) are recorded"
        );
    }

    #[tokio::test]
    async fn branch_does_not_record_a_closed_failure() {
        let failing = error_pf(FailureMode::Closed);
        let id = failing.filter_id;
        let branches = vec![make_branch("fail", None, RejoinTarget::Next, None, vec![failing])];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);

        let result = evaluate_branches(&branches, &mut ctx).await;

        assert!(
            matches!(&result, Err(e) if e.to_string().contains("branch error")),
            "failure_mode=closed should propagate the error"
        );
        assert!(
            !branch_filter_executed(&ctx.executed_branch_filters, id),
            "a filter whose error propagated must not be paired with on_response"
        );
    }

    #[tokio::test]
    async fn branch_free_evaluation_records_nothing() {
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);

        drop(evaluate_branches(&[], &mut ctx).await.unwrap());

        assert!(
            ctx.executed_branch_filters.is_empty(),
            "a request that runs no branch filter must leave the record empty"
        );
    }

    // -------------------------------------------------------------------------
    // Test Utilities
    // -------------------------------------------------------------------------

    /// Noop HTTP filter for branch testing.
    struct NoopFilter;

    #[async_trait]
    impl HttpFilter for NoopFilter {
        fn name(&self) -> &'static str {
            "noop"
        }

        async fn on_request(&self, _ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
            Ok(FilterAction::Continue)
        }
    }

    /// Noop filter with a caller-selected type name.
    struct NamedFilter(&'static str);

    #[async_trait]
    impl HttpFilter for NamedFilter {
        fn name(&self) -> &'static str {
            self.0
        }

        async fn on_request(&self, _ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
            Ok(FilterAction::Continue)
        }
    }

    /// Records whether trace context existed before this branch filter ran.
    struct TraceContextProbe {
        saw_context: Arc<AtomicBool>,
    }

    #[async_trait]
    impl HttpFilter for TraceContextProbe {
        fn name(&self) -> &'static str {
            "trace_context_probe"
        }

        async fn on_request(&self, ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
            self.saw_context.store(
                ctx.extensions.get::<crate::trace_context::TraceContext>().is_some(),
                Ordering::SeqCst,
            );
            Ok(FilterAction::Continue)
        }
    }

    /// HTTP filter that counts on_request invocations.
    struct CountFilter {
        counter: Arc<AtomicUsize>,
    }

    #[async_trait]
    impl HttpFilter for CountFilter {
        fn name(&self) -> &'static str {
            "count"
        }

        async fn on_request(&self, _ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
            self.counter.fetch_add(1, Ordering::SeqCst);
            Ok(FilterAction::Continue)
        }
    }

    /// HTTP filter that rejects with a given status.
    struct RejectFilter {
        status: u16,
    }

    #[async_trait]
    impl HttpFilter for RejectFilter {
        fn name(&self) -> &'static str {
            "reject"
        }

        async fn on_request(&self, _ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
            Ok(FilterAction::Reject(Rejection::status(self.status)))
        }
    }

    /// HTTP filter that returns an injected error.
    struct ErrorFilter;

    #[async_trait]
    impl HttpFilter for ErrorFilter {
        fn name(&self) -> &'static str {
            "error"
        }

        async fn on_request(&self, _ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
            Err("branch error".into())
        }
    }

    type ObsLog = Arc<std::sync::Mutex<Vec<(u64, &'static str)>>>;

    /// Filter that stores its id as typed state and records observations.
    struct BranchStatefulFilter {
        id: u64,
        log: ObsLog,
    }

    #[async_trait]
    impl HttpFilter for BranchStatefulFilter {
        fn name(&self) -> &'static str {
            "branch_stateful"
        }

        async fn on_request(&self, ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
            ctx.insert_filter_state(self.id);
            self.log.lock().unwrap().push((self.id, "insert"));
            Ok(FilterAction::Continue)
        }
    }

    /// Monotonic counter for test filter IDs.
    static NEXT_TEST_ID: AtomicUsize = AtomicUsize::new(1000);

    /// Build a counting [`PipelineFilter`].
    fn counting_pf(counter: Arc<AtomicUsize>) -> PipelineFilter {
        PipelineFilter {
            filter_id: NEXT_TEST_ID.fetch_add(1, Ordering::SeqCst),
            is_security: false,
            branches: vec![],
            conditions: vec![],
            failure_mode: FailureMode::default(),
            filter: AnyFilter::Http(Box::new(CountFilter { counter })),
            name: None,
            response_conditions: vec![],
        }
    }

    /// Build a named no-op [`PipelineFilter`] with request conditions.
    fn named_pf(name: &'static str, conditions: Vec<praxis_core::config::Condition>) -> PipelineFilter {
        PipelineFilter {
            filter_id: NEXT_TEST_ID.fetch_add(1, Ordering::SeqCst),
            is_security: false,
            branches: vec![],
            conditions,
            failure_mode: FailureMode::default(),
            filter: AnyFilter::Http(Box::new(NamedFilter(name))),
            name: None,
            response_conditions: vec![],
        }
    }

    /// Build a filter that observes early branch trace initialization.
    fn trace_probe_pf(saw_context: Arc<AtomicBool>) -> PipelineFilter {
        PipelineFilter {
            filter_id: NEXT_TEST_ID.fetch_add(1, Ordering::SeqCst),
            is_security: false,
            branches: vec![],
            conditions: vec![],
            failure_mode: FailureMode::default(),
            filter: AnyFilter::Http(Box::new(TraceContextProbe { saw_context })),
            name: None,
            response_conditions: vec![],
        }
    }

    /// Build a rejecting [`PipelineFilter`].
    fn reject_pf(status: u16) -> PipelineFilter {
        PipelineFilter {
            filter_id: NEXT_TEST_ID.fetch_add(1, Ordering::SeqCst),
            is_security: false,
            branches: vec![],
            conditions: vec![],
            failure_mode: FailureMode::default(),
            filter: AnyFilter::Http(Box::new(RejectFilter { status })),
            name: None,
            response_conditions: vec![],
        }
    }

    /// Build an erroring [`PipelineFilter`].
    fn error_pf(failure_mode: FailureMode) -> PipelineFilter {
        PipelineFilter {
            filter_id: NEXT_TEST_ID.fetch_add(1, Ordering::SeqCst),
            is_security: false,
            branches: vec![],
            conditions: vec![],
            failure_mode,
            filter: AnyFilter::Http(Box::new(ErrorFilter)),
            name: None,
            response_conditions: vec![],
        }
    }

    /// Build a stateful [`PipelineFilter`].
    fn stateful_pf(id: u64, log: &ObsLog) -> PipelineFilter {
        PipelineFilter {
            filter_id: NEXT_TEST_ID.fetch_add(1, Ordering::SeqCst),
            is_security: false,
            branches: vec![],
            conditions: vec![],
            failure_mode: FailureMode::default(),
            filter: AnyFilter::Http(Box::new(BranchStatefulFilter {
                id,
                log: Arc::clone(log),
            })),
            name: None,
            response_conditions: vec![],
        }
    }

    #[tokio::test]
    async fn streaming_terminal_response_propagates_from_branch() {
        use crate::StreamingResponseBody;

        struct NullStreamBody;

        #[async_trait::async_trait]
        impl StreamingResponseBody for NullStreamBody {
            async fn next_chunk(&mut self) -> Result<Option<Bytes>, FilterError> {
                Ok(None)
            }

            async fn suppress(&mut self) -> Result<(), FilterError> {
                Ok(())
            }

            async fn cancel(&mut self) {}
        }

        struct StreamingTerminalFilter;

        #[async_trait::async_trait]
        impl HttpFilter for StreamingTerminalFilter {
            fn name(&self) -> &'static str {
                "streaming_terminal_test"
            }

            async fn on_request(&self, _ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
                Ok(FilterAction::StreamingTerminalResponse(Box::new(
                    StreamingTerminalResponse::new(200, Box::new(NullStreamBody)),
                )))
            }
        }

        let streaming_pf = PipelineFilter {
            filter_id: 99,
            is_security: false,
            branches: vec![],
            conditions: vec![],
            failure_mode: FailureMode::default(),
            filter: AnyFilter::Http(Box::new(StreamingTerminalFilter)),
            name: None,
            response_conditions: vec![],
        };
        let branches = vec![make_branch(
            "stream_branch",
            None,
            RejoinTarget::Next,
            None,
            vec![streaming_pf],
        )];
        let req = crate::test_utils::make_request(Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let outcome = evaluate_branches(&branches, &mut ctx).await.unwrap();
        assert!(
            matches!(outcome, BranchOutcome::StreamingTerminalResponse(_)),
            "streaming terminal response should propagate from branch"
        );
    }

    /// Build a [`ResolvedBranch`] for testing.
    fn make_branch(
        name: &str,
        condition: Option<(&str, &str, &str)>,
        rejoin: RejoinTarget,
        max_iterations: Option<u32>,
        filters: Vec<PipelineFilter>,
    ) -> ResolvedBranch {
        ResolvedBranch {
            condition: condition.map(|(filter, key, value)| ResolvedBranchCondition {
                filter_name: Arc::from(filter),
                key: Arc::from(key),
                value: Arc::from(value),
            }),
            filters,
            max_iterations,
            name: Arc::from(name),
            rejoin,
        }
    }
}