use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, OnceLock};
use std::time::{Duration, Instant};
use alloy::primitives::{Address, B256, U256};
use alloy::signers::local::PrivateKeySigner;
use bao_tree::ChunkRanges;
use decdn_bao_range::RangedStore;
use decdn_bao_range::{AlignedRange, CHUNK_GROUP_BYTES, align_range};
use decdn_cache::{CacheEngine, CacheError, DownstreamWatch, FillError, FillSession, Hash};
use decdn_client::sink::PullReader;
use decdn_client::source::BlobSource as _;
use decdn_client::{
CoveredRun, DownstreamFrontier, Fault, HashMismatch as ClientPullHashMismatch, LegNoProgress,
PacingWait, PeerSource, PoolLedger, PrimedSource, RampPacer, RecoveryGate, SharedPool,
UpstreamPullHeader, WaitReason, classify, drive, first_leg,
};
use decdn_reputation::Outcome;
use iroh::{EndpointAddr, PublicKey};
use tokio_util::sync::CancellationToken;
use tracing::{Instrument as _, debug, warn};
use super::admit_store::NodeAdmitStore;
use super::backend_source::BackendSource;
use super::funder::{SETTLE_POLL_STEP, settle_window};
use super::ranged_pull::{AssembleOutcome, RecoveryEnd, RunOutcome, RunSink, assemble};
use super::timed_source::{TimedReader, TimedSource, timed_open};
use super::{
EconGate, NodeOrigin, NodeOriginDeps, ProbeGather, PullMiss, PullOutcome, PullVerdict,
SettleOnDrop, bind_upstream_ctx, cached_candidates, classify_pull_failure, coverage_spans,
discover, economic_ceiling, heat_of, lane_ledger, mb_of, now_micros, probe_and_rank,
record_backpressure_exhausted, record_backpressure_refusal, record_outcome,
record_pool_open_failure,
};
use crate::dht::negative_cache::Hash as DhtHash;
use crate::dht::routing::NodeId as DhtNodeId;
use crate::handlers::client::PROOF_WAIT_CEILING;
use crate::runtime::QUIC_MAX_IDLE_TIMEOUT;
use crate::selection::{Candidate, MAX_PROVIDER_ATTEMPTS, POOL_OPEN_CALLER_BUDGET};
use decdn_client::{
PoolContext, PullDeadlines, open_progressive_pull as open_progressive_upstream,
};
const CHUNK_BYTES: u64 = 1024;
pub(crate) struct PullLegTarget {
pub(crate) total_bytes: u64,
namespace_id: [u8; 32],
candidates: Vec<Candidate>,
primed: Option<PrimedHandshake>,
}
impl PullLegTarget {
pub(crate) const fn candidate_count(&self) -> usize {
self.candidates.len()
}
pub(crate) fn uncovered(&self, gap: &ChunkRanges) -> ChunkRanges {
let coverages: Vec<decdn_protocol::Coverage> =
self.candidates.iter().map(|c| c.coverage.clone()).collect();
super::ranged_pull::uncovered(gap, self.total_bytes, &coverages)
}
pub(crate) fn keep_prime_for(&mut self, owns: bool, pull_range: Option<(u64, u64)>) {
let keep = self.primed.as_ref().is_some_and(|primed| {
owns && pull_range == Some((primed.prime.offset, primed.prime.len))
});
if !keep && self.primed.take().is_some() {
debug!("node-origin: the serve does not own the primed pull's range; closing it");
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct PrimeLeg {
pub(crate) offset: u64,
pub(crate) len: u64,
pub(crate) window: u64,
}
impl PrimeLeg {
pub(crate) fn new(
offset: u64,
len: u64,
divisor: u64,
floor: u64,
credit_max: u64,
paid_carried: u64,
) -> Self {
let window =
decdn_incentive::ramped_credit_window(divisor, floor, credit_max, paid_carried);
Self {
offset,
len,
window: window - window % CHUNK_GROUP_BYTES,
}
}
fn predicted_leg(
&self,
total_bytes: u64,
coverage: &decdn_protocol::Coverage,
max_blob_size_bytes: u64,
) -> Option<AlignedRange> {
let block_bytes = decdn_protocol::discovery_block_bytes();
let start = self.offset - self.offset % CHUNK_GROUP_BYTES;
let request_end = if self.len == 0 {
total_bytes
} else {
self.offset
.saturating_add(self.len)
.div_ceil(CHUNK_GROUP_BYTES)
.saturating_mul(CHUNK_GROUP_BYTES)
.min(total_bytes)
};
let first_block = u32::try_from(start / block_bytes).ok()?;
if start >= request_end || !coverage.covers(first_block) {
return None;
}
let blocks = decdn_protocol::num_blocks(total_bytes);
let mut last_block = first_block;
while last_block.saturating_add(1) < blocks && coverage.covers(last_block + 1) {
last_block += 1;
}
let span_end = (u64::from(last_block) + 1)
.saturating_mul(block_bytes)
.min(request_end);
let mut draw = (span_end - start).min(self.window);
if max_blob_size_bytes > 0 {
let cap_end = max_blob_size_bytes.saturating_add(CHUNK_GROUP_BYTES);
draw = draw.min(cap_end.saturating_sub(start));
}
align_range(start, draw, total_bytes).ok()
}
fn first_run_source(
&self,
total_bytes: u64,
coverages: &[decdn_protocol::Coverage],
) -> Option<usize> {
let end = if self.len == 0 {
total_bytes
} else {
self.offset.saturating_add(self.len).min(total_bytes)
};
if self.offset >= end {
return None;
}
let gap = whole_range_chunks(self.offset, end - self.offset);
let rank: Vec<usize> = (0..coverages.len()).collect();
let (runs, _) = super::ranged_pull::plan_over(&gap, total_bytes, coverages, &rank);
let run = runs.first()?;
let block_bytes = decdn_protocol::discovery_block_bytes();
(run.offset / block_bytes == self.offset / block_bytes).then_some(run.source_ix)
}
}
fn handshake_order(
coverages: &[decdn_protocol::Coverage],
total_bytes: Option<u64>,
prime: Option<PrimeLeg>,
) -> Vec<usize> {
let mut order: Vec<usize> = (0..coverages.len()).collect();
let first = prime
.zip(total_bytes)
.and_then(|(prime, total)| prime.first_run_source(total, coverages));
if let Some(first) = first {
order.retain(|&ix| ix != first);
order.insert(0, first);
}
order
}
fn covered_handshake_range(
coverage: &decdn_protocol::Coverage,
total_bytes: u64,
) -> Option<(u64, u64)> {
let blocks = decdn_protocol::num_blocks(total_bytes);
let first = coverage.covered_blocks().find(|&block| block < blocks)?;
let start = u64::from(first).checked_mul(decdn_protocol::discovery_block_bytes())?;
(start < total_bytes).then(|| (start, CHUNK_GROUP_BYTES.min(total_bytes - start)))
}
#[derive(Debug, Clone)]
struct PrimeKey {
candidate_ix: usize,
pool_id: B256,
ledger: Arc<PoolLedger>,
range: AlignedRange,
opened_at: tokio::time::Instant,
}
impl PrimeKey {
fn answers(
&self,
source_ix: usize,
pool_id: B256,
ledger: &Arc<PoolLedger>,
first_leg: Option<&AlignedRange>,
) -> bool {
self.candidate_ix == source_ix
&& self.pool_id == pool_id
&& Arc::ptr_eq(&self.ledger, ledger)
&& first_leg == Some(&self.range)
}
}
pub(crate) struct PrimedHandshake {
key: PrimeKey,
prime: PrimeLeg,
header: UpstreamPullHeader,
reader: TimedReader<PullReader>,
}
struct DownstreamWait {
hash: Hash,
watch: DownstreamWatch,
metrics: Arc<crate::metrics::Metrics>,
}
const PULL_WAIT_WARN_AFTER: Duration = Duration::from_secs(45);
const PULL_WAIT_WARN_MARGIN: Duration = Duration::from_secs(10);
const _: () = assert!(
PULL_WAIT_WARN_AFTER.as_secs()
>= QUIC_MAX_IDLE_TIMEOUT.as_secs() + PULL_WAIT_WARN_MARGIN.as_secs()
&& PULL_WAIT_WARN_AFTER.as_secs()
>= PROOF_WAIT_CEILING.as_secs() + PULL_WAIT_WARN_MARGIN.as_secs(),
"PULL_WAIT_WARN_AFTER must exceed the QUIC idle timeout and the proof-wait ceiling \
by PULL_WAIT_WARN_MARGIN, or a vanished payer trips the warning"
);
impl DownstreamWait {
fn for_session(
session: &FillSession,
hash: Hash,
metrics: Arc<crate::metrics::Metrics>,
) -> Self {
Self {
hash,
watch: session.downstream_watch(),
metrics,
}
}
fn frontier(&self) -> DownstreamFrontier {
DownstreamFrontier {
served_paid: self.watch.served_paid(),
serve_demand: self.watch.serve_demand(),
}
}
}
impl PacingWait for DownstreamWait {
fn wait(
&self,
observed: DownstreamFrontier,
reason: WaitReason,
) -> Pin<Box<dyn Future<Output = ()> + Send + '_>> {
match reason {
WaitReason::WindowFull => self.metrics.node_pull_through_window_paused(),
WaitReason::MinDraw => self.metrics.node_pull_through_min_draw_waits(),
}
Box::pin(async move {
let _timer = PauseTimer {
started: tokio::time::Instant::now(),
metrics: &self.metrics,
};
let mut past =
std::pin::pin!(self.watch.past(observed.served_paid, observed.serve_demand));
if tokio::time::timeout(PULL_WAIT_WARN_AFTER, &mut past)
.await
.is_err()
{
warn!(
hash = %self.hash,
?reason,
served_paid = observed.served_paid,
serve_demand = observed.serve_demand,
waited_secs = PULL_WAIT_WARN_AFTER.as_secs(),
"pull still paused on its downstream frontiers; a stalled payer or a \
pacing regression holds it",
);
past.await;
}
})
}
}
struct PauseTimer<'a> {
started: tokio::time::Instant,
metrics: &'a crate::metrics::Metrics,
}
impl Drop for PauseTimer<'_> {
fn drop(&mut self) {
self.metrics.node_pull_through_wait(self.started.elapsed());
}
}
fn is_internal_fault(err: &anyhow::Error) -> bool {
err.chain().any(|cause| {
matches!(
cause.downcast_ref::<CacheError>(),
Some(CacheError::Internal(_))
)
})
}
fn content_len(ranges: &ChunkRanges, total: u64) -> u64 {
let boundaries = ranges.boundaries();
let mut it = boundaries.iter();
let mut sum = 0u64;
while let Some(a) = it.next() {
let start = a.0.saturating_mul(CHUNK_BYTES).min(total);
let end = match it.next() {
Some(b) => b.0.saturating_mul(CHUNK_BYTES).min(total),
None => total,
};
sum = sum.saturating_add(end.saturating_sub(start));
}
sum
}
fn is_bao_corruption(err: &anyhow::Error) -> bool {
err.chain().any(|cause| {
if cause.downcast_ref::<CacheError>().is_some_and(|e| {
matches!(
e,
CacheError::VerifyFailed { .. } | CacheError::HashMismatch { .. }
)
}) || cause.downcast_ref::<ClientPullHashMismatch>().is_some()
{
return true;
}
let msg = cause.to_string();
msg.contains("more than the") || msg.contains("after the promised total")
})
}
async fn spanning_cached_candidates(
deps: &NodeOriginDeps,
target: DhtHash,
hash: Hash,
requester: [u8; 32],
) -> Option<Vec<Candidate>> {
let mut cached = cached_candidates(deps, target).await?;
cached.retain(|candidate| candidate.node_id != requester);
if coverage_spans(&cached) {
return Some(cached);
}
debug!(
%hash,
cached = cached.len(),
"node-origin: probe-cache entry does not span the blob; running a fresh lookup"
);
None
}
impl NodeOrigin {
pub(crate) async fn open_pull_leg(
&self,
hash: Hash,
namespace_id: U256,
prime: Option<PrimeLeg>,
requester: [u8; 32],
) -> Result<PullLegTarget, PullMiss> {
let deps = self.deps.get().ok_or(PullMiss::Clean)?;
let _pending = self.enter_open(hash);
let hash_bytes = *hash.as_bytes();
let target = DhtHash::from_bytes(hash_bytes);
let namespace_bytes = namespace_id.to_be_bytes::<32>();
let mut budget = MAX_PROVIDER_ATTEMPTS;
let mut attempt_metered = false;
let mut miss = PullMiss::Clean;
if let Some(cached) = spanning_cached_candidates(deps, target, hash, requester).await {
deps.metrics.probe_cache_hit();
deps.metrics.node_pull_attempt();
attempt_metered = true;
let outcome = self
.handshake_from_candidates(deps, &cached, hash_bytes, namespace_id, budget, prime)
.await;
match outcome.payload {
Ok((total_bytes, primed)) => {
return Ok(PullLegTarget {
total_bytes,
namespace_id: namespace_bytes,
candidates: cached,
primed,
});
}
Err(failed) => miss = miss.or(failed),
}
budget = budget.saturating_sub(outcome.attempts);
deps.probe_cache.invalidate(&target);
if budget == 0 {
debug!(%hash, "node-origin: probe-cache candidates exhausted the pull-leg budget");
return Err(miss);
}
} else {
deps.metrics.probe_cache_miss();
}
let mut providers = discover(deps, hash_bytes, namespace_id).await;
let requester_id = DhtNodeId::from_bytes(requester);
providers.retain(|provider| *provider != requester_id);
if providers.is_empty() {
if !attempt_metered {
deps.metrics.node_pull_no_providers();
}
debug!(%hash, "node-origin: no providers discovered for the pull leg");
return Err(miss);
}
if !attempt_metered {
deps.metrics.node_pull_attempt();
}
let mut ranked = probe_and_rank(
deps,
providers,
hash_bytes,
ProbeGather::CoverageUnion,
namespace_id,
)
.await;
ranked.retain(|candidate| candidate.node_id != requester);
let outcome = self
.handshake_from_candidates(deps, &ranked, hash_bytes, namespace_id, budget, prime)
.await;
match outcome.payload {
Ok((total_bytes, primed)) => Ok(PullLegTarget {
total_bytes,
namespace_id: namespace_bytes,
candidates: ranked,
primed,
}),
Err(failed) => Err(miss.or(failed)),
}
}
async fn handshake_from_candidates(
&self,
deps: &NodeOriginDeps,
ranked: &[Candidate],
hash_bytes: [u8; 32],
namespace_id: U256,
budget: usize,
prime: Option<PrimeLeg>,
) -> PullOutcome<(u64, Option<PrimedHandshake>)> {
let mut attempts = 0;
let mut miss = PullMiss::Clean;
let coverages: Vec<decdn_protocol::Coverage> =
ranked.iter().map(|c| c.coverage.clone()).collect();
let total_hint = ranked.iter().find_map(|c| c.total_bytes_hint);
let visits = handshake_order(&coverages, total_hint, prime)
.into_iter()
.filter_map(|ix| ranked.get(ix).map(|candidate| (ix, candidate)));
for (candidate_ix, candidate) in visits.take(budget) {
attempts += 1;
match self
.handshake_total_bytes(
deps,
candidate,
candidate_ix,
hash_bytes,
namespace_id,
prime,
)
.await
{
Ok(answer) => {
return PullOutcome {
payload: Ok(answer),
attempts,
unfunded: Vec::new(),
};
}
Err(failed) => miss = miss.or(failed),
}
}
PullOutcome {
payload: Err(miss),
attempts,
unfunded: Vec::new(),
}
}
#[allow(clippy::too_many_lines)]
async fn handshake_total_bytes(
&self,
deps: &NodeOriginDeps,
candidate: &Candidate,
candidate_ix: usize,
hash_bytes: [u8; 32],
namespace_id: U256,
prime: Option<PrimeLeg>,
) -> Result<(u64, Option<PrimedHandshake>), PullMiss> {
let Ok(pk) = PublicKey::from_bytes(&candidate.node_id) else {
return Err(PullMiss::Clean);
};
let Some(provider_addr) = deps
.addr_resolver
.address_of(&DhtNodeId::from_bytes(candidate.node_id))
else {
debug!("node-origin: pull-leg candidate has no resolvable operator address; skipping");
return Err(PullMiss::Clean);
};
let heat = heat_of(deps, hash_bytes);
let rate_ceiling =
match economic_ceiling(deps, candidate.node_id.into(), heat, candidate.rate_per_mb) {
EconGate::Allow { rate_ceiling, .. } => rate_ceiling,
EconGate::Skip => return Err(PullMiss::BelowMargin),
};
let ctx = match deps
.buyer
.open_or_reuse_pool(provider_addr, POOL_OPEN_CALLER_BUDGET)
.await
{
Ok(ctx) => ctx,
Err(err) => return Err(record_pool_open_failure(deps, provider_addr, &err)),
};
let local_miss = |err: &anyhow::Error| {
PullMiss::for_verdict(classify_pull_failure(
deps,
pk,
provider_addr,
hash_bytes,
None,
err,
))
};
let ctx = bind_upstream_ctx(deps, ctx).map_err(|err| local_miss(&err))?;
let deadlines = deps.config.deadlines().map_err(|err| local_miss(&err))?;
let ledger = lane_ledger(deps, provider_addr, &ctx);
let namespace_bytes = namespace_id.to_be_bytes::<32>();
let stream_guard = deps.metrics.outbound_stream_guard();
let hint = candidate.total_bytes_hint;
let mut hint_wrong = false;
let predicted = prime.zip(hint).and_then(|(prime, total)| {
prime
.predicted_leg(total, &candidate.coverage, deps.config.max_blob_size_bytes)
.map(|range| (prime, range))
});
if let Some((prime, range)) = predicted {
let source = PeerSource::new(
&deps.endpoint,
EndpointAddr::new(pk),
Arc::new(std::sync::Mutex::new(ctx.clone())),
Arc::clone(&ledger),
&deps.slash_domain,
provider_addr,
namespace_bytes,
deps.config.max_blob_size_bytes,
rate_ceiling,
deadlines,
None,
)
.with_dial_runtime(deps.dial_runtime.clone());
let handshake = timed_open(
source.open(hash_bytes, range.clone()),
Arc::clone(&deps.metrics),
);
match handshake.await {
Ok((header, reader)) => {
drop(stream_guard);
let total_bytes = header.total_bytes;
if hint != Some(total_bytes) {
warn!(
peer = %pk,
hash = %Hash::from(hash_bytes),
signed = total_bytes,
?hint,
"node-origin: the provider signs another size than it probed; \
closing the first-leg handshake pull"
);
return Ok((total_bytes, None));
}
let primed = PrimedHandshake {
key: PrimeKey {
candidate_ix,
pool_id: ctx.pool_id,
ledger,
range,
opened_at: tokio::time::Instant::now(),
},
prime,
header,
reader,
};
return Ok((total_bytes, Some(primed)));
}
Err(err) if decdn_client::is_range_past_end(&err) => {
hint_wrong = true;
debug!(
peer = %pk,
hash = %Hash::from(hash_bytes),
"node-origin: the first-leg handshake runs past the blob's end ({err:#}); \
opening the whole blob, as the probed size is wrong"
);
}
Err(err) => {
drop(stream_guard);
let verdict =
handshake_verdict(deps, pk, provider_addr, hash_bytes, ctx.pool_id, &err);
return Err(PullMiss::for_verdict(verdict));
}
}
}
let (byte_offset, byte_len) = hint
.filter(|_| !hint_wrong)
.and_then(|total| covered_handshake_range(&candidate.coverage, total))
.unwrap_or((0, 0));
let (header, probe) = match open_progressive_upstream(
&deps.endpoint,
EndpointAddr::new(pk),
&ctx,
Arc::clone(&ledger),
&deps.slash_domain,
provider_addr,
hash_bytes,
namespace_bytes,
byte_offset,
now_micros(),
deps.config.max_blob_size_bytes,
rate_ceiling,
deadlines,
byte_len,
Some(&deps.dial_runtime),
)
.await
{
Ok(pair) => pair,
Err(err) => {
drop(stream_guard);
let verdict =
handshake_verdict(deps, pk, provider_addr, hash_bytes, ctx.pool_id, &err);
return Err(PullMiss::for_verdict(verdict));
}
};
let total_bytes = header.total_bytes;
let _ = probe.abort();
drop(stream_guard);
Ok((total_bytes, None))
}
}
fn handshake_verdict(
deps: &NodeOriginDeps,
pk: PublicKey,
provider_addr: Address,
hash_bytes: [u8; 32],
pool_id: B256,
err: &anyhow::Error,
) -> PullVerdict {
record_backpressure_refusal(deps, provider_addr, err).unwrap_or_else(|| {
classify_pull_failure(deps, pk, provider_addr, hash_bytes, Some(pool_id), err)
})
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn run_pull_leg(
deps_lock: Arc<OnceLock<NodeOriginDeps>>,
target: PullLegTarget,
engine: CacheEngine,
hash: Hash,
offset: u64,
len: u64,
credit_ramp_divisor: u64,
credit_floor: u64,
credit_max: u64,
paid_carried: u64,
session: Arc<FillSession>,
cancel: CancellationToken,
) {
let hash_bytes = *hash.as_bytes();
let PullLegTarget {
total_bytes,
namespace_id,
candidates,
primed,
} = target;
let Some(deps) = deps_lock.get() else {
session.mark_ended(Err(FillError::new(
"node-origin pull leg lost its dependencies mid-serve",
)));
return;
};
let deadlines = match deps.config.deadlines() {
Ok(deadlines) => deadlines,
Err(err) => {
session.mark_ended(Err(FillError::new(format!(
"node-origin ranged pull has unusable deadlines: {err:#}"
))));
return;
}
};
let admit_store = NodeAdmitStore::new(engine, hash, total_bytes, Some(Arc::clone(&session)))
.counting_received(Arc::clone(&deps.metrics));
let pacer = RampPacer {
divisor: credit_ramp_divisor,
floor: credit_floor,
credit_max,
paid_base: session.served_start(),
paid_carried,
};
let pacing_wait = DownstreamWait::for_session(&session, hash, Arc::clone(&deps.metrics));
let coverages: Vec<decdn_protocol::Coverage> =
candidates.iter().map(|c| c.coverage.clone()).collect();
let sink = PeerRunSink {
deps,
deps_lock: &deps_lock,
candidates: &candidates,
hash_bytes,
namespace_id,
total_bytes,
deadlines,
admit_store: &admit_store,
pacer: &pacer,
pacing_wait: &pacing_wait,
recovery: RecoveryGate::with_settle(settle_window(deps.config.event_poll_interval)),
gap_seen: std::sync::Mutex::new(None),
seen_deposit: std::sync::Mutex::new(deps.buyer.pool_deposit().unwrap_or(U256::ZERO)),
cancel: &cancel,
primed: std::sync::Mutex::new(primed),
};
let outcome = assemble(&sink, &coverages, offset, len, total_bytes).await;
match &outcome {
AssembleOutcome::Complete => deps.metrics.outbound_stream_ended(true),
AssembleOutcome::Unavailable(_)
| AssembleOutcome::Backpressured
| AssembleOutcome::FundingNeeded
| AssembleOutcome::RecoveryFailed
| AssembleOutcome::Terminal(_) => {
deps.metrics.outbound_stream_ended(false);
}
AssembleOutcome::Cancelled => {}
}
session.mark_ended(assembly_result(
outcome,
hash,
offset,
len,
candidates.len(),
));
}
fn assembly_result(
outcome: AssembleOutcome,
hash: Hash,
offset: u64,
len: u64,
candidates: usize,
) -> Result<(), FillError> {
match outcome {
AssembleOutcome::Complete | AssembleOutcome::Cancelled => Ok(()),
AssembleOutcome::Unavailable(cause) => {
warn!(
%hash,
offset,
len,
candidates,
cause = cause.as_str(),
"node-origin ranged pull ended before the range was filled"
);
Err(FillError::new(format!(
"node-origin ranged pull: {}",
cause.as_str()
)))
}
AssembleOutcome::Backpressured => Err(FillError::new(
"node-origin ranged pull: the only holder of a still-missing range kept refusing \
for backpressure",
)),
AssembleOutcome::FundingNeeded => {
warn!(
%hash,
offset,
len,
"node-origin ranged pull ended funding needed: the holders of a still-missing \
range refuse this node's funding and no recovery step can raise it"
);
Err(FillError::new(
"node-origin ranged pull: funding needed (this node's buyer pool)",
))
}
AssembleOutcome::RecoveryFailed => Err(FillError::new(
"node-origin ranged pull: the funding recovery step failed (this node's buyer pool)",
)),
AssembleOutcome::Terminal(err) => Err(err),
}
}
struct PeerRunSink<'a> {
deps: &'a NodeOriginDeps,
deps_lock: &'a Arc<OnceLock<NodeOriginDeps>>,
candidates: &'a [Candidate],
hash_bytes: [u8; 32],
namespace_id: [u8; 32],
total_bytes: u64,
deadlines: PullDeadlines,
admit_store: &'a NodeAdmitStore,
pacer: &'a RampPacer,
pacing_wait: &'a DownstreamWait,
recovery: RecoveryGate,
gap_seen: std::sync::Mutex<Option<u64>>,
seen_deposit: std::sync::Mutex<U256>,
cancel: &'a CancellationToken,
primed: std::sync::Mutex<Option<PrimedHandshake>>,
}
impl PeerRunSink<'_> {
async fn adopt_primed(
&self,
source: &PrimedSource<TimedSource<PeerSource<'_>>>,
run: &CoveredRun,
pool_id: B256,
ledger: &Arc<PoolLedger>,
) {
let taken = self
.primed
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.take();
let Some(primed) = taken else {
return;
};
let leg = match first_leg(
self.admit_store,
&[(run.offset, run.len)],
primed.prime.window,
self.deps.config.max_blob_size_bytes,
)
.await
{
Ok(leg) => leg,
Err(err) => {
warn!(
error = %format!("{err:#}"),
"node-origin: the store read to match the primed handshake leg failed; \
closing it"
);
return;
}
};
if primed
.key
.answers(run.source_ix, pool_id, ledger, leg.as_ref())
{
source.prime(
self.hash_bytes,
primed.key.range,
primed.header,
primed.reader,
primed.key.opened_at,
);
} else {
debug!(
hash = %Hash::from(self.hash_bytes),
source_ix = run.source_ix,
primed_ix = primed.key.candidate_ix,
"node-origin: the first run does not open the primed handshake leg on its \
lane; closing it"
);
}
}
}
const BACKPRESSURE_BACKOFF_BASE: Duration = Duration::from_millis(250);
const BACKPRESSURE_BACKOFF_MAX: Duration = Duration::from_secs(4);
fn backpressure_backoff(attempt: u32) -> Duration {
let doublings = attempt.saturating_sub(1).min(8);
BACKPRESSURE_BACKOFF_BASE
.saturating_mul(1 << doublings)
.min(BACKPRESSURE_BACKOFF_MAX)
}
impl RunSink for PeerRunSink<'_> {
async fn backoff(&self, attempt: u32) -> bool {
tokio::select! {
() = self.cancel.cancelled() => false,
() = tokio::time::sleep(backpressure_backoff(attempt)) => {
self.deps.metrics.node_pull_backpressure_backoffs();
true
}
}
}
fn backpressure_exhausted(&self, source_ix: usize, waits: u32) {
let Some(pk) = self
.candidates
.get(source_ix)
.and_then(|c| PublicKey::from_bytes(&c.node_id).ok())
else {
return;
};
record_backpressure_exhausted(self.deps, pk, self.hash_bytes, waits);
}
async fn recover(&self, gap_chunks: u64) -> Result<(), RecoveryEnd> {
{
let mut seen = self
.gap_seen
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if let Some(prev) = *seen
&& gap_chunks < prev
{
self.recovery
.record_verified((prev - gap_chunks).saturating_mul(CHUNK_BYTES));
}
*seen = Some(gap_chunks);
}
if self.recovery.settling(tokio::time::Instant::now()) {
return tokio::select! {
() = self.cancel.cancelled() => Err(RecoveryEnd::FundingNeeded),
() = tokio::time::sleep(SETTLE_POLL_STEP) => Ok(()),
};
}
let seen = *self
.seen_deposit
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
match super::node_step(&self.recovery, self.deps, seen, self.candidates.len()).await {
Ok(deposit) => {
*self
.seen_deposit
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner) = deposit;
Ok(())
}
Err(super::PullMiss::LocalFault) => Err(RecoveryEnd::Failed),
Err(_) => Err(RecoveryEnd::FundingNeeded),
}
}
async fn missing(&self, offset: u64, len: u64) -> ChunkRanges {
self.admit_store
.missing_ranges(offset, len)
.await
.unwrap_or_else(|_| whole_range_chunks(offset, len))
}
async fn drive_run(&self, run: CoveredRun) -> RunOutcome {
let span = tracing::info_span!(
"upstream_stream",
otel.kind = "client",
direction = "outbound",
peer = tracing::field::Empty,
local_node_id = %self.deps.endpoint.id(),
hash = %DhtHash::from_bytes(self.hash_bytes),
pool_id = tracing::field::Empty,
outcome = tracing::field::Empty,
);
if let Some(candidate) = self.candidates.get(run.source_ix) {
span.record(
"peer",
tracing::field::display(DhtNodeId::from_bytes(candidate.node_id)),
);
}
let outcome = self.drive_run_in_span(run).instrument(span.clone()).await;
span.record("outcome", outcome.as_str());
outcome
}
}
impl PeerRunSink<'_> {
#[allow(clippy::cognitive_complexity, clippy::too_many_lines)]
async fn drive_run_in_span(&self, run: CoveredRun) -> RunOutcome {
if self.cancel.is_cancelled() {
return RunOutcome::Cancelled;
}
let Some(candidate) = self.candidates.get(run.source_ix) else {
return RunOutcome::Reassign;
};
let Ok(pk) = PublicKey::from_bytes(&candidate.node_id) else {
return RunOutcome::Reassign;
};
let Some(provider_addr) = self
.deps
.addr_resolver
.address_of(&DhtNodeId::from_bytes(candidate.node_id))
else {
debug!("node-origin: ranged-pull run candidate has no resolvable operator address");
return RunOutcome::Reassign;
};
let heat = heat_of(self.deps, self.hash_bytes);
let (rate_ceiling, speculative) = match economic_ceiling(
self.deps,
candidate.node_id.into(),
heat,
candidate.rate_per_mb,
) {
EconGate::Allow {
rate_ceiling,
speculative,
} => (rate_ceiling, speculative),
EconGate::Skip => return RunOutcome::Reassign,
};
let ctx = match self
.deps
.buyer
.open_or_reuse_pool(provider_addr, POOL_OPEN_CALLER_BUDGET)
.await
{
Ok(ctx) => ctx,
Err(err) => {
return match record_pool_open_failure(self.deps, provider_addr, &err) {
PullMiss::LocalFault | PullMiss::FundingNeeded => {
RunOutcome::Terminal(FillError::new(format!(
"node-origin ranged pull: local buyer fault opening a lane: {err:#}"
)))
}
_ => RunOutcome::Reassign,
};
}
};
let ctx = match bind_upstream_ctx(self.deps, ctx) {
Ok(ctx) => ctx,
Err(err) => {
let verdict = classify_pull_failure(
self.deps,
pk,
provider_addr,
self.hash_bytes,
None,
&err,
);
return match PullMiss::for_verdict(verdict) {
PullMiss::LocalFault => RunOutcome::Terminal(FillError::new(format!(
"node-origin ranged pull: local bind fault: {err:#}"
))),
_ => RunOutcome::Reassign,
};
}
};
let pool_id = ctx.pool_id;
tracing::Span::current().record("pool_id", tracing::field::display(pool_id));
let prior_amount = ctx.prior_amount;
let ledger = lane_ledger(self.deps, provider_addr, &ctx);
let ctx = Arc::new(std::sync::Mutex::new(ctx));
let run_bytes = match self.admit_store.missing_ranges(run.offset, run.len).await {
Ok(ranges) => content_len(&ranges, self.total_bytes),
Err(_) => run.len,
};
let _settle = SettleOnDrop {
deps: Arc::clone(self.deps_lock),
provider_addr,
pool_id,
prior_amount,
ledger: Arc::clone(&ledger),
};
let peer_source = PeerSource::new(
&self.deps.endpoint,
EndpointAddr::new(pk),
Arc::clone(&ctx),
Arc::clone(&ledger),
&self.deps.slash_domain,
provider_addr,
self.namespace_id,
self.deps.config.max_blob_size_bytes,
rate_ceiling,
self.deadlines,
None,
)
.with_dial_runtime(self.deps.dial_runtime.clone());
let peer_source = PrimedSource::new(TimedSource::new(
peer_source,
Arc::clone(&self.deps.metrics),
));
self.adopt_primed(&peer_source, &run, pool_id, &ledger)
.await;
let pacing_wait = self.pacing_wait;
let downstream_reader = move || pacing_wait.frontier();
let ledgers = &self.deps.ledgers;
let spent = move || ledgers.pool_committed(pool_id);
let pool = SharedPool {
spent: &spent,
quotes: None,
};
let started = Instant::now();
let cancelled;
let result = tokio::select! {
biased;
r = drive(
self.admit_store,
&peer_source,
self.pacer,
&ctx,
&ledger,
self.hash_bytes,
run.offset,
run.len,
None,
Some(self.pacing_wait),
Some(&downstream_reader),
Some(&pool),
) => {
cancelled = false;
r
}
() = self.cancel.cancelled() => {
cancelled = true;
Ok(())
}
};
let elapsed = started.elapsed();
peer_source.clear();
if cancelled {
return RunOutcome::Cancelled;
}
match result {
Ok(()) => {
if speculative {
self.deps.config.warming.debit_speculative(
(*pk.as_bytes()).into(),
Hash::from_bytes(self.hash_bytes),
candidate.rate_per_mb.saturating_mul(mb_of(run_bytes)),
);
}
record_outcome(
self.deps,
pk,
&Outcome::Delivered {
bytes: run_bytes,
elapsed,
},
);
RunOutcome::Filled
}
Err(err) => {
if speculative && let Some(stuck) = err.downcast_ref::<LegNoProgress>() {
self.deps.config.warming.debit_speculative(
(*pk.as_bytes()).into(),
Hash::from_bytes(self.hash_bytes),
candidate.rate_per_mb.saturating_mul(mb_of(stuck.len)),
);
}
if is_bao_corruption(&err) {
tracing::warn!(
provider = %pk, %provider_addr,
"node ranged pull: upstream served bao-corrupt bytes; scoring Corruption"
);
record_outcome(self.deps, pk, &Outcome::Corruption);
self.deps.metrics.node_pull_through_upstream_verify_failed();
} else if record_backpressure_refusal(self.deps, provider_addr, &err).is_some() {
return RunOutcome::Backpressure;
} else {
let _ = classify_pull_failure(
self.deps,
pk,
provider_addr,
self.hash_bytes,
Some(pool_id),
&err,
);
}
if ends_the_assembly(&err) {
RunOutcome::Terminal(FillError::new(format!("{err:#}")))
} else if classify(&err) == Fault::Unaffordable {
RunOutcome::Unfunded
} else {
RunOutcome::Reassign
}
}
}
}
}
fn ends_the_assembly(err: &anyhow::Error) -> bool {
if super::is_local_store_fault(err) {
return true;
}
match classify(err) {
Fault::Fatal(_) => true,
Fault::Unaffordable | Fault::Source | Fault::Transient => false,
}
}
fn whole_range_chunks(offset: u64, len: u64) -> ChunkRanges {
let start = offset / CHUNK_BYTES;
let end = offset.saturating_add(len).div_ceil(CHUNK_BYTES);
ChunkRanges::from(bao_tree::ChunkNum(start)..bao_tree::ChunkNum(end))
}
fn local_bookkeeping_ctx() -> PoolContext {
PoolContext::new(
B256::ZERO,
U256::MAX,
Arc::new(PrivateKeySigner::random()),
decdn_incentive::voucher_domain(0, Address::ZERO),
)
}
#[allow(
clippy::too_many_arguments,
reason = "the pull leg's inputs are all owned so it can run on its own runtime"
)]
pub(crate) async fn run_local_pull_leg(
metrics: Arc<crate::metrics::Metrics>,
engine: CacheEngine,
source: BackendSource,
hash: Hash,
offset: u64,
len: u64,
credit_ramp_divisor: u64,
credit_floor: u64,
credit_max: u64,
paid_carried: u64,
total_bytes: u64,
session: Arc<FillSession>,
cancel: CancellationToken,
) {
let hash_bytes = *hash.as_bytes();
let admit_store = NodeAdmitStore::new(engine, hash, total_bytes, Some(Arc::clone(&session)));
let ledger = source.ledger();
let ctx = Arc::new(std::sync::Mutex::new(local_bookkeeping_ctx()));
let pacer = RampPacer {
divisor: credit_ramp_divisor,
floor: credit_floor,
credit_max,
paid_base: session.served_start(),
paid_carried,
};
let pacing_wait = DownstreamWait::for_session(&session, hash, Arc::clone(&metrics));
let downstream_reader = || pacing_wait.frontier();
let cancelled;
let result = tokio::select! {
biased;
r = drive(
&admit_store,
&source,
&pacer,
&ctx,
&ledger,
hash_bytes,
offset,
len,
None,
Some(&pacing_wait),
Some(&downstream_reader),
None,
) => {
cancelled = false;
r
}
() = cancel.cancelled() => {
cancelled = true;
Ok(())
}
};
if !cancelled && let Err(err) = &result {
if let Some(stuck) = err.downcast_ref::<LegNoProgress>() {
super::warn_leg_no_progress(&metrics, *hash.as_bytes(), None, stuck);
} else if is_internal_fault(err) {
metrics.node_pull_local_fault();
tracing::error!(
%hash,
error = %format_args!("{err:#}"),
"own-origin pull leg failed on an internal fault (code bug), not the origin"
);
} else {
metrics.node_pull_local_fault();
tracing::warn!(
%hash,
error = %format_args!("{err:#}"),
"own-origin pull leg failed; local-origin fault (no upstream to score)"
);
}
}
session.mark_ended(result.map_err(|e| FillError::new(format!("{e:#}"))));
}
#[allow(
clippy::unwrap_used,
clippy::expect_used,
clippy::indexing_slicing,
clippy::cast_possible_truncation,
reason = "tests"
)]
#[cfg(test)]
mod local_pull_leg_tests;
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] mod downstream_wait_tests;
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] mod prime_tests;
#[cfg(test)]
mod backpressure_backoff_tests;
#[cfg(test)]
mod assembly_fault_tests;