headless_lms_credit_registration/use_cases/verify/
lease.rs

1//! Leasing the rows a verify iteration asks about, so no concurrent iteration asks about them too.
2
3use std::collections::HashMap;
4
5use chrono::{DateTime, Utc};
6use headless_lms_models::credit_registrations::{
7    VerifyFlow, claim_due_for_verify, increment_verify_attempt_counts, schedule_next_attempts,
8};
9use headless_lms_models::library::credit_registration::course_unit_poll_slots::next_course_unit_polls;
10use headless_lms_models::library::credit_registration::outcomes::{
11    RowFacts, verify_poll_lease_until,
12};
13use sqlx::PgConnection;
14use uuid::Uuid;
15
16use crate::error::CreditRegistrationResult;
17use crate::use_cases::batch_flow::BatchFlowContext;
18use crate::workflow::{Claimed, ClaimedRegistration};
19
20/// A claimed row, leased where it stands, with the verify attempt the lease counted for it.
21pub(super) type Leased = Claimed<VerifyAttempt>;
22
23/// What the lease wrote for its row: the attempt count that sets an uncertain row's backoff, and
24/// the next course unit poll. Only [`claim_and_lease`] makes one.
25pub(super) struct VerifyAttempt {
26    count: i32,
27    course_unit_poll_at: Option<DateTime<Utc>>,
28}
29
30/// The row's facts under the attempt the lease counted, not the count the row was claimed with, so
31/// the backoff advances once per attempt.
32pub(super) fn attempt_facts(leased: &Leased, now: DateTime<Utc>) -> RowFacts {
33    RowFacts {
34        verify_attempt_count: leased.extra.count,
35        course_unit_poll_at: leased.extra.course_unit_poll_at,
36        ..leased.claim.facts(now)
37    }
38}
39
40/// Claims up to `limit` of `flow`'s due rows, counts an attempt on each and leases it until its
41/// next poll would be due, or the iteration's registry calls run out, so a concurrent iteration cannot
42/// poll the same row. Each answer overwrites its own row's schedule.
43pub(super) async fn claim_and_lease(
44    ctx: &BatchFlowContext<'_>,
45    conn: &mut PgConnection,
46    flow: VerifyFlow,
47    limit: usize,
48) -> CreditRegistrationResult<Vec<Leased>> {
49    let claimed = claim_due_for_verify(
50        conn,
51        flow,
52        ctx.scope,
53        i64::try_from(limit).unwrap_or(i64::MAX),
54    )
55    .await?;
56    let attempts = increment_verify_attempt_counts(
57        conn,
58        &claimed.iter().map(|row| row.id).collect::<Vec<_>>(),
59    )
60    .await?;
61    let now = Utc::now();
62    let course_unit_polls: HashMap<Uuid, DateTime<Utc>> =
63        if flow == VerifyFlow::Poll && !claimed.is_empty() {
64            let first_seen_at: Vec<_> = claimed
65                .iter()
66                .map(|row| row.partially_registered_at.unwrap_or(now))
67                .collect();
68            claimed
69                .iter()
70                .map(|row| row.id)
71                .zip(next_course_unit_polls(conn, now, &first_seen_at).await?)
72                .collect()
73        } else {
74            HashMap::new()
75        };
76    let leased: Vec<Leased> = claimed
77        .into_iter()
78        .filter_map(|row| {
79            // Not expected while the claim holds the row's lock; if it happens anyway, the row is
80            // left unpolled rather than polled under an attempt the lease never wrote.
81            let Some(attempt) = attempts.get(&row.id).copied() else {
82                warn!(
83                    credit_registration_id = %row.id,
84                    "Verify attempt was not counted for a claimed credit registration; leaving it unpolled"
85                );
86                return None;
87            };
88            let course_unit_poll_at = course_unit_polls.get(&row.id).copied();
89            Some(Claimed {
90                claim: ClaimedRegistration::left_in_place(row),
91                extra: VerifyAttempt {
92                    count: attempt,
93                    course_unit_poll_at,
94                },
95            })
96        })
97        .collect();
98    let scheduled: Vec<_> = leased
99        .iter()
100        .map(|leased| {
101            let row = leased.claim.registration();
102            (
103                row.id,
104                verify_poll_lease_until(
105                    row.state,
106                    &attempt_facts(leased, now),
107                    ctx.study_registry_wait,
108                ),
109            )
110        })
111        .collect();
112    schedule_next_attempts(conn, &scheduled).await?;
113    Ok(leased)
114}
headless_lms_credit_registration/use_cases/verify/lease.rs

headless_lms_credit_registration/use_cases/verify/
lease.rs
