PR feedback

- remove active field from nym_nodes
- delete obsolete nym_nodes

nym-node-status-api/nym-node-status-api/migrations/005_node_self_described.sql

@@ -1,3 +1,61 @@
 ALTER TABLE nym_nodes ADD COLUMN self_described TEXT;
 ALTER TABLE nym_nodes ADD COLUMN bond_info TEXT;
-ALTER TABLE nym_nodes ADD COLUMN active INTEGER CHECK (active in (0, 1)) NOT NULL DEFAULT 0;
+
+-- # Why recreate tables?
+-- I need DELETE with CASCADE functionality, but ALTER TABLE doesn't support
+-- adding constraints (which CASCADE is). So I recreate tables with proper
+-- constraints and fill them with existing data.
+
+-- To avoid invalidating existing FK references, temporarily disable FK enforcement.
+PRAGMA foreign_keys=off;
+
+DROP INDEX IF EXISTS idx_nym_nodes_packet_stats_raw_node_id_timestamp_utc;
+
+ALTER TABLE nym_node_descriptions RENAME TO _nym_node_descriptions_old;
+ALTER TABLE nym_nodes_packet_stats_raw RENAME TO _nym_nodes_packet_stats_raw_old;
+ALTER TABLE nym_node_daily_mixing_stats RENAME TO _nym_node_daily_mixing_stats_old;
+
+CREATE TABLE nym_node_descriptions (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    node_id INTEGER UNIQUE NOT NULL,
+    moniker VARCHAR,
+    website VARCHAR,
+    security_contact VARCHAR,
+    details VARCHAR,
+    last_updated_utc INTEGER NOT NULL,
+    FOREIGN KEY (node_id) REFERENCES nym_nodes (node_id) ON DELETE CASCADE
+);
+
+CREATE TABLE nym_nodes_packet_stats_raw (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    node_id INTEGER NOT NULL,
+    timestamp_utc INTEGER NOT NULL,
+    packets_received INTEGER,
+    packets_sent INTEGER,
+    packets_dropped INTEGER,
+    FOREIGN KEY (node_id) REFERENCES nym_nodes (node_id) ON DELETE CASCADE
+);
+
+CREATE INDEX idx_nym_nodes_packet_stats_raw_node_id_timestamp_utc ON nym_nodes_packet_stats_raw (node_id, timestamp_utc);
+
+CREATE TABLE nym_node_daily_mixing_stats (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    node_id INTEGER NOT NULL,
+    total_stake BIGINT NOT NULL,
+    date_utc VARCHAR NOT NULL,
+    packets_received INTEGER DEFAULT 0,
+    packets_sent INTEGER DEFAULT 0,
+    packets_dropped INTEGER DEFAULT 0,
+    FOREIGN KEY (node_id) REFERENCES nym_nodes (node_id) ON DELETE CASCADE,
+    UNIQUE (node_id, date_utc) -- This constraint automatically creates an index
+);
+
+INSERT INTO nym_node_descriptions SELECT * FROM _nym_node_descriptions_old;
+INSERT INTO nym_nodes_packet_stats_raw SELECT * FROM _nym_nodes_packet_stats_raw_old;
+INSERT INTO nym_node_daily_mixing_stats SELECT * FROM _nym_node_daily_mixing_stats_old;
+
+DROP TABLE _nym_node_descriptions_old;
+DROP TABLE _nym_nodes_packet_stats_raw_old;
+DROP TABLE _nym_node_daily_mixing_stats_old;
+
+PRAGMA foreign_keys=on;

nym-node-status-api/nym-node-status-api/src/db/models.rs

@@ -435,7 +435,6 @@ pub(crate) struct NymNodeDto {
     pub performance: String,
     pub self_described: Option<serde_json::Value>,
     pub bond_info: Option<serde_json::Value>,
-    pub active: bool,
 }
 
 #[allow(dead_code)] // it's not dead code but clippy doesn't detect usage in sqlx macros
@@ -453,7 +452,6 @@ pub(crate) struct NymNodeInsertRecord {
     pub entry: Option<serde_json::Value>,
     pub self_described: Option<serde_json::Value>,
     pub bond_info: Option<serde_json::Value>,
-    pub active: bool,
     pub last_updated_utc: String,
 }
 
@@ -465,12 +463,6 @@ impl NymNodeInsertRecord {
     ) -> anyhow::Result<Self> {
         let now = OffsetDateTime::now_utc().to_string();
 
-        // if a node doesn't expose its self-described endpoint, it can't route traffic
-        // https://nym.com/docs/operators/nodes/nym-node/bonding
-        // same if it's not bonded in the mixnet smart contract
-        // https://nym.com/docs/operators/tokenomics/mixnet-rewards#rewarded-set-selection
-        let active = self_described.is_some() && bond_info.is_some();
-
         // if bond info is missing, set stake to 0
         let total_stake = bond_info
             .map(|info| decimal_to_i64(info.total_stake()))
@@ -492,7 +484,6 @@ impl NymNodeInsertRecord {
             entry,
             self_described,
             bond_info,
-            active,
             last_updated_utc: now,
         };
 

nym-node-status-api/nym-node-status-api/src/db/queries/mixnodes.rs

@@ -11,12 +11,13 @@ use crate::{
     http::models::{DailyStats, Mixnode},
 };
 
-pub(crate) async fn update_bonded_mixnodes(
+pub(crate) async fn update_mixnodes(
     pool: &DbPool,
     mixnodes: Vec<MixnodeRecord>,
 ) -> anyhow::Result<()> {
     let mut tx = pool.begin().await?;
 
+    // mark all as unbonded
     sqlx::query!(
         r#"UPDATE
             mixnodes
@@ -27,6 +28,7 @@ pub(crate) async fn update_bonded_mixnodes(
     .execute(&mut *tx)
     .await?;
 
+    // existing nodes get updated on insert
     for record in mixnodes.iter() {
         // https://www.sqlite.org/lang_upsert.html
         sqlx::query!(

nym-node-status-api/nym-node-status-api/src/db/queries/mod.rs

@@ -13,12 +13,10 @@ pub(crate) use gateways::{
 };
 pub(crate) use gateways_stats::{delete_old_records, get_sessions_stats, insert_session_records};
 pub(crate) use misc::insert_summaries;
-pub(crate) use mixnodes::{
-    get_all_mixnodes, get_bonded_mix_ids, get_daily_stats, update_bonded_mixnodes,
-};
+pub(crate) use mixnodes::{get_all_mixnodes, get_bonded_mix_ids, get_daily_stats, update_mixnodes};
 pub(crate) use nym_nodes::{
-    get_active_node_bond_info, get_active_node_descriptions, get_active_nym_nodes,
-    get_all_nym_nodes, update_nym_nodes,
+    get_active_nym_nodes, get_all_nym_nodes, get_described_node_bond_info, get_node_descriptions,
+    update_nym_nodes,
 };
 pub(crate) use packet_stats::{
     get_raw_node_stats, insert_daily_node_stats, insert_node_packet_stats,

nym-node-status-api/nym-node-status-api/src/db/queries/nym_nodes.rs

@@ -1,11 +1,10 @@
-use std::collections::HashMap;
-
 use anyhow::Context;
 use futures_util::TryStreamExt;
 use nym_validator_client::{
     client::{NodeId, NymNodeDetails},
     models::NymNodeDescription,
 };
+use std::collections::{HashMap, HashSet};
 use tracing::instrument;
 
 use crate::db::{
@@ -30,8 +29,7 @@ pub(crate) async fn get_all_nym_nodes(pool: &DbPool) -> anyhow::Result<Vec<NymNo
             entry as "entry: serde_json::Value",
             performance,
             self_described as "self_described: serde_json::Value",
-            bond_info as "bond_info: serde_json::Value",
-            active as "active: bool"
+            bond_info as "bond_info: serde_json::Value"
         FROM
             nym_nodes
         "#,
@@ -42,6 +40,11 @@ pub(crate) async fn get_all_nym_nodes(pool: &DbPool) -> anyhow::Result<Vec<NymNo
     .map_err(From::from)
 }
 
+/// if a node doesn't expose its self-described endpoint, it can't route traffic
+/// - https://nym.com/docs/operators/nodes/nym-node/bonding
+///
+/// same if it's not bonded in the mixnet smart contract
+/// - https://nym.com/docs/operators/tokenomics/mixnet-rewards#rewarded-set-selection
 pub(crate) async fn get_active_nym_nodes(pool: &DbPool) -> anyhow::Result<Vec<NymNodeDto>> {
     let mut conn = pool.acquire().await?;
 
@@ -59,12 +62,13 @@ pub(crate) async fn get_active_nym_nodes(pool: &DbPool) -> anyhow::Result<Vec<Ny
             entry as "entry: serde_json::Value",
             performance,
             self_described as "self_described: serde_json::Value",
-            bond_info as "bond_info: serde_json::Value",
-            active as "active: bool"
+            bond_info as "bond_info: serde_json::Value"
         FROM
             nym_nodes
         WHERE
-            active = true
+            self_described IS NOT NULL
+        AND
+            bond_info IS NOT NULL
         "#,
     )
     .fetch(&mut *conn)
@@ -73,23 +77,24 @@ pub(crate) async fn get_active_nym_nodes(pool: &DbPool) -> anyhow::Result<Vec<Ny
     .map_err(From::from)
 }
 
-#[instrument(level = "debug", skip_all)]
+#[instrument(level = "debug", skip_all, fields(node_records=node_records.len()))]
 pub(crate) async fn update_nym_nodes(
     pool: &DbPool,
     node_records: Vec<NymNodeInsertRecord>,
 ) -> anyhow::Result<usize> {
     let mut tx = pool.begin().await?;
 
-    // set inactive all nodes
-    sqlx::query!(
-        r#"UPDATE
+    let mut nodes_to_delete = sqlx::query!(
+        r#"SELECT
+            node_id as "node_id!: i64"
+        FROM
             nym_nodes
-        SET
-            active = false
         "#,
     )
-    .execute(&mut *tx)
-    .await?;
+    .fetch_all(&mut *tx)
+    .await
+    .map(|records| records.into_iter().map(|record| record.node_id as NodeId))?
+    .collect::<HashSet<NodeId>>();
 
     // active nodes will get updated on insert
     let inserted = node_records.len();
@@ -104,10 +109,9 @@ pub(crate) async fn update_nym_nodes(
                     supported_roles, entry,
                     self_described,
                     bond_info,
-                    active,
                     performance, last_updated_utc
                 )
-                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                 ON CONFLICT(node_id) DO UPDATE SET
                 ed25519_identity_pubkey=excluded.ed25519_identity_pubkey,
                 ip_addresses=excluded.ip_addresses,
@@ -118,7 +122,6 @@ pub(crate) async fn update_nym_nodes(
                 entry=excluded.entry,
                 self_described=excluded.self_described,
                 bond_info=excluded.bond_info,
-                active=excluded.active,
                 performance=excluded.performance,
                 last_updated_utc=excluded.last_updated_utc
                 ;",
@@ -133,13 +136,31 @@ pub(crate) async fn update_nym_nodes(
             record.entry,
             record.self_described,
             record.bond_info,
-            record.active,
             record.performance,
             record.last_updated_utc,
         )
         .execute(&mut *tx)
         .await
-        .with_context(|| format!("node_id={}", record.node_id))?;
+        .with_context(|| format!("Failed to INSERT node_id={}", record.node_id))?;
+
+        // if node was updated, remove it from the list
+        nodes_to_delete.remove(&(record.node_id as NodeId));
+    }
+
+    if !nodes_to_delete.is_empty() {
+        tracing::debug!("DELETE {} obsolete nodes", nodes_to_delete.len());
+    }
+
+    // clean up leftover nodes, which weren't inserted/updated
+    for node_id in nodes_to_delete {
+        sqlx::query!(
+            "DELETE FROM nym_nodes
+            WHERE node_id = ?",
+            node_id,
+        )
+        .execute(&mut *tx)
+        .await
+        .map_err(|e| anyhow::anyhow!("Failed to DELETE node_id={}: {}", node_id, e))?;
     }
 
     tx.commit().await?;
@@ -147,7 +168,7 @@ pub(crate) async fn update_nym_nodes(
     Ok(inserted)
 }
 
-pub(crate) async fn get_active_node_bond_info(
+pub(crate) async fn get_described_node_bond_info(
     pool: &DbPool,
 ) -> anyhow::Result<HashMap<NodeId, NymNodeDetails>> {
     let mut conn = pool.acquire().await?;
@@ -161,7 +182,7 @@ pub(crate) async fn get_active_node_bond_info(
         WHERE
             bond_info IS NOT NULL
         AND
-            active = true
+            self_described IS NOT NULL
         "#,
     )
     .fetch_all(&mut *conn)
@@ -181,7 +202,7 @@ pub(crate) async fn get_active_node_bond_info(
     .map_err(From::from)
 }
 
-pub(crate) async fn get_active_node_descriptions(
+pub(crate) async fn get_node_descriptions(
     pool: &DbPool,
 ) -> anyhow::Result<HashMap<NodeId, NymNodeDescription>> {
     let mut conn = pool.acquire().await?;
@@ -194,8 +215,6 @@ pub(crate) async fn get_active_node_descriptions(
             nym_nodes
         WHERE
             self_described IS NOT NULL
-        AND
-            active = true
         "#,
     )
     .fetch_all(&mut *conn)

nym-node-status-api/nym-node-status-api/src/http/state.rs

@@ -347,8 +347,8 @@ impl HttpCache {
 
 #[instrument(level = "info", skip_all)]
 async fn aggregate_node_info_from_db(pool: &DbPool) -> anyhow::Result<Vec<ExtendedNymNode>> {
-    let node_bond_info = queries::get_active_node_bond_info(pool).await?;
-    tracing::debug!("Active nodes with bond info: {}", node_bond_info.len());
+    let node_bond_info = queries::get_described_node_bond_info(pool).await?;
+    tracing::debug!("Described nodes with bond info: {}", node_bond_info.len());
 
     let skimmed_nodes = queries::get_all_nym_nodes(pool).await.map(|records| {
         records
@@ -359,8 +359,8 @@ async fn aggregate_node_info_from_db(pool: &DbPool) -> anyhow::Result<Vec<Extend
     })?;
     tracing::debug!("Skimmed nodes: {}", skimmed_nodes.len());
 
-    let described_nodes = queries::get_active_node_descriptions(pool).await?;
-    tracing::debug!("Active described nodes: {}", described_nodes.len());
+    let described_nodes = queries::get_node_descriptions(pool).await?;
+    tracing::debug!("Described nodes: {}", described_nodes.len());
 
     let mut parsed_nym_nodes = Vec::new();
     for (node_id, described_node) in described_nodes {

nym-node-status-api/nym-node-status-api/src/monitor/mod.rs

@@ -237,7 +237,7 @@ impl Monitor {
             delegation_program_members,
         )?;
         let mixnodes_count = mixnode_records.len();
-        queries::update_bonded_mixnodes(&pool, mixnode_records)
+        queries::update_mixnodes(&pool, mixnode_records)
             .await
             .map(|_| {
                 tracing::debug!("{} mixnode info written to DB!", mixnodes_count);