use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::str::FromStr;
use std::sync::{Arc, Mutex, OnceLock, PoisonError};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use alloy::dyn_abi::Eip712Domain;
use alloy::primitives::{Address, B256, Bytes, U256};
use alloy::signers::local::PrivateKeySigner;
use decdn_client::buyer_pool::{
LOW_WATER_DIVISOR, PoolNotOpen, ProgressWrite, TopUpUnconfirmed, ToppedUpPool, WalletShortfall,
ensure_allowance, escrowed_but_untracked, grade_deposit_credit, open_pool, pool_accepts_funds,
refill_amount, self_owned_lane_ctx, top_up, topped_up_effect,
};
use decdn_client::source::{Funder, SourceFuture};
use decdn_client::{
Connections, Cumulative, DownloadTarget, Downloader, Holder, LaneHandle, LaneLedgers,
NoAffordableSource, NoCache, NoSourceHasBlob, PeerHealth, PeerSource, PoolContext, PoolLedger,
PoolReplaced, ProgressClock, PullConfig, PullDeadlines, Recovery, RecoveryGate, StopPolicy,
Streamer, UpstreamRefused, UpstreamVoucherRejected, VoucherProgress, sign_client_binding,
};
use super::cli_sources::CliSources;
use decdn_common::cli::{self, common::expand_tilde};
use decdn_common::config::{DEFAULT_CHAIN_ID, FileConfig, load_file_config};
use decdn_incentive::buyer_pool::{AdvanceOutcome, BuyerPoolState, BuyerPoolStore};
use decdn_incentive::buyer_pool_redb::RedbBuyerPoolStore;
use decdn_incentive::eth_identity::{self, PasswordUse, load_signer};
use decdn_incentive::payment_pool::{PaymentPool, newest_solvent_owned_pool};
use anyhow::Context as _;
use decdn_incentive::rate::min_payment;
use decdn_incentive::{
CapabilityGrant, Deployment, LaneKey, PoolId, bind_node_id_domain, slash_judge_domain,
voucher_domain,
};
use decdn_protocol::client::StreamError;
use iroh::{Endpoint, EndpointAddr, PublicKey, RelayUrl};
use tokio_util::task::AbortOnDropHandle;
use decdn_client::UpstreamRateLimited;
use decdn_client::discovery::{self, NodeCandidate};
use decdn_client::endpoint as client_endpoint;
use decdn_client::probe::probe_once;
use decdn_client::provider;
use super::buyer_store::{ChainAdoption, DataDirSource, open_client_store_for_buy};
use super::fetch_timings::{FetchTimings, Mark};
use super::interrupt::{Interrupt, Interrupted};
use super::ordered_writes::OrderedWrites;
use super::tab_progress::TabProgress;
const SELECT_PROBE_TIMEOUT_MS: u64 = 5_000;
pub(crate) fn parse_hash(s: &str) -> anyhow::Result<[u8; 32]> {
let hex = s
.strip_prefix("0x")
.or_else(|| s.strip_prefix("b3:"))
.unwrap_or(s);
let h = blake3::Hash::from_hex(hex).map_err(|e| {
anyhow::anyhow!(
"invalid hash {s:?}: expected 64 hex chars (BLAKE3 digest), optional `0x`/`b3:` \
prefix: {e}"
)
})?;
Ok(*h.as_bytes())
}
pub(crate) fn micros_now() -> u64 {
u64::try_from(
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_or(0, |d| d.as_micros()),
)
.unwrap_or(u64::MAX)
}
fn new_progress_bar() -> indicatif::ProgressBar {
let style = indicatif::ProgressStyle::with_template(
"{spinner:.green} {bytes}/{total_bytes} {msg}[{wide_bar:.cyan/blue}]",
)
.unwrap_or_else(|_| indicatif::ProgressStyle::default_bar())
.progress_chars("=>-");
let bar = crate::logging::attach_progress_bar(
indicatif::ProgressBar::new(0).with_finish(indicatif::ProgressFinish::AndClear),
);
bar.set_style(style);
bar.enable_steady_tick(Duration::from_millis(120));
bar
}
#[derive(Debug)]
pub(crate) struct ResolvedChain {
pub(crate) rpc_url: String,
pub(crate) payment_pool: Address,
pub(crate) slash_judge: Address,
pub(crate) capacity_bond: Option<Address>,
pub(crate) chain_id: u64,
pub(crate) keystore: PathBuf,
pub(crate) keystore_password_file: Option<PathBuf>,
pub(crate) data_dir: PathBuf,
pub(crate) data_dir_source: DataDirSource,
pub(crate) region: Option<String>,
pub(crate) region_allowlist: Vec<decdn_protocol::Region>,
pub(crate) working_deposit: U256,
pub(crate) max_approve: bool,
}
impl ResolvedChain {
pub(crate) const fn deployment(&self) -> Deployment {
Deployment {
chain_id: self.chain_id,
payment_pool: self.payment_pool,
}
}
}
fn parse_region_allowlist(file: &FileConfig) -> Vec<decdn_protocol::Region> {
file.client
.as_ref()
.and_then(|c| c.region_allowlist.as_ref())
.into_iter()
.flatten()
.filter_map(|code| {
let parsed = decdn_protocol::Region::parse(code);
if parsed.is_none() {
tracing::warn!(
"client.region_allowlist entry {code:?} is not a recognized \
region code; dropping it from the filter"
);
}
parsed
})
.collect()
}
pub(crate) fn resolve_chain(
args: &cli::ClientFetchArgs,
file: &FileConfig,
) -> anyhow::Result<ResolvedChain> {
let bc = file.blockchain.as_ref();
let rpc_url = args
.rpc_url
.clone()
.or_else(|| bc.and_then(|b| b.rpc_url.clone()))
.ok_or_else(|| anyhow::anyhow!("rpc_url not set (--rpc-url or blockchain.rpc_url)"))?;
let pp_raw = args
.payment_pool_address
.clone()
.or_else(|| bc.and_then(|b| b.payment_pool_address.clone()))
.ok_or_else(|| {
anyhow::anyhow!(
"payment_pool_address not set (--payment-pool-address or \
blockchain.payment_pool_address)"
)
})?;
let payment_pool = super::chain_ctx::parse_nonzero_address(&pp_raw, "payment_pool_address")?;
let sj_raw = args
.slash_judge_address
.clone()
.or_else(|| bc.and_then(|b| b.slash_judge_address.clone()))
.ok_or_else(|| {
anyhow::anyhow!(
"slash_judge_address not set (--slash-judge-address or \
blockchain.slash_judge_address)"
)
})?;
let slash_judge = super::chain_ctx::parse_nonzero_address(&sj_raw, "slash_judge_address")?;
let capacity_bond = args
.capacity_bond_address
.clone()
.or_else(|| bc.and_then(|b| b.capacity_bond_address.clone()))
.map(|raw| super::chain_ctx::parse_nonzero_address(&raw, "capacity_bond_address"))
.transpose()?;
let region = args
.region
.clone()
.or_else(|| file.identity.as_ref().and_then(|i| i.region.clone()));
let region_allowlist = parse_region_allowlist(file);
let chain_id = args
.chain_id
.or_else(|| bc.and_then(|b| b.chain_id))
.unwrap_or(DEFAULT_CHAIN_ID);
let (data_dir, data_dir_source) = args
.data_dir
.clone()
.map(|p| (p, DataDirSource::Flag))
.or_else(|| {
file.identity
.as_ref()
.and_then(|i| i.data_dir.clone())
.map(|p| (p, DataDirSource::Config))
})
.map(|(p, source)| (expand_tilde(&p), source))
.or_else(|| cli::default_client_data_dir().map(|p| (p, DataDirSource::Default)))
.ok_or_else(|| {
anyhow::anyhow!("data_dir not set and no default available (pass --data-dir)")
})?;
let keystore = args
.keystore
.clone()
.or_else(|| bc.and_then(|b| b.eth_keystore.clone()))
.map_or_else(
|| eth_identity::keystore_path(&data_dir),
|p| expand_tilde(&p),
);
let working_deposit_micro_usdc = args
.working_deposit_micro_usdc
.or_else(|| bc.and_then(|b| b.buyer_working_deposit_micro_usdc))
.unwrap_or(decdn_common::config::DEFAULT_BUYER_WORKING_DEPOSIT_MICRO_USDC);
anyhow::ensure!(
working_deposit_micro_usdc > 0,
"buyer_working_deposit_micro_usdc must be > 0 (openPool reverts ZeroAmount on a \
zero deposit) — set --working-deposit-micro-usdc or \
blockchain.buyer_working_deposit_micro_usdc"
);
let working_deposit = U256::from(working_deposit_micro_usdc);
let max_approve = bc.and_then(|b| b.buyer_max_approve).unwrap_or(false);
Ok(ResolvedChain {
rpc_url,
payment_pool,
slash_judge,
capacity_bond,
chain_id,
keystore,
keystore_password_file: args.keystore_password_file.as_deref().map(expand_tilde),
data_dir,
data_dir_source,
region,
region_allowlist,
working_deposit,
max_approve,
})
}
#[derive(Debug, Clone, Copy)]
pub(crate) struct ProxyWarmingParams {
pub enabled: bool,
pub rtt_threshold_ms: f64,
pub margin_ms: f64,
}
impl ProxyWarmingParams {
pub(crate) fn from_args(args: &cli::ClientFetchArgs) -> Self {
let ms = |v: u64| f64::from(u32::try_from(v).unwrap_or(u32::MAX));
Self {
enabled: args.proxy_warming,
rtt_threshold_ms: ms(args.proxy_warming_rtt_threshold_ms),
margin_ms: ms(args.proxy_warming_margin_ms),
}
}
}
fn no_serve_target_error(
probe_count: usize,
unreachable: usize,
rate_limited: usize,
unverifiable: usize,
) -> anyhow::Error {
if unverifiable > 0 {
return anyhow::Error::new(ResolveConfigFault(format!(
"{unverifiable} of {probe_count} probed node(s) answered, but their probe \
signatures did not recover to the operator address each is registered \
under. That is usually local configuration rather than missing content: \
check that blockchain.slash_judge_address and blockchain.chain_id match \
the deployment these nodes registered against"
)));
}
let unusable = probe_count
.saturating_sub(unreachable)
.saturating_sub(rate_limited);
let shed = if rate_limited > 0 {
format!(", {rate_limited} rate-limited this client's probes")
} else {
String::new()
};
anyhow::anyhow!(
"none of the {probe_count} probed node(s) could serve the blob \
({unreachable} did not answer{shed}, {unusable} answered but were unusable)"
)
}
const PROBE_SHED_BACKOFFS_MS: [u64; 3] = [250, 500, 1000];
async fn probe_candidate(
endpoint: &Endpoint,
target: EndpointAddr,
hash: [u8; 32],
timestamp_us: u64,
) -> anyhow::Result<(
decdn_protocol::message::ProbeResponse,
decdn_protocol::ProbeResponseExt,
decdn_client::probe::ProbeRtt,
)> {
let mut backoffs = PROBE_SHED_BACKOFFS_MS.iter();
loop {
let res = probe_once(
endpoint,
target.clone(),
hash,
timestamp_us,
Duration::from_millis(SELECT_PROBE_TIMEOUT_MS),
)
.await;
match (&res, backoffs.next()) {
(Err(e), Some(&ms)) if probe_shed(e) => {
tokio::time::sleep(Duration::from_millis(ms)).await;
}
(Err(e), None) if probe_shed(e) => {
tracing::info!(
"{} kept rate-limiting this client's probe: {e:#}",
target.id
);
return res;
}
_ => return res,
}
}
}
fn probe_shed(err: &anyhow::Error) -> bool {
err.downcast_ref::<UpstreamRateLimited>().is_some()
}
const PROBE_SETTLE_AFTER_HOLDER: Duration = Duration::from_millis(250);
async fn settle_probes<T, F>(
probes: impl IntoIterator<Item = F>,
grace: Duration,
cold_grace: Option<Duration>,
is_holder: impl Fn(&T) -> bool,
is_answer: impl Fn(&T) -> bool,
) -> (Vec<T>, futures_util::stream::FuturesUnordered<F>)
where
F: std::future::Future<Output = T>,
{
use futures_util::StreamExt as _;
let mut pending: futures_util::stream::FuturesUnordered<F> = probes.into_iter().collect();
let mut settled = Vec::with_capacity(pending.len());
let mut deadline: Option<tokio::time::Instant> = None;
let mut holder_seen = false;
loop {
let next = match deadline {
None => pending.next().await,
Some(at) => match tokio::time::timeout_at(at, pending.next()).await {
Ok(next) => next,
Err(_) => break,
},
};
let Some(outcome) = next else { break };
let now = tokio::time::Instant::now();
if !holder_seen && is_holder(&outcome) {
holder_seen = true;
let at = now + grace;
deadline = Some(deadline.map_or(at, |cold| cold.min(at)));
} else if deadline.is_none()
&& let Some(cold) = cold_grace
&& is_answer(&outcome)
{
deadline = Some(now + cold);
}
settled.push(outcome);
}
if !pending.is_empty() {
tracing::debug!(
pending = pending.len(),
holder = holder_seen,
"probe round settled; the rest continue as its tail"
);
}
(settled, pending)
}
fn log_probe_result(
cand: &NodeCandidate,
res: &anyhow::Result<(
decdn_protocol::message::ProbeResponse,
decdn_protocol::ProbeResponseExt,
decdn_client::probe::ProbeRtt,
)>,
started: std::time::Instant,
) {
match res {
Ok((resp, _, rtt)) => tracing::debug!(
node = %cand.node_id,
elapsed_ms = started.elapsed().as_millis(),
rtt_ms = rtt.ms,
rtt_direct = rtt.direct,
has_blob = resp.body.has_blob,
"probe answered"
),
Err(error) => tracing::debug!(
node = %cand.node_id,
elapsed_ms = started.elapsed().as_millis(),
error = %format_args!("{error:#}"),
"probe failed"
),
}
}
pub(crate) enum ProbeOutcome {
Holder(discovery::Probed, (PublicKey, f64, u64)),
NonHolder(discovery::WarmingCandidate, (PublicKey, f64, u64)),
Unreachable,
RateLimited,
Unverifiable,
Unusable,
}
impl ProbeOutcome {
const fn is_holder(&self) -> bool {
matches!(self, Self::Holder(..))
}
const fn is_answer(&self) -> bool {
matches!(self, Self::Holder(..) | Self::NonHolder(..))
}
pub(crate) const fn sample(&self) -> Option<(PublicKey, f64, u64)> {
match self {
Self::Holder(_, sample) | Self::NonHolder(_, sample) => Some(*sample),
Self::Unreachable | Self::RateLimited | Self::Unverifiable | Self::Unusable => None,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ProbeRound {
Stream,
Settle,
}
impl ProbeRound {
const fn grace(self) -> Duration {
match self {
Self::Stream => Duration::ZERO,
Self::Settle => PROBE_SETTLE_AFTER_HOLDER,
}
}
const fn cold_grace(self) -> Option<Duration> {
match self {
Self::Stream => Some(PROBE_SETTLE_AFTER_HOLDER),
Self::Settle => None,
}
}
}
pub(crate) struct LateProbes {
pub(crate) tail: std::pin::Pin<Box<dyn futures_util::Stream<Item = ProbeOutcome> + Send>>,
pub(crate) best_holder_rtt_ms: f64,
pub(crate) warming: ProxyWarmingParams,
}
#[derive(Clone, Copy)]
pub(crate) struct ProbeOpts<'a> {
pub(crate) timings: Option<&'a FetchTimings>,
pub(crate) round: ProbeRound,
}
#[derive(Default)]
pub(crate) struct LateSlot(std::sync::Mutex<Option<LateProbes>>);
impl LateSlot {
pub(crate) const fn new(late: LateProbes) -> Self {
Self(std::sync::Mutex::new(Some(late)))
}
pub(crate) fn take(&self) -> Option<LateProbes> {
self.0
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.take()
}
}
struct ProbeTally {
holders: Vec<discovery::Probed>,
non_holders: Vec<discovery::WarmingCandidate>,
probed_samples: Vec<(PublicKey, f64, u64)>,
unreachable: usize,
rate_limited: usize,
unverifiable: usize,
}
impl ProbeTally {
fn of(outcomes: Vec<ProbeOutcome>) -> Self {
let mut tally = Self {
holders: Vec::new(),
non_holders: Vec::new(),
probed_samples: Vec::new(),
unreachable: 0,
rate_limited: 0,
unverifiable: 0,
};
for outcome in outcomes {
match outcome {
ProbeOutcome::Holder(holder, sample) => {
tally.probed_samples.push(sample);
tally.holders.push(holder);
}
ProbeOutcome::NonHolder(candidate, sample) => {
tally.probed_samples.push(sample);
tally.non_holders.push(candidate);
}
ProbeOutcome::Unreachable => tally.unreachable += 1,
ProbeOutcome::RateLimited => tally.rate_limited += 1,
ProbeOutcome::Unverifiable => tally.unverifiable += 1,
ProbeOutcome::Unusable => {}
}
}
tally
}
}
fn classify_probe(
cand: &NodeCandidate,
res: anyhow::Result<(
decdn_protocol::message::ProbeResponse,
decdn_protocol::ProbeResponseExt,
decdn_client::probe::ProbeRtt,
)>,
hash: [u8; 32],
timestamp_us: u64,
slash_domain: &alloy::sol_types::Eip712Domain,
) -> ProbeOutcome {
let (resp, resp_ext, rtt) = match res {
Ok(answer) => answer,
Err(e) if probe_shed(&e) => return ProbeOutcome::RateLimited,
Err(_) => return ProbeOutcome::Unreachable,
};
if let Err(e) = decdn_client::probe::verify_probe_response(
&resp,
cand.eth_address,
slash_domain,
hash,
timestamp_us,
) {
tracing::warn!(
"dropping an unverifiable probe response from {}: {e}",
cand.node_id
);
return ProbeOutcome::Unverifiable;
}
if !resp_ext.consistent_with(resp.body.has_blob) {
tracing::warn!(
"dropping a probe response from {} with has_blob/coverage mismatch",
cand.node_id
);
return ProbeOutcome::Unusable;
}
let rtt_ms = rtt.ms;
let sample = (cand.node_id, rtt_ms, resp.body.rate_per_mb);
if resp.body.has_blob {
ProbeOutcome::Holder(
discovery::Probed {
candidate: cand.clone(),
rtt_ms,
total_bytes: resp_ext.total_bytes,
coverage: resp_ext.coverage,
},
sample,
)
} else {
ProbeOutcome::NonHolder(
discovery::WarmingCandidate {
node_id: cand.node_id,
eth_address: cand.eth_address,
rtt_ms,
multiaddrs: cand.multiaddrs.clone(),
},
sample,
)
}
}
pub(crate) async fn probe_and_order(
endpoint: &Endpoint,
candidates: &[NodeCandidate],
relay_hint: Option<&RelayUrl>,
hash: [u8; 32],
warming: ProxyWarmingParams,
slash_domain: &alloy::sol_types::Eip712Domain,
round: ProbeRound,
) -> anyhow::Result<ResolvedTargets> {
let timestamp_us = micros_now();
let started = std::time::Instant::now();
let probes = candidates.iter().cloned().map(|cand| {
let target = probe_target(&cand, relay_hint.cloned());
let endpoint = endpoint.clone();
let slash_domain = slash_domain.clone();
async move {
let res = probe_candidate(&endpoint, target, hash, timestamp_us).await;
log_probe_result(&cand, &res, started);
classify_probe(&cand, res, hash, timestamp_us, &slash_domain)
}
});
let (outcomes, tail) = settle_probes(
probes,
round.grace(),
round.cold_grace(),
ProbeOutcome::is_holder,
ProbeOutcome::is_answer,
)
.await;
let ProbeTally {
holders,
non_holders,
probed_samples,
unreachable,
rate_limited,
unverifiable,
} = ProbeTally::of(outcomes);
if holders.is_empty() && non_holders.is_empty() {
return Err(no_serve_target_error(
candidates.len(),
unreachable,
rate_limited,
unverifiable,
));
}
if holders.is_empty() {
tracing::info!(
"no node that has answered holds the blob in cache; starting on {} reachable \
bonded non-holder(s) as pull-through serve targets — a node serves an authorized \
miss from its own origin (#1911), and a holder that answers later joins the fetch",
non_holders.len()
);
}
let size_hint = nearest_size_hint(&holders);
let late = late_slot(round, tail, &holders, warming);
let ordered = failover_order(holders, &non_holders, warming);
if let Some((node_id, proxy_rtt, best_holder_rtt)) = ordered.warming_lead {
tracing::info!(
"proxy-warming: routing through nearer non-holder {node_id} ({proxy_rtt:.1}ms) \
instead of the best holder ({best_holder_rtt:.1}ms) to seed a regional copy, \
falling back to the holder if it declines (ADR 037)",
);
}
Ok(ResolvedTargets {
candidates: ordered.order,
coverage_by_node: ordered.coverage_by_node,
probed_samples,
pinned: false,
size_hint,
late,
})
}
fn late_slot<F>(
round: ProbeRound,
tail: futures_util::stream::FuturesUnordered<F>,
holders: &[discovery::Probed],
warming: ProxyWarmingParams,
) -> LateSlot
where
F: std::future::Future<Output = ProbeOutcome> + Send + 'static,
{
use futures_util::StreamExt as _;
if round != ProbeRound::Stream || tail.is_empty() {
return LateSlot::default();
}
let best_holder_rtt_ms = holders
.iter()
.map(|h| h.rtt_ms)
.fold(f64::INFINITY, f64::min);
let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
let task = AbortOnDropHandle::new(tokio::spawn(tail.for_each(move |outcome| {
let _ = tx.send(outcome);
std::future::ready(())
})));
let answers = futures_util::stream::unfold((rx, task), |(mut rx, task)| async move {
rx.recv().await.map(|outcome| (outcome, (rx, task)))
});
LateSlot::new(LateProbes {
tail: Box::pin(answers),
best_holder_rtt_ms,
warming,
})
}
fn nearest_size_hint(holders: &[discovery::Probed]) -> Option<u64> {
holders
.iter()
.filter(|h| h.total_bytes.is_some())
.min_by(|a, b| a.rtt_ms.total_cmp(&b.rtt_ms))
.and_then(|h| h.total_bytes)
}
struct FailoverOrder {
order: Vec<NodeCandidate>,
warming_lead: Option<(PublicKey, f64, f64)>,
coverage_by_node: HashMap<PublicKey, decdn_protocol::Coverage>,
}
fn failover_order(
mut holders: Vec<discovery::Probed>,
non_holders: &[discovery::WarmingCandidate],
warming: ProxyWarmingParams,
) -> FailoverOrder {
if holders.is_empty() {
let mut cold = non_holders.iter().collect::<Vec<_>>();
cold.sort_by(|a, b| a.rtt_ms.total_cmp(&b.rtt_ms));
let order = cold
.iter()
.map(|c| NodeCandidate {
node_id: c.node_id,
eth_address: c.eth_address,
region_hint: None,
multiaddrs: c.multiaddrs.clone(),
})
.collect();
return FailoverOrder {
order,
warming_lead: None,
coverage_by_node: HashMap::new(),
};
}
holders.sort_by(|a, b| a.rtt_ms.total_cmp(&b.rtt_ms));
let coverage_by_node: HashMap<PublicKey, decdn_protocol::Coverage> = holders
.iter()
.map(|h| (h.candidate.node_id, h.coverage.clone()))
.collect();
let best_holder_rtt = holders
.iter()
.map(|h| h.rtt_ms)
.fold(f64::INFINITY, f64::min);
let proxy_order = if warming.enabled {
discovery::proxy_warming_order(
best_holder_rtt,
warming.rtt_threshold_ms,
warming.margin_ms,
non_holders,
)
} else {
Vec::new()
};
let warming_lead = proxy_order
.first()
.map(|nearest| (nearest.node_id, nearest.rtt_ms, best_holder_rtt));
let proxies = proxy_order.iter().map(|proxy| NodeCandidate {
node_id: proxy.node_id,
eth_address: proxy.eth_address,
region_hint: None,
multiaddrs: proxy.multiaddrs.clone(),
});
let order = proxies
.chain(holders.iter().map(|h| h.candidate.clone()))
.collect();
FailoverOrder {
order,
warming_lead,
coverage_by_node,
}
}
async fn discover_provider(
endpoint: &Endpoint,
chain: &ResolvedChain,
capacity_bond: Address,
relay_hint: Option<&RelayUrl>,
hash: [u8; 32],
args: &cli::ClientFetchArgs,
probe: ProbeOpts<'_>,
) -> anyhow::Result<ResolvedTargets> {
let mark_probe_start = || {
if let Some(timings) = probe.timings {
timings.mark(Mark::ProbeStart);
}
};
let peer_store = decdn_client::PeerStore::open(&chain.data_dir);
let store_cfg = decdn_client::StoreConfig::default();
if !args.rediscover
&& let Some(targets) =
store_fast_path(&peer_store, &store_cfg, args.max_sources, now_secs_cli())
{
return Ok(targets);
}
if !args.rediscover {
let cached = identity_fresh_candidates(&peer_store, &store_cfg, now_secs_cli());
let selected = select_with_widening(
cached,
chain.region.as_deref(),
chain.region_allowlist.as_slice(),
&store_cfg,
);
if selected.len() >= store_cfg.min_fresh_candidates {
let warming = ProxyWarmingParams::from_args(args);
let slash_dom = slash_judge_domain(chain.chain_id, chain.slash_judge);
mark_probe_start();
let targets = probe_and_order(
endpoint,
&selected,
relay_hint,
hash,
warming,
&slash_dom,
probe.round,
)
.await?;
drop(spawn_harvest(
&chain.data_dir,
Vec::new(),
targets.probed_samples.clone(),
));
return Ok(targets);
}
}
let bootstrap = discovery::bootstrap_nodes(
&chain.rpc_url,
capacity_bond,
&chain.data_dir,
args.discovery_cap(),
)
.await?;
let is_live_registry = matches!(bootstrap, discovery::Bootstrap::Live { .. });
if let Some(warning) = bootstrap.warning() {
tracing::warn!("{warning}");
}
let all = bootstrap.into_peers();
if all.is_empty() {
anyhow::bail!("no active nodes in the CapacityBond registry at {capacity_bond}");
}
let allow: &[decdn_protocol::Region] = if is_live_registry {
chain.region_allowlist.as_slice()
} else {
&[]
};
let selected = select_with_widening(all, chain.region.as_deref(), allow, &store_cfg);
let warming = ProxyWarmingParams::from_args(args);
let slash_dom = slash_judge_domain(chain.chain_id, chain.slash_judge);
mark_probe_start();
let targets = probe_and_order(
endpoint,
&selected,
relay_hint,
hash,
warming,
&slash_dom,
probe.round,
)
.await?;
drop(spawn_harvest(
&chain.data_dir,
Vec::new(),
targets.probed_samples.clone(),
));
Ok(targets)
}
fn store_fast_path(
store: &decdn_client::PeerStore,
cfg: &decdn_client::StoreConfig,
max_sources: usize,
now_secs: u64,
) -> Option<ResolvedTargets> {
let mut fresh: Vec<_> = store
.load_all()
.into_iter()
.filter(|r| r.selectable(now_secs, cfg) && !r.identity_prunable(now_secs, cfg))
.collect();
if fresh.len() < cfg.min_fresh_candidates {
return None;
}
fresh.sort_by(|a, b| {
a.latency_ms
.unwrap_or(f64::MAX)
.total_cmp(&b.latency_ms.unwrap_or(f64::MAX))
});
let ordered: Vec<NodeCandidate> = fresh
.iter()
.map(decdn_client::PeerRecord::as_candidate)
.collect();
let candidates = discovery::admit_sources(ordered, max_sources);
if candidates.len() < cfg.min_fresh_candidates {
return None;
}
Some(ResolvedTargets {
candidates,
coverage_by_node: HashMap::new(),
probed_samples: Vec::new(),
pinned: false,
size_hint: None,
late: LateSlot::default(),
})
}
pub(crate) fn identity_fresh_candidates(
store: &decdn_client::PeerStore,
cfg: &decdn_client::StoreConfig,
now_secs: u64,
) -> Vec<NodeCandidate> {
store
.load_all()
.into_iter()
.filter(|r| {
r.identity_fresh(now_secs, cfg)
&& !r.identity_prunable(now_secs, cfg)
&& !r.failure_suppressed(now_secs, cfg)
})
.map(|r| r.as_candidate())
.collect()
}
fn cached_pool_token(
store: &RedbBuyerPoolStore,
self_address: Address,
deployment: Deployment,
) -> anyhow::Result<Option<Address>> {
Ok(store
.get_by_owner(self_address)?
.filter(|state| state.is_on(deployment))
.map(|state| state.token))
}
fn select_with_widening(
candidates: Vec<NodeCandidate>,
region: Option<&str>,
allow: &[decdn_protocol::Region],
cfg: &decdn_client::StoreConfig,
) -> Vec<NodeCandidate> {
let widen_fallback = (!allow.is_empty()).then(|| candidates.clone());
let selected =
discovery::select_candidates_filtered(candidates, region, discovery::SELECT_K, allow);
if let Some(widened) = widen_fallback
&& selected.len() < cfg.min_fresh_candidates
{
return discovery::select_candidates_filtered(widened, region, discovery::SELECT_K, &[]);
}
selected
}
pub(crate) fn spawn_harvest(
data_dir: &Path,
registry: Vec<NodeCandidate>,
probed: Vec<(PublicKey, f64, u64)>,
) -> tokio::task::JoinHandle<()> {
let dir = data_dir.to_path_buf();
tokio::spawn(async move {
let store = decdn_client::PeerStore::open(&dir);
let cfg = decdn_client::StoreConfig::default();
harvest(&store, ®istry, probed, now_secs_cli(), &cfg);
})
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct HarvestTally {
writes: usize,
failed: usize,
}
fn harvest(
store: &decdn_client::PeerStore,
registry: &[NodeCandidate],
probed: Vec<(PublicKey, f64, u64)>,
now: u64,
cfg: &decdn_client::StoreConfig,
) -> HarvestTally {
let mut tally = HarvestTally {
writes: 0,
failed: 0,
};
let mut file = |node_id: &PublicKey, what: &str, written: anyhow::Result<()>| {
tally.writes = tally.writes.saturating_add(1);
if let Err(e) = written {
tally.failed = tally.failed.saturating_add(1);
tracing::debug!(%node_id, "could not record {what} in the peer store: {e:#}");
}
};
for cand in registry {
file(
&cand.node_id,
"an identity",
store.upsert_identity(cand, now),
);
}
for (node_id, rtt_ms, rate) in probed {
let written = store.record_sample(&node_id, rtt_ms, rate, now, cfg);
file(&node_id, "a probe sample", written);
}
if let Err(e) = store.prune_and_cap(now, cfg) {
tracing::debug!("could not prune the peer store: {e:#}");
}
if tally.writes > 0 && tally.failed == tally.writes {
tracing::warn!(
writes = tally.writes,
"the peer store refused every write of this probe round; \
its latency ranking stays at its last state"
);
}
tally
}
pub(crate) fn now_secs_cli() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_or(0, |d| d.as_secs())
}
#[derive(Debug)]
pub(crate) struct ResolveConfigFault(pub(crate) String);
impl std::fmt::Display for ResolveConfigFault {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.0)
}
}
impl std::error::Error for ResolveConfigFault {}
pub(crate) fn holders_or_none(
resolved: anyhow::Result<ResolvedTargets>,
) -> anyhow::Result<ResolvedTargets> {
match resolved {
Ok(targets) => Ok(targets),
Err(err) if err.downcast_ref::<ResolveConfigFault>().is_some() => Err(err),
Err(err) => {
tracing::warn!(
error = %decdn_common::redact::sanitize_err_chain(&err),
"no holder resolved yet; the fetch keeps looking"
);
Ok(ResolvedTargets {
candidates: Vec::new(),
coverage_by_node: HashMap::new(),
probed_samples: Vec::new(),
pinned: false,
size_hint: None,
late: LateSlot::default(),
})
}
}
}
pub(crate) struct ResolvedTargets {
pub(crate) candidates: Vec<NodeCandidate>,
pub(crate) coverage_by_node: HashMap<PublicKey, decdn_protocol::Coverage>,
pub(crate) probed_samples: Vec<(PublicKey, f64, u64)>,
pub(crate) pinned: bool,
pub(crate) size_hint: Option<u64>,
pub(crate) late: LateSlot,
}
fn pinned_targets(node_id: PublicKey, provider: Address) -> ResolvedTargets {
ResolvedTargets {
candidates: vec![NodeCandidate {
node_id,
eth_address: provider,
region_hint: None,
multiaddrs: Bytes::new(),
}],
coverage_by_node: HashMap::new(),
probed_samples: Vec::new(),
pinned: true,
size_hint: None,
late: LateSlot::default(),
}
}
pub(crate) async fn resolve_target_node(
args: &cli::ClientFetchArgs,
chain: &ResolvedChain,
endpoint: &Endpoint,
relays: &[RelayUrl],
hash: [u8; 32],
probe: ProbeOpts<'_>,
) -> anyhow::Result<ResolvedTargets> {
if let Some(raw) = &args.node_id {
let config = |e: anyhow::Error| anyhow::Error::new(ResolveConfigFault(format!("{e:#}")));
let node_id = PublicKey::from_str(raw)
.map_err(|e| config(anyhow::anyhow!("invalid --node-id {raw:?}: {e}")))?;
let provider_raw = args.provider_address.as_deref().ok_or_else(|| {
config(anyhow::anyhow!(
"--provider-address is required with --node-id"
))
})?;
let provider =
super::chain_ctx::parse_address(provider_raw, "--provider-address").map_err(config)?;
return Ok(pinned_targets(node_id, provider));
}
let capacity_bond = chain.capacity_bond.ok_or_else(|| {
anyhow::Error::new(ResolveConfigFault(
"auto-discovery needs capacity_bond_address (--capacity-bond-address or \
blockchain.capacity_bond_address), or pass --node-id to dial directly"
.to_string(),
))
})?;
let order = discover_provider(
endpoint,
chain,
capacity_bond,
relays.first(),
hash,
args,
probe,
)
.await?;
if !order.candidates.is_empty() {
tracing::info!("discovered {} node(s):", order.candidates.len());
for c in &order.candidates {
tracing::info!(
" * {} / {} / {}",
c.region_hint.as_ref().map_or("??", |r| r.as_str()),
node_id_tail(&c.node_id),
short_address(&c.eth_address),
);
}
}
Ok(order)
}
fn node_id_tail(node_id: &PublicKey) -> String {
let s = node_id.to_string();
s.get(s.len().saturating_sub(4)..).unwrap_or(&s).to_owned()
}
fn short_address(addr: &Address) -> String {
let s = addr.to_string();
match (s.get(..6), s.get(s.len().saturating_sub(4)..)) {
(Some(head), Some(tail)) if s.len() > 10 => format!("{head}...{tail}"),
_ => s,
}
}
pub(crate) fn annotate_unbound_cache_miss(err: anyhow::Error, ctx: &PoolContext) -> anyhow::Error {
let refused = err
.downcast_ref::<UpstreamRefused>()
.filter(|refused| matches!(refused.error(), StreamError::NotFound));
if refused.is_none() && err.downcast_ref::<NoSourceHasBlob>().is_none() {
return err;
}
let mut causes: Vec<String> = Vec::new();
if let Some(quoted_rate) = refused
.and_then(UpstreamRefused::evidence)
.map(|resp| resp.body.rate_per_mb)
{
let headroom = ctx.deposit.saturating_sub(ctx.prior_amount);
let estimate = min_payment(decdn_protocol::client::CHUNK_BYTES, quoted_rate);
if quoted_rate > 0 && headroom < estimate {
causes.push(format!(
"this pool's remaining deposit ({headroom}) is below the ~{estimate} the node \
reserves before serving at its quoted rate of {quoted_rate} per MB. The fetch \
path already auto-refills below a low-water mark, so reaching this means the \
configured working deposit is itself too small: raise \
`--working-deposit-micro-usdc` (or `blockchain.buyer_working_deposit_micro_usdc`) \
and retry"
));
}
}
if ctx.client_binding.is_none() {
causes.push(
"no client identity binding was sent because \
blockchain.capacity_bond_address is unset, so the node could not \
reactively pull this cache-missed blob from its origin; set \
blockchain.capacity_bond_address to enable reactive pull-through"
.to_string(),
);
}
match causes.len() {
0 => err,
1 => err.context(causes.concat()),
n => err.context(format!(
"{n} possible causes: {}",
causes
.iter()
.enumerate()
.map(|(i, c)| format!("({}) {c}", i + 1))
.collect::<Vec<_>>()
.join("; ")
)),
}
}
pub(crate) struct SettleOnDrop<F: FnOnce()> {
settle: Option<F>,
}
impl<F: FnOnce()> SettleOnDrop<F> {
pub(crate) const fn new(settle: F) -> Self {
Self {
settle: Some(settle),
}
}
pub(crate) fn disarm(mut self) {
self.settle = None;
}
}
impl<F: FnOnce()> Drop for SettleOnDrop<F> {
fn drop(&mut self) {
if let Some(settle) = self.settle.take() {
settle();
}
}
}
fn persist_watermark(
store: &RedbBuyerPoolStore,
owner: Address,
pool_id: PoolId,
lane: LaneKey,
progress: &VoucherProgress,
) {
let Some(write) = ProgressWrite::of(progress) else {
return;
};
let label = match write {
ProgressWrite::Rebase { .. } => "rebased",
ProgressWrite::Advance { .. } => "advanced",
_ => "recorded",
};
let outcome = write.apply(store, owner, pool_id, lane);
let (bytes_delivered, amount) = progress.totals();
match outcome {
Ok(AdvanceOutcome::Advanced) => {}
Ok(other) => tracing::warn!(
write = label,
%bytes_delivered,
%amount,
"{label} voucher watermark not persisted for pool {pool_id} (provider {}): \
{other:?}; the next reuse may re-sign a stale watermark, which that provider \
rejects",
lane.provider
),
Err(e) => tracing::warn!(
write = label,
%bytes_delivered,
%amount,
"failed to persist {label} voucher watermark for pool {pool_id} (provider {}): \
{e}; the next reuse may re-sign a stale watermark, which that provider rejects",
lane.provider
),
}
}
pub async fn fetch(args: &cli::FetchArgs, config_path: Option<&Path>) -> anyhow::Result<()> {
let timings = FetchTimings::start();
let hash = parse_hash(&args.hash)?;
let common = &args.common;
common.validate()?;
let relays = client_endpoint::resolve_relays(common.relay_url.as_deref(), config_path)?;
let disc = client_endpoint::client_discovery(config_path)?;
let file = load_file_config(config_path)?;
let mut chain = resolve_chain(common, &file)?;
let grant = resolve_delegation_grant(common, &mut chain)?;
let store = Arc::new(open_client_store_for_buy(
&chain.data_dir,
chain.data_dir_source,
"fetch",
)?);
let endpoint = client_endpoint::client_endpoint(&relays, &disc).await?;
timings.mark(Mark::Endpoint);
let mut interrupt = Interrupt::watch();
let writes = OrderedWrites::default();
let result = tokio::select! {
result = fetch_over(args, hash, &relays, &chain, grant, &store, &endpoint, &writes, &timings) => {
result
}
() = interrupt.wait() => Err(Interrupted.into()),
};
let output = if wants_stdout(&args.output) {
"stdout"
} else {
"file"
};
timings.log(output, result.is_ok());
writes.settle().await;
endpoint.close().await;
result
}
pub(crate) const STALL_WINDOW: Duration = Duration::from_secs(30);
#[allow(clippy::too_many_lines, clippy::too_many_arguments)]
async fn fetch_over(
args: &cli::FetchArgs,
hash: [u8; 32],
relays: &[RelayUrl],
chain: &ResolvedChain,
grant: Option<CapabilityGrant>,
store: &Arc<RedbBuyerPoolStore>,
endpoint: &Endpoint,
writes: &OrderedWrites,
timings: &FetchTimings,
) -> anyhow::Result<()> {
let common = &args.common;
let probe = ProbeOpts {
timings: Some(timings),
round: ProbeRound::Stream,
};
let resolved =
holders_or_none(resolve_target_node(common, chain, endpoint, relays, hash, probe).await)?;
timings.mark(Mark::Resolved);
let password = super::chain_ctx::read_keystore_password(
&super::chain_ctx::password_sources(
chain.keystore_password_file.as_deref(),
PasswordUse::Unlock,
),
"eth keystore password",
)?
.into_secret();
let signer = Arc::new(load_signer(&chain.keystore, &password)?);
timings.mark(Mark::Unlocked);
let self_address = signer.address();
let rpc = provider::build_provider(&chain.rpc_url, &signer)?;
let contract = PaymentPool::new(chain.payment_pool, rpc.clone());
let voucher_dom = voucher_domain(chain.chain_id, chain.payment_pool);
let slash_dom = slash_judge_domain(chain.chain_id, chain.slash_judge);
let token = match cached_pool_token(store, self_address, chain.deployment())? {
Some(token) => token,
None => contract
.usdc()
.call()
.await
.map_err(|e| anyhow::Error::new(e).context("read PaymentPool.usdc()"))?,
};
let max_blob_bytes = common.max_blob_mb.saturating_mul(1024 * 1024);
let namespace_id = args
.namespace
.map_or(decdn_protocol::client::NO_NAMESPACE, |n| {
alloy::primitives::U256::from(n).to_be_bytes()
});
let deadlines = PullDeadlines::new(STALL_WINDOW, STALL_WINDOW, 0)?;
let connections = Connections::new(endpoint.clone());
let clock = Arc::new(ProgressClock::new());
let stop = StopPolicy::new(
std::io::IsTerminal::is_terminal(&std::io::stderr()),
common.give_up_after(),
clock,
);
let health = Arc::new(PeerHealth::default());
let recovery = Arc::new(RecoveryGate::new());
let mut replaced: Option<PoolReplaced> = None;
loop {
let funding = RunFunding::default();
let deps = DriveFetchDeps {
endpoint,
store,
contract: &contract,
rpc: &rpc,
slash_dom: &slash_dom,
self_address,
token,
chain,
namespace_id,
max_rate_per_mb: common.max_rate_per_mb,
max_blob_bytes,
deadlines,
connections: &connections,
writes,
funding: &funding,
timings: Some(timings),
};
let open_lock = tokio::sync::Mutex::new(());
let sources = CliSources::new(
&deps,
common,
relays,
grant.as_ref(),
&signer,
&voucher_dom,
Some(&open_lock),
None,
None,
&recovery,
);
let holders = sources.holders_from(&resolved);
if replaced.is_none() {
timings.set_holders_start(holders.len());
}
let on_drop = SettleOnDrop::new(|| sources.persist_watermarks_detached());
let result = async {
let claim = sources.first_claim(hash, holders, &health, &stop).await?;
if wants_stdout(&args.output) {
return stream_to_stdout(
&deps,
&sources,
claim.holders,
Arc::clone(&health),
hash,
claim.total_bytes,
stop.clone(),
common.max_sources,
)
.await;
}
download_to_file(
&deps,
&sources,
claim.holders,
Arc::clone(&health),
DownloadTarget {
hash,
total_bytes: claim.total_bytes,
dest: &args.output,
ranges: None,
},
&stop,
common.max_sources,
)
.await
}
.await;
on_drop.disarm();
sources.persist_watermarks().await;
if let Some(shortfall) = funding.shortfall() {
eprintln!("warning: {shortfall}");
}
let result = result.map_err(|err| sources.annotate(err));
match next_pass(result, &mut replaced, wants_stdout(&args.output)) {
Some(pass_result) => return pass_result,
None => recovery.start_next_pass(),
}
}
}
fn next_pass(
result: anyhow::Result<()>,
replaced: &mut Option<PoolReplaced>,
to_stdout: bool,
) -> Option<anyhow::Result<()>> {
let now_replaced = result
.as_ref()
.err()
.and_then(|err| err.downcast_ref::<PoolReplaced>().copied());
if let Some(new) = now_replaced
&& replaced.is_none()
&& !to_stdout
{
eprintln!("note: {new}");
*replaced = Some(new);
return None;
}
if let Some(earlier) = (*replaced).or(now_replaced) {
eprintln!(
"note: pool {} accepts no more funds and may still hold a balance; once its \
dispute window ends, reclaim it with: decdn pool reclaim --pool {}",
earlier.closed, earlier.closed
);
}
Some(result.map_err(|err| match (now_replaced, *replaced) {
(Some(_), Some(_)) => err.context(
"funding needed: the pool this command opened accepts no more funds either; run \
the command again",
),
(Some(new), None) => err.context(format!(
"pool {} is now the current pool; run the command again to stream from it",
new.opened
)),
_ => err,
}))
}
pub(crate) fn annotate_delegated_exhaustion(err: anyhow::Error) -> anyhow::Error {
use decdn_protocol::client::VoucherRejectReason;
let funding_rejection = err
.downcast_ref::<UpstreamVoucherRejected>()
.is_some_and(|rejected| {
matches!(
rejected.reason,
VoucherRejectReason::SpendingCapExhausted
| VoucherRejectReason::CapabilityExpired
| VoucherRejectReason::PoolExhausted
| VoucherRejectReason::SignerCapExhausted
)
});
let underpaid = err
.downcast_ref::<UpstreamVoucherRejected>()
.is_some_and(|rejected| rejected.reason == VoucherRejectReason::Underpaid);
if let Some(&decdn_client::FundingNeeded::PublisherPool { pool }) = err.downcast_ref() {
return err.context(format!(
"ask the owner of pool {pool} to top it up: the capability still covers the fetch, \
and a delegated client cannot top up a pool it does not own. A node refuses a \
signer whose registered terms are spent the same way; once this signer key is \
registered on-chain its terms are write-once, so the owner then issues a \
capability for a new signer key"
));
}
if funding_rejection
|| err.downcast_ref::<NoAffordableSource>().is_some()
|| err.downcast_ref::<decdn_client::FundingNeeded>().is_some()
{
err.context(
"funding needed: no provider serves this capability at its current funding. Ask \
the pool owner to top up the pool or issue a fresh capability; once this signer \
key is registered on-chain its terms are write-once, so the fresh capability must \
name a new signer key (a delegated client cannot top up a pool it does not own)",
)
} else if underpaid {
err.context(
"the node kept refusing this lane's vouchers as underpaid after the resync attempts \
ran out — retry the fetch; a lane with no accepted voucher yet means the signed \
price is below the node's quote",
)
} else {
err
}
}
pub(crate) struct DriveFetchDeps<'a, P> {
pub(crate) endpoint: &'a Endpoint,
pub(crate) store: &'a Arc<RedbBuyerPoolStore>,
pub(crate) contract: &'a PaymentPool::PaymentPoolInstance<P>,
pub(crate) rpc: &'a P,
pub(crate) slash_dom: &'a Eip712Domain,
pub(crate) self_address: Address,
pub(crate) token: Address,
pub(crate) chain: &'a ResolvedChain,
pub(crate) namespace_id: [u8; 32],
pub(crate) max_rate_per_mb: u64,
pub(crate) max_blob_bytes: u64,
pub(crate) deadlines: PullDeadlines,
pub(crate) connections: &'a Connections,
pub(crate) writes: &'a OrderedWrites,
pub(crate) funding: &'a RunFunding,
pub(crate) timings: Option<&'a FetchTimings>,
}
fn lane_ledger(
ledgers: Option<&LaneLedgers>,
lane: LaneKey,
ctx: PoolContext,
) -> (Arc<PoolLedger>, Arc<Mutex<PoolContext>>) {
let seed = Cumulative {
bytes: ctx.prior_bytes_delivered,
amount: ctx.prior_amount,
};
let fresh_deposit = ctx.deposit;
match ledgers {
Some(reg) => {
let h = reg.get_or_insert(lane, || LaneHandle {
ledger: Arc::new(PoolLedger::new(seed)),
ctx: Arc::new(Mutex::new(ctx)),
});
{
let mut g = h.ctx.lock().unwrap_or_else(PoisonError::into_inner);
g.deposit = g.deposit.max(fresh_deposit);
}
(h.ledger, h.ctx)
}
None => (Arc::new(PoolLedger::new(seed)), Arc::new(Mutex::new(ctx))),
}
}
pub(crate) struct MultiLane<'a> {
pub(crate) provider: Address,
pub(crate) pool_id: PoolId,
pub(crate) prior_amount: U256,
pub(crate) ctx: Arc<Mutex<PoolContext>>,
pub(crate) ledger: Arc<PoolLedger>,
pub(crate) source: PeerSource<'a>,
}
fn probe_target(cand: &NodeCandidate, relay: Option<RelayUrl>) -> EndpointAddr {
let mut target = EndpointAddr::new(cand.node_id);
if let Some(url) = relay {
target = target.with_relay_url(url);
}
discovery::with_dial_addrs(target, cand)
}
pub(crate) fn lane_target(
candidate: &NodeCandidate,
addr: Option<std::net::SocketAddr>,
relays: &[RelayUrl],
) -> EndpointAddr {
let mut target = EndpointAddr::new(candidate.node_id);
if let Some(addr) = addr {
target = target.with_ip_addr(addr);
}
if let Some(url) = relays.first() {
target = target.with_relay_url(url.clone());
}
discovery::with_dial_addrs(target, candidate)
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn build_multi_lane<'a, P>(
deps: &DriveFetchDeps<'a, P>,
grant: Option<&CapabilityGrant>,
signer: &Arc<PrivateKeySigner>,
voucher_dom: &Eip712Domain,
provider: Address,
target: EndpointAddr,
open_lock: Option<&tokio::sync::Mutex<()>>,
ledgers: Option<&LaneLedgers>,
) -> anyhow::Result<MultiLane<'a>>
where
P: alloy::providers::Provider + Clone,
{
let (ctx, spend) = {
let _open_guard = match (open_lock, grant) {
(Some(lock), None) => Some(lock.lock().await),
_ => None,
};
build_ctx_for_fetch(
grant,
deps.store,
deps.contract,
deps.rpc,
signer,
voucher_dom,
provider,
deps.self_address,
deps.chain,
deps.endpoint,
deps.funding,
)
.await?
};
let pool_id = ctx.pool_id;
let prior_amount = ctx.prior_amount;
let lane = LaneKey {
pool_id,
signer: deps.self_address,
provider,
};
let (ledger, ctx) = lane_ledger(ledgers, lane, ctx);
if let Some(spend) = spend {
deps.funding.join_lane(lane, spend, &ledger);
}
let source = lane_source(
deps,
target,
Arc::clone(&ctx),
Arc::clone(&ledger),
provider,
);
if let Some(timings) = deps.timings {
timings.mark(Mark::Lane);
}
Ok(MultiLane {
provider,
pool_id,
prior_amount,
ctx,
ledger,
source,
})
}
fn lane_source<'a, P>(
deps: &DriveFetchDeps<'a, P>,
target: EndpointAddr,
ctx: Arc<Mutex<PoolContext>>,
ledger: Arc<PoolLedger>,
provider: Address,
) -> PeerSource<'a> {
PeerSource::new(
deps.endpoint,
target,
ctx,
ledger,
deps.slash_dom,
provider,
deps.namespace_id,
deps.max_blob_bytes,
deps.max_rate_per_mb,
deps.deadlines,
Some(deps.connections.clone()),
)
}
pub(crate) struct FaceLaneHandle {
pub(crate) pool_id: PoolId,
pub(crate) provider: Address,
pub(crate) prior_amount: U256,
pub(crate) ledger: Arc<PoolLedger>,
pub(crate) ctx: Arc<Mutex<PoolContext>>,
}
fn stream_lane_watermarks(lanes: &[FaceLaneHandle]) -> Vec<LaneWatermark> {
let mut per_ledger: Vec<(&FaceLaneHandle, U256)> = Vec::new();
for l in lanes {
match per_ledger
.iter_mut()
.find(|(first, _)| Arc::ptr_eq(&first.ledger, &l.ledger))
{
Some((_, prior)) => *prior = (*prior).min(l.prior_amount),
None => per_ledger.push((l, l.prior_amount)),
}
}
per_ledger
.into_iter()
.map(|(l, prior_amount)| LaneWatermark {
pool_id: l.pool_id,
provider: l.provider,
prior_amount,
rebase_anchor: l.ledger.take_unsaved_rebase(),
settlement: l.ledger.settlement(),
})
.collect()
}
async fn copy_verified<R, W>(
reader: &mut R,
sink: &mut W,
total_bytes: u64,
on_progress: Option<&dyn Fn(u64, u64)>,
) -> anyhow::Result<u64>
where
R: tokio::io::AsyncRead + Unpin,
W: tokio::io::AsyncWrite + Unpin,
{
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let mut buf = vec![0u8; 256 * 1024];
let mut drained: u64 = 0;
loop {
let n = reader
.read(&mut buf)
.await
.map_err(|e| anyhow::anyhow!("read verified stream: {e}"))?;
if n == 0 {
break;
}
let chunk = buf
.get(..n)
.ok_or_else(|| anyhow::anyhow!("read returned more than the buffer holds"))?;
sink.write_all(chunk)
.await
.map_err(|e| anyhow::anyhow!("write to stdout: {e}"))?;
drained = drained.saturating_add(u64::try_from(n).unwrap_or(u64::MAX));
if let Some(cb) = on_progress {
cb(drained, total_bytes);
}
}
sink.flush()
.await
.map_err(|e| anyhow::anyhow!("flush stdout: {e}"))?;
Ok(drained)
}
type LaneWrite = (LaneKey, VoucherProgress);
fn face_watermark_writes(self_address: Address, handles: &[FaceLaneHandle]) -> Vec<LaneWrite> {
multi_lane_watermarks(self_address, &stream_lane_watermarks(handles))
}
fn write_watermarks(store: &RedbBuyerPoolStore, self_address: Address, writes: Vec<LaneWrite>) {
for (lane, vprogress) in writes {
persist_watermark(store, self_address, lane.pool_id, lane, &vprogress);
}
}
pub(crate) fn queue_face_watermarks(
writes: &OrderedWrites,
store: &Arc<RedbBuyerPoolStore>,
self_address: Address,
handles: &[FaceLaneHandle],
) -> tokio::sync::oneshot::Receiver<()> {
queue_watermark_writes(
writes,
store,
self_address,
face_watermark_writes(self_address, handles),
)
}
fn queue_watermark_writes(
writes: &OrderedWrites,
store: &Arc<RedbBuyerPoolStore>,
self_address: Address,
lane_writes: Vec<LaneWrite>,
) -> tokio::sync::oneshot::Receiver<()> {
let store = Arc::clone(store);
writes.queue_awaitable(move || write_watermarks(&store, self_address, lane_writes))
}
#[allow(clippy::too_many_arguments)]
async fn stream_to_stdout<P>(
deps: &DriveFetchDeps<'_, P>,
sources: &CliSources<'_, P>,
holders: Vec<Holder>,
health: Arc<PeerHealth>,
hash: [u8; 32],
total_bytes: u64,
stop: StopPolicy,
max_sources: usize,
) -> anyhow::Result<()>
where
P: alloy::providers::Provider + Clone,
{
let pull_config = PullConfig {
streamer_lane_cap: max_sources.max(1),
..PullConfig::new()
};
let funder = sources.funder();
let scratch = tempfile::tempdir_in(&deps.chain.data_dir)
.map_err(|e| anyhow::anyhow!("open stream scratch dir: {e}"))?;
let mut streamer = Streamer::new(sources, funder, scratch.path())
.holders(holders)
.health(health)
.recovery(Arc::clone(sources.recovery()))
.max_blob_bytes(deps.max_blob_bytes);
if let Some(slot) = sources.credentials() {
streamer = streamer.credentials(slot);
}
let (mut reader, mut drive) = streamer
.open(hash, total_bytes, &pull_config, Arc::new(NoCache), stop)
.await?;
let bar = std::io::IsTerminal::is_terminal(&std::io::stderr()).then(delivery_progress);
let on_progress = |received: u64, expected: u64| {
if received > 0
&& let Some(timings) = deps.timings
{
timings.mark(Mark::FirstByte);
}
if let Some((_, cb, _)) = &bar {
cb(received, expected);
}
};
let mut stdout = tokio::io::stdout();
let copy_result = drive
.alongside(copy_verified(
&mut reader,
&mut stdout,
total_bytes,
Some(&on_progress),
))
.await;
if let Some((bar, _, _)) = &bar {
bar.finish_and_clear();
}
let copied = copy_result.map_err(|read_err| stream_error(read_err, drive.take_error()))?;
tracing::info!("streamed {copied} verified bytes to stdout");
Ok(())
}
fn stream_error(read_err: anyhow::Error, fetch_err: Option<anyhow::Error>) -> anyhow::Error {
fetch_err.unwrap_or(read_err)
}
async fn download_to_file<P>(
deps: &DriveFetchDeps<'_, P>,
sources: &CliSources<'_, P>,
holders: Vec<Holder>,
health: Arc<PeerHealth>,
target: DownloadTarget<'_>,
stop: &StopPolicy,
max_sources: usize,
) -> anyhow::Result<()>
where
P: alloy::providers::Provider + Clone,
{
let (bar, on_bar, meter) = delivery_progress();
let on_progress = |received: u64, expected: u64| {
if received > 0
&& let Some(timings) = deps.timings
{
timings.mark(Mark::FirstByte);
}
on_bar(received, expected);
};
let mut downloader = Downloader::new(sources, sources.funder())
.holders(holders)
.health(health)
.recovery(Arc::clone(sources.recovery()))
.max_lanes(max_sources)
.max_blob_bytes(deps.max_blob_bytes);
if let Some(slot) = sources.credentials() {
downloader = downloader.credentials(slot);
}
let result =
Box::pin(downloader.fetch_to_paths_until(&[target], Some(&on_progress), stop)).await;
bar.finish_and_clear();
result?;
let proven = tokio::fs::metadata(target.dest)
.await
.map_or(target.total_bytes, |m| m.len());
print_fetch_summary(proven, &meter, target.dest);
Ok(())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct LaneWatermark {
pool_id: PoolId,
provider: Address,
prior_amount: U256,
settlement: Cumulative,
rebase_anchor: Option<Cumulative>,
}
fn multi_lane_watermarks(
signer: Address,
lanes: &[LaneWatermark],
) -> Vec<(LaneKey, VoucherProgress)> {
lanes
.iter()
.map(|l| {
(
LaneKey {
pool_id: l.pool_id,
signer,
provider: l.provider,
},
VoucherProgress::from_cumulative(l.settlement, l.prior_amount)
.with_rebase_anchor(l.rebase_anchor),
)
})
.collect()
}
#[derive(Debug, Default)]
pub(crate) struct RunFunding {
spend: Mutex<RunSpend>,
shortfall: Mutex<Option<String>>,
pool_id: OnceLock<PoolId>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct LaneSpend {
pub(crate) outside: U256,
pub(crate) prior: U256,
pub(crate) recorded: bool,
pub(crate) from_chain: bool,
}
#[derive(Debug, Default)]
struct RunSpend {
baseline: Option<U256>,
from_chain: bool,
lanes: HashMap<LaneKey, Vec<Arc<PoolLedger>>>,
}
impl RunFunding {
pub(crate) fn note_pool(&self, pool_id: PoolId) {
let _ = self.pool_id.set(pool_id);
}
pub(crate) fn pool_id(&self) -> Option<PoolId> {
self.pool_id.get().copied()
}
pub(crate) fn join_lane(&self, lane: LaneKey, spend: LaneSpend, ledger: &Arc<PoolLedger>) {
let mut run = self.spend.lock().unwrap_or_else(PoisonError::into_inner);
if let Some(ledgers) = run.lanes.get_mut(&lane) {
if !ledgers.iter().any(|known| Arc::ptr_eq(known, ledger)) {
ledgers.push(Arc::clone(ledger));
}
return;
}
run.lanes.insert(lane, vec![Arc::clone(ledger)]);
run.baseline = Some(match run.baseline {
None => {
run.from_chain = spend.from_chain;
spend.outside
}
Some(baseline) if spend.recorded || run.from_chain => {
baseline.saturating_sub(spend.prior)
}
Some(baseline) => baseline,
});
}
pub(crate) fn pool_spent(&self) -> Option<U256> {
let run = self.spend.lock().unwrap_or_else(PoisonError::into_inner);
let baseline = run.baseline?;
Some(run.lanes.values().fold(baseline, |total, ledgers| {
let lane = ledgers
.iter()
.map(|ledger| ledger.committed().amount)
.max()
.unwrap_or(U256::ZERO);
total.saturating_add(lane)
}))
}
pub(crate) fn shortfall(&self) -> Option<String> {
self.shortfall
.lock()
.unwrap_or_else(PoisonError::into_inner)
.clone()
}
pub(crate) async fn check_wallet<P>(
&self,
rpc: &P,
token: Address,
owner: Address,
pool_id: PoolId,
additional: U256,
err: &anyhow::Error,
) -> Option<String>
where
P: alloy::providers::Provider + Clone,
{
let error = decdn_common::redact::sanitize_err_chain(err);
let balance = match decdn_incentive::Erc20::new(token, rpc.clone())
.balanceOf(owner)
.call()
.await
{
Ok(balance) => balance,
Err(read) => {
let read = decdn_common::redact::sanitize_err_chain(&anyhow::Error::new(read));
tracing::warn!(
%pool_id,
%additional,
%error,
wallet_usdc_error = %read,
"buyer pool top-up failed, and the wallet's USDC balance could not be read"
);
return None;
}
};
if balance >= additional {
tracing::warn!(
%pool_id,
%additional,
wallet_usdc = %balance,
%error,
"buyer pool top-up failed although the wallet holds enough USDC"
);
return None;
}
let warning = format!(
"wallet {owner} holds {balance} µUSDC, less than the {additional} µUSDC top-up \
buyer pool {pool_id} needs; fund the wallet so the pool can be topped up"
);
let mut shortfall = self
.shortfall
.lock()
.unwrap_or_else(PoisonError::into_inner);
if shortfall.is_none() {
tracing::warn!(
%pool_id,
%additional,
wallet_usdc = %balance,
%error,
"the wallet holds too little USDC to top up the buyer pool"
);
*shortfall = Some(warning.clone());
}
Some(warning)
}
}
pub(crate) struct CliFunder<'a, P> {
pub(crate) contract: &'a PaymentPool::PaymentPoolInstance<P>,
pub(crate) rpc: &'a P,
pub(crate) store: &'a RedbBuyerPoolStore,
pub(crate) owner: Address,
pub(crate) signer: &'a Arc<PrivateKeySigner>,
pub(crate) deployment: Deployment,
pub(crate) token: Address,
pub(crate) payment_pool_addr: Address,
pub(crate) working_deposit: U256,
pub(crate) max_approve: bool,
pub(crate) funding: &'a RunFunding,
}
impl<P> Funder for CliFunder<'_, P>
where
P: alloy::providers::Provider + Clone,
{
fn pool_spent(&self) -> Option<U256> {
self.funding.pool_spent()
}
fn recover(&self, remaining: U256) -> SourceFuture<'_, Recovery> {
Box::pin(async move {
if self.working_deposit.is_zero() {
return Ok(Recovery::Unavailable);
}
let pool_id = self
.funding
.pool_id()
.ok_or_else(|| anyhow::anyhow!("no payment lane is built, so no pool to top up"))?;
let additional = self.working_deposit.saturating_sub(remaining);
if additional.is_zero() {
return self.settle_or_replace(pool_id, remaining).await;
}
if let Some(shortfall) = self.funding.shortfall() {
return Err(
anyhow::anyhow!("the wallet cannot fund a top-up: {shortfall}")
.context(WalletShortfall),
);
}
let escrowed = async {
ensure_allowance(
self.rpc,
self.token,
self.owner,
self.payment_pool_addr,
if self.max_approve {
None
} else {
Some(additional)
},
)
.await?;
top_up(self.contract, self.owner, pool_id, additional).await
}
.await;
let ToppedUpPool { credited, tx, .. } = match escrowed {
Ok(topped_up) => topped_up,
Err(err) if err.downcast_ref::<TopUpUnconfirmed>().is_some() => return Err(err),
Err(err) if err.downcast_ref::<PoolNotOpen>().is_some() => {
tracing::info!(
%pool_id,
"the current buyer pool no longer accepts funds; opening a new pool"
);
let opened = self.open_replacement().await?;
return Ok(Recovery::Replaced(PoolReplaced {
closed: pool_id,
opened,
}));
}
Err(err) => {
let short = self
.funding
.check_wallet(self.rpc, self.token, self.owner, pool_id, additional, &err)
.await;
return Err(match short {
Some(_) => err.context(WalletShortfall),
None => err,
});
}
};
let effect = topped_up_effect(pool_id, credited);
let new_deposit = grade_deposit_credit(
self.store.add_deposit(self.owner, pool_id, credited),
&effect,
tx,
)?;
Ok(Recovery::ToppedUp(new_deposit))
})
}
}
impl<P> CliFunder<'_, P>
where
P: alloy::providers::Provider + Clone,
{
async fn settle_or_replace(
&self,
pool_id: PoolId,
remaining: U256,
) -> anyhow::Result<Recovery> {
if !pool_accepts_funds(self.contract, self.owner, pool_id).await {
tracing::info!(
%pool_id,
"the current buyer pool holds its working deposit but accepts no more funds; \
opening a new pool"
);
let opened = self.open_replacement().await?;
return Ok(Recovery::Replaced(PoolReplaced {
closed: pool_id,
opened,
}));
}
let estimate = || remaining.saturating_add(self.funding.pool_spent().unwrap_or(U256::ZERO));
let deposit = match self.store.get_by_pool_id(pool_id) {
Ok(Some(row)) => row.deposit,
Ok(None) => estimate(),
Err(error) => {
tracing::warn!(
%pool_id,
%error,
"could not read the buyer pool's row; estimating its deposit from the run"
);
estimate()
}
};
tracing::info!(
%pool_id,
%deposit,
"the buyer pool already holds its working deposit; waiting for the nodes to see it"
);
Ok(Recovery::ToppedUp(deposit))
}
async fn open_replacement(&self) -> anyhow::Result<PoolId> {
let deposit = self.working_deposit;
ensure_allowance(
self.rpc,
self.token,
self.owner,
self.payment_pool_addr,
if self.max_approve {
None
} else {
Some(deposit)
},
)
.await?;
let opened = open_pool(
self.contract,
Arc::clone(self.signer),
self.deployment,
self.token,
self.owner,
deposit,
)
.await?;
self.store
.record(&opened.state)
.map_err(|e| escrowed_but_untracked("buyer pool opened", opened.tx, e))?;
tracing::info!(pool_id = %opened.state.pool_id, %deposit, "opened a new buyer pool");
Ok(opened.state.pool_id)
}
}
fn wants_stdout(output: &Path) -> bool {
output == Path::new("-")
}
pub(crate) fn ranged_store_location(output: &Path) -> anyhow::Result<(PathBuf, String)> {
let dir = match output.parent() {
Some(p) if !p.as_os_str().is_empty() => p.to_path_buf(),
_ => PathBuf::from("."),
};
let stem = output
.file_name()
.and_then(|n| n.to_str())
.ok_or_else(|| anyhow::anyhow!("--output {} has no usable file name", output.display()))?
.to_string();
Ok((dir, stem))
}
const RATE_SMOOTHING_TAU_SECS: f64 = 3.0;
#[derive(Default)]
pub(crate) struct SpeedState {
started: Option<(Instant, u64)>,
last: Option<(Instant, u64)>,
ewma_bps: Option<f64>,
}
impl SpeedState {
pub(crate) fn observe(&mut self, now: Instant, received: u64) -> f64 {
self.started.get_or_insert((now, received));
if let Some((prev_at, prev_bytes)) = self.last {
let dt = now.saturating_duration_since(prev_at).as_secs_f64();
if dt > 0.0 {
let inst = bytes_as_f64(received.saturating_sub(prev_bytes)) / dt;
let alpha = 1.0 - (-dt / RATE_SMOOTHING_TAU_SECS).exp();
let prev_bps = self.ewma_bps.unwrap_or(0.0);
self.ewma_bps = Some(prev_bps + alpha * (inst - prev_bps));
}
}
self.last = Some((now, received));
self.ewma_bps.unwrap_or(0.0)
}
}
#[expect(
clippy::cast_precision_loss,
reason = "byte counts stay far below f64's 2^53 exact-integer ceiling"
)]
pub(crate) const fn bytes_as_f64(n: u64) -> f64 {
n as f64
}
pub(crate) fn fmt_rate(bps: f64) -> String {
if bps.is_finite() && bps >= 1.0 {
#[expect(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
reason = "bps is finite and >= 1.0 here; the cast cannot wrap or go negative"
)]
let whole = bps.min(bytes_as_f64(u64::MAX)) as u64;
format!("{}/s", indicatif::HumanBytes(whole))
} else {
"--".to_string()
}
}
pub(crate) fn fmt_eta(remaining_bytes: u64, bps: f64) -> String {
if bps >= 1.0 {
let secs = (bytes_as_f64(remaining_bytes) / bps).clamp(0.0, 8.64e7);
format!(
"ETA {}",
indicatif::HumanDuration(Duration::from_secs_f64(secs))
)
} else {
"ETA --".to_string()
}
}
#[derive(Clone)]
pub(crate) struct DeliveryMeter {
state: Arc<Mutex<SpeedState>>,
}
impl DeliveryMeter {
fn summary(&self) -> Option<(Duration, u64)> {
let state = self.state.lock().ok()?;
let (start_at, start_bytes) = state.started?;
let (last_at, last_bytes) = state.last?;
Some((
last_at.saturating_duration_since(start_at),
last_bytes.saturating_sub(start_bytes),
))
}
}
fn delivery_progress() -> (
indicatif::ProgressBar,
impl Fn(u64, u64) + 'static,
DeliveryMeter,
) {
let bar = new_progress_bar();
let (on_progress, meter) = bar_callback(bar.clone(), TabProgress::detect(true));
(bar, on_progress, meter)
}
fn bar_callback(
bar: indicatif::ProgressBar,
tab: Option<Arc<TabProgress>>,
) -> (impl Fn(u64, u64) + 'static, DeliveryMeter) {
let length = std::sync::atomic::AtomicU64::new(u64::MAX);
let state = Arc::new(Mutex::new(SpeedState::default()));
let cb_state = Arc::clone(&state);
let on_progress = move |received: u64, expected: u64| {
if length.swap(expected, std::sync::atomic::Ordering::Relaxed) != expected {
bar.set_length(expected);
}
bar.set_position(received);
if let Some(tab) = &tab {
tab.update(received, expected);
}
if let Ok(mut s) = cb_state.lock() {
let bps = s.observe(Instant::now(), received);
bar.set_message(format!(
"({}, {}) ",
fmt_rate(bps),
fmt_eta(expected.saturating_sub(received), bps)
));
}
};
(on_progress, DeliveryMeter { state })
}
fn print_fetch_summary(content_bytes: u64, meter: &DeliveryMeter, output: &Path) {
match meter.summary() {
Some((elapsed, moved_bytes)) if elapsed > Duration::ZERO && moved_bytes > 0 => {
let avg_bps = bytes_as_f64(moved_bytes) / elapsed.as_secs_f64();
println!(
"fetched {content_bytes} bytes in {} (avg {}) -> {}",
indicatif::HumanDuration(elapsed),
fmt_rate(avg_bps),
output.display()
);
}
_ => println!("fetched {content_bytes} bytes -> {}", output.display()),
}
}
pub(crate) fn resolve_delegation_grant(
common: &cli::ClientFetchArgs,
chain: &mut ResolvedChain,
) -> anyhow::Result<Option<CapabilityGrant>> {
let grant = common
.resolve_capability_token()?
.map(|token| CapabilityGrant::from_token(&token))
.transpose()
.map_err(|e| anyhow::anyhow!("invalid --capability token: {e}"))?;
if grant.is_some() {
chain.working_deposit = U256::ZERO;
}
Ok(grant)
}
#[allow(clippy::too_many_arguments)]
async fn build_ctx_for_fetch<P>(
grant: Option<&CapabilityGrant>,
store: &RedbBuyerPoolStore,
contract: &PaymentPool::PaymentPoolInstance<P>,
rpc: &P,
signer: &Arc<PrivateKeySigner>,
voucher_dom: &Eip712Domain,
provider: Address,
self_address: Address,
chain: &ResolvedChain,
endpoint: &Endpoint,
funding: &RunFunding,
) -> anyhow::Result<(PoolContext, Option<LaneSpend>)>
where
P: alloy::providers::Provider + Clone,
{
if let Some(grant) = grant {
build_delegated_pool_ctx(
store,
contract,
signer,
voucher_dom,
provider,
self_address,
chain,
endpoint,
grant,
)
.await
.map(|ctx| (ctx, None))
} else {
build_pool_ctx(
store,
contract,
rpc,
signer,
provider,
self_address,
chain,
endpoint,
funding,
)
.await
.map(|(ctx, spend)| (ctx, Some(spend)))
}
}
#[allow(clippy::too_many_arguments)]
async fn build_pool_ctx<P>(
store: &RedbBuyerPoolStore,
contract: &PaymentPool::PaymentPoolInstance<P>,
rpc: &P,
signer: &Arc<PrivateKeySigner>,
provider: Address,
self_address: Address,
chain: &ResolvedChain,
endpoint: &Endpoint,
funding: &RunFunding,
) -> anyhow::Result<(PoolContext, LaneSpend)>
where
P: alloy::providers::Provider + Clone,
{
let (ctx, spend) = open_or_reuse_pool(
store,
contract,
rpc,
funding,
signer,
provider,
self_address,
chain.deployment(),
chain.working_deposit,
chain.max_approve,
ChainAdoption::for_buy(&chain.data_dir, &chain.keystore)?,
)
.await?;
Ok((attach_client_binding(ctx, chain, endpoint, signer)?, spend))
}
fn ensure_delegate_signer(self_address: Address, authorized: Address) -> anyhow::Result<()> {
anyhow::ensure!(
self_address == authorized,
"this capability authorizes signer {authorized}, but the loaded key is {self_address} — \
load the delegate keystore the pool owner assigned (--keystore / \
blockchain.eth_keystore), or ask for a capability issued to {self_address}",
);
Ok(())
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn build_delegated_pool_ctx<P>(
store: &RedbBuyerPoolStore,
contract: &PaymentPool::PaymentPoolInstance<P>,
signer: &Arc<PrivateKeySigner>,
voucher_dom: &Eip712Domain,
provider: Address,
self_address: Address,
chain: &ResolvedChain,
endpoint: &Endpoint,
grant: &CapabilityGrant,
) -> anyhow::Result<PoolContext>
where
P: alloy::providers::Provider + Clone,
{
ensure_delegate_signer(self_address, grant.signer)?;
let signed_capability = grant.to_signed_capability().map_err(|e| {
anyhow::anyhow!("capability token carries a malformed owner signature: {e}")
})?;
let pool_id = grant.pool_id;
let pool = contract.getPool(pool_id).call().await.map_err(|e| {
anyhow::Error::new(e).context(format!("read on-chain pool {pool_id} for the capability"))
})?;
anyhow::ensure!(
pool.owner != Address::ZERO,
"pool {pool_id} named by the capability does not exist on this PaymentPool contract \
({}) — wrong --payment-pool-address/chain, or a stale token",
chain.payment_pool,
);
let lane = LaneKey {
pool_id,
signer: self_address,
provider,
};
let (prior_bytes, prior_amount) = store
.get_by_pool_id(pool_id)?
.and_then(|state| state.lane_progress(lane))
.map_or((U256::ZERO, U256::ZERO), |p| (p.last_bytes, p.last_amount));
let token = contract
.usdc()
.call()
.await
.map_err(|e| anyhow::Error::new(e).context("read PaymentPool.usdc()"))?;
let state = BuyerPoolState::new(
pool_id,
chain.deployment(),
pool.owner,
token,
U256::from(pool.deposit),
);
let ctx = PoolContext::for_pool(&state, Arc::clone(signer), voucher_dom.clone())
.with_provider(provider, prior_bytes, prior_amount)
.with_capability(signed_capability);
attach_client_binding(ctx, chain, endpoint, signer)
}
pub(crate) fn attach_client_binding(
ctx: PoolContext,
chain: &ResolvedChain,
endpoint: &Endpoint,
signer: &Arc<PrivateKeySigner>,
) -> anyhow::Result<PoolContext> {
let Some(capacity_bond) = chain.capacity_bond else {
return Ok(ctx);
};
let bind_dom = bind_node_id_domain(chain.chain_id, capacity_bond);
let own_node_id = B256::from(*endpoint.id().as_bytes());
Ok(ctx.with_client_binding(sign_client_binding(signer, own_node_id, &bind_dom)?))
}
async fn adopt_owned_pool<P>(
store: &RedbBuyerPoolStore,
contract: &PaymentPool::PaymentPoolInstance<P>,
owner: Address,
deployment: Deployment,
) -> anyhow::Result<Option<BuyerPoolState>>
where
P: alloy::providers::Provider + Clone,
{
let Some((pool_id, pool)) = newest_solvent_owned_pool(contract, owner).await.context(
"could not ask the chain whether this wallet already owns a pool, so refusing to \
open one blind — rerun, or check the RPC endpoint",
)?
else {
return Ok(None);
};
let token =
contract.usdc().call().await.with_context(|| {
format!("read PaymentPool.usdc() while adopting live pool {pool_id}")
})?;
let state = BuyerPoolState::adopt(
pool_id,
deployment,
owner,
token,
U256::from(pool.deposit),
U256::from(pool.totalRedeemed),
);
store.record(&state).with_context(|| {
format!(
"found live pool {pool_id} on chain but could not record it in the local buyer \
store; nothing was escrowed, so rerunning retries the adoption"
)
})?;
tracing::info!(
%pool_id,
deposit = pool.deposit,
redeemed = pool.totalRedeemed,
"adopted a live buyer pool this wallet already owns on chain, instead of opening a \
second one"
);
Ok(Some(state))
}
async fn lane_watermark<P>(
contract: &PaymentPool::PaymentPoolInstance<P>,
lane: LaneKey,
) -> anyhow::Result<(U256, U256)>
where
P: alloy::providers::Provider + Clone,
{
let onchain = contract
.getWatermark(lane.pool_id, lane.signer, lane.provider)
.call()
.await
.context(
"could not read this lane's on-chain watermark, so refusing to resume it from zero, \
where it would stream bytes the provider cannot cash",
)?;
Ok((
U256::from(onchain.bytesDelivered),
U256::from(onchain.amount),
))
}
async fn pool_to_reuse<P>(
store: &RedbBuyerPoolStore,
contract: &PaymentPool::PaymentPoolInstance<P>,
funding: &RunFunding,
self_address: Address,
deployment: Deployment,
adoption: ChainAdoption,
) -> anyhow::Result<Option<BuyerPoolState>>
where
P: alloy::providers::Provider + Clone,
{
let tracked = match store.get_by_owner(self_address)? {
Some(state) if state.is_on(deployment) => Some(state),
Some(state) => {
tracing::warn!(
pool_id = %state.pool_id,
foreign_payment_pool = %state.deployment.payment_pool,
foreign_chain_id = state.deployment.chain_id,
configured_payment_pool = %deployment.payment_pool,
configured_chain_id = deployment.chain_id,
"ignoring a tracked buyer pool from another PaymentPool deployment; its deposit, \
if any, is recoverable only against `foreign_payment_pool`"
);
None
}
None => None,
};
let tracked = match tracked {
Some(state) if funding.pool_id() != Some(state.pool_id) => {
let kept = still_open(contract, self_address, state).await;
if let Some(state) = &kept {
funding.note_pool(state.pool_id);
}
kept
}
other => other,
};
Ok(match (tracked, adoption) {
(Some(state), _) => Some(state),
(None, ChainAdoption::Refused) => None,
(None, ChainAdoption::Allowed) => {
adopt_owned_pool(store, contract, self_address, deployment).await?
}
})
}
async fn still_open<P>(
contract: &PaymentPool::PaymentPoolInstance<P>,
self_address: Address,
state: BuyerPoolState,
) -> Option<BuyerPoolState>
where
P: alloy::providers::Provider + Clone,
{
match contract.getPool(state.pool_id).call().await {
Ok(pool)
if pool.owner == self_address && !matches!(pool.status, PaymentPool::Status::Open) =>
{
tracing::info!(
pool_id = %state.pool_id,
status = super::pool::status_label(pool.status),
"the tracked buyer pool no longer accepts work; a new pool replaces it"
);
None
}
Ok(_) => Some(state),
Err(error) => {
tracing::warn!(
pool_id = %state.pool_id,
error = %error,
"could not read the tracked buyer pool's status; reusing it"
);
Some(state)
}
}
}
fn seed_lane(
store: &RedbBuyerPoolStore,
mut state: BuyerPoolState,
lane: LaneKey,
bytes: U256,
amount: U256,
) -> anyhow::Result<(BuyerPoolState, U256, U256)> {
if bytes.is_zero() && amount.is_zero() {
return Ok((state, bytes, amount));
}
let pool_id = state.pool_id;
let outcome = store
.seed_progress(state.owner, pool_id, lane, bytes, amount)
.with_context(|| {
format!(
"record the on-chain watermark of lane {} in pool {pool_id}",
lane.provider
)
})?;
match outcome {
AdvanceOutcome::Advanced => {
if let Err(err) = state.seed_lane(lane, bytes, amount) {
tracing::debug!(error = %err, "lane seed did not advance the in-memory row");
}
Ok((state, bytes, amount))
}
AdvanceOutcome::Regressed(_) => {
let row = store.get_by_pool_id(pool_id)?.ok_or_else(|| {
anyhow::anyhow!("buyer row for pool {pool_id} vanished during a lane seed")
})?;
let progress = row.lane_progress(lane).ok_or_else(|| {
anyhow::anyhow!(
"buyer row for pool {pool_id} lost lane {} during a seed",
lane.provider
)
})?;
Ok((row, progress.last_bytes, progress.last_amount))
}
AdvanceOutcome::UnknownPool | AdvanceOutcome::PoolMismatch => Err(anyhow::anyhow!(
"the buyer row for pool {pool_id} was forgotten or replaced while lane {} was built; \
the build retries from the store",
lane.provider
)),
}
}
fn cannot_pay(state: &BuyerPoolState, shortfall: &str) -> anyhow::Error {
anyhow::Error::new(NoAffordableSource {
deposit: state.deposit,
})
.context(format!(
"buyer pool {} has no unspent deposit, and the wallet cannot fund a top-up: \
{shortfall}",
state.pool_id
))
}
#[allow(clippy::too_many_arguments)]
async fn refill_if_low<P>(
store: &RedbBuyerPoolStore,
contract: &PaymentPool::PaymentPoolInstance<P>,
rpc: &P,
funding: &RunFunding,
state: BuyerPoolState,
self_address: Address,
spent: U256,
working_deposit: U256,
max_approve: bool,
) -> anyhow::Result<BuyerPoolState>
where
P: alloy::providers::Provider + Clone,
{
let low_water = working_deposit / U256::from(LOW_WATER_DIVISOR);
let additional = refill_amount(state.deposit, spent, working_deposit, low_water);
if additional.is_zero() {
return Ok(state);
}
let remaining = state.deposit.saturating_sub(spent);
if let Some(shortfall) = funding.shortfall() {
if remaining.is_zero() {
return Err(cannot_pay(&state, &shortfall));
}
return Ok(state);
}
tracing::info!(
"buyer pool {} low on deposit ({remaining} µUSDC remaining of {} deposited); \
topping up {additional} µUSDC",
state.pool_id,
state.deposit,
);
let escrowed = async {
ensure_allowance(
rpc,
state.token,
self_address,
*contract.address(),
if max_approve { None } else { Some(additional) },
)
.await?;
top_up(contract, self_address, state.pool_id, additional).await
}
.await;
let err = match escrowed {
Ok(ToppedUpPool { credited, tx, .. }) => {
let effect = topped_up_effect(state.pool_id, credited);
grade_deposit_credit(
store.add_deposit(self_address, state.pool_id, credited),
&effect,
tx,
)?;
return store
.get_by_pool_id(state.pool_id)?
.ok_or_else(|| escrowed_but_untracked(&effect, tx, "the credited row vanished"));
}
Err(err) => err,
};
if err.downcast_ref::<TopUpUnconfirmed>().is_some() {
return Err(err);
}
let Some(shortfall) = funding
.check_wallet(
rpc,
state.token,
self_address,
state.pool_id,
additional,
&err,
)
.await
else {
return Err(err);
};
if remaining.is_zero() {
return Err(cannot_pay(&state, &shortfall));
}
tracing::info!(
pool_id = %state.pool_id,
%remaining,
"continuing on the pool's remaining deposit (µUSDC) without a top-up"
);
Ok(state)
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn open_or_reuse_pool<P>(
store: &RedbBuyerPoolStore,
contract: &PaymentPool::PaymentPoolInstance<P>,
rpc: &P,
funding: &RunFunding,
signer: &Arc<PrivateKeySigner>,
provider: Address,
self_address: Address,
deployment: Deployment,
working_deposit: U256,
max_approve: bool,
adoption: ChainAdoption,
) -> anyhow::Result<(PoolContext, LaneSpend)>
where
P: alloy::providers::Provider + Clone,
{
let payment_pool_addr = deployment.payment_pool;
let tracked =
pool_to_reuse(store, contract, funding, self_address, deployment, adoption).await?;
if let Some(state) = tracked {
let lane = LaneKey {
pool_id: state.pool_id,
signer: self_address,
provider,
};
let recorded = state.lane_progress(lane).is_some();
let from_chain = !state.redeemed_elsewhere().is_zero();
let (state, prior_bytes, prior_amount) = if let Some(p) = state.lane_progress(lane) {
(state, p.last_bytes, p.last_amount)
} else {
let (bytes, amount) = lane_watermark(contract, lane).await?;
seed_lane(store, state, lane, bytes, amount)?
};
let spent = state.pool_spend();
let state = refill_if_low(
store,
contract,
rpc,
funding,
state,
self_address,
spent,
working_deposit,
max_approve,
)
.await?;
let lane_spend = LaneSpend {
outside: spent.saturating_sub(prior_amount),
prior: prior_amount,
recorded,
from_chain,
};
return self_owned_lane_ctx(
&state,
signer,
&deployment.voucher_domain(),
provider,
prior_bytes,
prior_amount,
)
.map(|ctx| (ctx, lane_spend));
}
let token = contract
.usdc()
.call()
.await
.map_err(|e| anyhow::Error::new(e).context("read PaymentPool.usdc()"))?;
let deposit = working_deposit;
let approve_amount = if max_approve { None } else { Some(deposit) };
ensure_allowance(rpc, token, self_address, payment_pool_addr, approve_amount).await?;
let opened = open_pool(
contract,
Arc::clone(signer),
deployment,
token,
self_address,
deposit,
)
.await?;
store
.record(&opened.state)
.map_err(|e| escrowed_but_untracked("buyer pool opened", opened.tx, e))?;
let spend = LaneSpend {
outside: U256::ZERO,
prior: U256::ZERO,
recorded: false,
from_chain: false,
};
Ok((
opened
.ctx
.with_provider(provider, U256::ZERO, U256::ZERO)
.with_capability(opened.capability),
spend,
))
}
pub(crate) fn temp_in_parent(target: &Path) -> std::io::Result<tempfile::NamedTempFile> {
match target.parent().filter(|p| !p.as_os_str().is_empty()) {
Some(p) => tempfile::NamedTempFile::new_in(p),
None => tempfile::NamedTempFile::new_in("."),
}
}
#[cfg(test)]
#[allow(
clippy::unwrap_used,
clippy::expect_used,
clippy::indexing_slicing,
clippy::panic
)]
mod tests;
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used, clippy::indexing_slicing)]
mod adoption_tests;
#[cfg(test)]
pub(crate) mod tests_support;