diff --git a/.env.example b/.env.example index 331e9f3..1a51c20 100644 --- a/.env.example +++ b/.env.example @@ -32,9 +32,6 @@ BAD_HOSTS="tweet.whateveritworks.org" REFERER="https://localhost" # Ping AVG period in hours PING_RANGE_H="3" -# enable/disable startup scanning -# set to to false unless debugging / developing -DISABLE_STARTUP_SCAN=false # don't emit errors when re-checking offline hosts AUTO_MUTE=true # branch to use for fetching the latest nitter commit @@ -46,4 +43,6 @@ SESSION_TTL_SECONDS="2592000" # admin instance domains ADMIN_DOMAINS="example.com,foo.net" # token name used for logging in - has to be a valid DNS TXT key -LOGIN_TOKEN_NAME="localhost" \ No newline at end of file +LOGIN_TOKEN_NAME="localhost" +# completely disable any health checks - debugging only +DISABLE_HEALTH_CHECKS=false \ No newline at end of file diff --git a/entities/src/state/scanner.rs b/entities/src/state/scanner.rs index 16b13e1..141edf7 100644 --- a/entities/src/state/scanner.rs +++ b/entities/src/state/scanner.rs @@ -1,6 +1,7 @@ // SPDX-License-Identifier: AGPL-3.0-only use std::{sync::Arc, time::Duration}; pub type ScannerConfig = Arc; +#[derive(Debug)] pub struct Config { /// time until next instance list fetch pub list_fetch_interval: Duration, diff --git a/scanner/src/instance_check.rs b/scanner/src/instance_check.rs index 5ae52db..c63d34b 100644 --- a/scanner/src/instance_check.rs +++ b/scanner/src/instance_check.rs @@ -47,7 +47,7 @@ impl Scanner { while let Some(_) = join_set.join_next().await {} let end = Instant::now(); let took_ms = end.saturating_duration_since(start).as_millis(); - *self.inner.last_uptime_check.lock().unwrap() = end; + *self.inner.last_uptime_check.lock().unwrap() = Utc::now(); tracing::debug!(hosts = tasks, took_ms = took_ms, "checked uptime"); Ok(()) } diff --git a/scanner/src/lib.rs b/scanner/src/lib.rs index 01ff573..31637c8 100644 --- a/scanner/src/lib.rs +++ b/scanner/src/lib.rs @@ -6,11 +6,11 @@ use std::{ }; use about_parser::AboutParser; -use chrono::Utc; +use chrono::{Utc, DateTime, TimeZone, Duration}; use entities::{ host, prelude::*, - state::{error_cache::HostError, scanner::ScannerConfig, AppState}, + state::{error_cache::HostError, scanner::ScannerConfig, AppState}, health_check, }; use instance_parser::InstanceParser; use profile_parser::ProfileParser; @@ -22,7 +22,7 @@ use reqwest::{ use sea_orm::{ sea_query::OnConflict, ActiveModelTrait, ActiveValue, ColumnTrait, ConnectionTrait, DatabaseConnection, DbBackend, EntityTrait, FromQueryResult, QueryFilter, Statement, - TransactionTrait, + TransactionTrait, QuerySelect, }; use thiserror::Error; use tokio::{task::JoinSet, time::sleep}; @@ -113,12 +113,16 @@ pub async fn run_scanner( db: DatabaseConnection, config: ScannerConfig, app_state: AppState, - disable_startup_scan: bool, + disable_health_checks: bool, ) -> Result<()> { let scanner = Scanner::new(db, config, app_state).await?; + if disable_health_checks { + tracing::error!("Health checks disabled!"); + return Ok(()) + } tokio::spawn(async move { - scanner.run(disable_startup_scan).await.unwrap(); + scanner.run().await.unwrap(); }); Ok(()) @@ -137,8 +141,8 @@ struct InnerScanner { instance_parser: InstanceParser, about_parser: AboutParser, profile_parser: ProfileParser, - last_list_fetch: Mutex, - last_uptime_check: Mutex, + last_list_fetch: Mutex>, + last_uptime_check: Mutex>, rss_check_regex: Regex, client_ipv4: Client, client_ipv6: Client, @@ -187,6 +191,9 @@ impl Scanner { .local_address("::".parse::().unwrap()) .build() .unwrap(); + + let last_uptime_check = Self::query_last_fetch(&db).await?; + tracing::info!(?last_uptime_check); let scanner = Self { inner: Arc::new(InnerScanner { db, @@ -198,8 +205,8 @@ impl Scanner { instance_parser: InstanceParser::new(), about_parser: AboutParser::new(), profile_parser: ProfileParser::new(), - last_list_fetch: Mutex::new(Instant::now()), - last_uptime_check: Mutex::new(Instant::now()), + last_list_fetch: Mutex::new(last_uptime_check), + last_uptime_check: Mutex::new(last_uptime_check), rss_check_regex: builder_regex_rss .build() .expect("Invalid RSS Content regex!"), @@ -209,15 +216,24 @@ impl Scanner { Ok(scanner) } - pub async fn run(mut self, disable_startup_scan: bool) -> Result<()> { - let mut first_run = !disable_startup_scan; + /// Retrieves the last uptime fetch that happened + pub async fn query_last_fetch(db: &DatabaseConnection) -> Result> { + let time = health_check::Entity::find().column_as(health_check::Column::Time.max(), health_check::Column::Time) + .one(db) + .await? + .map(|model| model.time) + .unwrap_or_default(); + Ok(Utc.timestamp_opt(time,0).unwrap()) + } + + pub async fn run(mut self) -> Result<()> { loop { - if first_run || self.is_instance_list_outdated() { + if self.is_instance_list_outdated() { if let Err(e) = self.update_instacelist().await { tracing::error!(error=?e,"Failed updating instance list"); } } - if first_run || self.is_instance_check_outdated() { + if self.is_instance_check_outdated() { if let Err(e) = self.check_uptime().await { tracing::error!(error=?e,"Failed checking instance"); } @@ -226,7 +242,6 @@ impl Scanner { tracing::error!(error=?e,"Failed updating cache!"); } self.sleep_till_deadline().await; - first_run = false; } } @@ -235,28 +250,35 @@ impl Scanner { self.last_uptime_check() + self.inner.config.instance_check_interval; let delay_list_update = self.last_list_fetch() + self.inner.config.list_fetch_interval; - - let delay = delay_instance_check.min(delay_list_update); - tracing::trace!(duration=?delay-Instant::now(),"scanner sleeping"); - sleep(delay.duration_since(Instant::now())).await; + tracing::debug!(?delay_list_update,?delay_instance_check); + let next_deadline = delay_instance_check.min(delay_list_update); + let now = Utc::now(); + let sleep_time = next_deadline.signed_duration_since(now); + if sleep_time <= Duration::zero() { + // schedule right now, also std duration can't be negative + return; + } + let sleep_time = sleep_time.to_std().unwrap(); + tracing::trace!(duration=?sleep_time,"scanner sleeping"); + sleep(sleep_time).await; } - fn last_uptime_check(&self) -> Instant { + fn last_uptime_check(&self) -> DateTime { *self.inner.last_uptime_check.lock().unwrap() } - fn last_list_fetch(&self) -> Instant { + fn last_list_fetch(&self) -> DateTime { *self.inner.last_list_fetch.lock().unwrap() } fn is_instance_check_outdated(&self) -> bool { let val = self.last_uptime_check(); - Instant::now().saturating_duration_since(val) >= self.inner.config.instance_check_interval + Utc::now().signed_duration_since(val).to_std().unwrap() >= self.inner.config.instance_check_interval } fn is_instance_list_outdated(&self) -> bool { let val = self.last_list_fetch(); - Instant::now().saturating_duration_since(val) >= self.inner.config.list_fetch_interval + Utc::now().signed_duration_since(val).to_std().unwrap() >= self.inner.config.list_fetch_interval } #[instrument] @@ -371,7 +393,7 @@ impl Scanner { let end = Instant::now(); let took_ms = end.saturating_duration_since(start).as_millis(); { - *self.inner.last_list_fetch.lock().unwrap() = end; + *self.inner.last_list_fetch.lock().unwrap() = Utc::now(); } tracing::debug!( removed = removed, @@ -527,7 +549,7 @@ mod test { #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[ignore] - async fn update_instacelist() { + async fn update_instancelist() { let db = db_init().await; let scanner = Scanner::new(db, Config::test_defaults(), entities::state::new()) .await diff --git a/src/main.rs b/src/main.rs index 8588b31..22236a2 100644 --- a/src/main.rs +++ b/src/main.rs @@ -75,13 +75,13 @@ async fn _main() -> miette::Result<()> { let cache = entities::state::new(); - let disable_startup_scan = require_env_str("DISABLE_STARTUP_SCAN")? == "true"; + let disable_health_checks = require_env_str("DISABLE_HEALTH_CHECKS")? == "true"; scanner::run_scanner( pool.clone(), scanner_config.clone(), cache.clone(), - disable_startup_scan, + disable_health_checks, ) .await .into_diagnostic()