diff --git a/bin/dipper-service/src/admin_rpc_server/handlers/indexing_requests.rs b/bin/dipper-service/src/admin_rpc_server/handlers/indexing_requests.rs index b846dd8e..af36ee22 100644 --- a/bin/dipper-service/src/admin_rpc_server/handlers/indexing_requests.rs +++ b/bin/dipper-service/src/admin_rpc_server/handlers/indexing_requests.rs @@ -364,6 +364,15 @@ mod tests { Ok(JobId::default()) } + async fn cancel_rejected_agreement_on_chain( + &self, + _agreement_id: IndexingAgreementId, + _accept: Option, + _priority: crate::worker::queue::JobPriority, + ) -> anyhow::Result { + unimplemented!() + } + async fn submit_offer( &self, _agreement_id: IndexingAgreementId, diff --git a/bin/dipper-service/src/cancel_dispatch.rs b/bin/dipper-service/src/cancel_dispatch.rs index 1bb03b3c..a32029bd 100644 --- a/bin/dipper-service/src/cancel_dispatch.rs +++ b/bin/dipper-service/src/cancel_dispatch.rs @@ -85,10 +85,10 @@ pub enum CancelReason { } impl CancelReason { - fn ended_status(self) -> &'static str { + fn ended_status(self) -> IndexingAgreementStatus { match self { - Self::NotWanted => "CANCELED_BY_REQUESTER", - Self::Abandoned => "ABANDONED_BY_INDEXER", + Self::NotWanted => IndexingAgreementStatus::CanceledByRequester, + Self::Abandoned => IndexingAgreementStatus::AbandonedByIndexer, } } } @@ -162,6 +162,26 @@ pub async fn mark_cancelling( } } +/// Mark an agreement dipper's cancel ended in the status its `reason` ends it in. +async fn mark_ended( + registry: &R, + agreement: &IndexingAgreement, + reason: CancelReason, +) -> RegistryResult<()> { + match reason { + CancelReason::NotWanted => { + registry + .mark_indexing_agreement_as_canceled_by_requester(&agreement.id) + .await + } + CancelReason::Abandoned => { + registry + .mark_indexing_agreement_as_abandoned_by_indexer(&agreement.id) + .await + } + } +} + /// The second half of [`start_cancel`]: cancel a marked agreement if the chain shows it live. pub async fn send_marked_cancel( registry: &R, @@ -219,10 +239,10 @@ async fn finish_marked_cancel( } } -/// 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. One the chain listener already marked -/// ended counts as ended. +/// Mark an agreement the chain shows dipper ended as ended, as its `reason` says, 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. One the chain listener +/// already marked ended counts as ended. pub async fn confirm_cancelled( registry: &R, agreement: &IndexingAgreement, @@ -234,10 +254,7 @@ pub async fn confirm_cancelled( if tx_hash.is_some() { record_cancel(registry, agreement, tx_hash, config).await; } - match registry - .mark_indexing_agreement_as_canceled_by_requester(&agreement.id) - .await - { + match mark_ended(registry, agreement, reason).await { Ok(()) => {} Err(crate::registry::Error::NoRecordsUpdated) => { tracing::debug!( @@ -258,8 +275,8 @@ pub async fn confirm_cancelled( tracing::info!( agreement_id = %agreement.id, indexing_request_id = %agreement.indexing_request_id, - old_status = "CANCELLING", - new_status = reason.ended_status(), + old_status = %IndexingAgreementStatus::Cancelling, + new_status = %reason.ended_status(), reason = "cancel_confirmed_on_chain", "agreement state transition" ); @@ -288,15 +305,25 @@ async fn record_cancel( } } -/// Move an agreement dipper had already rejected or cancelled back into `Cancelling` when the -/// chain shows it live after all, so the cancel retry ends it. The chain is read first, so a -/// subgraph report from before dipper's cancel landed reopens nothing; an unreadable chain -/// reopens it anyway, as the retry reads again before sending. True if it was reopened. +/// What [`reopen_if_live`] did. +#[derive(Debug)] +pub enum Reopen { + /// Moved back into `Cancelling`. + Reopened, + /// Left as it was: the chain shows it ended, or it is no longer rejected or cancelled. + Left, + /// Left as it was, as the chain couldn't be read, so it needs reading again. + Unread(ChainClientError), +} + +/// Move an agreement dipper rejected or cancelled back into `Cancelling` if the chain shows it +/// live, for the cancel retry to end. An unreadable chain reopens only a rejected one, whose +/// indexer may be paid; a cancelled one likely shows accepted only as the subgraph lags its end. pub async fn reopen_if_live( registry: &R, chain_client: &T, agreement: &IndexingAgreement, -) -> RegistryResult +) -> RegistryResult where R: AgreementRegistry + Sync, T: ChainClient, @@ -306,12 +333,15 @@ where .await { Ok(AgreementOnChain::Live) => true, - Ok(_) => return Ok(false), + Ok(_) => return Ok(Reopen::Left), + Err(err) if agreement.status != IndexingAgreementStatus::Rejected => { + return Ok(Reopen::Unread(err)); + } Err(err) => { tracing::warn!( agreement_id = %agreement.id, error = %err, - "Failed to read an ended agreement reported live; the cancel retry checks it" + "Failed to read a rejected agreement reported live; the cancel retry checks it" ); false } @@ -321,7 +351,7 @@ where .await { Ok(()) => {} - Err(crate::registry::Error::NoRecordsUpdated) => return Ok(false), + Err(crate::registry::Error::NoRecordsUpdated) => return Ok(Reopen::Left), Err(err) => return Err(err), } tracing::warn!( @@ -329,11 +359,11 @@ where indexer_id = %agreement.indexer.id, indexing_request_id = %agreement.indexing_request_id, old_status = %agreement.status, - new_status = "CANCELLING", + new_status = %IndexingAgreementStatus::Cancelling, reason = "live_on_chain_after_end", "agreement state transition" ); - Ok(true) + Ok(Reopen::Reopened) } /// Log a cancel that mined but could not be read back, naming its transaction so it isn't lost; @@ -407,7 +437,7 @@ pub(crate) mod tests { use time::OffsetDateTime; use url::Url; - use super::{LiveCancel, SCOPE_BOTH, cancel_agreement_on_chain}; + use super::{CancelReason, LiveCancel, SCOPE_BOTH, cancel_agreement_on_chain}; use crate::{ chain_client::{AgreementOnChain, ChainClient, ChainClientError}, config::IndexingAgreementConfig, @@ -547,6 +577,18 @@ pub(crate) mod tests { } } + #[test] + fn each_reason_ends_an_agreement_in_its_own_status() { + assert_eq!( + CancelReason::NotWanted.ended_status(), + IndexingAgreementStatus::CanceledByRequester + ); + assert_eq!( + CancelReason::Abandoned.ended_status(), + IndexingAgreementStatus::AbandonedByIndexer + ); + } + #[tokio::test] async fn manager_cancel_uses_both_scopes_for_accepted() { // comp-1: a manager cancel passes BOTH scope bits so the collector @@ -724,11 +766,8 @@ pub(crate) mod tests { #[tokio::test] async fn a_reopen_clears_the_end_on_record_only_when_the_chain_shows_it_live() { - // An unread chain reopens it all the same, but it may have ended, so its record stays. - let ag = agreement( - IndexingAgreementStatus::CanceledByRequester, - Some(vec![7u8; 32]), - ); + // An unread chain reopens a rejected one all the same, but it may have ended, so its + // record stays. let live = RecordingChainClient { still_active_after_cancel: true, ..Default::default() @@ -739,14 +778,37 @@ pub(crate) mod tests { }; for (client, seen_live) in [(live, true), (unread, false)] { + let ag = agreement(IndexingAgreementStatus::Rejected, Some(vec![7u8; 32])); let registry = ReopenRegistry::default(); - let reopened = super::reopen_if_live(®istry, &client, &ag) + let reopen = super::reopen_if_live(®istry, &client, &ag) .await .expect("reopen"); - assert!(reopened); + assert!(matches!(reopen, super::Reopen::Reopened)); assert_eq!(*registry.seen_live.lock().unwrap(), vec![seen_live]); } } + + #[tokio::test] + async fn an_unread_chain_leaves_a_cancelled_agreement_to_be_read_again() { + // The subgraph lagging behind dipper's own cancel is the likelier reason it shows + // accepted, and reopening an ended one counts it live until the cancel retry reads it. + let ag = agreement( + IndexingAgreementStatus::CanceledByRequester, + Some(vec![7u8; 32]), + ); + let unread = RecordingChainClient { + read_back_fails: true, + ..Default::default() + }; + let registry = ReopenRegistry::default(); + + let reopen = super::reopen_if_live(®istry, &unread, &ag) + .await + .expect("reopen"); + + assert!(matches!(reopen, super::Reopen::Unread(_))); + assert!(registry.seen_live.lock().unwrap().is_empty()); + } } diff --git a/bin/dipper-service/src/chain_client/client.rs b/bin/dipper-service/src/chain_client/client.rs index 842d5021..beaf922f 100644 --- a/bin/dipper-service/src/chain_client/client.rs +++ b/bin/dipper-service/src/chain_client/client.rs @@ -28,7 +28,7 @@ use tokio::sync::Mutex; use super::{ abi::{IRecurringAgreementManager, IRecurringCollector}, gas::{GasEstimator, calculate_max_fee, exceeds_max_gas_price, get_gas_prices}, - rpc_provider::{BEHIND_A_SEEN_BLOCK, RpcProviderPool}, + rpc_provider::{BEHIND_A_SEEN_BLOCK, HttpProvider, RpcProviderPool}, }; use crate::{ chain_client::{ @@ -77,6 +77,11 @@ const BLOCKS_PER_SECOND: u64 = 4; /// the newest block dipper has seen raises an ERROR; nearer is lag that clears by itself. const ALERT_GAP_BLOCKS: u64 = 14_400; +/// A minute of blocks at the chain's fastest: how far apart endpoints may be and still agree. +/// Ones that are right differ by a few blocks, and the chain's progress is allowed for apart from +/// this, so it only bounds how far ahead a faulty endpoint can move a confirmed newest block seen. +const AGREE_WITHIN_BLOCKS: u64 = 60 * BLOCKS_PER_SECOND; + /// How a read refused by an endpoint whose latest block is too far ahead to be real says so. const FAR_AHEAD_OF_A_SEEN_BLOCK: &str = "too far ahead of the newest block seen"; @@ -87,12 +92,12 @@ const CROSS_CHECK_INTERVAL: Duration = Duration::from_secs(60); /// An answer takes well under a second, so this only bounds how long a hung endpoint holds a read. const CROSS_CHECK_DEADLINE: Duration = Duration::from_secs(3); -/// The newest block dipper has seen, from reads, receipts and latest-block lookups, and when it -/// last moved. It never goes backwards, so a lagging endpoint can't show state from before it. -/// Blocks too far ahead of it are refused only once it is confirmed, by the receipt for one of -/// dipper's transactions or by 2 endpoints agreeing, so an endpoint stuck far behind, or far -/// ahead, that answers first after a restart can't shut out the endpoints that are right. -/// Once confirmed, 1 endpoint alone can't move it far ahead either, for the same reason. +/// The newest block dipper has seen and when it last moved; it never goes back, so a lagging +/// endpoint can't show state from before it. Nothing ahead is refused until a receipt or 2 agreeing +/// endpoints confirm it, which replaces a block far ahead set by a faulty endpoint that answered +/// first after a restart. Until then any endpoint's block is taken: with only 2 that disagree, +/// which is right can't be told, so 1 far ahead has the other refused as behind until a receipt +/// or the chain catches up. Refusing it instead would let 1 stuck far behind serve stale reads. #[derive(Debug)] struct SeenBlock { number: u64, @@ -145,12 +150,12 @@ impl SeenBlock { (self.number, highest) } - /// The highest block 1 endpoint alone may move a confirmed newest block to: an hour of + /// The highest block 1 endpoint alone may move a confirmed newest block to: a minute of /// blocks past it, plus what the chain can have added since. fn lone_limit(&self, now: Instant) -> u64 { let since = now.saturating_duration_since(self.moved_at).as_secs(); self.number - .saturating_add(ALERT_GAP_BLOCKS) + .saturating_add(AGREE_WITHIN_BLOCKS) .saturating_add(since.saturating_mul(BLOCKS_PER_SECOND)) } @@ -179,7 +184,7 @@ impl SeenBlock { } /// Take `block` as one the chain is known to have reached. An unconfirmed newest block more - /// than an hour of blocks past it came from a faulty endpoint, so it is replaced. + /// than a minute of blocks past it came from a faulty endpoint, so it is replaced. fn confirm(&mut self, block: u64, now: Instant) { self.jump_reported = false; if self.confirmed { @@ -188,7 +193,7 @@ impl SeenBlock { } return; } - if block > self.number || self.number > block.saturating_add(ALERT_GAP_BLOCKS) { + if block > self.number || self.number > block.saturating_add(AGREE_WITHIN_BLOCKS) { self.number = block; self.moved_at = now; } @@ -210,13 +215,13 @@ impl SeenBlock { } /// The newest of the latest blocks that the most endpoints agree on, at least 2 of them within -/// an hour of blocks of each other, or `None` when no 2 agree. +/// a minute of blocks of each other, or `None` when no 2 agree. fn agreed_head(mut heads: Vec) -> Option { heads.sort_unstable(); let mut agreed: Option<(usize, u64)> = None; let mut start = 0; for (end, &head) in heads.iter().enumerate() { - while head - heads[start] > ALERT_GAP_BLOCKS { + while head - heads[start] > AGREE_WITHIN_BLOCKS { start += 1; } let agreeing = end - start + 1; @@ -299,6 +304,61 @@ impl Refusals { } } +/// How long a read reuses an endpoint's latest block before asking it again: about 4 blocks. +/// A read at it is still checked against the newest block seen, and asking before every read +/// doubled the calls a run of reads makes. +const HEAD_REUSE: Duration = Duration::from_secs(1); + +/// Each endpoint's latest block and when it was asked for, by the endpoint's place in the list. +struct EndpointHeads(std::sync::Mutex>>); + +impl EndpointHeads { + fn new(endpoints: usize) -> Self { + Self(std::sync::Mutex::new(vec![None; endpoints])) + } + + fn slots(&self) -> std::sync::MutexGuard<'_, Vec>> { + self.0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + + /// `endpoint`'s latest block, if it was asked for under [`HEAD_REUSE`] before `now`. + fn fresh(&self, endpoint: usize, now: Instant) -> Option { + let (head, asked_at) = self.slots()[endpoint]?; + (now.saturating_duration_since(asked_at) < HEAD_REUSE).then_some(head) + } + + fn remember(&self, endpoint: usize, head: u64, asked_at: Instant) { + self.slots()[endpoint] = Some((head, asked_at)); + } + + fn forget(&self, endpoint: usize) { + self.slots()[endpoint] = None; + } + + /// `endpoint`'s latest block, reused if fresh and not behind `seen`, the newest block seen, + /// otherwise asked for and remembered. + async fn latest( + &self, + provider: &HttpProvider, + endpoint: usize, + seen: u64, + refusals: &Refusals, + ) -> Result { + let reusable = self + .fresh(endpoint, Instant::now()) + .filter(|&head| head >= seen); + if let Some(head) = reusable { + return Ok(head); + } + let asked_at = Instant::now(); + let head = refusals.other(provider.get_block_number().await)?; + self.remember(endpoint, head, asked_at); + Ok(head) + } +} + /// `AgreementDetails.state` flags from `IAgreementCollector.sol` (REGISTERED=1, /// ACCEPTED=2, NOTICE_GIVEN=4, SETTLED=8, BY_PROVIDER=32). `getAgreementDetails` keeps /// ACCEPTED set on a canceled agreement and ORs in NOTICE_GIVEN, so a cancel must clear it, @@ -496,6 +556,8 @@ struct AlloyChainClientInner { /// behind it, so a lagging endpoint can't undo what dipper saw, nor from one too far /// ahead of it to be real, so a faulty endpoint can't refuse every read after it. seen_block: std::sync::Mutex, + /// Each endpoint's latest block, so a run of reads asks for it once a second, not every read. + endpoint_heads: EndpointHeads, } impl AlloyChainClient { @@ -532,6 +594,7 @@ impl AlloyChainClient { ); let submit_deadline = derive_submit_deadline(&rpc_pool); + let endpoint_heads = EndpointHeads::new(rpc_pool.endpoint_count()); Ok(Self { inner: Arc::new(AlloyChainClientInner { @@ -547,6 +610,7 @@ impl AlloyChainClient { submit_lock: Mutex::new(()), submit_deadline, seen_block: std::sync::Mutex::new(SeenBlock::new()), + endpoint_heads, }), }) } @@ -877,9 +941,9 @@ impl AlloyChainClient { .state) } - /// Run a read-only contract call at the endpoint's latest block, refusing an endpoint - /// whose latest block is older than one dipper has already seen, or too far ahead of it - /// to be real; the pool moves on to the next endpoint instead. + /// Run a read-only contract call at the endpoint's latest block, as of [`HEAD_REUSE`] ago at + /// most, refusing an endpoint whose latest block is older than one dipper has already seen, + /// or too far ahead of it to be real; the pool moves on to the next endpoint instead. async fn view_at_seen_block( &self, to: Address, @@ -890,19 +954,23 @@ impl AlloyChainClient { self.cross_check_seen_block().await; let (seen, highest) = self.block_bounds(); let refusals = Refusals::new(); + let heads = &self.inner.endpoint_heads; let read = self .inner .rpc_pool - .execute(operation, |provider| { + .execute_with_endpoint(operation, |provider, endpoint| { let calldata = calldata.clone(); let refusals = &refusals; async move { - let head = refusals.other(provider.get_block_number().await)?; - refusals.behind(head, seen)?; - refusals.far_ahead(head, seen, highest)?; - let tx = TransactionRequest::default().to(to).input(calldata.into()); - let output = - refusals.other(provider.call(tx).block(BlockId::number(head)).await)?; + let head = heads.latest(&provider, endpoint, seen, refusals).await?; + let read = async { + refusals.behind(head, seen)?; + refusals.far_ahead(head, seen, highest)?; + let tx = TransactionRequest::default().to(to).input(calldata.into()); + refusals.other(provider.call(tx).block(BlockId::number(head)).await) + }; + // An endpoint refused or failing may have moved on by the next attempt. + let output = read.await.inspect_err(|_| heads.forget(endpoint))?; Ok((head, output)) } }) @@ -933,18 +1001,20 @@ impl AlloyChainClient { if pool.endpoint_count() < 2 || !self.seen_block().cross_check_due(Instant::now()) { return; } - match agreed_head(pool.latest_blocks(CROSS_CHECK_DEADLINE).await) { + let heads = pool.latest_blocks(CROSS_CHECK_DEADLINE).await; + match agreed_head(heads.clone()) { Some(head) => self.seen_block().confirm(head, Instant::now()), None => tracing::warn!( - "No 2 RPC endpoints agree on the chain's latest block; reads aren't checked \ - against blocks too far ahead, nor the newest block seen moved far ahead, until \ - they do" + latest_blocks = ?heads, + "No 2 RPC endpoints agree on the chain's latest block; until they do, or a \ + receipt confirms it, any endpoint's block is taken as the newest seen and \ + endpoints behind it are refused" ), } } /// Remember a block dipper has seen, so later reads are never older. One too far ahead to - /// be real is ignored, so it can't refuse every read after it, and one over an hour ahead + /// be real is ignored, so it can't refuse every read after it, and one over a minute ahead /// waits for 2 endpoints to agree on it. fn note_block(&self, block: u64) { let only_endpoint = self.inner.rpc_pool.endpoint_count() < 2; @@ -960,7 +1030,7 @@ impl AlloyChainClient { Noted::Unagreed => tracing::warn!( block, seen_block = seen, - "Not moving to a block over an hour ahead of the newest block dipper has seen \ + "Not moving to a block over a minute ahead of the newest block dipper has seen \ until 2 RPC endpoints agree on it" ), Noted::TooFarAhead => tracing::warn!( @@ -1381,7 +1451,10 @@ impl ManagerEscrowReader for AlloyChainClient { #[cfg(test)] mod tests { use url::Url; - use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate, matchers::method}; + use wiremock::{ + Mock, MockServer, Request, Respond, ResponseTemplate, + matchers::{body_partial_json, method}, + }; use super::*; @@ -2503,6 +2576,88 @@ mod tests { assert!(live); } + /// How many of `requests` called `method`. + fn calls_to(requests: &[wiremock::Request], method: &str) -> usize { + requests + .iter() + .filter(|r| { + let body: serde_json::Value = + serde_json::from_slice(&r.body).expect("JSON-RPC request body"); + body["method"] == method + }) + .count() + } + + #[tokio::test] + async fn reads_in_quick_succession_ask_for_the_latest_block_once() { + let endpoint = server_at_block(100, STATE_REGISTERED).await; + let client = client_over(vec![endpoint.uri().parse().expect("provider URL")]); + + for _ in 0..3 { + client.agreement_on_chain(&[0xab; 16]).await.expect("read"); + } + + let requests = endpoint.received_requests().await.unwrap_or_default(); + assert_eq!(calls_to(&requests, "eth_blockNumber"), 1); + assert_eq!(calls_to(&requests, "eth_call"), 3); + } + + #[test] + fn reuses_an_endpoints_latest_block_for_a_second_unless_forgotten() { + let asked_at = Instant::now(); + let heads = EndpointHeads::new(2); + heads.remember(0, 100, asked_at); + + let soon = asked_at + Duration::from_millis(900); + assert_eq!(heads.fresh(0, soon), Some(100)); + assert_eq!(heads.fresh(0, asked_at + HEAD_REUSE), None, "asked again"); + assert_eq!(heads.fresh(1, soon), None, "each endpoint gives its own"); + + heads.forget(0); + assert_eq!(heads.fresh(0, soon), None, "asked again after a failure"); + } + + #[tokio::test] + async fn asks_for_the_latest_block_again_once_a_newer_one_is_seen() { + // As when dipper's own transaction has just mined: the read must not come from before it. + let endpoint = server_catching_up(100, 1, STATE_REGISTERED).await; + let client = client_over(vec![endpoint.uri().parse().expect("provider URL")]); + client.agreement_on_chain(&[0xab; 16]).await.expect("read"); + trust_block(&client, 101); + + client.agreement_on_chain(&[0xab; 16]).await.expect("read"); + + let requests = endpoint.received_requests().await.unwrap_or_default(); + assert_eq!(calls_to(&requests, "eth_blockNumber"), 2); + } + + #[tokio::test] + async fn asks_for_the_latest_block_again_after_a_failed_read() { + // The endpoint may have answered from a node that lacks the block it gave. + let endpoint = server_at_block(100, STATE_REGISTERED).await; + Mock::given(method("POST")) + .and(body_partial_json( + serde_json::json!({ "method": "eth_call" }), + )) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "jsonrpc": "2.0", + "id": 0, + "error": { "code": -32000, "message": "missing trie node" }, + }))) + .up_to_n_times(1) + .with_priority(1) + .mount(&endpoint) + .await; + let client = client_over(vec![endpoint.uri().parse().expect("provider URL")]); + let failed = client.agreement_on_chain(&[0xab; 16]).await; + assert!(failed.is_err(), "got {failed:?}"); + + client.agreement_on_chain(&[0xab; 16]).await.expect("read"); + + let requests = endpoint.received_requests().await.unwrap_or_default(); + assert_eq!(calls_to(&requests, "eth_blockNumber"), 2); + } + #[tokio::test] async fn never_reads_an_agreement_from_an_endpoint_behind_a_block_already_seen() { // The lagging endpoint still shows the offer it hasn't seen accepted and cancelled. @@ -2679,7 +2834,7 @@ mod tests { let mut seen = SeenBlock::new(); let real = 5_000_000; seen.confirm(real, now); - let jump = real + ALERT_GAP_BLOCKS + 1_000; + let jump = real + AGREE_WITHIN_BLOCKS + 1_000; assert_eq!(seen.advance(real + 10, now, false), Noted::Taken); assert_eq!(seen.advance(jump, now, false), Noted::Unagreed); @@ -2691,7 +2846,7 @@ mod tests { assert_eq!(seen.bounds(now).0, jump, "taken once 2 agree"); assert!(!seen.cross_check_due(now + CROSS_CHECK_INTERVAL)); assert_eq!( - seen.advance(jump + ALERT_GAP_BLOCKS + 1, now, true), + seen.advance(jump + AGREE_WITHIN_BLOCKS + 1, now, true), Noted::Taken, "the only endpoint has no other to agree with" ); @@ -2724,6 +2879,115 @@ mod tests { assert_eq!(agreed_head(vec![real]), None, "1 can't agree with itself"); } + #[test] + fn a_faulty_endpoint_well_ahead_neither_agrees_nor_moves_a_confirmed_block() { + // Its block taken as seen would have the endpoints that are right refused as behind + // until the chain caught up with it: 10,000 blocks is about 42 minutes. + let now = Instant::now(); + let real = 5_000_000; + let faulty = real + 10_000; + assert_eq!(agreed_head(vec![real, real + 2, faulty]), Some(real + 2)); + + let mut seen = SeenBlock::new(); + seen.confirm(real, now); + assert_eq!(seen.advance(faulty, now, false), Noted::Unagreed); + assert_eq!(seen.bounds(now).0, real); + } + + #[test] + fn a_block_from_a_faulty_endpoint_well_ahead_is_replaced_once_confirmed() { + // After a restart the faulty endpoint may answer first, before anything is confirmed. + let now = Instant::now(); + let real = 5_000_000; + let mut seen = SeenBlock::new(); + assert_eq!(seen.advance(real + 10_000, now, false), Noted::Taken); + + seen.confirm(real, now); + + assert_eq!(seen.bounds(now).0, real); + } + + #[test] + fn takes_the_higher_block_from_2_endpoints_that_disagree() { + // Which is right can't be told, so 1 stuck far behind is refused, and so is a right one + // behind 1 faulty far ahead, until a receipt replaces its block. + let now = Instant::now(); + let real = 5_000_000; + let stuck = real - 10_000; + let faulty = real + 10_000; + assert_eq!(agreed_head(vec![stuck, real]), None); + assert_eq!(agreed_head(vec![real, faulty]), None); + + let mut seen = SeenBlock::new(); + assert_eq!(seen.advance(stuck, now, false), Noted::Taken); + assert_eq!(seen.advance(real, now, false), Noted::Taken); + assert_eq!(seen.bounds(now).0, real, "the stuck one is refused"); + assert_eq!(seen.advance(faulty, now, false), Noted::Taken); + assert_eq!(seen.bounds(now).0, faulty, "the right one is refused"); + + seen.confirm(real + 4, now); + assert_eq!(seen.bounds(now).0, real + 4); + } + + #[test] + fn takes_the_few_blocks_endpoints_that_are_right_differ_by() { + let now = Instant::now(); + let real = 5_000_000; + let minute_ahead = real + AGREE_WITHIN_BLOCKS; + assert_eq!(agreed_head(vec![real, real + 3]), Some(real + 3)); + assert_eq!(agreed_head(vec![real, minute_ahead]), Some(minute_ahead)); + + let mut seen = SeenBlock::new(); + assert_eq!(seen.advance(minute_ahead, now, false), Noted::Taken); + seen.confirm(real, now); + assert_eq!( + seen.bounds(now).0, + minute_ahead, + "a lead of a minute is kept" + ); + + let mut seen = SeenBlock::new(); + seen.confirm(real, now); + assert_eq!(seen.advance(real + 40, now, false), Noted::Taken); + assert_eq!(seen.advance(minute_ahead + 40, now, false), Noted::Taken); + + // 10 quiet minutes, in which the chain can add 2,400 blocks. + let mut quiet = SeenBlock { + number: real, + moved_at: now.checked_sub(Duration::from_secs(600)).expect("instant"), + confirmed: true, + cross_checked_at: None, + jump_reported: false, + }; + assert_eq!(quiet.advance(real + 2_400 + 40, now, false), Noted::Taken); + } + + #[tokio::test] + async fn a_faulty_endpoint_well_ahead_does_not_shut_out_the_ones_that_are_right() { + let real = 5_000_000; + let ended = STATE_REGISTERED | STATE_ACCEPTED | STATE_NOTICE_GIVEN; + let right = server_at_block(real, ended).await; + let also_right = server_at_block(real, ended).await; + let faulty = server_at_block(real + 10_000, STATE_REGISTERED).await; + let client = client_over(vec![ + right.uri().parse().expect("provider URL"), + also_right.uri().parse().expect("provider URL"), + faulty.uri().parse().expect("provider URL"), + ]); + // The faulty endpoint answered first after a restart. + client.note_block(real + 10_000); + + let live = client + .agreement_on_chain(&[0xab; 16]) + .await + .expect("read") + .is_live(); + + assert!(!live, "read from an endpoint that is right"); + assert!(client.seen_block().confirmed); + assert_eq!(client.seen_block().number, real); + } + #[tokio::test] async fn confirms_the_newest_block_from_endpoints_that_agree() { // The first endpoint is stuck far behind; the other 2 agree, so it is passed over diff --git a/bin/dipper-service/src/chain_client/rpc_provider.rs b/bin/dipper-service/src/chain_client/rpc_provider.rs index f05bd1f7..3db14d3e 100644 --- a/bin/dipper-service/src/chain_client/rpc_provider.rs +++ b/bin/dipper-service/src/chain_client/rpc_provider.rs @@ -223,6 +223,21 @@ impl RpcProviderPool { where F: Fn(HttpProvider) -> Fut, Fut: Future>, + { + self.execute_with_endpoint(operation, |provider, _| f(provider)) + .await + } + + /// As [`execute`](Self::execute), also telling `f` which endpoint it is asking, by its place + /// in the list, so a caller can keep something for each endpoint. + pub async fn execute_with_endpoint( + &self, + operation: &str, + f: F, + ) -> Result + where + F: Fn(HttpProvider, usize) -> Fut, + Fut: Future>, { self.execute_with_retries(operation, self.max_retries, f) .await @@ -235,7 +250,7 @@ impl RpcProviderPool { f: F, ) -> Result where - F: Fn(HttpProvider) -> Fut, + F: Fn(HttpProvider, usize) -> Fut, Fut: Future>, { // What each endpoint said, in the order they were tried. `sign_and_send` matches @@ -254,11 +269,12 @@ impl RpcProviderPool { let start = self.current_index.load(Ordering::Relaxed); loop { - let current_url = self.url_at(start + providers_tried).clone(); + let position = (start + providers_tried) % self.providers.len(); + let current_url = self.url_at(position).clone(); let endpoint = endpoint_name(¤t_url); let endpoint_error = match self - .ask_endpoint(operation, ¤t_url, max_retries, &f) + .ask_endpoint(operation, position, ¤t_url, max_retries, &f) .await { Ok(result) => return Ok(result), @@ -315,17 +331,18 @@ impl RpcProviderPool { } } - /// Run a call against one endpoint, retrying a fault worth another go, and return what it + /// Run a call against 1 endpoint, retrying a fault worth another go, and return what it /// answered or why it last failed. async fn ask_endpoint( &self, operation: &str, + position: usize, url: &Url, max_retries: u32, f: &F, ) -> Result where - F: Fn(HttpProvider) -> Fut, + F: Fn(HttpProvider, usize) -> Fut, Fut: Future>, { let endpoint = endpoint_name(url); @@ -335,7 +352,7 @@ impl RpcProviderPool { let mut endpoint_error: Option = None; let mut lag_retries = 0; for attempt in 0..=max_retries { - match f(provider.clone()).await { + match f(provider.clone(), position).await { Ok(result) => return Ok(result), Err(e) if Self::is_behind(&e) @@ -537,6 +554,34 @@ mod tests { } } + /// A caller keeping something for each endpoint must be told the one it is really asking. + #[tokio::test] + async fn a_call_is_told_which_endpoint_it_is_asking() { + let servers = [server_answering_block(7).await, server_refusing().await]; + let pool = RpcProviderPool::new( + servers + .iter() + .map(|s| s.uri().parse().expect("server URL")) + .collect(), + Duration::from_secs(5), + 0, + ) + .expect("pool"); + pool.rotate(); + + let asked = std::sync::Mutex::new(Vec::new()); + let block = pool + .execute_with_endpoint("probe", |provider, endpoint| { + asked.lock().expect("asked").push(endpoint); + async move { provider.get_block_number().await } + }) + .await + .expect("the first endpoint answers"); + + assert_eq!(block, 7); + assert_eq!(*asked.lock().expect("asked"), vec![1, 0], "wrapped round"); + } + /// Hosted RPC endpoints carry their API key in the path, and this failure text reaches the /// logs and every error built from it, so it has to say which endpoint refused without /// repeating the key that gets it in. diff --git a/bin/dipper-service/src/main.rs b/bin/dipper-service/src/main.rs index 17810755..2b4fce79 100644 --- a/bin/dipper-service/src/main.rs +++ b/bin/dipper-service/src/main.rs @@ -539,6 +539,7 @@ pub async fn main() -> anyhow::Result<()> { chain_listener_notify: chain_listener_notify.clone(), subgraph_indexing_agreements_events_emitter: subgraph_indexing_agreements_events_emitter.clone(), + worker_queue: worker_handle.queue().clone(), }; let (handle, service) = network::service::chain_listener::new(ctx); Some((handle, service)) diff --git a/bin/dipper-service/src/network/service/cancel_retry.rs b/bin/dipper-service/src/network/service/cancel_retry.rs index 3f3fa6b9..11a786f2 100644 --- a/bin/dipper-service/src/network/service/cancel_retry.rs +++ b/bin/dipper-service/src/network/service/cancel_retry.rs @@ -16,7 +16,7 @@ use crate::{ config::IndexingAgreementConfig, registry::{ AgreementRegistry, CancelKind, CancellingAgreement, IndexingAgreement, - IndexingRequestRegistry, + IndexingAgreementStatus, IndexingRequestRegistry, }, worker::service::WorkerQueue, }; @@ -288,8 +288,8 @@ async fn confirm_if_over( } /// Mark an agreement the indexer ended `CanceledByIndexer` when the chain listener hasn't in -/// time, so it doesn't stay cancelling for good. The indexer is recorded as ending it first, -/// so its announcement names them; the time recorded is when dipper noticed. +/// time, so it doesn't stay cancelling for good. The indexer is recorded as ending it first, at +/// the time dipper noticed: an estimate the chain's end, once the listener reads it, replaces. async fn record_end_by_indexer( registry: &R, agreement: &IndexingAgreement, @@ -329,8 +329,8 @@ fn log_end_by_indexer(agreement: &IndexingAgreement, changed: bool) { tracing::info!( agreement_id = %agreement.id, indexing_request_id = %agreement.indexing_request_id, - old_status = "CANCELLING", - new_status = "CANCELED_BY_INDEXER", + old_status = %IndexingAgreementStatus::Cancelling, + new_status = %IndexingAgreementStatus::CanceledByIndexer, reason = "indexer_ended_it_before_dipper_cancelled", "agreement state transition" ); @@ -366,6 +366,10 @@ async fn note_check( log_failed_cancel(row, attempts, failed_attempts, err); } } + Err(crate::registry::Error::NoRecordsUpdated) => tracing::debug!( + agreement_id = %agreement.id, + "Agreement no longer cancelling when its check was recorded, as the chain listener can end it first" + ), Err(err) => tracing::warn!( agreement_id = %agreement.id, error = %err, @@ -382,7 +386,7 @@ 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 +/// Log 1 ERROR as an agreement reaches the limit, for an operator to look into, and a WARN for /// every other failed cancel. fn log_failed_cancel( row: &CancellingAgreement, @@ -442,10 +446,8 @@ mod tests { use super::*; use crate::{ - cancel_dispatch::tests::agreement, - chain_client::AgreementOnChain, - registry::{IndexingAgreementStatus, StubAgreementRegistry}, - worker::service::JobPriority, + cancel_dispatch::tests::agreement, chain_client::AgreementOnChain, + registry::StubAgreementRegistry, test_support::logged, worker::service::JobPriority, }; const DEADLINE: u64 = 1_000; @@ -460,6 +462,7 @@ mod tests { struct MockRegistry { cancelling: Vec, marked_cancelled: Mutex>, + marked_abandoned: Mutex>, marked_by_indexer: Mutex>, audits: Mutex>>, audited_by: Mutex>, @@ -468,6 +471,8 @@ mod tests { found_ended: Mutex>>, /// The chain listener marks it ended before the retry's own mark lands. listener_ended_it: bool, + /// Recording a check fails, as with the database down. + check_fails: bool, writes: Mutex>, /// Ended agreements awaiting replacement, until noted as replaced. awaiting_replacement: Vec, @@ -514,6 +519,16 @@ mod tests { self.writes.lock().unwrap().push("ended"); Ok(()) } + async fn mark_indexing_agreement_as_abandoned_by_indexer( + &self, + id: &IndexingAgreementId, + ) -> crate::registry::Result<()> { + if self.listener_ended_it { + return Err(crate::registry::Error::NoRecordsUpdated); + } + self.marked_abandoned.lock().unwrap().push(*id); + Ok(()) + } async fn record_cancel_audit( &self, _id: &IndexingAgreementId, @@ -548,6 +563,14 @@ mod tests { failed_attempts: u32, ended: Option, ) -> crate::registry::Result { + if self.listener_ended_it { + return Err(crate::registry::Error::NoRecordsUpdated); + } + if self.check_fails { + return Err(crate::registry::Error::BackendError( + dipper_pgregistry::Error::DbError(sqlx::Error::PoolTimedOut), + )); + } self.checks.fetch_add(1, Ordering::SeqCst); self.found_ended.lock().unwrap().push(ended); Ok(self.attempts.fetch_add(failed_attempts, Ordering::SeqCst) + failed_attempts) @@ -758,6 +781,14 @@ mod tests { self.0.lock().unwrap().push(indexing_request_id); Ok(dipper_pgmq::JobId::default()) } + async fn cancel_rejected_agreement_on_chain( + &self, + _agreement_id: IndexingAgreementId, + _accept: Option, + _priority: JobPriority, + ) -> anyhow::Result { + unimplemented!() + } async fn submit_offer( &self, _agreement_id: IndexingAgreementId, @@ -866,6 +897,19 @@ mod tests { assert_eq!(*registry.audits.lock().unwrap(), vec![Some(tx)]); } + #[tokio::test] + async fn ends_an_agreement_its_indexer_stopped_serving_as_abandoned() { + let mut registry = registry_with_one(true); + registry.cancelling[0].abandoned = true; + + let logs = logged(retry(®istry, &live_chain(), 0)).await; + + let id = registry.cancelling[0].agreement.id; + assert_eq!(*registry.marked_abandoned.lock().unwrap(), vec![id]); + assert!(registry.marked_cancelled.lock().unwrap().is_empty()); + assert!(logs.contains("ABANDONED_BY_INDEXER"), "{logs}"); + } + #[tokio::test] async fn records_the_cancel_before_marking_the_agreement_ended() { // The end is announced once the mark lands; recorded after, the announcement could @@ -996,30 +1040,6 @@ mod tests { } } - /// What dipper logs, at every level, while `run` runs. - async fn logged(run: impl Future) -> String { - struct Buffer(Arc>>); - impl std::io::Write for Buffer { - fn write(&mut self, bytes: &[u8]) -> std::io::Result { - self.0.lock().unwrap().extend_from_slice(bytes); - Ok(bytes.len()) - } - fn flush(&mut self) -> std::io::Result<()> { - Ok(()) - } - } - let lines = Arc::new(Mutex::new(Vec::new())); - let writer = Arc::clone(&lines); - let subscriber = tracing_subscriber::fmt() - .with_ansi(false) - .with_max_level(tracing::Level::DEBUG) - .with_writer(move || Buffer(Arc::clone(&writer))) - .finish(); - let _logging = tracing::subscriber::set_default(subscriber); - run.await; - String::from_utf8(lines.lock().unwrap().clone()).unwrap() - } - #[tokio::test] async fn logs_an_end_by_the_indexer_as_a_status_change_only_when_it_made_one() { for listener_ended_it in [false, true] { @@ -1047,6 +1067,11 @@ mod tests { changed, "{logs}" ); + assert_eq!( + logs.contains("old_status=CANCELLING new_status=CANCELED_BY_INDEXER"), + changed, + "{logs}" + ); assert_eq!( logs.contains("already recorded the indexer's end"), !changed, @@ -1055,6 +1080,38 @@ mod tests { } } + #[tokio::test] + async fn logs_a_check_the_listener_beat_at_debug_and_a_failed_one_as_a_warning() { + // The listener ending the agreement mid-sweep is expected; only a failed write needs a look. + let check = async |listener_ended_it, check_fails| { + let registry = MockRegistry { + listener_ended_it, + check_fails, + ..registry_with_one(false) + }; + // Withdrawn before its deadline, so the check is recorded and it stays cancelling. + logged(retry(®istry, &MockChain::default(), DEADLINE)).await + }; + + let beaten = check(true, false).await; + let failed = check(false, true).await; + + let line = |logs: &str, message: &str| { + let line = logs.lines().find(|line| line.contains(message)); + line.unwrap_or_else(|| panic!("no {message:?} in {logs}")) + .to_owned() + }; + assert!( + line(&beaten, "no longer cancelling").contains("DEBUG"), + "{beaten}" + ); + assert!(!beaten.contains("WARN"), "{beaten}"); + assert!( + line(&failed, "Failed to record a check").contains("WARN"), + "{failed}" + ); + } + #[tokio::test] async fn reads_an_ended_agreement_once_to_learn_who_ended_it() { let mut registry = registry_with_one(true); diff --git a/bin/dipper-service/src/network/service/chain_listener.rs b/bin/dipper-service/src/network/service/chain_listener.rs index 071c2e30..babae7de 100644 --- a/bin/dipper-service/src/network/service/chain_listener.rs +++ b/bin/dipper-service/src/network/service/chain_listener.rs @@ -17,7 +17,8 @@ //! | AcceptedOnChain | CanceledBy* | mark canceled (requester vs indexer from the CanceledBy* state) | //! | Expired | Accepted | mark AcceptedOnChain (recovery), run pending cancellations | //! | Expired | CanceledBy* | recover to AcceptedOnChain, run pending cancellations, then cancel | -//! | Rejected | Accepted | queue cancel-on-chain (adversarial path, dead code under proposal-first) | +//! | Rejected | Accepted | queue a chain check, which reopens its cancel if live (adversarial path) | +//! | CanceledByRequester | Accepted | queue a chain check, which reopens its cancel if live | //! | Canceled* | * | no-op (terminal) | //! //! ## Data Source @@ -57,6 +58,7 @@ use crate::{ AgreementRegistry, CancelKind, IndexingAgreement, IndexingAgreementStatus, PendingCancellationRegistry, ReconciliationItem, }, + worker::service::{ChainAccept, JobPriority, WorkerQueue}, }; /// Idle interval used when no `Created` agreements are awaiting acceptance. @@ -153,7 +155,7 @@ impl Handle { } /// Context required by the chain listener service -pub struct Ctx { +pub struct Ctx { /// Registry for querying and updating agreements pub registry: R, /// Chain event source (subgraph) @@ -171,6 +173,8 @@ pub struct Ctx { /// Emits `accepted` and `terminated` lifecycle events as on-chain state is reconciled. pub subgraph_indexing_agreements_events_emitter: Arc, + /// Queues the chain check of an agreement dipper ended that the subgraph shows accepted. + pub worker_queue: W, } /// State persisted in the database @@ -195,11 +199,12 @@ pub struct ChainListenerState { clippy::too_many_lines, reason = "predates this lint; fix when next touched" )] -pub fn new(ctx: Ctx) -> (Handle, impl Future>) +pub fn new(ctx: Ctx) -> (Handle, impl Future>) where R: AgreementRegistry + ChainListenerStateRegistry + PendingCancellationRegistry + Send + Sync, E: ChainEventSource, T: ChainClient + Send + Sync, + W: WorkerQueue + Send + Sync, { let (tx_stop, mut rx_stop) = mpsc::channel(1); @@ -211,6 +216,7 @@ where config, chain_listener_notify, subgraph_indexing_agreements_events_emitter, + worker_queue, } = ctx; let service = async move { @@ -287,6 +293,7 @@ where // prematurely. Persisted in memory only; on restart it resets // and the first poll's bound is strict. let mut last_chain_ts_persist_wall: std::time::Instant = std::time::Instant::now(); + let mut live_checks = LiveChecks::new(worker_queue); loop { tokio::select! { @@ -354,6 +361,7 @@ where bypass_chain_clock_defenses, ®istry, &chain_client, + &mut live_checks, &event_source, &mut rx_stop, &agreement_conf, @@ -471,7 +479,7 @@ struct DrainOutcome { clippy::too_many_lines, reason = "predates this lint; fix when next touched" )] -async fn drain_once( +async fn drain_once( cursor: &mut Cursor, cursor_hold: &mut Option, last_persisted_timestamp: &mut Option, @@ -485,6 +493,7 @@ async fn drain_once( bypass_chain_clock_defenses: bool, registry: &R, chain_client: &T, + live_checks: &mut LiveChecks, event_source: &E, rx_stop: &mut mpsc::Receiver<()>, config: &crate::config::IndexingAgreementConfig, @@ -493,6 +502,7 @@ where R: AgreementRegistry + ChainListenerStateRegistry + PendingCancellationRegistry + Send + Sync, E: ChainEventSource, T: ChainClient + Send + Sync, + W: WorkerQueue + Sync, { // Read-side cursor backs off by `reorg_buffer_blocks` so a // reorg that moves a state change across the boundary is still @@ -509,6 +519,7 @@ where } else { cursor.clone() }; + live_checks.forget_before(effective_cursor.block); let drain_start = std::time::Instant::now(); let mut pages_per_tick: u32 = 0; @@ -637,7 +648,7 @@ where } let agreement = agreements_by_id.remove(&snapshot.agreement_id); - match prepare_reconciliation(&snapshot, agreement, registry, chain_client).await { + match prepare_reconciliation(&snapshot, agreement, registry, live_checks).await { Ok(Some(prep)) => prepared.push(prep), Ok(None) => {} Err(err) => { @@ -881,15 +892,15 @@ fn apply_chain_ts_drift_cap( clippy::cognitive_complexity, reason = "predates this lint; fix when next touched" )] -async fn prepare_reconciliation( +async fn prepare_reconciliation( snapshot: &AgreementStateSnapshot, agreement: Option, registry: &R, - chain_client: &T, + live_checks: &mut LiveChecks, ) -> anyhow::Result> where R: AgreementRegistry + Sync, - T: ChainClient, + W: WorkerQueue + Sync, { tracing::debug!( agreement_id = %snapshot.agreement_id, @@ -919,31 +930,17 @@ where // (race with gRPC rejection or stale read), not the indexer. // Skipped when remote is already canceled because the cancel tx // would revert against canceled contract state. - if matches!(agreement.status, IndexingAgreementStatus::Rejected) - && snapshot.state.reached_accepted() - && !snapshot.state.is_canceled() - { - tracing::warn!( - agreement_id = %agreement.id, - indexer = %snapshot.indexer, - "Rejected agreement accepted on-chain, cancelling it" - ); - // 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 { - registry - .record_accepted_audit(&agreement.id, snapshot.accepted_at, &snapshot.accepted_tx) - .await - .context("failed to record the on-chain accept of a rejected agreement")?; + if ended_but_accepted(snapshot, &agreement) { + if agreement.status == IndexingAgreementStatus::Rejected { + tracing::warn!( + agreement_id = %agreement.id, + indexer = %snapshot.indexer, + "Rejected agreement accepted on-chain, checking whether to cancel it" + ); } + live_checks.queue(snapshot).await?; } - - 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, registry).await?; // Both transitions are applied atomically downstream so the // Accept-then-Cancel-in-one-snapshot path can't leak an intermediate @@ -1009,8 +1006,8 @@ where } /// 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. +/// write, as the terminated sweep waits only for the accept. What is on record wins, unless its end +/// predates the accept or has no transaction. The end is its last change, so a failure is returned. async fn record_accept_and_cancel_from_chain( snapshot: &AgreementStateSnapshot, agreement: &IndexingAgreement, @@ -1032,42 +1029,75 @@ async fn record_accept_and_cancel_from_chain( .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 -/// anyway, such as one that landed after dipper's cancel. Nothing else would end -/// it: reconciliation ignores an accept on a cancelled row. It goes back to -/// `Cancelling` for the cancel retry, unless the chain shows it already ended. Returns whether it -/// moved it back. -async fn reopen_if_cancelled_but_accepted( - snapshot: &AgreementStateSnapshot, - agreement: &IndexingAgreement, - registry: &R, - chain_client: &T, -) -> anyhow::Result -where - R: AgreementRegistry + Sync, - T: ChainClient, -{ - if agreement.status == IndexingAgreementStatus::CanceledByRequester - && snapshot.state.reached_accepted() +/// Whether dipper rejected or cancelled an agreement the subgraph shows accepted and not ended, +/// such as an offer that landed after dipper's cancel. Nothing else would end it, as +/// reconciliation ignores an accept on an ended row, so a job checks the chain for it. +fn ended_but_accepted(snapshot: &AgreementStateSnapshot, agreement: &IndexingAgreement) -> bool { + matches!( + agreement.status, + IndexingAgreementStatus::Rejected | IndexingAgreementStatus::CanceledByRequester + ) && snapshot.state.reached_accepted() && !snapshot.state.is_canceled() - { - return Ok( - crate::cancel_dispatch::reopen_if_live(registry, chain_client, agreement).await?, - ); +} + +/// Queues the chain check of an agreement dipper ended that the subgraph shows accepted, once +/// per change, as the reorg buffer and a held cursor read a change again. The job runs off the +/// drain, so a slow chain read can't hold up the rest of the page. +struct LiveChecks { + queue: W, + /// The changes checks were queued for, by agreement and block. + queued: std::collections::HashSet<(IndexingAgreementId, u64)>, +} + +impl LiveChecks { + fn new(queue: W) -> Self { + Self { + queue, + queued: std::collections::HashSet::new(), + } + } + + /// Queue the check for this change, unless it is already queued. + async fn queue(&mut self, snapshot: &AgreementStateSnapshot) -> anyhow::Result<()> { + let change = (snapshot.agreement_id, snapshot.last_state_change_block); + if self.queued.contains(&change) { + tracing::debug!( + agreement_id = %snapshot.agreement_id, + "On-chain check already queued for this change" + ); + return Ok(()); + } + let accept = (snapshot.accepted_at != 0).then(|| ChainAccept { + at: snapshot.accepted_at, + tx: snapshot.accepted_tx.clone(), + }); + self.queue + .cancel_rejected_agreement_on_chain( + snapshot.agreement_id, + accept, + JobPriority::Background, + ) + .await + .context("failed to queue the on-chain check of an agreement dipper ended")?; + self.queued.insert(change); + Ok(()) + } + + /// Forget the changes before `block`, which the drain no longer reads. + fn forget_before(&mut self, block: u64) { + self.queued.retain(|&(_, changed_at)| changed_at >= block); } - Ok(false) } -/// 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, to be announced with its -/// end once the cancel lands. A withdrawn offer reads as cancelled with no accept time. +/// An agreement dipper is cancelling stays `Cancelling` when the chain shows it accepted, so its +/// accept is recorded here, to be announced with its end once the cancel lands. A withdrawn offer +/// reads as cancelled with no accept time. async fn record_accept_of_cancelling( snapshot: &AgreementStateSnapshot, agreement: &IndexingAgreement, - reopened: bool, registry: &R, ) -> anyhow::Result<()> { - if (agreement.status != IndexingAgreementStatus::Cancelling && !reopened) + if agreement.status != IndexingAgreementStatus::Cancelling || !snapshot.state.reached_accepted() || snapshot.accepted_at == 0 { @@ -1149,32 +1179,35 @@ where /// Reconcile a single agreement snapshot against dipper's local DB. /// /// Compares the snapshot's remote state to the local `IndexingAgreementStatus` -/// and applies whatever transitions the diff implies. See the module-level -/// transition table for the full mapping. +/// and applies whatever transitions the diff implies, returning the chain checks it queued. +/// See the module-level transition table for the full mapping. #[cfg(test)] async fn reconcile_agreement( snapshot: &AgreementStateSnapshot, registry: &R, chain_client: &T, config: &crate::config::IndexingAgreementConfig, -) -> anyhow::Result<()> +) -> anyhow::Result> where R: AgreementRegistry + PendingCancellationRegistry + Sync, T: ChainClient, { + let mut live_checks = LiveChecks::new(tests::MockWorkerQueue::default()); let agreement = registry .get_indexing_agreement_by_id(&snapshot.agreement_id) .await?; - let Some(prep) = prepare_reconciliation(snapshot, agreement, registry, chain_client).await? - else { - return Ok(()); + let prep = prepare_reconciliation(snapshot, agreement, registry, &mut live_checks).await?; + let queued = live_checks.queue.queued(); + let Some(prep) = prep else { + return Ok(queued); }; let outcome = registry .apply_reconciliation(&prep.agreement.id, prep.item.apply_accept, prep.item.cancel) .await?; - finalize_reconciliation(&prep, outcome, registry, chain_client, config).await + finalize_reconciliation(&prep, outcome, registry, chain_client, config).await?; + Ok(queued) } /// Execute pending cancellations linked to a newly-accepted agreement. @@ -1595,7 +1628,7 @@ where .await { // Sent but not stamped: the next sweep re-sends (deduped by - // consumers on agreement_id). Better than dropping. + // consumers on agreement_id and event type). Better than dropping. tracing::warn!( agreement_id = %p.agreement_id, error = %err, @@ -2725,6 +2758,80 @@ mod tests { } } + /// A chain check the listener queued: the agreement, and the accept it carried. + pub(super) type QueuedCheck = (IndexingAgreementId, Option); + + /// Records the chain checks queued; queuing fails while `fail` is set. + #[derive(Clone, Default)] + pub(super) struct MockWorkerQueue { + checks: Arc>>, + fail: Arc, + } + + impl MockWorkerQueue { + pub(super) fn queued(&self) -> Vec { + self.checks.lock().unwrap().clone() + } + } + + #[async_trait::async_trait] + impl WorkerQueue for MockWorkerQueue { + async fn send_indexing_agreement_proposal( + &self, + _candidate_url: url::Url, + _agreement_id: IndexingAgreementId, + _indexing_request_id: IndexingRequestId, + _deployment_id: DeploymentId, + _deployment_chain_id: ChainId, + _priority: JobPriority, + ) -> anyhow::Result { + unimplemented!() + } + + async fn reassess_indexing_request( + &self, + _indexing_request_id: IndexingRequestId, + _deployment_id: DeploymentId, + _deployment_chain_id: ChainId, + _num_candidates: usize, + _priority: JobPriority, + ) -> anyhow::Result { + unimplemented!() + } + + async fn cancel_rejected_agreement_on_chain( + &self, + agreement_id: IndexingAgreementId, + accept: Option, + _priority: JobPriority, + ) -> anyhow::Result { + if self.fail.load(std::sync::atomic::Ordering::SeqCst) { + anyhow::bail!("queue unavailable"); + } + self.checks.lock().unwrap().push((agreement_id, accept)); + Ok(dipper_pgmq::JobId::default()) + } + + async fn submit_offer( + &self, + _agreement_id: IndexingAgreementId, + _indexing_request_id: IndexingRequestId, + _indexer_url: url::Url, + _deployment_id: DeploymentId, + _deployment_chain_id: ChainId, + _priority: JobPriority, + ) -> anyhow::Result { + unimplemented!() + } + } + + fn snapshot_accept() -> Option { + Some(ChainAccept { + at: TEST_ACCEPTED_AT, + tx: TEST_ACCEPTED_TX.to_string(), + }) + } + // -- reconcile_agreement tests -- #[tokio::test] @@ -2942,52 +3049,42 @@ mod tests { } #[tokio::test] - async fn test_reconcile_reopens_the_cancel_of_a_rejected_agreement_live_on_chain() { - let registry = MockRegistry::new(); - let chain_client = MockChainClient { - live_until_cancelled: true, - ..MockChainClient::default() - }; - let agreement_id = IndexingAgreementId::from_bytes(rand::random()); - - registry.add_agreement(agreement_id, IndexingAgreementStatus::Rejected); - - let snapshot = make_snapshot(agreement_id, AgreementState::Accepted, Address::ZERO); - let result = reconcile_agreement( - &snapshot, - ®istry, - &chain_client, - test_agreement_conf().as_ref(), - ) - .await; - - assert!(result.is_ok()); - assert!(!registry.was_marked_accepted_on_chain(&agreement_id)); - assert!(registry.was_reopened(&agreement_id)); - } - - #[tokio::test] - async fn test_reconcile_records_no_accept_for_a_rejected_agreement_already_ended() { - // The snapshot shows it accepted, but by the time dipper reads the chain the indexer - // has ended it. It stays Rejected, so an accept on record would be announced with no - // end ever following it. - let registry = MockRegistry::new(); - let chain_client = MockChainClient::default(); - let agreement_id = IndexingAgreementId::from_bytes(rand::random()); - registry.add_agreement(agreement_id, IndexingAgreementStatus::Rejected); + async fn test_reconcile_queues_a_chain_check_for_an_ended_agreement_accepted_on_chain() { + // Dipper rejected or cancelled it, but the indexer accepted it on-chain anyway, such as an + // offer that landed after dipper's cancel. The chain is read by the queued job, not here, + // so a slow read can't hold up the page, and the job records the accept if it reopens it. + for status in [ + IndexingAgreementStatus::Rejected, + IndexingAgreementStatus::CanceledByRequester, + ] { + let registry = MockRegistry::new(); + let chain_client = live_chain(); + let agreement_id = IndexingAgreementId::from_bytes(rand::random()); + registry.add_agreement(agreement_id, status); - let snapshot = make_snapshot(agreement_id, AgreementState::Accepted, Address::ZERO); - let result = reconcile_agreement( - &snapshot, - ®istry, - &chain_client, - test_agreement_conf().as_ref(), - ) - .await; + let snapshot = make_snapshot(agreement_id, AgreementState::Accepted, Address::ZERO); + let queued = reconcile_agreement( + &snapshot, + ®istry, + &chain_client, + test_agreement_conf().as_ref(), + ) + .await + .expect("reconcile ok"); - assert!(result.is_ok()); - assert!(!registry.was_reopened(&agreement_id)); - assert!(registry.audit_writes().is_empty()); + assert_eq!(queued, vec![(agreement_id, snapshot_accept())], "{status}"); + assert_eq!( + chain_client.reads.load(std::sync::atomic::Ordering::SeqCst), + 0, + "{status}" + ); + assert!(!registry.was_reopened(&agreement_id), "{status}"); + assert!(registry.audit_writes().is_empty(), "{status}"); + assert!( + !registry.was_marked_accepted_on_chain(&agreement_id), + "{status}" + ); + } } #[tokio::test] @@ -3019,8 +3116,7 @@ mod tests { ) .await; - assert!(result.is_ok()); - assert!(!registry.was_reopened(&agreement_id)); + assert!(result.expect("reconcile ok").is_empty(), "no check queued"); assert!(registry.was_marked_canceled_by_requester(&agreement_id)); assert!(!registry.was_marked_accepted_on_chain(&agreement_id)); } @@ -3154,65 +3250,11 @@ mod tests { ) .await; - assert!(result.is_ok()); + // Already canceled on-chain: must not queue a fresh check. + assert!(result.expect("reconcile ok").is_empty()); assert!(registry.was_marked_canceled_by_requester(&agreement_id)); assert!(!registry.was_marked_canceled_by_indexer(&agreement_id)); assert!(!registry.was_marked_accepted_on_chain(&agreement_id)); - // Already canceled on-chain: must not queue a fresh cancel job. - assert!(!registry.was_reopened(&agreement_id)); - } - - #[tokio::test] - async fn test_reconcile_reopens_the_cancel_of_a_cancelled_agreement_live_on_chain() { - // Dipper cancelled the agreement locally, but the indexer accepted its offer - // (for example one that landed after dipper's cancel). Nothing else would end - // it, so it goes back to cancelling for the cancel retry to end. - let registry = MockRegistry::new(); - let chain_client = MockChainClient { - live_until_cancelled: true, - ..MockChainClient::default() - }; - let agreement_id = IndexingAgreementId::from_bytes(rand::random()); - registry.add_agreement(agreement_id, IndexingAgreementStatus::CanceledByRequester); - - let snapshot = make_snapshot(agreement_id, AgreementState::Accepted, Address::ZERO); - let result = reconcile_agreement( - &snapshot, - ®istry, - &chain_client, - test_agreement_conf().as_ref(), - ) - .await; - - assert!(result.is_ok()); - assert!(registry.was_reopened(&agreement_id)); - assert!(!registry.was_marked_accepted_on_chain(&agreement_id)); - assert!( - registry.audit_writes().contains(&("accept", agreement_id)), - "its accept is recorded, so the cancel retry treats it as paying" - ); - } - - #[tokio::test] - async fn test_reconcile_leaves_a_cancelled_agreement_the_chain_shows_ended() { - // The subgraph can still report an agreement accepted for a few polls after - // dipper's cancel lands; reopening it would only bring it back to cancelling. - let registry = MockRegistry::new(); - let chain_client = MockChainClient::default(); - let agreement_id = IndexingAgreementId::from_bytes(rand::random()); - registry.add_agreement(agreement_id, IndexingAgreementStatus::CanceledByRequester); - - let snapshot = make_snapshot(agreement_id, AgreementState::Accepted, Address::ZERO); - reconcile_agreement( - &snapshot, - ®istry, - &chain_client, - test_agreement_conf().as_ref(), - ) - .await - .expect("reconcile ok"); - - assert!(!registry.was_reopened(&agreement_id)); } #[tokio::test] @@ -3231,8 +3273,7 @@ mod tests { ) .await; - assert!(result.is_ok()); - assert!(!registry.was_reopened(&agreement_id)); + assert!(result.expect("reconcile ok").is_empty()); } #[tokio::test] @@ -3532,7 +3573,7 @@ mod tests { assert!(!registry.was_cancel_audit_recorded(&old_id)); } - /// Runs the pending cancellation of one old agreement in `status`, returning + /// Runs the pending cancellation of 1 old agreement in `status`, returning /// whether it was already marked cancelling when its on-chain cancel went out. async fn marked_at_pending_cancel(status: IndexingAgreementStatus) -> Vec { let registry = MockRegistry::new(); @@ -3992,6 +4033,7 @@ mod tests { subgraph_indexing_agreements_events_emitter: Arc::new( SubgraphIndexingAgreementsEventsEmitter::disabled(), ), + worker_queue: MockWorkerQueue::default(), }; let (handle, service) = new(ctx); @@ -4075,6 +4117,7 @@ mod tests { subgraph_indexing_agreements_events_emitter: Arc::new( SubgraphIndexingAgreementsEventsEmitter::disabled(), ), + worker_queue: MockWorkerQueue::default(), }; let (handle, service) = new(ctx); @@ -4155,6 +4198,7 @@ mod tests { subgraph_indexing_agreements_events_emitter: Arc::new( SubgraphIndexingAgreementsEventsEmitter::disabled(), ), + worker_queue: MockWorkerQueue::default(), }; let (handle, service) = new(ctx); @@ -4218,6 +4262,7 @@ mod tests { let mut stall_count: u32 = 0; let mut consecutive_failures: u32 = 0; let (_tx_stop, mut rx_stop) = mpsc::channel::<()>(1); + let mut live_checks = LiveChecks::new(MockWorkerQueue::default()); let outcome = drain_once( &mut cursor, @@ -4233,6 +4278,7 @@ mod tests { false, ®istry, &chain_client, + &mut live_checks, &event_source, &mut rx_stop, test_agreement_conf().as_ref(), @@ -4294,6 +4340,7 @@ mod tests { let mut stall_count: u32 = 0; let mut consecutive_failures: u32 = 0; let (_tx_stop, mut rx_stop) = mpsc::channel::<()>(1); + let mut live_checks = LiveChecks::new(MockWorkerQueue::default()); for fails in apply_fails { registry.set_fail_batch(*fails); @@ -4311,6 +4358,7 @@ mod tests { false, ®istry, &chain_client, + &mut live_checks, &event_source, &mut rx_stop, test_agreement_conf().as_ref(), @@ -4352,13 +4400,18 @@ mod tests { } /// 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. + /// audit writes and job queuing failing, then working, then from the start again, as the + /// reorg buffer re-reads it. The agreement, the registry, the queue and each drain's cursor. async fn drain_a_page_whose_audit_write_fails( status: IndexingAgreementStatus, state: AgreementState, - chain_client: &MockChainClient, - ) -> (IndexingAgreementId, MockRegistry, Vec) { + ) -> ( + IndexingAgreementId, + MockRegistry, + MockWorkerQueue, + Vec, + ) { + let chain_client = live_chain(); let registry = MockRegistry::new(); let event_source = super::super::chain_events::mock::MockEventSource::new(); event_source.set_latest_block(100); @@ -4374,6 +4427,8 @@ mod tests { let mut stall_count: u32 = 0; let mut consecutive_failures: u32 = 0; let (_tx_stop, mut rx_stop) = mpsc::channel::<()>(1); + let queue = MockWorkerQueue::default(); + let mut live_checks = LiveChecks::new(queue.clone()); let mut cursors = Vec::new(); for (writes_fail, from_start) in [(true, false), (false, false), (false, true)] { @@ -4382,6 +4437,9 @@ mod tests { mock.fail_accept_audit = writes_fail; mock.fail_cancel_audit = writes_fail; } + queue + .fail + .store(writes_fail, std::sync::atomic::Ordering::SeqCst); if from_start { cursor = Cursor::genesis(); } @@ -4398,7 +4456,8 @@ mod tests { 10, false, ®istry, - chain_client, + &chain_client, + &mut live_checks, &event_source, &mut rx_stop, test_agreement_conf().as_ref(), @@ -4407,7 +4466,12 @@ mod tests { .expect("subgraph fetch succeeded"); cursors.push(cursor.clone()); } - (agreement_id, registry, cursors) + assert_eq!( + chain_client.reads.load(std::sync::atomic::Ordering::SeqCst), + 0, + "the drain reads nothing from the chain" + ); + (agreement_id, registry, queue, cursors) } /// The end of an agreement dipper already ended is its last change on-chain, so if recording @@ -4415,10 +4479,9 @@ mod tests { /// 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( + let (id, registry, _, cursors) = drain_a_page_whose_audit_write_fails( IndexingAgreementStatus::CanceledByRequester, AgreementState::CanceledByPayer, - &MockChainClient::default(), ) .await; @@ -4437,30 +4500,43 @@ mod tests { 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. + /// The same holds for the accept of an agreement being cancelled. #[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( + IndexingAgreementStatus::Cancelling, + AgreementState::Accepted, + ) + .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", id); 2]); + } + + /// A rejected or cancelled agreement accepted on-chain is read again until its chain check + /// is queued, and only queued once, however often the change is read. + #[tokio::test] + async fn test_drain_queues_the_chain_check_of_an_ended_agreement_once() { + for status in [ + IndexingAgreementStatus::Rejected, + IndexingAgreementStatus::CanceledByRequester, ] { - let (id, registry, cursors) = drain_a_page_whose_audit_write_fails( - status, - AgreementState::Accepted, - &chain_client, - ) - .await; + let (id, registry, queue, cursors) = + drain_a_page_whose_audit_write_fails(status, AgreementState::Accepted).await; assert_eq!( cursors[0], Cursor::genesis(), - "{status}: held before the failed write" + "{status}: held before the failed queuing" ); assert!( cursors[1..] @@ -4468,12 +4544,9 @@ mod tests { .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}" - ); + assert_eq!(queue.queued(), vec![(id, snapshot_accept())], "{status}"); + assert!(!registry.was_reopened(&id), "{status}"); + assert!(registry.audit_writes().is_empty(), "{status}"); } } @@ -4539,6 +4612,7 @@ mod tests { let mut stall_count: u32 = 0; let mut consecutive_failures: u32 = 0; let (_tx_stop, mut rx_stop) = mpsc::channel::<()>(1); + let mut live_checks = LiveChecks::new(MockWorkerQueue::default()); drain_once( &mut cursor, @@ -4554,6 +4628,7 @@ mod tests { false, ®istry, &chain_client, + &mut live_checks, &event_source, &mut rx_stop, test_agreement_conf().as_ref(), @@ -4582,6 +4657,7 @@ mod tests { false, ®istry, &chain_client, + &mut live_checks, &event_source, &mut rx_stop, test_agreement_conf().as_ref(), @@ -4671,6 +4747,7 @@ mod tests { let mut stall_count: u32 = 0; let mut consecutive_failures: u32 = 0; let (_tx_stop, mut rx_stop) = mpsc::channel::<()>(1); + let mut live_checks = LiveChecks::new(MockWorkerQueue::default()); drain_once( &mut cursor, @@ -4686,6 +4763,7 @@ mod tests { false, ®istry, &chain_client, + &mut live_checks, &event_source, &mut rx_stop, test_agreement_conf().as_ref(), @@ -4735,6 +4813,7 @@ mod tests { subgraph_indexing_agreements_events_emitter: Arc::new( SubgraphIndexingAgreementsEventsEmitter::disabled(), ), + worker_queue: MockWorkerQueue::default(), }; let (handle, service) = new(ctx); diff --git a/bin/dipper-service/src/network/service/expiration.rs b/bin/dipper-service/src/network/service/expiration.rs index 12802fcf..85162713 100644 --- a/bin/dipper-service/src/network/service/expiration.rs +++ b/bin/dipper-service/src/network/service/expiration.rs @@ -10,7 +10,10 @@ use tokio::{sync::mpsc, time::MissedTickBehavior}; use crate::{ config::ExpirationConfig, network::service::chain_listener::ChainListenerStateRegistry, - registry::{AgreementRegistry, IndexingRequestRegistry, PendingCancellationRegistry}, + registry::{ + AgreementRegistry, IndexingAgreementStatus, IndexingRequestRegistry, + PendingCancellationRegistry, + }, worker::service::{JobPriority, WorkerQueue}, }; @@ -185,8 +188,8 @@ where tracing::info!( agreement_id = %agreement.id, indexing_request_id = %agreement.indexing_request_id, - old_status = "Created", - new_status = "Expired", + old_status = %IndexingAgreementStatus::Created, + new_status = %IndexingAgreementStatus::Expired, "agreement state transition" ); @@ -538,6 +541,14 @@ mod tests { ) -> anyhow::Result { Ok(JobId::default()) } + async fn cancel_rejected_agreement_on_chain( + &self, + _agreement_id: IndexingAgreementId, + _accept: Option, + _priority: JobPriority, + ) -> anyhow::Result { + unimplemented!() + } async fn submit_offer( &self, _agreement_id: IndexingAgreementId, diff --git a/bin/dipper-service/src/network/service/liveness_checker.rs b/bin/dipper-service/src/network/service/liveness_checker.rs index bf0a59eb..c0d84c82 100644 --- a/bin/dipper-service/src/network/service/liveness_checker.rs +++ b/bin/dipper-service/src/network/service/liveness_checker.rs @@ -1453,6 +1453,14 @@ mod tests { self.calls.reassessments.lock().unwrap().push(req_id); Ok(JobId::default()) } + async fn cancel_rejected_agreement_on_chain( + &self, + _agreement_id: IndexingAgreementId, + _accept: Option, + _priority: JobPriority, + ) -> anyhow::Result { + unimplemented!() + } async fn submit_offer( &self, _agreement_id: IndexingAgreementId, diff --git a/bin/dipper-service/src/registry.rs b/bin/dipper-service/src/registry.rs index e194517d..bb07b366 100644 --- a/bin/dipper-service/src/registry.rs +++ b/bin/dipper-service/src/registry.rs @@ -99,6 +99,30 @@ fn filter_map_with_logging( } } +/// Warn when a write changed an agreement's end after its `terminated` event went out. The event +/// goes out again with the new end, but consumers dedupe on the agreement id and event type, so +/// they may drop it and keep the old end. +fn warn_if_end_corrected( + agreement_id: &IndexingAgreementId, + correction: Option, +) { + let Some(dipper_pgregistry::EndCorrection { old, new }) = correction else { + return; + }; + tracing::warn!( + event = "agreement_end_corrected_after_announcement", + %agreement_id, + old_canceled_at = ?old.canceled_at, + old_canceled_by = ?old.canceled_by, + old_canceled_tx = ?old.canceled_tx, + new_canceled_at = ?new.canceled_at, + new_canceled_by = ?new.canceled_by, + new_canceled_tx = ?new.canceled_tx, + "An agreement's end changed after its terminated event went out; the event goes out \ + again, but consumers may drop it as a repeat" + ); +} + /// A service for interacting with the registry, including registering new /// indexing requests and indexing agreements. #[derive(Clone)] @@ -391,6 +415,16 @@ impl AgreementRegistry for RegistryProvider { .map_err(Into::into) } + async fn mark_indexing_agreement_as_abandoned_by_indexer( + &self, + id: &IndexingAgreementId, + ) -> RegistryResult<()> { + self.inner + .mark_indexing_agreement_as_abandoned_by_indexer(id) + .await + .map_err(Into::into) + } + async fn mark_indexing_agreement_as_cancelling( &self, id: &IndexingAgreementId, @@ -416,10 +450,12 @@ impl AgreementRegistry for RegistryProvider { id: &IndexingAgreementId, seen_live: bool, ) -> RegistryResult<()> { - self.inner + let correction = self + .inner .reopen_indexing_agreement_cancel(id, seen_live) - .await - .map_err(Into::into) + .await?; + warn_if_end_corrected(id, correction); + Ok(()) } async fn get_cancelling_agreements( @@ -620,9 +656,11 @@ impl AgreementRegistry for RegistryProvider { canceled_by: &str, canceled_tx: Option<&str>, ) -> RegistryResult<()> { - self.inner + let correction = self + .inner .record_cancel_audit(agreement_id, canceled_at, canceled_by, canceled_tx) .await?; + warn_if_end_corrected(agreement_id, correction); Ok(()) } @@ -635,7 +673,8 @@ impl AgreementRegistry for RegistryProvider { canceled_by: &str, canceled_tx: Option<&str>, ) -> RegistryResult<()> { - self.inner + let correction = self + .inner .record_accept_and_cancel_audit( agreement_id, accepted_at, @@ -645,6 +684,7 @@ impl AgreementRegistry for RegistryProvider { canceled_tx, ) .await?; + warn_if_end_corrected(agreement_id, correction); Ok(()) } @@ -894,3 +934,53 @@ impl crate::network::service::chain_listener::ChainListenerStateRegistry for Reg .map_err(Into::into) } } + +#[cfg(test)] +mod tests { + use dipper_core::ids::IndexingAgreementId; + use dipper_pgregistry::{EndCorrection, RecordedEnd}; + + use super::warn_if_end_corrected; + use crate::test_support::logged; + + #[tokio::test] + async fn warns_with_both_ends_when_an_announced_end_changes() { + let id = IndexingAgreementId::from_bytes([0xaa; 16]); + let correction = EndCorrection { + old: RecordedEnd { + canceled_at: Some(9_000), + canceled_by: Some("0xindexer".to_owned()), + canceled_tx: None, + }, + new: RecordedEnd { + canceled_at: Some(5_000), + canceled_by: Some("0xprovider".to_owned()), + canceled_tx: Some("0xend".to_owned()), + }, + }; + + let logs = logged(async { warn_if_end_corrected(&id, Some(correction)) }).await; + + for expected in [ + "WARN", + "event=\"agreement_end_corrected_after_announcement\"", + &format!("agreement_id={id}"), + "old_canceled_at=Some(9000)", + "old_canceled_tx=None", + "new_canceled_at=Some(5000)", + "new_canceled_tx=Some(\"0xend\")", + "consumers may drop it", + ] { + assert!(logs.contains(expected), "missing {expected} in {logs}"); + } + } + + #[tokio::test] + async fn stays_quiet_when_no_announced_end_changes() { + let id = IndexingAgreementId::from_bytes([0xaa; 16]); + + let logs = logged(async { warn_if_end_corrected(&id, None) }).await; + + assert!(logs.is_empty(), "{logs}"); + } +} diff --git a/bin/dipper-service/src/registry/agreement.rs b/bin/dipper-service/src/registry/agreement.rs index e4416cb5..d4a87a39 100644 --- a/bin/dipper-service/src/registry/agreement.rs +++ b/bin/dipper-service/src/registry/agreement.rs @@ -260,13 +260,20 @@ pub trait AgreementRegistry { /// /// If there is no indexing agreement with the given ID, or if the agreement is not in the /// `CREATED`, `ACCEPTED_ON_CHAIN`, `REJECTED` or `CANCELLING` state, this method returns a - /// [`NoRecordUpdated`](Error::NoRecordsUpdated) error. One dipper was cancelling because its - /// indexer stopped serving it becomes `ABANDONED_BY_INDEXER` instead. + /// [`NoRecordUpdated`](Error::NoRecordsUpdated) error. async fn mark_indexing_agreement_as_canceled_by_requester( &self, id: &IndexingAgreementId, ) -> RegistryResult<()>; + /// Mark a `CANCELLING` agreement dipper was ending because its indexer stopped serving it + /// `ABANDONED_BY_INDEXER`, once the chain confirms dipper's cancel; + /// [`NoRecordUpdated`](Error::NoRecordsUpdated) otherwise. + async fn mark_indexing_agreement_as_abandoned_by_indexer( + &self, + id: &IndexingAgreementId, + ) -> RegistryResult<()>; + /// Mark a `CREATED`, `ACCEPTED_ON_CHAIN`, `REJECTED` or `EXPIRED` agreement `CANCELLING`, /// before its cancel is sent; [`NoRecordUpdated`](Error::NoRecordsUpdated) otherwise. async fn mark_indexing_agreement_as_cancelling( @@ -324,7 +331,8 @@ pub trait AgreementRegistry { /// - `apply_accept`: attempt `Created | Expired → AcceptedOnChain`. /// May be a no-op if the row is already in another status. /// - `cancel`: optionally apply `CanceledByRequester` or - /// `CanceledByIndexer` from an accepting/pre-accept state. + /// `CanceledByIndexer` from an accepting/pre-accept state. Dipper's cancel of one it was + /// ending because its indexer stopped serving it applies `AbandonedByIndexer` instead. /// /// Returns which transitions actually affected a row so callers can /// gate post-commit side effects (e.g. running diff --git a/bin/dipper-service/src/registry/agreement_stub.rs b/bin/dipper-service/src/registry/agreement_stub.rs index 1253c754..ef5bc214 100644 --- a/bin/dipper-service/src/registry/agreement_stub.rs +++ b/bin/dipper-service/src/registry/agreement_stub.rs @@ -133,6 +133,16 @@ pub trait StubAgreementRegistry: Send + Sync { unimplemented!("mark_indexing_agreement_as_canceled_by_requester") } + /// Defaults to the requester mark, so a mock that records only that dipper ended an + /// agreement sees both kinds of end; override it to tell them apart. + async fn mark_indexing_agreement_as_abandoned_by_indexer( + &self, + id: &IndexingAgreementId, + ) -> Result<()> { + self.mark_indexing_agreement_as_canceled_by_requester(id) + .await + } + async fn mark_indexing_agreement_as_cancelling(&self, _id: &IndexingAgreementId) -> Result<()> { unimplemented!("mark_indexing_agreement_as_cancelling") } @@ -473,6 +483,13 @@ impl AgreementRegistry for T { StubAgreementRegistry::mark_indexing_agreement_as_canceled_by_requester(self, id).await } + async fn mark_indexing_agreement_as_abandoned_by_indexer( + &self, + id: &IndexingAgreementId, + ) -> Result<()> { + StubAgreementRegistry::mark_indexing_agreement_as_abandoned_by_indexer(self, id).await + } + async fn mark_indexing_agreement_as_cancelling(&self, id: &IndexingAgreementId) -> Result<()> { StubAgreementRegistry::mark_indexing_agreement_as_cancelling(self, id).await } diff --git a/bin/dipper-service/src/test_support.rs b/bin/dipper-service/src/test_support.rs index 4ef6eb37..d3c4112e 100644 --- a/bin/dipper-service/src/test_support.rs +++ b/bin/dipper-service/src/test_support.rs @@ -3,7 +3,8 @@ //! Currently provides [`CapturingEventsProducer`], a test double for //! [`SubgraphIndexingAgreementEventsProducer`] that records every produced event //! so handler/service tests can assert exactly which lifecycle events were emitted -//! (and with which payloads) without a Kafka backend. +//! (and with which payloads) without a Kafka backend, and [`logged`], which captures what +//! dipper logs. use std::sync::{Arc, Mutex}; @@ -157,3 +158,27 @@ impl SubgraphIndexingAgreementEventsProducer for CapturingEventsProducer { Ok(()) } } + +/// What dipper logs, at every level, while `run` runs. +pub async fn logged(run: impl std::future::Future) -> String { + struct Buffer(Arc>>); + impl std::io::Write for Buffer { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(bytes); + Ok(bytes.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + let lines = Arc::new(Mutex::new(Vec::new())); + let writer = Arc::clone(&lines); + let subscriber = tracing_subscriber::fmt() + .with_ansi(false) + .with_max_level(tracing::Level::DEBUG) + .with_writer(move || Buffer(Arc::clone(&writer))) + .finish(); + let _logging = tracing::subscriber::set_default(subscriber); + run.await; + String::from_utf8(lines.lock().unwrap().clone()).unwrap() +} diff --git a/bin/dipper-service/src/worker/handlers.rs b/bin/dipper-service/src/worker/handlers.rs index fe545d82..bfab3b83 100644 --- a/bin/dipper-service/src/worker/handlers.rs +++ b/bin/dipper-service/src/worker/handlers.rs @@ -6,8 +6,8 @@ mod send_indexing_agreement_proposal; mod submit_offer; pub use cancel_rejected_agreement_on_chain::{ - Ctx as CancelRejectedAgreementOnChainCtx, Message as CancelRejectedAgreementOnChain, - handle as cancel_rejected_agreement_on_chain, + ChainAccept, Ctx as CancelRejectedAgreementOnChainCtx, + Message as CancelRejectedAgreementOnChain, handle as cancel_rejected_agreement_on_chain, }; pub use reassess_indexing_request::{ Ctx as ReassessIndexingRequestCtx, Message as ReassessIndexingRequest, 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 c6d5735c..aadb9f8d 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 @@ -1,31 +1,56 @@ -//! Kept so jobs queued before an upgrade still run. The chain listener now moves an -//! agreement dipper rejected or cancelled that went live on-chain anyway back into -//! `Cancelling` itself, and the cancel retry ends it; this job does the same. +//! Moves an agreement dipper rejected or cancelled that is live on-chain after all back into +//! `Cancelling`, with the accept the chain listener saw on record. The listener queues this +//! rather than reading the chain itself, so a slow read can't hold up the rest of its page. use std::time::Duration; use dipper_core::ids::IndexingAgreementId; use crate::{ - cancel_dispatch::reopen_if_live, + cancel_dispatch::{Reopen, reopen_if_live}, chain_client::ChainClient, - registry::{AgreementRegistry, IndexingAgreementStatus}, + registry::{AgreementRegistry, IndexingAgreement, IndexingAgreementStatus}, worker::result::{JobError, JobResult}, }; +/// How long a job whose database read or write failed waits before running again. It keeps +/// running until the database answers: the listener queues it once per change, and nothing else +/// checks an ended agreement for a live one, so a job given up on would leave its indexer paid. +const RETRY_DELAY: Duration = Duration::from_secs(30); + +/// How long a job that couldn't read a cancelled agreement on-chain waits to read it again, at +/// the cancel retry's own pace. It keeps reading until the chain answers. +const CHAIN_READ_RETRY_DELAY: Duration = Duration::from_secs(300); + pub struct Ctx { pub registry: R, pub chain_client: T, } -/// Cancel on-chain an agreement dipper rejected or cancelled that was accepted anyway. +/// Check on-chain an agreement dipper rejected or cancelled that was accepted anyway. #[derive(Debug, serde::Serialize, serde::Deserialize)] pub struct Message { pub agreement_id: IndexingAgreementId, + /// The accept the chain listener saw; absent from jobs queued before it was carried. + #[serde(default)] + pub accept: Option, +} + +/// When an agreement was accepted on-chain, and in which transaction. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct ChainAccept { + pub at: u64, + pub tx: String, } /// Hand an agreement dipper rejected or cancelled that is live on-chain to the cancel retry. -pub async fn handle(ctx: Ctx, Message { agreement_id }: &Message) -> JobResult<()> +pub async fn handle( + ctx: Ctx, + Message { + agreement_id, + accept, + }: &Message, +) -> JobResult<()> where R: AgreementRegistry + Sync, T: ChainClient, @@ -34,21 +59,66 @@ where .registry .get_indexing_agreement_by_id(agreement_id) .await - .map_err(|err| JobError::Fatal(err.into()))? + .map_err(|err| retry(agreement_id, "read", err.into()))? else { - tracing::warn!(%agreement_id, "Agreement not found for on-chain cancellation"); + tracing::warn!(%agreement_id, "Agreement not found for on-chain check"); return Ok(()); }; - if !matches!( - agreement.status, - IndexingAgreementStatus::Rejected | IndexingAgreementStatus::CanceledByRequester - ) { + let cancelling = match agreement.status { + IndexingAgreementStatus::Rejected | IndexingAgreementStatus::CanceledByRequester => { + reopen(&ctx, &agreement).await? + } + // Moved back by an earlier run of this job, or a copy of it, so only its accept is left. + IndexingAgreementStatus::Cancelling => true, + _ => return Ok(()), + }; + if !cancelling { return Ok(()); } - reopen_if_live(&ctx.registry, &ctx.chain_client, &agreement) + let Some(accept) = accept else { + return Ok(()); + }; + ctx.registry + .record_accepted_audit(agreement_id, accept.at, &accept.tx) .await - .map(|_| ()) - .map_err(|err| JobError::Retryable(err.into(), Duration::from_secs(30))) + .map_err(|err| retry(agreement_id, "record the on-chain accept of", err.into())) +} + +/// Move an agreement back into `Cancelling` if the chain shows it live; whether it moved. +async fn reopen(ctx: &Ctx, agreement: &IndexingAgreement) -> JobResult +where + R: AgreementRegistry + Sync, + T: ChainClient, +{ + let agreement_id = &agreement.id; + match reopen_if_live(&ctx.registry, &ctx.chain_client, agreement) + .await + .map_err(|err| retry(agreement_id, "reopen the cancel of", err.into()))? + { + Reopen::Reopened => Ok(true), + Reopen::Left => Ok(false), + Reopen::Unread(err) => { + tracing::warn!( + %agreement_id, + error = %err, + retry_secs = CHAIN_READ_RETRY_DELAY.as_secs(), + "Failed to read whether a cancelled agreement reported live is live; will read again" + ); + Err(JobError::Deferred(CHAIN_READ_RETRY_DELAY)) + } + } +} + +/// Log a failed database read or write and run the job again later, without using up its +/// attempts. +fn retry(agreement_id: &IndexingAgreementId, action: &str, err: anyhow::Error) -> JobError { + tracing::warn!( + %agreement_id, + error = format!("{err:#}"), + retry_secs = RETRY_DELAY.as_secs(), + "Failed to {action} an agreement dipper ended that was accepted on-chain; will retry" + ); + JobError::Deferred(RETRY_DELAY) } #[cfg(test)] @@ -63,12 +133,27 @@ mod tests { use crate::{ cancel_dispatch::tests::agreement, chain_client::{AgreementOnChain, ChainClientError}, - registry::{IndexingAgreement, StubAgreementRegistry}, + registry::StubAgreementRegistry, }; - struct MockRegistry { - agreement: IndexingAgreement, - reopened: Arc>>, + #[derive(Default)] + struct State { + agreement: Option, + /// What each reopen was told about the chain. + reopened: Vec, + accepts: Vec, + fail_read: bool, + fail_reopen: bool, + fail_accept: bool, + } + + #[derive(Clone, Default)] + struct MockRegistry(Arc>); + + fn db_down() -> crate::registry::Error { + crate::registry::Error::BackendError(dipper_pgregistry::Error::DbError( + sqlx::Error::Protocol("simulated outage".into()), + )) } #[async_trait] @@ -77,20 +162,47 @@ mod tests { &self, _id: &IndexingAgreementId, ) -> crate::registry::Result> { - Ok(Some(self.agreement.clone())) + let state = self.0.lock().unwrap(); + if state.fail_read { + return Err(db_down()); + } + Ok(state.agreement.clone()) } async fn reopen_indexing_agreement_cancel( &self, - id: &IndexingAgreementId, - _seen_live: bool, + _id: &IndexingAgreementId, + seen_live: bool, + ) -> crate::registry::Result<()> { + let mut state = self.0.lock().unwrap(); + if state.fail_reopen { + return Err(db_down()); + } + let agreement = state.agreement.as_mut().expect("agreement"); + agreement.status = IndexingAgreementStatus::Cancelling; + state.reopened.push(seen_live); + Ok(()) + } + async fn record_accepted_audit( + &self, + _agreement_id: &IndexingAgreementId, + accepted_at: u64, + accepted_tx: &str, ) -> crate::registry::Result<()> { - self.reopened.lock().unwrap().push(*id); + let mut state = self.0.lock().unwrap(); + if state.fail_accept { + return Err(db_down()); + } + state.accepts.push(ChainAccept { + at: accepted_at, + tx: accepted_tx.to_string(), + }); Ok(()) } } + /// Every agreement reads as live or not, or the read fails when `live` is `None`. struct MockChain { - live: bool, + live: Option, } #[async_trait] @@ -128,30 +240,39 @@ mod tests { &self, _agreement_id: &[u8; 16], ) -> Result { - Ok(AgreementOnChain::live_if(self.live)) + self.live + .map(AgreementOnChain::live_if) + .ok_or_else(|| ChainClientError::RpcError(anyhow::anyhow!("every endpoint failed"))) } async fn latest_block_timestamp(&self) -> Result { unimplemented!() } } - async fn run(status: IndexingAgreementStatus, live: bool) -> Vec { - let reopened = Arc::new(Mutex::new(Vec::new())); - let agreement = agreement(status, Some(vec![7u8; 32])); + fn chain_accept() -> ChainAccept { + ChainAccept { + at: 1_700_000_001, + tx: "0x11".to_string(), + } + } + + fn registry(status: IndexingAgreementStatus) -> MockRegistry { + let registry = MockRegistry::default(); + registry.0.lock().unwrap().agreement = Some(agreement(status, Some(vec![7u8; 32]))); + registry + } + + async fn run_job(registry: &MockRegistry, live: Option) -> JobResult<()> { + let agreement_id = registry.0.lock().unwrap().agreement.as_ref().unwrap().id; let message = Message { - agreement_id: agreement.id, + agreement_id, + accept: Some(chain_accept()), }; let ctx = Ctx { - registry: MockRegistry { - agreement, - reopened: Arc::clone(&reopened), - }, + registry: registry.clone(), chain_client: MockChain { live }, }; - - handle(ctx, &message).await.expect("job ok"); - - reopened.lock().unwrap().clone() + handle(ctx, &message).await } #[tokio::test] @@ -160,21 +281,129 @@ mod tests { IndexingAgreementStatus::Rejected, IndexingAgreementStatus::CanceledByRequester, ] { - assert_eq!(run(status, true).await.len(), 1, "{status}"); + let registry = registry(status); + + run_job(®istry, Some(true)).await.expect("job ok"); + + let state = registry.0.lock().unwrap(); + assert_eq!(state.reopened, vec![true], "{status}"); + assert_eq!( + state.accepts, + vec![chain_accept()], + "{status}: its accept is recorded, so the cancel retry treats it as paying" + ); } } #[tokio::test] async fn leaves_an_agreement_that_already_ended_or_is_still_wanted() { - assert!( - run(IndexingAgreementStatus::CanceledByRequester, false) - .await - .is_empty() - ); - assert!( - run(IndexingAgreementStatus::AcceptedOnChain, true) - .await - .is_empty() - ); + // An accept on record for one left as it is would be announced with no end to follow. + for (status, live) in [ + (IndexingAgreementStatus::Rejected, false), + (IndexingAgreementStatus::CanceledByRequester, false), + (IndexingAgreementStatus::AcceptedOnChain, true), + ] { + let registry = registry(status); + + run_job(®istry, Some(live)).await.expect("job ok"); + + let state = registry.0.lock().unwrap(); + assert!(state.reopened.is_empty(), "{status}"); + assert!(state.accepts.is_empty(), "{status}"); + } + } + + #[tokio::test] + async fn reopens_a_rejected_agreement_whose_chain_read_fails() { + // Its indexer may be paid, so it is reopened; it may have ended, so the end on record + // stays, and the cancel retry reads it again. + let registry = registry(IndexingAgreementStatus::Rejected); + + run_job(®istry, None).await.expect("job ok"); + + let state = registry.0.lock().unwrap(); + assert_eq!(state.reopened, vec![false]); + assert_eq!(state.accepts, vec![chain_accept()]); + } + + #[tokio::test] + async fn reads_a_cancelled_agreement_again_when_its_chain_read_fails() { + // Most likely dipper's cancel landed and the subgraph is behind, so it isn't reopened on + // a guess, but it may be an offer accepted after dipper withdrew it, so it is read again. + let registry = registry(IndexingAgreementStatus::CanceledByRequester); + + let unread = run_job(®istry, None).await; + let read = run_job(®istry, Some(true)).await; + + assert!(matches!( + unread, + Err(JobError::Deferred(CHAIN_READ_RETRY_DELAY)) + )); + read.expect("job ok"); + let state = registry.0.lock().unwrap(); + assert_eq!(state.reopened, vec![true], "reopened only once read live"); + assert_eq!(state.accepts, vec![chain_accept()]); + } + + #[tokio::test] + async fn retries_an_accept_that_failed_to_record_without_reopening_again() { + let registry = registry(IndexingAgreementStatus::CanceledByRequester); + registry.0.lock().unwrap().fail_accept = true; + + let first = run_job(®istry, Some(true)).await; + registry.0.lock().unwrap().fail_accept = false; + run_job(®istry, Some(true)).await.expect("retry ok"); + + assert!(matches!(first, Err(JobError::Deferred(RETRY_DELAY)))); + let state = registry.0.lock().unwrap(); + assert_eq!(state.reopened, vec![true]); + assert_eq!(state.accepts, vec![chain_accept()]); + } + + #[tokio::test] + async fn keeps_a_job_whose_database_keeps_failing() { + // The listener queues this check once per change, so a job given up on would leave an + // agreement live on-chain marked ended; deferring reruns it without using up attempts. + type Failure = (&'static str, fn(&mut State)); + let failures: [Failure; 3] = [ + ("read", |state| state.fail_read = true), + ("reopen", |state| state.fail_reopen = true), + ("accept", |state| state.fail_accept = true), + ]; + for (failing, fail) in failures { + let registry = registry(IndexingAgreementStatus::CanceledByRequester); + fail(&mut registry.0.lock().unwrap()); + + for _ in 0..5 { + let result = run_job(®istry, Some(true)).await; + assert!( + matches!(result, Err(JobError::Deferred(RETRY_DELAY))), + "{failing}: {result:?}" + ); + } + { + let mut state = registry.0.lock().unwrap(); + state.fail_read = false; + state.fail_reopen = false; + state.fail_accept = false; + } + run_job(®istry, Some(true)).await.expect("job ok"); + + let state = registry.0.lock().unwrap(); + assert_eq!(state.reopened, vec![true], "{failing}"); + assert_eq!(state.accepts, vec![chain_accept()], "{failing}"); + } + } + + /// A job queued before the listener carried the accept still runs. + #[test] + fn reads_a_job_queued_without_an_accept() { + let id = IndexingAgreementId::from_bytes([1; 16]); + let queued = serde_json::json!({ "agreement_id": id }); + + let message: Message = serde_json::from_value(queued).expect("old message"); + + assert_eq!(message.agreement_id, id); + assert_eq!(message.accept, None); } } 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 1d9f19fc..80a1df0c 100644 --- a/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs +++ b/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs @@ -27,9 +27,9 @@ use crate::{ }, }, registry::{ - AgreementRegistry, IndexerDenylistRegistry, IndexingAgreementTerms, - IndexingAgreementTermsMetadata, IndexingRequestRegistry, NewAgreementParams, - PendingCancellationRegistry, + AgreementRegistry, IndexerDenylistRegistry, IndexingAgreementStatus, + IndexingAgreementTerms, IndexingAgreementTermsMetadata, IndexingRequestRegistry, + NewAgreementParams, PendingCancellationRegistry, }, signing::eip712::Eip712Signer, worker::{ @@ -556,7 +556,7 @@ where agreement_id = %id, indexing_request_id = %indexing_request_id, old_status = "none", - new_status = "CREATED", + new_status = %IndexingAgreementStatus::Created, reason = "reassessment_replacement", "agreement state transition" ); @@ -594,7 +594,7 @@ where agreement_id = %id, indexing_request_id = %indexing_request_id, old_status = "none", - new_status = "CREATED", + new_status = %IndexingAgreementStatus::Created, reason = "reassessment_addition", "agreement state transition" ); @@ -674,7 +674,7 @@ where agreement_id = %old_agreement.id, indexing_request_id = %indexing_request_id, old_status = %old_agreement.status, - new_status, + new_status = %new_status, reason = "reassessment_not_in_target_group", "agreement state transition" ); @@ -958,6 +958,15 @@ mod lifecycle_event_tests { unimplemented!("not exercised by reassess handler") } + async fn cancel_rejected_agreement_on_chain( + &self, + _agreement_id: IndexingAgreementId, + _accept: Option, + _priority: crate::worker::queue::JobPriority, + ) -> anyhow::Result { + unimplemented!() + } + async fn submit_offer( &self, _agreement_id: IndexingAgreementId, @@ -2006,8 +2015,8 @@ mod lifecycle_event_tests { /// What became of an old agreement taken out of the target group. #[derive(Debug, PartialEq, Eq)] enum Unpaired { - /// Moved to the status named. - Moved(&'static str), + /// Moved to this status. + Moved(IndexingAgreementStatus), /// Already ended or being cancelled, as a race with another path can leave it. AlreadyEnding, /// Couldn't be marked; logged. @@ -2033,7 +2042,7 @@ where .mark_indexing_agreement_as_canceled_by_requester(&agreement.id) .await { - Ok(()) => Unpaired::Moved("CANCELED_BY_REQUESTER"), + Ok(()) => Unpaired::Moved(IndexingAgreementStatus::CanceledByRequester), Err(err) => unmarked(agreement, &err), }; } @@ -2047,12 +2056,12 @@ where .await { Ok(crate::cancel_dispatch::CancelStarted::Ended) => { - Unpaired::Moved("CANCELED_BY_REQUESTER") + Unpaired::Moved(IndexingAgreementStatus::CanceledByRequester) } Ok( crate::cancel_dispatch::CancelStarted::NotLive | crate::cancel_dispatch::CancelStarted::MayBeLive, - ) => Unpaired::Moved("CANCELLING"), + ) => Unpaired::Moved(IndexingAgreementStatus::Cancelling), Err(err) => unmarked(agreement, &err), } } 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 e3e6fbc4..3825f91a 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 @@ -190,8 +190,8 @@ where tracing::info!( agreement_id = %agreement_id, indexing_request_id = %indexing_request_id, - old_status = "CREATED", - new_status = "REJECTED", + old_status = %IndexingAgreementStatus::Created, + new_status = %IndexingAgreementStatus::Rejected, reason = %format_args!("rejected_{rejection_reason_str}"), detail = %rejected.detail, "agreement state transition" @@ -232,8 +232,8 @@ where tracing::info!( agreement_id = %agreement_id, indexing_request_id = %indexing_request_id, - old_status = "CREATED", - new_status = "UNRESPONSIVE", + old_status = %IndexingAgreementStatus::Created, + new_status = %IndexingAgreementStatus::Unresponsive, reason = "unresponsive", "agreement state transition" ); @@ -729,6 +729,15 @@ mod tests { Ok(JobId::default()) } + async fn cancel_rejected_agreement_on_chain( + &self, + _agreement_id: IndexingAgreementId, + _accept: Option, + _priority: JobPriority, + ) -> anyhow::Result { + unimplemented!() + } + async fn submit_offer( &self, _agreement_id: IndexingAgreementId, diff --git a/bin/dipper-service/src/worker/messages.rs b/bin/dipper-service/src/worker/messages.rs index 479d25b4..f279f27d 100644 --- a/bin/dipper-service/src/worker/messages.rs +++ b/bin/dipper-service/src/worker/messages.rs @@ -22,12 +22,8 @@ pub enum Message { /// See [`SendIndexingAgreementProposal`] for more details. SendIndexingAgreementProposal(SendIndexingAgreementProposal), - /// Cancel a rejected agreement on-chain. - /// - /// When an indexer rejects an agreement off-chain but later accepts it on-chain, - /// this message triggers on-chain cancellation via `cancelIndexingAgreementByPayer`. - /// - /// See [`CancelRejectedAgreementOnChain`] for more details. + /// Check on-chain an agreement dipper rejected or cancelled that the chain listener saw + /// accepted. See [`CancelRejectedAgreementOnChain`] for more details. CancelRejectedAgreementOnChain(CancelRejectedAgreementOnChain), /// Submit an RCA offer on-chain after the indexer has accepted the diff --git a/bin/dipper-service/src/worker/service.rs b/bin/dipper-service/src/worker/service.rs index 659ca90c..1d97edfd 100644 --- a/bin/dipper-service/src/worker/service.rs +++ b/bin/dipper-service/src/worker/service.rs @@ -20,6 +20,7 @@ use super::{ result::{JobError, JobResult, calculate_backoff_delay, retries_within_window}, }; pub use super::{ + handlers::ChainAccept, queue::JobPriority, service_queue::{WorkerQueue, WorkerQueueHandle}, }; diff --git a/bin/dipper-service/src/worker/service_queue.rs b/bin/dipper-service/src/worker/service_queue.rs index 49ad37e1..d86b36b5 100644 --- a/bin/dipper-service/src/worker/service_queue.rs +++ b/bin/dipper-service/src/worker/service_queue.rs @@ -4,7 +4,10 @@ use thegraph_core::{DeploymentId, alloy::primitives::ChainId}; use url::Url; use super::{ - handlers::{ReassessIndexingRequest, SendIndexingAgreementProposal, SubmitOffer}, + handlers::{ + CancelRejectedAgreementOnChain, ChainAccept, ReassessIndexingRequest, + SendIndexingAgreementProposal, SubmitOffer, + }, messages::Message, queue::{JobId, JobPriority, Queue}, }; @@ -30,6 +33,15 @@ pub trait WorkerQueue { priority: JobPriority, ) -> anyhow::Result; + /// Check on-chain an agreement dipper rejected or cancelled that was accepted anyway, and + /// move it back into `Cancelling` if it is live, recording `accept`. + async fn cancel_rejected_agreement_on_chain( + &self, + agreement_id: IndexingAgreementId, + accept: Option, + priority: JobPriority, + ) -> anyhow::Result; + /// Submit an RCA offer on-chain as the first step of a new proposal. The /// job retries until the indexer's window to accept has closed, so a /// provider outage costs one offer only if it outlasts that window. @@ -112,6 +124,23 @@ where .await } + async fn cancel_rejected_agreement_on_chain( + &self, + agreement_id: IndexingAgreementId, + accept: Option, + priority: JobPriority, + ) -> anyhow::Result { + self.queue + .push( + Message::CancelRejectedAgreementOnChain(CancelRejectedAgreementOnChain { + agreement_id, + accept, + }), + priority, + ) + .await + } + async fn submit_offer( &self, agreement_id: IndexingAgreementId, diff --git a/dipper-pgregistry/src/lib.rs b/dipper-pgregistry/src/lib.rs index 8e0ea869..c78ed6c0 100644 --- a/dipper-pgregistry/src/lib.rs +++ b/dipper-pgregistry/src/lib.rs @@ -20,9 +20,9 @@ pub use indexing_request::{ Status as IndexingRequestStatus, }; pub use postgres::{ - CancelKind, CancellingAgreement, ChainListenerStateRow, NewAgreementParams, + CancelKind, CancellingAgreement, ChainListenerStateRow, EndCorrection, NewAgreementParams, PendingAcceptedEvent, PendingExpiredEvent, PendingTerminatedEvent, PgRegistry, - ReconciliationAudit, ReconciliationItem, ReconciliationOutcome, + ReconciliationAudit, ReconciliationItem, ReconciliationOutcome, RecordedEnd, }; pub use result::{Error, Result}; diff --git a/dipper-pgregistry/src/postgres.rs b/dipper-pgregistry/src/postgres.rs index e25800ea..3a36f0bf 100644 --- a/dipper-pgregistry/src/postgres.rs +++ b/dipper-pgregistry/src/postgres.rs @@ -95,6 +95,46 @@ pub struct ReconciliationAudit { pub canceled_tx: Option, } +/// An agreement's end as recorded: when, in chain seconds, who ended it, and in which transaction. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct RecordedEnd { + pub canceled_at: Option, + pub canceled_by: Option, + pub canceled_tx: Option, +} + +/// An announced end that a write replaced or cleared, so its `terminated` event goes out again. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct EndCorrection { + pub old: RecordedEnd, + pub new: RecordedEnd, +} + +/// What a write that can replace an end returns: whether it cleared that end's announcement, then +/// the end before and after it. The end before is read under a row lock, so it is the one replaced. +type EndChangeRow = ( + bool, + Option, + Option, + Option, + Option, + Option, + Option, +); + +fn end_correction(row: EndChangeRow) -> Option { + let (announcement_cleared, old_at, old_by, old_tx, new_at, new_by, new_tx) = row; + let end = |at: Option, canceled_by, canceled_tx| RecordedEnd { + canceled_at: at.and_then(|at| u64::try_from(at).ok()), + canceled_by, + canceled_tx, + }; + announcement_cleared.then(|| EndCorrection { + old: end(old_at, old_by, old_tx), + new: end(new_at, new_by, new_tx), + }) +} + /// A row that needs a `terminated` lifecycle event emitted. Sourced entirely /// from the agreement row so the emission sweep is a pure DB -> broker drain. #[derive(Debug, Clone)] @@ -950,7 +990,7 @@ impl PgRegistry { Ok(()) } - /// One being cancelled because its indexer stopped serving it ends `AbandonedByIndexer`. + /// End an agreement dipper no longer wants `CanceledByRequester`. pub async fn mark_indexing_agreement_as_canceled_by_requester( &self, agreement_id: &IndexingAgreementId, @@ -963,6 +1003,20 @@ impl PgRegistry { .await } + /// End an agreement dipper was cancelling because its indexer stopped serving it, once the + /// chain confirms dipper's cancel. + pub async fn mark_indexing_agreement_as_abandoned_by_indexer( + &self, + agreement_id: &IndexingAgreementId, + ) -> Result<(), Error> { + self.set_status_from( + agreement_id, + IndexingAgreementStatus::AbandonedByIndexer, + &[IndexingAgreementStatus::Cancelling], + ) + .await + } + /// Mark an agreement that may be live on-chain `Cancelling`, before dipper sends its /// on-chain cancel. One marked `Expired` may have been accepted unseen by a lagging listener. pub async fn mark_indexing_agreement_as_cancelling( @@ -1015,13 +1069,14 @@ impl PgRegistry { /// as checked, since no cancel is sent with it, so the retry takes it on its next sweep rather /// than waiting for one to be mined. When the chain was read and showed it live (`seen_live`), /// the end on record, and any announcement of it, no longer stands, so both are cleared for - /// the end still to come; an unread chain leaves them, as the agreement may have ended. + /// the end still to come, and the end returned if it was announced; an unread chain leaves + /// them, as the agreement may have ended. pub async fn reopen_indexing_agreement_cancel( &self, agreement_id: &IndexingAgreementId, seen_live: bool, - ) -> Result<(), Error> { - let updated = sqlx::query( + ) -> Result, Error> { + let row: Option = sqlx::query_as( r#" UPDATE dipper_reg_indexing_agreements SET @@ -1035,7 +1090,14 @@ impl PgRegistry { terminated_event_emitted_at = CASE WHEN $5 THEN NULL ELSE terminated_event_emitted_at END, updated_at = timezone('UTC', now()) + FROM ( + SELECT canceled_at AS old_at, canceled_by AS old_by, canceled_tx AS old_tx, + terminated_event_emitted_at IS NOT NULL AS was_announced + FROM dipper_reg_indexing_agreements WHERE id = $2 FOR UPDATE + ) AS before WHERE id = $2 AND status IN ($3, $4) + RETURNING was_announced AND terminated_event_emitted_at IS NULL, + old_at, old_by, old_tx, canceled_at, canceled_by, canceled_tx "#, ) .bind(IndexingAgreementStatus::Cancelling) @@ -1043,12 +1105,9 @@ impl PgRegistry { .bind(IndexingAgreementStatus::CanceledByRequester) .bind(IndexingAgreementStatus::Rejected) .bind(seen_live) - .execute(&self.pool) + .fetch_optional(&self.pool) .await?; - if updated.rows_affected() == 0 { - return Err(Error::NoRecordsUpdated); - } - Ok(()) + row.map(end_correction).ok_or(Error::NoRecordsUpdated) } /// `Cancelling` agreements, those checked longest ago first, leaving out any never checked @@ -1212,18 +1271,7 @@ impl PgRegistry { let mut did_cancel = false; if let Some(kind) = cancel { - let (new_status, allowed_from): (_, &[IndexingAgreementStatus]) = match kind { - CancelKind::ByRequester => ( - IndexingAgreementStatus::CanceledByRequester, - CANCEL_BY_REQUESTER_FROM, - ), - CancelKind::ByIndexer => ( - IndexingAgreementStatus::CanceledByIndexer, - CANCEL_BY_INDEXER_FROM, - ), - }; - did_cancel = - update_status_from(&mut tx, agreement_id, new_status, allowed_from).await?; + did_cancel = apply_cancel(&mut tx, agreement_id, kind).await?; if did_accept && !did_cancel { return Err(Error::NoRecordsUpdated); } @@ -1248,7 +1296,6 @@ impl PgRegistry { /// stale outcome flags from the first iteration; debug_asserted. #[expect( clippy::expect_used, - clippy::too_many_lines, reason = "predates this lint; fix when next touched" )] pub async fn apply_reconciliation_batch( @@ -1306,18 +1353,7 @@ impl PgRegistry { ], ) .await?; - let (new_status, allowed_from): (_, &[IndexingAgreementStatus]) = match cancel_kind { - CancelKind::ByRequester => ( - IndexingAgreementStatus::CanceledByRequester, - CANCEL_BY_REQUESTER_FROM, - ), - CancelKind::ByIndexer => ( - IndexingAgreementStatus::CanceledByIndexer, - CANCEL_BY_INDEXER_FROM, - ), - }; - let did_cancel = - update_status_from(&mut tx, &item.agreement_id, new_status, allowed_from).await?; + let did_cancel = apply_cancel(&mut tx, &item.agreement_id, cancel_kind).await?; if did_accept && !did_cancel { return Err(Error::NoRecordsUpdated); } @@ -1344,25 +1380,12 @@ impl PgRegistry { outcomes.entry(id).or_default().did_accept = true; } - for id in batch_update_status_from( - &mut tx, - &cancel_by_requester, - IndexingAgreementStatus::CanceledByRequester, - CANCEL_BY_REQUESTER_FROM, - ) - .await? + for id in batch_apply_cancel(&mut tx, &cancel_by_requester, CancelKind::ByRequester).await? { outcomes.entry(id).or_default().did_cancel = true; } - for id in batch_update_status_from( - &mut tx, - &cancel_by_indexer, - IndexingAgreementStatus::CanceledByIndexer, - CANCEL_BY_INDEXER_FROM, - ) - .await? - { + for id in batch_apply_cancel(&mut tx, &cancel_by_indexer, CancelKind::ByIndexer).await? { outcomes.entry(id).or_default().did_cancel = true; } @@ -1558,13 +1581,9 @@ impl PgRegistry { Ok(()) } - /// Record the cancel audit payload for a dipper-initiated cancel so the - /// emission sweep can populate the `terminated` event's tx/by/at fields. - /// `COALESCE` keeps any value already observed on-chain. Best-effort - /// enrichment: the event still emits (with fallbacks) if never recorded. - /// An end recorded before the agreement's accept, such as its offer's withdrawal before the - /// offer landed after all, can't be its end, so a later one replaces it and is announced. - /// That is decided once per row, so the end's 4 fields always come from the same end. + /// Record an agreement's end for its `terminated` event. A new end from the accept on replaces + /// the end on record whole, to be announced again, if that came before the accept, so can't + /// be the end, or is dipper's estimate, with no transaction, and the new one has it. #[expect( clippy::cast_possible_wrap, reason = "predates this lint; fix when next touched" @@ -1575,8 +1594,8 @@ impl PgRegistry { canceled_at: u64, canceled_by: &str, canceled_tx: Option<&str>, - ) -> Result<(), Error> { - sqlx::query( + ) -> Result, Error> { + let row: Option = sqlx::query_as( r#" UPDATE dipper_reg_indexing_agreements AS agreement SET (canceled_at, canceled_by, canceled_tx, terminated_event_emitted_at) = ( @@ -1586,20 +1605,29 @@ impl PgRegistry { CASE WHEN replace_end THEN $4 ELSE COALESCE(agreement.canceled_tx, $4) END, CASE WHEN replace_end THEN NULL ELSE agreement.terminated_event_emitted_at END FROM ( - SELECT agreement.canceled_at < agreement.accepted_at - AND $2 >= agreement.accepted_at AS replace_end + SELECT $2 >= agreement.accepted_at + AND (agreement.canceled_at < agreement.accepted_at + OR (agreement.canceled_tx IS NULL AND $4 IS NOT NULL)) AS replace_end ) AS end_rule ) + FROM ( + SELECT canceled_at AS old_at, canceled_by AS old_by, canceled_tx AS old_tx, + terminated_event_emitted_at IS NOT NULL AS was_announced + FROM dipper_reg_indexing_agreements WHERE id = $1 FOR UPDATE + ) AS before WHERE agreement.id = $1 + RETURNING was_announced AND agreement.terminated_event_emitted_at IS NULL, + old_at, old_by, old_tx, + agreement.canceled_at, agreement.canceled_by, agreement.canceled_tx "#, ) .bind(agreement_id) .bind(canceled_at as i64) .bind(canceled_by) .bind(canceled_tx) - .execute(&self.pool) + .fetch_optional(&self.pool) .await?; - Ok(()) + Ok(row.and_then(end_correction)) } /// Record an agreement's accept and its end together, as the chain shows them, in 1 write, @@ -1617,8 +1645,8 @@ impl PgRegistry { canceled_at: u64, canceled_by: &str, canceled_tx: Option<&str>, - ) -> Result<(), Error> { - sqlx::query( + ) -> Result, Error> { + let row: Option = sqlx::query_as( r#" UPDATE dipper_reg_indexing_agreements AS agreement SET accepted_at = COALESCE(agreement.accepted_at, $2), @@ -1631,11 +1659,21 @@ impl PgRegistry { CASE WHEN replace_end THEN NULL ELSE agreement.terminated_event_emitted_at END FROM ( - SELECT agreement.canceled_at < COALESCE(agreement.accepted_at, $2) - AND $4 >= COALESCE(agreement.accepted_at, $2) AS replace_end + SELECT $4 >= COALESCE(agreement.accepted_at, $2) + AND (agreement.canceled_at < COALESCE(agreement.accepted_at, $2) + OR (agreement.canceled_tx IS NULL AND $6 IS NOT NULL)) + AS replace_end ) AS end_rule ) + FROM ( + SELECT canceled_at AS old_at, canceled_by AS old_by, canceled_tx AS old_tx, + terminated_event_emitted_at IS NOT NULL AS was_announced + FROM dipper_reg_indexing_agreements WHERE id = $1 FOR UPDATE + ) AS before WHERE agreement.id = $1 + RETURNING was_announced AND agreement.terminated_event_emitted_at IS NULL, + old_at, old_by, old_tx, + agreement.canceled_at, agreement.canceled_by, agreement.canceled_tx "#, ) .bind(agreement_id) @@ -1644,9 +1682,9 @@ impl PgRegistry { .bind(canceled_at as i64) .bind(canceled_by) .bind(canceled_tx) - .execute(&self.pool) + .fetch_optional(&self.pool) .await?; - Ok(()) + Ok(row.and_then(end_correction)) } // ========================================================================= @@ -2453,41 +2491,84 @@ impl PgRegistry { } } -/// Batched form of `update_status_from`: transitions all rows whose `id` -/// is in `agreement_ids` and whose current status is in `allowed_from` to -/// `new_status`, in one statement; a row noted abandoned that `new_status` would make -/// `CanceledByRequester` becomes `AbandonedByIndexer` instead. Returns the ids of the rows that -/// actually flipped (matched the CAS guard) so callers can build per-id -/// outcome maps. Empty input is a fast-path no-op. +/// Batched form of `update_status_from`: moves every row in `agreement_ids` whose status is in +/// `allowed_from` to `new_status`, in 1 statement. Returns the ids of the rows that moved, so +/// callers can build per-id outcome maps; empty input is a no-op. async fn batch_update_status_from( tx: &mut sqlx::Transaction<'_, Postgres>, agreement_ids: &[IndexingAgreementId], new_status: IndexingAgreementStatus, allowed_from: &[IndexingAgreementStatus], +) -> Result, Error> { + batch_set_status(tx, agreement_ids, "$1", &[new_status], allowed_from).await +} + +/// Record a cancel by `kind`, seen on the chain, in 1 row; true if it ended the row. +async fn apply_cancel( + tx: &mut sqlx::Transaction<'_, Postgres>, + agreement_id: &IndexingAgreementId, + kind: CancelKind, +) -> Result { + let ended = batch_apply_cancel(tx, std::slice::from_ref(agreement_id), kind).await?; + Ok(!ended.is_empty()) +} + +/// Record a cancel by `kind`, seen on the chain, in each row it ends, returning their ids. The +/// chain doesn't say why dipper cancelled, so its cancel ends a row noted abandoned (its indexer +/// stopped serving it) `AbandonedByIndexer`, and the rest `CanceledByRequester`. +async fn batch_apply_cancel( + tx: &mut sqlx::Transaction<'_, Postgres>, + agreement_ids: &[IndexingAgreementId], + kind: CancelKind, +) -> Result, Error> { + match kind { + CancelKind::ByRequester => { + let ended = [ + IndexingAgreementStatus::AbandonedByIndexer, + IndexingAgreementStatus::CanceledByRequester, + ]; + let status = "CASE WHEN abandoned THEN $1 ELSE $2 END"; + batch_set_status(tx, agreement_ids, status, &ended, CANCEL_BY_REQUESTER_FROM).await + } + CancelKind::ByIndexer => { + let ended = IndexingAgreementStatus::CanceledByIndexer; + batch_update_status_from(tx, agreement_ids, ended, CANCEL_BY_INDEXER_FROM).await + } + } +} + +/// Set the status of the rows in `agreement_ids` whose status is in `allowed_from` to +/// `status_sql`, which reads `new_statuses` as `$1`, `$2` and so on, returning their ids. +async fn batch_set_status( + tx: &mut sqlx::Transaction<'_, Postgres>, + agreement_ids: &[IndexingAgreementId], + status_sql: &str, + new_statuses: &[IndexingAgreementStatus], + allowed_from: &[IndexingAgreementStatus], ) -> Result, Error> { if agreement_ids.is_empty() { return Ok(Vec::new()); } + let ids_at = new_statuses.len() + 1; let placeholders = (0..allowed_from.len()) - .map(|i| format!("${}", i + 5)) + .map(|i| format!("${}", ids_at + 1 + i)) .collect::>() .join(", "); - // An agreement dipper ended because its indexer stopped serving it ends as abandoned. let sql = format!( r#" UPDATE dipper_reg_indexing_agreements SET - status = CASE WHEN abandoned AND $1 = $3 THEN $4 ELSE $1 END, + status = {status_sql}, updated_at = timezone('UTC', now()) - WHERE id = ANY($2) AND status IN ({placeholders}) + WHERE id = ANY(${ids_at}) AND status IN ({placeholders}) RETURNING id "# ); - let mut query = sqlx::query_as::<_, (IndexingAgreementId,)>(&sql) - .bind(new_status) - .bind(agreement_ids) - .bind(IndexingAgreementStatus::CanceledByRequester) - .bind(IndexingAgreementStatus::AbandonedByIndexer); + let mut query = sqlx::query_as::<_, (IndexingAgreementId,)>(&sql); + for status in new_statuses { + query = query.bind(*status); + } + query = query.bind(agreement_ids); for status in allowed_from { query = query.bind(*status); } @@ -2495,8 +2576,6 @@ async fn batch_update_status_from( Ok(rows.into_iter().map(|(id,)| id).collect()) } -/// Transition an agreement's status inside a transaction. Thin wrapper -/// over `batch_update_status_from` for the single-row case. /// Persist the accepted-transition audit payload for a single row. `COALESCE` /// keeps any already-stored value rather than clobbering it with `NULL`. #[expect( @@ -2559,6 +2638,8 @@ async fn persist_cancel_audit( Ok(()) } +/// Move an agreement whose status is in `allowed_from` to `new_status` inside a transaction, +/// returning whether it moved. Thin wrapper over `batch_update_status_from` for 1 row. async fn update_status_from( tx: &mut sqlx::Transaction<'_, Postgres>, agreement_id: &IndexingAgreementId, @@ -2574,3 +2655,40 @@ async fn update_status_from( .await? .is_empty()) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn an_end_counts_as_corrected_only_when_its_announcement_was_cleared() { + let row = |cleared| { + ( + cleared, + Some(9_000), + Some("0xindexer".to_owned()), + None, + Some(5_000), + Some("0xprovider".to_owned()), + Some("0xend".to_owned()), + ) + }; + + assert_eq!(end_correction(row(false)), None); + assert_eq!( + end_correction(row(true)), + Some(EndCorrection { + old: RecordedEnd { + canceled_at: Some(9_000), + canceled_by: Some("0xindexer".to_owned()), + canceled_tx: None, + }, + new: RecordedEnd { + canceled_at: Some(5_000), + canceled_by: Some("0xprovider".to_owned()), + canceled_tx: Some("0xend".to_owned()), + }, + }) + ); + } +} diff --git a/dipper-pgregistry/tests/it_registry_postgres.rs b/dipper-pgregistry/tests/it_registry_postgres.rs index d1f0a006..a09aa096 100644 --- a/dipper-pgregistry/tests/it_registry_postgres.rs +++ b/dipper-pgregistry/tests/it_registry_postgres.rs @@ -4,8 +4,9 @@ use std::collections::HashSet; use dipper_core::ids::{IndexingAgreementId, IndexingRequestId}; use dipper_pgregistry::{ - CancelKind, Error, IndexingAgreementStatus, IndexingAgreementTerms, IndexingRequestStatus, - NewAgreementParams, PgRegistry, ReconciliationAudit, ReconciliationItem, + CancelKind, EndCorrection, Error, IndexingAgreementStatus, IndexingAgreementTerms, + IndexingRequestStatus, NewAgreementParams, PgRegistry, ReconciliationAudit, ReconciliationItem, + RecordedEnd, }; use fake::{Fake, Faker}; use pgtemp::PgTempDB; @@ -697,7 +698,7 @@ async fn get_declined_indexers_by_deployment_includes_an_indexer_that_abandoned_ .await .expect("abandon"); registry - .mark_indexing_agreement_as_canceled_by_requester(&agreement_id) + .mark_indexing_agreement_as_abandoned_by_indexer(&agreement_id) .await .expect("ended abandoned"); @@ -1905,7 +1906,7 @@ where async fn an_abandoned_agreement_dipper_cancels_ends_abandoned() { let (again, status) = abandon_then_end(async |registry, id| { registry - .mark_indexing_agreement_as_canceled_by_requester(id) + .mark_indexing_agreement_as_abandoned_by_indexer(id) .await .expect("dipper's cancel confirmed"); }) @@ -1918,6 +1919,40 @@ async fn an_abandoned_agreement_dipper_cancels_ends_abandoned() { ); } +#[tokio::test] +async fn the_requester_mark_writes_its_own_status_whatever_the_abandoned_flag() { + // The caller picks the end status, so it logs the one written. + let (_, status) = abandon_then_end(async |registry, id| { + registry + .mark_indexing_agreement_as_canceled_by_requester(id) + .await + .expect("marked"); + }) + .await; + + assert_eq!(status, IndexingAgreementStatus::CanceledByRequester); +} + +#[tokio::test] +async fn an_abandoned_agreement_the_listener_sees_cancelled_in_a_batch_ends_abandoned() { + let (_, status) = abandon_then_end(async |registry, id| { + let items = [ReconciliationItem { + agreement_id: *id, + apply_accept: false, + cancel: Some(CancelKind::ByRequester), + audit: ReconciliationAudit::default(), + }]; + let outcomes = registry + .apply_reconciliation_batch(&items) + .await + .expect("reconciliation"); + assert!(outcomes[id].did_cancel); + }) + .await; + + assert_eq!(status, IndexingAgreementStatus::AbandonedByIndexer); +} + #[tokio::test] async fn an_abandoned_agreement_the_listener_sees_cancelled_ends_abandoned() { let (_, status) = abandon_then_end(async |registry, id| { @@ -3483,7 +3518,7 @@ async fn count_created_agreements_by_indexer_counts_offers_being_withdrawn() { .duration_since(std::time::UNIX_EPOCH) .unwrap() .as_secs(); - // 3 agreements being cancelled: an offer still open, one accepted, one past its deadline. + // 3 agreements being cancelled: an offer still open, 1 accepted, 1 past its deadline. let mut ids = Vec::new(); for (deployment_id, deadline) in [ ( @@ -3975,6 +4010,117 @@ async fn an_end_recorded_before_the_accept_gives_way_to_the_real_one() { ); } +/// Each write that can change an agreement's end reports a change to an end already announced, +/// whose `terminated` event then goes out again, and nothing otherwise. +#[tokio::test] +async fn a_change_to_an_announced_end_is_reported() { + 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"); + let registry = PgRegistry::new(db.clone()); + let announce = async |id: IndexingAgreementId| { + registry + .mark_terminated_event_emitted(&id) + .await + .expect("announce"); + }; + let end = |at: u64, by: &str, tx: &str| RecordedEnd { + canceled_at: Some(at), + canceled_by: Some(by.to_owned()), + canceled_tx: Some(tx.to_owned()), + }; + + // An offer withdrawn and announced, then accepted after all, so its real end replaces it. + let id = fixture_agreement(0xaa); + let first = registry + .record_cancel_audit(&id, 1_000, "0xwithdrawer", Some("0xwithdrawal")) + .await + .expect("cancel record"); + assert_eq!(first, None, "nothing was announced yet"); + announce(id).await; + registry + .record_accepted_audit(&id, 2_000, "0xaccept") + .await + .expect("accept record"); + let replaced = registry + .record_cancel_audit(&id, 3_000, "0xpayer", Some("0xcancel")) + .await + .expect("cancel record"); + assert_eq!( + replaced, + Some(EndCorrection { + old: end(1_000, "0xwithdrawer", "0xwithdrawal"), + new: end(3_000, "0xpayer", "0xcancel"), + }) + ); + announce(id).await; + let kept = registry + .record_cancel_audit(&id, 4_000, "0xlater", Some("0xlater")) + .await + .expect("cancel record"); + assert_eq!(kept, None, "an end kept is no change"); + + // Found live on-chain after its announced end, which no longer stands. + registry + .mark_indexing_agreement_as_canceled_by_requester(&id) + .await + .expect("mark ended"); + let unread = registry + .reopen_indexing_agreement_cancel(&id, false) + .await + .expect("reopen"); + assert_eq!( + unread, None, + "an unread chain keeps the end and its announcement" + ); + registry + .mark_indexing_agreement_as_canceled_by_requester(&id) + .await + .expect("mark ended"); + let reopened = registry + .reopen_indexing_agreement_cancel(&id, true) + .await + .expect("reopen"); + assert_eq!( + reopened, + Some(EndCorrection { + old: end(3_000, "0xpayer", "0xcancel"), + new: RecordedEnd::default(), + }) + ); + + // The chain's accept and end, read together, replace an announced withdrawal the same way. + let other = + IndexingAgreementId::from_bytes([0xaa, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2]); + sqlx::query( + "UPDATE dipper_reg_indexing_agreements SET canceled_at = 50, canceled_by = '0xwd', \ + canceled_tx = '0xwd', terminated_event_emitted_at = now() WHERE id = $1", + ) + .bind(other) + .execute(&db) + .await + .expect("withdraw"); + let read = async || { + registry + .record_accept_and_cancel_audit(&other, 100, "0xacc", 200, "0xpayer", Some("0xend")) + .await + .expect("record") + }; + assert_eq!( + read().await, + Some(EndCorrection { + old: end(50, "0xwd", "0xwd"), + new: end(200, "0xpayer", "0xend"), + }) + ); + announce(other).await; + assert_eq!(read().await, None, "a replay changes nothing"); +} + #[tokio::test] async fn a_cancelling_agreement_stays_live_and_unannounced_until_it_ends() { let (db, _temp_db) = temp_registry_db().await; @@ -4134,7 +4280,7 @@ async fn an_abandoned_agreement_awaits_replacement_once_ended_until_queued() { ); registry - .mark_indexing_agreement_as_canceled_by_requester(&id) + .mark_indexing_agreement_as_abandoned_by_indexer(&id) .await .expect("end it"); assert_eq!(awaiting().await, vec![id]); @@ -4218,6 +4364,108 @@ async fn an_accept_and_end_from_the_chain_are_recorded_in_1_write() { ); } +/// An end dipper records itself when the chain listener is slow, with no transaction, is an +/// estimate: the chain's end replaces it whole, never mixed with it, and is announced again. +#[tokio::test] +async fn an_estimated_end_gives_way_whole_to_the_chains() { + 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"); + let registry = PgRegistry::new(db.clone()); + // The end on record, and whether it is yet to be announced. + let end = async |id: IndexingAgreementId| { + sqlx::query_as::<_, (Option, Option, Option, bool)>( + "SELECT canceled_at, canceled_by, canceled_tx, terminated_event_emitted_at IS NULL \ + FROM dipper_reg_indexing_agreements WHERE id = $1", + ) + .bind(id) + .fetch_one(&db) + .await + .expect("cancel record query") + }; + let chains = |unannounced: bool| { + ( + Some(5_000), + Some("0xprovider".to_owned()), + Some("0xend".to_owned()), + unannounced, + ) + }; + + let id = fixture_agreement(0xaa); + registry + .mark_indexing_agreement_as_cancelling(&id) + .await + .expect("mark cancelling"); + registry + .record_accepted_audit(&id, 1_000, "0xacc") + .await + .expect("accept record"); + registry + .record_cancel_audit(&id, 9_000, "0xindexer", None) + .await + .expect("estimated end"); + let marked = registry + .apply_reconciliation(&id, false, Some(CancelKind::ByIndexer)) + .await + .expect("mark ended by the indexer"); + assert!(marked.did_cancel); + // Its terminated event waited an hour for a transaction, then went out with the estimate. + registry + .mark_terminated_event_emitted(&id) + .await + .expect("announce"); + + let read_from_chain = async || { + registry + .record_accept_and_cancel_audit(&id, 1_000, "0xacc", 5_000, "0xprovider", Some("0xend")) + .await + .expect("the chain's end"); + }; + read_from_chain().await; + assert_eq!( + end(id).await, + chains(true), + "the chain's end replaces the estimate whole, to be announced again" + ); + registry + .mark_terminated_event_emitted(&id) + .await + .expect("announce again"); + read_from_chain().await; + assert_eq!(end(id).await, chains(false), "a replay changes nothing"); + registry + .record_cancel_audit(&id, 9_500, "0xindexer", None) + .await + .expect("another estimate"); + assert_eq!( + end(id).await, + chains(false), + "an estimate never replaces the chain's end" + ); + + // The other writer of an end follows the same rule. + let other = + IndexingAgreementId::from_bytes([0xaa, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2]); + registry + .record_accepted_audit(&other, 1_000, "0xacc") + .await + .expect("accept record"); + registry + .record_cancel_audit(&other, 9_000, "0xindexer", None) + .await + .expect("estimated end"); + registry + .record_cancel_audit(&other, 5_000, "0xprovider", Some("0xend")) + .await + .expect("end with its transaction"); + assert_eq!(end(other).await, chains(true)); +} + /// An ended agreement's `terminated` event waits for the transaction that ended it, which the /// chain listener can record after dipper marks it ended, but not for ever. #[tokio::test]