server.rs (32882B)
1 //! Bounded HTTP/1.1 operations server with an exact passive route inventory. 2 3 use core::fmt; 4 use std::convert::Infallible; 5 use std::error::Error; 6 use std::io; 7 use std::net::SocketAddr; 8 use std::pin::Pin; 9 use std::sync::Arc; 10 use std::task::{Context, Poll}; 11 12 use bytes::Bytes; 13 use http::header::{CACHE_CONTROL, CONTENT_TYPE, HeaderValue}; 14 use http::{Method, Request, Response, StatusCode, Version}; 15 use http_body_util::Full; 16 use hyper::body::Incoming; 17 use hyper::server::conn::http1; 18 use hyper::service::service_fn; 19 use hyper_util::rt::{TokioIo, TokioTimer}; 20 use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf}; 21 use tokio::net::{TcpListener, TcpStream}; 22 use tokio::sync::Semaphore; 23 use tokio::task::JoinSet; 24 25 use super::{ 26 BoundedMetricsSnapshot, LIVEZ_PATH, METRICS_CONTENT_TYPE, OPERATIONS_HEALTH_CONTENT_TYPE, 27 OperationsListenerConfig, OperationsTransportLimits, READYZ_PATH, livez, readyz, 28 }; 29 use crate::{CachedServiceStateReader, CancellationToken, MonotonicClock, SystemMonotonicClock}; 30 31 pub const METRICS_PATH: &str = "/metrics"; 32 33 pub const OPERATIONS_HTTP_MIN_HEADER_BYTES: u32 = 8 * 1024; 34 const NOT_FOUND_BODY: &[u8] = b"not found\n"; 35 const VERSION_UNSUPPORTED_BODY: &[u8] = b"HTTP/1.1 required\n"; 36 const HEADERS_TOO_LARGE_BODY: &[u8] = b"request headers too large\n"; 37 const METRICS_UNAVAILABLE_BODY: &[u8] = b"metrics unavailable\n"; 38 const REQUEST_TIMEOUT_BODY: &[u8] = b"request timeout\n"; 39 const NO_STORE: HeaderValue = HeaderValue::from_static("no-store"); 40 41 /// Safe runtime failure for the operations server. 42 #[derive(Clone, Copy, Debug, PartialEq, Eq)] 43 pub enum OperationsServerError { 44 Disabled, 45 HeaderLimitBelowParserFloor, 46 Bind { kind: io::ErrorKind }, 47 LocalAddress { kind: io::ErrorKind }, 48 Accept { kind: io::ErrorKind }, 49 ConnectionTaskPanicked, 50 } 51 52 impl fmt::Display for OperationsServerError { 53 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { 54 formatter.write_str("bounded operations server failed") 55 } 56 } 57 58 impl Error for OperationsServerError {} 59 60 struct OperationsServerState { 61 cache: CachedServiceStateReader<BoundedMetricsSnapshot>, 62 limits: OperationsTransportLimits, 63 clock: Arc<dyn MonotonicClock>, 64 } 65 66 /// An unbound operations server with no route-registration extension point. 67 pub struct OperationsServer { 68 listen: SocketAddr, 69 state: Arc<OperationsServerState>, 70 } 71 72 impl fmt::Debug for OperationsServer { 73 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { 74 formatter 75 .debug_struct("OperationsServer") 76 .field("listen", &"[redacted]") 77 .field("limits", &self.state.limits) 78 .finish() 79 } 80 } 81 82 impl OperationsServer { 83 pub fn new( 84 config: OperationsListenerConfig, 85 cache: CachedServiceStateReader<BoundedMetricsSnapshot>, 86 ) -> Result<Self, OperationsServerError> { 87 Self::new_with_clock(config, cache, SystemMonotonicClock::new()) 88 } 89 90 pub fn new_with_clock<C>( 91 config: OperationsListenerConfig, 92 cache: CachedServiceStateReader<BoundedMetricsSnapshot>, 93 clock: C, 94 ) -> Result<Self, OperationsServerError> 95 where 96 C: MonotonicClock + 'static, 97 { 98 let listen = config 99 .listen() 100 .ok_or(OperationsServerError::Disabled)? 101 .socket_addr(); 102 let limits = config.limits().ok_or(OperationsServerError::Disabled)?; 103 if limits.header_bytes() < OPERATIONS_HTTP_MIN_HEADER_BYTES { 104 return Err(OperationsServerError::HeaderLimitBelowParserFloor); 105 } 106 Ok(Self { 107 listen, 108 state: Arc::new(OperationsServerState { 109 cache, 110 limits, 111 clock: Arc::new(clock), 112 }), 113 }) 114 } 115 116 /// Binds the exact validated socket address without starting admission. 117 pub async fn bind(self) -> Result<BoundOperationsServer, OperationsServerError> { 118 let listener = TcpListener::bind(self.listen) 119 .await 120 .map_err(|error| OperationsServerError::Bind { kind: error.kind() })?; 121 let local_address = listener 122 .local_addr() 123 .map_err(|error| OperationsServerError::LocalAddress { kind: error.kind() })?; 124 Ok(BoundOperationsServer { 125 listener, 126 local_address, 127 state: self.state, 128 }) 129 } 130 } 131 132 /// One successfully bound listener ready for explicit cancellation-owned service. 133 pub struct BoundOperationsServer { 134 listener: TcpListener, 135 local_address: SocketAddr, 136 state: Arc<OperationsServerState>, 137 } 138 139 impl fmt::Debug for BoundOperationsServer { 140 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { 141 formatter 142 .debug_struct("BoundOperationsServer") 143 .field("local_address", &"[redacted]") 144 .field("limits", &self.state.limits) 145 .finish() 146 } 147 } 148 149 impl BoundOperationsServer { 150 #[must_use] 151 pub const fn local_address(&self) -> SocketAddr { 152 self.local_address 153 } 154 155 /// Stops admission on cancellation and drains every bounded connection task. 156 pub async fn serve(self, cancellation: CancellationToken) -> Result<(), OperationsServerError> { 157 let permits = Arc::new(Semaphore::new( 158 self.state.limits.concurrent_connections() as usize 159 )); 160 let mut tasks = JoinSet::new(); 161 162 let mut result = loop { 163 tokio::select! { 164 biased; 165 () = cancellation.cancelled() => break Ok(()), 166 joined = tasks.join_next(), if !tasks.is_empty() => { 167 if joined.is_some_and(|result| result.is_err()) { 168 cancellation.cancel(); 169 break Err(OperationsServerError::ConnectionTaskPanicked); 170 } 171 } 172 accepted = self.listener.accept() => { 173 let (stream, _) = match accepted { 174 Ok(accepted) => accepted, 175 Err(error) => { 176 cancellation.cancel(); 177 break Err(OperationsServerError::Accept { kind: error.kind() }); 178 } 179 }; 180 let Ok(permit) = Arc::clone(&permits).try_acquire_owned() else { 181 drop(stream); 182 continue; 183 }; 184 let state = Arc::clone(&self.state); 185 let connection_cancellation = cancellation.clone(); 186 tasks.spawn(async move { 187 let _permit = permit; 188 serve_connection(stream, state, connection_cancellation).await; 189 }); 190 } 191 } 192 }; 193 194 while let Some(joined) = tasks.join_next().await { 195 if joined.is_err() { 196 result = Err(OperationsServerError::ConnectionTaskPanicked); 197 } 198 } 199 result 200 } 201 } 202 203 async fn serve_connection( 204 mut stream: TcpStream, 205 state: Arc<OperationsServerState>, 206 cancellation: CancellationToken, 207 ) { 208 let admitted = tokio::select! { 209 biased; 210 () = cancellation.cancelled() => return, 211 admitted = tokio::time::timeout( 212 state.limits.idle_timeout(), 213 read_request_head(&mut stream, state.limits.header_bytes() as usize), 214 ) => admitted, 215 }; 216 let prefix = match admitted { 217 Ok(Ok(prefix)) => prefix, 218 Ok(Err(RequestHeadError::TooLarge)) => { 219 let _ = tokio::time::timeout( 220 state.limits.request_deadline(), 221 write_header_limit_response(&mut stream, state.limits), 222 ) 223 .await; 224 return; 225 } 226 Ok(Err(RequestHeadError::Incomplete | RequestHeadError::Read)) | Err(_) => return, 227 }; 228 229 let service_state = Arc::clone(&state); 230 let service = service_fn(move |request| { 231 let state = Arc::clone(&service_state); 232 async move { Ok::<_, Infallible>(serve_request(request, state).await) } 233 }); 234 235 let mut builder = http1::Builder::new(); 236 builder 237 .keep_alive(false) 238 .auto_date_header(false) 239 .max_headers(state.limits.header_count() as usize) 240 .max_buf_size(state.limits.header_bytes() as usize) 241 .header_read_timeout(state.limits.idle_timeout()) 242 .timer(TokioTimer::new()); 243 244 let connection_deadline = state 245 .limits 246 .request_deadline() 247 .saturating_add(state.limits.idle_timeout()); 248 let stream = PrefixedTcpStream::new(prefix, stream); 249 let mut connection = Box::pin(builder.serve_connection(TokioIo::new(stream), service)); 250 tokio::select! { 251 biased; 252 () = cancellation.cancelled() => { 253 connection.as_mut().graceful_shutdown(); 254 let _ = tokio::time::timeout(connection_deadline, connection).await; 255 } 256 _ = tokio::time::timeout(connection_deadline, &mut connection) => {} 257 } 258 } 259 260 async fn serve_request( 261 request: Request<Incoming>, 262 state: Arc<OperationsServerState>, 263 ) -> Response<Full<Bytes>> { 264 let deadline = match state.clock.deadline_after(state.limits.request_deadline()) { 265 Ok(deadline) => deadline, 266 Err(_) => { 267 return fixed_response( 268 StatusCode::GATEWAY_TIMEOUT, 269 OPERATIONS_HEALTH_CONTENT_TYPE, 270 REQUEST_TIMEOUT_BODY, 271 state.limits, 272 ); 273 } 274 }; 275 if deadline.is_reached_at(state.clock.now_monotonic()) { 276 return fixed_response( 277 StatusCode::GATEWAY_TIMEOUT, 278 OPERATIONS_HEALTH_CONTENT_TYPE, 279 REQUEST_TIMEOUT_BODY, 280 state.limits, 281 ); 282 } 283 let response = process_request(request, Arc::clone(&state)); 284 if deadline.is_reached_at(state.clock.now_monotonic()) { 285 fixed_response( 286 StatusCode::GATEWAY_TIMEOUT, 287 OPERATIONS_HEALTH_CONTENT_TYPE, 288 REQUEST_TIMEOUT_BODY, 289 state.limits, 290 ) 291 } else { 292 response 293 } 294 } 295 296 fn process_request( 297 request: Request<Incoming>, 298 state: Arc<OperationsServerState>, 299 ) -> Response<Full<Bytes>> { 300 if request.version() != Version::HTTP_11 { 301 return fixed_response( 302 StatusCode::HTTP_VERSION_NOT_SUPPORTED, 303 OPERATIONS_HEALTH_CONTENT_TYPE, 304 VERSION_UNSUPPORTED_BODY, 305 state.limits, 306 ); 307 } 308 if request.headers().len() > state.limits.header_count() as usize { 309 return fixed_response( 310 StatusCode::REQUEST_HEADER_FIELDS_TOO_LARGE, 311 OPERATIONS_HEALTH_CONTENT_TYPE, 312 HEADERS_TOO_LARGE_BODY, 313 state.limits, 314 ); 315 } 316 if request.method() != Method::GET || request.uri().query().is_some() { 317 return not_found(state.limits); 318 } 319 320 match request.uri().path() { 321 LIVEZ_PATH => { 322 let response = livez(&state.cache); 323 fixed_response( 324 response.status(), 325 response.content_type(), 326 response.body(), 327 state.limits, 328 ) 329 } 330 READYZ_PATH => { 331 let response = readyz(&state.cache); 332 fixed_response( 333 response.status(), 334 response.content_type(), 335 response.body(), 336 state.limits, 337 ) 338 } 339 METRICS_PATH => { 340 let snapshot = state.cache.snapshot(); 341 match snapshot 342 .metrics() 343 .render(state.limits.response_body_utf8_bytes() as usize) 344 { 345 Ok(body) => response(StatusCode::OK, METRICS_CONTENT_TYPE, body), 346 Err(_) => fixed_response( 347 StatusCode::SERVICE_UNAVAILABLE, 348 OPERATIONS_HEALTH_CONTENT_TYPE, 349 METRICS_UNAVAILABLE_BODY, 350 state.limits, 351 ), 352 } 353 } 354 _ => not_found(state.limits), 355 } 356 } 357 358 fn not_found(limits: OperationsTransportLimits) -> Response<Full<Bytes>> { 359 fixed_response( 360 StatusCode::NOT_FOUND, 361 OPERATIONS_HEALTH_CONTENT_TYPE, 362 NOT_FOUND_BODY, 363 limits, 364 ) 365 } 366 367 fn fixed_response( 368 status: StatusCode, 369 content_type: &'static str, 370 body: &'static [u8], 371 limits: OperationsTransportLimits, 372 ) -> Response<Full<Bytes>> { 373 if body.len() > limits.response_body_utf8_bytes() as usize { 374 return response( 375 StatusCode::SERVICE_UNAVAILABLE, 376 OPERATIONS_HEALTH_CONTENT_TYPE, 377 Vec::new(), 378 ); 379 } 380 response(status, content_type, body.to_vec()) 381 } 382 383 fn response( 384 status: StatusCode, 385 content_type: &'static str, 386 body: Vec<u8>, 387 ) -> Response<Full<Bytes>> { 388 let mut response = Response::new(Full::new(Bytes::from(body))); 389 *response.status_mut() = status; 390 response 391 .headers_mut() 392 .insert(CONTENT_TYPE, HeaderValue::from_static(content_type)); 393 response.headers_mut().insert(CACHE_CONTROL, NO_STORE); 394 response 395 } 396 397 #[derive(Clone, Copy, Debug, PartialEq, Eq)] 398 enum RequestHeadError { 399 TooLarge, 400 Incomplete, 401 Read, 402 } 403 404 async fn read_request_head( 405 stream: &mut TcpStream, 406 maximum: usize, 407 ) -> Result<Vec<u8>, RequestHeadError> { 408 let allocation = maximum.checked_add(1).ok_or(RequestHeadError::TooLarge)?; 409 let mut head = Vec::with_capacity(allocation); 410 let mut scan_from = 0; 411 loop { 412 if let Some(index) = head[scan_from..] 413 .windows(4) 414 .position(|window| window == b"\r\n\r\n") 415 { 416 let end = scan_from + index + 4; 417 return if end <= maximum { 418 Ok(head) 419 } else { 420 Err(RequestHeadError::TooLarge) 421 }; 422 } 423 if head.len() > maximum { 424 return Err(RequestHeadError::TooLarge); 425 } 426 scan_from = head.len().saturating_sub(3); 427 let remaining = allocation.saturating_sub(head.len()); 428 if remaining == 0 { 429 return Err(RequestHeadError::TooLarge); 430 } 431 let mut chunk = [0_u8; 1024]; 432 let chunk_limit = remaining.min(chunk.len()); 433 let read = stream 434 .read(&mut chunk[..chunk_limit]) 435 .await 436 .map_err(|_| RequestHeadError::Read)?; 437 if read == 0 { 438 return Err(RequestHeadError::Incomplete); 439 } 440 head.extend_from_slice(&chunk[..read]); 441 } 442 } 443 444 async fn write_header_limit_response( 445 stream: &mut TcpStream, 446 limits: OperationsTransportLimits, 447 ) -> io::Result<()> { 448 let body = if HEADERS_TOO_LARGE_BODY.len() <= limits.response_body_utf8_bytes() as usize { 449 HEADERS_TOO_LARGE_BODY 450 } else { 451 &[] 452 }; 453 let head = format!( 454 concat!( 455 "HTTP/1.1 431 Request Header Fields Too Large\r\n", 456 "content-type: text/plain; charset=utf-8\r\n", 457 "cache-control: no-store\r\n", 458 "connection: close\r\n", 459 "content-length: {}\r\n\r\n" 460 ), 461 body.len() 462 ); 463 stream.write_all(head.as_bytes()).await?; 464 stream.write_all(body).await?; 465 stream.shutdown().await 466 } 467 468 struct PrefixedTcpStream { 469 prefix: Vec<u8>, 470 offset: usize, 471 stream: TcpStream, 472 } 473 474 impl PrefixedTcpStream { 475 fn new(prefix: Vec<u8>, stream: TcpStream) -> Self { 476 Self { 477 prefix, 478 offset: 0, 479 stream, 480 } 481 } 482 } 483 484 impl AsyncRead for PrefixedTcpStream { 485 fn poll_read( 486 mut self: Pin<&mut Self>, 487 context: &mut Context<'_>, 488 output: &mut ReadBuf<'_>, 489 ) -> Poll<io::Result<()>> { 490 if self.offset < self.prefix.len() { 491 let count = output 492 .remaining() 493 .min(self.prefix.len().saturating_sub(self.offset)); 494 output.put_slice(&self.prefix[self.offset..self.offset + count]); 495 self.offset += count; 496 if self.offset == self.prefix.len() { 497 self.prefix = Vec::new(); 498 self.offset = 0; 499 } 500 Poll::Ready(Ok(())) 501 } else { 502 Pin::new(&mut self.stream).poll_read(context, output) 503 } 504 } 505 } 506 507 impl AsyncWrite for PrefixedTcpStream { 508 #[cfg_attr(coverage_nightly, coverage(off))] 509 fn poll_write( 510 mut self: Pin<&mut Self>, 511 context: &mut Context<'_>, 512 buffer: &[u8], 513 ) -> Poll<Result<usize, io::Error>> { 514 Pin::new(&mut self.stream).poll_write(context, buffer) 515 } 516 517 #[cfg_attr(coverage_nightly, coverage(off))] 518 fn poll_flush( 519 mut self: Pin<&mut Self>, 520 context: &mut Context<'_>, 521 ) -> Poll<Result<(), io::Error>> { 522 Pin::new(&mut self.stream).poll_flush(context) 523 } 524 525 #[cfg_attr(coverage_nightly, coverage(off))] 526 fn poll_shutdown( 527 mut self: Pin<&mut Self>, 528 context: &mut Context<'_>, 529 ) -> Poll<Result<(), io::Error>> { 530 Pin::new(&mut self.stream).poll_shutdown(context) 531 } 532 } 533 534 #[cfg(test)] 535 mod tests { 536 use core::time::Duration; 537 use std::net::{IpAddr, Ipv4Addr, SocketAddr}; 538 use std::sync::atomic::{AtomicUsize, Ordering}; 539 540 use tokio::io::{AsyncReadExt, AsyncWriteExt}; 541 542 use super::*; 543 use crate::{ 544 CachedServiceState, CommonMetricGroup, MetricDescriptor, MetricKind, MetricLabel, 545 MetricLabelKey, MetricName, MetricSample, MetricValue, MonotonicTime, OperationsBindPolicy, 546 OperationsListenAddress, OperationsTransportLimitValues, Readiness, ReasonCodes, 547 ServiceOperationalState, ServicePhase, cached_service_state, 548 }; 549 550 fn limits() -> OperationsTransportLimits { 551 OperationsTransportLimits::new(OperationsTransportLimitValues { 552 header_count: 16, 553 header_bytes: OPERATIONS_HTTP_MIN_HEADER_BYTES, 554 response_body_utf8_bytes: 4096, 555 concurrent_connections: 4, 556 request_deadline: Duration::from_millis(200), 557 idle_timeout: Duration::from_millis(200), 558 }) 559 .unwrap() 560 } 561 562 #[test] 563 fn fixed_response_fails_closed_when_the_configured_body_cap_is_smaller() { 564 let mut values = limits().values(); 565 values.response_body_utf8_bytes = 1; 566 let limits = OperationsTransportLimits::new(values).unwrap(); 567 let response = fixed_response( 568 StatusCode::OK, 569 OPERATIONS_HEALTH_CONTENT_TYPE, 570 b"too large", 571 limits, 572 ); 573 assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); 574 } 575 576 fn snapshot( 577 phase: ServicePhase, 578 readiness: Readiness, 579 ) -> CachedServiceStateReader<BoundedMetricsSnapshot> { 580 let descriptor = MetricDescriptor::new( 581 CommonMetricGroup::Phase, 582 MetricName::new("radroots_phase").unwrap(), 583 "current phase", 584 MetricKind::Gauge, 585 [MetricLabelKey::Phase], 586 ) 587 .unwrap(); 588 let sample = MetricSample::new( 589 MetricName::new("radroots_phase").unwrap(), 590 MetricValue::Gauge(1), 591 [MetricLabel::phase(phase)], 592 ) 593 .unwrap(); 594 let metrics = BoundedMetricsSnapshot::new([descriptor], [sample]).unwrap(); 595 let operational = 596 ServiceOperationalState::new(phase, readiness, ReasonCodes::empty()).unwrap(); 597 cached_service_state(CachedServiceState::new(operational, metrics)).1 598 } 599 600 fn config(address: SocketAddr, limits: OperationsTransportLimits) -> OperationsListenerConfig { 601 OperationsListenerConfig::enabled( 602 OperationsListenAddress::new(address).unwrap(), 603 OperationsBindPolicy::LoopbackOnly, 604 limits, 605 ) 606 .unwrap() 607 } 608 609 async fn bound(limits: OperationsTransportLimits) -> BoundOperationsServer { 610 let address = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 9); 611 OperationsServer::new( 612 config(address, limits), 613 snapshot(ServicePhase::Ready, Readiness::READY), 614 ) 615 .unwrap() 616 .bind_with_ephemeral_port_for_test() 617 .await 618 .unwrap() 619 } 620 621 impl OperationsServer { 622 async fn bind_with_ephemeral_port_for_test( 623 self, 624 ) -> Result<BoundOperationsServer, OperationsServerError> { 625 let listener = TcpListener::bind(SocketAddr::new(self.listen.ip(), 0)) 626 .await 627 .map_err(|error| OperationsServerError::Bind { kind: error.kind() })?; 628 let local_address = listener 629 .local_addr() 630 .map_err(|error| OperationsServerError::LocalAddress { kind: error.kind() })?; 631 Ok(BoundOperationsServer { 632 listener, 633 local_address, 634 state: self.state, 635 }) 636 } 637 } 638 639 async fn raw_request(address: SocketAddr, request: &[u8]) -> Vec<u8> { 640 let mut stream = TcpStream::connect(address).await.unwrap(); 641 stream.write_all(request).await.unwrap(); 642 let mut response = Vec::new(); 643 stream.read_to_end(&mut response).await.unwrap(); 644 response 645 } 646 647 fn response_text(response: &[u8]) -> &str { 648 std::str::from_utf8(response).unwrap() 649 } 650 651 fn padded_request_head(total_bytes: usize) -> Vec<u8> { 652 const PREFIX: &[u8] = b"GET /livez HTTP/1.1\r\nx-pad: "; 653 const SUFFIX: &[u8] = b"\r\n\r\n"; 654 assert!(total_bytes >= PREFIX.len() + SUFFIX.len()); 655 let mut request = Vec::with_capacity(total_bytes); 656 request.extend_from_slice(PREFIX); 657 request.resize(total_bytes - SUFFIX.len(), b'a'); 658 request.extend_from_slice(SUFFIX); 659 assert_eq!(request.len(), total_bytes); 660 request 661 } 662 663 #[tokio::test] 664 async fn serves_only_exact_passive_routes_with_exact_content_types() { 665 let server = bound(limits()).await; 666 let address = server.local_address(); 667 let cancellation = CancellationToken::new(); 668 let serve_cancel = cancellation.clone(); 669 let task = tokio::spawn(server.serve(serve_cancel)); 670 671 let live = raw_request(address, b"GET /livez HTTP/1.1\r\nhost: localhost\r\n\r\n").await; 672 let ready = raw_request(address, b"GET /readyz HTTP/1.1\r\nhost: localhost\r\n\r\n").await; 673 let metrics = 674 raw_request(address, b"GET /metrics HTTP/1.1\r\nhost: localhost\r\n\r\n").await; 675 assert!( 676 response_text(&live).starts_with("HTTP/1.1 200 OK\r\n"), 677 "{}", 678 response_text(&live) 679 ); 680 assert!(response_text(&live).contains("content-type: text/plain; charset=utf-8\r\n")); 681 assert!(response_text(&live).ends_with("live\n")); 682 assert!(response_text(&ready).ends_with("ready\n")); 683 assert!( 684 response_text(&metrics) 685 .contains("content-type: text/plain; version=0.0.4; charset=utf-8\r\n") 686 ); 687 assert!(response_text(&metrics).contains("# TYPE radroots_phase gauge\n")); 688 689 for request in [ 690 &b"GET /status HTTP/1.1\r\nhost: localhost\r\n\r\n"[..], 691 &b"POST /readyz HTTP/1.1\r\nhost: localhost\r\ncontent-length: 0\r\n\r\n"[..], 692 &b"GET /readyz?probe=1 HTTP/1.1\r\nhost: localhost\r\n\r\n"[..], 693 &b"GET /v1/status HTTP/1.1\r\nhost: localhost\r\n\r\n"[..], 694 ] { 695 let rejected = raw_request(address, request).await; 696 assert!(response_text(&rejected).starts_with("HTTP/1.1 404 Not Found\r\n")); 697 assert!(response_text(&rejected).ends_with("not found\n")); 698 } 699 700 cancellation.cancel(); 701 assert_eq!(task.await.unwrap(), Ok(())); 702 } 703 704 #[tokio::test] 705 async fn enforces_http_header_and_metrics_response_limits() { 706 let mut values = limits().values(); 707 values.response_body_utf8_bytes = 32; 708 let server = bound(OperationsTransportLimits::new(values).unwrap()).await; 709 let address = server.local_address(); 710 let cancellation = CancellationToken::new(); 711 let task = tokio::spawn(server.serve(cancellation.clone())); 712 713 let exact_head = padded_request_head(OPERATIONS_HTTP_MIN_HEADER_BYTES as usize); 714 let exact = raw_request(address, &exact_head).await; 715 assert!(response_text(&exact).starts_with("HTTP/1.1 200 OK\r\n")); 716 let over_head = padded_request_head(OPERATIONS_HTTP_MIN_HEADER_BYTES as usize + 1); 717 let headers = raw_request(address, &over_head).await; 718 assert!( 719 response_text(&headers).starts_with("HTTP/1.1 431 Request Header Fields Too Large\r\n"), 720 "{}", 721 response_text(&headers) 722 ); 723 let metrics = raw_request(address, b"GET /metrics HTTP/1.1\r\nh: x\r\n\r\n").await; 724 assert!(response_text(&metrics).starts_with("HTTP/1.1 503 Service Unavailable\r\n")); 725 assert!(response_text(&metrics).ends_with("metrics unavailable\n")); 726 let version = raw_request(address, b"GET /livez HTTP/1.0\r\n\r\n").await; 727 assert!( 728 response_text(&version).starts_with("HTTP/1.0 505 HTTP Version Not Supported\r\n"), 729 "{}", 730 response_text(&version) 731 ); 732 733 cancellation.cancel(); 734 assert_eq!(task.await.unwrap(), Ok(())); 735 } 736 737 #[tokio::test] 738 async fn bind_failure_and_disabled_configuration_are_typed() { 739 assert_eq!( 740 OperationsServer::new( 741 OperationsListenerConfig::disabled(), 742 snapshot(ServicePhase::Ready, Readiness::READY), 743 ) 744 .unwrap_err(), 745 OperationsServerError::Disabled 746 ); 747 748 let mut below_floor = limits().values(); 749 below_floor.header_bytes = OPERATIONS_HTTP_MIN_HEADER_BYTES - 1; 750 let address = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 9); 751 assert_eq!( 752 OperationsServer::new( 753 config( 754 address, 755 OperationsTransportLimits::new(below_floor).unwrap() 756 ), 757 snapshot(ServicePhase::Ready, Readiness::READY), 758 ) 759 .unwrap_err(), 760 OperationsServerError::HeaderLimitBelowParserFloor 761 ); 762 763 let occupied = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap(); 764 let address = occupied.local_addr().unwrap(); 765 let error = OperationsServer::new( 766 config(address, limits()), 767 snapshot(ServicePhase::Ready, Readiness::READY), 768 ) 769 .unwrap() 770 .bind() 771 .await 772 .unwrap_err(); 773 assert_eq!( 774 error, 775 OperationsServerError::Bind { 776 kind: io::ErrorKind::AddrInUse 777 } 778 ); 779 } 780 781 #[tokio::test] 782 async fn cancellation_stops_admission_and_drains_partial_connections() { 783 let mut values = limits().values(); 784 values.request_deadline = Duration::from_millis(20); 785 values.idle_timeout = Duration::from_millis(20); 786 let server = bound(OperationsTransportLimits::new(values).unwrap()).await; 787 let address = server.local_address(); 788 let cancellation = CancellationToken::new(); 789 let task = tokio::spawn(server.serve(cancellation.clone())); 790 let mut partial = TcpStream::connect(address).await.unwrap(); 791 partial.write_all(b"GET /livez HTTP/1.1\r\n").await.unwrap(); 792 793 cancellation.cancel(); 794 assert_eq!( 795 tokio::time::timeout(Duration::from_secs(1), task) 796 .await 797 .unwrap() 798 .unwrap(), 799 Ok(()) 800 ); 801 assert!(TcpStream::connect(address).await.is_err()); 802 } 803 804 #[tokio::test] 805 async fn connection_saturation_sheds_and_recovers_without_queueing() { 806 let mut values = limits().values(); 807 values.concurrent_connections = 1; 808 values.request_deadline = Duration::from_millis(500); 809 values.idle_timeout = Duration::from_millis(500); 810 let server = bound(OperationsTransportLimits::new(values).unwrap()).await; 811 let address = server.local_address(); 812 let cancellation = CancellationToken::new(); 813 let task = tokio::spawn(server.serve(cancellation.clone())); 814 815 let mut occupied = TcpStream::connect(address).await.unwrap(); 816 occupied 817 .write_all(b"GET /livez HTTP/1.1\r\n") 818 .await 819 .unwrap(); 820 tokio::time::sleep(Duration::from_millis(20)).await; 821 822 let mut shed_stream = TcpStream::connect(address).await.unwrap(); 823 shed_stream 824 .write_all(b"GET /livez HTTP/1.1\r\nhost: localhost\r\n\r\n") 825 .await 826 .unwrap(); 827 let mut shed = Vec::new(); 828 let shed_result = shed_stream.read_to_end(&mut shed).await; 829 assert!(shed.is_empty()); 830 assert!( 831 shed_result.is_ok() 832 || shed_result.is_err_and(|error| error.kind() == io::ErrorKind::ConnectionReset) 833 ); 834 drop(occupied); 835 tokio::time::sleep(Duration::from_millis(20)).await; 836 837 let recovered = 838 raw_request(address, b"GET /livez HTTP/1.1\r\nhost: localhost\r\n\r\n").await; 839 assert!(response_text(&recovered).starts_with("HTTP/1.1 200 OK\r\n")); 840 841 cancellation.cancel(); 842 assert_eq!(task.await.unwrap(), Ok(())); 843 } 844 845 struct PostRenderDeadlineClock { 846 calls: AtomicUsize, 847 } 848 849 struct PreRenderDeadlineClock { 850 calls: AtomicUsize, 851 } 852 853 impl MonotonicClock for PreRenderDeadlineClock { 854 fn now_monotonic(&self) -> MonotonicTime { 855 let call = self.calls.fetch_add(1, Ordering::SeqCst); 856 let elapsed = if call == 0 { 857 Duration::ZERO 858 } else { 859 Duration::from_millis(2) 860 }; 861 MonotonicTime::from_duration_since_origin(elapsed) 862 } 863 } 864 865 impl MonotonicClock for PostRenderDeadlineClock { 866 fn now_monotonic(&self) -> MonotonicTime { 867 let call = self.calls.fetch_add(1, Ordering::SeqCst); 868 let elapsed = if call < 2 { 869 Duration::ZERO 870 } else { 871 Duration::from_millis(2) 872 }; 873 MonotonicTime::from_duration_since_origin(elapsed) 874 } 875 } 876 877 #[tokio::test] 878 async fn synchronous_render_cannot_return_success_after_request_deadline() { 879 let mut values = limits().values(); 880 values.request_deadline = Duration::from_millis(1); 881 let limits = OperationsTransportLimits::new(values).unwrap(); 882 let address = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 9); 883 let server = OperationsServer::new_with_clock( 884 config(address, limits), 885 snapshot(ServicePhase::Ready, Readiness::READY), 886 PostRenderDeadlineClock { 887 calls: AtomicUsize::new(0), 888 }, 889 ) 890 .unwrap() 891 .bind_with_ephemeral_port_for_test() 892 .await 893 .unwrap(); 894 let address = server.local_address(); 895 let cancellation = CancellationToken::new(); 896 let task = tokio::spawn(server.serve(cancellation.clone())); 897 898 let response = 899 raw_request(address, b"GET /metrics HTTP/1.1\r\nhost: localhost\r\n\r\n").await; 900 assert!(response_text(&response).starts_with("HTTP/1.1 504 Gateway Timeout\r\n")); 901 assert!(response_text(&response).ends_with("request timeout\n")); 902 903 cancellation.cancel(); 904 assert_eq!(task.await.unwrap(), Ok(())); 905 } 906 907 #[tokio::test] 908 async fn already_reached_request_deadline_never_dispatches_the_route() { 909 let mut values = limits().values(); 910 values.request_deadline = Duration::from_millis(1); 911 let limits = OperationsTransportLimits::new(values).unwrap(); 912 let address = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 9); 913 let server = OperationsServer::new_with_clock( 914 config(address, limits), 915 snapshot(ServicePhase::Ready, Readiness::READY), 916 PreRenderDeadlineClock { 917 calls: AtomicUsize::new(0), 918 }, 919 ) 920 .unwrap() 921 .bind_with_ephemeral_port_for_test() 922 .await 923 .unwrap(); 924 let address = server.local_address(); 925 let cancellation = CancellationToken::new(); 926 let task = tokio::spawn(server.serve(cancellation.clone())); 927 928 let response = 929 raw_request(address, b"GET /livez HTTP/1.1\r\nhost: localhost\r\n\r\n").await; 930 assert!(response_text(&response).starts_with("HTTP/1.1 504 Gateway Timeout\r\n")); 931 assert!(response_text(&response).ends_with("request timeout\n")); 932 933 cancellation.cancel(); 934 assert_eq!(task.await.unwrap(), Ok(())); 935 } 936 }