feat(tips): stage 3 — attach-funding + funding-verification worker (money-in)

- attach_funding_to_tip: atomic guarded UPDATE (COALESCE + CASE), idempotent on
  same txid, 400 on a different txid; non-drafts land pending_confirmation so the
  verifier (not the caller) owns the ->pending transition.
- worker (src/tips/confirmation.rs): XMR/WOW confirmation counting via LWS;
  amount verification for BTC/LTC (Electrum full history, confirmed-only) +
  XMR/WOW (classify_funding_amount, separate-then-subtract net). Verified->pending,
  Short->funding_mismatch; Unfunded/Drained/LwsError never mutate status (outage-
  safe). Spawned behind FEATURE_TIPS on a 60s loop w/ catch_unwind guard.
- POST /tips/social/:id/attach-funding endpoint + openapi.
- 5 DB tests + 6 classify unit tests; build clean.
This commit is contained in:
jwinterm
2026-07-10 14:53:46 -04:00
parent f3d6be95b8
commit b63cac19e2
8 changed files with 1058 additions and 0 deletions
+2
View File
@@ -72,6 +72,7 @@ use utoipa::OpenApi;
crate::api::tips::get_sent_social_tips,
crate::api::tips::get_public_social_tip,
crate::api::tips::cancel_social_tip,
crate::api::tips::attach_funding,
),
components(schemas(
crate::api::health::HealthResponse,
@@ -168,6 +169,7 @@ use utoipa::OpenApi;
crate::api::tips::SocialTipsResponse,
crate::api::tips::PublicTipInfo,
crate::api::tips::CancelTipResponse,
crate::api::tips::AttachFundingRequest,
)),
tags(
(name = "system", description = "Service health and metadata."),
+72
View File
@@ -118,6 +118,13 @@ pub struct CancelTipResponse {
pub ok: bool,
}
/// Attach a broadcast funding transaction to a draft tip.
#[derive(Debug, Deserialize, utoipa::ToSchema)]
pub struct AttachFundingRequest {
/// The on-chain funding transaction id (XMR/WOW txid, BTC/LTC txid).
pub funding_txid: String,
}
// ── helpers ──────────────────────────────────────────────────────────────────
/// Reject when tips are not enabled on this instance.
@@ -408,10 +415,75 @@ pub async fn cancel_social_tip(
}
}
/// Attach a broadcast funding tx to a draft tip, advancing it into the funding
/// lifecycle (`pending_confirmation`; the funding worker later flips it to
/// `pending` once the on-chain receipt is confirmed AND covers `amount`).
///
/// Idempotent: re-attaching the SAME txid is a no-op that returns the row; a
/// DIFFERENT txid on a tip that already has one (or a non-draft tip) is a 400;
/// an unknown / not-owned tip is a 404. Response mirrors create: the same
/// `CreateSocialTipResponse { tip_id, status, share_url }`.
#[utoipa::path(
security(("bearer_auth" = [])),
post,
path = "/tips/social/{tip_id}/attach-funding",
params(("tip_id" = String, Path, description = "Tip id")),
request_body = AttachFundingRequest,
responses(
(status = 200, description = "Funding attached", body = CreateSocialTipResponse),
(status = 400, description = "Tips off, empty/oversized txid, or a conflicting funding tx"),
(status = 401, description = "Missing or invalid token"),
(status = 404, description = "No such tip owned by the caller")
),
tag = "tips"
)]
#[instrument(skip(state, headers, req))]
pub async fn attach_funding(
State(state): State<Arc<AppState>>,
headers: HeaderMap,
Path(tip_id): Path<Uuid>,
Json(req): Json<AttachFundingRequest>,
) -> Result<Json<CreateSocialTipResponse>, AppError> {
let user_id = extract_user_id_from_token(&state, &headers).await?;
ensure_tips_enabled(&state)?;
// Validate the txid shape so a malformed value is a clean 400 rather than a
// DB length-overflow 500 (funding_txid is VARCHAR(128)).
let funding_txid = req.funding_txid.trim();
if funding_txid.is_empty() {
return Err(AppError::ValidationError("funding_txid is required.".into()));
}
if funding_txid.len() > 128 {
return Err(AppError::ValidationError("funding_txid is too long.".into()));
}
// The DB fn returns Ok(row) / NotFound / ValidationError; just surface it.
let tip = state
.db
.attach_funding_to_tip(tip_id, user_id, funding_txid)
.await?;
let share_url = state
.config
.tip_share_base
.as_ref()
.map(|base| format!("{}/{}", base.trim_end_matches('/'), tip.id));
Ok(Json(CreateSocialTipResponse {
tip_id: tip.id.to_string(),
status: tip.status,
share_url,
}))
}
pub fn routes() -> Router<Arc<AppState>> {
Router::new()
.route("/tips/social", post(create_social_tip))
.route("/tips/social/sent", get(get_sent_social_tips))
.route("/tips/social/:tip_id/public", get(get_public_social_tip))
.route("/tips/social/:tip_id/cancel", post(cancel_social_tip))
.route(
"/tips/social/:tip_id/attach-funding",
post(attach_funding),
)
}
+242
View File
@@ -184,4 +184,246 @@ impl Database {
.await?;
Ok(cancelled)
}
// ── Stage 3: money-in (attach funding + confirmation/amount verifier) ──────
//
// Every method below is a single guarded `UPDATE ... WHERE ... RETURNING`
// (or a filtered SELECT for the worker). The WHERE guards are the
// money-safety surface: the amount verifier only ever transitions a row it
// still finds in `pending_confirmation` with `funding_amount_verified =
// FALSE`, so an LWS/Electrum outage that re-drives the verifier can never
// double-fire or clobber a claimed/clawed-back row.
/// Attach a broadcast funding tx to a `draft` tip, advancing it into the
/// funding lifecycle. Idempotent: attaching the SAME txid again is a no-op
/// that returns the row; a DIFFERENT txid (or a non-draft row) is a
/// [`AppError::ValidationError`]; an unknown / not-owned id is
/// [`AppError::NotFound`].
///
/// One atomic guarded UPDATE encodes all three legitimate outcomes:
/// 1. `draft` + `funding_txid IS NULL` → attach and flip to
/// `pending_confirmation`.
/// 2. row already carries the same `funding_txid` → the `funding_txid =
/// $3` arm matches, `COALESCE` keeps it, status stays (no-op).
/// 3. anything else → no row, disambiguated by the follow-up SELECT.
///
/// ALL non-draft tips land in `pending_confirmation` (never `pending`), so
/// the funding-amount verifier — never the caller — owns the transition to
/// claimable. `COALESCE(funding_txid, $3)` can never overwrite an existing
/// txid, which is what makes the same-txid retry a true no-op.
#[instrument(skip(self))]
pub async fn attach_funding_to_tip(
&self,
tip_id: Uuid,
sender_user_id: Uuid,
funding_txid: &str,
) -> Result<SocialTipRow, AppError> {
let target_status = TipStatus::PendingConfirmation.as_str();
let result = sqlx::query_as::<_, SocialTipRow>(&format!(
"UPDATE social_tips \
SET funding_txid = COALESCE(funding_txid, $3), \
status = CASE \
WHEN status = 'draft' AND funding_txid IS NULL THEN $4 \
ELSE status END, \
updated_at = CASE \
WHEN status = 'draft' AND funding_txid IS NULL THEN NOW() \
ELSE updated_at END \
WHERE id = $1 \
AND sender_user_id = $2 \
AND ((status = 'draft' AND funding_txid IS NULL) OR funding_txid = $3) \
RETURNING {TIP_COLS}"
))
.bind(tip_id)
.bind(sender_user_id)
.bind(funding_txid)
.bind(target_status)
.fetch_optional(self.pool())
.await?;
if let Some(tip) = result {
return Ok(tip);
}
// The guarded UPDATE matched nothing. A single follow-up SELECT (error
// path only) distinguishes "not your tip / doesn't exist" from "already
// has a different funding_txid" from "wrong status".
let current = sqlx::query_as::<_, SocialTipRow>(&format!(
"SELECT {TIP_COLS} FROM social_tips WHERE id = $1 AND sender_user_id = $2"
))
.bind(tip_id)
.bind(sender_user_id)
.fetch_optional(self.pool())
.await?
.ok_or_else(|| AppError::NotFound(format!("Tip {tip_id} not found or not yours")))?;
if let Some(existing_txid) = current.funding_txid.as_ref() {
if existing_txid != funding_txid {
return Err(AppError::ValidationError(format!(
"Tip {tip_id} already has a different funding_txid attached"
)));
}
// Same txid — a race where the UPDATE matched but a concurrent write
// modified the row between statements. Return the current row as-is.
return Ok(current);
}
Err(AppError::ValidationError(format!(
"Tip {tip_id} is in status {} — funding can only be attached to drafts",
current.status
)))
}
/// Tips of `asset` that still need on-chain confirmation counting: in
/// `pending`/`pending_confirmation`, with a non-zero threshold not yet
/// reached, a funding txid attached, and not checked in the last 30s (per-row
/// rate limit). XMR/WOW only in practice — BTC/LTC have
/// `confirmations_required = 0` and are excluded by the `> 0` guard.
#[instrument(skip(self))]
pub async fn get_tips_pending_confirmation(
&self,
asset: &str,
) -> Result<Vec<SocialTipRow>, AppError> {
let rows = sqlx::query_as::<_, SocialTipRow>(&format!(
"SELECT {TIP_COLS} FROM social_tips \
WHERE status IN ('pending', 'pending_confirmation') \
AND asset = $1 \
AND confirmations_required > 0 \
AND funding_confirmations < confirmations_required \
AND funding_txid IS NOT NULL \
AND (last_confirmation_check IS NULL \
OR last_confirmation_check < NOW() - INTERVAL '30 seconds') \
ORDER BY created_at ASC \
LIMIT 50"
))
.bind(asset)
.fetch_all(self.pool())
.await?;
Ok(rows)
}
/// Record the latest `funding_confirmations` and stamp `funding_confirmed_at`
/// the first time the threshold is met. Deliberately does NOT flip status:
/// the `pending_confirmation` → `pending` transition is owned by the
/// funding-AMOUNT verifier (a tip with enough confirmations but a short
/// receipt must land in `funding_mismatch`, never become claimable).
#[instrument(skip(self))]
pub async fn update_tip_confirmations(
&self,
tip_id: Uuid,
confirmations: i32,
) -> Result<Option<SocialTipRow>, AppError> {
let row = sqlx::query_as::<_, SocialTipRow>(&format!(
"UPDATE social_tips \
SET funding_confirmations = $2, \
last_confirmation_check = NOW(), \
funding_confirmed_at = CASE \
WHEN $2 >= confirmations_required AND funding_confirmed_at IS NULL \
THEN NOW() ELSE funding_confirmed_at END, \
updated_at = NOW() \
WHERE id = $1 \
RETURNING {TIP_COLS}"
))
.bind(tip_id)
.bind(confirmations)
.fetch_optional(self.pool())
.await?;
Ok(row)
}
/// Stamp `last_confirmation_check = NOW()` only, touching nothing else. The
/// verifier calls this on its non-mutating branches (Unfunded / Drained /
/// LwsError) so the 30s per-row rate-limit predicate actually starves a
/// re-polled row during an outage instead of re-hitting it every cycle.
#[instrument(skip(self))]
pub async fn touch_tip_last_check(&self, tip_id: Uuid) -> Result<(), AppError> {
sqlx::query("UPDATE social_tips SET last_confirmation_check = NOW() WHERE id = $1")
.bind(tip_id)
.execute(self.pool())
.await?;
Ok(())
}
/// Tips that have reached their confirmation threshold and await
/// funding-amount verification: still `pending_confirmation`, not yet
/// amount-verified, `funding_confirmations >= confirmations_required`, and
/// past the 30s per-row rate limit. (BTC/LTC with threshold 0 qualify
/// immediately.)
#[instrument(skip(self))]
pub async fn get_tips_awaiting_amount_verification(
&self,
) -> Result<Vec<SocialTipRow>, AppError> {
let rows = sqlx::query_as::<_, SocialTipRow>(&format!(
"SELECT {TIP_COLS} FROM social_tips \
WHERE status = 'pending_confirmation' \
AND funding_amount_verified = FALSE \
AND funding_confirmations >= confirmations_required \
AND (last_confirmation_check IS NULL \
OR last_confirmation_check < NOW() - INTERVAL '30 seconds') \
ORDER BY created_at ASC \
LIMIT 50"
))
.fetch_all(self.pool())
.await?;
Ok(rows)
}
/// Mark funding amount-verified and flip `pending_confirmation` → `pending`
/// (claimable). The `status = 'pending_confirmation' AND
/// funding_amount_verified = FALSE` guard makes this idempotent: a second
/// call (or a row clawed back mid-poll) returns `None` and the caller skips
/// its side effects. `observed_amount` is the on-chain receipt total (>=
/// `amount`), in atomic units.
#[instrument(skip(self))]
pub async fn mark_tip_funding_verified(
&self,
tip_id: Uuid,
observed_amount: i64,
) -> Result<Option<SocialTipRow>, AppError> {
let row = sqlx::query_as::<_, SocialTipRow>(&format!(
"UPDATE social_tips \
SET status = 'pending', \
funding_amount_verified = TRUE, \
funding_amount_observed = $2, \
funding_amount_verified_at = NOW(), \
updated_at = NOW() \
WHERE id = $1 \
AND status = 'pending_confirmation' \
AND funding_amount_verified = FALSE \
RETURNING {TIP_COLS}"
))
.bind(tip_id)
.bind(observed_amount)
.fetch_optional(self.pool())
.await?;
Ok(row)
}
/// Mark a short-funded tip: flip `pending_confirmation` → `funding_mismatch`
/// (never claimable; the sender can claw it back). Same idempotency guard as
/// [`mark_tip_funding_verified`]. `observed_amount` is the on-chain receipt
/// total (< `amount`), in atomic units.
#[instrument(skip(self))]
pub async fn mark_tip_funding_mismatch(
&self,
tip_id: Uuid,
observed_amount: i64,
) -> Result<Option<SocialTipRow>, AppError> {
let row = sqlx::query_as::<_, SocialTipRow>(&format!(
"UPDATE social_tips \
SET status = 'funding_mismatch', \
funding_amount_verified = TRUE, \
funding_amount_observed = $2, \
funding_amount_verified_at = NOW(), \
updated_at = NOW() \
WHERE id = $1 \
AND status = 'pending_confirmation' \
AND funding_amount_verified = FALSE \
RETURNING {TIP_COLS}"
))
.bind(tip_id)
.bind(observed_amount)
.fetch_optional(self.pool())
.await?;
Ok(row)
}
}
+1
View File
@@ -14,6 +14,7 @@ pub mod core;
pub mod error;
pub mod infra;
pub mod models;
pub mod tips;
use std::collections::HashMap;
use std::sync::Arc;
+23
View File
@@ -262,6 +262,29 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
}
}
// Public-tips "money-in" worker: one funding pass per 60s (confirmation
// counting for XMR/WOW, then the amount verifier across all
// pending_confirmation tips). Only runs when tips are enabled. A panic in a
// single pass is caught + backed off so the loop survives (visible in logs)
// rather than silently dying and stranding every pending tip.
if state.config.features.tips {
use futures::FutureExt;
let state = state.clone();
tokio::spawn(async move {
let mut tick = tokio::time::interval(std::time::Duration::from_secs(60));
loop {
tick.tick().await;
let cycle = std::panic::AssertUnwindSafe(
smirk_backend_core::tips::run_tip_confirmation_cycle(state.clone()),
);
if cycle.catch_unwind().await.is_err() {
tracing::error!("tip confirmation cycle panicked; continuing after backoff");
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
}
}
});
}
let app = build_router(state);
let listener = tokio::net::TcpListener::bind(&addr).await?;
+467
View File
@@ -0,0 +1,467 @@
//! Funding confirmation + amount verification for public social tips.
//!
//! Ported (public subset) from the legacy `core/tip_confirmation.rs`. Grin and
//! all DM/announcement side-effects are dropped: public tips deliver by share
//! URL, so the "fire the recipient a notification" branches become no-ops here.
//!
//! Confirmation thresholds (see [`crate::models::tip_status::confirmations_for_asset`]):
//! XMR 10, WOW 4, BTC/LTC 0. BTC/LTC skip the confirmation-count pass entirely
//! and go straight to amount verification.
use std::sync::Arc;
use tracing::{debug, info, instrument, warn};
use crate::error::AppError;
use crate::infra::db::SocialTipRow;
use crate::infra::lws::AddressTxsResponse;
use crate::AppState;
/// CryptoNote assets that require an on-chain confirmation count before the
/// amount pass. BTC/LTC are deliberately absent (`confirmations_required = 0`).
const CONFIRMATION_ASSETS: [&str; 2] = ["xmr", "wow"];
/// Run one full funding pass: confirmation counting for XMR/WOW, then the
/// amount-verification pass across all `pending_confirmation` tips. Never
/// returns an error — every fallible step is logged and swallowed so a single
/// bad tip or a chain outage can't abort the cycle (or, via the spawn loop,
/// stall every other tip).
pub async fn run_tip_confirmation_cycle(state: Arc<AppState>) {
for asset in CONFIRMATION_ASSETS {
if let Err(e) = check_asset_confirmations(&state, asset).await {
warn!(error = %e, asset, "tip confirmation-count pass failed");
}
}
if let Err(e) = process_pending_verifications(&state).await {
warn!(error = %e, "tip funding-amount verification pass failed");
}
}
/// Update the confirmation counter for every funded tip of `asset` awaiting
/// confirmations. Skips silently when the chain is disabled on this instance.
///
/// Money-safety: `update_tip_confirmations` only writes the counter + timestamps
/// (never status), and a daemon error leaves the row entirely untouched (retried
/// next poll) — the count can never regress a tip out of a settled state.
#[instrument(skip(state))]
async fn check_asset_confirmations(state: &AppState, asset: &str) -> Result<(), AppError> {
// Poll-only CryptoNote assets reach the daemon via the LWS client. If the
// chain is disabled here, there's nothing to poll.
let lws = match asset {
"xmr" => state.chains.xmr.as_ref(),
"wow" => state.chains.wow.as_ref(),
_ => None,
};
let Some(lws) = lws else {
return Ok(());
};
let tips = state.db.get_tips_pending_confirmation(asset).await?;
if tips.is_empty() {
debug!(asset, "no tips awaiting confirmation");
return Ok(());
}
info!(asset, count = tips.len(), "checking tip funding confirmations");
for tip in &tips {
let funding_txid = match tip.funding_txid.as_deref() {
Some(txid) if !txid.is_empty() => txid,
_ => continue,
};
match lws.get_transaction_confirmations(funding_txid).await {
Ok(Some(confs)) => {
let confs_i32 = confs.min(i32::MAX as u64) as i32;
if let Err(e) = state.db.update_tip_confirmations(tip.id, confs_i32).await {
warn!(tip_id = %tip.id, error = %e, "failed to update tip confirmations");
}
}
Ok(None) => {
// Tx not found / still in the mempool without a height. Stamp the
// check timestamp (counter stays 0) so we don't re-poll it hot.
if let Err(e) = state.db.update_tip_confirmations(tip.id, 0).await {
warn!(tip_id = %tip.id, error = %e, "failed to stamp tip check timestamp");
}
}
Err(e) => {
// Daemon/LWS error: leave the row untouched for the next poll.
// NEVER mutate status here.
warn!(tip_id = %tip.id, error = %e, "failed to fetch confirmation count");
}
}
}
Ok(())
}
/// Result of checking whether the on-chain receipt at a tip's address covers the
/// sender-declared `amount`. The verifier acts on this; only `Verified` /
/// `Short` mutate status.
#[derive(Debug, PartialEq, Eq)]
pub enum FundingVerification {
/// Confirmed net receipt covers `amount`. Carries the observed total
/// (atomic units) so the row can record what was actually received.
Verified { observed: u64 },
/// Confirmed net receipt is positive but strictly less than `amount`
/// (sender underfunded). Flips the row to `funding_mismatch`.
Short { observed: u64 },
/// Confirmed net receipt is zero (never funded, or only mempool entries so
/// far). Row stays in `pending_confirmation` for the next poll.
Unfunded,
/// The address HAS seen funds (`total_sent > 0`) but currently holds less
/// than `amount` — funded then (partially) drained before the verifier ran
/// (clawback, or a URL holder claimed first). We do NOT announce (funds are
/// gone) and do NOT auto-flip `funding_mismatch` (the sender may have
/// legitimately recovered). Distinguished from `Unfunded` for operator
/// visibility.
Drained { net: u64, total_sent: u64 },
/// LWS / Electrum / upstream errored, or a required field was missing. Row
/// stays in `pending_confirmation` for the next poll. This variant MUST NOT
/// flip a row to `funding_mismatch` — that would mass-mismatch every pending
/// tip during an outage.
LwsError,
}
/// Classify an LWS `get_address_txs` response against `expected` (atomic units).
///
/// Sums `total_received` and `total_sent` SEPARATELY over confirmed
/// (non-mempool, `height > 0`) txs, then signed-subtracts once at the end
/// (`net = received.saturating_sub(sent)`). This separate-then-subtract order is
/// load-bearing: the old per-tx `saturating_sub` counted a fully-swept address
/// (funding tx received=N, sweep tx sent=N) as `Verified` for N, phantom-DMing a
/// recipient about funds the address no longer held. Pure function (no I/O) so
/// it is unit-testable without an LWS client.
pub(crate) fn classify_funding_amount(
txs: &AddressTxsResponse,
expected: i64,
) -> FundingVerification {
let total_received: u64 = txs
.transactions
.iter()
.filter(|tx| !tx.mempool && tx.height > 0)
.map(|tx| tx.total_received)
.sum();
let total_sent: u64 = txs
.transactions
.iter()
.filter(|tx| !tx.mempool && tx.height > 0)
.map(|tx| tx.total_sent)
.sum();
let net = total_received.saturating_sub(total_sent);
// `expected` is always non-negative in practice (create-tip validates
// amount > 0); saturate a defensive negative to 0.
let expected_u64: u64 = if expected < 0 { 0 } else { expected as u64 };
// Drained before Unfunded/Short: the address SAW funds but is now short.
if net < expected_u64 && total_sent > 0 {
return FundingVerification::Drained { net, total_sent };
}
if net == 0 {
return FundingVerification::Unfunded;
}
if net >= expected_u64 {
FundingVerification::Verified { observed: net }
} else {
FundingVerification::Short { observed: net }
}
}
/// Verify that a tip's on-chain funding covers `amount`.
///
/// XMR/WOW: `get_address_txs(tip_address, tip_view_key)` → `classify_funding_amount`.
/// BTC/LTC: the address history, summing received-at vs sent-from over CONFIRMED
/// (`height > 0`) txs via the verbose-tx accessors.
///
/// This never returns an error: EVERY failure path (disabled chain, missing
/// address/view-key, upstream error, a hostile-amount rejection from the
/// verbose-tx converters) maps to [`FundingVerification::LwsError`], so the
/// caller only ever stamps `last_confirmation_check` and retries — status is
/// never mutated on a failure.
async fn verify_funding_amount(state: &AppState, tip: &SocialTipRow) -> FundingVerification {
let asset = tip.asset.to_lowercase();
match asset.as_str() {
"xmr" | "wow" => {
let lws = if asset == "xmr" {
state.chains.xmr.as_ref()
} else {
state.chains.wow.as_ref()
};
let Some(lws) = lws else {
warn!(tip_id = %tip.id, %asset, "verify: LWS client disabled — staying in pending_confirmation");
return FundingVerification::LwsError;
};
let (Some(address), Some(view_key)) =
(tip.tip_address.as_deref(), tip.tip_view_key.as_deref())
else {
warn!(tip_id = %tip.id, %asset, "verify: tip_address/tip_view_key missing — treating as LwsError");
return FundingVerification::LwsError;
};
match lws.get_address_txs(address, view_key).await {
Ok(txs) => classify_funding_amount(&txs, tip.amount),
Err(e) => {
warn!(tip_id = %tip.id, %asset, error = %e, "verify: LWS get_address_txs failed — staying in pending_confirmation");
FundingVerification::LwsError
}
}
}
"btc" | "ltc" => {
let electrum = if asset == "btc" {
state.chains.btc.as_ref()
} else {
state.chains.ltc.as_ref()
};
let Some(electrum) = electrum else {
warn!(tip_id = %tip.id, %asset, "verify: Electrum client disabled — staying in pending_confirmation");
return FundingVerification::LwsError;
};
let Some(address) = tip.tip_address.as_deref() else {
warn!(tip_id = %tip.id, %asset, "verify: tip_address missing — treating as LwsError");
return FundingVerification::LwsError;
};
let history = match electrum.get_history(address).await {
Ok(h) => h,
Err(e) => {
warn!(tip_id = %tip.id, %asset, error = %e, "verify: Electrum get_history failed — staying in pending_confirmation");
return FundingVerification::LwsError;
}
};
// Sum received-at-address and sent-from-address over CONFIRMED
// (height > 0) history entries. A single-use tip address is a pure
// RECIPIENT in its funding tx and a pure SPENDER in its sweep tx;
// gating the sent-sum on `recv == 0` isolates the genuine sweep and
// avoids public-ElectrumX's prevout-less "sent" heuristic misreading
// funding change as a send (a false Drained that would block claims).
let mut total_received: u64 = 0;
let mut total_sent: u64 = 0;
let mut confirmed_seen = 0usize;
for entry in history.iter().filter(|e| e.height > 0) {
confirmed_seen += 1;
let tx = match electrum.get_transaction_verbose(&entry.tx_hash).await {
Ok(tx) => tx,
Err(e) => {
warn!(tip_id = %tip.id, tx_hash = %entry.tx_hash, error = %e, "verify: get_transaction_verbose failed — bailing this tick");
return FundingVerification::LwsError;
}
};
// Core's accessors are Result-returning (they reject hostile /
// non-finite amounts). Any rejection bails the tick as an
// LwsError-tier failure rather than mutating status.
let recv = match tx.total_received_at(address) {
Ok(r) => r,
Err(e) => {
warn!(tip_id = %tip.id, tx_hash = %entry.tx_hash, error = %e, "verify: total_received_at rejected an amount — bailing this tick");
return FundingVerification::LwsError;
}
};
total_received = total_received.saturating_add(recv);
if recv == 0 {
match tx.total_sent_from(address) {
Ok(Some(sent)) => total_sent = total_sent.saturating_add(sent),
Ok(None) => {}
Err(e) => {
warn!(tip_id = %tip.id, tx_hash = %entry.tx_hash, error = %e, "verify: total_sent_from rejected an amount — bailing this tick");
return FundingVerification::LwsError;
}
}
}
}
if confirmed_seen == 0 {
// History empty or all-mempool: no confirmed receipt yet.
return FundingVerification::Unfunded;
}
let net = total_received.saturating_sub(total_sent);
let expected_u64 = if tip.amount < 0 { 0 } else { tip.amount as u64 };
if net < expected_u64 && total_sent > 0 {
FundingVerification::Drained { net, total_sent }
} else if net == 0 {
FundingVerification::Unfunded
} else if net >= expected_u64 {
FundingVerification::Verified { observed: net }
} else {
FundingVerification::Short { observed: net }
}
}
other => {
// Only btc/ltc/xmr/wow can ever create a tip, so this is unreachable
// in practice. Do NOT auto-verify an unknown asset (that would be a
// money-safety hole) — treat it as an error so the row stays put.
warn!(tip_id = %tip.id, asset = other, "verify: unsupported asset — staying in pending_confirmation");
FundingVerification::LwsError
}
}
}
/// Amount-verify every `pending_confirmation` tip that has reached its
/// confirmation threshold.
///
/// Per-tip three-way branch:
/// - `Verified` → `mark_tip_funding_verified` (→ `pending`, claimable).
/// - `Short` → `mark_tip_funding_mismatch` (→ `funding_mismatch`).
/// - `Unfunded` / `Drained` / `LwsError` → NO status mutation; only
/// `touch_tip_last_check` (rate-limit stamp), retried next poll.
///
/// The DB guards (`status = 'pending_confirmation' AND funding_amount_verified =
/// FALSE`) make the `mark_*` transitions idempotent, so a re-driven verifier
/// never double-processes a row.
#[instrument(skip(state))]
async fn process_pending_verifications(state: &AppState) -> Result<(), AppError> {
let tips = state.db.get_tips_awaiting_amount_verification().await?;
if tips.is_empty() {
debug!("no tips awaiting funding-amount verification");
return Ok(());
}
info!(count = tips.len(), "verifying tip funding amounts");
for tip in &tips {
match verify_funding_amount(state, tip).await {
FundingVerification::Verified { observed } => {
let observed_i64 = i64::try_from(observed).unwrap_or(i64::MAX);
match state.db.mark_tip_funding_verified(tip.id, observed_i64).await {
Ok(Some(_updated)) => {
info!(
tip_id = %tip.id, asset = %tip.asset,
declared = tip.amount, observed = observed_i64,
"tip funding amount verified — now claimable"
);
// Public tips deliver by share URL: no recipient DM /
// announcement side-effect to fire here.
}
Ok(None) => debug!(tip_id = %tip.id, "verified: row no longer pending_confirmation — skipping"),
Err(e) => warn!(tip_id = %tip.id, error = %e, "mark_tip_funding_verified DB error"),
}
}
FundingVerification::Short { observed } => {
let observed_i64 = i64::try_from(observed).unwrap_or(i64::MAX);
match state.db.mark_tip_funding_mismatch(tip.id, observed_i64).await {
Ok(Some(_updated)) => warn!(
tip_id = %tip.id, asset = %tip.asset,
declared = tip.amount, observed = observed_i64,
"tip funding mismatch — sender funded LESS than declared"
),
Ok(None) => debug!(tip_id = %tip.id, "mismatch: row no longer pending_confirmation — skipping"),
Err(e) => warn!(tip_id = %tip.id, error = %e, "mark_tip_funding_mismatch DB error"),
}
}
FundingVerification::Unfunded => {
debug!(tip_id = %tip.id, asset = %tip.asset, "verify: Unfunded — staying in pending_confirmation");
if let Err(e) = state.db.touch_tip_last_check(tip.id).await {
debug!(tip_id = %tip.id, error = %e, "touch_tip_last_check failed (Unfunded)");
}
}
FundingVerification::Drained { net, total_sent } => {
warn!(
tip_id = %tip.id, asset = %tip.asset,
declared = tip.amount, net_received = net, total_sent,
"verify: Drained — funded address has been (partially) emptied; not announcing"
);
if let Err(e) = state.db.touch_tip_last_check(tip.id).await {
debug!(tip_id = %tip.id, error = %e, "touch_tip_last_check failed (Drained)");
}
}
FundingVerification::LwsError => {
debug!(tip_id = %tip.id, asset = %tip.asset, "verify: LwsError — staying in pending_confirmation");
if let Err(e) = state.db.touch_tip_last_check(tip.id).await {
debug!(tip_id = %tip.id, error = %e, "touch_tip_last_check failed (LwsError)");
}
}
}
}
Ok(())
}
#[cfg(test)]
mod tests {
//! Unit tests for the pure `classify_funding_amount` branch logic. The async
//! DB-backed paths are covered by `tests/tips.rs` (gated on TEST_DATABASE_URL).
use super::*;
use crate::infra::lws::{AddressTx, AddressTxsResponse};
fn tx(received: u64, sent: u64, height: u64, mempool: bool) -> AddressTx {
AddressTx {
hash: format!("h_{received}_{sent}_{height}"),
height,
timestamp: String::new(),
total_received: received,
total_sent: sent,
mempool,
unlock_time: 0,
payment_id: None,
spent_outputs: Vec::new(),
}
}
fn resp(transactions: Vec<AddressTx>) -> AddressTxsResponse {
AddressTxsResponse {
transactions,
total_received: 0,
scanned_height: 0,
blockchain_height: 0,
}
}
#[test]
fn exact_and_overpay_are_verified() {
assert_eq!(
classify_funding_amount(&resp(vec![tx(1_000, 0, 100, false)]), 1_000),
FundingVerification::Verified { observed: 1_000 }
);
assert_eq!(
classify_funding_amount(&resp(vec![tx(2_000, 0, 100, false)]), 1_000),
FundingVerification::Verified { observed: 2_000 }
);
}
#[test]
fn short_receipt_is_short() {
assert_eq!(
classify_funding_amount(&resp(vec![tx(10, 0, 100, false)]), 1_000),
FundingVerification::Short { observed: 10 }
);
}
#[test]
fn mempool_and_zero_height_do_not_count() {
// Only-mempool → Unfunded.
assert_eq!(
classify_funding_amount(&resp(vec![tx(5_000, 0, 0, true)]), 1_000),
FundingVerification::Unfunded
);
// height == 0 but mempool == false is excluded defensively.
assert_eq!(
classify_funding_amount(&resp(vec![tx(5_000, 0, 0, false)]), 1_000),
FundingVerification::Unfunded
);
// A confirmed short tx plus a huge mempool tx stays Short (mempool excluded).
assert_eq!(
classify_funding_amount(
&resp(vec![tx(100, 0, 100, false), tx(9_999_999, 0, 0, true)]),
1_000
),
FundingVerification::Short { observed: 100 }
);
}
#[test]
fn separate_then_subtract_marks_swept_as_drained() {
// REGRESSION: funding (received=N) + sweep (sent=N) must net to 0 and be
// Drained, not Verified — the separate-then-subtract guarantee.
let r = classify_funding_amount(
&resp(vec![tx(1_000, 0, 100, false), tx(0, 1_000, 110, false)]),
1_000,
);
assert_eq!(r, FundingVerification::Drained { net: 0, total_sent: 1_000 });
}
#[test]
fn no_txs_is_unfunded_not_drained() {
assert_eq!(
classify_funding_amount(&resp(vec![]), 1_000),
FundingVerification::Unfunded
);
}
#[test]
fn lws_error_distinct_from_unfunded() {
assert_ne!(FundingVerification::LwsError, FundingVerification::Unfunded);
}
}
+20
View File
@@ -0,0 +1,20 @@
//! Background funding worker for public social tips (the "money-in" path).
//!
//! One periodic pass ([`run_tip_confirmation_cycle`], spawned from `main` on a
//! 60s interval when `FEATURE_TIPS` is on):
//!
//! 1. Confirmation-count pass for the poll-only CryptoNote assets (XMR/WOW):
//! read the daemon's confirmation count for each funded tip and record it.
//! 2. Amount-verification pass across every `pending_confirmation` tip that
//! has reached its confirmation threshold — check that the on-chain receipt
//! at the tip address actually covers the sender-declared `amount` before
//! flipping the row claimable. BTC/LTC (threshold 0) enter here directly.
//!
//! Money-safety invariant: any upstream failure (LWS/Electrum outage, a hostile
//! response, a missing address/view-key) leaves the row untouched except for the
//! `last_confirmation_check` rate-limit stamp — it NEVER mutates status. An
//! outage can therefore never mass-flip pending tips to `funding_mismatch`.
pub mod confirmation;
pub use confirmation::run_tip_confirmation_cycle;