rws_gateway_api/gateway_api/mesh/
config.rs

1//! The controller's mesh configuration endpoint: `GET /mesh/v1/snapshot`
2//! answers the current [`MeshSnapshot`] as JSON with its version in
3//! `X-Rws-Mesh-Version`; with `?version=<n>` equal to the current one it
4//! waits (up to [`LONG_POLL`]) for the next. Every controller replica serves
5//! it from its own watches.
6//!
7//! Stage 1 serves it without authentication: it holds routes, Service
8//! addresses and endpoint addresses — what every pod can already learn from
9//! DNS and traffic — and nothing secret.
10
11use std::sync::{Arc, Mutex};
12use std::time::Duration;
13
14use crate::application::Application;
15use crate::request::Request;
16use crate::response::{Response, STATUS_CODE_REASON_PHRASE};
17use crate::server::ConnectionInfo;
18
19use super::MeshSnapshot;
20
21/// Where the endpoint answers.
22pub const PATH: &str = "/mesh/v1/snapshot";
23/// The version header.
24pub const VERSION_HEADER: &str = "X-Rws-Mesh-Version";
25/// The default port of the endpoint.
26pub const DEFAULT_PORT: u16 = 15010;
27/// How long a request for a newer version waits.
28pub const LONG_POLL: Duration = Duration::from_secs(30);
29
30/// The snapshot being served, and a signal when it changes.
31#[derive(Clone)]
32pub struct Publisher {
33    current: Arc<Mutex<(u64, Arc<String>)>>,
34    changed: tokio::sync::watch::Sender<u64>,
35}
36
37impl Default for Publisher {
38    fn default() -> Self {
39        Publisher { current: Arc::new(Mutex::new((0, Arc::new(String::new())))), changed: tokio::sync::watch::channel(0).0 }
40    }
41}
42
43impl Publisher {
44    pub fn new() -> Publisher {
45        Publisher::default()
46    }
47
48    /// Serve `mesh` from now on. The version is a hash of the content, so
49    /// every controller replica gives the same content the same version
50    /// (sidecars reach them through one Service); an unchanged snapshot
51    /// wakes no sidecar.
52    pub fn publish(&self, mesh: &MeshSnapshot) {
53        let Ok(body) = serde_json::to_string(mesh) else { return };
54        let version = content_version(&body);
55        let mut current = self.current.lock().unwrap_or_else(|e| e.into_inner());
56        if current.0 == version {
57            return;
58        }
59        *current = (version, Arc::new(body));
60        self.changed.send_replace(version);
61    }
62
63    /// The version and JSON being served (version 0: nothing yet).
64    pub fn current(&self) -> (u64, Arc<String>) {
65        self.current.lock().unwrap_or_else(|e| e.into_inner()).clone()
66    }
67
68    fn response(&self) -> Response {
69        let (version, body) = self.current();
70        if version == 0 {
71            return Response::text(STATUS_CODE_REASON_PHRASE.n503_service_unavailable, "no snapshot yet");
72        }
73        let mut response = Response::text(STATUS_CODE_REASON_PHRASE.n200_ok, &body);
74        response.headers.retain(|h| !h.name.eq_ignore_ascii_case("content-type"));
75        response.headers.push(crate::header::Header::new("Content-Type", "application/json"));
76        response.headers.push(crate::header::Header::new(VERSION_HEADER, version.to_string()));
77        response.headers.push(crate::header::Header::new("Cache-Control", "no-store"));
78        response
79    }
80
81    /// Wait until the version isn't `seen` (or [`LONG_POLL`] passes).
82    async fn wait_for_change(&self, seen: u64) {
83        let mut changed = self.changed.subscribe();
84        if *changed.borrow_and_update() != seen {
85            return;
86        }
87        let _ = tokio::time::timeout(LONG_POLL, changed.changed()).await;
88    }
89}
90
91/// A snapshot's version: the first 8 bytes of its SHA-256, never 0 (0
92/// means "nothing yet").
93pub fn content_version(body: &str) -> u64 {
94    let digest = crate::core::sha256::sha256(body.as_bytes());
95    u64::from_be_bytes(digest[..8].try_into().unwrap_or_default()).max(1)
96}
97
98/// `version=<n>` of a request target.
99pub fn requested_version(uri: &str) -> Option<u64> {
100    let query = uri.split_once('?')?.1;
101    query.split('&').find_map(|pair| pair.strip_prefix("version=")?.parse().ok())
102}
103
104fn is_snapshot_request(request: &Request) -> bool {
105    request.method == "GET" && request.request_uri.split('?').next() == Some(PATH)
106}
107
108impl Application for Publisher {
109    fn execute(&self, request: &Request, _connection: &ConnectionInfo) -> Result<Response, String> {
110        if !is_snapshot_request(request) {
111            return Ok(Response::text(STATUS_CODE_REASON_PHRASE.n404_not_found, "not found"));
112        }
113        Ok(self.response())
114    }
115
116    fn execute_async<'a>(&'a self, request: &'a Request, _connection: &'a ConnectionInfo) -> Option<crate::application::AsyncResponse<'a>> {
117        if !is_snapshot_request(request) {
118            return None;
119        }
120        Some(Box::pin(async move {
121            if let Some(seen) = requested_version(&request.request_uri) {
122                self.wait_for_change(seen).await;
123            }
124            Ok(self.response())
125        }))
126    }
127}
128
129/// Serve the endpoint on `port` (all interfaces) until the runtime stops.
130pub async fn serve(publisher: Publisher, port: u16) {
131    let listener = match tokio::net::TcpListener::bind(("0.0.0.0", port)).await {
132        Ok(listener) => listener,
133        Err(e) => {
134            eprintln!("[gateway-api] mesh: can't serve the configuration on port {port}: {e}");
135            return;
136        }
137    };
138    loop {
139        let Ok((stream, peer)) = listener.accept().await else {
140            tokio::time::sleep(Duration::from_millis(50)).await;
141            continue;
142        };
143        tokio::spawn(crate::server::serve_plain_connection(stream, peer, publisher.clone()));
144    }
145}
rws_gateway_api/gateway_api/mesh/config.rs

rws_gateway_api/gateway_api/mesh/
config.rs
