Implement presumed_origin

Before we yield a block for scanning, we save all of the contained script
public keys. Then, when we want the address credited for creating an output,
we read the script public key of the spent output from the database.

Fixes #559.

processor/bitcoin/src/block.rs

@@ -1,3 +1,4 @@
+use core::fmt;
 use std::collections::HashMap;
 
 use ciphersuite::{Ciphersuite, Secp256k1};
@@ -6,6 +7,7 @@ use bitcoin_serai::bitcoin::block::{Header, Block as BBlock};
 
 use serai_client::networks::bitcoin::Address;
 
+use serai_db::Db;
 use primitives::{ReceivedOutput, EventualityTracker};
 
 use crate::{hash_bytes, scan::scanner, output::Output, transaction::Eventuality};
@@ -21,11 +23,16 @@ impl primitives::BlockHeader for BlockHeader {
   }
 }
 
-#[derive(Clone, Debug)]
-pub(crate) struct Block(pub(crate) BBlock);
+#[derive(Clone)]
+pub(crate) struct Block<D: Db>(pub(crate) D, pub(crate) BBlock);
+impl<D: Db> fmt::Debug for Block<D> {
+  fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
+    fmt.debug_struct("Block").field("1", &self.1).finish_non_exhaustive()
+  }
+}
 
 #[async_trait::async_trait]
-impl primitives::Block for Block {
+impl<D: Db> primitives::Block for Block<D> {
   type Header = BlockHeader;
 
   type Key = <Secp256k1 as Ciphersuite>::G;
@@ -34,17 +41,17 @@ impl primitives::Block for Block {
   type Eventuality = Eventuality;
 
   fn id(&self) -> [u8; 32] {
-    primitives::BlockHeader::id(&BlockHeader(self.0.header))
+    primitives::BlockHeader::id(&BlockHeader(self.1.header))
   }
 
   fn scan_for_outputs_unordered(&self, key: Self::Key) -> Vec<Self::Output> {
     let scanner = scanner(key);
 
     let mut res = vec![];
     // We skip the coinbase transaction as its burdened by maturity
-    for tx in &self.0.txdata[1 ..] {
+    for tx in &self.1.txdata[1 ..] {
       for output in scanner.scan_transaction(tx) {
-        res.push(Output::new(key, tx, output));
+        res.push(Output::new(&self.0, key, tx, output));
       }
     }
     res
@@ -59,7 +66,7 @@ impl primitives::Block for Block {
     Self::Eventuality,
   > {
     let mut res = HashMap::new();
-    for tx in &self.0.txdata[1 ..] {
+    for tx in &self.1.txdata[1 ..] {
       let id = hash_bytes(tx.compute_txid().to_raw_hash());
       if let Some(eventuality) = eventualities.active_eventualities.remove(id.as_slice()) {
         res.insert(id, eventuality);

processor/bitcoin/src/db.rs

@@ -0,0 +1,8 @@
+use serai_db::{Get, DbTxn, create_db};
+
+create_db! {
+  BitcoinProcessor {
+    LatestBlockToYieldAsFinalized: () -> u64,
+    ScriptPubKey: (tx: [u8; 32], vout: u32) -> Vec<u8>,
+  }
+}

processor/bitcoin/src/main.rs

@@ -18,6 +18,10 @@ mod block;
 mod rpc;
 mod scheduler;
 
+// Our custom code for Bitcoin
+mod db;
+mod txindex;
+
 pub(crate) fn hash_bytes(hash: bitcoin_serai::bitcoin::hashes::sha256d::Hash) -> [u8; 32] {
   use bitcoin_serai::bitcoin::hashes::Hash;
 

processor/bitcoin/src/output.rs

@@ -15,6 +15,7 @@ use bitcoin_serai::{
 
 use scale::{Encode, Decode, IoReader};
 use borsh::{BorshSerialize, BorshDeserialize};
+use serai_db::Get;
 
 use serai_client::{
   primitives::{Coin, Amount, Balance, ExternalAddress},
@@ -52,13 +53,35 @@ pub(crate) struct Output {
 }
 
 impl Output {
-  pub fn new(key: <Secp256k1 as Ciphersuite>::G, tx: &Transaction, output: WalletOutput) -> Self {
+  pub fn new(
+    getter: &impl Get,
+    key: <Secp256k1 as Ciphersuite>::G,
+    tx: &Transaction,
+    output: WalletOutput,
+  ) -> Self {
     Self {
       kind: offsets_for_key(key)
         .into_iter()
         .find_map(|(kind, offset)| (offset == output.offset()).then_some(kind))
         .expect("scanned output for unknown offset"),
-      presumed_origin: presumed_origin(tx),
+      presumed_origin: presumed_origin(getter, tx),
+      output,
+      data: extract_serai_data(tx),
+    }
+  }
+
+  pub fn new_with_presumed_origin(
+    key: <Secp256k1 as Ciphersuite>::G,
+    tx: &Transaction,
+    presumed_origin: Option<Address>,
+    output: WalletOutput,
+  ) -> Self {
+    Self {
+      kind: offsets_for_key(key)
+        .into_iter()
+        .find_map(|(kind, offset)| (offset == output.offset()).then_some(kind))
+        .expect("scanned output for unknown offset"),
+      presumed_origin,
       output,
       data: extract_serai_data(tx),
     }

processor/bitcoin/src/rpc.rs

@@ -2,50 +2,55 @@ use bitcoin_serai::rpc::{RpcError, Rpc as BRpc};
 
 use serai_client::primitives::{NetworkId, Coin, Amount};
 
+use serai_db::Db;
 use scanner::ScannerFeed;
 use signers::TransactionPublisher;
 
 use crate::{
+  db,
   transaction::Transaction,
   block::{BlockHeader, Block},
 };
 
 #[derive(Clone)]
-pub(crate) struct Rpc(BRpc);
+pub(crate) struct Rpc<D: Db> {
+  pub(crate) db: D,
+  pub(crate) rpc: BRpc,
+}
 
 #[async_trait::async_trait]
-impl ScannerFeed for Rpc {
+impl<D: Db> ScannerFeed for Rpc<D> {
   const NETWORK: NetworkId = NetworkId::Bitcoin;
   const CONFIRMATIONS: u64 = 6;
   const WINDOW_LENGTH: u64 = 6;
 
   const TEN_MINUTES: u64 = 1;
 
-  type Block = Block;
+  type Block = Block<D>;
 
   type EphemeralError = RpcError;
 
   async fn latest_finalized_block_number(&self) -> Result<u64, Self::EphemeralError> {
-    u64::try_from(self.0.get_latest_block_number().await?)
-      .unwrap()
-      .checked_sub(Self::CONFIRMATIONS)
-      .ok_or(RpcError::ConnectionError)
+    db::LatestBlockToYieldAsFinalized::get(&self.db).ok_or(RpcError::ConnectionError)
   }
 
   async fn unchecked_block_header_by_number(
     &self,
     number: u64,
   ) -> Result<<Self::Block as primitives::Block>::Header, Self::EphemeralError> {
     Ok(BlockHeader(
-      self.0.get_block(&self.0.get_block_hash(number.try_into().unwrap()).await?).await?.header,
+      self.rpc.get_block(&self.rpc.get_block_hash(number.try_into().unwrap()).await?).await?.header,
     ))
   }
 
   async fn unchecked_block_by_number(
     &self,
     number: u64,
   ) -> Result<Self::Block, Self::EphemeralError> {
-    Ok(Block(self.0.get_block(&self.0.get_block_hash(number.try_into().unwrap()).await?).await?))
+    Ok(Block(
+      self.db.clone(),
+      self.rpc.get_block(&self.rpc.get_block_hash(number.try_into().unwrap()).await?).await?,
+    ))
   }
 
   fn dust(coin: Coin) -> Amount {
@@ -98,10 +103,10 @@ impl ScannerFeed for Rpc {
 }
 
 #[async_trait::async_trait]
-impl TransactionPublisher<Transaction> for Rpc {
+impl<D: Db> TransactionPublisher<Transaction> for Rpc<D> {
   type EphemeralError = RpcError;
 
   async fn publish(&self, tx: Transaction) -> Result<(), Self::EphemeralError> {
-    self.0.send_raw_transaction(&tx.0).await.map(|_| ())
+    self.rpc.send_raw_transaction(&tx.0).await.map(|_| ())
   }
 }

processor/bitcoin/src/scan.rs

@@ -13,8 +13,11 @@ use bitcoin_serai::{
 
 use serai_client::networks::bitcoin::Address;
 
+use serai_db::Get;
 use primitives::OutputType;
 
+use crate::{db, hash_bytes};
+
 const KEY_DST: &[u8] = b"Serai Bitcoin Processor Key Offset";
 static BRANCH_BASE_OFFSET: LazyLock<<Secp256k1 as Ciphersuite>::F> =
   LazyLock::new(|| Secp256k1::hash_to_F(KEY_DST, b"branch"));
@@ -55,26 +58,17 @@ pub(crate) fn scanner(key: <Secp256k1 as Ciphersuite>::G) -> Scanner {
   scanner
 }
 
-pub(crate) fn presumed_origin(tx: &Transaction) -> Option<Address> {
-  todo!("TODO")
-
-  /*
-  let spent_output = {
-    let input = &tx.input[0];
-    let mut spent_tx = input.previous_output.txid.as_raw_hash().to_byte_array();
-    spent_tx.reverse();
-    let mut tx;
-    while {
-      tx = self.rpc.get_transaction(&spent_tx).await;
-      tx.is_err()
-    } {
-      log::error!("couldn't get transaction from bitcoin node: {tx:?}");
-      sleep(Duration::from_secs(5)).await;
+pub(crate) fn presumed_origin(getter: &impl Get, tx: &Transaction) -> Option<Address> {
+  for input in &tx.input {
+    let txid = hash_bytes(input.previous_output.txid.to_raw_hash());
+    let vout = input.previous_output.vout;
+    if let Some(address) = Address::new(ScriptBuf::from_bytes(
+      db::ScriptPubKey::get(getter, txid, vout).expect("unknown output being spent by input"),
+    )) {
+      return Some(address);
     }
-    tx.unwrap().output.swap_remove(usize::try_from(input.previous_output.vout).unwrap())
-  };
-  Address::new(spent_output.script_pubkey)
-  */
+  }
+  None?
 }
 
 // Checks if this script matches SHA256 PUSH MSG_HASH OP_EQUALVERIFY ..

processor/bitcoin/src/scheduler.rs

@@ -10,6 +10,7 @@ use serai_client::{
   networks::bitcoin::Address,
 };
 
+use serai_db::Db;
 use primitives::{OutputType, ReceivedOutput, Payment};
 use scanner::{KeyFor, AddressFor, OutputFor, BlockFor};
 use utxo_scheduler::{PlannedTransaction, TransactionPlanner};
@@ -31,17 +32,24 @@ fn address_from_serai_key(key: <Secp256k1 as Ciphersuite>::G, kind: OutputType)
   .expect("couldn't create Serai-representable address for P2TR script")
 }
 
-fn signable_transaction(
+fn signable_transaction<D: Db>(
   fee_per_vbyte: u64,
-  inputs: Vec<OutputFor<Rpc>>,
-  payments: Vec<Payment<AddressFor<Rpc>>>,
-  change: Option<KeyFor<Rpc>>,
+  inputs: Vec<OutputFor<Rpc<D>>>,
+  payments: Vec<Payment<AddressFor<Rpc<D>>>>,
+  change: Option<KeyFor<Rpc<D>>>,
 ) -> Result<(SignableTransaction, BSignableTransaction), TransactionError> {
-  assert!(inputs.len() < Planner::MAX_INPUTS);
-  assert!((payments.len() + usize::from(u8::from(change.is_some()))) < Planner::MAX_OUTPUTS);
+  assert!(
+    inputs.len() <
+      <Planner as TransactionPlanner<Rpc<D>, EffectedReceivedOutputs<Rpc<D>>>>::MAX_INPUTS
+  );
+  assert!(
+    (payments.len() + usize::from(u8::from(change.is_some()))) <
+      <Planner as TransactionPlanner<Rpc<D>, EffectedReceivedOutputs<Rpc<D>>>>::MAX_OUTPUTS
+  );
 
   let inputs = inputs.into_iter().map(|input| input.output).collect::<Vec<_>>();
-  let payments = payments
+
+  let mut payments = payments
     .into_iter()
     .map(|payment| {
       (payment.address().clone(), {
@@ -51,7 +59,8 @@ fn signable_transaction(
       })
     })
     .collect::<Vec<_>>();
-  let change = change.map(Planner::change_address);
+  let change = change
+    .map(<Planner as TransactionPlanner<Rpc<D>, EffectedReceivedOutputs<Rpc<D>>>>::change_address);
 
   // TODO: ACP output
   BSignableTransaction::new(
@@ -69,7 +78,7 @@ fn signable_transaction(
 }
 
 pub(crate) struct Planner;
-impl TransactionPlanner<Rpc, EffectedReceivedOutputs<Rpc>> for Planner {
+impl<D: Db> TransactionPlanner<Rpc<D>, EffectedReceivedOutputs<Rpc<D>>> for Planner {
   type FeeRate = u64;
 
   type SignableTransaction = SignableTransaction;
@@ -94,29 +103,29 @@ impl TransactionPlanner<Rpc, EffectedReceivedOutputs<Rpc>> for Planner {
   // to unstick any transactions which had too low of a fee.
   const MAX_OUTPUTS: usize = 519;
 
-  fn fee_rate(block: &BlockFor<Rpc>, coin: Coin) -> Self::FeeRate {
+  fn fee_rate(block: &BlockFor<Rpc<D>>, coin: Coin) -> Self::FeeRate {
     assert_eq!(coin, Coin::Bitcoin);
     // TODO
     1
   }
 
-  fn branch_address(key: KeyFor<Rpc>) -> AddressFor<Rpc> {
+  fn branch_address(key: KeyFor<Rpc<D>>) -> AddressFor<Rpc<D>> {
     address_from_serai_key(key, OutputType::Branch)
   }
-  fn change_address(key: KeyFor<Rpc>) -> AddressFor<Rpc> {
+  fn change_address(key: KeyFor<Rpc<D>>) -> AddressFor<Rpc<D>> {
     address_from_serai_key(key, OutputType::Change)
   }
-  fn forwarding_address(key: KeyFor<Rpc>) -> AddressFor<Rpc> {
+  fn forwarding_address(key: KeyFor<Rpc<D>>) -> AddressFor<Rpc<D>> {
     address_from_serai_key(key, OutputType::Forwarded)
   }
 
   fn calculate_fee(
     fee_rate: Self::FeeRate,
-    inputs: Vec<OutputFor<Rpc>>,
-    payments: Vec<Payment<AddressFor<Rpc>>>,
-    change: Option<KeyFor<Rpc>>,
+    inputs: Vec<OutputFor<Rpc<D>>>,
+    payments: Vec<Payment<AddressFor<Rpc<D>>>>,
+    change: Option<KeyFor<Rpc<D>>>,
   ) -> Amount {
-    match signable_transaction(fee_rate, inputs, payments, change) {
+    match signable_transaction::<D>(fee_rate, inputs, payments, change) {
       Ok(tx) => Amount(tx.1.needed_fee()),
       Err(
         TransactionError::NoInputs | TransactionError::NoOutputs | TransactionError::DustPayment,
@@ -133,17 +142,17 @@ impl TransactionPlanner<Rpc, EffectedReceivedOutputs<Rpc>> for Planner {
 
   fn plan(
     fee_rate: Self::FeeRate,
-    inputs: Vec<OutputFor<Rpc>>,
-    payments: Vec<Payment<AddressFor<Rpc>>>,
-    change: Option<KeyFor<Rpc>>,
-  ) -> PlannedTransaction<Rpc, Self::SignableTransaction, EffectedReceivedOutputs<Rpc>> {
+    inputs: Vec<OutputFor<Rpc<D>>>,
+    payments: Vec<Payment<AddressFor<Rpc<D>>>>,
+    change: Option<KeyFor<Rpc<D>>>,
+  ) -> PlannedTransaction<Rpc<D>, Self::SignableTransaction, EffectedReceivedOutputs<Rpc<D>>> {
     let key = inputs.first().unwrap().key();
     for input in &inputs {
       assert_eq!(key, input.key());
     }
 
     let singular_spent_output = (inputs.len() == 1).then(|| inputs[0].id());
-    match signable_transaction(fee_rate, inputs, payments, change) {
+    match signable_transaction::<D>(fee_rate, inputs.clone(), payments, change) {
       Ok(tx) => PlannedTransaction {
         signable: tx.0,
         eventuality: Eventuality { txid: tx.1.txid(), singular_spent_output },
@@ -153,7 +162,14 @@ impl TransactionPlanner<Rpc, EffectedReceivedOutputs<Rpc>> for Planner {
 
           let mut res = vec![];
           for output in scanner.scan_transaction(tx) {
-            res.push(Output::new(key, tx, output));
+            res.push(Output::new_with_presumed_origin(
+              key,
+              tx,
+              // It shouldn't matter if this is wrong as we should never try to return these
+              // We still provide an accurate value to ensure a lack of discrepancies
+              Some(Address::new(inputs[0].output.output().script_pubkey.clone()).unwrap()),
+              output,
+            ));
           }
           res
         }),
@@ -174,4 +190,4 @@ impl TransactionPlanner<Rpc, EffectedReceivedOutputs<Rpc>> for Planner {
   }
 }
 
-pub(crate) type Scheduler = GenericScheduler<Rpc, Planner>;
+pub(crate) type Scheduler<D> = GenericScheduler<Rpc<D>, Planner>;