diff --git a/bin/dipper-service/src/cancel_dispatch.rs b/bin/dipper-service/src/cancel_dispatch.rs index cedd4326..c6e40059 100644 --- a/bin/dipper-service/src/cancel_dispatch.rs +++ b/bin/dipper-service/src/cancel_dispatch.rs @@ -119,32 +119,56 @@ where R: AgreementRegistry + Sync, T: ChainClient, { + mark_cancelling(registry, agreement, reason).await?; + Ok(send_marked_cancel(registry, chain_client, agreement, reason, config).await) +} + +/// The first half of [`start_cancel`]: mark the agreement `Cancelling`, or abandoning. +pub async fn mark_cancelling( + registry: &R, + agreement: &IndexingAgreement, + reason: CancelReason, +) -> RegistryResult<()> { match reason { CancelReason::NotWanted => { registry .mark_indexing_agreement_as_cancelling(&agreement.id) - .await? + .await } CancelReason::Abandoned => { registry .mark_indexing_agreement_as_abandoning(&agreement.id) - .await? + .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, + chain_client: &T, + agreement: &IndexingAgreement, + reason: CancelReason, + config: &IndexingAgreementConfig, +) -> CancelStarted +where + R: AgreementRegistry + Sync, + T: ChainClient, +{ let tx_hash = match cancel_if_live(chain_client, agreement, config).await { LiveCancel::Ended(tx_hash) => tx_hash, - LiveCancel::NotLive { .. } => return Ok(CancelStarted::NotLive), + LiveCancel::NotLive { .. } => return CancelStarted::NotLive, LiveCancel::ReadFailed(err) | LiveCancel::CancelFailed(err) => { tracing::warn!( agreement_id = %agreement.id, error = %err, "On-chain cancel failed; the cancel retry sends it again" ); - return Ok(CancelStarted::MayBeLive); + return CancelStarted::MayBeLive; } LiveCancel::Unconfirmed { tx_hash, err } => { log_unconfirmed(agreement, tx_hash, &err); - return Ok(CancelStarted::MayBeLive); + return CancelStarted::MayBeLive; } }; tracing::info!( @@ -154,15 +178,13 @@ where ); // An offer never accepted could still land and be accepted until its deadline. if agreement.status != IndexingAgreementStatus::AcceptedOnChain { - return Ok(CancelStarted::NotLive); + return CancelStarted::NotLive; + } + if confirm_cancelled(registry, agreement, reason, tx_hash, config).await { + CancelStarted::Ended + } else { + CancelStarted::NotLive } - Ok( - if confirm_cancelled(registry, agreement, reason, tx_hash, config).await { - CancelStarted::Ended - } else { - CancelStarted::NotLive - }, - ) } /// Mark an agreement the chain shows dipper ended as ended, recording the cancel when its diff --git a/bin/dipper-service/src/chain_client/client.rs b/bin/dipper-service/src/chain_client/client.rs index 9376c63c..842d5021 100644 --- a/bin/dipper-service/src/chain_client/client.rs +++ b/bin/dipper-service/src/chain_client/client.rs @@ -92,12 +92,26 @@ const CROSS_CHECK_DEADLINE: Duration = Duration::from_secs(3); /// 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. #[derive(Debug)] struct SeenBlock { number: u64, moved_at: Instant, confirmed: bool, cross_checked_at: Option, + /// An endpoint reported a block further ahead than 1 endpoint alone may move it, so the + /// endpoints are asked whether they agree. + jump_reported: bool, +} + +/// What became of a block 1 endpoint reported. +#[derive(Debug, PartialEq, Eq)] +enum Noted { + Taken, + /// Too far ahead for 1 endpoint alone; it waits for 2 to agree. + Unagreed, + /// Too far ahead to be real. + TooFarAhead, } impl SeenBlock { @@ -107,6 +121,7 @@ impl SeenBlock { moved_at: Instant::now(), confirmed: false, cross_checked_at: None, + jump_reported: false, } } @@ -130,24 +145,47 @@ impl SeenBlock { (self.number, highest) } - /// Take `block` as seen, unless it is too far ahead of a confirmed one to be real; false - /// when it is. - fn advance(&mut self, block: u64, now: Instant) -> bool { - if self.confirmed && block > self.believable_limit(now) { - return false; + /// The highest block 1 endpoint alone may move a confirmed newest block to: an hour 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(since.saturating_mul(BLOCKS_PER_SECOND)) + } + + /// Take `block`, reported by 1 endpoint, as seen. Once the newest block is confirmed, one + /// too far ahead to be real is refused, and one past [`Self::lone_limit`] waits for 2 + /// endpoints to agree, unless this is the `only_endpoint` there is to ask. + fn advance(&mut self, block: u64, now: Instant, only_endpoint: bool) -> Noted { + if self.confirmed { + if block > self.believable_limit(now) { + return Noted::TooFarAhead; + } + if !only_endpoint && block > self.lone_limit(now) { + self.jump_reported = true; + return Noted::Unagreed; + } } + self.take(block, now); + Noted::Taken + } + + fn take(&mut self, block: u64, now: Instant) { if block > self.number { self.number = block; self.moved_at = now; } - true } /// 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. fn confirm(&mut self, block: u64, now: Instant) { + self.jump_reported = false; if self.confirmed { - self.advance(block, now); + if block <= self.believable_limit(now) { + self.take(block, now); + } return; } if block > self.number || self.number > block.saturating_add(ALERT_GAP_BLOCKS) { @@ -157,13 +195,13 @@ impl SeenBlock { self.confirmed = true; } - /// Whether to ask every endpoint for its latest block: only while unconfirmed, and at most - /// once every [`CROSS_CHECK_INTERVAL`]. + /// Whether to ask every endpoint for its latest block: only while unconfirmed or after a + /// jump 1 endpoint reported, and at most once every [`CROSS_CHECK_INTERVAL`]. fn cross_check_due(&mut self, now: Instant) -> bool { let checked_lately = self .cross_checked_at .is_some_and(|at| now.saturating_duration_since(at) < CROSS_CHECK_INTERVAL); - if self.confirmed || checked_lately { + if (self.confirmed && !self.jump_reported) || checked_lately { return false; } self.cross_checked_at = Some(now); @@ -899,24 +937,37 @@ impl AlloyChainClient { 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 until they do" + against blocks too far ahead, nor the newest block seen moved far ahead, until \ + they do" ), } } /// 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. + /// be real is ignored, so it can't refuse every read after it, and one over an hour ahead + /// waits for 2 endpoints to agree on it. fn note_block(&self, block: u64) { - let (taken, seen) = { + let only_endpoint = self.inner.rpc_pool.endpoint_count() < 2; + let (noted, seen) = { let mut seen = self.seen_block(); - (seen.advance(block, Instant::now()), seen.number) + ( + seen.advance(block, Instant::now(), only_endpoint), + seen.number, + ) }; - if !taken { - tracing::warn!( + match noted { + Noted::Taken => {} + Noted::Unagreed => tracing::warn!( + block, + seen_block = seen, + "Not moving to a block over an hour ahead of the newest block dipper has seen \ + until 2 RPC endpoints agree on it" + ), + Noted::TooFarAhead => tracing::warn!( block, seen_block = seen, "Ignoring a block too far ahead of the newest block dipper has seen" - ); + ), } } @@ -2549,6 +2600,7 @@ mod tests { moved_at: now.checked_sub(Duration::from_secs(600)).expect("instant"), confirmed: true, cross_checked_at: None, + jump_reported: false, }; assert_eq!(seen.bounds(now), (100, 100 + WEEK_OF_BLOCKS + 2_400)); @@ -2590,8 +2642,8 @@ mod tests { // So a read just after dipper's cancel mined can't come from an endpoint behind it. let now = Instant::now(); let mut seen = SeenBlock::new(); - assert!(seen.advance(100, now)); - assert!(seen.advance(90, now)); + assert_eq!(seen.advance(100, now, false), Noted::Taken); + assert_eq!(seen.advance(90, now, false), Noted::Taken); assert_eq!(seen.bounds(now), (100, u64::MAX)); } @@ -2603,21 +2655,57 @@ mod tests { let now = Instant::now(); let mut seen = SeenBlock::new(); let real = 1_000 + WEEK_OF_BLOCKS * 3; - assert!(seen.advance(1_000, now)); - assert!(seen.advance(real, now), "taken while unconfirmed"); + assert_eq!(seen.advance(1_000, now, false), Noted::Taken); + assert_eq!( + seen.advance(real, now, false), + Noted::Taken, + "taken while unconfirmed" + ); seen.confirm(real, now); - assert!(!seen.advance(real + WEEK_OF_BLOCKS * 2, now)); + assert_eq!( + seen.advance(real + WEEK_OF_BLOCKS * 2, now, false), + Noted::TooFarAhead + ); assert_eq!(seen.bounds(now).0, real); } + #[test] + fn waits_for_2_endpoints_to_agree_before_a_jump_past_a_confirmed_block() { + // 1 faulty endpoint moving it far ahead would have every endpoint that is right + // refused as behind, for as long as the real chain takes to catch up. + let now = Instant::now(); + let mut seen = SeenBlock::new(); + let real = 5_000_000; + seen.confirm(real, now); + let jump = real + ALERT_GAP_BLOCKS + 1_000; + + assert_eq!(seen.advance(real + 10, now, false), Noted::Taken); + assert_eq!(seen.advance(jump, now, false), Noted::Unagreed); + assert_eq!(seen.bounds(now).0, real + 10); + assert!(seen.cross_check_due(now), "the endpoints are asked"); + + seen.confirm(jump, now); + + 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), + Noted::Taken, + "the only endpoint has no other to agree with" + ); + } + #[test] fn a_confirmed_block_replaces_one_far_ahead_that_was_never_confirmed() { let now = Instant::now(); let mut seen = SeenBlock::new(); let real = 5_000_000; - assert!(seen.advance(real + WEEK_OF_BLOCKS * 10, now)); + assert_eq!( + seen.advance(real + WEEK_OF_BLOCKS * 10, now, false), + Noted::Taken + ); seen.confirm(real, now); diff --git a/bin/dipper-service/src/config.rs b/bin/dipper-service/src/config.rs index 8bc4619c..abab64f5 100644 --- a/bin/dipper-service/src/config.rs +++ b/bin/dipper-service/src/config.rs @@ -2118,6 +2118,23 @@ mod tests { ); } + #[test] + fn the_example_configmap_alerts_section_parses() { + let example = include_str!("../../../k8s/configmap-example.yaml"); + let json: String = example + .split_once("config.json: |\n") + .expect("the example holds a config.json") + .1 + .lines() + .map(|line| format!("{}\n", line.strip_prefix(" ").unwrap_or(line))) + .collect(); + let config: serde_json::Value = serde_json::from_str(&json).expect("config.json is JSON"); + + let alerts = serde_json::from_value::(config["alerts"].clone()) + .expect("the example's alerts section is a valid alerts config"); + assert!(alerts.slack_webhook_url.is_some()); + } + #[test] fn alerts_config_takes_a_webhook_and_keeps_the_default_events() { let alerts = serde_json::from_str::( diff --git a/bin/dipper-service/src/network/service/cancel_retry.rs b/bin/dipper-service/src/network/service/cancel_retry.rs index 3417b30e..322ff341 100644 --- a/bin/dipper-service/src/network/service/cancel_retry.rs +++ b/bin/dipper-service/src/network/service/cancel_retry.rs @@ -41,9 +41,11 @@ const DB_TIMEOUT: Duration = Duration::from_secs(30); const QUEUE_TIMEOUT: Duration = Duration::from_secs(10); /// Minutes an agreement stays out of the retry after it is marked, so the cancel sent -/// when it was marked can be mined first instead of being sent again. One moved back to -/// cancelling had none sent, so it doesn't wait. -const SETTLE_MINUTES: i32 = 2; +/// when it was marked can be mined first instead of being sent again. That cancel can wait +/// behind another submission for its whole submit deadline, which stays under the 7-minute +/// worker job limit, so this outlasts that limit. One moved back to cancelling had none sent, +/// so it doesn't wait. +const SETTLE_MINUTES: i32 = 8; /// How long the chain listener gets, from when a check first finds an agreement ended, to /// record when and in which transaction it ended, before the retry marks it without them. @@ -147,7 +149,8 @@ where /// Retry the cancel of agreements still `Cancelling`. The chain's own latest block time /// decides when an offer that was never accepted no longer can be, so a subgraph that has -/// fallen behind doesn't hold that up. +/// fallen behind doesn't hold that up. Without it, such an offer is still withdrawn, but only +/// accepted agreements can be marked ended. pub async fn retry_cancelling_agreements( registry: &R, chain_client: &T, @@ -169,8 +172,10 @@ pub async fn retry_cancelling_agreements( if cancelling.is_empty() { return; } - let Some(chain_now) = chain_time(chain_client).await else { - return; + let chain_now = if cancelling.iter().any(|row| !row.accepted_on_chain) { + chain_time(chain_client).await + } else { + None }; let started = std::time::Instant::now(); for (done, row) in cancelling.iter().enumerate() { @@ -191,7 +196,7 @@ async fn chain_time(chain_client: &T) -> Option { Err(err) => { tracing::warn!( error = %err, - "Failed to read the chain's time; cancels are retried next sweep" + "Failed to read the chain's time; offers never accepted stay cancelling until a later sweep" ); None } @@ -203,7 +208,7 @@ async fn retry_cancel( chain_client: &T, config: &IndexingAgreementConfig, row: &CancellingAgreement, - chain_now: u64, + chain_now: Option, ) where R: AgreementRegistry + Sync, T: ChainClient, @@ -253,7 +258,7 @@ async fn confirm_if_over( row: &CancellingAgreement, tx_hash: Option, by_indexer: bool, - chain_now: u64, + chain_now: Option, ) -> bool { let agreement = &row.agreement; let past_grace = row @@ -262,7 +267,7 @@ async fn confirm_if_over( let can_confirm = if row.accepted_on_chain { tx_hash.is_some() || past_grace } else { - chain_now > agreement.terms.deadline + chain_now.is_some_and(|now| now > agreement.terms.deadline) }; if !can_confirm { return false; @@ -433,6 +438,12 @@ mod tests { const DEADLINE: u64 = 1_000; + #[test] + fn a_marked_agreement_waits_out_a_whole_submit_deadline_before_a_retry() { + let settle = Duration::from_secs(u64::from(SETTLE_MINUTES.unsigned_abs()) * 60); + assert!(settle > crate::worker::service::PROCESS_JOB_TIMEOUT); + } + #[derive(Default)] struct MockRegistry { cancelling: Vec, @@ -762,7 +773,8 @@ mod tests { let ended = ended_abandoned(); let registry = MockRegistry { awaiting_replacement: vec![ended.clone()], - ..registry_with_one(true) + // Never accepted, so each sweep reads the chain's time, which counts the sweeps. + ..registry_with_one(false) }; let chain = Arc::new(live_chain()); let queue = MockQueue::default(); @@ -792,17 +804,31 @@ mod tests { } #[tokio::test] - async fn waits_for_the_next_sweep_when_the_chain_time_cannot_be_read() { + async fn cancels_an_accepted_agreement_without_the_chain_time() { + // Only an offer never accepted needs the chain's time, to tell if its deadline passed. let registry = registry_with_one(true); + let chain = live_chain(); + + retry(®istry, &chain, 0).await; + + assert_eq!(chain.clock_reads.load(Ordering::SeqCst), 0); + assert_eq!(chain.cancels_sent.load(Ordering::SeqCst), 1); + assert_eq!(registry.marked_cancelled.lock().unwrap().len(), 1); + } + + #[tokio::test] + async fn withdraws_an_offer_but_keeps_it_cancelling_when_the_chain_time_cannot_be_read() { + let registry = registry_with_one(false); let chain = MockChain { clock_fails: true, ..live_chain() }; - retry(®istry, &chain, 0).await; + retry(®istry, &chain, DEADLINE + 1).await; - assert_eq!(chain.cancels_sent.load(Ordering::SeqCst), 0); - assert_eq!(registry.checks.load(Ordering::SeqCst), 0); + assert_eq!(chain.cancels_sent.load(Ordering::SeqCst), 1); + assert!(registry.marked_cancelled.lock().unwrap().is_empty()); + assert_eq!(registry.checks.load(Ordering::SeqCst), 1); } #[tokio::test] diff --git a/bin/dipper-service/src/network/service/chain_listener.rs b/bin/dipper-service/src/network/service/chain_listener.rs index a3ce327c..2e8ad5c2 100644 --- a/bin/dipper-service/src/network/service/chain_listener.rs +++ b/bin/dipper-service/src/network/service/chain_listener.rs @@ -75,6 +75,39 @@ const STALL_ERROR_THRESHOLD: u32 = 15; /// Sanity bound on a single tick's drain — not an operating point; /// steady-state drains are 0–1 pages. const MAX_PAGES_PER_DRAIN: u32 = 1000; +/// How long the listener keeps re-reading a page whose changes fail to apply before moving past +/// it: long enough to ride out a database failover, short enough that a change that can never +/// apply doesn't hold up every later accept and cancel. +const MAX_CURSOR_HOLD: Duration = Duration::from_secs(600); + +/// Where the cursor has been held, and since when, before a page whose changes failed to apply. +#[derive(Debug)] +struct CursorHold { + cursor: Cursor, + since: std::time::Instant, +} + +/// Whether to move past a page at `cursor` whose changes failed to apply: once the cursor has +/// been held before it for [`MAX_CURSOR_HOLD`]. Until then the hold is noted, so the page is +/// read again. +fn hold_expired(hold: &mut Option, cursor: &Cursor, now: std::time::Instant) -> bool { + match hold { + Some(held) if held.cursor == *cursor => { + if now.saturating_duration_since(held.since) < MAX_CURSOR_HOLD { + return false; + } + *hold = None; + true + } + _ => { + *hold = Some(CursorHold { + cursor: cursor.clone(), + since: now, + }); + false + } + } +} /// Cap on the cancellation-sweep batch so a backlog drains across polls /// instead of blocking one tick. const SWEEP_BATCH_SIZE: i64 = 1000; @@ -213,6 +246,7 @@ where // Track consecutive failures for adaptive backoff let mut consecutive_failures: u32 = 0; + let mut cursor_hold: Option = None; // Observability: heartbeat and stall detection let mut polls_since_last_event: u64 = 0; @@ -289,6 +323,7 @@ where let outcome = match drain_once( &mut cursor, + &mut cursor_hold, &mut last_persisted_timestamp, &mut last_chain_ts_persist_wall, &mut last_subgraph_head, @@ -419,6 +454,7 @@ struct DrainOutcome { )] async fn drain_once( cursor: &mut Cursor, + cursor_hold: &mut Option, last_persisted_timestamp: &mut Option, last_chain_ts_persist_wall: &mut std::time::Instant, last_subgraph_head: &mut u64, @@ -555,6 +591,9 @@ where // another payer's) drop out as `None` below. let snapshot_ids: Vec = result.snapshots.iter().map(|s| s.agreement_id).collect(); + // Agreements whose change this page failed to apply. The cursor stays before the page + // until none fail, as nothing else would look at them again. + let mut unapplied: Vec = Vec::new(); let mut agreements_by_id = match registry.get_indexing_agreements_by_ids(&snapshot_ids).await { Ok(m) => m, @@ -566,6 +605,7 @@ where "Failed to batch-fetch agreements for page; counting snapshots as errors" ); drain_errors += snapshot_ids.len() as u64; + unapplied.extend(&snapshot_ids); std::collections::HashMap::new() } }; @@ -588,12 +628,13 @@ where "Failed to prepare reconciliation for snapshot" ); drain_errors += 1; + unapplied.push(snapshot.agreement_id); } } } - // On batch error the tx rolls back; the next tick re-reads - // these rows via the reorg buffer. The successful path + // On batch error the tx rolls back; the cursor is held so the + // next tick re-reads these rows. The successful path // pre-fills every input id in `outcomes` (with `default()` // for rows whose CAS guard didn't match), so a missing id // is an unambiguous signal that the whole batch failed. @@ -611,6 +652,7 @@ where "Batched apply_reconciliation failed; counting page items as errors" ); drain_errors += items.len() as u64; + unapplied.extend(items.iter().map(|item| item.agreement_id)); std::collections::HashMap::new() } } @@ -672,7 +714,27 @@ where // Persist per-page so a crash mid-drain replays at most // one page. Skip the write when neither cursor nor // ratcheted timestamp moved. - let advance_cursor = *cursor < new_cursor; + let skip = !unapplied.is_empty() && hold_expired(cursor_hold, cursor, now_instant); + if unapplied.is_empty() { + *cursor_hold = None; + } + let advance_cursor = (unapplied.is_empty() || skip) && *cursor < new_cursor; + if skip { + tracing::error!( + event = "chain_listener_changes_skipped", + cursor_block = cursor.block, + agreement_ids = ?unapplied, + "Skipped on-chain changes to agreements that kept failing to apply; they won't \ + be read again until their next change" + ); + } else if !unapplied.is_empty() { + tracing::error!( + event = "chain_listener_cursor_held", + cursor_block = cursor.block, + agreement_ids = ?unapplied, + "Failed to apply on-chain changes to agreements; re-reading them next poll" + ); + } let timestamp_changed = ratchet_timestamp != *last_persisted_timestamp; if advance_cursor || timestamp_changed { let cursor_to_persist = if advance_cursor { @@ -701,8 +763,8 @@ where // nothing to do this tick. break 'drain; } else { - // Held-back cursor (parse failure path); reconcile - // already ran on what we did read, retry next tick. + // Held-back cursor (parse failure, or a change that failed to + // apply); reconcile already ran on what we did read, retry next tick. break 'drain; } @@ -845,16 +907,18 @@ where indexer = %snapshot.indexer, "Rejected agreement accepted on-chain, cancelling it" ); - crate::cancel_dispatch::reopen_if_live(registry, chain_client, &agreement).await?; - // This row goes Rejected -> Canceled without ever transiting // AcceptedOnChain, so `apply_reconciliation` never records the accept. // Persist it from the snapshot so the eventual `terminated` (emitted by // the sweep after the cancel) is eligible (`accepted_at IS NOT NULL`) and - // carries accurate accept data. - if let Err(err) = registry - .record_accepted_audit(&agreement.id, snapshot.accepted_at, &snapshot.accepted_tx) - .await + // carries accurate accept data. One left Rejected, as the chain shows it + // already ended, gets no accept on record, so none is announced. + let reopened = + crate::cancel_dispatch::reopen_if_live(registry, chain_client, &agreement).await?; + if reopened + && let Err(err) = registry + .record_accepted_audit(&agreement.id, snapshot.accepted_at, &snapshot.accepted_tx) + .await { tracing::warn!( agreement_id = %agreement.id, @@ -2879,6 +2943,30 @@ mod tests { 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); + + 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.audit_writes().is_empty()); + } + #[tokio::test] async fn test_reconcile_rejected_already_canceled_skips_queue_and_marks_local_cancel() { // The Rejected agreement was canceled on-chain between polls. Dipper @@ -4125,6 +4213,7 @@ mod tests { let outcome = drain_once( &mut cursor, + &mut None, &mut last_persisted_timestamp, &mut last_chain_ts_persist_wall, &mut last_subgraph_head, @@ -4161,6 +4250,118 @@ mod tests { } } + /// Drain 1 page holding a change that accepts an agreement, its apply failing or not, with + /// the cursor held before it since `held_since` if given; the cursor it leaves, and whether + /// the agreement was accepted. + async fn drain_a_page( + apply_fails: &[bool], + held_since: Option, + ) -> (Cursor, Option, bool) { + let registry = MockRegistry::new(); + let chain_client = MockChainClient::default(); + let event_source = super::super::chain_events::mock::MockEventSource::new(); + event_source.set_latest_block(100); + event_source.set_latest_block_timestamp(Some(1_700_000_000)); + let agreement_id = IndexingAgreementId::from_bytes(rand::random()); + registry.add_agreement(agreement_id, IndexingAgreementStatus::Created); + event_source.add_snapshots(vec![AgreementStateSnapshot { + agreement_id, + indexer: Address::ZERO, + state: super::super::chain_events::AgreementState::Accepted, + canceled_by: Address::ZERO, + last_state_change_block: 10, + accepted_at: 0, + accepted_tx: String::new(), + canceled_at: 0, + canceled_tx: String::new(), + }]); + let mut cursor = Cursor::genesis(); + let mut cursor_hold = held_since.map(|since| CursorHold { + cursor: Cursor::genesis(), + since, + }); + let mut last_persisted_timestamp: Option = None; + let mut last_chain_ts_persist_wall = std::time::Instant::now(); + let mut last_subgraph_head: u64 = 0; + let mut stall_count: u32 = 0; + let mut consecutive_failures: u32 = 0; + let (_tx_stop, mut rx_stop) = mpsc::channel::<()>(1); + + for fails in apply_fails { + registry.set_fail_batch(*fails); + drain_once( + &mut cursor, + &mut cursor_hold, + &mut last_persisted_timestamp, + &mut last_chain_ts_persist_wall, + &mut last_subgraph_head, + &mut stall_count, + &mut consecutive_failures, + 1337, + 0, + 10, + false, + ®istry, + &chain_client, + &event_source, + &mut rx_stop, + test_agreement_conf().as_ref(), + ) + .await + .expect("subgraph fetch succeeded"); + } + let accepted = registry.was_marked_accepted_on_chain(&agreement_id); + (cursor, cursor_hold, accepted) + } + + /// A page whose changes fail to apply keeps the cursor before it, so the next poll + /// re-reads it: no later change to those agreements may come to apply it instead. + #[tokio::test] + async fn test_drain_holds_the_cursor_until_a_failed_page_applies() { + let (cursor, hold, _) = drain_a_page(&[true], None).await; + assert_eq!(cursor, Cursor::genesis(), "held before the failed page"); + assert!(hold.is_some()); + + let (cursor, hold, accepted) = drain_a_page(&[true, false], None).await; + assert!(accepted); + assert!(cursor > Cursor::genesis()); + assert!(hold.is_none()); + } + + /// A page that keeps failing is passed over once the cursor has been held long enough, so + /// it can't hold up every later change. + #[tokio::test] + async fn test_drain_moves_past_a_page_that_keeps_failing() { + let long_ago = std::time::Instant::now() + .checked_sub(MAX_CURSOR_HOLD + Duration::from_secs(1)) + .expect("instant"); + + let (cursor, hold, accepted) = drain_a_page(&[true], Some(long_ago)).await; + + assert!(!accepted); + assert!(cursor > Cursor::genesis()); + assert!(hold.is_none()); + } + + #[test] + fn holds_the_cursor_before_a_failing_page_for_a_while() { + let now = std::time::Instant::now(); + let page = Cursor::genesis(); + let mut hold = None; + + assert!( + !hold_expired(&mut hold, &page, now), + "the first failure holds" + ); + assert!(!hold_expired( + &mut hold, + &page, + now + MAX_CURSOR_HOLD - Duration::from_secs(1) + )); + assert!(hold_expired(&mut hold, &page, now + MAX_CURSOR_HOLD)); + assert!(hold.is_none()); + } + /// Drives a hostile timestamp drift through `drain_once` and asserts /// the persisted chain timestamp is capped near /// `CHAIN_TS_DRIFT_TOLERANCE_SECS` instead of accepting the huge @@ -4189,6 +4390,7 @@ mod tests { drain_once( &mut cursor, + &mut None, &mut last_persisted_timestamp, &mut last_chain_ts_persist_wall, &mut last_subgraph_head, @@ -4216,6 +4418,7 @@ mod tests { drain_once( &mut cursor, + &mut None, &mut last_persisted_timestamp, &mut last_chain_ts_persist_wall, &mut last_subgraph_head, @@ -4319,6 +4522,7 @@ mod tests { drain_once( &mut cursor, + &mut None, &mut last_persisted_timestamp, &mut last_chain_ts_persist_wall, &mut last_subgraph_head, diff --git a/bin/dipper-service/src/network/service/liveness_checker.rs b/bin/dipper-service/src/network/service/liveness_checker.rs index a95e2682..6f1a87c6 100644 --- a/bin/dipper-service/src/network/service/liveness_checker.rs +++ b/bin/dipper-service/src/network/service/liveness_checker.rs @@ -507,8 +507,14 @@ async fn cancel_and_reassess( ); return; } - let Some(started) = - start_abandoned_cancel(agreement, registry, chain_client, agreement_conf).await + let Some(started) = start_abandoned_cancel( + agreement, + registry, + chain_client, + agreement_conf, + db_timeout, + ) + .await else { return; }; @@ -520,9 +526,39 @@ async fn cancel_and_reassess( ); return; } + if started == CancelStarted::NotLive { + note_seen_ended(agreement, registry, db_timeout).await; + } replace_abandoned(agreement, registry, worker_queue, db_timeout, queue_timeout).await; } +/// Note that the chain shows a stale agreement still cancelling has ended, so it no longer +/// holds its slot in the request and the reassessment queued next fills it. Bounded by +/// `db_timeout`, so a hung database can't stall the liveness cycle. +async fn note_seen_ended( + agreement: &IndexingAgreement, + registry: &R, + db_timeout: Duration, +) { + let noted = tokio::time::timeout( + db_timeout, + registry.record_cancel_check(&agreement.id, 0, Some(true)), + ) + .await; + match noted { + Ok(Ok(_) | Err(crate::registry::Error::NoRecordsUpdated)) => {} + Ok(Err(err)) => tracing::warn!( + agreement_id = %agreement.id, + error = %err, + "Failed to note a stale agreement ended; its replacement may wait for a later reassessment" + ), + Err(_) => tracing::warn!( + agreement_id = %agreement.id, + "Timeout noting a stale agreement ended; its replacement may wait for a later reassessment" + ), + } +} + /// Queue the replacement of an agreement whose indexer stopped serving it, and note it queued /// so it isn't queued again. pub(crate) async fn replace_abandoned( @@ -553,45 +589,62 @@ async fn start_abandoned_cancel( registry: &R, chain_client: &C, agreement_conf: &crate::config::IndexingAgreementConfig, + db_timeout: Duration, ) -> Option where R: AgreementRegistry + Sync, C: ChainClient, { - match crate::cancel_dispatch::start_cancel( + if !mark_abandoning(agreement, registry, db_timeout).await { + return None; + } + let started = crate::cancel_dispatch::send_marked_cancel( registry, chain_client, agreement, CancelReason::Abandoned, agreement_conf, ) - .await - { - Ok(started) => { - tracing::info!( - agreement_id = %agreement.id, - ?started, - reason = "indexer_stale", - "Cancelling stale agreement" - ); - Some(started) - } - Err(crate::registry::Error::NoRecordsUpdated) => { - tracing::debug!( - agreement_id = %agreement.id, - "Stale agreement already ended or being cancelled" - ); - None - } - Err(err) => { - tracing::error!( - agreement_id = %agreement.id, - error = %err, - "failed to mark stale agreement cancelling, will retry next cycle" - ); - None - } + .await; + tracing::info!( + agreement_id = %agreement.id, + ?started, + reason = "indexer_stale", + "Cancelling stale agreement" + ); + Some(started) +} + +/// Mark a stale agreement cancelling as abandoned; false, logged, when it isn't marked. Bounded +/// by `db_timeout`, so a hung database can't stall the liveness cycle; a mark that lands after +/// the timeout leaves the agreement for the cancel retry. +async fn mark_abandoning( + agreement: &IndexingAgreement, + registry: &R, + db_timeout: Duration, +) -> bool { + let marked = tokio::time::timeout( + db_timeout, + crate::cancel_dispatch::mark_cancelling(registry, agreement, CancelReason::Abandoned), + ) + .await; + match marked { + Ok(Ok(())) => return true, + Ok(Err(crate::registry::Error::NoRecordsUpdated)) => tracing::debug!( + agreement_id = %agreement.id, + "Stale agreement already ended or being cancelled" + ), + Ok(Err(err)) => tracing::error!( + agreement_id = %agreement.id, + error = %err, + "failed to mark stale agreement cancelling, will retry next cycle" + ), + Err(_) => tracing::error!( + agreement_id = %agreement.id, + "timeout marking stale agreement cancelling, will retry next cycle" + ), } + false } /// Drop the pending cancellations an abandoned agreement holds as a replacement, so the @@ -955,11 +1008,15 @@ mod tests { cancel_audits: Arc>>, /// Ids noted as having their replacement queued. replacements_noted: Arc>>, + /// Ids noted as seen ended on-chain while still cancelling. + seen_ended: Arc>>, } struct MockRegistry { calls: MockCalls, already_ending: bool, + mark_hangs: bool, + check_hangs: bool, get_request_result: Arc>>>>, } @@ -969,6 +1026,8 @@ mod tests { Self { calls, already_ending: false, + mark_hangs: false, + check_hangs: false, get_request_result: Arc::new(Mutex::new(Some(Ok(Some(request))))), } } @@ -1003,6 +1062,9 @@ mod tests { if self.already_ending { return Err(crate::registry::Error::NoRecordsUpdated); } + if self.mark_hangs { + std::future::pending::<()>().await; + } self.calls.abandoning.lock().unwrap().push(*id); Ok(()) } @@ -1012,6 +1074,21 @@ mod tests { Ok(()) } + async fn record_cancel_check( + &self, + id: &IndexingAgreementId, + _failed_attempts: u32, + ended: Option, + ) -> RegistryResult { + if self.check_hangs { + std::future::pending::<()>().await; + } + if ended == Some(true) { + self.calls.seen_ended.lock().unwrap().push(*id); + } + Ok(0) + } + async fn mark_indexing_agreement_as_canceled_by_requester( &self, id: &IndexingAgreementId, @@ -1568,6 +1645,11 @@ mod tests { end_stale(&agreement, ®istry, &chain).await; assert!(calls.chain_cancels.lock().unwrap().is_empty()); + assert_eq!( + calls.seen_ended.lock().unwrap().as_slice(), + &[agreement.id], + "so it no longer holds the slot its replacement fills" + ); assert_eq!( calls.reassessments.lock().unwrap().as_slice(), &[agreement.indexing_request_id] @@ -1610,6 +1692,42 @@ mod tests { assert!(calls.reassessments.lock().unwrap().is_empty()); } + #[tokio::test(start_paused = true)] + async fn gives_up_on_a_stale_agreement_whose_mark_hangs() { + // A hung database must not hold up the rest of the liveness cycle. + let agreement = stale_agreement(); + let calls = MockCalls::default(); + let registry = MockRegistry { + mark_hangs: true, + ..MockRegistry::new(calls.clone(), agreement.clone()) + }; + let chain = MockChainClient::success(calls.clone()); + + end_stale(&agreement, ®istry, &chain).await; + + assert!(calls.chain_cancels.lock().unwrap().is_empty()); + assert!(calls.reassessments.lock().unwrap().is_empty()); + } + + #[tokio::test(start_paused = true)] + async fn still_replaces_an_ended_stale_agreement_when_noting_its_end_hangs() { + let agreement = stale_agreement(); + let calls = MockCalls::default(); + let registry = MockRegistry { + check_hangs: true, + ..MockRegistry::new(calls.clone(), agreement.clone()) + }; + let chain = MockChainClient::rpc_error(calls.clone()); + chain.live.store(false, std::sync::atomic::Ordering::SeqCst); + + end_stale(&agreement, ®istry, &chain).await; + + assert_eq!( + calls.reassessments.lock().unwrap().as_slice(), + &[agreement.indexing_request_id] + ); + } + #[tokio::test] async fn test_record_progress_calls_registry() { // Arrange diff --git a/bin/dipper-service/src/registry.rs b/bin/dipper-service/src/registry.rs index 71470cda..e194517d 100644 --- a/bin/dipper-service/src/registry.rs +++ b/bin/dipper-service/src/registry.rs @@ -733,6 +733,16 @@ impl AgreementRegistry for RegistryProvider { .map_err(Into::into) } + async fn get_abandoned_indexers_holding_slots( + &self, + request_id: &IndexingRequestId, + ) -> RegistryResult> { + self.inner + .get_abandoned_indexers_holding_slots(request_id) + .await + .map_err(Into::into) + } + async fn update_agreement_sync_progress( &self, id: &IndexingAgreementId, diff --git a/bin/dipper-service/src/registry/agreement.rs b/bin/dipper-service/src/registry/agreement.rs index 7f8618b1..fde09894 100644 --- a/bin/dipper-service/src/registry/agreement.rs +++ b/bin/dipper-service/src/registry/agreement.rs @@ -359,101 +359,76 @@ pub trait AgreementRegistry { } /// Fetch agreements awaiting a `terminated` event: terminal-cancel status, - /// genuinely accepted on-chain, marker unset. Default returns empty so mocks - /// need not override; the production registry runs the real query. + /// genuinely accepted on-chain, marker unset. async fn get_agreements_pending_terminated_emission( &self, - _limit: i64, - ) -> RegistryResult> { - Ok(Vec::new()) - } + limit: i64, + ) -> RegistryResult>; /// Stamp the `terminated` emission marker after a confirmed broker send. - /// Default no-op so mocks need not override. async fn mark_terminated_event_emitted( &self, - _agreement_id: &IndexingAgreementId, - ) -> RegistryResult<()> { - Ok(()) - } + agreement_id: &IndexingAgreementId, + ) -> RegistryResult<()>; /// Fetch agreements awaiting an `accepted` event: `AcceptedOnChain`, accept - /// observed by the new reconcile path, marker unset. Default returns empty so - /// mocks need not override; the production registry runs the real query. + /// observed by the new reconcile path, marker unset. async fn get_agreements_pending_accepted_emission( &self, - _limit: i64, - ) -> RegistryResult> { - Ok(Vec::new()) - } + limit: i64, + ) -> RegistryResult>; /// Stamp the `accepted` emission marker after a confirmed broker send. - /// Default no-op so mocks need not override. async fn mark_accepted_event_emitted( &self, - _agreement_id: &IndexingAgreementId, - ) -> RegistryResult<()> { - Ok(()) - } + agreement_id: &IndexingAgreementId, + ) -> RegistryResult<()>; /// Fetch agreements awaiting a `request.expired` event: `Expired`, marker - /// unset. Default returns empty so mocks need not override; the production - /// registry runs the real query. + /// unset. async fn get_agreements_pending_expired_emission( &self, - _limit: i64, - ) -> RegistryResult> { - Ok(Vec::new()) - } + limit: i64, + ) -> RegistryResult>; /// Stamp the `request.expired` emission marker after a confirmed broker send. - /// Default no-op so mocks need not override. async fn mark_expired_event_emitted( &self, - _agreement_id: &IndexingAgreementId, - ) -> RegistryResult<()> { - Ok(()) - } + agreement_id: &IndexingAgreementId, + ) -> RegistryResult<()>; /// Record the accepted audit payload out-of-band (rejected-then-accepted /// anomaly), marking the row as genuinely accepted so its eventual - /// `terminated` is sweep-eligible. Default no-op so mocks need not override. + /// `terminated` is sweep-eligible. async fn record_accepted_audit( &self, - _agreement_id: &IndexingAgreementId, - _accepted_at: u64, - _accepted_tx: &str, - ) -> RegistryResult<()> { - Ok(()) - } + agreement_id: &IndexingAgreementId, + accepted_at: u64, + accepted_tx: &str, + ) -> RegistryResult<()>; /// Record an agreement's accept and its end together, in 1 write, so nothing reads one - /// without the other. Default no-op so mocks need not override. + /// without the other. #[allow(clippy::too_many_arguments)] async fn record_accept_and_cancel_audit( &self, - _agreement_id: &IndexingAgreementId, - _accepted_at: u64, - _accepted_tx: &str, - _canceled_at: u64, - _canceled_by: &str, - _canceled_tx: Option<&str>, - ) -> RegistryResult<()> { - Ok(()) - } + agreement_id: &IndexingAgreementId, + accepted_at: u64, + accepted_tx: &str, + canceled_at: u64, + canceled_by: &str, + canceled_tx: Option<&str>, + ) -> RegistryResult<()>; /// Record the cancel audit payload for a dipper-initiated cancel so the - /// emission sweep can populate the `terminated` event fields. Default no-op - /// so mocks need not override. + /// emission sweep can populate the `terminated` event fields. async fn record_cancel_audit( &self, - _agreement_id: &IndexingAgreementId, - _canceled_at: u64, - _canceled_by: &str, - _canceled_tx: Option<&str>, - ) -> RegistryResult<()> { - Ok(()) - } + agreement_id: &IndexingAgreementId, + canceled_at: u64, + canceled_by: &str, + canceled_tx: Option<&str>, + ) -> RegistryResult<()>; /// Get `Created` agreements whose deadline has passed (by block timestamp). async fn get_expired_created_agreements( @@ -522,6 +497,13 @@ pub trait AgreementRegistry { /// Note that an agreement's replacement has been queued. async fn mark_replacement_queued(&self, id: &IndexingAgreementId) -> RegistryResult<()>; + /// The indexers holding a slot in a request with an agreement they stopped serving that may + /// still be paid, so a reassessment fills only the rest and never picks them again. + async fn get_abandoned_indexers_holding_slots( + &self, + request_id: &IndexingRequestId, + ) -> RegistryResult>; + /// Update the sync progress for an agreement. /// /// Called when the liveness checker observes the block height has changed @@ -542,9 +524,9 @@ pub trait AgreementRegistry { &self, ) -> RegistryResult>; - /// Count `Created` (in-flight, not yet accepted) agreements per indexer, - /// plus the global total. Offer pacing reads both to decide whether an - /// indexer or the network has spare acceptance capacity. + /// Count in-flight offers per indexer, plus the global total: `Created` agreements, and + /// `Cancelling` ones never accepted until their offer deadline. Offer pacing reads both to + /// decide whether an indexer or the network has spare acceptance capacity. async fn count_created_agreements_by_indexer( &self, ) -> RegistryResult<(std::collections::HashMap, u64)>; diff --git a/bin/dipper-service/src/registry/agreement_stub.rs b/bin/dipper-service/src/registry/agreement_stub.rs index dd2950b4..1253c754 100644 --- a/bin/dipper-service/src/registry/agreement_stub.rs +++ b/bin/dipper-service/src/registry/agreement_stub.rs @@ -236,6 +236,13 @@ pub trait StubAgreementRegistry: Send + Sync { unimplemented!("mark_replacement_queued") } + async fn get_abandoned_indexers_holding_slots( + &self, + _request_id: &IndexingRequestId, + ) -> Result> { + unimplemented!("get_abandoned_indexers_holding_slots") + } + async fn update_agreement_sync_progress( &self, _id: &IndexingAgreementId, @@ -271,8 +278,8 @@ pub trait StubAgreementRegistry: Send + Sync { unimplemented!("count_accepted_agreements_by_deployment") } - // The emission-sweep and audit methods mirror the real trait's defaults - // (empty batch / no-op) so only emission-focused mocks override them. + // The emission-sweep and audit methods default to an empty batch or a + // no-op, here only, so only emission-focused mocks override them. async fn get_agreements_pending_terminated_emission( &self, _limit: i64, @@ -568,6 +575,13 @@ impl AgreementRegistry for T { StubAgreementRegistry::mark_replacement_queued(self, id).await } + async fn get_abandoned_indexers_holding_slots( + &self, + request_id: &IndexingRequestId, + ) -> Result> { + StubAgreementRegistry::get_abandoned_indexers_holding_slots(self, request_id).await + } + async fn update_agreement_sync_progress( &self, id: &IndexingAgreementId, 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 584e08d3..1d9f19fc 100644 --- a/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs +++ b/bin/dipper-service/src/worker/handlers/reassess_indexing_request.rs @@ -147,6 +147,17 @@ where .map(|agreement| agreement.indexer.id) .collect(); + // An agreement whose indexer stopped serving it keeps its slot while it may still be paid, + // so it isn't replaced while both are paid; once it can't be, its replacement is queued. + // Held slots only limit what is added, never what is cancelled. + let held: HashSet = ctx + .registry + .get_abandoned_indexers_holding_slots(indexing_request_id) + .await + .map_err(|err| JobError::Fatal(err.into()))? + .into_iter() + .collect(); + // In-flight offer counts, fetched once and reused by both the saturated // short-circuit (global figure only) and the per-indexer cap after IISA. let (in_flight_per_indexer, global_in_flight) = ctx @@ -162,7 +173,7 @@ where if should_defer_saturated( global_in_flight, global_cap, - *num_candidates, + num_candidates.saturating_sub(held.len()), active_agreements.len(), ) { tracing::debug!( @@ -273,9 +284,18 @@ where let target_pricing: HashMap = target_selected.iter().map(|s| (s.id, s)).collect(); - // Compute the diff + // Compute the diff. New offers fill only what the agreements kept and the held slots leave + // of the target, in IISA's order, and never for an indexer holding a slot. let to_cancel: HashSet<&IndexerId> = current_ids.difference(&target_ids).collect(); - let to_add: HashSet<&IndexerId> = target_ids.difference(¤t_ids).collect(); + let diff_add_count = target_ids.difference(¤t_ids).count(); + let kept = current_ids.intersection(&target_ids).count(); + let room = num_candidates.saturating_sub(held.len() + kept); + let to_add: HashSet<&IndexerId> = target_selected + .iter() + .map(|s| &s.id) + .filter(|id| !current_ids.contains(*id) && !held.contains(*id)) + .take(room) + .collect(); // Surface the IISA outcome alongside the diff so the operator can tell // "everyone is declined" apart from "we're already synced" without @@ -286,6 +306,7 @@ where indexing_request_id=%indexing_request_id, deployment_id=%deployment_id, requested_num_candidates = num_candidates, + held_slots = held.len(), iisa_returned_count = target_selected.len(), current_active_count = current_ids.len(), to_add_count = to_add.len(), @@ -345,8 +366,11 @@ where // Offer pacing: only create as many agreements as the network can accept // before the deadline, so a burst does not queue offers that expire unaccepted. + // Adds left out for held slots count as withheld, so the old agreements they would have + // replaced stay until a later pass can replace them. let (capped_to_add, offer_pacing_withheld, reserved_cancel_count) = if to_add.is_empty() { - (HashSet::new(), 0usize, 0usize) + let reserved = reserved_cancel_count(old_to_cancel.len(), diff_add_count, 0); + (HashSet::new(), 0usize, reserved) } else { // Reuse the in-flight counts and global cap fetched at the top of the pass. let per_indexer_cap = ctx.agreement_conf.max_in_flight_offers_per_indexer(); @@ -358,7 +382,7 @@ where per_indexer_cap, global_cap, ); - let reserved = reserved_cancel_count(old_to_cancel.len(), to_add.len(), allowed.len()); + let reserved = reserved_cancel_count(old_to_cancel.len(), diff_add_count, allowed.len()); if withheld > 0 { tracing::info!( event = "offer_pacing_capped", @@ -871,7 +895,7 @@ mod lifecycle_event_tests { // ---- Mock: IISA candidate selection ------------------------------------- - /// Returns a fixed list of [`SelectedIndexer`] regardless of arguments. + /// Returns a fixed list of [`SelectedIndexer`], at most as many as asked for. struct MockIisa { selected: Vec, } @@ -881,10 +905,10 @@ mod lifecycle_event_tests { async fn select_indexers( &self, _deployment_id: DeploymentId, - _num_candidates: usize, + num_candidates: usize, _context: &SelectionContext, ) -> std::result::Result, SelectionError> { - Ok(self.selected.clone()) + Ok(self.selected.iter().take(num_candidates).cloned().collect()) } async fn dips_accepting_indexers( @@ -1039,6 +1063,8 @@ mod lifecycle_event_tests { /// Ids marked CanceledByRequester locally. marked_cancelled: Arc>>, marked_cancelling: Arc>>, + /// Indexers holding a slot with an agreement they stopped serving. + held: Vec, } #[async_trait] @@ -1119,6 +1145,12 @@ mod lifecycle_event_tests { ) -> RegistryResult> { Ok(self.active_agreements.clone()) } + async fn get_abandoned_indexers_holding_slots( + &self, + _request_id: &IndexingRequestId, + ) -> RegistryResult> { + Ok(self.held.clone()) + } // Populates the `terminated` event's remaining count. async fn count_accepted_agreements_by_deployment( &self, @@ -1500,6 +1532,7 @@ mod lifecycle_event_tests { shortfall_active: std::sync::Mutex::new(true), marked_cancelled: Arc::default(), marked_cancelling: Arc::default(), + held: Vec::new(), }; let ctx = build_ctx( registry, @@ -1538,6 +1571,7 @@ mod lifecycle_event_tests { shortfall_active: std::sync::Mutex::new(false), marked_cancelled: Arc::default(), marked_cancelling: Arc::default(), + held: Vec::new(), }; let ctx = build_ctx( registry, @@ -1594,6 +1628,73 @@ mod lifecycle_event_tests { assert_eq!(unavailable[0].candidates_returned, 0); } + #[tokio::test] + async fn fills_only_the_slots_an_abandoned_agreement_does_not_hold() { + // The abandoned agreement may still be paid until its cancel lands, so filling its + // slot now would pay 2 indexers for it; its replacement is queued once it can't be. + let held = indexer_id(0x20); + let candidates = [held, indexer_id(0x21), indexer_id(0x22), indexer_id(0x23)]; + let url = |id: IndexerId| Url::parse(&format!("https://{id}.example")).unwrap(); + let mut snapshot = indexer_urls::Snapshot::new(); + for id in candidates { + snapshot.insert(id, url(id)); + } + let queue = MockQueue::default(); + let ctx = build_ctx( + MockRegistry { + held: vec![held], + ..MockRegistry::default() + }, + MockIisa { + selected: candidates.iter().copied().map(selected).collect(), + }, + queue.clone(), + MockChainClient::default(), + CapturingEventsProducer::new(), + snapshot, + ); + + handle(ctx, &test_message(3)).await.expect("handler ok"); + + let proposed: std::collections::HashSet = + queue.proposals.lock().unwrap().iter().cloned().collect(); + assert_eq!( + proposed, + std::collections::HashSet::from([url(candidates[1]), url(candidates[2])]), + "2 offers, the first in IISA's order, none to the indexer holding a slot" + ); + } + + #[tokio::test] + async fn never_cancels_an_agreement_for_a_slot_another_holds() { + // Target of 1: the healthy agreement stays, though an abandoned one holds a slot too. + let healthy = indexer_id(0x11); + let chain_client = MockChainClient::default(); + let queue = MockQueue::default(); + let registry = MockRegistry { + active_agreements: vec![agreement(healthy, IndexingAgreementStatus::AcceptedOnChain)], + held: vec![indexer_id(0x20)], + ..MockRegistry::default() + }; + let marked_cancelling = Arc::clone(®istry.marked_cancelling); + let ctx = build_ctx( + registry, + MockIisa { + selected: vec![selected(healthy)], + }, + queue.clone(), + chain_client.clone(), + CapturingEventsProducer::new(), + indexer_urls::Snapshot::new(), + ); + + handle(ctx, &test_message(1)).await.expect("handler ok"); + + assert!(chain_client.cancelled.lock().unwrap().is_empty()); + assert!(marked_cancelling.lock().unwrap().is_empty()); + assert!(queue.proposals.lock().unwrap().is_empty()); + } + #[tokio::test] async fn emits_shortfall_when_coverage_drops_below_target() { // Item 9: an indexer leaves the target group and IISA can't replace it, so @@ -1610,6 +1711,7 @@ mod lifecycle_event_tests { shortfall_active: std::sync::Mutex::new(false), marked_cancelled: Arc::default(), marked_cancelling: Arc::default(), + held: Vec::new(), }; let ctx = build_ctx( registry, @@ -1665,6 +1767,7 @@ mod lifecycle_event_tests { shortfall_active: std::sync::Mutex::new(false), marked_cancelled: Arc::default(), marked_cancelling: Arc::default(), + held: Vec::new(), }; let ctx = build_ctx( registry, @@ -1730,6 +1833,7 @@ mod lifecycle_event_tests { shortfall_active: std::sync::Mutex::new(false), marked_cancelled: Arc::default(), marked_cancelling: Arc::default(), + held: Vec::new(), }; let ctx = build_ctx( registry, @@ -1862,6 +1966,7 @@ mod lifecycle_event_tests { shortfall_active: std::sync::Mutex::new(false), marked_cancelled: Arc::default(), marked_cancelling: Arc::default(), + held: Vec::new(), }; let ctx = build_ctx( registry, diff --git a/dipper-pgregistry/src/postgres.rs b/dipper-pgregistry/src/postgres.rs index e9b7411e..254c4b00 100644 --- a/dipper-pgregistry/src/postgres.rs +++ b/dipper-pgregistry/src/postgres.rs @@ -1951,6 +1951,30 @@ impl PgRegistry { .map_err(Into::into) } + /// The indexers of a request's agreements whose indexer stopped serving them that may still + /// be paid: being cancelled, not yet seen ended on-chain, and with no replacement queued. + /// Each keeps its slot in the request until then, so it isn't replaced while both are paid. + pub async fn get_abandoned_indexers_holding_slots( + &self, + request_id: &IndexingRequestId, + ) -> Result, Error> { + let rows: Vec<(PgIndexerId,)> = sqlx::query_as( + r#" + SELECT indexer_id + FROM dipper_reg_indexing_agreements + WHERE indexing_request_id = $1 + AND status = $2 + AND replacement_pending + AND ended_seen_at IS NULL + "#, + ) + .bind(request_id) + .bind(IndexingAgreementStatus::Cancelling) + .fetch_all(&self.pool) + .await?; + Ok(rows.into_iter().map(|(id,)| id.0).collect()) + } + /// Note that an agreement's replacement has been queued. pub async fn mark_replacement_queued( &self, @@ -2029,9 +2053,10 @@ impl PgRegistry { .collect()) } - /// Count `Created` (in-flight, not yet accepted) agreements per indexer, - /// returning the per-indexer map and global total in one round-trip. Offer - /// pacing reads both to gauge spare acceptance capacity before creating more. + /// Count in-flight offers per indexer, returning the per-indexer map and global total in + /// one round-trip. Offer pacing reads both to gauge spare acceptance capacity before + /// creating more. Besides `Created` agreements, a `Cancelling` one never accepted counts + /// until its offer deadline, since its offer may be accepted before its cancel lands. #[expect( clippy::cast_sign_loss, reason = "predates this lint; fix when next touched" @@ -2046,10 +2071,14 @@ impl PgRegistry { SELECT indexer_id, COUNT(*) as count FROM dipper_reg_indexing_agreements WHERE status = $1 + OR (status = $2 + AND accepted_at IS NULL + AND CAST(terms->>'deadline' AS bigint) > EXTRACT(EPOCH FROM now())) GROUP BY GROUPING SETS ((indexer_id), ()) "#, ) .bind(IndexingAgreementStatus::Created) + .bind(IndexingAgreementStatus::Cancelling) .fetch_all(&self.pool) .await?; diff --git a/dipper-pgregistry/tests/it_registry_postgres.rs b/dipper-pgregistry/tests/it_registry_postgres.rs index d5fd3cef..02b1c05c 100644 --- a/dipper-pgregistry/tests/it_registry_postgres.rs +++ b/dipper-pgregistry/tests/it_registry_postgres.rs @@ -3424,6 +3424,164 @@ async fn count_created_agreements_by_indexer_counts_only_created() { assert_eq!(global, 3, "global counts only the 3 Created rows"); } +#[tokio::test] +async fn count_created_agreements_by_indexer_counts_offers_being_withdrawn() { + //* Given + let (db, _temp_db) = temp_registry_db().await; + let registry = PgRegistry::new(db); + let requested_by = address!("8f8c426f956876325b1e037c6eae9b189952994c"); + let indexer = indexer_id!("cccccccccccccccccccccccccccccccccccccccc"); + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs(); + // 3 agreements being cancelled: an offer still open, one accepted, one past its deadline. + let mut ids = Vec::new(); + for (deployment_id, deadline) in [ + ( + deployment_id!("QmUzRg2HHMpbgf6Q4VHKNDbtBEJnyp5JWCh2gUX9AV6jXv"), + now + 3_600, + ), + ( + deployment_id!("QmXbNL4EMkQ6DAPUcBjYSDXZJdpu1Kb1XkKvNvS8JdT7Hs"), + now + 3_600, + ), + ( + deployment_id!("QmYbNL4EMkQ6DAPUcBjYSDXZJdpu1Kb1XkKvNvS8JdT7Hs"), + now - 60, + ), + ] { + let request_id = match registry + .set_indexing_target_candidates(requested_by, deployment_id, 42161, 3) + .await + .expect("set target candidates") + { + dipper_pgregistry::IndexingRequestSetTargetOutcome::Inserted { id } => id, + other => panic!("unexpected set-target outcome: {other:?}"), + }; + let mut terms = Faker.fake::(); + terms.metadata.subgraph_deployment_id = deployment_id; + terms.deadline = deadline; + let id = registry + .register_new_indexing_agreement(NewAgreementParams { + agreement_id: Faker.fake::(), + nonce_uuid: uuid::Uuid::now_v7(), + request_id, + deployment_id, + indexer_id: indexer, + indexer_url: "http://localhost:8020".parse().expect("Invalid URL"), + terms, + terms_version_hash: None, + }) + .await + .expect("register agreement"); + registry + .mark_indexing_agreement_as_cancelling(&id) + .await + .expect("mark cancelling"); + ids.push(id); + } + registry + .record_accepted_audit(&ids[1], now, "0xacc") + .await + .expect("accept record"); + + //* When + let (per_indexer, global) = registry + .count_created_agreements_by_indexer() + .await + .expect("count in-flight offers"); + + //* Then + assert_eq!( + per_indexer.get(&indexer).copied(), + Some(1), + "only the open offer can still be accepted" + ); + assert_eq!(global, 1); +} + +#[tokio::test] +async fn an_abandoned_agreement_holds_its_slot_while_it_may_still_be_paid() { + //* Given + let (db, _temp_db) = temp_registry_db().await; + let registry = PgRegistry::new(db); + let deployment_id = deployment_id!("QmUzRg2HHMpbgf6Q4VHKNDbtBEJnyp5JWCh2gUX9AV6jXv"); + let request_id = match registry + .set_indexing_target_candidates( + address!("8f8c426f956876325b1e037c6eae9b189952994c"), + deployment_id, + 42161, + 3, + ) + .await + .expect("set target candidates") + { + dipper_pgregistry::IndexingRequestSetTargetOutcome::Inserted { id } => id, + other => panic!("unexpected set-target outcome: {other:?}"), + }; + let mut ids = Vec::new(); + for indexer_id in [ + indexer_id!("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), + indexer_id!("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"), + ] { + let mut terms = Faker.fake::(); + terms.metadata.subgraph_deployment_id = deployment_id; + let id = registry + .register_new_indexing_agreement(NewAgreementParams { + agreement_id: Faker.fake::(), + nonce_uuid: uuid::Uuid::now_v7(), + request_id, + deployment_id, + indexer_id, + indexer_url: "http://localhost:8020".parse().expect("Invalid URL"), + terms, + terms_version_hash: None, + }) + .await + .expect("register agreement"); + registry + .apply_reconciliation(&id, true, None) + .await + .expect("accept"); + ids.push(id); + } + let (abandoned, unwanted) = (ids[0], ids[1]); + registry + .mark_indexing_agreement_as_abandoning(&abandoned) + .await + .expect("abandon"); + registry + .mark_indexing_agreement_as_cancelling(&unwanted) + .await + .expect("cancel"); + let held = || async { + registry + .get_abandoned_indexers_holding_slots(&request_id) + .await + .expect("held slots") + .len() + }; + + //* Then + assert_eq!(held().await, 1, "only the abandoned one holds its slot"); + registry + .record_cancel_check(&abandoned, 0, Some(true)) + .await + .expect("seen ended"); + assert_eq!(held().await, 0, "not once the chain shows it ended"); + registry + .record_cancel_check(&abandoned, 0, Some(false)) + .await + .expect("seen live again"); + assert_eq!(held().await, 1); + registry + .mark_replacement_queued(&abandoned) + .await + .expect("replacement queued"); + assert_eq!(held().await, 0, "not once its replacement is queued"); +} + fn fixture_agreement(prefix: u8) -> IndexingAgreementId { let mut bytes = [0u8; 16]; bytes[0] = prefix; diff --git a/k8s/configmap-example.yaml b/k8s/configmap-example.yaml index 88222ea5..ae114cba 100644 --- a/k8s/configmap-example.yaml +++ b/k8s/configmap-example.yaml @@ -120,7 +120,7 @@ data: "threshold": 840 }, "alerts": { - "slack_webhook_url": "REPLACE_ME_OR_REMOVE_FOR_NO_ALERTS", + "slack_webhook_url": "https://hooks.slack.com/services/REPLACE_ME_OR_REMOVE_FOR_NO_ALERTS", "events": ["agreement_cancel_stuck", "rpc_blocks_refused", "nonce_gap_fill_failed"], "throttle": 900 }