From 18025bf870a04bcc851456eb157b967035962f6f Mon Sep 17 00:00:00 2001 From: Bendik Aagaard Lynghaug Date: Wed, 5 Aug 2026 07:20:40 +0200 Subject: [PATCH] Hot-reload content on a NATS trigger instead of requiring a restart questions is now Arc>> - readers do a lock-free atomic load (state.questions.load().get(&id).cloned()), never blocking on or blocked by a reload. content::watch_for_reload subscribes to portal.content.reload (published by the questions repo's own CI after it lints a push - see that repo's lint-and-reload.yml) and swaps in a freshly re-fetched HashMap on each message. A fetch/parse failure logs and keeps serving the last-good content rather than clearing it. --- Cargo.lock | 1 + Cargo.toml | 2 ++ src/answers.rs | 2 ++ src/app.rs | 6 ++++-- src/content.rs | 51 +++++++++++++++++++++++++++++++++++++++++++---- src/main.rs | 11 +++++++++- src/resource.rs | 2 ++ src/server/mod.rs | 7 ++++++- 8 files changed, 74 insertions(+), 8 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 65aa92f..486c64a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2751,6 +2751,7 @@ name = "portal" version = "0.1.0" dependencies = [ "anyhow", + "arc-swap", "async-nats", "aws-sdk-s3", "axum", diff --git a/Cargo.toml b/Cargo.toml index 64cd5e9..9838fa6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -22,6 +22,7 @@ tower = { version = "0.5", optional = true } tower-http = { version = "0.6", features = ["fs"], optional = true } tower-sessions = { version = "0.14", optional = true } async-nats = { version = "0.38", optional = true } +arc-swap = { version = "1", optional = true } url = { version = "2", optional = true } serde_yaml = { version = "0.9", optional = true } sha2 = { version = "0.10", optional = true } @@ -76,6 +77,7 @@ ssr = [ "dep:tower-http", "dep:tower-sessions", "dep:async-nats", + "dep:arc-swap", "dep:url", "dep:serde_yaml", "dep:sha2", diff --git a/src/answers.rs b/src/answers.rs index db39be6..830d451 100644 --- a/src/answers.rs +++ b/src/answers.rs @@ -96,7 +96,9 @@ pub async fn transition_answer( let state = expect_context::(); let question = state .questions + .load() .get(&question_id) + .cloned() .ok_or_else(|| ServerFnError::new("unknown question"))?; let feature = question .alternatives diff --git a/src/app.rs b/src/app.rs index dcf9c9c..37d4c3d 100644 --- a/src/app.rs +++ b/src/app.rs @@ -742,7 +742,7 @@ fn NotFound() -> impl IntoView { pub async fn get_question(path: String) -> Result, ServerFnError> { use crate::server::AppState; let state = expect_context::(); - Ok(state.questions.get(&path).cloned()) + Ok(state.questions.load().get(&path).cloned()) } #[derive(Clone, Debug, Serialize, Deserialize)] @@ -768,7 +768,9 @@ pub async fn submit_answer( let state = expect_context::(); let question = state .questions + .load() .get(&question_id) + .cloned() .ok_or_else(|| ServerFnError::new("unknown question"))?; let session: tower_sessions::Session = leptos_axum::extract().await?; @@ -776,7 +778,7 @@ pub async fn submit_answer( .get::(SESSION_USER_KEY) .await .map_err(|e| ServerFnError::new(e.to_string()))?; - if !is_qualified(user.as_ref(), question) { + if !is_qualified(user.as_ref(), &question) { return Err(ServerFnError::new("not authorized for this question")); } diff --git a/src/content.rs b/src/content.rs index 5b3d1de..de089c5 100644 --- a/src/content.rs +++ b/src/content.rs @@ -146,10 +146,10 @@ impl Requirement { /// Fetches every `*.yaml` file under `subdir` in a Gitea repo as a /// `Question`, keyed by its own `id`. `repo_url` is the repo's normal /// browser URL (e.g. `https://project.uhhm.no/uhhm/questions`) - the -/// Gitea host, owner and repo name are all read from it. Runs once at -/// startup, over Gitea's public contents API (no auth - the content -/// repo is public); no hot-reload yet - restart the process (or add -/// polling later) to pick up content changes. +/// Gitea host, owner and repo name are all read from it. Called once at +/// startup, and again on every `CONTENT_RELOAD_SUBJECT` message (see +/// `watch_for_reload`), over Gitea's public contents API (no auth - the +/// content repo is public). #[cfg(feature = "ssr")] pub async fn load_questions_from_gitea( repo_url: &str, @@ -218,3 +218,46 @@ pub async fn load_questions_from_gitea( } Ok(out) } + +/// Published by the content repo's own CI (after it lints a push) to +/// tell every running instance to pick up the change - a plain fire +/// and forget NATS publish, no payload, matching `events.rs`'s +/// `ANSWERS_SUBJECT` pattern. +#[cfg(feature = "ssr")] +pub const CONTENT_RELOAD_SUBJECT: &str = "portal.content.reload"; + +/// Runs for the life of the process: re-fetches `repo_url`/`branch` and +/// atomically swaps it into `questions` on every `CONTENT_RELOAD_SUBJECT` +/// message. A fetch/parse failure logs and keeps serving the last-good +/// content rather than clearing it - a bad push to the content repo +/// (which should already have been caught by its own lint step) doesn't +/// take the site down. +#[cfg(feature = "ssr")] +pub async fn watch_for_reload( + nats: async_nats::Client, + repo_url: String, + branch: String, + subdir: String, + questions: std::sync::Arc>>, +) { + let mut sub = match nats.subscribe(CONTENT_RELOAD_SUBJECT).await { + Ok(sub) => sub, + Err(e) => { + tracing::error!(error = %e, "failed to subscribe to content reload subject"); + return; + } + }; + use futures::StreamExt; + while sub.next().await.is_some() { + match load_questions_from_gitea(&repo_url, &branch, &subdir).await { + Ok(loaded) => { + let count = loaded.len(); + questions.store(std::sync::Arc::new(loaded)); + tracing::info!(count, "reloaded content"); + } + Err(e) => { + tracing::error!(error = %e, "content reload failed, keeping last-good content"); + } + } + } +} diff --git a/src/main.rs b/src/main.rs index f413699..7b8ec41 100644 --- a/src/main.rs +++ b/src/main.rs @@ -30,6 +30,7 @@ async fn main() -> anyhow::Result<()> { let content_branch = std::env::var("CONTENT_BRANCH").unwrap_or_else(|_| "main".to_string()); let questions = content::load_questions_from_gitea(&content_repo, &content_branch, "questions").await?; tracing::info!(count = questions.len(), repo = %content_repo, branch = %content_branch, "loaded content"); + let questions = Arc::new(arc_swap::ArcSwap::from_pointee(questions)); let nats_url = std::env::var("NATS_URL").unwrap_or_else(|_| "nats://127.0.0.1:4222".to_string()); @@ -56,11 +57,19 @@ async fn main() -> anyhow::Result<()> { tracing::warn!("GARAGE_* env vars not set - file upload fields will fail closed"); } + tokio::spawn(content::watch_for_reload( + nats.clone(), + content_repo, + content_branch, + "questions".to_string(), + questions.clone(), + )); + let state = AppState { leptos_options: leptos_options.clone(), nats, jetstream, - questions: std::sync::Arc::new(questions), + questions, oidc: oidc_state, garage, }; diff --git a/src/resource.rs b/src/resource.rs index 5ed242e..cd7e234 100644 --- a/src/resource.rs +++ b/src/resource.rs @@ -22,7 +22,9 @@ pub async fn get_resource( let state = expect_context::(); let question = state .questions + .load() .get(&question_id) + .cloned() .ok_or_else(|| ServerFnError::new("unknown question"))?; let feature = question .alternatives diff --git a/src/server/mod.rs b/src/server/mod.rs index 78eea50..3b6a2d0 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -2,6 +2,7 @@ pub mod oidc; +use arc_swap::ArcSwap; use axum::extract::FromRef; use leptos::prelude::LeptosOptions; use std::collections::HashMap; @@ -16,7 +17,11 @@ pub struct AppState { /// JetStream context - source of every NATS KV bucket this app /// reads/writes (applicants, and whatever a `ResourceSpec` names). pub jetstream: async_nats::jetstream::Context, - pub questions: Arc>, + /// Swapped out wholesale on a `content::CONTENT_RELOAD_SUBJECT` + /// message (see `content::watch_for_reload`) - readers never hold + /// a lock, just an atomic pointer load, so a reload never blocks or + /// is blocked by an in-flight request. + pub questions: Arc>>, pub oidc: Arc, /// `None` when `GARAGE_*` env vars aren't set - uploads are the one /// optional feature, everything else works without Garage.