diff --git a/bin/dipper-service/src/config.rs b/bin/dipper-service/src/config.rs index abab64f5..661ede0b 100644 --- a/bin/dipper-service/src/config.rs +++ b/bin/dipper-service/src/config.rs @@ -127,6 +127,9 @@ impl Default for AlertsConfig { "agreement_cancel_stuck", "rpc_blocks_refused", "nonce_gap_fill_failed", + "chain_listener_changes_skipped", + "stale_agreement_uncancellable", + "liveness_abandonment_paused", ] .map(str::to_owned) .to_vec(), @@ -536,9 +539,22 @@ pub struct LivenessCheckerConfig { #[serde(default = "default_liveness_checker_request_timeout")] pub request_timeout: Duration, - /// Maximum agreements to fetch per cycle (default: 500). + /// Agreements to check per cycle, those longest without progress (default: 500). Every + /// other indexer holding an accepted agreement adds 1 more, so all are asked each cycle. #[serde(default = "default_liveness_checker_batch_size")] pub batch_size: i64, + + /// Share of the indexers checked in a cycle that, when unreachable at once (status call + /// failed, or missing from the network registry), stops abandoning agreements for + /// unreachability, for at most `max_tolerance_days` days, since so many points to a fault on + /// dipper's side (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; with fewer, 1 indexer going + /// down is too large a share to tell apart from a fault on dipper's side (default: 3). + #[serde(default = "default_liveness_checker_unreachable_trip_min_indexers")] + pub unreachable_trip_min_indexers: usize, } fn default_liveness_checker_enabled() -> bool { @@ -561,6 +577,14 @@ fn default_liveness_checker_batch_size() -> i64 { 500 } +fn default_liveness_checker_unreachable_trip_fraction() -> f64 { + 0.5 +} + +fn default_liveness_checker_unreachable_trip_min_indexers() -> usize { + 3 +} + impl Default for LivenessCheckerConfig { fn default() -> Self { Self { @@ -569,6 +593,8 @@ impl Default for LivenessCheckerConfig { max_tolerance_days: default_liveness_checker_max_tolerance_days(), request_timeout: default_liveness_checker_request_timeout(), batch_size: default_liveness_checker_batch_size(), + unreachable_trip_fraction: default_liveness_checker_unreachable_trip_fraction(), + unreachable_trip_min_indexers: default_liveness_checker_unreachable_trip_min_indexers(), } } } @@ -2118,8 +2144,8 @@ mod tests { ); } - #[test] - fn the_example_configmap_alerts_section_parses() { + /// The config.json in the example configmap. + fn example_config() -> serde_json::Value { let example = include_str!("../../../k8s/configmap-example.yaml"); let json: String = example .split_once("config.json: |\n") @@ -2128,11 +2154,39 @@ mod tests { .lines() .map(|line| format!("{}\n", line.strip_prefix(" ").unwrap_or(line))) .collect(); - let config: serde_json::Value = serde_json::from_str(&json).expect("config.json is JSON"); + serde_json::from_str(&json).expect("config.json is JSON") + } + + #[test] + fn the_example_configmap_liveness_checker_section_parses() { + let liveness = serde_json::from_value::( + example_config()["liveness_checker"].clone(), + ) + .expect("the example's liveness_checker section is a valid liveness checker config"); + let defaults = LivenessCheckerConfig::default(); + + assert_eq!( + liveness.unreachable_trip_fraction, + defaults.unreachable_trip_fraction + ); + assert_eq!( + liveness.unreachable_trip_min_indexers, + defaults.unreachable_trip_min_indexers + ); + } + + #[test] + fn the_example_configmap_alerts_section_parses() { + let config = example_config(); let alerts = serde_json::from_value::(config["alerts"].clone()) .expect("the example's alerts section is a valid alerts config"); assert!(alerts.slack_webhook_url.is_some()); + assert_eq!( + alerts.events, + AlertsConfig::default().events, + "the example lists every event alerted on by default" + ); } #[test] diff --git a/bin/dipper-service/src/network/service/chain_listener.rs b/bin/dipper-service/src/network/service/chain_listener.rs index 3e90e78a..071c2e30 100644 --- a/bin/dipper-service/src/network/service/chain_listener.rs +++ b/bin/dipper-service/src/network/service/chain_listener.rs @@ -36,6 +36,7 @@ use std::{ time::{Duration, Instant}, }; +use anyhow::Context as _; use dipper_core::ids::IndexingAgreementId; // The concrete emitter is only constructed in tests (`disabled()`); the runtime // paths depend on the `dyn` trait, so keep this import test-gated. @@ -108,6 +109,24 @@ fn hold_expired(hold: &mut Option, cursor: &Cursor, now: std::time:: } } } + +/// Most agreement ids a log line lists, so an alert for a whole page that failed stays readable. +const MAX_LOGGED_IDS: usize = 20; + +/// Agreement ids in the 0x form operators search for, at most [`MAX_LOGGED_IDS`] of them. +fn listed_ids(ids: &[IndexingAgreementId]) -> String { + let shown: Vec = ids + .iter() + .take(MAX_LOGGED_IDS) + .map(ToString::to_string) + .collect(); + let more = ids.len().saturating_sub(MAX_LOGGED_IDS); + if more == 0 { + shown.join(", ") + } else { + format!("{} and {more} more", shown.join(", ")) + } +} /// Cap on the cancellation-sweep batch so a backlog drains across polls /// instead of blocking one tick. const SWEEP_BATCH_SIZE: i64 = 1000; @@ -623,7 +642,7 @@ where Ok(None) => {} Err(err) => { tracing::warn!( - error = %err, + error = format!("{err:#}"), agreement_id = %snapshot.agreement_id, "Failed to prepare reconciliation for snapshot" ); @@ -723,7 +742,8 @@ where tracing::error!( event = "chain_listener_changes_skipped", cursor_block = cursor.block, - agreement_ids = ?unapplied, + agreement_count = unapplied.len(), + agreement_ids = %listed_ids(&unapplied), "Skipped on-chain changes to agreements that kept failing to apply; they won't \ be read again until their next change" ); @@ -731,7 +751,8 @@ where tracing::error!( event = "chain_listener_cursor_held", cursor_block = cursor.block, - agreement_ids = ?unapplied, + agreement_count = unapplied.len(), + agreement_ids = %listed_ids(&unapplied), "Failed to apply on-chain changes to agreements; re-reading them next poll" ); } @@ -907,30 +928,22 @@ where indexer = %snapshot.indexer, "Rejected agreement accepted on-chain, cancelling it" ); - // This row goes Rejected -> Canceled without ever transiting - // AcceptedOnChain, so `apply_reconciliation` never records the accept. - // Persist it from the snapshot so the eventual `terminated` (emitted by - // the sweep after the cancel) is eligible (`accepted_at IS NOT NULL`) and - // carries accurate accept data. One left Rejected, as the chain shows it - // already ended, gets no accept on record, so none is announced. + // It never passes AcceptedOnChain, so its accept is recorded here for its end to be + // announced; one left Rejected, as the chain shows it already ended, gets none. A failed + // write gets the snapshot read again, which finds it cancelling and records the accept. let reopened = crate::cancel_dispatch::reopen_if_live(registry, chain_client, &agreement).await?; - if reopened - && let Err(err) = registry + if reopened { + registry .record_accepted_audit(&agreement.id, snapshot.accepted_at, &snapshot.accepted_tx) .await - { - tracing::warn!( - agreement_id = %agreement.id, - error = %err, - "failed to record accepted audit for rejected-then-accepted agreement" - ); + .context("failed to record the on-chain accept of a rejected agreement")?; } } let reopened = reopen_if_cancelled_but_accepted(snapshot, &agreement, registry, chain_client).await?; - record_accept_of_cancelling(snapshot, &agreement, reopened, registry).await; + record_accept_of_cancelling(snapshot, &agreement, reopened, registry).await?; // Both transitions are applied atomically downstream so the // Accept-then-Cancel-in-one-snapshot path can't leak an intermediate @@ -989,26 +1002,24 @@ where status = %agreement.status, "Agreement already canceled, recording any accept dipper missed" ); - record_accept_and_cancel_from_chain(snapshot, &agreement, registry).await; + record_accept_and_cancel_from_chain(snapshot, &agreement, registry).await?; } Ok(None) } } -/// Record the accept and cancel of an agreement dipper had already marked -/// cancelled that went live on-chain first, so its accepted and terminated -/// events go out. Both go in 1 write, as the terminated sweep waits only for the accept. -/// Existing values win, so an agreement dipper already recorded is unchanged, -/// except an end recorded before the accept, such as its offer's withdrawal, which gives way. +/// Record the accept and end of an agreement dipper already ended that went live on-chain, in 1 +/// write, as the terminated sweep waits only for the accept; what is on record wins, but an end +/// before the accept gives way. The end is its last change on-chain, so a failure is returned. async fn record_accept_and_cancel_from_chain( snapshot: &AgreementStateSnapshot, agreement: &IndexingAgreement, registry: &R, -) { +) -> anyhow::Result<()> { if snapshot.accepted_at == 0 { - return; + return Ok(()); } - let recorded = registry + registry .record_accept_and_cancel_audit( &agreement.id, snapshot.accepted_at, @@ -1017,15 +1028,8 @@ async fn record_accept_and_cancel_from_chain( &snapshot.canceled_by.to_string(), Some(&snapshot.canceled_tx), ) - .await; - if let Err(err) = recorded { - tracing::warn!( - agreement_id = %agreement.id, - error = %err, - "failed to record the on-chain accept and cancel of a cancelled agreement; \ - its events go out only if the listener reads this agreement again" - ); - } + .await + .context("failed to record the on-chain accept and cancel of a cancelled agreement") } /// Safety net for an agreement dipper cancelled whose offer the indexer accepted @@ -1055,31 +1059,24 @@ where } /// An agreement dipper is cancelling, or has just moved back to cancelling, stays `Cancelling` -/// when the chain shows it accepted, so its accept is recorded here; its end is then announced, -/// along with the accept, once the cancel lands. A withdrawn offer reads as cancelled with no -/// accept time. +/// when the chain shows it accepted, so its accept is recorded here, to be announced with its +/// end once the cancel lands. A withdrawn offer reads as cancelled with no accept time. async fn record_accept_of_cancelling( snapshot: &AgreementStateSnapshot, agreement: &IndexingAgreement, reopened: bool, registry: &R, -) { +) -> anyhow::Result<()> { if (agreement.status != IndexingAgreementStatus::Cancelling && !reopened) || !snapshot.state.reached_accepted() || snapshot.accepted_at == 0 { - return; + return Ok(()); } - if let Err(err) = registry + registry .record_accepted_audit(&agreement.id, snapshot.accepted_at, &snapshot.accepted_tx) .await - { - tracing::warn!( - agreement_id = %agreement.id, - error = %err, - "failed to record the on-chain accept of an agreement being cancelled" - ); - } + .context("failed to record the on-chain accept of an agreement being cancelled") } /// Log the transition that landed and, on fresh accepts, fan out the @@ -1954,6 +1951,8 @@ mod tests { audit_writes: Vec<(&'static str, IndexingAgreementId)>, /// When true, writes that record a cancel fail. fail_cancel_audit: bool, + /// When true, writes that record only an accept fail. + fail_accept_audit: bool, pending_cancellations: std::collections::HashMap< IndexingAgreementId, Vec, @@ -2250,7 +2249,18 @@ mod tests { id: &IndexingAgreementId, _seen_live: bool, ) -> RegistryResult<()> { - self.state.lock().unwrap().reopened.push(*id); + let mut state = self.state.lock().unwrap(); + let Some(agreement) = state.agreements.get_mut(id).filter(|a| { + matches!( + a.status, + IndexingAgreementStatus::CanceledByRequester + | IndexingAgreementStatus::Rejected + ) + }) else { + return Err(crate::registry::Error::NoRecordsUpdated); + }; + agreement.status = IndexingAgreementStatus::Cancelling; + state.reopened.push(*id); Ok(()) } @@ -2295,11 +2305,15 @@ mod tests { _accepted_at: u64, _accepted_tx: &str, ) -> RegistryResult<()> { - self.state - .lock() - .unwrap() - .audit_writes - .push(("accept", *agreement_id)); + let mut state = self.state.lock().unwrap(); + if state.fail_accept_audit { + return Err(crate::registry::Error::BackendError( + dipper_pgregistry::Error::DbError(sqlx::Error::Protocol( + "simulated transient failure".into(), + )), + )); + } + state.audit_writes.push(("accept", *agreement_id)); Ok(()) } @@ -3275,27 +3289,6 @@ mod tests { ); } - #[tokio::test] - async fn test_reconcile_survives_a_failed_accept_and_cancel_record() { - let registry = MockRegistry::new(); - let chain_client = MockChainClient::default(); - let agreement_id = IndexingAgreementId::from_bytes(rand::random()); - registry.add_agreement(agreement_id, IndexingAgreementStatus::CanceledByRequester); - registry.state.lock().unwrap().fail_cancel_audit = true; - - let snapshot = make_snapshot(agreement_id, AgreementState::CanceledByPayer, Address::ZERO); - let result = reconcile_agreement( - &snapshot, - ®istry, - &chain_client, - test_agreement_conf().as_ref(), - ) - .await; - - assert!(result.is_ok(), "a failed record must not fail the snapshot"); - assert!(registry.audit_writes().is_empty()); - } - #[tokio::test] async fn test_reconcile_records_nothing_for_cancelled_agreement_never_accepted() { // A withdrawn offer was never accepted, so there is nothing to announce. @@ -4358,6 +4351,132 @@ mod tests { assert!(hold.is_none()); } + /// Drain a page holding a change in `state` to an agreement in `status` 3 times: with its + /// audit writes failing, then working, then from the start again, as the reorg buffer + /// re-reads it. The agreement, the registry and the cursor after each drain. + async fn drain_a_page_whose_audit_write_fails( + status: IndexingAgreementStatus, + state: AgreementState, + chain_client: &MockChainClient, + ) -> (IndexingAgreementId, MockRegistry, Vec) { + let registry = MockRegistry::new(); + let event_source = super::super::chain_events::mock::MockEventSource::new(); + event_source.set_latest_block(100); + event_source.set_latest_block_timestamp(Some(1_700_000_000)); + let agreement_id = IndexingAgreementId::from_bytes(rand::random()); + registry.add_agreement(agreement_id, status); + event_source.add_snapshots(vec![make_snapshot(agreement_id, state, Address::ZERO)]); + let mut cursor = Cursor::genesis(); + let mut cursor_hold = None; + let mut last_persisted_timestamp: Option = None; + let mut last_chain_ts_persist_wall = std::time::Instant::now(); + let mut last_subgraph_head: u64 = 0; + let mut stall_count: u32 = 0; + let mut consecutive_failures: u32 = 0; + let (_tx_stop, mut rx_stop) = mpsc::channel::<()>(1); + let mut cursors = Vec::new(); + + for (writes_fail, from_start) in [(true, false), (false, false), (false, true)] { + { + let mut mock = registry.state.lock().unwrap(); + mock.fail_accept_audit = writes_fail; + mock.fail_cancel_audit = writes_fail; + } + if from_start { + cursor = Cursor::genesis(); + } + drain_once( + &mut cursor, + &mut cursor_hold, + &mut last_persisted_timestamp, + &mut last_chain_ts_persist_wall, + &mut last_subgraph_head, + &mut stall_count, + &mut consecutive_failures, + 1337, + 0, + 10, + false, + ®istry, + chain_client, + &event_source, + &mut rx_stop, + test_agreement_conf().as_ref(), + ) + .await + .expect("subgraph fetch succeeded"); + cursors.push(cursor.clone()); + } + (agreement_id, registry, cursors) + } + + /// The end of an agreement dipper already ended is its last change on-chain, so if recording + /// it fails, only holding the cursor gets it read again. Reading it again once recorded only + /// repeats the write, which keeps what is already on record. + #[tokio::test] + async fn test_drain_rereads_an_accept_and_cancel_that_failed_to_record() { + let (id, registry, cursors) = drain_a_page_whose_audit_write_fails( + IndexingAgreementStatus::CanceledByRequester, + AgreementState::CanceledByPayer, + &MockChainClient::default(), + ) + .await; + + assert_eq!( + cursors[0], + Cursor::genesis(), + "held before the failed write" + ); + assert!( + cursors[1..] + .iter() + .all(|cursor| *cursor > Cursor::genesis()) + ); + assert_eq!(registry.audit_writes(), vec![("accept and cancel", id); 2]); + assert!(!registry.was_reopened(&id)); + assert!(!registry.was_marked_canceled_by_requester(&id)); + } + + /// The same holds for the accept of an agreement being cancelled, including a rejected or + /// cancelled one moved back to cancelling: read again, it is cancelling, so its accept is + /// recorded without moving it back a second time. + #[tokio::test] + async fn test_drain_rereads_the_accept_of_an_agreement_being_cancelled_that_failed_to_record() { + for (status, chain_client) in [ + ( + IndexingAgreementStatus::Cancelling, + MockChainClient::default(), + ), + (IndexingAgreementStatus::CanceledByRequester, live_chain()), + (IndexingAgreementStatus::Rejected, live_chain()), + ] { + let (id, registry, cursors) = drain_a_page_whose_audit_write_fails( + status, + AgreementState::Accepted, + &chain_client, + ) + .await; + + assert_eq!( + cursors[0], + Cursor::genesis(), + "{status}: held before the failed write" + ); + assert!( + cursors[1..] + .iter() + .all(|cursor| *cursor > Cursor::genesis()), + "{status}" + ); + assert_eq!(registry.audit_writes(), vec![("accept", id); 2], "{status}"); + assert_eq!( + registry.state.lock().unwrap().reopened.len(), + usize::from(status != IndexingAgreementStatus::Cancelling), + "{status}" + ); + } + } + #[test] fn holds_the_cursor_before_a_failing_page_for_a_while() { let now = std::time::Instant::now(); @@ -4377,6 +4496,24 @@ mod tests { assert!(hold.is_none()); } + #[test] + fn lists_at_most_20_agreement_ids_in_their_0x_form() { + let ids: Vec = (0..25u8) + .map(|n| IndexingAgreementId::from_bytes([n; 16])) + .collect(); + + assert_eq!( + listed_ids(&ids[1..3]), + "0x01010101010101010101010101010101, 0x02020202020202020202020202020202" + ); + let many = listed_ids(&ids); + assert_eq!(many.matches("0x").count(), 20, "{many}"); + assert!( + many.ends_with("0x13131313131313131313131313131313 and 5 more"), + "{many}" + ); + } + /// Drives a hostile timestamp drift through `drain_once` and asserts /// the persisted chain timestamp is capped near /// `CHAIN_TS_DRIFT_TOLERANCE_SECS` instead of accepting the huge diff --git a/bin/dipper-service/src/network/service/liveness_checker.rs b/bin/dipper-service/src/network/service/liveness_checker.rs index 6f1a87c6..bf0a59eb 100644 --- a/bin/dipper-service/src/network/service/liveness_checker.rs +++ b/bin/dipper-service/src/network/service/liveness_checker.rs @@ -30,13 +30,26 @@ //! - Indexer unreachable → treat as no progress (threshold check still applies) //! - Deployment missing from response → treat as no progress //! +//! When at least `unreachable_trip_fraction` of the indexers checked in a cycle can't be +//! reached, the fault is likely dipper's own, so no agreement is abandoned for unreachability +//! until fewer are, for at most `max_tolerance_days` days; indexers that answered without +//! progress are still handled as above. Once 3 cycles in a row see fewer, or the hold runs out, +//! an indexer still unreachable is abandoned only after a full tolerance without progress from +//! the last cycle that held it, since dipper couldn't see it. +//! //! ## URL Refresh //! //! Indexer URLs are looked up fresh from the network topology on each cycle, //! not read from the stored agreement. This ensures URL changes are detected. -use std::{collections::HashMap, future::Future, sync::Arc, time::Duration}; +use std::{ + collections::{HashMap, HashSet}, + future::Future, + sync::Arc, + time::Duration, +}; +use dipper_core::ids::IndexingAgreementId; use thegraph_core::{DeploymentId, IndexerId}; use time::OffsetDateTime; use tokio::{sync::mpsc, time::MissedTickBehavior}; @@ -131,6 +144,8 @@ where .timeout(config.request_timeout) .build() .map_err(|e| anyhow::anyhow!("failed to build HTTP client: {e}"))?; + let mut uncancellable = Uncancellable::default(); + let mut breaker = UnreachableBreaker::default(); loop { tokio::select! { @@ -186,133 +201,49 @@ where "liveness check: checking agreements" ); - // 3. Group agreements by indexer ID for batched status queries + // 3. Ask every indexer for its progress before acting on any answer, so the share + // that can't be reached is known before any agreement is abandoned for it. let groups = group_by_indexer_id(agreements); - + let mut probed = Vec::with_capacity(groups.len()); for (indexer_id, group_agreements) in groups { - // Check for shutdown between indexer groups if rx_stop.try_recv().is_ok() { tracing::debug!("liveness checker stopping mid-cycle"); return Ok(()); } + let probe = + probe_indexer(&network, &http_client, &indexer_id, &group_agreements).await; + probed.push((group_agreements, probe)); + } + let unreachable = probed + .iter() + .filter(|(_, probe)| matches!(probe, Probe::Unreachable)) + .count(); + let policy = breaker.evaluate( + probed.len(), + unreachable, + OffsetDateTime::now_utc(), + &config, + ); - // 4. Look up fresh URL from network topology - let indexer_url = match network.get_indexer_by_id(&indexer_id) { - Some(indexer) => indexer.url, - None => { - // Indexer no longer in network registry — treat as unreachable - tracing::warn!( - indexer_id = %indexer_id, - "indexer not found in network registry, treating as unreachable" - ); - // Fall through with empty block_heights to trigger threshold checks - process_agreements_with_no_data( - &group_agreements, - &active_counts, - &config, - ®istry, - &worker_queue, - &chain_client, - &agreement_conf, - DB_UPDATE_TIMEOUT, - QUEUE_PUSH_TIMEOUT, - ) - .await; - continue; - } - }; - - // 5. Query the indexer status endpoint for all deployments in this group - let deployment_ids: Vec = group_agreements - .iter() - .map(|a| a.terms.metadata.subgraph_deployment_id.to_string()) - .collect(); - - let block_heights = - match query_indexer_status(&http_client, &indexer_url, &deployment_ids).await { - Ok(heights) => heights, - Err(err) => { - tracing::warn!( - indexer_url = %indexer_url, - error = %err, - "failed to query indexer status, treating as unreachable" - ); - // Unreachable indexer: use empty map so all get current_block = None - HashMap::new() - } - }; - - // 6. Process each agreement in the group - for agreement in group_agreements { - let deployment_id_str = - agreement.terms.metadata.subgraph_deployment_id.to_string(); - let current_block = block_heights.get(&deployment_id_str).copied().flatten(); - - let now = OffsetDateTime::now_utc(); - - let action = decide_liveness_action( - agreement.last_block_height, - current_block, - agreement.last_progress_at, - now, - agreement.terms.metadata.subgraph_deployment_id, - &active_counts, - config.max_tolerance_days, - ); - - match action { - LivenessAction::InitializeTracking(block) - | LivenessAction::ResetProgress(block) => { - record_progress(®istry, &agreement, block, now, DB_UPDATE_TIMEOUT) - .await; - } - LivenessAction::WithinTolerance => { - let last_progress = agreement.last_progress_at.unwrap_or(now); - let elapsed = now - last_progress; - let threshold = tolerance_duration( - agreement.terms.metadata.subgraph_deployment_id, - &active_counts, - config.max_tolerance_days, - ); - tracing::debug!( - agreement_id = %agreement.id, - deployment = %deployment_id_str, - last_block = agreement.last_block_height.unwrap_or(0), - elapsed_hours = elapsed.whole_hours(), - threshold_hours = threshold.whole_hours(), - "agreement block unchanged but within tolerance" - ); - } - LivenessAction::CancelAndReassess => { - let last_progress = agreement.last_progress_at.unwrap_or(now); - let elapsed = now - last_progress; - let threshold = tolerance_duration( - agreement.terms.metadata.subgraph_deployment_id, - &active_counts, - config.max_tolerance_days, - ); - tracing::warn!( - agreement_id = %agreement.id, - indexer_url = %indexer_url, - deployment = %deployment_id_str, - last_block = agreement.last_block_height.unwrap_or(0), - elapsed_hours = elapsed.whole_hours(), - threshold_hours = threshold.whole_hours(), - "agreement stale: no indexing progress detected" - ); - cancel_and_reassess( - &agreement, - ®istry, - &worker_queue, - &chain_client, - &agreement_conf, - DB_UPDATE_TIMEOUT, - QUEUE_PUSH_TIMEOUT, - ) - .await; - } - } + // 4. Act on each indexer's agreements + let checker = Checker { + registry: ®istry, + worker_queue: &worker_queue, + chain_client: &chain_client, + agreement_conf: &agreement_conf, + max_tolerance_days: config.max_tolerance_days, + active_counts: &active_counts, + db_timeout: DB_UPDATE_TIMEOUT, + queue_timeout: QUEUE_PUSH_TIMEOUT, + }; + for (group_agreements, probe) in probed { + if rx_stop.try_recv().is_ok() { + tracing::debug!("liveness checker stopping mid-cycle"); + return Ok(()); } + checker + .check_indexer(&mut uncancellable, group_agreements, &probe, policy) + .await; } tracing::debug!("liveness check completed"); @@ -356,90 +287,376 @@ fn tolerance_duration( time::Duration::days(days as i64) } -/// Process agreements when we have no status data (indexer not in registry). -/// -/// All agreements get `current_block = None`, triggering threshold checks. -#[allow(clippy::too_many_arguments)] -async fn process_agreements_with_no_data( +/// What a cycle learned from 1 indexer's status endpoint. +enum Probe { + /// It answered, with the latest block it reports for each deployment asked about. + Reached(HashMap>), + /// It could not be asked: its status call failed, or it is missing from the network registry. + Unreachable, +} + +impl Probe { + /// The latest block the indexer reports for an agreement's deployment, if any. + fn block(&self, agreement: &IndexingAgreement) -> Option { + match self { + Self::Reached(heights) => heights + .get(&agreement.terms.metadata.subgraph_deployment_id.to_string()) + .copied() + .flatten(), + Self::Unreachable => None, + } + } +} + +/// Ask 1 indexer, at its current URL in the network registry, for its progress on the +/// deployments of its agreements. +async fn probe_indexer( + network: &NetworkProviderService, + http_client: &reqwest::Client, + indexer_id: &IndexerId, agreements: &[IndexingAgreement], - active_counts: &HashMap, - config: &LivenessCheckerConfig, - registry: &R, - worker_queue: &W, - chain_client: &C, - agreement_conf: &crate::config::IndexingAgreementConfig, +) -> Probe { + let Some(indexer) = network.get_indexer_by_id(indexer_id) else { + tracing::warn!( + indexer_id = %indexer_id, + "indexer not found in network registry, treating as unreachable" + ); + return Probe::Unreachable; + }; + let deployment_ids: Vec = agreements + .iter() + .map(|a| a.terms.metadata.subgraph_deployment_id.to_string()) + .collect(); + match query_indexer_status(http_client, &indexer.url, &deployment_ids).await { + Ok(heights) => Probe::Reached(heights), + Err(err) => { + tracing::warn!( + indexer_id = %indexer_id, + indexer_url = %indexer.url, + error = %err, + "failed to query indexer status, treating as unreachable" + ); + Probe::Unreachable + } + } +} + +/// 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( + checked: usize, + unreachable: usize, + trip_fraction: f64, + min_indexers: usize, +) -> bool { + checked > 0 && checked >= min_indexers && unreachable as f64 / checked as f64 >= trip_fraction +} + +/// How a cycle treats the agreements of an indexer it couldn't reach. +#[derive(Clone, Copy, Debug, PartialEq)] +enum UnreachablePolicy { + /// So many indexers can't be reached that the fault is likely dipper's: abandon none of them + /// for it, and leave their progress clocks alone. + Hold, + /// Judge them as usual, but count their time without progress from no earlier than `since`, + /// the last cycle that held them, since dipper couldn't see them while it did. + Judge { since: Option }, +} + +impl UnreachablePolicy { + /// When an unreachable indexer's agreement counts its time without progress from. + fn progress_clock(self, last_progress_at: Option) -> Option { + match self { + Self::Judge { since: Some(since) } => last_progress_at.map(|at| at.max(since)), + Self::Judge { since: None } | Self::Hold => last_progress_at, + } + } +} + +/// 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; + +/// How often the alert repeats while the breaker stays tripped. +const UNREACHABLE_ALERT_REPEAT: time::Duration = time::Duration::hours(6); + +/// Stops agreements being abandoned for unreachability while so many indexers can't be reached +/// at once that the fault is likely dipper's own (its network, or the network registry), which +/// would hand every deployment to new indexers and bar the current ones for weeks. +#[derive(Default)] +struct UnreachableBreaker { + /// When the current trip began; `None` while clear. + tripped_since: Option, + /// Cycles in a row, while tripped, with too few indexers unreachable to trip it. + quiet_cycles: u32, + /// The last cycle that held agreements back. + last_held: Option, + /// When an operator was last alerted to the current trip. + alerted_at: Option, +} + +impl UnreachableBreaker { + /// How this cycle treats agreements whose indexer can't be reached. A trip holds them for at + /// most `max_tolerance_days` days, so indexers that are really gone can't keep it tripped. + fn evaluate( + &mut self, + checked: usize, + unreachable: usize, + now: OffsetDateTime, + config: &LivenessCheckerConfig, + ) -> UnreachablePolicy { + let trips = unreachable_trips( + checked, + unreachable, + config.unreachable_trip_fraction, + config.unreachable_trip_min_indexers, + ); + if trips { + self.note_trip(now); + } else if self.tripped_since.is_some() { + self.note_quiet_cycle(unreachable, checked); + } + let judge = UnreachablePolicy::Judge { + since: self.last_held, + }; + let Some(since) = self.tripped_since else { + return judge; + }; + let max_hold = time::Duration::days(i64::from(config.max_tolerance_days.max(1))); + if now - since > max_hold { + self.give_up(unreachable, checked, now, max_hold); + return judge; + } + self.report_hold(trips, unreachable, checked, now, max_hold); + self.last_held = Some(now); + UnreachablePolicy::Hold + } + + /// Note a cycle with enough indexers unreachable to trip the breaker. + fn note_trip(&mut self, now: OffsetDateTime) { + self.quiet_cycles = 0; + if self.tripped_since.is_none() { + self.tripped_since = Some(now); + self.alerted_at = None; + } + } + + /// Note a cycle, while tripped, with too few indexers unreachable to trip it, clearing the + /// breaker once enough of those come in a row. + fn note_quiet_cycle(&mut self, unreachable: usize, checked: usize) { + self.quiet_cycles += 1; + if self.quiet_cycles < UNREACHABLE_CLEAR_CYCLES { + tracing::info!( + unreachable, + checked, + quiet_cycles = self.quiet_cycles, + "Fewer indexers unreachable; the breaker clears if it stays so" + ); + return; + } + self.tripped_since = None; + self.quiet_cycles = 0; + tracing::info!( + unreachable, + checked, + "Fewer indexers unreachable; abandoning agreements for unreachability resumes, each \ + after a full tolerance without progress since the breaker last held it" + ); + } + + /// Log a cycle that holds agreements back: an ERROR, for an operator, as the breaker trips + /// and every `UNREACHABLE_ALERT_REPEAT` after, and otherwise a WARN while it stays tripped. + fn report_hold( + &mut self, + trips: bool, + unreachable: usize, + checked: usize, + now: OffsetDateTime, + max_hold: time::Duration, + ) { + let held_hours = self + .tripped_since + .map_or(0, |since| (now - since).whole_hours()); + if self + .alerted_at + .is_none_or(|at| now - at >= UNREACHABLE_ALERT_REPEAT) + { + self.alerted_at = Some(now); + tracing::error!( + event = "liveness_abandonment_paused", + unreachable, + checked, + held_hours, + max_hold_days = max_hold.whole_days(), + "Too many indexers unreachable at once: a fault on dipper's side (its network or \ + the network registry), or that many really gone; not abandoning agreements for \ + unreachability while it lasts, for up to max_hold_days" + ); + } else if trips { + tracing::warn!( + unreachable, + checked, + held_hours, + "Still too many indexers unreachable; not abandoning agreements for unreachability" + ); + } + } + + /// Log, once, a trip that has held agreements back as long as it may: an ERROR, since from + /// here each indexer still unreachable is abandoned after a full tolerance without progress. + fn give_up( + &mut self, + unreachable: usize, + checked: usize, + now: OffsetDateTime, + max_hold: time::Duration, + ) { + let Some(since) = self.tripped_since else { + return; + }; + if self.alerted_at.is_some_and(|at| at - since > max_hold) { + return; + } + self.alerted_at = Some(now); + tracing::error!( + event = "liveness_abandonment_paused", + unreachable, + checked, + held_days = max_hold.whole_days(), + "Too many indexers unreachable for longer than an outage on dipper's side should \ + last; abandoning agreements for unreachability again, each after a full tolerance \ + without progress from now" + ); + } +} + +/// What checking an indexer's agreements needs, the same for every indexer in a cycle. +struct Checker<'a, R, W, C> { + registry: &'a R, + worker_queue: &'a W, + chain_client: &'a C, + agreement_conf: &'a crate::config::IndexingAgreementConfig, + max_tolerance_days: u32, + active_counts: &'a HashMap, db_timeout: Duration, queue_timeout: Duration, -) where +} + +impl Checker<'_, R, W, C> +where R: AgreementRegistry + IndexingRequestRegistry + PendingCancellationRegistry + Send + Sync, W: WorkerQueue + Send + Sync, C: ChainClient + Send + Sync, { - let now = OffsetDateTime::now_utc(); + /// Act on 1 indexer's agreements given what its status endpoint said, under `policy` when + /// it couldn't be reached. + async fn check_indexer( + &self, + uncancellable: &mut Uncancellable, + agreements: Vec, + probe: &Probe, + policy: UnreachablePolicy, + ) { + let unreachable = matches!(probe, Probe::Unreachable).then_some(policy); + for agreement in agreements { + let current_block = probe.block(&agreement); + self.check_agreement(uncancellable, &agreement, current_block, unreachable) + .await; + } + } - for agreement in agreements { + /// Act on 1 agreement given the latest block its indexer reports, if any, and, when its + /// indexer couldn't be reached, how this cycle treats that. + async fn check_agreement( + &self, + uncancellable: &mut Uncancellable, + agreement: &IndexingAgreement, + current_block: Option, + unreachable: Option, + ) { + let now = OffsetDateTime::now_utc(); + let deployment_id = agreement.terms.metadata.subgraph_deployment_id; + let last_progress_at = unreachable.map_or(agreement.last_progress_at, |policy| { + policy.progress_clock(agreement.last_progress_at) + }); let action = decide_liveness_action( agreement.last_block_height, - None, // No data available - agreement.last_progress_at, + current_block, + last_progress_at, now, - agreement.terms.metadata.subgraph_deployment_id, - active_counts, - config.max_tolerance_days, + deployment_id, + self.active_counts, + self.max_tolerance_days, ); - + let elapsed_hours = (now - last_progress_at.unwrap_or(now)).whole_hours(); + let threshold_hours = + tolerance_duration(deployment_id, self.active_counts, self.max_tolerance_days) + .whole_hours(); match action { - LivenessAction::InitializeTracking(block) => { - // First check with no data — initialize at 0 - record_progress(registry, agreement, block, now, db_timeout).await; + LivenessAction::InitializeTracking(block) | LivenessAction::ResetProgress(block) => { + record_progress(self.registry, agreement, block, now, self.db_timeout).await; } - LivenessAction::ResetProgress(_) => { - // Should not happen when current_block is None - unreachable!("ResetProgress should not occur with current_block = None"); - } - LivenessAction::WithinTolerance => { - let last_progress = agreement.last_progress_at.unwrap_or(now); - let elapsed = now - last_progress; - let threshold = tolerance_duration( - agreement.terms.metadata.subgraph_deployment_id, - active_counts, - config.max_tolerance_days, - ); + LivenessAction::WithinTolerance => tracing::debug!( + agreement_id = %agreement.id, + deployment = %deployment_id, + unreachable = unreachable.is_some(), + last_block = agreement.last_block_height.unwrap_or(0), + elapsed_hours, + threshold_hours, + "agreement block unchanged but within tolerance" + ), + LivenessAction::CancelAndReassess if unreachable == Some(UnreachablePolicy::Hold) => { tracing::debug!( agreement_id = %agreement.id, - elapsed_hours = elapsed.whole_hours(), - threshold_hours = threshold.whole_hours(), - "indexer unreachable but within tolerance" + elapsed_hours, + threshold_hours, + "indexer unreachable past tolerance; not abandoning while too many are" ); } LivenessAction::CancelAndReassess => { - let last_progress = agreement.last_progress_at.unwrap_or(now); - let elapsed = now - last_progress; - let threshold = tolerance_duration( - agreement.terms.metadata.subgraph_deployment_id, - active_counts, - config.max_tolerance_days, - ); - tracing::warn!( - agreement_id = %agreement.id, - last_block = agreement.last_block_height.unwrap_or(0), - elapsed_hours = elapsed.whole_hours(), - threshold_hours = threshold.whole_hours(), - "indexer unreachable and tolerance exceeded" - ); - cancel_and_reassess( + self.abandon( + uncancellable, agreement, - registry, - worker_queue, - chain_client, - agreement_conf, - db_timeout, - queue_timeout, + unreachable.is_some(), + elapsed_hours, + threshold_hours, ) .await; } } } + + /// End an agreement whose indexer has gone too long without progress, and replace it. + async fn abandon( + &self, + uncancellable: &mut Uncancellable, + agreement: &IndexingAgreement, + unreachable: bool, + elapsed_hours: i64, + threshold_hours: i64, + ) { + tracing::warn!( + agreement_id = %agreement.id, + indexer_id = %agreement.indexer.id, + deployment = %agreement.terms.metadata.subgraph_deployment_id, + unreachable, + last_block = agreement.last_block_height.unwrap_or(0), + elapsed_hours, + threshold_hours, + "agreement stale: no indexing progress detected" + ); + cancel_and_reassess( + agreement, + self.registry, + self.worker_queue, + self.chain_client, + self.agreement_conf, + uncancellable, + self.db_timeout, + self.queue_timeout, + ) + .await; + } } /// Update sync progress for an agreement in the DB. @@ -491,6 +708,7 @@ async fn cancel_and_reassess( worker_queue: &W, chain_client: &C, agreement_conf: &crate::config::IndexingAgreementConfig, + uncancellable: &mut Uncancellable, db_timeout: Duration, queue_timeout: Duration, ) where @@ -501,10 +719,7 @@ async fn cancel_and_reassess( // Without a cancel that can ever be sent, replacing it would pay 2 indexers until an // operator ends it, so it stays as it is, active, for one to deal with. if crate::cancel_dispatch::cancel_hash(agreement).is_none() { - tracing::error!( - agreement_id = %agreement.id, - "cannot cancel stale agreement: missing terms_version_hash; leaving active for operator action" - ); + uncancellable.report(agreement); return; } let Some(started) = start_abandoned_cancel( @@ -532,6 +747,33 @@ async fn cancel_and_reassess( replace_abandoned(agreement, registry, worker_queue, db_timeout, queue_timeout).await; } +/// Stale agreements already reported as impossible to cancel. The check repeats every cycle +/// for as long as one stays live, so this lets an operator hear of each once. Only agreements +/// stored without a terms hash land here, so it stays small without pruning. +#[derive(Default)] +struct Uncancellable(HashSet); + +impl Uncancellable { + /// Log a stale agreement no cancel can be sent for: an ERROR the first time, for an operator + /// to act on, and a WARN on every later cycle. + fn report(&mut self, agreement: &IndexingAgreement) { + if self.0.insert(agreement.id) { + tracing::error!( + event = "stale_agreement_uncancellable", + agreement_id = %agreement.id, + indexer_id = %agreement.indexer.id, + indexing_request_id = %agreement.indexing_request_id, + "cannot cancel stale agreement: missing terms_version_hash; leaving active for operator action" + ); + } else { + tracing::warn!( + agreement_id = %agreement.id, + "stale agreement still cannot be cancelled: missing terms_version_hash" + ); + } + } +} + /// Note that the chain shows a stale agreement still cancelling has ended, so it no longer /// holds its slot in the request and the reassessment queued next fills it. Bounded by /// `db_timeout`, so a hung database can't stall the liveness cycle. @@ -912,8 +1154,9 @@ mod tests { use url::Url; use super::{ - LivenessAction, cancel_and_reassess, decide_liveness_action, group_by_indexer_id, - record_progress, tolerance_duration, + Checker, LivenessAction, Probe, Uncancellable, UnreachableBreaker, UnreachablePolicy, + cancel_and_reassess, decide_liveness_action, group_by_indexer_id, record_progress, + tolerance_duration, unreachable_trips, }; use crate::{ chain_client::{AgreementOnChain, ChainClient, ChainClientError}, @@ -1381,6 +1624,8 @@ mod tests { assert_eq!(config.max_tolerance_days, 4); assert_eq!(config.request_timeout, std::time::Duration::from_secs(10)); assert_eq!(config.batch_size, 500); + assert_eq!(config.unreachable_trip_fraction, 0.5); + assert_eq!(config.unreachable_trip_min_indexers, 3); } // ---- decide_liveness_action pure function tests ---- @@ -1569,6 +1814,15 @@ mod tests { agreement: &IndexingAgreement, registry: &MockRegistry, chain: &MockChainClient, + ) { + end_stale_reporting(agreement, registry, chain, &mut Uncancellable::default()).await; + } + + async fn end_stale_reporting( + agreement: &IndexingAgreement, + registry: &MockRegistry, + chain: &MockChainClient, + uncancellable: &mut Uncancellable, ) { let queue = MockWorkerQueue { calls: registry.calls.clone(), @@ -1579,12 +1833,37 @@ mod tests { &queue, chain, &test_agreement_conf(), + uncancellable, DB_TIMEOUT, QUEUE_TIMEOUT, ) .await; } + /// What dipper logs, at every level, while `run` runs. + async fn logged(run: impl std::future::Future) -> String { + struct Buffer(Arc>>); + impl std::io::Write for Buffer { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(bytes); + Ok(bytes.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + let lines = Arc::new(Mutex::new(Vec::new())); + let writer = Arc::clone(&lines); + let subscriber = tracing_subscriber::fmt() + .with_ansi(false) + .with_max_level(tracing::Level::DEBUG) + .with_writer(move || Buffer(Arc::clone(&writer))) + .finish(); + let _logging = tracing::subscriber::set_default(subscriber); + run.await; + String::from_utf8(lines.lock().unwrap().clone()).unwrap() + } + #[tokio::test] async fn ends_a_stale_agreement_through_cancelling_and_reassesses() { let agreement = stale_agreement(); @@ -1676,6 +1955,41 @@ mod tests { assert!(calls.reassessments.lock().unwrap().is_empty()); } + #[tokio::test] + async fn alerts_once_on_a_stale_agreement_it_can_never_cancel() { + // The check repeats every cycle while the agreement stays live; only the first goes to + // Slack, through its event tag. + let mut agreement = stale_agreement(); + agreement.terms_version_hash = None; + let calls = MockCalls::default(); + let registry = MockRegistry::new(calls.clone(), agreement.clone()); + let chain = MockChainClient::success(calls.clone()); + let mut uncancellable = Uncancellable::default(); + + let first = logged(end_stale_reporting( + &agreement, + ®istry, + &chain, + &mut uncancellable, + )) + .await; + let second = logged(end_stale_reporting( + &agreement, + ®istry, + &chain, + &mut uncancellable, + )) + .await; + + assert!(first.contains(" ERROR "), "{first}"); + assert!(first.contains("stale_agreement_uncancellable"), "{first}"); + assert!(second.contains(" WARN "), "{second}"); + assert!( + !second.contains("stale_agreement_uncancellable"), + "{second}" + ); + } + #[tokio::test] async fn leaves_an_agreement_already_ending_alone() { let agreement = stale_agreement(); @@ -1728,6 +2042,281 @@ mod tests { ); } + // ---- Holding back abandonment while many indexers are unreachable ---- + + #[test] + fn trips_when_at_least_the_trip_fraction_of_enough_indexers_is_unreachable() { + assert!(unreachable_trips(4, 2, 0.5, 3), "at the fraction"); + assert!(unreachable_trips(4, 4, 0.5, 3), "above it"); + assert!(unreachable_trips(3, 2, 0.5, 3), "at the minimum"); + assert!(!unreachable_trips(4, 1, 0.5, 3), "below the fraction"); + assert!(!unreachable_trips(2, 2, 0.5, 3), "below the minimum"); + assert!(!unreachable_trips(0, 0, 0.5, 0), "nothing checked"); + } + + /// The time of a liveness cycle `n` 5-minute cycles after `start`. + fn cycle(start: OffsetDateTime, n: i64) -> OffsetDateTime { + start + time::Duration::minutes(5 * n) + } + + #[tokio::test] + async fn alerts_once_as_the_breaker_trips_and_says_when_it_clears() { + let config = LivenessCheckerConfig::default(); + let mut breaker = UnreachableBreaker::default(); + let start = OffsetDateTime::now_utc(); + let mut evaluate = + |n, unreachable| breaker.evaluate(4, unreachable, cycle(start, n), &config); + + let tripped = logged(async { assert_eq!(evaluate(0, 3), UnreachablePolicy::Hold) }).await; + let still = logged(async { assert_eq!(evaluate(1, 4), UnreachablePolicy::Hold) }).await; + evaluate(2, 0); + evaluate(3, 0); + let cleared = logged(async { + assert_ne!(evaluate(4, 0), UnreachablePolicy::Hold); + }) + .await; + + assert!(tripped.contains(" ERROR "), "{tripped}"); + assert!(tripped.contains("liveness_abandonment_paused"), "{tripped}"); + assert!(still.contains(" WARN "), "{still}"); + assert!(!still.contains("liveness_abandonment_paused"), "{still}"); + assert!(cleared.contains(" INFO "), "{cleared}"); + assert!( + !cleared.contains("liveness_abandonment_paused"), + "{cleared}" + ); + } + + #[test] + fn clears_only_once_3_cycles_in_a_row_see_too_few_unreachable() { + let config = LivenessCheckerConfig::default(); + let mut breaker = UnreachableBreaker::default(); + let start = OffsetDateTime::now_utc(); + let mut evaluate = + |n, unreachable| breaker.evaluate(4, unreachable, cycle(start, n), &config); + + assert_eq!(evaluate(0, 2), UnreachablePolicy::Hold, "trips"); + assert_eq!(evaluate(1, 1), UnreachablePolicy::Hold); + assert_eq!(evaluate(2, 1), UnreachablePolicy::Hold); + assert_eq!( + evaluate(3, 2), + UnreachablePolicy::Hold, + "trips again, starting the count over" + ); + assert_eq!(evaluate(4, 1), UnreachablePolicy::Hold); + assert_eq!(evaluate(5, 1), UnreachablePolicy::Hold); + let cleared = UnreachablePolicy::Judge { + since: Some(cycle(start, 5)), + }; + assert_eq!(evaluate(6, 1), cleared, "counting from the last cycle held"); + assert_eq!(evaluate(7, 0), cleared, "and on every cycle after"); + } + + #[tokio::test] + async fn reminds_an_operator_every_6_hours_while_tripped() { + let config = LivenessCheckerConfig::default(); + let mut breaker = UnreachableBreaker::default(); + let start = OffsetDateTime::now_utc(); + let mut evaluate = + |hours| breaker.evaluate(4, 4, start + time::Duration::hours(hours), &config); + + let tripped = logged(async { assert_eq!(evaluate(0), UnreachablePolicy::Hold) }).await; + let soon = logged(async { assert_eq!(evaluate(5), UnreachablePolicy::Hold) }).await; + let later = logged(async { assert_eq!(evaluate(6), UnreachablePolicy::Hold) }).await; + + assert!(tripped.contains("liveness_abandonment_paused"), "{tripped}"); + assert!(!soon.contains("liveness_abandonment_paused"), "{soon}"); + assert!(later.contains(" ERROR "), "{later}"); + assert!(later.contains("liveness_abandonment_paused"), "{later}"); + } + + #[tokio::test] + async fn stops_holding_after_max_tolerance_days_and_says_so_once() { + // Indexers that are really gone can keep the share up for good; past the longest + // tolerance, an outage on dipper's side is no longer the likely cause. + let config = LivenessCheckerConfig::default(); + let mut breaker = UnreachableBreaker::default(); + let start = OffsetDateTime::now_utc(); + let last_held = start + time::Duration::days(i64::from(config.max_tolerance_days)); + let minutes = |m| last_held + time::Duration::minutes(m); + breaker.evaluate(4, 4, start, &config); + assert_eq!( + breaker.evaluate(4, 4, last_held, &config), + UnreachablePolicy::Hold + ); + + let mut evaluate = |at| breaker.evaluate(4, 4, at, &config); + let gave_up = logged(async { + assert_eq!( + evaluate(minutes(5)), + UnreachablePolicy::Judge { + since: Some(last_held) + } + ); + }) + .await; + let after = logged(async { + assert_eq!( + evaluate(minutes(10)), + UnreachablePolicy::Judge { + since: Some(last_held) + } + ); + }) + .await; + + assert!(gave_up.contains(" ERROR "), "{gave_up}"); + assert!(gave_up.contains("liveness_abandonment_paused"), "{gave_up}"); + assert!(!after.contains("liveness_abandonment_paused"), "{after}"); + } + + /// An agreement whose indexer last made progress 25 hours ago, past its 1-day tolerance. + fn long_stalled_agreement() -> IndexingAgreement { + let mut agreement = stale_agreement(); + agreement.last_progress_at = Some(OffsetDateTime::now_utc() - time::Duration::hours(25)); + agreement + } + + async fn check( + agreement: &IndexingAgreement, + registry: &MockRegistry, + chain: &MockChainClient, + probe: &Probe, + policy: UnreachablePolicy, + ) { + let queue = MockWorkerQueue { + calls: registry.calls.clone(), + }; + let conf = test_agreement_conf(); + let active_counts = HashMap::new(); + let checker = Checker { + registry, + worker_queue: &queue, + chain_client: chain, + agreement_conf: &conf, + max_tolerance_days: 4, + active_counts: &active_counts, + db_timeout: DB_TIMEOUT, + queue_timeout: QUEUE_TIMEOUT, + }; + checker + .check_indexer( + &mut Uncancellable::default(), + vec![agreement.clone()], + probe, + policy, + ) + .await; + } + + #[tokio::test] + async fn still_abandons_an_indexer_that_answers_without_progress_while_tripped() { + let agreement = long_stalled_agreement(); + let calls = MockCalls::default(); + let registry = MockRegistry::new(calls.clone(), agreement.clone()); + let chain = MockChainClient::success(calls.clone()); + let deployment = agreement.terms.metadata.subgraph_deployment_id.to_string(); + let stalled = Probe::Reached(HashMap::from([(deployment, agreement.last_block_height)])); + + check( + &agreement, + ®istry, + &chain, + &stalled, + UnreachablePolicy::Hold, + ) + .await; + + assert_eq!(calls.abandoning.lock().unwrap().as_slice(), &[agreement.id]); + } + + #[tokio::test] + async fn leaves_an_unreachable_indexer_and_its_progress_clock_alone_while_tripped() { + let agreement = long_stalled_agreement(); + let calls = MockCalls::default(); + let registry = MockRegistry::new(calls.clone(), agreement.clone()); + let chain = MockChainClient::success(calls.clone()); + + let hold = UnreachablePolicy::Hold; + check(&agreement, ®istry, &chain, &Probe::Unreachable, hold).await; + + assert!(calls.abandoning.lock().unwrap().is_empty()); + assert!(calls.chain_cancels.lock().unwrap().is_empty()); + assert!( + calls.progress_updates.lock().unwrap().is_empty(), + "its time without progress keeps counting, for when the breaker clears" + ); + } + + #[tokio::test] + async fn abandons_an_unreachable_indexer_past_tolerance_when_not_tripped() { + let agreement = long_stalled_agreement(); + let calls = MockCalls::default(); + let registry = MockRegistry::new(calls.clone(), agreement.clone()); + let chain = MockChainClient::success(calls.clone()); + + let judge = UnreachablePolicy::Judge { since: None }; + check(&agreement, ®istry, &chain, &Probe::Unreachable, judge).await; + + assert_eq!(calls.abandoning.lock().unwrap().as_slice(), &[agreement.id]); + } + + #[tokio::test] + async fn gives_an_indexer_still_unreachable_a_full_tolerance_once_the_breaker_clears() { + // A 30-hour outage held the agreement past its 1-day tolerance; dipper couldn't see the + // indexer through it, so it isn't abandoned for missing the first cycles after. + let agreement = long_stalled_agreement(); + let calls = MockCalls::default(); + let registry = MockRegistry::new(calls.clone(), agreement.clone()); + let chain = MockChainClient::success(calls.clone()); + let config = LivenessCheckerConfig::default(); + let mut breaker = UnreachableBreaker::default(); + let now = OffsetDateTime::now_utc(); + breaker.evaluate(4, 4, now - time::Duration::hours(30), &config); + breaker.evaluate(4, 4, now - time::Duration::minutes(20), &config); + breaker.evaluate(4, 1, now - time::Duration::minutes(15), &config); + breaker.evaluate(4, 1, now - time::Duration::minutes(10), &config); + let cleared = breaker.evaluate(4, 1, now - time::Duration::minutes(5), &config); + + check(&agreement, ®istry, &chain, &Probe::Unreachable, cleared).await; + + assert!(calls.abandoning.lock().unwrap().is_empty()); + assert!(calls.chain_cancels.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn abandons_an_indexer_still_unreachable_a_full_tolerance_after_the_breaker_cleared() { + let agreement = long_stalled_agreement(); + let calls = MockCalls::default(); + let registry = MockRegistry::new(calls.clone(), agreement.clone()); + let chain = MockChainClient::success(calls.clone()); + let cleared = UnreachablePolicy::Judge { + since: Some(OffsetDateTime::now_utc() - time::Duration::hours(25)), + }; + + check(&agreement, ®istry, &chain, &Probe::Unreachable, cleared).await; + + assert_eq!(calls.abandoning.lock().unwrap().as_slice(), &[agreement.id]); + } + + #[tokio::test] + async fn still_abandons_an_indexer_that_answers_without_progress_once_the_breaker_clears() { + // It answered through the outage, so its time without progress was seen and counts. + let agreement = long_stalled_agreement(); + let calls = MockCalls::default(); + let registry = MockRegistry::new(calls.clone(), agreement.clone()); + let chain = MockChainClient::success(calls.clone()); + let deployment = agreement.terms.metadata.subgraph_deployment_id.to_string(); + let stalled = Probe::Reached(HashMap::from([(deployment, agreement.last_block_height)])); + let cleared = UnreachablePolicy::Judge { + since: Some(OffsetDateTime::now_utc() - time::Duration::minutes(5)), + }; + + check(&agreement, ®istry, &chain, &stalled, cleared).await; + + assert_eq!(calls.abandoning.lock().unwrap().as_slice(), &[agreement.id]); + } + #[tokio::test] async fn test_record_progress_calls_registry() { // Arrange diff --git a/bin/dipper-service/src/registry/agreement.rs b/bin/dipper-service/src/registry/agreement.rs index fde09894..e4416cb5 100644 --- a/bin/dipper-service/src/registry/agreement.rs +++ b/bin/dipper-service/src/registry/agreement.rs @@ -463,7 +463,8 @@ pub trait AgreementRegistry { rejection_reason: Option<&str>, ) -> RegistryResult<()>; - /// Get all `AcceptedOnChain` agreements for liveness checking. + /// Get `AcceptedOnChain` agreements for liveness checking: the `batch_size` longest without + /// progress, plus 1 of every other indexer holding one, so every such indexer is asked. /// /// Results are ordered by `last_progress_at` ascending (NULLs first) so agreements /// that have never been checked are processed first. diff --git a/dipper-pgregistry/src/postgres.rs b/dipper-pgregistry/src/postgres.rs index 679ea805..e25800ea 100644 --- a/dipper-pgregistry/src/postgres.rs +++ b/dipper-pgregistry/src/postgres.rs @@ -1836,7 +1836,9 @@ impl PgRegistry { // Liveness tracking operations // ========================================================================= - /// Get all `AcceptedOnChain` agreements for liveness checking. + /// Get `AcceptedOnChain` agreements for liveness checking: the `batch_size` longest without + /// progress (all when it is 0 or less), plus the one longest without progress of every other + /// indexer holding one, so each cycle asks every such indexer whether it can be reached. /// /// Returns agreements ordered by `last_progress_at` ascending (NULLs first), /// so agreements that have never been checked are processed first. @@ -1846,6 +1848,22 @@ impl PgRegistry { ) -> Result, Error> { sqlx::query_as( r#" + WITH accepted AS ( + SELECT id, indexer_id, last_progress_at + FROM dipper_reg_indexing_agreements + WHERE status = $1 + ), + stalest AS ( + SELECT id + FROM accepted + ORDER BY last_progress_at ASC NULLS FIRST + LIMIT CASE WHEN $2 > 0 THEN $2 ELSE NULL END + ), + stalest_per_indexer AS ( + SELECT DISTINCT ON (indexer_id) id + FROM accepted + ORDER BY indexer_id, last_progress_at ASC NULLS FIRST + ) SELECT id, nonce_uuid, @@ -1862,9 +1880,8 @@ impl PgRegistry { rejection_reason, terms_version_hash FROM dipper_reg_indexing_agreements - WHERE status = $1 + WHERE id IN (SELECT id FROM stalest UNION SELECT id FROM stalest_per_indexer) ORDER BY last_progress_at ASC NULLS FIRST - LIMIT CASE WHEN $2 > 0 THEN $2 ELSE NULL END "#, ) .bind(IndexingAgreementStatus::AcceptedOnChain) diff --git a/dipper-pgregistry/tests/it_registry_postgres.rs b/dipper-pgregistry/tests/it_registry_postgres.rs index bbbabfe4..d1f0a006 100644 --- a/dipper-pgregistry/tests/it_registry_postgres.rs +++ b/dipper-pgregistry/tests/it_registry_postgres.rs @@ -1749,6 +1749,54 @@ async fn test_get_accepted_on_chain_agreements() { ); } +#[tokio::test] +async fn get_accepted_on_chain_agreements_adds_1_of_each_indexer_the_batch_leaves_out() { + //* 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 A holds 2 accepted agreements that stalled days ago; indexer B holds 1 that + // progressed a minute ago. + run_fixture( + &db, + r#" + UPDATE dipper_reg_indexing_agreements + SET status = 6, last_block_height = 1, last_progress_at = now() - interval '3 days' + WHERE id = '\xaa000000000000000000000000000001'::bytea; + UPDATE dipper_reg_indexing_agreements + SET last_block_height = 1, last_progress_at = now() - interval '2 days' + WHERE id = '\xaa000000000000000000000000000002'::bytea; + UPDATE dipper_reg_indexing_agreements + SET status = 6, last_block_height = 1, last_progress_at = now() - interval '1 minute' + WHERE id = '\xbb000000000000000000000000000001'::bytea; + "#, + ) + .await + .expect("Failed to set up accepted agreements"); + let registry = PgRegistry::new(db); + + //* When + let agreements = registry + .get_accepted_on_chain_agreements(1) + .await + .expect("Failed to get AcceptedOnChain agreements"); + + //* Then + let ids: Vec = agreements.iter().map(|a| a.id).collect(); + assert_eq!( + ids, + vec![ + IndexingAgreementId::from_bytes([0xaa, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1]), + IndexingAgreementId::from_bytes([0xbb, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1]), + ], + "the batch of 1 stalest, then 1 of indexer B, which the batch left out" + ); +} + #[tokio::test] async fn test_update_agreement_sync_progress() { //* Given @@ -4157,6 +4205,17 @@ async fn an_accept_and_end_from_the_chain_are_recorded_in_1_write() { (Some(100), Some(200), Some("0xend".to_owned()), true), "what is already known stays" ); + + registry + .mark_terminated_event_emitted(&id) + .await + .expect("announce the end"); + record(100, 200, "0xend").await; + assert_eq!( + recorded().await, + (Some(100), Some(200), Some("0xend".to_owned()), false), + "the chain listener reading the same change again doesn't announce it again" + ); } /// An ended agreement's `terminated` event waits for the transaction that ended it, which the diff --git a/k8s/configmap-example.yaml b/k8s/configmap-example.yaml index ae114cba..f69a6d38 100644 --- a/k8s/configmap-example.yaml +++ b/k8s/configmap-example.yaml @@ -98,7 +98,9 @@ data: "interval": 300, "max_tolerance_days": 4, "request_timeout": 10, - "batch_size": 500 + "batch_size": 500, + "unreachable_trip_fraction": 0.5, + "unreachable_trip_min_indexers": 3 }, "chain_client": { "enabled": true, @@ -121,7 +123,14 @@ data: }, "alerts": { "slack_webhook_url": "https://hooks.slack.com/services/REPLACE_ME_OR_REMOVE_FOR_NO_ALERTS", - "events": ["agreement_cancel_stuck", "rpc_blocks_refused", "nonce_gap_fill_failed"], + "events": [ + "agreement_cancel_stuck", + "rpc_blocks_refused", + "nonce_gap_fill_failed", + "chain_listener_changes_skipped", + "stale_agreement_uncancellable", + "liveness_abandonment_paused" + ], "throttle": 900 } }