From 432b1c916e9e7118dddfb8401df962d6586c1707 Mon Sep 17 00:00:00 2001 From: brunota20 Date: Mon, 22 Jun 2026 11:30:25 -0300 Subject: [PATCH] feat(ethflow-watcher): cap backoff: retries at MAX_BACKOFF_RETRIES (COW-1083) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The strategy's `apply_submit_retry` previously wrote an empty `backoff:{uid}` marker on every retriable submit failure (including the `TryNextBlock` fallback for unparseable orderbook envelopes). The marker was a presence flag with no payload, so on every supervisor reconnect / engine restart the same dead placement would retry indefinitely — bounded only by log re-delivery frequency. This change persists a per-UID retry count in the marker's value (ASCII `u32`) and upgrades to `dropped:` after `MAX_BACKOFF_RETRIES = 5` consecutive retries. The upgrade emits a Warn-level log line so the operator sees the structural issue (flaky CDN, indexer hiccup, poisoned envelope) rather than silently accumulating retries. ## Changes - `modules/ethflow-watcher/src/strategy.rs`: - New const `MAX_BACKOFF_RETRIES = 5`. - New helper `read_backoff_count` that reads + parses the marker payload; pre-COW-1083 empty markers decode to 0 so previously-set backoff: rows still get a fresh attempt (no premature drop on rollout). - `apply_submit_retry`'s retriable branch now reads the prior count, increments, and either writes the new count or upgrades to `dropped:` (clearing the stale `backoff:`) at the cap. - Cap-upgrade log line carries the retry-count and message: "... after 5 retries on transient/unparseable rejection ...". ## Tests - 19/19 ethflow-watcher tests pass. - New `submit_transient_error_at_cap_upgrades_to_dropped_warn`: seeds `backoff:{uid} = "4"`, triggers a `data: None` rejection (the unparseable case the issue names explicitly), asserts: * `dropped:{uid}` is now set * `backoff:{uid}` is cleared (single outcome marker at rest) * exactly one Warn log line containing "ethflow dropped" + "retries" - New `submit_transient_error_with_legacy_empty_marker_resets_counter`: backwards-compat — a pre-COW-1083 empty `b""` marker is treated as count 0, bumped to "1" on first retry rather than prematurely dropping. Protects in-flight backoffs across the rollout. - Existing `submit_transient_error_writes_backoff_marker_and_returns` extended with an assertion that the first retry persists `backoff:{uid} = "1"`. - `cargo clippy -p ethflow-watcher --all-targets -- -D warnings` clean. ## Why this is M4 Surfaced by jeffersonBastos's PR #55 (M3 mirror) review, thread on `crates/shepherd-sdk/src/cow/error.rs:82`. Latent in normal operation (the host forwards parseable envelopes after COW-1075, so `classify_api_error` returns `Drop` for permanent rejections), but the gap fires when the orderbook returns a non-JSON 4xx body (e.g. an HTML error page from a CDN) or if a future host change accidentally drops the envelope again. Bounded retry semantics close the latent risk without changing the safe-default classification (still `TryNextBlock` on `None` data — that part is explicitly out of scope per the issue). AI-assisted authoring with Claude (Opus 4.7); reviewed end-to-end and validated against the existing ethflow-watcher strategy tests before push. --- modules/ethflow-watcher/src/strategy.rs | 173 +++++++++++++++++++++++- 1 file changed, 168 insertions(+), 5 deletions(-) diff --git a/modules/ethflow-watcher/src/strategy.rs b/modules/ethflow-watcher/src/strategy.rs index 1c34d606..79d09e21 100644 --- a/modules/ethflow-watcher/src/strategy.rs +++ b/modules/ethflow-watcher/src/strategy.rs @@ -302,14 +302,49 @@ fn prior_outcome(host: &H, uid_hex: &str) -> Result(host: &H, err: &HostError, uid_hex: &str) -> Result<(), HostError> { match classify_api_error(err.data.as_deref()) { RetryAction::TryNextBlock | RetryAction::Backoff { .. } => { - host.set(&format!("backoff:{uid_hex}"), b"")?; - host.log( - LogLevel::Warn, - &format!("ethflow backoff {uid_hex} ({}): {}", err.code, err.message), - ); + let prior = read_backoff_count(host, uid_hex)?; + let next = prior + 1; + if next >= MAX_BACKOFF_RETRIES { + // Cap reached. Treat the persistent transient failure + // as terminal so dead placements stop re-arming on + // log re-delivery (COW-1083). + host.set(&format!("dropped:{uid_hex}"), b"")?; + let _ = host.delete(&format!("backoff:{uid_hex}")); + host.log( + LogLevel::Warn, + &format!( + "ethflow dropped {uid_hex} after {next} retries on transient/unparseable rejection ({}): {}", + err.code, err.message, + ), + ); + } else { + host.set( + &format!("backoff:{uid_hex}"), + next.to_string().as_bytes(), + )?; + host.log( + LogLevel::Warn, + &format!( + "ethflow backoff {uid_hex} retry {next}/{MAX_BACKOFF_RETRIES} ({}): {}", + err.code, err.message, + ), + ); + } } RetryAction::Drop => { host.set(&format!("dropped:{uid_hex}"), b"")?; @@ -337,6 +372,25 @@ fn apply_submit_retry(host: &H, err: &HostError, uid_hex: &str) -> Resu Ok(()) } +/// Decode the `backoff:{uid}` marker's counter payload. Pre-COW-1083 +/// markers were written as empty bytes (`b""`); those are treated as +/// zero so previously-set markers still get one fresh retry before +/// the cap kicks in. Garbage values (non-ASCII / non-u32) also reset +/// to zero to keep the strategy live in the face of a manual store +/// edit. +fn read_backoff_count(host: &H, uid_hex: &str) -> Result { + let Some(bytes) = host.get(&format!("backoff:{uid_hex}"))? else { + return Ok(0); + }; + if bytes.is_empty() { + return Ok(0); + } + Ok(std::str::from_utf8(&bytes) + .ok() + .and_then(|s| s.parse::().ok()) + .unwrap_or(0)) +} + /// Does this submit-side failure look like the documented Sepolia-orderbook /// rejection of EthFlow's canonical `validTo = u32::MAX`? The check is /// scoped to the `errorType` string the orderbook returns; the strategy @@ -723,6 +777,115 @@ mod tests { .contains_key(&format!("dropped:{uid}")) ); assert!(host.logging.contains("ethflow backoff")); + // COW-1083: the marker now carries an ASCII counter ("1" for + // the first retry) so subsequent attempts can detect the + // accumulated retry budget. + assert_eq!( + host.store.snapshot().get(&format!("backoff:{uid}")).map(Vec::as_slice), + Some(b"1".as_slice()), + "first retry persists count = 1" + ); + } + + #[test] + fn submit_transient_error_at_cap_upgrades_to_dropped_warn() { + // COW-1083 acceptance: after MAX_BACKOFF_RETRIES consecutive + // transient / unparseable rejections the strategy must Drop + // the UID so it stops re-arming on log re-delivery. The log + // line is Warn — this is the operator's signal that something + // is structurally wrong (a flaky CDN, an indexer hiccup, + // a poisoned envelope) rather than a normal transient. + let host = MockHost::new(); + let event = sample_event_for_decode(); + let (topics, data) = encode_log(&event); + let view = placement_log_view(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data); + let placement = + decode_order_placement(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data).unwrap(); + let uid = programmed_uid(&placement); + + // Seed `backoff:{uid}` at MAX-1 so the next retry trips the + // cap. ASCII bytes mirror the production marker payload. + host.store + .set( + &format!("backoff:{uid}"), + (MAX_BACKOFF_RETRIES - 1).to_string().as_bytes(), + ) + .unwrap(); + + // Unparseable rejection: `data = None` is the case the issue + // names explicitly (host failed to forward the envelope or + // CDN returned non-JSON). `classify_api_error` falls back to + // TryNextBlock here, which is exactly when the counter matters. + host.cow_api.respond(Err(HostError { + domain: "cow-api".into(), + kind: Kind::Internal, + code: 502, + message: "bad gateway".into(), + data: None, + })); + + on_logs(&host, &[view]).unwrap(); + + let snapshot = host.store.snapshot(); + assert!( + snapshot.contains_key(&format!("dropped:{uid}")), + "Nth retry of an unparseable rejection must upgrade to dropped:" + ); + assert!( + !snapshot.contains_key(&format!("backoff:{uid}")), + "terminal dropped: must clear the stale backoff: marker" + ); + let drop_lines: Vec<_> = host + .logging + .lines() + .into_iter() + .filter(|l| l.message.contains("ethflow dropped") && l.message.contains("retries")) + .collect(); + assert_eq!(drop_lines.len(), 1, "exactly one cap-upgrade line"); + assert_eq!( + drop_lines[0].level, + LogLevel::Warn, + "cap upgrade is a Warn — operator signal something is structurally wrong" + ); + } + + #[test] + fn submit_transient_error_with_legacy_empty_marker_resets_counter() { + // Backwards compat: pre-COW-1083 markers were written as + // empty bytes (`b""`). Treat those as count = 0 so a + // single in-flight backoff at upgrade time does not get + // prematurely dropped — the marker gets one fresh attempt, + // which counts as retry 1. + let host = MockHost::new(); + let event = sample_event_for_decode(); + let (topics, data) = encode_log(&event); + let view = placement_log_view(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data); + let placement = + decode_order_placement(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data).unwrap(); + let uid = programmed_uid(&placement); + + host.store.set(&format!("backoff:{uid}"), b"").unwrap(); + + host.cow_api.respond(Err(HostError { + domain: "cow-api".into(), + kind: Kind::Internal, + code: 502, + message: "bad gateway".into(), + data: None, + })); + + on_logs(&host, &[view]).unwrap(); + + let snapshot = host.store.snapshot(); + assert_eq!( + snapshot.get(&format!("backoff:{uid}")).map(Vec::as_slice), + Some(b"1".as_slice()), + "legacy empty marker bumps to count = 1, not premature drop" + ); + assert!( + !snapshot.contains_key(&format!("dropped:{uid}")), + "no upgrade to dropped: on first retry" + ); } #[test]