mirror of
https://github.com/Such-Software/grin-lws.git
synced 2026-10-17 15:41:43 +02:00
Phase 3: PMMR-index background scanner (fills the store, serves balances)
Replace the stub scanner with the real incremental algorithm: - discover: page get_unspent_outputs (with proofs) from the cursor to the tip, rewind each new output against caught-up accounts, store matches with their derivation path + a derived lock_height (coinbase maturity = 1440) - spend reconcile: mark any stored-unspent commit the node no longer lists (grin has no per-block inputs; spends = absence from the set) - reorg: detect via header-hash mismatch at the checkpoint height, roll back a bounded depth, reseek from the fork - backfill: bring a newly-registered account up from its birthday index - cursor only advances after discover AND reconcile succeed DB: chain_cursor gains output_mmr_index (tip height/hash become the reorg checkpoint); new helpers (list_accounts, all_unspent_commits, mark_unspent, account_unlocked_totals, scan-height advance/clamp). Read path: get_balance adds `unlocked` (spendable-now) and get_unspent_outs flags per-output `spendable`, so immature coinbase / locked outputs are never presented as spendable.
This commit is contained in:
jwinterm
@@ -49,10 +49,16 @@ CREATE INDEX IF NOT EXISTS idx_outputs_account_spent ON outputs (rewind_hash, sp
|
||||
-- Spend detection matches an on-chain input commitment against stored outputs.
|
||||
CREATE INDEX IF NOT EXISTS idx_outputs_commit ON outputs ("commit");
|
||||
|
||||
-- Single-row chain cursor for reorg-safe resume. On a reorg, roll back rows
|
||||
-- above the fork height and reset the cursor to the fork point.
|
||||
-- Single-row scanner cursor. Grin is scanned by walking the output PMMR in
|
||||
-- insertion-index order, so the resume point is an output index, not a block
|
||||
-- position. `height`/`block_hash` hold the chain tip last seen (the reorg-
|
||||
-- detection checkpoint). On a reorg the scanner rolls back rows above the fork
|
||||
-- and reseeks from there.
|
||||
CREATE TABLE IF NOT EXISTS chain_cursor (
|
||||
id INTEGER PRIMARY KEY CHECK (id = 1),
|
||||
height BIGINT NOT NULL,
|
||||
block_hash TEXT NOT NULL
|
||||
id INTEGER PRIMARY KEY CHECK (id = 1),
|
||||
-- Highest output PMMR index already scanned (forward-scan resume point).
|
||||
output_mmr_index BIGINT NOT NULL DEFAULT 0,
|
||||
-- Chain tip height + block hash at the last tick (reorg checkpoint).
|
||||
height BIGINT NOT NULL,
|
||||
block_hash TEXT NOT NULL
|
||||
);
|
||||
|
||||
+13
-3
@@ -53,8 +53,11 @@ pub struct RewindHashRequest {
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub struct BalanceResponse {
|
||||
/// Total unspent (nanogrin).
|
||||
/// Total unspent (nanogrin), including immature/locked outputs.
|
||||
pub total: u64,
|
||||
/// Spendable-now total: unspent AND unlocked at the current tip (mature
|
||||
/// coinbase, past any lock_height). Never overstates what a spend can use.
|
||||
pub unlocked: u64,
|
||||
/// Number of unspent outputs.
|
||||
pub count: u64,
|
||||
/// How far the scanner has processed this account.
|
||||
@@ -73,6 +76,9 @@ pub struct UnspentOut {
|
||||
pub mmr_index: u64,
|
||||
pub is_coinbase: bool,
|
||||
pub lock_height: u64,
|
||||
/// Spendable at the current tip (`lock_height <= blockchain_height`). A
|
||||
/// client must not use a non-spendable output as a transaction input.
|
||||
pub spendable: bool,
|
||||
pub key_id: Option<String>,
|
||||
pub n_child: Option<u32>,
|
||||
}
|
||||
@@ -160,13 +166,16 @@ pub async fn get_balance(
|
||||
Json(req): Json<RewindHashRequest>,
|
||||
) -> Result<Json<BalanceResponse>> {
|
||||
validate_hex64(&req.rewind_hash, "rewind_hash")?;
|
||||
let blockchain_height = state.node.get_tip_height().await?;
|
||||
let (total, count) = db::account_totals(&state.pool, &req.rewind_hash).await?;
|
||||
let (unlocked, _) =
|
||||
db::account_unlocked_totals(&state.pool, &req.rewind_hash, blockchain_height as i64).await?;
|
||||
let scanned_height = db::account_scan_height(&state.pool, &req.rewind_hash)
|
||||
.await?
|
||||
.unwrap_or(0);
|
||||
let blockchain_height = state.node.get_tip_height().await?;
|
||||
Ok(Json(BalanceResponse {
|
||||
total,
|
||||
unlocked,
|
||||
count,
|
||||
scanned_height,
|
||||
blockchain_height,
|
||||
@@ -179,8 +188,8 @@ pub async fn get_unspent_outs(
|
||||
Json(req): Json<RewindHashRequest>,
|
||||
) -> Result<Json<UnspentOutsResponse>> {
|
||||
validate_hex64(&req.rewind_hash, "rewind_hash")?;
|
||||
let stored = db::unspent_outputs(&state.pool, &req.rewind_hash).await?;
|
||||
let blockchain_height = state.node.get_tip_height().await?;
|
||||
let stored = db::unspent_outputs(&state.pool, &req.rewind_hash).await?;
|
||||
let outputs = stored
|
||||
.into_iter()
|
||||
.map(|o| UnspentOut {
|
||||
@@ -189,6 +198,7 @@ pub async fn get_unspent_outs(
|
||||
height: o.height,
|
||||
mmr_index: o.mmr_index,
|
||||
is_coinbase: o.is_coinbase,
|
||||
spendable: o.lock_height <= blockchain_height,
|
||||
lock_height: o.lock_height,
|
||||
key_id: o.key_id,
|
||||
n_child: o.n_child,
|
||||
|
||||
@@ -38,11 +38,24 @@ pub struct StoredOutput {
|
||||
pub n_child: Option<u32>,
|
||||
}
|
||||
|
||||
/// The scanner's resume point.
|
||||
/// The scanner's resume point: how far through the output PMMR it has scanned,
|
||||
/// plus the chain tip it last saw (for reorg detection).
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ChainCursor {
|
||||
pub height: u64,
|
||||
pub block_hash: String,
|
||||
/// Highest output PMMR index already processed (the forward-scan resume point).
|
||||
pub output_mmr_index: u64,
|
||||
/// Chain tip height at the last tick (reorg checkpoint).
|
||||
pub tip_height: u64,
|
||||
/// Chain tip block hash at the last tick (reorg detection probe).
|
||||
pub tip_hash: String,
|
||||
}
|
||||
|
||||
/// A registered account, for the scanner's per-account loop.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct AccountRow {
|
||||
pub rewind_hash: String,
|
||||
pub start_height: u64,
|
||||
pub scan_height: u64,
|
||||
}
|
||||
|
||||
/// Connect (lazily) to the database. Does not require a live DB at construction;
|
||||
@@ -74,7 +87,7 @@ pub async fn register_account(pool: &AnyPool, rewind_hash: &str, start_height: i
|
||||
Ok(res.rows_affected() > 0)
|
||||
}
|
||||
|
||||
/// All registered accounts' rewind_hashes (the scanner iterates these per block).
|
||||
/// All registered accounts' rewind_hashes (the admin `/list_accounts` view).
|
||||
pub async fn list_account_hashes(pool: &AnyPool) -> Result<Vec<String>> {
|
||||
let rows = sqlx::query("SELECT rewind_hash FROM accounts")
|
||||
.fetch_all(pool)
|
||||
@@ -82,6 +95,57 @@ pub async fn list_account_hashes(pool: &AnyPool) -> Result<Vec<String>> {
|
||||
Ok(rows.iter().map(|r| r.get::<String, _>("rewind_hash")).collect())
|
||||
}
|
||||
|
||||
/// All registered accounts with their scan progress (the scanner's per-account
|
||||
/// loop iterates these).
|
||||
pub async fn list_accounts(pool: &AnyPool) -> Result<Vec<AccountRow>> {
|
||||
let rows = sqlx::query("SELECT rewind_hash, start_height, scan_height FROM accounts")
|
||||
.fetch_all(pool)
|
||||
.await?;
|
||||
Ok(rows
|
||||
.iter()
|
||||
.map(|r| AccountRow {
|
||||
rewind_hash: r.get::<String, _>("rewind_hash"),
|
||||
start_height: r.get::<i64, _>("start_height").max(0) as u64,
|
||||
scan_height: r.get::<i64, _>("scan_height").max(0) as u64,
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
/// Advance the scan height of already-caught-up accounts (scan_height at/above
|
||||
/// the previous tip) to the new tip after a forward scan pass.
|
||||
pub async fn advance_caught_up_scan_heights(
|
||||
pool: &AnyPool,
|
||||
to_height: i64,
|
||||
from_height: i64,
|
||||
) -> Result<()> {
|
||||
sqlx::query("UPDATE accounts SET scan_height = $1 WHERE scan_height >= $2")
|
||||
.bind(to_height)
|
||||
.bind(from_height)
|
||||
.execute(pool)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Set one account's scan height (used after its backfill completes).
|
||||
pub async fn set_account_scan_height(pool: &AnyPool, rewind_hash: &str, height: i64) -> Result<()> {
|
||||
sqlx::query("UPDATE accounts SET scan_height = $1 WHERE rewind_hash = $2")
|
||||
.bind(height)
|
||||
.bind(rewind_hash)
|
||||
.execute(pool)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Clamp any account whose scan height sits above `fork_height` back down to it
|
||||
/// (a reorg rolled the chain back below their recorded progress).
|
||||
pub async fn clamp_account_scan_heights(pool: &AnyPool, fork_height: i64) -> Result<()> {
|
||||
sqlx::query("UPDATE accounts SET scan_height = $1 WHERE scan_height > $1")
|
||||
.bind(fork_height)
|
||||
.execute(pool)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Lower an account's scan progress to `height` for a backwards rescan (also
|
||||
/// clears its stored outputs at/above `height` so they are re-derived). SAFETY:
|
||||
/// backwards-only, mirroring the monero-lws rescan invariant.
|
||||
@@ -128,7 +192,7 @@ pub async fn insert_output(pool: &AnyPool, out: &StoredOutput, rewind_hash: &str
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Mark an output spent when its commitment appears as a tx input on-chain.
|
||||
/// Mark an output spent (its commitment left the node's unspent set).
|
||||
pub async fn mark_spent(pool: &AnyPool, commit: &str, spent_height: i64) -> Result<()> {
|
||||
sqlx::query("UPDATE outputs SET spent = true, spent_height = $2 WHERE \"commit\" = $1")
|
||||
.bind(commit)
|
||||
@@ -138,6 +202,25 @@ pub async fn mark_spent(pool: &AnyPool, commit: &str, spent_height: i64) -> Resu
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Un-mark a spend (a reorg brought a previously-spent output back into the
|
||||
/// unspent set). Self-heals an over-eager spend reconcile.
|
||||
pub async fn mark_unspent(pool: &AnyPool, commit: &str) -> Result<()> {
|
||||
sqlx::query("UPDATE outputs SET spent = false, spent_height = NULL WHERE \"commit\" = $1")
|
||||
.bind(commit)
|
||||
.execute(pool)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Every currently-unspent commitment across all accounts — the set the scanner
|
||||
/// re-checks against the node to detect spends by absence.
|
||||
pub async fn all_unspent_commits(pool: &AnyPool) -> Result<Vec<String>> {
|
||||
let rows = sqlx::query("SELECT \"commit\" FROM outputs WHERE spent = false")
|
||||
.fetch_all(pool)
|
||||
.await?;
|
||||
Ok(rows.iter().map(|r| r.get::<String, _>("commit")).collect())
|
||||
}
|
||||
|
||||
/// Unspent outputs for an account — the spendable set, WITH derivation paths.
|
||||
pub async fn unspent_outputs(pool: &AnyPool, rewind_hash: &str) -> Result<Vec<StoredOutput>> {
|
||||
let rows = sqlx::query(
|
||||
@@ -165,6 +248,28 @@ pub async fn account_totals(pool: &AnyPool, rewind_hash: &str) -> Result<(u64, u
|
||||
Ok((total.max(0) as u64, n.max(0) as u64))
|
||||
}
|
||||
|
||||
/// (unlocked total, count) for an account: unspent AND spendable at `tip_height`
|
||||
/// (`lock_height <= tip`). Immature coinbase / time-locked outputs are excluded,
|
||||
/// so this never reports coins a spend would reject. Mirrors monero-lws's
|
||||
/// unlocked-vs-total split.
|
||||
pub async fn account_unlocked_totals(
|
||||
pool: &AnyPool,
|
||||
rewind_hash: &str,
|
||||
tip_height: i64,
|
||||
) -> Result<(u64, u64)> {
|
||||
let row = sqlx::query(
|
||||
"SELECT COALESCE(SUM(value),0) AS total, COUNT(*) AS n \
|
||||
FROM outputs WHERE rewind_hash = $1 AND spent = false AND lock_height <= $2",
|
||||
)
|
||||
.bind(rewind_hash)
|
||||
.bind(tip_height)
|
||||
.fetch_one(pool)
|
||||
.await?;
|
||||
let total: i64 = row.try_get("total").unwrap_or(0);
|
||||
let n: i64 = row.try_get("n").unwrap_or(0);
|
||||
Ok((total.max(0) as u64, n.max(0) as u64))
|
||||
}
|
||||
|
||||
/// The height at which the scanner last processed this account.
|
||||
pub async fn account_scan_height(pool: &AnyPool, rewind_hash: &str) -> Result<Option<u64>> {
|
||||
let row = sqlx::query("SELECT scan_height FROM accounts WHERE rewind_hash = $1")
|
||||
@@ -193,25 +298,34 @@ fn row_to_output(r: &sqlx::any::AnyRow) -> StoredOutput {
|
||||
|
||||
// ── chain cursor (reorg-safe resume) ───────────────────────────────────────────
|
||||
|
||||
/// Read the scanner's global cursor (last block processed across all accounts).
|
||||
/// Read the scanner's global cursor (output-PMMR position + last-seen tip).
|
||||
pub async fn get_cursor(pool: &AnyPool) -> Result<Option<ChainCursor>> {
|
||||
let row = sqlx::query("SELECT height, block_hash FROM chain_cursor WHERE id = 1")
|
||||
let row = sqlx::query("SELECT output_mmr_index, height, block_hash FROM chain_cursor WHERE id = 1")
|
||||
.fetch_optional(pool)
|
||||
.await?;
|
||||
Ok(row.map(|r| ChainCursor {
|
||||
height: r.get::<i64, _>("height").max(0) as u64,
|
||||
block_hash: r.get::<String, _>("block_hash"),
|
||||
output_mmr_index: r.get::<i64, _>("output_mmr_index").max(0) as u64,
|
||||
tip_height: r.get::<i64, _>("height").max(0) as u64,
|
||||
tip_hash: r.get::<String, _>("block_hash"),
|
||||
}))
|
||||
}
|
||||
|
||||
/// Advance (or initialize) the scanner cursor.
|
||||
pub async fn set_cursor(pool: &AnyPool, height: i64, block_hash: &str) -> Result<()> {
|
||||
/// Advance (or initialize) the scanner cursor: the highest output PMMR index
|
||||
/// processed, plus the tip height + hash at that point (`height` / `block_hash`
|
||||
/// columns now hold the tip checkpoint, not a per-block position).
|
||||
pub async fn set_cursor(
|
||||
pool: &AnyPool,
|
||||
output_mmr_index: i64,
|
||||
tip_height: i64,
|
||||
tip_hash: &str,
|
||||
) -> Result<()> {
|
||||
sqlx::query(
|
||||
"INSERT INTO chain_cursor (id, height, block_hash) VALUES (1, $1, $2) \
|
||||
ON CONFLICT (id) DO UPDATE SET height = $1, block_hash = $2",
|
||||
"INSERT INTO chain_cursor (id, output_mmr_index, height, block_hash) VALUES (1, $1, $2, $3) \
|
||||
ON CONFLICT (id) DO UPDATE SET output_mmr_index = $1, height = $2, block_hash = $3",
|
||||
)
|
||||
.bind(height)
|
||||
.bind(block_hash)
|
||||
.bind(output_mmr_index)
|
||||
.bind(tip_height)
|
||||
.bind(tip_hash)
|
||||
.execute(pool)
|
||||
.await?;
|
||||
Ok(())
|
||||
|
||||
+212
-50
@@ -1,43 +1,53 @@
|
||||
//! Background chain scanner — the heart of a real light-wallet-server, and the
|
||||
//! bulk of the remaining work.
|
||||
//! Background chain scanner — the heart of a real light-wallet-server.
|
||||
//!
|
||||
//! It is a long-lived task that follows the grin chain tip and, for every new
|
||||
//! block, does two things per registered account:
|
||||
//! 1. CREATE: rewind each block output's rangeproof against the account's
|
||||
//! `rewind_hash`. On a match, persist the output WITH its recovered
|
||||
//! derivation path (so the client can spend directly — no client-side
|
||||
//! identify search).
|
||||
//! 2. SPEND: for each input commitment in the block, if it matches a stored
|
||||
//! output, mark that output spent at this height.
|
||||
//! It maintains a reorg-safe [`chain_cursor`](crate::db): on a reorg (a block
|
||||
//! whose `prev_hash` does not match our cursor), roll back rows above the fork
|
||||
//! and resume from the fork point.
|
||||
//! Grin's chain state is the set of unspent outputs in a Merkle Mountain Range.
|
||||
//! Scanning = walking that output set in PMMR insertion-index order, rewinding
|
||||
//! each output against every registered `rewind_hash` (view credential) and
|
||||
//! storing the matches WITH their recovered derivation path. It is incremental:
|
||||
//! a global cursor records the highest output index already processed, so each
|
||||
//! tick only rewinds outputs added since last time.
|
||||
//!
|
||||
//! STATUS: SCAFFOLD. The loop structure, cursor handling, and DB seams are real;
|
||||
//! the per-output rewind ([`crate::grin::rewind_output`]) and block parsing
|
||||
//! ([`crate::grin::GrinNode::get_block`]) are stubbed. Until this lands, the
|
||||
//! deployed on-demand `/wallet/grin/scan` proxy serves balance for grin's small
|
||||
//! UTXO set and the client uses a client-side identify helper for spend paths.
|
||||
//! Each `tick()` runs, against the shared cursor:
|
||||
//! 0. REORG check — if the chain's block at our checkpoint height no longer
|
||||
//! matches the hash we recorded, roll back and reseek.
|
||||
//! 1. DISCOVER — page new outputs (index > cursor) via `get_unspent_outputs`
|
||||
//! (with proofs), rewind against caught-up accounts, insert matches.
|
||||
//! 2. SPEND reconcile — a stored output whose commit is no longer in the
|
||||
//! node's unspent set has been spent; mark it. (Grin has no per-block input
|
||||
//! list to scan; spends are detected by absence.)
|
||||
//! 3. ADVANCE the cursor + caught-up accounts.
|
||||
//! 4. BACKFILL one pending new account (birthday → cursor) against its hash.
|
||||
//!
|
||||
//! Idempotent throughout (`insert_output`/`mark_spent` use ON CONFLICT), so a
|
||||
//! failed tick simply retries: the cursor is only advanced after discovery AND
|
||||
//! spend-reconcile both succeed, so balances never advance past unreconciled
|
||||
//! state.
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use sqlx::AnyPool;
|
||||
|
||||
use crate::config::Config;
|
||||
use crate::db::{self, StoredOutput};
|
||||
use crate::grin::GrinNode;
|
||||
|
||||
/// Grin coinbase maturity: a coinbase output is unspendable until this many
|
||||
/// blocks after its creation height (consensus `DAY_HEIGHT` = 24×60).
|
||||
const COINBASE_MATURITY: u64 = 1440;
|
||||
/// Max outputs per `get_unspent_outputs` page (node hard-caps at 10_000).
|
||||
const PAGE_MAX: u64 = 1000;
|
||||
/// How far back a reorg is auto-repaired. Grin's practical reorgs are a few
|
||||
/// blocks; deeper than this is not auto-healed (documented limitation).
|
||||
const REORG_DEPTH: u64 = 100;
|
||||
|
||||
/// Spawn the background scanner. Returns immediately; the task runs until the
|
||||
/// process exits.
|
||||
pub fn spawn(pool: AnyPool, node: GrinNode, cfg: Config) {
|
||||
tokio::spawn(async move {
|
||||
let poll = Duration::from_secs(cfg.scan_poll_secs.max(1));
|
||||
tracing::info!(
|
||||
poll_secs = cfg.scan_poll_secs,
|
||||
batch = cfg.scan_batch_blocks,
|
||||
"grin-lws scanner started (SCAFFOLD — block parse + rewind stubbed)"
|
||||
);
|
||||
tracing::info!(poll_secs = cfg.scan_poll_secs, "grin-lws scanner started");
|
||||
loop {
|
||||
if let Err(e) = tick(&pool, &node, &cfg).await {
|
||||
if let Err(e) = tick(&pool, &node).await {
|
||||
tracing::warn!(error = %e, "scanner tick failed; will retry");
|
||||
}
|
||||
tokio::time::sleep(poll).await;
|
||||
@@ -45,33 +55,185 @@ pub fn spawn(pool: AnyPool, node: GrinNode, cfg: Config) {
|
||||
});
|
||||
}
|
||||
|
||||
/// One scan pass: catch up from the cursor toward the tip, a batch at a time.
|
||||
async fn tick(pool: &AnyPool, node: &GrinNode, cfg: &Config) -> crate::error::Result<()> {
|
||||
let tip = node.get_tip_height().await?;
|
||||
let cursor = crate::db::get_cursor(pool).await?;
|
||||
let from = cursor.as_ref().map(|c| c.height + 1).unwrap_or(0);
|
||||
if from > tip {
|
||||
return Ok(()); // caught up
|
||||
}
|
||||
let to = (from + cfg.scan_batch_blocks).min(tip);
|
||||
tracing::debug!(from, to, tip, "scanner batch");
|
||||
/// One scan pass. See the module docs for the phase breakdown.
|
||||
async fn tick(pool: &AnyPool, node: &GrinNode) -> crate::error::Result<()> {
|
||||
let (tip_height, tip_hash) = node.get_tip().await?;
|
||||
let cursor = db::get_cursor(pool).await?;
|
||||
|
||||
for height in from..=to {
|
||||
// REAL IMPL (deferred):
|
||||
// let block = node.get_block(height).await?;
|
||||
// reorg check: if block.prev_hash != cursor.block_hash { rollback + reseek }
|
||||
// let accounts = db::list_account_hashes(pool).await?;
|
||||
// for out in &block.outputs {
|
||||
// for rh in &accounts {
|
||||
// if let Some(r) = grin::rewind_output(rh, out) {
|
||||
// db::insert_output(pool, &to_stored(out, &r), rh).await?;
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
// for commit in &block.input_commits { db::mark_spent(pool, commit, height).await?; }
|
||||
// db::set_cursor(pool, height, &block.hash).await?;
|
||||
let _ = height;
|
||||
return Ok(()); // scaffold: stop after entering the loop; no chain reads yet.
|
||||
// ── Phase 0: reorg check ────────────────────────────────────────────────
|
||||
if let Some(c) = &cursor {
|
||||
if c.tip_height > 0 {
|
||||
// Chain shorter than our checkpoint ⇒ reorg (header probe would just
|
||||
// 404). Otherwise, a changed block hash at the checkpoint height ⇒
|
||||
// reorg. A transient node error propagates and aborts the tick.
|
||||
if tip_height < c.tip_height {
|
||||
return handle_reorg(pool, node, tip_height).await;
|
||||
}
|
||||
let canon = node.header_hash_at(c.tip_height).await?;
|
||||
if canon != c.tip_hash {
|
||||
return handle_reorg(pool, node, tip_height).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let prev_tip = cursor.as_ref().map(|c| c.tip_height).unwrap_or(0);
|
||||
let from_index = cursor.as_ref().map(|c| c.output_mmr_index + 1).unwrap_or(1);
|
||||
|
||||
// ── Phase 1: discover new outputs for caught-up accounts ────────────────
|
||||
let accounts = db::list_accounts(pool).await?;
|
||||
let caught_up: Vec<String> = accounts
|
||||
.iter()
|
||||
.filter(|a| a.scan_height >= prev_tip)
|
||||
.map(|a| a.rewind_hash.clone())
|
||||
.collect();
|
||||
|
||||
let highest = forward_scan(pool, node, from_index, &caught_up).await?;
|
||||
let new_mmr_index = highest.max(cursor.as_ref().map(|c| c.output_mmr_index).unwrap_or(0));
|
||||
|
||||
// ── Phase 2: spend reconcile (by absence from the unspent set) ──────────
|
||||
reconcile_spends(pool, node, tip_height).await?;
|
||||
|
||||
// ── Phase 3: advance the cursor + caught-up accounts ────────────────────
|
||||
// (Only after discovery AND reconcile succeeded — Risk R1.)
|
||||
db::set_cursor(pool, new_mmr_index as i64, tip_height as i64, &tip_hash).await?;
|
||||
db::advance_caught_up_scan_heights(pool, tip_height as i64, prev_tip as i64).await?;
|
||||
|
||||
// ── Phase 4: backfill one pending new account (birthday → cursor) ───────
|
||||
if let Some(acct) = accounts.iter().find(|a| a.scan_height < prev_tip) {
|
||||
backfill_account(pool, node, acct, new_mmr_index, tip_height).await?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Page the output set from `from_index` to the current tip, rewinding each
|
||||
/// output against every `rewind_hash` in `accounts` and inserting matches.
|
||||
/// Returns the highest PMMR index reached. Rewinds nothing if `accounts` empty
|
||||
/// but still walks to learn `highest_index` so the cursor can advance.
|
||||
async fn forward_scan(
|
||||
pool: &AnyPool,
|
||||
node: &GrinNode,
|
||||
from_index: u64,
|
||||
accounts: &[String],
|
||||
) -> crate::error::Result<u64> {
|
||||
let mut start = from_index;
|
||||
let mut highest = from_index.saturating_sub(1);
|
||||
loop {
|
||||
let page = node.get_unspent_outputs(start, None, PAGE_MAX).await?;
|
||||
highest = highest.max(page.highest_index);
|
||||
for out in &page.outputs {
|
||||
store_first_match(pool, out, accounts).await?;
|
||||
}
|
||||
// Terminate when the page reached the tip, or made no forward progress
|
||||
// (defensive against a node that doesn't advance `last_retrieved_index`).
|
||||
if page.last_retrieved_index >= page.highest_index || page.last_retrieved_index < start {
|
||||
break;
|
||||
}
|
||||
start = page.last_retrieved_index + 1;
|
||||
}
|
||||
Ok(highest)
|
||||
}
|
||||
|
||||
/// Rewind `out` against each account; on the first match, store it (an output
|
||||
/// belongs to at most one wallet) and stop.
|
||||
async fn store_first_match(
|
||||
pool: &AnyPool,
|
||||
out: &crate::grin::ChainOutput,
|
||||
accounts: &[String],
|
||||
) -> crate::error::Result<()> {
|
||||
for rh in accounts {
|
||||
if let Some(r) = crate::grin::rewind_output(rh, out) {
|
||||
let lock_height = if out.is_coinbase {
|
||||
out.height.saturating_add(COINBASE_MATURITY)
|
||||
} else {
|
||||
out.height
|
||||
};
|
||||
db::insert_output(
|
||||
pool,
|
||||
&StoredOutput {
|
||||
commit: out.commit.clone(),
|
||||
value: r.value,
|
||||
height: out.height,
|
||||
mmr_index: out.mmr_index,
|
||||
is_coinbase: out.is_coinbase,
|
||||
lock_height,
|
||||
key_id: Some(r.key_id),
|
||||
n_child: Some(r.n_child),
|
||||
},
|
||||
rh,
|
||||
)
|
||||
.await?;
|
||||
break;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Mark any stored-unspent output that the node no longer lists as spent.
|
||||
async fn reconcile_spends(
|
||||
pool: &AnyPool,
|
||||
node: &GrinNode,
|
||||
tip_height: u64,
|
||||
) -> crate::error::Result<()> {
|
||||
let stored = db::all_unspent_commits(pool).await?;
|
||||
if stored.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
let present = node.outputs_present(&stored).await?;
|
||||
for commit in &stored {
|
||||
if !present.contains(commit) {
|
||||
db::mark_spent(pool, commit, tip_height as i64).await?;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Bring a newly-registered account up to date: scan the live output set from
|
||||
/// its birthday index up to the current cursor against ONLY its hash (its
|
||||
/// pre-existing outputs sit below the shared cursor), then mark it caught up so
|
||||
/// it joins the forward scan next tick.
|
||||
async fn backfill_account(
|
||||
pool: &AnyPool,
|
||||
node: &GrinNode,
|
||||
acct: &db::AccountRow,
|
||||
up_to_index: u64,
|
||||
tip_height: u64,
|
||||
) -> crate::error::Result<()> {
|
||||
let start = node.start_index_for_height(acct.start_height).await?;
|
||||
let only = [acct.rewind_hash.clone()];
|
||||
let mut idx = start;
|
||||
loop {
|
||||
let page = node.get_unspent_outputs(idx, Some(up_to_index), PAGE_MAX).await?;
|
||||
for out in &page.outputs {
|
||||
store_first_match(pool, out, &only).await?;
|
||||
}
|
||||
let reached = page.last_retrieved_index.min(page.highest_index);
|
||||
if reached >= up_to_index || page.last_retrieved_index < idx {
|
||||
break;
|
||||
}
|
||||
idx = page.last_retrieved_index + 1;
|
||||
}
|
||||
db::set_account_scan_height(pool, &acct.rewind_hash, tip_height as i64).await?;
|
||||
tracing::info!(height = tip_height, "grin-lws backfilled a new account");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// A reorg was detected: roll back everything above a bounded fork point and
|
||||
/// reseek from there. Idempotent re-discovery + spend-reconcile on subsequent
|
||||
/// ticks re-derive the canonical state. Reorgs deeper than `REORG_DEPTH` are not
|
||||
/// auto-repaired (the read-path maturity gate still protects spendability).
|
||||
async fn handle_reorg(
|
||||
pool: &AnyPool,
|
||||
node: &GrinNode,
|
||||
tip_height: u64,
|
||||
) -> crate::error::Result<()> {
|
||||
let fork = tip_height.saturating_sub(REORG_DEPTH);
|
||||
tracing::warn!(fork, tip = tip_height, "grin-lws reorg detected; rolling back");
|
||||
db::rollback_to(pool, fork as i64).await?;
|
||||
db::clamp_account_scan_heights(pool, fork as i64).await?;
|
||||
let fork_index = node.start_index_for_height(fork).await?;
|
||||
let fork_hash = node.header_hash_at(fork).await?;
|
||||
// Cursor sits just below the fork's first index so Phase 1 re-pages from it.
|
||||
db::set_cursor(pool, fork_index.saturating_sub(1) as i64, fork as i64, &fork_hash).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user