diff --git a/bin/dipper-service/src/cancel_dispatch.rs b/bin/dipper-service/src/cancel_dispatch.rs index cd068be4..4fd15e0a 100644 --- a/bin/dipper-service/src/cancel_dispatch.rs +++ b/bin/dipper-service/src/cancel_dispatch.rs @@ -6,7 +6,7 @@ use thegraph_core::alloy::primitives::B256; use crate::{ chain_client::{ChainClient, ChainClientError}, config::IndexingAgreementConfig, - registry::IndexingAgreement, + registry::{AgreementRegistry, IndexingAgreement}, }; /// Pass both ACTIVE and PENDING; local status lags the chain, so let the @@ -16,22 +16,59 @@ const SCOPE_ACTIVE: u16 = 1; const SCOPE_PENDING: u16 = 2; 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( +/// Resolve the version hash to cancel with: the locally stored one, or — if +/// missing or malformed (e.g. a pre-migration row with no `terms_version_hash` +/// column value) — the authoritative one read back from +/// `getAgreementDetails`. A recovered hash is best-effort persisted to the +/// registry so future cancels don't need to re-fetch it; a persistence +/// failure is logged and otherwise ignored, since the recovered hash is +/// still used for this call regardless. +async fn resolve_version_hash( chain_client: &T, + registry: &R, agreement: &IndexingAgreement, - config: &IndexingAgreementConfig, -) -> Result, ChainClientError> { - let version_hash = agreement +) -> Result { + if let Some(hash) = agreement .terms_version_hash .as_deref() .filter(|h| h.len() == 32) .map(B256::from_slice) + { + return Ok(hash); + } + + let recovered = chain_client + .fetch_agreement_version_hash(agreement.id.as_bytes()) + .await? .ok_or_else(|| ChainClientError::MissingTermsVersionHash { agreement_id: agreement.id.to_string(), })?; + + if let Err(err) = registry + .update_terms_version_hash(&agreement.id, recovered.as_slice().try_into().unwrap()) + .await + { + tracing::warn!( + agreement_id = %agreement.id, + error = %err, + "recovered terms_version_hash from chain but failed to persist it; will \ + re-recover on the next cancel attempt" + ); + } + Ok(recovered) +} + +/// Cancel an agreement on-chain through the RecurringAgreementManager. Passes +/// both scope bits so the collector cancels whichever scope the agreement is in. +/// If the local `terms_version_hash` is missing, first tries to recover it from +/// `getAgreementDetails` before giving up with `MissingTermsVersionHash`. +pub async fn cancel_agreement_on_chain( + chain_client: &T, + registry: &R, + agreement: &IndexingAgreement, + config: &IndexingAgreementConfig, +) -> Result, ChainClientError> { + let version_hash = resolve_version_hash(chain_client, registry, agreement).await?; // Hazard: the manager's cancel mines successfully even when it does nothing // (stale/wrong hash, unknown id, already-terminal). So after a submitted // cancel we re-read on-chain and surface CancelNotConfirmed if still active. @@ -78,21 +115,48 @@ mod tests { config::IndexingAgreementConfig, registry::{ IndexingAgreement, IndexingAgreementStatus, IndexingAgreementTerms, - IndexingAgreementTermsMetadata, + IndexingAgreementTermsMetadata, StubAgreementRegistry, }, }; + /// Panic-by-default registry: fine for every test that never exercises + /// the missing-hash recovery path (the only registry call dispatch makes). + struct StubRegistry; + impl StubAgreementRegistry for StubRegistry {} + + /// Records `update_terms_version_hash` calls for the recovery tests. + #[derive(Default)] + struct RecordingRegistry { + persisted_hashes: Mutex>, + } + + #[async_trait] + impl StubAgreementRegistry for RecordingRegistry { + async fn update_terms_version_hash( + &self, + id: &IndexingAgreementId, + hash: &[u8; 32], + ) -> crate::registry::Result<()> { + self.persisted_hashes.lock().unwrap().push((*id, *hash)); + Ok(()) + } + } + /// (collector, agreement_id, version_hash, options) per manager cancel. type ManagerCancelArgs = (Address, [u8; 16], B256, u16); /// Records which on-chain cancel ran and with what arguments. /// `still_active_after_cancel` is the post-cancel verification read result; /// `active_reads` counts how many times that read fired. + /// `on_chain_version_hash` is what a `fetch_agreement_version_hash` recovery + /// read returns; `version_hash_reads` counts how many times it fired. #[derive(Default)] struct RecordingChainClient { manager_cancels: Mutex>, still_active_after_cancel: bool, active_reads: Mutex, + on_chain_version_hash: Option, + version_hash_reads: Mutex, } #[async_trait] @@ -142,6 +206,14 @@ mod tests { *self.active_reads.lock().unwrap() += 1; Ok(self.still_active_after_cancel) } + + async fn fetch_agreement_version_hash( + &self, + _agreement_id: &[u8; 16], + ) -> Result, ChainClientError> { + *self.version_hash_reads.lock().unwrap() += 1; + Ok(self.on_chain_version_hash) + } } fn manager_conf(collector: Address) -> IndexingAgreementConfig { @@ -224,7 +296,7 @@ mod tests { Some(vec![7u8; 32]), ); - cancel_agreement_on_chain(&client, &ag, &manager_conf(collector)) + cancel_agreement_on_chain(&client, &StubRegistry, &ag, &manager_conf(collector)) .await .expect("cancel dispatch"); @@ -246,7 +318,7 @@ mod tests { let client = RecordingChainClient::default(); let ag = agreement(IndexingAgreementStatus::Rejected, Some(vec![9u8; 32])); - cancel_agreement_on_chain(&client, &ag, &manager_conf(Address::ZERO)) + cancel_agreement_on_chain(&client, &StubRegistry, &ag, &manager_conf(Address::ZERO)) .await .expect("cancel dispatch"); @@ -263,9 +335,10 @@ mod tests { let client = RecordingChainClient::default(); let ag = agreement(IndexingAgreementStatus::AcceptedOnChain, None); - let err = cancel_agreement_on_chain(&client, &ag, &manager_conf(Address::ZERO)) - .await - .unwrap_err(); + let err = + cancel_agreement_on_chain(&client, &StubRegistry, &ag, &manager_conf(Address::ZERO)) + .await + .unwrap_err(); assert!(matches!( err, @@ -283,9 +356,10 @@ mod tests { Some(vec![1u8; 16]), ); - let err = cancel_agreement_on_chain(&client, &ag, &manager_conf(Address::ZERO)) - .await - .unwrap_err(); + let err = + cancel_agreement_on_chain(&client, &StubRegistry, &ag, &manager_conf(Address::ZERO)) + .await + .unwrap_err(); assert!(matches!( err, @@ -308,9 +382,10 @@ mod tests { Some(vec![7u8; 32]), ); - let err = cancel_agreement_on_chain(&client, &ag, &manager_conf(Address::ZERO)) - .await - .unwrap_err(); + let err = + cancel_agreement_on_chain(&client, &StubRegistry, &ag, &manager_conf(Address::ZERO)) + .await + .unwrap_err(); assert!(matches!(err, ChainClientError::CancelNotConfirmed { .. })); assert_eq!(client.manager_cancels.lock().unwrap().len(), 1); @@ -330,12 +405,73 @@ mod tests { Some(vec![7u8; 32]), ); - let out = cancel_agreement_on_chain(&client, &ag, &manager_conf(Address::ZERO)) - .await - .expect("cancel confirmed"); + let out = + cancel_agreement_on_chain(&client, &StubRegistry, &ag, &manager_conf(Address::ZERO)) + .await + .expect("cancel confirmed"); assert!(out.is_some()); assert_eq!(client.manager_cancels.lock().unwrap().len(), 1); assert_eq!(*client.active_reads.lock().unwrap(), 1, "verified once"); } + + #[tokio::test] + async fn manager_cancel_recovers_missing_hash_from_chain_and_persists_it() { + // #638 item 3: a row with no local terms_version_hash (e.g. pre-migration) + // must not be permanently uncancelable. If the RecurringCollector still + // has the hash on record, recover it from there, use it for this cancel, + // and best-effort persist it so future cancels don't need to re-fetch. + let recovered_hash = B256::from_slice(&[3u8; 32]); + let client = RecordingChainClient { + on_chain_version_hash: Some(recovered_hash), + still_active_after_cancel: false, + ..Default::default() + }; + let registry = RecordingRegistry::default(); + let ag = agreement(IndexingAgreementStatus::AcceptedOnChain, None); + + let out = cancel_agreement_on_chain(&client, ®istry, &ag, &manager_conf(Address::ZERO)) + .await + .expect("recovered hash unblocks the cancel"); + + assert!(out.is_some()); + assert_eq!( + *client.version_hash_reads.lock().unwrap(), + 1, + "recovery read fired once" + ); + let calls = client.manager_cancels.lock().unwrap(); + assert_eq!(calls.len(), 1); + assert_eq!(calls[0].2, recovered_hash, "cancel used the recovered hash"); + + let persisted = registry.persisted_hashes.lock().unwrap(); + assert_eq!(persisted.len(), 1); + assert_eq!(persisted[0], (ag.id, *recovered_hash)); + } + + #[tokio::test] + async fn manager_cancel_missing_hash_with_nothing_on_chain_is_still_missing_hash_error() { + // The contract itself has no versionHash on record either (e.g. the + // agreement was never offered) — recovery has nothing to recover, so + // this must still surface as MissingTermsVersionHash, not attempt a + // cancel with a zero hash. + let client = RecordingChainClient { + on_chain_version_hash: None, + ..Default::default() + }; + let registry = RecordingRegistry::default(); + let ag = agreement(IndexingAgreementStatus::AcceptedOnChain, None); + + let err = cancel_agreement_on_chain(&client, ®istry, &ag, &manager_conf(Address::ZERO)) + .await + .unwrap_err(); + + assert!(matches!( + err, + ChainClientError::MissingTermsVersionHash { .. } + )); + assert_eq!(*client.version_hash_reads.lock().unwrap(), 1); + assert!(client.manager_cancels.lock().unwrap().is_empty()); + assert!(registry.persisted_hashes.lock().unwrap().is_empty()); + } } diff --git a/bin/dipper-service/src/chain_client.rs b/bin/dipper-service/src/chain_client.rs index bf39bcd1..57d584a9 100644 --- a/bin/dipper-service/src/chain_client.rs +++ b/bin/dipper-service/src/chain_client.rs @@ -119,6 +119,18 @@ pub trait ChainClient { agreement_id: &[u8; 16], ) -> Result; + /// Read the authoritative `versionHash` the RecurringCollector stored for + /// this agreement via `getAgreementDetails(id, VERSION_CURRENT)`. Returns + /// `None` if the contract has no hash on record (a zero `versionHash` — + /// the agreement was never offered, or the id is unknown), `Some(hash)` + /// otherwise. Used to recover a `terms_version_hash` missing locally + /// (e.g. a row from before the column existed) instead of leaving the + /// agreement permanently uncancelable. + async fn fetch_agreement_version_hash( + &self, + agreement_id: &[u8; 16], + ) -> Result, ChainClientError>; + /// Read the latest block's unix timestamp from the chain. Lets agreement /// deadlines be stamped from live chain time when the chain-clock bypass is /// on, instead of a cached listener timestamp that can lag a fast chain. @@ -165,6 +177,13 @@ impl ChainClient for Arc { (**self).agreement_still_active(agreement_id).await } + async fn fetch_agreement_version_hash( + &self, + agreement_id: &[u8; 16], + ) -> Result, ChainClientError> { + (**self).fetch_agreement_version_hash(agreement_id).await + } + async fn latest_block_timestamp(&self) -> Result { (**self).latest_block_timestamp().await } diff --git a/bin/dipper-service/src/chain_client/client.rs b/bin/dipper-service/src/chain_client/client.rs index d25ad58a..576d4726 100644 --- a/bin/dipper-service/src/chain_client/client.rs +++ b/bin/dipper-service/src/chain_client/client.rs @@ -585,6 +585,42 @@ impl AlloyChainClient { tokio::time::sleep(RECEIPT_POLL_INTERVAL).await; } } + + /// Read `getAgreementDetails(id, VERSION_CURRENT)` from the + /// RecurringCollector. Shared by `agreement_still_active` (checks + /// `state`) and `fetch_agreement_version_hash` (checks `versionHash`) so + /// there's one call site for the ABI encode/decode. + async fn get_agreement_details( + &self, + agreement_id: &[u8; 16], + ) -> Result { + let calldata = IRecurringCollector::getAgreementDetailsCall { + agreementId: FixedBytes::<16>::from_slice(agreement_id), + index: thegraph_core::alloy::primitives::U256::from(VERSION_CURRENT), + } + .abi_encode(); + + let collector = self.inner.recurring_collector_address; + let output = self + .inner + .rpc_pool + .execute("get_agreement_details", |provider| { + let calldata = calldata.clone(); + async move { + let tx = TransactionRequest::default() + .to(collector) + .input(calldata.into()); + provider.call(tx).await + } + }) + .await?; + + IRecurringCollector::getAgreementDetailsCall::abi_decode_returns(&output).map_err(|err| { + ChainClientError::RpcError(anyhow::anyhow!( + "undecodable getAgreementDetails from {collector}: {err}" + )) + }) + } } #[async_trait] @@ -710,33 +746,7 @@ impl ChainClient for AlloyChainClient { &self, agreement_id: &[u8; 16], ) -> Result { - let calldata = IRecurringCollector::getAgreementDetailsCall { - agreementId: FixedBytes::<16>::from_slice(agreement_id), - index: thegraph_core::alloy::primitives::U256::from(VERSION_CURRENT), - } - .abi_encode(); - - let collector = self.inner.recurring_collector_address; - let output = self - .inner - .rpc_pool - .execute("get_agreement_details", |provider| { - let calldata = calldata.clone(); - async move { - let tx = TransactionRequest::default() - .to(collector) - .input(calldata.into()); - provider.call(tx).await - } - }) - .await?; - - let details = IRecurringCollector::getAgreementDetailsCall::abi_decode_returns(&output) - .map_err(|err| { - ChainClientError::RpcError(anyhow::anyhow!( - "undecodable getAgreementDetails from {collector}: {err}" - )) - })?; + let details = self.get_agreement_details(agreement_id).await?; // Live iff the terms are accepted and no cancellation notice exists. // A cancel sets NOTICE_GIVEN while ACCEPTED stays set, so checking the @@ -745,6 +755,14 @@ impl ChainClient for AlloyChainClient { Ok(state & STATE_ACCEPTED != 0 && state & STATE_NOTICE_GIVEN == 0) } + async fn fetch_agreement_version_hash( + &self, + agreement_id: &[u8; 16], + ) -> Result, ChainClientError> { + let details = self.get_agreement_details(agreement_id).await?; + Ok(Some(details.versionHash).filter(|hash| *hash != B256::ZERO)) + } + async fn reconcile_provider( &self, collector: Address, diff --git a/bin/dipper-service/src/network/service/chain_listener.rs b/bin/dipper-service/src/network/service/chain_listener.rs index 6bdcba19..257dc538 100644 --- a/bin/dipper-service/src/network/service/chain_listener.rs +++ b/bin/dipper-service/src/network/service/chain_listener.rs @@ -1064,6 +1064,7 @@ where let mut on_chain_cancel_tx: Option = None; match crate::cancel_dispatch::cancel_agreement_on_chain( chain_client, + registry, &old_agreement, config, ) @@ -1213,8 +1214,13 @@ async fn sweep_orphan_canceled_agreements( for agreement in orphans { let mut on_chain_cancel_tx: Option = None; - match crate::cancel_dispatch::cancel_agreement_on_chain(chain_client, &agreement, config) - .await + match crate::cancel_dispatch::cancel_agreement_on_chain( + chain_client, + registry, + &agreement, + config, + ) + .await { Ok(Some(tx_hash)) => { tracing::info!( @@ -2070,6 +2076,14 @@ mod tests { Ok(()) } + async fn update_terms_version_hash( + &self, + _id: &IndexingAgreementId, + _hash: &[u8; 32], + ) -> RegistryResult<()> { + Ok(()) + } + async fn mark_indexing_agreement_as_canceled_by_requester( &self, id: &IndexingAgreementId, @@ -2474,6 +2488,18 @@ mod tests { // not-active here means "cancel confirmed", which these tests expect. Ok(false) } + + async fn fetch_agreement_version_hash( + &self, + _agreement_id: &[u8; 16], + ) -> Result< + Option, + crate::chain_client::ChainClientError, + > { + // These tests always construct agreements with a stored hash, so + // recovery is never exercised. + unimplemented!("not exercised by chain_listener tests") + } } #[async_trait::async_trait] diff --git a/bin/dipper-service/src/network/service/escrow_reconciler.rs b/bin/dipper-service/src/network/service/escrow_reconciler.rs index b611fa5e..75e4a8ef 100644 --- a/bin/dipper-service/src/network/service/escrow_reconciler.rs +++ b/bin/dipper-service/src/network/service/escrow_reconciler.rs @@ -251,6 +251,13 @@ mod tests { ) -> Result { unimplemented!() } + + async fn fetch_agreement_version_hash( + &self, + _agreement_id: &[u8; 16], + ) -> Result, ChainClientError> { + unimplemented!() + } } /// In-memory registry returning a fixed provider list. diff --git a/bin/dipper-service/src/network/service/liveness_checker.rs b/bin/dipper-service/src/network/service/liveness_checker.rs index 2d554e9f..ccb6a599 100644 --- a/bin/dipper-service/src/network/service/liveness_checker.rs +++ b/bin/dipper-service/src/network/service/liveness_checker.rs @@ -492,8 +492,13 @@ async fn cancel_and_reassess( { // 1. Cancel on-chain (mode-aware dispatch) let mut on_chain_cancel_tx: Option = None; - match crate::cancel_dispatch::cancel_agreement_on_chain(chain_client, agreement, agreement_conf) - .await + match crate::cancel_dispatch::cancel_agreement_on_chain( + chain_client, + registry, + agreement, + agreement_conf, + ) + .await { Ok(Some(tx_hash)) => { tracing::info!( @@ -1189,6 +1194,15 @@ mod tests { // not-active means "cancel confirmed", which these tests expect. Ok(false) } + + async fn fetch_agreement_version_hash( + &self, + _agreement_id: &[u8; 16], + ) -> Result, ChainClientError> { + // These tests always construct agreements with a stored hash, so + // recovery is never exercised. + unimplemented!("not exercised by liveness_checker tests") + } } const DB_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); diff --git a/bin/dipper-service/src/registry.rs b/bin/dipper-service/src/registry.rs index 0509cb78..3103801a 100644 --- a/bin/dipper-service/src/registry.rs +++ b/bin/dipper-service/src/registry.rs @@ -376,6 +376,17 @@ impl AgreementRegistry for RegistryProvider { .map_err(Into::into) } + async fn update_terms_version_hash( + &self, + id: &IndexingAgreementId, + hash: &[u8; 32], + ) -> RegistryResult<()> { + self.inner + .update_terms_version_hash(id, hash) + .await + .map_err(Into::into) + } + async fn mark_indexing_agreement_as_canceled_by_requester( &self, id: &IndexingAgreementId, diff --git a/bin/dipper-service/src/registry/agreement.rs b/bin/dipper-service/src/registry/agreement.rs index fac5d5fe..00dfe53f 100644 --- a/bin/dipper-service/src/registry/agreement.rs +++ b/bin/dipper-service/src/registry/agreement.rs @@ -256,6 +256,17 @@ pub trait AgreementRegistry { tx_hash: &[u8; 32], ) -> RegistryResult<()>; + /// Backfill a `terms_version_hash` recovered from on-chain state for an + /// agreement whose local copy is missing (e.g. a row from before the + /// column existed). Idempotent and safe to call even if a hash is + /// already stored — callers only invoke it when the local copy is + /// absent, so overwriting is not a concern in practice. + async fn update_terms_version_hash( + &self, + id: &IndexingAgreementId, + hash: &[u8; 32], + ) -> RegistryResult<()>; + /// Mark an indexing agreement as `CANCELED_BY_REQUESTER`. /// /// If there is no indexing agreement with the given ID, or if the agreement is not in the diff --git a/bin/dipper-service/src/registry/agreement_stub.rs b/bin/dipper-service/src/registry/agreement_stub.rs index 0b34dcaf..43ddcf14 100644 --- a/bin/dipper-service/src/registry/agreement_stub.rs +++ b/bin/dipper-service/src/registry/agreement_stub.rs @@ -129,6 +129,14 @@ pub trait StubAgreementRegistry: Send + Sync { unimplemented!("update_offer_tx_hash") } + async fn update_terms_version_hash( + &self, + _id: &IndexingAgreementId, + _hash: &[u8; 32], + ) -> Result<()> { + unimplemented!("update_terms_version_hash") + } + async fn mark_indexing_agreement_as_canceled_by_requester( &self, _id: &IndexingAgreementId, @@ -416,6 +424,14 @@ impl AgreementRegistry for T { StubAgreementRegistry::update_offer_tx_hash(self, id, tx_hash).await } + async fn update_terms_version_hash( + &self, + id: &IndexingAgreementId, + hash: &[u8; 32], + ) -> Result<()> { + StubAgreementRegistry::update_terms_version_hash(self, id, hash).await + } + async fn mark_indexing_agreement_as_canceled_by_requester( &self, id: &IndexingAgreementId, diff --git a/bin/dipper-service/src/worker/handlers/cancel_rejected_agreement_on_chain.rs b/bin/dipper-service/src/worker/handlers/cancel_rejected_agreement_on_chain.rs index a5cfc33d..520459a6 100644 --- a/bin/dipper-service/src/worker/handlers/cancel_rejected_agreement_on_chain.rs +++ b/bin/dipper-service/src/worker/handlers/cancel_rejected_agreement_on_chain.rs @@ -74,43 +74,49 @@ where ); // Send the cancellation transaction (mode-aware dispatch). - let on_chain_cancel_tx: Option = - match cancel_agreement_on_chain(&ctx.chain_client, &agreement, &ctx.agreement_conf).await { - Ok(Some(tx_hash)) => { - tracing::info!( - agreement_id = %agreement_id, - tx_hash = %tx_hash, - "Successfully submitted on-chain cancellation" - ); - Some(tx_hash.to_string()) - } - Ok(None) => { - tracing::info!( - agreement_id = %agreement_id, - "Rejected agreement already canceled on-chain; reconciling local state" - ); - None - } - Err(err @ ChainClientError::MissingTermsVersionHash { .. }) => { - // Permanent: the hash never appears, so retrying can't help. Fail - // terminally and leave the live agreement for operator action. - tracing::error!( - agreement_id = %agreement_id, - error = %err, - "Cannot cancel rejected agreement: missing terms_version_hash" - ); - return Err(JobError::Fatal(err.into())); - } - Err(err) => { - tracing::warn!( - agreement_id = %agreement_id, - error = %err, - "Failed to cancel agreement on-chain, will retry" - ); - // Retry with backoff - on-chain transactions can fail due to gas issues, nonce, etc. - return Err(JobError::Retryable(err.into(), Duration::from_secs(30))); - } - }; + let on_chain_cancel_tx: Option = match cancel_agreement_on_chain( + &ctx.chain_client, + &ctx.registry, + &agreement, + &ctx.agreement_conf, + ) + .await + { + Ok(Some(tx_hash)) => { + tracing::info!( + agreement_id = %agreement_id, + tx_hash = %tx_hash, + "Successfully submitted on-chain cancellation" + ); + Some(tx_hash.to_string()) + } + Ok(None) => { + tracing::info!( + agreement_id = %agreement_id, + "Rejected agreement already canceled on-chain; reconciling local state" + ); + None + } + Err(err @ ChainClientError::MissingTermsVersionHash { .. }) => { + // Permanent: the hash never appears, so retrying can't help. Fail + // terminally and leave the live agreement for operator action. + tracing::error!( + agreement_id = %agreement_id, + error = %err, + "Cannot cancel rejected agreement: missing terms_version_hash" + ); + return Err(JobError::Fatal(err.into())); + } + Err(err) => { + tracing::warn!( + agreement_id = %agreement_id, + error = %err, + "Failed to cancel agreement on-chain, will retry" + ); + // Retry with backoff - on-chain transactions can fail due to gas issues, nonce, etc. + return Err(JobError::Retryable(err.into(), Duration::from_secs(30))); + } + }; // When the row was actually flipped to terminal, record the cancel audit so // the chain_listener's `terminated` sweep announces it durably. The accept @@ -361,6 +367,14 @@ mod tests { Ok(()) } + async fn update_terms_version_hash( + &self, + _id: &IndexingAgreementId, + _hash: &[u8; 32], + ) -> crate::registry::Result<()> { + Ok(()) + } + async fn mark_indexing_agreement_as_canceled_by_requester( &self, id: &IndexingAgreementId, @@ -495,6 +509,15 @@ mod tests { ) -> Result { Ok(false) } + + async fn fetch_agreement_version_hash( + &self, + _agreement_id: &[u8; 16], + ) -> Result, ChainClientError> { + // These tests always construct agreements with a stored hash, so + // recovery is never exercised. + unimplemented!("not exercised by cancel_rejected_agreement_on_chain tests") + } } fn test_agreement_conf() -> Arc { diff --git a/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs b/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs index bcf8f25c..753a72fe 100644 --- a/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs +++ b/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs @@ -645,6 +645,7 @@ where if needs_on_chain_cancel { match crate::cancel_dispatch::cancel_agreement_on_chain( &ctx.chain_client, + &ctx.registry, old_agreement, &ctx.agreement_conf, ) @@ -780,8 +781,13 @@ where } /// Compute the EIP-712 terms hash persisted for the protocol-managed cancel -/// path, reusing the proposal signer's RCA-to-sol conversion and signing-hash so -/// the value matches the hash dipper signs over. +/// path, reusing the proposal signer's RCA-to-sol conversion and signing-hash. +/// In protocol-managed mode dipper signs nothing — this value is only the +/// cancel identifier the RecurringCollector stored at offer time and checks +/// against on cancel (see `cancel_dispatch::cancel_agreement_on_chain`, +/// which recovers it from chain if this column is empty, and confirms +/// on-chain afterward rather than trusting a mismatch to fail loudly on its +/// own). fn compute_terms_version_hash( nonce_uuid: uuid::Uuid, terms: &IndexingAgreementTerms, @@ -1071,6 +1077,15 @@ mod lifecycle_event_tests { // Cancel confirmed: agreement is no longer active on-chain. Ok(false) } + + async fn fetch_agreement_version_hash( + &self, + _agreement_id: &[u8; 16], + ) -> std::result::Result, ChainClientError> { + // These tests always construct agreements with a stored hash, so + // recovery is never exercised. + unimplemented!("not exercised by reassess handler") + } } // ---- Mock: registry (all five traits) ----------------------------------- @@ -1232,6 +1247,13 @@ mod lifecycle_event_tests { ) -> RegistryResult<()> { unimplemented!() } + async fn update_terms_version_hash( + &self, + _id: &IndexingAgreementId, + _hash: &[u8; 32], + ) -> RegistryResult<()> { + unimplemented!("not exercised by reassess handler tests") + } // Cancel path: pre-mark the local row terminal. async fn mark_indexing_agreement_as_canceled_by_requester( &self, @@ -2294,6 +2316,15 @@ mod deadline_clock_tests { ) -> Result { Ok(false) } + + async fn fetch_agreement_version_hash( + &self, + _agreement_id: &[u8; 16], + ) -> Result, ChainClientError> { + // These tests always construct agreements with a stored hash, so + // recovery is never exercised. + unimplemented!("not exercised by reassess handler") + } } /// `fail: true` makes the state lookup itself error, exercising the diff --git a/bin/dipper-service/src/worker/handlers/send_indexing_agreement_proposal.rs b/bin/dipper-service/src/worker/handlers/send_indexing_agreement_proposal.rs index 4bb06e56..4cce3cfd 100644 --- a/bin/dipper-service/src/worker/handlers/send_indexing_agreement_proposal.rs +++ b/bin/dipper-service/src/worker/handlers/send_indexing_agreement_proposal.rs @@ -509,6 +509,14 @@ mod tests { Ok(()) } + async fn update_terms_version_hash( + &self, + _id: &IndexingAgreementId, + _hash: &[u8; 32], + ) -> crate::registry::Result<()> { + Ok(()) + } + async fn mark_indexing_agreement_as_canceled_by_requester( &self, _id: &IndexingAgreementId, diff --git a/bin/dipper-service/src/worker/handlers/submit_offer.rs b/bin/dipper-service/src/worker/handlers/submit_offer.rs index 21511f8c..1327ef75 100644 --- a/bin/dipper-service/src/worker/handlers/submit_offer.rs +++ b/bin/dipper-service/src/worker/handlers/submit_offer.rs @@ -271,6 +271,12 @@ mod tests { ) -> Result { unimplemented!() } + async fn fetch_agreement_version_hash( + &self, + _agreement_id: &[u8; 16], + ) -> Result, ChainClientError> { + unimplemented!() + } async fn latest_block_timestamp(&self) -> Result { unimplemented!() } diff --git a/dipper-pgregistry/src/postgres.rs b/dipper-pgregistry/src/postgres.rs index 8f2254fc..36a3af47 100644 --- a/dipper-pgregistry/src/postgres.rs +++ b/dipper-pgregistry/src/postgres.rs @@ -895,6 +895,34 @@ impl PgRegistry { Ok(()) } + /// Backfill a `terms_version_hash` recovered from on-chain state for an + /// agreement whose local copy is missing — e.g. a row from before the + /// `terms_version_hash` column existed. No status guard: unlike + /// `update_offer_tx_hash` (which risks clobbering a live tx hash for an + /// agreement that has since moved on), a missing hash never gets a + /// legitimate DB-side update to race with, and cancellation from any + /// non-terminal status still needs the recovered value. + pub async fn update_terms_version_hash( + &self, + agreement_id: &IndexingAgreementId, + hash: &[u8; 32], + ) -> Result<(), Error> { + sqlx::query( + r#" + UPDATE dipper_reg_indexing_agreements + SET + terms_version_hash = $1, + updated_at = timezone('UTC', now()) + WHERE id = $2 AND terms_version_hash IS NULL + "#, + ) + .bind(&hash[..]) + .bind(agreement_id) + .execute(&self.pool) + .await?; + Ok(()) + } + pub async fn mark_indexing_agreement_as_canceled_by_requester( &self, agreement_id: &IndexingAgreementId,