X-Git-Url: http://these/git/?a=blobdiff_plain;f=src%2Fscheduled_tasks.rs;h=4928fe8dd5c0f836757f2bb1ce92431794f5fd6f;hb=92568956353f21649ed9aff68b42699c9d036f30;hp=9fb1ba70229f4d537e1867ce14ab80668448ee74;hpb=45818fb4c50dc06f6909312bd034f68af92c3146;p=lemmy.git diff --git a/src/scheduled_tasks.rs b/src/scheduled_tasks.rs index 9fb1ba70..4928fe8d 100644 --- a/src/scheduled_tasks.rs +++ b/src/scheduled_tasks.rs @@ -1,31 +1,39 @@ +use chrono::NaiveDateTime; use clokwerk::{Scheduler, TimeUnits as CTimeUnits}; use diesel::{ dsl::{now, IntervalDsl}, + sql_types::{Integer, Timestamp}, Connection, ExpressionMethods, + NullableExpressionMethods, QueryDsl, + QueryableByName, }; // Import week days and WeekDay use diesel::{sql_query, PgConnection, RunQueryDsl}; use lemmy_api_common::context::LemmyContext; use lemmy_db_schema::{ schema::{ - activity, - comment_aggregates, - community_aggregates, + captcha_answer, + comment, community_person_ban, instance, person, - post_aggregates, + post, + received_activity, + sent_activity, }, source::instance::{Instance, InstanceForm}, - utils::{functions::hot_rank, naive_now}, + utils::{naive_now, DELETED_REPLACEMENT_TEXT}, }; use lemmy_routes::nodeinfo::NodeInfo; -use lemmy_utils::{error::LemmyError, REQWEST_TIMEOUT}; +use lemmy_utils::{ + error::{LemmyError, LemmyResult}, + REQWEST_TIMEOUT, +}; use reqwest::blocking::Client; use std::{thread, time::Duration}; -use tracing::{error, info}; +use tracing::{error, info, warn}; /// Schedules various cleanup tasks for lemmy in a background thread pub fn setup( @@ -46,11 +54,18 @@ pub fn setup( update_banned_when_expired(&mut conn); }); - // Update hot ranks every 5 minutes + // 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); + }); + + // Delete any captcha answers older than ten minutes, every ten minutes let url = db_url.clone(); - scheduler.every(CTimeUnits::minutes(5)).run(move || { + scheduler.every(CTimeUnits::minutes(10)).run(move || { let mut conn = PgConnection::establish(&url).expect("could not establish connection"); - update_hot_ranks(&mut conn, true); + delete_expired_captcha_answers(&mut conn); }); // Clear old activities every week @@ -66,10 +81,19 @@ pub fn setup( context_1.settings_updated_channel().remove_older_than(hour); }); + // 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); + }); + // 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); + update_instance_software(&mut conn, &user_agent) + .map_err(|e| warn!("Failed to update instance software: {e}")) + .ok(); }); // Manually run the scheduler in an event loop @@ -83,85 +107,170 @@ 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); } /// Update the hot_rank columns for the aggregates tables -fn update_hot_ranks(conn: &mut PgConnection, last_week_only: bool) { - let mut post_update = diesel::update(post_aggregates::table).into_boxed(); - let mut comment_update = diesel::update(comment_aggregates::table).into_boxed(); - let mut community_update = diesel::update(community_aggregates::table).into_boxed(); - - // Only update for the last week of content - if last_week_only { - info!("Updating hot ranks for last week..."); - let last_week = now - diesel::dsl::IntervalDsl::weeks(1); - - post_update = post_update.filter(post_aggregates::published.gt(last_week)); - comment_update = comment_update.filter(comment_aggregates::published.gt(last_week)); - community_update = community_update.filter(community_aggregates::published.gt(last_week)); - } else { - info!("Updating hot ranks for all history..."); - } +/// Runs in batches until all necessary rows are updated once +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_hot_ranks_in_batches( + conn, + "comment_aggregates", + "a.hot_rank != 0", + "SET hot_rank = hot_rank(a.score, a.published)", + ); + + process_hot_ranks_in_batches( + conn, + "community_aggregates", + "a.hot_rank != 0", + "SET hot_rank = hot_rank(a.subscribers, a.published)", + ); + + info!("Finished hot ranks update!"); +} - match post_update - .set(( - post_aggregates::hot_rank.eq(hot_rank(post_aggregates::score, post_aggregates::published)), - post_aggregates::hot_rank_active.eq(hot_rank( - post_aggregates::score, - post_aggregates::newest_comment_time_necro, - )), +#[derive(QueryableByName)] +struct HotRanksUpdateResult { + #[diesel(sql_type = Timestamp)] + published: NaiveDateTime, +} + +/// 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, +) { + 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 + // in a single query (neither as a CTE, nor using a subquery) + let result = sql_query(format!( + r#"WITH batch AS (SELECT a.id + FROM {aggregates_table} a + WHERE a.published > $1 AND ({where_clause}) + ORDER BY a.published + LIMIT $2 + FOR UPDATE SKIP LOCKED) + UPDATE {aggregates_table} a {set_clause} + FROM batch WHERE a.id = batch.id RETURNING a.published; + "#, + aggregates_table = table_name, + set_clause = set_clause, + where_clause = where_clause )) - .execute(conn) - { - Ok(_) => {} - Err(e) => { - error!("Failed to update post_aggregates hot_ranks: {}", e) + .bind::(previous_batch_last_published) + .bind::(update_batch_size) + .get_results::(conn); + + match result { + 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; + } } } + info!( + "Finished process_hot_ranks_in_batches execution for {} (processed {} rows)", + table_name, processed_rows_count + ); +} - match comment_update - .set(comment_aggregates::hot_rank.eq(hot_rank( - comment_aggregates::score, - comment_aggregates::published, - ))) - .execute(conn) +fn delete_expired_captcha_answers(conn: &mut PgConnection) { + match diesel::delete( + captcha_answer::table.filter(captcha_answer::published.lt(now - IntervalDsl::minutes(10))), + ) + .execute(conn) { - Ok(_) => {} + Ok(_) => { + info!("Done."); + } Err(e) => { - error!("Failed to update comment_aggregates hot_ranks: {}", e) + error!("Failed to clear old captcha answers: {}", e) } } +} - match community_update - .set(community_aggregates::hot_rank.eq(hot_rank( - community_aggregates::subscribers, - community_aggregates::published, - ))) +/// Clear old activities (this table gets very large) +fn clear_old_activities(conn: &mut PgConnection) { + info!("Clearing old activities..."); + diesel::delete(sent_activity::table.filter(sent_activity::published.lt(now - 3.months()))) .execute(conn) + .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_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( + post::table + .filter(post::deleted.eq(true)) + .filter(post::updated.lt(now.nullable() - 1.months())) + .filter(post::body.ne(DELETED_REPLACEMENT_TEXT)), + ) + .set(( + post::body.eq(DELETED_REPLACEMENT_TEXT), + post::name.eq(DELETED_REPLACEMENT_TEXT), + )) + .execute(conn) { Ok(_) => { info!("Done."); } Err(e) => { - error!("Failed to update community_aggregates hot_ranks: {}", e) + error!("Failed to overwrite deleted posts: {}", e) } } -} -/// 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()))) - .execute(conn) + info!("Overwriting deleted comments..."); + match diesel::update( + comment::table + .filter(comment::deleted.eq(true)) + .filter(comment::updated.lt(now.nullable() - 1.months())) + .filter(comment::content.ne(DELETED_REPLACEMENT_TEXT)), + ) + .set(comment::content.eq(DELETED_REPLACEMENT_TEXT)) + .execute(conn) { Ok(_) => { info!("Done."); } Err(e) => { - error!("Failed to clear old activities: {}", e) + error!("Failed to overwrite deleted comments: {}", e) } } } @@ -179,7 +288,7 @@ fn active_counts(conn: &mut PgConnection) { for i in &intervals { let update_site_stmt = format!( - "update site_aggregates set users_active_{} = (select * from site_aggregates_activity('{}'))", + "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) { @@ -229,66 +338,72 @@ fn update_banned_when_expired(conn: &mut PgConnection) { } /// Updates the instance software and version -fn update_instance_software(conn: &mut PgConnection, user_agent: &str) { +/// +/// TODO: this should be async +/// TODO: if instance has been dead for a long time, it should be checked less frequently +fn update_instance_software(conn: &mut PgConnection, user_agent: &str) -> LemmyResult<()> { info!("Updating instances software and versions..."); - let client = match Client::builder() + let client = Client::builder() .user_agent(user_agent) .timeout(REQWEST_TIMEOUT) - .build() - { - Ok(client) => client, - Err(e) => { - error!("Failed to build reqwest client: {}", e); - return; - } - }; + .build()?; - let instances = match instance::table.get_results::(conn) { - Ok(instances) => instances, - Err(e) => { - error!("Failed to get instances: {}", e); - return; - } - }; + let instances = instance::table.get_results::(conn)?; for instance in instances { let node_info_url = format!("https://{}/nodeinfo/2.0.json", instance.domain); - // Skip it if it can't connect - let res = client - .get(&node_info_url) - .send() - .ok() - .and_then(|t| t.json::().ok()); - - if let Some(node_info) = res { - let software = node_info.software.as_ref(); - let form = InstanceForm::builder() - .domain(instance.domain) - .software(software.and_then(|s| s.name.clone())) - .version(software.and_then(|s| s.version.clone())) - .updated(Some(naive_now())) - .build(); - - match diesel::update(instance::table.find(instance.id)) - .set(form) - .execute(conn) - { - Ok(_) => { - info!("Done."); + // The `updated` column is used to check if instances are alive. If it is more than three days + // in the past, no outgoing activities will be sent to that instance. However not every + // Fediverse instance has a valid Nodeinfo endpoint (its not required for Activitypub). That's + // why we always need to mark instances as updated if they are alive. + let default_form = InstanceForm::builder() + .domain(instance.domain.clone()) + .updated(Some(naive_now())) + .build(); + let form = match client.get(&node_info_url).send() { + Ok(res) if res.status().is_client_error() => { + // Instance doesnt have nodeinfo but sent a response, consider it alive + Some(default_form) + } + Ok(res) => match res.json::() { + Ok(node_info) => { + // Instance sent valid nodeinfo, write it to db + Some( + InstanceForm::builder() + .domain(instance.domain) + .updated(Some(naive_now())) + .software(node_info.software.and_then(|s| s.name)) + .version(node_info.version.clone()) + .build(), + ) } - Err(e) => { - error!("Failed to update site instance software: {}", e); - return; + Err(_) => { + // No valid nodeinfo but valid HTTP response, consider instance alive + Some(default_form) } + }, + Err(_) => { + // dead instance, do nothing + None } + }; + if let Some(form) = form { + diesel::update(instance::table.find(instance.id)) + .set(form) + .execute(conn)?; } } + info!("Finished updating instances software and versions..."); + Ok(()) } #[cfg(test)] mod tests { + #![allow(clippy::unwrap_used)] + #![allow(clippy::indexing_slicing)] + use lemmy_routes::nodeinfo::NodeInfo; use reqwest::Client;