X-Git-Url: http://these/git/?a=blobdiff_plain;f=src%2Fscheduled_tasks.rs;h=4928fe8dd5c0f836757f2bb1ce92431794f5fd6f;hb=92568956353f21649ed9aff68b42699c9d036f30;hp=4d3c936e88b4cd49eac90c4473645be350c4ac9d;hpb=2aef6a5a338c9a6b5764fbc1ae42fa3edf096deb;p=lemmy.git diff --git a/src/scheduled_tasks.rs b/src/scheduled_tasks.rs index 4d3c936e..4928fe8d 100644 --- a/src/scheduled_tasks.rs +++ b/src/scheduled_tasks.rs @@ -13,15 +13,27 @@ 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}, }; 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,7 +58,7 @@ pub fn setup( 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); + update_hot_ranks(&mut conn); }); // Delete any captcha answers older than ten minutes, every ten minutes @@ -79,7 +91,9 @@ pub fn setup( // 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 @@ -93,7 +107,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); @@ -101,35 +115,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!"); @@ -141,18 +149,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 @@ -160,7 +170,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) @@ -168,14 +178,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; @@ -183,8 +197,8 @@ 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 ); } @@ -206,16 +220,17 @@ fn delete_expired_captcha_answers(conn: &mut PgConnection) { /// 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_err(|e| error!("Failed to clear old received activities: {}", e)) + .ok(); } /// overwrite posts and comments 30d after deletion @@ -273,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) { @@ -323,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;