Cake update v2

Cargo.toml

@@ -16,6 +16,7 @@ default-run = "electrs"
 [features]
 liquid = ["elements"]
 electrum-discovery = ["electrum-client"]
+silent-payments = []
 bench = []
 otlp-tracing = [
     "tracing",
@@ -40,6 +41,7 @@ dirs = "5.0.1"
 elements = { version = "0.25", features = ["serde"], optional = true }
 error-chain = "0.12.4"
 glob = "0.3"
+hex = "0.4"
 itertools = "0.12"
 lazy_static = "1.3.0"
 libc = "0.2.81"
@@ -62,6 +64,7 @@ tiny_http = "0.12.0"
 url = "2.2.0"
 hyper = "0.14"
 hyperlocal = "0.8"
+silentpayments = { git = "https://github.com/cygnet3/rust-silentpayments", branch = "master" }
 # close to same tokio version as dependent by hyper v0.14 and hyperlocal 0.8 -- things can go awry if they mismatch
 tokio = { version = "1", features = ["sync", "macros", "rt-multi-thread", "rt"] }
 opentelemetry = { version = "0.20.0", features = ["rt-tokio"], optional = true }

doc/schema.md

@@ -8,13 +8,14 @@ The index is stored as three RocksDB databases:
 
 ### Indexing process
 
-The indexing is done in the two phase, where each can be done concurrently within itself.
-The first phase populates the `txstore` database, the second phase populates the `history` database.
+The indexing is done in three phases, where each can be done concurrently within itself.
+The first phase populates the `txstore` database, the second phase populates the `history` database, and the third populates the `tweak` database.
 
 NOTE: in order to construct the history rows for spending inputs in phase #2, we rely on having the transactions being processed at phase #1, so they can be looked up efficiently (using parallel point lookups).
 
 After the indexing is completed, both funding and spending are indexed as independent rows under `H{scripthash}`, so that they can be queried in-order in one go.
 
+
 ### `txstore`
 
 Each block results in the following new rows:
@@ -60,6 +61,22 @@ Each block results in the following new row:
 
  * `"D{blockhash}" → ""` (signifies the block was indexed)
 
+#### Silent Payments only
+
+### `tweaks`
+
+Each block results in the following new rows:
+
+ * `"W{blockhash}" → "{tweak1}...{tweakN}"` (list of txids included in the block)
+
+Each transaction results in the following new rows:
+
+ * `"K{blockheight}{txid}" → "{tweak_data1}...{tweak_dataN}"` (txid -> tweak, and list of vout:amount:scripthash for each valid sp output)
+
+Every time a block is scanned for tweaks, the following row is updated:
+
+ * `"B{blockheight}" → "{tip}"` (the blockheight scanned -> the tip of the chain during scanning, this allows to cache the last tip and avoid running spent checks for the same block multiple times, until a new block comes in, then outputs need to be checked if they were spent the next time they are scanned)
+
 #### Elements only
 
 Assets (re)issuances results in the following new rows (only for user-issued assets):

flake.nix

@@ -58,7 +58,11 @@
             inherit src buildInputs nativeBuildInputs;
           } // envVars;
 
-          cargoArtifacts = craneLib.buildDepsOnly commonArgs;
+          # Build deps for all features (including silent-payments) so vendored/cached
+          # crates include bech32 0.9.1 required by silentpayments.
+          cargoArtifacts = craneLib.buildDepsOnly (commonArgs // {
+            cargoExtraArgs = "--all-features";
+          });
           bin = craneLib.buildPackage (commonArgs // {
             inherit cargoArtifacts;
           });

src/bin/electrs.rs

@@ -49,6 +49,12 @@ fn fetch_from(config: &Config, store: &Store) -> FetchFrom {
 }
 
 fn run_server(config: Arc<Config>, salt_rwlock: Arc<RwLock<String>>) -> Result<()> {
+    rayon::ThreadPoolBuilder::new()
+        .num_threads(16)
+        .thread_name(|i| format!("history-{}", i))
+        .build()
+        .unwrap();
+
     let (block_hash_notify, block_hash_receive) = channel::bounded(1);
     let signal = Waiter::start(block_hash_receive);
     let metrics = Metrics::new(config.monitoring_addr);
@@ -95,7 +101,6 @@ fn run_server(config: Arc<Config>, salt_rwlock: Arc<RwLock<String>>) -> Result<(
         &metrics,
         Arc::clone(&config),
     )));
-
     while !Mempool::update(&mempool, &daemon, &tip)? {
         // Mempool syncing was aborted because the chain tip moved;
         // Index the new block(s) and try again.

src/config.rs

@@ -78,6 +78,12 @@ pub struct Config {
     pub electrum_announce: bool,
     #[cfg(feature = "electrum-discovery")]
     pub tor_proxy: Option<std::net::SocketAddr>,
+    pub sp_begin_height: Option<usize>,
+    pub sp_min_dust: Option<usize>,
+    pub sp_check_spends: bool,
+    pub skip_history: bool,
+    pub skip_tweaks: bool,
+    pub skip_mempool: bool,
 }
 
 fn str_to_socketaddr(address: &str, what: &str) -> SocketAddr {
@@ -256,6 +262,14 @@ impl Config {
                     .long("zmq-addr")
                     .help("Optional zmq socket address of the bitcoind daemon")
                     .takes_value(true),
+            ).arg(
+            Arg::with_name("skip_history")
+                .long("skip-history")
+                .help("Skip history indexing"),
+        ).arg(
+                Arg::with_name("skip_mempool")
+                    .long("skip-mempool")
+                    .help("Skip local mempool"),
             );
 
         #[cfg(unix)]
@@ -298,6 +312,31 @@ impl Config {
                 .takes_value(true),
         );
 
+        #[cfg(feature = "silent-payments")]
+        let args = args
+            .arg(
+                Arg::with_name("sp_begin_height")
+                    .long("sp-begin-height")
+                    .help("Block height at which to begin scanning for silent payments")
+                    .takes_value(true),
+            )
+            .arg(
+                Arg::with_name("sp_min_dust")
+                    .long("sp-min-dust")
+                    .help("Minimum dust value for silent payments")
+                    .takes_value(true),
+            )
+            .arg(
+                Arg::with_name("sp_check_spends")
+                    .long("sp-check-spends")
+                    .help("Check spends of silent payments"),
+            )
+            .arg(
+                Arg::with_name("skip_tweaks")
+                    .long("skip-tweaks")
+                    .help("Skip tweaks indexing"),
+            );
+
         let m = args.get_matches();
 
         let network_name = m.value_of("network").unwrap_or("mainnet");
@@ -504,6 +543,12 @@ impl Config {
             electrum_announce: m.is_present("electrum_announce"),
             #[cfg(feature = "electrum-discovery")]
             tor_proxy: m.value_of("tor_proxy").map(|s| s.parse().unwrap()),
+            sp_begin_height: m.value_of("sp_begin_height").map(|s| s.parse().unwrap()),
+            sp_min_dust: m.value_of("sp_min_dust").map(|s| s.parse().unwrap()),
+            sp_check_spends: m.is_present("sp_check_spends"),
+            skip_history: m.is_present("skip_history"),
+            skip_tweaks: m.is_present("skip_tweaks"),
+            skip_mempool: m.is_present("skip_mempool"),
         };
         eprintln!("{:?}", config);
         config

src/daemon.rs

@@ -76,7 +76,7 @@ fn block_from_value(value: Value) -> Result<Block> {
     Ok(deserialize(&block_bytes).chain_err(|| format!("failed to parse block {}", block_hex))?)
 }
 
-fn tx_from_value(value: Value) -> Result<Transaction> {
+pub fn tx_from_value(value: Value) -> Result<Transaction> {
     let tx_hex = value.as_str().chain_err(|| "non-string tx")?;
     let tx_bytes = Vec::from_hex(tx_hex).chain_err(|| "non-hex tx")?;
     Ok(deserialize(&tx_bytes).chain_err(|| format!("failed to parse tx {}", tx_hex))?)
@@ -667,7 +667,8 @@ impl Daemon {
     pub fn gettransaction_raw(
         &self,
         txid: &Txid,
-        blockhash: &BlockHash,
+        // WARN: gettransaction_raw with blockhash=None requires bitcoind with txindex=1
+        blockhash: Option<&BlockHash>,
         verbose: bool,
     ) -> Result<Value> {
         self.request("getrawtransaction", json!([txid, verbose, blockhash]))