tidaldb/tidal-server/src/cluster/forward.rs

//! Cross-process HTTP forwarding, broadcast, and the internal-propagation
//! marker shared by the multi-process region routes (m8p10 task 03).
//!
//! The multi-process cluster is one coherent HTTP surface: any node accepts any
//! runbook operation and forwards it to the node that owns it. Three primitives
//! make that work, and they all live here so the routes cannot drift on the
//! loop-prevention contract:
//!
//! * the **internal-propagation marker** ([`INTERNAL_MARKER`]): a forwarded or
//!   broadcast request carries `x-tidal-internal: 1`. A handler that sees the
//!   marker applies the op LOCALLY ONLY — it never re-forwards or re-broadcasts.
//!   This is what terminates every forward and every fan-out (no infinite loops,
//!   no broadcast storms). The marker is an internal trust signal between sibling
//!   processes that already share one `TIDAL_API_KEY`; it is NOT an
//!   authentication bypass (the bearer middleware still runs first).
//! * **forward** ([`forward_json`]): relay one request body verbatim to a single
//!   peer (the leader, or a region owner) and hand its status + body straight
//!   back to the client. The `Authorization` header passes through verbatim so
//!   the peer's bearer middleware accepts it.
//! * **broadcast** ([`broadcast_marked`]): fan one body out to every peer
//!   concurrently with the marker set, collecting per-peer success/failure so the
//!   handler can report `{"replicated_to": n, "failed": [..]}` honestly.
//!
//! The shared [`reqwest::Client`] lives on [`ShardReplica`] (connection
//! pooling across requests); its timeouts are tight (connect 1s, request 5s) so
//! a dead peer degrades a forward into a clean 503/partial result instead of
//! hanging an axum worker. Operation-specific budgets override the client
//! default per request: forwards get the wider [`FORWARD_REQUEST_TIMEOUT`]
//! (the peer does real work — fsync, fan-out, merge — before answering), while
//! status aggregation uses the tighter [`STATUS_PEER_TIMEOUT`] because it
//! queries every peer on every poll.

use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

use axum::{
    Json,
    http::{HeaderMap, HeaderName, HeaderValue, StatusCode, header::AUTHORIZATION},
    response::{IntoResponse, Response},
};
use serde::Serialize;

/// Whether inter-node HTTP links use TLS (m11p7). A process is exactly ONE
/// cluster node with one TLS posture, so this process-global is the natural home
/// for the scheme decision — it avoids threading `https`-vs-`http` through every
/// `peer_url` call site and the membership view's address tables. Set once at
/// node construction (and at seed-join boot) when the cluster has `grpc_tls`;
/// read by [`peer_url`]. Default `false` (plaintext) keeps every existing
/// no-TLS test and deployment byte-for-byte unchanged.
static INTER_NODE_HTTPS: AtomicBool = AtomicBool::new(false);

/// Declare this node's inter-node HTTP scheme (m11p7): `true` ⇒ forwards,
/// broadcasts, scatter fetches, status aggregation, and seed-join dial `https://`.
/// Idempotent; called at node/seed-join construction when `grpc_tls` is present.
pub fn set_inter_node_https(on: bool) {
    INTER_NODE_HTTPS.store(on, Ordering::Relaxed);
}

/// Whether inter-node HTTP uses TLS (the scheme [`peer_url`] emits for a bare
/// `host:port`).
#[must_use]
pub fn inter_node_https() -> bool {
    INTER_NODE_HTTPS.load(Ordering::Relaxed)
}

/// Header name for the internal-propagation marker.
///
/// A request carrying `x-tidal-internal: 1` was generated by a sibling node
/// (a forward or a broadcast), so the receiving handler applies it LOCALLY and
/// never re-forwards/re-broadcasts. See the module docs.
pub const INTERNAL_MARKER: &str = "x-tidal-internal";

/// The marker value siblings set.
pub const INTERNAL_MARKER_VALUE: &str = "1";

/// Request header overriding the deployment's write-acknowledgment mode
/// (m11p3): `leader` or `quorum`. Forwarded verbatim with the write so the
/// leader honors the CALLER's choice, not the gateway's default.
pub const ACK_HEADER: &str = "x-tidal-ack";

/// Response header carrying a cluster write's assigned replicated-log seqno
/// (m11p3). Relayed verbatim on forwarded writes.
pub const SEQ_HEADER: &str = "x-tidal-seq";

/// Response header (value `1`) marking a 2xx write whose record was
/// dedup-suppressed by the WAL content-hash window (m11p3): an identical
/// record is already durable, no new log entry was created, so the response
/// carries this instead of [`SEQ_HEADER`]. Relayed verbatim on forwarded
/// writes so a gateway client can tell "deduplicated" from "no seqno
/// surface" without consulting the runbook.
pub const DEDUP_HEADER: &str = "x-tidal-deduplicated";

/// The marker value set on [`DEDUP_HEADER`].
pub const DEDUP_HEADER_VALUE: &str = "1";

/// Connect timeout for every forwarded/broadcast/aggregation request. A peer
/// whose TCP connect does not complete in 1s is treated as unreachable.
pub const CONNECT_TIMEOUT: Duration = Duration::from_secs(1);

/// Overall request timeout for a forwarded write/read (generous enough for a
/// leader-side WAL fsync + ship, tight enough that a hung peer never pins an
/// axum worker indefinitely).
pub const REQUEST_TIMEOUT: Duration = Duration::from_secs(5);

/// Per-request budget for [`forward_json`] (write forwards, the reconcile
/// snapshot exchange). Wider than the client-level [`REQUEST_TIMEOUT`] because
/// the peer's response covers REAL work, not just a network hop: a forwarded
/// write includes the leader's WAL fsync PLUS its own 2s-per-peer
/// item/embedding broadcast, and a reconcile exchange includes the remote's
/// CRDT merge+apply. Budget: connect (1s) + leader-side apply/fsync +
/// leader-side fan-out (2s) + cross-region RTT, with headroom for a congested
/// disk. A peer that exceeds this degrades into the caller's clean 503 — it
/// never pins an axum worker indefinitely.
pub const FORWARD_REQUEST_TIMEOUT: Duration = Duration::from_secs(15);

/// Per-peer overall budget for `/cluster/status` aggregation. Tighter than
/// [`REQUEST_TIMEOUT`] because status is polled frequently and an unreachable
/// peer must surface as `reachable: false` fast, not stall the whole view.
pub const STATUS_PEER_TIMEOUT: Duration = Duration::from_millis(500);

/// Per-peer timeout for the item/embedding leader broadcast and the promote
/// fan-out. A partitioned peer that does not answer in 2s is reported failed in
/// the response body (and catches up on the operator's next re-broadcast / heal)
/// rather than blocking the originating request.
pub const BROADCAST_PEER_TIMEOUT: Duration = Duration::from_secs(2);

/// Build the shared forwarding client with the standard connect/request
/// timeouts. One per node (held on [`ShardReplica`]) so connections pool
/// across requests.
///
/// When `ca_pem` is `Some` (m11p7 inter-node TLS), the cluster CA is added as a
/// trust anchor so `https://` forwards to a peer's cluster-signed server cert
/// verify. `None` ⇒ plaintext forwards (the trusted-loopback opt-out).
///
/// # Errors
///
/// Returns the underlying [`reqwest::Error`] if the client cannot be built
/// (e.g. the TLS backend fails to initialize or the CA PEM is invalid). The
/// caller treats this as a fatal startup error.
pub fn build_client(ca_pem: Option<&[u8]>) -> reqwest::Result<reqwest::Client> {
    let mut builder = reqwest::Client::builder()
        .connect_timeout(CONNECT_TIMEOUT)
        .timeout(REQUEST_TIMEOUT);
    if let Some(pem) = ca_pem {
        builder = builder.add_root_certificate(reqwest::Certificate::from_pem(pem)?);
    }
    builder.build()
}

/// True iff the request carries the internal-propagation marker.
///
/// A handler that sees this applies the op locally only — it does NOT forward or
/// broadcast (loop prevention). The value must be exactly `"1"`; any other value
/// is treated as absent so a stray header cannot disable forwarding.
#[must_use]
pub fn is_internal(headers: &HeaderMap) -> bool {
    headers
        .get(INTERNAL_MARKER)
        .and_then(|v| v.to_str().ok())
        .is_some_and(|v| v == INTERNAL_MARKER_VALUE)
}

/// Extract the verbatim `Authorization` header value, if present, so a forward
/// can pass the caller's bearer token straight through to the peer.
#[must_use]
pub fn forwarded_auth(headers: &HeaderMap) -> Option<String> {
    headers
        .get(AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
        .map(str::to_string)
}

/// The outcome of forwarding one request to a single peer: the peer's status
/// code and raw JSON body, ready to hand straight back to the original client.
pub struct ForwardedResponse {
    /// The peer's `x-tidal-seq` response header (a cluster write's assigned
    /// replicated-log seqno, m11p3), relayed verbatim when present.
    pub seq: Option<String>,
    /// Whether the peer marked the write dedup-suppressed
    /// ([`DEDUP_HEADER`]), relayed verbatim.
    pub deduplicated: bool,
    /// The peer's HTTP status, relayed verbatim.
    pub status: StatusCode,
    /// The peer's response body parsed as JSON (or a synthesized object if the
    /// body was empty / not JSON, so the relay always returns a JSON value).
    pub body: serde_json::Value,
}

/// Rebuild an axum [`Response`] from a [`ForwardedResponse`], relaying the
/// replicated-log verdict headers (`x-tidal-seq` / `x-tidal-deduplicated`) the
/// peer set. This is the ONE place a forwarded write's quorum/dedup verdict is
/// re-attached to the client-facing response, so the follower→leader hop
/// (`forward_write`) and the cross-shard gateway hop (`forward_to_group_node`)
/// cannot drift on the relay contract.
#[must_use]
pub fn relay_forwarded(resp: ForwardedResponse) -> Response {
    let mut response = (resp.status, Json(resp.body)).into_response();
    if let Some(seq) = resp.seq
        && let Ok(value) = HeaderValue::from_str(&seq)
    {
        response
            .headers_mut()
            .insert(HeaderName::from_static(SEQ_HEADER), value);
    }
    if resp.deduplicated {
        response.headers_mut().insert(
            HeaderName::from_static(DEDUP_HEADER),
            HeaderValue::from_static(DEDUP_HEADER_VALUE),
        );
    }
    response
}

/// The `x-tidal-ack` override (m11p3) as a forward passthrough entry, if the
/// caller set one — so a forwarded write reaches the leader with the CALLER's
/// ack mode, not the gateway's default. Empty when absent.
#[must_use]
pub fn ack_passthrough(headers: &HeaderMap) -> Vec<(&'static str, String)> {
    headers
        .get(ACK_HEADER)
        .and_then(|v| v.to_str().ok())
        .map(|v| vec![(ACK_HEADER, v.to_owned())])
        .unwrap_or_default()
}

/// Forward a JSON request to one peer and relay its status + body back.
///
/// `auth` is the caller's verbatim `Authorization` header (passed through so the
/// peer's bearer middleware accepts the forward); `internal` sets the
/// propagation marker so the peer applies the op locally and never re-forwards.
///
/// A connect/timeout/transport failure is returned as `Err(connect_error_string)`
/// so the caller can degrade it into a 503 naming the unreachable peer — the
/// forward never panics and never silently drops the request.
///
/// # Errors
///
/// Returns the stringified transport error when the peer cannot be reached or
/// the exchange fails before a status is received.
pub async fn forward_json<B: Serialize + Sync + ?Sized>(
    client: &reqwest::Client,
    url: &str,
    body: &B,
    auth: Option<&str>,
    internal: bool,
) -> Result<ForwardedResponse, String> {
    forward_json_with_headers(client, url, body, auth, internal, &[]).await
}

/// [`forward_json`] with extra request headers passed through verbatim
/// (m11p3: the caller's `x-tidal-ack` override must reach the leader).
pub async fn forward_json_with_headers<B: Serialize + Sync + ?Sized>(
    client: &reqwest::Client,
    url: &str,
    body: &B,
    auth: Option<&str>,
    internal: bool,
    passthrough: &[(&'static str, String)],
) -> Result<ForwardedResponse, String> {
    // Per-request override of the client-level REQUEST_TIMEOUT: a forward's
    // response covers the peer's real work (WAL fsync, a quorum wait, a
    // CRDT merge), not just a hop. See FORWARD_REQUEST_TIMEOUT.
    let mut req = client.post(url).timeout(FORWARD_REQUEST_TIMEOUT).json(body);
    if let Some(auth) = auth {
        req = req.header(AUTHORIZATION, auth);
    }
    if internal {
        req = req.header(INTERNAL_MARKER, INTERNAL_MARKER_VALUE);
    }
    for (name, value) in passthrough {
        req = req.header(*name, value);
    }
    let resp = req.send().await.map_err(|e| e.to_string())?;
    let status = resp.status();
    // Capture the seq/dedup headers BEFORE consuming the body: a forwarded
    // write's caller relays them so the client sees the leader's verdict.
    let seq = resp
        .headers()
        .get(SEQ_HEADER)
        .and_then(|v| v.to_str().ok())
        .map(str::to_owned);
    let deduplicated = resp
        .headers()
        .get(DEDUP_HEADER)
        .and_then(|v| v.to_str().ok())
        .is_some_and(|v| v == DEDUP_HEADER_VALUE);
    // Relay the body verbatim where possible; an empty/non-JSON body becomes a
    // null so the relay always yields a JSON value the caller can wrap.
    let bytes = resp.bytes().await.map_err(|e| e.to_string())?;
    let body = if bytes.is_empty() {
        serde_json::Value::Null
    } else {
        serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null)
    };
    Ok(ForwardedResponse {
        seq,
        deduplicated,
        status,
        body,
    })
}

/// Fan a JSON body out to every `(name, url)` peer concurrently with the
/// internal marker set, returning the names of peers that succeeded
/// (2xx) and the names that failed (transport error or non-2xx status).
///
/// Best-effort by contract: a failed peer is WARN-logged and reported in
/// `failed`, never fatal to the originating request. Each peer is bounded by
/// `per_peer_timeout` so one slow peer cannot stall the fan-out.
pub async fn broadcast_marked<B: Serialize + Sync>(
    client: &reqwest::Client,
    peers: Vec<(String, String)>,
    path: &str,
    body: &B,
    auth: Option<&str>,
    node_token: Option<&str>,
    per_peer_timeout: Duration,
) -> BroadcastOutcome {
    // Serialize once; every peer gets the identical body.
    let payload = serde_json::to_value(body).unwrap_or(serde_json::Value::Null);
    let futures = peers.into_iter().map(|(name, http_addr)| {
        let client = client.clone();
        let payload = payload.clone();
        let auth = auth.map(str::to_string);
        let node_token = node_token.map(str::to_string);
        let path = path.to_string();
        async move {
            let url = peer_url(&http_addr, &path);
            let mut req = client
                .post(&url)
                .timeout(per_peer_timeout)
                .header(INTERNAL_MARKER, INTERNAL_MARKER_VALUE)
                .json(&payload);
            if let Some(auth) = auth {
                req = req.header(AUTHORIZATION, auth);
            }
            // m11p7: the per-node token proves this broadcast originates from a
            // verified sibling, so the receiver's marker guard honors the marker.
            if let Some(token) = node_token {
                req = req.header(super::security::NODE_TOKEN_HEADER, token);
            }
            match req.send().await {
                Ok(resp) if resp.status().is_success() => (name, true),
                Ok(resp) => {
                    tracing::warn!(
                        peer = %name,
                        status = %resp.status(),
                        path = %path,
                        "cluster broadcast peer returned non-success status"
                    );
                    (name, false)
                }
                Err(e) => {
                    tracing::warn!(
                        peer = %name,
                        path = %path,
                        error = %e,
                        "cluster broadcast peer unreachable"
                    );
                    (name, false)
                }
            }
        }
    });
    let results = futures_util::future::join_all(futures).await;
    let mut acked = Vec::new();
    let mut failed = Vec::new();
    for (name, ok) in results {
        if ok {
            acked.push(name);
        } else {
            failed.push(name);
        }
    }
    BroadcastOutcome { acked, failed }
}

/// Result of a marked broadcast/fan-out: which peers acknowledged and which
/// failed (by region name).
pub struct BroadcastOutcome {
    /// Peers that returned a 2xx for the marked request.
    pub acked: Vec<String>,
    /// Peers that were unreachable or returned a non-2xx status.
    pub failed: Vec<String>,
}

/// Join a peer's HTTP base address and a path into a full URL.
///
/// The topology `http_addr` is a bare `host:port`. The scheme is `https://` when
/// the node runs inter-node TLS ([`set_inter_node_https`], m11p7) and `http://`
/// otherwise (the trusted-loopback/VPC opt-out). An address that already carries
/// a scheme is honored verbatim (never doubled).
#[must_use]
pub fn peer_url(http_addr: &str, path: &str) -> String {
    let base = http_addr.trim_end_matches('/');
    let scheme = if base.starts_with("http://") || base.starts_with("https://") {
        ""
    } else if inter_node_https() {
        "https://"
    } else {
        "http://"
    };
    let path = if path.starts_with('/') {
        path.to_string()
    } else {
        format!("/{path}")
    };
    format!("{scheme}{base}{path}")
}

#[cfg(test)]
mod tests {
    use axum::http::HeaderValue;

    use super::*;

    #[test]
    fn internal_marker_detected_only_for_exact_value() {
        let mut headers = HeaderMap::new();
        assert!(!is_internal(&headers), "absent marker is not internal");
        headers.insert(INTERNAL_MARKER, HeaderValue::from_static("1"));
        assert!(is_internal(&headers), "x-tidal-internal: 1 is internal");
        headers.insert(INTERNAL_MARKER, HeaderValue::from_static("0"));
        assert!(!is_internal(&headers), "any other value is not internal");
        headers.insert(INTERNAL_MARKER, HeaderValue::from_static("true"));
        assert!(!is_internal(&headers), "only the literal 1 counts");
    }

    #[test]
    fn forwarded_auth_passes_through_verbatim() {
        let mut headers = HeaderMap::new();
        assert_eq!(forwarded_auth(&headers), None);
        headers.insert(AUTHORIZATION, HeaderValue::from_static("Bearer secret"));
        assert_eq!(forwarded_auth(&headers).as_deref(), Some("Bearer secret"));
    }

    #[test]
    fn peer_url_joins_bare_host_port() {
        assert_eq!(
            peer_url("127.0.0.1:9500", "/signals"),
            "http://127.0.0.1:9500/signals"
        );
        assert_eq!(
            peer_url("127.0.0.1:9500/", "signals"),
            "http://127.0.0.1:9500/signals"
        );
        assert_eq!(
            peer_url("http://10.0.0.1:80", "/cluster/status/local"),
            "http://10.0.0.1:80/cluster/status/local"
        );
    }

    #[tokio::test]
    async fn join_all_preserves_order_and_concurrency() {
        // Futures that resolve after a varying number of yields still come back
        // in index order (the property `broadcast_marked` relies on to pair each
        // result with the right peer name). One async fn ⇒ one future type, so a
        // homogeneous `Vec` is well-formed.
        async fn yield_then(value: u32, yields: u32) -> u32 {
            for _ in 0..yields {
                tokio::task::yield_now().await;
            }
            value
        }
        let futs = vec![yield_then(1, 1), yield_then(2, 0), yield_then(3, 2)];
        let out = futures_util::future::join_all(futs).await;
        assert_eq!(out, vec![1, 2, 3]);
    }
}