use std::io::Write;
use std::path::{Path, PathBuf};
use std::str::FromStr;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use alloy::primitives::{Address, B256, TxHash, U256};
use alloy::signers::local::PrivateKeySigner;
use anyhow::Context as _;
use decdn_client::buyer_pool::{
ToppedUpPool, UnconfirmedOpens, ensure_allowance, escrowed_but_untracked, grade_deposit_credit,
open_pool, top_up, topped_up_effect,
};
use decdn_common::admin::{AdminRpcClient as _, BuyerPoolsResponse};
use decdn_common::cli::{self, common::expand_tilde};
use decdn_common::config::{
DEFAULT_CHAIN_ID, DEFAULT_REDEEM_INTERVAL_SECS, FileConfig, capability_expiry_margin_secs,
load_file_config,
};
use decdn_common::redact::{sanitize_err_chain, sanitize_error_sources};
use decdn_incentive::buyer_pool::{BuyerLoad, BuyerPoolState, BuyerPoolStore};
use decdn_incentive::buyer_pool_redb::{ReadOnlyBuyerPoolStore, RedbBuyerPoolStore};
use decdn_incentive::eth_identity::{self, PasswordUse, load_signer};
use decdn_incentive::payment_pool::{PaymentPool, enumerate_owned_pools};
use decdn_incentive::{
Capability, CapabilityGrant, Deployment, PoolId, PoolLifecycleRevert, voucher_domain,
};
use serde::Serialize;
use decdn_client::provider;
use super::buyer_store::{
BuyerStoreOwner, classify_buyer_store, client_buyer_db, node_buyer_db,
open_client_store_for_escrow,
};
use super::chain_ctx;
use super::interrupt::{Interrupt, Interrupted, name_unconfirmed_open};
pub async fn pool_dispatch(args: &cli::PoolArgs, config_path: Option<&Path>) -> anyhow::Result<()> {
match &args.command {
cli::PoolCommand::List(a) => list(a, config_path).await,
cli::PoolCommand::Open(a) => open(a, config_path).await,
cli::PoolCommand::TopUp(a) => top_up_cmd(a, config_path).await,
cli::PoolCommand::Close(a) => close(a, config_path).await,
cli::PoolCommand::Reclaim(a) => reclaim(a, config_path).await,
cli::PoolCommand::Assign(a) => assign(a, config_path).await,
}
}
#[derive(Debug)]
struct Resolved {
rpc_url: String,
payment_pool: Address,
chain_id: u64,
data_dir: PathBuf,
keystore: PathBuf,
keystore_password_file: Option<PathBuf>,
}
impl Resolved {
const fn deployment(&self) -> Deployment {
Deployment {
chain_id: self.chain_id,
payment_pool: self.payment_pool,
}
}
}
fn resolve_data_dir(data_dir: Option<PathBuf>, file: &FileConfig) -> anyhow::Result<PathBuf> {
data_dir
.or_else(|| file.identity.as_ref().and_then(|i| i.data_dir.clone()))
.map(|p| expand_tilde(&p))
.or_else(cli::default_client_data_dir)
.ok_or_else(|| {
anyhow::anyhow!("data_dir not set and no default available (pass --data-dir)")
})
}
fn resolve_chain(args: &cli::PoolChainArgs, file: &FileConfig) -> anyhow::Result<Resolved> {
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 payment_pool_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(&payment_pool_raw, "payment_pool_address")?;
let chain_id = args
.chain_id
.or_else(|| bc.and_then(|b| b.chain_id))
.unwrap_or(DEFAULT_CHAIN_ID);
let data_dir = resolve_data_dir(args.data_dir.clone(), file)?;
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),
);
Ok(Resolved {
rpc_url,
payment_pool,
chain_id,
data_dir,
keystore,
keystore_password_file: args.keystore_password_file.as_deref().map(expand_tilde),
})
}
fn load_buyer_signer(chain: &Resolved) -> anyhow::Result<PrivateKeySigner> {
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();
load_signer(&chain.keystore, &password)
}
fn parse_pool_id(s: &str) -> anyhow::Result<PoolId> {
B256::from_str(s)
.map_err(|e| anyhow::anyhow!("invalid --pool {s:?}: expected a 32-byte hex hash: {e}"))
}
fn ensure_owned(on_chain_owner: Address, ours: Address, pool_id: PoolId) -> anyhow::Result<()> {
anyhow::ensure!(
on_chain_owner == ours,
"pool {pool_id} on-chain owner {on_chain_owner} is not this keystore's address {ours} \
— wrong keystore/contract, or a mistyped --pool"
);
Ok(())
}
async fn open(args: &cli::PoolOpenArgs, config_path: Option<&Path>) -> anyhow::Result<()> {
let file = load_file_config(config_path)?;
let data_dir = resolve_data_dir(args.chain.data_dir.clone(), &file)?;
let store = open_client_store_for_escrow(&data_dir, "open a pool")?;
let chain = resolve_chain(&args.chain, &file)?;
let signer = Arc::new(load_buyer_signer(&chain)?);
let owner = signer.address();
let rpc = provider::build_provider(&chain.rpc_url, &signer)?;
let contract = PaymentPool::new(chain.payment_pool, rpc.clone());
let deployment = chain.deployment();
let token = contract
.usdc()
.call()
.await
.map_err(|e| anyhow::Error::new(e).context("read PaymentPool.usdc()"))?;
let deposit = U256::from(args.deposit_micro_usdc);
ensure_allowance(&rpc, token, owner, chain.payment_pool, Some(deposit)).await?;
let opens = UnconfirmedOpens::default();
let mut interrupt = Interrupt::watch(opens.clone());
let opened = async {
let opened = open_pool(
&contract,
Arc::clone(&signer),
deployment,
token,
owner,
deposit,
Some(&opens),
)
.await?;
store
.record(&opened.state)
.map_err(|e| escrowed_but_untracked("buyer pool opened", opened.tx, e))?;
anyhow::Ok(opened.state.pool_id)
};
let result = tokio::select! {
result = opened => result,
() = interrupt.wait() => Err(Interrupted.into()),
};
let pool_id = name_unconfirmed_open(result, &opens.pending())?;
let out = PoolOpenJson {
pool_id: format!("{pool_id:#x}"),
};
serde_json::to_writer_pretty(std::io::stdout(), &out)?;
println!();
Ok(())
}
#[derive(Serialize)]
struct PoolOpenJson {
#[serde(rename = "poolId")]
pool_id: String,
}
async fn top_up_cmd(args: &cli::PoolTopUpArgs, config_path: Option<&Path>) -> anyhow::Result<()> {
let file = load_file_config(config_path)?;
let data_dir = resolve_data_dir(args.chain.data_dir.clone(), &file)?;
let store = open_client_store_for_escrow(&data_dir, "top up a pool")?;
let chain = resolve_chain(&args.chain, &file)?;
let pool_id = parse_pool_id(&args.pool)?;
let signer = Arc::new(load_buyer_signer(&chain)?);
let owner = signer.address();
let rpc = provider::build_provider(&chain.rpc_url, &signer)?;
let contract = PaymentPool::new(chain.payment_pool, rpc.clone());
let pool = contract
.getPool(pool_id)
.call()
.await
.map_err(|e| anyhow::Error::new(e).context("getPool failed"))?;
ensure_owned(pool.owner, owner, pool_id)?;
let additional = U256::from(args.amount_micro_usdc);
let token = contract
.usdc()
.call()
.await
.map_err(|e| anyhow::Error::new(e).context("read PaymentPool.usdc()"))?;
ensure_allowance(&rpc, token, owner, chain.payment_pool, Some(additional)).await?;
let ToppedUpPool { credited, tx, .. } = top_up(&contract, owner, pool_id, additional).await?;
let effect = topped_up_effect(pool_id, credited);
let new_deposit =
grade_deposit_credit(store.add_deposit(owner, pool_id, credited), &effect, tx)?;
println!("topped up pool {pool_id} by {credited} µUSDC; deposit now {new_deposit} µUSDC");
Ok(())
}
#[derive(Clone, Copy)]
enum LocalBookkeeping<'a> {
Store {
store: &'a dyn BuyerPoolStore,
deployment: Deployment,
},
DaemonOwned(&'a Path),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum NotOpenClear {
Cleared,
Untouched,
Faulted,
}
impl NotOpenClear {
const fn suffix(self) -> &'static str {
match self {
Self::Cleared => "; cleared its row from the local store",
Self::Untouched => "",
Self::Faulted => "; its local row could not be cleared (see the warning above)",
}
}
}
impl LocalBookkeeping<'_> {
const fn new<'a>(
store: Option<&'a RedbBuyerPoolStore>,
buyer_db: &'a Path,
deployment: Deployment,
) -> LocalBookkeeping<'a> {
match store {
Some(store) => LocalBookkeeping::Store { store, deployment },
None => LocalBookkeeping::DaemonOwned(buyer_db),
}
}
fn report_daemon_row_untouched(buyer_db: &Path, pool_id: PoolId, verb: &str) {
eprintln!(
"warning: pool {pool_id} {verb} on-chain, but the daemon's row in {} was NOT \
cleared — `decdn pool` writes only the client store, and a running decdn-node holds \
its own exclusively. Run `decdn node pools`: if that pool is listed, the daemon is \
still using it — restart decdn-node so its bootstrap reconciles against the chain.",
buyer_db.display()
);
}
fn forget_after_close(
self,
owner: Address,
pool_id: PoolId,
tx: TxHash,
reclaim_note: &str,
) -> anyhow::Result<()> {
match self {
Self::Store { store, deployment } => grade_local_forget(
forget_on(store, deployment, owner, pool_id),
pool_id,
tx,
reclaim_note,
),
Self::DaemonOwned(buyer_db) => {
Self::report_daemon_row_untouched(buyer_db, pool_id, "closed");
Ok(())
}
}
}
fn forget_after_reclaim(self, owner: Address, pool_id: PoolId) {
match self {
Self::Store { store, deployment } => {
if let Err(e) = forget_on(store, deployment, owner, pool_id) {
eprintln!(
"warning: pool {pool_id} reclaimed on-chain but clearing it from the \
local store failed: {e}; if the row survived, a later fetch reuses this \
closed pool and its vouchers are rejected — re-run `decdn pool reclaim \
--pool {pool_id}` to clear it"
);
}
}
Self::DaemonOwned(buyer_db) => {
Self::report_daemon_row_untouched(buyer_db, pool_id, "reclaimed");
}
}
}
fn forget_not_open(
self,
owner: Address,
pool_id: PoolId,
status: PaymentPool::Status,
) -> NotOpenClear {
if !no_longer_open(status) {
return NotOpenClear::Untouched;
}
match self {
Self::Store { store, deployment } => {
match forget_on(store, deployment, owner, pool_id) {
Ok(true) => NotOpenClear::Cleared,
Ok(false) => NotOpenClear::Untouched,
Err(e) => {
eprintln!(
"warning: pool {pool_id} is {} on-chain but clearing it from the \
local store failed: {e}; if the row survived, a later fetch may \
reuse this pool, whose vouchers stop being redeemable at its \
dispute deadline — re-run this command to retry the clear",
status_label(status)
);
NotOpenClear::Faulted
}
}
}
Self::DaemonOwned(buyer_db) => {
let verb = format!("is already {}", status_label(status));
Self::report_daemon_row_untouched(buyer_db, pool_id, &verb);
NotOpenClear::Untouched
}
}
}
}
fn forget_on(
store: &dyn BuyerPoolStore,
deployment: Deployment,
owner: Address,
pool_id: PoolId,
) -> Result<bool, decdn_incentive::StoreError> {
match store.get_by_pool_id(pool_id)? {
Some(state) if state.is_on(deployment) => store.forget_if_pool(owner, pool_id),
_ => Ok(false),
}
}
const fn no_longer_open(status: PaymentPool::Status) -> bool {
matches!(
status,
PaymentPool::Status::Closing | PaymentPool::Status::Closed
)
}
fn grade_local_forget(
outcome: Result<bool, decdn_incentive::StoreError>,
pool_id: PoolId,
tx: TxHash,
reclaim_note: &str,
) -> anyhow::Result<()> {
match outcome {
Ok(_) => Ok(()),
Err(cause) => Err(anyhow::Error::new(CloseUnrecorded {
pool_id,
tx,
reclaim_note: reclaim_note.to_string(),
cause,
})),
}
}
#[derive(Debug)]
struct CloseUnrecorded {
pool_id: PoolId,
tx: TxHash,
reclaim_note: String,
cause: decdn_incentive::StoreError,
}
impl std::fmt::Display for CloseUnrecorded {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let Self {
pool_id,
tx,
reclaim_note,
cause,
} = self;
write!(
f,
"pool {pool_id} closed on-chain (tx {tx}) but clearing it from the local store \
failed: {cause}; the row may survive and send a later fetch back to this closing \
pool, whose vouchers stop being redeemable at the dispute deadline — re-run `decdn \
pool close --pool {pool_id}` to clear it, or {reclaim_note}, which also clears the \
row"
)
}
}
impl std::error::Error for CloseUnrecorded {}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum TxOutcome {
Landed,
Reverted,
}
const fn receipt_outcome(landed: bool) -> TxOutcome {
if landed {
TxOutcome::Landed
} else {
TxOutcome::Reverted
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ClosePlan {
Close,
Skip(&'static str),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ReclaimPlan {
Reclaim,
SkipOpen,
SkipInWindow(u64),
SkipClosed,
SkipUnknown,
}
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
struct BatchTally {
acted: usize,
skipped: usize,
failed: usize,
unconfirmed: usize,
unrecorded: usize,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ErrorClass {
Failed,
Unconfirmed,
LandedUnrecorded,
}
impl BatchTally {
const fn record_skip(&mut self, cleared: NotOpenClear) {
self.skipped += 1;
if matches!(cleared, NotOpenClear::Faulted) {
self.unrecorded += 1;
}
}
fn record_error(&mut self, err: &anyhow::Error) -> ErrorClass {
if err.downcast_ref::<CloseUnrecorded>().is_some() {
self.acted += 1;
self.unrecorded += 1;
ErrorClass::LandedUnrecorded
} else if err.downcast_ref::<TxUnconfirmed>().is_some() {
self.unconfirmed += 1;
ErrorClass::Unconfirmed
} else {
self.failed += 1;
ErrorClass::Failed
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct TxUnconfirmed {
label: &'static str,
tx: Option<TxHash>,
}
impl std::fmt::Display for TxUnconfirmed {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let label = self.label;
match self.tx {
Some(tx) => write!(
f,
"{label} sent (tx {tx:#x}) but the receipt could not be fetched; it may still \
mine — check the tx before re-running"
),
None => write!(
f,
"{label} send failed: the request may have reached the node and broadcast the \
transaction — check this signer's pending and mined transactions before \
re-running"
),
}
}
}
impl std::error::Error for TxUnconfirmed {}
const fn plan_close(status: PaymentPool::Status) -> ClosePlan {
match status {
PaymentPool::Status::Open => ClosePlan::Close,
PaymentPool::Status::Closing => ClosePlan::Skip("already closing"),
PaymentPool::Status::Closed => ClosePlan::Skip("already closed"),
_ => ClosePlan::Skip("unknown on-chain status"),
}
}
const fn plan_reclaim(status: PaymentPool::Status, dispute_deadline: u64, now: u64) -> ReclaimPlan {
match status {
PaymentPool::Status::Open => ReclaimPlan::SkipOpen,
PaymentPool::Status::Closing if now >= dispute_deadline => ReclaimPlan::Reclaim,
PaymentPool::Status::Closing => ReclaimPlan::SkipInWindow(dispute_deadline),
PaymentPool::Status::Closed => ReclaimPlan::SkipClosed,
_ => ReclaimPlan::SkipUnknown,
}
}
fn single_reclaim_gate(
pool_id: PoolId,
owner: Address,
status: PaymentPool::Status,
dispute_deadline: u64,
now: u64,
) -> anyhow::Result<()> {
anyhow::ensure!(
owner != Address::ZERO,
"pool {pool_id} does not exist on this PaymentPool contract — a mistyped --pool, or \
the wrong payment_pool address"
);
match plan_reclaim(status, dispute_deadline, now) {
ReclaimPlan::Reclaim => Ok(()),
ReclaimPlan::SkipOpen => anyhow::bail!(
"pool {pool_id} is still Open — run `decdn pool close --pool {pool_id}` first"
),
ReclaimPlan::SkipInWindow(deadline) => anyhow::bail!(
"pool {pool_id} is still in its dispute window; reclaimable after {}",
format_expiry(deadline, now)
),
ReclaimPlan::SkipClosed => {
anyhow::bail!("pool {pool_id} is already Closed — its residual was already reclaimed")
}
ReclaimPlan::SkipUnknown => {
anyhow::bail!("pool {pool_id} has an unrecognized on-chain status — nothing to reclaim")
}
}
}
fn batch_result(tally: &BatchTally, verb: &str) -> anyhow::Result<()> {
let failed = (tally.failed > 0).then(|| format!("{} pool(s) failed to {verb}", tally.failed));
let unconfirmed = (tally.unconfirmed > 0).then(|| {
format!(
"{} pool(s) may have broadcast a {verb} transaction — check this signer's pending \
and mined transactions before re-running",
tally.unconfirmed
)
});
let unrecorded = (tally.unrecorded > 0).then(|| {
format!(
"{} pool(s) are no longer Open on-chain but kept a local row that may be stale — \
re-run the same command to clear it",
tally.unrecorded
)
});
let causes: Vec<String> = [failed, unconfirmed, unrecorded]
.into_iter()
.flatten()
.collect();
if causes.is_empty() {
return Ok(());
}
anyhow::bail!("{}; see the summary above", causes.join(", and "))
}
async fn close_and_forget<P>(
contract: &PaymentPool::PaymentPoolInstance<P>,
store: LocalBookkeeping<'_>,
owner: Address,
pool_id: PoolId,
) -> anyhow::Result<String>
where
P: alloy::providers::Provider + Clone,
{
let pending = match contract.closePool(pool_id).send().await {
Ok(pending) => pending,
Err(e) if e.as_revert_data().is_some() => {
return Err(send_reverted(e, "closePool", pool_id, None));
}
Err(e) => return Err(send_failed(e, "closePool")),
};
let tx = *pending.tx_hash();
let receipt = pending.get_receipt().await.map_err(|e| {
anyhow::Error::new(e).context(TxUnconfirmed {
label: "closePool",
tx: Some(tx),
})
})?;
match receipt_outcome(receipt.status()) {
TxOutcome::Landed => {
let deadline_note = match contract.getPool(pool_id).call().await {
Ok(p) => format!("after Unix {}", p.disputeDeadline),
Err(e) => format!(
"after the dispute window (couldn't read the exact deadline: {})",
sanitize_error_sources(&e)
),
};
let reclaim_note = format!("run `decdn pool reclaim --pool {pool_id}` {deadline_note}");
store.forget_after_close(owner, pool_id, tx, &reclaim_note)?;
Ok(reclaim_note)
}
TxOutcome::Reverted => anyhow::bail!(
"closePool reverted on-chain for pool {pool_id} (tx {tx:#x}; it may have raced a \
concurrent close)"
),
}
}
async fn reclaim_and_forget<P>(
contract: &PaymentPool::PaymentPoolInstance<P>,
store: LocalBookkeeping<'_>,
owner: Address,
pool_id: PoolId,
) -> anyhow::Result<()>
where
P: alloy::providers::Provider + Clone,
{
let pending = match contract.reclaim(pool_id).send().await {
Ok(pending) => pending,
Err(e) if e.as_revert_data().is_some() => {
return Err(send_reverted(
e,
"reclaim",
pool_id,
Some(RECLAIM_UNKNOWN_REVERT),
));
}
Err(e) => return Err(send_failed(e, "reclaim")),
};
let tx = *pending.tx_hash();
let receipt = pending.get_receipt().await.map_err(|e| {
anyhow::Error::new(e).context(TxUnconfirmed {
label: "reclaim",
tx: Some(tx),
})
})?;
match receipt_outcome(receipt.status()) {
TxOutcome::Landed => {
store.forget_after_reclaim(owner, pool_id);
Ok(())
}
TxOutcome::Reverted => anyhow::bail!(reclaim_reverted(pool_id, tx)),
}
}
const RECLAIM_REVERT_GUESS: &str =
"most likely a concurrent reclaim; the USDC refund transfer to the owner can also revert";
const RECLAIM_UNKNOWN_REVERT: &str = "not a PaymentPool lifecycle error; most likely the USDC \
refund transfer to the owner reverted";
fn reclaim_reverted(pool_id: PoolId, tx: TxHash) -> String {
format!(
"reclaim reverted on-chain for pool {pool_id} (tx {tx:#x}) after the pre-check passed \
({RECLAIM_REVERT_GUESS})"
)
}
fn send_reverted(
err: alloy::contract::Error,
label: &str,
pool_id: PoolId,
unknown_cause: Option<&str>,
) -> anyhow::Error {
let cause = match PoolLifecycleRevert::decode(err.as_revert_data().as_ref()) {
Some(revert) => {
let hint = match revert {
PoolLifecycleRevert::PoolNotOpen => {
"the pool is no longer Open; it may have raced a concurrent close"
}
PoolLifecycleRevert::NotPoolOwner => "this keystore does not own the pool",
PoolLifecycleRevert::PoolNotClosing => {
"the pool is not Closing; most likely a concurrent reclaim already closed it"
}
PoolLifecycleRevert::GraceWindowActive => "the dispute window has not elapsed yet",
};
format!(": {} — {hint}", revert.signature())
}
None => unknown_cause.map_or_else(String::new, |cause| format!(" ({cause})")),
};
anyhow::Error::new(err).context(format!(
"{label} would revert for pool {pool_id}{cause} (rejected at gas estimation; nothing was \
sent)"
))
}
fn send_failed(err: alloy::contract::Error, label: &'static str) -> anyhow::Error {
if decdn_incentive::tx::send_broadcast_unknown(&err) {
anyhow::Error::new(err).context(TxUnconfirmed { label, tx: None })
} else {
anyhow::Error::new(err).context(format!("{label} send failed"))
}
}
async fn close(args: &cli::PoolCloseArgs, config_path: Option<&Path>) -> anyhow::Result<()> {
let file = load_file_config(config_path)?;
let target = args.pool.as_deref().map(parse_pool_id).transpose()?;
let store_owner = classify_buyer_store(&resolve_data_dir(args.chain.data_dir.clone(), &file)?)?;
if target.is_none() {
store_owner.refuse_sweep("close")?;
}
let chain = resolve_chain(&args.chain, &file)?;
let store = store_owner.open_for_write()?;
let buyer_db = node_buyer_db(&chain.data_dir);
let books = LocalBookkeeping::new(store.as_ref(), &buyer_db, chain.deployment());
let signer = Arc::new(load_buyer_signer(&chain)?);
let owner = signer.address();
let rpc = provider::build_provider(&chain.rpc_url, &signer)?;
let contract = PaymentPool::new(chain.payment_pool, rpc);
let Some(pool_id) = target else {
return close_all(&contract, books, owner).await;
};
let pool = contract
.getPool(pool_id)
.call()
.await
.map_err(|e| anyhow::Error::new(e).context("getPool failed"))?;
ensure_owned(pool.owner, owner, pool_id)?;
if !matches!(pool.status, PaymentPool::Status::Open) {
return Err(refuse_close(books, owner, pool_id, pool.status));
}
let reclaim_note = close_and_forget(&contract, books, owner, pool_id).await?;
println!("closed pool {pool_id}; dispute window open — {reclaim_note}");
Ok(())
}
fn refuse_close(
books: LocalBookkeeping<'_>,
owner: Address,
pool_id: PoolId,
status: PaymentPool::Status,
) -> anyhow::Error {
let cleared = books.forget_not_open(owner, pool_id, status).suffix();
anyhow::anyhow!(
"pool {pool_id} is not Open — nothing to close (already closing or closed){cleared}"
)
}
async fn close_all<P>(
contract: &PaymentPool::PaymentPoolInstance<P>,
store: LocalBookkeeping<'_>,
owner: Address,
) -> anyhow::Result<()>
where
P: alloy::providers::Provider + Clone,
{
let ids = enumerate_owned_pools(contract, owner).await?;
if ids.is_empty() {
println!("no pools found for this keystore — nothing to close");
return Ok(());
}
let mut tally = BatchTally::default();
for pool_id in ids {
let status = match contract.getPool(pool_id).call().await {
Ok(pool) => pool.status,
Err(e) => {
eprintln!(
"failed to read pool {pool_id}: {}",
sanitize_error_sources(&e)
);
tally.failed += 1;
continue;
}
};
match plan_close(status) {
ClosePlan::Close => match close_and_forget(contract, store, owner, pool_id).await {
Ok(note) => {
println!("closed pool {pool_id}; dispute window open — {note}");
tally.acted += 1;
}
Err(e) => {
let msg = sanitize_err_chain(&e);
match tally.record_error(&e) {
ErrorClass::LandedUnrecorded => eprintln!("warning: {msg}"),
ErrorClass::Unconfirmed => {
eprintln!("close of pool {pool_id} unconfirmed: {msg}");
}
ErrorClass::Failed => eprintln!("failed to close pool {pool_id}: {msg}"),
}
}
},
ClosePlan::Skip(reason) => {
let cleared = store.forget_not_open(owner, pool_id, status);
println!("skipped pool {pool_id}: {reason}{}", cleared.suffix());
tally.record_skip(cleared);
}
}
}
println!(
"close --all: closed {}, skipped {}, failed {}, unconfirmed {}, row not cleared {}",
tally.acted, tally.skipped, tally.failed, tally.unconfirmed, tally.unrecorded
);
batch_result(&tally, "close")
}
async fn reclaim(args: &cli::PoolReclaimArgs, config_path: Option<&Path>) -> anyhow::Result<()> {
let file = load_file_config(config_path)?;
let target = args.pool.as_deref().map(parse_pool_id).transpose()?;
let store_owner = classify_buyer_store(&resolve_data_dir(args.chain.data_dir.clone(), &file)?)?;
if target.is_none() {
store_owner.refuse_sweep("reclaim")?;
}
let chain = resolve_chain(&args.chain, &file)?;
let deployment = chain.deployment();
if let Some(pool_id) = target {
let reader = PaymentPool::new(
chain.payment_pool,
provider::build_read_provider(&chain.rpc_url)?,
);
let pool = reader
.getPool(pool_id)
.call()
.await
.map_err(|e| anyhow::Error::new(e).context("getPool failed"))?;
let now = chain_ctx::head_timestamp(reader.provider()).await?;
if let Err(refusal) =
single_reclaim_gate(pool_id, pool.owner, pool.status, pool.disputeDeadline, now)
{
if !no_longer_open(pool.status) {
return Err(refusal);
}
let opened = store_owner.open_for_write();
let buyer_db = node_buyer_db(&chain.data_dir);
let books = opened
.as_ref()
.map(|store| LocalBookkeeping::new(store.as_ref(), &buyer_db, deployment));
return Err(refuse_reclaim(
&refusal,
books,
pool.owner,
pool_id,
pool.status,
));
}
}
let store = store_owner.open_for_write()?;
let buyer_db = node_buyer_db(&chain.data_dir);
let books = LocalBookkeeping::new(store.as_ref(), &buyer_db, deployment);
let signer = Arc::new(load_buyer_signer(&chain)?);
let owner = signer.address();
let rpc = provider::build_provider(&chain.rpc_url, &signer)?;
let contract = PaymentPool::new(chain.payment_pool, rpc);
let Some(pool_id) = target else {
return reclaim_all(&contract, books, owner).await;
};
reclaim_and_forget(&contract, books, owner, pool_id).await?;
println!("reclaimed pool {pool_id}; residual deposit refunded to its owner");
Ok(())
}
fn refuse_reclaim(
refusal: &anyhow::Error,
books: Result<LocalBookkeeping<'_>, &anyhow::Error>,
owner: Address,
pool_id: PoolId,
status: PaymentPool::Status,
) -> anyhow::Error {
let cleared = match books {
Ok(books) => books.forget_not_open(owner, pool_id, status),
Err(e) => {
eprintln!(
"warning: pool {pool_id} is {} on-chain but the local store could not be opened \
to clear its row: {}; re-run this command once the store opens",
status_label(status),
sanitize_err_chain(e)
);
NotOpenClear::Faulted
}
};
anyhow::anyhow!("{refusal:#}{}", cleared.suffix())
}
async fn reclaim_all<P>(
contract: &PaymentPool::PaymentPoolInstance<P>,
store: LocalBookkeeping<'_>,
owner: Address,
) -> anyhow::Result<()>
where
P: alloy::providers::Provider + Clone,
{
let ids = enumerate_owned_pools(contract, owner).await?;
if ids.is_empty() {
println!("no pools found for this keystore — nothing to reclaim");
return Ok(());
}
let now = chain_ctx::head_timestamp(contract.provider()).await?;
let mut tally = BatchTally::default();
for pool_id in ids {
let pool = match contract.getPool(pool_id).call().await {
Ok(pool) => pool,
Err(e) => {
eprintln!(
"failed to read pool {pool_id}: {}",
sanitize_error_sources(&e)
);
tally.failed += 1;
continue;
}
};
match plan_reclaim(pool.status, pool.disputeDeadline, now) {
ReclaimPlan::Reclaim => {
match reclaim_and_forget(contract, store, owner, pool_id).await {
Ok(()) => {
println!(
"reclaimed pool {pool_id}; residual deposit refunded to its owner"
);
tally.acted += 1;
}
Err(e) => {
let msg = sanitize_err_chain(&e);
match tally.record_error(&e) {
ErrorClass::LandedUnrecorded => eprintln!("warning: {msg}"),
ErrorClass::Unconfirmed => {
eprintln!("reclaim of pool {pool_id} unconfirmed: {msg}");
}
ErrorClass::Failed => {
eprintln!("failed to reclaim pool {pool_id}: {msg}");
}
}
}
}
}
ReclaimPlan::SkipOpen => {
println!("skipped pool {pool_id}: still Open — run `pool close` first");
tally.skipped += 1;
}
ReclaimPlan::SkipInWindow(deadline) => {
let cleared = store.forget_not_open(owner, pool_id, pool.status);
println!(
"skipped pool {pool_id}: dispute window open — reclaimable after {}{}",
format_expiry(deadline, now),
cleared.suffix()
);
tally.record_skip(cleared);
}
ReclaimPlan::SkipClosed => {
let cleared = store.forget_not_open(owner, pool_id, pool.status);
println!(
"skipped pool {pool_id}: already reclaimed{}",
cleared.suffix()
);
tally.record_skip(cleared);
}
ReclaimPlan::SkipUnknown => {
println!("skipped pool {pool_id}: unrecognized on-chain status");
tally.skipped += 1;
}
}
}
println!(
"reclaim --all: reclaimed {}, skipped {}, failed {}, unconfirmed {}, row not cleared {}",
tally.acted, tally.skipped, tally.failed, tally.unconfirmed, tally.unrecorded
);
batch_result(&tally, "reclaim")
}
fn resolve_expiry(
now: u64,
expiry_secs: Option<u64>,
expiry_at: Option<u64>,
) -> anyhow::Result<u64> {
let expiry = match (expiry_secs, expiry_at) {
(Some(secs), None) => now.checked_add(secs).ok_or_else(|| {
anyhow::anyhow!("--expiry-secs {secs} overflows the Unix epoch from now ({now})")
})?,
(None, Some(at)) => at,
_ => anyhow::bail!("exactly one of --expiry-secs or --expiry-at is required"),
};
anyhow::ensure!(
expiry > now,
"capability expiry {expiry} is not in the future (now is {now}); it would be dead on \
arrival — pick a later --expiry-at or a positive --expiry-secs"
);
Ok(expiry)
}
const DEFAULT_NODE_EXPIRY_MARGIN_SECS: u64 =
capability_expiry_margin_secs(DEFAULT_REDEEM_INTERVAL_SECS);
const fn expires_within_default_node_margin(expiry: u64, now: u64) -> bool {
expiry.saturating_sub(now) <= DEFAULT_NODE_EXPIRY_MARGIN_SECS
}
fn unix_now() -> anyhow::Result<u64> {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.map_err(|e| anyhow::anyhow!("the system clock is before the Unix epoch: {e}"))
}
const HEAD_READ_TIMEOUT: Duration = Duration::from_secs(5);
async fn issuance_now(rpc_url: &str, timeout: Duration) -> anyhow::Result<u64> {
let head = match provider::build_read_provider(rpc_url) {
Ok(reader) => tokio::time::timeout(timeout, chain_ctx::head_timestamp(&reader))
.await
.unwrap_or_else(|_| Err(anyhow::anyhow!("no answer within {timeout:?}"))),
Err(e) => Err(e),
};
match head {
Ok(now) => Ok(now),
Err(e) => {
eprintln!(
"warning: could not read the chain head time ({}); checking the capability \
expiry against the local clock instead",
sanitize_err_chain(&e)
);
unix_now()
}
}
}
fn format_expiry(expiry: u64, now: u64) -> String {
let remaining = expiry.saturating_sub(now);
let days = remaining / 86_400;
let hours = (remaining % 86_400) / 3_600;
let mins = (remaining % 3_600) / 60;
format!("Unix {expiry} (in {days}d {hours}h {mins}m)")
}
async fn assign(args: &cli::PoolAssignArgs, config_path: Option<&Path>) -> anyhow::Result<()> {
let file = load_file_config(config_path)?;
let chain = resolve_chain(&args.chain, &file)?;
let pool_id = parse_pool_id(&args.pool)?;
let signer_addr = super::chain_ctx::parse_nonzero_address(&args.signer, "--signer")?;
let now = issuance_now(&chain.rpc_url, HEAD_READ_TIMEOUT).await?;
let expiry = resolve_expiry(now, args.expiry_secs, args.expiry_at)?;
if expires_within_default_node_margin(expiry, now) {
eprintln!(
"warning: the capability expires in {}s; a node with the default redeem interval \
refuses vouchers within {DEFAULT_NODE_EXPIRY_MARGIN_SECS}s of a capability's \
expiry, so such a node serves this delegate nothing",
expiry.saturating_sub(now)
);
}
let owner_signer = load_buyer_signer(&chain)?;
let owner = owner_signer.address();
let domain = voucher_domain(chain.chain_id, chain.payment_pool);
let spending_cap = args.cap_micro_usdc;
let capability = Capability {
signer: signer_addr,
spending_cap,
pool_id,
expiry,
};
let signed = capability.sign(&owner_signer, &domain)?;
let grant = CapabilityGrant::from_signed_capability(&signed);
let token = grant.to_token();
let recovered = grant
.owner(&domain)
.map_err(|e| anyhow::anyhow!("re-recovering the owner from the fresh token failed: {e}"))?;
match provider::build_provider(&chain.rpc_url, &owner_signer) {
Ok(rpc) => {
let contract = PaymentPool::new(chain.payment_pool, rpc);
ensure_on_chain_owner(&contract, pool_id, owner).await?;
}
Err(e) => eprintln!(
"warning: could not build an RPC provider to check the on-chain owner of pool \
{pool_id} ({}); issuing anyway — the node verifies the owner signature at redemption",
sanitize_err_chain(&e)
),
}
println!("pool: {pool_id}");
println!("delegate: {signer_addr} (the voucher-signing key this authorizes)");
println!(
"spending cap: {} USDC ({} µUSDC)",
format_usdc_u256(U256::from(spending_cap)),
args.cap_micro_usdc
);
println!("expiry: {}", format_expiry(expiry, now));
println!("owner: {recovered} (recovered from the signature)");
println!("Token (give this to the delegated client):");
println!("{token}");
Ok(())
}
async fn ensure_on_chain_owner<P>(
contract: &PaymentPool::PaymentPoolInstance<P>,
pool_id: PoolId,
expected: Address,
) -> anyhow::Result<()>
where
P: alloy::providers::Provider + Clone,
{
match contract.getPool(pool_id).call().await {
Ok(pool) => grade_on_chain_owner(pool.owner, expected).into_result(pool_id, expected),
Err(e) => {
eprintln!(
"warning: could not read pool {pool_id} on-chain to confirm ownership ({}); \
issuing anyway — the node verifies the owner signature at redemption",
sanitize_error_sources(&e)
);
Ok(())
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum OwnerVerdict {
Owned,
NoSuchPool,
OwnedByOther(Address),
}
fn grade_on_chain_owner(on_chain: Address, expected: Address) -> OwnerVerdict {
if on_chain.is_zero() {
OwnerVerdict::NoSuchPool
} else if on_chain == expected {
OwnerVerdict::Owned
} else {
OwnerVerdict::OwnedByOther(on_chain)
}
}
impl OwnerVerdict {
fn into_result(self, pool_id: PoolId, expected: Address) -> anyhow::Result<()> {
match self {
Self::Owned => Ok(()),
Self::NoSuchPool => anyhow::bail!(
"pool {pool_id} does not exist on this PaymentPool contract; a capability for a \
pool that was never opened is rejected at redemption, so no token was issued — \
check --pool, --payment-pool-address, and --chain-id"
),
Self::OwnedByOther(on_chain) => anyhow::bail!(
"pool {pool_id} on-chain owner {on_chain} is not this keystore's address \
{expected}; a capability signed by a non-owner is rejected at redemption, so no \
token was issued — sign with the owner keystore, or check --pool"
),
}
}
}
async fn list(args: &cli::PoolListArgs, config_path: Option<&Path>) -> anyhow::Result<()> {
let data_dir = match args.chain.data_dir.clone() {
Some(dir) => expand_tilde(&dir),
None => resolve_data_dir(None, &load_file_config(config_path)?)?,
};
if args.all {
return list_all(args, config_path, &data_dir).await;
}
match classify_buyer_store(&data_dir)? {
BuyerStoreOwner::Node { data_dir, .. } => {
list_from_daemon(args, config_path, &data_dir).await
}
BuyerStoreOwner::Client { data_dir } => list_from_client_store(args, &data_dir),
}
}
async fn list_all(
args: &cli::PoolListArgs,
config_path: Option<&Path>,
data_dir: &Path,
) -> anyhow::Result<()> {
let file = load_file_config(config_path)?;
let chain = resolve_chain(&args.chain, &file)?;
let signer = Arc::new(load_buyer_signer(&chain)?);
let owner = signer.address();
let rpc = provider::build_provider(&chain.rpc_url, &signer)?;
let contract = PaymentPool::new(chain.payment_pool, rpc);
let ids = enumerate_owned_pools(&contract, owner).await?;
let tracked = tracked_pool_ids(args, config_path, data_dir).await;
let now = chain_ctx::head_timestamp(contract.provider()).await?;
let mut rows = Vec::with_capacity(ids.len());
for pool_id in ids {
let pool = contract.getPool(pool_id).call().await.map_err(|e| {
anyhow::anyhow!("getPool({pool_id}) failed: {}", sanitize_error_sources(&e))
})?;
rows.push(ChainPoolRow {
pool_id,
status: pool.status,
deposit: U256::from(pool.deposit),
total_redeemed: U256::from(pool.totalRedeemed),
dispute_deadline: pool.disputeDeadline,
tracked: tracked.as_ref().and_then(|local| local.verdict(pool_id)),
});
}
if let Some(local) = tracked.as_ref() {
write_skipped_pools(&mut std::io::stderr().lock(), &local.undecodable())?;
}
let mut out = std::io::stdout().lock();
if args.json {
let view = PoolListAllJson {
source: SOURCE_CHAIN,
owner: format!("{owner:#x}"),
chain_id: chain.chain_id,
payment_pool: format!("{:#x}", chain.payment_pool),
local_store_read: tracked.is_some(),
pools: rows.iter().map(|r| ChainPoolJson::at(r, now)).collect(),
};
serde_json::to_writer_pretty(&mut out, &view)?;
writeln!(out)?;
} else {
write_chain_pools(
&mut out,
owner,
chain.chain_id,
&rows,
now,
tracked.is_some(),
)?;
}
Ok(())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum RowState {
Decoded,
Undecodable,
}
#[derive(Debug, Default)]
struct TrackedPools(std::collections::BTreeMap<PoolId, RowState>);
impl TrackedPools {
fn verdict(&self, pool_id: PoolId) -> Option<bool> {
match self.0.get(&pool_id) {
Some(RowState::Decoded) => Some(true),
Some(RowState::Undecodable) => None,
None => Some(false),
}
}
fn undecodable(&self) -> Vec<PoolId> {
self.0
.iter()
.filter(|(_, state)| **state == RowState::Undecodable)
.map(|(id, _)| *id)
.collect()
}
fn from_wire(resp: &BuyerPoolsResponse) -> Option<Self> {
let mut rows = std::collections::BTreeMap::new();
for (id, state) in resp
.pools
.iter()
.map(|p| (p.pool_id.as_str(), RowState::Decoded))
.chain(
resp.skipped
.iter()
.map(|p| (p.as_str(), RowState::Undecodable)),
)
{
rows.insert(PoolId::from_str(id).ok()?, state);
}
Some(Self(rows))
}
}
impl From<BuyerLoad> for TrackedPools {
fn from(load: BuyerLoad) -> Self {
Self(
load.pools
.iter()
.map(|p| (p.pool_id, RowState::Decoded))
.chain(
load.skipped
.into_iter()
.map(|id| (id, RowState::Undecodable)),
)
.collect(),
)
}
}
async fn tracked_pool_ids(
args: &cli::PoolListArgs,
config_path: Option<&Path>,
data_dir: &Path,
) -> Option<TrackedPools> {
let owner = match classify_buyer_store(data_dir) {
Ok(owner) => owner,
Err(err) => {
let mut w = std::io::stderr().lock();
let _ = writeln!(
w,
"warning: {err:#}; every TRACKED value below is unknown, which is not the same \
as untracked"
);
return None;
}
};
match owner {
BuyerStoreOwner::Client { data_dir } => read_local_pools(&client_buyer_db(&data_dir)),
BuyerStoreOwner::Node { data_dir, .. } => {
let buyer_db = node_buyer_db(&data_dir);
match ReadOnlyBuyerPoolStore::open_file(&buyer_db) {
Ok(reader) => load_local_pools(&reader, &buyer_db),
Err(decdn_incentive::StoreError::AlreadyOpen { .. }) => {
daemon_pool_ids(args, config_path).await
}
Err(err) => {
warn_unreadable_store(&buyer_db, &err);
None
}
}
}
}
}
fn read_local_pools(path: &Path) -> Option<TrackedPools> {
match ReadOnlyBuyerPoolStore::open_file(path) {
Ok(reader) => load_local_pools(&reader, path),
Err(decdn_incentive::StoreError::Absent { .. }) => Some(TrackedPools::default()),
Err(err) => {
warn_unreadable_store(path, &err);
None
}
}
}
fn load_local_pools(reader: &ReadOnlyBuyerPoolStore, path: &Path) -> Option<TrackedPools> {
match reader.load_all() {
Ok(load) => Some(TrackedPools::from(load)),
Err(err) => {
warn_unreadable_store(path, &err);
None
}
}
}
fn warn_unreadable_store(path: &Path, err: &decdn_incentive::StoreError) {
let mut w = std::io::stderr().lock();
let _ = writeln!(
w,
"warning: the local buyer store at {} could not be read ({err}); every TRACKED value \
below is unknown, which is not the same as untracked",
path.display()
);
}
async fn daemon_pool_ids(
args: &cli::PoolListArgs,
config_path: Option<&Path>,
) -> Option<TrackedPools> {
if args.timeout_ms == 0 {
let mut w = std::io::stderr().lock();
let _ = writeln!(
w,
"warning: --timeout-ms 0 skips the daemon lookup (jsonrpsee reads a zero duration \
as 'never'), so every TRACKED value below is unknown"
);
return None;
}
let url =
crate::commands::node::resolve_admin_url(args.admin_url.as_deref(), config_path).ok()?;
let client = jsonrpsee::http_client::HttpClientBuilder::default()
.request_timeout(std::time::Duration::from_millis(args.timeout_ms))
.build(&url)
.ok()?;
let resp: BuyerPoolsResponse = client.pools().await.ok()?;
let tracked = TrackedPools::from_wire(&resp);
if tracked.is_none() {
let mut w = std::io::stderr().lock();
let _ = writeln!(
w,
"warning: the daemon reported a pool id this binary cannot parse, so every TRACKED \
value below is unknown. The daemon and this CLI disagree on the wire format — \
check that both are the same build."
);
}
tracked
}
fn list_from_client_store(args: &cli::PoolListArgs, data_dir: &Path) -> anyhow::Result<()> {
let store_path = client_buyer_db(data_dir);
let store = RedbBuyerPoolStore::open(data_dir)?;
let load = store.load_all()?;
write_store_listing(args, &store_path, SOURCE_CLIENT_STORE, "", load)
}
fn write_store_listing(
args: &cli::PoolListArgs,
store_path: &Path,
source: &'static str,
provenance: &str,
load: BuyerLoad,
) -> anyhow::Result<()> {
let BuyerLoad {
mut pools,
mut skipped,
} = load;
pools.sort_by_key(|p| p.pool_id);
skipped.sort_unstable();
write_skipped_pools(&mut std::io::stderr().lock(), &skipped)?;
let mut out = std::io::stdout().lock();
if args.json {
let view = PoolListJson {
store: store_path.display().to_string(),
source,
pools: pools.iter().map(PoolJson::from).collect(),
skipped: skipped.iter().map(|p| format!("{p:#x}")).collect(),
};
serde_json::to_writer_pretty(&mut out, &view)?;
writeln!(out)?;
} else {
writeln!(out, "store={}{provenance}", store_path.display())?;
write_pools(&mut out, &pools, &skipped)?;
}
Ok(())
}
async fn list_from_daemon(
args: &cli::PoolListArgs,
config_path: Option<&Path>,
data_dir: &Path,
) -> anyhow::Result<()> {
let buyer_db = node_buyer_db(data_dir);
anyhow::ensure!(
args.timeout_ms > 0,
"--timeout-ms must be > 0 (jsonrpsee treats Duration::ZERO as \
'never' rather than 'sub-millisecond deadline')"
);
let url = crate::commands::node::resolve_admin_url(args.admin_url.as_deref(), config_path)?;
let client = jsonrpsee::http_client::HttpClientBuilder::default()
.request_timeout(std::time::Duration::from_millis(args.timeout_ms))
.build(&url)
.with_context(|| format!("failed to build admin JSON-RPC client for {url}"))?;
let resp: BuyerPoolsResponse = match client.pools().await {
Ok(resp) => resp,
Err(jsonrpsee::core::client::Error::Transport(inner))
if crate::commands::node::is_connection_refused(inner.as_ref()) =>
{
return list_from_stopped_daemon(args, data_dir, &buyer_db, &url, inner.as_ref());
}
Err(err) => {
return Err(unreachable_daemon_error(
args, data_dir, &buyer_db, &url, err,
));
}
};
let mut out = std::io::stdout().lock();
if args.json {
let view = DaemonPoolListJson {
store: buyer_db.display().to_string(),
source: SOURCE_DAEMON,
pools: &resp.pools,
skipped: &resp.skipped,
};
serde_json::to_writer_pretty(&mut out, &view)?;
writeln!(out)?;
} else {
write_skipped_buyer_pools(&mut std::io::stderr().lock(), &resp.skipped)?;
writeln!(
out,
"store={} (read from the running daemon)",
buyer_db.display()
)?;
write_buyer_pools(&mut out, &resp)?;
}
Ok(())
}
fn unreachable_daemon_error(
args: &cli::PoolListArgs,
data_dir: &Path,
buyer_db: &Path,
url: &str,
err: jsonrpsee::core::client::Error,
) -> anyhow::Error {
{
let answered = matches!(err, jsonrpsee::core::client::Error::Call(_));
let classified = crate::commands::node::classify_client_error(url, args.timeout_ms, err);
if answered {
classified.context(format!(
"{} is a decdn-node data dir, so its buyer pools live in {}, which this command \
reads through the daemon — a running daemon holds that file exclusively. The \
daemon answered and refused the read.",
data_dir.display(),
buyer_db.display(),
))
} else {
classified.context(format!(
"{} is a decdn-node data dir, so the pools that matter are in {} — which a \
running daemon holds exclusively and no other process can open. This command \
therefore asks the daemon, and the daemon did not answer. Start decdn-node, or \
pass --data-dir <client dir> to inspect a client store instead.",
data_dir.display(),
buyer_db.display(),
))
}
}
}
fn list_from_stopped_daemon(
args: &cli::PoolListArgs,
data_dir: &Path,
buyer_db: &Path,
url: &str,
transport: &(dyn std::error::Error + Send + Sync + 'static),
) -> anyhow::Result<()> {
match ReadOnlyBuyerPoolStore::open_file(buyer_db) {
Ok(reader) => {
let load = reader.load_all()?;
write_store_listing(
args,
buyer_db,
SOURCE_NODE_STORE_OFFLINE,
" (read from disk; no daemon running)",
load,
)
}
Err(decdn_incentive::StoreError::AlreadyOpen { .. }) => anyhow::bail!(
"admin at {url} refused the connection ({transport}), but a process holds {} \
exclusively — so a decdn-node IS running against {}. The admin URL or admin_port \
is wrong, not the node down. Pass --admin-url, or check admin_port in the node's \
config.",
buyer_db.display(),
data_dir.display(),
),
Err(decdn_incentive::StoreError::NeedsRepair { .. }) => anyhow::bail!(
"admin at {url} refused the connection ({transport}), and {} was not shut down \
cleanly — it needs a repair pass that only a writable open can run. Start \
decdn-node once against {} to repair it, then read it again.",
buyer_db.display(),
data_dir.display(),
),
Err(decdn_incentive::StoreError::Absent { .. }) => anyhow::bail!(
"admin at {url} refused the connection ({transport}), and {} does not exist — this \
node holds no buyer pools on disk. It has not opened one, or the store was deleted \
to force re-adoption. Start decdn-node and read it again, or run `decdn pool list \
--all` to see what this keystore owns on chain.",
buyer_db.display(),
),
Err(store_err) => Err(anyhow::anyhow!(
"admin at {url} refused the connection ({transport}); is the node running, and is \
admin_port configured correctly? Reading {} from disk instead did not work either: \
{store_err}",
buyer_db.display(),
)
.context(format!(
"{} is a decdn-node data dir, so the pools that matter are in {} — not in a client \
store. Pass --data-dir <client dir> to inspect a client store instead.",
data_dir.display(),
buyer_db.display(),
))),
}
}
struct ChainPoolRow {
pool_id: PoolId,
status: PaymentPool::Status,
deposit: U256,
total_redeemed: U256,
dispute_deadline: u64,
tracked: Option<bool>,
}
pub(crate) const fn status_label(status: PaymentPool::Status) -> &'static str {
match status {
PaymentPool::Status::Open => "open",
PaymentPool::Status::Closing => "closing",
PaymentPool::Status::Closed => "closed",
_ => "?",
}
}
fn reclaimable_label(row: &ChainPoolRow, now: u64) -> String {
match plan_reclaim(row.status, row.dispute_deadline, now) {
ReclaimPlan::Reclaim => "now".to_string(),
ReclaimPlan::SkipOpen => "close it first".to_string(),
ReclaimPlan::SkipInWindow(deadline) => format_expiry(deadline, now),
ReclaimPlan::SkipClosed => "already reclaimed".to_string(),
ReclaimPlan::SkipUnknown => "?".to_string(),
}
}
const fn tracked_label(tracked: Option<bool>) -> &'static str {
match tracked {
Some(true) => "yes",
Some(false) => "no",
None => "?",
}
}
fn write_chain_pools(
w: &mut impl Write,
owner: Address,
chain_id: u64,
rows: &[ChainPoolRow],
now: u64,
local_store_read: bool,
) -> std::io::Result<()> {
writeln!(
w,
"owner={owner:#x} chain_id={chain_id} pools={} local_store={}",
rows.len(),
if local_store_read {
"read"
} else {
"unreadable"
},
)?;
if rows.is_empty() {
writeln!(w, "(this keystore owns no pools on this contract)")?;
return Ok(());
}
writeln!(
w,
"{:<14} {:<8} {:>14} {:>14} {:>7} RECLAIMABLE",
"POOL", "STATUS", "DEPOSIT", "REDEEMED", "TRACKED"
)?;
for row in rows {
writeln!(
w,
"{:<14} {:<8} {:>14} {:>14} {:>7} {}",
short_hex(&format!("{:#x}", row.pool_id)),
status_label(row.status),
format_usdc_u256(row.deposit),
format_usdc_u256(row.total_redeemed),
tracked_label(row.tracked),
reclaimable_label(row, now),
)?;
}
Ok(())
}
#[derive(Serialize)]
struct PoolListAllJson {
source: &'static str,
owner: String,
chain_id: u64,
payment_pool: String,
local_store_read: bool,
pools: Vec<ChainPoolJson>,
}
#[derive(Serialize)]
struct ChainPoolJson {
pool_id: String,
status: &'static str,
deposit_micro_usdc: String,
total_redeemed_micro_usdc: String,
dispute_deadline: u64,
reclaimable_now: bool,
tracked: Option<bool>,
}
impl ChainPoolJson {
fn at(row: &ChainPoolRow, now: u64) -> Self {
Self {
pool_id: format!("{:#x}", row.pool_id),
status: status_label(row.status),
deposit_micro_usdc: row.deposit.to_string(),
total_redeemed_micro_usdc: row.total_redeemed.to_string(),
dispute_deadline: row.dispute_deadline,
reclaimable_now: matches!(
plan_reclaim(row.status, row.dispute_deadline, now),
ReclaimPlan::Reclaim
),
tracked: row.tracked,
}
}
}
const SOURCE_CLIENT_STORE: &str = "client_store";
const SOURCE_DAEMON: &str = "daemon";
const SOURCE_NODE_STORE_OFFLINE: &str = "node_store_offline";
const SOURCE_CHAIN: &str = "chain";
fn write_skipped_pools(w: &mut impl Write, skipped: &[PoolId]) -> std::io::Result<()> {
for pool_id in skipped {
writeln!(
w,
"warning: buyer pool {pool_id:#x} could not be decoded; its deposit remains \
escrowed but untracked until the record is repaired"
)?;
}
Ok(())
}
fn write_pools(
w: &mut impl Write,
pools: &[BuyerPoolState],
skipped: &[PoolId],
) -> std::io::Result<()> {
writeln!(w, "pools={}", pools.len())?;
if pools.is_empty() {
if skipped.is_empty() {
writeln!(w, "(no tracked pools)")?;
}
return Ok(());
}
writeln!(
w,
"{:<14} {:<14} {:<14} {:>14} {:>8}",
"POOL", "OWNER", "TOKEN", "DEPOSIT", "LANES"
)?;
for p in pools {
writeln!(
w,
"{:<14} {:<14} {:<14} {:>14} {:>8}",
short_hex(&format!("{:#x}", p.pool_id)),
short_hex(&format!("{:#x}", p.owner)),
short_hex(&format!("{:#x}", p.token)),
format_usdc_u256(p.deposit),
p.lane_count(),
)?;
}
Ok(())
}
pub(crate) fn write_buyer_pools(
w: &mut impl Write,
resp: &BuyerPoolsResponse,
) -> std::io::Result<()> {
writeln!(w, "pools={}", resp.pools.len())?;
if resp.pools.is_empty() {
if resp.skipped.is_empty() {
writeln!(w, "(no tracked pools)")?;
}
return Ok(());
}
writeln!(
w,
"{:<14} {:<14} {:<14} {:>14} {:>8}",
"POOL", "OWNER", "TOKEN", "DEPOSIT", "LANES"
)?;
for p in &resp.pools {
writeln!(
w,
"{:<14} {:<14} {:<14} {:>14} {:>8}",
short_hex(&p.pool_id),
short_hex(&p.owner),
short_hex(&p.token),
format_usdc_u256(U256::from(p.deposit_micro_usdc)),
p.lanes.len(),
)?;
}
Ok(())
}
pub(crate) fn write_skipped_buyer_pools(
w: &mut impl Write,
skipped: &[String],
) -> std::io::Result<()> {
for pool_id in skipped {
writeln!(
w,
"warning: buyer pool {pool_id} could not be decoded; its deposit remains escrowed but \
untracked until the record is repaired"
)?;
}
Ok(())
}
#[derive(Serialize)]
struct PoolListJson {
store: String,
source: &'static str,
pools: Vec<PoolJson>,
skipped: Vec<String>,
}
#[derive(Serialize)]
struct DaemonPoolListJson<'a> {
store: String,
source: &'static str,
pools: &'a [decdn_common::admin::BuyerPoolSnapshot],
skipped: &'a [String],
}
#[derive(Serialize)]
struct PoolJson {
pool_id: String,
chain_id: u64,
payment_pool: String,
owner: String,
token: String,
deposit_usdc: String,
lanes: Vec<LaneJson>,
}
#[derive(Serialize)]
struct LaneJson {
signer: String,
provider: String,
last_amount_usdc: String,
last_bytes_delivered: String,
}
impl From<&BuyerPoolState> for PoolJson {
fn from(p: &BuyerPoolState) -> Self {
Self {
pool_id: format!("{:#x}", p.pool_id),
chain_id: p.deployment.chain_id,
payment_pool: format!("{:#x}", p.deployment.payment_pool),
owner: format!("{:#x}", p.owner),
token: format!("{:#x}", p.token),
deposit_usdc: format_usdc_u256(p.deposit),
lanes: p
.lanes()
.map(|(lane, progress)| LaneJson {
signer: format!("{:#x}", lane.signer),
provider: format!("{:#x}", lane.provider),
last_amount_usdc: format_usdc_u256(progress.last_amount),
last_bytes_delivered: progress.last_bytes.to_string(),
})
.collect(),
}
}
}
fn format_usdc_u256(micro: U256) -> String {
let scale = U256::from(1_000_000u64);
let whole = micro / scale;
let frac = (micro % scale).to::<u64>();
format!("{whole}.{frac:06}")
}
fn short_hex(hex: &str) -> String {
match hex.get(..12) {
Some(prefix) if hex.len() > 12 => format!("{prefix}…"),
_ => hex.to_string(),
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests;