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 sync_expiry(&self, node: &GASPNode, outpoint: &str) -> Result<Option<u64>, GASPError> {
771        let Some(manager) = self.topic_manager else {
772            return Ok(None);
773        };
774        let tx = Transaction::from_hex(&node.raw_tx)
775            .map_err(|e| GASPError::Other(format!("sync root parse: {e}")))?;
776        if format!("{}.{}", tx.id(), node.output_index) != outpoint {
777            return Err(GASPError::Other(
778                "sync root content-address mismatch".into(),
779            ));
780        }
781        let output = tx
782            .outputs
783            .get(node.output_index as usize)
784            .ok_or_else(|| GASPError::Other("sync root output missing".into()))?;
785        Ok(manager.sync_expiry(&output.locking_script))
786    }
787
788    /// Returns UTXOs known for this topic since the given score/timestamp.
789    ///
790    /// Delegates to `Storage::find_utxos_for_topic()`, converting `Output`
791    /// records to `GASPOutput` with txid, output_index, and score.
792    async fn find_known_utxos(
793        &self,
794        since: u64,
795        limit: Option<u64>,
796    ) -> Result<Vec<GASPOutput>, GASPError> {
797        let outputs = self
798            .storage
799            .find_utxos_for_topic(
800                &self.topic,
801                Some(since as f64),
802                limit,
803                false, // don't need BEEF for listing
804            )
805            .await
806            .map_err(|e| GASPError::StorageError(e.to_string()))?;
807
808        Ok(outputs
809            .iter()
810            .map(|o| GASPOutput {
811                txid: o.txid.clone(),
812                output_index: o.output_index,
813                score: o.score.unwrap_or(0.0),
814            })
815            .collect())
816    }
817
818    /// Hydrate a GASP node with transaction data from storage.
819    ///
820    /// Looks up the root output (from graph_id) and searches its BEEF tree
821    /// for the requested txid. This is a simplified version that returns
822    /// the raw transaction data and proof if available.
823    ///
824    /// For a full implementation, this would delegate to
825    /// `Engine::provide_foreign_gasp_node()`, but since we only hold a
826    /// `Storage` reference (not the full Engine), we do a simpler lookup.
827    async fn hydrate_gasp_node(
828        &self,
829        graph_id: &str,
830        txid: &str,
831        output_index: u32,
832        _metadata: bool,
833    ) -> Result<GASPNode, GASPError> {
834        // Try to find the output with BEEF data
835        let output = self
836            .storage
837            .find_output(txid, output_index, Some(&self.topic), None, true)
838            .await
839            .map_err(|e| GASPError::StorageError(e.to_string()))?;
840
841        if let Some(output) = output {
842            if let Some(ref beef) = output.beef {
843                // Parse the BEEF to extract raw transaction hex
844                match beef_limits::transaction_from_beef(
845                    beef,
846                    None,
847                    &beef_limits::STORED_BEEF_LIMITS,
848                ) {
849                    Ok(tx) => {
850                        let mut node = GASPNode {
851                            graph_id: graph_id.to_string(),
852                            raw_tx: tx.to_hex(),
853                            output_index,
854                            proof: None,
855                            tx_metadata: None,
856                            output_metadata: None,
857                            inputs: None,
858                        };
859                        if let Some(ref merkle_path) = tx.merkle_path {
860                            node.proof = Some(merkle_path.to_hex());
861                        }
862                        return Ok(node);
863                    }
864                    Err(e) => {
865                        warn!("Failed to parse BEEF for {txid}: {e}");
866                    }
867                }
868            }
869        }
870
871        // Fallback: look up without topic filter
872        let output = self
873            .storage
874            .find_output(txid, output_index, None, None, true)
875            .await
876            .map_err(|e| GASPError::StorageError(e.to_string()))?
877            .ok_or_else(|| {
878                GASPError::NodeNotFound(format!("Output {txid}.{output_index} not found"))
879            })?;
880
881        let beef = output.beef.as_ref().ok_or_else(|| {
882            GASPError::NodeNotFound(format!("No BEEF data for {txid}.{output_index}"))
883        })?;
884
885        let tx = beef_limits::transaction_from_beef(beef, None, &beef_limits::STORED_BEEF_LIMITS)
886            .map_err(|e| GASPError::Other(format!("BEEF parse error: {e}")))?;
887
888        let mut node = GASPNode {
889            graph_id: graph_id.to_string(),
890            raw_tx: tx.to_hex(),
891            output_index,
892            proof: None,
893            tx_metadata: None,
894            output_metadata: None,
895            inputs: None,
896        };
897        if let Some(ref merkle_path) = tx.merkle_path {
898            node.proof = Some(merkle_path.to_hex());
899        }
900        Ok(node)
901    }
902
903    /// Determine which inputs are needed to validate this node.
904    ///
905    /// - A proven node ends the walk if its topic manager names no inputs it
906    ///   still needs: for a node whose output is NOT admissible, every named
907    ///   input not held (D13); for one whose output IS admissible, every named
908    ///   input neither held nor landed (lane E1D, an addition: the reference
909    ///   stops at an admissible node). Without a manager, proven nodes always
910    ///   stop.
911    /// - If no proof, parses the raw transaction and requests all inputs,
912    ///   filtering out any inputs already known in local storage.
913    /// - Admission errors propagate to GASP's incoming UTXO handler. Errors
914    ///   identifying needed inputs are logged and cut off this node, like TS.
915    /// - Unproven raw_tx parse failures retain the legacy graceful fallback.
916    async fn find_needed_inputs(
917        &self,
918        node: &GASPNode,
919    ) -> Result<Option<GASPNodeResponse>, GASPError> {
920        let mut requested_inputs: HashMap<String, GASPInputRequest> = HashMap::new();
921        // A proven node whose own output the dry run admits (lane E1D).
922        let mut admitted = false;
923
924        if let Some(proof_hex) = &node.proof {
925            let Some(manager) = self.topic_manager else {
926                return Ok(None);
927            };
928            let mut tx = Transaction::from_hex(&node.raw_tx)
929                .map_err(|e| GASPError::Other(format!("Failed to parse raw_tx: {e}")))?;
930            tx.merkle_path = Some(
931                beef_limits::proof_from_hex(proof_hex)
932                    .map_err(|e| GASPError::Other(format!("Failed to parse proof: {e}")))?,
933            );
934
935            // TS passes { dryRun: true }: the node is the peer's and nothing
936            // is admitted here, so a manager that writes on admission must not.
937            // Utils.toArray(string) defaults to UTF-8, even for hex-like text.
938            // This call is outside the needed-input catch in the reference.
939            let admittance = manager
940                .identify_admissible_outputs(
941                    &tx,
942                    &[],
943                    node.tx_metadata.as_deref().map(str::as_bytes),
944                    SubmitMode::HistoricalTx,
945                    &TopicAdmittanceContext::DRY_RUN,
946                )
947                .await
948                .map_err(|e| GASPError::Other(e.to_string()))?;
949            // The reference stops here (D13). Lane E1D (bsv-low #575,
950            // zanaadu-v2 #365) asks the named inputs of an ADMITTED node too:
951            // a manager that admits a successor with no coin (the pf head
952            // manager over its own head state) ended the walk at the
953            // successor while its predecessor had not landed, the graph was
954            // the successor alone, and its finalize submit recorded it with
955            // no coin; the predecessor, landed on a later tick, stood unspent
956            // beside the tip for good. So the predecessor is asked FIRST: a
957            // named input that is neither held nor landed (below) is
958            // requested and lands before the node in the graph's order, a
959            // finalize submit of it that does not land stops the graph, and
960            // a named input the peer cannot serve prunes (D14). When every
961            // named input is known the walk stops here, as the reference's.
962            admitted = admittance.outputs_to_admit.contains(&node.output_index);
963            // A node with no input to name stops here, the manager not
964            // asked (the reference's call count for such a node).
965            if admitted
966                && !tx
967                    .inputs
968                    .iter()
969                    .any(|input| input.get_source_txid().is_ok_and(|s| !s.is_empty()))
970            {
971                return Ok(None);
972            }
973
974            // A serialization RED is unreachable: bsv-rs stops collecting ancestors at a merkle path.
975            let needed_inputs = async {
976                let beef = tx
977                    .to_beef(false)
978                    .map_err(|e| format!("Failed to serialize proven BEEF: {e}"))?;
979                manager
980                    .identify_needed_inputs(&beef, None)
981                    .await
982                    .map_err(|e| e.to_string())
983            }
984            .await;
985            match needed_inputs {
986                Ok(inputs) => {
987                    for input in inputs {
988                        requested_inputs
989                            .insert(input.to_graph_id(), GASPInputRequest { metadata: false });
990                    }
991                }
992                Err(e) => {
993                    error!(
994                        "An error occurred when identifying needed inputs for transaction: {}.{}: {e}",
995                        tx.id(), node.output_index
996                    );
997                    return Ok(None);
998                }
999            }
1000        } else {
1001            // Unproven nodes always need all inputs, independently of the manager.
1002            let tx = match Transaction::from_hex(&node.raw_tx) {
1003                Ok(tx) => tx,
1004                Err(e) => {
1005                    warn!(
1006                        "Cannot parse raw_tx for find_needed_inputs (graph_id={}): {e}",
1007                        node.graph_id
1008                    );
1009                    return Ok(None);
1010                }
1011            };
1012            for input in &tx.inputs {
1013                let source_txid = input.get_source_txid().unwrap_or_default();
1014                if source_txid.is_empty() {
1015                    continue;
1016                }
1017                let outpoint = format!("{}.{}", source_txid, input.source_output_index);
1018                requested_inputs.insert(outpoint, GASPInputRequest { metadata: false });
1019            }
1020        }
1021
1022        if requested_inputs.is_empty() {
1023            return Ok(None);
1024        }
1025
1026        // Strip inputs that are already known in local storage.
1027        let mut to_remove = Vec::new();
1028        for outpoint in requested_inputs.keys() {
1029            if let Some((txid, oi)) = crate::gasp::parse_outpoint(outpoint) {
1030                match self
1031                    .storage
1032                    .find_output(&txid, oi, Some(&self.topic), None, false)
1033                    .await
1034                {
1035                    Ok(Some(_)) => {
1036                        to_remove.push(outpoint.clone());
1037                    }
1038                    Ok(None) => {}
1039                    Err(e) => {
1040                        debug!("Storage lookup failed for {outpoint}: {e}");
1041                    }
1042                }
1043            }
1044        }
1045        for key in &to_remove {
1046            requested_inputs.remove(key);
1047        }
1048
1049        // An admitted node's named inputs are requested only while their
1050        // landing is unknown (E1D): one whose transaction has an applied row
1051        // in the topic, or an output held there, has landed, and the walk
1052        // stops as the reference's. A read that faults keeps the input: one
1053        // round trip to the peer is the cheaper error. So does a transaction
1054        // past [`LANDED_READS_PER_NODE`] (the E1D lens fold, L3: a manager
1055        // names as many inputs as it likes, D14's decoys included, and each
1056        // costs up to two reads): it is requested, and a landed one the peer
1057        // serves is a dupe at its finalize submit.
1058        if admitted {
1059            let mut landed: HashSet<String> = HashSet::new();
1060            let mut unlanded: HashSet<String> = HashSet::new();
1061            let mut reads = 0usize;
1062            let mut outpoints: Vec<&String> = requested_inputs.keys().collect();
1063            outpoints.sort();
1064            for outpoint in outpoints {
1065                let Some((txid, _)) = crate::gasp::parse_outpoint(outpoint) else {
1066                    continue;
1067                };
1068                if landed.contains(&txid) || unlanded.contains(&txid) {
1069                    continue;
1070                }
1071                if reads + 2 > LANDED_READS_PER_NODE {
1072                    break;
1073                }
1074                reads += 1;
1075                let record = crate::types::AppliedTransaction {
1076                    txid: txid.clone(),
1077                    topic: self.topic.clone(),
1078                };
1079                let mut holds = matches!(
1080                    self.storage.does_applied_transaction_exist(&record).await,
1081                    Ok(true)
1082                );
1083                if !holds {
1084                    reads += 1;
1085                    holds = matches!(
1086                        self.storage.find_outputs_for_transaction(&txid, false).await,
1087                        Ok(outputs) if outputs.iter().any(|o| o.topic == self.topic)
1088                    );
1089                }
1090                if holds {
1091                    landed.insert(txid);
1092                } else {
1093                    unlanded.insert(txid);
1094                }
1095            }
1096            requested_inputs.retain(|outpoint, _| {
1097                crate::gasp::parse_outpoint(outpoint)
1098                    .is_none_or(|(txid, _)| !landed.contains(&txid))
1099            });
1100        }
1101
1102        if requested_inputs.is_empty() {
1103            return Ok(None);
1104        }
1105
1106        Ok(Some(GASPNodeResponse { requested_inputs }))
1107    }
1108
1109    /// Append a node to a temporary graph being constructed during sync.
1110    ///
1111    /// If `spent_by` is `None`, this is the root node and is keyed by `graph_id`.
1112    /// Otherwise, the node is keyed by its computed txid.outputIndex, and is
1113    /// registered as a child of the parent node identified by `spent_by`.
1114    async fn append_to_graph(
1115        &self,
1116        node: &GASPNode,
1117        spent_by: Option<&str>,
1118    ) -> Result<(), GASPError> {
1119        debug!(
1120            "Appending node to graph {}: tx={}..., oi={}, spent_by={:?}",
1121            node.graph_id,
1122            &node.raw_tx[..node.raw_tx.len().min(16)],
1123            node.output_index,
1124            spent_by,
1125        );
1126
1127        // Compute the key for this node.
1128        let node_key = if spent_by.is_none() {
1129            // Root node — keyed by graph_id
1130            node.graph_id.clone()
1131        } else {
1132            // Child node — keyed by txid.outputIndex
1133            match Transaction::from_hex(&node.raw_tx) {
1134                Ok(tx) => format!("{}.{}", tx.id(), node.output_index),
1135                Err(_) => {
1136                    // Fallback: use raw_tx prefix as key
1137                    format!(
1138                        "{}.{}",
1139                        &node.raw_tx[..node.raw_tx.len().min(64)],
1140                        node.output_index
1141                    )
1142                }
1143            }
1144        };
1145
1146        // The pending graphs change: nothing kept of an anchor check holds.
1147        self.take_anchored();
1148        let mut refs = self
1149            .pending_graphs
1150            .lock()
1151            .map_err(|e| GASPError::Other(format!("Lock poisoned: {e}")))?;
1152
1153        // Insert this node
1154        refs.insert(
1155            node_key.clone(),
1156            PendingNode {
1157                node: node.clone(),
1158                spent_by: spent_by.map(std::string::ToString::to_string),
1159                children: Vec::new(),
1160            },
1161        );
1162
1163        // If spent_by is set, register this node as a child of the parent.
1164        if let Some(parent_key) = spent_by {
1165            if let Some(parent) = refs.get_mut(parent_key) {
1166                parent.children.push(node_key);
1167            }
1168        }
1169
1170        Ok(())
1171    }
1172
1173    /// Check a completed graph before anything of it is finalized (bsv-low
1174    /// #551). The reference's `validateGraphAnchor`
1175    /// (`OverlayGASPStorage.ts` at f999e0c1a, lines 261 to 297), in its two
1176    /// steps. An `Err` is a discard: `GASPSync::complete_graph` drops the
1177    /// graph and NOTHING of it reaches the sink.
1178    ///
1179    /// **1. Bitcoin.** The ROOT node's BEEF (`GraphAssembly::beef_of`) is verified
1180    /// by the engine's own check, `verify_spv_like_the_reference`: the
1181    /// reference's `spvTx.verify(this.engine.chainTracker)`. With a tracker
1182    /// every merkle root in that BEEF is checked against it and every unproven
1183    /// input's script is executed; with none it is `'scripts only'` (roots
1184    /// accepted unchecked, scripts still run). One verifier, the same switch
1185    /// (`with_script_verification`) as `Engine::submit`.
1186    ///
1187    /// **2. Overlay, all or nothing.** The ordered BEEFs (ancestors first, root
1188    /// last) are replayed through the topic manager as the reference's
1189    /// `admitHistoricalBEEF` does: mode `historical-tx`, no off-chain values,
1190    /// `previous_coins` the inputs whose source outpoint is a coin (the
1191    /// engine's little-endian u32 input indices), and every output the manager
1192    /// admits joins the coins. If the ROOT's outpoint is not a coin at the
1193    /// end, the graph is refused. The set only GROWS, as in the reference at
1194    /// f999e0c1a: it reads `outputsToAdmit` alone, never `coinsToRetain`, and
1195    /// removes nothing ("a Set of all historical coins to retain (no need to
1196    /// remove them)"). The reference passes `{ dryRun: true }` and so does
1197    /// this replay (`TopicAdmittanceContext::DRY_RUN`): a manager that writes
1198    /// on admission writes nothing here.
1199    ///
1200    /// WITHOUT a topic manager step 2 does not run: there is nothing to
1201    /// replay, and `Engine::submit` refuses such a topic (`UnsupportedTopic`),
1202    /// so nothing can be admitted. `Engine::start_gasp_sync` always wires the
1203    /// manager of a topic it can admit.
1204    ///
1205    /// **Divergences by addition** (each with its why):
1206    ///
1207    /// - *Coins the storage already holds count.* `find_needed_inputs` strips
1208    ///   an input the local storage already holds (as the reference does), so
1209    ///   its node is not in the graph and the replay's own set never sees it.
1210    ///   The reference then refuses a root that extends a coin it already
1211    ///   holds: the next head of a head chain, synced one tick after the
1212    ///   last, would be discarded forever. A previous coin is therefore an
1213    ///   input in the set OR in the storage for this topic, the exact lookup
1214    ///   `Engine::submit` makes when finalize submits the same BEEF.
1215    /// - *Sources the storage already holds are merged for step 1.* For the
1216    ///   same reason an UNPROVEN node may lack an input's source transaction
1217    ///   (the reference throws in `getBEEFForNode` and discards). The stored
1218    ///   BEEF of that output is merged into the checked copy, so the spend is
1219    ///   executed against its real source and that source's own ancestry is
1220    ///   verified too. If the stored BEEF is itself incomplete the verifier
1221    ///   refuses, and the graph is discarded as in the reference. Only the
1222    ///   CHECKED copy is merged; the finalized bytes are untouched.
1223    /// - *Every node is an input of its parent* (`read_anchor_input`).
1224    /// - *A conflicting spend refuses the graph.* Two transactions of one
1225    ///   graph that spend the same outpoint cannot both be admitted by the
1226    ///   finalize submits (the first one consumes the coin), so a replay that
1227    ///   showed the coin to both would promise an admission finalize cannot
1228    ///   keep. Upstream added the same rule after f999e0c1a (`spentOutpoints`,
1229    ///   "Historical GASP graph contains a conflicting spend").
1230    /// - *A fault of the moment is not a verdict.* A chain tracker that could
1231    ///   not answer, or a storage read that failed, says nothing about the
1232    ///   graph: the error is `GASPError::AnchorUnavailable`, the graph is
1233    ///   still discarded, but `complete_graph` FAILS the UTXO so the cursor
1234    ///   gap guard asks for it again. The reference discards and moves on,
1235    ///   which here would lose the UTXO for good behind an advanced cursor.
1236    /// - *A manager error in the replay is "not now", not a refusal.* The
1237    ///   replay is where a manager is shown the coins, so it is where one
1238    ///   whose own state lags can say it cannot place a transaction YET. An
1239    ///   `Err` from it is `GASPError::AnchorUnavailable` too: the UTXO fails
1240    ///   and is asked for again. `Ok` with nothing admitted stays the final
1241    ///   refusal, and there the cursor moves, as in the reference (which
1242    ///   moves it past a refused and a failed graph alike, `GASP.ts` 389 to
1243    ///   392, 419 to 423, 559 to 572, `Engine.ts` 1600 to 1604). The cost: a
1244    ///   manager that errors FOREVER on one UTXO holds that peer's cursor
1245    ///   below it, and what lies above and is not yet held is walked again
1246    ///   on every tick.
1247    ///
1248    /// **Cost.** Step 1 runs the script of every input of every unproven
1249    /// transaction in the root's BEEF and asks the tracker once per merkle
1250    /// path in it (a proven root: one question, no script). The ancestors a
1251    /// manager named BEHIND a proven node are not in that BEEF: their own
1252    /// merkle paths are not checked, here or in the reference; they are bound
1253    /// to the verified root by txid instead. Step 2 is one manager call per
1254    /// transaction and one storage read per input that is not already a coin
1255    /// of the set, which is what the finalize submits cost again.
1256    ///
1257    /// Since bsv-low #592 the walk runs under the engine's walk budget
1258    /// (`DoorBudget`, the door's two limbs, estimated from the bytes before
1259    /// anything runs). A breach is no verdict on the graph, and nothing
1260    /// stands behind this check (no network accepted these bytes, and the
1261    /// finalize submits them `historical-tx-no-spv`), so a breach is never
1262    /// the replay's to judge (the E592 lens fold, H1: as built at `4060ffd`
1263    /// it went on, and a peer's root that spent a held coin with no
1264    /// signature was finalized). It is HELD: `GASPError::AnchorWalkHeld`,
1265    /// the graph not finalized, the UTXO failed (or, under a per-graph
1266    /// budget, its record kept with nothing pending), the cursor below it,
1267    /// counted (`GASPSync::anchor_walk_held`). The roots the walk reached are
1268    /// checked first, so a bad proof is still the refusal it is. The hold
1269    /// ends when the peer serves the root PROVEN: #555's resume re-asks an
1270    /// unproven root, a proven one restarts the walk (`root_proven`), and a
1271    /// proven root is trusted without a script run. So an honest 0-conf graph
1272    /// too big to walk lands after its block; a forgery never lands. What a
1273    /// forging peer costs: its own slot (the UTXO held below the cursor, its
1274    /// record re-asked each pass up to `DEFERRED_GRAPH_MAX_PASSES`), inside
1275    /// its per-peer budget, and, as it never yields a finalized graph or a
1276    /// cursor move on that UTXO, D16's yieldless bound quarantines it if it
1277    /// serves nothing else, and ONLY with a per-graph budget
1278    /// (`Engine::set_graph_budget`): the yieldless bound exists only there
1279    /// (the E592 delta lens, D1-L2). With NO per-graph budget (a library
1280    /// consumer that sets none) there is no record and no yieldless count:
1281    /// the held UTXO fails on every sync, the sync runs to its end and is a
1282    /// success under #302, the cursor stays below that UTXO for good, and
1283    /// the graph is fetched and its walk estimated again on every tick;
1284    /// nothing is admitted, and the cost is ours, the residual D15 states for
1285    /// `AnchorUnavailable`. The reference serialises this method behind
1286    /// `acquireAnchorValidationSlot` (at most 4 at once); here graphs are
1287    /// completed one at a time by one sync on one thread, so there is
1288    /// nothing to serialise.
1289    async fn validate_graph_anchor(&self, graph_id: &str) -> Result<(), GASPError> {
1290        let anchor = {
1291            let refs = self
1292                .pending_graphs
1293                .lock()
1294                .map_err(|e| GASPError::Other(format!("Lock poisoned: {e}")))?;
1295            Self::read_anchor_input(graph_id, &refs, self.strict_beef)?
1296        };
1297        let unavailable =
1298            |what: String| GASPError::AnchorUnavailable(format!("graph {graph_id}: {what}"));
1299        let refused = |why: String| GASPError::ValidationFailed(format!("graph {graph_id}: {why}"));
1300
1301        // ── 1. Bitcoin ──────────────────────────────────────────────────
1302        let mut root_beef = anchor.root_beef;
1303        if !anchor.absent_sources.is_empty() {
1304            let mut merged = beef_limits::parse_beef(&root_beef, &beef_limits::PEER_BEEF_LIMITS)
1305                .map_err(|e| refused(format!("root BEEF does not parse: {e}")))?;
1306            for (txid, output_index) in &anchor.absent_sources {
1307                let held = self
1308                    .storage
1309                    .find_output(txid, *output_index, Some(&self.topic), None, true)
1310                    .await
1311                    .map_err(|e| unavailable(format!("reading {txid}.{output_index}: {e}")))?;
1312                if let Some(stored) = held.and_then(|output| output.beef) {
1313                    if let Ok(stored) =
1314                        beef_limits::parse_beef(&stored, &beef_limits::STORED_BEEF_LIMITS)
1315                    {
1316                        merged.merge_beef(&stored);
1317                    }
1318                }
1319            }
1320            root_beef = merged.to_binary();
1321        }
1322        let tracker = self.chain_tracker.map(|inner| FaultNotingTracker {
1323            inner,
1324            faulted: AtomicBool::new(false),
1325        });
1326        let verdict = crate::engine::verify_spv_like_the_reference(
1327            tracker.as_ref().map(|t| t as &dyn ChainTracker),
1328            self.verify_scripts,
1329            &root_beef,
1330            &anchor.root_txid,
1331            self.walk_budget,
1332        )
1333        .await;
1334        // The checked copy has done its work (bsv-low #586: not held through
1335        // the replay).
1336        drop(root_beef);
1337        // A walk that could not run within the budget (bsv-low #592) is not
1338        // a verdict on the graph, and nothing else judges it: HELD until the
1339        // root is served proven (the E592 lens fold, H1; the doc above).
1340        let verdict = match verdict {
1341            Ok(Some(stop)) => {
1342                warn!(
1343                    "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)",
1344                    stop.subject_txid,
1345                    stop.limb.as_str(),
1346                    stop.at_txid,
1347                    stop.what
1348                );
1349                return Err(GASPError::AnchorWalkHeld(format!(
1350                    "graph {graph_id}: the script walk of {} could not run within the budget ({}: {})",
1351                    stop.subject_txid,
1352                    stop.limb.as_str(),
1353                    stop.what
1354                )));
1355            }
1356            Ok(None) => Ok(()),
1357            Err(e) => Err(e),
1358        };
1359        if let Err(e) = verdict {
1360            if tracker
1361                .as_ref()
1362                .is_some_and(|t| t.faulted.load(Ordering::Relaxed))
1363            {
1364                return Err(unavailable(format!("the chain tracker faulted: {e}")));
1365            }
1366            return Err(refused(format!(
1367                "The graph is not well-anchored according to the rules of Bitcoin: {e}"
1368            )));
1369        }
1370
1371        // ── 2. Overlay ──────────────────────────────────────────────────
1372        let Some(manager) = self.topic_manager else {
1373            self.keep_anchored(graph_id, anchor.order, anchor.assembly);
1374            return Ok(());
1375        };
1376        let mut coins: HashSet<String> = HashSet::new();
1377        let mut spent: HashSet<String> = HashSet::new();
1378        let mut replayed: HashSet<String> = HashSet::new();
1379        for key in &anchor.order {
1380            // One node's BEEF at a time, the bytes `finalize_graph` will
1381            // assemble, dropped when its turn is over (bsv-low #586).
1382            let beef = &anchor.assembly.beef_of(key);
1383            // The parse `Engine::submit` makes of the same bytes at finalize.
1384            let subject = beef_limits::parse_beef(beef, &beef_limits::PEER_BEEF_LIMITS)
1385                .map(|mut b| crate::subject::subject_txid_of(&mut b))
1386                .map_err(|e| refused(format!("a graph BEEF does not parse: {e}")))?;
1387            let tx = beef_limits::transaction_from_beef(
1388                beef,
1389                subject.as_deref(),
1390                &beef_limits::PEER_BEEF_LIMITS,
1391            )
1392            .map_err(|e| refused(format!("a graph BEEF does not parse: {e}")))?;
1393            let txid = tx.id();
1394            // Two outputs of one transaction are two nodes with one BEEF.
1395            if !replayed.insert(txid.clone()) {
1396                continue;
1397            }
1398            let mut previous_coins: Vec<u8> = Vec::new();
1399            for (input_index, input) in tx.inputs.iter().enumerate() {
1400                let source_txid = input.get_source_txid().unwrap_or_default();
1401                if source_txid.is_empty() {
1402                    continue;
1403                }
1404                let outpoint = format!("{}.{}", source_txid, input.source_output_index);
1405                if !spent.insert(outpoint.clone()) {
1406                    return Err(refused(format!(
1407                        "the graph contains a conflicting spend of {outpoint}"
1408                    )));
1409                }
1410                let is_coin = coins.contains(&outpoint)
1411                    || self
1412                        .storage
1413                        .find_output(
1414                            &source_txid,
1415                            input.source_output_index,
1416                            Some(&self.topic),
1417                            None,
1418                            false,
1419                        )
1420                        .await
1421                        .map_err(|e| unavailable(format!("reading {outpoint}: {e}")))?
1422                        .is_some();
1423                if is_coin {
1424                    previous_coins.extend_from_slice(&(input_index as u32).to_le_bytes());
1425                }
1426            }
1427            // TS passes { dryRun: true }: the replay only asks, the finalize
1428            // submit is the admission.
1429            // An `Err` is the manager's "not now" (a fault of the moment, see
1430            // the divergences above); `Ok` with nothing is its refusal.
1431            let admittance = manager
1432                .identify_admissible_outputs(
1433                    &tx,
1434                    &previous_coins,
1435                    None,
1436                    SubmitMode::HistoricalTx,
1437                    &TopicAdmittanceContext::DRY_RUN,
1438                )
1439                .await
1440                .map_err(|e| unavailable(format!("the topic manager failed on {txid}: {e}")))?;
1441            for output_index in admittance.outputs_to_admit {
1442                coins.insert(format!("{txid}.{output_index}"));
1443            }
1444        }
1445        if !coins.contains(graph_id) {
1446            return Err(refused(
1447                "This graph did not result in topical admittance of the root node. Rejecting."
1448                    .to_string(),
1449            ));
1450        }
1451        self.keep_anchored(graph_id, anchor.order, anchor.assembly);
1452        Ok(())
1453    }
1454
1455    /// Finalize a graph — convert accumulated nodes to ordered BEEFs and push
1456    /// them to the shared sink for later Engine submission.
1457    ///
1458    /// Computes ordered BEEF byte arrays (ancestors first, root last) from the
1459    /// temporary graph nodes and stores them in the `FinalizedGraphSink`. The
1460    /// Engine drains it (after the sync, or per completed UTXO under a
1461    /// per-peer budget) and submits each one with `HistoricalTxNoSpv` mode, which skips SPV because
1462    /// `validate_graph_anchor` has already done it (the reference's
1463    /// `finalizeGraph`, lines 317 to 333).
1464    async fn finalize_graph(&self, graph_id: &str) -> Result<(), GASPError> {
1465        let refs = self
1466            .pending_graphs
1467            .lock()
1468            .map_err(|e| GASPError::Other(format!("Lock poisoned: {e}")))?;
1469
1470        if !refs.contains_key(graph_id) {
1471            return Err(GASPError::Other(format!("No pending graph for {graph_id}")));
1472        }
1473
1474        let node_count = refs
1475            .values()
1476            .filter(|pn| pn.node.graph_id == graph_id)
1477            .count();
1478
1479        debug!("Finalizing graph {graph_id} ({node_count} nodes). Computing ordered BEEFs.");
1480
1481        // Compute ordered BEEFs for the graph.
1482        let beefs =
1483            Self::compute_ordered_beefs(graph_id, &refs, self.strict_beef, self.take_anchored())?;
1484
1485        // Push to shared sink for later Engine submission.
1486        drop(refs);
1487        self.finalized_sink
1488            .lock()
1489            .map_err(|e| GASPError::Other(format!("Lock poisoned: {e}")))?
1490            .push(FinalizedGraph {
1491                topic: self.topic.clone(),
1492                beefs,
1493            });
1494
1495        // Remove all nodes belonging to this graph from pending refs.
1496        let mut refs = self
1497            .pending_graphs
1498            .lock()
1499            .map_err(|e| GASPError::Other(format!("Lock poisoned: {e}")))?;
1500        let keys_to_remove: Vec<String> = refs
1501            .iter()
1502            .filter(|(_, pn)| pn.node.graph_id == graph_id)
1503            .map(|(k, _)| k.clone())
1504            .collect();
1505        for key in keys_to_remove {
1506            refs.remove(&key);
1507        }
1508
1509        Ok(())
1510    }
1511
1512    /// Discard a temporary graph that failed validation.
1513    async fn discard_graph(&self, graph_id: &str) -> Result<(), GASPError> {
1514        debug!("Discarding graph: {graph_id}");
1515        self.take_anchored();
1516        let mut refs = self
1517            .pending_graphs
1518            .lock()
1519            .map_err(|e| GASPError::Other(format!("Lock poisoned: {e}")))?;
1520        let keys_to_remove: Vec<String> = refs
1521            .iter()
1522            .filter(|(_, pn)| pn.node.graph_id == graph_id)
1523            .map(|(k, _)| k.clone())
1524            .collect();
1525        for key in keys_to_remove {
1526            refs.remove(&key);
1527        }
1528        Ok(())
1529    }
1530
1531    async fn read_fault_memos(
1532        &self,
1533        outpoints: Option<&[String]>,
1534    ) -> Result<crate::fault_memo::FaultMemoBatch, GASPError> {
1535        let batch = self
1536            .storage
1537            .read_fault_memos(&self.peer, &self.topic, outpoints)
1538            .await
1539            .map_err(|e| GASPError::StorageError(e.to_string()))?;
1540        let mut cache = self.fault_memos.lock().unwrap();
1541        // Only the active batch is cached. A walk spanning many pages does not
1542        // keep the complete memo in isolate memory.
1543        cache.clear();
1544        for record in &batch.records {
1545            cache.insert(record.outpoint.clone(), record.clone().into_record());
1546        }
1547        Ok(batch)
1548    }
1549
1550    async fn load_deferred_graphs(&self) -> Result<Vec<crate::gasp::DeferredGraphKey>, GASPError> {
1551        self.storage
1552            .find_deferred_graphs(&self.peer, &self.topic)
1553            .await
1554            .map_err(|e| GASPError::StorageError(e.to_string()))
1555    }
1556
1557    async fn get_deferred_graph(
1558        &self,
1559        outpoint: &str,
1560    ) -> Result<Option<crate::gasp::DeferredGraph>, GASPError> {
1561        if let Some(record) = self.fault_memos.lock().unwrap().get(outpoint).cloned() {
1562            return Ok(Some(record));
1563        }
1564        self.storage
1565            .get_deferred_graph(&self.peer, &self.topic, outpoint)
1566            .await
1567            .map_err(|e| GASPError::StorageError(e.to_string()))
1568    }
1569
1570    async fn save_deferred_graph(
1571        &self,
1572        record: &crate::gasp::DeferredGraph,
1573    ) -> Result<crate::gasp::DeferredGraphSave, GASPError> {
1574        let mut record = record.clone();
1575        record.peer.clone_from(&self.peer);
1576        record.topic.clone_from(&self.topic);
1577        record.configured = self.configured_peer;
1578        let result = self
1579            .storage
1580            .put_deferred_graph(&record)
1581            .await
1582            .map_err(|e| GASPError::StorageError(e.to_string()))?;
1583        if result == crate::gasp::DeferredGraphSave::Saved {
1584            let mut cache = self.fault_memos.lock().unwrap();
1585            if record.is_struck() {
1586                cache.insert(record.outpoint.clone(), record);
1587            } else {
1588                cache.remove(&record.outpoint);
1589            }
1590        }
1591        Ok(result)
1592    }
1593
1594    async fn reserve_unanchorable_retry(
1595        &self,
1596        record: &crate::gasp::DeferredGraph,
1597    ) -> Result<bool, GASPError> {
1598        let mut record = record.clone();
1599        record.peer.clone_from(&self.peer);
1600        record.topic.clone_from(&self.topic);
1601        record.configured = self.configured_peer;
1602        let reserved = self
1603            .storage
1604            .reserve_unanchorable_retry(&record)
1605            .await
1606            .map_err(|e| GASPError::StorageError(e.to_string()))?;
1607        self.fault_memos.lock().unwrap().remove(&record.outpoint);
1608        Ok(reserved)
1609    }
1610
1611    async fn delete_deferred_graph(&self, outpoint: &str) -> Result<(), GASPError> {
1612        self.fault_memos.lock().unwrap().remove(outpoint);
1613        self.storage
1614            .delete_deferred_graph(&self.peer, &self.topic, outpoint)
1615            .await
1616            .map_err(|e| GASPError::StorageError(e.to_string()))
1617    }
1618
1619    // Keyed by topic, not peer: the expiry is read from the outpoint's own
1620    // bytes, so any peer that serves it serves the same dead advert.
1621    async fn remember_expired(&self, outpoint: &str, expiry: u64) -> Result<(), GASPError> {
1622        self.storage
1623            .remember_expired_outpoint(&self.topic, outpoint, expiry)
1624            .await
1625            .map_err(|e| GASPError::StorageError(e.to_string()))
1626    }
1627
1628    async fn remembered_expired(&self, outpoints: &[String]) -> Result<Vec<String>, GASPError> {
1629        self.storage
1630            .find_expired_outpoints(&self.topic, outpoints)
1631            .await
1632            .map_err(|e| GASPError::StorageError(e.to_string()))
1633    }
1634}
1635
1636// ============================================================================
1637// Tests
1638// ============================================================================
1639
1640#[cfg(test)]
1641mod tests {
1642    use super::*;
1643    use crate::gasp::GASPStorage;
1644    use crate::storage::memory::MemoryStorage;
1645    use crate::storage::Storage;
1646    use crate::types::Output;
1647
1648    fn make_sink() -> FinalizedGraphSink {
1649        new_finalized_graph_sink()
1650    }
1651
1652    fn make_output(txid: &str, index: u32, topic: &str, score: f64) -> Output {
1653        Output {
1654            txid: txid.to_string(),
1655            output_index: index,
1656            output_script: vec![0x76, 0xa9],
1657            satoshis: 1000,
1658            topic: topic.to_string(),
1659            spent: false,
1660            outputs_consumed: vec![],
1661            consumed_by: vec![],
1662            beef: Some(vec![0xBE, 0xEF]),
1663            block_height: None,
1664            score: Some(score),
1665        }
1666    }
1667
1668    /// Build a minimal valid transaction hex with one input and one output.
1669    fn make_valid_tx_hex(source_txid: &str, source_oi: u32) -> String {
1670        let mut tx = Transaction::new();
1671        let input = bsv_rs::transaction::TransactionInput::new(source_txid.to_string(), source_oi);
1672        tx.inputs.push(input);
1673        tx.outputs.push(bsv_rs::transaction::TransactionOutput::new(
1674            100,
1675            bsv_rs::script::LockingScript::from_hex(
1676                "76a914000000000000000000000000000000000000000088ac",
1677            )
1678            .unwrap(),
1679        ));
1680        tx.to_hex()
1681    }
1682
1683    /// Build a coinbase-like tx (no meaningful inputs).
1684    /// A raw transaction with no input, read by `Transaction::from_hex`
1685    /// alone (no BEEF door, which refuses it since bsv-rs 0.4.1).
1686    fn make_no_input_tx_hex() -> String {
1687        let mut tx = Transaction::new();
1688        tx.outputs.push(bsv_rs::transaction::TransactionOutput::new(
1689            5000,
1690            bsv_rs::script::LockingScript::from_hex(
1691                "76a914000000000000000000000000000000000000000088ac",
1692            )
1693            .unwrap(),
1694        ));
1695        tx.to_hex()
1696    }
1697
1698    fn make_coinbase_tx_hex() -> String {
1699        let mut tx = Transaction::new();
1700        // A coinbase's null outpoint: a transaction with no input is refused
1701        // since bsv-rs 0.4.1 (NL-8 W6).
1702        let mut input = bsv_rs::transaction::TransactionInput::new("00".repeat(32), u32::MAX);
1703        input.unlocking_script =
1704            Some(bsv_rs::script::UnlockingScript::from_binary(&[0x51]).unwrap());
1705        tx.inputs.push(input);
1706        tx.outputs.push(bsv_rs::transaction::TransactionOutput::new(
1707            5000,
1708            bsv_rs::script::LockingScript::from_hex(
1709                "76a914000000000000000000000000000000000000000088ac",
1710            )
1711            .unwrap(),
1712        ));
1713        tx.to_hex()
1714    }
1715
1716    // ── find_known_utxos tests ────────────────────────────────────────
1717
1718    #[tokio::test]
1719    async fn find_known_utxos_returns_correct_outputs() {
1720        let store = MemoryStorage::new();
1721        store
1722            .insert_output(&make_output("tx1", 0, "tm_test", 100.0))
1723            .await
1724            .unwrap();
1725        store
1726            .insert_output(&make_output("tx2", 0, "tm_test", 200.0))
1727            .await
1728            .unwrap();
1729        store
1730            .insert_output(&make_output("tx3", 0, "tm_other", 300.0))
1731            .await
1732            .unwrap();
1733
1734        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
1735
1736        let utxos = gasp_storage.find_known_utxos(0, None).await.unwrap();
1737        assert_eq!(utxos.len(), 2, "Should return only tm_test UTXOs");
1738        assert_eq!(utxos[0].txid, "tx1");
1739        assert_eq!(utxos[1].txid, "tx2");
1740    }
1741
1742    #[tokio::test]
1743    async fn find_known_utxos_respects_since() {
1744        let store = MemoryStorage::new();
1745        store
1746            .insert_output(&make_output("tx1", 0, "tm_test", 100.0))
1747            .await
1748            .unwrap();
1749        store
1750            .insert_output(&make_output("tx2", 0, "tm_test", 200.0))
1751            .await
1752            .unwrap();
1753        store
1754            .insert_output(&make_output("tx3", 0, "tm_test", 300.0))
1755            .await
1756            .unwrap();
1757
1758        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
1759
1760        let utxos = gasp_storage.find_known_utxos(200, None).await.unwrap();
1761        assert_eq!(utxos.len(), 2, "Should return UTXOs with score >= 200");
1762        assert_eq!(utxos[0].txid, "tx2");
1763        assert_eq!(utxos[1].txid, "tx3");
1764    }
1765
1766    #[tokio::test]
1767    async fn find_known_utxos_respects_limit() {
1768        let store = MemoryStorage::new();
1769        store
1770            .insert_output(&make_output("tx1", 0, "tm_test", 100.0))
1771            .await
1772            .unwrap();
1773        store
1774            .insert_output(&make_output("tx2", 0, "tm_test", 200.0))
1775            .await
1776            .unwrap();
1777        store
1778            .insert_output(&make_output("tx3", 0, "tm_test", 300.0))
1779            .await
1780            .unwrap();
1781
1782        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
1783
1784        let utxos = gasp_storage.find_known_utxos(0, Some(2)).await.unwrap();
1785        assert_eq!(utxos.len(), 2, "Should limit to 2 results");
1786    }
1787
1788    #[tokio::test]
1789    async fn find_known_utxos_excludes_spent() {
1790        let store = MemoryStorage::new();
1791        store
1792            .insert_output(&make_output("tx1", 0, "tm_test", 100.0))
1793            .await
1794            .unwrap();
1795        store
1796            .insert_output(&make_output("tx2", 0, "tm_test", 200.0))
1797            .await
1798            .unwrap();
1799        store.mark_utxo_as_spent("tx1", 0, "tm_test").await.unwrap();
1800
1801        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
1802
1803        let utxos = gasp_storage.find_known_utxos(0, None).await.unwrap();
1804        assert_eq!(utxos.len(), 1, "Should exclude spent UTXOs");
1805        assert_eq!(utxos[0].txid, "tx2");
1806    }
1807
1808    #[tokio::test]
1809    async fn find_known_utxos_empty_topic() {
1810        let store = MemoryStorage::new();
1811        let gasp_storage = OverlayGASPStorage::new(&store, "tm_empty", make_sink());
1812
1813        let utxos = gasp_storage.find_known_utxos(0, None).await.unwrap();
1814        assert!(utxos.is_empty());
1815    }
1816
1817    // ── find_needed_inputs tests ──────────────────────────────────────
1818
1819    #[tokio::test]
1820    async fn find_needed_inputs_returns_none_for_unparsable_tx() {
1821        let store = MemoryStorage::new();
1822        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
1823
1824        let node = GASPNode {
1825            graph_id: "abc.0".to_string(),
1826            raw_tx: "deadbeef".to_string(), // not valid tx hex
1827            output_index: 0,
1828            proof: None,
1829            tx_metadata: None,
1830            output_metadata: None,
1831            inputs: None,
1832        };
1833
1834        let result = gasp_storage.find_needed_inputs(&node).await.unwrap();
1835        assert!(result.is_none(), "Unparsable tx: graceful fallback to None");
1836    }
1837
1838    #[tokio::test]
1839    async fn find_needed_inputs_returns_none_for_proved_tx() {
1840        let store = MemoryStorage::new();
1841        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
1842
1843        // A node with a proof (merkle path) should not need inputs.
1844        let raw_tx = make_valid_tx_hex(
1845            "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
1846            0,
1847        );
1848        let node = GASPNode {
1849            graph_id: "abc.0".to_string(),
1850            raw_tx,
1851            output_index: 0,
1852            proof: Some("some_proof_hex".to_string()),
1853            tx_metadata: None,
1854            output_metadata: None,
1855            inputs: None,
1856        };
1857
1858        let result = gasp_storage.find_needed_inputs(&node).await.unwrap();
1859        assert!(result.is_none(), "Proved tx should not need inputs");
1860    }
1861
1862    #[tokio::test]
1863    async fn find_needed_inputs_requests_inputs_for_unproved_tx() {
1864        let store = MemoryStorage::new();
1865        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
1866
1867        let source_txid = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
1868        let raw_tx = make_valid_tx_hex(source_txid, 0);
1869
1870        let node = GASPNode {
1871            graph_id: "abc.0".to_string(),
1872            raw_tx,
1873            output_index: 0,
1874            proof: None,
1875            tx_metadata: None,
1876            output_metadata: None,
1877            inputs: None,
1878        };
1879
1880        let result = gasp_storage.find_needed_inputs(&node).await.unwrap();
1881        assert!(result.is_some(), "Unproved tx should request inputs");
1882        let response = result.unwrap();
1883        assert_eq!(response.requested_inputs.len(), 1);
1884        let key = format!("{source_txid}.0");
1885        assert!(
1886            response.requested_inputs.contains_key(&key),
1887            "Should request the source input"
1888        );
1889    }
1890
1891    #[tokio::test]
1892    async fn find_needed_inputs_strips_known_inputs() {
1893        let store = MemoryStorage::new();
1894        let source_txid = "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc";
1895
1896        // Insert the source output so it's already known
1897        store
1898            .insert_output(&make_output(source_txid, 0, "tm_test", 50.0))
1899            .await
1900            .unwrap();
1901
1902        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
1903
1904        let raw_tx = make_valid_tx_hex(source_txid, 0);
1905        let node = GASPNode {
1906            graph_id: "abc.0".to_string(),
1907            raw_tx,
1908            output_index: 0,
1909            proof: None,
1910            tx_metadata: None,
1911            output_metadata: None,
1912            inputs: None,
1913        };
1914
1915        let result = gasp_storage.find_needed_inputs(&node).await.unwrap();
1916        assert!(
1917            result.is_none(),
1918            "Should strip already-known inputs, resulting in None"
1919        );
1920    }
1921
1922    #[tokio::test]
1923    async fn find_needed_inputs_returns_none_for_coinbase() {
1924        let store = MemoryStorage::new();
1925        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
1926
1927        let raw_tx = make_no_input_tx_hex();
1928        let node = GASPNode {
1929            graph_id: "abc.0".to_string(),
1930            raw_tx,
1931            output_index: 0,
1932            proof: None,
1933            tx_metadata: None,
1934            output_metadata: None,
1935            inputs: None,
1936        };
1937
1938        let result = gasp_storage.find_needed_inputs(&node).await.unwrap();
1939        assert!(
1940            result.is_none(),
1941            "Coinbase-like tx (no inputs) should need nothing"
1942        );
1943    }
1944
1945    // ── append / finalize / discard tests ─────────────────────────────
1946
1947    #[tokio::test]
1948    async fn append_and_finalize_graph() {
1949        let store = MemoryStorage::new();
1950        let sink = make_sink();
1951        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", sink.clone());
1952
1953        let raw_tx = make_coinbase_tx_hex();
1954        let node = GASPNode {
1955            graph_id: "abc.0".to_string(),
1956            raw_tx,
1957            output_index: 0,
1958            proof: None,
1959            tx_metadata: None,
1960            output_metadata: None,
1961            inputs: None,
1962        };
1963
1964        gasp_storage.append_to_graph(&node, None).await.unwrap();
1965        assert_eq!(gasp_storage.pending_graph_count(), 1);
1966
1967        gasp_storage.finalize_graph("abc.0").await.unwrap();
1968        assert_eq!(gasp_storage.pending_graph_count(), 0);
1969
1970        // Check that finalized data was produced
1971        let finalized = sink.lock().unwrap();
1972        assert_eq!(finalized.len(), 1, "Should have one finalized graph");
1973        assert_eq!(finalized[0].topic, "tm_test");
1974        assert!(!finalized[0].beefs.is_empty(), "Should have BEEF data");
1975    }
1976
1977    #[tokio::test]
1978    async fn finalize_graph_produces_submittable_beef() {
1979        let store = MemoryStorage::new();
1980        let sink = make_sink();
1981        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", sink.clone());
1982
1983        let raw_tx = make_coinbase_tx_hex();
1984        let node = GASPNode {
1985            graph_id: "root.0".to_string(),
1986            raw_tx,
1987            output_index: 0,
1988            proof: None,
1989            tx_metadata: None,
1990            output_metadata: None,
1991            inputs: None,
1992        };
1993
1994        gasp_storage.append_to_graph(&node, None).await.unwrap();
1995        gasp_storage.finalize_graph("root.0").await.unwrap();
1996
1997        let finalized = sink.lock().unwrap();
1998        assert_eq!(finalized.len(), 1);
1999
2000        // Each BEEF should be parseable
2001        for beef_bytes in &finalized[0].beefs {
2002            let tx = Transaction::from_beef(beef_bytes, None);
2003            assert!(
2004                tx.is_ok(),
2005                "Finalized BEEF should be parseable: {:?}",
2006                tx.err()
2007            );
2008        }
2009    }
2010
2011    #[tokio::test]
2012    async fn append_multiple_nodes_same_graph() {
2013        let store = MemoryStorage::new();
2014        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
2015
2016        let raw_tx1 = make_coinbase_tx_hex();
2017        let node1 = GASPNode {
2018            graph_id: "abc.0".to_string(),
2019            raw_tx: raw_tx1,
2020            output_index: 0,
2021            proof: None,
2022            tx_metadata: None,
2023            output_metadata: None,
2024            inputs: None,
2025        };
2026        let raw_tx2 = make_coinbase_tx_hex();
2027        let node2 = GASPNode {
2028            graph_id: "abc.0".to_string(),
2029            raw_tx: raw_tx2,
2030            output_index: 0,
2031            proof: None,
2032            tx_metadata: None,
2033            output_metadata: None,
2034            inputs: None,
2035        };
2036
2037        gasp_storage.append_to_graph(&node1, None).await.unwrap();
2038        gasp_storage
2039            .append_to_graph(&node2, Some("abc.0"))
2040            .await
2041            .unwrap();
2042        assert_eq!(gasp_storage.pending_graph_count(), 1);
2043    }
2044
2045    #[tokio::test]
2046    async fn discard_graph_removes_pending() {
2047        let store = MemoryStorage::new();
2048        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
2049
2050        let raw_tx = make_coinbase_tx_hex();
2051        let node = GASPNode {
2052            graph_id: "abc.0".to_string(),
2053            raw_tx,
2054            output_index: 0,
2055            proof: None,
2056            tx_metadata: None,
2057            output_metadata: None,
2058            inputs: None,
2059        };
2060
2061        gasp_storage.append_to_graph(&node, None).await.unwrap();
2062        assert_eq!(gasp_storage.pending_graph_count(), 1);
2063
2064        gasp_storage.discard_graph("abc.0").await.unwrap();
2065        assert_eq!(gasp_storage.pending_graph_count(), 0);
2066    }
2067
2068    #[tokio::test]
2069    async fn validate_graph_anchor_refuses_an_unknown_graph() {
2070        let store = MemoryStorage::new();
2071        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
2072
2073        // The reference throws `Graph node with ID ... not found`.
2074        let error = gasp_storage
2075            .validate_graph_anchor("nonexistent.0")
2076            .await
2077            .unwrap_err();
2078        assert!(matches!(error, GASPError::ValidationFailed(_)), "{error}");
2079    }
2080
2081    #[tokio::test]
2082    async fn finalize_empty_graph_errors() {
2083        let store = MemoryStorage::new();
2084        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
2085
2086        let result = gasp_storage.finalize_graph("nonexistent.0").await;
2087        assert!(result.is_err(), "Should error on unknown graph");
2088    }
2089
2090    #[tokio::test]
2091    async fn gasp_storage_is_object_safe() {
2092        let store = MemoryStorage::new();
2093        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", make_sink());
2094
2095        // Verify it can be used as dyn GASPStorage
2096        let boxed: Box<dyn GASPStorage> = Box::new(gasp_storage);
2097        let utxos = boxed.find_known_utxos(0, None).await.unwrap();
2098        assert!(utxos.is_empty());
2099    }
2100
2101    /// Integration test: OverlayGASPStorage works with GASPSync.
2102    #[tokio::test]
2103    async fn overlay_gasp_storage_with_gasp_sync() {
2104        use crate::gasp::{GASPRemote, GASPSync};
2105        use crate::types::{GASPInitialReply, GASPInitialRequest, GASPInitialResponse};
2106
2107        // Set up local storage with one UTXO
2108        let store = MemoryStorage::new();
2109        store
2110            .insert_output(&make_output("local_tx", 0, "tm_test", 50.0))
2111            .await
2112            .unwrap();
2113
2114        let sink = make_sink();
2115        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", sink.clone());
2116
2117        // Mock remote with two UTXOs — provides unparsable raw_tx so
2118        // find_needed_inputs returns None (graceful fallback).
2119        struct TestRemote;
2120
2121        #[async_trait(?Send)]
2122        impl GASPRemote for TestRemote {
2123            async fn get_initial_response(
2124                &self,
2125                request: &GASPInitialRequest,
2126            ) -> Result<GASPInitialResponse, GASPError> {
2127                Ok(GASPInitialResponse {
2128                    utxo_list: vec![
2129                        GASPOutput {
2130                            txid: "remote_tx1".to_string(),
2131                            output_index: 0,
2132                            score: 100.0,
2133                        },
2134                        GASPOutput {
2135                            txid: "remote_tx2".to_string(),
2136                            output_index: 0,
2137                            score: 200.0,
2138                        },
2139                    ],
2140                    since: request.since,
2141                })
2142            }
2143            async fn get_initial_reply(
2144                &self,
2145                _: &GASPInitialResponse,
2146            ) -> Result<GASPInitialReply, GASPError> {
2147                Ok(GASPInitialReply {
2148                    utxo_list: Vec::new(),
2149                })
2150            }
2151            async fn request_node(
2152                &self,
2153                graph_id: &str,
2154                txid: &str,
2155                output_index: u32,
2156                _: bool,
2157            ) -> Result<GASPNode, GASPError> {
2158                Ok(GASPNode {
2159                    graph_id: graph_id.to_string(),
2160                    raw_tx: format!("rawtx_{txid}"),
2161                    output_index,
2162                    proof: None,
2163                    tx_metadata: None,
2164                    output_metadata: None,
2165                    inputs: None,
2166                })
2167            }
2168            async fn submit_node(
2169                &self,
2170                _: &GASPNode,
2171            ) -> Result<Option<GASPNodeResponse>, GASPError> {
2172                Ok(None)
2173            }
2174        }
2175
2176        let mut sync = GASPSync::new(
2177            Box::new(gasp_storage),
2178            Box::new(TestRemote),
2179            0,
2180            "[TEST]",
2181            true, // unidirectional
2182        );
2183
2184        sync.sync(None).await.unwrap();
2185        // The synthetic graph cannot assemble a BEEF. Nothing completed,
2186        // so the cursor remains at entry and both outputs are retried.
2187        assert_eq!(sync.last_interaction, 0);
2188    }
2189
2190    /// Integration test: full sync with valid transactions produces submittable BEEFs.
2191    #[tokio::test]
2192    async fn gasp_sync_with_valid_tx_produces_finalized_beefs() {
2193        use crate::gasp::{GASPRemote, GASPSync};
2194        use crate::types::{GASPInitialReply, GASPInitialRequest, GASPInitialResponse};
2195
2196        let store = MemoryStorage::new();
2197        let sink = make_sink();
2198        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", sink.clone());
2199
2200        // Build a valid coinbase-style tx for the remote to provide.
2201        let valid_tx_hex = make_coinbase_tx_hex();
2202
2203        struct ValidTxRemote {
2204            tx_hex: String,
2205        }
2206
2207        #[async_trait(?Send)]
2208        impl GASPRemote for ValidTxRemote {
2209            async fn get_initial_response(
2210                &self,
2211                request: &GASPInitialRequest,
2212            ) -> Result<GASPInitialResponse, GASPError> {
2213                Ok(GASPInitialResponse {
2214                    utxo_list: vec![GASPOutput {
2215                        txid: "valid_tx".to_string(),
2216                        output_index: 0,
2217                        score: 100.0,
2218                    }],
2219                    since: request.since,
2220                })
2221            }
2222            async fn get_initial_reply(
2223                &self,
2224                _: &GASPInitialResponse,
2225            ) -> Result<GASPInitialReply, GASPError> {
2226                Ok(GASPInitialReply {
2227                    utxo_list: Vec::new(),
2228                })
2229            }
2230            async fn request_node(
2231                &self,
2232                graph_id: &str,
2233                _txid: &str,
2234                output_index: u32,
2235                _: bool,
2236            ) -> Result<GASPNode, GASPError> {
2237                // Served PROVEN: since bsv-low #551 the anchor check refuses
2238                // an unproven transaction that creates satoshis (this one has
2239                // no inputs). With no chain tracker a merkle path is accepted
2240                // unchecked, the reference's 'scripts only'.
2241                let txid = Transaction::from_hex(&self.tx_hex).unwrap().id();
2242                let proof = MerklePath::new(
2243                    100,
2244                    vec![vec![bsv_rs::transaction::MerklePathLeaf::new_txid(0, txid)]],
2245                )
2246                .unwrap();
2247                Ok(GASPNode {
2248                    graph_id: graph_id.to_string(),
2249                    raw_tx: self.tx_hex.clone(),
2250                    output_index,
2251                    proof: Some(proof.to_hex()),
2252                    tx_metadata: None,
2253                    output_metadata: None,
2254                    inputs: None,
2255                })
2256            }
2257            async fn submit_node(
2258                &self,
2259                _: &GASPNode,
2260            ) -> Result<Option<GASPNodeResponse>, GASPError> {
2261                Ok(None)
2262            }
2263        }
2264
2265        let remote = ValidTxRemote {
2266            tx_hex: valid_tx_hex,
2267        };
2268
2269        let mut sync = GASPSync::new(Box::new(gasp_storage), Box::new(remote), 0, "[TEST]", true);
2270
2271        sync.sync(None).await.unwrap();
2272
2273        let finalized = sink.lock().unwrap();
2274        assert_eq!(finalized.len(), 1, "Should have one finalized graph");
2275        assert!(!finalized[0].beefs.is_empty(), "Should have BEEF bytes");
2276
2277        // Verify the BEEF is parseable
2278        for beef_bytes in &finalized[0].beefs {
2279            let parsed = Transaction::from_beef(beef_bytes, None);
2280            assert!(parsed.is_ok(), "BEEF should be parseable");
2281        }
2282    }
2283
2284    #[tokio::test]
2285    async fn take_finalized_graphs_drains_list() {
2286        let store = MemoryStorage::new();
2287        let sink = make_sink();
2288        let gasp_storage = OverlayGASPStorage::new(&store, "tm_test", sink.clone());
2289
2290        let raw_tx = make_coinbase_tx_hex();
2291        let node = GASPNode {
2292            graph_id: "abc.0".to_string(),
2293            raw_tx,
2294            output_index: 0,
2295            proof: None,
2296            tx_metadata: None,
2297            output_metadata: None,
2298            inputs: None,
2299        };
2300
2301        gasp_storage.append_to_graph(&node, None).await.unwrap();
2302        gasp_storage.finalize_graph("abc.0").await.unwrap();
2303
2304        let taken = gasp_storage.take_finalized_graphs();
2305        assert_eq!(taken.len(), 1);
2306
2307        // Should be empty after draining
2308        let taken2 = gasp_storage.take_finalized_graphs();
2309        assert!(taken2.is_empty());
2310    }
2311
2312    // ── bsv-low #586: the assembly gives the bytes the hydrated tree gave ──
2313
2314    /// `get_beef_for_node` as it was on `fbb7fa8`, verbatim: the hydrating
2315    /// recursion [`GraphAssembly`] replaced. Kept as the reference the
2316    /// assembly is compared with, byte for byte and error for error.
2317    fn legacy_get_beef_for_node(
2318        node_key: &str,
2319        refs: &HashMap<String, PendingNode>,
2320        strict_beef: bool,
2321    ) -> Result<(Transaction, Vec<u8>), GASPError> {
2322        let pending = refs
2323            .get(node_key)
2324            .ok_or_else(|| GASPError::Other(format!("Node {node_key} not found in graph refs")))?;
2325
2326        let mut tx = Transaction::from_hex(&pending.node.raw_tx)
2327            .map_err(|e| GASPError::Other(format!("Failed to parse raw_tx for {node_key}: {e}")))?;
2328
2329        if let Some(ref proof_hex) = pending.node.proof {
2330            tx.merkle_path = Some(beef_limits::proof_from_hex(proof_hex).map_err(|e| {
2331                GASPError::Other(format!("Failed to parse proof for {node_key}: {e}"))
2332            })?);
2333        } else {
2334            let child_info: Vec<(usize, String)> = tx
2335                .inputs
2336                .iter()
2337                .enumerate()
2338                .filter_map(|(idx, input)| {
2339                    let source_txid = input.get_source_txid().unwrap_or_default();
2340                    if source_txid.is_empty() {
2341                        return None;
2342                    }
2343                    let child_key = format!("{}.{}", source_txid, input.source_output_index);
2344                    if refs.contains_key(&child_key) {
2345                        Some((idx, child_key))
2346                    } else {
2347                        None
2348                    }
2349                })
2350                .collect();
2351
2352            for (input_idx, child_key) in child_info {
2353                let (child_tx, _) = legacy_get_beef_for_node(&child_key, refs, strict_beef)?;
2354                tx.inputs[input_idx].source_transaction = Some(Box::new(child_tx));
2355            }
2356        }
2357
2358        let allow_partial = !strict_beef;
2359        let beef = tx.to_beef(allow_partial).map_err(|e| {
2360            GASPError::Other(format!("Failed to serialize BEEF for {node_key}: {e}"))
2361        })?;
2362
2363        Ok((tx, beef))
2364    }
2365
2366    /// `compute_ordered_beefs` as it was on `fbb7fa8`, verbatim.
2367    fn legacy_compute_ordered_beefs(
2368        graph_id: &str,
2369        refs: &HashMap<String, PendingNode>,
2370        strict_beef: bool,
2371    ) -> Result<Vec<Vec<u8>>, GASPError> {
2372        let mut beefs: Vec<Vec<u8>> = Vec::new();
2373        let mut visited: std::collections::HashSet<String> = std::collections::HashSet::new();
2374
2375        fn hydrate(
2376            node_key: &str,
2377            refs: &HashMap<String, PendingNode>,
2378            beefs: &mut Vec<Vec<u8>>,
2379            visited: &mut std::collections::HashSet<String>,
2380            strict_beef: bool,
2381        ) -> Result<(), GASPError> {
2382            if visited.contains(node_key) {
2383                return Ok(());
2384            }
2385            visited.insert(node_key.to_string());
2386
2387            let Some(pending) = refs.get(node_key) else {
2388                return Ok(());
2389            };
2390
2391            for child_key in &pending.children {
2392                hydrate(child_key, refs, beefs, visited, strict_beef)?;
2393            }
2394
2395            let (_, beef) = legacy_get_beef_for_node(node_key, refs, strict_beef)?;
2396            beefs.push(beef);
2397            Ok(())
2398        }
2399
2400        hydrate(graph_id, refs, &mut beefs, &mut visited, strict_beef)?;
2401        Ok(beefs)
2402    }
2403
2404    /// A small deterministic generator (no dependency, the same graphs on
2405    /// every run).
2406    struct Lcg(u64);
2407
2408    impl Lcg {
2409        fn below(&mut self, n: u64) -> u64 {
2410            self.0 = self
2411                .0
2412                .wrapping_mul(6_364_136_223_846_793_005)
2413                .wrapping_add(1_442_695_040_888_963_407);
2414            (self.0 >> 33) % n
2415        }
2416    }
2417
2418    fn pending(raw_tx: String, output_index: u32, proof: Option<String>) -> PendingNode {
2419        PendingNode {
2420            node: GASPNode {
2421                graph_id: "g".to_string(),
2422                raw_tx,
2423                output_index,
2424                proof,
2425                tx_metadata: None,
2426                output_metadata: None,
2427                inputs: None,
2428            },
2429            spent_by: None,
2430            children: Vec::new(),
2431        }
2432    }
2433
2434    /// A random pending graph: up to 9 transactions, each spending up to
2435    /// four outputs of earlier ones (two outputs of one transaction too, the
2436    /// diamond), some proven (two of a pair sharing one block, so their
2437    /// proofs COMBINE), some referenced outpoints absent from the graph (a
2438    /// partial BEEF, a strict refusal), and now and then a proof or a raw
2439    /// transaction that does not parse. Returns the graph and its root key.
2440    fn random_graph(rng: &mut Lcg) -> (HashMap<String, PendingNode>, String) {
2441        use bsv_rs::transaction::{MerklePathLeaf, TransactionInput, TransactionOutput};
2442        let count = 2 + rng.below(8) as usize;
2443        let mut txs: Vec<Transaction> = Vec::new();
2444        for i in 0..count {
2445            let mut tx = Transaction::new();
2446            let mut spent: HashSet<(usize, u32)> = HashSet::new();
2447            if i > 0 {
2448                for _ in 0..rng.below(5) {
2449                    let source = rng.below(i as u64) as usize;
2450                    let vout = rng.below(3) as u32;
2451                    if spent.insert((source, vout)) {
2452                        tx.inputs
2453                            .push(TransactionInput::new(txs[source].id(), vout));
2454                    }
2455                }
2456            }
2457            for _ in 0..3 {
2458                tx.outputs.push(TransactionOutput::new(
2459                    1000 + i as u64,
2460                    bsv_rs::script::LockingScript::from_hex("51").unwrap(),
2461                ));
2462            }
2463            txs.push(tx);
2464        }
2465        // Proofs: a transaction alone in its block, or a pair in one block.
2466        let mut proofs: Vec<Option<String>> = vec![None; count];
2467        let mut i = 0;
2468        while i < count {
2469            match rng.below(5) {
2470                0 => {
2471                    let path = vec![vec![MerklePathLeaf::new_txid(0, txs[i].id())]];
2472                    proofs[i] = Some(MerklePath::new(100 + i as u32, path).unwrap().to_hex());
2473                }
2474                1 if i + 1 < count => {
2475                    let (a, b) = (txs[i].id(), txs[i + 1].id());
2476                    let height = 100 + i as u32;
2477                    let of_a = vec![vec![
2478                        MerklePathLeaf::new_txid(0, a.clone()),
2479                        MerklePathLeaf::new(1, b.clone()),
2480                    ]];
2481                    let of_b = vec![vec![
2482                        MerklePathLeaf::new(0, a),
2483                        MerklePathLeaf::new_txid(1, b),
2484                    ]];
2485                    proofs[i] = Some(MerklePath::new(height, of_a).unwrap().to_hex());
2486                    proofs[i + 1] = Some(MerklePath::new(height, of_b).unwrap().to_hex());
2487                    i += 1;
2488                }
2489                _ => {}
2490            }
2491            i += 1;
2492        }
2493        let root_key = format!("{}.0", txs[count - 1].id());
2494        let mut refs: HashMap<String, PendingNode> = HashMap::new();
2495        refs.insert(
2496            root_key.clone(),
2497            pending(txs[count - 1].to_hex(), 0, proofs[count - 1].clone()),
2498        );
2499        // Every referenced outpoint is a node, most of the time; a node is
2500        // a child of the node(s) of its spender.
2501        for (i, tx) in txs.iter().enumerate().rev() {
2502            let own_keys: Vec<String> = (0..3)
2503                .map(|vout| format!("{}.{vout}", tx.id()))
2504                .filter(|key| refs.contains_key(key))
2505                .collect();
2506            if own_keys.is_empty() || proofs[i].is_some() && rng.below(3) > 0 {
2507                continue;
2508            }
2509            for input in &tx.inputs {
2510                let source_txid = input.get_source_txid().unwrap();
2511                let source = txs.iter().position(|t| t.id() == source_txid).unwrap();
2512                let key = format!("{source_txid}.{}", input.source_output_index);
2513                if rng.below(6) == 0 {
2514                    continue;
2515                }
2516                let mut node = pending(
2517                    txs[source].to_hex(),
2518                    input.source_output_index,
2519                    proofs[source].clone(),
2520                );
2521                match rng.below(40) {
2522                    0 => node.node.proof = Some("zz".to_string()),
2523                    1 => node.node.raw_tx = "00".to_string(),
2524                    _ => {}
2525                }
2526                refs.insert(key.clone(), node);
2527                for own in &own_keys {
2528                    let children = &mut refs.get_mut(own).unwrap().children;
2529                    if !children.contains(&key) {
2530                        children.push(key.clone());
2531                    }
2532                }
2533            }
2534        }
2535        (refs, root_key)
2536    }
2537
2538    // The assembly against the recursion it replaced, over 600 random
2539    // graphs, tolerant and strict: the BEEF of every node, the ordered BEEFs
2540    // of the graph, and every refusal, are the same bytes and the same words.
2541    #[test]
2542    fn e586_the_assembly_gives_the_hydrated_trees_bytes_and_errors() {
2543        let mut rng = Lcg(0x586);
2544        let (mut beefs, mut refusals, mut proven, mut partial) = (0, 0, 0, 0);
2545        for _ in 0..600 {
2546            let (refs, root_key) = random_graph(&mut rng);
2547            for strict in [false, true] {
2548                let mut keys: Vec<&String> = refs.keys().collect();
2549                keys.sort();
2550                for key in keys {
2551                    let legacy = legacy_get_beef_for_node(key, &refs, strict)
2552                        .map(|(_, beef)| beef)
2553                        .map_err(|e| e.to_string());
2554                    // A fresh assembly per node, and one shared by the graph
2555                    // below (`compute_ordered_beefs`): both orders of use.
2556                    let mut assembly = GraphAssembly::default();
2557                    let ours = assembly
2558                        .check(key, &refs, strict)
2559                        .map(|()| assembly.beef_of(key))
2560                        .map_err(|e| e.to_string());
2561                    assert_eq!(ours, legacy, "node {key}, strict {strict}");
2562                    match &ours {
2563                        Ok(beef) => {
2564                            beefs += 1;
2565                            let parsed = Beef::from_binary(beef).unwrap();
2566                            proven += usize::from(!parsed.bumps.is_empty());
2567                            partial += usize::from(
2568                                !strict && legacy_get_beef_for_node(key, &refs, true).is_err(),
2569                            );
2570                        }
2571                        Err(_) => refusals += 1,
2572                    }
2573                }
2574                let legacy = legacy_compute_ordered_beefs(&root_key, &refs, strict)
2575                    .map_err(|e| e.to_string());
2576                let ours =
2577                    OverlayGASPStorage::compute_ordered_beefs(&root_key, &refs, strict, None)
2578                        .map_err(|e| e.to_string());
2579                assert_eq!(ours, legacy, "graph {root_key}, strict {strict}");
2580            }
2581        }
2582        println!(
2583            "#586: {beefs} BEEFs equal ({proven} with a proof, {partial} partial), {refusals} refusals equal"
2584        );
2585        // The generator reached every class.
2586        assert!(beefs > 2000 && proven > 300 && partial > 100 && refusals > 300);
2587    }
2588
2589    // A node unknown to the graph, and the SDK's own words for a missing
2590    // source under `strict_beef`.
2591    #[test]
2592    fn e586_the_refusals_are_the_recursions_words() {
2593        let parent = "ab".repeat(32);
2594        let raw = make_valid_tx_hex(&parent, 3);
2595        let key = format!("{}.0", Transaction::from_hex(&raw).unwrap().id());
2596        let refs = HashMap::from([(key.clone(), pending(raw, 0, None))]);
2597        let mut assembly = GraphAssembly::default();
2598        assert_eq!(
2599            assembly
2600                .check("nope.0", &refs, false)
2601                .unwrap_err()
2602                .to_string(),
2603            legacy_get_beef_for_node("nope.0", &refs, false)
2604                .unwrap_err()
2605                .to_string()
2606        );
2607        let strict = assembly.check(&key, &refs, true).unwrap_err().to_string();
2608        assert_eq!(
2609            strict,
2610            legacy_get_beef_for_node(&key, &refs, true)
2611                .unwrap_err()
2612                .to_string()
2613        );
2614        assert!(strict.contains(&format!(
2615            "Missing source transaction for input 0 (txid: {parent})"
2616        )));
2617        // Tolerant, the same node assembles: a partial BEEF of one transaction.
2618        assembly.check(&key, &refs, false).unwrap();
2619        assert_eq!(
2620            assembly.beef_of(&key),
2621            legacy_get_beef_for_node(&key, &refs, false).unwrap().1
2622        );
2623    }
2624
2625    // The walks keep their work on the heap. A chain of 20,000 unproven
2626    // links is assembled; the recursion it replaced used the stack once per
2627    // link. And a link back into a node the walk is still inside (the root
2628    // is filed under the graph id the PEER names, so a peer can file a root
2629    // under an outpoint its own ancestry spends) ends there: the recursion
2630    // never returned.
2631    #[test]
2632    fn e586_a_deep_chain_and_a_cycle_do_not_use_the_stack() {
2633        use bsv_rs::transaction::{TransactionInput, TransactionOutput};
2634        let mut refs: HashMap<String, PendingNode> = HashMap::new();
2635        let mut previous: Option<String> = None;
2636        let mut tip = String::new();
2637        for i in 0..20_000u64 {
2638            let mut tx = Transaction::new();
2639            if let Some(txid) = &previous {
2640                tx.inputs.push(TransactionInput::new(txid.clone(), 0));
2641            }
2642            tx.outputs.push(TransactionOutput::new(
2643                i,
2644                bsv_rs::script::LockingScript::from_hex("51").unwrap(),
2645            ));
2646            let txid = tx.id();
2647            tip = format!("{txid}.0");
2648            let mut node = pending(tx.to_hex(), 0, None);
2649            if let Some(below) = &previous {
2650                node.children.push(format!("{below}.0"));
2651            }
2652            refs.insert(tip.clone(), node);
2653            previous = Some(txid);
2654        }
2655        assert_eq!(OverlayGASPStorage::ordered_keys(&tip, &refs).len(), 20_000);
2656        let mut assembly = GraphAssembly::default();
2657        assembly.check(&tip, &refs, true).unwrap();
2658        let beef = Beef::from_binary(&assembly.beef_of(&tip)).unwrap();
2659        assert_eq!(beef.txs.len(), 20_000);
2660
2661        // root (filed under `decoy.0`) spends y.0; y spends `decoy.0`.
2662        let decoy = "cd".repeat(32);
2663        let y_raw = make_valid_tx_hex(&decoy, 0);
2664        let y = Transaction::from_hex(&y_raw).unwrap().id();
2665        let root_raw = make_valid_tx_hex(&y, 0);
2666        let root_key = format!("{decoy}.0");
2667        let mut root = pending(root_raw, 0, None);
2668        root.children.push(format!("{y}.0"));
2669        let refs = HashMap::from([
2670            (root_key.clone(), root),
2671            (format!("{y}.0"), pending(y_raw, 0, None)),
2672        ]);
2673        let mut assembly = GraphAssembly::default();
2674        assembly.check(&root_key, &refs, true).unwrap();
2675        let beef = Beef::from_binary(&assembly.beef_of(&root_key)).unwrap();
2676        assert_eq!(beef.txs.len(), 2);
2677        let beefs =
2678            OverlayGASPStorage::compute_ordered_beefs(&root_key, &refs, true, None).unwrap();
2679        assert_eq!(beefs.len(), 2);
2680    }
2681}