Server overhaul: shared helpers, dead code, synchronous tasks::start, stale comments

crate::lock replaces three poison-tolerant lock helpers; the rate limiter reuses auth::bearer_token; the landing page
reuses invites::display_code. Removed: AppError::not_implemented (D-004 placeholder, only its own test used it) and
the unused Mutation::mutation_id/entity_type. tasks::start never awaited or failed: now a plain fn returning AppState.
Milestone-ownership comments (M2a/M3/M4) rewritten to say what the code does.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TQd3pEP9UeN2uifbHknQ7B
This commit is contained in:
2026-10-18 22:28:51 +02:00
parent 4b30888dc2
commit 6c405e142d
17 changed files with 42 additions and 89 deletions
+2 -11
View File
@@ -12,7 +12,7 @@ use axum::response::{IntoResponse, Response};
use percent_encoding::{NON_ALPHANUMERIC, percent_decode_str, utf8_percent_encode};
use crate::AppState;
use crate::api::invites::FROM_PARAM;
use crate::api::invites::{FROM_PARAM, display_code};
use crate::api::{apk, validate};
/// Longest code the page accepts (E-4).
@@ -110,15 +110,6 @@ fn clean_code(code: &str) -> String {
.collect()
}
fn grouped(code: &str) -> String {
let chars: Vec<char> = code.chars().collect();
chars
.chunks(4)
.map(|c| c.iter().collect::<String>())
.collect::<Vec<_>>()
.join("-")
}
const STYLE: &str = "body{font-family:system-ui,sans-serif;max-width:28rem;margin:3rem auto;padding:0 1rem;\
text-align:center;color:#1b1b1b;background:#fff}h1{font-size:2rem}.code{font-family:monospace;font-size:1.5rem;\
letter-spacing:.1em;margin:1.5rem 0}a.button{display:block;margin:.75rem 0;padding:.9rem;border-radius:.75rem;\
@@ -184,7 +175,7 @@ pub fn render(
lang = copy.html_lang,
app = copy.app_name,
instruction = copy.instruction,
code = escape(&grouped(&code)),
code = escape(&display_code(&code)),
open = escape(&open_url),
open_label = copy.open,
)
+3 -4
View File
@@ -1,7 +1,6 @@
//! HTTP handlers, one file per area (SPEC §14.7), and the extractors and response types they share.
//!
//! Every handler is implemented: `health` (E-1) and `setup` (E-2) by M2a, E-3 to E-30 and E-33 by M3, the sync
//! handlers E-31 and E-32 (`sync.rs`, protocol in `crate::sync`) by M4.
//! HTTP handlers, one file per area (SPEC §14.7), and the extractors and response types they share. The sync
//! handlers E-31 and E-32 are in `sync.rs`, the protocol itself in `crate::sync`; `rows` and `validate` hold the
//! row access and input rules the handlers, the sync protocol and the importer share.
pub mod accounts;
pub mod apk;
@@ -21,7 +21,7 @@
use std::collections::HashMap;
use std::convert::Infallible;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::{Arc, Mutex, MutexGuard};
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll};
use std::time::Instant;
@@ -31,11 +31,11 @@ use axum::response::{IntoResponse, Response};
use chrono::{DateTime, Duration, Utc};
use tower::{Layer, Service};
use crate::AppState;
use crate::audit::BoxFuture;
use crate::auth::{is_token_shaped, sha256_hex};
use crate::auth::{bearer_token, is_token_shaped, sha256_hex};
use crate::error::AppError;
use crate::time::Clock;
use crate::{AppState, lock};
/// Unauthenticated requests per window per client IP (SPEC §8.6).
pub const UNAUTHENTICATED_LIMIT: u32 = 60;
@@ -161,13 +161,6 @@ pub struct Limiter {
buckets: Mutex<Buckets>,
}
fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
match mutex.lock() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
}
}
impl Limiter {
pub fn new(clock: Arc<dyn Clock>, trusted_proxy: String) -> Self {
Self {
@@ -230,15 +223,7 @@ impl Limiter {
if PUBLIC_ROUTES.contains(&route) || route.is_empty() {
return Key::Ip(ip);
}
let token = request
.headers()
.get(axum::http::header::AUTHORIZATION)
.and_then(|v| v.to_str().ok())
.and_then(|v| v.split_once(' '))
.filter(|(scheme, _)| scheme.eq_ignore_ascii_case("bearer"))
.map(|(_, token)| token.trim())
.filter(|token| is_token_shaped(token));
match token {
match bearer_token(request.headers()).filter(|token| is_token_shaped(token)) {
Some(token) => Key::Token {
ip,
hash: sha256_hex(token),
@@ -1,4 +1,4 @@
//! Input validation shared by the M3 handlers (SPEC §8.6, §9.6, §9.7).
//! Input validation shared by the REST handlers, the sync push checks and the importer (SPEC §8.6, §9.6, §9.7).
//!
//! Strings are trimmed of leading and trailing whitespace and their length is counted in Unicode code points after
//! trimming; the trimmed value is what the server stores.
+2 -2
View File
@@ -1,7 +1,7 @@
//! The audit log (SPEC §14.4.7, FR-12). M2a defined the trait; M4 owns `src/audit/**` and implements it with
//! The audit log (SPEC §14.4.7, FR-12): the [`AuditWriter`] seam (SPEC §6.3) and its implementation
//! [`SqlAuditWriter`], which computes the field diff ([`diff::changes`]) and inserts the `audit_log` row.
//!
//! Callers (sync push in M4, recurring generation in E-28 and the scheduler, the importer in M9) call
//! Callers (sync push, recurring generation in E-28, E-29 and the scheduler, the importer) call
//! [`AuditWriter::write`] inside their own `BEGIN IMMEDIATE` transaction, passing the wire object before and after
//! the change; the writer derives `changes` per SPEC §14.4.7 and takes the entry's `server_version` from the same
//! transaction.
+4 -2
View File
@@ -87,7 +87,8 @@ pub struct AuthDevice {
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AdminDevice(pub AuthDevice);
fn bearer_token(headers: &HeaderMap) -> Option<&str> {
/// The token of `Authorization: Bearer <token>` (scheme case-insensitive, token trimmed), shape unchecked.
pub fn bearer_token(headers: &HeaderMap) -> Option<&str> {
let value = headers
.get(axum::http::header::AUTHORIZATION)?
.to_str()
@@ -96,7 +97,8 @@ fn bearer_token(headers: &HeaderMap) -> Option<&str> {
scheme.eq_ignore_ascii_case("bearer").then(|| token.trim())
}
fn app_version_code(headers: &HeaderMap) -> Option<i64> {
/// The `X-Kvitt-App-Version` value: a non-negative integer; anything else is ignored (the stored value is kept).
pub fn app_version_code(headers: &HeaderMap) -> Option<i64> {
headers
.get(APP_VERSION_HEADER)?
.to_str()
+1 -1
View File
@@ -40,7 +40,7 @@ pub struct Config {
/// `KVITT_FCM_PROJECT_ID`.
pub fcm_project_id: String,
/// `KVITT_TRUSTED_PROXY`: host name whose resolved address may set `X-Forwarded-For`. Resolved lazily by the
/// rate limiter (M3); an unresolvable name means "no proxy".
/// rate limiter; an unresolvable name means "no proxy".
pub trusted_proxy: String,
}
+2 -6
View File
@@ -213,10 +213,6 @@ impl AppError {
)
}
pub fn not_implemented() -> Self {
Self::new(ErrorCode::NotImplemented, "not implemented")
}
pub fn rate_limited(retry_after_secs: u64) -> Self {
Self {
retry_after_secs: Some(retry_after_secs),
@@ -308,10 +304,10 @@ mod tests {
#[test]
fn body_has_field_null() {
let v = serde_json::to_value(AppError::not_implemented().body()).unwrap();
let v = serde_json::to_value(AppError::not_found().body()).unwrap();
assert_eq!(
v,
serde_json::json!({"code": "not_implemented", "message": "not implemented", "field": null})
serde_json::json!({"code": "not_found", "message": "not found", "field": null})
);
}
}
+7 -1
View File
@@ -4,7 +4,7 @@
// No unwrap/expect outside tests (CLAUDE.md server rule); integration tests are separate crates and may use them.
#![cfg_attr(not(test), deny(clippy::unwrap_used, clippy::expect_used))]
use std::sync::Arc;
use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
pub mod api;
pub mod audit;
@@ -30,6 +30,12 @@ use time::{Clock, SystemClock};
/// The crate version, sent in `X-Kvitt-Server-Version` and in `GET /health` (SPEC §14.3.1).
pub const SERVER_VERSION: &str = env!("CARGO_PKG_VERSION");
/// Locks `mutex`, taking over the data of a poisoned lock: a panic in one holder never disables the rate limiter,
/// the push queue or a test clock for everyone else.
pub(crate) fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
mutex.lock().unwrap_or_else(PoisonError::into_inner)
}
/// Shared state of every handler. Cheap to clone.
#[derive(Clone)]
pub struct AppState {
+3 -3
View File
@@ -1,6 +1,6 @@
//! Push notifications (SPEC §14.6). M2a defined the trait; M4 owns `src/push/**` and implements it with the bounded
//! queue [`PushHub`] drained by the FCM [`Sender`] (started by `tasks::start`). [`NoopPushQueue`] remains for push
//! disabled (no credentials file).
//! Push notifications (SPEC §14.6): the [`PushQueue`] seam (SPEC §6.3), implemented by the bounded queue
//! [`PushHub`] drained by the FCM [`Sender`] (started by `tasks::start`). [`NoopPushQueue`] is push disabled (no
//! usable credentials file).
//!
//! [`PushQueue::enqueue`] never blocks and never fails the caller: sending happens on a background task and never
//! delays an HTTP response (SPEC §14.6). Call it after the write transaction has committed.
+3 -4
View File
@@ -4,6 +4,8 @@
use std::collections::VecDeque;
use std::sync::{Mutex, MutexGuard};
use crate::lock;
use tokio::sync::Notify;
use super::{PushMessage, PushQueue};
@@ -24,10 +26,7 @@ impl PushHub {
}
fn lock(&self) -> MutexGuard<'_, VecDeque<PushMessage>> {
match self.queue.lock() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
}
lock(&self.queue)
}
/// Messages waiting.
+2 -4
View File
@@ -1,6 +1,4 @@
//! The router: every route of SPEC §7 and the layers of SPEC §14.3.1 and §8.6.
//!
//! M2a owns this file. M3 and M4 fill in handler bodies in their own `api/*.rs` files; the routes are all here.
//! The router: every route of SPEC §7 and the layers of SPEC §14.3.1 and §8.6. The handlers are in `api/*.rs`.
use std::time::Instant;
@@ -106,7 +104,7 @@ pub fn router(state: AppState) -> Router {
.merge(authenticated)
.fallback(fallback)
.layer(DefaultBodyLimit::max(BODY_LIMIT_BYTES))
.layer(rate_limit::layer(&state)) // M3: the rate limiter of SPEC §8.6 attaches here (src/api/rate_limit.rs).
.layer(rate_limit::layer(&state))
.layer(middleware::from_fn(log_request))
.layer(middleware::map_response(add_server_version))
.with_state(state)
+3 -4
View File
@@ -6,7 +6,6 @@ pub mod scheduler;
use std::sync::Arc;
use crate::AppState;
use crate::error::AppError;
use crate::push::{Endpoints, FcmClient, NoopPushQueue, PushHub, PushQueue, RetryPolicy, Sender};
/// The push queue for `state`: the real queue with its sender spawned when the credentials file exists and parses,
@@ -49,8 +48,8 @@ pub fn start_push(
}
/// Starts the background tasks and returns `state` with the real push queue installed (the router must be built
/// from the returned state). The tasks are spawned on the runtime; this returns once they are started.
pub async fn start(mut state: AppState) -> Result<AppState, AppError> {
/// from the returned state). The tasks are spawned on the current runtime; this returns once they are started.
pub fn start(mut state: AppState) -> AppState {
state.push = start_push(&state, Endpoints::google(), RetryPolicy::default());
let generation = state.clone();
@@ -78,5 +77,5 @@ pub async fn start(mut state: AppState) -> Result<AppState, AppError> {
)
.await;
});
Ok(state)
state
}
@@ -90,8 +90,8 @@ async fn plan(after: DateTime<Utc>, at: NaiveTime) -> DateTime<Utc> {
}
}
/// The 06:00 job: M3's generation for every non-deleted, non-paused template (SPEC §9.5). Errors are logged; the
/// next run retries (generation is idempotent).
/// The 06:00 job: the recurring generation for every non-deleted, non-paused template (SPEC §9.5). Errors are
/// logged; the next run retries (generation is idempotent).
pub async fn generation_job(state: &AppState) {
match recurring::generate_all_due(state).await {
Ok(created) => tracing::info!(created, "recurring generation done"),
+2 -8
View File
@@ -62,19 +62,13 @@ impl FixedClock {
/// Sets the instant every clone of this clock returns from now on.
pub fn set(&self, instant: DateTime<Utc>) {
match self.0.lock() {
Ok(mut guard) => *guard = instant,
Err(poisoned) => *poisoned.into_inner() = instant,
}
*crate::lock(&self.0) = instant;
}
}
impl Clock for FixedClock {
fn now(&self) -> DateTime<Utc> {
match self.0.lock() {
Ok(guard) => *guard,
Err(poisoned) => *poisoned.into_inner(),
}
*crate::lock(&self.0)
}
}
@@ -131,22 +131,6 @@ pub enum Mutation {
Settlement(SettlementMutation),
}
impl Mutation {
pub fn mutation_id(&self) -> Uuid {
match self {
Self::Expense(m) => m.mutation_id,
Self::Settlement(m) => m.mutation_id,
}
}
pub fn entity_type(&self) -> EntityType {
match self {
Self::Expense(_) => EntityType::Expense,
Self::Settlement(_) => EntityType::Settlement,
}
}
}
/// E-32 `POST /sync` request: 1 to 100 mutations (more is 422 `batch_too_large`).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
pub struct SyncPushRequest {
@@ -184,7 +184,7 @@ async fn start_runs_generation_catching_up_after_downtime_and_skips_paused_and_d
insert_template(&app, &paused, 1, "2026-06", None, true, false).await;
insert_template(&app, &deleted, 1, "2026-06", None, false, true).await;
// Missing credentials in the test config: push disabled, the scheduler still runs.
let state = kvitt_server::tasks::start(app.state.clone()).await.unwrap();
let state = kvitt_server::tasks::start(app.state.clone());
app.state = state;
let deadline = std::time::Instant::now() + Duration::from_secs(5);
while periods(&app, &active).await.len() < 3 && std::time::Instant::now() < deadline {