mirror of
https://github.com/Such-Software/smirk-backend-core.git
synced 2026-10-17 15:31:38 +02:00
feat(tips): stage 4 — claim / clawback / confirm-sweep + sweep reconciler (money-out)
Escrow settlement, audited guards ported verbatim:
- mark_tip_claiming: status IN('pending','claiming') AND confs>=required AND
funding_amount_verified AND sweep_confirmed_at IS NULL (double-claim guard).
- confirm_tip_sweep: RECORD-not-settle, first-recorder-wins (COALESCE +
sweep_txid IS NULL OR =$3) — RBF-steal safe; reconciler owns claiming->claimed.
- confirm_sweep_onchain: SOLE writer of 'claimed', one-way status='claiming' AND
sweep_confirmed_at IS NULL lock; settles only after on-chain sweep confirms to
per-asset depth (btc/ltc 1, wow 4, xmr 10).
- clawback: 5-status set incl 'cancelled', blocked once sweep_confirmed_at set.
- revert_sweep_on_reorg: only from 'claimed', re-opens to 'claiming' + clears txid.
- sweep_reconciler: address-history-driven (not recorded txid), pick_winner
(earliest height, lexi-smallest tie), pg advisory locks on a pinned conn for
single-instance safety; reorg pass reverts only on affirmative vanish, never on
probe error. DMs stubbed, exactly-once stamps kept.
- POST claim/confirm-sweep/clawback endpoints + openapi.
- 16 integration + 12 reconciler unit tests; build clean; SocialTipRow dead_code
allow removed (all fields now read).
This commit is contained in:
jwinterm
@@ -73,6 +73,9 @@ use utoipa::OpenApi;
|
||||
crate::api::tips::get_public_social_tip,
|
||||
crate::api::tips::cancel_social_tip,
|
||||
crate::api::tips::attach_funding,
|
||||
crate::api::tips::claim_social_tip,
|
||||
crate::api::tips::confirm_sweep,
|
||||
crate::api::tips::clawback_social_tip,
|
||||
),
|
||||
components(schemas(
|
||||
crate::api::health::HealthResponse,
|
||||
@@ -170,6 +173,10 @@ use utoipa::OpenApi;
|
||||
crate::api::tips::PublicTipInfo,
|
||||
crate::api::tips::CancelTipResponse,
|
||||
crate::api::tips::AttachFundingRequest,
|
||||
crate::api::tips::ClaimTipResponse,
|
||||
crate::api::tips::ConfirmSweepRequest,
|
||||
crate::api::tips::ConfirmSweepResponse,
|
||||
crate::api::tips::ClawbackTipResponse,
|
||||
)),
|
||||
tags(
|
||||
(name = "system", description = "Service health and metadata."),
|
||||
|
||||
+157
@@ -125,6 +125,37 @@ pub struct AttachFundingRequest {
|
||||
pub funding_txid: String,
|
||||
}
|
||||
|
||||
/// Claim a tip: lock it into `claiming` and receive the encrypted claim key +
|
||||
/// tip address to sweep. Load-bearing INCONSISTENT key: `success` (not `ok`).
|
||||
#[derive(Debug, Serialize, utoipa::ToSchema)]
|
||||
pub struct ClaimTipResponse {
|
||||
pub success: bool,
|
||||
/// Hex-encoded AES-GCM ciphertext of the claim secret; `null` if the tip
|
||||
/// stored none. Useless without the URL-fragment key.
|
||||
pub encrypted_key: Option<String>,
|
||||
pub tip_address: Option<String>,
|
||||
}
|
||||
|
||||
/// Record the claimer's broadcast sweep transaction (first-recorder-wins).
|
||||
#[derive(Debug, Deserialize, utoipa::ToSchema)]
|
||||
pub struct ConfirmSweepRequest {
|
||||
/// The broadcast sweep transaction id.
|
||||
pub sweep_txid: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, utoipa::ToSchema)]
|
||||
pub struct ConfirmSweepResponse {
|
||||
/// The WINNING (first-recorded) sweep txid, or `null` if none is recorded.
|
||||
pub sweep_txid: Option<String>,
|
||||
pub status: String,
|
||||
}
|
||||
|
||||
/// Sender clawback of a tip. Load-bearing INCONSISTENT key: `success` (not `ok`).
|
||||
#[derive(Debug, Serialize, utoipa::ToSchema)]
|
||||
pub struct ClawbackTipResponse {
|
||||
pub success: bool,
|
||||
}
|
||||
|
||||
// ── helpers ──────────────────────────────────────────────────────────────────
|
||||
|
||||
/// Reject when tips are not enabled on this instance.
|
||||
@@ -476,6 +507,129 @@ pub async fn attach_funding(
|
||||
}))
|
||||
}
|
||||
|
||||
/// Claim a public tip: lock it into `claiming` and return the encrypted claim
|
||||
/// key + tip address so the caller can sweep the funds to their own wallet.
|
||||
///
|
||||
/// AUTHED, no body. The `claiming` state is a UX signal, not a cryptographic
|
||||
/// lock — any URL holder may claim while the sweep hasn't confirmed (whoever
|
||||
/// wins the on-chain sweep race wins the tip; the reconciler resolves the
|
||||
/// winner). A non-claimable tip (missing, underfunded, clawed back, already
|
||||
/// swept) is a 400.
|
||||
#[utoipa::path(
|
||||
security(("bearer_auth" = [])),
|
||||
post,
|
||||
path = "/tips/social/{tip_id}/claim",
|
||||
params(("tip_id" = String, Path, description = "Tip id")),
|
||||
responses(
|
||||
(status = 200, description = "Tip locked into claiming", body = ClaimTipResponse),
|
||||
(status = 400, description = "Tips off or tip not claimable"),
|
||||
(status = 401, description = "Missing or invalid token")
|
||||
),
|
||||
tag = "tips"
|
||||
)]
|
||||
#[instrument(skip(state, headers))]
|
||||
pub async fn claim_social_tip(
|
||||
State(state): State<Arc<AppState>>,
|
||||
headers: HeaderMap,
|
||||
Path(tip_id): Path<Uuid>,
|
||||
) -> Result<Json<ClaimTipResponse>, AppError> {
|
||||
let user_id = extract_user_id_from_token(&state, &headers).await?;
|
||||
ensure_tips_enabled(&state)?;
|
||||
|
||||
let row = state
|
||||
.db
|
||||
.mark_tip_claiming(tip_id, user_id)
|
||||
.await?
|
||||
.ok_or_else(|| AppError::ValidationError("tip not claimable".into()))?;
|
||||
|
||||
Ok(Json(ClaimTipResponse {
|
||||
success: true,
|
||||
encrypted_key: row.encrypted_key.map(hex::encode),
|
||||
tip_address: row.tip_address,
|
||||
}))
|
||||
}
|
||||
|
||||
/// Record the claimer's broadcast sweep txid (first-recorder-wins). RECORD not
|
||||
/// settle: the tip stays `claiming` and the reconciler owns the settle to
|
||||
/// `claimed` once the sweep confirms on-chain. Idempotent; a second, different
|
||||
/// txid does NOT overwrite the recorded winner. A tip that was never claimed is
|
||||
/// a 400.
|
||||
#[utoipa::path(
|
||||
security(("bearer_auth" = [])),
|
||||
post,
|
||||
path = "/tips/social/{tip_id}/confirm-sweep",
|
||||
params(("tip_id" = String, Path, description = "Tip id")),
|
||||
request_body = ConfirmSweepRequest,
|
||||
responses(
|
||||
(status = 200, description = "Sweep txid recorded", body = ConfirmSweepResponse),
|
||||
(status = 400, description = "Tips off, empty/oversized txid, or tip not claiming"),
|
||||
(status = 401, description = "Missing or invalid token")
|
||||
),
|
||||
tag = "tips"
|
||||
)]
|
||||
#[instrument(skip(state, headers, req))]
|
||||
pub async fn confirm_sweep(
|
||||
State(state): State<Arc<AppState>>,
|
||||
headers: HeaderMap,
|
||||
Path(tip_id): Path<Uuid>,
|
||||
Json(req): Json<ConfirmSweepRequest>,
|
||||
) -> Result<Json<ConfirmSweepResponse>, AppError> {
|
||||
let user_id = extract_user_id_from_token(&state, &headers).await?;
|
||||
ensure_tips_enabled(&state)?;
|
||||
|
||||
let sweep_txid = req.sweep_txid.trim();
|
||||
if sweep_txid.is_empty() {
|
||||
return Err(AppError::ValidationError("sweep_txid is required.".into()));
|
||||
}
|
||||
if sweep_txid.len() > 128 {
|
||||
return Err(AppError::ValidationError("sweep_txid is too long.".into()));
|
||||
}
|
||||
|
||||
let row = state
|
||||
.db
|
||||
.confirm_tip_sweep(tip_id, user_id, sweep_txid)
|
||||
.await?
|
||||
.ok_or_else(|| AppError::ValidationError("tip not claiming".into()))?;
|
||||
|
||||
Ok(Json(ConfirmSweepResponse {
|
||||
sweep_txid: row.sweep_txid,
|
||||
status: row.status,
|
||||
}))
|
||||
}
|
||||
|
||||
/// Sender reclaims a tip's funds. AUTHED, no body. Succeeds for the sender on a
|
||||
/// pending / pending_confirmation / claiming / funding_mismatch / cancelled tip
|
||||
/// while its sweep hasn't confirmed; a settled or non-clawable tip is a 400.
|
||||
#[utoipa::path(
|
||||
security(("bearer_auth" = [])),
|
||||
post,
|
||||
path = "/tips/social/{tip_id}/clawback",
|
||||
params(("tip_id" = String, Path, description = "Tip id")),
|
||||
responses(
|
||||
(status = 200, description = "Tip clawed back", body = ClawbackTipResponse),
|
||||
(status = 400, description = "Tips off or tip not clawable"),
|
||||
(status = 401, description = "Missing or invalid token")
|
||||
),
|
||||
tag = "tips"
|
||||
)]
|
||||
#[instrument(skip(state, headers))]
|
||||
pub async fn clawback_social_tip(
|
||||
State(state): State<Arc<AppState>>,
|
||||
headers: HeaderMap,
|
||||
Path(tip_id): Path<Uuid>,
|
||||
) -> Result<Json<ClawbackTipResponse>, AppError> {
|
||||
let user_id = extract_user_id_from_token(&state, &headers).await?;
|
||||
ensure_tips_enabled(&state)?;
|
||||
|
||||
state
|
||||
.db
|
||||
.clawback_social_tip(tip_id, user_id)
|
||||
.await?
|
||||
.ok_or_else(|| AppError::ValidationError("tip not clawable".into()))?;
|
||||
|
||||
Ok(Json(ClawbackTipResponse { success: true }))
|
||||
}
|
||||
|
||||
pub fn routes() -> Router<Arc<AppState>> {
|
||||
Router::new()
|
||||
.route("/tips/social", post(create_social_tip))
|
||||
@@ -486,4 +640,7 @@ pub fn routes() -> Router<Arc<AppState>> {
|
||||
"/tips/social/:tip_id/attach-funding",
|
||||
post(attach_funding),
|
||||
)
|
||||
.route("/tips/social/:tip_id/claim", post(claim_social_tip))
|
||||
.route("/tips/social/:tip_id/confirm-sweep", post(confirm_sweep))
|
||||
.route("/tips/social/:tip_id/clawback", post(clawback_social_tip))
|
||||
}
|
||||
|
||||
+355
-3
@@ -29,10 +29,8 @@ pub(crate) const TIP_COLS: &str = "\
|
||||
lws_deactivated_at, created_at, updated_at";
|
||||
|
||||
/// A persisted public social tip. Field names/order match [`TIP_COLS`] (sqlx
|
||||
/// `FromRow` maps by name). Fields not read until later port stages are held
|
||||
/// behind `allow(dead_code)`.
|
||||
/// `FromRow` maps by name).
|
||||
#[derive(Debug, Clone, FromRow)]
|
||||
#[allow(dead_code)]
|
||||
pub struct SocialTipRow {
|
||||
pub id: Uuid,
|
||||
pub sender_user_id: Uuid,
|
||||
@@ -426,4 +424,358 @@ impl Database {
|
||||
.await?;
|
||||
Ok(row)
|
||||
}
|
||||
|
||||
// ── Stage 4: money-out (claim / confirm-sweep / clawback + reconciler) ──────
|
||||
//
|
||||
// The HIGHEST money-safety surface. Every method below is a single guarded
|
||||
// `UPDATE ... WHERE ... RETURNING` and the WHERE guards are non-negotiable —
|
||||
// a wrong conjunct loses funds. Guards ported VERBATIM from the audited
|
||||
// legacy `smirk-backend/src/infra/db/social_tips.rs` (public subset: the
|
||||
// targeted/recipient branches are dropped, `is_public = TRUE` is kept).
|
||||
|
||||
/// Lock a claimable tip into `claiming` and hand the claimer the encrypted
|
||||
/// key + tip address. THE DOUBLE-CLAIM GUARD — every conjunct is
|
||||
/// non-negotiable:
|
||||
/// - `status IN ('pending', 'claiming')` (audit T1): without it a
|
||||
/// `funding_mismatch` / `clawed_back` row would re-pass the UPDATE
|
||||
/// because it still satisfies the confirmation/amount/sweep predicates —
|
||||
/// a URL holder could curl `/claim` on an underfunded tip and sweep the
|
||||
/// short amount before the sender's clawback wins.
|
||||
/// - `funding_confirmations >= confirmations_required AND
|
||||
/// funding_amount_verified = TRUE`: only a fully-funded, amount-verified
|
||||
/// tip is claimable.
|
||||
/// - `sweep_confirmed_at IS NULL`: never re-hand the key on a settled tip.
|
||||
/// - `is_public = TRUE`: public-only port; a public tip's claim state is a
|
||||
/// UX signal, not a cryptographic lock (whoever wins the on-chain sweep
|
||||
/// race wins the tip). `claimed_by_user_id` is recorded for DM routing
|
||||
/// only, carrying no exclusive rights.
|
||||
///
|
||||
/// Returns `None` (not claimable) for a missing / non-claimable / already-
|
||||
/// swept row.
|
||||
#[instrument(skip(self))]
|
||||
pub async fn mark_tip_claiming(
|
||||
&self,
|
||||
tip_id: Uuid,
|
||||
claimed_by_user_id: Uuid,
|
||||
) -> Result<Option<SocialTipRow>, AppError> {
|
||||
let tip = sqlx::query_as::<_, SocialTipRow>(&format!(
|
||||
"UPDATE social_tips \
|
||||
SET status = 'claiming', \
|
||||
claimed_at = NOW(), \
|
||||
claimed_by_user_id = $2, \
|
||||
updated_at = NOW() \
|
||||
WHERE id = $1 \
|
||||
AND status IN ('pending', 'claiming') \
|
||||
AND funding_confirmations >= confirmations_required \
|
||||
AND funding_amount_verified = TRUE \
|
||||
AND sweep_confirmed_at IS NULL \
|
||||
AND is_public = TRUE \
|
||||
RETURNING {TIP_COLS}"
|
||||
))
|
||||
.bind(tip_id)
|
||||
.bind(claimed_by_user_id)
|
||||
.fetch_optional(self.pool())
|
||||
.await?;
|
||||
Ok(tip)
|
||||
}
|
||||
|
||||
/// RECORD (not settle) the claimer's broadcast sweep txid, leaving the row in
|
||||
/// `claiming`. The reconciler — never this call — owns `claiming -> claimed`,
|
||||
/// and only once the sweep CONFIRMS on-chain. This closes the RBF-steal
|
||||
/// false-"claimed" vector: a URL holder who RBFs the recorded sweep can't
|
||||
/// trigger a premature settlement.
|
||||
///
|
||||
/// First-recorder-wins + idempotent: the guard
|
||||
/// `(sweep_txid IS NULL OR sweep_txid = $3)` records the first txid and makes
|
||||
/// a second call with the SAME txid a no-op; a second call with a DIFFERENT
|
||||
/// txid does NOT overwrite (the UPDATE matches nothing) — junk-txid poisoning
|
||||
/// stays blocked. In that case the fallthrough SELECT returns the current
|
||||
/// row (with the WINNING first txid) so the caller renders live state; both
|
||||
/// the first and the different-second call therefore return `Some`.
|
||||
///
|
||||
/// Returns `None` only when the tip doesn't exist or isn't in
|
||||
/// `claiming`/`claimed` (e.g. `/confirm-sweep` on a never-claimed tip).
|
||||
#[instrument(skip(self))]
|
||||
pub async fn confirm_tip_sweep(
|
||||
&self,
|
||||
tip_id: Uuid,
|
||||
claimer_user_id: Uuid,
|
||||
sweep_txid: &str,
|
||||
) -> Result<Option<SocialTipRow>, AppError> {
|
||||
let recorded = sqlx::query_as::<_, SocialTipRow>(&format!(
|
||||
"UPDATE social_tips \
|
||||
SET sweep_txid = COALESCE(sweep_txid, $3), \
|
||||
claimed_by_user_id = COALESCE(claimed_by_user_id, $2), \
|
||||
claimed_at = COALESCE(claimed_at, NOW()), \
|
||||
updated_at = NOW() \
|
||||
WHERE id = $1 \
|
||||
AND status = 'claiming' \
|
||||
AND sweep_confirmed_at IS NULL \
|
||||
AND is_public = TRUE \
|
||||
AND (sweep_txid IS NULL OR sweep_txid = $3) \
|
||||
RETURNING {TIP_COLS}"
|
||||
))
|
||||
.bind(tip_id)
|
||||
.bind(claimer_user_id)
|
||||
.bind(sweep_txid)
|
||||
.fetch_optional(self.pool())
|
||||
.await?;
|
||||
|
||||
if let Some(tip) = recorded {
|
||||
return Ok(Some(tip));
|
||||
}
|
||||
|
||||
// Did not record (already settled, a DIFFERENT txid already recorded, or
|
||||
// not a claiming/claimed row): return the current row so the caller can
|
||||
// render live state — first-recorder's txid wins.
|
||||
let existing = sqlx::query_as::<_, SocialTipRow>(&format!(
|
||||
"SELECT {TIP_COLS} FROM social_tips \
|
||||
WHERE id = $1 AND status IN ('claiming', 'claimed')"
|
||||
))
|
||||
.bind(tip_id)
|
||||
.fetch_optional(self.pool())
|
||||
.await?;
|
||||
Ok(existing)
|
||||
}
|
||||
|
||||
/// Sender reclaims the funds. Guard ported verbatim (evolved across two
|
||||
/// audits):
|
||||
/// - `sender_user_id = $2`: only the sender can claw back.
|
||||
/// - `status IN ('pending', 'pending_confirmation', 'claiming',
|
||||
/// 'funding_mismatch', 'cancelled')`: the 5-status set lets a sender
|
||||
/// recover a tip racing a claimer (`claiming`), an underfunded tip
|
||||
/// (`funding_mismatch`), and a GC-cancelled-but-still-funded draft
|
||||
/// (`cancelled`).
|
||||
/// - `sweep_confirmed_at IS NULL`: never roll back a settled `claimed` row.
|
||||
///
|
||||
/// Returns `None` when no such clawable row (wrong owner, wrong status, or
|
||||
/// already settled).
|
||||
#[instrument(skip(self))]
|
||||
pub async fn clawback_social_tip(
|
||||
&self,
|
||||
tip_id: Uuid,
|
||||
sender_user_id: Uuid,
|
||||
) -> Result<Option<SocialTipRow>, AppError> {
|
||||
let tip = sqlx::query_as::<_, SocialTipRow>(&format!(
|
||||
"UPDATE social_tips \
|
||||
SET status = 'clawed_back', \
|
||||
clawed_back_at = NOW(), \
|
||||
updated_at = NOW() \
|
||||
WHERE id = $1 \
|
||||
AND sender_user_id = $2 \
|
||||
AND status IN ( \
|
||||
'pending', \
|
||||
'pending_confirmation', \
|
||||
'claiming', \
|
||||
'funding_mismatch', \
|
||||
'cancelled' \
|
||||
) \
|
||||
AND sweep_confirmed_at IS NULL \
|
||||
RETURNING {TIP_COLS}"
|
||||
))
|
||||
.bind(tip_id)
|
||||
.bind(sender_user_id)
|
||||
.fetch_optional(self.pool())
|
||||
.await?;
|
||||
Ok(tip)
|
||||
}
|
||||
|
||||
/// The SOLE new writer of `status = 'claimed'`. Settle a tip on-chain after
|
||||
/// its sweep confirmed to the per-asset depth.
|
||||
///
|
||||
/// One-way lock: the `status = 'claiming' AND sweep_confirmed_at IS NULL`
|
||||
/// guard means the FIRST reconciler tick to observe the on-chain winner wins
|
||||
/// the transition; any later tick (or an instance that lost the advisory-lock
|
||||
/// race) gets `None` and fires no side effects.
|
||||
///
|
||||
/// `claimed_by`: `Some` overwrites `claimed_by_user_id`; `None` PRESERVES it
|
||||
/// (the `CASE WHEN $5 IS NOT NULL` guard). This method can only set-or-keep,
|
||||
/// never clear — to NULL the attribution use `clear_claimed_by_on_settle`.
|
||||
#[instrument(skip(self))]
|
||||
pub async fn confirm_sweep_onchain(
|
||||
&self,
|
||||
tip_id: Uuid,
|
||||
winner_txid: &str,
|
||||
block_height: i32,
|
||||
block_hash: Option<&str>,
|
||||
claimed_by: Option<Uuid>,
|
||||
) -> Result<Option<SocialTipRow>, AppError> {
|
||||
let tip = sqlx::query_as::<_, SocialTipRow>(&format!(
|
||||
"UPDATE social_tips \
|
||||
SET status = 'claimed', \
|
||||
sweep_confirmed_at = NOW(), \
|
||||
sweep_txid = $2, \
|
||||
sweep_block_height = $3, \
|
||||
sweep_block_hash = $4, \
|
||||
claimed_by_user_id = CASE \
|
||||
WHEN $5::uuid IS NOT NULL THEN $5 \
|
||||
ELSE claimed_by_user_id END, \
|
||||
updated_at = NOW() \
|
||||
WHERE id = $1 \
|
||||
AND status = 'claiming' \
|
||||
AND sweep_confirmed_at IS NULL \
|
||||
RETURNING {TIP_COLS}"
|
||||
))
|
||||
.bind(tip_id)
|
||||
.bind(winner_txid)
|
||||
.bind(block_height)
|
||||
.bind(block_hash)
|
||||
.bind(claimed_by)
|
||||
.fetch_optional(self.pool())
|
||||
.await?;
|
||||
Ok(tip)
|
||||
}
|
||||
|
||||
/// Settle to `claimed` AND NULL `claimed_by_user_id`. Used when the on-chain
|
||||
/// winner's txid doesn't match the recorded `sweep_txid` (RBF / unknown
|
||||
/// sweeper) or none was recorded — a public tip swept by a URL holder who
|
||||
/// isn't the recorded claimer. Same one-way `claiming -> claimed` lock as
|
||||
/// `confirm_sweep_onchain`; `None` if already settled.
|
||||
#[instrument(skip(self))]
|
||||
pub async fn clear_claimed_by_on_settle(
|
||||
&self,
|
||||
tip_id: Uuid,
|
||||
winner_txid: &str,
|
||||
block_height: i32,
|
||||
block_hash: Option<&str>,
|
||||
) -> Result<Option<SocialTipRow>, AppError> {
|
||||
let tip = sqlx::query_as::<_, SocialTipRow>(&format!(
|
||||
"UPDATE social_tips \
|
||||
SET status = 'claimed', \
|
||||
sweep_confirmed_at = NOW(), \
|
||||
sweep_txid = $2, \
|
||||
sweep_block_height = $3, \
|
||||
sweep_block_hash = $4, \
|
||||
claimed_by_user_id = NULL, \
|
||||
updated_at = NOW() \
|
||||
WHERE id = $1 \
|
||||
AND status = 'claiming' \
|
||||
AND sweep_confirmed_at IS NULL \
|
||||
RETURNING {TIP_COLS}"
|
||||
))
|
||||
.bind(tip_id)
|
||||
.bind(winner_txid)
|
||||
.bind(block_height)
|
||||
.bind(block_hash)
|
||||
.fetch_optional(self.pool())
|
||||
.await?;
|
||||
Ok(tip)
|
||||
}
|
||||
|
||||
/// Revert a settled (`claimed`) tip back to `claiming` after a reorg orphaned
|
||||
/// its recorded sweep. Clearing `sweep_txid` is LOAD-BEARING: the next
|
||||
/// confirm cycle re-scans the address and records the NEW on-chain winner,
|
||||
/// not the orphaned one. Also clears `sweep_confirmed_dm_sent_at` so a fresh
|
||||
/// "claimed" notify can fire once re-confirmation lands. The
|
||||
/// `status = 'claimed' AND sweep_confirmed_at IS NOT NULL` guard is the only
|
||||
/// path off `claimed`; `None` if the row isn't settled.
|
||||
#[instrument(skip(self))]
|
||||
pub async fn revert_sweep_on_reorg(
|
||||
&self,
|
||||
tip_id: Uuid,
|
||||
) -> Result<Option<SocialTipRow>, AppError> {
|
||||
let tip = sqlx::query_as::<_, SocialTipRow>(&format!(
|
||||
"UPDATE social_tips \
|
||||
SET status = 'claiming', \
|
||||
sweep_confirmed_at = NULL, \
|
||||
sweep_txid = NULL, \
|
||||
sweep_block_height = NULL, \
|
||||
sweep_block_hash = NULL, \
|
||||
sweep_confirmed_dm_sent_at = NULL, \
|
||||
updated_at = NOW() \
|
||||
WHERE id = $1 \
|
||||
AND status = 'claimed' \
|
||||
AND sweep_confirmed_at IS NOT NULL \
|
||||
RETURNING {TIP_COLS}"
|
||||
))
|
||||
.bind(tip_id)
|
||||
.fetch_optional(self.pool())
|
||||
.await?;
|
||||
Ok(tip)
|
||||
}
|
||||
|
||||
/// Reconciler poll: `claiming` tips whose sweep hasn't confirmed yet. Includes
|
||||
/// rows with `sweep_txid IS NULL` — the reconciler scans the ADDRESS, not any
|
||||
/// recorded txid, since a public tip's recorded sweep may not be the on-chain
|
||||
/// winner (RBF / race). `funding_amount_verified = TRUE` keeps a
|
||||
/// `funding_mismatch`-adjacent row out. Rate-limited on `last_confirmation_check`
|
||||
/// (30s) so an outage doesn't burn one probe per row per cycle.
|
||||
#[instrument(skip(self))]
|
||||
pub async fn get_tips_awaiting_sweep_confirmation(
|
||||
&self,
|
||||
limit: i64,
|
||||
) -> Result<Vec<SocialTipRow>, AppError> {
|
||||
let tips = sqlx::query_as::<_, SocialTipRow>(&format!(
|
||||
"SELECT {TIP_COLS} FROM social_tips \
|
||||
WHERE status = 'claiming' \
|
||||
AND sweep_confirmed_at IS NULL \
|
||||
AND funding_amount_verified = TRUE \
|
||||
AND (last_confirmation_check IS NULL \
|
||||
OR last_confirmation_check < NOW() - INTERVAL '30 seconds') \
|
||||
ORDER BY claimed_at ASC \
|
||||
LIMIT $1"
|
||||
))
|
||||
.bind(limit)
|
||||
.fetch_all(self.pool())
|
||||
.await?;
|
||||
Ok(tips)
|
||||
}
|
||||
|
||||
/// Reorg-check poll: settled tips the reconciler settled ITSELF (they carry a
|
||||
/// `sweep_block_height` witness) and hasn't re-probed in the last 30s. Rows
|
||||
/// with `sweep_block_height IS NULL` are excluded on purpose (no recorded
|
||||
/// height to reorg-check against).
|
||||
#[instrument(skip(self))]
|
||||
pub async fn get_settled_tips_for_reorg_check(
|
||||
&self,
|
||||
limit: i64,
|
||||
) -> Result<Vec<SocialTipRow>, AppError> {
|
||||
let tips = sqlx::query_as::<_, SocialTipRow>(&format!(
|
||||
"SELECT {TIP_COLS} FROM social_tips \
|
||||
WHERE status = 'claimed' \
|
||||
AND sweep_confirmed_at IS NOT NULL \
|
||||
AND sweep_block_height IS NOT NULL \
|
||||
AND (last_confirmation_check IS NULL \
|
||||
OR last_confirmation_check < NOW() - INTERVAL '30 seconds') \
|
||||
ORDER BY sweep_confirmed_at ASC \
|
||||
LIMIT $1"
|
||||
))
|
||||
.bind(limit)
|
||||
.fetch_all(self.pool())
|
||||
.await?;
|
||||
Ok(tips)
|
||||
}
|
||||
|
||||
/// Exactly-once stamp for the reconciler-owned "tip claimed" notify. The
|
||||
/// `WHERE ... IS NULL` guard makes it idempotent so a retried cycle never
|
||||
/// re-stamps. (The notify itself is stubbed to a no-op in this port, but the
|
||||
/// stamp is preserved so re-enabling notifications stays exactly-once.)
|
||||
#[instrument(skip(self))]
|
||||
pub async fn stamp_sweep_confirmed_dm_sent(&self, tip_id: Uuid) -> Result<(), AppError> {
|
||||
sqlx::query(
|
||||
"UPDATE social_tips \
|
||||
SET sweep_confirmed_dm_sent_at = NOW(), updated_at = NOW() \
|
||||
WHERE id = $1 AND sweep_confirmed_dm_sent_at IS NULL",
|
||||
)
|
||||
.bind(tip_id)
|
||||
.execute(self.pool())
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Exactly-once stamp for the reconciler-owned "claim reversed by reorg"
|
||||
/// notify. Same `WHERE ... IS NULL` idempotency as
|
||||
/// `stamp_sweep_confirmed_dm_sent`.
|
||||
#[instrument(skip(self))]
|
||||
pub async fn stamp_reorg_notified(&self, tip_id: Uuid) -> Result<(), AppError> {
|
||||
sqlx::query(
|
||||
"UPDATE social_tips \
|
||||
SET reorg_notified_at = NOW(), updated_at = NOW() \
|
||||
WHERE id = $1 AND reorg_notified_at IS NULL",
|
||||
)
|
||||
.bind(tip_id)
|
||||
.execute(self.pool())
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
+45
@@ -285,6 +285,51 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
});
|
||||
}
|
||||
|
||||
// Public-tips "money-out" reconciler: the settle-on-sweep-confirmation pass
|
||||
// (probe each claiming tip's ADDRESS, settle claiming -> claimed once the
|
||||
// sweep confirms to the per-asset depth). Sole new writer of 'claimed'. Runs
|
||||
// an immediate startup pass (so a restart doesn't wait a full interval before
|
||||
// settling anything that confirmed while we were down) then every 60s. Same
|
||||
// catch_unwind guard as the money-in worker.
|
||||
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_sweep_reconcile_cycle(state.clone()),
|
||||
);
|
||||
if cycle.catch_unwind().await.is_err() {
|
||||
tracing::error!("sweep reconcile cycle panicked; continuing after backoff");
|
||||
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// Public-tips reorg-check pass: re-probe settled tips and revert any whose
|
||||
// recorded sweep was orphaned by a chain reorg. Lowest-frequency, novel, and
|
||||
// recoverable, so it runs on a slower cadence (300s) than the confirm pass.
|
||||
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(300));
|
||||
loop {
|
||||
tick.tick().await;
|
||||
let cycle = std::panic::AssertUnwindSafe(
|
||||
smirk_backend_core::tips::run_sweep_reorg_cycle(state.clone()),
|
||||
);
|
||||
if cycle.catch_unwind().await.is_err() {
|
||||
tracing::error!("sweep reorg 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?;
|
||||
|
||||
@@ -16,5 +16,7 @@
|
||||
//! outage can therefore never mass-flip pending tips to `funding_mismatch`.
|
||||
|
||||
pub mod confirmation;
|
||||
pub mod sweep_reconciler;
|
||||
|
||||
pub use confirmation::run_tip_confirmation_cycle;
|
||||
pub use sweep_reconciler::{run_sweep_reconcile_cycle, run_sweep_reorg_cycle};
|
||||
|
||||
@@ -0,0 +1,643 @@
|
||||
//! Settle-on-sweep-confirmation reconciler for public social tips (the
|
||||
//! "money-out" path). Ported (public subset) from the legacy
|
||||
//! `smirk-backend/src/core/sweep_reconciler.rs`.
|
||||
//!
|
||||
//! Watches `claiming` tips and settles them to `claimed` ONLY after the sweep
|
||||
//! transaction confirms on-chain to the per-asset depth. It is the sole new
|
||||
//! writer of `status = 'claimed'` (via [`Database::confirm_sweep_onchain`] /
|
||||
//! [`Database::clear_claimed_by_on_settle`]) and the sole writer of the reorg
|
||||
//! revert ([`Database::revert_sweep_on_reorg`]).
|
||||
//!
|
||||
//! ## Source of truth = the ADDRESS, not the recorded `sweep_txid`
|
||||
//!
|
||||
//! Public tips carry the spend key in the URL fragment, so multiple claimers can
|
||||
//! race; only one sweep wins on-chain (and a claimer may RBF-replace its own
|
||||
//! sweep). So the source of truth is the tip ADDRESS's confirmed spend history:
|
||||
//! we probe the address, collect confirmed spends, and pick the deterministic
|
||||
//! winner ([`pick_winner`]).
|
||||
//!
|
||||
//! ## Confirmation thresholds (mirror the funding thresholds)
|
||||
//! BTC/LTC 1, WOW 4, XMR 10. Grin is OUT of this port.
|
||||
//!
|
||||
//! ## Money-safety invariants
|
||||
//! - A probe ERROR (Electrum/LWS unreachable, hostile amount) NEVER settles or
|
||||
//! reverts — it only stamps `last_confirmation_check` and retries.
|
||||
//! - Settle requires an affirmative confirmed spend at the required depth.
|
||||
//! - Reorg-revert requires an affirmative "the recorded sweep is GONE from
|
||||
//! confirmed history"; a probe error can never trigger it.
|
||||
//!
|
||||
//! ## Notifications
|
||||
//! The sender-DM side-effects are STUBBED to no-ops in this port (public tips
|
||||
//! deliver by share URL), but the exactly-once stamp writes
|
||||
//! (`stamp_sweep_confirmed_dm_sent` / `stamp_reorg_notified`) are preserved so
|
||||
//! re-enabling notifications later stays exactly-once.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use tracing::{debug, info, instrument, warn};
|
||||
|
||||
use crate::error::AppError;
|
||||
use crate::infra::db::SocialTipRow;
|
||||
use crate::infra::electrum::VerboseTransaction;
|
||||
use crate::AppState;
|
||||
|
||||
/// Max rows processed per cycle (per pass).
|
||||
const BATCH_SIZE: i64 = 50;
|
||||
|
||||
/// Advisory-lock key for the confirm/settle cycle. A fixed, arbitrary 64-bit
|
||||
/// constant — single-instance guard so two replicas don't both probe + settle
|
||||
/// the same batch. Distinct from the reorg key.
|
||||
const SWEEP_CONFIRM_ADVISORY_LOCK_KEY: i64 = 0x53EE_D0C1;
|
||||
/// Advisory-lock key for the reorg-check cycle (distinct from the confirm cycle
|
||||
/// so the two passes run independently).
|
||||
const SWEEP_REORG_ADVISORY_LOCK_KEY: i64 = 0x53EE_D0C2;
|
||||
|
||||
/// A confirmed spend of the tip address, distilled to what winner-determination
|
||||
/// needs. Pure data, no I/O.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct ConfirmedSpend {
|
||||
/// On-chain identifier of the spend (BTC/LTC + XMR/WOW tx hash).
|
||||
pub txid: String,
|
||||
/// Block height the spend confirmed at.
|
||||
pub block_height: i64,
|
||||
/// Block hash at that height when trivially available (BTC/LTC don't fetch
|
||||
/// the merkle proof, so usually `None`).
|
||||
pub block_hash: Option<String>,
|
||||
}
|
||||
|
||||
/// Per-asset sweep confirmation threshold (confirmations the winning spend needs
|
||||
/// before we settle to `claimed`). Mirrors the funding-side thresholds. Unknown
|
||||
/// assets default to the most conservative (10) so a misconfigured asset never
|
||||
/// settles early. Grin is not supported by this port.
|
||||
pub(crate) fn sweep_confirm_threshold(asset: &str) -> u64 {
|
||||
match asset.to_lowercase().as_str() {
|
||||
"btc" | "ltc" => 1,
|
||||
"wow" => 4,
|
||||
"xmr" => 10,
|
||||
_ => 10,
|
||||
}
|
||||
}
|
||||
|
||||
/// Pick the canonical winning spend: EARLIEST `block_height`, ties broken by the
|
||||
/// lexicographically smallest `txid`. Deterministic + idempotent, so a re-probed
|
||||
/// address always settles to the same winner. `None` for an empty slice.
|
||||
pub(crate) fn pick_winner(spends: &[ConfirmedSpend]) -> Option<&ConfirmedSpend> {
|
||||
spends.iter().min_by(|a, b| {
|
||||
a.block_height
|
||||
.cmp(&b.block_height)
|
||||
.then_with(|| a.txid.cmp(&b.txid))
|
||||
})
|
||||
}
|
||||
|
||||
/// Is this verbose BTC/LTC tx a SPEND of `address`? A single-use tip address is
|
||||
/// a pure RECIPIENT in its funding tx and a pure SPENDER only in its sweep, so a
|
||||
/// tx is a spend of the address iff the address received nothing in it
|
||||
/// (`total_received_at(address) == 0`). Same `recv == 0` isolation the funding
|
||||
/// verifier uses to avoid misreading a funding tx's change output as a send.
|
||||
///
|
||||
/// Returns `Err` when core's Result-returning accessor rejects a hostile /
|
||||
/// non-finite amount — the caller bails the probe (never settles) on that.
|
||||
pub(crate) fn is_verbose_tx_spend_of(
|
||||
tx: &VerboseTransaction,
|
||||
address: &str,
|
||||
) -> Result<bool, AppError> {
|
||||
Ok(tx.total_received_at(address)? == 0)
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Advisory-lock single-instance guard
|
||||
// ============================================================================
|
||||
|
||||
/// Acquire a session-scoped Postgres advisory lock on a PINNED connection. pg
|
||||
/// advisory locks are connection(session)-scoped, so the lock MUST be acquired
|
||||
/// and released on the SAME connection — releasing on a different pooled
|
||||
/// connection leaks the lock. Returns the held connection on success (the caller
|
||||
/// keeps it alive for the cycle, then passes it to `release_advisory_lock`), or
|
||||
/// `None` if another holder has it (skip the cycle).
|
||||
async fn acquire_advisory_lock(
|
||||
state: &AppState,
|
||||
key: i64,
|
||||
) -> Result<Option<sqlx::pool::PoolConnection<sqlx::Postgres>>, AppError> {
|
||||
let mut conn = state.db.pool().acquire().await?;
|
||||
let acquired: bool = sqlx::query_scalar("SELECT pg_try_advisory_lock($1)")
|
||||
.bind(key)
|
||||
.fetch_one(&mut *conn)
|
||||
.await?;
|
||||
if acquired {
|
||||
Ok(Some(conn))
|
||||
} else {
|
||||
Ok(None)
|
||||
}
|
||||
}
|
||||
|
||||
/// Release the advisory lock on the SAME connection it was acquired on, then
|
||||
/// drop the connection back to the pool. Best-effort — the lock auto-releases
|
||||
/// when the connection closes.
|
||||
async fn release_advisory_lock(
|
||||
mut conn: sqlx::pool::PoolConnection<sqlx::Postgres>,
|
||||
key: i64,
|
||||
) {
|
||||
if let Err(e) = sqlx::query("SELECT pg_advisory_unlock($1)")
|
||||
.bind(key)
|
||||
.execute(&mut *conn)
|
||||
.await
|
||||
{
|
||||
debug!(error = %e, key, "pg_advisory_unlock failed (lock auto-releases on conn close)");
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Confirm cycle
|
||||
// ============================================================================
|
||||
|
||||
/// One confirm/settle cycle: probe each awaiting tip's address, settle the
|
||||
/// winner once it reaches the per-asset threshold. Guarded by a Postgres
|
||||
/// advisory lock so only one instance processes a batch. Never returns an
|
||||
/// error to the caller — a single tip's probe/settle failure is logged and
|
||||
/// swallowed so it can't tank the batch (or the spawn loop).
|
||||
pub async fn run_sweep_reconcile_cycle(state: Arc<AppState>) {
|
||||
if let Err(e) = run_confirm_cycle(&state).await {
|
||||
warn!(error = %e, "sweep confirm cycle failed");
|
||||
}
|
||||
}
|
||||
|
||||
#[instrument(skip(state))]
|
||||
async fn run_confirm_cycle(state: &AppState) -> Result<(), AppError> {
|
||||
let Some(lock_conn) = acquire_advisory_lock(state, SWEEP_CONFIRM_ADVISORY_LOCK_KEY).await? else {
|
||||
debug!("sweep confirm cycle: advisory lock held by another instance — skipping");
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
let result = confirm_cycle_inner(state).await;
|
||||
|
||||
release_advisory_lock(lock_conn, SWEEP_CONFIRM_ADVISORY_LOCK_KEY).await;
|
||||
result
|
||||
}
|
||||
|
||||
async fn confirm_cycle_inner(state: &AppState) -> Result<(), AppError> {
|
||||
let tips = state.db.get_tips_awaiting_sweep_confirmation(BATCH_SIZE).await?;
|
||||
if tips.is_empty() {
|
||||
debug!("no tips awaiting sweep confirmation");
|
||||
return Ok(());
|
||||
}
|
||||
info!(count = tips.len(), "reconciling tip sweeps");
|
||||
|
||||
for tip in &tips {
|
||||
if let Err(e) = process_awaiting_tip(state, tip).await {
|
||||
// A single tip's probe/settle error must not tank the batch.
|
||||
warn!(tip_id = %tip.id, error = %e, "process_awaiting_tip failed");
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Process a single `claiming` tip: probe its address, pick the winner, require
|
||||
/// the per-asset depth, settle, stamp the (stubbed) notify.
|
||||
async fn process_awaiting_tip(state: &AppState, tip: &SocialTipRow) -> Result<(), AppError> {
|
||||
let asset = tip.asset.to_lowercase();
|
||||
|
||||
// Probe the address for confirmed spends + the current chain tip.
|
||||
let (spends, tip_height) = match probe_confirmed_spends(state, tip).await {
|
||||
Ok(v) => v,
|
||||
Err(e) => {
|
||||
// Probe error (chain unreachable / disabled / hostile response):
|
||||
// NEVER settle. Stamp last_check so the rate-limit backs off, skip.
|
||||
warn!(tip_id = %tip.id, %asset, error = %e, "sweep probe failed — skipping (retry next cycle)");
|
||||
if let Err(e2) = state.db.touch_tip_last_check(tip.id).await {
|
||||
debug!(tip_id = %tip.id, error = %e2, "touch_tip_last_check failed after probe error");
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
|
||||
let winner = match pick_winner(&spends) {
|
||||
Some(w) => w.clone(),
|
||||
None => {
|
||||
// No confirmed spend of the address yet — sweep not mined, or still
|
||||
// in mempool. Back off via last_check and retry.
|
||||
debug!(tip_id = %tip.id, %asset, "no confirmed sweep yet");
|
||||
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 (no winner)");
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
|
||||
// Threshold gate: confirmations = tip_height - winner_height + 1.
|
||||
let confirmations = tip_height
|
||||
.saturating_sub(winner.block_height)
|
||||
.saturating_add(1);
|
||||
let needed = sweep_confirm_threshold(&asset);
|
||||
if (confirmations as u64) < needed {
|
||||
debug!(tip_id = %tip.id, %asset, confirmations, needed, "sweep seen on-chain, waiting for depth");
|
||||
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 (below threshold)");
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
// Attribution (public-only): if the on-chain winner matches the recorded
|
||||
// sweep_txid, keep claimed_by (via confirm_sweep_onchain with claimed_by =
|
||||
// None). Otherwise the tip was swept by a URL holder who isn't the recorded
|
||||
// claimer (RBF / unknown sweeper / no recorded txid) — settle the funds-moved
|
||||
// fact but drop the now-unreliable attribution.
|
||||
let recorded = tip.sweep_txid.as_deref();
|
||||
let txid_matches = recorded == Some(winner.txid.as_str());
|
||||
let block_hash = winner.block_hash.as_deref();
|
||||
|
||||
let settled = if txid_matches {
|
||||
state
|
||||
.db
|
||||
.confirm_sweep_onchain(tip.id, &winner.txid, winner.block_height as i32, block_hash, None)
|
||||
.await?
|
||||
} else {
|
||||
debug!(
|
||||
tip_id = %tip.id, %asset,
|
||||
recorded_sweep_txid = ?recorded, onchain_winner_txid = %winner.txid,
|
||||
"public tip on-chain winner differs from recorded sweep — clearing claimed_by"
|
||||
);
|
||||
state
|
||||
.db
|
||||
.clear_claimed_by_on_settle(tip.id, &winner.txid, winner.block_height as i32, block_hash)
|
||||
.await?
|
||||
};
|
||||
|
||||
let settled = match settled {
|
||||
Some(row) => row,
|
||||
None => {
|
||||
// Another tick (or instance) already settled this row — side effects
|
||||
// belong to whoever won the UPDATE.
|
||||
debug!(tip_id = %tip.id, "sweep already settled by another tick — skipping side effects");
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
|
||||
info!(
|
||||
tip_id = %settled.id, %asset,
|
||||
winner_txid = %winner.txid, block_height = winner.block_height, confirmations,
|
||||
"tip sweep confirmed on-chain — settled to claimed"
|
||||
);
|
||||
|
||||
// Notify side-effect is stubbed to a no-op in this port; keep the
|
||||
// exactly-once stamp so re-enabling it later stays exactly-once.
|
||||
send_sweep_confirmed_dm_if_needed(state, &settled).await;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Per-asset on-chain spend probe (source of truth = the ADDRESS)
|
||||
// ============================================================================
|
||||
|
||||
/// Probe the tip address for confirmed spends and the current chain tip height.
|
||||
/// Returns `(spends, tip_height)`. Any upstream error / disabled chain / missing
|
||||
/// field is `Err` — the caller treats that as "skip this row, retry next cycle"
|
||||
/// (never settle / revert on an error).
|
||||
async fn probe_confirmed_spends(
|
||||
state: &AppState,
|
||||
tip: &SocialTipRow,
|
||||
) -> Result<(Vec<ConfirmedSpend>, i64), AppError> {
|
||||
let asset = tip.asset.to_lowercase();
|
||||
match asset.as_str() {
|
||||
"btc" | "ltc" => probe_electrum_spends(state, tip, &asset).await,
|
||||
"xmr" | "wow" => probe_lws_spends(state, tip, &asset).await,
|
||||
other => {
|
||||
// Grin is out of this port; only btc/ltc/xmr/wow ever create a tip.
|
||||
warn!(tip_id = %tip.id, asset = %other, "sweep probe: unsupported asset");
|
||||
Err(AppError::ValidationError(format!(
|
||||
"sweep probe: unsupported asset {other}"
|
||||
)))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// BTC/LTC probe via Electrum. For each confirmed history entry (`height > 0`)
|
||||
/// fetch the verbose tx and treat it as a SPEND iff the address received nothing
|
||||
/// in it. `block_hash` stays `None` (no merkle fetch); reorg detection keys on
|
||||
/// the txid vanishing from confirmed history, not the hash.
|
||||
async fn probe_electrum_spends(
|
||||
state: &AppState,
|
||||
tip: &SocialTipRow,
|
||||
asset: &str,
|
||||
) -> Result<(Vec<ConfirmedSpend>, i64), AppError> {
|
||||
let electrum = if asset == "btc" {
|
||||
state.chains.btc.as_ref()
|
||||
} else {
|
||||
state.chains.ltc.as_ref()
|
||||
}
|
||||
.ok_or_else(|| AppError::NodeError(format!("{asset} Electrum client disabled")))?;
|
||||
|
||||
let address = tip
|
||||
.tip_address
|
||||
.as_deref()
|
||||
.ok_or_else(|| AppError::ValidationError("tip_address missing for Electrum probe".into()))?;
|
||||
|
||||
let history = electrum.get_history(address).await?;
|
||||
// Chain height comes from ELECTRUM, not a local node (the backend runs none).
|
||||
let tip_height = electrum.get_tip_height().await?;
|
||||
|
||||
let mut spends = Vec::new();
|
||||
for entry in history.iter().filter(|e| e.height > 0) {
|
||||
let tx = electrum.get_transaction_verbose(&entry.tx_hash).await?;
|
||||
if is_verbose_tx_spend_of(&tx, address)? {
|
||||
spends.push(ConfirmedSpend {
|
||||
txid: entry.tx_hash.clone(),
|
||||
block_height: entry.height,
|
||||
block_hash: None,
|
||||
});
|
||||
}
|
||||
}
|
||||
Ok((spends, tip_height))
|
||||
}
|
||||
|
||||
/// XMR/WOW probe via LWS `get_address_txs`. A SPEND is a confirmed (`height > 0`,
|
||||
/// not mempool) tx with `total_sent > 0 && total_received == 0` (the single-use
|
||||
/// tip address only ever spends in its sweep). `tip_height` comes from the LWS
|
||||
/// response's `blockchain_height`, falling back to the node height if it's 0.
|
||||
async fn probe_lws_spends(
|
||||
state: &AppState,
|
||||
tip: &SocialTipRow,
|
||||
asset: &str,
|
||||
) -> Result<(Vec<ConfirmedSpend>, i64), AppError> {
|
||||
let lws = if asset == "xmr" {
|
||||
state.chains.xmr.as_ref()
|
||||
} else {
|
||||
state.chains.wow.as_ref()
|
||||
}
|
||||
.ok_or_else(|| AppError::NodeError(format!("{asset} LWS client disabled")))?;
|
||||
|
||||
let address = tip
|
||||
.tip_address
|
||||
.as_deref()
|
||||
.ok_or_else(|| AppError::ValidationError("tip_address missing for LWS probe".into()))?;
|
||||
let view_key = tip
|
||||
.tip_view_key
|
||||
.as_deref()
|
||||
.ok_or_else(|| AppError::ValidationError("tip_view_key missing for LWS probe".into()))?;
|
||||
|
||||
let txs = lws.get_address_txs(address, view_key).await?;
|
||||
|
||||
let tip_height = if txs.blockchain_height > 0 {
|
||||
txs.blockchain_height as i64
|
||||
} else {
|
||||
lws.get_blockchain_height().await? as i64
|
||||
};
|
||||
|
||||
let spends: Vec<ConfirmedSpend> = txs
|
||||
.transactions
|
||||
.iter()
|
||||
.filter(|t| !t.mempool && t.height > 0 && t.total_sent > 0 && t.total_received == 0)
|
||||
.map(|t| ConfirmedSpend {
|
||||
txid: t.hash.clone(),
|
||||
block_height: t.height as i64,
|
||||
block_hash: None,
|
||||
})
|
||||
.collect();
|
||||
|
||||
Ok((spends, tip_height))
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Reorg cycle
|
||||
// ============================================================================
|
||||
|
||||
/// One reorg-check cycle: re-probe settled tips; if a recorded sweep is no longer
|
||||
/// a confirmed spend of the address, revert to `claiming`. Guarded by its own
|
||||
/// advisory lock. Runs on a slower cadence than the confirm cycle (the reorg
|
||||
/// path is the lowest-frequency one and reverting is recoverable).
|
||||
pub async fn run_sweep_reorg_cycle(state: Arc<AppState>) {
|
||||
if let Err(e) = run_reorg_cycle(&state).await {
|
||||
warn!(error = %e, "sweep reorg cycle failed");
|
||||
}
|
||||
}
|
||||
|
||||
#[instrument(skip(state))]
|
||||
async fn run_reorg_cycle(state: &AppState) -> Result<(), AppError> {
|
||||
let Some(lock_conn) = acquire_advisory_lock(state, SWEEP_REORG_ADVISORY_LOCK_KEY).await? else {
|
||||
debug!("sweep reorg cycle: advisory lock held by another instance — skipping");
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
let result = reorg_cycle_inner(state).await;
|
||||
|
||||
release_advisory_lock(lock_conn, SWEEP_REORG_ADVISORY_LOCK_KEY).await;
|
||||
result
|
||||
}
|
||||
|
||||
async fn reorg_cycle_inner(state: &AppState) -> Result<(), AppError> {
|
||||
let tips = state.db.get_settled_tips_for_reorg_check(BATCH_SIZE).await?;
|
||||
if tips.is_empty() {
|
||||
debug!("no settled tips to reorg-check");
|
||||
return Ok(());
|
||||
}
|
||||
debug!(count = tips.len(), "reorg-checking settled tip sweeps");
|
||||
|
||||
for tip in &tips {
|
||||
if let Err(e) = process_settled_tip(state, tip).await {
|
||||
warn!(tip_id = %tip.id, error = %e, "process_settled_tip failed");
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Re-probe a settled tip. If its recorded `sweep_txid` is still a confirmed
|
||||
/// spend of the address, do nothing (stamp). If it VANISHED (orphaned by a
|
||||
/// reorg), revert to `claiming`. A probe ERROR never reverts.
|
||||
async fn process_settled_tip(state: &AppState, tip: &SocialTipRow) -> Result<(), AppError> {
|
||||
let recorded_txid = match tip.sweep_txid.as_deref() {
|
||||
Some(t) => t,
|
||||
None => {
|
||||
// Settled with no recorded txid — nothing to reorg-check against.
|
||||
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 (no recorded txid)");
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
|
||||
let (spends, _tip_height) = match probe_confirmed_spends(state, tip).await {
|
||||
Ok(v) => v,
|
||||
Err(e) => {
|
||||
// NEVER revert on a probe error. Back off.
|
||||
warn!(tip_id = %tip.id, error = %e, "reorg re-probe failed — skipping (will retry)");
|
||||
if let Err(e2) = state.db.touch_tip_last_check(tip.id).await {
|
||||
debug!(tip_id = %tip.id, error = %e2, "touch_tip_last_check failed after reorg probe error");
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
|
||||
// Still confirmed iff the recorded txid appears among the confirmed spends.
|
||||
// Absence = orphaned by a reorg.
|
||||
let still_confirmed = spends.iter().any(|s| s.txid == recorded_txid);
|
||||
if still_confirmed {
|
||||
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 (still confirmed)");
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
warn!(
|
||||
tip_id = %tip.id, asset = %tip.asset, recorded_sweep_txid = %recorded_txid,
|
||||
"recorded sweep no longer confirmed — reverting claim (chain reorg)"
|
||||
);
|
||||
|
||||
match state.db.revert_sweep_on_reorg(tip.id).await? {
|
||||
Some(reverted) => send_reorg_dm_if_needed(state, &reverted).await,
|
||||
None => debug!(tip_id = %tip.id, "reorg revert no-op (already reverted) — skipping notify"),
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Notify side-effects (STUBBED to no-ops; exactly-once stamps preserved)
|
||||
// ============================================================================
|
||||
|
||||
/// Sender "your tip was claimed" notify. STUBBED to a no-op in this port (public
|
||||
/// tips deliver by share URL), but we still stamp `sweep_confirmed_dm_sent_at`
|
||||
/// (guarded `IS NULL`) so re-enabling notifications later stays exactly-once.
|
||||
async fn send_sweep_confirmed_dm_if_needed(state: &AppState, tip: &SocialTipRow) {
|
||||
if tip.sweep_confirmed_dm_sent_at.is_some() {
|
||||
return;
|
||||
}
|
||||
// TODO(notify): send the sender DM here when notifications are re-enabled.
|
||||
if let Err(e) = state.db.stamp_sweep_confirmed_dm_sent(tip.id).await {
|
||||
debug!(tip_id = %tip.id, error = %e, "stamp_sweep_confirmed_dm_sent failed");
|
||||
}
|
||||
}
|
||||
|
||||
/// Sender "claim reversed by reorg" notify. STUBBED to a no-op; stamps
|
||||
/// `reorg_notified_at` (guarded `IS NULL`) to keep exactly-once semantics.
|
||||
async fn send_reorg_dm_if_needed(state: &AppState, tip: &SocialTipRow) {
|
||||
if tip.reorg_notified_at.is_some() {
|
||||
return;
|
||||
}
|
||||
// TODO(notify): send the sender reorg DM here when notifications are re-enabled.
|
||||
if let Err(e) = state.db.stamp_reorg_notified(tip.id).await {
|
||||
debug!(tip_id = %tip.id, error = %e, "stamp_reorg_notified failed");
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::infra::electrum::{ScriptPubKeyInfo, VerboseTransaction, VerboseVin, VerboseVout};
|
||||
|
||||
fn spend(txid: &str, height: i64) -> ConfirmedSpend {
|
||||
ConfirmedSpend {
|
||||
txid: txid.to_string(),
|
||||
block_height: height,
|
||||
block_hash: None,
|
||||
}
|
||||
}
|
||||
|
||||
// ---- pick_winner -------------------------------------------------
|
||||
|
||||
#[test]
|
||||
fn pick_winner_empty_is_none() {
|
||||
let spends: Vec<ConfirmedSpend> = vec![];
|
||||
assert!(pick_winner(&spends).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn pick_winner_earliest_height_wins() {
|
||||
let spends = vec![spend("zzz", 110), spend("aaa", 100), spend("mmm", 105)];
|
||||
let w = pick_winner(&spends).unwrap();
|
||||
assert_eq!(w.txid, "aaa");
|
||||
assert_eq!(w.block_height, 100);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn pick_winner_same_height_smallest_txid_tiebreak() {
|
||||
let spends = vec![spend("ffff", 100), spend("aaaa", 100), spend("cccc", 100)];
|
||||
assert_eq!(pick_winner(&spends).unwrap().txid, "aaaa");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn pick_winner_is_deterministic_across_input_order() {
|
||||
let a = vec![spend("bbb", 100), spend("aaa", 100)];
|
||||
let b = vec![spend("aaa", 100), spend("bbb", 100)];
|
||||
assert_eq!(pick_winner(&a).unwrap().txid, pick_winner(&b).unwrap().txid);
|
||||
assert_eq!(pick_winner(&a).unwrap().txid, "aaa");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn pick_winner_height_beats_txid() {
|
||||
let spends = vec![spend("aaa", 200), spend("zzz", 100)];
|
||||
assert_eq!(pick_winner(&spends).unwrap().txid, "zzz");
|
||||
}
|
||||
|
||||
// ---- sweep_confirm_threshold -------------------------------------
|
||||
|
||||
#[test]
|
||||
fn threshold_btc_ltc_is_one() {
|
||||
assert_eq!(sweep_confirm_threshold("btc"), 1);
|
||||
assert_eq!(sweep_confirm_threshold("ltc"), 1);
|
||||
assert_eq!(sweep_confirm_threshold("BTC"), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn threshold_wow_is_four() {
|
||||
assert_eq!(sweep_confirm_threshold("wow"), 4);
|
||||
assert_eq!(sweep_confirm_threshold("WOW"), 4);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn threshold_xmr_is_ten() {
|
||||
assert_eq!(sweep_confirm_threshold("xmr"), 10);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn threshold_unknown_defaults_conservative() {
|
||||
assert_eq!(sweep_confirm_threshold("doge"), 10);
|
||||
}
|
||||
|
||||
// ---- is_verbose_tx_spend_of --------------------------------------
|
||||
|
||||
fn spk(addr: &str) -> ScriptPubKeyInfo {
|
||||
ScriptPubKeyInfo {
|
||||
address: Some(addr.to_string()),
|
||||
addresses: None,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn funding_tx_is_not_a_spend_of_tip_address() {
|
||||
// Funding tx: tip address RECEIVES; recv != 0 → not a spend.
|
||||
let tx = VerboseTransaction {
|
||||
txid: "fund".into(),
|
||||
vin: vec![VerboseVin { txid: Some("alice_prev".into()), vout: Some(0), prevout: None }],
|
||||
vout: vec![
|
||||
VerboseVout { value: 0.001, n: 0, script_pub_key: spk("ltc1qtip") },
|
||||
VerboseVout { value: 0.296, n: 1, script_pub_key: spk("ltc1qalice") },
|
||||
],
|
||||
};
|
||||
assert!(!is_verbose_tx_spend_of(&tx, "ltc1qtip").unwrap());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sweep_tx_is_a_spend_of_tip_address() {
|
||||
// Sweep: tip address absent from outputs (recv == 0) → spend.
|
||||
let tx = VerboseTransaction {
|
||||
txid: "sweep".into(),
|
||||
vin: vec![VerboseVin { txid: Some("fund".into()), vout: Some(0), prevout: None }],
|
||||
vout: vec![VerboseVout { value: 0.00099, n: 0, script_pub_key: spk("ltc1qbob") }],
|
||||
};
|
||||
assert!(is_verbose_tx_spend_of(&tx, "ltc1qtip").unwrap());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn incoming_payment_with_change_is_not_a_spend() {
|
||||
// Tip address receives with the payer's change to a third party —
|
||||
// recv != 0 → not a spend (guards the funding-with-change case).
|
||||
let tx = VerboseTransaction {
|
||||
txid: "pay".into(),
|
||||
vin: vec![VerboseVin { txid: Some("payer_prev".into()), vout: Some(0), prevout: None }],
|
||||
vout: vec![
|
||||
VerboseVout { value: 0.5, n: 0, script_pub_key: spk("ltc1qtip") },
|
||||
VerboseVout { value: 0.3, n: 1, script_pub_key: spk("ltc1qpayer") },
|
||||
],
|
||||
};
|
||||
assert!(!is_verbose_tx_spend_of(&tx, "ltc1qtip").unwrap());
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user