rate limit instance update, better startup errors, reorganize list fetching, report connectivity check in about

Cargo.lock

@@ -3218,6 +3218,7 @@ dependencies = [
  "csv",
  "entities",
  "git2",
+ "miette",
  "migration",
  "regex",
  "reqwest",

entities/src/state/scanner.rs

@@ -41,6 +41,8 @@ pub struct Config {
     pub cleanup_interval: Duration,
     /// Amount of latest errors to keep per instance/host
     pub error_retention_per_host: usize,
+    /// Path for connectivity checks
+    pub connectivity_path: String,
 }
 
 impl Config {
@@ -65,6 +67,7 @@ impl Config {
             bad_hosts: vec![String::from("tweet.whateveritworks.org")],
             cleanup_interval: Duration::from_secs(24*60*60),
             error_retention_per_host: 100,
+            connectivity_path: String::from("/"),
         })
     }
 }

scanner/Cargo.toml

@@ -10,6 +10,7 @@ reqwest = { workspace = true, features = ["deflate","gzip","brotli","cookies", "
 sea-orm = { workspace = true, features = [ "sqlx-sqlite", "runtime-tokio-native-tls", "macros" ] }
 sea-query = { workspace = true }
 thiserror = { workspace = true }
+miette = { workspace = true }
 tracing = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
 scraper = "0.17.1"

scanner/src/cache_update.rs

@@ -1,5 +1,5 @@
-use std::cmp;
 // SPDX-License-Identifier: AGPL-3.0-only
+use std::cmp;
 use std::collections::HashMap;
 
 use chrono::{TimeZone, Duration};

scanner/src/lib.rs

@@ -2,31 +2,26 @@
 use std::{
     net::IpAddr,
     sync::{Arc, Mutex},
-    time::Instant,
 };
 
 use about_parser::AboutParser;
 use chrono::{Utc, DateTime, TimeZone, Duration};
-use entities::{
-    host,
-    prelude::*,
-    state::{error_cache::HostError, scanner::ScannerConfig, AppState}, health_check,
-};
+use entities::{state::{error_cache::HostError, scanner::ScannerConfig, AppState}, health_check};
 use instance_parser::InstanceParser;
+use miette::{IntoDiagnostic, Context};
 use profile_parser::ProfileParser;
 use regex::{Regex, RegexBuilder};
 use reqwest::{
     header::{HeaderMap, HeaderValue},
-    Client, ClientBuilder, Url,
+    Client, ClientBuilder
 };
 use sea_orm::{
-    sea_query::OnConflict, ActiveModelTrait, ActiveValue, ColumnTrait, ConnectionTrait,
-    DatabaseConnection, DbBackend, EntityTrait, FromQueryResult, QueryFilter, Statement,
-    TransactionTrait, QuerySelect,
+    ColumnTrait, ConnectionTrait,
+    DatabaseConnection, DbBackend, EntityTrait, FromQueryResult, Statement,
+    QuerySelect,
 };
 use thiserror::Error;
-use tokio::{task::JoinSet, time::sleep};
-use tracing::instrument;
+use tokio::time::sleep;
 
 type Result<T> = std::result::Result<T, ScannerError>;
 
@@ -37,6 +32,7 @@ mod instance_parser;
 mod profile_parser;
 mod version_check;
 mod cleanup;
+mod list_update;
 
 const CAPTCHA_TEXT: &'static str = "Enable JavaScript and cookies to continue";
 const CAPTCHA_CODE: u16 = 403;
@@ -110,13 +106,21 @@ impl FetchError {
     }
 }
 
+#[derive(Debug, FromQueryResult, Default)]
+pub(crate) struct LatestCheck {
+    pub host: i32,
+    pub healthy: bool,
+    pub domain: String,
+}
+
 pub async fn run_scanner(
     db: DatabaseConnection,
     config: ScannerConfig,
     app_state: AppState,
     disable_health_checks: bool,
-) -> Result<()> {
-    let scanner = Scanner::new(db, config, app_state).await?;
+) -> miette::Result<()> {
+    let scanner = Scanner::new(db, config, app_state).await
+        .wrap_err("Initializing scanner!")?;
     scanner.schedule_cleanup().unwrap();
 
     if disable_health_checks {
@@ -124,7 +128,7 @@ pub async fn run_scanner(
         return Ok(())
     }
     tokio::spawn(async move {
-        scanner.run().await.unwrap();
+        scanner.run().await.expect("Failed to run scanner daemon!");
     });
 
     Ok(())
@@ -150,13 +154,6 @@ struct InnerScanner {
     client_ipv6: Client,
 }
 
-#[derive(Debug, FromQueryResult, Default)]
-pub struct LatestCheck {
-    pub host: i32,
-    pub healthy: bool,
-    pub domain: String,
-}
-
 impl Scanner {
     pub(crate) fn client_builder(config: &ScannerConfig) -> ClientBuilder {
         let mut headers = HeaderMap::with_capacity(HEADERS.len());
@@ -181,20 +178,22 @@ impl Scanner {
         db: DatabaseConnection,
         config: ScannerConfig,
         app_state: AppState,
-    ) -> Result<Self> {
+    ) -> miette::Result<Self> {
         let mut builder_regex_rss = RegexBuilder::new(&config.rss_content);
         builder_regex_rss.case_insensitive(true);
         let http_client = Self::client_builder(&config);
         let client_ipv4 = Scanner::client_builder(&config)
             .local_address("0.0.0.0".parse::<IpAddr>().unwrap())
             .build()
-            .unwrap();
+            .into_diagnostic()?;
         let client_ipv6 = Scanner::client_builder(&config)
             .local_address("::".parse::<IpAddr>().unwrap())
             .build()
-            .unwrap();
+            .into_diagnostic()?;
 
-        let last_uptime_check = Self::query_last_fetch(&db).await?;
+        let last_uptime_check = Self::query_last_fetch(&db).await
+            .into_diagnostic()
+            .wrap_err("Fetching last uptime check failed!")?;
         tracing::info!(?last_uptime_check);
         let scanner = Self {
             inner: Arc::new(InnerScanner {
@@ -211,10 +210,12 @@ impl Scanner {
                 last_uptime_check: Mutex::new(last_uptime_check),
                 rss_check_regex: builder_regex_rss
                     .build()
-                    .expect("Invalid RSS Content regex!"),
+                    .into_diagnostic()
+                    .wrap_err("Invalid RSS Content regex!")?,
             }),
         };
-        scanner.update_cache().await?;
+        scanner.update_cache().await.into_diagnostic()
+        .wrap_err("Initial cache update failed!")?;
         Ok(scanner)
     }
 
@@ -228,6 +229,29 @@ impl Scanner {
         Ok(Utc.timestamp_opt(time,0).unwrap())
     }
 
+    pub async fn query_latest_check<T: ConnectionTrait>(
+        &self,
+        connection: &T,
+    ) -> Result<Vec<LatestCheck>> {
+        let health_checks = LatestCheck::find_by_statement(Statement::from_sql_and_values(
+            DbBackend::Sqlite,
+            r#"
+            WITH latest AS(
+                SELECT u.host,MAX(u.time) as time FROM health_check u
+                GROUP BY u.host
+            )
+            SELECT u.host,healthy,h.domain FROM health_check u
+            JOIN host h ON h.id = u.host
+            JOIN latest l ON l.host = u.host AND l.time = u.time
+            WHERE h.enabled = true
+            "#,
+            [],
+        ))
+        .all(connection)
+        .await?;
+        Ok(health_checks)
+    }
+
     pub async fn run(mut self) -> Result<()> {
         loop {
             if self.is_instance_list_outdated() {
@@ -283,158 +307,11 @@ impl Scanner {
         Utc::now().signed_duration_since(val).to_std().unwrap() >= self.inner.config.list_fetch_interval
     }
 
-    #[instrument]
-    async fn update_instacelist(&mut self) -> Result<()> {
-        let start = Instant::now();
-        let html: String = self.fetch_instance_list().await?;
-        let parsed_instances = self.inner.instance_parser.parse_instancelist(
-            &html,
-            &self.inner.config.additional_hosts,
-            &self.inner.config.additional_host_country,
-            false,
-        )?;
-
-        let transaction = self.inner.db.begin().await?;
-
-        // find all currently enabled instances
-        let enabled_hosts = Host::find()
-            .filter(host::Column::Enabled.eq(true))
-            .all(&transaction)
-            .await?;
-        // make a diff and remove the ones not found while parsing
-        let time: chrono::DateTime<Utc> = Utc::now();
-        let mut removed = 0;
-        for host in enabled_hosts.iter() {
-            if !parsed_instances.contains_key(&host.domain) {
-                host::ActiveModel {
-                    id: ActiveValue::Set(host.id),
-                    enabled: ActiveValue::Set(false),
-                    updated: ActiveValue::Set(time.timestamp()),
-                    ..Default::default()
-                }
-                .update(&transaction)
-                .await?;
-                removed += 1;
-            }
-        }
-        // now update/insert the existing ones
-        let found_instances: usize = parsed_instances.len();
-        // find last update checks to detect spam
-        let last_status = self.query_latest_check(&transaction).await?;
-        let mut join_set = JoinSet::new();
-        for (_, instance) in parsed_instances {
-            // TODO: parallelize this!
-            let scanner_c = self.clone();
-            // detect already offline host and prevent log spam
-            let muted_host = match self.inner.config.auto_mute {
-                false => false,
-                true => last_status
-                    .iter()
-                    .find(|v| v.domain == instance.domain)
-                    .map_or(false, |check| !check.healthy),
-            };
-            // tracing::trace!(muted_host,instance=?instance,last_status=?last_status);
-            join_set.spawn(async move {
-                let (connectivity, rss, version, version_url) = match Url::parse(&instance.url) {
-                    Err(_) => {
-                        if !muted_host {
-                            tracing::info!(url = instance.url, "Instance URL invalid");
-                        }
-                        (None, false, None, None)
-                    }
-                    Ok(mut url) => {
-                        let connectivity = scanner_c.check_connectivity(url.as_str()).await;
-                        let rss = scanner_c.has_rss(&mut url, muted_host).await;
-                        match scanner_c.nitter_version(&mut url, muted_host).await {
-                            Some(version) => (
-                                connectivity,
-                                rss,
-                                Some(version.version_name),
-                                Some(version.url),
-                            ),
-                            None => (connectivity, rss, None, None),
-                        }
-                    }
-                };
-
-                host::ActiveModel {
-                    id: ActiveValue::NotSet,
-                    domain: ActiveValue::Set(instance.domain),
-                    country: ActiveValue::Set(instance.country),
-                    url: ActiveValue::Set(instance.url),
-                    enabled: ActiveValue::Set(true),
-                    version: ActiveValue::Set(version),
-                    version_url: ActiveValue::Set(version_url),
-                    rss: ActiveValue::Set(rss),
-                    updated: ActiveValue::Set(time.timestamp()),
-                    connectivity: ActiveValue::Set(connectivity),
-                }
-            });
-        }
-        while let Some(update_model) = join_set.join_next().await.map(|v| v.unwrap()) {
-            Host::insert(update_model)
-                .on_conflict(
-                    OnConflict::column(host::Column::Domain)
-                        .update_columns([
-                            host::Column::Enabled,
-                            host::Column::Updated,
-                            host::Column::Url,
-                            host::Column::Rss,
-                            host::Column::Version,
-                            host::Column::VersionUrl,
-                            host::Column::Country,
-                            host::Column::Connectivity,
-                        ])
-                        .to_owned(),
-                )
-                .exec(&transaction)
-                .await?;
-        }
-
-        transaction.commit().await?;
-        let end = Instant::now();
-        let took_ms = end.saturating_duration_since(start).as_millis();
-        {
-            *self.inner.last_list_fetch.lock().unwrap() = Utc::now();
-        }
-        tracing::debug!(
-            removed = removed,
-            found = found_instances,
-            took_ms = took_ms
-        );
-        Ok(())
-    }
-
     async fn fetch_instance_list(&self) -> Result<String> {
         let (_, body) = self.fetch_url(&self.inner.config.instance_list_url).await?;
         Ok(body)
     }
 
-    /// Check ipv4/6 connectivity of host
-    async fn check_connectivity(&self, url: &str) -> Option<host::Connectivity> {
-        let ipv4 = self
-            .inner
-            .client_ipv4
-            .get(url)
-            .send()
-            .await
-            .map_or(false, |res| res.status().is_success());
-        let ipv6 = self
-            .inner
-            .client_ipv6
-            .get(url)
-            .send()
-            .await
-            .map_or(false, |res| res.status().is_success());
-
-        match (ipv4, ipv6) {
-            (true, true) => Some(host::Connectivity::All),
-            (false, true) => Some(host::Connectivity::IPv6),
-            (true, false) => Some(host::Connectivity::IPv4),
-            (false, false) => None,
-        }
-    }
-
     async fn fetch_url(&self, url: &str) -> std::result::Result<(u16, String), FetchError> {
         let fetch_res = self.inner.client.get(url).send().await?;
         let code = fetch_res.status().as_u16();
@@ -481,29 +358,6 @@ impl Scanner {
 
         Ok((code, body))
     }
-
-    pub async fn query_latest_check<T: ConnectionTrait>(
-        &self,
-        connection: &T,
-    ) -> Result<Vec<LatestCheck>> {
-        let health_checks = LatestCheck::find_by_statement(Statement::from_sql_and_values(
-            DbBackend::Sqlite,
-            r#"
-            WITH latest AS(
-                SELECT u.host,MAX(u.time) as time FROM health_check u
-                GROUP BY u.host
-            )
-            SELECT u.host,healthy,h.domain FROM health_check u
-            JOIN host h ON h.id = u.host
-            JOIN latest l ON l.host = u.host AND l.time = u.time
-            WHERE h.enabled = true
-            "#,
-            [],
-        ))
-        .all(connection)
-        .await?;
-        Ok(health_checks)
-    }
 }
 
 #[cfg(test)]
@@ -514,10 +368,10 @@ mod test {
     use entities::health_check;
     use entities::state::scanner::Config;
     use migration::MigratorTrait;
-    use sea_orm::{ConnectOptions, Database};
+    use sea_orm::{ConnectOptions, Database, ActiveValue, ActiveModelTrait};
     use tokio::{fs::File, io::AsyncWriteExt};
 
-    async fn db_init() -> DatabaseConnection {
+    pub(crate) async fn db_init() -> DatabaseConnection {
         let db = Database::connect(ConnectOptions::new(
             "sqlite:./test_db.db?mode=rwc".to_owned(),
         ))
@@ -582,27 +436,6 @@ mod test {
             .unwrap();
         dbg!(scanner.generate_cache_data().await.unwrap());
     }
-
-    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-    #[ignore]
-    async fn connectivity_test() {
-        let db = db_init().await;
-        let scanner = Scanner::new(db, Config::test_defaults(), entities::state::new())
-            .await
-            .unwrap();
-        assert_eq!(
-            scanner.check_connectivity("https://v4.ipv6test.app").await,
-            Some(host::Connectivity::IPv4)
-        );
-        assert_eq!(
-            scanner.check_connectivity("https://ipv6test.app").await,
-            Some(host::Connectivity::All)
-        );
-        assert_eq!(
-            scanner.check_connectivity("https://v6.ipv6test.app").await,
-            Some(host::Connectivity::IPv6)
-        );
-    }
 }
 
 impl std::fmt::Debug for Scanner {

scanner/src/list_update.rs

@@ -0,0 +1,205 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+//! Updates the list of available instances, fetching all required fields
+
+use std::time::Duration;
+use std::time::Instant;
+
+use chrono::Utc;
+use entities::host;
+use entities::prelude::Host;
+use reqwest::Url;
+use sea_orm::{QueryFilter, TransactionTrait, ActiveModelTrait, ActiveValue, ColumnTrait, EntityTrait};
+use sea_query::OnConflict;
+use tokio::task::JoinSet;
+use tracing::instrument;
+
+use crate::Scanner;
+use crate::Result;
+
+impl Scanner {
+    /// Fetches the list of all instances from the wiki.  
+    /// Updates all fields for host::Model, including connectivity, rss, version and enabled.
+    #[instrument]
+    pub(crate) async fn update_instacelist(&mut self) -> Result<()> {
+        let start = Instant::now();
+        let html: String = self.fetch_instance_list().await?;
+        let parsed_instances = self.inner.instance_parser.parse_instancelist(
+            &html,
+            &self.inner.config.additional_hosts,
+            &self.inner.config.additional_host_country,
+            false,
+        )?;
+
+        let transaction = self.inner.db.begin().await?;
+
+        // find all currently enabled instances
+        let enabled_hosts = Host::find()
+            .filter(host::Column::Enabled.eq(true))
+            .all(&transaction)
+            .await?;
+        // make a diff and remove the ones not found while parsing
+        let time: chrono::DateTime<Utc> = Utc::now();
+        let mut removed = 0;
+        for host in enabled_hosts.iter() {
+            if !parsed_instances.contains_key(&host.domain) {
+                host::ActiveModel {
+                    id: ActiveValue::Set(host.id),
+                    enabled: ActiveValue::Set(false),
+                    updated: ActiveValue::Set(time.timestamp()),
+                    ..Default::default()
+                }
+                .update(&transaction)
+                .await?;
+                removed += 1;
+            }
+        }
+        // now update/insert the existing ones
+        let found_instances: usize = parsed_instances.len();
+        // find last update checks to detect spam
+        let last_status = self.query_latest_check(&transaction).await?;
+        let mut join_set = JoinSet::new();
+        for (_, instance) in parsed_instances {
+            // TODO: parallelize this!
+            let scanner_c = self.clone();
+            // detect already offline host and prevent log spam
+            let muted_host = match self.inner.config.auto_mute {
+                false => false,
+                true => last_status
+                    .iter()
+                    .find(|v| v.domain == instance.domain)
+                    .map_or(false, |check| !check.healthy),
+            };
+            // tracing::trace!(muted_host,instance=?instance,last_status=?last_status);
+            join_set.spawn(async move {
+                let (connectivity, rss, version, version_url) = match Url::parse(&instance.url) {
+                    Err(_) => {
+                        if !muted_host {
+                            tracing::info!(url = instance.url, "Instance URL invalid");
+                        }
+                        (None, false, None, None)
+                    }
+                    Ok(mut url) => {
+                        let connectivity = scanner_c.check_connectivity(&mut url).await;
+                        // prevent DoS
+                        tokio::time::sleep(Duration::from_secs(1)).await;
+                        let rss = scanner_c.has_rss(&mut url, muted_host).await;
+                        tokio::time::sleep(Duration::from_secs(1)).await;
+                        match scanner_c.nitter_version(&mut url, muted_host).await {
+                            Some(version) => (
+                                connectivity,
+                                rss,
+                                Some(version.version_name),
+                                Some(version.url),
+                            ),
+                            None => (connectivity, rss, None, None),
+                        }
+                    }
+                };
+
+                host::ActiveModel {
+                    id: ActiveValue::NotSet,
+                    domain: ActiveValue::Set(instance.domain),
+                    country: ActiveValue::Set(instance.country),
+                    url: ActiveValue::Set(instance.url),
+                    enabled: ActiveValue::Set(true),
+                    version: ActiveValue::Set(version),
+                    version_url: ActiveValue::Set(version_url),
+                    rss: ActiveValue::Set(rss),
+                    updated: ActiveValue::Set(time.timestamp()),
+                    connectivity: ActiveValue::Set(connectivity),
+                }
+            });
+        }
+        while let Some(update_model) = join_set.join_next().await.map(|v| v.unwrap()) {
+            Host::insert(update_model)
+                .on_conflict(
+                    OnConflict::column(host::Column::Domain)
+                        .update_columns([
+                            host::Column::Enabled,
+                            host::Column::Updated,
+                            host::Column::Url,
+                            host::Column::Rss,
+                            host::Column::Version,
+                            host::Column::VersionUrl,
+                            host::Column::Country,
+                            host::Column::Connectivity,
+                        ])
+                        .to_owned(),
+                )
+                .exec(&transaction)
+                .await?;
+        }
+
+        transaction.commit().await?;
+        let end = Instant::now();
+        let took_ms = end.saturating_duration_since(start).as_millis();
+        {
+            *self.inner.last_list_fetch.lock().unwrap() = Utc::now();
+        }
+        tracing::debug!(
+            removed = removed,
+            found = found_instances,
+            took_ms = took_ms
+        );
+        Ok(())
+    }
+
+    /// Check ipv4/6 connectivity of host
+    async fn check_connectivity(&self, url: &mut Url) -> Option<host::Connectivity> {
+        url.set_path(&self.inner.config.connectivity_path);
+        let ipv4 = self
+            .inner
+            .client_ipv4
+            .get(url.as_str())
+            .send()
+            .await
+            .map_or(false, |res| res.status().is_success());
+        // prevent DoS
+        tokio::time::sleep(Duration::from_secs(1)).await;
+        let ipv6 = self
+            .inner
+            .client_ipv6
+            .get(url.as_str())
+            .send()
+            .await
+            .map_or(false, |res| res.status().is_success());
+
+        match (ipv4, ipv6) {
+            (true, true) => Some(host::Connectivity::All),
+            (false, true) => Some(host::Connectivity::IPv6),
+            (true, false) => Some(host::Connectivity::IPv4),
+            (false, false) => None,
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use entities::state::scanner::Config;
+    use tracing_test::traced_test;
+
+    use crate::{Scanner, test::db_init};
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    #[traced_test]
+    #[ignore]
+    async fn connectivity_test() {
+        let db = db_init().await;
+        let scanner = Scanner::new(db, Config::test_defaults(), entities::state::new())
+            .await
+            .unwrap();
+        assert_eq!(
+            scanner.check_connectivity(&mut Url::parse("https://v4.ipv6test.app").unwrap()).await,
+            Some(host::Connectivity::IPv4)
+        );
+        assert_eq!(
+            scanner.check_connectivity(&mut Url::parse("https://ipv6test.app").unwrap()).await,
+            Some(host::Connectivity::All)
+        );
+        assert_eq!(
+            scanner.check_connectivity(&mut Url::parse("https://v6.ipv6test.app").unwrap()).await,
+            Some(host::Connectivity::IPv6)
+        );
+    }
+}

server/src/website.rs

@@ -49,6 +49,7 @@ pub async fn about(
     paths.push(&scanner_config.about_path);
     paths.push(&scanner_config.rss_path);
     paths.push(&scanner_config.profile_path);
+    paths.push(&scanner_config.connectivity_path);
     context.insert("checked_paths", &paths);
     context.insert(
         "uptime_interval_s",

src/main.rs

@@ -91,8 +91,7 @@ async fn _main() -> miette::Result<()> {
         disable_health_checks,
     )
     .await
-    .into_diagnostic()
-    .wrap_err("Failed starting background scanner")?;
+    .wrap_err("Crash starting background scanner")?;
 
     let addr = std::net::SocketAddr::from(([127, 0, 0, 1], port));
     server::start(&addr, pool, server_config, scanner_config, cache)
@@ -157,6 +156,7 @@ fn read_scanner_cfg() -> miette::Result<ScannerConfig> {
         bad_hosts,
         cleanup_interval: Duration::from_secs(cleanup_interval),
         error_retention_per_host,
+        connectivity_path: String::from("/"),
     }))
 }