fix: end stale agreements through the cancelling status

bin/dipper-service/src/cancel_dispatch.rs

@@ -1,5 +1,5 @@
-//! On-chain cancel dispatch. Every cancel goes through
-//! [`cancel_agreement_on_chain`] so the manager-routed path lives in one place.
+//! On-chain cancel dispatch. Every cancel starts with [`start_cancel`] and goes out through
+//! `cancel_agreement_on_chain`, so the manager-routed path lives in one place.
 
 use dipper_core::time::now_secs;
 use thegraph_core::alloy::primitives::B256;
@@ -23,17 +23,12 @@ const SCOPE_BOTH: u16 = SCOPE_ACTIVE | SCOPE_PENDING;
 /// Cancel an agreement on-chain through the RecurringAgreementManager. Passes
 /// both scope bits so the collector cancels whichever scope the agreement is in,
 /// and treats a missing or short stored hash as `MissingTermsVersionHash`.
-pub async fn cancel_agreement_on_chain<T: ChainClient>(
+async fn cancel_agreement_on_chain<T: ChainClient>(
     chain_client: &T,
     agreement: &IndexingAgreement,
     config: &IndexingAgreementConfig,
 ) -> LiveCancel {
-    let Some(version_hash) = agreement
-        .terms_version_hash
-        .as_deref()
-        .filter(|h| h.len() == 32)
-        .map(B256::from_slice)
-    else {
+    let Some(version_hash) = cancel_hash(agreement) else {
         return LiveCancel::CancelFailed(ChainClientError::MissingTermsVersionHash {
             agreement_id: agreement.id.to_string(),
         });
@@ -68,10 +63,39 @@ pub async fn cancel_agreement_on_chain<T: ChainClient>(
     }
 }
 
+/// The stored terms hash an on-chain cancel needs, or `None` when the agreement has no 32-byte
+/// one, so no cancel can ever be sent for it.
+pub fn cancel_hash(agreement: &IndexingAgreement) -> Option<B256> {
+    agreement
+        .terms_version_hash
+        .as_deref()
+        .filter(|h| h.len() == 32)
+        .map(B256::from_slice)
+}
+
+/// Why dipper is ending an agreement, which decides the status it ends in once the chain
+/// confirms dipper's cancel.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum CancelReason {
+    /// Dipper no longer wants it: it ends `CanceledByRequester`.
+    NotWanted,
+    /// Its indexer stopped serving it: it ends `AbandonedByIndexer`.
+    Abandoned,
+}
+
+impl CancelReason {
+    fn ended_status(self) -> &'static str {
+        match self {
+            Self::NotWanted => "CANCELED_BY_REQUESTER",
+            Self::Abandoned => "ABANDONED_BY_INDEXER",
+        }
+    }
+}
+
 /// What [`start_cancel`] left an agreement as.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum CancelStarted {
-    /// It was accepted and its cancel landed: now `CanceledByRequester`.
+    /// It was accepted and its cancel landed: now ended, as its [`CancelReason`] says.
     Ended,
     /// Still `Cancelling`; the chain listener finishes it once it can't go live.
     Cancelling,
@@ -84,15 +108,25 @@ pub async fn start_cancel<R, T>(
     registry: &R,
     chain_client: &T,
     agreement: &IndexingAgreement,
+    reason: CancelReason,
     config: &IndexingAgreementConfig,
 ) -> RegistryResult<CancelStarted>
 where
     R: AgreementRegistry + Sync,
     T: ChainClient,
 {
-    registry
-        .mark_indexing_agreement_as_cancelling(&agreement.id)
-        .await?;
+    match reason {
+        CancelReason::NotWanted => {
+            registry
+                .mark_indexing_agreement_as_cancelling(&agreement.id)
+                .await?
+        }
+        CancelReason::Abandoned => {
+            registry
+                .mark_indexing_agreement_as_abandoning(&agreement.id)
+                .await?
+        }
+    }
     let tx_hash = match cancel_if_live(chain_client, agreement, config).await {
         LiveCancel::Ended(tx_hash) => tx_hash,
         LiveCancel::NotLive { .. } => return Ok(CancelStarted::Cancelling),
@@ -119,20 +153,21 @@ where
         return Ok(CancelStarted::Cancelling);
     }
     Ok(
-        if confirm_cancelled(registry, agreement, tx_hash, config).await {
+        if confirm_cancelled(registry, agreement, reason, tx_hash, config).await {
             CancelStarted::Ended
         } else {
             CancelStarted::Cancelling
         },
     )
 }
 
-/// Mark an agreement the chain shows dipper ended `CanceledByRequester`, recording the cancel
-/// when its transaction is known, so the `terminated` sweep announces it. False, logged, when
-/// the mark fails; it stays `Cancelling` for the cancel retry.
+/// Mark an agreement the chain shows dipper ended as ended, recording the cancel when its
+/// transaction is known, so the `terminated` sweep announces it. False, logged, when the mark
+/// fails; it stays `Cancelling` for the cancel retry.
 pub async fn confirm_cancelled<R: AgreementRegistry + Sync>(
     registry: &R,
     agreement: &IndexingAgreement,
+    reason: CancelReason,
     tx_hash: Option<B256>,
     config: &IndexingAgreementConfig,
 ) -> bool {
@@ -155,7 +190,7 @@ pub async fn confirm_cancelled<R: AgreementRegistry + Sync>(
         agreement_id = %agreement.id,
         indexing_request_id = %agreement.indexing_request_id,
         old_status = "CANCELLING",
-        new_status = "CANCELED_BY_REQUESTER",
+        new_status = reason.ended_status(),
         reason = "cancel_confirmed_on_chain",
         "agreement state transition"
     );

bin/dipper-service/src/network/service/cancel_retry.rs

@@ -1,12 +1,15 @@
 //! Finishes the cancels dipper starts. An agreement dipper wants ended is marked
 //! `Cancelling` before its on-chain cancel goes out; this sweep re-sends the cancel while
-//! the chain shows it live, and marks it `CanceledByRequester` once it can no longer be.
+//! the chain shows it live, and marks it ended once it can no longer be: `CanceledByRequester`,
+//! or `AbandonedByIndexer` for one dipper ended because its indexer stopped serving it.
 
 use dipper_core::time::now_secs;
 use thegraph_core::alloy::primitives::B256;
 
 use crate::{
-    cancel_dispatch::{LiveCancel, cancel_if_live, confirm_cancelled, log_unconfirmed},
+    cancel_dispatch::{
+        CancelReason, LiveCancel, cancel_if_live, confirm_cancelled, log_unconfirmed,
+    },
     chain_client::{ChainClient, ChainClientError},
     config::IndexingAgreementConfig,
     registry::{AgreementRegistry, CancelKind, CancellingAgreement, IndexingAgreement},
@@ -132,7 +135,7 @@ async fn retry_cancel<R, T>(
     note_check(registry, row, failure.as_ref(), Some(failure.is_none())).await;
 }
 
-/// Mark the agreement `CanceledByRequester` once it can't go live again: this sweep's cancel
+/// Mark the agreement ended by dipper once it can't go live again: this sweep's cancel
 /// ended it, or nobody accepted its offer before the deadline to. One ended otherwise is left
 /// to the chain listener for a while; one the indexer ended then becomes `CanceledByIndexer`.
 async fn confirm_if_over<R: AgreementRegistry + Sync>(
@@ -162,7 +165,12 @@ async fn confirm_if_over<R: AgreementRegistry + Sync>(
         );
         return past_grace && record_end_by_indexer(registry, agreement).await;
     }
-    confirm_cancelled(registry, agreement, tx_hash, config).await
+    let reason = if row.abandoned {
+        CancelReason::Abandoned
+    } else {
+        CancelReason::NotWanted
+    };
+    confirm_cancelled(registry, agreement, reason, tx_hash, config).await
 }
 
 /// Mark an agreement the indexer ended `CanceledByIndexer` when the chain listener hasn't in
@@ -229,7 +237,7 @@ async fn note_check<R: AgreementRegistry + Sync>(
     {
         Ok(attempts) => {
             if let Some(err) = failure.filter(|_| failed_attempts > 0) {
-                log_failed_cancel(agreement, attempts, failed_attempts, err);
+                log_failed_cancel(row, attempts, failed_attempts, err);
             }
         }
         Err(err) => tracing::warn!(
@@ -251,11 +259,12 @@ fn log_uncounted_failure(agreement: &IndexingAgreement, err: &ChainClientError)
 /// One ERROR as an agreement reaches the limit, for an operator to look into; a WARN for
 /// every other failed cancel.
 fn log_failed_cancel(
-    agreement: &IndexingAgreement,
+    row: &CancellingAgreement,
     attempts: u32,
     failed: u32,
     err: &ChainClientError,
 ) {
+    let agreement = &row.agreement;
     let reached_limit =
         attempts >= MAX_CANCEL_ATTEMPTS && attempts.saturating_sub(failed) < MAX_CANCEL_ATTEMPTS;
     if !reached_limit {
@@ -272,6 +281,7 @@ fn log_failed_cancel(
         agreement_id = %agreement.id,
         indexer_id = %agreement.indexer.id,
         indexing_request_id = %agreement.indexing_request_id,
+        abandoned = row.abandoned,
         attempts,
         error = %err,
         "Cancelling an agreement keeps failing; it may still be live. Dipper now retries it hourly"
@@ -497,6 +507,7 @@ mod tests {
                 agreement: cancelling,
                 accepted_on_chain,
                 ended_seen_at: None,
+                abandoned: false,
             }],
             ..MockRegistry::default()
         }

bin/dipper-service/src/network/service/chain_listener.rs

@@ -1086,7 +1086,7 @@ where
         match prep.item.cancel {
             Some(CancelKind::ByRequester) => tracing::info!(
                 agreement_id = %prep.agreement.id,
-                "Agreement marked as CanceledByRequester (on-chain confirmation)"
+                "Agreement marked as ended by dipper (on-chain confirmation)"
             ),
             Some(CancelKind::ByIndexer) => tracing::info!(
                 agreement_id = %prep.agreement.id,
@@ -1224,8 +1224,14 @@ where
             None => return Ok(false),
         }
     }
-    let started =
-        crate::cancel_dispatch::start_cancel(registry, chain_client, &old_agreement, config).await;
+    let started = crate::cancel_dispatch::start_cancel(
+        registry,
+        chain_client,
+        &old_agreement,
+        crate::cancel_dispatch::CancelReason::NotWanted,
+        config,
+    )
+    .await;
     Ok(note_replaced_cancel(
         new_agreement_id,
         &old_agreement,
@@ -1314,8 +1320,14 @@ async fn sweep_orphan_canceled_agreements<R, T>(
     };
 
     for agreement in orphans {
-        let started =
-            crate::cancel_dispatch::start_cancel(registry, chain_client, &agreement, config).await;
+        let started = crate::cancel_dispatch::start_cancel(
+            registry,
+            chain_client,
+            &agreement,
+            crate::cancel_dispatch::CancelReason::NotWanted,
+            config,
+        )
+        .await;
         log_orphan_cancel(&agreement, started);
     }
 }
@@ -2439,13 +2451,6 @@ mod tests {
             Ok((per_indexer, global))
         }
 
-        async fn mark_indexing_agreement_as_abandoned(
-            &self,
-            _id: &IndexingAgreementId,
-        ) -> RegistryResult<IndexingAgreement> {
-            Err(crate::registry::Error::NoRecordsUpdated)
-        }
-
         async fn get_agreement_fee_rates(&self) -> RegistryResult<Vec<AgreementFeeRate>> {
             Ok(vec![])
         }