Mempool

Gemfile

@@ -7,7 +7,7 @@ gem 'bitcoin-ruby',
   # path: '../bitcoin-ruby'
 
 gem 'bitcoin-ruby-blockchain',
-  git: "git://github.com/mhanne/bitcoin-ruby-blockchain"
+  git: "git://github.com/mhanne/bitcoin-ruby-blockchain", branch: "mempool"
   # path: '../bitcoin-ruby-blockchain'
 
 gem 'bitcoin-ruby-wallet',

bin/bitcoin_node

@@ -56,6 +56,11 @@ optparse = OptionParser.new do |opts|
     options[:storage] = storage
   end
 
+  opts.on("-m", "--mempool [DB]",
+    "Use given mempool db (default: #{options[:mempool]})") do |mempool|
+    options[:mempool][:db] = mempool
+  end
+
   opts.on("--skip-validation", "Skip validation of blocks and transactions") do
     options[:skip_validation] = true
   end

lib/bitcoin/node.rb

@@ -102,6 +102,12 @@ class Node
       skip_validation: false,
       check_blocks: 1000,
       connection_timeout: 10,
+      mempool: {
+        db: "sqlite:/",
+        max_age: 3600,
+        log_level: :debug,
+      },
+
     }
 
     def initialize config = {}
@@ -127,7 +133,8 @@ def set_store
         skip_validation: @config[:skip_validation],
         index_nhash: @config[:index_nhash],
         index_p2sh_type: @config[:index_p2sh_type],
-        log_level: @config[:log][:storage]
+        log_level: @config[:log][:storage],
+        mempool: @config[:mempool],
       })
       @store.log.level = @config[:log][:storage]
       @store.check_consistency(@config[:check_blocks])
@@ -301,6 +308,10 @@ def run
           end
         end
 
+        @store.mempool.subscribe(:accepted) do |data|
+          tx = Bitcoin::P::Tx.new(data[:payload].htb)
+          push_notification :tx, [tx, 0]
+        end
       end
     end
 
@@ -440,6 +451,7 @@ def work_queue
               exit
             end
           else
+            @store.mempool.add(obj[1])  if @store.in_sync?
             drop = @unconfirmed.size - @config[:max][:unconfirmed] + 1
             drop.times { @unconfirmed.shift }  if drop > 0
             unless @unconfirmed[obj[1].hash]
@@ -467,7 +479,9 @@ def work_inv_queue
       return  if @queue.size >= @config[:max][:queue]
       while inv = @inv_queue.shift
         next  if [email protected]_sync? && inv[0] == :tx && @notifiers.empty?
+        @store.mempool.inv(inv[1])  if inv[0] == :tx && @store.in_sync?
         next  if @queue.map{|i|i[1]}.map(&:hash).include?(inv[1])
+        next  if @store.in_sync? && @store.mempool.exists?(inv[1])
         inv[2].send("send_getdata_#{inv[0]}", inv[1])
       end
     end

lib/bitcoin/node/command_handler.rb

@@ -108,6 +108,15 @@ def handle_unmonitor request
     { id: id }
   end
 
+  [:accepted, :rejected, :doublespend, :seen, :confirmed].each do |type|
+    define_method("handle_monitor_mempool_#{type}") do |request, params|
+      monitor_id = @monitors.size
+      id = @node.store.mempool.subscribe(type) do |data|
+        respond(request, data)
+      end
+    end
+  end
+
   # Handle +monitor block+ command;
   def handle_monitor_block request, params
     monitor_id = @monitors.size
@@ -127,7 +136,6 @@ def respond_missed_blocks request, monitor_id
   end
 
   def respond_monitor_block request, block, height = nil
-    p :mbb
     height ||= block.height
     respond(request, { hash: block.hash, hex: block.to_payload.hth, height: height, depth: height })
   end
@@ -283,6 +291,12 @@ def handle_info
         peers: (blocks.inject{|a,b| a+=b; a } / blocks.size rescue '?' ),
         sync: @node.store.in_sync?,
       },
+      mempool: {
+        total: @node.store.mempool.transactions.count,
+        accepted: @node.store.mempool.accepted.count,
+        rejected: @node.store.mempool.rejected.count,
+        doublespends: @node.store.mempool.doublespend.count,
+      },
       addrs: {
         alive: @node.addrs.select{|a| a.alive?}.size,
         total: @node.addrs.size,
@@ -305,6 +319,17 @@ def handle_info
     Bitcoin.namecoin? ? {names: @node.store.db[:names].count}.merge(info) : info
   end
 
+  def handle_mempool_stats
+    mempool = @node.store.mempool
+    {
+      total: mempool.transactions.count,
+      accepted: mempool.accepted.count,
+      rejected: mempool.rejected.count,
+      doublespend: mempool.doublespend.count,
+      oldest: (Time.now - mempool.transactions.order(:created_at).first[:created_at]).to_i
+    }
+  end
+
   def add_monitor params, channels
     @monitors << { params: params, channels: channels }
     @monitors.size - 1
@@ -491,6 +516,7 @@ def handle_relay_tx request, params = {}
                      details: validator.error })
     end
 
+    @node.store.mempool.add(tx)
     #@node.store.store_tx(tx)
     @node.relay_tx[tx.hash] = tx
     @node.relay_propagation[tx.hash] = 0
@@ -533,8 +559,10 @@ def handle_store_block params
   #  { method: "store_tx", params: { hex: <tx data in hex> } }
   def handle_store_tx params
     tx = Bitcoin::P::Tx.new(params[:hex].htb)
-    @node.queue << [:tx, tx]
-    { queued: tx.hash }
+#    @node.queue << [:tx, tx]
+#    { queued: tx.hash }
+    @node.store.mempool.add(tx, params[:skip_validation])
+    { added: tx.hash }
   end
 
   # # format node uptime

lib/bitcoin/node/connection_handler.rb

@@ -109,6 +109,9 @@ def complete_handshake
       @node.push_notification(:connection, info.merge(type: :connected))
       @node.addrs << addr
       send_data P::Addr.pkt(@node.addr)  if @node.config[:announce]
+      # send_data Bitcoin::P.pkt("mempool", "")
+      send_getblocks
+
     end
 
     # error parsing a message, log as warning but otherwise ignore
@@ -120,9 +123,12 @@ def on_error(type, data)
     # add to inv_queue, unlesss maximum is reached
     def on_inv_transaction(hash)
       log.debug { ">> inv transaction: #{hash.hth}" }
+
+      # TODO: get relay propagation from mempool
       if @node.relay_propagation.keys.include?(hash.hth)
         @node.relay_propagation[hash.hth] += 1
       end
+
       return  if @node.inv_queue.size >= @node.config[:max][:inv]
       @node.queue_inv([:tx, hash, self])
     end

spec/node/command_api_spec.rb

@@ -522,10 +522,9 @@ def store_block block
         r1 = send "monitor", channel: "tx"
         should_receive r1, id: 0
         tx = @block.tx[0]
-        r2 = send "store_tx", hex: tx.to_payload.hth
-        should_receive r2, { "queued" => tx.hash }
-
+        r2 = send "store_tx", hex: tx.to_payload.hth, skip_validation: true
         should_receive_tx(r1, tx, 0)
+        should_receive r2, { "added" => tx.hash }
       end
 
       it "should unmonitor tx" do
@@ -537,7 +536,7 @@ def store_block block
 
         tx = @block.tx[0]
         r3 = send "store_tx", hex: tx.to_payload.hth
-        should_receive r3, { "queued" => tx.hash }
+        should_receive r3, { "added" => tx.hash }
 
         test_command("tslb") {|r| (0..TSLB_TIMEOUT).include?(r['tslb']).should == true }
       end
@@ -610,9 +609,9 @@ def store_block block
         r1 = send "monitor", channel: "output"
         should_receive r1, id: 0
         tx = @block.tx[0]
-        r2 = send "store_tx", hex: tx.to_payload.hth
-        should_receive r2, { "queued" => tx.hash }
+        r2 = send "store_tx", hex: tx.to_payload.hth, skip_validation: true
         should_receive_output(r1, tx, 0, 0)
+        should_receive r2, { "added" => tx.hash }
       end
 
       it "should unmonitor outputs" do
@@ -621,7 +620,7 @@ def store_block block
 
         tx = @block.tx[0]
         r2 = send "store_tx", hex: tx.to_payload.hth
-        should_receive r2, { "queued" => tx.hash }
+        should_receive r2, { "added" => tx.hash }
 
         test_command("tslb") {|r| (0..TSLB_TIMEOUT).include?(r['tslb']).should == true }
       end