zakura-state 11.0.0

//! Startup fixtures for the single-protocol header DAG cutover.

use std::sync::Arc;

use zakura_chain::{
    block::{self, Height},
    parameters::{Network, NetworkKind},
    work::difficulty::U256,
};
use zakura_header_chain::{
    prepare_headers, CheckpointSet, EngineConfig, EngineMode, Frontier, HeaderBatchInput,
    HeaderChainDiskVersion, HeaderRules, RowLimit, StoreAuditRead, StoreAuditSnapshot, SystemClock,
    TrustedAnchor, MAX_NON_FINALIZED_NODES_V1,
};

use super::{
    super::{
        ZAKURA_HEADER_BODY_SIZE_BY_HEIGHT, ZAKURA_HEADER_BY_HEIGHT, ZAKURA_HEADER_HASH_BY_HEIGHT,
        ZAKURA_HEADER_HEIGHT_BY_HASH,
    },
    common::{mainnet_block, state_with_genesis_config, write_full_block_header_and_transactions},
};
use crate::{
    service::finalized_state::{
        disk_db::{DiskWriteBatch, WriteDisk},
        disk_format::{
            header_chain_values::{decode_v4_engine_metadata, HeaderChainValueError},
            RawBytes,
        },
        header_chain::{
            migration::{initialize_header_chain_reconciled, HeaderChainInitializationError},
            HeaderChainStore, HeaderChainStoreError,
        },
        FallibleDiskValue, ZakuraDb,
    },
    Config,
};

use crate::service::finalized_state::{HEADER_ENGINE_META, HEADER_VALIDATION_CONTEXT};

/// Writes a synthetic finalized header chain from `genesis` to `chain_tip`, and
/// returns its headers indexed by height.
fn write_synthetic_finalized_headers(
    state: &ZakuraDb,
    genesis: &Arc<block::Block>,
    chain_tip: u32,
) -> Vec<Arc<block::Header>> {
    let header_cf = state
        .db
        .cf_handle("block_header_by_height")
        .expect("the full-state header column exists");
    let hash_cf = state
        .db
        .cf_handle("hash_by_height")
        .expect("the full-state hash column exists");
    let height_cf = state
        .db
        .cf_handle("height_by_hash")
        .expect("the full-state reverse hash column exists");
    let mut headers = vec![genesis.header.clone()];
    let mut full_state = DiskWriteBatch::new();
    for height in 1..=chain_tip {
        let previous = headers
            .last()
            .expect("the synthetic chain starts at genesis");
        let mut header = **previous;
        header.previous_block_hash = previous.hash();
        header.time += chrono::Duration::seconds(1);
        header.nonce.0[0] =
            u8::try_from(height).expect("the synthetic chain is shorter than 256 blocks");
        let header = Arc::new(header);
        let hash = header.hash();
        let height = Height(height);
        full_state.zs_insert(&header_cf, height, &header);
        full_state.zs_insert(&hash_cf, height, hash);
        full_state.zs_insert(&height_cf, hash, height);
        headers.push(header);
    }
    state
        .db
        .write(full_state)
        .expect("the synthetic finalized header chain writes");

    headers
}

fn engine_config(network: Network, genesis: &Arc<block::Block>) -> EngineConfig {
    let frontier = Frontier::new(Height(0), genesis.hash());
    EngineConfig::new(
        EngineMode::Integrated,
        network,
        TrustedAnchor {
            frontier,
            header: genesis.header.clone(),
        },
        CheckpointSet::default(),
    )
    .expect("the fixture has an authenticated genesis anchor")
}

#[test]
fn clean_store_initializes_only_from_finalized_full_state() {
    let _init_guard = zakura_test::init();
    let network = Network::Mainnet;
    let genesis = mainnet_block(0);
    let block1 = mainnet_block(1);
    let block2 = mainnet_block(2);
    let state = state_with_genesis_config(&network, genesis.clone(), Config::ephemeral());
    write_full_block_header_and_transactions(&state, block1.clone());
    let config = engine_config(network.clone(), &genesis);

    let (runtime, report) = initialize_header_chain_reconciled(&state, &config, Vec::new())
        .expect("an empty overlay initializes from authenticated full state");
    let anchor = Frontier::new(Height(1), block1.hash());
    assert_eq!(report.anchor, anchor);
    assert_eq!(report.validation_context_rows, 1);
    assert_eq!(report.startup.current.frontiers.header_best, anchor);
    assert_eq!(runtime.publisher().snapshot(), report.startup.current);
    let store = HeaderChainStore::new(state.header_chain_disk_db());
    let audit = store.audit_snapshot().expect("the audit snapshot opens");
    let metadata = audit.metadata().expect("the initialized metadata decodes");
    assert_eq!(metadata.work_origin, anchor);
    let mut nodes = Vec::new();
    audit
        .visit_header_nodes(RowLimit::new(MAX_NON_FINALIZED_NODES_V1 + 1), &mut |node| {
            nodes.push(node);
            Ok(())
        })
        .expect("the initialized node decodes");
    assert_eq!(nodes.len(), 1);
    assert_eq!(nodes[0].work_coordinate().origin_hash(), anchor.hash);
    assert_eq!(nodes[0].work_coordinate().cumulative_work(), U256::zero());
    assert_eq!(
        store
            .selected_projection()
            .expect("the initialized selection decodes"),
        vec![anchor]
    );

    let lease = runtime
        .reader()
        .validation_context(anchor.hash)
        .expect("the authenticated context read succeeds")
        .expect("the finalized anchor is retained");
    assert_eq!(
        lease.predecessors().len(),
        2,
        "the lease contains the anchor and its one available predecessor"
    );
    let rules = HeaderRules::for_validation_lease(&lease)
        .expect("the production validation policy is authenticated");
    prepare_headers(
        HeaderBatchInput::new(std::slice::from_ref(&block2.header)),
        lease.parent(),
        &rules,
        &SystemClock,
    )
    .expect("the first post-anchor header validates from the seeded context");
}

#[test]
fn predecessor_overlay_is_atomically_replaced_from_finalized_state() {
    let _init_guard = zakura_test::init();
    let network = Network::Mainnet;
    let genesis = mainnet_block(0);
    let block1 = mainnet_block(1);
    let block2 = mainnet_block(2);
    let block3 = mainnet_block(3);
    let state = state_with_genesis_config(&network, genesis.clone(), Config::ephemeral());
    write_full_block_header_and_transactions(&state, block1.clone());
    let header_cf = state
        .db
        .cf_handle(ZAKURA_HEADER_BY_HEIGHT)
        .expect("the obsolete column remains physically present");
    let hash_cf = state
        .db
        .cf_handle(ZAKURA_HEADER_HASH_BY_HEIGHT)
        .expect("the obsolete column remains physically present");
    let height_cf = state
        .db
        .cf_handle(ZAKURA_HEADER_HEIGHT_BY_HASH)
        .expect("the obsolete column remains physically present");
    let body_size_cf = state
        .db
        .cf_handle(ZAKURA_HEADER_BODY_SIZE_BY_HEIGHT)
        .expect("the advisory body-size column remains physically present");
    let mut legacy = DiskWriteBatch::new();
    legacy.zs_insert(&header_cf, Height(2), &block2.header);
    legacy.zs_insert(&header_cf, Height(3), &block3.header);
    legacy.zs_insert(&hash_cf, Height(2), block2.hash());
    legacy.zs_insert(&hash_cf, Height(3), block3.hash());
    legacy.zs_insert(&height_cf, block2.hash(), Height(2));
    legacy.zs_insert(&height_cf, block3.hash(), Height(3));
    legacy.zs_insert(
        &body_size_cf,
        Height(2),
        RawBytes::new_raw_bytes(1_u32.to_be_bytes().to_vec()),
    );
    state
        .db
        .write(legacy)
        .expect("the legacy fixture row writes");
    let config = engine_config(network, &genesis);

    let (runtime, report) = initialize_header_chain_reconciled(&state, &config, Vec::new())
        .expect("obsolete overlay rows are replaced from authenticated full state");
    let anchor = Frontier::new(Height(1), block1.hash());
    assert_eq!(report.anchor, anchor);
    assert_eq!(report.startup.current.frontiers.header_best, anchor);
    assert_eq!(runtime.publisher().snapshot(), report.startup.current);
    assert_eq!(state.hash(Height(1)), Some(block1.hash()));
    assert_eq!(
        HeaderChainStore::new(state.header_chain_disk_db())
            .selected_projection()
            .expect("the initialized selection decodes"),
        vec![anchor]
    );
    for family in [
        ZAKURA_HEADER_BY_HEIGHT,
        ZAKURA_HEADER_HASH_BY_HEIGHT,
        ZAKURA_HEADER_HEIGHT_BY_HASH,
    ] {
        let cf = state
            .db
            .cf_handle(family)
            .expect("the obsolete column remains physically present");
        assert!(
            state
                .db
                .raw_range_cf(&cf, &[], None)
                .expect("the obsolete column remains readable")
                .is_empty(),
            "{family} is empty after migration"
        );
    }
    assert_eq!(
        state
            .db
            .raw_range_cf(&body_size_cf, &[], None)
            .expect("the advisory body-size column remains readable")
            .len(),
        1,
        "migration preserves the advisory body-size column"
    );
    assert!(matches!(
        initialize_header_chain_reconciled(&state, &config, Vec::new()),
        Err(HeaderChainInitializationError::AlreadyInitialized)
    ));
    assert_eq!(
        runtime
            .reader()
            .validation_context(anchor.hash)
            .expect("the authenticated context read succeeds")
            .expect("the finalized anchor is retained")
            .predecessors()
            .len(),
        2
    );
}

#[test]
fn existing_narrow_validation_context_is_backfilled_before_startup() {
    let _init_guard = zakura_test::init();
    let network = Network::Mainnet;
    let genesis = mainnet_block(0);
    let state = state_with_genesis_config(&network, genesis.clone(), Config::ephemeral());
    let predecessor_span = zakura_header_chain::MAX_POW_PREDECESSOR_CONTEXT_SPAN;
    let chain_tip = u32::try_from(predecessor_span + 1)
        .expect("the validation context span fits in a block height");
    write_synthetic_finalized_headers(&state, &genesis, chain_tip);

    let config = engine_config(network, &genesis);
    let (runtime, report) = initialize_header_chain_reconciled(&state, &config, Vec::new())
        .expect("the complete validation context initializes");
    assert_eq!(report.validation_context_rows, predecessor_span);
    drop(runtime);

    let store = HeaderChainStore::new(state.header_chain_disk_db());
    let mut contexts = Vec::new();
    store
        .audit_snapshot()
        .expect("the initialized store has an audit snapshot")
        .visit_validation_context_records(RowLimit::new(predecessor_span), &mut |record| {
            contexts.push(record);
            Ok(())
        })
        .expect("the validation context rows decode");
    contexts.sort_unstable_by_key(|record| record.height);

    // A build without ZIP 218 retained 17 averaging-window headers plus 10
    // additional median-time headers below the finalized anchor.
    let old_predecessor_span = 27;
    let mut downgrade = DiskWriteBatch::new();
    let context_cf = state
        .db
        .cf_handle(HEADER_VALIDATION_CONTEXT)
        .expect("the validation context column exists");
    for context in contexts
        .iter()
        .take(predecessor_span - old_predecessor_span)
    {
        downgrade.zs_delete(&context_cf, context.header.hash());
    }
    // That build also recorded header-chain disk format 4.
    let metadata_cf = state
        .db
        .cf_handle(HEADER_ENGINE_META)
        .expect("the header-chain metadata column exists");
    let mut metadata = state
        .db
        .raw_get_cf(&metadata_cf, b"")
        .expect("the metadata row reads")
        .expect("the initialized store has metadata");
    metadata[..4].copy_from_slice(&4_u32.to_be_bytes());
    downgrade.zs_insert(
        &metadata_cf,
        RawBytes::new_raw_bytes(Vec::new()),
        RawBytes::new_raw_bytes(metadata.clone()),
    );
    state
        .db
        .write(downgrade)
        .expect("the pre-ZIP 218 context fixture writes");

    // A different network kind must still be rejected without changing its bytes,
    // even though Mainnet permits a network policy change.
    let mut incompatible =
        decode_v4_engine_metadata(&metadata).expect("the version-four metadata fixture decodes");
    incompatible.network_id = NetworkKind::Testnet;
    let incompatible = incompatible
        .encode()
        .expect("the incompatible metadata fixture encodes");
    let mut corrupt = DiskWriteBatch::new();
    corrupt.zs_insert(
        &metadata_cf,
        RawBytes::new_raw_bytes(Vec::new()),
        RawBytes::new_raw_bytes(incompatible.clone()),
    );
    state
        .db
        .write(corrupt)
        .expect("the incompatible metadata fixture writes");
    assert!(matches!(
        store.migrate_to_current(&config),
        Err(HeaderChainStoreError::Incoherent(
            "legacy network kind does not match the configured network"
        ))
    ));
    assert_eq!(
        state
            .db
            .raw_get_cf(&metadata_cf, b"")
            .expect("the rejected metadata row reads")
            .expect("the rejected metadata row remains present"),
        incompatible,
    );
    let mut restore = DiskWriteBatch::new();
    restore.zs_insert(
        &metadata_cf,
        RawBytes::new_raw_bytes(Vec::new()),
        RawBytes::new_raw_bytes(metadata),
    );
    state
        .db
        .write(restore)
        .expect("the valid version-four metadata fixture is restored");

    assert!(store
        .migrate_to_current(&config)
        .expect("version four migrates to the current format"));
    let metadata = state
        .db
        .raw_get_cf(&metadata_cf, b"")
        .expect("the metadata row reads")
        .expect("the migrated store has metadata");
    assert_eq!(
        metadata[..4],
        HeaderChainDiskVersion::CURRENT.0.to_be_bytes()
    );
    // The previous release accepts only format 4, so it reports the newer format
    // instead of reading a validation context wider than its row limit.
    assert_eq!(
        decode_v4_engine_metadata(&metadata),
        Err(HeaderChainValueError::UnsupportedDiskFormat(
            HeaderChainDiskVersion::CURRENT.0
        ))
    );

    // Model a restart after the version marker commits but before the context
    // resize. A fresh store must complete the idempotent backfill.
    let store = HeaderChainStore::new(state.header_chain_disk_db());
    assert_eq!(
        store
            .resize_validation_context(&state)
            .expect("authenticated full state backfills the wider context"),
        predecessor_span - old_predecessor_span,
    );
    let (_, startup) = store
        .startup(&config)
        .expect("startup accepts the backfilled validation context");
    assert!(startup.publication_allowed);
    let store = HeaderChainStore::new(state.header_chain_disk_db());
    assert_eq!(store.resize_validation_context(&state).unwrap(), 0);
    assert!(store.startup(&config).unwrap().1.publication_allowed);
}

#[test]
fn predecessor_overlay_is_preserved_when_full_state_authentication_fails() {
    let _init_guard = zakura_test::init();
    let network = Network::Mainnet;
    let genesis = mainnet_block(0);
    let block1 = mainnet_block(1);
    let state = state_with_genesis_config(&network, genesis, Config::ephemeral());
    let header_cf = state
        .db
        .cf_handle(ZAKURA_HEADER_BY_HEIGHT)
        .expect("the obsolete column remains physically present");
    let mut legacy = DiskWriteBatch::new();
    legacy.zs_insert(&header_cf, Height(1), &block1.header);
    state
        .db
        .write(legacy)
        .expect("the legacy fixture row writes");
    let before = state
        .db
        .raw_range_cf(&header_cf, &[], None)
        .expect("the predecessor row can be observed without decoding it");
    let config = EngineConfig::new(
        EngineMode::Integrated,
        network,
        TrustedAnchor {
            frontier: Frontier::new(Height(1), block1.hash()),
            header: block1.header.clone(),
        },
        CheckpointSet::default(),
    )
    .expect("the fixture has an authenticated but unavailable anchor");

    assert!(matches!(
        initialize_header_chain_reconciled(&state, &config, Vec::new()),
        Err(HeaderChainInitializationError::AnchorMismatch)
    ));
    assert!(HeaderChainStore::new(state.header_chain_disk_db())
        .metadata()
        .is_err());
    assert_eq!(
        state
            .db
            .raw_range_cf(&header_cf, &[], None)
            .expect("the rejected startup leaves the predecessor row untouched"),
        before
    );
}

#[test]
fn initialization_rejects_finalized_tip_header_mismatch_before_cleanup() {
    let _init_guard = zakura_test::init();
    let network = Network::Mainnet;
    let genesis = mainnet_block(0);
    let block1 = mainnet_block(1);
    let block2 = mainnet_block(2);
    let state = state_with_genesis_config(&network, genesis.clone(), Config::ephemeral());
    write_full_block_header_and_transactions(&state, block1);
    let finalized_header_cf = state
        .db
        .cf_handle("block_header_by_height")
        .expect("the finalized header column exists");
    let legacy_header_cf = state
        .db
        .cf_handle(ZAKURA_HEADER_BY_HEIGHT)
        .expect("the obsolete column remains physically present");
    let mut corrupted = DiskWriteBatch::new();
    corrupted.zs_insert(&finalized_header_cf, Height(1), &genesis.header);
    corrupted.zs_insert(&legacy_header_cf, Height(2), &block2.header);
    state
        .db
        .write(corrupted)
        .expect("the mismatched tip and legacy fixture write");
    let legacy_before = state
        .db
        .raw_range_cf(&legacy_header_cf, &[], None)
        .expect("the predecessor row can be observed without decoding it");
    let config = engine_config(network, &genesis);

    assert!(matches!(
        initialize_header_chain_reconciled(&state, &config, Vec::new()),
        Err(HeaderChainInitializationError::AnchorMismatch)
    ));
    assert!(HeaderChainStore::new(state.header_chain_disk_db())
        .metadata()
        .is_err());
    assert_eq!(
        state
            .db
            .raw_range_cf(&legacy_header_cf, &[], None)
            .expect("the rejected startup leaves the predecessor row untouched"),
        legacy_before
    );
}

#[test]
fn initialization_does_not_fill_finalized_gaps_from_legacy_overlay() {
    let _init_guard = zakura_test::init();
    let network = Network::Mainnet;
    let genesis = mainnet_block(0);
    let block1 = mainnet_block(1);
    let state = state_with_genesis_config(&network, genesis.clone(), Config::ephemeral());
    write_full_block_header_and_transactions(&state, block1);
    let finalized_hash_cf = state
        .db
        .cf_handle("hash_by_height")
        .expect("the finalized hash column exists");
    let header_cf = state
        .db
        .cf_handle(ZAKURA_HEADER_BY_HEIGHT)
        .expect("the obsolete column remains physically present");
    let hash_cf = state
        .db
        .cf_handle(ZAKURA_HEADER_HASH_BY_HEIGHT)
        .expect("the obsolete column remains physically present");
    let height_cf = state
        .db
        .cf_handle(ZAKURA_HEADER_HEIGHT_BY_HASH)
        .expect("the obsolete column remains physically present");
    let mut corrupted = DiskWriteBatch::new();
    corrupted.zs_delete(&finalized_hash_cf, Height(0));
    corrupted.zs_insert(&header_cf, Height(0), &genesis.header);
    corrupted.zs_insert(&hash_cf, Height(0), genesis.hash());
    corrupted.zs_insert(&height_cf, genesis.hash(), Height(0));
    state
        .db
        .write(corrupted)
        .expect("the finalized gap and legacy fallback fixture write");
    let legacy_before = [
        ZAKURA_HEADER_BY_HEIGHT,
        ZAKURA_HEADER_HASH_BY_HEIGHT,
        ZAKURA_HEADER_HEIGHT_BY_HASH,
    ]
    .map(|family| {
        let cf = state
            .db
            .cf_handle(family)
            .expect("the obsolete column remains physically present");
        state
            .db
            .raw_range_cf(&cf, &[], None)
            .expect("the predecessor rows can be observed without decoding them")
    });
    let config = engine_config(network, &genesis);

    assert!(matches!(
        initialize_header_chain_reconciled(&state, &config, Vec::new()),
        Err(HeaderChainInitializationError::AnchorMismatch)
    ));
    assert!(HeaderChainStore::new(state.header_chain_disk_db())
        .metadata()
        .is_err());
    for (family, before) in [
        ZAKURA_HEADER_BY_HEIGHT,
        ZAKURA_HEADER_HASH_BY_HEIGHT,
        ZAKURA_HEADER_HEIGHT_BY_HASH,
    ]
    .into_iter()
    .zip(legacy_before)
    {
        let cf = state
            .db
            .cf_handle(family)
            .expect("the obsolete column remains physically present");
        assert_eq!(
            state
                .db
                .raw_range_cf(&cf, &[], None)
                .expect("the rejected startup leaves predecessor rows untouched"),
            before
        );
    }
}