diff --git a/Cargo.lock b/Cargo.lock index 6b3403c9b0..450fd53c06 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -134,6 +134,19 @@ version = "0.2.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5c6cb57a04249c6480766f7f7cef5467412af1490f8d1e243141daddada3264f" +[[package]] +name = "ammonia" +version = "4.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ab99eae5ee58501ab236beb6f20f6ca39be615267b014899c89b2f0bc18a459" +dependencies = [ + "html5ever", + "maplit", + "once_cell", + "tendril", + "url", +] + [[package]] name = "android-tzdata" version = "0.1.1" @@ -2595,6 +2608,16 @@ version = "2.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6d5a32815ae3f33302d95fdcb2ce17862f8c65363dcfd29360480ba1001fc9c" +[[package]] +name = "futf" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df420e2e84819663797d1ec6544b13c5be84629e7bb00dc960d6917db2987843" +dependencies = [ + "mac", + "new_debug_unreachable", +] + [[package]] name = "futures" version = "0.3.31" @@ -3095,6 +3118,20 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "html5ever" +version = "0.27.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c13771afe0e6e846f1e67d038d4cb29998a6779f93c809212e4e9c32efd244d4" +dependencies = [ + "log", + "mac", + "markup5ever", + "proc-macro2", + "quote", + "syn 2.0.90", +] + [[package]] name = "http" version = "0.2.12" @@ -3985,6 +4022,12 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "mac" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4" + [[package]] name = "macroific" version = "1.3.1" @@ -4032,6 +4075,26 @@ dependencies = [ "syn 2.0.90", ] +[[package]] +name = "maplit" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3e2e65a1a2e43cfcb47a895c4c8b10d1f4a61097f9f254f183aee60cad9c651d" + +[[package]] +name = "markup5ever" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "16ce3abbeba692c8b8441d036ef91aea6df8da2c6b6e21c7e14d3c18e526be45" +dependencies = [ + "log", + "phf", + "phf_codegen", + "string_cache", + "string_cache_codegen", + "tendril", +] + [[package]] name = "matchers" version = "0.1.0" @@ -4315,6 +4378,12 @@ dependencies = [ "log", ] +[[package]] +name = "new_debug_unreachable" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "650eef8c711430f1a879fdd01d4745a7deea475becfb90269c06775983bbf086" + [[package]] name = "newline-converter" version = "0.2.2" @@ -6070,6 +6139,7 @@ dependencies = [ name = "nym-node-status-api" version = "1.0.0-rc.7" dependencies = [ + "ammonia", "anyhow", "axum 0.7.7", "chrono", @@ -7379,7 +7449,46 @@ version = "0.11.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ade2d8b8f33c7333b51bcf0428d37e217e9f32192ae4772156f65063b8ce03dc" dependencies = [ - "phf_shared", + "phf_shared 0.11.2", +] + +[[package]] +name = "phf_codegen" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aef8048c789fa5e851558d709946d6d79a8ff88c0440c587967f8e94bfb1216a" +dependencies = [ + "phf_generator 0.11.3", + "phf_shared 0.11.2", +] + +[[package]] +name = "phf_generator" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d5285893bb5eb82e6aaf5d59ee909a06a16737a8970984dd7746ba9283498d6" +dependencies = [ + "phf_shared 0.10.0", + "rand", +] + +[[package]] +name = "phf_generator" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3c80231409c20246a13fddb31776fb942c38553c51e871f8cbd687a4cfb5843d" +dependencies = [ + "phf_shared 0.11.2", + "rand", +] + +[[package]] +name = "phf_shared" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6796ad771acdc0123d2a88dc428b5e38ef24456743ddb1744ed628f9815c096" +dependencies = [ + "siphasher 0.3.11", ] [[package]] @@ -7570,6 +7679,12 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5b40af805b3121feab8a3c29f04d8ad262fa8e0561883e7653e024ae4479e6de" +[[package]] +name = "precomputed-hash" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "925383efa346730478fb4838dbe9137d2a47675ad789c546d150a6e1dd4ab31c" + [[package]] name = "pretty_assertions" version = "1.4.1" @@ -9297,6 +9412,32 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" +[[package]] +name = "string_cache" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f91138e76242f575eb1d3b38b4f1362f10d3a43f47d182a5b359af488a02293b" +dependencies = [ + "new_debug_unreachable", + "once_cell", + "parking_lot", + "phf_shared 0.10.0", + "precomputed-hash", + "serde", +] + +[[package]] +name = "string_cache_codegen" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6bb30289b722be4ff74a408c3cc27edeaad656e06cb1fe8fa9231fa59c728988" +dependencies = [ + "phf_generator 0.10.0", + "phf_shared 0.10.0", + "proc-macro2", + "quote", +] + [[package]] name = "stringprep" version = "0.1.5" @@ -9653,6 +9794,17 @@ dependencies = [ "walkdir", ] +[[package]] +name = "tendril" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d24a120c5fc464a3458240ee02c299ebcb9d67b5249c8848b09d639dca8d7bb0" +dependencies = [ + "futf", + "mac", + "utf-8", +] + [[package]] name = "termcolor" version = "1.4.1" diff --git a/nym-node-status-api/nym-node-status-api/.sqlx/query-06065394c157927e4002ddd5c7c1af626ae15728d615f539470cd7c189312385.json b/nym-node-status-api/nym-node-status-api/.sqlx/query-06065394c157927e4002ddd5c7c1af626ae15728d615f539470cd7c189312385.json new file mode 100644 index 0000000000..205bfdc79b --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/.sqlx/query-06065394c157927e4002ddd5c7c1af626ae15728d615f539470cd7c189312385.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "\n INSERT INTO mixnode_description (\n mix_id, moniker, website, security_contact, details, last_updated_utc\n ) VALUES (?, ?, ?, ?, ?, ?)\n ON CONFLICT (mix_id) DO UPDATE SET\n moniker = excluded.moniker,\n website = excluded.website,\n security_contact = excluded.security_contact,\n details = excluded.details,\n last_updated_utc = excluded.last_updated_utc\n ", + "describe": { + "columns": [], + "parameters": { + "Right": 6 + }, + "nullable": [] + }, + "hash": "06065394c157927e4002ddd5c7c1af626ae15728d615f539470cd7c189312385" +} diff --git a/nym-node-status-api/nym-node-status-api/.sqlx/query-f5048d9926a5f5329f7f3b96d43b925e033ceec4f8112258feb4ac9e96fc5924.json b/nym-node-status-api/nym-node-status-api/.sqlx/query-0cf0e4d4f30e90caecffd6255ef85dab12730e538be194438f19ed7f198bd50e.json similarity index 66% rename from nym-node-status-api/nym-node-status-api/.sqlx/query-f5048d9926a5f5329f7f3b96d43b925e033ceec4f8112258feb4ac9e96fc5924.json rename to nym-node-status-api/nym-node-status-api/.sqlx/query-0cf0e4d4f30e90caecffd6255ef85dab12730e538be194438f19ed7f198bd50e.json index 06d098bebc..fda05c400a 100644 --- a/nym-node-status-api/nym-node-status-api/.sqlx/query-f5048d9926a5f5329f7f3b96d43b925e033ceec4f8112258feb4ac9e96fc5924.json +++ b/nym-node-status-api/nym-node-status-api/.sqlx/query-0cf0e4d4f30e90caecffd6255ef85dab12730e538be194438f19ed7f198bd50e.json @@ -1,6 +1,6 @@ { "db_name": "SQLite", - "query": "UPDATE\n testruns\n SET\n status = ?\n WHERE\n status = ?\n AND\n timestamp_utc < ?\n ", + "query": "UPDATE\n testruns\n SET\n status = ?\n WHERE\n status = ?\n AND\n last_assigned_utc < ?\n ", "describe": { "columns": [], "parameters": { @@ -8,5 +8,5 @@ }, "nullable": [] }, - "hash": "f5048d9926a5f5329f7f3b96d43b925e033ceec4f8112258feb4ac9e96fc5924" + "hash": "0cf0e4d4f30e90caecffd6255ef85dab12730e538be194438f19ed7f198bd50e" } diff --git a/nym-node-status-api/nym-node-status-api/.sqlx/query-3243cf5646255a9430d1e6710970505d0dbcc62703f40e090e80ff48c77723c4.json b/nym-node-status-api/nym-node-status-api/.sqlx/query-3243cf5646255a9430d1e6710970505d0dbcc62703f40e090e80ff48c77723c4.json new file mode 100644 index 0000000000..7aeb02d48b --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/.sqlx/query-3243cf5646255a9430d1e6710970505d0dbcc62703f40e090e80ff48c77723c4.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "INSERT OR IGNORE INTO gateway_session_stats\n (gateway_identity_key, node_id, day,\n unique_active_clients, session_started, users_hashes,\n vpn_sessions, mixnet_sessions, unknown_sessions)\n VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + "describe": { + "columns": [], + "parameters": { + "Right": 9 + }, + "nullable": [] + }, + "hash": "3243cf5646255a9430d1e6710970505d0dbcc62703f40e090e80ff48c77723c4" +} diff --git a/nym-node-status-api/nym-node-status-api/.sqlx/query-3d5fc502f976f5081f01352856b8632c29c81bfafb043bb8744129cf9e0266ad.json b/nym-node-status-api/nym-node-status-api/.sqlx/query-3d5fc502f976f5081f01352856b8632c29c81bfafb043bb8744129cf9e0266ad.json index 2a834fe87e..e8725a21d1 100644 --- a/nym-node-status-api/nym-node-status-api/.sqlx/query-3d5fc502f976f5081f01352856b8632c29c81bfafb043bb8744129cf9e0266ad.json +++ b/nym-node-status-api/nym-node-status-api/.sqlx/query-3d5fc502f976f5081f01352856b8632c29c81bfafb043bb8744129cf9e0266ad.json @@ -30,7 +30,7 @@ "nullable": [ true, false, - true, + false, true ] }, diff --git a/nym-node-status-api/nym-node-status-api/.sqlx/query-3c584e211d07c511644c8079187965acf3bcfb3f84ba8d24ed645d79976cf784.json b/nym-node-status-api/nym-node-status-api/.sqlx/query-4d865e873c9cb133883da94db72dcdebd4969e1f240def9fb1bf946f4a1f342f.json similarity index 61% rename from nym-node-status-api/nym-node-status-api/.sqlx/query-3c584e211d07c511644c8079187965acf3bcfb3f84ba8d24ed645d79976cf784.json rename to nym-node-status-api/nym-node-status-api/.sqlx/query-4d865e873c9cb133883da94db72dcdebd4969e1f240def9fb1bf946f4a1f342f.json index 56898ca5ad..08b9e93fff 100644 --- a/nym-node-status-api/nym-node-status-api/.sqlx/query-3c584e211d07c511644c8079187965acf3bcfb3f84ba8d24ed645d79976cf784.json +++ b/nym-node-status-api/nym-node-status-api/.sqlx/query-4d865e873c9cb133883da94db72dcdebd4969e1f240def9fb1bf946f4a1f342f.json @@ -1,6 +1,6 @@ { "db_name": "SQLite", - "query": "INSERT INTO testruns (gateway_id, status, ip_address, timestamp_utc, log) VALUES (?, ?, ?, ?, ?)", + "query": "INSERT INTO testruns (gateway_id, status, ip_address, created_utc, log) VALUES (?, ?, ?, ?, ?)", "describe": { "columns": [], "parameters": { @@ -8,5 +8,5 @@ }, "nullable": [] }, - "hash": "3c584e211d07c511644c8079187965acf3bcfb3f84ba8d24ed645d79976cf784" + "hash": "4d865e873c9cb133883da94db72dcdebd4969e1f240def9fb1bf946f4a1f342f" } diff --git a/nym-node-status-api/nym-node-status-api/.sqlx/query-5e9cbb39f5fb53774803270f422989e199aac4d4a71913c7074359b4bd676b02.json b/nym-node-status-api/nym-node-status-api/.sqlx/query-5e9cbb39f5fb53774803270f422989e199aac4d4a71913c7074359b4bd676b02.json new file mode 100644 index 0000000000..e665d23924 --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/.sqlx/query-5e9cbb39f5fb53774803270f422989e199aac4d4a71913c7074359b4bd676b02.json @@ -0,0 +1,26 @@ +{ + "db_name": "SQLite", + "query": "UPDATE testruns\n SET\n status = ?,\n last_assigned_utc = ?\n WHERE rowid =\n (\n SELECT rowid\n FROM testruns\n WHERE status = ?\n ORDER BY created_utc asc\n LIMIT 1\n )\n RETURNING\n id as \"id!\",\n gateway_id\n ", + "describe": { + "columns": [ + { + "name": "id!", + "ordinal": 0, + "type_info": "Int64" + }, + { + "name": "gateway_id", + "ordinal": 1, + "type_info": "Int64" + } + ], + "parameters": { + "Right": 3 + }, + "nullable": [ + true, + false + ] + }, + "hash": "5e9cbb39f5fb53774803270f422989e199aac4d4a71913c7074359b4bd676b02" +} diff --git a/nym-node-status-api/nym-node-status-api/.sqlx/query-71a455c705f9c25d3843ff2fb8629d1320a5eb10797cdb5a435455d22c6aeac1.json b/nym-node-status-api/nym-node-status-api/.sqlx/query-71a455c705f9c25d3843ff2fb8629d1320a5eb10797cdb5a435455d22c6aeac1.json index f061747404..983f04a451 100644 --- a/nym-node-status-api/nym-node-status-api/.sqlx/query-71a455c705f9c25d3843ff2fb8629d1320a5eb10797cdb5a435455d22c6aeac1.json +++ b/nym-node-status-api/nym-node-status-api/.sqlx/query-71a455c705f9c25d3843ff2fb8629d1320a5eb10797cdb5a435455d22c6aeac1.json @@ -82,7 +82,7 @@ false, false, false, - true, + false, true, true, true, diff --git a/nym-node-status-api/nym-node-status-api/.sqlx/query-7b3e44e1dea0568d331017216e5375684086bbeb80f3b251adc761bae0dafe92.json b/nym-node-status-api/nym-node-status-api/.sqlx/query-7b3e44e1dea0568d331017216e5375684086bbeb80f3b251adc761bae0dafe92.json new file mode 100644 index 0000000000..61863d2d8a --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/.sqlx/query-7b3e44e1dea0568d331017216e5375684086bbeb80f3b251adc761bae0dafe92.json @@ -0,0 +1,32 @@ +{ + "db_name": "SQLite", + "query": "\n SELECT packets_received, packets_sent, packets_dropped\n FROM mixnode_packet_stats_raw\n WHERE mix_id = ?\n ORDER BY timestamp_utc DESC\n LIMIT 1 OFFSET 1\n ", + "describe": { + "columns": [ + { + "name": "packets_received", + "ordinal": 0, + "type_info": "Int64" + }, + { + "name": "packets_sent", + "ordinal": 1, + "type_info": "Int64" + }, + { + "name": "packets_dropped", + "ordinal": 2, + "type_info": "Int64" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + true, + true, + true + ] + }, + "hash": "7b3e44e1dea0568d331017216e5375684086bbeb80f3b251adc761bae0dafe92" +} diff --git a/nym-node-status-api/nym-node-status-api/.sqlx/query-84e784b7ceb479f37fa312dd9a8bb0b3debc04cd251cdfaa438bce7aca7c5992.json b/nym-node-status-api/nym-node-status-api/.sqlx/query-84e784b7ceb479f37fa312dd9a8bb0b3debc04cd251cdfaa438bce7aca7c5992.json new file mode 100644 index 0000000000..34f356d1b6 --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/.sqlx/query-84e784b7ceb479f37fa312dd9a8bb0b3debc04cd251cdfaa438bce7aca7c5992.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "\n INSERT INTO mixnode_packet_stats_raw (\n mix_id, timestamp_utc, packets_received, packets_sent, packets_dropped\n ) VALUES (?, ?, ?, ?, ?)\n ", + "describe": { + "columns": [], + "parameters": { + "Right": 5 + }, + "nullable": [] + }, + "hash": "84e784b7ceb479f37fa312dd9a8bb0b3debc04cd251cdfaa438bce7aca7c5992" +} diff --git a/nym-node-status-api/nym-node-status-api/.sqlx/query-8c6c1c67df2d06da74825872f448d30505801b56506668499f9da718f1d6b8f8.json b/nym-node-status-api/nym-node-status-api/.sqlx/query-8c6c1c67df2d06da74825872f448d30505801b56506668499f9da718f1d6b8f8.json new file mode 100644 index 0000000000..8ce60c2b8c --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/.sqlx/query-8c6c1c67df2d06da74825872f448d30505801b56506668499f9da718f1d6b8f8.json @@ -0,0 +1,32 @@ +{ + "db_name": "SQLite", + "query": "\n SELECT mix_id as node_id, host, http_api_port \n FROM mixnodes \n WHERE bonded = true\n ", + "describe": { + "columns": [ + { + "name": "node_id", + "ordinal": 0, + "type_info": "Int64" + }, + { + "name": "host", + "ordinal": 1, + "type_info": "Text" + }, + { + "name": "http_api_port", + "ordinal": 2, + "type_info": "Int64" + } + ], + "parameters": { + "Right": 0 + }, + "nullable": [ + false, + false, + false + ] + }, + "hash": "8c6c1c67df2d06da74825872f448d30505801b56506668499f9da718f1d6b8f8" +} diff --git a/nym-node-status-api/nym-node-status-api/.sqlx/query-969acbff4b128d3a3dc998bb5db1a4c8b4c106767e0276435dc23fcfd6acfe4f.json b/nym-node-status-api/nym-node-status-api/.sqlx/query-969acbff4b128d3a3dc998bb5db1a4c8b4c106767e0276435dc23fcfd6acfe4f.json new file mode 100644 index 0000000000..007f4ecb4f --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/.sqlx/query-969acbff4b128d3a3dc998bb5db1a4c8b4c106767e0276435dc23fcfd6acfe4f.json @@ -0,0 +1,20 @@ +{ + "db_name": "SQLite", + "query": "\n SELECT\n total_stake\n FROM mixnodes\n WHERE mix_id = ?\n ", + "describe": { + "columns": [ + { + "name": "total_stake", + "ordinal": 0, + "type_info": "Int64" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + false + ] + }, + "hash": "969acbff4b128d3a3dc998bb5db1a4c8b4c106767e0276435dc23fcfd6acfe4f" +} diff --git a/nym-node-status-api/nym-node-status-api/.sqlx/query-a14d5767512bc88ac47b667c47a773b548ccb20bc539839766080d8455439339.json b/nym-node-status-api/nym-node-status-api/.sqlx/query-a14d5767512bc88ac47b667c47a773b548ccb20bc539839766080d8455439339.json new file mode 100644 index 0000000000..d729065fcb --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/.sqlx/query-a14d5767512bc88ac47b667c47a773b548ccb20bc539839766080d8455439339.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "\n INSERT INTO mixnode_daily_stats (\n mix_id, date_utc, total_stake, packets_received, packets_sent, packets_dropped\n ) VALUES (?, ?, ?, ?, ?, ?)\n ON CONFLICT(mix_id, date_utc) DO UPDATE SET\n total_stake = excluded.total_stake,\n packets_received = mixnode_daily_stats.packets_received + excluded.packets_received,\n packets_sent = mixnode_daily_stats.packets_sent + excluded.packets_sent,\n packets_dropped = mixnode_daily_stats.packets_dropped + excluded.packets_dropped\n ", + "describe": { + "columns": [], + "parameters": { + "Right": 6 + }, + "nullable": [] + }, + "hash": "a14d5767512bc88ac47b667c47a773b548ccb20bc539839766080d8455439339" +} diff --git a/nym-node-status-api/nym-node-status-api/.sqlx/query-46d76bc6d3fba2dae3b21511a36289dd776749dd7a20cda61b0480f2fba60889.json b/nym-node-status-api/nym-node-status-api/.sqlx/query-a1d9eb816acd1a91ed0975c801c9295c01a78861a2a0597ad28b1579a14bf008.json similarity index 63% rename from nym-node-status-api/nym-node-status-api/.sqlx/query-46d76bc6d3fba2dae3b21511a36289dd776749dd7a20cda61b0480f2fba60889.json rename to nym-node-status-api/nym-node-status-api/.sqlx/query-a1d9eb816acd1a91ed0975c801c9295c01a78861a2a0597ad28b1579a14bf008.json index 2cb8760009..b5cc3e88fa 100644 --- a/nym-node-status-api/nym-node-status-api/.sqlx/query-46d76bc6d3fba2dae3b21511a36289dd776749dd7a20cda61b0480f2fba60889.json +++ b/nym-node-status-api/nym-node-status-api/.sqlx/query-a1d9eb816acd1a91ed0975c801c9295c01a78861a2a0597ad28b1579a14bf008.json @@ -1,6 +1,6 @@ { "db_name": "SQLite", - "query": "SELECT\n id as \"id!\",\n gateway_id as \"gateway_id!\",\n status as \"status!\",\n timestamp_utc as \"timestamp_utc!\",\n ip_address as \"ip_address!\",\n log as \"log!\"\n FROM testruns\n WHERE gateway_id = ? AND status != 2\n ORDER BY id DESC\n LIMIT 1", + "query": "SELECT\n id as \"id!\",\n gateway_id as \"gateway_id!\",\n status as \"status!\",\n created_utc as \"created_utc!\",\n ip_address as \"ip_address!\",\n log as \"log!\",\n last_assigned_utc\n FROM testruns\n WHERE gateway_id = ? AND status != 2\n ORDER BY id DESC\n LIMIT 1", "describe": { "columns": [ { @@ -19,7 +19,7 @@ "type_info": "Int64" }, { - "name": "timestamp_utc!", + "name": "created_utc!", "ordinal": 3, "type_info": "Int64" }, @@ -32,6 +32,11 @@ "name": "log!", "ordinal": 5, "type_info": "Text" + }, + { + "name": "last_assigned_utc", + "ordinal": 6, + "type_info": "Int64" } ], "parameters": { @@ -43,8 +48,9 @@ false, false, false, - false + false, + true ] }, - "hash": "46d76bc6d3fba2dae3b21511a36289dd776749dd7a20cda61b0480f2fba60889" + "hash": "a1d9eb816acd1a91ed0975c801c9295c01a78861a2a0597ad28b1579a14bf008" } diff --git a/nym-node-status-api/nym-node-status-api/.sqlx/query-a79a61b87325f3f1d9c5a4fb386ccd585be0641e5878acb6283b879f22ed2b4c.json b/nym-node-status-api/nym-node-status-api/.sqlx/query-a79a61b87325f3f1d9c5a4fb386ccd585be0641e5878acb6283b879f22ed2b4c.json new file mode 100644 index 0000000000..37dd4ac669 --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/.sqlx/query-a79a61b87325f3f1d9c5a4fb386ccd585be0641e5878acb6283b879f22ed2b4c.json @@ -0,0 +1,20 @@ +{ + "db_name": "SQLite", + "query": "SELECT\n COUNT(id) as \"count: i64\"\n FROM testruns\n WHERE\n status = ?\n ", + "describe": { + "columns": [ + { + "name": "count: i64", + "ordinal": 0, + "type_info": "Int" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + false + ] + }, + "hash": "a79a61b87325f3f1d9c5a4fb386ccd585be0641e5878acb6283b879f22ed2b4c" +} diff --git a/nym-node-status-api/nym-node-status-api/.sqlx/query-c9e61180ec35dfab8623333fafa3b216b93440d0fddc0a37dd1b6c1813741f6a.json b/nym-node-status-api/nym-node-status-api/.sqlx/query-c9e61180ec35dfab8623333fafa3b216b93440d0fddc0a37dd1b6c1813741f6a.json new file mode 100644 index 0000000000..0d313098b6 --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/.sqlx/query-c9e61180ec35dfab8623333fafa3b216b93440d0fddc0a37dd1b6c1813741f6a.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "DELETE FROM gateway_session_stats WHERE day <= ?", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "c9e61180ec35dfab8623333fafa3b216b93440d0fddc0a37dd1b6c1813741f6a" +} diff --git a/nym-node-status-api/nym-node-status-api/.sqlx/query-6d7967b831b355d5f2c77950abc56f816956b0824c66a25da611dce688105d36.json b/nym-node-status-api/nym-node-status-api/.sqlx/query-f0c64794cbaed87a1d3166251d8e6720c9a9fc929144188460849be85d915004.json similarity index 61% rename from nym-node-status-api/nym-node-status-api/.sqlx/query-6d7967b831b355d5f2c77950abc56f816956b0824c66a25da611dce688105d36.json rename to nym-node-status-api/nym-node-status-api/.sqlx/query-f0c64794cbaed87a1d3166251d8e6720c9a9fc929144188460849be85d915004.json index e12434a85a..cf1df0017b 100644 --- a/nym-node-status-api/nym-node-status-api/.sqlx/query-6d7967b831b355d5f2c77950abc56f816956b0824c66a25da611dce688105d36.json +++ b/nym-node-status-api/nym-node-status-api/.sqlx/query-f0c64794cbaed87a1d3166251d8e6720c9a9fc929144188460849be85d915004.json @@ -1,6 +1,6 @@ { "db_name": "SQLite", - "query": "SELECT\n id as \"id!\",\n gateway_id as \"gateway_id!\",\n status as \"status!\",\n timestamp_utc as \"timestamp_utc!\",\n ip_address as \"ip_address!\",\n log as \"log!\"\n FROM testruns\n WHERE\n id = ?\n AND\n status = ?\n ORDER BY timestamp_utc", + "query": "SELECT\n id as \"id!\",\n gateway_id as \"gateway_id!\",\n status as \"status!\",\n created_utc as \"created_utc!\",\n ip_address as \"ip_address!\",\n log as \"log!\",\n last_assigned_utc\n FROM testruns\n WHERE\n id = ?\n AND\n status = ?\n ORDER BY created_utc\n LIMIT 1", "describe": { "columns": [ { @@ -19,7 +19,7 @@ "type_info": "Int64" }, { - "name": "timestamp_utc!", + "name": "created_utc!", "ordinal": 3, "type_info": "Int64" }, @@ -32,6 +32,11 @@ "name": "log!", "ordinal": 5, "type_info": "Text" + }, + { + "name": "last_assigned_utc", + "ordinal": 6, + "type_info": "Int64" } ], "parameters": { @@ -43,8 +48,9 @@ false, false, false, - false + false, + true ] }, - "hash": "6d7967b831b355d5f2c77950abc56f816956b0824c66a25da611dce688105d36" + "hash": "f0c64794cbaed87a1d3166251d8e6720c9a9fc929144188460849be85d915004" } diff --git a/nym-node-status-api/nym-node-status-api/.sqlx/query-ff9334ba7b670b218b2f9100e9ab5d2f2d08b2e53203aab9f07ea9b52acbd407.json b/nym-node-status-api/nym-node-status-api/.sqlx/query-ff9334ba7b670b218b2f9100e9ab5d2f2d08b2e53203aab9f07ea9b52acbd407.json deleted file mode 100644 index 24e735e96d..0000000000 --- a/nym-node-status-api/nym-node-status-api/.sqlx/query-ff9334ba7b670b218b2f9100e9ab5d2f2d08b2e53203aab9f07ea9b52acbd407.json +++ /dev/null @@ -1,26 +0,0 @@ -{ - "db_name": "SQLite", - "query": "UPDATE testruns\n SET status = ?\n WHERE rowid =\n (\n SELECT rowid\n FROM testruns\n WHERE status = ?\n ORDER BY timestamp_utc asc\n LIMIT 1\n )\n RETURNING\n id as \"id!\",\n gateway_id\n ", - "describe": { - "columns": [ - { - "name": "id!", - "ordinal": 0, - "type_info": "Int64" - }, - { - "name": "gateway_id", - "ordinal": 1, - "type_info": "Int64" - } - ], - "parameters": { - "Right": 2 - }, - "nullable": [ - true, - false - ] - }, - "hash": "ff9334ba7b670b218b2f9100e9ab5d2f2d08b2e53203aab9f07ea9b52acbd407" -} diff --git a/nym-node-status-api/nym-node-status-api/Cargo.toml b/nym-node-status-api/nym-node-status-api/Cargo.toml index 98684f75f5..81460981ec 100644 --- a/nym-node-status-api/nym-node-status-api/Cargo.toml +++ b/nym-node-status-api/nym-node-status-api/Cargo.toml @@ -13,6 +13,7 @@ license.workspace = true rust-version.workspace = true [dependencies] +ammonia = "4" anyhow = { workspace = true } axum = { workspace = true, features = ["tokio", "macros"] } chrono = { workspace = true } diff --git a/nym-node-status-api/nym-node-status-api/migrations/003_scraper_tables.sql b/nym-node-status-api/nym-node-status-api/migrations/003_scraper_tables.sql new file mode 100644 index 0000000000..9d9b5842b1 --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/migrations/003_scraper_tables.sql @@ -0,0 +1,11 @@ +CREATE TABLE mixnode_packet_stats_raw ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + mix_id INTEGER NOT NULL, + timestamp_utc INTEGER NOT NULL, + packets_received INTEGER, + packets_sent INTEGER, + packets_dropped INTEGER, + FOREIGN KEY (mix_id) REFERENCES mixnodes (mix_id) + ); + +CREATE INDEX idx_mixnode_packet_stats_raw_mix_id_timestamp_utc ON mixnode_packet_stats_raw (mix_id, timestamp_utc); \ No newline at end of file diff --git a/nym-node-status-api/nym-node-status-api/src/db/models.rs b/nym-node-status-api/nym-node-status-api/src/db/models.rs index 9de09271bf..7901b0f13c 100644 --- a/nym-node-status-api/nym-node-status-api/src/db/models.rs +++ b/nym-node-status-api/nym-node-status-api/src/db/models.rs @@ -376,3 +376,9 @@ impl TryFrom for http::models::SessionStats { }) } } + +pub(crate) struct ScraperNodeInfo { + pub node_id: i64, + pub host: String, + pub http_api_port: i64, +} diff --git a/nym-node-status-api/nym-node-status-api/src/db/queries/mod.rs b/nym-node-status-api/nym-node-status-api/src/db/queries/mod.rs index 8c87ab2fc0..1a8c4f579b 100644 --- a/nym-node-status-api/nym-node-status-api/src/db/queries/mod.rs +++ b/nym-node-status-api/nym-node-status-api/src/db/queries/mod.rs @@ -2,6 +2,7 @@ mod gateways; mod gateways_stats; mod misc; mod mixnodes; +pub(crate) mod scraper; mod summary; pub(crate) mod testruns; @@ -13,6 +14,7 @@ pub(crate) use misc::insert_summaries; pub(crate) use mixnodes::{ ensure_mixnodes_still_bonded, get_all_mixnodes, get_daily_stats, insert_mixnodes, }; +pub(crate) use scraper::fetch_active_nodes; pub(crate) use summary::{get_summary, get_summary_history}; pub(crate) use gateways_stats::{delete_old_records, get_sessions_stats, insert_session_records}; diff --git a/nym-node-status-api/nym-node-status-api/src/db/queries/scraper.rs b/nym-node-status-api/nym-node-status-api/src/db/queries/scraper.rs new file mode 100644 index 0000000000..848273cc9c --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/src/db/queries/scraper.rs @@ -0,0 +1,20 @@ +use crate::db::{models::ScraperNodeInfo, DbPool}; +use anyhow::Result; + +pub(crate) async fn fetch_active_nodes(pool: &DbPool) -> Result> { + let mut conn = pool.acquire().await?; + let nodes = sqlx::query_as!( + ScraperNodeInfo, + r#" + SELECT mix_id as node_id, host, http_api_port + FROM mixnodes + WHERE bonded = true + "# + ) + .fetch_all(&mut *conn) + .await?; + + Ok(nodes) +} + +// TODO: add stuff for gateways diff --git a/nym-node-status-api/nym-node-status-api/src/main.rs b/nym-node-status-api/nym-node-status-api/src/main.rs index 4fd01d7960..afa93f6424 100644 --- a/nym-node-status-api/nym-node-status-api/src/main.rs +++ b/nym-node-status-api/nym-node-status-api/src/main.rs @@ -6,6 +6,7 @@ mod cli; mod db; mod http; mod logging; +mod mixnet_scraper; mod monitor; mod node_scraper; mod testruns; @@ -28,6 +29,15 @@ async fn main() -> anyhow::Result<()> { let storage = db::Storage::init(connection_url).await?; let db_pool = storage.pool_owned(); + + // Start the node scraper + let scraper = mixnet_scraper::Scraper::new(storage.pool_owned()); + tokio::spawn(async move { + scraper.start().await; + }); + tracing::info!("Started node scraper task"); + + // Start the monitor let args_clone = args.clone(); tokio::spawn(async move { diff --git a/nym-node-status-api/nym-node-status-api/src/mixnet_scraper/helpers.rs b/nym-node-status-api/nym-node-status-api/src/mixnet_scraper/helpers.rs new file mode 100644 index 0000000000..273dff2176 --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/src/mixnet_scraper/helpers.rs @@ -0,0 +1,315 @@ +use crate::db::models::ScraperNodeInfo; +use ammonia::Builder; +use anyhow::Result; +use chrono::{DateTime, Datelike, Utc}; +use nym_network_defaults::DEFAULT_NYM_NODE_HTTP_PORT; +use reqwest; +use serde::{Deserialize, Serialize}; +use sqlx::SqlitePool; +use std::time::Duration; + +#[derive(Debug, Serialize, Deserialize)] +pub struct NodeStats { + pub packets_received: i64, + pub packets_sent: i64, + pub packets_dropped: i64, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct NodeDescriptionResponse { + pub moniker: Option, + pub website: Option, + pub security_contact: Option, + pub details: Option, +} + +// Eventhough they are `/api/v1/*`, newer nodes have a redirect to the correct endpoints +const DESCRIPTION_URL: &str = "/description"; +const PACKET_STATS_URL: &str = "/stats"; + +// We need this as some of the mixnodes respond with float values for the packet statistics (?????) +pub fn get_packet_value(response: &serde_json::Value, key: &str) -> Option { + response + .get(key) + .and_then(|value| value.as_i64().or_else(|| value.as_f64().map(|f| f as i64))) +} + +pub fn parse_mixnet_stats(response: serde_json::Value) -> Option { + // Try to parse the response return by old (deprecated) mixnodes + if let Some(packets_dropped) = + get_packet_value(&response, "packets_explicitly_dropped_since_startup") + { + return Some(NodeStats { + packets_dropped, + packets_sent: get_packet_value(&response, "packets_sent_since_startup") + .unwrap_or_default(), + + packets_received: get_packet_value(&response, "packets_received_since_startup") + .unwrap_or_default(), + }); + } + + // Try to parse the response returned by nym-nodes + if let Some(packets_dropped) = get_packet_value(&response, "dropped_since_startup") { + return Some(NodeStats { + packets_dropped, + packets_sent: get_packet_value(&response, "sent_since_startup").unwrap_or_default(), + + packets_received: get_packet_value(&response, "received_since_startup") + .unwrap_or_default(), + }); + } + + // If neither format matches, return None + None +} + +pub fn calculate_packet_difference(current: i64, previous: i64) -> i64 { + if current >= previous { + current - previous + } else { + current // Node likely restarted, use current value + } +} + +pub fn build_client() -> Result { + reqwest::Client::builder() + .timeout(Duration::from_secs(3)) + // Since we have a mix of TLS and non-TLS nodes, we need to accept invalid certs + // when accessing IP:port + .danger_accept_invalid_certs(true) + .build() + .map_err(|e| anyhow::anyhow!("Failed to build HTTP client: {}", e)) +} + +pub fn sanitize_description(description: NodeDescriptionResponse) -> NodeDescriptionResponse { + let mut sanitizer = Builder::new(); + sanitizer + .tags(std::collections::HashSet::new()) + .generic_attributes(std::collections::HashSet::new()) + .url_schemes(std::collections::HashSet::new()); + + let sanitize_field = |opt: Option| -> Option { + Some( + opt.filter(|s| !s.trim().is_empty()) + .map_or_else(|| "N/A".to_string(), |s| sanitizer.clean(&s).to_string()), + ) + }; + + NodeDescriptionResponse { + moniker: sanitize_field(description.moniker), + website: sanitize_field(description.website), + security_contact: sanitize_field(description.security_contact), + details: sanitize_field(description.details), + } +} + +pub async fn scrape_and_store_description(pool: &SqlitePool, node: &ScraperNodeInfo) -> Result<()> { + let client = build_client()?; + let mut urls = vec![ + format!("http://{}:{DEFAULT_NYM_NODE_HTTP_PORT}", node.host), + format!("http://{}:8000", node.host), + format!("https://{}", node.host), + format!("http://{}", node.host), + ]; + + if node.http_api_port != DEFAULT_NYM_NODE_HTTP_PORT as i64 { + urls.insert(0, format!("http://{}:{}", node.host, node.http_api_port)); + } + + let mut description = None; + let mut error = None; + + for mut url in urls { + url = format!("{}{}", url.trim_end_matches('/'), DESCRIPTION_URL); + + match client.get(&url).send().await { + Ok(response) => { + if let Ok(desc) = response.json::().await { + description = Some(desc); + break; + } + } + Err(e) => error = Some(e), + } + } + + let description = description.ok_or_else(|| { + let err_msg = error.map_or_else(|| "Unknown error".to_string(), |e| e.to_string()); + anyhow::anyhow!("Failed to fetch description from any URL: {}", err_msg) + })?; + + let sanitized_description = sanitize_description(description); + + let mut conn = pool.acquire().await?; + let timestamp = Utc::now().timestamp(); + + sqlx::query!( + r#" + INSERT INTO mixnode_description ( + mix_id, moniker, website, security_contact, details, last_updated_utc + ) VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT (mix_id) DO UPDATE SET + moniker = excluded.moniker, + website = excluded.website, + security_contact = excluded.security_contact, + details = excluded.details, + last_updated_utc = excluded.last_updated_utc + "#, + node.node_id, + sanitized_description.moniker, + sanitized_description.website, + sanitized_description.security_contact, + sanitized_description.details, + timestamp, + ) + .execute(&mut *conn) + .await?; + + Ok(()) +} + +pub async fn scrape_and_store_packet_stats( + pool: &SqlitePool, + node: &ScraperNodeInfo, +) -> Result<()> { + let client = build_client()?; + let mut urls = vec![ + format!("http://{}:{DEFAULT_NYM_NODE_HTTP_PORT}", node.host), + format!("http://{}:8000", node.host), + format!("https://{}", node.host), + format!("http://{}", node.host), + ]; + + if node.http_api_port != DEFAULT_NYM_NODE_HTTP_PORT as i64 { + urls.insert(0, format!("http://{}:{}", node.host, node.http_api_port)); + } + + let mut stats = None; + let mut error = None; + + for mut url in urls { + url = format!("{}{}", url.trim_end_matches('/'), PACKET_STATS_URL); + + match client.get(&url).send().await { + Ok(response) => { + if let Some(node_stats) = parse_mixnet_stats(response.json().await?) { + stats = Some(node_stats); + break; + } + } + Err(e) => error = Some(e), + } + } + + let stats = stats.ok_or_else(|| { + let err_msg = error.map_or_else(|| "Unknown error".to_string(), |e| e.to_string()); + anyhow::anyhow!("Failed to fetch stats from any URL: {}", err_msg) + })?; + + let timestamp = Utc::now(); + let timestamp_utc = timestamp.timestamp(); + let mut conn = pool.acquire().await?; + + // Store raw stats + sqlx::query!( + r#" + INSERT INTO mixnode_packet_stats_raw ( + mix_id, timestamp_utc, packets_received, packets_sent, packets_dropped + ) VALUES (?, ?, ?, ?, ?) + "#, + node.node_id, + timestamp_utc, + stats.packets_received, + stats.packets_sent, + stats.packets_dropped, + ) + .execute(&mut *conn) + .await?; + + // Update daily stats + update_daily_stats(pool, node.node_id, timestamp, &stats).await?; + + Ok(()) +} + +pub async fn update_daily_stats( + pool: &SqlitePool, + node_id: i64, + timestamp: DateTime, + current_stats: &NodeStats, +) -> Result<()> { + let mut conn = pool.acquire().await?; + + let date_utc = format!( + "{:04}-{:02}-{:02}", + timestamp.year(), + timestamp.month(), + timestamp.day() + ); + + let total_stake = sqlx::query_scalar!( + r#" + SELECT + total_stake + FROM mixnodes + WHERE mix_id = ? + "#, + node_id + ) + .fetch_one(&mut *conn) + .await?; + + // Get previous stats + let previous_stats = sqlx::query!( + r#" + SELECT packets_received, packets_sent, packets_dropped + FROM mixnode_packet_stats_raw + WHERE mix_id = ? + ORDER BY timestamp_utc DESC + LIMIT 1 OFFSET 1 + "#, + node_id + ) + .fetch_optional(&mut *conn) + .await?; + + let (diff_received, diff_sent, diff_dropped) = if let Some(prev) = previous_stats { + ( + calculate_packet_difference( + current_stats.packets_received, + prev.packets_received.unwrap_or(0), + ), + calculate_packet_difference(current_stats.packets_sent, prev.packets_sent.unwrap_or(0)), + calculate_packet_difference( + current_stats.packets_dropped, + prev.packets_dropped.unwrap_or(0), + ), + ) + } else { + (0, 0, 0) // No previous stats available + }; + + sqlx::query!( + r#" + INSERT INTO mixnode_daily_stats ( + mix_id, date_utc, total_stake, packets_received, packets_sent, packets_dropped + ) VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(mix_id, date_utc) DO UPDATE SET + total_stake = excluded.total_stake, + packets_received = mixnode_daily_stats.packets_received + excluded.packets_received, + packets_sent = mixnode_daily_stats.packets_sent + excluded.packets_sent, + packets_dropped = mixnode_daily_stats.packets_dropped + excluded.packets_dropped + "#, + node_id, + date_utc, + total_stake, + diff_received, + diff_sent, + diff_dropped, + ) + .execute(&mut *conn) + .await?; + + Ok(()) +} diff --git a/nym-node-status-api/nym-node-status-api/src/mixnet_scraper/mod.rs b/nym-node-status-api/nym-node-status-api/src/mixnet_scraper/mod.rs new file mode 100644 index 0000000000..f9fcd21b22 --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/src/mixnet_scraper/mod.rs @@ -0,0 +1,202 @@ +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; +pub mod helpers; +use anyhow::Result; +use helpers::{scrape_and_store_description, scrape_and_store_packet_stats}; +use sqlx::SqlitePool; +use tracing::{debug, error, warn}; + +use crate::db::models::ScraperNodeInfo; +use crate::db::queries::fetch_active_nodes; + +const DESCRIPTION_SCRAPE_INTERVAL: Duration = Duration::from_secs(60 * 60 * 4); // 4 hours +const PACKET_SCRAPE_INTERVAL: Duration = Duration::from_secs(60 * 60); // 1 hour +const QUEUE_CHECK_INTERVAL: Duration = Duration::from_millis(250); +const MAX_CONCURRENT_TASKS: usize = 5; + +static TASK_COUNTER: AtomicUsize = AtomicUsize::new(0); +static TASK_ID_COUNTER: AtomicUsize = AtomicUsize::new(0); + +pub struct Scraper { + pool: SqlitePool, + description_queue: Arc>>, + packet_queue: Arc>>, +} + +impl Scraper { + pub fn new(pool: SqlitePool) -> Self { + Self { + pool, + description_queue: Arc::new(Mutex::new(Vec::new())), + packet_queue: Arc::new(Mutex::new(Vec::new())), + } + } + + pub async fn start(&self) { + self.spawn_description_scraper().await; + self.spawn_packet_scraper().await; + } + + async fn spawn_description_scraper(&self) { + let pool = self.pool.clone(); + let queue = self.description_queue.clone(); + tracing::info!("Starting description scraper"); + tokio::spawn(async move { + loop { + if let Err(e) = Self::run_description_scraper(&pool, queue.clone()).await { + error!("Description scraper failed: {}", e); + } + tokio::time::sleep(DESCRIPTION_SCRAPE_INTERVAL).await; + } + }); + } + + async fn spawn_packet_scraper(&self) { + let pool = self.pool.clone(); + let queue = self.packet_queue.clone(); + tracing::info!("Starting packet scraper"); + tokio::spawn(async move { + loop { + if let Err(e) = Self::run_packet_scraper(&pool, queue.clone()).await { + error!("Packet scraper failed: {}", e); + } + tokio::time::sleep(PACKET_SCRAPE_INTERVAL).await; + } + }); + } + + async fn run_description_scraper( + pool: &SqlitePool, + queue: Arc>>, + ) -> Result<()> { + let nodes = fetch_active_nodes(pool).await?; + if let Ok(mut queue_lock) = queue.lock() { + queue_lock.extend(nodes); + } else { + warn!("Failed to acquire description queue lock"); + return Ok(()); + } + + Self::process_description_queue(pool, queue).await?; + Ok(()) + } + + async fn run_packet_scraper( + pool: &SqlitePool, + queue: Arc>>, + ) -> Result<()> { + let nodes = fetch_active_nodes(pool).await?; + tracing::info!("Found {} active nodes", nodes.len()); + if let Ok(mut queue_lock) = queue.lock() { + queue_lock.extend(nodes); + } else { + warn!("Failed to acquire packet queue lock"); + return Ok(()); + } + + Self::process_packet_queue(pool, queue).await?; + Ok(()) + } + + async fn process_description_queue( + pool: &SqlitePool, + queue: Arc>>, + ) -> Result<()> { + loop { + let running_tasks = TASK_COUNTER.load(Ordering::Relaxed); + + if running_tasks < MAX_CONCURRENT_TASKS { + let node = { + if let Ok(mut queue_lock) = queue.lock() { + if queue_lock.is_empty() { + TASK_ID_COUNTER.store(0, Ordering::Relaxed); + break; + } + queue_lock.remove(0) + } else { + warn!("Failed to acquire description queue lock"); + break; + } + }; + + TASK_COUNTER.fetch_add(1, Ordering::Relaxed); + let task_id = TASK_ID_COUNTER.fetch_add(1, Ordering::Relaxed); + let pool = pool.clone(); + + tokio::spawn(async move { + match Self::scrape_and_store_description(&pool, &node).await { + Ok(_) => debug!( + "✅ Description task #{} for node {} complete", + task_id, node.node_id + ), + Err(e) => debug!( + "❌ Description task #{} for node {} failed: {}", + task_id, node.node_id, e + ), + } + TASK_COUNTER.fetch_sub(1, Ordering::Relaxed); + }); + } else { + tokio::time::sleep(QUEUE_CHECK_INTERVAL).await; + } + } + Ok(()) + } + + async fn process_packet_queue( + pool: &SqlitePool, + queue: Arc>>, + ) -> Result<()> { + loop { + let running_tasks = TASK_COUNTER.load(Ordering::Relaxed); + + if running_tasks < MAX_CONCURRENT_TASKS { + let node = { + if let Ok(mut queue_lock) = queue.lock() { + if queue_lock.is_empty() { + TASK_ID_COUNTER.store(0, Ordering::Relaxed); + break; + } + queue_lock.remove(0) + } else { + warn!("Failed to acquire packet queue lock"); + break; + } + }; + + TASK_COUNTER.fetch_add(1, Ordering::Relaxed); + let task_id = TASK_ID_COUNTER.fetch_add(1, Ordering::Relaxed); + let pool = pool.clone(); + + tokio::spawn(async move { + match Self::scrape_and_store_packet_stats(&pool, &node).await { + Ok(_) => debug!( + "✅ Packet stats task #{} for node {} complete", + task_id, node.node_id + ), + Err(e) => debug!( + "❌ Packet stats task #{} for node {} failed: {}", + task_id, node.node_id, e + ), + } + TASK_COUNTER.fetch_sub(1, Ordering::Relaxed); + }); + } else { + tokio::time::sleep(QUEUE_CHECK_INTERVAL).await; + } + } + Ok(()) + } + + async fn scrape_and_store_description(pool: &SqlitePool, node: &ScraperNodeInfo) -> Result<()> { + scrape_and_store_description(pool, node).await + } + + async fn scrape_and_store_packet_stats( + pool: &SqlitePool, + node: &ScraperNodeInfo, + ) -> Result<()> { + scrape_and_store_packet_stats(pool, node).await + } +}