dwara_core/dataplane/

split.rs

//! Service-level traffic splitting (DW-040, feature analysis section
//! 5 "Traffic"): canary releases, blue-green switches, and the sticky
//! cookie that pins a session through them.
//!
//! # The pick
//!
//! A split service dispatches each request to ONE of its target
//! upstreams by a deterministic weighted pick over the SAME FNV-1a
//! hash the balancer's consistent-hash ring uses
//! (`balance::key_hash` — the stability argument for that choice,
//! toolchain-stable hashing, applies identically here):
//!
//! - key = the sticky cookie's VALUE when the service configures
//!   [`crate::config::StickyAffinity`] and the request carries the
//!   cookie — a session lands on the same branch for every request
//!   while the weights are unchanged;
//! - key = the request id otherwise — per-request distribution whose
//!   realized ratios converge on the configured weights statistically
//!   (the issue's acceptance shape: "split ratios verified
//!   statistically").
//!
//! The slot is `hash % total_weight`, walked over the targets'
//! cumulative weights. DISPLACEMENT IS BOUNDED BY AN INVARIANT TOTAL:
//! when a weight change KEEPS the total constant, only the changed
//! share of hash space moves — 95/5 -> 90/10 moves exactly the 5% that
//! became canary, and the blue-green flip 100/0 -> 0/100 (total still
//! 100) moves exactly the stable side. A change that ALTERS the total
//! changes the modulus and therefore reshuffles EVERY key — ramp a
//! canary by RE-BALANCING the pair (95/5 -> 90/10), never by growing
//! the canary alone (95/5 -> 95/10 reshuffles all sessions). The pick
//! is STATELESS — no counters, no locks, no per-service runtime state
//! — which is what makes the blue-green switch instant: a re-publish
//! rebuilds the compiled split, and the very next request dispatches
//! by the new generation.
//!
//! # Stickiness is layered
//!
//! The cookie guarantees BRANCH affinity (which upstream). Endpoint
//! affinity within a branch comes from the branch upstream's own
//! balancer: when it runs `ip_hash`, the sticky value becomes the
//! ring key (the proxy passes it as the dispatch hash key), so the
//! session pins one endpoint through the same ketama machinery that
//! pins client IPs. With other balancers the branch pick is the
//! affinity guarantee and the endpoint is free to float — documented,
//! not hidden: ring-based endpoint pinning and round-robin spreading
//! are different tools and the operator picks per branch.
//!
//! The cookie value is an opaque affinity handle generated by the
//! gateway (time + a process counter, hex): NOT a secret, carrying no
//! identity, never trusted as one. Reusing a presented cookie is safe
//! precisely because its only power is choosing among upstreams the
//! operator already configured — a client "forging" one can only
//! pick a branch it could have landed on anyway.
//!
//! One documented edge: a session whose FIRST request is a response
//! cache HIT is served without a dispatch, so no cookie is minted
//! until its first cache MISS — while hits last there is nothing to
//! pin (no upstream was contacted), and the miss mints exactly as the
//! uncached path would.

use std::sync::Arc;

use crate::dataplane::balance::key_hash;
use crate::dataplane::upstream::UpstreamHandle;

/// One compiled target of a service split (DW-040).
pub struct SplitEntry {
    handle: Arc<UpstreamHandle>,
    weight: u64,
    /// Inclusive upper bound of this entry's hash-slot range (the
    /// cumulative weight through this entry); the entry owns slots
    /// `(bound - weight, bound]`.
    bound: u64,
}

/// The compiled weighted split of one service (DW-040): built per
/// config generation by the upstream registry, held by the dataplane
/// dispatch path. Immutable once built; the pick is stateless.
pub struct ServiceSplit {
    entries: Vec<SplitEntry>,
    /// Sum of all weights (the hash modulus; validation guarantees > 0).
    total: u64,
}

impl ServiceSplit {
    /// Compile from validated config targets plus the registry's
    /// handles. Targets with weight 0 are kept (their handles exist,
    /// their slots are empty — the parked side of a blue-green pair),
    /// preserving index stability across weight flips.
    pub fn new(targets: &[(Arc<UpstreamHandle>, u64)]) -> Self {
        let mut entries = Vec::with_capacity(targets.len());
        let mut total = 0u64;
        for (handle, weight) in targets {
            total += weight;
            entries.push(SplitEntry {
                handle: Arc::clone(handle),
                weight: *weight,
                bound: total,
            });
        }
        ServiceSplit { entries, total }
    }

    /// The total weight (0 only for an unvalidated empty split).
    pub fn total_weight(&self) -> u64 {
        self.total
    }

    /// Pick the target upstream for one dispatch by the deterministic
    /// weighted hash of `key` (the sticky cookie value or the request
    /// id — see the module docs). The handle is the branch; its own
    /// balancer (and the hash key threaded to it) picks the endpoint.
    pub fn pick(&self, key: &str) -> &Arc<UpstreamHandle> {
        let slot = key_hash(key) % self.total;
        // The entry whose cumulative bound first exceeds the slot.
        // Linear over at most 8 entries — cheaper than any index.
        for e in &self.entries {
            if slot < e.bound {
                return &e.handle;
            }
        }
        // Unreachable: the last bound is the total, and slot < total.
        // Guarded fallthrough keeps a hostile zero-weight total from
        // panicking: dispatch to the first entry rather than crash.
        &self.entries[0].handle
    }

    /// DW-091: pick the target upstream AND its index (0 = baseline,
    /// 1 = canary for a 2-target canary split). The auto-canary
    /// controller needs the index to record the outcome on the
    /// correct side. Same hash as [`Self::pick`].
    pub fn pick_with_index(&self, key: &str) -> (&Arc<UpstreamHandle>, usize) {
        let slot = key_hash(key) % self.total;
        for (i, e) in self.entries.iter().enumerate() {
            if slot < e.bound {
                return (&e.handle, i);
            }
        }
        (&self.entries[0].handle, 0)
    }

    /// The entry weights in config order (diagnostics and tests).
    pub fn weights(&self) -> Vec<u64> {
        self.entries.iter().map(|e| e.weight).collect()
    }
}

/// Read one cookie's value from a request's `Cookie` headers (the
/// first match across all header lines wins; `Cookie` is a
/// semicolon-separated `name=value` list per RFC 6265). Returns the
/// trimmed value; a malformed pair cannot match a validated cookie
/// name and is skipped silently (the gateway then mints a fresh
/// handle — the safe direction: a bad cookie re-sessions rather than
/// mis-routing).
pub fn read_cookie(headers: &hyper::HeaderMap, name: &str) -> Option<String> {
    for value in headers.get_all(hyper::header::COOKIE) {
        let Ok(line) = value.to_str() else { continue };
        for pair in line.split(';') {
            let pair = pair.trim();
            if let Some((k, v)) = pair.split_once('=') {
                if k.trim() == name {
                    let v = v.trim();
                    if !v.is_empty() {
                        return Some(v.to_string());
                    }
                }
            }
        }
    }
    None
}

/// One process-wide affinity-handle counter (monotonic; combined with
/// the wall clock it makes handles unique per mint).
static AFFINITY_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

/// Mint a fresh affinity handle: `hex(unix ms) hex(counter)` — 24
/// printable hex characters. Opaque, unique per process, NOT a secret
/// and never treated as one (its only power is choosing among
/// upstreams the operator already configured).
pub fn mint_affinity_id() -> String {
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0);
    let n = AFFINITY_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    format!("{now:012x}{n:06x}")
}