headless_lms_server/domain/credit_registration_phases/
worker_loop.rs

1//! The loop both credit registration workers run.
2//!
3//! The processes differ only in which phases they own and how often they look, so the scheduling
4//! lives here instead of in each of them.
5
6use std::{collections::HashMap, sync::Arc, time::Duration};
7
8use chrono::{DateTime, Utc};
9use sqlx::PgPool;
10
11use headless_lms_base::config::ApplicationConfiguration;
12use headless_lms_models::credit_registration_phase_state::{
13    self, CreditRegistrationPhaseState, set_next_run_at,
14};
15use headless_lms_models::suotar_api_calls::PgSuotarCallAudit;
16use headless_lms_utils::services::suotar::SuotarClient;
17
18use crate::programs::periodic_worker::{
19    PeriodicWorkerConfig, is_db_disconnect, run_periodic_worker,
20};
21
22use super::{CreditRegistrationPhase, PhaseContext, PhaseScope, PhaseTick, run_phase_once};
23
24/// How often the loop looks for a due phase; each phase's own interval lives in
25/// `credit_registration_phase_state`.
26const TICK_INTERVAL_SECS: u64 = 10;
27
28/// Ten minutes of ticks. The per-phase heartbeat in the database is the machine-readable half.
29const STILL_RUNNING_MESSAGE_TICKS: u32 = 60;
30
31/// Runs the phases this process owns until it is stopped, matching `process_name` against
32/// [`CreditRegistrationPhase::process_name`].
33pub async fn run(
34    process_name: &'static str,
35    db_pool: PgPool,
36    app_configuration: ApplicationConfiguration,
37    still_running_message: &str,
38) -> anyhow::Result<()> {
39    let suotar_client = SuotarClient::new(
40        &app_configuration.suotar_configuration,
41        Arc::new(PgSuotarCallAudit::new(db_pool.clone())),
42    );
43    let phases: Vec<CreditRegistrationPhase> = CreditRegistrationPhase::ALL
44        .into_iter()
45        .filter(|phase| phase.process_name() == process_name)
46        .collect();
47
48    run_periodic_worker(
49        PeriodicWorkerConfig {
50            tick_interval: Duration::from_secs(TICK_INTERVAL_SECS),
51            still_running_every: STILL_RUNNING_MESSAGE_TICKS,
52            still_running_message,
53            initial_ticks: 0,
54            // A slow iteration should push later ticks out, not fire them back to back (tokio's
55            // default).
56            delay_missed_ticks: true,
57        },
58        async || {
59            let ctx =
60                PhaseContext::from_app(&db_pool, &suotar_client, &app_configuration, process_name);
61            let states = match phase_states(&db_pool).await {
62                Ok(states) => states,
63                Err(error) => {
64                    log_failure(
65                        process_name,
66                        "Reading the credit registration phase states",
67                        &error,
68                    );
69                    return Ok(());
70                }
71            };
72            for phase in &phases {
73                let Some(state) = states.get(phase.as_str()) else {
74                    error!(
75                        "Credit registration phase {} has no phase-state row.",
76                        phase.as_str()
77                    );
78                    continue;
79                };
80                if !is_due(state, Utc::now()) {
81                    continue;
82                }
83                // Logged and swallowed: one phase failing must not stop the others, and the phase-state
84                // row already carries the failure for the dashboard.
85                if let Err(error) = run_due_phase(&ctx, *phase, state).await {
86                    log_failure(
87                        process_name,
88                        &format!("Credit registration phase {}", phase.as_str()),
89                        &error,
90                    );
91                }
92            }
93            Ok(())
94        },
95    )
96    .await
97}
98
99/// Every phase's state in one read, so a tick costs one query rather than one per owned phase.
100async fn phase_states(
101    pool: &PgPool,
102) -> anyhow::Result<HashMap<String, CreditRegistrationPhaseState>> {
103    let mut conn = pool.acquire().await?;
104    Ok(credit_registration_phase_state::get_all(&mut conn)
105        .await?
106        .into_iter()
107        .map(|state| (state.phase.clone(), state))
108        .collect())
109}
110
111fn log_failure(process_name: &str, subject: &str, error: &anyhow::Error) {
112    error!("{subject} failed: {error}");
113    if is_db_disconnect(error.source()) {
114        info!("{process_name} may have lost its connection to the database.");
115    }
116}
117
118/// Runs one due phase, and schedules the next run.
119async fn run_due_phase(
120    ctx: &PhaseContext<'_>,
121    phase: CreditRegistrationPhase,
122    state: &CreditRegistrationPhaseState,
123) -> anyhow::Result<()> {
124    let mut conn = ctx.pool.acquire().await?;
125    // Stamped before the work, so a phase whose iteration takes longer than its interval does not
126    // run back to back.
127    set_next_run_at(
128        &mut conn,
129        phase.as_str(),
130        Utc::now() + chrono::Duration::seconds(state.expected_interval_secs.into()),
131    )
132    .await?;
133    drop(conn);
134
135    // Always unscoped: a worker that narrowed would leave rows nobody sweeps.
136    match run_phase_once(ctx, phase, &PhaseScope::default()).await? {
137        PhaseTick::Ran(outcome) if outcome.items_processed > 0 || outcome.items_failed > 0 => {
138            info!(
139                "Credit registration phase {} processed {} rows, {} of them unsuccessfully.",
140                phase.as_str(),
141                outcome.items_processed,
142                outcome.items_failed
143            );
144        }
145        // Nothing to do, paused, or waiting out a cooldown: quiet on purpose, because the heartbeat
146        // is what says the loop is alive.
147        _ => {}
148    }
149    Ok(())
150}
151
152/// A phase is due when an admin asked for it, or when its interval has elapsed since it last began.
153fn is_due(state: &CreditRegistrationPhaseState, now: DateTime<Utc>) -> bool {
154    if let Some(next_run_at) = state.next_run_at {
155        return next_run_at <= now;
156    }
157    state.last_run_started_at.is_none_or(|started| {
158        (now - started).num_seconds() >= i64::from(state.expected_interval_secs)
159    })
160}
161
162#[cfg(test)]
163mod tests {
164    use super::*;
165    use uuid::Uuid;
166
167    fn state(
168        next_run_at: Option<DateTime<Utc>>,
169        last_run_started_at: Option<DateTime<Utc>>,
170    ) -> CreditRegistrationPhaseState {
171        CreditRegistrationPhaseState {
172            id: Uuid::new_v4(),
173            created_at: Utc::now(),
174            updated_at: Utc::now(),
175            deleted_at: None,
176            phase: "verify".to_string(),
177            process_name: "credit-registrar".to_string(),
178            expected_interval_secs: 60,
179            last_heartbeat_at: None,
180            last_run_started_at,
181            last_run_finished_at: None,
182            last_success_at: None,
183            next_run_at,
184            items_processed_last_run: None,
185            items_failed_last_run: None,
186            consecutive_failures: 0,
187            last_error: None,
188            paused_at: None,
189            paused_by_user_id: None,
190            pause_reason: None,
191        }
192    }
193
194    #[test]
195    fn a_phase_that_has_never_run_is_due() {
196        assert!(is_due(&state(None, None), Utc::now()));
197    }
198
199    #[test]
200    fn a_phase_is_due_again_once_its_interval_has_elapsed() {
201        let now = Utc::now();
202        assert!(!is_due(
203            &state(None, Some(now - chrono::Duration::seconds(30))),
204            now
205        ));
206        assert!(is_due(
207            &state(None, Some(now - chrono::Duration::seconds(90))),
208            now
209        ));
210    }
211
212    /// How "run now" works: the admin endpoint stamps the timestamp and the loop notices.
213    #[test]
214    fn an_explicit_next_run_beats_the_interval() {
215        let now = Utc::now();
216        let asked_for = state(Some(now), Some(now));
217        assert!(is_due(&asked_for, now));
218
219        let scheduled = state(Some(now + chrono::Duration::seconds(30)), None);
220        assert!(!is_due(&scheduled, now));
221    }
222
223    /// A phase belonging to neither process would look merely idle rather than unrun.
224    #[test]
225    fn the_two_processes_between_them_own_every_phase() {
226        let mut owned: Vec<&str> = Vec::new();
227        for process in ["credit-registrar", "suotar-syncer"] {
228            owned.extend(
229                CreditRegistrationPhase::ALL
230                    .into_iter()
231                    .filter(|phase| phase.process_name() == process)
232                    .map(|phase| phase.as_str()),
233            );
234        }
235        assert_eq!(owned.len(), CreditRegistrationPhase::ALL.len());
236    }
237}
headless_lms_server/domain/credit_registration_phases/worker_loop.rs

headless_lms_server/domain/credit_registration_phases/
worker_loop.rs
