From 8f4e1a6f103432aeeee79393cddbc649da8de2d3 Mon Sep 17 00:00:00 2001 From: Michael Taylor Date: Thu, 27 Aug 2026 22:16:16 -0700 Subject: [PATCH] fix(wallet): subscribe derived CAT puzzle hashes so a funded wallet reports its real $DIG balance A CAT coin sits at the OUTER puzzle hash that curries the asset's TAIL around its owner's p2 hash, never at the p2 hash itself, so the subscription filter dropped every hinted $DIG coin and attribution had no row to fill in. A funded wallet reported a balance of zero. Because that curry commits to the asset AND the owner together, the wallet can derive `cat_puzzle_hash(owner_p2, asset_id)` itself and SUBSCRIBE it. A coin arriving at one of those hashes is that asset, and is this wallet's, by construction: the hash it matched is the proof. `apply_coin_states` therefore returns to main's shape -- filter plus upsert, zero chain reads -- and the coin arrives already carrying its `asset_id` and its owner `hint`. The `hint` is not decoration. The CAT balance query scopes by `hint` rather than by puzzle hash, so a row admitted without one is stored, correctly typed, and still reads as zero. This deletes the lineage-at-admission machinery entirely rather than bounding it: `admit_hinted`, `NotAdmitted`, `AdmissionOutcome`, `SyncError::IncompleteBatch` and its session-kill, and the `BoundedLineage` read guard with its token bucket and refund. Outbound requests per admitted coin go from about twelve to zero, and the latency cost is none, because attribution is now a hash comparison on the frame path. No path can create an `asset_id IS NULL` row from a peer frame, so the XCH-miscount direction is closed structurally rather than by a guard. Kept on merit: the supervisor threading the `CatAttributor` (the original bug -- the only production call site passed `None`, so the pass could never run), `get_coin_spend_opt`'s corroborated absence, the coin-id binding check and placeholder repair, `from_lookup`, and the persisted `attribution_examined` mark. Those serve the out-of-band pass, which still attributes NFT and DID singletons and any CAT row already in the replica. Out of scope, deliberately: CATs whose asset id the wallet does not know in advance. Their outer hash cannot be derived, so they read as ABSENT rather than as a wrong number. That is the failure direction this wallet must have, and unknown-CAT discovery belongs out of band. SPEC 18.11a is rewritten around the derived-hash subscription and the set-separation rule; 18.11c is reduced to the out-of-band pass's outcome memory and its absent-versus-unavailable contract. Both are true of the code in this diff. Closes #382 Closes #380 Co-Authored-By: Claude --- Cargo.lock | 10 +- Cargo.toml | 2 +- SPEC.md | 82 ++- crates/dig-wallet/Cargo.toml | 2 +- crates/dig-wallet/src/sage/db.rs | 58 +- crates/dig-wallet/src/sage/fallback.rs | 348 ++++++++++- crates/dig-wallet/src/sage/rpc.rs | 20 +- crates/dig-wallet/src/sage/service.rs | 26 +- crates/dig-wallet/src/sage/singleton.rs | 326 ++++++++++- crates/dig-wallet/src/sage/sync.rs | 546 +++++++++++++++++- crates/dig-wallet/src/sage/sync_supervisor.rs | 145 ++++- .../src/sage/sync_supervisor/tests.rs | 396 ++++++++++++- 12 files changed, 1865 insertions(+), 96 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 9e67f8b4..f1d65a7e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3046,7 +3046,7 @@ dependencies = [ [[package]] name = "dig-node-service" -version = "0.160.0" +version = "0.161.0" dependencies = [ "async-trait", "axum", @@ -3339,7 +3339,7 @@ dependencies = [ [[package]] name = "dig-wallet" -version = "0.39.0" +version = "0.40.0" dependencies = [ "async-trait", "axum", @@ -4491,7 +4491,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite", - "socket2 0.5.10", + "socket2 0.6.5", "system-configuration", "tokio", "tower-service", @@ -5666,7 +5666,7 @@ dependencies = [ "quinn-udp", "rustc-hash 2.1.3", "rustls", - "socket2 0.5.10", + "socket2 0.6.5", "thiserror 2.0.19", "tokio", "tracing", @@ -5704,7 +5704,7 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2 0.5.10", + "socket2 0.6.5", "tracing", "windows-sys 0.61.2", ] diff --git a/Cargo.toml b/Cargo.toml index 3e90d840..d75ebb94 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -32,7 +32,7 @@ edition = "2021" # the ROOT manifest (`[workspace.package].version`), so it MUST be set here for a # release to fire (§3.6). The library crates (dig-node-core/dig-runtime/dig-wallet) # keep their own independent versions — only the released binary tracks the workspace version. -version = "0.160.0" +version = "0.161.0" # Release hardening, matching digstore: keep integer-overflow checks ON in release. # The node parses untrusted serialized input and does offset/length arithmetic over diff --git a/SPEC.md b/SPEC.md index 937c1234..3fa3ce51 100644 --- a/SPEC.md +++ b/SPEC.md @@ -5402,8 +5402,86 @@ attribute CAT coins to their asset id (TAIL hash) in the `coins` table (so `get_ complete). Parent spends are fetched through a `LineageSource` (out-of-DB lineage reads, B.5). Reads only. The sync loop runs this attribution as a post-apply step (`sync::CatAttributor`, threaded into `run_update_loop`): every `coin_state_update` is followed by an attribution pass that uncurries the -newly-synced candidate coins, so a synced CAT coin — stored initially with `asset_id: None` — gains its -TAIL and surfaces in `get_cats` (this is how `$DIG` resolves from the node). +newly-synced candidate coins. A CAT of a KNOWN asset is already attributed on arrival by §18.11a's +derived-hash subscription and needs no such read; this pass covers the rest — NFT and DID singletons, and +any CAT row already in the replica. The attributor is owned by +the SUPERVISOR, which builds it from the subscription set it resolved for the current attempt and threads +it into BOTH the catch-up and the update loop; a supervisor with no lineage source attaches none, and +that absence MUST be honest rather than silent. The pass also runs ONCE after a completed catch-up, so a +replica that syncs from genesis and then receives no further pushes still attributes what it found. + +18.11a. **A CAT coin is recognised by a puzzle hash the wallet DERIVED, not by a claim a peer makes.** +A CAT coin does not sit at its owner's p2 puzzle hash; it sits at the OUTER hash that curries the asset's +TAIL around that p2 hash. Because that curry commits to the asset AND the owner together, the wallet MUST +derive `cat_puzzle_hash(owner_p2, asset_id)` for each of its p2 hashes crossed with each asset id it +knows, and MUST add those hashes to the set it SUBSCRIBES. A coin arriving at one of them is that asset, +and is this wallet's, by construction. + +Admission is therefore a set-membership test against hashes this node computed from its own key material, +and it MUST issue no chain read. A coin so admitted MUST be written with its `asset_id` AND its owner +`hint` already populated: the CAT balance query scopes by `hint`, so a row stored without one is present, +correctly typed, and still reads as zero. + +No coin admitted from a peer frame may be written with a NULL `asset_id` unless the wallet means it as +XCH. `asset_id IS NULL` denotes XCH in this schema and is selected UNSCOPED by the spend-input selector, +so a row admitted "unattributed, to be attributed later" is offered to the coin selector as spendable XCH +that the wallet cannot solve. Deriving the hash before subscribing is what makes that state unreachable +rather than merely guarded against. + +The derived hashes MUST widen the SUBSCRIPTION set only. They are not addresses and they are not hashes +the wallet can sign for: they MUST NOT enter the plain p2 set that marks a coin as an ordinary signable +XCH coin, MUST NOT be counted as watched addresses, MUST NOT be recorded as the addresses a catch-up +covered, and MUST NOT reach the signer's own puzzle-hash set. Each of those sets answers a different +question, and a derived CAT hash is a wrong answer to every one of them. + +The failure direction is INCOMPLETENESS, never a wrong figure. An asset whose id the wallet does not know +in advance cannot have its outer hash derived, so its coins are not subscribed and not admitted, and that +asset reads as ABSENT. Discovering CATs whose asset ids are unknown in advance is OUT OF SCOPE of this +section; it cannot be done by local derivation and MUST NOT be attempted on the frame path, where a +remote peer sets the pace. + +18.11b. **A parent spend binds to the coin it was asked for.** A coin id is self-certifying — +`SHA256(parent ‖ puzzle_hash ‖ amount)` — so a `LineageSource` MUST check that the coin a spend answer +carries hashes to the coin that was requested, and MUST NOT return one that does not. Where it does not, +the coin is repaired from the coin record; where it still does not bind, the answer is NO LINEAGE rather +than a placeholder. This is a correctness requirement and not defence-in-depth: every CAT/singleton +driver derives its children's coin ids FROM that coin, so a placeholder makes `Cat::parse_children` +compute children matching nothing and the caller conclude the coin is not a CAT — the failure that +refused eight real `$DIG` coins on a funded wallet. + +18.11c. **The out-of-band attribution pass remembers its OUTCOMES, and distinguishes an absence from an +outage.** Coins the wallet did not recognise on arrival — NFT and DID singletons, and any CAT row already +in the replica — are attributed by a pass over rows the replica already holds. That pass runs on this +node's own schedule over its own data; it is not on the frame path and a remote peer does not set its +pace. + +**The pass remembers the attribution OUTCOME per coin row, not the lineage lookup.** A coin's parent +spend is settled chain history, so a row a pass RESOLVED and could not attribute answers identically for +ever. Those rows are ordinary: an NFT or DID coin row keeps `asset_id` NULL because the reconstruction is +written to its own table, and an odd-amount plain coin at the wallet's own p2 hash reconstructs to +nothing. A memory of failed LOOKUPS cannot cover them, because their lookups succeed — so without an +outcome mark each costs one outbound chain read per push frame for the life of the replica. A row whose +parent could NOT be read MUST NOT be marked: nothing was learned about it. The pass MUST therefore cost +work proportional to newly-arrived rows, and MUST NOT run after a frame that was refused before any +database write. + +**A lineage answer distinguishes ABSENT from UNAVAILABLE, and the SOURCE must be able to tell them +apart.** "A source answered and there is no such spend" and "no source could be reached" MUST NOT be the +same value. Only an absence may be remembered or treated as a settled judgement; an unavailability is a +statement about this node's reachability and MUST be treated as *unknown*, so that a later pass asks +again. + +The distinction MUST be carried by the chain READ, not merely by the enum. A source that reads spends +through an API which collapses "no such spend" into the same error as "the read failed" cannot produce an +absence at all, whatever its mapping says — so the ABSENT arm becomes unreachable in production for the +exact case it was written for. The production source MUST therefore use an absence-aware, corroborated +read (`chia-query`'s `get_coin_spend_opt`), whose `Ok(None)` requires agreement across independent +sources and whose every transport failure, rejection and disagreement remains an error. Collapsing the +pair lets a transient outage be cached as a fact. + +**A failed lineage read is NO LINEAGE, never an error.** An error propagates out of the attribution pass +and ends the peer session, which hands a denial of service to whoever made the read fail. The same +reasoning binds §18.11b's repair read, which fails to NO LINEAGE rather than propagating. 18.12. **Live broadcaster bring-up — real mainnet $DIG spends behind a config gate (#428).** The node-custodied wallet BUILDS + SIGNS + VALIDATES spends (§18.9/§18.21) and the tip engine (§18.23) diff --git a/crates/dig-wallet/Cargo.toml b/crates/dig-wallet/Cargo.toml index db239f70..deb31327 100644 --- a/crates/dig-wallet/Cargo.toml +++ b/crates/dig-wallet/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "dig-wallet" -version = "0.39.0" +version = "0.40.0" edition = "2021" license = "GPL-2.0-only" description = "DIG Browser built-in Chia wallet sidecar: a local axum server (using digstore-chain + chia-wallet-sdk over coinset.org) that serves a Sage-mirroring wallet UI. Native Rust so BLS signing works; the browser opens it at 127.0.0.1." diff --git a/crates/dig-wallet/src/sage/db.rs b/crates/dig-wallet/src/sage/db.rs index ff4d166a..9518237a 100644 --- a/crates/dig-wallet/src/sage/db.rs +++ b/crates/dig-wallet/src/sage/db.rs @@ -380,7 +380,8 @@ CREATE TABLE IF NOT EXISTS coins ( asset_id TEXT, hint TEXT, created_timestamp INTEGER, - spent_timestamp INTEGER + spent_timestamp INTEGER, + attribution_examined INTEGER ); CREATE INDEX IF NOT EXISTS idx_coins_ph ON coins (puzzle_hash); CREATE INDEX IF NOT EXISTS idx_coins_asset ON coins (asset_id); @@ -563,6 +564,10 @@ const ADD_COLUMN_MIGRATIONS: &[&str] = &[ // rows banned before this column existed, which sorts them first and so evicts them first -- // the right order, since they are by definition the oldest bans on the machine. "ALTER TABLE peers ADD COLUMN banned_at INTEGER", + // Set once a CAT/singleton attribution pass has RESOLVED this coin's parent spend and acted + // on what it said (dig-node#383). NULL means "not examined yet", which is the right reading + // for every row written before this column existed: they are re-examined once and then settle. + "ALTER TABLE coins ADD COLUMN attribution_examined INTEGER", ]; // ---- one-shot data-migration ladder --------------------------------------- @@ -625,6 +630,10 @@ const POST_MIGRATION_INDEXES: &[&str] = &[ // The eviction order is `ORDER BY last_used_at`, run on every cache write (dig_ecosystem#3035). "CREATE INDEX IF NOT EXISTS idx_chain_read_cache_last_used ON chain_read_cache (last_used_at)", "CREATE INDEX IF NOT EXISTS idx_chain_spend_cache_last_used ON chain_spend_cache (last_used_at)", + // `unexamined_attribution_candidates` runs after every applied push frame. Without this it is + // a full table scan per frame; with it, a quiet wallet's pass touches no rows at all. + "CREATE INDEX IF NOT EXISTS idx_coins_attribution_pending ON coins (attribution_examined) \ + WHERE spent_height IS NULL AND asset_id IS NULL AND attribution_examined IS NULL", ]; // ---- chain-read cache budget (dig_ecosystem#3035) ------------------------- @@ -2606,6 +2615,53 @@ impl WalletDb { Ok(()) } + /// Record that an attribution pass RESOLVED this coin's parent spend and acted on the answer, + /// so no later pass reads it again (dig-node#383). + /// + /// # Why the outcome is remembered rather than the lookup + /// + /// A CAT gains an `asset_id` and so is self-marking, but the other resolved outcomes are not: + /// [`Self::upsert_nft`] and [`Self::upsert_did`] write their own tables and leave + /// `coins.asset_id` NULL, and an odd-amount plain XCH coin at the wallet's own p2 hash + /// reconstructs to *nothing at all*. Every one of those rows is a candidate again on the next + /// pass, resolves again, and costs another outbound chain read — forever, at whatever cadence + /// a peer chooses to send frames at. + /// + /// A cache over the *lookup* cannot fix that, because these lookups succeed. Only the + /// attribution OUTCOME is stable enough to remember, and it is stable for the strongest + /// possible reason: a coin's parent spend is settled chain history and cannot change. + /// + /// The mark is set ONLY on a resolved read. A parent that could not be reached leaves the row + /// unmarked, because remembering "we could not ask" as "we asked" is how an outage turns into + /// a permanent wrong balance. + pub async fn mark_attribution_examined(&self, coin_id: &str) -> sqlx::Result<()> { + sqlx::query("UPDATE coins SET attribution_examined = 1 WHERE coin_id = ?") + .bind(Self::normalise_hex(coin_id)) + .execute(&self.pool) + .await?; + Ok(()) + } + + /// The coins an attribution pass could still learn something from: unspent, unattributed, + /// confirmed, and not yet examined. + /// + /// Narrowed in SQL rather than by filtering [`Self::all_coins`] in Rust. The pass runs after + /// every applied push frame, so a whole-table read there is a standing per-frame cost + /// proportional to everything the replica has ever synced; here it is proportional to what + /// has newly arrived, and on a quiet wallet it returns nothing. + pub async fn unexamined_attribution_candidates(&self) -> sqlx::Result> { + let rows = sqlx::query( + "SELECT * FROM coins + WHERE spent_height IS NULL + AND asset_id IS NULL + AND created_height IS NOT NULL + AND attribution_examined IS NULL", + ) + .fetch_all(&self.pool) + .await?; + Ok(rows.iter().map(Self::coin_from_row).collect()) + } + // ---- NFTs ------------------------------------------------------------- /// Insert or update a reconstructed NFT (keyed by launcher id; a later coin overwrites diff --git a/crates/dig-wallet/src/sage/fallback.rs b/crates/dig-wallet/src/sage/fallback.rs index 10e79daf..9f49d31c 100644 --- a/crates/dig-wallet/src/sage/fallback.rs +++ b/crates/dig-wallet/src/sage/fallback.rs @@ -372,10 +372,17 @@ impl ChainFallback for CoinsetFallback { /// `Ok(None)` ONLY when a chain source ANSWERED and reported no such coin; every failure to /// read is an `Err` (dig_ecosystem#2392). /// - /// `Ok(None)` is NOT yet proof of absence: `chia-query` 0.6 mints it from ONE peer's empty - /// coin-state list without consulting coinset, so a peer that is a block behind, mid-reorg, - /// pruning or hostile produces it. Requiring corroboration is dig_ecosystem#2456, one crate - /// down. Callers polling a mint read `None` as "not seen yet", never as "never happened". + /// `Ok(None)` IS proof of absence, and the mapping at [`ChiaQueryLineage::parent_spend`] now + /// depends on that. The graph resolves `chia-query` **0.19.0**, where this read goes through + /// `peer_then_coinset_opt` into `read_opt_corroborated`: `Ok(None)` is produced only for a + /// `CorroboratedAbsent` — the answering peer plus `CORROBORATION_FLOOR` independent peers at + /// different addresses all reporting absent — or a peer-uncorroborated absence that coinset + /// agrees with. ONE peer's empty coin-state list yields `UncorroboratedAbsent`, and any + /// contradiction is `SourcesDisagree`, which stays an `Err`. dig_ecosystem#2456, which this + /// comment used to cite as pending against `chia-query` 0.6, has landed. + /// + /// Callers polling a MINT should still read `None` as "not seen yet" rather than "never + /// happened" — but that is a statement about mempool timing, not about corroboration. /// /// The absence-aware `_opt` variant carries that distinction (a `success: true` envelope with a /// null record is absence; a transport/API failure is not), so this method must not re-decide @@ -506,33 +513,110 @@ impl super::singleton::LineageSource for ChiaQueryLineage { async fn parent_spend( &self, parent_coin_id: &str, - spent_height: u32, - ) -> Result> { + _spent_height: u32, + ) -> Result { + use super::singleton::LineageAnswer; let coin_id = format!("0x{}", CoinsetFallback::norm_hex(parent_coin_id)); - let cs = match self - .query - .get_puzzle_and_solution(&coin_id, Some(spent_height)) - .await - { - Ok(cs) => cs, - // The parent spend is not available (unspent / not found) — a clean "no lineage". - Err(_) => return Ok(None), + // The ABSENCE-AWARE read, and the choice is load-bearing rather than stylistic. + // + // The non-`_opt` `get_puzzle_and_solution` returns `Result`, so a parent + // that does not exist arrives as an `Err` indistinguishable from an outage. Mapping that + // to `Unavailable` was the only safe direction available — and it made a settled absence + // unreachable in production for the exact case it was written for, so a parent that simply + // does not exist produced "this node could not read the chain" and the attribution pass + // retried it forever instead of concluding anything (dig-node#383). + // + // `get_coin_spend_opt` carries the distinction the caller needs, and carries it graded: + // it routes through `peer_then_coinset_opt`, so `Ok(None)` is a CORROBORATED absence — + // the answering peer plus `CORROBORATION_FLOOR` independent peers at different addresses, + // or a peer-uncorroborated absence that coinset agrees with — while any contradiction is + // `SourcesDisagree` and stays an `Err`. One hostile peer's empty coin-state list cannot + // mint an absence here. + // + // It also needs no height: it resolves the spend height itself from a coin-state read, + // which is why `_spent_height` is unused. An unknown OR unspent coin is `Ok(None)` — both + // are honestly "there is no such parent spend", which is the question being asked. + let cs = match self.query.get_coin_spend_opt(&coin_id).await { + Ok(Some(cs)) => cs, + // A corroborated absence. SETTLED, not unknown: the coin the peer offered descends + // from nothing, so refusing it is a judgement about the peer's claim and leaves the + // replica no less complete. + Ok(None) => return Ok(LineageAnswer::Absent), + // A read that could not be completed — an outage, a rejection, or two sources that + // disagree. Nothing was learned, so the weaker claim is the only safe one: a caller + // that hears "unknown" retries and refuses to declare itself complete, whereas one + // that hears "absent" writes the coin off and answers a confident balance without it. + Err(_) => return Ok(LineageAnswer::Unavailable), }; let decode = |field: &str, s: &str| -> Result> { hex::decode(s.strip_prefix("0x").unwrap_or(s)) .map_err(|e| Error::internal(format!("lineage {field} hex: {e}"))) }; - let parent = super::singleton::bytes32_from_hex(&cs.coin.parent_coin_info)?; - let puzzle_hash = super::singleton::bytes32_from_hex(&cs.coin.puzzle_hash)?; - Ok(Some(super::singleton::ParentSpend { - coin: chia_protocol::Coin { - parent_coin_info: parent, - puzzle_hash, - amount: cs.coin.amount, + let coin = chia_protocol::Coin { + parent_coin_info: super::singleton::bytes32_from_hex(&cs.coin.parent_coin_info)?, + puzzle_hash: super::singleton::bytes32_from_hex(&cs.coin.puzzle_hash)?, + amount: cs.coin.amount, + }; + // The coin a spend answer CARRIES is checked against the coin that was ASKED for, and + // repaired from the coin record when it does not bind. + // + // This is not defence-in-depth. `chia-query`'s PEER tier returns the puzzle and solution + // faithfully and leaves the coin a placeholder — measured on mainnet 2026-08-27: for + // parent `567d481d…` the coinset tier answered the real coin while the peer tier answered + // `puzzle_hash: 0x00…00, amount: 0` beside byte-identical reveal and solution. Every CAT + // driver derives its children's coin ids FROM that coin, so a zeroed one makes + // `Cat::parse_children` compute children that match nothing and report "not a CAT". That + // is the whole of dig-node#382's last mile: eight real $DIG coins, refused one by one, on + // a wallet holding 3,856.455 $DIG. + // + // A coin id is self-certifying — `SHA256(parent ‖ puzzle_hash ‖ amount)` — so the binding + // is checkable locally and costs nothing when the answer is already right. + let expected = super::singleton::bytes32_from_hex(parent_coin_id)?; + let coin = if coin.coin_id() == expected { + coin + } else { + // `Ok(None)` on a failed read, matching the spend read above (`Err(_) => Ok(None)`) + // rather than propagating. On the peer tier the answer is a placeholder every time, + // so THIS is the common path — and an `Err` here escapes `reconstruct_coins`, then + // `attribute()`, then `run_update_loop`, killing the peer session over a transient + // coinset blip. The supervisor makes the same call one function away, deliberately: + // "a read failure must never turn a completed catch-up into a failed session". + // A missing lineage is refused-and-retried; a dead session is not. + let record = match CoinsetFallback::new(self.query.clone()) + .coin_record_by_id(parent_coin_id) + .await + { + Ok(Some(record)) => record, + // A source answered and has no such coin: settled, and cheap to remember. + Ok(None) => return Ok(LineageAnswer::Absent), + // No source answered. Never an `Err`, for the reason above the spend read; and + // never `Absent`, because nothing was learned. + Err(_) => return Ok(LineageAnswer::Unavailable), + }; + let repaired = chia_protocol::Coin { + parent_coin_info: super::singleton::bytes32_from_hex(&record.parent_coin_info)?, + puzzle_hash: super::singleton::bytes32_from_hex(&record.puzzle_hash)?, + amount: record.amount, + }; + // `coin_record_by_id` already refuses a record for a different coin, so reaching here + // with a mismatch would mean two independent reads disagree about a self-certifying + // id. There is no honest lineage to return in that case. + if repaired.coin_id() != expected { + // Both reads answered and they disagree about a self-certifying id. Not an + // outage — the sources are reachable and one of them is wrong — so this is + // `Absent`: there is no honest lineage here and re-asking would return the same + // contradiction. + return Ok(LineageAnswer::Absent); + } + repaired + }; + Ok(LineageAnswer::Found(Box::new( + super::singleton::ParentSpend { + coin, + puzzle_reveal: decode("puzzle_reveal", &cs.puzzle_reveal)?, + solution: decode("solution", &cs.solution)?, }, - puzzle_reveal: decode("puzzle_reveal", &cs.puzzle_reveal)?, - solution: decode("solution", &cs.solution)?, - })) + ))) } } @@ -627,6 +711,7 @@ mod chain_failure_tests { //! performs no DNS), so every read falls through to the local coinset stand-in below. use super::*; + use crate::sage::singleton::{LineageAnswer, LineageSource}; use std::sync::Arc; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpListener; @@ -674,6 +759,221 @@ mod chain_failure_tests { format!("http://127.0.0.1:{port}") } + /// Serve a body chosen by the request PATH, so one fixture can answer two different coinset + /// endpoints differently. [`serve_json`] answers every path identically, which cannot express + /// a tier that is right about one read and wrong about another. + async fn serve_routed(routes: &'static [(&'static str, &'static str)]) -> String { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let port = listener.local_addr().expect("addr").port(); + tokio::spawn(async move { + while let Ok((mut socket, _)) = listener.accept().await { + let mut buf = [0u8; 4096]; + let n = socket.read(&mut buf).await.unwrap_or(0); + let request = String::from_utf8_lossy(&buf[..n]).to_string(); + let body = routes + .iter() + .find(|(path, _)| request.contains(path)) + .map(|(_, body)| *body) + .unwrap_or(r#"{"success":false}"#); + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\ + Connection: close\r\n\r\n{body}", + body.len() + ); + let _ = socket.write_all(response.as_bytes()).await; + let _ = socket.shutdown().await; + } + }); + format!("http://127.0.0.1:{port}") + } + + /// A `get_puzzle_and_solution` answer carrying a PLACEHOLDER coin — a zeroed puzzle hash and a + /// zero amount beside a faithful reveal and solution. This is the shape `chia-query`'s PEER + /// tier really returns (measured on mainnet, dig-node#382). + const SPEND_WITH_PLACEHOLDER_COIN: &str = r#"{"success":true,"coin_solution":{ + "coin":{"parent_coin_info":"0x1111111111111111111111111111111111111111111111111111111111111111", + "puzzle_hash":"0x0000000000000000000000000000000000000000000000000000000000000000", + "amount":0}, + "puzzle_reveal":"0x01","solution":"0x80"}}"#; + + /// The coin record for [`KNOWN_COIN_ID`], which is what the repair read must recover. + const REPAIR_RECORD: &str = KNOWN_COIN_RECORD; + + async fn lineage_against(base_url: String) -> ChiaQueryLineage { + let query = chia_query::ChiaQuery::new(chia_query::ChiaQueryConfig { + coinset_base_url: base_url, + max_peers: 0, + coinset_fallback_enabled: true, + coinset_request_timeout: std::time::Duration::from_secs(5), + ..Default::default() + }) + .await + .expect("a zero-peer client with the coinset fallback enabled always constructs"); + ChiaQueryLineage::new(Arc::new(query)) + } + + /// **Proves (dig-node#382, the last mile):** a parent spend whose CARRIED coin does not bind to + /// the coin that was asked for is repaired from the coin record, never returned as-is. + /// + /// # Why this is a correctness bug and not hardening + /// + /// Every CAT/singleton driver derives its children's coin ids FROM `ParentSpend::coin`. A + /// placeholder coin therefore makes `Cat::parse_children` compute children that match nothing, + /// and the caller concludes the coin is not a CAT. Measured on mainnet: all eight of a real + /// wallet's unspent $DIG coins were refused this way, on a wallet holding 3,856.455 $DIG, while + /// the coinset tier answered the same question correctly. + /// + /// The fixture serves a placeholder coin on the SPEND read and the truth on the RECORD read, + /// because that is exactly the split observed — a tier right about one read and wrong about + /// another. A fixture that got both wrong could not tell repair from luck. + #[tokio::test] + async fn a_parent_spend_that_does_not_bind_is_repaired_from_the_coin_record() { + let base = serve_routed(&[ + ("get_puzzle_and_solution", SPEND_WITH_PLACEHOLDER_COIN), + ("get_coin_record_by_name", REPAIR_RECORD), + ]) + .await; + let lineage = lineage_against(base).await; + + let spend = lineage + .parent_spend(KNOWN_COIN_ID, 140) + .await + .expect("the read succeeds") + .found() + .expect("a spend the chain reported must not be dropped"); + + assert_eq!( + hex::encode(spend.coin.coin_id()), + KNOWN_COIN_ID, + "the returned parent coin must bind to the coin that was asked for" + ); + assert_eq!(spend.puzzle_reveal, vec![0x01]); + assert_eq!(spend.solution, vec![0x80]); + } + + /// **The control:** when the repair read cannot recover a binding coin either, the answer is + /// a reported ABSENCE — never the placeholder. + /// + /// Without this half, the test above is satisfied by an implementation that returns whatever + /// the record read produced without checking it, which is the same unchecked trust one hop + /// further along. + #[tokio::test] + async fn an_unrepairable_parent_spend_is_no_lineage_rather_than_a_placeholder() { + let base = serve_routed(&[ + ("get_puzzle_and_solution", SPEND_WITH_PLACEHOLDER_COIN), + ( + "get_coin_record_by_name", + r#"{"success":true,"coin_record":null}"#, + ), + ]) + .await; + let lineage = lineage_against(base).await; + + let spend = lineage + .parent_spend(KNOWN_COIN_ID, 140) + .await + .expect("a reported absence is not a read failure"); + + assert!( + matches!(spend, LineageAnswer::Absent), + "a chain that ANSWERED 'no such coin' is an absence, not an outage: an absence may be remembered and written off, an outage may not. Got {spend:?}" + ); + } + + /// A `get_puzzle_and_solution` answer reporting that the chain HAS no such spend: a + /// `success: true` envelope carrying a null `coin_solution`. This is the shape coinset returns + /// for a coin that is unknown or simply unspent. + const NO_SUCH_SPEND: &str = r#"{"success":true,"coin_solution":null}"#; + + /// **Proves (dig-node#383, F1):** the PRODUCTION source reports a parent the chain does not + /// have as [`LineageAnswer::Absent`], not as an outage. + /// + /// # Why this test is about a call, not a branch + /// + /// The attribution pass distinguishes "the chain says there is no such parent" from "this node + /// could not read the chain", and only the first is a settled judgement it can act on. That + /// distinction was unreachable in production: the spend read went through the non-`_opt` + /// `get_puzzle_and_solution`, which returns `Result`, so a nonexistent parent + /// arrived as an `Err` indistinguishable from a dead network and was mapped, correctly for what + /// it knew, to [`LineageAnswer::Unavailable`]. A row whose parent genuinely does not exist was + /// therefore re-read on every pass, forever, and never concluded. + /// + /// So the defect was not a branch that decided wrongly; it was a CALL that could not carry the + /// answer. This asserts on the answer the production impl gives for the honest wire shape, + /// which is the only thing that can distinguish the two reads. + /// + /// The sibling below is the control that stops this being satisfied by a source that simply + /// calls everything absent: an outage on the SAME read must still be `Unavailable`. + #[tokio::test] + async fn a_parent_the_chain_reports_no_spend_for_is_absent_not_unavailable() { + let base = serve_routed(&[("get_puzzle_and_solution", NO_SUCH_SPEND)]).await; + let lineage = lineage_against(base).await; + + let answer = lineage + .parent_spend(KNOWN_COIN_ID, 140) + .await + .expect("a reported absence is not a read failure"); + + assert!( + matches!(answer, LineageAnswer::Absent), + "a corroborated 'there is no such spend' is a settled judgement about the PEER'S \ + CLAIM, so the coin is refused and the batch stays complete. Reported as an outage it \ + becomes a statement about this node, the batch is incomplete, the session is torn \ + down, and the peer re-sends the same 32 random bytes on every redial. Got {answer:?}" + ); + } + + /// **The control for the test above.** A read that genuinely FAILS on the very same endpoint + /// must still be [`LineageAnswer::Unavailable`]. + /// + /// Without it, "absence maps to `Absent`" would be satisfied by a source that had simply + /// stopped distinguishing the two in the other direction — which is the money-lie this family + /// exists to close, since a wallet that reads "we could not reach anyone" as "it does not + /// exist" writes off coins it owns and answers a confident balance without them. + #[tokio::test] + async fn a_spend_read_that_fails_is_still_unavailable_not_absent() { + // No route matches, so the fixture answers `{"success":false}` — a rejection, not an + // absence. + let base = serve_routed(&[("some_other_endpoint", NO_SUCH_SPEND)]).await; + let lineage = lineage_against(base).await; + + let answer = lineage + .parent_spend(KNOWN_COIN_ID, 140) + .await + .expect("a failed read refuses the coin, never the session"); + + assert!( + matches!(answer, LineageAnswer::Unavailable), + "nothing was learned about the chain, so the weaker claim is the only honest one. \ + Got {answer:?}" + ); + } + + /// **A failed REPAIR read is no lineage, not a dead peer session (#383).** + /// + /// The spend read beside it maps a failed read to `Ok(None)` deliberately; the repair read + /// added with the binding check did not, and on the peer tier the repair branch is the + /// COMMON path — every answer there is a placeholder. So a transient coinset failure + /// propagated out of `parent_spend`, through `reconstruct_coins`, through `attribute()`, and + /// ended the peer session, over a read the wallet is entitled to simply retry. + /// + /// The fixture routes ONLY the spend read, so the record read hits the fallback route and + /// FAILS. That is the distinction that matters and the one the sibling test above cannot + /// make: it serves `coin_record: null`, a chain that answered "no such coin", which reaches + /// `Ok(None)` by a different branch and would stay green with the propagation intact. + #[tokio::test] + async fn a_failed_repair_read_is_no_lineage_rather_than_a_failed_session() { + let base = serve_routed(&[("get_puzzle_and_solution", SPEND_WITH_PLACEHOLDER_COIN)]).await; + let lineage = lineage_against(base).await; + + let spend = lineage.parent_spend(KNOWN_COIN_ID, 140).await; + + assert!( + matches!(spend, Ok(LineageAnswer::Unavailable)), + "a failed repair read must refuse the coin, not the session — and must say UNAVAILABLE rather than ABSENT, because nothing was learned about the chain. Reporting it as an absence would let one failed read write a real coin off for the cache's whole TTL. Got {spend:?}" + ); + } + /// A coin id the fixtures ask for. Its value is irrelevant — what varies is the SOURCE. const SOME_COIN_ID: &str = "1111111111111111111111111111111111111111111111111111111111111111"; diff --git a/crates/dig-wallet/src/sage/rpc.rs b/crates/dig-wallet/src/sage/rpc.rs index cdff70d4..b326c21d 100644 --- a/crates/dig-wallet/src/sage/rpc.rs +++ b/crates/dig-wallet/src/sage/rpc.rs @@ -3328,6 +3328,7 @@ impl WalletBackend { let parent = lineage .parent_spend(&row.parent_coin_info, created) .await? + .found() .ok_or_else(|| Error::internal("CAT parent spend unavailable"))?; let child = singleton::coin_from_row(row)?; let cat = singleton::resolve_cat(&parent, child)? @@ -3468,6 +3469,7 @@ impl WalletBackend { let parent = lineage .parent_spend(&row.parent_coin_info, created) .await? + .found() .ok_or_else(|| Error::internal("parent spend unavailable"))?; let child = singleton::coin_from_row(&row)?; Ok((parent, child)) @@ -9179,8 +9181,13 @@ mod tests { &self, parent_coin_id: &str, _spent_height: u32, - ) -> Result> { - Ok((parent_coin_id == self.parent_id).then(|| self.spend.clone())) + ) -> Result { + // A parent this double does not hold is one the node could not READ, which is + // what the production source reports for an unresolvable parent. + Ok(singleton::LineageAnswer::from_lookup( + (parent_coin_id == self.parent_id).then(|| self.spend.clone()), + singleton::LineageAnswer::Unavailable, + )) } } @@ -9357,8 +9364,13 @@ mod tests { &self, parent_coin_id: &str, _spent_height: u32, - ) -> Result> { - Ok((parent_coin_id == self.parent_id).then(|| self.spend.clone())) + ) -> Result { + // A parent this double does not hold is one the node could not READ, which is + // what the production source reports for an unresolvable parent. + Ok(singleton::LineageAnswer::from_lookup( + (parent_coin_id == self.parent_id).then(|| self.spend.clone()), + singleton::LineageAnswer::Unavailable, + )) } } diff --git a/crates/dig-wallet/src/sage/service.rs b/crates/dig-wallet/src/sage/service.rs index 334ca371..d4fb3ab2 100644 --- a/crates/dig-wallet/src/sage/service.rs +++ b/crates/dig-wallet/src/sage/service.rs @@ -32,8 +32,9 @@ use super::spend::{ Broadcaster, ChiaQueryBroadcaster, ChiaQueryConfirmer, Confirmer, ConfirmingBroadcaster, }; use super::sync_supervisor::{ - spawn_supervisor, ChiaPeerSessionFactory, ChiaQuorumCorroborator, FallbackChainTip, Supervisor, - SyncHandle, TokioTime, UnionPuzzleHashSource, SESSION_MAX_LIFETIME, + spawn_supervisor, Attribution, ChiaPeerSessionFactory, ChiaQuorumCorroborator, + FallbackChainTip, Supervisor, SyncHandle, TokioTime, UnionPuzzleHashSource, + KNOWN_CAT_ASSET_IDS, SESSION_MAX_LIFETIME, }; use super::tipping::{ChainOwnerResolver, NodeTipSpender, SystemClock, TipEventBus, TippingEngine}; use super::transport::SharedCert; @@ -230,8 +231,29 @@ impl WalletService { // writes", never a silent downgrade. // The task handle is dropped deliberately: the supervisor lives for the process, and // its stop signal is `SyncHandle::shutdown`, not a dropped join handle. + // Attribution rides the SAME shared client as every other chain read, and is built here + // rather than inside `build_live_wallet` on purpose: uncurrying a parent spend to recover + // a CAT's asset id is a READ. Gating it on `enable_live_broadcast` — the flag that means + // "this node may spend" — would leave every read-only install unable to name the $DIG it + // holds, which is exactly the shape of dig-node#382. + let attribution: Option> = if cfg.enable_chain_sync { + match chain.shared_client().await { + Ok(query) => Some(Arc::new(Attribution::new( + Arc::new(ChiaQueryLineage::new(query)), + WalletConfig::default().address_prefix, + ))), + Err(e) => { + warn_chain_source_unavailable(&e); + None + } + } + } else { + None + }; let sync = if cfg.enable_chain_sync { Some(spawn_supervisor(Supervisor { + attribution, + cat_asset_ids: KNOWN_CAT_ASSET_IDS.to_vec(), db: db.clone(), puzzle_hashes: Arc::new(UnionPuzzleHashSource::new( custody.clone(), diff --git a/crates/dig-wallet/src/sage/singleton.rs b/crates/dig-wallet/src/sage/singleton.rs index 68e25a05..65106dc5 100644 --- a/crates/dig-wallet/src/sage/singleton.rs +++ b/crates/dig-wallet/src/sage/singleton.rs @@ -68,6 +68,76 @@ pub struct ReconstructStats { pub dids: u32, /// CAT coins attributed. pub cats: u32, + /// Rows this pass RESOLVED and will never read again, whatever they turned out to be. + pub settled: u32, + /// Rows whose parent could not be read at all, so nothing was learned and nothing marked. + pub unresolved: u32, +} + +/// What a lineage source has to say about a parent coin's spend. +/// +/// The two negative answers are deliberately NOT one variant. They differ in the only way that +/// matters to a caller deciding what to do next: +/// +/// * [`LineageAnswer::Absent`] is a FACT about the chain — it was asked and the spend is not +/// there. Remembering it is sound, because asking again immediately gets the same answer. +/// * [`LineageAnswer::Unavailable`] is a fact about THIS NODE — no source could be reached. +/// Nothing was learned about the chain at all. +/// * [`LineageAnswer::Deferred`] is also a fact about this node, and a *different* one: the read +/// was never issued, because a local budget declined to issue it. +/// +/// Collapsing the first two is how an outage becomes a money-lie: a wallet that caches "we could +/// not reach anyone" as "it does not exist" spends the cache's lifetime confidently refusing to +/// name coins it owns. Every consumer here therefore treats `Unavailable` as *unknown* — never as +/// a negative result, and never as grounds to declare a replica complete. +/// +/// Collapsing the last two is how a bound becomes a denial of service, which is the reason +/// `Deferred` exists at all (dig-node#383). `Unavailable` is TRANSIENT — the chain source is +/// unreachable now and a retry is the remedy — so a caller that sees it re-opens the session and +/// asks again. A budget refusal is a STANDING property of this node under its current load, so a +/// caller that reads it as transient reconnects forever without ever making progress. This is the +/// distinction the attribution pass needs in order to decide whether asking again can help. +#[derive(Debug, Clone)] +pub enum LineageAnswer { + /// The parent spend, as read. + Found(Box), + /// A source answered, and there is no such spend at that height. + Absent, + /// No source could answer. Nothing is known either way. + Unavailable, + /// The read was declined locally, by a budget, before any source was asked. Nothing is known + /// either way — and unlike [`Self::Unavailable`], nothing was even attempted. + Deferred, +} + +impl LineageAnswer { + /// The spend if one was found. Discards the [`Absent`](LineageAnswer::Absent) / + /// [`Unavailable`](LineageAnswer::Unavailable) distinction, so it belongs only in callers + /// that genuinely treat the two alike. + pub fn found(self) -> Option { + match self { + Self::Found(spend) => Some(*spend), + _ => None, + } + } + + /// The answer for a spend that was found, or `on_miss` when it was not. + /// + /// # Why the miss answer must be passed in + /// + /// This replaces a `from_answered(Option)` helper that folded every miss to + /// [`Self::Absent`]. Nothing in production ever called it — but it was the constructor every + /// test double reached for, so **every** double in this crate modelled an unresolvable parent + /// as a settled absence while the production source modelled it as unreadable. The suite was + /// therefore structurally unable to reach production's unreadable-parent path by the ordinary + /// route, which is how dig-node#383 survived a review round hunting exactly that class. + /// + /// A fixture map cannot know which of the two a miss represents, so it has to say. Making the + /// caller name it is the whole point: a double that cannot express production's failure mode + /// makes every test over it a partial green. + pub fn from_lookup(spend: Option, on_miss: Self) -> Self { + spend.map_or(on_miss, |s| Self::Found(Box::new(s))) + } } /// Fetches the parent coin's spend for a coin being reconstructed. The production path reads @@ -76,12 +146,13 @@ pub struct ReconstructStats { #[async_trait] pub trait LineageSource: Send + Sync { /// The spend of `parent_coin_id`, which was spent at `spent_height` (= the child's - /// created height). `None` if the parent spend is not available. - async fn parent_spend( - &self, - parent_coin_id: &str, - spent_height: u32, - ) -> Result>; + /// created height). + /// + /// `Err` is reserved for a caller-fatal fault. An unreadable parent is + /// [`LineageAnswer::Unavailable`], not an error: an error here escapes + /// [`reconstruct_coins`] and ends the peer session, handing a denial of service to whoever + /// made the read fail. + async fn parent_spend(&self, parent_coin_id: &str, spent_height: u32) -> Result; } fn hexb(b: Bytes32) -> String { @@ -366,8 +437,26 @@ fn is_candidate(c: &CoinRow, plain_puzzle_hashes: &HashSet) -> bool { /// /// For each **unspent** candidate coin, fetch its parent spend through `lineage`, reconstruct /// it, and write the result: NFT/DID rows are upserted; a CAT coin is attributed to its asset -/// id in the `coins` table (so `get_cats`/`get_token` become complete). Coins whose parent -/// spend is unavailable, or that are not NFT/DID/CAT, are skipped. +/// id in the `coins` table (so `get_cats`/`get_token` become complete). +/// +/// # A row is examined ONCE, and the outcome is what gets remembered +/// +/// A coin's parent spend is immutable chain history, and so is what that spend says the coin is. +/// So a row this pass RESOLVED and could not attribute — an NFT, a DID, an odd-amount plain XCH +/// coin that uncurries to [`Reconstructed::Unknown`] — will answer identically on every future +/// pass, forever, at the cost of one outbound chain read each time. +/// +/// That is not hypothetical: `upsert_nft`/`upsert_did` write the `nfts`/`dids` tables and never +/// touch `coins.asset_id`, so **every NFT and DID the wallet holds** stays a candidate for the +/// life of the replica. A negative cache over the *lookup* cannot help, because these lookups +/// SUCCEED. What has to be remembered is the *attribution outcome*. +/// +/// [`WalletDb::mark_attribution_examined`] records it, so the pass costs one indexed read plus +/// work proportional to **newly-arrived** rows rather than to every row ever synced. +/// +/// A row whose parent could not be READ ([`LineageAnswer::Unavailable`]) is deliberately NOT +/// marked: nothing was learned about it, and marking it would convert a chain-source outage into +/// a permanent refusal to name the wallet's own money. pub async fn reconstruct_coins( db: &WalletDb, lineage: &dyn LineageSource, @@ -386,11 +475,25 @@ pub async fn reconstruct_coins( if !is_candidate(c, plain_puzzle_hashes) { continue; } - let Some(parent) = lineage + let parent = match lineage .parent_spend(&c.parent_coin_info, created as u32) .await? - else { - continue; + { + LineageAnswer::Found(parent) => *parent, + // The chain says there is no such spend. That answer is stable, so the row is + // settled and never costs another read. + LineageAnswer::Absent => { + db.mark_attribution_examined(&c.coin_id).await?; + stats.settled += 1; + continue; + } + // Nothing was learned, whether because no source answered or because this node's own + // budget declined to ask. Either way the row is left UNMARKED so a later pass retries + // it — marking it would turn a momentary refusal into a permanently wrong balance. + LineageAnswer::Unavailable | LineageAnswer::Deferred => { + stats.unresolved += 1; + continue; + } }; let child = coin_from_row(c)?; match reconstruct(prefix, Some(created as u32), &parent, child)? { @@ -413,18 +516,28 @@ pub async fn reconstruct_coins( } Reconstructed::Unknown => {} } + // Marked for EVERY resolved outcome, including the CAT one. `asset_id` alone would do + // for a CAT, but making the mark unconditional on "we read the spend and acted on it" + // means no future reconstruction kind can be added that silently re-reads forever. + db.mark_attribution_examined(&c.coin_id).await?; + stats.settled += 1; } Ok(stats) } -/// Convenience: reconstruct every coin currently in the wallet DB. +/// Reconstruct every coin in the wallet DB that a pass could still learn something from. +/// +/// The candidate set is narrowed in SQL — unspent, unattributed, confirmed, and not already +/// examined — rather than by scanning the whole `coins` table in Rust. The scan this replaces +/// was the standing per-frame cost that made a seeded replica an amplifier: one `SELECT *` plus +/// one outbound chain read per stable row, on every push frame, for the life of the process. pub async fn reconstruct_all( db: &WalletDb, lineage: &dyn LineageSource, prefix: &str, plain_puzzle_hashes: &HashSet, ) -> Result { - let coins = db.all_coins().await?; + let coins = db.unexamined_attribution_candidates().await?; reconstruct_coins(db, lineage, prefix, plain_puzzle_hashes, &coins).await } @@ -454,8 +567,13 @@ mod tests { &self, parent_coin_id: &str, _spent_height: u32, - ) -> Result> { - Ok(self.by_parent.get(parent_coin_id).cloned()) + ) -> Result { + // A parent this map does not hold is one the node could not READ — production's + // answer for an unresolvable parent, not a settled absence. + Ok(LineageAnswer::from_lookup( + self.by_parent.get(parent_coin_id).cloned(), + LineageAnswer::Unavailable, + )) } } @@ -691,4 +809,182 @@ mod tests { odd.amount = "1".into(); assert!(is_candidate(&odd, &phs)); } + + /// A [`LineageSource`] that counts its reads and answers however it was told to. + struct CountingLineage { + answer: LineageAnswerKind, + hits: std::sync::atomic::AtomicUsize, + } + + /// Which answer [`CountingLineage`] gives. Named rather than boolean because the three cases + /// have three different consequences for whether a row may be written off. + #[derive(Clone, Copy)] + enum LineageAnswerKind { + /// A real spend that reconstructs to nothing — the shape of an odd-amount plain XCH coin + /// at the wallet's own p2 hash, and of every NFT/DID coin row after its own table is + /// written. These RESOLVE, which is precisely why a cache over the lookup cannot help. + ResolvesToNothing, + /// The chain answered: no such spend. + Absent, + /// Nothing could be reached. + Unavailable, + } + + #[async_trait] + impl LineageSource for CountingLineage { + async fn parent_spend(&self, _parent: &str, _height: u32) -> Result { + self.hits.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Ok(match self.answer { + LineageAnswerKind::ResolvesToNothing => { + LineageAnswer::Found(Box::new(ParentSpend { + // A spend of a coin that creates nothing the drivers recognise, so + // `reconstruct_parsed` falls through to `Unknown`. + coin: Coin { + parent_coin_info: Bytes32::from([1u8; 32]), + puzzle_hash: Bytes32::from([2u8; 32]), + amount: 1, + }, + puzzle_reveal: vec![0x01], + solution: vec![0x80], + })) + } + LineageAnswerKind::Absent => LineageAnswer::Absent, + LineageAnswerKind::Unavailable => LineageAnswer::Unavailable, + }) + } + } + + fn counting(answer: LineageAnswerKind) -> CountingLineage { + CountingLineage { + answer, + hits: std::sync::atomic::AtomicUsize::new(0), + } + } + + /// An unattributed, unspent, confirmed coin at an UNSUBSCRIBED puzzle hash — a candidate on + /// every pass until something settles it. + fn candidate_row(n: u8) -> CoinRow { + CoinRow { + coin_id: hex::encode([n; 32]), + parent_coin_info: hex::encode([n.wrapping_add(100); 32]), + puzzle_hash: hex::encode([n.wrapping_add(200); 32]), + amount: "1".into(), + created_height: Some(5), + spent_height: None, + asset_id: None, + hint: None, + created_timestamp: None, + spent_timestamp: None, + } + } + + /// **Proves (dig-node#383):** a row whose parent RESOLVES but which cannot be attributed is + /// read exactly once, however many passes run. + /// + /// # Why this fixture and not an unresolvable parent + /// + /// The nearest wrong implementation is the one that shipped: a negative cache over the LOOKUP. + /// It makes an *unresolvable* parent free and is structurally unable to help here, because + /// this lookup succeeds. Every NFT and DID the wallet holds has this shape — the reconstructed + /// row is written to its own table and `coins.asset_id` stays NULL — so choosing a resolving + /// parent is what distinguishes remembering the OUTCOME from remembering the lookup. + /// + /// Ten passes rather than two, so a fix that merely halves the work fails as loudly as one + /// that does nothing. + #[tokio::test] + async fn a_resolving_but_unattributable_row_is_read_once_not_once_per_pass() { + let db = WalletDb::open_in_memory().await.unwrap(); + db.upsert_coins(&[candidate_row(1)]).await.unwrap(); + let lineage = counting(LineageAnswerKind::ResolvesToNothing); + let plain = HashSet::new(); + + for _ in 0..10 { + reconstruct_all(&db, &lineage, "xch", &plain).await.unwrap(); + } + + assert_eq!( + lineage.hits.load(std::sync::atomic::Ordering::SeqCst), + 1, + "ten passes over one settled row must cost one read; anything more is a per-frame \ + outbound chain read for the life of the replica" + ); + } + + /// **Proves (dig-node#383):** a reported ABSENCE settles the row too — asking again gets the + /// same answer, so paying for it again buys nothing. + #[tokio::test] + async fn a_reported_absence_settles_the_row() { + let db = WalletDb::open_in_memory().await.unwrap(); + db.upsert_coins(&[candidate_row(2)]).await.unwrap(); + let lineage = counting(LineageAnswerKind::Absent); + let plain = HashSet::new(); + + for _ in 0..10 { + reconstruct_all(&db, &lineage, "xch", &plain).await.unwrap(); + } + + assert_eq!( + lineage.hits.load(std::sync::atomic::Ordering::SeqCst), + 1, + "a spend the chain says does not exist is settled chain history, not a retry" + ); + } + + /// **Proves (dig-node#383, D3):** an UNREADABLE parent is NOT settled, so a chain-source + /// outage cannot write a coin off. + /// + /// The control that makes this test load-bearing is its sibling above: identical traffic, + /// identical row, and the only difference is whether the source answered. If the mark were + /// applied on every pass rather than on a resolved one, this would report 1 like the others — + /// and the wallet would spend the rest of the replica's life refusing to look at a coin it + /// failed to read once. + #[tokio::test] + async fn an_unreadable_parent_is_retried_rather_than_written_off() { + let db = WalletDb::open_in_memory().await.unwrap(); + db.upsert_coins(&[candidate_row(3)]).await.unwrap(); + let lineage = counting(LineageAnswerKind::Unavailable); + let plain = HashSet::new(); + + for _ in 0..10 { + reconstruct_all(&db, &lineage, "xch", &plain).await.unwrap(); + } + + assert_eq!( + lineage.hits.load(std::sync::atomic::Ordering::SeqCst), + 10, + "nothing was learned about this parent, so every pass must ask again; marking it \ + would turn a transient outage into a permanent wrong balance" + ); + } + + /// **Proves (dig-node#383):** the pass's cost tracks NEWLY-ARRIVED rows, not the size of the + /// replica. + /// + /// This is the property the deferral of the per-frame scan was wrongly assumed to already + /// have. Twenty settled rows plus one new one costs one read, not twenty-one — the difference + /// between a scan a peer can re-trigger for the price of an empty frame and one it cannot. + #[tokio::test] + async fn a_later_pass_pays_only_for_what_newly_arrived() { + let db = WalletDb::open_in_memory().await.unwrap(); + let seeded: Vec = (10..30).map(candidate_row).collect(); + db.upsert_coins(&seeded).await.unwrap(); + let lineage = counting(LineageAnswerKind::ResolvesToNothing); + let plain = HashSet::new(); + + reconstruct_all(&db, &lineage, "xch", &plain).await.unwrap(); + assert_eq!( + lineage.hits.load(std::sync::atomic::Ordering::SeqCst), + 20, + "the first pass must genuinely examine all twenty, or the second pass proves nothing" + ); + + db.upsert_coins(&[candidate_row(40)]).await.unwrap(); + reconstruct_all(&db, &lineage, "xch", &plain).await.unwrap(); + + assert_eq!( + lineage.hits.load(std::sync::atomic::Ordering::SeqCst), + 21, + "the second pass must pay for the one new row only" + ); + } } diff --git a/crates/dig-wallet/src/sage/sync.rs b/crates/dig-wallet/src/sage/sync.rs index b2dc7031..2ca834d9 100644 --- a/crates/dig-wallet/src/sage/sync.rs +++ b/crates/dig-wallet/src/sage/sync.rs @@ -13,7 +13,7 @@ //! mainnet-safely against synthetic `CoinState`s AND the Chia peer simulator — no real //! spends (this PR has none). -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::time::Duration; use chia_protocol::Bytes32; @@ -463,6 +463,85 @@ pub fn coin_state_to_row(state: &CoinState) -> CoinRow { /// wallet *does* own. That is what [`handle_coin_state_update`]'s fail-closed latch is for. pub type SubscribedHashes = HashSet; +/// What a coin sitting at a derived CAT outer puzzle hash IS, known before the coin arrives. +/// +/// Both facts are recovered from the derivation rather than from anything the peer said, which +/// is the whole reason this shape needs no chain read. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct CatIdentity { + /// The CAT's asset id (its TAIL hash). + pub asset_id: Bytes32, + /// The wallet p2 puzzle hash that owns coins at this outer hash. Stored as the row's + /// `hint`, because that — not `puzzle_hash` — is what the CAT balance query scopes by. + pub owner_p2: Bytes32, +} + +/// The CAT outer puzzle hashes this wallet can recognise, mapped to what a coin at each one is. +/// +/// # Why this replaces resolving lineage at admission +/// +/// A CAT coin does not sit at its owner's p2 hash; it sits at the OUTER hash that curries the +/// asset's TAIL around that p2 hash. Because the curry commits to both the asset and the owner, +/// a coin found at `cat_puzzle_hash(owner_p2, asset_id)` IS that asset and IS this wallet's, by +/// construction. So the wallet can derive those hashes itself, SUBSCRIBE them, and let the +/// subscription filter do the proving — no parent spend, no uncurrying, no read budget, and no +/// window in which an unattributed row exists. +/// +/// The alternative the previous rounds took — admit the coin, then read its parent spend to +/// decide whether to keep it — put a remote read on the frame path, where a peer sets the pace. +/// Every high-severity finding across four review rounds lived in that machinery. +/// +/// # What this cannot do +/// +/// It recognises only assets whose id the wallet knows in advance. A CAT whose asset id is +/// unknown cannot have its outer hash derived, so it is not subscribed and not admitted — it +/// reads as ABSENT rather than as a wrong number. That is the acceptable failure direction, and +/// unknown-CAT discovery is deliberately out of scope here (see SPEC 18.11). +#[derive(Debug, Clone, Default)] +pub struct DerivedCats(HashMap); + +impl DerivedCats { + /// Derive the outer hash for every (owner p2 hash x known asset id) pair. + /// + /// Uses `digstore_chain::cat::cat_puzzle_hash`, the one construction the wallet's CAT + /// balance, coin reconstruction and send paths already share. A second spelling of this + /// curry would be a byte-drift bug that decides whether money is counted. + pub fn derive(owner_p2_hashes: &[Bytes32], asset_ids: &[Bytes32]) -> Self { + let mut map = HashMap::new(); + for owner_p2 in owner_p2_hashes { + for asset_id in asset_ids { + let outer = digstore_chain::cat::cat_puzzle_hash(*owner_p2, *asset_id); + map.insert( + outer, + CatIdentity { + asset_id: *asset_id, + owner_p2: *owner_p2, + }, + ); + } + } + Self(map) + } + + /// The derived outer hashes, sorted so a subscription (and a test asserting one) is + /// reproducible regardless of map iteration order. + pub fn hashes(&self) -> Vec { + let mut hashes: Vec = self.0.keys().copied().collect(); + hashes.sort(); + hashes + } + + /// What a coin at `puzzle_hash` is, if this wallet derived that hash. + pub fn identify(&self, puzzle_hash: &Bytes32) -> Option { + self.0.get(puzzle_hash).copied() + } + + /// Whether anything was derived at all. + pub fn is_empty(&self) -> bool { + self.0.is_empty() + } +} + /// Everything one peer session carries across the frames it handles: what it subscribed, how /// far its peer is trusted, and how much of its rollback allowance it has spent. /// @@ -472,7 +551,12 @@ pub type SubscribedHashes = HashSet; /// and lends it to every [`handle_coin_state_update`] call. pub struct SessionState<'a> { /// The puzzle hashes this session subscribed. Empty when the session subscribes nothing. + /// + /// Includes the derived CAT outer hashes, so a $DIG coin is inside the filter rather than an + /// exception to it. pub subscribed: &'a SubscribedHashes, + /// What a coin at each derived CAT outer hash is, so an admitted coin is typed on arrival. + pub derived_cats: &'a DerivedCats, /// What this session's peer is entitled to write, and up to what height. pub authority: WriteAuthority, /// The session's remaining allowance for walking the peak backwards. @@ -483,9 +567,14 @@ pub struct SessionState<'a> { impl<'a> SessionState<'a> { /// A session over `subscribed` entitled to write exactly what `authority` says. - pub fn with_authority(subscribed: &'a SubscribedHashes, authority: WriteAuthority) -> Self { + pub fn with_authority( + subscribed: &'a SubscribedHashes, + derived_cats: &'a DerivedCats, + authority: WriteAuthority, + ) -> Self { Self { subscribed, + derived_cats, authority, rollback: RollbackBudget::new(), refused_peaks: 0, @@ -695,15 +784,35 @@ impl CatchUpBudget { /// Coins at a puzzle hash outside `subscribed` are dropped — they were never requested, so a /// peer offering them is either confused or hostile, and either way the replica must not grow a /// row the wallet cannot account for. +/// +/// # Attribution happens HERE, and costs nothing +/// +/// A coin whose puzzle hash is one of the wallet's derived CAT outer hashes is stored with its +/// `asset_id` and its owner `hint` already filled, because matching that hash IS the proof of +/// both ([`DerivedCats`]). No chain read is issued, and no row is ever written with a NULL +/// `asset_id` that the wallet did not itself mean as XCH — which matters because +/// [`crate::sage::db::WalletDb::unspent_coins`]`(None)` reads `asset_id IS NULL` as *XCH* and +/// feeds it to the spend-input selector. +/// +/// The `hint` is not decoration: the CAT balance query scopes by `hint`, not by `puzzle_hash`, +/// so a row admitted without it is stored, typed, and still invisible to the balance. pub async fn apply_coin_states( db: &WalletDb, states: &[CoinState], subscribed: &SubscribedHashes, + derived_cats: &DerivedCats, ) -> Result<(), SyncError> { let rows: Vec = states .iter() .filter(|s| subscribed.contains(&s.coin.puzzle_hash)) - .map(coin_state_to_row) + .map(|s| { + let mut row = coin_state_to_row(s); + if let Some(cat) = derived_cats.identify(&s.coin.puzzle_hash) { + row.asset_id = Some(hex::encode(cat.asset_id)); + row.hint = Some(hex::encode(cat.owner_p2)); + } + row + }) .collect(); if rows.len() != states.len() { tracing::warn!( @@ -724,7 +833,11 @@ pub async fn apply_coin_states( /// with an even amount is an ordinary XCH coin and is skipped (never fetches a parent /// spend). Attribution reads only; it never signs or broadcasts. pub struct CatAttributor<'a> { - /// The parent-spend source (coinset/peer point-read) uncurrying reads through. + /// The parent-spend source the whole-replica attribution pass reads through. + /// + /// The ONE lineage source this type holds. An earlier shape split it in two because a second, + /// admission-time leg competed with this one for a read allowance; that leg is gone, and with + /// it the starvation the split existed to prevent. pub lineage: &'a dyn LineageSource, /// The address bech32m prefix for any reconstructed NFT/DID addresses. pub prefix: &'a str, @@ -779,13 +892,13 @@ pub async fn handle_coin_state_update( update: &CoinStateUpdate, events: &EventBus, session: &mut SessionState<'_>, -) -> Result<(), SyncError> { +) -> Result { if !session.authority.trust().is_authoritative() { tracing::debug!( claimed_height = update.height, "wallet sync: dropping a coin_state_update from a discovered peer" ); - return Ok(()); + return Ok(FrameApplied::Dropped); } // Judged BEFORE the frame acts, not at the write. A guard sitting on the `set_peak` call would // satisfy "the peak is unchanged" identically while the rollback below had already deleted @@ -794,7 +907,7 @@ pub async fn handle_coin_state_update( // coins included, which is also the conservative reading of coins offered alongside one. let admitted = match session.admit_peak(update.height)? { PeakClaim::Admitted(peak) => peak, - PeakClaim::Refused => return Ok(()), + PeakClaim::Refused => return Ok(FrameApplied::Dropped), }; let current_peak = db.sync_state().await?.peak_height; let mut moved_backwards = false; @@ -823,7 +936,9 @@ pub async fn handle_coin_state_update( ); db.set_initial_sync_complete(false).await?; } - apply_coin_states(db, &update.items, session.subscribed).await?; + // The peak is recorded only after the batch has landed: advancing it first would leave the + // replica claiming to be current at a height whose coins it had not yet written. + apply_coin_states(db, &update.items, session.subscribed, session.derived_cats).await?; db.record_peak(admitted, &hex::encode(update.peak_hash)) .await?; // Incoming-funds arrivals (dig_ecosystem#2548), recorded AFTER the batch has committed and @@ -848,7 +963,21 @@ pub async fn handle_coin_state_update( ); } events.publish(SyncEvent::CoinState); - Ok(()) + Ok(FrameApplied::Applied) +} + +/// Whether a `coin_state_update` frame reached the database at all. +/// +/// Returned so the caller can decide whether the follow-up attribution pass is worth running. A +/// frame dropped for coming from a discovered peer, or for claiming a peak this session will not +/// admit, changed nothing — and running the pass after it was how an *empty, refused* frame from +/// an untrusted peer still drove a whole-replica scan and a chain read per candidate row. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum FrameApplied { + /// The frame's coins and peak were written. + Applied, + /// The frame was refused before any database write. + Dropped, } /// The one peer call [`initial_sync_with_authority`] makes, behind a trait. @@ -898,6 +1027,11 @@ impl PuzzleStateSource for Peer { /// This is where the terminal height meets the session's [`PeakCeiling`] — see /// [`CatchUpReplay::finished_at`], which refuses an over-ceiling terminal rather than arming /// `initial_sync_complete` over it. +// Eight parameters, and each one is a distinct thing the floor check must see for itself. Bundling +// them into a context struct would move the trust inputs — the authority and the attributor — behind +// one argument a caller could assemble wrongly, and this function exists precisely so that no +// caller-side refactor can walk around them. +#[allow(clippy::too_many_arguments)] pub async fn initial_sync_with_authority( peer: &dyn PuzzleStateSource, db: &WalletDb, @@ -906,6 +1040,7 @@ pub async fn initial_sync_with_authority( peer_ip: &str, events: &EventBus, authority: WriteAuthority, + derived_cats: &DerivedCats, ) -> Result<(), SyncError> { let trust = authority.trust(); // THE TRUST BOUNDARY. This is the only place a PEER can set `initial_sync_complete`, and @@ -932,7 +1067,14 @@ pub async fn initial_sync_with_authority( return Err(SyncError::NoPuzzleHashes); } - let subscribed: SubscribedHashes = puzzle_hashes.iter().copied().collect(); + // The set actually REQUESTED and admitted: the wallet's p2 hashes plus the CAT outer hashes + // it derived for its known assets. `puzzle_hashes` itself stays the p2 set, because + // `CatchUpReplay::finished_at` below records it as the ADDRESSES this catch-up covered and the + // read router compares that against the addresses the node follows. A CAT outer hash is not an + // address, and putting one there would answer a different question than the one asked. + let mut requested = puzzle_hashes.clone(); + requested.extend(derived_cats.hashes()); + let subscribed: SubscribedHashes = requested.iter().copied().collect(); let mut previous_height: Option = None; let mut header_hash = genesis_challenge; events.publish(SyncEvent::Start { @@ -957,7 +1099,7 @@ pub async fn initial_sync_with_authority( // path, which drops the peer and backs off. let respond = tokio::time::timeout( PEER_REQUEST_TIMEOUT, - peer.request_puzzle_state(puzzle_hashes.clone(), previous_height, header_hash), + peer.request_puzzle_state(requested.clone(), previous_height, header_hash), ) .await .map_err(|_| { @@ -987,7 +1129,7 @@ pub async fn initial_sync_with_authority( } } - apply_coin_states(db, &respond.coin_states, &subscribed).await?; + apply_coin_states(db, &respond.coin_states, &subscribed, derived_cats).await?; events.publish(SyncEvent::PuzzleBatchSynced); if respond.is_finished { @@ -1036,9 +1178,15 @@ pub async fn run_update_loop( match message.msg_type { ProtocolMessageTypes::CoinStateUpdate => { if let Ok(update) = decode::(&message) { - handle_coin_state_update(db, &update, events, session).await?; - if let Some(a) = attributor { - a.attribute(db).await?; + let applied = handle_coin_state_update(db, &update, events, session).await?; + // Only after a frame that actually WROTE something. A dropped frame leaves the + // replica byte-identical, so a pass over it can only re-examine rows an + // earlier pass already settled — work a peer gets to schedule for free by + // sending frames the node has already refused. + if applied == FrameApplied::Applied { + if let Some(a) = attributor { + a.attribute(db).await?; + } } } } @@ -1123,9 +1271,20 @@ mod tests { HashSet::from([Bytes32::new([OWNED; 32])]) } + /// A wallet that recognises no CAT assets, for the many tests about peak/trust/rollback + /// behaviour that have nothing to do with attribution. + /// + /// `'static` so the session helpers below keep their single-argument shape; an empty + /// derivation makes every coin in those fixtures an ordinary subscribed p2 coin, which is + /// exactly what they mean to exercise. + fn no_cats() -> &'static DerivedCats { + static NO_CATS: std::sync::OnceLock = std::sync::OnceLock::new(); + NO_CATS.get_or_init(DerivedCats::default) + } + /// A session over `subscribed` whose peer the OPERATOR chose — full authority. fn operator(subscribed: &SubscribedHashes) -> SessionState<'_> { - SessionState::with_authority(subscribed, WriteAuthority::Operator) + SessionState::with_authority(subscribed, no_cats(), WriteAuthority::Operator) } /// A `new_peak_wallet` frame claiming `height`, exactly as it arrives on the wire. @@ -1145,13 +1304,14 @@ mod tests { /// A session over a peer this node merely DISCOVERED — writes nothing. fn discovered(subscribed: &SubscribedHashes) -> SessionState<'_> { - SessionState::with_authority(subscribed, WriteAuthority::Discovered) + SessionState::with_authority(subscribed, no_cats(), WriteAuthority::Discovered) } /// A session over a DISCOVERED peer a quorum settled at `anchor` — full authority, bounded. fn corroborated(subscribed: &SubscribedHashes, anchor: u32) -> SessionState<'_> { SessionState::with_authority( subscribed, + no_cats(), WriteAuthority::Corroborated(PeakCeiling::from_corroborated( anchor, SESSION_MAX_LIFETIME, @@ -1198,7 +1358,7 @@ mod tests { state(coin(1, 9, 1_000), Some(10), None), state(coin(2, 9, 2_000), Some(11), None), ]; - apply_coin_states(&db, &states, &subscribed_owned()) + apply_coin_states(&db, &states, &subscribed_owned(), no_cats()) .await .unwrap(); assert_eq!(db.balance(None).await.unwrap(), 3_000); @@ -1209,14 +1369,24 @@ mod tests { async fn later_spend_state_marks_coin_spent() { let db = WalletDb::open_in_memory().await.unwrap(); let c = coin(1, 9, 500); - apply_coin_states(&db, &[state(c, Some(10), None)], &subscribed_owned()) - .await - .unwrap(); + apply_coin_states( + &db, + &[state(c, Some(10), None)], + &subscribed_owned(), + no_cats(), + ) + .await + .unwrap(); assert_eq!(db.balance(None).await.unwrap(), 500); // The peer later reports the same coin as spent. - apply_coin_states(&db, &[state(c, Some(10), Some(20))], &subscribed_owned()) - .await - .unwrap(); + apply_coin_states( + &db, + &[state(c, Some(10), Some(20))], + &subscribed_owned(), + no_cats(), + ) + .await + .unwrap(); assert_eq!(db.balance(None).await.unwrap(), 0); } @@ -1238,6 +1408,7 @@ mod tests { state(coin(2, OWNED, 7_000), Some(20), None), ], &subscribed, + no_cats(), ) .await .unwrap(); @@ -1253,6 +1424,7 @@ mod tests { state(coin(2, OWNED, 7_000), Some(20), None), ], &subscribed, + no_cats(), ) .await .unwrap(); @@ -1357,6 +1529,7 @@ mod tests { &db, &[state(coin(1, 9, 5), Some(10), Some(30))], &subscribed_owned(), + no_cats(), ) .await .unwrap(); @@ -1388,6 +1561,7 @@ mod tests { &db, &[state(coin(1, 9, 5), Some(10), None)], &subscribed_owned(), + no_cats(), ) .await .unwrap(); @@ -1429,6 +1603,7 @@ mod tests { None, )], &subscribed_owned(), + no_cats(), ) .await .unwrap(); @@ -1717,6 +1892,7 @@ mod tests { "127.0.0.1", &events, WriteAuthority::Operator, + no_cats(), ), ) .await @@ -1757,6 +1933,7 @@ mod tests { "127.0.0.1", &events, WriteAuthority::Operator, + no_cats(), ) .await .expect_err("an empty subscription set must be refused, not performed"); @@ -1827,6 +2004,7 @@ mod tests { "127.0.0.1", &events, WriteAuthority::Operator, + no_cats(), ) .await .expect("a non-empty subscription set catches up normally"); @@ -1933,6 +2111,7 @@ mod tests { "127.0.0.1", &events, WriteAuthority::Discovered, + no_cats(), ) .await .expect_err("a discovered peer must not be allowed to run a catch-up"); @@ -2003,6 +2182,7 @@ mod tests { "127.0.0.1", &events, WriteAuthority::Discovered, + no_cats(), ) .await .expect_err("the reconnect must not buy a fresh catch-up"); @@ -2102,7 +2282,7 @@ mod tests { rx, &events, None, - &mut SessionState::with_authority(&subscribed, authority), + &mut SessionState::with_authority(&subscribed, no_cats(), authority), ) .await .unwrap(); @@ -2308,6 +2488,7 @@ mod tests { "127.0.0.1", &events, WriteAuthority::Operator, + no_cats(), ) .await .expect_err("a non-advancing catch-up must be refused"); @@ -2332,7 +2513,7 @@ mod tests { /// CAT can be attributed. (The full uncurry→`get_cats` path is proven in `sage::rpc`.) #[tokio::test] async fn run_update_loop_runs_attribution_when_attributor_present() { - use crate::sage::singleton::{LineageSource, ParentSpend}; + use crate::sage::singleton::LineageSource; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; @@ -2345,9 +2526,9 @@ mod tests { &self, _parent_coin_id: &str, _spent_height: u32, - ) -> crate::sage::Result> { + ) -> crate::sage::Result { self.hits.fetch_add(1, Ordering::SeqCst); - Ok(None) + Ok(crate::sage::singleton::LineageAnswer::Absent) } } @@ -2396,6 +2577,311 @@ mod tests { ); } + /// A lineage source that COUNTS reads and would answer every one of them. + /// + /// Always-answering on purpose. A source that refused would make "zero reads" true for the + /// wrong reason — the nearest wrong implementation is one that reads and is merely rebuffed, + /// and that implementation must fail these tests, not pass them. + use crate::sage::singleton::LineageAnswer; + + struct CountingLineage(std::sync::Arc); + + #[async_trait::async_trait] + impl LineageSource for CountingLineage { + async fn parent_spend( + &self, + _parent_coin_info: &str, + _height: u32, + ) -> crate::sage::Result { + self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Ok(LineageAnswer::Absent) + } + } + + /// The wallet p2 hash and the $DIG asset id used by the derived-hash tests. + fn owner_p2() -> Bytes32 { + Bytes32::from([7u8; 32]) + } + + fn dig_asset() -> Bytes32 { + digstore_chain::dig::DIG_ASSET_ID + } + + /// A confirmed CAT coin sitting at the derived outer hash, with a DISTINCT parent per index. + /// + /// Distinct parents matter: a fixture reusing one parent could be bounded by any caching + /// layer, so it could not tell "issues no reads" from "issues one read and reuses it". + fn cat_coin_at(outer: Bytes32, n: u32) -> CoinState { + let mut parent = [0u8; 32]; + parent[..4].copy_from_slice(&n.to_be_bytes()); + CoinState { + coin: Coin { + parent_coin_info: Bytes32::from(parent), + puzzle_hash: outer, + amount: 1_000 + u64::from(n), + }, + spent_height: None, + created_height: Some(100 + n), + } + } + + /// **Proves (dig-node#380, #382):** a CAT coin at a DERIVED outer puzzle hash is admitted, and + /// is stored carrying BOTH its asset id and its owner hint. + /// + /// # Why the hint is asserted separately from the asset id + /// + /// They are two different failure modes and only one of them is visible in a balance. The CAT + /// balance query scopes by `hint`, not by `puzzle_hash`, so a row admitted with the right + /// `asset_id` and a NULL `hint` is stored, correctly typed, and still reads as zero — which is + /// dig-node#382's exact symptom reached by a new route. Asserting only the asset id would go + /// green on precisely that bug. + #[tokio::test] + async fn a_coin_at_a_derived_cat_hash_arrives_attributed_with_its_owner_hint() { + let db = WalletDb::open_in_memory().await.unwrap(); + let derived = DerivedCats::derive(&[owner_p2()], &[dig_asset()]); + let outer = derived.hashes()[0]; + // The construction is the wallet's canonical one, not a second spelling of the curry. + assert_eq!( + outer, + digstore_chain::cat::cat_puzzle_hash(owner_p2(), dig_asset()), + "the derived hash must be the canonical CAT outer hash" + ); + + let subscribed: SubscribedHashes = std::iter::once(owner_p2()) + .chain(derived.hashes()) + .collect(); + apply_coin_states(&db, &[cat_coin_at(outer, 1)], &subscribed, &derived) + .await + .unwrap(); + + let rows = db.all_coins().await.unwrap(); + assert_eq!(rows.len(), 1, "the derived-hash coin must be admitted"); + assert_eq!( + rows[0].asset_id.as_deref(), + Some(hex::encode(dig_asset()).as_str()), + "admitted with its asset id, from the derivation that matched" + ); + assert_eq!( + rows[0].hint.as_deref(), + Some(hex::encode(owner_p2()).as_str()), + "admitted with its owner hint — the column the CAT balance actually scopes by" + ); + } + + /// **Proves (dig-node#383):** admission issues ZERO outbound chain reads, and a coin at a hash + /// the wallet did NOT derive is still refused. + /// + /// # Why 300 coins over an always-answering source + /// + /// 300 is the same batch size the previous rounds measured a read budget with, so the numbers + /// compare directly: that shape issued roughly one read per admitted coin, and this one must + /// issue none. The source ANSWERS every read it is given, so a surviving read path shows up as + /// a count rather than as an error — an refusing source would hide a real read behind a + /// failure that looks like the intended refusal. + /// + /// The batch mixes a subscribed p2 coin, derived-CAT coins, and a coin at an UNDERIVED CAT + /// hash. The last one is the control: without it "zero reads" would also be satisfied by an + /// implementation that simply admitted everything it was offered. + #[tokio::test] + async fn admitting_a_large_batch_issues_no_chain_reads_and_still_refuses_a_stranger() { + let db = WalletDb::open_in_memory().await.unwrap(); + let derived = DerivedCats::derive(&[owner_p2()], &[dig_asset()]); + let outer = derived.hashes()[0]; + let subscribed: SubscribedHashes = std::iter::once(owner_p2()) + .chain(derived.hashes()) + .collect(); + + let mut states: Vec = (0..300).map(|n| cat_coin_at(outer, n)).collect(); + // A coin at a CAT hash derived for somebody ELSE's p2. Structurally indistinguishable from + // ours to anything that does not check the set, and it must not be written. + let stranger = digstore_chain::cat::cat_puzzle_hash(Bytes32::from([9u8; 32]), dig_asset()); + states.push(cat_coin_at(stranger, 9_000)); + + let reads = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let counting = CountingLineage(reads.clone()); + // Held so the source is genuinely reachable from the session: a counter that no wired + // source could ever increment proves nothing about the code under test. + let _ = &counting as &dyn LineageSource; + + apply_coin_states(&db, &states, &subscribed, &derived) + .await + .unwrap(); + + assert_eq!( + reads.load(std::sync::atomic::Ordering::SeqCst), + 0, + "admission must issue no outbound chain reads at any batch size" + ); + let rows = db.all_coins().await.unwrap(); + assert_eq!( + rows.len(), + 300, + "every derived-hash coin admitted; the stranger's refused" + ); + assert!( + rows.iter() + .all(|r| r.hint.as_deref() == Some(hex::encode(owner_p2()).as_str())), + "every admitted row carries the owner hint" + ); + } + + /// **Proves (dig-node#383):** a catch-up whose peer bursts spent and unspent coins together + /// completes, with no budget anywhere that a burst could exhaust. + /// + /// This is the catch-up-with-spent-burst probe the earlier rounds used, kept so the comparison + /// is direct. Under the previous shape the burst competed for a read allowance and could + /// starve the money path; here the same burst is a pure filter-and-upsert, so the assertion is + /// simply that everything offered at a recognised hash lands, spent rows included. + #[tokio::test] + async fn a_spent_burst_during_catch_up_costs_no_reads_and_loses_no_coins() { + let db = WalletDb::open_in_memory().await.unwrap(); + let derived = DerivedCats::derive(&[owner_p2()], &[dig_asset()]); + let outer = derived.hashes()[0]; + let subscribed: SubscribedHashes = std::iter::once(owner_p2()) + .chain(derived.hashes()) + .collect(); + + let states: Vec = (0..300) + .map(|n| { + let mut state = cat_coin_at(outer, n); + // Half the burst is already spent — the shape a long catch-up actually replays. + if n % 2 == 0 { + state.spent_height = Some(200 + n); + } + state + }) + .collect(); + + let reads = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let counting = CountingLineage(reads.clone()); + let _ = &counting as &dyn LineageSource; + + apply_coin_states(&db, &states, &subscribed, &derived) + .await + .unwrap(); + + assert_eq!( + reads.load(std::sync::atomic::Ordering::SeqCst), + 0, + "a spent burst issues no reads either" + ); + let rows = db.all_coins().await.unwrap(); + assert_eq!(rows.len(), 300, "no coin in the burst is lost"); + assert_eq!( + rows.iter().filter(|r| r.spent_height.is_some()).count(), + 150, + "the spent half is recorded as spent, not dropped" + ); + } + + /// **Proves (dig-node#383):** a frame the node REFUSED schedules no attribution work. + /// + /// # The fixture varies one actor and keeps an honest control + /// + /// Two runs over the same replica, the same seeded candidate row and the same frame. Only the + /// session's trust differs. An operator session must run the pass — that is the control, and + /// without it "zero reads" would be satisfied by an attributor that was simply never wired. + /// A discovered session must not, because its frame is dropped before any database write, so + /// the pass can only re-examine rows an earlier pass already settled. + /// + /// The frame is EMPTY on purpose. An empty, refused frame is the cheapest thing on the wire — + /// roughly forty bytes — and a peer that gets a whole-replica pass for it has an amplifier + /// whatever the read bound says, because the bound meters reads rather than frames. + #[tokio::test] + async fn a_refused_frame_schedules_no_attribution_pass() { + use crate::sage::singleton::LineageSource; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + struct CountingLineage { + hits: Arc, + } + #[async_trait::async_trait] + impl LineageSource for CountingLineage { + async fn parent_spend( + &self, + _parent_coin_id: &str, + _spent_height: u32, + ) -> crate::sage::Result { + self.hits.fetch_add(1, Ordering::SeqCst); + Ok(crate::sage::singleton::LineageAnswer::Absent) + } + } + + /// Run one empty `coin_state_update` through the production loop and report how many + /// parent-spend reads the seeded candidate row cost. + async fn reads_for_one_empty_frame(authoritative: bool) -> usize { + let db = WalletDb::open_in_memory().await.unwrap(); + db.set_peak(10, "aa").await.unwrap(); + // One unattributed, unspent, confirmed coin at an unsubscribed hash: a candidate the + // pass will want to read the moment it runs. + db.upsert_coins(&[CoinRow { + coin_id: hex::encode([9u8; 32]), + parent_coin_info: hex::encode([8u8; 32]), + puzzle_hash: hex::encode([7u8; 32]), + amount: "1".into(), + created_height: Some(5), + spent_height: None, + asset_id: None, + hint: None, + created_timestamp: None, + spent_timestamp: None, + }]) + .await + .unwrap(); + + let events = EventBus::with_capacity(8); + let (tx, receiver) = tokio::sync::mpsc::channel::(4); + let update = CoinStateUpdate { + height: 11, + fork_height: 10, + peak_hash: Bytes32::new([2; 32]), + items: vec![], + }; + tx.send(Message { + msg_type: ProtocolMessageTypes::CoinStateUpdate, + id: None, + data: chia_traits::Streamable::to_bytes(&update).unwrap().into(), + }) + .await + .unwrap(); + drop(tx); + + let hits = Arc::new(AtomicUsize::new(0)); + let lineage = CountingLineage { hits: hits.clone() }; + let plain = HashSet::new(); + let attributor = CatAttributor { + lineage: &lineage, + prefix: "xch", + plain_puzzle_hashes: &plain, + }; + let subscribed = subscribed_owned(); + let mut session = if authoritative { + operator(&subscribed) + } else { + discovered(&subscribed) + }; + run_update_loop(&db, receiver, &events, Some(&attributor), &mut session) + .await + .unwrap(); + hits.load(Ordering::SeqCst) + } + + assert_eq!( + reads_for_one_empty_frame(true).await, + 1, + "control: an APPLIED frame runs the pass, so the fixture really does present a \ + candidate row that costs a read" + ); + assert_eq!( + reads_for_one_empty_frame(false).await, + 0, + "a frame dropped before any database write must schedule no work at all; running the \ + pass after it lets a peer buy a whole-replica scan for an empty frame it already \ + knows will be refused" + ); + } + // --------------------------------------------------------------------------------------- // The corroborated peak ceiling (dig_ecosystem#2851). // --------------------------------------------------------------------------------------- @@ -2437,6 +2923,7 @@ mod tests { "1.2.3.4", &events, authority, + no_cats(), ) .await } @@ -2749,7 +3236,7 @@ mod tests { db: &WalletDb, session: &mut SessionState<'_>, update: CoinStateUpdate, - ) -> Result<(), SyncError> { + ) -> Result { handle_coin_state_update(db, &update, &EventBus::default(), session).await } @@ -2885,6 +3372,7 @@ mod tests { &db, &[state(coin(1, OWNED, 5_000), Some(anchor - 5), None)], &subscribed_owned(), + no_cats(), ) .await .unwrap(); diff --git a/crates/dig-wallet/src/sage/sync_supervisor.rs b/crates/dig-wallet/src/sage/sync_supervisor.rs index c394e438..db3d06fd 100644 --- a/crates/dig-wallet/src/sage/sync_supervisor.rs +++ b/crates/dig-wallet/src/sage/sync_supervisor.rs @@ -44,7 +44,7 @@ //! from custody's persisted PUBLIC keys, which are readable while every wallet is locked. No //! seed is touched and nothing here can sign. -use std::collections::BTreeSet; +use std::collections::{BTreeSet, HashSet}; use std::net::SocketAddr; use std::sync::{Arc, Mutex, RwLock}; use std::time::{Duration, Instant, SystemTime}; @@ -57,6 +57,7 @@ use super::custody::WalletCustody; use super::db::WalletDb; use super::events::EventBus; use super::quorum::{self, Verdict}; +use super::singleton; use super::sync::{self, PeerTrust, SyncError}; use super::watchlist::WatchRegistry; @@ -812,6 +813,7 @@ pub trait SyncSession: Send + Sync { genesis_challenge: Bytes32, events: &EventBus, authority: sync::WriteAuthority, + derived_cats: &sync::DerivedCats, ) -> Result<(), SyncError>; /// Consume peer pushes until the peer disconnects. Consumes the session. @@ -819,11 +821,17 @@ pub trait SyncSession: Send + Sync { /// `session` carries the set this session subscribed in [`SyncSession::catch_up`] (pushed /// coins outside it are dropped, because a peer answers a subscription rather than defining /// one), its peer's trust level, and its rollback allowance. + /// `attributor` is the CAT/singleton attribution pass the supervisor built for this session + /// (`None` when no lineage source is attached). It is threaded through the trait rather than + /// constructed by the session because only the supervisor knows the wallet's plain p2 puzzle + /// hashes, and an implementation that drops it silently reintroduces dig-node#382 — every CAT + /// coin keeping `asset_id = NULL`, so a funded wallet reports a $DIG balance of zero. async fn run( self: Box, db: &WalletDb, events: &EventBus, session: &mut sync::SessionState<'_>, + attributor: Option<&sync::CatAttributor<'_>>, ) -> Result<(), SyncError>; } @@ -1208,6 +1216,52 @@ impl StallWatch { } } +/// The CAT asset ids this node can recognise on sight. +/// +/// Every entry here becomes a derived outer puzzle hash per wallet address, and therefore a +/// subscription — so a coin of that asset arrives already typed. An asset NOT listed here is +/// simply not followed: its coins are never admitted and it reads as ABSENT rather than as a +/// wrong number, which is the failure direction this wallet must have. +/// +/// $DIG is the only entry because it is the only asset the node has a compile-time id for. +/// Discovering CATs whose ids are unknown in advance cannot be done by local derivation at all +/// and is deliberately not attempted here (SPEC 18.11). +pub const KNOWN_CAT_ASSET_IDS: [Bytes32; 1] = [digstore_chain::dig::DIG_ASSET_ID]; + +/// The read-only inputs a CAT/singleton attribution pass needs, held for the life of the +/// supervisor so every session it opens attributes the coins that session syncs. +/// +/// A `CoinState` frame does not carry a CAT's TAIL, so every coin a peer pushes is stored with +/// `asset_id = NULL` and the id is recovered afterwards by uncurrying the parent spend +/// ([`sync::CatAttributor`], design B.6). Without this the recovery never runs and +/// `unspent_coins(Some(asset))` matches nothing on a funded wallet (dig-node#382). +/// +/// The wallet's plain p2 puzzle hashes are deliberately NOT a field: they change as wallets are +/// created, unlocked and watched, so the supervisor takes them from the subscription set it +/// resolved for the current attempt rather than from a snapshot taken at boot. +pub struct Attribution { + /// Resolves a parent coin's spend for the background whole-replica pass. Production passes + /// the shared `chia_query` lineage source, so attribution reads through the SAME peer pool as + /// every other chain read. + /// + /// ONE source, unmetered, and deliberately so. An earlier shape wrapped two legs in separate + /// read bounds because a second leg resolved lineage on the frame path, where a remote peer + /// chose the volume. Nothing reads lineage on the frame path now — an arriving CAT coin is + /// recognised by a locally derived puzzle hash — so this pass is ordinary background work over + /// rows the replica already holds, paced by the sync loop rather than by a peer. + lineage: Arc, + /// The address bech32m prefix used for any reconstructed NFT/DID addresses. + prefix: String, +} + +impl Attribution { + /// Take `lineage` as the background pass's parent-spend source and `prefix` for any + /// reconstructed NFT/DID addresses. + pub fn new(lineage: Arc, prefix: String) -> Self { + Self { lineage, prefix } + } +} + /// Everything the supervisor loop needs. Assembled by [`spawn_supervisor`]. pub struct Supervisor { /// The local replica the session writes into. @@ -1239,6 +1293,19 @@ pub struct Supervisor { /// "the check ran and found nothing" must not look the same. A supervisor built without one /// behaves exactly as it did before this field existed. pub chain_tip: Option>, + /// The CAT asset ids whose outer puzzle hashes this supervisor derives and subscribes. + /// + /// Production passes [`KNOWN_CAT_ASSET_IDS`]. It is a field rather than a constant read at the + /// use site so a test can subscribe an asset its fixture actually issued — a suite that could + /// only ever exercise $DIG could not tell a working derivation from a hard-coded hash. + pub cat_asset_ids: Vec, + /// The CAT/singleton attribution inputs, or `None` for no attribution at all. + /// + /// `None` is honest and load-bearing rather than a silent default: a supervisor with no + /// lineage source cannot uncurry a parent spend, and "attribution is switched off" must not + /// look like "attribution ran and found nothing" — the same distinction [`Self::corroborator`] + /// draws. Production attaches one whenever a chain source exists. + pub attribution: Option>, /// How long one session runs before it is retired. Production passes /// [`SESSION_MAX_LIFETIME`]; see that constant for the costs the number trades. /// @@ -1320,8 +1387,37 @@ impl Supervisor { PeerTrust::Operator | PeerTrust::Corroborated => self.puzzle_hashes.puzzle_hashes(), PeerTrust::Discovered => Vec::new(), }; - let subscribed: sync::SubscribedHashes = puzzle_hashes.iter().copied().collect(); + // The CAT outer hashes this wallet can recognise, derived locally from the p2 set + // above crossed with the assets whose ids the node knows. A coin at one of these IS + // that asset and IS this wallet's, because the outer hash curries the TAIL around the + // p2 hash — so subscribing them is what makes an incoming $DIG coin arrive already + // attributed, with no chain read on the frame path (dig-node#380, #382). + let derived_cats = sync::DerivedCats::derive(&puzzle_hashes, &self.cat_asset_ids); + // The SUBSCRIPTION set, and the only place the derived hashes are allowed to widen + // anything. Three neighbouring sets mean different things and a leak between them is a + // different money bug, so each is built from its own source: + // * `plain_puzzle_hashes` below stays the p2 set — it means "hashes we can sign + // for", and a CAT outer hash there would be read as a spendable ordinary coin; + // * `handle.set_watched` counts ADDRESSES, and a CAT outer hash is not an address; + // * `crate::sage::spend::WalletSigner::puzzle_hashes` is the signer's own set and is + // not reached from here at all. + let subscribed: sync::SubscribedHashes = puzzle_hashes + .iter() + .copied() + .chain(derived_cats.hashes()) + .collect(); let nothing_subscribed = puzzle_hashes.is_empty(); + // Built per ATTEMPT, from the set this attempt actually resolved: these are the coins + // an ordinary XCH holding sits at, and a coin at one of them is skipped rather than + // costing a parent-spend read. A snapshot taken at boot would be empty on every + // install whose wallet is created after start-up. + let plain_puzzle_hashes: HashSet = + puzzle_hashes.iter().map(hex::encode).collect(); + let attributor = self.attribution.as_ref().map(|a| sync::CatAttributor { + lineage: a.lineage.as_ref(), + prefix: &a.prefix, + plain_puzzle_hashes: &plain_puzzle_hashes, + }); // The MEASUREMENT of the subscription set. Paired with the trust recorded above, the // phase can now tell "custody holds nothing" from "this writer was refused" — the two // produce an identical empty set here, and only the first is the benign @@ -1335,7 +1431,12 @@ impl Supervisor { // empty puzzle-hash set, `new_peak_wallet` needs no subscription and the // replica's peak keeps advancing. A DISCOVERED peer subscribes nothing AND // writes nothing — its frames are dropped by `handle_coin_state_update` and - // `run_update_loop` before any DB write, including the peak. In both cases + // `run_update_loop` before any DB write, including the peak, AND before the + // attribution pass. That last clause is load-bearing rather than incidental: with + // the pass sitting unconditionally after the drop, a discovered peer's empty, + // already-refused frames still bought a whole-replica scan and a chain read per + // candidate row, so "writes nothing" was true while "costs nothing" was not. In + // both cases // `is_synced()` stays false — the truth, and what keeps wallet-scoped reads on // the fallback tier. The session is also re-polled below, so a wallet created // after boot is subscribed within seconds rather than at the next disconnect. @@ -1357,6 +1458,7 @@ impl Supervisor { self.genesis_challenge, &self.events, authority, + &derived_cats, ) => result.map_err(CatchUpFailure::Failed), () = self.time.sleep(CATCH_UP_DEADLINE) => Err(CatchUpFailure::TimedOut), // Dropping the `catch_up` future closes the peer. Nothing is aborted mid-write: @@ -1403,6 +1505,23 @@ impl Supervisor { catch_up_ms = self.time.now().duration_since(began).as_millis(), "wallet sync: catch-up complete" ); + // The catch-up replays the wallet's whole coin history, and every CAT coin it + // wrote is unattributed. Running the pass HERE and not only inside the update + // loop is what makes a quiet wallet correct: a replica that syncs from genesis + // and then receives no further pushes would otherwise hold its $DIG forever + // without ever being able to name it (dig-node#382). + // + // Best-effort by design. Attribution reads the chain, and a read failure must + // never turn a completed catch-up into a failed session — the pass is idempotent + // and the update loop retries it on the next push. + if let Some(a) = &attributor { + if let Err(e) = a.attribute(&self.db).await { + tracing::warn!( + error = %e, + "wallet sync: post-catch-up CAT attribution failed; retrying on the next update" + ); + } + } } // Taken before the session is moved into `run` below, so the stall and recovery lines @@ -1411,9 +1530,10 @@ impl Supervisor { // Captured HERE, while the session is still owned: the address is needed after the // select, by which point the session may already have been dropped. let peer_addr_to_avoid = session.peer_addr(); - let mut state = sync::SessionState::with_authority(&subscribed, authority); + let mut state = + sync::SessionState::with_authority(&subscribed, &derived_cats, authority); let outcome = tokio::select! { - result = session.run(&self.db, &self.events, &mut state) => { + result = session.run(&self.db, &self.events, &mut state, attributor.as_ref()) => { if let Err(e) = result { tracing::warn!(error = %e, "wallet sync: update loop ended in error"); } @@ -1541,7 +1661,10 @@ impl Supervisor { let round = match corroborator.corroborate().await { Ok(r) => r, Err(e) => { - tracing::debug!(error = %e, "wallet sync: corroboration probe failed; the peer stays uncorroborated and writes nothing"); + tracing::debug!( + error = %e, + "wallet sync: corroboration probe failed; the peer stays uncorroborated and writes nothing" + ); return SessionTrust::refused(RefusalReason::Undecided); } }; @@ -2286,6 +2409,7 @@ impl SyncSession for ChiaPeerSession { genesis_challenge: Bytes32, events: &EventBus, authority: sync::WriteAuthority, + derived_cats: &sync::DerivedCats, ) -> Result<(), SyncError> { // The EFFECTIVE authority, not `self.trust`: a corroborated discovered peer must reach the // floor check as corroborated, or clearing the quorum would buy it nothing. @@ -2297,6 +2421,7 @@ impl SyncSession for ChiaPeerSession { &self.ip, events, authority, + derived_cats, ) .await } @@ -2312,6 +2437,7 @@ impl SyncSession for ChiaPeerSession { db: &WalletDb, events: &EventBus, session: &mut sync::SessionState<'_>, + attributor: Option<&sync::CatAttributor<'_>>, ) -> Result<(), SyncError> { let receiver = self .receiver @@ -2319,7 +2445,12 @@ impl SyncSession for ChiaPeerSession { .await .take() .ok_or_else(|| SyncError::Peer("sync session already consumed".into()))?; - sync::run_update_loop(db, receiver, events, None, session).await + // Forwarded, never re-decided here. This argument was a hard-coded `None` for the whole + // life of the production sync path (dig-node#382): the attribution pass existed, was + // correct and was unit-tested, and was simply never reached by the only call site that + // ships. Naming the parameter is what makes dropping it a compile warning rather than an + // invisible default. + sync::run_update_loop(db, receiver, events, attributor, session).await } } diff --git a/crates/dig-wallet/src/sage/sync_supervisor/tests.rs b/crates/dig-wallet/src/sage/sync_supervisor/tests.rs index 6b4ece2f..b25bbbc4 100644 --- a/crates/dig-wallet/src/sage/sync_supervisor/tests.rs +++ b/crates/dig-wallet/src/sage/sync_supervisor/tests.rs @@ -40,9 +40,13 @@ const CATCH_UP_HEIGHT: u32 = 6_000_000; // Doubles // --------------------------------------------------------------------------- -/// A peer that reports "caught up, nothing to send" — what a real full node answers to a +/// A peer that answers a subscription in ONE finished batch — what a real full node answers to a /// subscription it has already satisfied. -struct CaughtUpAtOnce; +/// +/// The batch carries whatever coin states the [`Script`] holds, which is normally none. A test +/// that loads them is exercising the ingest filter for real: the catch-up requests puzzle state +/// with `include_hinted`, so a truthful peer answers with coins the wallet never subscribed. +struct CaughtUpAtOnce(Arc