headless_lms_server/programs/
regrader.rs

1use std::{env, error::Error, sync::Arc, time::Duration};
2
3use crate::config::FileStoreRuntimeConfig;
4use crate::config::program_config::ProgramConfig;
5use crate::domain::models_requests::{self, JwtKey};
6use crate::programs::periodic_worker::{
7    PeriodicWorkerConfig, is_db_disconnect, run_periodic_worker,
8};
9use headless_lms_base::config::ApplicationConfiguration;
10use headless_lms_models as models;
11use models::library::regrading;
12use sqlx::PgPool;
13
14/**
15Starts a thread that will periodically send regrading submissions to the corresponding exercise services for regrading.
16*/
17pub async fn main() -> anyhow::Result<()> {
18    // TODO: Audit that the environment access only happens in single-threaded code.
19    unsafe { env::set_var("RUST_LOG", "info,actix_web=info,sqlx=warn") };
20    dotenvy::dotenv().ok();
21    crate::setup_tracing()?;
22    let db_url = ProgramConfig::database_url_with_default();
23    let jwt_password = secrecy::SecretString::new(ProgramConfig::required("JWT_PASSWORD")?.into());
24    let jwt_key = Arc::new(JwtKey::new(&jwt_password)?);
25    let app_conf = ApplicationConfiguration::try_from_env()?;
26    let file_store =
27        crate::setup_file_store(&FileStoreRuntimeConfig::try_from_env()?, &app_conf.base_url).await;
28
29    // Since this is repeating every 10 seconds we can keep the connection open.
30    let db_pool = PgPool::connect(&db_url).await?;
31    let mut conn = db_pool.acquire().await?;
32
33    run_periodic_worker(
34        PeriodicWorkerConfig {
35            tick_interval: Duration::from_secs(10),
36            still_running_every: 60,
37            still_running_message: "running the regrader",
38            initial_ticks: 60,
39            delay_missed_ticks: false,
40        },
41        async || {
42            let exercise_services_by_type =
43                models::exercise_service_info::get_upsert_all_exercise_services_by_type(
44                    &mut conn,
45                    models_requests::fetch_service_info,
46                )
47                .await?;
48            // do not stop the thread on error, report it and try again next tick
49            if let Err(err) = regrading::regrade(
50                &mut conn,
51                &exercise_services_by_type,
52                models_requests::make_grading_request_sender(
53                    Arc::clone(&jwt_key),
54                    app_conf.base_url.clone(),
55                ),
56                file_store.as_ref(),
57                &app_conf,
58            )
59            .await
60            {
61                tracing::error!("Error in regrader: {}", err);
62                if is_db_disconnect(err.source()) {
63                    // this usually happens if the database is reset while running bin/dev etc.
64                    tracing::info!(
65                        "regrader may have lost its connection to the db, trying to reconnect"
66                    );
67                    conn = db_pool.acquire().await?;
68                }
69            }
70            Ok(())
71        },
72    )
73    .await
74}
headless_lms_server/programs/regrader.rs

headless_lms_server/programs/
regrader.rs
