From fd2b686b1a03425c782aded8e1875375880d54bd Mon Sep 17 00:00:00 2001 From: Shoaib Ahmed Date: Tue, 3 Nov 2026 18:48:49 +0400 Subject: [PATCH 1/2] fix(zeronym): requeue an unknown anchor once, then burn it A unanimous unknown anchor is recorded with the observed height of that flush. A later one burns when the tip is fresh and the observed height is above the mark. Co-authored-by: Cursor --- zeronym/hub/src/batcher.rs | 279 ++++++++++++++++++++++++++++++++++--- zeronym/hub/src/chain.rs | 120 +++++++++++++--- zeronym/hub/src/queue.rs | 237 ++++++++++++++++++++++++++++--- 3 files changed, 574 insertions(+), 62 deletions(-) diff --git a/zeronym/hub/src/batcher.rs b/zeronym/hub/src/batcher.rs index 1d56ac36..5c37fbd5 100644 --- a/zeronym/hub/src/batcher.rs +++ b/zeronym/hub/src/batcher.rs @@ -543,9 +543,10 @@ async fn run_with_poll_interval( /// Publish everything held, all at once, in an unpredictable order. /// /// Returns the achieved batch size, which is the honest measure of the privacy -/// the flush actually delivered. A unanimous `AlreadyMined` or `Rejected` is -/// gone. A split `Rejected` is back in the queue. Public so the cadence is not -/// the only thing that can exercise it: a +/// the flush actually delivered. A unanimous `AlreadyMined` or `Rejected`, and +/// a unanimous `Anchor` once the tip is fresh and the observed height is above +/// the mark, are gone. Anything else is back in the queue. +/// Public so the cadence is not the only thing that can exercise it: a /// flush is the security-critical operation here and it must be testable /// directly. pub async fn flush( @@ -577,9 +578,16 @@ pub async fn flush( let mut rejected = 0usize; let mut sample_failure: Option = None; let mut unplaced = Vec::new(); - // A burn is a unanimous AlreadyMined or Rejected. A split Rejected is put - // back. An accepted or already-known entry stays unburned. + // A burn is a unanimous AlreadyMined or Rejected, or a later unanimous + // Anchor once the tip is fresh and the observed height is above the mark. + // An accepted or already-known entry stays unburned. let mut judged = Vec::new(); + let mut anchor_burns = Vec::new(); + let mut anchors = Vec::new(); + // The verdict is judged against the height the hub held when it asked. + // That is the conservative choice. + let marked_at = tip.observed_height(); + let fresh = !tip.is_stale(); let proven = batch.iter().filter(|e| e.lane == Lane::Proven).count(); let transparent = batch.iter().filter(|e| e.lane == Lane::Transparent).count(); for (i, entry) in batch.into_iter().enumerate() { @@ -593,6 +601,14 @@ pub async fn flush( rejected += 1; judged.push(entry); } + Some(Publish::Anchor) => { + if queue.anchor_due(&entry, marked_at, fresh) { + rejected += 1; + anchor_burns.push(entry); + } else { + anchors.push(entry); + } + } Some(Publish::Retryable { reason }) => { if reason != SPLIT_VERDICT { sample_failure.get_or_insert_with(|| reason.clone()); @@ -606,29 +622,37 @@ pub async fn flush( } } queue.burn(&judged); - - // A transport failure goes back into the queue for the next cadence. This is - // the only place such a failure can be recovered: the shim answered the - // wallet error_code 0 the moment the frame reached the mixnet and keeps no - // record, so once the entry left this queue there is no other copy anywhere - // that anyone will retry. Dropping it because the indexer restarted during - // the flush window would lose the migration outright while the wallet - // believes it was sent. The entry keeps its original expiry; when the - // indexer answers again a stale one gets the node's verdict and leaves. A - // burn is a unanimous AlreadyMined or Rejected. A split Rejected is put - // back. + queue.burn_anchor(&anchor_burns); + // Before requeue: a resend already resident makes requeue drop the + // returning copy, and the mark has to outlive that skip. + queue.note_anchor(&anchors, marked_at); + + // A Retryable goes back into the queue for the next cadence. This is the + // only place such a failure can be recovered: the shim answered the wallet + // error_code 0 the moment the frame reached the mixnet and keeps no record, + // so once the entry left this queue there is no other copy anywhere that + // anyone will retry. Dropping it because the indexer restarted during the + // flush window would lose the migration outright while the wallet believes + // it was sent. The entry keeps its original expiry; when the indexer + // answers again a stale one gets the node's verdict and leaves. A burn is + // a unanimous AlreadyMined or Rejected. A split Rejected is put back. A + // unanimous Anchor is put back. A later one burns when the tip is fresh + // and the observed height is above the mark. // // `observed_height`, not `admission_height`. On a stale hub the recorded // boundary follows the free-running estimate, which runs ahead of a stalled // chain. Judging a new arrival on that clock only refuses it, and the // wallet can rebuild; judging an acknowledged entry on it would drop one - // that may still be mined. + // that may still be mined. Read the height again: publishing can take the + // whole call budget, and expiry is judged on the height after it. + let height = tip.observed_height(); let requeued = queue.requeue( unplaced, - tip.observed_height(), + height, params.flush_interval, params.mining_margin, ); + let anchored = queue.requeue(anchors, height, params.flush_interval, params.mining_margin); // Aggregates only. Never a txid, never a body, never a per-entry // identifier: in a Nitro enclave the tracing output reaches the parent host, @@ -642,11 +666,18 @@ pub async fn flush( achieved_batch_size = achieved, rejected, requeued = requeued.held, + anchor_requeued = anchored.held, dropped_expired = requeued.dropped_expired, dropped_exhausted = requeued.dropped_exhausted, elapsed_ms = started.elapsed().as_millis() as u64, "flush published" ); + if anchored.held > 0 { + tracing::info!( + anchor_requeued = anchored.held, + "unknown anchor held for the next flush" + ); + } if requeued.held > 0 { if let Some(reason) = sample_failure.as_deref() { tracing::warn!( @@ -690,6 +721,13 @@ pub async fn flush( "gave up on migrations that could not be placed; they exist nowhere else" ); } + if anchored.dropped_expired > 0 || anchored.dropped_exhausted > 0 { + tracing::error!( + dropped_expired = anchored.dropped_expired, + dropped_exhausted = anchored.dropped_exhausted, + "gave up on an unknown anchor; it exists nowhere else" + ); + } if achieved <= 1 { // Honest telemetry, not an error. At batch size 1 the anonymity set is @@ -1359,10 +1397,10 @@ mod tests { Answer::Send(-26, "16: bad-txns-orchard-binding-signature-invalid"), true, ), - // The hub leaves anchors unchecked, so an unknown anchor burns too. + // A first unanimous unknown anchor is put back, not burned. ( Answer::Send(-26, "16: bad-txns-orchard-unknown-anchor"), - true, + false, ), // Mined: a copy off the chain buys one window. ( @@ -1465,6 +1503,207 @@ mod tests { assert!(!text.contains(MARK), "{text}"); } + #[tokio::test(flavor = "current_thread")] + async fn a_split_verdict_is_not_logged_as_unreachable() { + let anchor = spawn_mock_indexer( + vec![Answer::Send(-26, "16: bad-txns-orchard-unknown-anchor")], + Duration::ZERO, + ) + .await; + let rejected = spawn_mock_indexer( + vec![Answer::Send( + -26, + "16: bad-txns-orchard-binding-signature-invalid", + )], + Duration::ZERO, + ) + .await; + let chain = Arc::new(ChainClient::new(vec![anchor.addr, rejected.addr], None).unwrap()); + let queue = queue_holding(&[junk(5)]); + let (_guard, buf) = error_log(tracing::Level::WARN); + let _ = flush(&queue, &chain, &test_tip(), BatchParams::default()).await; + let text = String::from_utf8(buf.lock().unwrap().clone()).unwrap(); + assert_eq!(queue.len(), 1); + assert!(!text.contains("could not be reached"), "{text}"); + assert!(!text.contains("binding-signature"), "{text}"); + } + + #[tokio::test] + async fn an_unknown_anchor_requeues_until_the_height_rises_then_burns() { + const PROVEN: &[u8] = include_bytes!("../tests/fixtures/v6_proven_a.bin"); + let anchor = Answer::Send(-26, "16: bad-txns-orchard-unknown-anchor"); + let queue = queue_holding(&[PROVEN.to_vec()]); + + let one = spawn_mock_indexer(vec![anchor], Duration::ZERO).await; + let split = + Arc::new(ChainClient::new(vec![one.addr, unreachable_addr().await], None).unwrap()); + let _ = flush(&queue, &split, &test_tip(), BatchParams::default()).await; + assert_eq!(queue.len(), 1, "a mixed anchor poll does not record"); + assert!(!refused_rejected(queue.admit(PROVEN, TIP, N, MARGIN))); + + let record = |a, b| async move { Arc::new(ChainClient::new(vec![a, b], None).unwrap()) }; + let tip = test_tip(); + let left = spawn_mock_indexer(vec![anchor], Duration::ZERO).await; + let right = spawn_mock_indexer(vec![anchor], Duration::ZERO).await; + let _ = flush( + &queue, + &record(left.addr, right.addr).await, + &tip, + BatchParams::default(), + ) + .await; + assert_eq!(queue.len(), 1, "the first unanimous anchor requeues"); + assert!(!refused_rejected(queue.admit(PROVEN, TIP, N, MARGIN))); + + let left = spawn_mock_indexer(vec![anchor], Duration::ZERO).await; + let right = spawn_mock_indexer(vec![anchor], Duration::ZERO).await; + let _ = flush( + &queue, + &record(left.addr, right.addr).await, + &tip, + BatchParams::default(), + ) + .await; + assert_eq!(queue.len(), 1, "the same observed height does not burn"); + + tip.observe(&[TIP + 1]); + let left = spawn_mock_indexer(vec![anchor], Duration::ZERO).await; + let right = spawn_mock_indexer(vec![anchor], Duration::ZERO).await; + let _ = flush( + &queue, + &record(left.addr, right.addr).await, + &tip, + BatchParams::default(), + ) + .await; + assert!( + queue.is_empty(), + "the unanimous anchor burns once the height has risen" + ); + assert!(refused_rejected(queue.admit(PROVEN, TIP, N, MARGIN))); + } + + #[tokio::test] + async fn a_resend_during_the_recording_flush_does_not_clear_the_mark() { + const PROVEN: &[u8] = include_bytes!("../tests/fixtures/v6_proven_a.bin"); + let anchor = Answer::Send(-26, "16: bad-txns-orchard-unknown-anchor"); + let queue = queue_holding(&[PROVEN.to_vec()]); + + let one = spawn_mock_indexer(vec![anchor], Duration::ZERO).await; + let split = + Arc::new(ChainClient::new(vec![one.addr, unreachable_addr().await], None).unwrap()); + let _ = flush(&queue, &split, &test_tip(), BatchParams::default()).await; + + let left = spawn_mock_indexer(vec![anchor], Duration::from_secs(3)).await; + let right = spawn_mock_indexer(vec![anchor], Duration::from_secs(3)).await; + let both = Arc::new(ChainClient::new(vec![left.addr, right.addr], None).unwrap()); + let tip = test_tip(); + let wait = left.on_send.notified(); + tokio::pin!(wait); + let flushing = flush(&queue, &both, &tip, BatchParams::default()); + tokio::pin!(flushing); + tokio::select! { + () = &mut wait => {} + _ = &mut flushing => panic!("the flush returned before the indexer was asked"), + } + assert!(matches!( + queue.admit(PROVEN, TIP, N, MARGIN), + crate::queue::Admission::Admitted { .. } + )); + flushing.await; + assert_eq!(queue.len(), 1); + + tip.observe(&[TIP + 1]); + let left = spawn_mock_indexer(vec![anchor], Duration::ZERO).await; + let right = spawn_mock_indexer(vec![anchor], Duration::ZERO).await; + let both = Arc::new(ChainClient::new(vec![left.addr, right.addr], None).unwrap()); + let _ = flush(&queue, &both, &tip, BatchParams::default()).await; + assert!(queue.is_empty()); + assert!(refused_rejected(queue.admit(PROVEN, TIP, N, MARGIN))); + } + + #[tokio::test] + async fn a_keyless_unknown_anchor_requeues_once_then_burns() { + let payload = junk(7); + let queue = queue_holding(std::slice::from_ref(&payload)); + let anchor = Answer::Send(-26, "16: bad-txns-orchard-unknown-anchor"); + let mock = spawn_mock_indexer(vec![anchor], Duration::ZERO).await; + let chain = client(mock.addr); + let tip = test_tip(); + let _ = flush(&queue, &chain, &tip, BatchParams::default()).await; + assert_eq!(queue.len(), 1); + assert!(!refused_rejected(queue.admit(&payload, TIP, N, MARGIN))); + tip.observe(&[TIP + 1]); + let _ = flush(&queue, &chain, &tip, BatchParams::default()).await; + assert!(queue.is_empty()); + assert!(refused_rejected(queue.admit(&payload, TIP, N, MARGIN))); + } + + #[tokio::test] + async fn a_stale_tip_does_not_burn_an_anchor_mark() { + let payload = junk(8); + let queue = queue_holding(std::slice::from_ref(&payload)); + let anchor = Answer::Send(-26, "16: bad-txns-orchard-unknown-anchor"); + let mock = spawn_mock_indexer(vec![anchor], Duration::ZERO).await; + let tip = test_tip(); + let _ = flush(&queue, &client(mock.addr), &tip, BatchParams::default()).await; + tip.observe(&[TIP + 1]); + tip.age_last_advance_for_test(TIP_STALE_AFTER + Duration::from_secs(1)); + let _ = flush(&queue, &client(mock.addr), &tip, BatchParams::default()).await; + assert!(tip.is_stale()); + assert_eq!(queue.len(), 1); + assert!(!refused_rejected(queue.admit(&payload, TIP, N, MARGIN))); + } + + #[tokio::test(flavor = "current_thread")] + async fn an_anchor_requeue_beside_an_acceptance_is_not_a_selective_hold_back() { + let mock = spawn_mock_indexer( + vec![ + Answer::Send(0, "txid"), + Answer::Send(-26, "16: bad-txns-orchard-unknown-anchor"), + ], + Duration::ZERO, + ) + .await; + let queue = queue_holding(&[junk(1), junk(2)]); + let (_guard, buf) = error_log(tracing::Level::ERROR); + let _ = flush( + &queue, + &client(mock.addr), + &test_tip(), + BatchParams::default(), + ) + .await; + let text = String::from_utf8(buf.lock().unwrap().clone()).unwrap(); + assert_eq!(queue.len(), 1); + assert!(!text.contains("SELECTIVE HOLD-BACK"), "{text}"); + } + + #[tokio::test(flavor = "current_thread")] + async fn an_anchor_that_expires_logs_the_anchor_give_up() { + let tx = fixture_expiring_at(1024); + let queue = Arc::new(Queue::new()); + let tip = Arc::new(TipTracker::new(TipGuard::Unguarded)); + tip.observe(&[1000]); + assert!(matches!( + queue.admit(&tx, 1000, N, MARGIN), + crate::queue::Admission::Admitted { .. } + )); + tip.observe(&[1020]); + let mock = spawn_mock_indexer( + vec![Answer::Send(-26, "16: bad-txns-orchard-unknown-anchor")], + Duration::ZERO, + ) + .await; + let (guard, buf) = error_log(tracing::Level::ERROR); + let _ = flush(&queue, &client(mock.addr), &tip, BatchParams::default()).await; + drop(guard); + let text = String::from_utf8(buf.lock().unwrap().clone()).unwrap(); + assert!(queue.is_empty()); + assert!(!text.contains("no verdict"), "{text}"); + assert!(text.contains("unknown anchor"), "{text}"); + } + #[tokio::test] async fn a_transport_flavoured_grpc_status_is_held_but_invalid_argument_is_not() { // The same wire (a non-OK gRPC status) carries both kinds of failure; diff --git a/zeronym/hub/src/chain.rs b/zeronym/hub/src/chain.rs index 94f183c0..589226df 100644 --- a/zeronym/hub/src/chain.rs +++ b/zeronym/hub/src/chain.rs @@ -137,18 +137,26 @@ impl std::error::Error for GrpcStatusError {} /// transaction and refused it. `Retryable` means this endpoint did not deliver /// one: the connection was refused or reset, TLS or the gRPC call timed out, /// the endpoint said it was unavailable or overloaded, or it answered something -/// this client could not read. +/// this client could not read. `Anchor` is an unknown anchor. /// /// `AlreadyMined` names a block. `AlreadyKnown` is every other already-known /// reply, a mempool among them, and leaves the proofs unburned so a resend can -/// follow an eviction. +/// follow an eviction. A unanimous `Anchor` is not yet a burn. #[derive(Debug, Clone, PartialEq, Eq)] pub enum Publish { - Accepted { txid: String }, + Accepted { + txid: String, + }, AlreadyKnown, AlreadyMined, - Rejected { reason: String }, - Retryable { reason: String }, + Rejected { + reason: String, + }, + /// A unanimous unknown anchor. Not yet a burn. + Anchor, + Retryable { + reason: String, + }, } /// gRPC INVALID_ARGUMENT and FAILED_PRECONDITION: the two statuses under which @@ -689,8 +697,9 @@ pub(crate) const SPLIT_VERDICT: &str = "split verdict"; /// Any [`Publish::Accepted`] wins. Otherwise any [`Publish::AlreadyKnown`] wins. /// Otherwise a burn is every verdict [`Publish::AlreadyMined`] or /// [`Publish::Rejected`]: [`Publish::AlreadyMined`] if any of them is, otherwise -/// one [`Publish::Rejected`]. Anything else is [`Publish::Retryable`]. An empty -/// vec is [`Publish::Rejected`] with reason `no endpoints configured`. A split +/// one [`Publish::Rejected`]. Otherwise a unanimous [`Publish::Anchor`] stays +/// [`Publish::Anchor`]. Anything else is [`Publish::Retryable`]. An empty vec is +/// [`Publish::Rejected`] with reason `no endpoints configured`. A split /// [`Publish::Retryable`] keeps a transport reason when an endpoint returned /// one, and never a [`Publish::Rejected`] reason. fn best_of(outcomes: Vec) -> Publish { @@ -726,6 +735,12 @@ fn best_of(outcomes: Vec) -> Publish { } return rejected.unwrap_or(Publish::AlreadyMined); } + if outcomes + .iter() + .all(|outcome| matches!(outcome, Publish::Anchor)) + { + return Publish::Anchor; + } Publish::Retryable { reason: outcomes .iter() @@ -820,18 +835,28 @@ const PERMANENT_REJECTIONS: &[&str] = &[ // The inputs are gone. A retry re-offers bytes that can never be valid. "conflict", "already spent", - // An unknown anchor: zebra's spelling, which lacks the `bad txns` prefix - // zcashd's `bad-txns-orchard-unknown-anchor` carries. The one entry here a - // waiting node could change, for a recent root at a node behind the tip. - // Listed on purpose: admission leaves the anchor unchecked, so a proof - // against an anchor the chain never had fails the same way at every node, - // and retried it would hold a proven slot for every attempt. +]; + +/// Both word orders, for each pool. Matched before [`PERMANENT_REJECTIONS`]: +/// zcashd's `bad-txns-orchard-unknown-anchor` folds to a string that contains +/// `orchard unknown anchor` and also `bad txns`. +const UNKNOWN_ANCHORS: &[&str] = &[ "unknown sprout anchor", + "sprout unknown anchor", "unknown sapling anchor", + "sapling unknown anchor", "unknown orchard anchor", + "orchard unknown anchor", "unknown ironwood anchor", + "ironwood unknown anchor", ]; +fn is_unknown_anchor(message: &str) -> bool { + UNKNOWN_ANCHORS + .iter() + .any(|pattern| message.contains(pattern)) +} + fn classify_publish_error(message: &str) -> Publish { // Hyphens folded to spaces before matching. Bitcoin-derived nodes report // these as hyphenated reject reasons (`txn-already-known`) while the longer @@ -842,6 +867,10 @@ fn classify_publish_error(message: &str) -> Publish { // what this component exists to avoid emitting. let m = message.to_ascii_lowercase().replace('-', " "); + if is_unknown_anchor(&m) { + return Publish::Anchor; + } + // A NAMED CONSENSUS FAILURE IS CHECKED FIRST, and the order is load-bearing. // `bad-txns-inputs-duplicate` is a consensus rejection that happens to // contain the substring "duplicate", so with already-known matched first it @@ -1076,12 +1105,6 @@ mod tests { "{message:?}" ); } - assert!(matches!( - classify_publish_error( - "missing transparent output: possible double-spend of OutPoint(..)" - ), - Publish::Retryable { .. } - )); } #[test] @@ -1141,18 +1164,60 @@ mod tests { } #[test] - fn an_unknown_anchor_is_a_rejection_in_both_vocabularies() { + fn an_unknown_anchor_is_its_own_verdict_in_both_word_orders() { for message in [ "16: bad-txns-orchard-unknown-anchor", + "16: bad-txns-sprout-unknown-anchor", + "16: bad-txns-sapling-unknown-anchor", + "16: bad-txns-ironwood-unknown-anchor", "unknown Orchard anchor: Root(..)", "consensus validation: unknown Orchard anchor: Root(..)", "unknown Ironwood anchor: Root(..)", + "unknown sprout anchor", + "unknown sapling anchor", + "orchard unknown anchor", + "sapling unknown anchor", ] { - assert!( - matches!(classify_publish_error(message), Publish::Rejected { .. }), + assert_eq!( + classify_publish_error(message), + Publish::Anchor, "{message:?}" ); } + assert!(matches!( + classify_publish_error("16: bad-txns-orchard-duplicate-nullifier"), + Publish::Rejected { .. } + )); + } + + #[test] + fn an_anchor_beside_a_burn_verdict_is_retried() { + let anchor = || Publish::Anchor; + let rejected = || Publish::Rejected { + reason: "bad-txns".into(), + }; + assert!(matches!( + best_of(vec![anchor(), rejected()]), + Publish::Retryable { .. } + )); + assert!(matches!( + best_of(vec![anchor(), Publish::AlreadyMined]), + Publish::Retryable { .. } + )); + assert!(matches!( + best_of(vec![ + anchor(), + Publish::Retryable { + reason: "down".into() + } + ]), + Publish::Retryable { .. } + )); + assert_eq!(best_of(vec![anchor(), anchor()]), Publish::Anchor); + assert_eq!( + best_of(vec![anchor(), Publish::AlreadyKnown]), + Publish::AlreadyKnown + ); } #[test] @@ -1354,6 +1419,17 @@ mod tests { reason: "connection refused".into(), } ); + assert_eq!( + best_of(vec![ + Publish::Anchor, + Publish::Rejected { + reason: node.into(), + }, + ]), + Publish::Retryable { + reason: SPLIT_VERDICT.into(), + } + ); let outcomes = vec![ Publish::Retryable { reason: "connection refused".into(), diff --git a/zeronym/hub/src/queue.rs b/zeronym/hub/src/queue.rs index 0c9258b2..61208cb3 100644 --- a/zeronym/hub/src/queue.rs +++ b/zeronym/hub/src/queue.rs @@ -33,7 +33,9 @@ //! * **One entry per proof, and one publish per proof.** Entries are keyed on //! their payload hash AND on their bundle keys, so re-signing one proof //! collapses into the held entry. A bundle key is burned on a unanimous -//! `AlreadyMined` or `Rejected`. A split answer is put back. +//! `AlreadyMined` or `Rejected`, or on a later unanimous `Anchor` once the +//! tip is fresh and the observed height is above the mark. A keyless +//! `Anchor` on that path burns the payload hash. //! * **Bytes are the budget, not entries.** Real Orchard migrations are 2 to //! 16 KB, so fixed-size slots would waste most of the budget and hand an //! attacker a cheaper occupancy attack than the memory attack. @@ -109,9 +111,9 @@ pub const TRANSPARENT_LANE_ENTRIES: usize = 112; /// traffic here is close to zero while hub and shim share one parser. pub const UNVERIFIED_LANE_ENTRIES: usize = 16; -/// How many bundle keys the hub remembers having published to a verdict. +/// Cap for the burn set and for the anchor-mark set. /// -/// Oldest first out, about 100 MB at the cap, and lost on restart. +/// Each is oldest first out, about 100 MB at the cap, and lost on restart. pub const MAX_BURNED_KEYS: usize = 1 << 20; /// How many admissions may verify proofs at once. @@ -144,8 +146,10 @@ fn lane_index(lane: Lane) -> usize { /// How many flushes may put one entry back before it is given up on. /// -/// A unanimous `AlreadyMined` or `Rejected` is a burn. A split answer is put -/// back, and this counts those flushes on the entry. A resend starts at zero. +/// A unanimous `AlreadyMined` or `Rejected` is a burn, and so is a later +/// unanimous `Anchor` once the tip is fresh and the observed height is above +/// the mark. A split answer and an `Anchor` that is not due are put back, and +/// this counts those flushes on the entry. A resend starts at zero. /// Eight flushes is a few hours at the production cadence. pub const MAX_REQUEUE_ATTEMPTS: u32 = 8; @@ -259,6 +263,20 @@ struct Inner { /// Bundle key to the payload hash of the entry that holds it. bundle_keys: HashMap<[u8; 32], [u8; 32]>, burned: Burned, + /// Keys seen in a unanimous `Anchor`, with the observed height of that + /// flush. A later unanimous `Anchor` burns when the tip is fresh and the + /// observed height is above the mark. A later `Accepted` does not remove a + /// key. A burn does. + anchored: AnchorMarks, +} + +/// Keys an anchor mark tracks. A keyless entry uses the hash of its bytes. +fn anchor_keys(entry: &Entry) -> Vec<[u8; 32]> { + if entry.bundle_keys.is_empty() { + vec![entry.key] + } else { + entry.bundle_keys.clone() + } } /// Whether `spends` is a TEX second leg of `parent_txid`: exactly one input, @@ -314,6 +332,21 @@ impl Inner { Some(entry) } + fn forget_anchor(&mut self, entry: &Entry) { + for key in anchor_keys(entry) { + self.anchored.forget(&key); + } + } + + fn remember_burn(&mut self, key: [u8; 32]) { + self.burned.insert(key); + if let Some(resident) = self.bundle_keys.get(&key).copied() { + self.remove(&resident); + } else if self.entries.contains_key(&key) { + self.remove(&key); + } + } + /// Whether any of these bundle keys is already held. fn holds_any(&self, bundle_keys: &[[u8; 32]]) -> bool { bundle_keys @@ -371,11 +404,15 @@ impl Inner { txid: self.entries.get(held).and_then(|entry| entry.txid.clone()), }); } - if inspected - .bundle_keys - .iter() - .any(|bundle_key| self.burned.contains(bundle_key)) - { + let burned = if inspected.bundle_keys.is_empty() { + self.burned.contains(key) + } else { + inspected + .bundle_keys + .iter() + .any(|bundle_key| self.burned.contains(bundle_key)) + }; + if burned { return Err(Admission::Refused(Refusal::Rejected)); } @@ -404,7 +441,11 @@ impl Inner { } } -/// Bundle keys this hub has published to a verdict, oldest first out. +/// Oldest first out, capped at [`MAX_BURNED_KEYS`]. +/// +/// The burn set holds bundle keys from a unanimous `AlreadyMined` or +/// `Rejected`, and the keys of an `Anchor` that burns. A keyless `Anchor` +/// burn stores the payload hash. #[derive(Default)] struct Burned { keys: HashSet<[u8; 32]>, @@ -428,6 +469,39 @@ impl Burned { } } +/// Anchor marks. Same FIFO and cap as [`Burned`]. An existing key keeps the +/// height it was first recorded at. +#[derive(Default)] +struct AnchorMarks { + height: HashMap<[u8; 32], u32>, + order: VecDeque<[u8; 32]>, +} + +impl AnchorMarks { + fn height(&self, key: &[u8; 32]) -> Option { + self.height.get(key).copied() + } + + fn insert(&mut self, key: [u8; 32], at: u32) { + if self.height.contains_key(&key) { + return; + } + self.height.insert(key, at); + self.order.push_back(key); + if self.order.len() > MAX_BURNED_KEYS { + if let Some(oldest) = self.order.pop_front() { + self.height.remove(&oldest); + } + } + } + + /// Drop `key` from the map. A consumed key may stay in `order` until it + /// reaches the front, and its eviction drops a later mark for the same key. + fn forget(&mut self, key: &[u8; 32]) { + self.height.remove(key); + } +} + /// What one requeue did with the entries a flush could not place. /// /// Counts only. An operator needs to see that entries are being given up on and @@ -484,6 +558,7 @@ impl Queue { lanes: [0; 3], bundle_keys: HashMap::new(), burned: Burned::default(), + anchored: AnchorMarks::default(), }), max_bytes, draining: std::sync::atomic::AtomicBool::new(false), @@ -665,7 +740,8 @@ impl Queue { /// Burn bundle keys from a unanimous `AlreadyMined` or `Rejected`. /// - /// A split `Rejected` is put back, so it is not passed here. An accepted or + /// The anchor mark for each entry is removed. A keyless entry has no bundle + /// key, so this does not store its payload hash. An accepted or /// already-known entry is not passed here. /// /// A resend admitted while the batch was out holds the same proofs, and the @@ -673,11 +749,50 @@ impl Queue { pub fn burn<'a>(&self, entries: impl IntoIterator) { let mut inner = self.lock(); for entry in entries { + inner.forget_anchor(entry); for bundle_key in &entry.bundle_keys { - inner.burned.insert(*bundle_key); - if let Some(resident) = inner.bundle_keys.get(bundle_key).copied() { - inner.remove(&resident); - } + inner.remember_burn(*bundle_key); + } + } + } + + /// Burn a unanimous `Anchor` that is due. A keyless entry is stored by its + /// payload hash. The anchor mark for each entry is removed. + pub fn burn_anchor<'a>(&self, entries: impl IntoIterator) { + let mut inner = self.lock(); + for entry in entries { + inner.forget_anchor(entry); + for key in anchor_keys(entry) { + inner.remember_burn(key); + } + } + } + + /// Whether a unanimous `Anchor` for `entry` is a burn at `height`. + /// + /// The tip must be fresh, and `height` must be above the observed height + /// stored with a marked key. Any one such key is enough. + pub fn anchor_due(&self, entry: &Entry, height: u32, fresh: bool) -> bool { + if !fresh { + return false; + } + let inner = self.lock(); + anchor_keys(entry).iter().any(|key| { + inner + .anchored + .height(key) + .is_some_and(|marked| height > marked) + }) + } + + /// Record a unanimous `Anchor` at `height`. Called before [`Self::requeue`], + /// so a returning copy dropped because a resend is already resident still + /// leaves its keys here. A key already recorded keeps its height. + pub fn note_anchor<'a>(&self, entries: impl IntoIterator, height: u32) { + let mut inner = self.lock(); + for entry in entries { + for key in anchor_keys(entry) { + inner.anchored.insert(key, height); } } } @@ -713,10 +828,9 @@ impl Queue { } // Burned while this entry was out, through another signing judged // in the same window: the burn set refuses these proofs. - if entry - .bundle_keys + if anchor_keys(&entry) .iter() - .any(|bundle_key| inner.burned.contains(bundle_key)) + .any(|key| inner.burned.contains(key)) { continue; } @@ -739,7 +853,10 @@ impl Queue { } if entry.attempts > MAX_REQUEUE_ATTEMPTS { // Dropped, never burned. A burn is a unanimous AlreadyMined or - // Rejected. A split Rejected is put back. A resend starts clean. + // Rejected, or a later unanimous Anchor once the tip is fresh + // and the observed height is above the mark. A split Rejected + // is put back, and so is an Anchor that is not yet due. A + // resend starts clean. out.dropped_exhausted += 1; continue; } @@ -1111,6 +1228,86 @@ mod tests { assert!(admitted(q.admit(PROVEN_B, 100, N, MARGIN))); } + #[test] + fn any_one_marked_bundle_key_is_the_anchor_repeat() { + let q = Queue::new(); + let marked = [1u8; 32]; + let fresh = [2u8; 32]; + let entry = Entry { + key: [3u8; 32], + txid: None, + expiry: None, + tx_bytes: Zeroizing::new(vec![1]), + attempts: 0, + received_height: 100, + lane: Lane::Proven, + bundle_keys: vec![marked], + transparent_outputs: 0, + has_child: false, + spends: Vec::new(), + }; + q.note_anchor([&entry], 100); + let beside = Entry { + key: [4u8; 32], + bundle_keys: vec![marked, fresh], + tx_bytes: Zeroizing::new(vec![2]), + ..entry + }; + assert!(q.anchor_due(&beside, 101, true)); + let only_fresh = Entry { + key: [5u8; 32], + bundle_keys: vec![fresh], + ..beside + }; + assert!(!q.anchor_due(&only_fresh, 101, true)); + } + + #[test] + fn an_anchor_mark_burns_only_after_the_height_rises_on_a_fresh_tip() { + let q = Queue::new(); + let entry = Entry { + key: [7u8; 32], + txid: None, + expiry: None, + tx_bytes: Zeroizing::new(vec![1]), + attempts: 0, + received_height: 100, + lane: Lane::Proven, + bundle_keys: vec![[8u8; 32]], + transparent_outputs: 0, + has_child: false, + spends: Vec::new(), + }; + q.note_anchor([&entry], 100); + assert!(!q.anchor_due(&entry, 100, true)); + assert!(!q.anchor_due(&entry, 101, false)); + assert!(q.anchor_due(&entry, 101, true)); + q.note_anchor([&entry], 105); + assert!(q.anchor_due(&entry, 101, true)); + } + + #[test] + fn a_burn_removes_the_anchor_mark_and_a_keyless_rejection_does_not_store_the_hash() { + let q = Queue::new(); + let payload = vec![9u8; 64]; + assert!(admitted(q.admit(&payload, 100, N, MARGIN))); + let held = q.drain_shuffled(); + q.note_anchor(&held, 100); + assert!(q.anchor_due(&held[0], 101, true)); + q.burn(&held); + assert!(!q.anchor_due(&held[0], 101, true)); + assert!(admitted(q.admit(&payload, 100, N, MARGIN))); + + let held = q.drain_shuffled(); + q.note_anchor(&held, 100); + q.burn_anchor(&held); + assert!(!q.anchor_due(&held[0], 101, true)); + assert_eq!( + q.admit(&payload, 100, N, MARGIN), + Admission::Refused(Refusal::Rejected) + ); + } + #[test] fn a_copy_of_a_mined_transaction_is_refused_once_every_indexer_has_it() { let revealed = Arc::new(ChainNullifiers::new(2)); From ed17d5cb8f22e3740074ee743b3133c39832a459 Mon Sep 17 00:00:00 2001 From: Shoaib Ahmed Date: Wed, 4 Nov 2026 13:09:29 +0400 Subject: [PATCH 2/2] build(zeronym): re-baseline the hub EXPECTED_SHA256 for the unknown anchor Co-authored-by: Cursor --- zeronym/hub/deploy/EXPECTED_SHA256 | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/zeronym/hub/deploy/EXPECTED_SHA256 b/zeronym/hub/deploy/EXPECTED_SHA256 index 4699d33f..63640f82 100644 --- a/zeronym/hub/deploy/EXPECTED_SHA256 +++ b/zeronym/hub/deploy/EXPECTED_SHA256 @@ -1 +1 @@ -2897b521ee4bdc15868b5cb082c2a18540d23cd9d1816b690f69aed9b1c80d0b +fa789e6b99a6f8f6ccf256e04ba9cd646f80a5600dae9b2f3d69853390c77775