From 75542f6358fb34aa15f4e9e27596992958a71bf4 Mon Sep 17 00:00:00 2001 From: jwinterm Date: Thu, 9 Jul 2026 13:06:17 -0400 Subject: [PATCH] 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. --- migrations/0001_init.sql | 16 ++- src/api.rs | 16 ++- src/db.rs | 144 ++++++++++++++++++--- src/scanner.rs | 262 +++++++++++++++++++++++++++++++-------- 4 files changed, 365 insertions(+), 73 deletions(-) diff --git a/migrations/0001_init.sql b/migrations/0001_init.sql index 800a0e5..bbc570b 100644 --- a/migrations/0001_init.sql +++ b/migrations/0001_init.sql @@ -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 ); diff --git a/src/api.rs b/src/api.rs index 01fd318..f75c77d 100644 --- a/src/api.rs +++ b/src/api.rs @@ -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, pub n_child: Option, } @@ -160,13 +166,16 @@ pub async fn get_balance( Json(req): Json, ) -> Result> { 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, ) -> Result> { 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, diff --git a/src/db.rs b/src/db.rs index f420715..a712c87 100644 --- a/src/db.rs +++ b/src/db.rs @@ -38,11 +38,24 @@ pub struct StoredOutput { pub n_child: Option, } -/// 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> { 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> { Ok(rows.iter().map(|r| r.get::("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> { + 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::("rewind_hash"), + start_height: r.get::("start_height").max(0) as u64, + scan_height: r.get::("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> { + let rows = sqlx::query("SELECT \"commit\" FROM outputs WHERE spent = false") + .fetch_all(pool) + .await?; + Ok(rows.iter().map(|r| r.get::("commit")).collect()) +} + /// Unspent outputs for an account — the spendable set, WITH derivation paths. pub async fn unspent_outputs(pool: &AnyPool, rewind_hash: &str) -> Result> { 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> { 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> { - 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::("height").max(0) as u64, - block_hash: r.get::("block_hash"), + output_mmr_index: r.get::("output_mmr_index").max(0) as u64, + tip_height: r.get::("height").max(0) as u64, + tip_hash: r.get::("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(()) diff --git a/src/scanner.rs b/src/scanner.rs index d588bee..80ec996 100644 --- a/src/scanner.rs +++ b/src/scanner.rs @@ -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 = 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 { + 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(()) +}