From caa3a43eccbc1079f47446b6192d94564deaefc1 Mon Sep 17 00:00:00 2001 From: TaprootFreak <142087526+TaprootFreak@users.noreply.github.com> Date: Sun, 31 May 2026 14:02:49 +0200 Subject: [PATCH] fix(commit): sync SMT update in broadcast_commit_and_deliver MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bring the send-commit path in line with the mint-commit Phase E so the SMT integration completes synchronously before /api/commit returns 200, closing the race window where a follow-up send reads `account.commitment_public_key` from server state but finds no matching SMT entry yet (the async scanner has not observed the on-chain inscription). The race surfaced as 422 "Unable to get merkle proofs for provided public key" in the regression test `second_send_roundtrip_succeeds_without_prev_commitment_pubkey_field` when run against dev-api.zkcoins.app: a wallet that chains /api/send + /api/commit + /api/send hit the second send before the scanner finished its ~20 s reveal-observation lap on Mutinynet. There was no server code regression — the send path had always relied exclusively on the scanner to integrate the commit, while the mint path already did it inline after the broadcast (Phase E, router.rs::mint_handler). The asymmetry was latent and Mutinynet latency made it visible. Changes: * Extract the Phase E body (state.update + atomic persist_state_and_mark_complete_tx) into a shared helper `apply_commit_and_persist_phase_e` in router.rs. The helper takes a `flow_label` for logging and returns a structured `PhaseEFailure` so the two call sites can preserve their existing flow-specific public error strings ("mint broadcast..." vs "commit broadcast..."). * `mint_handler` now delegates Phase E to the helper. Behaviour is byte-identical for happy path and both Err arms; existing tests (`mint_handler_advances_state_synchronously_with_broadcast`, `mint_handler_atomic_tx_rollback_leaves_state_and_row_consistent`, `mint_handler_in_process_state_advance_collision_returns_503`) continue to pass. * `broadcast_commit_and_deliver` now invokes the helper synchronously between the Bitcoin broadcast and the recipient `receive_coin` mutation. On failure: 503, no retry, no fallback — scanner-replay remains the single source of repair, exactly as in the mint flow. * `broadcast_commit_and_deliver` also switches from the process-wide `NETWORK_CONFIG` lazy_static to `state.esplora_config` so the send-commit path becomes testable with a wiremock Esplora, matching the testability shape already in place for `mint_handler`. Production behaviour is unchanged because `start_rest_node` clones `NETWORK_CONFIG` into that slot. * Update the `commit_handler` doc to describe the new Phase E symmetry and remove the stale "no analogue of the mint state-desync class here" sentence. * Add two new tests in `router_tests.rs` mirroring the existing mint Phase E coverage: - `commit_handler_advances_state_synchronously_with_broadcast` — happy path: /api/send + /api/commit with mocked accepting Esplora and live Postgres, then verify SMT contains pk_0, MMR leaf_count == 1, root_indices has the new entry, and pending_inscriptions row sits at `complete` so `should_skip_scanner_state_update` fires. - `commit_handler_atomic_tx_rollback_leaves_state_and_row_consistent` — install a trigger that fails the in-tx UPDATE to `complete`, assert 503 with the expected error substring, and verify on-disk SMT/MMR/root_index stays untouched and the row stays at `reveal_broadcast` for scanner-replay to integrate from chain. Lock topology and crash-recovery contract are preserved verbatim (both documented in the helper's docstring). The scanner remains the authoritative path for external recovery inscriptions but is now a redundant observer for our own send commits too — exactly as it already was for mint commits. --- node/src/router.rs | 314 ++++++++++++++++----------- node/src/router_tests.rs | 443 +++++++++++++++++++++++++++++++++++++++ node/src/runtime.rs | 84 ++++++-- 3 files changed, 700 insertions(+), 141 deletions(-) diff --git a/node/src/router.rs b/node/src/router.rs index ee1572c7..a53a367a 100644 --- a/node/src/router.rs +++ b/node/src/router.rs @@ -96,6 +96,150 @@ pub(crate) fn lock_or_recover(mutex: &Mutex) -> MutexGuard<'_, T> { }) } +/// Phase E failure modes returned by [`apply_commit_and_persist_phase_e`]. +/// +/// Each variant maps 1:1 to the two distinct error arms in the shared +/// helper: an in-process `state.update` rejection (typically an SMT +/// key-collision-with-different-value, observed-but-rare), or a +/// post-update durable-write rollback. The caller (mint or send) maps +/// the variant onto its own flow-tagged response string so the public +/// error message stays exactly as the wallet-side +/// `KNOWN_SERVER_ERRORS` table expects per endpoint. +#[derive(Debug)] +pub(crate) enum PhaseEFailure { + /// `update_and_snapshot_for_persist` returned an `Err` — the + /// in-process SMT/MMR could not be advanced (typical cause: SMT + /// key collision with different value). The broadcast already + /// landed on chain; the scanner-replay path will reconcile. + StateUpdate, + /// `persist_state_and_mark_complete_tx` failed — the atomic tx + /// rolled back so SMT/MMR/root_index AND the + /// `pending_inscriptions.status -> 'complete'` advance all stayed + /// at their pre-call values on disk. The in-memory SMT/MMR HAVE + /// already mutated; on restart `State::load_from_pg` returns the + /// pre-update on-disk state and the scanner-replay path heals. + DurablePersist, +} + +/// Apply a freshly-broadcast commitment to the in-memory SMT + MMR +/// and persist the resulting snapshot **atomically** with the matching +/// `pending_inscriptions.status -> 'complete'` advance. +/// +/// This is the shared Phase E body invoked by both flows that originate +/// inscriptions on this node: +/// * [`mint_handler`] — for mint commits, immediately after +/// `create_and_broadcast_inscription` returns Ok. +/// * [`crate::runtime::broadcast_commit_and_deliver`] — for send +/// commits, immediately after the user-signed commitment is +/// broadcast. +/// +/// The symmetry matters: before this helper existed, the send path +/// relied exclusively on the async scanner to observe the commit on +/// chain and run `state.update` itself. That left a race window in +/// which a wallet could chain `/api/send` + `/api/commit` and then +/// issue a second `/api/send` whose proof-build walks the SMT for the +/// first send's commitment — and finds it missing because the scanner +/// hadn't yet observed the new inscription (especially on Mutinynet +/// where reveal-broadcast → scanner-observe sits at tens of seconds). +/// Running Phase E synchronously here closes that window: by the time +/// the handler responds 200, the SMT entry for the just-broadcast +/// commitment is committed in memory AND on disk, and the scanner +/// will skip its redundant integration via +/// `should_skip_scanner_state_update`. The scanner remains the +/// authoritative path for external / recovery inscriptions. +/// +/// ## Lock topology (preserved across both callers) +/// The function acquires `state.account_node` only to clone its +/// `Arc>` reference, then drops the account-node guard +/// **before** acquiring the state guard. `std::sync::Mutex` is held +/// only across the synchronous `update_and_snapshot_for_persist` call +/// and is released before the async `persist_state_and_mark_complete_tx` +/// — keeping a `std::sync::Mutex` off any `.await` boundary. +/// +/// ## Error handling (no fallbacks) +/// On Err the caller logs and converts to 503. There is **no in-process +/// retry, no spawn-async-retry, no half-state cleanup attempt** — the +/// scanner-replay path is the single source of repair, identical for +/// mint and send. See the memory rule on no-fallbacks for why this is +/// not a robustness gap. +pub(crate) async fn apply_commit_and_persist_phase_e( + state: &AppState, + commitment: &Commitment, + commit_txid_bytes: &[u8; 32], + flow_label: &'static str, +) -> Result { + // Test-only deterministic hold between the broadcast result and + // the phase-3b state advance. Pre-unlocked in all `test_state` + // constructors so production-shaped tests acquire + drop in one + // step. Holding the guard across a colliding SMT injection lets + // the in-process state.update Err test observe the collision when + // the handler's `state.update` finally runs. Production builds + // compile this out entirely (the field does not exist). + #[cfg(test)] + drop(state.state_advance_release_lock.lock().await); + + let state_advance_outcome = { + let state_arc_for_advance = { + let account_node_guard = lock_or_recover(&state.account_node); + account_node_guard.state().clone() + }; + let mut state_guard = lock_or_recover(&state_arc_for_advance); + state_guard.update_and_snapshot_for_persist(std::slice::from_ref(commitment)) + }; + let (new_root, smt_bytes, mmr_bytes, root_index_entry) = match state_advance_outcome { + Ok(snapshot) => snapshot, + Err(e) => { + // The in-process SMT/MMR could not be advanced — typically + // an SMT key-collision-with-different-value. The broadcast + // already landed on chain; the publisher already advanced + // the row to `reveal_broadcast` BEFORE the broadcast call, + // so the scanner-replay path will pick the inscription up + // from chain and run state.update against the un-mutated + // SMT. + eprintln!( + "{}: in-process state.update failed: {} (broadcast already landed; scanner-replay will reconcile)", + flow_label, e + ); + return Err(PhaseEFailure::StateUpdate); + } + }; + let root_index_ref = root_index_entry.as_ref().map(|(p, s, i)| (p, s, *i as u64)); + match db::persist_state_and_mark_complete_tx( + &state.pool, + &smt_bytes, + &mmr_bytes, + root_index_ref, + &commit_txid_bytes[..], + ) + .await + { + Ok(()) => { + println!( + "{}: state.update persisted + row marked complete. New MMR root: {}", + flow_label, + hex::encode(zkcoins_program::hash::digest_to_bytes(&new_root)) + ); + Ok(new_root) + } + Err(e) => { + // The atomic tx rolled back: SMT/MMR/root_index AND the + // row advance all stayed at their pre-call values on disk. + // The in-memory SMT/MMR HAVE already been mutated (that + // happened above before the await), so they are now ahead + // of disk by exactly one leaf. On restart, + // `State::load_from_pg` returns the pre-update on-disk + // state and the scanner-replay path walks the block, + // observes the row at `reveal_broadcast`, and integrates + // the inscription itself — a clean heal. + eprintln!( + "{}: atomic persist + mark-complete failed: {} (scanner-replay will heal)", + flow_label, e + ); + Err(PhaseEFailure::DurablePersist) + } + } +} + // Define a struct for our application state #[derive(Clone)] pub(crate) struct AppState { @@ -1192,122 +1336,27 @@ async fn mint_handler( // Apply the freshly-broadcast commitment to the in-memory SMT + MMR // and persist the resulting snapshot — together with the // `pending_inscriptions.status = 'complete'` row advance — in ONE - // atomic Postgres transaction (`persist_state_and_mark_complete_tx`). - // The scanner's pre-state.update lookup uses that `complete` marker - // to skip its own redundant integration when it later observes the - // same commit on chain. - // - // Rationale (this is the regression Phase E fixes): the scanner - // observed a mint's commit ~20-30 s after `/api/mint` returned 200. - // A wallet that issued a second mint inside that window walked - // `derive_num_pubkeys_from_smt` against the un-updated SMT, signed - // with the same pubkey index as the first mint, and surfaced - // `Unable to get mmr inclusion proof for the previous root` at the - // prover. Advancing `state.update` synchronously here closes the - // window: the second mint's SMT walk sees the first mint's entry - // immediately. The scanner becomes a redundant observer for our - // own inscriptions and remains the authoritative path for external - // recovery inscriptions and out-of-band commits. - // - // Lock topology: the state lock is acquired AFTER the broadcast - // completes (broadcasting is slow and would otherwise serialize - // all `/api/mint` requests behind a single in-flight inscription). - // - // Crash-recovery contract (the BLOCKER this commit fixed): the - // previous two-step shape (persist SMT/MMR/root_index, then a - // standalone UPDATE to `complete`) opened a window where the - // SMT/MMR/root_index could land on disk while the row stayed at - // `reveal_broadcast`. On restart, `State::load_from_pg` rebuilt the - // in-memory state WITH the new leaf, the scanner re-scanned the - // block, observed `reveal_broadcast` → `should_skip_scanner_state_update` - // returned `false`, and `state.update` ran a second time — the SMT - // insert was an idempotent no-op (same key+value) but - // `mmr.append(leaf)` appended a DUPLICATE leaf, diverging the MMR - // root. The atomic single-tx persist + mark-complete below - // guarantees that on success, the scanner-skip predicate will - // correctly fire on replay. On tx failure, the row stays at - // `reveal_broadcast` and the in-memory state advance was NOT - // persisted to disk (transaction atomicity); the scanner will - // replay cleanly. - // Test-only deterministic hold between the broadcast result and - // the phase-3b state advance. Pre-unlocked in all `test_state` - // constructors so production-shaped tests acquire + drop in one - // step. The in-process state.update Err test holds the guard - // across a colliding SMT injection so the handler observes the - // collision when its `state.update` finally runs. Production - // builds compile this out entirely (the field does not exist). - #[cfg(test)] - drop(state.state_advance_release_lock.lock().await); - - let state_advance_outcome = { - let state_arc_for_advance = { - let account_node_guard = lock_or_recover(&state.account_node); - account_node_guard.state().clone() - }; - let mut state_guard = lock_or_recover(&state_arc_for_advance); - state_guard.update_and_snapshot_for_persist(std::slice::from_ref(&commitment)) - }; - let (new_root, smt_bytes, mmr_bytes, root_index_entry) = match state_advance_outcome { - Ok(snapshot) => snapshot, - Err(e) => { - // The in-process SMT/MMR could not be advanced — typically - // an SMT key-collision-with-different-value (a concurrent - // mint race that slipped the phase-2 re-derive gate, or a - // genuine bug). The broadcast already landed on chain, but - // the caller's mint was NOT integrated synchronously. The - // publisher already advanced the row to `reveal_broadcast` - // BEFORE the broadcast call; we keep it there so the - // scanner-replay path will pick the inscription up from - // chain and run state.update against the un-mutated SMT. - // Return 503 so the wallet knows the mint did NOT land - // synchronously and can poll for completion. - eprintln!( - "mint_handler: in-process state.update failed: {} (broadcast already landed; scanner-replay will reconcile)", - e - ); - return handler_error_response( - StatusCode::SERVICE_UNAVAILABLE, - "mint broadcast landed on chain but in-process state advance failed; scanner will reconcile", - ); - } - }; - let root_index_ref = root_index_entry.as_ref().map(|(p, s, i)| (p, s, *i as u64)); - match db::persist_state_and_mark_complete_tx( - &state.pool, - &smt_bytes, - &mmr_bytes, - root_index_ref, - &commit_txid_bytes, - ) - .await + // atomic Postgres transaction. The shared implementation lives in + // [`apply_commit_and_persist_phase_e`], which is also invoked from + // the send path in [`crate::runtime::broadcast_commit_and_deliver`] + // so the two flows that originate inscriptions on this node both + // integrate them synchronously and the scanner becomes a redundant + // observer for our own commits. See the helper's docstring for the + // full rationale (race window, lock topology, crash-recovery + // contract). + if let Err(failure) = + apply_commit_and_persist_phase_e(&state, &commitment, &commit_txid_bytes, "mint_handler") + .await { - Ok(()) => { - println!( - "mint_handler: state.update persisted + row marked complete. New MMR root: {}", - hex::encode(zkcoins_program::hash::digest_to_bytes(&new_root)) - ); - } - Err(e) => { - // The atomic tx rolled back: SMT/MMR/root_index AND - // the row advance all stayed at their pre-call values - // on disk. The in-memory SMT/MMR HAVE already been - // mutated (that happened above before the await), so - // they are now ahead of disk by exactly one leaf. - // On restart, `State::load_from_pg` returns the - // pre-update on-disk state and the scanner-replay path - // walks the block, observes the row at - // `reveal_broadcast`, and integrates the inscription - // itself — a clean heal. Return 503 so the caller - // knows the durable state did not advance. - eprintln!( - "mint_handler: atomic persist + mark-complete failed: {} (scanner-replay will heal)", - e - ); - return handler_error_response( - StatusCode::SERVICE_UNAVAILABLE, - "mint broadcast landed on chain but durable state advance failed; scanner will reconcile", - ); - } + let msg: &'static str = match failure { + PhaseEFailure::StateUpdate => { + "mint broadcast landed on chain but in-process state advance failed; scanner will reconcile" + } + PhaseEFailure::DurablePersist => { + "mint broadcast landed on chain but durable state advance failed; scanner will reconcile" + } + }; + return handler_error_response(StatusCode::SERVICE_UNAVAILABLE, msg); } // ---- 4. COMMIT phase (broadcast OK) --------------------------------- @@ -1454,18 +1503,29 @@ async fn get_proof_handler( /// Accepts a client-signed commitment for a previously generated proof. /// Broadcasts the commitment as a Taproot inscription and delivers the coin to the recipient. /// -/// **Broadcast-then-deliver invariant (zk-coins/node#89).** Unlike -/// the mint flow, the `/api/commit` endpoint receives a *proof_id* the -/// node already generated (in an earlier `/api/send` call), looks up -/// the persisted `CoinProof`, broadcasts its commitment, and only then -/// hands the proof to `receive_coin` for the recipient mutation. The -/// in-memory mutation lives in [`broadcast_commit_and_deliver`] in -/// `runtime.rs`; the broadcast call sits at the very top of -/// that function and returns 503 on failure with NO subsequent state -/// mutation, so there is no analogue of the mint state-desync class -/// here. DO NOT reorder the broadcast and the `receive_coin` call — -/// the audit in zk-coins/node#89 verified this ordering is correct -/// and any future refactor must preserve it. +/// **Broadcast-then-deliver invariant (zk-coins/node#89).** The +/// `/api/commit` endpoint receives a *proof_id* the node already +/// generated (in an earlier `/api/send` call), looks up the persisted +/// `CoinProof`, broadcasts its commitment, advances the SMT/MMR via +/// the shared Phase E helper synchronously, and only then hands the +/// proof to `receive_coin` for the recipient mutation. The in-memory +/// mutation + persistence lives in [`broadcast_commit_and_deliver`] in +/// `runtime.rs`; the broadcast call sits at the very top of that +/// function and returns 503 on failure with NO subsequent state +/// mutation. DO NOT reorder the broadcast and the `receive_coin` call. +/// +/// **Phase E symmetry (this branch).** The send-commit path now runs +/// [`apply_commit_and_persist_phase_e`] synchronously between the +/// broadcast and `receive_coin`, matching `mint_handler`. Before this +/// change the send-commit SMT integration relied exclusively on the +/// async scanner, which left a race window where a wallet that +/// followed `/api/send` + `/api/commit` with a second `/api/send` +/// would walk the SMT for the first commit's pubkey and find it +/// missing — surfacing as 422 `"Unable to get merkle proofs for +/// provided public key"`. The synchronous Phase E call closes that +/// window; the scanner remains the authoritative path for external +/// recovery inscriptions but is now a redundant observer for our own +/// send commits, exactly as for mint commits. async fn commit_handler( State(state): State, Json(request): Json, diff --git a/node/src/router_tests.rs b/node/src/router_tests.rs index 12a3a880..731672d3 100644 --- a/node/src/router_tests.rs +++ b/node/src/router_tests.rs @@ -5864,3 +5864,446 @@ async fn r2_probe_history_limit_clamped_to_max() { let arr: Vec = serde_json::from_str(&body).expect("valid JSON"); assert!(arr.is_empty()); } + +// --------------------------------------------------------------------------- +// Phase E (send-commit branch) — mirrors the mint Phase E tests above. +// +// `broadcast_commit_and_deliver` runs the shared +// `apply_commit_and_persist_phase_e` helper synchronously after the +// Bitcoin broadcast. The tests below assert the two load-bearing +// observable properties from outside the handler: +// +// 1. Happy path: after a 200 response the SMT contains the commit's +// pubkey, the MMR has advanced by one leaf, the matching +// `mmr_root_index` row is present, and the `pending_inscriptions` +// row sits at `complete` — so a scanner re-observation hits +// `should_skip_scanner_state_update`. +// +// 2. Atomic rollback (`PhaseEFailure::DurablePersist`): a trigger that +// blocks the in-tx UPDATE to `complete` rolls the whole transaction +// back. The handler surfaces 503; on-disk SMT/MMR/root_index stays +// unchanged; the row stays at `reveal_broadcast` so scanner-replay +// will integrate the inscription from chain. +// --------------------------------------------------------------------------- + +/// End-to-end mirror of `mint_handler_advances_state_synchronously_with_broadcast` +/// for the send-commit path. Runs `/api/send` (real prover) followed by +/// `/api/commit` (real broadcast against a wiremock Esplora that +/// accepts both the UTXO lookup and the `POST /tx`), then asserts the +/// in-memory and on-disk Phase E aftermath that closes the second-send +/// race (the regression that `second_send_roundtrip_succeeds_without_prev_commitment_pubkey_field` +/// surfaced against Mutinynet). +#[tokio::test] +async fn commit_handler_advances_state_synchronously_with_broadcast() { + use bitcoin::bip32::{ChildNumber, Xpriv, Xpub}; + use bitcoin::hashes::Hash as _; + use bitcoin::secp256k1::{Keypair, PublicKey, SecretKey}; + use testcontainers::{runners::AsyncRunner, ImageExt}; + use testcontainers_modules::postgres::Postgres; + + let pg_container = Postgres::default() + .with_tag("17") + .start() + .await + .expect("failed to start postgres container"); + let host = pg_container.get_host().await.unwrap(); + let port = pg_container.get_host_port_ipv4(5432).await.unwrap(); + let url = format!("postgres://postgres:postgres@{}:{}/postgres", host, port); + let pool = Arc::new( + crate::db::connect_and_migrate(&url) + .await + .expect("connect_and_migrate failed"), + ); + + // `mint_broadcast_mock_server` is publisher-key-agnostic — same + // `0x01` SecretKey used for every test-mode broadcast. It accepts + // the UTXO GET and the `POST /tx` so the commit-side + // `create_and_broadcast_inscription` succeeds. + let mock_server = mint_broadcast_mock_server().await; + + // `test_state()` carries `dead_pool`; swap in the live pool and the + // mock Esplora before exercising the handler. + let mut state = test_state(); + state.pool = Arc::clone(&pool); + state.esplora_config = Arc::new(crate::publisher::EsploraConfig { + url: mock_server.uri(), + is_mainnet: false, + network_name: "Mutinynet".to_string(), + ws_url: None, + }); + + // Derive the same BIP-32 child keys the other commit tests use. + let secret_bytes = include_bytes!("../minting_secret.bin"); + let xpriv = Xpriv::new_master(bitcoin::Network::Signet, secret_bytes).unwrap(); + let secp = secp::Secp256k1::new(); + let pk_0: PublicKey = Xpub::from_priv(&secp, &xpriv) + .derive_pub(&secp, &[ChildNumber::Normal { index: 0 }]) + .unwrap() + .public_key; + let pk_1: PublicKey = Xpub::from_priv(&secp, &xpriv) + .derive_pub(&secp, &[ChildNumber::Normal { index: 1 }]) + .unwrap() + .public_key; + let sk_0: SecretKey = xpriv + .derive_priv(&secp, &[ChildNumber::Normal { index: 0 }]) + .unwrap() + .private_key; + + // ---- /api/send (real prover, returns the post-state hashes) ---- + let account_address = "0x".to_string() + + &hex::encode(zkcoins_program::hash::digest_to_bytes( + &zkcoins_program::types::MINTING_ADDRESS, + )); + let recipient = "0x".to_string() + &hex::encode([0xa1u8; 32]); + let amount: u64 = 1; + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs(); + let mut hasher = Sha256::new(); + hasher.update(account_address.as_bytes()); + hasher.update(recipient.as_bytes()); + hasher.update(amount.to_le_bytes()); + hasher.update(now.to_le_bytes()); + let hash: [u8; 32] = hasher.finalize().into(); + let msg = Message::from_digest(hash); + let kp = Keypair::from_secret_key(&secp, &sk_0); + let sig = secp.sign_schnorr(&msg, &kp); + + let send_body = serde_json::json!({ + "account_address": account_address, + "recipient": recipient, + "amount": amount, + "public_key": hex::encode(pk_0.serialize()), + "next_public_key": hex::encode(pk_1.serialize()), + "signature": hex::encode(sig.serialize()), + "timestamp": now, + }); + let send_req = Request::post("/api/send") + .header("content-type", "application/json") + .body(Body::from(send_body.to_string())) + .unwrap(); + let (send_status, send_body_text) = send_request_with_state(state.clone(), send_req).await; + assert_eq!(send_status, StatusCode::OK, "send failed: {send_body_text}"); + let send_resp: serde_json::Value = serde_json::from_str(&send_body_text).unwrap(); + let proof_id = send_resp["proof_id"].as_u64().unwrap(); + let ash_hex = send_resp["account_state_hash"] + .as_str() + .unwrap() + .to_string(); + let ocr_hex = send_resp["output_coins_root"].as_str().unwrap().to_string(); + + // Pre-commit sanity: pk_0 is NOT yet in the SMT (send_coin_handler + // does not advance the SMT; that is exactly the Phase E gap this + // commit closes for the send branch). + let pk0_smt_key = bitcoin::hashes::sha256::Hash::hash(&pk_0.serialize()).to_byte_array(); + { + let node_guard = state.account_node.lock().unwrap(); + let state_arc = node_guard.state().clone(); + let state_guard = state_arc.lock().unwrap(); + assert!( + state_guard.smt.get(&pk0_smt_key).is_none(), + "post-send / pre-commit: pk_0 must NOT be in SMT yet (commit's Phase E inserts it)" + ); + assert_eq!( + state_guard.mmr.leaf_count(), + 0, + "post-send / pre-commit: MMR must be empty" + ); + } + + // ---- /api/commit (broadcast OK → Phase E runs synchronously) ---- + let ash_bytes = hex::decode(&ash_hex).unwrap(); + let ocr_bytes = hex::decode(&ocr_hex).unwrap(); + let mut commit_message = Vec::with_capacity(ash_bytes.len() + ocr_bytes.len()); + commit_message.extend_from_slice(&ash_bytes); + commit_message.extend_from_slice(&ocr_bytes); + let commitment = shared::commitment::Commitment::new(&sk_0, commit_message.clone()) + .expect("commitment creation"); + assert!(commitment.verify(), "test commitment must verify locally"); + + let commit_body = serde_json::json!({ + "proof_id": proof_id, + "public_key": hex::encode(commitment.public_key.serialize()), + "signature": hex::encode(commitment.signature.serialize()), + "message": hex::encode(&commitment.message), + }); + let commit_req = Request::post("/api/commit") + .header("content-type", "application/json") + .body(Body::from(commit_body.to_string())) + .unwrap(); + let (commit_status, commit_resp_body) = + send_request_with_state(state.clone(), commit_req).await; + assert_eq!( + commit_status, + StatusCode::OK, + "commit must succeed against accepting Esplora + live pool: {}", + commit_resp_body + ); + + // ---- Phase E aftermath: SMT/MMR/root_indices reflect the commit ---- + let state_arc = { + let node_guard = state.account_node.lock().unwrap(); + node_guard.state().clone() + }; + { + let state_guard = state_arc.lock().unwrap(); + assert!( + state_guard.smt.get(&pk0_smt_key).is_some(), + "Phase E regression: commit_handler must advance SMT with pk_0 before returning 200" + ); + assert_eq!( + state_guard.mmr.leaf_count(), + 1, + "Phase E: MMR must hold exactly one new leaf after the commit" + ); + assert!( + state_guard + .root_indices + .contains_key(&state_guard.prev_mmr_root), + "Phase E: root_indices must hold the freshly written prev_mmr_root" + ); + } + + // ---- pending_inscriptions row marked `complete` atomically ---- + let (pending_status,): (String,) = + sqlx::query_as("SELECT status FROM pending_inscriptions ORDER BY id DESC LIMIT 1") + .fetch_one(&*pool) + .await + .expect("a pending row must exist for the broadcast commit"); + assert_eq!( + pending_status, + crate::db::PENDING_STATUS_COMPLETE, + "Phase E: commit_handler must mark pending_inscriptions complete after state.update" + ); + let (commit_txid_bytes,): (Vec,) = + sqlx::query_as("SELECT commit_txid FROM pending_inscriptions ORDER BY id DESC LIMIT 1") + .fetch_one(&*pool) + .await + .expect("commit_txid column must populate"); + assert!( + crate::scanner::should_skip_scanner_state_update( + crate::db::pending_inscription_status_by_commit_txid(&pool, &commit_txid_bytes) + .await + .unwrap() + .as_deref() + ), + "scanner must skip its redundant state.update for a Phase-E-completed send commit" + ); +} + +/// Mirror of `mint_handler_atomic_tx_rollback_leaves_state_and_row_consistent` +/// for the send-commit path. Installs a `BEFORE UPDATE` trigger on +/// `pending_inscriptions` that raises an exception when the new +/// `status` value is `complete`. The trigger fires inside the atomic +/// `persist_state_and_mark_complete_tx` envelope so the SMT/MMR/ +/// root_index UPSERTs and the mark-complete UPDATE all roll back +/// together. `broadcast_commit_and_deliver` converts the failure to +/// 503; on-disk durable state is unchanged; the row stays at +/// `reveal_broadcast` so the scanner-replay path can integrate the +/// inscription from chain without doubling up the MMR leaf. +#[tokio::test] +async fn commit_handler_atomic_tx_rollback_leaves_state_and_row_consistent() { + use bitcoin::bip32::{ChildNumber, Xpriv, Xpub}; + use bitcoin::secp256k1::{Keypair, PublicKey, SecretKey}; + use testcontainers::{runners::AsyncRunner, ImageExt}; + use testcontainers_modules::postgres::Postgres; + + let pg_container = Postgres::default() + .with_tag("17") + .start() + .await + .expect("failed to start postgres container"); + let host = pg_container.get_host().await.unwrap(); + let port = pg_container.get_host_port_ipv4(5432).await.unwrap(); + let url = format!("postgres://postgres:postgres@{}:{}/postgres", host, port); + let pool = Arc::new( + crate::db::connect_and_migrate(&url) + .await + .expect("connect_and_migrate failed"), + ); + + // Trigger raises on every UPDATE that sets `status = 'complete'`. + sqlx::query( + "CREATE OR REPLACE FUNCTION fail_complete_commit() RETURNS trigger AS $$ + BEGIN + IF NEW.status = 'complete' THEN + RAISE EXCEPTION 'simulated mark-complete failure (commit)'; + END IF; + RETURN NEW; + END; + $$ LANGUAGE plpgsql", + ) + .execute(&*pool) + .await + .unwrap(); + sqlx::query( + "CREATE TRIGGER block_complete_commit BEFORE UPDATE ON pending_inscriptions \ + FOR EACH ROW EXECUTE FUNCTION fail_complete_commit()", + ) + .execute(&*pool) + .await + .unwrap(); + + let mock_server = mint_broadcast_mock_server().await; + let mut state = test_state(); + state.pool = Arc::clone(&pool); + state.esplora_config = Arc::new(crate::publisher::EsploraConfig { + url: mock_server.uri(), + is_mainnet: false, + network_name: "Mutinynet".to_string(), + ws_url: None, + }); + + let secret_bytes = include_bytes!("../minting_secret.bin"); + let xpriv = Xpriv::new_master(bitcoin::Network::Signet, secret_bytes).unwrap(); + let secp = secp::Secp256k1::new(); + let pk_0: PublicKey = Xpub::from_priv(&secp, &xpriv) + .derive_pub(&secp, &[ChildNumber::Normal { index: 0 }]) + .unwrap() + .public_key; + let pk_1: PublicKey = Xpub::from_priv(&secp, &xpriv) + .derive_pub(&secp, &[ChildNumber::Normal { index: 1 }]) + .unwrap() + .public_key; + let sk_0: SecretKey = xpriv + .derive_priv(&secp, &[ChildNumber::Normal { index: 0 }]) + .unwrap() + .private_key; + + let account_address = "0x".to_string() + + &hex::encode(zkcoins_program::hash::digest_to_bytes( + &zkcoins_program::types::MINTING_ADDRESS, + )); + let recipient = "0x".to_string() + &hex::encode([0xa2u8; 32]); + let amount: u64 = 1; + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs(); + let mut hasher = Sha256::new(); + hasher.update(account_address.as_bytes()); + hasher.update(recipient.as_bytes()); + hasher.update(amount.to_le_bytes()); + hasher.update(now.to_le_bytes()); + let hash: [u8; 32] = hasher.finalize().into(); + let msg = Message::from_digest(hash); + let kp = Keypair::from_secret_key(&secp, &sk_0); + let sig = secp.sign_schnorr(&msg, &kp); + + let send_body = serde_json::json!({ + "account_address": account_address, + "recipient": recipient, + "amount": amount, + "public_key": hex::encode(pk_0.serialize()), + "next_public_key": hex::encode(pk_1.serialize()), + "signature": hex::encode(sig.serialize()), + "timestamp": now, + }); + let send_req = Request::post("/api/send") + .header("content-type", "application/json") + .body(Body::from(send_body.to_string())) + .unwrap(); + let (send_status, send_body_text) = send_request_with_state(state.clone(), send_req).await; + assert_eq!(send_status, StatusCode::OK, "send failed: {send_body_text}"); + let send_resp: serde_json::Value = serde_json::from_str(&send_body_text).unwrap(); + let proof_id = send_resp["proof_id"].as_u64().unwrap(); + let ash_hex = send_resp["account_state_hash"] + .as_str() + .unwrap() + .to_string(); + let ocr_hex = send_resp["output_coins_root"].as_str().unwrap().to_string(); + + let ash_bytes = hex::decode(&ash_hex).unwrap(); + let ocr_bytes = hex::decode(&ocr_hex).unwrap(); + let mut commit_message = Vec::with_capacity(ash_bytes.len() + ocr_bytes.len()); + commit_message.extend_from_slice(&ash_bytes); + commit_message.extend_from_slice(&ocr_bytes); + let commitment = shared::commitment::Commitment::new(&sk_0, commit_message.clone()) + .expect("commitment creation"); + assert!(commitment.verify(), "test commitment must verify locally"); + + let commit_body = serde_json::json!({ + "proof_id": proof_id, + "public_key": hex::encode(commitment.public_key.serialize()), + "signature": hex::encode(commitment.signature.serialize()), + "message": hex::encode(&commitment.message), + }); + let commit_req = Request::post("/api/commit") + .header("content-type", "application/json") + .body(Body::from(commit_body.to_string())) + .unwrap(); + let (commit_status, commit_resp_body) = send_request_with_state(state, commit_req).await; + + // Trigger fires inside the atomic tx → handler converts to 503. + assert_eq!( + commit_status, + StatusCode::SERVICE_UNAVAILABLE, + "atomic tx rollback must surface 503, body: {}", + commit_resp_body + ); + let v: serde_json::Value = serde_json::from_str(&commit_resp_body).expect("valid JSON"); + assert_eq!(v["success"], false); + assert!( + v["error"] + .as_str() + .unwrap_or("") + .contains("durable state advance failed"), + "response error must explain the durable-persist failure, got: {}", + v["error"] + ); + + // On-disk SMT/MMR/root_index did NOT advance — the atomic + // envelope rolled them back together with the failed UPDATE. + assert_eq!( + crate::db::load_smt(&pool).await.unwrap(), + None, + "atomic-tx rollback must leave smt_state untouched" + ); + assert_eq!( + crate::db::load_mmr(&pool).await.unwrap(), + None, + "atomic-tx rollback must leave mmr_state untouched" + ); + assert!( + crate::db::load_root_indices(&pool) + .await + .unwrap() + .is_empty(), + "atomic-tx rollback must leave mmr_root_index untouched" + ); + + // Pending row stays at `reveal_broadcast`: publisher set it there + // before the broadcast, mark-complete was the call the trigger + // blocked. Scanner-replay on next boot picks up the inscription + // and integrates it via its own state.update path. + let (pending_status,): (String,) = + sqlx::query_as("SELECT status FROM pending_inscriptions ORDER BY id DESC LIMIT 1") + .fetch_one(&*pool) + .await + .expect("a pending row must exist for the broadcasted commitment"); + assert_eq!( + pending_status, + crate::db::PENDING_STATUS_REVEAL_BROADCAST, + "atomic-tx rollback: pending row must stay at reveal_broadcast for scanner-replay to pick up" + ); + let (commit_txid_bytes,): (Vec,) = + sqlx::query_as("SELECT commit_txid FROM pending_inscriptions ORDER BY id DESC LIMIT 1") + .fetch_one(&*pool) + .await + .unwrap(); + assert!( + !crate::scanner::should_skip_scanner_state_update( + crate::db::pending_inscription_status_by_commit_txid(&pool, &commit_txid_bytes) + .await + .unwrap() + .as_deref() + ), + "scanner must NOT skip its state.update for a send commit whose mark-complete failed" + ); + + sqlx::query("DROP TRIGGER block_complete_commit ON pending_inscriptions") + .execute(&*pool) + .await + .unwrap(); +} diff --git a/node/src/runtime.rs b/node/src/runtime.rs index 5739532c..fc58e2f2 100644 --- a/node/src/runtime.rs +++ b/node/src/runtime.rs @@ -22,7 +22,10 @@ use tokio::net::TcpListener; use crate::account_node::{persist_account, CoinProof}; use crate::db; use crate::publisher::{create_and_broadcast_inscription, resume_pending_inscriptions}; -use crate::router::{lock_or_recover, SendCoinResponse}; +use crate::router::{ + apply_commit_and_persist_phase_e, handler_error_response, lock_or_recover, PhaseEFailure, + SendCoinResponse, +}; use crate::NETWORK_CONFIG; use shared::ProofData; use zkcoins_program::hash::digest_to_bytes; @@ -231,12 +234,12 @@ pub async fn start_rest_node( Ok(()) } -/// Broadcast the commit inscription and, on success, deliver the coin -/// to the recipient and persist the account state. This contains the -/// network call (Bitcoin broadcast) and the post-broadcast bookkeeping, -/// plus the success/failure response dispatch — all of which cannot be -/// exercised by unit tests, so the whole function lives in the runtime -/// module that is excluded from the coverage scope. +/// Broadcast the commit inscription and, on success, run the shared +/// Phase E (SMT/MMR advance + atomic persist + `pending_inscriptions` +/// row marked `complete`), then deliver the coin to the recipient and +/// persist the account state. This contains the network call (Bitcoin +/// broadcast) and the post-broadcast bookkeeping, plus the +/// success/failure response dispatch. /// /// **Invariant (zk-coins/node#89).** The broadcast `if let Err(...) /// { return 503 }` MUST stay above every `receive_coin`/`upsert_account` @@ -245,6 +248,20 @@ pub async fn start_rest_node( /// function does not have that bug because its broadcast is already /// the first effect. Any future refactor that moves a state mutation /// above the broadcast re-introduces the state-desync class — do not. +/// +/// **Phase E symmetry.** Between the broadcast and the recipient +/// `receive_coin` mutation, we invoke +/// [`apply_commit_and_persist_phase_e`] synchronously — identical +/// shape to `mint_handler`. Prior to this, the send-commit SMT +/// integration ran only via the async scanner, which surfaced as a +/// race for back-to-back `/api/send` + `/api/commit` + `/api/send` +/// flows: the second send walked the SMT for the first commit's +/// pubkey and found no entry, returning 422 `"Unable to get merkle +/// proofs for provided public key"`. Running Phase E inline closes +/// that window. The scanner remains the recovery path for external +/// inscriptions and re-scans of our own commits hit +/// `should_skip_scanner_state_update` because the `complete` row +/// advance lands atomically here. pub(crate) async fn broadcast_commit_and_deliver( state: &AppState, commitment: Commitment, @@ -256,19 +273,58 @@ pub(crate) async fn broadcast_commit_and_deliver( "Broadcasting user commitment ({} bytes)", commitment_data.len() ); - if let Err(err) = create_and_broadcast_inscription( + // Use `state.esplora_config` (instead of the process-wide + // `NETWORK_CONFIG` lazy_static) so tests can redirect Esplora calls + // at a `wiremock::MockServer`, matching the testability shape + // already in place for `mint_handler`. In production + // `start_rest_node` clones `NETWORK_CONFIG` into this slot so the + // runtime behaviour is unchanged. + let broadcast_outcome = create_and_broadcast_inscription( &commitment_data, crate::db::InscriptionKind::Send, - &NETWORK_CONFIG, + &state.esplora_config, Some(&state.pool), ) + .await; + let commit_txid_bytes: [u8; 32] = match broadcast_outcome { + Ok((commit_txid, _reveal_txid)) => { + use bitcoin::hashes::Hash as _; + commit_txid.to_byte_array() + } + Err(err) => { + eprintln!("Error broadcasting commit inscription: {}", err); + return handler_error_response( + StatusCode::SERVICE_UNAVAILABLE, + "Failed to broadcast commitment inscription on-chain", + ); + } + }; + + // ---- Phase E (broadcast OK) ----------------------------------------- + // Run the shared SMT/MMR advance + atomic persist + mark-complete + // BEFORE the recipient `receive_coin` mutation. Locked-step with + // `mint_handler::Phase E`; see [`apply_commit_and_persist_phase_e`] + // for the full rationale, lock topology, and crash-recovery + // contract. On failure the broadcast already landed on chain — we + // surface 503 (no fallback, no retry) and the scanner-replay path + // is the single source of repair. + if let Err(failure) = apply_commit_and_persist_phase_e( + state, + &commitment, + &commit_txid_bytes, + "broadcast_commit_and_deliver", + ) .await { - eprintln!("Error broadcasting commit inscription: {}", err); - return crate::router::handler_error_response( - StatusCode::SERVICE_UNAVAILABLE, - "Failed to broadcast commitment inscription on-chain", - ); + let msg: &'static str = match failure { + PhaseEFailure::StateUpdate => { + "commit broadcast landed on chain but in-process state advance failed; scanner will reconcile" + } + PhaseEFailure::DurablePersist => { + "commit broadcast landed on chain but durable state advance failed; scanner will reconcile" + } + }; + return handler_error_response(StatusCode::SERVICE_UNAVAILABLE, msg); } let mut updated_proof = coin_proof;