diff --git a/Cargo.lock b/Cargo.lock index 4583f71bae..c32f9fba12 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5743,6 +5743,7 @@ dependencies = [ "serde", "serde_json", "sha2 0.11.0", + "socket2", "tempfile", "thiserror 2.0.20", "time", diff --git a/crates/apps/data/plane/svc/src/main.rs b/crates/apps/data/plane/svc/src/main.rs index a490a5e3d4..ee527b363a 100644 --- a/crates/apps/data/plane/svc/src/main.rs +++ b/crates/apps/data/plane/svc/src/main.rs @@ -6,7 +6,7 @@ use pplx_space_data_plane_server::{ DEFAULT_MAX_ATTACHED_STREAMS, DEFAULT_MAX_ATTACHED_STREAMS_PER_ORG, DEFAULT_MAX_ATTACHED_STREAMS_PER_SANDBOX, DEFAULT_REVOCATION_CHECK_INTERVAL_MS, DEFAULT_ROUTE_CACHE_TTL_MS, DEFAULT_UNAUTHENTICATED_UPGRADE_CAP_PER_IP, DataPlaneConfig, - MIN_SHUTDOWN_GRACE, UploadAdmissionArgs, + MAX_REVOCATION_CHECK_INTERVAL_MS, MIN_SHUTDOWN_GRACE, UploadAdmissionArgs, }; // Same allocator choice as space-api-svc: glibc arenas keep the peak of an @@ -75,8 +75,13 @@ struct Cli { #[arg(long, default_value_t = DEFAULT_BACKEND_CHANNEL_POOL_SIZE)] backend_channel_pool_size: usize, - /// Interval between revocation checks for long-lived attached streams. - #[arg(long, default_value_t = DEFAULT_REVOCATION_CHECK_INTERVAL_MS)] + /// Interval between revocation checks for long-lived attached streams; at + /// most 19000, so a failed check can keep the last confirmed verdict. + #[arg( + long, + default_value_t = DEFAULT_REVOCATION_CHECK_INTERVAL_MS, + value_parser = clap::value_parser!(u64).range(..=MAX_REVOCATION_CHECK_INTERVAL_MS) + )] revocation_check_interval_ms: u64, /// Short-lived TTL for token-scoped route resolution cache entries. diff --git a/crates/apps/data/plane/svc/src/tests.rs b/crates/apps/data/plane/svc/src/tests.rs index a232143db6..563c46caa6 100644 --- a/crates/apps/data/plane/svc/src/tests.rs +++ b/crates/apps/data/plane/svc/src/tests.rs @@ -53,3 +53,21 @@ fn shutdown_grace_below_the_stream_open_budget_is_rejected() { Some(MIN_SHUTDOWN_GRACE) ); } + +#[test] +fn revocation_check_interval_above_the_maximum_is_rejected() { + let maximum = MAX_REVOCATION_CHECK_INTERVAL_MS.to_string(); + assert_eq!( + config(&["--revocation-check-interval-ms", &maximum]).revocation_check_interval, + Duration::from_millis(MAX_REVOCATION_CHECK_INTERVAL_MS) + ); + let above = (MAX_REVOCATION_CHECK_INTERVAL_MS + 1).to_string(); + assert!( + Cli::try_parse_from([ + "space-data-plane-svc", + "--revocation-check-interval-ms", + &above + ]) + .is_err() + ); +} diff --git a/crates/libs/data/plane/server/Cargo.toml b/crates/libs/data/plane/server/Cargo.toml index 85008b3c55..54cdd3ceb9 100644 --- a/crates/libs/data/plane/server/Cargo.toml +++ b/crates/libs/data/plane/server/Cargo.toml @@ -38,6 +38,9 @@ rand = { workspace = true } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } sha2 = { workspace = true } +# `all` gates TcpKeepalive::with_retries, which the listener sets, and the +# socket option getters the listener test reads. +socket2 = { workspace = true, features = ["all"] } tempfile = { workspace = true } thiserror = { workspace = true } time = { workspace = true, features = ["formatting"] } diff --git a/crates/libs/data/plane/server/src/drain.rs b/crates/libs/data/plane/server/src/drain.rs index fe972c737f..46d07efa72 100644 --- a/crates/libs/data/plane/server/src/drain.rs +++ b/crates/libs/data/plane/server/src/drain.rs @@ -52,7 +52,7 @@ enum DrainEnding { /// second stop signal cancels `force_token`, which ends the drain the same way /// the deadline does. pub(super) async fn serve_and_drain( - listener: pplx_space_common_svc::http_listener::NodelayListener, + listener: DataPlaneListener, app: Router, lifecycle: DrainLifecycle, ) -> std::io::Result<()> { diff --git a/crates/libs/data/plane/server/src/drain/tests.rs b/crates/libs/data/plane/server/src/drain/tests.rs index f8a8891c89..e2a5c7e654 100644 --- a/crates/libs/data/plane/server/src/drain/tests.rs +++ b/crates/libs/data/plane/server/src/drain/tests.rs @@ -1,5 +1,5 @@ use super::*; -use crate::tests::test_data_plane_state; +use crate::tests::{read_server_frame, test_data_plane_state}; use bytes::Bytes; use tokio::sync::mpsc; use tokio_stream::wrappers::ReceiverStream; @@ -28,7 +28,7 @@ async fn start_drain( let force_token = CancellationToken::new(); let tasks = TaskTracker::new(); let server = tokio::spawn(serve_and_drain( - pplx_space_common_svc::http_listener::nodelay(listener), + data_plane_listener(listener), app(stream_shutdown_token.clone()), DrainLifecycle { shutdown_token: shutdown_token.clone(), @@ -372,28 +372,6 @@ async fn drain_refusals_close_http1_connections_only() { assert!(response.headers().get(header::CONNECTION).is_none()); } -/// Read one server WebSocket frame header from `tcp` and return its opcode and -/// payload. Server frames are never masked, so the header is two bytes plus an -/// optional extended length. -async fn read_server_frame(tcp: &mut tokio::net::TcpStream) -> (u8, Vec) { - let mut header = [0_u8; 2]; - tcp.read_exact(&mut header).await.expect("frame header"); - let opcode = header[0] & 0x0f; - let len = match header[1] & 0x7f { - 126 => { - let mut extended = [0_u8; 2]; - tcp.read_exact(&mut extended) - .await - .expect("extended length"); - usize::from(u16::from_be_bytes(extended)) - } - len => usize::from(len), - }; - let mut payload = vec![0_u8; len]; - tcp.read_exact(&mut payload).await.expect("frame payload"); - (opcode, payload) -} - #[tokio::test] async fn unattached_websocket_is_told_to_reconnect_when_the_drain_starts() { if !insecure_dev_test_process() { @@ -409,7 +387,7 @@ async fn unattached_websocket_is_told_to_reconnect_when_the_drain_starts() { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); let server = tokio::spawn(serve_and_drain( - pplx_space_common_svc::http_listener::nodelay(listener), + data_plane_listener(listener), router(state), DrainLifecycle { shutdown_token: shutdown_token.clone(), diff --git a/crates/libs/data/plane/server/src/lib.rs b/crates/libs/data/plane/server/src/lib.rs index b23c8e5573..41b4ef5558 100644 --- a/crates/libs/data/plane/server/src/lib.rs +++ b/crates/libs/data/plane/server/src/lib.rs @@ -143,6 +143,21 @@ pub const DEFAULT_MAX_ATTACHED_STREAMS_PER_SANDBOX: usize = 512; pub const DEFAULT_BACKEND_CHANNEL_POOL_SIZE: usize = 4; pub const DEFAULT_REVOCATION_CHECK_INTERVAL_MS: u64 = 10_000; pub const DEFAULT_ROUTE_CACHE_TTL_MS: u64 = 1_000; +/// How long after its last confirmed revocation check an attached stream may +/// keep running while later checks fail, before it fails closed. +const REVOCATION_MAX_STALENESS: Duration = Duration::from_secs(60); +/// How long one revocation check may wait on the cluster API, across all of +/// its reads. A check that takes longer counts as unavailable. +const REVOCATION_CHECK_TIMEOUT: Duration = Duration::from_secs(3); +/// The longest revocation check interval the service accepts: a third of +/// what is left of the staleness bound after one check timeout, 19 s at the +/// defaults. A failed check keeps a verdict until it is that bound less one +/// interval and one timeout old, so at this interval or below that is at +/// least two intervals: a stream rides out at least the first failed check +/// after a confirmed one, and a refresh another stream has in flight can +/// still answer for it. +pub const MAX_REVOCATION_CHECK_INTERVAL_MS: u64 = + ((REVOCATION_MAX_STALENESS.as_millis() - REVOCATION_CHECK_TIMEOUT.as_millis()) / 3) as u64; /// The shortest drain deadline the service accepts: the 8-second stream-open /// budget, so a `Process.Start` already opening when SIGTERM arrives finishes /// opening, and can report whether the command started, before the deadline. @@ -205,6 +220,8 @@ pub struct DataPlaneConfig { pub max_attached_streams_per_org: usize, pub max_attached_streams_per_sandbox: usize, pub backend_channel_pool_size: usize, + /// At most [`MAX_REVOCATION_CHECK_INTERVAL_MS`]; [`run`] refuses a longer + /// one. pub revocation_check_interval: Duration, pub route_cache_ttl: Duration, /// `None` leaves `POST /files` unbounded. @@ -252,6 +269,8 @@ struct DataPlaneState { ecw_process_input_locks: Arc, ecw_hosts: Arc, revocation_check_interval: Duration, + revocation_max_staleness: Duration, + revocation_check_timeout: Duration, route_cache_ttl: Duration, /// Cancelled on SIGTERM: a child of the caller's shutdown token. New /// streams are refused once it is cancelled, and sockets still @@ -361,8 +380,26 @@ fn cluster_api_flag_attributes() -> Attributes { attributes } +impl DataPlaneConfig { + /// Check the settings [`run`] cannot serve with. + pub fn validate(&self) -> Result<(), RunDataPlaneError> { + if self.revocation_check_interval > Duration::from_millis(MAX_REVOCATION_CHECK_INTERVAL_MS) + { + return Err(RunDataPlaneError::RevocationCheckIntervalTooLong { + interval_ms: self.revocation_check_interval.as_millis(), + }); + } + Ok(()) + } +} + #[derive(Debug, thiserror::Error)] pub enum RunDataPlaneError { + #[error( + "revocation check interval of {interval_ms} ms is above the maximum of {} ms", + MAX_REVOCATION_CHECK_INTERVAL_MS + )] + RevocationCheckIntervalTooLong { interval_ms: u128 }, #[error("failed to connect to SPACE internal ClusterApi: {0}")] ConnectClusterApi(#[source] ConnectClusterApiError), #[error("failed to bind data-plane listener at {listen}: {source}")] @@ -382,6 +419,7 @@ pub async fn run( config: DataPlaneConfig, stop: pplx_space_common_svc::StopTokens, ) -> Result<(), RunDataPlaneError> { + config.validate()?; let pplx_space_common_svc::StopTokens { shutdown: shutdown_token, forced: force_token, @@ -420,12 +458,20 @@ pub async fn run( revocation_check_interval: config .revocation_check_interval .max(Duration::from_millis(1)), + revocation_max_staleness: REVOCATION_MAX_STALENESS, + revocation_check_timeout: REVOCATION_CHECK_TIMEOUT, route_cache_ttl: config.route_cache_ttl.max(Duration::from_millis(1)), drain_started, stream_shutdown_token: stream_shutdown_token.clone(), upgrade_tasks: upgrade_tasks.clone(), }; + tracing::info!( + revocation_check_interval_ms = state.revocation_check_interval.as_millis(), + revocation_tolerance_ms = native::revocation_tolerance(&state).as_millis(), + "attached streams keep a confirmed revocation verdict through failed checks \ + for up to the revocation tolerance" + ); let app = router(state); let listener = tokio::net::TcpListener::bind(&config.listen) .await @@ -436,7 +482,7 @@ pub async fn run( tracing::info!(listen = %config.listen, "space-data-plane-svc listening"); serve_and_drain( - pplx_space_common_svc::http_listener::nodelay(listener), + data_plane_listener(listener), app, DrainLifecycle { shutdown_token, @@ -451,6 +497,49 @@ pub async fn run( .map_err(RunDataPlaneError::Serve) } +/// How long an accepted connection may sit idle before its first keepalive +/// probe. +const DATA_PLANE_TCP_KEEPALIVE_TIME: Duration = Duration::from_secs(30); +/// How long between keepalive probes. +const DATA_PLANE_TCP_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(10); +/// How many unanswered keepalive probes drop the connection. +const DATA_PLANE_TCP_KEEPALIVE_RETRIES: u32 = 3; + +type DataPlaneListener = axum::serve::TapIo< + pplx_space_common_svc::http_listener::NodelayListener, + fn(&mut tokio::net::TcpStream), +>; + +/// Accept connections with `TCP_NODELAY` and the data plane's dead-peer +/// detection set, on every route this listener serves, ECW's included. +/// +/// Keepalive probes a connection that carries nothing for 30 seconds and drops +/// it after about a minute of silence, so an idle connection that dies +/// silently releases its stream, and the node's per-sandbox permit behind it. +/// It detects the immediate TCP peer, which is often a proxy rather than the +/// end client. It never closes a connection whose peer still answers, so it +/// sets no limit on how long a client may stop reading. A stream that sends +/// regularly, such as SSE with its 15-second keep-alive comments, is never +/// idle, so a dead peer on it waits out the kernel's retransmission timeout. +/// No `TCP_USER_TIMEOUT` is set: this listener serves ECW routes too, and on +/// Linux that option would also decide when keepalive gives up. +fn data_plane_listener(listener: tokio::net::TcpListener) -> DataPlaneListener { + fn set_dead_peer_detection(tcp: &mut tokio::net::TcpStream) { + let socket = socket2::SockRef::from(&*tcp); + let keepalive = socket2::TcpKeepalive::new() + .with_time(DATA_PLANE_TCP_KEEPALIVE_TIME) + .with_interval(DATA_PLANE_TCP_KEEPALIVE_INTERVAL) + .with_retries(DATA_PLANE_TCP_KEEPALIVE_RETRIES); + if let Err(error) = socket.set_tcp_keepalive(&keepalive) { + tracing::debug!(%error, "failed to set TCP keepalive on an accepted connection"); + } + } + axum::serve::ListenerExt::tap_io( + pplx_space_common_svc::http_listener::nodelay(listener), + set_dead_peer_detection as fn(&mut tokio::net::TcpStream), + ) +} + async fn connect_cluster_api_clients_for_startup( shutdown_token: &CancellationToken, ) -> Result { diff --git a/crates/libs/data/plane/server/src/native.rs b/crates/libs/data/plane/server/src/native.rs index 2214ab2a7c..6562900ad6 100644 --- a/crates/libs/data/plane/server/src/native.rs +++ b/crates/libs/data/plane/server/src/native.rs @@ -47,17 +47,17 @@ async fn resolve_session_route_with_cache( && use_cache && !space_api_route_token_needs_refresh(&attached.space_api_route_token) { - state.revocation_cache.remember_active( + state.revocation_cache.remember_cached_active( session_id, attached.revocation_epoch, attached.org_id, attached.org_revocation_epoch, - state.revocation_check_interval, ); return Ok(attached); } let mut client = state.cluster_api.for_session(&session_id); + let read = start_revocation_read(state); let response = client .resolve_data_plane_session_route(ResolveDataPlaneSessionRouteRequest { session_id: session_id.to_string(), @@ -87,7 +87,7 @@ async fn resolve_session_route_with_cache( attached.revocation_epoch, attached.org_id, attached.org_revocation_epoch, - state.revocation_check_interval, + read, ); attached.cache_key = Some(cache_key.clone()); state @@ -367,6 +367,7 @@ pub(super) async fn resolve_session_route_internal( session_id: DataPlaneSessionId, ) -> Result { let mut client = state.cluster_api.for_session(&session_id); + let read = start_revocation_read(state); let response = client .resolve_data_plane_session_route(ResolveDataPlaneSessionRouteRequest { session_id: session_id.to_string(), @@ -389,7 +390,7 @@ pub(super) async fn resolve_session_route_internal( attached.revocation_epoch, attached.org_id, attached.org_revocation_epoch, - state.revocation_check_interval, + read, ); Ok(attached) } @@ -782,12 +783,121 @@ pub(super) fn revocation_check_interval(state: &DataPlaneState) -> tokio::time:: interval } +/// Whether an attached stream's session has been revoked, checked on each +/// revocation tick. The whole check, which can take an org read and then a +/// session read, gets one [`REVOCATION_CHECK_TIMEOUT`]; a check that outlives +/// it counts as one the cluster API could not answer. +/// +/// A check the cluster API could not answer keeps the session's last +/// confirmed verdict while that verdict is young enough, so a brief cluster +/// API or database outage does not end every attached stream. Once it is not, +/// the error is returned and the stream fails closed: the first failed check +/// that finishes once the verdict is older than [`revocation_tolerance`] ends +/// it, at most the staleness bound ([`REVOCATION_MAX_STALENESS`]) after the +/// read that confirmed it. The verdict's age is its org entry's; see +/// [`RevocationCache::confirmed_active`]. Any other failure, such as a +/// malformed answer, fails closed at once, and so does a confirmed revocation. pub(super) async fn session_is_revoked( state: &DataPlaneState, session_id: DataPlaneSessionId, ) -> Result { let kind = session_id.kind().as_str(); - match state.revocation_cache.probe(session_id) { + let checked = tokio::time::timeout( + state.revocation_check_timeout, + check_session_revocation(state, session_id), + ) + .await + .unwrap_or_else(|_| { + pplx_space_metrics::data_plane::revocation_check(kind, "timeout"); + Err(StreamBridgeError::RevocationStatus( + revocation_check_timed_out(), + )) + }); + match checked { + Err(error) if keeps_last_verdict(state, session_id, &error) => { + pplx_space_metrics::data_plane::revocation_check(kind, "kept_last_verdict"); + tracing::warn!(%error, "revocation check failed; keeping the last confirmed verdict"); + Ok(false) + } + result => result, + } +} + +/// Whether a failed check leaves the stream on the session's last confirmed +/// verdict rather than failing it closed. +fn keeps_last_verdict( + state: &DataPlaneState, + session_id: DataPlaneSessionId, + error: &StreamBridgeError, +) -> bool { + revocation_check_unavailable(error) + && state + .revocation_cache + .confirmed_active(session_id, revocation_tolerance(state)) +} + +/// How old a confirmed verdict may be and still stand in for a failed check. +pub(super) fn revocation_tolerance(state: &DataPlaneState) -> Duration { + revocation_tolerance_within( + state.revocation_max_staleness, + state.revocation_check_interval, + state.revocation_check_timeout, + ) +} + +/// The check that finds a verdict too old can start up to one interval after +/// the verdict became too old and take up to one check timeout, so both come +/// off the staleness bound and the stream ends within it. +pub(super) fn revocation_tolerance_within( + max_staleness: Duration, + check_interval: Duration, + check_timeout: Duration, +) -> Duration { + max_staleness + .saturating_sub(check_interval) + .saturating_sub(check_timeout) +} + +/// A revocation read beginning now, whose answer stays fresh for just under +/// one check interval. A stream's checks are an interval apart, but each +/// starts a little late by a varying amount; with a full interval, a stream +/// alone in its org could start a check just before the refresh its own last +/// check set, and re-read the org only on every other check. +fn start_revocation_read(state: &DataPlaneState) -> RevocationRead { + let interval = state.revocation_check_interval; + RevocationRead::start(interval - interval / 10) +} + +/// Whether a revocation check failed for want of a usable answer from the +/// cluster API: it was unreachable, overloaded or timed out, the call was +/// cancelled, or it failed in a way tonic reports as `Internal` or `Unknown`, +/// which includes an answer that could not be decoded at all. A refusal, or +/// an answer that decoded but could not be used, is not one. +fn revocation_check_unavailable(error: &StreamBridgeError) -> bool { + matches!( + error, + StreamBridgeError::RevocationStatus(status) + if matches!( + status.code(), + tonic::Code::Unavailable + | tonic::Code::DeadlineExceeded + | tonic::Code::Cancelled + | tonic::Code::ResourceExhausted + | tonic::Code::Internal + | tonic::Code::Unknown + ) + ) +} + +async fn check_session_revocation( + state: &DataPlaneState, + session_id: DataPlaneSessionId, +) -> Result { + let kind = session_id.kind().as_str(); + match state + .revocation_cache + .probe(session_id, revocation_tolerance(state)) + { RevocationProbe::Fresh(active) => { pplx_space_metrics::data_plane::revocation_cache_probe(kind, "fresh"); Ok(!active) @@ -798,6 +908,7 @@ pub(super) async fn session_is_revoked( } RevocationProbe::RefreshSession => { pplx_space_metrics::data_plane::revocation_cache_probe(kind, "refresh"); + let read = start_revocation_read(state); let Some((org_id, revocation)) = fetch_session_revocation_state(state, session_id) .await .inspect_err(|error| { @@ -813,16 +924,14 @@ pub(super) async fn session_is_revoked( let active = revocation_state_is_active(&revocation); let result = if active { "active" } else { "revoked" }; pplx_space_metrics::data_plane::revocation_check(kind, result); - state.revocation_cache.remember( - session_id, - org_id, - &revocation, - state.revocation_check_interval, - ); + state + .revocation_cache + .remember(session_id, org_id, &revocation, read); Ok(!active) } RevocationProbe::RefreshOrg(refresh) => { pplx_space_metrics::data_plane::revocation_cache_probe(kind, "refresh"); + let read = start_revocation_read(state); let org_epoch = fetch_org_revocation_epoch(state, refresh.org_id) .await .inspect_err(|error| { @@ -831,11 +940,9 @@ pub(super) async fn session_is_revoked( revocation_check_error_label(error), ); })?; - state.revocation_cache.remember_org_epoch( - refresh.org_id, - org_epoch, - state.revocation_check_interval, - ); + state + .revocation_cache + .remember_org_epoch(refresh.org_id, org_epoch, read); if org_epoch <= refresh.cached_org_epoch { pplx_space_metrics::data_plane::revocation_check(kind, "org_epoch_unchanged"); drop(refresh.guard); @@ -843,6 +950,7 @@ pub(super) async fn session_is_revoked( } pplx_space_metrics::data_plane::revocation_check(kind, "org_epoch_changed"); drop(refresh.guard); + let read = start_revocation_read(state); let Some((org_id, revocation)) = fetch_session_revocation_state(state, session_id) .await .inspect_err(|error| { @@ -858,12 +966,9 @@ pub(super) async fn session_is_revoked( let active = revocation_state_is_active(&revocation); let result = if active { "active" } else { "revoked" }; pplx_space_metrics::data_plane::revocation_check(kind, result); - state.revocation_cache.remember( - session_id, - org_id, - &revocation, - state.revocation_check_interval, - ); + state + .revocation_cache + .remember(session_id, org_id, &revocation, read); Ok(!active) } } @@ -910,6 +1015,12 @@ pub(super) async fn fetch_org_revocation_epoch( Ok(response.epoch) } +/// A revocation check that outlived its timeout, reported as the deadline +/// status the cluster API would have returned. +fn revocation_check_timed_out() -> tonic::Status { + tonic::Status::deadline_exceeded("revocation check timed out") +} + pub(super) fn revocation_check_error_label(error: &StreamBridgeError) -> &'static str { match error { StreamBridgeError::MissingRevocationState => "missing_state", diff --git a/crates/libs/data/plane/server/src/native/sse.rs b/crates/libs/data/plane/server/src/native/sse.rs index bf35308f32..8b8fe037f5 100644 --- a/crates/libs/data/plane/server/src/native/sse.rs +++ b/crates/libs/data/plane/server/src/native/sse.rs @@ -348,7 +348,7 @@ impl ProcessSseState { self.finished = true; self.push_frame(DataPlaneServerFrame::Error { code: DataPlaneErrorCode::BackendStreamError, - message: error.to_string(), + message: PROCESS_STREAM_FAILED.to_string(), retryable: true, }); } @@ -658,7 +658,7 @@ impl FileWatchSseState { self.finished = true; self.push_frame(DataPlaneServerFrame::Error { code: DataPlaneErrorCode::BackendStreamError, - message: error.to_string(), + message: FILE_WATCH_STREAM_FAILED.to_string(), retryable: true, }); } diff --git a/crates/libs/data/plane/server/src/native/streams.rs b/crates/libs/data/plane/server/src/native/streams.rs index 7c4e752110..e1ff4adf2e 100644 --- a/crates/libs/data/plane/server/src/native/streams.rs +++ b/crates/libs/data/plane/server/src/native/streams.rs @@ -198,6 +198,16 @@ pub(super) fn process_stream_start_http_error(error: &StreamBridgeError) -> Resp ) } +/// Public messages for the error frames that report a backend or internal +/// error. That error can name internal endpoints or carry the guest's own +/// wording, so these frames carry one of these instead and the detail goes to +/// the log. Frames about the client's own request carry their own fixed text. +pub(super) const PROCESS_STREAM_FAILED: &str = "process event stream failed"; +pub(super) const FILE_WATCH_STREAM_FAILED: &str = "file watch event stream failed"; +pub(super) const STREAM_OPEN_FAILED: &str = "event stream could not be opened"; +pub(super) const STREAM_OPEN_REFUSED: &str = "this request was refused"; +pub(super) const RESIZE_REFUSED: &str = "terminal resize is not supported by this process"; + pub(super) fn stream_start_error_frame(error: &StreamBridgeError) -> DataPlaneServerFrame { // A refusal made before anything started is retryable here as on SSE. if let StreamBridgeError::BackendStatus(status) = error @@ -205,7 +215,7 @@ pub(super) fn stream_start_error_frame(error: &StreamBridgeError) -> DataPlaneSe { return DataPlaneServerFrame::Error { code: DataPlaneErrorCode::BackendStreamError, - message: format!("this request was refused: {}", status.message()), + message: STREAM_OPEN_REFUSED.to_string(), retryable: true, }; } @@ -215,9 +225,10 @@ pub(super) fn stream_start_error_frame(error: &StreamBridgeError) -> DataPlaneSe if !pplx_space_common_grpc::is_terminal_guest_process_error(status) && backend_stream_status_retryable(status) ); + // The bridge logs the error. DataPlaneServerFrame::Error { code: DataPlaneErrorCode::BackendStreamError, - message: error.to_string(), + message: STREAM_OPEN_FAILED.to_string(), retryable, } } diff --git a/crates/libs/data/plane/server/src/native/tests.rs b/crates/libs/data/plane/server/src/native/tests.rs index 1aa60f8e30..8730392b31 100644 --- a/crates/libs/data/plane/server/src/native/tests.rs +++ b/crates/libs/data/plane/server/src/native/tests.rs @@ -96,6 +96,7 @@ async fn process_start_http_errors_preserve_permanent_guest_failures() { let body: serde_json::Value = serde_json::from_slice(&body).unwrap(); assert_eq!(body["error"]["retryable"], retryable, "{code:?}: {body}"); let DataPlaneServerFrame::Error { + message, retryable: actual_websocket_retryable, .. } = stream_start_error_frame(&error) @@ -106,6 +107,7 @@ async fn process_start_http_errors_preserve_permanent_guest_failures() { actual_websocket_retryable, websocket_retryable, "WebSocket opening errors retain the established stream retry contract" ); + assert_eq!(message, STREAM_OPEN_FAILED, "{code:?}"); } } @@ -378,10 +380,14 @@ async fn refusals_before_start_are_retryable_with_retry_after() { assert_eq!(body["error"]["code"], expected_code, "{body}"); assert_eq!(body["error"]["retryable"], true, "{body}"); assert!(!body.to_string().contains("synthetic guest wording")); - let DataPlaneServerFrame::Error { retryable, .. } = stream_start_error_frame(&error) else { + let DataPlaneServerFrame::Error { + message, retryable, .. + } = stream_start_error_frame(&error) + else { panic!("expected error frame") }; assert!(retryable, "WebSocket agrees with SSE for {code:?}"); + assert_eq!(message, STREAM_OPEN_REFUSED, "{code:?}"); } } @@ -715,3 +721,991 @@ async fn binding_survives_a_client_that_disconnects_mid_bind() { .expect("the binding completed after the client left"); assert_eq!(fake.bound_process_ids.lock().unwrap().len(), 1); } + +/// How a fake cluster API answers revocation checks. +#[derive(Clone, Copy)] +enum FakeRevocation { + Active, + /// The cluster API is unreachable; its status names an internal address. + Down, + /// The cluster API takes the check and never answers. + Hung, + /// The org epoch moved on, and the session check then never answers. + HungAfterOrgEpochChange, + /// The org read takes most of a check timeout and finds the epoch moved + /// on, and the session read then never answers. + SlowOrgThenHungSession, + /// Every call, route resolves included, answers as `Active` does after + /// [`FAKE_SLOW_ANSWER`]. + Slow, + Revoked, + /// The cluster API refuses the check. + Rejected, +} + +/// How long [`FakeRevocation::Slow`] takes to answer. +const FAKE_SLOW_ANSWER: Duration = Duration::from_millis(300); +/// The check timeout for [`FakeRevocation::SlowOrgThenHungSession`], and how +/// long its org read takes. +const SPLIT_CHECK_TIMEOUT: Duration = Duration::from_millis(2_000); +const SPLIT_CHECK_ORG_READ: Duration = Duration::from_millis(1_200); + +/// A fake cluster API and worker for one running process. The test sets how +/// revocation checks are answered and sends stdout chunks down the open +/// process stream. +struct RevocationFake { + endpoint: String, + revocation: Arc>, + /// Revocation checks answered with an error. + failed_answers: Arc, + stdout: tokio::sync::mpsc::UnboundedSender<&'static str>, +} + +impl RevocationFake { + fn answer(&self, revocation: FakeRevocation) { + *self.revocation.lock().unwrap() = revocation; + } + + fn print(&self, data: &'static str) { + self.stdout.send(data).unwrap(); + } + + fn failed_answers(&self) -> usize { + self.failed_answers.load(Ordering::SeqCst) + } +} + +/// Make the session's org due for a refresh, so its next check reads the +/// cluster API however soon the last read answered. +fn make_org_refresh_due(state: &DataPlaneState, session_id: DataPlaneSessionId) { + let org_id = state + .revocation_cache + .entries + .get(&session_id.to_string()) + .unwrap() + .org_id; + state + .revocation_cache + .orgs + .get_mut(&org_id) + .unwrap() + .refresh_after = tokio::time::Instant::now(); +} + +const FAKE_CLUSTER_API_OUTAGE: &str = "connection refused: 192.0.2.10:9443"; + +fn grpc_message(message: &impl prost::Message) -> bytes::Bytes { + let encoded = message.encode_to_vec(); + let mut framed = vec![0]; + framed.extend_from_slice(&(encoded.len() as u32).to_be_bytes()); + framed.extend(encoded); + bytes::Bytes::from(framed) +} + +async fn spawn_revocation_fake(session_id: DataPlaneSessionId) -> RevocationFake { + use pplx_space_node_api_client::proto::{ + GetDataPlaneOrgRevocationEpochResponse, GetDataPlaneSessionRevocationStateResponse, + }; + use prost::Message as _; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let endpoint = format!("http://{}", listener.local_addr().unwrap()); + let mut route = crate::tests::route_response_with_token(session_id, None); + route.backend.as_mut().unwrap().endpoint = endpoint.clone(); + route.adapter_state_json = r#"{"command":["sleep","600"]}"#.to_string(); + let org_id = route.org_id.clone(); + let route = route.encode_to_vec(); + let process_id = pplx_space_public_api::process_id_for_session(session_id).to_string(); + let revocation = Arc::new(std::sync::Mutex::new(FakeRevocation::Active)); + let failed_answers = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let (stdout, stdout_rx) = tokio::sync::mpsc::unbounded_channel(); + let stdout_rx = Arc::new(std::sync::Mutex::new(Some(stdout_rx))); + let fake = RevocationFake { + endpoint, + revocation: Arc::clone(&revocation), + failed_answers: Arc::clone(&failed_answers), + stdout, + }; + let service = axum::Router::new().fallback(move |request: Request| { + let route = route.clone(); + let org_id = org_id.clone(); + let process_id = process_id.clone(); + let answer = *revocation.lock().unwrap(); + let failed_answers = Arc::clone(&failed_answers); + let stdout_rx = Arc::clone(&stdout_rx); + async move { + let path = request.uri().path().to_string(); + if path.ends_with("/Check") { + return crate::tests::grpc_test_response(vec![8, 1]); + } + if matches!(answer, FakeRevocation::Slow) { + tokio::time::sleep(FAKE_SLOW_ANSWER).await; + } + if path.ends_with("/ResolveDataPlaneSessionRoute") { + return crate::tests::grpc_test_response(route); + } + if path.ends_with("/BindDataPlaneSessionProcess") { + return crate::tests::grpc_test_response(Vec::new()); + } + let org_epoch_request = path.ends_with("/GetDataPlaneOrgRevocationEpoch"); + if org_epoch_request || path.ends_with("/GetDataPlaneSessionRevocationState") { + let failed = || failed_answers.fetch_add(1, Ordering::SeqCst); + let (epoch, status) = match answer { + FakeRevocation::Active | FakeRevocation::Slow => { + (1, DataPlaneRevocationStatus::Active) + } + FakeRevocation::Down => { + failed(); + return tonic::Status::unavailable(FAKE_CLUSTER_API_OUTAGE) + .into_http::(); + } + FakeRevocation::Rejected => { + failed(); + return tonic::Status::permission_denied(FAKE_CLUSTER_API_OUTAGE) + .into_http::(); + } + FakeRevocation::Hung => std::future::pending().await, + FakeRevocation::HungAfterOrgEpochChange + | FakeRevocation::SlowOrgThenHungSession + if !org_epoch_request => + { + std::future::pending().await + } + FakeRevocation::SlowOrgThenHungSession => { + tokio::time::sleep(SPLIT_CHECK_ORG_READ).await; + (2, DataPlaneRevocationStatus::Active) + } + FakeRevocation::Revoked => (2, DataPlaneRevocationStatus::Revoked), + FakeRevocation::HungAfterOrgEpochChange => { + (2, DataPlaneRevocationStatus::Active) + } + }; + if org_epoch_request { + return crate::tests::grpc_test_response( + GetDataPlaneOrgRevocationEpochResponse { + org_id, + epoch, + updated_at: None, + } + .encode_to_vec(), + ); + } + return crate::tests::grpc_test_response( + GetDataPlaneSessionRevocationStateResponse { + session_id: session_id.to_string(), + protocol_session_id: None, + revocation: Some(DataPlaneRevocationState { + epoch, + status: status as i32, + updated_at: None, + reason: None, + org_epoch: epoch, + }), + org_id, + } + .encode_to_vec(), + ); + } + assert!( + path.ends_with("/SpacedStartProcess"), + "unexpected call {path}" + ); + let event = move |event| { + Ok::<_, Infallible>(hyper::body::Frame::data(grpc_message(&ProcessEvent { + process_id: process_id.clone(), + pid: Some(42), + event: Some(event), + }))) + }; + let started = event(spaced_process_event::Event::Started( + pplx_space_sbx_mgr_client::operations::ProcessStarted { start_time: None }, + )); + let output = tokio_stream::wrappers::UnboundedReceiverStream::new( + stdout_rx + .lock() + .unwrap() + .take() + .expect("one process stream"), + ) + .map(move |data: &'static str| { + event(spaced_process_event::Event::Stdout( + pplx_space_sbx_mgr_client::proto::SpacedProcessOutput { + data: data.as_bytes().to_vec(), + }, + )) + }); + let frames = futures_util::stream::iter([started]).chain(output); + Response::builder() + .header("content-type", "application/grpc") + .body(Body::new(http_body_util::StreamBody::new(frames))) + .unwrap() + } + }); + tokio::spawn(async move { + axum::serve(listener, service).await.unwrap(); + }); + fake +} + +/// Revocation timings for the stream tests below. Checks run every 20ms. The +/// staleness bound and check timeout are production's unless a test shortens +/// them; their assertions count checks or wait on outcomes rather than +/// measure time. +struct RevocationTimings { + max_staleness: Duration, + check_timeout: Duration, +} + +impl Default for RevocationTimings { + fn default() -> Self { + Self { + max_staleness: REVOCATION_MAX_STALENESS, + check_timeout: REVOCATION_CHECK_TIMEOUT, + } + } +} + +/// Opens a process session's event stream against the fake and reads it until +/// the process has started. +async fn open_revocation_checked_events( + fake: &RevocationFake, + session_id: DataPlaneSessionId, + timings: RevocationTimings, +) -> (tokio::sync::mpsc::UnboundedReceiver, String) { + let mut state = crate::tests::test_data_plane_state().await; + state.cluster_api = ClusterApiClients::service_only( + pplx_space_node_api_client::cluster::connect(fake.endpoint.clone(), "test") + .await + .unwrap(), + ); + state.revocation_check_interval = Duration::from_millis(20); + state.revocation_max_staleness = timings.max_staleness; + state.revocation_check_timeout = timings.check_timeout; + let token = format!("{DATA_PLANE_TOKEN_PREFIX}{}", "c".repeat(64)); + let response = tower::ServiceExt::oneshot( + router(state), + Request::get(format!("/v1/sessions/{session_id}/events")) + .header(header::AUTHORIZATION, format!("Bearer {token}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let mut body = response.into_body(); + let (chunk_tx, mut chunks) = tokio::sync::mpsc::unbounded_channel(); + tokio::spawn(async move { + while let Some(Ok(frame)) = body.frame().await { + if let Ok(data) = frame.into_data() + && chunk_tx + .send(String::from_utf8_lossy(&data).into_owned()) + .is_err() + { + break; + } + } + }); + let mut seen = String::new(); + read_until(&mut chunks, &mut seen, "process_started").await; + (chunks, seen) +} + +async fn assert_stream_ends(chunks: &mut tokio::sync::mpsc::UnboundedReceiver) { + tokio::time::timeout(Duration::from_secs(5), async { + while chunks.recv().await.is_some() {} + }) + .await + .expect("the stream ends"); +} + +fn stdout_marker(data: &str) -> String { + BASE64_STANDARD.encode(data) +} + +#[tokio::test] +async fn process_stream_survives_a_revocation_outage_within_the_staleness_bound() { + if !binding_fake_test_process() { + return; + } + let session_id = DataPlaneSessionId::new_for_kind(SessionKind::Process); + let fake = spawn_revocation_fake(session_id).await; + let (mut chunks, mut seen) = + open_revocation_checked_events(&fake, session_id, RevocationTimings::default()).await; + + // Failed checks well inside the bound: output keeps flowing. + fake.answer(FakeRevocation::Down); + tokio::time::timeout(Duration::from_secs(5), async { + while fake.failed_answers() < 5 { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("checks failed"); + fake.print("during the outage"); + read_until(&mut chunks, &mut seen, &stdout_marker("during the outage")).await; + + fake.answer(FakeRevocation::Active); + tokio::time::sleep(Duration::from_millis(100)).await; + fake.print("after the outage"); + read_until(&mut chunks, &mut seen, &stdout_marker("after the outage")).await; + assert!( + !seen.contains("backend_stream_error") && !seen.contains("closed"), + "{seen}" + ); +} + +#[tokio::test] +async fn process_stream_keeps_delivering_while_revocation_checks_hang() { + if !binding_fake_test_process() { + return; + } + let session_id = DataPlaneSessionId::new_for_kind(SessionKind::Process); + let fake = spawn_revocation_fake(session_id).await; + let timings = RevocationTimings { + check_timeout: Duration::from_millis(100), + ..RevocationTimings::default() + }; + let (mut chunks, mut seen) = open_revocation_checked_events(&fake, session_id, timings).await; + + // Output sent while checks hang arrives as each one times out, long + // before the cluster API client's own 30-second deadline. + fake.answer(FakeRevocation::Hung); + tokio::time::sleep(Duration::from_millis(300)).await; + fake.print("during the hang"); + read_until(&mut chunks, &mut seen, &stdout_marker("during the hang")).await; + + fake.answer(FakeRevocation::Active); + tokio::time::sleep(Duration::from_millis(300)).await; + fake.print("after the hang"); + read_until(&mut chunks, &mut seen, &stdout_marker("after the hang")).await; + assert!( + !seen.contains("backend_stream_error") && !seen.contains("closed"), + "{seen}" + ); +} + +#[tokio::test] +async fn hung_check_after_an_org_revocation_fails_closed_once_it_times_out() { + if !binding_fake_test_process() { + return; + } + let session_id = DataPlaneSessionId::new_for_kind(SessionKind::Process); + let fake = spawn_revocation_fake(session_id).await; + let timings = RevocationTimings { + check_timeout: Duration::from_millis(100), + ..RevocationTimings::default() + }; + let (mut chunks, mut seen) = open_revocation_checked_events(&fake, session_id, timings).await; + + // The org's epoch moved on, so the cached verdict no longer counts, and + // the session check that would settle it never answers. + fake.answer(FakeRevocation::HungAfterOrgEpochChange); + read_until(&mut chunks, &mut seen, "backend_stream_error").await; + assert_stream_ends(&mut chunks).await; +} + +#[tokio::test] +async fn process_stream_fails_closed_once_its_verdict_is_older_than_the_bound() { + if !binding_fake_test_process() { + return; + } + let session_id = DataPlaneSessionId::new_for_kind(SessionKind::Process); + let fake = spawn_revocation_fake(session_id).await; + let timings = RevocationTimings { + max_staleness: Duration::from_millis(500), + check_timeout: Duration::from_millis(50), + }; + let (mut chunks, mut seen) = open_revocation_checked_events(&fake, session_id, timings).await; + + fake.answer(FakeRevocation::Down); + read_until(&mut chunks, &mut seen, "backend_stream_error").await; + // About 20 checks fit in the 430ms the verdict is kept; well over a few + // of them were ridden out before the stream failed closed. + assert!( + fake.failed_answers() >= 5, + "failed closed after {} failed checks", + fake.failed_answers() + ); + assert!(seen.contains(r#""retryable":true"#), "{seen}"); + assert!( + seen.contains(r#""message":"process event stream failed""#), + "{seen}" + ); + assert!( + !seen.contains("192.0.2.10"), + "internal detail leaked: {seen}" + ); + assert_stream_ends(&mut chunks).await; +} + +#[tokio::test] +async fn refused_revocation_check_fails_closed_at_once() { + if !binding_fake_test_process() { + return; + } + let session_id = DataPlaneSessionId::new_for_kind(SessionKind::Process); + let fake = spawn_revocation_fake(session_id).await; + let (mut chunks, mut seen) = + open_revocation_checked_events(&fake, session_id, RevocationTimings::default()).await; + + // The verdict is fresh, but a refusal is not an outage to ride out: the + // first one ends the stream. + fake.answer(FakeRevocation::Rejected); + read_until(&mut chunks, &mut seen, "backend_stream_error").await; + assert_eq!(fake.failed_answers(), 1, "{seen}"); + assert_stream_ends(&mut chunks).await; +} + +/// Which failed checks keep a fresh verdict: every status the cluster API +/// could not answer with, and nothing else. +#[tokio::test] +async fn only_unanswered_revocation_checks_keep_the_last_verdict() { + if !binding_fake_test_process() { + return; + } + let state = crate::tests::test_data_plane_state().await; + let session_id = DataPlaneSessionId::new_for_kind(SessionKind::Process); + state.revocation_cache.remember_active( + session_id, + 1, + OrgId::new(), + 1, + start_revocation_read(&state), + ); + for code in (0..=16).map(tonic::Code::from) { + let keeps = match code { + tonic::Code::Unavailable + | tonic::Code::DeadlineExceeded + | tonic::Code::Cancelled + | tonic::Code::ResourceExhausted + | tonic::Code::Internal + | tonic::Code::Unknown => true, + tonic::Code::Ok + | tonic::Code::InvalidArgument + | tonic::Code::NotFound + | tonic::Code::AlreadyExists + | tonic::Code::PermissionDenied + | tonic::Code::FailedPrecondition + | tonic::Code::Aborted + | tonic::Code::OutOfRange + | tonic::Code::Unimplemented + | tonic::Code::DataLoss + | tonic::Code::Unauthenticated => false, + }; + let error = StreamBridgeError::RevocationStatus(tonic::Status::new(code, "synthetic")); + assert_eq!( + keeps_last_verdict(&state, session_id, &error), + keeps, + "{code:?}" + ); + } + // Answers that decoded but could not be used. + for error in [ + StreamBridgeError::MissingRevocationState, + StreamBridgeError::RouteState("invalid_org_id"), + ] { + assert!(!keeps_last_verdict(&state, session_id, &error), "{error}"); + } +} + +#[tokio::test] +async fn revoked_session_ends_its_stream_at_once() { + if !binding_fake_test_process() { + return; + } + let session_id = DataPlaneSessionId::new_for_kind(SessionKind::Process); + let fake = spawn_revocation_fake(session_id).await; + let (mut chunks, mut seen) = + open_revocation_checked_events(&fake, session_id, RevocationTimings::default()).await; + + // Even mid-outage, a revocation the next check confirms ends the stream + // without waiting out the minute-long staleness bound. + fake.answer(FakeRevocation::Down); + tokio::time::sleep(Duration::from_millis(100)).await; + fake.answer(FakeRevocation::Revoked); + read_until(&mut chunks, &mut seen, r#""reason":"revoked""#).await; + assert_stream_ends(&mut chunks).await; +} + +/// At production timings and on a paused clock, for every phase of the stream's +/// checks against its last confirmed read: failed checks keep the verdict +/// while the stream can still end within the bound, and no longer. +#[tokio::test] +async fn failed_checks_keep_a_verdict_only_while_the_stream_ends_within_the_bound() { + if !binding_fake_test_process() { + return; + } + let mut state = crate::tests::test_data_plane_state().await; + // Nothing below reaches the network, so the clock moves only as told. + tokio::time::pause(); + state.revocation_check_interval = + Duration::from_millis(crate::DEFAULT_REVOCATION_CHECK_INTERVAL_MS); + let interval = state.revocation_check_interval; + // The check that fails closed may take the whole check timeout, so it + // must start no later than this after the read that confirmed the + // verdict; and since checks are an interval apart, no earlier than one + // interval before it. + let latest_start = REVOCATION_MAX_STALENESS - REVOCATION_CHECK_TIMEOUT; + let unavailable = + StreamBridgeError::RevocationStatus(tonic::Status::unavailable("synthetic outage")); + let step = Duration::from_millis(250); + let phases = (interval.as_millis() / step.as_millis()) as u32; + for phase in (0..phases).map(|n| step * n) { + let session_id = DataPlaneSessionId::new_for_kind(SessionKind::Process); + state.revocation_cache.remember_active( + session_id, + 1, + OrgId::new(), + 1, + start_revocation_read(&state), + ); + let confirmed = tokio::time::Instant::now(); + // The first check `phase` after the confirmation, then one per + // interval, each decided as soon as it starts. + tokio::time::advance(phase).await; + let failing_check_started = loop { + if !keeps_last_verdict(&state, session_id, &unavailable) { + break confirmed.elapsed(); + } + tokio::time::advance(interval).await; + }; + assert!( + failing_check_started <= latest_start, + "phase {phase:?}: the failing check started {failing_check_started:?} after \ + the confirmed read" + ); + assert!( + failing_check_started >= latest_start - interval, + "phase {phase:?}: the failing check started {failing_check_started:?} after \ + the confirmed read, a whole interval early" + ); + } +} + +/// A stream alone in its org re-reads the org on each of its checks, even +/// when a check starts a little sooner than one interval after the previous +/// read began. +#[tokio::test] +async fn lone_stream_rereads_its_org_on_every_check() { + if !binding_fake_test_process() { + return; + } + let mut state = crate::tests::test_data_plane_state().await; + tokio::time::pause(); + state.revocation_check_interval = + Duration::from_millis(crate::DEFAULT_REVOCATION_CHECK_INTERVAL_MS); + let interval = state.revocation_check_interval; + let session_id = DataPlaneSessionId::new_for_kind(SessionKind::Process); + let org_id = OrgId::new(); + state + .revocation_cache + .remember_active(session_id, 1, org_id, 1, start_revocation_read(&state)); + for early in [0, 1, 50, 500].map(Duration::from_millis) { + tokio::time::advance(interval - early).await; + let RevocationProbe::RefreshOrg(refresh) = state + .revocation_cache + .probe(session_id, revocation_tolerance(&state)) + else { + panic!("a check {early:?} early answered from the cache"); + }; + state + .revocation_cache + .remember_org_epoch(org_id, 1, start_revocation_read(&state)); + drop(refresh.guard); + } +} + +/// The org's confirmation is dated from when each read that confirmed it +/// began, whether a route resolve or a revocation check, not from when its +/// slow answer arrived. +#[tokio::test] +async fn revocation_verdicts_are_dated_from_when_their_read_began() { + if !binding_fake_test_process() { + return; + } + let session_id = DataPlaneSessionId::new_for_kind(SessionKind::Process); + let fake = spawn_revocation_fake(session_id).await; + let mut state = crate::tests::test_data_plane_state().await; + state.cluster_api = ClusterApiClients::service_only( + pplx_space_node_api_client::cluster::connect(fake.endpoint.clone(), "test") + .await + .unwrap(), + ); + // Every answer is due for a refresh by the time the next read starts. + state.revocation_check_interval = Duration::from_millis(1); + fake.answer(FakeRevocation::Slow); + let assert_dated_from = |read: &str, began: tokio::time::Instant, state: &DataPlaneState| { + let org_id = state + .revocation_cache + .entries + .get(&session_id.to_string()) + .unwrap() + .org_id; + let confirmed_at = state + .revocation_cache + .orgs + .get(&org_id) + .unwrap() + .confirmed_at; + assert!( + confirmed_at >= began && confirmed_at < began + FAKE_SLOW_ANSWER, + "{read}: confirmed {:?} after the read began", + confirmed_at - began + ); + }; + + let token = format!("{DATA_PLANE_TOKEN_PREFIX}{}", "c".repeat(64)); + let began = tokio::time::Instant::now(); + resolve_session_route(&state, session_id, &token) + .await + .unwrap(); + assert_dated_from("route resolve", began, &state); + + let began = tokio::time::Instant::now(); + resolve_session_route_internal(&state, session_id) + .await + .unwrap(); + assert_dated_from("internal route resolve", began, &state); + + let began = tokio::time::Instant::now(); + assert!(!session_is_revoked(&state, session_id).await.unwrap()); + assert_dated_from("org check", began, &state); + + state.revocation_cache.forget(session_id); + let began = tokio::time::Instant::now(); + assert!(!session_is_revoked(&state, session_id).await.unwrap()); + assert_dated_from("session check", began, &state); +} + +/// A check that reads the org and then the session gets one timeout for +/// both reads, not one each. +#[tokio::test] +async fn revocation_check_has_one_timeout_across_its_reads() { + if !binding_fake_test_process() { + return; + } + let session_id = DataPlaneSessionId::new_for_kind(SessionKind::Process); + let fake = spawn_revocation_fake(session_id).await; + let mut state = crate::tests::test_data_plane_state().await; + state.cluster_api = ClusterApiClients::service_only( + pplx_space_node_api_client::cluster::connect(fake.endpoint.clone(), "test") + .await + .unwrap(), + ); + state.revocation_check_timeout = SPLIT_CHECK_TIMEOUT; + let token = format!("{DATA_PLANE_TOKEN_PREFIX}{}", "c".repeat(64)); + resolve_session_route(&state, session_id, &token) + .await + .unwrap(); + + fake.answer(FakeRevocation::SlowOrgThenHungSession); + make_org_refresh_due(&state, session_id); + let started = tokio::time::Instant::now(); + let error = session_is_revoked(&state, session_id) + .await + .expect_err("the org moved on, so the hung session read fails closed"); + let took = started.elapsed(); + assert!( + matches!(&error, StreamBridgeError::RevocationStatus(status) + if status.code() == tonic::Code::DeadlineExceeded), + "{error}" + ); + // Halfway between one timeout and the org read plus a second timeout. + assert!( + took < SPLIT_CHECK_TIMEOUT + SPLIT_CHECK_ORG_READ / 2, + "the check took {took:?}" + ); +} + +#[tokio::test] +async fn route_cache_hit_does_not_confirm_the_revocation_verdict() { + if !binding_fake_test_process() { + return; + } + let session_id = DataPlaneSessionId::new_for_kind(SessionKind::Process); + let fake = spawn_revocation_fake(session_id).await; + let mut state = crate::tests::test_data_plane_state().await; + state.cluster_api = ClusterApiClients::service_only( + pplx_space_node_api_client::cluster::connect(fake.endpoint.clone(), "test") + .await + .unwrap(), + ); + let interval = Duration::from_millis(crate::DEFAULT_REVOCATION_CHECK_INTERVAL_MS); + state.revocation_check_interval = interval; + state.route_cache_ttl = 10 * REVOCATION_MAX_STALENESS; + let token = format!("{DATA_PLANE_TOKEN_PREFIX}{}", "c".repeat(64)); + resolve_session_route(&state, session_id, &token) + .await + .unwrap(); + let tolerance = Duration::from_secs(1); + assert!( + state + .revocation_cache + .confirmed_active(session_id, tolerance) + ); + + // Past the tolerance, an attach answered from the route cache asked the + // cluster API nothing, so the verdict is as old as it was. Nothing below + // reaches the network, so the paused clock moves only as told. + tokio::time::pause(); + tokio::time::advance(2 * tolerance).await; + state.revocation_cache.forget(session_id); + resolve_session_route(&state, session_id, &token) + .await + .unwrap(); + assert!( + !state + .revocation_cache + .confirmed_active(session_id, tolerance) + ); + assert!(matches!( + state + .revocation_cache + .probe(session_id, revocation_tolerance(&state)), + RevocationProbe::Fresh(true) + )); + // The resolve set the next refresh just under one interval after it + // began; the cache hit must not have moved it later. This lands a + // tolerance past that refresh, and a tolerance short of where one set by + // the cache hit would fall. + let fresh_for = interval - interval / 10; + tokio::time::advance(fresh_for - tolerance).await; + assert!( + matches!( + state + .revocation_cache + .probe(session_id, revocation_tolerance(&state)), + RevocationProbe::RefreshOrg(_) + ), + "the cache hit pushed the org's next refresh out" + ); +} + +/// How the fake backend fails a native WebSocket session. +#[derive(Clone, Copy)] +enum BackendFailure { + /// The backend stream errors after its first event. + MidStream, + /// The stream stays open, and a terminal resize is refused. + ResizeRefused, +} + +/// Backend status text naming an internal endpoint, which must stay out of +/// public frames. +const BACKEND_FAILURE_DETAIL: &str = "dial 192.0.2.10:9443: connection refused"; + +/// A fake cluster API and worker for one native session that fail it as +/// `failure` says. Returns its endpoint. +async fn spawn_backend_failure_fake( + session_id: DataPlaneSessionId, + failure: BackendFailure, +) -> String { + use prost::Message as _; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let endpoint = format!("http://{}", listener.local_addr().unwrap()); + let mut route = crate::tests::route_response_with_token(session_id, None); + route.backend.as_mut().unwrap().endpoint = endpoint.clone(); + route.adapter_state_json = match session_id.kind() { + SessionKind::FileWatch => r#"{"path":"/tmp"}"#, + _ => r#"{"command":["sleep","600"]}"#, + } + .to_string(); + let route = route.encode_to_vec(); + let process_id = pplx_space_public_api::process_id_for_session(session_id).to_string(); + let service = axum::Router::new().fallback(move |request: Request| { + let route = route.clone(); + let process_id = process_id.clone(); + async move { + let path = request.uri().path().to_string(); + if path.ends_with("/Check") { + return crate::tests::grpc_test_response(vec![8, 1]); + } + if path.ends_with("/ResolveDataPlaneSessionRoute") { + return crate::tests::grpc_test_response(route); + } + if path.ends_with("/BindDataPlaneSessionProcess") { + return crate::tests::grpc_test_response(Vec::new()); + } + if path.ends_with("/SpacedResizePty") { + return tonic::Status::failed_precondition(BACKEND_FAILURE_DETAIL) + .into_http::(); + } + let first = if path.ends_with("/SpacedWatch") { + grpc_message(&FileEvent { + path: "/tmp/changed".to_string(), + kind: SpacedFileEventKind::Modified as i32, + ..Default::default() + }) + } else { + assert!( + path.ends_with("/SpacedStartProcess"), + "unexpected call {path}" + ); + grpc_message(&ProcessEvent { + process_id, + pid: Some(42), + event: Some(spaced_process_event::Event::Started( + pplx_space_sbx_mgr_client::operations::ProcessStarted { start_time: None }, + )), + }) + }; + let mut trailers = HeaderMap::new(); + trailers.insert("grpc-status", HeaderValue::from_static("13")); + trailers.insert( + "grpc-message", + HeaderValue::from_static(BACKEND_FAILURE_DETAIL), + ); + let end = match failure { + BackendFailure::MidStream => Some(hyper::body::Frame::trailers(trailers)), + BackendFailure::ResizeRefused => None, + }; + let frames = futures_util::stream::iter( + [Some(hyper::body::Frame::data(first)), end] + .into_iter() + .flatten() + .map(Ok::<_, Infallible>), + ) + .chain(futures_util::stream::pending()); + Response::builder() + .header("content-type", "application/grpc") + .body(Body::new(http_body_util::StreamBody::new(frames))) + .unwrap() + } + }); + tokio::spawn(async move { + axum::serve(listener, service).await.unwrap(); + }); + endpoint +} + +/// Serves the data plane against the fake at `endpoint` and attaches a native +/// WebSocket to the session, returning the upgraded connection. +async fn attach_native_websocket( + endpoint: &str, + session_id: DataPlaneSessionId, +) -> tokio::net::TcpStream { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let mut state = crate::tests::test_data_plane_state().await; + state.cluster_api = ClusterApiClients::service_only( + pplx_space_node_api_client::cluster::connect(endpoint.to_string(), "test") + .await + .unwrap(), + ); + state.revocation_check_interval = Duration::from_secs(3600); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve( + listener, + router(state).into_make_service_with_connect_info::(), + ) + .await + .unwrap(); + }); + let mut tcp = tokio::net::TcpStream::connect(addr).await.unwrap(); + tcp.write_all( + format!( + "GET /v1/sessions/{session_id}/attach HTTP/1.1\r\n\ + Host: data-plane.test\r\n\ + Authorization: Bearer {DATA_PLANE_TOKEN_PREFIX}{}\r\n\ + Connection: Upgrade\r\n\ + Upgrade: websocket\r\n\ + Sec-WebSocket-Version: 13\r\n\ + Sec-WebSocket-Key: {}\r\n\r\n", + "c".repeat(64), + BASE64_STANDARD.encode(b"native-test-nonc"), + ) + .as_bytes(), + ) + .await + .unwrap(); + let mut response = Vec::new(); + while !response.ends_with(b"\r\n\r\n") { + let mut byte = [0_u8; 1]; + tcp.read_exact(&mut byte).await.expect("upgrade response"); + response.push(byte[0]); + } + assert!(response.starts_with(b"HTTP/1.1 101")); + tcp +} + +/// Reads server text frames into `seen` until one contains `marker`. +async fn read_websocket_until(tcp: &mut tokio::net::TcpStream, seen: &mut String, marker: &str) { + tokio::time::timeout(Duration::from_secs(5), async { + while !seen.contains(marker) { + let (opcode, payload) = crate::tests::read_server_frame(tcp).await; + assert_eq!(opcode, 0x1, "text frames until {marker}; saw {seen}"); + seen.push_str(&String::from_utf8_lossy(&payload)); + } + }) + .await + .unwrap_or_else(|_| panic!("{marker} within 5s; saw {seen}")); +} + +/// Sends one short client text frame. Client frames must be masked; an +/// all-zero key leaves the payload as written. +async fn send_websocket_text(tcp: &mut tokio::net::TcpStream, text: &str) { + use tokio::io::AsyncWriteExt; + let len = u8::try_from(text.len()) + .ok() + .filter(|len| *len < 126) + .unwrap(); + let mut frame = vec![0x81, 0x80 | len, 0, 0, 0, 0]; + frame.extend_from_slice(text.as_bytes()); + tcp.write_all(&frame).await.unwrap(); +} + +#[tokio::test] +async fn websocket_backend_stream_errors_carry_a_fixed_message() { + if !binding_fake_test_process() { + return; + } + for (kind, live, message) in [ + ( + SessionKind::Process, + "process_started", + PROCESS_STREAM_FAILED, + ), + ( + SessionKind::FileWatch, + "file_event", + FILE_WATCH_STREAM_FAILED, + ), + ] { + let session_id = DataPlaneSessionId::new_for_kind(kind); + let endpoint = spawn_backend_failure_fake(session_id, BackendFailure::MidStream).await; + let mut tcp = attach_native_websocket(&endpoint, session_id).await; + let mut seen = String::new(); + read_websocket_until(&mut tcp, &mut seen, live).await; + read_websocket_until(&mut tcp, &mut seen, "backend_stream_error").await; + assert!( + seen.contains(&format!(r#""message":"{message}""#)), + "{kind:?}: {seen}" + ); + assert!( + !seen.contains("192.0.2.10"), + "{kind:?}: internal detail leaked: {seen}" + ); + } +} + +#[tokio::test] +async fn websocket_resize_refusal_carries_a_fixed_message() { + if !binding_fake_test_process() { + return; + } + let session_id = DataPlaneSessionId::new_for_kind(SessionKind::Process); + let endpoint = spawn_backend_failure_fake(session_id, BackendFailure::ResizeRefused).await; + let mut tcp = attach_native_websocket(&endpoint, session_id).await; + let mut seen = String::new(); + read_websocket_until(&mut tcp, &mut seen, "process_started").await; + send_websocket_text(&mut tcp, r#"{"type":"resize","cols":80,"rows":24}"#).await; + read_websocket_until(&mut tcp, &mut seen, "resize_not_supported").await; + assert!( + seen.contains(r#""message":"terminal resize is not supported by this process""#), + "{seen}" + ); + assert!( + !seen.contains("192.0.2.10"), + "internal detail leaked: {seen}" + ); +} diff --git a/crates/libs/data/plane/server/src/native/websocket.rs b/crates/libs/data/plane/server/src/native/websocket.rs index c0f135c793..2cd23832f6 100644 --- a/crates/libs/data/plane/server/src/native/websocket.rs +++ b/crates/libs/data/plane/server/src/native/websocket.rs @@ -660,7 +660,7 @@ pub(crate) async fn bridge_process_session( &mut sink, &DataPlaneServerFrame::Error { code: DataPlaneErrorCode::BackendStreamError, - message: status.message().to_string(), + message: PROCESS_STREAM_FAILED.to_string(), retryable: backend_stream_status_retryable(&status), }, ) @@ -928,7 +928,7 @@ pub(crate) async fn bridge_file_watch_session( &mut sink, &DataPlaneServerFrame::Error { code: DataPlaneErrorCode::BackendStreamError, - message: status.message().to_string(), + message: FILE_WATCH_STREAM_FAILED.to_string(), retryable: backend_stream_status_retryable(&status), }, ) @@ -1104,11 +1104,12 @@ where Err(StreamBridgeError::BackendStatus(status)) if status.code() == tonic::Code::FailedPrecondition => { + tracing::debug!(grpc_message = %status.message(), "terminal resize was refused"); send_server_frame( sink, &DataPlaneServerFrame::Error { code: DataPlaneErrorCode::ResizeNotSupported, - message: status.message().to_string(), + message: RESIZE_REFUSED.to_string(), retryable: false, }, ) diff --git a/crates/libs/data/plane/server/src/state.rs b/crates/libs/data/plane/server/src/state.rs index 6dcc8f1393..9e6df714b0 100644 --- a/crates/libs/data/plane/server/src/state.rs +++ b/crates/libs/data/plane/server/src/state.rs @@ -497,9 +497,9 @@ impl RevocationCache { epoch: u64, org_id: OrgId, org_epoch: u64, - refresh_after: Duration, + read: RevocationRead, ) { - self.remember_state(session_id, epoch, true, org_id, org_epoch, refresh_after); + self.remember_state(session_id, epoch, true, org_id, org_epoch, read); } pub(super) fn remember( @@ -507,7 +507,7 @@ impl RevocationCache { session_id: DataPlaneSessionId, org_id: OrgId, revocation: &DataPlaneRevocationState, - refresh_after: Duration, + read: RevocationRead, ) { self.remember_state( session_id, @@ -515,7 +515,7 @@ impl RevocationCache { revocation_state_is_active(revocation), org_id, revocation.org_epoch, - refresh_after, + read, ); } @@ -526,13 +526,41 @@ impl RevocationCache { active: bool, org_id: OrgId, org_epoch: u64, - refresh_after: Duration, + read: RevocationRead, ) { + if self.remember_session(session_id, epoch, active, org_id, org_epoch) { + self.remember_org_epoch(org_id, org_epoch, read); + } + } + + /// Record an active session from a route this replica resolved earlier + /// and still caches. The cluster API was not asked, so the org's last + /// confirmation and next refresh stay as the resolve left them. + pub(super) fn remember_cached_active( + &self, + session_id: DataPlaneSessionId, + epoch: u64, + org_id: OrgId, + org_epoch: u64, + ) { + self.remember_session(session_id, epoch, true, org_id, org_epoch); + } + + /// Record the session's verdict unless a newer epoch is already cached; + /// returns whether it was recorded. + fn remember_session( + &self, + session_id: DataPlaneSessionId, + epoch: u64, + active: bool, + org_id: OrgId, + org_epoch: u64, + ) -> bool { let key = session_id.to_string(); if let Some(existing) = self.entries.get(&key) && epoch < existing.epoch { - return; + return false; } self.entries.insert( key, @@ -543,26 +571,45 @@ impl RevocationCache { org_epoch, }, ); - self.remember_org_epoch(org_id, org_epoch, refresh_after); + true } - pub(super) fn remember_org_epoch(&self, org_id: OrgId, epoch: u64, refresh_after: Duration) { - if let Some(existing) = self.orgs.get(&org_id) - && epoch < existing.epoch - { - return; + /// Record the org epoch `read` returned. An older epoch than the cached + /// one is ignored, and the same epoch only moves its confirmation and next + /// refresh later, so an answer that arrives late cannot undo a newer one. + pub(super) fn remember_org_epoch(&self, org_id: OrgId, epoch: u64, read: RevocationRead) { + let refresh_after = read.started_at + read.fresh_for; + match self.orgs.entry(org_id) { + dashmap::mapref::entry::Entry::Occupied(mut current) => { + let current = current.get_mut(); + if epoch < current.epoch + || (epoch == current.epoch && read.started_at < current.confirmed_at) + { + return; + } + current.epoch = epoch; + current.confirmed_at = read.started_at; + current.refresh_after = refresh_after; + } + dashmap::mapref::entry::Entry::Vacant(slot) => { + slot.insert(OrgRevocationCacheEntry { + epoch, + confirmed_at: read.started_at, + refresh_after, + refreshing: Arc::new(AtomicBool::new(false)), + }); + } } - self.orgs.insert( - org_id, - OrgRevocationCacheEntry { - epoch, - refresh_after: tokio::time::Instant::now() + refresh_after, - refreshing: Arc::new(AtomicBool::new(false)), - }, - ); } - pub(super) fn probe(self: &Arc, session_id: DataPlaneSessionId) -> RevocationProbe { + /// Probe the session's cached verdict. A refresh another stream has in + /// flight answers from the cache only while that verdict is younger than + /// `tolerance`; past it, this session is re-read instead. + pub(super) fn probe( + self: &Arc, + session_id: DataPlaneSessionId, + tolerance: Duration, + ) -> RevocationProbe { let key = session_id.to_string(); let Some(entry) = self.entries.get(&key) else { return RevocationProbe::RefreshSession; @@ -589,6 +636,7 @@ impl RevocationCache { return RevocationProbe::Fresh(active); } let cached_org_epoch = org.epoch.max(session_org_epoch); + let confirmed_recently = org.confirmed_at.elapsed() < tolerance; let refreshing = Arc::clone(&org.refreshing); drop(org); if refreshing @@ -603,11 +651,33 @@ impl RevocationCache { refreshing: Some(refreshing), }, }) - } else { + } else if confirmed_recently { RevocationProbe::RefreshInFlight(active) + } else { + RevocationProbe::RefreshSession } } + /// Whether the session's last confirmed verdict is active and younger + /// than `tolerance`. The verdict's age is its org entry's: every + /// revocation advances the org epoch, so a check that confirms the org's + /// epoch unchanged also confirms each session read under that epoch. A + /// newer org epoch than the session was read under means a revocation in + /// its org since, so the verdict no longer counts. + pub(super) fn confirmed_active( + &self, + session_id: DataPlaneSessionId, + tolerance: Duration, + ) -> bool { + let Some(entry) = self.entries.get(&session_id.to_string()) else { + return false; + }; + let Some(org) = self.orgs.get(&entry.org_id) else { + return false; + }; + entry.active && entry.org_epoch >= org.epoch && org.confirmed_at.elapsed() < tolerance + } + pub(super) fn forget(&self, session_id: DataPlaneSessionId) { self.entries.remove(&session_id.to_string()); } @@ -633,9 +703,30 @@ pub(super) struct RevocationCacheEntry { pub(super) org_epoch: u64, } +/// When a revocation read began, and how long its answer stays fresh. The +/// cluster API may have read its rows at any point before it answered, so an +/// answer is dated from when the read began, not from when it arrived. +#[derive(Debug, Clone, Copy)] +pub(super) struct RevocationRead { + pub(super) started_at: tokio::time::Instant, + pub(super) fresh_for: Duration, +} + +impl RevocationRead { + /// A read beginning now. + pub(super) fn start(fresh_for: Duration) -> Self { + Self { + started_at: tokio::time::Instant::now(), + fresh_for, + } + } +} + #[derive(Debug, Clone)] pub(super) struct OrgRevocationCacheEntry { pub(super) epoch: u64, + /// When the latest read that confirmed this epoch began. + pub(super) confirmed_at: tokio::time::Instant, pub(super) refresh_after: tokio::time::Instant, pub(super) refreshing: Arc, } diff --git a/crates/libs/data/plane/server/src/tests.rs b/crates/libs/data/plane/server/src/tests.rs index a2107975be..3824a3a9d0 100644 --- a/crates/libs/data/plane/server/src/tests.rs +++ b/crates/libs/data/plane/server/src/tests.rs @@ -870,6 +870,36 @@ async fn spawn_cluster_api_stub() -> std::net::SocketAddr { addr } +#[tokio::test] +async fn accepted_connections_detect_dead_peers() { + use axum::serve::Listener; + + let mut listener = + data_plane_listener(tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap()); + let address = listener.local_addr().unwrap(); + let _client = tokio::net::TcpStream::connect(address).await.unwrap(); + let (accepted, _peer) = listener.accept().await; + assert!(accepted.nodelay().unwrap()); + let socket = socket2::SockRef::from(&accepted); + assert!(socket.keepalive().unwrap()); + assert_eq!( + socket.tcp_keepalive_time().unwrap(), + DATA_PLANE_TCP_KEEPALIVE_TIME + ); + assert_eq!( + socket.tcp_keepalive_interval().unwrap(), + DATA_PLANE_TCP_KEEPALIVE_INTERVAL + ); + assert_eq!( + socket.tcp_keepalive_retries().unwrap(), + DATA_PLANE_TCP_KEEPALIVE_RETRIES + ); + // A user timeout would cap how long a client may stop reading, ECW's + // included, and on Linux would decide when keepalive gives up. + #[cfg(target_os = "linux")] + assert_eq!(socket.tcp_user_timeout().unwrap(), None); +} + pub(crate) async fn test_data_plane_state() -> DataPlaneState { let addr = spawn_cluster_api_stub().await; let cluster_api = @@ -891,6 +921,8 @@ pub(crate) async fn test_data_plane_state() -> DataPlaneState { ecw_process_input_locks: Arc::new(EcwProcessInputLocks::default()), ecw_hosts: Arc::default(), revocation_check_interval: Duration::from_secs(1), + revocation_max_staleness: REVOCATION_MAX_STALENESS, + revocation_check_timeout: REVOCATION_CHECK_TIMEOUT, route_cache_ttl: Duration::from_secs(60), drain_started: CancellationToken::new(), stream_shutdown_token: CancellationToken::new(), @@ -1815,6 +1847,41 @@ fn native_session_limiter_allows_one_terminal_or_process_attach() { ); } +/// The tolerance production derives from its default check interval. +fn production_revocation_tolerance() -> Duration { + crate::native::revocation_tolerance_within( + REVOCATION_MAX_STALENESS, + Duration::from_millis(DEFAULT_REVOCATION_CHECK_INTERVAL_MS), + REVOCATION_CHECK_TIMEOUT, + ) +} + +#[tokio::test] +async fn run_refuses_a_revocation_check_interval_above_the_maximum() { + let maximum = Duration::from_millis(MAX_REVOCATION_CHECK_INTERVAL_MS); + let config = |revocation_check_interval| DataPlaneConfig { + revocation_check_interval, + ..DataPlaneConfig::default() + }; + config(maximum).validate().unwrap(); + let above = config(maximum + Duration::from_millis(1)); + assert!(matches!( + above.validate(), + Err(RunDataPlaneError::RevocationCheckIntervalTooLong { .. }) + )); + // Refused before it connects to anything. + let stop = pplx_space_common_svc::StopTokens { + shutdown: CancellationToken::new(), + forced: CancellationToken::new(), + }; + assert!(matches!( + tokio::time::timeout(Duration::from_secs(5), run(above, stop)).await, + Ok(Err( + RunDataPlaneError::RevocationCheckIntervalTooLong { .. } + )) + )); +} + #[test] fn revocation_cache_uses_epoch_and_coalesces_refresh() { let cache = Arc::new(RevocationCache::default()); @@ -1822,41 +1889,67 @@ fn revocation_cache_uses_epoch_and_coalesces_refresh() { let org_id = OrgId::new(); assert!(matches!( - cache.probe(session_id), + cache.probe(session_id, production_revocation_tolerance()), RevocationProbe::RefreshSession )); - cache.remember_active(session_id, 10, org_id, 20, Duration::from_secs(60)); + cache.remember_active( + session_id, + 10, + org_id, + 20, + RevocationRead::start(Duration::from_secs(60)), + ); assert_eq!(cache.cached_epoch(session_id), Some(10)); assert_eq!(cache.cached_org_epoch(org_id), Some(20)); assert!(matches!( - cache.probe(session_id), + cache.probe(session_id, production_revocation_tolerance()), RevocationProbe::Fresh(true) )); - cache.remember_state(session_id, 9, false, org_id, 19, Duration::from_secs(60)); + cache.remember_state( + session_id, + 9, + false, + org_id, + 19, + RevocationRead::start(Duration::from_secs(60)), + ); assert_eq!(cache.cached_epoch(session_id), Some(10)); assert_eq!(cache.cached_org_epoch(org_id), Some(20)); assert!(matches!( - cache.probe(session_id), + cache.probe(session_id, production_revocation_tolerance()), RevocationProbe::Fresh(true) )); - cache.remember_state(session_id, 11, false, org_id, 21, Duration::from_secs(60)); + cache.remember_state( + session_id, + 11, + false, + org_id, + 21, + RevocationRead::start(Duration::from_secs(60)), + ); assert_eq!(cache.cached_epoch(session_id), Some(11)); assert_eq!(cache.cached_org_epoch(org_id), Some(21)); assert!(matches!( - cache.probe(session_id), + cache.probe(session_id, production_revocation_tolerance()), RevocationProbe::Fresh(false) )); - cache.remember_active(session_id, 12, org_id, 22, Duration::from_millis(0)); - let guard = match cache.probe(session_id) { + cache.remember_active( + session_id, + 12, + org_id, + 22, + RevocationRead::start(Duration::from_millis(0)), + ); + let guard = match cache.probe(session_id, production_revocation_tolerance()) { RevocationProbe::RefreshOrg(refresh) => refresh.guard, other => panic!("expected refresh probe for expired entry, got {other:?}"), }; assert!(matches!( - cache.probe(session_id), + cache.probe(session_id, production_revocation_tolerance()), RevocationProbe::RefreshInFlight(true) )); drop(guard); @@ -1871,9 +1964,15 @@ fn revocation_cache_refreshes_session_behind_org_epoch() { let session_id = DataPlaneSessionId::Process(Default::default()); let org_id = OrgId::new(); - cache.remember_active(session_id, 10, org_id, 20, Duration::from_secs(60)); + cache.remember_active( + session_id, + 10, + org_id, + 20, + RevocationRead::start(Duration::from_secs(60)), + ); assert!(matches!( - cache.probe(session_id), + cache.probe(session_id, production_revocation_tolerance()), RevocationProbe::Fresh(true) )); @@ -1881,17 +1980,23 @@ fn revocation_cache_refreshes_session_behind_org_epoch() { // refresh recorded the new epoch before this session's tick. The org // entry's TTL is fresh, but this session's own row was last read under // the old epoch, so it must be re-fetched rather than reported fresh. - cache.remember_org_epoch(org_id, 21, Duration::from_secs(60)); + cache.remember_org_epoch(org_id, 21, RevocationRead::start(Duration::from_secs(60))); assert!(matches!( - cache.probe(session_id), + cache.probe(session_id, production_revocation_tolerance()), RevocationProbe::RefreshSession )); // Once the session row has been re-read under the current epoch the // cache answers from memory again. - cache.remember_active(session_id, 11, org_id, 21, Duration::from_secs(60)); + cache.remember_active( + session_id, + 11, + org_id, + 21, + RevocationRead::start(Duration::from_secs(60)), + ); assert!(matches!( - cache.probe(session_id), + cache.probe(session_id, production_revocation_tolerance()), RevocationProbe::Fresh(true) )); } @@ -1901,9 +2006,15 @@ async fn revocation_cache_coalesces_concurrent_org_refreshes() { let cache = Arc::new(RevocationCache::default()); let session_id = DataPlaneSessionId::Process(Default::default()); let org_id = OrgId::new(); - cache.remember_active(session_id, 10, org_id, 20, Duration::ZERO); + cache.remember_active( + session_id, + 10, + org_id, + 20, + RevocationRead::start(Duration::ZERO), + ); - let guard = match cache.probe(session_id) { + let guard = match cache.probe(session_id, production_revocation_tolerance()) { RevocationProbe::RefreshOrg(refresh) => refresh.guard, other => panic!("expected first probe to own refresh, got {other:?}"), }; @@ -1912,7 +2023,7 @@ async fn revocation_cache_coalesces_concurrent_org_refreshes() { let cache = Arc::clone(&cache); tokio::spawn(async move { matches!( - cache.probe(session_id), + cache.probe(session_id, production_revocation_tolerance()), RevocationProbe::RefreshInFlight(true) ) }) @@ -1926,6 +2037,90 @@ async fn revocation_cache_coalesces_concurrent_org_refreshes() { drop(guard); } +#[test] +fn revocation_cache_keeps_a_confirmed_verdict_only_within_the_staleness_bound() { + let cache = Arc::new(RevocationCache::default()); + let session_id = DataPlaneSessionId::Process(Default::default()); + let org_id = OrgId::new(); + assert!(!cache.confirmed_active(session_id, production_revocation_tolerance())); + + // Confirmed just now, and already due for a refresh. + cache.remember_active( + session_id, + 10, + org_id, + 20, + RevocationRead::start(Duration::ZERO), + ); + assert!(cache.confirmed_active(session_id, production_revocation_tolerance())); + assert!(!cache.confirmed_active(session_id, Duration::ZERO)); + + // While another stream's refresh is in flight the cached verdict answers + // within the bound; past it, this session is re-read. + let _guard = match cache.probe(session_id, production_revocation_tolerance()) { + RevocationProbe::RefreshOrg(refresh) => refresh.guard, + other => panic!("expected the first probe to own the refresh, got {other:?}"), + }; + assert!(matches!( + cache.probe(session_id, production_revocation_tolerance()), + RevocationProbe::RefreshInFlight(true) + )); + assert!(matches!( + cache.probe(session_id, Duration::ZERO), + RevocationProbe::RefreshSession + )); + + // A verdict read under an older org epoch no longer counts, nor does a + // revoked one. + cache.remember_org_epoch(org_id, 21, RevocationRead::start(Duration::from_secs(60))); + assert!(!cache.confirmed_active(session_id, production_revocation_tolerance())); + cache.remember_state( + session_id, + 11, + false, + org_id, + 21, + RevocationRead::start(Duration::from_secs(60)), + ); + assert!(!cache.confirmed_active(session_id, production_revocation_tolerance())); +} + +#[test] +fn revocation_cache_org_confirmation_only_moves_forward() { + let cache = RevocationCache::default(); + let org_id = OrgId::new(); + let fresh_for = Duration::from_secs(60); + let earlier = RevocationRead::start(fresh_for); + let later = RevocationRead { + started_at: earlier.started_at + Duration::from_secs(5), + fresh_for, + }; + let org = |cache: &RevocationCache| { + let entry = cache.orgs.get(&org_id).unwrap(); + (entry.epoch, entry.confirmed_at, entry.refresh_after) + }; + cache.remember_org_epoch(org_id, 20, later); + let confirmed = (20, later.started_at, later.started_at + fresh_for); + assert_eq!(org(&cache), confirmed); + + // A slower read that began first and answers last moves nothing back. + cache.remember_org_epoch(org_id, 20, earlier); + assert_eq!(org(&cache), confirmed); + // Nor does an older epoch, however recent its read. + let latest = RevocationRead { + started_at: later.started_at + Duration::from_secs(5), + fresh_for, + }; + cache.remember_org_epoch(org_id, 19, latest); + assert_eq!(org(&cache), confirmed); + // A newer epoch replaces it, dated from its own read. + cache.remember_org_epoch(org_id, 21, earlier); + assert_eq!( + org(&cache), + (21, earlier.started_at, earlier.started_at + fresh_for) + ); +} + #[test] fn route_cache_expires_and_forgets_native_routes() { let cache = RouteCache::default(); @@ -3954,6 +4149,28 @@ async fn legacy_ingress_eof_delivers_buffered_body_without_completion_frames() { ); } +/// Read one server WebSocket frame header from `tcp` and return its opcode and +/// payload. Server frames are never masked, so the header is two bytes plus an +/// optional extended length. +pub(crate) async fn read_server_frame(tcp: &mut tokio::net::TcpStream) -> (u8, Vec) { + let mut header = [0_u8; 2]; + tcp.read_exact(&mut header).await.expect("frame header"); + let opcode = header[0] & 0x0f; + let len = match header[1] & 0x7f { + 126 => { + let mut extended = [0_u8; 2]; + tcp.read_exact(&mut extended) + .await + .expect("extended length"); + usize::from(u16::from_be_bytes(extended)) + } + len => usize::from(len), + }; + let mut payload = vec![0_u8; len]; + tcp.read_exact(&mut payload).await.expect("frame payload"); + (opcode, payload) +} + pub(crate) fn grpc_test_response(bytes: Vec) -> Response { let mut message = vec![0]; message.extend_from_slice(&(bytes.len() as u32).to_be_bytes()); diff --git a/crates/libs/metrics/src/data_plane.rs b/crates/libs/metrics/src/data_plane.rs index aa2e0c9208..3acf86f0af 100644 --- a/crates/libs/metrics/src/data_plane.rs +++ b/crates/libs/metrics/src/data_plane.rs @@ -508,6 +508,11 @@ pub fn revocation_cache_probe(kind: &str, result: &str) { ); } +/// Counts one revocation check outcome. A check can record more than one: an +/// org read that finds the epoch changed records `org_epoch_changed` and then +/// the session read's result, and a failed check that keeps the last verdict +/// records `status_error` or `timeout` and then `kept_last_verdict`. The sum +/// over results is therefore not a count of checks. pub fn revocation_check(kind: &str, result: &str) { counter( "data_plane.revocation.check",