fn sync_next_to_send_after_gap_stream(next_to_send: &mut u64, end_height: u64) {
let dedup = super::memory::GAP_STREAM_DEDUP_HEIGHT.load(Ordering::Relaxed);
if dedup < *next_to_send {
return;
}
let advanced = dedup.saturating_add(1).min(end_height.saturating_add(1));
if advanced > *next_to_send {
*next_to_send = advanced;
}
}
#[allow(dead_code)] pub(crate) fn rewind_gap_stream_dedup_over_missing_hole(
gap: u64,
last_streamed: u64,
) -> Option<u64> {
if last_streamed < gap {
return None;
}
if !super::IBD_TIP_GAP_MISSING.load(Ordering::Relaxed) {
return None;
}
let rewind = gap.saturating_sub(1);
match super::memory::GAP_STREAM_DEDUP_HEIGHT.compare_exchange(
last_streamed,
rewind,
Ordering::Relaxed,
Ordering::Relaxed,
) {
Ok(_) => Some(rewind),
Err(_) => None,
}
}
fn resync_next_to_send_with_validation_tip(
validation_height: Option<&Arc<AtomicU64>>,
received: &mut BTreeMap<u64, (SharedBlock, SharedWitnesses)>,
next_to_send: &mut u64,
end_height: u64,
network: &Arc<NetworkManager>,
peer_addr: SocketAddr,
block_hash_by_height: &BTreeMap<u64, [u8; 32]>,
in_flight_heights: &HashSet<u64>,
last_gap_at: &mut std::time::Instant,
) -> bool {
let Some(vh) = validation_height else {
return false;
};
let tip = vh.load(Ordering::Relaxed);
let need = tip.saturating_add(1);
if tip >= end_height || need > end_height {
for &h in in_flight_heights.iter() {
if let Some(&hash) = block_hash_by_height.get(&h) {
network.cancel_block_request(peer_addr, hash);
}
}
*next_to_send = end_height.saturating_add(1);
*last_gap_at = std::time::Instant::now();
return true;
}
if need <= *next_to_send {
return false;
}
while *next_to_send < need {
let h = *next_to_send;
if h == need || (h == need.saturating_sub(1) && received.contains_key(&h)) {
if in_flight_heights.contains(&h) {
if let Some(&hash) = block_hash_by_height.get(&h) {
network.cancel_block_request(peer_addr, hash);
}
}
*next_to_send += 1;
continue;
}
let _ = received_take(received, h);
if in_flight_heights.contains(&h) {
if let Some(&hash) = block_hash_by_height.get(&h) {
network.cancel_block_request(peer_addr, hash);
}
}
*next_to_send += 1;
}
*last_gap_at = std::time::Instant::now();
false
}
async fn try_stream_validation_gap(
validation_height: Option<&Arc<AtomicU64>>,
received: &mut BTreeMap<u64, (SharedBlock, SharedWitnesses)>,
block_tx: Option<&tokio::sync::mpsc::Sender<(u64, SharedBlock, SharedWitnesses)>>,
start_height: u64,
end_height: u64,
) -> Result<bool> {
let Some(vh) = validation_height else {
return Ok(false);
};
let gap = vh.load(Ordering::Relaxed).saturating_add(1);
if gap < start_height || gap > end_height {
return Ok(false);
}
let last_streamed = super::memory::GAP_STREAM_DEDUP_HEIGHT.load(Ordering::Relaxed);
let Some(tx) = block_tx else {
return Ok(false);
};
if gap <= last_streamed {
if super::tip_stage::tip_taken_by_validation(gap)
|| super::tip_release::tip_release_holds(gap)
{
return Ok(false);
}
let Some((block, block_witnesses)) = received_clone(received, gap) else {
return Ok(false);
};
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
let last_h = super::memory::GAP_STREAM_LAST_RESEND_HEIGHT.load(Ordering::Relaxed);
let last_ms = super::memory::GAP_STREAM_LAST_RESEND_MS.load(Ordering::Relaxed);
if gap == last_h && now_ms.saturating_sub(last_ms) < 750 {
return Ok(false);
}
let tip_on_channel = stream_tip_to_coordinator(tx, gap, block, block_witnesses).await?;
super::memory::GAP_STREAM_LAST_RESEND_HEIGHT.store(gap, Ordering::Relaxed);
super::memory::GAP_STREAM_LAST_RESEND_MS.store(now_ms, Ordering::Relaxed);
tracing::warn!(
"[IBD_GAP_STREAM_RESEND] height {} (last_streamed={}) — tip still missing after prior drain",
gap,
last_streamed
);
super::memory::bump_gap_stream_dedup(gap);
if tip_on_channel {
let drained_n =
drain_consecutive_received_after(received, block_tx, gap, end_height).await?;
if drained_n > 0 {
tracing::warn!(
"[IBD_GAP_DRAIN] after={} drained={} through={} (chunk {}-{})",
gap,
drained_n,
gap.saturating_add(drained_n),
start_height,
end_height
);
}
}
return Ok(true);
}
let Some((block, block_witnesses)) = received_clone(received, gap) else {
return Ok(false);
};
let tip_on_channel = stream_tip_to_coordinator(tx, gap, block, block_witnesses).await?;
super::memory::bump_gap_stream_dedup(gap);
if tip_on_channel {
let drained_n =
drain_consecutive_received_after(received, block_tx, gap, end_height).await?;
if drained_n > 0 {
tracing::warn!(
"[IBD_GAP_DRAIN] after={} drained={} through={} (chunk {}-{})",
gap,
drained_n,
gap.saturating_add(drained_n),
start_height,
end_height
);
}
}
Ok(true)
}
async fn stream_tip_to_coordinator(
tx: &tokio::sync::mpsc::Sender<(u64, SharedBlock, SharedWitnesses)>,
gap: u64,
block: SharedBlock,
witnesses: SharedWitnesses,
) -> Result<bool> {
if !super::tip_release::release_side_drain_enabled() {
if tx.send((gap, block, witnesses)).await.is_err() {
return Err(anyhow::anyhow!(
"block_tx closed during gap stream - chunk needs retry"
));
}
return Ok(true);
}
match tx.try_send((gap, block, witnesses)) {
Ok(()) => Ok(true),
Err(tokio::sync::mpsc::error::TrySendError::Full((h, b, w))) => {
let prev = super::tip_release::offer_tip_release(h, b, w);
warn!(
"[IBD_TIP_RELEASE_LATCH] h={} channel_full capacity={} prev_latched={:?} — release-side (no send.await)",
h,
tx.capacity(),
prev
);
Ok(false)
}
Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => Err(anyhow::anyhow!(
"block_tx closed during gap stream - chunk needs retry"
)),
}
}
async fn drain_consecutive_received_after(
received: &mut BTreeMap<u64, (SharedBlock, SharedWitnesses)>,
block_tx: Option<&tokio::sync::mpsc::Sender<(u64, SharedBlock, SharedWitnesses)>>,
after_height: u64,
end_height: u64,
) -> Result<u64> {
let Some(tx) = block_tx else {
return Ok(0);
};
let mut next = after_height.saturating_add(1);
let mut drained = 0u64;
while next <= end_height {
let Some((block, block_witnesses)) = received_take(received, next) else {
break;
};
await_block_tx_tip_reserve(tx, next, Some(after_height)).await;
if tx.send((next, block, block_witnesses)).await.is_err() {
return Err(anyhow::anyhow!(
"block_tx closed during consecutive gap drain - chunk needs retry"
));
}
super::memory::bump_gap_stream_dedup(next);
drained = drained.saturating_add(1);
next = next.saturating_add(1);
}
Ok(drained)
}
async fn flush_received_on_abort(
received: &mut BTreeMap<u64, (SharedBlock, SharedWitnesses)>,
block_tx: Option<&tokio::sync::mpsc::Sender<(u64, SharedBlock, SharedWitnesses)>>,
start_height: u64,
end_height: u64,
next_to_send: u64,
validation_height: Option<&AtomicU64>,
) -> usize {
let Some(tx) = block_tx else {
return 0;
};
if received.is_empty() {
return 0;
}
let buffered = received.len();
let mut flushed = 0usize;
if let Some(vh) = validation_height {
let tip_needed = vh.load(Ordering::Relaxed).saturating_add(1);
let min_h = received.keys().next().copied();
let max_h = received.keys().next_back().copied();
let ahead_only = min_h.is_some_and(|m| m > tip_needed);
if ahead_only {
let skipped = received.len();
info!(
"[IBD_FLUSH_ON_ABORT] chunk {}-{}: flushed 0 tip-contiguous block(s) (kept_ahead={}, buffered={}, next_to_send={}, tip_needed={})",
start_height, end_height, skipped, buffered, next_to_send, tip_needed
);
} else if min_h.is_some_and(|m| m == tip_needed) || received.contains_key(&tip_needed) {
let mut h = tip_needed;
while let Some((block, witnesses)) = received_take(received, h) {
if tx.send((h, block, witnesses)).await.is_err() {
break;
}
flushed += 1;
h = h.saturating_add(1);
}
let skipped = received.len();
received.clear();
crate::node::parallel_ibd::memory::GAP_FLUSH_ON_ABORT_BLOCKS
.fetch_add(flushed as u64, Ordering::Relaxed);
info!(
"[IBD_FLUSH_ON_ABORT] chunk {}-{}: flushed {} tip-contiguous block(s) (skipped_ahead={}, buffered={}, next_to_send={}, tip_needed={})",
start_height, end_height, flushed, skipped, buffered, next_to_send, tip_needed
);
} else {
let _ = max_h;
while let Some((&h, _)) = received.iter().next() {
if h > tip_needed {
break;
}
let Some((block, witnesses)) = received_take(received, h) else {
break;
};
if h < tip_needed {
}
if tx.send((h, block, witnesses)).await.is_err() {
break;
}
flushed += 1;
}
let skipped = received.len();
received.clear();
if flushed > 0 || skipped > 0 {
crate::node::parallel_ibd::memory::GAP_FLUSH_ON_ABORT_BLOCKS
.fetch_add(flushed as u64, Ordering::Relaxed);
info!(
"[IBD_FLUSH_ON_ABORT] chunk {}-{}: flushed {} block(s) (skipped_ahead={}, buffered={}, next_to_send={}, tip_needed={})",
start_height, end_height, flushed, skipped, buffered, next_to_send, tip_needed
);
}
}
} else {
while let Some((&h, _)) = received.iter().next() {
let Some((block, witnesses)) = received_take(received, h) else {
break;
};
if tx.send((h, block, witnesses)).await.is_err() {
break;
}
flushed += 1;
}
if flushed > 0 {
crate::node::parallel_ibd::memory::GAP_FLUSH_ON_ABORT_BLOCKS
.fetch_add(flushed as u64, Ordering::Relaxed);
info!(
"[IBD_FLUSH_ON_ABORT] chunk {}-{}: flushed {} buffered block(s) to coordinator (next_to_send={})",
start_height, end_height, flushed, next_to_send
);
}
}
flushed
}
async fn abort_on_outer_deadline(
received: &mut BTreeMap<u64, (SharedBlock, SharedWitnesses)>,
block_tx: Option<&tokio::sync::mpsc::Sender<(u64, SharedBlock, SharedWitnesses)>>,
start_height: u64,
end_height: u64,
next_to_send: u64,
validation_height: Option<&AtomicU64>,
in_flight_heights: &HashSet<u64>,
block_hash_by_height: &BTreeMap<u64, [u8; 32]>,
network: &NetworkManager,
peer_addr: SocketAddr,
peer_scorer: &crate::network::peer_scoring::PeerScorer,
outer_secs: u64,
) -> Result<DownloadChunkResult> {
warn!(
"[IBD] chunk {}-{} cooperative outer deadline ({}s) — flushing {} buffered block(s) before abort",
start_height,
end_height,
outer_secs,
received.len()
);
peer_scorer.record_failure(peer_addr);
flush_received_on_abort(
received,
block_tx,
start_height,
end_height,
next_to_send,
validation_height,
)
.await;
received_drain_all(received);
for &h in in_flight_heights {
if let Some(&h_hash) = block_hash_by_height.get(&h) {
network.cancel_block_request(peer_addr, h_hash);
}
}
Err(anyhow::anyhow!(
"Chunk {}-{}: outer deadline {}s",
start_height,
end_height,
outer_secs
))
}
pub(crate) fn should_extend_outer_while_streaming(
last_gap_at: std::time::Instant,
gap_streams: u64,
chunk_started: std::time::Instant,
extensions_used: u32,
) -> bool {
const MAX_EXTENDS: u32 = 4; if extensions_used >= MAX_EXTENDS {
return false;
}
if gap_streams == 0 {
return false;
}
if last_gap_at.elapsed() < Duration::from_secs(8) {
return true;
}
let secs = chunk_started.elapsed().as_secs_f64().max(1.0);
gap_streams as f64 / secs >= 1.0
}
fn outer_extend_secs() -> u64 {
std::env::var("BLVM_IBD_OUTER_EXTEND_SECS")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(60)
.clamp(30, 120)
}
fn try_soft_extend_outer(
outer_deadline: &mut Option<tokio::time::Instant>,
outer_extends: &mut u32,
outer_deadline_secs: &mut u64,
last_gap_at: std::time::Instant,
gap_streams: u64,
chunk_start_time: std::time::Instant,
start_height: u64,
end_height: u64,
peer_id: &str,
) -> bool {
if !should_extend_outer_while_streaming(
last_gap_at,
gap_streams,
chunk_start_time,
*outer_extends,
) {
return false;
}
let add = outer_extend_secs();
*outer_extends = outer_extends.saturating_add(1);
*outer_deadline_secs = outer_deadline_secs.saturating_add(add);
*outer_deadline = Some(tokio::time::Instant::now() + Duration::from_secs(add));
warn!(
"[IBD_OUTER_EXTEND] chunk {}-{} peer={} extend={}s total_budget={}s streams={} extends={}/4",
start_height, end_height, peer_id, add, *outer_deadline_secs, gap_streams, *outer_extends
);
true
}
async fn wait_cooperative_outer(deadline: Option<tokio::time::Instant>) {
match deadline {
Some(d) => tokio::time::sleep_until(d).await,
None => std::future::pending::<()>().await,
}
}
pub(crate) fn empty_witness_hit_cap() -> u32 {
std::env::var("BLVM_IBD_EMPTY_WITNESS_MAX")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(4)
.clamp(2, 32)
}
pub(crate) fn far_ahead_band() -> u64 {
latch_env!(u64, {
std::env::var("BLVM_IBD_FAR_AHEAD_BAND")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(128)
.clamp(32, 1024)
})
}
pub(crate) fn tip_gap_timeout_secs() -> u64 {
tip_gap_timeout_secs_ex(false)
}
pub(crate) fn tip_gap_timeout_secs_ex(_ahead_buffered: bool) -> u64 {
tip_gap_timeout_secs_for_chunk(_ahead_buffered, 0, 0)
}
pub(crate) fn tip_gap_timeout_secs_for_chunk(
ahead_buffered: bool,
chunk_start: u64,
chunk_end: u64,
) -> u64 {
let base = latch_env!(u64, {
std::env::var("BLVM_IBD_TIP_GAP_TIMEOUT_SECS")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(16)
.clamp(8, 60)
});
let holes = crate::node::parallel_ibd::IBD_TIP_BRIDGE_HOLES.load(Ordering::Relaxed);
let hole_cap = {
let raw = latch_env!(u64, {
std::env::var("BLVM_IBD_TIP_HOLE_GAP_TIMEOUT_SECS")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(5)
});
raw.clamp(3, base)
};
let holey_cap = {
let raw = latch_env!(u64, {
std::env::var("BLVM_IBD_TIP_HOLEY_PENDING_CAP_SECS")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(16)
});
raw.clamp(hole_cap, base)
};
let trigger = {
let raw = latch_env!(u64, {
std::env::var("BLVM_IBD_TIP_HOLE_CAP_TRIGGER_SECS")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(3)
});
raw.clamp(2, hole_cap)
};
let empty_trigger = {
let raw = latch_env!(u64, {
std::env::var("BLVM_IBD_TIP_EMPTY_BRIDGE_CAP_TRIGGER_SECS")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(0)
});
raw.clamp(0, base)
};
let awaiting = super::tip_stage::tip_awaiting_secs_for_cap();
let pending = crate::node::parallel_ibd::memory::BRIDGE_PENDING_COUNT.load(Ordering::Relaxed);
let gap_missing = crate::node::parallel_ibd::IBD_TIP_GAP_MISSING.load(Ordering::Relaxed);
let deep = chunk_end > chunk_start;
let cheese =
ahead_buffered || crate::node::parallel_ibd::IBD_REORDER_AHEAD.load(Ordering::Relaxed) > 0;
if holes > 0 {
if awaiting >= trigger {
if holes == 1 && pending > 0 {
return hole_cap;
}
if pending > 0 && !cheese {
return tip_cap_during_export(holey_cap, base);
}
return tip_cap_during_export(hole_cap, base);
}
return base;
}
if gap_missing && pending == 0 && awaiting >= empty_trigger {
return if deep && !cheese {
tip_cap_during_export(holey_cap, base)
} else {
tip_cap_during_export(hole_cap, base)
};
}
base
}
#[inline]
fn tip_cap_during_export(cap: u64, base: u64) -> u64 {
if !crate::node::parallel_ibd::IBD_CHECKPOINT_EXPORT_ACTIVE.load(Ordering::Relaxed) {
return cap;
}
let floor = {
let raw = latch_env!(u64, {
std::env::var("BLVM_IBD_TIP_EXPORT_ACTIVE_CAP_SECS")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(16)
});
raw.clamp(8, base)
};
cap.max(floor).min(base)
}
pub(crate) fn far_ahead_timeout_secs() -> u64 {
latch_env!(u64, {
std::env::var("BLVM_IBD_FAR_AHEAD_TIMEOUT_SECS")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(10)
.clamp(5, 30)
})
}
pub(crate) fn tip_gap_soft_retries() -> u32 {
latch_env!(u32, {
std::env::var("BLVM_IBD_TIP_GAP_SOFT_RETRIES")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(3)
.clamp(1, 12)
})
}
pub(crate) fn download_byte_budget() -> Option<u64> {
let raw = std::env::var("BLVM_IBD_DOWNLOAD_BYTE_BUDGET").ok()?;
let n: u64 = raw.parse().ok()?;
if n == 0 {
return None;
}
Some(n.clamp(8 * 1024 * 1024, 256 * 1024 * 1024))
}
fn download_est_block_bytes() -> u64 {
static EST: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1_000_000);
EST.load(std::sync::atomic::Ordering::Relaxed).max(50_000)
}
fn note_download_block_bytes(nbytes: u64) {
if nbytes == 0 {
return;
}
static EST: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1_000_000);
let old = EST.load(std::sync::atomic::Ordering::Relaxed).max(50_000);
let next = old.saturating_mul(7).saturating_add(nbytes) / 8;
EST.store(next.max(50_000), std::sync::atomic::Ordering::Relaxed);
}
#[allow(clippy::if_same_then_else)] pub(crate) fn gap_soft_retry_budget(height: u64, validation_tip: u64) -> u32 {
let tip_needed = validation_tip.saturating_add(1);
if height < tip_needed {
0
} else if height > tip_needed.saturating_add(far_ahead_band()) {
0
} else if height == tip_needed {
tip_gap_soft_retries()
} else {
3
}
}
pub(crate) fn gap_soft_retry_budget_for_chunk(
height: u64,
validation_tip: u64,
chunk_start: u64,
chunk_end: u64,
) -> u32 {
gap_soft_retry_budget_for_chunk_ex(height, validation_tip, chunk_start, chunk_end, false, false)
}
pub(crate) fn gap_soft_retry_budget_for_chunk_ex(
height: u64,
validation_tip: u64,
chunk_start: u64,
chunk_end: u64,
ahead_buffered: bool,
hot_tip_streamer: bool,
) -> u32 {
let tip_needed = validation_tip.saturating_add(1);
if height == tip_needed {
let pending =
crate::node::parallel_ibd::memory::BRIDGE_PENDING_COUNT.load(Ordering::Relaxed);
let gap_missing = crate::node::parallel_ibd::IBD_TIP_GAP_MISSING.load(Ordering::Relaxed);
if gap_missing && pending == 0 {
if chunk_end > chunk_start {
let _ = (ahead_buffered, hot_tip_streamer);
return 2;
}
let _ = (ahead_buffered, hot_tip_streamer);
return 2;
}
if chunk_end > chunk_start {
let _ = (ahead_buffered, hot_tip_streamer);
let holes = crate::node::parallel_ibd::IBD_TIP_BRIDGE_HOLES.load(Ordering::Relaxed);
if holes >= 32 || (holes >= 20 && pending == 0) {
return 1;
}
return 2;
}
let _ = (hot_tip_streamer, ahead_buffered);
let holes = crate::node::parallel_ibd::IBD_TIP_BRIDGE_HOLES.load(Ordering::Relaxed);
if holes >= 32 || (holes >= 20 && pending == 0) {
return 1;
}
return 2;
}
let _ = (chunk_start, chunk_end, ahead_buffered, hot_tip_streamer);
gap_soft_retry_budget(height, validation_tip)
}
pub(crate) fn tip_pipe_has_ahead_buffered(
received: &std::collections::BTreeMap<u64, (SharedBlock, SharedWitnesses)>,
tip_needed: u64,
) -> bool {
received.keys().any(|&h| h > tip_needed)
}
pub(crate) fn tip_covering_fail_is_mute(err_str: &str) -> bool {
if err_str.contains("tip-SLA") || err_str.contains("tip-enter walk-in") {
return false;
}
err_str.contains("tip-gap timeout")
|| err_str.contains("Block timeout for gap height")
|| err_str.contains("no first block in")
|| err_str.contains("PIPE_FILL mute")
}
pub(crate) fn mute_pipe_ms() -> u64 {
latch_env!(u64, {
std::env::var("BLVM_IBD_MUTE_PIPE_MS")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(3000)
.clamp(1000, 15_000)
})
}
pub(crate) fn pipe_mute_episode_active(
gap_streams: u64,
pipe_fill_recv0: bool,
saw_network_body: bool,
next_in_chunk: bool,
) -> bool {
(gap_streams > 0 || pipe_fill_recv0) && !saw_network_body && next_in_chunk
}
pub(crate) fn pipe_mute_should_clear_clock(
episode_active: bool,
gap_streams: u64,
pipe_fill_recv0: bool,
saw_network_body: bool,
) -> bool {
!episode_active && (saw_network_body || !(gap_streams > 0 || pipe_fill_recv0))
}
pub(crate) fn pipe_mute_may_fire(tip_buffered: bool, tip_is_local_inflight: bool) -> bool {
!(tip_buffered || tip_is_local_inflight)
}
pub(crate) fn tip_hole_grow_enabled() -> bool {
super::policy::tip_hole_grow()
}
pub(crate) fn tip_hole_pipe_cap() -> usize {
super::policy::tip_hole_pipe()
}
pub(crate) fn tip_hole_grow_cap() -> usize {
if !tip_hole_grow_enabled() {
return tip_hole_pipe_cap();
}
super::policy::tip_hole_grow_cap_raw()
.clamp(2, 128)
.min(tip_hole_pipe_cap())
}
pub(crate) fn tip_hole_gd_fast_enabled() -> bool {
super::policy::tip_hole_gd_fast()
}
pub(crate) fn tip_hole_gd_fast_ms() -> u64 {
super::policy::tip_hole_gd_fast_ms()
}
pub(crate) fn tip_hole_gd_fast_n() -> u64 {
super::policy::tip_hole_gd_fast_n()
}
pub(crate) fn tip_hole_grow_fast_cap() -> usize {
let cold = tip_hole_grow_cap();
super::policy::tip_hole_grow_fast_cap_raw()
.clamp(cold, 96)
.min(tip_hole_pipe_cap())
.max(cold)
}
pub(crate) fn tip_hole_gd_fast_min_h() -> u64 {
super::policy::tip_hole_gd_fast_min_h()
}
pub(crate) fn tip_hole_grow_cap_effective() -> usize {
let cold = tip_hole_grow_cap();
if !tip_hole_grow_enabled() || !tip_hole_gd_fast_enabled() {
return cold;
}
let min_h = tip_hole_gd_fast_min_h();
if min_h > 0 && super::tip_stage::tracked_tip_height() < min_h {
return cold;
}
let fast = tip_hole_grow_fast_cap();
if fast <= cold {
return cold;
}
match super::tip_stage::getdata_body_ewma_ms_min_n(tip_hole_gd_fast_n()) {
Some((ms, _n)) if ms < tip_hole_gd_fast_ms() => fast,
_ => cold,
}
}
pub(crate) fn tip_hole_gd_slow_enabled() -> bool {
super::policy::tip_hole_gd_slow()
}
pub(crate) fn tip_hole_gd_slow_ms() -> u64 {
super::policy::tip_hole_gd_slow_ms()
}
pub(crate) fn tip_hole_gd_slow_n() -> u64 {
super::policy::tip_hole_gd_slow_n()
}
pub(crate) fn tip_hole_slow_fill_cap() -> usize {
super::policy::tip_hole_slow_fill_cap_raw()
.unwrap_or_else(tip_hole_grow_start)
.clamp(2, tip_hole_grow_cap())
}
pub(crate) fn tip_hole_gd_slow_ratchet_enabled() -> bool {
super::policy::tip_hole_gd_slow_ratchet()
}
pub(crate) fn tip_hole_gd_slow_next_depth(grown: usize) -> usize {
let slow = tip_hole_slow_fill_cap();
if grown <= slow {
return grown;
}
if !tip_hole_gd_slow_ratchet_enabled() {
return slow;
}
grown.saturating_sub(tip_hole_grow_step()).max(slow)
}
pub(crate) fn tip_hole_gd_slow_sole_keep(ibd_ready: usize) -> bool {
ibd_ready <= 1 && tip_hole_gd_slow()
}
pub(crate) fn tip_hole_sole_gd_slow_floor() -> usize {
super::policy::tip_hole_sole_floor().clamp(tip_hole_slow_fill_cap(), tip_hole_grow_cap())
}
pub(crate) fn tip_hole_sole_floor_recover_ms() -> u64 {
super::policy::tip_hole_sole_floor_recover_ms_raw()
.unwrap_or_else(tip_hole_gd_slow_ms)
.clamp(50, tip_hole_gd_slow_ms())
}
pub(crate) fn tip_hole_sole_floor_blocks_grow() -> bool {
if !super::tip_stage::sole_floor_latched() {
return false;
}
match super::tip_stage::getdata_body_ewma_ms_min_n(tip_hole_gd_slow_n()) {
Some((ms, _)) if ms < tip_hole_sole_floor_recover_ms() => {
super::tip_stage::clear_sole_floor_latch();
false
}
_ => true,
}
}
pub(crate) fn tip_hole_sole_no_fast_clear_n() -> u32 {
super::policy::tip_hole_sole_no_fast_clear_n()
}
pub(crate) fn tip_hole_sole_no_fast_min_hold_ms() -> u64 {
super::policy::tip_hole_sole_no_fast_min_hold_ms()
}
pub(crate) fn tip_hole_sole_no_fast_arm_min_h() -> u64 {
super::policy::tip_hole_sole_no_fast_arm_min_h()
}
fn maybe_note_sole_no_fast_latch(height: u64) {
let min_h = tip_hole_sole_no_fast_arm_min_h();
if min_h > 0 && height < min_h {
return;
}
super::tip_stage::note_sole_no_fast_latch();
}
pub(crate) fn tip_hole_sole_floor_max_h() -> u64 {
super::policy::tip_hole_sole_floor_max_h()
}
pub(crate) fn tip_hole_sole_floor_applies(height: u64) -> bool {
let max_h = tip_hole_sole_floor_max_h();
if max_h == 0 {
return true;
}
if height >= max_h {
if super::tip_stage::sole_floor_latched() {
super::tip_stage::clear_sole_floor_latch();
}
return false;
}
true
}
pub(crate) fn tip_hole_sole_no_fast_active() -> bool {
if !super::tip_stage::sole_no_fast_latched() {
return false;
}
let held_long_enough = super::tip_stage::sole_no_fast_armed_age_ms()
.map(|age| age >= tip_hole_sole_no_fast_min_hold_ms())
.unwrap_or(false);
match super::tip_stage::getdata_body_ewma_ms_min_n(tip_hole_gd_fast_n()) {
Some((ms, _)) if ms < tip_hole_gd_fast_ms() => {
if !held_long_enough {
let _ = super::tip_stage::sole_no_fast_note_clear_sample(false);
return true;
}
let streak = super::tip_stage::sole_no_fast_note_clear_sample(true);
if streak >= tip_hole_sole_no_fast_clear_n() {
super::tip_stage::clear_sole_no_fast_latch();
false
} else {
true
}
}
_ => {
let _ = super::tip_stage::sole_no_fast_note_clear_sample(false);
true
}
}
}
pub(crate) fn tip_hole_cap_for_sole(sole_ready: bool, cap: usize) -> usize {
if sole_ready && tip_hole_sole_no_fast_active() {
cap.min(tip_hole_grow_cap())
} else {
cap
}
}
pub(crate) fn tip_hole_gd_slow() -> bool {
if !tip_hole_grow_enabled() || !tip_hole_gd_slow_enabled() {
return false;
}
matches!(
super::tip_stage::getdata_body_ewma_ms_min_n(tip_hole_gd_slow_n()),
Some((ms, _n)) if ms >= tip_hole_gd_slow_ms()
)
}
pub(crate) fn tip_hole_warm_enabled() -> bool {
super::policy::tip_hole_warm()
}
fn tip_hole_warm_cap_raw() -> usize {
super::policy::tip_hole_warm_cap_raw()
.unwrap_or_else(tip_hole_pipe_cap)
.clamp(2, 128)
.min(tip_hole_pipe_cap())
}
pub(crate) fn tip_hole_grow_cap_for_peer(hot_tip_streamer: bool) -> usize {
let cold = tip_hole_grow_cap_effective();
if !tip_hole_grow_enabled() || !tip_hole_warm_enabled() || !hot_tip_streamer {
return cold;
}
tip_hole_warm_cap_raw()
.clamp(cold, 128)
.min(tip_hole_pipe_cap())
.max(cold)
}
pub(crate) fn tip_hole_grow_start() -> usize {
if !tip_hole_grow_enabled() {
return tip_hole_pipe_cap();
}
super::policy::tip_hole_grow_start()
.clamp(2, 32)
.min(tip_hole_grow_cap())
}
pub(crate) fn tip_hole_grow_step() -> usize {
let configured = super::policy::tip_hole_grow_step();
if tip_hole_gd_fast_enabled()
&& tip_hole_grow_enabled()
&& tip_hole_grow_cap_effective() > tip_hole_grow_cap()
{
return configured.max(16);
}
configured
}
pub(crate) fn tip_hole_sticky_abs_cap(hot_tip_streamer: bool) -> usize {
let cold = tip_hole_grow_cap();
let mut cap = if tip_hole_gd_fast_enabled() {
tip_hole_grow_fast_cap().max(cold)
} else {
cold
};
if tip_hole_warm_enabled() && hot_tip_streamer {
let warm = tip_hole_warm_cap_raw()
.clamp(cap, 128)
.min(tip_hole_pipe_cap());
cap = cap.max(warm);
}
cap.min(tip_hole_pipe_cap())
}
pub(crate) fn tip_hole_grow_on_delivery(current: usize) -> usize {
tip_hole_grow_on_delivery_capped(current, tip_hole_grow_cap_effective())
}
pub(crate) fn tip_hole_grow_on_delivery_capped(current: usize, cap: usize) -> usize {
if !tip_hole_grow_enabled() {
return tip_hole_pipe_cap();
}
current.saturating_add(tip_hole_grow_step()).min(cap)
}
pub(crate) fn gap_timeout_for_chunk(
start_height: u64,
end_height: u64,
validation_tip: u64,
default_secs: u64,
) -> u64 {
let tip_needed = validation_tip.saturating_add(1);
if start_height <= tip_needed && tip_needed <= end_height {
tip_gap_timeout_secs_for_chunk(false, start_height, end_height)
} else if start_height > tip_needed.saturating_add(far_ahead_band()) {
far_ahead_timeout_secs()
} else {
default_secs
}
}
pub(crate) fn wan_deep_pipe_timeout_secs(
height: u64,
validation_tip: u64,
confirmed_body_height: u64,
) -> Option<u64> {
if confirmed_body_height == 0 || height <= confirmed_body_height {
return None;
}
let tip_needed = validation_tip.saturating_add(1);
if height < tip_needed {
return None;
}
let offset = height.saturating_sub(tip_needed);
let secs = match offset {
0 => tip_gap_timeout_secs(),
1..=31 => latch_env!(u64, {
std::env::var("BLVM_IBD_PIPE_MID_TIMEOUT_SECS")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(30)
.clamp(15, 60)
}),
_ => latch_env!(u64, {
std::env::var("BLVM_IBD_PIPE_DEEP_TIMEOUT_SECS")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(45)
.clamp(25, 90)
}),
};
Some(secs)
}
pub(crate) fn block_gap_timeout_secs(
height: u64,
validation_tip: u64,
confirmed_body_height: u64,
chunk_start: u64,
chunk_end: u64,
default_secs: u64,
) -> u64 {
wan_deep_pipe_timeout_secs(height, validation_tip, confirmed_body_height).unwrap_or_else(|| {
gap_timeout_for_chunk(chunk_start, chunk_end, validation_tip, default_secs)
})
}
pub(crate) fn tip_gap_inflight_exceeded(started: Instant, tip_gap_secs: u64) -> bool {
started.elapsed().as_secs() >= tip_gap_secs.max(1)
}
pub(crate) fn rebase_tip_cap_clock(
tip_needed: u64,
tip_cap_clock_h: &mut Option<u64>,
in_flight_heights: &HashSet<u64>,
inflight_started: &mut HashMap<u64, Instant>,
) -> bool {
if Some(tip_needed) == *tip_cap_clock_h {
return false;
}
*tip_cap_clock_h = Some(tip_needed);
if !in_flight_heights.contains(&tip_needed) {
return false;
}
if let Some(started) = inflight_started.get_mut(&tip_needed) {
*started = Instant::now();
return true;
}
false
}
fn wall_ms_now() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0)
}
fn sync_inflight_started(
inflight_started: &mut HashMap<u64, Instant>,
inflight_deadlines: &mut HashMap<u64, Arc<AtomicU64>>,
in_flight_heights: &HashSet<u64>,
) {
inflight_started.retain(|h, _| in_flight_heights.contains(h));
inflight_deadlines.retain(|h, _| in_flight_heights.contains(h));
for &h in in_flight_heights {
inflight_started.entry(h).or_insert_with(Instant::now);
}
}
async fn await_block_with_deadline(
mut rx: tokio::sync::oneshot::Receiver<(Block, Vec<Vec<Witness>>, Option<Vec<u8>>)>,
deadline_ms: Arc<AtomicU64>,
) -> Result<
Result<(Block, Vec<Vec<Witness>>, Option<Vec<u8>>), tokio::sync::oneshot::error::RecvError>,
tokio::time::error::Elapsed,
> {
loop {
let now = wall_ms_now();
let dl = deadline_ms.load(Ordering::Relaxed);
if now >= dl {
return Err(timeout(Duration::ZERO, std::future::pending::<()>())
.await
.unwrap_err());
}
let slice = Duration::from_millis((dl - now).min(500));
tokio::select! {
biased;
r = &mut rx => return Ok(r),
_ = tokio::time::sleep(slice) => {}
}
}
}
fn push_network_inflight(
in_flight: &mut FuturesUnordered<PendingBlockFuture>,
in_flight_heights: &mut HashSet<u64>,
inflight_deadlines: &mut HashMap<u64, Arc<AtomicU64>>,
height: u64,
block_hash: [u8; 32],
rx: tokio::sync::oneshot::Receiver<(Block, Vec<Vec<Witness>>, Option<Vec<u8>>)>,
permit: Option<tokio::sync::OwnedSemaphorePermit>,
timeout_secs: u64,
) {
let deadline = Arc::new(AtomicU64::new(
wall_ms_now().saturating_add(timeout_secs.saturating_mul(1000)),
));
inflight_deadlines.insert(height, Arc::clone(&deadline));
in_flight_heights.insert(height);
let request_start = Instant::now();
in_flight.push(Box::pin(async move {
let r = await_block_with_deadline(rx, deadline).await;
(height, block_hash, request_start, r, permit)
}));
}
async fn wait_tip_enter_abort(
tip_enter: &Option<Arc<super::chunk_assigner::ChunkAssigner>>,
peer_id: &str,
start_height: u64,
end_height: u64,
) {
let Some(assigner) = tip_enter.as_ref() else {
std::future::pending::<()>().await;
return;
};
let mut tick = tokio::time::interval(Duration::from_millis(100));
tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop {
tick.tick().await;
if assigner.should_abort_tip_walk_in(peer_id, start_height, end_height) {
return;
}
}
}
async fn wait_blacklist_abort(
tip_enter: &Option<Arc<super::chunk_assigner::ChunkAssigner>>,
peer_id: &str,
) {
let Some(assigner) = tip_enter.as_ref() else {
std::future::pending::<()>().await;
return;
};
let mut tick = tokio::time::interval(Duration::from_millis(100));
tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop {
tick.tick().await;
if assigner.is_peer_blacklisted(peer_id) {
return;
}
}
}
pub(crate) fn wan_deep_tip_pipe_chunk_deadline_secs(
start_height: u64,
end_height: u64,
confirmed_body_height: u64,
default_secs: u64,
) -> u64 {
if confirmed_body_height > 0
&& start_height > confirmed_body_height
&& end_height.saturating_sub(start_height) >= 63
{
super::tip_stage::tip_sla_secs()
.saturating_mul(2)
.clamp(90, 180)
} else {
default_secs
}
}
pub(crate) fn wan_tip_stream_credit_count(
from_local: bool,
tip_adjacent: bool,
gap_streamed: bool,
) -> u8 {
if from_local {
return 0;
}
let mut n = 0u8;
if tip_adjacent {
n = n.saturating_add(1);
}
if gap_streamed && !tip_adjacent {
n = n.saturating_add(1);
}
n
}