headless_lms_server/controllers/main_frontend/credit_registration_admin/
phases.rs

1//! The Workers tab: one row per pipeline phase, with the queue each is responsible for.
2//!
3//! Phases, not pods. Pausing `import` while `verify` keeps running is a real incident move that no
4//! pod-level control can express, and the pause here is our own flag — the k8s status page still
5//! answers whether the process hosting a phase is up.
6
7use std::collections::HashMap;
8
9use headless_lms_models::credit_registration_phase_state::{
10    self, CreditRegistrationPhaseState as PhaseStateRow,
11};
12use headless_lms_models::credit_registrations::{self, CreditRegistrationState};
13use utoipa::ToSchema;
14
15use crate::domain::credit_registration::health::{
16    PHASE_CONSECUTIVE_FAILURE_LIMIT, PHASE_HEARTBEAT_INTERVAL_MULTIPLIER, is_heartbeat_late,
17};
18use crate::domain::credit_registration_phases::CreditRegistrationPhase;
19use crate::prelude::*;
20
21use super::authorize_credit_registration_admin;
22
23/// One phase as the Workers tab renders it.
24///
25/// Wider than `CreditRegistrationPhaseStatus`, which the pause/resume/run-now responses return: this
26/// one also carries the last error, the run window and the queue.
27#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, ToSchema)]
28pub struct CreditRegistrationPhaseRow {
29    pub phase: String,
30    /// The worker process whose loop runs the phase. Rows are grouped by it, because a dead pod
31    /// makes every phase inside it go stale at once and that reads as one fault, not seven.
32    pub process_name: String,
33    pub expected_interval_secs: i32,
34    pub last_heartbeat_at: Option<DateTime<Utc>>,
35    pub last_run_started_at: Option<DateTime<Utc>>,
36    pub last_run_finished_at: Option<DateTime<Utc>>,
37    pub last_success_at: Option<DateTime<Utc>>,
38    pub next_run_at: Option<DateTime<Utc>>,
39    pub items_processed_last_run: Option<i32>,
40    pub items_failed_last_run: Option<i32>,
41    pub consecutive_failures: i32,
42    /// Our own wording or the study registry's code, never its prose.
43    pub last_error: Option<String>,
44    pub paused_at: Option<DateTime<Utc>>,
45    pub paused_by_user_id: Option<Uuid>,
46    pub pause_reason: Option<String>,
47    /// No implementation is registered for the phase yet, so it has never reported and will not.
48    pub implemented: bool,
49    /// Computed server-side: a page comparing its own clock against a server timestamp misjudges
50    /// this on a skewed client.
51    pub seconds_since_heartbeat: Option<i64>,
52    pub last_run_duration_secs: Option<i64>,
53    /// Always `false` while paused or never heartbeated.
54    pub heartbeat_late: bool,
55    pub failing: bool,
56    /// The ledger states nothing but this phase moves a row out of. Empty for the phases whose work
57    /// is not a ledger state: `materialize` waits on completions, the syncer's phases on modules.
58    pub owned_states: Vec<CreditRegistrationState>,
59    /// Live rows in `owned_states`, or `None` where there are none to own — which is not the same
60    /// as an empty queue.
61    pub queue_depth: Option<i64>,
62}
63
64#[derive(Debug, Serialize, Deserialize, PartialEq, Clone, ToSchema)]
65pub struct CreditRegistrationPhaseList {
66    /// In process then pipeline order, so the grouping is a fold over the list.
67    pub phases: Vec<CreditRegistrationPhaseRow>,
68    pub heartbeat_interval_multiplier: i32,
69    pub consecutive_failure_limit: i32,
70    /// Every phase is stopped, which is what the kill switch does.
71    pub paused_globally: bool,
72}
73
74/**
75GET `/api/v0/main-frontend/credit-registration-admin/phases` - Every pipeline phase, its heartbeat
76and the queue it is responsible for.
77*/
78#[instrument(skip(pool))]
79#[utoipa::path(
80    get,
81    path = "/phases",
82    operation_id = "listCreditRegistrationPhases",
83    tag = "credit-registration-admin",
84    responses(
85        (status = 200, description = "One row per pipeline phase", body = CreditRegistrationPhaseList)
86    )
87)]
88pub async fn list_credit_registration_phases(
89    user: AuthUser,
90    pool: web::Data<PgPool>,
91) -> ControllerResult<web::Json<CreditRegistrationPhaseList>> {
92    let mut conn = pool.acquire().await?;
93    let token = authorize_credit_registration_admin(&mut conn, user.id).await?;
94
95    let depths: HashMap<CreditRegistrationState, i64> =
96        credit_registrations::count_by_state(&mut conn)
97            .await?
98            .into_iter()
99            .collect();
100    let now = Utc::now();
101    let mut phases: Vec<CreditRegistrationPhaseRow> =
102        credit_registration_phase_state::get_all(&mut conn)
103            .await?
104            .into_iter()
105            .map(|row| to_phase_row(row, now, &depths))
106            .collect();
107    phases.sort_by_key(|row| {
108        (
109            row.process_name.clone(),
110            CreditRegistrationPhase::from_phase_name(&row.phase)
111                .and_then(|phase| {
112                    CreditRegistrationPhase::ALL
113                        .iter()
114                        .position(|p| *p == phase)
115                })
116                .unwrap_or(usize::MAX),
117        )
118    });
119
120    token.authorized_ok(web::Json(CreditRegistrationPhaseList {
121        paused_globally: !phases.is_empty() && phases.iter().all(|row| row.paused_at.is_some()),
122        phases,
123        heartbeat_interval_multiplier: PHASE_HEARTBEAT_INTERVAL_MULTIPLIER,
124        consecutive_failure_limit: PHASE_CONSECUTIVE_FAILURE_LIMIT,
125    }))
126}
127
128fn to_phase_row(
129    row: PhaseStateRow,
130    now: DateTime<Utc>,
131    depths: &HashMap<CreditRegistrationState, i64>,
132) -> CreditRegistrationPhaseRow {
133    let known = CreditRegistrationPhase::from_phase_name(&row.phase);
134    let owned_states: Vec<CreditRegistrationState> = known
135        .map(|phase| phase.owned_states().to_vec())
136        .unwrap_or_default();
137    let seconds_since_heartbeat = row.last_heartbeat_at.map(|at| (now - at).num_seconds());
138    let heartbeat_late = is_heartbeat_late(
139        row.last_heartbeat_at,
140        row.expected_interval_secs,
141        row.paused_at,
142        now,
143    );
144    CreditRegistrationPhaseRow {
145        implemented: known.is_some(),
146        queue_depth: (!owned_states.is_empty()).then(|| {
147            owned_states
148                .iter()
149                .map(|state| depths.get(state).copied().unwrap_or(0))
150                .sum()
151        }),
152        owned_states,
153        seconds_since_heartbeat,
154        heartbeat_late,
155        failing: row.paused_at.is_none()
156            && row.consecutive_failures >= PHASE_CONSECUTIVE_FAILURE_LIMIT,
157        last_run_duration_secs: row
158            .last_run_started_at
159            .zip(row.last_run_finished_at)
160            .map(|(started, finished)| (finished - started).num_seconds()),
161        phase: row.phase,
162        process_name: row.process_name,
163        expected_interval_secs: row.expected_interval_secs,
164        last_heartbeat_at: row.last_heartbeat_at,
165        last_run_started_at: row.last_run_started_at,
166        last_run_finished_at: row.last_run_finished_at,
167        last_success_at: row.last_success_at,
168        next_run_at: row.next_run_at,
169        items_processed_last_run: row.items_processed_last_run,
170        items_failed_last_run: row.items_failed_last_run,
171        consecutive_failures: row.consecutive_failures,
172        last_error: row.last_error,
173        paused_at: row.paused_at,
174        paused_by_user_id: row.paused_by_user_id,
175        pause_reason: row.pause_reason,
176    }
177}
178
179pub fn _add_routes(cfg: &mut ServiceConfig) {
180    cfg.route("/phases", web::get().to(list_credit_registration_phases));
181}
headless_lms_server/controllers/main_frontend/credit_registration_admin/phases.rs

headless_lms_server/controllers/main_frontend/credit_registration_admin/
phases.rs
