zakura-state 11.0.0

use super::*;

#[test]
fn cached_finalized_continuation_still_chooses_the_nearest_locator() {
    let (runtime, db, _, path) = reconciled_store_with_finalized_prefix(7);
    let reader = runtime.reader();
    let source = SourceId::from_digest([0xd5; 32]);
    let target = path[5].hash;
    let scope = HeaderWorkAuthority::for_target(&runtime.publisher().snapshot(), target);
    let RetainedPathLeaseOutcome::Acquired(first) = reader
        .acquire_retained_path(source, 7, target, &[path[2].hash], scope)
        .unwrap()
    else {
        panic!("the initial path is available");
    };
    assert!(matches!(
        reader
            .read_retained_path(source, 7, first.lease_id, scope, path[2].hash, 1)
            .unwrap(),
        RetainedPathReadOutcome::Page(_)
    ));
    finalize_serving_test_path(&runtime, &db, &path[3..]);
    let scope = HeaderWorkAuthority::for_target(&runtime.publisher().snapshot(), target);
    let RetainedPathLeaseOutcome::Acquired(next) = reader
        .acquire_retained_path(source, 7, target, &[path[3].hash, path[4].hash], scope)
        .unwrap()
    else {
        panic!("a cache miss can use the nearest finalized locator");
    };
    assert_eq!(next.common_ancestor.hash, path[4].hash);
    let RetainedPathReadOutcome::Page(page) = reader
        .read_retained_path(source, 7, next.lease_id, scope, path[4].hash, 1)
        .unwrap()
    else {
        panic!("the nearest finalized suffix is available");
    };
    assert_eq!(page.headers[0].hash(), target);
    assert!(page.complete);
}

#[test]
fn idle_continuation_indexes_are_bounded_and_cannot_block_new_readers() {
    let (runtime, _db, _, path) = reconciled_store_with_finalized_prefix(6);
    let reader = runtime.reader();
    let target = path[5].hash;
    let scope = HeaderWorkAuthority::for_target(&runtime.publisher().snapshot(), target);
    for marker in 0..=MAX_RETAINED_PATH_LEASES {
        let source = SourceId::from_digest([u8::try_from(marker).unwrap(); 32]);
        let RetainedPathLeaseOutcome::Acquired(lease) = reader
            .acquire_retained_path(source, 7, target, &[path[2].hash], scope)
            .unwrap()
        else {
            panic!("idle continuations cannot consume active capacity");
        };
        let RetainedPathReadOutcome::Page(page) = reader
            .read_retained_path(source, 7, lease.lease_id, scope, path[2].hash, 1)
            .unwrap()
        else {
            panic!("the first page is available");
        };
        assert!(!page.complete);
        let leases = runtime.leases.lock().unwrap();
        assert!(leases.by_peer.is_empty());
        assert!(leases.continuations.len() <= MAX_RETAINED_PATH_LEASES);
    }
    {
        let leases = runtime.leases.lock().unwrap();
        assert_eq!(leases.continuations.len(), MAX_RETAINED_PATH_LEASES);
        assert!(!leases
            .continuations
            .contains_key(&SourceId::from_digest([0; 32])));
    }
    // An evicted index is only a cache miss, even when the request is a continuation.
    // Ordinary slots fill first, and the last short path uses the reserve.
    for marker in 0..MAX_RETAINED_PATH_LEASES {
        let source = SourceId::from_digest([u8::try_from(marker).unwrap(); 32]);
        assert!(matches!(
            reader
                .acquire_retained_path(source, 7, target, &[path[3].hash], scope)
                .unwrap(),
            RetainedPathLeaseOutcome::Acquired(_)
        ));
    }
    assert!(matches!(
        reader
            .acquire_retained_path(
                SourceId::from_digest([0xff; 32]),
                7,
                target,
                &[path[3].hash],
                scope
            )
            .unwrap(),
        RetainedPathLeaseOutcome::CapacityBusy(_)
    ));
}

#[test]
fn continuation_cache_matches_session_target_and_first_locator() {
    let (runtime, _db, _, path) = reconciled_store_with_finalized_prefix(6);
    let reader = runtime.reader();
    let source = SourceId::from_digest([0xd3; 32]);
    let target = path[5].hash;
    let scope = HeaderWorkAuthority::for_target(&runtime.publisher().snapshot(), target);
    for (session, next_target, locators, ancestor) in [
        (8, target, vec![path[3].hash], path[3].hash),
        (7, path[4].hash, vec![path[3].hash], path[3].hash),
        (7, target, vec![path[2].hash, path[3].hash], path[3].hash),
    ] {
        let RetainedPathLeaseOutcome::Acquired(first) = reader
            .acquire_retained_path(source, 7, target, &[path[2].hash], scope)
            .unwrap()
        else {
            panic!("the initial page has capacity");
        };
        let original = runtime
            .leases
            .lock()
            .unwrap()
            .get(source, 7, first.lease_id, Instant::now())
            .unwrap()
            .retained_path;
        assert!(matches!(
            reader
                .read_retained_path(source, 7, first.lease_id, scope, path[2].hash, 1)
                .unwrap(),
            RetainedPathReadOutcome::Page(_)
        ));
        let next_scope =
            HeaderWorkAuthority::for_target(&runtime.publisher().snapshot(), next_target);
        let RetainedPathLeaseOutcome::Acquired(next) = reader
            .acquire_retained_path(source, session, next_target, &locators, next_scope)
            .unwrap()
        else {
            panic!("a mismatched cache entry must fall back to normal acquisition");
        };
        assert_eq!(next.common_ancestor.hash, ancestor);
        let fresh = runtime
            .leases
            .lock()
            .unwrap()
            .get(source, session, next.lease_id, Instant::now())
            .unwrap();
        assert!(!Arc::ptr_eq(&original, &fresh.retained_path));
        assert!(!reader
            .release_retained_path(source, 7, first.lease_id, scope)
            .unwrap());
        assert!(reader
            .release_retained_path(source, session, next.lease_id, next_scope)
            .unwrap());
    }
}

#[test]
fn continuation_index_survives_its_target_moving_to_finalized_storage() {
    let (runtime, db, _, path) = reconciled_store_with_finalized_prefix(7);
    let reader = runtime.reader();
    let source = SourceId::from_digest([0xd4; 32]);
    let target = path[5].hash;
    let scope = HeaderWorkAuthority::for_target(&runtime.publisher().snapshot(), target);
    let RetainedPathLeaseOutcome::Acquired(first) = reader
        .acquire_retained_path(source, 7, target, &[path[2].hash], scope)
        .unwrap()
    else {
        panic!("the retained target is available");
    };
    let original = runtime
        .leases
        .lock()
        .unwrap()
        .get(source, 7, first.lease_id, Instant::now())
        .unwrap()
        .retained_path;
    assert!(matches!(
        reader
            .read_retained_path(source, 7, first.lease_id, scope, path[2].hash, 1)
            .unwrap(),
        RetainedPathReadOutcome::Page(_)
    ));
    finalize_serving_test_path(&runtime, &db, &path[3..]);
    assert!(runtime.store.header_node(target).unwrap().is_none());
    let scope = HeaderWorkAuthority::for_target(&runtime.publisher().snapshot(), target);
    let RetainedPathLeaseOutcome::Acquired(next) = reader
        .acquire_retained_path(source, 7, target, &[path[3].hash], scope)
        .unwrap()
    else {
        panic!("the finalized target can reuse its bounded index");
    };
    let resumed = runtime
        .leases
        .lock()
        .unwrap()
        .get(source, 7, next.lease_id, Instant::now())
        .unwrap();
    assert!(Arc::ptr_eq(&original, &resumed.retained_path));
    let RetainedPathReadOutcome::Page(page) = reader
        .read_retained_path(source, 7, next.lease_id, scope, path[3].hash, 2)
        .unwrap()
    else {
        panic!("the remaining canonical headers are still available");
    };
    assert_eq!(
        page.headers,
        vec![path[4].header.clone(), path[5].header.clone()]
    );
    assert!(page.complete);
    assert!(runtime.leases.lock().unwrap().continuations.is_empty());
}

#[test]
fn continuations_reuse_the_path_index_through_finality_and_late_cleanup() {
    let (runtime, db, _, path) = reconciled_store_with_finalized_prefix(64);
    let reader = runtime.reader();
    let source = SourceId::from_digest([0xd2; 32]);
    let target = path.last().unwrap().hash;
    let mut after = path[2].hash;
    let mut original_path = None;
    let mut previous_lease = None;
    for expected in &path[3..] {
        if expected.height == block::Height(10) {
            finalize_serving_test_path(&runtime, &db, &path[3..20]);
        }
        let scope = HeaderWorkAuthority::for_target(&runtime.publisher().snapshot(), target);
        let RetainedPathLeaseOutcome::Acquired(lease) = reader
            .acquire_retained_path(source, 7, target, &[after], scope)
            .unwrap()
        else {
            panic!("every continuation has capacity");
        };
        let cursor = runtime
            .leases
            .lock()
            .unwrap()
            .get(source, 7, lease.lease_id, Instant::now())
            .unwrap();
        if let Some(original) = &original_path {
            assert!(
                Arc::ptr_eq(original, &cursor.retained_path),
                "continuations must reuse the immutable path index, not rebuild it"
            );
        } else {
            original_path = Some(cursor.retained_path.clone());
        }
        if let Some((old_id, old_scope)) = previous_lease {
            assert!(!reader
                .release_retained_path(source, 7, old_id, old_scope)
                .unwrap());
        }
        let RetainedPathReadOutcome::Page(page) = reader
            .read_retained_path(source, 7, lease.lease_id, scope, after, 1)
            .unwrap()
        else {
            panic!("the exact next page remains available");
        };
        assert_eq!(
            page.headers.as_slice(),
            std::slice::from_ref(&expected.header)
        );
        assert_eq!(page.complete, expected.hash == target);
        assert!(runtime.leases.lock().unwrap().by_peer.is_empty());
        previous_lease = Some((lease.lease_id, scope));
        after = expected.hash;
    }
}

#[test]
fn completed_page_releases_capacity_before_continuation_and_late_cleanup() {
    let (runtime, _db, _, path) = reconciled_store_with_finalized_prefix(5);
    let reader = runtime.reader();
    let source = SourceId::from_digest([0xd1; 32]);
    let target = path[4].hash;
    let scope = HeaderWorkAuthority::for_target(&runtime.publisher().snapshot(), target);
    let RetainedPathLeaseOutcome::Acquired(first) = reader
        .acquire_retained_path(source, 7, target, &[path[2].hash], scope)
        .unwrap()
    else {
        panic!("the first page has capacity");
    };
    let RetainedPathReadOutcome::Page(page) = reader
        .read_retained_path(source, 7, first.lease_id, scope, path[2].hash, 1)
        .unwrap()
    else {
        panic!("the first page remains available");
    };
    assert!(!page.complete);
    let RetainedPathLeaseOutcome::Acquired(second) = reader
        .acquire_retained_path(source, 7, target, &[path[3].hash], scope)
        .unwrap()
    else {
        panic!("the continuation must acquire before cleanup of the first page runs");
    };
    assert_ne!(second.lease_id, first.lease_id);
    assert!(!reader
        .release_retained_path(source, 7, first.lease_id, scope)
        .unwrap());
    assert_eq!(
        reader
            .read_retained_path(source, 7, first.lease_id, scope, path[3].hash, 1)
            .unwrap(),
        RetainedPathReadOutcome::Unavailable,
        "the consumed lease cannot race the new reader",
    );
    let RetainedPathReadOutcome::Page(page) = reader
        .read_retained_path(source, 7, second.lease_id, scope, path[3].hash, 1)
        .unwrap()
    else {
        panic!("late cleanup must not revoke the continuation");
    };
    assert_eq!(page.headers[0].hash(), target);
    assert!(page.complete);
    assert!(runtime.leases.lock().unwrap().by_peer.is_empty());
}

#[test]
fn serving_leases_do_not_block_a_better_work_branch() {
    let (engine_config, anchor, metadata) = fixture();
    let store = HeaderChainStore::new(open(&Config::ephemeral(), engine_config.network()));
    store.initialize(metadata, anchor.clone()).unwrap();
    let (runtime, _) = store.startup(&engine_config).unwrap();
    let reader = runtime.reader();
    let parent = Frontier::new(anchor.height, anchor.hash);
    let validation = reader.validation_context(anchor.hash).unwrap().unwrap();
    let rules = HeaderRules::for_validation_lease(&validation).unwrap();
    let context = TransitionContext {
        config: &engine_config,
        clock: &SystemClock,
        full_state_authority: None,
        retention_references: &[],
    };
    let insert = |headers: &[Arc<block::Header>]| {
        let snapshot = runtime.publisher().snapshot();
        let target = headers.last().unwrap().hash();
        runtime
            .apply(
                TransitionRequest {
                    expected_version: snapshot.state_version,
                    event: TransitionEvent::InsertHeaders(Box::new(InsertHeaders {
                        owner: header_owner(&snapshot, target, 1, 1),
                        source: SourceId::from_digest([0xee; 32]),
                        parent_hash: parent.hash,
                        target_tip_hash: target,
                        completion: TargetCompletion::TargetComplete {
                            common_ancestor: parent,
                        },
                        batch: zakura_header_chain::prepare_headers(
                            HeaderBatchInput::new(headers),
                            parent,
                            &rules,
                            &SystemClock,
                        )
                        .unwrap(),
                        aux: Vec::new(),
                    })),
                },
                &context,
            )
            .unwrap()
    };
    let mut captured_paths = Vec::new();
    for index in 0..zakura_header_chain::MAX_CANDIDATE_TIPS_V1 {
        let marker = u8::try_from(index).unwrap();
        let mut header = *anchor.header;
        header.previous_block_hash = parent.hash;
        header.time += chrono::Duration::seconds(1);
        header.nonce.0[0] = marker;
        let header = Arc::new(header);
        assert_eq!(
            insert(std::slice::from_ref(&header)),
            ApplyResult::Committed
        );
        let source = SourceId::from_digest([marker; 32]);
        let scope = zakura_header_chain::HeaderWorkAuthority::for_target(
            &runtime.publisher().snapshot(),
            header.hash(),
        );
        let RetainedPathLeaseOutcome::Acquired(lease) = reader
            .acquire_retained_path(source, 7, header.hash(), &[parent.hash], scope)
            .unwrap()
        else {
            panic!("every candidate tip fits in the general serving capacity");
        };
        let cursor = runtime
            .leases
            .lock()
            .unwrap()
            .get(source, 7, lease.lease_id, Instant::now())
            .unwrap();
        let captured = reader.capture_path_read(&cursor, 1).unwrap();
        captured_paths.push((cursor, captured));
    }

    let mut first = *anchor.header;
    first.previous_block_hash = parent.hash;
    first.time += chrono::Duration::seconds(1);
    first.nonce.0[0] = 0xff;
    let first = Arc::new(first);
    let mut second = *first;
    second.previous_block_hash = first.hash();
    second.time += chrono::Duration::seconds(1);
    let second = Arc::new(second);
    let target = Frontier::new(block::Height(2), second.hash());
    assert_eq!(
        insert(&[first, second]),
        ApplyResult::Committed,
        "peer serving leases must not veto admission of the better-work branch",
    );
    let snapshot = runtime.publisher().snapshot();
    assert_eq!(snapshot.frontiers.header_best, target);
    assert!(!snapshot.alarms.resource_stalled);

    let (cursor, captured) = captured_paths
        .iter()
        .find(|(cursor, _)| {
            runtime
                .store
                .header_node(cursor.target.hash)
                .unwrap()
                .is_none()
        })
        .expect("retention evicts an old fork to admit the better branch");
    assert_eq!(
        reader
            .read_retained_path(
                cursor.peer,
                7,
                cursor.lease_id,
                cursor.scope,
                parent.hash,
                1
            )
            .unwrap(),
        RetainedPathReadOutcome::Unavailable,
    );
    let (page, _, _) = captured.read_page(cursor, 1).unwrap().unwrap();
    assert_eq!(page.headers[0].hash(), cursor.target.hash);
    assert!(
        page.complete,
        "an already captured page survives pruning coherently"
    );
}

#[test]
fn selected_chain_membership_spans_the_finalized_boundary_and_follows_a_reorg() {
    use zakura_node_services::header_chain::SelectedChainMembership::{
        NotRetained, NotSelected, Selected,
    };
    // Heights 1 through 3 are finalized, and 4 through 6 are retained above them.
    let (runtime, db, genesis, path) = reconciled_store_with_finalized_prefix(6);
    let reader = runtime.reader();
    let membership = |header: &VerifiedHeaderRef| {
        reader
            .selected_chain_membership(header.hash)
            .expect("the membership query reads coherent state")
    };
    let selected = |header: &VerifiedHeaderRef| Selected(Frontier::new(header.height, header.hash));
    for header in std::iter::once(&genesis).chain(&path) {
        assert_eq!(membership(header), selected(header));
    }
    assert_eq!(
        reader
            .selected_chain_membership(block::Hash([0xab; 32]))
            .unwrap(),
        NotRetained
    );

    // Finality moves to height 5, so height 4 now answers from the finalized index.
    finalize_serving_test_path(&runtime, &db, &path[3..5]);
    assert_eq!(
        runtime.publisher().snapshot().frontiers.finalized,
        Frontier::new(path[4].height, path[4].hash)
    );
    assert!(
        runtime
            .transition_engine
            .lock()
            .unwrap()
            .graph()
            .header_node(path[3].hash)
            .is_none(),
        "finality pruned height 4 from the retained graph"
    );
    for header in std::iter::once(&genesis).chain(&path) {
        assert_eq!(membership(header), selected(header));
    }

    // A two-header branch from height 5 has more work and replaces height 6.
    let validation = reader.validation_context(path[4].hash).unwrap().unwrap();
    let rules = HeaderRules::for_validation_lease(&validation).unwrap();
    let parent = Frontier::new(path[4].height, path[4].hash);
    let mut branch = Vec::new();
    let mut previous = path[4].header.clone();
    for marker in [0xf1, 0xf2] {
        let mut header = *previous;
        header.previous_block_hash = previous.hash();
        header.time += chrono::Duration::seconds(1);
        header.nonce.0[31] = marker;
        let header = Arc::new(header);
        branch.push(header.clone());
        previous = header;
    }
    let snapshot = runtime.publisher().snapshot();
    let target = previous.hash();
    assert_eq!(
        runtime
            .apply(
                TransitionRequest {
                    expected_version: snapshot.state_version,
                    event: TransitionEvent::InsertHeaders(Box::new(InsertHeaders {
                        owner: header_owner(&snapshot, target, 1, 1),
                        source: SourceId::from_digest([0xee; 32]),
                        parent_hash: parent.hash,
                        target_tip_hash: target,
                        completion: TargetCompletion::TargetComplete {
                            common_ancestor: parent,
                        },
                        batch: zakura_header_chain::prepare_headers(
                            HeaderBatchInput::new(&branch),
                            parent,
                            &rules,
                            &SystemClock,
                        )
                        .unwrap(),
                        aux: Vec::new(),
                    })),
                },
                &TransitionContext {
                    config: &runtime.config,
                    clock: &SystemClock,
                    full_state_authority: None,
                    retention_references: &[],
                },
            )
            .unwrap(),
        ApplyResult::Committed
    );
    assert_eq!(
        runtime.publisher().snapshot().frontiers.header_best.hash,
        target
    );
    assert_eq!(membership(&path[5]), NotSelected);
    for header in std::iter::once(&genesis).chain(&path[..5]) {
        assert_eq!(membership(header), selected(header));
    }
    for (header, height) in branch.iter().zip([6, 7]) {
        assert_eq!(
            reader.selected_chain_membership(header.hash()).unwrap(),
            Selected(Frontier::new(block::Height(height), header.hash()))
        );
    }
}

#[test]
fn atomic_finality_context_can_use_a_newly_staged_anchor_path() {
    let db_config = Config::ephemeral();
    let (engine_config, anchor, metadata) = fixture();
    let store = HeaderChainStore::new(open(&db_config, engine_config.network()));
    store
        .initialize(metadata, anchor.clone())
        .expect("the empty schema initializes");

    // Stage one node past the retained predecessor span, so the context cap
    // binds whatever difficulty averaging window this build uses.
    let predecessor_span = zakura_header_chain::MAX_POW_PREDECESSOR_CONTEXT_SPAN;
    let staged_len =
        u32::try_from(predecessor_span + 1).expect("the retained predecessor span fits in u32");

    let mut nodes = Vec::new();
    let mut parent = anchor;
    for height in 1..=staged_len {
        let mut header = *parent.header;
        header.previous_block_hash = parent.hash;
        header.time += chrono::Duration::seconds(1);
        header.nonce.0[0] = u8::try_from(height).expect("the staged test path is shorter than 256");
        let header = Arc::new(header);
        let hash = header.hash();
        let node = HeaderNode::from_durable_parts(
            header,
            hash,
            parent.hash,
            block::Height(height),
            parent.block_work,
            parent
                .work_coordinate()
                .checked_add(parent.block_work)
                .expect("the short staged path cannot exhaust cumulative work"),
            HeaderValidationState::Valid,
            Default::default(),
            BodyValidationState::Unknown,
            Vec::new(),
        )
        .expect("the staged node fields are coherent");
        parent = node.clone();
        nodes.push(node);
    }
    let staged: HashMap<_, _> = nodes.iter().map(|node| (node.hash, node)).collect();
    let contexts = authenticated_context_headers(&store, parent.hash, Some(&staged))
        .expect("the atomic batch can authenticate context from its staged node overlay");
    assert_eq!(contexts.len(), predecessor_span);
    assert_eq!(
        contexts.first().map(|context| context.height),
        Some(block::Height(1))
    );
    assert_eq!(
        contexts.last().map(|context| context.height),
        Some(block::Height(staged_len - 1))
    );
    assert_eq!(
        parent.header.previous_block_hash,
        contexts
            .last()
            .expect("the context is nonempty")
            .header
            .hash()
    );
}

#[test]
fn publisher_mirror_stays_absent_until_attachment_then_tracks_commits() {
    let (_, _, metadata) = fixture();
    let initial = metadata.snapshot();
    let publisher = Publisher::new(initial.clone());
    let (mirror_sender, mirror_receiver) = watch::channel(None);

    assert_eq!(*mirror_receiver.borrow(), None);

    publisher.mirror_to(mirror_sender);
    assert_eq!(*mirror_receiver.borrow(), Some(initial.clone()));

    let mut committed = initial;
    committed.state_version = StateVersion::new(2);
    publisher.publish(committed.clone(), TransitionEffect::none());
    assert_eq!(*mirror_receiver.borrow(), Some(committed));
}

#[test]
fn coherent_reader_builds_locator_from_the_durable_selected_projection() {
    let db_config = Config::ephemeral();
    let (engine_config, anchor, metadata) = fixture();
    let store = HeaderChainStore::new(open(&db_config, engine_config.network()));
    store
        .initialize(metadata, anchor.clone())
        .expect("the empty schema initializes");
    let (runtime, _) = store
        .startup(&engine_config)
        .expect("the initialized store audits");

    let reader = runtime.reader();
    let durable = reader
        .selected_locator()
        .expect("the durable selected projection is coherent");
    let committed = reader
        .committed_selected_locator()
        .expect("the committed selected projection is coherent");
    assert_eq!(committed, durable);
    assert_eq!(
        durable.entries(),
        &[Frontier::new(anchor.height, anchor.hash)]
    );
}

#[test]
fn body_refill_snapshot_holds_the_complete_transition_barrier() {
    let db_config = Config::ephemeral();
    let (engine_config, anchor, metadata) = fixture();
    let store = HeaderChainStore::new(open(&db_config, engine_config.network()));
    store
        .initialize(metadata, anchor.clone())
        .expect("the empty schema initializes");
    let (runtime, _) = store
        .startup(&engine_config)
        .expect("the initialized store audits");
    let reader = runtime.reader();
    let cloned_reader = reader.clone();

    assert!(Arc::ptr_eq(&reader.config, &cloned_reader.config));

    let (full_state, selected_projection) = reader
        .with_selected_projection(|| {
            assert!(reader.store.writer.try_lock().is_err());
            assert!(reader.transition_engine.try_lock().is_err());
            Frontier::new(anchor.height, anchor.hash)
        })
        .expect("the body refill snapshot is coherent");

    assert_eq!(full_state, Frontier::new(anchor.height, anchor.hash));
    assert_eq!(selected_projection, vec![full_state]);
}

#[test]
fn body_refill_does_not_attach_committed_sizes_to_another_fork() {
    use crate::arbitrary::Prepare;
    use crate::tests::setup::{new_state_with_mainnet_genesis, transaction_v4_from_coinbase};
    use zakura_chain::serialization::ZcashSerialize;

    let _init_guard = zakura_test::init();
    let (engine_config, anchor, metadata) = mainnet_fixture();
    let (finalized, mut full_state, _) = new_state_with_mainnet_genesis();
    let store = HeaderChainStore::new(finalized.db.header_chain_disk_db());
    store
        .initialize(metadata, anchor.clone())
        .expect("header schema initializes");

    let mut old: block::Block = zakura_test::vectors::BLOCK_MAINNET_1_BYTES
        .zcash_deserialize_into()
        .expect("height one parses");
    let selected_header = old.header.clone();
    let mut old_header = *old.header;
    old_header.nonce.0[0] ^= 1;
    old.header = Arc::new(old_header);
    // The state fixture requires a modern coinbase transaction version.
    old.transactions[0] = transaction_v4_from_coinbase(&old.transactions[0]).into();
    let old = Arc::new(old);
    full_state
        .commit_new_chain(old.clone().prepare(), &finalized)
        .expect("old branch commits");

    let selected_hash = selected_header.hash();
    let (runtime, _) = store
        .startup_reconciled(
            &engine_config,
            Frontier::new(anchor.height, anchor.hash),
            Vec::new(),
            vec![VerifiedHeaderRef {
                height: block::Height(1),
                hash: selected_hash,
                header: selected_header,
            }],
        )
        .expect("selected fork restores");

    let old_size = u32::try_from(
        old.zcash_serialize_to_vec()
            .expect("old block serializes")
            .len(),
    )
    .expect("test block size fits u32");
    assert_eq!(
        crate::service::read::block_info(
            full_state.best_chain(),
            &finalized.db,
            old.hash().into(),
        )
        .map(|info| info.size()),
        Some(old_size),
        "the old branch has a confirmed size at the same height",
    );
    let metadata = crate::service::missing_block_body_metadata(
        || full_state.best_chain(),
        &finalized.db,
        Some(&runtime.reader()),
        block::Height(2),
        10,
    )
    .expect("fork refill metadata reads");
    assert_eq!(metadata.anchor, Frontier::new(anchor.height, anchor.hash));
    assert_eq!(
        metadata.blocks,
        vec![(block::Height(1), selected_hash, None)],
        "a selected fork must not inherit the other block's confirmed size"
    );
}

#[test]
fn selected_body_window_reads_four_thousand_hashes_in_one_coherent_range() {
    let db_config = Config::ephemeral();
    let (engine_config, anchor, metadata) = fixture();
    let store = HeaderChainStore::new(open(&db_config, engine_config.network()));
    store
        .initialize(metadata, anchor.clone())
        .expect("the empty schema initializes");

    let genesis = VerifiedHeaderRef {
        height: anchor.height,
        hash: anchor.hash,
        header: anchor.header.clone(),
    };
    let mut parent = genesis.clone();
    let mut restored = Vec::new();
    for height in 1_u32..=4_000 {
        let mut header = *parent.header;
        header.previous_block_hash = parent.hash;
        header.time += chrono::Duration::seconds(1);
        header.nonce.0[..4].copy_from_slice(&height.to_le_bytes());
        let header = Arc::new(header);
        let child = VerifiedHeaderRef {
            height: block::Height(height),
            hash: header.hash(),
            header,
        };
        parent = child.clone();
        restored.push(child);
    }

    let (runtime, _) = store
        .startup_reconciled(
            &engine_config,
            Frontier::new(genesis.height, genesis.hash),
            Vec::new(),
            restored.clone(),
        )
        .expect("the genesis-finalized scratch path reconciles");
    let selected = runtime
        .reader()
        .selected_hashes(block::Height(1), 4_000)
        .expect("the full block-sync window is one coherent projection read");

    assert_eq!(selected.len(), 4_000);
    assert_eq!(
        selected.first().copied(),
        Some(Frontier::new(restored[0].height, restored[0].hash))
    );
    assert_eq!(
        selected.last().copied(),
        restored
            .last()
            .map(|header| Frontier::new(header.height, header.hash))
    );
    let snapshot = runtime.publisher().snapshot();
    let owner = zakura_header_chain::BodyWorkAuthority::for_snapshot(&snapshot)
        .bind(5, NonZeroU64::new(6).expect("six is nonzero"));
    let repair = runtime
        .reader()
        .vct_repair_context(owner, block::Height(1))
        .expect("the long selected repair context is coherent")
        .expect("the empty selected range needs repair");
    assert_eq!(repair.selected_header_count(), 4_000);
    assert_eq!(
        repair.request_target(),
        Frontier::new(restored[3_999].height, restored[3_999].hash)
    );
}

/// A reconciled store over a genesis and `path_len` descendant headers.
///
/// The genesis and the first three path headers are finalized and indexed in the canonical
/// finalized columns. Remaining headers sit in the retained graph above the finalized frontier.
/// Returns the runtime, its open database, the genesis header, and the requested path.
fn reconciled_store_with_finalized_prefix(
    path_len: u32,
) -> (
    HeaderChainRuntime,
    DiskDb,
    VerifiedHeaderRef,
    Vec<VerifiedHeaderRef>,
) {
    reconciled_store_with_finality(path_len, 3)
}

fn reconciled_store_with_finality(
    path_len: u32,
    finalized_count: usize,
) -> (
    HeaderChainRuntime,
    DiskDb,
    VerifiedHeaderRef,
    Vec<VerifiedHeaderRef>,
) {
    let db_config = Config::ephemeral();
    let (engine_config, anchor, metadata) = fixture();
    let db = open(&db_config, engine_config.network());
    let store = HeaderChainStore::new(db.clone());
    store
        .initialize(metadata, anchor.clone())
        .expect("the empty schema initializes");

    let genesis = VerifiedHeaderRef {
        height: anchor.height,
        hash: anchor.hash,
        header: anchor.header.clone(),
    };
    let mut path = Vec::new();
    let mut parent = genesis.clone();
    for marker in 1..=path_len {
        let mut header = *parent.header;
        header.previous_block_hash = parent.hash;
        header.time += chrono::Duration::seconds(1);
        header.nonce.0[..4].copy_from_slice(&marker.to_le_bytes());
        let header = Arc::new(header);
        let height = parent
            .height
            .next()
            .expect("the four-header fixture stays in range");
        let hash = header.hash();
        let child = VerifiedHeaderRef {
            height,
            hash,
            header,
        };
        path.push(child.clone());
        parent = child;
    }

    let hash_by_height = db
        .cf_handle("hash_by_height")
        .expect("the finalized hash index exists");
    let height_by_hash = db
        .cf_handle("height_by_hash")
        .expect("the finalized height index exists");
    let block_header_by_height = db
        .cf_handle("block_header_by_height")
        .expect("the finalized header column exists");
    let mut batch = DiskWriteBatch::new();
    for header in std::iter::once(&genesis).chain(path[..finalized_count].iter()) {
        batch.zs_insert(&hash_by_height, header.height, header.hash);
        batch.zs_insert(&height_by_hash, header.hash, header.height);
        batch.zs_insert(
            &block_header_by_height,
            header.height,
            header.header.as_ref(),
        );
    }
    db.write(batch)
        .expect("the canonical finalized header fixture commits");

    let finalized = Frontier::new(
        path[finalized_count - 1].height,
        path[finalized_count - 1].hash,
    );
    let (runtime, _) = store
        .startup_reconciled(
            &engine_config,
            finalized,
            path[..finalized_count].to_vec(),
            path[finalized_count..].to_vec(),
        )
        .expect("the finalized prefix and retained suffix reconcile");
    (runtime, db, genesis, path)
}

#[test]
fn repair_context_range_is_bounded_by_reclaimable_capacity() {
    let (mut runtime, _db, _genesis, path) = reconciled_store_with_finalized_prefix(6);
    // Even an empty store cannot retain a range larger than its aggregate limit.
    runtime.set_auxiliary_limits_for_test(1, 1);
    let snapshot = runtime.publisher().snapshot();
    let owner = zakura_header_chain::BodyWorkAuthority::for_snapshot(&snapshot)
        .bind(7, NonZeroU64::new(8).unwrap());
    for index in [3, 4, 5] {
        let context = runtime
            .reader()
            .vct_repair_context(owner, path[index].height)
            .unwrap()
            .unwrap();
        assert!(context.admission_capacity_available);
        assert_eq!(context.selected_header_count(), 1);
        assert_eq!(
            context.target,
            Frontier::new(path[index].height, path[index].hash)
        );
    }
}

#[test]
fn repair_context_reconstructs_rejected_input_after_engine_hydration() {
    let (mut runtime, db, _genesis, path) = reconciled_store_with_finalized_prefix(5);
    runtime.set_auxiliary_limits_for_test(1, 3);
    let target = Frontier::new(path[3].height, path[3].hash);
    let snapshot = runtime.publisher().snapshot();
    let owner = zakura_header_chain::BodyWorkAuthority::for_snapshot(&snapshot)
        .bind(7, NonZeroU64::new(8).expect("eight is nonzero"));
    let before = runtime
        .reader()
        .vct_repair_context(owner, target.height)
        .expect("the initial repair context is coherent")
        .expect("the retained selected target needs a repair context");
    let input = zakura_header_chain::TreeAuxRecordV1 {
        height: target.height,
        sapling_root: Default::default(),
        orchard_root: Default::default(),
        ironwood_root: Default::default(),
        sapling_tx_count: 1,
        orchard_tx_count: 2,
        ironwood_tx_count: 3,
        auth_data_root: zakura_chain::block::merkle::AuthDataRoot::from([4; 32]),
    };
    let rejected = AuxDelivery::new(
        EvidenceId::from_digest([0x91; 32]),
        target.hash,
        SourceId::from_digest([0x92; 32]),
        owner.into(),
        zakura_header_chain::BodySizeHint::Unknown,
        Some(input),
    )
    .test_only_with_outcome(2, [Some([0x93; 32]), None], Some(path[4].hash))
    .expect("the rejected auxiliary outcome is coherent");
    let mut target_node = runtime
        .store
        .header_node(target.hash)
        .expect("the selected target row decodes")
        .expect("the selected target remains retained");
    target_node.aux_delivery_ids.push(rejected.delivery_id);
    let mut batch = DiskWriteBatch::new();
    runtime
        .store
        .put_value(
            &mut batch,
            HEADER_NODE_BY_HASH,
            target.hash.0,
            &HeaderNodeDisk::from_domain(&target_node),
        )
        .expect("the selected target with rejection evidence encodes");
    runtime
        .store
        .put_value(
            &mut batch,
            HEADER_AUX_DELIVERY,
            HeaderAuxDeliveryKey {
                header: target.hash,
                delivery: rejected.delivery_id,
            }
            .as_bytes(),
            &rejected,
        )
        .expect("the rejected auxiliary outcome encodes");
    runtime
        .store
        .db
        .write(batch)
        .expect("the durable rejection fixture commits");

    let engine_config = runtime.config.clone();
    drop(runtime);
    let (runtime, _) = HeaderChainStore::new(db)
        .startup(&engine_config)
        .expect("startup reconstructs the durable delivery base and rejection constraints");
    let recovered = runtime
        .reader()
        .vct_repair_context(owner, target.height)
        .expect("the recovered repair context is coherent")
        .expect("the recovered selected target still needs repair");

    assert_ne!(recovered.episode, before.episode);
    assert!(
        recovered.admission_capacity_available,
        "a full recovered input bucket permits a selected replacement"
    );
    assert!(recovered.excludes(input));
    assert!(recovered.retains_payload(input));
    assert!(recovered.retains_source(rejected.source));

    let parent = Frontier::new(path[2].height, path[2].hash);
    let lease = runtime
        .reader()
        .validation_context(parent.hash)
        .expect("the repair parent validation context is coherent")
        .expect("the repair parent remains retained");
    let rules = HeaderRules::for_validation_lease(&lease)
        .expect("the repair parent produces validation rules");
    let headers = [path[3].header.clone()];
    let batch = zakura_header_chain::prepare_headers(
        HeaderBatchInput::new(&headers),
        parent,
        &rules,
        &SystemClock,
    )
    .expect("the replacement header passes deterministic preparation");
    let mut replacement_input = input;
    replacement_input.sapling_tx_count = replacement_input.sapling_tx_count.saturating_add(1);
    let source = SourceId::from_digest([0x94; 32]);
    let replacement = AuxDelivery::new(
        EvidenceId::from_digest([0x95; 32]),
        target.hash,
        source,
        owner.into(),
        zakura_header_chain::BodySizeHint::Unknown,
        Some(replacement_input),
    );
    let repair_request = |episode, delivery: AuxDelivery| TransitionRequest {
        expected_version: StateVersion::default(),
        event: TransitionEvent::InsertHeaders(Box::new(InsertHeaders {
            owner: owner.into(),
            source: delivery.source,
            parent_hash: parent.hash,
            target_tip_hash: target.hash,
            completion: TargetCompletion::SelectedAuxiliaryRepair {
                common_ancestor: parent,
                selected_target: target,
                episode,
            },
            batch: batch.clone(),
            aux: vec![delivery],
        })),
    };
    let context = TransitionContext {
        config: &runtime.config,
        clock: &SystemClock,
        full_state_authority: None,
        retention_references: &[],
    };

    let normal_headers = [path[3].header.clone(), path[4].header.clone()];
    let normal_batch = zakura_header_chain::prepare_headers(
        HeaderBatchInput::new(&normal_headers),
        parent,
        &rules,
        &SystemClock,
    )
    .expect("the duplicate normal path passes deterministic preparation");
    let normal_owner = zakura_header_chain::HeaderWorkAuthority::for_target(
        &runtime.publisher().snapshot(),
        path[4].hash,
    )
    .bind(9, NonZeroU64::new(10).expect("ten is nonzero"));
    let repeated = AuxDelivery::new(
        EvidenceId::from_digest([0x96; 32]),
        target.hash,
        source,
        normal_owner.into(),
        zakura_header_chain::BodySizeHint::Unknown,
        Some(input),
    );
    let successor = AuxDelivery::new(
        EvidenceId::from_digest([0x97; 32]),
        path[4].hash,
        source,
        normal_owner.into(),
        zakura_header_chain::BodySizeHint::Unknown,
        None,
    );
    assert!(matches!(
        runtime
            .apply(
                TransitionRequest {
                    expected_version: StateVersion::default(),
                    event: TransitionEvent::InsertHeaders(Box::new(InsertHeaders {
                        owner: normal_owner.into(),
                        source,
                        parent_hash: parent.hash,
                        target_tip_hash: path[4].hash,
                        completion: TargetCompletion::TargetComplete {
                            common_ancestor: parent,
                        },
                        batch: normal_batch,
                        aux: vec![repeated, successor],
                    })),
                },
                &context,
            )
            .expect("rejected semantic input is a normal stale apply outcome"),
        ApplyResult::Stale(_)
    ));
    assert!(runtime
        .store
        .aux_deliveries(target.hash)
        .expect("the target delivery rows remain readable")
        .iter()
        .all(|delivery| delivery.delivery_id != repeated.delivery_id));

    assert!(matches!(
        runtime
            .apply(repair_request(before.episode, replacement), &context)
            .expect("a stale repair episode is a normal apply outcome"),
        ApplyResult::Stale(_)
    ));
    assert!(runtime
        .store
        .aux_deliveries(target.hash)
        .expect("the target delivery rows remain readable")
        .iter()
        .all(|delivery| delivery.delivery_id != replacement.delivery_id));

    let same_source_replacement = AuxDelivery::new(
        EvidenceId::from_digest([0x98; 32]),
        target.hash,
        rejected.source,
        owner.into(),
        zakura_header_chain::BodySizeHint::Unknown,
        Some(replacement_input),
    );
    assert!(matches!(
        runtime
            .apply(
                repair_request(recovered.episode, same_source_replacement),
                &context,
            )
            .expect("a retained supplier cannot consume another rooted slot"),
        ApplyResult::Stale(_)
    ));
    assert!(runtime
        .store
        .aux_deliveries(target.hash)
        .expect("the target delivery rows remain readable")
        .iter()
        .all(|delivery| delivery.delivery_id != same_source_replacement.delivery_id));
    let duplicate_input = AuxDelivery::new(
        EvidenceId::from_digest([0x99; 32]),
        target.hash,
        source,
        owner.into(),
        zakura_header_chain::BodySizeHint::Unknown,
        Some(input),
    );
    assert!(matches!(
        runtime
            .apply(repair_request(recovered.episode, duplicate_input), &context)
            .expect("a retained semantic input cannot complete the repair again"),
        ApplyResult::Stale(_)
    ));
    assert!(matches!(
        runtime
            .apply(repair_request(recovered.episode, replacement), &context)
            .expect("the current repair episode can apply replacement input"),
        ApplyResult::Committed
    ));
}

#[test]
fn selected_range_repair_rejects_atomically_then_commits_every_delivery() {
    let (runtime, _db, _genesis, path) = reconciled_store_with_finalized_prefix(5);
    let parent = Frontier::new(path[2].height, path[2].hash);
    let targets = [
        Frontier::new(path[3].height, path[3].hash),
        Frontier::new(path[4].height, path[4].hash),
    ];
    let snapshot = runtime.publisher().snapshot();
    let owner = zakura_header_chain::BodyWorkAuthority::for_snapshot(&snapshot)
        .bind(17, NonZeroU64::new(18).expect("eighteen is nonzero"));
    let repair = runtime
        .reader()
        .vct_repair_context(owner, targets[0].height)
        .expect("the range repair context is coherent")
        .expect("the selected empty range needs repair");
    assert_eq!(repair.selected_header_count(), targets.len());
    assert_eq!(repair.request_target(), targets[1]);

    let lease = runtime
        .reader()
        .validation_context(parent.hash)
        .expect("the repair parent validation context is coherent")
        .expect("the repair parent remains retained");
    let rules =
        HeaderRules::for_validation_lease(&lease).expect("the repair parent produces header rules");
    let headers = [path[3].header.clone(), path[4].header.clone()];
    let batch = zakura_header_chain::prepare_headers(
        HeaderBatchInput::new(&headers),
        parent,
        &rules,
        &SystemClock,
    )
    .expect("the selected range passes deterministic preparation");
    let source = SourceId::from_digest([0xa1; 32]);
    let delivery = |index: usize, height| {
        let marker = u8::try_from(index).expect("the delivery index fits u8");
        AuxDelivery::new(
            EvidenceId::from_digest([0xa2_u8.saturating_add(marker); 32]),
            targets[index].hash,
            source,
            owner.into(),
            zakura_header_chain::BodySizeHint::Unknown,
            Some(zakura_header_chain::TreeAuxRecordV1 {
                height,
                sapling_root: Default::default(),
                orchard_root: Default::default(),
                ironwood_root: Default::default(),
                sapling_tx_count: u64::try_from(index).expect("the delivery index fits u64"),
                orchard_tx_count: 0,
                ironwood_tx_count: 0,
                auth_data_root: zakura_chain::block::merkle::AuthDataRoot::from(
                    [0xa4_u8.saturating_add(marker); 32],
                ),
            }),
        )
    };
    let request = |aux| TransitionRequest {
        expected_version: StateVersion::default(),
        event: TransitionEvent::InsertHeaders(Box::new(InsertHeaders {
            owner: owner.into(),
            source,
            parent_hash: parent.hash,
            target_tip_hash: targets[1].hash,
            completion: TargetCompletion::SelectedAuxiliaryRepair {
                common_ancestor: parent,
                selected_target: targets[1],
                episode: repair.episode,
            },
            batch: batch.clone(),
            aux,
        })),
    };
    let context = TransitionContext {
        config: &runtime.config,
        clock: &SystemClock,
        full_state_authority: None,
        retention_references: &[],
    };

    let malformed = request(vec![
        delivery(0, targets[0].height),
        delivery(1, targets[0].height),
    ]);
    assert!(runtime.apply(malformed, &context).is_err());
    for target in targets {
        assert!(runtime
            .store
            .aux_deliveries(target.hash)
            .expect("the target auxiliary rows remain readable")
            .is_empty());
    }

    assert!(matches!(
        runtime
            .apply(
                request(vec![
                    delivery(0, targets[0].height),
                    delivery(1, targets[1].height),
                ]),
                &context,
            )
            .expect("the complete repair range applies"),
        ApplyResult::Committed
    ));
    for target in targets {
        assert_eq!(
            runtime
                .store
                .aux_deliveries(target.hash)
                .expect("the committed target auxiliary rows are readable")
                .len(),
            1
        );
    }
}

/// Store each delivery as durable input on its retained header and refresh the runtime mirror.
///
/// Each delivery must name a distinct header.
fn retain_aux_deliveries_for_test(runtime: &HeaderChainRuntime, deliveries: &[AuxDelivery]) {
    let mut batch = DiskWriteBatch::new();
    for delivery in deliveries {
        let mut node = runtime
            .store
            .header_node(delivery.header_hash)
            .expect("the retained header row decodes")
            .expect("the delivery header is retained");
        node.aux_delivery_ids.push(delivery.delivery_id);
        runtime
            .store
            .put_value(
                &mut batch,
                HEADER_NODE_BY_HASH,
                delivery.header_hash.0,
                &HeaderNodeDisk::from_domain(&node),
            )
            .expect("the header with auxiliary input encodes");
        runtime
            .store
            .put_value(
                &mut batch,
                HEADER_AUX_DELIVERY,
                HeaderAuxDeliveryKey {
                    header: delivery.header_hash,
                    delivery: delivery.delivery_id,
                }
                .as_bytes(),
                delivery,
            )
            .expect("the auxiliary delivery encodes");
    }
    runtime
        .store
        .db
        .write(batch)
        .expect("the auxiliary input fixture commits");
    *runtime
        .transition_engine
        .lock()
        .expect("the transition engine mutex is not poisoned") =
        load_transition_engine(&runtime.store)
            .expect("the direct durable test fixture refreshes the runtime mirror");
}

/// Repair three headers that hold rootless near-tip rows in one range, below one rooted header.
///
/// `aggregate_limit` replaces the aggregate input limit. Returns the durable rows of each
/// repaired target and the number of durable rows in the store.
fn repair_rootless_near_tip_range(
    aggregate_limit: Option<usize>,
) -> (Vec<Vec<AuxDelivery>>, usize) {
    let (mut runtime, _db, _genesis, path) = reconciled_store_with_finalized_prefix(8);
    if let Some(total) = aggregate_limit {
        let per_header = runtime.config.limits.max_aux_deliveries_per_header.get();
        runtime.set_auxiliary_limits_for_test(per_header, total);
    }
    let parent = Frontier::new(path[2].height, path[2].hash);
    let targets: Vec<_> = path[3..6]
        .iter()
        .map(|header| Frontier::new(header.height, header.hash))
        .collect();
    let rooted_header = Frontier::new(path[6].height, path[6].hash);
    let snapshot = runtime.publisher().snapshot();
    let owner = zakura_header_chain::BodyWorkAuthority::for_snapshot(&snapshot)
        .bind(37, NonZeroU64::new(38).expect("thirty-eight is nonzero"));
    let near_tip_owner =
        zakura_header_chain::HeaderWorkAuthority::for_target(&snapshot, path[7].hash)
            .bind(39, NonZeroU64::new(40).expect("forty is nonzero"));
    let near_tip_source = SourceId::from_digest([0xc1; 32]);
    let record = |height: block::Height, marker: u8| zakura_header_chain::TreeAuxRecordV1 {
        height,
        sapling_root: Default::default(),
        orchard_root: Default::default(),
        ironwood_root: Default::default(),
        sapling_tx_count: u64::from(marker),
        orchard_tx_count: 0,
        ironwood_tx_count: 0,
        auth_data_root: zakura_chain::block::merkle::AuthDataRoot::from([marker; 32]),
    };
    let mut near_tip_input: Vec<_> = targets
        .iter()
        .zip(0xc2_u8..)
        .map(|(target, marker)| {
            AuxDelivery::new(
                EvidenceId::from_digest([marker; 32]),
                target.hash,
                near_tip_source,
                near_tip_owner.into(),
                zakura_header_chain::BodySizeHint::Unknown,
                None,
            )
        })
        .collect();
    near_tip_input.push(AuxDelivery::new(
        EvidenceId::from_digest([0xc8; 32]),
        rooted_header.hash,
        near_tip_source,
        near_tip_owner.into(),
        zakura_header_chain::BodySizeHint::Unknown,
        Some(record(rooted_header.height, 0xc9)),
    ));
    retain_aux_deliveries_for_test(&runtime, &near_tip_input);

    let repair = runtime
        .reader()
        .vct_repair_context(owner, targets[0].height)
        .expect("the rootless range repair context is coherent")
        .expect("the rootless selected range needs repair");
    assert_eq!(
        repair.selected_header_count(),
        targets.len(),
        "the range stops before the first rooted row"
    );
    assert_eq!(repair.request_target(), targets[2]);
    assert_eq!(
        runtime
            .reader()
            .vct_repair_context(owner, rooted_header.height)
            .expect("the rooted repair context is coherent")
            .expect("the rooted target remains repairable")
            .selected_header_count(),
        1,
        "rooted input still constrains its replacement to one target"
    );

    let lease = runtime
        .reader()
        .validation_context(parent.hash)
        .expect("the repair parent validation context is coherent")
        .expect("the repair parent remains retained");
    let rules =
        HeaderRules::for_validation_lease(&lease).expect("the repair parent produces header rules");
    let headers: Vec<_> = path[3..6]
        .iter()
        .map(|header| header.header.clone())
        .collect();
    let batch = zakura_header_chain::prepare_headers(
        HeaderBatchInput::new(&headers),
        parent,
        &rules,
        &SystemClock,
    )
    .expect("the selected range passes deterministic preparation");
    let source = SourceId::from_digest([0xca; 32]);
    let replacements = targets
        .iter()
        .zip(0xcb_u8..)
        .map(|(target, marker)| {
            AuxDelivery::new(
                EvidenceId::from_digest([marker; 32]),
                target.hash,
                source,
                owner.into(),
                zakura_header_chain::BodySizeHint::Unknown,
                Some(record(target.height, marker)),
            )
        })
        .collect();
    let context = TransitionContext {
        config: &runtime.config,
        clock: &SystemClock,
        full_state_authority: None,
        retention_references: &[],
    };
    assert!(matches!(
        runtime
            .apply(
                TransitionRequest {
                    expected_version: StateVersion::default(),
                    event: TransitionEvent::InsertHeaders(Box::new(InsertHeaders {
                        owner: owner.into(),
                        source,
                        parent_hash: parent.hash,
                        target_tip_hash: targets[2].hash,
                        completion: TargetCompletion::SelectedAuxiliaryRepair {
                            common_ancestor: parent,
                            selected_target: targets[2],
                            episode: repair.episode,
                        },
                        batch,
                        aux: replacements,
                    })),
                },
                &context,
            )
            .expect("the rootless range repair applies"),
        ApplyResult::Committed
    ));
    let rows = targets
        .iter()
        .map(|target| {
            runtime
                .store
                .aux_deliveries(target.hash)
                .expect("the repaired target rows are readable")
        })
        .collect();
    let total = runtime
        .store
        .load_aux_deliveries()
        .expect("the durable input rows are readable")
        .len();
    (rows, total)
}

#[test]
fn repair_range_spans_rootless_near_tip_input() {
    // Suppliers attach roots only to their finalized prefix, so headers received near the
    // network tip keep rootless rows. When a later handoff moves above those headers, one repair
    // must still cover them all instead of one header per round trip.
    let (rows, _) = repair_rootless_near_tip_range(None);
    for rows in rows {
        assert_eq!(rows.len(), 2, "the repair adds a rooted row");
        assert!(rows.iter().any(|row| row.tree_aux.is_some()));
    }
}

#[test]
fn saturated_rootless_range_repair_retains_every_root() {
    // The four near-tip rows leave one free aggregate slot. The two lowest targets sit in the
    // commit window, so their rows are protected, and the range admits three new roots by
    // reclaiming the two rows above it. Aggregate pressure must evict those older rows, never a
    // root that the range repair just supplied.
    let (rows, total) = repair_rootless_near_tip_range(Some(5));
    assert_eq!(total, 5, "the repair stays within the aggregate limit");
    let repair_source = SourceId::from_digest([0xca; 32]);
    for rows in &rows {
        assert!(
            rows.iter()
                .any(|row| row.source == repair_source && row.tree_aux.is_some()),
            "every target retains its new root"
        );
    }
    assert_eq!(rows[2].len(), 1, "eviction removes the suffix rootless row");
}

#[test]
fn one_header_range_prefix_keeps_its_state_bound_episode() {
    let (runtime, _db, _genesis, path) = reconciled_store_with_finalized_prefix(5);
    let parent = Frontier::new(path[2].height, path[2].hash);
    let target = Frontier::new(path[3].height, path[3].hash);
    let successor = Frontier::new(path[4].height, path[4].hash);
    let snapshot = runtime.publisher().snapshot();
    let owner = zakura_header_chain::BodyWorkAuthority::for_snapshot(&snapshot)
        .bind(27, NonZeroU64::new(28).expect("twenty-eight is nonzero"));
    let range = runtime
        .reader()
        .vct_repair_context(owner, target.height)
        .expect("the range repair context is coherent")
        .expect("the selected empty range needs repair");
    let prefix = range
        .bounded_prefix(1)
        .expect("the one-header range prefix exists");
    let stale_episode = zakura_header_chain::VctRepairContext::from_durable_rows(
        target,
        HeaderLocator::for_continuation(parent),
        StateVersion::new(snapshot.state_version.get().saturating_add(1)),
        Some(successor.hash),
        true,
        &[],
    )
    .expect("the stale exact context is coherent")
    .extend_empty_selected_range(&[], Some(successor.hash))
    .expect("the stale one-header range is coherent")
    .episode;

    let lease = runtime
        .reader()
        .validation_context(parent.hash)
        .expect("the repair parent validation context is coherent")
        .expect("the repair parent remains retained");
    let rules =
        HeaderRules::for_validation_lease(&lease).expect("the repair parent produces header rules");
    let batch = zakura_header_chain::prepare_headers(
        HeaderBatchInput::new(std::slice::from_ref(&path[3].header)),
        parent,
        &rules,
        &SystemClock,
    )
    .expect("the selected prefix passes deterministic preparation");
    let source = SourceId::from_digest([0xb1; 32]);
    let request = |episode| TransitionRequest {
        expected_version: StateVersion::default(),
        event: TransitionEvent::InsertHeaders(Box::new(InsertHeaders {
            owner: owner.into(),
            source,
            parent_hash: parent.hash,
            target_tip_hash: target.hash,
            completion: TargetCompletion::SelectedAuxiliaryRepair {
                common_ancestor: parent,
                selected_target: target,
                episode,
            },
            batch: batch.clone(),
            aux: vec![AuxDelivery::new(
                EvidenceId::from_digest([0xb2; 32]),
                target.hash,
                source,
                owner.into(),
                zakura_header_chain::BodySizeHint::Unknown,
                Some(zakura_header_chain::TreeAuxRecordV1 {
                    height: target.height,
                    sapling_root: Default::default(),
                    orchard_root: Default::default(),
                    ironwood_root: Default::default(),
                    sapling_tx_count: 1,
                    orchard_tx_count: 0,
                    ironwood_tx_count: 0,
                    auth_data_root: [0xb3; 32].into(),
                }),
            )],
        })),
    };
    let context = TransitionContext {
        config: &runtime.config,
        clock: &SystemClock,
        full_state_authority: None,
        retention_references: &[],
    };

    assert!(matches!(
        runtime
            .apply(request(stale_episode), &context)
            .expect("a stale range episode is a normal apply outcome"),
        ApplyResult::Stale(_)
    ));
    assert!(runtime
        .store
        .aux_deliveries(target.hash)
        .expect("the stale target auxiliary rows remain readable")
        .is_empty());
    assert!(matches!(
        runtime
            .apply(request(prefix.episode), &context)
            .expect("the current one-header range prefix applies"),
        ApplyResult::Committed
    ));
}

#[tokio::test(start_paused = true)]
async fn retained_path_serves_a_locator_before_the_header_retention_window() {
    let (runtime, db, genesis, path) = reconciled_store_with_finalized_prefix(4);
    let hash_by_height = db
        .cf_handle("hash_by_height")
        .expect("the finalized hash index exists");
    let reader = runtime.reader();
    let target = Frontier::new(path[3].height, path[3].hash);
    let scope = zakura_header_chain::HeaderWorkAuthority::for_target(
        &runtime.publisher().snapshot(),
        target.hash,
    );
    let RetainedPathLeaseOutcome::Acquired(mut lease) = reader
        .acquire_retained_path(
            SourceId::from_digest([0x71; 32]),
            9,
            target.hash,
            &[genesis.hash],
            scope,
        )
        .expect("the finalized locator is a coherent retained path")
    else {
        panic!("the finalized locator should acquire a lease");
    };
    assert_eq!(
        lease.common_ancestor,
        Frontier::new(genesis.height, genesis.hash)
    );

    let owner = SourceId::from_digest([0x71; 32]);
    let mut after = genesis.hash;
    for (expected, complete) in path.iter().zip([false, false, false, true]) {
        let RetainedPathReadOutcome::Page(page) = reader
            .read_retained_path(owner, 9, lease.lease_id, scope, after, 1)
            .expect("the historical path page is coherent")
        else {
            panic!("the historical path lease should remain available");
        };
        assert_eq!(
            page.headers.as_slice(),
            std::slice::from_ref(&expected.header)
        );
        assert_eq!(page.aux_deliveries, vec![Vec::new()]);
        assert_eq!(page.complete, complete);
        after = expected.hash;
        if !complete {
            let RetainedPathLeaseOutcome::Acquired(next) = reader
                .acquire_retained_path(owner, 9, target.hash, &[after], scope)
                .unwrap()
            else {
                panic!("the next page can acquire fresh capacity");
            };
            lease = next;
        }
    }
    assert!(!reader
        .release_retained_path(owner, 9, lease.lease_id, scope)
        .expect("the one-header-page cursor releases"));

    for (marker, page_count) in [(0x72, 2), (0x73, 3)] {
        let page_owner = SourceId::from_digest([marker; 32]);
        let RetainedPathLeaseOutcome::Acquired(mut lease) = reader
            .acquire_retained_path(page_owner, 9, target.hash, &[genesis.hash], scope)
            .expect("the tier-boundary page cursor acquires")
        else {
            panic!("the tier-boundary cursor should be retained");
        };
        let mut after = genesis.hash;
        let mut served = Vec::new();
        loop {
            let RetainedPathReadOutcome::Page(page) = reader
                .read_retained_path(page_owner, 9, lease.lease_id, scope, after, page_count)
                .expect("the page spanning the storage-tier boundary is coherent")
            else {
                panic!("the tier-boundary cursor should remain available");
            };
            served.extend(page.headers.iter().map(|header| header.hash()));
            if page.complete {
                break;
            }
            after = page
                .headers
                .last()
                .expect("an incomplete page contains at least one header")
                .hash();
            let RetainedPathLeaseOutcome::Acquired(next) = reader
                .acquire_retained_path(page_owner, 9, target.hash, &[after], scope)
                .unwrap()
            else {
                panic!("the next page can acquire across the storage boundary");
            };
            lease = next;
        }
        assert_eq!(
            served,
            path.iter().map(|header| header.hash).collect::<Vec<_>>(),
            "page counts ending at and after the tier boundary serve one canonical sequence",
        );
    }

    let retry_owner = SourceId::from_digest([0x74; 32]);
    let RetainedPathLeaseOutcome::Acquired(retry_lease) = reader
        .acquire_retained_path(retry_owner, 9, target.hash, &[genesis.hash], scope)
        .expect("the corruption-retry cursor acquires")
    else {
        panic!("the corruption-retry cursor should be retained");
    };
    let mut corrupt = DiskWriteBatch::new();
    corrupt.zs_delete(&hash_by_height, path[0].height);
    db.write(corrupt)
        .expect("the test removes one finalized path hash");
    assert!(reader
        .read_retained_path(retry_owner, 9, retry_lease.lease_id, scope, genesis.hash, 1,)
        .is_err());
    let mut restore = DiskWriteBatch::new();
    restore.zs_insert(&hash_by_height, path[0].height, path[0].hash);
    db.write(restore)
        .expect("the test restores the finalized path hash");
    let RetainedPathReadOutcome::Page(retried) = reader
        .read_retained_path(retry_owner, 9, retry_lease.lease_id, scope, genesis.hash, 1)
        .expect("a repaired local row can retry the same cursor position")
    else {
        panic!("the failed page did not advance the cursor");
    };
    assert_eq!(retried.headers[0].hash(), path[0].hash);

    let expiry_owner = SourceId::from_digest([0x75; 32]);
    let RetainedPathLeaseOutcome::Acquired(expiry_lease) = reader
        .acquire_retained_path(expiry_owner, 9, target.hash, &[genesis.hash], scope)
        .expect("the failed-read expiry cursor acquires")
    else {
        panic!("the failed-read expiry cursor should be retained");
    };
    tokio::time::advance(RETAINED_PATH_LEASE_IDLE.saturating_sub(Duration::from_secs(1))).await;
    let mut corrupt = DiskWriteBatch::new();
    corrupt.zs_delete(&hash_by_height, path[0].height);
    db.write(corrupt)
        .expect("the test removes the expiring cursor's next hash");
    assert!(reader
        .read_retained_path(
            expiry_owner,
            9,
            expiry_lease.lease_id,
            scope,
            genesis.hash,
            1,
        )
        .is_err());
    tokio::time::advance(Duration::from_secs(2)).await;
    let mut restore = DiskWriteBatch::new();
    restore.zs_insert(&hash_by_height, path[0].height, path[0].hash);
    db.write(restore)
        .expect("the test restores the expiring cursor's next hash");
    assert_eq!(
        reader
            .read_retained_path(
                expiry_owner,
                9,
                expiry_lease.lease_id,
                scope,
                genesis.hash,
                1,
            )
            .expect("an expired cursor is a normal unavailable outcome"),
        RetainedPathReadOutcome::Unavailable,
        "a failed page must not renew its cursor deadline",
    );
}

#[tokio::test(start_paused = true)]
async fn retained_path_leases_are_exact_bounded_session_scoped_and_expiring() {
    let db_config = Config::ephemeral();
    let (engine_config, anchor, metadata) = fixture();
    let anchor_frontier = Frontier::new(anchor.height, anchor.hash);
    let store = HeaderChainStore::new(open(&db_config, engine_config.network()));
    store
        .initialize(metadata, anchor.clone())
        .expect("the empty schema initializes");
    let mut child_header = *anchor.header;
    child_header.previous_block_hash = anchor.hash;
    child_header.time += chrono::Duration::seconds(1);
    let child_header = Arc::new(child_header);
    let child = VerifiedHeaderRef {
        height: anchor.height.next().expect("genesis has a successor"),
        hash: child_header.hash(),
        header: child_header,
    };
    let mut grandchild_header = *anchor.header;
    grandchild_header.previous_block_hash = child.hash;
    grandchild_header.time += chrono::Duration::seconds(2);
    let grandchild_header = Arc::new(grandchild_header);
    let grandchild = VerifiedHeaderRef {
        height: child.height.next().expect("the child has a successor"),
        hash: grandchild_header.hash(),
        header: grandchild_header,
    };
    let (runtime, _) = store
        .startup_reconciled(
            &engine_config,
            anchor_frontier,
            Vec::new(),
            vec![child.clone(), grandchild.clone()],
        )
        .expect("the selected two-header path reconciles");
    let reader = runtime.reader();
    let validation_lease = reader
        .validation_context(anchor.hash)
        .expect("the retained parent context is coherent")
        .expect("the retained anchor has validation context");
    assert_eq!(validation_lease.parent(), anchor_frontier);
    assert_eq!(
        validation_lease.trust_anchor_digest(),
        engine_config.trust_anchor_digest()
    );
    assert_eq!(
        reader
            .validation_context(block::Hash([0xff; 32]))
            .expect("an absent parent is a normal stale read"),
        None
    );
    let durable_window = reader
        .selected_auxiliary_window(child.height, child.hash)
        .expect("the exact selected auxiliary window is coherent")
        .expect("the selected child is retained");
    let window = runtime
        .selected_auxiliary_window(child.height, child.hash)
        .expect("the in-memory selected auxiliary window is coherent")
        .expect("the selected child is retained in the committed engine");
    assert_eq!(window, durable_window);
    let captured_projection = runtime
        .capture_selected_projection()
        .expect("the in-memory selected projection is coherent");
    let child_index = captured_projection
        .frontiers
        .binary_search_by_key(&child.height, |frontier| frontier.height)
        .expect("the selected projection contains the child");
    assert_eq!(
        runtime
            .selected_auxiliary_window_at_projection_index(
                child_index,
                Frontier::new(child.height, child.hash),
            )
            .expect("the captured projection index is coherent"),
        Some(window.clone())
    );
    assert_eq!(
        runtime
            .selected_auxiliary_window_at_projection_index(
                child_index + 1,
                Frontier::new(child.height, child.hash),
            )
            .expect("a stale projection index is a normal read outcome"),
        None
    );
    assert_eq!(
        window.engine_snapshot,
        runtime.publisher().snapshot(),
        "the auxiliary window carries the snapshot read under the same transition lock"
    );
    assert_eq!(window.delivery_header.header_node.hash, child.hash);
    assert!(window.delivery_header.auxiliary_deliveries.is_empty());
    let successor_header = window
        .successor_header
        .expect("the selected grandchild follows");
    assert_eq!(successor_header.header_node.hash, grandchild.hash);
    assert!(successor_header.auxiliary_deliveries.is_empty());
    assert_eq!(
        reader
            .selected_auxiliary_window(child.height, block::Hash([0xfe; 32]))
            .expect("a stale branch hash is a normal read outcome"),
        None
    );
    let snapshot = runtime.publisher().snapshot();
    let owner = zakura_header_chain::BodyWorkAuthority::for_snapshot(&snapshot)
        .bind(7, NonZeroU64::new(8).expect("eight is nonzero"));
    let repair = reader
        .vct_repair_context(owner, child.height)
        .expect("the selected repair context is coherent")
        .expect("the current owner resolves its selected header");
    assert_eq!(repair.target, Frontier::new(child.height, child.hash));
    assert_eq!(repair.locator.entries(), &[anchor_frontier]);
    assert_eq!(repair.selected_header_count(), 2);
    assert_eq!(
        repair.request_target(),
        Frontier::new(grandchild.height, grandchild.hash)
    );
    let checkpoint_bounded = reader
        .vct_repair_context_bounded(owner, child.height, child.height)
        .expect("the checkpoint-bounded repair context is coherent")
        .expect("the checkpoint permits the blocking target");
    assert_eq!(checkpoint_bounded.selected_header_count(), 1);
    assert_eq!(checkpoint_bounded.request_target(), repair.target);
    assert_eq!(
        reader
            .vct_repair_context_bounded(owner, child.height, anchor.height)
            .expect("a repair above the checkpoint is a normal stale outcome"),
        None
    );

    let mut stale_owner = owner;
    stale_owner.authority.verified_generation = VerifiedGeneration::new(
        owner
            .verified_generation
            .get()
            .checked_add(1)
            .expect("the fixture state version can advance"),
    );
    assert_eq!(
        reader
            .vct_repair_context(stale_owner, child.height)
            .expect("a stale repair owner is a normal read outcome"),
        None
    );
    assert_eq!(
        reader
            .vct_repair_context(owner, anchor.height)
            .expect("a finalized repair height is a normal stale outcome"),
        None
    );

    let aux = zakura_header_chain::TreeAuxRecordV1 {
        height: child.height,
        sapling_root: Default::default(),
        orchard_root: Default::default(),
        ironwood_root: Default::default(),
        sapling_tx_count: 13,
        orchard_tx_count: 14,
        ironwood_tx_count: 15,
        auth_data_root: zakura_chain::block::merkle::AuthDataRoot::from([16; 32]),
    };
    let delivery = AuxDelivery::new(
        EvidenceId::from_digest([0x91; 32]),
        child.hash,
        SourceId::from_digest([0x92; 32]),
        owner.into(),
        zakura_header_chain::BodySizeHint::Unknown,
        Some(aux),
    );
    let mut child_node = runtime
        .store
        .header_node(child.hash)
        .expect("the selected child row decodes")
        .expect("the selected child is retained");
    child_node.aux_delivery_ids.push(delivery.delivery_id);
    let mut aux_batch = DiskWriteBatch::new();
    runtime
        .store
        .put_value(
            &mut aux_batch,
            HEADER_NODE_BY_HASH,
            child.hash.0,
            &HeaderNodeDisk::from_domain(&child_node),
        )
        .expect("the selected child with auxiliary evidence encodes");
    runtime
        .store
        .put_value(
            &mut aux_batch,
            HEADER_AUX_DELIVERY,
            HeaderAuxDeliveryKey {
                header: child.hash,
                delivery: delivery.delivery_id,
            }
            .as_bytes(),
            &delivery,
        )
        .expect("the selected auxiliary delivery encodes");
    runtime
        .store
        .db
        .write(aux_batch)
        .expect("the coherent selected auxiliary fixture commits");
    *runtime
        .transition_engine
        .lock()
        .expect("the transition engine mutex is not poisoned") =
        load_transition_engine(&runtime.store)
            .expect("the direct durable test fixture refreshes the runtime mirror");
    let evidence_constrained = reader
        .vct_repair_context(owner, child.height)
        .expect("the evidence-constrained repair context is coherent")
        .expect("the selected target remains repairable");
    assert_eq!(evidence_constrained.selected_header_count(), 1);
    assert_eq!(evidence_constrained.request_target(), repair.target);
    let roots = reader
        .selected_block_roots(child.height, 2)
        .expect("selected auxiliary roots are coherent");
    assert_eq!(roots.len(), 1, "the read stops at the first missing height");
    assert_eq!(roots[0].height, child.height);
    assert_eq!(roots[0].sapling_tx, aux.sapling_tx_count);
    assert_eq!(roots[0].orchard_tx, aux.orchard_tx_count);
    assert_eq!(roots[0].ironwood_tx, aux.ironwood_tx_count);
    assert_eq!(roots[0].auth_data_root, aux.auth_data_root);
    let crate::service::write::VctAuxiliaryWindowRead::Ready(window) =
        crate::service::write::HeaderChainWriter::new(runtime.clone(), engine_config.clone())
            .vct_auxiliary_window(child.height, child.hash)
            .expect("the selected auxiliary window is coherent")
    else {
        panic!("the current delivery remains usable without successor auxiliary data");
    };
    assert_eq!(window.successor_height, Some(grandchild.height));
    assert!(window.successor.is_none());

    let owner = SourceId::from_digest([1; 32]);
    let lease_scope = zakura_header_chain::HeaderWorkAuthority::for_target(
        &runtime.publisher().snapshot(),
        grandchild.hash,
    );
    let acquired = reader
        .acquire_retained_path(owner, 7, grandchild.hash, &[anchor.hash], lease_scope)
        .expect("the coherent target path is readable");
    let RetainedPathLeaseOutcome::Acquired(lease) = acquired else {
        panic!("the exact retained target should acquire a lease");
    };
    assert_eq!(
        lease.target,
        Frontier::new(grandchild.height, grandchild.hash)
    );
    assert_eq!(lease.common_ancestor, anchor_frontier);
    assert_eq!(lease.scope, lease_scope);
    let mut wrong_scope = lease_scope;
    wrong_scope.header_generation = wrong_scope
        .header_generation
        .checked_next()
        .expect("the fixture generation has a successor");
    assert_eq!(
        reader
            .acquire_retained_path(
                SourceId::from_digest([0xee; 32]),
                7,
                grandchild.hash,
                &[anchor.hash],
                wrong_scope,
            )
            .expect("a stale acquisition scope is a normal refusal"),
        RetainedPathLeaseOutcome::Busy
    );
    assert!(matches!(
        reader
            .acquire_retained_path(owner, 7, grandchild.hash, &[anchor.hash], lease_scope,)
            .expect("the lease bound is a normal outcome"),
        RetainedPathLeaseOutcome::CapacityBusy(_)
    ));
    assert!(matches!(
        reader
            .acquire_retained_path(owner, 8, grandchild.hash, &[anchor.hash], lease_scope)
            .expect("a new session cannot replace a live lease"),
        RetainedPathLeaseOutcome::CapacityBusy(_)
    ));
    assert_eq!(
        reader
            .read_retained_path(owner, 8, lease.lease_id, lease_scope, anchor.hash, 1)
            .expect("a mismatched session is non-fatal"),
        RetainedPathReadOutcome::Unavailable
    );
    assert_eq!(
        reader
            .read_retained_path(owner, 7, lease.lease_id, wrong_scope, anchor.hash, 1)
            .expect("a mismatched branch scope is non-fatal"),
        RetainedPathReadOutcome::Unavailable
    );
    assert!(!reader
        .release_retained_path(owner, 7, lease.lease_id, wrong_scope)
        .expect("a mismatched release scope is non-fatal"));
    let RetainedPathReadOutcome::Page(page) = reader
        .read_retained_path(owner, 7, lease.lease_id, lease_scope, anchor.hash, 1)
        .expect("a lease page read validates against the serialized publication gate")
    else {
        panic!("the current owner should read its lease");
    };
    assert_eq!(page.headers.len(), 1);
    assert_eq!(page.headers[0].hash(), child.hash);
    assert_eq!(page.common_ancestor, anchor_frontier);
    assert_eq!(page.scope, lease_scope);
    assert_eq!(page.aux_deliveries, vec![vec![delivery]]);
    assert!(!page.complete);
    assert_eq!(
        reader
            .read_retained_path(owner, 7, lease.lease_id, lease_scope, anchor.hash, 1)
            .expect("a replayed cursor position is a normal refusal"),
        RetainedPathReadOutcome::Unavailable,
        "the opaque cursor advances exactly once and cannot be rewound",
    );

    let RetainedPathLeaseOutcome::Acquired(lease) = reader
        .acquire_retained_path(owner, 7, grandchild.hash, &[child.hash], lease_scope)
        .unwrap()
    else {
        panic!("the next page acquires its own lease");
    };
    let cursor = runtime
        .leases
        .lock()
        .expect("the lease registry is available")
        .get(owner, 7, lease.lease_id, Instant::now())
        .expect("the lease remains active");
    let captured_read = reader
        .capture_path_read(&cursor, 1)
        .expect("the page captures a coherent read view");
    assert!(
        reader.store.writer.try_lock().is_ok(),
        "page I/O does not hold the writer"
    );
    assert!(
        reader.transition_engine.try_lock().is_ok(),
        "page I/O does not hold the engine"
    );

    let before = runtime.publisher().snapshot();
    let evidence = EvidenceId::from_digest([3; 32]);
    let id = zakura_header_chain::OperatorInvalidationId::new([3; 16]);
    let mut hasher = sha2::Sha256::new();
    use sha2::Digest as _;
    hasher.update(b"zakura-operator-invalidation-v1");
    hasher.update(child.hash.0);
    hasher.update(id.bytes());
    let authority = Authority(evidence);
    runtime
        .apply(
            TransitionRequest {
                expected_version: before.state_version,
                event: TransitionEvent::OperatorInvalidate(
                    zakura_header_chain::OperatorInvalidate {
                        target: child.hash,
                        id,
                        operator_reason_digest: hasher.finalize().into(),
                        evidence,
                    },
                ),
            },
            &TransitionContext {
                config: &engine_config,
                clock: &SystemClock,
                full_state_authority: Some(&authority),
                retention_references: &[],
            },
        )
        .expect("the selected path can change while the lease is active");
    assert_eq!(
        runtime.publisher().snapshot().frontiers.header_best,
        anchor_frontier
    );

    let (captured_page, _, _) = captured_read
        .read_page(&cursor, 1)
        .expect("the old snapshot remains coherent after the branch changes")
        .expect("the captured path remains readable");
    assert_eq!(captured_page.headers[0].hash(), grandchild.hash);
    assert!(captured_page.complete);

    let stale_source = SourceId::from_digest([0xed; 32]);
    let RetainedPathLeaseOutcome::Acquired(stale_lease) = reader
        .acquire_retained_path(
            stale_source,
            7,
            grandchild.hash,
            &[anchor.hash],
            lease_scope,
        )
        .expect("a request queued before the head changed can still acquire its exact path")
    else {
        panic!("head changes do not invalidate an exact retained target");
    };
    assert_eq!(stale_lease.scope, lease_scope);
    assert!(reader
        .release_retained_path(stale_source, 7, stale_lease.lease_id, lease_scope)
        .unwrap());

    let RetainedPathReadOutcome::Page(continuation) = reader
        .read_retained_path(owner, 7, lease.lease_id, lease_scope, child.hash, 1)
        .expect("the immutable cursor continues after reselection")
    else {
        panic!("the current owner should read its continuation");
    };
    assert_eq!(
        continuation.common_ancestor,
        Frontier::new(child.height, child.hash)
    );
    assert_eq!(continuation.headers[0].hash(), grandchild.hash);
    assert!(continuation.complete);

    assert_eq!(
        reader
            .acquire_retained_path(
                SourceId::from_digest([2; 32]),
                7,
                block::Hash([0xfe; 32]),
                &[anchor.hash],
                zakura_header_chain::HeaderWorkAuthority::for_target(
                    &runtime.publisher().snapshot(),
                    block::Hash([0xfe; 32]),
                ),
            )
            .expect("an absent target is a normal outcome"),
        RetainedPathLeaseOutcome::TargetNotRetained
    );
    assert_eq!(
        reader
            .acquire_retained_path(
                SourceId::from_digest([2; 32]),
                7,
                child.hash,
                &[block::Hash([0xfd; 32])],
                zakura_header_chain::HeaderWorkAuthority::for_target(
                    &runtime.publisher().snapshot(),
                    child.hash,
                ),
            )
            .expect("a disjoint locator is a normal outcome"),
        RetainedPathLeaseOutcome::NoLocatorIntersection
    );
    let RetainedPathLeaseOutcome::Acquired(target_intersection) = reader
        .acquire_retained_path(
            SourceId::from_digest([2; 32]),
            7,
            child.hash,
            &[child.hash, anchor.hash],
            zakura_header_chain::HeaderWorkAuthority::for_target(
                &runtime.publisher().snapshot(),
                child.hash,
            ),
        )
        .expect("the nearest locator ancestor is selected")
    else {
        panic!("the target itself intersects the locator");
    };
    assert_eq!(target_intersection.common_ancestor.hash, child.hash);
    let RetainedPathReadOutcome::Page(completed) = reader
        .read_retained_path(
            SourceId::from_digest([2; 32]),
            7,
            target_intersection.lease_id,
            target_intersection.scope,
            child.hash,
            1,
        )
        .expect("a cursor acquired at its target is readable")
    else {
        panic!("the target-intersection cursor remains available");
    };
    assert!(completed.headers.is_empty());
    assert!(completed.complete);
    assert!(!reader
        .release_retained_path(
            SourceId::from_digest([2; 32]),
            7,
            target_intersection.lease_id,
            target_intersection.scope,
        )
        .expect("the target-intersection lease releases"));

    assert!(!reader
        .release_retained_path(owner, 7, lease.lease_id, lease_scope)
        .expect("the exact owner can release its lease"));
    // Ordinary leases fill the general slots, and the last short path uses the reserve.
    for marker in 1..=MAX_RETAINED_PATH_LEASES {
        let marker = u8::try_from(marker).expect("the lease cap fits in one byte");
        assert!(matches!(
            reader
                .acquire_retained_path(
                    SourceId::from_digest([marker; 32]),
                    9,
                    child.hash,
                    &[anchor.hash],
                    zakura_header_chain::HeaderWorkAuthority::for_target(
                        &runtime.publisher().snapshot(),
                        child.hash,
                    ),
                )
                .expect("bounded acquisition returns an outcome"),
            RetainedPathLeaseOutcome::Acquired(_)
        ));
    }
    assert!(matches!(
        reader
            .acquire_retained_path(
                SourceId::from_digest([0xff; 32]),
                9,
                child.hash,
                &[anchor.hash],
                zakura_header_chain::HeaderWorkAuthority::for_target(
                    &runtime.publisher().snapshot(),
                    child.hash,
                ),
            )
            .expect("capacity refusal is a normal outcome"),
        RetainedPathLeaseOutcome::CapacityBusy(_)
    ));
    tokio::time::advance(RETAINED_PATH_LEASE_IDLE + Duration::from_secs(1)).await;
    assert!(matches!(
        reader
            .acquire_retained_path(
                SourceId::from_digest([0xff; 32]),
                10,
                child.hash,
                &[anchor.hash],
                zakura_header_chain::HeaderWorkAuthority::for_target(
                    &runtime.publisher().snapshot(),
                    child.hash,
                ),
            )
            .expect("expired slots are reclaimed"),
        RetainedPathLeaseOutcome::Acquired(_)
    ));

    let snapshot = runtime.publisher().snapshot();
    let delivery = AuxDelivery::new(
        EvidenceId::from_digest([0xa1; 32]),
        anchor.hash,
        SourceId::from_digest([0xa2; 32]),
        body_owner(&snapshot, 11, 12).into(),
        zakura_header_chain::BodySizeHint::Unknown,
        None,
    );
    let mut corrupt = DiskWriteBatch::new();
    runtime
        .store
        .put_value(
            &mut corrupt,
            HEADER_AUX_DELIVERY,
            HeaderAuxDeliveryKey {
                header: anchor.hash,
                delivery: delivery.delivery_id,
            }
            .as_bytes(),
            &delivery,
        )
        .expect("the contradictory auxiliary row encodes");
    runtime
        .store
        .db
        .write(corrupt)
        .expect("the contradictory auxiliary row commits");
    assert!(matches!(
        reader.selected_auxiliary_window(anchor.height, anchor.hash),
        Err(HeaderChainStoreError::Store(StoreError::Incoherent(
            "retained node and auxiliary delivery index disagree"
        )))
    ));
}

#[tokio::test(start_paused = true)]
async fn retained_path_serves_a_bounded_finalized_range_below_the_header_frontier() {
    let (runtime, _db, genesis, path) = reconciled_store_with_finalized_prefix(4);
    let finalized = Frontier::new(path[2].height, path[2].hash);
    let reader = runtime.reader();

    // A VCT repair can ask for a selected range below the retained suffix. A supplier that has
    // finalized that range serves it from the finalized indexes.
    let target = Frontier::new(path[1].height, path[1].hash);
    assert!(target.height < finalized.height);
    let scope = zakura_header_chain::HeaderWorkAuthority::for_target(
        &runtime.publisher().snapshot(),
        target.hash,
    );
    // Transfers occupy every ordinary slot. The reserve still admits a bounded repair.
    let retained_target = Frontier::new(path[3].height, path[3].hash);
    let retained_scope = zakura_header_chain::HeaderWorkAuthority::for_target(
        &runtime.publisher().snapshot(),
        retained_target.hash,
    );
    for marker in 1..MAX_RETAINED_PATH_LEASES {
        let marker = u8::try_from(marker).expect("the lease cap fits in one byte");
        assert!(matches!(
            reader
                .acquire_retained_path(
                    SourceId::from_digest([marker; 32]),
                    10,
                    retained_target.hash,
                    &[genesis.hash],
                    retained_scope,
                )
                .expect("the general path acquisition is coherent"),
            RetainedPathLeaseOutcome::Acquired(_)
        ));
    }

    let owner = SourceId::from_digest([0x81; 32]);
    let RetainedPathLeaseOutcome::Acquired(lease) = reader
        .acquire_retained_path(owner, 11, target.hash, &[path[0].hash], scope)
        .expect("the finalized target resolves through the finalized indexes")
    else {
        panic!("the finalized target should acquire a lease");
    };
    assert_eq!(
        lease.common_ancestor,
        Frontier::new(path[0].height, path[0].hash)
    );
    assert_eq!(lease.target, target);

    let RetainedPathReadOutcome::Page(page) = reader
        .read_retained_path(owner, 11, lease.lease_id, scope, path[0].hash, 4)
        .expect("the finalized target page is coherent")
    else {
        panic!("the finalized target lease should remain available");
    };
    assert_eq!(
        page.headers.as_slice(),
        std::slice::from_ref(&path[1].header)
    );
    assert_eq!(page.target, target);
    assert!(page.complete);
    assert!(!reader
        .release_retained_path(owner, 11, lease.lease_id, scope)
        .expect("the finalized target cursor releases"));

    // The same acquisition serves a range from an earlier canonical locator.
    let long_path_owner = SourceId::from_digest([0x84; 32]);
    let RetainedPathLeaseOutcome::Acquired(long_path_lease) = reader
        .acquire_retained_path(long_path_owner, 11, target.hash, &[genesis.hash], scope)
        .expect("the bounded finalized path lookup is coherent")
    else {
        panic!("the bounded finalized path should acquire a lease");
    };
    let RetainedPathReadOutcome::Page(long_path_page) = reader
        .read_retained_path(
            long_path_owner,
            11,
            long_path_lease.lease_id,
            scope,
            genesis.hash,
            4,
        )
        .expect("the bounded finalized path page is coherent")
    else {
        panic!("the bounded finalized path lease should remain available");
    };
    assert_eq!(
        long_path_page.headers,
        vec![path[0].header.clone(), path[1].header.clone()]
    );
    assert_eq!(long_path_page.target, target);
    assert!(long_path_page.complete);
    assert!(!reader
        .release_retained_path(long_path_owner, 11, long_path_lease.lease_id, scope,)
        .expect("the bounded finalized path cursor releases"));

    // The nearest canonical locator bounds the lease. Honouring list order instead would let a
    // requester place a distant ancestor first and lease a complete range it never needed.
    let nearest_owner = SourceId::from_digest([0x85; 32]);
    let RetainedPathLeaseOutcome::Acquired(nearest_lease) = reader
        .acquire_retained_path(
            nearest_owner,
            11,
            target.hash,
            &[genesis.hash, path[0].hash],
            scope,
        )
        .expect("the nearest canonical locator resolves")
    else {
        panic!("the nearest canonical locator should acquire a lease");
    };
    assert_eq!(
        nearest_lease.common_ancestor,
        Frontier::new(path[0].height, path[0].hash),
        "an ordered-first distant locator must not widen the leased range"
    );
    assert!(reader
        .release_retained_path(nearest_owner, 11, nearest_lease.lease_id, scope)
        .expect("the nearest canonical locator cursor releases"));

    // A locator above the target cannot intersect its ancestry.
    let above_owner = SourceId::from_digest([0x82; 32]);
    assert!(matches!(
        reader
            .acquire_retained_path(above_owner, 11, target.hash, &[path[2].hash], scope)
            .expect("the locator lookup is coherent"),
        RetainedPathLeaseOutcome::NoLocatorIntersection
    ));

    // An unknown target stays unservable.
    let unknown_owner = SourceId::from_digest([0x83; 32]);
    let unknown = zakura_chain::block::Hash([0x9c; 32]);
    let unknown_scope = zakura_header_chain::HeaderWorkAuthority::for_target(
        &runtime.publisher().snapshot(),
        unknown,
    );
    assert!(matches!(
        reader
            .acquire_retained_path(unknown_owner, 11, unknown, &[genesis.hash], unknown_scope)
            .expect("the unknown target lookup is coherent"),
        RetainedPathLeaseOutcome::TargetNotRetained
    ));
}

#[test]
fn body_size_hints_by_hash_read_known_sizes_from_retained_deliveries() {
    let (runtime, db, _genesis, path) = reconciled_store_with_finalized_prefix(5);
    let parent = Frontier::new(path[2].height, path[2].hash);
    let target = Frontier::new(path[3].height, path[3].hash);
    let successor = Frontier::new(path[4].height, path[4].hash);
    let snapshot = runtime.publisher().snapshot();
    let owner = zakura_header_chain::BodyWorkAuthority::for_snapshot(&snapshot)
        .bind(27, NonZeroU64::new(28).expect("twenty-eight is nonzero"));
    let range = runtime
        .reader()
        .vct_repair_context(owner, target.height)
        .expect("the range repair context is coherent")
        .expect("the selected empty range needs repair");
    let prefix = range
        .bounded_prefix(1)
        .expect("the one-header range prefix exists");
    let lease = runtime
        .reader()
        .validation_context(parent.hash)
        .expect("the repair parent validation context is coherent")
        .expect("the repair parent remains retained");
    let rules =
        HeaderRules::for_validation_lease(&lease).expect("the repair parent produces header rules");
    let batch = zakura_header_chain::prepare_headers(
        HeaderBatchInput::new(std::slice::from_ref(&path[3].header)),
        parent,
        &rules,
        &SystemClock,
    )
    .expect("the selected prefix passes deterministic preparation");
    let source = SourceId::from_digest([0xc1; 32]);
    let known_size = std::num::NonZeroU32::new(123_456).expect("the fixture size is nonzero");
    let request = TransitionRequest {
        expected_version: StateVersion::default(),
        event: TransitionEvent::InsertHeaders(Box::new(InsertHeaders {
            owner: owner.into(),
            source,
            parent_hash: parent.hash,
            target_tip_hash: target.hash,
            completion: TargetCompletion::SelectedAuxiliaryRepair {
                common_ancestor: parent,
                selected_target: target,
                episode: prefix.episode,
            },
            batch,
            aux: vec![AuxDelivery::new(
                EvidenceId::from_digest([0xc2; 32]),
                target.hash,
                source,
                owner.into(),
                zakura_header_chain::BodySizeHint::Unknown,
                Some(zakura_header_chain::TreeAuxRecordV1 {
                    height: target.height,
                    sapling_root: Default::default(),
                    orchard_root: Default::default(),
                    ironwood_root: Default::default(),
                    sapling_tx_count: 1,
                    orchard_tx_count: 0,
                    ironwood_tx_count: 0,
                    auth_data_root: [0xc3; 32].into(),
                }),
            )],
        })),
    };
    let original_request = request.clone();
    let context = TransitionContext {
        config: &runtime.config,
        clock: &SystemClock,
        full_state_authority: None,
        retention_references: &[],
    };
    assert!(matches!(
        runtime
            .apply(request, &context)
            .expect("the one-header range prefix applies"),
        ApplyResult::Committed
    ));

    let hints_by_hash = || {
        runtime
            .reader()
            .body_size_hints_by_hash(&[parent.hash, target.hash, successor.hash])
            .expect("the delivery hints read")
    };
    assert_eq!(hints_by_hash(), vec![None, None, None]);
    assert!(runtime
        .store
        .scan_raw(HEADER_AUX_BODY_SIZE)
        .unwrap()
        .is_empty());
    let replay = |request_id, marker, size| {
        let view = runtime.publisher().view();
        let mut replay = original_request.clone();
        replay.expected_version = view.state_version;
        let TransitionEvent::InsertHeaders(insert) = &mut replay.event else {
            unreachable!()
        };
        let new_owner = HeaderWorkAuthority::for_target(&view.snapshot, target.hash)
            .bind(27, NonZeroU64::new(request_id).unwrap());
        insert.owner = new_owner.into();
        insert.completion = TargetCompletion::TargetPrefix {
            common_ancestor: parent,
        };
        insert.aux[0].owner = new_owner.into();
        insert.aux[0].delivery_id = EvidenceId::from_digest([marker; 32]);
        insert.aux[0].body_size = zakura_header_chain::BodySizeHint::new(size).unwrap();
        replay
    };
    let before = runtime.publisher().view();
    let filled = runtime
        .apply(replay(29, 0xc4, known_size.get()), &context)
        .unwrap();
    assert!(matches!(filled, ApplyResult::Committed), "{filled:?}");
    let after = runtime.publisher().view();
    assert_eq!(after.header_generation, before.header_generation);
    assert_eq!(after.body_work_epoch, before.body_work_epoch);
    assert_eq!(
        after.body_size_hint_revision,
        before.body_size_hint_revision + 1
    );
    assert_eq!(
        runtime.store.scan_raw(HEADER_AUX_DELIVERY).unwrap().len(),
        1
    );
    assert_eq!(
        runtime.store.scan_raw(HEADER_AUX_BODY_SIZE).unwrap().len(),
        1
    );
    let persisted = runtime.store.untrusted_aux_deliveries(target.hash).unwrap();
    assert_eq!(
        persisted[0].delivery().body_size,
        zakura_header_chain::BodySizeHint::Unknown
    );
    assert_eq!(hints_by_hash(), vec![None, Some(known_size), None]);

    // A differing known hint is false, so it cannot replace the filled size.
    let conflicting = runtime.apply(replay(30, 0xc5, 3_146), &context).unwrap();
    assert!(
        matches!(conflicting, ApplyResult::NoChange(_)),
        "{conflicting:?}"
    );
    assert_eq!(
        runtime.publisher().view().body_size_hint_revision,
        after.body_size_hint_revision
    );
    let config = runtime.config.clone();
    drop(runtime);
    let (reopened, _) = HeaderChainStore::new(db).startup(&config).unwrap();
    assert_eq!(
        reopened
            .reader()
            .body_size_hints_by_hash(&[target.hash])
            .unwrap(),
        vec![Some(known_size)]
    );
}

#[test]
fn retained_path_lookup_stops_at_the_locator_outside_commit_locks() {
    let (runtime, db, _, path) = reconciled_store_with_finalized_prefix(64);
    let reader = runtime.reader();
    let target = path.last().unwrap();
    let parent = &path[path.len() - 2];
    let snapshot = runtime.publisher().snapshot();
    let scope = HeaderWorkAuthority::for_target(&snapshot, target.hash);

    // Make an older, unrequested row unreadable so a walk past the locator fails
    // deterministically. The target's requested suffix is still intact.
    let mut batch = DiskWriteBatch::new();
    runtime
        .store
        .delete_raw(&mut batch, HEADER_NODE_BY_HASH, path[3].hash.0)
        .unwrap();
    db.write(batch).unwrap();
    let read = reader.store.audit_snapshot().unwrap();
    let node = read.retained_path_node(target.hash).unwrap().unwrap();
    {
        let _writer = reader.store.writer.lock().unwrap();
        let _engine = reader.transition_engine.lock().unwrap();
        let ancestors = read
            .retained_ancestry_to_nearest_locator(
                node,
                snapshot.frontiers.finalized,
                &HashSet::from([parent.hash]),
                None,
            )
            .unwrap();
        assert_eq!(
            ancestors,
            serving_snapshot::RetainedAncestry::Path(vec![
                Frontier::new(target.height, target.hash),
                Frontier::new(parent.height, parent.hash),
            ])
        );
    }

    let source = SourceId::from_digest([0xc1; 32]);
    let RetainedPathLeaseOutcome::Acquired(lease) = reader
        .acquire_retained_path(source, 7, target.hash, &[parent.hash, path[2].hash], scope)
        .unwrap()
    else {
        panic!("a one-header request only needs the intact suffix");
    };
    let RetainedPathReadOutcome::Page(page) = reader
        .read_retained_path(source, 7, lease.lease_id, scope, parent.hash, 1)
        .unwrap()
    else {
        panic!("the requested suffix can be served");
    };
    assert_eq!(page.headers[0].hash(), target.hash);
    assert!(page.complete);
    // The nearest locator wins regardless of list order, so the walk never reaches the gap.
    let RetainedPathLeaseOutcome::Acquired(reordered) = reader
        .acquire_retained_path(
            SourceId::from_digest([0xc2; 32]),
            7,
            target.hash,
            &[path[2].hash, parent.hash],
            scope,
        )
        .unwrap()
    else {
        panic!("the nearest locator only needs the intact suffix");
    };
    assert_eq!(reordered.common_ancestor.hash, parent.hash);
    assert!(matches!(
        reader
            .acquire_retained_path(
                SourceId::from_digest([0xc3; 32]),
                7,
                target.hash,
                &[path[2].hash],
                scope,
            )
            .unwrap(),
        RetainedPathLeaseOutcome::HistoryPruned
    ));
}

#[test]
fn retained_path_lookup_chooses_the_nearest_locator_and_skips_other_branches() {
    let (runtime, db, genesis, path) = reconciled_store_with_finalized_prefix(6);
    let reader = runtime.reader();
    let target = &path[4];
    let snapshot = runtime.publisher().snapshot();
    let scope = HeaderWorkAuthority::for_target(&snapshot, target.hash);
    let mut fork = runtime.store.header_node(path[3].hash).unwrap().unwrap();
    let mut header = *fork.header;
    header.nonce.0[0] = 0xff;
    fork.header = Arc::new(header);
    fork.hash = fork.header.hash();
    let mut batch = DiskWriteBatch::new();
    runtime
        .store
        .put_value(
            &mut batch,
            HEADER_NODE_BY_HASH,
            fork.hash.0,
            &HeaderNodeDisk::from_domain(&fork),
        )
        .unwrap();
    db.write(batch).unwrap();
    for (index, (locators, ancestor)) in [
        (vec![path[2].hash, path[3].hash], &path[3]),
        (vec![path[3].hash, path[2].hash], &path[3]),
        (vec![genesis.hash, path[3].hash], &path[3]),
        (vec![path[3].hash, genesis.hash], &path[3]),
        (vec![path[5].hash, path[3].hash], &path[3]),
        (vec![fork.hash, target.hash], target),
        (vec![fork.hash, path[3].hash], &path[3]),
        (vec![fork.hash, genesis.hash], &genesis),
        (vec![target.hash, path[3].hash], target),
    ]
    .into_iter()
    .enumerate()
    {
        let source = SourceId::from_digest([u8::try_from(index).unwrap(); 32]);
        let RetainedPathLeaseOutcome::Acquired(lease) = reader
            .acquire_retained_path(source, 7, target.hash, &locators, scope)
            .unwrap()
        else {
            panic!("each locator list intersects the retained target's ancestry");
        };
        assert_eq!(lease.common_ancestor.hash, ancestor.hash);
        let RetainedPathReadOutcome::Page(page) = reader
            .read_retained_path(source, 7, lease.lease_id, scope, ancestor.hash, 10)
            .unwrap()
        else {
            panic!("the path from the chosen locator remains contiguous");
        };
        assert_eq!(
            page.headers.len(),
            usize::try_from(target.height.0 - ancestor.height.0).unwrap()
        );
        assert_eq!(page.target.hash, target.hash);
        assert!(page.complete);
    }
}

#[test]
fn retained_path_survives_finalization_during_and_between_page_reads() {
    let (runtime, db, _, path) = reconciled_store_with_finalized_prefix(6);
    let reader = runtime.reader();
    let target = Frontier::new(path[3].height, path[3].hash);
    let source = SourceId::from_digest([0xa1; 32]);
    let before = runtime.publisher().snapshot();
    let scope = HeaderWorkAuthority::for_target(&before, target.hash);
    let RetainedPathLeaseOutcome::Acquired(lease) = reader
        .acquire_retained_path(source, 7, target.hash, &[path[2].hash], scope)
        .expect("the retained target can be leased")
    else {
        panic!("the retained target has capacity");
    };
    let cursor = runtime
        .leases
        .lock()
        .unwrap()
        .get(source, 7, lease.lease_id, Instant::now())
        .unwrap();
    let captured = reader.capture_path_read(&cursor, 1).unwrap();
    let ancestry_read = reader.store.audit_snapshot().unwrap();
    finalize_serving_test_path(&runtime, &db, &path[3..5]);
    assert!(runtime.publisher().snapshot().header_generation > before.header_generation);
    assert!(
        runtime.store.header_node(target.hash).unwrap().is_none(),
        "finality removes the old retained row"
    );
    let ancestors = ancestry_read
        .retained_ancestry_to_nearest_locator(
            ancestry_read
                .retained_path_node(target.hash)
                .unwrap()
                .unwrap(),
            before.frontiers.finalized,
            &HashSet::from([path[2].hash]),
            None,
        )
        .unwrap();
    assert_eq!(
        ancestors,
        serving_snapshot::RetainedAncestry::Path(vec![target, before.frontiers.finalized])
    );
    let (page, _, _) = captured.read_page(&cursor, 1).unwrap().unwrap();
    assert_eq!(page.headers[0].hash(), target.hash);
    assert!(page.complete);
    let RetainedPathReadOutcome::Page(page) = reader
        .read_retained_path(source, 7, lease.lease_id, scope, path[2].hash, 1)
        .unwrap()
    else {
        panic!("a fresh page reads the same hash from finalized storage");
    };
    assert_eq!(page.headers[0].hash(), target.hash);
    assert!(page.complete);
    let RetainedPathLeaseOutcome::Acquired(fresh) = reader
        .acquire_retained_path(
            SourceId::from_digest([0xa2; 32]),
            7,
            target.hash,
            &[path[2].hash],
            scope,
        )
        .unwrap()
    else {
        panic!("an acquisition queued before finality can still serve the exact hash");
    };
    assert_eq!(fresh.target, target);
    assert_eq!(fresh.scope, scope);
}

#[test]
fn retained_path_rejects_a_target_pruned_before_lease_commit_or_page_capture() {
    let (runtime, db, _, path) = reconciled_store_with_finalized_prefix(6);
    let reader = runtime.reader();
    let source = SourceId::from_digest([0xb1; 32]);
    let target = Frontier::new(path[3].height, path[3].hash);
    let before = runtime.publisher().snapshot();
    let scope = HeaderWorkAuthority::for_target(&before, target.hash);
    let RetainedPathLeaseOutcome::Acquired(lease) = reader
        .acquire_retained_path(source, 7, target.hash, &[path[2].hash], scope)
        .unwrap()
    else {
        panic!("the original target is retained");
    };
    let cursor = runtime
        .leases
        .lock()
        .unwrap()
        .get(source, 7, lease.lease_id, Instant::now())
        .unwrap();
    let captured = reader.capture_path_read(&cursor, 1).unwrap();
    let ancestry_read = reader.store.audit_snapshot().unwrap();

    let cached_source = SourceId::from_digest([0xb4; 32]);
    let cached_target = path[4].hash;
    let cached_scope = HeaderWorkAuthority::for_target(&before, cached_target);
    let RetainedPathLeaseOutcome::Acquired(cached_lease) = reader
        .acquire_retained_path(
            cached_source,
            8,
            cached_target,
            &[path[2].hash],
            cached_scope,
        )
        .unwrap()
    else {
        panic!("the longer losing path can prepare a continuation");
    };
    let RetainedPathReadOutcome::Page(page) = reader
        .read_retained_path(
            cached_source,
            8,
            cached_lease.lease_id,
            cached_scope,
            path[2].hash,
            1,
        )
        .unwrap()
    else {
        panic!("the first losing-path page is available before finality");
    };
    assert!(!page.complete);

    let pending_source = SourceId::from_digest([0xb2; 32]);
    let (reservation_id, capacity) = runtime
        .leases
        .lock()
        .unwrap()
        .reserve(
            pending_source,
            Instant::now(),
            RetainedPathCapacity::General,
        )
        .unwrap();
    let reservation = RetainedPathReservation {
        leases: runtime.leases.clone(),
        peer: pending_source,
        reservation_id,
        active: true,
    };
    let pending = RetainedPathLeaseSpec {
        peer: pending_source,
        session_id: 8,
        target,
        common_ancestor: cursor.common_ancestor,
        scope,
        position: cursor.position,
        retained_path: cursor.retained_path.clone(),
        capacity,
    };

    let mut parent = path[2].clone();
    let mut fork = Vec::new();
    for _ in 0..3 {
        let mut header = *parent.header;
        header.previous_block_hash = parent.hash;
        header.time += chrono::Duration::seconds(10);
        let header = Arc::new(header);
        parent = VerifiedHeaderRef {
            height: parent.height.next().unwrap(),
            hash: header.hash(),
            header,
        };
        fork.push(parent.clone());
    }
    let evidence = EvidenceId::from_digest([0xb3; 32]);
    let authority = Authority(evidence);
    runtime
        .apply(
            TransitionRequest {
                expected_version: before.state_version,
                event: TransitionEvent::VerifiedChainChanged(VerifiedChainChanged {
                    full_state_transition_id: evidence,
                    old_tip: before.frontiers.verified_best,
                    new_path: fork.clone(),
                    cause: VerifiedChangeCause::Reset,
                }),
            },
            &TransitionContext {
                config: &runtime.config,
                clock: &SystemClock,
                full_state_authority: Some(&authority),
                retention_references: &[],
            },
        )
        .expect("full state can select the competing verified path");
    finalize_serving_test_path(&runtime, &db, &fork[..2]);
    assert_eq!(
        reader
            .acquire_retained_path(
                cached_source,
                8,
                cached_target,
                &[path[3].hash],
                cached_scope
            )
            .unwrap(),
        RetainedPathLeaseOutcome::TargetNotRetained,
        "a cached path must not resurrect a branch removed by finality"
    );
    assert!(runtime.store.header_node(target.hash).unwrap().is_none());
    assert!(reader
        .store
        .audit_snapshot()
        .unwrap()
        .finalized_frontier(target.hash)
        .unwrap()
        .is_none());

    let ancestors = ancestry_read
        .retained_ancestry_to_nearest_locator(
            ancestry_read
                .retained_path_node(target.hash)
                .unwrap()
                .unwrap(),
            before.frontiers.finalized,
            &HashSet::from([path[2].hash]),
            None,
        )
        .unwrap();
    assert_eq!(
        ancestors,
        serving_snapshot::RetainedAncestry::Path(vec![target, before.frontiers.finalized])
    );

    let (page, _, _) = captured.read_page(&cursor, 1).unwrap().unwrap();
    assert_eq!(page.headers[0].hash(), target.hash);
    assert!(page.complete);
    assert_eq!(
        reader
            .read_retained_path(source, 7, lease.lease_id, scope, path[2].hash, 1)
            .unwrap(),
        RetainedPathReadOutcome::Unavailable
    );
    assert!(matches!(
        reader.commit_path_lease(reservation, pending).unwrap(),
        RetainedPathLeaseOutcome::TargetNotRetained
    ));
    let leases = runtime.leases.lock().unwrap();
    assert!(!leases.reservations.contains_key(&pending_source));
    assert!(!leases.by_peer.contains_key(&pending_source));
}

fn finalize_serving_test_path(
    runtime: &HeaderChainRuntime,
    db: &DiskDb,
    rows: &[VerifiedHeaderRef],
) {
    let before = runtime.publisher().snapshot();
    let last = rows.last().expect("the test finalizes a nonempty range");
    let finalized = Frontier::new(last.height, last.hash);
    let proof: Vec<_> = runtime
        .verified_projection()
        .unwrap()
        .into_iter()
        .take_while(|frontier| frontier.height <= finalized.height)
        .map(|frontier| frontier.hash)
        .collect();
    let evidence =
        zakura_header_chain::full_state_finality_evidence(before.state_version, finalized, &proof);
    let authority = Authority(evidence);
    let mut batch = DiskWriteBatch::new();
    for header in rows {
        batch.zs_insert(
            &db.cf_handle("hash_by_height").unwrap(),
            header.height,
            header.hash,
        );
        batch.zs_insert(
            &db.cf_handle("height_by_hash").unwrap(),
            header.hash,
            header.height,
        );
        batch.zs_insert(
            &db.cf_handle("block_header_by_height").unwrap(),
            header.height,
            header.header.as_ref(),
        );
    }
    runtime
        .apply_combined(
            TransitionRequest {
                expected_version: before.state_version,
                event: TransitionEvent::FullStateFinalized(FullStateFinalized {
                    full_state_transition_id: evidence,
                    new_finalized: finalized,
                    verified_path_proof: proof,
                }),
            },
            &TransitionContext {
                config: &runtime.config,
                clock: &SystemClock,
                full_state_authority: Some(&authority),
                retention_references: &[],
            },
            batch,
            || {},
        )
        .expect("the authoritative test path finalizes");
}

#[test]
fn capacity_signals_follow_reservations_and_capacity_classes() {
    let mut leases = RetainedPathLeaseRegistry::default();
    let now = Instant::now();
    let first = SourceId::from_digest([0; 32]);
    let (first_id, _) = leases
        .reserve(first, now, RetainedPathCapacity::General)
        .unwrap();
    let Err(RetainedPathLeaseOutcome::CapacityBusy(peer_wait)) =
        leases.reserve(first, now, RetainedPathCapacity::General)
    else {
        panic!("the reservation occupies this peer's slot")
    };
    for marker in 1..MAX_RETAINED_PATH_LEASES - 1 {
        leases
            .reserve(
                SourceId::from_digest([u8::try_from(marker).unwrap(); 32]),
                now,
                RetainedPathCapacity::General,
            )
            .unwrap();
    }
    let waiting = SourceId::from_digest([0xfe; 32]);
    let Err(RetainedPathLeaseOutcome::CapacityBusy(general_wait)) =
        leases.reserve(waiting, now, RetainedPathCapacity::General)
    else {
        panic!("general capacity is full")
    };
    let Err(RetainedPathLeaseOutcome::CapacityBusy(other_wait)) = leases.reserve(
        SourceId::from_digest([0xfd; 32]),
        now,
        RetainedPathCapacity::General,
    ) else {
        panic!("another peer shares the same full capacity class")
    };
    assert_eq!(general_wait, other_wait);
    let fallback = SourceId::from_digest([0xff; 32]);
    let (fallback_id, _) = leases
        .reserve(fallback, now, RetainedPathCapacity::Bounded)
        .unwrap();
    let Err(RetainedPathLeaseOutcome::CapacityBusy(fallback_wait)) =
        leases.reserve(waiting, now, RetainedPathCapacity::Bounded)
    else {
        panic!("fallback capacity is full")
    };
    leases.release_reservation(fallback, fallback_id + 1);
    assert!(!fallback_wait.is_released());
    leases.release_reservation(fallback, fallback_id);
    assert!(fallback_wait.is_released());
    assert!(!general_wait.is_released());
    assert!(!peer_wait.is_released());
    leases.release_reservation(first, first_id);
    assert!(general_wait.is_released());
    assert!(other_wait.is_released());
    assert!(peer_wait.is_released());
    let (replacement, _) = leases
        .reserve(first, now, RetainedPathCapacity::General)
        .unwrap();
    let Err(RetainedPathLeaseOutcome::CapacityBusy(next_general_wait)) =
        leases.reserve(waiting, now, RetainedPathCapacity::General)
    else {
        panic!("another peer acquired the available general slot")
    };
    assert_ne!(general_wait, next_general_wait);
    let Err(RetainedPathLeaseOutcome::CapacityBusy(next_wait)) =
        leases.reserve(first, now, RetainedPathCapacity::General)
    else {
        panic!("the replacement occupies this peer's slot")
    };
    leases.release_reservation(first, first_id);
    assert!(!next_wait.is_released());
    assert!(!next_general_wait.is_released());
    leases.release_reservation(first, replacement);
    assert!(next_wait.is_released());
    assert!(next_general_wait.is_released());
}

#[test]
fn capacity_signal_survives_reservation_commit_and_page_release() {
    let (runtime, _db, _, path) = reconciled_store_with_finalized_prefix(6);
    let reader = runtime.reader();
    let source = SourceId::from_digest([0xd8; 32]);
    let target = path[5].hash;
    let scope = HeaderWorkAuthority::for_target(&runtime.publisher().snapshot(), target);
    let RetainedPathLeaseOutcome::Acquired(lease) = reader
        .acquire_retained_path(source, 7, target, &[path[2].hash], scope)
        .unwrap()
    else {
        panic!("the target is retained")
    };
    let RetainedPathLeaseOutcome::CapacityBusy(signal) = reader
        .acquire_retained_path(source, 8, target, &[path[2].hash], scope)
        .unwrap()
    else {
        panic!("the first session holds this peer's slot")
    };
    assert!(!reader
        .release_retained_path(source, 8, lease.lease_id, scope)
        .unwrap());
    assert!(!signal.is_released());
    assert!(matches!(
        reader
            .read_retained_path(source, 7, lease.lease_id, scope, path[2].hash, 1)
            .unwrap(),
        RetainedPathReadOutcome::Page(_)
    ));
    assert!(signal.is_released());
    assert!(!reader
        .release_retained_path(source, 7, lease.lease_id, scope)
        .unwrap());
}

#[test]
fn dropped_reservation_notifies_waiters() {
    let leases = Arc::new(Mutex::new(RetainedPathLeaseRegistry::default()));
    let peer = SourceId::from_digest([0xd9; 32]);
    let (id, _) = leases
        .lock()
        .unwrap()
        .reserve(peer, Instant::now(), RetainedPathCapacity::General)
        .unwrap();
    let reservation = RetainedPathReservation {
        leases: leases.clone(),
        peer,
        reservation_id: id,
        active: true,
    };
    let Err(RetainedPathLeaseOutcome::CapacityBusy(signal)) =
        leases
            .lock()
            .unwrap()
            .reserve(peer, Instant::now(), RetainedPathCapacity::General)
    else {
        panic!("the reservation occupies this peer's slot")
    };
    assert!(!signal.is_released());
    drop(reservation);
    assert!(signal.is_released());
}

#[test]
fn reservation_commit_keeps_the_waiter_until_exact_release_or_expiry() {
    for expire in [false, true] {
        let mut leases = RetainedPathLeaseRegistry::default();
        let peer = SourceId::from_digest([0xda; 32]);
        let now = Instant::now();
        let (reservation, capacity) = leases
            .reserve(peer, now, RetainedPathCapacity::General)
            .unwrap();
        let Err(RetainedPathLeaseOutcome::CapacityBusy(signal)) =
            leases.reserve(peer, now, RetainedPathCapacity::General)
        else {
            panic!("the reservation occupies this peer's slot")
        };
        let frontier = Frontier::new(block::Height(0), block::Hash([0; 32]));
        let scope = HeaderWorkAuthority {
            header_generation: zakura_header_chain::HeaderGeneration::new(1),
            branch: zakura_header_chain::BranchId::new(frontier.hash, frontier.hash),
        };
        let RetainedPathLeaseOutcome::Acquired(lease) = leases.commit_reservation(
            peer,
            reservation,
            RetainedPathLeaseSpec {
                peer,
                session_id: 7,
                target: frontier,
                common_ancestor: frontier,
                scope,
                position: CanonicalHeaderPathPosition::Complete,
                capacity,
                retained_path: Arc::from([]),
            },
            now,
        ) else {
            panic!("the reservation commits its lease")
        };
        assert!(!signal.is_released());
        assert!(!leases.release(peer, 8, lease.lease_id, scope));
        assert!(!signal.is_released());
        if expire {
            leases.expire(now + RETAINED_PATH_LEASE_IDLE);
        } else {
            assert!(leases.release(peer, 7, lease.lease_id, scope));
        }
        assert!(signal.is_released());
    }
}

#[tokio::test]
async fn abandoned_blocking_acquisition_releases_its_lease_and_notifies_waiters() {
    let (runtime, _db, _, path) = reconciled_store_with_finalized_prefix(7);
    let reader = runtime.reader();
    let source = SourceId::from_digest([0xe5; 32]);
    let target = path[5].hash;
    let scope = HeaderWorkAuthority::for_target(&runtime.publisher().snapshot(), target);
    let outcome = reader
        .acquire_retained_path(source, 7, target, &[path[2].hash], scope)
        .unwrap();
    assert!(matches!(outcome, RetainedPathLeaseOutcome::Acquired(_)));
    let RetainedPathLeaseOutcome::CapacityBusy(signal) = reader
        .acquire_retained_path(source, 7, target, &[path[2].hash], scope)
        .unwrap()
    else {
        panic!("the first acquisition owns the peer slot");
    };
    let (resume, blocked) = std::sync::mpsc::channel();
    let task_reader = reader.clone();
    let task = tokio::task::spawn_blocking(move || {
        blocked.recv().unwrap();
        PendingRetainedPathAcquisition::new(task_reader, outcome)
    });
    drop(task);
    assert!(!signal.is_released());
    resume.send(()).unwrap();
    tokio::time::timeout(std::time::Duration::from_secs(5), signal.released())
        .await
        .unwrap();
    let outcome = reader
        .acquire_retained_path(source, 7, target, &[path[2].hash], scope)
        .unwrap();
    let pending = PendingRetainedPathAcquisition::new(reader.clone(), outcome);
    let RetainedPathLeaseOutcome::Acquired(lease) = pending.into_outcome() else {
        panic!("abandonment must return the peer slot");
    };
    // Receiving the result transfers ownership to the caller.
    assert!(reader
        .release_retained_path(source, 7, lease.lease_id, scope)
        .unwrap());
}

fn retained_checkpoint_fixture() -> (HeaderChainRuntime, DiskDb, Vec<VerifiedHeaderRef>) {
    let (runtime, db, _genesis, path) = reconciled_store_with_finalized_prefix(5);
    let before = runtime.publisher().snapshot();
    let evidence = EvidenceId::from_digest([0xc4; 32]);
    let authority = Authority(evidence);
    runtime
        .apply(
            TransitionRequest {
                expected_version: before.state_version,
                event: TransitionEvent::VerifiedChainChanged(VerifiedChainChanged {
                    full_state_transition_id: evidence,
                    old_tip: before.frontiers.verified_best,
                    new_path: Vec::new(),
                    cause: VerifiedChangeCause::Reset,
                }),
            },
            &TransitionContext {
                config: &runtime.config,
                clock: &SystemClock,
                full_state_authority: Some(&authority),
                retention_references: &[],
            },
        )
        .expect("the fixture resets the verified tip while retaining its headers");
    assert_eq!(
        runtime.publisher().snapshot().frontiers.verified_best,
        before.frontiers.finalized
    );
    (runtime, db, path)
}

#[test]
fn retained_checkpoint_omits_context_but_missing_headers_and_other_causes_keep_it() {
    let (runtime, _db, path) = retained_checkpoint_fixture();
    let before = runtime.publisher().snapshot();
    let make_request = |header: VerifiedHeaderRef, cause| TransitionRequest {
        expected_version: before.state_version,
        event: TransitionEvent::VerifiedChainChanged(VerifiedChainChanged {
            full_state_transition_id: zakura_header_chain::checkpoint_finality_evidence(
                before.state_version,
                Frontier::new(header.height, header.hash),
            ),
            old_tip: before.frontiers.verified_best,
            new_path: vec![header],
            cause,
        }),
    };
    let engine = runtime
        .transition_engine
        .lock()
        .expect("the fixture writer is unlocked");
    let retained = make_request(
        path[3].clone(),
        VerifiedChangeCause::CheckpointFinalizedGrow,
    );
    let input = runtime
        .build_transition_input(retained.clone(), &before, runtime.config.network(), &engine)
        .expect("retained checkpoint input needs no durable context");
    assert!(input
        .header_validation_facts()
        .expect("checkpoint facts exist")
        .validation_leases
        .is_empty());
    let grow = runtime
        .build_transition_input(
            make_request(path[3].clone(), VerifiedChangeCause::Grow),
            &before,
            runtime.config.network(),
            &engine,
        )
        .expect("ordinary growth keeps its context");
    assert_eq!(
        grow.header_validation_facts()
            .expect("growth facts exist")
            .validation_leases
            .len(),
        1
    );
    let mut missing = path[3].clone();
    Arc::make_mut(&mut missing.header).nonce.0[0] ^= 0x80;
    missing.hash = missing.header.hash();
    let input = runtime
        .build_transition_input(
            make_request(missing, VerifiedChangeCause::CheckpointFinalizedGrow),
            &before,
            runtime.config.network(),
            &engine,
        )
        .expect("a missing header receives durable context");
    assert_eq!(
        input
            .header_validation_facts()
            .expect("checkpoint facts exist")
            .validation_leases
            .len(),
        1
    );
    drop(engine);

    let TransitionEvent::VerifiedChainChanged(event) = &retained.event else {
        unreachable!()
    };
    let authority = Authority(event.full_state_transition_id);
    let context = TransitionContext {
        config: &runtime.config,
        clock: &SystemClock,
        full_state_authority: Some(&authority),
        retention_references: &[],
    };
    let mut batch = DiskWriteBatch::new();
    let accepted = Frontier::new(path[3].height, path[3].hash);
    stage_full_state_canonical_hash(&runtime.store, &mut batch, accepted);
    assert!(matches!(
        runtime
            .apply_combined(retained, &context, batch, || {})
            .expect("retained checkpoint commits through the full adapter"),
        ApplyResult::Committed
    ));
    assert_eq!(runtime.publisher().snapshot().frontiers.finalized, accepted);
}

#[test]
fn retained_checkpoint_still_rejects_mismatched_header_identity() {
    let (runtime, _db, path) = retained_checkpoint_fixture();
    let before = runtime.publisher().snapshot();
    let accepted = Frontier::new(path[3].height, path[3].hash);
    let evidence =
        zakura_header_chain::checkpoint_finality_evidence(before.state_version, accepted);
    let authority = Authority(evidence);
    let context = TransitionContext {
        config: &runtime.config,
        clock: &SystemClock,
        full_state_authority: Some(&authority),
        retention_references: &[],
    };
    let mut tampered = path[3].clone();
    Arc::make_mut(&mut tampered.header).nonce.0[0] ^= 0x80;
    let request = TransitionRequest {
        expected_version: before.state_version,
        event: TransitionEvent::VerifiedChainChanged(VerifiedChainChanged {
            full_state_transition_id: evidence,
            old_tip: before.frontiers.verified_best,
            new_path: vec![tampered],
            cause: VerifiedChangeCause::CheckpointFinalizedGrow,
        }),
    };
    assert!(runtime
        .apply_combined(request, &context, DiskWriteBatch::new(), || {})
        .is_err());
    assert_eq!(runtime.publisher().snapshot(), before);
}

/// Read a complete path, reacquiring the target with a continuation locator after each page.
fn read_whole_retained_path(
    reader: &HeaderChainReader,
    owner: SourceId,
    target: block::Hash,
    locators: &[block::Hash],
    scope: HeaderWorkAuthority,
    page_count: u32,
) -> (Frontier, Vec<Arc<block::Header>>) {
    let mut locators = locators.to_vec();
    let mut first_ancestor = None;
    let mut headers = Vec::new();
    loop {
        let RetainedPathLeaseOutcome::Acquired(lease) = reader
            .acquire_retained_path(owner, 1, target, &locators, scope)
            .expect("the target resolves on either side of finality")
        else {
            panic!("the same locator must acquire every target");
        };
        first_ancestor.get_or_insert(lease.common_ancestor);
        let RetainedPathReadOutcome::Page(page) = reader
            .read_retained_path(
                owner,
                1,
                lease.lease_id,
                scope,
                lease.common_ancestor.hash,
                page_count,
            )
            .expect("the cursor reads a coherent page")
        else {
            panic!("the path must remain available");
        };
        assert_eq!(page.target.hash, target);
        assert!(page.aux_deliveries.iter().all(Vec::is_empty));
        headers.extend(page.headers.iter().cloned());
        if page.complete {
            break;
        }
        locators = vec![page
            .headers
            .last()
            .expect("an incomplete page is nonempty")
            .hash()];
    }
    (
        first_ancestor.expect("the loop acquires at least once"),
        headers,
    )
}

#[tokio::test(start_paused = true)]
async fn retained_path_requests_have_identical_rules_across_finality() {
    for finalized_count in [1, 3, 5] {
        let (runtime, _db, genesis, path) = reconciled_store_with_finality(5, finalized_count);
        let reader = runtime.reader();
        let owner = SourceId::from_digest([0xc1; 32]);
        let target = Frontier::new(path[2].height, path[2].hash);
        let scope = HeaderWorkAuthority::for_target(&runtime.publisher().snapshot(), target.hash);
        for page_count in [1, 4] {
            let (ancestor, headers) = read_whole_retained_path(
                &reader,
                owner,
                target.hash,
                &[
                    genesis.hash,
                    path[0].hash,
                    path[4].hash,
                    block::Hash([0xff; 32]),
                ],
                scope,
                page_count,
            );
            assert_eq!(ancestor.hash, path[0].hash);
            assert_eq!(
                headers,
                vec![path[1].header.clone(), path[2].header.clone()]
            );
        }
        let RetainedPathLeaseOutcome::Acquired(lease) = reader
            .acquire_retained_path(owner, 1, target.hash, &[genesis.hash, target.hash], scope)
            .unwrap()
        else {
            panic!("the target itself must intersect in every storage band");
        };
        assert_eq!(lease.common_ancestor, target);
        let RetainedPathReadOutcome::Page(page) = reader
            .read_retained_path(owner, 1, lease.lease_id, scope, target.hash, 1)
            .unwrap()
        else {
            panic!("the empty path must be readable");
        };
        assert!(page.headers.is_empty());
        assert!(page.complete);
        assert_eq!(
            reader
                .acquire_retained_path(owner, 1, target.hash, &[path[4].hash], scope)
                .unwrap(),
            RetainedPathLeaseOutcome::NoLocatorIntersection
        );
    }
}

#[tokio::test(start_paused = true)]
async fn retained_path_reserves_bounded_repair_capacity_for_every_storage_band() {
    let (runtime, _db, genesis, path) =
        reconciled_store_with_finalized_prefix(crate::constants::MAX_HEADER_SYNC_HEIGHT_RANGE + 4);
    let reader = runtime.reader();
    let ordinary_target = path.last().unwrap();
    let ordinary_scope =
        HeaderWorkAuthority::for_target(&runtime.publisher().snapshot(), ordinary_target.hash);
    let mut ordinary_leases = Vec::new();
    for marker in 1..MAX_RETAINED_PATH_LEASES {
        let RetainedPathLeaseOutcome::Acquired(lease) = reader
            .acquire_retained_path(
                SourceId::from_digest([u8::try_from(marker).unwrap(); 32]),
                1,
                ordinary_target.hash,
                &[genesis.hash],
                ordinary_scope,
            )
            .unwrap()
        else {
            panic!("ordinary transfers must fill the general slots");
        };
        ordinary_leases.push(lease);
    }
    let owner = SourceId::from_digest([0xc2; 32]);
    let RetainedPathLeaseOutcome::CapacityBusy(long_wait) = reader
        .acquire_retained_path(
            owner,
            1,
            ordinary_target.hash,
            &[genesis.hash],
            ordinary_scope,
        )
        .unwrap()
    else {
        panic!("a long path rejected from the reserve waits for ordinary capacity");
    };
    assert!(!long_wait.is_released());
    let RetainedPathLeaseOutcome::CapacityBusy(missing_locator_wait) = reader
        .acquire_retained_path(
            owner,
            1,
            ordinary_target.hash,
            &[block::Hash([0xee; 32])],
            ordinary_scope,
        )
        .unwrap()
    else {
        panic!("reserve traversal stops at the range limit before resolving a missing locator");
    };
    assert_eq!(missing_locator_wait, long_wait);

    // Each two-header repair uses the reserve and stops at its admitted target.
    for (target, ancestor) in [
        (&path[1], &genesis),
        (&path[2], &path[0]),
        (&path[3], &path[1]),
    ] {
        let scope = HeaderWorkAuthority::for_target(&runtime.publisher().snapshot(), target.hash);
        let RetainedPathLeaseOutcome::Acquired(lease) = reader
            .acquire_retained_path(owner, 1, target.hash, &[ancestor.hash], scope)
            .unwrap()
        else {
            panic!("bounded repair must acquire the reserved slot");
        };
        assert_eq!(
            runtime.leases.lock().unwrap().by_peer[&owner].capacity,
            RetainedPathCapacity::Bounded
        );
        let RetainedPathLeaseOutcome::CapacityBusy(capacity_signal) = reader
            .acquire_retained_path(
                SourceId::from_digest([0xc3; 32]),
                1,
                target.hash,
                &[ancestor.hash],
                scope,
            )
            .unwrap()
        else {
            panic!("one bounded lease fills the reserve and reports its release signal");
        };
        assert!(!capacity_signal.is_released());
        let RetainedPathReadOutcome::Page(page) = reader
            .read_retained_path(
                owner,
                1,
                lease.lease_id,
                scope,
                ancestor.hash,
                crate::constants::MAX_HEADER_SYNC_HEIGHT_RANGE,
            )
            .unwrap()
        else {
            panic!("bounded repair must read the reserved path");
        };
        assert_eq!(page.headers.len(), 2);
        assert_eq!(page.headers.last().unwrap().hash(), target.hash);
        assert!(page.complete);
        assert!(capacity_signal.is_released());
    }

    let target = &path[3];
    let scope = HeaderWorkAuthority::for_target(&runtime.publisher().snapshot(), target.hash);
    assert!(
        matches!(
            reader
                .acquire_retained_path(owner, 1, target.hash, &[path[1].hash], scope)
                .unwrap(),
            RetainedPathLeaseOutcome::Acquired(_)
        ),
        "a completed page releases the reserve"
    );
    assert!(!long_wait.is_released());
    let released = ordinary_leases.pop().unwrap();
    assert!(reader
        .release_retained_path(released.peer, 1, released.lease_id, ordinary_scope)
        .unwrap());
    assert!(
        long_wait.is_released(),
        "a freed ordinary slot wakes long paths rejected from the reserve"
    );
    assert!(
        matches!(
            reader
                .acquire_retained_path(
                    released.peer,
                    2,
                    ordinary_target.hash,
                    &[genesis.hash],
                    ordinary_scope,
                )
                .unwrap(),
            RetainedPathLeaseOutcome::Acquired(_)
        ),
        "a reserve lease must not consume a released general slot"
    );
}

#[tokio::test(start_paused = true)]
async fn retained_path_long_transfers_use_ordinary_capacity_across_finality() {
    let range = crate::constants::MAX_HEADER_SYNC_HEIGHT_RANGE;
    let target_index = usize::try_from(range).unwrap();
    for finalized_count in [1, target_index + 2] {
        let (runtime, _db, genesis, path) =
            reconciled_store_with_finality(range + 2, finalized_count);
        let reader = runtime.reader();
        let target = &path[target_index];
        let scope = HeaderWorkAuthority::for_target(&runtime.publisher().snapshot(), target.hash);
        let owner = SourceId::from_digest([0xc4; 32]);
        let RetainedPathLeaseOutcome::Acquired(lease) = reader
            .acquire_retained_path(owner, 1, target.hash, &[genesis.hash], scope)
            .unwrap()
        else {
            panic!("long paths must use ordinary capacity in either storage band");
        };
        assert_eq!(
            runtime.leases.lock().unwrap().by_peer[&owner].capacity,
            RetainedPathCapacity::General
        );
        let RetainedPathReadOutcome::Page(first) = reader
            .read_retained_path(owner, 1, lease.lease_id, scope, genesis.hash, range)
            .unwrap()
        else {
            panic!("the first protocol range must be readable");
        };
        assert_eq!(first.headers.len(), target_index);
        assert!(!first.complete);
        let continuation = path[target_index - 1].hash;
        let RetainedPathLeaseOutcome::Acquired(lease) = reader
            .acquire_retained_path(owner, 1, target.hash, &[continuation], scope)
            .unwrap()
        else {
            panic!("the continuation must reacquire the target");
        };
        let RetainedPathReadOutcome::Page(last) = reader
            .read_retained_path(owner, 1, lease.lease_id, scope, continuation, range)
            .unwrap()
        else {
            panic!("the final page must be readable");
        };
        assert_eq!(last.headers, vec![target.header.clone()]);
        assert!(last.complete);
        for marker in 1..MAX_RETAINED_PATH_LEASES {
            assert!(matches!(
                reader
                    .acquire_retained_path(
                        SourceId::from_digest([u8::try_from(marker).unwrap(); 32]),
                        1,
                        target.hash,
                        &[genesis.hash],
                        scope
                    )
                    .unwrap(),
                RetainedPathLeaseOutcome::Acquired(_)
            ));
        }
        let repair_owner = SourceId::from_digest([0xc5; 32]);
        let RetainedPathLeaseOutcome::CapacityBusy(long_wait) = reader
            .acquire_retained_path(repair_owner, 1, target.hash, &[genesis.hash], scope)
            .unwrap()
        else {
            panic!("a long path waits for ordinary capacity in either storage band");
        };
        assert!(!long_wait.is_released());
        assert!(
            matches!(
                reader
                    .acquire_retained_path(repair_owner, 1, target.hash, &[path[0].hash], scope)
                    .unwrap(),
                RetainedPathLeaseOutcome::Acquired(_)
            ),
            "a protocol-sized repair must still use the reserve in either storage band"
        );
    }
}

#[test]
fn retained_path_reservations_keep_capacity_classes_separate() {
    let mut registry = RetainedPathLeaseRegistry::default();
    let now = Instant::now();
    let peers: Vec<_> = (0..MAX_RETAINED_PATH_LEASES)
        .map(|marker| SourceId::from_digest([u8::try_from(marker).unwrap(); 32]))
        .collect();
    let mut reservations = Vec::new();
    for peer in &peers {
        reservations.push(
            registry
                .reserve(*peer, now, RetainedPathCapacity::Bounded)
                .unwrap(),
        );
    }
    assert_eq!(
        reservations.last().unwrap().1,
        RetainedPathCapacity::Bounded
    );
    let Err(RetainedPathLeaseOutcome::CapacityBusy(general_wait)) = registry.reserve(
        SourceId::from_digest([0xfe; 32]),
        now,
        RetainedPathCapacity::General,
    ) else {
        panic!("ordinary reservations fill the general class")
    };
    registry.release_reservation(peers[0], reservations[0].0);
    assert!(
        general_wait.is_released(),
        "the reserve must not hide a freed ordinary slot"
    );
    assert_eq!(
        registry
            .reserve(peers[0], now, RetainedPathCapacity::General)
            .unwrap()
            .1,
        RetainedPathCapacity::General
    );
    assert!(registry
        .reserve(peers[0], now, RetainedPathCapacity::Bounded)
        .is_err());
    assert!(registry
        .reserve(
            SourceId::from_digest([0xff; 32]),
            now,
            RetainedPathCapacity::Bounded
        )
        .is_err());
}

#[test]
fn reserve_rejection_reports_an_ordinary_slot_freed_after_admission() {
    let mut registry = RetainedPathLeaseRegistry::default();
    let now = Instant::now();
    let peers: Vec<_> = (0..MAX_RETAINED_PATH_LEASES)
        .map(|marker| SourceId::from_digest([u8::try_from(marker).unwrap(); 32]))
        .collect();
    let reservations: Vec<_> = peers
        .iter()
        .map(|peer| {
            registry
                .reserve(*peer, now, RetainedPathCapacity::Bounded)
                .unwrap()
        })
        .collect();
    let (reserve_id, class) = *reservations.last().unwrap();
    assert_eq!(class, RetainedPathCapacity::Bounded);
    let Err(RetainedPathLeaseOutcome::CapacityBusy(reserve_wait)) = registry.reserve(
        SourceId::from_digest([0xfe; 32]),
        now,
        RetainedPathCapacity::Bounded,
    ) else {
        panic!("the reserve is occupied");
    };

    // An ordinary slot frees while the reserve-admitted request walks its ancestry.
    registry.release_reservation(peers[0], reservations[0].0);
    let RetainedPathLeaseOutcome::CapacityBusy(ordinary_wait) =
        registry.reject_beyond_reserve(*peers.last().unwrap(), reserve_id)
    else {
        panic!("a reserve rejection returns the ordinary capacity signal");
    };
    assert!(
        ordinary_wait.is_released(),
        "the requester must not wait for a slot that is already free"
    );
    assert!(
        reserve_wait.is_released(),
        "freeing any slot wakes bounded waiters"
    );
    assert!(!registry.reservations.contains_key(peers.last().unwrap()));
}

#[test]
fn retained_path_busy_admission_does_not_lock_the_engine() {
    let (runtime, _db, genesis, path) = reconciled_store_with_finalized_prefix(4);
    let reader = runtime.reader();
    let target = path.last().unwrap();
    let scope = HeaderWorkAuthority::for_target(&runtime.publisher().snapshot(), target.hash);
    let owner = SourceId::from_digest([0xc6; 32]);
    assert!(matches!(
        reader
            .acquire_retained_path(owner, 1, target.hash, &[genesis.hash], scope)
            .unwrap(),
        RetainedPathLeaseOutcome::Acquired(_)
    ));
    let engine = reader.transition_engine.clone();
    let _ = std::thread::spawn(move || {
        let _guard = engine.lock().unwrap();
        panic!("poison the engine to detect any access after failed admission");
    })
    .join();
    assert!(matches!(
        reader
            .acquire_retained_path(owner, 2, target.hash, &[genesis.hash], scope)
            .unwrap(),
        RetainedPathLeaseOutcome::CapacityBusy(_)
    ));
    for marker in 0..MAX_RETAINED_PATH_LEASES - 1 {
        reader
            .leases
            .lock()
            .unwrap()
            .reserve(
                SourceId::from_digest([u8::try_from(marker).unwrap(); 32]),
                Instant::now(),
                RetainedPathCapacity::Bounded,
            )
            .unwrap();
    }
    assert!(matches!(
        reader
            .acquire_retained_path(
                SourceId::from_digest([0xff; 32]),
                1,
                target.hash,
                &[genesis.hash],
                scope
            )
            .unwrap(),
        RetainedPathLeaseOutcome::CapacityBusy(_)
    ));
}

#[test]
fn stored_validation_context_equals_the_lease_built_from_rehashed_headers() {
    let path_len = u8::try_from(zakura_header_chain::MAX_POW_PREDECESSOR_CONTEXT_SPAN + 6)
        .expect("the predecessor span fixture is shorter than 256");
    let (runtime, _, _, path) = reconciled_store_with_finalized_prefix(u32::from(path_len));
    let (engine_config, _, _) = fixture();
    let network = engine_config.network();
    let parent = path.last().expect("the fixture path is nonempty");

    let lease = runtime
        .store
        .validation_context(parent.hash, network)
        .expect("the retained parent has a coherent validation context");

    // Rebuild the lease the way it was built before hash reuse: hash every context header.
    let parent_node = runtime
        .store
        .header_node(parent.hash)
        .expect("the parent row decodes")
        .expect("the parent is retained");
    let mut facts = vec![zakura_header_chain::HeaderContextFact {
        frontier: Frontier::new(parent_node.height, parent.hash),
        header: parent_node.header.clone(),
    }];
    facts.extend(
        authenticated_context_headers(&runtime.store, parent.hash, None)
            .expect("the stored context authenticates")
            .into_iter()
            .rev()
            .map(|context| context.fact()),
    );
    assert_eq!(
        facts.len(),
        zakura_header_chain::MAX_POW_PREDECESSOR_CONTEXT_SPAN + 1
    );
    for fact in &facts {
        assert_eq!(fact.frontier.hash, fact.header.hash());
    }
    let expected = zakura_header_chain::ValidationLease::new(
        lease.parent(),
        facts,
        network.clone(),
        runtime
            .store
            .metadata()
            .expect("the metadata row exists")
            .anchor_manifest_digest,
    );
    assert_eq!(lease, expected);
}

#[test]
fn stored_validation_context_rejects_a_predecessor_row_with_altered_header_bytes() {
    let (runtime, _, _, path) = reconciled_store_with_finalized_prefix(8);
    let (engine_config, _, _) = fixture();
    let parent = &path[7];
    let altered = path[5].hash;
    runtime
        .store
        .validation_context(parent.hash, engine_config.network())
        .expect("the unaltered context is coherent");

    let cf = runtime
        .store
        .cf(HEADER_NODE_BY_HASH)
        .expect("the node column family exists");
    let mut row = runtime
        .store
        .db
        .raw_get_cf(&cf, &altered.0)
        .expect("the node row reads")
        .expect("the predecessor is retained");
    // The row starts with a four-byte header length, then the header: version (4), previous
    // hash (32), merkle root (32), and commitment (32) precede the four-byte time.
    row[4 + 100] ^= 0x01;
    let mut batch = DiskWriteBatch::new();
    runtime
        .store
        .put_raw(&mut batch, HEADER_NODE_BY_HASH, altered.0, &row)
        .expect("the altered row is addressable");
    runtime
        .store
        .db
        .write(batch)
        .expect("the fixture corruption is durable");

    assert!(matches!(
        runtime
            .store
            .validation_context(parent.hash, engine_config.network()),
        Err(StoreError::Incoherent("durable header-chain read failed"))
    ));
}

/// Rewrites the stored node row of `hash` with `rewrite`, durably.
fn rewrite_node_row(
    runtime: &HeaderChainRuntime,
    hash: block::Hash,
    rewrite: impl FnOnce(&mut Vec<u8>),
) {
    let cf = runtime
        .store
        .cf(HEADER_NODE_BY_HASH)
        .expect("the node column family exists");
    let mut row = runtime
        .store
        .db
        .raw_get_cf(&cf, &hash.0)
        .expect("the node row reads")
        .expect("the node is retained");
    rewrite(&mut row);
    let mut batch = DiskWriteBatch::new();
    runtime
        .store
        .put_raw(&mut batch, HEADER_NODE_BY_HASH, hash.0, &row)
        .expect("the row is addressable");
    runtime
        .store
        .db
        .write(batch)
        .expect("the fixture corruption is durable");
}

#[test]
fn validation_context_rejects_truncated_and_hash_flipped_predecessor_rows() {
    use zakura_chain::serialization::ZcashSerialize;

    for corruption in ["truncated", "stored hash flipped"] {
        let (runtime, path, network) = hashed_evidence_fixture();
        let parent = path.last().expect("the fixture path is nonempty").hash;
        let corrupted = path[path.len() - 5].hash;
        rewrite_node_row(&runtime, corrupted, |row| {
            let (node, _) = HeaderNodeDisk::decode_hashed(row).expect("the stored row decodes");
            let header_len = node
                .header
                .zcash_serialize_to_vec()
                .expect("a decoded header serializes")
                .len();
            match corruption {
                // Cut the row inside the stored hash that follows the header.
                "truncated" => row.truncate(4 + header_len + 16),
                // The row is a four-byte length, the header, then the redundant stored hash.
                _ => row[4 + header_len] ^= 0x01,
            }
        });

        assert!(
            matches!(
                runtime.store.validation_context(parent, &network),
                Err(StoreError::Incoherent("durable header-chain read failed"))
            ),
            "{corruption} row must fail the integrity-checking decode"
        );
    }
}

#[test]
fn validation_context_rejects_rows_with_tampered_parent_or_work_fields() {
    use crate::service::finalized_state::disk_format::FallibleDiskValue;

    for field in ["parent_hash", "block_work"] {
        let (runtime, path, network) = hashed_evidence_fixture();
        let parent = path.last().expect("the fixture path is nonempty").hash;
        let tampered = path[path.len() - 5].hash;
        // Re-encode a well-formed row: the header bytes and stored hash stay consistent, so
        // only the hashed-header node construction can notice the contradicted field.
        rewrite_node_row(&runtime, tampered, |row| {
            let (mut node, _) = HeaderNodeDisk::decode_hashed(row).expect("the stored row decodes");
            match field {
                "parent_hash" => node.parent_hash.0[0] ^= 0x01,
                // Valid headers have nonzero work, so doubling it changes the field.
                _ => node.block_work <<= 1,
            }
            *row = node.encode().expect("a tampered field still encodes");
        });

        assert!(
            matches!(
                runtime.store.validation_context(parent, &network),
                Err(StoreError::Incoherent("invalid durable node"))
            ),
            "a row with a tampered {field} must be rejected"
        );
    }
}

#[test]
fn validation_context_notices_corruption_written_between_two_builds() {
    let (runtime, path, network) = hashed_evidence_fixture();
    let parent = path.last().expect("the fixture path is nonempty").hash;
    runtime
        .store
        .validation_context(parent, &network)
        .expect("the uncorrupted context is coherent");

    // A later build must read the row again rather than reuse anything from the first build.
    rewrite_node_row(&runtime, path[path.len() - 5].hash, |row| {
        row[4 + 100] ^= 0x01;
    });
    assert!(matches!(
        runtime.store.validation_context(parent, &network),
        Err(StoreError::Incoherent("durable header-chain read failed"))
    ));
}

fn hashed_evidence_fixture() -> (HeaderChainRuntime, Vec<VerifiedHeaderRef>, Network) {
    let path_len = u8::try_from(zakura_header_chain::MAX_POW_PREDECESSOR_CONTEXT_SPAN + 6)
        .expect("the predecessor span fixture is shorter than 256");
    let (runtime, _, _, path) = reconciled_store_with_finalized_prefix(u32::from(path_len));
    let (engine_config, _, _) = fixture();
    (runtime, path, engine_config.network().clone())
}

#[test]
fn repeated_validation_contexts_reread_every_row_and_build_identical_leases() {
    let (runtime, path, network) = hashed_evidence_fixture();
    let parent = path.last().expect("the fixture path is nonempty").hash;

    HeaderChainStore::reset_header_node_disk_reads();
    let first = runtime
        .store
        .validation_context(parent, &network)
        .expect("coherent");
    let first_reads = HeaderChainStore::header_node_disk_reads();
    HeaderChainStore::reset_header_node_disk_reads();
    let second = runtime
        .store
        .validation_context(parent, &network)
        .expect("coherent");
    let second_reads = HeaderChainStore::header_node_disk_reads();

    assert_eq!(first, second);
    // The parent is read by the build and again by the context walk, then one read per
    // predecessor: no row is served from an earlier build.
    let expected = u64::try_from(zakura_header_chain::MAX_POW_PREDECESSOR_CONTEXT_SPAN + 2)
        .expect("the span fits in u64");
    assert_eq!(first_reads, expected);
    assert_eq!(second_reads, expected);
}

#[test]
fn validation_context_rejects_a_missing_predecessor_row() {
    let (runtime, path, network) = hashed_evidence_fixture();
    let parent = path.last().expect("the fixture path is nonempty").hash;
    let missing = path[path.len() - 5].hash;
    let mut batch = DiskWriteBatch::new();
    runtime
        .store
        .delete_raw(&mut batch, HEADER_NODE_BY_HASH, missing.0)
        .expect("the node row is addressable");
    runtime
        .store
        .db
        .write(batch)
        .expect("the deletion is durable");

    assert!(matches!(
        runtime.store.validation_context(parent, &network),
        Err(StoreError::Incoherent("validation context has a gap"))
    ));
}

#[test]
fn validation_context_rejects_a_valid_row_stored_under_another_hash() {
    let (runtime, path, network) = hashed_evidence_fixture();
    let parent = path.last().expect("the fixture path is nonempty").hash;
    let replaced = path[path.len() - 5].hash;
    let other = path[path.len() - 6].hash;
    let cf = runtime
        .store
        .cf(HEADER_NODE_BY_HASH)
        .expect("the node column family exists");
    let other_row = runtime
        .store
        .db
        .raw_get_cf(&cf, &other.0)
        .expect("the node row reads")
        .expect("the other predecessor is retained");
    let mut batch = DiskWriteBatch::new();
    runtime
        .store
        .put_raw(&mut batch, HEADER_NODE_BY_HASH, replaced.0, &other_row)
        .expect("the row is addressable");
    runtime
        .store
        .db
        .write(batch)
        .expect("the replacement is durable");

    assert!(matches!(
        runtime.store.validation_context(parent, &network),
        Err(StoreError::Incoherent("node key/hash mismatch"))
    ));
}

#[test]
fn header_evidence_from_another_row_is_not_used() {
    let (runtime, path, _) = hashed_evidence_fixture();
    let cf = runtime
        .store
        .cf(HEADER_NODE_BY_HASH)
        .expect("the node column family exists");
    let raw = |hash: block::Hash| {
        runtime
            .store
            .db
            .raw_get_cf(&cf, &hash.0)
            .expect("the node row reads")
            .expect("the node is retained")
    };
    let a = path[path.len() - 2].hash;
    let b = path[path.len() - 3].hash;
    let (row_a, evidence_a) = HeaderNodeDisk::decode_hashed(&raw(a)).expect("row a decodes");
    let (_, evidence_b) = HeaderNodeDisk::decode_hashed(&raw(b)).expect("row b decodes");
    assert_eq!(evidence_a.hash(), a);
    assert_eq!(evidence_b.hash(), b);

    let expected = row_a
        .clone()
        .into_domain(Vec::new())
        .expect("row a is coherent");
    let with_foreign = row_a
        .clone()
        .into_domain_hashed(evidence_b, Vec::new())
        .expect("foreign evidence is ignored and row a is re-hashed");
    let with_own = row_a
        .into_domain_hashed(evidence_a, Vec::new())
        .expect("row a is coherent");
    assert_eq!(with_foreign, expected);
    assert_eq!(with_own, expected);
    assert_eq!(expected.hash, a);

    // Evidence for an equal header in a different allocation is not used either.
    let (mut row_copy, evidence_copy) =
        HeaderNodeDisk::decode_hashed(&raw(a)).expect("row a decodes");
    row_copy.header = Arc::new(*row_copy.header);
    assert!(!Arc::ptr_eq(evidence_copy.header(), &row_copy.header));
    assert_eq!(
        row_copy
            .into_domain_hashed(evidence_copy, Vec::new())
            .expect("row a is coherent"),
        expected
    );
}

/// Replace one retained predecessor's node row with a `HEADER_VALIDATION_CONTEXT` row.
fn replace_node_with_context_row(
    runtime: &HeaderChainRuntime,
    target: &VerifiedHeaderRef,
    context: HeaderValidationContextDisk,
) {
    let mut batch = DiskWriteBatch::new();
    runtime
        .store
        .delete_raw(&mut batch, HEADER_NODE_BY_HASH, target.hash.0)
        .expect("the node row is addressable");
    runtime
        .store
        .put_value(
            &mut batch,
            HEADER_VALIDATION_CONTEXT,
            target.hash.0,
            &context,
        )
        .expect("the context row encodes");
    runtime
        .store
        .db
        .write(batch)
        .expect("the replacement is durable");
}

#[test]
fn validation_context_fallback_rows_build_the_same_lease_and_reject_tampering() {
    let (runtime, path, network) = hashed_evidence_fixture();
    let parent = path.last().expect("the fixture path is nonempty").hash;
    let target = path[path.len() - 5].clone();
    let original = runtime
        .store
        .validation_context(parent, &network)
        .expect("coherent");

    // A coherent fallback row builds exactly the same lease.
    replace_node_with_context_row(
        &runtime,
        &target,
        HeaderValidationContextDisk {
            header: target.header.clone(),
            height: target.height,
        },
    );
    assert_eq!(
        runtime
            .store
            .validation_context(parent, &network)
            .expect("coherent"),
        original
    );

    // A wrong height is rejected.
    replace_node_with_context_row(
        &runtime,
        &target,
        HeaderValidationContextDisk {
            header: target.header.clone(),
            height: target
                .height
                .next()
                .expect("the fixture height has a successor"),
        },
    );
    assert!(matches!(
        runtime.store.validation_context(parent, &network),
        Err(StoreError::Incoherent(
            "invalid immutable validation context"
        ))
    ));

    // Different header bytes under the same key are rejected.
    let mut altered = *target.header;
    altered.time += chrono::Duration::seconds(1);
    replace_node_with_context_row(
        &runtime,
        &target,
        HeaderValidationContextDisk {
            header: Arc::new(altered),
            height: target.height,
        },
    );
    assert!(matches!(
        runtime.store.validation_context(parent, &network),
        Err(StoreError::Incoherent(
            "invalid immutable validation context"
        ))
    ));
}