bsv-overlay-engine 0.1.4

//! OverlayGASPStorage — bridges the Engine's storage to the GASP protocol.
//!
//! Wraps a reference to the Engine's `Storage` trait plus a topic name, providing
//! the `GASPStorage` interface that `GASPSync` needs to synchronize UTXOs with
//! remote peers.
//!
//! ## finalize_graph
//!
//! When a graph is finalized, the nodes are converted into ordered BEEF byte
//! arrays (ancestors first, root last) and stored in a shared
//! `FinalizedGraphSink`. The caller (Engine) drains the sink and submits each
//! BEEF to `Engine::submit()` with `HistoricalTxNoSpv` mode: after
//! `GASPSync::sync()` returns, or, under a per-peer sync budget, after every
//! completed UTXO (`gasp::FinalizedGraphHook`, bsv-low #552).
//!
//! ## find_needed_inputs
//!
//! Parses the node's raw transaction hex to determine what inputs are needed:
//! - A proven node ends the walk unless its topic manager names inputs needed
//!   for overlay history: when its output is not yet admissible, those not
//!   held; when it is admissible, those neither held nor landed (lane E1D).
//! - If no proof, all transaction inputs are requested (minus any already
//!   known in local storage).
//!
//! ## validate_graph_anchor (bsv-low #551)
//!
//! Before a graph is finalized it is checked the way the reference checks it
//! (`OverlayGASPStorage.ts` `validateGraphAnchor`): the ROOT node's BEEF is
//! verified by the engine's own SPV check, then the ordered BEEFs are replayed
//! through the topic manager over a set of coins, and the graph is discarded
//! WHOLE unless its root is a coin at the end. That is what makes the
//! `HistoricalTxNoSpv` submits of `finalize_graph` rest on a true premise.
//!
//! Ported from `~/bsv/overlay-services/src/GASP/OverlayGASPStorage.ts` (388 lines).

use crate::beef_limits;

use std::collections::{HashMap, HashSet};
use std::rc::Rc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Mutex;

use async_trait::async_trait;
use bsv_rs::transaction::{Beef, ChainTracker, ChainTrackerError, MerklePath, Transaction};
use tracing::{debug, error, warn};

use crate::gasp::{GASPError, GASPStorage};
use crate::storage::Storage;
use crate::topic_manager::TopicManager;
use crate::types::{
    GASPInputRequest, GASPNode, GASPNodeResponse, GASPOutput, SubmitMode, TopicAdmittanceContext,
};

/// The most store reads [`OverlayGASPStorage::find_needed_inputs`] spends
/// on asking whether an admitted node's named inputs have LANDED (E1D's
/// re-ask; the E1D lens fold, L3): two per transaction at most (its applied
/// row, then its held outputs). A transaction past it is requested from the
/// peer. The figure is the door's own per-topic bound (`PREDECESSOR_READS`).
const LANDED_READS_PER_NODE: usize = 16;

/// A GASP node stored during in-progress graph construction.
#[derive(Debug, Clone)]
struct PendingNode {
    node: GASPNode,
    #[allow(dead_code)]
    spent_by: Option<String>,
    /// Children of this node (nodes whose transactions are inputs to this one).
    children: Vec<String>, // keys into the pending_graphs map
}

/// A finalized graph ready for Engine submission.
///
/// Contains ordered BEEF byte arrays (ancestors first, root last) and
/// the topic they belong to.
#[derive(Debug, Clone)]
pub struct FinalizedGraph {
    /// Topic this graph belongs to.
    pub topic: String,
    /// Ordered BEEF byte arrays — ancestors first, root transaction last.
    pub beefs: Vec<Vec<u8>>,
}

/// Shared container for finalized graph data.
///
/// The Engine creates this before sync and passes it to `OverlayGASPStorage`.
/// After sync completes, the Engine drains the list and submits each BEEF.
pub type FinalizedGraphSink = Rc<Mutex<Vec<FinalizedGraph>>>;

/// Create a new empty finalized graph sink.
pub fn new_finalized_graph_sink() -> FinalizedGraphSink {
    Rc::new(Mutex::new(Vec::new()))
}

/// `GASPStorage` implementation that delegates to the Engine's `Storage` trait.
///
/// Tracks one topic at a time — create a new instance per (topic, peer) sync.
/// During sync, incoming graph nodes are accumulated in memory. On `finalize_graph`,
/// the accumulated nodes are converted to BEEF format and pushed to the shared
/// `FinalizedGraphSink` for later Engine submission.
pub struct OverlayGASPStorage<'a> {
    /// The Engine's storage backend.
    storage: &'a dyn Storage,
    /// Topic being synchronized.
    topic: String,
    /// Optional topic manager for overlay-specific history behind proven nodes.
    topic_manager: Option<&'a dyn TopicManager>,
    /// Temporary graphs being constructed during sync.
    /// Key: node identifier (graph_id for root, "txid.outputIndex" for children).
    /// Value: the pending node with its relationship data.
    pending_graphs: Mutex<HashMap<String, PendingNode>>,
    /// Shared sink for completed graphs ready for Engine submission.
    finalized_sink: FinalizedGraphSink,
    /// If true, finalize uses strict `to_beef(false)` so a missing ancestor
    /// fails loud rather than silently emitting a partial BEEF. Defaults to
    /// `false` (tolerant `to_beef(true)`) — byte-identical to today. Only set
    /// `true` when ancestor hydration is enabled, so post-hydration
    /// completeness is enforced.
    strict_beef: bool,
    /// The engine's chain tracker, for the anchor's Bitcoin check. `None` is
    /// the reference's `'scripts only'`: a merkle path is accepted unchecked,
    /// every unproven input's script still runs.
    chain_tracker: Option<&'a dyn ChainTracker>,
    /// [`crate::engine::Engine::set_script_verification`], handed over so the
    /// anchor check obeys the same switch as `Engine::submit`. Default `true`.
    verify_scripts: bool,
    /// [`crate::engine::Engine::set_walk_budget`], handed over so the anchor
    /// check's script walk runs under the same two limbs as `Engine::submit`
    /// (bsv-low #592). Default [`crate::engine::DoorBudget::DEFAULT`].
    walk_budget: crate::engine::DoorBudget,
    /// The peer this instance syncs from: the key of its deferred graphs
    /// (bsv-low #555, [`Self::with_peer`]).
    peer: String,
    /// Whether that peer is CONFIGURED (`SyncTarget::Peers`), written into
    /// each record saved (the delta-2 fold's D2-M2, [`Self::with_configured_peer`]).
    configured_peer: bool,
    /// The graph the anchor check last PASSED, parsed and checked, kept for
    /// its `finalize_graph` so a graph is parsed once (bsv-low #586). Dropped
    /// by anything that changes or discards a pending graph.
    anchored: Mutex<Option<AnchoredGraph>>,
}

/// A graph the anchor check passed: what `finalize_graph` assembles from.
struct AnchoredGraph {
    graph_id: String,
    strict_beef: bool,
    order: Vec<String>,
    assembly: GraphAssembly,
}

/// Lends the engine's tracker to the anchor check and notes whether it
/// FAULTED: a tracker that could not answer is a fault of the moment, not a
/// verdict on the graph (see `validate_graph_anchor`). Noted here, on the
/// call, so no error text is matched.
struct FaultNotingTracker<'a> {
    inner: &'a dyn ChainTracker,
    faulted: AtomicBool,
}

#[async_trait]
impl ChainTracker for FaultNotingTracker<'_> {
    async fn is_valid_root_for_height(
        &self,
        root: &str,
        height: u32,
    ) -> Result<bool, ChainTrackerError> {
        let answer = self.inner.is_valid_root_for_height(root, height).await;
        if answer.is_err() {
            self.faulted.store(true, Ordering::Relaxed);
        }
        answer
    }

    async fn current_height(&self) -> Result<u32, ChainTrackerError> {
        self.inner.current_height().await
    }
}

/// What the anchor check needs of a pending graph, assembled under the lock.
struct AnchorInput {
    /// The root node's transaction id.
    root_txid: String,
    /// The root node's BEEF (tolerant: a source the graph does not carry is
    /// absent, and merged from local storage or refused by the verifier).
    root_beef: Vec<u8>,
    /// Inputs of the root's UNPROVEN ancestry that the graph does not carry.
    absent_sources: Vec<(String, u32)>,
    /// The keys of the graph's nodes in finalize order: ancestors first,
    /// root last. Each one's BEEF is assembled when the replay reaches it
    /// and dropped after it (bsv-low #586): the check never holds the
    /// graph's BEEFs together.
    order: Vec<String>,
    /// The graph's transactions and proofs, parsed once and CHECKED: every
    /// BEEF of `order` can be assembled from it.
    assembly: GraphAssembly,
}

/// One transaction of a pending graph, parsed once.
struct ParsedTx {
    tx: Transaction,
    txid: String,
}

/// One node of a pending graph as its BEEFs need it.
struct AssemblyNode {
    /// Shared by every node of one transaction (one node per output index
    /// an unproven spender asked for).
    parsed: Rc<ParsedTx>,
    /// The node's own proof. A proven node is a leaf of every BEEF.
    proof: Option<MerklePath>,
    /// Per input of an UNPROVEN node: the key of the node that holds its
    /// source, when the graph has one. Empty for a proven node.
    sources: Vec<Option<String>>,
}

/// A pending graph's BEEFs, assembled from its raw transactions and proofs
/// parsed ONCE each (bsv-low #586, zanaadu-v2 #377).
///
/// Before it, `get_beef_for_node` hydrated a `source_transaction` tree by
/// recursion: every reference to a shared ancestor rebuilt that ancestor's
/// whole subtree (parse, raw bytes and hex of each transaction, cloned per
/// parent), and serialized it at every level, so an unmined graph with
/// fan-in cost its PATHS, not its transactions: Zanaadu measured 36x the
/// fetched bytes on 13 nodes, each further unproven layer multiplying by its
/// fan-in. The reference hydrates the same way (`OverlayGASPStorage.ts`
/// `getBEEFForNode`, f999e0c1a: `hydrator` parses a node again for every
/// input that reaches it; JavaScript shares nothing between the copies
/// either), so it has the same multiplier.
///
/// Here a transaction is parsed once per graph (keyed by its bytes), a node
/// names its sources by KEY, and a node's BEEF is the walk
/// `Transaction::to_beef` made over the hydrated tree (`bsv-rs`
/// `collect_ancestors`: depth first, inputs from the last to the first, a
/// proven node a leaf, a txid collected once, a transaction after its
/// sources; proofs deduplicated by height and root), made over the keys
/// instead, then the SDK's own `Beef` (`merge_bump`, `merge_transaction`,
/// `to_binary`, whose sort is the SDK's). The bytes are the ones the tree
/// gave, for every graph: the frozen digests of `tests/gasp_topic_manager.rs`
/// and of the witness hold.
///
/// Both walks keep their work on the heap (the recursion they replace used
/// the stack, once per unproven link), and neither follows a link back into
/// a node it is still inside.
#[derive(Default)]
struct GraphAssembly {
    /// By the SHA-256 of a transaction's raw bytes.
    txs: HashMap<[u8; 32], Rc<ParsedTx>>,
    /// The transaction of each node key asked for.
    parsed: HashMap<String, Rc<ParsedTx>>,
    /// By node key.
    nodes: HashMap<String, Rc<AssemblyNode>>,
    /// (node key, strict) whose BEEF is known to assemble.
    checked: HashSet<(String, bool)>,
}

impl GraphAssembly {
    /// The transaction of the node `key`, parsed once per graph.
    fn parsed_tx(
        &mut self,
        key: &str,
        refs: &HashMap<String, PendingNode>,
    ) -> Result<Rc<ParsedTx>, GASPError> {
        if let Some(parsed) = self.parsed.get(key) {
            return Ok(parsed.clone());
        }
        let pending = refs
            .get(key)
            .ok_or_else(|| GASPError::Other(format!("Node {key} not found in graph refs")))?;
        let unparsed =
            |e: bsv_rs::Error| GASPError::Other(format!("Failed to parse raw_tx for {key}: {e}"));
        let raw = bsv_rs::primitives::from_hex(&pending.node.raw_tx).map_err(unparsed)?;
        let parsed = match self.txs.entry(bsv_rs::primitives::hash::sha256(&raw)) {
            std::collections::hash_map::Entry::Occupied(known) => known.get().clone(),
            std::collections::hash_map::Entry::Vacant(new) => {
                let tx = Transaction::from_binary(&raw).map_err(unparsed)?;
                let txid = tx.id();
                new.insert(Rc::new(ParsedTx { tx, txid })).clone()
            }
        };
        self.parsed.insert(key.to_string(), parsed.clone());
        Ok(parsed)
    }

    /// The node `key`: its transaction, its proof, and where the graph
    /// holds each input's source.
    fn node(
        &mut self,
        key: &str,
        refs: &HashMap<String, PendingNode>,
    ) -> Result<Rc<AssemblyNode>, GASPError> {
        if let Some(node) = self.nodes.get(key) {
            return Ok(node.clone());
        }
        let parsed = self.parsed_tx(key, refs)?;
        let proof =
            match refs
                .get(key)
                .and_then(|pending| pending.node.proof.as_ref())
            {
                Some(proof_hex) => Some(beef_limits::proof_from_hex(proof_hex).map_err(|e| {
                    GASPError::Other(format!("Failed to parse proof for {key}: {e}"))
                })?),
                None => None,
            };
        // A proven node is a leaf. An unproven one takes each input's source
        // from the node filed under that outpoint, when the graph has it.
        let sources = if proof.is_some() {
            Vec::new()
        } else {
            parsed
                .tx
                .inputs
                .iter()
                .map(|input| {
                    let source_txid = input.get_source_txid().unwrap_or_default();
                    if source_txid.is_empty() {
                        return None;
                    }
                    let child_key = format!("{}.{}", source_txid, input.source_output_index);
                    refs.contains_key(&child_key).then_some(child_key)
                })
                .collect()
        };
        let node = Rc::new(AssemblyNode {
            parsed,
            proof,
            sources,
        });
        self.nodes.insert(key.to_string(), node.clone());
        Ok(node)
    }

    /// Check that the BEEF of `key` can be assembled: every node it reaches
    /// parses (its transaction, its proof), and under `strict` no unproven
    /// node lacks a source (`to_beef(false)`'s rule; tolerant, a missing
    /// source is left out silently, `to_beef(true)`). The errors are the ones
    /// the hydrating recursion gave, in its order: a node on entry, its
    /// sources in input order, its own missing source on the way out. A node
    /// is checked once per graph.
    fn check(
        &mut self,
        key: &str,
        refs: &HashMap<String, PendingNode>,
        strict: bool,
    ) -> Result<(), GASPError> {
        // (node key, the node, the next input to follow)
        let mut stack: Vec<(String, Rc<AssemblyNode>, usize)> = Vec::new();
        let mut inside: HashSet<String> = HashSet::new();
        let mut entering: Option<String> = Some(key.to_string());
        loop {
            if let Some(key) = entering.take() {
                if !self.checked.contains(&(key.clone(), strict)) && inside.insert(key.clone()) {
                    let node = self.node(&key, refs)?;
                    stack.push((key, node, 0));
                }
            }
            let Some((_, node, next)) = stack.last_mut() else {
                return Ok(());
            };
            if let Some(source) = node.sources.get(*next) {
                *next += 1;
                entering.clone_from(source);
                continue;
            }
            let (key, node, _) = stack.pop().expect("the stack is not empty");
            if strict {
                if let Some(missing) = node.sources.iter().rposition(Option::is_none) {
                    let source_txid = node.parsed.tx.inputs[missing]
                        .source_txid
                        .as_deref()
                        .unwrap_or("unknown");
                    // The SDK's own refusal, word for word (`to_beef(false)`).
                    let e = bsv_rs::Error::TransactionError(format!(
                        "Missing source transaction for input {missing} (txid: {source_txid}). \
                         Set allow_partial=true to skip missing source transactions."
                    ));
                    return Err(GASPError::Other(format!(
                        "Failed to serialize BEEF for {key}: {e}"
                    )));
                }
            }
            inside.remove(&key);
            self.checked.insert((key, strict));
        }
    }

    /// The txid of a checked node.
    fn txid_of(&self, key: &str) -> Option<&str> {
        self.nodes.get(key).map(|node| node.parsed.txid.as_str())
    }

    /// The BEEF of a CHECKED node: its transaction and its unproven
    /// ancestry down to the first proven node of every branch, each
    /// transaction once. Nothing is hydrated: the only copies made are the
    /// ones the `Beef` being written holds, dropped with it.
    fn beef_of(&self, key: &str) -> Vec<u8> {
        struct Frame<'n> {
            node: &'n AssemblyNode,
            inputs_left: usize,
        }
        let mut seen: HashSet<&str> = HashSet::new();
        let mut inside: HashSet<&str> = HashSet::new();
        let mut ancestors: Vec<&AssemblyNode> = Vec::new();
        let mut bumps: Vec<MerklePath> = Vec::new();
        let mut bump_index_by_root: HashMap<String, usize> = HashMap::new();
        let mut stack: Vec<Frame> = Vec::new();
        let mut entering: Option<&AssemblyNode> = self.nodes.get(key).map(Rc::as_ref);
        loop {
            if let Some(node) = entering.take() {
                let txid = node.parsed.txid.as_str();
                if seen.contains(txid) || inside.contains(txid) {
                    // Already collected (or being collected): skip it.
                } else if let Some(merkle_path) = &node.proof {
                    // Proven: its proof and itself, and no further. Proofs
                    // are deduplicated by block height and computed root.
                    let root = merkle_path.compute_root(Some(txid)).unwrap_or_default();
                    let slot = format!("{}:{}", merkle_path.block_height, root);
                    match bump_index_by_root.get(&slot) {
                        Some(&existing) if bumps[existing].combine(merkle_path).is_ok() => {}
                        _ => {
                            bump_index_by_root.insert(slot, bumps.len());
                            bumps.push(merkle_path.clone());
                        }
                    }
                    seen.insert(txid);
                    ancestors.push(node);
                } else {
                    inside.insert(txid);
                    stack.push(Frame {
                        node,
                        inputs_left: node.sources.len(),
                    });
                }
            }
            let Some(top) = stack.last_mut() else {
                break;
            };
            if top.inputs_left == 0 {
                let node = stack.pop().expect("the stack is not empty").node;
                inside.remove(node.parsed.txid.as_str());
                seen.insert(node.parsed.txid.as_str());
                ancestors.push(node);
                continue;
            }
            top.inputs_left -= 1;
            if let Some(source) = &top.node.sources[top.inputs_left] {
                entering = self.nodes.get(source).map(Rc::as_ref);
            }
        }

        let mut beef = Beef::new();
        for bump in bumps {
            beef.merge_bump(bump);
        }
        for node in ancestors {
            beef.merge_transaction(node.parsed.tx.clone());
        }
        // The writer grows by doubling: a graph's BEEFs are held together
        // by the finalize, so each gives its slack back.
        let mut bytes = beef.to_binary();
        bytes.shrink_to_fit();
        bytes
    }
}

impl<'a> OverlayGASPStorage<'a> {
    /// Create a new `OverlayGASPStorage` for the given topic.
    ///
    /// The `sink` parameter is a shared container where finalized graphs are
    /// pushed. The Engine holds a clone and drains it after sync.
    pub fn new(
        storage: &'a dyn Storage,
        topic: impl Into<String>,
        sink: FinalizedGraphSink,
    ) -> Self {
        Self {
            storage,
            topic: topic.into(),
            topic_manager: None,
            pending_graphs: Mutex::new(HashMap::new()),
            finalized_sink: sink,
            strict_beef: false,
            chain_tracker: None,
            verify_scripts: true,
            walk_budget: crate::engine::DoorBudget::DEFAULT,
            peer: String::new(),
            configured_peer: false,
            anchored: Mutex::new(None),
        }
    }

    /// Keep a graph the anchor check passed for its finalize.
    fn keep_anchored(&self, graph_id: &str, order: Vec<String>, assembly: GraphAssembly) {
        *self
            .anchored
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(AnchoredGraph {
            graph_id: graph_id.to_string(),
            strict_beef: self.strict_beef,
            order,
            assembly,
        });
    }

    /// Take the graph kept by the anchor check (leaving none).
    fn take_anchored(&self) -> Option<AnchoredGraph> {
        self.anchored
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .take()
    }

    /// Name the peer this instance syncs from, the key under which its
    /// deferred graphs are kept (bsv-low #555). `Engine::start_gasp_sync`
    /// wires it.
    #[must_use]
    pub fn with_peer(mut self, peer: impl Into<String>) -> Self {
        self.peer = peer.into();
        self
    }

    /// Say whether the peer is a CONFIGURED one (`SyncTarget::Peers`) or
    /// one `ls_ship` discovered (bsv-low #555, the delta-2 fold's D2-M2):
    /// each record saved carries it (`DeferredGraph::configured`), so a
    /// storage can reserve part of its ceiling for configured peers.
    /// Default `false`. `Engine::start_gasp_sync` wires it.
    #[must_use]
    pub fn with_configured_peer(mut self, configured: bool) -> Self {
        self.configured_peer = configured;
        self
    }

    /// Check merkle roots of the anchor against this chain tracker (the
    /// engine's: `Engine::start_gasp_sync` wires it). Without one the anchor
    /// check is the reference's `'scripts only'` walk.
    #[must_use]
    pub fn with_chain_tracker(mut self, tracker: &'a dyn ChainTracker) -> Self {
        self.chain_tracker = Some(tracker);
        self
    }

    /// Carry [`crate::engine::Engine::set_script_verification`] into the
    /// anchor check. `false` is that switch's escape hatch, not a mode: the
    /// pre-2026-09-08 structural check (roots only, and nothing without a
    /// tracker). Default `true`.
    #[must_use]
    pub fn with_script_verification(mut self, enabled: bool) -> Self {
        self.verify_scripts = enabled;
        self
    }

    /// Carry [`crate::engine::Engine::set_walk_budget`] into the anchor
    /// check (bsv-low #592).
    #[must_use]
    pub fn with_walk_budget(mut self, budget: crate::engine::DoorBudget) -> Self {
        self.walk_budget = budget;
        self
    }

    /// Consult the topic manager when a proven node needs overlay history.
    /// Without a manager, proven nodes retain the legacy stopping rule.
    #[must_use]
    pub fn with_topic_manager(mut self, manager: &'a dyn TopicManager) -> Self {
        self.topic_manager = Some(manager);
        self
    }

    /// Enable strict-BEEF finalize (`to_beef(false)`), failing loud on a
    /// missing ancestor instead of silently emitting a partial BEEF.
    ///
    /// Defaults to OFF (tolerant). Only enable this together with ancestor
    /// hydration so that, after the chain fallback has spliced in ancestors,
    /// an incomplete graph is discarded + retried rather than stored partial.
    #[must_use]
    pub fn with_strict_beef(mut self, strict: bool) -> Self {
        self.strict_beef = strict;
        self
    }

    /// Take all finalized graphs from the shared sink, leaving it empty.
    ///
    /// Convenience method for callers who hold a reference to this storage
    /// rather than the raw sink.
    pub fn take_finalized_graphs(&self) -> Vec<FinalizedGraph> {
        self.finalized_sink
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .drain(..)
            .collect()
    }

    /// Get the number of pending graphs (for testing).
    #[cfg(test)]
    pub fn pending_graph_count(&self) -> usize {
        let refs = self.pending_graphs.lock().unwrap();
        let graph_ids: std::collections::HashSet<&str> =
            refs.values().map(|pn| pn.node.graph_id.as_str()).collect();
        graph_ids.len()
    }

    /// Get the number of finalized graphs (for testing).
    #[cfg(test)]
    pub fn finalized_graph_count(&self) -> usize {
        self.finalized_sink.lock().unwrap().len()
    }

    /// The keys of a graph's nodes in the order their BEEFs are finalized
    /// (ancestors first, root last): depth first from the root over each
    /// node's `children`, a node after its children, each key once. An
    /// explicit stack (bsv-low #586; it was a recursion, in this order).
    fn ordered_keys(graph_id: &str, refs: &HashMap<String, PendingNode>) -> Vec<String> {
        let mut order: Vec<String> = Vec::new();
        let mut visited: HashSet<&str> = HashSet::new();
        // (node key, its children, the next child to visit)
        let mut stack: Vec<(&str, &[String], usize)> = Vec::new();
        let mut entering: Option<&str> = Some(graph_id);
        loop {
            if let Some(key) = entering.take() {
                if visited.insert(key) {
                    if let Some(pending) = refs.get(key) {
                        stack.push((key, &pending.children, 0));
                    }
                }
            }
            let Some((key, children, next)) = stack.last_mut() else {
                return order;
            };
            if let Some(child) = children.get(*next) {
                *next += 1;
                entering = Some(child.as_str());
            } else {
                order.push((*key).to_string());
                stack.pop();
            }
        }
    }

    /// Compute ordered BEEFs for a graph (ancestors first, root last), each
    /// node's BEEF assembled from the graph's transactions parsed ONCE
    /// ([`GraphAssembly`]).
    ///
    /// `anchored`: the graph as the anchor check left it, already parsed and
    /// checked; without one (a finalize with no anchor check before it) it is
    /// parsed and checked here.
    fn compute_ordered_beefs(
        graph_id: &str,
        refs: &HashMap<String, PendingNode>,
        strict_beef: bool,
        anchored: Option<AnchoredGraph>,
    ) -> Result<Vec<Vec<u8>>, GASPError> {
        let (order, assembly) = match anchored {
            Some(kept) if kept.graph_id == graph_id && kept.strict_beef == strict_beef => {
                (kept.order, kept.assembly)
            }
            _ => {
                let mut assembly = GraphAssembly::default();
                let order = Self::ordered_keys(graph_id, refs);
                for key in &order {
                    assembly.check(key, refs, strict_beef)?;
                }
                (order, assembly)
            }
        };
        Ok(order.iter().map(|key| assembly.beef_of(key)).collect())
    }

    /// Everything `validate_graph_anchor` needs of the pending graph, read in
    /// one pass under the lock (nothing awaits while the guard is held).
    ///
    /// It also enforces the one structural rule the rest relies on: every
    /// node hangs off its parent by an input the parent REALLY spends. The
    /// walk asks the peer for an outpoint and files the answer under the
    /// answer's own txid, so a peer that answers with some other transaction
    /// would otherwise ride into the ordered BEEFs unlinked to anything the
    /// Bitcoin check covers. With the rule, every node of the graph is tied
    /// to the verified root by a chain of txids. (A divergence by addition
    /// from the reference at f999e0c1a, which has no such rule; upstream added
    /// it later in `appendToGraph`: "The GASP child node is not an input of
    /// its declared parent".)
    ///
    /// A graph that cannot be ASSEMBLED (a raw transaction or a proof that
    /// does not parse, or under `strict_beef` a missing ancestor) is
    /// `GASPError::AnchorUnavailable`, not a refusal: before #551 that error
    /// surfaced from `finalize_graph` and failed the UTXO, so the cursor gap
    /// guard asked for it again (#43), and it still does.
    fn read_anchor_input(
        graph_id: &str,
        refs: &HashMap<String, PendingNode>,
        strict_beef: bool,
    ) -> Result<AnchorInput, GASPError> {
        let unassembled = |e: GASPError| {
            GASPError::AnchorUnavailable(format!("graph {graph_id} cannot be assembled: {e}"))
        };
        if !refs.contains_key(graph_id) {
            return Err(GASPError::ValidationFailed(format!(
                "Graph node with ID {graph_id} not found"
            )));
        }

        let mut assembly = GraphAssembly::default();
        let mut absent_sources: Vec<(String, u32)> = Vec::new();
        let mut visited: HashSet<&str> = HashSet::new();
        // (node key, whether a path of UNPROVEN nodes leads from the root to it)
        let mut stack: Vec<(&str, bool)> = vec![(graph_id, true)];
        while let Some((key, in_root_beef)) = stack.pop() {
            if !visited.insert(key) {
                continue;
            }
            let Some(pending) = refs.get(key) else {
                continue;
            };
            let parsed = assembly.parsed_tx(key, refs).map_err(unassembled)?;
            let spends: HashSet<String> = parsed
                .tx
                .inputs
                .iter()
                .filter_map(|input| {
                    let source_txid = input.get_source_txid().ok()?;
                    Some(format!("{}.{}", source_txid, input.source_output_index))
                })
                .collect();
            for child in &pending.children {
                if !spends.contains(child) {
                    return Err(GASPError::ValidationFailed(format!(
                        "node {child} is not an input of its declared parent {key}"
                    )));
                }
            }
            let unproven = pending.node.proof.is_none();
            if in_root_beef && unproven {
                for outpoint in &spends {
                    if !refs.contains_key(outpoint) {
                        if let Some(source) = crate::gasp::parse_outpoint(outpoint) {
                            absent_sources.push(source);
                        }
                    }
                }
            }
            for child in &pending.children {
                stack.push((child.as_str(), in_root_beef && unproven));
            }
        }

        // The root's CHECKED copy is assembled tolerant whatever `strict_beef`
        // says: it is never stored, sources the storage holds are merged into
        // it, and the verifier refuses any source still missing. The ordered
        // BEEFs are assembled exactly as `finalize_graph` will assemble them.
        assembly.check(graph_id, refs, false).map_err(unassembled)?;
        let root_beef = assembly.beef_of(graph_id);
        let root_txid = assembly.txid_of(graph_id).unwrap_or_default().to_string();
        let order = Self::ordered_keys(graph_id, refs);
        for key in &order {
            assembly
                .check(key, refs, strict_beef)
                .map_err(unassembled)?;
        }
        Ok(AnchorInput {
            root_txid,
            root_beef,
            absent_sources,
            order,
            assembly,
        })
    }
}

#[async_trait(?Send)]
impl GASPStorage for OverlayGASPStorage<'_> {
    /// Returns UTXOs known for this topic since the given score/timestamp.
    ///
    /// Delegates to `Storage::find_utxos_for_topic()`, converting `Output`
    /// records to `GASPOutput` with txid, output_index, and score.
    async fn find_known_utxos(
        &self,
        since: u64,
        limit: Option<u64>,
    ) -> Result<Vec<GASPOutput>, GASPError> {
        let outputs = self
            .storage
            .find_utxos_for_topic(
                &self.topic,
                Some(since as f64),
                limit,
                false, // don't need BEEF for listing
            )
            .await
            .map_err(|e| GASPError::StorageError(e.to_string()))?;

        Ok(outputs
            .iter()
            .map(|o| GASPOutput {
                txid: o.txid.clone(),
                output_index: o.output_index,
                score: o.score.unwrap_or(0.0),
            })
            .collect())
    }

    /// Hydrate a GASP node with transaction data from storage.
    ///
    /// Looks up the root output (from graph_id) and searches its BEEF tree
    /// for the requested txid. This is a simplified version that returns
    /// the raw transaction data and proof if available.
    ///
    /// For a full implementation, this would delegate to
    /// `Engine::provide_foreign_gasp_node()`, but since we only hold a
    /// `Storage` reference (not the full Engine), we do a simpler lookup.
    async fn hydrate_gasp_node(
        &self,
        graph_id: &str,
        txid: &str,
        output_index: u32,
        _metadata: bool,
    ) -> Result<GASPNode, GASPError> {
        // Try to find the output with BEEF data
        let output = self
            .storage
            .find_output(txid, output_index, Some(&self.topic), None, true)
            .await
            .map_err(|e| GASPError::StorageError(e.to_string()))?;

        if let Some(output) = output {
            if let Some(ref beef) = output.beef {
                // Parse the BEEF to extract raw transaction hex
                match beef_limits::transaction_from_beef(
                    beef,
                    None,
                    &beef_limits::STORED_BEEF_LIMITS,
                ) {
                    Ok(tx) => {
                        let mut node = GASPNode {
                            graph_id: graph_id.to_string(),
                            raw_tx: tx.to_hex(),
                            output_index,
                            proof: None,
                            tx_metadata: None,
                            output_metadata: None,
                            inputs: None,
                        };
                        if let Some(ref merkle_path) = tx.merkle_path {
                            node.proof = Some(merkle_path.to_hex());
                        }
                        return Ok(node);
                    }
                    Err(e) => {
                        warn!("Failed to parse BEEF for {txid}: {e}");
                    }
                }
            }
        }

        // Fallback: look up without topic filter
        let output = self
            .storage
            .find_output(txid, output_index, None, None, true)
            .await
            .map_err(|e| GASPError::StorageError(e.to_string()))?
            .ok_or_else(|| {
                GASPError::NodeNotFound(format!("Output {txid}.{output_index} not found"))
            })?;

        let beef = output.beef.as_ref().ok_or_else(|| {
            GASPError::NodeNotFound(format!("No BEEF data for {txid}.{output_index}"))
        })?;

        let tx = beef_limits::transaction_from_beef(beef, None, &beef_limits::STORED_BEEF_LIMITS)
            .map_err(|e| GASPError::Other(format!("BEEF parse error: {e}")))?;

        let mut node = GASPNode {
            graph_id: graph_id.to_string(),
            raw_tx: tx.to_hex(),
            output_index,
            proof: None,
            tx_metadata: None,
            output_metadata: None,
            inputs: None,
        };
        if let Some(ref merkle_path) = tx.merkle_path {
            node.proof = Some(merkle_path.to_hex());
        }
        Ok(node)
    }

    /// Determine which inputs are needed to validate this node.
    ///
    /// - A proven node ends the walk if its topic manager names no inputs it
    ///   still needs: for a node whose output is NOT admissible, every named
    ///   input not held (D13); for one whose output IS admissible, every named
    ///   input neither held nor landed (lane E1D, an addition: the reference
    ///   stops at an admissible node). Without a manager, proven nodes always
    ///   stop.
    /// - If no proof, parses the raw transaction and requests all inputs,
    ///   filtering out any inputs already known in local storage.
    /// - Admission errors propagate to GASP's incoming UTXO handler. Errors
    ///   identifying needed inputs are logged and cut off this node, like TS.
    /// - Unproven raw_tx parse failures retain the legacy graceful fallback.
    async fn find_needed_inputs(
        &self,
        node: &GASPNode,
    ) -> Result<Option<GASPNodeResponse>, GASPError> {
        let mut requested_inputs: HashMap<String, GASPInputRequest> = HashMap::new();
        // A proven node whose own output the dry run admits (lane E1D).
        let mut admitted = false;

        if let Some(proof_hex) = &node.proof {
            let Some(manager) = self.topic_manager else {
                return Ok(None);
            };
            let mut tx = Transaction::from_hex(&node.raw_tx)
                .map_err(|e| GASPError::Other(format!("Failed to parse raw_tx: {e}")))?;
            tx.merkle_path = Some(
                beef_limits::proof_from_hex(proof_hex)
                    .map_err(|e| GASPError::Other(format!("Failed to parse proof: {e}")))?,
            );

            // TS passes { dryRun: true }: the node is the peer's and nothing
            // is admitted here, so a manager that writes on admission must not.
            // Utils.toArray(string) defaults to UTF-8, even for hex-like text.
            // This call is outside the needed-input catch in the reference.
            let admittance = manager
                .identify_admissible_outputs(
                    &tx,
                    &[],
                    node.tx_metadata.as_deref().map(str::as_bytes),
                    SubmitMode::HistoricalTx,
                    &TopicAdmittanceContext::DRY_RUN,
                )
                .await
                .map_err(|e| GASPError::Other(e.to_string()))?;
            // The reference stops here (D13). Lane E1D (bsv-low #575,
            // zanaadu-v2 #365) asks the named inputs of an ADMITTED node too:
            // a manager that admits a successor with no coin (the pf head
            // manager over its own head state) ended the walk at the
            // successor while its predecessor had not landed, the graph was
            // the successor alone, and its finalize submit recorded it with
            // no coin; the predecessor, landed on a later tick, stood unspent
            // beside the tip for good. So the predecessor is asked FIRST: a
            // named input that is neither held nor landed (below) is
            // requested and lands before the node in the graph's order, a
            // finalize submit of it that does not land stops the graph, and
            // a named input the peer cannot serve prunes (D14). When every
            // named input is known the walk stops here, as the reference's.
            admitted = admittance.outputs_to_admit.contains(&node.output_index);
            // A node with no input to name stops here, the manager not
            // asked (the reference's call count for such a node).
            if admitted
                && !tx
                    .inputs
                    .iter()
                    .any(|input| input.get_source_txid().is_ok_and(|s| !s.is_empty()))
            {
                return Ok(None);
            }

            // A serialization RED is unreachable: bsv-rs stops collecting ancestors at a merkle path.
            let needed_inputs = async {
                let beef = tx
                    .to_beef(false)
                    .map_err(|e| format!("Failed to serialize proven BEEF: {e}"))?;
                manager
                    .identify_needed_inputs(&beef, None)
                    .await
                    .map_err(|e| e.to_string())
            }
            .await;
            match needed_inputs {
                Ok(inputs) => {
                    for input in inputs {
                        requested_inputs
                            .insert(input.to_graph_id(), GASPInputRequest { metadata: false });
                    }
                }
                Err(e) => {
                    error!(
                        "An error occurred when identifying needed inputs for transaction: {}.{}: {e}",
                        tx.id(), node.output_index
                    );
                    return Ok(None);
                }
            }
        } else {
            // Unproven nodes always need all inputs, independently of the manager.
            let tx = match Transaction::from_hex(&node.raw_tx) {
                Ok(tx) => tx,
                Err(e) => {
                    warn!(
                        "Cannot parse raw_tx for find_needed_inputs (graph_id={}): {e}",
                        node.graph_id
                    );
                    return Ok(None);
                }
            };
            for input in &tx.inputs {
                let source_txid = input.get_source_txid().unwrap_or_default();
                if source_txid.is_empty() {
                    continue;
                }
                let outpoint = format!("{}.{}", source_txid, input.source_output_index);
                requested_inputs.insert(outpoint, GASPInputRequest { metadata: false });
            }
        }

        if requested_inputs.is_empty() {
            return Ok(None);
        }

        // Strip inputs that are already known in local storage.
        let mut to_remove = Vec::new();
        for outpoint in requested_inputs.keys() {
            if let Some((txid, oi)) = crate::gasp::parse_outpoint(outpoint) {
                match self
                    .storage
                    .find_output(&txid, oi, Some(&self.topic), None, false)
                    .await
                {
                    Ok(Some(_)) => {
                        to_remove.push(outpoint.clone());
                    }
                    Ok(None) => {}
                    Err(e) => {
                        debug!("Storage lookup failed for {outpoint}: {e}");
                    }
                }
            }
        }
        for key in &to_remove {
            requested_inputs.remove(key);
        }

        // An admitted node's named inputs are requested only while their
        // landing is unknown (E1D): one whose transaction has an applied row
        // in the topic, or an output held there, has landed, and the walk
        // stops as the reference's. A read that faults keeps the input: one
        // round trip to the peer is the cheaper error. So does a transaction
        // past [`LANDED_READS_PER_NODE`] (the E1D lens fold, L3: a manager
        // names as many inputs as it likes, D14's decoys included, and each
        // costs up to two reads): it is requested, and a landed one the peer
        // serves is a dupe at its finalize submit.
        if admitted {
            let mut landed: HashSet<String> = HashSet::new();
            let mut unlanded: HashSet<String> = HashSet::new();
            let mut reads = 0usize;
            let mut outpoints: Vec<&String> = requested_inputs.keys().collect();
            outpoints.sort();
            for outpoint in outpoints {
                let Some((txid, _)) = crate::gasp::parse_outpoint(outpoint) else {
                    continue;
                };
                if landed.contains(&txid) || unlanded.contains(&txid) {
                    continue;
                }
                if reads + 2 > LANDED_READS_PER_NODE {
                    break;
                }
                reads += 1;
                let record = crate::types::AppliedTransaction {
                    txid: txid.clone(),
                    topic: self.topic.clone(),
                };
                let mut holds = matches!(
                    self.storage.does_applied_transaction_exist(&record).await,
                    Ok(true)
                );
                if !holds {
                    reads += 1;
                    holds = matches!(
                        self.storage.find_outputs_for_transaction(&txid, false).await,
                        Ok(outputs) if outputs.iter().any(|o| o.topic == self.topic)
                    );
                }
                if holds {
                    landed.insert(txid);
                } else {
                    unlanded.insert(txid);
                }
            }
            requested_inputs.retain(|outpoint, _| {
                crate::gasp::parse_outpoint(outpoint)
                    .is_none_or(|(txid, _)| !landed.contains(&txid))
            });
        }

        if requested_inputs.is_empty() {
            return Ok(None);
        }

        Ok(Some(GASPNodeResponse { requested_inputs }))
    }

    /// Append a node to a temporary graph being constructed during sync.
    ///
    /// If `spent_by` is `None`, this is the root node and is keyed by `graph_id`.
    /// Otherwise, the node is keyed by its computed txid.outputIndex, and is
    /// registered as a child of the parent node identified by `spent_by`.
    async fn append_to_graph(
        &self,
        node: &GASPNode,
        spent_by: Option<&str>,
    ) -> Result<(), GASPError> {
        debug!(
            "Appending node to graph {}: tx={}..., oi={}, spent_by={:?}",
            node.graph_id,
            &node.raw_tx[..node.raw_tx.len().min(16)],
            node.output_index,
            spent_by,
        );

        // Compute the key for this node.
        let node_key = if spent_by.is_none() {
            // Root node — keyed by graph_id
            node.graph_id.clone()
        } else {
            // Child node — keyed by txid.outputIndex
            match Transaction::from_hex(&node.raw_tx) {
                Ok(tx) => format!("{}.{}", tx.id(), node.output_index),
                Err(_) => {
                    // Fallback: use raw_tx prefix as key
                    format!(
                        "{}.{}",
                        &node.raw_tx[..node.raw_tx.len().min(64)],
                        node.output_index
                    )
                }
            }
        };

        // The pending graphs change: nothing kept of an anchor check holds.
        self.take_anchored();
        let mut refs = self
            .pending_graphs
            .lock()
            .map_err(|e| GASPError::Other(format!("Lock poisoned: {e}")))?;

        // Insert this node
        refs.insert(
            node_key.clone(),
            PendingNode {
                node: node.clone(),
                spent_by: spent_by.map(std::string::ToString::to_string),
                children: Vec::new(),
            },
        );

        // If spent_by is set, register this node as a child of the parent.
        if let Some(parent_key) = spent_by {
            if let Some(parent) = refs.get_mut(parent_key) {
                parent.children.push(node_key);
            }
        }

        Ok(())
    }

    /// Check a completed graph before anything of it is finalized (bsv-low
    /// #551). The reference's `validateGraphAnchor`
    /// (`OverlayGASPStorage.ts` at f999e0c1a, lines 261 to 297), in its two
    /// steps. An `Err` is a discard: `GASPSync::complete_graph` drops the
    /// graph and NOTHING of it reaches the sink.
    ///
    /// **1. Bitcoin.** The ROOT node's BEEF (`GraphAssembly::beef_of`) is verified
    /// by the engine's own check, `verify_spv_like_the_reference`: the
    /// reference's `spvTx.verify(this.engine.chainTracker)`. With a tracker
    /// every merkle root in that BEEF is checked against it and every unproven
    /// input's script is executed; with none it is `'scripts only'` (roots
    /// accepted unchecked, scripts still run). One verifier, the same switch
    /// (`with_script_verification`) as `Engine::submit`.
    ///
    /// **2. Overlay, all or nothing.** The ordered BEEFs (ancestors first, root
    /// last) are replayed through the topic manager as the reference's
    /// `admitHistoricalBEEF` does: mode `historical-tx`, no off-chain values,
    /// `previous_coins` the inputs whose source outpoint is a coin (the
    /// engine's little-endian u32 input indices), and every output the manager
    /// admits joins the coins. If the ROOT's outpoint is not a coin at the
    /// end, the graph is refused. The set only GROWS, as in the reference at
    /// f999e0c1a: it reads `outputsToAdmit` alone, never `coinsToRetain`, and
    /// removes nothing ("a Set of all historical coins to retain (no need to
    /// remove them)"). The reference passes `{ dryRun: true }` and so does
    /// this replay (`TopicAdmittanceContext::DRY_RUN`): a manager that writes
    /// on admission writes nothing here.
    ///
    /// WITHOUT a topic manager step 2 does not run: there is nothing to
    /// replay, and `Engine::submit` refuses such a topic (`UnsupportedTopic`),
    /// so nothing can be admitted. `Engine::start_gasp_sync` always wires the
    /// manager of a topic it can admit.
    ///
    /// **Divergences by addition** (each with its why):
    ///
    /// - *Coins the storage already holds count.* `find_needed_inputs` strips
    ///   an input the local storage already holds (as the reference does), so
    ///   its node is not in the graph and the replay's own set never sees it.
    ///   The reference then refuses a root that extends a coin it already
    ///   holds: the next head of a head chain, synced one tick after the
    ///   last, would be discarded forever. A previous coin is therefore an
    ///   input in the set OR in the storage for this topic, the exact lookup
    ///   `Engine::submit` makes when finalize submits the same BEEF.
    /// - *Sources the storage already holds are merged for step 1.* For the
    ///   same reason an UNPROVEN node may lack an input's source transaction
    ///   (the reference throws in `getBEEFForNode` and discards). The stored
    ///   BEEF of that output is merged into the checked copy, so the spend is
    ///   executed against its real source and that source's own ancestry is
    ///   verified too. If the stored BEEF is itself incomplete the verifier
    ///   refuses, and the graph is discarded as in the reference. Only the
    ///   CHECKED copy is merged; the finalized bytes are untouched.
    /// - *Every node is an input of its parent* (`read_anchor_input`).
    /// - *A conflicting spend refuses the graph.* Two transactions of one
    ///   graph that spend the same outpoint cannot both be admitted by the
    ///   finalize submits (the first one consumes the coin), so a replay that
    ///   showed the coin to both would promise an admission finalize cannot
    ///   keep. Upstream added the same rule after f999e0c1a (`spentOutpoints`,
    ///   "Historical GASP graph contains a conflicting spend").
    /// - *A fault of the moment is not a verdict.* A chain tracker that could
    ///   not answer, or a storage read that failed, says nothing about the
    ///   graph: the error is `GASPError::AnchorUnavailable`, the graph is
    ///   still discarded, but `complete_graph` FAILS the UTXO so the cursor
    ///   gap guard asks for it again. The reference discards and moves on,
    ///   which here would lose the UTXO for good behind an advanced cursor.
    /// - *A manager error in the replay is "not now", not a refusal.* The
    ///   replay is where a manager is shown the coins, so it is where one
    ///   whose own state lags can say it cannot place a transaction YET. An
    ///   `Err` from it is `GASPError::AnchorUnavailable` too: the UTXO fails
    ///   and is asked for again. `Ok` with nothing admitted stays the final
    ///   refusal, and there the cursor moves, as in the reference (which
    ///   moves it past a refused and a failed graph alike, `GASP.ts` 389 to
    ///   392, 419 to 423, 559 to 572, `Engine.ts` 1600 to 1604). The cost: a
    ///   manager that errors FOREVER on one UTXO holds that peer's cursor
    ///   below it, and what lies above and is not yet held is walked again
    ///   on every tick.
    ///
    /// **Cost.** Step 1 runs the script of every input of every unproven
    /// transaction in the root's BEEF and asks the tracker once per merkle
    /// path in it (a proven root: one question, no script). The ancestors a
    /// manager named BEHIND a proven node are not in that BEEF: their own
    /// merkle paths are not checked, here or in the reference; they are bound
    /// to the verified root by txid instead. Step 2 is one manager call per
    /// transaction and one storage read per input that is not already a coin
    /// of the set, which is what the finalize submits cost again.
    ///
    /// Since bsv-low #592 the walk runs under the engine's walk budget
    /// (`DoorBudget`, the door's two limbs, estimated from the bytes before
    /// anything runs). A breach is no verdict on the graph, and nothing
    /// stands behind this check (no network accepted these bytes, and the
    /// finalize submits them `historical-tx-no-spv`), so a breach is never
    /// the replay's to judge (the E592 lens fold, H1: as built at `4060ffd`
    /// it went on, and a peer's root that spent a held coin with no
    /// signature was finalized). It is HELD: `GASPError::AnchorWalkHeld`,
    /// the graph not finalized, the UTXO failed (or, under a per-graph
    /// budget, its record kept with nothing pending), the cursor below it,
    /// counted (`GASPSync::anchor_walk_held`). The roots the walk reached are
    /// checked first, so a bad proof is still the refusal it is. The hold
    /// ends when the peer serves the root PROVEN: #555's resume re-asks an
    /// unproven root, a proven one restarts the walk (`root_proven`), and a
    /// proven root is trusted without a script run. So an honest 0-conf graph
    /// too big to walk lands after its block; a forgery never lands. What a
    /// forging peer costs: its own slot (the UTXO held below the cursor, its
    /// record re-asked each pass up to `DEFERRED_GRAPH_MAX_PASSES`), inside
    /// its per-peer budget, and, as it never yields a finalized graph or a
    /// cursor move on that UTXO, D16's yieldless bound quarantines it if it
    /// serves nothing else, and ONLY with a per-graph budget
    /// (`Engine::set_graph_budget`): the yieldless bound exists only there
    /// (the E592 delta lens, D1-L2). With NO per-graph budget (a library
    /// consumer that sets none) there is no record and no yieldless count:
    /// the held UTXO fails on every sync, the sync runs to its end and is a
    /// success under #302, the cursor stays below that UTXO for good, and
    /// the graph is fetched and its walk estimated again on every tick;
    /// nothing is admitted, and the cost is ours, the residual D15 states for
    /// `AnchorUnavailable`. The reference serialises this method behind
    /// `acquireAnchorValidationSlot` (at most 4 at once); here graphs are
    /// completed one at a time by one sync on one thread, so there is
    /// nothing to serialise.
    async fn validate_graph_anchor(&self, graph_id: &str) -> Result<(), GASPError> {
        let anchor = {
            let refs = self
                .pending_graphs
                .lock()
                .map_err(|e| GASPError::Other(format!("Lock poisoned: {e}")))?;
            Self::read_anchor_input(graph_id, &refs, self.strict_beef)?
        };
        let unavailable =
            |what: String| GASPError::AnchorUnavailable(format!("graph {graph_id}: {what}"));
        let refused = |why: String| GASPError::ValidationFailed(format!("graph {graph_id}: {why}"));

        // ── 1. Bitcoin ──────────────────────────────────────────────────
        let mut root_beef = anchor.root_beef;
        if !anchor.absent_sources.is_empty() {
            let mut merged = beef_limits::parse_beef(&root_beef, &beef_limits::PEER_BEEF_LIMITS)
                .map_err(|e| refused(format!("root BEEF does not parse: {e}")))?;
            for (txid, output_index) in &anchor.absent_sources {
                let held = self
                    .storage
                    .find_output(txid, *output_index, Some(&self.topic), None, true)
                    .await
                    .map_err(|e| unavailable(format!("reading {txid}.{output_index}: {e}")))?;
                if let Some(stored) = held.and_then(|output| output.beef) {
                    if let Ok(stored) =
                        beef_limits::parse_beef(&stored, &beef_limits::STORED_BEEF_LIMITS)
                    {
                        merged.merge_beef(&stored);
                    }
                }
            }
            root_beef = merged.to_binary();
        }
        let tracker = self.chain_tracker.map(|inner| FaultNotingTracker {
            inner,
            faulted: AtomicBool::new(false),
        });
        let verdict = crate::engine::verify_spv_like_the_reference(
            tracker.as_ref().map(|t| t as &dyn ChainTracker),
            self.verify_scripts,
            &root_beef,
            &anchor.root_txid,
            self.walk_budget,
        )
        .await;
        // The checked copy has done its work (bsv-low #586: not held through
        // the replay).
        drop(root_beef);
        // A walk that could not run within the budget (bsv-low #592) is not
        // a verdict on the graph, and nothing else judges it: HELD until the
        // root is served proven (the E592 lens fold, H1; the doc above).
        let verdict = match verdict {
            Ok(Some(stop)) => {
                warn!(
                    "GASP anchor: the script walk of {} could not run ({}: at {}; {}): held, the graph is not finalized until its root is served proven (bsv-low #592, E592 fold)",
                    stop.subject_txid,
                    stop.limb.as_str(),
                    stop.at_txid,
                    stop.what
                );
                return Err(GASPError::AnchorWalkHeld(format!(
                    "graph {graph_id}: the script walk of {} could not run within the budget ({}: {})",
                    stop.subject_txid,
                    stop.limb.as_str(),
                    stop.what
                )));
            }
            Ok(None) => Ok(()),
            Err(e) => Err(e),
        };
        if let Err(e) = verdict {
            if tracker
                .as_ref()
                .is_some_and(|t| t.faulted.load(Ordering::Relaxed))
            {
                return Err(unavailable(format!("the chain tracker faulted: {e}")));
            }
            return Err(refused(format!(
                "The graph is not well-anchored according to the rules of Bitcoin: {e}"
            )));
        }

        // ── 2. Overlay ──────────────────────────────────────────────────
        let Some(manager) = self.topic_manager else {
            self.keep_anchored(graph_id, anchor.order, anchor.assembly);
            return Ok(());
        };
        let mut coins: HashSet<String> = HashSet::new();
        let mut spent: HashSet<String> = HashSet::new();
        let mut replayed: HashSet<String> = HashSet::new();
        for key in &anchor.order {
            // One node's BEEF at a time, the bytes `finalize_graph` will
            // assemble, dropped when its turn is over (bsv-low #586).
            let beef = &anchor.assembly.beef_of(key);
            // The parse `Engine::submit` makes of the same bytes at finalize.
            let subject = beef_limits::parse_beef(beef, &beef_limits::PEER_BEEF_LIMITS)
                .map(|mut b| crate::subject::subject_txid_of(&mut b))
                .map_err(|e| refused(format!("a graph BEEF does not parse: {e}")))?;
            let tx = beef_limits::transaction_from_beef(
                beef,
                subject.as_deref(),
                &beef_limits::PEER_BEEF_LIMITS,
            )
            .map_err(|e| refused(format!("a graph BEEF does not parse: {e}")))?;
            let txid = tx.id();
            // Two outputs of one transaction are two nodes with one BEEF.
            if !replayed.insert(txid.clone()) {
                continue;
            }
            let mut previous_coins: Vec<u8> = Vec::new();
            for (input_index, input) in tx.inputs.iter().enumerate() {
                let source_txid = input.get_source_txid().unwrap_or_default();
                if source_txid.is_empty() {
                    continue;
                }
                let outpoint = format!("{}.{}", source_txid, input.source_output_index);
                if !spent.insert(outpoint.clone()) {
                    return Err(refused(format!(
                        "the graph contains a conflicting spend of {outpoint}"
                    )));
                }
                let is_coin = coins.contains(&outpoint)
                    || self
                        .storage
                        .find_output(
                            &source_txid,
                            input.source_output_index,
                            Some(&self.topic),
                            None,
                            false,
                        )
                        .await
                        .map_err(|e| unavailable(format!("reading {outpoint}: {e}")))?
                        .is_some();
                if is_coin {
                    previous_coins.extend_from_slice(&(input_index as u32).to_le_bytes());
                }
            }
            // TS passes { dryRun: true }: the replay only asks, the finalize
            // submit is the admission.
            // An `Err` is the manager's "not now" (a fault of the moment, see
            // the divergences above); `Ok` with nothing is its refusal.
            let admittance = manager
                .identify_admissible_outputs(
                    &tx,
                    &previous_coins,
                    None,
                    SubmitMode::HistoricalTx,
                    &TopicAdmittanceContext::DRY_RUN,
                )
                .await
                .map_err(|e| unavailable(format!("the topic manager failed on {txid}: {e}")))?;
            for output_index in admittance.outputs_to_admit {
                coins.insert(format!("{txid}.{output_index}"));
            }
        }
        if !coins.contains(graph_id) {
            return Err(refused(
                "This graph did not result in topical admittance of the root node. Rejecting."
                    .to_string(),
            ));
        }
        self.keep_anchored(graph_id, anchor.order, anchor.assembly);
        Ok(())
    }

    /// Finalize a graph — convert accumulated nodes to ordered BEEFs and push
    /// them to the shared sink for later Engine submission.
    ///
    /// Computes ordered BEEF byte arrays (ancestors first, root last) from the
    /// temporary graph nodes and stores them in the `FinalizedGraphSink`. The
    /// Engine drains it (after the sync, or per completed UTXO under a
    /// per-peer budget) and submits each one with `HistoricalTxNoSpv` mode, which skips SPV because
    /// `validate_graph_anchor` has already done it (the reference's
    /// `finalizeGraph`, lines 317 to 333).
    async fn finalize_graph(&self, graph_id: &str) -> Result<(), GASPError> {
        let refs = self
            .pending_graphs
            .lock()
            .map_err(|e| GASPError::Other(format!("Lock poisoned: {e}")))?;

        if !refs.contains_key(graph_id) {
            return Err(GASPError::Other(format!("No pending graph for {graph_id}")));
        }

        let node_count = refs
            .values()
            .filter(|pn| pn.node.graph_id == graph_id)
            .count();

        debug!("Finalizing graph {graph_id} ({node_count} nodes). Computing ordered BEEFs.");

        // Compute ordered BEEFs for the graph.
        let beefs =
            Self::compute_ordered_beefs(graph_id, &refs, self.strict_beef, self.take_anchored())?;

        // Push to shared sink for later Engine submission.
        drop(refs);
        self.finalized_sink
            .lock()
            .map_err(|e| GASPError::Other(format!("Lock poisoned: {e}")))?
            .push(FinalizedGraph {
                topic: self.topic.clone(),
                beefs,
            });

        // Remove all nodes belonging to this graph from pending refs.
        let mut refs = self
            .pending_graphs
            .lock()
            .map_err(|e| GASPError::Other(format!("Lock poisoned: {e}")))?;
        let keys_to_remove: Vec<String> = refs
            .iter()
            .filter(|(_, pn)| pn.node.graph_id == graph_id)
            .map(|(k, _)| k.clone())
            .collect();
        for key in keys_to_remove {
            refs.remove(&key);
        }

        Ok(())
    }

    /// Discard a temporary graph that failed validation.
    async fn discard_graph(&self, graph_id: &str) -> Result<(), GASPError> {
        debug!("Discarding graph: {graph_id}");
        self.take_anchored();
        let mut refs = self
            .pending_graphs
            .lock()
            .map_err(|e| GASPError::Other(format!("Lock poisoned: {e}")))?;
        let keys_to_remove: Vec<String> = refs
            .iter()
            .filter(|(_, pn)| pn.node.graph_id == graph_id)
            .map(|(k, _)| k.clone())
            .collect();
        for key in keys_to_remove {
            refs.remove(&key);
        }
        Ok(())
    }

    async fn load_deferred_graphs(&self) -> Result<Vec<crate::gasp::DeferredGraphKey>, GASPError> {
        self.storage
            .find_deferred_graphs(&self.peer, &self.topic)
            .await
            .map_err(|e| GASPError::StorageError(e.to_string()))
    }

    async fn get_deferred_graph(
        &self,
        outpoint: &str,
    ) -> Result<Option<crate::gasp::DeferredGraph>, GASPError> {
        self.storage
            .get_deferred_graph(&self.peer, &self.topic, outpoint)
            .await
            .map_err(|e| GASPError::StorageError(e.to_string()))
    }

    async fn save_deferred_graph(
        &self,
        record: &crate::gasp::DeferredGraph,
    ) -> Result<crate::gasp::DeferredGraphSave, GASPError> {
        let mut record = record.clone();
        record.peer.clone_from(&self.peer);
        record.topic.clone_from(&self.topic);
        record.configured = self.configured_peer;
        self.storage
            .put_deferred_graph(&record)
            .await
            .map_err(|e| GASPError::StorageError(e.to_string()))
    }

    async fn delete_deferred_graph(&self, outpoint: &str) -> Result<(), GASPError> {
        self.storage
            .delete_deferred_graph(&self.peer, &self.topic, outpoint)
            .await
            .map_err(|e| GASPError::StorageError(e.to_string()))
    }
}

// ============================================================================
// Tests
// ============================================================================

#[cfg(test)]
mod tests {
    use super::*;
    use crate::gasp::GASPStorage;
    use crate::storage::memory::MemoryStorage;
    use crate::storage::Storage;
    use crate::types::Output;

    fn make_sink() -> FinalizedGraphSink {
        new_finalized_graph_sink()
    }

    fn make_output(txid: &str, index: u32, topic: &str, score: f64) -> Output {
        Output {
            txid: txid.to_string(),
            output_index: index,
            output_script: vec![0x76, 0xa9],
            satoshis: 1000,
            topic: topic.to_string(),
            spent: false,
            outputs_consumed: vec![],
            consumed_by: vec![],
            beef: Some(vec![0xBE, 0xEF]),
            block_height: None,
            score: Some(score),
        }
    }

    /// Build a minimal valid transaction hex with one input and one output.
    fn make_valid_tx_hex(source_txid: &str, source_oi: u32) -> String {
        let mut tx = Transaction::new();
        let input = bsv_rs::transaction::TransactionInput::new(source_txid.to_string(), source_oi);
        tx.inputs.push(input);
        tx.outputs.push(bsv_rs::transaction::TransactionOutput::new(
            100,
            bsv_rs::script::LockingScript::from_hex(
                "76a914000000000000000000000000000000000000000088ac",
            )
            .unwrap(),
        ));
        tx.to_hex()
    }

    /// Build a coinbase-like tx (no meaningful inputs).
    /// A raw transaction with no input, read by `Transaction::from_hex`
    /// alone (no BEEF door, which refuses it since bsv-rs 0.4.1).
    fn make_no_input_tx_hex() -> String {
        let mut tx = Transaction::new();
        tx.outputs.push(bsv_rs::transaction::TransactionOutput::new(
            5000,
            bsv_rs::script::LockingScript::from_hex(
                "76a914000000000000000000000000000000000000000088ac",
            )
            .unwrap(),
        ));
        tx.to_hex()
    }

    fn make_coinbase_tx_hex() -> String {
        let mut tx = Transaction::new();
        // A coinbase's null outpoint: a transaction with no input is refused
        // since bsv-rs 0.4.1 (NL-8 W6).
        let mut input = bsv_rs::transaction::TransactionInput::new("00".repeat(32), u32::MAX);
        input.unlocking_script =
            Some(bsv_rs::script::UnlockingScript::from_binary(&[0x51]).unwrap());
        tx.inputs.push(input);
        tx.outputs.push(bsv_rs::transaction::TransactionOutput::new(
            5000,
            bsv_rs::script::LockingScript::from_hex(
                "76a914000000000000000000000000000000000000000088ac",
            )
            .unwrap(),
        ));
        tx.to_hex()
    }

    // ── find_known_utxos tests ────────────────────────────────────────

    #[tokio::test]
    async fn find_known_utxos_returns_correct_outputs() {
        let store = MemoryStorage::new();
        store
            .insert_output(&make_output("tx1", 0, "tm_test", 100.0))
            .await
            .unwrap();
        store
            .insert_output(&make_output("tx2", 0, "tm_test", 200.0))
            .await
            .unwrap();
        store
            .insert_output(&make_output("tx3", 0, "tm_other", 300.0))
            .await
            .unwrap();

        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());

        let utxos = gasp_storage.find_known_utxos(0, None).await.unwrap();
        assert_eq!(utxos.len(), 2, "Should return only tm_test UTXOs");
        assert_eq!(utxos[0].txid, "tx1");
        assert_eq!(utxos[1].txid, "tx2");
    }

    #[tokio::test]
    async fn find_known_utxos_respects_since() {
        let store = MemoryStorage::new();
        store
            .insert_output(&make_output("tx1", 0, "tm_test", 100.0))
            .await
            .unwrap();
        store
            .insert_output(&make_output("tx2", 0, "tm_test", 200.0))
            .await
            .unwrap();
        store
            .insert_output(&make_output("tx3", 0, "tm_test", 300.0))
            .await
            .unwrap();

        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());

        let utxos = gasp_storage.find_known_utxos(200, None).await.unwrap();
        assert_eq!(utxos.len(), 2, "Should return UTXOs with score >= 200");
        assert_eq!(utxos[0].txid, "tx2");
        assert_eq!(utxos[1].txid, "tx3");
    }

    #[tokio::test]
    async fn find_known_utxos_respects_limit() {
        let store = MemoryStorage::new();
        store
            .insert_output(&make_output("tx1", 0, "tm_test", 100.0))
            .await
            .unwrap();
        store
            .insert_output(&make_output("tx2", 0, "tm_test", 200.0))
            .await
            .unwrap();
        store
            .insert_output(&make_output("tx3", 0, "tm_test", 300.0))
            .await
            .unwrap();

        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());

        let utxos = gasp_storage.find_known_utxos(0, Some(2)).await.unwrap();
        assert_eq!(utxos.len(), 2, "Should limit to 2 results");
    }

    #[tokio::test]
    async fn find_known_utxos_excludes_spent() {
        let store = MemoryStorage::new();
        store
            .insert_output(&make_output("tx1", 0, "tm_test", 100.0))
            .await
            .unwrap();
        store
            .insert_output(&make_output("tx2", 0, "tm_test", 200.0))
            .await
            .unwrap();
        store.mark_utxo_as_spent("tx1", 0, "tm_test").await.unwrap();

        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());

        let utxos = gasp_storage.find_known_utxos(0, None).await.unwrap();
        assert_eq!(utxos.len(), 1, "Should exclude spent UTXOs");
        assert_eq!(utxos[0].txid, "tx2");
    }

    #[tokio::test]
    async fn find_known_utxos_empty_topic() {
        let store = MemoryStorage::new();
        let gasp_storage = OverlayGASPStorage::new(&store, "tm_empty", make_sink());

        let utxos = gasp_storage.find_known_utxos(0, None).await.unwrap();
        assert!(utxos.is_empty());
    }

    // ── find_needed_inputs tests ──────────────────────────────────────

    #[tokio::test]
    async fn find_needed_inputs_returns_none_for_unparsable_tx() {
        let store = MemoryStorage::new();
        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());

        let node = GASPNode {
            graph_id: "abc.0".to_string(),
            raw_tx: "deadbeef".to_string(), // not valid tx hex
            output_index: 0,
            proof: None,
            tx_metadata: None,
            output_metadata: None,
            inputs: None,
        };

        let result = gasp_storage.find_needed_inputs(&node).await.unwrap();
        assert!(result.is_none(), "Unparsable tx: graceful fallback to None");
    }

    #[tokio::test]
    async fn find_needed_inputs_returns_none_for_proved_tx() {
        let store = MemoryStorage::new();
        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());

        // A node with a proof (merkle path) should not need inputs.
        let raw_tx = make_valid_tx_hex(
            "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
            0,
        );
        let node = GASPNode {
            graph_id: "abc.0".to_string(),
            raw_tx,
            output_index: 0,
            proof: Some("some_proof_hex".to_string()),
            tx_metadata: None,
            output_metadata: None,
            inputs: None,
        };

        let result = gasp_storage.find_needed_inputs(&node).await.unwrap();
        assert!(result.is_none(), "Proved tx should not need inputs");
    }

    #[tokio::test]
    async fn find_needed_inputs_requests_inputs_for_unproved_tx() {
        let store = MemoryStorage::new();
        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());

        let source_txid = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
        let raw_tx = make_valid_tx_hex(source_txid, 0);

        let node = GASPNode {
            graph_id: "abc.0".to_string(),
            raw_tx,
            output_index: 0,
            proof: None,
            tx_metadata: None,
            output_metadata: None,
            inputs: None,
        };

        let result = gasp_storage.find_needed_inputs(&node).await.unwrap();
        assert!(result.is_some(), "Unproved tx should request inputs");
        let response = result.unwrap();
        assert_eq!(response.requested_inputs.len(), 1);
        let key = format!("{source_txid}.0");
        assert!(
            response.requested_inputs.contains_key(&key),
            "Should request the source input"
        );
    }

    #[tokio::test]
    async fn find_needed_inputs_strips_known_inputs() {
        let store = MemoryStorage::new();
        let source_txid = "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc";

        // Insert the source output so it's already known
        store
            .insert_output(&make_output(source_txid, 0, "tm_test", 50.0))
            .await
            .unwrap();

        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());

        let raw_tx = make_valid_tx_hex(source_txid, 0);
        let node = GASPNode {
            graph_id: "abc.0".to_string(),
            raw_tx,
            output_index: 0,
            proof: None,
            tx_metadata: None,
            output_metadata: None,
            inputs: None,
        };

        let result = gasp_storage.find_needed_inputs(&node).await.unwrap();
        assert!(
            result.is_none(),
            "Should strip already-known inputs, resulting in None"
        );
    }

    #[tokio::test]
    async fn find_needed_inputs_returns_none_for_coinbase() {
        let store = MemoryStorage::new();
        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());

        let raw_tx = make_no_input_tx_hex();
        let node = GASPNode {
            graph_id: "abc.0".to_string(),
            raw_tx,
            output_index: 0,
            proof: None,
            tx_metadata: None,
            output_metadata: None,
            inputs: None,
        };

        let result = gasp_storage.find_needed_inputs(&node).await.unwrap();
        assert!(
            result.is_none(),
            "Coinbase-like tx (no inputs) should need nothing"
        );
    }

    // ── append / finalize / discard tests ─────────────────────────────

    #[tokio::test]
    async fn append_and_finalize_graph() {
        let store = MemoryStorage::new();
        let sink = make_sink();
        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", sink.clone());

        let raw_tx = make_coinbase_tx_hex();
        let node = GASPNode {
            graph_id: "abc.0".to_string(),
            raw_tx,
            output_index: 0,
            proof: None,
            tx_metadata: None,
            output_metadata: None,
            inputs: None,
        };

        gasp_storage.append_to_graph(&node, None).await.unwrap();
        assert_eq!(gasp_storage.pending_graph_count(), 1);

        gasp_storage.finalize_graph("abc.0").await.unwrap();
        assert_eq!(gasp_storage.pending_graph_count(), 0);

        // Check that finalized data was produced
        let finalized = sink.lock().unwrap();
        assert_eq!(finalized.len(), 1, "Should have one finalized graph");
        assert_eq!(finalized[0].topic, "tm_test");
        assert!(!finalized[0].beefs.is_empty(), "Should have BEEF data");
    }

    #[tokio::test]
    async fn finalize_graph_produces_submittable_beef() {
        let store = MemoryStorage::new();
        let sink = make_sink();
        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", sink.clone());

        let raw_tx = make_coinbase_tx_hex();
        let node = GASPNode {
            graph_id: "root.0".to_string(),
            raw_tx,
            output_index: 0,
            proof: None,
            tx_metadata: None,
            output_metadata: None,
            inputs: None,
        };

        gasp_storage.append_to_graph(&node, None).await.unwrap();
        gasp_storage.finalize_graph("root.0").await.unwrap();

        let finalized = sink.lock().unwrap();
        assert_eq!(finalized.len(), 1);

        // Each BEEF should be parseable
        for beef_bytes in &finalized[0].beefs {
            let tx = Transaction::from_beef(beef_bytes, None);
            assert!(
                tx.is_ok(),
                "Finalized BEEF should be parseable: {:?}",
                tx.err()
            );
        }
    }

    #[tokio::test]
    async fn append_multiple_nodes_same_graph() {
        let store = MemoryStorage::new();
        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());

        let raw_tx1 = make_coinbase_tx_hex();
        let node1 = GASPNode {
            graph_id: "abc.0".to_string(),
            raw_tx: raw_tx1,
            output_index: 0,
            proof: None,
            tx_metadata: None,
            output_metadata: None,
            inputs: None,
        };
        let raw_tx2 = make_coinbase_tx_hex();
        let node2 = GASPNode {
            graph_id: "abc.0".to_string(),
            raw_tx: raw_tx2,
            output_index: 0,
            proof: None,
            tx_metadata: None,
            output_metadata: None,
            inputs: None,
        };

        gasp_storage.append_to_graph(&node1, None).await.unwrap();
        gasp_storage
            .append_to_graph(&node2, Some("abc.0"))
            .await
            .unwrap();
        assert_eq!(gasp_storage.pending_graph_count(), 1);
    }

    #[tokio::test]
    async fn discard_graph_removes_pending() {
        let store = MemoryStorage::new();
        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());

        let raw_tx = make_coinbase_tx_hex();
        let node = GASPNode {
            graph_id: "abc.0".to_string(),
            raw_tx,
            output_index: 0,
            proof: None,
            tx_metadata: None,
            output_metadata: None,
            inputs: None,
        };

        gasp_storage.append_to_graph(&node, None).await.unwrap();
        assert_eq!(gasp_storage.pending_graph_count(), 1);

        gasp_storage.discard_graph("abc.0").await.unwrap();
        assert_eq!(gasp_storage.pending_graph_count(), 0);
    }

    #[tokio::test]
    async fn validate_graph_anchor_refuses_an_unknown_graph() {
        let store = MemoryStorage::new();
        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());

        // The reference throws `Graph node with ID ... not found`.
        let error = gasp_storage
            .validate_graph_anchor("nonexistent.0")
            .await
            .unwrap_err();
        assert!(matches!(error, GASPError::ValidationFailed(_)), "{error}");
    }

    #[tokio::test]
    async fn finalize_empty_graph_errors() {
        let store = MemoryStorage::new();
        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());

        let result = gasp_storage.finalize_graph("nonexistent.0").await;
        assert!(result.is_err(), "Should error on unknown graph");
    }

    #[tokio::test]
    async fn gasp_storage_is_object_safe() {
        let store = MemoryStorage::new();
        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());

        // Verify it can be used as dyn GASPStorage
        let boxed: Box<dyn GASPStorage> = Box::new(gasp_storage);
        let utxos = boxed.find_known_utxos(0, None).await.unwrap();
        assert!(utxos.is_empty());
    }

    /// Integration test: OverlayGASPStorage works with GASPSync.
    #[tokio::test]
    async fn overlay_gasp_storage_with_gasp_sync() {
        use crate::gasp::{GASPRemote, GASPSync};
        use crate::types::{GASPInitialReply, GASPInitialRequest, GASPInitialResponse};

        // Set up local storage with one UTXO
        let store = MemoryStorage::new();
        store
            .insert_output(&make_output("local_tx", 0, "tm_test", 50.0))
            .await
            .unwrap();

        let sink = make_sink();
        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", sink.clone());

        // Mock remote with two UTXOs — provides unparsable raw_tx so
        // find_needed_inputs returns None (graceful fallback).
        struct TestRemote;

        #[async_trait(?Send)]
        impl GASPRemote for TestRemote {
            async fn get_initial_response(
                &self,
                request: &GASPInitialRequest,
            ) -> Result<GASPInitialResponse, GASPError> {
                Ok(GASPInitialResponse {
                    utxo_list: vec![
                        GASPOutput {
                            txid: "remote_tx1".to_string(),
                            output_index: 0,
                            score: 100.0,
                        },
                        GASPOutput {
                            txid: "remote_tx2".to_string(),
                            output_index: 0,
                            score: 200.0,
                        },
                    ],
                    since: request.since,
                })
            }
            async fn get_initial_reply(
                &self,
                _: &GASPInitialResponse,
            ) -> Result<GASPInitialReply, GASPError> {
                Ok(GASPInitialReply {
                    utxo_list: Vec::new(),
                })
            }
            async fn request_node(
                &self,
                graph_id: &str,
                txid: &str,
                output_index: u32,
                _: bool,
            ) -> Result<GASPNode, GASPError> {
                Ok(GASPNode {
                    graph_id: graph_id.to_string(),
                    raw_tx: format!("rawtx_{txid}"),
                    output_index,
                    proof: None,
                    tx_metadata: None,
                    output_metadata: None,
                    inputs: None,
                })
            }
            async fn submit_node(
                &self,
                _: &GASPNode,
            ) -> Result<Option<GASPNodeResponse>, GASPError> {
                Ok(None)
            }
        }

        let mut sync = GASPSync::new(
            Box::new(gasp_storage),
            Box::new(TestRemote),
            0,
            "[TEST]",
            true, // unidirectional
        );

        sync.sync(None).await.unwrap();
        // The synthetic graph cannot assemble a BEEF. Nothing completed,
        // so the cursor remains at entry and both outputs are retried.
        assert_eq!(sync.last_interaction, 0);
    }

    /// Integration test: full sync with valid transactions produces submittable BEEFs.
    #[tokio::test]
    async fn gasp_sync_with_valid_tx_produces_finalized_beefs() {
        use crate::gasp::{GASPRemote, GASPSync};
        use crate::types::{GASPInitialReply, GASPInitialRequest, GASPInitialResponse};

        let store = MemoryStorage::new();
        let sink = make_sink();
        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", sink.clone());

        // Build a valid coinbase-style tx for the remote to provide.
        let valid_tx_hex = make_coinbase_tx_hex();

        struct ValidTxRemote {
            tx_hex: String,
        }

        #[async_trait(?Send)]
        impl GASPRemote for ValidTxRemote {
            async fn get_initial_response(
                &self,
                request: &GASPInitialRequest,
            ) -> Result<GASPInitialResponse, GASPError> {
                Ok(GASPInitialResponse {
                    utxo_list: vec![GASPOutput {
                        txid: "valid_tx".to_string(),
                        output_index: 0,
                        score: 100.0,
                    }],
                    since: request.since,
                })
            }
            async fn get_initial_reply(
                &self,
                _: &GASPInitialResponse,
            ) -> Result<GASPInitialReply, GASPError> {
                Ok(GASPInitialReply {
                    utxo_list: Vec::new(),
                })
            }
            async fn request_node(
                &self,
                graph_id: &str,
                _txid: &str,
                output_index: u32,
                _: bool,
            ) -> Result<GASPNode, GASPError> {
                // Served PROVEN: since bsv-low #551 the anchor check refuses
                // an unproven transaction that creates satoshis (this one has
                // no inputs). With no chain tracker a merkle path is accepted
                // unchecked, the reference's 'scripts only'.
                let txid = Transaction::from_hex(&self.tx_hex).unwrap().id();
                let proof = MerklePath::new(
                    100,
                    vec![vec![bsv_rs::transaction::MerklePathLeaf::new_txid(0, txid)]],
                )
                .unwrap();
                Ok(GASPNode {
                    graph_id: graph_id.to_string(),
                    raw_tx: self.tx_hex.clone(),
                    output_index,
                    proof: Some(proof.to_hex()),
                    tx_metadata: None,
                    output_metadata: None,
                    inputs: None,
                })
            }
            async fn submit_node(
                &self,
                _: &GASPNode,
            ) -> Result<Option<GASPNodeResponse>, GASPError> {
                Ok(None)
            }
        }

        let remote = ValidTxRemote {
            tx_hex: valid_tx_hex,
        };

        let mut sync = GASPSync::new(Box::new(gasp_storage), Box::new(remote), 0, "[TEST]", true);

        sync.sync(None).await.unwrap();

        let finalized = sink.lock().unwrap();
        assert_eq!(finalized.len(), 1, "Should have one finalized graph");
        assert!(!finalized[0].beefs.is_empty(), "Should have BEEF bytes");

        // Verify the BEEF is parseable
        for beef_bytes in &finalized[0].beefs {
            let parsed = Transaction::from_beef(beef_bytes, None);
            assert!(parsed.is_ok(), "BEEF should be parseable");
        }
    }

    #[tokio::test]
    async fn take_finalized_graphs_drains_list() {
        let store = MemoryStorage::new();
        let sink = make_sink();
        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", sink.clone());

        let raw_tx = make_coinbase_tx_hex();
        let node = GASPNode {
            graph_id: "abc.0".to_string(),
            raw_tx,
            output_index: 0,
            proof: None,
            tx_metadata: None,
            output_metadata: None,
            inputs: None,
        };

        gasp_storage.append_to_graph(&node, None).await.unwrap();
        gasp_storage.finalize_graph("abc.0").await.unwrap();

        let taken = gasp_storage.take_finalized_graphs();
        assert_eq!(taken.len(), 1);

        // Should be empty after draining
        let taken2 = gasp_storage.take_finalized_graphs();
        assert!(taken2.is_empty());
    }

    // ── bsv-low #586: the assembly gives the bytes the hydrated tree gave ──

    /// `get_beef_for_node` as it was on `fbb7fa8`, verbatim: the hydrating
    /// recursion [`GraphAssembly`] replaced. Kept as the reference the
    /// assembly is compared with, byte for byte and error for error.
    fn legacy_get_beef_for_node(
        node_key: &str,
        refs: &HashMap<String, PendingNode>,
        strict_beef: bool,
    ) -> Result<(Transaction, Vec<u8>), GASPError> {
        let pending = refs
            .get(node_key)
            .ok_or_else(|| GASPError::Other(format!("Node {node_key} not found in graph refs")))?;

        let mut tx = Transaction::from_hex(&pending.node.raw_tx)
            .map_err(|e| GASPError::Other(format!("Failed to parse raw_tx for {node_key}: {e}")))?;

        if let Some(ref proof_hex) = pending.node.proof {
            tx.merkle_path = Some(beef_limits::proof_from_hex(proof_hex).map_err(|e| {
                GASPError::Other(format!("Failed to parse proof for {node_key}: {e}"))
            })?);
        } else {
            let child_info: Vec<(usize, String)> = tx
                .inputs
                .iter()
                .enumerate()
                .filter_map(|(idx, input)| {
                    let source_txid = input.get_source_txid().unwrap_or_default();
                    if source_txid.is_empty() {
                        return None;
                    }
                    let child_key = format!("{}.{}", source_txid, input.source_output_index);
                    if refs.contains_key(&child_key) {
                        Some((idx, child_key))
                    } else {
                        None
                    }
                })
                .collect();

            for (input_idx, child_key) in child_info {
                let (child_tx, _) = legacy_get_beef_for_node(&child_key, refs, strict_beef)?;
                tx.inputs[input_idx].source_transaction = Some(Box::new(child_tx));
            }
        }

        let allow_partial = !strict_beef;
        let beef = tx.to_beef(allow_partial).map_err(|e| {
            GASPError::Other(format!("Failed to serialize BEEF for {node_key}: {e}"))
        })?;

        Ok((tx, beef))
    }

    /// `compute_ordered_beefs` as it was on `fbb7fa8`, verbatim.
    fn legacy_compute_ordered_beefs(
        graph_id: &str,
        refs: &HashMap<String, PendingNode>,
        strict_beef: bool,
    ) -> Result<Vec<Vec<u8>>, GASPError> {
        let mut beefs: Vec<Vec<u8>> = Vec::new();
        let mut visited: std::collections::HashSet<String> = std::collections::HashSet::new();

        fn hydrate(
            node_key: &str,
            refs: &HashMap<String, PendingNode>,
            beefs: &mut Vec<Vec<u8>>,
            visited: &mut std::collections::HashSet<String>,
            strict_beef: bool,
        ) -> Result<(), GASPError> {
            if visited.contains(node_key) {
                return Ok(());
            }
            visited.insert(node_key.to_string());

            let Some(pending) = refs.get(node_key) else {
                return Ok(());
            };

            for child_key in &pending.children {
                hydrate(child_key, refs, beefs, visited, strict_beef)?;
            }

            let (_, beef) = legacy_get_beef_for_node(node_key, refs, strict_beef)?;
            beefs.push(beef);
            Ok(())
        }

        hydrate(graph_id, refs, &mut beefs, &mut visited, strict_beef)?;
        Ok(beefs)
    }

    /// A small deterministic generator (no dependency, the same graphs on
    /// every run).
    struct Lcg(u64);

    impl Lcg {
        fn below(&mut self, n: u64) -> u64 {
            self.0 = self
                .0
                .wrapping_mul(6_364_136_223_846_793_005)
                .wrapping_add(1_442_695_040_888_963_407);
            (self.0 >> 33) % n
        }
    }

    fn pending(raw_tx: String, output_index: u32, proof: Option<String>) -> PendingNode {
        PendingNode {
            node: GASPNode {
                graph_id: "g".to_string(),
                raw_tx,
                output_index,
                proof,
                tx_metadata: None,
                output_metadata: None,
                inputs: None,
            },
            spent_by: None,
            children: Vec::new(),
        }
    }

    /// A random pending graph: up to 9 transactions, each spending up to
    /// four outputs of earlier ones (two outputs of one transaction too, the
    /// diamond), some proven (two of a pair sharing one block, so their
    /// proofs COMBINE), some referenced outpoints absent from the graph (a
    /// partial BEEF, a strict refusal), and now and then a proof or a raw
    /// transaction that does not parse. Returns the graph and its root key.
    fn random_graph(rng: &mut Lcg) -> (HashMap<String, PendingNode>, String) {
        use bsv_rs::transaction::{MerklePathLeaf, TransactionInput, TransactionOutput};
        let count = 2 + rng.below(8) as usize;
        let mut txs: Vec<Transaction> = Vec::new();
        for i in 0..count {
            let mut tx = Transaction::new();
            let mut spent: HashSet<(usize, u32)> = HashSet::new();
            if i > 0 {
                for _ in 0..rng.below(5) {
                    let source = rng.below(i as u64) as usize;
                    let vout = rng.below(3) as u32;
                    if spent.insert((source, vout)) {
                        tx.inputs
                            .push(TransactionInput::new(txs[source].id(), vout));
                    }
                }
            }
            for _ in 0..3 {
                tx.outputs.push(TransactionOutput::new(
                    1000 + i as u64,
                    bsv_rs::script::LockingScript::from_hex("51").unwrap(),
                ));
            }
            txs.push(tx);
        }
        // Proofs: a transaction alone in its block, or a pair in one block.
        let mut proofs: Vec<Option<String>> = vec![None; count];
        let mut i = 0;
        while i < count {
            match rng.below(5) {
                0 => {
                    let path = vec![vec![MerklePathLeaf::new_txid(0, txs[i].id())]];
                    proofs[i] = Some(MerklePath::new(100 + i as u32, path).unwrap().to_hex());
                }
                1 if i + 1 < count => {
                    let (a, b) = (txs[i].id(), txs[i + 1].id());
                    let height = 100 + i as u32;
                    let of_a = vec![vec![
                        MerklePathLeaf::new_txid(0, a.clone()),
                        MerklePathLeaf::new(1, b.clone()),
                    ]];
                    let of_b = vec![vec![
                        MerklePathLeaf::new(0, a),
                        MerklePathLeaf::new_txid(1, b),
                    ]];
                    proofs[i] = Some(MerklePath::new(height, of_a).unwrap().to_hex());
                    proofs[i + 1] = Some(MerklePath::new(height, of_b).unwrap().to_hex());
                    i += 1;
                }
                _ => {}
            }
            i += 1;
        }
        let root_key = format!("{}.0", txs[count - 1].id());
        let mut refs: HashMap<String, PendingNode> = HashMap::new();
        refs.insert(
            root_key.clone(),
            pending(txs[count - 1].to_hex(), 0, proofs[count - 1].clone()),
        );
        // Every referenced outpoint is a node, most of the time; a node is
        // a child of the node(s) of its spender.
        for (i, tx) in txs.iter().enumerate().rev() {
            let own_keys: Vec<String> = (0..3)
                .map(|vout| format!("{}.{vout}", tx.id()))
                .filter(|key| refs.contains_key(key))
                .collect();
            if own_keys.is_empty() || proofs[i].is_some() && rng.below(3) > 0 {
                continue;
            }
            for input in &tx.inputs {
                let source_txid = input.get_source_txid().unwrap();
                let source = txs.iter().position(|t| t.id() == source_txid).unwrap();
                let key = format!("{source_txid}.{}", input.source_output_index);
                if rng.below(6) == 0 {
                    continue;
                }
                let mut node = pending(
                    txs[source].to_hex(),
                    input.source_output_index,
                    proofs[source].clone(),
                );
                match rng.below(40) {
                    0 => node.node.proof = Some("zz".to_string()),
                    1 => node.node.raw_tx = "00".to_string(),
                    _ => {}
                }
                refs.insert(key.clone(), node);
                for own in &own_keys {
                    let children = &mut refs.get_mut(own).unwrap().children;
                    if !children.contains(&key) {
                        children.push(key.clone());
                    }
                }
            }
        }
        (refs, root_key)
    }

    // The assembly against the recursion it replaced, over 600 random
    // graphs, tolerant and strict: the BEEF of every node, the ordered BEEFs
    // of the graph, and every refusal, are the same bytes and the same words.
    #[test]
    fn e586_the_assembly_gives_the_hydrated_trees_bytes_and_errors() {
        let mut rng = Lcg(0x586);
        let (mut beefs, mut refusals, mut proven, mut partial) = (0, 0, 0, 0);
        for _ in 0..600 {
            let (refs, root_key) = random_graph(&mut rng);
            for strict in [false, true] {
                let mut keys: Vec<&String> = refs.keys().collect();
                keys.sort();
                for key in keys {
                    let legacy = legacy_get_beef_for_node(key, &refs, strict)
                        .map(|(_, beef)| beef)
                        .map_err(|e| e.to_string());
                    // A fresh assembly per node, and one shared by the graph
                    // below (`compute_ordered_beefs`): both orders of use.
                    let mut assembly = GraphAssembly::default();
                    let ours = assembly
                        .check(key, &refs, strict)
                        .map(|()| assembly.beef_of(key))
                        .map_err(|e| e.to_string());
                    assert_eq!(ours, legacy, "node {key}, strict {strict}");
                    match &ours {
                        Ok(beef) => {
                            beefs += 1;
                            let parsed = Beef::from_binary(beef).unwrap();
                            proven += usize::from(!parsed.bumps.is_empty());
                            partial += usize::from(
                                !strict && legacy_get_beef_for_node(key, &refs, true).is_err(),
                            );
                        }
                        Err(_) => refusals += 1,
                    }
                }
                let legacy = legacy_compute_ordered_beefs(&root_key, &refs, strict)
                    .map_err(|e| e.to_string());
                let ours =
                    OverlayGASPStorage::compute_ordered_beefs(&root_key, &refs, strict, None)
                        .map_err(|e| e.to_string());
                assert_eq!(ours, legacy, "graph {root_key}, strict {strict}");
            }
        }
        println!(
            "#586: {beefs} BEEFs equal ({proven} with a proof, {partial} partial), {refusals} refusals equal"
        );
        // The generator reached every class.
        assert!(beefs > 2000 && proven > 300 && partial > 100 && refusals > 300);
    }

    // A node unknown to the graph, and the SDK's own words for a missing
    // source under `strict_beef`.
    #[test]
    fn e586_the_refusals_are_the_recursions_words() {
        let parent = "ab".repeat(32);
        let raw = make_valid_tx_hex(&parent, 3);
        let key = format!("{}.0", Transaction::from_hex(&raw).unwrap().id());
        let refs = HashMap::from([(key.clone(), pending(raw, 0, None))]);
        let mut assembly = GraphAssembly::default();
        assert_eq!(
            assembly
                .check("nope.0", &refs, false)
                .unwrap_err()
                .to_string(),
            legacy_get_beef_for_node("nope.0", &refs, false)
                .unwrap_err()
                .to_string()
        );
        let strict = assembly.check(&key, &refs, true).unwrap_err().to_string();
        assert_eq!(
            strict,
            legacy_get_beef_for_node(&key, &refs, true)
                .unwrap_err()
                .to_string()
        );
        assert!(strict.contains(&format!(
            "Missing source transaction for input 0 (txid: {parent})"
        )));
        // Tolerant, the same node assembles: a partial BEEF of one transaction.
        assembly.check(&key, &refs, false).unwrap();
        assert_eq!(
            assembly.beef_of(&key),
            legacy_get_beef_for_node(&key, &refs, false).unwrap().1
        );
    }

    // The walks keep their work on the heap. A chain of 20,000 unproven
    // links is assembled; the recursion it replaced used the stack once per
    // link. And a link back into a node the walk is still inside (the root
    // is filed under the graph id the PEER names, so a peer can file a root
    // under an outpoint its own ancestry spends) ends there: the recursion
    // never returned.
    #[test]
    fn e586_a_deep_chain_and_a_cycle_do_not_use_the_stack() {
        use bsv_rs::transaction::{TransactionInput, TransactionOutput};
        let mut refs: HashMap<String, PendingNode> = HashMap::new();
        let mut previous: Option<String> = None;
        let mut tip = String::new();
        for i in 0..20_000u64 {
            let mut tx = Transaction::new();
            if let Some(txid) = &previous {
                tx.inputs.push(TransactionInput::new(txid.clone(), 0));
            }
            tx.outputs.push(TransactionOutput::new(
                i,
                bsv_rs::script::LockingScript::from_hex("51").unwrap(),
            ));
            let txid = tx.id();
            tip = format!("{txid}.0");
            let mut node = pending(tx.to_hex(), 0, None);
            if let Some(below) = &previous {
                node.children.push(format!("{below}.0"));
            }
            refs.insert(tip.clone(), node);
            previous = Some(txid);
        }
        assert_eq!(OverlayGASPStorage::ordered_keys(&tip, &refs).len(), 20_000);
        let mut assembly = GraphAssembly::default();
        assembly.check(&tip, &refs, true).unwrap();
        let beef = Beef::from_binary(&assembly.beef_of(&tip)).unwrap();
        assert_eq!(beef.txs.len(), 20_000);

        // root (filed under `decoy.0`) spends y.0; y spends `decoy.0`.
        let decoy = "cd".repeat(32);
        let y_raw = make_valid_tx_hex(&decoy, 0);
        let y = Transaction::from_hex(&y_raw).unwrap().id();
        let root_raw = make_valid_tx_hex(&y, 0);
        let root_key = format!("{decoy}.0");
        let mut root = pending(root_raw, 0, None);
        root.children.push(format!("{y}.0"));
        let refs = HashMap::from([
            (root_key.clone(), root),
            (format!("{y}.0"), pending(y_raw, 0, None)),
        ]);
        let mut assembly = GraphAssembly::default();
        assembly.check(&root_key, &refs, true).unwrap();
        let beef = Beef::from_binary(&assembly.beef_of(&root_key)).unwrap();
        assert_eq!(beef.txs.len(), 2);
        let beefs =
            OverlayGASPStorage::compute_ordered_beefs(&root_key, &refs, true, None).unwrap();
        assert_eq!(beefs.len(), 2);
    }
}