Hot-reload content on a NATS trigger instead of requiring a restart
Deploy / deploy (push) Successful in 29s
Deploy / deploy (push) Successful in 29s
questions is now Arc<ArcSwap<HashMap<...>>> - 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.
This commit is contained in:
@@ -96,7 +96,9 @@ pub async fn transition_answer(
|
||||
let state = expect_context::<AppState>();
|
||||
let question = state
|
||||
.questions
|
||||
.load()
|
||||
.get(&question_id)
|
||||
.cloned()
|
||||
.ok_or_else(|| ServerFnError::new("unknown question"))?;
|
||||
let feature = question
|
||||
.alternatives
|
||||
|
||||
+4
-2
@@ -742,7 +742,7 @@ fn NotFound() -> impl IntoView {
|
||||
pub async fn get_question(path: String) -> Result<Option<Question>, ServerFnError> {
|
||||
use crate::server::AppState;
|
||||
let state = expect_context::<AppState>();
|
||||
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::<AppState>();
|
||||
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::<User>(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"));
|
||||
}
|
||||
|
||||
|
||||
+47
-4
@@ -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<arc_swap::ArcSwap<std::collections::HashMap<String, Question>>>,
|
||||
) {
|
||||
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");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+10
-1
@@ -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,
|
||||
};
|
||||
|
||||
@@ -22,7 +22,9 @@ pub async fn get_resource(
|
||||
let state = expect_context::<AppState>();
|
||||
let question = state
|
||||
.questions
|
||||
.load()
|
||||
.get(&question_id)
|
||||
.cloned()
|
||||
.ok_or_else(|| ServerFnError::new("unknown question"))?;
|
||||
let feature = question
|
||||
.alternatives
|
||||
|
||||
+6
-1
@@ -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<HashMap<String, Question>>,
|
||||
/// 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<ArcSwap<HashMap<String, Question>>>,
|
||||
pub oidc: Arc<oidc::Oidc>,
|
||||
/// `None` when `GARAGE_*` env vars aren't set - uploads are the one
|
||||
/// optional feature, everything else works without Garage.
|
||||
|
||||
Reference in New Issue
Block a user