cleanup old errors

.env.example

@@ -45,4 +45,8 @@ ADMIN_DOMAINS="example.com,foo.net"
 # token name used for logging in - has to be a valid DNS TXT key
 LOGIN_TOKEN_NAME="localhost"
 # completely disable any health checks - debugging only
-DISABLE_HEALTH_CHECKS=false
+DISABLE_HEALTH_CHECKS=false
+# interval in seconds to cleanup old data like errors stored
+CLEANUP_INTERVAL_S=86400
+# amount of errors to keep per host
+ERROR_RETENTION_PER_HOST=100

Cargo.lock

@@ -3201,6 +3201,7 @@ dependencies = [
  "reqwest",
  "scraper",
  "sea-orm",
+ "sea-query",
  "serde",
  "thiserror",
  "tokio",

Cargo.toml

@@ -15,6 +15,7 @@ chrono = "0.4.22"
 serde = "1.0.144"
 sea-orm-migration = "0.12.0"
 sea-orm = "0.12.0"
+sea-query = "0.30" # keep in step sea-orm
 futures = "0.3"
 tracing = "0.1"
 tracing-subscriber = "0.3"

entities/src/state/scanner.rs

@@ -37,6 +37,10 @@ pub struct Config {
     pub source_git_branch: String,
     /// Hosts known to be bad (ip block), domain per entry
     pub bad_hosts: Vec<String>,
+    /// Interval to run cleanup operations in, to remove old data
+    pub cleanup_interval: Duration,
+    /// Amount of latest errors to keep per instance/host
+    pub error_retention_per_host: usize,
 }
 
 impl Config {
@@ -59,6 +63,8 @@ impl Config {
             source_git_branch: String::from("master"),
             source_git_url: String::from("https://github.com/zedeus/nitter.git"),
             bad_hosts: vec![String::from("tweet.whateveritworks.org")],
+            cleanup_interval: Duration::from_secs(24*60*60),
+            error_retention_per_host: 100,
         })
     }
 }

scanner/Cargo.toml

@@ -8,6 +8,7 @@ edition = "2021"
 [dependencies]
 reqwest = { workspace = true, features = ["deflate","gzip","brotli","cookies", "rustls-tls"] }
 sea-orm = { workspace = true, features = [ "sqlx-sqlite", "runtime-tokio-native-tls", "macros" ] }
+sea-query = { workspace = true }
 thiserror = { workspace = true }
 tracing = { workspace = true }
 tokio = { workspace = true, features = ["full"] }

scanner/src/cleanup.rs

@@ -0,0 +1,59 @@
+use entities::check_errors;
+use entities::health_check;
+use entities::host;
+use sea_orm::ColumnTrait;
+use sea_orm::EntityTrait;
+use sea_orm::Order;
+use sea_orm::QueryFilter;
+use sea_orm::QueryOrder;
+use sea_orm::QuerySelect;
+use sea_query::Query;
+use tokio::time::interval;
+
+use crate::Scanner;
+use crate::Result;
+
+impl Scanner {
+    /// Setup scheduled job for cleaning up old data
+    pub(crate) fn schedule_cleanup(&self) -> Result<()> {
+        let c = self.clone();
+        tokio::spawn(async move {
+            let mut interval = interval(c.inner.config.cleanup_interval);
+            loop {
+                interval.tick().await;
+                if let Err(e) = c.cleanup().await {
+                    tracing::error!(error=%e);
+                }
+            }
+        });
+        Ok(())
+    }
+    /// Perform cleanup of outdated data
+    async fn cleanup(&self) -> Result<()> {
+        self.cleanup_errors().await?;
+        Ok(())
+    }
+
+    /// Remove all but recent host error entries
+    async fn cleanup_errors(&self) -> Result<()> {
+        // I wish this was easier without row ids or giant SQL queries
+        let hosts = host::Entity::find().all(&self.inner.db).await?;
+        for host in hosts {
+            let res = check_errors::Entity::delete_many()
+            .filter(check_errors::Column::Host.eq(host.id))
+            .filter(check_errors::Column::Time.not_in_subquery(
+                Query::select()
+                .column(check_errors::Column::Time)
+                .from(check_errors::Entity)
+                .and_where(check_errors::Column::Host.eq(host.id))
+                .order_by(check_errors::Column::Time, Order::Desc)
+                .limit(self.inner.config.error_retention_per_host as _)
+                .to_owned()
+            ))
+            .exec(&self.inner.db).await?;
+            tracing::debug!(host=host.id,deleted_errors=res.rows_affected);
+        }
+
+        Ok(())
+    }
+}

scanner/src/lib.rs

@@ -36,6 +36,7 @@ mod instance_check;
 mod instance_parser;
 mod profile_parser;
 mod version_check;
+mod cleanup;
 
 const CAPTCHA_TEXT: &'static str = "Enable JavaScript and cookies to continue";
 const CAPTCHA_CODE: u16 = 403;
@@ -116,6 +117,7 @@ pub async fn run_scanner(
     disable_health_checks: bool,
 ) -> Result<()> {
     let scanner = Scanner::new(db, config, app_state).await?;
+    scanner.schedule_cleanup().unwrap();
 
     if disable_health_checks {
         tracing::error!("Health checks disabled!");
@@ -284,7 +286,7 @@ impl Scanner {
     #[instrument]
     async fn update_instacelist(&mut self) -> Result<()> {
         let start = Instant::now();
-        let html: String = self.fetch_instancelist().await?;
+        let html: String = self.fetch_instance_list().await?;
         let parsed_instances = self.inner.instance_parser.parse_instancelist(
             &html,
             &self.inner.config.additional_hosts,
@@ -403,7 +405,7 @@ impl Scanner {
         Ok(())
     }
 
-    async fn fetch_instancelist(&self) -> Result<String> {
+    async fn fetch_instance_list(&self) -> Result<String> {
         let (_, body) = self.fetch_url(&self.inner.config.instance_list_url).await?;
         Ok(body)
     }
@@ -554,7 +556,7 @@ mod test {
         let scanner = Scanner::new(db, Config::test_defaults(), entities::state::new())
             .await
             .unwrap();
-        let res = scanner.fetch_instancelist().await.unwrap();
+        let res = scanner.fetch_instance_list().await.unwrap();
         let mut file = File::create("test_data/instancelist.html").await.unwrap();
         file.write_all(&res.as_bytes()).await.unwrap();
     }

src/main.rs

@@ -130,6 +130,12 @@ fn read_scanner_cfg() -> miette::Result<ScannerConfig> {
     let auto_mute = require_env_str("AUTO_MUTE")? == "true";
     let source_git_branch = require_env_str("ORIGIN_SOURCE_GIT_BRANCH")?;
     let source_git_url = require_env_str("ORIGIN_SOURCE_GIT_URL")?;
+    let cleanup_interval: u64 = require_env_str("CLEANUP_INTERVAL_S")?
+        .parse()
+        .expect("CLEANUP_INTERVAL_S must be a number");
+    let error_retention_per_host: usize = require_env_str("ERROR_RETENTION_PER_HOST")?
+        .parse()
+        .expect("CLEANUP_INTERVAL_S must be a number");
 
     Ok(Arc::new(entities::state::scanner::Config {
         list_fetch_interval: Duration::from_secs(instance_list_interval),
@@ -149,6 +155,8 @@ fn read_scanner_cfg() -> miette::Result<ScannerConfig> {
         source_git_branch,
         source_git_url,
         bad_hosts,
+        cleanup_interval: Duration::from_secs(cleanup_interval),
+        error_retention_per_host,
     }))
 }