optimize database

datetime as ints, strict and no rowid

Cargo.lock

@@ -1794,6 +1794,7 @@ version = "0.1.0"
 dependencies = [
  "async-std",
  "sea-orm-migration",
+ "tracing",
 ]
 
 [[package]]

entities/src/health_check.rs

@@ -4,10 +4,10 @@ use sea_orm::entity::prelude::*;
 use serde::Serialize;
 
 #[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize)]
-#[sea_orm(table_name = "update_check")]
+#[sea_orm(table_name = "health_check")]
 pub struct Model {
     #[sea_orm(primary_key, auto_increment = false)]
-    pub time: DateTimeUtc,
+    pub time: i64,
     #[sea_orm(primary_key, auto_increment = false)]
     pub host: i32,
     pub resp_time: Option<i32>,

entities/src/host.rs

@@ -1,9 +1,18 @@
 //! `SeaORM` Entity. Generated by sea-orm-codegen 0.11.3
 
-use sea_orm::entity::prelude::*;
+use sea_orm::{entity::prelude::*, sea_query::{Expr, SimpleExpr}};
 use serde::Serialize;
 
-#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize)]
+#[derive(Copy, Clone, Default, Debug, DeriveEntity)]
+pub struct Entity;
+
+impl EntityName for Entity {
+    fn table_name(&self) -> &str {
+        "host"
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, DeriveModel, DeriveActiveModel, Eq, Serialize)]
 #[sea_orm(table_name = "host")]
 pub struct Model {
     #[sea_orm(primary_key)]
@@ -14,16 +23,72 @@ pub struct Model {
     pub rss: bool,
     pub version: Option<String>,
     /// Last time the url and enabled were updated, *not* the rss
-    pub updated: DateTimeUtc,
+    pub updated: i64,
 }
 
-#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+#[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
+pub enum Column {
+    Id,
+    Domain,
+    Url,
+    Version,
+    Enabled,
+    Rss,
+    Updated,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DerivePrimaryKey)]
+pub enum PrimaryKey {
+    Id,
+}
+
+impl PrimaryKeyTrait for PrimaryKey {
+    type ValueType = i32;
+    fn auto_increment() -> bool {
+        true
+    }
+}
+
+#[derive(Copy, Clone, Debug, EnumIter)]
 pub enum Relation {
-    #[sea_orm(has_many = "super::update_check::Entity")]
     UpdateCheck,
 }
 
-impl Related<super::update_check::Entity> for Entity {
+impl ColumnTrait for Column {
+    type EntityName = Entity;
+    fn def(&self) -> ColumnDef {
+        match self {
+            Self::Id => ColumnType::Integer.def(),
+            Self::Domain => ColumnType::String(None).def(),
+            Self::Url => ColumnType::String(None).def(),
+            Self::Version => ColumnType::String(None).def().null(),
+            Self::Enabled => ColumnType::Integer.def(),
+            Self::Rss => ColumnType::Integer.def(),
+            Self::Updated => ColumnType::Integer.def(),
+        }
+    }
+
+    // fn select_as(&self, expr: Expr) -> SimpleExpr {
+    //     Column::CaseInsensitiveText => expr.cast_as(Alias::new("text")),
+    //     _ => self.select_enum_as(expr),
+    // }
+
+    // /// Cast value of a column into the correct type for database storage.
+    // fn save_as(&self, val: Expr) -> SimpleExpr {
+    //     Column::CaseInsensitiveText => val.cast_as(Alias::new("citext")),
+    //     _ => self.save_enum_as(val),
+    // }
+}
+
+impl RelationTrait for Relation {
+    fn def(&self) -> RelationDef {
+        match self {
+            Self::UpdateCheck => Entity::has_many(super::health_check::Entity).into(),
+        }
+    }
+}
+
+impl Related<super::health_check::Entity> for Entity {
     fn to() -> RelationDef {
         Relation::UpdateCheck.def()
     }

entities/src/mod.rs

@@ -3,7 +3,7 @@
 pub mod prelude;
 
 pub mod host;
-pub mod update_check;
+pub mod health_check;
 
 // has to be re-added
 pub mod state;

entities/src/prelude.rs

@@ -1,4 +1,4 @@
 //! `SeaORM` Entity. Generated by sea-orm-codegen 0.11.3
 
 pub use super::host::Entity as Host;
-pub use super::update_check::Entity as UpdateCheck;
+pub use super::health_check::Entity as UpdateCheck;

migration/Cargo.toml

@@ -10,7 +10,7 @@ path = "src/lib.rs"
 
 [dependencies]
 async-std = { version = "^1", features = ["attributes", "tokio1"] }
-
+tracing = { workspace = true }
 [dependencies.sea-orm-migration]
 workspace = true
 features = [

migration/src/lib.rs

@@ -1,12 +1,18 @@
 pub use sea_orm_migration::prelude::*;
 
 mod m20220101_000001_create_table;
+mod m20230729_010231_datetime_rowid;
+mod m20230729_230909_datetime_int_host;
 
 pub struct Migrator;
 
 #[async_trait::async_trait]
 impl MigratorTrait for Migrator {
     fn migrations() -> Vec<Box<dyn MigrationTrait>> {
-        vec![Box::new(m20220101_000001_create_table::Migration)]
+        vec![
+            Box::new(m20220101_000001_create_table::Migration),
+            Box::new(m20230729_010231_datetime_rowid::Migration),
+            Box::new(m20230729_230909_datetime_int_host::Migration),
+        ]
     }
 }

migration/src/m20220101_000001_create_table.rs

@@ -1,3 +1,4 @@
+// SPDX-License-Identifier: AGPL-3.0-only
 use sea_orm_migration::prelude::*;
 
 #[derive(DeriveMigrationName)]

migration/src/m20230729_010231_datetime_rowid.rs

@@ -0,0 +1,104 @@
+use std::borrow::Cow;
+
+use sea_orm_migration::{
+    prelude::*,
+    sea_orm::{
+        prelude::DateTimeUtc, DbBackend, DeriveEntityModel, FromQueryResult, RuntimeErr, Statement,
+    },
+};
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[derive(Debug, FromQueryResult)]
+pub struct UpdateCheck {
+    time: DateTimeUtc,
+    host: i32,
+    resp_time: Option<i32>,
+    healthy: bool,
+    response_code: Option<i32>,
+}
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        let cmd = r#"CREATE TABLE "health_check_new" (
+            "time" integer NOT NULL,
+            "host" integer NOT NULL,
+            "resp_time" integer,
+            "healthy" integer NOT NULL,
+            "response_code" integer,
+            CONSTRAINT "pk_health_check_new" PRIMARY KEY ("host", "time"),
+            FOREIGN KEY ("host") REFERENCES "host" ("id") ON DELETE CASCADE ON UPDATE CASCADE
+        ) WITHOUT ROWID, STRICT;"#;
+        let db = manager.get_connection();
+        db.execute_unprepared("BEGIN EXCLUSIVE").await?;
+        db.execute_unprepared(cmd).await?;
+        tracing::info!("fetching all healtcheck entries..");
+        let data = UpdateCheck::find_by_statement(Statement::from_sql_and_values(
+            DbBackend::Sqlite,
+            r#"SELECT * FROM update_check"#,
+            [],
+        ))
+        .all(db)
+        .await?;
+
+        tracing::info!("migrating {} entries to temp table..", data.len());
+        let mut duplicates = 0;
+        for entry in data.into_iter() {
+            if let Err(err) = db
+                .execute(Statement::from_sql_and_values(
+                    DbBackend::Sqlite,
+                    r#"INSERT INTO health_check_new (
+                    time,
+                    host,
+                    resp_time,
+                    healthy,
+                    response_code) VALUES ($1,$2,$3,$4,$5)"#,
+                    [
+                        entry.time.timestamp().into(),
+                        entry.host.into(),
+                        entry.resp_time.into(),
+                        entry.healthy.into(),
+                        entry.response_code.into(),
+                    ],
+                ))
+                .await
+            {
+                if let DbErr::Exec(RuntimeErr::SqlxError(e)) = &err {
+                    if let Some(err) = e.as_database_error() {
+                        if err.code() == Some(Cow::Borrowed("1555")) {
+                            duplicates += 1;
+                            continue;
+                        }
+                    }
+                }
+                tracing::info!(entry=?entry);
+                return Err(err);
+            }
+        }
+        tracing::info!("dropped {} duplicate timestamps..", duplicates);
+        tracing::info!("dropping old table..");
+        db.execute_unprepared("DROP TABLE update_check").await?;
+
+        db.execute_unprepared(&cmd.replace("health_check_new", "health_check"))
+            .await?;
+        tracing::info!("inserting back into final table..");
+        db.execute_unprepared(
+            r#"INSERT INTO health_check
+            SELECT * 
+            FROM health_check_new WHERE true"#,
+        )
+        .await?;
+        tracing::info!("dropping temp table..");
+        db.execute_unprepared("DROP TABLE health_check_new").await?;
+        tracing::info!("cleaning up db..");
+        db.execute_unprepared("COMMIT TRANSACTION").await?;
+        db.execute_unprepared("VACUUM").await?;
+        Ok(())
+    }
+
+    async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> {
+        panic!("Can't migrate down");
+    }
+}

migration/src/m20230729_230909_datetime_int_host.rs

@@ -0,0 +1,102 @@
+use sea_orm_migration::{
+    prelude::*,
+    sea_orm::{prelude::DateTimeUtc, DbBackend, FromQueryResult, Statement},
+};
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[derive(Debug, FromQueryResult)]
+pub struct Host {
+    pub id: i32,
+    pub domain: String,
+    pub url: String,
+    pub enabled: bool,
+    pub rss: bool,
+    pub version: Option<String>,
+    pub updated: DateTimeUtc,
+}
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        let cmd = r#"CREATE TABLE "host_new" (
+            "id" integer NOT NULL PRIMARY KEY AUTOINCREMENT,
+            "domain" text NOT NULL UNIQUE,
+            "url" text NOT NULL,
+            "version" text,
+            "enabled" integer NOT NULL,
+            "rss" integer NOT NULL,
+            "updated" integer NOT NULL
+        ) STRICT;"#;
+        let db = manager.get_connection();
+        db.execute_unprepared("PRAGMA foreign_keys = 0").await?;
+        db.execute_unprepared("BEGIN EXCLUSIVE").await?;
+        db.execute_unprepared(cmd).await?;
+        tracing::info!("fetching all host entries..");
+        let data = Host::find_by_statement(Statement::from_sql_and_values(
+            DbBackend::Sqlite,
+            r#"SELECT * FROM host"#,
+            [],
+        ))
+        .all(db)
+        .await?;
+
+        tracing::info!("migrating {} entries to temp table..", data.len());
+        for entry in data.into_iter() {
+            db.execute(Statement::from_sql_and_values(
+                DbBackend::Sqlite,
+                r#"INSERT INTO host_new (
+                    id,
+                    domain,
+                    url,
+                    version,
+                    enabled,
+                    rss,
+                    updated ) VALUES ($1,$2,$3,$4,$5,$6,$7)"#,
+                [
+                    entry.id.into(),
+                    entry.domain.into(),
+                    entry.url.into(),
+                    entry.version.into(),
+                    entry.enabled.into(),
+                    entry.rss.into(),
+                    entry.updated.timestamp().into(),
+                ],
+            ))
+            .await?;
+        }
+        tracing::info!("dropping old table..");
+        db.execute_unprepared("DROP TABLE host").await?;
+
+        db.execute_unprepared(&cmd.replace("host_new", "host"))
+            .await?;
+        tracing::info!("inserting back into final table..");
+        db.execute_unprepared(
+            r#"INSERT INTO host
+            SELECT * 
+            FROM host_new WHERE true"#,
+        )
+        .await?;
+        tracing::info!("dropping temp table..");
+        db.execute_unprepared("DROP TABLE host_new").await?;
+        tracing::info!("cleaning up db..");
+        db.execute_unprepared("COMMIT TRANSACTION").await?;
+        db.execute_unprepared("PRAGMA foreign_keys = 1").await?;
+        db.execute_unprepared("VACUUM").await?;
+        Ok(())
+    }
+
+    async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> {
+        panic!("Can't migrate down");
+    }
+}
+
+/// Learn more at https://docs.rs/sea-query#iden
+#[derive(Iden)]
+enum Post {
+    Table,
+    Id,
+    Title,
+    Text,
+}

scanner/src/cache_update.rs

@@ -139,12 +139,12 @@ impl Scanner {
         let last_pings = PingEntry::find_by_statement(Statement::from_sql_and_values(
             DbBackend::Sqlite,
             r#"
-            SELECT u.host,(CASE u.healthy WHEN true THEN u.resp_time ELSE null END) as ping FROM update_check u
+            SELECT u.host,(CASE u.healthy WHEN true THEN u.resp_time ELSE null END) as ping FROM health_check u
             JOIN host h ON h.id = u.host
             WHERE h.enabled = true AND u.time >= $1
             ORDER BY u.host,u.time ASC
             "#,
-            [age.into()],
+            [age.timestamp().into()],
         ))
         .all(&self.inner.db)
         .await?;
@@ -157,6 +157,12 @@ impl Scanner {
         let mut current_entry = LastPings::default();
         let mut last_host = first.host;
         let mut non_null_entries = first.ping.as_ref().map_or(0, |_|1);
+        if let Some(ping) = first.ping.as_ref() {
+            current_entry.avg = Some(*ping);
+            non_null_entries += 1;
+            current_entry.min = Some(*ping);
+            current_entry.max = Some(*ping);
+        }
         current_entry.pings.push(first.ping);
         for ping in iter {
             if last_host != ping.host {
@@ -192,11 +198,11 @@ impl Scanner {
         let stats = Version::find_by_statement(Statement::from_sql_and_values(
             DbBackend::Sqlite,
             r#"SELECT version FROM host h
-            JOIN update_check u ON u.host = h.id
+            JOIN health_check u ON u.host = h.id
             WHERE h.enabled = true AND u.time >= $1 AND version IS NOT NULL
             GROUP BY version
             ORDER BY version ASC"#,
-            [age.into()],
+            [age.timestamp().into()],
         ))
         .all(&self.inner.db)
         .await?;
@@ -219,11 +225,11 @@ impl Scanner {
     ) -> Result<Vec<HostStats>> {
         let stats: Vec<HostStats> = HostStats::find_by_statement(Statement::from_sql_and_values(
             DbBackend::Sqlite,
-            r#"SELECT u.host, COUNT(CASE WHEN healthy = true THEN 1 END) as good,COUNT(*) as total FROM update_check u
+            r#"SELECT u.host, COUNT(CASE WHEN healthy = true THEN 1 END) as good,COUNT(*) as total FROM health_check u
             JOIN host h ON h.id = u.host
             WHERE h.enabled = true AND u.time BETWEEN $1 AND $2
             GROUP BY u.host "#,
-            [from.into(), to.into()],
+            [from.timestamp().into(), to.timestamp().into()],
         ))
         .all(&self.inner.db)
         .await?;

scanner/src/instance_check.rs

@@ -3,7 +3,7 @@
 use std::time::Instant;
 
 use chrono::Utc;
-use entities::update_check;
+use entities::health_check;
 use entities::{host, prelude::*};
 use reqwest::Url;
 use sea_orm::prelude::DateTimeUtc;
@@ -35,7 +35,7 @@ impl Scanner {
             let scanner = self.clone();
             let muted_host = last_check.iter().find(|v|v.host == model.id).map_or(false,|check|!check.healthy);
             join_set.spawn(async move {
-                scanner.update_check_host(model, muted_host).await;
+                scanner.health_check_host(model, muted_host).await;
             });
         }
         // wait till all of them are finished, preventing DoS
@@ -49,14 +49,14 @@ impl Scanner {
     }
 
     #[instrument]
-    async fn update_check_host(&self, host: host::Model, muted: bool) {
+    async fn health_check_host(&self, host: host::Model, muted: bool) {
         let now = Utc::now();
         let mut url = match Url::parse(&host.url) {
             Err(e) => {
                 if !muted {
                     tracing::error!(error=?e, url=host.url,"failed to parse instance URL");
                 }
-                self.insert_failed_update_check(host.id, now, None, None)
+                self.insert_failed_health_check(host.id, now, None, None)
                     .await;
                 return;
             }
@@ -82,7 +82,7 @@ impl Scanner {
                     }
                     _ => (None, None),
                 };
-                self.insert_failed_update_check(host.id, now, resp_time, http_code)
+                self.insert_failed_health_check(host.id, now, resp_time, http_code)
                     .await;
             }
             Ok((http_code, content)) => {
@@ -97,7 +97,7 @@ impl Scanner {
                             "host doesn't contain expected profile content"
                         );
                     }
-                    self.insert_failed_update_check(
+                    self.insert_failed_health_check(
                         host.id,
                         now,
                         Some(took_ms as _),
@@ -107,8 +107,8 @@ impl Scanner {
                 }
 
                 // create successfull uptime entry
-                if let Err(e) = (update_check::ActiveModel {
-                    time: ActiveValue::Set(now),
+                if let Err(e) = (health_check::ActiveModel {
+                    time: ActiveValue::Set(now.timestamp()),
                     host: ActiveValue::Set(host.id),
                     resp_time: ActiveValue::Set(Some(took_ms as _)),
                     response_code: ActiveValue::Set(Some(http_code as _)),
@@ -168,15 +168,15 @@ impl Scanner {
         }
     }
 
-    async fn insert_failed_update_check(
+    async fn insert_failed_health_check(
         &self,
         host: i32,
         time: DateTimeUtc,
         resp_time: Option<i32>,
         http_code: Option<i32>,
     ) {
-        if let Err(e) = (update_check::ActiveModel {
-            time: ActiveValue::Set(time),
+        if let Err(e) = (health_check::ActiveModel {
+            time: ActiveValue::Set(time.timestamp()),
             host: ActiveValue::Set(host),
             resp_time: ActiveValue::Set(resp_time),
             healthy: ActiveValue::Set(false),

scanner/src/lib.rs

@@ -240,7 +240,7 @@ impl Scanner {
                 host::ActiveModel {
                     id: ActiveValue::Set(host.id),
                     enabled: ActiveValue::Set(false),
-                    updated: ActiveValue::Set(time),
+                    updated: ActiveValue::Set(time.timestamp()),
                     ..Default::default()
                 }
                 .update(&transaction)
@@ -283,7 +283,7 @@ impl Scanner {
                     enabled: ActiveValue::Set(true),
                     version: ActiveValue::Set(version),
                     rss: ActiveValue::Set(rss),
-                    updated: ActiveValue::Set(time),
+                    updated: ActiveValue::Set(time.timestamp()),
                 }
             });
         }
@@ -367,14 +367,14 @@ impl Scanner {
     }
 
     pub async fn query_latest_check<T: ConnectionTrait>(&self, connection: &T) -> Result<Vec<LatestCheck>> {
-        let update_checks = LatestCheck::find_by_statement(Statement::from_sql_and_values(
+        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 update_check u
+                SELECT u.host,MAX(u.time) as time FROM health_check u
                 GROUP BY u.host
             )
-            SELECT u.host,healthy,h.domain FROM update_check u
+            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
@@ -383,7 +383,7 @@ impl Scanner {
         ))
         .all(connection)
         .await?;
-        Ok(update_checks)
+        Ok(health_checks)
     }
 }
 

src/main.rs

@@ -29,7 +29,7 @@ async fn _main() -> miette::Result<()> {
         .with(tracing_subscriber::EnvFilter::new(
             var("RUST_LOG").unwrap_or_else(|_| {
                 format!(
-                    "warn,tower_http=debug,scanner=trace,server=debug,{}=debug",
+                    "warn,tower_http=debug,migration=debug,scanner=trace,server=debug,{}=debug",
                     env!("CARGO_PKG_NAME")
                 )
                 .into()