headless_lms_credit_registration/runtime/
dispatch.rs

1//! Running one iteration of one phase, and the bookkeeping around it: the pause and scope checks,
2//! the heartbeat, the circuit breakers and the limiter.
3
4use chrono::{DateTime, TimeDelta, Utc};
5use headless_lms_models::credit_registration_phase_state::{self, PhaseRunOutcome};
6use headless_lms_models::library::credit_registration::scrub::scrub_text;
7use headless_lms_models::library::credit_registration::sisu_day_gap;
8use headless_lms_utils::services::suotar::SuotarClient;
9use sqlx::PgPool;
10use tokio_util::sync::CancellationToken;
11
12use super::heartbeat::keep_alive;
13use super::suotar::{
14    SuotarStudyRegistry, max_study_registry_wait, report_breakers, report_rate_limits,
15};
16use crate::error::CreditRegistrationResult;
17use crate::error_reports::ErrorReporter;
18use crate::phase::{CreditRegistrationPhase, WorkerProcess};
19use crate::use_cases::batch_flow::BatchFlowContext;
20use crate::use_cases::{
21    config_validation, enrolment_discovery, import, ledger_snapshot, legacy_mirror, link_emails,
22    materialize, preconditions, resolve_enrolments, retention_sweep, student_notifications, verify,
23};
24use crate::workflow::Counts;
25use headless_lms_models::credit_registrations::RegistrationScope;
26
27/// What one dispatch attempt did.
28#[derive(Debug, Clone, PartialEq)]
29pub enum PhaseTick {
30    Ran(PhaseRunOutcome),
31    /// The phase legitimately did nothing; not counted as a failure.
32    Skipped(PhaseSkipReason),
33    /// The scope names something this phase cannot narrow on; refused rather than run wide.
34    ScopeNotSupported,
35}
36
37#[derive(Debug, Clone, Copy, PartialEq, Eq)]
38pub enum PhaseSkipReason {
39    Paused,
40    CircuitBreakerOpen,
41    AccountLinkingDisabled,
42    /// Finland is on the next day and Sisu is not; see [`sisu_day_gap`].
43    SisuDayGap,
44}
45
46/// Who runs a phase iteration.
47#[derive(Debug, Clone, Copy, PartialEq, Eq)]
48pub enum Runner<'a> {
49    /// A worker process's loop, the only holder of the breaker and limiter state the dashboard
50    /// shows for the phases it owns.
51    Worker(WorkerProcess),
52    /// Anyone else, such as the test tick endpoint, named for the audit log. Its in-memory state
53    /// would overwrite the worker's on the dashboard, so it reports none.
54    Other(&'a str),
55}
56
57impl Runner<'_> {
58    /// What goes into the audit log's `worker_name` alongside the phase.
59    pub fn caller(&self) -> &str {
60        match self {
61            Self::Worker(process) => process.as_str(),
62            Self::Other(caller) => caller,
63        }
64    }
65
66    pub fn owning_process(&self) -> Option<WorkerProcess> {
67        match self {
68            Self::Worker(process) => Some(*process),
69            Self::Other(_) => None,
70        }
71    }
72}
73
74/// Everything a phase iteration needs from its caller: the worker loop or the test tick endpoint.
75pub struct PhaseContext<'a> {
76    pub pool: &'a PgPool,
77    pub suotar_client: &'a SuotarClient,
78    /// Shortens the circuit breaker's cooldown to something a test can wait out.
79    pub test_mode: bool,
80    pub runner: Runner<'a>,
81    /// Absolute base for links in queued mail, which outlive the process that wrote them.
82    pub base_url: &'a str,
83    /// Linking mails go only to people enrolled at or after it. `None`, none are sent and discovery
84    /// only wakes linked students' registrations.
85    pub account_linking_since: Option<DateTime<Utc>>,
86    /// The worker's SIGTERM; `None` for a run no signal can stop, such as an on-demand one.
87    pub shutdown: Option<&'a CancellationToken>,
88}
89
90impl<'a> PhaseContext<'a> {
91    /// Builds a context from the application configuration, the shape every construction site
92    /// starts from.
93    pub fn from_app(
94        pool: &'a PgPool,
95        suotar_client: &'a SuotarClient,
96        app_conf: &'a headless_lms_base::config::ApplicationConfiguration,
97        runner: Runner<'a>,
98    ) -> Self {
99        Self {
100            pool,
101            suotar_client,
102            test_mode: app_conf.test_mode,
103            runner,
104            base_url: &app_conf.base_url,
105            account_linking_since: app_conf.suotar_configuration.account_linking_since,
106            shutdown: None,
107        }
108    }
109}
110
111/// The audit log's `worker_name`, which the database caps at 64 characters.
112pub(super) fn worker_name(caller: &str, phase: CreditRegistrationPhase) -> String {
113    format!("{caller}/{}", phase.as_str())
114}
115
116/// Runs exactly one iteration of one phase.
117#[tracing::instrument(
118    skip_all,
119    fields(phase = phase.as_str(), caller = ctx.runner.caller(), scope = ?scope)
120)]
121pub async fn run_phase_once(
122    ctx: &PhaseContext<'_>,
123    phase: CreditRegistrationPhase,
124    scope: &RegistrationScope,
125) -> CreditRegistrationResult<PhaseTick> {
126    // Before the pause check: a caller whose narrowing cannot be honoured must not be told it ran.
127    if !phase.spec().scope.covers(scope) {
128        return Ok(PhaseTick::ScopeNotSupported);
129    }
130    let mut conn = ctx.pool.acquire().await?;
131    if credit_registration_phase_state::is_paused(&mut conn, phase.as_str()).await? {
132        return Ok(PhaseTick::Skipped(PhaseSkipReason::Paused));
133    }
134    // A scoped run writes nothing to the phase-state row: that row describes the workers, and a
135    // test's traffic in it would make a dead worker look alive to the heartbeat alert.
136    let bookkeeping = scope.is_unscoped();
137    let is_own_worker = ctx.runner.owning_process() == Some(phase.spec().process);
138    // Before the breaker check, unlike the pause above, which health.rs excludes from the staleness
139    // alert by itself. A cooldown is a worker deliberately waiting, not a worker that died, and
140    // skipping the heartbeat through it would raise a critical alert within a tick or two.
141    if bookkeeping {
142        credit_registration_phase_state::heartbeat(&mut conn, phase.as_str()).await?;
143        trace!(phase = phase.as_str(), "Wrote phase heartbeat");
144    }
145    // After the heartbeat, like the breaker check below: a switched-off phase is idle, not dead.
146    if phase.spec().is_account_linking_only && ctx.account_linking_since.is_none() {
147        return Ok(PhaseTick::Skipped(PhaseSkipReason::AccountLinkingDisabled));
148    }
149    // Not in test mode: the system tests run at every hour, the gap's included.
150    if phase.spec().waits_out_sisu_day_gap
151        && !ctx.test_mode
152        && let Some(gap_end) = sisu_day_gap::current_gap_end(&mut conn).await?
153    {
154        let moved = sisu_day_gap::spread_imports_past_gap(&mut conn, gap_end).await?;
155        if moved > 0 {
156            info!(
157                phase = phase.as_str(),
158                moved,
159                %gap_end,
160                "Spread imports due in the Sisu day gap over the hours after it"
161            );
162        }
163        if bookkeeping {
164            credit_registration_phase_state::record_deliberate_idle_run(&mut conn, phase.as_str())
165                .await?;
166            if is_own_worker {
167                report_breakers(&mut conn, phase).await?;
168            }
169        }
170        return Ok(PhaseTick::Skipped(PhaseSkipReason::SisuDayGap));
171    }
172    let mut registry = match SuotarStudyRegistry::admit(
173        ctx.suotar_client,
174        worker_name(ctx.runner.caller(), phase),
175        phase,
176        scope,
177        ctx.test_mode,
178    ) {
179        Ok(registry) => registry,
180        Err(skip) => {
181            if bookkeeping && is_own_worker {
182                report_breakers(&mut conn, phase).await?;
183            }
184            return Ok(PhaseTick::Skipped(skip));
185        }
186    };
187    drop(conn);
188
189    let body = if bookkeeping {
190        tokio::select! {
191            body = run_body(ctx, phase, scope, &mut registry) => body,
192            never = keep_alive(ctx.pool, phase) => match never {},
193        }
194    } else {
195        run_body(ctx, phase, scope, &mut registry).await
196    };
197    let failure = registry.finish();
198    // Only a real `CreditRegistrationError` carries a backtrace and span trace; `failure` and
199    // `counts.finding` are already plain messages.
200    let stack_trace = match &body {
201        Err(error) => Some(format!("{error:?}")),
202        Ok(_) => None,
203    };
204    let outcome = match body {
205        Ok(counts) => {
206            let outcome = PhaseRunOutcome {
207                items_processed: counts.processed_count(),
208                items_waiting: counts.waiting_count(),
209                items_failed: counts.failed_count(),
210                error: failure.or(counts.into_finding()),
211            };
212            if let Some(error) = &outcome.error {
213                warn!(phase = phase.as_str(), error = %error, "Credit registration phase iteration recorded a failure");
214            }
215            outcome
216        }
217        Err(error) => {
218            error!(phase = phase.as_str(), error = %error, "Credit registration phase iteration aborted");
219            PhaseRunOutcome {
220                error: Some(scrub_text(&error.cause_chain())),
221                ..PhaseRunOutcome::default()
222            }
223        }
224    };
225    if let Some(error) = &outcome.error {
226        ErrorReporter::new(ctx.pool, ctx.runner.owning_process(), phase)
227            .report(error, stack_trace, serde_json::json!({}))
228            .await;
229    }
230    if bookkeeping {
231        let mut conn = ctx.pool.acquire().await?;
232        credit_registration_phase_state::record_run(&mut conn, phase.as_str(), &outcome).await?;
233        if is_own_worker {
234            report_rate_limits(&mut conn, phase).await?;
235            report_breakers(&mut conn, phase).await?;
236        }
237    }
238    Ok(PhaseTick::Ran(outcome))
239}
240
241/// The one place a phase implementation is registered: exhaustive over [`CreditRegistrationPhase`],
242/// so a variant added there without an arm here fails to compile. Each phase gets only what it
243/// touches, and a phase that asks the study registry gets the iteration's registry beside it.
244async fn run_body(
245    ctx: &PhaseContext<'_>,
246    phase: CreditRegistrationPhase,
247    scope: &RegistrationScope,
248    registry: &mut SuotarStudyRegistry<'_>,
249) -> CreditRegistrationResult<Counts> {
250    let pool = ctx.pool;
251    let batch_flow = BatchFlowContext {
252        pool,
253        scope,
254        phase,
255        errors: ErrorReporter::new(pool, ctx.runner.owning_process(), phase),
256        shutdown: ctx.shutdown,
257        // Request timeouts are minutes, far inside the range.
258        study_registry_wait: TimeDelta::from_std(max_study_registry_wait(phase))
259            .unwrap_or(TimeDelta::zero()),
260    };
261    match phase {
262        CreditRegistrationPhase::Materialize => materialize::run(pool, scope).await,
263        CreditRegistrationPhase::Preconditions => preconditions::run(pool, scope).await,
264        CreditRegistrationPhase::ResolveEnrolments => {
265            resolve_enrolments::run(&batch_flow, registry).await
266        }
267        CreditRegistrationPhase::Import => import::run(&batch_flow, registry).await,
268        CreditRegistrationPhase::Verify => verify::run(&batch_flow, registry).await,
269        CreditRegistrationPhase::LegacyMirror => legacy_mirror::run(pool, scope).await,
270        CreditRegistrationPhase::StudentNotifications => {
271            student_notifications::run(
272                pool,
273                scope,
274                ctx.base_url,
275                ctx.account_linking_since.is_some(),
276            )
277            .await
278        }
279        CreditRegistrationPhase::EnrolmentDiscovery => {
280            enrolment_discovery::run(pool, scope, ctx.account_linking_since, registry).await
281        }
282        CreditRegistrationPhase::LinkEmails => link_emails::run(pool, scope, ctx.base_url).await,
283        CreditRegistrationPhase::ConfigValidation => {
284            config_validation::run(pool, scope, registry).await
285        }
286        CreditRegistrationPhase::RetentionSweep => retention_sweep::run(pool).await,
287        CreditRegistrationPhase::LedgerSnapshot => {
288            ledger_snapshot::run(pool, ctx.account_linking_since).await
289        }
290    }
291}
292
293#[cfg(test)]
294mod tests {
295    use super::*;
296
297    #[test]
298    fn the_audit_name_says_who_ran_the_phase() {
299        for caller in [WorkerProcess::CreditRegistrar.as_str(), "run-tick"] {
300            for phase in CreditRegistrationPhase::ALL {
301                let name = worker_name(caller, phase);
302                assert!(name.starts_with(caller));
303                assert!(name.ends_with(phase.as_str()));
304                assert!(name.len() <= 64, "{name}");
305            }
306        }
307    }
308}
headless_lms_credit_registration/runtime/dispatch.rs

headless_lms_credit_registration/runtime/
dispatch.rs
