Stop reactor leaks, TLS handshake hangs and reactor spin

CHANGELOG.md

@@ -1,3 +1,22 @@
+# 3.2.6
+Bug Fixes:
+* `Cluster#close` / `Cluster#close_async` now always stop the ione IO reactor, so a closed cluster can never leave its reactor thread behind.
+* Closing a cluster whose control connection never connected used to return a future that never resolved and left the reactor running; `ControlConnection#close_async` now stops the reactor in that state too.
+* The connect timeout now bounds the TLS handshake as well as the TCP connect. A peer that accepts TCP but never answers the handshake used to hang connects forever.
+* A TLS socket waiting for the server's handshake bytes is selected for readability. Selecting it for writability made the reactor spin at 100% CPU for as long as the handshake was pending.
+* Closing a TLS connection now closes the underlying TCP socket too, instead of leaking the file descriptor until GC.
+* The reactor sleeps until there is IO, a timer is due or it is unblocked, instead of waking every second, and evicts sockets whose file descriptor was closed underneath it.
+* The reactor thread is named `io_reactor` so leaked reactors can be counted per process.
+* `Reconnection::Policies::Exponential` accepts a `jitter:` fraction (e.g. `Exponential.new(1, 60, 2, jitter: 0.25)`) so a fleet of processes does not reconnect in lockstep; the ceiling is always honoured.
+* Reactor restarts restore the wake-up pipe atomically with the start transition, including when shutdown completes during a restart request; shutdown drains use a fixed tick even when timers are overdue.
+* Connection attempts sleep until their nearest deadline, and timers scheduled on the reactor thread avoid redundant wake-ups.
+* Closing or draining connections and listeners wakes the reactor so idle sockets and listening ports are released promptly.
+* Socket `IOError` and `EBADF` failures during connect, read, or flush close only the affected socket; transient accept errors leave the listener available for retry.
+* Completed TCP connections and TLS handshakes take precedence over expired deadlines, including connections queued while the reactor is stopped.
+* `connect_timeout: Float::INFINITY` continues to allow unbounded TCP connects and TLS handshakes.
+* Reconnection jitter samples within the bounded window, avoiding a concentration of retries at the maximum interval.
+* Requires ione 1.3.x (`~> 1.3.0`).
+
 # 3.2.5
 Bug Fixes:
 * [RUBY-293](https://datastax-oss.atlassian.net/browse/RUBY-293) Infinite loop when connecting with allow_beta_protocol

Gemfile.lock

@@ -8,8 +8,8 @@ GIT
 PATH
   remote: .
   specs:
-    cassandra-driver (3.2.5)
-      ione
+    cassandra-driver (3.2.6)
+      ione (~> 1.3.0)
       sorted_set
 
 GEM

cassandra-driver.gemspec

@@ -30,7 +30,7 @@ Gem::Specification.new do |s|
     s.files << 'ext/cassandra_murmur3/cassandra_murmur3.c'
   end
 
-  s.add_runtime_dependency 'ione'
+  s.add_runtime_dependency 'ione', '~> 1.3.0'
   s.add_runtime_dependency 'sorted_set'
 
   s.add_development_dependency 'bundler'

lib/cassandra/cluster.rb

@@ -325,6 +325,7 @@ def inspect
 require 'cassandra/cluster/connector'
 require 'cassandra/cluster/control_connection'
 require 'cassandra/cluster/failed_connection'
+require 'cassandra/cluster/io_reactor'
 require 'cassandra/cluster/metadata'
 require 'cassandra/cluster/options'
 require 'cassandra/cluster/registry'

lib/cassandra/cluster/control_connection.rb

@@ -40,7 +40,7 @@ def initialize(logger, io_reactor, cluster_registry, cluster_schema,
         @schema_fetcher        = schema_fetcher
         @refreshing_statuses   = ::Hash.new(false)
         @refresh_schema_future = nil
-        @status                = :closed
+        @status                = :disconnected
         @refreshing_hosts      = false
         @refreshing_host       = ::Hash.new(false)
         @closed_promise        = Ione::Promise.new
@@ -59,6 +59,10 @@ def on_close(&block)
       def connect_async
         synchronize do
           return Ione::Future.resolved if @status == :connecting || @status == :connected
+          if @status == :closing
+            return Ione::Future.failed(Errors::ClientError.new('Control connection is closing'))
+          end
+          @closed_promise = Ione::Promise.new if @status == :closed
           @status = :connecting
         end
 
@@ -87,7 +91,7 @@ def host_up(host)
           @io_reactor.cancel_timer(timer) if timer
 
           unless @connection ||
-                 (@status == :closing || @status == :closed) ||
+                 (@status == :disconnected || @status == :closing || @status == :closed) ||
                  @load_balancing_policy.distance(host) == :ignore
             return connect_to_first_available(
               @load_balancing_policy.plan(nil, VOID_STATEMENT, VOID_OPTIONS)
@@ -127,20 +131,27 @@ def host_down(host)
       end
 
       def close_async
-        synchronize do
-          return @closed_promise.future if @status == :closing || @status == :closed
+        promise = synchronize do
+          if @status == :closing || @status == :closed
+            return @closed_promise.future
+          end
           @status = :closing
+          @closed_promise
         end
         f = @io_reactor.stop
 
         f.on_value(&method(:connection_closed))
         f.on_failure(&method(:connection_closed))
 
-        @closed_promise.future
+        promise.future
       end
 
       def connection_closed(cause)
-        @closed_promise.fulfill
+        promise = synchronize do
+          @status = :closed
+          @closed_promise
+        end
+        promise.fulfill
       end
 
       def inspect
@@ -633,9 +644,7 @@ def connect_to_first_available(plan, errors = nil)
 
               synchronize do
                 if connection == @connection
-                  if @status == :closing
-                    @status = :closed
-                  else
+                  unless @status == :closing || @status == :closed
                     @status = :reconnecting
                     reconnect = true
                   end