From aa430ba419e5b4eb5d7ede6872baee735855cd80 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 30 Oct 2026 15:02:39 +0100 Subject: [PATCH 01/11] feat(registry): store how long each indexer's DIPs info has been missing A new table keeps, for each indexer holding an accepted agreement, how long its /dips/info has given no readable answer. Each cycle adds time for the missing ones and forgets the rest. --- ...30000000_add_indexer_dips_info_missing.sql | 10 ++ dipper-pgregistry/src/lib.rs | 6 +- dipper-pgregistry/src/postgres.rs | 74 ++++++++++++ .../tests/it_registry_postgres.rs | 113 +++++++++++++++++- 4 files changed, 199 insertions(+), 4 deletions(-) create mode 100644 dipper-pgregistry/migrations/20261030000000_add_indexer_dips_info_missing.sql 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 // ============================================================================= From e65810c5b4632b81d877733e4c64f2e60de1e27c Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 30 Oct 2026 15:02:39 +0100 Subject: [PATCH 02/11] feat(dipper): alert when an agreement holder's DIPs info stays unreadable IISA can't score an indexer without DIPs info, so nothing replaces it. A new watcher asks each agreement holder for its DIPs info every 5 minutes and alerts after 24 hours without an answer. --- bin/dipper-service/src/config.rs | 158 ++++++ bin/dipper-service/src/main.rs | 32 ++ bin/dipper-service/src/network/service.rs | 1 + .../src/network/service/dips_info_watcher.rs | 490 ++++++++++++++++++ .../src/network/service/liveness_checker.rs | 2 +- bin/dipper-service/src/registry.rs | 39 ++ bin/dipper-service/src/registry/dips_info.rs | 31 ++ k8s/configmap-example.yaml | 9 + 8 files changed, 761 insertions(+), 1 deletion(-) create mode 100644 bin/dipper-service/src/network/service/dips_info_watcher.rs create mode 100644 bin/dipper-service/src/registry/dips_info.rs diff --git a/bin/dipper-service/src/config.rs b/bin/dipper-service/src/config.rs index bbef7f05..6c3fe5a0 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,107 @@ 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, when none of them answers readably, makes + /// the cycle count toward nobody's 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() { + return Err("interval 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 +2016,57 @@ 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 { + 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 +2637,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..98cdf550 --- /dev/null +++ b/bin/dipper-service/src/network/service/dips_info_watcher.rs @@ -0,0 +1,490 @@ +//! 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::{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_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" + ); + + const DB_TIMEOUT: Duration = Duration::from_secs(30); + + 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() => {}, + } + let elapsed = cycles.start(Instant::now(), config.interval); + + let indexers = match tokio::time::timeout( + DB_TIMEOUT, + registry.get_indexers_with_accepted_agreements(), + ) + .await + { + Ok(Ok(indexers)) => indexers, + Ok(Err(err)) => { + tracing::error!(error = %err, "failed to query indexers with accepted agreements"); + continue; + } + Err(_) => { + tracing::error!("timeout querying indexers with accepted agreements"); + continue; + } + }; + + let missing = missing_dips_info( + &network, + &http_client, + &indexers, + config.max_concurrent_requests, + ) + .await; + let added = cycles.judge(indexers.len(), missing.len(), elapsed, &config); + + let checks = match tokio::time::timeout( + DB_TIMEOUT, + registry.record_dips_info_checks(&missing, added), + ) + .await + { + Ok(Ok(checks)) => checks, + Ok(Err(err)) => { + tracing::error!(error = %err, "failed to record DIPs info checks"); + continue; + } + Err(_) => { + tracing::error!("timeout recording DIPs info checks"); + continue; + } + }; + report(&checks, added, config.alert_after()); + } + + tracing::debug!("DIPs info watcher stopped"); + Ok(()) + }; + + (Handle { tx_stop }, service) +} + +/// What the watcher remembers between cycles: when the last one started, and whether it was +/// paused because too many indexers gave no readable answer at once. +#[derive(Default)] +struct Cycles { + last_started: Option, + paused: bool, +} + +impl Cycles { + /// Note a cycle starting, returning the time since the last one, at most 2 intervals so a + /// stall doesn't count as missing time. The first cycle after startup counts nothing. + fn start(&mut self, now: Instant, interval: Duration) -> Duration { + let elapsed = self + .last_started + .map_or(Duration::ZERO, |at| now.saturating_duration_since(at)); + self.last_started = Some(now); + elapsed.min(interval.saturating_mul(2)) + } + + /// 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. + fn judge( + &mut self, + checked: usize, + missing: usize, + elapsed: Duration, + config: &DipsInfoWatcherConfig, + ) -> Duration { + let trips = unreachable_trips( + checked, + missing, + config.unreachable_trip_fraction, + config.unreachable_trip_min_indexers, + ); + if trips && !self.paused { + tracing::warn!( + event = "dips_info_checks_paused", + missing, + checked, + "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" + ); + } else if !trips && self.paused { + tracing::info!(missing, checked, "Counting missing DIPs info again"); + } + self.paused = trips; + if trips { Duration::ZERO } else { elapsed } + } +} + +/// Log what a recorded cycle means: an alert for each indexer whose missing time passed +/// `alert_after` in it, and a note for each one that had been alerted on and has answered again. +fn report(checks: &DipsInfoChecks, added: Duration, alert_after: Duration) { + for (indexer_id, missing) in &checks.missing { + if crossed(*missing, added, alert_after) { + tracing::warn!( + event = "dips_info_missing", + indexer_id = %indexer_id, + missing_hours = missing.as_secs() / 3600, + "Indexer holding an accepted agreement has given no readable /dips/info answer \ + for over missing_hours; IISA keeps it in its groups but can't score it" + ); + } + } + 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 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 permits = Arc::new(Semaphore::new(max_concurrent.max(1))); + let mut asks = JoinSet::new(); + for indexer_id in indexers { + let indexer_id = *indexer_id; + let url = network + .get_indexer_by_id(&indexer_id) + .map(|indexer| indexer.url); + let http_client = http_client.clone(); + let permits = permits.clone(); + asks.spawn(async move { + let _permit = permits.acquire_owned().await; + let answer = match url { + Some(url) => ask_dips_info(&http_client, &url).await, + None => Err("not in the network registry".to_string()), + }; + (indexer_id, answer) + }); + } + + let mut missing = Vec::new(); + while let Some(asked) = asks.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) => tracing::error!(error = %err, "DIPs info check task failed"), + } + } + 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_up_to_2_intervals() { + let interval = Duration::from_secs(300); + let start = Instant::now(); + let mut cycles = Cycles::default(); + + assert_eq!(cycles.start(start, interval), Duration::ZERO, "first cycle"); + assert_eq!( + cycles.start(start + Duration::from_secs(310), interval), + Duration::from_secs(310) + ); + assert_eq!( + cycles.start(start + 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); + + assert_eq!(cycles.judge(4, 1, elapsed, &config), elapsed); + assert_eq!( + cycles.judge(4, 2, elapsed, &config), + Duration::ZERO, + "half missing" + ); + assert!(cycles.paused); + assert_eq!( + cycles.judge(4, 1, elapsed, &config), + elapsed, + "counting resumes" + ); + assert!(!cycles.paused); + assert_eq!( + cycles.judge(2, 2, elapsed, &config), + elapsed, + "too few indexers checked for the share to count" + ); + } + + #[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 report_alerts_once_an_indexer_reaches_the_limit() { + let reached = indexer_id!("1111111111111111111111111111111111111111"); + let short = indexer_id!("2222222222222222222222222222222222222222"); + let checks = DipsInfoChecks { + missing: HashMap::from([(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(&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..681bfef6 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, diff --git a/bin/dipper-service/src/registry.rs b/bin/dipper-service/src/registry.rs index bb07b366..a74e06f5 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,43 @@ 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 { + 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/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": [ From b0882b221ea2a244c8759a272485cbf9fb900f16 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 30 Oct 2026 15:11:25 +0100 Subject: [PATCH 03/11] fix(dipper): repeat the paused-counting alert every 6 hours Counting missing DIPs info can stay paused for as long as most agreement holders lack it, as after a release that changes /dips/info, so its alert now repeats instead of firing once. --- .../src/network/service/dips_info_watcher.rs | 75 +++++++++++++++---- 1 file changed, 59 insertions(+), 16 deletions(-) diff --git a/bin/dipper-service/src/network/service/dips_info_watcher.rs b/bin/dipper-service/src/network/service/dips_info_watcher.rs index 98cdf550..4caa4534 100644 --- a/bin/dipper-service/src/network/service/dips_info_watcher.rs +++ b/bin/dipper-service/src/network/service/dips_info_watcher.rs @@ -80,7 +80,8 @@ where _ = rx_stop.recv() => break, _ = timer.tick() => {}, } - let elapsed = cycles.start(Instant::now(), config.interval); + let now = Instant::now(); + let elapsed = cycles.start(now, config.interval); let indexers = match tokio::time::timeout( DB_TIMEOUT, @@ -106,7 +107,7 @@ where config.max_concurrent_requests, ) .await; - let added = cycles.judge(indexers.len(), missing.len(), elapsed, &config); + let added = cycles.judge(indexers.len(), missing.len(), elapsed, now, &config); let checks = match tokio::time::timeout( DB_TIMEOUT, @@ -134,12 +135,16 @@ where (Handle { tx_stop }, service) } -/// What the watcher remembers between cycles: when the last one started, and whether it was -/// paused because too many indexers gave no readable answer at once. +/// 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 one started, and since when, and when +/// last alerted on, counting has been paused because too many indexers gave no readable answer. #[derive(Default)] struct Cycles { last_started: Option, - paused: bool, + paused_since: Option, + pause_alerted_at: Option, } impl Cycles { @@ -155,11 +160,13 @@ impl Cycles { /// 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( @@ -168,20 +175,34 @@ impl Cycles { config.unreachable_trip_fraction, config.unreachable_trip_min_indexers, ); - if trips && !self.paused { + if !trips { + if self.paused_since.take().is_some() { + tracing::info!(missing, checked, "Counting missing DIPs info again"); + } + return elapsed; + } + let since = *self.paused_since.get_or_insert(now); + 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" ); - } else if !trips && self.paused { - tracing::info!(missing, checked, "Counting missing DIPs info again"); } - self.paused = trips; - if trips { Duration::ZERO } else { elapsed } + Duration::ZERO + } + + /// Whether counting is paused. + fn paused(&self) -> bool { + self.paused_since.is_some() } } @@ -396,27 +417,49 @@ mod tests { 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, &config), elapsed); + assert_eq!(cycles.judge(4, 1, elapsed, now, &config), elapsed); assert_eq!( - cycles.judge(4, 2, elapsed, &config), + cycles.judge(4, 2, elapsed, now, &config), Duration::ZERO, "half missing" ); - assert!(cycles.paused); + assert!(cycles.paused()); assert_eq!( - cycles.judge(4, 1, elapsed, &config), + cycles.judge(4, 1, elapsed, now, &config), elapsed, "counting resumes" ); - assert!(!cycles.paused); + assert!(!cycles.paused()); assert_eq!( - cycles.judge(2, 2, elapsed, &config), + 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 From 8c91b887d0caa31585f527a9f3897e69eb2c6998 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 30 Oct 2026 15:12:28 +0100 Subject: [PATCH 04/11] fix(dipper): resume counting missing DIPs info only after 3 calm cycles A missing share hovering around the trip fraction paused and resumed counting every cycle, repeating its alert. Counting now resumes after 3 calm cycles in a row. --- .../src/network/service/dips_info_watcher.rs | 37 ++++++++++++++++--- .../src/network/service/liveness_checker.rs | 2 +- 2 files changed, 32 insertions(+), 7 deletions(-) diff --git a/bin/dipper-service/src/network/service/dips_info_watcher.rs b/bin/dipper-service/src/network/service/dips_info_watcher.rs index 4caa4534..5a4b5425 100644 --- a/bin/dipper-service/src/network/service/dips_info_watcher.rs +++ b/bin/dipper-service/src/network/service/dips_info_watcher.rs @@ -11,7 +11,7 @@ use tokio::{ time::{Instant, MissedTickBehavior}, }; -use super::liveness_checker::unreachable_trips; +use super::liveness_checker::{UNREACHABLE_CLEAR_CYCLES, unreachable_trips}; use crate::{ config::DipsInfoWatcherConfig, network::provider::NetworkProviderService, @@ -145,6 +145,8 @@ struct Cycles { last_started: 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 { @@ -175,13 +177,24 @@ impl Cycles { config.unreachable_trip_fraction, config.unreachable_trip_min_indexers, ); - if !trips { - if self.paused_since.take().is_some() { + // 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"); } - return elapsed; } - let since = *self.paused_since.get_or_insert(now); + 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) @@ -426,10 +439,22 @@ mod tests { "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" + "counting resumes on the 3rd calm cycle" ); assert!(!cycles.paused()); assert_eq!( diff --git a/bin/dipper-service/src/network/service/liveness_checker.rs b/bin/dipper-service/src/network/service/liveness_checker.rs index 681bfef6..7f0897a5 100644 --- a/bin/dipper-service/src/network/service/liveness_checker.rs +++ b/bin/dipper-service/src/network/service/liveness_checker.rs @@ -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); From 63469d26b8b99b9f3b40249cd2c850af8707cc0f Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 30 Oct 2026 15:13:05 +0100 Subject: [PATCH 05/11] fix(dipper): name every indexer missing DIPs info in 1 alert per cycle Slack posts 1 message per event in each throttle window and only counts the rest, so indexers reaching the limit together lost their IDs. 1 alert per cycle now names all of them. --- .../src/network/service/dips_info_watcher.rs | 53 +++++++++++++------ 1 file changed, 37 insertions(+), 16 deletions(-) diff --git a/bin/dipper-service/src/network/service/dips_info_watcher.rs b/bin/dipper-service/src/network/service/dips_info_watcher.rs index 5a4b5425..e000f207 100644 --- a/bin/dipper-service/src/network/service/dips_info_watcher.rs +++ b/bin/dipper-service/src/network/service/dips_info_watcher.rs @@ -219,19 +219,25 @@ impl Cycles { } } -/// Log what a recorded cycle means: an alert for each indexer whose missing time passed -/// `alert_after` in it, and a note for each one that had been alerted on and has answered again. +/// 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) { - for (indexer_id, missing) in &checks.missing { - if crossed(*missing, added, alert_after) { - tracing::warn!( - event = "dips_info_missing", - indexer_id = %indexer_id, - missing_hours = missing.as_secs() / 3600, - "Indexer holding an accepted agreement has given no readable /dips/info answer \ - for over missing_hours; IISA keeps it in its groups but can't score it" - ); - } + 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 { @@ -450,7 +456,10 @@ mod tests { 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!( + cycles.paused(), + "a missing share back up starts the calm cycles again" + ); assert_eq!( cycles.judge(4, 1, elapsed, now, &config), elapsed, @@ -473,7 +482,13 @@ mod tests { 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); + cycles.judge( + 4, + 4, + elapsed, + start + Duration::from_secs(minutes * 60), + &config, + ); } }) .await; @@ -541,11 +556,16 @@ mod tests { } #[tokio::test] - async fn report_alerts_once_an_indexer_reaches_the_limit() { + 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), (short, 23 * HOUR)]), + missing: HashMap::from([ + (reached, 24 * HOUR), + (also_reached, 24 * HOUR), + (short, 23 * HOUR), + ]), cleared: HashMap::new(), }; @@ -553,6 +573,7 @@ mod tests { 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}"); } } From 657fcd30d3f7203c47b8433a76c11939b39cfb8a Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 30 Oct 2026 15:14:02 +0100 Subject: [PATCH 06/11] fix(dipper): keep a failed cycle's time for the next DIPs info check A cycle that failed to read or write the database threw away the time since the last one, so missing DIPs info added up slower. That time now carries into the next recorded cycle. --- .../src/network/service/dips_info_watcher.rs | 49 ++++++++++++------- 1 file changed, 32 insertions(+), 17 deletions(-) diff --git a/bin/dipper-service/src/network/service/dips_info_watcher.rs b/bin/dipper-service/src/network/service/dips_info_watcher.rs index e000f207..a2288d69 100644 --- a/bin/dipper-service/src/network/service/dips_info_watcher.rs +++ b/bin/dipper-service/src/network/service/dips_info_watcher.rs @@ -81,7 +81,7 @@ where _ = timer.tick() => {}, } let now = Instant::now(); - let elapsed = cycles.start(now, config.interval); + let elapsed = cycles.elapsed(now, config.interval); let indexers = match tokio::time::timeout( DB_TIMEOUT, @@ -125,6 +125,7 @@ where continue; } }; + cycles.recorded(now); report(&checks, added, config.alert_after()); } @@ -138,11 +139,11 @@ where /// 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 one started, and since when, and when -/// last alerted on, counting has been paused because too many indexers gave no readable answer. +/// 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_started: Option, + last_recorded: Option, paused_since: Option, pause_alerted_at: Option, /// Cycles in a row, while paused, with too few indexers missing to pause it. @@ -150,14 +151,18 @@ struct Cycles { } impl Cycles { - /// Note a cycle starting, returning the time since the last one, at most 2 intervals so a - /// stall doesn't count as missing time. The first cycle after startup counts nothing. - fn start(&mut self, now: Instant, interval: Duration) -> Duration { - let elapsed = self - .last_started - .map_or(Duration::ZERO, |at| now.saturating_duration_since(at)); - self.last_started = Some(now); - elapsed.min(interval.saturating_mul(2)) + /// 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 @@ -414,18 +419,28 @@ mod tests { } #[test] - fn a_cycle_counts_the_time_since_the_last_up_to_2_intervals() { + 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.start(start, interval), Duration::ZERO, "first cycle"); assert_eq!( - cycles.start(start + Duration::from_secs(310), interval), - Duration::from_secs(310) + 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.start(start + Duration::from_secs(5000), interval), + cycles.elapsed(next + Duration::from_secs(5000), interval), Duration::from_secs(600), "a stall counts as 2 intervals at most" ); From 95fc4b254e29ce858b7cbc46d427c1bb15fb9b1e Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 30 Oct 2026 15:15:48 +0100 Subject: [PATCH 07/11] fix(dipper): let a stop end a DIPs info check cycle midway A stop waited for every /dips/info request in the cycle, up to 10 seconds each, well past the 5 seconds shutdown allows each service. It now drops the cycle before anything is recorded. --- .../src/network/service/dips_info_watcher.rs | 183 +++++++++++++----- 1 file changed, 134 insertions(+), 49 deletions(-) diff --git a/bin/dipper-service/src/network/service/dips_info_watcher.rs b/bin/dipper-service/src/network/service/dips_info_watcher.rs index a2288d69..c40682a2 100644 --- a/bin/dipper-service/src/network/service/dips_info_watcher.rs +++ b/bin/dipper-service/src/network/service/dips_info_watcher.rs @@ -65,8 +65,6 @@ where "DIPs info watcher started" ); - const DB_TIMEOUT: Duration = Duration::from_secs(30); - let mut timer = tokio::time::interval(config.interval); timer.set_missed_tick_behavior(MissedTickBehavior::Skip); let http_client = reqwest::Client::builder() @@ -80,53 +78,12 @@ where _ = rx_stop.recv() => break, _ = timer.tick() => {}, } - let now = Instant::now(); - let elapsed = cycles.elapsed(now, config.interval); - - let indexers = match tokio::time::timeout( - DB_TIMEOUT, - registry.get_indexers_with_accepted_agreements(), - ) - .await - { - Ok(Ok(indexers)) => indexers, - Ok(Err(err)) => { - tracing::error!(error = %err, "failed to query indexers with accepted agreements"); - continue; - } - Err(_) => { - tracing::error!("timeout querying indexers with accepted agreements"); - continue; - } - }; - - 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 checks = match tokio::time::timeout( - DB_TIMEOUT, - registry.record_dips_info_checks(&missing, added), - ) - .await - { - Ok(Ok(checks)) => checks, - Ok(Err(err)) => { - tracing::error!(error = %err, "failed to record DIPs info checks"); - continue; - } - Err(_) => { - tracing::error!("timeout recording DIPs info checks"); - continue; - } - }; - cycles.recorded(now); - report(&checks, added, config.alert_after()); + // 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"); @@ -136,6 +93,62 @@ where (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); @@ -570,6 +583,78 @@ mod tests { assert_eq!(missing, expected); } + /// 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"); From 096bd970cbf603a90db29a6d7ec1f58a7de8196f Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 30 Oct 2026 15:17:26 +0100 Subject: [PATCH 08/11] fix(dipper): count an indexer whose DIPs info check crashed as missing A check that crashed left its indexer off the missing list, so its built-up time was wiped as if it had answered. It now counts as missing, and the error log names the indexer. --- .../src/network/service/dips_info_watcher.rs | 72 ++++++++++++++----- 1 file changed, 56 insertions(+), 16 deletions(-) diff --git a/bin/dipper-service/src/network/service/dips_info_watcher.rs b/bin/dipper-service/src/network/service/dips_info_watcher.rs index c40682a2..5c01c3a4 100644 --- a/bin/dipper-service/src/network/service/dips_info_watcher.rs +++ b/bin/dipper-service/src/network/service/dips_info_watcher.rs @@ -2,7 +2,7 @@ //! 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::{future::Future, sync::Arc, time::Duration}; +use std::{collections::HashMap, future::Future, sync::Arc, time::Duration}; use thegraph_core::IndexerId; use tokio::{ @@ -232,6 +232,7 @@ impl Cycles { } /// Whether counting is paused. + #[cfg(test)] fn paused(&self) -> bool { self.paused_since.is_some() } @@ -273,42 +274,65 @@ 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 most `max_concurrent` at a time, returning those -/// that gave no readable answer. +/// 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 permits = Arc::new(Semaphore::new(max_concurrent.max(1))); - let mut asks = JoinSet::new(); - for indexer_id in indexers { - let indexer_id = *indexer_id; + let asks = indexers.iter().map(|indexer_id| { let url = network - .get_indexer_by_id(&indexer_id) + .get_indexer_by_id(indexer_id) .map(|indexer| indexer.url); let http_client = http_client.clone(); - let permits = permits.clone(); - asks.spawn(async move { - let _permit = permits.acquire_owned().await; - let answer = match url { + 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, answer) + } + }; + (*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) = asks.join_next().await { + 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) => tracing::error!(error = %err, "DIPs info check task failed"), + 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 @@ -583,6 +607,22 @@ mod tests { 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, From 35b0e47c06ec8cbed5a734f82487fca6f3fdb1e9 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 30 Oct 2026 15:17:49 +0100 Subject: [PATCH 09/11] fix(config): refuse a DIPs info watcher request timeout of 0 A timeout of 0 makes every /dips/info request fail at once, so every agreement holder would look as if its DIPs info were missing. Dipper now refuses to start with it. --- bin/dipper-service/src/config.rs | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/bin/dipper-service/src/config.rs b/bin/dipper-service/src/config.rs index 6c3fe5a0..ae2ebe57 100644 --- a/bin/dipper-service/src/config.rs +++ b/bin/dipper-service/src/config.rs @@ -699,6 +699,9 @@ impl DipsInfoWatcherConfig { if self.interval.is_zero() { return Err("interval must be above 0".to_string()); } + 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()); } @@ -2041,6 +2044,10 @@ mod tests { interval: Duration::ZERO, ..enabled.clone() }, + DipsInfoWatcherConfig { + request_timeout: Duration::ZERO, + ..enabled.clone() + }, DipsInfoWatcherConfig { alert_after_hours: 0, ..enabled.clone() From 500552faf0540586d2d8c96d7a26539c4d928d85 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 30 Oct 2026 15:18:13 +0100 Subject: [PATCH 10/11] fix(config): cap the DIPs info watcher's interval at 1 day A huge interval let a cycle's time overflow the stored missing-time count, which then failed to save on every cycle. Dipper now refuses an interval over 1 day. --- bin/dipper-service/src/config.rs | 11 +++++++++-- bin/dipper-service/src/registry.rs | 1 + 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/bin/dipper-service/src/config.rs b/bin/dipper-service/src/config.rs index ae2ebe57..a90604f7 100644 --- a/bin/dipper-service/src/config.rs +++ b/bin/dipper-service/src/config.rs @@ -696,8 +696,11 @@ impl DipsInfoWatcherConfig { if !self.enabled { return Ok(()); } - if self.interval.is_zero() { - return Err("interval must be above 0".to_string()); + 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()); @@ -2044,6 +2047,10 @@ mod tests { interval: Duration::ZERO, ..enabled.clone() }, + DipsInfoWatcherConfig { + interval: Duration::from_secs(24 * 3600 + 1), + ..enabled.clone() + }, DipsInfoWatcherConfig { request_timeout: Duration::ZERO, ..enabled.clone() diff --git a/bin/dipper-service/src/registry.rs b/bin/dipper-service/src/registry.rs index a74e06f5..a2f0b9e0 100644 --- a/bin/dipper-service/src/registry.rs +++ b/bin/dipper-service/src/registry.rs @@ -865,6 +865,7 @@ impl DipsInfoRegistry for RegistryProvider { 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 From fcd6458092250719f8b3131c863a12e062f2f17a Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Fri, 30 Oct 2026 15:18:30 +0100 Subject: [PATCH 11/11] docs(config): say plainly what the DIPs info watcher's pause share does The setting's description said the share applied when none of the indexers answered, which misread it. It now says that share answering unreadably at once pauses counting. --- bin/dipper-service/src/config.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/bin/dipper-service/src/config.rs b/bin/dipper-service/src/config.rs index a90604f7..78ff8550 100644 --- a/bin/dipper-service/src/config.rs +++ b/bin/dipper-service/src/config.rs @@ -662,9 +662,9 @@ pub struct DipsInfoWatcherConfig { #[serde(default = "default_dips_info_watcher_max_concurrent_requests")] pub max_concurrent_requests: usize, - /// Share of the indexers checked in a cycle that, when none of them answers readably, makes - /// the cycle count toward nobody's missing time, since so many points to a fault on dipper's - /// side or a release that changed the endpoint (default: 0.5). + /// 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,