● refactor(status): back NamedAtomicStatus with tokio::sync::watch by zancas · Pull Request #1055 · zingolabs/zaino

zancas · 2026-04-25T00:35:55Z

Subscribe to the chain-index sync loop's NamedAtomicStatus via
tokio::sync::watch so test setup wakes on the first Ready
transition instead of after the next 2 s polling tick. A single
sleep(sync_timings.interval) after rendezvous preserves the
load-bearing teardown alignment at ~500 ms (down from 2 s); that shim
disappears when the indexer gains CancellationToken-based shutdown.
wait_for_indexer_tip is rewired against the same watch with a
tokio::time::timeout budget.

Co-Authored-By: Claude Opus 4.7 (1M context) noreply@anthropic.com

Fixes: #1037

Subscribe to the chain-index sync loop's `NamedAtomicStatus` via `tokio::sync::watch` so test setup wakes on the first `Ready` transition instead of after the next 2 s polling tick. A single `sleep(sync_timings.interval)` after rendezvous preserves the load-bearing teardown alignment at ~500 ms (down from 2 s); that shim disappears when the indexer gains `CancellationToken`-based shutdown. `wait_for_indexer_tip` is rewired against the same watch with a `tokio::time::timeout` budget. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…hanges Add `BlockchainSource::wait_for_change` (default returns a never-resolving future), have the success-path interval sleep `select!` between the timer and that future. MockchainSource overrides `wait_for_change` with a `Notify` pinged from `mine_blocks`, so the chain-index sync loop wakes immediately on a mocked block instead of after the next 500 ms tick. Per-block test latency drops from ~510 ms (one full `interval`) to ~10 ms (sync work only). `sync_blocks_after_startup` ~12 s → ~1–2 s, hitting #1039's "under 2 s" target. Real backends keep the default and pace on the timer unchanged. `NonFinalizedState::initialize` now takes `source.clone()` so the outer `source` survives the inner async block's `.await` and is reachable in the new `select!` arm. Known regression: `get_mempool_stream_for_stale_snapshot` flips from a probabilistic flake (#1037) to a deterministic failure. The chain-index sync loop now wakes immediately on `mine_blocks`, but the mempool serve loop still polls on its 100 ms cadence, so the chain-index/mempool tip-skew window is now always observable. The principled fix (per-subscriber notify + select! in the mempool loop) is the work tracked under #1037 and follows in a TDD'd PR. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…tion #1 Rename `get_mempool_stream_for_stale_snapshot` to `skew_chain_index_ahead_returns_stale_stream` and add a doc-comment identifying it as direction #1 of the #1037 tip-skew suite. The companion test for the mempool-ahead direction follows. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

new `mempool_skew` mod (1:1 with `tests/mempool_skew.rs`); add the companion direction-#2 test alongside. - `mempool_skew::chain_index_ahead_returns_stale_stream` — pre-existing direction-#1 spec, deterministically failing post-Option-2 for the same reason it was a flake before (#1037). - `mempool_skew::mempool_ahead_rejects_fresh_snapshot` — new direction-#2 spec. Drives the source via `mine_blocks_silent` (no chain-index notify) so the mempool's poll observes the new tip first, exposing the symmetric false-negative case where the API rejects its own freshly-issued snapshot. Test-only infrastructure to enable the new test: - `MockchainSource::mine_blocks_silent` advances the active height without firing the change-notify, the only way to put the chain-index *behind* the mempool now that `mine_blocks` wakes the sync loop. - `MempoolSubscriber::mempool_tip` (cfg(test)) clones the `mempool_chain_tip` watch receiver; forwarder on `NodeBackedChainIndexSubscriber::mempool_tip`. - `wait_for_mempool_tip_change` helper in `mempool_skew.rs` waits on that receiver, gating direction #2 to act exactly in the skew window. - `mockchain_tests::wait_for_indexer_tip` widened to `pub(super)` so the sibling mod can reuse it. Direction #2 is expected to fail until the principled fix lands. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Add `tips_converge_within_bounded_time_after_mining` — a 20-sample property test that mines one block per iteration, awaits the chain-index tip via the existing event-driven helper, then samples the mempool's tracked tip immediately and counts iterations where the mempool still has the pre-mining hash. Asserts `lag_count == 0`. Today the chain-index sync loop wakes on `mine_blocks` while the mempool serve loop polls on its own 100 ms cadence, so the mempool lags in the vast majority of iterations and the test fails deterministically. Once the principled fix wakes both subsystems on `mine_blocks`, both converge within a single iteration and `lag_count` is zero. Test-only support: - `MockchainSource::active_block_hash` returns the block hash at the current active height — needed to compare the mempool's tracked tip against the expected post-mining hash. - `wait_for_mempool_tip` helper waits for a specific tip hash on the mempool's watch (companion to the existing `wait_for_mempool_tip_change`); used to resync between iterations so the test isn't chasing stacked updates. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…shness (#1037 part 1/2) `NodeBackedChainIndexSubscriber::get_mempool_stream` previously delegated the snapshot-freshness check to `Mempool`, which compared the snapshot's tip against the mempool serve loop's `mempool_chain_tip`. Because the chain-index sync loop and the mempool serve loop poll the source on independent cadences, the two views diverge for a window after every block — and the API exposed the divergence as caller-visible internal contradictions. Move the freshness check into the chain-index subscriber and compare against the chain-index's *current* non-finalized tip (synchronous `arc-swap` load on `non_finalized_state`, same source of truth as `snapshot_nonfinalized_state`). Pass `None` to `MempoolSubscriber::get_mempool_stream` so the mempool no longer gates on tip — staleness is the chain-index's responsibility. This flips the two skew-direction regression tests added under #1037: - `mempool_skew::chain_index_ahead_returns_stale_stream` (direction #1): FAIL → PASS. After `mine_blocks` + `wait_for_indexer_tip`, the chain-index has the new tip while the mempool still has the old one. The pre-fix check matched the stale snapshot against the mempool's lagging view and accepted it; the new check matches against the chain-index's advanced view and correctly returns `None`. - `mempool_skew::mempool_ahead_rejects_fresh_snapshot` (direction #2): FAIL → PASS. After `mine_blocks_silent` + `wait_for_mempool_tip_change`, the mempool has the new tip while the chain-index still has the old one. The pre-fix check rejected the freshly-issued snapshot because the mempool's tip had moved past it; the new check matches against the chain-index's authoritative (lagging) view and correctly returns `Some`. `mempool_skew::tips_converge_within_bounded_time_after_mining` still fails — the mempool serve loop still polls on its own cadence, so it lags the chain-index in raw timing. That's the next work and lands in part 2/2 (per-subscriber wake on source change via `tokio::sync::broadcast`). The dead `Some(Err(MempoolError::IncorrectChainTip { .. }))` arm in `get_mempool_stream`'s outer match is left in place as defensive code — the mempool can no longer return that error from this call site, but the arm doesn't cost anything and keeps the match exhaustive against future Mempool surface changes. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…art 2/2) Both the chain-index sync loop and the mempool serve loop pace themselves on a fixed-cadence timer; only the chain-index used to wake early on source state change (Option-2 `Notify`). After a `mine_blocks`, the chain-index would converge in the same iteration but the mempool would not pick up the new tip until its 100 ms timer expired. That asymmetric lag was the failure mode behind `mempool_skew::tips_converge_within_bounded_time_after_mining`. Replace `BlockchainSource::wait_for_change` (a single shared never-resolves future) with `change_subscribe` returning `Option<broadcast::Receiver<()>>`. Each subscriber gets its own buffered receiver, so a `mine_blocks` that fires while one subsystem is mid-iteration is preserved on its receiver and consumed on the next `recv().await` — no missed-wake gap. Push-capable sources (the mockchain) override to return `Some`; poll-only sources (real validators) keep the `None` default and fall through to the timer. Fix the actual convergence bug: the post-`send_replace` cool-down at the top of the mempool serve loop's tip-change branch was a bare `tokio::time::sleep(100ms)` not wired to any wake source. After iter-0's mine event the mempool would land in this sleep and stay deaf to the broadcast for the next iter-1 mine — so the test saw 19 / 20 events lag (iter 0 the only exception, since the mempool started in the broadcast-aware bottom `select!` after init). Both the cool-down and the inter-iteration sleep now go through the shared helper. DRY the now-three call sites of the `match change_rx { Some => select!(sleep, recv), None => sleep }` pattern into `source::wait_or_source_change(change_rx.as_mut(), duration)`. The helper is `pub(super)` — only the chain-index sync loop and the mempool serve loop need it. `mempool_skew::tips_converge_within_bounded_time_after_mining`: FAIL (19/20 lags) → PASS (0 lags). The two direction tests (`chain_index_ahead_returns_stale_stream`, `mempool_ahead_rejects_fresh_snapshot`) keep passing — `mine_blocks_silent` still suppresses the broadcast, and the mempool's 100 ms timer alone is enough to put it ahead of the chain-index for direction #2. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

idky137

this is mostly a good. I think the watch-based status subscription and event-driven test waits are an improvement over fixed polling sleeps.

I left three comments:

NamedAtomicStatus::store should use send_if_modified() so duplicate suppression is atomic with the update.
The broadcast docs should describe these notifications as coalescible wakeups, not preserved events, since lagging receivers can skip messages.
I do not think get_mempool_stream(expected_chain_tip) should be replaced with get_mempool_stream(None). We should keep the expected-tip guard and fix the skew handling in the mempool layer rather than bypassing that check.

With those addressed, this looks good to me.

… split merge · esc to interrupt The merge from `dev` (#1065) split `chain_index/source.rs` into a `source/` directory without reconciling this branch's #1037 / mempool_skew work, which had been written against the old layout. Restore the dropped pieces on `source/mockchain_source.rs`: * `mine_blocks_silent` and `active_block_hash` inherent methods, lifted from a5ff985 / 934ce01. * `change_broadcast` field + init in both ctors, `mine_blocks` send on actual advance, and the `change_subscribe` override — the broadcast plumbing 901afc1 added on the old `source.rs`. Without it, `mine_blocks` and `mine_blocks_silent` were behaviourally identical and consumers fell through to the trait default's `None`. Fix `tests/mockchain_tests.rs`: import `BlockchainSource` (t-address RPCs became trait methods in #1065) and `GetAddressDeltasParams` / `Response`. Add a regression test adjacent to the methods (`mod mine_blocks`) pinning the broadcast contract directly, so future drift surfaces at the source instead of as flake in the skew tests. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…commit rule: fix(status): make NamedAtomicStatus::store compare-and-update atomic Replace the load-then-send_replace pattern in `NamedAtomicStatus::store` with `watch::Sender::send_if_modified`, fusing the duplicate-suppression check and the value write under the watch channel's internal lock. The previous implementation read the current value, compared, then called `send_replace`, which opened a TOCTOU window where two concurrent writers could both observe the same stale value and both broadcast a single transition, or where a "duplicate" suppression decision could be made against a snapshot a concurrent writer had already invalidated. `store` now returns `bool` matching `send_if_modified`'s shape (true if the value changed and subscribers were notified). All existing call sites discard the return value; no behaviour change for callers. Addresses item 1 of the #1055 review feedback. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

The Broadcast type wraps a tokio::sync::watch sender, which only retains the most recent value: rapid sends collapse, and a subscriber that hasn't polled between two sends sees only the second one. The previous docs used "broadcasts an update" / "notifies clients of updates" / "listen for state update notifications" language that read as if each call delivered a discrete event, which would mislead a reader writing code that expects to observe every transition. Rewrite the module, type, and method docs to name the primitive (watch), state the contract ("wake up and re-read the map"), and make clear that intermediate values are dropped by the channel. The BroadcastSubscriber doc was previously a copy-paste of the Broadcast doc; replace it with subscriber-appropriate text. Addresses item 2 of the #1055 review feedback. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…ts_replace_sleeps

Extract two protocol-level concepts to `chain_index::`: - `pub(crate) const NON_FINALIZED_DEPTH: u32 = 100` — depth of Zaino's non-finalized state in blocks above the finalized seam. Doc cross-references Zebra's `MAX_BLOCK_REORG_HEIGHT` (99 = `MIN_TRANSPARENT_COINBASE_MATURITY - 1`); preserves Zaino's off-by-one-vs-Zebra choice (one extra block of margin) pending review. - `pub(crate) fn anchor_height(chain_height: u32) -> types::Height` — computes the finalized-DB seam from the chain tip via `saturating_sub(NON_FINALIZED_DEPTH)`. Returns a typed `Height` so callers don't double-wrap. Replace six literal `100`s at protocol-depth sites with `anchor_height(...)`, and rename local variables binding the seam value (`finalised_height`, `seam_height`, `initial_seam_height`) to the `anchor_height` shape. No behavior change. The race in #1126 is still present and unfixed — this only names the concept so the follow-up race-fix commit reads as a single contained change. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

The existing `block_is_evicted_from_nfs_when_finalized_advances_past_it` test exercises the same property — *blocks at the iter's pre-mine anchor should be evicted once the source advances past them* — but its race window depends on scheduler timing, so it can pass while the bug is fully present. The first run that surfaced the race went 2/4 failed; a later run on the same code went 4/4 passed. This commit adds a deterministic reproducer. `MockchainSource` gains a test-only one-shot `arm_one_shot_get_block_hook`: the next time `get_block(Height(trigger))` is invoked, a caller-supplied closure fires before the active-height check, mutating source state visible to the same call's lookup. The test arms it for the height the worker's NFS-sync while loop fetches *first*, then has the hook call `mine_blocks_silent(20)`. Now the worker's iter has already committed to the pre-mine `chain_height` (via `fs.sync_to_height(anchor_height(...))` as a no-op) and entered the NFS while loop — exactly the #1126 window — when the source jumps under it. `mine_blocks_silent` (no broadcast) keeps the worker on its current iter through the publish, so the buggy snapshot is observable by the polling test. The assertion `block at pre-mine anchor is evicted` fails 100% while the race is present. Once the iter is bounded at its committed `chain_height`, the silently-mined blocks land in iter N+1 (which trims to the correct post-mine anchor) and the assertion passes. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

#1126) The chain_index sync worker reads `chain_height` at the top of each iter and uses it as the seam target for `fs.sync_to_height(anchor_height(...))`. The NFS sync that follows used an *unbounded* `while let Some(get_block(...))` loop that kept extending NFS as long as the source returned blocks — so a source advance mid-iter (the validator producing new blocks during this iter's NFS sync) silently absorbed those blocks into the *current* iter's publish, while `update`'s trim used the *iter-start* anchor. The published snapshot then overshot its seam. Fix: pre-fetch the iter's NFS window at the worker, against the same `chain_height`, and pass it to `sync` as a `Vec<Arc<Block>>`. NFS extent is now structurally bounded by the iter's commitment; source advances mid-iter are picked up by iter N+1, which pre-fetches its own window against its own fresh `chain_height`. Changes: - `chain_index.rs`: worker pre-fetches blocks at `anchor_height + 1 ..= chain_height` after `fs.sync_to_height`. Length is `NON_FINALIZED_DEPTH` in steady production, shorter on early chains. - `non_finalised_state.rs::sync` signature becomes `sync(finalized_db, window: Vec<Arc<Block>>)`. The unbounded while-loop becomes a `for block in window`. New helper `NonfinalizedBlockCacheSnapshot::block_is_already_in_nfs` skips entries the snapshot has already absorbed (covers the reset path and the idle-iter case where the window overlaps the current tip). - `handle_reorg`, mid-loop `update`, `handle_nfs_change_listener`, and the final trim+CAS are all unchanged — so non-best-chain tracking and reorg defense are preserved. The proptest's random-branch `MockchainSource::get_block` (which used to drive `handle_reorg` reconciliation across multiple iters) still works the same way: the pre-fetch returns whatever branch the source selects per height, and `sync` reconciles via `handle_reorg` on parent-hash mismatch. Test: - `race_pre_mine_anchor_block_is_evicted_when_source_advances_mid_iter` (added in the previous commit) now passes — iter N pre-fetches against the pre-mine `chain_height`, the hook's silent advance is captured in iter N+1, which trims to the correct post-mine anchor. - `block_is_evicted_from_nfs_when_finalized_advances_past_it` becomes reliable for the same reason. - All other chain_index tests pass. The remaining `shutdown_terminates_ sync_loop_cleanly` flake is #1098 (separate; documented as "papering over" in its introducing commit). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…1098) The sync worker used `status.load() == Closing` as its cancellation signal, checked once at the top of each iter. But the iter body itself unconditionally writes `status.store(Syncing | Ready | RecoverableError)`, overwriting any `Closing` flag set by `shutdown()` that landed *during* the iter. Once overwritten, the worker has no way to learn shutdown was requested until it next reaches the failure-escalation path's `tokio::time::sleep(current_backoff)` — and even then, only by waiting out the full `max_consecutive_failures` ladder (~40 s under default timings) once `fs.shutdown()` makes every subsequent `fs.*` call fail. `shutdown_terminates_sync_loop_cleanly`'s introducing commit (dc53a8a) calls this out explicitly: it papered over the race by picking 500 ms default timings, betting that `shutdown()` would *usually* land in the post-iter sleep where the status check would be reached. Under container/CI load, the iter body often dominates the sleep, and the test panics at the 5 s `handle` timeout. Fix: separate cancellation from operational status. Add `cancel_token: CancellationToken` to `NodeBackedChainIndex`, `cancel_token.cancel()` *before* `fs.shutdown()` in the shutdown path (so the worker exits before its in-flight `fs.*` call can fail), and `tokio::select!` the worker's two sleep points against `cancel_token.cancelled()`: - Post-success `wait_or_source_change` (the dominant idle wait). - Failure-path `tokio::time::sleep(current_backoff)` (so a worker that was already mid-iter when shutdown landed exits in one extra iter, not after ~40 s of backoff). Top-of-loop check switches from `status.load() == Closing` to `cancel_token.is_cancelled()`. Status writes in the iter body are untouched — they remain the operational signal for `status()` readers and `status_subscribe` watchers; they just can no longer hide the shutdown signal, which now travels in its own channel. Pattern matches existing CancellationToken usage in `finalised_state`'s LMDB lifecycle (`db/v1.rs`, `db/v0.rs`, `db.rs`). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…op test shim Follow-up to 99dbbd1 (CancellationToken-based shutdown). Two pieces: 1. Wrap the sync worker's iter body in `tokio::select!` against `cancel_token.cancelled()`, not just the two sleep points. Every await inside the body — `source.get_best_block_height`, mid-loop `fs.sync_to_height`, the pre-fetch RPCs, `non_finalized_state.sync`'s internals — is now a cancellation checkpoint. `biased;` makes the cancellation arm take priority over body completion. All in-flight ops are cancellation-safe: LMDB writes are per-block atomic, ArcSwap CAS swaps complete in a single tick, and local `Vec`/`HashMap` state is scoped to the dropped future. 2. `impl Drop for NodeBackedChainIndex` calls `self.cancel_token.cancel()`. Tests that drop the indexer without calling `shutdown()` — most of them — now signal the worker cooperatively. With body-level cancellation in place, the worker exits at its next await checkpoint instead of racing the runtime teardown over a mid-iter LMDB write. These two together close the outstanding promise from PR #1055: > A single `sleep(sync_timings.interval)` after rendezvous preserves > the load-bearing teardown alignment at ~500 ms (down from 2 s); that > shim disappears when the indexer gains `CancellationToken`-based > shutdown. The harness `tokio::time::sleep(sync_timings.interval).await;` shim in `chain_index/tests.rs::load_with_settings` is removed accordingly, along with its 11-line comment explaining the workaround. Net effect: every test calling `load_test_vectors_and_sync_chain_index*` drops 500 ms (default timings) / 50 ms (fast timings) of unnecessary wall time, and teardown ordering is now driven by the same cancellation channel the explicit shutdown path uses. `shutdown_terminates_sync_loop_cleanly` semantics unchanged from 99dbbd1 — explicit `shutdown()` continues to cancel-then-`fs.shutdown`- then-`mempool.close`. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…review) Addresses idky137's review comment on PR #1055: #1055 (comment) The doc on `BlockchainSource::change_subscribe` (and the matching field comment on `MockchainSource::change_broadcast`) claimed every change- event is preserved on its receiver: > Each subscriber receives every change-event on its own buffered > receiver, so wakes between iterations are preserved (no missed-wake > gap). That overstates the `tokio::sync::broadcast` guarantee. The channel is bounded; a lagging receiver gets `RecvError::Lagged` and misses intermediate messages. The current code is unaffected because every consumer treats the wakeup as a coalescible "source state changed" signal and re-reads source state after each wake — a missed intermediate ping just defers them to the next fixed-cadence timer tick. But the doc text shouldn't promise a lossless event log. Reword both sites to describe broadcast as an independent wakeup path, explicitly call out `RecvError::Lagged`, and document why losing intermediate pings is safe given the consumer's re-read-on-wake discipline. No behavior change. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

… review) Addresses idky137's review on PR #1055: #1055 (comment) PR #1055 fixed the #1037 mempool-tip flake by moving snapshot-freshness validation out of the mempool layer (where the laggy `mempool_chain_tip` caused false rejections) into the chain-index layer (where the authoritative NFS tip is the source of truth). The mempool layer was called with `expected_chain_tip = None`, disabling both the up-front and in-stream tip guards. Problem: the in-stream guard is load-bearing. Without it, a reorg or chain advance landing *while a mempool stream is open* doesn't terminate the stream — the caller keeps receiving txs aligned to a chain tip that no longer matches their snapshot. The chain-index's new up-front check only validates freshness at stream-creation time. Fix: keep the chain-index's up-front authoritative check; restore `expected_chain_tip` passing into `get_mempool_stream`; and in the mempool layer distinguish *transient lag* from *real divergence*: - Remove `get_mempool_stream`'s up-front `IncorrectChainTip` rejection. That was the #1037 trigger — laggy mempool would reject snapshots the chain-index considered fresh. - In `wait_on_mempool_updates`, replace immediate-rejection-on-mismatch with `tokio::time::timeout(MEMPOOL_TIP_CATCHUP, wait_for_tip_match)`. Lag (mempool's ~100 ms poll cycle hasn't seen the source advance yet) converges within one cycle; divergence (caller's snapshot is stale, or a reorg) doesn't converge and falls through to the existing `Syncing → end-stream` path on timeout. `MEMPOOL_TIP_CATCHUP = 500 ms` — 5× the mempool serve loop's poll cadence, so transient lag has comfortable headroom while real divergence is detected within an interactive timeframe. `get_mempool_stream_stale_expected_chain_tip` is updated to pin the new contract: stale tip → stream is created (no synchronous error) → streamer task's wait_on_mempool_updates times out → `Syncing` arm short-circuits the loop → channel closes within `MEMPOOL_TIP_CATCHUP + slack`. The old assertion expected `Err(MempoolError::IncorrectChainTip)` synchronously, which is exactly the path the review fix removes. `MempoolError::IncorrectChainTip` is no longer constructed but the variant remains in the public error surface to avoid a breaking change; the corresponding match arm at `chain_index.rs::get_mempool_stream` is now defensive (unreachable under current code paths). Both can be scrubbed in a follow-up API-cleanup commit if desired. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Follow-up to 91359a8 (mempool tip lag-vs-divergence handling). Now that the contract surface no longer constructs `IncorrectChainTip` — tip mismatch is signalled by stream closure via the `Syncing` arm in `wait_on_mempool_updates`, not by a synchronous error — the variant is dead code with no remaining producers or consumers in the workspace. - `error.rs`: drop the `IncorrectChainTip { expected_chain_tip, current_chain_tip }` variant on `MempoolError` and its arm in the `From<MempoolError> for ChainIndexError` impl. - `chain_index.rs::get_mempool_stream`: drop the dead `Some(Err(MempoolError::IncorrectChainTip { .. })) => None` match arm; the general `Some(Err(e)) => …` arm already handles every remaining `MempoolError` variant. - `CHANGELOG.md` under `### Removed`: document the breaking change. Consumers that pattern-matched on this variant should observe stream closure instead (the `Receiver` drains and returns `None`). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

zancas · 2026-05-20T03:34:47Z

Comments to this point have been addressed and all tests pass locally.

…nge notify The `MockchainSource::change_broadcast: broadcast::Sender<()>` field used `broadcast::channel(16)` to wake subscribers on `mine_blocks`. `broadcast` is the wrong primitive for this use: - It's a multi-subscriber *event log* with a per-subscriber bounded buffer — overcommits to a contract the consumer side doesn't want. Subscribers re-read source state on each wake; they care only about *whether* something changed, not *how many* mine events fired. - The `(16)` buffer size was an arbitrary magic number, recently documented as such after PR #1055 review #3147926393 ("`broadcast` gives each subscriber an independent wakeup path, but the notification stream is not a lossless event log…"). `tokio::sync::watch` is the idiomatic Tokio primitive for "wake multiple subscribers when state has changed since they last looked," which is exactly what the consumers want: - `send_replace(())` always triggers `changed()` on every receiver, regardless of value-equality (no PartialEq check). - Any number of `send_replace` calls between two `changed().await` calls collapse into a single wake by construction — coalescing is the contract, not an accident of buffer sizing. - No capacity to pick; no lag semantics to explain. Renames in this commit: - `MockchainSource::change_broadcast: broadcast::Sender<()>` → `MockchainSource::blocks_received_broadcaster: watch::Sender<()>`. New name reflects what's actually being announced (mine_blocks advanced the active height = the source received new blocks) at the concrete-impl level. The abstract trait method `change_subscribe` stays — it's the API surface for any source-state change, not just block receipt. - `BlockchainSource::change_subscribe` return type: `Option<broadcast::Receiver<()>>` → `Option<watch::Receiver<()>>`. - `wait_or_source_change` parameter type updates; `rx.recv()` → `rx.changed()` inside its `select!`. - The pinning regression test (mockchain_source.rs::mine_blocks:: `mine_blocks_fires_broadcast_silent_does_not` → `mine_blocks_fires_broadcaster_silent_does_not`) rewritten against `watch::Receiver::mark_unchanged()` / `has_changed()`. No behaviour change at the consumer level — `chain_index.rs` sync loop and `mempool.rs` serve loop both used type-inferred local bindings and propagated through unchanged. The doc comments that previously needed careful framing around `RecvError::Lagged` and "not a lossless event log" are simpler now: the type itself documents the contract. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…tive_height helper Per AloeareV's review on PR #1055: #1055 (comment) `mine_blocks_silent` was an almost-verbatim copy of `mine_blocks`, differing only in whether it fired `blocks_received_broadcaster` afterwards. Extract the shared advance logic into a private `advance_active_height(&self, blocks: u32) -> bool` that returns whether the height actually moved; the two public methods become one-liners whose difference is exactly the broadcaster call — matching what the comment on `mine_blocks_silent` already said the distinction was. No behaviour change. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

The function and surrounding variables introduced as `anchor_height` described the height of the finalized-DB seam — i.e., a finalized height. "Anchor" was just a synonym, and a confusing one: in Zcash, "anchor" already has an entrenched cryptographic meaning (the Merkle root of the Sapling/Orchard note-commitment tree at a given block, used by zk-proofs). Picking the same word for the finalization seam was an unforced overload. Rename per protocol-domain accuracy: - `chain_index::anchor_height(chain_height: u32) -> types::Height` → `chain_index::finalized_height`. - All local bindings: `let anchor_height = …`, `validator_anchor_height`, `initial_anchor_height`, `pre_mine_anchor` → corresponding `…finalized_height…` forms. - Test function `race_pre_mine_anchor_block_is_evicted_…` → `race_pre_mine_finalized_height_block_is_evicted_…`. - Doc comments referring to "anchor", "pre-mine anchor", "post-mine anchor", etc. → "finalized height" / "pre-mine finalized height". Skipped: `chain_index::types::db::commitment.rs:78` — `/// Sapling note-commitment tree root (anchor) at this block.` That "anchor" is the cryptographic-anchor sense (NU5/Orchard / Sapling proof anchor) and is correct as written. Renaming it would be wrong. The struct field `ChainIndexSnapshot::StillSyncingFinalizedState { validator_finalized_height }` was already named `validator_finalized_height`; after this rename, its local-binding init can use field-init shorthand (`{ validator_finalized_height }` instead of `{ validator_finalized_height: validator_anchor_height }`). −1 line of redundancy. Pure rename. No behaviour change. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…oor (#1128) The function `finalized_height(chain_height)` returns `chain_height - NON_FINALIZED_DEPTH`, a chain-tip-derived lower bound. The name implies "zaino's authoritative finalized tip," but the value isn't that: zaino's finalized-DB tip (`fs.db_height`) is monotonic (once a block is added it never leaves), while the chain-derived value can drop after a reorg shortens the best-known-chain tip below the prior finalization boundary. The actual finalized tip is therefore `max(fs.db_height, finalized_height_floor(chain_tip))`. Rename to `finalized_height_floor` so the `_floor` suffix cues readers that: - this is the chain-derived *lower bound*, not the monotonic actual tip, - `fs.db_height()` is the right call when you want "what zaino has actually finalized," - the two diverge only after a reorg that shortens the chain, never in steady state. Doc rewritten on the function to spell out the steady-state / divergence semantics, cross-reference #1128 (which tracks the read-routing call-site impact at `ChainIndexSnapshot::StillSyncingFinalizedState`), and frame the chain tip explicitly as the *best-known-chain* tip (since "best chain" overloads with "the network's best chain," which can differ from what the local node has observed). Variable bindings that hold the returned value (`let finalized_height = …`, `validator_finalized_height`, `pre_mine_finalized_height`, etc.) are left as-is — within a single sync iter the value *is* the finalized height zaino is operating against; the floor distinction lives on the function's name and doc where it carries weight for the reader. Pure rename + doc. No behaviour change. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

AloeareV

Partial review: 14 of 19 files read.

AloeareV · 2026-05-20T17:15:10Z

+    /// don't extend the chain. Blocks without a coinbase height (malformed
+    /// input) are not considered already-in-NFS; the caller's parent-hash
+    /// chain check will surface the problem.
+    fn block_is_already_in_nfs(&self, block: &zebra_chain::block::Block) -> bool {


This name will see non-best-chain blocks with a height the same as a block on the best chain as already being in the nfs. This method is only sound to call on blocks in the best chain. If this is only intended to be used for best-chain blocks, can that be made more clear in the name?

AloeareV · 2026-05-20T17:33:15Z

+    /// Sync the NFS to the iter's committed chain tip, trimming to the
+    /// finalised tip.
+    ///
+    /// `window` is the iter's pre-fetched view of the non-finalized region —


Currently, we can start work as soon as the first block is fetched. This delays the start of work until the entire window is fetched. Could we make this a collection of futures? A futures::stream::FuturesOrdered may be the best way...although a Vec<F> may get the job done fine. Not too clear on the advantages of the more specialized type here.

AloeareV · 2026-05-20T17:34:24Z

+            // Window may start below `working_snapshot.best_tip` (when iter
+            // N-1 already advanced past `finalized_height_floor(chain_height)`); skip
+            // those entries.
+            if working_snapshot.block_is_already_in_nfs(&block) {


I think this has the potential to skip blocks not on the best chain, if there's already a block on the best chain with the same height. See comment on the block_is_already_in_nfs method.

AloeareV · 2026-05-20T18:06:42Z

+                    let window_first = finalized_height.0 + 1;
+                    let mut window: Vec<Arc<zebra_chain::block::Block>> =
+                        Vec::with_capacity(chain_height.0.saturating_sub(finalized_height.0) as usize);
+                    for h in window_first..=chain_height.0 {


A couple things.

First of all, this is not reorg-aware. If a reorg happens between two block fetches, we could end up serving some blocks from one chain, and then others from another chain.

(also, out of the scope of this PR, but maybe block fetch could be parallelized).

…1129) `NonfinalizedBlockCacheSnapshot::block_is_already_in_nfs` claims, per its docstring, to answer "is this block already represented in the NFS?" The implementation checks `block.coinbase_height() <= self.best_tip.height`, not membership in `self.blocks`. For a block that *is* in `self.blocks` but at a height above `best_tip.height` — the legitimate shape for a non-best-chain block retained via `handle_nfs_change_listener` → `add_nonbest_block` — the predicate returns false even though the block is in the NFS. Today the bug doesn't bite in production because the only call site (`sync()`'s for-loop ~line 419) feeds the predicate the iter's pre-fetched best-chain window, where the height heuristic happens to give the right answer. The contract is broken, the call-site discipline papers over it, and a refactor that broadens the input domain (or changes the listener-drain order) would silently re-process blocks the predicate said weren't there. Add a unit test under `#[cfg(test)] mod block_is_already_in_nfs_predicate` that: - Builds a snapshot with `best_tip.height = 10`. - Inserts a block at height 50 directly into `snapshot.blocks` (bypassing `heights_to_hashes`, matching `add_nonbest_block`'s shape). - Calls `snapshot.block_is_already_in_nfs(&zebra_block_at_50)` and asserts the answer matches the docstring's contract ("yes — the block is in `self.blocks`"). Test fails with the current implementation. Will pass once `block_is_already_in_nfs` is changed to check by hash, or renamed + documented to reflect the height heuristic — whichever direction #1129 lands. The module docstring spells out the bug and points at #1129 for fix discussion. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

AloeareV · 2026-05-20T19:49:59Z

+
+#[cfg(test)]
+mod block_is_already_in_nfs_predicate {
+    //! Reproducer for [#1129](https://github.com/zingolabs/zaino/issues/1129):


We were talking about this test just as we split. I think there was something about it you are changing, that I didn't fully follow.

zancas · 2026-05-21T14:53:11Z

I am factoring this work into a series of more coherent follow up commits.

Threads a "blocks received at the source" wake channel through the chain-index sync worker so the worker re-syncs within a tick of a new block landing at the source, instead of waiting up to `timings.interval` on its periodic timer. `BlockchainSource` grows a default `subscribe_to_blocks_received() -> Option<watch::Receiver<()>>` returning `None` — poll-only sources (real validators) keep pacing themselves on the timer. A free `source::wait_or_source_change(rx, duration)` helper races the optional receiver against `tokio::time::sleep`, degrading to a plain sleep when `rx` is `None`. The chain-index sync worker subscribes once at task start and uses the helper inside the existing post- success `tokio::select!` (which already races against `cancel_token.cancelled()`); the failure-path backoff sleep is unchanged. `MockchainSource` overrides `subscribe_to_blocks_received` to hand out a receiver from a new `blocks_received_broadcaster: watch::Sender<()>` field. `mine_blocks` fires `send_replace(())` when the height actually moves; a sibling `mine_blocks_silent` exists so race regression tests can advance the active height without waking subscribers. Both are one-liners over a new private `advance_active_height(&self, blocks) -> bool` helper that owns the `fetch_update` logic — the public split is now exactly "does this variant fire the broadcaster?" The `mine_blocks_fires_broadcaster_silent_does_not` unit test pins the contract at the source so future drift (field removed, `send_replace` dropped, override removed) fails locally instead of leaking into higher-level tests. While touching `MockchainSource`, collapse the two constructors: `new` is now a one-line wrapper over `new_with_active_height` with `active_chain_height = tip_height`. The duplicated assertion blocks, `build_txid_index` call, and struct initializer now live in one place — the new fields (`get_block_hook`, `blocks_received_broadcaster`) no longer have to be initialised twice. Test fallout: `get_mempool_stream_for_stale_snapshot` was relying on the chain-index sync loop being *slower* than the mempool's 100 ms serve-loop poll — after `mine_blocks + wait_for_indexer_tip` the chain-index had updated, the mempool's next poll had also fired, and the stale-snapshot rejection was observable in a single shot. With the chain-index now waking immediately on the broadcaster, the test races past the mempool's first post-mine tick. Replace the single `assert!(mempool_stream.is_none())` with a `poll_until` on the same condition; the mempool's 100 ms tick still catches up well inside the 10 s budget. Wiring the mempool serve loop to `subscribe_to_blocks_received` would close the underlying gap and belongs with the mempool tip-skew work in PR #1055. Naming choices come from PR #1055 review threads: - `broadcast(16)` → `watch` for the change-notify transport: #1055 (review) - DRY `mine_blocks{,_silent}` via `advance_active_height`: #1055 (comment) Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…e reproducer from PR zingolabs#1055 Sibling test `block_is_evicted_from_nfs_when_finalized_advances_past_it` checks the same property — blocks at an iter's pre-mine finalized height should be evicted from the NFS once the source advances past them — but relies on calling `mine_blocks` from the test thread and racing the sync worker for the iter-start window. The race fires non-deterministically; CI can pass while the bug is fully present. Port the deterministic reproducer (e91d152 on PR zingolabs#1055's `accelerate_unit_tests_replace_sleeps`): a one-shot hook on `MockchainSource::get_block` that fires before the active-height check on the first `get_block(Height(_))` call, mines 20 blocks from inside the hook, and forces the same call to return a block above the iter's committed `chain_height`. The worker's NFS-sync while loop then extends past its commitment while the iter's `update` trims at the pre-mine finalized height, publishing a snapshot whose lowest block sits below the post-mine seam. The test asserts that block is evicted; it fails every run until the iter is bounded at its committed `chain_height` (commit 52ae3ae on the same PR), which lands separately. Minimum surface needed for the hook only: `get_block_hook` field (`Arc<Mutex<Option<Box<dyn FnOnce + Send + Sync>>>`) on `MockchainSource`, both constructors initialise it, an `arm_one_shot_get_block_hook` setter, and the firing block at the top of `get_block(HashOrHeight::Height(_))`. PR zingolabs#1055's `mine_blocks_silent`, `advance_active_height` refactor, `blocks_received_broadcaster`, `change_subscribe`, and `active_block_hash` are not pulled — they're load-bearing only once the sync worker subscribes to source change-notifications, which is a separate concern. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…ht (zingolabs#1126) The chain_index sync worker reads `chain_height` at the top of each iter and uses `finalized_height_floor(chain_height)` as the seam target for `fs.sync_to_height`. The NFS sync that followed kept an *unbounded* `while let Some(get_block(best_tip+1))` loop: a source advance mid-iter (the validator producing new blocks while this iter is still in NFS sync) silently widened the iter's published snapshot past the seam its `fs.sync_to_height` had already committed to. `update` then trimmed at the iter-start floor, so the block at that floor stuck around in the NFS even though the post-mine floor was now strictly higher. Pass `chain_height` into `non_finalised_state::sync` and break the existing speculative-fetch loop between the `get_block(...)` and the apply when `next_height > chain_height`. The speculative `get_block` still runs (which the zingolabs#1126 race-driver test relies on to fire its one-shot hook), but applying a block above the iter's committed seam is suppressed; iter N+1 reads a fresh `chain_height` from the source and walks the new range normally, with the correct finalised floor for its trim. Minimum-invasive variant of PR zingolabs#1055's window pre-fetch — same bound, no new `Vec`, no `block_is_already_in_nfs` predicate, no rework of "what's already in the NFS." Closes the zingolabs#1126 regression test extracted in the previous commit. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…Token Pulls zingolabs#1055's two cancellation commits (99dbbd1, 02bd686) adapted to this branch. The sync worker previously used `status.load() == StatusType::Closing` as its cancellation signal, checked once at the top of each iter. But the iter body itself unconditionally writes `status.store(Syncing | Ready | RecoverableError)`, overwriting any `Closing` flag set by `shutdown()` that landed *during* the iter — so the worker had no way to learn shutdown was requested until the next failure-escalation `tokio::time::sleep(current_backoff)`, and even then only after walking the full `max_consecutive_failures` ladder once `fs.shutdown()` made every subsequent `fs.*` call fail. Tests that dropped the indexer without calling `shutdown()` papered over this with a load-bearing 2 s setup-poll interval that aimed to park the worker in its post-iter sleep before runtime teardown raced a mid-iter LMDB write. Add `cancel_token: CancellationToken` to `NodeBackedChainIndex`. `shutdown()` fires `cancel_token.cancel()` *before* `fs.shutdown()` so the worker exits via the cancellation arm instead of cycling through the failure backoff. Top-of-iter check switches from `status.load() == Closing` → `cancel_token.is_cancelled()`. Both sleep points (post-success `tokio::time::sleep(timings.interval)`, failure-path `tokio::time::sleep(current_backoff)`) are wrapped in `tokio::select! { biased; _ = cancel_token.cancelled() => return Ok(()), _ = ... => {} }`. The iter body itself is wrapped in the same shape so every await inside — `source.get_best_block_height`, `fs.sync_to_height`, `non_finalized_state.sync` — is a cancellation checkpoint; in-flight ops drop cleanly (LMDB writes are per-block atomic, ArcSwap CAS is single-tick, local Vec/HashMap state is scoped to the dropped future). `impl Drop for NodeBackedChainIndex` fires `self.cancel_token.cancel()` so tests that drop without `shutdown()` get the same cooperative exit. Status writes in the iter body are untouched — they remain the operational signal for `status()` readers; they just can no longer hide the shutdown signal, which now travels in its own channel. Harness fallout: the `load_test_vectors_and_sync_chain_index` setup- poll interval drops from 2 s to 25 ms (matching `_with_timings`). The 11-line "load-bearing for the teardown race" comment is replaced with a record of the structural fix. Skipped from zingolabs#1055: the `wait_or_source_change` rename of the post- success sleep (needs the source-change broadcaster, which is a separate concern), `active_block_hash`, `mine_blocks_silent`, and the watch-backed `NamedAtomicStatus`. The mempool/finalised-state status-flag sites are left for follow-up — this lands the pattern at the chain_index level only. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

zancas added 8 commits April 24, 2026 17:31

Merge branch 'dev' into accelerate_unit_tests_replace_sleeps

858557f

zancas marked this pull request as ready for review April 25, 2026 02:28

zancas requested a review from AloeareV April 25, 2026 02:31

zancas assigned AloeareV Apr 25, 2026

zancas mentioned this pull request Apr 25, 2026

Analyze make chain #1056

Closed

idky137 requested changes Apr 27, 2026

View reviewed changes

Comment thread packages/zaino-state/src/status.rs Outdated

Comment thread packages/zaino-state/src/chain_index/source.rs Outdated

Comment thread packages/zaino-state/src/chain_index.rs Outdated

zancas and others added 6 commits April 27, 2026 09:29

Merge branch 'dev' into accelerate_unit_tests_replace_sleeps

c3bbb18

Merge branch 'dev' into accelerate_unit_tests_replace_sleeps

3e99ca2

Merge branch 'dev' into accelerate_unit_tests_replace_sleeps

96e4a19

Merge branch 'dev' into accelerate_unit_tests_replace_sleeps

038c305

Merge branch 'dev' into accelerate_unit_tests_replace_sleeps

462841e

zancas mentioned this pull request May 14, 2026

v0_to_v1_interrupted migration test can hang on wait_until_ready #1122

Open

zancas added 10 commits May 14, 2026 10:52

Merge remote-tracking branch 'zingolabs/dev' into accelerate_unit_tes…

03845e9

…ts_replace_sleeps

Merge branch 'dev' into accelerate_unit_tests_replace_sleeps

9c2a7d4

fix conflict resolution error

0ec291a

fmt

98d79e7

clippy

9050f44

zancas added 6 commits May 19, 2026 19:21

fmt

d028d6b

AloeareV requested changes May 20, 2026

View reviewed changes

zancas added 2 commits May 20, 2026 09:10

zancas requested a review from idky137 May 20, 2026 16:21

zancas added 2 commits May 20, 2026 09:46

AloeareV requested changes May 20, 2026

View reviewed changes

AloeareV reviewed May 20, 2026

View reviewed changes

zancas marked this pull request as draft May 21, 2026 14:52

This was referenced May 23, 2026

Add wake channel to mock chain source #1138

Merged

this uses a tokio_watch to update a status (NamedAtomicStatus) #1139

Closed

Uh oh!

Conversation

zancas commented Apr 25, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

idky137 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

zancas commented May 20, 2026

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

AloeareV left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

AloeareV May 20, 2026

Choose a reason for hiding this comment

Uh oh!

AloeareV May 20, 2026

Choose a reason for hiding this comment

Uh oh!

AloeareV May 20, 2026

Choose a reason for hiding this comment

Uh oh!

AloeareV May 20, 2026

Choose a reason for hiding this comment

Uh oh!

AloeareV May 20, 2026

Choose a reason for hiding this comment

Uh oh!

zancas commented May 21, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

zancas commented Apr 25, 2026 •

edited

Loading