X-Git-Url: http://these/git/?a=blobdiff_plain;f=src%2Fscheduled_tasks.rs;h=28315beaae815c7bfd36a3858611fbbcc5396f98;hb=55e383ae38ecc54fb19948eac23ffefa60c829ea;hp=f20e61e12821d155ed0bdb6e02ffbc1a893ed1dd;hpb=7d8cb93b5323eb50bd90afde270685d1ecd07061;p=lemmy.git diff --git a/src/scheduled_tasks.rs b/src/scheduled_tasks.rs index f20e61e1..28315bea 100644 --- a/src/scheduled_tasks.rs +++ b/src/scheduled_tasks.rs @@ -13,7 +13,16 @@ use diesel::{ use diesel::{sql_query, PgConnection, RunQueryDsl}; use lemmy_api_common::context::LemmyContext; use lemmy_db_schema::{ - schema::{activity, captcha_answer, comment, community_person_ban, instance, person, post}, + schema::{ + captcha_answer, + comment, + community_person_ban, + instance, + person, + post, + received_activity, + sent_activity, + }, source::instance::{Instance, InstanceForm}, utils::{naive_now, DELETED_REPLACEMENT_TEXT}, }; @@ -40,30 +49,54 @@ pub fn setup( // Update active counts every hour let url = db_url.clone(); scheduler.every(CTimeUnits::hour(1)).run(move || { - let mut conn = PgConnection::establish(&url).expect("could not establish connection"); - active_counts(&mut conn); - update_banned_when_expired(&mut conn); + PgConnection::establish(&url) + .map(|mut conn| { + active_counts(&mut conn); + update_banned_when_expired(&mut conn); + }) + .map_err(|e| { + error!("Failed to establish db connection for active counts update: {e}"); + }) + .ok(); }); // Update hot ranks every 15 minutes let url = db_url.clone(); scheduler.every(CTimeUnits::minutes(15)).run(move || { - let mut conn = PgConnection::establish(&url).expect("could not establish connection"); - update_hot_ranks(&mut conn, true); + PgConnection::establish(&url) + .map(|mut conn| { + update_hot_ranks(&mut conn); + }) + .map_err(|e| { + error!("Failed to establish db connection for hot ranks update: {e}"); + }) + .ok(); }); // Delete any captcha answers older than ten minutes, every ten minutes let url = db_url.clone(); scheduler.every(CTimeUnits::minutes(10)).run(move || { - let mut conn = PgConnection::establish(&url).expect("could not establish connection"); - delete_expired_captcha_answers(&mut conn); + PgConnection::establish(&url) + .map(|mut conn| { + delete_expired_captcha_answers(&mut conn); + }) + .map_err(|e| { + error!("Failed to establish db connection for captcha cleanup: {e}"); + }) + .ok(); }); // Clear old activities every week let url = db_url.clone(); scheduler.every(CTimeUnits::weeks(1)).run(move || { - let mut conn = PgConnection::establish(&url).expect("could not establish connection"); - clear_old_activities(&mut conn); + PgConnection::establish(&url) + .map(|mut conn| { + clear_old_activities(&mut conn); + }) + .map_err(|e| { + error!("Failed to establish db connection for activity cleanup: {e}"); + }) + .ok(); }); // Remove old rate limit buckets after 1 to 2 hours of inactivity @@ -75,15 +108,27 @@ pub fn setup( // Overwrite deleted & removed posts and comments every day let url = db_url.clone(); scheduler.every(CTimeUnits::days(1)).run(move || { - let mut conn = PgConnection::establish(&url).expect("could not establish connection"); - overwrite_deleted_posts_and_comments(&mut conn); + PgConnection::establish(&db_url) + .map(|mut conn| { + overwrite_deleted_posts_and_comments(&mut conn); + }) + .map_err(|e| { + error!("Failed to establish db connection for deleted content cleanup: {e}"); + }) + .ok(); }); // Update the Instance Software scheduler.every(CTimeUnits::days(1)).run(move || { - let mut conn = PgConnection::establish(&db_url).expect("could not establish connection"); - update_instance_software(&mut conn, &user_agent) - .map_err(|e| warn!("Failed to update instance software: {e}")) + PgConnection::establish(&url) + .map(|mut conn| { + update_instance_software(&mut conn, &user_agent) + .map_err(|e| warn!("Failed to update instance software: {e}")) + .ok(); + }) + .map_err(|e| { + error!("Failed to establish db connection for instance software update: {e}"); + }) .ok(); }); @@ -98,7 +143,7 @@ pub fn setup( fn startup_jobs(db_url: &str) { let mut conn = PgConnection::establish(db_url).expect("could not establish connection"); active_counts(&mut conn); - update_hot_ranks(&mut conn, false); + update_hot_ranks(&mut conn); update_banned_when_expired(&mut conn); clear_old_activities(&mut conn); overwrite_deleted_posts_and_comments(&mut conn); @@ -106,35 +151,29 @@ fn startup_jobs(db_url: &str) { /// Update the hot_rank columns for the aggregates tables /// Runs in batches until all necessary rows are updated once -fn update_hot_ranks(conn: &mut PgConnection, last_week_only: bool) { - let process_start_time = if last_week_only { - info!("Updating hot ranks for last week..."); - naive_now() - chrono::Duration::days(7) - } else { - info!("Updating hot ranks for all history..."); - NaiveDateTime::from_timestamp_opt(0, 0).expect("0 timestamp creation") - }; +fn update_hot_ranks(conn: &mut PgConnection) { + info!("Updating hot ranks for all history..."); process_hot_ranks_in_batches( conn, "post_aggregates", + "a.hot_rank != 0 OR a.hot_rank_active != 0", "SET hot_rank = hot_rank(a.score, a.published), hot_rank_active = hot_rank(a.score, a.newest_comment_time_necro)", - process_start_time, ); process_hot_ranks_in_batches( conn, "comment_aggregates", + "a.hot_rank != 0", "SET hot_rank = hot_rank(a.score, a.published)", - process_start_time, ); process_hot_ranks_in_batches( conn, "community_aggregates", + "a.hot_rank != 0", "SET hot_rank = hot_rank(a.subscribers, a.published)", - process_start_time, ); info!("Finished hot ranks update!"); @@ -146,18 +185,20 @@ struct HotRanksUpdateResult { published: NaiveDateTime, } -/// Runs the hot rank update query in batches until all rows after `process_start_time` have been -/// processed. -/// In `set_clause`, "a" will refer to the current aggregates table. +/// Runs the hot rank update query in batches until all rows have been processed. +/// In `where_clause` and `set_clause`, "a" will refer to the current aggregates table. /// Locked rows are skipped in order to prevent deadlocks (they will likely get updated on the next /// run) fn process_hot_ranks_in_batches( conn: &mut PgConnection, table_name: &str, + where_clause: &str, set_clause: &str, - process_start_time: NaiveDateTime, ) { + let process_start_time = NaiveDateTime::from_timestamp_opt(0, 0).expect("0 timestamp creation"); + let update_batch_size = 1000; // Bigger batches than this tend to cause seq scans + let mut processed_rows_count = 0; let mut previous_batch_result = Some(process_start_time); while let Some(previous_batch_last_published) = previous_batch_result { // Raw `sql_query` is used as a performance optimization - Diesel does not support doing this @@ -165,7 +206,7 @@ fn process_hot_ranks_in_batches( let result = sql_query(format!( r#"WITH batch AS (SELECT a.id FROM {aggregates_table} a - WHERE a.published > $1 + WHERE a.published > $1 AND ({where_clause}) ORDER BY a.published LIMIT $2 FOR UPDATE SKIP LOCKED) @@ -173,14 +214,18 @@ fn process_hot_ranks_in_batches( FROM batch WHERE a.id = batch.id RETURNING a.published; "#, aggregates_table = table_name, - set_clause = set_clause + set_clause = set_clause, + where_clause = where_clause )) .bind::(previous_batch_last_published) .bind::(update_batch_size) .get_results::(conn); match result { - Ok(updated_rows) => previous_batch_result = updated_rows.last().map(|row| row.published), + Ok(updated_rows) => { + processed_rows_count += updated_rows.len(); + previous_batch_result = updated_rows.last().map(|row| row.published); + } Err(e) => { error!("Failed to update {} hot_ranks: {}", table_name, e); break; @@ -188,45 +233,44 @@ fn process_hot_ranks_in_batches( } } info!( - "Finished process_hot_ranks_in_batches execution for {}", - table_name + "Finished process_hot_ranks_in_batches execution for {} (processed {} rows)", + table_name, processed_rows_count ); } fn delete_expired_captcha_answers(conn: &mut PgConnection) { - match diesel::delete( + diesel::delete( captcha_answer::table.filter(captcha_answer::published.lt(now - IntervalDsl::minutes(10))), ) .execute(conn) - { - Ok(_) => { - info!("Done."); - } - Err(e) => { - error!("Failed to clear old captcha answers: {}", e) - } - } + .map(|_| { + info!("Done."); + }) + .map_err(|e| error!("Failed to clear old captcha answers: {e}")) + .ok(); } /// Clear old activities (this table gets very large) fn clear_old_activities(conn: &mut PgConnection) { info!("Clearing old activities..."); - match diesel::delete(activity::table.filter(activity::published.lt(now - 6.months()))) + diesel::delete(sent_activity::table.filter(sent_activity::published.lt(now - 3.months()))) .execute(conn) - { - Ok(_) => { - info!("Done."); - } - Err(e) => { - error!("Failed to clear old activities: {}", e) - } - } + .map_err(|e| error!("Failed to clear old sent activities: {e}")) + .ok(); + + diesel::delete( + received_activity::table.filter(received_activity::published.lt(now - 3.months())), + ) + .execute(conn) + .map(|_| info!("Done.")) + .map_err(|e| error!("Failed to clear old received activities: {e}")) + .ok(); } /// overwrite posts and comments 30d after deletion fn overwrite_deleted_posts_and_comments(conn: &mut PgConnection) { info!("Overwriting deleted posts..."); - match diesel::update( + diesel::update( post::table .filter(post::deleted.eq(true)) .filter(post::updated.lt(now.nullable() - 1.months())) @@ -237,17 +281,14 @@ fn overwrite_deleted_posts_and_comments(conn: &mut PgConnection) { post::name.eq(DELETED_REPLACEMENT_TEXT), )) .execute(conn) - { - Ok(_) => { - info!("Done."); - } - Err(e) => { - error!("Failed to overwrite deleted posts: {}", e) - } - } + .map(|_| { + info!("Done."); + }) + .map_err(|e| error!("Failed to overwrite deleted posts: {e}")) + .ok(); info!("Overwriting deleted comments..."); - match diesel::update( + diesel::update( comment::table .filter(comment::deleted.eq(true)) .filter(comment::updated.lt(now.nullable() - 1.months())) @@ -255,14 +296,11 @@ fn overwrite_deleted_posts_and_comments(conn: &mut PgConnection) { ) .set(comment::content.eq(DELETED_REPLACEMENT_TEXT)) .execute(conn) - { - Ok(_) => { - info!("Done."); - } - Err(e) => { - error!("Failed to overwrite deleted comments: {}", e) - } - } + .map(|_| { + info!("Done."); + }) + .map_err(|e| error!("Failed to overwrite deleted comments: {e}")) + .ok(); } /// Re-calculate the site and community active counts every 12 hours @@ -281,20 +319,16 @@ fn active_counts(conn: &mut PgConnection) { "update site_aggregates set users_active_{} = (select * from site_aggregates_activity('{}')) where site_id = 1", i.1, i.0 ); - match sql_query(update_site_stmt).execute(conn) { - Ok(_) => {} - Err(e) => { - error!("Failed to update site stats: {}", e) - } - } + sql_query(update_site_stmt) + .execute(conn) + .map_err(|e| error!("Failed to update site stats: {e}")) + .ok(); let update_community_stmt = format!("update community_aggregates ca set users_active_{} = mv.count_ from community_aggregates_activity('{}') mv where ca.community_id = mv.community_id_", i.1, i.0); - match sql_query(update_community_stmt).execute(conn) { - Ok(_) => {} - Err(e) => { - error!("Failed to update community stats: {}", e) - } - } + sql_query(update_community_stmt) + .execute(conn) + .map_err(|e| error!("Failed to update community stats: {e}")) + .ok(); } info!("Done."); @@ -304,27 +338,20 @@ fn active_counts(conn: &mut PgConnection) { fn update_banned_when_expired(conn: &mut PgConnection) { info!("Updating banned column if it expires ..."); - match diesel::update( + diesel::update( person::table .filter(person::banned.eq(true)) .filter(person::ban_expires.lt(now)), ) .set(person::banned.eq(false)) .execute(conn) - { - Ok(_) => {} - Err(e) => { - error!("Failed to update person.banned when expires: {}", e) - } - } - match diesel::delete(community_person_ban::table.filter(community_person_ban::expires.lt(now))) + .map_err(|e| error!("Failed to update person.banned when expires: {e}")) + .ok(); + + diesel::delete(community_person_ban::table.filter(community_person_ban::expires.lt(now))) .execute(conn) - { - Ok(_) => {} - Err(e) => { - error!("Failed to remove community_ban expired rows: {}", e) - } - } + .map_err(|e| error!("Failed to remove community_ban expired rows: {e}")) + .ok(); } /// Updates the instance software and version @@ -360,12 +387,13 @@ fn update_instance_software(conn: &mut PgConnection, user_agent: &str) -> LemmyR Ok(res) => match res.json::() { Ok(node_info) => { // Instance sent valid nodeinfo, write it to db + let software = node_info.software.as_ref(); Some( InstanceForm::builder() .domain(instance.domain) .updated(Some(naive_now())) - .software(node_info.software.and_then(|s| s.name)) - .version(node_info.version.clone()) + .software(software.and_then(|s| s.name.clone())) + .version(software.and_then(|s| s.version.clone())) .build(), ) } @@ -391,6 +419,9 @@ fn update_instance_software(conn: &mut PgConnection, user_agent: &str) -> LemmyR #[cfg(test)] mod tests { + #![allow(clippy::unwrap_used)] + #![allow(clippy::indexing_slicing)] + use lemmy_routes::nodeinfo::NodeInfo; use reqwest::Client;