ddidderr/tdkpin
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(highscores): bound service resource usage

Browse Source 
The high-score endpoints previously accepted unbounded request bodies and ran
SQLite work directly in async handlers, allowing oversized input or database
contention to consume server resources. Add a 1 KiB route body limit, admit
only one database operation at a time, shed excess requests with a clear 503,
and run accepted SQLite work on blocking threads while retaining admission
until that work finishes. Extend the Nginx example with matching request,
connection, body, and proxy time limits, and cover the limits, health
availability, contention, and cancellation behavior with tests.

Test Plan:
- `just --justfile tdkpin-rs/justfile test` -- passed (144 tests)
- `just --justfile tdkpin-rs/justfile clippy` -- passed
- `cargo +nightly fmt --manifest-path tdkpin-rs/highscore-server/Cargo.toml -- --check` -- passed
- `rumdl check --flavor commonmark tdkpin-rs/highscore-server/README.md` -- passed
- `git diff --cached --check` -- passed
This commit is contained in:
ddidderr
2026-08-29 19:36:53 +02:00
parent 86434aaa2b
commit 3dff722535
4 changed files with 273 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tdkpin-rs/highscore-server/src/lib.rs
+243 -9
View File
@@ -6,16 +6,19 @@ use std::{
use axum::{
Json,
Router,
extract::State,
extract::{DefaultBodyLimit, State},
http::StatusCode,
response::{IntoResponse, Response},
routing::get,
};
use rusqlite::{Connection, params, types::Type};
use serde::{Deserialize, Serialize};
use tokio::{sync::Semaphore, task};
const MAX_HIGH_SCORES: usize = 10;
const MAX_NAME_CHARS: usize = 21;
const MAX_SUBMISSION_BODY_BYTES: usize = 1024;
const MAX_CONCURRENT_DATABASE_OPERATIONS: usize = 1;
#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
pub struct HighScore {
@@ -127,6 +130,50 @@ impl HighScoreStore {
}
}
#[derive(Clone)]
struct AppState {
store: HighScoreStore,
database_slots: Arc<Semaphore>,
}
impl AppState {
fn new(store: HighScoreStore) -> Self {
Self {
store,
database_slots: Arc::new(Semaphore::new(MAX_CONCURRENT_DATABASE_OPERATIONS)),
}
}
}
#[derive(Debug)]
enum DatabaseRequestError {
Busy,
Failed,
}
async fn run_database_operation<T, F>(
state: &AppState,
operation: F,
) -> Result<T, DatabaseRequestError>
where
T: Send + 'static,
F: FnOnce(&HighScoreStore) -> Result<T, StoreError> + Send + 'static,
{
let permit = state
.database_slots
.clone()
.try_acquire_owned()
.map_err(|_| DatabaseRequestError::Busy)?;
let store = state.store.clone();
task::spawn_blocking(move || {
let _permit = permit;
operation(&store)
})
.await
.map_err(|_| DatabaseRequestError::Failed)?
.map_err(|_| DatabaseRequestError::Failed)
}
#[derive(Debug, Deserialize)]
struct SubmitRequest {
name: String,
@@ -146,6 +193,19 @@ fn invalid_request(message: &'static str) -> Response {
.into_response()
}
fn database_error_response(error: &DatabaseRequestError) -> Response {
match error {
DatabaseRequestError::Busy => (
StatusCode::SERVICE_UNAVAILABLE,
Json(ErrorResponse {
error: "service is busy",
}),
)
.into_response(),
DatabaseRequestError::Failed => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
}
}
fn validate_request(request: &SubmitRequest) -> Result<HighScore, &'static str> {
let name = request.name.trim();
if name.is_empty() {
@@ -167,39 +227,47 @@ async fn health() -> &'static str {
"ok"
}
async fn list_high_scores(State(store): State<HighScoreStore>) -> Response {
match store.list() {
async fn list_high_scores(State(state): State<AppState>) -> Response {
match run_database_operation(&state, HighScoreStore::list).await {
Ok(scores) => Json(scores).into_response(),
Err(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
Err(error) => database_error_response(&error),
}
}
async fn submit_high_score(
State(store): State<HighScoreStore>,
State(state): State<AppState>,
Json(request): Json<SubmitRequest>,
) -> Response {
let entry = match validate_request(&request) {
Ok(entry) => entry,
Err(message) => return invalid_request(message),
};
match store.submit(&entry) {
match run_database_operation(&state, move |store| store.submit(&entry)).await {
Ok(scores) => (StatusCode::CREATED, Json(scores)).into_response(),
Err(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
Err(error) => database_error_response(&error),
}
}
pub fn router(store: HighScoreStore) -> Router {
router_with_state(AppState::new(store))
}
fn router_with_state(state: AppState) -> Router {
Router::new()
.route("/healthz", get(health))
.route(
"/api/highscores",
get(list_high_scores).post(submit_high_score),
get(list_high_scores)
.post(submit_high_score)
.layer(DefaultBodyLimit::max(MAX_SUBMISSION_BODY_BYTES)),
)
.with_state(store)
.with_state(state)
}
#[cfg(test)]
mod tests {
use std::{sync::mpsc, time::Duration};
use axum::{
body::Body,
http::{Request, StatusCode},
@@ -229,6 +297,44 @@ mod tests {
.status()
}
async fn request(app: &Router, request: Request<Body>) -> StatusCode {
app.clone()
.oneshot(request)
.await
.expect("router should respond")
.status()
}
fn hold_database_lock(
store: HighScoreStore,
) -> (mpsc::SyncSender<()>, std::thread::JoinHandle<()>) {
let (locked_sender, locked_receiver) = mpsc::sync_channel(0);
let (release_sender, release_receiver) = mpsc::sync_channel(0);
let lock_thread = std::thread::spawn(move || {
let _connection = store.0.lock().expect("database lock should succeed");
locked_sender
.send(())
.expect("test should observe the held database lock");
release_receiver
.recv()
.expect("test should release the database lock");
});
locked_receiver
.recv()
.expect("database lock thread should start");
(release_sender, lock_thread)
}
async fn wait_until_database_is_busy(database_slots: &Semaphore) {
tokio::time::timeout(Duration::from_secs(1), async {
while database_slots.available_permits() != 0 {
tokio::task::yield_now().await;
}
})
.await
.expect("first database request should acquire admission");
}
async fn list(app: &Router) -> Vec<HighScore> {
let response = app
.clone()
@@ -284,6 +390,134 @@ mod tests {
assert_eq!(submit(&app, "ok\nno", 10).await, StatusCode::BAD_REQUEST);
}
#[tokio::test]
async fn api_rejects_oversized_submission_bodies() {
let app = router(HighScoreStore::open_in_memory().expect("database should open"));
let body = serde_json::to_vec(&serde_json::json!({
"name": "Player",
"score": 10,
"padding": "x".repeat(MAX_SUBMISSION_BODY_BYTES),
}))
.expect("request JSON should encode");
let status = request(
&app,
Request::post("/api/highscores")
.header("content-type", "application/json")
.body(Body::from(body))
.expect("request should build"),
)
.await;
assert_eq!(status, StatusCode::PAYLOAD_TOO_LARGE);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
async fn busy_database_load_sheds_api_without_blocking_health() {
let store = HighScoreStore::open_in_memory().expect("database should open");
let (release_sender, lock_thread) = hold_database_lock(store.clone());
let state = AppState::new(store);
let database_slots = state.database_slots.clone();
let app = router_with_state(state);
let accepted_app = app.clone();
let accepted = tokio::spawn(async move { submit(&accepted_app, "Player", 10).await });
wait_until_database_is_busy(&database_slots).await;
assert_eq!(
tokio::time::timeout(
Duration::from_millis(250),
request(
&app,
Request::get("/healthz")
.body(Body::empty())
.expect("request should build"),
),
)
.await
.expect("health request should not wait for the database"),
StatusCode::OK
);
assert_eq!(
submit(&app, "Other Player", 20).await,
StatusCode::SERVICE_UNAVAILABLE
);
assert_eq!(
request(
&app,
Request::get("/api/highscores")
.body(Body::empty())
.expect("request should build"),
)
.await,
StatusCode::SERVICE_UNAVAILABLE
);
assert_eq!(
request(
&app,
Request::builder()
.method("HEAD")
.uri("/api/highscores")
.body(Body::empty())
.expect("request should build"),
)
.await,
StatusCode::SERVICE_UNAVAILABLE
);
release_sender
.send(())
.expect("database lock should be released");
assert_eq!(
accepted.await.expect("accepted request should complete"),
StatusCode::CREATED
);
lock_thread
.join()
.expect("database lock thread should stop");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
async fn canceled_request_holds_admission_until_blocking_work_stops() {
let store = HighScoreStore::open_in_memory().expect("database should open");
let (release_sender, lock_thread) = hold_database_lock(store.clone());
let state = AppState::new(store);
let database_slots = state.database_slots.clone();
let app = router_with_state(state);
let canceled_app = app.clone();
let canceled = tokio::spawn(async move { submit(&canceled_app, "Player", 10).await });
wait_until_database_is_busy(&database_slots).await;
canceled.abort();
assert!(
canceled
.await
.expect_err("request should be canceled")
.is_cancelled()
);
assert_eq!(database_slots.available_permits(), 0);
assert_eq!(
submit(&app, "Other Player", 20).await,
StatusCode::SERVICE_UNAVAILABLE
);
release_sender
.send(())
.expect("database lock should be released");
lock_thread
.join()
.expect("database lock thread should stop");
tokio::time::timeout(Duration::from_secs(1), async {
while database_slots.available_permits() == 0 {
tokio::task::yield_now().await;
}
})
.await
.expect("completed blocking work should release admission");
assert_eq!(submit(&app, "Other Player", 20).await, StatusCode::CREATED);
}
#[tokio::test]
async fn health_endpoint_is_available_for_nginx() {
let app = router(HighScoreStore::open_in_memory().expect("database should open"));