mod attachment_and_vct_aux;
mod deferred;
mod failure_exit;
mod full_state_coherence;
mod selection_and_evidence;
use super::*;
use std::{
collections::VecDeque,
sync::{
atomic::{AtomicUsize, Ordering},
Arc, Mutex,
},
time::{Duration, Instant},
};
use tokio::sync::{mpsc, oneshot, watch};
use zakura_chain::{
block::{self, genesis::regtest_genesis_block, Block, ChainHistoryBlockTxAuthCommitmentHash},
fmt::HexDebug,
history_tree::HistoryTree,
parameters::{
testnet::{ConfiguredActivationHeights, ConfiguredCheckpoints, ParametersBuilder},
Network, NetworkUpgrade, GENESIS_PREVIOUS_BLOCK_HASH,
},
serialization::ZcashDeserializeInto,
transaction::{arbitrary::transaction_to_fake_v5, Transaction},
transparent,
work::{difficulty::ParameterDifficulty as _, equihash},
};
use crate::{
arbitrary::Prepare,
service::{
finalized_state::{
header_chain::{HeaderChainStore, HeaderChainStoreError},
FinalizedState, VctAuxiliaryFailureAttribution, VctAuxiliaryWindow,
VctSuccessorWitness,
},
non_finalized_state::NonFinalizedState,
write::{
classify_verified_change, commit_contextual_finalization, commit_operator_change,
receive_until_deferred_deadline, recover_resource_stall, verified_request,
BlockWriteSender, BlockWriteTaskExit, HeaderChainAttachmentError,
HeaderChainMaintenance, HeaderChainObservers, HeaderChainWriter,
NonFinalizedWriteMessage, PreparedFullStateTransition,
},
ChainTipSender,
},
tests::FakeChainHelper,
CheckpointVerifiedBlock, CommitBlockError, Config, ValidateContextError,
};
use zakura_header_chain::{
AdjustedDifficulty, AlarmSet, ApplyResult, BodyRuleId, BodyUnavailableSummary,
BodyValidationState, BodyViolation, ChainScore, CheckpointSet, ConsensusBodyInvalid,
EngineConfig, EngineMetadata, EngineMode, EngineSnapshot, EvidenceId, FinalityEpoch, Frontier,
FrontierSet, HeaderBatchInput, HeaderChainDiskVersion, HeaderGeneration, HeaderNode,
HeaderRules, HeaderValidationState, InsertHeaders, InvalidTransitionEvidence, SourceId,
StateVersion, SuffixWork, SystemClock, TargetCompletion, TransientBodyFailure,
TransientBodyFailureKind, TransitionContext, TransitionEvent, TransitionFailure,
TransitionRequest, TrustedAnchor, VerifiedChangeCause, VerifiedGeneration, VerifiedHeaderRef,
WorkCoordinate, MAX_CANDIDATE_TIPS_V1, POW_ADJUSTMENT_BLOCK_SPAN,
};
#[test]
fn write_failure_kind_preserves_invalid_and_retryable_context() {
let retryable = CommitBlockError::ValidateContextError(Box::new(
ValidateContextError::VctSuppliedRootUnavailable {
height: block::Height(1),
},
));
assert_eq!(
NonFinalizedWriteFailureKind::from_error(&retryable),
NonFinalizedWriteFailureKind::Retryable
);
let invalid = CommitBlockError::ValidateContextError(Box::new(
ValidateContextError::DuplicateTransparentSpend {
outpoint: transparent::OutPoint {
hash: [1; 32].into(),
index: 0,
},
location: "test chain",
},
));
assert_eq!(
NonFinalizedWriteFailureKind::from_error(&invalid),
NonFinalizedWriteFailureKind::Invalid
);
let invalid_ancestor = CommitBlockError::ValidateContextError(Box::new(
ValidateContextError::InvalidAncestorBlock(block::Hash([2; 32])),
));
assert_eq!(
NonFinalizedWriteFailureKind::from_error(&invalid_ancestor),
NonFinalizedWriteFailureKind::Invalid
);
}
fn header_owner(
snapshot: &EngineSnapshot,
target: block::Hash,
session_id: u64,
request_id: u64,
) -> zakura_header_chain::HeaderSyncWorkOwner {
zakura_header_chain::HeaderWorkAuthority::for_target(snapshot, target)
.bind(
session_id,
std::num::NonZeroU64::new(request_id).expect("fixture request IDs are nonzero"),
)
.into()
}
struct TestDeferredMaintenance {
deadlines: Mutex<VecDeque<chrono::DateTime<chrono::Utc>>>,
sender: Mutex<Option<mpsc::UnboundedSender<NonFinalizedWriteMessage>>>,
reevaluations: AtomicUsize,
}
impl HeaderChainMaintenance for TestDeferredMaintenance {
fn earliest_deferred(
&self,
) -> Result<Option<chrono::DateTime<chrono::Utc>>, HeaderChainStoreError> {
Ok(self
.deadlines
.lock()
.map_err(|_| HeaderChainStoreError::WriterPoisoned)?
.front()
.copied())
}
fn now(&self) -> chrono::DateTime<chrono::Utc> {
chrono::Utc::now()
}
fn reevaluate_deferred(&self) -> Result<(), HeaderChainStoreError> {
self.reevaluations.fetch_add(1, Ordering::SeqCst);
let mut deadlines = self
.deadlines
.lock()
.map_err(|_| HeaderChainStoreError::WriterPoisoned)?;
deadlines.pop_front();
if deadlines.is_empty() {
self.sender
.lock()
.map_err(|_| HeaderChainStoreError::WriterPoisoned)?
.take();
}
Ok(())
}
}
fn header_writer(
finalized_state: &FinalizedState,
network: &Network,
anchor_height: block::Height,
anchor_block: &Arc<zakura_chain::block::Block>,
) -> HeaderChainWriter {
let frontier = Frontier::new(anchor_height, anchor_block.hash());
let config = EngineConfig::new(
EngineMode::Integrated,
network.clone(),
TrustedAnchor {
frontier,
header: anchor_block.header.clone(),
},
CheckpointSet::default(),
)
.expect("the full-state fixture anchor is coherent");
let work = anchor_block
.header
.difficulty_threshold
.to_work()
.expect("the fixture target has exact work");
let anchor = HeaderNode::from_durable_parts(
anchor_block.header.clone(),
frontier.hash,
anchor_block.header.previous_block_hash,
frontier.height,
work,
WorkCoordinate::new(frontier.hash, work.as_u256()),
HeaderValidationState::Valid,
Default::default(),
BodyValidationState::Verified {
evidence: EvidenceId::from_digest([0x70; 32]),
},
Vec::new(),
)
.expect("the anchor node fields agree");
let metadata = EngineMetadata {
disk_format: HeaderChainDiskVersion::CURRENT,
mode: EngineMode::Integrated,
network_id: config.network().kind(),
network_policy_digest: config.network_policy_digest(),
anchor_manifest_digest: config.trust_anchor_digest(),
work_origin: frontier,
state_version: StateVersion::new(1),
header_generation: HeaderGeneration::new(1),
verified_generation: VerifiedGeneration::new(1),
finality_epoch: FinalityEpoch::new(0),
headers_only_migration_epoch: None,
frontiers: FrontierSet {
finalized: frontier,
header_best: frontier,
verified_best: frontier,
},
header_best_score: ChainScore::new(SuffixWork::zero(), frontier.hash),
oldest_retained_height: frontier.height,
alarms: AlarmSet::default(),
last_transition: None,
};
let store = HeaderChainStore::new(finalized_state.db.db().clone());
store
.initialize(metadata, anchor)
.expect("the fixture header store initializes");
let (runtime, _) = store
.startup(&config)
.expect("the fixture header store audits");
HeaderChainWriter::new(runtime, config)
}
#[test]
fn only_the_latest_block_write_attempt_can_reject_waiters() {
let hash = block::Hash([42; 32]);
for reject_before_retry in [false, true] {
let (sender, receiver) = watch::channel(BlockWriteNotice::default());
let old_attempt = start_block_write(&sender, hash);
if reject_before_retry {
notify_block_rejected(&sender, hash, old_attempt);
assert!(receiver.borrow().is_rejected(&hash));
}
let retry = start_block_write(&sender, hash);
assert!(!receiver.borrow().is_rejected(&hash));
if !reject_before_retry {
notify_block_rejected(&sender, hash, old_attempt);
}
assert!(
!receiver.borrow().is_rejected(&hash),
"an older write failing after retry admission must not poison that retry"
);
notify_block_rejected(&sender, hash, retry);
assert!(receiver.borrow().is_rejected(&hash));
notify_block_committed(&sender, hash);
assert!(!receiver.borrow().is_rejected(&hash));
}
}
#[test]
fn expiring_a_block_write_attempt_preserves_its_newer_retry() {
let (sender, receiver) = watch::channel(BlockWriteNotice::default());
let hash = block::Hash([42; 32]);
let old_attempt = start_block_write(&sender, hash);
let retry = start_block_write(&sender, hash);
for i in 1..REJECTED_ANCESTOR_MAP_LIMIT {
let mut other = [0; 32];
other[..8].copy_from_slice(&u64::try_from(i).unwrap().to_le_bytes());
start_block_write(&sender, block::Hash(other));
}
assert_eq!(
receiver.borrow().recent_attempts.len(),
REJECTED_ANCESTOR_MAP_LIMIT
);
assert!(receiver.borrow().attempts.len() <= REJECTED_ANCESTOR_MAP_LIMIT);
notify_block_rejected(&sender, hash, old_attempt);
assert!(!receiver.borrow().is_rejected(&hash));
notify_block_rejected(&sender, hash, retry);
assert!(receiver.borrow().is_rejected(&hash));
start_block_write(&sender, block::Hash([43; 32]));
notify_block_rejected(&sender, hash, retry);
assert!(
!receiver.borrow().is_rejected(&hash),
"a late result cannot reinsert an evicted attempt"
);
}