diff --git a/bin/dipper-service/src/config.rs b/bin/dipper-service/src/config.rs index bbef7f05..78ff8550 100644 --- a/bin/dipper-service/src/config.rs +++ b/bin/dipper-service/src/config.rs @@ -70,6 +70,9 @@ pub struct Config { /// The liveness checker service configuration (detects silent agreement abandonment) #[serde(default)] pub liveness_checker: Option, + /// The DIPs info watcher configuration (alerts on agreement holders missing DIPs info) + #[serde(default)] + pub dips_info_watcher: Option, /// The escrow reconciler service configuration (runs only when this section is present) #[serde(default)] pub escrow_reconciler: Option, @@ -141,6 +144,8 @@ impl Default for AlertsConfig { "stale_agreement_uncancellable", "liveness_abandonment_paused", "iisa_request_rejected", + "dips_info_missing", + "dips_info_checks_paused", ] .map(str::to_owned) .to_vec(), @@ -628,6 +633,113 @@ impl Default for LivenessCheckerConfig { } } +/// Configuration for the DIPs info watcher, which alerts when an indexer holding an accepted +/// agreement goes too long without a readable answer from its `/dips/info` endpoint. +#[serde_as] +#[derive(Debug, Clone, serde::Deserialize)] +#[serde(deny_unknown_fields)] +pub struct DipsInfoWatcherConfig { + /// Whether the DIPs info watcher is enabled (default: false). + #[serde(default)] + pub enabled: bool, + + /// Interval between checks in seconds (default: 300s / 5 min). + #[serde_as(as = "serde_with::DurationSeconds")] + #[serde(default = "default_dips_info_watcher_interval")] + pub interval: Duration, + + /// Hours an indexer may go without a readable answer before the `dips_info_missing` + /// alert (default: 24). + #[serde(default = "default_dips_info_watcher_alert_after_hours")] + pub alert_after_hours: u32, + + /// Timeout per `/dips/info` request in seconds (default: 10s). + #[serde_as(as = "serde_with::DurationSeconds")] + #[serde(default = "default_dips_info_watcher_request_timeout")] + pub request_timeout: Duration, + + /// Most indexers asked at once (default: 16). + #[serde(default = "default_dips_info_watcher_max_concurrent_requests")] + pub max_concurrent_requests: usize, + + /// Share of the indexers checked in a cycle that, giving no readable answer at once, pauses + /// counting missing time, since so many points to a fault on dipper's side or a release that + /// changed the endpoint (default: 0.5). + #[serde(default = "default_liveness_checker_unreachable_trip_fraction")] + pub unreachable_trip_fraction: f64, + + /// Fewest indexers checked in a cycle for that share to count (default: 3). + #[serde(default = "default_liveness_checker_unreachable_trip_min_indexers")] + pub unreachable_trip_min_indexers: usize, +} + +fn default_dips_info_watcher_interval() -> Duration { + Duration::from_secs(300) +} + +fn default_dips_info_watcher_alert_after_hours() -> u32 { + 24 +} + +fn default_dips_info_watcher_request_timeout() -> Duration { + Duration::from_secs(10) +} + +fn default_dips_info_watcher_max_concurrent_requests() -> usize { + 16 +} + +impl DipsInfoWatcherConfig { + /// Reject settings the watcher can't run with. A disabled watcher uses none of them, so only + /// an enabled one is checked. + pub fn validate(&self) -> Result<(), String> { + if !self.enabled { + return Ok(()); + } + if self.interval.is_zero() || self.interval > Duration::from_secs(24 * 3600) { + return Err(format!( + "interval ({}s) must be above 0 and at most 1 day", + self.interval.as_secs() + )); + } + if self.request_timeout.is_zero() { + return Err("request_timeout must be above 0".to_string()); + } + if self.alert_after_hours == 0 { + return Err("alert_after_hours must be above 0".to_string()); + } + if self.max_concurrent_requests == 0 { + return Err("max_concurrent_requests must be above 0".to_string()); + } + let fraction = self.unreachable_trip_fraction; + if fraction.is_nan() || fraction <= 0.0 || fraction > 1.0 { + return Err(format!( + "unreachable_trip_fraction ({fraction}) must be above 0 and at most 1" + )); + } + Ok(()) + } + + /// How long an indexer may go without a readable answer before it is alerted on. + pub fn alert_after(&self) -> Duration { + Duration::from_secs(u64::from(self.alert_after_hours) * 3600) + } +} + +impl Default for DipsInfoWatcherConfig { + fn default() -> Self { + Self { + enabled: false, + interval: default_dips_info_watcher_interval(), + alert_after_hours: default_dips_info_watcher_alert_after_hours(), + request_timeout: default_dips_info_watcher_request_timeout(), + max_concurrent_requests: default_dips_info_watcher_max_concurrent_requests(), + unreachable_trip_fraction: default_liveness_checker_unreachable_trip_fraction(), + unreachable_trip_min_indexers: default_liveness_checker_unreachable_trip_min_indexers(), + } + } +} + /// Configuration for the on-chain event listener service. /// /// This service monitors the SubgraphService contract for `IndexingAgreementAccepted` @@ -1910,6 +2022,65 @@ mod tests { ); } + #[test] + fn dips_info_watcher_defaults_alert_after_a_day() { + let conf: DipsInfoWatcherConfig = + serde_json::from_str(r#"{"enabled": true}"#).expect("deserialization"); + + assert!(conf.validate().is_ok()); + assert_eq!(conf.alert_after(), Duration::from_secs(24 * 3600)); + assert_eq!(conf.interval, Duration::from_secs(300)); + assert!( + !DipsInfoWatcherConfig::default().enabled, + "off until configured" + ); + } + + #[test] + fn dips_info_watcher_validate_rejects_settings_it_cannot_run_with() { + let enabled = DipsInfoWatcherConfig { + enabled: true, + ..DipsInfoWatcherConfig::default() + }; + for bad in [ + DipsInfoWatcherConfig { + interval: Duration::ZERO, + ..enabled.clone() + }, + DipsInfoWatcherConfig { + interval: Duration::from_secs(24 * 3600 + 1), + ..enabled.clone() + }, + DipsInfoWatcherConfig { + request_timeout: Duration::ZERO, + ..enabled.clone() + }, + DipsInfoWatcherConfig { + alert_after_hours: 0, + ..enabled.clone() + }, + DipsInfoWatcherConfig { + max_concurrent_requests: 0, + ..enabled.clone() + }, + DipsInfoWatcherConfig { + unreachable_trip_fraction: f64::NAN, + ..enabled.clone() + }, + ] { + assert!(bad.validate().is_err(), "{bad:?}"); + } + let disabled = DipsInfoWatcherConfig { + enabled: false, + alert_after_hours: 0, + ..enabled + }; + assert!( + disabled.validate().is_ok(), + "a disabled watcher uses none of it" + ); + } + #[test] fn validate_rejects_a_duration_too_short_for_its_deadline() { // With a bounded duration the agreement must outlive the acceptance @@ -2480,6 +2651,7 @@ mod tests { assert_rejects::("expiration"); assert_rejects::("escrow_reconciler"); assert_rejects::("liveness_checker"); + assert_rejects::("dips_info_watcher"); assert_rejects::("indexer_client"); assert_rejects::("event_streaming_config"); } diff --git a/bin/dipper-service/src/main.rs b/bin/dipper-service/src/main.rs index 707ce765..ae216fe5 100644 --- a/bin/dipper-service/src/main.rs +++ b/bin/dipper-service/src/main.rs @@ -111,6 +111,11 @@ pub async fn main() -> anyhow::Result<()> { { anyhow::bail!("invalid liveness checker config: {err}"); } + if let Some(watcher_conf) = &conf.dips_info_watcher + && let Err(err) = watcher_conf.validate() + { + anyhow::bail!("invalid DIPs info watcher config: {err}"); + } // TODO: Decouple the config file format from the internal representation let (agreement_conf, pricing_table) = conf.dips.into(); @@ -587,6 +592,20 @@ pub async fn main() -> anyhow::Result<()> { _ => None, }; + //- The DIPs info watcher (optional, enabled by config) + // Alerts on agreement holders whose /dips/info stays unreadable, which IISA can't score + let dips_info_watcher_handle = match conf.dips_info_watcher { + Some(ref watcher_conf) if watcher_conf.enabled => { + let ctx = network::service::dips_info_watcher::Ctx { + registry: registry.clone(), + network: network_provider.clone(), + config: watcher_conf.clone(), + }; + Some(network::service::dips_info_watcher::new(ctx)) + } + _ => None, + }; + //- The escrow reconciler service (AgreementManager mode only) // Reclaims orphaned manager-owned escrow the indexer has no reason to touch. let escrow_reconciler_handle = if network::service::escrow_reconciler::should_run( @@ -698,6 +717,15 @@ pub async fn main() -> anyhow::Result<()> { None }; + // Spawn the DIPs info watcher if enabled + let dips_info_watcher_stop_handle = if let Some((handle, service)) = dips_info_watcher_handle { + let task_handle = task_tree.spawn(service); + tracing::debug!(task_id=%task_handle.id(), "DIPs info watcher started"); + Some(handle) + } else { + None + }; + // Spawn the chain listener service if enabled let chain_listener_stop_handle = if let Some((handle, service)) = chain_listener_handle { let task_handle = task_tree.spawn(service); @@ -785,6 +813,10 @@ pub async fn main() -> anyhow::Result<()> { all_stopped &= stop_service("Liveness checker", handle.stop()).await; } + if let Some(handle) = dips_info_watcher_stop_handle { + all_stopped &= stop_service("DIPs info watcher", handle.stop()).await; + } + // Stop chain listener service before worker (it depends on worker queue) if let Some(handle) = chain_listener_stop_handle { all_stopped &= stop_service("Chain listener", handle.stop()).await; diff --git a/bin/dipper-service/src/network/service.rs b/bin/dipper-service/src/network/service.rs index 3fc9d0f0..e7e1c39e 100644 --- a/bin/dipper-service/src/network/service.rs +++ b/bin/dipper-service/src/network/service.rs @@ -1,6 +1,7 @@ pub mod cancel_retry; pub mod chain_events; pub mod chain_listener; +pub mod dips_info_watcher; pub mod domain_refresh; pub mod entity_count_cache; pub mod escrow_reconciler; diff --git a/bin/dipper-service/src/network/service/dips_info_watcher.rs b/bin/dipper-service/src/network/service/dips_info_watcher.rs new file mode 100644 index 00000000..5c01c3a4 --- /dev/null +++ b/bin/dipper-service/src/network/service/dips_info_watcher.rs @@ -0,0 +1,719 @@ +//! DIPs info watcher: counts how long each indexer holding an accepted agreement goes without a +//! readable answer from its `/dips/info` endpoint, and alerts once that passes a limit. IISA keeps +//! such an indexer in its group but can't score it, so nothing else would ever replace it. + +use std::{collections::HashMap, future::Future, sync::Arc, time::Duration}; + +use thegraph_core::IndexerId; +use tokio::{ + sync::{Semaphore, mpsc}, + task::JoinSet, + time::{Instant, MissedTickBehavior}, +}; + +use super::liveness_checker::{UNREACHABLE_CLEAR_CYCLES, unreachable_trips}; +use crate::{ + config::DipsInfoWatcherConfig, + network::provider::NetworkProviderService, + registry::{DipsInfoChecks, DipsInfoRegistry}, +}; + +/// Handle for controlling the DIPs info watcher's lifecycle. +#[derive(Clone)] +pub struct Handle { + tx_stop: mpsc::Sender<()>, +} + +impl Handle { + /// Stop the DIPs info watcher gracefully. + pub async fn stop(&self) { + if self.tx_stop.is_closed() { + return; + } + let _ = self.tx_stop.send(()).await; + self.tx_stop.closed().await; + } +} + +/// Context required by the DIPs info watcher. +pub struct Ctx { + /// Registry for the indexers to check and their missing time. + pub registry: R, + /// Network provider for looking up fresh indexer URLs. + pub network: NetworkProviderService, + /// Service configuration. + pub config: DipsInfoWatcherConfig, +} + +/// Create a new DIPs info watcher. Returns a handle for stopping it and a future that must be +/// spawned on a runtime. +pub fn new(ctx: Ctx) -> (Handle, impl Future>) +where + R: DipsInfoRegistry + Send + Sync, +{ + let (tx_stop, mut rx_stop) = mpsc::channel(1); + let Ctx { + registry, + network, + config, + } = ctx; + + let service = async move { + tracing::info!( + interval_secs = config.interval.as_secs(), + alert_after_hours = config.alert_after_hours, + "DIPs info watcher started" + ); + + let mut timer = tokio::time::interval(config.interval); + timer.set_missed_tick_behavior(MissedTickBehavior::Skip); + let http_client = reqwest::Client::builder() + .timeout(config.request_timeout) + .build() + .map_err(|e| anyhow::anyhow!("failed to build HTTP client: {e}"))?; + let mut cycles = Cycles::default(); + + loop { + tokio::select! { + _ = rx_stop.recv() => break, + _ = timer.tick() => {}, + } + // A stop ends a cycle midway, dropping its requests before anything is recorded, so + // shutdown doesn't wait on slow indexers. + tokio::select! { + _ = rx_stop.recv() => break, + () = run_cycle(®istry, &network, &http_client, &mut cycles, &config) => {}, + } + } + + tracing::debug!("DIPs info watcher stopped"); + Ok(()) + }; + + (Handle { tx_stop }, service) +} + +/// Run 1 cycle: ask every agreement holder for its DIPs info, record how long each without it has +/// gone, and alert on any that reached the limit. A failed registry call ends the cycle early. +async fn run_cycle( + registry: &R, + network: &NetworkProviderService, + http_client: &reqwest::Client, + cycles: &mut Cycles, + config: &DipsInfoWatcherConfig, +) { + let now = Instant::now(); + let elapsed = cycles.elapsed(now, config.interval); + let Some(indexers) = db( + "query indexers with accepted agreements", + registry.get_indexers_with_accepted_agreements(), + ) + .await + else { + return; + }; + let missing = missing_dips_info( + network, + http_client, + &indexers, + config.max_concurrent_requests, + ) + .await; + let added = cycles.judge(indexers.len(), missing.len(), elapsed, now, config); + let Some(checks) = db( + "record DIPs info checks", + registry.record_dips_info_checks(&missing, added), + ) + .await + else { + return; + }; + cycles.recorded(now); + report(&checks, added, config.alert_after()); +} + +/// Run 1 registry call with a timeout, logging and returning nothing when it fails, so the cycle +/// can end early; the next one tries again. +async fn db(what: &str, call: impl Future>) -> Option { + const DB_TIMEOUT: Duration = Duration::from_secs(30); + match tokio::time::timeout(DB_TIMEOUT, call).await { + Ok(Ok(value)) => Some(value), + Ok(Err(err)) => { + tracing::error!(error = %err, "failed to {what}"); + None + } + Err(_) => { + tracing::error!("timeout trying to {what}"); + None + } + } +} + +/// How often the pause alert repeats while counting stays paused. +const PAUSE_ALERT_REPEAT: Duration = Duration::from_secs(6 * 3600); + +/// What the watcher remembers between cycles: when the last recorded one started, and, while too +/// many indexers giving no readable answer pause counting, since when and when last alerted on. +#[derive(Default)] +struct Cycles { + last_recorded: Option, + paused_since: Option, + pause_alerted_at: Option, + /// Cycles in a row, while paused, with too few indexers missing to pause it. + quiet_cycles: u32, +} + +impl Cycles { + /// The time since the last recorded cycle, at most 2 intervals so a stall doesn't count as + /// missing time. The first cycle after startup counts nothing. + fn elapsed(&self, now: Instant, interval: Duration) -> Duration { + self.last_recorded + .map_or(Duration::ZERO, |at| now.saturating_duration_since(at)) + .min(interval.saturating_mul(2)) + } + + /// Note that the cycle started at `now` was recorded, so a cycle that fails before then + /// leaves its time to the next one. + fn recorded(&mut self, now: Instant) { + self.last_recorded = Some(now); + } + + /// The missing time this cycle adds: none while so many of the indexers checked give no + /// readable answer that the fault is likely dipper's, or a release that changed the endpoint. + /// The pause has no time limit, so its alert repeats every `PAUSE_ALERT_REPEAT` while it lasts. + fn judge( + &mut self, + checked: usize, + missing: usize, + elapsed: Duration, + now: Instant, + config: &DipsInfoWatcherConfig, + ) -> Duration { + let trips = unreachable_trips( + checked, + missing, + config.unreachable_trip_fraction, + config.unreachable_trip_min_indexers, + ); + // Clearing takes several calm cycles in a row, so a share hovering around the trip + // fraction doesn't pause and resume counting every cycle. + if trips { + self.quiet_cycles = 0; + } else if self.paused_since.is_some() { + self.quiet_cycles += 1; + if self.quiet_cycles >= UNREACHABLE_CLEAR_CYCLES { + self.paused_since = None; + self.quiet_cycles = 0; + tracing::info!(missing, checked, "Counting missing DIPs info again"); + } + } + if trips { + self.paused_since.get_or_insert(now); + } + let Some(since) = self.paused_since else { + return elapsed; + }; + if self + .pause_alerted_at + .is_none_or(|at| at < since || now.saturating_duration_since(at) >= PAUSE_ALERT_REPEAT) + { + self.pause_alerted_at = Some(now); + tracing::warn!( + event = "dips_info_checks_paused", + missing, + checked, + paused_hours = now.saturating_duration_since(since).as_secs() / 3600, + "Too many indexers without readable DIPs info at once: a fault on dipper's side \ + or an indexer-service release that changed /dips/info; not counting missing \ + DIPs info while it lasts" + ); + } + Duration::ZERO + } + + /// Whether counting is paused. + #[cfg(test)] + fn paused(&self) -> bool { + self.paused_since.is_some() + } +} + +/// Log what a recorded cycle means: 1 alert naming every indexer whose missing time passed +/// `alert_after` in it, since Slack posts 1 message per event in each throttle window and only +/// counts the rest, and a note for each one that had been alerted on and has answered again. +fn report(checks: &DipsInfoChecks, added: Duration, alert_after: Duration) { + let mut reached: Vec = checks + .missing + .iter() + .filter(|(_, missing)| crossed(**missing, added, alert_after)) + .map(|(indexer_id, _)| indexer_id.to_string()) + .collect(); + if !reached.is_empty() { + reached.sort(); + tracing::warn!( + event = "dips_info_missing", + indexer_ids = reached.join(", "), + alert_after_hours = alert_after.as_secs() / 3600, + "Indexers holding an accepted agreement have given no readable /dips/info answer for \ + alert_after_hours; IISA keeps them in their groups but can't score them" + ); + } + for (indexer_id, missing) in &checks.cleared { + if *missing >= alert_after { + tracing::info!( + indexer_id = %indexer_id, + missing_hours = missing.as_secs() / 3600, + "Indexer answers /dips/info readably again, or holds no accepted agreement" + ); + } + } +} + +/// Whether adding `added` took a total of `total` to `limit` or past it. +fn crossed(total: Duration, added: Duration, limit: Duration) -> bool { + !added.is_zero() && total >= limit && total.saturating_sub(added) < limit +} + +/// Ask each of `indexers` for its DIPs info, at its current URL in the network registry, at most +/// `max_concurrent` at a time, returning those that gave no readable answer. +async fn missing_dips_info( + network: &NetworkProviderService, + http_client: &reqwest::Client, + indexers: &[IndexerId], + max_concurrent: usize, +) -> Vec { + let asks = indexers.iter().map(|indexer_id| { + let url = network + .get_indexer_by_id(indexer_id) + .map(|indexer| indexer.url); + let http_client = http_client.clone(); + let ask = async move { + match url { + Some(url) => ask_dips_info(&http_client, &url).await, + None => Err("not in the network registry".to_string()), + } + }; + (*indexer_id, ask) + }); + missing_answers(asks, max_concurrent).await +} + +/// Run each indexer's `ask`, at most `max_concurrent` at a time, returning the indexers whose ask +/// failed. One that crashed counts as failed, so its indexer keeps its count rather than being +/// forgotten as if it had answered. +async fn missing_answers( + asks: impl IntoIterator, + max_concurrent: usize, +) -> Vec +where + A: Future> + Send + 'static, +{ + let permits = Arc::new(Semaphore::new(max_concurrent.max(1))); + let mut tasks = JoinSet::new(); + let mut asked_by_task = HashMap::new(); + for (indexer_id, ask) in asks { + let permits = permits.clone(); + let task = tasks.spawn(async move { + let _permit = permits.acquire_owned().await; + (indexer_id, ask.await) + }); + asked_by_task.insert(task.id(), indexer_id); + } + + let mut missing = Vec::new(); + while let Some(asked) = tasks.join_next().await { + match asked { + Ok((_, Ok(()))) => {} + Ok((indexer_id, Err(why))) => { + tracing::debug!(indexer_id = %indexer_id, why, "no readable DIPs info"); + missing.push(indexer_id); + } + Err(err) => { + let indexer_id = asked_by_task.get(&err.id()).copied(); + tracing::error!(error = %err, indexer_id = ?indexer_id, "DIPs info check failed"); + missing.extend(indexer_id); + } + } + } + missing +} + +/// Ask 1 indexer for its DIPs info, returning why its answer isn't readable, if it isn't. +async fn ask_dips_info( + http_client: &reqwest::Client, + indexer_url: &url::Url, +) -> Result<(), String> { + let url = indexer_url + .join("dips/info") + .map_err(|e| format!("can't build the /dips/info URL: {e}"))?; + let response = http_client + .get(url) + .send() + .await + .and_then(reqwest::Response::error_for_status) + .map_err(|e| e.to_string())?; + let body = response.bytes().await.map_err(|e| e.to_string())?; + readable_dips_info(&body).map_err(str::to_string) +} + +/// Check a `/dips/info` body the way IISA's score job does: a JSON object, with any per-entity +/// price, top-level or under `pricing`, a finite number of at least 0. IISA counts anything else +/// as no DIPs info, so an invalid per-entity price hides an indexer from scoring as well. +fn readable_dips_info(body: &[u8]) -> Result<(), &'static str> { + let value: serde_json::Value = + serde_json::from_slice(body).map_err(|_| "the body is not JSON")?; + let object = value.as_object().ok_or("the body is not a JSON object")?; + let pricing = object + .get("pricing") + .and_then(serde_json::Value::as_object) + .unwrap_or(object); + match pricing.get("min_grt_per_billion_entities_per_30_days") { + None | Some(serde_json::Value::Null) => Ok(()), + Some(price) if valid_price(price) => Ok(()), + Some(_) => Err("the per-entity price is not a finite number of at least 0"), + } +} + +/// Whether an advertised price is a finite number of at least 0, given as a JSON number or as +/// text. Text is read as loosely as IISA reads it, ignoring surrounding spaces and underscores. +fn valid_price(price: &serde_json::Value) -> bool { + let parsed = match price { + serde_json::Value::Number(number) => number.as_f64(), + serde_json::Value::String(text) => text.trim().replace('_', "").parse::().ok(), + _ => None, + }; + parsed.is_some_and(|price| price.is_finite() && price >= 0.0) +} + +#[cfg(test)] +mod tests { + use std::collections::{BTreeMap, HashMap}; + + use thegraph_core::indexer_id; + use wiremock::{ + Mock, MockServer, ResponseTemplate, + matchers::{method, path}, + }; + + use super::*; + use crate::network::service::indexer_urls; + + const HOUR: Duration = Duration::from_secs(3600); + + #[test] + fn readable_dips_info_takes_an_object_with_a_valid_or_no_per_entity_price() { + for body in [ + r#"{}"#, + r#"{"min_grt_per_30_days": {"arbitrum-one": "450"}}"#, + r#"{"min_grt_per_billion_entities_per_30_days": 2000}"#, + r#"{"min_grt_per_billion_entities_per_30_days": " 1_000.5 "}"#, + r#"{"min_grt_per_billion_entities_per_30_days": null}"#, + r#"{"pricing": {"min_grt_per_billion_entities_per_30_days": "0"}}"#, + ] { + assert_eq!(readable_dips_info(body.as_bytes()), Ok(()), "{body}"); + } + } + + #[test] + fn readable_dips_info_refuses_what_iisa_counts_as_no_dips_info() { + for body in [ + "", + "not json", + "[]", + "null", + r#""text""#, + r#"{"min_grt_per_billion_entities_per_30_days": -1}"#, + r#"{"min_grt_per_billion_entities_per_30_days": "NaN"}"#, + r#"{"min_grt_per_billion_entities_per_30_days": "inf"}"#, + r#"{"min_grt_per_billion_entities_per_30_days": "cheap"}"#, + r#"{"min_grt_per_billion_entities_per_30_days": true}"#, + r#"{"min_grt_per_billion_entities_per_30_days": [1]}"#, + r#"{"pricing": {"min_grt_per_billion_entities_per_30_days": -5}}"#, + ] { + assert!(readable_dips_info(body.as_bytes()).is_err(), "{body}"); + } + } + + #[test] + fn readable_dips_info_reads_the_top_level_price_when_pricing_is_not_an_object() { + let body = r#"{"pricing": 1, "min_grt_per_billion_entities_per_30_days": -1}"#; + assert!(readable_dips_info(body.as_bytes()).is_err()); + } + + #[test] + fn crossed_fires_only_on_the_cycle_that_reaches_the_limit() { + let limit = 24 * HOUR; + assert!(!crossed(23 * HOUR, HOUR, limit)); + assert!(crossed(24 * HOUR, HOUR, limit)); + assert!( + !crossed(25 * HOUR, HOUR, limit), + "already past it last cycle" + ); + assert!( + !crossed(24 * HOUR, Duration::ZERO, limit), + "a cycle that added nothing reached nothing" + ); + } + + #[test] + fn a_cycle_counts_the_time_since_the_last_recorded_up_to_2_intervals() { + let interval = Duration::from_secs(300); + let start = Instant::now(); + let mut cycles = Cycles::default(); + + assert_eq!( + cycles.elapsed(start, interval), + Duration::ZERO, + "first cycle" + ); + cycles.recorded(start); + let failed = start + Duration::from_secs(300); + assert_eq!(cycles.elapsed(failed, interval), Duration::from_secs(300)); + let next = start + Duration::from_secs(590); + assert_eq!( + cycles.elapsed(next, interval), + Duration::from_secs(590), + "a cycle that failed to record leaves its time to the next" + ); + cycles.recorded(next); + assert_eq!( + cycles.elapsed(next + Duration::from_secs(5000), interval), + Duration::from_secs(600), + "a stall counts as 2 intervals at most" + ); + } + + #[test] + fn a_cycle_with_too_many_indexers_missing_adds_no_time() { + let config = DipsInfoWatcherConfig::default(); + let mut cycles = Cycles::default(); + let elapsed = Duration::from_secs(300); + let now = Instant::now(); + + assert_eq!(cycles.judge(4, 1, elapsed, now, &config), elapsed); + assert_eq!( + cycles.judge(4, 2, elapsed, now, &config), + Duration::ZERO, + "half missing" + ); + assert!(cycles.paused()); + for _ in 0..2 { + assert_eq!( + cycles.judge(4, 1, elapsed, now, &config), + Duration::ZERO, + "still paused until 3 calm cycles in a row" + ); + } + assert_eq!(cycles.judge(4, 2, elapsed, now, &config), Duration::ZERO); + for _ in 0..2 { + cycles.judge(4, 1, elapsed, now, &config); + } + assert!( + cycles.paused(), + "a missing share back up starts the calm cycles again" + ); + assert_eq!( + cycles.judge(4, 1, elapsed, now, &config), + elapsed, + "counting resumes on the 3rd calm cycle" + ); + assert!(!cycles.paused()); + assert_eq!( + cycles.judge(2, 2, elapsed, now, &config), + elapsed, + "too few indexers checked for the share to count" + ); + } + + #[tokio::test] + async fn the_pause_alert_repeats_while_counting_stays_paused() { + let config = DipsInfoWatcherConfig::default(); + let mut cycles = Cycles::default(); + let elapsed = Duration::from_secs(300); + let start = Instant::now(); + + let logs = crate::test_support::logged(async { + for minutes in [0, 5, 10, 6 * 60 + 5] { + cycles.judge( + 4, + 4, + elapsed, + start + Duration::from_secs(minutes * 60), + &config, + ); + } + }) + .await; + + assert_eq!( + logs.matches("dips_info_checks_paused").count(), + 2, + "once as it pauses, again 6 hours on: {logs}" + ); + } + + #[tokio::test] + async fn missing_dips_info_lists_the_indexers_without_a_readable_answer() { + //* Given + let readable = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/dips/info")) + .respond_with( + ResponseTemplate::new(200).set_body_string(r#"{"supported_networks": []}"#), + ) + .mount(&readable) + .await; + let removed = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/dips/info")) + .respond_with(ResponseTemplate::new(404)) + .mount(&removed) + .await; + let bad_price = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/dips/info")) + .respond_with( + ResponseTemplate::new(200) + .set_body_string(r#"{"min_grt_per_billion_entities_per_30_days": "free"}"#), + ) + .mount(&bad_price) + .await; + + let answers = indexer_id!("1111111111111111111111111111111111111111"); + let gone = indexer_id!("2222222222222222222222222222222222222222"); + let invalid = indexer_id!("3333333333333333333333333333333333333333"); + let unregistered = indexer_id!("4444444444444444444444444444444444444444"); + let url = |server: &MockServer| format!("{}/", server.uri()).parse().unwrap(); + let snapshot: indexer_urls::Snapshot = BTreeMap::from([ + (answers, url(&readable)), + (gone, url(&removed)), + (invalid, url(&bad_price)), + ]); + let network = NetworkProviderService::new(indexer_urls::Handle::for_test(snapshot)); + + //* When + let mut missing = missing_dips_info( + &network, + &reqwest::Client::new(), + &[answers, gone, invalid, unregistered], + 2, + ) + .await; + + //* Then + missing.sort(); + let mut expected = vec![gone, invalid, unregistered]; + expected.sort(); + assert_eq!(missing, expected); + } + + #[tokio::test] + async fn a_check_that_crashes_counts_its_indexer_as_missing() { + let answers = indexer_id!("1111111111111111111111111111111111111111"); + let crashes = indexer_id!("2222222222222222222222222222222222222222"); + let asks = [answers, crashes].map(|indexer_id| { + (indexer_id, async move { + assert_ne!(indexer_id, crashes, "the check crashes"); + Ok(()) + }) + }); + + let missing = missing_answers(asks, 2).await; + + assert_eq!(missing, vec![crashes]); + } + + /// A registry holding a fixed set of agreement holders, recording each cycle it is given. + struct StubRegistry { + holders: Vec, + recorded: Arc>>>, + } + + #[async_trait::async_trait] + impl DipsInfoRegistry for StubRegistry { + async fn get_indexers_with_accepted_agreements( + &self, + ) -> crate::registry::Result> { + Ok(self.holders.clone()) + } + + async fn record_dips_info_checks( + &self, + missing: &[IndexerId], + _added: Duration, + ) -> crate::registry::Result { + self.recorded.lock().unwrap().push(missing.to_vec()); + Ok(DipsInfoChecks::default()) + } + } + + #[tokio::test] + async fn stop_ends_a_cycle_waiting_on_a_slow_indexer() { + //* Given + let slow = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/dips/info")) + .respond_with(ResponseTemplate::new(200).set_delay(Duration::from_secs(60))) + .mount(&slow) + .await; + let indexer = indexer_id!("1111111111111111111111111111111111111111"); + let snapshot: indexer_urls::Snapshot = + BTreeMap::from([(indexer, format!("{}/", slow.uri()).parse().unwrap())]); + let recorded = Arc::new(std::sync::Mutex::new(Vec::new())); + let registry = StubRegistry { + holders: vec![indexer], + recorded: recorded.clone(), + }; + let (handle, service) = new(Ctx { + registry, + network: NetworkProviderService::new(indexer_urls::Handle::for_test(snapshot)), + config: DipsInfoWatcherConfig { + enabled: true, + request_timeout: Duration::from_secs(60), + ..DipsInfoWatcherConfig::default() + }, + }); + let task = tokio::spawn(service); + while slow + .received_requests() + .await + .unwrap_or_default() + .is_empty() + { + tokio::time::sleep(Duration::from_millis(10)).await; + } + + //* When + let stopped = tokio::time::timeout(Duration::from_secs(5), handle.stop()).await; + + //* Then + assert!(stopped.is_ok(), "stop waited on the slow indexer"); + assert!(task.await.unwrap().is_ok()); + assert!( + recorded.lock().unwrap().is_empty(), + "the interrupted cycle recorded nothing" + ); + } + + #[tokio::test] + async fn report_alerts_once_naming_every_indexer_that_reached_the_limit() { + let reached = indexer_id!("1111111111111111111111111111111111111111"); + let also_reached = indexer_id!("3333333333333333333333333333333333333333"); + let short = indexer_id!("2222222222222222222222222222222222222222"); + let checks = DipsInfoChecks { + missing: HashMap::from([ + (reached, 24 * HOUR), + (also_reached, 24 * HOUR), + (short, 23 * HOUR), + ]), + cleared: HashMap::new(), + }; + + let logs = crate::test_support::logged(async { report(&checks, HOUR, 24 * HOUR) }).await; + + assert_eq!(logs.matches("dips_info_missing").count(), 1, "{logs}"); + assert!(logs.contains(&reached.to_string()), "{logs}"); + assert!(logs.contains(&also_reached.to_string()), "{logs}"); + assert!(!logs.contains(&short.to_string()), "{logs}"); + } +} diff --git a/bin/dipper-service/src/network/service/liveness_checker.rs b/bin/dipper-service/src/network/service/liveness_checker.rs index 2e8fcba0..7f0897a5 100644 --- a/bin/dipper-service/src/network/service/liveness_checker.rs +++ b/bin/dipper-service/src/network/service/liveness_checker.rs @@ -344,7 +344,7 @@ async fn probe_indexer( /// Whether so many of a cycle's indexers are unreachable at once that the fault is likely /// dipper's own: at least `trip_fraction` of them, out of at least `min_indexers` checked. -fn unreachable_trips( +pub(super) fn unreachable_trips( checked: usize, unreachable: usize, trip_fraction: f64, @@ -393,7 +393,7 @@ impl UnreachablePolicy { /// Cycles in a row that must see too few indexers unreachable to trip the breaker before it /// clears, so a share hovering around the trip fraction doesn't trip and clear it over and over. -const UNREACHABLE_CLEAR_CYCLES: u32 = 3; +pub(super) const UNREACHABLE_CLEAR_CYCLES: u32 = 3; /// How often the alert repeats while the breaker stays tripped. const UNREACHABLE_ALERT_REPEAT: time::Duration = time::Duration::hours(6); diff --git a/bin/dipper-service/src/registry.rs b/bin/dipper-service/src/registry.rs index bb07b366..a2f0b9e0 100644 --- a/bin/dipper-service/src/registry.rs +++ b/bin/dipper-service/src/registry.rs @@ -1,6 +1,7 @@ mod agreement; #[cfg(test)] mod agreement_stub; +mod dips_info; mod indexer_denylist; mod indexing_request; mod pending_cancellation; @@ -28,6 +29,7 @@ pub use self::{ Status as IndexingAgreementStatus, Terms as IndexingAgreementTerms, TermsMetadata as IndexingAgreementTermsMetadata, }, + dips_info::{DipsInfoChecks, DipsInfoRegistry}, indexer_denylist::IndexerDenylistRegistry, indexing_request::{ IndexingRequest, IndexingRequestRegistry, SetTargetOutcome, Status as IndexingRequestStatus, @@ -849,6 +851,44 @@ impl IndexerDenylistRegistry for RegistryProvider { } } +#[async_trait] +impl DipsInfoRegistry for RegistryProvider { + async fn get_indexers_with_accepted_agreements(&self) -> RegistryResult> { + self.inner + .get_indexers_with_accepted_agreements() + .await + .map_err(Into::into) + } + + async fn record_dips_info_checks( + &self, + missing: &[IndexerId], + added: std::time::Duration, + ) -> RegistryResult { + // The watcher adds at most 2 intervals, and its config keeps an interval to 1 day. + let added_secs = i64::try_from(added.as_secs()).unwrap_or(i64::MAX); + let checks = self + .inner + .record_dips_info_checks(missing, added_secs) + .await?; + let to_durations = |times: std::collections::HashMap| { + times + .into_iter() + .map(|(id, secs)| { + ( + id, + std::time::Duration::from_secs(secs.max(0).unsigned_abs()), + ) + }) + .collect() + }; + Ok(DipsInfoChecks { + missing: to_durations(checks.missing), + cleared: to_durations(checks.cleared), + }) + } +} + #[async_trait] impl PendingCancellationRegistry for RegistryProvider { async fn get_pending_cancellations_by_new_agreement( diff --git a/bin/dipper-service/src/registry/dips_info.rs b/bin/dipper-service/src/registry/dips_info.rs new file mode 100644 index 00000000..3cf16694 --- /dev/null +++ b/bin/dipper-service/src/registry/dips_info.rs @@ -0,0 +1,31 @@ +//! DIPs info watcher registry trait. + +use std::{collections::HashMap, time::Duration}; + +use async_trait::async_trait; +use thegraph_core::IndexerId; + +use super::result::Result as RegistryResult; + +/// What recording a cycle of DIPs info checks left: how long each indexer still missing its +/// DIPs info has gone without a readable answer, and the time each forgotten one had built up. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct DipsInfoChecks { + pub missing: HashMap, + pub cleared: HashMap, +} + +/// Trait for tracking how long each agreement holder's DIPs info has been missing. +#[async_trait] +pub trait DipsInfoRegistry { + /// Get every indexer holding at least 1 agreement accepted on-chain, each once. + async fn get_indexers_with_accepted_agreements(&self) -> RegistryResult>; + + /// Record 1 cycle of checks: add `added` to the time each of `missing` has gone without a + /// readable answer, and forget every other indexer, which starts its count again from 0. + async fn record_dips_info_checks( + &self, + missing: &[IndexerId], + added: Duration, + ) -> RegistryResult; +} diff --git a/dipper-pgregistry/migrations/20261030000000_add_indexer_dips_info_missing.sql b/dipper-pgregistry/migrations/20261030000000_add_indexer_dips_info_missing.sql new file mode 100644 index 00000000..c9a188a1 --- /dev/null +++ b/dipper-pgregistry/migrations/20261030000000_add_indexer_dips_info_missing.sql @@ -0,0 +1,10 @@ +-- Table: dipper_indexer_dips_info_missing +-- How long each indexer holding an accepted agreement has gone without a readable /dips/info +-- answer, counted only over the checks dipper could judge. A readable answer removes the row, +-- so a row exists only while the indexer's DIPs info is missing. +CREATE TABLE dipper_indexer_dips_info_missing +( + indexer_id BYTEA PRIMARY KEY, + missing_secs BIGINT NOT NULL CHECK (missing_secs >= 0), + updated_at TIMESTAMPTZ NOT NULL DEFAULT timezone('UTC', now()) +); diff --git a/dipper-pgregistry/src/lib.rs b/dipper-pgregistry/src/lib.rs index c78ed6c0..7d0fb6a6 100644 --- a/dipper-pgregistry/src/lib.rs +++ b/dipper-pgregistry/src/lib.rs @@ -20,9 +20,9 @@ pub use indexing_request::{ Status as IndexingRequestStatus, }; pub use postgres::{ - CancelKind, CancellingAgreement, ChainListenerStateRow, EndCorrection, NewAgreementParams, - PendingAcceptedEvent, PendingExpiredEvent, PendingTerminatedEvent, PgRegistry, - ReconciliationAudit, ReconciliationItem, ReconciliationOutcome, RecordedEnd, + CancelKind, CancellingAgreement, ChainListenerStateRow, DipsInfoChecks, EndCorrection, + NewAgreementParams, PendingAcceptedEvent, PendingExpiredEvent, PendingTerminatedEvent, + PgRegistry, ReconciliationAudit, ReconciliationItem, ReconciliationOutcome, RecordedEnd, }; pub use result::{Error, Result}; diff --git a/dipper-pgregistry/src/postgres.rs b/dipper-pgregistry/src/postgres.rs index 3a36f0bf..e361e37d 100644 --- a/dipper-pgregistry/src/postgres.rs +++ b/dipper-pgregistry/src/postgres.rs @@ -103,6 +103,14 @@ pub struct RecordedEnd { pub canceled_tx: Option, } +/// What recording a cycle of DIPs info checks left, in seconds without a readable answer: each +/// indexer still missing its DIPs info, and each one forgotten, with the time it had built up. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct DipsInfoChecks { + pub missing: HashMap, + pub cleared: HashMap, +} + /// An announced end that a write replaced or cleared, so its `terminated` event goes out again. #[derive(Debug, Clone, PartialEq, Eq)] pub struct EndCorrection { @@ -2196,6 +2204,72 @@ impl PgRegistry { Ok(rows.into_iter().map(|(id,)| id.0).collect()) } + // ========================================================================= + // DIPs info watcher operations + // ========================================================================= + + /// Get every indexer holding at least 1 `AcceptedOnChain` agreement, each once. + pub async fn get_indexers_with_accepted_agreements(&self) -> Result, Error> { + let rows: Vec<(PgIndexerId,)> = sqlx::query_as( + r#" + SELECT DISTINCT indexer_id + FROM dipper_reg_indexing_agreements + WHERE status = $1 + "#, + ) + .bind(IndexingAgreementStatus::AcceptedOnChain) + .fetch_all(&self.pool) + .await?; + + Ok(rows.into_iter().map(|(id,)| id.0).collect()) + } + + /// Record 1 cycle of DIPs info checks: add `added_secs` to the time each of `missing` has + /// gone without a readable answer, and forget every other indexer, so a readable answer, or + /// no longer being checked, starts its count again from 0. + pub async fn record_dips_info_checks( + &self, + missing: &[IndexerId], + added_secs: i64, + ) -> Result { + let pg_missing: Vec = missing.iter().map(|id| PgIndexerId(*id)).collect(); + let mut tx = self.pool.begin().await?; + + let cleared: Vec<(PgIndexerId, i64)> = sqlx::query_as( + r#" + DELETE FROM dipper_indexer_dips_info_missing + WHERE NOT (indexer_id = ANY($1)) + RETURNING indexer_id, missing_secs + "#, + ) + .bind(&pg_missing[..]) + .fetch_all(&mut *tx) + .await?; + + let still_missing: Vec<(PgIndexerId, i64)> = sqlx::query_as( + r#" + INSERT INTO dipper_indexer_dips_info_missing (indexer_id, missing_secs) + SELECT indexer_id, $2 FROM unnest($1::bytea[]) AS t(indexer_id) + ON CONFLICT (indexer_id) DO UPDATE SET + missing_secs = dipper_indexer_dips_info_missing.missing_secs + + EXCLUDED.missing_secs, + updated_at = timezone('UTC', now()) + RETURNING indexer_id, missing_secs + "#, + ) + .bind(&pg_missing[..]) + .bind(added_secs) + .fetch_all(&mut *tx) + .await?; + + tx.commit().await?; + + Ok(DipsInfoChecks { + missing: still_missing.into_iter().map(|(id, s)| (id.0, s)).collect(), + cleared: cleared.into_iter().map(|(id, s)| (id.0, s)).collect(), + }) + } + // ========================================================================= // Optimistic DIPs fees // ========================================================================= diff --git a/dipper-pgregistry/tests/it_registry_postgres.rs b/dipper-pgregistry/tests/it_registry_postgres.rs index a09aa096..febf486d 100644 --- a/dipper-pgregistry/tests/it_registry_postgres.rs +++ b/dipper-pgregistry/tests/it_registry_postgres.rs @@ -1,6 +1,6 @@ #![cfg(feature = "fake")] -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use dipper_core::ids::{IndexingAgreementId, IndexingRequestId}; use dipper_pgregistry::{ @@ -1709,6 +1709,117 @@ async fn get_open_indexing_requests_for_reassessment_zero_batch_returns_all() { assert_eq!(requests.len(), 3, "Expected all 3 eligible requests"); } +// ============================================================================= +// DIPs info watcher DB operation tests +// ============================================================================= + +#[tokio::test] +async fn get_indexers_with_accepted_agreements_lists_each_holder_once() { + //* Given + let (db, _temp_db) = temp_registry_db().await; + run_fixture( + &db, + include_str!("fixtures/0003_multi_indexer_agreements.sql"), + ) + .await + .expect("Failed to run fixture"); + // Indexer 0x1111 holds 2 accepted agreements and indexer 0x2222 holds 1; indexer 0x3333's + // only agreement expired. + run_fixture( + &db, + r#" + UPDATE dipper_reg_indexing_agreements SET status = 6 + WHERE id IN ('\xaa000000000000000000000000000001'::bytea, + '\xbb000000000000000000000000000001'::bytea); + "#, + ) + .await + .expect("Failed to set up accepted agreements"); + let registry = PgRegistry::new(db); + + //* When + let indexers: HashSet = registry + .get_indexers_with_accepted_agreements() + .await + .expect("Failed to get indexers with accepted agreements") + .into_iter() + .collect(); + + //* Then + assert_eq!( + indexers, + HashSet::from([ + indexer_id!("1111111111111111111111111111111111111111"), + indexer_id!("2222222222222222222222222222222222222222"), + ]) + ); +} + +#[tokio::test] +async fn record_dips_info_checks_adds_up_missing_time_until_an_indexer_answers() { + //* Given + let (db, _temp_db) = temp_registry_db().await; + let registry = PgRegistry::new(db); + let indexer_a = indexer_id!("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"); + let indexer_b = indexer_id!("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"); + + //* When + registry + .record_dips_info_checks(&[indexer_a, indexer_b], 300) + .await + .expect("Failed to record first cycle"); + let second = registry + .record_dips_info_checks(&[indexer_a, indexer_b], 600) + .await + .expect("Failed to record second cycle"); + let third = registry + .record_dips_info_checks(&[indexer_a], 300) + .await + .expect("Failed to record third cycle"); + + //* Then + assert_eq!(second.missing.get(&indexer_a), Some(&900)); + assert_eq!(second.missing.get(&indexer_b), Some(&900)); + assert!(second.cleared.is_empty()); + assert_eq!(third.missing, HashMap::from([(indexer_a, 1200)])); + assert_eq!( + third.cleared, + HashMap::from([(indexer_b, 900)]), + "indexer B answered, so its count is forgotten" + ); +} + +#[tokio::test] +async fn record_dips_info_checks_with_no_indexer_missing_forgets_them_all() { + //* Given + let (db, _temp_db) = temp_registry_db().await; + let registry = PgRegistry::new(db); + let indexer_a = indexer_id!("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"); + registry + .record_dips_info_checks(&[indexer_a], 300) + .await + .expect("Failed to record first cycle"); + + //* When + let checks = registry + .record_dips_info_checks(&[], 300) + .await + .expect("Failed to record second cycle"); + let after = registry + .record_dips_info_checks(&[indexer_a], 0) + .await + .expect("Failed to record third cycle"); + + //* Then + assert!(checks.missing.is_empty()); + assert_eq!(checks.cleared, HashMap::from([(indexer_a, 300)])); + assert_eq!( + after.missing, + HashMap::from([(indexer_a, 0)]), + "a forgotten indexer starts again from 0" + ); +} + // ============================================================================= // Liveness checker DB operation tests // ============================================================================= diff --git a/k8s/configmap-example.yaml b/k8s/configmap-example.yaml index e0ad4471..9b340cc2 100644 --- a/k8s/configmap-example.yaml +++ b/k8s/configmap-example.yaml @@ -103,6 +103,15 @@ data: "unreachable_trip_fraction": 0.5, "unreachable_trip_min_indexers": 3 }, + "dips_info_watcher": { + "enabled": true, + "interval": 300, + "alert_after_hours": 24, + "request_timeout": 10, + "max_concurrent_requests": 16, + "unreachable_trip_fraction": 0.5, + "unreachable_trip_min_indexers": 3 + }, "chain_client": { "enabled": true, "providers": [