mini-serve 0.13.12

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

use hyper::StatusCode;
use mini_serve::{RouteBuilder, handler};
use tokio::net::TcpListener;
use tokio::sync::oneshot;

/// Permit acquisition must be raced against the shutdown signal, not awaited
/// bare inside the accept branch — otherwise a fully saturated server ignores
/// SIGTERM indefinitely, under exactly the load where a restart is most needed.
/// With `max_connections = 1`, one
/// long-lived request holds the only permit; a second request queues behind
/// it. Shutdown fires while the second is still queued. A correct
/// implementation drops the queued connection immediately and completes as
/// soon as the first (already in-flight) request finishes — it must not wait
/// for, or go on to serve, the second.
#[tokio::test]
async fn shutdown_is_not_starved_by_a_saturated_semaphore() {
	let (started_tx, started_rx) = oneshot::channel::<()>();
	let (release_tx, release_rx) = oneshot::channel::<()>();
	let started_tx = Arc::new(Mutex::new(Some(started_tx)));
	let release_rx = Arc::new(Mutex::new(Some(release_rx)));
	let fast_invoked = Arc::new(AtomicBool::new(false));
	let fast_invoked_for_handler = fast_invoked.clone();

	let app = RouteBuilder::new(())
		.with_max_connections(1)
		.get("/slow", handler(move |_req, _state| {
			let started_tx = started_tx.clone();
			let release_rx = release_rx.clone();
			async move {
				if let Some(tx) = started_tx.lock().unwrap().take() {
					let _ = tx.send(());
				}
				let rx = release_rx.lock().unwrap().take().expect("/slow hit more than once");
				let _ = rx.await;
				mini_serve::json(StatusCode::OK, &serde_json::json!({"ok": true}))
			}
		}))
		.get("/fast", handler(move |_req, _state| {
			let fast_invoked = fast_invoked_for_handler.clone();
			async move {
				fast_invoked.store(true, Ordering::SeqCst);
				mini_serve::json(StatusCode::OK, &serde_json::json!({"ok": true}))
			}
		}))
		.seal();

	let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
	let addr = listener.local_addr().unwrap();

	let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();
	let mut run_task = tokio::spawn(async move {
		app.run(listener, async move {
			let _ = shutdown_rx.await;
		})
		.await
	});

	// Occupy the only permit with a long-lived request.
	let slow_client = tokio::spawn(async move {
		reqwest::get(format!("http://{addr}/slow")).await
	});
	started_rx.await.expect("/slow handler never started");

	// This one queues behind the saturated semaphore, blocked waiting for a
	// permit that will not free until /slow is released below.
	let fast_client = tokio::spawn(async move {
		reqwest::get(format!("http://{addr}/fast")).await
	});
	tokio::time::sleep(Duration::from_millis(100)).await;

	shutdown_tx.send(()).expect("run task dropped the shutdown receiver");

	// Give the accept loop a chance to observe shutdown and drop the queued
	// /fast connection. The run task must NOT have finished yet — it still
	// owes /slow a chance to complete.
	tokio::time::sleep(Duration::from_millis(50)).await;
	let still_running = tokio::time::timeout(Duration::from_millis(10), &mut run_task).await;
	assert!(
		still_running.is_err(),
		"shutdown must wait for the in-flight /slow request, but run() already returned"
	);

	release_tx.send(()).expect("/slow handler dropped the release receiver");

	let result = tokio::time::timeout(Duration::from_secs(2), run_task)
		.await
		.expect("run() did not complete promptly after the in-flight request finished")
		.expect("run task panicked");
	assert!(result.is_ok(), "run() returned an error: {result:?}");

	assert!(
		slow_client.await.unwrap().unwrap().status().is_success(),
		"the in-flight request should have been allowed to complete"
	);

	assert!(
		!fast_invoked.load(Ordering::SeqCst),
		"the queued request must be dropped on shutdown, not served after the fact"
	);

	// The client for the dropped connection just sees a closed connection.
	let _ = fast_client.await;
}

/// A handler that never returns must not hold shutdown open forever — and the drain
/// must actually wait for it first.
///
/// The previous version of this test asserted only that shutdown *completed*, and it
/// completed in ~1s: the in-flight request never reached the drain, so the grace period
/// was never exercised. Setting the drain to 24 hours left it green. It was measuring
/// nothing.
///
/// This asserts both halves, which is what makes it able to fail. Shutdown must take at
/// least most of the grace period (proving it waited for the wedged connection rather
/// than dropping it) and must not take much more (proving the wait is bounded). The
/// handler signals when it has actually been entered, rather than the test sleeping and
/// hoping.
#[tokio::test]
async fn a_wedged_handler_does_not_hold_shutdown_open_forever() {
	const DRAIN: Duration = Duration::from_secs(5);

	let (entered_tx, entered_rx) = oneshot::channel::<()>();
	let entered_tx = Arc::new(Mutex::new(Some(entered_tx)));

	let app = RouteBuilder::stateless()
		.get(
			"/wedge",
			handler(move |_req, _state| {
				let entered_tx = entered_tx.clone();
				async move {
					if let Some(tx) = entered_tx.lock().unwrap().take() {
						let _ = tx.send(());
					}
					// Never returns; stands in for a hung handler or an endless stream.
					std::future::pending::<()>().await;
					unreachable!("pending() never resolves")
				}
			}),
		)
		.seal();

	let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
	let addr = listener.local_addr().unwrap();
	let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();

	let run_task = tokio::spawn(async move {
		app.run(listener, async move {
			let _ = shutdown_rx.await;
		})
		.await
	});

	// Hold the request open, and wait until the handler has genuinely been entered — a
	// sleep here is what let the previous version race past the case it meant to test.
	let wedged = tokio::spawn(async move {
		let _ = reqwest::get(format!("http://{addr}/wedge")).await;
	});
	entered_rx.await.expect("the wedged handler was never entered");

	let asked_to_stop = std::time::Instant::now();
	shutdown_tx.send(()).unwrap();
	let stopped = tokio::time::timeout(DRAIN * 4, run_task).await;
	let took = asked_to_stop.elapsed();

	assert!(
		stopped.is_ok(),
		"shutdown never returned — the wedged handler held it open"
	);
	assert!(
		took >= DRAIN.mul_f32(0.8),
		"shutdown returned after {took:?}, before the grace period — the in-flight \
		 connection was dropped rather than drained"
	);
	assert!(
		took <= DRAIN * 2,
		"shutdown took {took:?} — the grace period is not bounding the drain"
	);
	wedged.abort();
}