headless_lms_server/domain/credit_registration_phases/
verify.rs

1//! The `verify` phase: asking the study registry what became of a submission.
2//!
3//! The only way out of `submission_uncertain`, and never a way back to `import`. Nothing here ever
4//! fails a row: the attainment may exist, and a failed row is one an admin retries, which for an
5//! uncertain submission would mean sending it twice.
6
7use headless_lms_base::error::backend_error::BackendError;
8use headless_lms_models::credit_registration_events::CreditRegistrationEventKind;
9use headless_lms_models::credit_registration_phase_state::PhaseRunOutcome;
10use headless_lms_models::credit_registrations::{
11    CreditRegistration, CreditRegistrationState, RequestPurpose, Transition, claim_due,
12    increment_verify_attempt_counts, request_item_id, schedule_next_attempts,
13    set_sisu_attainment_if_unclaimed, transition,
14};
15use headless_lms_models::library::credit_registration::classification::{map_code, settled_state};
16use headless_lms_models::library::credit_registration::enrolment_selection::attainment_matching_submission;
17use headless_lms_models::library::credit_registration::outcomes::{
18    Outcome, RowFacts, uncertain_recheck_outcome, verify_error_outcome,
19    verify_not_registered_outcome, verify_poll_lease_until,
20};
21use headless_lms_models::library::credit_registration::submission_context::get_submission_contexts;
22use headless_lms_utils::error::util_error::UtilError;
23use headless_lms_utils::prelude::Utc;
24use headless_lms_utils::services::suotar::{
25    EnrolmentResolutionResult, ResolveEnrolmentRequestItem, SuotarBatchResponse, SuotarCallContext,
26    SuotarEndpoint, SuotarItemStatus, SuotarResponseItem, VerifyAttainmentRequestItem,
27    VerifyAttainmentResult,
28};
29use sqlx::{Connection, PgConnection};
30
31use super::{
32    CreditRegistrationPhase, OutcomeEvent, PhaseContext, PhaseScope, Prepared, SuotarBatchPhase,
33    apply_outcome, counts_as_failed, row_facts, run_suotar_batch_phase,
34};
35
36/// Both states the poller owns. Withdrawal moves a row out of both, which is what stops the polling
37/// without any query having to know about withdrawal.
38const CLAIMED_STATES: [CreditRegistrationState; 2] = [
39    CreditRegistrationState::AwaitingVerification,
40    CreditRegistrationState::SubmissionUncertain,
41];
42
43/// One claimed row and the poll it was claimed for. The attempt count travels with it because it
44/// names the request item and sets the backoff the answer is scheduled by.
45struct Poll {
46    row: CreditRegistration,
47    attempt: i32,
48    submitted_attainment_id: String,
49}
50
51/// A row whose submission we lost track of: nothing to poll by, so the lookup goes through
52/// `resolve-enrolments` instead.
53struct Recovery {
54    row: CreditRegistration,
55    attempt: i32,
56}
57
58pub async fn run(ctx: &PhaseContext<'_>, scope: &PhaseScope) -> anyhow::Result<PhaseRunOutcome> {
59    let mut conn = ctx.pool.acquire().await?;
60    let mut tx = conn.begin().await?;
61    let claimed = claim_due(
62        &mut tx,
63        &CLAIMED_STATES,
64        scope,
65        SuotarEndpoint::VerifyAttainments.max_batch_size() as i64,
66    )
67    .await?;
68    // Counted before the call: the count is part of the request item id, so two polls of one row
69    // are distinguishable in the registry's log.
70    let attempts = increment_verify_attempt_counts(
71        &mut tx,
72        &claimed.iter().map(|row| row.id).collect::<Vec<_>>(),
73    )
74    .await?;
75    // Pushed out of reach before the request leaves, so a concurrent iteration cannot poll the same
76    // row. Each answer overwrites its own row's schedule.
77    let now = Utc::now();
78    let scheduled: Vec<_> = attempts
79        .iter()
80        .map(|(id, attempt)| (*id, verify_poll_lease_until(now, *attempt)))
81        .collect();
82    schedule_next_attempts(&mut tx, &scheduled).await?;
83
84    let mut polls = Vec::new();
85    let mut recoveries = Vec::new();
86    for row in claimed {
87        let Some(attempt) = attempts.get(&row.id).copied() else {
88            continue;
89        };
90        match row.submitted_attainment_id.clone() {
91            Some(submitted_attainment_id) => polls.push(Poll {
92                row,
93                attempt,
94                submitted_attainment_id,
95            }),
96            None if row.state == CreditRegistrationState::SubmissionUncertain => {
97                recoveries.push(Recovery { row, attempt })
98            }
99            None => {
100                warn!(
101                    "Credit registration {} is awaiting verification with no submitted attainment id.",
102                    row.id
103                );
104            }
105        }
106    }
107    tx.commit().await?;
108    // Held only for the claim; the two flows below re-acquire around their own Suotar calls.
109    drop(conn);
110
111    let mut outcome = PhaseRunOutcome::default();
112    if !polls.is_empty() {
113        add(
114            &mut outcome,
115            run_suotar_batch_phase(&mut VerifyPoll { polls }, ctx, scope).await?,
116        );
117    }
118    // Recoveries go out on `resolve-enrolments`, whose batch limit is smaller than the one these
119    // rows were claimed at, so an oversized set would be refused whole before anything was sent.
120    let batch_size = SuotarEndpoint::ResolveEnrolments.max_batch_size();
121    while !recoveries.is_empty() {
122        let rest = recoveries.split_off(batch_size.min(recoveries.len()));
123        let mut flow = UncertainRecovery { recoveries };
124        add(
125            &mut outcome,
126            run_suotar_batch_phase(&mut flow, ctx, scope).await?,
127        );
128        recoveries = rest;
129    }
130    Ok(outcome)
131}
132
133/// Sums what the two flows of one iteration did; the first error stands for the iteration.
134fn add(total: &mut PhaseRunOutcome, part: PhaseRunOutcome) {
135    total.items_processed += part.items_processed;
136    total.items_failed += part.items_failed;
137    total.error = total.error.take().or(part.error);
138}
139
140/// Polls the rows that have something to poll by.
141struct VerifyPoll {
142    polls: Vec<Poll>,
143}
144
145impl SuotarBatchPhase for VerifyPoll {
146    type Row = Poll;
147    type Item = VerifyAttainmentRequestItem;
148    type Result = VerifyAttainmentResult;
149
150    const ALL_TRANSIENT_ERROR: &'static str =
151        "Every verify poll came back transiently unavailable.";
152
153    /// The rows are claimed by the phase itself, which splits them between this flow and the
154    /// recovery one, so there is nothing left to decide here.
155    async fn prepare(
156        &mut self,
157        _ctx: &PhaseContext<'_>,
158        _conn: &mut PgConnection,
159        _scope: &PhaseScope,
160    ) -> anyhow::Result<Prepared<Self::Row, Self::Item>> {
161        Ok(Prepared {
162            sendable: std::mem::take(&mut self.polls)
163                .into_iter()
164                .map(|poll| {
165                    let item = VerifyAttainmentRequestItem {
166                        request_item_id: Self::request_item_id(&poll),
167                        submitted_attainment_id: poll.submitted_attainment_id.clone(),
168                    };
169                    (poll, item)
170                })
171                .collect(),
172            ..Prepared::default()
173        })
174    }
175
176    fn registration(poll: &Self::Row) -> &CreditRegistration {
177        &poll.row
178    }
179
180    fn request_item_id(poll: &Self::Row) -> String {
181        request_item_id(&poll.row, RequestPurpose::VerifyPoll(poll.attempt))
182    }
183
184    async fn send(
185        &self,
186        ctx: &PhaseContext<'_>,
187        rows: &[Self::Row],
188        items: Vec<Self::Item>,
189    ) -> Result<SuotarBatchResponse<Self::Result>, UtilError> {
190        ctx.suotar_client
191            .verify_attainments(
192                SuotarCallContext::new(ctx.worker_name(CreditRegistrationPhase::Verify))
193                    .for_registrations(rows.iter().map(|poll| poll.row.id).collect()),
194                items,
195            )
196            .await
197    }
198
199    async fn apply(
200        &self,
201        conn: &mut PgConnection,
202        poll: &Self::Row,
203        item: Option<&SuotarResponseItem<Self::Result>>,
204        event: OutcomeEvent<'_>,
205    ) -> anyhow::Result<bool> {
206        apply_poll_answer(conn, poll, item, event).await
207    }
208
209    /// Deliberately not the shared request-level outcome: a failure to ask proves nothing was or
210    /// was not created, and moving the row towards `failed_retryable` would let an admin resubmit
211    /// it. The iteration is still reported as failed, so the breaker sees it.
212    async fn apply_request_rejection(
213        &self,
214        conn: &mut PgConnection,
215        poll: &Self::Row,
216        request: &serde_json::Value,
217        error: &UtilError,
218    ) -> anyhow::Result<bool> {
219        apply_outcome(
220            conn,
221            &poll.row,
222            &verify_not_registered_outcome(poll.row.state, &poll.facts()),
223            OutcomeEvent {
224                message: Some("Could not verify this submission this time."),
225                error_message: Some(error.message()),
226                request: Some(request),
227                ..OutcomeEvent::default()
228            },
229            Some(poll.row.state),
230        )
231        .await?;
232        Ok(false)
233    }
234}
235
236impl Poll {
237    /// The count this poll was made under, not the one the row was claimed with, so the backoff
238    /// doubles once per poll.
239    fn facts(&self) -> RowFacts {
240        RowFacts {
241            verify_attempt_count: self.attempt,
242            ..row_facts(&self.row)
243        }
244    }
245}
246
247/// Applies one poll's answer. Everything but `registered` keeps the row where it is: `notRegistered`
248/// is a normal polling answer, and any other answer is one we will not act on blindly.
249async fn apply_poll_answer(
250    conn: &mut PgConnection,
251    poll: &Poll,
252    item: Option<&SuotarResponseItem<VerifyAttainmentResult>>,
253    event: OutcomeEvent<'_>,
254) -> anyhow::Result<bool> {
255    let row = &poll.row;
256    let registered = item.is_some_and(|item| {
257        item.status == SuotarItemStatus::Ok
258            && settled_state(SuotarEndpoint::VerifyAttainments, &item.code)
259                == Some(CreditRegistrationState::Registered)
260    });
261    if registered {
262        if let Some(result) = item.and_then(|item| item.result.as_ref()) {
263            set_sisu_attainment_if_unclaimed(
264                conn,
265                row.id,
266                &result.attainment.id,
267                Some(&result.attainment.attainment_type),
268            )
269            .await?;
270        }
271        apply_outcome(
272            conn,
273            row,
274            &Outcome {
275                // Confirmed, so whatever an operator was asked to look at is settled.
276                needs_admin_attention: Some(false),
277                ..Outcome::to(CreditRegistrationState::Registered)
278            },
279            event,
280            Some(row.state),
281        )
282        .await?;
283        return Ok(false);
284    }
285    let facts = poll.facts();
286    let outcome = item
287        .and_then(|item| map_code(SuotarEndpoint::VerifyAttainments, &item.code))
288        .map(|code| verify_error_outcome(row.state, code, &facts))
289        .unwrap_or_else(|| verify_not_registered_outcome(row.state, &facts));
290    apply_outcome(
291        conn,
292        row,
293        &outcome,
294        OutcomeEvent {
295            error_message: item
296                .and_then(|item| item.error.as_ref())
297                .map(|error| error.message.as_str()),
298            ..event
299        },
300        Some(row.state),
301    )
302    .await?;
303    Ok(counts_as_failed(&outcome))
304}
305
306/// Looks for the attainment a submission we lost track of would have produced. The row stays
307/// `submission_uncertain` unless it is found: never failed, never re-imported.
308struct UncertainRecovery {
309    recoveries: Vec<Recovery>,
310}
311
312impl SuotarBatchPhase for UncertainRecovery {
313    type Row = Recovery;
314    type Item = ResolveEnrolmentRequestItem;
315    type Result = EnrolmentResolutionResult;
316
317    const ALL_TRANSIENT_ERROR: &'static str =
318        "Every recovery lookup came back transiently unavailable.";
319
320    /// A row with nothing to ask about is left where it is: it is uncertain, which no answer of
321    /// ours may turn into a failure, and it is already scheduled for the next check.
322    async fn prepare(
323        &mut self,
324        _ctx: &PhaseContext<'_>,
325        conn: &mut PgConnection,
326        _scope: &PhaseScope,
327    ) -> anyhow::Result<Prepared<Self::Row, Self::Item>> {
328        let recoveries = std::mem::take(&mut self.recoveries);
329        let contexts = get_submission_contexts(
330            conn,
331            &recoveries
332                .iter()
333                .map(|recovery| recovery.row.id)
334                .collect::<Vec<_>>(),
335        )
336        .await?;
337        let mut prepared = Prepared::default();
338        for recovery in recoveries {
339            let Some(context) = contexts.get(&recovery.row.id) else {
340                continue;
341            };
342            let (Some(student_number), Some(course_code)) = (
343                recovery
344                    .row
345                    .student_number
346                    .clone()
347                    .or_else(|| context.student_number.clone()),
348                recovery
349                    .row
350                    .uh_course_code
351                    .clone()
352                    .or_else(|| context.uh_course_code.clone()),
353            ) else {
354                continue;
355            };
356            let item = ResolveEnrolmentRequestItem {
357                request_item_id: Self::request_item_id(&recovery),
358                student_number,
359                course_code,
360            };
361            prepared.sendable.push((recovery, item));
362        }
363        Ok(prepared)
364    }
365
366    fn registration(recovery: &Self::Row) -> &CreditRegistration {
367        &recovery.row
368    }
369
370    fn request_item_id(recovery: &Self::Row) -> String {
371        request_item_id(
372            &recovery.row,
373            RequestPurpose::UncertainRecovery(recovery.attempt),
374        )
375    }
376
377    async fn send(
378        &self,
379        ctx: &PhaseContext<'_>,
380        rows: &[Self::Row],
381        items: Vec<Self::Item>,
382    ) -> Result<SuotarBatchResponse<Self::Result>, UtilError> {
383        ctx.suotar_client
384            .resolve_enrolments(
385                SuotarCallContext::new(ctx.worker_name(CreditRegistrationPhase::Verify))
386                    .for_registrations(rows.iter().map(|recovery| recovery.row.id).collect()),
387                items,
388            )
389            .await
390    }
391
392    async fn apply(
393        &self,
394        conn: &mut PgConnection,
395        recovery: &Self::Row,
396        item: Option<&SuotarResponseItem<Self::Result>>,
397        event: OutcomeEvent<'_>,
398    ) -> anyhow::Result<bool> {
399        apply_recovery_answer(conn, recovery, item, event).await
400    }
401
402    /// Not the shared request-level outcome either: these rows must stay uncertain whatever the
403    /// call did.
404    async fn apply_request_rejection(
405        &self,
406        conn: &mut PgConnection,
407        recovery: &Self::Row,
408        request: &serde_json::Value,
409        error: &UtilError,
410    ) -> anyhow::Result<bool> {
411        apply_outcome(
412            conn,
413            &recovery.row,
414            &uncertain_recheck_outcome(&recovery.facts()),
415            OutcomeEvent {
416                message: Some("Could not look for the attainment this time."),
417                error_message: Some(error.message()),
418                request: Some(request),
419                ..OutcomeEvent::default()
420            },
421            Some(recovery.row.state),
422        )
423        .await?;
424        Ok(false)
425    }
426}
427
428impl Recovery {
429    /// The count this lookup was made under, so the recheck cadence advances once per lookup.
430    fn facts(&self) -> RowFacts {
431        RowFacts {
432            verify_attempt_count: self.attempt,
433            ..row_facts(&self.row)
434        }
435    }
436}
437
438async fn apply_recovery_answer(
439    conn: &mut PgConnection,
440    recovery: &Recovery,
441    item: Option<&SuotarResponseItem<EnrolmentResolutionResult>>,
442    event: OutcomeEvent<'_>,
443) -> anyhow::Result<bool> {
444    let row = &recovery.row;
445    let found = match item {
446        Some(item) if item.status == SuotarItemStatus::Ok => item
447            .result
448            .as_ref()
449            .zip(row.attainment_date)
450            .and_then(|(result, attainment_date)| {
451                attainment_matching_submission(
452                    &result.existing_attainments,
453                    attainment_date,
454                    row.grade_scale_id.as_deref().unwrap_or_default(),
455                    row.grade_id.as_deref().unwrap_or_default(),
456                )
457            }),
458        _ => None,
459    };
460    let Some(attainment) = found else {
461        apply_outcome(
462            conn,
463            row,
464            &uncertain_recheck_outcome(&recovery.facts()),
465            OutcomeEvent {
466                message: Some(
467                    "No matching attainment yet, so whether the submission landed is still unknown.",
468                ),
469                ..event
470            },
471            Some(row.state),
472        )
473        .await?;
474        // A row still waiting to be resolved is not a failed item; the recheck raises the admin
475        // flag after enough tries instead.
476        return Ok(false);
477    };
478    set_sisu_attainment_if_unclaimed(
479        conn,
480        row.id,
481        &attainment.id,
482        Some(&attainment.attainment_type),
483    )
484    .await?;
485    transition(
486        conn,
487        row.id,
488        &Transition {
489            event_kind: CreditRegistrationEventKind::SuotarResponse,
490            event_message: Some(
491                "The attainment this submission would have created is in the study registry, so \
492                 it landed after all."
493                    .to_string(),
494            ),
495            needs_admin_attention: Some(false),
496            suotar_api_call_id: event.suotar_api_call_id,
497            event_details: Some(
498                headless_lms_models::credit_registration_events::suotar_exchange_details(
499                    event.request,
500                    event.response,
501                ),
502            ),
503            expected_from_state: Some(row.state),
504            ..Transition::to(CreditRegistrationState::Duplicate)
505        },
506    )
507    .await?;
508    Ok(false)
509}
510
511#[cfg(test)]
512mod tests {
513    use headless_lms_models::credit_registrations::{
514        recovery_request_item_id, verify_request_item_id,
515    };
516
517    use super::*;
518
519    /// Nothing this phase claims can lead back to a batch.
520    #[test]
521    fn the_poller_owns_exactly_the_two_states_a_submission_can_still_be_in() {
522        assert!(CLAIMED_STATES.contains(&CreditRegistrationState::AwaitingVerification));
523        assert!(CLAIMED_STATES.contains(&CreditRegistrationState::SubmissionUncertain));
524        assert!(!CLAIMED_STATES.contains(&CreditRegistrationState::Submitting));
525        assert!(!CLAIMED_STATES.contains(&CreditRegistrationState::Cancelled));
526    }
527
528    /// A registry log line has to name one call, not one row: two polls of a row, and a recovery
529    /// lookup against the row's own resolve call, are separate lines.
530    #[test]
531    fn two_calls_about_one_row_are_addressed_apart() {
532        let id = uuid::Uuid::new_v4();
533        assert_eq!(verify_request_item_id(id, 1), format!("vf-{id}-1"));
534        assert_ne!(verify_request_item_id(id, 1), verify_request_item_id(id, 2));
535        assert_ne!(
536            recovery_request_item_id(id, 1),
537            verify_request_item_id(id, 1)
538        );
539        assert_ne!(recovery_request_item_id(id, 1), format!("cr-{id}"));
540    }
541}
headless_lms_server/domain/credit_registration_phases/verify.rs

headless_lms_server/domain/credit_registration_phases/
verify.rs
