praxis-proxy-filter 0.7.2

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

//! Cluster name extraction from filter pipeline capabilities.
//!
//! Collects the set of cluster names declared by cluster-selecting
//! filters (routers, endpoint selectors) and load-balancer filters.
//! The ordering checks in [`checks`] compare these two sets to detect
//! misaligned or orphaned cluster references at build time.
//!
//! [`checks`]: super::checks

use std::collections::HashSet;

use super::filter::PipelineFilter;
#[cfg(feature = "upstream-binding")]
use crate::any_filter::AnyFilter;

// -----------------------------------------------------------------------------
// Cluster Extraction
// -----------------------------------------------------------------------------

/// Cluster selectors declare every cluster name they may assign.
///
/// Recurses into branch sub-chains: a cluster selected inside a branch is
/// assigned to `ctx.cluster` when the branch runs, so it must be checked
/// against the load balancers just like a top-level selection. Otherwise a
/// branch selecting an undefined cluster passes the build and 502s at request
/// time.
pub(super) fn extract_selected_clusters(filters: &[PipelineFilter]) -> HashSet<String> {
    let mut out = HashSet::new();
    for pf in filters {
        out.extend(pf.filter.selected_clusters());
        for branch in &pf.branches {
            out.extend(extract_selected_clusters(&branch.filters));
        }
    }
    out
}

/// Load-balancers declare the cluster names they can consume.
///
/// Recurses into branch sub-chains for the same reason as
/// [`extract_selected_clusters`].
pub(super) fn extract_lb_clusters(filters: &[PipelineFilter]) -> HashSet<String> {
    let mut out = HashSet::new();
    for pf in filters {
        out.extend(pf.filter.load_balancer_clusters());
        for branch in &pf.branches {
            out.extend(extract_lb_clusters(&branch.filters));
        }
    }
    out
}

/// Cluster names selected by this level's filters only (no branch recursion).
///
/// Branch-level demands are checked per branch with that branch's own
/// availability; see `check_misaligned_clusters`.
pub(super) fn level_selected_clusters(filters: &[PipelineFilter]) -> HashSet<String> {
    filters.iter().flat_map(|pf| pf.filter.selected_clusters()).collect()
}

/// Cluster names provided by load balancers guaranteed to run for every
/// request that reaches this level.
///
/// This is this level's own load balancers plus those inside *unconditional*
/// branches (`condition: None`) hung off *unconditional* host filters (no
/// filter conditions), recursively. Such a branch always fires and its filters
/// run against the same `ctx`, so a load balancer inside it sets `ctx.upstream`
/// for the enclosing selection exactly like a top-level one, an inlined chain
/// in all but syntax. A *conditional* branch (or one on a conditional host) is
/// excluded: it may not run, so its load balancer cannot be relied on to serve
/// an enclosing selection.
///
/// The branch's rejoin target is irrelevant here: whether it rejoins `Next` or
/// `Terminal`, the branch's own filters still run and set `ctx.upstream` before
/// control leaves the branch; rejoin only governs which *later* top-level
/// filters run.
pub(super) fn reachable_lb_clusters(filters: &[PipelineFilter]) -> HashSet<String> {
    let mut out = HashSet::new();
    for pf in filters {
        out.extend(pf.filter.load_balancer_clusters());
        if pf.conditions.is_empty() {
            for branch in &pf.branches {
                if branch.condition.is_none() {
                    out.extend(reachable_lb_clusters(&branch.filters));
                }
            }
        }
    }
    out
}

/// Cluster names that a binding filter (`binds_upstream`) may publish as the
/// logical upstream, recursing into branch sub-chains.
///
/// The binding router records only a cluster name in the logical binding; these
/// are exactly the names a bound-consuming load balancer must be able to
/// resolve. Pipeline validation compares this set against reachable consumer
/// coverage so every bindable cluster is served on its request path.
#[cfg(feature = "upstream-binding")]
pub(super) fn bindable_clusters(filters: &[PipelineFilter]) -> HashSet<String> {
    let mut out = HashSet::new();
    for pf in filters {
        if let AnyFilter::Http(f) = &pf.filter
            && f.binds_upstream()
        {
            out.extend(f.selected_clusters());
        }
        for branch in &pf.branches {
            out.extend(bindable_clusters(&branch.filters));
        }
    }
    out
}

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

#[cfg(test)]
#[expect(clippy::allow_attributes, reason = "blanket test suppressions")]
#[allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::indexing_slicing,
    clippy::panic,
    reason = "tests"
)]
mod tests {
    use super::*;
    #[cfg(feature = "upstream-binding")]
    use crate::pipeline::test_filters::binding_router;
    use crate::pipeline::test_filters::{lb_filter, noop_filter, selector_filter};

    #[test]
    fn extracts_selected_clusters() {
        let filters = vec![selector_filter("router", &["web", "api"])];
        let clusters = extract_selected_clusters(&filters);
        assert_eq!(clusters.len(), 2, "should extract two clusters");
        assert!(clusters.contains("web"), "should contain 'web'");
        assert!(clusters.contains("api"), "should contain 'api'");
    }

    #[test]
    fn extracts_lb_clusters() {
        let filters = vec![lb_filter(&["web", "api"])];
        let clusters = extract_lb_clusters(&filters);
        assert_eq!(clusters.len(), 2, "should extract two clusters");
        assert!(clusters.contains("web"), "should contain 'web'");
        assert!(clusters.contains("api"), "should contain 'api'");
    }

    #[test]
    fn skips_non_cluster_selecting_entries() {
        let filters = vec![noop_filter("ip_acl")];
        let clusters = extract_selected_clusters(&filters);
        assert!(
            clusters.is_empty(),
            "non-cluster-selecting entries should yield no clusters"
        );
    }

    #[test]
    fn merges_selected_clusters_from_multiple_filters() {
        let filters = vec![
            selector_filter("router", &["web"]),
            selector_filter("custom_selector", &["weather-backend"]),
        ];
        let clusters = extract_selected_clusters(&filters);
        assert_eq!(clusters.len(), 2, "should merge selected clusters");
        assert!(clusters.contains("web"), "should contain router cluster");
        assert!(
            clusters.contains("weather-backend"),
            "should contain custom selector cluster"
        );
    }

    #[test]
    fn skips_non_load_balancer_entries() {
        let filters = vec![selector_filter("router", &["web"])];
        let clusters = extract_lb_clusters(&filters);
        assert!(clusters.is_empty(), "non-LB entries should yield no clusters");
    }

    #[test]
    fn deduplicates_selected_clusters() {
        let filters = vec![
            selector_filter("router", &["web"]),
            selector_filter("custom_selector", &["web"]),
        ];
        let clusters = extract_selected_clusters(&filters);
        assert_eq!(clusters.len(), 1, "duplicate cluster names should be deduplicated");
        assert!(clusters.contains("web"), "should contain 'web'");
    }

    #[test]
    fn recurses_into_branch_subchains() {
        use std::sync::Arc;

        use crate::pipeline::branch::{RejoinTarget, ResolvedBranch};

        let mut host = noop_filter("headers");
        host.branches = vec![ResolvedBranch {
            condition: None,
            filters: vec![selector_filter("router", &["branch-cluster"])],
            max_iterations: None,
            name: Arc::from("br"),
            rejoin: RejoinTarget::Terminal,
        }];
        let selected = extract_selected_clusters(&[host]);
        assert!(
            selected.contains("branch-cluster"),
            "a cluster selected inside a branch sub-chain must be collected"
        );
    }

    #[test]
    fn empty_entries_yields_empty() {
        let filters = vec![];
        assert!(
            extract_selected_clusters(&filters).is_empty(),
            "empty input should yield empty set"
        );
        assert!(
            extract_lb_clusters(&filters).is_empty(),
            "empty input should yield empty set"
        );
    }

    use std::sync::Arc;

    use praxis_core::config::{Condition, ConditionMatch};

    use crate::pipeline::branch::{RejoinTarget, ResolvedBranch, ResolvedBranchCondition};

    /// Build a host filter carrying one branch (condition controls reachability).
    fn host_with(condition: Option<ResolvedBranchCondition>, branch_filters: Vec<PipelineFilter>) -> PipelineFilter {
        let mut host = noop_filter("headers");
        host.branches = vec![ResolvedBranch {
            condition,
            filters: branch_filters,
            max_iterations: None,
            name: Arc::from("br"),
            rejoin: RejoinTarget::Next,
        }];
        host
    }

    fn cond() -> ResolvedBranchCondition {
        ResolvedBranchCondition {
            filter_name: Arc::from("classifier"),
            key: Arc::from("kind"),
            value: Arc::from("premium"),
        }
    }

    #[test]
    fn reachable_includes_this_levels_load_balancers() {
        let filters = vec![lb_filter(&["web", "api"])];
        let clusters = reachable_lb_clusters(&filters);
        assert!(clusters.contains("web") && clusters.contains("api"));
    }

    #[test]
    fn reachable_folds_unconditional_branch_lb() {
        let filters = vec![host_with(None, vec![lb_filter(&["x"])])];
        assert!(
            reachable_lb_clusters(&filters).contains("x"),
            "unconditional branch LB must be reachable"
        );
    }

    #[test]
    fn reachable_excludes_conditional_branch_lb() {
        let filters = vec![host_with(Some(cond()), vec![lb_filter(&["x"])])];
        assert!(
            !reachable_lb_clusters(&filters).contains("x"),
            "conditional branch LB must not be reachable"
        );
    }

    #[test]
    fn reachable_excludes_branch_lb_on_conditional_host() {
        let mut host = noop_filter("headers");
        host.conditions = vec![Condition::When(ConditionMatch {
            grpc: None,
            path: None,
            path_prefix: Some("/x".to_owned()),
            methods: None,
            headers: None,
            bound_upstream: None,
            selected_upstream: None,
        })];
        host.branches = vec![ResolvedBranch {
            condition: None,
            filters: vec![lb_filter(&["x"])],
            max_iterations: None,
            name: Arc::from("br"),
            rejoin: RejoinTarget::Next,
        }];
        assert!(
            !reachable_lb_clusters(&[host]).contains("x"),
            "a branch on a conditional host must not be reachable"
        );
    }

    #[test]
    fn reachable_folds_nested_unconditional_branches() {
        let inner = host_with(None, vec![lb_filter(&["deep"])]);
        let outer = host_with(None, vec![inner]);
        assert!(
            reachable_lb_clusters(&[outer]).contains("deep"),
            "nested unconditional branch LBs must fold up"
        );
    }

    #[test]
    fn reachable_stops_folding_at_conditional_nesting() {
        let inner = host_with(Some(cond()), vec![lb_filter(&["deep"])]);
        let outer = host_with(None, vec![inner]);
        assert!(
            !reachable_lb_clusters(&[outer]).contains("deep"),
            "a conditional nested branch stops the reachability fold"
        );
    }

    #[cfg(feature = "upstream-binding")]
    #[test]
    fn bindable_clusters_collects_only_binding_publishers() {
        let filters = vec![
            selector_filter("ordinary_router", &["ordinary"]),
            binding_router(&["bound-a", "bound-b"]),
            lb_filter(&["endpoint-only"]),
        ];

        let clusters = bindable_clusters(&filters);

        assert_eq!(
            clusters.len(),
            2,
            "only the binding router's clusters are bindable: {clusters:?}"
        );
        assert!(
            clusters.contains("bound-a"),
            "the binding router's first cluster is bindable: {clusters:?}"
        );
        assert!(
            clusters.contains("bound-b"),
            "the binding router's second cluster is bindable: {clusters:?}"
        );
        assert!(
            !clusters.contains("ordinary"),
            "an ordinary router's cluster is not bindable: {clusters:?}"
        );
        assert!(
            !clusters.contains("endpoint-only"),
            "a load balancer's cluster is not bindable: {clusters:?}"
        );
    }

    #[cfg(feature = "upstream-binding")]
    #[test]
    fn bindable_clusters_recurses_into_branches() {
        let filters = vec![host_with(None, vec![binding_router(&["nested"])])];

        assert_eq!(
            bindable_clusters(&filters),
            HashSet::from(["nested".to_owned()]),
            "a binding router inside a branch should contribute its bindable cluster"
        );
    }
}