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