diff --git a/nym-node-status-agent/run.sh b/nym-node-status-agent/run.sh index a0cf362e16..0b38466031 100755 --- a/nym-node-status-agent/run.sh +++ b/nym-node-status-agent/run.sh @@ -16,30 +16,13 @@ set -a source "${monorepo_root}/envs/${ENVIRONMENT}.env" set +a -export RUST_LOG="debug" +export RUST_LOG="info" export NODE_STATUS_AGENT_SERVER_ADDRESS="http://127.0.0.1" export NODE_STATUS_AGENT_SERVER_PORT="8000" export NODE_STATUS_AGENT_PROBE_PATH="$crate_root/nym-gateway-probe" - -# each key is required for a separate agent when running in parallel: their -# public counterparts need to be registered with NS API -private_keys=("BjyC9SsHAZUzPRkQR4sPTvVrp4GgaquTh5YfSJksvvWT" - "4RqSKydrEuyGF8Xtwyauvja62SAjqxFPYQzW2neZdkQD" - "CfudaSaaLTkgR8rkBijUnYocdFciWTbKqkSNYevepnbn" - "Dd3fDyPUg4edTpiCAkSweE17NdWJ7gAchbtqAeSj3MBc" - "HAtfcfnpw5XdpcVzAH6Qsxp6Zf75oe2W54HjTD8ngVbU" - "8aqWP8wZyhX5W8gfrvyh1SmS6CEgfLAR95DBhWXRUpTm" - "234U1PMkWpAsn7hD98g1D8hfRFkJJS91T2sJBQDyXmqx" - "5qUUFu83fgqpACsr3dC6iwGJxhTqN4JJDTecT2QSqwEe" - "4Pp7Cd9G3aMku9biFcxRMEja8RBMbBRGZuDAZt1yTS7H" - "4U136QykP8G831EZSDNLLvgWCGYA8naYtT8BQ9kLeL5B" -) +export NODE_STATUS_AGENT_AUTH_KEY="BjyC9SsHAZUzPRkQR4sPTvVrp4GgaquTh5YfSJksvvWT" workers=${1:-1} -if ((workers > ${#private_keys[@]})); then - echo "Error: ${workers} is larger than the number of private keys available (${#private_keys[@]})." - exit 1 -fi echo "Running $workers workers in parallel" # build & copy over GW probe @@ -61,7 +44,6 @@ function swarm() { local workers=$1 for ((i = 1; i <= workers; i++)); do - export NODE_STATUS_AGENT_AUTH_KEY=${private_keys[i]} ../target/release/nym-node-status-agent run-probe & done diff --git a/nym-node-status-agent/src/cli/generate_keypair.rs b/nym-node-status-agent/src/cli/generate_keypair.rs index e89bf3e8d5..e8360af47f 100644 --- a/nym-node-status-agent/src/cli/generate_keypair.rs +++ b/nym-node-status-agent/src/cli/generate_keypair.rs @@ -31,10 +31,7 @@ mod test { use super::*; - use std::{ - fs::{self}, - path::PathBuf, - }; + use std::{fs, path::PathBuf}; #[test] fn can_generate_valid_keypair() { diff --git a/nym-node-status-api/migrations/001_assigned_agent.sql b/nym-node-status-api/migrations/001_assigned_agent.sql deleted file mode 100644 index 96ec895924..0000000000 --- a/nym-node-status-api/migrations/001_assigned_agent.sql +++ /dev/null @@ -1,2 +0,0 @@ -ALTER TABLE testruns -ADD COLUMN assigned_agent VARCHAR; diff --git a/nym-node-status-api/migrations/002_last_assigned_utc.sql b/nym-node-status-api/migrations/001_last_assigned_utc.sql similarity index 100% rename from nym-node-status-api/migrations/002_last_assigned_utc.sql rename to nym-node-status-api/migrations/001_last_assigned_utc.sql index edd91053cf..2a801763e4 100644 --- a/nym-node-status-api/migrations/002_last_assigned_utc.sql +++ b/nym-node-status-api/migrations/001_last_assigned_utc.sql @@ -1,5 +1,5 @@ -ALTER TABLE testruns -ADD COLUMN last_assigned_utc INTEGER; - ALTER TABLE testruns RENAME COLUMN timestamp_utc TO created_utc; + +ALTER TABLE testruns +ADD COLUMN last_assigned_utc INTEGER; diff --git a/nym-node-status-api/src/cli/mod.rs b/nym-node-status-api/src/cli/mod.rs index fa3d3c12bf..10485ab495 100644 --- a/nym-node-status-api/src/cli/mod.rs +++ b/nym-node-status-api/src/cli/mod.rs @@ -73,6 +73,13 @@ pub(crate) struct Cli { #[clap(env = "NODE_STATUS_API_AGENT_KEY_LIST")] #[arg(value_delimiter = ',')] pub(crate) agent_key_list: Vec, + + #[clap( + long, + default_value_t = 40, + env = "NYM_NODE_STATUS_API_NYM_HTTP_CACHE_TTL" + )] + pub(crate) max_agent_count: i64, } fn parse_duration(arg: &str) -> Result { diff --git a/nym-node-status-api/src/db/models.rs b/nym-node-status-api/src/db/models.rs index 4062e1451d..596f634f2e 100644 --- a/nym-node-status-api/src/db/models.rs +++ b/nym-node-status-api/src/db/models.rs @@ -309,7 +309,6 @@ pub struct TestRunDto { pub created_utc: i64, pub ip_address: String, pub log: String, - pub assigned_agent: Option, pub last_assigned_utc: Option, } diff --git a/nym-node-status-api/src/db/queries/testruns.rs b/nym-node-status-api/src/db/queries/testruns.rs index 465444c8ec..dc1e20ee2e 100644 --- a/nym-node-status-api/src/db/queries/testruns.rs +++ b/nym-node-status-api/src/db/queries/testruns.rs @@ -5,14 +5,29 @@ use crate::{ testruns::now_utc, }; use chrono::Duration; -use nym_crypto::asymmetric::ed25519::PublicKey; use sqlx::{pool::PoolConnection, Sqlite}; -pub(crate) async fn testrun_in_progress_assigned_to_agent( +pub(crate) async fn count_testruns_in_progress( conn: &mut PoolConnection, - agent_key: &PublicKey, -) -> sqlx::Result { - let agent_key = agent_key.to_base58_string(); +) -> anyhow::Result { + sqlx::query_scalar!( + r#"SELECT + COUNT(id) as "count: i64" + FROM testruns + WHERE + status = ? + "#, + TestRunStatus::InProgress as i64, + ) + .fetch_one(conn.as_mut()) + .await + .map_err(anyhow::Error::from) +} + +pub(crate) async fn get_in_progress_testrun_by_id( + conn: &mut PoolConnection, + testrun_id: i64, +) -> anyhow::Result { sqlx::query_as!( TestRunDto, r#"SELECT @@ -22,19 +37,20 @@ pub(crate) async fn testrun_in_progress_assigned_to_agent( created_utc as "created_utc!", ip_address as "ip_address!", log as "log!", - assigned_agent, last_assigned_utc FROM testruns WHERE - assigned_agent = ? + id = ? AND status = ? - ORDER BY created_utc"#, - agent_key, + ORDER BY created_utc + LIMIT 1"#, + testrun_id, TestRunStatus::InProgress as i64, ) .fetch_one(conn.as_mut()) .await + .map_err(|e| anyhow::anyhow!("Couldn't retrieve testrun {testrun_id}: {e}")) } pub(crate) async fn update_testruns_assigned_before( @@ -49,8 +65,7 @@ pub(crate) async fn update_testruns_assigned_before( r#"UPDATE testruns SET - status = ?, - assigned_agent = NULL + status = ? WHERE status = ? AND @@ -77,16 +92,13 @@ pub(crate) async fn update_testruns_assigned_before( pub(crate) async fn assign_oldest_testrun( conn: &mut PoolConnection, - agent_key: PublicKey, ) -> anyhow::Result> { - let agent_key = agent_key.to_base58_string(); let now = now_utc().timestamp(); // find & mark as "In progress" in the same transaction to avoid race conditions let returning = sqlx::query!( r#"UPDATE testruns SET status = ?, - assigned_agent = ?, last_assigned_utc = ? WHERE rowid = ( @@ -101,7 +113,6 @@ pub(crate) async fn assign_oldest_testrun( gateway_id "#, TestRunStatus::InProgress as i64, - agent_key, now, TestRunStatus::Queued as i64, ) diff --git a/nym-node-status-api/src/http/api/testruns.rs b/nym-node-status-api/src/http/api/testruns.rs index bc965186d0..6f5ab862ea 100644 --- a/nym-node-status-api/src/http/api/testruns.rs +++ b/nym-node-status-api/src/http/api/testruns.rs @@ -9,6 +9,7 @@ use reqwest::StatusCode; use crate::db::models::TestRunStatus; use crate::db::queries; +use crate::testruns::now_utc; use crate::{ db, http::{ @@ -35,15 +36,9 @@ async fn request_testrun( ) -> HttpResult> { // TODO dz log agent's network probe version authenticate(&request, &state)?; - state - .update_last_request_time( - &request.payload.agent_public_key, - &request.payload.timestamp, - ) - .await?; + is_fresh(&request.payload.timestamp)?; - let agent_pubkey = request.payload.agent_public_key; - tracing::debug!("Agent {} requested testrun", agent_pubkey); + tracing::debug!("Agent requested testrun"); let db = state.db_pool(); let mut conn = db @@ -51,31 +46,29 @@ async fn request_testrun( .await .map_err(HttpError::internal_with_logging)?; - if let Ok(testrun) = - db::queries::testruns::testrun_in_progress_assigned_to_agent(&mut conn, &agent_pubkey).await - { + let active_testruns = db::queries::testruns::count_testruns_in_progress(&mut conn) + .await + .map_err(HttpError::internal_with_logging)?; + if active_testruns >= state.agent_max_count() { tracing::warn!( - "Testrun {} already in progress for agent {:?}, rejecting", - testrun.id, - testrun.assigned_agent + "{}/{} testruns in progress, rejecting", + active_testruns, + state.agent_max_count() ); - return Err(HttpError::invalid_input( - "Testrun already in progress for this agent", - )); - }; + return Err(HttpError::no_testruns_available()); + } - return match db::queries::testruns::assign_oldest_testrun(&mut conn, agent_pubkey).await { + return match db::queries::testruns::assign_oldest_testrun(&mut conn).await { Ok(res) => { if let Some(testrun) = res { tracing::info!( - "🏃‍ Assigned testrun row_id {} gateway {} to agent {}", + "🏃‍ Assigned testrun row_id {} gateway {} to agent", &testrun.testrun_id, testrun.gateway_identity_key, - agent_pubkey ); Ok(Json(testrun)) } else { - tracing::debug!("No testruns available for agent {}", agent_pubkey); + tracing::debug!("No testruns available"); Err(HttpError::no_testruns_available()) } } @@ -97,23 +90,17 @@ async fn submit_testrun( .await .map_err(HttpError::internal_with_logging)?; - let submitter_pubkey = submitted_result.payload.agent_public_key; let assigned_testrun = - queries::testruns::testrun_in_progress_assigned_to_agent(&mut conn, &submitter_pubkey) + queries::testruns::get_in_progress_testrun_by_id(&mut conn, submitted_testrun_id) .await .map_err(|err| { - tracing::warn!("No testruns in progress for agent {submitter_pubkey}: {err}"); + tracing::warn!( + "No testruns in progress for testrun_id {}: {}", + submitted_testrun_id, + err + ); HttpError::invalid_input("Invalid testrun submitted") })?; - if submitted_testrun_id != assigned_testrun.id { - tracing::warn!( - "Agent {} submitted testrun {} but {} was expected", - submitter_pubkey, - submitted_testrun_id, - assigned_testrun.id - ); - return Err(HttpError::invalid_input("Invalid testrun submitted")); - } if Some(submitted_result.payload.assigned_at_utc) != assigned_testrun.last_assigned_utc { tracing::warn!( "Submitted testrun timestamp mismatch: {} != {:?}, rejecting", @@ -132,8 +119,7 @@ async fn submit_testrun( )) })?; tracing::debug!( - "Agent {} submitted testrun {} for gateway {} ({} bytes)", - submitter_pubkey, + "Agent submitted testrun {} for gateway {} ({} bytes)", submitted_testrun_id, gw_identity, &submitted_result.payload.probe_result.len(), @@ -165,10 +151,20 @@ async fn submit_testrun( .await .map_err(HttpError::internal_with_logging)?; + let created_at = chrono::DateTime::from_timestamp(assigned_testrun.created_utc, 0) + .map(|d| d.to_rfc3339()) + .unwrap_or_default(); + let last_assigned = assigned_testrun + .last_assigned_utc + .and_then(|d| chrono::DateTime::from_timestamp(d, 0)) + .map(|d| d.to_rfc3339()) + .unwrap_or_default(); tracing::info!( - "✅ Testrun row_id {} for gateway {} complete", + "✅ Testrun row_id {} for gateway {} complete (last assigned at {}, created at {})", assigned_testrun.id, - gw_identity + gw_identity, + last_assigned, + created_at ); Ok(StatusCode::CREATED) @@ -190,6 +186,17 @@ fn authenticate(request: &impl VerifiableRequest, state: &AppState) -> HttpResul Ok(()) } +fn is_fresh(request_time: &i64) -> HttpResult<()> { + // if a request took longer than N minutes to reach NS API, something is very wrong + let freshness_cutoff = chrono::Duration::minutes(1); + let cutoff_timestamp = (now_utc() - freshness_cutoff).timestamp(); + if *request_time < cutoff_timestamp { + tracing::warn!("Request older than {}s, rejecting", cutoff_timestamp); + return Err(HttpError::unauthorized()); + } + Ok(()) +} + fn get_result_from_log(log: &str) -> String { let re = regex::Regex::new(r"\n\{\s").unwrap(); let result: Vec<_> = re.splitn(log, 2).collect(); diff --git a/nym-node-status-api/src/http/server.rs b/nym-node-status-api/src/http/server.rs index 714a4836e4..c8be0d63a1 100644 --- a/nym-node-status-api/src/http/server.rs +++ b/nym-node-status-api/src/http/server.rs @@ -16,10 +16,11 @@ pub(crate) async fn start_http_api( http_port: u16, nym_http_cache_ttl: u64, agent_key_list: Vec, + agent_max_count: i64, ) -> anyhow::Result { let router_builder = RouterBuilder::with_default_routes(); - let state = AppState::new(db_pool, nym_http_cache_ttl, agent_key_list); + let state = AppState::new(db_pool, nym_http_cache_ttl, agent_key_list, agent_max_count); let router = router_builder.with_state(state); // TODO dz do we need this to be configurable? diff --git a/nym-node-status-api/src/http/state.rs b/nym-node-status-api/src/http/state.rs index fb273027ac..393da5c0b8 100644 --- a/nym-node-status-api/src/http/state.rs +++ b/nym-node-status-api/src/http/state.rs @@ -1,4 +1,4 @@ -use std::{collections::HashMap, sync::Arc, time::Duration}; +use std::{sync::Arc, time::Duration}; use moka::{future::Cache, Entry}; use nym_crypto::asymmetric::ed25519::PublicKey; @@ -6,11 +6,7 @@ use tokio::sync::RwLock; use crate::{ db::DbPool, - http::{ - error::{HttpError, HttpResult}, - models::{DailyStats, Gateway, Mixnode, SummaryHistory}, - }, - testruns::now_utc, + http::models::{DailyStats, Gateway, Mixnode, SummaryHistory}, }; #[derive(Debug, Clone)] @@ -18,18 +14,21 @@ pub(crate) struct AppState { db_pool: DbPool, cache: HttpCache, agent_key_list: Vec, - /// last time agent requested a testrun - // if performance becomes a problem, consider a faster hashmap like `scc`` - agent_last_request_times: Arc>>, + agent_max_count: i64, } impl AppState { - pub(crate) fn new(db_pool: DbPool, cache_ttl: u64, agent_key_list: Vec) -> Self { + pub(crate) fn new( + db_pool: DbPool, + cache_ttl: u64, + agent_key_list: Vec, + agent_max_count: i64, + ) -> Self { Self { db_pool, cache: HttpCache::new(cache_ttl), agent_key_list, - agent_last_request_times: Arc::new(RwLock::new(HashMap::new())), + agent_max_count, } } @@ -45,45 +44,8 @@ impl AppState { self.agent_key_list.contains(agent_pubkey) } - /// Only updates if request is not a replay. Otherwise return error - pub(crate) async fn update_last_request_time( - &self, - agent_key: &PublicKey, - request_time: &i64, - ) -> HttpResult<()> { - // if a request took longer than N minutes to reach NS API, something is very wrong - let cutoff_duration = chrono::Duration::minutes(1); - let cutoff_timestamp = (now_utc() - cutoff_duration).timestamp(); - if *request_time < cutoff_timestamp { - tracing::warn!("Request older than {}s, rejecting", cutoff_timestamp); - return Err(HttpError::unauthorized()); - } - - // if a previous entry has a newer time than this request's submit time, - // it's a repeated request - let agent_key = agent_key.to_base58_string(); - let request_times = self.agent_last_request_times.read().await; - if let Some(previous_request_time) = request_times.get(&agent_key) { - if request_time <= previous_request_time { - tracing::warn!( - "Request has timestamp {} but previous request was at {}, rejecting", - request_time, - previous_request_time - ); - return Err(HttpError::unauthorized()); - } - } - drop(request_times); - - // otherwise this is a newer (or a first) request - self.agent_last_request_times - .write() - .await - .entry(agent_key) - .and_modify(|value| *value = *request_time) - .or_insert(*request_time); - - Ok(()) + pub(crate) fn agent_max_count(&self) -> i64 { + self.agent_max_count } } diff --git a/nym-node-status-api/src/main.rs b/nym-node-status-api/src/main.rs index 0833c6a7b4..6fcdf34473 100644 --- a/nym-node-status-api/src/main.rs +++ b/nym-node-status-api/src/main.rs @@ -47,6 +47,7 @@ async fn main() -> anyhow::Result<()> { args.http_port, args.nym_http_cache_ttl, agent_key_list.to_owned(), + args.max_agent_count, ) .await .expect("Failed to start server"); diff --git a/nym-node-status-api/src/testruns/queue.rs b/nym-node-status-api/src/testruns/queue.rs index a0e7a1422f..7ce0a851ef 100644 --- a/nym-node-status-api/src/testruns/queue.rs +++ b/nym-node-status-api/src/testruns/queue.rs @@ -56,7 +56,6 @@ pub(crate) async fn try_queue_testrun( created_utc as "created_utc!", ip_address as "ip_address!", log as "log!", - assigned_agent, last_assigned_utc FROM testruns WHERE gateway_id = ? AND status != 2