forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
18.3 KiB
1//! Page: standby replication (`/admin/replication`). Instance admins only; anyone else gets 404.
2//!
3//! Shows the `[replication]` target, the last database snapshot shipped, and one row per repo
4//! with its health: ok, lagging (no success within twice the interval), error, or never replicated.
5//! "Reconcile now" queues a full pass for the worker. It is a POST with a CSRF token, and it is
6//! audited as `replication_reconcile_started`.
7
8use std::collections::HashMap;
9use std::time::Duration;
10
11use forge_core::store::replication::{DB_SUBJECT, ReplicationStatus};
12use forge_core::time::stamp_ago;
13use forge_core::tz::format_in_tz;
14use forge_core::{AuditAction, Error};
15use serde::Deserialize;
16use topcoat::Result;
17use topcoat::context::{Cx, app_context};
18use topcoat::router::content::Form;
19use topcoat::router::error::{forbidden, see_other};
20use topcoat::router::response::Response;
21use topcoat::router::{page, query_params, route};
22use topcoat::view::{View, component, view};
23
24use crate::audit::{self, Actor, Target};
25use crate::auth::check_csrf;
26use crate::i18n::admin::{self as g, K};
27use crate::i18n::{self, Key, Lang};
28use crate::pages::admin_users::admin_only;
29use crate::pages::issues_list::stamp_unix;
30use crate::state::AppState;
31use crate::views::{enc, shell};
32
33#[query_params(error = bad_request)]
34pub(crate) struct NoticeQuery {
35 notice: Option<String>,
36}
37
38/// How a subject reads on the page.
39#[derive(Debug, Clone, Copy, PartialEq, Eq)]
40enum Health {
41 Ok,
42 Lagging,
43 Error,
44 Never,
45}
46
47impl Health {
48 fn label(self, lang: Lang) -> &'static str {
49 match self {
50 Health::Ok => g::t(lang, K::HealthOk),
51 Health::Lagging => g::t(lang, K::HealthLagging),
52 Health::Error => g::t(lang, K::HealthError),
53 Health::Never => g::t(lang, K::HealthNever),
54 }
55 }
56}
57
58/// Health of one subject. `cutoff` is the oldest success that still counts as current; stamps are
59/// fixed-width UTC, so a text comparison orders them.
60fn health(status: Option<&ReplicationStatus>, cutoff: &str) -> Health {
61 match status {
62 None => Health::Never,
63 Some(s) if s.last_error.is_some() => Health::Error,
64 Some(s) => match s.last_ok_at.as_deref() {
65 None => Health::Never,
66 Some(ok) if ok >= cutoff => Health::Ok,
67 Some(_) => Health::Lagging,
68 },
69 }
70}
71
72/// One line of the repos table, already formatted.
73struct RepoRow {
74 name: String,
75 health: &'static str,
76 last_ok: String,
77 error: String,
78}
79
80/// Everything the page shows, already formatted.
81struct PageData {
82 enabled: bool,
83 target: String,
84 db_health: &'static str,
85 db_last_ok: String,
86 db_last_attempt: String,
87 db_error: String,
88 interval: String,
89 healthcheck: &'static str,
90 repos: Vec<RepoRow>,
91}
92
93fn ok_text(stamp: Option<&str>, tz: &str, lang: Lang) -> String {
94 match stamp {
95 Some(stamp) => format_in_tz(stamp_unix(stamp), tz),
96 None => i18n::t(lang, Key::Never).to_owned(),
97 }
98}
99
100/// Builds the page data from the config, the status rows and the repo list.
101async fn page_data(state: &AppState, tz: &str, lang: Lang) -> forge_core::Result<PageData> {
102 let config = &state.config.replication;
103 let db_cutoff = stamp_ago(Duration::from_secs(config.db_snapshot_minutes * 60 * 2));
104 let repo_cutoff = stamp_ago(Duration::from_secs(config.reconcile_minutes * 60 * 2));
105
106 let statuses: HashMap<String, ReplicationStatus> = state
107 .store
108 .list_replication_status()
109 .await?
110 .into_iter()
111 .map(|s| (s.subject.clone(), s))
112 .collect();
113
114 let db = statuses.get(DB_SUBJECT);
115 let mut repos = Vec::new();
116 let ids = state.store.repo_ids().await?;
117 for id in &ids {
118 let status = statuses.get(id);
119 let name = match state.store.repo_by_id(id).await? {
120 Some(repo) => format!("{}/{}", repo.owner_name, repo.name),
121 None => id.clone(),
122 };
123 repos.push(RepoRow {
124 name,
125 health: health(status, &repo_cutoff).label(lang),
126 last_ok: ok_text(status.and_then(|s| s.last_ok_at.as_deref()), tz, lang),
127 error: status
128 .and_then(|s| s.last_error.clone())
129 .unwrap_or_default(),
130 });
131 }
132 // Repos deleted since their last push: their standby copy is removed at the next reconcile.
133 for status in statuses.values() {
134 if status.subject != DB_SUBJECT && !ids.contains(&status.subject) {
135 repos.push(RepoRow {
136 name: g::deleted_repo_name(lang, &status.subject),
137 health: health(Some(status), &repo_cutoff).label(lang),
138 last_ok: ok_text(status.last_ok_at.as_deref(), tz, lang),
139 error: status.last_error.clone().unwrap_or_default(),
140 });
141 }
142 }
143
144 Ok(PageData {
145 enabled: config.enabled(),
146 target: config.target.clone(),
147 db_health: health(db, &db_cutoff).label(lang),
148 db_last_ok: ok_text(db.and_then(|s| s.last_ok_at.as_deref()), tz, lang),
149 db_last_attempt: ok_text(db.and_then(|s| s.last_attempt_at.as_deref()), tz, lang),
150 db_error: db.and_then(|s| s.last_error.clone()).unwrap_or_default(),
151 interval: g::interval_line(lang, config.db_snapshot_minutes, config.reconcile_minutes),
152 healthcheck: if config.healthcheck_url.is_empty() {
153 g::t(lang, K::HealthcheckNotSet)
154 } else {
155 g::t(lang, K::HealthcheckSet)
156 },
157 repos,
158 })
159}
160
161#[component]
162async fn replication_view(
163 lang: Lang,
164 csrf: String,
165 data: PageData,
166 notice: Option<String>,
167) -> Result<impl View> {
168 Ok(view! {
169 <section class="page-head">
170 <h1>(g::t(lang, K::Replication))</h1>
171 <p class="muted">(g::t(lang, K::ReplicationIntro))</p>
172 </section>
173 if let Some(message) = notice {
174 <section class="card notice" role="status">
175 <p>(message.as_str())</p>
176 </section>
177 }
178 <section class="card">
179 <h2>(g::t(lang, K::Standby))</h2>
180 if data.enabled {
181 <dl class="details">
182 <dt>(g::t(lang, K::ReplicationTarget))</dt><dd><code>(data.target.as_str())</code></dd>
183 <dt>(g::t(lang, K::Interval))</dt><dd>(data.interval.as_str())</dd>
184 <dt>(g::t(lang, K::Healthcheck))</dt><dd>(data.healthcheck)</dd>
185 <dt>(g::t(lang, K::Database))</dt><dd>(data.db_health)</dd>
186 <dt>(g::t(lang, K::LastDbSnapshotShipped))</dt><dd class="small">(data.db_last_ok.as_str())</dd>
187 <dt>(g::t(lang, K::LastDbAttempt))</dt><dd class="small">(data.db_last_attempt.as_str())</dd>
188 </dl>
189 if !data.db_error.is_empty() {
190 <p class="error">(g::t(lang, K::DatabaseErrorPrefix))(data.db_error.as_str())</p>
191 }
192 } else {
193 <p>(g::t(lang, K::ReplicationOff))</p>
194 }
195 <div class="actions">
196 <form method="post" action="/admin/replication/reconcile">
197 <input type="hidden" name="csrf" value=(csrf.as_str())>
198 <button type="submit" class="primary">(g::t(lang, K::ReconcileNow))</button>
199 </form>
200 </div>
201 </section>
202 <section class="card">
203 <h2>(g::t(lang, K::Repos))</h2>
204 if data.repos.is_empty() {
205 <p class="muted">(g::t(lang, K::NoReposYet))</p>
206 } else {
207 <div class="table-wrap">
208 <table class="table">
209 <thead>
210 <tr>
211 <th>(g::t(lang, K::TableRepo))</th>
212 <th>(g::t(lang, K::TableStatus))</th>
213 <th>(g::t(lang, K::LastSuccess))</th>
214 <th>(g::t(lang, K::TableError))</th>
215 </tr>
216 </thead>
217 <tbody>
218 for row in data.repos.into_iter() {
219 <tr>
220 <td>(row.name.as_str())</td>
221 <td>(row.health)</td>
222 <td class="small">(row.last_ok.as_str())</td>
223 <td class="small">(row.error.as_str())</td>
224 </tr>
225 }
226 </tbody>
227 </table>
228 </div>
229 }
230 </section>
231 })
232}
233
234#[page("/admin/replication")]
235async fn replication_page(cx: &Cx) -> Result<impl View> {
236 let viewer = admin_only(cx).await?;
237 let query = query_params::<NoticeQuery>(cx)?;
238 let lang = i18n::lang(cx, Some(&viewer.0));
239 let state = app_context::<AppState>(cx);
240 let data = page_data(state, viewer.0.timezone.as_str(), lang).await?;
241 let csrf = viewer.1.csrf_token.clone();
242 let title = g::t(lang, K::Replication);
243 Ok(view! {
244 shell(title: title, viewer: Some(&viewer), replication_view(lang: lang, csrf: csrf, data: data, notice: query.notice.clone()))
245 })
246}
247
248#[derive(Deserialize)]
249struct ReconcileForm {
250 #[serde(default)]
251 csrf: String,
252}
253
254#[route(POST "/admin/replication/reconcile")]
255async fn replication_reconcile(cx: &Cx, Form(form): Form<ReconcileForm>) -> Result<Response> {
256 let viewer = admin_only(cx).await?;
257 if !check_csrf(&viewer.1, &form.csrf) {
258 return Err(forbidden().into());
259 }
260 let lang = i18n::lang(cx, Some(&viewer.0));
261 let state = app_context::<AppState>(cx);
262 let notice = match reconcile_now(cx, state, &viewer.0, lang).await {
263 Ok(()) => g::t(lang, K::ReconcileStarted).to_owned(),
264 Err(Error::Config(message)) => message,
265 Err(other) => other.to_string(),
266 };
267 let location = format!("/admin/replication?notice={}", enc(&notice));
268 Err(see_other(&location).into())
269}
270
271/// Queues a reconcile pass and audits the request. Refused when replication is not configured.
272async fn reconcile_now(
273 cx: &Cx,
274 state: &AppState,
275 user: &forge_core::User,
276 lang: Lang,
277) -> forge_core::Result<()> {
278 if !state.config.replication.enabled() {
279 return Err(Error::Config(
280 g::t(lang, K::ErrReplicationNotConfigured).to_owned(),
281 ));
282 }
283 audit::record(
284 &state.store,
285 Some(cx),
286 Actor::User(user),
287 AuditAction::ReplicationReconcileStarted,
288 Target {
289 kind: "replication",
290 id: "",
291 name: "",
292 },
293 serde_json::json!({ "trigger": "manual" }),
294 )
295 .await;
296 state.replication.request_reconcile();
297 Ok(())
298}
299
300#[cfg(test)]
301mod tests {
302 use super::*;
303 use forge_core::model::Visibility;
304 use topcoat::router::request::Request;
305 use topcoat::router::{Body, Router, to_bytes};
306
307 use crate::auth::SESSION_COOKIE;
308 use crate::test_support::{app_router, create_session, test_config, test_state};
309
310 const ADMIN_TOKEN: &str = "replication-admin-session";
311 const USER_TOKEN: &str = "replication-user-session";
312 const CSRF: &str = "replication-csrf";
313 const TARGET: &str = "sb@standby.example:/srv/forge-standby";
314
315 struct Fixture {
316 _dir: tempfile::TempDir,
317 router: Router,
318 state: AppState,
319 repo_id: String,
320 }
321
322 /// An instance with an admin, a plain user and one repo. Replication is configured when asked.
323 async fn fixture(configured: bool) -> Fixture {
324 let dir = tempfile::tempdir().expect("tempdir");
325 let data_dir = dir.path().join("data");
326 std::fs::create_dir_all(&data_dir).expect("data dir");
327 let mut state = test_state(&data_dir).await;
328 let mut config = test_config(&data_dir);
329 if configured {
330 config.replication.target = TARGET.into();
331 }
332 state.config = std::sync::Arc::new(config);
333 let admin = state
334 .store
335 .create_user("root", None, "h", true)
336 .await
337 .expect("admin");
338 let user = state
339 .store
340 .create_user("plain", None, "h", false)
341 .await
342 .expect("user");
343 create_session(&state.store, &admin.id, ADMIN_TOKEN, CSRF).await;
344 create_session(&state.store, &user.id, USER_TOKEN, CSRF).await;
345 let repo = state
346 .store
347 .create_repo(&admin.id, "demo", "", Visibility::Private, "main")
348 .await
349 .expect("repo");
350 let router = app_router(state.clone());
351 Fixture {
352 _dir: dir,
353 router,
354 state,
355 repo_id: repo.id,
356 }
357 }
358
359 async fn send(
360 router: &Router,
361 method: &str,
362 uri: &str,
363 token: Option<&str>,
364 form: Option<String>,
365 ) -> (u16, String) {
366 let mut builder = Request::builder().method(method).uri(uri);
367 if let Some(token) = token {
368 builder = builder.header("cookie", format!("__Host-{SESSION_COOKIE}={token}"));
369 }
370 let body = match form {
371 Some(form) => {
372 builder = builder.header("content-type", "application/x-www-form-urlencoded");
373 Body::from(form)
374 }
375 None => Body::empty(),
376 };
377 let res = router.handle(builder.body(body).expect("request")).await;
378 let status = res.status().as_u16();
379 let bytes = to_bytes(res.into_body(), usize::MAX).await.expect("body");
380 (status, String::from_utf8_lossy(&bytes).into_owned())
381 }
382
383 #[tokio::test]
384 async fn page_and_reconcile_are_404_for_anonymous_and_non_admins() {
385 let fx = fixture(true).await;
386 let (anon, _) = send(&fx.router, "GET", "/admin/replication", None, None).await;
387 assert_eq!(anon, 404);
388 let (user, _) = send(
389 &fx.router,
390 "GET",
391 "/admin/replication",
392 Some(USER_TOKEN),
393 None,
394 )
395 .await;
396 assert_eq!(user, 404);
397 let (posted, _) = send(
398 &fx.router,
399 "POST",
400 "/admin/replication/reconcile",
401 Some(USER_TOKEN),
402 Some(format!("csrf={CSRF}")),
403 )
404 .await;
405 assert_eq!(posted, 404);
406 assert!(!fx.state.replication.take_reconcile_request());
407 }
408
409 #[tokio::test]
410 async fn page_shows_target_db_and_repo_health_without_secrets() {
411 let fx = fixture(true).await;
412 let store = &fx.state.store;
413 store
414 .record_replication_ok(DB_SUBJECT, None)
415 .await
416 .expect("db ok");
417 store
418 .record_replication_ok(&fx.repo_id, Some("abc"))
419 .await
420 .expect("repo ok");
421 store
422 .record_replication_error("gone-repo-id", "remove failed")
423 .await
424 .expect("stale");
425
426 let (status, body) = send(
427 &fx.router,
428 "GET",
429 "/admin/replication",
430 Some(ADMIN_TOKEN),
431 None,
432 )
433 .await;
434 assert_eq!(status, 200);
435 assert!(body.contains("sb@standby.example:/srv/forge-standby"));
436 assert!(body.contains("root/demo"), "repo name is shown");
437 assert!(body.contains("ok"));
438 assert!(
439 body.contains("gone-repo-id"),
440 "stale rows are shown until cleaned"
441 );
442 assert!(body.contains("pending cleanup"));
443 assert!(body.contains("Reconcile now"));
444 }
445
446 #[tokio::test]
447 async fn unconfigured_page_says_so_and_reconcile_is_refused() {
448 let fx = fixture(false).await;
449 let (status, body) = send(
450 &fx.router,
451 "GET",
452 "/admin/replication",
453 Some(ADMIN_TOKEN),
454 None,
455 )
456 .await;
457 assert_eq!(status, 200);
458 assert!(body.contains("Replication is off"));
459
460 let (status, _) = send(
461 &fx.router,
462 "POST",
463 "/admin/replication/reconcile",
464 Some(ADMIN_TOKEN),
465 Some(format!("csrf={CSRF}")),
466 )
467 .await;
468 assert_eq!(status, 303);
469 assert!(!fx.state.replication.take_reconcile_request());
470 let audit = fx
471 .state
472 .store
473 .list_audit(&Default::default(), None, 10)
474 .await
475 .expect("audit");
476 assert!(
477 !audit
478 .iter()
479 .any(|e| e.action == AuditAction::ReplicationReconcileStarted)
480 );
481 }
482
483 #[tokio::test]
484 async fn reconcile_needs_csrf_and_is_audited() {
485 let fx = fixture(true).await;
486 let (bad, _) = send(
487 &fx.router,
488 "POST",
489 "/admin/replication/reconcile",
490 Some(ADMIN_TOKEN),
491 Some("csrf=wrong".into()),
492 )
493 .await;
494 assert_eq!(bad, 403);
495 assert!(!fx.state.replication.take_reconcile_request());
496
497 let (status, _) = send(
498 &fx.router,
499 "POST",
500 "/admin/replication/reconcile",
501 Some(ADMIN_TOKEN),
502 Some(format!("csrf={CSRF}")),
503 )
504 .await;
505 assert_eq!(status, 303);
506 assert!(fx.state.replication.take_reconcile_request());
507 let audit = fx
508 .state
509 .store
510 .list_audit(&Default::default(), None, 10)
511 .await
512 .expect("audit");
513 assert!(
514 audit
515 .iter()
516 .any(|e| e.action == AuditAction::ReplicationReconcileStarted),
517 "the request is audited"
518 );
519 }
520
521 #[test]
522 fn health_reads_errors_before_age() {
523 let cutoff = "2026-01-01T00:00:00.000000000Z";
524 let fresh = ReplicationStatus {
525 subject: "r".into(),
526 last_ok_at: Some("2026-06-01T00:00:00.000000000Z".into()),
527 last_error: None,
528 last_attempt_at: None,
529 refs_sha256: None,
530 };
531 assert_eq!(health(Some(&fresh), cutoff), Health::Ok);
532 let old = ReplicationStatus {
533 last_ok_at: Some("2025-06-01T00:00:00.000000000Z".into()),
534 ..fresh.clone()
535 };
536 assert_eq!(health(Some(&old), cutoff), Health::Lagging);
537 let failing = ReplicationStatus {
538 last_error: Some("boom".into()),
539 ..fresh.clone()
540 };
541 assert_eq!(health(Some(&failing), cutoff), Health::Error);
542 assert_eq!(health(None, cutoff), Health::Never);
543 }
544}