Lines
0 %
Functions
Branches
100 %
#[macro_use]
extern crate rust_i18n;
i18n!("locales", fallback = "en");
mod auth_keys;
mod client_ip;
mod config;
mod files;
mod handler;
mod jwt_auth;
mod model;
mod pages;
mod rate_limit;
mod redirect_middleware;
mod response;
mod route;
mod telemetry;
mod token;
use axum::{
Router,
extract::{MatchedPath, Request},
middleware::{self, Next},
response::IntoResponse,
routing::get,
};
use axum::http::{
HeaderValue, Method,
header::{ACCEPT, AUTHORIZATION, CONTENT_TYPE},
use redis::Client;
use std::sync::Arc;
use std::time::Duration;
use std::{future::ready, time::Instant};
use tower_http::{cors::CorsLayer, services::ServeDir};
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
use config::Config;
use route::create_scripts_router;
use route::{
create_accounts_router, create_admin_pages_router, create_api_router, create_pages_router,
create_reports_router, create_tags_router, create_transactions_router,
pub struct AppState {
conf: Config,
redis_client: Client,
frac: i64,
}
/// The metrics listener. Fails rather than panics: a recorder that cannot be
/// installed is a startup fault, and `main` already reports those.
fn metrics_app() -> anyhow::Result<Router> {
let recorder_handle = telemetry::install_recorder()?;
Ok(Router::new().route("/metrics", get(move || ready(recorder_handle.render()))))
async fn track_metrics(req: Request, next: Next) -> impl IntoResponse {
let start = Instant::now();
let path = if let Some(matched_path) = req.extensions().get::<MatchedPath>() {
matched_path.as_str().to_owned()
} else {
req.uri().path().to_owned()
let method = req.method().clone();
let response = next.run(req).await;
let latency = start.elapsed().as_secs_f64();
let status = response.status().as_u16().to_string();
let labels = [
("method", method.to_string()),
("path", path),
("status", status),
];
metrics::counter!(telemetry::HTTP_REQUESTS, &labels).increment(1);
metrics::histogram!(telemetry::HTTP_REQUEST_DURATION, &labels).record(latency);
response
/// How long the session store may take to accept a connection at startup.
///
/// Redis runs beside web in the same pod and the two start together, so a
/// refused connection here means "not listening yet", not "misconfigured".
/// Bounded rather than infinite: a Redis that is genuinely absent must fail the
/// process, not leave it waiting and reported healthy.
const REDIS_WAIT: Duration = Duration::from_secs(30);
/// Pause between connection attempts.
const REDIS_RETRY_PAUSE: Duration = Duration::from_secs(1);
/// Whether an error means "not up yet" rather than "wrong".
/// Only the store being absent is worth waiting out. A refused password or a
/// bad database number will be refused just as firmly in thirty seconds, so
/// retrying one buries a configuration error under a startup delay.
/// Deliberately narrower than `is_io_error`, which is every `io::Error` the
/// crate wraps — a name that does not resolve included. That is a wrong
/// address, not a slow one, and it should fail at once.
fn redis_not_up_yet(e: &redis::RedisError) -> bool {
// Refusal covers the case this exists for: the sidecar has not bound its
// port yet. On unix it also covers a socket file that is not there yet.
e.is_connection_refusal() || e.is_timeout()
/// Opens the session store, waiting for it to come up.
/// Reports the address rather than the configured URL: that URL may carry
/// credentials, and this error reaches the logs.
async fn connect_redis(url: &str) -> anyhow::Result<Client> {
let client =
Client::open(url.to_string()).map_err(|e| anyhow::anyhow!("redis url unusable: {e}"))?;
let addr = client.get_connection_info().addr().to_string();
let deadline = Instant::now() + REDIS_WAIT;
let mut last = "no attempt completed".to_string();
loop {
let remaining = deadline.saturating_duration_since(Instant::now());
if remaining.is_zero() {
return Err(anyhow::anyhow!(
"redis at {addr} did not accept a connection within {REDIS_WAIT:?}: {last}"
));
// Each attempt carries the remaining budget: the connect future has no
// timeout of its own, so a half-open socket would hang past REDIS_WAIT
// and make the bound a fiction.
//
// The multiplexed async connection, not `get_connection`: the blocking
// one stalls the runtime this is awaited on.
match tokio::time::timeout(remaining, client.get_multiplexed_async_connection()).await {
Ok(Ok(_)) => {
log::info!("connected to redis at {addr}");
return Ok(client);
Ok(Err(e)) if !redis_not_up_yet(&e) => {
"redis at {addr} refused the connection: {e}"
Ok(Err(e)) => last = e.to_string(),
Err(_) => last = "the connection attempt timed out".to_string(),
log::warn!("redis at {addr} not ready ({last}); retrying");
let left = deadline.saturating_duration_since(Instant::now());
tokio::time::sleep(REDIS_RETRY_PAUSE.min(left)).await;
#[tokio::main]
async fn main() -> anyhow::Result<()> {
tracing_subscriber::registry()
.with(
tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()),
)
.with(tracing_subscriber::fmt::layer())
.try_init()?;
// Migrate + seed the admin DB before reading config: seeding moved out of
// the migration set into the `seed_complete`-guarded bootstrap, so a fresh
// DB has no `site_url` until `boot` runs. Idempotent on an already-booted
// DB. Without this, `Config::init` panics with NoConfig("site_url").
server::boot()
.await
.map_err(|e| anyhow::anyhow!("server boot failed: {e:?}"))?;
log::debug!("Server boot complete");
let conf = Config::init()
.map_err(|e| anyhow::anyhow!("config init failed: {e:?}"))?;
log::debug!("Config ready");
let redis_client = connect_redis(&conf.redis_url).await?;
let cors = CorsLayer::new()
.allow_origin(conf.site_url.parse::<HeaderValue>()?)
.allow_methods([Method::GET, Method::POST, Method::PATCH, Method::DELETE])
.allow_credentials(true)
.allow_headers([AUTHORIZATION, ACCEPT, CONTENT_TYPE]);
let state = Arc::new(AppState {
conf,
redis_client: redis_client.clone(),
frac: 0,
});
// Validated before anything serves: getting this wrong either throttles
// every user into one shared budget or lets them forge their identity, and
// both failures are quiet.
let limiter = Arc::new(rate_limit::AuthLimiter::new(
client_ip::TrustedProxies::from_env()?,
let router = Router::new()
.route("/", get(pages::index))
.route("/login", get(pages::login))
.merge(create_admin_pages_router(state.clone()))
.merge(create_pages_router(state.clone()))
.merge(create_accounts_router(state.clone()))
.merge(create_transactions_router(state.clone()))
.merge(create_tags_router(state.clone()))
.merge(create_reports_router(state.clone()));
let router = router.merge(create_scripts_router(state.clone()));
let router = router
.nest(
"/api",
create_api_router(state.clone(), Arc::clone(&limiter)),
// Every authenticated request decodes a JWT, and decoding reads the
// claimed subject's key from the directory before anything is verified.
// Applied here so it covers every page and API route; static assets are
// nested below and need no database.
.route_layer(middleware::from_fn_with_state(
limiter,
rate_limit::throttle_verification,
))
.nest_service(
"/static",
ServeDir::new(std::env::var("STATIC_PATH").unwrap_or("web/static".to_string())),
.with_state(state)
.layer(cors)
.route_layer(middleware::from_fn(track_metrics));
let app = metrics_app()?;
let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await?;
let metrics_listener = tokio::net::TcpListener::bind("0.0.0.0:3001").await?;
// Either listener dying is fatal: a live app port with a dead metrics port
// (or vice versa) is a half-running process that k8s cannot see is broken.
let (metrics_server, main_server) = tokio::join!(
run_server(metrics_listener, app),
run_server(listener, router)
);
metrics_server?;
main_server?;
Ok(())
async fn run_server(listener: tokio::net::TcpListener, app: Router) -> std::io::Result<()> {
// `with_connect_info` is what makes the peer address reachable from a
// handler. Without it the throttle has nothing trustworthy to key on.
axum::serve(
listener,
app.into_make_service_with_connect_info::<std::net::SocketAddr>(),