From a507612209db9bba5ab3e8014aaf2478300deb05 Mon Sep 17 00:00:00 2001
From: MoonBoi9001 <contact@samuelmetcalfe.com>
Date: Mon, 12 Oct 2026 15:34:08 +0100
Subject: [PATCH 1/7] fix(alerts): report skipped chain changes and
 uncancellable agreements

Slack gets only the log lines whose tag is listed, and 2 errors needing an operator were not:
on-chain changes the listener gave up applying, and a stale agreement no cancel can be sent for.
The second repeats every 5 minutes, so it now alerts once per agreement and only warns after.
---
 bin/dipper-service/src/config.rs              |   7 ++
 .../src/network/service/liveness_checker.rs   | 119 ++++++++++++++++--
 k8s/configmap-example.yaml                    |   8 +-
 3 files changed, 126 insertions(+), 8 deletions(-)

diff --git a/bin/dipper-service/src/config.rs b/bin/dipper-service/src/config.rs
index abab64f5..5eaae2d8 100644
--- a/bin/dipper-service/src/config.rs
+++ b/bin/dipper-service/src/config.rs
@@ -127,6 +127,8 @@ impl Default for AlertsConfig {
                 "agreement_cancel_stuck",
                 "rpc_blocks_refused",
                 "nonce_gap_fill_failed",
+                "chain_listener_changes_skipped",
+                "stale_agreement_uncancellable",
             ]
             .map(str::to_owned)
             .to_vec(),
@@ -2133,6 +2135,11 @@ mod tests {
         let alerts = serde_json::from_value::<AlertsConfig>(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/liveness_checker.rs b/bin/dipper-service/src/network/service/liveness_checker.rs
index 6f1a87c6..3469fb9f 100644
--- a/bin/dipper-service/src/network/service/liveness_checker.rs
+++ b/bin/dipper-service/src/network/service/liveness_checker.rs
@@ -35,8 +35,14 @@
 //! 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 +137,7 @@ where
             .timeout(config.request_timeout)
             .build()
             .map_err(|e| anyhow::anyhow!("failed to build HTTP client: {e}"))?;
+        let mut uncancellable = Uncancellable::default();
 
         loop {
             tokio::select! {
@@ -214,6 +221,7 @@ where
                             &worker_queue,
                             &chain_client,
                             &agreement_conf,
+                            &mut uncancellable,
                             DB_UPDATE_TIMEOUT,
                             QUEUE_PUSH_TIMEOUT,
                         )
@@ -306,6 +314,7 @@ where
                                 &worker_queue,
                                 &chain_client,
                                 &agreement_conf,
+                                &mut uncancellable,
                                 DB_UPDATE_TIMEOUT,
                                 QUEUE_PUSH_TIMEOUT,
                             )
@@ -368,6 +377,7 @@ async fn process_agreements_with_no_data<R, W, C>(
     worker_queue: &W,
     chain_client: &C,
     agreement_conf: &crate::config::IndexingAgreementConfig,
+    uncancellable: &mut Uncancellable,
     db_timeout: Duration,
     queue_timeout: Duration,
 ) where
@@ -433,6 +443,7 @@ async fn process_agreements_with_no_data<R, W, C>(
                     worker_queue,
                     chain_client,
                     agreement_conf,
+                    uncancellable,
                     db_timeout,
                     queue_timeout,
                 )
@@ -491,6 +502,7 @@ async fn cancel_and_reassess<R, W, C>(
     worker_queue: &W,
     chain_client: &C,
     agreement_conf: &crate::config::IndexingAgreementConfig,
+    uncancellable: &mut Uncancellable,
     db_timeout: Duration,
     queue_timeout: Duration,
 ) where
@@ -501,10 +513,7 @@ async fn cancel_and_reassess<R, W, C>(
     // 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 +541,33 @@ async fn cancel_and_reassess<R, W, C>(
     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<IndexingAgreementId>);
+
+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 +948,8 @@ mod tests {
     use url::Url;
 
     use super::{
-        LivenessAction, cancel_and_reassess, decide_liveness_action, group_by_indexer_id,
-        record_progress, tolerance_duration,
+        LivenessAction, Uncancellable, cancel_and_reassess, decide_liveness_action,
+        group_by_indexer_id, record_progress, tolerance_duration,
     };
     use crate::{
         chain_client::{AgreementOnChain, ChainClient, ChainClientError},
@@ -1569,6 +1605,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 +1624,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<Output = ()>) -> String {
+        struct Buffer(Arc<Mutex<Vec<u8>>>);
+        impl std::io::Write for Buffer {
+            fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
+                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 +1746,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,
+            &registry,
+            &chain,
+            &mut uncancellable,
+        ))
+        .await;
+        let second = logged(end_stale_reporting(
+            &agreement,
+            &registry,
+            &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();
diff --git a/k8s/configmap-example.yaml b/k8s/configmap-example.yaml
index ae114cba..32d97d94 100644
--- a/k8s/configmap-example.yaml
+++ b/k8s/configmap-example.yaml
@@ -121,7 +121,13 @@ 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"
+        ],
         "throttle": 900
       }
     }

From cc98edee2d853d6508727da4fe0d0641e4eddc75 Mon Sep 17 00:00:00 2001
From: MoonBoi9001 <contact@samuelmetcalfe.com>
Date: Mon, 12 Oct 2026 15:44:48 +0100
Subject: [PATCH 2/7] fix(liveness): pause abandonment when most indexers can't
 be reached

An outage on dipper's side made every indexer look dead, so a day later all agreements ended,
were replaced and their indexers were kept out for 30 days. Now when half of 3 or more indexers
can't be reached at once, that cycle ends none for it, alerts once, and still ends real stalls.
---
 bin/dipper-service/src/config.rs              |  51 +-
 .../src/network/service/liveness_checker.rs   | 574 ++++++++++++------
 k8s/configmap-example.yaml                    |   7 +-
 3 files changed, 439 insertions(+), 193 deletions(-)

diff --git a/bin/dipper-service/src/config.rs b/bin/dipper-service/src/config.rs
index 5eaae2d8..e2488e79 100644
--- a/bin/dipper-service/src/config.rs
+++ b/bin/dipper-service/src/config.rs
@@ -129,6 +129,7 @@ impl Default for AlertsConfig {
                 "nonce_gap_fill_failed",
                 "chain_listener_changes_skipped",
                 "stale_agreement_uncancellable",
+                "liveness_abandonment_paused",
             ]
             .map(str::to_owned)
             .to_vec(),
@@ -541,6 +542,17 @@ pub struct LivenessCheckerConfig {
     /// Maximum agreements to fetch per cycle (default: 500).
     #[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 that cycle abandoning agreements for
+    /// unreachability, 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 {
@@ -563,6 +575,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 {
@@ -571,6 +591,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(),
         }
     }
 }
@@ -2120,8 +2142,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")
@@ -2130,7 +2152,30 @@ 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::<LivenessCheckerConfig>(
+            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::<AlertsConfig>(config["alerts"].clone())
             .expect("the example's alerts section is a valid alerts config");
diff --git a/bin/dipper-service/src/network/service/liveness_checker.rs b/bin/dipper-service/src/network/service/liveness_checker.rs
index 3469fb9f..ac809c90 100644
--- a/bin/dipper-service/src/network/service/liveness_checker.rs
+++ b/bin/dipper-service/src/network/service/liveness_checker.rs
@@ -30,6 +30,10 @@
 //! - 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 that cycle abandons no agreement for
+//! unreachability; indexers that answered without progress are still handled as above.
+//!
 //! ## URL Refresh
 //!
 //! Indexer URLs are looked up fresh from the network topology on each cycle,
@@ -138,6 +142,7 @@ where
             .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! {
@@ -193,135 +198,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(());
                 }
-
-                // 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,
-                            &registry,
-                            &worker_queue,
-                            &chain_client,
-                            &agreement_conf,
-                            &mut uncancellable,
-                            DB_UPDATE_TIMEOUT,
-                            QUEUE_PUSH_TIMEOUT,
-                        )
-                        .await;
-                        continue;
-                    }
-                };
-
-                // 5. Query the indexer status endpoint for all deployments in this group
-                let deployment_ids: Vec<String> = 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(&registry, &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,
-                                &registry,
-                                &worker_queue,
-                                &chain_client,
-                                &agreement_conf,
-                                &mut uncancellable,
-                                DB_UPDATE_TIMEOUT,
-                                QUEUE_PUSH_TIMEOUT,
-                            )
-                            .await;
-                        }
-                    }
+                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 hold_unreachable = breaker.evaluate(probed.len(), unreachable, &config);
+
+            // 4. Act on each indexer's agreements
+            let checker = Checker {
+                registry: &registry,
+                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,
+                        hold_unreachable,
+                    )
+                    .await;
             }
 
             tracing::debug!("liveness check completed");
@@ -365,92 +284,254 @@ 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<R, W, C>(
+/// 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<String, Option<u64>>),
+    /// 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<u64> {
+        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<DeploymentId, usize>,
-    config: &LivenessCheckerConfig,
-    registry: &R,
-    worker_queue: &W,
-    chain_client: &C,
-    agreement_conf: &crate::config::IndexingAgreementConfig,
-    uncancellable: &mut Uncancellable,
+) -> 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<String> = 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
+}
+
+/// Stops agreements being abandoned because their indexer can't be reached while so many
+/// indexers can't be reached at once that the fault is likely dipper's own (its network, or the
+/// network registry). Abandoning them all would hand every deployment to new indexers and keep
+/// the old ones out for weeks.
+#[derive(Default)]
+struct UnreachableBreaker {
+    tripped: bool,
+}
+
+impl UnreachableBreaker {
+    /// Whether this cycle holds back abandoning agreements for unreachability, logging an ERROR
+    /// as it trips, a WARN for each later cycle it stays tripped and an INFO as it clears.
+    fn evaluate(
+        &mut self,
+        checked: usize,
+        unreachable: usize,
+        config: &LivenessCheckerConfig,
+    ) -> bool {
+        let trip_fraction = config.unreachable_trip_fraction;
+        let trips = unreachable_trips(
+            checked,
+            unreachable,
+            trip_fraction,
+            config.unreachable_trip_min_indexers,
+        );
+        match (self.tripped, trips) {
+            (false, true) => tracing::error!(
+                event = "liveness_abandonment_paused",
+                unreachable,
+                checked,
+                trip_fraction,
+                "Too many indexers unreachable at once, likely a fault on dipper's side; not \
+                 abandoning agreements for unreachability until fewer are"
+            ),
+            (true, true) => tracing::warn!(
+                unreachable,
+                checked,
+                trip_fraction,
+                "Still too many indexers unreachable; not abandoning agreements for unreachability"
+            ),
+            (true, false) => tracing::info!(
+                unreachable,
+                checked,
+                trip_fraction,
+                "Fewer indexers unreachable; abandoning agreements for unreachability resumes"
+            ),
+            (false, false) => {}
+        }
+        self.tripped = trips;
+        trips
+    }
+}
+
+/// 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<DeploymentId, usize>,
     db_timeout: Duration,
     queue_timeout: Duration,
-) where
+}
+
+impl<R, W, C> 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. With
+    /// `hold_unreachable` set, an unreachable indexer's stale agreements are left as they are,
+    /// their progress clock untouched, so they are judged as before once the breaker clears.
+    async fn check_indexer(
+        &self,
+        uncancellable: &mut Uncancellable,
+        agreements: Vec<IndexingAgreement>,
+        probe: &Probe,
+        hold_unreachable: bool,
+    ) {
+        let unreachable = matches!(probe, Probe::Unreachable);
+        for agreement in agreements {
+            let current_block = probe.block(&agreement);
+            self.check_agreement(
+                uncancellable,
+                &agreement,
+                current_block,
+                unreachable,
+                unreachable && hold_unreachable,
+            )
+            .await;
+        }
+    }
 
-    for agreement in agreements {
+    /// Act on 1 agreement given the latest block its indexer reports, if any; with `hold` set,
+    /// leave it as it is however long it has gone without progress.
+    async fn check_agreement(
+        &self,
+        uncancellable: &mut Uncancellable,
+        agreement: &IndexingAgreement,
+        current_block: Option<u64>,
+        unreachable: bool,
+        hold: bool,
+    ) {
+        let now = OffsetDateTime::now_utc();
+        let deployment_id = agreement.terms.metadata.subgraph_deployment_id;
         let action = decide_liveness_action(
             agreement.last_block_height,
-            None, // No data available
+            current_block,
             agreement.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 - agreement.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::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,
-                );
-                tracing::debug!(
-                    agreement_id = %agreement.id,
-                    elapsed_hours = elapsed.whole_hours(),
-                    threshold_hours = threshold.whole_hours(),
-                    "indexer unreachable but within tolerance"
-                );
+            LivenessAction::InitializeTracking(block) | LivenessAction::ResetProgress(block) => {
+                record_progress(self.registry, agreement, block, now, self.db_timeout).await;
             }
+            LivenessAction::WithinTolerance => tracing::debug!(
+                agreement_id = %agreement.id,
+                deployment = %deployment_id,
+                unreachable,
+                last_block = agreement.last_block_height.unwrap_or(0),
+                elapsed_hours,
+                threshold_hours,
+                "agreement block unchanged but within tolerance"
+            ),
+            LivenessAction::CancelAndReassess if hold => tracing::debug!(
+                agreement_id = %agreement.id,
+                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(
-                    agreement,
-                    registry,
-                    worker_queue,
-                    chain_client,
-                    agreement_conf,
+                self.abandon(
                     uncancellable,
-                    db_timeout,
-                    queue_timeout,
+                    agreement,
+                    unreachable,
+                    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.
@@ -948,8 +1029,9 @@ mod tests {
     use url::Url;
 
     use super::{
-        LivenessAction, Uncancellable, cancel_and_reassess, decide_liveness_action,
-        group_by_indexer_id, record_progress, tolerance_duration,
+        Checker, LivenessAction, Probe, Uncancellable, UnreachableBreaker, cancel_and_reassess,
+        decide_liveness_action, group_by_indexer_id, record_progress, tolerance_duration,
+        unreachable_trips,
     };
     use crate::{
         chain_client::{AgreementOnChain, ChainClient, ChainClientError},
@@ -1417,6 +1499,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 ----
@@ -1833,6 +1917,120 @@ 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");
+    }
+
+    #[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 tripped = logged(async { assert!(breaker.evaluate(4, 3, &config)) }).await;
+        let still = logged(async { assert!(breaker.evaluate(4, 4, &config)) }).await;
+        let cleared = logged(async { assert!(!breaker.evaluate(4, 0, &config)) }).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}"
+        );
+    }
+
+    /// 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,
+        hold_unreachable: bool,
+    ) {
+        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,
+                hold_unreachable,
+            )
+            .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, &registry, &chain, &stalled, true).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());
+
+        check(&agreement, &registry, &chain, &Probe::Unreachable, true).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());
+
+        check(&agreement, &registry, &chain, &Probe::Unreachable, false).await;
+
+        assert_eq!(calls.abandoning.lock().unwrap().as_slice(), &[agreement.id]);
+    }
+
     #[tokio::test]
     async fn test_record_progress_calls_registry() {
         // Arrange
diff --git a/k8s/configmap-example.yaml b/k8s/configmap-example.yaml
index 32d97d94..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,
@@ -126,7 +128,8 @@ data:
           "rpc_blocks_refused",
           "nonce_gap_fill_failed",
           "chain_listener_changes_skipped",
-          "stale_agreement_uncancellable"
+          "stale_agreement_uncancellable",
+          "liveness_abandonment_paused"
         ],
         "throttle": 900
       }

From 73364c808f9323b8421401861afb09476d36c8ff Mon Sep 17 00:00:00 2001
From: MoonBoi9001 <contact@samuelmetcalfe.com>
Date: Mon, 12 Oct 2026 15:35:46 +0100
Subject: [PATCH 3/7] fix(listener): re-read a change whose accept or end
 failed to record

When the database failed to record an agreement's on-chain accept, or its accept and end, the
listener moved past the change anyway. An end is the last change on-chain, so its accepted and
terminated events were lost for good. Now the change is read again, like others that fail.
---
 .../src/network/service/chain_listener.rs     | 247 ++++++++++++------
 .../tests/it_registry_postgres.rs             |  11 +
 2 files changed, 184 insertions(+), 74 deletions(-)

diff --git a/bin/dipper-service/src/network/service/chain_listener.rs b/bin/dipper-service/src/network/service/chain_listener.rs
index 3e90e78a..1dd12227 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.
@@ -623,7 +624,7 @@ where
                 Ok(None) => {}
                 Err(err) => {
                     tracing::warn!(
-                        error = %err,
+                        error = format!("{err:#}"),
                         agreement_id = %snapshot.agreement_id,
                         "Failed to prepare reconciliation for snapshot"
                     );
@@ -907,30 +908,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 +982,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<R: AgreementRegistry + Sync>(
     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 +1008,8 @@ async fn record_accept_and_cancel_from_chain<R: AgreementRegistry + Sync>(
             &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 +1039,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<R: AgreementRegistry + Sync>(
     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 +1931,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<crate::registry::PendingCancellation>,
@@ -2250,7 +2229,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 +2285,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 +3269,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,
-            &registry,
-            &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 +4331,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<Cursor>) {
+        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<u64> = 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,
+                &registry,
+                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();
diff --git a/dipper-pgregistry/tests/it_registry_postgres.rs b/dipper-pgregistry/tests/it_registry_postgres.rs
index bbbabfe4..17ff4501 100644
--- a/dipper-pgregistry/tests/it_registry_postgres.rs
+++ b/dipper-pgregistry/tests/it_registry_postgres.rs
@@ -4157,6 +4157,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

From 2446d4d83392ba62e48b0e279d924de941d20748 Mon Sep 17 00:00:00 2001
From: MoonBoi9001 <contact@samuelmetcalfe.com>
Date: Mon, 12 Oct 2026 16:24:01 +0100
Subject: [PATCH 4/7] fix(liveness): give a held indexer a full tolerance once
 outages end

While most indexers can't be reached, dipper holds off abandoning them but their clocks keep
running, so after a long outage any still unreachable on the first good cycle was dropped.
Now the hold lifts after 3 good cycles in a row, and idle time counts from the last held cycle.
---
 .../src/network/service/liveness_checker.rs   | 308 ++++++++++++++----
 1 file changed, 238 insertions(+), 70 deletions(-)

diff --git a/bin/dipper-service/src/network/service/liveness_checker.rs b/bin/dipper-service/src/network/service/liveness_checker.rs
index ac809c90..053bf879 100644
--- a/bin/dipper-service/src/network/service/liveness_checker.rs
+++ b/bin/dipper-service/src/network/service/liveness_checker.rs
@@ -32,7 +32,9 @@
 //!
 //! 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 that cycle abandons no agreement for
-//! unreachability; indexers that answered without progress are still handled as above.
+//! unreachability; indexers that answered without progress are still handled as above. Once
+//! 3 cycles in a row see fewer, 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
 //!
@@ -215,7 +217,12 @@ where
                 .iter()
                 .filter(|(_, probe)| matches!(probe, Probe::Unreachable))
                 .count();
-            let hold_unreachable = breaker.evaluate(probed.len(), unreachable, &config);
+            let policy = breaker.evaluate(
+                probed.len(),
+                unreachable,
+                OffsetDateTime::now_utc(),
+                &config,
+            );
 
             // 4. Act on each indexer's agreements
             let checker = Checker {
@@ -234,12 +241,7 @@ where
                     return Ok(());
                 }
                 checker
-                    .check_indexer(
-                        &mut uncancellable,
-                        group_agreements,
-                        &probe,
-                        hold_unreachable,
-                    )
+                    .check_indexer(&mut uncancellable, group_agreements, &probe, policy)
                     .await;
             }
 
@@ -349,24 +351,52 @@ fn unreachable_trips(
     checked > 0 && checked >= min_indexers && unreachable as f64 / checked as f64 >= trip_fraction
 }
 
-/// Stops agreements being abandoned because their indexer can't be reached while so many
-/// indexers can't be reached at once that the fault is likely dipper's own (its network, or the
-/// network registry). Abandoning them all would hand every deployment to new indexers and keep
-/// the old ones out for weeks.
+/// 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<OffsetDateTime> },
+}
+
+impl UnreachablePolicy {
+    /// When an unreachable indexer's agreement counts its time without progress from.
+    fn progress_clock(self, last_progress_at: Option<OffsetDateTime>) -> Option<OffsetDateTime> {
+        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;
+
+/// 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 {
     tripped: bool,
+    /// 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<OffsetDateTime>,
 }
 
 impl UnreachableBreaker {
-    /// Whether this cycle holds back abandoning agreements for unreachability, logging an ERROR
-    /// as it trips, a WARN for each later cycle it stays tripped and an INFO as it clears.
+    /// How this cycle treats agreements whose indexer can't be reached.
     fn evaluate(
         &mut self,
         checked: usize,
         unreachable: usize,
+        now: OffsetDateTime,
         config: &LivenessCheckerConfig,
-    ) -> bool {
+    ) -> UnreachablePolicy {
         let trip_fraction = config.unreachable_trip_fraction;
         let trips = unreachable_trips(
             checked,
@@ -374,31 +404,69 @@ impl UnreachableBreaker {
             trip_fraction,
             config.unreachable_trip_min_indexers,
         );
-        match (self.tripped, trips) {
-            (false, true) => tracing::error!(
-                event = "liveness_abandonment_paused",
+        let holds = if trips {
+            self.note_trip(unreachable, checked, trip_fraction);
+            true
+        } else {
+            self.tripped && !self.note_quiet_cycle(unreachable, checked, trip_fraction)
+        };
+        if !holds {
+            return UnreachablePolicy::Judge {
+                since: self.last_held,
+            };
+        }
+        self.tripped = true;
+        self.last_held = Some(now);
+        UnreachablePolicy::Hold
+    }
+
+    /// Note a cycle with enough indexers unreachable to trip the breaker: an ERROR as it trips,
+    /// a WARN while it stays tripped.
+    fn note_trip(&mut self, unreachable: usize, checked: usize, trip_fraction: f64) {
+        self.quiet_cycles = 0;
+        if self.tripped {
+            tracing::warn!(
                 unreachable,
                 checked,
                 trip_fraction,
-                "Too many indexers unreachable at once, likely a fault on dipper's side; not \
-                 abandoning agreements for unreachability until fewer are"
-            ),
-            (true, true) => tracing::warn!(
+                "Still too many indexers unreachable; not abandoning agreements for unreachability"
+            );
+        } else {
+            tracing::error!(
+                event = "liveness_abandonment_paused",
                 unreachable,
                 checked,
                 trip_fraction,
-                "Still too many indexers unreachable; not abandoning agreements for unreachability"
-            ),
-            (true, false) => tracing::info!(
+                "Too many indexers unreachable at once, likely a fault on dipper's side; not \
+                 abandoning agreements for unreachability until fewer are"
+            );
+        }
+    }
+
+    /// Note a cycle, while tripped, with too few indexers unreachable to trip it; true, logged,
+    /// once enough of those in a row clear it.
+    fn note_quiet_cycle(&mut self, unreachable: usize, checked: usize, trip_fraction: f64) -> bool {
+        self.quiet_cycles += 1;
+        if self.quiet_cycles < UNREACHABLE_CLEAR_CYCLES {
+            tracing::info!(
                 unreachable,
                 checked,
-                trip_fraction,
-                "Fewer indexers unreachable; abandoning agreements for unreachability resumes"
-            ),
-            (false, false) => {}
+                quiet_cycles = self.quiet_cycles,
+                "Fewer indexers unreachable; still not abandoning agreements for unreachability \
+                 until it stays so"
+            );
+            return false;
         }
-        self.tripped = trips;
-        trips
+        self.tripped = false;
+        self.quiet_cycles = 0;
+        tracing::info!(
+            unreachable,
+            checked,
+            trip_fraction,
+            "Fewer indexers unreachable; abandoning agreements for unreachability resumes, each \
+             after a full tolerance from now without progress"
+        );
+        true
     }
 }
 
@@ -420,52 +488,47 @@ where
     W: WorkerQueue + Send + Sync,
     C: ChainClient + Send + Sync,
 {
-    /// Act on 1 indexer's agreements given what its status endpoint said. With
-    /// `hold_unreachable` set, an unreachable indexer's stale agreements are left as they are,
-    /// their progress clock untouched, so they are judged as before once the breaker clears.
+    /// 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<IndexingAgreement>,
         probe: &Probe,
-        hold_unreachable: bool,
+        policy: UnreachablePolicy,
     ) {
-        let unreachable = matches!(probe, Probe::Unreachable);
+        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,
-                unreachable && hold_unreachable,
-            )
-            .await;
+            self.check_agreement(uncancellable, &agreement, current_block, unreachable)
+                .await;
         }
     }
 
-    /// Act on 1 agreement given the latest block its indexer reports, if any; with `hold` set,
-    /// leave it as it is however long it has gone without progress.
+    /// 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<u64>,
-        unreachable: bool,
-        hold: bool,
+        unreachable: Option<UnreachablePolicy>,
     ) {
         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,
             current_block,
-            agreement.last_progress_at,
+            last_progress_at,
             now,
             deployment_id,
             self.active_counts,
             self.max_tolerance_days,
         );
-        let elapsed_hours = (now - agreement.last_progress_at.unwrap_or(now)).whole_hours();
+        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();
@@ -476,23 +539,25 @@ where
             LivenessAction::WithinTolerance => tracing::debug!(
                 agreement_id = %agreement.id,
                 deployment = %deployment_id,
-                unreachable,
+                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 hold => tracing::debug!(
-                agreement_id = %agreement.id,
-                elapsed_hours,
-                threshold_hours,
-                "indexer unreachable past tolerance; not abandoning while too many are"
-            ),
+            LivenessAction::CancelAndReassess if unreachable == Some(UnreachablePolicy::Hold) => {
+                tracing::debug!(
+                    agreement_id = %agreement.id,
+                    elapsed_hours,
+                    threshold_hours,
+                    "indexer unreachable past tolerance; not abandoning while too many are"
+                );
+            }
             LivenessAction::CancelAndReassess => {
                 self.abandon(
                     uncancellable,
                     agreement,
-                    unreachable,
+                    unreachable.is_some(),
                     elapsed_hours,
                     threshold_hours,
                 )
@@ -1029,9 +1094,9 @@ mod tests {
     use url::Url;
 
     use super::{
-        Checker, LivenessAction, Probe, Uncancellable, UnreachableBreaker, cancel_and_reassess,
-        decide_liveness_action, group_by_indexer_id, record_progress, tolerance_duration,
-        unreachable_trips,
+        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},
@@ -1929,14 +1994,27 @@ mod tests {
         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 tripped = logged(async { assert!(breaker.evaluate(4, 3, &config)) }).await;
-        let still = logged(async { assert!(breaker.evaluate(4, 4, &config)) }).await;
-        let cleared = logged(async { assert!(!breaker.evaluate(4, 0, &config)) }).await;
+        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}");
@@ -1949,6 +2027,31 @@ mod tests {
         );
     }
 
+    #[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");
+    }
+
     /// 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();
@@ -1961,7 +2064,7 @@ mod tests {
         registry: &MockRegistry,
         chain: &MockChainClient,
         probe: &Probe,
-        hold_unreachable: bool,
+        policy: UnreachablePolicy,
     ) {
         let queue = MockWorkerQueue {
             calls: registry.calls.clone(),
@@ -1983,7 +2086,7 @@ mod tests {
                 &mut Uncancellable::default(),
                 vec![agreement.clone()],
                 probe,
-                hold_unreachable,
+                policy,
             )
             .await;
     }
@@ -1997,7 +2100,14 @@ mod tests {
         let deployment = agreement.terms.metadata.subgraph_deployment_id.to_string();
         let stalled = Probe::Reached(HashMap::from([(deployment, agreement.last_block_height)]));
 
-        check(&agreement, &registry, &chain, &stalled, true).await;
+        check(
+            &agreement,
+            &registry,
+            &chain,
+            &stalled,
+            UnreachablePolicy::Hold,
+        )
+        .await;
 
         assert_eq!(calls.abandoning.lock().unwrap().as_slice(), &[agreement.id]);
     }
@@ -2009,7 +2119,8 @@ mod tests {
         let registry = MockRegistry::new(calls.clone(), agreement.clone());
         let chain = MockChainClient::success(calls.clone());
 
-        check(&agreement, &registry, &chain, &Probe::Unreachable, true).await;
+        let hold = UnreachablePolicy::Hold;
+        check(&agreement, &registry, &chain, &Probe::Unreachable, hold).await;
 
         assert!(calls.abandoning.lock().unwrap().is_empty());
         assert!(calls.chain_cancels.lock().unwrap().is_empty());
@@ -2026,7 +2137,64 @@ mod tests {
         let registry = MockRegistry::new(calls.clone(), agreement.clone());
         let chain = MockChainClient::success(calls.clone());
 
-        check(&agreement, &registry, &chain, &Probe::Unreachable, false).await;
+        let judge = UnreachablePolicy::Judge { since: None };
+        check(&agreement, &registry, &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, &registry, &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, &registry, &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, &registry, &chain, &stalled, cleared).await;
 
         assert_eq!(calls.abandoning.lock().unwrap().as_slice(), &[agreement.id]);
     }

From 65e9ee5eae50aca8b455373e16d11ec9113a5c80 Mon Sep 17 00:00:00 2001
From: MoonBoi9001 <contact@samuelmetcalfe.com>
Date: Mon, 12 Oct 2026 16:26:24 +0100
Subject: [PATCH 5/7] fix(liveness): stop holding unreachable indexers after 4
 days

When half of a few indexers are really gone, dipper reads it as its own outage and never ends
their agreements, alerting just once. It now holds them at most the longest tolerance (4 days),
repeats the alert every 6 hours, and the alert no longer rules out indexers being really gone.
---
 bin/dipper-service/src/config.rs              |   5 +-
 .../src/network/service/liveness_checker.rs   | 212 ++++++++++++++----
 2 files changed, 168 insertions(+), 49 deletions(-)

diff --git a/bin/dipper-service/src/config.rs b/bin/dipper-service/src/config.rs
index e2488e79..c24207d4 100644
--- a/bin/dipper-service/src/config.rs
+++ b/bin/dipper-service/src/config.rs
@@ -544,8 +544,9 @@ pub struct LivenessCheckerConfig {
     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 that cycle abandoning agreements for
-    /// unreachability, since so many points to a fault on dipper's side (default: 0.5).
+    /// 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,
 
diff --git a/bin/dipper-service/src/network/service/liveness_checker.rs b/bin/dipper-service/src/network/service/liveness_checker.rs
index 053bf879..bf0a59eb 100644
--- a/bin/dipper-service/src/network/service/liveness_checker.rs
+++ b/bin/dipper-service/src/network/service/liveness_checker.rs
@@ -31,10 +31,11 @@
 //! - 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 that cycle abandons no agreement for
-//! unreachability; indexers that answered without progress are still handled as above. Once
-//! 3 cycles in a row see fewer, 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.
+//! 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
 //!
@@ -376,20 +377,27 @@ impl UnreachablePolicy {
 /// 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 {
-    tripped: bool,
+    /// When the current trip began; `None` while clear.
+    tripped_since: Option<OffsetDateTime>,
     /// 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<OffsetDateTime>,
+    /// When an operator was last alerted to the current trip.
+    alerted_at: Option<OffsetDateTime>,
 }
 
 impl UnreachableBreaker {
-    /// How this cycle treats agreements whose indexer can't be reached.
+    /// 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,
@@ -397,76 +405,128 @@ impl UnreachableBreaker {
         now: OffsetDateTime,
         config: &LivenessCheckerConfig,
     ) -> UnreachablePolicy {
-        let trip_fraction = config.unreachable_trip_fraction;
         let trips = unreachable_trips(
             checked,
             unreachable,
-            trip_fraction,
+            config.unreachable_trip_fraction,
             config.unreachable_trip_min_indexers,
         );
-        let holds = if trips {
-            self.note_trip(unreachable, checked, trip_fraction);
-            true
-        } else {
-            self.tripped && !self.note_quiet_cycle(unreachable, checked, trip_fraction)
+        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,
         };
-        if !holds {
-            return 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.tripped = true;
+        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: an ERROR as it trips,
-    /// a WARN while it stays tripped.
-    fn note_trip(&mut self, unreachable: usize, checked: usize, trip_fraction: f64) {
+    /// 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 {
-            tracing::warn!(
-                unreachable,
-                checked,
-                trip_fraction,
-                "Still too many indexers unreachable; not abandoning agreements for unreachability"
-            );
-        } else {
-            tracing::error!(
-                event = "liveness_abandonment_paused",
-                unreachable,
-                checked,
-                trip_fraction,
-                "Too many indexers unreachable at once, likely a fault on dipper's side; not \
-                 abandoning agreements for unreachability until fewer are"
-            );
+        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; true, logged,
-    /// once enough of those in a row clear it.
-    fn note_quiet_cycle(&mut self, unreachable: usize, checked: usize, trip_fraction: f64) -> bool {
+    /// 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; still not abandoning agreements for unreachability \
-                 until it stays so"
+                "Fewer indexers unreachable; the breaker clears if it stays so"
             );
-            return false;
+            return;
         }
-        self.tripped = false;
+        self.tripped_since = None;
         self.quiet_cycles = 0;
         tracing::info!(
             unreachable,
             checked,
-            trip_fraction,
             "Fewer indexers unreachable; abandoning agreements for unreachability resumes, each \
-             after a full tolerance from now without progress"
+             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"
         );
-        true
     }
 }
 
@@ -2052,6 +2112,64 @@ mod tests {
         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();

From d11bb8d8b7257cc4285c7683f0772779f2a9c2fc Mon Sep 17 00:00:00 2001
From: MoonBoi9001 <contact@samuelmetcalfe.com>
Date: Mon, 12 Oct 2026 16:27:51 +0100
Subject: [PATCH 6/7] fix(liveness): ask every indexer with an accepted
 agreement each cycle

Each cycle checks the 500 agreements longest without progress, which can leave out most healthy
indexers, so dead ones looked like half the network and paused abandonment with no outage.
Now 1 agreement of every indexer the batch leaves out is checked too, so the share counts all.
---
 bin/dipper-service/src/config.rs              |  3 +-
 bin/dipper-service/src/registry/agreement.rs  |  3 +-
 dipper-pgregistry/src/postgres.rs             | 23 +++++++--
 .../tests/it_registry_postgres.rs             | 48 +++++++++++++++++++
 4 files changed, 72 insertions(+), 5 deletions(-)

diff --git a/bin/dipper-service/src/config.rs b/bin/dipper-service/src/config.rs
index c24207d4..661ede0b 100644
--- a/bin/dipper-service/src/config.rs
+++ b/bin/dipper-service/src/config.rs
@@ -539,7 +539,8 @@ 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,
 
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<Vec<IndexingAgreement>, 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 17ff4501..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<IndexingAgreementId> = 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

From 6732c278ce56dfb9e1daae2f91c6a3cffa0475b4 Mon Sep 17 00:00:00 2001
From: MoonBoi9001 <contact@samuelmetcalfe.com>
Date: Mon, 12 Oct 2026 16:29:01 +0100
Subject: [PATCH 7/7] fix(listener): list skipped agreement ids in 0x form, at
 most 20

The Slack alert for chain changes the listener gave up on printed agreement ids as raw byte
arrays nobody can search for, and a whole failed page put up to 1,000 of them in 1 message.
---
 .../src/network/service/chain_listener.rs     | 42 ++++++++++++++++++-
 1 file changed, 40 insertions(+), 2 deletions(-)

diff --git a/bin/dipper-service/src/network/service/chain_listener.rs b/bin/dipper-service/src/network/service/chain_listener.rs
index 1dd12227..071c2e30 100644
--- a/bin/dipper-service/src/network/service/chain_listener.rs
+++ b/bin/dipper-service/src/network/service/chain_listener.rs
@@ -109,6 +109,24 @@ fn hold_expired(hold: &mut Option<CursorHold>, 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<String> = 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;
@@ -724,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"
             );
@@ -732,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"
             );
         }
@@ -4476,6 +4496,24 @@ mod tests {
         assert!(hold.is_none());
     }
 
+    #[test]
+    fn lists_at_most_20_agreement_ids_in_their_0x_form() {
+        let ids: Vec<IndexingAgreementId> = (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