Skip to main content

bsv_overlay_engine/
gasp_overlay.rs

1//! OverlayGASPStorage — bridges the Engine's storage to the GASP protocol.
2//!
3//! Wraps a reference to the Engine's `Storage` trait plus a topic name, providing
4//! the `GASPStorage` interface that `GASPSync` needs to synchronize UTXOs with
5//! remote peers.
6//!
7//! ## finalize_graph
8//!
9//! When a graph is finalized, the nodes are converted into ordered BEEF byte
10//! arrays (ancestors first, root last) and stored in a shared
11//! `FinalizedGraphSink`. The caller (Engine) drains the sink and submits each
12//! BEEF to `Engine::submit()` with `HistoricalTxNoSpv` mode: after
13//! `GASPSync::sync()` returns, or, under a per-peer sync budget, after every
14//! completed UTXO (`gasp::FinalizedGraphHook`, bsv-low #552).
15//!
16//! ## find_needed_inputs
17//!
18//! Parses the node's raw transaction hex to determine what inputs are needed:
19//! - A proven node ends the walk unless its topic manager names inputs needed
20//!   for overlay history: when its output is not yet admissible, those not
21//!   held; when it is admissible, those neither held nor landed (lane E1D).
22//! - If no proof, all transaction inputs are requested (minus any already
23//!   known in local storage).
24//!
25//! ## validate_graph_anchor (bsv-low #551)
26//!
27//! Before a graph is finalized it is checked the way the reference checks it
28//! (`OverlayGASPStorage.ts` `validateGraphAnchor`): the ROOT node's BEEF is
29//! verified by the engine's own SPV check, then the ordered BEEFs are replayed
30//! through the topic manager over a set of coins, and the graph is discarded
31//! WHOLE unless its root is a coin at the end. That is what makes the
32//! `HistoricalTxNoSpv` submits of `finalize_graph` rest on a true premise.
33//!
34//! Ported from `~/bsv/overlay-services/src/GASP/OverlayGASPStorage.ts` (388 lines).
35
36use crate::beef_limits;
37
38use std::collections::{HashMap, HashSet};
39use std::rc::Rc;
40use std::sync::atomic::{AtomicBool, Ordering};
41use std::sync::Mutex;
42
43use async_trait::async_trait;
44use bsv_rs::transaction::{Beef, ChainTracker, ChainTrackerError, MerklePath, Transaction};
45use tracing::{debug, error, warn};
46
47use crate::gasp::{GASPError, GASPStorage};
48use crate::storage::Storage;
49use crate::topic_manager::TopicManager;
50use crate::types::{
51    GASPInputRequest, GASPNode, GASPNodeResponse, GASPOutput, SubmitMode, TopicAdmittanceContext,
52};
53
54/// The most store reads [`OverlayGASPStorage::find_needed_inputs`] spends
55/// on asking whether an admitted node's named inputs have LANDED (E1D's
56/// re-ask; the E1D lens fold, L3): two per transaction at most (its applied
57/// row, then its held outputs). A transaction past it is requested from the
58/// peer. The figure is the door's own per-topic bound (`PREDECESSOR_READS`).
59const LANDED_READS_PER_NODE: usize = 16;
60
61/// A GASP node stored during in-progress graph construction.
62#[derive(Debug, Clone)]
63struct PendingNode {
64    node: GASPNode,
65    #[allow(dead_code)]
66    spent_by: Option<String>,
67    /// Children of this node (nodes whose transactions are inputs to this one).
68    children: Vec<String>, // keys into the pending_graphs map
69}
70
71/// A finalized graph ready for Engine submission.
72///
73/// Contains ordered BEEF byte arrays (ancestors first, root last) and
74/// the topic they belong to.
75#[derive(Debug, Clone)]
76pub struct FinalizedGraph {
77    /// Topic this graph belongs to.
78    pub topic: String,
79    /// Ordered BEEF byte arrays — ancestors first, root transaction last.
80    pub beefs: Vec<Vec<u8>>,
81}
82
83/// Shared container for finalized graph data.
84///
85/// The Engine creates this before sync and passes it to `OverlayGASPStorage`.
86/// After sync completes, the Engine drains the list and submits each BEEF.
87pub type FinalizedGraphSink = Rc<Mutex<Vec<FinalizedGraph>>>;
88
89/// Create a new empty finalized graph sink.
90pub fn new_finalized_graph_sink() -> FinalizedGraphSink {
91    Rc::new(Mutex::new(Vec::new()))
92}
93
94/// `GASPStorage` implementation that delegates to the Engine's `Storage` trait.
95///
96/// Tracks one topic at a time — create a new instance per (topic, peer) sync.
97/// During sync, incoming graph nodes are accumulated in memory. On `finalize_graph`,
98/// the accumulated nodes are converted to BEEF format and pushed to the shared
99/// `FinalizedGraphSink` for later Engine submission.
100pub struct OverlayGASPStorage<'a> {
101    fault_memos: Mutex<HashMap<String, crate::gasp::DeferredGraph>>,
102    /// The Engine's storage backend.
103    storage: &'a dyn Storage,
104    /// Topic being synchronized.
105    topic: String,
106    /// Optional topic manager for overlay-specific history behind proven nodes.
107    topic_manager: Option<&'a dyn TopicManager>,
108    /// Temporary graphs being constructed during sync.
109    /// Key: node identifier (graph_id for root, "txid.outputIndex" for children).
110    /// Value: the pending node with its relationship data.
111    pending_graphs: Mutex<HashMap<String, PendingNode>>,
112    /// Shared sink for completed graphs ready for Engine submission.
113    finalized_sink: FinalizedGraphSink,
114    /// If true, finalize uses strict `to_beef(false)` so a missing ancestor
115    /// fails loud rather than silently emitting a partial BEEF. Defaults to
116    /// `false` (tolerant `to_beef(true)`) — byte-identical to today. Only set
117    /// `true` when ancestor hydration is enabled, so post-hydration
118    /// completeness is enforced.
119    strict_beef: bool,
120    /// The engine's chain tracker, for the anchor's Bitcoin check. `None` is
121    /// the reference's `'scripts only'`: a merkle path is accepted unchecked,
122    /// every unproven input's script still runs.
123    chain_tracker: Option<&'a dyn ChainTracker>,
124    /// [`crate::engine::Engine::set_script_verification`], handed over so the
125    /// anchor check obeys the same switch as `Engine::submit`. Default `true`.
126    verify_scripts: bool,
127    /// [`crate::engine::Engine::set_walk_budget`], handed over so the anchor
128    /// check's script walk runs under the same two limbs as `Engine::submit`
129    /// (bsv-low #592). Default [`crate::engine::DoorBudget::DEFAULT`].
130    walk_budget: crate::engine::DoorBudget,
131    /// The peer this instance syncs from: the key of its deferred graphs
132    /// (bsv-low #555, [`Self::with_peer`]).
133    peer: String,
134    /// Whether that peer is CONFIGURED (`SyncTarget::Peers`), written into
135    /// each record saved (the delta-2 fold's D2-M2, [`Self::with_configured_peer`]).
136    configured_peer: bool,
137    /// The graph the anchor check last PASSED, parsed and checked, kept for
138    /// its `finalize_graph` so a graph is parsed once (bsv-low #586). Dropped
139    /// by anything that changes or discards a pending graph.
140    anchored: Mutex<Option<AnchoredGraph>>,
141}
142
143/// A graph the anchor check passed: what `finalize_graph` assembles from.
144struct AnchoredGraph {
145    graph_id: String,
146    strict_beef: bool,
147    order: Vec<String>,
148    assembly: GraphAssembly,
149}
150
151/// Lends the engine's tracker to the anchor check and notes whether it
152/// FAULTED: a tracker that could not answer is a fault of the moment, not a
153/// verdict on the graph (see `validate_graph_anchor`). Noted here, on the
154/// call, so no error text is matched.
155struct FaultNotingTracker<'a> {
156    inner: &'a dyn ChainTracker,
157    faulted: AtomicBool,
158}
159
160#[async_trait]
161impl ChainTracker for FaultNotingTracker<'_> {
162    async fn is_valid_root_for_height(
163        &self,
164        root: &str,
165        height: u32,
166    ) -> Result<bool, ChainTrackerError> {
167        let answer = self.inner.is_valid_root_for_height(root, height).await;
168        if answer.is_err() {
169            self.faulted.store(true, Ordering::Relaxed);
170        }
171        answer
172    }
173
174    async fn current_height(&self) -> Result<u32, ChainTrackerError> {
175        self.inner.current_height().await
176    }
177}
178
179/// What the anchor check needs of a pending graph, assembled under the lock.
180struct AnchorInput {
181    /// The root node's transaction id.
182    root_txid: String,
183    /// The root node's BEEF (tolerant: a source the graph does not carry is
184    /// absent, and merged from local storage or refused by the verifier).
185    root_beef: Vec<u8>,
186    /// Inputs of the root's UNPROVEN ancestry that the graph does not carry.
187    absent_sources: Vec<(String, u32)>,
188    /// The keys of the graph's nodes in finalize order: ancestors first,
189    /// root last. Each one's BEEF is assembled when the replay reaches it
190    /// and dropped after it (bsv-low #586): the check never holds the
191    /// graph's BEEFs together.
192    order: Vec<String>,
193    /// The graph's transactions and proofs, parsed once and CHECKED: every
194    /// BEEF of `order` can be assembled from it.
195    assembly: GraphAssembly,
196}
197
198/// One transaction of a pending graph, parsed once.
199struct ParsedTx {
200    tx: Transaction,
201    txid: String,
202}
203
204/// One node of a pending graph as its BEEFs need it.
205struct AssemblyNode {
206    /// Shared by every node of one transaction (one node per output index
207    /// an unproven spender asked for).
208    parsed: Rc<ParsedTx>,
209    /// The node's own proof. A proven node is a leaf of every BEEF.
210    proof: Option<MerklePath>,
211    /// Per input of an UNPROVEN node: the key of the node that holds its
212    /// source, when the graph has one. Empty for a proven node.
213    sources: Vec<Option<String>>,
214}
215
216/// A pending graph's BEEFs, assembled from its raw transactions and proofs
217/// parsed ONCE each (bsv-low #586, zanaadu-v2 #377).
218///
219/// Before it, `get_beef_for_node` hydrated a `source_transaction` tree by
220/// recursion: every reference to a shared ancestor rebuilt that ancestor's
221/// whole subtree (parse, raw bytes and hex of each transaction, cloned per
222/// parent), and serialized it at every level, so an unmined graph with
223/// fan-in cost its PATHS, not its transactions: Zanaadu measured 36x the
224/// fetched bytes on 13 nodes, each further unproven layer multiplying by its
225/// fan-in. The reference hydrates the same way (`OverlayGASPStorage.ts`
226/// `getBEEFForNode`, f999e0c1a: `hydrator` parses a node again for every
227/// input that reaches it; JavaScript shares nothing between the copies
228/// either), so it has the same multiplier.
229///
230/// Here a transaction is parsed once per graph (keyed by its bytes), a node
231/// names its sources by KEY, and a node's BEEF is the walk
232/// `Transaction::to_beef` made over the hydrated tree (`bsv-rs`
233/// `collect_ancestors`: depth first, inputs from the last to the first, a
234/// proven node a leaf, a txid collected once, a transaction after its
235/// sources; proofs deduplicated by height and root), made over the keys
236/// instead, then the SDK's own `Beef` (`merge_bump`, `merge_transaction`,
237/// `to_binary`, whose sort is the SDK's). The bytes are the ones the tree
238/// gave, for every graph: the frozen digests of `tests/gasp_topic_manager.rs`
239/// and of the witness hold.
240///
241/// Both walks keep their work on the heap (the recursion they replace used
242/// the stack, once per unproven link), and neither follows a link back into
243/// a node it is still inside.
244#[derive(Default)]
245struct GraphAssembly {
246    /// By the SHA-256 of a transaction's raw bytes.
247    txs: HashMap<[u8; 32], Rc<ParsedTx>>,
248    /// The transaction of each node key asked for.
249    parsed: HashMap<String, Rc<ParsedTx>>,
250    /// By node key.
251    nodes: HashMap<String, Rc<AssemblyNode>>,
252    /// (node key, strict) whose BEEF is known to assemble.
253    checked: HashSet<(String, bool)>,
254}
255
256impl GraphAssembly {
257    /// The transaction of the node `key`, parsed once per graph.
258    fn parsed_tx(
259        &mut self,
260        key: &str,
261        refs: &HashMap<String, PendingNode>,
262    ) -> Result<Rc<ParsedTx>, GASPError> {
263        if let Some(parsed) = self.parsed.get(key) {
264            return Ok(parsed.clone());
265        }
266        let pending = refs
267            .get(key)
268            .ok_or_else(|| GASPError::Other(format!("Node {key} not found in graph refs")))?;
269        let unparsed =
270            |e: bsv_rs::Error| GASPError::Other(format!("Failed to parse raw_tx for {key}: {e}"));
271        let raw = bsv_rs::primitives::from_hex(&pending.node.raw_tx).map_err(unparsed)?;
272        let parsed = match self.txs.entry(bsv_rs::primitives::hash::sha256(&raw)) {
273            std::collections::hash_map::Entry::Occupied(known) => known.get().clone(),
274            std::collections::hash_map::Entry::Vacant(new) => {
275                let tx = Transaction::from_binary(&raw).map_err(unparsed)?;
276                let txid = tx.id();
277                new.insert(Rc::new(ParsedTx { tx, txid })).clone()
278            }
279        };
280        self.parsed.insert(key.to_string(), parsed.clone());
281        Ok(parsed)
282    }
283
284    /// The node `key`: its transaction, its proof, and where the graph
285    /// holds each input's source.
286    fn node(
287        &mut self,
288        key: &str,
289        refs: &HashMap<String, PendingNode>,
290    ) -> Result<Rc<AssemblyNode>, GASPError> {
291        if let Some(node) = self.nodes.get(key) {
292            return Ok(node.clone());
293        }
294        let parsed = self.parsed_tx(key, refs)?;
295        let proof =
296            match refs
297                .get(key)
298                .and_then(|pending| pending.node.proof.as_ref())
299            {
300                Some(proof_hex) => Some(beef_limits::proof_from_hex(proof_hex).map_err(|e| {
301                    GASPError::Other(format!("Failed to parse proof for {key}: {e}"))
302                })?),
303                None => None,
304            };
305        // A proven node is a leaf. An unproven one takes each input's source
306        // from the node filed under that outpoint, when the graph has it.
307        let sources = if proof.is_some() {
308            Vec::new()
309        } else {
310            parsed
311                .tx
312                .inputs
313                .iter()
314                .map(|input| {
315                    let source_txid = input.get_source_txid().unwrap_or_default();
316                    if source_txid.is_empty() {
317                        return None;
318                    }
319                    let child_key = format!("{}.{}", source_txid, input.source_output_index);
320                    refs.contains_key(&child_key).then_some(child_key)
321                })
322                .collect()
323        };
324        let node = Rc::new(AssemblyNode {
325            parsed,
326            proof,
327            sources,
328        });
329        self.nodes.insert(key.to_string(), node.clone());
330        Ok(node)
331    }
332
333    /// Check that the BEEF of `key` can be assembled: every node it reaches
334    /// parses (its transaction, its proof), and under `strict` no unproven
335    /// node lacks a source (`to_beef(false)`'s rule; tolerant, a missing
336    /// source is left out silently, `to_beef(true)`). The errors are the ones
337    /// the hydrating recursion gave, in its order: a node on entry, its
338    /// sources in input order, its own missing source on the way out. A node
339    /// is checked once per graph.
340    fn check(
341        &mut self,
342        key: &str,
343        refs: &HashMap<String, PendingNode>,
344        strict: bool,
345    ) -> Result<(), GASPError> {
346        // (node key, the node, the next input to follow)
347        let mut stack: Vec<(String, Rc<AssemblyNode>, usize)> = Vec::new();
348        let mut inside: HashSet<String> = HashSet::new();
349        let mut entering: Option<String> = Some(key.to_string());
350        loop {
351            if let Some(key) = entering.take() {
352                if !self.checked.contains(&(key.clone(), strict)) && inside.insert(key.clone()) {
353                    let node = self.node(&key, refs)?;
354                    stack.push((key, node, 0));
355                }
356            }
357            let Some((_, node, next)) = stack.last_mut() else {
358                return Ok(());
359            };
360            if let Some(source) = node.sources.get(*next) {
361                *next += 1;
362                entering.clone_from(source);
363                continue;
364            }
365            let (key, node, _) = stack.pop().expect("the stack is not empty");
366            if strict {
367                if let Some(missing) = node.sources.iter().rposition(Option::is_none) {
368                    let source_txid = node.parsed.tx.inputs[missing]
369                        .source_txid
370                        .as_deref()
371                        .unwrap_or("unknown");
372                    // The SDK's own refusal, word for word (`to_beef(false)`).
373                    let e = bsv_rs::Error::TransactionError(format!(
374                        "Missing source transaction for input {missing} (txid: {source_txid}). \
375                         Set allow_partial=true to skip missing source transactions."
376                    ));
377                    return Err(GASPError::Other(format!(
378                        "Failed to serialize BEEF for {key}: {e}"
379                    )));
380                }
381            }
382            inside.remove(&key);
383            self.checked.insert((key, strict));
384        }
385    }
386
387    /// The txid of a checked node.
388    fn txid_of(&self, key: &str) -> Option<&str> {
389        self.nodes.get(key).map(|node| node.parsed.txid.as_str())
390    }
391
392    /// The BEEF of a CHECKED node: its transaction and its unproven
393    /// ancestry down to the first proven node of every branch, each
394    /// transaction once. Nothing is hydrated: the only copies made are the
395    /// ones the `Beef` being written holds, dropped with it.
396    fn beef_of(&self, key: &str) -> Vec<u8> {
397        struct Frame<'n> {
398            node: &'n AssemblyNode,
399            inputs_left: usize,
400        }
401        let mut seen: HashSet<&str> = HashSet::new();
402        let mut inside: HashSet<&str> = HashSet::new();
403        let mut ancestors: Vec<&AssemblyNode> = Vec::new();
404        let mut bumps: Vec<MerklePath> = Vec::new();
405        let mut bump_index_by_root: HashMap<String, usize> = HashMap::new();
406        let mut stack: Vec<Frame> = Vec::new();
407        let mut entering: Option<&AssemblyNode> = self.nodes.get(key).map(Rc::as_ref);
408        loop {
409            if let Some(node) = entering.take() {
410                let txid = node.parsed.txid.as_str();
411                if seen.contains(txid) || inside.contains(txid) {
412                    // Already collected (or being collected): skip it.
413                } else if let Some(merkle_path) = &node.proof {
414                    // Proven: its proof and itself, and no further. Proofs
415                    // are deduplicated by block height and computed root.
416                    let root = merkle_path.compute_root(Some(txid)).unwrap_or_default();
417                    let slot = format!("{}:{}", merkle_path.block_height, root);
418                    match bump_index_by_root.get(&slot) {
419                        Some(&existing) if bumps[existing].combine(merkle_path).is_ok() => {}
420                        _ => {
421                            bump_index_by_root.insert(slot, bumps.len());
422                            bumps.push(merkle_path.clone());
423                        }
424                    }
425                    seen.insert(txid);
426                    ancestors.push(node);
427                } else {
428                    inside.insert(txid);
429                    stack.push(Frame {
430                        node,
431                        inputs_left: node.sources.len(),
432                    });
433                }
434            }
435            let Some(top) = stack.last_mut() else {
436                break;
437            };
438            if top.inputs_left == 0 {
439                let node = stack.pop().expect("the stack is not empty").node;
440                inside.remove(node.parsed.txid.as_str());
441                seen.insert(node.parsed.txid.as_str());
442                ancestors.push(node);
443                continue;
444            }
445            top.inputs_left -= 1;
446            if let Some(source) = &top.node.sources[top.inputs_left] {
447                entering = self.nodes.get(source).map(Rc::as_ref);
448            }
449        }
450
451        let mut beef = Beef::new();
452        for bump in bumps {
453            beef.merge_bump(bump);
454        }
455        for node in ancestors {
456            beef.merge_transaction(node.parsed.tx.clone());
457        }
458        // The writer grows by doubling: a graph's BEEFs are held together
459        // by the finalize, so each gives its slack back.
460        let mut bytes = beef.to_binary();
461        bytes.shrink_to_fit();
462        bytes
463    }
464}
465
466impl<'a> OverlayGASPStorage<'a> {
467    /// Create a new `OverlayGASPStorage` for the given topic.
468    ///
469    /// The `sink` parameter is a shared container where finalized graphs are
470    /// pushed. The Engine holds a clone and drains it after sync.
471    pub fn new(
472        storage: &'a dyn Storage,
473        topic: impl Into<String>,
474        sink: FinalizedGraphSink,
475    ) -> Self {
476        Self {
477            storage,
478            topic: topic.into(),
479            topic_manager: None,
480            pending_graphs: Mutex::new(HashMap::new()),
481            fault_memos: Mutex::new(HashMap::new()),
482            finalized_sink: sink,
483            strict_beef: false,
484            chain_tracker: None,
485            verify_scripts: true,
486            walk_budget: crate::engine::DoorBudget::DEFAULT,
487            peer: String::new(),
488            configured_peer: false,
489            anchored: Mutex::new(None),
490        }
491    }
492
493    /// Keep a graph the anchor check passed for its finalize.
494    fn keep_anchored(&self, graph_id: &str, order: Vec<String>, assembly: GraphAssembly) {
495        *self
496            .anchored
497            .lock()
498            .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(AnchoredGraph {
499            graph_id: graph_id.to_string(),
500            strict_beef: self.strict_beef,
501            order,
502            assembly,
503        });
504    }
505
506    /// Take the graph kept by the anchor check (leaving none).
507    fn take_anchored(&self) -> Option<AnchoredGraph> {
508        self.anchored
509            .lock()
510            .unwrap_or_else(std::sync::PoisonError::into_inner)
511            .take()
512    }
513
514    /// Name the peer this instance syncs from, the key under which its
515    /// deferred graphs are kept (bsv-low #555). `Engine::start_gasp_sync`
516    /// wires it.
517    #[must_use]
518    pub fn with_peer(mut self, peer: impl Into<String>) -> Self {
519        self.peer = peer.into();
520        self
521    }
522
523    /// Say whether the peer is a CONFIGURED one (`SyncTarget::Peers`) or
524    /// one `ls_ship` discovered (bsv-low #555, the delta-2 fold's D2-M2):
525    /// each record saved carries it (`DeferredGraph::configured`), so a
526    /// storage can reserve part of its ceiling for configured peers.
527    /// Default `false`. `Engine::start_gasp_sync` wires it.
528    #[must_use]
529    pub fn with_configured_peer(mut self, configured: bool) -> Self {
530        self.configured_peer = configured;
531        self
532    }
533
534    /// Check merkle roots of the anchor against this chain tracker (the
535    /// engine's: `Engine::start_gasp_sync` wires it). Without one the anchor
536    /// check is the reference's `'scripts only'` walk.
537    #[must_use]
538    pub fn with_chain_tracker(mut self, tracker: &'a dyn ChainTracker) -> Self {
539        self.chain_tracker = Some(tracker);
540        self
541    }
542
543    /// Carry [`crate::engine::Engine::set_script_verification`] into the
544    /// anchor check. `false` is that switch's escape hatch, not a mode: the
545    /// pre-2026-09-08 structural check (roots only, and nothing without a
546    /// tracker). Default `true`.
547    #[must_use]
548    pub fn with_script_verification(mut self, enabled: bool) -> Self {
549        self.verify_scripts = enabled;
550        self
551    }
552
553    /// Carry [`crate::engine::Engine::set_walk_budget`] into the anchor
554    /// check (bsv-low #592).
555    #[must_use]
556    pub fn with_walk_budget(mut self, budget: crate::engine::DoorBudget) -> Self {
557        self.walk_budget = budget;
558        self
559    }
560
561    /// Consult the topic manager when a proven node needs overlay history.
562    /// Without a manager, proven nodes retain the legacy stopping rule.
563    #[must_use]
564    pub fn with_topic_manager(mut self, manager: &'a dyn TopicManager) -> Self {
565        self.topic_manager = Some(manager);
566        self
567    }
568
569    /// Enable strict-BEEF finalize (`to_beef(false)`), failing loud on a
570    /// missing ancestor instead of silently emitting a partial BEEF.
571    ///
572    /// Defaults to OFF (tolerant). Only enable this together with ancestor
573    /// hydration so that, after the chain fallback has spliced in ancestors,
574    /// an incomplete graph is discarded + retried rather than stored partial.
575    #[must_use]
576    pub fn with_strict_beef(mut self, strict: bool) -> Self {
577        self.strict_beef = strict;
578        self
579    }
580
581    /// Take all finalized graphs from the shared sink, leaving it empty.
582    ///
583    /// Convenience method for callers who hold a reference to this storage
584    /// rather than the raw sink.
585    pub fn take_finalized_graphs(&self) -> Vec<FinalizedGraph> {
586        self.finalized_sink
587            .lock()
588            .unwrap_or_else(std::sync::PoisonError::into_inner)
589            .drain(..)
590            .collect()
591    }
592
593    /// Get the number of pending graphs (for testing).
594    #[cfg(test)]
595    pub fn pending_graph_count(&self) -> usize {
596        let refs = self.pending_graphs.lock().unwrap();
597        let graph_ids: std::collections::HashSet<&str> =
598            refs.values().map(|pn| pn.node.graph_id.as_str()).collect();
599        graph_ids.len()
600    }
601
602    /// Get the number of finalized graphs (for testing).
603    #[cfg(test)]
604    pub fn finalized_graph_count(&self) -> usize {
605        self.finalized_sink.lock().unwrap().len()
606    }
607
608    /// The keys of a graph's nodes in the order their BEEFs are finalized
609    /// (ancestors first, root last): depth first from the root over each
610    /// node's `children`, a node after its children, each key once. An
611    /// explicit stack (bsv-low #586; it was a recursion, in this order).
612    fn ordered_keys(graph_id: &str, refs: &HashMap<String, PendingNode>) -> Vec<String> {
613        let mut order: Vec<String> = Vec::new();
614        let mut visited: HashSet<&str> = HashSet::new();
615        // (node key, its children, the next child to visit)
616        let mut stack: Vec<(&str, &[String], usize)> = Vec::new();
617        let mut entering: Option<&str> = Some(graph_id);
618        loop {
619            if let Some(key) = entering.take() {
620                if visited.insert(key) {
621                    if let Some(pending) = refs.get(key) {
622                        stack.push((key, &pending.children, 0));
623                    }
624                }
625            }
626            let Some((key, children, next)) = stack.last_mut() else {
627                return order;
628            };
629            if let Some(child) = children.get(*next) {
630                *next += 1;
631                entering = Some(child.as_str());
632            } else {
633                order.push((*key).to_string());
634                stack.pop();
635            }
636        }
637    }
638
639    /// Compute ordered BEEFs for a graph (ancestors first, root last), each
640    /// node's BEEF assembled from the graph's transactions parsed ONCE
641    /// ([`GraphAssembly`]).
642    ///
643    /// `anchored`: the graph as the anchor check left it, already parsed and
644    /// checked; without one (a finalize with no anchor check before it) it is
645    /// parsed and checked here.
646    fn compute_ordered_beefs(
647        graph_id: &str,
648        refs: &HashMap<String, PendingNode>,
649        strict_beef: bool,
650        anchored: Option<AnchoredGraph>,
651    ) -> Result<Vec<Vec<u8>>, GASPError> {
652        let (order, assembly) = match anchored {
653            Some(kept) if kept.graph_id == graph_id && kept.strict_beef == strict_beef => {
654                (kept.order, kept.assembly)
655            }
656            _ => {
657                let mut assembly = GraphAssembly::default();
658                let order = Self::ordered_keys(graph_id, refs);
659                for key in &order {
660                    assembly.check(key, refs, strict_beef)?;
661                }
662                (order, assembly)
663            }
664        };
665        Ok(order.iter().map(|key| assembly.beef_of(key)).collect())
666    }
667
668    /// Everything `validate_graph_anchor` needs of the pending graph, read in
669    /// one pass under the lock (nothing awaits while the guard is held).
670    ///
671    /// It also enforces the one structural rule the rest relies on: every
672    /// node hangs off its parent by an input the parent REALLY spends. The
673    /// walk asks the peer for an outpoint and files the answer under the
674    /// answer's own txid, so a peer that answers with some other transaction
675    /// would otherwise ride into the ordered BEEFs unlinked to anything the
676    /// Bitcoin check covers. With the rule, every node of the graph is tied
677    /// to the verified root by a chain of txids. (A divergence by addition
678    /// from the reference at f999e0c1a, which has no such rule; upstream added
679    /// it later in `appendToGraph`: "The GASP child node is not an input of
680    /// its declared parent".)
681    ///
682    /// A graph that cannot be ASSEMBLED (a raw transaction or a proof that
683    /// does not parse, or under `strict_beef` a missing ancestor) is
684    /// `GASPError::AnchorUnavailable`, not a refusal: before #551 that error
685    /// surfaced from `finalize_graph` and failed the UTXO, so the cursor gap
686    /// guard asked for it again (#43), and it still does.
687    fn read_anchor_input(
688        graph_id: &str,
689        refs: &HashMap<String, PendingNode>,
690        strict_beef: bool,
691    ) -> Result<AnchorInput, GASPError> {
692        let unassembled = |e: GASPError| {
693            GASPError::AnchorUnavailable(format!("graph {graph_id} cannot be assembled: {e}"))
694        };
695        if !refs.contains_key(graph_id) {
696            return Err(GASPError::ValidationFailed(format!(
697                "Graph node with ID {graph_id} not found"
698            )));
699        }
700
701        let mut assembly = GraphAssembly::default();
702        let mut absent_sources: Vec<(String, u32)> = Vec::new();
703        let mut visited: HashSet<&str> = HashSet::new();
704        // (node key, whether a path of UNPROVEN nodes leads from the root to it)
705        let mut stack: Vec<(&str, bool)> = vec![(graph_id, true)];
706        while let Some((key, in_root_beef)) = stack.pop() {
707            if !visited.insert(key) {
708                continue;
709            }
710            let Some(pending) = refs.get(key) else {
711                continue;
712            };
713            let parsed = assembly.parsed_tx(key, refs).map_err(unassembled)?;
714            let spends: HashSet<String> = parsed
715                .tx
716                .inputs
717                .iter()
718                .filter_map(|input| {
719                    let source_txid = input.get_source_txid().ok()?;
720                    Some(format!("{}.{}", source_txid, input.source_output_index))
721                })
722                .collect();
723            for child in &pending.children {
724                if !spends.contains(child) {
725                    return Err(GASPError::ValidationFailed(format!(
726                        "node {child} is not an input of its declared parent {key}"
727                    )));
728                }
729            }
730            let unproven = pending.node.proof.is_none();
731            if in_root_beef && unproven {
732                for outpoint in &spends {
733                    if !refs.contains_key(outpoint) {
734                        if let Some(source) = crate::gasp::parse_outpoint(outpoint) {
735                            absent_sources.push(source);
736                        }
737                    }
738                }
739            }
740            for child in &pending.children {
741                stack.push((child.as_str(), in_root_beef && unproven));
742            }
743        }
744
745        // The root's CHECKED copy is assembled tolerant whatever `strict_beef`
746        // says: it is never stored, sources the storage holds are merged into
747        // it, and the verifier refuses any source still missing. The ordered
748        // BEEFs are assembled exactly as `finalize_graph` will assemble them.
749        assembly.check(graph_id, refs, false).map_err(unassembled)?;
750        let root_beef = assembly.beef_of(graph_id);
751        let root_txid = assembly.txid_of(graph_id).unwrap_or_default().to_string();
752        let order = Self::ordered_keys(graph_id, refs);
753        for key in &order {
754            assembly
755                .check(key, refs, strict_beef)
756                .map_err(unassembled)?;
757        }
758        Ok(AnchorInput {
759            root_txid,
760            root_beef,
761            absent_sources,
762            order,
763            assembly,
764        })
765    }
766}
767
768#[async_trait(?Send)]
769impl GASPStorage for OverlayGASPStorage<'_> {
770    fn open_page(&self) {
771        if let Some(page) = crate::page_batch::installed() {
772            page.open();
773        }
774    }
775
776    async fn close_page(&self) -> Result<(), GASPError> {
777        let Some(page) = crate::page_batch::installed() else {
778            return Ok(());
779        };
780        page.close();
781        match page.flush(self.storage).await {
782            Ok(()) => Ok(()),
783            Err(e) => {
784                page.discard();
785                Err(GASPError::StorageError(format!(
786                    "the page's storage batch failed, nothing of it admitted: {e}"
787                )))
788            }
789        }
790    }
791
792    async fn prefetch_page(&self, roots: &[String], nodes: &[&GASPNode]) {
793        if crate::page_batch::current().is_none() {
794            return;
795        }
796        let mut inputs = Vec::new();
797        for node in nodes {
798            let Ok(tx) = Transaction::from_hex(&node.raw_tx) else {
799                continue;
800            };
801            for input in &tx.inputs {
802                let source = input.get_source_txid().unwrap_or_default();
803                if !source.is_empty() {
804                    inputs.push((source, input.source_output_index));
805                }
806            }
807        }
808        crate::page_batch::prefetch(self.storage, &self.topic, &inputs, roots).await;
809    }
810
811    fn sync_expiry(&self, node: &GASPNode, outpoint: &str) -> Result<Option<u64>, GASPError> {
812        let Some(manager) = self.topic_manager else {
813            return Ok(None);
814        };
815        let tx = Transaction::from_hex(&node.raw_tx)
816            .map_err(|e| GASPError::Other(format!("sync root parse: {e}")))?;
817        if format!("{}.{}", tx.id(), node.output_index) != outpoint {
818            return Err(GASPError::Other(
819                "sync root content-address mismatch".into(),
820            ));
821        }
822        let output = tx
823            .outputs
824            .get(node.output_index as usize)
825            .ok_or_else(|| GASPError::Other("sync root output missing".into()))?;
826        Ok(manager.sync_expiry(&output.locking_script))
827    }
828
829    /// Returns UTXOs known for this topic since the given score/timestamp.
830    ///
831    /// Delegates to `Storage::find_utxos_for_topic()`, converting `Output`
832    /// records to `GASPOutput` with txid, output_index, and score.
833    async fn find_known_utxos(
834        &self,
835        since: u64,
836        limit: Option<u64>,
837    ) -> Result<Vec<GASPOutput>, GASPError> {
838        let outputs = self
839            .storage
840            .find_utxos_for_topic(
841                &self.topic,
842                Some(since as f64),
843                limit,
844                false, // don't need BEEF for listing
845            )
846            .await
847            .map_err(|e| GASPError::StorageError(e.to_string()))?;
848
849        Ok(outputs
850            .iter()
851            .map(|o| GASPOutput {
852                txid: o.txid.clone(),
853                output_index: o.output_index,
854                score: o.score.unwrap_or(0.0),
855            })
856            .collect())
857    }
858
859    /// Hydrate a GASP node with transaction data from storage.
860    ///
861    /// Looks up the root output (from graph_id) and searches its BEEF tree
862    /// for the requested txid. This is a simplified version that returns
863    /// the raw transaction data and proof if available.
864    ///
865    /// For a full implementation, this would delegate to
866    /// `Engine::provide_foreign_gasp_node()`, but since we only hold a
867    /// `Storage` reference (not the full Engine), we do a simpler lookup.
868    async fn hydrate_gasp_node(
869        &self,
870        graph_id: &str,
871        txid: &str,
872        output_index: u32,
873        _metadata: bool,
874    ) -> Result<GASPNode, GASPError> {
875        // Try to find the output with BEEF data
876        let output = self
877            .storage
878            .find_output(txid, output_index, Some(&self.topic), None, true)
879            .await
880            .map_err(|e| GASPError::StorageError(e.to_string()))?;
881
882        if let Some(output) = output {
883            if let Some(ref beef) = output.beef {
884                // Parse the BEEF to extract raw transaction hex
885                match beef_limits::transaction_from_beef(
886                    beef,
887                    None,
888                    &beef_limits::STORED_BEEF_LIMITS,
889                ) {
890                    Ok(tx) => {
891                        let mut node = GASPNode {
892                            graph_id: graph_id.to_string(),
893                            raw_tx: tx.to_hex(),
894                            output_index,
895                            proof: None,
896                            tx_metadata: None,
897                            output_metadata: None,
898                            inputs: None,
899                        };
900                        if let Some(ref merkle_path) = tx.merkle_path {
901                            node.proof = Some(merkle_path.to_hex());
902                        }
903                        return Ok(node);
904                    }
905                    Err(e) => {
906                        warn!("Failed to parse BEEF for {txid}: {e}");
907                    }
908                }
909            }
910        }
911
912        // Fallback: look up without topic filter
913        let output = self
914            .storage
915            .find_output(txid, output_index, None, None, true)
916            .await
917            .map_err(|e| GASPError::StorageError(e.to_string()))?
918            .ok_or_else(|| {
919                GASPError::NodeNotFound(format!("Output {txid}.{output_index} not found"))
920            })?;
921
922        let beef = output.beef.as_ref().ok_or_else(|| {
923            GASPError::NodeNotFound(format!("No BEEF data for {txid}.{output_index}"))
924        })?;
925
926        let tx = beef_limits::transaction_from_beef(beef, None, &beef_limits::STORED_BEEF_LIMITS)
927            .map_err(|e| GASPError::Other(format!("BEEF parse error: {e}")))?;
928
929        let mut node = GASPNode {
930            graph_id: graph_id.to_string(),
931            raw_tx: tx.to_hex(),
932            output_index,
933            proof: None,
934            tx_metadata: None,
935            output_metadata: None,
936            inputs: None,
937        };
938        if let Some(ref merkle_path) = tx.merkle_path {
939            node.proof = Some(merkle_path.to_hex());
940        }
941        Ok(node)
942    }
943
944    /// Determine which inputs are needed to validate this node.
945    ///
946    /// - A proven node ends the walk if its topic manager names no inputs it
947    ///   still needs: for a node whose output is NOT admissible, every named
948    ///   input not held (D13); for one whose output IS admissible, every named
949    ///   input neither held nor landed (lane E1D, an addition: the reference
950    ///   stops at an admissible node). Without a manager, proven nodes always
951    ///   stop.
952    /// - If no proof, parses the raw transaction and requests all inputs,
953    ///   filtering out any inputs already known in local storage.
954    /// - Admission errors propagate to GASP's incoming UTXO handler. Errors
955    ///   identifying needed inputs are logged and cut off this node, like TS.
956    /// - Unproven raw_tx parse failures retain the legacy graceful fallback.
957    async fn find_needed_inputs(
958        &self,
959        node: &GASPNode,
960    ) -> Result<Option<GASPNodeResponse>, GASPError> {
961        let mut requested_inputs: HashMap<String, GASPInputRequest> = HashMap::new();
962        // A proven node whose own output the dry run admits (lane E1D).
963        let mut admitted = false;
964
965        if let Some(proof_hex) = &node.proof {
966            let Some(manager) = self.topic_manager else {
967                return Ok(None);
968            };
969            let mut tx = Transaction::from_hex(&node.raw_tx)
970                .map_err(|e| GASPError::Other(format!("Failed to parse raw_tx: {e}")))?;
971            tx.merkle_path = Some(
972                beef_limits::proof_from_hex(proof_hex)
973                    .map_err(|e| GASPError::Other(format!("Failed to parse proof: {e}")))?,
974            );
975
976            // TS passes { dryRun: true }: the node is the peer's and nothing
977            // is admitted here, so a manager that writes on admission must not.
978            // Utils.toArray(string) defaults to UTF-8, even for hex-like text.
979            // This call is outside the needed-input catch in the reference.
980            let admittance = manager
981                .identify_admissible_outputs(
982                    &tx,
983                    &[],
984                    node.tx_metadata.as_deref().map(str::as_bytes),
985                    SubmitMode::HistoricalTx,
986                    &TopicAdmittanceContext::DRY_RUN,
987                )
988                .await
989                .map_err(|e| GASPError::Other(e.to_string()))?;
990            // The reference stops here (D13). Lane E1D (bsv-low #575,
991            // zanaadu-v2 #365) asks the named inputs of an ADMITTED node too:
992            // a manager that admits a successor with no coin (the pf head
993            // manager over its own head state) ended the walk at the
994            // successor while its predecessor had not landed, the graph was
995            // the successor alone, and its finalize submit recorded it with
996            // no coin; the predecessor, landed on a later tick, stood unspent
997            // beside the tip for good. So the predecessor is asked FIRST: a
998            // named input that is neither held nor landed (below) is
999            // requested and lands before the node in the graph's order, a
1000            // finalize submit of it that does not land stops the graph, and
1001            // a named input the peer cannot serve prunes (D14). When every
1002            // named input is known the walk stops here, as the reference's.
1003            admitted = admittance.outputs_to_admit.contains(&node.output_index);
1004            // A node with no input to name stops here, the manager not
1005            // asked (the reference's call count for such a node).
1006            if admitted
1007                && !tx
1008                    .inputs
1009                    .iter()
1010                    .any(|input| input.get_source_txid().is_ok_and(|s| !s.is_empty()))
1011            {
1012                return Ok(None);
1013            }
1014
1015            // A serialization RED is unreachable: bsv-rs stops collecting ancestors at a merkle path.
1016            let needed_inputs = async {
1017                let beef = tx
1018                    .to_beef(false)
1019                    .map_err(|e| format!("Failed to serialize proven BEEF: {e}"))?;
1020                manager
1021                    .identify_needed_inputs(&beef, None)
1022                    .await
1023                    .map_err(|e| e.to_string())
1024            }
1025            .await;
1026            match needed_inputs {
1027                Ok(inputs) => {
1028                    for input in inputs {
1029                        requested_inputs
1030                            .insert(input.to_graph_id(), GASPInputRequest { metadata: false });
1031                    }
1032                }
1033                Err(e) => {
1034                    error!(
1035                        "An error occurred when identifying needed inputs for transaction: {}.{}: {e}",
1036                        tx.id(), node.output_index
1037                    );
1038                    return Ok(None);
1039                }
1040            }
1041        } else {
1042            // Unproven nodes always need all inputs, independently of the manager.
1043            let tx = match Transaction::from_hex(&node.raw_tx) {
1044                Ok(tx) => tx,
1045                Err(e) => {
1046                    warn!(
1047                        "Cannot parse raw_tx for find_needed_inputs (graph_id={}): {e}",
1048                        node.graph_id
1049                    );
1050                    return Ok(None);
1051                }
1052            };
1053            for input in &tx.inputs {
1054                let source_txid = input.get_source_txid().unwrap_or_default();
1055                if source_txid.is_empty() {
1056                    continue;
1057                }
1058                let outpoint = format!("{}.{}", source_txid, input.source_output_index);
1059                requested_inputs.insert(outpoint, GASPInputRequest { metadata: false });
1060            }
1061        }
1062
1063        if requested_inputs.is_empty() {
1064            return Ok(None);
1065        }
1066
1067        // Strip inputs that are already known in local storage.
1068        let mut to_remove = Vec::new();
1069        for outpoint in requested_inputs.keys() {
1070            if let Some((txid, oi)) = crate::gasp::parse_outpoint(outpoint) {
1071                match self
1072                    .storage
1073                    .find_output(&txid, oi, Some(&self.topic), None, false)
1074                    .await
1075                {
1076                    Ok(Some(_)) => {
1077                        to_remove.push(outpoint.clone());
1078                    }
1079                    Ok(None) => {}
1080                    Err(e) => {
1081                        debug!("Storage lookup failed for {outpoint}: {e}");
1082                    }
1083                }
1084            }
1085        }
1086        for key in &to_remove {
1087            requested_inputs.remove(key);
1088        }
1089
1090        // An admitted node's named inputs are requested only while their
1091        // landing is unknown (E1D): one whose transaction has an applied row
1092        // in the topic, or an output held there, has landed, and the walk
1093        // stops as the reference's. A read that faults keeps the input: one
1094        // round trip to the peer is the cheaper error. So does a transaction
1095        // past [`LANDED_READS_PER_NODE`] (the E1D lens fold, L3: a manager
1096        // names as many inputs as it likes, D14's decoys included, and each
1097        // costs up to two reads): it is requested, and a landed one the peer
1098        // serves is a dupe at its finalize submit.
1099        if admitted {
1100            let mut landed: HashSet<String> = HashSet::new();
1101            let mut unlanded: HashSet<String> = HashSet::new();
1102            let mut reads = 0usize;
1103            let mut outpoints: Vec<&String> = requested_inputs.keys().collect();
1104            outpoints.sort();
1105            for outpoint in outpoints {
1106                let Some((txid, _)) = crate::gasp::parse_outpoint(outpoint) else {
1107                    continue;
1108                };
1109                if landed.contains(&txid) || unlanded.contains(&txid) {
1110                    continue;
1111                }
1112                if reads + 2 > LANDED_READS_PER_NODE {
1113                    break;
1114                }
1115                reads += 1;
1116                let record = crate::types::AppliedTransaction {
1117                    txid: txid.clone(),
1118                    topic: self.topic.clone(),
1119                };
1120                let mut holds = matches!(
1121                    self.storage.does_applied_transaction_exist(&record).await,
1122                    Ok(true)
1123                );
1124                if !holds {
1125                    reads += 1;
1126                    holds = matches!(
1127                        self.storage.find_outputs_for_transaction(&txid, false).await,
1128                        Ok(outputs) if outputs.iter().any(|o| o.topic == self.topic)
1129                    );
1130                }
1131                if holds {
1132                    landed.insert(txid);
1133                } else {
1134                    unlanded.insert(txid);
1135                }
1136            }
1137            requested_inputs.retain(|outpoint, _| {
1138                crate::gasp::parse_outpoint(outpoint)
1139                    .is_none_or(|(txid, _)| !landed.contains(&txid))
1140            });
1141        }
1142
1143        if requested_inputs.is_empty() {
1144            return Ok(None);
1145        }
1146
1147        Ok(Some(GASPNodeResponse { requested_inputs }))
1148    }
1149
1150    /// Append a node to a temporary graph being constructed during sync.
1151    ///
1152    /// If `spent_by` is `None`, this is the root node and is keyed by `graph_id`.
1153    /// Otherwise, the node is keyed by its computed txid.outputIndex, and is
1154    /// registered as a child of the parent node identified by `spent_by`.
1155    async fn append_to_graph(
1156        &self,
1157        node: &GASPNode,
1158        spent_by: Option<&str>,
1159    ) -> Result<(), GASPError> {
1160        debug!(
1161            "Appending node to graph {}: tx={}..., oi={}, spent_by={:?}",
1162            node.graph_id,
1163            &node.raw_tx[..node.raw_tx.len().min(16)],
1164            node.output_index,
1165            spent_by,
1166        );
1167
1168        // Compute the key for this node.
1169        let node_key = if spent_by.is_none() {
1170            // Root node — keyed by graph_id
1171            node.graph_id.clone()
1172        } else {
1173            // Child node — keyed by txid.outputIndex
1174            match Transaction::from_hex(&node.raw_tx) {
1175                Ok(tx) => format!("{}.{}", tx.id(), node.output_index),
1176                Err(_) => {
1177                    // Fallback: use raw_tx prefix as key
1178                    format!(
1179                        "{}.{}",
1180                        &node.raw_tx[..node.raw_tx.len().min(64)],
1181                        node.output_index
1182                    )
1183                }
1184            }
1185        };
1186
1187        // The pending graphs change: nothing kept of an anchor check holds.
1188        self.take_anchored();
1189        let mut refs = self
1190            .pending_graphs
1191            .lock()
1192            .map_err(|e| GASPError::Other(format!("Lock poisoned: {e}")))?;
1193
1194        // Insert this node
1195        refs.insert(
1196            node_key.clone(),
1197            PendingNode {
1198                node: node.clone(),
1199                spent_by: spent_by.map(std::string::ToString::to_string),
1200                children: Vec::new(),
1201            },
1202        );
1203
1204        // If spent_by is set, register this node as a child of the parent.
1205        if let Some(parent_key) = spent_by {
1206            if let Some(parent) = refs.get_mut(parent_key) {
1207                parent.children.push(node_key);
1208            }
1209        }
1210
1211        Ok(())
1212    }
1213
1214    /// Check a completed graph before anything of it is finalized (bsv-low
1215    /// #551). The reference's `validateGraphAnchor`
1216    /// (`OverlayGASPStorage.ts` at f999e0c1a, lines 261 to 297), in its two
1217    /// steps. An `Err` is a discard: `GASPSync::complete_graph` drops the
1218    /// graph and NOTHING of it reaches the sink.
1219    ///
1220    /// **1. Bitcoin.** The ROOT node's BEEF (`GraphAssembly::beef_of`) is verified
1221    /// by the engine's own check, `verify_spv_like_the_reference`: the
1222    /// reference's `spvTx.verify(this.engine.chainTracker)`. With a tracker
1223    /// every merkle root in that BEEF is checked against it and every unproven
1224    /// input's script is executed; with none it is `'scripts only'` (roots
1225    /// accepted unchecked, scripts still run). One verifier, the same switch
1226    /// (`with_script_verification`) as `Engine::submit`.
1227    ///
1228    /// **2. Overlay, all or nothing.** The ordered BEEFs (ancestors first, root
1229    /// last) are replayed through the topic manager as the reference's
1230    /// `admitHistoricalBEEF` does: mode `historical-tx`, no off-chain values,
1231    /// `previous_coins` the inputs whose source outpoint is a coin (the
1232    /// engine's little-endian u32 input indices), and every output the manager
1233    /// admits joins the coins. If the ROOT's outpoint is not a coin at the
1234    /// end, the graph is refused. The set only GROWS, as in the reference at
1235    /// f999e0c1a: it reads `outputsToAdmit` alone, never `coinsToRetain`, and
1236    /// removes nothing ("a Set of all historical coins to retain (no need to
1237    /// remove them)"). The reference passes `{ dryRun: true }` and so does
1238    /// this replay (`TopicAdmittanceContext::DRY_RUN`): a manager that writes
1239    /// on admission writes nothing here.
1240    ///
1241    /// WITHOUT a topic manager step 2 does not run: there is nothing to
1242    /// replay, and `Engine::submit` refuses such a topic (`UnsupportedTopic`),
1243    /// so nothing can be admitted. `Engine::start_gasp_sync` always wires the
1244    /// manager of a topic it can admit.
1245    ///
1246    /// **Divergences by addition** (each with its why):
1247    ///
1248    /// - *Coins the storage already holds count.* `find_needed_inputs` strips
1249    ///   an input the local storage already holds (as the reference does), so
1250    ///   its node is not in the graph and the replay's own set never sees it.
1251    ///   The reference then refuses a root that extends a coin it already
1252    ///   holds: the next head of a head chain, synced one tick after the
1253    ///   last, would be discarded forever. A previous coin is therefore an
1254    ///   input in the set OR in the storage for this topic, the exact lookup
1255    ///   `Engine::submit` makes when finalize submits the same BEEF.
1256    /// - *Sources the storage already holds are merged for step 1.* For the
1257    ///   same reason an UNPROVEN node may lack an input's source transaction
1258    ///   (the reference throws in `getBEEFForNode` and discards). The stored
1259    ///   BEEF of that output is merged into the checked copy, so the spend is
1260    ///   executed against its real source and that source's own ancestry is
1261    ///   verified too. If the stored BEEF is itself incomplete the verifier
1262    ///   refuses, and the graph is discarded as in the reference. Only the
1263    ///   CHECKED copy is merged; the finalized bytes are untouched.
1264    /// - *Every node is an input of its parent* (`read_anchor_input`).
1265    /// - *A conflicting spend refuses the graph.* Two transactions of one
1266    ///   graph that spend the same outpoint cannot both be admitted by the
1267    ///   finalize submits (the first one consumes the coin), so a replay that
1268    ///   showed the coin to both would promise an admission finalize cannot
1269    ///   keep. Upstream added the same rule after f999e0c1a (`spentOutpoints`,
1270    ///   "Historical GASP graph contains a conflicting spend").
1271    /// - *A fault of the moment is not a verdict.* A chain tracker that could
1272    ///   not answer, or a storage read that failed, says nothing about the
1273    ///   graph: the error is `GASPError::AnchorUnavailable`, the graph is
1274    ///   still discarded, but `complete_graph` FAILS the UTXO so the cursor
1275    ///   gap guard asks for it again. The reference discards and moves on,
1276    ///   which here would lose the UTXO for good behind an advanced cursor.
1277    /// - *A manager error in the replay is "not now", not a refusal.* The
1278    ///   replay is where a manager is shown the coins, so it is where one
1279    ///   whose own state lags can say it cannot place a transaction YET. An
1280    ///   `Err` from it is `GASPError::AnchorUnavailable` too: the UTXO fails
1281    ///   and is asked for again. `Ok` with nothing admitted stays the final
1282    ///   refusal, and there the cursor moves, as in the reference (which
1283    ///   moves it past a refused and a failed graph alike, `GASP.ts` 389 to
1284    ///   392, 419 to 423, 559 to 572, `Engine.ts` 1600 to 1604). The cost: a
1285    ///   manager that errors FOREVER on one UTXO holds that peer's cursor
1286    ///   below it, and what lies above and is not yet held is walked again
1287    ///   on every tick.
1288    ///
1289    /// **Cost.** Step 1 runs the script of every input of every unproven
1290    /// transaction in the root's BEEF and asks the tracker once per merkle
1291    /// path in it (a proven root: one question, no script). The ancestors a
1292    /// manager named BEHIND a proven node are not in that BEEF: their own
1293    /// merkle paths are not checked, here or in the reference; they are bound
1294    /// to the verified root by txid instead. Step 2 is one manager call per
1295    /// transaction and one storage read per input that is not already a coin
1296    /// of the set, which is what the finalize submits cost again.
1297    ///
1298    /// Since bsv-low #592 the walk runs under the engine's walk budget
1299    /// (`DoorBudget`, the door's two limbs, estimated from the bytes before
1300    /// anything runs). A breach is no verdict on the graph, and nothing
1301    /// stands behind this check (no network accepted these bytes, and the
1302    /// finalize submits them `historical-tx-no-spv`), so a breach is never
1303    /// the replay's to judge (the E592 lens fold, H1: as built at `4060ffd`
1304    /// it went on, and a peer's root that spent a held coin with no
1305    /// signature was finalized). It is HELD: `GASPError::AnchorWalkHeld`,
1306    /// the graph not finalized, the UTXO failed (or, under a per-graph
1307    /// budget, its record kept with nothing pending), the cursor below it,
1308    /// counted (`GASPSync::anchor_walk_held`). The roots the walk reached are
1309    /// checked first, so a bad proof is still the refusal it is. The hold
1310    /// ends when the peer serves the root PROVEN: #555's resume re-asks an
1311    /// unproven root, a proven one restarts the walk (`root_proven`), and a
1312    /// proven root is trusted without a script run. So an honest 0-conf graph
1313    /// too big to walk lands after its block; a forgery never lands. What a
1314    /// forging peer costs: its own slot (the UTXO held below the cursor, its
1315    /// record re-asked each pass up to `DEFERRED_GRAPH_MAX_PASSES`), inside
1316    /// its per-peer budget, and, as it never yields a finalized graph or a
1317    /// cursor move on that UTXO, D16's yieldless bound quarantines it if it
1318    /// serves nothing else, and ONLY with a per-graph budget
1319    /// (`Engine::set_graph_budget`): the yieldless bound exists only there
1320    /// (the E592 delta lens, D1-L2). With NO per-graph budget (a library
1321    /// consumer that sets none) there is no record and no yieldless count:
1322    /// the held UTXO fails on every sync, the sync runs to its end and is a
1323    /// success under #302, the cursor stays below that UTXO for good, and
1324    /// the graph is fetched and its walk estimated again on every tick;
1325    /// nothing is admitted, and the cost is ours, the residual D15 states for
1326    /// `AnchorUnavailable`. The reference serialises this method behind
1327    /// `acquireAnchorValidationSlot` (at most 4 at once); here graphs are
1328    /// completed one at a time by one sync on one thread, so there is
1329    /// nothing to serialise.
1330    async fn validate_graph_anchor(&self, graph_id: &str) -> Result<(), GASPError> {
1331        let anchor = {
1332            let refs = self
1333                .pending_graphs
1334                .lock()
1335                .map_err(|e| GASPError::Other(format!("Lock poisoned: {e}")))?;
1336            Self::read_anchor_input(graph_id, &refs, self.strict_beef)?
1337        };
1338        let unavailable =
1339            |what: String| GASPError::AnchorUnavailable(format!("graph {graph_id}: {what}"));
1340        let refused = |why: String| GASPError::ValidationFailed(format!("graph {graph_id}: {why}"));
1341
1342        // ── 1. Bitcoin ──────────────────────────────────────────────────
1343        let mut root_beef = anchor.root_beef;
1344        if !anchor.absent_sources.is_empty() {
1345            let mut merged = beef_limits::parse_beef(&root_beef, &beef_limits::PEER_BEEF_LIMITS)
1346                .map_err(|e| refused(format!("root BEEF does not parse: {e}")))?;
1347            for (txid, output_index) in &anchor.absent_sources {
1348                let held = self
1349                    .storage
1350                    .find_output(txid, *output_index, Some(&self.topic), None, true)
1351                    .await
1352                    .map_err(|e| unavailable(format!("reading {txid}.{output_index}: {e}")))?;
1353                if let Some(stored) = held.and_then(|output| output.beef) {
1354                    if let Ok(stored) =
1355                        beef_limits::parse_beef(&stored, &beef_limits::STORED_BEEF_LIMITS)
1356                    {
1357                        merged.merge_beef(&stored);
1358                    }
1359                }
1360            }
1361            root_beef = merged.to_binary();
1362        }
1363        let tracker = self.chain_tracker.map(|inner| FaultNotingTracker {
1364            inner,
1365            faulted: AtomicBool::new(false),
1366        });
1367        let verdict = crate::engine::verify_spv_like_the_reference(
1368            tracker.as_ref().map(|t| t as &dyn ChainTracker),
1369            self.verify_scripts,
1370            &root_beef,
1371            &anchor.root_txid,
1372            self.walk_budget,
1373        )
1374        .await;
1375        // The checked copy has done its work (bsv-low #586: not held through
1376        // the replay).
1377        drop(root_beef);
1378        // A walk that could not run within the budget (bsv-low #592) is not
1379        // a verdict on the graph, and nothing else judges it: HELD until the
1380        // root is served proven (the E592 lens fold, H1; the doc above).
1381        let verdict = match verdict {
1382            Ok(Some(stop)) => {
1383                warn!(
1384                    "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)",
1385                    stop.subject_txid,
1386                    stop.limb.as_str(),
1387                    stop.at_txid,
1388                    stop.what
1389                );
1390                return Err(GASPError::AnchorWalkHeld(format!(
1391                    "graph {graph_id}: the script walk of {} could not run within the budget ({}: {})",
1392                    stop.subject_txid,
1393                    stop.limb.as_str(),
1394                    stop.what
1395                )));
1396            }
1397            Ok(None) => Ok(()),
1398            Err(e) => Err(e),
1399        };
1400        if let Err(e) = verdict {
1401            if tracker
1402                .as_ref()
1403                .is_some_and(|t| t.faulted.load(Ordering::Relaxed))
1404            {
1405                return Err(unavailable(format!("the chain tracker faulted: {e}")));
1406            }
1407            return Err(refused(format!(
1408                "The graph is not well-anchored according to the rules of Bitcoin: {e}"
1409            )));
1410        }
1411
1412        // ── 2. Overlay ──────────────────────────────────────────────────
1413        let Some(manager) = self.topic_manager else {
1414            self.keep_anchored(graph_id, anchor.order, anchor.assembly);
1415            return Ok(());
1416        };
1417        let mut coins: HashSet<String> = HashSet::new();
1418        let mut spent: HashSet<String> = HashSet::new();
1419        let mut replayed: HashSet<String> = HashSet::new();
1420        for key in &anchor.order {
1421            // One node's BEEF at a time, the bytes `finalize_graph` will
1422            // assemble, dropped when its turn is over (bsv-low #586).
1423            let beef = &anchor.assembly.beef_of(key);
1424            // The parse `Engine::submit` makes of the same bytes at finalize.
1425            let subject = beef_limits::parse_beef(beef, &beef_limits::PEER_BEEF_LIMITS)
1426                .map(|mut b| crate::subject::subject_txid_of(&mut b))
1427                .map_err(|e| refused(format!("a graph BEEF does not parse: {e}")))?;
1428            let tx = beef_limits::transaction_from_beef(
1429                beef,
1430                subject.as_deref(),
1431                &beef_limits::PEER_BEEF_LIMITS,
1432            )
1433            .map_err(|e| refused(format!("a graph BEEF does not parse: {e}")))?;
1434            let txid = tx.id();
1435            // Two outputs of one transaction are two nodes with one BEEF.
1436            if !replayed.insert(txid.clone()) {
1437                continue;
1438            }
1439            let mut previous_coins: Vec<u8> = Vec::new();
1440            for (input_index, input) in tx.inputs.iter().enumerate() {
1441                let source_txid = input.get_source_txid().unwrap_or_default();
1442                if source_txid.is_empty() {
1443                    continue;
1444                }
1445                let outpoint = format!("{}.{}", source_txid, input.source_output_index);
1446                if !spent.insert(outpoint.clone()) {
1447                    return Err(refused(format!(
1448                        "the graph contains a conflicting spend of {outpoint}"
1449                    )));
1450                }
1451                let is_coin = coins.contains(&outpoint)
1452                    || self
1453                        .storage
1454                        .find_output(
1455                            &source_txid,
1456                            input.source_output_index,
1457                            Some(&self.topic),
1458                            None,
1459                            false,
1460                        )
1461                        .await
1462                        .map_err(|e| unavailable(format!("reading {outpoint}: {e}")))?
1463                        .is_some();
1464                if is_coin {
1465                    previous_coins.extend_from_slice(&(input_index as u32).to_le_bytes());
1466                }
1467            }
1468            // TS passes { dryRun: true }: the replay only asks, the finalize
1469            // submit is the admission.
1470            // An `Err` is the manager's "not now" (a fault of the moment, see
1471            // the divergences above); `Ok` with nothing is its refusal.
1472            let admittance = manager
1473                .identify_admissible_outputs(
1474                    &tx,
1475                    &previous_coins,
1476                    None,
1477                    SubmitMode::HistoricalTx,
1478                    &TopicAdmittanceContext::DRY_RUN,
1479                )
1480                .await
1481                .map_err(|e| unavailable(format!("the topic manager failed on {txid}: {e}")))?;
1482            for output_index in admittance.outputs_to_admit {
1483                coins.insert(format!("{txid}.{output_index}"));
1484            }
1485        }
1486        if !coins.contains(graph_id) {
1487            return Err(refused(
1488                "This graph did not result in topical admittance of the root node. Rejecting."
1489                    .to_string(),
1490            ));
1491        }
1492        self.keep_anchored(graph_id, anchor.order, anchor.assembly);
1493        Ok(())
1494    }
1495
1496    /// Finalize a graph — convert accumulated nodes to ordered BEEFs and push
1497    /// them to the shared sink for later Engine submission.
1498    ///
1499    /// Computes ordered BEEF byte arrays (ancestors first, root last) from the
1500    /// temporary graph nodes and stores them in the `FinalizedGraphSink`. The
1501    /// Engine drains it (after the sync, or per completed UTXO under a
1502    /// per-peer budget) and submits each one with `HistoricalTxNoSpv` mode, which skips SPV because
1503    /// `validate_graph_anchor` has already done it (the reference's
1504    /// `finalizeGraph`, lines 317 to 333).
1505    async fn finalize_graph(&self, graph_id: &str) -> Result<(), GASPError> {
1506        let refs = self
1507            .pending_graphs
1508            .lock()
1509            .map_err(|e| GASPError::Other(format!("Lock poisoned: {e}")))?;
1510
1511        if !refs.contains_key(graph_id) {
1512            return Err(GASPError::Other(format!("No pending graph for {graph_id}")));
1513        }
1514
1515        let node_count = refs
1516            .values()
1517            .filter(|pn| pn.node.graph_id == graph_id)
1518            .count();
1519
1520        debug!("Finalizing graph {graph_id} ({node_count} nodes). Computing ordered BEEFs.");
1521
1522        // Compute ordered BEEFs for the graph.
1523        let beefs =
1524            Self::compute_ordered_beefs(graph_id, &refs, self.strict_beef, self.take_anchored())?;
1525
1526        // Push to shared sink for later Engine submission.
1527        drop(refs);
1528        self.finalized_sink
1529            .lock()
1530            .map_err(|e| GASPError::Other(format!("Lock poisoned: {e}")))?
1531            .push(FinalizedGraph {
1532                topic: self.topic.clone(),
1533                beefs,
1534            });
1535
1536        // Remove all nodes belonging to this graph from pending refs.
1537        let mut refs = self
1538            .pending_graphs
1539            .lock()
1540            .map_err(|e| GASPError::Other(format!("Lock poisoned: {e}")))?;
1541        let keys_to_remove: Vec<String> = refs
1542            .iter()
1543            .filter(|(_, pn)| pn.node.graph_id == graph_id)
1544            .map(|(k, _)| k.clone())
1545            .collect();
1546        for key in keys_to_remove {
1547            refs.remove(&key);
1548        }
1549
1550        Ok(())
1551    }
1552
1553    /// Discard a temporary graph that failed validation.
1554    async fn discard_graph(&self, graph_id: &str) -> Result<(), GASPError> {
1555        debug!("Discarding graph: {graph_id}");
1556        self.take_anchored();
1557        let mut refs = self
1558            .pending_graphs
1559            .lock()
1560            .map_err(|e| GASPError::Other(format!("Lock poisoned: {e}")))?;
1561        let keys_to_remove: Vec<String> = refs
1562            .iter()
1563            .filter(|(_, pn)| pn.node.graph_id == graph_id)
1564            .map(|(k, _)| k.clone())
1565            .collect();
1566        for key in keys_to_remove {
1567            refs.remove(&key);
1568        }
1569        Ok(())
1570    }
1571
1572    async fn read_fault_memos(
1573        &self,
1574        outpoints: Option<&[String]>,
1575    ) -> Result<crate::fault_memo::FaultMemoBatch, GASPError> {
1576        let batch = self
1577            .storage
1578            .read_fault_memos(&self.peer, &self.topic, outpoints)
1579            .await
1580            .map_err(|e| GASPError::StorageError(e.to_string()))?;
1581        let mut cache = self.fault_memos.lock().unwrap();
1582        // Only the active batch is cached. A walk spanning many pages does not
1583        // keep the complete memo in isolate memory.
1584        cache.clear();
1585        for record in &batch.records {
1586            cache.insert(record.outpoint.clone(), record.clone().into_record());
1587        }
1588        Ok(batch)
1589    }
1590
1591    async fn load_deferred_graphs(&self) -> Result<Vec<crate::gasp::DeferredGraphKey>, GASPError> {
1592        self.storage
1593            .find_deferred_graphs(&self.peer, &self.topic)
1594            .await
1595            .map_err(|e| GASPError::StorageError(e.to_string()))
1596    }
1597
1598    async fn get_deferred_graph(
1599        &self,
1600        outpoint: &str,
1601    ) -> Result<Option<crate::gasp::DeferredGraph>, GASPError> {
1602        if let Some(record) = self.fault_memos.lock().unwrap().get(outpoint).cloned() {
1603            return Ok(Some(record));
1604        }
1605        self.storage
1606            .get_deferred_graph(&self.peer, &self.topic, outpoint)
1607            .await
1608            .map_err(|e| GASPError::StorageError(e.to_string()))
1609    }
1610
1611    async fn save_deferred_graph(
1612        &self,
1613        record: &crate::gasp::DeferredGraph,
1614    ) -> Result<crate::gasp::DeferredGraphSave, GASPError> {
1615        let mut record = record.clone();
1616        record.peer.clone_from(&self.peer);
1617        record.topic.clone_from(&self.topic);
1618        record.configured = self.configured_peer;
1619        let result = self
1620            .storage
1621            .put_deferred_graph(&record)
1622            .await
1623            .map_err(|e| GASPError::StorageError(e.to_string()))?;
1624        if result == crate::gasp::DeferredGraphSave::Saved {
1625            let mut cache = self.fault_memos.lock().unwrap();
1626            if record.is_struck() {
1627                cache.insert(record.outpoint.clone(), record);
1628            } else {
1629                cache.remove(&record.outpoint);
1630            }
1631        }
1632        Ok(result)
1633    }
1634
1635    async fn reserve_unanchorable_retry(
1636        &self,
1637        record: &crate::gasp::DeferredGraph,
1638    ) -> Result<bool, GASPError> {
1639        let mut record = record.clone();
1640        record.peer.clone_from(&self.peer);
1641        record.topic.clone_from(&self.topic);
1642        record.configured = self.configured_peer;
1643        let reserved = self
1644            .storage
1645            .reserve_unanchorable_retry(&record)
1646            .await
1647            .map_err(|e| GASPError::StorageError(e.to_string()))?;
1648        self.fault_memos.lock().unwrap().remove(&record.outpoint);
1649        Ok(reserved)
1650    }
1651
1652    async fn delete_deferred_graph(&self, outpoint: &str) -> Result<(), GASPError> {
1653        self.fault_memos.lock().unwrap().remove(outpoint);
1654        self.storage
1655            .delete_deferred_graph(&self.peer, &self.topic, outpoint)
1656            .await
1657            .map_err(|e| GASPError::StorageError(e.to_string()))
1658    }
1659
1660    // Keyed by topic, not peer: the expiry is read from the outpoint's own
1661    // bytes, so any peer that serves it serves the same dead advert.
1662    async fn remember_expired(&self, outpoint: &str, expiry: u64) -> Result<(), GASPError> {
1663        self.storage
1664            .remember_expired_outpoint(&self.topic, outpoint, expiry)
1665            .await
1666            .map_err(|e| GASPError::StorageError(e.to_string()))
1667    }
1668
1669    async fn remembered_expired(&self, outpoints: &[String]) -> Result<Vec<String>, GASPError> {
1670        self.storage
1671            .find_expired_outpoints(&self.topic, outpoints)
1672            .await
1673            .map_err(|e| GASPError::StorageError(e.to_string()))
1674    }
1675}
1676
1677// ============================================================================
1678// Tests
1679// ============================================================================
1680
1681#[cfg(test)]
1682mod tests {
1683    use super::*;
1684    use crate::gasp::GASPStorage;
1685    use crate::storage::memory::MemoryStorage;
1686    use crate::storage::Storage;
1687    use crate::types::Output;
1688
1689    fn make_sink() -> FinalizedGraphSink {
1690        new_finalized_graph_sink()
1691    }
1692
1693    fn make_output(txid: &str, index: u32, topic: &str, score: f64) -> Output {
1694        Output {
1695            txid: txid.to_string(),
1696            output_index: index,
1697            output_script: vec![0x76, 0xa9],
1698            satoshis: 1000,
1699            topic: topic.to_string(),
1700            spent: false,
1701            outputs_consumed: vec![],
1702            consumed_by: vec![],
1703            beef: Some(vec![0xBE, 0xEF]),
1704            block_height: None,
1705            score: Some(score),
1706        }
1707    }
1708
1709    /// Build a minimal valid transaction hex with one input and one output.
1710    fn make_valid_tx_hex(source_txid: &str, source_oi: u32) -> String {
1711        let mut tx = Transaction::new();
1712        let input = bsv_rs::transaction::TransactionInput::new(source_txid.to_string(), source_oi);
1713        tx.inputs.push(input);
1714        tx.outputs.push(bsv_rs::transaction::TransactionOutput::new(
1715            100,
1716            bsv_rs::script::LockingScript::from_hex(
1717                "76a914000000000000000000000000000000000000000088ac",
1718            )
1719            .unwrap(),
1720        ));
1721        tx.to_hex()
1722    }
1723
1724    /// Build a coinbase-like tx (no meaningful inputs).
1725    /// A raw transaction with no input, read by `Transaction::from_hex`
1726    /// alone (no BEEF door, which refuses it since bsv-rs 0.4.1).
1727    fn make_no_input_tx_hex() -> String {
1728        let mut tx = Transaction::new();
1729        tx.outputs.push(bsv_rs::transaction::TransactionOutput::new(
1730            5000,
1731            bsv_rs::script::LockingScript::from_hex(
1732                "76a914000000000000000000000000000000000000000088ac",
1733            )
1734            .unwrap(),
1735        ));
1736        tx.to_hex()
1737    }
1738
1739    fn make_coinbase_tx_hex() -> String {
1740        let mut tx = Transaction::new();
1741        // A coinbase's null outpoint: a transaction with no input is refused
1742        // since bsv-rs 0.4.1 (NL-8 W6).
1743        let mut input = bsv_rs::transaction::TransactionInput::new("00".repeat(32), u32::MAX);
1744        input.unlocking_script =
1745            Some(bsv_rs::script::UnlockingScript::from_binary(&[0x51]).unwrap());
1746        tx.inputs.push(input);
1747        tx.outputs.push(bsv_rs::transaction::TransactionOutput::new(
1748            5000,
1749            bsv_rs::script::LockingScript::from_hex(
1750                "76a914000000000000000000000000000000000000000088ac",
1751            )
1752            .unwrap(),
1753        ));
1754        tx.to_hex()
1755    }
1756
1757    // ── find_known_utxos tests ────────────────────────────────────────
1758
1759    #[tokio::test]
1760    async fn find_known_utxos_returns_correct_outputs() {
1761        let store = MemoryStorage::new();
1762        store
1763            .insert_output(&make_output("tx1", 0, "tm_test", 100.0))
1764            .await
1765            .unwrap();
1766        store
1767            .insert_output(&make_output("tx2", 0, "tm_test", 200.0))
1768            .await
1769            .unwrap();
1770        store
1771            .insert_output(&make_output("tx3", 0, "tm_other", 300.0))
1772            .await
1773            .unwrap();
1774
1775        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
1776
1777        let utxos = gasp_storage.find_known_utxos(0, None).await.unwrap();
1778        assert_eq!(utxos.len(), 2, "Should return only tm_test UTXOs");
1779        assert_eq!(utxos[0].txid, "tx1");
1780        assert_eq!(utxos[1].txid, "tx2");
1781    }
1782
1783    #[tokio::test]
1784    async fn find_known_utxos_respects_since() {
1785        let store = MemoryStorage::new();
1786        store
1787            .insert_output(&make_output("tx1", 0, "tm_test", 100.0))
1788            .await
1789            .unwrap();
1790        store
1791            .insert_output(&make_output("tx2", 0, "tm_test", 200.0))
1792            .await
1793            .unwrap();
1794        store
1795            .insert_output(&make_output("tx3", 0, "tm_test", 300.0))
1796            .await
1797            .unwrap();
1798
1799        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
1800
1801        let utxos = gasp_storage.find_known_utxos(200, None).await.unwrap();
1802        assert_eq!(utxos.len(), 2, "Should return UTXOs with score >= 200");
1803        assert_eq!(utxos[0].txid, "tx2");
1804        assert_eq!(utxos[1].txid, "tx3");
1805    }
1806
1807    #[tokio::test]
1808    async fn find_known_utxos_respects_limit() {
1809        let store = MemoryStorage::new();
1810        store
1811            .insert_output(&make_output("tx1", 0, "tm_test", 100.0))
1812            .await
1813            .unwrap();
1814        store
1815            .insert_output(&make_output("tx2", 0, "tm_test", 200.0))
1816            .await
1817            .unwrap();
1818        store
1819            .insert_output(&make_output("tx3", 0, "tm_test", 300.0))
1820            .await
1821            .unwrap();
1822
1823        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
1824
1825        let utxos = gasp_storage.find_known_utxos(0, Some(2)).await.unwrap();
1826        assert_eq!(utxos.len(), 2, "Should limit to 2 results");
1827    }
1828
1829    #[tokio::test]
1830    async fn find_known_utxos_excludes_spent() {
1831        let store = MemoryStorage::new();
1832        store
1833            .insert_output(&make_output("tx1", 0, "tm_test", 100.0))
1834            .await
1835            .unwrap();
1836        store
1837            .insert_output(&make_output("tx2", 0, "tm_test", 200.0))
1838            .await
1839            .unwrap();
1840        store.mark_utxo_as_spent("tx1", 0, "tm_test").await.unwrap();
1841
1842        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
1843
1844        let utxos = gasp_storage.find_known_utxos(0, None).await.unwrap();
1845        assert_eq!(utxos.len(), 1, "Should exclude spent UTXOs");
1846        assert_eq!(utxos[0].txid, "tx2");
1847    }
1848
1849    #[tokio::test]
1850    async fn find_known_utxos_empty_topic() {
1851        let store = MemoryStorage::new();
1852        let gasp_storage = OverlayGASPStorage::new(&store, "tm_empty", make_sink());
1853
1854        let utxos = gasp_storage.find_known_utxos(0, None).await.unwrap();
1855        assert!(utxos.is_empty());
1856    }
1857
1858    // ── find_needed_inputs tests ──────────────────────────────────────
1859
1860    #[tokio::test]
1861    async fn find_needed_inputs_returns_none_for_unparsable_tx() {
1862        let store = MemoryStorage::new();
1863        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
1864
1865        let node = GASPNode {
1866            graph_id: "abc.0".to_string(),
1867            raw_tx: "deadbeef".to_string(), // not valid tx hex
1868            output_index: 0,
1869            proof: None,
1870            tx_metadata: None,
1871            output_metadata: None,
1872            inputs: None,
1873        };
1874
1875        let result = gasp_storage.find_needed_inputs(&node).await.unwrap();
1876        assert!(result.is_none(), "Unparsable tx: graceful fallback to None");
1877    }
1878
1879    #[tokio::test]
1880    async fn find_needed_inputs_returns_none_for_proved_tx() {
1881        let store = MemoryStorage::new();
1882        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
1883
1884        // A node with a proof (merkle path) should not need inputs.
1885        let raw_tx = make_valid_tx_hex(
1886            "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
1887            0,
1888        );
1889        let node = GASPNode {
1890            graph_id: "abc.0".to_string(),
1891            raw_tx,
1892            output_index: 0,
1893            proof: Some("some_proof_hex".to_string()),
1894            tx_metadata: None,
1895            output_metadata: None,
1896            inputs: None,
1897        };
1898
1899        let result = gasp_storage.find_needed_inputs(&node).await.unwrap();
1900        assert!(result.is_none(), "Proved tx should not need inputs");
1901    }
1902
1903    #[tokio::test]
1904    async fn find_needed_inputs_requests_inputs_for_unproved_tx() {
1905        let store = MemoryStorage::new();
1906        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
1907
1908        let source_txid = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
1909        let raw_tx = make_valid_tx_hex(source_txid, 0);
1910
1911        let node = GASPNode {
1912            graph_id: "abc.0".to_string(),
1913            raw_tx,
1914            output_index: 0,
1915            proof: None,
1916            tx_metadata: None,
1917            output_metadata: None,
1918            inputs: None,
1919        };
1920
1921        let result = gasp_storage.find_needed_inputs(&node).await.unwrap();
1922        assert!(result.is_some(), "Unproved tx should request inputs");
1923        let response = result.unwrap();
1924        assert_eq!(response.requested_inputs.len(), 1);
1925        let key = format!("{source_txid}.0");
1926        assert!(
1927            response.requested_inputs.contains_key(&key),
1928            "Should request the source input"
1929        );
1930    }
1931
1932    #[tokio::test]
1933    async fn find_needed_inputs_strips_known_inputs() {
1934        let store = MemoryStorage::new();
1935        let source_txid = "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc";
1936
1937        // Insert the source output so it's already known
1938        store
1939            .insert_output(&make_output(source_txid, 0, "tm_test", 50.0))
1940            .await
1941            .unwrap();
1942
1943        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
1944
1945        let raw_tx = make_valid_tx_hex(source_txid, 0);
1946        let node = GASPNode {
1947            graph_id: "abc.0".to_string(),
1948            raw_tx,
1949            output_index: 0,
1950            proof: None,
1951            tx_metadata: None,
1952            output_metadata: None,
1953            inputs: None,
1954        };
1955
1956        let result = gasp_storage.find_needed_inputs(&node).await.unwrap();
1957        assert!(
1958            result.is_none(),
1959            "Should strip already-known inputs, resulting in None"
1960        );
1961    }
1962
1963    #[tokio::test]
1964    async fn find_needed_inputs_returns_none_for_coinbase() {
1965        let store = MemoryStorage::new();
1966        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
1967
1968        let raw_tx = make_no_input_tx_hex();
1969        let node = GASPNode {
1970            graph_id: "abc.0".to_string(),
1971            raw_tx,
1972            output_index: 0,
1973            proof: None,
1974            tx_metadata: None,
1975            output_metadata: None,
1976            inputs: None,
1977        };
1978
1979        let result = gasp_storage.find_needed_inputs(&node).await.unwrap();
1980        assert!(
1981            result.is_none(),
1982            "Coinbase-like tx (no inputs) should need nothing"
1983        );
1984    }
1985
1986    // ── append / finalize / discard tests ─────────────────────────────
1987
1988    #[tokio::test]
1989    async fn append_and_finalize_graph() {
1990        let store = MemoryStorage::new();
1991        let sink = make_sink();
1992        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", sink.clone());
1993
1994        let raw_tx = make_coinbase_tx_hex();
1995        let node = GASPNode {
1996            graph_id: "abc.0".to_string(),
1997            raw_tx,
1998            output_index: 0,
1999            proof: None,
2000            tx_metadata: None,
2001            output_metadata: None,
2002            inputs: None,
2003        };
2004
2005        gasp_storage.append_to_graph(&node, None).await.unwrap();
2006        assert_eq!(gasp_storage.pending_graph_count(), 1);
2007
2008        gasp_storage.finalize_graph("abc.0").await.unwrap();
2009        assert_eq!(gasp_storage.pending_graph_count(), 0);
2010
2011        // Check that finalized data was produced
2012        let finalized = sink.lock().unwrap();
2013        assert_eq!(finalized.len(), 1, "Should have one finalized graph");
2014        assert_eq!(finalized[0].topic, "tm_test");
2015        assert!(!finalized[0].beefs.is_empty(), "Should have BEEF data");
2016    }
2017
2018    #[tokio::test]
2019    async fn finalize_graph_produces_submittable_beef() {
2020        let store = MemoryStorage::new();
2021        let sink = make_sink();
2022        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", sink.clone());
2023
2024        let raw_tx = make_coinbase_tx_hex();
2025        let node = GASPNode {
2026            graph_id: "root.0".to_string(),
2027            raw_tx,
2028            output_index: 0,
2029            proof: None,
2030            tx_metadata: None,
2031            output_metadata: None,
2032            inputs: None,
2033        };
2034
2035        gasp_storage.append_to_graph(&node, None).await.unwrap();
2036        gasp_storage.finalize_graph("root.0").await.unwrap();
2037
2038        let finalized = sink.lock().unwrap();
2039        assert_eq!(finalized.len(), 1);
2040
2041        // Each BEEF should be parseable
2042        for beef_bytes in &finalized[0].beefs {
2043            let tx = Transaction::from_beef(beef_bytes, None);
2044            assert!(
2045                tx.is_ok(),
2046                "Finalized BEEF should be parseable: {:?}",
2047                tx.err()
2048            );
2049        }
2050    }
2051
2052    #[tokio::test]
2053    async fn append_multiple_nodes_same_graph() {
2054        let store = MemoryStorage::new();
2055        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
2056
2057        let raw_tx1 = make_coinbase_tx_hex();
2058        let node1 = GASPNode {
2059            graph_id: "abc.0".to_string(),
2060            raw_tx: raw_tx1,
2061            output_index: 0,
2062            proof: None,
2063            tx_metadata: None,
2064            output_metadata: None,
2065            inputs: None,
2066        };
2067        let raw_tx2 = make_coinbase_tx_hex();
2068        let node2 = GASPNode {
2069            graph_id: "abc.0".to_string(),
2070            raw_tx: raw_tx2,
2071            output_index: 0,
2072            proof: None,
2073            tx_metadata: None,
2074            output_metadata: None,
2075            inputs: None,
2076        };
2077
2078        gasp_storage.append_to_graph(&node1, None).await.unwrap();
2079        gasp_storage
2080            .append_to_graph(&node2, Some("abc.0"))
2081            .await
2082            .unwrap();
2083        assert_eq!(gasp_storage.pending_graph_count(), 1);
2084    }
2085
2086    #[tokio::test]
2087    async fn discard_graph_removes_pending() {
2088        let store = MemoryStorage::new();
2089        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
2090
2091        let raw_tx = make_coinbase_tx_hex();
2092        let node = GASPNode {
2093            graph_id: "abc.0".to_string(),
2094            raw_tx,
2095            output_index: 0,
2096            proof: None,
2097            tx_metadata: None,
2098            output_metadata: None,
2099            inputs: None,
2100        };
2101
2102        gasp_storage.append_to_graph(&node, None).await.unwrap();
2103        assert_eq!(gasp_storage.pending_graph_count(), 1);
2104
2105        gasp_storage.discard_graph("abc.0").await.unwrap();
2106        assert_eq!(gasp_storage.pending_graph_count(), 0);
2107    }
2108
2109    #[tokio::test]
2110    async fn validate_graph_anchor_refuses_an_unknown_graph() {
2111        let store = MemoryStorage::new();
2112        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
2113
2114        // The reference throws `Graph node with ID ... not found`.
2115        let error = gasp_storage
2116            .validate_graph_anchor("nonexistent.0")
2117            .await
2118            .unwrap_err();
2119        assert!(matches!(error, GASPError::ValidationFailed(_)), "{error}");
2120    }
2121
2122    #[tokio::test]
2123    async fn finalize_empty_graph_errors() {
2124        let store = MemoryStorage::new();
2125        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
2126
2127        let result = gasp_storage.finalize_graph("nonexistent.0").await;
2128        assert!(result.is_err(), "Should error on unknown graph");
2129    }
2130
2131    #[tokio::test]
2132    async fn gasp_storage_is_object_safe() {
2133        let store = MemoryStorage::new();
2134        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
2135
2136        // Verify it can be used as dyn GASPStorage
2137        let boxed: Box<dyn GASPStorage> = Box::new(gasp_storage);
2138        let utxos = boxed.find_known_utxos(0, None).await.unwrap();
2139        assert!(utxos.is_empty());
2140    }
2141
2142    /// Integration test: OverlayGASPStorage works with GASPSync.
2143    #[tokio::test]
2144    async fn overlay_gasp_storage_with_gasp_sync() {
2145        use crate::gasp::{GASPRemote, GASPSync};
2146        use crate::types::{GASPInitialReply, GASPInitialRequest, GASPInitialResponse};
2147
2148        // Set up local storage with one UTXO
2149        let store = MemoryStorage::new();
2150        store
2151            .insert_output(&make_output("local_tx", 0, "tm_test", 50.0))
2152            .await
2153            .unwrap();
2154
2155        let sink = make_sink();
2156        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", sink.clone());
2157
2158        // Mock remote with two UTXOs — provides unparsable raw_tx so
2159        // find_needed_inputs returns None (graceful fallback).
2160        struct TestRemote;
2161
2162        #[async_trait(?Send)]
2163        impl GASPRemote for TestRemote {
2164            async fn get_initial_response(
2165                &self,
2166                request: &GASPInitialRequest,
2167            ) -> Result<GASPInitialResponse, GASPError> {
2168                Ok(GASPInitialResponse {
2169                    utxo_list: vec![
2170                        GASPOutput {
2171                            txid: "remote_tx1".to_string(),
2172                            output_index: 0,
2173                            score: 100.0,
2174                        },
2175                        GASPOutput {
2176                            txid: "remote_tx2".to_string(),
2177                            output_index: 0,
2178                            score: 200.0,
2179                        },
2180                    ],
2181                    since: request.since,
2182                })
2183            }
2184            async fn get_initial_reply(
2185                &self,
2186                _: &GASPInitialResponse,
2187            ) -> Result<GASPInitialReply, GASPError> {
2188                Ok(GASPInitialReply {
2189                    utxo_list: Vec::new(),
2190                })
2191            }
2192            async fn request_node(
2193                &self,
2194                graph_id: &str,
2195                txid: &str,
2196                output_index: u32,
2197                _: bool,
2198            ) -> Result<GASPNode, GASPError> {
2199                Ok(GASPNode {
2200                    graph_id: graph_id.to_string(),
2201                    raw_tx: format!("rawtx_{txid}"),
2202                    output_index,
2203                    proof: None,
2204                    tx_metadata: None,
2205                    output_metadata: None,
2206                    inputs: None,
2207                })
2208            }
2209            async fn submit_node(
2210                &self,
2211                _: &GASPNode,
2212            ) -> Result<Option<GASPNodeResponse>, GASPError> {
2213                Ok(None)
2214            }
2215        }
2216
2217        let mut sync = GASPSync::new(
2218            Box::new(gasp_storage),
2219            Box::new(TestRemote),
2220            0,
2221            "[TEST]",
2222            true, // unidirectional
2223        );
2224
2225        sync.sync(None).await.unwrap();
2226        // The synthetic graph cannot assemble a BEEF. Nothing completed,
2227        // so the cursor remains at entry and both outputs are retried.
2228        assert_eq!(sync.last_interaction, 0);
2229    }
2230
2231    /// Integration test: full sync with valid transactions produces submittable BEEFs.
2232    #[tokio::test]
2233    async fn gasp_sync_with_valid_tx_produces_finalized_beefs() {
2234        use crate::gasp::{GASPRemote, GASPSync};
2235        use crate::types::{GASPInitialReply, GASPInitialRequest, GASPInitialResponse};
2236
2237        let store = MemoryStorage::new();
2238        let sink = make_sink();
2239        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", sink.clone());
2240
2241        // Build a valid coinbase-style tx for the remote to provide.
2242        let valid_tx_hex = make_coinbase_tx_hex();
2243
2244        struct ValidTxRemote {
2245            tx_hex: String,
2246        }
2247
2248        #[async_trait(?Send)]
2249        impl GASPRemote for ValidTxRemote {
2250            async fn get_initial_response(
2251                &self,
2252                request: &GASPInitialRequest,
2253            ) -> Result<GASPInitialResponse, GASPError> {
2254                Ok(GASPInitialResponse {
2255                    utxo_list: vec![GASPOutput {
2256                        txid: "valid_tx".to_string(),
2257                        output_index: 0,
2258                        score: 100.0,
2259                    }],
2260                    since: request.since,
2261                })
2262            }
2263            async fn get_initial_reply(
2264                &self,
2265                _: &GASPInitialResponse,
2266            ) -> Result<GASPInitialReply, GASPError> {
2267                Ok(GASPInitialReply {
2268                    utxo_list: Vec::new(),
2269                })
2270            }
2271            async fn request_node(
2272                &self,
2273                graph_id: &str,
2274                _txid: &str,
2275                output_index: u32,
2276                _: bool,
2277            ) -> Result<GASPNode, GASPError> {
2278                // Served PROVEN: since bsv-low #551 the anchor check refuses
2279                // an unproven transaction that creates satoshis (this one has
2280                // no inputs). With no chain tracker a merkle path is accepted
2281                // unchecked, the reference's 'scripts only'.
2282                let txid = Transaction::from_hex(&self.tx_hex).unwrap().id();
2283                let proof = MerklePath::new(
2284                    100,
2285                    vec![vec![bsv_rs::transaction::MerklePathLeaf::new_txid(0, txid)]],
2286                )
2287                .unwrap();
2288                Ok(GASPNode {
2289                    graph_id: graph_id.to_string(),
2290                    raw_tx: self.tx_hex.clone(),
2291                    output_index,
2292                    proof: Some(proof.to_hex()),
2293                    tx_metadata: None,
2294                    output_metadata: None,
2295                    inputs: None,
2296                })
2297            }
2298            async fn submit_node(
2299                &self,
2300                _: &GASPNode,
2301            ) -> Result<Option<GASPNodeResponse>, GASPError> {
2302                Ok(None)
2303            }
2304        }
2305
2306        let remote = ValidTxRemote {
2307            tx_hex: valid_tx_hex,
2308        };
2309
2310        let mut sync = GASPSync::new(Box::new(gasp_storage), Box::new(remote), 0, "[TEST]", true);
2311
2312        sync.sync(None).await.unwrap();
2313
2314        let finalized = sink.lock().unwrap();
2315        assert_eq!(finalized.len(), 1, "Should have one finalized graph");
2316        assert!(!finalized[0].beefs.is_empty(), "Should have BEEF bytes");
2317
2318        // Verify the BEEF is parseable
2319        for beef_bytes in &finalized[0].beefs {
2320            let parsed = Transaction::from_beef(beef_bytes, None);
2321            assert!(parsed.is_ok(), "BEEF should be parseable");
2322        }
2323    }
2324
2325    #[tokio::test]
2326    async fn take_finalized_graphs_drains_list() {
2327        let store = MemoryStorage::new();
2328        let sink = make_sink();
2329        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", sink.clone());
2330
2331        let raw_tx = make_coinbase_tx_hex();
2332        let node = GASPNode {
2333            graph_id: "abc.0".to_string(),
2334            raw_tx,
2335            output_index: 0,
2336            proof: None,
2337            tx_metadata: None,
2338            output_metadata: None,
2339            inputs: None,
2340        };
2341
2342        gasp_storage.append_to_graph(&node, None).await.unwrap();
2343        gasp_storage.finalize_graph("abc.0").await.unwrap();
2344
2345        let taken = gasp_storage.take_finalized_graphs();
2346        assert_eq!(taken.len(), 1);
2347
2348        // Should be empty after draining
2349        let taken2 = gasp_storage.take_finalized_graphs();
2350        assert!(taken2.is_empty());
2351    }
2352
2353    // ── bsv-low #586: the assembly gives the bytes the hydrated tree gave ──
2354
2355    /// `get_beef_for_node` as it was on `fbb7fa8`, verbatim: the hydrating
2356    /// recursion [`GraphAssembly`] replaced. Kept as the reference the
2357    /// assembly is compared with, byte for byte and error for error.
2358    fn legacy_get_beef_for_node(
2359        node_key: &str,
2360        refs: &HashMap<String, PendingNode>,
2361        strict_beef: bool,
2362    ) -> Result<(Transaction, Vec<u8>), GASPError> {
2363        let pending = refs
2364            .get(node_key)
2365            .ok_or_else(|| GASPError::Other(format!("Node {node_key} not found in graph refs")))?;
2366
2367        let mut tx = Transaction::from_hex(&pending.node.raw_tx)
2368            .map_err(|e| GASPError::Other(format!("Failed to parse raw_tx for {node_key}: {e}")))?;
2369
2370        if let Some(ref proof_hex) = pending.node.proof {
2371            tx.merkle_path = Some(beef_limits::proof_from_hex(proof_hex).map_err(|e| {
2372                GASPError::Other(format!("Failed to parse proof for {node_key}: {e}"))
2373            })?);
2374        } else {
2375            let child_info: Vec<(usize, String)> = tx
2376                .inputs
2377                .iter()
2378                .enumerate()
2379                .filter_map(|(idx, input)| {
2380                    let source_txid = input.get_source_txid().unwrap_or_default();
2381                    if source_txid.is_empty() {
2382                        return None;
2383                    }
2384                    let child_key = format!("{}.{}", source_txid, input.source_output_index);
2385                    if refs.contains_key(&child_key) {
2386                        Some((idx, child_key))
2387                    } else {
2388                        None
2389                    }
2390                })
2391                .collect();
2392
2393            for (input_idx, child_key) in child_info {
2394                let (child_tx, _) = legacy_get_beef_for_node(&child_key, refs, strict_beef)?;
2395                tx.inputs[input_idx].source_transaction = Some(Box::new(child_tx));
2396            }
2397        }
2398
2399        let allow_partial = !strict_beef;
2400        let beef = tx.to_beef(allow_partial).map_err(|e| {
2401            GASPError::Other(format!("Failed to serialize BEEF for {node_key}: {e}"))
2402        })?;
2403
2404        Ok((tx, beef))
2405    }
2406
2407    /// `compute_ordered_beefs` as it was on `fbb7fa8`, verbatim.
2408    fn legacy_compute_ordered_beefs(
2409        graph_id: &str,
2410        refs: &HashMap<String, PendingNode>,
2411        strict_beef: bool,
2412    ) -> Result<Vec<Vec<u8>>, GASPError> {
2413        let mut beefs: Vec<Vec<u8>> = Vec::new();
2414        let mut visited: std::collections::HashSet<String> = std::collections::HashSet::new();
2415
2416        fn hydrate(
2417            node_key: &str,
2418            refs: &HashMap<String, PendingNode>,
2419            beefs: &mut Vec<Vec<u8>>,
2420            visited: &mut std::collections::HashSet<String>,
2421            strict_beef: bool,
2422        ) -> Result<(), GASPError> {
2423            if visited.contains(node_key) {
2424                return Ok(());
2425            }
2426            visited.insert(node_key.to_string());
2427
2428            let Some(pending) = refs.get(node_key) else {
2429                return Ok(());
2430            };
2431
2432            for child_key in &pending.children {
2433                hydrate(child_key, refs, beefs, visited, strict_beef)?;
2434            }
2435
2436            let (_, beef) = legacy_get_beef_for_node(node_key, refs, strict_beef)?;
2437            beefs.push(beef);
2438            Ok(())
2439        }
2440
2441        hydrate(graph_id, refs, &mut beefs, &mut visited, strict_beef)?;
2442        Ok(beefs)
2443    }
2444
2445    /// A small deterministic generator (no dependency, the same graphs on
2446    /// every run).
2447    struct Lcg(u64);
2448
2449    impl Lcg {
2450        fn below(&mut self, n: u64) -> u64 {
2451            self.0 = self
2452                .0
2453                .wrapping_mul(6_364_136_223_846_793_005)
2454                .wrapping_add(1_442_695_040_888_963_407);
2455            (self.0 >> 33) % n
2456        }
2457    }
2458
2459    fn pending(raw_tx: String, output_index: u32, proof: Option<String>) -> PendingNode {
2460        PendingNode {
2461            node: GASPNode {
2462                graph_id: "g".to_string(),
2463                raw_tx,
2464                output_index,
2465                proof,
2466                tx_metadata: None,
2467                output_metadata: None,
2468                inputs: None,
2469            },
2470            spent_by: None,
2471            children: Vec::new(),
2472        }
2473    }
2474
2475    /// A random pending graph: up to 9 transactions, each spending up to
2476    /// four outputs of earlier ones (two outputs of one transaction too, the
2477    /// diamond), some proven (two of a pair sharing one block, so their
2478    /// proofs COMBINE), some referenced outpoints absent from the graph (a
2479    /// partial BEEF, a strict refusal), and now and then a proof or a raw
2480    /// transaction that does not parse. Returns the graph and its root key.
2481    fn random_graph(rng: &mut Lcg) -> (HashMap<String, PendingNode>, String) {
2482        use bsv_rs::transaction::{MerklePathLeaf, TransactionInput, TransactionOutput};
2483        let count = 2 + rng.below(8) as usize;
2484        let mut txs: Vec<Transaction> = Vec::new();
2485        for i in 0..count {
2486            let mut tx = Transaction::new();
2487            let mut spent: HashSet<(usize, u32)> = HashSet::new();
2488            if i > 0 {
2489                for _ in 0..rng.below(5) {
2490                    let source = rng.below(i as u64) as usize;
2491                    let vout = rng.below(3) as u32;
2492                    if spent.insert((source, vout)) {
2493                        tx.inputs
2494                            .push(TransactionInput::new(txs[source].id(), vout));
2495                    }
2496                }
2497            }
2498            for _ in 0..3 {
2499                tx.outputs.push(TransactionOutput::new(
2500                    1000 + i as u64,
2501                    bsv_rs::script::LockingScript::from_hex("51").unwrap(),
2502                ));
2503            }
2504            txs.push(tx);
2505        }
2506        // Proofs: a transaction alone in its block, or a pair in one block.
2507        let mut proofs: Vec<Option<String>> = vec![None; count];
2508        let mut i = 0;
2509        while i < count {
2510            match rng.below(5) {
2511                0 => {
2512                    let path = vec![vec![MerklePathLeaf::new_txid(0, txs[i].id())]];
2513                    proofs[i] = Some(MerklePath::new(100 + i as u32, path).unwrap().to_hex());
2514                }
2515                1 if i + 1 < count => {
2516                    let (a, b) = (txs[i].id(), txs[i + 1].id());
2517                    let height = 100 + i as u32;
2518                    let of_a = vec![vec![
2519                        MerklePathLeaf::new_txid(0, a.clone()),
2520                        MerklePathLeaf::new(1, b.clone()),
2521                    ]];
2522                    let of_b = vec![vec![
2523                        MerklePathLeaf::new(0, a),
2524                        MerklePathLeaf::new_txid(1, b),
2525                    ]];
2526                    proofs[i] = Some(MerklePath::new(height, of_a).unwrap().to_hex());
2527                    proofs[i + 1] = Some(MerklePath::new(height, of_b).unwrap().to_hex());
2528                    i += 1;
2529                }
2530                _ => {}
2531            }
2532            i += 1;
2533        }
2534        let root_key = format!("{}.0", txs[count - 1].id());
2535        let mut refs: HashMap<String, PendingNode> = HashMap::new();
2536        refs.insert(
2537            root_key.clone(),
2538            pending(txs[count - 1].to_hex(), 0, proofs[count - 1].clone()),
2539        );
2540        // Every referenced outpoint is a node, most of the time; a node is
2541        // a child of the node(s) of its spender.
2542        for (i, tx) in txs.iter().enumerate().rev() {
2543            let own_keys: Vec<String> = (0..3)
2544                .map(|vout| format!("{}.{vout}", tx.id()))
2545                .filter(|key| refs.contains_key(key))
2546                .collect();
2547            if own_keys.is_empty() || proofs[i].is_some() && rng.below(3) > 0 {
2548                continue;
2549            }
2550            for input in &tx.inputs {
2551                let source_txid = input.get_source_txid().unwrap();
2552                let source = txs.iter().position(|t| t.id() == source_txid).unwrap();
2553                let key = format!("{source_txid}.{}", input.source_output_index);
2554                if rng.below(6) == 0 {
2555                    continue;
2556                }
2557                let mut node = pending(
2558                    txs[source].to_hex(),
2559                    input.source_output_index,
2560                    proofs[source].clone(),
2561                );
2562                match rng.below(40) {
2563                    0 => node.node.proof = Some("zz".to_string()),
2564                    1 => node.node.raw_tx = "00".to_string(),
2565                    _ => {}
2566                }
2567                refs.insert(key.clone(), node);
2568                for own in &own_keys {
2569                    let children = &mut refs.get_mut(own).unwrap().children;
2570                    if !children.contains(&key) {
2571                        children.push(key.clone());
2572                    }
2573                }
2574            }
2575        }
2576        (refs, root_key)
2577    }
2578
2579    // The assembly against the recursion it replaced, over 600 random
2580    // graphs, tolerant and strict: the BEEF of every node, the ordered BEEFs
2581    // of the graph, and every refusal, are the same bytes and the same words.
2582    #[test]
2583    fn e586_the_assembly_gives_the_hydrated_trees_bytes_and_errors() {
2584        let mut rng = Lcg(0x586);
2585        let (mut beefs, mut refusals, mut proven, mut partial) = (0, 0, 0, 0);
2586        for _ in 0..600 {
2587            let (refs, root_key) = random_graph(&mut rng);
2588            for strict in [false, true] {
2589                let mut keys: Vec<&String> = refs.keys().collect();
2590                keys.sort();
2591                for key in keys {
2592                    let legacy = legacy_get_beef_for_node(key, &refs, strict)
2593                        .map(|(_, beef)| beef)
2594                        .map_err(|e| e.to_string());
2595                    // A fresh assembly per node, and one shared by the graph
2596                    // below (`compute_ordered_beefs`): both orders of use.
2597                    let mut assembly = GraphAssembly::default();
2598                    let ours = assembly
2599                        .check(key, &refs, strict)
2600                        .map(|()| assembly.beef_of(key))
2601                        .map_err(|e| e.to_string());
2602                    assert_eq!(ours, legacy, "node {key}, strict {strict}");
2603                    match &ours {
2604                        Ok(beef) => {
2605                            beefs += 1;
2606                            let parsed = Beef::from_binary(beef).unwrap();
2607                            proven += usize::from(!parsed.bumps.is_empty());
2608                            partial += usize::from(
2609                                !strict && legacy_get_beef_for_node(key, &refs, true).is_err(),
2610                            );
2611                        }
2612                        Err(_) => refusals += 1,
2613                    }
2614                }
2615                let legacy = legacy_compute_ordered_beefs(&root_key, &refs, strict)
2616                    .map_err(|e| e.to_string());
2617                let ours =
2618                    OverlayGASPStorage::compute_ordered_beefs(&root_key, &refs, strict, None)
2619                        .map_err(|e| e.to_string());
2620                assert_eq!(ours, legacy, "graph {root_key}, strict {strict}");
2621            }
2622        }
2623        println!(
2624            "#586: {beefs} BEEFs equal ({proven} with a proof, {partial} partial), {refusals} refusals equal"
2625        );
2626        // The generator reached every class.
2627        assert!(beefs > 2000 && proven > 300 && partial > 100 && refusals > 300);
2628    }
2629
2630    // A node unknown to the graph, and the SDK's own words for a missing
2631    // source under `strict_beef`.
2632    #[test]
2633    fn e586_the_refusals_are_the_recursions_words() {
2634        let parent = "ab".repeat(32);
2635        let raw = make_valid_tx_hex(&parent, 3);
2636        let key = format!("{}.0", Transaction::from_hex(&raw).unwrap().id());
2637        let refs = HashMap::from([(key.clone(), pending(raw, 0, None))]);
2638        let mut assembly = GraphAssembly::default();
2639        assert_eq!(
2640            assembly
2641                .check("nope.0", &refs, false)
2642                .unwrap_err()
2643                .to_string(),
2644            legacy_get_beef_for_node("nope.0", &refs, false)
2645                .unwrap_err()
2646                .to_string()
2647        );
2648        let strict = assembly.check(&key, &refs, true).unwrap_err().to_string();
2649        assert_eq!(
2650            strict,
2651            legacy_get_beef_for_node(&key, &refs, true)
2652                .unwrap_err()
2653                .to_string()
2654        );
2655        assert!(strict.contains(&format!(
2656            "Missing source transaction for input 0 (txid: {parent})"
2657        )));
2658        // Tolerant, the same node assembles: a partial BEEF of one transaction.
2659        assembly.check(&key, &refs, false).unwrap();
2660        assert_eq!(
2661            assembly.beef_of(&key),
2662            legacy_get_beef_for_node(&key, &refs, false).unwrap().1
2663        );
2664    }
2665
2666    // The walks keep their work on the heap. A chain of 20,000 unproven
2667    // links is assembled; the recursion it replaced used the stack once per
2668    // link. And a link back into a node the walk is still inside (the root
2669    // is filed under the graph id the PEER names, so a peer can file a root
2670    // under an outpoint its own ancestry spends) ends there: the recursion
2671    // never returned.
2672    #[test]
2673    fn e586_a_deep_chain_and_a_cycle_do_not_use_the_stack() {
2674        use bsv_rs::transaction::{TransactionInput, TransactionOutput};
2675        let mut refs: HashMap<String, PendingNode> = HashMap::new();
2676        let mut previous: Option<String> = None;
2677        let mut tip = String::new();
2678        for i in 0..20_000u64 {
2679            let mut tx = Transaction::new();
2680            if let Some(txid) = &previous {
2681                tx.inputs.push(TransactionInput::new(txid.clone(), 0));
2682            }
2683            tx.outputs.push(TransactionOutput::new(
2684                i,
2685                bsv_rs::script::LockingScript::from_hex("51").unwrap(),
2686            ));
2687            let txid = tx.id();
2688            tip = format!("{txid}.0");
2689            let mut node = pending(tx.to_hex(), 0, None);
2690            if let Some(below) = &previous {
2691                node.children.push(format!("{below}.0"));
2692            }
2693            refs.insert(tip.clone(), node);
2694            previous = Some(txid);
2695        }
2696        assert_eq!(OverlayGASPStorage::ordered_keys(&tip, &refs).len(), 20_000);
2697        let mut assembly = GraphAssembly::default();
2698        assembly.check(&tip, &refs, true).unwrap();
2699        let beef = Beef::from_binary(&assembly.beef_of(&tip)).unwrap();
2700        assert_eq!(beef.txs.len(), 20_000);
2701
2702        // root (filed under `decoy.0`) spends y.0; y spends `decoy.0`.
2703        let decoy = "cd".repeat(32);
2704        let y_raw = make_valid_tx_hex(&decoy, 0);
2705        let y = Transaction::from_hex(&y_raw).unwrap().id();
2706        let root_raw = make_valid_tx_hex(&y, 0);
2707        let root_key = format!("{decoy}.0");
2708        let mut root = pending(root_raw, 0, None);
2709        root.children.push(format!("{y}.0"));
2710        let refs = HashMap::from([
2711            (root_key.clone(), root),
2712            (format!("{y}.0"), pending(y_raw, 0, None)),
2713        ]);
2714        let mut assembly = GraphAssembly::default();
2715        assembly.check(&root_key, &refs, true).unwrap();
2716        let beef = Beef::from_binary(&assembly.beef_of(&root_key)).unwrap();
2717        assert_eq!(beef.txs.len(), 2);
2718        let beefs =
2719            OverlayGASPStorage::compute_ordered_beefs(&root_key, &refs, true, None).unwrap();
2720        assert_eq!(beefs.len(), 2);
2721    }
2722}