node: send pings, disconnect peer if it doesn't complete handshake or pong in time (lian#132)

Author: mhanne

bin/bitcoin_node

@@ -67,6 +67,11 @@ optparse = OptionParser.new do |opts|
     options[:check_blocks] = check.to_i
   end
 
+  opts.on("--connection-timeout TIMEOUT",
+          "Disconnect peer if it doesn't respond to handshake or ping within TIMEOUT seconds (default: 10)") do |timeout|
+    options[:connection_timeout] = timeout.to_i
+  end
+
   opts.on("--ho", "--headers-only",
     "Download only block headers") do
     options[:headers_only] = true
@@ -109,7 +114,7 @@ optparse = OptionParser.new do |opts|
     end
   end
 
-  [:queue, :inv_queue, :blocks, :addrs, :connect].each do |name|
+  [:queue, :inv_queue, :addrs, :connect, :relay, :ping].each do |name|
     opts.on("--i#{name.to_s[0]}", "--interval-#{name} [SECONDS]",
       "Interval for #{name} worker (default: #{options[:intervals][name]})") do |sec|
       options[:intervals][name] = sec.to_i

lib/bitcoin/network/command_handler.rb

@@ -322,18 +322,19 @@ def handle_config
   #  bitcoin_node connections
   def handle_connections
     @node.connections.sort{|x,y| y.uptime <=> x.uptime}.map{|c|
-      {
+      res = {
         type: c.direction, host: c.host, port: c.port, state: c.state,
         uptime: c.uptime,
-        version: {
+        version: c.version ? {
           version: c.version.version,
           services: c.version.services,
           time: c.version.time,
           nonce: c.version.nonce,
           block: c.version.last_block,
-          client: (c.version.user_agent rescue '?'),
+          client: (c.version.user_agent),
           relay: c.version.relay,
-        }
+        } : nil,
+        latency: c.latency_ms.to_i,
       }
     }
   end

lib/bitcoin/network/connection_handler.rb

@@ -7,8 +7,6 @@ module Bitcoin::Network
   # Node network connection to a peer. Handles all the communication with a specific peer.
   class ConnectionHandler < EM::Connection
 
-    LATENCY_MAX = (5*60*1000) # 5min in ms
-
     include Bitcoin
     include Bitcoin::Storage
 
@@ -35,10 +33,10 @@ def initialize node, host, port, direction
       @parser = Bitcoin::Protocol::Parser.new(self)
       @state = :new
       @version = nil
-      @started = nil
+      @started = Time.now
       @port, @host = *Socket.unpack_sockaddr_in(get_peername)  if get_peername
       @ping_nonce = nil
-      @latency_ms = nil
+      @latency_ms = @node.config[:connection_timeout] * 1000
       @lock = Monitor.new
       @last_getblocks = []  # the last few getblocks messages received
     rescue
@@ -48,9 +46,7 @@ def initialize node, host, port, direction
 
     # check if connection is wanted, begin handshake if it is, disconnect if not
     def post_init
-      if incoming?
-        begin_handshake
-      end
+      begin_handshake  if incoming?
     rescue
       log.fatal { "Error in #post_init" }
       p $!; puts *$@
@@ -83,15 +79,23 @@ def unbind
     end
 
     # begin handshake
-    # TODO: disconnect if we don't complete within a reasonable time
     def begin_handshake
       if incoming? && !@node.accept_connections?
         return close_connection  unless @node.config[:connect].include?([@host, @port.to_s])
       end
       log.info { "Established #{@direction} connection" }
       @node.connections << self
       @state = :handshake
+
+      EM.add_timer(@node.config[:connection_timeout]) do
+        unless @state == :connected
+          log.debug { "Handshake timed out." }
+          close_connection
+        end
+      end
+
       send_version
+
     rescue
       log.fatal { "Error in #begin_handshake" }
       p $!; puts *$@
@@ -102,6 +106,7 @@ def complete_handshake
       if @state == :handshake
         log.debug { 'Handshake completed' }
         @state = :connected
+        @latency_ms = Time.now - @started
         @started = Time.now
         @node.push_notification(:connection, info.merge(type: :connected))
         @node.addrs << addr
@@ -301,18 +306,23 @@ def send_getaddr
       send_data(Protocol.pkt("getaddr", ""))
     end
 
-    # send +ping+ message
-    # TODO: wait for pong and disconnect if it doesn't arrive (and version is new enough)
-    def send_ping
+    # send +ping+ message. if +wait+ is true, wait for pong and disconnect
+    # if it doesn't arrive in time (and version is new enough)
+    def send_ping wait = true
       if @version.version > Bitcoin::Protocol::BIP0031_VERSION
-        @latency_ms = LATENCY_MAX
         @ping_nonce = rand(0xffffffff)
         @ping_time = Time.now
         log.debug { "<< ping (#{@ping_nonce})" }
+        EM.add_timer(@node.config[:connection_timeout]) do
+          unless @latency_ms / 1000 < @node.config[:connection_timeout]
+            log.debug { "Ping #{@ping_nonce} timed out." }
+            close_connection
+          end
+        end
         send_data(Protocol.ping_pkt(@ping_nonce))
       else
-        # set latency to 5 seconds, terrible but this version should be obsolete now
-        @latency_ms = (5*1000)
+        # set latency to maximum allowed, terrible but this version should be obsolete now
+        @latency_ms = @node.config[:connection_timeout]
         log.debug { "<< ping" }
         send_data(Protocol.ping_pkt)
       end

lib/bitcoin/network/node.rb

@@ -84,10 +84,12 @@ class Node
         :addrs => 5,
         :connect => 5,
         :relay => 0,
+        :ping => 90,
       },
       :import => nil,
       :skip_validation => false,
       :check_blocks => 1000,
+      :connection_timeout => 10,
     }
 
     def initialize config = {}
@@ -177,7 +179,7 @@ def uptime
 
     def start_timers
       return EM.add_timer(1) { start_timers }  if @importing
-      [:queue, :inv_queue, :addrs, :connect, :relay].each do |name|
+      [:queue, :inv_queue, :addrs, :connect, :relay, :ping].each do |name|
         interval = @config[:intervals][name].to_f
         next  if !interval || interval == 0.0
         @timers[name] = EM.add_periodic_timer(interval, method("work_#{name}"))
@@ -379,6 +381,12 @@ def getblocks locator = store.get_locator
       end
     end
 
+    # send ping to every connected peer (which will disconnect if it doesn't respond in time)
+    def work_ping
+      log.debug { "ping worker running" }
+      @connections.select(&:connected?).each(&:send_ping)
+    end
+
     # check if the addr store is full and request new addrs
     # from a random peer if it isn't
     def work_addrs